Use span links for consumer distributed tracing, fix prototype instrumentation

Author: onurtemizkan

dev-packages/browser-integration-tests/suites/integrations/supabase/queues-rpc/init.js

@@ -20,7 +20,7 @@ async function performQueueOperations() {
   try {
     await supabaseClient.rpc('send', {
       queue_name: 'todos',
-      msg: { title: 'Test Todo' },
+      message: { title: 'Test Todo' },
     });
 
     await supabaseClient.rpc('pop', {

dev-packages/browser-integration-tests/suites/integrations/supabase/queues-schema/init.js

@@ -20,7 +20,7 @@ async function performQueueOperations() {
   try {
     await supabaseClient.schema('pgmq_public').rpc('send', {
       queue_name: 'todos',
-      msg: { title: 'Test Todo' },
+      message: { title: 'Test Todo' },
     });
 
     await supabaseClient.schema('pgmq_public').rpc('pop', {

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

@@ -271,7 +271,10 @@ test('Sends queue publish spans with `schema(...).rpc(...)`', async ({ page, bas
   const result = await fetch(`${baseURL}/api/queue/producer-schema`);
 
   expect(result.status).toBe(200);
-  expect(await result.json()).toEqual({ data: [1] });
+  const responseData = await result.json();
+  expect(responseData.data).toHaveLength(1);
+  expect(typeof responseData.data[0]).toBe('number');
+  const messageId = responseData.data[0];
 
   const transactionEvent = await httpTransactionPromise;
 
@@ -280,7 +283,7 @@ test('Sends queue publish spans with `schema(...).rpc(...)`', async ({ page, bas
     data: {
       'messaging.destination.name': 'todos',
       'messaging.system': 'supabase',
-      'messaging.message.id': '1',
+      'messaging.message.id': String(messageId),
       'messaging.operation.type': 'publish',
       'messaging.operation.name': 'send',
       'messaging.message.body.size': expect.any(Number),
@@ -305,7 +308,7 @@ test('Sends queue publish spans with `schema(...).rpc(...)`', async ({ page, bas
     message: 'queue.publish(todos)',
     data: {
       'messaging.destination.name': 'todos',
-      'messaging.message.id': '1',
+      'messaging.message.id': String(messageId),
       'messaging.message.body.size': expect.any(Number),
     },
   });
@@ -323,14 +326,17 @@ test('Sends queue publish spans with `rpc(...)`', async ({ page, baseURL }) => {
   const transactionEvent = await httpTransactionPromise;
 
   expect(result.status).toBe(200);
-  expect(await result.json()).toEqual({ data: [2] });
+  const responseData = await result.json();
+  expect(responseData.data).toHaveLength(1);
+  expect(typeof responseData.data[0]).toBe('number');
+  const messageId = responseData.data[0];
 
   expect(transactionEvent.spans).toHaveLength(2);
   expect(transactionEvent.spans).toContainEqual({
     data: {
       'messaging.destination.name': 'todos',
       'messaging.system': 'supabase',
-      'messaging.message.id': '2',
+      'messaging.message.id': String(messageId),
       'messaging.operation.type': 'publish',
       'messaging.operation.name': 'send',
       'messaging.message.body.size': expect.any(Number),
@@ -355,7 +361,7 @@ test('Sends queue publish spans with `rpc(...)`', async ({ page, baseURL }) => {
     message: 'queue.publish(todos)',
     data: {
       'messaging.destination.name': 'todos',
-      'messaging.message.id': '2',
+      'messaging.message.id': String(messageId),
       'messaging.message.body.size': expect.any(Number),
     },
   });
@@ -453,6 +459,11 @@ test('Sends queue process spans with `schema(...).rpc(...)`', async ({ page, bas
     },
   });
 
+  // CRITICAL: Verify the link actually points to the producer span from the first request
+  // This ensures distributed tracing works correctly across separate HTTP transactions
+  expect(producerLink?.trace_id).toBe(producerSpan?.trace_id);
+  expect(producerLink?.span_id).toBe(producerSpan?.span_id);
+
   expect(transactionEvent.breadcrumbs).toContainEqual({
     timestamp: expect.any(Number),
     type: 'supabase',
@@ -558,6 +569,11 @@ test('Sends queue process spans with `rpc(...)`', async ({ page, baseURL }) => {
     },
   });
 
+  // CRITICAL: Verify the link actually points to the producer span from the first request
+  // This ensures distributed tracing works correctly across separate HTTP transactions
+  expect(producerLink?.trace_id).toBe(producerSpan?.trace_id);
+  expect(producerLink?.span_id).toBe(producerSpan?.span_id);
+
   expect(transactionEvent.breadcrumbs).toContainEqual({
     timestamp: expect.any(Number),
     type: 'supabase',

packages/core/src/integrations/supabase.ts

@@ -3,7 +3,7 @@
 
 /* eslint-disable max-lines */
 import { addBreadcrumb } from '../breadcrumbs';
-import { getClient, getCurrentScope, withIsolationScope } from '../currentScopes';
+import { getClient, getCurrentScope } from '../currentScopes';
 import { DEBUG_BUILD } from '../debug-build';
 import { captureException } from '../exports';
 import { defineIntegration } from '../integration';
@@ -15,7 +15,7 @@ import {
 } from '../tracing/dynamicSamplingContext';
 import type { IntegrationFn } from '../types-hoist/integration';
 import type { Span, SpanAttributes } from '../types-hoist/span';
-import { dynamicSamplingContextToSentryBaggageHeader, parseBaggageHeader } from '../utils/baggage';
+import { dynamicSamplingContextToSentryBaggageHeader } from '../utils/baggage';
 import { debug } from '../utils/debug-logger';
 import { isPlainObject } from '../utils/is';
 import { addExceptionMechanism } from '../utils/misc';
@@ -27,6 +27,7 @@ export interface SupabaseClientConstructorType {
   prototype: {
     from: (table: string) => PostgRESTQueryBuilder;
     schema: (schema: string) => { rpc: (...args: unknown[]) => Promise<unknown> };
+    rpc: (...args: unknown[]) => Promise<unknown>;
   };
   rpc: (fn: string, params: Record<string, unknown>) => Promise<unknown>;
 }
@@ -648,7 +649,7 @@ const _instrumentRpcConsumer = (target: unknown, thisArg: unknown, argumentsList
       DEBUG_BUILD && debug.log('Consumer RPC call completed', { queueName, hasData: !!res.data });
 
       // Extract trace context from message for distributed tracing
-      const { sentryTrace, baggage } = _extractTraceAndBaggageFromMessage(res.data?.[0]?.message || {});
+      const { sentryTrace } = _extractTraceAndBaggageFromMessage(res.data?.[0]?.message || {});
 
       if (Array.isArray(res.data)) {
         res.data.forEach(item => {
@@ -658,11 +659,8 @@ const _instrumentRpcConsumer = (target: unknown, thisArg: unknown, argumentsList
         });
       }
 
-      // Extract producer trace context for span link and propagation
+      // Extract producer trace context for span link
       let producerSpanContext: { traceId: string; spanId: string; traceFlags: number } | undefined;
-      let producerPropagationContext:
-        | { traceId: string; parentSpanId: string; sampled: boolean; dsc?: Record<string, string> }
-        | undefined;
 
       if (sentryTrace) {
         const traceparentData = extractTraceparentData(sentryTrace);
@@ -676,117 +674,58 @@ const _instrumentRpcConsumer = (target: unknown, thisArg: unknown, argumentsList
             spanId: traceparentData.parentSpanId,
             traceFlags,
           };
-
-          // Prepare propagation context for isolated scope
-          producerPropagationContext = {
-            traceId: traceparentData.traceId,
-            parentSpanId: traceparentData.parentSpanId,
-            sampled: traceparentData.parentSampled ?? false,
-            dsc: baggage ? parseBaggageHeader(baggage) : undefined,
-          };
         }
       }
 
       const runWithSpan = (): SupabaseResponse => {
-        // If we have producer trace context, use isolated scope to prevent pollution
-        if (producerPropagationContext) {
-          return withIsolationScope(isolatedScope => {
-            // Set producer's propagation context in isolated scope
-            // This ensures the consumer span continues the producer's trace
-            isolatedScope.setPropagationContext({
-              ...producerPropagationContext,
-              sampleRand: Math.random(), // Generate new sample rand for current execution context
-            });
-
-            // Force transaction to make it a root span (not child of current span)
-            // This is critical to prevent the consumer span from becoming a child of
-            // an active HTTP request or other unrelated transaction
-            return startSpan(
-              {
-                name: spanName,
-                op: 'queue.process',
-                startTime: spanStartTime,
-                attributes: spanAttributes,
-                forceTransaction: true, // Makes this a root span, not a child
-                // 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);
-
-                  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;
-                    });
-                    scope.setContext('supabase', { queueName });
-                    return scope;
-                  });
-
-                  span.setStatus({ code: SPAN_STATUS_ERROR });
-                  throw err;
-                }
-              },
-            );
-          });
-          // Isolated scope automatically discarded here, original scope restored
-        } else {
-          // No producer context, create regular span without isolation
-          return startSpan(
-            {
-              name: spanName,
-              op: 'queue.process',
-              startTime: spanStartTime,
-              attributes: spanAttributes,
-            },
-            span => {
-              try {
-                const processedResponse = _processConsumerSpan(span, res, queueName);
+        // 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);
 
-                DEBUG_BUILD && debug.log('Consumer span processed successfully', { 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 });
+              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 });
-                throw err;
-              }
-            },
-          );
-        }
+              span.setStatus({ code: SPAN_STATUS_ERROR });
+              throw err;
+            }
+          },
+        );
       };
 
-      // Create consumer span with isolated scope and forced transaction
+      // Create consumer span as child of current transaction with span links for distributed tracing
       return runWithSpan();
     })
     .catch((err: unknown) => {
@@ -1026,20 +965,32 @@ function _instrumentRpcProducer(target: unknown, thisArg: unknown, argumentsList
 }
 
 /**
- * Instruments direct RPC calls on a Supabase client.
+ * Instruments direct RPC calls on a Supabase client's constructor prototype.
  * This handles the pattern: `client.rpc('function_name', params)`
  * Uses the shared proxy handler to route queue operations.
  *
- * @param SupabaseClient - The Supabase client instance to instrument
+ * We instrument the prototype rather than individual instances to ensure consistent
+ * behavior across all clients sharing the same constructor and to avoid issues with
+ * Proxy property forwarding affecting the instrumentation marker on the original function.
+ *
+ * @param SupabaseClientConstructor - The Supabase client constructor to instrument
  */
-function _instrumentRpc(SupabaseClient: unknown): void {
-  const client = SupabaseClient as SupabaseClientInstance;
+function _instrumentRpc(SupabaseClientConstructor: unknown): void {
+  const prototype = (SupabaseClientConstructor as SupabaseClientConstructorType).prototype;
+
+  if (!prototype?.rpc) {
+    return;
+  }
 
-  if (!client.rpc) {
+  // Prevent double-wrapping if instrumentSupabaseClient is called multiple times
+  if (_isInstrumented(prototype.rpc)) {
     return;
   }
 
-  client.rpc = new Proxy(client.rpc, _createRpcProxyHandler());
+  const wrappedRpc = new Proxy(prototype.rpc, _createRpcProxyHandler());
+  prototype.rpc = wrappedRpc;
+
+  _markAsInstrumented(prototype.rpc);
 }
 
 /**
@@ -1454,7 +1405,7 @@ export const instrumentSupabaseClient = (supabaseClient: unknown): void => {
 
   _instrumentSupabaseClientConstructor(SupabaseClientConstructor);
   _instrumentRpcReturnedFromSchemaCall(SupabaseClientConstructor);
-  _instrumentRpc(supabaseClient as SupabaseClientInstance);
+  _instrumentRpc(SupabaseClientConstructor);
   _instrumentSupabaseAuthClient(supabaseClient as SupabaseClientInstance);
 };