refactor(producer): extract captureStreamingStage (single-machine fusion)

Move the streaming encode fusion path (`useStreamingEncode === true` with
successful encoder spawn) out of `executeRenderJob` into
`services/render/stages/captureStreamingStage.ts`. The stage owns:

- `spawnStreamingEncoder` invocation, including the abort-rethrow vs.
  graceful-fallback handling.
- Parallel + sequential capture-to-stdin loops (Stage 4 absorbs Stage 5
  for streaming renders).
- The streaming encoder's `close()` + result check.
- Defensive cleanup of the streaming encoder in the stage's own
  `try/finally`.

The stage returns either `{ success: true, ... }` (sequencer skips the
disk path AND inline Stage 5) or `{ success: false }` (sequencer falls
back to the disk path). The sequencer's `useStreamingEncode` flag is
no longer flipped imperatively — the result type makes the branch
selection explicit.

Hard constraints preserved verbatim:
- `probeSession` is closed at the same code points (parallel: after
  capture; sequential: in session finally). The local binding nulls
  via the returned result.
- `lastBrowserConsole` is set to the buffer of whichever session was
  active last (probe close path or sequential session finally).
- `job.framesRendered` is updated per-frame; `Streaming frame N/M
  [(K workers)]` `updateJobStatus` payloads fire at the same 30-frame
  and completion checkpoints (parallel) or every frame (sequential),
  with the same percentage math `25 + frameProgress * 55`.
- `Streaming encode failed: <err>` still throws on the encoder's
  `success: false` close result.
- The defensive `try/finally` close-on-throw is preserved, now inside
  the stage instead of the orchestrator.
- `perfStages.captureMs` is still set by the sequencer from
  `stage4Start`; the stage also returns `encodeMs` for the encoder's
  overlapped duration (assigned to `perfStages.encodeMs`).

Removes the orphaned `createFrameReorderBuffer` and
`prepareCaptureSessionForReuse` imports from the orchestrator after
the streaming code moved.

Verified inside `Dockerfile.test`:
- 5/5 fixtures PASS (font-variant-numeric, many-cuts, variables-prod,
  sub-composition-video, gsap-letters-render-compat).
- `gsap-letters-render-compat` (single-worker render, 4s duration)
  exercises the new streaming stage end-to-end —
  `streaming-encode gate enabled=true` confirmed in the log.
- The other 4 fixtures exercise the disk path (workerCount > 1).

