From 68028bf9adaa233ebee0e078820fbc75ea1bf73f Mon Sep 17 00:00:00 2001 From: miga-heygen Date: Tue, 25 Aug 2026 04:28:29 +0000 Subject: [PATCH] fix(producer): trip DE parallel-router circuit breaker on stalls and hangs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause: the per-worker capture calls in captureFrameRange (parallelCoordinator.ts) take no abort signal of their own, and only checked `signal.aborted` BEFORE starting each frame — a no-op once a worker is already awaiting an in-flight call. On WSL2, the native drawElement/BeginFrame capture call can hang indefinitely at frame 0 with no error. The DE parallel-router's existing stall watchdog (captureStreamingStage.ts) correctly fires `stallController.abort()` after HF_DE_STALL_MS, but that abort had no way to reach a worker already wedged inside a hung capture call — so executeParallelCapture's Promise.all waited forever, the render hung indefinitely, and the CLI's circuit breaker (which only runs after executeRenderJob settles) never got a chance to trip. Fix: race each per-frame capture call against the signal actually firing (raceAgainstAbort), the same "can't cancel, only race" pattern already used by the sequential capture path. Once the watchdog's abort is observed, the wedged worker rejects, executeParallelCapture settles, and the existing pinned-fallback retry / "reverted" outcome / circuit breaker machinery (already correct) runs end to end. Also widen the CLI breaker's trip condition from the literal string "reverted" to "not a clean routed success", so any future non-success outcome the observability layer records also latches the breaker instead of silently falling through. Closes #3441 Co-Authored-By: Miga --- packages/cli/src/commands/render.ts | 23 +++++--- .../parallelCoordinator-peerAbort.test.ts | 53 +++++++++++++++++++ .../src/services/parallelCoordinator.ts | 49 +++++++++++++++-- 3 files changed, 115 insertions(+), 10 deletions(-) diff --git a/packages/cli/src/commands/render.ts b/packages/cli/src/commands/render.ts index 102a45bcc8..f53af70355 100644 --- a/packages/cli/src/commands/render.ts +++ b/packages/cli/src/commands/render.ts @@ -1314,8 +1314,9 @@ function persistDeParallelRouterTrialFired(): boolean { /** * After a trial-armed render, persist that the router's OWN bet actually - * failed — its self-verify/generic-failure safety net fired - * (`deParallelRouter === "reverted"`) — or that the render-count backstop + * failed — its self-verify/generic-failure safety net fired (recorded as + * anything other than a clean `"routed"`, e.g. `"reverted"`, or a stall/hang + * outcome — heygen-com/hyperframes#3441) — or that the render-count backstop * (`DE_PARALLEL_ROUTER_TRIAL_MAX_RENDERS`) was reached, so it's never * enabled again for this install. A clean "routed" (the render succeeded * with no fallback) does NOT consume the trial by itself — the whole point @@ -1352,11 +1353,19 @@ function maybeConsumeDeParallelRouterTrial( const config = readConfigFresh(); const renderCount = (config.deParallelRouterTrialRenderCount ?? 0) + 1; config.deParallelRouterTrialRenderCount = renderCount; - // Trip ONLY on an actual fallback. The old trial also tripped at a - // 25-render exposure cap, which was sampling logic: bound how long an - // experiment force-enables itself. Under a shipped default that would - // switch the feature off behind the user's back after 25 good renders. - const fired = outcome === "reverted"; + // Trip on any recorded non-success, not only the literal string + // "reverted". The old trial also tripped at a 25-render exposure cap, + // which was sampling logic: bound how long an experiment force-enables + // itself. Under a shipped default that would switch the feature off + // behind the user's back after 25 good renders — so a clean "routed" (no + // fallback needed) must NOT trip. But narrowing the positive check to the + // single string "reverted" (heygen-com/hyperframes#3441) meant any other + // non-success signal the observability layer might ever record — a stall, + // a timeout, a future outcome value — would silently fall through to "not + // fired" instead of tripping. `outcome` is `undefined`-filtered above, so + // by this point it is a real recorded outcome; the only one that means + // "no fallback happened" is "routed" itself. + const fired = outcome !== "routed"; if (fired) { config.deParallelRouterTrialFired = true; // Latch BEFORE attempting persistence — the decision holds for this diff --git a/packages/engine/src/services/parallelCoordinator-peerAbort.test.ts b/packages/engine/src/services/parallelCoordinator-peerAbort.test.ts index 5c82440a02..c389a80f53 100644 --- a/packages/engine/src/services/parallelCoordinator-peerAbort.test.ts +++ b/packages/engine/src/services/parallelCoordinator-peerAbort.test.ts @@ -61,4 +61,57 @@ describe("executeParallelCapture peer abort", () => { rmSync(root, { recursive: true, force: true }); } }); + + // heygen-com/hyperframes#3441: a worker wedged INSIDE a native capture call + // (WSL2 hangs the very first drawElement/BeginFrame call at frame 0 with no + // error) must actually be unstuck once the caller's `signal` aborts — e.g. + // the DE parallel-router stall watchdog in `captureStreamingStage.ts` firing + // after HF_DE_STALL_MS. Before this fix, `captureFrameRange` only checked + // `signal.aborted` BEFORE starting each frame's capture call, which is a + // no-op for a call that is already in flight and never settles — the abort + // had no way to reach it, so `executeParallelCapture`'s `Promise.all` (and + // therefore the whole render) hung forever, and the CLI circuit breaker + // (which only runs after `executeRenderJob` settles) never got a chance to + // trip. + it("rejects promptly when the signal aborts while a worker is wedged inside a capture call that never settles", async () => { + const root = mkdtempSync(join(tmpdir(), "hf-stall-abort-")); + // Simulates the native capture call hanging indefinitely (never resolves, + // never rejects) — exactly the WSL2 shape from the field report. + const captureFrame = vi.fn(() => new Promise(() => {})); + const closeCaptureSession = vi.fn().mockResolvedValue(undefined); + vi.doMock("./frameCapture.js", () => ({ + createCaptureSession: vi.fn( + async () => ({ workerId: 0, browserConsoleBuffer: [] }) as unknown as CaptureSession, + ), + initializeSession: vi.fn(async () => {}), + captureFrame, + captureFrameToBuffer: vi.fn(), + captureFrameToBufferPipelined: vi.fn(), + closeCaptureSession, + getCapturePerfSummary: vi.fn(() => ({ frames: 0 })), + })); + + try { + const { executeParallelCapture } = await import("./parallelCoordinator.js"); + const controller = new AbortController(); + const result = executeParallelCapture( + "http://127.0.0.1", + root, + [{ workerId: 0, startFrame: 0, endFrame: 3, outputDir: join(root, "worker-0") }], + { width: 320, height: 180, fps: { num: 30, den: 1 } }, + () => null, + controller.signal, + ); + + // Give the worker a tick to actually enter the (never-resolving) + // capture call before simulating the watchdog's abort. + await new Promise((resolve) => setTimeout(resolve, 10)); + controller.abort(); + + await expect(result).rejects.toMatchObject({ name: "CaptureFailure" }); + expect(captureFrame).toHaveBeenCalledTimes(1); + } finally { + rmSync(root, { recursive: true, force: true }); + } + }); }); diff --git a/packages/engine/src/services/parallelCoordinator.ts b/packages/engine/src/services/parallelCoordinator.ts index 35cb06d395..41a641aca5 100644 --- a/packages/engine/src/services/parallelCoordinator.ts +++ b/packages/engine/src/services/parallelCoordinator.ts @@ -461,6 +461,43 @@ export function shouldVerifyWorkerGpu(workerId: number, config?: Partial(promise: Promise, signal: AbortSignal | undefined): Promise { + if (!signal) return promise; + if (signal.aborted) return Promise.reject(new Error("Parallel worker cancelled")); + return new Promise((resolve, reject) => { + const onAbort = () => reject(new Error("Parallel worker cancelled")); + signal.addEventListener("abort", onAbort, { once: true }); + promise.then( + (value) => { + signal.removeEventListener("abort", onAbort); + resolve(value); + }, + (error) => { + signal.removeEventListener("abort", onAbort); + reject(error); + }, + ); + }); +} + // fallow-ignore-next-line complexity async function captureFrameRange( session: CaptureSession, @@ -498,7 +535,10 @@ async function captureFrameRange( 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); + const { encodeResult } = await raceAgainstAbort( + captureFrameToBufferPipelined(session, i - outputOffset, time), + signal, + ); // 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 @@ -542,10 +582,13 @@ async function captureFrameRange( const fileFrameIdx = i - outputOffset; if (onFrameBuffer) { - const { buffer } = await captureFrameToBuffer(session, fileFrameIdx, time); + const { buffer } = await raceAgainstAbort( + captureFrameToBuffer(session, fileFrameIdx, time), + signal, + ); await onFrameBuffer(i, buffer, session); } else { - await captureFrame(session, fileFrameIdx, time); + await raceAgainstAbort(captureFrame(session, fileFrameIdx, time), signal); } framesCaptured++; if (onFrameCaptured) onFrameCaptured(task.workerId, i);