diff --git a/src/js/internal/cluster/primary.ts b/src/js/internal/cluster/primary.ts index d4840fccbcbf..963a13d2d3df 100644 --- a/src/js/internal/cluster/primary.ts +++ b/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,9 +262,6 @@ 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"; @@ -270,15 +269,10 @@ function queryServer(worker, message) { 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); diff --git a/src/runtime/ipc.rs b/src/runtime/ipc.rs index 0d97b065e3d1..6aec97c8426e 100644 --- a/src/runtime/ipc.rs +++ b/src/runtime/ipc.rs @@ -922,6 +922,9 @@ pub struct SendQueue { root: Cell>>, pub(crate) queue: JsCell>, pub(crate) waiting_for_ack: JsCell>, + /// 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>, pub(crate) retry_count: Cell, pub(crate) keep_alive: JsCell, @@ -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); diff --git a/test/js/node/cluster.test.ts b/test/js/node/cluster.test.ts index 64d101f15e8f..f6262e46c43b 100644 --- a/test/js/node/cluster.test.ts +++ b/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 () => {