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

@@ -377,16 +377,19 @@
     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) {
@@ -413,6 +416,7 @@
       return this;
     },
   };
+  return asyncHook;
 }
 
 const executionAsyncIdNotImpl = createWarning(

src/js/node/worker_threads.ts

@@ -100,6 +100,11 @@
 // 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
@@ -928,6 +933,9 @@
   #stderr;
   #stdoutAutoPipe = false;
   #stderrAutoPipe = false;
+  // 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
@@ -1088,6 +1096,45 @@
       }
       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() {
@@ -1105,13 +1152,17 @@
     if (!this.#stdoutAutoPipe) this.#stdoutPort?.ref();
     if (!this.#stderrAutoPipe) this.#stderrPort?.ref();
     this.#stdinPort?.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.#stdoutAutoPipe) this.#stdoutPort?.unref();
     if (!this.#stderrAutoPipe) this.#stderrPort?.unref();
     this.#stdinPort?.unref();
+    if (!this.#exited) this.#hasRef = false;
   }
 
   get stdin() {
@@ -1305,6 +1356,10 @@
     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

@@ -1978,6 +1978,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.
     ///
@@ -2069,8 +2078,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 @@ static inline JSC::EncodedJSValue jsBroadcastChannelPrototypeFunction_refBody(JS
     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>
@@ -539,6 +540,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;
 

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());
+}

test/js/node/test/parallel/test-worker-error-stack-getter-throws.js

@@ -0,0 +1,22 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const { Worker } = require('worker_threads');
+
+const w = new Worker(
+  `const fn = (err) => {
+    if (err.message === 'fhqwhgads')
+      throw new Error('come on');
+    return 'This is my custom stack trace!';
+  };
+  Error.prepareStackTrace = fn;
+  throw new Error('fhqwhgads');
+  `,
+  { eval: true }
+);
+w.on('message', common.mustNotCall());
+w.on('error', common.mustCall((err) => {
+  assert.strictEqual(err.stack, undefined);
+  assert.strictEqual(err.message, 'fhqwhgads');
+  assert.strictEqual(err.name, 'Error');
+}));

test/js/node/test/parallel/test-worker-hasref.js

@@ -0,0 +1,33 @@
+'use strict';
+const common = require('../common');
+
+const { Worker } = require('worker_threads');
+const { createHook } = require('async_hooks');
+const assert = require('assert');
+
+let handle;
+
+createHook({
+  init(asyncId, type, triggerAsyncId, resource) {
+    if (type === 'WORKER') {
+      handle = resource;
+      this.disable();
+    }
+  }
+}).enable();
+
+const w = new Worker('', { eval: true });
+
+assert.strictEqual(handle.hasRef(), true);
+w.unref();
+assert.strictEqual(handle.hasRef(), false);
+w.ref();
+assert.strictEqual(handle.hasRef(), true);
+
+w.on('exit', common.mustCall((exitCode) => {
+  assert.strictEqual(exitCode, 0);
+  assert.strictEqual(handle.hasRef(), true);
+  setTimeout(common.mustCall(() => {
+    assert.strictEqual(handle.hasRef(), undefined);
+  }), 0);
+}));

test/js/node/test/parallel/test-worker-http2-stream-terminate.js

@@ -0,0 +1,63 @@
+'use strict';
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const { duplexPair } = require('stream');
+const { parentPort, Worker } = require('worker_threads');
+
+// This test ensures that workers can be terminated without error while
+// stream activity is ongoing, in particular the C++ function
+// ReportWritesToJSStreamListener::OnStreamAfterReqFinished.
+
+const MAX_ITERATIONS = 5;
+const MAX_THREADS = 6;
+
+// Do not use isMainThread so that this test itself can be run inside a Worker.
+if (!process.env.HAS_STARTED_WORKER) {
+  process.env.HAS_STARTED_WORKER = 1;
+
+  function spinWorker(iter) {
+    const w = new Worker(__filename);
+    w.on('message', common.mustCall((msg) => {
+      assert.strictEqual(msg, 'terminate');
+      w.terminate();
+    }));
+
+    w.on('exit', common.mustCall(() => {
+      if (iter < MAX_ITERATIONS)
+        spinWorker(++iter);
+    }));
+  }
+
+  for (let i = 0; i < MAX_THREADS; i++) {
+    spinWorker(0);
+  }
+} else {
+  const server = http2.createServer();
+  let i = 0;
+  server.on('stream', (stream, headers) => {
+    if (i === 1) {
+      parentPort.postMessage('terminate');
+    }
+    i++;
+
+    stream.end('');
+  });
+
+  const [ clientSide, serverSide ] = duplexPair();
+  server.emit('connection', serverSide);
+
+  const client = http2.connect('http://localhost:80', {
+    createConnection: () => clientSide,
+  });
+
+  function makeRequests() {
+    for (let i = 0; i < 3; i++) {
+      client.request().end();
+    }
+    setImmediate(makeRequests);
+  }
+  makeRequests();
+}