Clean up

Author: onurtemizkan

dev-packages/e2e-tests/test-applications/supabase-nextjs/pages/api/queue/consumer-error.ts

@@ -1,4 +1,4 @@
-// Enqueue a job to the queue
+// Consume from a non-existing queue to test error handling
 
 import { NextApiRequest, NextApiResponse } from 'next';
 import { createClient } from '@supabase/supabase-js';
@@ -14,7 +14,7 @@ const supabaseClient = createClient(NEXT_PUBLIC_SUPABASE_URL, SUPABASE_SERVICE_R
 Sentry.instrumentSupabaseClient(supabaseClient);
 
 export default async function handler(req: NextApiRequest, res: NextApiResponse) {
-  // Enqueue a job to the queue
+  // Pop from a non-existing queue to trigger an error
   const { data, error } = await supabaseClient.schema('pgmq_public').rpc('pop', {
     queue_name: 'non-existing-queue',
   });

dev-packages/e2e-tests/test-applications/supabase-nextjs/pages/api/queue/consumer-rpc.ts

@@ -1,4 +1,4 @@
-// Enqueue a job to the queue
+// Consume a message from the queue using direct rpc() call
 
 import { NextApiRequest, NextApiResponse } from 'next';
 import { createClient } from '@supabase/supabase-js';
@@ -18,7 +18,7 @@ const supabaseClient = createClient(NEXT_PUBLIC_SUPABASE_URL, SUPABASE_SERVICE_R
 Sentry.instrumentSupabaseClient(supabaseClient);
 
 export default async function handler(req: NextApiRequest, res: NextApiResponse) {
-  // Enqueue a job to the queue
+  // Pop a message from the queue
   const { data, error } = await supabaseClient.rpc('pop', {
     queue_name: 'todos',
   });

dev-packages/e2e-tests/test-applications/supabase-nextjs/tests/performance.test.ts

@@ -39,7 +39,7 @@ test('Sends server-side Supabase RPC spans and breadcrumbs', async ({ page, base
   const httpTransactionPromise = waitForTransaction('supabase-nextjs', transactionEvent => {
     return Boolean(
       transactionEvent?.contexts?.trace?.op === 'http.server' &&
-        transactionEvent?.transaction === 'GET /api/rpc/status',
+      transactionEvent?.transaction === 'GET /api/rpc/status',
     );
   });
 
@@ -352,8 +352,8 @@ test('Sends queue process spans with `schema(...).rpc(...)`', async ({ page, bas
   const producerTransactionPromise = waitForTransaction('supabase-nextjs', transactionEvent => {
     return Boolean(
       transactionEvent?.contexts?.trace?.op === 'http.server' &&
-        transactionEvent?.transaction === 'GET /api/queue/producer-schema' &&
-        transactionEvent?.spans?.some((span: any) => span.op === 'queue.publish'),
+      transactionEvent?.transaction === 'GET /api/queue/producer-schema' &&
+      transactionEvent?.spans?.some((span: any) => span.op === 'queue.publish'),
     );
   });
 
@@ -369,8 +369,8 @@ test('Sends queue process spans with `schema(...).rpc(...)`', async ({ page, bas
   const httpTransactionPromise = waitForTransaction('supabase-nextjs', transactionEvent => {
     return Boolean(
       transactionEvent?.contexts?.trace?.op === 'http.server' &&
-        transactionEvent?.transaction === 'GET /api/queue/consumer-schema' &&
-        transactionEvent?.spans?.some((span: any) => span.op === 'queue.process'),
+      transactionEvent?.transaction === 'GET /api/queue/consumer-schema' &&
+      transactionEvent?.spans?.some((span: any) => span.op === 'queue.process'),
     );
   });
 
@@ -575,7 +575,7 @@ test('Sends queue process error spans with `rpc(...)`', async ({ page, baseURL }
   const httpTransactionPromise = waitForTransaction('supabase-nextjs', transactionEvent => {
     return Boolean(
       transactionEvent?.contexts?.trace?.op === 'http.server' &&
-        transactionEvent?.transaction === 'GET /api/queue/consumer-error',
+      transactionEvent?.transaction === 'GET /api/queue/consumer-error',
     );
   });
 

packages/core/src/integrations/supabase/constants.ts

@@ -52,8 +52,6 @@ export const FILTER_MAPPINGS = {
 
 export const DB_OPERATIONS_TO_INSTRUMENT = ['select', 'insert', 'upsert', 'update', 'delete'];
 
-export const QUEUE_RPC_OPERATIONS = new Set(['send', 'send_batch', 'pop', 'receive', 'read']);
-
 export const INTEGRATION_NAME = 'Supabase';
 
 /**

packages/core/src/integrations/supabase/queue-consumer.ts

@@ -45,11 +45,7 @@ function _processConsumerSpan(span: Span, res: SupabaseResponse, queueName: stri
     span.setAttribute('messaging.message.retry.count', 0);
     span.setStatus({ code: res.error ? SPAN_STATUS_ERROR : SPAN_STATUS_OK });
 
-    const breadcrumbData: Record<string, unknown> = {};
-    if (queueName) {
-      breadcrumbData['messaging.destination.name'] = queueName;
-    }
-    _createQueueBreadcrumb('queue.process', queueName, Object.keys(breadcrumbData).length ? breadcrumbData : undefined);
+    _createQueueBreadcrumb('queue.process', queueName, queueName ? { 'messaging.destination.name': queueName } : undefined);
 
     if (res.error) {
       _captureQueueError(res.error, queueName);
@@ -136,9 +132,7 @@ export function _instrumentRpcConsumer(target: unknown, thisArg: unknown, argume
       queueName,
     });
 
-  const spanName = `process ${queueName || 'unknown'}`;
-  // Cloudflare pattern: op='db.queue' for valid transactions, 'queue.process' for Queue Insights.
-  // Works both as child spans and root spans.
+  const spanName = `process ${queueName}`;
   const spanAttributes = {
     [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: 'auto.db.supabase.queue.consumer',
     [SEMANTIC_ATTRIBUTE_SENTRY_OP]: 'queue.process',
@@ -188,6 +182,8 @@ export function _instrumentRpcConsumer(target: unknown, thisArg: unknown, argume
           const sentryTrace = firstMessage?._sentry?.sentry_trace;
 
           // Clean up _sentry metadata from messages before returning to user
+          // Shallow copy to avoid mutating the Supabase response object
+          let cleanedRes = res;
           if (Array.isArray(res.data)) {
             const hasMetadata = res.data.some(
               item =>
@@ -199,14 +195,15 @@ export function _instrumentRpcConsumer(target: unknown, thisArg: unknown, argume
             );
 
             if (hasMetadata) {
-              res.data = res.data.map(item => {
+              const cleanedData = res.data.map(item => {
                 if (item && typeof item === 'object' && item.message && typeof item.message === 'object') {
                   const messageCopy = { ...(item.message as Record<string, unknown>) };
                   delete messageCopy._sentry;
                   return { ...item, message: messageCopy };
                 }
                 return item;
               });
+              cleanedRes = { ...res, data: cleanedData };
             }
           }
 
@@ -227,7 +224,7 @@ export function _instrumentRpcConsumer(target: unknown, thisArg: unknown, argume
           }
 
           try {
-            const processedResponse = _processConsumerSpan(span, res, queueName);
+            const processedResponse = _processConsumerSpan(span, cleanedRes, queueName);
             DEBUG_BUILD && debug.log('Consumer span processed successfully', { queueName });
             return processedResponse;
           } catch (err: unknown) {
@@ -236,7 +233,7 @@ export function _instrumentRpcConsumer(target: unknown, thisArg: unknown, argume
             captureSupabaseError(err, 'auto.db.supabase.queue', { queueName });
 
             span.setStatus({ code: SPAN_STATUS_ERROR });
-            return res;
+            return cleanedRes;
           }
         },
         (err: unknown) => {

packages/core/src/integrations/supabase/queue-producer.ts

@@ -60,11 +60,9 @@ export function _instrumentRpcProducer(target: unknown, thisArg: unknown, argume
 
   const messageBodySize = _calculateMessageBodySize(queueParams?.message || queueParams?.messages);
 
-  // Cloudflare pattern: op='db.queue' for valid transactions, 'queue.publish' for Queue Insights.
-  // Works both as child spans and root spans.
   return startSpan(
     {
-      name: `publish ${queueName || 'unknown'}`,
+      name: `publish ${queueName}`,
       op: 'db.queue',
       attributes: {
         [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: 'auto.db.supabase.queue.producer',

packages/core/src/integrations/supabase/queue-utils.ts

@@ -61,7 +61,7 @@ export function _createQueueBreadcrumb(
   addBreadcrumb(breadcrumb);
 }
 
-/** Calculates the size of a message body in bytes, or undefined if too large or not serializable. */
+/** Calculates the approximate size of a message body (character count), or undefined if too large or not serializable. */
 export function _calculateMessageBodySize(message: unknown): number | undefined {
   if (!message) {
     return undefined;
@@ -95,7 +95,7 @@ export function _captureQueueError(
   captureSupabaseError(err, 'auto.db.supabase.queue', { queueName, messageId, ...extraContext });
 }
 
-/** Parses an enqueued_at timestamp and returns the latency in milliseconds. */
+/** Parses an enqueued_at timestamp and returns the latency in milliseconds. Assumes Supabase/PGMQ timestamptz (ISO 8601 with timezone). */
 export function _parseEnqueuedAtLatency(enqueuedAt: string | undefined): number | undefined {
   if (!enqueuedAt) {
     return undefined;

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

@@ -711,7 +711,7 @@ describe('Supabase Queue Instrumentation', () => {
   });
 
   describe('Non-Queue RPC Operations', () => {
-    it('should not instrument non-queue RPC calls', async () => {
+    it('should not instrument non-queue RPC calls as queue operations', async () => {
       const mockResponse = { data: { result: 'success' } };
       mockRpcFunction.mockResolvedValue(mockResponse);
       instrumentSupabaseClient(mockSupabaseClient);