diff --git a/packages/producer/src/services/render/capturePlan.test.ts b/packages/producer/src/services/render/capturePlan.test.ts index 07e29f742..a8b8c0ac0 100644 --- a/packages/producer/src/services/render/capturePlan.test.ts +++ b/packages/producer/src/services/render/capturePlan.test.ts @@ -74,6 +74,11 @@ describe("CapturePlan", () => { 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" }); @@ -87,11 +92,16 @@ describe("CapturePlan", () => { expect(Object.isFrozen(next.routing)).toBe(true); }); - it("reduces an OOM fallback to one worker without losing immutable routing state", () => { + it("retries an inversion OOM in single-worker screenshot streaming mode", () => { const initial = streaming({ - kind: "parallel_router", + 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: "capture_failure", @@ -99,7 +109,31 @@ describe("CapturePlan", () => { }); expect(next).toMatchObject({ - kind: "sdr_disk", + kind: "sdr_streaming", + workerCount: 1, + forceScreenshot: true, + routing: { kind: "worker_inversion", state: "reverted" }, + }); + }); + + it("retries a parallel-router OOM in single-worker screenshot streaming mode", () => { + 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: "capture_failure", + memoryExhaustion: true, + }); + + expect(next).toMatchObject({ + kind: "sdr_streaming", workerCount: 1, forceScreenshot: true, routing: { kind: "parallel_router", state: "reverted" }, diff --git a/packages/producer/src/services/render/capturePlan.ts b/packages/producer/src/services/render/capturePlan.ts index e4a5037eb..d704f4f95 100644 --- a/packages/producer/src/services/render/capturePlan.ts +++ b/packages/producer/src/services/render/capturePlan.ts @@ -21,6 +21,7 @@ export type CaptureRouting = kind: "worker_inversion" | "parallel_router"; state: "active" | "reverted"; fallback: CapturePlanTarget; + memoryExhaustionFallback: CapturePlanTarget; }>; interface CapturePlanBase { @@ -83,7 +84,11 @@ function freezeTarget(target: CapturePlanTarget): CapturePlanTarget { function freezeRouting(routing: CaptureRouting | undefined): CaptureRouting { if (!routing || routing.kind === "default") return Object.freeze({ kind: "default" }); - return Object.freeze({ ...routing, fallback: freezeTarget(routing.fallback) }); + return Object.freeze({ + ...routing, + fallback: freezeTarget(routing.fallback), + memoryExhaustionFallback: freezeTarget(routing.memoryExhaustionFallback), + }); } export function createCapturePlan(input: CreateCapturePlanInput): CapturePlan { @@ -132,15 +137,20 @@ export function replanAfterFailure(plan: CapturePlan, failure: CapturePlanFailur }); } + const isMemoryExhaustion = failure.kind === "capture_failure" && failure.memoryExhaustion; const fallback = plan.routing.kind === "default" - ? { kind: plan.kind, workerCount: plan.workerCount, forceParallelStream: false } - : plan.routing.fallback; - const workerCount = - failure.kind === "capture_failure" && failure.memoryExhaustion ? 1 : fallback.workerCount; + ? { + kind: plan.kind, + workerCount: isMemoryExhaustion ? 1 : plan.workerCount, + forceParallelStream: false, + } + : isMemoryExhaustion + ? plan.routing.memoryExhaustionFallback + : plan.routing.fallback; return createCapturePlan({ ...plan, - workerCount, + workerCount: fallback.workerCount, forceScreenshot: true, forceParallelStream: fallback.forceParallelStream, useStreamingEncode: fallback.kind === "sdr_streaming", diff --git a/packages/producer/src/services/renderOrchestrator.ts b/packages/producer/src/services/renderOrchestrator.ts index 4f2b33545..cb9e3cdb1 100644 --- a/packages/producer/src/services/renderOrchestrator.ts +++ b/packages/producer/src/services/renderOrchestrator.ts @@ -2707,6 +2707,14 @@ async function executeRenderPipeline(input: { durationSeconds: job.duration, isMemoryExhaustion: false, }); + const inversionMemoryExhaustionFallback = resolveInversionRetryPlan({ + deWorkerInversion, + preInversionWorkerCount: preRoutingWorkerCount, + cfg, + outputFormat, + durationSeconds: job.duration, + isMemoryExhaustion: true, + }); const parallelRouterFallback = resolveParallelRouterRetryPlan({ deParallelRouter, preRouterWorkerCount: preRoutingWorkerCount, @@ -2715,27 +2723,50 @@ async function executeRenderPipeline(input: { durationSeconds: job.duration, isMemoryExhaustion: false, }); - const captureRouting: CaptureRouting = inversionFallback - ? { - kind: "worker_inversion", - state: "active", - fallback: { - kind: inversionFallback.useStreamingEncode ? "sdr_streaming" : "sdr_disk", - workerCount: inversionFallback.workerCount, - forceParallelStream: false, - }, - } - : parallelRouterFallback + const parallelRouterMemoryExhaustionFallback = resolveParallelRouterRetryPlan({ + deParallelRouter, + preRouterWorkerCount: preRoutingWorkerCount, + cfg, + outputFormat, + durationSeconds: job.duration, + isMemoryExhaustion: true, + }); + const captureRouting: CaptureRouting = + inversionFallback && inversionMemoryExhaustionFallback ? { - kind: "parallel_router", + kind: "worker_inversion", state: "active", fallback: { - kind: parallelRouterFallback.useStreamingEncode ? "sdr_streaming" : "sdr_disk", - workerCount: parallelRouterFallback.workerCount, + kind: inversionFallback.useStreamingEncode ? "sdr_streaming" : "sdr_disk", + workerCount: inversionFallback.workerCount, + forceParallelStream: false, + }, + memoryExhaustionFallback: { + kind: inversionMemoryExhaustionFallback.useStreamingEncode + ? "sdr_streaming" + : "sdr_disk", + workerCount: inversionMemoryExhaustionFallback.workerCount, forceParallelStream: false, }, } - : { kind: "default" }; + : parallelRouterFallback && parallelRouterMemoryExhaustionFallback + ? { + kind: "parallel_router", + state: "active", + fallback: { + kind: parallelRouterFallback.useStreamingEncode ? "sdr_streaming" : "sdr_disk", + workerCount: parallelRouterFallback.workerCount, + forceParallelStream: false, + }, + memoryExhaustionFallback: { + kind: parallelRouterMemoryExhaustionFallback.useStreamingEncode + ? "sdr_streaming" + : "sdr_disk", + workerCount: parallelRouterMemoryExhaustionFallback.workerCount, + forceParallelStream: false, + }, + } + : { kind: "default" }; let capturePlan: CapturePlan = createCapturePlan({ workerCount, forceScreenshot: captureForceScreenshot,