perf(producer): chunk capture workers boot while the BeginFrame probe checks

packages/engine/src/services/parallelCoordinator-peerAbort.test.ts

@@ -238,4 +238,62 @@ describe("executeParallelCapture peer abort", () => {
       rmSync(root, { recursive: true, force: true });
     }
   });
+
+  it("boots every worker at once but captures no frame until beforeFirstFrame opens, and fails them if it rejects", async () => {
+    const root = mkdtempSync(join(tmpdir(), "hf-first-frame-gate-"));
+    const order: string[] = [];
+    const captureFrame = vi.fn(async () => {
+      order.push("frame");
+      return { path: join(root, "frame.jpg") };
+    });
+    vi.doMock("./frameCapture.js", () => ({
+      createCaptureSession: vi.fn(async () => ({ browserConsoleBuffer: [] })),
+      initializeSession: vi.fn(async () => {
+        order.push("init");
+      }),
+      captureFrame,
+      captureFrameToBuffer: vi.fn(),
+      captureFrameToBufferPipelined: vi.fn(),
+      closeCaptureSession: vi.fn().mockResolvedValue(undefined),
+      getCapturePerfSummary: vi.fn(() => ({ frames: 0 })),
+    }));
+    const tasks = [0, 1].map((workerId) => ({
+      workerId,
+      startFrame: workerId,
+      endFrame: workerId + 1,
+      outputDir: join(root, `worker-${workerId}`),
+    }));
+    const run = async (beforeFirstFrame: Promise<void>) => {
+      const { executeParallelCapture } = await import("./parallelCoordinator.js");
+      return executeParallelCapture(
+        "http://127.0.0.1",
+        root,
+        tasks,
+        { width: 320, height: 180, fps: { num: 30, den: 1 } },
+        () => null,
+        undefined,
+        undefined,
+        undefined,
+        undefined,
+        { beforeFirstFrame },
+      );
+    };
+
+    try {
+      let open!: () => void;
+      const gated = run(new Promise<void>((resolve) => (open = resolve)));
+      await vi.waitFor(() => expect(order).toEqual(["init", "init"]));
+      for (let turn = 0; turn < 20; turn++) await new Promise((resolve) => setImmediate(resolve));
+      expect(order).toEqual(["init", "init"]);
+      open();
+      await gated;
+      expect(order).toEqual(["init", "init", "frame", "frame"]);
+
+      order.length = 0;
+      await expect(run(Promise.reject(new Error("probe said no")))).rejects.toThrow(/probe said no/);
+      expect(order).toEqual(["init", "init"]);
+    } finally {
+      rmSync(root, { recursive: true, force: true });
+    }
+  });
 });

packages/engine/src/services/parallelCoordinator.ts

@@ -847,6 +847,7 @@ async function executeWorkerTask(
   parallel?: boolean,
   onFailure?: (failure: CaptureFailure) => void,
   onWorkerPhase?: (diagnostic: ParallelWorkerPhaseDiagnostic) => void,
+  beforeFirstFrame?: Promise<void>,
 ): Promise<WorkerResult> {
   const startTime = Date.now();
   let framesCaptured = 0;
@@ -915,6 +916,7 @@ async function executeWorkerTask(
         await assertSwiftShader(activeSession.page, readWebGlVendorInfoFromCanvas);
       }
       await initializeSession(activeSession);
+      await beforeFirstFrame;
     });
     canvasDrawElement = activeSession.captureMode === "drawelement";
     gpuBackend = activeSession.gpuRenderer ?? "unknown";
@@ -1021,6 +1023,7 @@ export interface ParallelCaptureHooks {
    * frame (the ordered writer) with the original error, not a stall.
    */
   onWorkerFailure?: (failure: CaptureFailure) => void;
+  beforeFirstFrame?: Promise<void>;
 }
 
 function pageErrorLinesOf(session: CaptureSession): string[] {
@@ -1152,6 +1155,7 @@ export async function executeParallelCapture(
         parallel,
         onFailure,
         onWorkerPhase,
+        hooks?.beforeFirstFrame,
       ),
     ),
   );

