From 650977d1d8a695203047c3020c8a61465ed55bb8 Mon Sep 17 00:00:00 2001 From: Vance Ingalls Date: Wed, 15 Jul 2026 01:53:58 -0700 Subject: [PATCH 1/2] fix(producer): extend DE stall watchdog to the single-worker streaming path --- .../stages/captureStreamingStage.test.ts | 85 ++++++++++++++--- .../render/stages/captureStreamingStage.ts | 94 +++++++++++++++---- 2 files changed, 148 insertions(+), 31 deletions(-) diff --git a/packages/producer/src/services/render/stages/captureStreamingStage.test.ts b/packages/producer/src/services/render/stages/captureStreamingStage.test.ts index 9c25aa505..61bbec32a 100644 --- a/packages/producer/src/services/render/stages/captureStreamingStage.test.ts +++ b/packages/producer/src/services/render/stages/captureStreamingStage.test.ts @@ -17,6 +17,8 @@ const spawnStreamingEncoder = mock(async () => ({ let failCaptureFrameToBuffer = false; let failInitializeSession = false; let hangParallelUntilAbort = false; +let hangSequentialUntilStall = false; +let sessionWorkerEncodeEnabled = false; let initializeSessionErrorMessage = "initialize failed"; const browserConsoleBuffer = ["[FrameCapture:ERROR] page.goto failed"]; const closeCaptureSession = mock(async () => {}); @@ -26,12 +28,25 @@ mock.module("@hyperframes/engine", () => ({ calculateOptimalWorkers: () => 1, convertTransfer: () => {}, captureFrame: async () => {}, - captureFrameToBufferPipelined: async () => ({ encodeResult: { buffer: Buffer.from("frame") } }), - captureFramesBatchPipelined: async () => [], + captureFrameToBufferPipelined: async () => { + if (hangSequentialUntilStall) { + return new Promise(() => {}); + } + return { encodeResult: Promise.resolve(Buffer.from("frame")) }; + }, + captureFramesBatchPipelined: async () => { + if (hangSequentialUntilStall) { + return new Promise(() => {}); + } + return []; + }, captureFrameToBuffer: async () => { if (failCaptureFrameToBuffer) { throw new Error("captureFrameToBuffer failed"); } + if (hangSequentialUntilStall) { + return new Promise(() => {}); + } return { buffer: Buffer.from("frame"), captureTimeMs: 1 }; }, closeCaptureSession, @@ -40,7 +55,7 @@ mock.module("@hyperframes/engine", () => ({ isInitialized: false, browserConsoleBuffer, options: { captureBeyondViewport: false }, - workerEncodeEnabled: false, + workerEncodeEnabled: sessionWorkerEncodeEnabled, }), createFrameReorderBuffer: () => ({ waitForFrame: async () => {}, @@ -213,8 +228,8 @@ describe("runCaptureStreamingStage", () => { it("trips the stall watchdog and rethrows a non-cancellation error when the parallel path makes no frame progress", async () => { hangParallelUntilAbort = true; - const prev = process.env.HF_DE_PARALLEL_STALL_MS; - process.env.HF_DE_PARALLEL_STALL_MS = "50"; + const prev = process.env.HF_DE_STALL_MS; + process.env.HF_DE_STALL_MS = "50"; const { runCaptureStreamingStage } = await import("./captureStreamingStage.js"); const cfg = { forceScreenshot: false, ffmpegStreamingTimeout: 3_600_000 }; const input = { @@ -231,8 +246,8 @@ describe("runCaptureStreamingStage", () => { caught = error; } finally { hangParallelUntilAbort = false; - if (prev === undefined) delete process.env.HF_DE_PARALLEL_STALL_MS; - else process.env.HF_DE_PARALLEL_STALL_MS = prev; + if (prev === undefined) delete process.env.HF_DE_STALL_MS; + else process.env.HF_DE_STALL_MS = prev; } expect(caught).toBeInstanceOf(Error); @@ -245,9 +260,9 @@ describe("runCaptureStreamingStage", () => { it("does not relabel a genuine parent-abort as a stall", async () => { hangParallelUntilAbort = true; - const prev = process.env.HF_DE_PARALLEL_STALL_MS; + const prev = process.env.HF_DE_STALL_MS; // Huge window so the watchdog never trips; the parent abort is what ends it. - process.env.HF_DE_PARALLEL_STALL_MS = "600000"; + process.env.HF_DE_STALL_MS = "600000"; const controller = new AbortController(); const { runCaptureStreamingStage } = await import("./captureStreamingStage.js"); const cfg = { forceScreenshot: false, ffmpegStreamingTimeout: 3_600_000 }; @@ -267,12 +282,60 @@ describe("runCaptureStreamingStage", () => { await run; hangParallelUntilAbort = false; - if (prev === undefined) delete process.env.HF_DE_PARALLEL_STALL_MS; - else process.env.HF_DE_PARALLEL_STALL_MS = prev; + if (prev === undefined) delete process.env.HF_DE_STALL_MS; + else process.env.HF_DE_STALL_MS = prev; expect(caught).toBeInstanceOf(Error); expect((caught as Error).message).not.toContain("stalled"); }); + + it("trips the stall watchdog on the single-worker worker-encode pipeline when capture makes no progress", async () => { + hangSequentialUntilStall = true; + sessionWorkerEncodeEnabled = true; + const prev = process.env.HF_DE_STALL_MS; + process.env.HF_DE_STALL_MS = "50"; + const { runCaptureStreamingStage } = await import("./captureStreamingStage.js"); + const cfg = { forceScreenshot: false, ffmpegStreamingTimeout: 3_600_000 }; + const input = { ...createInput(cfg), totalFrames: 10, workerCount: 1 }; + + let caught: unknown; + try { + await runCaptureStreamingStage(input); + } catch (error) { + caught = error; + } finally { + hangSequentialUntilStall = false; + sessionWorkerEncodeEnabled = false; + if (prev === undefined) delete process.env.HF_DE_STALL_MS; + else process.env.HF_DE_STALL_MS = prev; + } + + expect(caught).toBeInstanceOf(Error); + expect((caught as Error).message).toContain("stalled"); + }); + + it("trips the stall watchdog on the single-worker plain capture loop when capture makes no progress", async () => { + hangSequentialUntilStall = true; + const prev = process.env.HF_DE_STALL_MS; + process.env.HF_DE_STALL_MS = "50"; + const { runCaptureStreamingStage } = await import("./captureStreamingStage.js"); + const cfg = { forceScreenshot: false, ffmpegStreamingTimeout: 3_600_000 }; + const input = { ...createInput(cfg), totalFrames: 10, workerCount: 1 }; + + let caught: unknown; + try { + await runCaptureStreamingStage(input); + } catch (error) { + caught = error; + } finally { + hangSequentialUntilStall = false; + if (prev === undefined) delete process.env.HF_DE_STALL_MS; + else process.env.HF_DE_STALL_MS = prev; + } + + expect(caught).toBeInstanceOf(Error); + expect((caught as Error).message).toContain("stalled"); + }); }); describe("runCaptureStage", () => { diff --git a/packages/producer/src/services/render/stages/captureStreamingStage.ts b/packages/producer/src/services/render/stages/captureStreamingStage.ts index d94b669d6..41b3e3ea2 100644 --- a/packages/producer/src/services/render/stages/captureStreamingStage.ts +++ b/packages/producer/src/services/render/stages/captureStreamingStage.ts @@ -80,23 +80,50 @@ import { ensureFrameWritten } from "./captureHdrFrameShared.js"; import { updateJobStatus } from "../shared.js"; /** - * No-frame-progress watchdog for the parallel DE streaming path. The router - * auto-enables this path for the ≥24GB macOS trial cohort, so a worker that - * wedges mid-capture (a hung seek/screenshot at an early frame) would - * otherwise sit until the per-frame CDP `protocolTimeout` (~5 min) fires — - * a silent multi-minute hang that only THEN reaches the pinned fallback. - * Trip well before that: if no NEW frame lands within this window, abort the - * pool so the orchestrator re-renders via screenshot. Default 60s ≫ any real - * per-frame budget (15–32 ms), so a legit slow frame won't false-trip; a - * false trip only costs the (slower, never-wrong) screenshot fallback. + * No-frame-progress watchdog for DE streaming capture. A worker (parallel + * path) or the single in-flight capture (sequential path, worker-encode or + * plain) can wedge mid-capture (a hung seek/screenshot at an early frame), + * which would otherwise sit until the per-frame CDP `protocolTimeout` + * (~5 min) fires — a silent multi-minute hang that only THEN reaches the + * pinned fallback. Trip well before that: if no NEW frame lands within this + * window, fail fast so the orchestrator re-renders via screenshot. Default + * 60s ≫ any real per-frame budget (15–32 ms), so a legit slow frame won't + * false-trip; a false trip only costs the (slower, never-wrong) screenshot + * fallback. */ -const DEFAULT_DE_PARALLEL_STALL_MS = 60_000; -const DE_PARALLEL_STALL_POLL_MS = 5_000; +const DEFAULT_DE_STALL_MS = 60_000; +const DE_STALL_POLL_MS = 5_000; -function resolveParallelStallTimeoutMs(): number { - const raw = process.env.HF_DE_PARALLEL_STALL_MS; +function resolveDeStallTimeoutMs(): number { + const raw = process.env.HF_DE_STALL_MS; const parsed = raw ? Number(raw) : Number.NaN; - return Number.isFinite(parsed) && parsed > 0 ? parsed : DEFAULT_DE_PARALLEL_STALL_MS; + return Number.isFinite(parsed) && parsed > 0 ? parsed : DEFAULT_DE_STALL_MS; +} + +/** + * Race a single sequential capture call against a stall deadline. Unlike the + * parallel path (which threads an AbortSignal into executeParallelCapture), + * captureFrameToBuffer/captureFrameToBufferPipelined/captureFramesBatchPipelined + * take no signal — a wedged call can't be cancelled, only raced. A tripped + * guard abandons the in-flight capture (same "orphaned, never awaited" + * contract as the worker-encode pipeline's encodeResult) and rejects so the + * caller fails fast to the pinned screenshot fallback instead of waiting out + * the ~5min CDP protocol timeout. + */ +function raceAgainstStall(promise: Promise, deadlineMs: number, message: string): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error(message)), Math.max(0, deadlineMs)); + promise.then( + (value) => { + clearTimeout(timer); + resolve(value); + }, + (error) => { + clearTimeout(timer); + reject(error); + }, + ); + }); } /** @@ -371,6 +398,15 @@ async function runWorkerEncodePipelineLoop( const guard = createDrainFrameGuard({ log, stats, frameTime }); const guardFrame = (idx: number, buf: Buffer): Promise => guard(session, idx, buf); + const stallTimeoutMs = resolveDeStallTimeoutMs(); + let lastProgressAt = Date.now(); + const captureWithStallGuard = (idx: number, promise: Promise): Promise => + raceAgainstStall( + promise, + stallTimeoutMs - (Date.now() - lastProgressAt), + `[Render] Sequential drawElement capture stalled: no frame progress for ${stallTimeoutMs}ms (stuck at frame ${idx}/${totalFrames}).`, + ); + const drainPrev = async (): Promise => { if (!prev) return; // Observe aborts while parked here (the encode wait + ffmpeg write are the @@ -382,6 +418,7 @@ async function runWorkerEncodePipelineLoop( ensureFrameWritten(await currentEncoder.writeFrame(buf), prev.idx, currentEncoder); reorderBuffer.advanceTo(prev.idx + 1); job.framesRendered = prev.idx + 1; + lastProgressAt = Date.now(); updateJobStatus( job, "rendering", @@ -408,6 +445,7 @@ async function runWorkerEncodePipelineLoop( ensureFrameWritten(await currentEncoder.writeFrame(buf), item.idx, currentEncoder); reorderBuffer.advanceTo(item.idx + 1); job.framesRendered = item.idx + 1; + lastProgressAt = Date.now(); updateJobStatus( job, "rendering", @@ -437,11 +475,17 @@ async function runWorkerEncodePipelineLoop( // pixels on a deterministic comp (87dB vs ∞ noise floor — main-thread // contention shifting a paint-wait to the timeout path). Keep the // drain strictly between batch evaluates. - const results = await captureFramesBatchPipelined(session, idxs, idxs.map(frameTime)); + const results = await captureWithStallGuard( + idxs[0] ?? i, + captureFramesBatchPipelined(session, idxs, idxs.map(frameTime)), + ); await drainBatch(prevBatch); prevBatch = results.map((r) => ({ idx: r.frameIndex, encodeResult: r.encodeResult })); } else { - const { encodeResult } = await captureFrameToBufferPipelined(session, i, frameTime(i)); + const { encodeResult } = await captureWithStallGuard( + i, + captureFrameToBufferPipelined(session, i, frameTime(i)), + ); await drainBatch(prevBatch); prevBatch = [{ idx: i, encodeResult }]; i++; @@ -459,7 +503,10 @@ async function runWorkerEncodePipelineLoop( for (let i = 0; i < totalFrames; i++) { assertNotAborted(); const time = frameTime(i); - const { encodeResult } = await captureFrameToBufferPipelined(session, i, time); + const { encodeResult } = await captureWithStallGuard( + i, + captureFrameToBufferPipelined(session, i, time), + ); await drainPrev(); prev = { idx: i, encodeResult }; } @@ -626,7 +673,7 @@ export async function runCaptureStreamingStage( if (abortSignal.aborted) stallController.abort(); else abortSignal.addEventListener("abort", forwardParentAbort, { once: true }); } - const stallTimeoutMs = resolveParallelStallTimeoutMs(); + const stallTimeoutMs = resolveDeStallTimeoutMs(); let lastCapturedFrames = 0; let lastProgressAt = Date.now(); let stalled = false; @@ -641,7 +688,7 @@ export async function runCaptureStreamingStage( reorderBuffer.abort(stallErr); stallController.abort(); }, - Math.min(stallTimeoutMs, DE_PARALLEL_STALL_POLL_MS), + Math.min(stallTimeoutMs, DE_STALL_POLL_MS), ); let workerResults; @@ -770,14 +817,21 @@ export async function runCaptureStreamingStage( deDrainStats, ); } else { + const stallTimeoutMs = resolveDeStallTimeoutMs(); + let lastProgressAt = Date.now(); for (let i = 0; i < totalFrames; i++) { assertNotAborted(); const time = (i * job.config.fps.den) / job.config.fps.num; - const { buffer } = await captureFrameToBuffer(session, i, time); + const { buffer } = await raceAgainstStall( + captureFrameToBuffer(session, i, time), + stallTimeoutMs - (Date.now() - lastProgressAt), + `[Render] Sequential drawElement capture stalled: no frame progress for ${stallTimeoutMs}ms (stuck at frame ${i}/${totalFrames}).`, + ); await reorderBuffer.waitForFrame(i); ensureFrameWritten(await currentEncoder.writeFrame(buffer), i, currentEncoder); reorderBuffer.advanceTo(i + 1); job.framesRendered = i + 1; + lastProgressAt = Date.now(); const frameProgress = (i + 1) / totalFrames; const progress = 25 + frameProgress * 55; From b70d849a4728ee08d070ac49cd6e6af4b23f4a2e Mon Sep 17 00:00:00 2001 From: Vance Ingalls Date: Wed, 15 Jul 2026 11:53:03 -0700 Subject: [PATCH 2/2] fix(producer): bridge stall-timeout env var rename, disambiguate abort from stall on sequential path Review feedback on #2473: HF_DE_PARALLEL_STALL_MS had no backwards-compat shim after the rename to HF_DE_STALL_MS, and a parent abort during a wedged sequential capture would surface as "stalled" instead of "aborted" in downstream logs/telemetry (functionally harmless since isCancellation is gated on abortSignal.aborted, not message text, but misleading to read). --- .../stages/captureStreamingStage.test.ts | 58 +++++++++++++++++++ .../render/stages/captureStreamingStage.ts | 36 +++++++++++- 2 files changed, 91 insertions(+), 3 deletions(-) diff --git a/packages/producer/src/services/render/stages/captureStreamingStage.test.ts b/packages/producer/src/services/render/stages/captureStreamingStage.test.ts index 61bbec32a..0df053e5a 100644 --- a/packages/producer/src/services/render/stages/captureStreamingStage.test.ts +++ b/packages/producer/src/services/render/stages/captureStreamingStage.test.ts @@ -336,6 +336,64 @@ describe("runCaptureStreamingStage", () => { expect(caught).toBeInstanceOf(Error); expect((caught as Error).message).toContain("stalled"); }); + + it("still honors the pre-rename HF_DE_PARALLEL_STALL_MS env var for one release", async () => { + hangSequentialUntilStall = true; + const prevNew = process.env.HF_DE_STALL_MS; + const prevOld = process.env.HF_DE_PARALLEL_STALL_MS; + delete process.env.HF_DE_STALL_MS; + process.env.HF_DE_PARALLEL_STALL_MS = "50"; + const { runCaptureStreamingStage } = await import("./captureStreamingStage.js"); + const cfg = { forceScreenshot: false, ffmpegStreamingTimeout: 3_600_000 }; + const input = { ...createInput(cfg), totalFrames: 10, workerCount: 1 }; + + let caught: unknown; + try { + await runCaptureStreamingStage(input); + } catch (error) { + caught = error; + } finally { + hangSequentialUntilStall = false; + if (prevNew === undefined) delete process.env.HF_DE_STALL_MS; + else process.env.HF_DE_STALL_MS = prevNew; + if (prevOld === undefined) delete process.env.HF_DE_PARALLEL_STALL_MS; + else process.env.HF_DE_PARALLEL_STALL_MS = prevOld; + } + + expect(caught).toBeInstanceOf(Error); + expect((caught as Error).message).toContain("stalled"); + }); + + it("does not relabel a genuine parent-abort as a stall on the sequential path", async () => { + hangSequentialUntilStall = true; + const prev = process.env.HF_DE_STALL_MS; + process.env.HF_DE_STALL_MS = "50"; + const controller = new AbortController(); + controller.abort(); + const { runCaptureStreamingStage } = await import("./captureStreamingStage.js"); + const cfg = { forceScreenshot: false, ffmpegStreamingTimeout: 3_600_000 }; + const input = { + ...createInput(cfg), + totalFrames: 10, + workerCount: 1, + abortSignal: controller.signal, + }; + + let caught: unknown; + try { + await runCaptureStreamingStage(input); + } catch (error) { + caught = error; + } finally { + hangSequentialUntilStall = false; + if (prev === undefined) delete process.env.HF_DE_STALL_MS; + else process.env.HF_DE_STALL_MS = prev; + } + + expect(caught).toBeInstanceOf(Error); + expect((caught as Error).message).not.toContain("stalled"); + expect((caught as Error).message).toContain("aborted"); + }); }); describe("runCaptureStage", () => { diff --git a/packages/producer/src/services/render/stages/captureStreamingStage.ts b/packages/producer/src/services/render/stages/captureStreamingStage.ts index 41b3e3ea2..5ce2ac01e 100644 --- a/packages/producer/src/services/render/stages/captureStreamingStage.ts +++ b/packages/producer/src/services/render/stages/captureStreamingStage.ts @@ -95,7 +95,11 @@ const DEFAULT_DE_STALL_MS = 60_000; const DE_STALL_POLL_MS = 5_000; function resolveDeStallTimeoutMs(): number { - const raw = process.env.HF_DE_STALL_MS; + // HF_DE_PARALLEL_STALL_MS is the pre-rename name (this config used to guard + // only the parallel path). Bridged for one release so an already-deployed + // ops surface (runbook, ConfigMap, ...) tuning the old name doesn't + // silently no-op; drop once nothing sets it anymore. + const raw = process.env.HF_DE_STALL_MS ?? process.env.HF_DE_PARALLEL_STALL_MS; const parsed = raw ? Number(raw) : Number.NaN; return Number.isFinite(parsed) && parsed > 0 ? parsed : DEFAULT_DE_STALL_MS; } @@ -109,10 +113,32 @@ function resolveDeStallTimeoutMs(): number { * contract as the worker-encode pipeline's encodeResult) and rejects so the * caller fails fast to the pinned screenshot fallback instead of waiting out * the ~5min CDP protocol timeout. + * + * `signal` is read only at trip time to label the rejection, never to cancel + * the race early — a parent abort during a wedge still has to wait out the + * same deadline (nothing can unstick the underlying call), but the message + * must say "aborted", not "stalled", so downstream logs/telemetry don't + * misreport a deliberate cancellation as a capture failure. */ -function raceAgainstStall(promise: Promise, deadlineMs: number, message: string): Promise { +function raceAgainstStall( + promise: Promise, + deadlineMs: number, + message: string, + signal?: AbortSignal, +): Promise { return new Promise((resolve, reject) => { - const timer = setTimeout(() => reject(new Error(message)), Math.max(0, deadlineMs)); + const timer = setTimeout( + () => { + reject( + new Error( + signal?.aborted + ? "[Render] Sequential drawElement capture aborted while a capture call was in flight." + : message, + ), + ); + }, + Math.max(0, deadlineMs), + ); promise.then( (value) => { clearTimeout(timer); @@ -392,6 +418,7 @@ async function runWorkerEncodePipelineLoop( onProgress: CaptureStreamingStageInput["onProgress"], log: CaptureStreamingStageInput["log"], stats: DeDrainStats, + abortSignal: AbortSignal | undefined, ): Promise { let prev: { idx: number; encodeResult: Promise } | null = null; const frameTime = (i: number) => (i * job.config.fps.den) / job.config.fps.num; @@ -405,6 +432,7 @@ async function runWorkerEncodePipelineLoop( promise, stallTimeoutMs - (Date.now() - lastProgressAt), `[Render] Sequential drawElement capture stalled: no frame progress for ${stallTimeoutMs}ms (stuck at frame ${idx}/${totalFrames}).`, + abortSignal, ); const drainPrev = async (): Promise => { @@ -815,6 +843,7 @@ export async function runCaptureStreamingStage( onProgress, log, deDrainStats, + abortSignal, ); } else { const stallTimeoutMs = resolveDeStallTimeoutMs(); @@ -826,6 +855,7 @@ export async function runCaptureStreamingStage( captureFrameToBuffer(session, i, time), stallTimeoutMs - (Date.now() - lastProgressAt), `[Render] Sequential drawElement capture stalled: no frame progress for ${stallTimeoutMs}ms (stuck at frame ${i}/${totalFrames}).`, + abortSignal, ); await reorderBuffer.waitForFrame(i); ensureFrameWritten(await currentEncoder.writeFrame(buffer), i, currentEncoder);