node:worker_threads: implement postMessageToThread [1bx5ty]

src/codegen/bake-codegen.ts

@@ -56,7 +56,10 @@ async function run() {
           side: JSON.stringify(side),
           IS_ERROR_RUNTIME: String(file === "error"),
           IS_BUN_DEVELOPMENT: String(!!debug),
-          OVERLAY_CSS: css("../runtime/bake/client/overlay.css", !!debug),
+          // JSON.stringify so the value is an explicit JS string literal;
+          // relying on Bun's define auto-quote fallback breaks when the
+          // bootstrap bun predates the `*`/`?` JSON-lexer fix (314d044c0a).
+          OVERLAY_CSS: JSON.stringify(css("../runtime/bake/client/overlay.css", !!debug)),
         },
         minify: {
           syntax: !debug,

src/js/builtins.d.ts

@@ -680,6 +680,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;

src/js/node/worker_threads.ts

@@ -6,16 +6,19 @@ type WebWorker = InstanceType<typeof globalThis.Worker>;
 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<typeof globalThis.Worker>, 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<typeof globalThis.Worker>, 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<unknown, unknown>,
+  MessagePort | undefined,
 ];
 
 type NodeWorkerOptions = import("node:worker_threads").WorkerOptions;
@@ -439,6 +444,224 @@ 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.
+//
+// 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 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;
+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<number, MessagePort> | 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);
 }
@@ -462,6 +685,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
@@ -491,9 +720,12 @@ 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();
       // Restore any transferList handles that were already neutered by
       // packJSTransferables, so their fds aren't orphaned.
       options[kRestoreJSTransferables]?.();
@@ -502,6 +734,8 @@ class Worker extends EventEmitter {
       }
       throw e;
     }
+    this.#threadId = this.#worker.threadId;
+    createMainThreadPort(this.#threadId, mainThreadPortToMain);
     // The transfer is committed - release fds that were transferred but are
     // not referenced from workerData (nothing will deserialize them).
     options[kFinalizeJSTransferables]?.();
@@ -532,7 +766,7 @@ class Worker extends EventEmitter {
   }
 
   get threadId() {
-    return this.#worker.threadId;
+    return this.#threadId;
   }
 
   ref() {
@@ -609,6 +843,8 @@ class Worker extends EventEmitter {
   }
 
   #onClose(e) {
+    destroyMainThreadPort(this.#threadId);
+    this.#threadId = -1;
     this.#onExitPromise = e.code;
     this.#stdout?.push(null);
     this.#stderr?.push(null);
@@ -682,6 +918,7 @@ export default {
   },
   markAsUntransferable,
   moveMessagePortToContext,
+  postMessageToThread,
   receiveMessageOnPort,
   SHARE_ENV,
   threadId,

src/jsc/ErrorCode.rs

@@ -717,9 +717,17 @@ impl ErrorCode {
     pub const TRACE_EVENTS_CATEGORY_REQUIRED: ErrorCode = ErrorCode(329);
     /// `ERR_TRACE_EVENTS_UNAVAILABLE` (instanceof Error)
     pub const TRACE_EVENTS_UNAVAILABLE: ErrorCode = ErrorCode(330);
+    /// `ERR_WORKER_MESSAGING_ERRORED` (instanceof Error)
+    pub const WORKER_MESSAGING_ERRORED: ErrorCode = ErrorCode(331);
+    /// `ERR_WORKER_MESSAGING_FAILED` (instanceof Error)
+    pub const WORKER_MESSAGING_FAILED: ErrorCode = ErrorCode(332);
+    /// `ERR_WORKER_MESSAGING_SAME_THREAD` (instanceof Error)
+    pub const WORKER_MESSAGING_SAME_THREAD: ErrorCode = ErrorCode(333);
+    /// `ERR_WORKER_MESSAGING_TIMEOUT` (instanceof Error)
+    pub const WORKER_MESSAGING_TIMEOUT: ErrorCode = ErrorCode(334);
 
     /// == C++ `NODE_ERROR_COUNT`.
-    pub const COUNT: u16 = 331;
+    pub const COUNT: u16 = 335;
 }
 
 // ──────────────────────────────────────────────────────────────────────────
@@ -1094,6 +1102,10 @@ impl ErrorCode {
     pub const ERR_SECRETS_INTERACTION_REQUIRED: ErrorCode = ErrorCode::SECRETS_INTERACTION_REQUIRED;
     pub const ERR_HTTP2_GOAWAY_SESSION: ErrorCode = ErrorCode::HTTP2_GOAWAY_SESSION;
     pub const ERR_PROXY_TUNNEL: ErrorCode = ErrorCode::PROXY_TUNNEL;
+    pub const ERR_WORKER_MESSAGING_ERRORED: ErrorCode = ErrorCode::WORKER_MESSAGING_ERRORED;
+    pub const ERR_WORKER_MESSAGING_FAILED: ErrorCode = ErrorCode::WORKER_MESSAGING_FAILED;
+    pub const ERR_WORKER_MESSAGING_SAME_THREAD: ErrorCode = ErrorCode::WORKER_MESSAGING_SAME_THREAD;
+    pub const ERR_WORKER_MESSAGING_TIMEOUT: ErrorCode = ErrorCode::WORKER_MESSAGING_TIMEOUT;
 
     // NOTE: `ERR_SYSTEM_ERROR` / `ERR_CHILD_CLOSED_BEFORE_REPLY` intentionally
     // do NOT live here. They belong to the unrelated enum
@@ -1442,6 +1454,10 @@ static CODE_STR: [&str; ErrorCode::COUNT as usize] = [
     "ERR_INVALID_BUFFER_SIZE",
     "ERR_TRACE_EVENTS_CATEGORY_REQUIRED",
     "ERR_TRACE_EVENTS_UNAVAILABLE",
+    "ERR_WORKER_MESSAGING_ERRORED",
+    "ERR_WORKER_MESSAGING_FAILED",
+    "ERR_WORKER_MESSAGING_SAME_THREAD",
+    "ERR_WORKER_MESSAGING_TIMEOUT",
 ];
 
 // ──────────────────────────────────────────────────────────────────────────