diff --git a/src/js/internal/cluster/child.ts b/src/js/internal/cluster/child.ts index 28aa980e08cb..0ccb2c74c93e 100644 --- a/src/js/internal/cluster/child.ts +++ b/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; diff --git a/src/js/internal/cluster/primary.ts b/src/js/internal/cluster/primary.ts index 0cf353fee8a1..e84add590ac6 100644 --- a/src/js/internal/cluster/primary.ts +++ b/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; diff --git a/src/js/internal/shared.ts b/src/js/internal/shared.ts index ad88418760dd..2efec5719ee6 100644 --- a/src/js/internal/shared.ts +++ b/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"), diff --git a/src/js/node/net.ts b/src/js/node/net.ts index 8d2305b5ef50..a257dc98b20b 100644 --- a/src/js/node/net.ts +++ b/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, diff --git a/test/js/node/cluster.test.ts b/test/js/node/cluster.test.ts index bd8506b3b071..6dfe5c91e0a7 100644 --- a/test/js/node/cluster.test.ts +++ b/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); +});