mirror of
https://github.com/heygen-com/hyperframes.git
synced 2026-09-02 20:18:35 +00:00
feat(producer): add services/distributed/plan.ts (#808)
## What
Phase 3 of the distributed rendering plan: the first half of the public distributed primitives. Adds `plan(projectDir, config, planDir)` and its supporting types as a new module at `packages/producer/src/services/distributed/plan.ts`. See `DISTRIBUTED-RENDERING-PLAN.md` §11 Phase 3.
## Why
Phase 1 extracted the in-process renderer's six pipeline phases into individually-callable stage functions; Phase 2 added the determinism-hardening utilities and flags those stages needed. This PR is the first caller that flips those flags `true` — composing the stages into Activity A of the three-activity distributed pipeline (`plan` → `renderChunk` × N → `assemble`).
Output is a self-contained `<planDir>/` with the documented §4.1 layout plus a content-addressed `planHash` (§4.2). Adapter authors (Temporal, AWS Lambda + Step Functions, etc.) consume the directory + hash; the OSS library never touches transport.
## How
`plan()` composes (in order):
1. `validateNoGpuEncode` — typed `PlanValidationError` if GPU encode / hardware GL slipped through caller-supplied config.
2. `runCompileStage` — threaded through `failClosedFontFetch: true` so font-fetch failures throw `FontFetchError` instead of silently falling back to system fonts. Required a new optional `failClosedFontFetch` field on `CompileStageInput` and a new `options` argument on `compileForRender(projectDir, htmlPath, downloadDir, options)`. Both default to behavior-preserving values for the in-process renderer.
3. `validateNoSystemFonts(compiled.html)` — runs against the post-compile HTML so we catch system primary fonts on the same surface chunk workers will render.
4. `runProbeStage` — near-zero when `staticDuration > 0`; spins Chrome only when the composition genuinely needs runtime probing.
5. `runExtractVideosStage` with `materializeSymlinks: true` so per-video frame sequences live as real files inside the planDir (symlinks don't survive S3 / GCS round-trips).
6. `runAudioStage` — produces `<planDir>/audio.aac` if the composition has audio.
7. Materialize the `<planDir>/{compiled,video-frames,audio.aac,meta}/...` layout from the staged work tree.
8. `freezePlan` — writes `meta/{composition,encoder,chunks}.json` + `plan.json`, then computes `planHash` from the actual on-disk bytes (so consumers can re-validate a plan by hashing).
`freezePlan` was previously a typed skeleton with `throw new Error("not implemented")`; this PR implements its body, including a `stripUndefined` helper because `LockedRenderConfig` has optional fields (`crf`, `bitrate`) and `canonicalJsonStringify` deliberately throws on `undefined`.
Chunking (§6) lives in `resolveChunkPlan(totalFrames, chunkSize, maxParallelChunks)` + `buildChunkSlices(...)` — exported from `plan.ts` so PR 3.2 (renderChunk) and adapter code can import them directly.
### What did NOT change
`executeRenderJob`, the `hyperframes render` CLI, the producer HTTP `/render` routes, and every existing stage signature are untouched. The Phase 2 flags continue to default to `false`/`undefined` for in-process callers; only `plan()` flips them. PSNR baselines for the regression harness should be unchanged.
## Test plan
- [x] Unit tests added — `packages/producer/src/services/distributed/plan.test.ts`. 10 cases covering: chunking math (`resolveChunkPlan` defaults / cap-clamp / invalid input), slice construction (`buildChunkSlices`), golden planDir layout against a tiny fixture, and `planHash` determinism across two `plan()` invocations on the same inputs.
- [x] `bun test packages/producer/src/services/distributed/` — 10 pass.
- [x] `bun test packages/producer/src/` — 312 pass, 1 fail. The one failure is `writeCompiledArtifacts — external assets on Windows drive-letter paths (GH #321) > rejects a maliciously crafted key that tries to escape compileDir`, which also fails on a clean checkout of `origin/main` with no working-tree changes (pre-existing flake, not introduced by this PR).
- [x] `bun run --filter @hyperframes/producer typecheck` — clean.
- [x] `bun run --filter @hyperframes/producer build` — clean.
- [x] `bunx oxlint` + `bunx oxfmt --check` — clean on changed files.
- [ ] Producer Docker regression harness — pending CI run. `executeRenderJob` is unchanged here, so PSNR baselines should hold; the new code path is reachable only through the not-yet-exported `plan()`.
This is PR 1 of a 6-PR Phase 3 stack:
- **3.1 (this PR)** — `services/distributed/plan.ts`
- 3.2 — `services/distributed/renderChunk.ts`
- 3.3 — `services/distributed/assemble.ts`
- 3.4 — `planDir` size cap (`PLAN_TOO_LARGE`)
- 3.5 — distributed format banlist (webm + HDR mp4)
- 3.6 — public exports + `@hyperframes/producer/distributed` subpath
🤖 Generated with [Claude Code](https://claude.com/claude-code)
This commit is contained in:
James Russo
@@ -0,0 +1,211 @@
|
||||
/**
|
||||
* Unit tests for `services/distributed/plan.ts`.
|
||||
*
|
||||
* Covers:
|
||||
* - Golden planDir layout produced from a tiny fixture (no browser probe
|
||||
* required — the fixture declares `data-duration` so the probe stage
|
||||
* short-circuits).
|
||||
* - planHash determinism across two `plan()` calls on the same inputs.
|
||||
* - Chunking math (`resolveChunkPlan`, `buildChunkSlices`).
|
||||
*
|
||||
* The "no browser probe" path is deliberate: spinning Chrome inside `bun test`
|
||||
* is expensive and flaky. The chunking helpers + planDir layout are tested
|
||||
* with synchronous compile-only fixtures; the BeginFrame / probe path lives
|
||||
* inside the regression harness (`bun run --cwd packages/producer docker:test`).
|
||||
*/
|
||||
|
||||
import { afterAll, beforeAll, describe, expect, it } from "bun:test";
|
||||
import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import {
|
||||
buildChunkSlices,
|
||||
DEFAULT_CHUNK_SIZE,
|
||||
DEFAULT_MAX_PARALLEL_CHUNKS,
|
||||
plan,
|
||||
resolveChunkPlan,
|
||||
} from "./plan.js";
|
||||
|
||||
// Composition the tests render. `data-duration="1"` keeps the probe stage's
|
||||
// `needsBrowser` gate `false` so plan() completes without launching Chrome.
|
||||
const FIXTURE_HTML = `<!doctype html>
|
||||
<html>
|
||||
<head><meta charset="utf-8"><title>plan-test fixture</title></head>
|
||||
<body>
|
||||
<div data-composition-id="root" data-width="320" data-height="240" data-duration="1">
|
||||
<p>plan-test fixture</p>
|
||||
</div>
|
||||
</body>
|
||||
</html>`;
|
||||
|
||||
let projectDir: string;
|
||||
let runRoot: string;
|
||||
|
||||
beforeAll(() => {
|
||||
runRoot = mkdtempSync(join(tmpdir(), "hf-plan-test-"));
|
||||
projectDir = join(runRoot, "project");
|
||||
mkdirSync(projectDir, { recursive: true });
|
||||
writeFileSync(join(projectDir, "index.html"), FIXTURE_HTML, "utf-8");
|
||||
});
|
||||
|
||||
afterAll(() => {
|
||||
rmSync(runRoot, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
describe("resolveChunkPlan", () => {
|
||||
it("returns 1 chunk when totalFrames fits in configChunkSize", () => {
|
||||
const result = resolveChunkPlan(60, 240, 16);
|
||||
expect(result.chunkCount).toBe(1);
|
||||
expect(result.effectiveChunkSize).toBeGreaterThanOrEqual(60);
|
||||
});
|
||||
|
||||
it("caps chunkCount at maxParallelChunks for very long renders", () => {
|
||||
// 54000 frames / 240 = 225 naive chunks → must cap at 16.
|
||||
const result = resolveChunkPlan(54000, 240, 16);
|
||||
expect(result.chunkCount).toBe(16);
|
||||
// 54000 / 16 = 3375 → chunkSize at least that big so the union covers
|
||||
// all frames in 16 slices.
|
||||
expect(result.effectiveChunkSize).toBeGreaterThanOrEqual(Math.ceil(54000 / 16));
|
||||
});
|
||||
|
||||
it("naive count drives chunkCount when below cap", () => {
|
||||
// 600 frames / 240 = 3 naive chunks; well below the 16 cap.
|
||||
const result = resolveChunkPlan(600, 240, 16);
|
||||
expect(result.chunkCount).toBe(3);
|
||||
expect(result.effectiveChunkSize).toBe(240);
|
||||
});
|
||||
|
||||
it("rejects non-positive totalFrames", () => {
|
||||
expect(() => resolveChunkPlan(0, 240, 16)).toThrow();
|
||||
expect(() => resolveChunkPlan(-1, 240, 16)).toThrow();
|
||||
expect(() => resolveChunkPlan(Number.NaN, 240, 16)).toThrow();
|
||||
});
|
||||
|
||||
it("rejects non-positive configChunkSize / maxParallelChunks", () => {
|
||||
expect(() => resolveChunkPlan(60, 0, 16)).toThrow();
|
||||
expect(() => resolveChunkPlan(60, 240, 0)).toThrow();
|
||||
});
|
||||
|
||||
it("rejects non-integer inputs (would produce fractional endFrames)", () => {
|
||||
expect(() => resolveChunkPlan(10.5, 240, 16)).toThrow(/positive integer/);
|
||||
expect(() => resolveChunkPlan(60, 240.5, 16)).toThrow(/positive integer/);
|
||||
expect(() => resolveChunkPlan(60, 240, 16.5)).toThrow(/positive integer/);
|
||||
expect(() => resolveChunkPlan(60, 240, Number.POSITIVE_INFINITY)).toThrow(/positive integer/);
|
||||
});
|
||||
});
|
||||
|
||||
describe("buildChunkSlices", () => {
|
||||
it("produces consecutive non-overlapping ranges covering all frames", () => {
|
||||
const slices = buildChunkSlices(700, 3, 240);
|
||||
expect(slices).toHaveLength(3);
|
||||
expect(slices[0]).toEqual({ index: 0, startFrame: 0, endFrame: 240 });
|
||||
expect(slices[1]).toEqual({ index: 1, startFrame: 240, endFrame: 480 });
|
||||
// Last chunk absorbs the remainder so endFrame === totalFrames exactly.
|
||||
expect(slices[2]).toEqual({ index: 2, startFrame: 480, endFrame: 700 });
|
||||
});
|
||||
|
||||
it("handles a single-chunk render", () => {
|
||||
const slices = buildChunkSlices(50, 1, 240);
|
||||
expect(slices).toHaveLength(1);
|
||||
expect(slices[0]).toEqual({ index: 0, startFrame: 0, endFrame: 50 });
|
||||
});
|
||||
});
|
||||
|
||||
describe("plan() defaults", () => {
|
||||
it("exports the documented chunking defaults", () => {
|
||||
expect(DEFAULT_CHUNK_SIZE).toBe(240);
|
||||
expect(DEFAULT_MAX_PARALLEL_CHUNKS).toBe(16);
|
||||
});
|
||||
});
|
||||
|
||||
describe("plan() — golden planDir + planHash determinism", () => {
|
||||
// Each `plan()` call is reasonably expensive (compile pass parses + inlines
|
||||
// the HTML), so we run it once for the layout assertions and once more for
|
||||
// the determinism assertion. The 30s timeout absorbs cold-start font /
|
||||
// runtime resolution variance on the CI host.
|
||||
const TIMEOUT_MS = 30_000;
|
||||
|
||||
it(
|
||||
"produces the documented planDir layout",
|
||||
async () => {
|
||||
const planDir = join(runRoot, "plan-layout");
|
||||
mkdirSync(planDir, { recursive: true });
|
||||
const result = await plan(
|
||||
projectDir,
|
||||
{ fps: 30, width: 320, height: 240, format: "mp4" },
|
||||
planDir,
|
||||
);
|
||||
|
||||
// planDir directory layout
|
||||
expect(existsSync(join(planDir, "plan.json"))).toBe(true);
|
||||
expect(existsSync(join(planDir, "compiled", "index.html"))).toBe(true);
|
||||
expect(existsSync(join(planDir, "video-frames"))).toBe(true);
|
||||
// No audio in the fixture — audio.aac must NOT exist.
|
||||
expect(existsSync(join(planDir, "audio.aac"))).toBe(false);
|
||||
expect(existsSync(join(planDir, "meta", "composition.json"))).toBe(true);
|
||||
expect(existsSync(join(planDir, "meta", "encoder.json"))).toBe(true);
|
||||
expect(existsSync(join(planDir, "meta", "chunks.json"))).toBe(true);
|
||||
// The temporary work tree must be cleaned up.
|
||||
expect(existsSync(join(planDir, ".plan-work"))).toBe(false);
|
||||
|
||||
// ── PlanResult contract ─────────────────────────────────────────────
|
||||
expect(result.planDir).toBe(planDir);
|
||||
expect(result.planHash).toMatch(/^[0-9a-f]{64}$/);
|
||||
expect(result.chunkCount).toBeGreaterThanOrEqual(1);
|
||||
expect(result.totalFrames).toBe(30); // 1s @ 30fps
|
||||
expect(result.width).toBe(320);
|
||||
expect(result.height).toBe(240);
|
||||
expect(result.format).toBe("mp4");
|
||||
expect(result.ffmpegVersion).toMatch(/ffmpeg/i);
|
||||
expect(result.producerVersion).toMatch(/^\d+\.\d+\.\d+/);
|
||||
|
||||
// ── chunks.json shape ───────────────────────────────────────────────
|
||||
const chunks = JSON.parse(
|
||||
readFileSync(join(planDir, "meta", "chunks.json"), "utf-8"),
|
||||
) as Array<{ index: number; startFrame: number; endFrame: number }>;
|
||||
expect(chunks).toHaveLength(result.chunkCount);
|
||||
// Slices must cover [0, totalFrames) with no gaps.
|
||||
let cursor = 0;
|
||||
for (const chunk of chunks) {
|
||||
expect(chunk.startFrame).toBe(cursor);
|
||||
cursor = chunk.endFrame;
|
||||
}
|
||||
expect(cursor).toBe(result.totalFrames);
|
||||
|
||||
// ── plan.json shape ─────────────────────────────────────────────────
|
||||
const planJson = JSON.parse(readFileSync(join(planDir, "plan.json"), "utf-8")) as Record<
|
||||
string,
|
||||
unknown
|
||||
>;
|
||||
expect(planJson.planHash).toBe(result.planHash);
|
||||
expect(planJson.hasAudio).toBe(false);
|
||||
expect(planJson.totalFrames).toBe(result.totalFrames);
|
||||
},
|
||||
TIMEOUT_MS,
|
||||
);
|
||||
|
||||
it(
|
||||
"produces a byte-identical planHash on a second invocation",
|
||||
async () => {
|
||||
const planDirA = join(runRoot, "plan-determinism-a");
|
||||
const planDirB = join(runRoot, "plan-determinism-b");
|
||||
mkdirSync(planDirA, { recursive: true });
|
||||
mkdirSync(planDirB, { recursive: true });
|
||||
|
||||
const config = { fps: 30 as const, width: 320, height: 240, format: "mp4" as const };
|
||||
const a = await plan(projectDir, config, planDirA);
|
||||
const b = await plan(projectDir, config, planDirB);
|
||||
|
||||
expect(a.planHash).toBe(b.planHash);
|
||||
expect(a.chunkCount).toBe(b.chunkCount);
|
||||
expect(a.totalFrames).toBe(b.totalFrames);
|
||||
|
||||
// Encoder JSON must be byte-identical — its bytes feed planHash, so any
|
||||
// drift here would silently change the hash framing.
|
||||
const encoderA = readFileSync(join(planDirA, "meta", "encoder.json"));
|
||||
const encoderB = readFileSync(join(planDirB, "meta", "encoder.json"));
|
||||
expect(encoderA.equals(encoderB)).toBe(true);
|
||||
},
|
||||
TIMEOUT_MS,
|
||||
);
|
||||
});
|
||||
@@ -0,0 +1,614 @@
|
||||
/**
|
||||
* Activity A of the distributed render pipeline.
|
||||
*
|
||||
* `plan(projectDir, config, planDir)` composes the existing render stages
|
||||
* (compile → probe → extract videos → audio → freeze) into a self-contained
|
||||
* `<planDir>/` directory tree that downstream chunk workers consume:
|
||||
*
|
||||
* <planDir>/
|
||||
* ├── plan.json
|
||||
* ├── compiled/ # compileForRender output (self-contained)
|
||||
* ├── video-frames/ # per-video JPEG sequences (dereferenced)
|
||||
* ├── audio.aac # only when composition has audio
|
||||
* └── meta/
|
||||
* ├── composition.json
|
||||
* ├── encoder.json # LockedRenderConfig
|
||||
* └── chunks.json
|
||||
*
|
||||
* Pure function over local paths. No networking. Two invocations with the
|
||||
* same inputs produce the same `planHash` — adapters use that contract to
|
||||
* short-circuit `plan()` on workflow replay.
|
||||
*
|
||||
* Banned configurations (GPU encode, hardware browser GL, system primary
|
||||
* fonts) are rejected at plan time via `planValidation.ts` so chunk workers
|
||||
* never have to handle them.
|
||||
*/
|
||||
|
||||
import { execFile as execFileCallback } from "node:child_process";
|
||||
import { existsSync, mkdirSync, readFileSync, renameSync, rmSync } from "node:fs";
|
||||
import { dirname, join } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { promisify } from "node:util";
|
||||
import { type CanvasResolution, type Fps } from "@hyperframes/core";
|
||||
import { type EngineConfig, resolveConfig } from "@hyperframes/engine";
|
||||
import { defaultLogger, type ProducerLogger } from "../../logger.js";
|
||||
import { type RenderConfig, type RenderJob, createRenderJob } from "../renderOrchestrator.js";
|
||||
import { runAudioStage } from "../render/stages/audioStage.js";
|
||||
import { runCompileStage } from "../render/stages/compileStage.js";
|
||||
import { runExtractVideosStage } from "../render/stages/extractVideosStage.js";
|
||||
import { runProbeStage } from "../render/stages/probeStage.js";
|
||||
import {
|
||||
type ChunkSliceJson,
|
||||
type CompositionMetadataJson,
|
||||
freezePlan,
|
||||
type LockedRenderConfig,
|
||||
} from "../render/stages/freezePlan.js";
|
||||
import {
|
||||
canonicalJsonStringify,
|
||||
type PlanDimensions,
|
||||
sha256Hex,
|
||||
} from "../render/stages/planHash.js";
|
||||
import { validateNoGpuEncode, validateNoSystemFonts } from "../render/planValidation.js";
|
||||
import { snapshotRuntimeEnv } from "../render/runtimeEnvSnapshot.js";
|
||||
|
||||
const execFile = promisify(execFileCallback);
|
||||
|
||||
/**
|
||||
* Caller-supplied configuration for a distributed render. `fps`, `width`,
|
||||
* `height`, and `format` are required; everything else carries a default
|
||||
* sensible for AWS Lambda fan-out.
|
||||
*/
|
||||
export interface DistributedRenderConfig {
|
||||
/** Integer frame rate. Distributed renders only accept integer fps; the in-process renderer's `Fps` rational handles NTSC. */
|
||||
fps: 24 | 30 | 60;
|
||||
width: number;
|
||||
height: number;
|
||||
/**
|
||||
* Output container format. webm and HDR mp4 are not supported in
|
||||
* distributed mode — `plan()` refuses them up front with a typed
|
||||
* `FormatNotSupportedInDistributedError`. The in-process renderer
|
||||
* supports both.
|
||||
*/
|
||||
format: "mp4" | "mov" | "png-sequence";
|
||||
quality?: "draft" | "standard" | "high";
|
||||
/** Constant-rate-factor override; mutually exclusive with `bitrate`. */
|
||||
crf?: number;
|
||||
/** Target video bitrate (e.g. `"10M"`); mutually exclusive with `crf`. */
|
||||
bitrate?: string;
|
||||
/** Output resolution preset; engages Chrome `deviceScaleFactor` supersampling. */
|
||||
outputResolution?: CanvasResolution;
|
||||
|
||||
/** Default `240` frames (~8s @ 30fps; fits Lambda's 15-min cap). */
|
||||
chunkSize?: number;
|
||||
/** Default `16`. Caps long renders to fewer-but-longer chunks for operational fairness. */
|
||||
maxParallelChunks?: number;
|
||||
/** Runtime hint; consumed by future per-runtime budget checks. The current implementation records the value but does not enforce. */
|
||||
runtimeCap?: "lambda" | "temporal" | "cloud-run-job" | "k8s-job" | "none";
|
||||
|
||||
/**
|
||||
* Reject compositions whose primary font-family resolves to a host-OS /
|
||||
* generic family. Default `true` for distributed renders — overriding to
|
||||
* `false` is unsupported and exists only as an escape hatch for tests.
|
||||
*/
|
||||
rejectOnSystemFonts?: boolean;
|
||||
/**
|
||||
* Threaded into the `injectDeterministicFontFaces` font loader. Default
|
||||
* `true` — distributed renders must not silently fall back to system fonts.
|
||||
*/
|
||||
failClosedFontFetch?: boolean;
|
||||
|
||||
/** HDR is not supported in distributed mode; `force-hdr` trips a `FormatNotSupportedInDistributedError`. Defaults to `force-sdr`. */
|
||||
hdrMode?: "auto" | "force-sdr";
|
||||
|
||||
logger?: ProducerLogger;
|
||||
/** Optional engine config override (env vars are not read when provided). */
|
||||
producerConfig?: EngineConfig;
|
||||
/** Entry HTML file relative to `projectDir`. Defaults to `"index.html"`. */
|
||||
entryFile?: string;
|
||||
/** Caller-supplied AbortSignal. Threaded through compile / probe / extract / audio stages. */
|
||||
abortSignal?: AbortSignal;
|
||||
}
|
||||
|
||||
/**
|
||||
* Result of {@link plan}. The `planHash` is the content-addressed identifier
|
||||
* that adapters key replay short-circuits off of.
|
||||
*/
|
||||
export interface PlanResult {
|
||||
planDir: string;
|
||||
planHash: string;
|
||||
chunkCount: number;
|
||||
totalFrames: number;
|
||||
fps: 24 | 30 | 60;
|
||||
width: number;
|
||||
height: number;
|
||||
format: "mp4" | "mov" | "png-sequence";
|
||||
ffmpegVersion: string;
|
||||
producerVersion: string;
|
||||
}
|
||||
|
||||
/** Default chunk size in frames (~8s @ 30fps; fits Lambda's 15-min cap). */
|
||||
export const DEFAULT_CHUNK_SIZE = 240;
|
||||
/** Default cap on parallel chunks for operational fairness across renders. */
|
||||
export const DEFAULT_MAX_PARALLEL_CHUNKS = 16;
|
||||
|
||||
/**
|
||||
* Compute `(chunkCount, effectiveChunkSize)` from total frames and the
|
||||
* caller's chunking knobs:
|
||||
*
|
||||
* chunkCount = min(maxParallelChunks, ceil(totalFrames / chunkSize))
|
||||
* effectiveChunkSize = max(configChunkSize, ceil(totalFrames / maxParallelChunks))
|
||||
*
|
||||
* Long renders auto-rescale to fewer-but-longer chunks rather than
|
||||
* fragmenting infinitely. Returned `chunkCount >= 1` (`totalFrames === 0`
|
||||
* is rejected upstream); `effectiveChunkSize >= configChunkSize`.
|
||||
*/
|
||||
export function resolveChunkPlan(
|
||||
totalFrames: number,
|
||||
configChunkSize: number,
|
||||
maxParallelChunks: number,
|
||||
): { chunkCount: number; effectiveChunkSize: number } {
|
||||
// Integer-only inputs: a fractional `totalFrames` (e.g. 10.5) would
|
||||
// otherwise produce a last chunk with non-integer `endFrame`, and the
|
||||
// chunk worker's `for (i = startFrame; i < endFrame; i++)` loop would
|
||||
// silently truncate.
|
||||
assertPositiveInteger("totalFrames", totalFrames);
|
||||
assertPositiveInteger("configChunkSize", configChunkSize);
|
||||
assertPositiveInteger("maxParallelChunks", maxParallelChunks);
|
||||
const naiveCount = Math.ceil(totalFrames / configChunkSize);
|
||||
const chunkCount = Math.min(maxParallelChunks, Math.max(1, naiveCount));
|
||||
const effectiveChunkSize = Math.max(configChunkSize, Math.ceil(totalFrames / chunkCount));
|
||||
return { chunkCount, effectiveChunkSize };
|
||||
}
|
||||
|
||||
function assertPositiveInteger(name: string, value: number): void {
|
||||
if (!Number.isInteger(value) || value <= 0) {
|
||||
throw new Error(
|
||||
`[plan] resolveChunkPlan: ${name} must be a positive integer (received ${String(value)})`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Slice `totalFrames` into `chunkCount` consecutive ranges. Each chunk gets
|
||||
* `effectiveChunkSize` frames except the last, which absorbs the remainder
|
||||
* so the union is exactly `[0, totalFrames)`. `endFrame` is the EXCLUSIVE
|
||||
* upper bound — chunk workers iterate `i in [startFrame, endFrame)`.
|
||||
*/
|
||||
export function buildChunkSlices(
|
||||
totalFrames: number,
|
||||
chunkCount: number,
|
||||
effectiveChunkSize: number,
|
||||
): ChunkSliceJson[] {
|
||||
const slices: ChunkSliceJson[] = [];
|
||||
for (let i = 0; i < chunkCount; i++) {
|
||||
const startFrame = i * effectiveChunkSize;
|
||||
const endFrame =
|
||||
i === chunkCount - 1 ? totalFrames : Math.min(totalFrames, startFrame + effectiveChunkSize);
|
||||
slices.push({ index: i, startFrame, endFrame });
|
||||
}
|
||||
return slices;
|
||||
}
|
||||
|
||||
/**
|
||||
* Map a `DistributedRenderConfig` onto the in-process `RenderConfig` shape
|
||||
* the stage functions consume. Distributed plan() is the first caller of
|
||||
* the staged renderer that operates without a full RenderJob — we synthesize
|
||||
* one from the distributed config so the existing stage interfaces don't
|
||||
* need a parallel "distributed mode" overload.
|
||||
*/
|
||||
function buildSyntheticRenderJob(config: DistributedRenderConfig): RenderJob {
|
||||
const renderConfig: RenderConfig = {
|
||||
fps: { num: config.fps, den: 1 } satisfies Fps,
|
||||
quality: config.quality ?? "standard",
|
||||
format: config.format,
|
||||
crf: config.crf,
|
||||
videoBitrate: config.bitrate,
|
||||
outputResolution: config.outputResolution,
|
||||
// Distributed mode hard-pins to software GPU. The plan-time validator
|
||||
// (see validateNoGpuEncode) refuses to fan out otherwise.
|
||||
useGpu: false,
|
||||
debug: false,
|
||||
entryFile: config.entryFile ?? "index.html",
|
||||
logger: config.logger ?? defaultLogger,
|
||||
// HDR is banned in distributed mode. force-sdr keeps the
|
||||
// extract / encoder paths off the HDR branches entirely.
|
||||
hdrMode: config.hdrMode ?? "force-sdr",
|
||||
producerConfig: config.producerConfig,
|
||||
};
|
||||
return createRenderJob(renderConfig);
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve the producer package version by walking up from the calling module
|
||||
* until a `package.json` whose `name === "@hyperframes/producer"` is found.
|
||||
* Works for both the bundled `dist/index.js` (1 level up) and the unbundled
|
||||
* source tree (`src/services/distributed/plan.ts` → 4 levels up).
|
||||
*/
|
||||
function readProducerVersion(): string {
|
||||
const startDir = dirname(fileURLToPath(import.meta.url));
|
||||
let current = startDir;
|
||||
for (let i = 0; i < 10; i++) {
|
||||
const candidate = join(current, "package.json");
|
||||
if (existsSync(candidate)) {
|
||||
try {
|
||||
const pkg = JSON.parse(readFileSync(candidate, "utf-8")) as {
|
||||
name?: string;
|
||||
version?: string;
|
||||
};
|
||||
if (pkg.name === "@hyperframes/producer" && typeof pkg.version === "string") {
|
||||
return pkg.version;
|
||||
}
|
||||
} catch {
|
||||
// Fall through to the next ancestor.
|
||||
}
|
||||
}
|
||||
const parent = dirname(current);
|
||||
if (parent === current) break;
|
||||
current = parent;
|
||||
}
|
||||
return "0.0.0-unknown";
|
||||
}
|
||||
|
||||
/**
|
||||
* Spawn `ffmpeg -version` and return the first line (e.g. `"ffmpeg version 6.1.1"`).
|
||||
* The string is opaque — `planHash` mixes it in verbatim, so any drift across
|
||||
* worker hosts trips a `FFMPEG_VERSION_MISMATCH` rather than producing pixels
|
||||
* that subtly disagree with the plan's baked-in encoder args.
|
||||
*/
|
||||
async function readFfmpegVersion(): Promise<string> {
|
||||
const { stdout } = await execFile("ffmpeg", ["-version"], { maxBuffer: 1024 * 1024 });
|
||||
const firstLine = stdout.split(/\r?\n/)[0]?.trim() ?? "";
|
||||
if (!firstLine) {
|
||||
throw new Error("[plan] ffmpeg -version returned empty output");
|
||||
}
|
||||
return firstLine;
|
||||
}
|
||||
|
||||
/**
|
||||
* Hash the deterministic-font bundle that ships inside `@hyperframes/producer`.
|
||||
* The compiled HTML already inlines per-family `@font-face` data URIs, so the
|
||||
* snapshot SHA exists primarily to detect cross-version font-bundle drift on
|
||||
* chunk workers. Mixed into `planHash`.
|
||||
*
|
||||
* Pulled lazily because the generated module is large and only the
|
||||
* distributed pipeline needs it.
|
||||
*/
|
||||
async function readFontSnapshotSha(): Promise<string> {
|
||||
const module = (await import("../fontData.generated.js")) as {
|
||||
EMBEDDED_FONT_DATA?: unknown;
|
||||
};
|
||||
const data = module.EMBEDDED_FONT_DATA;
|
||||
if (!data || typeof data !== "object") {
|
||||
throw new Error(
|
||||
"[plan] EMBEDDED_FONT_DATA missing from fontData.generated.js — was `bun run build:fonts` run?",
|
||||
);
|
||||
}
|
||||
// Hash a canonical key fingerprint, not the raw font bytes — the bytes are
|
||||
// already mixed in through `compositionHtml` (the @font-face data URIs the
|
||||
// compiler injects). What we really want to detect here is "the bundle on
|
||||
// worker B is a different version of the producer than on controller A",
|
||||
// which is fully captured by the sorted family names + per-family byte
|
||||
// lengths.
|
||||
const dataObj = data as Record<string, unknown>;
|
||||
const fingerprint: Record<string, number> = {};
|
||||
for (const key of Object.keys(dataObj).sort()) {
|
||||
const value = dataObj[key];
|
||||
fingerprint[key] =
|
||||
typeof value === "string" ? value.length : JSON.stringify(value ?? null).length;
|
||||
}
|
||||
return sha256Hex(canonicalJsonStringify(fingerprint));
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the `LockedRenderConfig` frozen into `meta/encoder.json`.
|
||||
* Captures everything chunk workers need to reproduce the controller's
|
||||
* encode decisions byte-for-byte. Validated by the chunk worker on boot —
|
||||
* the same input here must round-trip to an identical config.
|
||||
*/
|
||||
function buildLockedRenderConfig(input: {
|
||||
config: DistributedRenderConfig;
|
||||
forceScreenshot: boolean;
|
||||
deviceScaleFactor: number;
|
||||
ffmpegVersion: string;
|
||||
effectiveChunkSize: number;
|
||||
chunkCount: number;
|
||||
runtimeEnv: Record<string, string>;
|
||||
}): LockedRenderConfig {
|
||||
const { config, forceScreenshot, deviceScaleFactor, ffmpegVersion } = input;
|
||||
const { encoder, pixelFormat, preset } = FORMAT_ENCODER_TABLE[config.format];
|
||||
return {
|
||||
captureMode: forceScreenshot ? "screenshot" : "beginframe",
|
||||
forceScreenshot,
|
||||
deviceScaleFactor,
|
||||
useLayeredHdrComposite: false,
|
||||
browserGpuMode: "software",
|
||||
// Match `LOCKED_WARMUP_TICKS` in `frameCapture.ts` — kept as a literal so
|
||||
// a worker that ships a different value will trip `PLAN_HASH_MISMATCH`
|
||||
// (the locked config flows into planHash via the canonical JSON).
|
||||
warmupTicks: 60,
|
||||
encoder,
|
||||
quality: config.quality ?? "standard",
|
||||
ffmpegVersion,
|
||||
preset,
|
||||
crf: config.crf,
|
||||
bitrate: config.bitrate,
|
||||
// GOP === chunkSize so every chunk's first frame is an IDR keyframe and
|
||||
// ffmpeg concat-copy round-trips losslessly.
|
||||
gopSize: input.effectiveChunkSize,
|
||||
closedGop: true,
|
||||
forceKeyframes: "n=0",
|
||||
pixelFormat,
|
||||
chunkSize: input.effectiveChunkSize,
|
||||
chunkCount: input.chunkCount,
|
||||
runtimeEnv: input.runtimeEnv,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Per-format encoder + pixel-format + preset triple. Distributed mode is
|
||||
* SDR-only: H.264 8-bit for mp4, ProRes 4444 for mov, raw RGBA for
|
||||
* png-sequence.
|
||||
*/
|
||||
const FORMAT_ENCODER_TABLE: Record<
|
||||
DistributedRenderConfig["format"],
|
||||
{ encoder: LockedRenderConfig["encoder"]; pixelFormat: string; preset: string }
|
||||
> = {
|
||||
mp4: { encoder: "libx264-software", pixelFormat: "yuv420p", preset: "medium" },
|
||||
mov: { encoder: "prores-software", pixelFormat: "yuva444p10le", preset: "4444" },
|
||||
"png-sequence": { encoder: "png-sequence", pixelFormat: "rgba", preset: "lossless" },
|
||||
};
|
||||
|
||||
/**
|
||||
* Activity A of the distributed render pipeline. Produces a self-contained
|
||||
* `<planDir>/` from a project + config. See module docstring for the
|
||||
* directory layout.
|
||||
*/
|
||||
export async function plan(
|
||||
projectDir: string,
|
||||
config: DistributedRenderConfig,
|
||||
planDir: string,
|
||||
): Promise<PlanResult> {
|
||||
// ── Plan-time validation ──
|
||||
// Rejections here surface as typed `PlanValidationError`s with non-retryable
|
||||
// codes so workflow adapters don't waste retry budget on banned configs.
|
||||
validateNoGpuEncode({
|
||||
useGpu: false,
|
||||
browserGpuMode: "software",
|
||||
});
|
||||
|
||||
if (!existsSync(planDir)) mkdirSync(planDir, { recursive: true });
|
||||
|
||||
const log = config.logger ?? defaultLogger;
|
||||
const abortSignal = config.abortSignal;
|
||||
const assertNotAborted = (): void => {
|
||||
if (abortSignal?.aborted) {
|
||||
throw new Error("[plan] render_cancelled");
|
||||
}
|
||||
};
|
||||
const cfg: EngineConfig = {
|
||||
...(config.producerConfig ?? resolveConfig()),
|
||||
browserGpuMode: "software",
|
||||
forceScreenshot: false,
|
||||
};
|
||||
|
||||
const job = buildSyntheticRenderJob(config);
|
||||
const entryFile = config.entryFile ?? "index.html";
|
||||
const htmlPath = join(projectDir, entryFile);
|
||||
if (!existsSync(htmlPath)) {
|
||||
throw new Error(`[plan] entry file not found: ${htmlPath}`);
|
||||
}
|
||||
|
||||
const workDir = join(planDir, ".plan-work");
|
||||
if (!existsSync(workDir)) mkdirSync(workDir, { recursive: true });
|
||||
const compiledDir = join(workDir, "compiled");
|
||||
|
||||
// The compiled directory lives at `<planDir>/compiled/` in the final
|
||||
// layout. The stages write under `<planDir>/.plan-work/compiled/`; we
|
||||
// move the contents over once the staged work completes.
|
||||
const finalCompiledDir = join(planDir, "compiled");
|
||||
|
||||
// mov + png-sequence carry alpha — flip force-screenshot so compileStage
|
||||
// takes the alpha-aware capture path (BeginFrame doesn't preserve alpha
|
||||
// on Linux headless-shell).
|
||||
const needsAlpha = config.format === "png-sequence" || config.format === "mov";
|
||||
|
||||
// ── Compile ──
|
||||
const compileResult = await runCompileStage({
|
||||
projectDir,
|
||||
workDir,
|
||||
htmlPath,
|
||||
entryFile,
|
||||
job,
|
||||
cfg,
|
||||
needsAlpha,
|
||||
log,
|
||||
assertNotAborted,
|
||||
// Distributed renders fail closed on font-fetch errors so the planDir
|
||||
// is content-addressed against deterministic fonts only.
|
||||
failClosedFontFetch: config.failClosedFontFetch !== false,
|
||||
});
|
||||
let compiled = compileResult.compiled;
|
||||
const composition = compileResult.composition;
|
||||
const { deviceScaleFactor, forceScreenshot } = compileResult;
|
||||
// composition.{width,height} are the authored page dimensions. The
|
||||
// post-supersample output dims are `compileResult.outputWidth/outputHeight`
|
||||
// — chunks render at output dims, but planHash + composition.json record
|
||||
// the page dims so cross-machine consistency keys off the composition's
|
||||
// own intent rather than a knob the planner could tweak.
|
||||
const { width, height } = composition;
|
||||
|
||||
// ── Reject system primary fonts ──
|
||||
// Runs against the post-compile HTML (which has @font-face declarations
|
||||
// injected) so we evaluate the same surface the chunk worker would render.
|
||||
if (config.rejectOnSystemFonts !== false) {
|
||||
validateNoSystemFonts(compiled.html);
|
||||
}
|
||||
|
||||
// ── Probe ──
|
||||
// Browser probe runs only when needed. For statically-resolvable durations
|
||||
// this is a near-zero pass.
|
||||
const probeResult = await runProbeStage({
|
||||
projectDir,
|
||||
workDir,
|
||||
job,
|
||||
cfg,
|
||||
log,
|
||||
assertNotAborted,
|
||||
compiled,
|
||||
composition,
|
||||
width,
|
||||
height,
|
||||
needsAlpha,
|
||||
deviceScaleFactor,
|
||||
});
|
||||
compiled = probeResult.compiled;
|
||||
job.duration = probeResult.duration;
|
||||
job.totalFrames = probeResult.totalFrames;
|
||||
const totalFrames = probeResult.totalFrames;
|
||||
if (probeResult.fileServer) probeResult.fileServer.close();
|
||||
if (probeResult.probeSession) {
|
||||
// Close inside a try/catch — leaking a Chrome process here would mask
|
||||
// the original plan() result on cancellation paths.
|
||||
try {
|
||||
const { closeCaptureSession } = await import("@hyperframes/engine");
|
||||
await closeCaptureSession(probeResult.probeSession);
|
||||
} catch (err) {
|
||||
log.warn("[plan] probe session close failed", {
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// ── Extract videos ──
|
||||
// `materializeSymlinks: true` recursively copies frames so the planDir is
|
||||
// self-contained (symlinks don't survive S3/GCS round-trips).
|
||||
const extractResult = await runExtractVideosStage({
|
||||
projectDir,
|
||||
compiledDir,
|
||||
job,
|
||||
cfg,
|
||||
composition,
|
||||
abortSignal,
|
||||
assertNotAborted,
|
||||
materializeSymlinks: true,
|
||||
});
|
||||
if (extractResult.frameLookup) extractResult.frameLookup.cleanup();
|
||||
|
||||
// ── Audio ──
|
||||
const audioResult = await runAudioStage({
|
||||
projectDir,
|
||||
workDir,
|
||||
compiledDir,
|
||||
duration: job.duration,
|
||||
audios: composition.audios,
|
||||
abortSignal,
|
||||
assertNotAborted,
|
||||
});
|
||||
|
||||
// Promote staged artifacts from the temp work tree into the final planDir
|
||||
// shape. `workDir` is `<planDir>/.plan-work/` — always the same filesystem
|
||||
// as `planDir`, so `renameSync` succeeds without copying. Video frames
|
||||
// alone can be hundreds of MB; copying once instead of twice (the prior
|
||||
// approach left a duplicate under `compiled/__hyperframes_video_frames/`)
|
||||
// halves peak disk usage during `plan()`.
|
||||
const stagedVideoFrames = join(compiledDir, "__hyperframes_video_frames");
|
||||
const videoFramesDst = join(planDir, "video-frames");
|
||||
if (existsSync(videoFramesDst)) rmSync(videoFramesDst, { recursive: true, force: true });
|
||||
if (existsSync(stagedVideoFrames)) {
|
||||
renameSync(stagedVideoFrames, videoFramesDst);
|
||||
} else {
|
||||
mkdirSync(videoFramesDst, { recursive: true });
|
||||
}
|
||||
|
||||
if (existsSync(finalCompiledDir)) rmSync(finalCompiledDir, { recursive: true, force: true });
|
||||
renameSync(compiledDir, finalCompiledDir);
|
||||
|
||||
const planAudioPath = join(planDir, "audio.aac");
|
||||
if (audioResult.hasAudio && existsSync(audioResult.audioOutputPath)) {
|
||||
renameSync(audioResult.audioOutputPath, planAudioPath);
|
||||
}
|
||||
|
||||
// ── Chunking decisions + locked config ──
|
||||
const configChunkSize = config.chunkSize ?? DEFAULT_CHUNK_SIZE;
|
||||
const maxParallel = config.maxParallelChunks ?? DEFAULT_MAX_PARALLEL_CHUNKS;
|
||||
const { chunkCount, effectiveChunkSize } = resolveChunkPlan(
|
||||
totalFrames,
|
||||
configChunkSize,
|
||||
maxParallel,
|
||||
);
|
||||
const chunks = buildChunkSlices(totalFrames, chunkCount, effectiveChunkSize);
|
||||
|
||||
const ffmpegVersion = await readFfmpegVersion();
|
||||
const producerVersion = readProducerVersion();
|
||||
const fontSnapshotSha = await readFontSnapshotSha();
|
||||
const runtimeEnv = snapshotRuntimeEnv();
|
||||
const lockedConfig = buildLockedRenderConfig({
|
||||
config,
|
||||
forceScreenshot,
|
||||
deviceScaleFactor,
|
||||
ffmpegVersion,
|
||||
effectiveChunkSize,
|
||||
chunkCount,
|
||||
runtimeEnv,
|
||||
});
|
||||
|
||||
// ── Freeze the plan ──
|
||||
// `freezePlan` writes meta/{composition,encoder,chunks}.json then walks
|
||||
// the planDir to compute planHash from the actual bytes the chunk worker
|
||||
// will read.
|
||||
const compositionJson: CompositionMetadataJson = {
|
||||
durationSeconds: job.duration ?? 0,
|
||||
width,
|
||||
height,
|
||||
fps: job.config.fps,
|
||||
videoCount: composition.videos.length,
|
||||
audioCount: composition.audios.length,
|
||||
imageCount: composition.images.length,
|
||||
};
|
||||
const dimensions: PlanDimensions = {
|
||||
fpsNum: config.fps,
|
||||
fpsDen: 1,
|
||||
width,
|
||||
height,
|
||||
format: config.format,
|
||||
};
|
||||
const freezeResult = await freezePlan({
|
||||
planDir,
|
||||
composition: compositionJson,
|
||||
encoder: lockedConfig,
|
||||
chunks,
|
||||
dimensions,
|
||||
producerVersion,
|
||||
fontSnapshotSha,
|
||||
durationSeconds: job.duration ?? 0,
|
||||
totalFrames,
|
||||
hasAudio: audioResult.hasAudio,
|
||||
});
|
||||
const planHash = freezeResult.planHash;
|
||||
|
||||
// Clean up the temp work tree. `.plan-work/` holds intermediate
|
||||
// compileStage artifacts that are now promoted into `planDir/`; leaving
|
||||
// it would inflate the planDir-size check and confuse chunk workers' file
|
||||
// walks.
|
||||
try {
|
||||
rmSync(workDir, { recursive: true, force: true });
|
||||
} catch (err) {
|
||||
log.warn("[plan] failed to remove temp work dir", {
|
||||
workDir,
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
}
|
||||
|
||||
return {
|
||||
planDir,
|
||||
planHash,
|
||||
chunkCount,
|
||||
totalFrames,
|
||||
fps: config.fps,
|
||||
width,
|
||||
height,
|
||||
format: config.format,
|
||||
ffmpegVersion,
|
||||
producerVersion,
|
||||
};
|
||||
}
|
||||
@@ -915,6 +915,22 @@ export function collectExternalAssets(
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Optional behavior toggles for {@link compileForRender}. All fields are
|
||||
* additive; omitting `options` preserves the in-process renderer's defaults.
|
||||
*/
|
||||
export interface CompileForRenderOptions {
|
||||
/**
|
||||
* Threaded through to {@link injectDeterministicFontFaces}. When `true`,
|
||||
* any external font fetch failure throws `FontFetchError` instead of
|
||||
* silently falling back to system fonts. Distributed `plan()` sets this
|
||||
* to `true` so font availability is part of the planDir's content-addressed
|
||||
* hash and fetch failures surface as typed non-retryable errors. Default
|
||||
* `false` preserves the in-process behavior.
|
||||
*/
|
||||
failClosedFontFetch?: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Compile an HTML composition project into a single self-contained HTML string
|
||||
* with all media metadata resolved.
|
||||
@@ -923,6 +939,7 @@ export async function compileForRender(
|
||||
projectDir: string,
|
||||
htmlPath: string,
|
||||
downloadDir: string,
|
||||
options: CompileForRenderOptions = {},
|
||||
): Promise<CompiledComposition> {
|
||||
const rawHtml = readFileSync(htmlPath, "utf-8");
|
||||
const { html: compiledHtml, unresolvedCompositions } = await compileHtmlFile(
|
||||
@@ -963,6 +980,7 @@ export async function compileForRender(
|
||||
|
||||
const coalescedHtml = await injectDeterministicFontFaces(
|
||||
coalesceHeadStylesAndBodyScripts(promoteCssImportsToLinkTags(sanitizedHtml)),
|
||||
{ failClosedFontFetch: options.failClosedFontFetch === true },
|
||||
);
|
||||
|
||||
// Download CDN scripts and inline them AFTER coalescing. This order matters:
|
||||
|
||||
@@ -18,9 +18,9 @@
|
||||
* and the orchestrator has told us whether the output format demands an
|
||||
* alpha channel. The resolved boolean is also returned on the stage's
|
||||
* result so downstream stages can consume the value as an explicit
|
||||
* parameter instead of reading `cfg.forceScreenshot` directly. See the
|
||||
* distributed-render plan §4.3 — `LockedRenderConfig.forceScreenshot`
|
||||
* is computed here and frozen for the rest of the pipeline.
|
||||
* parameter instead of reading `cfg.forceScreenshot` directly. The
|
||||
* resolved value also flows into `LockedRenderConfig.forceScreenshot`
|
||||
* for distributed renders, where it must be frozen at plan time.
|
||||
*
|
||||
* Hard constraints preserved verbatim from the in-process renderer:
|
||||
* - `perfStages.compileOnlyMs` is set to wall-clock ms around the
|
||||
@@ -70,6 +70,14 @@ export interface CompileStageInput {
|
||||
log: ProducerLogger;
|
||||
/** Cooperative-cancellation probe; throws `RenderCancelledError` when aborted. */
|
||||
assertNotAborted: () => void;
|
||||
/**
|
||||
* When `true`, `compileForRender` threads through to
|
||||
* `injectDeterministicFontFaces` and any external font fetch failure
|
||||
* throws `FontFetchError` instead of silently falling back to system
|
||||
* fonts. Distributed `plan()` passes `true`; the in-process renderer
|
||||
* leaves it `undefined` to preserve current behavior.
|
||||
*/
|
||||
failClosedFontFetch?: boolean;
|
||||
}
|
||||
|
||||
export interface CompileStageResult {
|
||||
@@ -92,11 +100,23 @@ export interface CompileStageResult {
|
||||
}
|
||||
|
||||
export async function runCompileStage(input: CompileStageInput): Promise<CompileStageResult> {
|
||||
const { projectDir, workDir, htmlPath, entryFile, job, cfg, needsAlpha, log, assertNotAborted } =
|
||||
input;
|
||||
const {
|
||||
projectDir,
|
||||
workDir,
|
||||
htmlPath,
|
||||
entryFile,
|
||||
job,
|
||||
cfg,
|
||||
needsAlpha,
|
||||
log,
|
||||
assertNotAborted,
|
||||
failClosedFontFetch,
|
||||
} = input;
|
||||
|
||||
const compileStart = Date.now();
|
||||
const compiled = await compileForRender(projectDir, htmlPath, join(workDir, "downloads"));
|
||||
const compiled = await compileForRender(projectDir, htmlPath, join(workDir, "downloads"), {
|
||||
failClosedFontFetch: failClosedFontFetch === true,
|
||||
});
|
||||
assertNotAborted();
|
||||
const compileOnlyMs = Date.now() - compileStart;
|
||||
// Fold three signals into a single capture-mode decision: caller's
|
||||
|
||||
@@ -3,20 +3,27 @@
|
||||
* manifest at the end of `plan()`, compute the planHash from the frozen
|
||||
* artifacts, and return the manifest path.
|
||||
*
|
||||
* Signature-only skeleton: there are no callers yet. The function body
|
||||
* lands when `services/distributed/plan.ts` is added and composes the
|
||||
* stage primitives.
|
||||
*
|
||||
* See DISTRIBUTED-RENDERING-PLAN.md §2.1 phase 6, §4.1 directory layout,
|
||||
* §4.3 LockedRenderConfig.
|
||||
* Called from `services/distributed/plan.ts` after all earlier phases have
|
||||
* materialized their on-disk artifacts under `<planDir>/`. The function is
|
||||
* deliberately the last step so `planHash` is computed from the actual bytes
|
||||
* the chunk worker will read — not from intermediate values the controller
|
||||
* has in memory.
|
||||
*/
|
||||
|
||||
import { existsSync, mkdirSync, readFileSync, readdirSync, writeFileSync } from "node:fs";
|
||||
import { join, relative, resolve } from "node:path";
|
||||
|
||||
import type { Fps } from "@hyperframes/core";
|
||||
import type { PlanDimensions } from "./planHash.js";
|
||||
import {
|
||||
canonicalJsonStringify,
|
||||
computePlanHash,
|
||||
type PlanAssetHash,
|
||||
type PlanDimensions,
|
||||
sha256Hex,
|
||||
} from "./planHash.js";
|
||||
|
||||
/**
|
||||
* The encoder configuration locked in at plan time. Mirrors §4.3
|
||||
* LockedRenderConfig in the design doc.
|
||||
* The encoder configuration locked in at plan time.
|
||||
*/
|
||||
export interface LockedRenderConfig {
|
||||
// Capture
|
||||
@@ -30,6 +37,13 @@ export interface LockedRenderConfig {
|
||||
|
||||
// Encode
|
||||
encoder: "libx264-software" | "libx265-software" | "prores-software" | "png-sequence";
|
||||
/**
|
||||
* Caller-supplied quality enum, persisted so chunk workers can rebuild
|
||||
* the matching `getEncoderPreset(quality, format, …)` instead of
|
||||
* inferring quality from the encoder discriminant (which loses
|
||||
* information when the encoder→quality table grows non-injective).
|
||||
*/
|
||||
quality: "draft" | "standard" | "high";
|
||||
ffmpegVersion: string;
|
||||
preset: string;
|
||||
crf?: number;
|
||||
@@ -61,14 +75,15 @@ export interface CompositionMetadataJson {
|
||||
export interface ChunkSliceJson {
|
||||
index: number;
|
||||
startFrame: number;
|
||||
/** Inclusive end frame for the chunk. */
|
||||
/** Exclusive upper bound — chunk workers iterate frames in `[startFrame, endFrame)`. */
|
||||
endFrame: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Inputs to `freezePlan`. `planDir` already contains `compiled/`,
|
||||
* `video-frames/`, and (optionally) `audio.aac` by the time freezePlan
|
||||
* runs — see §2.1 phases 1-5.
|
||||
* runs — those are materialized by the upstream compile/probe/extract/audio
|
||||
* stages composed in `services/distributed/plan.ts`.
|
||||
*/
|
||||
export interface FreezePlanInput {
|
||||
/** Absolute path to the plan directory being frozen. */
|
||||
@@ -80,12 +95,18 @@ export interface FreezePlanInput {
|
||||
producerVersion: string;
|
||||
/** Hash of the deterministic-font snapshot baked into the plan. */
|
||||
fontSnapshotSha: string;
|
||||
/** Composition duration in seconds (mirrors `composition.durationSeconds`; carried separately for `plan.json`). */
|
||||
durationSeconds: number;
|
||||
/** Total frame count, separately materialized for callers that read `plan.json` without parsing chunks.json. */
|
||||
totalFrames: number;
|
||||
/** Whether `<planDir>/audio.aac` was produced. */
|
||||
hasAudio: boolean;
|
||||
}
|
||||
|
||||
export interface FreezePlanResult {
|
||||
/** Absolute path to `plan.json`. */
|
||||
planJsonPath: string;
|
||||
/** Content-addressed planHash; see §4.2. */
|
||||
/** Content-addressed planHash; see {@link computePlanHash}. */
|
||||
planHash: string;
|
||||
}
|
||||
|
||||
@@ -95,17 +116,174 @@ export interface FreezePlanResult {
|
||||
* `../runtimeEnvSnapshot.ts` — chunk workers re-apply the snapshot during
|
||||
* boot, so it needs to be importable without dragging in the freeze pipeline.
|
||||
*/
|
||||
export { snapshotRuntimeEnv, RUNTIME_ENV_PREFIXES } from "../runtimeEnvSnapshot.js";
|
||||
export { RUNTIME_ENV_PREFIXES, snapshotRuntimeEnv } from "../runtimeEnvSnapshot.js";
|
||||
|
||||
/** The relative path inside `<planDir>/` to the compiled HTML. */
|
||||
const COMPILED_INDEX_RELATIVE_PATH = "compiled/index.html";
|
||||
/** Files whose contents are framing of the plan itself, not assets. */
|
||||
const HASH_EXCLUDED_PLAN_FILES = new Set<string>([
|
||||
"plan.json",
|
||||
"meta/encoder.json",
|
||||
COMPILED_INDEX_RELATIVE_PATH,
|
||||
]);
|
||||
|
||||
/**
|
||||
* Recursively drop keys whose value is `undefined`. Preserves arrays and
|
||||
* primitive leaves. Used to sanitize the `LockedRenderConfig` before
|
||||
* canonical-JSON serialization so optional fields collapse to "absent"
|
||||
* rather than tripping `canonicalJsonStringify`'s undefined-rejection.
|
||||
*/
|
||||
function stripUndefined(value: unknown): unknown {
|
||||
if (Array.isArray(value)) return value.map(stripUndefined);
|
||||
if (value !== null && typeof value === "object") {
|
||||
const obj = value as Record<string, unknown>;
|
||||
const out: Record<string, unknown> = {};
|
||||
for (const key of Object.keys(obj)) {
|
||||
const v = obj[key];
|
||||
if (v === undefined) continue;
|
||||
out[key] = stripUndefined(v);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
/**
|
||||
* Walk `<planDir>/` depth-first; return a sorted, deterministic list of
|
||||
* `{ planRelativePath, absolutePath }` entries. Symlinks are skipped — the
|
||||
* `extractVideosStage` materializes them when `materializeSymlinks: true`,
|
||||
* so anything that slips through is by definition something the caller did
|
||||
* not intend to expose to chunk workers across machines.
|
||||
*/
|
||||
function listPlanFiles(planDir: string): Array<{ planRelativePath: string; absolutePath: string }> {
|
||||
const results: Array<{ planRelativePath: string; absolutePath: string }> = [];
|
||||
const rootResolved = resolve(planDir);
|
||||
|
||||
function walk(dir: string): void {
|
||||
const entries = readdirSync(dir, { withFileTypes: true });
|
||||
for (const entry of entries) {
|
||||
const full = join(dir, entry.name);
|
||||
if (entry.isDirectory()) {
|
||||
walk(full);
|
||||
} else if (entry.isFile()) {
|
||||
results.push({
|
||||
planRelativePath: relative(rootResolved, full)
|
||||
.split(/[\\/]+/)
|
||||
.join("/"),
|
||||
absolutePath: full,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
walk(rootResolved);
|
||||
results.sort((a, b) => (a.planRelativePath < b.planRelativePath ? -1 : 1));
|
||||
return results;
|
||||
}
|
||||
|
||||
/**
|
||||
* Read `compiled/index.html` and SHA every other regular file under `<planDir>/`
|
||||
* except the ones whose contents constitute the plan framing itself. Returns
|
||||
* the compiled HTML bytes (mixed verbatim into `planHash`) and the
|
||||
* sorted-by-path asset hashes.
|
||||
*/
|
||||
function collectPlanAssetShas(planDir: string): {
|
||||
compositionHtml: Uint8Array;
|
||||
assets: PlanAssetHash[];
|
||||
} {
|
||||
const files = listPlanFiles(planDir);
|
||||
let compositionHtml: Uint8Array | null = null;
|
||||
const assets: PlanAssetHash[] = [];
|
||||
for (const file of files) {
|
||||
if (file.planRelativePath === COMPILED_INDEX_RELATIVE_PATH) {
|
||||
compositionHtml = readFileSync(file.absolutePath);
|
||||
continue;
|
||||
}
|
||||
if (HASH_EXCLUDED_PLAN_FILES.has(file.planRelativePath)) continue;
|
||||
const bytes = readFileSync(file.absolutePath);
|
||||
assets.push({ path: file.planRelativePath, sha256: sha256Hex(bytes) });
|
||||
}
|
||||
if (compositionHtml === null) {
|
||||
throw new Error(
|
||||
`[freezePlan] compiled HTML missing at ${COMPILED_INDEX_RELATIVE_PATH} ` +
|
||||
`— upstream compile stage did not materialize the expected file.`,
|
||||
);
|
||||
}
|
||||
return { compositionHtml, assets };
|
||||
}
|
||||
|
||||
/**
|
||||
* Freeze a plan directory: write `meta/*.json` + top-level `plan.json`, then
|
||||
* compute `planHash` over the canonicalized contents.
|
||||
*
|
||||
* Skeleton — body lands when the distributed-render primitives compose the
|
||||
* stage functions. The body will resolve `input.encoder.runtimeEnv ||=
|
||||
* snapshotRuntimeEnv()` so callers can optionally pre-populate the field,
|
||||
* with the live env as the default.
|
||||
* The encoder JSON is written via {@link canonicalJsonStringify} so the bytes
|
||||
* fed into {@link computePlanHash} match the bytes on disk exactly. Consumers
|
||||
* can re-validate a plan by hashing `meta/encoder.json` directly.
|
||||
*/
|
||||
export async function freezePlan(_input: FreezePlanInput): Promise<FreezePlanResult> {
|
||||
throw new Error("freezePlan is not implemented yet.");
|
||||
export async function freezePlan(input: FreezePlanInput): Promise<FreezePlanResult> {
|
||||
const {
|
||||
planDir,
|
||||
composition,
|
||||
encoder,
|
||||
chunks,
|
||||
dimensions,
|
||||
producerVersion,
|
||||
fontSnapshotSha,
|
||||
durationSeconds,
|
||||
totalFrames,
|
||||
hasAudio,
|
||||
} = input;
|
||||
|
||||
if (!existsSync(planDir)) {
|
||||
throw new Error(`[freezePlan] planDir does not exist: ${planDir}`);
|
||||
}
|
||||
|
||||
const metaDir = join(planDir, "meta");
|
||||
if (!existsSync(metaDir)) mkdirSync(metaDir, { recursive: true });
|
||||
|
||||
writeFileSync(
|
||||
join(metaDir, "composition.json"),
|
||||
`${JSON.stringify(composition, null, 2)}\n`,
|
||||
"utf-8",
|
||||
);
|
||||
|
||||
// `LockedRenderConfig` has optional fields (`crf`, `bitrate`) that may be
|
||||
// `undefined`. `canonicalJsonStringify` deliberately throws on `undefined`
|
||||
// — JSON has no representation for it, and allowing it would silently
|
||||
// collapse two distinct configs (`{crf: 23}` vs `{crf: 23, bitrate: undefined}`)
|
||||
// into the same hash. Strip undefined values before canonicalizing so the
|
||||
// hashed config matches what is realistically a "missing field".
|
||||
const encoderForCanonical = stripUndefined(encoder) as Record<string, unknown>;
|
||||
const encoderConfigCanonicalJson = canonicalJsonStringify(encoderForCanonical);
|
||||
writeFileSync(join(metaDir, "encoder.json"), encoderConfigCanonicalJson, "utf-8");
|
||||
|
||||
writeFileSync(join(metaDir, "chunks.json"), `${JSON.stringify(chunks, null, 2)}\n`, "utf-8");
|
||||
|
||||
const { compositionHtml, assets } = collectPlanAssetShas(planDir);
|
||||
|
||||
const planHash = computePlanHash({
|
||||
compositionHtml,
|
||||
assets,
|
||||
fontSnapshotSha,
|
||||
encoderConfigCanonicalJson,
|
||||
producerVersion,
|
||||
ffmpegVersion: encoder.ffmpegVersion,
|
||||
dimensions,
|
||||
});
|
||||
|
||||
const planJson = {
|
||||
planHash,
|
||||
producerVersion,
|
||||
ffmpegVersion: encoder.ffmpegVersion,
|
||||
fontSnapshotSha,
|
||||
dimensions,
|
||||
chunkCount: chunks.length,
|
||||
totalFrames,
|
||||
duration: durationSeconds,
|
||||
hasAudio,
|
||||
};
|
||||
const planJsonPath = join(planDir, "plan.json");
|
||||
writeFileSync(planJsonPath, `${JSON.stringify(planJson, null, 2)}\n`, "utf-8");
|
||||
|
||||
return { planJsonPath, planHash };
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user