MessagePort: close a port whose peer closed, so it is rejected as a transferable

src/jsc/bindings/webcore/MessagePort.cpp

@@ -50,6 +50,8 @@ Ref<MessagePort> MessagePort::create(ScriptExecutionContext& context, Ref<Messag
 {
     auto messagePort = adoptRef(*new MessagePort(context, WTF::move(pipe), side));
     messagePort->suspendIfNeeded();
+    // So the peer's close() reaches this port (peerClosed()) even if it never gets a listener.
+    messagePort->m_pipe->registerCloseContext(side, context.identifier(), ThreadSafeWeakPtr<MessagePort> { messagePort.get() });
     return messagePort;
 }
 
@@ -271,24 +273,33 @@ void MessagePort::dispatchCloseEvent()
 
 void MessagePort::peerClosed()
 {
-    if (m_isDetached)
+    if (m_isDetached || m_isClosing)
         return;
     auto* context = scriptExecutionContext();
     if (!context || !context->globalObject())
         return;
     Ref protectedThis { *this };
-    // Deliver whatever the peer sent before it closed, then fire 'close'. Node orders
-    // them that way, and registerCloseContext()'s retroactive notify can land before any
-    // drain is scheduled -- e.g. on('close') registered before on('message').
+    // Node delivers what the peer sent before closing first; this task can run before the drain.
     if (m_started && hasMessageEventListener())
         flushQueuedMessagesBeforeClose();
-    // 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.
-    auto* globalObject = defaultGlobalObject(context->globalObject());
-    jsUnref(globalObject);
+    // A handler in that flush closed or transferred this port.
+    if (m_isDetached || m_isClosing)
+        return;
+    if (!m_pipe->isOtherSideClosedByRequest(m_side)) {
+        // Peer collected, not closed: node never closes a channel over that (see jsRef()), so
+        // only release the loop refs of a port that is started or listening for 'close'. An
+        // idle port keeps its one 'close' event for its own close(cb).
+        if (m_started || m_hasCloseEventListener.load(std::memory_order_relaxed)) {
+            dispatchCloseEvent();
+            jsUnref(defaultGlobalObject(context->globalObject()));
+        }
+        return;
+    }
+    // Not receiving, messages still queued: node's close notification waits behind them.
+    // attach() notifies again once the port starts; tryTakeMessage() closes on the empty read.
+    if (MessagePortPipe::queuedCount(m_pipe->state(m_side)) > 0)
+        return;
+    close();
 }
 
 TransferredMessagePort MessagePort::disentangle()
@@ -381,8 +392,14 @@ JSValue MessagePort::tryTakeMessage(JSGlobalObject* lexicalGlobalObject, bool& h
         return jsUndefined();
 
     auto message = m_pipe->takeOne(m_side);
-    if (!message)
+    if (!message) {
+        // This read reached the peer's close: node closes the port here (see peerClosed()).
+        // Peer state before our queue, as in virtualHasPendingActivity(): a message the peer
+        // sent between takeOne() and its close() is then visible and must not be dropped.
+        if (m_pipe->isOtherSideClosedByRequest(m_side) && !MessagePortPipe::queuedCount(m_pipe->state(m_side)))
+            close();
         return jsUndefined();
+    }
 
     hadMessage = true;
     auto ports = MessagePort::entanglePorts(*context, WTF::move(message->transferredPorts));
@@ -525,12 +542,6 @@ bool MessagePort::addEventListener(const AtomString& eventType, Ref<EventListene
         }
     } else if (eventType == eventNames().closeEvent) {
         m_hasCloseEventListener.store(true, std::memory_order_release);
-        if (isEntangled()) {
-            // Record our context with the pipe so the peer's close() can deliver a
-            // 'close' event even if we never started (no 'message' listener).
-            if (auto* context = scriptExecutionContext())
-                m_pipe->registerCloseContext(m_side, context->identifier(), ThreadSafeWeakPtr<MessagePort> { *this });
-        }
     }
     return EventTarget::addEventListener(eventType, WTF::move(listener), options);
 }

src/jsc/bindings/webcore/MessagePort.h

@@ -79,8 +79,7 @@ class MessagePort final : public ActiveDOMObject, public EventTarget, public Thr
     // The worker's entry module finished evaluating: a start() requested before that takes effect now.
     void entrySettled();
     void close();
-    // Called on the entangled peer when this side closes: dispatches a
-    // 'close' event and releases the event-loop ref so the loop can idle.
+    // Posted by the pipe to the peer of a side that closed: closes this port too once its queue is drained.
     void peerClosed();
     void dispatchCloseEvent();
 

