node:worker_threads: implement postMessageToThread [1bx5ty]

src/bun.js/bindings/ErrorCode.cpp

@@ -2390,6 +2390,14 @@ JSC_DEFINE_HOST_FUNCTION(Bun::jsFunctionMakeErrorWithCode, (JSC::JSGlobalObject
         return JSC::JSValue::encode(createError(globalObject, ErrorCode::ERR_IPC_CHANNEL_CLOSED, "Channel closed."_s));
     case ErrorCode::ERR_SOCKET_BAD_TYPE:
         return JSC::JSValue::encode(createError(globalObject, ErrorCode::ERR_SOCKET_BAD_TYPE, "Bad socket type specified. Valid types are: udp4, udp6"_s));
+    case ErrorCode::ERR_WORKER_MESSAGING_ERRORED:
+        return JSC::JSValue::encode(createError(globalObject, ErrorCode::ERR_WORKER_MESSAGING_ERRORED, "The destination thread threw an error while processing the message"_s));
+    case ErrorCode::ERR_WORKER_MESSAGING_FAILED:
+        return JSC::JSValue::encode(createError(globalObject, ErrorCode::ERR_WORKER_MESSAGING_FAILED, "Cannot find the destination thread or listener"_s));
+    case ErrorCode::ERR_WORKER_MESSAGING_SAME_THREAD:
+        return JSC::JSValue::encode(createError(globalObject, ErrorCode::ERR_WORKER_MESSAGING_SAME_THREAD, "Cannot sent a message to the same thread"_s));
+    case ErrorCode::ERR_WORKER_MESSAGING_TIMEOUT:
+        return JSC::JSValue::encode(createError(globalObject, ErrorCode::ERR_WORKER_MESSAGING_TIMEOUT, "Sending a message to another thread timed out"_s));
     case ErrorCode::ERR_ZLIB_INITIALIZATION_FAILED:
         return JSC::JSValue::encode(createError(globalObject, ErrorCode::ERR_ZLIB_INITIALIZATION_FAILED, "Initialization failed"_s));
     case ErrorCode::ERR_IPC_ONE_PIPE:

src/bun.js/bindings/ErrorCode.ts

@@ -272,6 +272,10 @@ const errors: ErrorCodeMapping = [
   ["ERR_WASI_NOT_STARTED", Error],
   ["ERR_WEBASSEMBLY_RESPONSE", TypeError],
   ["ERR_WORKER_INIT_FAILED", Error],
+  ["ERR_WORKER_MESSAGING_ERRORED", Error],
+  ["ERR_WORKER_MESSAGING_FAILED", Error],
+  ["ERR_WORKER_MESSAGING_SAME_THREAD", Error],
+  ["ERR_WORKER_MESSAGING_TIMEOUT", Error],
   ["ERR_WORKER_NOT_RUNNING", Error],
   ["ERR_WORKER_UNSUPPORTED_OPERATION", TypeError],
   ["ERR_ZLIB_INITIALIZATION_FAILED", Error],

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

@@ -116,8 +116,7 @@ bool EventEmitter::emitForBindings(const Identifier& eventType, const MarkedArgu
     if (!scriptExecutionContext())
         return false;
 
-    emit(eventType, arguments);
-    return true;
+    return emit(eventType, arguments);
 }
 
 bool EventEmitter::emit(const Identifier& eventType, const MarkedArgumentBuffer& arguments)

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

@@ -202,7 +202,11 @@ static inline bool setJSMessagePort_onmessageSetter(JSGlobalObject& lexicalGloba
     vm.writeBarrier(&thisObject, value);
     ensureStillAliveHere(value);
 
-    thisObject.wrapped().jsRef(&lexicalGlobalObject);
+    // Node.js's onmessage setter refs when assigning a function and unrefs otherwise.
+    if (value.isCallable())
+        thisObject.wrapped().jsRef(&lexicalGlobalObject);
+    else
+        thisObject.wrapped().jsUnref(&lexicalGlobalObject);
 
     return true;
 }
@@ -231,7 +235,7 @@ static inline bool setJSMessagePort_onmessageerrorSetter(JSGlobalObject& lexical
     vm.writeBarrier(&thisObject, value);
     ensureStillAliveHere(value);
 
-    thisObject.wrapped().jsRef(&lexicalGlobalObject);
+    // Unlike onmessage, Node.js does not ref the port when only onmessageerror is set.
 
     return true;
 }

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

@@ -145,8 +145,10 @@ template<> JSC::EncodedJSValue JSC_HOST_CALL_ATTRIBUTES JSWorkerDOMConstructor::
 
     WorkerOptions options {};
     JSValue nodeWorkerObject {};
-    if (callFrame->argumentCount() == 3) {
+    JSValue mainThreadPort {};
+    if (callFrame->argumentCount() >= 3) {
         nodeWorkerObject = callFrame->argument(2);
+        mainThreadPort = callFrame->argument(3);
         options.kind = WorkerOptions::Kind::Node;
     }
     JSValue workerData = jsUndefined();
@@ -307,13 +309,22 @@ template<> JSC::EncodedJSValue JSC_HOST_CALL_ATTRIBUTES JSWorkerDOMConstructor::
         }
     }
 
+    // The internal port linking this worker to the main thread for worker_threads.postMessageToThread.
+    // Only present when constructed via node:worker_threads.
+    if (mainThreadPort && mainThreadPort.isObject()) {
+        transferList.append({ vm, mainThreadPort.getObject() });
+    } else {
+        mainThreadPort = jsUndefined();
+    }
+
     Vector<RefPtr<MessagePort>> ports;
-    auto* valueToTransfer = constructEmptyArray(globalObject, nullptr, 2);
+    auto* valueToTransfer = constructEmptyArray(globalObject, nullptr, 3);
     RETURN_IF_EXCEPTION(throwScope, {});
     valueToTransfer->putDirectIndex(globalObject, 0, workerData);
     auto* environmentData = globalObject->nodeWorkerEnvironmentData();
     // If node:worker_threads has not been imported, environment data will not be set up yet.
     valueToTransfer->putDirectIndex(globalObject, 1, environmentData ? environmentData : jsUndefined());
+    valueToTransfer->putDirectIndex(globalObject, 2, mainThreadPort);
 
     ExceptionOr<Ref<SerializedScriptValue>> serialized = SerializedScriptValue::create(*lexicalGlobalObject, valueToTransfer, WTF::move(transferList), ports, SerializationForStorage::No, SerializationContext::WorkerPostMessage);
     if (serialized.hasException()) {

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

@@ -27,7 +27,6 @@
 #include "config.h"
 #include "MessagePort.h"
 
-#include "BunClientData.h"
 #include "EventNames.h"
 #include "MessageEvent.h"
 #include "MessagePortPipe.h"
@@ -36,8 +35,6 @@
 #include "WebCoreOpaqueRoot.h"
 #include <wtf/TZoneMallocInlines.h>
 
-extern "C" void Bun__eventLoop__incrementRefConcurrently(void* bunVM, int delta);
-
 namespace WebCore {
 
 WTF_MAKE_TZONE_ALLOCATED_IMPL(MessagePort);
@@ -60,6 +57,17 @@
 {
     if (!m_isDetached)
         m_pipe->close(m_side);
+    // Third teardown path alongside close() and disentangle(): a ref'd port
+    // (onmessage / .ref()) whose peer has closed is GC-collectible via
+    // hasPendingActivity() → PeerClosed, and we reach here with
+    // m_isRefingEventLoop still set. scriptExecutionContext() is still valid
+    // — ContextDestructionObserver's destructor runs after this body, and if
+    // the context died first contextDestroyed() already routed through
+    // close() → updateEventLoopRef().
+    if (m_isRefingEventLoop) {
+        if (auto* context = scriptExecutionContext())
+            context->unrefEventLoop();
+    }
 }
 
 ExceptionOr<void> MessagePort::postMessage(JSC::JSGlobalObject& state, JSC::JSValue messageValue, StructuredSerializeOptions&& options)
@@ -113,20 +121,10 @@
     m_pipe->close(m_side);
 
     removeAllEventListeners();
-
-    // Release the self-reference taken by jsRef() (set when .onmessage is
-    // assigned or .ref() is called from JS). The JS .close() binding calls
-    // jsUnref() first, so m_hasRef is already false on that path; we only
-    // reach this branch when close() runs without a preceding jsUnref() —
-    // most importantly from contextDestroyed() during Worker teardown.
-    // Without this, the self-ref pins the MessagePort past the JS wrapper
-    // sweep and it leaks forever.
-    if (m_hasRef) {
-        m_hasRef = false;
-        if (auto* context = scriptExecutionContext())
-            context->unrefEventLoop();
-        deref();
-    }
+    // m_isDetached flipped above; ensure any explicit (.ref() / onmessage)
+    // event-loop ref is released too — removeAllEventListeners() only covers
+    // the listener-count path.
+    updateEventLoopRef();
 }
 
 TransferredMessagePort MessagePort::disentangle()
@@ -139,18 +137,6 @@
     removeAllEventListeners();
     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.
-    // The caller (disentanglePorts) holds a RefPtr, so deref() is safe.
-    if (m_hasRef) {
-        m_hasRef = false;
-        if (auto* context = scriptExecutionContext())
-            context->unrefEventLoop();
-        deref();
-    }
-
     // Hand the pipe endpoint to its next owner. Messages that arrive while
     // in transit buffer in the pipe; the receiving context's entangle()
     // re-attaches and flushes them. We keep our own ref to the pipe so the
@@ -160,6 +146,11 @@
     m_isDetached = true;
     m_started = false;
 
+    // Release any explicit event-loop ref before we drop our context; after
+    // observeContext(nullptr) updateEventLoopRef() would see a null context
+    // and just clear the flag without balancing the ref.
+    updateEventLoopRef();
+
     if (auto* context = scriptExecutionContext())
         context->willDestroyDestructionObserver(*this);
     observeContext(nullptr);
@@ -226,12 +217,6 @@
 
 void MessagePort::contextDestroyed()
 {
-    // close() releases the jsRef() self-reference, which may be the last
-    // strong ref if the JS wrapper was already swept. Protect across the
-    // call so we can cleanly detach from the dying ScriptExecutionContext
-    // first — otherwise ~ContextDestructionObserver() would call back into
-    // it while it is mid-destruction.
-    Ref protectedThis { *this };
     close();
     ContextDestructionObserver::contextDestroyed();
 }
@@ -240,19 +225,22 @@
 {
     // Called from the GC thread concurrently with the mutator; must be
     // lockless. m_pipe is a Ref<> held for the port's whole lifetime, so
-    // the dereference is always safe; state() and isOtherSideOpen() are
-    // 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.
+    // the dereference is always safe. The plain bool reads can observe
+    // stale values but cannot crash.
     if (!scriptExecutionContext() || 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);
+    // Single atomic load: queued count, DrainScheduled (a message has been
+    // popped from the inbox but not yet dispatched — queuedCount is already
+    // decremented in that window), and the PeerClosed mirror bit all live in
+    // our side's state word. Reading the peer's state separately would let
+    // the GC observe {queuedCount=0, !DrainScheduled} from before the peer's
+    // send, then Closed from after the peer's close — and collect the wrapper
+    // with a message in flight (ASSERT(m_wrapper) in debug, silently dropped
+    // event → hang in release).
+    return MessagePortPipe::isActivityPending(m_pipe->state(m_side));
 }
 
 ExceptionOr<Vector<TransferredMessagePort>> MessagePort::disentanglePorts(Vector<RefPtr<MessagePort>>&& ports)
@@ -281,30 +269,47 @@
     });
 }
 
