From dbc535c75b71782da3b30b90f05f5228bf2fe95f Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 13 Aug 2026 04:11:30 +0000 Subject: [PATCH] MessagePort: ref the port when its first 'message' listener is added Node's setupPortReferencing calls port.ref() when the first 'message' listener is added, so a port that was unref()'d before anyone listened is ref'd again once someone does. Bun only took the listener loop ref while m_isRefd was still set, so after unref() a later on('message') kept the port unref'd and the process exited with the listener installed. The first 'message' listener now goes through the same gated path as .ref() (setRefd()). The onmessage setter no longer takes a ref of its own: installing a callable handler is a listener add, so a handler installed next to other listeners leaves an unref() in force as in node; replacing the only handler still refs, since node's setter re-registers a replaced handler; anything non-callable still unrefs. The listener loop ref is also released when the port itself is destroyed. A listening port becomes collectable as soon as its peer closes, which can happen before the posted peer-close notification (a weak pointer) runs, and the ref then outlived the port and pinned the loop forever. The onmessage path was shielded from this by the self-reference the setter used to take; on()/addEventListener() ports were not. --- src/jsc/bindings/webcore/JSMessagePort.cpp | 10 +- src/jsc/bindings/webcore/MessagePort.cpp | 66 ++-- src/jsc/bindings/webcore/MessagePort.h | 18 +- .../worker_threads/worker_threads.test.ts | 282 ++++++++++++++++++ .../message-port-context-destroy-leak.test.ts | 14 +- 5 files changed, 355 insertions(+), 35 deletions(-) diff --git a/src/jsc/bindings/webcore/JSMessagePort.cpp b/src/jsc/bindings/webcore/JSMessagePort.cpp index 3b70232ef476..2311bf6c9bdd 100644 --- a/src/jsc/bindings/webcore/JSMessagePort.cpp +++ b/src/jsc/bindings/webcore/JSMessagePort.cpp @@ -203,10 +203,14 @@ static inline bool setJSMessagePort_onmessageSetter(JSGlobalObject& lexicalGloba vm.writeBarrier(&thisObject, value); ensureStillAliveHere(value); - // node: a callable handler starts the port and keeps the loop alive; assigning anything else - // clears the handler and lets the loop exit again. + // An installed handler is a 'message' listener: registering it started the port and, if it + // is the first listener, ref'd it (node's setupPortReferencing). The setter takes no ref of + // its own, so a handler added next to other listeners leaves an earlier unref() in force, + // as in node. Anything that cannot be called (a cleared handler, or an object + // setAttributeEventListener stores but JSEventListener never invokes) lets the loop exit + // again, as node's setter also only counts functions. if (value.isCallable()) - thisObject.wrapped().jsRef(&lexicalGlobalObject); + thisObject.wrapped().didSetMessageHandler(); else thisObject.wrapped().jsUnref(&lexicalGlobalObject); diff --git a/src/jsc/bindings/webcore/MessagePort.cpp b/src/jsc/bindings/webcore/MessagePort.cpp index 1e1681c17d85..c622a035ea63 100644 --- a/src/jsc/bindings/webcore/MessagePort.cpp +++ b/src/jsc/bindings/webcore/MessagePort.cpp @@ -60,14 +60,21 @@ MessagePort::MessagePort(ScriptExecutionContext& context, Ref&& { // The WeakPtrFactory must be initialized on the owning thread. EventTarget::initializeWeakPtrFactory(); - // Any port with a 'message' listener refs the event loop (matching node: a - // listening port keeps its thread alive until closed or unref'd); otherwise a - // buffered message could be lost if its listener is added late. + // A port's first 'message' listener refs it (see onDidChangeListenerImpl), matching node: + // a listening port keeps its thread alive until it is closed or unref'd, so a buffered + // message is not lost to the loop exiting. onDidChangeListener = &MessagePort::onDidChangeListenerImpl; } MessagePort::~MessagePort() { + // A listening port becomes collectable as soon as its peer closes, possibly before the + // posted peerClosed() (which only holds a weak pointer to us) runs and releases the + // listener loop-ref. Release it here too, or it outlives the port and the loop never + // idles. m_hasRef cannot be set here: it holds a reference to this object. + ASSERT(!m_hasRef); + m_isRefd = false; + updateListenerEventLoopRef(); if (!m_isDetached) m_pipe->close(m_side, MessagePortPipe::CloseKind::Collected); } @@ -224,9 +231,8 @@ void MessagePort::close() // it in hasPendingActivity()); marking our side Closed is sufficient. m_pipe->close(m_side, MessagePortPipe::CloseKind::Explicit); - // Release the self-reference taken by jsRef() (set when .onmessage is - // assigned or .ref() is called from JS). The JS .close() binding calls - // jsUnref() first; stop() and contextDestroyed() do not. + // Release the self-reference taken by jsRef() (.ref() from JS). The JS .close() + // binding calls jsUnref() first; stop() and contextDestroyed() do not. if (m_hasRef) { m_hasRef = false; if (auto* context = scriptExecutionContext()) @@ -285,8 +291,8 @@ void MessagePort::peerClosed() // Fire 'close' (guarded against a double dispatch) and release this side's loop refs // so the loop can idle, matching node. dispatchCloseEvent(); - // jsUnref() clears both the listener loop-ref (m_isRefd) and the onmessage/ref() - // keepalive (m_hasRef), so a listening transferred port stops pinning the loop. + // jsUnref() clears both the listener loop-ref (m_isRefd) and the .ref() keepalive + // (m_hasRef), so a listening transferred port stops pinning the loop. auto* globalObject = defaultGlobalObject(context->globalObject()); jsUnref(globalObject); } @@ -499,7 +505,11 @@ void MessagePort::onDidChangeListenerImpl(EventTarget& self, const AtomString& e auto& port = static_cast(self); switch (kind) { case Add: - port.m_messageEventCount++; + // Node's setupPortReferencing calls port.ref() for the first 'message' listener, so + // listening re-refs a port that was unref()'d before anyone listened. on(), + // addEventListener(), once() and an installed onmessage handler all register here. + if (++port.m_messageEventCount == 1) + port.setRefd(); break; case Remove: if (port.m_messageEventCount > 0) @@ -512,6 +522,12 @@ void MessagePort::onDidChangeListenerImpl(EventTarget& self, const AtomString& e port.updateListenerEventLoopRef(); } +void MessagePort::didSetMessageHandler() +{ + if (m_messageEventCount == 1) + setRefd(); +} + bool MessagePort::addEventListener(const AtomString& eventType, Ref&& listener, const AddEventListenerOptions& options) { if (eventType == eventNames().messageEvent) { @@ -550,24 +566,28 @@ WebCoreOpaqueRoot root(MessagePort* port) return WebCoreOpaqueRoot { port }; } -void MessagePort::jsRef(JSGlobalObject* lexicalGlobalObject) +bool MessagePort::setRefd() { // A closed or transferred-away port can never receive messages again, so - // taking a self-ref (and an event-loop ref) here would only leak: - // close()/disentangle() have already run and nothing will ever release a - // ref taken afterwards. Same once the peer has closed: peerClosed() already - // ran jsUnref(), and nothing releases a ref re-taken after it, so `.ref()` - // or a late `onmessage =` would pin the loop forever. Node no-ops both. - // Only an explicit peer close counts: node never closes a channel because a - // port was collected, so keying on Closed alone made this GC-dependent. + // a ref taken now would only leak: close()/disentangle() have already run + // and nothing will ever release a ref taken afterwards. Same once the peer + // has closed: peerClosed() already ran jsUnref(), and nothing releases a ref + // re-taken after it, so `.ref()` or a late listener would pin the loop + // forever. Node no-ops both. Only an explicit peer close counts: node never + // closes a channel because a port was collected, so keying on Closed alone + // made this GC-dependent. if (!isEntangled() || m_pipe->isOtherSideClosedByRequest(m_side)) - return; + return false; - // Re-acquire the message-listener loop-ref (if a listener is present) that .unref() released. - if (!m_isRefd) { - m_isRefd = true; - updateListenerEventLoopRef(); - } + m_isRefd = true; + updateListenerEventLoopRef(); + return true; +} + +void MessagePort::jsRef(JSGlobalObject* lexicalGlobalObject) +{ + if (!setRefd()) + return; if (!m_hasRef) { m_hasRef = true; diff --git a/src/jsc/bindings/webcore/MessagePort.h b/src/jsc/bindings/webcore/MessagePort.h index c08911798f25..348ac623052c 100644 --- a/src/jsc/bindings/webcore/MessagePort.h +++ b/src/jsc/bindings/webcore/MessagePort.h @@ -115,6 +115,11 @@ class MessagePort final : public ActiveDOMObject, public EventTarget, public Thr void jsRef(JSGlobalObject*); void jsUnref(JSGlobalObject*); + // The .onmessage setter installed a callable handler. Installing one is a listener add, + // which refs the port like any first 'message' listener; but node's setter also + // re-registers a replaced handler (removeListener, then newListener), so replacing the + // port's only 'message' listener refs it again even after an unref(). This covers that. + void didSetMessageHandler(); // Report the actual loop-ref state (matches Node's uv_has_ref), not the intent flag. bool jsHasRef() { return m_hasRef || m_listenerLoopRefActive; } @@ -159,17 +164,24 @@ class MessagePort final : public ActiveDOMObject, public EventTarget, public Thr // Read from the GC thread: a port whose only listener is 'close' must survive // until that event is delivered, or the peer's close is lost to a collection. std::atomic m_hasCloseEventListener { false }; + // The explicit .ref() hold: an event-loop ref plus a self-ref, so a ref()'d port with no + // listener pins the loop and outlives its wrapper (node never collects an open port). bool m_hasRef { false }; - // Whether .ref()/.unref() want this port to keep the loop alive (default refd); - // independent of m_hasRef (the .onmessage=/.ref() keepalive). - bool m_isRefd { true }; + // Node's ref flag as far as listeners are concerned: set by .ref() and by the first + // 'message' listener (node's setupPortReferencing calls port.ref() for it), cleared by + // .unref() and by close()/transfer/peer close. A fresh port is unref'd, as in node. It + // keeps the loop alive only while a 'message' listener is present. + bool m_isRefd { false }; // Whether the message-listener mechanism currently holds an event-loop ref // (held iff m_isRefd && m_messageEventCount > 0). bool m_listenerLoopRefActive { false }; uint32_t m_messageEventCount { 0 }; static void onDidChangeListenerImpl(EventTarget& self, const AtomString& eventType, OnDidChangeListenerKind kind); + // Sets m_isRefd for .ref() and for the first 'message' listener; declines (returning false) + // once the port can no longer receive, see the definition. + bool setRefd(); // Reconciles the listener event-loop ref with (m_isRefd && m_messageEventCount > 0). void updateListenerEventLoopRef(); }; diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index 023ea833ffa1..3a5f3f59addc 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -997,6 +997,288 @@ test("hasRef() survives collection of the unreferenced peer", () => { port1.close(); }); +// Node's setupPortReferencing (lib/internal/worker/io.js) calls port.ref() when the first +// 'message' listener is added, so listening re-refs a port that was unref()'d earlier; the +// onmessage setter is just one way of registering that listener and refs nothing on its own. +// Every expected value below is what node prints. +test("the first 'message' listener refs the port like ref() does", () => { + const channels: MessageChannel[] = []; + function port(): MessagePort { + const channel = new MessageChannel(); + channels.push(channel); + return channel.port1; + } + const f = () => {}; + const g = () => {}; + const seen: Record = {}; + let p: MessagePort; + + // the first listener refs an unref()'d port, however it is registered + p = port(); + p.unref(); + p.on("message", f); + seen["unref(); on()"] = p.hasRef(); + + p = port(); + p.unref(); + p.addEventListener("message", f); + seen["unref(); addEventListener()"] = p.hasRef(); + + p = port(); + p.unref(); + p.addEventListener("message", { handleEvent() {} }); + seen["unref(); addEventListener(handleEvent object)"] = p.hasRef(); + + p = port(); + p.unref(); + p.once("message", f); + seen["unref(); once()"] = p.hasRef(); + + p = port(); + p.unref(); + p.onmessage = f; + seen["unref(); onmessage ="] = p.hasRef(); + + p = port(); + p.unref(); + p.on("message", f); + p.off("message", f); + p.on("message", f); + seen["unref(); on(); off(); on()"] = p.hasRef(); + + // only the first listener refs: a handler installed next to another listener (or replacing + // one of several) leaves an unref() in force, while node's setter re-registers a replaced + // handler, so replacing the only listener refs again + p = port(); + p.on("message", g); + p.unref(); + p.onmessage = f; + seen["on(g); unref(); onmessage = f"] = p.hasRef(); + + p = port(); + p.on("message", g); + p.unref(); + p.on("message", f); + seen["on(g); unref(); on(f)"] = p.hasRef(); + + p = port(); + p.on("message", f); + p.onmessage = g; + p.unref(); + p.onmessage = () => {}; + seen["on(f); onmessage = g; unref(); onmessage = h"] = p.hasRef(); + + p = port(); + p.onmessage = f; + p.unref(); + p.onmessage = g; + seen["onmessage = f; unref(); onmessage = g"] = p.hasRef(); + + // ...while replacing it with something that cannot be called unrefs, as clearing it does: + // the object stays stored as the handler but is never invoked + p = port(); + p.onmessage = f; + p.onmessage = {} as any; + seen["onmessage = f; onmessage = {}"] = p.hasRef(); + + p = port(); + p.onmessage = f; + p.unref(); + p.onmessage = { handleEvent() {} } as any; + seen["onmessage = f; unref(); onmessage = { handleEvent }"] = p.hasRef(); + + // explicit ref()/unref() still win over a present listener + p = port(); + p.on("message", f); + p.unref(); + const afterUnref = p.hasRef(); + p.ref(); + seen["on(); unref(); ref()"] = [afterUnref, p.hasRef()]; + + // a listener added to a port that can no longer receive refs nothing + p = port(); + p.close(); + p.on("message", f); + seen["close(); on()"] = p.hasRef(); + + { + p = port(); + const carrier = port(); + carrier.postMessage(null, [p]); + p.on("message", f); + seen["transferred away; on()"] = p.hasRef(); + } + + // node's test-messageport-hasref up to its listener step (the file itself also needs the + // async_hooks MESSAGEPORT resource and close-event timing, so it is not vendored), plus + // the off() that undoes it + { + p = port(); + const sequence = [p.hasRef()]; + p.unref(); + sequence.push(p.hasRef()); + p.ref(); + sequence.push(p.hasRef()); + p.unref(); + sequence.push(p.hasRef()); + p.on("message", f); + sequence.push(p.hasRef()); + p.off("message", f); + sequence.push(p.hasRef()); + seen["fresh; unref(); ref(); unref(); on(); off()"] = sequence; + } + + for (const { port1, port2 } of channels) { + port1.close(); + port2.close(); + } + + expect(seen).toEqual({ + "unref(); on()": true, + "unref(); addEventListener()": true, + "unref(); addEventListener(handleEvent object)": true, + "unref(); once()": true, + "unref(); onmessage =": true, + "unref(); on(); off(); on()": true, + "on(g); unref(); onmessage = f": false, + "on(g); unref(); on(f)": false, + "on(f); onmessage = g; unref(); onmessage = h": false, + "onmessage = f; unref(); onmessage = g": true, + "onmessage = f; onmessage = {}": false, + "onmessage = f; unref(); onmessage = { handleEvent }": false, + "on(); unref(); ref()": [false, true], + "close(); on()": false, + "transferred away; on()": false, + "fresh; unref(); ref(); unref(); on(); off()": [false, false, true, false, true, false], + }); +}); + +// Once the peer's close has been delivered the port is as good as closed (node closes it +// outright), so a first listener added afterwards must not take a loop ref either. +test("a first 'message' listener added after the peer closed does not ref the port", async () => { + const { port1, port2 } = new MessageChannel(); + const closed = new Promise(resolve => port1.once("close", resolve)); + port2.close(); + await closed; + port1.on("message", () => {}); + expect(port1.hasRef()).toBe(false); + port1.close(); +}); + +// The process-level symptom of the flag above: whether the port keeps the loop alive. Each +// script's only other work is an unref()'d timer, which can only fire if the port is +// keeping the process up; a script that must exit on its own uses a long unref()'d timer +// as a bounded way to report that it did not. +describe.concurrent("a port's first 'message' listener and process lifetime", () => { + async function run(script: string) { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", `const { MessageChannel } = require("worker_threads");\n${script}`], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + return { stdout, stderr, exitCode }; + } + + test("unref() then on('message') keeps the process alive until the port closes", async () => { + const result = await run(` + const { port1, port2 } = new MessageChannel(); + port1.unref(); + port1.on("message", m => { console.log("got " + m); port1.close(); port2.close(); }); + setTimeout(() => port2.postMessage("hi"), 50).unref(); + `); + expect(result).toEqual({ stdout: "got hi\n", stderr: "", exitCode: 0 }); + }); + + test("unref() then addEventListener('message') keeps the process alive until the port closes", async () => { + const result = await run(` + const { port1, port2 } = new MessageChannel(); + port1.unref(); + port1.addEventListener("message", e => { console.log("got " + e.data); port1.close(); port2.close(); }); + setTimeout(() => port2.postMessage("hi"), 50).unref(); + `); + expect(result).toEqual({ stdout: "got hi\n", stderr: "", exitCode: 0 }); + }); + + test("onmessage = next to an existing listener does not re-ref an unref()'d port", async () => { + const result = await run(` + const { port1, port2 } = new MessageChannel(); + globalThis.keep = port2; + port1.on("message", () => {}); + port1.unref(); + port1.onmessage = () => {}; + setTimeout(() => { console.log("still pinned"); process.exit(1); }, 5_000).unref(); + console.log("done"); + `); + expect(result).toEqual({ stdout: "done\n", stderr: "", exitCode: 0 }); + }); + + // The listener's loop ref has to die with the port. A listening port nobody references is + // collectable as soon as its peer closes, and a GC can get to it before the peer-close + // notification (which only holds a weak pointer to the port) runs and releases the ref; + // the port must release it itself when it is destroyed, or the process never exits. + const listen = { + "on()": `port.on("message", () => {})`, + "addEventListener()": `port.addEventListener("message", () => {})`, + "onmessage =": `port.onmessage = () => {}`, + }; + test.each(Object.entries(listen))( + "a port listening via %s that is collected after its peer closes releases its ref", + async (_, register) => { + const result = await run(` + (function () { + for (let i = 0; i < 20; i++) { + const { port1, port2: port } = new MessageChannel(); + ${register}; + port1.close(); + } + })(); + Bun.gc(true); + Bun.gc(true); + setTimeout(() => { console.log("still pinned"); process.exit(1); }, 5_000).unref(); + console.log("done"); + `); + expect(result).toEqual({ stdout: "done\n", stderr: "", exitCode: 0 }); + }, + ); + + test("a listening port collected along with its peer releases its ref", async () => { + const result = await run(` + (function () { + for (let i = 0; i < 20; i++) new MessageChannel().port2.on("message", () => {}); + })(); + // The first collection takes the peers (nothing listens on them); that closes the + // listening sides' channels, and the second collection takes the listening sides. + Bun.gc(true); + Bun.gc(true); + setTimeout(() => { console.log("still pinned"); process.exit(1); }, 5_000).unref(); + console.log("done"); + `); + expect(result).toEqual({ stdout: "done\n", stderr: "", exitCode: 0 }); + }); + + // parentPort is the same MessagePort: unref()'d first and listened to afterwards, it keeps + // the worker alive (node), instead of the worker exiting before any message can reach it. + test("parentPort.unref() followed by on('message') keeps the worker alive", async () => { + const w = new Worker( + `const { parentPort } = require("worker_threads"); + process.on("beforeExit", () => parentPort.postMessage("loop drained")); + parentPort.unref(); + parentPort.on("message", m => { parentPort.postMessage("got " + m); parentPort.close(); }); + setTimeout(() => parentPort.postMessage("still alive"), 50).unref();`, + { eval: true }, + ); + const seen: string[] = []; + const exited = new Promise(resolve => w.on("exit", resolve)); + w.on("message", m => { + seen.push(m); + if (m === "still alive") w.postMessage("hi"); + }); + expect({ exitCode: await exited, seen }).toEqual({ exitCode: 0, seen: ["still alive", "got hi"] }); + }); +}); + // markAsUncloneable blocks *cloning*, not transfer: a marked port in the transfer // list is moved, so node lets it through and it still works on the far side. test("markAsUncloneable blocks cloning a port but not transferring it", async () => { diff --git a/test/js/web/workers/message-port-context-destroy-leak.test.ts b/test/js/web/workers/message-port-context-destroy-leak.test.ts index 270d44248a85..d21203501b8b 100644 --- a/test/js/web/workers/message-port-context-destroy-leak.test.ts +++ b/test/js/web/workers/message-port-context-destroy-leak.test.ts @@ -2,11 +2,11 @@ import { expect, test } from "bun:test"; import { bunEnv, bunExe, isASAN, isWindows } from "harness"; // MessagePort::jsRef() takes a self-ref() on the C++ MessagePort (plus an -// event-loop ref) when .onmessage is assigned or .ref() is called. The only -// path that released it was an explicit .close()/.unref() from JS. When a -// Worker (or any owning context) is torn down without that, contextDestroyed() -// → close() ran but never dropped the self-ref, so every such MessagePort -// leaked for the lifetime of the process. +// event-loop ref) when .ref() is called. The only path that released it was +// an explicit .close()/.unref() from JS. When a Worker (or any owning context) +// is torn down without that, contextDestroyed() → close() ran but never +// dropped the self-ref, so every such MessagePort leaked for the lifetime of +// the process. // // Skipped on Windows: RSS there does not drop after worker threads exit (the // per-thread mimalloc arenas stay committed), so the allocator residue from @@ -25,7 +25,9 @@ test.skipIf(isWindows)( const keep = []; for (let i = 0; i < 8000; i++) { const { port1, port2 } = new MessageChannel(); - // Assigning onmessage calls MessagePort::jsRef() → self-ref(). + // ref() calls MessagePort::jsRef() → self-ref(). (A listener alone takes + // only an event-loop ref, so it is not what this test measures.) + port1.ref(); port1.onmessage = () => {}; keep.push(port1, port2); }