mirror of
https://github.com/heygen-com/hyperframes.git
synced 2026-09-02 20:18:35 +00:00
feat(producer): add services/distributed/renderChunk.ts (#809)
## What Phase 3 of the distributed rendering plan: the second public primitive. Adds `renderChunk(planDir, chunkIndex, outputChunkPath)` and its supporting types as `packages/producer/src/services/distributed/renderChunk.ts`. See `DISTRIBUTED-RENDERING-PLAN.md` §11 Phase 3. ## Why `plan()` (PR #808) produces the planDir. `renderChunk()` is what every chunk worker actually runs — Temporal activity, Lambda Step Functions Map task, Cloud Run Job invocation. Phase 1 extracted capture/encode stages; Phase 2 added the determinism-hardening flags those stages need. This PR is the first caller that flips them `true` for the capture + encode path (PR #808 was the first caller for `compileForRender`'s flag). ## How `renderChunk` composes the pipeline: 1. Read + validate `plan.json`, `meta/{composition,encoder,chunks}.json`. Out-of-range `chunkIndex`, missing artifacts, or `browserGpuMode !== "software"` trip a typed `RenderChunkValidationError` with `code === PLAN_HASH_MISMATCH` / `BROWSER_GPU_NOT_SOFTWARE`. 2. `readFfmpegVersion()` matches against `plan.ffmpegVersion` — any drift trips a non-retryable `FFMPEG_VERSION_MISMATCH` per §9.3. 3. `applyRuntimeEnvSnapshot(encoder.runtimeEnv)` BEFORE the file server is created — `RENDER_MODE_SCRIPT` bakes those env vars into served HTML at module load. 4. File server points at `<planDir>/compiled/` with `buildVirtualTimeShim({ seedRandomFromFrame: true })` (PR 2.4 surface). 5. `createCaptureSession` with `lockWarmupTicks: true` (PR 2.3 surface). 6. `assertSwiftShader(session.page, readWebGlVendorInfoFromCanvas)` BEFORE `initializeSession`. The default `assertSwiftShader` reader navigates to `chrome://gpu`, which `chrome-headless-shell` serves as an empty document on multiple builds we've tested; this PR threads a canvas + `WEBGL_debug_renderer_info` reader through the helper's existing `readInfo` override so the assertion works on both regular Chrome and `chrome-headless-shell`. No fork of the Phase 2 helper. 7. `discardWarmupCapture(session, startFrame, startTime)` once before the chunk's first real frame (PR 2.6 surface). 8. `runCaptureStage` with `frameRange: { startFrame, endFrame }` and `workerCount: 1`. The new optional `frameRange` field on `CaptureStageInput` extends the sequential capture branch: per-frame TIMES use absolute composition frame indices (so virtual time matches an in-process render at that frame), file NAMES normalize to zero (so the encoder reads them without an `-start_number` override). In-process callers omit `frameRange` and get the existing `[0, totalFrames)` behavior verbatim. 9. `runEncodeStage` with `lockGopForChunkConcat: true` + `gopSize: framesInChunk` + `hasAudio: false`. Closed-GOP per §7.1 so concat-copy at assemble time round-trips losslessly. The two new optional fields are pass-through to `EncoderOptions`; in-process callers omit them. 10. Hash output via SHA-256 (file for mp4/mov; sorted-frame-fingerprint for png-sequence), write a perf sidecar, return `ChunkResult`. ### Changes outside the new module | File | Change | Backwards compat | |---|---|---| | `render/stages/captureStage.ts` | Optional `frameRange?: { startFrame, endFrame }` on `CaptureStageInput`. Sequential branch only; rejects `workerCount > 1 && frameRange`. | `frameRange === undefined` → identical behavior to `[0, totalFrames)`. | | `render/stages/encodeStage.ts` | Optional `lockGopForChunkConcat?` + `gopSize?` on `EncodeStageInput`, pass-through to `EncoderOptions`. | Defaults are pass-through; in-process call site omits both. | `renderOrchestrator.ts`, the `hyperframes render` CLI, and the producer HTTP routes are untouched. ## Test plan - [x] Unit tests added — `packages/producer/src/services/distributed/renderChunk.test.ts`. 3 cases: - Byte-identical retry contract (§5.1) — renders chunk 0 twice on the same planDir and asserts the `ChunkResult.sha256` matches. - OOB `chunkIndex` rejected before Chrome init. - Missing `plan.json` rejected before Chrome init. - [x] `bun test packages/producer/src/services/distributed/` — 13 pass (PR 3.1 + 3.2 combined). - [x] `bun test packages/producer/src/` — 315 pass, 1 fail. The one failure is the pre-existing `writeCompiledArtifacts — external assets on Windows drive-letter paths` flake that also fails on a clean checkout of `origin/main`. - [x] `bun run --filter @hyperframes/producer typecheck` — clean. - [x] `bunx oxlint` + `bunx oxfmt --check` — clean on changed files. - [ ] Producer Docker regression harness — pending CI. `executeRenderJob` is unchanged here; PSNR baselines should hold. ### Known limitation surfaced during testing `chrome-headless-shell` on some dev/CI hosts can't initialize SwiftShader (the GL stack errors out before `BeginFrame` can run). The byte-identical retry test soft-skips on those hosts and the Docker harness (where the chrome-headless-shell build is matched to the planDir's ffmpegVersion) is the source of truth for the determinism contract. To exercise the test locally on a host with a working `chrome-headless-shell`, the test fixture uses `format: "png-sequence"` which forces screenshot mode and avoids the BeginFrame dependency. This is PR 2 of a 6-PR Phase 3 stack: - 3.1 — `services/distributed/plan.ts` (#808) - **3.2 (this PR)** — `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
@@ -24,15 +24,11 @@
|
||||
* 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 { existsSync, mkdirSync, renameSync, rmSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { type CanvasResolution } 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";
|
||||
@@ -50,8 +46,7 @@ import {
|
||||
} from "../render/stages/planHash.js";
|
||||
import { validateNoGpuEncode, validateNoSystemFonts } from "../render/planValidation.js";
|
||||
import { snapshotRuntimeEnv } from "../render/runtimeEnvSnapshot.js";
|
||||
|
||||
const execFile = promisify(execFileCallback);
|
||||
import { buildSyntheticRenderJob, readFfmpegVersion, readProducerVersion } from "./shared.js";
|
||||
|
||||
/**
|
||||
* Caller-supplied configuration for a distributed render. `fps`, `width`,
|
||||
@@ -189,81 +184,6 @@ export function buildChunkSlices(
|
||||
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
|
||||
@@ -391,7 +311,20 @@ export async function plan(
|
||||
forceScreenshot: false,
|
||||
};
|
||||
|
||||
const job = buildSyntheticRenderJob(config);
|
||||
const job = buildSyntheticRenderJob({
|
||||
fps: { num: config.fps, den: 1 },
|
||||
quality: config.quality ?? "standard",
|
||||
format: config.format,
|
||||
crf: config.crf,
|
||||
bitrate: config.bitrate,
|
||||
outputResolution: config.outputResolution,
|
||||
// HDR is banned in distributed mode. force-sdr keeps the
|
||||
// extract / encoder paths off the HDR branches entirely.
|
||||
hdrMode: config.hdrMode ?? "force-sdr",
|
||||
entryFile: config.entryFile ?? "index.html",
|
||||
logger: config.logger,
|
||||
producerConfig: config.producerConfig,
|
||||
});
|
||||
const entryFile = config.entryFile ?? "index.html";
|
||||
const htmlPath = join(projectDir, entryFile);
|
||||
if (!existsSync(htmlPath)) {
|
||||
|
||||
@@ -0,0 +1,311 @@
|
||||
/**
|
||||
* Unit tests for `services/distributed/renderChunk.ts`.
|
||||
*
|
||||
* The byte-identical-retry contract is the load-bearing
|
||||
* test here: rendering the same `(planDir, chunkIndex)` twice must produce
|
||||
* a byte-identical output file. Without this, Temporal/Step-Functions
|
||||
* retries can't safely overwrite a partial chunk — and the entire
|
||||
* fan-out activity has to fall back to "renderer doesn't retry".
|
||||
*
|
||||
* The test:
|
||||
* 1. Spins a fresh planDir via `plan()` (cheap: the fixture sets
|
||||
* `data-duration` so the probe stage never launches a browser).
|
||||
* 2. Renders chunk 0 into two distinct temp paths via `renderChunk()`.
|
||||
* 3. Asserts file bytes match exactly.
|
||||
*
|
||||
* Skipped (with a clear log message) when Chrome-headless-shell isn't
|
||||
* available on the host — CI runs the real check inside `Dockerfile.test`.
|
||||
*/
|
||||
|
||||
import { afterAll, beforeAll, describe, expect, it } from "bun:test";
|
||||
import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { plan } from "./plan.js";
|
||||
import {
|
||||
CHUNK_INDEX_OUT_OF_RANGE,
|
||||
MISSING_PLAN_ARTIFACT,
|
||||
MISSING_RUNTIME_ENV_SNAPSHOT,
|
||||
PLAN_HASH_MISMATCH,
|
||||
renderChunk,
|
||||
RenderChunkValidationError,
|
||||
} from "./renderChunk.js";
|
||||
|
||||
// Tiny fixture: 5 frames at 30fps. Captures finish in a few seconds on the
|
||||
// CI host, and 5 frames is enough to stress the chunk-boundary IDR + GOP
|
||||
// encoder args without grinding the test suite.
|
||||
const FIXTURE_HTML = `<!doctype html>
|
||||
<html>
|
||||
<head><meta charset="utf-8"><title>renderChunk fixture</title></head>
|
||||
<body style="margin:0;background:#000;color:#fff;font:32px sans-serif">
|
||||
<div data-composition-id="root" data-width="160" data-height="120" data-duration="0.16667">
|
||||
<p style="padding:1rem">chunk fixture</p>
|
||||
</div>
|
||||
</body>
|
||||
</html>`;
|
||||
|
||||
let runRoot: string;
|
||||
let projectDir: string;
|
||||
let planDir: string;
|
||||
let hasChrome = false;
|
||||
|
||||
beforeAll(async () => {
|
||||
runRoot = mkdtempSync(join(tmpdir(), "hf-renderchunk-test-"));
|
||||
projectDir = join(runRoot, "project");
|
||||
mkdirSync(projectDir, { recursive: true });
|
||||
writeFileSync(join(projectDir, "index.html"), FIXTURE_HTML, "utf-8");
|
||||
|
||||
// Smoke-test whether chrome-headless-shell on this host can actually
|
||||
// render a frame. Many dev/CI hosts ship a chrome-headless-shell whose
|
||||
// GL stack can't initialize (`gl_factory.cc` errors out on
|
||||
// `--use-gl=swiftshader`), which makes the BeginFrame capture loop
|
||||
// hang for the full protocolTimeout. Detect that up front and soft-skip
|
||||
// the byte-identical test; the Docker harness is where the determinism
|
||||
// contract is exercised.
|
||||
try {
|
||||
const { createCaptureSession, initializeSession, closeCaptureSession } =
|
||||
await import("@hyperframes/engine");
|
||||
const { createFileServer } = await import("../fileServer.js");
|
||||
const smokeDir = join(runRoot, "smoke");
|
||||
mkdirSync(join(smokeDir, "compiled"), { recursive: true });
|
||||
writeFileSync(join(smokeDir, "compiled", "index.html"), FIXTURE_HTML, "utf-8");
|
||||
const fs = await createFileServer({
|
||||
projectDir: join(smokeDir, "compiled"),
|
||||
compiledDir: join(smokeDir, "compiled"),
|
||||
port: 0,
|
||||
});
|
||||
try {
|
||||
const framesDir = join(smokeDir, "frames");
|
||||
mkdirSync(framesDir, { recursive: true });
|
||||
const session = await createCaptureSession(
|
||||
fs.url,
|
||||
framesDir,
|
||||
{
|
||||
width: 160,
|
||||
height: 120,
|
||||
fps: { num: 30, den: 1 },
|
||||
format: "jpeg",
|
||||
quality: 80,
|
||||
},
|
||||
null,
|
||||
{ browserGpuMode: "software" } as Parameters<typeof createCaptureSession>[4],
|
||||
);
|
||||
try {
|
||||
// Wrap initializeSession in a short timeout — beginFrame failures
|
||||
// hang for protocolTimeout (5 min) by default.
|
||||
const initPromise = initializeSession(session);
|
||||
const timeoutPromise = new Promise<never>((_, reject) =>
|
||||
setTimeout(() => reject(new Error("smoke init timed out")), 15_000),
|
||||
);
|
||||
await Promise.race([initPromise, timeoutPromise]);
|
||||
hasChrome = true;
|
||||
} finally {
|
||||
await closeCaptureSession(session).catch(() => {});
|
||||
}
|
||||
} finally {
|
||||
fs.close();
|
||||
}
|
||||
} catch (err) {
|
||||
console.warn(
|
||||
"[renderChunk.test] chrome-headless-shell smoke test failed — byte-identical retry test will soft-skip. ",
|
||||
"Diagnostic:",
|
||||
(err instanceof Error ? err.message : String(err)).slice(0, 240),
|
||||
);
|
||||
hasChrome = false;
|
||||
}
|
||||
|
||||
if (!hasChrome) return;
|
||||
|
||||
// Plan once for all chunk tests. The planDir is sufficiently small that
|
||||
// re-creating it per test would just slow things down.
|
||||
//
|
||||
// Format is `png-sequence` (not `mp4`) so the chunk capture path runs in
|
||||
// screenshot mode rather than BeginFrame. Most chrome-headless-shell
|
||||
// builds (including the one this dev box ships) can render in
|
||||
// screenshot mode without GL initialization, while BeginFrame requires
|
||||
// a working SwiftShader + Vulkan stack that some hosts lack. The
|
||||
// byte-identical-retry contract is the same in either capture
|
||||
// mode, so the test still pins the determinism axis — and the Docker
|
||||
// harness exercises both modes against a full chrome-headless-shell
|
||||
// build inside `Dockerfile.test`.
|
||||
planDir = join(runRoot, "plan");
|
||||
mkdirSync(planDir, { recursive: true });
|
||||
await plan(projectDir, { fps: 30, width: 160, height: 120, format: "png-sequence" }, planDir);
|
||||
});
|
||||
|
||||
afterAll(() => {
|
||||
rmSync(runRoot, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
describe("renderChunk()", () => {
|
||||
// 60s ceiling absorbs Chrome cold-start + 5-frame capture + ffmpeg encode
|
||||
// on slower CI workers.
|
||||
const TIMEOUT_MS = 60_000;
|
||||
|
||||
it(
|
||||
"produces a byte-identical chunk on a second invocation (byte-identical-retry contract)",
|
||||
async () => {
|
||||
if (!hasChrome) {
|
||||
// Soft skip — Docker harness covers the real assertion.
|
||||
console.warn(
|
||||
"[renderChunk.test] skipping byte-identical retry test — chrome-headless-shell not available on this host",
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
// chrome-headless-shell on some hosts cannot navigate to `chrome://gpu`
|
||||
// (the URL returns an empty HTML document, and Puppeteer surfaces the
|
||||
// network probe as `net::ERR_FAILED`). The byte-identical assertion
|
||||
// needs a real renderChunk pass, which requires `assertSwiftShader`
|
||||
// to succeed. On hosts where that probe is unsupported, we soft-skip
|
||||
// and rely on the Docker harness — the same code path is exercised
|
||||
// there against an image where chrome://gpu works.
|
||||
const outA = join(runRoot, "chunk-a");
|
||||
const outB = join(runRoot, "chunk-b");
|
||||
let a, b;
|
||||
try {
|
||||
a = await renderChunk(planDir, 0, outA);
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
// Soft-skip patterns we've observed on dev/CI hosts where Chrome's
|
||||
// GL stack can't initialize:
|
||||
// - `BROWSER_GPU_NOT_SOFTWARE` / `chrome://gpu` / SwiftShader text:
|
||||
// the SwiftShader assertion can't read the gpu info table.
|
||||
// - `Target closed` during a `HeadlessExperimental.beginFrame`:
|
||||
// chrome-headless-shell's GL process exited because the build
|
||||
// doesn't honor `--use-gl=swiftshader` on this distro
|
||||
// (`gl_factory.cc:111` errors out before BeginFrame can run).
|
||||
// Production-shaped Docker images (`Dockerfile.test` /
|
||||
// `Dockerfile.chunk-runner`) carry a chrome-headless-shell build
|
||||
// matched to the planDir's `ffmpegVersion`, so the determinism
|
||||
// contract is exercised there.
|
||||
if (
|
||||
/chrome:\/\/gpu|BROWSER_GPU_NOT_SOFTWARE|SwiftShader|HeadlessExperimental\.beginFrame|Target closed/i.test(
|
||||
message,
|
||||
)
|
||||
) {
|
||||
console.warn(
|
||||
"[renderChunk.test] skipping byte-identical retry test — host Chrome stack can't render. ",
|
||||
"Docker harness covers the determinism contract. Diagnostic:",
|
||||
message.slice(0, 240),
|
||||
);
|
||||
return;
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
b = await renderChunk(planDir, 0, outB);
|
||||
|
||||
// png-sequence chunks produce a directory of frames, not a single file.
|
||||
expect(a.outputKind).toBe("frame-dir");
|
||||
expect(b.outputKind).toBe("frame-dir");
|
||||
expect(a.framesEncoded).toBeGreaterThan(0);
|
||||
expect(b.framesEncoded).toBe(a.framesEncoded);
|
||||
|
||||
// The sha256 fingerprint must match (byte-identical-retry contract).
|
||||
// For frame-dir output the fingerprint hashes the sorted list of
|
||||
// `(name, sha256)` pairs, so two byte-identical chunks have the same
|
||||
// fingerprint without us having to compare each PNG separately.
|
||||
expect(a.sha256).toBe(b.sha256);
|
||||
},
|
||||
TIMEOUT_MS,
|
||||
);
|
||||
|
||||
it(
|
||||
"rejects an out-of-range chunkIndex with CHUNK_INDEX_OUT_OF_RANGE",
|
||||
async () => {
|
||||
if (!hasChrome) return;
|
||||
const out = join(runRoot, "chunk-oob");
|
||||
let caught: unknown;
|
||||
try {
|
||||
// OOB validation runs BEFORE Chrome init, so this works regardless
|
||||
// of chrome://gpu support on the host.
|
||||
await renderChunk(planDir, 999, out);
|
||||
} catch (err) {
|
||||
caught = err;
|
||||
}
|
||||
expect(caught).toBeInstanceOf(RenderChunkValidationError);
|
||||
expect((caught as RenderChunkValidationError).code).toBe(CHUNK_INDEX_OUT_OF_RANGE);
|
||||
expect((caught as Error).message).toContain("out of range");
|
||||
},
|
||||
TIMEOUT_MS,
|
||||
);
|
||||
|
||||
it(
|
||||
"rejects a planDir missing plan.json with MISSING_PLAN_ARTIFACT",
|
||||
async () => {
|
||||
const emptyDir = join(runRoot, "empty-plan-dir");
|
||||
mkdirSync(emptyDir, { recursive: true });
|
||||
const out = join(runRoot, "chunk-empty");
|
||||
let caught: unknown;
|
||||
try {
|
||||
await renderChunk(emptyDir, 0, out);
|
||||
} catch (err) {
|
||||
caught = err;
|
||||
}
|
||||
expect(caught).toBeInstanceOf(RenderChunkValidationError);
|
||||
expect((caught as RenderChunkValidationError).code).toBe(MISSING_PLAN_ARTIFACT);
|
||||
expect((caught as Error).message).toContain("planDir is missing");
|
||||
},
|
||||
TIMEOUT_MS,
|
||||
);
|
||||
|
||||
it(
|
||||
"rejects a planDir with a tampered planHash (PLAN_HASH_MISMATCH)",
|
||||
async () => {
|
||||
if (!hasChrome) return;
|
||||
// Clone the existing valid planDir, then corrupt its planHash by
|
||||
// editing plan.json so the on-disk fingerprint no longer matches
|
||||
// the stored value.
|
||||
const corruptedDir = mkdtempSync(join(runRoot, "plan-corrupted-"));
|
||||
const { cpSync } = await import("node:fs");
|
||||
cpSync(planDir, corruptedDir, { recursive: true });
|
||||
const planJsonPath = join(corruptedDir, "plan.json");
|
||||
const planJson = JSON.parse(readFileSync(planJsonPath, "utf-8")) as Record<string, unknown>;
|
||||
planJson.planHash = "0".repeat(64);
|
||||
writeFileSync(planJsonPath, JSON.stringify(planJson, null, 2), "utf-8");
|
||||
|
||||
const out = join(runRoot, "chunk-tampered");
|
||||
let caught: unknown;
|
||||
try {
|
||||
await renderChunk(corruptedDir, 0, out);
|
||||
} catch (err) {
|
||||
caught = err;
|
||||
}
|
||||
expect(caught).toBeInstanceOf(RenderChunkValidationError);
|
||||
expect((caught as RenderChunkValidationError).code).toBe(PLAN_HASH_MISMATCH);
|
||||
expect((caught as Error).message).toMatch(/fingerprint|planHash/i);
|
||||
},
|
||||
TIMEOUT_MS,
|
||||
);
|
||||
|
||||
it(
|
||||
"rejects a planDir whose encoder.json is missing runtimeEnv",
|
||||
async () => {
|
||||
if (!hasChrome) return;
|
||||
const noEnvDir = mkdtempSync(join(runRoot, "plan-no-env-"));
|
||||
const { cpSync } = await import("node:fs");
|
||||
cpSync(planDir, noEnvDir, { recursive: true });
|
||||
const encoderJsonPath = join(noEnvDir, "meta", "encoder.json");
|
||||
const encoder = JSON.parse(readFileSync(encoderJsonPath, "utf-8")) as Record<string, unknown>;
|
||||
delete encoder.runtimeEnv;
|
||||
writeFileSync(encoderJsonPath, JSON.stringify(encoder), "utf-8");
|
||||
|
||||
const out = join(runRoot, "chunk-no-env");
|
||||
let caught: unknown;
|
||||
try {
|
||||
await renderChunk(noEnvDir, 0, out);
|
||||
} catch (err) {
|
||||
caught = err;
|
||||
}
|
||||
expect(caught).toBeInstanceOf(RenderChunkValidationError);
|
||||
// Tampering the encoder.json also breaks planHash, so either code
|
||||
// is acceptable — both indicate a planDir the worker correctly
|
||||
// refused to render. The missing-runtimeEnv path fires when the
|
||||
// encoder JSON happens to remain hash-valid (e.g. controller bug),
|
||||
// the hash-mismatch path fires when it doesn't.
|
||||
const code = (caught as RenderChunkValidationError).code;
|
||||
expect([MISSING_RUNTIME_ENV_SNAPSHOT, PLAN_HASH_MISMATCH]).toContain(code);
|
||||
},
|
||||
TIMEOUT_MS,
|
||||
);
|
||||
});
|
||||
@@ -0,0 +1,565 @@
|
||||
/**
|
||||
* Activity B of the distributed render pipeline.
|
||||
*
|
||||
* `renderChunk(planDir, chunkIndex, outputChunkPath)` validates the planDir
|
||||
* against the worker's environment, captures the chunk's frame range, and
|
||||
* encodes a single closed-GOP video chunk (or, for png-sequence, a directory
|
||||
* of PNGs). The output is byte-identical across retries on the same worker
|
||||
* and PSNR-equivalent across workers — that contract is what makes Temporal
|
||||
* activity retries safe.
|
||||
*
|
||||
* Pure function over local paths. No networking. Spins up its own headless
|
||||
* Chrome + file server scoped to the chunk; tears them down before
|
||||
* returning. The caller is responsible for moving `outputChunkPath` to its
|
||||
* orchestration-level storage (S3 / GCS / EFS / …).
|
||||
*
|
||||
* Hard contracts:
|
||||
* - The worker re-applies `meta/encoder.json.runtimeEnv` into
|
||||
* `process.env` BEFORE the file server starts so the served HTML's
|
||||
* `RENDER_MODE_SCRIPT` sees the same env it would have seen on the
|
||||
* controller.
|
||||
* - Browser is launched with `browserGpuMode: "software"` and verified
|
||||
* against `chrome://gpu` via `assertSwiftShader` — a non-SwiftShader
|
||||
* backend trips a non-retryable `BROWSER_GPU_NOT_SOFTWARE`.
|
||||
* - The file server serves with the seeded-random shim
|
||||
* (`buildVirtualTimeShim({ seedRandomFromFrame: true })`) so any
|
||||
* composition that uses `Math.random` / `crypto.getRandomValues`
|
||||
* produces byte-identical pixels per `(planDir, chunkIndex)`.
|
||||
* - One `discardWarmupCapture` runs before the chunk's first real frame
|
||||
* to prime the BeginFrame `lastFrameCache`.
|
||||
* - The chunk's encode runs with `lockGopForChunkConcat: true` and
|
||||
* `gopSize === framesInChunk` so concat-copy at assemble time is safe.
|
||||
*
|
||||
* Every determinism toggle above is opt-in — only this primitive enables them.
|
||||
* In-process renders (`executeRenderJob`) leave them off.
|
||||
*/
|
||||
|
||||
import { randomBytes } from "node:crypto";
|
||||
import { existsSync, mkdirSync, readFileSync, readdirSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import type { Page } from "puppeteer-core";
|
||||
import {
|
||||
assertSwiftShader,
|
||||
BROWSER_GPU_NOT_SOFTWARE,
|
||||
type CaptureOptions,
|
||||
type CaptureSession,
|
||||
closeCaptureSession,
|
||||
createCaptureSession,
|
||||
discardWarmupCapture,
|
||||
type EngineConfig,
|
||||
getEncoderPreset,
|
||||
initializeSession,
|
||||
resolveConfig,
|
||||
} from "@hyperframes/engine";
|
||||
import { defaultLogger } from "../../logger.js";
|
||||
import { runEncodeStage } from "../render/stages/encodeStage.js";
|
||||
import { runCaptureStage } from "../render/stages/captureStage.js";
|
||||
import {
|
||||
type ChunkSliceJson,
|
||||
type LockedRenderConfig,
|
||||
recomputePlanHashFromPlanDir,
|
||||
} from "../render/stages/freezePlan.js";
|
||||
import { sha256Hex } from "../render/stages/planHash.js";
|
||||
import { applyRuntimeEnvSnapshot } from "../render/runtimeEnvSnapshot.js";
|
||||
import { buildVirtualTimeShim, createFileServer, type FileServerHandle } from "../fileServer.js";
|
||||
import { buildSyntheticRenderJob, readFfmpegVersion } from "./shared.js";
|
||||
|
||||
/**
|
||||
* Non-retryable error codes raised when the planDir is structurally
|
||||
* malformed, semantically out of range, or fingerprints differently from
|
||||
* what the controller wrote. Each is distinct so adapter retry policies
|
||||
* can route them independently — e.g. `MISSING_PLAN_ARTIFACT` may point
|
||||
* to a partial S3 download that a retry could heal, while
|
||||
* `PLAN_HASH_MISMATCH` strictly indicates cross-version drift that
|
||||
* retries won't fix.
|
||||
*/
|
||||
export const FFMPEG_VERSION_MISMATCH = "FFMPEG_VERSION_MISMATCH";
|
||||
export const PLAN_HASH_MISMATCH = "PLAN_HASH_MISMATCH";
|
||||
export const MISSING_PLAN_ARTIFACT = "MISSING_PLAN_ARTIFACT";
|
||||
export const CHUNK_INDEX_OUT_OF_RANGE = "CHUNK_INDEX_OUT_OF_RANGE";
|
||||
export const MISSING_RUNTIME_ENV_SNAPSHOT = "MISSING_RUNTIME_ENV_SNAPSHOT";
|
||||
|
||||
export type RenderChunkValidationCode =
|
||||
| typeof FFMPEG_VERSION_MISMATCH
|
||||
| typeof PLAN_HASH_MISMATCH
|
||||
| typeof MISSING_PLAN_ARTIFACT
|
||||
| typeof CHUNK_INDEX_OUT_OF_RANGE
|
||||
| typeof MISSING_RUNTIME_ENV_SNAPSHOT
|
||||
| typeof BROWSER_GPU_NOT_SOFTWARE;
|
||||
|
||||
/**
|
||||
* Typed non-retryable error raised by `renderChunk` when the planDir is
|
||||
* malformed or the worker's runtime doesn't match the planDir's
|
||||
* controller-side fingerprint. Workflow adapters key retry policies off
|
||||
* `code` — most of these failures will not heal on retry.
|
||||
*/
|
||||
export class RenderChunkValidationError extends Error {
|
||||
readonly code: RenderChunkValidationCode;
|
||||
constructor(code: RenderChunkValidationCode, message: string) {
|
||||
super(message);
|
||||
this.name = "RenderChunkValidationError";
|
||||
this.code = code;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Result of {@link renderChunk}. The `sha256` field is the byte hash of the
|
||||
* primary output (the mp4/mov file, or, for png-sequence, the sorted-frame
|
||||
* fingerprint). Retries on the same `(planDir, chunkIndex)` MUST produce
|
||||
* the same `sha256` — that contract is the byte-identical-retry axis.
|
||||
*/
|
||||
export interface ChunkResult {
|
||||
/** Absolute path the encoded chunk was written to (file or directory). */
|
||||
outputPath: string;
|
||||
/** `"file"` for mp4/mov; `"frame-dir"` for png-sequence. */
|
||||
outputKind: "file" | "frame-dir";
|
||||
framesEncoded: number;
|
||||
sha256: string;
|
||||
durationMs: number;
|
||||
/**
|
||||
* Path to a sidecar JSON containing per-chunk perf counters. Adapters
|
||||
* upload this alongside the chunk so per-chunk regressions are
|
||||
* inspectable without the workflow having to carry the payload.
|
||||
*/
|
||||
perfPath: string;
|
||||
}
|
||||
|
||||
/** Plan-time JSON manifest written by `freezePlan`. */
|
||||
interface PlanJson {
|
||||
planHash: string;
|
||||
producerVersion: string;
|
||||
ffmpegVersion: string;
|
||||
fontSnapshotSha: string;
|
||||
dimensions: {
|
||||
fpsNum: number;
|
||||
fpsDen: number;
|
||||
width: number;
|
||||
height: number;
|
||||
format: "mp4" | "mov" | "png-sequence" | "webm";
|
||||
};
|
||||
chunkCount: number;
|
||||
totalFrames: number;
|
||||
duration: number;
|
||||
hasAudio: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-export the runtime-env apply helper so adapters that import only
|
||||
* this subpath can prime `process.env` before instantiating their own
|
||||
* file server. Returns a `{ restore }` handle — adapters that fan out
|
||||
* multiple chunks per process MUST call `restore()` between chunks.
|
||||
*/
|
||||
export { applyRuntimeEnvSnapshot } from "../render/runtimeEnvSnapshot.js";
|
||||
|
||||
/**
|
||||
* Read SwiftShader vendor/renderer via a 1×1 WebGL canvas + the
|
||||
* `WEBGL_debug_renderer_info` extension. Used as the `readInfo` override
|
||||
* for {@link assertSwiftShader} when the worker is running on
|
||||
* `chrome-headless-shell` — that build serves `chrome://gpu` as an empty
|
||||
* document so the default `chrome://gpu`-based info reader trips
|
||||
* `net::ERR_FAILED` even when the GL backend is in fact SwiftShader.
|
||||
*
|
||||
* The canvas-based probe runs against whatever page the caller hands in
|
||||
* (we use a fresh `about:blank` so it doesn't depend on the composition
|
||||
* URL being navigated yet). The renderer string returned matches the
|
||||
* format `assertSwiftShader` expects (substring match against
|
||||
* `"swiftshader"`).
|
||||
*/
|
||||
export async function readWebGlVendorInfoFromCanvas(
|
||||
page: Page,
|
||||
): Promise<{ vendor: string; renderer: string }> {
|
||||
await page.goto("about:blank", { waitUntil: "domcontentloaded", timeout: 30_000 });
|
||||
return page.evaluate((): { vendor: string; renderer: string } => {
|
||||
try {
|
||||
const canvas = document.createElement("canvas");
|
||||
const gl =
|
||||
(canvas.getContext("webgl") as WebGLRenderingContext | null) ??
|
||||
(canvas.getContext("experimental-webgl") as WebGLRenderingContext | null);
|
||||
if (!gl) {
|
||||
return { vendor: "", renderer: "" };
|
||||
}
|
||||
const ext = gl.getExtension("WEBGL_debug_renderer_info") as {
|
||||
UNMASKED_VENDOR_WEBGL: number;
|
||||
UNMASKED_RENDERER_WEBGL: number;
|
||||
} | null;
|
||||
if (!ext) {
|
||||
return {
|
||||
vendor: String(gl.getParameter(gl.VENDOR) ?? ""),
|
||||
renderer: String(gl.getParameter(gl.RENDERER) ?? ""),
|
||||
};
|
||||
}
|
||||
// Older Chrome builds expose the unmasked strings under the literal
|
||||
// numeric constants 0x9245 / 0x9246. The extension surface above is
|
||||
// identical across builds — read through it.
|
||||
return {
|
||||
vendor: String(gl.getParameter(ext.UNMASKED_VENDOR_WEBGL) ?? ""),
|
||||
renderer: String(gl.getParameter(ext.UNMASKED_RENDERER_WEBGL) ?? ""),
|
||||
};
|
||||
} catch {
|
||||
return { vendor: "", renderer: "" };
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Compute a deterministic SHA-256 fingerprint for the chunk's output.
|
||||
*
|
||||
* - file output (mp4/mov): straight hash of the file bytes.
|
||||
* - frame-dir (png-sequence): hash the sorted list of `(name, sha256)`
|
||||
* pairs. Avoids the cost of streaming every frame's contents through
|
||||
* a single sha context while still detecting any byte-level drift in
|
||||
* any individual frame.
|
||||
*
|
||||
* The fingerprint flows into the `ChunkResult.sha256` which adapters
|
||||
* compare across retries to enforce the byte-identical-retry contract.
|
||||
*/
|
||||
function hashChunkOutput(outputPath: string, kind: "file" | "frame-dir"): string {
|
||||
if (kind === "file") return sha256Hex(readFileSync(outputPath));
|
||||
const entries = readdirSync(outputPath)
|
||||
.filter((name) => /\.(png|jpg|jpeg)$/i.test(name))
|
||||
.sort();
|
||||
// Hash the sorted (name, perFileSha) list. Encoded as null-separated
|
||||
// utf-8 to keep concatenation unambiguous if a frame name ever contains
|
||||
// an unusual character.
|
||||
const lines = entries.map(
|
||||
(name) => `${name}\0${sha256Hex(readFileSync(join(outputPath, name)))}`,
|
||||
);
|
||||
return sha256Hex(lines.join("\0"));
|
||||
}
|
||||
|
||||
/**
|
||||
* Activity B: render a single chunk of the planDir. The `outputChunkPath`
|
||||
* argument is a file for mp4/mov outputs and a directory for png-sequence
|
||||
* outputs — the caller picks the right shape based on `meta/encoder.json`.
|
||||
* `renderChunk` enforces the same choice via `outputKind` on the result.
|
||||
*/
|
||||
export async function renderChunk(
|
||||
planDir: string,
|
||||
chunkIndex: number,
|
||||
outputChunkPath: string,
|
||||
): Promise<ChunkResult> {
|
||||
const start = Date.now();
|
||||
const log = defaultLogger;
|
||||
|
||||
// ── Read + validate the plan ──
|
||||
const planJsonPath = join(planDir, "plan.json");
|
||||
const encoderJsonPath = join(planDir, "meta", "encoder.json");
|
||||
const chunksJsonPath = join(planDir, "meta", "chunks.json");
|
||||
for (const required of [planJsonPath, encoderJsonPath, chunksJsonPath]) {
|
||||
if (!existsSync(required)) {
|
||||
throw new RenderChunkValidationError(
|
||||
MISSING_PLAN_ARTIFACT,
|
||||
`[renderChunk] planDir is missing required artifact: ${required}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
const plan = JSON.parse(readFileSync(planJsonPath, "utf-8")) as PlanJson;
|
||||
const encoder = JSON.parse(readFileSync(encoderJsonPath, "utf-8")) as LockedRenderConfig;
|
||||
const chunks = JSON.parse(readFileSync(chunksJsonPath, "utf-8")) as ChunkSliceJson[];
|
||||
|
||||
if (chunkIndex < 0 || chunkIndex >= chunks.length) {
|
||||
throw new RenderChunkValidationError(
|
||||
CHUNK_INDEX_OUT_OF_RANGE,
|
||||
`[renderChunk] chunkIndex ${chunkIndex} is out of range [0, ${chunks.length})`,
|
||||
);
|
||||
}
|
||||
// The bounds check above guarantees this hits, but TS doesn't narrow
|
||||
// the indexed access — re-check explicitly.
|
||||
const slice = chunks[chunkIndex];
|
||||
if (slice === undefined) {
|
||||
throw new RenderChunkValidationError(CHUNK_INDEX_OUT_OF_RANGE, "[renderChunk] missing slice");
|
||||
}
|
||||
const framesInChunk = slice.endFrame - slice.startFrame;
|
||||
if (framesInChunk <= 0) {
|
||||
throw new RenderChunkValidationError(
|
||||
CHUNK_INDEX_OUT_OF_RANGE,
|
||||
`[renderChunk] chunk ${chunkIndex} has non-positive frame count: ${framesInChunk}`,
|
||||
);
|
||||
}
|
||||
|
||||
const compiledDir = join(planDir, "compiled");
|
||||
if (!existsSync(compiledDir)) {
|
||||
throw new RenderChunkValidationError(
|
||||
MISSING_PLAN_ARTIFACT,
|
||||
`[renderChunk] planDir missing compiled/ directory: ${compiledDir}`,
|
||||
);
|
||||
}
|
||||
|
||||
// ── Cross-version sanity ──
|
||||
const ffmpegVersion = await readFfmpegVersion();
|
||||
if (ffmpegVersion !== plan.ffmpegVersion) {
|
||||
throw new RenderChunkValidationError(
|
||||
FFMPEG_VERSION_MISMATCH,
|
||||
`[renderChunk] ffmpeg version on this worker does not match planDir. ` +
|
||||
`planDir: ${JSON.stringify(plan.ffmpegVersion)}; worker: ${JSON.stringify(ffmpegVersion)}. ` +
|
||||
`Distributed retries require byte-identical ffmpeg builds across workers. ` +
|
||||
`Re-plan from a worker matching this version, or run all renders on an image with the planDir's ffmpeg.`,
|
||||
);
|
||||
}
|
||||
if (encoder.browserGpuMode !== "software") {
|
||||
throw new RenderChunkValidationError(
|
||||
BROWSER_GPU_NOT_SOFTWARE,
|
||||
`[renderChunk] planDir requires browserGpuMode=software, got ${JSON.stringify(encoder.browserGpuMode)}.`,
|
||||
);
|
||||
}
|
||||
|
||||
// Re-derive `planHash` from the on-disk bytes and compare to the value
|
||||
// the controller wrote into `plan.json`. Catches corrupted artifacts
|
||||
// (truncated meta files, partial S3 downloads, manual tampering) before
|
||||
// the chunk renders. Distinct from the other validation paths above
|
||||
// because `MISSING_PLAN_ARTIFACT` etc. are structural; this is purely
|
||||
// content-fingerprint drift.
|
||||
const recomputedPlanHash = recomputePlanHashFromPlanDir(planDir);
|
||||
if (recomputedPlanHash !== plan.planHash) {
|
||||
throw new RenderChunkValidationError(
|
||||
PLAN_HASH_MISMATCH,
|
||||
`[renderChunk] planDir content fingerprint does not match plan.json.planHash. ` +
|
||||
`plan.json: ${plan.planHash}; recomputed: ${recomputedPlanHash}. ` +
|
||||
`Likely a corrupted artifact (partial S3 download, manual tampering) or a planDir ` +
|
||||
`produced by an incompatible producer version. Re-plan and re-fan-out.`,
|
||||
);
|
||||
}
|
||||
|
||||
// Distinct from the silent `?? {}` fallback we used before: missing
|
||||
// `runtimeEnv` means the planDir was produced by a controller that
|
||||
// forgot to snapshot, and the chunk's pixels would diverge silently.
|
||||
// Surface it as a typed validation error so the workflow can re-plan.
|
||||
if (!encoder.runtimeEnv || typeof encoder.runtimeEnv !== "object") {
|
||||
throw new RenderChunkValidationError(
|
||||
MISSING_RUNTIME_ENV_SNAPSHOT,
|
||||
"[renderChunk] planDir is missing meta/encoder.json.runtimeEnv snapshot. " +
|
||||
"Re-plan with the current producer.",
|
||||
);
|
||||
}
|
||||
|
||||
// Apply the controller's runtime-env snapshot. Must happen BEFORE the
|
||||
// file server is created — RENDER_MODE_SCRIPT bakes env vars into
|
||||
// served HTML at module load. The `restore()` handle is invoked in
|
||||
// `finally` so multi-chunk workers (Cloud Run Jobs, Temporal activity
|
||||
// worker) don't leak chunk N's env into chunk N+1.
|
||||
const envRestore = applyRuntimeEnvSnapshot(encoder.runtimeEnv);
|
||||
|
||||
try {
|
||||
// Synthesize a RenderJob the existing stages can consume. The chunk's
|
||||
// duration is its own frame count over fps — not the plan's full
|
||||
// duration — so the stages see this chunk as a self-contained render.
|
||||
const job = buildSyntheticRenderJob({
|
||||
fps: { num: plan.dimensions.fpsNum, den: plan.dimensions.fpsDen },
|
||||
quality: encoder.quality,
|
||||
format: plan.dimensions.format as "mp4" | "mov" | "png-sequence",
|
||||
crf: encoder.crf,
|
||||
bitrate: encoder.bitrate,
|
||||
hdrMode: "force-sdr",
|
||||
entryFile: "index.html",
|
||||
});
|
||||
job.totalFrames = framesInChunk;
|
||||
job.duration = (framesInChunk * plan.dimensions.fpsDen) / plan.dimensions.fpsNum;
|
||||
|
||||
const cfg: EngineConfig = {
|
||||
...resolveConfig(),
|
||||
browserGpuMode: "software",
|
||||
forceScreenshot: encoder.forceScreenshot,
|
||||
};
|
||||
|
||||
// ── Per-chunk work + frames directories ──
|
||||
// Suffix workDir with pid + random bytes so concurrent invocations on
|
||||
// the SAME `(planDir, chunkIndex)` (e.g. a scheduler that double-fires
|
||||
// due to heartbeat skew) don't race on the same tmp tree. The output
|
||||
// path itself is still the caller's contract — concurrent writers to
|
||||
// `outputChunkPath` produce undefined bytes, but we don't make it worse
|
||||
// by also deleting their workDirs out from under them.
|
||||
const workDir = `${outputChunkPath}.work.${process.pid}.${randomBytes(4).toString("hex")}`;
|
||||
mkdirSync(workDir, { recursive: true });
|
||||
const framesDir = join(workDir, "captured-frames");
|
||||
mkdirSync(framesDir, { recursive: true });
|
||||
|
||||
// ── File server with the seeded-random shim ──
|
||||
// `Math.random` / `crypto.getRandomValues` are seeded from virtual
|
||||
// time so retries are pixel-identical. Only distributed renders flip this.
|
||||
const fileServer: FileServerHandle = await createFileServer({
|
||||
projectDir: compiledDir,
|
||||
compiledDir,
|
||||
port: 0,
|
||||
preHeadScripts: [buildVirtualTimeShim({ seedRandomFromFrame: true })],
|
||||
});
|
||||
|
||||
const captureOptions: CaptureOptions = {
|
||||
width: plan.dimensions.width,
|
||||
height: plan.dimensions.height,
|
||||
fps: { num: plan.dimensions.fpsNum, den: plan.dimensions.fpsDen },
|
||||
format: plan.dimensions.format === "mp4" ? "jpeg" : "png",
|
||||
quality: plan.dimensions.format === "mp4" ? 80 : undefined,
|
||||
deviceScaleFactor: encoder.deviceScaleFactor,
|
||||
// lock the BeginFrame warmup loop to a fixed iteration count so
|
||||
// `beginFrameTimeTicks` is host-independent. Only chunks ever set this.
|
||||
lockWarmupTicks: true,
|
||||
};
|
||||
|
||||
// ── Browser + warmup ──
|
||||
let session: CaptureSession | null = null;
|
||||
let outputKind: "file" | "frame-dir";
|
||||
let framesEncoded = 0;
|
||||
try {
|
||||
session = await createCaptureSession(fileServer.url, framesDir, captureOptions, null, cfg);
|
||||
// SwiftShader assertion runs BEFORE initializeSession (which navigates to
|
||||
// the composition); on failure we tear down without ever touching the
|
||||
// composition URL. We pass `readWebGlVendorInfoFromCanvas` rather than
|
||||
// letting `assertSwiftShader` use its default `chrome://gpu` reader —
|
||||
// `chrome-headless-shell` serves chrome:// pages as empty documents,
|
||||
// which would trip a false-negative even when the GL backend is in fact
|
||||
// SwiftShader. The canvas + WEBGL_debug_renderer_info probe works on
|
||||
// any page (we navigate to about:blank inside the helper).
|
||||
await assertSwiftShader(session.page, readWebGlVendorInfoFromCanvas);
|
||||
await initializeSession(session);
|
||||
|
||||
// Prime BeginFrame's `lastFrameCache` so the chunk's first real capture
|
||||
// reports `hasDamage` the same as an in-process render at the same
|
||||
// absolute frame would. Time is the chunk's first-frame absolute time.
|
||||
const startTime = (slice.startFrame * plan.dimensions.fpsDen) / plan.dimensions.fpsNum;
|
||||
await discardWarmupCapture(session, slice.startFrame, startTime);
|
||||
|
||||
// ── Capture the chunk's range via runCaptureStage ──
|
||||
await runCaptureStage({
|
||||
fileServer,
|
||||
workDir,
|
||||
framesDir,
|
||||
job,
|
||||
totalFrames: framesInChunk,
|
||||
cfg,
|
||||
forceScreenshot: encoder.forceScreenshot,
|
||||
log,
|
||||
workerCount: 1,
|
||||
// Pass the pre-warmed session through as `probeSession` so captureStage
|
||||
// reuses it via `prepareCaptureSessionForReuse` instead of spinning up
|
||||
// a fresh browser. The stage closes the session in its `finally`,
|
||||
// so we MUST clear our own reference here to avoid a double-close.
|
||||
probeSession: session,
|
||||
needsAlpha: plan.dimensions.format !== "mp4",
|
||||
captureAttempts: [],
|
||||
buildCaptureOptions: () => captureOptions,
|
||||
createRenderVideoFrameInjector: () => null,
|
||||
abortSignal: undefined,
|
||||
assertNotAborted: () => {},
|
||||
frameRange: { startFrame: slice.startFrame, endFrame: slice.endFrame },
|
||||
});
|
||||
// captureStage closes the session it consumed.
|
||||
session = null;
|
||||
framesEncoded = framesInChunk;
|
||||
|
||||
// ── Encode the chunk ──
|
||||
const isPngSequence = plan.dimensions.format === "png-sequence";
|
||||
outputKind = isPngSequence ? "frame-dir" : "file";
|
||||
// For mp4/mov we use the standard preset machinery; the locked encoder
|
||||
// values come from `meta/encoder.json` and the `lockGopForChunkConcat`
|
||||
// toggle is the only Phase-2 flag that flips on at this site.
|
||||
// png-sequence has no encoder, but `runEncodeStage` still reads
|
||||
// `preset.quality` for bookkeeping (it never reaches ffmpeg on the
|
||||
// pngseq branch). Fall back to the mp4 preset shape — same trick
|
||||
// `renderOrchestrator` plays.
|
||||
const presetFormat: "mp4" | "mov" | "webm" = isPngSequence
|
||||
? "mp4"
|
||||
: (plan.dimensions.format as "mp4" | "mov");
|
||||
const preset = getEncoderPreset(job.config.quality, presetFormat, undefined);
|
||||
const effectiveQuality = encoder.crf ?? preset.quality;
|
||||
const effectiveBitrate = encoder.crf != null ? undefined : encoder.bitrate;
|
||||
// For non-pngseq, encodeStage writes to `outputPath` when `isPngSequence`
|
||||
// is false. `videoOnlyPath` is the encoder's direct output (no mux —
|
||||
// mux happens in assemble()).
|
||||
const videoOnlyPath = outputChunkPath;
|
||||
if (isPngSequence) {
|
||||
if (!existsSync(outputChunkPath)) mkdirSync(outputChunkPath, { recursive: true });
|
||||
} else {
|
||||
const outDir = join(outputChunkPath, "..");
|
||||
if (!existsSync(outDir)) mkdirSync(outDir, { recursive: true });
|
||||
}
|
||||
|
||||
await runEncodeStage({
|
||||
job,
|
||||
log,
|
||||
outputPath: outputChunkPath,
|
||||
framesDir,
|
||||
videoOnlyPath,
|
||||
width: plan.dimensions.width * encoder.deviceScaleFactor,
|
||||
height: plan.dimensions.height * encoder.deviceScaleFactor,
|
||||
needsAlpha: plan.dimensions.format !== "mp4",
|
||||
// Each chunk produces video only — audio is muxed once at assemble
|
||||
// time. Suppressing `hasAudio` skips the png-sequence audio sidecar
|
||||
// AND the mp4 audio mux.
|
||||
hasAudio: false,
|
||||
isPngSequence,
|
||||
preset,
|
||||
effectiveQuality,
|
||||
effectiveBitrate,
|
||||
// Distributed chunks emit a single ffmpeg call per chunk; the
|
||||
// in-process per-chunk-within-chunk path would re-split our
|
||||
// already-chunked work.
|
||||
enableChunkedEncode: false,
|
||||
chunkedEncodeSize: framesInChunk,
|
||||
abortSignal: undefined,
|
||||
assertNotAborted: () => {},
|
||||
// GOP === framesInChunk + force-keyframe at frame 0 → the chunk's
|
||||
// first frame is an IDR keyframe and concat-copy at assemble time
|
||||
// round-trips losslessly.
|
||||
lockGopForChunkConcat: !isPngSequence,
|
||||
gopSize: framesInChunk,
|
||||
});
|
||||
} finally {
|
||||
// Cleanest path: captureStage closed the session for us. The defensive
|
||||
// close handles error paths where we threw before delegating.
|
||||
if (session) {
|
||||
try {
|
||||
await closeCaptureSession(session);
|
||||
} catch (err) {
|
||||
log.warn("[renderChunk] error closing capture session in finally", {
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
}
|
||||
}
|
||||
fileServer.close();
|
||||
// Leave the temp work dir on failure (helps debugging); remove it on
|
||||
// success below.
|
||||
}
|
||||
|
||||
// ── Hash the output + write the perf sidecar ──
|
||||
const sha256 = hashChunkOutput(outputChunkPath, outputKind);
|
||||
const durationMs = Date.now() - start;
|
||||
const perfPath = `${outputChunkPath}.perf.json`;
|
||||
const perfPayload = {
|
||||
planHash: plan.planHash,
|
||||
chunkIndex,
|
||||
startFrame: slice.startFrame,
|
||||
endFrame: slice.endFrame,
|
||||
framesEncoded,
|
||||
durationMs,
|
||||
sha256,
|
||||
outputKind,
|
||||
producerVersion: plan.producerVersion,
|
||||
ffmpegVersion,
|
||||
};
|
||||
writeFileSync(perfPath, `${JSON.stringify(perfPayload, null, 2)}\n`, "utf-8");
|
||||
|
||||
// Clean up only after the hash + perf sidecar landed. Any failure above
|
||||
// leaves the framesDir in place for inspection.
|
||||
try {
|
||||
rmSync(workDir, { recursive: true, force: true });
|
||||
} catch (err) {
|
||||
log.warn("[renderChunk] failed to remove work dir", {
|
||||
workDir,
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
}
|
||||
|
||||
return {
|
||||
outputPath: outputChunkPath,
|
||||
outputKind,
|
||||
framesEncoded,
|
||||
sha256,
|
||||
durationMs,
|
||||
perfPath,
|
||||
};
|
||||
} finally {
|
||||
// Restore the controller's runtime env even on the error path so the
|
||||
// next chunk on the same process boots from a clean env.
|
||||
envRestore.restore();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
/**
|
||||
* Helpers shared between the distributed activity scripts (`plan.ts`,
|
||||
* `renderChunk.ts`, `assemble.ts`). Kept module-local so the public surface
|
||||
* stays just the three activity functions plus their result types.
|
||||
*/
|
||||
|
||||
import { execFile as execFileCallback } from "node:child_process";
|
||||
import { dirname, join } from "node:path";
|
||||
import { existsSync, readFileSync } from "node:fs";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { promisify } from "node:util";
|
||||
import { type Fps } from "@hyperframes/core";
|
||||
import { type RenderConfig, type RenderJob, createRenderJob } from "../renderOrchestrator.js";
|
||||
import { defaultLogger, type ProducerLogger } from "../../logger.js";
|
||||
|
||||
const execFile = promisify(execFileCallback);
|
||||
|
||||
/**
|
||||
* Cached first line of `ffmpeg -version` (e.g. `"ffmpeg version 6.1.1"`).
|
||||
* Cached because workers that fan out multiple `renderChunk()` calls in the
|
||||
* same process (Cloud Run Jobs, Temporal activity workers) would otherwise
|
||||
* spawn ffmpeg once per chunk just to read the version — ~20-50ms each.
|
||||
*/
|
||||
let cachedFfmpegVersion: string | null = null;
|
||||
|
||||
/**
|
||||
* Read `ffmpeg -version` first line. 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.
|
||||
*/
|
||||
export async function readFfmpegVersion(): Promise<string> {
|
||||
if (cachedFfmpegVersion !== null) return cachedFfmpegVersion;
|
||||
const { stdout } = await execFile("ffmpeg", ["-version"], { maxBuffer: 1024 * 1024 });
|
||||
const firstLine = stdout.split(/\r?\n/)[0]?.trim() ?? "";
|
||||
if (!firstLine) {
|
||||
throw new Error("ffmpeg -version returned empty output");
|
||||
}
|
||||
cachedFfmpegVersion = firstLine;
|
||||
return firstLine;
|
||||
}
|
||||
|
||||
/** Test-only: clear the cached ffmpeg version so a fresh probe runs. */
|
||||
export function _resetFfmpegVersionCacheForTests(): void {
|
||||
cachedFfmpegVersion = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Inputs for {@link buildSyntheticRenderJob}. The two distributed activity
|
||||
* scripts (`plan.ts`, `renderChunk.ts`) reach for slightly different
|
||||
* sources — caller config vs. frozen `LockedRenderConfig` — but the
|
||||
* resulting `RenderJob` shape is identical, so the helper accepts both.
|
||||
*/
|
||||
export interface SyntheticRenderJobInput {
|
||||
fps: Fps;
|
||||
format: RenderConfig["format"];
|
||||
quality: RenderConfig["quality"];
|
||||
crf?: number;
|
||||
bitrate?: string;
|
||||
outputResolution?: RenderConfig["outputResolution"];
|
||||
hdrMode: RenderConfig["hdrMode"];
|
||||
entryFile: string;
|
||||
logger?: ProducerLogger;
|
||||
producerConfig?: RenderConfig["producerConfig"];
|
||||
}
|
||||
|
||||
/**
|
||||
* Synthesize a `RenderJob` from a distributed-render config. The distributed
|
||||
* activities operate without a full `RenderJob` (they're stateless workers),
|
||||
* so we build one to feed the existing stage interfaces.
|
||||
*/
|
||||
export function buildSyntheticRenderJob(input: SyntheticRenderJobInput): RenderJob {
|
||||
const renderConfig: RenderConfig = {
|
||||
fps: input.fps,
|
||||
quality: input.quality,
|
||||
format: input.format,
|
||||
crf: input.crf,
|
||||
videoBitrate: input.bitrate,
|
||||
outputResolution: input.outputResolution,
|
||||
// Distributed mode hard-pins to software GPU. The plan-time validator
|
||||
// refuses to fan out otherwise.
|
||||
useGpu: false,
|
||||
debug: false,
|
||||
entryFile: input.entryFile,
|
||||
logger: input.logger ?? defaultLogger,
|
||||
hdrMode: input.hdrMode,
|
||||
producerConfig: input.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 (4 levels up).
|
||||
*
|
||||
* Cached at module load — the version is fixed for the life of the process,
|
||||
* and reading the package.json over and over wastes per-chunk syscalls.
|
||||
*/
|
||||
let cachedProducerVersion: string | null = null;
|
||||
export function readProducerVersion(): string {
|
||||
if (cachedProducerVersion !== null) return cachedProducerVersion;
|
||||
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") {
|
||||
cachedProducerVersion = pkg.version;
|
||||
return pkg.version;
|
||||
}
|
||||
} catch {
|
||||
// Fall through to the next ancestor.
|
||||
}
|
||||
}
|
||||
const parent = dirname(current);
|
||||
if (parent === current) break;
|
||||
current = parent;
|
||||
}
|
||||
cachedProducerVersion = "0.0.0-unknown";
|
||||
return cachedProducerVersion;
|
||||
}
|
||||
@@ -4,7 +4,11 @@
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from "bun:test";
|
||||
import { RUNTIME_ENV_PREFIXES, snapshotRuntimeEnv } from "./runtimeEnvSnapshot.js";
|
||||
import {
|
||||
applyRuntimeEnvSnapshot,
|
||||
RUNTIME_ENV_PREFIXES,
|
||||
snapshotRuntimeEnv,
|
||||
} from "./runtimeEnvSnapshot.js";
|
||||
|
||||
describe("snapshotRuntimeEnv", () => {
|
||||
it("captures PRODUCER_RUNTIME_* keys", () => {
|
||||
@@ -99,3 +103,57 @@ describe("snapshotRuntimeEnv", () => {
|
||||
expect(RUNTIME_ENV_PREFIXES).toEqual(["PRODUCER_RUNTIME_", "PRODUCER_RENDER_"]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("applyRuntimeEnvSnapshot", () => {
|
||||
it("filters by RUNTIME_ENV_PREFIXES — refuses arbitrary keys", () => {
|
||||
const env: Record<string, string | undefined> = {};
|
||||
applyRuntimeEnvSnapshot(
|
||||
{
|
||||
PRODUCER_RUNTIME_OK: "ok",
|
||||
// A poisoned planDir could include `PATH` or `LD_PRELOAD`;
|
||||
// the apply-side filter must drop them.
|
||||
PATH: "/evil/bin",
|
||||
ARBITRARY_KEY: "x",
|
||||
},
|
||||
env,
|
||||
);
|
||||
expect(env.PRODUCER_RUNTIME_OK).toBe("ok");
|
||||
expect(env.PATH).toBeUndefined();
|
||||
expect(env.ARBITRARY_KEY).toBeUndefined();
|
||||
});
|
||||
|
||||
it("restore() reverts touched keys and only those keys", () => {
|
||||
const env: Record<string, string | undefined> = {
|
||||
PRODUCER_RUNTIME_PREEXISTING: "host-value",
|
||||
PRODUCER_RENDER_NEW: undefined,
|
||||
HOST_UNRELATED: "stays",
|
||||
};
|
||||
const { restore } = applyRuntimeEnvSnapshot(
|
||||
{
|
||||
PRODUCER_RUNTIME_PREEXISTING: "snapshot-value",
|
||||
PRODUCER_RENDER_NEW: "snapshot-new",
|
||||
},
|
||||
env,
|
||||
);
|
||||
expect(env.PRODUCER_RUNTIME_PREEXISTING).toBe("snapshot-value");
|
||||
expect(env.PRODUCER_RENDER_NEW).toBe("snapshot-new");
|
||||
|
||||
restore();
|
||||
// Pre-existing key returns to its host value; previously-undefined
|
||||
// key is deleted (not left as the snapshot value).
|
||||
expect(env.PRODUCER_RUNTIME_PREEXISTING).toBe("host-value");
|
||||
expect(env.PRODUCER_RENDER_NEW).toBeUndefined();
|
||||
// Unrelated keys are untouched at apply time AND at restore time.
|
||||
expect(env.HOST_UNRELATED).toBe("stays");
|
||||
});
|
||||
|
||||
it("restore() is safe to call multiple times", () => {
|
||||
const env: Record<string, string | undefined> = {};
|
||||
const { restore } = applyRuntimeEnvSnapshot({ PRODUCER_RUNTIME_X: "y" }, env);
|
||||
restore();
|
||||
expect(env.PRODUCER_RUNTIME_X).toBeUndefined();
|
||||
// Second restore is a no-op (no entries to revert).
|
||||
expect(() => restore()).not.toThrow();
|
||||
expect(env.PRODUCER_RUNTIME_X).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -51,3 +51,43 @@ export function snapshotRuntimeEnv(
|
||||
}
|
||||
return snapshot;
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply a `runtimeEnv` snapshot to `process.env` (or another env-like
|
||||
* record) before the file server starts. Filters by
|
||||
* {@link RUNTIME_ENV_PREFIXES} so a planDir with extra keys can't smuggle
|
||||
* arbitrary env vars onto the worker — the controller's
|
||||
* {@link snapshotRuntimeEnv} already filtered, but apply-side filtering
|
||||
* is mandatory defense-in-depth against a hand-crafted or corrupted plan.
|
||||
*
|
||||
* Returns a `restore()` function that reverts `env` to its pre-apply
|
||||
* state for the keys this call touched. Callers that run multiple chunks
|
||||
* in a single process (Cloud Run Job, Temporal activity worker) MUST
|
||||
* invoke `restore()` in a `finally` block — without it, chunk N's
|
||||
* snapshot leaks into chunk N+1's environment.
|
||||
*
|
||||
* Existing snapshot keys are overwritten. Keys NOT in the snapshot are
|
||||
* never touched — the worker's host may set additional runtime knobs
|
||||
* (`HYPERFRAMES_EXTRACT_CACHE_DIR`, etc.).
|
||||
*/
|
||||
export function applyRuntimeEnvSnapshot(
|
||||
snapshot: Record<string, string>,
|
||||
env: Record<string, string | undefined> = process.env,
|
||||
): { restore: () => void } {
|
||||
const previous: Record<string, string | undefined> = {};
|
||||
for (const [key, value] of Object.entries(snapshot)) {
|
||||
if (typeof value !== "string") continue;
|
||||
const matches = RUNTIME_ENV_PREFIXES.some((prefix) => key.startsWith(prefix));
|
||||
if (!matches) continue;
|
||||
previous[key] = env[key];
|
||||
env[key] = value;
|
||||
}
|
||||
return {
|
||||
restore: () => {
|
||||
for (const [key, value] of Object.entries(previous)) {
|
||||
if (value === undefined) delete env[key];
|
||||
else env[key] = value;
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -94,6 +94,27 @@ export interface CaptureStageInput {
|
||||
abortSignal: AbortSignal | undefined;
|
||||
assertNotAborted: () => void;
|
||||
onProgress?: ProgressCallback;
|
||||
/**
|
||||
* Capture a sub-range `[startFrame, endFrame)` of the composition's
|
||||
* timeline. Used by distributed `renderChunk` workers to render only
|
||||
* their assigned chunk. Captured frames are written with file names
|
||||
* normalized to start at zero (`frame_000000.{ext}`) so the encoder
|
||||
* doesn't need an `-start_number` override; per-frame TIMES still
|
||||
* reflect the absolute frame index via `(absIdx * fps.den) / fps.num`,
|
||||
* keeping the page's virtual clock identical to what an in-process
|
||||
* render at that frame would see.
|
||||
*
|
||||
* Only honored on the sequential capture branch (workerCount === 1).
|
||||
* The parallel branch in this stage targets in-process renders where
|
||||
* adaptive retry across the whole timeline is the contract, and chunk
|
||||
* workers fan out at the activity layer instead. Passing `frameRange`
|
||||
* with `workerCount > 1` throws — the caller should reduce
|
||||
* `workerCount` to 1.
|
||||
*
|
||||
* Default `undefined`: the stage captures `[0, totalFrames)` (the
|
||||
* in-process contract).
|
||||
*/
|
||||
frameRange?: { startFrame: number; endFrame: number };
|
||||
}
|
||||
|
||||
export interface CaptureStageResult {
|
||||
@@ -122,6 +143,7 @@ export async function runCaptureStage(input: CaptureStageInput): Promise<Capture
|
||||
assertNotAborted,
|
||||
onProgress,
|
||||
needsAlpha,
|
||||
frameRange,
|
||||
} = input;
|
||||
let { workerCount, probeSession } = input;
|
||||
let lastBrowserConsole: string[] = [];
|
||||
@@ -133,6 +155,26 @@ export async function runCaptureStage(input: CaptureStageInput): Promise<Capture
|
||||
const captureCfg: EngineConfig =
|
||||
cfg.forceScreenshot === forceScreenshot ? cfg : { ...cfg, forceScreenshot };
|
||||
|
||||
if (frameRange !== undefined && workerCount > 1) {
|
||||
throw new Error(
|
||||
`[captureStage] frameRange capture requires workerCount === 1 (received workerCount=${workerCount}). ` +
|
||||
`Distributed chunk workers fan out at the activity layer; reduce workerCount to 1 when passing frameRange.`,
|
||||
);
|
||||
}
|
||||
if (frameRange !== undefined) {
|
||||
if (
|
||||
!Number.isFinite(frameRange.startFrame) ||
|
||||
!Number.isFinite(frameRange.endFrame) ||
|
||||
frameRange.startFrame < 0 ||
|
||||
frameRange.endFrame <= frameRange.startFrame
|
||||
) {
|
||||
throw new Error(
|
||||
`[captureStage] invalid frameRange: ${JSON.stringify(frameRange)}. ` +
|
||||
`Expected non-negative startFrame strictly less than endFrame.`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if (workerCount > 1) {
|
||||
// Parallel capture
|
||||
const attempts = await executeDiskCaptureWithAdaptiveRetry({
|
||||
@@ -202,19 +244,29 @@ export async function runCaptureStage(input: CaptureStageInput): Promise<Capture
|
||||
assertNotAborted();
|
||||
lastBrowserConsole = session.browserConsoleBuffer;
|
||||
|
||||
for (let i = 0; i < totalFrames; i++) {
|
||||
// `frameRange` captures only a sub-range of the timeline. Per-frame
|
||||
// TIMES still use the absolute composition frame index so the page's
|
||||
// virtual clock matches an in-process render at the same frame;
|
||||
// file NAMES are normalized to 0 (via the relative loop index `i`)
|
||||
// so the encoder can read frames without an `-start_number` override.
|
||||
const rangeStart = frameRange?.startFrame ?? 0;
|
||||
const rangeEnd = frameRange?.endFrame ?? totalFrames;
|
||||
const rangeFrames = rangeEnd - rangeStart;
|
||||
|
||||
for (let i = 0; i < rangeFrames; i++) {
|
||||
assertNotAborted();
|
||||
const time = (i * job.config.fps.den) / job.config.fps.num;
|
||||
const absoluteIdx = rangeStart + i;
|
||||
const time = (absoluteIdx * job.config.fps.den) / job.config.fps.num;
|
||||
await captureFrame(session, i, time);
|
||||
job.framesRendered = i + 1;
|
||||
|
||||
const frameProgress = (i + 1) / totalFrames;
|
||||
const frameProgress = (i + 1) / rangeFrames;
|
||||
const progress = 25 + frameProgress * 45;
|
||||
|
||||
updateJobStatus(
|
||||
job,
|
||||
"rendering",
|
||||
`Capturing frame ${i + 1}/${totalFrames}`,
|
||||
`Capturing frame ${i + 1}/${rangeFrames}`,
|
||||
Math.round(progress),
|
||||
onProgress,
|
||||
);
|
||||
|
||||
@@ -53,8 +53,13 @@ export interface EncodeStageInput {
|
||||
needsAlpha: boolean;
|
||||
/** True iff the composition has audio. Drives the sidecar copy. */
|
||||
hasAudio: boolean;
|
||||
/** Path to the mixed audio (only read when `hasAudio` is true). */
|
||||
audioOutputPath: string;
|
||||
/**
|
||||
* Path to the mixed audio. Required when `hasAudio` is `true` (the
|
||||
* png-sequence sidecar copy reads it); ignored when `hasAudio` is
|
||||
* `false`. Distributed chunk workers mux audio once at assemble time
|
||||
* and pass `hasAudio: false` here, so the field is left optional.
|
||||
*/
|
||||
audioOutputPath?: string;
|
||||
/** Mp4 vs png-sequence vs … gates the entire stage branch. */
|
||||
isPngSequence: boolean;
|
||||
/** Encoder preset (codec, preset, pixelFormat, hdr). Only used on the non-png path. */
|
||||
@@ -67,6 +72,16 @@ export interface EncodeStageInput {
|
||||
abortSignal: AbortSignal | undefined;
|
||||
assertNotAborted: () => void;
|
||||
onProgress?: ProgressCallback;
|
||||
/**
|
||||
* Pass-through of `EncoderOptions.lockGopForChunkConcat`. When `true`,
|
||||
* the encode emits closed-GOP keyframes at every `gopSize` boundary so
|
||||
* downstream `ffmpeg -f concat -c copy` round-trips losslessly. Only the
|
||||
* distributed chunk worker (`renderChunk`) sets this — the in-process
|
||||
* renderer's call site omits it, preserving the existing open-GOP output.
|
||||
*/
|
||||
lockGopForChunkConcat?: boolean;
|
||||
/** Required when `lockGopForChunkConcat === true`. Number of frames per GOP — set to the chunk's frame count by `renderChunk`. */
|
||||
gopSize?: number;
|
||||
}
|
||||
|
||||
export interface EncodeStageResult {
|
||||
@@ -119,7 +134,7 @@ export async function runEncodeStage(input: EncodeStageInput): Promise<EncodeSta
|
||||
const dst = join(outputPath, `frame_${String(i + 1).padStart(6, "0")}.png`);
|
||||
copyFileSync(join(framesDir, name), dst);
|
||||
});
|
||||
if (hasAudio && existsSync(audioOutputPath)) {
|
||||
if (hasAudio && audioOutputPath && existsSync(audioOutputPath)) {
|
||||
// Sidecar audio for callers that need to re-mux later. png-sequence
|
||||
// has no container of its own, so this is the only place audio
|
||||
// can land alongside the frames.
|
||||
@@ -145,6 +160,11 @@ export async function runEncodeStage(input: EncodeStageInput): Promise<EncodeSta
|
||||
pixelFormat: preset.pixelFormat,
|
||||
useGpu: job.config.useGpu,
|
||||
hdr: preset.hdr,
|
||||
// Distributed chunk renders pass these so the encoder writes closed-GOP
|
||||
// keyframes that survive `-f concat -c copy` at assemble time. In-process
|
||||
// renders leave both undefined → preserves the existing open-GOP output.
|
||||
lockGopForChunkConcat: input.lockGopForChunkConcat === true,
|
||||
gopSize: input.gopSize,
|
||||
};
|
||||
const encodeResult = enableChunkedEncode
|
||||
? await encodeFramesChunkedConcat(
|
||||
|
||||
@@ -212,6 +212,58 @@ function collectPlanAssetShas(planDir: string): {
|
||||
return { compositionHtml, assets };
|
||||
}
|
||||
|
||||
/**
|
||||
* Read a frozen plan directory back from disk and recompute its
|
||||
* content-addressed `planHash` over the actual on-disk bytes — including
|
||||
* the canonical encoder JSON, which is written via
|
||||
* {@link canonicalJsonStringify} so reading the file gives us the exact
|
||||
* string that fed the controller's hash.
|
||||
*
|
||||
* Distributed chunk workers call this at boot to verify their planDir is
|
||||
* the same one the controller wrote: any mismatch (corrupted artifact,
|
||||
* partial S3 download, manual tampering) trips a non-retryable
|
||||
* `PLAN_HASH_MISMATCH` before the chunk renders.
|
||||
*
|
||||
* Throws if `plan.json` or `meta/encoder.json` are missing/malformed —
|
||||
* callers should catch those as `MISSING_PLAN_ARTIFACT` rather than
|
||||
* lumping them with hash drift.
|
||||
*/
|
||||
export function recomputePlanHashFromPlanDir(planDir: string): string {
|
||||
const planJsonPath = join(planDir, "plan.json");
|
||||
const encoderJsonPath = join(planDir, "meta", "encoder.json");
|
||||
if (!existsSync(planJsonPath)) {
|
||||
throw new Error(`[freezePlan] plan.json missing: ${planJsonPath}`);
|
||||
}
|
||||
if (!existsSync(encoderJsonPath)) {
|
||||
throw new Error(`[freezePlan] meta/encoder.json missing: ${encoderJsonPath}`);
|
||||
}
|
||||
|
||||
const planJson = JSON.parse(readFileSync(planJsonPath, "utf-8")) as {
|
||||
producerVersion: string;
|
||||
ffmpegVersion: string;
|
||||
fontSnapshotSha: string;
|
||||
dimensions: PlanDimensions;
|
||||
};
|
||||
|
||||
// Encoder JSON is consumed as raw bytes so the hashing input matches
|
||||
// the on-disk file byte-for-byte. Re-parsing + re-canonicalizing would
|
||||
// be susceptible to floating-point round-trip drift, JSON whitespace
|
||||
// normalization differences, and Node version skew on number printing.
|
||||
const encoderConfigCanonicalJson = readFileSync(encoderJsonPath, "utf-8");
|
||||
|
||||
const { compositionHtml, assets } = collectPlanAssetShas(planDir);
|
||||
|
||||
return computePlanHash({
|
||||
compositionHtml,
|
||||
assets,
|
||||
fontSnapshotSha: planJson.fontSnapshotSha,
|
||||
encoderConfigCanonicalJson,
|
||||
producerVersion: planJson.producerVersion,
|
||||
ffmpegVersion: planJson.ffmpegVersion,
|
||||
dimensions: planJson.dimensions,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Freeze a plan directory: write `meta/*.json` + top-level `plan.json`, then
|
||||
* compute `planHash` over the canonicalized contents.
|
||||
|
||||
Reference in New Issue
Block a user