mirror of
https://github.com/heygen-com/hyperframes.git
synced 2026-10-31 02:45:40 +00:00
fix(producer): a render's page-error count includes what the capture workers saw (#6196)
* fix(producer): a render's page-error count includes what the capture workers saw * fix(producer): a failed parallel capture still carries its workers' page errors * fix(producer): every capture failure carries the page's errors, in every stage * chore(engine): drop a comment the capture classifier's name already says
This commit is contained in:
@@ -102,3 +102,22 @@ describe("classifyCaptureFailure", () => {
|
||||
).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("classifyCaptureFailure page errors", () => {
|
||||
it("carries the page's error lines onto the classified failure", () => {
|
||||
const failure = classifyCaptureFailure(new Error("Target closed"), {
|
||||
pageErrorLines: ["[Browser:PAGEERROR] script threw"],
|
||||
});
|
||||
expect(failure.pageErrorLines).toEqual(["[Browser:PAGEERROR] script threw"]);
|
||||
});
|
||||
|
||||
it("keeps the lines when an existing failure is reclassified", () => {
|
||||
const original = new CaptureFailure({
|
||||
kind: "transient_browser",
|
||||
message: "Target closed",
|
||||
pageErrorLines: ["[Browser:PAGEERROR] script threw"],
|
||||
});
|
||||
const failure = classifyCaptureFailure(original, { workerDiagnostics: [] });
|
||||
expect(failure.pageErrorLines).toEqual(["[Browser:PAGEERROR] script threw"]);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -19,12 +19,14 @@ export class CaptureFailure extends Error {
|
||||
readonly kind: CaptureFailureKind;
|
||||
readonly cause: unknown;
|
||||
readonly workerDiagnostics: readonly CaptureWorkerDiagnostic[];
|
||||
readonly pageErrorLines: readonly string[];
|
||||
|
||||
constructor(input: {
|
||||
kind: CaptureFailureKind;
|
||||
message: string;
|
||||
cause?: unknown;
|
||||
workerDiagnostics?: readonly CaptureWorkerDiagnostic[];
|
||||
pageErrorLines?: readonly string[];
|
||||
}) {
|
||||
super(input.message);
|
||||
this.name = "CaptureFailure";
|
||||
@@ -35,6 +37,7 @@ export class CaptureFailure extends Error {
|
||||
Object.freeze({ ...diagnostic, lines: Object.freeze([...diagnostic.lines]) }),
|
||||
),
|
||||
);
|
||||
this.pageErrorLines = Object.freeze([...(input.pageErrorLines ?? [])]);
|
||||
if (input.cause instanceof Error && input.cause.stack) this.stack = input.cause.stack;
|
||||
}
|
||||
}
|
||||
@@ -147,44 +150,54 @@ function ioError(error: unknown, message: string): boolean {
|
||||
);
|
||||
}
|
||||
|
||||
const ORDERED_KIND_RULES: readonly (readonly [CaptureFailureKind, (message: string) => boolean])[] =
|
||||
[
|
||||
[
|
||||
"memory_exhaustion",
|
||||
(message) =>
|
||||
BUN_MEMORY_EXHAUSTION_EXACT_MESSAGE.test(message.trim()) ||
|
||||
BUN_MEMORY_EXHAUSTION_WRAPPED_WORKER_MESSAGE.test(message) ||
|
||||
matchesAny(message, MEMORY_EXHAUSTION_ERROR_PATTERNS),
|
||||
],
|
||||
["verification", (message) => matchesAny(message, VERIFICATION_ERROR_PATTERNS)],
|
||||
["protocol_timeout", (message) => matchesAny(message, PROTOCOL_TIMEOUT_PATTERNS)],
|
||||
["transient_browser", (message) => matchesAny(message, TRANSIENT_BROWSER_ERROR_PATTERNS)],
|
||||
["authoring", (message) => matchesAny(message, AUTHORING_ERROR_PATTERNS)],
|
||||
];
|
||||
|
||||
function classifyKind(
|
||||
error: unknown,
|
||||
message: string,
|
||||
signal: AbortSignal | undefined,
|
||||
): CaptureFailureKind {
|
||||
if (signal?.aborted || /(?:render|capture)?_?cancelled|AbortError/i.test(message)) {
|
||||
return "cancelled";
|
||||
}
|
||||
const rule = ORDERED_KIND_RULES.find(([, matches]) => matches(message));
|
||||
if (rule) return rule[0];
|
||||
return ioError(error, message) ? "io" : "authoring";
|
||||
}
|
||||
|
||||
export function classifyCaptureFailure(
|
||||
error: unknown,
|
||||
options: {
|
||||
signal?: AbortSignal;
|
||||
workerDiagnostics?: readonly CaptureWorkerDiagnostic[];
|
||||
pageErrorLines?: readonly string[];
|
||||
} = {},
|
||||
): CaptureFailure {
|
||||
if (error instanceof CaptureFailure && !options.workerDiagnostics && !options.signal?.aborted) {
|
||||
return error;
|
||||
const prior = error instanceof CaptureFailure ? error : undefined;
|
||||
if (prior && !(options.workerDiagnostics || options.pageErrorLines || options.signal?.aborted)) {
|
||||
return prior;
|
||||
}
|
||||
const message = messageOf(error);
|
||||
let kind: CaptureFailureKind;
|
||||
if (options.signal?.aborted || /(?:render|capture)?_?cancelled|AbortError/i.test(message)) {
|
||||
kind = "cancelled";
|
||||
} else if (
|
||||
BUN_MEMORY_EXHAUSTION_EXACT_MESSAGE.test(message.trim()) ||
|
||||
BUN_MEMORY_EXHAUSTION_WRAPPED_WORKER_MESSAGE.test(message) ||
|
||||
matchesAny(message, MEMORY_EXHAUSTION_ERROR_PATTERNS)
|
||||
) {
|
||||
kind = "memory_exhaustion";
|
||||
} else if (matchesAny(message, VERIFICATION_ERROR_PATTERNS)) {
|
||||
kind = "verification";
|
||||
} else if (matchesAny(message, PROTOCOL_TIMEOUT_PATTERNS)) {
|
||||
kind = "protocol_timeout";
|
||||
} else if (matchesAny(message, TRANSIENT_BROWSER_ERROR_PATTERNS)) {
|
||||
kind = "transient_browser";
|
||||
} else if (matchesAny(message, AUTHORING_ERROR_PATTERNS)) {
|
||||
kind = "authoring";
|
||||
} else {
|
||||
kind = ioError(error, message) ? "io" : "authoring";
|
||||
}
|
||||
const kind = classifyKind(error, message, options.signal);
|
||||
return new CaptureFailure({
|
||||
kind,
|
||||
message,
|
||||
cause: error,
|
||||
workerDiagnostics:
|
||||
options.workerDiagnostics ??
|
||||
(error instanceof CaptureFailure ? error.workerDiagnostics : undefined),
|
||||
workerDiagnostics: options.workerDiagnostics ?? prior?.workerDiagnostics,
|
||||
pageErrorLines: options.pageErrorLines ?? prior?.pageErrorLines,
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -62,6 +62,92 @@ describe("executeParallelCapture peer abort", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("carries every worker's page-error lines on the failure, including a worker that did not fail", async () => {
|
||||
const root = mkdtempSync(join(tmpdir(), "hf-failure-page-errors-"));
|
||||
vi.doMock("./frameCapture.js", () => ({
|
||||
createCaptureSession: vi.fn(async (_url: string, outputDir: string) => {
|
||||
const workerId = outputDir.endsWith("worker-0") ? 0 : 1;
|
||||
return {
|
||||
workerId,
|
||||
browserConsoleBuffer: [`[Browser:PAGEERROR] worker ${workerId} script threw`],
|
||||
} as unknown as CaptureSession & { workerId: number };
|
||||
}),
|
||||
initializeSession: vi.fn(async (session: CaptureSession & { workerId: number }) => {
|
||||
if (session.workerId === 0) throw new Error("Composition has zero duration.");
|
||||
await Promise.resolve();
|
||||
}),
|
||||
captureFrame: vi.fn().mockResolvedValue(undefined),
|
||||
captureFrameToBuffer: vi.fn(),
|
||||
captureFrameToBufferPipelined: vi.fn(),
|
||||
closeCaptureSession: vi.fn().mockResolvedValue(undefined),
|
||||
getCapturePerfSummary: vi.fn(() => ({ frames: 0 })),
|
||||
}));
|
||||
|
||||
try {
|
||||
const { executeParallelCapture } = await import("./parallelCoordinator.js");
|
||||
const result = executeParallelCapture(
|
||||
"http://127.0.0.1",
|
||||
root,
|
||||
[
|
||||
{ workerId: 0, startFrame: 0, endFrame: 1, outputDir: join(root, "worker-0") },
|
||||
{ workerId: 1, startFrame: 1, endFrame: 2, outputDir: join(root, "worker-1") },
|
||||
],
|
||||
{ width: 320, height: 180, fps: { num: 30, den: 1 } },
|
||||
() => null,
|
||||
);
|
||||
|
||||
await expect(result).rejects.toMatchObject({
|
||||
name: "CaptureFailure",
|
||||
pageErrorLines: [
|
||||
"[Browser:PAGEERROR] worker 0 script threw",
|
||||
"[Browser:PAGEERROR] worker 1 script threw",
|
||||
],
|
||||
});
|
||||
} finally {
|
||||
rmSync(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("hands the pool the failed worker's own page-error lines", async () => {
|
||||
const root = mkdtempSync(join(tmpdir(), "hf-worker-failure-lines-"));
|
||||
vi.doMock("./frameCapture.js", () => ({
|
||||
createCaptureSession: vi.fn(async () => ({
|
||||
browserConsoleBuffer: ["[Browser:PAGEERROR] script threw at load"],
|
||||
})),
|
||||
initializeSession: vi.fn(async () => {
|
||||
throw new Error("Protocol error (Target.createTarget): Target closed");
|
||||
}),
|
||||
captureFrame: vi.fn(),
|
||||
captureFrameToBuffer: vi.fn(),
|
||||
captureFrameToBufferPipelined: vi.fn(),
|
||||
closeCaptureSession: vi.fn().mockResolvedValue(undefined),
|
||||
getCapturePerfSummary: vi.fn(() => ({ frames: 0 })),
|
||||
}));
|
||||
|
||||
try {
|
||||
const { executeParallelCapture } = await import("./parallelCoordinator.js");
|
||||
const failures: { pageErrorLines: readonly string[] }[] = [];
|
||||
await executeParallelCapture(
|
||||
"http://127.0.0.1",
|
||||
root,
|
||||
[{ workerId: 0, startFrame: 0, endFrame: 1, outputDir: join(root, "worker-0") }],
|
||||
{ width: 320, height: 180, fps: { num: 30, den: 1 } },
|
||||
() => null,
|
||||
undefined,
|
||||
undefined,
|
||||
async () => {},
|
||||
undefined,
|
||||
{ onWorkerFailure: (failure) => failures.push(failure) },
|
||||
).catch(() => {});
|
||||
|
||||
expect(failures.map((failure) => failure.pageErrorLines)).toEqual([
|
||||
["[Browser:PAGEERROR] script threw at load"],
|
||||
]);
|
||||
} finally {
|
||||
rmSync(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("closes a browser whose launch finishes after the cancel before the render rejects", async () => {
|
||||
const root = mkdtempSync(join(tmpdir(), "hf-late-launch-"));
|
||||
const lateSession = {} as CaptureSession;
|
||||
|
||||
@@ -79,6 +79,8 @@ export interface WorkerResult {
|
||||
perf?: CapturePerfSummary;
|
||||
error?: string;
|
||||
diagnostics?: string[];
|
||||
/** The worker session's `[Browser:PAGEERROR]` console lines, kept so the render's error count sees them. */
|
||||
pageErrorLines?: string[];
|
||||
failure?: CaptureFailure;
|
||||
}
|
||||
|
||||
@@ -941,6 +943,7 @@ async function executeWorkerTask(
|
||||
frameStride: task.frameStride,
|
||||
durationMs: Date.now() - startTime,
|
||||
perf,
|
||||
pageErrorLines: pageErrorLinesOf(activeSession),
|
||||
};
|
||||
} catch (error) {
|
||||
const diagnostics = session ? selectWorkerDiagnostics(session.browserConsoleBuffer) : [];
|
||||
@@ -954,6 +957,7 @@ async function executeWorkerTask(
|
||||
const failure = classifyCaptureFailure(error, {
|
||||
signal,
|
||||
workerDiagnostics: [workerDiagnostic],
|
||||
pageErrorLines: session ? pageErrorLinesOf(session) : undefined,
|
||||
});
|
||||
onFailure?.(failure);
|
||||
return {
|
||||
@@ -966,6 +970,7 @@ async function executeWorkerTask(
|
||||
perf,
|
||||
error: failure.message,
|
||||
diagnostics: diagnostics.length > 0 ? diagnostics : undefined,
|
||||
pageErrorLines: session ? pageErrorLinesOf(session) : undefined,
|
||||
failure,
|
||||
};
|
||||
} finally {
|
||||
@@ -1016,6 +1021,10 @@ export interface ParallelCaptureHooks {
|
||||
onWorkerFailure?: (failure: CaptureFailure) => void;
|
||||
}
|
||||
|
||||
function pageErrorLinesOf(session: CaptureSession): string[] {
|
||||
return session.browserConsoleBuffer.filter((line) => line.includes("PAGEERROR"));
|
||||
}
|
||||
|
||||
/**
|
||||
* The pool's single failure gate. Both halves of the contract the streaming
|
||||
* stage relies on live here so they can be pinned without a browser: the
|
||||
@@ -1157,6 +1166,7 @@ export async function executeParallelCapture(
|
||||
message: `[Parallel] Capture failed: ${errorMessages}`,
|
||||
cause: representative,
|
||||
workerDiagnostics,
|
||||
pageErrorLines: [...new Set(results.flatMap((result) => result.pageErrorLines ?? []))],
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user