packages/producer/src/services/distributed/renderChunk.startupOverlap.test.ts

@@ -0,0 +1,148 @@
+import { afterEach, beforeEach, expect, it, vi } from "vitest";
+import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { distributedCaptureSessionDependencies, renderChunk } from "./renderChunk.js";
+
+// Entry: renderChunk on a BeginFrame chunk with 4 workers. The capture stage is a stand-in that records
+// when it starts and what its workers' gate does; the probe browser is a fake whose verdict the test holds.
+const probe = vi.hoisted(() => ({ liveness: undefined as undefined | (() => Promise<boolean>) }));
+const stage = vi.hoisted(() => ({
+  calls: [] as { screenshot: boolean; gate: Promise<void> | undefined }[],
+  run: undefined as undefined | ((call: number) => Promise<never>),
+}));
+
+vi.mock("@hyperframes/engine", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("@hyperframes/engine")>()),
+  probeBeginFrameLiveness: () => probe.liveness?.(),
+}));
+vi.mock("../render/stages/captureStage.js", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("../render/stages/captureStage.js")>()),
+  runCaptureStage: (input: {
+    plan: { forceScreenshot: boolean };
+    beforeFirstFrame?: Promise<void>;
+  }) => {
+    stage.calls.push({ screenshot: input.plan.forceScreenshot, gate: input.beforeFirstFrame });
+    return stage.run?.(stage.calls.length) ?? Promise.reject(new Error("no stage"));
+  },
+}));
+vi.mock("../render/stages/encodeStage.js", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("../render/stages/encodeStage.js")>()),
+  canEncodeWhileCapturing: () => false,
+}));
+vi.mock("../fileServer.js", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("../fileServer.js")>()),
+  createFileServer: async () => ({ url: "http://127.0.0.1:1", port: 1, close: () => {} }),
+}));
+vi.mock("../render/stages/freezePlan.js", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("../render/stages/freezePlan.js")>()),
+  recomputePlanHashFromPlanDir: () => "plan-hash",
+}));
+vi.mock("./shared.js", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("./shared.js")>()),
+  readFfmpegVersion: async () => "ffmpeg-test",
+}));
+
+const STOP = new Error("stop after capture");
+const original = { ...distributedCaptureSessionDependencies };
+let planDir = "";
+let closed = 0;
+
+const flush = async () => {
+  for (let turn = 0; turn < 20; turn++) await new Promise((resolve) => setImmediate(resolve));
+};
+const render = () => renderChunk(planDir, 0, join(planDir, "chunk.mp4"), { maxCaptureWorkers: 4 });
+
+beforeEach(() => {
+  stage.calls = [];
+  stage.run = undefined;
+  closed = 0;
+  Object.assign(distributedCaptureSessionDependencies, {
+    createCaptureSession: async () => ({
+      page: { close: async () => void closed++ },
+      launchCaptureMode: "beginframe",
+      beginFrameTimeTicks: 0,
+      beginFrameIntervalMs: 33,
+      browserConsoleBuffer: [],
+      options: {},
+    }),
+    assertSwiftShader: async () => {},
+    initializeSession: async () => {},
+  });
+  planDir = mkdtempSync(join(tmpdir(), "hf-chunk-overlap-"));
+  const write = (path: string, value: unknown) =>
+    writeFileSync(join(planDir, path), typeof value === "string" ? value : JSON.stringify(value));
+  mkdirSync(join(planDir, "meta"));
+  mkdirSync(join(planDir, "compiled"));
+  write("plan.json", {
+    planHash: "plan-hash",
+    producerVersion: "test",
+    ffmpegVersion: "ffmpeg-test",
+    dimensions: { fpsNum: 30, fpsDen: 1, width: 16, height: 16, format: "mp4" },
+  });
+  write("meta/encoder.json", {
+    browserGpuMode: "software",
+    forceScreenshot: false,
+    deviceScaleFactor: 1,
+    encoder: "libx264-software",
+    quality: "standard",
+    runtimeEnv: {},
+  });
+  write("meta/chunks.json", [{ index: 0, startFrame: 0, endFrame: 240 }]);
+  write("compiled/index.html", "<div></div>");
+});
+afterEach(() => {
+  Object.assign(distributedCaptureSessionDependencies, original);
+  rmSync(planDir, { recursive: true, force: true });
+});
+
+it("the workers start while the probe is still checking, and capture once it passes", async () => {
+  let verdict!: (alive: boolean) => void;
+  probe.liveness = () => new Promise<boolean>((resolve) => (verdict = resolve));
+  const opened: boolean[] = [];
+  stage.run = async () => {
+    await stage.calls[0]?.gate;
+    opened.push(true);
+    throw STOP;
+  };
+  const rendered = render().catch((error: unknown) => error);
+  await flush();
+  expect(stage.calls).toHaveLength(1);
+  expect(opened).toEqual([]);
+  verdict(true);
+  expect(await rendered).toBe(STOP);
+  expect(opened).toEqual([true]);
+  expect(closed).toBe(1);
+});
+
+it("a probe that finds BeginFrame stalled sends the whole chunk to screenshot capture", async () => {
+  probe.liveness = async () => false;
+  stage.run = async (call) => {
+    if (call === 1) {
+      const gate = stage.calls[0]?.gate;
+      await gate?.catch((error: Error) => {
+        throw new Error(`[Parallel] Capture failed: worker 0: ${error.message}`);
+      });
+    }
+    throw STOP;
+  };
+  expect(await render().catch((error: unknown) => error)).toBe(STOP);
+  expect(stage.calls.map((call) => call.screenshot)).toEqual([false, true]);
+  expect(stage.calls[1]?.gate).toBeUndefined();
+  expect(closed).toBe(1);
+});
+
+it("a probe browser that fails surfaces its own error and the chunk is not retried", async () => {
+  const failure = new Error("HeadlessExperimental.beginFrame: software GL required");
+  distributedCaptureSessionDependencies.assertSwiftShader = async () => {
+    throw failure;
+  };
+  stage.run = async () => {
+    await stage.calls[0]?.gate?.catch((error: Error) => {
+      throw new Error(`[Parallel] Capture failed: worker 0: ${error.message}`);
+    });
+    throw STOP;
+  };
+  expect(await render().catch((error: unknown) => error)).toBe(failure);
+  expect(stage.calls).toHaveLength(1);
+});

