fix(producer): give each parallel segmented-capture worker its own browser (#4475)

* fix(producer): give each parallel segmented-capture worker its own browser

Segmented capture with two or more workers failed at the first frame with
"Invalid JPEG input at frame 0: empty frame". Every worker opened its page in
one pooled Chrome, and beginFrame capture does not survive a shared browser:
a worker's first beginFrame returned no screenshot. Fresh worker sessions now
follow shouldDisableBrowserPoolForParallelWorker, the rule the parallel disk
path already uses.

The default session factory was also built once per worker, so with a probe
session every worker reused that same page. It is now built once per run, so
only the first worker takes the probe.

Workers are now created inside the try, so when a later worker fails to
launch, the finally still closes the browsers opened before it.

* fix(producer): close a segmented worker session whose initialization fails

The default segmented session factory opened a fresh browser and then
initialized it with nothing owning it on failure, so a page-load or init
timeout leaked that Chrome. It now closes the session and rethrows, as the
streaming stage does.

packages/engine/src/index.ts

@@ -269,6 +269,7 @@ export {
   executeParallelCapture,
   mergeWorkerFrames,
   getSystemResources,
+  shouldDisableBrowserPoolForParallelWorker,
   type WorkerTask,
   type WorkerResult,
   type WorkerSizing,

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

@@ -28,6 +28,7 @@ let failPrepareCaptureSessionForReuse = false;
 let initializeSessionErrorMessage = "initialize failed";
 const browserConsoleBuffer = ["[FrameCapture:ERROR] page.goto failed"];
 const closeCaptureSession = mock(async () => {});
+const createdSessionConfigs: Array<{ enableBrowserPool?: boolean }> = [];
 class DrawElementVerificationError extends Error {}
 
 mock.module("@hyperframes/engine", () => ({
@@ -57,13 +58,16 @@ mock.module("@hyperframes/engine", () => ({
   },
   closeCaptureSession,
   completeDeferredDrawElementInit: async () => {},
-  createCaptureSession: async () => ({
-    isInitialized: false,
-    browserConsoleBuffer,
-    options: { captureBeyondViewport: false },
-    workerEncodeEnabled: sessionWorkerEncodeEnabled,
-    captureMode: captureSessionMode,
-  }),
+  createCaptureSession: async (...args: unknown[]) => {
+    createdSessionConfigs.push(args[4] as { enableBrowserPool?: boolean });
+    return {
+      isInitialized: false,
+      browserConsoleBuffer,
+      options: { captureBeyondViewport: false },
+      workerEncodeEnabled: sessionWorkerEncodeEnabled,
+      captureMode: captureSessionMode,
+    };
+  },
   createFrameReorderBuffer: () => ({
     waitForFrame: async () => {},
     advanceTo: () => {},
@@ -130,6 +134,9 @@ mock.module("@hyperframes/engine", () => ({
     }
   },
   recaptureDrawElementFrameForVerify: async () => Buffer.from("frame"),
+  resolveHeadlessShellPath: () => "/headless-shell",
+  shouldDisableBrowserPoolForParallelWorker: ({ parallel }: { parallel?: boolean }) =>
+    Boolean(parallel),
   spawnStreamingEncoder,
   writeCapturedFrame: async () => {},
 }));
@@ -674,6 +681,70 @@ describe("runCaptureSegmentedStage", () => {
     );
   });
 
+  it("gives each parallel worker its own browser and the probe session to one worker", async () => {
+    createdSessionConfigs.length = 0;
+    const result = await runCaptureSegmentedStage({
+      ...fakeStageInput({ totalFrames: 6 }),
+      segmentFrames: 3,
+      segmentDir: join(fixtureRoot, "own-browser"),
+      workerCount: 2,
+      deps: {
+        spawnEncoder: mock(async () => okEncoder()),
+        captureFrame: mock(async () => ({ buffer: Buffer.alloc(1) })),
+        concat: mock(async () => ({ success: true as const })),
+        closeSession: mock(async () => {}),
+        removeFile: () => {},
+      },
+    });
+    expect(result.success).toBe(true);
+    expect(createdSessionConfigs).toEqual([expect.objectContaining({ enableBrowserPool: false })]);
+  });
+
+  it("closes a fresh session whose initialization fails", async () => {
+    closeCaptureSession.mockClear();
+    failInitializeSession = true;
+    try {
+      await expect(
+        runCaptureSegmentedStage({
+          ...fakeStageInput({ totalFrames: 3 }),
+          probeSession: null,
+          segmentFrames: 3,
+          segmentDir: join(fixtureRoot, "initfail"),
+        }),
+      ).rejects.toThrow(/initialize failed/);
+    } finally {
+      failInitializeSession = false;
+    }
+    expect(closeCaptureSession.mock.calls.map((c) => c[0])).toEqual([
+      expect.objectContaining({ isInitialized: false }),
+    ]);
+  });
+
+  it("closes the workers already opened when a later worker fails to launch", async () => {
+    const closed: number[] = [];
+    let id = 0;
+    const result = runCaptureSegmentedStage({
+      ...fakeStageInput({ totalFrames: 9 }),
+      probeSession: null,
+      segmentFrames: 3,
+      segmentDir: join(fixtureRoot, "launchfail"),
+      workerCount: 3,
+      sessionFactory: {
+        create: async () => {
+          if (id === 2) throw new Error("browser launch timed out");
+          return fakeSession(id++);
+        },
+      },
+      deps: {
+        closeSession: mock(async (s: { id: number }) => {
+          closed.push(s.id);
+        }),
+      },
+    });
+    await expect(result).rejects.toThrow(/browser launch timed out/);
+    expect(closed.sort()).toEqual([0, 1]);
+  });
+
   it("closes every worker session when one worker fails hard", async () => {
     const closed: number[] = [];
     let id = 0;

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

@@ -23,6 +23,8 @@ import {
   getCapturePerfSummary,
   initializeSession,
   prepareCaptureSessionForReuse,
+  resolveHeadlessShellPath,
+  shouldDisableBrowserPoolForParallelWorker,
   spawnStreamingEncoder,
   type BeforeCaptureHook,
   type CaptureOptions,
@@ -289,15 +291,18 @@ function defaultSessionFactory(input: CaptureSegmentedStageInput): SessionFactor
   let probeSession = input.probeSession;
   return {
     create: async () => {
-      const session = probeSession
-        ? probeSession
-        : await openSegmentedSession({ ...input, probeSession: null });
+      const session = probeSession ?? (await openSegmentedSession(input));
       if (probeSession) {
         openSegmentedSessionReuse(input, probeSession);
         probeSession = null;
       }
-      if (!session.isInitialized) await initializeSession(session);
-      await completeDeferredDrawElementInit(session);
+      try {
+        if (!session.isInitialized) await initializeSession(session);
+        await completeDeferredDrawElementInit(session);
+      } catch (error) {
+        await closeCaptureSession(session).catch(() => {});
+        throw error;
+      }
       return session;
     },
   };
@@ -310,7 +315,6 @@ function openSegmentedSessionReuse(
   prepareCaptureSessionForReuse(session, input.framesDir, input.createRenderVideoFrameInjector());
 }
 
-/** Reuse the probe session when there is one, else open a fresh one. */
 async function openSegmentedSession(input: CaptureSegmentedStageInput): Promise<CaptureSession> {
   // Same reasoning as the streaming stage: the resolved forceScreenshot comes
   // from the immutable plan, not from the caller-owned cfg.
@@ -319,16 +323,19 @@ async function openSegmentedSession(input: CaptureSegmentedStageInput): Promise<
       ? input.cfg
       : { ...input.cfg, forceScreenshot: input.plan.forceScreenshot };
   const videoInjector = input.createRenderVideoFrameInjector();
-  if (input.probeSession) {
-    prepareCaptureSessionForReuse(input.probeSession, input.framesDir, videoInjector);
-    return input.probeSession;
-  }
+  const captureOptions = input.buildCaptureOptions();
+  const ownBrowser = shouldDisableBrowserPoolForParallelWorker({
+    parallel: (input.workerCount ?? 1) > 1,
+    platform: process.platform,
+    deviceScaleFactor: captureOptions.deviceScaleFactor,
+    headlessShellPath: resolveHeadlessShellPath(captureCfg),
+  });
   return createCaptureSession(
     input.fileServer.url,
     input.framesDir,
-    input.buildCaptureOptions(),
+    captureOptions,
     videoInjector,
-    captureCfg,
+    ownBrowser ? { ...captureCfg, enableBrowserPool: false } : captureCfg,
   );
 }
 
@@ -373,12 +380,13 @@ interface SegmentRun {
   skipSize: number;
   workerCount: number;
   recycleEvery: number;
+  /** One for all workers, so only the first create() takes the probe session. */
+  sessionFactory: SessionFactory;
 }
 
 async function createSegmentWorker(run: SegmentRun, id: number): Promise<SegmentWorker> {
   const { input, deps } = run;
-  const factory =
-    input.sessionFactoryForWorker?.(id) ?? input.sessionFactory ?? defaultSessionFactory(input);
+  const factory = input.sessionFactoryForWorker?.(id) ?? run.sessionFactory;
   return {
     id,
     factory,
@@ -551,6 +559,7 @@ export async function runCaptureSegmentedStage(
     skipSize: skip.size,
     workerCount: Math.max(1, input.workerCount ?? 1),
     recycleEvery: input.browserRecycleEverySegments ?? 0,
+    sessionFactory: input.sessionFactory ?? defaultSessionFactory(input),
   };
   const queue = createSegmentQueue(pending);
   // Built from the plan, not from completion order: workers finish out of
@@ -560,9 +569,8 @@ export async function runCaptureSegmentedStage(
   let fellBackToStreaming = false;
 
   const workers: SegmentWorker[] = [];
-  for (let id = 0; id < run.workerCount; id++) workers.push(await createSegmentWorker(run, id));
   const consoleOf = () => workers[0]?.ctx.session.browserConsoleBuffer ?? [];
-  let lastBrowserConsole: string[] = consoleOf();
+  let lastBrowserConsole: string[] = [];
 
   /** Pull segments until the queue drains, another worker failed, or we fell back. */
   const runWorkerLoop = async (worker: SegmentWorker): Promise<void> => {
@@ -580,6 +588,7 @@ export async function runCaptureSegmentedStage(
 
   try {
     assertNotAborted();
+    for (let id = 0; id < run.workerCount; id++) workers.push(await createSegmentWorker(run, id));
     // allSettled, not all: closing a session out from under a worker that is
     // still capturing orphans its ffmpeg and races the CDP connection, so
     // every worker has to stop before the finally runs. The first error is