node:cluster: worker.disconnect() drops queued send() messages and the worker never sees 'disconnect'

src/js/internal/cluster/child.ts

@@ -1,6 +1,7 @@
 const EventEmitter = require("node:events");
 const Worker = require("internal/cluster/Worker");
 const path = require("node:path");
+const { owner_symbol } = require("internal/shared");
 
 const sendHelper = $newRustFunction("node_cluster_binding.rs", "sendHelperChild", 3);
 const onInternalMessage = $newRustFunction("node_cluster_binding.rs", "onInternalMessageChild", 2);
@@ -15,7 +16,6 @@ const indexes = new Map();
 const noop = FunctionPrototype;
 const TIMEOUT_MAX = 2 ** 31 - 1;
 const kNoFailure = 0;
-const owner_symbol = Symbol("owner_symbol");
 
 export default cluster;
 

src/js/internal/cluster/primary.ts

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

src/js/internal/shared.ts

@@ -290,6 +290,9 @@ export default {
   NodeEntryObserver,
 
   kHandle: Symbol("kHandle"),
+  // node:net and internal/cluster/child must agree on this symbol so the
+  // cluster disconnect protocol can find the net.Server owning a faux handle.
+  owner_symbol: Symbol("owner_symbol"),
   kAutoDestroyed: Symbol("kAutoDestroyed"),
   kResistStopPropagation: Symbol("kResistStopPropagation"),
   kWeakHandler: Symbol("kWeak"),

src/js/node/net.ts

@@ -34,6 +34,7 @@ const {
   hasObserver,
   startPerf,
   stopPerf,
+  owner_symbol,
 } = require("internal/shared");
 import type { Socket, SocketHandler, SocketListener } from "bun";
 import type { Server as NetServer, Socket as NetSocket, ServerOpts } from "node:net";
@@ -115,7 +116,6 @@ const getBufferedAmount = $newRustFunction("runtime/socket/socket.rs", "jsGetBuf
 
 const bunTlsSymbol = Symbol.for("::buntls::");
 const bunSocketServerOptions = Symbol.for("::bunnetserveroptions::");
-const owner_symbol = Symbol("owner_symbol");
 
 const kServerSocket = Symbol("kServerSocket");
 const kBytesWritten = Symbol("kBytesWritten");
@@ -3643,6 +3643,13 @@ function listenInCluster(
     if (err) {
       throw new ExceptionWithHostPort(err, "bind", address, port);
     }
+    // Bun keeps the real Bun.listen() handle as server._handle rather than the
+    // cluster faux handle, so link them here: Worker#_disconnect finds the
+    // server via owner_symbol, and closing the server closes the faux handle.
+    if (handle) {
+      handle[owner_symbol] = server;
+      server.once("close", () => handle.close());
+    }
     server[kRealListen](
       path,
       port,

test/js/node/cluster.test.ts

@@ -1,5 +1,6 @@
 import { expect, test } from "bun:test";
-import { bunEnv, bunExe, bunRun, joinP, tempDirWithFiles } from "harness";
+import { bunEnv, bunExe, bunRun, joinP, tempDir, tempDirWithFiles } from "harness";
+import path from "node:path";
 
 test("cloneable and transferable equals", () => {
   const dir = tempDirWithFiles("bun-test", {
@@ -185,3 +186,66 @@ test("disconnect() on a cluster.Worker built around a plain object does not abor
   const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
   expect({ stdout: stdout.trim(), exitCode }).toEqual({ stdout: "returned self: true", exitCode: 0 });
 });
+
+// worker.disconnect() must deliver every message already accepted by
+// worker.send() and let the worker observe 'disconnect'. In Node the primary
+// only enqueues {act:'disconnect'} (ordered after the user messages on the
+// same channel); the worker closes the channel from its side once received.
+test("worker.disconnect() delivers queued send() messages and the worker sees 'disconnect'", async () => {
+  using dir = tempDir("cluster-disconnect-drain", {
+    "main.js": `
+      const cluster = require("node:cluster");
+      const fs = require("node:fs");
+      const N = 50;
+      if (cluster.isPrimary) {
+        const OUT = process.argv[2];
+        const worker = cluster.fork({ OUT });
+        worker.once("message", msg => {
+          if (msg !== "ready") return;
+          // 64 KiB per message so the IPC write queue backs up past the
+          // kernel socket buffer before disconnect() is called.
+          const payload = Buffer.alloc(64 * 1024, "z").toString();
+          let acked = 0;
+          for (let i = 0; i < N; i++) worker.send({ q: "p", i, payload }, err => { if (!err) acked++; });
+          worker.disconnect();
+          let sawDisconnect = false;
+          worker.once("disconnect", () => { sawDisconnect = true; });
+          worker.once("exit", (code, signal) => {
+            const got = JSON.parse(fs.readFileSync(OUT, "utf8"));
+            console.log(JSON.stringify({
+              acked, received: got.n, workerSawDisconnect: got.saw,
+              primarySawDisconnect: sawDisconnect, exitedAfterDisconnect: worker.exitedAfterDisconnect,
+              code, signal,
+            }));
+          });
+        });
+      } else {
+        let n = 0, saw = false;
+        const dump = () => { try { fs.writeFileSync(process.env.OUT, JSON.stringify({ n, saw })); } catch {} };
+        process.on("message", m => { if (m && m.q === "p") n++; });
+        process.on("disconnect", () => { saw = true; dump(); process.exit(0); });
+        process.on("exit", dump);
+        process.send("ready");
+      }
+    `,
+  });
+  const out = path.join(String(dir), "out.json");
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "main.js", out],
+    env: bunEnv,
+    cwd: String(dir),
+    stdout: "pipe",
+    stderr: "inherit",
+  });
+  const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
+  expect(JSON.parse(stdout.trim())).toEqual({
+    acked: 50,
+    received: 50,
+    workerSawDisconnect: true,
+    primarySawDisconnect: true,
+    exitedAfterDisconnect: true,
+    code: 0,
+    signal: null,
+  });
+  expect(exitCode).toBe(0);
+});