Skip to content

Commit 2db2ea7

Browse files
jrusso1020jrusso1020
committed
refactor(producer): make capture plans immutable
1 parent e08d6df commit 2db2ea7

8 files changed

Lines changed: 457 additions & 154 deletions

File tree

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

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,7 @@ import { defaultLogger } from "../../logger.js";
5959
import { runEncodeStage } from "../render/stages/encodeStage.js";
6060
import { runCaptureStage } from "../render/stages/captureStage.js";
6161
import { resolveVideoCaptureBeyondViewport } from "../render/captureBeyondViewport.js";
62+
import { createCapturePlan } from "../render/capturePlan.js";
6263
import {
6364
type ChunkSliceJson,
6465
type LockedRenderConfig,
@@ -589,18 +590,28 @@ export async function renderChunk(
589590
// one Chrome session per worker, so captureStageMs includes those
590591
// boots; sessionBootMs stays 0 there.
591592
const captureStarted = Date.now();
593+
const capturePlan = createCapturePlan({
594+
workerCount: chunkWorkerCount,
595+
forceScreenshot: encoder.forceScreenshot,
596+
useStreamingEncode: false,
597+
useLayeredComposite: false,
598+
usePageSideCompositing: false,
599+
hasHdrContent: false,
600+
needsAlpha: plan.dimensions.format !== "mp4",
601+
});
602+
if (capturePlan.kind !== "sdr_disk") {
603+
throw new Error(`Distributed chunk requires sdr_disk plan; got ${capturePlan.kind}`);
604+
}
592605
await runCaptureStage({
593606
fileServer,
594607
workDir,
595608
framesDir,
596609
job,
597610
totalFrames: framesInChunk,
598611
cfg,
599-
forceScreenshot: encoder.forceScreenshot,
612+
plan: capturePlan,
600613
log,
601-
workerCount: chunkWorkerCount,
602614
probeSession: session,
603-
needsAlpha: plan.dimensions.format !== "mp4",
604615
captureAttempts: [],
605616
// Distributed chunks run on Linux (beginframe) where dedup never arms;
606617
// a throwaway sink satisfies the type without per-chunk dedup reporting.

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

Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,123 @@
1+
import { describe, expect, it } from "vitest";
2+
import { createCapturePlan, replanAfterFailure, type CaptureRouting } from "./capturePlan.js";
3+
4+
function streaming(routing?: CaptureRouting) {
5+
return createCapturePlan({
6+
workerCount: 1,
7+
forceScreenshot: false,
8+
forceParallelStream: false,
9+
useStreamingEncode: true,
10+
useLayeredComposite: false,
11+
usePageSideCompositing: false,
12+
hasHdrContent: false,
13+
needsAlpha: false,
14+
routing,
15+
});
16+
}
17+
18+
describe("CapturePlan", () => {
19+
it("makes layered capture dominant and enforces its screenshot invariant", () => {
20+
const plan = createCapturePlan({
21+
workerCount: 3,
22+
forceScreenshot: false,
23+
forceParallelStream: true,
24+
useStreamingEncode: true,
25+
useLayeredComposite: true,
26+
usePageSideCompositing: false,
27+
hasHdrContent: true,
28+
needsAlpha: false,
29+
});
30+
31+
expect(plan).toMatchObject({
32+
kind: "hdr_layered",
33+
workerCount: 3,
34+
forceScreenshot: true,
35+
forceParallelStream: false,
36+
});
37+
expect(Object.isFrozen(plan)).toBe(true);
38+
expect(Object.isFrozen(plan.routing)).toBe(true);
39+
});
40+
41+
it("falls back from an unavailable streaming encoder to the same disk route", () => {
42+
const initial = streaming();
43+
const next = replanAfterFailure(initial, { kind: "streaming_unavailable" });
44+
45+
expect(next).toMatchObject({ kind: "sdr_disk", workerCount: 1, forceScreenshot: false });
46+
expect(initial.kind).toBe("sdr_streaming");
47+
});
48+
49+
it("makes page-side compositing force screenshot capture", () => {
50+
const plan = createCapturePlan({
51+
workerCount: 1,
52+
forceScreenshot: false,
53+
forceParallelStream: false,
54+
useStreamingEncode: true,
55+
useLayeredComposite: false,
56+
usePageSideCompositing: true,
57+
hasHdrContent: false,
58+
needsAlpha: false,
59+
});
60+
expect(plan).toMatchObject({ kind: "sdr_streaming", forceScreenshot: true });
61+
});
62+
63+
it("retries an ordinary verification failure in streaming screenshot mode", () => {
64+
expect(replanAfterFailure(streaming(), { kind: "draw_element_verification" })).toMatchObject({
65+
kind: "sdr_streaming",
66+
workerCount: 1,
67+
forceScreenshot: true,
68+
routing: { kind: "default" },
69+
});
70+
});
71+
72+
it("atomically restores the pre-inversion disk route after verification failure", () => {
73+
const initial = streaming({
74+
kind: "worker_inversion",
75+
state: "active",
76+
fallback: { kind: "sdr_disk", workerCount: 5, forceParallelStream: false },
77+
});
78+
const next = replanAfterFailure(initial, { kind: "draw_element_verification" });
79+
80+
expect(next).toMatchObject({
81+
kind: "sdr_disk",
82+
workerCount: 5,
83+
forceScreenshot: true,
84+
routing: { kind: "worker_inversion", state: "reverted" },
85+
});
86+
expect(initial).toMatchObject({ workerCount: 1, routing: { state: "active" } });
87+
expect(Object.isFrozen(next.routing)).toBe(true);
88+
});
89+
90+
it("reduces an OOM fallback to one worker without losing immutable routing state", () => {
91+
const initial = streaming({
92+
kind: "parallel_router",
93+
state: "active",
94+
fallback: { kind: "sdr_disk", workerCount: 5, forceParallelStream: false },
95+
});
96+
const next = replanAfterFailure(initial, {
97+
kind: "capture_failure",
98+
memoryExhaustion: true,
99+
});
100+
101+
expect(next).toMatchObject({
102+
kind: "sdr_disk",
103+
workerCount: 1,
104+
forceScreenshot: true,
105+
routing: { kind: "parallel_router", state: "reverted" },
106+
});
107+
});
108+
109+
it("rejects a streaming transition from a non-streaming plan", () => {
110+
const disk = createCapturePlan({
111+
workerCount: 2,
112+
forceScreenshot: true,
113+
useStreamingEncode: false,
114+
useLayeredComposite: false,
115+
usePageSideCompositing: false,
116+
hasHdrContent: false,
117+
needsAlpha: false,
118+
});
119+
expect(() => replanAfterFailure(disk, { kind: "streaming_unavailable" })).toThrow(
120+
"Cannot apply streaming_unavailable to sdr_disk",
121+
);
122+
});
123+
});

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

Lines changed: 150 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,150 @@
1+
/**
2+
* Immutable routing decision for the capture phase.
3+
*
4+
* The orchestrator used to carry the same decision in several mutable booleans
5+
* (`useStreamingEncode`, `useLayeredComposite`, `forceScreenshot`) plus worker
6+
* routing state. Keeping those values independently mutable made invalid
7+
* combinations representable during fallback. A CapturePlan is the single
8+
* value consumed by capture stages, and `replanAfterFailure` is the only
9+
* transition between variants.
10+
*/
11+
12+
export type CapturePlanTarget = Readonly<{
13+
kind: "sdr_streaming" | "sdr_disk";
14+
workerCount: number;
15+
forceParallelStream: boolean;
16+
}>;
17+
18+
export type CaptureRouting =
19+
| Readonly<{ kind: "default" }>
20+
| Readonly<{
21+
kind: "worker_inversion" | "parallel_router";
22+
state: "active" | "reverted";
23+
fallback: CapturePlanTarget;
24+
}>;
25+
26+
interface CapturePlanBase {
27+
readonly workerCount: number;
28+
readonly forceScreenshot: boolean;
29+
readonly forceParallelStream: boolean;
30+
readonly usePageSideCompositing: boolean;
31+
readonly hasHdrContent: boolean;
32+
readonly needsAlpha: boolean;
33+
readonly routing: CaptureRouting;
34+
}
35+
36+
export interface SdrStreamingCapturePlan extends CapturePlanBase {
37+
readonly kind: "sdr_streaming";
38+
}
39+
40+
export interface SdrDiskCapturePlan extends CapturePlanBase {
41+
readonly kind: "sdr_disk";
42+
readonly forceParallelStream: false;
43+
}
44+
45+
export interface HdrLayeredCapturePlan extends CapturePlanBase {
46+
readonly kind: "hdr_layered";
47+
readonly forceScreenshot: true;
48+
readonly forceParallelStream: false;
49+
}
50+
51+
export type CapturePlan = SdrStreamingCapturePlan | SdrDiskCapturePlan | HdrLayeredCapturePlan;
52+
53+
export interface CreateCapturePlanInput {
54+
workerCount: number;
55+
forceScreenshot: boolean;
56+
forceParallelStream?: boolean;
57+
useStreamingEncode: boolean;
58+
useLayeredComposite: boolean;
59+
usePageSideCompositing: boolean;
60+
hasHdrContent: boolean;
61+
needsAlpha: boolean;
62+
routing?: CaptureRouting;
63+
}
64+
65+
export type CapturePlanFailure =
66+
| Readonly<{ kind: "streaming_unavailable" }>
67+
| Readonly<{ kind: "draw_element_verification" }>
68+
| Readonly<{ kind: "capture_failure"; memoryExhaustion: boolean }>;
69+
70+
function assertWorkerCount(workerCount: number): void {
71+
if (!Number.isInteger(workerCount) || workerCount < 1) {
72+
throw new Error(`CapturePlan workerCount must be a positive integer; got ${workerCount}`);
73+
}
74+
}
75+
76+
function freezeTarget(target: CapturePlanTarget): CapturePlanTarget {
77+
assertWorkerCount(target.workerCount);
78+
if (target.kind === "sdr_disk" && target.forceParallelStream) {
79+
throw new Error("CapturePlan disk fallback cannot force parallel streaming");
80+
}
81+
return Object.freeze({ ...target });
82+
}
83+
84+
function freezeRouting(routing: CaptureRouting | undefined): CaptureRouting {
85+
if (!routing || routing.kind === "default") return Object.freeze({ kind: "default" });
86+
return Object.freeze({ ...routing, fallback: freezeTarget(routing.fallback) });
87+
}
88+
89+
export function createCapturePlan(input: CreateCapturePlanInput): CapturePlan {
90+
assertWorkerCount(input.workerCount);
91+
const base = {
92+
workerCount: input.workerCount,
93+
forceScreenshot: input.forceScreenshot || input.usePageSideCompositing,
94+
forceParallelStream: input.useStreamingEncode ? (input.forceParallelStream ?? false) : false,
95+
usePageSideCompositing: input.usePageSideCompositing,
96+
hasHdrContent: input.hasHdrContent,
97+
needsAlpha: input.needsAlpha,
98+
routing: freezeRouting(input.routing),
99+
};
100+
101+
if (input.useLayeredComposite) {
102+
return Object.freeze({
103+
...base,
104+
kind: "hdr_layered",
105+
forceScreenshot: true,
106+
forceParallelStream: false,
107+
});
108+
}
109+
if (input.useStreamingEncode) {
110+
return Object.freeze({ ...base, kind: "sdr_streaming" });
111+
}
112+
return Object.freeze({ ...base, kind: "sdr_disk", forceParallelStream: false });
113+
}
114+
115+
function revertedRouting(routing: CaptureRouting): CaptureRouting {
116+
if (routing.kind === "default") return routing;
117+
return freezeRouting({ ...routing, state: "reverted" });
118+
}
119+
120+
/** Pure, exhaustive capture fallback transition. The input plan is never mutated. */
121+
export function replanAfterFailure(plan: CapturePlan, failure: CapturePlanFailure): CapturePlan {
122+
if (plan.kind !== "sdr_streaming") {
123+
throw new Error(`Cannot apply ${failure.kind} to ${plan.kind} capture plan`);
124+
}
125+
126+
if (failure.kind === "streaming_unavailable") {
127+
return createCapturePlan({
128+
...plan,
129+
useStreamingEncode: false,
130+
useLayeredComposite: false,
131+
forceParallelStream: false,
132+
});
133+
}
134+
135+
const fallback =
136+
plan.routing.kind === "default"
137+
? { kind: plan.kind, workerCount: plan.workerCount, forceParallelStream: false }
138+
: plan.routing.fallback;
139+
const workerCount =
140+
failure.kind === "capture_failure" && failure.memoryExhaustion ? 1 : fallback.workerCount;
141+
return createCapturePlan({
142+
...plan,
143+
workerCount,
144+
forceScreenshot: true,
145+
forceParallelStream: fallback.forceParallelStream,
146+
useStreamingEncode: fallback.kind === "sdr_streaming",
147+
useLayeredComposite: false,
148+
routing: revertedRouting(plan.routing),
149+
});
150+
}

0 commit comments

Comments
 (0)