ref(redis): expose diagnostics-channel subscription as an integration

packages/deno/src/integrations/redis.ts

@@ -1,11 +1,7 @@
-// * import so that loading this module doesn't error on Deno versions
-// lacking `tracingChannel` (added in Deno 1.44.3).
-// On older runtimes the integration becomes a no-op.
-import * as dc from 'node:diagnostics_channel';
-import type { RedisDiagnosticChannelResponseHook, RedisTracingChannelFactory } from '@sentry/server-utils';
-import { subscribeRedisDiagnosticChannels } from '@sentry/server-utils';
+import type { RedisDiagnosticChannelResponseHook } from '@sentry/server-utils';
+import { redisIntegration as redisChannelIntegration } from '@sentry/server-utils';
 import type { Integration, IntegrationFn } from '@sentry/core';
-import { defineIntegration } from '@sentry/core';
+import { defineIntegration, extendIntegration } from '@sentry/core';
 import { setAsyncLocalStorageAsyncContextStrategy } from '../async';
 
 const INTEGRATION_NAME = 'DenoRedis' as const;
@@ -19,18 +15,17 @@ export interface DenoRedisIntegrationOptions {
 }
 
 const _denoRedisIntegration = ((options: DenoRedisIntegrationOptions = {}) => {
-  return {
+  // The diagnostics_channel subscription lives in server-utils so it is shared across runtimes; we
+  // extend it here to install Deno's AsyncLocalStorage async-context strategy, which the channel
+  // binding reads via `getTracingChannelBinding`. `extendIntegration` runs the base `setupOnce`
+  // first, but its subscribe is deferred a tick when no binding exists yet, so the strategy set
+  // synchronously below is in place by the time the deferred subscribe runs.
+  return extendIntegration(redisChannelIntegration({ responseHook: options.responseHook }), {
     name: INTEGRATION_NAME,
     setupOnce() {
-      if (!dc.tracingChannel) {
-        return;
-      }
-      // The span is bound into Deno's AsyncLocalStorage context via the async-context
-      // strategy's `getTracingChannelBinding`, so the native channel can be passed directly.
       setAsyncLocalStorageAsyncContextStrategy();
-      subscribeRedisDiagnosticChannels(dc.tracingChannel as RedisTracingChannelFactory, options.responseHook);
     },
-  };
+  });
 }) satisfies IntegrationFn;
 
 /**

packages/node/src/integrations/tracing/redis/index.ts

@@ -1,7 +1,7 @@
 import type { IntegrationFn } from '@sentry/core';
-import { defineIntegration, waitForTracingChannelBinding } from '@sentry/core';
+import { defineIntegration, extendIntegration } from '@sentry/core';
 import * as dc from 'node:diagnostics_channel';
-import { subscribeRedisDiagnosticChannels, type RedisTracingChannelFactory } from '@sentry/server-utils';
+import { redisIntegration as redisChannelIntegration } from '@sentry/server-utils';
 import { generateInstrumentOnce } from '@sentry/node-core';
 import { isDiagnosticsChannelInjectionEnabled } from '../../../sdk/diagnosticsChannelInjection';
 import { cacheResponseHook, type RedisOptions, setRedisOptions } from './cache';
@@ -41,17 +41,6 @@ export const instrumentRedis = Object.assign(
       instrumentIORedis();
     }
     instrumentRedisModule();
-    // node-redis >= 5.12.0 and ioredis >= 5.11.0 publish via diagnostics_channel.
-    // `bindTracingChannelToSpan` (inside the subscriber) makes the span the active
-    // OTel context via `bindStore`, which needs the Sentry OTel context manager to
-    // be registered — `initOpenTelemetry()` does that after integration `setupOnce`,
-    // so defer to the next tick.
-    // Check this here to ensure this does not fail at runtime for Node <= 18.18.0
-    if (dc.tracingChannel) {
-      waitForTracingChannelBinding(() => {
-        subscribeRedisDiagnosticChannels(dc.tracingChannel as RedisTracingChannelFactory, cacheResponseHook);
-      });
-    }
 
     // todo: implement them gradually
     // new LegacyRedisInstrumentation({}),
@@ -60,13 +49,17 @@ export const instrumentRedis = Object.assign(
 );
 
 const _redisIntegration = ((options: RedisOptions = {}) => {
-  return {
+  // The diagnostics_channel subscription (node-redis >= 5.12.0, ioredis >= 5.11.0) lives in
+  // server-utils so it is shared across server runtimes; we extend it here to also run the vendored
+  // OTel patchers for older client versions. `cacheResponseHook` reads options set in the extension's
+  // `setupOnce` below, but it only runs at command time, by which point those options are set.
+  return extendIntegration(redisChannelIntegration({ responseHook: cacheResponseHook }), {
     name: INTEGRATION_NAME,
     setupOnce() {
       setRedisOptions(options);
       instrumentRedis();
     },
-  };
+  });
 }) satisfies IntegrationFn;
 
 /**

packages/node/test/integrations/tracing/redis-ioredis-gating.test.ts

@@ -20,12 +20,6 @@ vi.mock('@sentry/node-core', async importOriginal => {
   };
 });
 
-// The >=5.11.0 diagnostics_channel subscription is irrelevant here; keep it inert.
-vi.mock('@sentry/server-utils', async importOriginal => {
-  const actual = (await importOriginal()) as Record<string, unknown>;
-  return { ...actual, subscribeRedisDiagnosticChannels: () => undefined };
-});
-
 import { instrumentRedis } from '../../../src/integrations/tracing/redis';
 
 describe('instrumentRedis ioredis gating', () => {

packages/server-utils/src/index.ts

@@ -5,22 +5,8 @@
  */
 
 export { mongooseIntegration } from './mongoose';
