mirror of
https://github.com/heygen-com/hyperframes.git
synced 2026-09-11 06:30:03 +00:00
fix(producer): close orphaned probe session before verify-triggered retries
On a parallel-capture disk-verify or streaming-drain breach, the outer catch cleared probeSession without first closing the still-owned session, orphaning the probe Chrome process precisely when the retry was recovering from GPU/memory pressure. Introduce closeOrphanedProbeForRetry so both retry catches close the session (with defensive .catch that logs on close error) before releasing the reference, and cover it with a focused unit test asserting closure-before-clear and the swallow-and-warn behaviour. Addresses Magi's REQUEST_CHANGES on #2749; also closes Rames' sibling concern at the streaming-retry path (renderOrchestrator.ts:3093). — Via
This commit is contained in:
@@ -17,6 +17,7 @@ vi.mock("@hyperframes/engine", async (importOriginal) => {
|
|||||||
import {
|
import {
|
||||||
buildMissingFrameRetryBatches,
|
buildMissingFrameRetryBatches,
|
||||||
captureAttemptMadeProgress,
|
captureAttemptMadeProgress,
|
||||||
|
closeOrphanedProbeForRetry,
|
||||||
describeMemoryExhaustion,
|
describeMemoryExhaustion,
|
||||||
executeDiskCaptureWithAdaptiveRetry,
|
executeDiskCaptureWithAdaptiveRetry,
|
||||||
collectVideoMetadataHints,
|
collectVideoMetadataHints,
|
||||||
@@ -2130,3 +2131,63 @@ describe("shouldStreamParallelCapture (non-DE parallel streaming router)", () =>
|
|||||||
expect(shouldStreamParallelCapture({ ...eligible, layeredOrEffectRoute: true })).toBe(false);
|
expect(shouldStreamParallelCapture({ ...eligible, layeredOrEffectRoute: true })).toBe(false);
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("closeOrphanedProbeForRetry (probe cleanup before verify-triggered retry)", () => {
|
||||||
|
// Enough of a CaptureSession stand-in to exercise the closer path — the
|
||||||
|
// helper never inspects the object; it just hands it to the injected closer.
|
||||||
|
const stubSession = { browserConsoleBuffer: [] } as unknown as Parameters<
|
||||||
|
typeof closeOrphanedProbeForRetry
|
||||||
|
>[0];
|
||||||
|
|
||||||
|
it("hands the still-owned probe to the closer before the caller clears it", async () => {
|
||||||
|
const closer = vi.fn(async () => {});
|
||||||
|
const log = { warn: vi.fn() };
|
||||||
|
|
||||||
|
await closeOrphanedProbeForRetry(stubSession, closer, log, "streaming");
|
||||||
|
|
||||||
|
expect(closer).toHaveBeenCalledTimes(1);
|
||||||
|
expect(closer).toHaveBeenCalledWith(stubSession);
|
||||||
|
expect(log.warn).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("swallows a close failure with a warn so the caller's retry can proceed", async () => {
|
||||||
|
const closer = vi.fn(async () => {
|
||||||
|
throw new Error("chrome zombie");
|
||||||
|
});
|
||||||
|
const log = { warn: vi.fn() };
|
||||||
|
|
||||||
|
await expect(
|
||||||
|
closeOrphanedProbeForRetry(stubSession, closer, log, "disk verify"),
|
||||||
|
).resolves.toBeUndefined();
|
||||||
|
|
||||||
|
expect(closer).toHaveBeenCalledTimes(1);
|
||||||
|
expect(log.warn).toHaveBeenCalledTimes(1);
|
||||||
|
const [message, meta] = log.warn.mock.calls[0];
|
||||||
|
expect(message).toContain("disk verify");
|
||||||
|
expect((meta as { error: string }).error).toBe("chrome zombie");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("preserves the retry context in the warn message so the audit trail names which retry path leaked", async () => {
|
||||||
|
const closer = vi.fn(async () => {
|
||||||
|
throw new Error("session already closed");
|
||||||
|
});
|
||||||
|
const log = { warn: vi.fn() };
|
||||||
|
|
||||||
|
await closeOrphanedProbeForRetry(stubSession, closer, log, "streaming");
|
||||||
|
|
||||||
|
expect(log.warn.mock.calls[0][0]).toContain("streaming");
|
||||||
|
expect(log.warn.mock.calls[0][0]).not.toContain("disk verify");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("stringifies non-Error rejections so the log entry still names the cause", async () => {
|
||||||
|
const closer = vi.fn(async () => Promise.reject("string-only rejection"));
|
||||||
|
const log = { warn: vi.fn() };
|
||||||
|
|
||||||
|
await closeOrphanedProbeForRetry(stubSession, closer, log, "streaming");
|
||||||
|
|
||||||
|
expect(log.warn).toHaveBeenCalledTimes(1);
|
||||||
|
expect((log.warn.mock.calls[0][1] as { error: string }).error).toBe(
|
||||||
|
"string-only rejection",
|
||||||
|
);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|||||||
@@ -1468,6 +1468,32 @@ export function shouldRetryViaPinnedFallback(args: {
|
|||||||
return args.deWorkerInversion === "inverted" || args.deParallelRouter === "routed";
|
return args.deWorkerInversion === "inverted" || args.deParallelRouter === "routed";
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* When a self-verify (or pinned-fallback) retry is triggered mid-capture, the
|
||||||
|
* caller may still hold a live probe session that the failed stage was passed
|
||||||
|
* but did not (or could not) close in its own `finally` before it threw. Left
|
||||||
|
* behind, that session's Chrome process orphans until the containing render
|
||||||
|
* exits — precisely when we are recovering from GPU/memory pressure and can
|
||||||
|
* least afford an unaccounted Chrome. Close it before the caller clears its
|
||||||
|
* reference; swallow any close error with a warn so the retry itself is never
|
||||||
|
* derailed by a shutdown hiccup.
|
||||||
|
*/
|
||||||
|
export async function closeOrphanedProbeForRetry(
|
||||||
|
probe: CaptureSession,
|
||||||
|
closer: (session: CaptureSession) => Promise<void>,
|
||||||
|
log: Pick<ProducerLogger, "warn">,
|
||||||
|
retryContext: string,
|
||||||
|
): Promise<void> {
|
||||||
|
try {
|
||||||
|
await closer(probe);
|
||||||
|
} catch (closeErr) {
|
||||||
|
log.warn(
|
||||||
|
`[Render] probe close before ${retryContext} retry failed; continuing with retry`,
|
||||||
|
{ error: closeErr instanceof Error ? closeErr.message : String(closeErr) },
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Parallel-streaming router for NON-drawElement capture (screenshot on
|
* Parallel-streaming router for NON-drawElement capture (screenshot on
|
||||||
* macOS/Windows/forced-screenshot, BeginFrame on Linux): should this
|
* macOS/Windows/forced-screenshot, BeginFrame on Linux): should this
|
||||||
@@ -3090,7 +3116,16 @@ async function executeRenderPipeline(input: {
|
|||||||
deWorkerInversion,
|
deWorkerInversion,
|
||||||
deParallelRouter,
|
deParallelRouter,
|
||||||
});
|
});
|
||||||
|
// Streaming stage aims to close the probe in its own finally; if it
|
||||||
|
// threw before doing so, the Chrome process would orphan through the
|
||||||
|
// pinned-fallback retry. Close defensively before we release the
|
||||||
|
// reference — see closeOrphanedProbeForRetry.
|
||||||
|
if (probeSession) {
|
||||||
|
lastBrowserConsole = probeSession.browserConsoleBuffer;
|
||||||
|
const orphaned = probeSession;
|
||||||
probeSession = null;
|
probeSession = null;
|
||||||
|
await closeOrphanedProbeForRetry(orphaned, closeCaptureSession, log, "streaming");
|
||||||
|
}
|
||||||
if (failedRouting === "worker_inversion") {
|
if (failedRouting === "worker_inversion") {
|
||||||
// The inversion bet on drawElement and lost — re-render on the
|
// The inversion bet on drawElement and lost — re-render on the
|
||||||
// pre-inversion parallel screenshot path instead of single-worker
|
// pre-inversion parallel screenshot path instead of single-worker
|
||||||
@@ -3243,7 +3278,17 @@ async function executeRenderPipeline(input: {
|
|||||||
resetCaptureAttemptProgress(job);
|
resetCaptureAttemptProgress(job);
|
||||||
dedupPerfs.length = 0;
|
dedupPerfs.length = 0;
|
||||||
cfg.useDrawElement = false;
|
cfg.useDrawElement = false;
|
||||||
|
// Same shape as the streaming retry above: `runCaptureStage` was
|
||||||
|
// passed the probe and threw before it could close it, so we must
|
||||||
|
// release the Chrome process ourselves before starting the
|
||||||
|
// screenshot-baseline retry — otherwise it orphans until render
|
||||||
|
// exit. See closeOrphanedProbeForRetry.
|
||||||
|
if (probeSession) {
|
||||||
|
lastBrowserConsole = probeSession.browserConsoleBuffer;
|
||||||
|
const orphaned = probeSession;
|
||||||
probeSession = null;
|
probeSession = null;
|
||||||
|
await closeOrphanedProbeForRetry(orphaned, closeCaptureSession, log, "disk verify");
|
||||||
|
}
|
||||||
capturePlan = replanAfterFailure(capturePlan, { kind: "draw_element_verification" });
|
capturePlan = replanAfterFailure(capturePlan, { kind: "draw_element_verification" });
|
||||||
syncCapturePlan();
|
syncCapturePlan();
|
||||||
updateCaptureObservability({
|
updateCaptureObservability({
|
||||||
|
|||||||
Reference in New Issue
Block a user