node:cluster: let workers shut down gracefully on worker.disconnect()

src/js/internal/cluster/child.ts

@@ -155,6 +155,12 @@ cluster._getServer = function (obj, options, cb) {
   });
 };
 
+// For servers that listen without _getServer (node:http), so _disconnect still closes them.
+cluster._trackServer = function (server) {
+  handles.set(server, server);
+  server.once("close", () => handles.delete(server));
+};
+
 function removeIndexesKey(indexesKey, index) {
   const indexSet = indexes.get(indexesKey);
   if (!indexSet) {
@@ -310,6 +316,7 @@ Worker.prototype.disconnect = function () {
 };
 
 Worker.prototype._disconnect = function (this: typeof Worker, primaryInitiated?) {
+  if (this.exitedAfterDisconnect) return;
   this.exitedAfterDisconnect = true;
   let waitingCount = 1;
 

src/js/internal/cluster/primary.ts

@@ -342,7 +342,6 @@ function send(worker, message, handle?, cb?) {
 Worker.prototype.disconnect = function () {
   this.exitedAfterDisconnect = true;
   send(this, { act: "disconnect" });
-  this.process.disconnect();
   removeHandlesForWorker(this, false);
   removeWorker(this);
   return this;

src/js/node/_http_server.ts

@@ -601,6 +601,9 @@ Server.prototype.listen = function () {
 
     if (cluster === undefined) cluster = require("node:cluster");
 
+    // worker.disconnect() only closes servers the cluster child knows about.
+    cluster._trackServer(server);
+
     server.once("listening", () => {
       // No channel (NODE_UNIQUE_ID inherited by a plain child, or already disconnected): nothing to notify.
       if (!process.connected) return;

test/js/node/cluster.test.ts

@@ -1,4 +1,4 @@
-import { expect, test } from "bun:test";
+import { expect, setDefaultTimeout, test } from "bun:test";
 import {
   bunEnv,
   bunExe,
@@ -13,6 +13,8 @@ import {
 } from "harness";
 import net from "node:net";
 
+setDefaultTimeout(40_000);
+
 test.concurrent("cloneable and transferable equals", async () => {
   const dir = tempDirWithFiles("bun-test", {
     "index.ts": `
@@ -1276,3 +1278,137 @@ if (cluster.isPrimary) {
   });
   expect(exitCode).toBe(0);
 }, 30_000);
+
+// Per https://nodejs.org/api/cluster.html#workerdisconnect, disconnect() closes the
+// worker's servers, waits for their 'close' events, then disconnects the channel.
+// node:http binds its own socket instead of using _getServer, so it needs tracking.
+test.concurrent(
+  "primary-initiated worker.disconnect() closes the worker's http server and the worker exits",
+  async () => {
+    using dir = tempDir("cluster-disconnect-http", {
+      "main.js": `
+        const cluster = require("node:cluster");
+        const http = require("node:http");
+        const net = require("node:net");
+
+        if (cluster.isPrimary) {
+          const worker = cluster.fork();
+          const disconnected = new Promise(resolve => worker.once("disconnect", resolve));
+          const exited = new Promise(resolve => worker.once("exit", (code, signal) => resolve({ code, signal })));
+
+          worker.on("message", async msg => {
+            if (msg.cmd !== "listening") return;
+            const port = msg.port;
+            const before = await (await fetch(\`http://127.0.0.1:\${port}/\`)).text();
+
+            worker.disconnect();
+
+            // The worker must close its server, disconnect the channel and
+            // exit on its own. If it never does (the bug), kill it so it
+            // cannot outlive the test, and report the failure.
+            let timer;
+            const timedOut = new Promise(resolve => {
+              timer = setTimeout(resolve, 20_000, "timeout");
+            });
+            const exit = await Promise.race([Promise.all([exited, disconnected]).then(([e]) => e), timedOut]);
+            clearTimeout(timer);
+            if (exit === "timeout") {
+              worker.process.kill("SIGKILL");
+              console.log(JSON.stringify({ fail: "worker did not exit after disconnect()" }));
+              process.exit(1);
+            }
+
+            // The worker is gone, so nothing may be listening on its port.
+            const after = await new Promise(resolve => {
+              const socket = net.connect(port, "127.0.0.1");
+              socket.once("connect", () => {
+                socket.destroy();
+                resolve("still listening");
+              });
+              socket.once("error", () => resolve("refused"));
+            });
+
+            console.log(JSON.stringify({ before, exit, exitedAfterDisconnect: worker.exitedAfterDisconnect, after }));
+            process.exit(0);
+          });
+        } else {
+          const server = http.createServer((req, res) => res.end("served-by-worker"));
+          server.listen(0, "127.0.0.1", () => {
+            process.send({ cmd: "listening", port: server.address().port });
+          });
+        }
+      `,
+    });
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "main.js"],
+      env: bunEnv,
+      cwd: String(dir),
+      // Inherited so that on regression the worker's output reaches the
+      // runner log instead of filling an unread pipe.
+      stderr: "inherit",
+    });
+    const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
+    expect(stdout.trim()).toBe(
+      '{"before":"served-by-worker","exit":{"code":0,"signal":null},"exitedAfterDisconnect":true,"after":"refused"}',
+    );
+    expect(exitCode).toBe(0);
+  },
+);
+
+// A shutdown path and an error path both asking a worker to go away is the normal
+// race in a cluster manager, so Node makes the second disconnect() a no-op.
+test.concurrent("calling worker.disconnect() twice does not throw", async () => {
+  using dir = tempDir("cluster-double-disconnect", {
+    "main.js": `
+      const cluster = require("node:cluster");
+      if (cluster.isPrimary) {
+        const worker = cluster.fork();
+        worker.on("online", () => {
+          worker.disconnect();
+          worker.disconnect();
+        });
+        worker.on("exit", (code, signal) => console.log("worker-exit", code, signal));
+        cluster.on("disconnect", () => console.log("cluster-disconnect"));
+      }
+    `,
+  });
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "main.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(stderr).not.toContain("ERR_IPC_DISCONNECTED");
+  // The primary and the worker share stdout, so their interleaving is not fixed.
+  const lines = stdout.split("\n").filter(Boolean).sort();
+  expect(lines).toEqual(["cluster-disconnect", "worker-exit 0 null"]);
+  expect(exitCode).toBe(0);
+});
+
+// The primary keeps its end of the channel open, so a "disconnect" sent before the
+// worker booted does not swallow the "online" message coming the other way.
+test.concurrent("disconnecting a worker before it comes online still emits 'online'", async () => {
+  using dir = tempDir("cluster-disconnect-before-online", {
+    "main.js": `
+      const cluster = require("node:cluster");
+      if (cluster.isPrimary) {
+        const worker = cluster.fork();
+        worker.disconnect();
+        worker.on("online", () => console.log("online"));
+        worker.on("exit", code => console.log("worker-exit", code));
+      }
+    `,
+  });
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "main.js"],
+    env: bunEnv,
+    cwd: String(dir),
+    stdout: "pipe",
+    stderr: "inherit",
+  });
+  const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
+  expect(stdout.split("\n").filter(Boolean).sort()).toEqual(["online", "worker-exit 0"]);
+  expect(exitCode).toBe(0);
+});