+void MessagePort::updateEventLoopRef()
+{
+    bool shouldRef = m_hasRef && (m_messageEventCount > 0 || m_wantsExplicitRef) && !m_isDetached;
+    if (shouldRef == m_isRefingEventLoop)
+        return;
+    auto* context = scriptExecutionContext();
+    if (!context) {
+        m_isRefingEventLoop = false;
+        return;
+    }
+    m_isRefingEventLoop = shouldRef;
+    if (shouldRef)
+        context->refEventLoop();
+    else
+        context->unrefEventLoop();
+}
+
 void MessagePort::onDidChangeListenerImpl(EventTarget& self, const AtomString& eventType, OnDidChangeListenerKind kind)
 {
     if (eventType != eventNames().messageEvent)
         return;
 
     auto& port = static_cast<MessagePort&>(self);
-    auto* context = port.scriptExecutionContext();
     switch (kind) {
     case Add:
-        if (port.m_messageEventCount == 0 && context)
-            context->refEventLoop();
+        // Node's setupPortReferencing installs a 'newListener' hook that
+        // calls this.ref() on the first 'message' listener. Mirror that so
+        // port.unref(); port.on('message', fn) re-refs (matching Node), while
+        // port.on('message', fn); port.unref() stays unref'd.
+        if (port.m_messageEventCount == 0)
+            port.m_hasRef = true;
         port.m_messageEventCount++;
         break;
     case Remove:
         port.m_messageEventCount--;
-        if (port.m_messageEventCount == 0 && context)
-            context->unrefEventLoop();
         break;
     case Clear:
-        if (port.m_messageEventCount > 0 && context)
-            context->unrefEventLoop();
         port.m_messageEventCount = 0;
         break;
     }
+    port.updateEventLoopRef();
 }
 
 bool MessagePort::addEventListener(const AtomString& eventType, Ref<EventListener>&& listener, const AddEventListenerOptions& options)
