fix(nestjs): Isolate BullMQ lifecycle handlers

dev-packages/e2e-tests/test-applications/nestjs-bullmq/src/app.controller.ts

@@ -9,8 +9,8 @@ export class AppController {
 
   @Get('enqueue/:name')
   async enqueue(@Param('name') name: string) {
-    await this.queue.add(name, { timestamp: Date.now() });
-    return { queued: true };
+    const job = await this.queue.add(name, { timestamp: Date.now() });
+    return { queued: true, jobId: job.id };
   }
 
   @Get('check-isolation')

dev-packages/e2e-tests/test-applications/nestjs-bullmq/src/app.module.ts

@@ -4,6 +4,7 @@ import { BullModule } from '@nestjs/bullmq';
 import { SentryGlobalFilter, SentryModule } from '@sentry/nestjs/setup';
 import { AppController } from './app.controller';
 import { TestProcessor } from './jobs/test.processor';
+import { TestQueueEventsListener } from './jobs/test-queue-events.listener';
 
 @Module({
   imports: [
@@ -16,6 +17,7 @@ import { TestProcessor } from './jobs/test.processor';
   controllers: [AppController],
   providers: [
     TestProcessor,
+    TestQueueEventsListener,
     {
       provide: APP_FILTER,
       useClass: SentryGlobalFilter,

dev-packages/e2e-tests/test-applications/nestjs-bullmq/src/jobs/test-queue-events.listener.ts

@@ -0,0 +1,29 @@
+import { OnQueueEvent, QueueEventsHost, QueueEventsListener } from '@nestjs/bullmq';
+import * as Sentry from '@sentry/nestjs';
+
+@QueueEventsListener('test-queue', { lastEventId: '0-0' })
+export class TestQueueEventsListener extends QueueEventsHost {
+  @OnQueueEvent('completed')
+  async onCompleted({ jobId, returnvalue }: { jobId: string; returnvalue: { scenario?: string } }) {
+    if (returnvalue?.scenario !== 'queue-event-isolation') {
+      return;
+    }
+
+    const scopeData = Sentry.getIsolationScope().getScopeData();
+    const previousJobId = scopeData.tags['queue_event.job_id'] ?? null;
+    const previousBreadcrumbs = scopeData.breadcrumbs
+      .filter(breadcrumb => breadcrumb.category === 'bullmq.queue_event')
+      .map(breadcrumb => breadcrumb.message);
+
+    Sentry.setTag('queue_event.job_id', jobId);
+    Sentry.addBreadcrumb({
+      category: 'bullmq.queue_event',
+      message: `leaked-breadcrumb-from-queue-event-${jobId}`,
+    });
+    await Promise.resolve();
+
+    Sentry.captureMessage(`BullMQ queue lifecycle callback ${jobId}`, {
+      extra: { previousJobId, previousBreadcrumbs },
+    });
+  }
+}

dev-packages/e2e-tests/test-applications/nestjs-bullmq/src/jobs/test.processor.ts

@@ -5,6 +5,18 @@ import * as Sentry from '@sentry/nestjs';
 @Processor('test-queue')
 export class TestProcessor extends WorkerHost {
   async process(job: Job): Promise<any> {
+    if (job.name === 'shared-lifecycle-context') {
+      Sentry.addBreadcrumb({ category: 'bullmq.lifecycle', message: 'process' });
+      await job.updateProgress(50);
+      Sentry.addBreadcrumb({ category: 'bullmq.lifecycle', message: 'process-after-progress' });
+      Sentry.captureMessage('BullMQ shared lifecycle process');
+      return { processed: true };
+    }
+
+    if (job.name === 'queue-event-isolation') {
+      return { scenario: 'queue-event-isolation' };
+    }
+
     if (job.name === 'fail') {
       throw new Error('Test error from BullMQ processor');
     }
@@ -30,22 +42,36 @@ export class TestProcessor extends WorkerHost {
   }
 
   @OnWorkerEvent('completed')
-  onCompleted(job: Job) {
+  async onCompleted(job: Job) {
+    if (job.name === 'shared-lifecycle-context') {
+      await Promise.resolve();
+      Sentry.addBreadcrumb({ category: 'bullmq.lifecycle', message: 'completed' });
+      Sentry.captureMessage('BullMQ shared lifecycle completed');
+    }
+
     if (job.name === 'lifecycle-breadcrumb-test') {
       Sentry.addBreadcrumb({
         message: 'leaked-breadcrumb-from-lifecycle-event',
         level: 'info',
       });
+      Sentry.captureMessage('BullMQ completed lifecycle callback');
     }
   }
 
   @OnWorkerEvent('active')
   onActive(job: Job) {
+    if (job.name === 'shared-lifecycle-context') {
+      Sentry.setTag('lifecycle.job_id', job.id);
+      Sentry.setUser({ id: job.id });
+      Sentry.addBreadcrumb({ category: 'bullmq.lifecycle', message: 'active' });
+    }
+
     if (job.name === 'lifecycle-active-breadcrumb-test') {
       Sentry.addBreadcrumb({
         message: 'leaked-breadcrumb-from-active-event',
         level: 'info',
       });
+      Sentry.captureMessage('BullMQ active lifecycle callback');
     }
   }
 
@@ -56,11 +82,17 @@ export class TestProcessor extends WorkerHost {
         message: 'leaked-breadcrumb-from-failed-event',
         level: 'info',
       });
+      Sentry.captureMessage('BullMQ failed lifecycle callback');
     }
   }
 
   @OnWorkerEvent('progress')
   onProgress(job: Job) {
+    if (job.name === 'shared-lifecycle-context') {
+      Sentry.setTag('lifecycle.progress', 'received');
+      Sentry.addBreadcrumb({ category: 'bullmq.lifecycle', message: 'progress' });
+    }
+
     if (job.name === 'lifecycle-progress-breadcrumb-test') {
       Sentry.addBreadcrumb({
         message: 'leaked-breadcrumb-from-progress-event',

dev-packages/e2e-tests/test-applications/nestjs-bullmq/tests/bullmq.test.ts

@@ -75,56 +75,42 @@ test('BullMQ processor breadcrumbs do not leak into subsequent HTTP requests', a
   expect(await getLeakedBreadcrumbs(baseURL!)).not.toContain('leaked-breadcrumb-from-bullmq-processor');
 });
 
-// TODO: @OnWorkerEvent('completed') handlers run outside the isolation scope created by process().
-// They are registered via worker.on() (EventEmitter), so breadcrumbs/tags set there
-// leak into the default isolation scope and appear on subsequent HTTP requests.
-test('BullMQ @OnWorkerEvent completed lifecycle breadcrumbs currently leak into subsequent HTTP requests', async ({
+test('BullMQ @OnWorkerEvent completed lifecycle breadcrumbs do not leak into subsequent HTTP requests', async ({
   baseURL,
 }) => {
-  const processSpanPromise = waitForProcessSpan();
+  const callbackPromise = waitForError(APP_NAME, event => event.message === 'BullMQ completed lifecycle callback');
 
-  // Enqueue a job (the completed event fires right after the job is processed)
   await fetch(`${baseURL}/enqueue/lifecycle-breadcrumb-test`);
 
-  await processSpanPromise;
+  await callbackPromise;
 
-  // This SHOULD be not.toContain() once lifecycle event isolation is implemented.
-  expect(await getLeakedBreadcrumbs(baseURL!)).toContain('leaked-breadcrumb-from-lifecycle-event');
+  expect(await getLeakedBreadcrumbs(baseURL!)).not.toContain('leaked-breadcrumb-from-lifecycle-event');
 });
 
-// TODO: @OnWorkerEvent('active') handlers run outside the isolation scope created by process().
-// Breadcrumbs set there leak into the default isolation scope and appear on subsequent HTTP requests.
-test('BullMQ @OnWorkerEvent active lifecycle breadcrumbs currently leak into subsequent HTTP requests', async ({
+test('BullMQ @OnWorkerEvent active lifecycle breadcrumbs do not leak into subsequent HTTP requests', async ({
   baseURL,
 }) => {
-  const processSpanPromise = waitForProcessSpan();
+  const callbackPromise = waitForError(APP_NAME, event => event.message === 'BullMQ active lifecycle callback');
 
   await fetch(`${baseURL}/enqueue/lifecycle-active-breadcrumb-test`);
 
-  await processSpanPromise;
+  await callbackPromise;
 
-  // This SHOULD be not.toContain() once lifecycle event isolation is implemented.
-  expect(await getLeakedBreadcrumbs(baseURL!)).toContain('leaked-breadcrumb-from-active-event');
+  expect(await getLeakedBreadcrumbs(baseURL!)).not.toContain('leaked-breadcrumb-from-active-event');
 });
 
-// TODO: @OnWorkerEvent('failed') handlers run outside the isolation scope created by process().
-// Breadcrumbs set there leak into the default isolation scope and appear on subsequent HTTP requests.
-test('BullMQ @OnWorkerEvent failed lifecycle breadcrumbs currently leak into subsequent HTTP requests', async ({
+test('BullMQ @OnWorkerEvent failed lifecycle breadcrumbs do not leak into subsequent HTTP requests', async ({
   baseURL,
 }) => {
-  const processSpanPromise = waitForProcessSpan();
+  const callbackPromise = waitForError(APP_NAME, event => event.message === 'BullMQ failed lifecycle callback');
 
   await fetch(`${baseURL}/enqueue/lifecycle-failed-breadcrumb-test`);
 
-  await processSpanPromise;
+  await callbackPromise;
 
-  // This SHOULD be not.toContain() once lifecycle event isolation is implemented.
-  expect(await getLeakedBreadcrumbs(baseURL!)).toContain('leaked-breadcrumb-from-failed-event');
+  expect(await getLeakedBreadcrumbs(baseURL!)).not.toContain('leaked-breadcrumb-from-failed-event');
 });
 
-// The 'progress' event does NOT leak breadcrumbs — unlike 'active', 'completed', and 'failed',
-// BullMQ emits it inside the process() call (via job.updateProgress()), so it runs within
-// the isolation scope already established by the instrumentation.
 test('BullMQ @OnWorkerEvent progress lifecycle breadcrumbs do not leak into subsequent HTTP requests', async ({
   baseURL,
 }) => {
@@ -136,3 +122,66 @@ test('BullMQ @OnWorkerEvent progress lifecycle breadcrumbs do not leak into subs
 
   expect(await getLeakedBreadcrumbs(baseURL!)).not.toContain('leaked-breadcrumb-from-progress-event');
 });
+
+test('Shares job context across worker lifecycle callbacks and async boundaries', async ({ baseURL }) => {
+  const processEventPromise = waitForError(APP_NAME, event => event.message === 'BullMQ shared lifecycle process');
+  const completedEventPromise = waitForError(APP_NAME, event => event.message === 'BullMQ shared lifecycle completed');
+
+  const response = await fetch(`${baseURL}/enqueue/shared-lifecycle-context`);
+  const { jobId } = await response.json();
+  const [processEvent, completedEvent] = await Promise.all([processEventPromise, completedEventPromise]);
+
+  expect([processEvent.tags?.['lifecycle.job_id'], completedEvent.tags?.['lifecycle.job_id']]).toEqual([jobId, jobId]);
+  expect([processEvent.tags?.['lifecycle.progress'], completedEvent.tags?.['lifecycle.progress']]).toEqual([
+    'received',
+    'received',
+  ]);
+  expect([processEvent.user?.id, completedEvent.user?.id]).toEqual([jobId, jobId]);
+  expect(
+    processEvent.breadcrumbs
+      ?.filter(breadcrumb => breadcrumb.category === 'bullmq.lifecycle')
+      .map(breadcrumb => breadcrumb.message),
+  ).toEqual(['active', 'process', 'progress', 'process-after-progress']);
+  expect(
+    completedEvent.breadcrumbs
+      ?.filter(breadcrumb => breadcrumb.category === 'bullmq.lifecycle')
+      .map(breadcrumb => breadcrumb.message),
+  ).toEqual(['active', 'process', 'progress', 'process-after-progress', 'completed']);
+});
+
+test('Isolates async queue event listeners from other notifications and HTTP requests', async ({ baseURL }) => {
+  const firstEventPromise = waitForError(APP_NAME, event =>
+    event.message?.startsWith('BullMQ queue lifecycle callback '),
+  );
+
+  const firstResponse = await fetch(`${baseURL}/enqueue/queue-event-isolation`);
+  const { jobId: firstJobId } = await firstResponse.json();
+  const firstEvent = await firstEventPromise;
+
+  const secondEventPromise = waitForError(APP_NAME, event =>
+    event.message?.startsWith('BullMQ queue lifecycle callback '),
+  );
+  const secondResponse = await fetch(`${baseURL}/enqueue/queue-event-isolation`);
+  const { jobId: secondJobId } = await secondResponse.json();
+  const secondEvent = await secondEventPromise;
+
+  expect([firstEvent.tags?.['queue_event.job_id'], secondEvent.tags?.['queue_event.job_id']]).toEqual([
+    firstJobId,
+    secondJobId,
+  ]);
+  expect([firstEvent.extra, secondEvent.extra]).toEqual([
+    { previousJobId: null, previousBreadcrumbs: [] },
+    { previousJobId: null, previousBreadcrumbs: [] },
+  ]);
+  expect(
+    firstEvent.breadcrumbs
+      ?.filter(breadcrumb => breadcrumb.category === 'bullmq.queue_event')
+      .map(breadcrumb => breadcrumb.message),
+  ).toEqual([`leaked-breadcrumb-from-queue-event-${firstJobId}`]);
+  expect(
+    secondEvent.breadcrumbs
+      ?.filter(breadcrumb => breadcrumb.category === 'bullmq.queue_event')
+      .map(breadcrumb => breadcrumb.message),
+  ).toEqual([`leaked-breadcrumb-from-queue-event-${secondJobId}`]);
+  expect(await getLeakedBreadcrumbs(baseURL!)).toEqual([]);
+});

packages/nestjs/src/integrations/orchestrion-subscriber.ts

@@ -15,12 +15,14 @@ import {
   MECHANISM_TIMEOUT,
   patchMethodDescriptor,
   patchProcessorTarget,
+  wrapBullMQEventHandler,
   wrapEventHandler,
   wrapScheduleHandler,
 } from './wrap-handlers';
 import { getAppCreationSpanOptions, wrapRequestContextHandler, wrapRouteHandler } from './wrap-route';
 
 const NOOP = (): void => {};
+const JOB_SCOPED_WORKER_EVENTS = new Set<unknown>(['active', 'progress', 'completed', 'failed']);
 
 /**
  * Subscribe to a decorator channel (`Injectable`/`Catch`).
@@ -199,4 +201,12 @@ export function subscribeToNestChannels(): void {
   subscribeFactoryDecorator(CHANNELS.NESTJS_PROCESSOR, (decorator, data) =>
     makeProcessorDecorator(decorator, extractQueueName(data.arguments?.[0])),
   );
+  subscribeFactoryDecorator(CHANNELS.NESTJS_BULLMQ_WORKER_EVENT, (decorator, data) =>
+    makeMethodDecorator(decorator, handler =>
+      wrapBullMQEventHandler(handler, JOB_SCOPED_WORKER_EVENTS.has(data.arguments?.[0])),
+    ),
+  );
+  subscribeFactoryDecorator(CHANNELS.NESTJS_BULLMQ_QUEUE_EVENT, decorator =>
+    makeMethodDecorator(decorator, handler => wrapBullMQEventHandler(handler)),
+  );
 }