cluster: layering, protocol, and coverage follow-ups from PR review

- primary/RoundRobinHandle: translate listen errors to a uv-domain errno
  (uvTranslateSysError) at source and drop the Bun-only 'errcode' string
  from the wire protocol; net.ts uses ExceptionWithHostPort like Node.
- primary: reuse enobufs/einval bindings for the numeric errnos so
  getSystemErrorName(ex.errno) matches ex.code on every platform.
- primary/SharedHandle/RoundRobinHandle: port the rest of nodejs/node#60141
  (cachedHandle + has(worker)) so a worker re-asking for a key it already
  holds gets a fresh handle that EADDRINUSEs; re-enable the debug asserts.
- primary: replace the Windows SCHED_RR '// TODO' with the accurate reason
  (TCP SCHED_NONE works via WSADuplicateSocketW; keeping RR default until
  named-pipe DuplicateHandle export lands).
- SharedHandle: refuse a plain worker joining a TLS shared-only key (the
  symmetric case of the existing EINVAL guard) and skip Linux abstract-
  socket paths from the pipe-unlink cleanup.
- node_cluster_binding: use bun_sys wrappers for cloexec/nonblock/close
  instead of raw libc.
- usockets: us_create_udp_socket_from_fd surfaces the poll-registration
  errno (matching us_socket_group_listen_fd) via a new *err out-param.
- tests: TLS cluster worker happy path, TLS-first mixed-key EINVAL,
  round-robin byte-0 preservation, RR socket state (connecting=false /
  remoteAddress), SCHED_NONE listen({fd:2}) leaves the primary's stderr
  open, and send(socket, {keepOpen: true}) keeps the sender's copy live.

Author: alii

src/js/internal/cluster/RoundRobinHandle.ts

@@ -5,6 +5,7 @@ let net;
 
 const sendHelper = $newRustFunction("node_cluster_binding.rs", "sendHelperPrimary", 4);
 const uvTranslateSysError = $newRustFunction("node_util_binding.rs", "uvTranslateSysError", 1);
+const einvalErrorCode = $newRustFunction("node_util_binding.rs", "einvalErrorCode", 0);
 
 const ArrayIsArray = Array.isArray;
 
@@ -91,8 +92,8 @@ export default class RoundRobinHandle {
       // getSystemErrorName) expects. On Windows the WSA code goes through
       // uv_translate_sys_error so `getSystemErrorName(errno)` matches
       // `err.code` — same as node's send(err.errno, null).
-      const errno = uvTranslateSysError(typeof err.errno === "number" ? err.errno : 0) || uvTranslateSysError(-1);
-      send(errno, null, null);
+      const raw = typeof err.errno === "number" && err.errno !== 0 ? err.errno : null;
+      send(raw != null ? uvTranslateSysError(raw) : einvalErrorCode(), null, null);
     });
   }
 

src/jsc/ipc.rs

@@ -1510,7 +1510,7 @@ impl SendQueue {
         // close_on_complete and re-fire the callback with null.
         let will_push_fresh = handle.is_some()
             || self.queue.is_empty()
-            || self.queue.last().map_or(true, |l| {
+            || self.queue.last().is_none_or(|l| {
                 l.handle.is_some()
                     || l.is_ack_nack()
                     || (self.queue.len() == 1 && self.write_in_progress)

src/uws_sys/udp.rs

@@ -69,7 +69,7 @@ impl Socket {
                 close_cb,
                 recv_error_cb,
                 fd,
-                err.map_or(core::ptr::null_mut(), |e| e as *mut _),
+                err.map_or(core::ptr::null_mut(), core::ptr::from_mut),
                 user_data,
             )
         }

test/js/node/child_process/child_process_ipc_handle.test.ts

@@ -279,4 +279,70 @@ process.on('message', (m, server) => {
         response: true,
       });
     });
+
+  // send(msg, socket, {keepOpen: true}): both parent and child hold a live
+  // dup of the connection. Node's test/parallel/test-child-process-send-keep-open.js.
+  test.concurrent("net.Socket handle sent with {keepOpen: true} stays open in the sender", async () => {
+    using dir = tempDir("ipc-handle-keepopen", {
+      "parent.js": `
+const { fork } = require('node:child_process');
+const net = require('node:net');
+
+const child = fork('child.js');
+let closed = false;
+const server = net.createServer(socket => {
+  socket.on('close', () => { closed = true; });
+  child.send('socket', socket, { keepOpen: true }, err => {
+    if (err) return finish(false, 'send:' + err.message);
+    // The parent's copy must still be usable after the ack.
+    socket.write('parent', () => {});
+  });
+  child.on('message', m => {
+    if (m !== 'child-wrote') return;
+    // Only end after the child has also written.
+    setTimeout(() => {
+      if (closed) return finish(false, 'parent socket closed by keepOpen send');
+      socket.end();
+    }, 50);
+  });
+}).listen(0, '127.0.0.1', () => {
+  const client = net.connect(server.address().port, '127.0.0.1');
+  client.setEncoding('utf8');
+  let data = '';
+  client.on('data', c => (data += c));
+  client.on('end', () => finish(data.includes('parent') && data.includes('child'), data));
+  client.on('error', e => finish(false, 'client:' + e.message));
+});
+
+function finish(ok, detail) {
+  console.log(ok ? 'RESPONSE:' + detail : 'FAILED:' + detail);
+  try { child.kill(); } catch {}
+  try { server.close(); } catch {}
+  process.exit(ok ? 0 : 1);
+}
+`,
+      "child.js": `
+const net = require('node:net');
+process.on('message', (m, socket) => {
+  if (!(socket instanceof net.Socket)) return process.send({ error: 'handle was ' + typeof socket });
+  socket.write('child', () => process.send('child-wrote'));
+});
+`,
+    });
+
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "parent.js"],
+      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({ exitCode, stderr, hasParent: stdout.includes("parent"), hasChild: stdout.includes("child") }).toEqual({
+      exitCode: 0,
+      stderr: expect.any(String),
+      hasParent: true,
+      hasChild: true,
+    });
+  });
 });

