refactor(producer): extract captureStage (SDR disk path)

Move the SDR / DOM-only-HDR disk-capture body out of `executeRenderJob`
into `services/render/stages/captureStage.ts`. Covers both branches of
the disk path: parallel capture via `executeDiskCaptureWithAdaptiveRetry`
(`workerCount > 1`) and sequential per-process capture (`workerCount === 1`,
reusing `probeSession` when available).

The HDR layered branch (`useLayeredComposite === true`) and the streaming
encode fusion path (`useStreamingEncode === true` with successful encoder
spawn) stay inline in the sequencer — they will be extracted by the next
two PRs in the stack.

Hard constraints preserved verbatim:
- `probeSession` is closed (and the sequencer's `let probeSession`
  nulled via the returned result) at the same points.
- `captureAttempts` is mutated in place — the parallel retry loop still
  pushes each attempt onto the array the sequencer owns.
- `workerCount` reassignment from adaptive retry survives via the
  returned result.
- `lastBrowserConsole` is set to the buffer of whichever session was
  active last (probe close path or sequential capture finally).
- `job.framesRendered` is updated at the same per-frame / per-progress
  points; `Capturing frame N/M [(K workers)]` `updateJobStatus` payloads
  fire at the same 30-frame and completion checkpoints.
- `perfStages.captureMs` is still computed by the sequencer from the
  outer `stage4Start` so its window covers both the in-sequencer setup
  (fileServer init, calibration, worker resolution, preset selection)
  AND the capture call.

Two small new exports on `renderOrchestrator.ts`:
- `executeDiskCaptureWithAdaptiveRetry` — was a private helper; the
  stage calls it directly.
- `updateJobStatus` — was a private helper; the stage uses it for the
  per-frame progress callbacks so the `completedAt` branch matches.

These re-introduce a small runtime cycle between the stage and the
orchestrator (orchestrator imports `runCaptureStage`; stage imports
helpers back). The cycle is safe (both modules finish loading before
any stage function is invoked at runtime) and will be flattened in a
follow-up PR that consolidates capture helpers into a shared module.
Removes the now-orphaned `captureFrame` import from the orchestrator.

Verified inside `Dockerfile.test`:
- `font-variant-numeric`: audio correlation 1.000
- `many-cuts`: 0 failed frames, audio correlation 0.994
- `variables-prod`: PSNR ~69 dB, audio correlation 0.975
- `sub-composition-video`: PSNR ~43-52 dB, audio correlation 0.947
  (exercises video extraction + capture end-to-end)
- `gsap-letters-render-compat`: PSNR ~53-55 dB, audio correlation 1.000
  (exercises the parallel capture path; 5/5 PASS overall)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
