feat(node): Capture the original job error for failed BullMQ jobs

dev-packages/node-integration-tests/suites/tracing/bullmq/test.ts

@@ -1,6 +1,6 @@
 import type { Event, SerializedMetric } from '@sentry/core';
 import { SDK_VERSION } from '@sentry/core';
-import { afterAll, expect } from 'vitest';
+import { afterAll, describe, expect } from 'vitest';
 import { EXPECTED_SDK_NAME } from '../../../utils';
 import { cleanupChildProcesses, createEsmAndCjsTests, describeWithDockerCompose } from '../../../utils/runner';
 
@@ -210,96 +210,121 @@ describeWithDockerCompose('bullmq', { workingDirectory: [__dirname] }, () => {
     });
   });
 
-  createEsmAndCjsTests(__dirname, 'scenario-failed-job.mjs', 'instrument.mjs', (createRunner, test) => {
-    test(
-      'captures an error for a failed job and sets the error status on its process span',
-      { timeout: 90_000 },
-      async () => {
-        let errorEvent: Event | undefined;
-        const runner = createRunner().ignore('trace_metric');
-        const consumerSpansPromise = runner.collectStreamedSpansUntilSegment('process test-failed-queue');
+  // BullMQ passes the job error to the telemetry adapter since 6.7.5, so only then does the error have a stack.
+  describe.each([
+    ['bullmq v5', {}, undefined],
+    [
+      'bullmq v6',
+      { bullmq: '^6.7.5' },
+      {
+        frames: expect.arrayContaining([
+          expect.objectContaining({ filename: expect.stringContaining('scenario-failed-job') }),
+        ]),
+      },
+    ],
+  ])('%s', (_version, additionalDependencies, expectedStacktrace) => {
+    createEsmAndCjsTests(
+      __dirname,
+      'scenario-failed-job.mjs',
+      'instrument.mjs',
+      (createRunner, test) => {
+        test(
+          'captures an error for a failed job and sets the error status on its process span',
+          { timeout: 90_000 },
+          async () => {
+            let errorEvent: Event | undefined;
+            const runner = createRunner().ignore('trace_metric');
+            const consumerSpansPromise = runner.collectStreamedSpansUntilSegment('process test-failed-queue');
 
-        await runner
-          .expect({
-            event: event => {
-              errorEvent = event;
-            },
-          })
-          .start()
-          .completed();
+            await runner
+              .expect({
+                event: event => {
+                  errorEvent = event;
+                },
+              })
+              .start()
+              .completed();
 
-        const consumerSpans = await consumerSpansPromise;
-        const processSpan = consumerSpans.find(span => span.is_segment && span.name === 'process test-failed-queue')!;
+            const consumerSpans = await consumerSpansPromise;
+            const processSpan = consumerSpans.find(
+              span => span.is_segment && span.name === 'process test-failed-queue',
+            )!;
 
-        expect(
-          consumerSpans
-            .filter(span => span.attributes['messaging.system']?.value === 'bullmq')
-            .sort((a, b) => a.start_timestamp - b.start_timestamp)
-            .map(span => ({
-              name: span.name,
-              status: span.status,
-              attributes: span.attributes,
-            })),
-        ).toEqual([
-          {
-            name: 'process test-failed-queue',
-            status: 'error',
-            attributes: {
-              'sentry.op': { type: 'string', value: 'queue.process' },
-              'sentry.origin': { type: 'string', value: 'auto.queue.bullmq.consumer' },
-              'messaging.system': { type: 'string', value: 'bullmq' },
-              'messaging.destination.name': { type: 'string', value: 'test-failed-queue' },
-              'messaging.operation.name': { type: 'string', value: 'process' },
-              'messaging.message.id': { type: 'string', value: expect.any(String) },
-              'messaging.message.retry.count': { type: 'integer', value: 0 },
-              'messaging.client.id': { type: 'string', value: expect.any(String) },
-              'messaging.bullmq.job.name': { type: 'string', value: 'failing-job' },
-              'messaging.bullmq.job.processed_timestamp': { type: 'integer', value: expect.any(Number) },
-              'messaging.bullmq.job.attempt_finished_timestamp': { type: 'integer', value: expect.any(Number) },
-              'sentry.environment': { type: 'string', value: 'production' },
-              'sentry.is_localhost': { type: 'boolean', value: false },
-              'sentry.release': { type: 'string', value: '1.0' },
-              'sentry.sdk.name': { type: 'string', value: EXPECTED_SDK_NAME },
-              'sentry.sdk.version': { type: 'string', value: SDK_VERSION },
-              'sentry.segment.id': { type: 'string', value: processSpan.span_id },
-              'sentry.segment.name': { type: 'string', value: 'process test-failed-queue' },
-              'sentry.trace_lifecycle': { type: 'string', value: 'stream' },
-              'sentry.sample_rate': { type: 'integer', value: 1 },
-              'sentry.sdk.integrations': { type: 'array', value: expect.any(Array) },
-              'sentry.segment.name.source': { type: 'string', value: 'custom' },
-              'sentry.status.message': { type: 'string', value: 'internal_error' },
-            },
-          },
-          {
-            name: 'fail test-failed-queue',
-            status: 'ok',
-            attributes: {
-              'sentry.op': { type: 'string', value: 'queue' },
-              'sentry.origin': { type: 'string', value: 'auto.queue.bullmq' },
-              'messaging.system': { type: 'string', value: 'bullmq' },
-              'messaging.destination.name': { type: 'string', value: 'test-failed-queue' },
-              'messaging.operation.name': { type: 'string', value: 'fail' },
-              'messaging.message.id': { type: 'string', value: expect.any(String) },
-              'messaging.bullmq.job.name': { type: 'string', value: 'failing-job' },
-              'sentry.environment': { type: 'string', value: 'production' },
-              'sentry.is_localhost': { type: 'boolean', value: false },
-              'sentry.release': { type: 'string', value: '1.0' },
-              'sentry.sdk.name': { type: 'string', value: EXPECTED_SDK_NAME },
-              'sentry.sdk.version': { type: 'string', value: SDK_VERSION },
-              'sentry.segment.id': { type: 'string', value: processSpan.span_id },
-              'sentry.segment.name': { type: 'string', value: 'process test-failed-queue' },
-              'sentry.trace_lifecycle': { type: 'string', value: 'stream' },
-            },
-          },
-        ]);
-        expect(errorEvent?.exception?.values).toHaveLength(1);
-        expect(errorEvent?.exception?.values?.[0]?.type).toBe('Error');
-        expect(errorEvent?.exception?.values?.[0]?.value).toBe('Job failed on purpose');
-        expect(errorEvent?.exception?.values?.[0]?.mechanism).toEqual({ type: 'auto.queue.bullmq', handled: false });
-        expect(errorEvent?.exception?.values?.[0]?.stacktrace).toBeUndefined();
-        expect(errorEvent?.contexts?.trace?.trace_id).toBe(processSpan.trace_id);
-        expect(errorEvent?.contexts?.trace?.span_id).toBe(processSpan.span_id);
+            expect(
+              consumerSpans
+                .filter(span => span.attributes['messaging.system']?.value === 'bullmq')
+                .sort((a, b) => a.start_timestamp - b.start_timestamp)
+                .map(span => ({
+                  name: span.name,
+                  status: span.status,
+                  attributes: span.attributes,
+                })),
+            ).toEqual([
+              {
+                name: 'process test-failed-queue',
+                status: 'error',
+                attributes: {
+                  'sentry.op': { type: 'string', value: 'queue.process' },
+                  'sentry.origin': { type: 'string', value: 'auto.queue.bullmq.consumer' },
+                  'messaging.system': { type: 'string', value: 'bullmq' },
+                  'messaging.destination.name': { type: 'string', value: 'test-failed-queue' },
+                  'messaging.operation.name': { type: 'string', value: 'process' },
+                  'messaging.message.id': { type: 'string', value: expect.any(String) },
+                  'messaging.message.retry.count': { type: 'integer', value: 0 },
+                  'messaging.client.id': { type: 'string', value: expect.any(String) },
+                  'messaging.bullmq.job.name': { type: 'string', value: 'failing-job' },
+                  'messaging.bullmq.job.processed_timestamp': { type: 'integer', value: expect.any(Number) },
+                  'messaging.bullmq.job.attempt_finished_timestamp': { type: 'integer', value: expect.any(Number) },
+                  'sentry.environment': { type: 'string', value: 'production' },
+                  'sentry.is_localhost': { type: 'boolean', value: false },
+                  'sentry.release': { type: 'string', value: '1.0' },
+                  'sentry.sdk.name': { type: 'string', value: EXPECTED_SDK_NAME },
+                  'sentry.sdk.version': { type: 'string', value: SDK_VERSION },
+                  'sentry.segment.id': { type: 'string', value: processSpan.span_id },
+                  'sentry.segment.name': { type: 'string', value: 'process test-failed-queue' },
+                  'sentry.trace_lifecycle': { type: 'string', value: 'stream' },
+                  'sentry.sample_rate': { type: 'integer', value: 1 },
+                  'sentry.sdk.integrations': { type: 'array', value: expect.any(Array) },
+                  'sentry.segment.name.source': { type: 'string', value: 'custom' },
+                  'sentry.status.message': { type: 'string', value: 'internal_error' },
+                },
+              },
+              {
+                name: 'fail test-failed-queue',
+                status: 'ok',
+                attributes: {
+                  'sentry.op': { type: 'string', value: 'queue' },
+                  'sentry.origin': { type: 'string', value: 'auto.queue.bullmq' },
+                  'messaging.system': { type: 'string', value: 'bullmq' },
+                  'messaging.destination.name': { type: 'string', value: 'test-failed-queue' },
+                  'messaging.operation.name': { type: 'string', value: 'fail' },
+                  'messaging.message.id': { type: 'string', value: expect.any(String) },
+                  'messaging.bullmq.job.name': { type: 'string', value: 'failing-job' },
+                  'sentry.environment': { type: 'string', value: 'production' },
+                  'sentry.is_localhost': { type: 'boolean', value: false },
+                  'sentry.release': { type: 'string', value: '1.0' },
+                  'sentry.sdk.name': { type: 'string', value: EXPECTED_SDK_NAME },
+                  'sentry.sdk.version': { type: 'string', value: SDK_VERSION },
+                  'sentry.segment.id': { type: 'string', value: processSpan.span_id },
+                  'sentry.segment.name': { type: 'string', value: 'process test-failed-queue' },
+                  'sentry.trace_lifecycle': { type: 'string', value: 'stream' },
+                },
+              },
+            ]);
+            expect(errorEvent?.exception?.values).toHaveLength(1);
+            expect(errorEvent?.exception?.values?.[0]?.type).toBe('Error');
+            expect(errorEvent?.exception?.values?.[0]?.value).toBe('Job failed on purpose');
+            expect(errorEvent?.exception?.values?.[0]?.mechanism).toEqual({
+              type: 'auto.queue.bullmq',
+              handled: false,
+            });
+            expect(errorEvent?.exception?.values?.[0]?.stacktrace).toEqual(expectedStacktrace);
+            expect(errorEvent?.contexts?.trace?.trace_id).toBe(processSpan.trace_id);
+            expect(errorEvent?.contexts?.trace?.span_id).toBe(processSpan.span_id);
+          },
+        );
       },
+      { additionalDependencies },
     );
   });
 

packages/node/src/integrations/tracing/bullmq/span.ts

@@ -142,6 +142,7 @@ export function toSentryAttributes(attributes: Record<string, AttributeValue>):
 export class SentryBullMQSpan implements TelemetrySpan {
   private _span: Span;
   private _scope: Scope;
+  private _recordedError: Error | undefined;
 
   public constructor(span: Span, scope: Scope) {
     this._span = span;
@@ -164,11 +165,16 @@ export class SentryBullMQSpan implements TelemetrySpan {
     if (name === 'job failed') {
       this._span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' });
 
-      const reason = attributes?.['bullmq.job.failed.reason'];
-      const error = new Error(String(reason || 'Unknown error'));
-      // BullMQ passes only the error message (https://github.com/taskforcesh/bullmq/issues/4902), so this stack has only
-      // BullMQ and SDK frames, the same for every job. Without it, Sentry groups the errors by their message.
-      error.stack = undefined;
+      // Since 6.7.5, BullMQ passes the job error to `recordException` right before this event
+      // (https://github.com/taskforcesh/bullmq/pull/4915). Older versions pass only the error message.
+      let error = this._recordedError;
+      this._recordedError = undefined;
+      if (!error) {
+        error = new Error(String(attributes?.['bullmq.job.failed.reason'] || 'Unknown error'));
+        // This stack has only BullMQ and SDK frames, the same for every job. Without it, Sentry groups the errors by
+        // their message.
+        error.stack = undefined;
+      }
       captureException(error, {
         mechanism: {
           handled: false,
@@ -180,13 +186,15 @@ export class SentryBullMQSpan implements TelemetrySpan {
 
   /**
    * Does not capture to Sentry. BullMQ calls this for Redis, lock and serialization errors that it
-   * retries itself, often several times for one fault. Job failures are captured in `addEvent`.
+   * retries itself, often several times for one fault, and for the job error right before the `job failed` event.
+   * Job failures are captured in `addEvent`.
    */
   public recordException(exception: Error | string | { code?: number; message?: string; name?: string }): void {
+    // BullMQ passes the retried errors as strings, so only the job error is kept.
+    this._recordedError = exception instanceof Error ? exception : undefined;
     const error =
-      exception instanceof Error
-        ? exception
-        : new Error(typeof exception === 'string' ? exception : exception.message || 'Unknown error');
+      this._recordedError ??
+      new Error(typeof exception === 'string' ? exception : exception.message || 'Unknown error');
 
     this._span.recordException(error);
   }

packages/node/test/integrations/tracing/bullmq.test.ts

@@ -397,6 +397,32 @@ describe('SentryBullMQSpan', () => {
       });
     });
 
+    it('captures the job error that BullMQ recorded before the "job failed" event', () => {
+      const { span } = createSpan();
+      const error = new TypeError('Cannot read properties of undefined');
+
+      span.recordException(error);
+      span.addEvent('job failed', { 'bullmq.job.failed.reason': error.message });
+
+      expect(SentryCore.captureException).toHaveBeenCalledTimes(1);
+      expect(SentryCore.captureException).toHaveBeenCalledWith(
+        expect.objectContaining({ name: 'TypeError', stack: error.stack }),
+        { mechanism: { handled: false, type: 'auto.queue.bullmq' } },
+      );
+    });
+
+    it('captures an error without a stack from the reason when BullMQ recorded only a string', () => {
+      const { span } = createSpan();
+
+      (span as any).recordException('Connection timed out');
+      span.addEvent('job failed', { 'bullmq.job.failed.reason': 'Connection refused' });
+
+      expect(SentryCore.captureException).toHaveBeenCalledWith(
+        expect.objectContaining({ message: 'Connection refused', stack: undefined }),
+        { mechanism: { handled: false, type: 'auto.queue.bullmq' } },
+      );
+    });
+
     it('sets the error status on the span when the event name is "job failed"', () => {
       const { span, otelSpan } = createSpan();