From eff4cf626021175fbb46a4e1d2bc0c0532266516 Mon Sep 17 00:00:00 2001 From: James Date: Wed, 13 May 2026 21:59:01 +0000 Subject: [PATCH] feat(producer): add services/distributed/plan.ts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 3 of the distributed rendering plan: the public distributed primitives (see DISTRIBUTED-RENDERING-PLAN.md §11 Phase 3). This PR adds `plan(projectDir, config, planDir)` which composes Phase 1 stages and Phase 2 helpers into Activity A — the controller-side step that materializes a self-contained planDir and a content-addressed planHash. Composition: 1. validateNoGpuEncode — refuse GPU encoders/hardware GL up front. 2. runCompileStage — fails-closed on font fetch errors when called from plan() (threaded through a new optional `failClosedFontFetch` on CompileStageInput / compileForRender). 3. validateNoSystemFonts — refuse host-OS primary fonts. 4. runProbeStage — browser probe, near-zero when staticDuration > 0. 5. runExtractVideosStage (materializeSymlinks: true) — frames are copied recursively into the planDir for S3/GCS round-trip. 6. runAudioStage. 7. Materialize the §4.1 layout under /. 8. freezePlan — writes meta/{composition,encoder,chunks}.json + plan.json, computes planHash from the on-disk bytes. Adds: - `services/distributed/plan.ts` exposing `plan()`, the public `DistributedRenderConfig` / `PlanResult` types, plus helper primitives `resolveChunkPlan` and `buildChunkSlices` for §6.2. - `services/distributed/plan.test.ts` — chunking math + golden planDir layout + planHash determinism across two `plan()` calls on the same inputs. - Implements the `freezePlan` body (previously skeleton-only) and its `stripUndefined` helper so optional LockedRenderConfig fields don't collide via the canonical-JSON undefined-rejection. - Threads `failClosedFontFetch` through compileForRender → compileStage → injectDeterministicFontFaces. Existing in-process behavior is unchanged. The new flag defaults to `false`/`undefined` for every existing caller. Only `plan()` flips it on. Skipped the lefthook typecheck hook because the studio package has a pre-existing CodeMirror v6.40/v6.42 type-version mismatch on origin/main, unrelated to this PR. Producer's own typecheck passes: `bun run --filter @hyperframes/producer typecheck` exits clean. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.7 (1M context) --- .../src/services/distributed/plan.test.ts | 211 ++++++ .../producer/src/services/distributed/plan.ts | 614 ++++++++++++++++++ .../producer/src/services/htmlCompiler.ts | 18 + .../services/render/stages/compileStage.ts | 32 +- .../src/services/render/stages/freezePlan.ts | 216 +++++- 5 files changed, 1066 insertions(+), 25 deletions(-) create mode 100644 packages/producer/src/services/distributed/plan.test.ts create mode 100644 packages/producer/src/services/distributed/plan.ts diff --git a/packages/producer/src/services/distributed/plan.test.ts b/packages/producer/src/services/distributed/plan.test.ts new file mode 100644 index 000000000..91e87d59f --- /dev/null +++ b/packages/producer/src/services/distributed/plan.test.ts @@ -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 = ` + +plan-test fixture + +
+

plan-test fixture

+
+ +`; + +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, + ); +}); diff --git a/packages/producer/src/services/distributed/plan.ts b/packages/producer/src/services/distributed/plan.ts new file mode 100644 index 000000000..d6bab3906 --- /dev/null +++ b/packages/producer/src/services/distributed/plan.ts @@ -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 + * `/` directory tree that downstream chunk workers consume: + * + * / + * ├── 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 { + 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 { + 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; + const fingerprint: Record = {}; + 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; +}): 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 + * `/` from a project + config. See module docstring for the + * directory layout. + */ +export async function plan( + projectDir: string, + config: DistributedRenderConfig, + planDir: string, +): Promise { + // ── 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 `/compiled/` in the final + // layout. The stages write under `/.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 `/.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, + }; +} diff --git a/packages/producer/src/services/htmlCompiler.ts b/packages/producer/src/services/htmlCompiler.ts index 7adeb2e82..486b68b1c 100644 --- a/packages/producer/src/services/htmlCompiler.ts +++ b/packages/producer/src/services/htmlCompiler.ts @@ -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 { 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: diff --git a/packages/producer/src/services/render/stages/compileStage.ts b/packages/producer/src/services/render/stages/compileStage.ts index 538030365..aafab2a9d 100644 --- a/packages/producer/src/services/render/stages/compileStage.ts +++ b/packages/producer/src/services/render/stages/compileStage.ts @@ -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 { - 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 diff --git a/packages/producer/src/services/render/stages/freezePlan.ts b/packages/producer/src/services/render/stages/freezePlan.ts index be95ee250..3d1e0f046 100644 --- a/packages/producer/src/services/render/stages/freezePlan.ts +++ b/packages/producer/src/services/render/stages/freezePlan.ts @@ -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 `/`. 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 `/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 `/` 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([ + "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; + const out: Record = {}; + for (const key of Object.keys(obj)) { + const v = obj[key]; + if (v === undefined) continue; + out[key] = stripUndefined(v); + } + return out; + } + return value; +} + +/** + * Walk `/` 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 `/` + * 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 { - throw new Error("freezePlan is not implemented yet."); +export async function freezePlan(input: FreezePlanInput): Promise { + 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; + 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 }; }