ext/hyperframes
2
0
Fork 0
mirror of https://github.com/heygen-com/hyperframes.git synced 2026-09-03 12:54:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(producer): extend DE stall watchdog to the single-worker streaming path

Browse Source
This commit is contained in:
Vance Ingalls
2026-07-15 11:53:54 -07:00
parent dbdf03cd6b
commit 650977d1d8
2 changed files with 148 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/producer/src/services/render/stages/captureStreamingStage.test.ts
+74 -11
View File
@@ -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", () => {
packages/producer/src/services/render/stages/captureStreamingStage.ts
+74 -20
View File
@@ -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<T>(promise: Promise<T>, deadlineMs: number, message: string): Promise<T> {
return new Promise<T>((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<Buffer> => guard(session, idx, buf);
const stallTimeoutMs = resolveDeStallTimeoutMs();
let lastProgressAt = Date.now();
const captureWithStallGuard = <T>(idx: number, promise: Promise<T>): Promise<T> =>
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<void> => {
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;