diff --git a/packages/producer/src/services/render/stages/captureStreamingStage.test.ts b/packages/producer/src/services/render/stages/captureStreamingStage.test.ts index 9c25aa505..0df053e5a 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,118 @@ 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"); + }); + + 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 d94b669d6..5ce2ac01e 100644 --- a/packages/producer/src/services/render/stages/captureStreamingStage.ts +++ b/packages/producer/src/services/render/stages/captureStreamingStage.ts @@ -80,23 +80,76 @@ 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 { + // 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_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. + * + * `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, + signal?: AbortSignal, +): Promise { + return new Promise((resolve, reject) => { + 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); + resolve(value); + }, + (error) => { + clearTimeout(timer); + reject(error); + }, + ); + }); } /** @@ -365,12 +418,23 @@ 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; 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}).`, + abortSignal, + ); + const drainPrev = async (): Promise => { if (!prev) return; // Observe aborts while parked here (the encode wait + ffmpeg write are the @@ -382,6 +446,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 +473,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 +503,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 +531,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 +701,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 +716,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; @@ -768,16 +843,25 @@ export async function runCaptureStreamingStage( onProgress, log, deDrainStats, + abortSignal, ); } 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}).`, + abortSignal, + ); 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;