diff --git a/packages/producer/src/services/distributed/plan.ts b/packages/producer/src/services/distributed/plan.ts index 8f11d7bd55..bed93e3141 100644 --- a/packages/producer/src/services/distributed/plan.ts +++ b/packages/producer/src/services/distributed/plan.ts @@ -24,16 +24,7 @@ * never have to handle them. */ -import { - cpSync, - existsSync, - mkdirSync, - readdirSync, - renameSync, - rmSync, - statSync, - writeFileSync, -} from "node:fs"; +import { cpSync, existsSync, mkdirSync, renameSync, rmSync, writeFileSync } from "node:fs"; import { join, relative, sep } from "node:path"; import { type CanvasResolution, fpsToNumber } from "@hyperframes/core"; import { @@ -79,6 +70,11 @@ import { readFfmpegVersion, readProducerVersion, } from "./shared.js"; +import { + measurePlanSizeBreakdown, + type PlanSizeBreakdown, + type PlanSizeRootKind, +} from "./planSize.js"; /** * Caller-supplied configuration for a distributed render. `fps`, `width`, @@ -341,17 +337,36 @@ export class PlanTooLargeError extends Error { readonly code: typeof PLAN_TOO_LARGE = PLAN_TOO_LARGE; readonly sizeBytes: number; readonly limitBytes: number; - constructor(sizeBytes: number, limitBytes: number) { + readonly breakdown?: PlanSizeBreakdown; + readonly observedAt?: string; + constructor( + sizeBytes: number, + limitBytes: number, + breakdown?: PlanSizeBreakdown, + observedAt?: string, + ) { + const breakdownSuffix = breakdown + ? ` Breakdown: video-frames=${formatBytes(breakdown.videoFramesBytes)}, ` + + `compiled=${formatBytes(breakdown.compiledBytes)} ` + + `(source-media=${formatBytes(breakdown.sourceMediaBytes)}), ` + + `audio=${formatBytes(breakdown.audioBytes)}, metadata=${formatBytes(breakdown.metadataBytes)}, ` + + `other=${formatBytes(breakdown.otherBytes)}, files=${breakdown.fileCount}.` + : ""; + const observationSuffix = observedAt ? ` Observed at ${observedAt}.` : ""; super( `[plan] planDir size ${formatBytes(sizeBytes)} exceeds the configured ceiling ` + `${formatBytes(limitBytes)} (PLAN_TOO_LARGE). The default 2 GB cap fits inside AWS ` + `Lambda's 10 GB /tmp budget alongside the chunk worker's frame buffer and ffmpeg's ` + `working set. To unblock: shorten the composition, lower the framerate, or use the ` + - `in-process renderer (\`executeRenderJob\`) — it has no planDir size cap.`, + `in-process renderer (\`executeRenderJob\`) — it has no planDir size cap.` + + observationSuffix + + breakdownSuffix, ); this.name = "PlanTooLargeError"; this.sizeBytes = sizeBytes; this.limitBytes = limitBytes; + this.breakdown = breakdown; + this.observedAt = observedAt; } } @@ -434,30 +449,33 @@ export function rejectUnsupportedDistributedFormat( * walker outside the planDir. */ export function measurePlanDirBytes(planDir: string): number { - let total = 0; - function walk(dir: string): void { - let entries; - try { - entries = readdirSync(dir, { withFileTypes: true }); - } catch { - return; - } - for (const entry of entries) { - const full = join(dir, entry.name); - if (entry.isDirectory()) { - walk(full); - } else if (entry.isFile()) { - try { - total += statSync(full).size; - } catch { - // Ignore — a file disappearing during the walk shouldn't crash - // the measurement. - } - } - } - } - walk(planDir); - return total; + return measurePlanSizeBreakdown(planDir).totalBytes; +} + +function assertPlanSizeWithinLimit(input: { + rootDir: string; + rootKind: PlanSizeRootKind; + limitBytes: number; + observedAt: string; + log: ProducerLogger; +}): PlanSizeBreakdown { + const breakdown = measurePlanSizeBreakdown(input.rootDir, input.rootKind); + if (breakdown.totalBytes <= input.limitBytes) return breakdown; + input.log.warn("[plan] size budget exceeded", { + observedAt: input.observedAt, + sizeBytes: breakdown.totalBytes, + limitBytes: input.limitBytes, + fileCount: breakdown.fileCount, + compiledBytes: breakdown.compiledBytes, + sourceMediaBytes: breakdown.sourceMediaBytes, + videoFramesBytes: breakdown.videoFramesBytes, + videoFrameFileCount: breakdown.videoFrameFileCount, + audioBytes: breakdown.audioBytes, + metadataBytes: breakdown.metadataBytes, + otherBytes: breakdown.otherBytes, + topComponents: breakdown.topComponents, + }); + throw new PlanTooLargeError(breakdown.totalBytes, input.limitBytes, breakdown, input.observedAt); } /** @@ -546,6 +564,19 @@ function assertPositiveInteger(name: string, value: number): void { } } +const FREEZE_OWNED_PLAN_FILES = [ + "plan.json", + join("meta", "composition.json"), + join("meta", "encoder.json"), + join("meta", "chunks.json"), +] as const; + +function removeFreezeOwnedPlanFiles(planDir: string): void { + for (const relativePath of FREEZE_OWNED_PLAN_FILES) { + rmSync(join(planDir, relativePath), { force: true }); + } +} + /** * Slice `totalFrames` into `chunkCount` consecutive ranges. Each chunk gets * `effectiveChunkSize` frames except the last, which absorbs the remainder @@ -750,6 +781,7 @@ export async function plan( if (!existsSync(planDir)) mkdirSync(planDir, { recursive: true }); const log = config.logger ?? defaultLogger; + const sizeLimitBytes = config.planDirSizeLimitBytes ?? PLAN_DIR_SIZE_LIMIT_BYTES; const abortSignal = config.abortSignal; const assertNotAborted = (): void => { if (abortSignal?.aborted) { @@ -915,6 +947,17 @@ export async function plan( } } + // The compiled tree is now stable. If it already exceeds the final plan + // budget, extraction/audio/freeze can only add retained bytes, so fail before + // generating a multi-GiB frame tree. + assertPlanSizeWithinLimit({ + rootDir: compiledDir, + rootKind: "compiled", + limitBytes: sizeLimitBytes, + observedAt: "pre-extract", + log, + }); + // ── Extract videos ── // `materializeSymlinks: true` recursively copies frames so the planDir is // self-contained (symlinks don't survive S3/GCS round-trips). @@ -1058,6 +1101,21 @@ export async function plan( }); } + // A caller may reuse an existing planDir. freezePlan overwrites these four + // files, so remove stale versions before the preliminary measurement; the + // exact post-freeze check below still includes the newly written metadata. + removeFreezeOwnedPlanFiles(planDir); + + // All retained heavy artifacts have been promoted and transient work files + // removed. Reject before freezePlan performs its full content-hash read. + assertPlanSizeWithinLimit({ + rootDir: planDir, + rootKind: "plan", + limitBytes: sizeLimitBytes, + observedAt: "pre-freeze", + log, + }); + const freezeResult = await freezePlan({ planDir, composition: compositionJson, @@ -1076,11 +1134,13 @@ export async function plan( // alongside the chunk worker's frame buffer + ffmpeg working set. The // check runs AFTER cleanup so the workDir tree doesn't double-count. // Non-retryable: the same planDir would trip the cap on every retry. - const sizeLimitBytes = config.planDirSizeLimitBytes ?? PLAN_DIR_SIZE_LIMIT_BYTES; - const planDirBytes = measurePlanDirBytes(planDir); - if (planDirBytes > sizeLimitBytes) { - throw new PlanTooLargeError(planDirBytes, sizeLimitBytes); - } + assertPlanSizeWithinLimit({ + rootDir: planDir, + rootKind: "plan", + limitBytes: sizeLimitBytes, + observedAt: "post-freeze", + log, + }); return { planDir, diff --git a/packages/producer/src/services/distributed/planSize.ts b/packages/producer/src/services/distributed/planSize.ts new file mode 100644 index 0000000000..c798745077 --- /dev/null +++ b/packages/producer/src/services/distributed/planSize.ts @@ -0,0 +1,196 @@ +import { createHash } from "node:crypto"; +import { lstatSync, readdirSync } from "node:fs"; +import { extname, join, relative } from "node:path"; + +const MAX_TOP_COMPONENTS = 10; +const SOURCE_MEDIA_EXTENSIONS = new Set([ + ".aac", + ".aiff", + ".avi", + ".flac", + ".m4a", + ".m4v", + ".mkv", + ".mov", + ".mp3", + ".mp4", + ".mpeg", + ".mpg", + ".ogg", + ".ogv", + ".ts", + ".wav", + ".webm", +]); + +export type PlanSizeRootKind = "compiled" | "plan"; + +export interface PlanSizeComponent { + /** Fixed category or a one-way hash for a video-frame directory; never a source path. */ + label: string; + sizeBytes: number; + fileCount: number; +} + +export interface PlanSizeBreakdown { + totalBytes: number; + fileCount: number; + compiledBytes: number; + /** Attribution within compiledBytes, not an additional mutually-exclusive bucket. */ + sourceMediaBytes: number; + videoFramesBytes: number; + videoFrameFileCount: number; + audioBytes: number; + metadataBytes: number; + otherBytes: number; + /** Largest fixed/hash-only buckets, deterministically ordered and bounded. */ + topComponents: PlanSizeComponent[]; +} + +type PlanSizeCategory = "audio" | "compiled" | "metadata" | "other" | "video-frames"; + +interface ClassifiedFile { + category: PlanSizeCategory; + componentLabel: string; +} + +const PLAN_ROOT_CATEGORY: Readonly> = { + compiled: "compiled", + "audio.aac": "audio", + meta: "metadata", + "plan.json": "metadata", +}; + +function hashComponent(value: string): string { + return createHash("sha256").update(value).digest("hex").slice(0, 12); +} + +function classifyFile(segments: string[], rootKind: PlanSizeRootKind): ClassifiedFile { + const rootName = segments[0] ?? ""; + const extractedRootName = rootKind === "compiled" ? "__hyperframes_video_frames" : "video-frames"; + if (rootName === extractedRootName) { + const videoKey = segments[1] ?? "unknown"; + return { + category: "video-frames", + componentLabel: `video-frames:${hashComponent(videoKey)}`, + }; + } + + const category = rootKind === "compiled" ? "compiled" : (PLAN_ROOT_CATEGORY[rootName] ?? "other"); + return { category, componentLabel: category }; +} + +function emptyBreakdown(): PlanSizeBreakdown { + return { + totalBytes: 0, + fileCount: 0, + compiledBytes: 0, + sourceMediaBytes: 0, + videoFramesBytes: 0, + videoFrameFileCount: 0, + audioBytes: 0, + metadataBytes: 0, + otherBytes: 0, + topComponents: [], + }; +} + +function addCategoryBytes( + result: PlanSizeBreakdown, + category: PlanSizeCategory, + filePath: string, + sizeBytes: number, +): void { + if (category === "compiled") { + result.compiledBytes += sizeBytes; + if (SOURCE_MEDIA_EXTENSIONS.has(extname(filePath).toLowerCase())) { + result.sourceMediaBytes += sizeBytes; + } + return; + } + if (category === "video-frames") { + result.videoFramesBytes += sizeBytes; + result.videoFrameFileCount += 1; + return; + } + if (category === "audio") { + result.audioBytes += sizeBytes; + return; + } + if (category === "metadata") { + result.metadataBytes += sizeBytes; + return; + } + result.otherBytes += sizeBytes; +} + +function recordFile( + result: PlanSizeBreakdown, + components: Map, + rootDir: string, + rootKind: PlanSizeRootKind, + filePath: string, +): void { + let sizeBytes: number; + try { + const stat = lstatSync(filePath); + if (!stat.isFile()) return; + sizeBytes = stat.size; + } catch { + return; + } + const rel = relative(rootDir, filePath); + const segments = rel.split(/[\\/]+/).filter(Boolean); + const classified = classifyFile(segments, rootKind); + + result.totalBytes += sizeBytes; + result.fileCount += 1; + addCategoryBytes(result, classified.category, filePath, sizeBytes); + + const component = components.get(classified.componentLabel) ?? { + label: classified.componentLabel, + sizeBytes: 0, + fileCount: 0, + }; + component.sizeBytes += sizeBytes; + component.fileCount += 1; + components.set(classified.componentLabel, component); +} + +function walkRegularFiles(dir: string, visit: (filePath: string) => void): void { + let entries; + try { + entries = readdirSync(dir, { withFileTypes: true }); + } catch { + return; + } + for (const entry of entries) { + const full = join(dir, entry.name); + if (entry.isDirectory()) { + walkRegularFiles(full, visit); + } else if (entry.isFile()) { + visit(full); + } + } +} + +/** + * Measure a staged compiled tree or finalized plan tree without following + * symlinks. Component labels are fixed or hashed, so callers can safely put + * the bounded result in logs/errors without exposing authored paths. + */ +export function measurePlanSizeBreakdown( + rootDir: string, + rootKind: PlanSizeRootKind = "plan", +): PlanSizeBreakdown { + const result = emptyBreakdown(); + const components = new Map(); + + walkRegularFiles(rootDir, (filePath) => + recordFile(result, components, rootDir, rootKind, filePath), + ); + result.topComponents = [...components.values()] + .sort((a, b) => b.sizeBytes - a.sizeBytes || a.label.localeCompare(b.label)) + .slice(0, MAX_TOP_COMPONENTS); + return result; +} diff --git a/packages/producer/src/services/distributed/planSizeCap.test.ts b/packages/producer/src/services/distributed/planSizeCap.test.ts index 7492407471..0de68e67ef 100644 --- a/packages/producer/src/services/distributed/planSizeCap.test.ts +++ b/packages/producer/src/services/distributed/planSizeCap.test.ts @@ -15,7 +15,7 @@ */ import { afterAll, beforeAll, describe, expect, it } from "bun:test"; -import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { existsSync, mkdirSync, mkdtempSync, rmSync, symlinkSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { @@ -25,6 +25,7 @@ import { PlanTooLargeError, plan, } from "./plan.js"; +import { measurePlanSizeBreakdown } from "./planSize.js"; import { DISTRIBUTED_DURATION_OUT_OF_RANGE } from "../render/planValidation.js"; const FIXTURE_HTML = ` @@ -57,17 +58,96 @@ describe("measurePlanDirBytes", () => { expect(measurePlanDirBytes(dir)).toBe(400); }); - it("ignores symlinks (not traversed into)", () => { + it("ignores file and directory symlinks instead of traversing outside the plan", () => { const dir = mkdtempSync(join(runRoot, "symlinks-")); + const outsideDir = mkdtempSync(join(runRoot, "outside-")); writeFileSync(join(dir, "real.bin"), Buffer.alloc(128)); - // We don't actually create a symlink here because the planDir - // materialization path strips them — but the function should still - // gracefully ignore broken entries if any slipped in. Confirm the - // baseline is correct (the real file's bytes). + writeFileSync(join(outsideDir, "large.bin"), Buffer.alloc(4_096)); + try { + symlinkSync(join(outsideDir, "large.bin"), join(dir, "file-link.bin")); + symlinkSync(outsideDir, join(dir, "dir-link"), "dir"); + } catch { + // Windows without Developer Mode may reject symlink creation. The + // production Linux path and privileged Windows CI exercise both links. + } expect(measurePlanDirBytes(dir)).toBe(128); }); }); +describe("measurePlanSizeBreakdown", () => { + it("classifies retained plan artifacts without exposing video directory names", () => { + const dir = mkdtempSync(join(runRoot, "breakdown-")); + mkdirSync(join(dir, "compiled", "assets"), { recursive: true }); + mkdirSync(join(dir, "video-frames", "customer-video-name"), { recursive: true }); + mkdirSync(join(dir, "meta"), { recursive: true }); + writeFileSync(join(dir, "compiled", "index.html"), Buffer.alloc(100)); + writeFileSync(join(dir, "compiled", "assets", "source.mp4"), Buffer.alloc(200)); + writeFileSync( + join(dir, "video-frames", "customer-video-name", "frame_000001.jpg"), + Buffer.alloc(300), + ); + writeFileSync(join(dir, "audio.aac"), Buffer.alloc(40)); + writeFileSync(join(dir, "meta", "encoder.json"), Buffer.alloc(20)); + writeFileSync(join(dir, "plan.json"), Buffer.alloc(10)); + writeFileSync(join(dir, "other.bin"), Buffer.alloc(5)); + + const result = measurePlanSizeBreakdown(dir); + + expect(result).toMatchObject({ + totalBytes: 675, + fileCount: 7, + compiledBytes: 300, + sourceMediaBytes: 200, + videoFramesBytes: 300, + videoFrameFileCount: 1, + audioBytes: 40, + metadataBytes: 30, + otherBytes: 5, + }); + const labels = result.topComponents.map((component) => component.label); + expect(labels).toContain("compiled"); + expect(labels.some((label) => label.startsWith("video-frames:"))).toBe(true); + expect(JSON.stringify(result)).not.toContain("customer-video-name"); + }); +}); + +describe("measurePlanSizeBreakdown path boundaries", () => { + it("does not misclassify an authored compiled/video-frames path", () => { + const dir = mkdtempSync(join(runRoot, "compiled-name-collision-")); + mkdirSync(join(dir, "compiled", "assets", "video-frames", "customer"), { + recursive: true, + }); + writeFileSync( + join(dir, "compiled", "assets", "video-frames", "customer", "source.mp4"), + Buffer.alloc(90), + ); + + expect(measurePlanSizeBreakdown(dir)).toMatchObject({ + totalBytes: 90, + compiledBytes: 90, + sourceMediaBytes: 90, + videoFramesBytes: 0, + }); + }); + + it("classifies staged extracted frames separately from compiled assets", () => { + const dir = mkdtempSync(join(runRoot, "staged-breakdown-")); + mkdirSync(join(dir, "__hyperframes_video_frames", "video-1"), { recursive: true }); + writeFileSync(join(dir, "index.html"), Buffer.alloc(25)); + writeFileSync( + join(dir, "__hyperframes_video_frames", "video-1", "frame_000001.jpg"), + Buffer.alloc(75), + ); + + expect(measurePlanSizeBreakdown(dir, "compiled")).toMatchObject({ + totalBytes: 100, + compiledBytes: 25, + videoFramesBytes: 75, + videoFrameFileCount: 1, + }); + }); +}); + describe("PLAN_DIR_SIZE_LIMIT_BYTES constant", () => { it("is the documented 2 GB ceiling", () => { expect(PLAN_DIR_SIZE_LIMIT_BYTES).toBe(2 * 1024 * 1024 * 1024); @@ -120,12 +200,20 @@ describe("plan() PLAN_TOO_LARGE throw path", () => { } expect(caught).toBeInstanceOf(PlanTooLargeError); - expect((caught as PlanTooLargeError).code).toBe(PLAN_TOO_LARGE); - expect((caught as PlanTooLargeError).sizeBytes).toBeGreaterThan(1024); - expect((caught as PlanTooLargeError).limitBytes).toBe(1024); + const error = caught as PlanTooLargeError; + expect(error.code).toBe(PLAN_TOO_LARGE); + expect(error.sizeBytes).toBeGreaterThan(1024); + expect(error.limitBytes).toBe(1024); + expect(error.breakdown?.totalBytes).toBe(error.sizeBytes); + expect(error.observedAt).toBe("post-freeze"); + expect(error.message).toContain("Breakdown:"); }, TIMEOUT_MS, ); +}); + +describe("plan() under size cap", () => { + const TIMEOUT_MS = 30_000; it( "succeeds when the default ceiling is well above the produced planDir size", @@ -146,6 +234,82 @@ describe("plan() PLAN_TOO_LARGE throw path", () => { ); }); +describe("plan() early size budget", () => { + const TIMEOUT_MS = 30_000; + + it( + "rejects an oversized stable compiled tree before extraction and freeze", + async () => { + const projectDir = mkdtempSync(join(runRoot, "project-pre-extract-too-large-")); + writeFileSync( + join(projectDir, "index.html"), + ` + +
+ +
+`, + "utf-8", + ); + // Deliberately invalid media: reaching extraction would fail in ffprobe. + // PLAN_TOO_LARGE at pre-extract proves that expensive stage was skipped. + writeFileSync(join(projectDir, "large-source.mp4"), Buffer.alloc(2_048)); + const planDir = mkdtempSync(join(runRoot, "plandir-pre-extract-too-large-")); + + let caught: unknown; + try { + await plan( + projectDir, + { + fps: 30, + width: 320, + height: 240, + format: "mp4", + planDirSizeLimitBytes: 1_024, + }, + planDir, + ); + } catch (err) { + caught = err; + } + + expect(caught).toBeInstanceOf(PlanTooLargeError); + const error = caught as PlanTooLargeError; + expect(error.observedAt).toBe("pre-extract"); + expect(error.breakdown?.compiledBytes).toBeGreaterThanOrEqual(2_048); + expect(error.breakdown?.sourceMediaBytes).toBe(2_048); + expect(existsSync(join(planDir, "plan.json"))).toBe(false); + }, + TIMEOUT_MS, + ); +}); + +describe("plan() reused directory size check", () => { + it("ignores stale freeze-owned metadata that the new plan overwrites", async () => { + const projectDir = mkdtempSync(join(runRoot, "project-reused-plan-dir-")); + writeFileSync(join(projectDir, "index.html"), FIXTURE_HTML, "utf-8"); + const planDir = mkdtempSync(join(runRoot, "plandir-reused-")); + mkdirSync(join(planDir, "meta"), { recursive: true }); + writeFileSync(join(planDir, "plan.json"), Buffer.alloc(100_000)); + writeFileSync(join(planDir, "meta", "encoder.json"), Buffer.alloc(100_000)); + + const result = await plan( + projectDir, + { + fps: 30, + width: 320, + height: 240, + format: "mp4", + planDirSizeLimitBytes: 4_096, + }, + planDir, + ); + + expect(result.planHash).toMatch(/^[0-9a-f]{64}$/); + expect(measurePlanDirBytes(planDir)).toBeLessThan(4_096); + }, 30_000); +}); + describe("plan() duration guard", () => { const TIMEOUT_MS = 30_000;