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,35 +52,48 @@ 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 });
     }
   };
 
   const flushAt = (frame: number): void => {
     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 };
 }
 
 // ---------------------------------------------------------------------------