fix(producer): recover from worker crashes instead of hanging the render (#1132)

* fix(producer): recover from worker crashes instead of hanging the render

Both the shader-transition and png-decode-blit worker pools freed a
crashed worker's slot (busy=false, current=null) but left it in the slot
list and never marked it dead. A later run() then selected the dead slot
via slots.find(s => !s.busy) and dispatched to its terminated worker,
where postMessage is a silent no-op (no throw, no reply) — so the task
promise never settled. In the HDR hybrid capture loop, which pipelines
blends across N DOM workers and awaits every dispatch, that wedges the
whole render with no fail-fast.

The crash handlers also never drained the queue, so a queued task could
wait forever for a slot that had died.

Mark a slot dead on error/exit, exclude dead slots from dispatch and from
run()'s slot selection, and fail fast: when no live workers remain, reject
queued tasks and reject new run() calls rather than hanging. This keeps
the pools' existing no-respawn, fail-fast intent; it just actually fails
fast instead of wedging.

Adds crash-recovery tests to both pools via a fixture worker that throws
on its first message, asserting the in-flight task, queued tasks, and
subsequent run() calls all settle rather than hang.

* fix(producer): address review nits on worker-pool crash recovery

- Reword the dead-marking comments in both onWorkerError handlers: the
  flag is set before rejecting and before draining the queue, not
  "before anything else" (current/busy are cleared first).
- Rename the shader pool's all-slots-die test to match the png pool's
  equivalent; the size-2 fixture crashes every worker, so there are no
  surviving workers serving.

---------

Co-authored-by: Carlos Alcaraz <193642530+calcarazgre646@users.noreply.github.com>

.fallowrc.jsonc

@@ -22,6 +22,9 @@
     // Worker entry points loaded dynamically by their *Pool.ts companions.
     "packages/producer/src/services/pngDecodeBlitWorker.ts",
     "packages/producer/src/services/shaderTransitionWorker.ts",
+    // Test fixture worker, spawned by path via the pools' workerEntryPath
+    // option from the crash-recovery tests; has no import-graph referrer.
+    "packages/producer/src/services/__fixtures__/crashOnMessageWorker.mjs",
     "scripts/*.{ts,mjs,js}",
     "scripts/*/run.mjs",
   ],

packages/producer/src/services/__fixtures__/crashOnMessageWorker.mjs

@@ -0,0 +1,19 @@
+/**
+ * Test fixture worker (not part of the production build).
+ *
+ * Simulates a worker that dies mid-task — e.g. an OOM kill on a heavy shader
+ * frame — by throwing an uncaught exception on the first message it receives.
+ * The throw surfaces as the Worker `error` event (followed by `exit`), which
+ * is exactly the crash path the shader / png-decode pools must recover from:
+ * the crashing slot has to be marked dead so a later task is never routed to
+ * its terminated worker (where `postMessage` is a silent no-op and the task
+ * would hang forever).
+ *
+ * Referenced only by path via the pools' `workerEntryPath` option in
+ * shaderTransitionWorkerPool.test.ts / pngDecodeBlitWorkerPool.test.ts.
+ */
+import { parentPort } from "node:worker_threads";
+
+parentPort?.on("message", () => {
+  throw new Error("simulated worker crash");
+});

packages/producer/src/services/pngDecodeBlitWorkerPool.test.ts

@@ -321,4 +321,64 @@ describe("PngDecodeBlitWorkerPool", () => {
     // both are acceptable.
     expect(r1).toBeDefined();
   });