@@ -329,29 +334,20 @@
     return WebCoreOpaqueRoot { port };
 }
 
-void MessagePort::jsRef(JSGlobalObject* lexicalGlobalObject)
+void MessagePort::jsRef(JSGlobalObject*)
 {
-    // A closed or transferred-away port can never receive messages again, so
-    // taking a self-ref (and an event-loop ref) here would only leak:
-    // close()/disentangle() have already run and nothing will ever release a
-    // ref taken afterwards.
-    if (!isEntangled())
-        return;
-
-    if (!m_hasRef) {
-        m_hasRef = true;
-        ref();
-        Bun__eventLoop__incrementRefConcurrently(WebCore::clientData(lexicalGlobalObject->vm())->bunVM, 1);
-    }
+    // updateEventLoopRef()'s predicate already gates on !m_isDetached, so a
+    // closed or transferred-away port cannot take an event-loop ref here.
+    m_hasRef = true;
+    m_wantsExplicitRef = true;
+    updateEventLoopRef();
 }
 
-void MessagePort::jsUnref(JSGlobalObject* lexicalGlobalObject)
+void MessagePort::jsUnref(JSGlobalObject*)
 {
-    if (m_hasRef) {
-        m_hasRef = false;
-        deref();
-        Bun__eventLoop__incrementRefConcurrently(WebCore::clientData(lexicalGlobalObject->vm())->bunVM, -1);
-    }
+    m_hasRef = false;
+    m_wantsExplicitRef = false;
+    updateEventLoopRef();
 }
 
 } // namespace WebCore

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