packages/producer/src/services/distributed/renderChunk.ts

@@ -348,6 +348,9 @@ export function observedChunkCaptureMode(
   return mode;
 }
 
+/** Thrown at the workers' gate when the probe chose screenshot capture; the whole-chunk fallback matches it. */
+const PROBE_SELECTED_SCREENSHOT = "[renderChunk] probe selected screenshot capture";
+
 /**
  * Only BeginFrame-specific failures are safe to retry in screenshot mode.
  * Cancellation, memory exhaustion, and unrelated authoring/IO failures must
@@ -359,6 +362,7 @@ export function shouldRetryChunkCaptureWithScreenshot(error: unknown): boolean {
   return (
     /HeadlessExperimental\.beginFrame/i.test(failure.message) ||
     /beginFrame probe timeout/i.test(failure.message) ||
+    failure.message.includes(PROBE_SELECTED_SCREENSHOT) ||
     /Another frame is pending|Frame still pending/i.test(failure.message)
   );
 }
@@ -369,6 +373,8 @@ export function shouldRetryChunkCaptureWithScreenshot(error: unknown): boolean {
  */
 export async function runCaptureWithScreenshotFallback<T>(input: {
   forceScreenshot: boolean;
+  /** A failure that is not capture's (the probe's own) is never retried. */
+  canRetry?: () => boolean;
   run: (forceScreenshot: boolean) => Promise<T>;
   resetForScreenshotRetry: () => Promise<void> | void;
   onFallback?: (error: unknown) => void;
@@ -376,7 +382,8 @@ export async function runCaptureWithScreenshotFallback<T>(input: {
   try {
     return await input.run(input.forceScreenshot);
   } catch (error) {
-    if (input.forceScreenshot || !shouldRetryChunkCaptureWithScreenshot(error)) throw error;
+    if (input.forceScreenshot || input.canRetry?.() === false) throw error;
+    if (!shouldRetryChunkCaptureWithScreenshot(error)) throw error;
     input.onFallback?.(error);
     await input.resetForScreenshotRetry();
     return await input.run(true);
@@ -765,102 +772,123 @@ export async function renderChunk(
     let forceScreenshotForChunk = encoder.forceScreenshot;
     let shownBoxes: Map<string, ShownBox> | undefined;
     let extraction: Promise<unknown> | undefined;
+    let probeRunning: Promise<void> | undefined;
+    let probeFailed = false;
     // Frames go to the encoder as they land; it is started again if the capture is retried.
     const overlap: { current: EncodeWhileCapturing | null } = { current: null };
     try {
-      if (chunkWorkerCount === 1 || !forceScreenshotForChunk) {
-        // Sequential capture reuses this session. Parallel BeginFrame capture
-        // uses it only for the composition liveness preflight.
-        // 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).
-        const bootStarted = Date.now();
-        session = await createVerifiedDistributedCaptureSession(
-          fileServer.url,
-          framesDir,
-          captureOptions,
-          cfg,
-        );
-        sessionBootMs = Date.now() - bootStarted;
-        // Measured before extraction so each video is extracted at the size it is drawn.
-        shownBoxes = await measureVideoShownBoxes(
-          session.page,
-          [...shownFrames.keys()],
-          chunkFrameNumbers.map((frame) => (frame * chunkFps.den) / chunkFps.num),
-        );
-        // `discardWarmupCapture` is intentionally NOT called: every frame
-        // seeks fresh DOM, so `lastFrameCache` is never read; priming it
-        // would deadlock Chrome's compositor by issuing a second beginFrame
-        // at a `frameTimeTicks` it had just advanced to.
-        const browserSelectedScreenshot = session.launchCaptureMode === "screenshot";
-        const beginFrameStalled =
-          !browserSelectedScreenshot && (await beginFrameSessionNeedsScreenshotFallback(session));
-        if (browserSelectedScreenshot) {
-          forceScreenshotForChunk = true;
-          log.warn(
-            "[renderChunk] Browser capability probe selected screenshot capture for the entire chunk",
-            { chunkIndex },
+      // Parallel BeginFrame capture boots its workers while this probe runs; they wait for its verdict.
+      const overlapStartup = chunkWorkerCount > 1 && !forceScreenshotForChunk;
+      const runProbe = async (): Promise<void> => {
+        if (chunkWorkerCount === 1 || !forceScreenshotForChunk) {
+          // Sequential capture reuses this session. Parallel BeginFrame capture
+          // uses it only for the composition liveness preflight.
+          // 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).
+          const bootStarted = Date.now();
+          session = await createVerifiedDistributedCaptureSession(
+            fileServer.url,
+            framesDir,
+            captureOptions,
+            cfg,
           );
-          if (chunkWorkerCount > 1) {
+          sessionBootMs = Date.now() - bootStarted;
+          // Measured before extraction so each video is extracted at the size it is drawn.
+          shownBoxes = await measureVideoShownBoxes(
+            session.page,
+            [...shownFrames.keys()],
+            chunkFrameNumbers.map((frame) => (frame * chunkFps.den) / chunkFps.num),
+          );
+          // `discardWarmupCapture` is intentionally NOT called: every frame
+          // seeks fresh DOM, so `lastFrameCache` is never read; priming it
+          // would deadlock Chrome's compositor by issuing a second beginFrame
+          // at a `frameTimeTicks` it had just advanced to.
+          const browserSelectedScreenshot = session.launchCaptureMode === "screenshot";
+          const beginFrameStalled =
+            !browserSelectedScreenshot && (await beginFrameSessionNeedsScreenshotFallback(session));
+          if (browserSelectedScreenshot) {
+            forceScreenshotForChunk = true;
+            log.warn(
+              "[renderChunk] Browser capability probe selected screenshot capture for the entire chunk",
+              { chunkIndex },
+            );
+            if (chunkWorkerCount > 1) {
+              await closeCaptureSession(session);
+              session = null;
+            }
+          } else if (beginFrameStalled) {
+            forceScreenshotForChunk = true;
+            log.warn(
+              "[renderChunk] BeginFrame liveness probe failed; using screenshot capture for the entire chunk",
+              { chunkIndex },
+            );
+            await closeCaptureSession(session).catch(() => {});
+            session = null;
+            if (chunkWorkerCount === 1) {
+              const screenshotBootStarted = Date.now();
+              const screenshotCfg: EngineConfig = {
+                ...cfg,
+                forceScreenshot: true,
+                forceScreenshotExplicitlyOptedOut: false,
+              };
+              session = await createVerifiedDistributedCaptureSession(
+                fileServer.url,
+                framesDir,
+                captureOptions,
+                screenshotCfg,
+              );
+              sessionBootMs += Date.now() - screenshotBootStarted;
+            }
+          } else if (chunkWorkerCount > 1) {
             await closeCaptureSession(session);
             session = null;
           }
-        } else if (beginFrameStalled) {
-          forceScreenshotForChunk = true;
-          log.warn(
-            "[renderChunk] BeginFrame liveness probe failed; using screenshot capture for the entire chunk",
-            { chunkIndex },
-          );
-          await closeCaptureSession(session).catch(() => {});
-          session = null;
-          if (chunkWorkerCount === 1) {
-            const screenshotBootStarted = Date.now();
-            const screenshotCfg: EngineConfig = {
-              ...cfg,
-              forceScreenshot: true,
-              forceScreenshotExplicitlyOptedOut: false,
-            };
-            session = await createVerifiedDistributedCaptureSession(
-              fileServer.url,
-              framesDir,
-              captureOptions,
-              screenshotCfg,
-            );
-            sessionBootMs += Date.now() - screenshotBootStarted;
-          }
-        } else if (chunkWorkerCount > 1) {
-          await closeCaptureSession(session);
-          session = null;
         }
-      }
+      };
+      const probeRun = runProbe();
+      probeRun.catch(() => {
+        probeFailed = true;
+      });
+      probeRunning = probeRun;
+      if (!overlapStartup) await probeRun;
+      const probeVerdict = overlapStartup
+        ? probeRun.then(() => {
+            if (forceScreenshotForChunk) throw new Error(PROBE_SELECTED_SCREENSHOT);
+          })
+        : undefined;
+      probeVerdict?.catch(() => {});
 
       // Build the immutable frame lookup once. Each browser session/worker gets
       // its own injector hook because the hook remembers the last injected frame
       // per video; sharing that state across a fresh retry page can suppress the
       // first injection and produce a blank/stale frame.
-      const extractStarted = Date.now();
       const deferredVideoFrames = planVideos
-        ? extractDeferredVideoFramesForChunk({
-            planDir,
-            planVideos,
-            frames: chunkFrameNumbers,
-            fps: chunkFps,
-            outputRoot: videoFramesDir,
-            cfg,
-            remoteSources,
-            shownBoxes,
-          }).then((frames) => {
-            videoExtractMs = Date.now() - extractStarted;
-            return frames;
+        ? probeRun.then(() => {
+            const extractStarted = Date.now();
+            return extractDeferredVideoFramesForChunk({
+              planDir,
+              planVideos,
+              frames: chunkFrameNumbers,
+              fps: chunkFps,
+              outputRoot: videoFramesDir,
+              cfg,
+              remoteSources,
+              shownBoxes,
+            }).then((frames) => {
+              videoExtractMs = Date.now() - extractStarted;
+              return frames;
+            });
           })
-        : Promise.resolve([]);
+        : probeRun.then(() => []);
       extraction = deferredVideoFrames;
+      deferredVideoFrames.catch(() => {});
       const videoFrameLookup =
         planVideos && planVideos.extracted.length > 0
           ? deferredVideoFrames.then((deferred) =>
@@ -951,6 +979,7 @@ export async function renderChunk(
       const captureStarted = Date.now();
       captureMode = await runCaptureWithScreenshotFallback({
         forceScreenshot: forceScreenshotForChunk,
+        canRetry: () => !probeFailed,
         // fallow-ignore-next-line complexity
         run: async (forceScreenshot) => {
           const captureCfg: EngineConfig =
@@ -984,8 +1013,8 @@ export async function renderChunk(
 
           // runCaptureStage owns and closes any probe session it receives,
           // including error paths. Clear our defensive handle before await.
-          const captureSession = session;
-          session = null;
+          const captureSession = overlapStartup ? null : session;
+          if (!overlapStartup) session = null;
           overlap.current = canEncodeWhileCapturing(encodeInput, chunkWorkerCount, cpus().length)
             ? encodeWhileCapturing(encodeInput, framesInChunk)
             : null;
@@ -1016,6 +1045,7 @@ export async function renderChunk(
                 onFrameLanded: (frame) => overlap.current?.onFrameLanded(frame),
                 interleaveFrames: interleavesCapture(overlap.current !== null, options),
                 openAttempt,
+                beforeFirstFrame: forceScreenshot ? undefined : probeVerdict,
               }),
             { stall: frameStallRule(job) },
           ).catch(async (captureError: unknown) => {
@@ -1046,6 +1076,8 @@ export async function renderChunk(
           );
         },
       }).catch(async (captureError: unknown) => {
+        // The probe's own failure, not the workers' echo of it.
+        await probeRun;
         await deferredVideoFrames;
         throw captureError;
       });
@@ -1085,6 +1117,7 @@ export async function renderChunk(
     } finally {
       // No ffmpeg left writing into a work dir the caller removes; its error surfaced above.
       await overlap.current?.abort().catch(() => {});
+      await probeRunning?.catch(() => {});
       await extraction?.catch(() => {});
       // Cleanest path: captureStage closed the session for us. The defensive
       // close handles error paths where we threw before delegating.

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

@@ -121,6 +121,7 @@ export interface CaptureStageInput {
   interleaveFrames?: boolean;
   /** Each capture attempt opens one, so a stalled attempt is retried by the ladders below. */
   openAttempt?: OpenStallAttempt;
+  beforeFirstFrame?: Promise<void>;
 }
 
 export interface CaptureStageResult {
@@ -385,6 +386,7 @@ export async function runCaptureStage(input: CaptureStageInput): Promise<Capture
       dedupPerfs,
       pageErrorLines: workerPageErrors,
       openAttempt: input.openAttempt,
+      beforeFirstFrame: input.beforeFirstFrame,
       onProgress: (progress) => {
         if (progress.latestWorkerPhase) {
           if (progress.capturedFrames === 0) reportWorkerStartup(job, progress, onProgress);

packages/producer/src/services/renderOrchestrator.ts

@@ -1253,6 +1253,7 @@ export async function executeDiskCaptureWithAdaptiveRetry(options: {
   /** Mutated in place: the workers' page-error console lines, for the render's error count. */
   pageErrorLines?: string[];
   openAttempt?: OpenStallAttempt;
+  beforeFirstFrame?: Promise<void>;
 }): Promise<CaptureAttemptSummary[]> {
   const attempts: CaptureAttemptSummary[] = [];
   let currentWorkers = options.initialWorkerCount;
@@ -1331,6 +1332,7 @@ export async function executeDiskCaptureWithAdaptiveRetry(options: {
               : undefined,
             undefined,
             options.cfg,
+            { beforeFirstFrame: options.beforeFirstFrame },
           );
           pushWorkerDedupPerfs(workerResults, options.dedupPerfs);
           if (options.pageErrorLines) pushWorkerPageErrors(workerResults, options.pageErrorLines);