mirror of
https://github.com/heygen-com/hyperframes.git
synced 2026-09-03 04:38:33 +00:00
653 lines
23 KiB
TypeScript
653 lines
23 KiB
TypeScript
// fallow-ignore-file unused-type code-duplication complexity
|
||
/**
|
||
* Streaming Encoder Service
|
||
*
|
||
* Pipes frame screenshot buffers directly to FFmpeg's stdin via `-f image2pipe`
|
||
* instead of writing them to disk and reading them back in a separate encode
|
||
* stage. Inspired by Remotion's approach to browser-based video rendering.
|
||
*
|
||
* Two building blocks:
|
||
* 1. Frame reorder buffer – ensures out-of-order parallel workers feed
|
||
* frames to FFmpeg stdin in sequential order.
|
||
* 2. Streaming FFmpeg encoder – spawns FFmpeg with `-f image2pipe` and
|
||
* exposes an async `writeFrame(buffer)` + `close()` API.
|
||
*/
|
||
|
||
import { spawn, type ChildProcess } from "child_process";
|
||
import { once } from "events";
|
||
import { trackChildProcess } from "../utils/processTracker.js";
|
||
import { existsSync, mkdirSync, statSync } from "fs";
|
||
import { dirname } from "path";
|
||
|
||
import {
|
||
type GpuEncoder,
|
||
getCachedGpuEncoder,
|
||
getGpuEncoderName,
|
||
mapPresetForGpuEncoder,
|
||
} from "../utils/gpuEncoder.js";
|
||
import { formatFfmpegError } from "../utils/runFfmpeg.js";
|
||
import { getFfmpegBinary } from "../utils/ffmpegBinaries.js";
|
||
import { getHdrEncoderColorParams } from "../utils/hdr.js";
|
||
import { withEvenDimensionPad } from "../utils/evenDimensions.js";
|
||
import { DEFAULT_CONFIG, type EngineConfig } from "../config.js";
|
||
import { fpsToFfmpegArg, type Fps } from "@hyperframes/core";
|
||
import { appendVp9CpuUsedArg } from "./vp9Options.js";
|
||
|
||
// Re-export EncoderOptions so callers can reference the type via this module.
|
||
export type { EncoderOptions } from "./chunkEncoder.types.js";
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// 1. Frame reorder buffer — ordered async barrier
|
||
// ---------------------------------------------------------------------------
|
||
//
|
||
// Parallel workers produce frames out of order; FFmpeg's stdin expects them in
|
||
// strict sequential order. Each worker calls `waitForFrame(n)` to block until
|
||
// its turn, writes, then calls `advanceTo(n + 1)` to release the next waiter.
|
||
//
|
||
// `pending` holds an array per frame index (not a single resolver) so that
|
||
// `waitForAllDone` can coexist with the writer still waiting on the final
|
||
// frame without one clobbering the other.
|
||
|
||
export interface FrameReorderBuffer {
|
||
waitForFrame: (frame: number) => Promise<void>;
|
||
advanceTo: (frame: number) => void;
|
||
waitForAllDone: () => Promise<void>;
|
||
/**
|
||
* Reject every parked and future waiter with `err`. Required by the
|
||
* interleaved parallel drain: when one worker fails (e.g. drawElement
|
||
* self-verification), its frames will never be written — peers parked in
|
||
* waitForFrame would otherwise deadlock the whole capture (the worker pool
|
||
* awaits ALL workers before surfacing the failure).
|
||
*/
|
||
abort: (err: Error) => void;
|
||
}
|
||
|
||
export function createFrameReorderBuffer(startFrame: number, endFrame: number): FrameReorderBuffer {
|
||
let cursor = startFrame;
|
||
let aborted: Error | null = null;
|
||
const pending = new Map<number, Array<{ resolve: () => void; reject: (e: Error) => void }>>();
|
||
|
||
const enqueueAt = (frame: number, resolve: () => void, reject: (e: Error) => void): void => {
|
||
const list = pending.get(frame);
|
||
if (list === undefined) {
|
||
pending.set(frame, [{ resolve, reject }]);
|
||
} else {
|
||
list.push({ resolve, reject });
|
||
}
|
||
};
|
||
|
||
const flushAt = (frame: number): void => {
|
||
const list = pending.get(frame);
|
||
if (list === undefined) return;
|
||
pending.delete(frame);
|
||
for (const waiter of list) waiter.resolve();
|
||
};
|
||
|
||
const waitForFrame = (frame: number): Promise<void> =>
|
||
new Promise<void>((resolve, reject) => {
|
||
if (aborted) {
|
||
reject(aborted);
|
||
return;
|
||
}
|
||
if (frame === cursor) {
|
||
resolve();
|
||
return;
|
||
}
|
||
enqueueAt(frame, resolve, reject);
|
||
});
|
||
|
||
const advanceTo = (frame: number): void => {
|
||
cursor = frame;
|
||
flushAt(frame);
|
||
};
|
||
|
||
const waitForAllDone = (): Promise<void> =>
|
||
new Promise<void>((resolve, reject) => {
|
||
if (aborted) {
|
||
reject(aborted);
|
||
return;
|
||
}
|
||
if (cursor >= endFrame) {
|
||
resolve();
|
||
return;
|
||
}
|
||
enqueueAt(endFrame, resolve, reject);
|
||
});
|
||
|
||
const abort = (err: Error): void => {
|
||
if (aborted) return;
|
||
aborted = err;
|
||
for (const [frame, list] of pending) {
|
||
pending.delete(frame);
|
||
for (const waiter of list) waiter.reject(err);
|
||
}
|
||
};
|
||
|
||
return { waitForFrame, advanceTo, waitForAllDone, abort };
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// 2. Streaming FFmpeg encoder
|
||
// ---------------------------------------------------------------------------
|
||
|
||
export interface StreamingEncoderOptions {
|
||
/** Frame rate as an exact rational; see `Fps` in @hyperframes/core. */
|
||
fps: Fps;
|
||
width: number;
|
||
height: number;
|
||
codec?: "h264" | "h265" | "vp9" | "prores";
|
||
preset?: string;
|
||
quality?: number;
|
||
bitrate?: string;
|
||
pixelFormat?: string;
|
||
/** libvpx-vp9 -cpu-used value. Defaults to the engine VP9 setting. */
|
||
vp9CpuUsed?: number;
|
||
useGpu?: boolean;
|
||
imageFormat?: "jpeg" | "png";
|
||
hdr?: { transfer: import("../utils/hdr.js").HdrTransfer };
|
||
/** When set, use rawvideo input instead of image2pipe. For HDR PQ-encoded frames. */
|
||
rawInputFormat?: "rgb48le";
|
||
}
|
||
|
||
export interface StreamingEncoderResult {
|
||
success: boolean;
|
||
durationMs: number;
|
||
fileSize: number;
|
||
error?: string;
|
||
}
|
||
|
||
export interface StreamingEncoder {
|
||
/**
|
||
* Write one frame to FFmpeg stdin, awaiting `drain` when the pipe is full
|
||
* so back-pressure propagates to the caller. Resolves `false` when FFmpeg
|
||
* is already gone. Callers must serialize calls — one in-flight writeFrame
|
||
* per encoder (the frame reorder buffer provides this ordering); concurrent
|
||
* calls would interleave frame bytes on the pipe and race the drain wait.
|
||
*/
|
||
writeFrame: (buffer: Buffer) => Promise<boolean>;
|
||
close: () => Promise<StreamingEncoderResult>;
|
||
getExitStatus: () => "running" | "success" | "error";
|
||
/**
|
||
* The FFmpeg failure reason (exit code + tail of stderr), or `undefined`
|
||
* while the process is still running / exited cleanly. Lets a `writeFrame`
|
||
* that returned `false` because FFmpeg died surface WHY it died (bad args,
|
||
* unsupported codec, disk full) instead of a bare "encoder exited" message.
|
||
*/
|
||
getExitError: () => string | undefined;
|
||
}
|
||
|
||
/**
|
||
* Build FFmpeg args for streaming (image2pipe) input.
|
||
* Reuses the same codec/quality/GPU logic as chunkEncoder's buildEncoderArgs
|
||
* but with `-f image2pipe` instead of `-i <pattern>`.
|
||
*
|
||
* Exported so unit tests can assert on the constructed CLI without spawning
|
||
* FFmpeg — see streamingEncoder.test.ts.
|
||
*/
|
||
export function buildStreamingArgs(
|
||
options: StreamingEncoderOptions,
|
||
outputPath: string,
|
||
gpuEncoder: GpuEncoder = null,
|
||
): string[] {
|
||
const {
|
||
fps,
|
||
codec = "h264",
|
||
preset = "medium",
|
||
quality = 23,
|
||
bitrate,
|
||
pixelFormat = "yuv420p",
|
||
vp9CpuUsed,
|
||
useGpu = false,
|
||
imageFormat = "jpeg",
|
||
} = options;
|
||
|
||
// Input args: pipe from stdin
|
||
const args: string[] = [];
|
||
if (options.rawInputFormat) {
|
||
// Raw pixel input (HLG/PQ-encoded rgb48le from FFmpeg extraction).
|
||
// Tag the input with the correct color space so FFmpeg uses the right
|
||
// YUV matrix when converting rgb48le → yuv420p10le for encoding.
|
||
// Without these tags FFmpeg assumes bt709 and applies the wrong matrix.
|
||
const hdrTransfer = options.hdr?.transfer;
|
||
const inputColorTrc =
|
||
hdrTransfer === "pq" ? "smpte2084" : hdrTransfer === "hlg" ? "arib-std-b67" : undefined;
|
||
args.push(
|
||
"-f",
|
||
"rawvideo",
|
||
"-pix_fmt",
|
||
options.rawInputFormat,
|
||
"-s",
|
||
`${options.width}x${options.height}`,
|
||
"-framerate",
|
||
fpsToFfmpegArg(fps),
|
||
);
|
||
if (inputColorTrc) {
|
||
args.push(
|
||
"-color_primaries",
|
||
"bt2020",
|
||
"-color_trc",
|
||
inputColorTrc,
|
||
"-colorspace",
|
||
"bt2020nc",
|
||
);
|
||
}
|
||
args.push("-i", "-");
|
||
} else {
|
||
const inputCodec = imageFormat === "png" ? "png" : "mjpeg";
|
||
args.push(
|
||
"-f",
|
||
"image2pipe",
|
||
"-vcodec",
|
||
inputCodec,
|
||
"-framerate",
|
||
fpsToFfmpegArg(fps),
|
||
"-i",
|
||
"-",
|
||
);
|
||
}
|
||
args.push("-r", fpsToFfmpegArg(fps));
|
||
|
||
const shouldUseGpu = useGpu && gpuEncoder !== null;
|
||
|
||
if (codec === "h264" || codec === "h265") {
|
||
if (shouldUseGpu) {
|
||
const encoderName = getGpuEncoderName(gpuEncoder, codec);
|
||
args.push("-c:v", encoderName);
|
||
|
||
switch (gpuEncoder) {
|
||
case "nvenc":
|
||
args.push("-preset", mapPresetForGpuEncoder("nvenc", preset));
|
||
if (bitrate) args.push("-b:v", bitrate);
|
||
else args.push("-cq", String(quality));
|
||
break;
|
||
case "videotoolbox":
|
||
if (bitrate) args.push("-b:v", bitrate);
|
||
else {
|
||
const vtQuality = Math.max(0, Math.min(100, 100 - quality * 2));
|
||
args.push("-q:v", String(vtQuality));
|
||
}
|
||
args.push("-allow_sw", "1");
|
||
break;
|
||
case "vaapi":
|
||
args.unshift("-vaapi_device", "/dev/dri/renderD128");
|
||
args.push("-vf", "format=nv12,hwupload");
|
||
if (bitrate) args.push("-b:v", bitrate);
|
||
else args.push("-qp", String(quality));
|
||
break;
|
||
case "qsv":
|
||
args.push("-preset", mapPresetForGpuEncoder("qsv", preset));
|
||
if (bitrate) args.push("-b:v", bitrate);
|
||
else args.push("-global_quality", String(quality));
|
||
break;
|
||
case "amf":
|
||
if (bitrate) args.push("-b:v", bitrate);
|
||
else args.push("-rc", "cqp", "-qp_i", String(quality), "-qp_p", String(quality));
|
||
break;
|
||
}
|
||
|
||
// Mirror SW branch: GPU h264 paths emit B-frames by default (nvenc, amf,
|
||
// qsv, vaapi) and produce the same negative-DTS freeze for downstream players.
|
||
// See chunkEncoder.buildEncoderArgs for the full explanation.
|
||
if (
|
||
codec === "h264" &&
|
||
(gpuEncoder === "nvenc" ||
|
||
gpuEncoder === "qsv" ||
|
||
gpuEncoder === "vaapi" ||
|
||
gpuEncoder === "amf")
|
||
) {
|
||
args.push("-bf", "0");
|
||
if (gpuEncoder === "qsv") {
|
||
args.push("-b_strategy", "0");
|
||
}
|
||
}
|
||
} else {
|
||
const encoderName = codec === "h264" ? "libx264" : "libx265";
|
||
args.push("-c:v", encoderName, "-preset", preset);
|
||
if (bitrate) args.push("-b:v", bitrate);
|
||
else args.push("-crf", String(quality));
|
||
|
||
// Mirrors chunkEncoder: disable B-frames for h264 so PTS == DTS, no
|
||
// negative DTS at stream start. Without this, files freeze on the
|
||
// first frame in VS Code preview, several browsers, and some HW
|
||
// decoders. See chunkEncoder.buildEncoderArgs for the full reasoning.
|
||
if (codec === "h264") {
|
||
args.push("-bf", "0");
|
||
}
|
||
|
||
// Encoder-specific params: anti-banding + color space tagging.
|
||
// For HDR, getHdrEncoderColorParams also emits the SMPTE ST 2086
|
||
// mastering-display and CTA-861.3 MaxCLL/MaxFALL SEI messages —
|
||
// without them, players (Apple, YouTube, HDR TVs) treat the file
|
||
// as SDR BT.2020 and tone-map incorrectly.
|
||
const xParamsFlag = codec === "h264" ? "-x264-params" : "-x265-params";
|
||
const colorParams =
|
||
options.rawInputFormat && options.hdr
|
||
? getHdrEncoderColorParams(options.hdr.transfer).x265ColorParams
|
||
: "colorprim=bt709:transfer=bt709:colormatrix=bt709";
|
||
if (preset === "ultrafast") {
|
||
args.push(xParamsFlag, `aq-mode=3:${colorParams}`);
|
||
} else {
|
||
args.push(xParamsFlag, `aq-mode=3:aq-strength=0.8:deblock=1,1:${colorParams}`);
|
||
}
|
||
// Apple devices require hvc1 tag for HEVC playback (default hev1 won't open in QuickTime)
|
||
if (codec === "h265") {
|
||
args.push("-tag:v", "hvc1");
|
||
}
|
||
}
|
||
} else if (codec === "vp9") {
|
||
args.push("-c:v", "libvpx-vp9", "-b:v", bitrate || "0", "-crf", String(quality));
|
||
args.push("-deadline", preset === "ultrafast" ? "realtime" : "good");
|
||
args.push("-row-mt", "1");
|
||
appendVp9CpuUsedArg(args, vp9CpuUsed);
|
||
if (pixelFormat === "yuva420p") {
|
||
args.push("-auto-alt-ref", "0");
|
||
args.push("-metadata:s:v:0", "alpha_mode=1");
|
||
}
|
||
} else if (codec === "prores") {
|
||
args.push("-c:v", "prores_ks", "-profile:v", preset, "-vendor", "apl0");
|
||
args.push("-pix_fmt", pixelFormat);
|
||
return [...args, "-y", outputPath];
|
||
}
|
||
|
||
// Color space metadata.
|
||
// When rawInputFormat is set, data comes from the WebGPU HDR pipeline
|
||
// (PQ-encoded) — tag with bt2020/PQ truthfully.
|
||
// Otherwise, Chrome captures sRGB — tag as bt709.
|
||
if (codec === "h264" || codec === "h265") {
|
||
if (options.rawInputFormat && options.hdr) {
|
||
args.push(
|
||
"-colorspace:v",
|
||
"bt2020nc",
|
||
"-color_primaries:v",
|
||
"bt2020",
|
||
"-color_trc:v",
|
||
options.hdr.transfer === "pq" ? "smpte2084" : "arib-std-b67",
|
||
"-color_range",
|
||
"tv",
|
||
);
|
||
} else {
|
||
args.push(
|
||
"-colorspace:v",
|
||
"bt709",
|
||
"-color_primaries:v",
|
||
"bt709",
|
||
"-color_trc:v",
|
||
"bt709",
|
||
"-color_range",
|
||
"tv",
|
||
);
|
||
}
|
||
|
||
// Video filter for range/color conversion.
|
||
// Raw HDR input (from WebGPU pipeline) is already PQ-encoded — no conversion needed.
|
||
// Chrome screenshots need full→TV range conversion.
|
||
if (options.rawInputFormat) {
|
||
// No filter needed — PQ data goes straight to encoder
|
||
} else if (gpuEncoder === "vaapi") {
|
||
// vaapi already runs `format=nv12,hwupload`; the nv12 conversion aligns
|
||
// odd dimensions before upload, so only prepend the range conversion.
|
||
const vfIdx = args.indexOf("-vf");
|
||
if (vfIdx !== -1) {
|
||
args[vfIdx + 1] = `scale=in_range=pc:out_range=tv,${args[vfIdx + 1]}`;
|
||
}
|
||
} else if (shouldUseGpu) {
|
||
// nvenc/videotoolbox/qsv/amf feed software frames straight to the HW
|
||
// encoder with no `-vf`. They hit the same "height not divisible by 2"
|
||
// abort as libx264 on an odd-sized 4:2:0 canvas, so pad odd dimensions
|
||
// up to even on the software side before the encode.
|
||
const vf = withEvenDimensionPad("", pixelFormat, options.width, options.height);
|
||
if (vf) args.push("-vf", vf);
|
||
} else {
|
||
// Range conversion: Chrome screenshots are full-range RGB. Pad odd
|
||
// dimensions up to even so libx264/libx265 (4:2:0) don't abort with
|
||
// "height not divisible by 2" on an odd-sized composition canvas.
|
||
args.push(
|
||
"-vf",
|
||
withEvenDimensionPad(
|
||
"scale=in_range=pc:out_range=tv",
|
||
pixelFormat,
|
||
options.width,
|
||
options.height,
|
||
),
|
||
);
|
||
}
|
||
|
||
// Fixed timescale for consistent A/V timing across platforms.
|
||
args.push("-video_track_timescale", "90000");
|
||
}
|
||
|
||
if (gpuEncoder !== "vaapi") {
|
||
args.push("-pix_fmt", pixelFormat);
|
||
}
|
||
|
||
// Belt-and-suspenders against negative DTS at stream start. See chunkEncoder
|
||
// for the full explanation; same playback compatibility class.
|
||
args.push("-avoid_negative_ts", "make_zero");
|
||
|
||
args.push("-y", outputPath);
|
||
return args;
|
||
}
|
||
|
||
/**
|
||
* Spawn a streaming FFmpeg encoder that accepts frame buffers on stdin.
|
||
*/
|
||
export async function spawnStreamingEncoder(
|
||
outputPath: string,
|
||
options: StreamingEncoderOptions,
|
||
signal?: AbortSignal,
|
||
config?: Partial<Pick<EngineConfig, "ffmpegStreamingTimeout">>,
|
||
): Promise<StreamingEncoder> {
|
||
const outputDir = dirname(outputPath);
|
||
if (!existsSync(outputDir)) mkdirSync(outputDir, { recursive: true });
|
||
|
||
let gpuEncoder: GpuEncoder = null;
|
||
if (options.useGpu) {
|
||
gpuEncoder = await getCachedGpuEncoder();
|
||
}
|
||
|
||
const args = buildStreamingArgs(options, outputPath, gpuEncoder);
|
||
|
||
const startTime = Date.now();
|
||
const ffmpeg: ChildProcess = spawn(getFfmpegBinary(), args, {
|
||
stdio: ["pipe", "pipe", "pipe"],
|
||
});
|
||
trackChildProcess(ffmpeg);
|
||
|
||
let exitStatus: "running" | "success" | "error" = "running";
|
||
let stderr = "";
|
||
let exitCode: number | null = null;
|
||
let exitPromiseResolve: ((value: void) => void) | null = null;
|
||
const exitPromise = new Promise<void>((resolve) => (exitPromiseResolve = resolve));
|
||
|
||
// Track stderr for progress and error messages
|
||
ffmpeg.stderr?.on("data", (data: Buffer) => {
|
||
stderr += data.toString();
|
||
});
|
||
|
||
ffmpeg.on("close", (code: number | null) => {
|
||
exitCode = code;
|
||
exitStatus = code === 0 ? "success" : "error";
|
||
exitPromiseResolve?.();
|
||
});
|
||
|
||
ffmpeg.on("error", (err: Error) => {
|
||
exitStatus = "error";
|
||
stderr += `\nProcess error: ${err.message}`;
|
||
exitPromiseResolve?.();
|
||
});
|
||
|
||
ffmpeg.stdin?.on("error", () => {});
|
||
ffmpeg.stdout?.on("error", () => {});
|
||
|
||
// Handle abort signal
|
||
const onAbort = () => {
|
||
if (exitStatus === "running") {
|
||
ffmpeg.kill("SIGTERM");
|
||
}
|
||
};
|
||
if (signal) {
|
||
if (signal.aborted) {
|
||
ffmpeg.kill("SIGTERM");
|
||
} else {
|
||
signal.addEventListener("abort", onAbort, { once: true });
|
||
}
|
||
}
|
||
|
||
// Inactivity timeout: fires only when no frame has been written for
|
||
// `ffmpegStreamingTimeout` ms. A slow-but-progressing capture (e.g. a CI
|
||
// runner under load) keeps resetting the timer on each writeFrame, so total
|
||
// wall-clock render time is unbounded — only a true hang (Chrome dead,
|
||
// capture stuck, no frames arriving) trips SIGTERM. The 600s default was
|
||
// previously a total-render cap, which intermittently killed legitimate
|
||
// slow renders mid-encode (FFmpeg got SIGTERM after most frames were sent;
|
||
// libx264 printed its summary and exited 255, observable as
|
||
// "Streaming encode failed: FFmpeg exited with code 255" with audio:0kB).
|
||
const streamingTimeout = config?.ffmpegStreamingTimeout ?? DEFAULT_CONFIG.ffmpegStreamingTimeout;
|
||
let timer: NodeJS.Timeout | null = null;
|
||
const resetTimer = () => {
|
||
if (timer) clearTimeout(timer);
|
||
timer = setTimeout(() => {
|
||
if (exitStatus === "running") {
|
||
ffmpeg.kill("SIGTERM");
|
||
}
|
||
}, streamingTimeout);
|
||
};
|
||
resetTimer();
|
||
|
||
const waitForDrainOrExit = async (
|
||
stdin: NonNullable<ChildProcess["stdin"]>,
|
||
): Promise<"drain" | "exit"> => {
|
||
// Back-pressure can hit once per frame. Do not race `exitPromise.then(...)`
|
||
// here: V8 retains `.then` reaction-list entries on an unsettled promise,
|
||
// so a one-hour 30fps render under steady back-pressure can accumulate
|
||
// ~108K closures + AbortControllers. Use one-shot listeners for this write
|
||
// instead, then abort them in finally. `close` is the event that flips
|
||
// `exitStatus`; re-check after listener attachment so a close emitted
|
||
// between `stdin.write(false)` and this await cannot hang forever.
|
||
const abortController = new AbortController();
|
||
try {
|
||
const drainPromise = once(stdin, "drain", { signal: abortController.signal }).then(
|
||
() => "drain" as const,
|
||
);
|
||
const closePromise = once(ffmpeg, "close", { signal: abortController.signal }).then(
|
||
() => "exit" as const,
|
||
);
|
||
const racePromise = Promise.race([drainPromise, closePromise]).catch((err: unknown) => {
|
||
if (err instanceof Error && err.name === "AbortError") {
|
||
return "exit" as const;
|
||
}
|
||
throw err;
|
||
});
|
||
|
||
if (exitStatus !== "running") {
|
||
return "exit";
|
||
}
|
||
|
||
return await racePromise;
|
||
} finally {
|
||
abortController.abort();
|
||
}
|
||
};
|
||
|
||
const encoder: StreamingEncoder = {
|
||
writeFrame: async (buffer: Buffer): Promise<boolean> => {
|
||
const stdin = ffmpeg.stdin;
|
||
if (exitStatus !== "running" || !stdin || stdin.destroyed) {
|
||
return false;
|
||
}
|
||
// Copy the buffer before writing — Node streams hold a reference to the
|
||
// provided buffer and drain it asynchronously. The HDR path's compositor
|
||
// reuses pre-allocated transOutput/normalCanvas buffers across frames,
|
||
// so without this copy the pipe would read partially-overwritten data
|
||
// and flicker.
|
||
const copy = Buffer.from(buffer);
|
||
const accepted = stdin.write(copy);
|
||
// Reset inactivity timer immediately ONLY on `accepted === true`. `true`
|
||
// means the write went through to the kernel pipe without buffering in
|
||
// Node — proof FFmpeg is actually consuming. `false` means Node's writable
|
||
// stream had to buffer (FFmpeg hasn't drained the pipe yet); we await
|
||
// `drain` before letting callers produce the next frame, and only reset
|
||
// after drain proves consumption. We deliberately don't reset before
|
||
// drain so a hung FFmpeg with a still-producing Chrome can't keep us
|
||
// alive forever while Node's stdin buffer grows to OOM. If FFmpeg exits
|
||
// before draining, waitForDrainOrExit returns "exit", removes its
|
||
// one-shot listeners, and callers see `false` instead of hanging.
|
||
if (accepted) {
|
||
resetTimer();
|
||
return true;
|
||
}
|
||
|
||
const drainResult = await waitForDrainOrExit(stdin);
|
||
if (drainResult !== "drain" || exitStatus !== "running") {
|
||
return false;
|
||
}
|
||
resetTimer();
|
||
return true;
|
||
},
|
||
|
||
close: async (): Promise<StreamingEncoderResult> => {
|
||
// INVARIANT: close() is idempotent. The renderOrchestrator HDR cleanup
|
||
// path tracks an `encoderClosed` flag and may still re-call close() in
|
||
// the outer finally if the inner cleanup raised before the flag flipped.
|
||
// Each step here must be safe to repeat:
|
||
// - clearTimeout: safe to call on an already-cleared/fired timer
|
||
// - removeEventListener: no-op if the listener was already removed
|
||
// (and {once: true} would have removed it on the first abort anyway)
|
||
// - stdin.end gated on !destroyed: skipped on the second call
|
||
// - exitPromise: a single shared Promise; awaiting an already-resolved
|
||
// Promise resolves immediately with the same captured exitCode
|
||
// The returned StreamingEncoderResult is therefore consistent across
|
||
// repeated calls. If you change this method, preserve idempotency or
|
||
// a regression here will silently double-close ffmpeg and produce
|
||
// harder-to-trace errors at the orchestrator layer.
|
||
if (timer) {
|
||
clearTimeout(timer);
|
||
timer = null;
|
||
}
|
||
if (signal) signal.removeEventListener("abort", onAbort);
|
||
|
||
const stdin = ffmpeg.stdin;
|
||
if (stdin && !stdin.destroyed) {
|
||
await new Promise<void>((resolve) => {
|
||
stdin.end(() => resolve());
|
||
});
|
||
}
|
||
|
||
await exitPromise;
|
||
|
||
const durationMs = Date.now() - startTime;
|
||
|
||
if (signal?.aborted) {
|
||
return {
|
||
success: false,
|
||
durationMs,
|
||
fileSize: 0,
|
||
error: "Streaming encode cancelled",
|
||
};
|
||
}
|
||
|
||
if (exitCode !== 0) {
|
||
return {
|
||
success: false,
|
||
durationMs,
|
||
fileSize: 0,
|
||
error: formatFfmpegError(exitCode, stderr),
|
||
};
|
||
}
|
||
|
||
const fileSize = existsSync(outputPath) ? statSync(outputPath).size : 0;
|
||
|
||
return { success: true, durationMs, fileSize };
|
||
},
|
||
|
||
getExitStatus: () => exitStatus,
|
||
|
||
getExitError: () => {
|
||
if (exitStatus !== "error") return undefined;
|
||
return formatFfmpegError(exitCode, stderr);
|
||
},
|
||
};
|
||
|
||
return encoder;
|
||
}
|