diff --git a/packages/producer/src/services/render/capturePlan.test.ts b/packages/producer/src/services/render/capturePlan.test.ts index 87ddfd9e30..b9e09cb787 100644 --- a/packages/producer/src/services/render/capturePlan.test.ts +++ b/packages/producer/src/services/render/capturePlan.test.ts @@ -1,5 +1,11 @@ import { describe, expect, it } from "vitest"; -import { createCapturePlan, replanAfterFailure, type CaptureRouting } from "./capturePlan.js"; +import { + createCapturePlan, + drawElementVerificationFailure, + replanAfterFailure, + streamingCaptureFailure, + type CaptureRouting, +} from "./capturePlan.js"; function streaming(routing?: CaptureRouting) { return createCapturePlan({ @@ -80,7 +86,10 @@ describe("CapturePlan", () => { forceParallelStream: false, }, }); - const next = replanAfterFailure(initial, { kind: "draw_element_verification" }); + const next = replanAfterFailure(initial, { + kind: "draw_element_verification", + diskFallbackAvailable: true, + }); expect(next).toMatchObject({ kind: "sdr_disk", @@ -92,6 +101,191 @@ describe("CapturePlan", () => { expect(Object.isFrozen(next.routing)).toBe(true); }); + it("keeps verification recovery streaming when the disk fallback lacks headroom", () => { + const initial = streaming({ + kind: "worker_inversion", + state: "active", + fallback: { kind: "sdr_disk", workerCount: 5, forceParallelStream: false }, + memoryExhaustionFallback: { + kind: "sdr_streaming", + workerCount: 1, + forceParallelStream: false, + }, + }); + + const next = replanAfterFailure(initial, { + kind: "draw_element_verification", + diskFallbackAvailable: false, + }); + + expect(next).toMatchObject({ + kind: "sdr_streaming", + workerCount: 1, + forceScreenshot: true, + routing: { kind: "worker_inversion", state: "reverted" }, + }); + expect(initial).toMatchObject({ workerCount: 1, routing: { state: "active" } }); + expect(Object.isFrozen(next.routing)).toBe(true); + }); + + it("keeps parallel-router verification recovery streaming when the disk fallback lacks headroom", () => { + const initial = streaming({ + kind: "parallel_router", + state: "active", + fallback: { kind: "sdr_disk", workerCount: 5, forceParallelStream: false }, + memoryExhaustionFallback: { + kind: "sdr_streaming", + workerCount: 1, + forceParallelStream: false, + }, + }); + + const next = replanAfterFailure(initial, { + kind: "draw_element_verification", + diskFallbackAvailable: false, + }); + + expect(next).toMatchObject({ + kind: "sdr_streaming", + workerCount: 1, + forceScreenshot: true, + routing: { kind: "parallel_router", state: "reverted" }, + }); + expect( + replanAfterFailure(initial, { + kind: "draw_element_verification", + diskFallbackAvailable: true, + }), + ).toMatchObject({ kind: "sdr_disk", workerCount: 5 }); + }); + + describe("drawElementVerificationFailure", () => { + const diskFallback = { kind: "sdr_disk", workerCount: 5, forceParallelStream: false } as const; + const streamFallback = { + kind: "sdr_streaming", + workerCount: 1, + forceParallelStream: false, + } as const; + + it("consults disk headroom for every routing whose disk fallback has an off-disk escape", () => { + for (const kind of ["worker_inversion", "parallel_router"] as const) { + const plan = streaming({ + kind, + state: "active", + fallback: diskFallback, + memoryExhaustionFallback: streamFallback, + }); + for (const available of [false, true]) { + let inspections = 0; + const failure = drawElementVerificationFailure(plan, () => { + inspections += 1; + return available; + }); + expect(failure).toEqual({ + kind: "draw_element_verification", + diskFallbackAvailable: available, + }); + expect(inspections).toBe(1); + } + } + }); + + it("skips the disk inspection when headroom cannot change the fallback", () => { + const inspect = () => { + throw new Error("disk must not be inspected"); + }; + const unchanged = { kind: "draw_element_verification", diskFallbackAvailable: undefined }; + + expect(drawElementVerificationFailure(streaming(), inspect)).toEqual(unchanged); + expect( + drawElementVerificationFailure( + streaming({ + kind: "parallel_router", + state: "active", + fallback: streamFallback, + memoryExhaustionFallback: streamFallback, + }), + inspect, + ), + ).toEqual(unchanged); + expect( + drawElementVerificationFailure( + streaming({ + kind: "worker_inversion", + state: "active", + fallback: diskFallback, + memoryExhaustionFallback: { ...diskFallback, workerCount: 1 }, + }), + inspect, + ), + ).toEqual(unchanged); + }); + }); + + describe("streamingCaptureFailure", () => { + // The routing shape for which a verification failure DOES consult disk + // headroom — so any inspector call below would be on that path alone. + const routedPlan = streaming({ + kind: "worker_inversion", + state: "active", + fallback: { kind: "sdr_disk", workerCount: 5, forceParallelStream: false }, + memoryExhaustionFallback: { + kind: "sdr_streaming", + workerCount: 1, + forceParallelStream: false, + }, + }); + const inspect = () => { + throw new Error("disk must not be inspected"); + }; + + // A canvas / paint-record capture error, a renderer stall and an OOM are + // not verification failures: each retries as a plain capture failure and + // must not touch the disk. + it.each([false, true])( + "never inspects disk headroom for a non-verification failure (memoryExhaustion: %s)", + (isMemoryExhaustion) => { + expect( + streamingCaptureFailure( + routedPlan, + { isVerifyError: false, isMemoryExhaustion }, + inspect, + ), + ).toEqual({ kind: "capture_failure", memoryExhaustion: isMemoryExhaustion }); + }, + ); + + it("reverts to the preferred disk fallback on a drawElement capture failure", () => { + // The streaming-side capture failure carries no headroom flag, so the + // retry takes the routing's preferred fallback without steering. + const failure = streamingCaptureFailure( + routedPlan, + { isVerifyError: false, isMemoryExhaustion: false }, + inspect, + ); + expect(replanAfterFailure(routedPlan, failure)).toMatchObject({ + kind: "sdr_disk", + workerCount: 5, + forceScreenshot: true, + routing: { kind: "worker_inversion", state: "reverted" }, + }); + }); + + it("routes a verification failure through the headroom-aware builder", () => { + let inspections = 0; + const failure = streamingCaptureFailure( + routedPlan, + { isVerifyError: true, isMemoryExhaustion: false }, + () => { + inspections += 1; + return false; + }, + ); + expect(failure).toEqual({ kind: "draw_element_verification", diskFallbackAvailable: false }); + expect(inspections).toBe(1); + }); + }); + it("retries an inversion OOM in single-worker screenshot streaming mode", () => { const initial = streaming({ kind: "worker_inversion", diff --git a/packages/producer/src/services/render/capturePlan.ts b/packages/producer/src/services/render/capturePlan.ts index 3a1e563ed9..3b5eac1c25 100644 --- a/packages/producer/src/services/render/capturePlan.ts +++ b/packages/producer/src/services/render/capturePlan.ts @@ -65,10 +65,58 @@ export interface CreateCapturePlanInput { export type CapturePlanFailure = | Readonly<{ kind: "streaming_unavailable" }> - | Readonly<{ kind: "draw_element_verification" }> + | Readonly<{ + kind: "draw_element_verification"; + /** + * Set by `drawElementVerificationFailure` only when the routing's + * preferred fallback writes frames to disk and its precomputed + * low-resource target does not. `false` means that disk route lacks + * headroom, so `replanAfterFailure` takes the off-disk target instead. + */ + diskFallbackAvailable?: boolean; + }> | Readonly<{ kind: "draw_element_capture" }> | Readonly<{ kind: "capture_failure"; memoryExhaustion: boolean }>; +/** + * Build the verification failure for `plan`, consulting disk headroom only + * when the answer can change the fallback: a non-default routing (worker + * inversion or parallel router) whose preferred fallback is `sdr_disk` while + * its memory-exhaustion fallback is off-disk. Every other plan leaves the flag + * `undefined` without calling `hasDiskFallbackHeadroom`. + */ +export function drawElementVerificationFailure( + plan: CapturePlan, + hasDiskFallbackHeadroom: () => boolean, +): CapturePlanFailure { + const { routing } = plan; + const needsHeadroom = + routing.kind !== "default" && + routing.fallback.kind === "sdr_disk" && + routing.memoryExhaustionFallback.kind === "sdr_streaming"; + return { + kind: "draw_element_verification", + diskFallbackAvailable: needsHeadroom ? hasDiskFallbackHeadroom() : undefined, + }; +} + +/** + * Build the failure the streaming drain retries with, from the caller's + * classification of the capture error. Only a drawElement self-verification + * failure goes through `drawElementVerificationFailure` (and so may consult + * disk headroom); every other failure is a plain capture failure and never + * calls `hasDiskFallbackHeadroom`. + */ +export function streamingCaptureFailure( + plan: CapturePlan, + classification: Readonly<{ isVerifyError: boolean; isMemoryExhaustion: boolean }>, + hasDiskFallbackHeadroom: () => boolean, +): CapturePlanFailure { + return classification.isVerifyError + ? drawElementVerificationFailure(plan, hasDiskFallbackHeadroom) + : { kind: "capture_failure", memoryExhaustion: classification.isMemoryExhaustion }; +} + function assertWorkerCount(workerCount: number): void { if (!Number.isInteger(workerCount) || workerCount < 1) { throw new Error(`CapturePlan workerCount must be a positive integer; got ${workerCount}`); @@ -154,6 +202,12 @@ export function replanAfterFailure(plan: CapturePlan, failure: CapturePlanFailur } const isMemoryExhaustion = failure.kind === "capture_failure" && failure.memoryExhaustion; + const diskFallbackUnavailable = + failure.kind === "draw_element_verification" && failure.diskFallbackAvailable === false; + // memoryExhaustionFallback is the routing decision's precomputed + // low-resource target. It is also the viable choice when disk, rather than + // RAM, makes the preferred fallback impossible — for any routing kind + // (see drawElementVerificationFailure for when that flag is populated). const fallback = plan.routing.kind === "default" ? { @@ -161,7 +215,7 @@ export function replanAfterFailure(plan: CapturePlan, failure: CapturePlanFailur workerCount: isMemoryExhaustion ? 1 : plan.workerCount, forceParallelStream: false, } - : isMemoryExhaustion + : isMemoryExhaustion || diskFallbackUnavailable ? plan.routing.memoryExhaustionFallback : plan.routing.fallback; return createCapturePlan({ diff --git a/packages/producer/src/services/render/stages/captureStage.test.ts b/packages/producer/src/services/render/stages/captureStage.test.ts index 60dc2703c5..45448f6048 100644 --- a/packages/producer/src/services/render/stages/captureStage.test.ts +++ b/packages/producer/src/services/render/stages/captureStage.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import { assertDiskCaptureHeadroom, estimateDiskCaptureBytes, + inspectDiskCaptureHeadroom, shouldAllowAdaptiveCaptureRetry, } from "./captureStage.js"; @@ -33,4 +34,39 @@ describe("disk capture capacity", () => { assertDiskCaptureHeadroom("/render/captured-frames", 10, captureOptions, () => 800_000), ).toThrow(/may need ~0\.8 MB.*0\.8 MB is free.*--low-memory-mode/s); }); + + it("exposes the same 90% headroom decision to fallback planning", () => { + const estimatedBytes = estimateDiskCaptureBytes(10, captureOptions); + + expect( + inspectDiskCaptureHeadroom( + "/render/captured-frames", + 10, + captureOptions, + () => estimatedBytes / 0.9 - 1, + ), + ).toEqual({ available: false, estimatedBytes, freeBytes: estimatedBytes / 0.9 - 1 }); + expect( + inspectDiskCaptureHeadroom("/render/captured-frames", 10, captureOptions, () => null) + .available, + ).toBe(true); + }); + + it("rejects the reported 5318-frame landscape disk route with 45 GiB free", () => { + const landscape = { + width: 1920, + height: 1080, + fps: { num: 30, den: 1 }, + format: "jpeg" as const, + }; + const headroom = inspectDiskCaptureHeadroom( + "/render/captured-frames", + 5318, + landscape, + () => 45 * 1024 ** 3, + ); + + expect(headroom.estimatedBytes).toBe(5318 * 1920 * 1080 * 4); + expect(headroom.available).toBe(false); + }); }); diff --git a/packages/producer/src/services/render/stages/captureStage.ts b/packages/producer/src/services/render/stages/captureStage.ts index f97a1de96c..b50ee2430c 100644 --- a/packages/producer/src/services/render/stages/captureStage.ts +++ b/packages/producer/src/services/render/stages/captureStage.ts @@ -147,7 +147,16 @@ export function estimateDiskCaptureBytes( return Math.ceil(totalFrames) * outputWidth * outputHeight * 4; } -export function assertDiskCaptureHeadroom( +/** + * Unknown free space (`freeBytes: null`) counts as available: the gate can + * only reject when it has a measurement. + */ +export type DiskCaptureHeadroom = + | { available: true; estimatedBytes: number; freeBytes: number | null } + | { available: false; estimatedBytes: number; freeBytes: number }; + +/** Shared 90% disk gate used by both fallback planning and disk execution. */ +export function inspectDiskCaptureHeadroom( framesDir: string, totalFrames: number, captureOptions: CaptureOptions, @@ -159,14 +168,31 @@ export function assertDiskCaptureHeadroom( return null; } }, -): void { +): DiskCaptureHeadroom { const freeBytes = freeDiskBytes(framesDir); - if (freeBytes === null) return; const estimatedBytes = estimateDiskCaptureBytes(totalFrames, captureOptions); - if (estimatedBytes <= freeBytes * 0.9) return; + if (freeBytes === null || estimatedBytes <= freeBytes * 0.9) { + return { available: true, estimatedBytes, freeBytes }; + } + return { available: false, estimatedBytes, freeBytes }; +} + +export function assertDiskCaptureHeadroom( + framesDir: string, + totalFrames: number, + captureOptions: CaptureOptions, + freeDiskBytes?: (path: string) => number | null, +): void { + const headroom = inspectDiskCaptureHeadroom( + framesDir, + totalFrames, + captureOptions, + freeDiskBytes, + ); + if (headroom.available) return; throw new Error( - `Disk capture may need ~${(estimatedBytes / 1e6).toFixed(1)} MB of temporary frame storage, ` + - `but only ${(freeBytes / 1e6).toFixed(1)} MB is free at ${framesDir}. ` + + `Disk capture may need ~${(headroom.estimatedBytes / 1e6).toFixed(1)} MB of temporary frame storage, ` + + `but only ${(headroom.freeBytes / 1e6).toFixed(1)} MB is free at ${framesDir}. ` + "Re-run with --low-memory-mode to stream frames, raise " + "PRODUCER_STREAMING_ENCODE_MAX_DURATION_SECONDS if streaming is supported, " + "or free up disk space.", diff --git a/packages/producer/src/services/renderOrchestrator.ts b/packages/producer/src/services/renderOrchestrator.ts index 2f0319c680..427f152991 100644 --- a/packages/producer/src/services/renderOrchestrator.ts +++ b/packages/producer/src/services/renderOrchestrator.ts @@ -119,6 +119,7 @@ import { ArtifactTransaction } from "./render/artifactTransaction.js"; import { createCapturePlan, replanAfterFailure, + streamingCaptureFailure, type CapturePlan, type SdrDiskCapturePlan, type CaptureRouting, @@ -164,7 +165,7 @@ import { runProbeStage } from "./render/stages/probeStage.js"; import { validateRenderDuration } from "./render/planValidation.js"; import { runExtractVideosStage } from "./render/stages/extractVideosStage.js"; import { runAudioStage } from "./render/stages/audioStage.js"; -import { runCaptureStage } from "./render/stages/captureStage.js"; +import { inspectDiskCaptureHeadroom, runCaptureStage } from "./render/stages/captureStage.js"; import { type CaptureStreamingStageResult, runCaptureStreamingStage, @@ -3780,12 +3781,13 @@ async function executeRenderPipeline(input: { : "capture failed; retrying with a fresh screenshot session", ); const failedRouting = capturePlan.routing.kind; - capturePlan = replanAfterFailure( + const failure = streamingCaptureFailure( capturePlan, - isVerifyError - ? { kind: "draw_element_verification" } - : { kind: "capture_failure", memoryExhaustion: isMemoryExhaustion }, + { isVerifyError, isMemoryExhaustion }, + () => + inspectDiskCaptureHeadroom(framesDir, totalFrames, buildCaptureOptions()).available, ); + capturePlan = replanAfterFailure(capturePlan, failure); syncCapturePlan(); updateCaptureObservability({ forceScreenshot: capturePlan.forceScreenshot, @@ -3809,24 +3811,26 @@ async function executeRenderPipeline(input: { probeSession = null; await closeOrphanedProbeForRetry(orphaned, closeCaptureSession, log, "streaming"); } - if (failedRouting === "worker_inversion") { - // The inversion bet on drawElement and lost — re-render on the - // pre-inversion parallel screenshot path instead of single-worker - // screenshot streaming (the slowest capture shape for this size). - // "reverted" (not cleared) so telemetry keeps the lost-inversion - // cohort distinguishable from renders that never inverted. + if (failedRouting !== "default") { + // The inversion's / router's bet on the pinned path lost. Prefer + // the pre-routing parallel screenshot route; a verification + // failure alone also retains the already supported low-worker + // screenshot stream when that disk route lacks storage headroom + // (OOM takes the same target for RAM; other capture failures + // revert unconditionally). Routing stays "reverted" (not + // cleared) so telemetry keeps the lost-bet cohort distinguishable + // from renders that never routed. + const route = + failedRouting === "worker_inversion" ? "worker inversion" : "parallel router"; + const diskFallbackLacksHeadroom = + failure.kind === "draw_element_verification" && + failure.diskFallbackAvailable === false; log.info( - `[Render] Reverting worker inversion for the retry: ${capturePlan.workerCount} workers, ` + - `plan=${capturePlan.kind}.`, - ); - } else if (failedRouting === "parallel_router") { - // The router's bet on verified parallel streaming lost — re-render - // on the ordinary (non-DE) parallel path at the pre-router worker - // count, same "reverted, not cleared" telemetry contract as the - // inversion above. - log.info( - `[Render] Reverting parallel router for the retry: ${capturePlan.workerCount} workers, ` + - `plan=${capturePlan.kind}.`, + diskFallbackLacksHeadroom + ? `[Render] ${route} disk fallback lacks headroom; retrying with ` + + `${capturePlan.workerCount}-worker screenshot streaming.` + : `[Render] Reverting ${route} for the retry: ${capturePlan.workerCount} workers, ` + + `plan=${capturePlan.kind}.`, ); } if (capturePlan.kind === "sdr_streaming") {