fix(engine,producer): close review gaps in parallel drawElement streaming

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).
This commit is contained in:
Vance Ingalls
2026-07-08 16:11:46 -07:00
parent 81a4f04360
commit 2dbe958a49
2 changed files with 27 additions and 1 deletions
@@ -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);
@@ -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,