From 2dbe958a49baa08319d091d36795edd51afc3988 Mon Sep 17 00:00:00 2001
From: Vance Ingalls <vance@heygen.com>
Date: Wed, 8 Jul 2026 15:24:33 -0700
Subject: [PATCH] fix(engine,producer): close review gaps in parallel
 drawElement streaming
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Address PR #2056 review feedback:
- Fix totalFrames progress inflation for interleaved tasks — divide
  each task's span by its frameStride to match the actual per-worker
  frame count (captureFrameRange steps by stride), instead of summing
  raw endFrame-startFrame which double(N)-counts interleaved tasks.
- Attach a no-op .catch to each frame's pipelined encodeResult at kick
  time so an abandoned promise (loop exits early on abort/error before
  draining it) can't surface as an unhandled rejection during teardown.
- Document why the pipelined branch's stride=1 path is validation-only
  in production (HF_DE_PARALLEL_STREAM always uses interleaved
  distribution) so a future refactor doesn't unknowingly widen it.
- Comment the intentional single shared parallelGuard/parallelStats
  across workers (safe single-threaded, better rolling-median signal).
---
 .../src/services/parallelCoordinator.ts       | 22 ++++++++++++++++++-
 .../render/stages/captureStreamingStage.ts    |  6 +++++
 2 files changed, 27 insertions(+), 1 deletion(-)

diff --git a/packages/engine/src/services/parallelCoordinator.ts b/packages/engine/src/services/parallelCoordinator.ts
index 2493d2507..f2b766a49 100644
--- a/packages/engine/src/services/parallelCoordinator.ts
+++ b/packages/engine/src/services/parallelCoordinator.ts
@@ -300,6 +300,12 @@ async function captureFrameRange(
   // 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();
@@ -312,6 +318,12 @@ async function captureFrameRange(
         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`);
       }
@@ -458,7 +470,15 @@ export async function executeParallelCapture(
   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);
diff --git a/packages/producer/src/services/render/stages/captureStreamingStage.ts b/packages/producer/src/services/render/stages/captureStreamingStage.ts
index 7fd5c526b..e22d2a3e9 100644
--- a/packages/producer/src/services/render/stages/captureStreamingStage.ts
+++ b/packages/producer/src/services/render/stages/captureStreamingStage.ts
@@ -525,6 +525,12 @@ export async function runCaptureStreamingStage(
       // 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,