+
+  describe("crash recovery", () => {
+    const CRASH_WORKER = resolve(
+      dirname(fileURLToPath(import.meta.url)),
+      "__fixtures__",
+      "crashOnMessageWorker.mjs",
+    );
+
+    async function makeCrashPool(size: number): Promise<PngDecodeBlitWorkerPool> {
+      const p = await createPngDecodeBlitWorkerPool({ size, workerEntryPath: CRASH_WORKER });
+      pools.push(p);
+      return p;
+    }
+
+    function blitReq(): Parameters<PngDecodeBlitWorkerPool["run"]>[0] {
+      return {
+        png: Buffer.from([0]),
+        dest: Buffer.alloc(RGB48_BYTES),
+        width: W,
+        height: H,
+        transfer: "srgb",
+      };
+    }
+
+    function settledWithin(p: Promise<unknown>, ms = 3000): Promise<string> {
+      return Promise.race([
+        p.then(
+          () => "resolved",
+          () => "rejected",
+        ),
+        new Promise<string>((r) => setTimeout(() => r("hung"), ms)),
+      ]);
+    }
+
+    it("rejects the in-flight task when its only worker crashes, then fails subsequent runs fast", async () => {
+      const pool = await makeCrashPool(1);
+      expect(await settledWithin(pool.run(blitReq()))).toBe("rejected");
+      // Dead slot must be excluded; a later run fails fast rather than hanging
+      // on a postMessage to the terminated worker.
+      expect(await settledWithin(pool.run(blitReq()))).toBe("rejected");
+    });
+
+    it("rejects a queued task on crash instead of leaving it to hang", async () => {
+      const pool = await makeCrashPool(1);
+      const inFlight = settledWithin(pool.run(blitReq()));
+      const queued = settledWithin(pool.run(blitReq()));
+      expect(await inFlight).toBe("rejected");
+      expect(await queued).toBe("rejected");
+    });
+
+    it("never wedges the pool when slots die", async () => {
+      const pool = await makeCrashPool(2);
+      const results = await Promise.all([
+        settledWithin(pool.run(blitReq())),
+        settledWithin(pool.run(blitReq())),
+        settledWithin(pool.run(blitReq())),
+      ]);
+      expect(results).not.toContain("hung");
+    });
+  });
 });

packages/producer/src/services/pngDecodeBlitWorkerPool.ts

