Distinguish queue producer and consumer error mechanisms

Author: onurtemizkan

packages/core/src/integrations/supabase.ts

@@ -611,14 +611,15 @@ function calculateMessageBodySize(message: unknown): number | undefined {
 function captureQueueError(
   error: { message: string; code?: string; details?: unknown },
   queueName: string | undefined,
+  mechanismType: string,
   messageId?: string,
   extraContext?: Record<string, unknown>,
 ): void {
   const err = new Error(error.message) as SupabaseError;
   if (error.code) err.code = error.code;
   if (error.details) err.details = error.details;
 
-  captureSupabaseError(err, 'auto.db.supabase.queue', { queueName, messageId, ...extraContext });
+  captureSupabaseError(err, mechanismType, { queueName, messageId, ...extraContext });
 }
 
 /** Returns latency from an enqueued_at timestamp in milliseconds. */
@@ -764,7 +765,9 @@ function instrumentRpcProducer(
           });
 
           if (res.error) {
-            captureQueueError(res.error, queueName, messageId, { operation: operationName });
+            captureQueueError(res.error, queueName, 'auto.db.supabase.queue.producer', messageId, {
+              operation: operationName,
+            });
           }
 
           span.setStatus({ code: res.error ? SPAN_STATUS_ERROR : SPAN_STATUS_OK });
@@ -774,7 +777,7 @@ function instrumentRpcProducer(
         (err: unknown) => {
           span.setStatus({ code: SPAN_STATUS_ERROR });
 
-          captureSupabaseError(err, 'auto.db.supabase.queue', { queueName, operation: operationName });
+          captureSupabaseError(err, 'auto.db.supabase.queue.producer', { queueName, operation: operationName });
 
           throw err;
         },
@@ -973,7 +976,7 @@ function instrumentRpcConsumer(
             });
 
             if (cleanedRes.error) {
-              captureQueueError(cleanedRes.error, queueName);
+              captureQueueError(cleanedRes.error, queueName, 'auto.db.supabase.queue.consumer');
             }
 
             return cleanedRes;
@@ -983,7 +986,7 @@ function instrumentRpcConsumer(
 
           if (cleanedRes.error) {
             const messageId = extractMessageIds(cleanedData);
-            captureQueueError(cleanedRes.error, queueName, messageId);
+            captureQueueError(cleanedRes.error, queueName, 'auto.db.supabase.queue.consumer', messageId);
           }
 
           span.setStatus({ code: cleanedRes.error ? SPAN_STATUS_ERROR : SPAN_STATUS_OK });
@@ -998,7 +1001,7 @@ function instrumentRpcConsumer(
             data: { 'messaging.destination.name': queueName },
           });
 
-          captureSupabaseError(err, 'auto.db.supabase.queue', { queueName });
+          captureSupabaseError(err, 'auto.db.supabase.queue.consumer', { queueName });
 
           span.setStatus({ code: SPAN_STATUS_ERROR });
           throw err;

packages/core/test/lib/integrations/supabase-queues.test.ts

@@ -1,6 +1,7 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import type { Client } from '../../../src';
 import * as CurrentScopes from '../../../src/currentScopes';
+import * as exports from '../../../src/exports';
 import type { SupabaseClientInstance, SupabaseResponse } from '../../../src/integrations/supabase';
 import { instrumentSupabaseClient } from '../../../src/integrations/supabase';
 import { SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN } from '../../../src/semanticAttributes';
@@ -156,6 +157,37 @@ describe('Supabase Queue Instrumentation', () => {
       ).rejects.toThrow('Queue send failed');
     });
 
+    it('should capture producer errors with producer mechanism type', async () => {
+      const captureExceptionSpy = vi.spyOn(exports, 'captureException').mockImplementation(() => '');
+
+      await callRpc(
+        mockRpcFunction,
+        mockSupabaseClient,
+        'send',
+        { queue_name: 'test-queue', message: { foo: 'bar' } },
+        ERROR_RESPONSE,
+      );
+
+      expect(captureExceptionSpy).toHaveBeenCalledTimes(1);
+
+      // Execute the scope callback to verify mechanism type
+      const scopeCallback = captureExceptionSpy.mock.calls[0]![1] as (scope: any) => any;
+      const mockScope = { addEventProcessor: vi.fn().mockReturnThis(), setContext: vi.fn().mockReturnThis() };
+      scopeCallback(mockScope);
+
+      const eventProcessor = mockScope.addEventProcessor.mock.calls[0]![0];
+      const event = { exception: { values: [{}] } };
+      eventProcessor(event);
+
+      expect(event.exception.values[0]).toEqual(
+        expect.objectContaining({
+          mechanism: expect.objectContaining({ type: 'auto.db.supabase.queue.producer' }),
+        }),
+      );
+
+      captureExceptionSpy.mockRestore();
+    });
+
     it('should not mutate original params for single send or batch send', async () => {
       const singleParams = { queue_name: 'test-queue', message: { foo: 'bar', nested: { value: 42 } } };
       const batchParams = { queue_name: 'test-queue', messages: [{ foo: 'bar' }, { baz: 'qux' }] };
@@ -308,6 +340,30 @@ describe('Supabase Queue Instrumentation', () => {
       expect(processSpanCall?.[0]?.name).toBe('process test-queue');
     });
 
+    it('should capture consumer errors with consumer mechanism type', async () => {
+      const captureExceptionSpy = vi.spyOn(exports, 'captureException').mockImplementation(() => '');
+
+      await callRpc(mockRpcFunction, mockSupabaseClient, 'pop', { queue_name: 'test-queue' }, ERROR_RESPONSE);
+
+      expect(captureExceptionSpy).toHaveBeenCalledTimes(1);
+
+      const scopeCallback = captureExceptionSpy.mock.calls[0]![1] as (scope: any) => any;
+      const mockScope = { addEventProcessor: vi.fn().mockReturnThis(), setContext: vi.fn().mockReturnThis() };
+      scopeCallback(mockScope);
+
+      const eventProcessor = mockScope.addEventProcessor.mock.calls[0]![0];
+      const event = { exception: { values: [{}] } };
+      eventProcessor(event);
+
+      expect(event.exception.values[0]).toEqual(
+        expect.objectContaining({
+          mechanism: expect.objectContaining({ type: 'auto.db.supabase.queue.consumer' }),
+        }),
+      );
+
+      captureExceptionSpy.mockRestore();
+    });
+
     it('should set correct attributes on consumer span', async () => {
       const startSpanSpy = vi.spyOn(Tracing, 'startSpan');