@@ -107,7 +107,7 @@ class MessagePort final : public ContextDestructionObserver, public EventTarget,
 
     void jsRef(JSGlobalObject*);
     void jsUnref(JSGlobalObject*);
-    bool jsHasRef() { return m_hasRef; }
+    bool jsHasRef() { return m_isRefingEventLoop; }
 
 private:
     MessagePort(ScriptExecutionContext&, Ref<MessagePortPipe>&&, uint8_t side);
@@ -128,9 +128,24 @@ class MessagePort final : public ContextDestructionObserver, public EventTarget,
     bool m_started { false };
     bool m_isDetached { false };
     bool m_hasMessageEventListener { false };
-    bool m_hasRef { false };
+
+    // Event-loop ref state machine. A port holds a single event-loop ref iff
+    //   m_hasRef && (m_messageEventCount > 0 || m_wantsExplicitRef) && !m_isDetached
+    // updateEventLoopRef() reconciles m_isRefingEventLoop with that predicate. This gives
+    // Node's .unref() its documented semantics: even a transferred port with a 'message'
+    // listener stops holding the process open once .unref() is called.
+    //
+    // m_hasRef: has .unref() NOT been called (default true; .unref() clears, .ref() sets).
+    // m_wantsExplicitRef: has .ref() or onmessage=fn been called (so .ref() on a fresh port
+    //   refs even without a listener).
+    // m_messageEventCount: only tracked for transferred ports (onDidChangeListener wired in
+    //   entangle()); fresh ports don't hold the process open via listeners alone.
+    bool m_hasRef { true };
+    bool m_wantsExplicitRef { false };
+    bool m_isRefingEventLoop { false };
 
     uint32_t m_messageEventCount { 0 };
+    void updateEventLoopRef();
     static void onDidChangeListenerImpl(EventTarget& self, const AtomString& eventType, OnDidChangeListenerKind kind);
 };
 

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

@@ -230,6 +230,7 @@ void MessagePortPipe::close(uint8_t side)
     while (!worklist.isEmpty()) {
         auto [pipe, sd] = worklist.takeLast();
         auto& s = pipe->m_sides[sd];
+        auto& peer = pipe->m_sides[1 - sd];
 
         Deque<MessageWithMessagePorts> dropped;
         {
@@ -240,6 +241,15 @@ void MessagePortPipe::close(uint8_t side)
             s.state.store(Closed, std::memory_order_release);
             dropped = std::exchange(s.inbox, {});
         }
+        {
+            // Mirror Closed into the peer's state word so its
+            // hasPendingActivity() — which must be a single atomic load to
+            // avoid observing a torn {before-send, after-close} snapshot —
+            // can see it. fetch_or under the peer's lock since all state
+            // writes are lock-guarded.
+            Locker locker { peer.lock };
+            peer.state.fetch_or(PeerClosed, std::memory_order_acq_rel);
+        }
 
         // Harvest transferred pipes before `dropped` destructs so their
         // ~TransferredMessagePort sees pipe == nullptr and is a no-op.