lib: improve diagnostic message support

Support serializable context.diagnostic messages and
optionally listen to diagnostic_channel messages

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode

Author: jasnell

doc/api/bench.md

@@ -316,6 +316,9 @@ added: REPLACEME
 * `name` {string} The benchmark name. **Default:** The `name` property of `fn`,
   or `'<anonymous>'` when `fn` has no name.
 * `options` {Object}
+  * `diagnosticChannels` {Array} String diagnostics channel names, deduplicated
+    and inherited from containing suites by union. Symbol values in the array
+    are silently ignored. **Default:** `[]`.
   * `only` {boolean} When any benchmark or containing suite has `only` set,
     benchmarks without `only` in their hierarchy are skipped. **Default:**
     `false`.
@@ -346,6 +349,12 @@ samples, but their samples are discarded. An exception, rejection, timeout,
 abort, missing timing call, or duplicate timing call stops the current
 benchmark. Later benchmarks continue to run.
 
+For each warmup and measured callback, the runner subscribes to the configured
+diagnostics channels. Each publication queues a context diagnostic whose
+`message` is `{ name, message }`, containing the string channel name and the
+published message. Subscriptions are removed when the callback settles or is
+aborted.
+
 A timeout or abort cannot interrupt synchronous JavaScript and does not forcibly
 cancel asynchronous work that ignores `context.signal`.
 
@@ -388,6 +397,9 @@ added: REPLACEME
 * `name` {string} The suite name. **Default:** The `name` property of `fn`, or
   `'<anonymous>'` when `fn` has no name.
 * `options` {Object}
+  * `diagnosticChannels` {Array} String diagnostics channel names inherited by
+    nested suites and benchmarks. Symbol values in the array are silently
+    ignored. **Default:** `[]`.
   * `only` {boolean} Selects all benchmarks nested in this suite. **Default:**
     `false`.
   * `skip` {boolean|string} Skips all benchmarks nested in this suite.
@@ -663,7 +675,8 @@ message transport from the duration. `record()` is mutually exclusive with
 added: REPLACEME
 -->
 
-* `message` {string} The diagnostic message.
+* `message` {any} A structured-cloneable diagnostic value. With CLI process
+  isolation, it must also be supported by advanced child process serialization.
 * `options` {Object}
   * `level` {string} Either `'info'` or `'warning'`. **Default:** `'info'`.
   * `detail` {any} Additional structured-cloneable diagnostic data. With CLI
@@ -679,10 +692,10 @@ before a callback failure are emitted before the failed `'bench:complete'`
 event and do not themselves cause the benchmark to fail. If a timeout or abort
 wins before the callback settles, queued diagnostics might not be emitted.
 
-The message and options are validated, and detail is cloned, synchronously.
-Calling `diagnostic()` between `context.start()` and `context.end()` therefore
-includes that work in the measured duration. Invalid arguments or an
-uncloneable detail violate the sample contract.
+The message and detail are cloned synchronously. Options are also validated
+synchronously. Calling `diagnostic()` between `context.start()` and
+`context.end()` therefore includes that work in the measured duration. Invalid
+arguments or an uncloneable message or detail violate the sample contract.
 
 ### `context.done()`
 
@@ -751,6 +764,8 @@ isolation, all files share one runner and their plans are emitted before any
 benchmark executes. Plan data contains the benchmark-scoped identity, location,
 tags, and parameters described in [benchmark result][], together with:
 
+* `diagnosticChannels` {string\[]} The inherited string channel names
+  subscribed to during each callback.
 * `samples` {number} The effective maximum number of measured callback
   invocations after run-level overrides.
 * `warmup` {number} The effective number of unreported warmup callback

doc/api/cli.md

@@ -492,9 +492,7 @@ benchmark runner process. This reduces startup overhead but allows module,
 heap, and process state to carry between files. User writes to stdout or stderr
 also share destinations with benchmark reporters in this mode.
 
-The supported modes are `'process'` and `'none'`. Worker-thread isolation is not
-a CLI mode. Higher-level tools can implement it using externally measured
-samples as described in the [benchmark runner][] documentation.
+The supported modes are `'process'` and `'none'`.
 
 ### `--bench-name-pattern=pattern`
 

