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);