diagnostics_channel: early-exit tracing channel trace methods

PR-URL: #51915
Reviewed-By: Yagiz Nizipli <[email protected]>
Reviewed-By: Gerhard Stöbich <[email protected]>
Reviewed-By: Chengzhong Wu <[email protected]>

Author: Stephen Belanger

doc/api/diagnostics_channel.md

@@ -781,6 +781,11 @@ if the given function throws an error. This will run the given function using
 [`channel.runStores(context, ...)`][] on the `start` channel which ensures all
 events should have any bound stores set to match this trace context.
 
+To ensure only correct trace graphs are formed, events will only be published
+if subscribers are present prior to starting the trace. Subscriptions which are
+added after the trace begins will not receive future events from that trace,
+only future traces will be seen.
+
 ```mjs
 import diagnostics_channel from 'node:diagnostics_channel';
 
@@ -829,6 +834,11 @@ returned promise rejects. This will run the given function using
 [`channel.runStores(context, ...)`][] on the `start` channel which ensures all
 events should have any bound stores set to match this trace context.
 
+To ensure only correct trace graphs are formed, events will only be published
+if subscribers are present prior to starting the trace. Subscriptions which are
+added after the trace begins will not receive future events from that trace,
+only future traces will be seen.
+
 ```mjs
 import diagnostics_channel from 'node:diagnostics_channel';
 
@@ -881,6 +891,11 @@ the callback is set. This will run the given function using
 [`channel.runStores(context, ...)`][] on the `start` channel which ensures all
 events should have any bound stores set to match this trace context.
 
