diff --git a/src/jsc/bindings/webcore/MessagePort.cpp b/src/jsc/bindings/webcore/MessagePort.cpp index f152366907f2..e4df194186c1 100644 --- a/src/jsc/bindings/webcore/MessagePort.cpp +++ b/src/jsc/bindings/webcore/MessagePort.cpp @@ -28,6 +28,7 @@ #include "MessagePort.h" #include "BunClientData.h" +#include "Event.h" #include "EventNames.h" #include "MessageEvent.h" #include "MessagePortPipe.h" @@ -108,11 +109,27 @@ void MessagePort::close() return; m_isDetached = true; - // m_pipe is held for the port's whole lifetime (the GC thread reads - // it in hasPendingActivity()); marking our side Closed is sufficient. - m_pipe->close(m_side); - - removeAllEventListeners(); + // m_pipe is held for the port's whole lifetime (the GC thread reads it in + // hasPendingActivity()); marking our side Closed is sufficient. Record + // whether this is an explicit script close() so the peer only fires its + // 'close' for that, not for a GC/teardown/drop close (which would make + // non-script death observable from JS). + bool byScript = canRunScript(); + m_pipe->close(m_side, byScript); + + // Wake the entangled peer so it can fire its own 'close' event after + // draining any queued messages — only for a real close() from script. + // contextDestroyed() during teardown and ~MessagePort (which calls the pipe + // directly, bypassing this method) leave it un-woken, so the peer is + // neither woken nor pinned waiting for a close that will never come. + if (byScript) + m_pipe->wakePeerForClose(m_side); + + // Fire our own 'close' event asynchronously (Node + HTML semantics), then + // tear down listeners. If JS can't run (context teardown) tear down + // synchronously as before. scheduleCloseEvent() takes its own strong ref, + // so it is safe to run before releasing m_hasRef. + bool scheduledClose = scheduleCloseEvent(); // Release the self-reference taken by jsRef() (set when .onmessage is // assigned or .ref() is called from JS). The JS .close() binding calls @@ -127,6 +144,110 @@ void MessagePort::close() context->unrefEventLoop(); deref(); } + + if (!scheduledClose) { + // No 'close' task could be posted (context teardown / postTaskTo + // failed), so no 'close' event will ever fire for this port. Mark the + // close consumed and tear down now, so a close listener added after + // close() cannot pin the already-closed wrapper via hasPendingActivity(). + m_closeEventDispatched = true; + removeAllEventListeners(); + } +} + +void MessagePort::startForClose() +{ + // Attach to the pipe so the peer can wake this port to dispatch 'close' + // (the wake comes from the peer's close() via wakePeerForClose(), or from + // this attach() itself if the peer has already script-closed). Unlike + // start(), this does not set m_started, so a later 'message' listener still + // runs start() and re-attaches to flush any buffered messages. attach() is + // idempotent, so calling it again from start() is harmless. + if (!isEntangled()) + return; + auto* context = scriptExecutionContext(); + if (!context) + return; + m_pipe->attach(m_side, context->identifier(), ThreadSafeWeakPtr { *this }); +} + +bool MessagePort::canRunScript() const +{ + auto* context = scriptExecutionContext(); + if (!context || !context->globalObject()) + return false; + auto* globalObject = defaultGlobalObject(context->globalObject()); + return Zig::GlobalObject::scriptExecutionStatus(globalObject, globalObject) == ScriptExecutionStatus::Running; +} + +void MessagePort::dispatchCloseEvent() +{ + if (m_closeEventDispatched) + return; + m_closeEventDispatched = true; + + if (!canRunScript()) + return; + + // Bypass MessagePort::dispatchEvent()'s detached guard: by the time the + // close task runs the port is already detached, but the 'close' event must + // still reach its listener. + EventTarget::dispatchEvent(Event::create(eventNames().closeEvent, Event::CanBubble::No, Event::IsCancelable::No)); +} + +void MessagePort::dispatchCloseEventSelf() +{ + dispatchCloseEvent(); + removeAllEventListeners(); + m_hasMessageEventListener = false; + m_hasCloseEventListener = false; +} + +bool MessagePort::scheduleCloseEvent() +{ + if (m_closeEventDispatched) + return false; + + // Post unconditionally when JS can run, even without a close listener yet: + // Node and the HTML spec queue the close task on close(), so a listener + // added synchronously afterwards (port.close(); port.on('close', cb)) still + // fires. dispatchCloseEventSelf() is a no-op dispatch when no listener + // exists and then tears the port down, so the wrapper still gets collected. + if (!canRunScript()) + return false; + + auto* context = scriptExecutionContext(); + return ScriptExecutionContext::postTaskTo(context->identifier(), [protectedThis = Ref { *this }](ScriptExecutionContext&) { + protectedThis->dispatchCloseEventSelf(); + }); +} + +void MessagePort::dispatchCloseEventFromPeer() +{ + if (m_isDetached || m_closeEventDispatched || !m_hasCloseEventListener) + return; + + // Runs JS (the close handler), which may drop the last external ref. The + // caller (the drain) holds a RefPtr and we take our own Ref here, so the + // C++ object survives; the JS wrapper is rooted for the handler by the + // event's target on the JS stack (same GC tolerance as the message path). + Ref protectedThis { *this }; + // Stop message delivery and make a re-entrant close() a no-op. + m_isDetached = true; + + dispatchCloseEvent(); + + m_pipe->close(m_side); + removeAllEventListeners(); + m_hasMessageEventListener = false; + m_hasCloseEventListener = false; + + if (m_hasRef) { + m_hasRef = false; + if (auto* context = scriptExecutionContext()) + context->unrefEventLoop(); + deref(); + } } TransferredMessagePort MessagePort::disentangle() @@ -244,12 +365,35 @@ bool MessagePort::hasPendingActivity() const // atomic loads. The plain bool reads can observe stale values but // cannot crash — at worst the wrapper is collected one cycle early // or late, which is the same tolerance as before this refactor. - if (!scriptExecutionContext() || m_isDetached) + if (!scriptExecutionContext()) + return false; + + uint64_t s = m_pipe->state(m_side); + + // Keep the wrapper (and its 'close' listener) alive until a pending close + // event is dispatched. A close is pending, and will actually fire, when a + // close listener is registered, it has not been dispatched yet, and either: + // - this side has closed: the closing port's own 'close' task (posted by + // close() while JS can run) will fire it, or close() already set + // m_closeEventDispatched when it couldn't post (teardown); or + // - the peer has closed AND a drain is scheduled on this side: that drain + // is the only thing that calls dispatchCloseEventFromPeer(), so the pin + // is tied to its existence. A peer closed via ~MessagePort / + // ~TransferredMessagePort / teardown never wakes us (no drain), so such + // a port is not pinned and stays collectable. + // The !m_closeEventDispatched guard lets an already-closed port be + // collected once its close has fired (or was marked consumed), so a close + // listener added afterwards cannot pin the wrapper forever. + if (m_hasCloseEventListener && !m_closeEventDispatched + && ((s & MessagePortPipe::Closed) + || ((s & MessagePortPipe::DrainScheduled) && !m_pipe->isOtherSideOpen(m_side)))) + return true; + + if (m_isDetached) return false; if (!m_hasMessageEventListener) return false; - uint64_t s = m_pipe->state(m_side); // Keep alive if there are messages already queued for us, or the peer // is still open and could send more. return MessagePortPipe::queuedCount(s) > 0 || m_pipe->isOtherSideOpen(m_side); @@ -312,6 +456,9 @@ bool MessagePort::addEventListener(const AtomString& eventType, Ref port; - size_t limit; + size_t limit = 0; + // Set when the inbox empties while the port is still attached: the port may + // then owe a 'close' event (dispatched below, outside the lock). + bool emptied = false; { Locker locker { s.lock }; // This task was posted to `expectedCtx` (and is running there). If @@ -112,9 +115,18 @@ void MessagePortPipe::drainAndDispatch(uint8_t side, ScriptExecutionContextIdent uint64_t st = s.state.load(std::memory_order_relaxed); if (!port || s.inbox.isEmpty()) { s.state.store(st & ~DrainScheduled, std::memory_order_release); - return; + emptied = port && (st & Attached) != 0; + } else { + limit = std::max(s.inbox.size(), 1000); } - limit = std::max(s.inbox.size(), 1000); + } + + if (!limit) { + // Nothing queued. If our peer has closed, the attached port still owes + // a 'close' event. + if (emptied && readyToDispatchClose(side)) + port->dispatchCloseEventFromPeer(); + return; } auto* context = port->scriptExecutionContext(); @@ -141,6 +153,20 @@ void MessagePortPipe::drainAndDispatch(uint8_t side, ScriptExecutionContextIdent break; uint64_t st = s.state.load(std::memory_order_relaxed); if (!(st & Attached) || s.inbox.isEmpty()) { + s.state.store(st & ~DrainScheduled, std::memory_order_release); + emptied = (st & Attached) != 0; + break; + } + // A port attached only for a 'close' listener (start() was never + // called) must not have its queued messages dispatched to no one; + // an unstarted port buffers per the HTML spec. Leave them queued — + // adding a 'message' listener (or start()) runs start(), which + // re-attaches and reschedules this drain. Don't set `emptied`: the + // inbox is non-empty, so a pending 'close' waits behind the + // undelivered messages (matching Node). A *started* port with no + // listener still dispatches (to no one) below, as the spec requires, + // which keeps its inbox draining and avoids stranding/pinning it. + if (!port->started()) { s.state.store(st & ~DrainScheduled, std::memory_order_release); break; } @@ -165,6 +191,30 @@ void MessagePortPipe::drainAndDispatch(uint8_t side, ScriptExecutionContextIdent if (rescheduleCtx) scheduleDrain(side, rescheduleCtx); + else if (emptied && readyToDispatchClose(side)) + // Inbox fully drained and the peer has closed: deliver 'close' after + // all queued messages, matching Node's ordering. + port->dispatchCloseEventFromPeer(); +} + +bool MessagePortPipe::readyToDispatchClose(uint8_t side) +{ + ASSERT(side < 2); + // The peer must have closed via an explicit script close() (a GC/teardown/ + // drop close must not surface as a 'close' event), and this side must still + // be drained. Re-check the inbox/drain state under the lock to close a + // cross-thread TOCTOU: the drain loop cleared DrainScheduled and released + // the lock before we get here, and a concurrent send() on the peer thread + // can re-enqueue a message and re-arm a drain in that window. If it did, + // dispatching 'close' now would drop that message (dispatchCloseEventFromPeer + // -> close() clears the inbox); instead leave it to the freshly scheduled + // drain, which will deliver the message and then dispatch close itself. + if (!isOtherSideClosedByScript(side)) + return false; + auto& s = m_sides[side]; + Locker locker { s.lock }; + uint64_t st = s.state.load(std::memory_order_relaxed); + return (st & Attached) && !(st & DrainScheduled) && queuedCount(st) == 0; } std::optional MessagePortPipe::takeOne(uint8_t side) @@ -188,8 +238,14 @@ void MessagePortPipe::attach(uint8_t side, ScriptExecutionContextIdentifier ctxI s.ctxId = ctxId; s.port = WTF::move(port); uint64_t st = s.state.load(std::memory_order_relaxed); - uint64_t ns = (st | Attached) & ~Closed; - if (queuedCount(st) > 0 && !(st & DrainScheduled)) { + uint64_t ns = (st | Attached) & ~(Closed | ClosedByScript); + // Schedule a drain if there is work to dispatch: queued messages, or a + // peer that has closed via an explicit script close() (a 'close' event + // is owed). The latter covers a 'close' listener added after the peer's + // close(), which would otherwise have missed the peer's wake-up. A peer + // closed by GC/teardown/drop is deliberately NOT woken here — firing + // 'close' for it would make that non-script death observable from JS. + if (!(st & DrainScheduled) && (queuedCount(st) > 0 || isOtherSideClosedByScript(side))) { ns |= DrainScheduled; wakeCtx = ctxId; } @@ -199,6 +255,34 @@ void MessagePortPipe::attach(uint8_t side, ScriptExecutionContextIdentifier ctxI scheduleDrain(side, wakeCtx); } +void MessagePortPipe::wakePeerForClose(uint8_t side) +{ + ASSERT(side < 2); + // Called by MessagePort::close() (a real close() from script) after this + // side has been marked Closed: schedule a drain on the entangled peer so + // it can dispatch its 'close' event after its queued messages drain. + // Deliberately reached only from an explicit close() (gated on + // canRunScript()), not from ~MessagePort / context teardown: firing the + // peer's 'close' on teardown/GC is intentionally not supported, and + // hasPendingActivity() matches by not pinning a port whose peer died that + // way (no drain is scheduled for it). + // + // If a drain is already in flight on the peer it will observe this side's + // Closed bit and dispatch close itself, so only schedule when none is. + auto& peer = m_sides[1 - side]; + ScriptExecutionContextIdentifier peerCtx = 0; + { + Locker locker { peer.lock }; + uint64_t ps = peer.state.load(std::memory_order_relaxed); + if ((ps & Attached) && !(ps & DrainScheduled)) { + peer.state.store(ps | DrainScheduled, std::memory_order_release); + peerCtx = peer.ctxId; + } + } + if (peerCtx) + scheduleDrain(1 - side, peerCtx); +} + void MessagePortPipe::detach(uint8_t side) { ASSERT(side < 2); @@ -214,7 +298,7 @@ void MessagePortPipe::detach(uint8_t side) s.state.fetch_and(~uint64_t(Attached | DrainScheduled), std::memory_order_acq_rel); } -void MessagePortPipe::close(uint8_t side) +void MessagePortPipe::close(uint8_t side, bool closedByScript) { ASSERT(side < 2); @@ -224,12 +308,19 @@ void MessagePortPipe::close(uint8_t side) // chain of nested transferred ports overflows the native stack. Drain the // cascade iteratively instead: steal transferred pipes from each batch of // dropped messages into a stack-local worklist and close them in a loop. + // + // `closedByScript` applies only to the top-level side being closed; the + // transferred ports harvested below are dropped in transit, not + // script-closed, so they never set ClosedByScript. Vector, uint8_t>> worklist; worklist.append({ this, side }); + bool topLevel = true; while (!worklist.isEmpty()) { auto [pipe, sd] = worklist.takeLast(); auto& s = pipe->m_sides[sd]; + uint64_t closedState = Closed | (topLevel && closedByScript ? ClosedByScript : 0); + topLevel = false; Deque dropped; { @@ -237,7 +328,7 @@ void MessagePortPipe::close(uint8_t side) s.ctxId = 0; s.port = nullptr; // Closed is terminal; queued messages are dropped. - s.state.store(Closed, std::memory_order_release); + s.state.store(closedState, std::memory_order_release); dropped = std::exchange(s.inbox, {}); } diff --git a/src/jsc/bindings/webcore/MessagePortPipe.h b/src/jsc/bindings/webcore/MessagePortPipe.h index ff421a0952ba..6d59e3787c43 100644 --- a/src/jsc/bindings/webcore/MessagePortPipe.h +++ b/src/jsc/bindings/webcore/MessagePortPipe.h @@ -44,6 +44,12 @@ class MessagePortPipe final : public ThreadSafeRefCounted { Closed = 1ull << 0, // close() was called on this side; drops further deliveries. DrainScheduled = 1ull << 1, // a drain task for this side is in flight. Attached = 1ull << 2, // ctxId/port are valid; ok to schedule drains. + // Closed specifically via an explicit script close() (MessagePort::close + // while JS can run), as opposed to GC (~MessagePort), drop-in-transit + // (~TransferredMessagePort), or context teardown. Only a script close + // fires the peer's 'close' event; deriving that from the bare Closed bit + // would make GC/teardown timing observable from JS. + ClosedByScript = 1ull << 3, QueuedShift = 8, QueuedOne = 1ull << QueuedShift, @@ -62,11 +68,21 @@ class MessagePortPipe final : public ThreadSafeRefCounted { // means "just buffer, don't dispatch" (used before start()). void attach(uint8_t side, ScriptExecutionContextIdentifier, ThreadSafeWeakPtr); void detach(uint8_t side); - void close(uint8_t side); + // `closedByScript` records whether this is an explicit script close() (which + // fires the peer's 'close') vs a GC/teardown/drop close (which does not). + void close(uint8_t side, bool closedByScript = false); + + // Schedules the entangled peer to dispatch its 'close' event after this + // side has been marked Closed. Only safe from a live (script-running) + // context, never from a destructor or teardown — see the impl comment. + void wakePeerForClose(uint8_t side); // Lockless snapshot for the GC visitor / hasPendingActivity. uint64_t state(uint8_t side) const { return m_sides[side].state.load(std::memory_order_acquire); } bool isOtherSideOpen(uint8_t side) const { return !(state(1 - side) & Closed); } + // Whether the peer closed via an explicit script close() (the only kind + // that should fire this side's 'close' event). + bool isOtherSideClosedByScript(uint8_t side) const { return state(1 - side) & ClosedByScript; } // Equality is by identity; used to reject "port posted through itself". bool operator==(const MessagePortPipe& other) const { return this == &other; } @@ -77,6 +93,11 @@ class MessagePortPipe final : public ThreadSafeRefCounted { void scheduleDrain(uint8_t side, ScriptExecutionContextIdentifier); void drainAndDispatch(uint8_t side, ScriptExecutionContextIdentifier expectedCtx); + // Whether `side` may dispatch its peer-close event now: peer is closed and, + // re-checked under the lock, this side is still drained with no drain + // re-armed by a concurrent send(). Guards a cross-thread TOCTOU. + bool readyToDispatchClose(uint8_t side); + struct Side { WTF::Lock lock; WTF::Deque inbox WTF_GUARDED_BY_LOCK(lock); diff --git a/test/js/web/workers/message-port-pipe.test.ts b/test/js/web/workers/message-port-pipe.test.ts index e928eb4f11bd..a6a1c88a876f 100644 --- a/test/js/web/workers/message-port-pipe.test.ts +++ b/test/js/web/workers/message-port-pipe.test.ts @@ -365,6 +365,249 @@ describe("MessagePort pipe", () => { }); }); +// Closing one end of a MessageChannel must fire a 'close' event on both the +// port that was closed and its entangled peer, asynchronously, after any +// already-queued messages have been delivered. This matches Node's +// node:worker_threads MessagePort and the HTML MessagePort 'close' event. +// https://github.com/oven-sh/bun/issues/32563 +describe("MessagePort close event", () => { + async function run(src: string) { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", src], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + return { stdout: stdout.trim(), stderr, exitCode }; + } + + test.concurrent("fires on the peer after queued messages drain (issue repro)", async () => { + const { stdout, stderr, exitCode } = await run(` + const { MessageChannel } = require("node:worker_threads"); + const { port1, port2 } = new MessageChannel(); + port2.on("message", (message) => console.log(message)); + port2.on("close", () => console.log("closed!")); + port1.postMessage("foobar"); + port1.close(); + `); + expect(stderr).toBe(""); + expect(stdout).toBe("foobar\nclosed!"); + expect(exitCode).toBe(0); + }); + + test.concurrent("fires on both ports; message delivered before close", async () => { + const { stdout, stderr, exitCode } = await run(` + const { MessageChannel } = require("node:worker_threads"); + const { port1, port2 } = new MessageChannel(); + const log = []; + let n = 0; + const finish = () => { if (++n === 2) console.log(JSON.stringify(log)); }; + port1.on("close", () => { log.push("port1 close"); finish(); }); + port2.on("message", (m) => log.push("port2 message: " + m)); + port2.on("close", () => { log.push("port2 close"); finish(); }); + port1.postMessage("foobar"); + port1.close(); + `); + expect(stderr).toBe(""); + const log = JSON.parse(stdout); + // Both ports get 'close'; the queued message drains first on port2. + expect(log).toContain("port1 close"); + expect(log).toContain("port2 close"); + expect(log.indexOf("port2 message: foobar")).toBeGreaterThanOrEqual(0); + expect(log.indexOf("port2 message: foobar")).toBeLessThan(log.indexOf("port2 close")); + expect(exitCode).toBe(0); + }); + + test.concurrent("closing port fires its own close event", async () => { + const { stdout, stderr, exitCode } = await run(` + const { MessageChannel } = require("node:worker_threads"); + const { port1, port2 } = new MessageChannel(); + port1.on("close", () => console.log("port1 closed")); + port2.on("message", () => {}); + port1.close(); + `); + expect(stderr).toBe(""); + expect(stdout).toBe("port1 closed"); + expect(exitCode).toBe(0); + }); + + test.concurrent("Web API: addEventListener('close') on MessageChannel", async () => { + const { stdout, stderr, exitCode } = await run(` + const { port1, port2 } = new MessageChannel(); + port2.addEventListener("message", (e) => console.log("msg:" + e.data)); + port2.addEventListener("close", () => console.log("closed")); + port1.postMessage("x"); + port1.close(); + `); + expect(stderr).toBe(""); + expect(stdout).toBe("msg:x\nclosed"); + expect(exitCode).toBe(0); + }); + + test.concurrent("close fires at most once", async () => { + const { stdout, stderr, exitCode } = await run(` + const { MessageChannel } = require("node:worker_threads"); + const { port1, port2 } = new MessageChannel(); + let count = 0; + port1.on("close", () => count++); + port2.on("message", () => {}); + port1.close(); + port1.close(); + process.on("exit", () => console.log("count=" + count)); + `); + expect(stderr).toBe(""); + expect(stdout).toBe("count=1"); + expect(exitCode).toBe(0); + }); + + test.concurrent("no close listener: channel closes and process exits cleanly", async () => { + const { stdout, stderr, exitCode } = await run(` + const { MessageChannel } = require("node:worker_threads"); + const { port1, port2 } = new MessageChannel(); + port2.on("message", (m) => console.log(m)); + port1.postMessage("only-message"); + port1.close(); + `); + expect(stderr).toBe(""); + expect(stdout).toBe("only-message"); + expect(exitCode).toBe(0); + }); + + test.concurrent("fires on a peer that has only a close listener (no message listener)", async () => { + const { stdout, stderr, exitCode } = await run(` + const { MessageChannel } = require("node:worker_threads"); + const { port1, port2 } = new MessageChannel(); + port2.on("close", () => console.log("closed")); + port1.close(); + `); + expect(stderr).toBe(""); + expect(stdout).toBe("closed"); + expect(exitCode).toBe(0); + }); + + test.concurrent("fires when the close listener is added after the peer already closed", async () => { + const { stdout, stderr, exitCode } = await run(` + const { MessageChannel } = require("node:worker_threads"); + const { port1, port2 } = new MessageChannel(); + port1.close(); + port2.on("close", () => console.log("closed")); + `); + expect(stderr).toBe(""); + expect(stdout).toBe("closed"); + expect(exitCode).toBe(0); + }); + + test.concurrent("fires on the closing port when its own close listener is added after close()", async () => { + const { stdout, stderr, exitCode } = await run(` + const { MessageChannel } = require("node:worker_threads"); + const { port1 } = new MessageChannel(); + port1.close(); + port1.on("close", () => console.log("closed")); + `); + expect(stderr).toBe(""); + expect(stdout).toBe("closed"); + expect(exitCode).toBe(0); + }); + + test.concurrent("a close-only port does not drop queued messages", async () => { + const { stdout, stderr, exitCode } = await run(` + const { MessageChannel, receiveMessageOnPort } = require("node:worker_threads"); + const { port1, port2 } = new MessageChannel(); + port2.on("close", () => {}); // close-only: attaches port2 but no message listener + port1.postMessage("buffered"); + port1.close(); + // Drain the scheduled task: with no message listener, "buffered" must + // stay queued (not dispatched to no one), so it is still retrievable. + await Bun.sleep(0); + const m = receiveMessageOnPort(port2); + console.log(m ? m.message : "LOST"); + `); + expect(stderr).toBe(""); + expect(stdout).toBe("buffered"); + expect(exitCode).toBe(0); + }); + + test.concurrent("a peer closed by drop-in-transit (not an explicit close) does not fire close", async () => { + const { stdout, stderr, exitCode } = await run(` + const { MessageChannel } = require("node:worker_threads"); + const { port1: A, port2: B } = new MessageChannel(); + B.close(); + const inner = new MessageChannel(); + // inner.port1 is dropped in transit (B is closed), closing its pipe side + // via ~TransferredMessagePort — NOT an explicit script close(). + A.postMessage(null, [inner.port1]); + A.close(); + const log = []; + inner.port2.on("close", () => log.push("close")); // listener added after the drop + await Bun.sleep(0); + // A non-script close must not surface as a 'close' event (otherwise GC / + // drop / teardown timing would be observable from JS). + console.log(JSON.stringify(log)); + `); + expect(stderr).toBe(""); + expect(stdout).toBe("[]"); + expect(exitCode).toBe(0); + }); + + // Regression guard: a close listener added after close() (or a close-only + // peer) must not pin the JS wrapper for the lifetime of the context. + test.concurrent("closed ports with close listeners are collected", async () => { + const { stdout, stderr, exitCode } = await run(` + const { heapStats } = require("bun:jsc"); + const { MessageChannel } = require("node:worker_threads"); + const count = () => heapStats().objectTypeCounts.MessagePort || 0; + async function settle() { for (let i = 0; i < 5; i++) { Bun.gc(true); await Bun.sleep(0); } } + + await settle(); + const base = count(); + const N = 100; + for (let i = 0; i < N; i++) { + const { port1, port2 } = new MessageChannel(); + port1.close(); + port1.addEventListener("close", () => {}); // listener added AFTER close + port2.addEventListener("close", () => {}); // close-only peer of a closed port + + // close-only port whose peer posted a message and then closed: close is + // blocked behind the undelivered buffered message (no message listener), + // so it has no dispatch path and must not be pinned. + const ch = new MessageChannel(); + ch.port2.addEventListener("close", () => {}); + ch.port1.postMessage("x"); + ch.port1.close(); + + // peer closed via drop-in-transit (no peer-wake): a port with message + // and close listeners must not be pinned when no drain was scheduled. + const { port1: A, port2: B } = new MessageChannel(); + B.close(); + const inner = new MessageChannel(); + inner.port2.addEventListener("message", () => {}); + inner.port2.addEventListener("close", () => {}); + A.postMessage(null, [inner.port1]); // inner.port1 dropped in transit + A.close(); + + // explicitly started port with no message listener: a buffered message + // on a closed peer must dispatch (to no one) and not strand/pin it. + const ch2 = new MessageChannel(); + ch2.port2.start(); + ch2.port1.postMessage("y"); + ch2.port1.close(); + } + await settle(); + // 10 MessagePorts per iteration; each of the four scenarios guards one + // port, so regressing a single scenario pins ~N of them. Threshold N/2 + // catches that while staying well above the single-digit residue that + // conservative stack scanning leaves after settle(). + const leaked = count() - base; + if (leaked > N / 2) { console.error("leaked " + leaked + " MessagePort"); process.exit(1); } + console.log("OK"); + `); + expect(stderr).toBe(""); + expect(stdout).toBe("OK"); + expect(exitCode).toBe(0); + }); +}); + // worker.postMessage / parentPort.postMessage go through the same coalesced // inbox+batch-drain path as MessagePortPipe. Verify the observable ordering // matches Node: messages arrive in order with a microtask checkpoint between