MessagePort: fire 'close' on a port that is transferred away

src/jsc/bindings/webcore/MessagePort.cpp

@@ -241,9 +241,15 @@ void MessagePort::close()
         updateListenerEventLoopRef();
     }
 
+    queueCloseEvent();
+}
+
+void MessagePort::queueCloseEvent()
+{
     // Defer 'close' to a task (node fires it at uv close-callback timing, i.e.
-    // after sync code and microtasks), so a listener added after close() still
-    // observes it and close(cb) interleaves with other listeners.
+    // after sync code and microtasks), so a listener added after close() or the
+    // transfer still observes it and close(cb) interleaves with other listeners.
+    // Listeners stay registered until then; the task drops them once it has fired.
     if (isContextStopped()) {
         removeAllEventListeners();
         return;
@@ -295,16 +301,14 @@ TransferredMessagePort MessagePort::disentangle()
 {
     ASSERT(isEntangled());
 
-    // Drop any message listeners (and the event-loop ref they carry) while
-    // this port is still attached to its context; after observeContext(null)
-    // there would be nothing to unref.
-    removeAllEventListeners();
+    // Nothing is dispatched to this object after the transfer except the 'close'
+    // queued below; its listeners are dropped by that task.
     m_hasMessageEventListener = false;
 
     // Release the self-reference taken by jsRef() on the sending side. After
     // transfer this object is inert (the receiving side gets a fresh
-    // MessagePort for the same pipe endpoint) and is no longer a destruction
-    // observer, so nothing else will ever release a ref taken here.
+    // MessagePort for the same pipe endpoint) and close() no-ops on it once
+    // m_isDetached is set, so nothing else will ever release a ref taken here.
     // The caller (disentanglePorts) holds a RefPtr, so deref() is safe.
     if (m_hasRef) {
         m_hasRef = false;
@@ -329,12 +333,12 @@ TransferredMessagePort MessagePort::disentangle()
     m_isDetached = true;
     m_started = false;
 
-    // We can't receive any messages or generate any events after this, so remove ourselves from the list of active ports.
-    if (auto* context = scriptExecutionContext()) {
-        context->willDestroyActiveDOMObject(*this);
-        context->willDestroyDestructionObserver(*this);
-    }
-    observeContext(nullptr);
+    // Node closes the transferred-away object's handle, so it fires 'close' like a
+    // close()d port while the channel lives on in the receiver's new port (the peer
+    // is not told): https://github.com/nodejs/node/blob/v26.3.0/src/node_messaging.cc#L921-L924
+    // The task dispatches through our context, and only pins the wrapper while we are
+    // attached to one, so this object stays attached until collected, like a closed port.
+    queueCloseEvent();
 
     return TransferredMessagePort { m_pipe.copyRef(), m_side };
 }

src/jsc/bindings/webcore/MessagePort.h

@@ -84,7 +84,8 @@ class MessagePort final : public ActiveDOMObject, public EventTarget, public Thr
     void peerClosed();
     void dispatchCloseEvent();
 
-    // Transfer machinery.
+    // Transfer machinery. Like node, disentangle() closes the local object: it
+    // fires 'close' on it (its pipe side is not closed, the receiver takes it over).
     static ExceptionOr<Vector<TransferredMessagePort>> disentanglePorts(Vector<RefPtr<MessagePort>>&&);
     static Vector<RefPtr<MessagePort>> entanglePorts(ScriptExecutionContext&, Vector<TransferredMessagePort>&&);
     static Ref<MessagePort> entangle(ScriptExecutionContext&, TransferredMessagePort&&);
@@ -131,6 +132,8 @@ class MessagePort final : public ActiveDOMObject, public EventTarget, public Thr
 
     // Deliver messages already queued when close() is called, before teardown.
     void flushQueuedMessagesBeforeClose();
+    // Shared tail of close() and disentangle(): fires 'close' from a task, then drops the listeners.
+    void queueCloseEvent();
 
     bool isEntangled() const { return !m_isDetached; }
 

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

@@ -1223,6 +1223,113 @@ test("dropping a transferred port notifies its peer", async () => {
   port1.close();
 });
 
+// Transferring a port closes the sender's object (node closes its handle), so that object
+// fires 'close' once, asynchronously. The channel itself moves to the receiver: the peer
+// sees nothing, and the channel's eventual close goes to the received port.
+describe("a port transferred away fires 'close' on the sender's object", () => {
+  test("postMessage to a local port", async () => {
+    const { port1, port2 } = new MessageChannel();
+    const carrier = new MessageChannel();
+    const events: string[] = [];
+    port1.on("close", () => events.push("port1 close (listener added before the transfer)"));
+    port2.on("close", () => events.push("port2 close"));
+    carrier.port1.postMessage(port1, [port1]);
+    port1.on("close", () => events.push("port1 close (listener added after the transfer)"));
+    events.push("postMessage returned");
+    for (let i = 0; i < 2; i++) await new Promise(r => setImmediate(r));
+    expect(events).toEqual([
+      "postMessage returned",
+      "port1 close (listener added before the transfer)",
+      "port1 close (listener added after the transfer)",
+    ]);
+
+    // The channel is intact: the received port is entangled with port2.
+    const received = await new Promise<MessagePort>(resolve => carrier.port2.once("message", resolve));
+    received.postMessage("via the received port");
+    expect(await new Promise(resolve => port2.once("message", resolve))).toBe("via the received port");
+
+    // Closing the channel now closes port2 and the received port; the object that was
+    // transferred away does not fire a second time.
+    received.on("message", () => {});
+    const receivedClosed = new Promise<void>(resolve => received.once("close", () => resolve()));
+    port2.close();
+    await receivedClosed;
+    for (let i = 0; i < 2; i++) await new Promise(r => setImmediate(r));
+    expect(events).toEqual([
+      "postMessage returned",
+      "port1 close (listener added before the transfer)",
+      "port1 close (listener added after the transfer)",
+      "port2 close",
+    ]);
+    carrier.port1.close();
+    carrier.port2.close();
+  });
+
+  // The transfer is the last thing the script does: the event still fires, and the
+  // pending event does not keep the process alive afterwards.
+  test("the event fires before the process exits and does not keep it alive", async () => {
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `const { MessageChannel } = require("worker_threads");
+         const { port1 } = new MessageChannel();
+         const carrier = new MessageChannel();
+         port1.on("close", () => console.log("port1 close"));
+         carrier.port1.postMessage(port1, [port1]);`,
+      ],
+      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, signalCode: proc.signalCode }).toEqual({
+      stdout: "port1 close",
+      stderr: "",
+      exitCode: 0,
+      signalCode: null,
+    });
+  });
+
+  // Ports transferred through the Worker constructor's transferList, through
+  // worker.postMessage(), and from the worker through parentPort.postMessage().
+  test("transfers to and from a Worker", async () => {
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `const { Worker, MessageChannel } = require("worker_threads");
+         const viaConstructor = new MessageChannel();
+         const viaPostMessage = new MessageChannel();
+         viaConstructor.port2.on("close", () => console.log("transferList: close"));
+         viaPostMessage.port2.on("close", () => console.log("worker.postMessage: close"));
+         const w = new Worker(
+           \`const { parentPort, workerData, MessageChannel } = require("worker_threads");
+            const { port1 } = new MessageChannel();
+            port1.on("close", () => parentPort.postMessage("parentPort.postMessage: close"));
+            parentPort.postMessage(port1, [port1]);\`,
+           { eval: true, workerData: viaConstructor.port2, transferList: [viaConstructor.port2] },
+         );
+         w.postMessage(viaPostMessage.port2, [viaPostMessage.port2]);
+         w.on("message", m => {
+           if (typeof m === "string") {
+             console.log(m);
+             w.unref();
+           }
+         });`,
+      ],
+      env: bunEnv,
+      stderr: "pipe",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect({ lines: stdout.trim().split("\n").sort(), stderr, exitCode, signalCode: proc.signalCode }).toEqual({
+      lines: ["parentPort.postMessage: close", "transferList: close", "worker.postMessage: close"],
+      stderr: "",
+      exitCode: 0,
+      signalCode: null,
+    });
+  });
+});
+
 // close() outside a dispatch drops whatever is queued; close() from inside a
 // 'message' handler lets the in-flight drain finish. Both are node's behaviour.
 test("close() drops queued messages unless it runs inside a dispatch", async () => {

test/js/web/workers/message-channel.test.ts

@@ -352,6 +352,36 @@ test("a pending close event survives GC after the port becomes unreachable", asy
   expect(fired).toBe(50);
 });
 
+// A transfer closes the sender's port object the same way (node fires 'close' on it), so
+// disentangle() queues the same task, and that object must stay attached to its context
+// until the task runs: the pending activity only pins the wrapper while it has a context.
+test("a transferred-away port's pending close event survives GC after the port becomes unreachable", async () => {
+  const { heapStats } = require("bun:jsc");
+  const count = () => heapStats().objectTypeCounts.MessagePort ?? 0;
+  Bun.gc(true);
+  const base = count();
+  let fired = 0;
+  const carrier = new MessageChannel();
+  for (let i = 0; i < 50; i++) {
+    (() => {
+      const { port1 } = new MessageChannel();
+      port1.addEventListener("close", () => fired++);
+      carrier.port1.postMessage(port1, [port1]);
+    })();
+    if (i % 10 === 0) Bun.gc(true);
+  }
+  Bun.gc(true);
+  for (let i = 0; i < 4; i++) await new Promise(r => setImmediate(r));
+  expect(fired).toBe(50);
+  // Once the event has fired nothing pins the transferred-away objects (or their
+  // unreferenced peers) any more; only the carrier pair is still reachable.
+  Bun.gc(true);
+  Bun.gc(true);
+  expect(count() - base).toBeLessThanOrEqual(2 + 10);
+  carrier.port1.close();
+  carrier.port2.close();
+});
+
 // The peer's notifyPeerClosed() task only holds a weak ref back to this port, so a
 // port whose only listener is 'close' must survive GC until the event is delivered.
 test("a close event from the peer survives GC of the unreachable port", async () => {

test/js/web/workers/worker-postmessage-transfer.test.ts

@@ -136,4 +136,47 @@ describe("self.postMessage transfer list", () => {
       URL.revokeObjectURL(url);
     }
   });
+
+  // Transferring a port closes the sender's object (node fires 'close' on it, asynchronously).
+  // Worker#postMessage and the worker-side self.postMessage each disentangle the port
+  // themselves, so both directions are checked.
+  test("a port transferred with postMessage fires 'close' on the sender's object, in both directions", async () => {
+    const url = URL.createObjectURL(
+      new Blob([
+        `
+          const { port1 } = new MessageChannel();
+          const events = [];
+          port1.addEventListener("close", () => events.push("close"));
+          self.postMessage(port1, [port1]);
+          events.push("postMessage returned");
+          setImmediate(() => setImmediate(() => self.postMessage(events)));
+        `,
+      ]),
+    );
+    const worker = new Worker(url);
+    try {
+      const messages: any[] = [];
+      const workerEvents = new Promise<string[]>((resolve, reject) => {
+        worker.onerror = e => reject(e.error ?? e.message ?? e);
+        worker.onmessage = e => {
+          messages.push(e.data);
+          if (Array.isArray(e.data)) resolve(e.data);
+        };
+      });
+
+      const { port1 } = new MessageChannel();
+      const events: string[] = [];
+      port1.addEventListener("close", () => events.push("close"));
+      worker.postMessage(port1, [port1]);
+      events.push("postMessage returned");
+      await new Promise(r => setImmediate(() => setImmediate(r)));
+      expect(events).toEqual(["postMessage returned", "close"]);
+
+      expect(await workerEvents).toEqual(["postMessage returned", "close"]);
+      expect(messages[0]).toBeInstanceOf(MessagePort);
+    } finally {
+      worker.terminate();
+      URL.revokeObjectURL(url);
+    }
+  });
 });