fix(server-utils): Ensure orchestrion instrumentation lazy registers on Node (#22518)

Co-Authored-By: isaacs <i@izs.me>
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: 3 people

dev-packages/node-integration-tests/suites/tracing/orchestrion-lazy-registration/instrument.mjs

@@ -0,0 +1,15 @@
+import * as Sentry from '@sentry/node';
+import { loggingTransport } from '@sentry-internal/node-integration-tests';
+
+// Opt into diagnostics-channel injection. This registers the runtime module
+// hook and swaps the OTel integrations for the channel-based ones. The scenario
+// then checks that the channel subscribers are NOT wired up until the module
+// is actually loaded.
+Sentry.experimentalUseDiagnosticsChannelInjection();
+
+Sentry.init({
+  dsn: 'https://public@dsn.ingest.sentry.io/1337',
+  release: '1.0',
+  tracesSampleRate: 1.0,
+  transport: loggingTransport,
+});

dev-packages/node-integration-tests/suites/tracing/orchestrion-lazy-registration/scenario.mjs

@@ -0,0 +1,26 @@
+import { strict as assert } from 'node:assert';
+import { tracingChannel } from 'node:diagnostics_channel';
+
+// `generic-pool` is a default channel integration and a pure, service-free
+// require, so loading it is enough to trigger the runtime hook's injection.
+const channel = tracingChannel('orchestrion:generic-pool:acquire');
+
+// After `init()` but BEFORE `generic-pool` is loaded, a lazily-registering
+// integration must not have subscribed to the channel yet — otherwise every
+// default channel integration would consume channel slots up front (Node caps
+// channels in use at 1024), even for modules the app never loads.
+assert.equal(
+  channel.start.hasSubscribers,
+  false,
+  'expected NO subscribers on orchestrion:generic-pool:acquire before generic-pool is loaded',
+);
+
+// Loading the module triggers the runtime hook to inject it, which is the point
+// at which the integration should wire up its channel subscriber.
+await import('generic-pool');
+
+assert.equal(
+  channel.start.hasSubscribers,
+  true,
+  'expected subscribers on orchestrion:generic-pool:acquire after generic-pool is loaded',
+);

dev-packages/node-integration-tests/suites/tracing/orchestrion-lazy-registration/test.ts

@@ -0,0 +1,24 @@
+import * as path from 'path';
+import { afterAll, test } from 'vitest';
+import { conditionalTest } from '../../../utils';
+import { cleanupChildProcesses, createRunner } from '../../../utils/runner';
+
+afterAll(() => {
+  cleanupChildProcesses();
+});
+
+// The runtime module hook needs Node >= 18.19; gate on 20 to stay on the stable
+// `Module.registerHooks` / `Channel.hasSubscribers` surface.
+conditionalTest({ min: 20 })('orchestrion lazy channel registration', () => {
+  // The scenario self-asserts (via `node:assert`) that a default channel
+  // integration has NOT subscribed to its channel until the instrumented module
+  // is loaded, then that it HAS once loaded. A violation throws, which
+  // `ensureNoErrorOutput` turns into a test failure.
+  test('does not attach channel listeners until the module is loaded', async () => {
+    await createRunner(__dirname, 'scenario.mjs')
+      .withInstrument(path.join(__dirname, 'instrument.mjs'))
+      .ensureNoErrorOutput()
+      .start()
+      .completed();
+  });
+});

packages/core/src/client.ts

@@ -932,6 +932,16 @@ export abstract class Client<O extends ClientOptions = ClientOptions> {
    */
   public on(hook: 'stopUIProfiler', callback: () => void): () => void;
 
+  /**
+   * A hook that is called when an orchestrion-instrumented module is injected at
+   * runtime (by the `--import` module hook). Channel-based integrations use it to
+   * subscribe their diagnostics-channel listeners lazily, only once the module
+   * they instrument is actually loaded. Receives the injected module name.
+   *
+   * @returns {() => void} A function that, when executed, removes the registered callback.
+   */
+  public on(hook: 'orchestrion.module-runtime-injected', callback: (moduleName: string) => void): () => void;
+
   /**
    * Register a hook on this client.
    */
@@ -1193,6 +1203,11 @@ export abstract class Client<O extends ClientOptions = ClientOptions> {
    */
   public emit(hook: 'stopUIProfiler'): void;
 
+  /**
+   * Emit a hook when an orchestrion-instrumented module is injected at runtime.
+   */
+  public emit(hook: 'orchestrion.module-runtime-injected', moduleName: string): void;
+
   /**
    * Emit a hook that was previously registered via `on()`.
    */

packages/server-utils/src/integrations/tracing-channel/amqplib.ts

@@ -3,14 +3,12 @@ import * as diagnosticsChannel from 'node:diagnostics_channel';
 import type { IntegrationFn, Span, SpanAttributes } from '@sentry/core';
 import {
   continueTrace,
-  debug,
   defineIntegration,
   getTraceData,
   SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN,
   SPAN_STATUS_ERROR,
   startInactiveSpan,
   timestampInSeconds,
-  waitForTracingChannelBinding,
 } from '@sentry/core';
 // eslint-disable-next-line typescript/no-deprecated -- NET_PEER_* emitted alongside SERVER_* for backwards compatibility (TODO(v11): remove)
 import {
@@ -27,7 +25,8 @@ import {
   SERVER_PORT,
   URL_FULL,
 } from '@sentry/conventions/attributes';
-import { DEBUG_BUILD } from '../../debug-build';
+import { amqplibModuleNames } from '../../orchestrion/config/amqplib';
+import { invokeOrchestrionInstrumentation } from '../../orchestrion/instrumentation';
 import { CHANNELS } from '../../orchestrion/channels';
 import { bindTracingChannelToSpan } from '../../tracing-channel';
 
@@ -156,36 +155,24 @@ interface AmqpConnectContext {
 
 const NOOP = (): void => {};
 
-// Guards against subscribing to the amqplib channels more than once in a process. Core dedupes
-// `setupOnce` by integration *name*, which is not enough here: the Deno SDK wraps this integration
-// under a different name (`DenoAmqplib`) via `extendIntegration`, so adding both would otherwise run
-// the subscribe logic twice and emit duplicate spans for every operation.
-let subscribed = false;
-
 const _amqplibChannelIntegration = (() => {
   return {
     name: INTEGRATION_NAME,
-    setupOnce() {
-      // `tracingChannel` is unavailable before Node 18.19 so do nothing in that case.
-      if (!diagnosticsChannel.tracingChannel || subscribed) {
-        return;
-      }
-      subscribed = true;
-
-      DEBUG_BUILD && debug.log('[orchestrion:amqplib] subscribing to amqplib tracing channels');
-
-      waitForTracingChannelBinding(() => {
-        subscribeConnect();
-        subscribePublish();
-        subscribeConfirmPublish();
-        subscribeConsume();
-        subscribeDispatch();
-        subscribeSettle();
-      });
+    setup(client) {
+      invokeOrchestrionInstrumentation(client, amqplibModuleNames, instrumentAmqplib, []);
     },
   };
 }) satisfies IntegrationFn;
 
+function instrumentAmqplib(): void {
+  subscribeConnect();
+  subscribePublish();
+  subscribeConfirmPublish();
+  subscribeConsume();
+  subscribeDispatch();
+  subscribeSettle();
+}
+
 /**
  * Producer span for `Channel.prototype.publish`. Creates a PRODUCER span, injects the trace headers
  * into the publish options, and ends when the (synchronous) publish call returns. Skips the confirm

packages/server-utils/src/integrations/tracing-channel/anthropic.ts

@@ -5,7 +5,6 @@ import {
   _INTERNAL_shouldSkipAiProviderWrapping,
   addAnthropicRequestAttributes,
   addAnthropicResponseAttributes,
-  debug,
   defineIntegration,
   extractAnthropicRequestAttributes,
   instrumentAsyncIterableStream,
@@ -14,11 +13,11 @@ import {
   SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN,
   shouldEnableTruncation,
   startInactiveSpan,
-  waitForTracingChannelBinding,
 } from '@sentry/core';
-import { DEBUG_BUILD } from '../../debug-build';
 import { CHANNELS } from '../../orchestrion/channels';
 import { bindTracingChannelToSpan } from '../../tracing-channel';
+import { anthropicAiModuleNames } from '../../orchestrion/config/anthropic-ai';
+import { invokeOrchestrionInstrumentation } from '../../orchestrion/instrumentation';
 
 // Same name as the OTel integration by design: when enabled, the OTel 'Anthropic_AI'
 // integration is dropped from the default set (see the Node opt-in loader).
@@ -47,43 +46,34 @@ interface AnthropicChannelContext {
   result?: unknown;
 }
 
-let subscribed = false;
-
 const _anthropicChannelIntegration = ((options: AnthropicAiOptions = {}) => {
   return {
     name: INTEGRATION_NAME,
-    setupOnce() {
-      // tracingChannel is unavailable before Node 18.19 and prevent double-subscribe
-      if (!diagnosticsChannel.tracingChannel || subscribed) {
-        return;
-      }
-      subscribed = true;
-
-      // `bindTracingChannelToSpan` needs the async-context binding that `initOpenTelemetry()` registers
-      // after `setupOnce` runs, so wait for it before subscribing.
-      waitForTracingChannelBinding(() => {
-        for (const { channel, operation, methodPath, stream } of INSTRUMENTED_CHANNELS) {
-          DEBUG_BUILD && debug.log(`[orchestrion:anthropic] subscribing to channel "${channel}"`);
-          bindTracingChannelToSpan(
-            diagnosticsChannel.tracingChannel<AnthropicChannelContext>(channel),
-            data => createGenAiSpan(data, operation, methodPath, options),
-            {
-              beforeSpanEnd: (span, data) => {
-                addAnthropicResponseAttributes(
-                  span,
-                  data.result as AnthropicAiResponse,
-                  resolveAIRecordingOptions(options).recordOutputs,
-                );
-              },
-              deferSpanEnd: ({ span, data }) => wrapStreamResult(span, data, stream, options),
-            },
-          );
-        }
-      });
+    setup(client) {
+      invokeOrchestrionInstrumentation(client, anthropicAiModuleNames, instrumentAnthropic, [options]);
     },
   };
 }) satisfies IntegrationFn;
 
+function instrumentAnthropic(options: AnthropicAiOptions): void {
+  for (const { channel, operation, methodPath, stream } of INSTRUMENTED_CHANNELS) {
+    bindTracingChannelToSpan(
+      diagnosticsChannel.tracingChannel<AnthropicChannelContext>(channel),
+      data => createGenAiSpan(data, operation, methodPath, options),
+      {
+        beforeSpanEnd: (span, data) => {
+          addAnthropicResponseAttributes(
+            span,
+            data.result as AnthropicAiResponse,
+            resolveAIRecordingOptions(options).recordOutputs,
+          );
+        },
+        deferSpanEnd: ({ span, data }) => wrapStreamResult(span, data, stream, options),
+      },
+    );
+  }
+}
+
 /**
  * Build the span for an instrumented call.
  * Returning `undefined` opts the payload out so no span is opened.