ext/hyperframes
2
0
Fork 0
mirror of https://github.com/heygen-com/hyperframes.git synced 2026-11-04 13:07:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(producer): a laptop that sleeps mid-render no longer counts the sleep as a capture stall (#7096)

Browse Source 
* fix(producer): a laptop that sleeps mid-render no longer counts the sleep as a capture stall

* style(producer): format the stall watchdog timers
This commit is contained in:
Miguel Ángel
2026-11-03 23:11:32 +00:00
committed by GitHub
parent 3dc82f18e3
commit e2edc93f5f
6 changed files with 238 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/producer/src/services/render/observability.test.ts
+37
View File
@@ -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<void>(() => {}), {
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(),
packages/producer/src/services/render/observability.ts
+20 -21
View File
@@ -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<typeof setInterval> | undefined;
let inTail = false;
const restartWindow = (ms: number) => {
last = rule.progress();
budgetMs = ms;
lastChangeAt = Date.now();
silence.reset();
};
const tripped = new Promise<never>((_, 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();
packages/producer/src/services/render/stages/captureStreamingStage.test.ts
+76
View File
@@ -518,6 +518,82 @@ describe("runCaptureStreamingStage", () => {
expect(input.abortSignal).toBeUndefined();
});
describe("a system sleep during capture", () => {
const settled = (promise: Promise<unknown>) => {
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<never>(() => {}), 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<never>(() => {}), 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[],
packages/producer/src/services/render/stages/captureStreamingStage.ts
+50 -44
View File
@@ -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<T>(
signal?: AbortSignal,
): Promise<T> {
return new Promise<T>((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<typeof setTimeout>;
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<Buffer> => {
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 = <T>(idx: number, promise: Promise<T>): Promise<T> =>
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<number, string>();
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;
packages/producer/src/services/render/stallClock.test.ts
+29
View File
@@ -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);
});
});
packages/producer/src/services/render/stallClock.ts
+26
View File
@@ -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;
},
};
}