src/jsc/bindings/webcore/MessagePortPipe.cpp

@@ -234,9 +234,8 @@ void MessagePortPipe::attach(uint8_t side, ScriptExecutionContextIdentifier ctxI
     }
     if (wakeCtx)
         scheduleDrain(side, wakeCtx);
-    // Peer already closed while this side was in transit (detach() cleared
-    // ContextKnown so notifyPeerClosed() early-returned): re-deliver to the new
-    // owner, or the receiving context's listener loop-ref is never released.
+    // Peer already closed: notify again now that this side is receiving. The task runs after the
+    // drain scheduled above, so a close that peerClosed() deferred behind queued messages completes.
     if (m_sides[1 - side].state.load(std::memory_order_acquire) & Closed)
         notifyPeerClosed(side);
 }

src/jsc/bindings/webcore/MessagePortPipe.h

@@ -64,9 +64,8 @@ class MessagePortPipe final : public ThreadSafeRefCounted<MessagePortPipe> {
     // already queued (e.g. after transfer). Passing a null port is allowed and
     // means "just buffer, don't dispatch" (used before start()).
     void attach(uint8_t side, ScriptExecutionContextIdentifier, ThreadSafeWeakPtr<MessagePort>);
-    // Like attach() but only records ctxId/port so the peer's close() can deliver
-    // a 'close' event to a port that never started (no 'message' listener). Does
-    // NOT enable drains. No-op if already attached/registered/closed.
+    // Called when a MessagePort is created for this side: records ctxId/port so the peer's close()
+    // reaches a port that never started, without enabling drains. No-op once attached/registered/closed.
     void registerCloseContext(uint8_t side, ScriptExecutionContextIdentifier, ThreadSafeWeakPtr<MessagePort>);
     void detach(uint8_t side);
     // Explicit == a real, permanent close: close(), context teardown, or an orphaned

test/js/node/worker_threads/worker_threads.test.ts

@@ -1785,6 +1785,267 @@ test("MessagePort: peer closing while a port is in transit still delivers 'close
   });
 });
 
