fix(producer): attribute and stop oversized plans early

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 {
@@ -82,6 +73,11 @@ import {
   readProducerVersion,
 } from "./shared.js";
 import { CURRENT_PLAN_PROTOCOL, type PlanProtocolV1Descriptor } from "./planProtocol.js";
+import {
+  measurePlanSizeBreakdown,
+  type PlanSizeBreakdown,
+  type PlanSizeRootKind,
+} from "./planSize.js";
 
 /**
  * Caller-supplied configuration for a distributed render. `fps`, `width`,
@@ -360,18 +356,37 @@ 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: use the content-addressed \`planV2()\` transport, shorten ` +
         `the composition, lower the framerate, or use the in-process renderer ` +
-        `(\`executeRenderJob\`) — it has no planDir size cap.`,
+        `(\`executeRenderJob\`) — it has no planDir size cap.` +
+        observationSuffix +
+        breakdownSuffix,
     );
     this.name = "PlanTooLargeError";
     this.sizeBytes = sizeBytes;
     this.limitBytes = limitBytes;
+    this.breakdown = breakdown;
+    this.observedAt = observedAt;
   }
 }
 
@@ -454,30 +469,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);
 }
 
 /**
@@ -566,6 +584,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
@@ -770,6 +801,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) {
@@ -935,6 +967,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).
@@ -1078,6 +1121,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,
@@ -1096,11 +1154,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,

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<Record<string, PlanSizeCategory>> = {
+  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<string, PlanSizeComponent>,
+  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<string, PlanSizeComponent>();
+
+  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;
+}

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 {
@@ -26,6 +26,7 @@ import {
   plan,
 } from "./plan.js";
 import { planV2, readPlanV2Manifest } from "./planV2.js";
+import { measurePlanSizeBreakdown } from "./planSize.js";
 import { DISTRIBUTED_DURATION_OUT_OF_RANGE } from "../render/planValidation.js";
 
 const FIXTURE_HTML = `<!doctype html>
@@ -58,17 +59,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);
@@ -121,12 +201,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",
@@ -181,6 +269,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"),
+        `<!doctype html>
+<html><body>
+  <div data-composition-id="root" data-width="320" data-height="240" data-duration="1">
+    <video id="source" src="large-source.mp4" data-start="0" data-end="1"></video>
+  </div>
+</body></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;