-export {
-  IOREDIS_DC_CHANNEL_COMMAND,
-  IOREDIS_DC_CHANNEL_CONNECT,
-  REDIS_DC_CHANNEL_BATCH,
-  REDIS_DC_CHANNEL_COMMAND,
-  REDIS_DC_CHANNEL_CONNECT,
-  subscribeRedisDiagnosticChannels,
-} from './redis/redis-dc-subscriber';
-export type {
-  IORedisCommandData,
-  RedisBatchData,
-  RedisCommandData,
-  RedisConnectData,
-  RedisDiagnosticChannelResponseHook,
-  RedisTracingChannelFactory,
-} from './redis/redis-dc-subscriber';
+export { redisIntegration, type RedisDiagnosticChannelsOptions } from './redis';
+export type { RedisDiagnosticChannelResponseHook } from './redis/redis-dc-subscriber';
 export { defaultDbStatementSerializer } from './redis/redis-statement-serializer';
 export { bindTracingChannelToSpan } from './tracing-channel';
 export type {

packages/server-utils/src/redis/index.ts

@@ -0,0 +1,35 @@
+import { defineIntegration, type IntegrationFn, waitForTracingChannelBinding } from '@sentry/core';
+import * as dc from 'node:diagnostics_channel';
+import { type RedisDiagnosticChannelResponseHook, subscribeRedisDiagnosticChannels } from './redis-dc-subscriber';
+
+/** Options controlling the redis diagnostics-channel subscription. */
+export interface RedisDiagnosticChannelsOptions {
+  /**
+   * Optional hook invoked once the redis command response arrives. Useful for attaching
+   * response-derived attributes (e.g. cache hit/miss, payload size).
+   */
+  responseHook?: RedisDiagnosticChannelResponseHook;
+}
+
+const _redisIntegration = ((options: RedisDiagnosticChannelsOptions = {}) => {
+  return {
+    name: 'Redis',
+    setupOnce() {
+      // Bail on runtimes without `tracingChannel` (Node <= 18.18.0).
+      if (!dc.tracingChannel) {
+        return;
+      }
+
+      waitForTracingChannelBinding(() => {
+        subscribeRedisDiagnosticChannels(dc.tracingChannel, options.responseHook);
+      });
+    },
+  };
+}) satisfies IntegrationFn;
+
+/**
+ * Auto-instrument the [redis](https://www.npmjs.com/package/redis) and
+ * [ioredis](https://www.npmjs.com/package/ioredis) libraries via their native
+ * `node:diagnostics_channel` tracing channels (node-redis >= 5.12.0, ioredis >= 5.11.0).
+ */
+export const redisIntegration = defineIntegration(_redisIntegration);

packages/server-utils/src/redis/redis-dc-subscriber.ts

@@ -7,8 +7,7 @@ import {
   SERVER_PORT,
 } from '@sentry/conventions/attributes';
 import type { Span } from '@sentry/core';
-import { debug, SEMANTIC_ATTRIBUTE_SENTRY_OP, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, startInactiveSpan } from '@sentry/core';
-import { DEBUG_BUILD } from '../debug-build';
+import { SEMANTIC_ATTRIBUTE_SENTRY_OP, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, startInactiveSpan } from '@sentry/core';
 import { bindTracingChannelToSpan } from '../tracing-channel';
 
 // Channel names published by node-redis >= 5.12.0 and ioredis >= 5.11.0.
@@ -106,10 +105,6 @@ export type RedisDiagnosticChannelResponseHook = (
  */
 export type RedisTracingChannelFactory = <T extends object>(name: string) => TracingChannel<T, T>;
 
-let subscribed = false;
-let currentResponseHook: RedisDiagnosticChannelResponseHook | undefined;
-let activeUnbinds: Array<() => void> = [];
-
 /**
  * Subscribe Sentry span handlers to node-redis and ioredis diagnostics-channel
  * events: `node-redis:command`/`:batch`/`:connect` (published by node-redis
@@ -118,46 +113,37 @@ let activeUnbinds: Array<() => void> = [];
  * On older client versions the channels are never published to, so subscribers
  * are inert — there is no double-instrumentation against any IITM-based
  * patcher gated to those older versions.
- *
- * Idempotent: subsequent calls update the response hook but do not
- * re-subscribe.
  */
 export function subscribeRedisDiagnosticChannels(
   tracingChannel: RedisTracingChannelFactory,
   responseHook?: RedisDiagnosticChannelResponseHook,
 ): void {
-  currentResponseHook = responseHook;
-  if (subscribed) return;
-  subscribed = true;
-
-  try {
-    // node-redis: command name appears as args[0] in the channel payload, so
-    // strip it before the statement and response hook see it.
-    activeUnbinds.push(
-      setupCommandChannel<RedisCommandData>(tracingChannel, REDIS_DC_CHANNEL_COMMAND, data => data.args.slice(1)),
-      setupBatchChannel(tracingChannel, REDIS_DC_CHANNEL_BATCH, data =>
-        data.batchMode === 'PIPELINE' ? 'PIPELINE' : 'MULTI',
-      ),
-      setupConnectChannel(tracingChannel, REDIS_DC_CHANNEL_CONNECT),
-      // ioredis: args already exclude the command name; no slicing needed. And
-      // ioredis has no separate batch channel — pipeline/MULTI metadata rides
-      // on the per-command payload via `batchMode`/`batchSize`.
-      setupCommandChannel<IORedisCommandData>(tracingChannel, IOREDIS_DC_CHANNEL_COMMAND, data => data.args),
-      setupConnectChannel(tracingChannel, IOREDIS_DC_CHANNEL_CONNECT),
-    );
-  } catch {
-    // The factory may rely on `node:diagnostics_channel`, which isn't always
-    // available. Fail closed; the SDK simply won't emit redis spans here.
-    DEBUG_BUILD && debug.log('Redis node:diagnostics_channel subscription failed.');
-  }
+  // node-redis: command name appears as args[0] in the channel payload, so
+  // strip it before the statement and response hook see it.
+  setupCommandChannel<RedisCommandData>(
+    tracingChannel,
+    REDIS_DC_CHANNEL_COMMAND,
+    data => data.args.slice(1),
+    responseHook,
+  );
+  setupBatchChannel(tracingChannel, REDIS_DC_CHANNEL_BATCH, data =>
+    data.batchMode === 'PIPELINE' ? 'PIPELINE' : 'MULTI',
+  );
+  setupConnectChannel(tracingChannel, REDIS_DC_CHANNEL_CONNECT);
+  // ioredis: args already exclude the command name; no slicing needed. And
+  // ioredis has no separate batch channel — pipeline/MULTI metadata rides
+  // on the per-command payload via `batchMode`/`batchSize`.
+  setupCommandChannel<IORedisCommandData>(tracingChannel, IOREDIS_DC_CHANNEL_COMMAND, data => data.args, responseHook);
+  setupConnectChannel(tracingChannel, IOREDIS_DC_CHANNEL_CONNECT);
 }
 
 function setupCommandChannel<T extends RedisCommandData | IORedisCommandData>(
   tracingChannel: RedisTracingChannelFactory,
   channelName: string,
   getCommandArgs: (data: T) => string[],
-): () => void {
-  return bindTracingChannelToSpan(
+  responseHook?: RedisDiagnosticChannelResponseHook,
+): void {
+  bindTracingChannelToSpan(
     tracingChannel<T>(channelName),
     data => {
       // `args` is already sanitized by the publishing library (node-redis /
@@ -183,18 +169,18 @@ function setupCommandChannel<T extends RedisCommandData | IORedisCommandData>(
       captureError: false,
       beforeSpanEnd(span, data) {
         if ('error' in data) return;
-        runResponseHook(span, data.command, getCommandArgs(data), data.result);
+        runResponseHook(responseHook, span, data.command, getCommandArgs(data), data.result);
       },
     },
-  ).unbind;
+  );
 }
 
 function setupBatchChannel(
   tracingChannel: RedisTracingChannelFactory,
   channelName: string,
   getOperationName: (data: RedisBatchData) => string,
-): () => void {
-  return bindTracingChannelToSpan(
+): void {
+  bindTracingChannelToSpan(
     tracingChannel<RedisBatchData>(channelName),
     data => {
       return startInactiveSpan({
@@ -212,11 +198,11 @@ function setupBatchChannel(
       });
     },
     { captureError: false },
-  ).unbind;
+  );
 }
 
-function setupConnectChannel(tracingChannel: RedisTracingChannelFactory, channelName: string): () => void {
-  return bindTracingChannelToSpan(
+function setupConnectChannel(tracingChannel: RedisTracingChannelFactory, channelName: string): void {
+  bindTracingChannelToSpan(
     tracingChannel<RedisConnectData>(channelName),
     data => {
       return startInactiveSpan({
@@ -231,23 +217,20 @@ function setupConnectChannel(tracingChannel: RedisTracingChannelFactory, channel
       });
     },
     { captureError: false },
-  ).unbind;
+  );
 }
 
-function runResponseHook(span: Span, command: string, args: string[], result: unknown): void {
-  const hook = currentResponseHook;
+function runResponseHook(
+  hook: RedisDiagnosticChannelResponseHook | undefined,
+  span: Span,
+  command: string,
+  args: string[],
+  result: unknown,
+): void {
   if (!hook) return;
   try {
     hook(span, command, args, result);
   } catch {
     // never let user hooks break instrumentation
   }
 }
-
-/** Test-only: detach all channel bindings and reset module-local subscribe state. */
-export function _resetRedisDiagnosticChannelsForTesting(): void {
-  activeUnbinds.forEach(unbind => unbind());
-  activeUnbinds = [];
-  subscribed = false;
-  currentResponseHook = undefined;
-}

packages/server-utils/test/redis/redis-dc-subscriber.test.ts

@@ -17,9 +17,8 @@ import {
   spanToJSON,
   startSpan,
 } from '@sentry/core';
-import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
 import {
-  _resetRedisDiagnosticChannelsForTesting,
   IOREDIS_DC_CHANNEL_COMMAND,
   IOREDIS_DC_CHANNEL_CONNECT,
   REDIS_DC_CHANNEL_BATCH,
@@ -127,23 +126,31 @@ async function traceCommand(
 
 const factory = tracingChannel as RedisTracingChannelFactory;
 
+// The response hook is bound into the channel handlers when we subscribe, so it must be a stable
+// reference for the whole file; `vi.clearAllMocks` in `afterEach` resets its recorded calls per test.
+const responseHook = vi.fn();
+
 describe('subscribeRedisDiagnosticChannels', () => {
-  let responseHook: ReturnType<typeof vi.fn>;
   let captureExceptionSpy: ReturnType<typeof vi.spyOn>;
 
-  // `node:diagnostics_channel` channels are process-global. `_reset…` calls each binding's `unbind`,
-  // so we can subscribe and fully detach per test without handlers leaking across tests.
-  beforeEach(() => {
+  // The subscriber binds handlers onto process-global channels with no teardown, so subscribe once
+  // here, mirroring production where `setupOnce` subscribes a single time. Per-test we only reset the
+  // client, scopes, and mock call history (in `afterEach`), so nothing leaks between tests.
+  beforeAll(() => {
     installTestAsyncContextStrategy();
+    subscribeRedisDiagnosticChannels(factory, responseHook);
+  });
+
+  afterAll(() => {
+    setAsyncContextStrategy(undefined);
+  });
+
+  beforeEach(() => {
     initTestClient();
-    responseHook = vi.fn();
     captureExceptionSpy = vi.spyOn(SentryCore, 'captureException').mockReturnValue('event-id');
-    subscribeRedisDiagnosticChannels(factory, responseHook);
   });
 
   afterEach(() => {
-    _resetRedisDiagnosticChannelsForTesting();
-    setAsyncContextStrategy(undefined);
     getCurrentScope().clear();
     getCurrentScope().setClient(undefined);
     getGlobalScope().clear();
@@ -261,21 +268,4 @@ describe('subscribeRedisDiagnosticChannels', () => {
       expect(spanToJSON(span!).timestamp).toBeDefined();
     });
   });
-
-  describe('idempotency', () => {
-    it('does not re-subscribe on a second call, but updates the response hook', async () => {
-      const secondHook = vi.fn();
-      subscribeRedisDiagnosticChannels(factory, secondHook);
-
-      const { span } = await traceCommand(
-        REDIS_DC_CHANNEL_COMMAND,
-        { command: 'GET', args: ['GET', 'k'] },
-        { result: 'v' },
-      );
-
-      expect(secondHook).toHaveBeenCalledTimes(1);
-      expect(secondHook).toHaveBeenCalledWith(span, 'GET', ['k'], 'v');
-      expect(responseHook).not.toHaveBeenCalled();
-    });
-  });
 });