diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index b174e8c22..e2b4b49c1 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -177,6 +177,7 @@ export type { export { calculateOptimalWorkers, distributeFrames, + distributeFramesInterleaved, executeParallelCapture, mergeWorkerFrames, getSystemResources, diff --git a/packages/engine/src/services/frameCapture.ts b/packages/engine/src/services/frameCapture.ts index 72d1ec030..b8a2d5f59 100644 --- a/packages/engine/src/services/frameCapture.ts +++ b/packages/engine/src/services/frameCapture.ts @@ -161,6 +161,11 @@ export interface CaptureSession { * worker-encode path (mirrors `lastFrameBuffer` on the screenshot path). Only set * when static-frame dedup is armed on the drawElement path. */ lastEncodeResult?: Promise; + /** Frame index lastEncodeResult belongs to — static-dedup reuse must verify + * every frame in (lastEncodeResultFrame, i] is predicted-static (under the + * interleaved parallel stride the "previous" produced frame is i−N, and + * reusing it for frame i is only valid when the whole gap is static). */ + lastEncodeResultFrame?: number; /** Per-render self-verification ground truth (ungated-release safety net): * K screenshot frames captured at init BEFORE the drawElement canvas is * injected (the only window where a page screenshot shows the live DOM, not @@ -2651,9 +2656,23 @@ export async function captureFrameToBufferPipelined( // and skip the seek + drawElement + encode entirely. Same predicate as the serial // path; clip-cut frames are excluded from staticFrames so they always capture. if (session.staticFrames?.has(frameIndex) && session.lastEncodeResult) { - session.staticDedupCount = (session.staticDedupCount ?? 0) + 1; - session.capturePerf.frames += 1; - return { encodeResult: session.lastEncodeResult, captureTimeMs: Date.now() - startTime }; + // Reuse is valid only when EVERY frame in (lastEncodeResultFrame, i] is + // predicted-static — then all of them (and the reused buffer) share the + // same pixels. Sequential capture reduces to has(i) (gap = {i}); the + // interleaved parallel stride makes the gap N frames wide. + const lastIdx = session.lastEncodeResultFrame ?? frameIndex - 1; + let gapStatic = true; + for (let j = lastIdx + 1; j <= frameIndex; j++) { + if (!session.staticFrames.has(j)) { + gapStatic = false; + break; + } + } + if (gapStatic) { + session.staticDedupCount = (session.staticDedupCount ?? 0) + 1; + session.capturePerf.frames += 1; + return { encodeResult: session.lastEncodeResult, captureTimeMs: Date.now() - startTime }; + } } try { @@ -2688,7 +2707,10 @@ export async function captureFrameToBufferPipelined( session.capturePerf.frameMs.push(boundaryMs); } const boundaryResult = Promise.resolve(buffer); - if (session.staticFrames) session.lastEncodeResult = boundaryResult; + if (session.staticFrames) { + session.lastEncodeResult = boundaryResult; + session.lastEncodeResultFrame = frameIndex; + } return { encodeResult: boundaryResult, captureTimeMs: Date.now() - startTime }; } @@ -2715,7 +2737,10 @@ export async function captureFrameToBufferPipelined( session.capturePerf.frameMs.push(captureTimeMs); // Task B: retain this encode result so a following static frame can reuse it. - if (session.staticFrames) session.lastEncodeResult = encodeResult; + if (session.staticFrames) { + session.lastEncodeResult = encodeResult; + session.lastEncodeResultFrame = frameIndex; + } return { encodeResult, captureTimeMs }; } catch (captureError) { diff --git a/packages/engine/src/services/parallelCoordinator.ts b/packages/engine/src/services/parallelCoordinator.ts index 6fc6f420f..f2b766a49 100644 --- a/packages/engine/src/services/parallelCoordinator.ts +++ b/packages/engine/src/services/parallelCoordinator.ts @@ -15,6 +15,7 @@ import { initializeSession, closeCaptureSession, captureFrame, + captureFrameToBufferPipelined, captureFrameToBuffer, getCapturePerfSummary, type CaptureSession, @@ -42,6 +43,13 @@ export interface WorkerTask { * calculation still uses the absolute frame index. */ outputFrameOffset?: number; + /** + * Frame stride for interleaved distribution (HF_DE_PARALLEL_STREAM spike): + * the worker captures startFrame, startFrame+stride, … < endFrame. Default 1 + * (contiguous range). Interleaving keeps the ordered streaming writer's + * reorder window at O(workerCount) frames instead of O(totalFrames/N). + */ + frameStride?: number; } export interface WorkerResult { @@ -234,6 +242,34 @@ export function distributeFrames( return tasks; } +/** + * Interleaved (round-robin) distribution: worker i captures frames + * i, i+N, i+2N, …. Seek-based capture makes stride access free (every frame + * is an absolute seek), and the streaming reorder window shrinks from + * totalFrames/N to N — contiguous chunks serialize workers behind the + * ordered writer (worker 1's first frame waits for ALL of worker 0's). + * HF_DE_PARALLEL_STREAM spike; disk-path capture keeps contiguous chunks. + */ +export function distributeFramesInterleaved( + totalFrames: number, + workerCount: number, + workDir: string, + rangeStart: number = 0, +): WorkerTask[] { + const tasks: WorkerTask[] = []; + for (let i = 0; i < workerCount && i < totalFrames; i++) { + tasks.push({ + workerId: i, + startFrame: rangeStart + i, + endFrame: rangeStart + totalFrames, + frameStride: workerCount, + outputDir: join(workDir, `worker-${i}`), + outputFrameOffset: rangeStart, + }); + } + return tasks; +} + /** * Decide whether a parallel worker should run the per-worker SwiftShader * assertion. Gated to worker 0 only: workers within a chunk share the same @@ -244,24 +280,89 @@ export function shouldVerifyWorkerGpu(workerId: number, config?: Partial void) | undefined, - onFrameBuffer: ((frameIndex: number, buffer: Buffer) => Promise) | undefined, + onFrameBuffer: + | ((frameIndex: number, buffer: Buffer, session: CaptureSession) => Promise) + | undefined, ): Promise { let framesCaptured = 0; const outputOffset = task.outputFrameOffset ?? 0; - for (let i = task.startFrame; i < task.endFrame; i++) { + const stride = task.frameStride ?? 1; + // Depth-2 pipelined drawElement produce (HF_DE_PARALLEL_STREAM spike): frame + // k's in-page worker encode overlaps frame k+stride's produce phase — the + // same shape as the sequential worker-encode loop. Only engaged when the + // session's encode worker initialized (drawElement mode) and frames stream + // back via onFrameBuffer; the ordered writer's waitForFrame provides the + // cross-worker backpressure (each worker runs at most `stride` frames ahead). + // NOTE: this branch fires for any stride, but production only ever reaches + // it via HF_DE_PARALLEL_STREAM, which always uses interleaved distribution + // (stride = workerCount). The stride=1 (contiguous) path through here is + // validation-only — exercised by tests wiring onFrameBuffer with a + // contiguous multi-worker task, not a shape real renders take. Don't + // "simplify" the flag checks around this without accounting for that. + if (onFrameBuffer && session.workerEncodeEnabled) { + const dbg = process.env.HF_DE_PAR_DEBUG === "1"; + const dbgT0 = Date.now(); + const dbgWin = 40 * stride; + let prev: { idx: number; encodeResult: Promise } | null = null; + for (let i = task.startFrame; i < task.endFrame; i += stride) { + if (signal?.aborted) throw new Error("Parallel worker cancelled"); + const time = (i * captureOptions.fps.den) / captureOptions.fps.num; + if (dbg && i < task.startFrame + dbgWin) { + console.log(`[par:w${task.workerId}] +${Date.now() - dbgT0}ms produce ${i} start`); + } + const { encodeResult } = await captureFrameToBufferPipelined(session, i - outputOffset, time); + // Marks the promise "handled" for Node's unhandled-rejection detector + // without affecting the real `await prev.encodeResult` below — if a + // later iteration throws (abort, downstream writeFrame failure) before + // this frame's encode is drained, it's abandoned rather than awaited, + // and would otherwise surface as an unhandled rejection during teardown. + encodeResult.catch(() => {}); + if (dbg && i < task.startFrame + dbgWin) { + console.log(`[par:w${task.workerId}] +${Date.now() - dbgT0}ms produce ${i} kicked`); + } + if (prev) { + if (dbg && prev.idx < task.startFrame + dbgWin) { + console.log( + `[par:w${task.workerId}] +${Date.now() - dbgT0}ms drain ${prev.idx} await-encode`, + ); + } + const buf = await prev.encodeResult; + if (dbg && prev.idx < task.startFrame + dbgWin) { + console.log( + `[par:w${task.workerId}] +${Date.now() - dbgT0}ms drain ${prev.idx} encoded ${buf.length}B`, + ); + } + await onFrameBuffer(prev.idx, buf, session); + if (dbg && prev.idx < task.startFrame + dbgWin) { + console.log(`[par:w${task.workerId}] +${Date.now() - dbgT0}ms drain ${prev.idx} written`); + } + framesCaptured++; + if (onFrameCaptured) onFrameCaptured(task.workerId, prev.idx); + } + prev = { idx: i, encodeResult }; + } + if (prev) { + await onFrameBuffer(prev.idx, await prev.encodeResult, session); + framesCaptured++; + if (onFrameCaptured) onFrameCaptured(task.workerId, prev.idx); + } + return framesCaptured; + } + for (let i = task.startFrame; i < task.endFrame; i += stride) { if (signal?.aborted) throw new Error("Parallel worker cancelled"); const time = (i * captureOptions.fps.den) / captureOptions.fps.num; const fileFrameIdx = i - outputOffset; if (onFrameBuffer) { const { buffer } = await captureFrameToBuffer(session, fileFrameIdx, time); - await onFrameBuffer(i, buffer); + await onFrameBuffer(i, buffer, session); } else { await captureFrame(session, fileFrameIdx, time); } @@ -278,7 +379,7 @@ async function executeWorkerTask( createBeforeCaptureHook: () => BeforeCaptureHook | null, signal?: AbortSignal, onFrameCaptured?: (workerId: number, frameIndex: number) => void, - onFrameBuffer?: (frameIndex: number, buffer: Buffer) => Promise, + onFrameBuffer?: (frameIndex: number, buffer: Buffer, session: CaptureSession) => Promise, config?: Partial, parallel?: boolean, ): Promise { @@ -309,11 +410,19 @@ async function executeWorkerTask( createBeforeCaptureHook(), workerConfig, ); + if (process.env.HF_DE_PAR_DEBUG === "1") { + console.log(`[par:w${task.workerId}] session created`); + } // Worker-0-only SwiftShader assertion — see `shouldVerifyWorkerGpu` and #955. if (shouldVerifyWorkerGpu(task.workerId, workerConfig)) { await assertSwiftShader(session.page, readWebGlVendorInfoFromCanvas); } await initializeSession(session); + if (process.env.HF_DE_PAR_DEBUG === "1") { + console.log( + `[par:w${task.workerId}] init done (mode=${session.captureMode} workerEncode=${session.workerEncodeEnabled === true})`, + ); + } framesCaptured = await captureFrameRange( session, task, @@ -358,10 +467,18 @@ export async function executeParallelCapture( createBeforeCaptureHook: () => BeforeCaptureHook | null, signal?: AbortSignal, onProgress?: (progress: ParallelProgress) => void, - onFrameBuffer?: (frameIndex: number, buffer: Buffer) => Promise, + onFrameBuffer?: (frameIndex: number, buffer: Buffer, session: CaptureSession) => Promise, config?: Partial, ): Promise { - const totalFrames = tasks.reduce((sum, t) => sum + (t.endFrame - t.startFrame), 0); + // `endFrame - startFrame` is the correct per-task frame count for contiguous + // tasks (stride 1), but for interleaved tasks (stride = workerCount) each + // task spans nearly the full range while only actually capturing 1/stride + // of it — dividing by stride here matches the loop in `captureFrameRange` + // (`i += stride`) so progress doesn't plateau at ~1/workerCount. + const totalFrames = tasks.reduce( + (sum, t) => sum + Math.ceil((t.endFrame - t.startFrame) / (t.frameStride ?? 1)), + 0, + ); const workerProgress = new Map(); for (const task of tasks) workerProgress.set(task.workerId, 0); diff --git a/packages/engine/src/services/streamingEncoder.test.ts b/packages/engine/src/services/streamingEncoder.test.ts index 4623af1d1..cd99c7554 100644 --- a/packages/engine/src/services/streamingEncoder.test.ts +++ b/packages/engine/src/services/streamingEncoder.test.ts @@ -905,3 +905,23 @@ describe("spawnStreamingEncoder lifecycle and cleanup", () => { } }); }); + +describe("createFrameReorderBuffer abort (interleaved parallel drain)", () => { + it("rejects parked waiters so a failed worker cannot deadlock its peers", async () => { + const buf = createFrameReorderBuffer(0, 100); + const parked = buf.waitForFrame(5); + const err = new Error("verify failed"); + buf.abort(err); + await expect(parked).rejects.toThrow("verify failed"); + // Future waiters reject immediately too. + await expect(buf.waitForFrame(6)).rejects.toThrow("verify failed"); + await expect(buf.waitForAllDone()).rejects.toThrow("verify failed"); + }); + + it("in-order waiters still resolve before an abort", async () => { + const buf = createFrameReorderBuffer(0, 10); + await expect(buf.waitForFrame(0)).resolves.toBeUndefined(); + buf.advanceTo(1); + await expect(buf.waitForFrame(1)).resolves.toBeUndefined(); + }); +}); diff --git a/packages/engine/src/services/streamingEncoder.ts b/packages/engine/src/services/streamingEncoder.ts index dba5ed59d..109cf4041 100644 --- a/packages/engine/src/services/streamingEncoder.ts +++ b/packages/engine/src/services/streamingEncoder.ts @@ -52,18 +52,27 @@ export interface FrameReorderBuffer { waitForFrame: (frame: number) => Promise; advanceTo: (frame: number) => void; waitForAllDone: () => Promise; + /** + * 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; - const pending = new Map void>>(); + let aborted: Error | null = null; + const pending = new Map void; reject: (e: Error) => void }>>(); - const enqueueAt = (frame: number, resolve: () => void): void => { + const enqueueAt = (frame: number, resolve: () => void, reject: (e: Error) => void): void => { const list = pending.get(frame); if (list === undefined) { - pending.set(frame, [resolve]); + pending.set(frame, [{ resolve, reject }]); } else { - list.push(resolve); + list.push({ resolve, reject }); } }; @@ -71,16 +80,20 @@ export function createFrameReorderBuffer(startFrame: number, endFrame: number): const list = pending.get(frame); if (list === undefined) return; pending.delete(frame); - for (const resolve of list) resolve(); + for (const waiter of list) waiter.resolve(); }; const waitForFrame = (frame: number): Promise => - new Promise((resolve) => { + new Promise((resolve, reject) => { + if (aborted) { + reject(aborted); + return; + } if (frame === cursor) { resolve(); return; } - enqueueAt(frame, resolve); + enqueueAt(frame, resolve, reject); }); const advanceTo = (frame: number): void => { @@ -89,15 +102,28 @@ export function createFrameReorderBuffer(startFrame: number, endFrame: number): }; const waitForAllDone = (): Promise => - new Promise((resolve) => { + new Promise((resolve, reject) => { + if (aborted) { + reject(aborted); + return; + } if (cursor >= endFrame) { resolve(); return; } - enqueueAt(endFrame, resolve); + enqueueAt(endFrame, resolve, reject); }); - return { waitForFrame, advanceTo, waitForAllDone }; + 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 }; } // --------------------------------------------------------------------------- diff --git a/packages/producer/src/services/render/stages/captureStreamingStage.ts b/packages/producer/src/services/render/stages/captureStreamingStage.ts index 721ef6688..e22d2a3e9 100644 --- a/packages/producer/src/services/render/stages/captureStreamingStage.ts +++ b/packages/producer/src/services/render/stages/captureStreamingStage.ts @@ -61,6 +61,7 @@ import { createCaptureSession, createFrameReorderBuffer, distributeFrames, + distributeFramesInterleaved, executeParallelCapture, getCapturePerfSummary, getFfmpegBinary, @@ -179,24 +180,19 @@ async function psnrDb(a: Buffer, b: Buffer): Promise { } } -async function runWorkerEncodePipelineLoop( - session: CaptureSession, - totalFrames: number, - job: CaptureStreamingStageInput["job"], - currentEncoder: StreamingEncoder, - reorderBuffer: ReturnType, - assertNotAborted: () => void, - onProgress: CaptureStreamingStageInput["onProgress"], - log: CaptureStreamingStageInput["log"], - stats: DeDrainStats, -): Promise { - let prev: { idx: number; encodeResult: Promise } | null = null; - const frameTime = (i: number) => (i * job.config.fps.den) / job.config.fps.num; - - // ── drawElement drain-time safety checks (ungated-release safety net) ── - // Runs on every drained frame. Drains execute strictly between capture - // evaluates (see the batch NOTE below), so a re-seek + recapture here is - // safe — nothing else is mutating page time. +// ── drawElement drain-time safety checks (ungated-release safety net) ── +// Shared by the sequential worker-encode loop and the interleaved parallel +// streaming drain (HF_DE_PARALLEL_STREAM): the guard is session-parameterized +// so each parallel worker's frames verify against ITS OWN session's +// pre-injection ground truth (all sessions arm the same K sample indices from +// CaptureOptions.compositionDurationSeconds, so every sample is checked by +// whichever worker owns that frame). +function createDrainFrameGuard(args: { + log: CaptureStreamingStageInput["log"]; + stats: DeDrainStats; + frameTime: (i: number) => number; +}): (session: CaptureSession, idx: number, buf: Buffer) => Promise { + const { log, stats, frameTime } = args; // // 1. Blank guard (worker-path analogue of the serial-path guard in // frameCapture): drawElement intermittently returns an anomalously small @@ -237,7 +233,7 @@ async function runWorkerEncodePipelineLoop( return Math.max(absFloor, median * 0.12); }; let acceptedSmall: Buffer | null = null; - const guardFrame = async (idx: number, buf: Buffer): Promise => { + return async (session: CaptureSession, idx: number, buf: Buffer): Promise => { if (process.env.HF_FORCE_DRAWELEMENT !== "1") { const floor = blankFloor(); if (floor > 0 && buf.length < floor && acceptedSmall?.equals(buf)) { @@ -326,6 +322,23 @@ async function runWorkerEncodePipelineLoop( } return buf; }; +} + +async function runWorkerEncodePipelineLoop( + session: CaptureSession, + totalFrames: number, + job: CaptureStreamingStageInput["job"], + currentEncoder: StreamingEncoder, + reorderBuffer: ReturnType, + assertNotAborted: () => void, + onProgress: CaptureStreamingStageInput["onProgress"], + log: CaptureStreamingStageInput["log"], + stats: DeDrainStats, +): Promise { + let prev: { idx: number; encodeResult: Promise } | null = null; + const frameTime = (i: number) => (i * job.config.fps.den) / job.config.fps.num; + const guard = createDrainFrameGuard({ log, stats, frameTime }); + const guardFrame = (idx: number, buf: Buffer): Promise => guard(session, idx, buf); const drainPrev = async (): Promise => { if (!prev) return; @@ -494,43 +507,120 @@ export async function runCaptureStreamingStage( if (workerCount > 1) { // Parallel capture → streaming encode - const tasks = distributeFrames(totalFrames, workerCount, workDir); + // HF_DE_PARALLEL_STREAM (opt-in): interleaved distribution — worker i + // takes frames i, i+N, i+2N… so the ordered writer's reorder window is + // N frames and workers run in lockstep instead of serializing behind + // contiguous ranges (see distributeFramesInterleaved). Each worker runs + // the depth-2 pipelined drawElement produce when its session initialized + // in drawelement mode. + const deParallelStream = process.env.HF_DE_PARALLEL_STREAM === "true"; + const tasks = deParallelStream + ? distributeFramesInterleaved(totalFrames, workerCount, workDir) + : distributeFrames(totalFrames, workerCount, workDir); - const onFrameBuffer = async (frameIndex: number, buffer: Buffer): Promise => { - await reorderBuffer.waitForFrame(frameIndex); - ensureFrameWritten(await currentEncoder.writeFrame(buffer), frameIndex, currentEncoder); - reorderBuffer.advanceTo(frameIndex + 1); + // Drain-time safety net for parallel drawElement frames: the SAME + // blank-guard + PSNR self-verify the sequential worker-encode drain + // runs, session-parameterized so each frame verifies against its + // owning worker's pre-injection ground truth. A verification failure + // rejects the worker, executeParallelCapture rethrows, and the + // orchestrator's DrawElementVerificationError handler re-renders via + // screenshot (post-#2026 it also reverts any worker inversion). + // Intentionally ONE guard shared by all workers rather than one per + // worker: Node is single-threaded and the guard's checks run + // synchronously between await points, so there's no cross-worker race + // on `parallelStats`. Sharing also means the rolling median it tracks + // is computed across every worker's interleaved frames together, + // which is a better signal than per-worker medians would be. + const parallelStats: DeDrainStats = { + verifyChecked: 0, + blankSuspects: 0, + blankDeterministicAccepts: 0, + blankRecaptures: 0, + }; + const parallelGuard = createDrainFrameGuard({ + log, + stats: parallelStats, + frameTime: (i: number) => (i * job.config.fps.den) / job.config.fps.num, + }); + let parallelGuardRan = false; + // First guard/write failure aborts the reorder buffer so peer workers + // parked in waitForFrame reject instead of deadlocking the pool (the + // pool awaits ALL workers before surfacing errors). The original error + // is rethrown below so DrawElementVerificationError keeps its type — + // executeParallelCapture flattens worker errors to strings. + let parallelDrainError: Error | null = null; + const onFrameBuffer = async ( + frameIndex: number, + buffer: Buffer, + workerSession: CaptureSession, + ): Promise => { + try { + if (deParallelStream && workerSession.captureMode === "drawelement") { + parallelGuardRan = true; + buffer = await parallelGuard(workerSession, frameIndex, buffer); + } + await reorderBuffer.waitForFrame(frameIndex); + ensureFrameWritten(await currentEncoder.writeFrame(buffer), frameIndex, currentEncoder); + reorderBuffer.advanceTo(frameIndex + 1); + } catch (err) { + const e = err instanceof Error ? err : new Error(String(err)); + if (!parallelDrainError) { + parallelDrainError = e; + reorderBuffer.abort(e); + } + throw e; + } }; - const workerResults = 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; + let workerResults; + try { + workerResults = 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, - captureCfg, - ); + 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, + // Interleaved DE workers each need their own browser PROCESS: + // pages co-tenant in one browser share one compositor, whose + // internal frame scheduling deprioritizes non-active pages — their + // canvas `paint` events starve and the drawElement paint-wait slows + // ~3x (measured: 86s vs 30s on a 3,245-frame rAF comp). This is + // Chromium's internal frame-production scheduling on ALL platforms, + // NOT the Linux-only HeadlessExperimental.beginFrame capture mode. + deParallelStream ? { ...captureCfg, enableBrowserPool: false } : captureCfg, + ); + } catch (err) { + // Surface the TYPED drain error (DrawElementVerificationError) so the + // orchestrator's verify-retry handler recognizes it — the worker pool + // flattens worker errors into a plain message string. + if (parallelDrainError) throw parallelDrainError; + throw err; + } + if (parallelDrainError) throw parallelDrainError; pushWorkerDedupPerfs(workerResults, dedupPerfs); + if (parallelGuardRan) { + deDrainStats = parallelStats; + } if (probeSession) { captureBeyondViewport = probeSession.options.captureBeyondViewport; diff --git a/packages/producer/src/services/renderOrchestrator.ts b/packages/producer/src/services/renderOrchestrator.ts index 45f4dfebf..4e241748d 100644 --- a/packages/producer/src/services/renderOrchestrator.ts +++ b/packages/producer/src/services/renderOrchestrator.ts @@ -963,6 +963,13 @@ export function shouldUseStreamingEncode( if (outputFormat === "gif") return false; if (!Number.isFinite(durationSeconds) || durationSeconds <= 0) return false; if (durationSeconds > cfg.streamingEncodeMaxDurationSeconds) return false; + // SPIKE (HF_DE_PARALLEL_STREAM): allow multi-worker streaming for the + // interleaved drawElement produce experiment. Contiguous-chunk parallel + // streaming stalls (worker k+1's first frame waits for ALL of worker k's), + // so this only makes sense with the interleaved distribution the capture + // stage selects under the same flag. Explicit opt-in, unverified — do not + // ship default-on without threading guardFrame through onFrameBuffer. + if (process.env.HF_DE_PARALLEL_STREAM === "true") return true; return workerCount === 1; } @@ -1706,7 +1713,10 @@ export async function executeRenderJob( probeSession !== null && probeSession.captureMode !== "drawelement" && !probeSession.deInitDeferred, - experimentalParallelDeOptIn: process.env.PRODUCER_EXPERIMENTAL_FAST_CAPTURE === "true", + experimentalParallelDeOptIn: + process.env.PRODUCER_EXPERIMENTAL_FAST_CAPTURE === "true" || + // Verified parallel DE streaming (opt-in) wants its parallelism kept. + process.env.HF_DE_PARALLEL_STREAM === "true", }); if ( job.config.workers === undefined && @@ -1816,10 +1826,17 @@ export async function executeRenderJob( // disk path (png-sequence / over the streaming duration cap) and parallel // capture ship frames no drain verifies — route those renders to the // screenshot baseline unless drawElement was explicitly opted into. + // HF_DE_PARALLEL_STREAM: multi-worker STREAMING renders now carry the + // full drain-time self-verification (per-worker ground truth + the shared + // drain guard), so the confinement rule is satisfied and the parallel + // clamp does not apply. The disk path stays clamped. + const deParallelStreamVerified = + process.env.HF_DE_PARALLEL_STREAM === "true" && useStreamingEncode && workerCount > 1; if ( cfg.useDrawElement && process.env.PRODUCER_EXPERIMENTAL_FAST_CAPTURE !== "true" && - (!useStreamingEncode || workerCount > 1) + (!useStreamingEncode || workerCount > 1) && + !deParallelStreamVerified ) { cfg.useDrawElement = false; deClampReason = workerCount > 1 ? "parallel" : "disk_path";