fix(producer): stop disk capture early when frames cannot fit on disk (#4164)

* fix(producer): stop disk capture early when frames cannot fit on disk

* test(producer): a measured disk shortfall is never retried as a browser death

packages/producer/src/services/render/stages/captureStage.test.ts

@@ -1,9 +1,9 @@
 import { afterEach, describe, expect, it, vi } from "vitest";
-import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
-import { DEFAULT_CONFIG } from "@hyperframes/engine";
-import { createRenderJob } from "../../renderOrchestrator.js";
+import { DEFAULT_CONFIG, classifyCaptureFailure } from "@hyperframes/engine";
+import { createRenderJob, isTransientCaptureRetryEligible } from "../../renderOrchestrator.js";
 import { formatCaptureFrameName } from "../../../utils/paths.js";
 
 // Mock only the engine session primitives; `classifyCaptureFailure` stays
@@ -14,6 +14,7 @@ const captureFrame = vi.fn();
 const closeCaptureSession = vi.fn(async () => {});
 const verifyDiskDrawElementSamples = vi.fn(async () => {});
 const getCapturePerfSummary = vi.fn(() => ({}));
+const executeParallelCapture = vi.fn();
 
 vi.mock("@hyperframes/engine", async (importOriginal) => {
   const actual = await importOriginal<typeof import("@hyperframes/engine")>();
@@ -25,11 +26,25 @@ vi.mock("@hyperframes/engine", async (importOriginal) => {
     closeCaptureSession: (...args: unknown[]) => closeCaptureSession(...args),
     verifyDiskDrawElementSamples: (...args: unknown[]) => verifyDiskDrawElementSamples(...args),
     getCapturePerfSummary: (...args: unknown[]) => getCapturePerfSummary(...args),
+    executeParallelCapture: (...args: unknown[]) => executeParallelCapture(...args),
+    mergeWorkerFrames: async () => {},
+  };
+});
+
+const freeBytes = vi.fn<(path: string) => { bavail: number; bsize: number }>();
+vi.mock("node:fs", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("node:fs")>();
+  return {
+    ...actual,
+    statfsSync: (path: string) => freeBytes(path) ?? actual.statfsSync(path),
   };
 });
 
 import {
   assertDiskCaptureHeadroom,
+  createDiskCaptureProjection,
+  diskCaptureShortfallError,
+  measureCaptureFrameBytes,
   estimateDiskCaptureBytes,
   inspectDiskCaptureHeadroom,
   runCaptureStage,
@@ -55,13 +70,19 @@ describe("disk capture capacity", () => {
     format: "jpeg" as const,
   };
 
+  const pngOptions = { ...captureOptions, format: "png" as const };
+
   it("estimates output-resolution frame storage conservatively", () => {
-    expect(estimateDiskCaptureBytes(10, captureOptions)).toBe(800_000);
+    expect(estimateDiskCaptureBytes(10, pngOptions)).toBe(800_000);
+  });
+
+  it("estimates jpeg frames well below the raw RGBA model", () => {
+    expect(estimateDiskCaptureBytes(10, captureOptions)).toBe(100_000);
   });
 
   it("fails before capture when estimated frames exceed available headroom", () => {
     expect(() =>
-      assertDiskCaptureHeadroom("/render/captured-frames", 10, captureOptions, () => 800_000),
+      assertDiskCaptureHeadroom("/render/captured-frames", 10, pngOptions, () => 800_000),
     ).toThrow(/may need ~0\.8 MB.*0\.8 MB is free.*--low-memory-mode/s);
   });
 
@@ -70,7 +91,7 @@ describe("disk capture capacity", () => {
       width: 1920,
       height: 1080,
       fps: { num: 30, den: 1 },
-      format: "jpeg" as const,
+      format: "png" as const,
     };
     let message = "";
     try {
@@ -92,28 +113,27 @@ describe("disk capture capacity", () => {
   });
 
   it("exposes the same 90% headroom decision to fallback planning", () => {
-    const estimatedBytes = estimateDiskCaptureBytes(10, captureOptions);
+    const estimatedBytes = estimateDiskCaptureBytes(10, pngOptions);
 
     expect(
       inspectDiskCaptureHeadroom(
         "/render/captured-frames",
         10,
-        captureOptions,
+        pngOptions,
         () => estimatedBytes / 0.9 - 1,
       ),
     ).toEqual({ available: false, estimatedBytes, freeBytes: estimatedBytes / 0.9 - 1 });
     expect(
-      inspectDiskCaptureHeadroom("/render/captured-frames", 10, captureOptions, () => null)
-        .available,
+      inspectDiskCaptureHeadroom("/render/captured-frames", 10, pngOptions, () => null).available,
     ).toBe(true);
   });
 
-  it("rejects the reported 5318-frame landscape disk route with 45 GiB free", () => {
+  it("rejects the reported 5318-frame landscape png disk route with 45 GiB free", () => {
     const landscape = {
       width: 1920,
       height: 1080,
       fps: { num: 30, den: 1 },
-      format: "jpeg" as const,
+      format: "png" as const,
     };
     const headroom = inspectDiskCaptureHeadroom(
       "/render/captured-frames",
@@ -125,6 +145,179 @@ describe("disk capture capacity", () => {
     expect(headroom.estimatedBytes).toBe(5318 * 1920 * 1080 * 4);
     expect(headroom.available).toBe(false);
   });
+
+  it("lets the same 5318-frame landscape route through as jpeg", () => {
+    const landscape = { ...captureOptions, width: 1920, height: 1080, deviceScaleFactor: 1 };
+    expect(
+      inspectDiskCaptureHeadroom("/render/captured-frames", 5318, landscape, () => 45 * 1024 ** 3)
+        .available,
+    ).toBe(true);
+  });
+});
+
+describe("createDiskCaptureProjection", () => {
+  it("throws a measured shortfall once ten frames show the remainder cannot fit", () => {
+    const measure = vi.fn(() => 10e6);
+    const check = createDiskCaptureProjection({
+      framesDir: "/frames",
+      totalFrames: 1010,
+      measureBytes: measure,
+      freeDiskBytes: () => 500e6,
+    });
+    check(9);
+    expect(measure).not.toHaveBeenCalled();
+    expect(() => check(10)).toThrow(
+      /~1000\.0 MB.*measured from the first 10 frames.*500\.0 MB is free at \/frames/s,
+    );
+    expect(() => check(11)).not.toThrow();
+  });
+
+  it("does not spend its one check on an empty measurement", () => {
+    let bytes = 0;
+    const check = createDiskCaptureProjection({
+      framesDir: "/frames",
+      totalFrames: 1010,
+      measureBytes: () => bytes,
+      freeDiskBytes: () => 500e6,
+    });
+    expect(() => check(10)).not.toThrow();
+    bytes = 10e6;
+    expect(() => check(11)).toThrow(/measured from the first/);
+  });
+
+  it("measures frames in retry-batch worker dirs as well as capture-attempt dirs", () => {
+    const workDir = mkdtempSync(join(tmpdir(), "hf-measure-"));
+    try {
+      const framesDir = join(workDir, "captured-frames");
+      for (const dir of [
+        framesDir,
+        join(workDir, "capture-attempt-0", "worker-0"),
+        join(workDir, "retry-1-batch-0-worker-0"),
+      ]) {
+        mkdirSync(dir, { recursive: true });
+        writeFileSync(join(dir, "frame_000001.jpg"), Buffer.alloc(1000));
+      }
+      expect(measureCaptureFrameBytes(workDir, framesDir)).toBe(3000);
+    } finally {
+      rmSync(workDir, { recursive: true, force: true });
+    }
+  });
+
+  it("stays quiet when the projected remainder fits", () => {
+    const check = createDiskCaptureProjection({
+      framesDir: "/frames",
+      totalFrames: 1010,
+      measureBytes: () => 10e6,
+      freeDiskBytes: () => 5000e6,
+    });
+    expect(() => check(10)).not.toThrow();
+  });
+});
+
+describe("runCaptureStage measured disk projection (both branches)", () => {
+  const roots: string[] = [];
+  const oneMb = Buffer.alloc(1_000_000);
+  const makeInput = (workerCount: number, workDir: string, framesDir: string) => ({
+    fileServer: { url: "http://localhost:0" } as never,
+    workDir,
+    framesDir,
+    job: createRenderJob({ fps: { num: 30, den: 1 }, quality: "draft" }),
+    totalFrames: 20,
+    cfg: DEFAULT_CONFIG,
+    plan: { workerCount, forceScreenshot: false },
+    log: { error: vi.fn(), warn: vi.fn(), info: vi.fn(), debug: vi.fn() } as never,
+    probeSession: null,
+    captureAttempts: [],
+    dedupPerfs: [],
+    buildCaptureOptions: () => ({
+      width: 64,
+      height: 64,
+      fps: { num: 30, den: 1 },
+      format: "jpeg" as const,
+    }),
+    createRenderVideoFrameInjector: () => null,
+    abortSignal: undefined,
+    assertNotAborted: () => {},
+  });
+  const makeDirs = () => {
+    const workDir = mkdtempSync(join(tmpdir(), "hf-disk-projection-"));
+    const framesDir = join(workDir, "captured-frames");
+    mkdirSync(framesDir);
+    roots.push(workDir);
+    return { workDir, framesDir };
+  };
+
+  afterEach(() => {
+    for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true });
+    createCaptureSession.mockReset();
+    captureFrame.mockReset();
+    executeParallelCapture.mockReset();
+    freeBytes.mockReset();
+    closeCaptureSession.mockReset().mockResolvedValue(undefined);
+  });
+
+  it("sequential: stops after ten frames when the measured remainder cannot fit", async () => {
+    const { workDir, framesDir } = makeDirs();
+    // 8 MB free: passes the static jpeg estimate, fails the 10 MB measured remainder.
+    freeBytes.mockReturnValue({ bavail: 8_000_000, bsize: 1 });
+    createCaptureSession.mockResolvedValue({
+      isInitialized: false,
+      browserConsoleBuffer: [],
+      options: { captureBeyondViewport: false, format: "jpeg" },
+      workerEncodeEnabled: false,
+      outputDir: framesDir,
+    });
+    captureFrame.mockImplementation(async (_s: unknown, index: number) => {
+      writeFileSync(join(framesDir, formatCaptureFrameName(index, "jpg")), oneMb);
+    });
+
+    await expect(runCaptureStage(makeInput(1, workDir, framesDir))).rejects.toThrow(
+      /measured from the first 10 frames/,
+    );
+    expect(captureFrame).toHaveBeenCalledTimes(10);
+    expect(createCaptureSession).toHaveBeenCalledTimes(1);
+  });
+
+  it("a measured shortfall is never classified as a retryable browser death", () => {
+    const shortfall = diskCaptureShortfallError(10e9, 1e9, "/frames", "measured");
+    const kind = classifyCaptureFailure(shortfall);
+    expect(kind).not.toBe("transient_browser");
+    expect(isTransientCaptureRetryEligible(kind, [{ startFrame: 10, endFrame: 20 }], 0)).toBe(
+      false,
+    );
+  });
+
+  it("parallel: the progress callback raises the same shortfall from worker dirs", async () => {
+    const { workDir, framesDir } = makeDirs();
+    freeBytes.mockReturnValue({ bavail: 8_000_000, bsize: 1 });
+    executeParallelCapture.mockImplementation(
+      async (
+        _url: string,
+        attemptWorkDir: string,
+        _tasks: unknown,
+        _opts: unknown,
+        _hook: unknown,
+        _signal: unknown,
+        onProgress: (p: {
+          capturedFrames: number;
+          totalFrames: number;
+          activeWorkers: number;
+        }) => void,
+      ) => {
+        const workerDir = join(attemptWorkDir, "worker-0");
+        mkdirSync(workerDir, { recursive: true });
+        for (let i = 0; i < 10; i++) {
+          writeFileSync(join(workerDir, formatCaptureFrameName(i, "jpg")), oneMb);
+        }
+        onProgress({ capturedFrames: 10, totalFrames: 20, activeWorkers: 2 });
+        return [];
+      },
+    );
+
+    await expect(runCaptureStage(makeInput(2, workDir, framesDir))).rejects.toThrow(
+      /measured from the first 10 frames/,
+    );
+  });
 });
 
 describe("runCaptureStage — sequential path transient Target-closed single retry (integration)", () => {

packages/producer/src/services/render/stages/captureStage.ts

@@ -37,7 +37,8 @@
  * the stages can import them without reaching back into the orchestrator.
  */
 
-import { statfsSync } from "node:fs";
+import { existsSync, readdirSync, statfsSync } from "node:fs";
+import { join } from "node:path";
 import {
   classifyCaptureFailure,
   frameFileExtension,
@@ -63,6 +64,7 @@ import {
   executeDiskCaptureWithAdaptiveRetry,
   findMissingFrameRanges,
   isTransientCaptureRetryEligible,
+  sampleDirectoryBytes,
   type CaptureAttemptSummary,
   type ProgressCallback,
   type RenderJob,
@@ -142,6 +144,11 @@ export function shouldAllowAdaptiveCaptureRetry(
   return workerCount > 1;
 }
 
+/** Jpeg frames are far smaller than raw pixels; this divisor is a conservative bound. */
+const JPEG_RAW_RGBA_DIVISOR = 8;
+/** Frames captured before the measured projection replaces the static estimate. */
+const DISK_PROJECTION_SAMPLE_FRAMES = 10;
+
 export function estimateDiskCaptureBytes(
   totalFrames: number,
   captureOptions: CaptureOptions,
@@ -149,7 +156,19 @@ export function estimateDiskCaptureBytes(
   const scale = captureOptions.deviceScaleFactor ?? 1;
   const outputWidth = Math.ceil(captureOptions.width * scale);
   const outputHeight = Math.ceil(captureOptions.height * scale);
-  return Math.ceil(totalFrames) * outputWidth * outputHeight * 4;
+  const rawBytes = Math.ceil(totalFrames) * outputWidth * outputHeight * 4;
+  return frameFileExtension(captureOptions.format) === "jpg"
+    ? Math.ceil(rawBytes / JPEG_RAW_RGBA_DIVISOR)
+    : rawBytes;
+}
+
+function freeDiskBytesAt(path: string): number | null {
+  try {
+    const stat = statfsSync(path);
+    return stat.bavail * stat.bsize;
+  } catch {
+    return null;
+  }
 }
 
 /**
@@ -165,14 +184,7 @@ export function inspectDiskCaptureHeadroom(
   framesDir: string,
   totalFrames: number,
   captureOptions: CaptureOptions,
-  freeDiskBytes: (path: string) => number | null = (path) => {
-    try {
-      const stat = statfsSync(path);
-      return stat.bavail * stat.bsize;
-    } catch {
-      return null;
-    }
-  },
+  freeDiskBytes: (path: string) => number | null = freeDiskBytesAt,
 ): DiskCaptureHeadroom {
   const freeBytes = freeDiskBytes(framesDir);
   const estimatedBytes = estimateDiskCaptureBytes(totalFrames, captureOptions);
@@ -182,6 +194,32 @@ export function inspectDiskCaptureHeadroom(
   return { available: false, estimatedBytes, freeBytes };
 }
 
+export function diskCaptureShortfallError(
+  needBytes: number,
+  freeBytes: number,
+  framesDir: string,
+  basis: "estimate" | "measured",
+): Error {
+  const basisNote =
+    basis === "measured"
+      ? ` (measured from the first ${DISK_PROJECTION_SAMPLE_FRAMES} frames)`
+      : "";
+  return new Error(
+    `Disk capture may need ~${(needBytes / 1e6).toFixed(1)} MB of temporary frame storage${basisNote}, ` +
+      `but only ${(freeBytes / 1e6).toFixed(1)} MB is free at ${framesDir}. ` +
+      "Disk capture stores every frame as an image file (JPEG, or PNG for alpha). mp4/mov renders stream instead: " +
+      "always at --workers 1, and multi-worker on Linux BeginFrame capture. " +
+      "This render landed on disk because the output is png-sequence or gif, because it " +
+      "is multi-worker screenshot capture (opt in with HF_CAPTURE_PARALLEL_STREAM=true), " +
+      "or because streaming was switched off: HF_CAPTURE_PARALLEL_STREAM=false, " +
+      "PRODUCER_ENABLE_STREAMING_ENCODE=false, or the duration cap " +
+      "(PRODUCER_STREAMING_ENCODE_DURATION_CAP_ENABLED=true) on a render longer than " +
+      "PRODUCER_STREAMING_ENCODE_MAX_DURATION_SECONDS. HDR and shader-transition renders " +
+      "never reach this check; they run their own compositor. " +
+      "Re-run with --workers 1, use --low-memory-mode, or free up disk space.",
+  );
+}
+
 export function assertDiskCaptureHeadroom(
   framesDir: string,
   totalFrames: number,
@@ -195,22 +233,56 @@ export function assertDiskCaptureHeadroom(
     freeDiskBytes,
   );
   if (headroom.available) return;
-  throw new Error(
-    `Disk capture may need ~${(headroom.estimatedBytes / 1e6).toFixed(1)} MB of temporary frame storage, ` +
-      `but only ${(headroom.freeBytes / 1e6).toFixed(1)} MB is free at ${framesDir}. ` +
-      "Disk capture stores every frame as raw RGBA. mp4/mov renders stream instead: " +
-      "always at --workers 1, and multi-worker on Linux BeginFrame capture. " +
-      "This render landed on disk because the output is png-sequence or gif, because it " +
-      "is multi-worker screenshot capture (opt in with HF_CAPTURE_PARALLEL_STREAM=true), " +
-      "or because streaming was switched off: HF_CAPTURE_PARALLEL_STREAM=false, " +
-      "PRODUCER_ENABLE_STREAMING_ENCODE=false, or the duration cap " +
-      "(PRODUCER_STREAMING_ENCODE_DURATION_CAP_ENABLED=true) on a render longer than " +
-      "PRODUCER_STREAMING_ENCODE_MAX_DURATION_SECONDS. HDR and shader-transition renders " +
-      "never reach this check; they run their own compositor. " +
-      "Re-run with --workers 1, use --low-memory-mode, or free up disk space.",
+  throw diskCaptureShortfallError(
+    headroom.estimatedBytes,
+    headroom.freeBytes,
+    framesDir,
+    "estimate",
   );
 }
 
+/** Frame bytes written so far: the merged frames dir plus every capture attempt's worker dirs. */
+export function measureCaptureFrameBytes(workDir: string, framesDir: string): number {
+  let total = sampleDirectoryBytes(framesDir);
+  if (!existsSync(workDir)) return total;
+  for (const attempt of readdirSync(workDir)) {
+    if (/^retry-\d+-batch-\d+-worker-\d+$/.test(attempt)) {
+      total += sampleDirectoryBytes(join(workDir, attempt));
+      continue;
+    }
+    if (!/^capture-attempt-\d+$/.test(attempt)) continue;
+    const attemptDir = join(workDir, attempt);
+    for (const name of readdirSync(attemptDir)) {
+      if (/^worker-\d+$/.test(name)) total += sampleDirectoryBytes(join(attemptDir, name));
+    }
+  }
+  return total;
+}
+
+/** Per-frame callback: after the sample frames, throws if the projected remainder exceeds 90% of free space. */
+export function createDiskCaptureProjection(input: {
+  framesDir: string;
+  totalFrames: number;
+  measureBytes: () => number;
+  freeDiskBytes?: (path: string) => number | null;
+}): (capturedFrames: number) => void {
+  const { framesDir, totalFrames, measureBytes, freeDiskBytes = freeDiskBytesAt } = input;
+  let checked = false;
+  return (capturedFrames) => {
+    if (checked || capturedFrames < DISK_PROJECTION_SAMPLE_FRAMES) return;
+    const measured = measureBytes();
+    if (measured <= 0) return;
+    checked = true;
+    const remaining = totalFrames - capturedFrames;
+    const freeBytes = freeDiskBytes(framesDir);
+    if (remaining <= 0 || freeBytes === null) return;
+    const needBytes = Math.ceil((measured / capturedFrames) * remaining);
+    if (needBytes > freeBytes * 0.9) {
+      throw diskCaptureShortfallError(needBytes, freeBytes, framesDir, "measured");
+    }
+  };
+}
+
 export async function runCaptureStage(input: CaptureStageInput): Promise<CaptureStageResult> {
   const {
     fileServer,
@@ -270,6 +342,11 @@ export async function runCaptureStage(input: CaptureStageInput): Promise<Capture
 
   const captureOptions = buildCaptureOptions();
   assertDiskCaptureHeadroom(framesDir, totalFrames, captureOptions);
+  const checkDiskProjection = createDiskCaptureProjection({
+    framesDir,
+    totalFrames,
+    measureBytes: () => measureCaptureFrameBytes(workDir, framesDir),
+  });
 
   if (workerCount > 1) {
     // Parallel capture. When `frameRange` is set (distributed chunk), pass
@@ -290,6 +367,7 @@ export async function runCaptureStage(input: CaptureStageInput): Promise<Capture
       dedupPerfs,
       onProgress: (progress) => {
         job.framesRendered = progress.capturedFrames;
+        checkDiskProjection(progress.capturedFrames);
         const frameProgress = progress.capturedFrames / progress.totalFrames;
         const progressPct = 25 + frameProgress * 45;
 
@@ -321,7 +399,12 @@ export async function runCaptureStage(input: CaptureStageInput): Promise<Capture
       probeSession = null;
     }
   } else {
-    const sequential = await runSequentialDiskCapture(input, captureOptions, captureCfg);
+    const sequential = await runSequentialDiskCapture(
+      input,
+      captureOptions,
+      captureCfg,
+      checkDiskProjection,
+    );
     probeSession = null;
     lastBrowserConsole = sequential.lastBrowserConsole;
     captureBeyondViewport = sequential.captureBeyondViewport;
@@ -339,6 +422,7 @@ async function runSequentialDiskCapture(
   input: CaptureStageInput,
   captureOptions: CaptureOptions,
   captureCfg: EngineConfig,
+  checkDiskProjection: (capturedFrames: number) => void,
 ): Promise<SequentialCaptureResult> {
   const {
     fileServer,
@@ -389,7 +473,13 @@ async function runSequentialDiskCapture(
       assertNotAborted();
       lastBrowserConsole = session.browserConsoleBuffer;
 
-      await captureSessionFrames(session, { input, rangeStart, rangeEnd, resumeFrom });
+      await captureSessionFrames(session, {
+        input,
+        rangeStart,
+        rangeEnd,
+        resumeFrom,
+        checkDiskProjection,
+      });
       break;
       // This must mirror streaming capture: catch wraps the original failure with
       // browser diagnostics, finally only handles cleanup.
@@ -471,16 +561,18 @@ interface SequentialFrameLoop {
   rangeStart: number;
   rangeEnd: number;
   resumeFrom: number;
+  checkDiskProjection: (capturedFrames: number) => void;
 }
 
 async function captureSessionFrames(
   session: CaptureSession,
-  { input, rangeStart, rangeEnd, resumeFrom }: SequentialFrameLoop,
+  { input, rangeStart, rangeEnd, resumeFrom, checkDiskProjection }: SequentialFrameLoop,
 ): Promise<void> {
   const { job, framesDir, dedupPerfs, assertNotAborted, onProgress } = input;
   const rangeFrames = rangeEnd - rangeStart;
   const reportFrame = (fileIndex: number): void => {
     job.framesRendered = fileIndex + 1;
+    checkDiskProjection(fileIndex + 1);
     // Keep status cadence identical to the streaming sequential path; the
     // capture error wrapper below must remain separate from finally so it
     // can throw with the browser console before cleanup overwrites flow.

packages/producer/src/services/render/stages/captureStreamingStage.test.ts

@@ -179,6 +179,7 @@ mock.module("../../renderOrchestrator.js", () => ({
   executeDiskCaptureWithAdaptiveRetry: async () => [],
   findMissingFrameRanges: () => [],
   isTransientCaptureRetryEligible: () => false,
+  sampleDirectoryBytes: () => 0,
   resolveCompositeTransfer: () => "srgb",
 }));
 

packages/producer/src/services/renderOrchestrator.ts

@@ -208,7 +208,7 @@ import { shouldUseLayeredComposite } from "./hdrCompositor.js";
 import { resolveCaptureImageFormat } from "./render/captureImageFormat.js";
 import { assertMotionBlurSupported } from "./render/motionBlurRoute.js";
 
-function sampleDirectoryBytes(dir: string): number {
+export function sampleDirectoryBytes(dir: string): number {
   let total = 0;
   const stack: string[] = [dir];
   while (stack.length > 0) {