MessagePort: don't drop messages when a GC runs during dispatch

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.

src/jsc/bindings/webcore/MessagePortPipe.h

@@ -48,6 +48,9 @@ class MessagePortPipe final : public ThreadSafeRefCounted<MessagePortPipe> {
         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.

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