diff --git a/dev-packages/e2e-tests/test-applications/nestjs-bullmq/src/app.controller.ts b/dev-packages/e2e-tests/test-applications/nestjs-bullmq/src/app.controller.ts index fcdb44542aae..0b3fa08d5695 100644 --- a/dev-packages/e2e-tests/test-applications/nestjs-bullmq/src/app.controller.ts +++ b/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') diff --git a/dev-packages/e2e-tests/test-applications/nestjs-bullmq/src/app.module.ts b/dev-packages/e2e-tests/test-applications/nestjs-bullmq/src/app.module.ts index be5fd107e4cb..80ba330e800a 100644 --- a/dev-packages/e2e-tests/test-applications/nestjs-bullmq/src/app.module.ts +++ b/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, diff --git a/dev-packages/e2e-tests/test-applications/nestjs-bullmq/src/jobs/test-queue-events.listener.ts b/dev-packages/e2e-tests/test-applications/nestjs-bullmq/src/jobs/test-queue-events.listener.ts new file mode 100644 index 000000000000..75bcdd6a41e0 --- /dev/null +++ b/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 }, + }); + } +} diff --git a/dev-packages/e2e-tests/test-applications/nestjs-bullmq/src/jobs/test.processor.ts b/dev-packages/e2e-tests/test-applications/nestjs-bullmq/src/jobs/test.processor.ts index 1c6cf8ef3052..2bf3fa1a903d 100644 --- a/dev-packages/e2e-tests/test-applications/nestjs-bullmq/src/jobs/test.processor.ts +++ b/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 { + 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', diff --git a/dev-packages/e2e-tests/test-applications/nestjs-bullmq/tests/bullmq.test.ts b/dev-packages/e2e-tests/test-applications/nestjs-bullmq/tests/bullmq.test.ts index 36b184dc5e23..3debc82061be 100644 --- a/dev-packages/e2e-tests/test-applications/nestjs-bullmq/tests/bullmq.test.ts +++ b/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([]); +}); diff --git a/packages/nestjs/src/integrations/orchestrion-subscriber.ts b/packages/nestjs/src/integrations/orchestrion-subscriber.ts index 53e21dda735f..b327fa7534f9 100644 --- a/packages/nestjs/src/integrations/orchestrion-subscriber.ts +++ b/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(['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)), + ); } diff --git a/packages/nestjs/src/integrations/wrap-handlers.ts b/packages/nestjs/src/integrations/wrap-handlers.ts index 62f677635eeb..5c158a212db6 100644 --- a/packages/nestjs/src/integrations/wrap-handlers.ts +++ b/packages/nestjs/src/integrations/wrap-handlers.ts @@ -1,10 +1,17 @@ +import type { Scope } from '@sentry/core'; import { captureException, isObjectLike, isThenable, startSpan, withIsolationScope } from '@sentry/core'; import type { AnyFn, ReflectWithMetadata } from './helpers'; -import { getBullMQProcessSpanOptions, getEventSpanOptions, isWrapped, markWrapped } from './helpers'; +import { + copyReflectMetadata, + getBullMQProcessSpanOptions, + getEventSpanOptions, + isWrapped, + markWrapped, +} from './helpers'; /** * Span-emitting / error-capturing logic for the `@Cron`/`@Interval`/ - * `@Timeout` (schedule), `@OnEvent` (event), and `@Processor` (bullmq) + * `@Timeout` (schedule), `@OnEvent` (event), and BullMQ processor and lifecycle * handlers. * * @module @@ -19,6 +26,24 @@ export const MECHANISM_TIMEOUT = 'auto.function.nestjs.timeout'; export const MECHANISM_EVENT = 'auto.event.nestjs'; export const MECHANISM_BULLMQ = 'auto.queue.nestjs.bullmq'; +const bullMQJobScopes = new WeakMap(); + +function withBullMQJobScope(job: unknown, callback: () => T): T { + if (!isObjectLike(job)) { + return withIsolationScope(callback); + } + + const existingScope = bullMQJobScopes.get(job); + if (existingScope) { + return withIsolationScope(existingScope, callback); + } + + return withIsolationScope(scope => { + bullMQJobScopes.set(job, scope); + return callback(); + }); +} + const EVENT_LISTENER_METADATA = 'EVENT_LISTENER_METADATA'; interface ProcessorTarget { @@ -31,13 +56,11 @@ function captureHandlerError(error: unknown, mechanismType: string): void { } /** - * 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. + * Preserve synchronous throws so the caller's error handling still applies. */ export function wrapScheduleHandler(handler: AnyFn, mechanismType: string): AnyFn { return function (this: unknown, ...args: unknown[]): unknown { - return withIsolationScope(() => { + const invoke = (): unknown => { let result: unknown; try { result = handler.apply(this, args); @@ -52,7 +75,19 @@ export function wrapScheduleHandler(handler: AnyFn, mechanismType: string): AnyF }); } return result; - }); + }; + + return withIsolationScope(invoke); + }; +} + +/** + * @param shareJobScope Reuse the first argument's (BullMQ Job) scope so lifecycle hooks share context with process(). + */ +export function wrapBullMQEventHandler(handler: AnyFn, shareJobScope: boolean = false): AnyFn { + return function (this: unknown, ...args: unknown[]): unknown { + const invoke = (): unknown => handler.apply(this, args); + return shareJobScope ? withBullMQJobScope(args[0], invoke) : withIsolationScope(invoke); }; } @@ -113,12 +148,12 @@ export function wrapEventHandler(handler: AnyFn, fallbackEvent: unknown): AnyFn } /** - * Wrap a BullMQ `process` method: fork the isolation scope, open a + * Wrap a BullMQ `process` method: reuse the job's isolation scope, open a * `queue.process` transaction, and capture errors. */ export function wrapBullMQProcess(process: AnyFn, queueName: string | undefined): AnyFn { return function (this: unknown, ...args: unknown[]): unknown { - return withIsolationScope(() => + return withBullMQJobScope(args[0], () => startSpan(getBullMQProcessSpanOptions(queueName), async () => { try { return await process.apply(this, args); @@ -145,6 +180,7 @@ export function patchMethodDescriptor( const handler = descriptor?.value as AnyFn | undefined; if (descriptor && handler && typeof handler === 'function' && !target?.__SENTRY_INTERNAL__ && !isWrapped(handler)) { const wrapped = wrapHandler(handler); + copyReflectMetadata(handler, wrapped); Object.defineProperty(wrapped, 'name', { value: handler.name || String(propertyKey), configurable: true, diff --git a/packages/nestjs/test/integrations/orchestrion-subscriber.test.ts b/packages/nestjs/test/integrations/orchestrion-subscriber.test.ts index 37409a4312d1..86d653690e5a 100644 --- a/packages/nestjs/test/integrations/orchestrion-subscriber.test.ts +++ b/packages/nestjs/test/integrations/orchestrion-subscriber.test.ts @@ -992,6 +992,46 @@ describe('NestJS orchestrion subscriber: schedule / event / bullmq', () => { }, ); + it.each([CHANNELS.NESTJS_BULLMQ_WORKER_EVENT, CHANNELS.NESTJS_BULLMQ_QUEUE_EVENT])( + '%s: propagates synchronous listener errors without capturing them', + channel => { + installTestAsyncContextStrategy(); + subscribeToNestChannels(); + const captureSpy = vi.spyOn(SentryCore, 'captureException'); + const error = new Error('listener failed'); + const descriptor: PropertyDescriptor = { + value: () => { + throw error; + }, + }; + const decorator = driveFactory(channel, ['completed'], (_t, _k, d) => d); + decorator({}, 'onCompleted', descriptor); + + expect(() => descriptor.value({})).toThrow(error); + expect(captureSpy).not.toHaveBeenCalled(); + }, + ); + + it.each([CHANNELS.NESTJS_BULLMQ_WORKER_EVENT, CHANNELS.NESTJS_BULLMQ_QUEUE_EVENT])( + '%s: propagates async listener errors without capturing them', + async channel => { + installTestAsyncContextStrategy(); + subscribeToNestChannels(); + const captureSpy = vi.spyOn(SentryCore, 'captureException'); + const error = new Error('async listener failed'); + const descriptor: PropertyDescriptor = { + value: async () => { + throw error; + }, + }; + const decorator = driveFactory(channel, ['completed'], (_t, _k, d) => d); + decorator({}, 'onCompleted', descriptor); + + await expect(descriptor.value({})).rejects.toBe(error); + expect(captureSpy).not.toHaveBeenCalled(); + }, + ); + it('schedule @Timeout: captures sync errors with the timeout mechanism', () => { installTestAsyncContextStrategy(); initTestClient(); diff --git a/packages/server-utils/src/orchestrion/config/nestjs.ts b/packages/server-utils/src/orchestrion/config/nestjs.ts index 344b35913ede..c0795d84d702 100644 --- a/packages/server-utils/src/orchestrion/config/nestjs.ts +++ b/packages/server-utils/src/orchestrion/config/nestjs.ts @@ -128,6 +128,24 @@ export const nestjsConfig = [ }, functionQuery: { functionName: 'Processor', kind: 'Sync' }, }, + { + channelName: 'onWorkerEventDecorator', + module: { + name: '@nestjs/bullmq', + versionRange: '>=10.0.0', + filePath: 'dist/decorators/on-worker-event.decorator.js', + }, + functionQuery: { expressionName: 'OnWorkerEvent', kind: 'Sync' }, + }, + { + channelName: 'onQueueEventDecorator', + module: { + name: '@nestjs/bullmq', + versionRange: '>=10.0.0', + filePath: 'dist/decorators/on-queue-event.decorator.js', + }, + functionQuery: { expressionName: 'OnQueueEvent', kind: 'Sync' }, + }, ] satisfies InstrumentationConfig[]; export const nestjsChannels = { @@ -140,4 +158,6 @@ export const nestjsChannels = { NESTJS_SCHEDULE_TIMEOUT: 'orchestrion:@nestjs/schedule:timeoutDecorator', NESTJS_ONEVENT: 'orchestrion:@nestjs/event-emitter:onEventDecorator', NESTJS_PROCESSOR: 'orchestrion:@nestjs/bullmq:processorDecorator', + NESTJS_BULLMQ_WORKER_EVENT: 'orchestrion:@nestjs/bullmq:onWorkerEventDecorator', + NESTJS_BULLMQ_QUEUE_EVENT: 'orchestrion:@nestjs/bullmq:onQueueEventDecorator', } as const;