Known follow-up: same runtime import cycle situation as captureStage —
the stage imports `updateJobStatus` and types from
`renderOrchestrator.ts`, which imports the stage back. Safe (deferred
to runtime); a future PR will flatten this once all 8 stages are
extracted.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
James
2026-05-12 01:15:03 +00:00
co-authored by Claude Opus 4.7
parent d39df2de6c
commit ba90b411ed
2 changed files with 439 additions and 262 deletions
@@ -0,0 +1,305 @@
/**
* captureStreamingStage — single-machine fused capture + encode path.
*
* Streaming mode pipes captured frame buffers directly into ffmpeg's stdin
* via `spawnStreamingEncoder`, skipping disk writes and the separate
* Stage 5 encode step. In effect, Stage 4 (capture) absorbs Stage 5
* (encode) for renders that fit the single-machine fusion path.
*
* The streaming path is gated by `shouldUseStreamingEncode(...)` upstream:
* - Disabled when output is png-sequence (no encoder).
* - Disabled for parallel renders auto-selected by calibration where the
* ordered streaming writer would stall later workers behind earlier
* ranges (the orchestrator decides this; the stage is told via input).
* - Disabled in distributed mode (which writes chunks to disk).
*
* If `spawnStreamingEncoder` fails for any non-abort reason, the stage
* returns `{ success: false }` and the sequencer falls back to the disk
* capture path. This mirrors the original orchestrator's flag-flip
* (`useStreamingEncode = false`).
*
* Hard constraints preserved verbatim from the in-process renderer:
* - `probeSession` is closed when the parallel path takes over, OR in
* the sequential session's `finally`. Either way the local binding
* is nulled and the result returns the updated value.
* - `lastBrowserConsole` is set to the buffer of whichever session
* was active last (probe close path, or sequential session finally).
* - `job.framesRendered` is updated per-frame; `Streaming frame N/M`
* `updateJobStatus` payloads fire at the same 30-frame and
* completion checkpoints (parallel) or every frame (sequential).
* - Encoder close + result inspection happens inside the stage; a
* `Streaming encode failed: ...` error throws on `success: false`.
* - Defensive cleanup of `streamingEncoder` happens in the stage's
* own `finally` regardless of success/failure, gated on
* `streamingEncoderClosed` so it's idempotent.
*
* Known follow-up (same as captureStage): this stage imports
* `updateJobStatus` from `renderOrchestrator.ts`, re-introducing the
* cycle PR 1.3.5 broke. A subsequent PR will consolidate capture
* helpers into a shared module.
*/
import {
type BeforeCaptureHook,
type CaptureOptions,
type CaptureSession,
type EngineConfig,
type StreamingEncoder,
captureFrameToBuffer,
closeCaptureSession,
createCaptureSession,
createFrameReorderBuffer,
distributeFrames,
executeParallelCapture,
initializeSession,
prepareCaptureSessionForReuse,
spawnStreamingEncoder,
} from "@hyperframes/engine";
import type { FileServerHandle } from "../../fileServer.js";
import type { ProducerLogger } from "../../../logger.js";
import {
updateJobStatus,
type ProgressCallback,
type RenderJob,
} from "../../renderOrchestrator.js";
/**
* Pre-built ffmpeg streaming-encoder options, exactly matching the
* second argument to `spawnStreamingEncoder`. The sequencer constructs
* this from its in-scope preset / dimensions / quality fields and
* passes it through so the stage doesn't have to reach back for the
* preset's internal shape.
*/
export type StreamingEncoderOptions = Parameters<typeof spawnStreamingEncoder>[1];
export interface CaptureStreamingStageInput {
fileServer: FileServerHandle;
workDir: string;
framesDir: string;
videoOnlyPath: string;
job: RenderJob;
/**
* `job.totalFrames` is `number | undefined` in the public type — the
* sequencer narrows it via the probeStage result before calling here.
*/
totalFrames: number;
cfg: EngineConfig;
log: ProducerLogger;
workerCount: number;
probeSession: CaptureSession | null;
/** For the spawn-failure log message context only. */
outputFormat: string;
/** Pre-built encoder options; passed straight to `spawnStreamingEncoder`. */
streamingEncoderOptions: StreamingEncoderOptions;
buildCaptureOptions: () => CaptureOptions;
createRenderVideoFrameInjector: () => BeforeCaptureHook | null;
abortSignal: AbortSignal | undefined;
assertNotAborted: () => void;
onProgress?: ProgressCallback;
}
export type CaptureStreamingStageResult =
| {
/** Streaming path ran successfully — sequencer should skip the disk path AND Stage 5 encode. */
success: true;
/** Wall-clock ms for the capture phase (`Date.now() - stage4Start` is the sequencer's job). */
captureDurationMs: number;
/** Wall-clock ms for the encode phase (overlapped with capture; from the encoder's own report). */
encodeMs: number;
probeSession: CaptureSession | null;
lastBrowserConsole: string[];
workerCount: number;
}
| {
/** Spawn failed (non-abort) — sequencer should fall back to the disk path. */
success: false;
};
export async function runCaptureStreamingStage(
input: CaptureStreamingStageInput,
): Promise<CaptureStreamingStageResult> {
const {
fileServer,
workDir,
framesDir,
videoOnlyPath,
job,
totalFrames,
cfg,
log,
outputFormat,
streamingEncoderOptions,
buildCaptureOptions,
createRenderVideoFrameInjector,
abortSignal,
assertNotAborted,
onProgress,
} = input;
let { workerCount, probeSession } = input;
let lastBrowserConsole: string[] = [];
let streamingEncoder: StreamingEncoder | null = null;
let streamingEncoderClosed = false;
try {
streamingEncoder = await spawnStreamingEncoder(
videoOnlyPath,
streamingEncoderOptions,
abortSignal,
);
assertNotAborted();
} catch (err) {
if (abortSignal?.aborted) {
if (streamingEncoder && !streamingEncoderClosed) {
await (streamingEncoder as StreamingEncoder).close().catch(() => {});
streamingEncoderClosed = true;
}
throw err;
}
log.warn("[Render] Streaming encoder spawn failed; falling back to disk-frame encode.", {
error: err instanceof Error ? err.message : String(err),
outputFormat,
workerCount,
durationSeconds: job.duration,
});
return { success: false };
}
const streamStart = Date.now();
const currentEncoder: StreamingEncoder = streamingEncoder;
try {
// ── Streaming capture + encode (Stage 4 absorbs Stage 5) ──────────
// Streaming encode is locked in here; capture retries may shrink
// workerCount later, but must not grow a streaming render past one worker.
const reorderBuffer = createFrameReorderBuffer(0, totalFrames);
if (workerCount > 1) {
// Parallel capture → streaming encode
const tasks = distributeFrames(totalFrames, workerCount, workDir);
const onFrameBuffer = async (frameIndex: number, buffer: Buffer): Promise<void> => {
await reorderBuffer.waitForFrame(frameIndex);
currentEncoder.writeFrame(buffer);
reorderBuffer.advanceTo(frameIndex + 1);
};
await executeParallelCapture(
fileServer.url,
workDir,
tasks,
buildCaptureOptions(),
createRenderVideoFrameInjector,
abortSignal,
(progress) => {
job.framesRendered = progress.capturedFrames;
const frameProgress = progress.capturedFrames / progress.totalFrames;
const progressPct = 25 + frameProgress * 55;
if (
progress.capturedFrames % 30 === 0 ||
progress.capturedFrames === progress.totalFrames
) {
updateJobStatus(
job,
"rendering",
`Streaming frame ${progress.capturedFrames}/${progress.totalFrames} (${workerCount} workers)`,
Math.round(progressPct),
onProgress,
);
}
},
onFrameBuffer,
cfg,
);
if (probeSession) {
lastBrowserConsole = probeSession.browserConsoleBuffer;
await closeCaptureSession(probeSession);
probeSession = null;
}
} else {
// Sequential capture → streaming encode
const videoInjector = createRenderVideoFrameInjector();
const session =
probeSession ??
(await createCaptureSession(
fileServer.url,
framesDir,
buildCaptureOptions(),
videoInjector,
cfg,
));
if (probeSession) {
prepareCaptureSessionForReuse(session, framesDir, videoInjector);
probeSession = null;
}
try {
if (!session.isInitialized) {
await initializeSession(session);
}
assertNotAborted();
lastBrowserConsole = session.browserConsoleBuffer;
for (let i = 0; i < totalFrames; i++) {
assertNotAborted();
const time = (i * job.config.fps.den) / job.config.fps.num;
const { buffer } = await captureFrameToBuffer(session, i, time);
await reorderBuffer.waitForFrame(i);
currentEncoder.writeFrame(buffer);
reorderBuffer.advanceTo(i + 1);
job.framesRendered = i + 1;
const frameProgress = (i + 1) / totalFrames;
const progress = 25 + frameProgress * 55;
updateJobStatus(
job,
"rendering",
`Streaming frame ${i + 1}/${totalFrames}`,
Math.round(progress),
onProgress,
);
}
} finally {
lastBrowserConsole = session.browserConsoleBuffer;
await closeCaptureSession(session);
}
}
// Close encoder and get result
const encodeResult = await currentEncoder.close();
streamingEncoderClosed = true;
assertNotAborted();
if (!encodeResult.success) {
throw new Error(`Streaming encode failed: ${encodeResult.error}`);
}
return {
success: true,
captureDurationMs: Date.now() - streamStart,
encodeMs: encodeResult.durationMs,
probeSession,
lastBrowserConsole,
workerCount,
};
} finally {
// Defensive cleanup: if the streaming branch threw before
// currentEncoder.close() (e.g. capture failure, abort, broken pipe),
// the ffmpeg subprocess would otherwise leak. close() is idempotent so
// this is safe to call alongside the success-path close — we just gate
// on the flag to avoid redundant work.
if (streamingEncoder && !streamingEncoderClosed) {
try {
await streamingEncoder.close();
} catch (err) {
log.warn("streamingEncoder defensive close failed", {
err: err instanceof Error ? err.message : String(err),
});
}
}
}
}
@@ -41,7 +41,6 @@ import {
initializeSession, initializeSession,
closeCaptureSession, closeCaptureSession,
captureFrameToBuffer, captureFrameToBuffer,
prepareCaptureSessionForReuse,
type CaptureOptions, type CaptureOptions,
type CaptureVideoMetadataHint, type CaptureVideoMetadataHint,
type CaptureSession, type CaptureSession,
@@ -59,7 +58,6 @@ import {
type ParallelProgress, type ParallelProgress,
type WorkerTask, type WorkerTask,
spawnStreamingEncoder, spawnStreamingEncoder,
createFrameReorderBuffer,
type StreamingEncoder, type StreamingEncoder,
analyzeCompositionHdr, analyzeCompositionHdr,
runFfmpeg, runFfmpeg,
@@ -101,6 +99,7 @@ import { runProbeStage } from "./render/stages/probeStage.js";
import { runExtractVideosStage } from "./render/stages/extractVideosStage.js"; import { runExtractVideosStage } from "./render/stages/extractVideosStage.js";
import { runAudioStage } from "./render/stages/audioStage.js"; import { runAudioStage } from "./render/stages/audioStage.js";
import { runCaptureStage } from "./render/stages/captureStage.js"; import { runCaptureStage } from "./render/stages/captureStage.js";
import { runCaptureStreamingStage } from "./render/stages/captureStreamingStage.js";
/** /**
* Wrap a cleanup operation so it never throws, but logs any failure. * Wrap a cleanup operation so it never throws, but logs any failure.
@@ -3081,281 +3080,154 @@ export async function executeRenderJob(
} }
hdrVideoFrameSources.clear(); hdrVideoFrameSources.clear();
} }
} else // ── Standard capture paths (SDR or DOM-only HDR) ────────────────── } else {
// Streaming encode mode: pipe frame buffers directly to FFmpeg stdin, // ── Standard capture paths (SDR or DOM-only HDR) ──────────────────
// skipping disk writes and the separate Stage 5 encode step. // Streaming encode mode pipes frame buffers directly to FFmpeg stdin,
{ // skipping disk writes and the separate Stage 5 encode step. If the
let streamingEncoder: StreamingEncoder | null = null; // streaming spawn fails (non-abort) the stage returns { success: false }
let streamingEncoderClosed = false; // and we fall back to the disk path below.
let streamingHandled = false;
if (useStreamingEncode) { if (useStreamingEncode) {
try { const streamingRes = await runCaptureStreamingStage({
streamingEncoder = await spawnStreamingEncoder( fileServer,
videoOnlyPath, workDir,
{ framesDir,
fps: job.config.fps, videoOnlyPath,
width, job,
height, totalFrames,
codec: preset.codec, cfg,
preset: preset.preset, log,
quality: effectiveQuality, workerCount,
bitrate: effectiveBitrate, probeSession,
pixelFormat: preset.pixelFormat, outputFormat,
useGpu: job.config.useGpu, streamingEncoderOptions: {
imageFormat: captureOptions.format || "jpeg", fps: job.config.fps,
hdr: preset.hdr, width,
}, height,
abortSignal, codec: preset.codec,
); preset: preset.preset,
assertNotAborted(); quality: effectiveQuality,
} catch (err) { bitrate: effectiveBitrate,
if (abortSignal?.aborted) { pixelFormat: preset.pixelFormat,
if (streamingEncoder && !streamingEncoderClosed) { useGpu: job.config.useGpu,
await streamingEncoder.close().catch(() => {}); imageFormat: captureOptions.format || "jpeg",
streamingEncoderClosed = true; hdr: preset.hdr,
} },
throw err; buildCaptureOptions,
} createRenderVideoFrameInjector,
abortSignal,
assertNotAborted,
onProgress,
});
if (streamingRes.success) {
streamingHandled = true;
workerCount = streamingRes.workerCount;
probeSession = streamingRes.probeSession;
lastBrowserConsole = streamingRes.lastBrowserConsole;
perfStages.captureMs = Date.now() - stage4Start;
perfStages.encodeMs = streamingRes.encodeMs; // Overlapped with capture
} else {
useStreamingEncode = false; useStreamingEncode = false;
streamingEncoder = null;
log.warn("[Render] Streaming encoder spawn failed; falling back to disk-frame encode.", {
error: err instanceof Error ? err.message : String(err),
outputFormat,
workerCount,
durationSeconds: job.duration,
});
} }
} }
try { if (!streamingHandled) {
if (useStreamingEncode && streamingEncoder) { // ── Disk-based capture (original flow) ────────────────────────────
// ── Streaming capture + encode (Stage 4 absorbs Stage 5) ────────── const captureRes = await runCaptureStage({
// Streaming encode is locked in here; capture retries may shrink fileServer,
// workerCount later, but must not grow a streaming render past one worker. workDir,
const reorderBuffer = createFrameReorderBuffer(0, totalFrames); framesDir,
const currentEncoder = streamingEncoder; job,
totalFrames,
cfg,
log,
workerCount,
probeSession,
needsAlpha,
captureAttempts,
buildCaptureOptions,
createRenderVideoFrameInjector,
abortSignal,
assertNotAborted,
onProgress,
});
workerCount = captureRes.workerCount;
probeSession = captureRes.probeSession;
lastBrowserConsole = captureRes.lastBrowserConsole;
if (workerCount > 1) { perfStages.captureMs = Date.now() - stage4Start;
// Parallel capture → streaming encode
const tasks = distributeFrames(job.totalFrames, workerCount, workDir);
const onFrameBuffer = async (frameIndex: number, buffer: Buffer): Promise<void> => { if (isPngSequence) {
await reorderBuffer.waitForFrame(frameIndex); // ── Stage 5 (png-sequence): copy captured PNGs to outputDir ──────
currentEncoder.writeFrame(buffer); // No encoder, no mux, no faststart — captured frames already carry
reorderBuffer.advanceTo(frameIndex + 1); // alpha and are the deliverable. We rename to `frame_NNNNNN.png`
}; // (zero-padded) so consumers (After Effects, Nuke, Fusion, ffmpeg
// image2 demuxer) can globbed-import without surprises.
await executeParallelCapture( const stage5Start = Date.now();
fileServer.url, updateJobStatus(job, "encoding", "Writing PNG sequence", 75, onProgress);
workDir, if (!existsSync(outputPath)) mkdirSync(outputPath, { recursive: true });
tasks, const captured = readdirSync(framesDir)
buildCaptureOptions(), .filter((name) => name.endsWith(".png"))
createRenderVideoFrameInjector, .sort();
abortSignal, if (captured.length === 0) {
(progress) => { throw new Error(
job.framesRendered = progress.capturedFrames; `[Render] png-sequence output requested but no PNGs were captured to ${framesDir}`,
const frameProgress = progress.capturedFrames / progress.totalFrames;
const progressPct = 25 + frameProgress * 55;
if (
progress.capturedFrames % 30 === 0 ||
progress.capturedFrames === progress.totalFrames
) {
updateJobStatus(
job,
"rendering",
`Streaming frame ${progress.capturedFrames}/${progress.totalFrames} (${workerCount} workers)`,
Math.round(progressPct),
onProgress,
);
}
},
onFrameBuffer,
cfg,
); );
if (probeSession) {
lastBrowserConsole = probeSession.browserConsoleBuffer;
await closeCaptureSession(probeSession);
probeSession = null;
}
} else {
// Sequential capture → streaming encode
const videoInjector = createRenderVideoFrameInjector();
const session =
probeSession ??
(await createCaptureSession(
fileServer.url,
framesDir,
buildCaptureOptions(),
videoInjector,
cfg,
));
if (probeSession) {
prepareCaptureSessionForReuse(session, framesDir, videoInjector);
probeSession = null;
}
try {
if (!session.isInitialized) {
await initializeSession(session);
}
assertNotAborted();
lastBrowserConsole = session.browserConsoleBuffer;
for (let i = 0; i < totalFrames; i++) {
assertNotAborted();
const time = (i * job.config.fps.den) / job.config.fps.num;
const { buffer } = await captureFrameToBuffer(session, i, time);
await reorderBuffer.waitForFrame(i);
currentEncoder.writeFrame(buffer);
reorderBuffer.advanceTo(i + 1);
job.framesRendered = i + 1;
const frameProgress = (i + 1) / totalFrames;
const progress = 25 + frameProgress * 55;
updateJobStatus(
job,
"rendering",
`Streaming frame ${i + 1}/${job.totalFrames}`,
Math.round(progress),
onProgress,
);
}
} finally {
lastBrowserConsole = session.browserConsoleBuffer;
await closeCaptureSession(session);
}
} }
captured.forEach((name, i) => {
const dst = join(outputPath, `frame_${String(i + 1).padStart(6, "0")}.png`);
copyFileSync(join(framesDir, name), dst);
});
if (hasAudio && existsSync(audioOutputPath)) {
// Sidecar audio for callers that need to re-mux later. png-sequence
// has no container of its own, so this is the only place audio
// can land alongside the frames.
copyFileSync(audioOutputPath, join(outputPath, "audio.aac"));
log.info(`[Render] png-sequence: audio.aac sidecar written to ${outputPath}/audio.aac`);
}
perfStages.encodeMs = Date.now() - stage5Start;
} else {
// ── Stage 5: Encode ───────────────────────────────────────────────
const stage5Start = Date.now();
updateJobStatus(job, "encoding", "Encoding video", 75, onProgress);
// Close encoder and get result const frameExt = needsAlpha ? "png" : "jpg";
const encodeResult = await currentEncoder.close(); const framePattern = `frame_%06d.${frameExt}`;
streamingEncoderClosed = true; const encoderOpts = {
fps: job.config.fps,
width,
height,
codec: preset.codec,
preset: preset.preset,
quality: effectiveQuality,
bitrate: effectiveBitrate,
pixelFormat: preset.pixelFormat,
useGpu: job.config.useGpu,
hdr: preset.hdr,
};
const encodeResult = enableChunkedEncode
? await encodeFramesChunkedConcat(
framesDir,
framePattern,
videoOnlyPath,
encoderOpts,
chunkedEncodeSize,
abortSignal,
)
: await encodeFramesFromDir(
framesDir,
framePattern,
videoOnlyPath,
encoderOpts,
abortSignal,
);
assertNotAborted(); assertNotAborted();
if (!encodeResult.success) { if (!encodeResult.success) {
throw new Error(`Streaming encode failed: ${encodeResult.error}`); throw new Error(`Encoding failed: ${encodeResult.error}`);
} }
perfStages.captureMs = Date.now() - stage4Start; perfStages.encodeMs = Date.now() - stage5Start;
perfStages.encodeMs = encodeResult.durationMs; // Overlapped with capture
} else {
// ── Disk-based capture (original flow) ────────────────────────────
const captureRes = await runCaptureStage({
fileServer,
workDir,
framesDir,
job,
totalFrames,
cfg,
log,
workerCount,
probeSession,
needsAlpha,
captureAttempts,
buildCaptureOptions,
createRenderVideoFrameInjector,
abortSignal,
assertNotAborted,
onProgress,
});
workerCount = captureRes.workerCount;
probeSession = captureRes.probeSession;
lastBrowserConsole = captureRes.lastBrowserConsole;
perfStages.captureMs = Date.now() - stage4Start;
if (isPngSequence) {
// ── Stage 5 (png-sequence): copy captured PNGs to outputDir ──────
// No encoder, no mux, no faststart — captured frames already carry
// alpha and are the deliverable. We rename to `frame_NNNNNN.png`
// (zero-padded) so consumers (After Effects, Nuke, Fusion, ffmpeg
// image2 demuxer) can globbed-import without surprises.
const stage5Start = Date.now();
updateJobStatus(job, "encoding", "Writing PNG sequence", 75, onProgress);
if (!existsSync(outputPath)) mkdirSync(outputPath, { recursive: true });
const captured = readdirSync(framesDir)
.filter((name) => name.endsWith(".png"))
.sort();
if (captured.length === 0) {
throw new Error(
`[Render] png-sequence output requested but no PNGs were captured to ${framesDir}`,
);
}
captured.forEach((name, i) => {
const dst = join(outputPath, `frame_${String(i + 1).padStart(6, "0")}.png`);
copyFileSync(join(framesDir, name), dst);
});
if (hasAudio && existsSync(audioOutputPath)) {
// Sidecar audio for callers that need to re-mux later. png-sequence
// has no container of its own, so this is the only place audio
// can land alongside the frames.
copyFileSync(audioOutputPath, join(outputPath, "audio.aac"));
log.info(
`[Render] png-sequence: audio.aac sidecar written to ${outputPath}/audio.aac`,
);
}
perfStages.encodeMs = Date.now() - stage5Start;
} else {
// ── Stage 5: Encode ───────────────────────────────────────────────
const stage5Start = Date.now();
updateJobStatus(job, "encoding", "Encoding video", 75, onProgress);
const frameExt = needsAlpha ? "png" : "jpg";
const framePattern = `frame_%06d.${frameExt}`;
const encoderOpts = {
fps: job.config.fps,
width,
height,
codec: preset.codec,
preset: preset.preset,
quality: effectiveQuality,
bitrate: effectiveBitrate,
pixelFormat: preset.pixelFormat,
useGpu: job.config.useGpu,
hdr: preset.hdr,
};
const encodeResult = enableChunkedEncode
? await encodeFramesChunkedConcat(
framesDir,
framePattern,
videoOnlyPath,
encoderOpts,
chunkedEncodeSize,
abortSignal,
)
: await encodeFramesFromDir(
framesDir,
framePattern,
videoOnlyPath,
encoderOpts,
abortSignal,
);
assertNotAborted();
if (!encodeResult.success) {
throw new Error(`Encoding failed: ${encodeResult.error}`);
}
perfStages.encodeMs = Date.now() - stage5Start;
}
}
} finally {
// Defensive cleanup: if the streaming encoder branch threw before
// currentEncoder.close() (e.g. capture failure, abort, broken pipe),
// the ffmpeg subprocess would otherwise leak. close() is idempotent so
// this is safe to call alongside the success-path close — we just gate
// on the flag to avoid redundant work.
if (streamingEncoder && !streamingEncoderClosed) {
try {
await streamingEncoder.close();
} catch (err) {
log.warn("streamingEncoder defensive close failed", {
err: err instanceof Error ? err.message : String(err),
});
}
} }
} }
} // end SDR capture paths block } // end SDR capture paths block