MessagePort alignment with Nodejs

.vscode/settings.json

@@ -148,6 +148,10 @@
     "ios": "cpp",
     "oxlint.json": "jsonc",
     "bun.lock": "jsonc",
+    "*.inc": "cpp",
+    "__locale": "cpp",
+    "locale": "cpp",
+    "fstream": "cpp",
   },
   "C_Cpp.files.exclude": {
     "**/.vscode": true,

src/bun.js/bindings/webcore/MessagePort.cpp

@@ -44,6 +44,7 @@
 #include <wtf/TZoneMallocInlines.h>
 #include <wtf/Lock.h>
 #include <wtf/Scope.h>
+#include "Event.h"
 
 extern "C" void Bun__eventLoop__incrementRefConcurrently(void* bunVM, int delta);
 
@@ -104,6 +105,7 @@ void MessagePort::notifyMessageAvailable(const MessagePortIdentifier& identifier
 Ref<MessagePort> MessagePort::create(ScriptExecutionContext& scriptExecutionContext, const MessagePortIdentifier& local, const MessagePortIdentifier& remote)
 {
     auto messagePort = adoptRef(*new MessagePort(scriptExecutionContext, local, remote));
+    messagePort->m_strongRef = messagePort.ptr();
     // messagePort->suspendIfNeeded();
     return messagePort;
 }
@@ -229,18 +231,37 @@ void MessagePort::start()
         return;
 
     m_started = true;
-    scriptExecutionContext()->processMessageWithMessagePortsSoon([pendingActivity = Ref { *this }] {});
+    auto* context = scriptExecutionContext();
+    context->processMessageWithMessagePortsSoon([pendingActivity = Ref { *this }] {});
+    context->refEventLoop();
 }
 
 void MessagePort::close()
 {
-    if (m_isDetached)
+    shouldClose = true;
+
+    tryClose();
+}
+
+void MessagePort::tryClose()
+{
+    if (!shouldClose|| m_isDetached || m_messagesInFlight > 0)
         return;
     m_isDetached = true;
-
+    dispatchCloseEvent();
     MessagePortChannelProvider::singleton().messagePortClosed(m_identifier);
 
     removeAllEventListeners();
+
+    // message port is closed, we can safely mark it for GC
+    this->m_strongRef = nullptr;
+
+    scriptExecutionContext()->unrefEventLoop();
+}
+
+void MessagePort::dispatchCloseEvent()
+{
+    dispatchEvent(Event::create(eventNames().closeEvent, Event::CanBubble::No, Event::IsCancelable::No));
 }
 
 void MessagePort::dispatchMessages()
@@ -266,6 +287,7 @@ void MessagePort::dispatchMessages()
         auto* globalObject = defaultGlobalObject(context->globalObject());
         Ref vm = globalObject->vm();
         auto scope = DECLARE_CATCH_SCOPE(vm);
+        protectedThis->m_messagesInFlight += messages.size();
 
         for (auto& message : messages) {
             // close() in Worker onmessage handler should prevent next message from dispatching.
@@ -287,6 +309,8 @@ void MessagePort::dispatchMessages()
             ScriptExecutionContext::postTaskTo(context->identifier(), [protectedThis = Ref { *this }, ports = WTFMove(ports), message = WTFMove(message)](ScriptExecutionContext& context) mutable {
                 auto event = MessageEvent::create(*context.jsGlobalObject(), message.message.releaseNonNull(), {}, {}, {}, WTFMove(ports));
                 protectedThis->dispatchEvent(event.event);
+                protectedThis->m_messagesInFlight -= 1;
+                protectedThis->tryClose();
             });
         }
     };
@@ -427,8 +451,7 @@ Ref<MessagePort> MessagePort::entangle(ScriptExecutionContext& context, Transfer
 bool MessagePort::addEventListener(const AtomString& eventType, Ref<EventListener>&& listener, const AddEventListenerOptions& options)
 {
     if (eventType == eventNames().messageEvent) {
-        if (listener->isAttribute())
-            start();
+        start();
         m_hasMessageEventListener = true;
     }
     return EventTarget::addEventListener(eventType, WTFMove(listener), options);

src/bun.js/bindings/webcore/MessagePort.h

@@ -75,6 +75,7 @@ class MessagePort final : /* public ActiveDOMObject, */ public ContextDestructio
     bool isDetached() const { return m_isDetached; }
 
     void dispatchMessages();
+    void dispatchCloseEvent();
 
     // Returns null if there is no entangled port, or if the entangled port is run by a different thread.
     // This is used solely to enable a GC optimization. Some platforms may not be able to determine ownership
@@ -149,6 +150,11 @@ class MessagePort final : /* public ActiveDOMObject, */ public ContextDestructio
 
     uint32_t m_messageEventCount { 0 };
     static void onDidChangeListenerImpl(EventTarget& self, const AtomString& eventType, OnDidChangeListenerKind kind);
+    // keeps MessagePort alive until we close()
+    RefPtr<MessagePort> m_strongRef = nullptr;
+    unsigned m_messagesInFlight = 0;
+    bool shouldClose = false;
+    void tryClose();
 };
 
 WebCoreOpaqueRoot root(MessagePort*);

src/bun.js/bindings/webcore/SerializedScriptValue.cpp

@@ -5779,6 +5779,7 @@ ExceptionOr<Ref<SerializedScriptValue>> SerializedScriptValue::create(JSGlobalOb
         if (auto port = JSMessagePort::toWrapped(vm, transferable.get())) {
             if (port->isDetached())
                 return Exception { DataCloneError, "MessagePort in transfer list is already detached"_s };
+            port->dispatchCloseEvent();
             messagePorts.append(WTFMove(port));
             continue;
         }