test/js/node/cluster.test.ts

@@ -1,5 +1,5 @@
 import { expect, test } from "bun:test";
-import { bunEnv, bunExe, bunRun, isIPv6, isWindows, joinP, tempDirWithFiles } from "harness";
+import { bunEnv, bunExe, bunRun, isIPv6, isWindows, joinP, tempDirWithFiles, tls as tlsCerts } from "harness";
 
 test("cloneable and transferable equals", () => {
   const dir = tempDirWithFiles("bun-test", {
@@ -556,3 +556,223 @@ test.each(["net", "http"])("cluster 'listening' reports the address a %s server
     if (!("path" in target)) expect(payloads[i].port).toBeWithin(1, 65536);
   }
 });
+
+test("round-robin worker connection socket has connecting=false and remoteAddress synchronously", () => {
+  const dir = tempDirWithFiles("bun-test", {
+    "main.ts": `
+const cluster = require("node:cluster");
+const net = require("node:net");
+
+if (cluster.isPrimary) {
+  const worker = cluster.fork();
+  worker.on("message", m => {
+    console.log(JSON.stringify(m));
+    worker.kill();
+    process.exit(0);
+  });
+  cluster.on("listening", (w, address) => {
+    net.connect(address.port, "127.0.0.1").on("error", () => {});
+  });
+} else {
+  net
+    .createServer(socket => {
+      // Captured synchronously in the connection listener: node's onconnection
+      // delivers accepted sockets already open, not connecting.
+      process.send({
+        connecting: socket.connecting,
+        readyState: socket.readyState,
+        remote: typeof socket.remoteAddress,
+      });
+      socket.end();
+    })
+    .listen(0, "127.0.0.1");
+}
+`,
+  });
+  const { stdout } = bunRun(joinP(dir, "main.ts"), bunEnv);
+  const m = JSON.parse(stdout.trim());
+  expect(m.connecting).toBe(false);
+  expect(m.readyState).toBe("open");
+  expect(m.remote).toBe("string");
+});
+
+test("round-robin: primary never consumes accepted-socket bytes before handoff", () => {
+  const dir = tempDirWithFiles("bun-test", {
+    "main.ts": `
+const cluster = require("node:cluster");
+const net = require("node:net");
+
+const N = 20;
+if (cluster.isPrimary) {
+  const worker = cluster.fork();
+  let got = 0;
+  worker.on("message", m => {
+    console.log(m);
+    if (++got === N) {
+      worker.kill();
+      process.exit(0);
+    }
+  });
+  cluster.on("listening", (w, address) => {
+    for (let i = 0; i < N; i++) {
+      const c = net.connect(address.port, "127.0.0.1", () => {
+        // Write immediately on connect: with pauseOnConnect at the primary
+        // accept, byte 0 must reach the worker, not the primary's Duplex.
+        c.write("MAGIC-" + i + "-" + "x".repeat(4096));
+        c.end();
+      });
+      c.on("error", () => {});
+    }
+  });
+} else {
+  net
+    .createServer(sock => {
+      let buf = "";
+      sock.on("data", d => (buf += d));
+      sock.on("end", () => process.send(buf.slice(0, 20) + " " + buf.length));
+    })
+    .listen(0, "127.0.0.1");
+}
+`,
+  });
+  const { stdout } = bunRun(joinP(dir, "main.ts"), bunEnv);
+  const lines = stdout.trim().split("\n").sort();
+  expect(lines.length).toBe(20);
+  for (const line of lines) {
+    expect(line).toMatch(/^MAGIC-\d+-x+ 41\d\d$/);
+  }
+});
+
+test("TLS cluster worker under SCHED_RR listens on a shared handle and completes handshakes", () => {
+  // Two forked workers + a TLS handshake in a debug build is well over the
+  // default 5s test budget.
+  const dir = tempDirWithFiles("bun-test", {
+    "cert.pem": tlsCerts.cert,
+    "key.pem": tlsCerts.key,
+    "main.ts": `
+const cluster = require("node:cluster");
+const tls = require("node:tls");
+const fs = require("node:fs");
+const path = require("node:path");
+const key = fs.readFileSync(path.join(__dirname, "key.pem"));
+const cert = fs.readFileSync(path.join(__dirname, "cert.pem"));
+
+if (cluster.isPrimary) {
+  const w1 = cluster.fork();
+  const w2 = cluster.fork();
+  const ports = new Set();
+  let listening = 0;
+  cluster.on("listening", (w, address) => {
+    ports.add(address.port);
+    if (++listening !== 2) return;
+    // Both workers must share the primary-bound port under SCHED_RR TLS.
+    console.log("distinct ports:", ports.size);
+    const port = address.port;
+    const c = tls.connect({ port, host: "127.0.0.1", rejectUnauthorized: false }, () => {
+      c.write("hi");
+    });
+    c.setEncoding("utf8");
+    c.on("data", d => {
+      console.log("reply:", d);
+      c.end();
+      w1.kill();
+      w2.kill();
+      process.exit(0);
+    });
+    c.on("error", e => {
+      console.log("client error:", e.code);
+      process.exit(1);
+    });
+  });
+} else {
+  tls
+    .createServer({ key, cert }, socket => {
+      socket.on("data", d => socket.end("echo:" + d));
+    })
+    .listen(0);
+}
+`,
+  });
+  const { stdout } = bunRun(joinP(dir, "main.ts"), bunEnv);
+  expect(stdout).toContain("distinct ports: 1");
+  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", () => {
+  const dir = tempDirWithFiles("bun-test", {
+    "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");
+const fs = require("node:fs");
+const path = require("node:path");
+const key = fs.readFileSync(path.join(__dirname, "key.pem"));
+const cert = fs.readFileSync(path.join(__dirname, "cert.pem"));
+
+if (cluster.isPrimary) {
+  // Reverse of the existing test: TLS worker claims first, plain worker second.
+  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);
+      tlsWorker.kill();
+      netWorker.kill();
+      process.exit(0);
+    });
+  });
+} else if (process.env.ROLE === "tls") {
+  tls.createServer({ key, cert }, () => {}).listen(0);
+} else {
+  const server = net.createServer(() => {});
+  server.on("error", err => process.send({ code: err.code }));
+  server.listen(0);
+}
+`,
+  });
+  const { stdout } = bunRun(joinP(dir, "main.ts"), bunEnv);
+  expect(stdout).toContain("net listen error code: EINVAL");
+}, 30_000);
+
+test.skipIf(isWindows)("SCHED_NONE listen({fd:2}) fails ENOTSOCK and does not close the primary's stderr", () => {
+  const dir = tempDirWithFiles("bun-test", {
+    "main.ts": `
+const cluster = require("node:cluster");
+const net = require("node:net");
+const fs = require("node:fs");
+
+cluster.schedulingPolicy = cluster.SCHED_NONE;
+
+if (cluster.isPrimary) {
+  const worker = cluster.fork();
+  worker.on("message", m => {
+    console.log("worker error code:", m.code);
+    worker.disconnect();
+  });
+  cluster.on("exit", () => {
+    // stderr fd must still be a valid open fd in the primary.
+    try {
+      fs.fstatSync(2);
+      console.log("stderr open: true");
+    } catch (e) {
+      console.log("stderr open: false");
+    }
+    process.exit(0);
+  });
+} else {
+  const server = net.createServer(() => {});
+  server.on("error", err => {
+    process.send({ code: err.code });
+  });
+  server.listen({ fd: 2 });
+}
+`,
+  });
+  const { stdout } = bunRun(joinP(dir, "main.ts"), bunEnv);
+  // ENOTSOCK when the primary's fd 2 is a pipe/tty; some paths surface EINVAL.
+  // The load-bearing invariant is that the primary's stderr survives remove().
+  expect(stdout).toMatch(/worker error code: (ENOTSOCK|EINVAL|EBADF)/);
+  expect(stdout).toContain("stderr open: true");
+});