From 864beeda19ebf7a92d65df3b156e7485cfb74ef9 Mon Sep 17 00:00:00 2001 From: James Date: Fri, 10 Jul 2026 13:02:31 -0700 Subject: [PATCH] refactor(producer): make capture plans immutable --- .../src/services/distributed/renderChunk.ts | 17 +- .../src/services/render/capturePlan.test.ts | 123 ++++++++++++ .../src/services/render/capturePlan.ts | 150 ++++++++++++++ .../services/render/stages/captureHdrStage.ts | 45 +---- .../services/render/stages/captureStage.ts | 26 +-- .../stages/captureStreamingStage.test.ts | 33 +++- .../render/stages/captureStreamingStage.ts | 31 +-- .../src/services/renderOrchestrator.ts | 186 +++++++++++------- 8 files changed, 457 insertions(+), 154 deletions(-) create mode 100644 packages/producer/src/services/render/capturePlan.test.ts create mode 100644 packages/producer/src/services/render/capturePlan.ts diff --git a/packages/producer/src/services/distributed/renderChunk.ts b/packages/producer/src/services/distributed/renderChunk.ts index a9e60f0c71..4f5009f15b 100644 --- a/packages/producer/src/services/distributed/renderChunk.ts +++ b/packages/producer/src/services/distributed/renderChunk.ts @@ -59,6 +59,7 @@ import { defaultLogger } from "../../logger.js"; import { runEncodeStage } from "../render/stages/encodeStage.js"; import { runCaptureStage } from "../render/stages/captureStage.js"; import { resolveVideoCaptureBeyondViewport } from "../render/captureBeyondViewport.js"; +import { createCapturePlan } from "../render/capturePlan.js"; import { type ChunkSliceJson, type LockedRenderConfig, @@ -589,6 +590,18 @@ export async function renderChunk( // one Chrome session per worker, so captureStageMs includes those // boots; sessionBootMs stays 0 there. const captureStarted = Date.now(); + const capturePlan = createCapturePlan({ + workerCount: chunkWorkerCount, + forceScreenshot: encoder.forceScreenshot, + useStreamingEncode: false, + useLayeredComposite: false, + usePageSideCompositing: false, + hasHdrContent: false, + needsAlpha: plan.dimensions.format !== "mp4", + }); + if (capturePlan.kind !== "sdr_disk") { + throw new Error(`Distributed chunk requires sdr_disk plan; got ${capturePlan.kind}`); + } await runCaptureStage({ fileServer, workDir, @@ -596,11 +609,9 @@ export async function renderChunk( job, totalFrames: framesInChunk, cfg, - forceScreenshot: encoder.forceScreenshot, + plan: capturePlan, log, - workerCount: chunkWorkerCount, probeSession: session, - needsAlpha: plan.dimensions.format !== "mp4", captureAttempts: [], // Distributed chunks run on Linux (beginframe) where dedup never arms; // a throwaway sink satisfies the type without per-chunk dedup reporting. diff --git a/packages/producer/src/services/render/capturePlan.test.ts b/packages/producer/src/services/render/capturePlan.test.ts new file mode 100644 index 0000000000..07e29f7426 --- /dev/null +++ b/packages/producer/src/services/render/capturePlan.test.ts @@ -0,0 +1,123 @@ +import { describe, expect, it } from "vitest"; +import { createCapturePlan, replanAfterFailure, type CaptureRouting } from "./capturePlan.js"; + +function streaming(routing?: CaptureRouting) { + return createCapturePlan({ + workerCount: 1, + forceScreenshot: false, + forceParallelStream: false, + useStreamingEncode: true, + useLayeredComposite: false, + usePageSideCompositing: false, + hasHdrContent: false, + needsAlpha: false, + routing, + }); +} + +describe("CapturePlan", () => { + it("makes layered capture dominant and enforces its screenshot invariant", () => { + const plan = createCapturePlan({ + workerCount: 3, + forceScreenshot: false, + forceParallelStream: true, + useStreamingEncode: true, + useLayeredComposite: true, + usePageSideCompositing: false, + hasHdrContent: true, + needsAlpha: false, + }); + + expect(plan).toMatchObject({ + kind: "hdr_layered", + workerCount: 3, + forceScreenshot: true, + forceParallelStream: false, + }); + expect(Object.isFrozen(plan)).toBe(true); + expect(Object.isFrozen(plan.routing)).toBe(true); + }); + + it("falls back from an unavailable streaming encoder to the same disk route", () => { + const initial = streaming(); + const next = replanAfterFailure(initial, { kind: "streaming_unavailable" }); + + expect(next).toMatchObject({ kind: "sdr_disk", workerCount: 1, forceScreenshot: false }); + expect(initial.kind).toBe("sdr_streaming"); + }); + + it("makes page-side compositing force screenshot capture", () => { + const plan = createCapturePlan({ + workerCount: 1, + forceScreenshot: false, + forceParallelStream: false, + useStreamingEncode: true, + useLayeredComposite: false, + usePageSideCompositing: true, + hasHdrContent: false, + needsAlpha: false, + }); + expect(plan).toMatchObject({ kind: "sdr_streaming", forceScreenshot: true }); + }); + + it("retries an ordinary verification failure in streaming screenshot mode", () => { + expect(replanAfterFailure(streaming(), { kind: "draw_element_verification" })).toMatchObject({ + kind: "sdr_streaming", + workerCount: 1, + forceScreenshot: true, + routing: { kind: "default" }, + }); + }); + + it("atomically restores the pre-inversion disk route after verification failure", () => { + const initial = streaming({ + kind: "worker_inversion", + state: "active", + fallback: { kind: "sdr_disk", workerCount: 5, forceParallelStream: false }, + }); + const next = replanAfterFailure(initial, { kind: "draw_element_verification" }); + + expect(next).toMatchObject({ + kind: "sdr_disk", + workerCount: 5, + forceScreenshot: true, + routing: { kind: "worker_inversion", state: "reverted" }, + }); + expect(initial).toMatchObject({ workerCount: 1, routing: { state: "active" } }); + expect(Object.isFrozen(next.routing)).toBe(true); + }); + + it("reduces an OOM fallback to one worker without losing immutable routing state", () => { + const initial = streaming({ + kind: "parallel_router", + state: "active", + fallback: { kind: "sdr_disk", workerCount: 5, forceParallelStream: false }, + }); + const next = replanAfterFailure(initial, { + kind: "capture_failure", + memoryExhaustion: true, + }); + + expect(next).toMatchObject({ + kind: "sdr_disk", + workerCount: 1, + forceScreenshot: true, + routing: { kind: "parallel_router", state: "reverted" }, + }); + }); + + it("rejects a streaming transition from a non-streaming plan", () => { + const disk = createCapturePlan({ + workerCount: 2, + forceScreenshot: true, + useStreamingEncode: false, + useLayeredComposite: false, + usePageSideCompositing: false, + hasHdrContent: false, + needsAlpha: false, + }); + expect(() => replanAfterFailure(disk, { kind: "streaming_unavailable" })).toThrow( + "Cannot apply streaming_unavailable to sdr_disk", + ); + }); +}); diff --git a/packages/producer/src/services/render/capturePlan.ts b/packages/producer/src/services/render/capturePlan.ts new file mode 100644 index 0000000000..e4a5037ebe --- /dev/null +++ b/packages/producer/src/services/render/capturePlan.ts @@ -0,0 +1,150 @@ +/** + * Immutable routing decision for the capture phase. + * + * The orchestrator used to carry the same decision in several mutable booleans + * (`useStreamingEncode`, `useLayeredComposite`, `forceScreenshot`) plus worker + * routing state. Keeping those values independently mutable made invalid + * combinations representable during fallback. A CapturePlan is the single + * value consumed by capture stages, and `replanAfterFailure` is the only + * transition between variants. + */ + +export type CapturePlanTarget = Readonly<{ + kind: "sdr_streaming" | "sdr_disk"; + workerCount: number; + forceParallelStream: boolean; +}>; + +export type CaptureRouting = + | Readonly<{ kind: "default" }> + | Readonly<{ + kind: "worker_inversion" | "parallel_router"; + state: "active" | "reverted"; + fallback: CapturePlanTarget; + }>; + +interface CapturePlanBase { + readonly workerCount: number; + readonly forceScreenshot: boolean; + readonly forceParallelStream: boolean; + readonly usePageSideCompositing: boolean; + readonly hasHdrContent: boolean; + readonly needsAlpha: boolean; + readonly routing: CaptureRouting; +} + +export interface SdrStreamingCapturePlan extends CapturePlanBase { + readonly kind: "sdr_streaming"; +} + +export interface SdrDiskCapturePlan extends CapturePlanBase { + readonly kind: "sdr_disk"; + readonly forceParallelStream: false; +} + +export interface HdrLayeredCapturePlan extends CapturePlanBase { + readonly kind: "hdr_layered"; + readonly forceScreenshot: true; + readonly forceParallelStream: false; +} + +export type CapturePlan = SdrStreamingCapturePlan | SdrDiskCapturePlan | HdrLayeredCapturePlan; + +export interface CreateCapturePlanInput { + workerCount: number; + forceScreenshot: boolean; + forceParallelStream?: boolean; + useStreamingEncode: boolean; + useLayeredComposite: boolean; + usePageSideCompositing: boolean; + hasHdrContent: boolean; + needsAlpha: boolean; + routing?: CaptureRouting; +} + +export type CapturePlanFailure = + | Readonly<{ kind: "streaming_unavailable" }> + | Readonly<{ kind: "draw_element_verification" }> + | Readonly<{ kind: "capture_failure"; memoryExhaustion: boolean }>; + +function assertWorkerCount(workerCount: number): void { + if (!Number.isInteger(workerCount) || workerCount < 1) { + throw new Error(`CapturePlan workerCount must be a positive integer; got ${workerCount}`); + } +} + +function freezeTarget(target: CapturePlanTarget): CapturePlanTarget { + assertWorkerCount(target.workerCount); + if (target.kind === "sdr_disk" && target.forceParallelStream) { + throw new Error("CapturePlan disk fallback cannot force parallel streaming"); + } + return Object.freeze({ ...target }); +} + +function freezeRouting(routing: CaptureRouting | undefined): CaptureRouting { + if (!routing || routing.kind === "default") return Object.freeze({ kind: "default" }); + return Object.freeze({ ...routing, fallback: freezeTarget(routing.fallback) }); +} + +export function createCapturePlan(input: CreateCapturePlanInput): CapturePlan { + assertWorkerCount(input.workerCount); + const base = { + workerCount: input.workerCount, + forceScreenshot: input.forceScreenshot || input.usePageSideCompositing, + forceParallelStream: input.useStreamingEncode ? (input.forceParallelStream ?? false) : false, + usePageSideCompositing: input.usePageSideCompositing, + hasHdrContent: input.hasHdrContent, + needsAlpha: input.needsAlpha, + routing: freezeRouting(input.routing), + }; + + if (input.useLayeredComposite) { + return Object.freeze({ + ...base, + kind: "hdr_layered", + forceScreenshot: true, + forceParallelStream: false, + }); + } + if (input.useStreamingEncode) { + return Object.freeze({ ...base, kind: "sdr_streaming" }); + } + return Object.freeze({ ...base, kind: "sdr_disk", forceParallelStream: false }); +} + +function revertedRouting(routing: CaptureRouting): CaptureRouting { + if (routing.kind === "default") return routing; + return freezeRouting({ ...routing, state: "reverted" }); +} + +/** Pure, exhaustive capture fallback transition. The input plan is never mutated. */ +export function replanAfterFailure(plan: CapturePlan, failure: CapturePlanFailure): CapturePlan { + if (plan.kind !== "sdr_streaming") { + throw new Error(`Cannot apply ${failure.kind} to ${plan.kind} capture plan`); + } + + if (failure.kind === "streaming_unavailable") { + return createCapturePlan({ + ...plan, + useStreamingEncode: false, + useLayeredComposite: false, + forceParallelStream: false, + }); + } + + const fallback = + plan.routing.kind === "default" + ? { kind: plan.kind, workerCount: plan.workerCount, forceParallelStream: false } + : plan.routing.fallback; + const workerCount = + failure.kind === "capture_failure" && failure.memoryExhaustion ? 1 : fallback.workerCount; + return createCapturePlan({ + ...plan, + workerCount, + forceScreenshot: true, + forceParallelStream: fallback.forceParallelStream, + useStreamingEncode: fallback.kind === "sdr_streaming", + useLayeredComposite: false, + routing: revertedRouting(plan.routing), + }); +} diff --git a/packages/producer/src/services/render/stages/captureHdrStage.ts b/packages/producer/src/services/render/stages/captureHdrStage.ts index 06d77cdb46..89f9c52140 100644 --- a/packages/producer/src/services/render/stages/captureHdrStage.ts +++ b/packages/producer/src/services/render/stages/captureHdrStage.ts @@ -25,9 +25,8 @@ * because `captureAlphaPng` hangs under `--enable-begin-frame-control`. * Previously the stage mutated `cfg.forceScreenshot = true` directly; * the value is now derived into a local `hdrCfg` so the caller-owned - * `cfg` survives the stage unchanged. The sequencer is expected to - * pass `forceScreenshot: true` for the layered branch as a contract - * check. + * `cfg` survives the stage unchanged. The sequencer passes an immutable + * `hdr_layered` plan whose construction guarantees screenshot mode. * * Resource setup (HDR video extraction, image decode, dim probing) lives * in `captureHdrResources.ts`; per-frame work lives in @@ -44,7 +43,6 @@ import { type EngineConfig, type HdrTransfer, type StreamingEncoder, - calculateOptimalWorkers, closeCaptureSession, createCaptureSession, getEncoderPreset, @@ -77,18 +75,13 @@ import { partitionTransitionFrames, shouldUseHybridLayeredPath } from "./capture import { runSequentialLayeredFrameLoop } from "./captureHdrSequentialLoop.js"; import { runHybridLayeredFrameLoop } from "./captureHdrHybridLoop.js"; import { wrapCaptureStageError } from "../captureStageError.js"; +import type { HdrLayeredCapturePlan } from "../capturePlan.js"; export interface CaptureHdrStageInput { job: RenderJob; cfg: EngineConfig; - /** - * Capture-mode flag threaded from `compileStage`. The HDR layered - * branch requires `true` (see file header for the - * `captureAlphaPng` / `--enable-begin-frame-control` constraint); - * the stage throws if called with `false`. Stored locally as - * `hdrCfg.forceScreenshot` so the caller-owned `cfg` is not mutated. - */ - forceScreenshot: boolean; + /** Immutable layered route selected by the sequencer. */ + plan: HdrLayeredCapturePlan; log: ProducerLogger; projectDir: string; @@ -120,13 +113,6 @@ export interface CaptureHdrStageInput { /** Mutated in place (counters incremented). */ hdrDiagnostics: HdrDiagnostics; - /** - * Worker budget for the hybrid layered path. Only consulted when the - * gating predicate (`shouldUseHybridLayeredPath`) returns true. The - * sequential loop always runs on a single DOM session. - */ - workerCount?: number; - abortSignal: AbortSignal | undefined; assertNotAborted: () => void; onProgress?: ProgressCallback; @@ -160,7 +146,7 @@ export async function runCaptureHdrStage( const { job, cfg, - forceScreenshot, + plan, log, projectDir, compiledDir, @@ -184,17 +170,11 @@ export async function runCaptureHdrStage( buildCaptureOptions, createRenderVideoFrameInjector, hdrDiagnostics, - workerCount, abortSignal, assertNotAborted, onProgress, } = input; - - if (!forceScreenshot) { - throw new Error( - "captureHdrStage requires forceScreenshot=true; the layered composite path uses captureAlphaPng which hangs under --enable-begin-frame-control.", - ); - } + const { workerCount } = plan; const stageStart = Date.now(); let lastBrowserConsole: string[] = []; @@ -378,16 +358,7 @@ export async function runCaptureHdrStage( }; // ── Dispatch to sequential or hybrid frame loop ──────────────────── - // Resolve the worker budget here rather than threading it through the - // renderOrchestrator call: keeps the renderOrchestrator diff zero - // (hf#732 PR 4 is intentionally a producer-stage-local change), at the - // cost of recomputing the same number the orchestrator already knows. - // The cost is negligible (one cpus() call) and the two values stay in - // lockstep because `calculateOptimalWorkers` is pure. - const effectiveWorkerCount = - workerCount !== undefined - ? Math.max(1, workerCount) - : calculateOptimalWorkers(totalFrames, job.config.workers, hdrCfg); + const effectiveWorkerCount = Math.max(1, workerCount); const transitionFrameCount = partitionTransitionFrames(transitionRanges, totalFrames).size; const useHybrid = shouldUseHybridLayeredPath({ hasHdrContent, diff --git a/packages/producer/src/services/render/stages/captureStage.ts b/packages/producer/src/services/render/stages/captureStage.ts index bf71e7afec..808b00bffe 100644 --- a/packages/producer/src/services/render/stages/captureStage.ts +++ b/packages/producer/src/services/render/stages/captureStage.ts @@ -62,6 +62,7 @@ import { } from "../../renderOrchestrator.js"; import { wrapCaptureStageError } from "../captureStageError.js"; import { updateJobStatus } from "../shared.js"; +import type { SdrDiskCapturePlan } from "../capturePlan.js"; export interface CaptureStageInput { fileServer: FileServerHandle; @@ -76,23 +77,11 @@ export interface CaptureStageInput { */ totalFrames: number; cfg: EngineConfig; - /** - * Capture-mode flag threaded from `compileStage`. The stage derives a - * local copy of `cfg` with this value applied to `forceScreenshot` - * before any engine call, so the caller-owned `cfg` is never mutated. - * The sequencer may override `compileResult.forceScreenshot` after a - * BeginFrame calibration timeout — passing the override through this - * parameter keeps the decision visible at the call site instead of - * hiding it inside a shared mutable config. - */ - forceScreenshot: boolean; + /** Immutable route selected by the sequencer. */ + plan: SdrDiskCapturePlan; 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[]; /** @@ -154,7 +143,7 @@ export async function runCaptureStage(input: CaptureStageInput): Promise {}, warn: () => {}, info: () => {}, debug: () => {}, }, - workerCount: 1, probeSession: null, outputFormat: "mp4", streamingEncoderOptions: { fps: { num: 30, den: 1 }, width: 1920, height: 1080 }, @@ -287,10 +295,18 @@ describe("runCaptureStage", () => { try { await runCaptureStage({ ...createInput(cfg), + plan: createCapturePlan({ + workerCount: 1, + forceScreenshot: false, + useStreamingEncode: false, + useLayeredComposite: false, + usePageSideCompositing: false, + hasHdrContent: false, + needsAlpha: false, + }), videoOnlyPath: undefined, outputFormat: undefined, streamingEncoderOptions: undefined, - needsAlpha: false, captureAttempts: [], }); } catch (error) { @@ -326,7 +342,15 @@ describe("runCaptureHdrStage", () => { duration: 1, }, cfg: { forceScreenshot: true }, - forceScreenshot: true, + plan: createCapturePlan({ + workerCount: 1, + forceScreenshot: true, + useStreamingEncode: false, + useLayeredComposite: true, + usePageSideCompositing: false, + hasHdrContent: false, + needsAlpha: false, + }), log: { error: () => {}, warn: () => {}, @@ -375,7 +399,6 @@ describe("runCaptureHdrStage", () => { videoExtractionFailures: 0, imageDecodeFailures: 0, }, - workerCount: 1, abortSignal: undefined, assertNotAborted: () => {}, }); diff --git a/packages/producer/src/services/render/stages/captureStreamingStage.ts b/packages/producer/src/services/render/stages/captureStreamingStage.ts index fe81ff1f48..41f4fd60e8 100644 --- a/packages/producer/src/services/render/stages/captureStreamingStage.ts +++ b/packages/producer/src/services/render/stages/captureStreamingStage.ts @@ -78,6 +78,7 @@ import { wrapCaptureStageError } from "../captureStageError.js"; import { pushWorkerDedupPerfs } from "../perfSummary.js"; import { ensureFrameWritten } from "./captureHdrFrameShared.js"; import { updateJobStatus } from "../shared.js"; +import type { SdrStreamingCapturePlan } from "../capturePlan.js"; /** * No-frame-progress watchdog for the parallel DE streaming path. The router @@ -120,27 +121,10 @@ export interface CaptureStreamingStageInput { */ totalFrames: number; cfg: EngineConfig; - /** - * Capture-mode flag threaded from `compileStage`. The stage derives a - * local copy of `cfg` with this value applied to `forceScreenshot` - * before any engine call, so the caller-owned `cfg` is never mutated. - * The sequencer may override `compileResult.forceScreenshot` after a - * BeginFrame calibration timeout — passing the override through this - * parameter keeps the decision visible at the call site instead of - * hiding it inside a shared mutable config. - */ - forceScreenshot: boolean; + /** Immutable route selected by the sequencer. */ + plan: SdrStreamingCapturePlan; log: ProducerLogger; - workerCount: number; probeSession: CaptureSession | null; - /** - * Per-render override from the DE parallel router — see - * deParallelStreamForced's declaration in renderOrchestrator.ts. Distinct - * from the `HF_DE_PARALLEL_STREAM` manual opt-in (still read directly by - * this stage) because the router's decision must not leak across - * concurrently-running renders sharing this process via a global env var. - */ - forceParallelStream?: boolean; /** For the spawn-failure log message context only. */ outputFormat: string; /** Pre-built encoder options; passed straight to `spawnStreamingEncoder`. */ @@ -474,7 +458,7 @@ export async function runCaptureStreamingStage( job, totalFrames, cfg, - forceScreenshot, + plan, log, outputFormat, streamingEncoderOptions, @@ -484,16 +468,17 @@ export async function runCaptureStreamingStage( assertNotAborted, onProgress, dedupPerfs, - forceParallelStream, } = input; - let { workerCount, probeSession } = input; + let { probeSession } = input; + let { workerCount } = plan; + const { forceScreenshot, forceParallelStream } = plan; let lastBrowserConsole: string[] = []; let deDrainStats: DeDrainStats | undefined; let captureBeyondViewport: boolean | undefined = probeSession?.options.captureBeyondViewport; // Derive a local cfg view rather than reading `forceScreenshot` from the // caller-owned `cfg`. The sequencer threads the resolved value via the - // explicit parameter; this keeps the engine-facing config a pure + // immutable plan; this keeps the engine-facing config a pure // pass-through. const captureCfg: EngineConfig = cfg.forceScreenshot === forceScreenshot ? cfg : { ...cfg, forceScreenshot }; diff --git a/packages/producer/src/services/renderOrchestrator.ts b/packages/producer/src/services/renderOrchestrator.ts index c7843966e9..e5f97d98b7 100644 --- a/packages/producer/src/services/renderOrchestrator.ts +++ b/packages/producer/src/services/renderOrchestrator.ts @@ -93,6 +93,12 @@ import { buildRenderErrorDetails } from "./render/cleanup.js"; import { publishRenderFailure } from "./render/renderEventPublisher.js"; import { RenderExecutionContext } from "./render/renderExecutionContext.js"; import { ArtifactTransaction } from "./render/artifactTransaction.js"; +import { + createCapturePlan, + replanAfterFailure, + type CapturePlan, + type CaptureRouting, +} from "./render/capturePlan.js"; import { normalizeErrorMessage } from "../utils/errorMessage.js"; import { formatCaptureFrameName } from "../utils/paths.js"; import { resolveEffectiveHdrMode } from "./render/hdrMode.js"; @@ -2547,22 +2553,85 @@ async function executeRenderPipeline(input: { hasShaderTransitions: compiled.hasShaderTransitions && !isGif, isPngSequence, }); - updateCaptureObservability({ + const inversionFallback = resolveInversionRetryPlan({ + deWorkerInversion, + preInversionWorkerCount: preRoutingWorkerCount, + cfg, + outputFormat, + durationSeconds: job.duration, + isMemoryExhaustion: false, + }); + const parallelRouterFallback = resolveParallelRouterRetryPlan({ + deParallelRouter, + preRouterWorkerCount: preRoutingWorkerCount, + cfg, + outputFormat, + durationSeconds: job.duration, + isMemoryExhaustion: false, + }); + const captureRouting: CaptureRouting = inversionFallback + ? { + kind: "worker_inversion", + state: "active", + fallback: { + kind: inversionFallback.useStreamingEncode ? "sdr_streaming" : "sdr_disk", + workerCount: inversionFallback.workerCount, + forceParallelStream: false, + }, + } + : parallelRouterFallback + ? { + kind: "parallel_router", + state: "active", + fallback: { + kind: parallelRouterFallback.useStreamingEncode ? "sdr_streaming" : "sdr_disk", + workerCount: parallelRouterFallback.workerCount, + forceParallelStream: false, + }, + } + : { kind: "default" }; + let capturePlan: CapturePlan = createCapturePlan({ workerCount, + forceScreenshot: captureForceScreenshot, + forceParallelStream: deParallelStreamForced || captureParallelStreamForced, useStreamingEncode, useLayeredComposite, usePageSideCompositing: usePageSideCompositingForTransitions, hasHdrContent, - forceScreenshot: captureForceScreenshot, + needsAlpha, + routing: captureRouting, + }); + const syncCapturePlan = (): void => { + workerCount = capturePlan.workerCount; + captureForceScreenshot = capturePlan.forceScreenshot; + useStreamingEncode = capturePlan.kind === "sdr_streaming"; + deParallelStreamForced = + capturePlan.kind === "sdr_streaming" && capturePlan.forceParallelStream; + if (capturePlan.routing.kind === "worker_inversion") { + deWorkerInversion = capturePlan.routing.state === "active" ? "inverted" : "reverted"; + } + if (capturePlan.routing.kind === "parallel_router") { + deParallelRouter = capturePlan.routing.state === "active" ? "routed" : "reverted"; + } + }; + syncCapturePlan(); + updateCaptureObservability({ + workerCount: capturePlan.workerCount, + useStreamingEncode: capturePlan.kind === "sdr_streaming", + useLayeredComposite: capturePlan.kind === "hdr_layered", + usePageSideCompositing: capturePlan.usePageSideCompositing, + hasHdrContent: capturePlan.hasHdrContent, + forceScreenshot: capturePlan.forceScreenshot, }); observability.checkpoint("capture_strategy", "resolved", { - workerCount, - forceScreenshot: captureForceScreenshot, + plan: capturePlan.kind, + workerCount: capturePlan.workerCount, + forceScreenshot: capturePlan.forceScreenshot, captureBeyondViewport: resolvedCaptureBeyondViewport ?? null, - useStreamingEncode, - useLayeredComposite, - usePageSideCompositing: usePageSideCompositingForTransitions, - hasHdrContent, + useStreamingEncode: capturePlan.kind === "sdr_streaming", + useLayeredComposite: capturePlan.kind === "hdr_layered", + usePageSideCompositing: capturePlan.usePageSideCompositing, + hasHdrContent: capturePlan.hasHdrContent, hasShaderTransitions: compiled.hasShaderTransitions, isPngSequence, }); @@ -2604,13 +2673,13 @@ async function executeRenderPipeline(input: { // into the active rgb48le signal space. Shader transitions use this same // path for SDR compositions so the engine can apply transition math to // isolated scene buffers instead of recording plain DOM screenshots. - if (useLayeredComposite) { + if (capturePlan.kind === "hdr_layered") { + const layeredPlan = capturePlan; // Layered composite always runs in screenshot mode — keep // `captureForceScreenshot` in sync so the perf summary and any // post-HDR diagnostic that reads the boolean see the same value // the stage uses internally. - captureForceScreenshot = true; - updateCaptureObservability({ forceScreenshot: captureForceScreenshot }); + updateCaptureObservability({ forceScreenshot: layeredPlan.forceScreenshot }); const hdrRes = await observeRenderStage( observability, "capture_hdr_layered", @@ -2619,7 +2688,7 @@ async function executeRenderPipeline(input: { runCaptureHdrStage({ job, cfg, - forceScreenshot: captureForceScreenshot, + plan: layeredPlan, log, projectDir, compiledDir, @@ -2662,9 +2731,13 @@ async function executeRenderPipeline(input: { // streaming spawn fails (non-abort) the stage returns { success: false } // and we fall back to the disk path below. let streamingHandled = false; - if (useStreamingEncode) { + if (capturePlan.kind === "sdr_streaming") { const captureFrameStart = Date.now(); const invokeStreaming = () => { + if (capturePlan.kind !== "sdr_streaming") { + throw new Error(`Cannot invoke streaming stage with ${capturePlan.kind} plan`); + } + const streamingPlan = capturePlan; resetCaptureAttemptProgress(job); return observeRenderStage( observability, @@ -2679,12 +2752,10 @@ async function executeRenderPipeline(input: { job, totalFrames, cfg, - forceScreenshot: captureForceScreenshot, + plan: streamingPlan, log, - workerCount, probeSession, outputFormat, - forceParallelStream: deParallelStreamForced || captureParallelStreamForced, streamingEncoderOptions: { fps: job.config.fps, width, @@ -2755,71 +2826,45 @@ async function executeRenderPipeline(input: { ? "drawElement self-verify failed; retrying with forceScreenshot" : "capture failed on pinned worker count; retrying with forceScreenshot", ); - captureForceScreenshot = true; + const failedRouting = capturePlan.routing.kind; + capturePlan = replanAfterFailure( + capturePlan, + isVerifyError + ? { kind: "draw_element_verification" } + : { kind: "capture_failure", memoryExhaustion: isMemoryExhaustion }, + ); + syncCapturePlan(); updateCaptureObservability({ - forceScreenshot: true, + forceScreenshot: capturePlan.forceScreenshot, deSelfVerifyFallback, deFallbackReason, - }); - probeSession = null; - // Must clear BEFORE resolveParallelRouterRetryPlan recomputes - // useStreamingEncode, or shouldUseStreamingEncode would keep - // resolving to the parallel-streaming shape on the retry instead - // of the well-tested parallel-disk fallback. - if (deParallelRouter === "routed") deParallelStreamForced = false; - const inversionRetryPlan = resolveInversionRetryPlan({ + workerCount: capturePlan.workerCount, + useStreamingEncode: capturePlan.kind === "sdr_streaming", deWorkerInversion, - preInversionWorkerCount: preRoutingWorkerCount, - cfg, - outputFormat, - durationSeconds: job.duration, - isMemoryExhaustion, - }); - const parallelRouterRetryPlan = resolveParallelRouterRetryPlan({ deParallelRouter, - preRouterWorkerCount: preRoutingWorkerCount, - cfg, - outputFormat, - durationSeconds: job.duration, - isMemoryExhaustion, }); - if (inversionRetryPlan) { + probeSession = null; + if (failedRouting === "worker_inversion") { // The inversion bet on drawElement and lost — re-render on the // pre-inversion parallel screenshot path instead of single-worker // screenshot streaming (the slowest capture shape for this size). // "reverted" (not cleared) so telemetry keeps the lost-inversion // cohort distinguishable from renders that never inverted. - deWorkerInversion = inversionRetryPlan.deWorkerInversion; - workerCount = inversionRetryPlan.workerCount; - useStreamingEncode = inversionRetryPlan.useStreamingEncode; - updateCaptureObservability({ - workerCount, - useStreamingEncode, - deWorkerInversion, - }); log.info( - `[Render] Reverting worker inversion for the retry: ${workerCount} workers, ` + - `streaming=${useStreamingEncode}.`, + `[Render] Reverting worker inversion for the retry: ${capturePlan.workerCount} workers, ` + + `plan=${capturePlan.kind}.`, ); - } else if (parallelRouterRetryPlan) { + } else if (failedRouting === "parallel_router") { // The router's bet on verified parallel streaming lost — re-render // on the ordinary (non-DE) parallel path at the pre-router worker // count, same "reverted, not cleared" telemetry contract as the // inversion above. - deParallelRouter = parallelRouterRetryPlan.deParallelRouter; - workerCount = parallelRouterRetryPlan.workerCount; - useStreamingEncode = parallelRouterRetryPlan.useStreamingEncode; - updateCaptureObservability({ - workerCount, - useStreamingEncode, - deParallelRouter, - }); log.info( - `[Render] Reverting parallel router for the retry: ${workerCount} workers, ` + - `streaming=${useStreamingEncode}.`, + `[Render] Reverting parallel router for the retry: ${capturePlan.workerCount} workers, ` + + `plan=${capturePlan.kind}.`, ); } - if (useStreamingEncode) { + if (capturePlan.kind === "sdr_streaming") { streamingRes = await invokeStreaming(); } else { // Parallel retry goes through the disk path below. @@ -2848,7 +2893,10 @@ async function executeRenderPipeline(input: { perfStages.captureSetupMs = Math.max(0, perfStages.captureMs - captureFrameMs); perfStages.encodeMs = streamingRes.encodeMs; // Overlapped with capture } else { - useStreamingEncode = false; + if (capturePlan.kind === "sdr_streaming") { + capturePlan = replanAfterFailure(capturePlan, { kind: "streaming_unavailable" }); + syncCapturePlan(); + } // The disk path has no drain-time self-verification — clamp // default-on drawElement here exactly like the pre-capture clamp // (verified-path confinement). Skipped when screenshots are already @@ -2856,7 +2904,7 @@ async function executeRenderPipeline(input: { // opt-in, mirroring the clamp above. if ( cfg.useDrawElement && - !captureForceScreenshot && + !capturePlan.forceScreenshot && process.env.PRODUCER_EXPERIMENTAL_FAST_CAPTURE !== "true" ) { cfg.useDrawElement = false; @@ -2872,19 +2920,23 @@ async function executeRenderPipeline(input: { probeSession = null; } } - updateCaptureObservability({ useStreamingEncode }); + updateCaptureObservability({ useStreamingEncode: false }); observability.checkpoint("capture_streaming", "spawn failed; falling back to disk"); } } if (!streamingHandled) { + if (capturePlan.kind !== "sdr_disk") { + throw new Error(`Disk capture requires sdr_disk plan; got ${capturePlan.kind}`); + } + const diskPlan = capturePlan; // ── Disk-based capture (original flow) ──────────────────────────── resetCaptureAttemptProgress(job); const captureFrameStart = Date.now(); const captureRes = await observeRenderStage( observability, "capture_disk", - captureStageObservationData({ needsAlpha }), + captureStageObservationData({ needsAlpha: diskPlan.needsAlpha }), () => runCaptureStage({ fileServer: activeFileServer, @@ -2893,11 +2945,9 @@ async function executeRenderPipeline(input: { job, totalFrames, cfg, - forceScreenshot: captureForceScreenshot, + plan: diskPlan, log, - workerCount, probeSession, - needsAlpha, captureAttempts, dedupPerfs, buildCaptureOptions,