feat(nest): final (still dormant) Orchestrion instrumentation

isaacs · isaacs · commit d314ba38d08c · 2026-07-02T16:17:56.000-07:00
Connect the `@Cron`/`@interval`/`@Timeout` (schedule), `@OnEvent`
(event), and `@Processor` (bullmq) instrumentations in the orchestrion
implementation.

At this point, it's still not wired up by default into the SDK, but all
of the functionality is there.

Next step is the final wire-up and opt-in to swap out the OTel NestJS
for this Orchestrion implementation.
diff --git a/packages/server-utils/src/integrations/tracing-channel/nestjs-decorators.ts b/packages/server-utils/src/integrations/tracing-channel/nestjs-decorators.ts
@@ -10,11 +10,7 @@ import {
   startSpanManual,
   withActiveSpan,
 } from '@sentry/core';
-
-/**
- * A function of unknown signature.
- */
-export type AnyFn = (this: unknown, ...args: unknown[]) => unknown;
+import type { AnyFn } from './nestjs-shared';
 
 const OP_MIDDLEWARE = 'middleware.nestjs';
 const ORIGIN_MIDDLEWARE = 'auto.middleware.nestjs';
diff --git a/packages/server-utils/src/integrations/tracing-channel/nestjs-handler-wrappers.ts b/packages/server-utils/src/integrations/tracing-channel/nestjs-handler-wrappers.ts
@@ -0,0 +1,264 @@
+import * as diagnosticsChannel from 'node:diagnostics_channel';
+import {
+  captureException,
+  isThenable,
+  SEMANTIC_ATTRIBUTE_SENTRY_OP,
+  SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN,
+  startSpan,
+  withIsolationScope,
+} from '@sentry/core';
+import { CHANNELS } from '../../orchestrion/channels';
+import type { AnyFn, ChannelContext } from './nestjs-shared';
+import { isWrapped, markWrapped } from './nestjs-shared';
+
+const NOOP = (): void => {};
+
+// Mechanism types for scheduled-handler error capture (no span)
+// match vendored `SentryNestScheduleInstrumentation`
+const MECHANISM_CRON = 'auto.function.nestjs.cron';
+const MECHANISM_INTERVAL = 'auto.function.nestjs.interval';
+const MECHANISM_TIMEOUT = 'auto.function.nestjs.timeout';
+const MECHANISM_EVENT = 'auto.event.nestjs';
+const MECHANISM_BULLMQ = 'auto.queue.nestjs.bullmq';
+
+const EVENT_LISTENER_METADATA = 'EVENT_LISTENER_METADATA';
+
+interface ReflectWithMetadata {
+  getMetadataKeys?: (target: object) => unknown[];
+  getMetadata?: (key: unknown, target: object) => unknown;
+}
+
+/**
+ * The class a `@Processor` decorator is applied to (a BullMQ queue processor). */
+interface ProcessorTarget {
+  __SENTRY_INTERNAL__?: boolean;
+  prototype?: { process?: AnyFn };
+}
+
+function captureHandlerError(error: unknown, mechanismType: string): void {
+  captureException(error, { mechanism: { handled: false, type: mechanismType } });
+}
+
+/**
+ * Wrap a scheduled handler (`@Cron`/`@Interval`/`@Timeout`): fork the
+ * isolation scope and capture errors. NOT async. Preserve the handler's sync
+ * return type, so sync and async errors are handled on separate paths
+ * matches vendored OTel implementation
+ */
+function wrapScheduleHandler(handler: AnyFn, mechanismType: string): AnyFn {
+  return function (this: unknown, ...args: unknown[]): unknown {
+    return withIsolationScope(() => {
+      let result: unknown;
+      try {
+        result = handler.apply(this, args);
+      } catch (error) {
+        captureHandlerError(error, mechanismType);
+        throw error;
+      }
+      if (isThenable(result)) {
+        return result.then(undefined, (error: unknown) => {
+          captureHandlerError(error, mechanismType);
+          throw error;
+        });
+      }
+      return result;
+    });
+  };
+}
+
+function eventNameFromEvent(event: unknown): string {
+  if (typeof event === 'string') {
+    return event;
+  }
+  if (Array.isArray(event)) {
+    return event.map(eventNameFromEvent).join(',');
+  }
+  return String(event);
+}
+
+/**
+ * Derive the event name(s) for an @OnEvent span. The wrapped handler carries
+ * `EVENT_LISTENER_METADATA` (set by the original decorator), which lists every
+ * event when multiple @OnEvent decorators are stacked; fall back to the event
+ * captured from the decorator factory.
+ */
+function deriveEventName(handler: AnyFn, fallbackEvent: unknown): string {
+  const R = Reflect as unknown as ReflectWithMetadata;
+  if (typeof R.getMetadataKeys === 'function' && typeof R.getMetadata === 'function') {
+    if (R.getMetadataKeys(handler)?.includes(EVENT_LISTENER_METADATA)) {
+      const eventData = R.getMetadata(EVENT_LISTENER_METADATA, handler);
+      if (Array.isArray(eventData)) {
+        return (eventData as unknown[])
+          .map(entry => {
+            const event = entry && typeof entry === 'object' ? (entry as { event?: unknown }).event : undefined;
+            return event ? eventNameFromEvent(event) : '';
+          })
+          .reverse() // decorators evaluate bottom to top
+          .join('|');
+      }
+    }
+  }
+  return eventNameFromEvent(fallbackEvent);
+}
+
+/**
+ * Wrap an @OnEvent handler: fork the isolation scope, open an `event.nestjs`
+ * transaction, and capture errors. (event-handler errors bypass the global
+ * filter)
+ */
+function wrapEventHandler(handler: AnyFn, fallbackEvent: unknown): AnyFn {
+  const wrapped = async function (this: unknown, ...args: unknown[]): Promise<unknown> {
+    const eventName = deriveEventName(wrapped, fallbackEvent);
+    return withIsolationScope(() =>
+      startSpan(
+        {
+          name: `event ${eventName}`,
+          attributes: {
+            [SEMANTIC_ATTRIBUTE_SENTRY_OP]: 'event.nestjs',
+            [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: MECHANISM_EVENT,
+          },
+          forceTransaction: true,
+        },
+        async () => {
+          try {
+            return await handler.apply(this, args);
+          } catch (error) {
+            captureHandlerError(error, MECHANISM_EVENT);
+            throw error;
+          }
+        },
+      ),
+    );
+  };
+  return wrapped;
+}
+
+/**
+ * Wrap a BullMQ `process` method: fork the isolation scope, open a
+ * `queue.process` transaction, and capture errors.
+ */
+function wrapBullMQProcess(process: AnyFn, queueName: string): AnyFn {
+  return function (this: unknown, ...args: unknown[]): unknown {
+    return withIsolationScope(() =>
+      startSpan(
+        {
+          name: `${queueName} process`,
+          attributes: {
+            [SEMANTIC_ATTRIBUTE_SENTRY_OP]: 'queue.process',
+            [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: MECHANISM_BULLMQ,
+            'messaging.system': 'bullmq',
+            'messaging.destination.name': queueName,
+          },
+          forceTransaction: true,
+        },
+        async () => {
+          try {
+            return await process.apply(this, args);
+          } catch (error) {
+            captureHandlerError(error, MECHANISM_BULLMQ);
+            throw error;
+          }
+        },
+      ),
+    );
+  };
+}
+
+/**
+ * Wrap a method decorator (the function the factory returns for
+ * `@Cron`/`@Interval`/`@Timeout`/`@OnEvent`) so it replaces
+ * `descriptor.value` with a wrapped handler before delegating to the
+ * original decorator (which then attaches its metadata to our wrapper).
+ */
+function makeMethodDecorator(original: AnyFn, wrapHandler: (handler: AnyFn) => AnyFn): AnyFn {
+  return function (this: unknown, ...args: unknown[]): unknown {
+    const target = args[0] as { __SENTRY_INTERNAL__?: boolean } | undefined;
+    const propertyKey = args[1];
+    const descriptor = args[2] as PropertyDescriptor | undefined;
+    const handler = descriptor?.value;
+    if (handler && typeof handler === 'function' && !target?.__SENTRY_INTERNAL__ && !isWrapped(handler as AnyFn)) {
+      const wrapped = wrapHandler(handler as AnyFn);
+      Object.defineProperty(wrapped, 'name', {
+        value: (handler as AnyFn).name || String(propertyKey),
+        configurable: true,
+      });
+      markWrapped(wrapped);
+      descriptor.value = wrapped;
+    }
+    return original.apply(this, args);
+  };
+}
+
+/**
+ * Wrap the class decorator @Processor returns so it patches
+ * `target.prototype.process` before delegating to the original decorator.
+ */
+function makeProcessorDecorator(original: AnyFn, queueName: string): AnyFn {
+  return function (this: unknown, ...args: unknown[]): unknown {
+    const target = args[0] as ProcessorTarget | undefined;
+    const process = target?.prototype?.process;
+    if (process && typeof process === 'function' && !target?.__SENTRY_INTERNAL__ && !isWrapped(process)) {
+      const wrapped = wrapBullMQProcess(process, queueName);
+      markWrapped(wrapped);
+      target.prototype!.process = wrapped;
+    }
+    return original.apply(this, args);
+  };
+}
+
+function extractQueueName(arg: unknown): string {
+  if (typeof arg === 'string') {
+    return arg;
+  }
+  if (arg && typeof arg === 'object' && 'name' in arg && typeof (arg as { name?: unknown }).name === 'string') {
+    return (arg as { name: string }).name;
+  }
+  return 'unknown';
+}
+
+/**
+ * Subscribe to a decorator-factory channel. `end` reassigns `data.result` (the
+ * decorator the factory returns) with a wrapped version -> `traceSync` returns
+ * whatever `end` leaves there. `wrap` receives the original decorator and the
+ * channel context (for the factory's args, e.g. the BullMQ queue name).
+ */
+function subscribeFactoryDecorator(channelName: string, wrap: (decorator: AnyFn, data: ChannelContext) => AnyFn): void {
+  diagnosticsChannel.tracingChannel<ChannelContext>(channelName).subscribe({
+    start: NOOP,
+    end(data) {
+      const decorator = data.result;
+      if (typeof decorator === 'function' && !isWrapped(decorator as AnyFn)) {
+        const wrapped = wrap(decorator as AnyFn, data);
+        markWrapped(wrapped);
+        data.result = wrapped;
+      }
+    },
+    asyncStart: NOOP,
+    asyncEnd: NOOP,
+    error: NOOP,
+  });
+}
+
+/**
+ * Subscribe the @Cron/@Interval/@Timeout (schedule), @OnEvent (event-emitter)
+ * and @Processor (bullmq) decorator channels. Each `end` handler reassigns the
+ * decorator the factory returns (via `data.result`) with one that wraps the
+ * user handler (schedule/event) or the `process` method (bullmq).
+ */
+export function subscribeNestHandlerDecorators(): void {
+  subscribeFactoryDecorator(CHANNELS.NESTJS_SCHEDULE_CRON, decorator =>
+    makeMethodDecorator(decorator, handler => wrapScheduleHandler(handler, MECHANISM_CRON)),
+  );
+  subscribeFactoryDecorator(CHANNELS.NESTJS_SCHEDULE_INTERVAL, decorator =>
+    makeMethodDecorator(decorator, handler => wrapScheduleHandler(handler, MECHANISM_INTERVAL)),
+  );
+  subscribeFactoryDecorator(CHANNELS.NESTJS_SCHEDULE_TIMEOUT, decorator =>
+    makeMethodDecorator(decorator, handler => wrapScheduleHandler(handler, MECHANISM_TIMEOUT)),
+  );
+  subscribeFactoryDecorator(CHANNELS.NESTJS_ONEVENT, (decorator, data) =>
+    makeMethodDecorator(decorator, handler => wrapEventHandler(handler, data.arguments?.[0])),
+  );
+  subscribeFactoryDecorator(CHANNELS.NESTJS_PROCESSOR, (decorator, data) =>
+    makeProcessorDecorator(decorator, extractQueueName(data.arguments?.[0])),
+  );
+}
diff --git a/packages/server-utils/src/integrations/tracing-channel/nestjs-shared.ts b/packages/server-utils/src/integrations/tracing-channel/nestjs-shared.ts
@@ -0,0 +1,30 @@
+/** A function of unknown signature, matching the methods/handlers we wrap. */
+export type AnyFn = (this: unknown, ...args: unknown[]) => unknown;
+
+/**
+ * The orchestrion tracing-channel context. `arguments` is the live call args
+ * array; `result` is the return value, which an `end` handler may reassign to
+ * substitute it (`traceSync`/`tracePromise` always return `ctx.result`).
+ */
+export interface ChannelContext {
+  arguments: unknown[];
+  moduleVersion?: string;
+  result?: unknown;
+  error?: unknown;
+}
+
+/**
+ * Marks a function as already wrapped so repeated subscriptions (eg a second
+ * `setupOnce`) or multiple decorators on one method don't double-wrap it.
+ */
+const SENTRY_WRAPPED = Symbol.for('sentry.orchestrion.nestjs.wrapped');
+
+/** Whether `fn` has already been wrapped by this integration. */
+export function isWrapped(fn: AnyFn): boolean {
+  return !!(fn as AnyFn & Record<symbol, unknown>)[SENTRY_WRAPPED];
+}
+
+/** Mark `fn` as wrapped (see {@link isWrapped}). */
+export function markWrapped(fn: AnyFn): void {
+  (fn as AnyFn & Record<symbol, unknown>)[SENTRY_WRAPPED] = true;
+}
diff --git a/packages/server-utils/src/integrations/tracing-channel/nestjs.ts b/packages/server-utils/src/integrations/tracing-channel/nestjs.ts
@@ -4,8 +4,11 @@ import { debug, defineIntegration, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, startInacti
 import { DEBUG_BUILD } from '../../debug-build';
 import { CHANNELS } from '../../orchestrion/channels';
 import { bindTracingChannelToSpan } from '../../tracing-channel';
-import type { AnyFn, CatchTarget, InjectableTarget } from './nestjs-decorators';
+import type { CatchTarget, InjectableTarget } from './nestjs-decorators';
 import { patchCatchTarget, patchInjectableTarget } from './nestjs-decorators';
+import { subscribeNestHandlerDecorators } from './nestjs-handler-wrappers';
+import type { AnyFn, ChannelContext } from './nestjs-shared';
+import { isWrapped, markWrapped } from './nestjs-shared';
 
 // NOTE: this uses the same name as the OTel integration by design.
 // When enabled, the OTel 'Nest' integration is omitted from the default set.
@@ -33,30 +36,6 @@ const TYPE_REQUEST_HANDLER = 'handler';
 
 const NOOP = (): void => {};
 
-// Marks a function as already wrapped so repeated subscriptions (e.g. a second
-// `setupOnce`) don't double-wrap a callback or returned handler.
-const SENTRY_WRAPPED = Symbol.for('sentry.orchestrion.nestjs.wrapped');
-
-function isWrapped(fn: AnyFn): boolean {
-  return !!(fn as AnyFn & Record<symbol, unknown>)[SENTRY_WRAPPED];
-}
-
-function markWrapped(fn: AnyFn): void {
-  (fn as AnyFn & Record<symbol, unknown>)[SENTRY_WRAPPED] = true;
-}
-
-/**
- * The orchestrion tracing-channel context. `arguments` is the live call args
- * array; `result` is the return value, which an `end` handler may reassign to
- * substitute it (`traceSync`/`tracePromise` always return `ctx.result`).
- */
-interface ChannelContext {
-  arguments: unknown[];
-  moduleVersion?: string;
-  result?: unknown;
-  error?: unknown;
-}
-
 /** Minimal request shape, across the express/fastify adapters. */
 interface NestRequest {
   route?: { path?: string };
@@ -263,6 +242,9 @@ const _nestjsChannelIntegration = (() => {
         patchInjectableTarget(target, seenInterceptorContexts),
       );
       subscribeDecoratorChannel<CatchTarget>(CHANNELS.NESTJS_CATCH, patchCatchTarget);
+
+      // @Cron/@Interval/@Timeout (schedule), @OnEvent (event), @Processor (bullmq).
+      subscribeNestHandlerDecorators();
     },
   };
 }) satisfies IntegrationFn;
diff --git a/packages/server-utils/src/orchestrion/channels.ts b/packages/server-utils/src/orchestrion/channels.ts
@@ -21,6 +21,11 @@ export const CHANNELS = {
   NESTJS_ROUTER_CONTEXT: 'orchestrion:@nestjs/core:routerExecutionContextCreate',
   NESTJS_INJECTABLE: 'orchestrion:@nestjs/common:injectableDecorator',
   NESTJS_CATCH: 'orchestrion:@nestjs/common:catchDecorator',
+  NESTJS_SCHEDULE_CRON: 'orchestrion:@nestjs/schedule:cronDecorator',
+  NESTJS_SCHEDULE_INTERVAL: 'orchestrion:@nestjs/schedule:intervalDecorator',
+  NESTJS_SCHEDULE_TIMEOUT: 'orchestrion:@nestjs/schedule:timeoutDecorator',
+  NESTJS_ONEVENT: 'orchestrion:@nestjs/event-emitter:onEventDecorator',
+  NESTJS_PROCESSOR: 'orchestrion:@nestjs/bullmq:processorDecorator',
 } as const;
 
 export type ChannelName = (typeof CHANNELS)[keyof typeof CHANNELS];
diff --git a/packages/server-utils/src/orchestrion/config.ts b/packages/server-utils/src/orchestrion/config.ts
diff --git a/packages/server-utils/test/orchestrion/nestjs.test.ts b/packages/server-utils/test/orchestrion/nestjs.test.ts
