Merge pull request #2056 from heygen-com/de2-15-parallel-de-verified

feat(engine,producer): verified interleaved parallel drawElement streaming (opt-in)

packages/engine/src/index.ts

@@ -177,6 +177,7 @@ export type {
 export {
   calculateOptimalWorkers,
   distributeFrames,
+  distributeFramesInterleaved,
   executeParallelCapture,
   mergeWorkerFrames,
   getSystemResources,

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<Buffer>;
+  /** 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
@@ -2657,9 +2662,28 @@ 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;
+      // Advance the watermark on reuse too, not just on real captures — the
+      // gap-check above starts from lastEncodeResultFrame, so leaving it
+      // pinned to the last REAL capture makes every consecutive reuse rescan
+      // an ever-widening window instead of just the one new frame (review).
+      session.lastEncodeResultFrame = frameIndex;
+      return { encodeResult: session.lastEncodeResult, captureTimeMs: Date.now() - startTime };
+    }
   }
 
   try {
@@ -2694,7 +2718,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 };
     }
 
@@ -2721,7 +2748,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) {

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<EngineC
   return config?.browserGpuMode === "software" && workerId === 0;
 }
 
+// fallow-ignore-next-line complexity
 async function captureFrameRange(
   session: CaptureSession,
   task: WorkerTask,
   captureOptions: CaptureOptions,
   signal: AbortSignal | undefined,
   onFrameCaptured: ((workerId: number, frameIndex: number) => void) | undefined,
-  onFrameBuffer: ((frameIndex: number, buffer: Buffer) => Promise<void>) | undefined,
+  onFrameBuffer:
+    | ((frameIndex: number, buffer: Buffer, session: CaptureSession) => Promise<void>)
+    | undefined,
 ): Promise<number> {
   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<Buffer> } | 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<void>,
+  onFrameBuffer?: (frameIndex: number, buffer: Buffer, session: CaptureSession) => Promise<void>,
   config?: Partial<EngineConfig>,
   parallel?: boolean,
 ): Promise<WorkerResult> {
@@ -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<void>,
+  onFrameBuffer?: (frameIndex: number, buffer: Buffer, session: CaptureSession) => Promise<void>,
   config?: Partial<EngineConfig>,
 ): Promise<WorkerResult[]> {
-  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<number, number>();
 
   for (const task of tasks) workerProgress.set(task.workerId, 0);

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();
+  });
+});

packages/engine/src/services/streamingEncoder.ts

@@ -52,18 +52,27 @@ export interface FrameReorderBuffer {
   waitForFrame: (frame: number) => Promise<void>;
   advanceTo: (frame: number) => void;
   waitForAllDone: () => Promise<void>;
+  /**
+   * Reject every parked and future waiter with `err`. Required by the
+   * interleaved parallel drain: when one worker fails (e.g. drawElement
+   * self-verification), its frames will never be written — peers parked in
+   * waitForFrame would otherwise deadlock the whole capture (the worker pool
+   * awaits ALL workers before surfacing the failure).
+   */
+  abort: (err: Error) => void;
 }
 
 export function createFrameReorderBuffer(startFrame: number, endFrame: number): FrameReorderBuffer {
   let cursor = startFrame;
-  const pending = new Map<number, Array<() => void>>();
+  let aborted: Error | null = null;
+  const pending = new Map<number, Array<{ resolve: () => 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<void> =>
-    new Promise<void>((resolve) => {
+    new Promise<void>((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<void> =>
-    new Promise<void>((resolve) => {
+    new Promise<void>((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 };
 }
 
 // ---------------------------------------------------------------------------

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<number> {
   }
 }
 
-async function runWorkerEncodePipelineLoop(
-  session: CaptureSession,
-  totalFrames: number,
-  job: CaptureStreamingStageInput["job"],
-  currentEncoder: StreamingEncoder,
-  reorderBuffer: ReturnType<typeof createFrameReorderBuffer>,
-  assertNotAborted: () => void,
-  onProgress: CaptureStreamingStageInput["onProgress"],
-  log: CaptureStreamingStageInput["log"],
-  stats: DeDrainStats,
-): Promise<void> {
-  let prev: { idx: number; encodeResult: Promise<Buffer> } | 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<Buffer> {
+  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<Buffer> => {
+  return async (session: CaptureSession, idx: number, buf: Buffer): Promise<Buffer> => {
     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<typeof createFrameReorderBuffer>,
+  assertNotAborted: () => void,
+  onProgress: CaptureStreamingStageInput["onProgress"],
+  log: CaptureStreamingStageInput["log"],
+  stats: DeDrainStats,
+): Promise<void> {
+  let prev: { idx: number; encodeResult: Promise<Buffer> } | 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<Buffer> => guard(session, idx, buf);
 
   const drainPrev = async (): Promise<void> => {
     if (!prev) return;
@@ -494,43 +507,127 @@ 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<void> => {
-        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, so the rolling median it tracks is computed across every
+      // worker's interleaved frames together — a better signal than
+      // per-worker medians would be. This IS touched from concurrent workers
+      // across real await points (recapture, PSNR), so the shared
+      // `sizes`/`absFloor`/`acceptedSmall` state can interleave — safe by
+      // construction rather than by single-threadedness: `absFloor` only
+      // ratchets down (order-independent min), `sizes` is append-only (order
+      // doesn't affect the median once ≥12 samples exist), and
+      // `acceptedSmall`'s fast path only ever fires on exact byte-equality
+      // against a buffer some worker already re-verified deterministic — so
+      // whichever worker's buffer lands there, the equality check itself is
+      // what re-validates it, not which worker wrote it (review).
+      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<void> => {
+        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;

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";