From e20f9f9e6f95e14729f6d753744c4c8573f38dfe Mon Sep 17 00:00:00 2001 From: robobun Date: Tue, 28 Apr 2026 11:38:11 +0000 Subject: [PATCH 01/30] node:worker_threads: implement postMessageToThread [1bx5ty] Each node:worker_threads Worker now gets a dedicated MessageChannel to the main thread, whose other end is plumbed through the existing serialized [workerData, environmentData] array (extended to include the port at index 2) in JSWorker.cpp/Worker.cpp. The main thread keeps a threadId -> port registry; postMessageToThread routes through it and uses a SharedArrayBuffer + Atomics.waitAsync to await delivery / timeout. New error codes: ERR_WORKER_MESSAGING_ERRORED ERR_WORKER_MESSAGING_FAILED ERR_WORKER_MESSAGING_SAME_THREAD ERR_WORKER_MESSAGING_TIMEOUT Also fixes uncovered while getting the tests to pass: - Native EventEmitter::emitForBindings (used by process) ignored the return value of fireEventListeners() and always returned true; process.emit() now returns false when there are no listeners. - MessagePort.prototype.unref() was a no-op for transferred ports: adding a 'message' listener called refEventLoop() through a counter unref() never touched. Unify into a single event-loop ref that is active when (m_hasRef && m_messageEventCount > 0); ref()/unref() toggle m_hasRef (default true) and hasRef() reflects whether the port is currently keeping the loop alive. - ~ScriptExecutionContext: drain pending processMessageWithMessagePortsSoon handlers before setting m_inScriptExecutionContextDestructor so a MessagePort whose last ref is held by such a handler (e.g. a worker terminated before its mainThreadPort ever dispatched) can be destroyed without tripping the assertion in willDestroyDestructionObserver. Passes: test/js/node/test/parallel/test-worker-messaging.js test/js/node/test/parallel/test-worker-messaging-errors-handler.js test/js/node/test/parallel/test-worker-messaging-errors-invalid.js test/js/node/test/parallel/test-worker-messaging-errors-timeout.js --- src/bun.js/bindings/ErrorCode.cpp | 8 + src/bun.js/bindings/ErrorCode.ts | 4 + .../bindings/ScriptExecutionContext.cpp | 10 +- src/bun.js/bindings/webcore/EventEmitter.cpp | 3 +- src/bun.js/bindings/webcore/JSWorker.cpp | 15 +- src/bun.js/bindings/webcore/MessagePort.cpp | 56 ++--- src/bun.js/bindings/webcore/MessagePort.h | 9 +- src/bun.js/bindings/webcore/Worker.cpp | 11 +- src/js/builtins.d.ts | 4 + src/js/node/worker_threads.ts | 227 +++++++++++++++++- .../test-worker-messaging-errors-handler.js | 34 +++ .../test-worker-messaging-errors-invalid.js | 48 ++++ .../test-worker-messaging-errors-timeout.js | 38 +++ .../test/parallel/test-worker-messaging.js | 112 +++++++++ .../worker_threads/worker_threads.test.ts | 129 ++++++++++ 15 files changed, 663 insertions(+), 45 deletions(-) create mode 100644 test/js/node/test/parallel/test-worker-messaging-errors-handler.js create mode 100644 test/js/node/test/parallel/test-worker-messaging-errors-invalid.js create mode 100644 test/js/node/test/parallel/test-worker-messaging-errors-timeout.js create mode 100644 test/js/node/test/parallel/test-worker-messaging.js diff --git a/src/bun.js/bindings/ErrorCode.cpp b/src/bun.js/bindings/ErrorCode.cpp index 346094c3d76c..2dcdfe1393c2 100644 --- a/src/bun.js/bindings/ErrorCode.cpp +++ b/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: diff --git a/src/bun.js/bindings/ErrorCode.ts b/src/bun.js/bindings/ErrorCode.ts index 047aa1eab1b9..51e1ad30766f 100644 --- a/src/bun.js/bindings/ErrorCode.ts +++ b/src/bun.js/bindings/ErrorCode.ts @@ -271,6 +271,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_ZLIB_INITIALIZATION_FAILED", Error], ["MODULE_NOT_FOUND", Error], diff --git a/src/bun.js/bindings/ScriptExecutionContext.cpp b/src/bun.js/bindings/ScriptExecutionContext.cpp index f1686a9fe2f6..b77b985b78f8 100644 --- a/src/bun.js/bindings/ScriptExecutionContext.cpp +++ b/src/bun.js/bindings/ScriptExecutionContext.cpp @@ -134,13 +134,21 @@ ScriptExecutionContext::~ScriptExecutionContext() Locker locker { allScriptExecutionContextsMapLock }; ASSERT_WITH_MESSAGE(!allScriptExecutionContextsMap().contains(m_identifier), "A ScriptExecutionContext subclass instance implementing postTask should have already removed itself from the map"); } - m_inScriptExecutionContextDestructor = true; #endif // ASSERT_ENABLED + // Draining these handlers may drop the last reference to a MessagePort whose JS wrapper is + // already gone (e.g. when a worker is terminated while a port still has a pending + // processMessageWithMessagePortsSoon callback). Do this before setting + // m_inScriptExecutionContextDestructor so the observer can unregister itself without + // tripping the assertion below; takeAny() makes the subsequent loop safe regardless. auto postMessageCompletionHandlers = WTF::move(m_processMessageWithMessagePortsSoonHandlers); for (auto& completionHandler : postMessageCompletionHandlers) completionHandler(); +#if ASSERT_ENABLED + m_inScriptExecutionContextDestructor = true; +#endif // ASSERT_ENABLED + while (auto* destructionObserver = m_destructionObservers.takeAny()) destructionObserver->contextDestroyed(); diff --git a/src/bun.js/bindings/webcore/EventEmitter.cpp b/src/bun.js/bindings/webcore/EventEmitter.cpp index c9ab566b4103..48148ff4e3c6 100644 --- a/src/bun.js/bindings/webcore/EventEmitter.cpp +++ b/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) diff --git a/src/bun.js/bindings/webcore/JSWorker.cpp b/src/bun.js/bindings/webcore/JSWorker.cpp index a3e129aaf8ae..a6bf547b499e 100644 --- a/src/bun.js/bindings/webcore/JSWorker.cpp +++ b/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> 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> serialized = SerializedScriptValue::create(*lexicalGlobalObject, valueToTransfer, WTF::move(transferList), ports, SerializationForStorage::No, SerializationContext::WorkerPostMessage); if (serialized.hasException()) { diff --git a/src/bun.js/bindings/webcore/MessagePort.cpp b/src/bun.js/bindings/webcore/MessagePort.cpp index 4290583aebe5..065228b24bb7 100644 --- a/src/bun.js/bindings/webcore/MessagePort.cpp +++ b/src/bun.js/bindings/webcore/MessagePort.cpp @@ -45,8 +45,6 @@ #include #include -extern "C" void Bun__eventLoop__incrementRefConcurrently(void* bunVM, int delta); - namespace WebCore { WTF_MAKE_TZONE_ALLOCATED_IMPL(MessagePort); @@ -383,36 +381,40 @@ void MessagePort::contextDestroyed() // ActiveDOMObject::contextDestroyed(); } +void MessagePort::updateEventLoopRef() +{ + bool shouldRef = m_hasRef && m_messageEventCount > 0 && !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) { auto& port = static_cast(self); switch (kind) { case Add: - if (port.m_messageEventCount == 0) { - auto* context = port.scriptExecutionContext(); - if (context) - context->refEventLoop(); - } port.m_messageEventCount++; break; case Remove: - port.m_messageEventCount--; - if (port.m_messageEventCount == 0) { - auto* context = port.scriptExecutionContext(); - if (context) - context->unrefEventLoop(); - } + if (port.m_messageEventCount > 0) + port.m_messageEventCount--; break; case Clear: - if (port.m_messageEventCount > 0) { - auto* context = port.scriptExecutionContext(); - if (context) - context->unrefEventLoop(); - } port.m_messageEventCount = 0; break; } + port.updateEventLoopRef(); } }; @@ -453,22 +455,16 @@ WebCoreOpaqueRoot root(MessagePort* port) return WebCoreOpaqueRoot { port }; } -void MessagePort::jsRef(JSGlobalObject* lexicalGlobalObject) +void MessagePort::jsRef(JSGlobalObject*) { - if (!m_hasRef) { - m_hasRef = true; - ref(); - Bun__eventLoop__incrementRefConcurrently(WebCore::clientData(lexicalGlobalObject->vm())->bunVM, 1); - } + m_hasRef = 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; + updateEventLoopRef(); } } // namespace WebCore diff --git a/src/bun.js/bindings/webcore/MessagePort.h b/src/bun.js/bindings/webcore/MessagePort.h index c90f8a5b47c6..4d382ebfc73b 100644 --- a/src/bun.js/bindings/webcore/MessagePort.h +++ b/src/bun.js/bindings/webcore/MessagePort.h @@ -113,7 +113,7 @@ class MessagePort final : /* public ActiveDOMObject, */ public ContextDestructio void jsRef(JSGlobalObject*); void jsUnref(JSGlobalObject*); - bool jsHasRef() { return m_hasRef; } + bool jsHasRef() { return m_isRefingEventLoop; } private: explicit MessagePort(ScriptExecutionContext&, const MessagePortIdentifier& local, const MessagePortIdentifier& remote); @@ -139,9 +139,14 @@ class MessagePort final : /* public ActiveDOMObject, */ public ContextDestructio mutable std::atomic m_refCount { 1 }; - bool m_hasRef { false }; + // Whether this port should keep the event loop alive when it has a 'message' listener. + // Toggled by ref()/unref() from JS. + bool m_hasRef { true }; + // Whether this port is currently holding a ref on the event loop. + bool m_isRefingEventLoop { false }; uint32_t m_messageEventCount { 0 }; + void updateEventLoopRef(); static void onDidChangeListenerImpl(EventTarget& self, const AtomString& eventType, OnDidChangeListenerKind kind); }; diff --git a/src/bun.js/bindings/webcore/Worker.cpp b/src/bun.js/bindings/webcore/Worker.cpp index 5695b2e171d4..96d723bd6c40 100644 --- a/src/bun.js/bindings/webcore/Worker.cpp +++ b/src/bun.js/bindings/webcore/Worker.cpp @@ -584,6 +584,7 @@ JSValue createNodeWorkerThreadsBinding(Zig::GlobalObject* globalObject) auto scope = DECLARE_THROW_SCOPE(globalObject->vm()); JSValue workerData = jsNull(); JSValue threadId = jsNumber(0); + JSValue mainThreadPort = jsUndefined(); JSMap* environmentData = nullptr; if (auto* worker = WebWorker__getParentWorker(globalObject->bunVM())) { @@ -592,17 +593,20 @@ JSValue createNodeWorkerThreadsBinding(Zig::GlobalObject* globalObject) RefPtr serialized = WTF::move(options.workerDataAndEnvironmentData); JSValue deserialized = serialized->deserialize(*globalObject, globalObject, WTF::move(ports)); RETURN_IF_EXCEPTION(scope, {}); - // Should always be set to an Array of length 2 in the constructor in JSWorker.cpp + // Should always be set to an Array of length 3 in the constructor in JSWorker.cpp auto* pair = uncheckedDowncast(deserialized); - ASSERT(pair->length() == 2); + ASSERT(pair->length() == 3); ASSERT(pair->canGetIndexQuickly(0u)); ASSERT(pair->canGetIndexQuickly(1u)); + ASSERT(pair->canGetIndexQuickly(2u)); workerData = pair->getIndexQuickly(0); RETURN_IF_EXCEPTION(scope, {}); auto environmentDataValue = pair->getIndexQuickly(1); // it might not be a Map if the parent had not set up environmentData yet environmentData = environmentDataValue ? dynamicDowncast(environmentDataValue) : nullptr; RETURN_IF_EXCEPTION(scope, {}); + mainThreadPort = pair->getIndexQuickly(2); + RETURN_IF_EXCEPTION(scope, {}); // Main thread starts at 1 threadId = jsNumber(worker->clientIdentifier() - 1); @@ -614,12 +618,13 @@ JSValue createNodeWorkerThreadsBinding(Zig::GlobalObject* globalObject) ASSERT(environmentData); globalObject->setNodeWorkerEnvironmentData(environmentData); - JSObject* array = constructEmptyArray(globalObject, nullptr, 4); + JSObject* array = constructEmptyArray(globalObject, nullptr, 5); RETURN_IF_EXCEPTION(scope, {}); array->putDirectIndex(globalObject, 0, workerData); array->putDirectIndex(globalObject, 1, threadId); array->putDirectIndex(globalObject, 2, JSFunction::create(vm, globalObject, 1, "receiveMessageOnPort"_s, jsReceiveMessageOnPort, ImplementationVisibility::Public, NoIntrinsic)); array->putDirectIndex(globalObject, 3, environmentData); + array->putDirectIndex(globalObject, 4, mainThreadPort); return array; } diff --git a/src/js/builtins.d.ts b/src/js/builtins.d.ts index 709b040fc550..9e134d48e2f9 100644 --- a/src/js/builtins.d.ts +++ b/src/js/builtins.d.ts @@ -710,6 +710,10 @@ declare function $ERR_IPC_DISCONNECTED(): Error; declare function $ERR_SERVER_NOT_RUNNING(): Error; declare function $ERR_IPC_CHANNEL_CLOSED(): Error; declare function $ERR_SOCKET_BAD_TYPE(): Error; +declare function $ERR_WORKER_MESSAGING_ERRORED(): Error; +declare function $ERR_WORKER_MESSAGING_FAILED(): Error; +declare function $ERR_WORKER_MESSAGING_SAME_THREAD(): Error; +declare function $ERR_WORKER_MESSAGING_TIMEOUT(): Error; declare function $ERR_ZLIB_INITIALIZATION_FAILED(): Error; declare function $ERR_IPC_ONE_PIPE(): Error; declare function $ERR_SOCKET_ALREADY_BOUND(): Error; diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index c6d0633dde98..9503154d7132 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -6,16 +6,19 @@ type WebWorker = InstanceType; const EventEmitter = require("node:events"); const Readable = require("internal/streams/readable"); const { throwNotImplemented, warnNotImplementedOnce } = require("internal/shared"); +const { validateNumber } = require("internal/validators"); const { MessageChannel, BroadcastChannel, Worker: WebWorker, } = globalThis as typeof globalThis & { - // The Worker constructor secretly takes an extra parameter to provide the node:worker_threads - // instance. This is so that it can emit the `worker` event on the process with the - // node:worker_threads instance instead of the Web Worker instance. - Worker: new (...args: [...ConstructorParameters, nodeWorker: Worker]) => WebWorker; + // The Worker constructor secretly takes two extra parameters: the node:worker_threads instance + // (so the `worker` event on process emits the node instance instead of the Web Worker), and the + // worker's end of the MessagePort connecting it to the main thread for postMessageToThread. + Worker: new ( + ...args: [...ConstructorParameters, nodeWorker: Worker, mainThreadPort: MessagePort] + ) => WebWorker; }; const SHARE_ENV = Symbol("nodejs.worker_threads.SHARE_ENV"); @@ -25,11 +28,13 @@ const { 1: _threadId, 2: _receiveMessageOnPort, 3: environmentData, + 4: _mainThreadPort, } = $cpp("Worker.cpp", "createNodeWorkerThreadsBinding") as [ unknown, number, (port: unknown) => unknown, Map, + MessagePort | undefined, ]; type NodeWorkerOptions = import("node:worker_threads").WorkerOptions; @@ -203,6 +208,212 @@ function fakeParentPort() { } let parentPort: MessagePort | null = isMainThread ? null : fakeParentPort(); +// --- postMessageToThread --- +// +// Every worker gets a direct MessageChannel to the main thread (mainThreadPort). The main thread +// keeps a Map of threadId -> port. postMessageToThread always routes through the main thread, and +// a SharedArrayBuffer carries the ack so the sender can await delivery with Atomics.waitAsync. + +const kRegisterMainThreadPort = 0; +const kUnregisterMainThreadPort = 1; +const kSendMessageToWorker = 2; +const kReceiveMessageFromWorker = 3; + +// SharedArrayBuffer must always be Int32, so it's * 4. +// We need one for the operation status (performing / performed) and one for the result (success / failure). +const WORKER_MESSAGING_SHARED_DATA = 2 * 4; +const WORKER_MESSAGING_STATUS_INDEX = 0; +const WORKER_MESSAGING_RESULT_INDEX = 1; + +// Response codes +const WORKER_MESSAGING_RESULT_DELIVERED = 0; +const WORKER_MESSAGING_RESULT_NO_LISTENERS = 1; +const WORKER_MESSAGING_RESULT_LISTENER_ERROR = 2; + +// This is only populated by the main thread and always empty in other threads. +let threadsPorts: Map | undefined; +// This is only populated in child threads and always undefined in the main thread. +let mainThreadPort: MessagePort | undefined; + +function ensureThreadsPorts() { + return (threadsPorts ??= new Map()); +} + +// This event handler is always executed on the main thread only. +function handleMessageFromThread(message) { + switch (message.type) { + case kRegisterMainThreadPort: { + const { threadId, port } = message; + // Register the port + ensureThreadsPorts().$set(threadId, port); + // Handle messages on this port. When a new thread wants to register a child this takes care + // of doing that. This way any thread can be linked to the main one. + port.addEventListener("message", event => handleMessageFromThread(event.data)); + // Never block the thread on this port + port.unref(); + break; + } + case kUnregisterMainThreadPort: { + const ports = ensureThreadsPorts(); + const port = ports.$get(message.threadId); + if (port) { + port.close(); + ports.$delete(message.threadId); + } + break; + } + case kSendMessageToWorker: { + // Send the message to the target thread + const { source, destination, value, transferList, memory } = message; + sendMessageToWorker(source, destination, value, transferList, memory); + break; + } + } +} + +function handleMessageFromMainThread(message) { + if (message.type === kReceiveMessageFromWorker) { + receiveMessageFromWorker(message.source, message.value, message.memory); + } +} + +function sendMessageToWorker(source, destination, value, transferList, memory) { + // We are on the main thread, we can directly process the message + if (destination === threadId) { + receiveMessageFromWorker(source, value, memory); + return; + } + + // Search the port to the target thread + const port = ensureThreadsPorts().$get(destination); + + if (!port) { + const status = new Int32Array(memory); + Atomics.store(status, WORKER_MESSAGING_RESULT_INDEX, WORKER_MESSAGING_RESULT_NO_LISTENERS); + Atomics.store(status, WORKER_MESSAGING_STATUS_INDEX, 1); + Atomics.notify(status, WORKER_MESSAGING_STATUS_INDEX, 1); + return; + } + + port.postMessage( + { + type: kReceiveMessageFromWorker, + source, + destination, + value, + memory, + }, + transferList, + ); +} + +function receiveMessageFromWorker(source, value, memory) { + let response = WORKER_MESSAGING_RESULT_NO_LISTENERS; + + // We need an exception in a listener to propagate here, but the native process.emit swallows + // listener exceptions and reports them as uncaught. Invoke the listeners directly instead. + try { + const listeners = process.listeners("workerMessage"); + for (let i = 0; i < listeners.length; i++) { + listeners[i].$call(process, value, source); + response = WORKER_MESSAGING_RESULT_DELIVERED; + } + } catch { + response = WORKER_MESSAGING_RESULT_LISTENER_ERROR; + } + + // Populate the result + const status = new Int32Array(memory); + Atomics.store(status, WORKER_MESSAGING_RESULT_INDEX, response); + Atomics.store(status, WORKER_MESSAGING_STATUS_INDEX, 1); + Atomics.notify(status, WORKER_MESSAGING_STATUS_INDEX, 1); +} + +function createMainThreadPort(childThreadId, port) { + const registrationMessage = { + type: kRegisterMainThreadPort, + threadId: childThreadId, + port, + }; + + if (mainThreadPort) { + mainThreadPort.postMessage(registrationMessage, [port]); + } else { + // Either we are the main thread, or we were created without the node:worker_threads plumbing + // (e.g. via the Web Worker constructor). Act as the local root for messaging. + handleMessageFromThread(registrationMessage); + } +} + +function destroyMainThreadPort(childThreadId) { + const unregistrationMessage = { + type: kUnregisterMainThreadPort, + threadId: childThreadId, + }; + + if (mainThreadPort) { + mainThreadPort.postMessage(unregistrationMessage); + } else { + handleMessageFromThread(unregistrationMessage); + } +} + +function setupMainThreadPort(port) { + mainThreadPort = port; + port.addEventListener("message", event => handleMessageFromMainThread(event.data)); + // Never block the process on this port + port.unref(); +} + +async function postMessageToThread(destination, value, transferList, timeout) { + if (typeof transferList === "number" && typeof timeout === "undefined") { + timeout = transferList; + transferList = []; + } + + if (typeof timeout !== "undefined") { + validateNumber(timeout, "timeout", 0); + } + + if (destination === threadId) { + throw $ERR_WORKER_MESSAGING_SAME_THREAD(); + } + + const memory = new SharedArrayBuffer(WORKER_MESSAGING_SHARED_DATA); + const status = new Int32Array(memory); + const promise = Atomics.waitAsync(status, WORKER_MESSAGING_STATUS_INDEX, 0, timeout).value; + + const message = { + type: kSendMessageToWorker, + source: threadId, + destination, + value, + memory, + transferList, + }; + + if (mainThreadPort) { + mainThreadPort.postMessage(message, transferList); + } else { + handleMessageFromThread(message); + } + + // Wait for the response + const response = await promise; + + if (response === "timed-out") { + throw $ERR_WORKER_MESSAGING_TIMEOUT(); + } else if (status[WORKER_MESSAGING_RESULT_INDEX] === WORKER_MESSAGING_RESULT_NO_LISTENERS) { + throw $ERR_WORKER_MESSAGING_FAILED(); + } else if (status[WORKER_MESSAGING_RESULT_INDEX] === WORKER_MESSAGING_RESULT_LISTENER_ERROR) { + throw $ERR_WORKER_MESSAGING_ERRORED(); + } +} + +if (!isMainThread && _mainThreadPort) { + setupMainThreadPort(_mainThreadPort); +} + function getEnvironmentData(key: unknown): unknown { return environmentData.get(key); } @@ -248,14 +459,18 @@ class Worker extends EventEmitter { this.#urlToRevoke = filename; } } + // Create a channel that links the new thread to the main thread for postMessageToThread. + const { port1: mainThreadPortToMain, port2: mainThreadPortToThread } = new MessageChannel(); try { - this.#worker = new WebWorker(filename, options as Bun.WorkerOptions, this); + this.#worker = new WebWorker(filename, options as Bun.WorkerOptions, this, mainThreadPortToThread); } catch (e) { + mainThreadPortToMain.close(); if (this.#urlToRevoke) { URL.revokeObjectURL(this.#urlToRevoke); } throw e; } + createMainThreadPort(this.#worker.threadId, mainThreadPortToMain); this.#worker.addEventListener("close", this.#onClose.bind(this), { once: true }); this.#worker.addEventListener("error", this.#onError.bind(this)); this.#worker.addEventListener("message", this.#onMessage.bind(this)); @@ -350,6 +565,7 @@ class Worker extends EventEmitter { } #onClose(e) { + destroyMainThreadPort(this.#worker.threadId); this.#onExitPromise = e.code; this.emit("exit", e.code); } @@ -420,6 +636,7 @@ export default { }, markAsUntransferable, moveMessagePortToContext, + postMessageToThread, receiveMessageOnPort, SHARE_ENV, threadId, diff --git a/test/js/node/test/parallel/test-worker-messaging-errors-handler.js b/test/js/node/test/parallel/test-worker-messaging-errors-handler.js new file mode 100644 index 000000000000..a4a0ae033d90 --- /dev/null +++ b/test/js/node/test/parallel/test-worker-messaging-errors-handler.js @@ -0,0 +1,34 @@ +'use strict'; + +const common = require('../common'); +const { + parentPort, + postMessageToThread, + Worker, + workerData, +} = require('node:worker_threads'); +const assert = require('node:assert'); + +async function test() { + const worker = new Worker(__filename, { workerData: { children: true } }); + + await assert.rejects(common.mustCall(function() { + return postMessageToThread(worker.threadId); + }), { + name: 'Error', + code: 'ERR_WORKER_MESSAGING_ERRORED', + }); + + worker.postMessage('success'); +} + +if (!workerData?.children) { + test(); +} else { + process.on('workerMessage', () => { + throw new Error('KABOOM'); + }); + + parentPort.postMessage('ready'); + parentPort.once('message', common.mustCall()); +} diff --git a/test/js/node/test/parallel/test-worker-messaging-errors-invalid.js b/test/js/node/test/parallel/test-worker-messaging-errors-invalid.js new file mode 100644 index 000000000000..e55e3485e904 --- /dev/null +++ b/test/js/node/test/parallel/test-worker-messaging-errors-invalid.js @@ -0,0 +1,48 @@ +'use strict'; + +const common = require('../common'); +const { once } = require('node:events'); +const { + parentPort, + postMessageToThread, + threadId, + Worker, + workerData, +} = require('node:worker_threads'); +const assert = require('node:assert'); + +async function test() { + await assert.rejects(common.mustCall(function() { + return postMessageToThread(threadId); + }), { + name: 'Error', + code: 'ERR_WORKER_MESSAGING_SAME_THREAD', + }); + + await assert.rejects(common.mustCall(function() { + return postMessageToThread(Date.now()); + }), { + name: 'Error', + code: 'ERR_WORKER_MESSAGING_FAILED', + }); + + // The delivery to the first worker will fail as there is no listener for `workerMessage` + const worker = new Worker(__filename, { workerData: { children: true } }); + await once(worker, 'message'); + + await assert.rejects(common.mustCall(function() { + return postMessageToThread(worker.threadId); + }), { + name: 'Error', + code: 'ERR_WORKER_MESSAGING_FAILED', + }); + + worker.postMessage('success'); +} + +if (!workerData?.children) { + test(); +} else { + parentPort.postMessage('ready'); + parentPort.once('message', common.mustCall()); +} diff --git a/test/js/node/test/parallel/test-worker-messaging-errors-timeout.js b/test/js/node/test/parallel/test-worker-messaging-errors-timeout.js new file mode 100644 index 000000000000..f069005c4110 --- /dev/null +++ b/test/js/node/test/parallel/test-worker-messaging-errors-timeout.js @@ -0,0 +1,38 @@ +'use strict'; + +const common = require('../common'); +const { + postMessageToThread, + workerData, + Worker, +} = require('node:worker_threads'); +const assert = require('node:assert'); + +const memory = new SharedArrayBuffer(4); + +async function test() { + const worker = new Worker(__filename, { workerData: { memory, children: true } }); + const array = new Int32Array(memory); + + await assert.rejects(common.mustCall(function() { + return postMessageToThread(worker.threadId, 0, common.platformTimeout(500)); + }), { + name: 'Error', + code: 'ERR_WORKER_MESSAGING_TIMEOUT', + }); + + Atomics.store(array, 0, 1); + Atomics.notify(array, 0); +} + +if (!workerData?.children) { + test(); +} else { + process.on('beforeExit', common.mustCall()); + + const array = new Int32Array(workerData.memory); + + // Starve this thread waiting for the status to be unlocked. + // This happens in the main thread AFTER the timeout. + Atomics.wait(array, 0, 0); +} diff --git a/test/js/node/test/parallel/test-worker-messaging.js b/test/js/node/test/parallel/test-worker-messaging.js new file mode 100644 index 000000000000..e29258bea130 --- /dev/null +++ b/test/js/node/test/parallel/test-worker-messaging.js @@ -0,0 +1,112 @@ +'use strict'; + +const common = require('../common'); +const Countdown = require('../common/countdown'); +const { + parentPort, + postMessageToThread, + threadId, + workerData, + Worker, +} = require('node:worker_threads'); +const assert = require('node:assert'); +const { once } = require('node:events'); + +// Spawn threads on three levels: 1 main thread, two children, four grand childrens. 7 threads total, max id = 6 +const MAX_LEVEL = 2; +const MAX_THREAD = 6; + +// This is to allow the test to run in --worker mode +const mainThread = workerData?.mainThread ?? threadId; +const level = workerData?.level ?? 0; + +const channel = new BroadcastChannel('nodejs:test-worker-connection'); +let completed; + +if (level === 0) { + completed = new Countdown(MAX_THREAD + 1, () => { + channel.postMessage('exit'); + channel.close(); + }); +} + +async function createChildren() { + const worker = new Worker(__filename, { workerData: { mainThread, level: level + 1 } }); + await once(worker, 'message'); +} + +async function ping() { + let target; + do { + target = mainThread + Math.floor(Math.random() * MAX_THREAD); + } while (target === threadId); + + const { port1, port2 } = new MessageChannel(); + await postMessageToThread(target, { level, port: port2 }, [port2]); + + port1.on('message', common.mustCall(function(message) { + assert.deepStrictEqual(message, { message: 'pong', source: target, destination: threadId }); + port1.close(); + + if (level === 0) { + completed.dec(); + } else { + channel.postMessage('end'); + } + })); + + port1.postMessage({ message: 'ping', source: threadId, destination: target }); +} + +// Do not use mustCall here as the thread might not receive any connection request +process.on('workerMessage', common.mustCallAtLeast(({ port, level }, source) => { + // Let's verify the source hierarchy + // Given we do depth first, the level is 1 for thread 1 and 4, 2 for other threads + if (source !== mainThread) { + const currentThread = source - mainThread; + assert.strictEqual(level, (currentThread === 1 || currentThread === 4) ? 1 : 2); + } else { + assert.strictEqual(level, 0); + } + + // Verify communication + port.on('message', common.mustCall(function(message) { + assert.deepStrictEqual(message, { message: 'ping', source, destination: threadId }); + port.postMessage({ message: 'pong', source: threadId, destination: source }); + port.close(); + })); +}, 0)); + +async function test() { + if (level < MAX_LEVEL) { + await createChildren(); + await createChildren(); + } + + channel.onmessage = function(message) { + switch (message.data) { + case 'start': + ping(); + break; + case 'end': + if (level === 0) { + completed.dec(); + } + break; + case 'exit': + channel.close(); + break; + } + }; + + if (level > 0) { + const currentThread = threadId - mainThread; + assert.strictEqual(level, (currentThread === 1 || currentThread === 4) ? 1 : 2); + parentPort.postMessage({ type: 'ready', threadId }); + } else { + channel.postMessage('start'); + ping(); + } +} + +test(); diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index 555c205c6fc7..9515112599c4 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -38,6 +38,7 @@ test("all worker_threads module properties are present", () => { expect(wt).toHaveProperty("markAsUntransferable"); expect(wt).toHaveProperty("moveMessagePortToContext"); expect(wt).toHaveProperty("parentPort"); + expect(wt).toHaveProperty("postMessageToThread"); expect(wt).toHaveProperty("receiveMessageOnPort"); expect(wt).toHaveProperty("resourceLimits"); expect(wt).toHaveProperty("SHARE_ENV"); @@ -487,3 +488,131 @@ describe("getHeapSnapshot", () => { worker.postMessage(0); }); }); + +describe("postMessageToThread", () => { + test("is exported", () => { + expect(wt).toHaveProperty("postMessageToThread"); + expect(wt.postMessageToThread).toBeFunction(); + }); + + test("rejects when targeting the same thread", async () => { + await expect(wt.postMessageToThread(threadId)).rejects.toMatchObject({ + name: "Error", + code: "ERR_WORKER_MESSAGING_SAME_THREAD", + }); + }); + + test("rejects when targeting an unknown thread", async () => { + await expect(wt.postMessageToThread(2 ** 30)).rejects.toMatchObject({ + name: "Error", + code: "ERR_WORKER_MESSAGING_FAILED", + }); + }); + + test("delivers to workerMessage listener in worker and back", async () => { + const worker = new Worker( + /* js */ ` + const wt = require("node:worker_threads"); + process.on("workerMessage", (value, source) => { + wt.postMessageToThread(source, { echo: value, from: wt.threadId }); + }); + wt.parentPort.postMessage("ready"); + wt.parentPort.once("message", () => {}); + `, + { eval: true }, + ); + try { + await once(worker, "message"); + const { promise, resolve } = Promise.withResolvers(); + process.once("workerMessage", (value, source) => resolve({ value, source })); + await wt.postMessageToThread(worker.threadId, "hello"); + const { value, source } = await promise; + expect(value).toEqual({ echo: "hello", from: worker.threadId }); + expect(source).toBe(worker.threadId); + } finally { + worker.postMessage("done"); + await worker.terminate(); + } + }); + + test("rejects with ERR_WORKER_MESSAGING_FAILED when worker has no listener", async () => { + const worker = new Worker( + /* js */ ` + const wt = require("node:worker_threads"); + wt.parentPort.postMessage("ready"); + wt.parentPort.once("message", () => {}); + `, + { eval: true }, + ); + try { + await once(worker, "message"); + await expect(wt.postMessageToThread(worker.threadId, "hello")).rejects.toMatchObject({ + name: "Error", + code: "ERR_WORKER_MESSAGING_FAILED", + }); + } finally { + worker.postMessage("done"); + await worker.terminate(); + } + }); + + test("rejects with ERR_WORKER_MESSAGING_ERRORED when handler throws", async () => { + const worker = new Worker( + /* js */ ` + const wt = require("node:worker_threads"); + process.on("workerMessage", () => { throw new Error("boom"); }); + wt.parentPort.postMessage("ready"); + wt.parentPort.once("message", () => {}); + `, + { eval: true }, + ); + try { + await once(worker, "message"); + await expect(wt.postMessageToThread(worker.threadId, "hello")).rejects.toMatchObject({ + name: "Error", + code: "ERR_WORKER_MESSAGING_ERRORED", + }); + } finally { + worker.postMessage("done"); + await worker.terminate(); + } + }); +}); + +test("process.emit returns false when there are no listeners", () => { + // process uses a native EventEmitter; it used to always return true. + expect(process.emit("__bun_test_no_listener_event__")).toBe(false); + let called = false; + process.once("__bun_test_with_listener_event__", () => { + called = true; + }); + expect(process.emit("__bun_test_with_listener_event__")).toBe(true); + expect(called).toBe(true); +}); + +test("worker does not stay alive after unref() on a transferred port with a listener", async () => { + // A transferred MessagePort with a 'message' listener used to hold a separate + // event loop ref that .unref() could not release, keeping the worker alive forever. + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + /* js */ ` + const { Worker, workerData } = require("node:worker_threads"); + const { port1, port2 } = new MessageChannel(); + const w = new Worker( + "const p = require('node:worker_threads').workerData.port; p.addEventListener('message', () => {}); p.unref();", + { ev${/* bundler hates eval */ ""}al: true, workerData: { port: port2 }, transferList: [port2] }, + ); + w.on("exit", code => { console.log("exit", code); port1.close(); }); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout.trim()).toBe("exit 0"); + expect(exitCode).toBe(0); +}); From c93e868763f9a412416d7f9d1c7182dc49ed8653 Mon Sep 17 00:00:00 2001 From: robobun Date: Tue, 28 Apr 2026 11:57:07 +0000 Subject: [PATCH 02/30] test: bump timeouts on slow worker_threads fixture tests The eval-source-leak fixture spawns six workers with 100 MiB of source each and the environmentdata-inherit fixture creates five nested workers; both exceed the 5s default in debug/ASAN builds. --- .../worker_threads/worker_threads.test.ts | 66 +++++++++++-------- 1 file changed, 38 insertions(+), 28 deletions(-) diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index 9515112599c4..080416d25435 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -281,19 +281,24 @@ describe("execArgv option", async () => { // TODO(@190n) get our handling of non-string array elements in line with Node's }); -test("eval does not leak source code", async () => { - const proc = Bun.spawn({ - cmd: [bunExe(), "eval-source-leak-fixture.js"], - env: bunEnv, - cwd: __dirname, - stderr: "pipe", - stdout: "ignore", - }); - await proc.exited; - const errors = await proc.stderr.text(); - if (errors.length > 0) throw new Error(errors); - expect(proc.exitCode).toBe(0); -}); +test( + "eval does not leak source code", + async () => { + const proc = Bun.spawn({ + cmd: [bunExe(), "eval-source-leak-fixture.js"], + env: bunEnv, + cwd: __dirname, + stderr: "pipe", + stdout: "ignore", + }); + await proc.exited; + const errors = await proc.stderr.text(); + if (errors.length > 0) throw new Error(errors); + expect(proc.exitCode).toBe(0); + }, + // Spawns six workers with 100 MiB of source each; slow in debug builds. + 60_000, +); describe("worker event", () => { test("is emitted on the next tick with the right value", () => { @@ -371,21 +376,26 @@ describe("environmentData", () => { expect(getEnvironmentData("does_not_exist")).toBeUndefined(); }); - test("is deeply inherited", async () => { - const proc = Bun.spawn({ - cmd: [bunExe(), "environmentdata-inherit-fixture.js"], - env: bunEnv, - cwd: __dirname, - stderr: "pipe", - stdout: "pipe", - }); - await proc.exited; - const errors = await proc.stderr.text(); - if (errors.length > 0) throw new Error(errors); - expect(proc.exitCode).toBe(0); - const out = await proc.stdout.text(); - expect(out).toBe("foo\n".repeat(5)); - }); + test( + "is deeply inherited", + async () => { + const proc = Bun.spawn({ + cmd: [bunExe(), "environmentdata-inherit-fixture.js"], + env: bunEnv, + cwd: __dirname, + stderr: "pipe", + stdout: "pipe", + }); + await proc.exited; + const errors = await proc.stderr.text(); + if (errors.length > 0) throw new Error(errors); + expect(proc.exitCode).toBe(0); + const out = await proc.stdout.text(); + expect(out).toBe("foo\n".repeat(5)); + }, + // Five nested workers; slow in debug builds. + 30_000, + ); test("can be used if parent thread had not imported worker_threads", async () => { const proc = Bun.spawn({ From 731b3319fdb213df64aeb9f0ff29f4719b430f48 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Tue, 28 Apr 2026 11:58:58 +0000 Subject: [PATCH 03/30] [autofix.ci] apply automated fixes --- .../worker_threads/worker_threads.test.ts | 68 ++++++++----------- 1 file changed, 30 insertions(+), 38 deletions(-) diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index 080416d25435..3e1494c31a7e 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -281,24 +281,20 @@ describe("execArgv option", async () => { // TODO(@190n) get our handling of non-string array elements in line with Node's }); -test( - "eval does not leak source code", - async () => { - const proc = Bun.spawn({ - cmd: [bunExe(), "eval-source-leak-fixture.js"], - env: bunEnv, - cwd: __dirname, - stderr: "pipe", - stdout: "ignore", - }); - await proc.exited; - const errors = await proc.stderr.text(); - if (errors.length > 0) throw new Error(errors); - expect(proc.exitCode).toBe(0); - }, - // Spawns six workers with 100 MiB of source each; slow in debug builds. - 60_000, -); +test("eval does not leak source code", async () => { + const proc = Bun.spawn({ + cmd: [bunExe(), "eval-source-leak-fixture.js"], + env: bunEnv, + cwd: __dirname, + stderr: "pipe", + stdout: "ignore", + }); + await proc.exited; + const errors = await proc.stderr.text(); + if (errors.length > 0) throw new Error(errors); + expect(proc.exitCode).toBe(0); +}, // Spawns six workers with 100 MiB of source each; slow in debug builds. +60_000); describe("worker event", () => { test("is emitted on the next tick with the right value", () => { @@ -376,26 +372,22 @@ describe("environmentData", () => { expect(getEnvironmentData("does_not_exist")).toBeUndefined(); }); - test( - "is deeply inherited", - async () => { - const proc = Bun.spawn({ - cmd: [bunExe(), "environmentdata-inherit-fixture.js"], - env: bunEnv, - cwd: __dirname, - stderr: "pipe", - stdout: "pipe", - }); - await proc.exited; - const errors = await proc.stderr.text(); - if (errors.length > 0) throw new Error(errors); - expect(proc.exitCode).toBe(0); - const out = await proc.stdout.text(); - expect(out).toBe("foo\n".repeat(5)); - }, - // Five nested workers; slow in debug builds. - 30_000, - ); + test("is deeply inherited", async () => { + const proc = Bun.spawn({ + cmd: [bunExe(), "environmentdata-inherit-fixture.js"], + env: bunEnv, + cwd: __dirname, + stderr: "pipe", + stdout: "pipe", + }); + await proc.exited; + const errors = await proc.stderr.text(); + if (errors.length > 0) throw new Error(errors); + expect(proc.exitCode).toBe(0); + const out = await proc.stdout.text(); + expect(out).toBe("foo\n".repeat(5)); + }, // Five nested workers; slow in debug builds. + 30_000); test("can be used if parent thread had not imported worker_threads", async () => { const proc = Bun.spawn({ From ac00c9ede0635903e900000f9d0803c2971e2525 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Tue, 28 Apr 2026 12:00:59 +0000 Subject: [PATCH 04/30] [autofix.ci] apply automated fixes (attempt 2/3) --- test/js/node/worker_threads/worker_threads.test.ts | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index 3e1494c31a7e..1d5d6243ae28 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -293,8 +293,7 @@ test("eval does not leak source code", async () => { const errors = await proc.stderr.text(); if (errors.length > 0) throw new Error(errors); expect(proc.exitCode).toBe(0); -}, // Spawns six workers with 100 MiB of source each; slow in debug builds. -60_000); +}, 60_000); // Spawns six workers with 100 MiB of source each; slow in debug builds. describe("worker event", () => { test("is emitted on the next tick with the right value", () => { @@ -386,8 +385,7 @@ describe("environmentData", () => { expect(proc.exitCode).toBe(0); const out = await proc.stdout.text(); expect(out).toBe("foo\n".repeat(5)); - }, // Five nested workers; slow in debug builds. - 30_000); + }, 30_000); // Five nested workers; slow in debug builds. test("can be used if parent thread had not imported worker_threads", async () => { const proc = Bun.spawn({ From 76848e4e5e1bf65fb8e1a09de2ca164759a92e8b Mon Sep 17 00:00:00 2001 From: robobun Date: Tue, 28 Apr 2026 12:33:17 +0000 Subject: [PATCH 05/30] test: use on/removeListener instead of once for workerMessage Avoids leaving a stale listener on process since the native EventEmitter handles once-removal inside emit(), which receiveMessageFromWorker bypasses. --- test/js/node/worker_threads/worker_threads.test.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index 1d5d6243ae28..bb1b018bd61d 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -521,15 +521,17 @@ describe("postMessageToThread", () => { `, { eval: true }, ); + const onWorkerMessage = (value, source) => resolve({ value, source }); + const { promise, resolve } = Promise.withResolvers(); try { await once(worker, "message"); - const { promise, resolve } = Promise.withResolvers(); - process.once("workerMessage", (value, source) => resolve({ value, source })); + process.on("workerMessage", onWorkerMessage); await wt.postMessageToThread(worker.threadId, "hello"); const { value, source } = await promise; expect(value).toEqual({ echo: "hello", from: worker.threadId }); expect(source).toBe(worker.threadId); } finally { + process.removeListener("workerMessage", onWorkerMessage); worker.postMessage("done"); await worker.terminate(); } From f5c6c081d824fe68060e05948fc8c2c4571f98d5 Mon Sep 17 00:00:00 2001 From: robobun Date: Tue, 28 Apr 2026 15:00:38 +0000 Subject: [PATCH 06/30] MessagePort: restore explicit-ref path for jsRef() Setting port.onmessage / calling port.ref() on a non-transferred MessagePort needs to keep the event loop alive even though m_messageEventCount is not tracked for such ports (onDidChangeListener is only wired for transferred ports). Introduce m_wantsExplicitRef so jsRef() marks the port as wanting a ref independently of the listener count, and make updateEventLoopRef() consider either source. Fixes message-channel.test.ts 'message channel created on other thread' where the worker in create-port-worker.js exited before the main thread could round-trip through port1.onmessage. --- src/bun.js/bindings/webcore/MessagePort.cpp | 4 +++- src/bun.js/bindings/webcore/MessagePort.h | 8 ++++++-- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/src/bun.js/bindings/webcore/MessagePort.cpp b/src/bun.js/bindings/webcore/MessagePort.cpp index 065228b24bb7..8cbdb5c70fe0 100644 --- a/src/bun.js/bindings/webcore/MessagePort.cpp +++ b/src/bun.js/bindings/webcore/MessagePort.cpp @@ -383,7 +383,7 @@ void MessagePort::contextDestroyed() void MessagePort::updateEventLoopRef() { - bool shouldRef = m_hasRef && m_messageEventCount > 0 && !m_isDetached; + bool shouldRef = m_hasRef && (m_messageEventCount > 0 || m_wantsExplicitRef) && !m_isDetached; if (shouldRef == m_isRefingEventLoop) return; auto* context = scriptExecutionContext(); @@ -458,12 +458,14 @@ WebCoreOpaqueRoot root(MessagePort* port) void MessagePort::jsRef(JSGlobalObject*) { m_hasRef = true; + m_wantsExplicitRef = true; updateEventLoopRef(); } void MessagePort::jsUnref(JSGlobalObject*) { m_hasRef = false; + m_wantsExplicitRef = false; updateEventLoopRef(); } diff --git a/src/bun.js/bindings/webcore/MessagePort.h b/src/bun.js/bindings/webcore/MessagePort.h index 4d382ebfc73b..c6ce7762983b 100644 --- a/src/bun.js/bindings/webcore/MessagePort.h +++ b/src/bun.js/bindings/webcore/MessagePort.h @@ -139,9 +139,13 @@ class MessagePort final : /* public ActiveDOMObject, */ public ContextDestructio mutable std::atomic m_refCount { 1 }; - // Whether this port should keep the event loop alive when it has a 'message' listener. - // Toggled by ref()/unref() from JS. + // Whether this port should keep the event loop alive when it is active. + // Toggled by ref()/unref() from JS; unref() wins over everything else. bool m_hasRef { true }; + // Whether jsRef() has been called (from .ref() or the onmessage setter). This is one of the + // ways a port becomes "active" for event-loop-ref purposes; the other is m_messageEventCount, + // which is only tracked for ports whose onDidChangeListener is wired (transferred ports). + bool m_wantsExplicitRef { false }; // Whether this port is currently holding a ref on the event loop. bool m_isRefingEventLoop { false }; From 9f5770e629813a524b2f48aee64ae8225e04d03b Mon Sep 17 00:00:00 2001 From: robobun Date: Wed, 29 Apr 2026 01:51:33 +0000 Subject: [PATCH 07/30] MessagePort: don't ref on onmessageerror; document lazy mainThreadPort setup - Drop jsRef() from the onmessageerror setter so a port with only onmessageerror set does not keep the event loop alive, matching Node. The onmessage setter still refs. - Document that the worker side of the postMessageToThread channel is wired lazily on first require('node:worker_threads'), so sending with no timeout to a worker that never imports it stays pending instead of rejecting with ERR_WORKER_MESSAGING_FAILED like Node does. --- src/bun.js/bindings/webcore/JSMessagePort.cpp | 2 +- src/js/node/worker_threads.ts | 7 +++++++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/src/bun.js/bindings/webcore/JSMessagePort.cpp b/src/bun.js/bindings/webcore/JSMessagePort.cpp index 543477b0caa6..c5106a105bbf 100644 --- a/src/bun.js/bindings/webcore/JSMessagePort.cpp +++ b/src/bun.js/bindings/webcore/JSMessagePort.cpp @@ -231,7 +231,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; } diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index 9503154d7132..26fdba91d099 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -213,6 +213,13 @@ let parentPort: MessagePort | null = isMainThread ? null : fakeParentPort(); // Every worker gets a direct MessageChannel to the main thread (mainThreadPort). The main thread // keeps a Map of threadId -> port. postMessageToThread always routes through the main thread, and // a SharedArrayBuffer carries the ack so the sender can await delivery with Atomics.waitAsync. +// +// Known limitation: the worker side of the channel is only set up the first time the worker +// evaluates node:worker_threads (see setupMainThreadPort below). Node.js wires it during worker +// bootstrap unconditionally, so postMessageToThread(id, value) with no timeout to a worker that +// never imports worker_threads rejects with ERR_WORKER_MESSAGING_FAILED there, whereas here the +// promise stays pending until the worker exits or a timeout is supplied. In practice a +// node:worker_threads Worker almost always imports the module for parentPort/workerData. const kRegisterMainThreadPort = 0; const kUnregisterMainThreadPort = 1; From d284e8362a43347d8833aff39af329f75db02572 Mon Sep 17 00:00:00 2001 From: robobun Date: Wed, 29 Apr 2026 02:22:04 +0000 Subject: [PATCH 08/30] test: bump timeouts on execArgv/environmentData subprocess tests These spawn --smol subprocesses that each create a worker; the first execArgv test does it twice and sits at ~4.5-5s in debug builds, which tips over the 5s default under load. --- test/js/node/worker_threads/worker_threads.test.ts | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index bb1b018bd61d..a01c9e70a0ef 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -267,17 +267,18 @@ describe("execArgv option", async () => { expect(await proc.stdout.text()).toBe(expected); } + // Each run() spawns a --smol subprocess that creates a worker; slow in debug builds. it("inherits the parent's execArgv when falsy or unspecified", async () => { await run("null", '["--smol"]\n'); await run("0", '["--smol"]\n'); - }); + }, 30_000); it("provides empty execArgv when passed an empty array", async () => { // empty array should result in empty execArgv, not inherited from parent thread await run("[]", "[]\n"); - }); + }, 15_000); it("can specify an array of strings", async () => { await run('["--no-warnings"]', '["--no-warnings"]\n'); - }); + }, 15_000); // TODO(@190n) get our handling of non-string array elements in line with Node's }); @@ -399,7 +400,7 @@ describe("environmentData", () => { const errors = await proc.stderr.text(); if (errors.length > 0) throw new Error(errors); expect(proc.exitCode).toBe(0); - }); + }, 30_000); // Two nested workers; slow in debug builds. }); describe("error event", () => { From 2869ce3b3b2e4bab25923f0ca651b51aab7c5e74 Mon Sep 17 00:00:00 2001 From: robobun Date: Wed, 29 Apr 2026 02:29:21 +0000 Subject: [PATCH 09/30] doc: clarify postMessageToThread pending-promise wording Worker exit closes the port but does not notify already-pending Atomics.waitAsync waiters, so the promise stays pending indefinitely without a timeout, not 'until the worker exits'. --- src/js/node/worker_threads.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index 26fdba91d099..0c818382646d 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -218,8 +218,9 @@ let parentPort: MessagePort | null = isMainThread ? null : fakeParentPort(); // evaluates node:worker_threads (see setupMainThreadPort below). Node.js wires it during worker // bootstrap unconditionally, so postMessageToThread(id, value) with no timeout to a worker that // never imports worker_threads rejects with ERR_WORKER_MESSAGING_FAILED there, whereas here the -// promise stays pending until the worker exits or a timeout is supplied. In practice a -// node:worker_threads Worker almost always imports the module for parentPort/workerData. +// promise stays pending indefinitely unless a timeout is supplied (worker exit closes the port +// but does not notify already-pending waiters). In practice a node:worker_threads Worker almost +// always imports the module for parentPort/workerData. const kRegisterMainThreadPort = 0; const kUnregisterMainThreadPort = 1; From 4e55aef2d520bcfe907e1b44cca489c95ec39b02 Mon Sep 17 00:00:00 2001 From: robobun Date: Wed, 29 Apr 2026 03:26:46 +0000 Subject: [PATCH 10/30] MessagePort: release event-loop ref in close() The JS close() binding already calls jsUnref() before impl.close(), so this was not observable from JS, but contextDestroyed() and any other native close() callers bypass that. Add updateEventLoopRef() at the end of close() so the !m_isDetached predicate is honoured regardless of the entry point. --- src/bun.js/bindings/webcore/MessagePort.cpp | 1 + 1 file changed, 1 insertion(+) diff --git a/src/bun.js/bindings/webcore/MessagePort.cpp b/src/bun.js/bindings/webcore/MessagePort.cpp index 8cbdb5c70fe0..b71ce4b3ff01 100644 --- a/src/bun.js/bindings/webcore/MessagePort.cpp +++ b/src/bun.js/bindings/webcore/MessagePort.cpp @@ -239,6 +239,7 @@ void MessagePort::close() MessagePortChannelProvider::singleton().messagePortClosed(m_identifier); removeAllEventListeners(); + updateEventLoopRef(); } void MessagePort::dispatchMessages() From 28964143fc5e5ece7368e18adfb731e4173f5c79 Mon Sep 17 00:00:00 2001 From: robobun Date: Wed, 29 Apr 2026 03:59:00 +0000 Subject: [PATCH 11/30] doc: note grandchild-orphan registry behaviour (matches Node) Terminating an intermediate worker leaves its grandchildren in the root's threadsPorts map since destroyMainThreadPort runs from the parent's #onClose. Node's kOnExit is identically parent-driven, so this is upstream parity; document it alongside the lazy-setup note. --- src/js/node/worker_threads.ts | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index 0c818382646d..4f67cc60a111 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -221,6 +221,10 @@ let parentPort: MessagePort | null = isMainThread ? null : fakeParentPort(); // promise stays pending indefinitely unless a timeout is supplied (worker exit closes the port // but does not notify already-pending waiters). In practice a node:worker_threads Worker almost // always imports the module for parentPort/workerData. +// +// Unregistration is driven from the parent's #onClose (matching Node's kOnExit), so terminating +// an intermediate worker orphans its live grandchildren in the root's threadsPorts map; sending +// to such a threadId without a timeout likewise stays pending. This matches Node.js behaviour. const kRegisterMainThreadPort = 0; const kUnregisterMainThreadPort = 1; From a56a2fe4d489f1d67c7539430c2b1f08715d6010 Mon Sep 17 00:00:00 2001 From: robobun Date: Wed, 29 Apr 2026 04:16:01 +0000 Subject: [PATCH 12/30] test: set 30s default timeout for worker_threads.test.ts Most tests in this file spawn a subprocess or a worker, each of which takes 1-3s in debug/ASAN builds; under heavy load several were tipping past the 5s default. A file-level default is simpler than continuing to bump individual tests. --- test/js/node/worker_threads/worker_threads.test.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index a01c9e70a0ef..1b7a922b6911 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -1,3 +1,4 @@ +import { describe, expect, it, setDefaultTimeout, test } from "bun:test"; import { bunEnv, bunExe } from "harness"; import { once } from "node:events"; import fs from "node:fs"; @@ -21,6 +22,10 @@ import wt, { workerData, } from "worker_threads"; +// Many tests here spawn subprocesses or workers that each take 1-3s in debug/ASAN builds, +// and the default 5s per-test timeout is too tight under load. +setDefaultTimeout(30_000); + test("support eval in worker", async () => { const worker = new Worker(`postMessage(1 + 1)`, { eval: true, From a916a10a6d57a582f7f8bb383d7b550a4f8253d3 Mon Sep 17 00:00:00 2001 From: robobun Date: Wed, 29 Apr 2026 04:41:12 +0000 Subject: [PATCH 13/30] MessagePort: unref when onmessage is set to a non-function Node's onmessage setter branches on typeof value === 'function', calling ref()/start() for functions and unref()/stop() otherwise. Previously Bun called jsRef() unconditionally, so port.onmessage = fn; port.onmessage = null left the port refing the event loop. --- src/bun.js/bindings/webcore/JSMessagePort.cpp | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/bun.js/bindings/webcore/JSMessagePort.cpp b/src/bun.js/bindings/webcore/JSMessagePort.cpp index c5106a105bbf..6664922b2fc2 100644 --- a/src/bun.js/bindings/webcore/JSMessagePort.cpp +++ b/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; } From 394f3c3aea73e5ae9df134a6f05cc7baeb253a6a Mon Sep 17 00:00:00 2001 From: robobun Date: Wed, 29 Apr 2026 05:08:31 +0000 Subject: [PATCH 14/30] doc: update WorkerOptions.h comment for 3-element serialized array The serialized bootstrap payload now carries [workerData, environmentData, mainThreadPort]; the Worker.cpp comment was updated but the header field comment was missed. --- src/bun.js/bindings/webcore/WorkerOptions.h | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/bun.js/bindings/webcore/WorkerOptions.h b/src/bun.js/bindings/webcore/WorkerOptions.h index 3feed8512c4e..29d7e7b4d64e 100644 --- a/src/bun.js/bindings/webcore/WorkerOptions.h +++ b/src/bun.js/bindings/webcore/WorkerOptions.h @@ -24,10 +24,11 @@ struct WorkerOptions { // Blob URL. bool evalMode { false }; Kind kind { Kind::Web }; - // Serialized array containing [workerData, environmentData] - // (environmentData is always a Map) + // Serialized array containing [workerData, environmentData, mainThreadPort] + // (environmentData is always a Map; mainThreadPort is the worker's end of the + // postMessageToThread channel, or undefined for Web workers) RefPtr workerDataAndEnvironmentData; - // Objects transferred for either data or environmentData in the transferList + // Objects transferred for data, environmentData, or mainThreadPort in the transferList Vector dataMessagePorts; Vector preloadModules; std::optional> env; // TODO(@190n) allow shared From 54ed0acd177e1d4d46a98562e5d8ca40a29c185a Mon Sep 17 00:00:00 2001 From: robobun Date: Wed, 29 Apr 2026 05:47:17 +0000 Subject: [PATCH 15/30] build(linux): only strip .eh_frame* when LTO is on MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Non-LTO release builds link with --eh-frame-hdr, which emits a PT_GNU_EH_FRAME program header. GNU strip cannot remove program headers, so stripping the .eh_frame* sections leaves an orphan phdr pointing at an unmapped vaddr. On Worker teardown, pthread_exit → _Unwind_ForcedUnwind → _Unwind_Find_FDE dereferences that vaddr and segfaults. CI release builds use LTO (linked with --no-eh-frame-hdr) so are unaffected; this only showed up in local/non-LTO release builds. Gate the section removal on c.lto so it matches the link-time phdr decision. --- scripts/build/flags.ts | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/scripts/build/flags.ts b/scripts/build/flags.ts index e62c9a0efcef..22be8a36fdbf 100644 --- a/scripts/build/flags.ts +++ b/scripts/build/flags.ts @@ -1020,13 +1020,16 @@ export const stripFlags: Flag[] = [ { // musl: no eh_frame handling differences, but CMake gates on NOT musl so we do too. // Strip only runs on plain release (shouldStrip gates debug/asan/valgrind/assertions) - // which in CI always has LTO on — in practice paired with --no-eh-frame-hdr. + // which in CI always has LTO on — paired with --no-eh-frame-hdr so no + // PT_GNU_EH_FRAME phdr is emitted. // - // GNU strip does not rewrite the program header table — so any - // PT_GNU_EH_FRAME phdr entry also survives as an orphan. See the - // --no-eh-frame-hdr rationale in linkFlags above. + // GNU strip does not rewrite the program header table. In a non-LTO + // release build (local dev / gate harness) we link with --eh-frame-hdr, + // so stripping these sections would leave an orphan PT_GNU_EH_FRAME + // phdr pointing at an unmapped vaddr — pthread_exit → _Unwind_Find_FDE + // then segfaults. Gate on c.lto to keep the sections when the phdr exists. flag: ["-R", ".eh_frame", "-R", ".eh_frame_hdr", "-R", ".gcc_except_table"], - when: c => c.linux && c.abi === "gnu", + when: c => c.linux && c.abi === "gnu" && c.lto, desc: "Remove unwind sections (GNU strip required — llvm-strip leaves [LOAD #2 [R]])", }, ]; From abc2aca4f9bf20865c640a77ea40ab834b69408b Mon Sep 17 00:00:00 2001 From: robobun Date: Wed, 29 Apr 2026 06:01:43 +0000 Subject: [PATCH 16/30] worker_threads: cache threadId so destroyMainThreadPort gets the real id The native Worker.threadId getter returns -1 once ClosingFlag is set, which happens before the 'close' event is dispatched. #onClose was therefore calling destroyMainThreadPort(-1) for every worker, leaking the port registration and making postMessageToThread(exitedId, v) pend until timeout instead of rejecting with ERR_WORKER_MESSAGING_FAILED. Cache threadId at construction (as Node does), use it in #onClose, and set it to -1 before emitting 'exit' so the public getter still matches Node's post-exit behaviour. --- src/js/node/worker_threads.ts | 14 +++++++++++--- .../js/node/worker_threads/worker_threads.test.ts | 15 +++++++++++++++ 2 files changed, 26 insertions(+), 3 deletions(-) diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index 4f67cc60a111..8b2b6ee8615a 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -449,6 +449,12 @@ function moveMessagePortToContext() { class Worker extends EventEmitter { #worker: WebWorker; #performance; + // Cached at construction. The native threadId getter returns -1 once the + // worker is closing; we need the real id in #onClose to unregister the + // postMessageToThread port, and for get threadId() to keep returning the + // id after terminate() (Node's getter also does this until [kDispose] nulls + // the handle, which happens after its [kOnExit] equivalent). + #threadId: number; // this is used by terminate(); // either is the exit code if exited, a promise resolving to the exit code, or undefined if we haven't sent .terminate() yet @@ -482,7 +488,8 @@ class Worker extends EventEmitter { } throw e; } - createMainThreadPort(this.#worker.threadId, mainThreadPortToMain); + this.#threadId = this.#worker.threadId; + createMainThreadPort(this.#threadId, mainThreadPortToMain); this.#worker.addEventListener("close", this.#onClose.bind(this), { once: true }); this.#worker.addEventListener("error", this.#onError.bind(this)); this.#worker.addEventListener("message", this.#onMessage.bind(this)); @@ -500,7 +507,7 @@ class Worker extends EventEmitter { } get threadId() { - return this.#worker.threadId; + return this.#threadId; } ref() { @@ -577,7 +584,8 @@ class Worker extends EventEmitter { } #onClose(e) { - destroyMainThreadPort(this.#worker.threadId); + destroyMainThreadPort(this.#threadId); + this.#threadId = -1; this.#onExitPromise = e.code; this.emit("exit", e.code); } diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index 1b7a922b6911..02ce7661be39 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -585,6 +585,21 @@ describe("postMessageToThread", () => { await worker.terminate(); } }); + + test("rejects with ERR_WORKER_MESSAGING_FAILED for an exited worker's threadId", async () => { + // The worker's port registration must be torn down on exit; previously + // #onClose read the native threadId (which is -1 once the worker is + // closing) and the stale port stayed in the map, so this would hang + // until the timeout instead of failing immediately. + const worker = new Worker("require('node:worker_threads')", { eval: true }); + const id = worker.threadId; + await once(worker, "exit"); + expect(worker.threadId).toBe(-1); + await expect(wt.postMessageToThread(id, "hello")).rejects.toMatchObject({ + name: "Error", + code: "ERR_WORKER_MESSAGING_FAILED", + }); + }); }); test("process.emit returns false when there are no listeners", () => { From 2a7ebc403930a29753b20e31eb2d34a8f1b47bb8 Mon Sep 17 00:00:00 2001 From: robobun Date: Wed, 29 Apr 2026 07:34:24 +0000 Subject: [PATCH 17/30] Worker: set OnlineFlag before posting 'open' to the parent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit dispatchOnline was posting the 'open' event to the parent thread and only then setting OnlineFlag. The parent could receive the online notification, call worker.getHeapSnapshot() (or anything gated on isOnline()), and observe isOnline() == false → ERR_WORKER_NOT_RUNNING. Seen as a rare flake on aarch64 CI across multiple unrelated PRs. Set the flag first so the parent never observes 'online but not yet isOnline()'. postTaskToWorkerGlobalScope already handles the flag flipping mid-call; fireEarlyMessages still drains anything that raced into the queue. --- src/bun.js/bindings/webcore/Worker.cpp | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/src/bun.js/bindings/webcore/Worker.cpp b/src/bun.js/bindings/webcore/Worker.cpp index 96d723bd6c40..018c550ebc78 100644 --- a/src/bun.js/bindings/webcore/Worker.cpp +++ b/src/bun.js/bindings/webcore/Worker.cpp @@ -363,6 +363,16 @@ void Worker::drainEvents() void Worker::dispatchOnline(Zig::GlobalObject* workerGlobalObject) { + // Mark online BEFORE posting the 'open' event to the parent. Otherwise the parent can + // receive 'online', call e.g. worker.getHeapSnapshot(), and observe isOnline() == false + // → ERR_WORKER_NOT_RUNNING. postTaskToWorkerGlobalScope switches from queueing to direct + // posting once this flag is set; fireEarlyMessages (called right after us) drains any + // tasks that raced into the queue. + { + Locker lock(this->m_pendingTasksMutex); + m_onlineClosingFlags.fetch_or(OnlineFlag); + } + auto* ctx = scriptExecutionContext(); if (ctx) { ScriptExecutionContext::postTaskTo(ctx->identifier(), [protectedThis = Ref { *this }](ScriptExecutionContext& context) -> void { @@ -373,9 +383,6 @@ void Worker::dispatchOnline(Zig::GlobalObject* workerGlobalObject) }); } - Locker lock(this->m_pendingTasksMutex); - - m_onlineClosingFlags.fetch_or(OnlineFlag); auto* thisContext = workerGlobalObject->scriptExecutionContext(); if (!thisContext) { return; From e7b3f3721bc04c4863ec1d57d683cf8aecbf7b19 Mon Sep 17 00:00:00 2001 From: robobun Date: Wed, 29 Apr 2026 08:33:47 +0000 Subject: [PATCH 18/30] MessagePort: release event-loop ref in disentangle() When a port that holds an event-loop ref (via onmessage / .ref() / a message listener on a transferred port) is itself transferred, disentangle() calls observeContext(nullptr) without first balancing the ref. updateEventLoopRef() can't help after that point because the context is gone. The source thread is left with a permanent +1 ref and hangs instead of exiting. Release the ref explicitly before detaching from the context. This completes the same refactor that 4e55aef2d5 applied to close(). --- src/bun.js/bindings/webcore/MessagePort.cpp | 9 +++++++ .../worker_threads/worker_threads.test.ts | 27 +++++++++++++++++++ 2 files changed, 36 insertions(+) diff --git a/src/bun.js/bindings/webcore/MessagePort.cpp b/src/bun.js/bindings/webcore/MessagePort.cpp index b71ce4b3ff01..07e3b400ee1b 100644 --- a/src/bun.js/bindings/webcore/MessagePort.cpp +++ b/src/bun.js/bindings/webcore/MessagePort.cpp @@ -193,6 +193,15 @@ TransferredMessagePort MessagePort::disentangle() auto& context = *scriptExecutionContext(); MessagePortChannelProvider::fromContext(context).messagePortDisentangled(m_identifier); + // Release any event-loop ref we hold on the source context before detaching. After + // observeContext(nullptr), updateEventLoopRef() has no context to unref, so a port that + // was ref'd (via onmessage / .ref() / a message listener on a transferred port) and then + // itself transferred would otherwise leave its source thread with a permanent +1 ref. + if (m_isRefingEventLoop) { + m_isRefingEventLoop = false; + context.unrefEventLoop(); + } + // We can't receive any messages or generate any events after this, so remove ourselves from the list of active ports. context.destroyedMessagePort(*this); // context.willDestroyActiveDOMObject(*this); diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index 02ce7661be39..e6856a0aad0c 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -613,6 +613,33 @@ test("process.emit returns false when there are no listeners", () => { expect(called).toBe(true); }); +test("transferring a ref'd MessagePort releases its event-loop ref on the source thread", async () => { + // port1.onmessage = fn takes an event-loop ref; transferring port1 detaches it from the + // source context. disentangle() used to leave that ref behind, so the source process hung. + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + /* js */ ` + const { Worker } = require("node:worker_threads"); + const { port1 } = new MessageChannel(); + port1.onmessage = () => {}; + const w = new Worker("setTimeout(() => {}, 100)", { ev${/* bundler hates eval */ ""}al: true }); + w.postMessage({ p: port1 }, [port1]); + w.unref(); + setTimeout(() => console.log("DONE"), 50); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout.trim()).toBe("DONE"); + expect(exitCode).toBe(0); +}); + test("worker does not stay alive after unref() on a transferred port with a listener", async () => { // A transferred MessagePort with a 'message' listener used to hold a separate // event loop ref that .unref() could not release, keeping the worker alive forever. From 7da988b557813bbffeed42ad0d5e7e6998d664b0 Mon Sep 17 00:00:00 2001 From: robobun Date: Fri, 1 May 2026 20:13:32 +0000 Subject: [PATCH 19/30] MessagePort: keep JS wrapper alive while a drain is in progress MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MessagePortPipe::drainAndDispatch pops a message from the inbox (decrementing queuedCount) under the lock, releases the lock, then dispatches. If the peer side has already closed, hasPendingActivity() sees queuedCount==0 && !isOtherSideOpen in that window and returns false, letting the GC collect the port's JS wrapper before the event is delivered — ASSERT(m_wrapper) in debug, silently dropped event (and hung test-worker-messaging.js) in release. DrainScheduled stays set for the whole drain loop; include it in hasPendingActivity() so the wrapper is rooted across the pop→dispatch window. Brings test-worker-messaging.js from ~3% hang rate to 0/300. --- src/bun.js/bindings/webcore/MessagePort.cpp | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/src/bun.js/bindings/webcore/MessagePort.cpp b/src/bun.js/bindings/webcore/MessagePort.cpp index c05e3930f81c..b2a552084418 100644 --- a/src/bun.js/bindings/webcore/MessagePort.cpp +++ b/src/bun.js/bindings/webcore/MessagePort.cpp @@ -224,9 +224,14 @@ bool MessagePort::hasPendingActivity() const 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); + // Keep alive if there are messages already queued for us, a drain is in + // progress (a message has been popped from the inbox but not yet + // dispatched — queuedCount is already decremented in that window), or the + // peer is still open and could send more. Without the DrainScheduled + // check, a port whose peer has closed can have its JS wrapper collected + // between drainAndDispatch's pop and dispatch → ASSERT(m_wrapper) in + // JSEventListener (debug) or a silently dropped event (release). + return MessagePortPipe::queuedCount(s) > 0 || (s & MessagePortPipe::DrainScheduled) || m_pipe->isOtherSideOpen(m_side); } ExceptionOr> MessagePort::disentanglePorts(Vector>&& ports) From b185dbe8731e2d1ca434b13be5bf7ab3578e5e4c Mon Sep 17 00:00:00 2001 From: robobun Date: Fri, 1 May 2026 20:33:11 +0000 Subject: [PATCH 20/30] MessagePortPipe: mirror peer's Closed bit so hasPendingActivity is a single atomic load MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit hasPendingActivity() previously read state(m_side) then state(1-m_side) for isOtherSideOpen — two atomic loads that could straddle the peer's send+close: the GC could observe {queuedCount=0, !DrainScheduled} from before the send, then Closed from after the close, and collect the JS wrapper with a message in flight. Add a PeerClosed bit to each side's state word; close() sets it on the other side under that side's lock (same lock send() uses to append to the inbox). hasPendingActivity() now reads only our own state word — queuedCount, DrainScheduled and PeerClosed together give a consistent snapshot. Removed the now-unused isOtherSideOpen(). --- src/bun.js/bindings/webcore/MessagePort.cpp | 24 +++++++++---------- .../bindings/webcore/MessagePortPipe.cpp | 10 ++++++++ src/bun.js/bindings/webcore/MessagePortPipe.h | 11 ++++++++- 3 files changed, 31 insertions(+), 14 deletions(-) diff --git a/src/bun.js/bindings/webcore/MessagePort.cpp b/src/bun.js/bindings/webcore/MessagePort.cpp index b2a552084418..76852afdd988 100644 --- a/src/bun.js/bindings/webcore/MessagePort.cpp +++ b/src/bun.js/bindings/webcore/MessagePort.cpp @@ -214,24 +214,22 @@ bool MessagePort::hasPendingActivity() const { // 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, a drain is in - // progress (a message has been popped from the inbox but not yet - // dispatched — queuedCount is already decremented in that window), or the - // peer is still open and could send more. Without the DrainScheduled - // check, a port whose peer has closed can have its JS wrapper collected - // between drainAndDispatch's pop and dispatch → ASSERT(m_wrapper) in - // JSEventListener (debug) or a silently dropped event (release). - return MessagePortPipe::queuedCount(s) > 0 || (s & MessagePortPipe::DrainScheduled) || 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> MessagePort::disentanglePorts(Vector>&& ports) diff --git a/src/bun.js/bindings/webcore/MessagePortPipe.cpp b/src/bun.js/bindings/webcore/MessagePortPipe.cpp index a782916564c0..0123100aa249 100644 --- a/src/bun.js/bindings/webcore/MessagePortPipe.cpp +++ b/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 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. diff --git a/src/bun.js/bindings/webcore/MessagePortPipe.h b/src/bun.js/bindings/webcore/MessagePortPipe.h index ff421a0952ba..ded98fae59c0 100644 --- a/src/bun.js/bindings/webcore/MessagePortPipe.h +++ b/src/bun.js/bindings/webcore/MessagePortPipe.h @@ -44,11 +44,21 @@ class MessagePortPipe final : public ThreadSafeRefCounted { Closed = 1ull << 0, // close() was called on this side; drops further deliveries. DrainScheduled = 1ull << 1, // a drain task for this side is in flight. Attached = 1ull << 2, // ctxId/port are valid; ok to schedule drains. + // Mirror of the peer's Closed bit. Lets hasPendingActivity() read a + // consistent snapshot from a single atomic load — otherwise the GC + // could observe {queuedCount=0, !DrainScheduled} from our side before + // the peer's send, then observe Closed from the peer's side after its + // close, and collect the wrapper with a message in flight. + PeerClosed = 1ull << 3, QueuedShift = 8, QueuedOne = 1ull << QueuedShift, }; static constexpr uint64_t queuedCount(uint64_t s) { return s >> QueuedShift; } + // True while the port attached to this side should be kept alive by the + // GC: there is a message queued for it, a drain task is in the + // pop→dispatch window, or the peer could still send more. + static constexpr bool isActivityPending(uint64_t s) { return queuedCount(s) > 0 || (s & DrainScheduled) || !(s & PeerClosed); } // Sender-thread operations. // `fromSide` is the sender's side; the message lands in the *other* side's inbox. @@ -66,7 +76,6 @@ class MessagePortPipe final : public ThreadSafeRefCounted { // Lockless snapshot for the GC visitor / hasPendingActivity. uint64_t state(uint8_t side) const { return m_sides[side].state.load(std::memory_order_acquire); } - bool isOtherSideOpen(uint8_t side) const { return !(state(1 - side) & Closed); } // Equality is by identity; used to reject "port posted through itself". bool operator==(const MessagePortPipe& other) const { return this == &other; } From 27d4a88f0ca25340760be16a760a3a432387f1c1 Mon Sep 17 00:00:00 2001 From: robobun Date: Mon, 4 May 2026 02:34:08 +0000 Subject: [PATCH 21/30] MessagePort: release event-loop ref in destructor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Third teardown path alongside close() and disentangle(): a port with onmessage/.ref() whose peer has closed becomes GC-collectible via hasPendingActivity() → PeerClosed, and ~MessagePort() runs with m_isRefingEventLoop still set. Nothing on that path called context->unrefEventLoop(), so the process hung with a permanent +1. Release it in the destructor body; scriptExecutionContext() is still valid there (ContextDestructionObserver's base destructor runs after). --- src/bun.js/bindings/webcore/MessagePort.cpp | 11 ++++++++ .../worker_threads/worker_threads.test.ts | 28 +++++++++++++++++++ 2 files changed, 39 insertions(+) diff --git a/src/bun.js/bindings/webcore/MessagePort.cpp b/src/bun.js/bindings/webcore/MessagePort.cpp index 654ae630117a..cafb060c24e5 100644 --- a/src/bun.js/bindings/webcore/MessagePort.cpp +++ b/src/bun.js/bindings/webcore/MessagePort.cpp @@ -57,6 +57,17 @@ MessagePort::~MessagePort() { 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 MessagePort::postMessage(JSC::JSGlobalObject& state, JSC::JSValue messageValue, StructuredSerializeOptions&& options) diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index e6856a0aad0c..d3781e4c2842 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -613,6 +613,34 @@ test("process.emit returns false when there are no listeners", () => { expect(called).toBe(true); }); +test("GC of a ref'd MessagePort whose peer closed releases its event-loop ref", async () => { + // port1.onmessage = fn takes an event-loop ref. port2.close() sets PeerClosed on + // port1's pipe side so hasPendingActivity() → false and port1's wrapper is + // collectible. ~MessagePort() used to not release the event-loop ref → hang. + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + /* js */ ` + (() => { + const { port1, port2 } = new MessageChannel(); + port1.onmessage = () => {}; + port2.close(); + })(); + Bun.gc(true); + setTimeout(() => { Bun.gc(true); console.log("DONE"); }, 50); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout.trim()).toBe("DONE"); + expect(exitCode).toBe(0); +}); + test("transferring a ref'd MessagePort releases its event-loop ref on the source thread", async () => { // port1.onmessage = fn takes an event-loop ref; transferring port1 detaches it from the // source context. disentangle() used to leave that ref behind, so the source process hung. From 5f762a4be5f7cfbdd536c89e050331b3c7ccd21e Mon Sep 17 00:00:00 2001 From: robobun Date: Mon, 4 May 2026 03:24:04 +0000 Subject: [PATCH 22/30] MessagePort: re-enable m_hasRef on first message listener (Node parity) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Node's setupPortReferencing installs a 'newListener' hook that calls this.ref() when the first 'message' listener is added. Mirror that in onDidChangeListenerImpl's Add case so port.unref(); port.on('message', fn) re-refs (matching Node), while port.on('message', fn); port.unref() stays unref'd (also matching Node). Before this, the updateEventLoopRef() predicate gated on m_hasRef which stayed false after unref(), so the unref→on ordering left the port without an event-loop ref — a regression from the refactor since the old Add case reffed unconditionally. --- src/bun.js/bindings/webcore/MessagePort.cpp | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/bun.js/bindings/webcore/MessagePort.cpp b/src/bun.js/bindings/webcore/MessagePort.cpp index cafb060c24e5..f30c48d0e74c 100644 --- a/src/bun.js/bindings/webcore/MessagePort.cpp +++ b/src/bun.js/bindings/webcore/MessagePort.cpp @@ -294,6 +294,12 @@ void MessagePort::onDidChangeListenerImpl(EventTarget& self, const AtomString& e auto& port = static_cast(self); switch (kind) { case Add: + // 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: From 000ac8fa86ad1e1b223b72b64592c20705e68206 Mon Sep 17 00:00:00 2001 From: robobun Date: Mon, 4 May 2026 04:09:22 +0000 Subject: [PATCH 23/30] test: cover port.unref(); port.on() re-ref ordering (Node newListener parity) --- .../worker_threads/worker_threads.test.ts | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index d3781e4c2842..2ad0096d8411 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -668,6 +668,41 @@ test("transferring a ref'd MessagePort releases its event-loop ref on the source expect(exitCode).toBe(0); }); +test("on() after unref() on a transferred port re-refs (Node's newListener hook)", async () => { + // Node's setupPortReferencing installs a 'newListener' hook that calls + // this.ref() on the first 'message' listener, so port.unref(); port.on('message', fn) + // re-refs and the worker stays alive. The reverse order (on; unref) stays unref'd. + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + /* js */ ` + const { Worker, workerData } = require("node:worker_threads"); + const { once } = require("node:events"); + const { port1, port2 } = new MessageChannel(); + const w = new Worker( + "const p = require('node:worker_threads').workerData.port;" + + "p.unref(); p.on('message', m => { console.log('got:' + m); p.close(); });" + + "require('node:worker_threads').parentPort.postMessage(p.hasRef());", + { ev${/* bundler hates eval */ ""}al: true, workerData: { port: port2 }, transferList: [port2] }, + ); + once(w, "message").then(([hasRef]) => { + console.log("hasRef:" + hasRef); + setTimeout(() => port1.postMessage("hi"), 50); + }); + w.on("exit", c => { console.log("exit:" + c); port1.close(); }); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout.trim().split("\n").sort()).toEqual(["exit:0", "got:hi", "hasRef:true"]); + expect(exitCode).toBe(0); +}); + test("worker does not stay alive after unref() on a transferred port with a listener", async () => { // A transferred MessagePort with a 'message' listener used to hold a separate // event loop ref that .unref() could not release, keeping the worker alive forever. From ca91e0ea4b178c27cb0a02e39dc16138abfe81f7 Mon Sep 17 00:00:00 2001 From: robobun Date: Mon, 4 May 2026 06:05:17 +0000 Subject: [PATCH 24/30] =?UTF-8?q?MessagePort:=20mirror=20Node's=20removeLi?= =?UTF-8?q?stener=20=E2=86=92=20unref()=20on=20last=20message=20listener?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Symmetric counterpart to 5f762a4b: Node's setupPortReferencing installs a 'removeListener' hook that calls this.unref() when the last 'message' listener is removed. Mirror that in the Remove/Clear N→0 transition so port.ref(); port.on(...); port.off(...); port.hasRef() → false, matching Node. Not a regression (pre-PR behaved the same), but completes the newListener/removeListener pair. --- src/bun.js/bindings/webcore/MessagePort.cpp | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/src/bun.js/bindings/webcore/MessagePort.cpp b/src/bun.js/bindings/webcore/MessagePort.cpp index f30c48d0e74c..455cf1e48dfd 100644 --- a/src/bun.js/bindings/webcore/MessagePort.cpp +++ b/src/bun.js/bindings/webcore/MessagePort.cpp @@ -294,18 +294,27 @@ void MessagePort::onDidChangeListenerImpl(EventTarget& self, const AtomString& e auto& port = static_cast(self); switch (kind) { case Add: - // 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. + // Node's setupPortReferencing installs 'newListener'/'removeListener' + // hooks that call this.ref() on the first 'message' listener and + // this.unref() when the last one is removed. Mirror that so + // port.unref(); port.on('message', fn) re-refs, and + // port.ref(); port.on(...); port.off(...) un-refs — both matching Node. if (port.m_messageEventCount == 0) port.m_hasRef = true; port.m_messageEventCount++; break; case Remove: port.m_messageEventCount--; + if (port.m_messageEventCount == 0) { + port.m_hasRef = false; + port.m_wantsExplicitRef = false; + } break; case Clear: + if (port.m_messageEventCount > 0) { + port.m_hasRef = false; + port.m_wantsExplicitRef = false; + } port.m_messageEventCount = 0; break; } From ba2b25b11fad46de2b777de31ab740e4838cfe28 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Fri, 15 May 2026 11:50:14 +0000 Subject: [PATCH 25/30] [autofix.ci] apply automated fixes --- src/crash_handler/lib.rs | 14 +++++++-- src/errno/lib.rs | 14 +++++++-- src/perf/tracy.rs | 7 ++++- src/runtime/cli/Arguments.rs | 5 +++- src/runtime/cli/run_command.rs | 45 ++++++++++++++++++++++------ src/runtime/cli/upgrade_command.rs | 7 ++++- src/runtime/jsc_hooks.rs | 7 +++-- src/runtime/webview/ChromeProcess.rs | 12 ++++++-- src/spawn/process.rs | 11 +++++-- src/spawn_sys/spawn_process.rs | 5 +++- 10 files changed, 103 insertions(+), 24 deletions(-) diff --git a/src/crash_handler/lib.rs b/src/crash_handler/lib.rs index e2773ce194de..4ff1536c53ab 100644 --- a/src/crash_handler/lib.rs +++ b/src/crash_handler/lib.rs @@ -1766,7 +1766,12 @@ mod draft { .store(handle as *mut core::ffi::c_void, Ordering::Relaxed); } } - #[cfg(any(target_os = "macos", target_os = "linux", target_os = "android", target_os = "freebsd"))] + #[cfg(any( + target_os = "macos", + target_os = "linux", + target_os = "android", + target_os = "freebsd" + ))] { reset_on_posix(); } @@ -2907,7 +2912,12 @@ mod draft { let _ = spawn_result; let _ = url; } - #[cfg(any(target_os = "macos", target_os = "linux", target_os = "android", target_os = "freebsd"))] + #[cfg(any( + target_os = "macos", + target_os = "linux", + target_os = "android", + target_os = "freebsd" + ))] { let mut buf = bun_core::PathBuffer::default(); let mut buf2 = bun_core::PathBuffer::default(); diff --git a/src/errno/lib.rs b/src/errno/lib.rs index abe6ee0ccf7a..8142a6c52ea5 100644 --- a/src/errno/lib.rs +++ b/src/errno/lib.rs @@ -409,7 +409,12 @@ mod errno_name_tests { assert_eq!(Error::from_errno(0), Error::UNEXPECTED); assert_eq!(Error::from_errno(9999), Error::UNEXPECTED); // errno 11 is platform-specific: EAGAIN on linux/windows, EDEADLK on darwin/bsd. - #[cfg(any(target_os = "linux", target_os = "android", windows, target_family = "wasm"))] + #[cfg(any( + target_os = "linux", + target_os = "android", + windows, + target_family = "wasm" + ))] { assert_eq!(Error::from_errno(11), Error::intern("EAGAIN")); assert_eq!(Error::from_errno(104), Error::intern("ECONNRESET")); @@ -464,7 +469,12 @@ mod errno_name_tests { coreutils_error_map::get(2), Some("No such file or directory") ); - #[cfg(any(target_os = "linux", target_os = "android", windows, target_family = "wasm"))] + #[cfg(any( + target_os = "linux", + target_os = "android", + windows, + target_family = "wasm" + ))] assert_eq!( coreutils_error_map::get(11), Some("Resource temporarily unavailable") diff --git a/src/perf/tracy.rs b/src/perf/tracy.rs index a4d304fd3fa3..16ee81aaddd0 100644 --- a/src/perf/tracy.rs +++ b/src/perf/tracy.rs @@ -756,7 +756,12 @@ fn dlsym(symbol: &'static core::ffi::CStr) -> Option { ]; #[cfg(windows)] const PATHS_TO_TRY: &[&core::ffi::CStr] = &[c"tracy.dll"]; - #[cfg(not(any(target_os = "macos", target_os = "linux", target_os = "android", windows)))] + #[cfg(not(any( + target_os = "macos", + target_os = "linux", + target_os = "android", + windows + )))] const PATHS_TO_TRY: &[&core::ffi::CStr] = &[]; // TODO(port): RTLD flags — Zig used `@bitCast(@as(i32, -2))` on diff --git a/src/runtime/cli/Arguments.rs b/src/runtime/cli/Arguments.rs index 464ee8767ba3..63c1a2520e8c 100644 --- a/src/runtime/cli/Arguments.rs +++ b/src/runtime/cli/Arguments.rs @@ -679,7 +679,10 @@ pub const BASE_RUNTIME_TRANSPILER_PARAMS: &[ParamType] = // built with `comptime_table!(.., cold)` and stay in plain `.rodata`, where // `src/startup.order` can still cluster the ones a sampled cold path actually // hits without weighing down the `.rodata.startup` fault-around window. -#[cfg_attr(any(target_os = "linux", target_os = "android"), unsafe(link_section = ".rodata.startup"))] +#[cfg_attr( + any(target_os = "linux", target_os = "android"), + unsafe(link_section = ".rodata.startup") +)] pub static AUTO_TABLE: &clap::ConvertedTable = clap::comptime_table!(AUTO_PARAMS); pub static RUN_TABLE: &clap::ConvertedTable = clap::comptime_table!(RUN_PARAMS, cold); pub static BUILD_TABLE: &clap::ConvertedTable = clap::comptime_table!(BUILD_PARAMS, cold); diff --git a/src/runtime/cli/run_command.rs b/src/runtime/cli/run_command.rs index 03a0613055e2..999e7d596a3f 100644 --- a/src/runtime/cli/run_command.rs +++ b/src/runtime/cli/run_command.rs @@ -552,7 +552,10 @@ Full documentation is available at https://bun.com/docs/cli/run /// not share `.text` pages with the hot `bun run