Use OTEL-compliant messaging.batch.message_count attribute.

onurtemizkan · onurtemizkan · commit 58858339d997 · 2025-12-09T14:29:19.000Z
diff --git a/dev-packages/e2e-tests/test-applications/supabase-nextjs/tests/performance.test.ts b/dev-packages/e2e-tests/test-applications/supabase-nextjs/tests/performance.test.ts
@@ -265,6 +265,7 @@ test('Sends queue publish spans with `schema(...).rpc(...)`', async ({ page, bas
       'messaging.message.body.size': expect.any(Number),
       'sentry.op': 'queue.publish',
       'sentry.origin': 'auto.db.supabase.queue.producer',
+      'sentry.source': 'task',
     },
     description: 'publish todos',
     op: 'queue.publish',
@@ -318,6 +319,7 @@ test('Sends queue publish spans with `rpc(...)`', async ({ page, baseURL }) => {
       'messaging.message.body.size': expect.any(Number),
       'sentry.op': 'queue.publish',
       'sentry.origin': 'auto.db.supabase.queue.producer',
+      'sentry.source': 'task',
     },
     description: 'publish todos',
     op: 'queue.publish',
@@ -664,6 +666,7 @@ test('Sends queue batch publish spans with `rpc(...)`', async ({ page, baseURL }
       'messaging.message.body.size': expect.any(Number),
       'sentry.op': 'queue.publish',
       'sentry.origin': 'auto.db.supabase.queue.producer',
+      'sentry.source': 'task',
     },
     description: 'publish todos',
     op: 'queue.publish',
diff --git a/packages/core/src/integrations/supabase.ts b/packages/core/src/integrations/supabase.ts
@@ -7,7 +7,11 @@ import { getClient, getCurrentScope } from '../currentScopes';
 import { DEBUG_BUILD } from '../debug-build';
 import { captureException } from '../exports';
 import { defineIntegration } from '../integration';
-import { SEMANTIC_ATTRIBUTE_SENTRY_OP, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN } from '../semanticAttributes';
+import {
+  SEMANTIC_ATTRIBUTE_SENTRY_OP,
+  SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN,
+  SEMANTIC_ATTRIBUTE_SENTRY_SOURCE,
+} from '../semanticAttributes';
 import {
   setHttpStatus,
   SPAN_STATUS_ERROR,
@@ -27,7 +31,6 @@ import { debug } from '../utils/debug-logger';
 import { isPlainObject } from '../utils/is';
 import { addExceptionMechanism } from '../utils/misc';
 import { spanToTraceContext, spanToTraceHeader } from '../utils/spanUtils';
-import { timestampInSeconds } from '../utils/time';
 import { extractTraceparentData } from '../utils/tracing';
 
 export interface SupabaseClientConstructorType {
@@ -714,6 +717,10 @@ function _processConsumerSpan(span: Span, res: SupabaseResponse, queueName: stri
   const messageId = _extractMessageIds(data);
 
   // Set span attributes
+  // Set batch message count for OTel semantic conventions compliance
+  // This is set for all consumer operations since pop() returns an array (batch-oriented)
+  span.setAttribute('messaging.batch.message_count', data.length);
+
   if (messageId) {
     span.setAttribute('messaging.message.id', messageId);
   }
@@ -793,161 +800,129 @@ const _instrumentRpcConsumer = (target: unknown, thisArg: unknown, argumentsList
     });
 
   const spanName = `process ${queueName || 'unknown'}`;
+  // Use the Cloudflare pattern for queue spans:
+  // - op: 'db.queue' makes the span a valid transaction when no parent exists
+  // - SEMANTIC_ATTRIBUTE_SENTRY_OP: 'queue.process' ensures Queue Insights recognition
+  // - SEMANTIC_ATTRIBUTE_SENTRY_SOURCE: 'task' marks it as a task source
+  // This pattern allows queue spans to work both as child spans (when inside a transaction)
+  // and as root spans (when called outside a transaction context).
   const spanAttributes = {
     [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: 'auto.db.supabase.queue.consumer',
     [SEMANTIC_ATTRIBUTE_SENTRY_OP]: 'queue.process',
+    [SEMANTIC_ATTRIBUTE_SENTRY_SOURCE]: 'task',
     'messaging.system': 'supabase',
     'messaging.destination.name': queueName,
     'messaging.operation.name': operationName,
     'messaging.operation.type': 'process',
   } as const;
-  const spanStartTime = timestampInSeconds();
-
-  const rpcPromise = Reflect.apply(
-    target as (...args: unknown[]) => Promise<unknown>,
-    thisArg,
-    argumentsList,
-  ) as Promise<SupabaseResponse>;
-
-  return rpcPromise
-    .then(res => {
-      DEBUG_BUILD && debug.log('Consumer RPC call completed', { queueName, hasData: !!res.data });
-
-      // Skip span creation for empty/null responses when there's no error - no messages to process
-      // Still create span if there's an error to capture the failure
-      if ((!res.data || (Array.isArray(res.data) && res.data.length === 0)) && !res.error) {
-        DEBUG_BUILD && debug.log('Skipping consumer span for empty response', { queueName });
-        return res;
-      }
 
-      // Extract trace context from message for distributed tracing
-      const { sentryTrace } = _extractTraceAndBaggageFromMessage(res.data?.[0]?.message || {});
-
-      // Clean up _sentry metadata from messages before returning to user
-      // Use immutable updates to avoid mutating the original response data
-      if (Array.isArray(res.data)) {
-        const hasMetadata = res.data.some(
-          item =>
-            item &&
-            typeof item === 'object' &&
-            item.message &&
-            typeof item.message === 'object' &&
-            '_sentry' in item.message,
-        );
-
-        if (hasMetadata) {
-          res.data = 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;
-          });
-        }
-      }
-
-      // Extract producer trace context for span link
-      let producerSpanContext: { traceId: string; spanId: string; traceFlags: number } | undefined;
+  // Wrap the entire RPC call with startSpan to ensure the span is created BEFORE the async operation.
+  // This matches the producer pattern and ensures the span is properly attached to the current
+  // transaction context. Previously, the span was created inside .then() which could cause
+  // the span to be orphaned if the transaction ended before the callback executed.
+  return startSpan(
+    {
+      name: spanName,
+      op: 'db.queue',
+      attributes: spanAttributes,
+    },
+    span => {
+      const rpcPromise = Reflect.apply(
+        target as (...args: unknown[]) => Promise<unknown>,
+        thisArg,
+        argumentsList,
+      ) as Promise<SupabaseResponse>;
 
-      if (sentryTrace) {
-        const traceparentData = extractTraceparentData(sentryTrace);
-        if (traceparentData?.traceId && traceparentData?.parentSpanId) {
-          // Convert parentSampled boolean to traceFlags (W3C trace context spec)
-          // traceFlags bit 0 (LSB) = sampled flag: 1 if sampled, 0 if not sampled
-          const traceFlags = traceparentData.parentSampled ? 1 : 0;
+      return rpcPromise.then(
+        (res: SupabaseResponse) => {
+          DEBUG_BUILD && debug.log('Consumer RPC call completed', { queueName, hasData: !!res.data });
+
+          // Handle empty responses - set span status but don't skip span creation
+          // The span is already created, we just mark it appropriately
+          if ((!res.data || (Array.isArray(res.data) && res.data.length === 0)) && !res.error) {
+            DEBUG_BUILD && debug.log('Consumer received empty response', { queueName });
+            span.setStatus({ code: SPAN_STATUS_OK });
+            span.setAttribute('messaging.batch.message_count', 0);
+            return res;
+          }
 
-          producerSpanContext = {
-            traceId: traceparentData.traceId,
-            spanId: traceparentData.parentSpanId,
-            traceFlags,
-          };
-        }
-      }
+          // Extract trace context from message for distributed tracing (before cleanup)
+          const { sentryTrace } = _extractTraceAndBaggageFromMessage(res.data?.[0]?.message || {});
+
+          // Clean up _sentry metadata from messages before returning to user
+          // Use immutable updates to avoid mutating the original response data
+          if (Array.isArray(res.data)) {
+            const hasMetadata = res.data.some(
+              item =>
+                item &&
+                typeof item === 'object' &&
+                item.message &&
+                typeof item.message === 'object' &&
+                '_sentry' in item.message,
+            );
 
-      const runWithSpan = (): SupabaseResponse => {
-        // Create consumer span as child of current transaction (e.g., HTTP request)
-        // Add span link to producer span for distributed tracing across async queue boundary
-        return startSpan(
-          {
-            name: spanName,
-            op: 'queue.process',
-            startTime: spanStartTime,
-            attributes: spanAttributes,
-            // Add span link to producer span for distributed tracing across async queue boundary
-            links: producerSpanContext
-              ? [
-                  {
-                    context: producerSpanContext,
-                    attributes: { 'sentry.link.type': 'queue.producer' },
-                  },
-                ]
-              : undefined,
-          },
-          span => {
-            try {
-              const processedResponse = _processConsumerSpan(span, res, queueName);
+            if (hasMetadata) {
+              res.data = 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;
+              });
+            }
+          }
 
-              DEBUG_BUILD && debug.log('Consumer span processed successfully', { queueName });
+          // Extract producer trace context and add span link for distributed tracing
+          if (sentryTrace) {
+            const traceparentData = extractTraceparentData(sentryTrace);
+            if (traceparentData?.traceId && traceparentData?.parentSpanId) {
+              // Convert parentSampled boolean to traceFlags (W3C trace context spec)
+              // traceFlags bit 0 (LSB) = sampled flag: 1 if sampled, 0 if not sampled
+              const traceFlags = traceparentData.parentSampled ? 1 : 0;
+
+              span.addLink({
+                context: {
+                  traceId: traceparentData.traceId,
+                  spanId: traceparentData.parentSpanId,
+                  traceFlags,
+                },
+                attributes: { 'sentry.link.type': 'queue.producer' },
+              });
+            }
+          }
 
-              return processedResponse;
-            } catch (err: unknown) {
-              // Handle span processing errors without re-throwing.
-              // The outer .catch() is for RPC promise rejections only.
-              // Re-throwing here would cause duplicate spans and duplicate captureException calls
-              // because the .then().catch() pattern catches errors from both promise rejections
-              // and errors thrown within the .then() callback.
-              DEBUG_BUILD && debug.log('Consumer span processing failed', { queueName, error: err });
+          // Process the span with response data
+          try {
+            const processedResponse = _processConsumerSpan(span, res, queueName);
+            DEBUG_BUILD && debug.log('Consumer span processed successfully', { queueName });
+            return processedResponse;
+          } catch (err: unknown) {
+            DEBUG_BUILD && debug.log('Consumer span processing failed', { queueName, error: err });
 
-              captureException(err, scope => {
-                scope.addEventProcessor(e => {
-                  addExceptionMechanism(e, {
-                    handled: false,
-                    type: 'auto.db.supabase.queue',
-                  });
-                  return e;
+            captureException(err, scope => {
+              scope.addEventProcessor(e => {
+                addExceptionMechanism(e, {
+                  handled: false,
+                  type: 'auto.db.supabase.queue',
                 });
-                scope.setContext('supabase', { queueName });
-                return scope;
+                return e;
               });
+              scope.setContext('supabase', { queueName });
+              return scope;
+            });
 
-              span.setStatus({ code: SPAN_STATUS_ERROR });
-
-              // Return the original response since the RPC call itself succeeded.
-              // Only span processing failed, which we've already captured.
-              return res;
-            }
-          },
-        );
-      };
-
-      // Create consumer span as child of current transaction with span links for distributed tracing
-      return runWithSpan();
-    })
-    .catch((err: unknown) => {
-      DEBUG_BUILD && debug.log('Consumer RPC call failed', { queueName, error: err });
-
-      return startSpan(
-        {
-          name: spanName,
-          op: 'queue.process',
-          startTime: spanStartTime,
-          attributes: spanAttributes,
-        },
-        span => {
-          if (queueName) {
-            span.setAttribute('messaging.destination.name', queueName);
+            span.setStatus({ code: SPAN_STATUS_ERROR });
+            return res;
           }
+        },
+        (err: unknown) => {
+          // Handle RPC promise rejection
+          DEBUG_BUILD && debug.log('Consumer RPC call failed', { queueName, error: err });
 
-          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, {
+            'messaging.destination.name': queueName,
+          });
 
           captureException(err, scope => {
             scope.addEventProcessor(e => {
@@ -965,7 +940,8 @@ const _instrumentRpcConsumer = (target: unknown, thisArg: unknown, argumentsList
           throw err;
         },
       );
-    });
+    },
+  );
 };
 
 /**
@@ -1008,13 +984,20 @@ function _instrumentRpcProducer(target: unknown, thisArg: unknown, argumentsList
   // Calculate message body size upfront for initial span attributes
   const messageBodySize = _calculateMessageBodySize(queueParams?.message || queueParams?.messages);
 
+  // Use the Cloudflare pattern for queue spans:
+  // - op: 'db.queue' makes the span a valid transaction when no parent exists
+  // - SEMANTIC_ATTRIBUTE_SENTRY_OP: 'queue.publish' ensures Queue Insights recognition
+  // - SEMANTIC_ATTRIBUTE_SENTRY_SOURCE: 'task' marks it as a task source
+  // This pattern allows queue spans to work both as child spans (when inside a transaction)
+  // and as root spans (when called outside a transaction context).
   return startSpan(
     {
       name: `publish ${queueName || 'unknown'}`,
-      op: 'queue.publish',
+      op: 'db.queue',
       attributes: {
         [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: 'auto.db.supabase.queue.producer',
         [SEMANTIC_ATTRIBUTE_SENTRY_OP]: 'queue.publish',
+        [SEMANTIC_ATTRIBUTE_SENTRY_SOURCE]: 'task',
         'messaging.system': 'supabase',
         'messaging.destination.name': queueName,
         'messaging.operation.name': operationName,
diff --git a/packages/core/test/lib/integrations/supabase-queues.test.ts b/packages/core/test/lib/integrations/supabase-queues.test.ts
