From a4fb86d4da67de5baae07261cfb3eab997fe0539 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Mon, 29 Jun 2026 07:33:33 +0000 Subject: [PATCH 1/2] MessagePort: keep the in-flight message counted until dispatch returns drainAndDispatch() popped a message and decremented the pipe's queued count before deserializing and dispatching it. Once the peer port is closed, that count is all that keeps hasPendingActivity() true, so a GC triggered by the allocating deserialization could collect the JS wrapper (and with it the weakly held message listener) while the message was being delivered: release builds silently dropped the message and debug builds hit "ASSERTION FAILED: m_wrapper" in JSEventListener::ensureJSFunction. Release the QueuedOne unit only after dispatchOneMessage() returns, so the wrapper stays reachable for the whole pop to dispatch window. close() resets the state word, so the release is skipped if the handler closed the port. Regression from #29937 (v1.3.14). --- src/jsc/bindings/webcore/MessagePortPipe.cpp | 13 ++++- src/jsc/bindings/webcore/MessagePortPipe.h | 3 ++ test/js/web/workers/message-port-pipe.test.ts | 53 +++++++++++++++++++ 3 files changed, 68 insertions(+), 1 deletion(-) diff --git a/src/jsc/bindings/webcore/MessagePortPipe.cpp b/src/jsc/bindings/webcore/MessagePortPipe.cpp index a782916564c0..0433b5c5c829 100644 --- a/src/jsc/bindings/webcore/MessagePortPipe.cpp +++ b/src/jsc/bindings/webcore/MessagePortPipe.cpp @@ -150,12 +150,23 @@ void MessagePortPipe::drainAndDispatch(uint8_t side, ScriptExecutionContextIdent rescheduleCtx = s.ctxId; break; } + // QueuedOne for this message is released after dispatch, below. message = s.inbox.takeFirst(); - s.state.store(st - QueuedOne, std::memory_order_release); } port->dispatchOneMessage(*context, WTF::move(*message)); + // Decrement only after dispatch: once the peer is closed this count is + // all that keeps hasPendingActivity() true, so a GC during the + // deserialize would collect the wrapper + listeners and drop the message. + { + Locker locker { s.lock }; + uint64_t st = s.state.load(std::memory_order_relaxed); + // close() from inside the handler already reset the state word. + if (queuedCount(st) > 0) + s.state.store(st - QueuedOne, std::memory_order_release); + } + // Node's MakeCallback wraps each emit in an InternalCallbackScope, // which drains nextTick + microtasks on exit; match that so // queueMicrotask(cb) inside onmessage runs before the next message. diff --git a/src/jsc/bindings/webcore/MessagePortPipe.h b/src/jsc/bindings/webcore/MessagePortPipe.h index ff421a0952ba..d0a2578db00a 100644 --- a/src/jsc/bindings/webcore/MessagePortPipe.h +++ b/src/jsc/bindings/webcore/MessagePortPipe.h @@ -48,6 +48,9 @@ class MessagePortPipe final : public ThreadSafeRefCounted { QueuedShift = 8, QueuedOne = 1ull << QueuedShift, }; + // Count of undelivered messages: the inbox plus the message currently + // being dispatched (its unit is released only once dispatch returns, so + // hasPendingActivity() keeps the receiving wrapper alive throughout). static constexpr uint64_t queuedCount(uint64_t s) { return s >> QueuedShift; } // Sender-thread operations. diff --git a/test/js/web/workers/message-port-pipe.test.ts b/test/js/web/workers/message-port-pipe.test.ts index e928eb4f11bd..9d6973de7258 100644 --- a/test/js/web/workers/message-port-pipe.test.ts +++ b/test/js/web/workers/message-port-pipe.test.ts @@ -228,6 +228,59 @@ describe("MessagePort pipe", () => { expect(exitCode).toBe(0); }); + // A message popped off the inbox but not yet dispatched must still count + // as pending activity: once the peer is closed, that count is all that + // keeps the receiving wrapper (and, via it, the listener functions) alive, + // and the deserialization inside the dispatch allocates. A GC there used + // to collect the wrapper and silently drop the message (on debug builds: + // "ASSERTION FAILED: m_wrapper" in JSEventListener::ensureJSFunction). + test("in-flight message keeps an otherwise-unreferenced listening port alive across GC", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const N = 100; + // Deserializing a large string reports its size as extra GC memory, + // so collections trigger inside the dispatch of some of the messages. + const big = Buffer.alloc(1 << 20, "x").toString(); + let fired = 0; + for (let i = 0; i < N; i++) { + const { port1, port2 } = new MessageChannel(); + // The handler must not capture port1; nothing but the pipe's + // undelivered-message count roots the wrapper once port2 closes. + port1.onmessage = () => { fired++; }; + port2.postMessage(big); + port2.close(); + } + // Every postMessage above scheduled its drain task, in order, on this + // context's task queue; this channel's drain therefore runs last. + await new Promise(resolve => { + const { port1, port2 } = new MessageChannel(); + port1.onmessage = () => { port1.close(); port2.close(); resolve(); }; + port2.postMessage(null); + }); + console.log("fired=" + fired + "/" + N); + process.exit(0); + `, + ], + env: { + ...bunEnv, + // Keep the JSC heap small so the deserialization inside the dispatch + // window reliably triggers a collection. + BUN_JSC_forceRAMSize: String(16 * 1024 * 1024), + }, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout: stdout.trim(), stderr, exitCode }).toEqual({ + stdout: "fired=100/100", + stderr: "", + exitCode: 0, + }); + }); + // A port transferred through a carrier whose destination is already // closed never reaches a new owner. The endpoint must be marked Closed // when the in-transit struct is dropped, otherwise the peer's From e613491641f417a50383eb2079fbd6e9178436b7 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Mon, 29 Jun 2026 07:44:00 +0000 Subject: [PATCH 2/2] test: report but do not constrain stderr in the GC stress subprocess --- test/js/web/workers/message-port-pipe.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/test/js/web/workers/message-port-pipe.test.ts b/test/js/web/workers/message-port-pipe.test.ts index 9d6973de7258..d46ee5a13907 100644 --- a/test/js/web/workers/message-port-pipe.test.ts +++ b/test/js/web/workers/message-port-pipe.test.ts @@ -274,9 +274,11 @@ describe("MessagePort pipe", () => { stderr: "pipe", }); const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + // stderr is reported but not constrained: debug/ASAN lanes may emit + // benign diagnostics under the GC pressure this child creates. expect({ stdout: stdout.trim(), stderr, exitCode }).toEqual({ stdout: "fired=100/100", - stderr: "", + stderr: expect.any(String), exitCode: 0, }); });