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;