From e2edc93f5f4127fd44d4fcfd6534db7bb0d8d4aa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miguel=20=C3=81ngel?= Date: Tue, 3 Nov 2026 23:11:32 +0000 Subject: [PATCH] fix(producer): a laptop that sleeps mid-render no longer counts the sleep as a capture stall (#7096) * fix(producer): a laptop that sleeps mid-render no longer counts the sleep as a capture stall * style(producer): format the stall watchdog timers --- .../src/services/render/observability.test.ts | 37 ++++++++ .../src/services/render/observability.ts | 41 ++++---- .../stages/captureStreamingStage.test.ts | 76 +++++++++++++++ .../render/stages/captureStreamingStage.ts | 94 ++++++++++--------- .../src/services/render/stallClock.test.ts | 29 ++++++ .../src/services/render/stallClock.ts | 26 +++++ 6 files changed, 238 insertions(+), 65 deletions(-) create mode 100644 packages/producer/src/services/render/stallClock.test.ts create mode 100644 packages/producer/src/services/render/stallClock.ts diff --git a/packages/producer/src/services/render/observability.test.ts b/packages/producer/src/services/render/observability.test.ts index e1fee535d4..0b43363a09 100644 --- a/packages/producer/src/services/render/observability.test.ts +++ b/packages/producer/src/services/render/observability.test.ts @@ -215,6 +215,43 @@ describe("RenderObservabilityRecorder", () => { vi.useRealTimers(); }); + describe("the stall watchdog across a system sleep", () => { + const watch = () => { + const recorder = new RenderObservabilityRecorder({ + pipelineStartMs: Date.now(), + log: makeLog(), + renderJobId: "render-stall-sleep", + }); + const outcome: { error?: unknown } = {}; + observeRenderStage(recorder, "capture_disk", undefined, () => new Promise(() => {}), { + stall: { progress: () => 0, noProgressMs: 10_000, firstProgressMs: 10_000 }, + }).catch((error: unknown) => { + outcome.error = error; + }); + return outcome; + }; + + it("does not read a ten-hour sleep as no progress", async () => { + vi.useFakeTimers(); + const outcome = watch(); + await vi.advanceTimersByTimeAsync(3_000); + vi.setSystemTime(Date.now() + 10 * 3_600_000); + await vi.advanceTimersByTimeAsync(9_000); + expect(outcome.error).toBeUndefined(); + await vi.advanceTimersByTimeAsync(10_000); + expect(outcome.error).toBeInstanceOf(Error); + vi.useRealTimers(); + }); + + it("still stalls when the process runs on time and no progress lands", async () => { + vi.useFakeTimers(); + const outcome = watch(); + await vi.advanceTimersByTimeAsync(11_000); + expect((outcome.error as Error).message).toContain("no progress for 10000ms"); + vi.useRealTimers(); + }); + }); + it("tells the stall rule each time a capture attempt opens", async () => { const recorder = new RenderObservabilityRecorder({ pipelineStartMs: Date.now(), diff --git a/packages/producer/src/services/render/observability.ts b/packages/producer/src/services/render/observability.ts index 64027108c0..0d39dfead1 100644 --- a/packages/producer/src/services/render/observability.ts +++ b/packages/producer/src/services/render/observability.ts @@ -1,4 +1,5 @@ import { createHash } from "node:crypto"; +import { createStallClock } from "./stallClock.js"; import { redactTelemetryString } from "@hyperframes/core"; import { CaptureFailure } from "@hyperframes/engine"; import type { ProducerLogger } from "../../logger.js"; @@ -646,35 +647,33 @@ function watchStall(phase: string, rule: StageStallRule) { const detail = rule.detail ?? (() => undefined); let last = rule.progress(); let budgetMs = rule.firstProgressMs; - let lastChangeAt = Date.now(); + const silence = createStallClock(); + const pollMs = Math.min(STALL_POLL_MS, rule.noProgressMs, rule.firstProgressMs); let attempt: AbortController | undefined; let timer: ReturnType | undefined; let inTail = false; const restartWindow = (ms: number) => { last = rule.progress(); budgetMs = ms; - lastChangeAt = Date.now(); + silence.reset(); }; const tripped = new Promise((_, reject) => { - timer = setInterval( - () => { - const current = rule.progress(); - if (rule.complete?.() && !inTail) { - inTail = true; - return restartWindow(rule.tailMs ?? rule.noProgressMs); - } - if (current !== last) return restartWindow(rule.noProgressMs); - if (Date.now() - lastChangeAt < budgetMs) return; - const error = new RenderStageStalledError(phase, budgetMs, current, detail()); - if (attempt && !attempt.signal.aborted) { - attempt.abort(error); - return restartWindow(rule.noProgressMs); - } - clearInterval(timer); - reject(error); - }, - Math.min(STALL_POLL_MS, rule.noProgressMs, rule.firstProgressMs), - ); + timer = setInterval(() => { + const current = rule.progress(); + if (rule.complete?.() && !inTail) { + inTail = true; + return restartWindow(rule.tailMs ?? rule.noProgressMs); + } + if (current !== last) return restartWindow(rule.noProgressMs); + if (silence.idle(pollMs) < budgetMs) return; + const error = new RenderStageStalledError(phase, budgetMs, current, detail()); + if (attempt && !attempt.signal.aborted) { + attempt.abort(error); + return restartWindow(rule.noProgressMs); + } + clearInterval(timer); + reject(error); + }, pollMs); }); const openAttempt: OpenStallAttempt = (parent) => { attempt = new AbortController(); diff --git a/packages/producer/src/services/render/stages/captureStreamingStage.test.ts b/packages/producer/src/services/render/stages/captureStreamingStage.test.ts index 3af404bf04..6ed213c5bc 100644 --- a/packages/producer/src/services/render/stages/captureStreamingStage.test.ts +++ b/packages/producer/src/services/render/stages/captureStreamingStage.test.ts @@ -518,6 +518,82 @@ describe("runCaptureStreamingStage", () => { expect(input.abortSignal).toBeUndefined(); }); + describe("a system sleep during capture", () => { + const settled = (promise: Promise) => { + const state = { rejected: undefined as unknown, done: false }; + promise.then( + () => (state.done = true), + (error) => { + state.done = true; + state.rejected = error; + }, + ); + return state; + }; + const after = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + const realNow = Date.now; + const sleepFor = (ms: number) => { + const skew = ms; + Date.now = () => realNow() + skew; + }; + const input = { + session: { captureMode: "screenshot", captureStep: "" }, + frameIndex: 3, + totalFrames: 9, + stallTimeoutMs: 400, + } as const; + + it("does not trip a sequential capture the moment the machine wakes", async () => { + const { raceAgainstStall } = await import("./captureStreamingStage.js"); + const state = settled(raceAgainstStall(new Promise(() => {}), 400, input)); + await after(100); + sleepFor(10 * 3_600_000); + try { + await after(500); + expect(state.done).toBe(false); + await after(1_000); + } finally { + Date.now = realNow; + } + expect((state.rejected as Error).name).toBe("SequentialCaptureStallError"); + }); + + it("trips a sequential capture that makes no progress on time", async () => { + const { raceAgainstStall } = await import("./captureStreamingStage.js"); + const state = settled(raceAgainstStall(new Promise(() => {}), 200, input)); + await after(700); + expect((state.rejected as Error).name).toBe("SequentialCaptureStallError"); + }); + + it("does not trip the parallel watchdog the moment the machine wakes", async () => { + hangParallelUntilAbort = true; + const prev = process.env.HF_DE_STALL_MS; + process.env.HF_DE_STALL_MS = "100"; + const { runCaptureStreamingStage } = await import("./captureStreamingStage.js"); + const baseInput = createInput({ forceScreenshot: false, ffmpegStreamingTimeout: 3_600_000 }); + const started = Date.now(); + const run = runCaptureStreamingStage({ + ...baseInput, + totalFrames: 100, + plan: { ...baseInput.plan, workerCount: 2, forceParallelStream: true }, + parallelCaptureLabel: "screenshot", + }).catch((error: unknown) => error); + await after(30); + sleepFor(10 * 3_600_000); + let caught: unknown; + try { + caught = await run; + } finally { + Date.now = realNow; + hangParallelUntilAbort = false; + if (prev === undefined) delete process.env.HF_DE_STALL_MS; + else process.env.HF_DE_STALL_MS = prev; + } + expect((caught as Error).name).toBe("ParallelCaptureStallError"); + expect(Date.now() - started).toBeGreaterThanOrEqual(190); + }); + }); + describe("a parallel capture stall", () => { const run = async ( afterFrames: number | number[], diff --git a/packages/producer/src/services/render/stages/captureStreamingStage.ts b/packages/producer/src/services/render/stages/captureStreamingStage.ts index 0f260bd23f..fb0882eb3c 100644 --- a/packages/producer/src/services/render/stages/captureStreamingStage.ts +++ b/packages/producer/src/services/render/stages/captureStreamingStage.ts @@ -79,6 +79,7 @@ import { pushWorkerPageErrors, withWorkerPageErrors, } from "../perfSummary.js"; +import { createStallClock } from "../stallClock.js"; import { ensureFrameWritten } from "./captureHdrFrameShared.js"; import { reportEncodeProgress, reportFrameProgress, reportWorkerStartup } from "../shared.js"; import { encoderFailureError } from "../encoderInterruption.js"; @@ -220,24 +221,31 @@ export function raceAgainstStall( signal?: AbortSignal, ): Promise { return new Promise((resolve, reject) => { - const timer = setTimeout( - () => { - reject( - signal?.aborted - ? new Error( - `[Render] Sequential ${captureModeLabel(input.session.captureMode)} capture aborted ` + - `while a capture call was in flight (frame ${input.frameIndex}/${input.totalFrames}).`, - ) - : new SequentialCaptureStallError( - input.session, - input.stallTimeoutMs, - input.frameIndex, - input.totalFrames, - ), - ); - }, - Math.max(0, deadlineMs), - ); + const clock = createStallClock(); + const trip = () => + reject( + signal?.aborted + ? new Error( + `[Render] Sequential ${captureModeLabel(input.session.captureMode)} capture aborted ` + + `while a capture call was in flight (frame ${input.frameIndex}/${input.totalFrames}).`, + ) + : new SequentialCaptureStallError( + input.session, + input.stallTimeoutMs, + input.frameIndex, + input.totalFrames, + ), + ); + const deadline = Math.max(0, deadlineMs); + let wait = Math.min(deadline, DE_STALL_POLL_MS); + let timer: ReturnType; + const look = () => { + const idle = clock.idle(wait); + if (idle >= deadline) return trip(); + wait = Math.min(deadline - idle, DE_STALL_POLL_MS); + timer = setTimeout(look, wait); + }; + timer = setTimeout(look, wait); promise.then( (value) => { clearTimeout(timer); @@ -514,21 +522,21 @@ async function runWorkerEncodePipelineLoop( item.trusted ? Promise.resolve(buf) : guard(session, item.idx, buf); const captureRisk = async (idx: number): Promise => { await risk!.open(); - lastProgressAt = Date.now(); + quiet.reset(); return raceAgainstStall( risk!.capture(idx, frameTime(idx)), - stallTimeoutMs - (Date.now() - lastProgressAt), + stallTimeoutMs - quiet.idle(0), { session, frameIndex: idx, totalFrames, stallTimeoutMs }, abortSignal, ); }; const stallTimeoutMs = resolveCaptureStallTimeoutMs(); - let lastProgressAt = Date.now(); + const quiet = createStallClock(); const captureWithStallGuard = (idx: number, promise: Promise): Promise => raceAgainstStall( promise, - stallTimeoutMs - (Date.now() - lastProgressAt), + stallTimeoutMs - quiet.idle(0), { session, frameIndex: idx, totalFrames, stallTimeoutMs }, abortSignal, ); @@ -545,7 +553,7 @@ async function runWorkerEncodePipelineLoop( reorderBuffer.advanceTo(prev.idx + 1); job.framesRendered = prev.idx + 1; resume.written = prev.idx + 1; - lastProgressAt = Date.now(); + quiet.reset(); reportFrameProgress( job, `Streaming frame ${prev.idx + 1}/${totalFrames}`, @@ -574,7 +582,7 @@ async function runWorkerEncodePipelineLoop( reorderBuffer.advanceTo(item.idx + 1); job.framesRendered = item.idx + 1; resume.written = item.idx + 1; - lastProgressAt = Date.now(); + quiet.reset(); reportFrameProgress( job, `Streaming frame ${item.idx + 1}/${totalFrames}`, @@ -837,7 +845,7 @@ export async function runCaptureStreamingStage( } const stallTimeoutMs = resolveCaptureStallTimeoutMs(); let lastCapturedFrames = 0; - let lastProgressAt = Date.now(); + const quiet = createStallClock(); const workerPhases = new Map(); const phaseDetail = (phase: ParallelWorkerPhaseDiagnostic) => `phase=${phase.phase} frame=${phase.frameIndex ?? "n/a"} step=${phase.readStep?.() ?? "n/a"} ` + @@ -856,23 +864,21 @@ export async function runCaptureStreamingStage( captureLabel === "BeginFrame", cfg, ); - const stallTimer = setInterval( - () => { - if (!watchdogArmed || Date.now() - lastProgressAt <= budget()) return; - stalled = true; - const stallErr = new ParallelCaptureStallError({ - variant: "drain", - captureLabel, - stallTimeoutMs: budget(), - lastCapturedFrames: resumeFrom + lastCapturedFrames, - totalFrames, - phaseSummary: phaseSummary(), - }); - parallelReorder.abort(stallErr); - stallController.abort(); - }, - Math.min(budget(), DE_STALL_POLL_MS), - ); + const pollMs = Math.min(budget(), DE_STALL_POLL_MS); + const stallTimer = setInterval(() => { + if (!watchdogArmed || quiet.idle(pollMs) <= budget()) return; + stalled = true; + const stallErr = new ParallelCaptureStallError({ + variant: "drain", + captureLabel, + stallTimeoutMs: budget(), + lastCapturedFrames: resumeFrom + lastCapturedFrames, + totalFrames, + phaseSummary: phaseSummary(), + }); + parallelReorder.abort(stallErr); + stallController.abort(); + }, pollMs); try { workerResults = await executeParallelCapture( @@ -888,7 +894,7 @@ export async function runCaptureStreamingStage( workerPhases.set(phase.workerId, phaseDetail(phase)); if (phase.phase === "frame_capture" && !watchdogArmed) { watchdogArmed = true; - lastProgressAt = Date.now(); + quiet.reset(); } log.info("[Render] Parallel capture worker phase", { workerId: phase.workerId, @@ -904,7 +910,7 @@ export async function runCaptureStreamingStage( } if (progress.capturedFrames > lastCapturedFrames) { lastCapturedFrames = progress.capturedFrames; - lastProgressAt = Date.now(); + quiet.reset(); watchdogArmed = true; } const doneFrames = resumeFrom + progress.capturedFrames; diff --git a/packages/producer/src/services/render/stallClock.test.ts b/packages/producer/src/services/render/stallClock.test.ts new file mode 100644 index 0000000000..ab314e51d9 --- /dev/null +++ b/packages/producer/src/services/render/stallClock.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it } from "vitest"; +import { createStallClock } from "./stallClock.js"; + +describe("createStallClock", () => { + it("does not count a suspended stretch as silence", () => { + let now = 0; + const clock = createStallClock(() => now); + now = 5_000; + expect(clock.idle(5_000)).toBe(5_000); + now += 10 * 3_600_000; + expect(clock.idle(5_000)).toBe(5_000); + now += 5_000; + expect(clock.idle(5_000)).toBe(10_000); + }); + + it("counts on-time looks without progress, and starts again at progress", () => { + let now = 0; + const clock = createStallClock(() => now); + let idle = 0; + for (let i = 0; i < 13; i++) { + now += 5_000; + idle = clock.idle(5_000); + } + expect(idle).toBe(65_000); + clock.reset(); + now += 5_000; + expect(clock.idle(5_000)).toBe(5_000); + }); +}); diff --git a/packages/producer/src/services/render/stallClock.ts b/packages/producer/src/services/render/stallClock.ts new file mode 100644 index 0000000000..0ad06eb040 --- /dev/null +++ b/packages/producer/src/services/render/stallClock.ts @@ -0,0 +1,26 @@ +/** A gap this much past a look's expected delay means the process was suspended, not that nothing landed. */ +const SUSPEND_GAP_MS = 15_000; + +export interface StallClock { + /** Progress landed: the silence starts again. */ + reset(): void; + /** Silent ms, not counting a suspended stretch; `expectedMs` is how long after the last look this one was due. */ + idle(expectedMs: number): number; +} + +export function createStallClock(now: () => number = () => Date.now()): StallClock { + let since = now(); + let seen = since; + return { + reset() { + since = seen = now(); + }, + idle(expectedMs) { + const at = now(); + const gap = at - seen; + seen = at; + if (gap > expectedMs + SUSPEND_GAP_MS) since += gap; + return at - since; + }, + }; +}