+// When one port of a channel is close()d, node closes the other one as well (once the
+// close has propagated): it fires 'close', drops its loop refs, and from then on it is
+// rejected as a transferable like any closed port. Bun fired 'close' on the peer but
+// left it open, so a dead endpoint could be transferred; a peer with no listeners was
+// not told at all.
+describe("a port whose peer closed is closed too", () => {
+  const detached = expect.objectContaining({
+    name: "DataCloneError",
+    message: "MessagePort in transfer list is already detached",
+  });
+  const tick = () => new Promise(r => setImmediate(r));
+
+  // Probe without transferring anything: the trailing plain object is itself invalid, so the
+  // post always throws, and the message says whether `port` (checked first, in list order)
+  // was rejected as detached.
+  function isRejectedAsDetached(port: MessagePort): boolean {
+    const carrier = new MessageChannel();
+    let message: string | undefined;
+    try {
+      carrier.port1.postMessage(null, [port, {} as any]);
+    } catch (e: any) {
+      message = e.message;
+    } finally {
+      carrier.port1.close();
+      carrier.port2.close();
+    }
+    if (message === undefined) throw new Error("the probe post must throw");
+    return message === "MessagePort in transfer list is already detached";
+  }
+
+  test("with a 'close' listener: rejected by postMessage and structuredClone, transfer stays atomic", async () => {
+    const { port1, port2 } = new MessageChannel();
+    const carrier = new MessageChannel();
+    const closed = once(port1, "close");
+    port2.close();
+    await closed;
+
+    const ab = new ArrayBuffer(8);
+    expect(() => carrier.port1.postMessage(null, [ab, port1])).toThrow(detached);
+    // Same as node: a bad port later in the list means nothing earlier in it is transferred.
+    expect(ab.byteLength).toBe(8);
+    expect(() => structuredClone(port1, { transfer: [port1] })).toThrow(detached);
+    expect(port1.hasRef()).toBe(false);
+    carrier.port1.close();
+    carrier.port2.close();
+  });
+
+  test("with a 'message' listener: what the peer sent first is delivered, then the port closes", async () => {
+    const { port1, port2 } = new MessageChannel();
+    const events: string[] = [];
+    port1.on("message", m => events.push(m));
+    const closed = once(port1, "close");
+    port2.postMessage("m1");
+    port2.close();
+    await closed;
+    events.push("close");
+
+    expect(events).toEqual(["m1", "close"]);
+    expect(isRejectedAsDetached(port1)).toBe(true);
+    expect(port1.hasRef()).toBe(false);
+  });
+
+  test("the port is already closed when its own 'close' listener runs", async () => {
+    const { port1, port2 } = new MessageChannel();
+    const { promise, resolve } = Promise.withResolvers<boolean>();
+    port1.on("close", () => resolve(isRejectedAsDetached(port1)));
+    port2.close();
+    expect(await promise).toBe(true);
+  });
+
+  test("with no listeners at all", async () => {
+    const { port1, port2 } = new MessageChannel();
+    port2.close();
+    // The close reaches port1 as a task; poll for it rather than assume how many turns it takes.
+    let rejected = false;
+    for (let i = 0; i < 100 && !rejected; i++) {
+      await tick();
+      rejected = isRejectedAsDetached(port1);
+    }
+    expect(rejected).toBe(true);
+  });
+
+  test("Worker.postMessage() and the Worker constructor's transferList reject it as well", async () => {
+    const { port1, port2 } = new MessageChannel();
+    const closed = once(port1, "close");
+    port2.close();
+    await closed;
+
+    const worker = new Worker("", { eval: true });
+    try {
+      expect(() => worker.postMessage(port1, [port1])).toThrow(detached);
+      expect(() => new Worker("", { eval: true, workerData: port1, transferList: [port1] })).toThrow(detached);
+    } finally {
+      await worker.terminate();
+    }
+  });
+
+  // node's close notification sits behind whatever the peer posted before closing: a port
+  // that is not receiving keeps those messages and stays open until they are consumed.
+  test("a port that is not receiving stays open until a 'message' listener drains the queue", async () => {
+    const { port1, port2 } = new MessageChannel();
+    const events: string[] = [];
+    const closed = once(port1, "close");
+    port2.postMessage("m1");
+    port2.postMessage("m2");
+    port2.close();
+    for (let i = 0; i < 4; i++) await tick();
+    expect({ events, rejected: isRejectedAsDetached(port1) }).toEqual({ events: [], rejected: false });
+
+    port1.on("message", m => events.push(m));
+    await closed;
+    events.push("close");
+    expect(events).toEqual(["m1", "m2", "close"]);
+    expect(isRejectedAsDetached(port1)).toBe(true);
+  });
+
+  test("a port that is not receiving stays open until receiveMessageOnPort() drains the queue", async () => {
+    const { port1, port2 } = new MessageChannel();
+    let closes = 0;
+    const closed = once(port1, "close");
+    port1.on("close", () => closes++);
+    port2.postMessage("m1");
+    port2.close();
+    for (let i = 0; i < 4; i++) await tick();
+    expect({ closes, rejected: isRejectedAsDetached(port1) }).toEqual({ closes: 0, rejected: false });
+
+    expect(receiveMessageOnPort(port1)).toEqual({ message: "m1" });
+    // Taking the last message does not close it yet (node closes on reaching the notification).
+    expect({ closes, rejected: isRejectedAsDetached(port1) }).toEqual({ closes: 0, rejected: false });
+
+    // This call reaches the peer's close: no message, and the port is closed on the spot.
+    expect(receiveMessageOnPort(port1)).toBeUndefined();
+    expect({ closes, rejected: isRejectedAsDetached(port1) }).toEqual({ closes: 0, rejected: true });
+    await closed;
+    expect(closes).toBe(1);
+  });
+
+  test("a port left open that way is still a valid transferable; the receiver gets the queue and the close", async () => {
+    const { port1, port2 } = new MessageChannel();
+    const carrier = new MessageChannel();
+    let senderCloses = 0;
+    port1.on("close", () => senderCloses++);
+    port2.postMessage("m1");
+    port2.close();
+    for (let i = 0; i < 4; i++) await tick();
+    expect(senderCloses).toBe(0);
+
+    carrier.port1.postMessage(port1, [port1]);
+    const [received] = (await once(carrier.port2, "message")) as [MessagePort];
+    const events: string[] = [];
+    const closed = once(received, "close");
+    received.on("message", m => events.push(m));
+    await closed;
+    events.push("close");
+    expect(events).toEqual(["m1", "close"]);
+    carrier.port1.close();
+    carrier.port2.close();
+  });
+
+  // Makes a channel, drops one port and collects it (its WeakRef clearing is the proof; the
+  // collection runs ~MessagePort, which posts the peer notification), then gives the surviving
+  // port a turn to process that notification and returns it.
+  async function survivorOfCollectedPeer(prepare: (survivor: MessagePort, doomed: MessagePort) => void) {
+    const { survivor, doomed } = (() => {
+      const { port1, port2 } = new MessageChannel();
+      prepare(port1, port2);
+      return { survivor: port1, doomed: new WeakRef(port2) };
+    })();
+    let collected = false;
+    for (let i = 0; i < 20 && !collected; i++) {
+      // new WeakRef() / deref() keep the target alive for the rest of the current turn.
+      await tick();
+      Bun.gc(true);
+      collected = doomed.deref() === undefined;
+    }
+    expect(collected).toBe(true);
+    await tick();
+    return survivor;
+  }
+
+  // Only an explicit close counts. Bun collects an entangled port whose wrapper is unreachable
+  // (node never does) and notifies the peer so it releases its loop refs; closing the peer on
+  // top of that would make its transfer fail or not depending on when GC ran.
+  test("a peer that was garbage collected rather than closed leaves the port transferable", async () => {
+    const carrier = new MessageChannel();
+    const port1 = await survivorOfCollectedPeer(survivor => survivor.on("close", () => {}));
+    expect(isRejectedAsDetached(port1)).toBe(false);
+    carrier.port1.postMessage(port1, [port1]);
+    carrier.port1.close();
+    carrier.port2.close();
+  });
+
+  // The wait-for-the-queue-to-drain rule is for an explicit close only: nothing would ever
+  // re-notify a 'close'-listening port that never starts, and until 'close' has been dispatched
+  // such a port is pinned in the heap. This is what main did before the change, kept as is.
+  test("a 'close'-listening port with an unread message is still told when its peer is garbage collected", async () => {
+    let closes = 0;
+    const port1 = await survivorOfCollectedPeer((survivor, doomed) => {
+      survivor.on("close", () => closes++);
+      doomed.postMessage("unread");
+    });
+    for (let i = 0; i < 20 && closes === 0; i++) await tick();
+    expect({ closes, unread: receiveMessageOnPort(port1), rejected: isRejectedAsDetached(port1) }).toEqual({
+      closes: 1,
+      unread: { message: "unread" },
+      rejected: false,
+    });
+  });
+
+  // ...and an idle port is not even told: its one 'close' event has to stay available for
+  // the close() that eventually happens (node's test-worker-message-port.js relies on this).
+  test("a port whose peer was garbage collected still delivers close(cb) when it is closed later", async () => {
+    const idle = await survivorOfCollectedPeer(() => {});
+    const unread = await survivorOfCollectedPeer((_survivor, doomed) => doomed.postMessage("queued"));
+
+    // 'close' is dispatched from a task queued by close(); give it a few turns, then give up.
+    function closeCallbackFires(port: MessagePort): Promise<string> {
+      return Promise.race([
+        new Promise<string>(resolve => port.close(() => resolve("called"))),
+        (async () => {
+          for (let i = 0; i < 20; i++) await tick();
+          return "not called";
+        })(),
+      ]);
+    }
+
+    expect(await closeCallbackFires(idle)).toBe("called");
+    // close() from inside the 'message' handler, as node's test does.
+    const unreadResult = await new Promise<string>(resolve => {
+      unread.on("message", m => closeCallbackFires(unread).then(cb => resolve(`${m}, close(cb) ${cb}`)));
+    });
+    expect(unreadResult).toBe("queued, close(cb) called");
+  });
+
+  // The same missing notification left a ref()'d port with no listeners pinning the loop
+  // forever after its peer closed; node's port closes and the process exits.
+  test("a ref()'d port with no listeners lets the process exit once its peer closes", async () => {
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `const { MessageChannel } = require("worker_threads");
+         const { port1, port2 } = new MessageChannel();
+         port1.ref();
+         port2.close();
+         // unref'd, so it only ever fires if something else (the port) is still holding the loop open.
+         setTimeout(() => { console.log("still running, hasRef=" + port1.hasRef()); process.exit(1); }, 3000).unref();
+         process.on("exit", code => console.log("exit " + code + ", hasRef=" + port1.hasRef()));`,
+      ],
+      env: bunEnv,
+      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: "exit 0, hasRef=false",
+      stderr: "",
+      exitCode: 0,
+    });
+  });
+});
+
 test("workerData is not unwrapped for a non-node globalThis.Worker", async () => {
   await using proc = Bun.spawn({
     cmd: [