refactor(producer): make capture plans immutable

packages/producer/src/services/distributed/renderChunk.ts

@@ -59,6 +59,7 @@ import { defaultLogger } from "../../logger.js";
 import { runEncodeStage } from "../render/stages/encodeStage.js";
 import { runCaptureStage } from "../render/stages/captureStage.js";
 import { resolveVideoCaptureBeyondViewport } from "../render/captureBeyondViewport.js";
+import { createCapturePlan } from "../render/capturePlan.js";
 import {
   type ChunkSliceJson,
   type LockedRenderConfig,
@@ -589,18 +590,28 @@ export async function renderChunk(
       // one Chrome session per worker, so captureStageMs includes those
       // boots; sessionBootMs stays 0 there.
       const captureStarted = Date.now();
+      const capturePlan = createCapturePlan({
+        workerCount: chunkWorkerCount,
+        forceScreenshot: encoder.forceScreenshot,
+        useStreamingEncode: false,
+        useLayeredComposite: false,
+        usePageSideCompositing: false,
+        hasHdrContent: false,
+        needsAlpha: plan.dimensions.format !== "mp4",
+      });
+      if (capturePlan.kind !== "sdr_disk") {
+        throw new Error(`Distributed chunk requires sdr_disk plan; got ${capturePlan.kind}`);
+      }
       await runCaptureStage({
         fileServer,
         workDir,
         framesDir,
         job,
         totalFrames: framesInChunk,
         cfg,
-        forceScreenshot: encoder.forceScreenshot,
+        plan: capturePlan,
         log,
-        workerCount: chunkWorkerCount,
         probeSession: session,
-        needsAlpha: plan.dimensions.format !== "mp4",
         captureAttempts: [],
         // Distributed chunks run on Linux (beginframe) where dedup never arms;
         // a throwaway sink satisfies the type without per-chunk dedup reporting.

packages/producer/src/services/render/capturePlan.test.ts

@@ -0,0 +1,123 @@
+import { describe, expect, it } from "vitest";
+import { createCapturePlan, replanAfterFailure, type CaptureRouting } from "./capturePlan.js";
+
+function streaming(routing?: CaptureRouting) {
+  return createCapturePlan({
+    workerCount: 1,
+    forceScreenshot: false,
+    forceParallelStream: false,
+    useStreamingEncode: true,
+    useLayeredComposite: false,
+    usePageSideCompositing: false,
+    hasHdrContent: false,
+    needsAlpha: false,
+    routing,
+  });
+}
+
+describe("CapturePlan", () => {
+  it("makes layered capture dominant and enforces its screenshot invariant", () => {
+    const plan = createCapturePlan({
+      workerCount: 3,
+      forceScreenshot: false,
+      forceParallelStream: true,
+      useStreamingEncode: true,
+      useLayeredComposite: true,
+      usePageSideCompositing: false,
+      hasHdrContent: true,
+      needsAlpha: false,
+    });
+
+    expect(plan).toMatchObject({
+      kind: "hdr_layered",
+      workerCount: 3,
+      forceScreenshot: true,
+      forceParallelStream: false,
+    });
+    expect(Object.isFrozen(plan)).toBe(true);
+    expect(Object.isFrozen(plan.routing)).toBe(true);
+  });
+
+  it("falls back from an unavailable streaming encoder to the same disk route", () => {
+    const initial = streaming();
+    const next = replanAfterFailure(initial, { kind: "streaming_unavailable" });
+
+    expect(next).toMatchObject({ kind: "sdr_disk", workerCount: 1, forceScreenshot: false });
+    expect(initial.kind).toBe("sdr_streaming");
+  });
+
+  it("makes page-side compositing force screenshot capture", () => {
+    const plan = createCapturePlan({
+      workerCount: 1,
+      forceScreenshot: false,
+      forceParallelStream: false,
+      useStreamingEncode: true,
+      useLayeredComposite: false,
+      usePageSideCompositing: true,
+      hasHdrContent: false,
+      needsAlpha: false,
+    });
+    expect(plan).toMatchObject({ kind: "sdr_streaming", forceScreenshot: true });
+  });
+
+  it("retries an ordinary verification failure in streaming screenshot mode", () => {
+    expect(replanAfterFailure(streaming(), { kind: "draw_element_verification" })).toMatchObject({
+      kind: "sdr_streaming",
+      workerCount: 1,
+      forceScreenshot: true,
+      routing: { kind: "default" },
+    });
+  });
+
+  it("atomically restores the pre-inversion disk route after verification failure", () => {
+    const initial = streaming({
+      kind: "worker_inversion",
+      state: "active",
+      fallback: { kind: "sdr_disk", workerCount: 5, forceParallelStream: false },
+    });
+    const next = replanAfterFailure(initial, { kind: "draw_element_verification" });
+
+    expect(next).toMatchObject({
+      kind: "sdr_disk",
+      workerCount: 5,
+      forceScreenshot: true,
+      routing: { kind: "worker_inversion", state: "reverted" },
+    });
+    expect(initial).toMatchObject({ workerCount: 1, routing: { state: "active" } });
+    expect(Object.isFrozen(next.routing)).toBe(true);
+  });
+
+  it("reduces an OOM fallback to one worker without losing immutable routing state", () => {
+    const initial = streaming({
+      kind: "parallel_router",
+      state: "active",
+      fallback: { kind: "sdr_disk", workerCount: 5, forceParallelStream: false },
+    });
+    const next = replanAfterFailure(initial, {
+      kind: "capture_failure",
+      memoryExhaustion: true,
+    });
+
+    expect(next).toMatchObject({
+      kind: "sdr_disk",
+      workerCount: 1,
+      forceScreenshot: true,
+      routing: { kind: "parallel_router", state: "reverted" },
+    });
+  });
+
+  it("rejects a streaming transition from a non-streaming plan", () => {
+    const disk = createCapturePlan({
+      workerCount: 2,
+      forceScreenshot: true,
+      useStreamingEncode: false,
+      useLayeredComposite: false,
+      usePageSideCompositing: false,
+      hasHdrContent: false,
+      needsAlpha: false,
+    });
+    expect(() => replanAfterFailure(disk, { kind: "streaming_unavailable" })).toThrow(
+      "Cannot apply streaming_unavailable to sdr_disk",
+    );
+  });
+});

packages/producer/src/services/render/capturePlan.ts

@@ -0,0 +1,150 @@
+/**
+ * Immutable routing decision for the capture phase.
+ *
+ * The orchestrator used to carry the same decision in several mutable booleans
+ * (`useStreamingEncode`, `useLayeredComposite`, `forceScreenshot`) plus worker
+ * routing state. Keeping those values independently mutable made invalid
+ * combinations representable during fallback. A CapturePlan is the single
+ * value consumed by capture stages, and `replanAfterFailure` is the only
+ * transition between variants.
+ */
+
+export type CapturePlanTarget = Readonly<{
+  kind: "sdr_streaming" | "sdr_disk";
+  workerCount: number;
+  forceParallelStream: boolean;
+}>;
+
+export type CaptureRouting =
+  | Readonly<{ kind: "default" }>
+  | Readonly<{
+      kind: "worker_inversion" | "parallel_router";
+      state: "active" | "reverted";
+      fallback: CapturePlanTarget;
+    }>;
+
+interface CapturePlanBase {
+  readonly workerCount: number;
+  readonly forceScreenshot: boolean;
+  readonly forceParallelStream: boolean;
+  readonly usePageSideCompositing: boolean;
+  readonly hasHdrContent: boolean;
+  readonly needsAlpha: boolean;
+  readonly routing: CaptureRouting;
+}
+
+export interface SdrStreamingCapturePlan extends CapturePlanBase {
+  readonly kind: "sdr_streaming";
+}
+
+export interface SdrDiskCapturePlan extends CapturePlanBase {
+  readonly kind: "sdr_disk";
+  readonly forceParallelStream: false;
+}
+
+export interface HdrLayeredCapturePlan extends CapturePlanBase {
+  readonly kind: "hdr_layered";
+  readonly forceScreenshot: true;
+  readonly forceParallelStream: false;
+}
+
+export type CapturePlan = SdrStreamingCapturePlan | SdrDiskCapturePlan | HdrLayeredCapturePlan;
+
+export interface CreateCapturePlanInput {
+  workerCount: number;
+  forceScreenshot: boolean;
+  forceParallelStream?: boolean;
+  useStreamingEncode: boolean;
+  useLayeredComposite: boolean;
+  usePageSideCompositing: boolean;
+  hasHdrContent: boolean;
+  needsAlpha: boolean;
+  routing?: CaptureRouting;
+}
+
+export type CapturePlanFailure =
+  | Readonly<{ kind: "streaming_unavailable" }>
+  | Readonly<{ kind: "draw_element_verification" }>
+  | Readonly<{ kind: "capture_failure"; memoryExhaustion: boolean }>;
+
+function assertWorkerCount(workerCount: number): void {
+  if (!Number.isInteger(workerCount) || workerCount < 1) {
+    throw new Error(`CapturePlan workerCount must be a positive integer; got ${workerCount}`);
+  }
+}
+
+function freezeTarget(target: CapturePlanTarget): CapturePlanTarget {
+  assertWorkerCount(target.workerCount);
+  if (target.kind === "sdr_disk" && target.forceParallelStream) {
+    throw new Error("CapturePlan disk fallback cannot force parallel streaming");
+  }
+  return Object.freeze({ ...target });
+}
+
+function freezeRouting(routing: CaptureRouting | undefined): CaptureRouting {
+  if (!routing || routing.kind === "default") return Object.freeze({ kind: "default" });
+  return Object.freeze({ ...routing, fallback: freezeTarget(routing.fallback) });
+}
+
+export function createCapturePlan(input: CreateCapturePlanInput): CapturePlan {
+  assertWorkerCount(input.workerCount);
+  const base = {
+    workerCount: input.workerCount,
+    forceScreenshot: input.forceScreenshot || input.usePageSideCompositing,
+    forceParallelStream: input.useStreamingEncode ? (input.forceParallelStream ?? false) : false,
+    usePageSideCompositing: input.usePageSideCompositing,
+    hasHdrContent: input.hasHdrContent,
+    needsAlpha: input.needsAlpha,
+    routing: freezeRouting(input.routing),
+  };
+
+  if (input.useLayeredComposite) {
+    return Object.freeze({
+      ...base,
+      kind: "hdr_layered",
+      forceScreenshot: true,
+      forceParallelStream: false,
+    });
+  }
+  if (input.useStreamingEncode) {
+    return Object.freeze({ ...base, kind: "sdr_streaming" });
+  }
+  return Object.freeze({ ...base, kind: "sdr_disk", forceParallelStream: false });
+}
+
+function revertedRouting(routing: CaptureRouting): CaptureRouting {
+  if (routing.kind === "default") return routing;
+  return freezeRouting({ ...routing, state: "reverted" });
+}
+
+/** Pure, exhaustive capture fallback transition. The input plan is never mutated. */
+export function replanAfterFailure(plan: CapturePlan, failure: CapturePlanFailure): CapturePlan {
+  if (plan.kind !== "sdr_streaming") {
+    throw new Error(`Cannot apply ${failure.kind} to ${plan.kind} capture plan`);
+  }
+
+  if (failure.kind === "streaming_unavailable") {
+    return createCapturePlan({
+      ...plan,
+      useStreamingEncode: false,
+      useLayeredComposite: false,
+      forceParallelStream: false,
+    });
+  }
+
+  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;
+  return createCapturePlan({
+    ...plan,
+    workerCount,
+    forceScreenshot: true,
+    forceParallelStream: fallback.forceParallelStream,
+    useStreamingEncode: fallback.kind === "sdr_streaming",
+    useLayeredComposite: false,
+    routing: revertedRouting(plan.routing),
+  });
+}