James
2026-05-12 01:05:31 +00:00
co-authored by Claude Opus 4.7
parent aec29fe599
commit d39df2de6c
2 changed files with 235 additions and 94 deletions
@@ -0,0 +1,211 @@
/**
* captureStage — SDR disk-capture path of `executeRenderJob`.
*
* Handles both branches of the SDR / DOM-only-HDR disk-capture flow:
* - `workerCount > 1`: parallel capture with adaptive retry via
* `executeDiskCaptureWithAdaptiveRetry`.
* - `workerCount === 1`: sequential capture in the orchestrator process,
* reusing `probeSession` when available.
*
* The HDR layered branch (`useLayeredComposite === true`) and the streaming
* encode fusion path (`useStreamingEncode === true` with successful encoder
* spawn) live in separate stages.
*
* Hard constraints preserved verbatim:
* - `probeSession` is closed (and the local binding nulled) once the
* stage no longer needs it. The sequencer's `let probeSession` is
* updated via the returned result.
* - `captureAttempts` is mutated in place — the parallel path appends
* each retry attempt to the array the sequencer owns.
* - `workerCount` may be reduced by an adaptive retry; the returned
* value reflects the final worker count for the perf summary.
* - `lastBrowserConsole` is set to the buffer of whichever session was
* active last (probe session in the parallel close path; sequential
* session in the sequential path).
* - `job.framesRendered` is updated at the same per-frame / per-progress
* points; the same `Capturing frame N/M` `updateJobStatus` payloads
* fire at 30-frame and completion checkpoints (parallel) or every
* frame (sequential).
*
* Known follow-up: this stage imports `executeDiskCaptureWithAdaptiveRetry`
* from `renderOrchestrator.ts`, which itself imports the stage — a runtime
* cycle that resolves at module-init time because no stage function is
* invoked during load. A subsequent PR will consolidate the capture
* helpers (`executeDiskCaptureWithAdaptiveRetry`, `countFrameRanges`,
* `safeCleanup`, `sampleDirectoryBytes`, etc.) into a shared module so
* the stages can import them without reaching back into the orchestrator.
*/
import {
type BeforeCaptureHook,
type CaptureOptions,
type CaptureSession,
type EngineConfig,
captureFrame,
closeCaptureSession,
createCaptureSession,
initializeSession,
prepareCaptureSessionForReuse,
} from "@hyperframes/engine";
import type { FileServerHandle } from "../../fileServer.js";
import type { ProducerLogger } from "../../../logger.js";
import {
executeDiskCaptureWithAdaptiveRetry,
updateJobStatus,
type CaptureAttemptSummary,
type ProgressCallback,
type RenderJob,
} from "../../renderOrchestrator.js";
export interface CaptureStageInput {
fileServer: FileServerHandle;
workDir: string;
framesDir: string;
job: RenderJob;
/**
* `job.totalFrames` is `number | undefined` in the public type — the
* sequencer narrows it to a `number` via the probeStage result before
* calling this stage. Passed in explicitly here so the stage doesn't
* have to re-narrow on every reference.
*/
totalFrames: number;
cfg: EngineConfig;
log: ProducerLogger;
/** Initial worker count from `resolveRenderWorkerCount`; adaptive retry may reduce it. */
workerCount: number;
/** Reused for the sequential path's first session if non-null. */
probeSession: CaptureSession | null;
/** True for webm / mov / png-sequence (controls capture format + extension). */
needsAlpha: boolean;
/** Mutated in place — each parallel retry attempt is appended. */
captureAttempts: CaptureAttemptSummary[];
buildCaptureOptions: () => CaptureOptions;
createRenderVideoFrameInjector: () => BeforeCaptureHook | null;
abortSignal: AbortSignal | undefined;
assertNotAborted: () => void;
onProgress?: ProgressCallback;
}
export interface CaptureStageResult {
/** Final worker count after any adaptive retry. */
workerCount: number;
/** Always `null` after the stage — the probe session is closed before the stage returns. */
probeSession: CaptureSession | null;
/** Browser console buffer from whichever session was active last. */
lastBrowserConsole: string[];
}
export async function runCaptureStage(input: CaptureStageInput): Promise<CaptureStageResult> {
const {
fileServer,
workDir,
framesDir,
job,
totalFrames,
cfg,
log,
captureAttempts,
buildCaptureOptions,
createRenderVideoFrameInjector,
abortSignal,
assertNotAborted,
onProgress,
needsAlpha,
} = input;
let { workerCount, probeSession } = input;
let lastBrowserConsole: string[] = [];
if (workerCount > 1) {
// Parallel capture
const attempts = await executeDiskCaptureWithAdaptiveRetry({
serverUrl: fileServer.url,
workDir,
framesDir,
totalFrames,
initialWorkerCount: workerCount,
allowRetry: job.config.workers === undefined,
frameExt: needsAlpha ? "png" : "jpg",
captureOptions: buildCaptureOptions(),
createBeforeCaptureHook: createRenderVideoFrameInjector,
abortSignal,
onProgress: (progress) => {
job.framesRendered = progress.capturedFrames;
const frameProgress = progress.capturedFrames / progress.totalFrames;
const progressPct = 25 + frameProgress * 45;
if (
progress.capturedFrames % 30 === 0 ||
progress.capturedFrames === progress.totalFrames
) {
updateJobStatus(
job,
"rendering",
`Capturing frame ${progress.capturedFrames}/${progress.totalFrames} (${progress.activeWorkers} workers)`,
Math.round(progressPct),
onProgress,
);
}
},
cfg,
log,
});
captureAttempts.push(...attempts);
const lastAttempt = attempts[attempts.length - 1];
if (lastAttempt) {
workerCount = lastAttempt.workers;
}
if (probeSession) {
lastBrowserConsole = probeSession.browserConsoleBuffer;
await closeCaptureSession(probeSession);
probeSession = null;
}
} else {
// Sequential capture
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;
await captureFrame(session, i, time);
job.framesRendered = i + 1;
const frameProgress = (i + 1) / totalFrames;
const progress = 25 + frameProgress * 45;
updateJobStatus(
job,
"rendering",
`Capturing frame ${i + 1}/${totalFrames}`,
Math.round(progress),
onProgress,
);
}
} finally {
lastBrowserConsole = session.browserConsoleBuffer;
await closeCaptureSession(session);
}
}
return { workerCount, probeSession, lastBrowserConsole };
}
@@ -40,7 +40,6 @@ import {
createCaptureSession,
initializeSession,
closeCaptureSession,
captureFrame,
captureFrameToBuffer,
prepareCaptureSessionForReuse,
type CaptureOptions,
@@ -101,6 +100,7 @@ import { runCompileStage } from "./render/stages/compileStage.js";
import { runProbeStage } from "./render/stages/probeStage.js";
import { runExtractVideosStage } from "./render/stages/extractVideosStage.js";
import { runAudioStage } from "./render/stages/audioStage.js";
import { runCaptureStage } from "./render/stages/captureStage.js";
/**
* Wrap a cleanup operation so it never throws, but logs any failure.
@@ -540,7 +540,7 @@ export class RenderCancelledError extends Error {
}
}
function updateJobStatus(
export function updateJobStatus(
job: RenderJob,
status: RenderStatus,
stage: string,
@@ -1013,7 +1013,7 @@ function createFailedCaptureCalibrationEstimate(reason: string): {
};
}
async function executeDiskCaptureWithAdaptiveRetry(options: {
export async function executeDiskCaptureWithAdaptiveRetry(options: {
serverUrl: string;
workDir: string;
framesDir: string;
@@ -3243,97 +3243,27 @@ export async function executeRenderJob(
perfStages.encodeMs = encodeResult.durationMs; // Overlapped with capture
} else {
// ── Disk-based capture (original flow) ────────────────────────────
if (workerCount > 1) {
// Parallel capture
const attempts = await executeDiskCaptureWithAdaptiveRetry({
serverUrl: fileServer.url,
workDir,
framesDir,
totalFrames: job.totalFrames,
initialWorkerCount: workerCount,
allowRetry: job.config.workers === undefined,
frameExt: needsAlpha ? "png" : "jpg",
captureOptions: buildCaptureOptions(),
createBeforeCaptureHook: createRenderVideoFrameInjector,
abortSignal,
onProgress: (progress) => {
job.framesRendered = progress.capturedFrames;
const frameProgress = progress.capturedFrames / progress.totalFrames;
const progressPct = 25 + frameProgress * 45;
if (
progress.capturedFrames % 30 === 0 ||
progress.capturedFrames === progress.totalFrames
) {
updateJobStatus(
job,
"rendering",
`Capturing frame ${progress.capturedFrames}/${progress.totalFrames} (${progress.activeWorkers} workers)`,
Math.round(progressPct),
onProgress,
);
}
},
cfg,
log,
});
captureAttempts.push(...attempts);
const lastAttempt = attempts[attempts.length - 1];
if (lastAttempt) {
workerCount = lastAttempt.workers;
}
if (probeSession) {
lastBrowserConsole = probeSession.browserConsoleBuffer;
await closeCaptureSession(probeSession);
probeSession = null;
}
} else {
// Sequential capture
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 < job.totalFrames; i++) {
assertNotAborted();
const time = (i * job.config.fps.den) / job.config.fps.num;
await captureFrame(session, i, time);
job.framesRendered = i + 1;
const frameProgress = (i + 1) / job.totalFrames;
const progress = 25 + frameProgress * 45;
updateJobStatus(
job,
"rendering",
`Capturing frame ${i + 1}/${job.totalFrames}`,
Math.round(progress),
onProgress,
);
}
} finally {
lastBrowserConsole = session.browserConsoleBuffer;
await closeCaptureSession(session);
}
}
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;