doc/node.1

@@ -304,6 +304,7 @@ When \fBmode\fR is \fB'none'\fR, all matching files and benchmarks run serially
 benchmark runner process. This reduces startup overhead but allows module,
 heap, and process state to carry between files. User writes to stdout or stderr
 also share destinations with benchmark reporters in this mode.
+The supported modes are \fB'process'\fR and \fB'none'\fR.
 .
 .It Fl -bench-name-pattern Ns = Ns Ar pattern
 Only runs benchmarks whose full hierarchical name matches the JavaScript

lib/internal/bench_runner/benchmark.js

@@ -48,6 +48,7 @@ const { structuredClone } = require('internal/worker/js_transferable');
 const { bigint: hrtime } = process.hrtime;
 const kDefaultSamples = 30;
 const kDefaultWarmup = 0;
+const kEmptyDiagnosticChannels = ObjectFreeze([]);
 const kEmptyNamePath = ObjectFreeze([]);
 const kEmptyParams = ObjectFreeze({ __proto__: null });
 const kEmptyTags = ObjectFreeze([]);
@@ -82,6 +83,33 @@ function canonicalizeTags(tags, parentTags = kEmptyTags) {
   return ObjectFreeze(result);
 }
 
+function canonicalizeDiagnosticChannels(
+  diagnosticChannels,
+  parentDiagnosticChannels = kEmptyDiagnosticChannels,
+) {
+  if (diagnosticChannels === undefined) return parentDiagnosticChannels;
+  if (!ArrayIsArray(diagnosticChannels)) {
+    throw new ERR_INVALID_ARG_TYPE(
+      'options.diagnosticChannels', 'Array', diagnosticChannels);
+  }
+
+  const result = ArrayPrototypeSlice(parentDiagnosticChannels);
+  const seen = new SafeSet(parentDiagnosticChannels);
+  for (let i = 0; i < diagnosticChannels.length; i++) {
+    const name = diagnosticChannels[i];
+    if (typeof name === 'symbol') continue;
+    if (typeof name !== 'string') {
+      throw new ERR_INVALID_ARG_TYPE(
+        `options.diagnosticChannels[${i}]`, ['string', 'symbol'], name);
+    }
+    if (!seen.has(name)) {
+      seen.add(name);
+      ArrayPrototypePush(result, name);
+    }
+  }
+  return ObjectFreeze(result);
+}
+
 function canonicalizeParams(params) {
   if (params === undefined) return kEmptyParams;
   validateObject(params, 'options.params');
@@ -106,15 +134,17 @@ function canonicalizeParams(params) {
   return ObjectFreeze(result);
 }
 
-function validateNodeOptions(options, parentTags) {
+function validateNodeOptions(options, parentTags, parentDiagnosticChannels) {
   validateObject(options, 'options');
-  const { only = false, skip, tags } = options;
+  const { diagnosticChannels, only = false, skip, tags } = options;
   if (typeof only !== 'boolean') {
     throw new ERR_INVALID_ARG_TYPE('options.only', 'boolean', only);
   }
   validateSkip(skip);
   return {
     __proto__: null,
+    diagnosticChannels: canonicalizeDiagnosticChannels(
+      diagnosticChannels, parentDiagnosticChannels),
     only,
     skip,
     tags: canonicalizeTags(tags, parentTags),
@@ -157,7 +187,10 @@ class Suite extends AsyncResource {
   constructor(harness, parent, name, options, fn, loc, isRoot = false) {
     super('BenchSuite');
     const validated = validateNodeOptions(
-      options, parent?.tags ?? kEmptyTags);
+      options,
+      parent?.tags ?? kEmptyTags,
+      parent?.diagnosticChannels ?? kEmptyDiagnosticChannels,
+    );
 
     this.harness = harness;
     this.parent = parent;
@@ -174,6 +207,7 @@ class Suite extends AsyncResource {
       this.namePath,
     ]);
     this.parentId = isRoot || parent.isRoot ? null : parent.suiteId;
+    this.diagnosticChannels = validated.diagnosticChannels;
     this.only = validated.only;
     this.skip = validated.skip;
     this.tags = validated.tags;
@@ -195,7 +229,8 @@ class Suite extends AsyncResource {
 class Bench extends AsyncResource {
   constructor(harness, parent, name, options, fn, loc) {
     super('Benchmark');
-    const validated = validateNodeOptions(options, parent.tags);
+    const validated = validateNodeOptions(
+      options, parent.tags, parent.diagnosticChannels);
     const {
       params,
       samples = kDefaultSamples,
@@ -216,6 +251,7 @@ class Bench extends AsyncResource {
     this.name = name;
     this.fn = fn;
     this.loc = createLocation(loc, harness.entryFile);
+    this.diagnosticChannels = validated.diagnosticChannels;
     this.only = validated.only;
     this.skip = validated.skip;
     this.tags = validated.tags;
@@ -275,15 +311,15 @@ class BenchContext {
       throw new ERR_INVALID_STATE('benchmark sample is no longer active');
     }
     try {
-      validateString(message, 'message');
+      const clonedMessage = structuredClone(message);
       validateObject(options, 'options');
       const { detail, level = 'info' } = options;
       validateString(level, 'options.level');
       if (level !== 'info' && level !== 'warning') {
         throw new ERR_INVALID_ARG_VALUE(
           'options.level', level, "must be 'info' or 'warning'");
       }
-      const diagnostic = { __proto__: null, level, message };
+      const diagnostic = { __proto__: null, level, message: clonedMessage };
       if (detail !== undefined) diagnostic.detail = structuredClone(detail);
       this.#onDiagnostic(diagnostic);
     } catch (error) {

lib/internal/bench_runner/cli.js

@@ -440,7 +440,8 @@ function validateRecord(record) {
       warmup,
       yieldBetweenSamples,
     } = record.data;
-    if (typeof record.data.file !== 'string' ||
+    if (!isStringArray(record.data.diagnosticChannels) ||
+        typeof record.data.file !== 'string' ||
         !NumberIsSafeInteger(record.data.line) || record.data.line < 0 ||
         !NumberIsSafeInteger(record.data.column) || record.data.column < 0 ||
         !isStringArray(record.data.tags) ||
@@ -471,7 +472,7 @@ function validateRecord(record) {
         record.data.phase !== 'measurement') ||
        !NumberIsSafeInteger(record.data.index) || record.data.index < 0 ||
        record.data.index > 0xFFFFFFFF ||
-       typeof record.data.message !== 'string' ||
+       !ObjectPrototypeHasOwnProperty(record.data, 'message') ||
        (record.data.level !== 'info' && record.data.level !== 'warning') ||
        typeof record.data.file !== 'string' ||
        !NumberIsSafeInteger(record.data.line) || record.data.line < 0 ||

lib/internal/bench_runner/harness.js

@@ -24,6 +24,10 @@ const {
 const { getCallerLocation } = internalBinding('util');
 const { exitCodes: { kGenericUserError } } = internalBinding('errors');
 const { AsyncLocalStorage } = require('async_hooks');
+const {
+  subscribe: subscribeToChannel,
+  unsubscribe: unsubscribeFromChannel,
+} = require('diagnostics_channel');
 const { AbortController } = require('internal/abort_controller');
 const {
   AbortError,
@@ -406,6 +410,7 @@ class Harness {
         parentId: benchmark.parentId,
         name: benchmark.name,
         namePath: ArrayPrototypeSlice(benchmark.namePath),
+        diagnosticChannels: ArrayPrototypeSlice(benchmark.diagnosticChannels),
         file: benchmark.loc.file,
         line: benchmark.loc.line,
         column: benchmark.loc.column,
@@ -720,9 +725,37 @@ class Harness {
     const context = new BenchContext(
       benchmark, signal, phase, index,
       (diagnostic) => ArrayPrototypePush(diagnostics, diagnostic));
+    const channels = benchmark.diagnosticChannels;
+    let abortSubscription;
+    let diagnosticError;
+    let diagnosticFailed = false;
+    let subscribed = 0;
+    const onMessage = (message, name) => {
+      if (signal.aborted || diagnosticFailed) return;
+      try {
+        context.diagnostic({ __proto__: null, name, message });
+      } catch (error) {
+        diagnosticError = error;
+        diagnosticFailed = true;
+      }
+    };
+    const unsubscribe = () => {
+      while (subscribed > 0) {
+        subscribed--;
+        unsubscribeFromChannel(channels[subscribed], onMessage);
+      }
+    };
     try {
+      for (let i = 0; i < channels.length; i++) {
+        subscribeToChannel(channels[i], onMessage);
+        subscribed++;
+      }
+      if (subscribed > 0) {
+        abortSubscription = addAbortListener(signal, unsubscribe);
+      }
       await this.#invoke(
         benchmark, benchmark, benchmark.fn, [context]);
+      if (diagnosticFailed) throw diagnosticError;
       const { done, sample } = context.finish();
       return {
         __proto__: null,
@@ -739,6 +772,9 @@ class Harness {
         error,
         failed: true,
       };
+    } finally {
+      abortSubscription?.[SymbolDispose]();
+      unsubscribe();
     }
   }
 

test/fixtures/bench-runner/diagnostic.cjs

@@ -1,11 +1,17 @@
 'use strict';
 
 const { bench } = require('node:bench');
+const { channel } = require('diagnostics_channel');
 
-bench('diagnostic relay', { samples: 1 }, (b) => {
-  b.diagnostic('relayed warning', {
-    detail: { value: 42n },
-    level: 'warning',
-  });
+const channelName = 'node:bench:test:diagnostic';
+const diagnosticChannel = channel(channelName);
+
+bench('diagnostic relay', {
+  diagnosticChannels: [channelName],
+  samples: 1,
+}, (b) => {
+  const message = { value: 42n };
+  diagnosticChannel.publish(message);
+  message.value = 0n;
   b.record({ duration_ns: 1n, operations: 1 });
 });

test/parallel/test-bench-cli.js

@@ -399,11 +399,14 @@ for (const isolation of ['process', 'none']) {
     ({ type }) => type === 'bench:diagnostic').data;
   const completion = records.find(
     ({ type }) => type === 'bench:complete').data;
-  assert.strictEqual(diagnostic.message, 'relayed warning');
-  assert.strictEqual(diagnostic.level, 'warning');
+  assert.deepStrictEqual(diagnostic.message, {
+    name: 'node:bench:test:diagnostic',
+    message: { value: '42' },
+  });
+  assert.strictEqual(diagnostic.level, 'info');
   assert.strictEqual(diagnostic.phase, 'measurement');
   assert.strictEqual(diagnostic.index, 0);
-  assert.deepStrictEqual(diagnostic.detail, { value: '42' });
+  assert.strictEqual(diagnostic.detail, undefined);
   assert.strictEqual(diagnostic.benchId, completion.benchId);
   assert.strictEqual(diagnostic.fileRunId, completion.fileRunId);
   assert.strictEqual(completion.error, undefined);

test/parallel/test-bench-context-errors.js

@@ -59,8 +59,8 @@ runner.bench('reentrant record', { samples: 1 }, (b) => {
 runner.bench('uncloneable detail', { samples: 1 }, (b) => {
   b.record({ duration_ns: 1n, operations: 1, detail: () => {} });
 });
-runner.bench('invalid diagnostic message', { samples: 1 }, (b) => {
-  b.diagnostic(1);
+runner.bench('uncloneable diagnostic message', { samples: 1 }, (b) => {
+  b.diagnostic(() => {});
 });
 runner.bench('invalid diagnostic level', { samples: 1 }, (b) => {
   b.diagnostic('invalid', { level: 'error' });
@@ -113,8 +113,8 @@ runner.bench('caught diagnostic violation', { samples: 1 },
                      'ERR_INVALID_STATE');
   assert.strictEqual(byName.get('uncloneable detail').error.name,
                      'DataCloneError');
-  assert.strictEqual(byName.get('invalid diagnostic message').error.code,
-                     'ERR_INVALID_ARG_TYPE');
+  assert.strictEqual(byName.get('uncloneable diagnostic message').error.name,
+                     'DataCloneError');
   assert.strictEqual(byName.get('invalid diagnostic level').error.code,
                      'ERR_INVALID_ARG_VALUE');
   assert.strictEqual(byName.get('uncloneable diagnostic detail').error.name,