Fire the close event on MessagePort

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<MessagePort> { *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<EventListene
     if (eventType == eventNames().messageEvent) {
         start();
         m_hasMessageEventListener = true;
+    } else if (eventType == eventNames().closeEvent) {
+        startForClose();
+        m_hasCloseEventListener = true;
     }
     return EventTarget::addEventListener(eventType, WTF::move(listener), options);
 }
@@ -321,6 +468,8 @@ bool MessagePort::removeEventListener(const AtomString& eventType, EventListener
     auto result = EventTarget::removeEventListener(eventType, listener, options);
     if (!hasEventListeners(eventNames().messageEvent))
         m_hasMessageEventListener = false;
+    if (!hasEventListeners(eventNames().closeEvent))
+        m_hasCloseEventListener = false;
     return result;
 }
 

src/jsc/bindings/webcore/MessagePort.h

@@ -84,6 +84,11 @@ class MessagePort final : public ContextDestructionObserver, public EventTarget,
     // Called by the pipe on this port's context thread with one dequeued message.
     void dispatchOneMessage(ScriptExecutionContext&, MessageWithMessagePorts&&);
 
+    // Called by the pipe drain on this port's context thread once the inbox is
+    // empty and the entangled peer has closed: fires 'close' (if a listener is
+    // registered) and tears the port down.
+    void dispatchCloseEventFromPeer();
+
     // Only here for JSMessagePortCustom's GC optimization; always null.
     MessagePort* locallyEntangledPort() { return nullptr; }
 
@@ -117,6 +122,24 @@ class MessagePort final : public ContextDestructionObserver, public EventTarget,
 
     void contextDestroyed() final;
 
+    // Attach to the pipe so the peer can wake this port to dispatch 'close',
+    // without enabling message delivery (unlike start(), which a 'message'
+    // listener calls). Idempotent with a later start().
+    void startForClose();
+
+    // Whether this port's context can currently run JS (not mid-teardown).
+    // Gates scheduling close-event work that would otherwise never run.
+    bool canRunScript() const;
+
+    // Fires the 'close' event on this port (once). Safe to call after the port
+    // is detached: it bypasses the detached guard in dispatchEvent().
+    void dispatchCloseEvent();
+    // Task body for the closing port's own asynchronous 'close' event.
+    void dispatchCloseEventSelf();
+    // Schedules dispatchCloseEventSelf() on this port's context. Returns true
+    // if a task was posted (in which case listener teardown is deferred to it).
+    bool scheduleCloseEvent();
+
     bool isEntangled() const { return !m_isDetached; }
 
     // Held for the port's entire lifetime — never nulled — so that the GC
@@ -128,6 +151,8 @@ class MessagePort final : public ContextDestructionObserver, public EventTarget,
     bool m_started { false };
     bool m_isDetached { false };
     bool m_hasMessageEventListener { false };
+    bool m_hasCloseEventListener { false };
+    bool m_closeEventDispatched { false };
     bool m_hasRef { false };
 
     uint32_t m_messageEventCount { 0 };