feat(producer): calibration-aware heartbeat + worker-death terminal-error contract

Field signals ts=1784019503 (heartbeat reports 0 frames during 64s
browser calibration — reads as broken but is healthy) and ts=1784042064
(1292s Windows render hard-exited during video frame extraction with
no final error string — silent worker crash).

Add calibrating/capturing state to heartbeat labels; surface synthetic
terminal error on unexpected worker exit when no explicit error was
emitted.

Stack: PR #6 of 9 (base via/overlay-count-lint).
Signed-off-by: Via <vance@heygen.com>
This commit is contained in:
Via
2026-07-15 23:14:49 +00:00
parent 75de927fb5
commit 971bcf39ae
5 changed files with 287 additions and 4 deletions
@@ -2,10 +2,12 @@ import { describe, it, expect } from "vitest";
import {
calculateOptimalWorkers,
distributeFrames,
expectedFramesForTask,
formatWorkerFailure,
selectWorkerDiagnostics,
shouldDisableBrowserPoolForParallelWorker,
shouldVerifyWorkerGpu,
synthesizeSilentWorkerExitError,
resolveParallelDeVerifySamples,
} from "./parallelCoordinator.js";
import type { EngineConfig } from "../config.js";
@@ -161,6 +163,91 @@ describe("worker failure diagnostics", () => {
"Worker 1: Navigation timeout of 60000 ms exceeded; diagnostics: [FrameCapture:ERROR] page.goto failed mode=screenshot timeoutMs=60000",
);
});
// Field signal ts=1784042064: a Windows render hard-exited during video
// frame extraction without emitting a terminal error string; the parent
// treated the result as success because `error` was falsy. The synthesized
// string surfaces the shortfall so downstream telemetry can classify the
// failure instead of it disappearing silently.
it("synthesizes a terminal error when a worker exits with no error string", () => {
const message = formatWorkerFailure({
workerId: 3,
framesCaptured: 12,
startFrame: 0,
endFrame: 60,
durationMs: 180_000,
});
expect(message).toContain("Worker 3: worker 3 exited without terminal error string");
expect(message).toContain("framesCaptured=12");
expect(message).toContain("expected=60");
expect(message).toContain("range=[0, 60)");
expect(message).toContain("ts=1784042064");
expect(message).toContain("--workers=1");
});
it("prefers an explicit error string over the synthesized silent-exit message", () => {
const message = formatWorkerFailure({
workerId: 3,
framesCaptured: 12,
startFrame: 0,
endFrame: 60,
durationMs: 180_000,
error: "Target closed",
});
expect(message).toBe("Worker 3: Target closed");
expect(message).not.toContain("ts=1784042064");
expect(message).not.toContain("exited without terminal error string");
});
it("treats an empty error string as a silent exit for the message contract", () => {
const message = formatWorkerFailure({
workerId: 3,
framesCaptured: 12,
startFrame: 0,
endFrame: 60,
durationMs: 180_000,
error: "",
});
expect(message).toContain("exited without terminal error string");
expect(message).toContain("ts=1784042064");
});
});
describe("expectedFramesForTask", () => {
it("returns the range length for contiguous tasks", () => {
expect(expectedFramesForTask({ startFrame: 0, endFrame: 30 })).toBe(30);
expect(expectedFramesForTask({ startFrame: 10, endFrame: 25 })).toBe(15);
});
it("divides by stride for interleaved tasks", () => {
// Worker 0 in a 3-way interleave over 30 frames captures 0, 3, 6, ..., 27 → 10 frames.
expect(expectedFramesForTask({ startFrame: 0, endFrame: 30, frameStride: 3 })).toBe(10);
});
it("rounds up so the last-stride-remainder frame is expected", () => {
// Worker 0 in a 3-way interleave over 31 frames captures 0, 3, ..., 30 → 11 frames.
expect(expectedFramesForTask({ startFrame: 0, endFrame: 31, frameStride: 3 })).toBe(11);
});
it("returns zero for empty ranges", () => {
expect(expectedFramesForTask({ startFrame: 10, endFrame: 10 })).toBe(0);
expect(expectedFramesForTask({ startFrame: 30, endFrame: 10 })).toBe(0);
});
});
describe("synthesizeSilentWorkerExitError", () => {
it("names the field signal and reruns hint so operators can classify the failure", () => {
const message = synthesizeSilentWorkerExitError(
{ workerId: 7, framesCaptured: 40, startFrame: 60, endFrame: 120 },
60,
);
expect(message).toContain("worker 7 exited without terminal error string");
expect(message).toContain("framesCaptured=40");
expect(message).toContain("expected=60");
expect(message).toContain("range=[60, 120)");
expect(message).toContain("Field signal ts=1784042064");
expect(message).toContain("--workers=1");
});
});
describe("shouldVerifyWorkerGpu", () => {
@@ -147,8 +147,49 @@ function compactDiagnosticLine(line: string): string {
return line.replace(/\s+/g, " ").trim();
}
/**
* Expected frame count for a worker task, honoring its stride. Contiguous
* tasks (stride 1) expect `endFrame - startFrame`; interleaved tasks
* (stride > 1) expect `ceil((endFrame - startFrame) / stride)`, matching
* the loop shape in `captureFrameRange`.
*/
export function expectedFramesForTask(task: {
startFrame: number;
endFrame: number;
frameStride?: number;
}): number {
const stride = task.frameStride ?? 1;
return Math.max(0, Math.ceil((task.endFrame - task.startFrame) / stride));
}
/**
* Synthetic terminal-error message for a worker whose exit didn't produce
* an explicit error string but under-captured its expected frame range.
* Field signal ts=1784042064: a 1292s Windows render hard-exited during
* capture with no final error string, leaving the operator with no
* actionable trace. This message surfaces the shortfall + reruns hint
* so downstream telemetry (and operators grepping logs) can classify the
* failure instead of it disappearing silently.
*/
export function synthesizeSilentWorkerExitError(
result: Pick<WorkerResult, "workerId" | "framesCaptured" | "startFrame" | "endFrame">,
expectedFrames: number,
): string {
return (
`worker ${result.workerId} exited without terminal error string ` +
`(framesCaptured=${result.framesCaptured}, expected=${expectedFrames}, ` +
`range=[${result.startFrame}, ${result.endFrame})). ` +
`Field signal ts=1784042064 — this class of failure has been reported; ` +
`consider re-run with --workers=1 to isolate.`
);
}
export function formatWorkerFailure(result: WorkerResult): string {
const base = `Worker ${result.workerId}: ${result.error ?? "unknown error"}`;
const errorText =
result.error && result.error.length > 0
? result.error
: synthesizeSilentWorkerExitError(result, expectedFramesForTask(result));
const base = `Worker ${result.workerId}: ${errorText}`;
if (!result.diagnostics || result.diagnostics.length === 0) return base;
const diagnostics = result.diagnostics.map(compactDiagnosticLine).join(" | ");
@@ -541,6 +582,17 @@ export async function executeParallelCapture(
),
);
// A worker may return without an error string yet with framesCaptured
// below the task's expected count — that's the silent-exit shape field
// signal ts=1784042064 called out. Synthesize a terminal error string
// in-place so the filter below treats it as a failure (and so the
// caller's failure message actually names what went wrong).
for (const r of results) {
if (!r.error && r.framesCaptured < expectedFramesForTask(r)) {
r.error = synthesizeSilentWorkerExitError(r, expectedFramesForTask(r));
}
}
const errors = results.filter((r) => r.error);
if (errors.length > 0) {
const errorMessages = errors.map(formatWorkerFailure).join("; ");
@@ -209,6 +209,98 @@ describe("RenderObservabilityRecorder", () => {
vi.useRealTimers();
});
// Field signal ts=1784019503: a healthy ~64s browser calibration
// pre-capture emitted heartbeats saying "stage still running /
// framesCompleted: 0" and read as broken to downstream consumers. The
// calibration call sites now pass `heartbeatMessage: "browser calibrating
// (frames not started)"` so operator-facing logs and metrics can
// distinguish healthy calibration waits from actual zero-frame stalls
// once capture is meant to be underway.
it("uses a custom heartbeat message during calibration stages", async () => {
vi.useFakeTimers();
const log = makeLog();
const recorder = new RenderObservabilityRecorder({
pipelineStartMs: Date.now(),
log,
renderJobId: "render-calibrating",
});
let resolveStage: (() => void) | undefined;
const stage = observeRenderStage(
recorder,
"capture_calibration",
{ stagePhase: "calibrating", framesCompleted: 0, totalFrames: 900 },
() =>
new Promise<void>((resolve) => {
resolveStage = resolve;
}),
{ heartbeatMessage: "browser calibrating (frames not started)" },
);
await vi.advanceTimersByTimeAsync(30_000);
const heartbeatCalls = log.info.mock.calls.filter(
([message, meta]) =>
message === "[Render:trace]" &&
meta?.phase === "capture_calibration" &&
meta?.status === "checkpoint",
);
expect(heartbeatCalls).toHaveLength(1);
expect(heartbeatCalls[0]?.[1]).toEqual(
expect.objectContaining({
message: "browser calibrating (frames not started)",
stagePhase: "calibrating",
framesCompleted: 0,
heartbeatIndex: 1,
}),
);
// No "stage still running" for this stage — the calibration override
// must fully replace the default, not sit alongside it.
expect(
log.info.mock.calls.some(
([message, meta]) =>
message === "[Render:trace]" && meta?.message === "stage still running",
),
).toBe(false);
resolveStage?.();
await stage;
vi.useRealTimers();
});
it("keeps the default heartbeat message when no override is passed", async () => {
vi.useFakeTimers();
const log = makeLog();
const recorder = new RenderObservabilityRecorder({
pipelineStartMs: Date.now(),
log,
renderJobId: "render-default",
});
let resolveStage: (() => void) | undefined;
const stage = observeRenderStage(
recorder,
"capture_streaming",
{ stagePhase: "capturing", framesCompleted: 42, totalFrames: 900 },
() =>
new Promise<void>((resolve) => {
resolveStage = resolve;
}),
);
await vi.advanceTimersByTimeAsync(30_000);
const heartbeatCalls = log.info.mock.calls.filter(
([message, meta]) => message === "[Render:trace]" && meta?.message === "stage still running",
);
expect(heartbeatCalls).toHaveLength(1);
expect(heartbeatCalls[0]?.[1]).toEqual(
expect.objectContaining({
stagePhase: "capturing",
framesCompleted: 42,
}),
);
resolveStage?.();
await stage;
vi.useRealTimers();
});
it("clears pending heartbeats when a stage rejects", async () => {
vi.useFakeTimers();
const log = makeLog();
@@ -164,6 +164,12 @@ const ALLOWED_STRING_DATA_KEYS = new Set([
"renderJobId",
"requestedHdrMode",
"requestedWorkers",
// "calibrating" during pre-capture browser warm-up / calibration stages;
// "capturing" during capture_disk / capture_streaming / capture_hdr_*
// stages. Distinguishes healthy 0-frame heartbeats (browser starting up)
// from actual zero-frame stalls once capture is meant to be underway.
// Field signal ts=1784019503.
"stagePhase",
]);
const RESERVED_LOG_KEYS = new Set([
"data",
@@ -416,12 +422,40 @@ function heartbeatTargetMs(index: number): number {
return HEARTBEAT_RAMP_END_MS + overflow * HEARTBEAT_REPEAT_MS;
}
/**
* Options for `observeRenderStage` heartbeat behavior.
*
* The default heartbeat message is "stage still running", chosen for the
* capture stages where a live frame count in the observation data already
* communicates progress. Stages that run BEFORE any frame count is
* meaningful — the ~64s browser calibration path in particular — inherit
* that default and confusingly report "stage still running / framesCompleted:
* 0" to downstream consumers. Field signal ts=1784019503 captured exactly
* that read-as-broken shape on a healthy 64s calibration.
*
* `heartbeatMessage` lets the calibration call sites override the message
* to "browser calibrating" so operator-facing logs and downstream metrics
* can distinguish healthy pre-capture waits from actual zero-frame stalls
* mid-capture. The `data` payload also flows through (callers pass a
* `stagePhase: "calibrating" | "capturing"` field) so structured consumers
* don't have to string-match on the message.
*/
export interface ObserveRenderStageOptions {
/**
* Message to attach to each heartbeat checkpoint for this stage.
* Defaults to "stage still running".
*/
heartbeatMessage?: string;
}
export async function observeRenderStage<T>(
recorder: RenderObservabilityRecorder,
phase: string,
data: RenderObservationData | undefined,
fn: () => Promise<T>,
options: ObserveRenderStageOptions = {},
): Promise<T> {
const heartbeatMessage = options.heartbeatMessage ?? "stage still running";
const startedAt = recorder.stageStart(phase, data);
let heartbeatCount = 0;
let lastFiredAtMs = 0;
@@ -431,7 +465,7 @@ export async function observeRenderStage<T>(
heartbeatTimer = setTimeout(() => {
lastFiredAtMs = targetMs;
heartbeatCount += 1;
recorder.checkpoint(phase, "stage still running", {
recorder.checkpoint(phase, heartbeatMessage, {
...data,
heartbeatIndex: heartbeatCount,
stageElapsedMs: Date.now() - startedAt,
@@ -1862,7 +1862,7 @@ export async function executeRenderJob(
const probeResult = await observeRenderStage(
observability,
"browser_probe",
{ forceScreenshot: captureForceScreenshot },
{ forceScreenshot: captureForceScreenshot, stagePhase: "calibrating" },
() =>
runProbeStage({
projectDir,
@@ -1879,6 +1879,10 @@ export async function executeRenderJob(
needsAlpha,
deviceScaleFactor,
}),
// Browser probe is pre-capture; report `browser calibrating` so a
// slow probe (~64s SwiftShader warm-up on Windows was the reported
// shape) doesn't read as a zero-frame stall. Field signal ts=1784019503.
{ heartbeatMessage: "browser calibrating (frames not started)" },
);
compiled = probeResult.compiled;
compositionHash = computeCompositionObservabilityHash(compiled.html);
@@ -2277,9 +2281,15 @@ export async function executeRenderJob(
// captureStageObservationData can close over it for the calibration
// stage itself — reads as undefined until resolveRenderWorkerCount runs.
let workerCount: number;
// Default `stagePhase` — spread FIRST so a caller can override via
// `extra` (the calibration call site passes `stagePhase: "calibrating"`
// to distinguish healthy pre-capture waits from actual zero-frame stalls
// during capture; heartbeats in `capture_calibration` otherwise emit
// `framesCompleted: 0` and read as broken). Field signal ts=1784019503.
const captureStageObservationData = (
extra: RenderObservationData = {},
): RenderObservationData => ({
stagePhase: "capturing",
...extra,
get workerCount() {
return workerCount;
@@ -2329,7 +2339,14 @@ export async function executeRenderJob(
const outcome = await observeRenderStage(
observability,
"capture_calibration",
captureStageObservationData({ forceScreenshot: captureForceScreenshot }),
captureStageObservationData({
forceScreenshot: captureForceScreenshot,
// Override the default `capturing` — calibration writes probe
// frames only, not `job.framesRendered`, so heartbeats reporting
// `framesCompleted: 0` misread as broken. Field signal
// ts=1784019503.
stagePhase: "calibrating",
}),
() =>
runCaptureCalibration({
cfg,
@@ -2344,6 +2361,7 @@ export async function executeRenderJob(
createRenderVideoFrameInjector,
assertNotAborted,
}),
{ heartbeatMessage: "browser calibrating (frames not started)" },
);
captureCalibration = outcome.calibration;
captureForceScreenshot = outcome.forceScreenshot;