@@ -123,6 +123,14 @@ interface WorkerSlot {
   worker: Worker;
   busy: boolean;
   current: PendingTask | null;
+  /**
+   * Set once the worker has crashed (`error`) or exited unexpectedly. A dead
+   * slot must never be dispatched to again: `postMessage` to a terminated
+   * Worker is a silent no-op (no throw, no reply), so a task routed to it
+   * would hang forever. The pool does not respawn mid-render, so a dead slot
+   * stays dead until teardown.
+   */
+  dead: boolean;
 }
 
 interface WorkerReply {
@@ -215,8 +223,18 @@ export async function createPngDecodeBlitWorkerPool(
 
   const execArgv = buildExecArgv(entryIsTs);
 
+  // When every worker has died there is no live thread left to drain the
+  // queue, so any waiting tasks would hang forever. Reject them instead.
+  const failQueueIfNoLiveSlots = (): void => {
+    if (slots.some((s) => !s.dead)) return;
+    while (queue.length > 0) {
+      const t = queue.shift();
+      if (t) t.reject(new Error("png-decode-blit pool has no live workers; task abandoned"));
+    }
+  };
+
   const dispatchNext = (slot: WorkerSlot): void => {
-    if (terminated || slot.busy) return;
+    if (terminated || slot.busy || slot.dead) return;
     const task = queue.shift();
     if (!task) return;
     slot.busy = true;
@@ -347,28 +365,35 @@ export async function createPngDecodeBlitWorkerPool(
     const task = slot.current;
     slot.current = null;
     slot.busy = false;
+    // Mark dead before rejecting and before draining the queue so this slot is
+    // excluded from future dispatch; postMessage to its terminated worker would
+    // be a silent no-op and any task routed here would hang.
+    slot.dead = true;
     if (task) {
       task.reject(
         new Error(`png-decode-blit worker crashed mid-task: ${err.message}; dest buffer lost`),
       );
     }
     log.warn?.("[pngDecodeBlitWorkerPool] worker errored", { err: err.message });
+    failQueueIfNoLiveSlots();
   };
 
   const onWorkerExit = (slot: WorkerSlot, code: number): void => {
     if (terminated) return;
+    slot.dead = true;
     if (slot.current) {
       slot.current.reject(new Error(`png-decode-blit worker exited (code=${code}) mid-task`));
       slot.current = null;
       slot.busy = false;
     }
     log.warn?.("[pngDecodeBlitWorkerPool] worker exited unexpectedly", { code });
+    failQueueIfNoLiveSlots();
   };
 
   try {
     for (let i = 0; i < size; i++) {
       const worker = new Worker(entry, { execArgv });
-      const slot: WorkerSlot = { worker, busy: false, current: null };
+      const slot: WorkerSlot = { worker, busy: false, current: null, dead: false };
       worker.on("message", (msg: WorkerReply) => onWorkerMessage(slot, msg));
       worker.on("error", (err: unknown) =>
         onWorkerError(slot, err instanceof Error ? err : new Error(String(err))),
@@ -394,12 +419,17 @@ export async function createPngDecodeBlitWorkerPool(
         const task: PendingTask = traceEnabled
           ? { req, resolve, reject, enqueuedAtMs: Date.now(), traceId: ++nextTaskId }
           : { req, resolve, reject };
-        const idle = slots.find((s) => !s.busy);
+        const idle = slots.find((s) => !s.busy && !s.dead);
         if (idle) {
           queue.unshift(task);
           dispatchNext(idle);
-        } else {
+        } else if (slots.some((s) => !s.dead)) {
+          // A live worker is busy; it drains the queue when it completes.
           queue.push(task);
+        } else {
+          // Every worker has died — don't hang waiting for a dispatch that
+          // can never happen.
+          reject(new Error("png-decode-blit pool has no live workers"));
         }
       });
     },

packages/producer/src/services/shaderTransitionWorkerPool.test.ts

@@ -316,4 +316,74 @@ describe("ShaderTransitionWorkerPool", () => {
     // Remove from `pools` so afterEach doesn't double-terminate.
     pools.pop();
   });
+
+  describe("crash recovery", () => {
+    const CRASH_WORKER = resolve(
+      dirname(fileURLToPath(import.meta.url)),
+      "__fixtures__",
+      "crashOnMessageWorker.mjs",
+    );
+
+    async function makeCrashPool(size: number): Promise<ShaderTransitionWorkerPool> {
+      const p = await createShaderTransitionWorkerPool({ size, workerEntryPath: CRASH_WORKER });
+      pools.push(p);
+      return p;
+    }
+
+    function blendReq(): Parameters<ShaderTransitionWorkerPool["run"]>[0] {
+      return {
+        shader: "crossfade",
+        bufferA: fillSolid(WIDTH, HEIGHT, 0, 0, 0),
+        bufferB: fillSolid(WIDTH, HEIGHT, 1, 1, 1),
+        output: Buffer.alloc(BUF_SIZE),
+        width: WIDTH,
+        height: HEIGHT,
+        progress: 0.5,
+      };
+    }
+
+    // Resolves to "hung" if `p` doesn't settle within `ms`. The crash paths
+    // settle near-instantly; "hung" only appears if the regression (dispatch
+    // to a dead worker) comes back.
+    function settledWithin(p: Promise<unknown>, ms = 3000): Promise<string> {
+      return Promise.race([
+        p.then(
+          () => "resolved",
+          () => "rejected",
+        ),
+        new Promise<string>((r) => setTimeout(() => r("hung"), ms)),
+      ]);
+    }
+
+    it("rejects the in-flight task when its only worker crashes, then fails subsequent runs fast", async () => {
+      const pool = await makeCrashPool(1);
+      // The worker throws on receipt: the in-flight task must reject, not hang.
+      expect(await settledWithin(pool.run(blendReq()))).toBe("rejected");
+      // The slot is now dead. A later run must fail fast rather than dispatch
+      // to the terminated worker (postMessage there is a silent no-op → hang).
+      expect(await settledWithin(pool.run(blendReq()))).toBe("rejected");
+    });
+
+    it("rejects a queued task on crash instead of leaving it to hang", async () => {
+      const pool = await makeCrashPool(1);
+      // First occupies the single worker (which crashes); second is queued
+      // behind it. When the worker dies, both must settle — never hang.
+      const inFlight = settledWithin(pool.run(blendReq()));
+      const queued = settledWithin(pool.run(blendReq()));
+      expect(await inFlight).toBe("rejected");
+      expect(await queued).toBe("rejected");
+    });
+
+    it("never wedges the pool when all slots die", async () => {
+      // Size 2: both slots run the crashing fixture, so there are no surviving
+      // workers; the pool must still never wedge — every run settles.
+      const pool = await makeCrashPool(2);
+      const results = await Promise.all([
+        settledWithin(pool.run(blendReq())),
+        settledWithin(pool.run(blendReq())),
+        settledWithin(pool.run(blendReq())),
+      ]);
+      expect(results).not.toContain("hung");
+    });
+  });
 });

packages/producer/src/services/shaderTransitionWorkerPool.ts

@@ -99,6 +99,15 @@ interface WorkerSlot {
   worker: Worker;
   busy: boolean;
   current: PendingTask | null;
+  /**
+   * Set once the worker has crashed (`error`) or exited unexpectedly. A dead
+   * slot must never be dispatched to again: `postMessage` to a terminated
+   * Worker is a silent no-op (no throw, no reply), so a task routed to it
+   * would hang forever. The pool does not respawn mid-render (the lost
+   * transferList buffers can't be reconstructed), so a dead slot stays dead
+   * until teardown.
+   */
+  dead: boolean;
 }
 
 interface WorkerReply {
@@ -214,8 +223,18 @@ export async function createShaderTransitionWorkerPool(
   // appends one so the `.ts` worker entry still loads.
   const execArgv = buildExecArgv(entryIsTs);
 
+  // When every worker has died there is no live thread left to drain the
+  // queue, so any waiting tasks would hang forever. Reject them instead.
+  const failQueueIfNoLiveSlots = (): void => {
+    if (slots.some((s) => !s.dead)) return;
+    while (queue.length > 0) {
+      const t = queue.shift();
+      if (t) t.reject(new Error("shader-blend pool has no live workers; task abandoned"));
+    }
+  };
+
   const dispatchNext = (slot: WorkerSlot): void => {
-    if (terminated || slot.busy) return;
+    if (terminated || slot.busy || slot.dead) return;
     const task = queue.shift();
     if (!task) return;
     slot.busy = true;
@@ -289,6 +308,11 @@ export async function createShaderTransitionWorkerPool(
     const task = slot.current;
     slot.current = null;
     slot.busy = false;
+    // Mark dead before rejecting and before draining the queue: this slot's
+    // worker can no longer accept a dispatch (postMessage would be a silent
+    // no-op), so it must be excluded from future slot selection or a later
+    // task would hang on it.
+    slot.dead = true;
     if (task) {
       // The in-flight task's buffers were transferred to the worker. They're
       // lost on the worker crash — the caller's original Buffers are
@@ -297,20 +321,24 @@ export async function createShaderTransitionWorkerPool(
       task.reject(new Error(`shader-blend worker crashed mid-task: ${err.message}; buffers lost`));
     }
     log.warn?.("[shaderTransitionWorkerPool] worker errored", { err: err.message });
+    failQueueIfNoLiveSlots();
   };
 
   const onWorkerExit = (slot: WorkerSlot, code: number): void => {
     if (terminated) return;
-    // Unexpected exit — fail any in-flight task and drop the slot. We don't
-    // auto-respawn in the middle of a render because the lost transferList
-    // buffers can't be reconstructed, and silently shrinking the pool would
-    // mask the real failure. Pool teardown handles graceful shutdown.
+    // Unexpected exit — fail any in-flight task and mark the slot dead. We
+    // don't auto-respawn in the middle of a render because the lost
+    // transferList buffers can't be reconstructed, and silently shrinking the
+    // pool would mask the real failure. Pool teardown handles graceful
+    // shutdown.
+    slot.dead = true;
     if (slot.current) {
       slot.current.reject(new Error(`shader-blend worker exited (code=${code}) mid-task`));
       slot.current = null;
       slot.busy = false;
     }
     log.warn?.("[shaderTransitionWorkerPool] worker exited unexpectedly", { code });
+    failQueueIfNoLiveSlots();
   };
 
   // Spawn workers. If any throws synchronously we still want to terminate
@@ -318,7 +346,7 @@ export async function createShaderTransitionWorkerPool(
   try {
     for (let i = 0; i < size; i++) {
       const worker = new Worker(entry, { execArgv });
-      const slot: WorkerSlot = { worker, busy: false, current: null };
+      const slot: WorkerSlot = { worker, busy: false, current: null, dead: false };
       worker.on("message", (msg: WorkerReply) => onWorkerMessage(slot, msg));
       worker.on("error", (err: unknown) =>
         onWorkerError(slot, err instanceof Error ? err : new Error(String(err))),
@@ -344,13 +372,18 @@ export async function createShaderTransitionWorkerPool(
         const task: PendingTask = traceEnabled
           ? { req, resolve, reject, enqueuedAtMs: Date.now(), traceId: ++nextTaskId }
           : { req, resolve, reject };
-        // Find an idle slot; otherwise queue.
-        const idle = slots.find((s) => !s.busy);
+        // Find a live idle slot; otherwise queue behind the live busy ones.
+        const idle = slots.find((s) => !s.busy && !s.dead);
         if (idle) {
           queue.unshift(task);
           dispatchNext(idle);
-        } else {
+        } else if (slots.some((s) => !s.dead)) {
+          // A live worker is busy; it drains the queue when it completes.
           queue.push(task);
+        } else {
+          // Every worker has died — don't hang waiting for a dispatch that
+          // can never happen.
+          reject(new Error("shader-blend pool has no live workers"));
         }
       });
     },