From 5deef4c0e4c906d4e1500739761c95c863eb0e2f Mon Sep 17 00:00:00 2001
From: Brandur <brandur@brandur.org>
Date: Mon, 5 Oct 2026 23:10:07 -0500
Subject: [PATCH] Fix intermittently failing test

Just noticed this test case was failing intermittently on:
https://github.com/riverqueue/river/pull/1451

Problem: the test case scheduled a `setImmediate` callback that'd wait
for 100 ms and exit the thread. After that handler returned, the test
submitted a second task, expecting the thread to crash before accepting
it, but it was possible for the second task's message to arrive and be
proceed before the `setImmediate` callback ran.

Replace the timing-dependent setup with a test handler that replaces the
thread's message listener and exits immediately when the next task
arrives, guaranteeing the reused thread dies before acknowledging it,
and exercising the pool's retry path deterministically. A side benefit
is that we kill the 100 ms busy wait.
---
 js/worker-threads/src/index.test.ts        |  8 +++++---
 js/worker-threads/src/testdata/handlers.ts | 23 ++++++++++------------
 2 files changed, 15 insertions(+), 16 deletions(-)

diff --git a/js/worker-threads/src/index.test.ts b/js/worker-threads/src/index.test.ts
index 8a3026aa0..76f70317b 100644
--- a/js/worker-threads/src/index.test.ts
+++ b/js/worker-threads/src/index.test.ts
@@ -781,10 +781,13 @@ describe("WorkerThreads", () => {
       }
     );
 
-    it("retries a task once when its reused thread dies before starting it", async () => {
+    it("retries a task once when its reused thread dies before starting it", async ({
+      onTestFinished,
+    }) => {
       const executor = new WorkerThreads({ maxThreads: 1 });
+      onTestFinished(() => executor.close());
       const exiting = executor.handler(testJob, {
-        exportName: "blockThenExit",
+        exportName: "exitBeforeNextTask",
         module: handlers,
       });
       const complete = executor.handler(testJob, {
@@ -805,7 +808,6 @@ describe("WorkerThreads", () => {
         crashedThreads: 1,
         totalThreads: 1,
       });
-      await executor.close();
     });
 
     it("fails an attempt that exceeds its thread's heap limit", async () => {
diff --git a/js/worker-threads/src/testdata/handlers.ts b/js/worker-threads/src/testdata/handlers.ts
index 360191491..c69f81e2d 100644
--- a/js/worker-threads/src/testdata/handlers.ts
+++ b/js/worker-threads/src/testdata/handlers.ts
@@ -5,6 +5,8 @@
  * the TypeScript source, so every test also exercises the thread's fallback
  * from a missing `.js` module to its `.ts` source.
  */
+import { parentPort } from "node:worker_threads";
+
 import {
   complete as completeOutcome,
   snooze as snoozeOutcome,
@@ -22,19 +24,6 @@ export const allocateForever: TestHandler = ({ logger }) => {
   for (;;) retained.push(new Array<number>(100_000).fill(retained.length));
 };
 
-export const blockThenExit: TestHandler = () => {
-  // Keep the thread busy after returning so the next run message is queued
-  // behind this callback, then exit before the thread can acknowledge it.
-  setImmediate(() => {
-    const end = Date.now() + 100;
-    while (Date.now() < end) {
-      // Deliberately block this isolated thread.
-    }
-    process.exit(9);
-  });
-  return completeOutcome();
-};
-
 export const complete: TestHandler = ({ job }) =>
   completeOutcome({ output: { value: job.args.value ?? null } });
 
@@ -93,6 +82,14 @@ export const echoExact: TestHandler = ({
   return completeOutcome({ output: { id } });
 };
 
+export const exitBeforeNextTask: TestHandler = () => {
+  // Replace the run listener so the next task kills this thread before it
+  // can acknowledge the task. A setImmediate callback races with that message.
+  parentPort?.removeAllListeners("message");
+  parentPort?.once("message", () => process.exit(9));
+  return completeOutcome();
+};
+
 export const exitWhenIdle: TestHandler = () => {
   setImmediate(() => process.exit(7));
   return completeOutcome();