worker: add diagnostic channel 'worker_threads.main.registry'

doc/api/diagnostics_channel.md

@@ -2002,6 +2002,19 @@ added: v16.18.0
 
 Emitted when a new thread is created.
 
+##### Event: `'worker_threads.main.registry'`
+
+* `threadIds` {number\[]} A snapshot of the registered worker thread IDs.
+
+Emitted only on the main thread whenever a worker's messaging port is registered
+or unregistered, including workers created by other workers. The main thread's
+ID (`0`) is not included.
+
+Registration can occur before a worker is online. Unregistration occurs when the
+parent thread processes the worker's exit, so it can be delayed if the parent
+thread is blocked. The snapshot represents registered messaging ports, not a
+guarantee that the listed workers are ready or still running.
+
 #### SQLite
 
 <!-- YAML

lib/internal/worker/messaging.js

@@ -1,6 +1,7 @@
 'use strict';
 
 const {
+  ArrayFrom,
   AtomicsNotify,
   AtomicsStore,
   AtomicsWaitAsync,
@@ -30,6 +31,8 @@ const { MessageChannel } = require('internal/worker/io');
 
 const { validateNumber } = require('internal/validators');
 
+const registryChannel = require('diagnostics_channel').channel('worker_threads.main.registry');
+
 const messageTypes = {
   REGISTER_MAIN_THREAD_PORT: 'registerMainThreadPort',
   UNREGISTER_MAIN_THREAD_PORT: 'unregisterMainThreadPort',
@@ -72,12 +75,20 @@ function handleMessageFromThread(message) {
 
         // Never block the thread on this port
         port.unref();
+
+        if (registryChannel.hasSubscribers) {
+          registryChannel.publish({ threadIds: ArrayFrom(threadsPorts.keys()) });
+        }
       }
 
       break;
     case messageTypes.UNREGISTER_MAIN_THREAD_PORT:
       threadsPorts.get(message.threadId).close();
       threadsPorts.delete(message.threadId);
+
+      if (registryChannel.hasSubscribers) {
+        registryChannel.publish({ threadIds: ArrayFrom(threadsPorts.keys()) });
+      }
       break;
     case messageTypes.SEND_MESSAGE_TO_WORKER:
       {

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

@@ -0,0 +1,81 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const { once } = require('events');
+const dc = require('diagnostics_channel');
+const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
+
+const channelName = 'worker_threads.main.registry';
+
+if (isMainThread) {
+  // Expect one snapshot for each registration and removal in the worker tree.
+  const snapshots = [];
+  const childRegistered = Promise.withResolvers();
+  const childRemoved = Promise.withResolvers();
+
+  dc.subscribe(channelName, common.mustCall((message) => {
+    assert.deepStrictEqual(Object.keys(message), ['threadIds']);
+    snapshots.push(message.threadIds);
+
+    if (snapshots.length === 2) {
+      childRegistered.resolve();
+    }
+    if (snapshots.length === 3) {
+      childRemoved.resolve();
+    }
+  }, 4));
+
+  async function test() {
+    const worker = new Worker(__filename);
+    const threadId = worker.threadId;
+
+    // The child ID and its registry update arrive on independent channels.
+    const [[childId]] = await Promise.all([
+      once(worker, 'message'),
+      once(worker, 'online'),
+      childRegistered.promise,
+    ]);
+
+    assert.deepStrictEqual(snapshots, [[threadId], [threadId, childId]]);
+
+    // Remove the nested worker first, keeping its parent alive for the update.
+    worker.postMessage('stop');
+    await childRemoved.promise;
+
+    assert.deepStrictEqual(snapshots, [[threadId], [threadId, childId], [threadId]]);
+
+    // Let the parent exit naturally, then check that earlier snapshots are intact.
+    const exited = once(worker, 'exit');
+    worker.postMessage('finish');
+    const [code] = await exited;
+
+    assert.strictEqual(code, 0);
+    assert.deepStrictEqual(snapshots, [[threadId], [threadId, childId], [threadId], []]);
+  }
+
+  test().then(common.mustCall());
+} else {
+  // Neither the parent worker nor its child should publish this channel locally.
+  dc.subscribe(channelName, common.mustNotCall());
+
+  if (workerData === 'child') {
+    // Keep the child alive until its parent explicitly terminates it.
+    parentPort.on('message', common.mustNotCall());
+  } else {
+    async function run() {
+      const worker = new Worker(__filename, { workerData: 'child' });
+      await once(worker, 'online');
+      parentPort.postMessage(worker.threadId);
+
+      // The main thread requests termination after observing both registrations.
+      await once(parentPort, 'message');
+      assert.strictEqual(await worker.terminate(), 1);
+
+      // Wait until the main thread observes the child's removal before exiting.
+      await once(parentPort, 'message');
+      parentPort.close();
+    }
+
+    run().then(common.mustCall());
+  }
+}