+To ensure only correct trace graphs are formed, events will only be published
+if subscribers are present prior to starting the trace. Subscriptions which are
+added after the trace begins will not receive future events from that trace,
+only future traces will be seen.
+
 ```mjs
 import diagnostics_channel from 'node:diagnostics_channel';
 
@@ -969,6 +984,11 @@ of the callback while the `error` will either be a thrown error visible in the
 `end` event or the first callback argument in either of the `asyncStart` or
 `asyncEnd` events.
 
+To ensure only correct trace graphs are formed, events should only be published
+if subscribers are present prior to starting the trace. Subscriptions which are
+added after the trace begins should not receive future events from that trace,
+only future traces will be seen.
+
 Tracing channels should follow a naming pattern of:
 
 * `tracing:module.class.method:start` or `tracing:module.function:start`

lib/diagnostics_channel.js

@@ -6,6 +6,7 @@ const {
   ArrayPrototypePush,
   ArrayPrototypeSplice,
   SafeFinalizationRegistry,
+  ObjectDefineProperty,
   ObjectGetPrototypeOf,
   ObjectSetPrototypeOf,
   Promise,
@@ -250,35 +251,40 @@ function assertChannel(value, name) {
   }
 }
 
+function tracingChannelFrom(nameOrChannels, name) {
+  if (typeof nameOrChannels === 'string') {
+    return channel(`tracing:${nameOrChannels}:${name}`);
+  }
+
+  if (typeof nameOrChannels === 'object' && nameOrChannels !== null) {
+    const channel = nameOrChannels[name];
+    assertChannel(channel, `nameOrChannels.${name}`);
+    return channel;
+  }
+
+  throw new ERR_INVALID_ARG_TYPE('nameOrChannels',
+                                 ['string', 'object', 'TracingChannel'],
+                                 nameOrChannels);
+}
+
 class TracingChannel {
   constructor(nameOrChannels) {
-    if (typeof nameOrChannels === 'string') {
-      this.start = channel(`tracing:${nameOrChannels}:start`);
-      this.end = channel(`tracing:${nameOrChannels}:end`);
-      this.asyncStart = channel(`tracing:${nameOrChannels}:asyncStart`);
-      this.asyncEnd = channel(`tracing:${nameOrChannels}:asyncEnd`);
-      this.error = channel(`tracing:${nameOrChannels}:error`);
-    } else if (typeof nameOrChannels === 'object') {
-      const { start, end, asyncStart, asyncEnd, error } = nameOrChannels;
-
-      assertChannel(start, 'nameOrChannels.start');
-      assertChannel(end, 'nameOrChannels.end');
-      assertChannel(asyncStart, 'nameOrChannels.asyncStart');
-      assertChannel(asyncEnd, 'nameOrChannels.asyncEnd');
-      assertChannel(error, 'nameOrChannels.error');
-
-      this.start = start;
-      this.end = end;
-      this.asyncStart = asyncStart;
-      this.asyncEnd = asyncEnd;
-      this.error = error;
-    } else {
-      throw new ERR_INVALID_ARG_TYPE('nameOrChannels',
-                                     ['string', 'object', 'Channel'],
-                                     nameOrChannels);
+    for (const eventName of traceEvents) {
+      ObjectDefineProperty(this, eventName, {
+        __proto__: null,
+        value: tracingChannelFrom(nameOrChannels, eventName),
+      });
     }
   }
 
+  get hasSubscribers() {
+    return this.start.hasSubscribers ||
+      this.end.hasSubscribers ||
+      this.asyncStart.hasSubscribers ||
+      this.asyncEnd.hasSubscribers ||
+      this.error.hasSubscribers;
+  }
+
   subscribe(handlers) {
     for (const name of traceEvents) {
       if (!handlers[name]) continue;
@@ -302,6 +308,10 @@ class TracingChannel {
   }
 
   traceSync(fn, context = {}, thisArg, ...args) {
+    if (!this.hasSubscribers) {
+      return ReflectApply(fn, thisArg, args);
+    }
+
     const { start, end, error } = this;
 
     return start.runStores(context, () => {
@@ -320,6 +330,10 @@ class TracingChannel {
   }
 
   tracePromise(fn, context = {}, thisArg, ...args) {
+    if (!this.hasSubscribers) {
+      return ReflectApply(fn, thisArg, args);
+    }
+
     const { start, end, asyncStart, asyncEnd, error } = this;
 
     function reject(err) {
@@ -358,6 +372,10 @@ class TracingChannel {
   }
 
   traceCallback(fn, position = -1, context = {}, thisArg, ...args) {
+    if (!this.hasSubscribers) {
+      return ReflectApply(fn, thisArg, args);
+    }
+
     const { start, end, asyncStart, asyncEnd, error } = this;
 
     function wrappedCallback(err, res) {

test/parallel/test-diagnostics-channel-tracing-channel-args-types.js

@@ -24,7 +24,7 @@ assert.throws(() => (channel = dc.tracingChannel(0)), {
   code: 'ERR_INVALID_ARG_TYPE',
   name: 'TypeError',
   message:
-    /The "nameOrChannels" argument must be of type string or an instance of Channel or Object/,
+    /The "nameOrChannels" argument must be of type string or an instance of TracingChannel or Object/,
 });
 
 // tracingChannel creating without instance of Channel must throw error

test/parallel/test-diagnostics-channel-tracing-channel-async.js

@@ -0,0 +1,19 @@
+'use strict';
+
+const common = require('../common');
+const dc = require('diagnostics_channel');
+
+const channel = dc.tracingChannel('test');
+
+const handlers = {
+  start: common.mustNotCall(),
+  end: common.mustNotCall(),
+  asyncStart: common.mustNotCall(),
+  asyncEnd: common.mustNotCall(),
+  error: common.mustNotCall()
+};
+
+// While subscribe occurs _before_ the callback executes,
+// no async events should be published.
+channel.traceCallback(setImmediate, 0, {}, null, common.mustCall());
+channel.subscribe(handlers);

test/parallel/test-diagnostics-channel-tracing-channel-callback-early-exit.js

@@ -15,14 +15,14 @@ function check(found) {
 }
 
 const handlers = {
-  start: common.mustCall(check, 2),
-  end: common.mustCall(check, 2),
-  asyncStart: common.mustCall(check, 2),
-  asyncEnd: common.mustCall(check, 2),
+  start: common.mustCall(check),
+  end: common.mustCall(check),
+  asyncStart: common.mustCall(check),
+  asyncEnd: common.mustCall(check),
   error: common.mustCall((found) => {
     check(found);
     assert.deepStrictEqual(found.error, expectedError);
-  }, 2)
+  })
 };
 
 channel.subscribe(handlers);
@@ -34,13 +34,3 @@ channel.traceCallback(function(cb, err) {
   assert.strictEqual(err, expectedError);
   assert.strictEqual(res, undefined);
 }), expectedError);
-
-channel.tracePromise(function(value) {
-  assert.deepStrictEqual(this, thisArg);
-  return Promise.reject(value);
-}, input, thisArg, expectedError).then(
-  common.mustNotCall(),
-  common.mustCall((value) => {
-    assert.deepStrictEqual(value, expectedError);
-  })
-);

test/parallel/test-diagnostics-channel-tracing-channel-async-error.js renamed to test/parallel/test-diagnostics-channel-tracing-channel-callback-error.js

@@ -0,0 +1,43 @@
+'use strict';
+
+const common = require('../common');
+const dc = require('diagnostics_channel');
+const assert = require('assert');
+
+const channel = dc.tracingChannel('test');
+
+const expectedResult = { foo: 'bar' };
+const input = { foo: 'bar' };
+const thisArg = { baz: 'buz' };
+
+function check(found) {
+  assert.deepStrictEqual(found, input);
+}
+
+function checkAsync(found) {
+  check(found);
+  assert.strictEqual(found.error, undefined);
+  assert.deepStrictEqual(found.result, expectedResult);
+}
+
+const handlers = {
+  start: common.mustCall(check),
+  end: common.mustCall(check),
+  asyncStart: common.mustCall(checkAsync),
+  asyncEnd: common.mustCall(checkAsync),
+  error: common.mustNotCall()
+};
+
+channel.subscribe(handlers);
+
+channel.traceCallback(function(cb, err, res) {
+  assert.deepStrictEqual(this, thisArg);
+  setImmediate(cb, err, res);
+}, 0, input, thisArg, common.mustCall((err, res) => {
+  assert.strictEqual(err, null);
+  assert.deepStrictEqual(res, expectedResult);
+}), null, expectedResult);
+
+assert.throws(() => {
+  channel.traceCallback(common.mustNotCall(), 0, input, thisArg, 1, 2, 3);
+}, /"callback" argument must be of type function/);

test/parallel/test-diagnostics-channel-tracing-channel-callback.js

@@ -0,0 +1,21 @@
+'use strict';
+
+const common = require('../common');
+const dc = require('diagnostics_channel');
+
+const channel = dc.tracingChannel('test');
+
+const handlers = {
+  start: common.mustNotCall(),
+  end: common.mustNotCall(),
+  asyncStart: common.mustNotCall(),
+  asyncEnd: common.mustNotCall(),
+  error: common.mustNotCall()
+};
+
+// While subscribe occurs _before_ the promise resolves,
+// no async events should be published.
+channel.tracePromise(() => {
+  return new Promise(setImmediate);
+}, {});
+channel.subscribe(handlers);

test/parallel/test-diagnostics-channel-tracing-channel-promise-early-exit.js

@@ -0,0 +1,38 @@
+'use strict';
+
+const common = require('../common');
+const dc = require('diagnostics_channel');
+const assert = require('assert');
+
+const channel = dc.tracingChannel('test');
+
+const expectedError = new Error('test');
+const input = { foo: 'bar' };
+const thisArg = { baz: 'buz' };
+
+function check(found) {
+  assert.deepStrictEqual(found, input);
+}
+
+const handlers = {
+  start: common.mustCall(check),
+  end: common.mustCall(check),
+  asyncStart: common.mustCall(check),
+  asyncEnd: common.mustCall(check),
+  error: common.mustCall((found) => {
+    check(found);
+    assert.deepStrictEqual(found.error, expectedError);
+  })
+};
+
+channel.subscribe(handlers);
+
+channel.tracePromise(function(value) {
+  assert.deepStrictEqual(this, thisArg);
+  return Promise.reject(value);
+}, input, thisArg, expectedError).then(
+  common.mustNotCall(),
+  common.mustCall((value) => {
+    assert.deepStrictEqual(value, expectedError);
+  })
+);