node:worker_threads: per-thread --use-system-ca, real eventLoopUtilization, --cpu-prof in workers, node's online timing, error.code / stack-getter / timeOrigin fixes, async_hooks WORKER resource, worker_threads dc channel (+10 upstream tests)

src/js/node/async_hooks.ts

@@ -493,16 +493,19 @@ function createHook(hook) {
     throw $ERR_ASYNC_CALLBACK("hook.promiseResolve");
 
   let enabledInit;
-  return {
+  const asyncHook = {
     enable() {
       if (init !== undefined && enabledInit === undefined) {
-        // init is delivered for TickObject resources (process.nextTick);
-        // other resource types are still unimplemented.
+        // init is delivered for TickObject (process.nextTick) and WORKER
+        // resources; other resource types are still unimplemented.
         // Per-instance wrapper: two hooks registered with the same init
         // function must stay independently removable (removal is by
         // identity, and removing the other instance's entry would reorder
         // its callback relative to unrelated hooks).
-        enabledInit = (asyncId, type, triggerAsyncId, resource) => init(asyncId, type, triggerAsyncId, resource);
+        // node invokes init as a method on the AsyncHook instance
+        // (lib/internal/async_hooks.js), so `this.disable()` inside init works.
+        enabledInit = (asyncId, type, triggerAsyncId, resource) =>
+          init.$call(asyncHook, asyncId, type, triggerAsyncId, resource);
         require("internal/async_hooks_tick").tickInitHooks.push(enabledInit);
       }
       if (before !== undefined || after !== undefined || destroy !== undefined || promiseResolve !== undefined) {
@@ -529,6 +532,7 @@ function createHook(hook) {
       return this;
     },
   };
+  return asyncHook;
 }
 
 const executionAsyncIdNotImpl = createWarning(

src/js/node/worker_threads.ts

@@ -100,6 +100,11 @@ type NodeWorkerOptions = import("node:worker_threads").WorkerOptions;
 // Used to ensure that Blobs created to hold the source code for `eval: true` Workers get cleaned up
 // after their Worker exits
 let urlRevokeRegistry: FinalizationRegistry<string> | undefined = undefined;
+// Resolved at module load, as node does (lib/internal/worker.js): resolving it
+// lazily inside the constructor would build the channel out of whatever
+// Map/Object user code had tampered with by then.
+const workerThreadsChannel = require("node:diagnostics_channel").channel("worker_threads");
+const { tickInitHooks, newAsyncId } = require("internal/async_hooks_tick");
 
 function injectFakeEmitter(Class) {
   // Per-instance registry mapping each event to (user listener -> wrapper), so
@@ -935,6 +940,9 @@ class Worker extends EventEmitter {
   #stdin;
   #stdout;
   #stderr;
+  // Mirrors ref()/unref() for the async_hooks WORKER resource's hasRef();
+  // undefined once the thread has exited, as node's handle reads back.
+  #hasRef: boolean | undefined = true;
 
   // 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
@@ -1092,6 +1100,45 @@ class Worker extends EventEmitter {
       }
       urlRevokeRegistry.register(this.#worker, this.#urlToRevoke);
     }
+    // node publishes the newly-constructed Worker here, at the end of the
+    // constructor (lib/internal/worker.js).
+    if (workerThreadsChannel.hasSubscribers) {
+      workerThreadsChannel.publish({ worker: this });
+    }
+    this.#emitAsyncHooksInit();
+  }
+
+  // node's WORKER resource is the C++ handle, so hasRef() follows ref()/unref()
+  // and reads back undefined once the handle is gone. Bun delivered init for
+  // TickObject only; this extends the same array to the type worker consumers
+  // look for. Only hasRef() is exposed, not the full handle.
+  #emitAsyncHooksInit() {
+    const count = tickInitHooks.length;
+    if (count === 0) return;
+    const worker = this;
+    const resource = {
+      hasRef() {
+        return worker.#hasRef;
+      },
+    };
+    const asyncId = newAsyncId();
+    // Snapshot: enable()/disable() from inside a hook must not affect the
+    // in-flight dispatch (node stages such mutations in tmp_array).
+    const snapshot = $newArrayWithSize<Function>(count);
+    for (let i = 0; i < count; i++) snapshot[i] = tickInitHooks[i];
+    for (let i = 0; i < count; i++) {
+      try {
+        snapshot[i](asyncId, "WORKER", 0, resource);
+      } catch (err) {
+        // node: a throwing init hook is fatal (fatalError: print + exit 1),
+        // never surfaced to the `new Worker` caller — which here has already
+        // spawned the thread. console is user-mutable, so shield the print.
+        try {
+          console.error(typeof err?.stack === "string" ? err.stack : err);
+        } catch {}
+        process.exit(1);
+      }
+    }
   }
 
   get threadId() {
@@ -1106,10 +1153,14 @@ class Worker extends EventEmitter {
     // stdio ports are not touched here (node's ref()/unref() only touch the
     // handle and the public port); their ref state tracks in-flight I/O.
     this.#worker.ref();
+    // node's ref()/unref() no-op once the handle is gone, leaving hasRef()
+    // undefined rather than resurrecting it.
+    if (!this.#exited) this.#hasRef = true;
   }
 
   unref() {
     this.#worker.unref();
+    if (!this.#exited) this.#hasRef = false;
   }
 
   get stdin() {
@@ -1291,6 +1342,10 @@ class Worker extends EventEmitter {
     this.#stdinPort?.close();
     this.#onExitPromise = e.code;
     this.emit("exit", e.code);
+    // node's WORKER handle is gone once the thread has exited, so its
+    // hasRef() reads back undefined. 'exit' listeners ran synchronously above
+    // and still saw the live value.
+    this.#hasRef = undefined;
   }
 
   #onError(event: ErrorEvent) {

src/jsc/VirtualMachine.rs

@@ -1988,6 +1988,15 @@ fn get_origin_timestamp() -> u64 {
     (now - ORIGIN_RELATIVE_EPOCH).max(0) as u64
 }
 
+/// `performance.timeOrigin` is the PROCESS start time and every thread reports
+/// the same one — a worker's timeOrigin equals the main thread's in node, and
+/// `performance.now()` inside a worker is relative to that same origin. Capture
+/// it on the first VM so worker VMs inherit it instead of restarting the clock.
+fn process_origin() -> (std::time::Instant, u64) {
+    static ORIGIN: std::sync::OnceLock<(std::time::Instant, u64)> = std::sync::OnceLock::new();
+    *ORIGIN.get_or_init(|| (std::time::Instant::now(), get_origin_timestamp()))
+}
+
 impl VirtualMachine {
     /// `VirtualMachine.init(opts)` — allocate + wire the per-thread VM.
     ///
@@ -2079,8 +2088,9 @@ impl VirtualMachine {
             addr_of_mut!((*vm).pending_internal_promise_reported_at).write(u32::MAX);
             addr_of_mut!((*vm).on_unhandled_rejection)
                 .write(VirtualMachine::default_on_unhandled_rejection);
-            addr_of_mut!((*vm).origin_timer).write(std::time::Instant::now());
-            addr_of_mut!((*vm).origin_timestamp).write(get_origin_timestamp());
+            let (origin_timer, origin_timestamp) = process_origin();
+            addr_of_mut!((*vm).origin_timer).write(origin_timer);
+            addr_of_mut!((*vm).origin_timestamp).write(origin_timestamp);
             addr_of_mut!((*vm).smol).write(opts.smol);
             // `Option<{CPU,Heap}ProfilerConfig>` are NOT zero-valid: each
             // payload contains a `bool`, and rustc picks that field's invalid

src/jsc/bindings/webcore/JSBroadcastChannel.cpp

@@ -387,7 +387,9 @@
     UNUSED_PARAM(throwScope);
     UNUSED_PARAM(callFrame);
     auto& impl = castedThis->wrapped();
-    RELEASE_AND_RETURN(throwScope, JSValue::encode(toJS<IDLUndefined>(*lexicalGlobalObject, throwScope, [&]() -> decltype(auto) { return impl.jsRef(lexicalGlobalObject); })));
+    impl.jsRef(lexicalGlobalObject);
+    // node's ref() returns the channel so it chains (lib/internal/worker/io.js).
+    RELEASE_AND_RETURN(throwScope, JSValue::encode(castedThis));
 }
 
 JSC_DEFINE_HOST_FUNCTION(jsBroadcastChannelPrototypeFunction_ref, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))

src/jsc/bindings/webcore/Worker.cpp

@@ -33,6 +33,7 @@
 #include "Event.h"
 #include "EventNames.h"
 #include "StructuredSerializeOptions.h"
+#include <JavaScriptCore/ErrorInstance.h>
 #include <JavaScriptCore/IteratorOperations.h>
 #include <JavaScriptCore/ScriptCallStack.h>
 #include <wtf/TZoneMallocInlines.h>
@@ -446,31 +447,29 @@ void Worker::rejectAllCrossVMRequests(JSC::JSGlobalObject* globalObject)
 
 // ---- Worker-thread entry points ---------------------------------------------
 
+// Posted before the entry point runs, matching node's 'online', so a worker whose
+// top-level never returns still reports online. Deliberately leaves m_state alone:
+// Pending queues in postTaskToWorkerGlobalScope, it does not reject.
+void Worker::dispatchOnlineEvent()
+{
+    postTaskToParent([protectedThis = Ref { *this }](ScriptExecutionContext&) {
+        if (protectedThis->hasEventListeners(eventNames().openEvent)) {
+            auto event = Event::create(eventNames().openEvent, Event::CanBubble::No, Event::IsCancelable::No);
+            protectedThis->dispatchEvent(event);
+        }
+    });
+}
+
 void Worker::dispatchOnline(Zig::GlobalObject* workerGlobalObject)
 {
     // Pending→Running under the same lock postTaskToWorkerGlobalScope uses, so
     // a message post racing this transition either queues (drained below by
     // fireEarlyMessages) or posts directly — never both, never neither.
-    //
-    // This MUST happen BEFORE the open event is posted to the parent: the
-    // parent's `online` handler may immediately call getHeapSnapshot() (or
-    // anything else gated on isOnline() / postTaskToWorkerGlobalScope()). If
-    // the state flip happens after the post, a fast parent thread can run the
-    // open task while m_state is still Pending and observe
-    // ERR_WORKER_NOT_RUNNING — flaky `await once(worker, "online");
-    // worker.getHeapSnapshot()` in worker_threads.test.ts.
     {
         Locker lock(m_pendingTasksMutex);
         m_state.store(State::Running);
     }
 
-    postTaskToParent([protectedThis = Ref { *this }](ScriptExecutionContext&) {
-        if (protectedThis->hasEventListeners(eventNames().openEvent)) {
-            auto event = Event::create(eventNames().openEvent, Event::CanBubble::No, Event::IsCancelable::No);
-            protectedThis->dispatchEvent(event);
-        }
-    });
-
     auto* thisContext = workerGlobalObject->scriptExecutionContext();
     if (!thisContext) {
         return;
@@ -539,6 +538,24 @@ bool Worker::dispatchErrorWithValue(Zig::GlobalObject* workerGlobalObject, JSVal
 
     auto serialized = SerializedScriptValue::create(*workerGlobalObject, value, SerializationForStorage::No, SerializationErrorMode::NonThrowing);
     CLEAR_IF_EXCEPTION(scope);
+    // Cloning an Error reads `stack`, so a throwing Error.prepareStackTrace takes
+    // the whole error down and the caller reports the pretty-printed text as the
+    // message instead. Node drops only the unreadable `stack`
+    // (lib/internal/error_serdes.js TryGetAllProperties); retry once with an own
+    // undefined `stack` so name/message/code still cross. Safe to mutate: the
+    // worker's own error event and the fallback message are both already
+    // materialized by the time we get here, and the thread is terminating.
+    if (!serialized && !scope.exception()) {
+        if (auto* errorInstance = dynamicDowncast<JSC::ErrorInstance>(value)) {
+            errorInstance->putDirect(vm, vm.propertyNames->stack, JSC::jsUndefined(), JSC::PropertyAttribute::DontEnum | 0);
+            // putDirect bypasses the ErrorInstance property overrides, so tell it
+            // not to re-materialize `stack` (and re-run prepareStackTrace) below.
+            errorInstance->setStackPropertyAlreadyMaterialized();
+            CLEAR_IF_EXCEPTION(scope);
+            serialized = SerializedScriptValue::create(*workerGlobalObject, value, SerializationForStorage::No, SerializationErrorMode::NonThrowing);
+            CLEAR_IF_EXCEPTION(scope);
+        }
+    }
     if (!serialized)
         return false;
 
@@ -713,6 +730,11 @@ extern "C" void WebWorker__entrySettled(Zig::GlobalObject* globalObject)
     CLEAR_IF_EXCEPTION(scope);
 }
 
+extern "C" void WebWorker__dispatchOnlineEvent(Worker* worker)
+{
+    worker->dispatchOnlineEvent();
+}
+
 extern "C" void WebWorker__dispatchOnline(Worker* worker, Zig::GlobalObject* globalObject)
 {
     WebWorker__entrySettled(globalObject);

src/jsc/bindings/webcore/Worker.h

@@ -130,6 +130,7 @@ class Worker final : public ThreadSafeRefCounted<Worker>, public EventTargetWith
     WorkerOptions& options() { return m_options; }
 
     // -- Worker-thread entry points (each posts to m_parentContextId) --------
+    void dispatchOnlineEvent();
     void dispatchOnline(Zig::GlobalObject* workerGlobalObject);
     void fireEarlyMessages(Zig::GlobalObject* workerGlobalObject);
     void dispatchErrorWithMessage(WTF::String message);

src/jsc/web_worker.rs

@@ -30,7 +30,7 @@
 //! Lifecycle of the worker thread (`threadMain`):
 //!   1. `startVM()`  — build a mimalloc arena, clone env, initialise a
 //!      `jsc.VirtualMachine`, publish `vm` under `vm_lock`.
-//!   2. `spin()`     — load the entry point, call `dispatchOnline` +
+//!   2. `spin()`     — post 'online', load the entry point, call `dispatchOnline` +
 //!      `fireEarlyMessages`, run the event loop until it drains or
 //!      `requested_terminate` is observed, run `beforeExit`.
 //!   3. `shutdown()` — call `vm.onExit()`, tear down the JSC VM, post
@@ -221,6 +221,7 @@ unsafe extern "C" {
     // Re-declared here (also private in VM.rs) so `thread_main` can take the
     // API lock as a raw FFI call with NO RAII guard — see the note there.
     safe fn JSC__VM__getAPILock(vm: &jsc::VM);
+    safe fn WebWorker__dispatchOnlineEvent(cpp_worker: *mut c_void);
     safe fn WebWorker__dispatchOnline(cpp_worker: *mut c_void, global: &JSGlobalObject);
     safe fn WebWorker__fireEarlyMessages(cpp_worker: *mut c_void, global: &JSGlobalObject);
     safe fn WebWorker__entrySettled(global: &JSGlobalObject);
@@ -1094,6 +1095,11 @@ impl WebWorker {
             return self.shutdown();
         }
 
+        // node reports 'online' before user code runs, so a worker whose entry
+        // point never returns still goes online. Only the event goes out here:
+        // the Pending→Running flip stays put, so message routing is unchanged.
+        WebWorker__dispatchOnlineEvent(self.cpp_worker);
+
         // `path` borrows the resolver's process-lifetime string store, the
         // standalone module graph, or `self.unresolved_specifier` — all of
         // which outlive the worker VM. `vm.main` stores it as a raw BACKREF
@@ -1146,13 +1152,9 @@ impl WebWorker {
 
         self.flush_logs(vm);
         log!("[{}] event loop start", self.execution_context_id);
-        // dispatchOnline fires the parent-side 'open' event and flips the C++
-        // state to Running (which routes postMessage directly instead of
-        // queuing). It is placed after the entry point has loaded so the parent
-        // observes 'online' only once the worker's top-level code has completed;
-        // moving it earlier would change that observable ordering.
-        // `cpp_worker` is the opaque C++-owned handle round-tripped via `safe fn`;
-        // `vm.global()` yields the live `&JSGlobalObject` published in start_vm.
+        // Flips the C++ state to Running, which routes postMessage directly
+        // instead of queuing; the 'online' event already went out before the
+        // entry point loaded.
         WebWorker__dispatchOnline(self.cpp_worker, vm.global());
         WebWorker__fireEarlyMessages(self.cpp_worker, vm.global());
         self.set_status(Status::Running);

test/js/node/test/parallel/test-diagnostics-channel-worker-threads.js

@@ -0,0 +1,11 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const { Worker } = require('worker_threads');
+const dc = require('diagnostics_channel');
+
+dc.subscribe('worker_threads', common.mustCall(({ worker }) => {
+  assert.strictEqual(worker instanceof Worker, true);
+}));
+
+new Worker('const a = 1;', { eval: true });

test/js/node/test/parallel/test-perf-hooks-worker-timeorigin.js

@@ -0,0 +1,20 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const { Worker } = require('worker_threads');
+
+const w = new Worker(`
+require('worker_threads').parentPort.postMessage(performance.timeOrigin);
+`, { eval: true });
+
+w.on('message', common.mustCall((timeOrigin) => {
+  // PerformanceNodeTiming exposes process milestones so the
+  // `performance.timeOrigin` in the `worker_threads.Worker` must be the start
+  // time of the process.
+  assert.strictEqual(timeOrigin, performance.timeOrigin);
+}));
+
+w.on('exit', common.mustCall((code) => {
+  assert.strictEqual(code, 0);
+}));

test/js/node/test/parallel/test-worker-cleanup-handles.js

@@ -0,0 +1,29 @@
+'use strict';
+const common = require('../common');
+
+const assert = require('assert');
+const fs = require('fs');
+const { Server } = require('net');
+const { Worker, isMainThread, parentPort } = require('worker_threads');
+
+if (isMainThread) {
+  const w = new Worker(__filename);
+  let fd = null;
+  w.on('message', common.mustCall((fd_) => {
+    assert.strictEqual(typeof fd_, 'number');
+    fd = fd_;
+  }));
+  w.on('exit', common.mustCall(() => {
+    if (fd === -1) {
+      // This happens when server sockets don’t have file descriptors,
+      // i.e. on Windows.
+      return;
+    }
+    assert.throws(() => fs.fstatSync(fd), { code: 'EBADF' });
+  }));
+} else {
+  const server = new Server();
+  server.listen(0);
+  parentPort.postMessage(server._handle.fd);
+  server.unref();
+}

test/js/node/test/parallel/test-worker-dispose.mjs

@@ -0,0 +1,8 @@
+import * as common from '../common/index.mjs';
+import { Worker } from 'node:worker_threads';
+
+{
+  // Verifies that the worker is async disposable
+  await using worker = new Worker('for(;;) {}', { eval: true });
+  worker.on('exit', common.mustCall());
+}