fix(producer): detach the risky probe before closing it in the segmented session factory

With several workers, openSegmentWorkersTogether calls create() on the shared
default factory concurrently. The probe was only cleared after its close was
awaited, so a second worker still saw it and closed it again. Clear the
binding first, then close.

The reuse branch has no such window: `probeSession ?? await open()` does not
await when the probe is present, and the binding is cleared synchronously.

— Via

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Vance Ingalls
2026-10-26 21:35:01 +00:00
co-authored by Claude Opus 5.5
parent c4d73299bd
commit b0e8eff4d1
2 changed files with 26 additions and 1 deletions
@@ -346,6 +346,30 @@ describe("runCaptureSegmentedStage", () => {
expect(closeCaptureSession.mock.calls.map((c) => c[0])).toContain(probeSession);
});
it("closes a risky probe once when several workers open sessions together", async () => {
closeCaptureSession.mockClear();
const probeSession = {
...fakeStageInput({ totalFrames: 1 }).probeSession,
deRiskFrames: new Set([2]),
};
const result = await runCaptureSegmentedStage({
...fakeStageInput({ totalFrames: 6 }),
probeSession: probeSession as never,
segmentFrames: 3,
workerCount: 2,
segmentDir: join(fixtureRoot, "risk-probe-workers"),
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(closeCaptureSession.mock.calls.filter((c) => c[0] === probeSession)).toHaveLength(1);
});
it("skips completed segments and still concats all of them in order", async () => {
const stableDir = join(fixtureRoot, "stable");
const spawned: string[] = [];
@@ -293,8 +293,9 @@ function defaultSessionFactory(input: CaptureSegmentedStageInput): SessionFactor
create: async () => {
const probeCannotCaptureEveryFrame = Boolean(probeSession?.deRiskFrames?.size);
if (probeSession && probeCannotCaptureEveryFrame) {
await closeCaptureSession(probeSession).catch(() => {});
const riskyProbe = probeSession;
probeSession = null;
await closeCaptureSession(riskyProbe).catch(() => {});
}
const session = probeSession ?? (await openSegmentedSession(input));
if (probeSession) {