cluster: fix two ways a Windows worker stops reacting (shared accept() race, early handle ack); fixes test-cluster-shared-leak timeouts

src/js/internal/cluster/primary.ts

@@ -216,6 +216,16 @@ function exitedAfterDisconnect(worker, message) {
   send(worker, { ack: message.seq });
 }
 
+// Windows cannot accept from copies of one listening socket in two processes: the worker losing the
+// race for a connection blocks inside accept() despite its copy being non-blocking, and libuv's
+// exclusive AFD polls from two processes cancel each other in a loop (node uses AcceptEx there).
+// So on Windows the primary distributes TCP connections under every policy; UDP and TLS always share.
+function usesRoundRobinHandle(message) {
+  if (message.sharedOnly === true || message.addressType === "udp4" || message.addressType === "udp6") return false;
+  if (schedulingPolicy === SCHED_RR) return true;
+  return process.platform === "win32" && (message.addressType === 4 || message.addressType === 6);
+}
+
 function queryServer(worker, message) {
   // Stop processing if worker already disconnecting
   if (worker.exitedAfterDisconnect) return;
@@ -231,21 +241,13 @@ function queryServer(worker, message) {
   if (cachedHandle && !cachedHandle.has(worker)) handle = cachedHandle;
 
   const kSharedOnlyHint =
-    "TLS and non-TLS cluster workers cannot share the same address:port under SCHED_RR " +
-    "(Bun's TLS accept is native and cannot adopt round-robin connection fds)";
+    "TLS and non-TLS cluster workers cannot share the same address:port while the primary distributes its connections " +
+    "(Bun's TLS accept is native and cannot adopt handed-off connections); listen on different ports";
   if (handle !== undefined && message.sharedOnly === true && handle instanceof RoundRobinHandle) {
     send(worker, { errno: UV_EINVAL, key, ack: message.seq, data: handle.data, bunHint: kSharedOnlyHint }, null);
     return;
   }
-  if (
-    schedulingPolicy === SCHED_RR &&
-    handle !== undefined &&
-    message.sharedOnly !== true &&
-    handle instanceof SharedHandle &&
-    handle.sharedOnly &&
-    message.addressType !== "udp4" &&
-    message.addressType !== "udp6"
-  ) {
+  if (handle instanceof SharedHandle && handle.sharedOnly && usesRoundRobinHandle(message)) {
     send(worker, { errno: UV_EINVAL, key, ack: message.seq, data: handle.data, bunHint: kSharedOnlyHint }, null);
     return;
   }
@@ -260,25 +262,17 @@ function queryServer(worker, message) {
       if (message.address.length < address.length) address = message.address;
     }
 
-    // UDP is exempt from round-robin connection balancing for what should
-    // be obvious reasons: it's connectionless. There is nothing to send to
-    // the workers except raw datagrams and that's pointless.
     if (process.platform === "win32" && (message.addressType === "udp4" || message.addressType === "udp6")) {
       const error = new Error(`write ENOTSUP - cannot share a dgram socket with a worker on Windows`);
       error.code = "ENOTSUP";
       error.syscall = "write";
       worker.emit("error", error);
       return;
     }
-    if (
-      schedulingPolicy !== SCHED_RR ||
-      message.sharedOnly === true ||
-      message.addressType === "udp4" ||
-      message.addressType === "udp6"
-    ) {
-      handle = new SharedHandle(key, address, message);
-    } else {
+    if (usesRoundRobinHandle(message)) {
       handle = new RoundRobinHandle(key, address, message);
+    } else {
+      handle = new SharedHandle(key, address, message);
     }
 
     if (!cachedHandle) handles.set(key, handle);

src/runtime/ipc.rs

@@ -922,6 +922,9 @@ pub struct SendQueue {
     root: Cell<Option<core::ptr::NonNull<SendQueue>>>,
     pub(crate) queue: JsCell<Vec<SendHandle>>,
     pub(crate) waiting_for_ack: JsCell<Option<SendHandle>>,
+    /// Reply to a handle message read before libuv reported that message's own write complete;
+    /// `on_write_complete` (which moves the message into `waiting_for_ack`) applies it.
+    early_ack_nack: Cell<Option<AckNack>>,
 
     pub(crate) retry_count: Cell<u32>,
     pub(crate) keep_alive: JsCell<KeepAlive>,
@@ -1046,6 +1049,7 @@ impl SendQueue {
             root: Cell::new(None),
             queue: JsCell::new(Vec::new()),
             waiting_for_ack: JsCell::new(None),
+            early_ack_nack: Cell::new(None),
             retry_count: Cell::new(0),
             keep_alive: JsCell::new(KeepAlive::default()),
             #[cfg(debug_assertions)]
@@ -1355,7 +1359,18 @@ impl SendQueue {
             Some(item) => Some(item.handle.is_some()),
         });
         let Some(has_handle) = waiting else {
-            log!("onAckNack: ack received but not waiting for ack");
+            let handle_write_in_progress = self.write_in_progress.get()
+                && self
+                    .queue
+                    .get()
+                    .first()
+                    .is_some_and(|first| first.handle.is_some());
+            if handle_write_in_progress {
+                log!("onAckNack: ack/nack arrived before the handle message's write completed");
+                self.early_ack_nack.set(Some(ack_nack));
+            } else {
+                log!("onAckNack: ack received but not waiting for ack");
+            }
             return;
         };
         if !has_handle {
@@ -1547,6 +1562,7 @@ impl SendQueue {
             return;
         }
         self.write_in_progress.set(false);
+        let early_ack_nack = self.early_ack_nack.take();
         let global_this = self.get_global_this();
 
         enum Done {
@@ -1590,9 +1606,10 @@ impl SendQueue {
             }
         });
         match done {
-            Done::AwaitAck => {
-                self.continue_send(&global_this, ContinueSendReason::OnWritable);
-            }
+            Done::AwaitAck => match early_ack_nack {
+                Some(ack_nack) => self.on_ack_nack(&global_this, ack_nack),
+                None => self.continue_send(&global_this, ContinueSendReason::OnWritable),
+            },
             Done::Completed(item) => {
                 item.complete(&global_this); // call the callback & deinit
                 self.continue_send(&global_this, ContinueSendReason::OnWritable);

test/js/node/cluster.test.ts

@@ -204,7 +204,7 @@ if (cluster.isPrimary) {
   const { stdout } = await bunRun(joinP(dir, "main.ts"), bunEnv);
   expect(stdout).toContain("tls listen error code: EINVAL");
   expect(stdout).toContain("TLS and non-TLS cluster workers cannot share");
-});
+}, 30_000);
 
 test("cluster pipe listen error carries no port suffix", async () => {
   const dir = tempDirWithFiles("bun-test", {
@@ -412,6 +412,8 @@ if (cluster.isPrimary) {
   expect(stdout).toContain("ipv6 connect ok");
 });
 
+// A SharedHandle binds its socket natively; only a RoundRobinHandle makes the primary itself
+// listen() with a net.Server. Windows serves SCHED_NONE TCP listens round-robin, see primary.ts.
 test("SCHED_NONE: a second worker listens on the same shared handle", async () => {
   const dir = tempDirWithFiles("bun-test", {
     "main.ts": `
@@ -421,6 +423,12 @@ const net = require("node:net");
 cluster.schedulingPolicy = cluster.SCHED_NONE;
 
 if (cluster.isPrimary) {
+  let primaryListenCalls = 0;
+  const listen = net.Server.prototype.listen;
+  net.Server.prototype.listen = function () {
+    primaryListenCalls++;
+    return listen.apply(this, arguments);
+  };
   const workers = [cluster.fork(), cluster.fork()];
   let listening = 0;
   const ports = new Set();
@@ -429,6 +437,7 @@ if (cluster.isPrimary) {
     ports.add(address.port);
     if (++listening !== 2) return;
     console.log("listening workers:", listening, "distinct ports:", ports.size);
+    console.log("primary listen() calls:", primaryListenCalls);
     for (const w of workers) w.kill();
     process.exit(0);
   });
@@ -449,8 +458,114 @@ if (cluster.isPrimary) {
   const { stdout } = await bunRun(joinP(dir, "main.ts"), bunEnv);
   expect(stdout).toContain("policy is SCHED_NONE: true");
   expect(stdout).toContain("listening workers: 2 distinct ports: 1");
+  expect(stdout).toContain(`primary listen() calls: ${isWindows ? 1 : 0}`);
 });
 
+// Every connection below is one chance for a worker to stop answering IPC; the primary pings both
+// workers after each one and the workers must still disconnect cleanly at the end.
+//
+// SCHED_NONE: two workers accepting on copies of one Windows listening socket race each other for
+// every connection, and the loser blocked inside accept(), so its event loop never ran again. TCP
+// listens under SCHED_NONE are served by the primary on Windows now.
+//
+// SCHED_RR: the primary sends each connection as a handle message. On Windows the worker's ack for
+// it could be read before libuv reported the write itself complete; the primary then dropped the ack
+// and waited for it forever, and every later message to that worker (pings, disconnect) sat behind
+// it. This ordering needs a busy machine, so before the fix this variant only failed under load.
+for (const policy of ["SCHED_NONE", "SCHED_RR"]) {
+  test.skipIf(!isWindows)(
+    `${policy}: two workers on one port keep answering IPC while connections arrive, then exit`,
+    async () => {
+      using dir = tempDir("cluster-worker-responsiveness", {
+        "main.ts": `
+const cluster = require("node:cluster");
+const net = require("node:net");
+cluster.schedulingPolicy = cluster.${policy};
+
+if (cluster.isPrimary) {
+  const CONNECTIONS = 100;
+  const workers = [cluster.fork(), cluster.fork()];
+  const ports = new Set();
+  const exits = [];
+  let listening = 0;
+
+  function fail(message) {
+    console.log(message);
+    for (const worker of workers) worker.process.kill();
+    process.exit(1);
+  }
+
+  cluster.on("listening", (worker, address) => {
+    ports.add(address.port);
+    if (++listening !== workers.length) return;
+    run(address.port).catch(err => fail("error: " + err.message));
+  });
+
+  cluster.on("exit", (worker, code, signal) => {
+    exits.push(code ?? signal);
+    if (exits.length === workers.length) console.log("exits:", exits.join(","));
+  });
+
+  function connectOnce(port) {
+    return new Promise((resolve, reject) => {
+      const c = net.connect(port, "127.0.0.1");
+      let data = "";
+      c.setEncoding("utf8");
+      c.on("data", chunk => (data += chunk));
+      c.on("error", reject);
+      c.on("close", () => (data === "hi" ? resolve() : reject(new Error("got " + JSON.stringify(data)))));
+    });
+  }
+
+  function pingWorkers(connection) {
+    return Promise.all(
+      workers.map(worker => {
+        return new Promise(resolve => {
+          const stuck = setTimeout(() => fail("worker " + worker.id + " stopped answering after connection " + connection), 10_000);
+          worker.once("message", () => {
+            clearTimeout(stuck);
+            resolve();
+          });
+          worker.send("ping");
+        });
+      }),
+    );
+  }
+
+  async function run(port) {
+    for (let connection = 1; connection <= CONNECTIONS; connection++) {
+      await connectOnce(port);
+      await pingWorkers(connection);
+    }
+    console.log("served:", CONNECTIONS, "ports:", ports.size);
+    for (const worker of workers) worker.disconnect();
+  }
+} else {
+  net.createServer(socket => socket.end("hi")).listen(0, "127.0.0.1");
+  process.on("message", message => {
+    if (message === "ping") process.send("pong");
+  });
+}
+`,
+      });
+      await using proc = Bun.spawn({
+        cmd: [bunExe(), "main.ts"],
+        env: bunEnv,
+        cwd: String(dir),
+        stdout: "pipe",
+        stderr: "pipe",
+      });
+      const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+      expect({ stdout, stderr, exitCode }).toEqual({
+        stdout: "served: 100 ports: 1\nexits: 0,0\n",
+        stderr: expect.any(String),
+        exitCode: 0,
+      });
+    },
+    30_000,
+  );
+}
+
 test("SCHED_NONE: close() releases the shared handle so the worker can re-listen on the same port", async () => {
   using dir = tempDir("cluster-shared-relisten", {
     "main.ts": `
@@ -807,11 +922,10 @@ if (cluster.isPrimary) {
   expect(stdout).toContain("reply: echo:hi");
 }, 30_000);
 
-test("plain worker listening on a key already owned by a TLS shared-only handle fails with EINVAL", async () => {
-  const dir = tempDirWithFiles("bun-test", {
-    "cert.pem": tlsCerts.cert,
-    "key.pem": tlsCerts.key,
-    "main.ts": `
+const netWorkerAfterTlsWorkerFixture = {
+  "cert.pem": tlsCerts.cert,
+  "key.pem": tlsCerts.key,
+  "main.ts": `
 const cluster = require("node:cluster");
 const net = require("node:net");
 const tls = require("node:tls");
@@ -824,11 +938,18 @@ if (cluster.isPrimary) {
   const tlsWorker = cluster.fork({ ROLE: "tls" });
   cluster.once("listening", () => {
     const netWorker = cluster.fork({ ROLE: "net" });
-    netWorker.on("message", msg => {
-      console.log("net listen error code:", msg.code, msg.msg);
+    const done = () => {
       tlsWorker.kill();
       netWorker.kill();
       process.exit(0);
+    };
+    netWorker.on("message", msg => {
+      console.log("net listen error code:", msg.code, msg.msg);
+      done();
+    });
+    netWorker.on("listening", () => {
+      console.log("net worker is listening on the TLS worker's socket");
+      done();
     });
   });
 } else if (process.env.ROLE === "tls") {
@@ -839,12 +960,31 @@ if (cluster.isPrimary) {
   server.listen(0);
 }
 `,
-  });
+};
+
+test("plain worker listening on a key already owned by a TLS shared-only handle fails with EINVAL", async () => {
+  const dir = tempDirWithFiles("bun-test", netWorkerAfterTlsWorkerFixture);
   const { stdout } = await bunRun(joinP(dir, "main.ts"), bunEnv);
   expect(stdout).toContain("net listen error code: EINVAL");
   expect(stdout).toContain("TLS and non-TLS cluster workers cannot share");
 }, 30_000);
 
+// Under SCHED_NONE the plain worker shares the TLS worker's socket, as in node. On Windows the primary
+// serves TCP listens under SCHED_NONE too (two workers accepting on copies of one socket is what hangs
+// a worker there), so it is refused there under either policy.
+test("SCHED_NONE: plain worker listening on a key owned by a TLS shared-only handle", async () => {
+  const dir = tempDirWithFiles("bun-test", netWorkerAfterTlsWorkerFixture);
+  const { stdout } = await bunRun(joinP(dir, "main.ts"), { NODE_CLUSTER_SCHED_POLICY: "none" });
+  if (isWindows) {
+    expect(stdout).toContain("net listen error code: EINVAL");
+    expect(stdout).toContain(
+      "TLS and non-TLS cluster workers cannot share the same address:port while the primary distributes its connections",
+    );
+  } else {
+    expect(stdout).toBe("net worker is listening on the TLS worker's socket");
+  }
+}, 30_000);
+
 test.skipIf(isWindows)(
   "SCHED_NONE listen({fd:2}) fails EINVAL like node and does not close the primary's stderr",
   async () => {