/** * 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)`. * - No `lastFrameCache` priming: every frame seeks fresh DOM so the * cache is never read, and priming would deadlock the compositor. * - 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 { extname, join } from "node:path"; import { assertSwiftShader, type BeforeCaptureHook, BROWSER_GPU_NOT_SOFTWARE, calculateOptimalWorkers, type CaptureOptions, type CaptureSession, closeCaptureSession, createCaptureSession, createFrameLookupTable, createVideoFrameInjector, type EngineConfig, type ExtractedFrames, getEncoderPreset, initializeSession, readWebGlVendorInfoFromCanvas, 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, closeFileServerSafely, createFileServer, type FileServerHandle, } from "../fileServer.js"; import { buildSyntheticRenderJob, type DistributedFormat, PLAN_VIDEOS_META_RELATIVE_PATH, type PlanVideosJson, 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"; const LEGACY_DISTRIBUTED_VP9_CPU_USED = 2; 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; /** * Stage wall-clock split of `durationMs`, for separating per-chunk fixed * overhead from frame-proportional work in fleet cost models: * * - `planHashMs` — full planDir content-hash recomputation (validation). * - `sessionBootMs` — sequential-branch Chrome boot + SwiftShader assert + * composition warmup. Stays 0 when `workers > 1` (each parallel worker * boots inside the capture stage instead). * - `captureStageMs` — the capture stage call; includes per-worker session * boots in the parallel branch. * - `encodeStageMs` — the encode stage call (single ffmpeg invocation, or * the frame-dir arrangement for png-sequence). * * The remainder of `durationMs` is validation + file-server setup + output * hashing + cleanup. */ planHashMs: number; sessionBootMs: number; captureStageMs: number; encodeStageMs: number; /** Capture workers used for this chunk (`calculateOptimalWorkers` result). */ workers: 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; } /** * Rebuild the engine's in-memory `ExtractedFrames[]` from the on-disk * planDir layout. `/video-frames//` holds the numbered * frame files plan() extracted; this lists each dir and rebuilds the * 1-based `framePaths` Map that `FrameLookupTable` / `videoFrameInjector` * both index against. */ function rebuildExtractedFramesFromPlanDir( planDir: string, videos: PlanVideosJson["extracted"], ): ExtractedFrames[] { const result: ExtractedFrames[] = []; for (const v of videos) { const outputDir = join(planDir, "video-frames", v.videoId); if (!existsSync(outputDir)) { throw new Error( `[renderChunk] planDir missing extracted video frames for ${JSON.stringify(v.videoId)}: ` + `${outputDir} not present. plan() should have written frames here; the planDir is malformed.`, ); } // framePattern looks like `frame_%05d.jpg`; sprintf isn't available at // runtime so list-and-sort the directory. Sorted-by-name matches // sorted-by-frame-index because the extractor writes zero-padded // monotonic indices. const ext = (extname(v.framePattern) || ".jpg").toLowerCase(); const frames = readdirSync(outputDir) .filter((name) => name.toLowerCase().endsWith(ext)) .sort(); const framePaths = new Map(); for (let i = 0; i < frames.length; i++) { const frameName = frames[i]; if (!frameName) continue; // FrameLookupTable indexes frames 1-based. framePaths.set(i + 1, join(outputDir, frameName)); } result.push({ videoId: v.videoId, srcPath: v.srcPath, outputDir, framePattern: v.framePattern, fps: v.fps, totalFrames: v.totalFrames, metadata: v.metadata, framePaths, // The chunk worker doesn't own the planDir's video-frames/ directory // (the controller does — adapters that fan out chunks across machines // share the planDir as read-only). Mark ownership as false so the // injector's eventual cleanup doesn't rm bytes another worker may // still be reading. ownedByLookup: false, }); } return result; } /** 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: DistributedFormat; }; 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"; // `readWebGlVendorInfoFromCanvas` lives in `@hyperframes/engine` (it's // used both here and by `parallelCoordinator.executeWorkerTask`). Re-exported // from this subpath so downstream consumers that already import it from // `@hyperframes/producer/distributed` keep working. export { readWebGlVendorInfoFromCanvas } from "@hyperframes/engine"; /** * 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")); } /** * Apply the planDir's locked-encoder choice on top of an * `EncoderPreset` from `getEncoderPreset`. `getEncoderPreset` returns * h265 only on the HDR branch, but distributed mode is SDR-only — for * an `libx265-software` planDir we still need to flip the preset's * codec to h265 so `runEncodeStage` invokes libx265. Exported so a * unit test can pin the override independently of the heavyweight * Docker fixture: a refactor that moves the override (e.g. into * `getEncoderPreset` itself) shouldn't be able to silently regress * the contract without a fast-test signal. */ export function resolvePresetForLockedEncoder< P extends { codec: "h264" | "h265" | "vp9" | "prores" }, >(basePreset: P, lockedEncoder: LockedRenderConfig["encoder"]): P { if (lockedEncoder === "libx265-software") { return { ...basePreset, codec: "h265" as const }; } return basePreset; } export function resolveLockedVp9CpuUsed( lockedEncoder: Pick, ): number | undefined { if (lockedEncoder.encoder !== "libvpx-vp9-software") return undefined; // Pre-vp9CpuUsed WebM planDirs used the old closed-GOP literal. Keep replay // bytes stable for those plans while new planDirs carry their resolved value. return lockedEncoder.vp9CpuUsed ?? LEGACY_DISTRIBUTED_VP9_CPU_USED; } /** * 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 { 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[]; // `meta/videos.json` only exists when the composition has `