feat(cloudflare): propagate tracing context through Queue messages

dev-packages/cloudflare-integration-tests/suites/queue/index.ts

@@ -10,6 +10,7 @@ export default Sentry.withSentry(
   (env: Env) => ({
     dsn: env.SENTRY_DSN,
     tracesSampleRate: 1,
+    enableQueueTracePropagation: true,
   }),
   {
     async fetch(request, env) {
@@ -37,6 +38,11 @@ export default Sentry.withSentry(
       return new Response('not found', { status: 404 });
     },
     async queue(batch: MessageBatch<{ trigger?: 'error'; payload?: string }>) {
+      Sentry.getActiveSpan()?.setAttribute(
+        'test.queue.message_bodies',
+        JSON.stringify(batch.messages.map(message => message.body)),
+      );
+
       for (const message of batch.messages) {
         if (message.body.trigger === 'error') {
           throw new Error('Boom from queue handler');

dev-packages/cloudflare-integration-tests/suites/queue/test.ts

@@ -51,13 +51,15 @@ it('captures errors thrown by the queue handler with the correct mechanism', asy
 it('emits a queue.publish span on env.MY_QUEUE.send and a queue.process transaction on the consumer', async ({
   signal,
 }) => {
+  let publishSpan: Record<string, unknown> | undefined;
+  let consumerTrace: Record<string, unknown> | undefined;
   const runner = createRunner(__dirname)
     .unordered()
     .expect((envelope: Envelope) => {
       // Producer transaction must contain a queue.publish child span
-      const publishSpan = findPublishSpan(envelope);
-      expect(publishSpan).toBeDefined();
-      expect(publishSpan).toMatchObject({
+      const candidate = findPublishSpan(envelope);
+      expect(candidate).toBeDefined();
+      expect(candidate).toMatchObject({
         op: 'queue.publish',
         description: 'send MY_QUEUE',
         data: expect.objectContaining({
@@ -68,6 +70,7 @@ it('emits a queue.publish span on env.MY_QUEUE.send and a queue.process transact
           'sentry.origin': 'auto.faas.cloudflare.queue',
         }),
       });
+      publishSpan = candidate;
     })
     .expect((envelope: Envelope) => {
       expect(isConsumerTransaction(envelope)).toBe(true);
@@ -83,22 +86,39 @@ it('emits a queue.publish span on env.MY_QUEUE.send and a queue.process transact
           'messaging.operation.name': 'process',
           'messaging.batch.message_count': 1,
           'faas.trigger': 'pubsub',
+          'test.queue.message_bodies': '[{"payload":"hello"}]',
         }),
       });
+      consumerTrace = trace;
     })
     .start(signal);
 
   await runner.makeRequest('post', '/enqueue/ok');
   await runner.completed();
+
+  expect(publishSpan).toBeDefined();
+  expect(consumerTrace).toBeDefined();
+  expect(consumerTrace!.trace_id).not.toBe(publishSpan!.trace_id);
+  expect(consumerTrace!.links).toEqual([
+    {
+      trace_id: publishSpan!.trace_id,
+      span_id: publishSpan!.span_id,
+      sampled: true,
+      isRemote: true,
+      attributes: { 'sentry.link.type': 'previous_trace' },
+    },
+  ]);
 });
 
 it('emits a queue.publish span with batch attributes on env.MY_QUEUE.sendBatch', async ({ signal }) => {
+  let publishSpan: Record<string, unknown> | undefined;
+  let consumerTrace: Record<string, unknown> | undefined;
   const runner = createRunner(__dirname)
     .unordered()
     .expect((envelope: Envelope) => {
-      const publishSpan = findPublishSpan(envelope);
-      expect(publishSpan).toBeDefined();
-      expect(publishSpan).toMatchObject({
+      const candidate = findPublishSpan(envelope);
+      expect(candidate).toBeDefined();
+      expect(candidate).toMatchObject({
         op: 'queue.publish',
         description: 'send MY_QUEUE',
         data: expect.objectContaining({
@@ -110,6 +130,7 @@ it('emits a queue.publish span with batch attributes on env.MY_QUEUE.sendBatch',
           'sentry.origin': 'auto.faas.cloudflare.queue',
         }),
       });
+      publishSpan = candidate;
     })
     .expect((envelope: Envelope) => {
       expect(isConsumerTransaction(envelope)).toBe(true);
@@ -120,9 +141,22 @@ it('emits a queue.publish span with batch attributes on env.MY_QUEUE.sendBatch',
           'messaging.batch.message_count': 3,
         }),
       });
+      consumerTrace = trace;
     })
     .start(signal);
 
   await runner.makeRequest('post', '/enqueue/batch');
   await runner.completed();
+
+  expect(publishSpan).toBeDefined();
+  expect(consumerTrace).toBeDefined();
+  expect(consumerTrace!.links).toEqual([
+    {
+      trace_id: publishSpan!.trace_id,
+      span_id: publishSpan!.span_id,
+      sampled: true,
+      isRemote: true,
+      attributes: { 'sentry.link.type': 'previous_trace' },
+    },
+  ]);
 });

packages/cloudflare/src/client.ts

@@ -210,6 +210,19 @@ interface BaseCloudflareOptions {
    */
   enableRpcTracePropagation?: boolean;
 
+  /**
+   * Enable trace propagation from Queue producers to Queue consumers using span links.
+   *
+   * Cloudflare Queues does not expose a message-header API, so the SDK adds trace context to record-like message
+   * bodies under `__sentry_queue_trace__`. The consumer removes this field before invoking the queue handler.
+   *
+   * **Important:** Enable this option on both producers and consumers. Consumers which are not instrumented with
+   * this option receive the additional field, which may break strict message validation or affect payload limits.
+   *
+   * @default false
+   */
+  enableQueueTracePropagation?: boolean;
+
   /**
    * @deprecated Use `enableRpcTracePropagation` instead. This option will be removed in a future major version.
    *

packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts

@@ -14,6 +14,7 @@ function isProxyable(item: unknown): item is object {
 }
 
 const instrumentedBindings = new WeakMap<object, unknown>();
+const instrumentedQueueBindings = new WeakMap<object, Map<string, unknown>>();
 
 /**
  * Wraps the Cloudflare `env` object in a Proxy that detects binding types
@@ -43,6 +44,22 @@ export function instrumentEnv<Env extends Record<string, unknown>>(env: Env, opt
         return item;
       }
 
+      if (isQueue(item)) {
+        const bindingName = typeof prop === 'string' ? prop : String(prop);
+        const enableTracePropagation = options?.enableQueueTracePropagation === true;
+        const cacheKey = `${bindingName}:${enableTracePropagation}`;
+        const cached = instrumentedQueueBindings.get(item)?.get(cacheKey);
+        if (cached) {
+          return cached;
+        }
+
+        const instrumented = instrumentQueueProducer(item, bindingName, enableTracePropagation);
+        const queueBindings = instrumentedQueueBindings.get(item) || new Map<string, unknown>();
+        queueBindings.set(cacheKey, instrumented);
+        instrumentedQueueBindings.set(item, queueBindings);
+        return instrumented;
+      }
+
       const cached = instrumentedBindings.get(item);
 
       if (cached) {
@@ -55,13 +72,6 @@ export function instrumentEnv<Env extends Record<string, unknown>>(env: Env, opt
         return instrumented;
       }
 
-      if (isQueue(item)) {
-        const bindingName = typeof prop === 'string' ? prop : String(prop);
-        const instrumented = instrumentQueueProducer(item, bindingName);
-        instrumentedBindings.set(item, instrumented);
-        return instrumented;
-      }
-
       if (isR2Bucket(item)) {
         const bindingName = typeof prop === 'string' ? prop : String(prop);
         const instrumented = instrumentR2Bucket(item, bindingName);

packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts

@@ -1,7 +1,10 @@
 import type { ExportedHandler, MessageBatch } from '@cloudflare/workers-types';
+import { TraceFlags } from '@opentelemetry/api';
 import type { env as cloudflareEnv, WorkerEntrypoint } from 'cloudflare:workers';
+import type { SpanLink } from '@sentry/core';
 import {
   captureException,
+  SEMANTIC_LINK_ATTRIBUTE_LINK_TYPE,
   SEMANTIC_ATTRIBUTE_SENTRY_OP,
   SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN,
   SEMANTIC_ATTRIBUTE_SENTRY_SOURCE,
@@ -15,8 +18,38 @@ import { getFinalOptions } from '../../options';
 import { addCloudResourceContext } from '../../scope-utils';
 import { init } from '../../sdk';
 import { instrumentContext } from '../../utils/instrumentContext';
+import { extractQueueTraceContext } from '../../utils/queueTraceMeta';
 import { instrumentEnv } from './instrumentEnv';
 
+function extractQueueTraceLinks(batch: MessageBatch): SpanLink[] {
+  const linkedSpans = new Set<string>();
+
+  return batch.messages.flatMap(message => {
+    const producerContext = extractQueueTraceContext(message.body);
+    if (!producerContext) {
+      return [];
+    }
+
+    const spanKey = `${producerContext.traceId}:${producerContext.spanId}`;
+    if (linkedSpans.has(spanKey)) {
+      return [];
+    }
+    linkedSpans.add(spanKey);
+
+    return [
+      {
+        context: {
+          traceId: producerContext.traceId,
+          spanId: producerContext.spanId,
+          traceFlags: producerContext.sampled ? TraceFlags.SAMPLED : TraceFlags.NONE,
+          isRemote: true,
+        },
+        attributes: { [SEMANTIC_LINK_ATTRIBUTE_LINK_TYPE]: 'previous_trace' },
+      },
+    ];
+  });
+}
+
 /**
  * Core queue handler logic - wraps execution with Sentry instrumentation.
  */
@@ -34,10 +67,15 @@ function wrapQueueHandler(
 
     addCloudResourceContext(isolationScope);
 
+    const links = options.enableQueueTracePropagation ? extractQueueTraceLinks(batch) : [];
+
     return startSpan(
       {
         op: 'faas.queue',
         name: `process ${batch.queue}`,
+        // A queue batch can fan in unrelated traces, so continuing one trace would invent false parentage.
+        // Keep queue.process as a root and link every producer instead: https://github.com/getsentry/sentry-javascript/issues/22298
+        links,
         attributes: {
           'faas.trigger': 'pubsub',
           'messaging.destination.name': batch.queue,

packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts

@@ -1,5 +1,7 @@
 import type { MessageSendRequest, Queue, QueueSendBatchOptions, QueueSendOptions } from '@cloudflare/workers-types';
-import { SEMANTIC_ATTRIBUTE_SENTRY_OP, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, startSpan } from '@sentry/core';
+import type { Span } from '@sentry/core';
+import { getTraceData, SEMANTIC_ATTRIBUTE_SENTRY_OP, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, startSpan } from '@sentry/core';
+import { addQueueTraceContext } from '../../utils/queueTraceMeta';
 
 const ORIGIN = 'auto.faas.cloudflare.queue';
 
@@ -9,7 +11,7 @@ function startPublishSpan<T>(
     bodySize: number | undefined;
     messageCount?: number;
   },
-  callback: () => T,
+  callback: (span: Span) => T,
 ): T {
   const { bindingName, bodySize, messageCount } = options;
 
@@ -56,23 +58,46 @@ function getBodySize(body: unknown): number | undefined {
   }
 }
 
+function getBatchBodySize(messages: MessageSendRequest[]): number | undefined {
+  return messages.reduce<number | undefined>((acc, message) => {
+    const size = getBodySize(message.body);
+    if (size === undefined) {
+      return acc;
+    }
+    return (acc ?? 0) + size;
+  }, undefined);
+}
+
 /**
  * Wraps a Queue producer binding to create `queue.publish` spans on
  * `send` and `sendBatch` calls.
  *
  * The queue's own name is not available on the binding object, so we use
  * the env binding key (e.g. `MY_QUEUE`) as `messaging.destination.name`.
  */
-export function instrumentQueueProducer<T extends Queue>(queue: T, bindingName: string): T {
+export function instrumentQueueProducer<T extends Queue>(
+  queue: T,
+  bindingName: string,
+  enableTracePropagation = false,
+): T {
   return new Proxy(queue, {
     get(target, prop, receiver) {
       if (prop === 'send') {
         const original = Reflect.get(target, prop, receiver) as Queue['send'];
 
         return function (this: unknown, message: unknown, options?: QueueSendOptions): ReturnType<Queue['send']> {
-          return startPublishSpan({ bindingName, bodySize: getBodySize(message) }, () =>
-            Reflect.apply(original, target, [message, options]),
-          );
+          return startPublishSpan({ bindingName, bodySize: getBodySize(message) }, span => {
+            // Read trace data here so the carrier points to queue.publish, not its parent span.
+            const tracedMessage = enableTracePropagation
+              ? addQueueTraceContext(message, getTraceData()['sentry-trace'])
+              : message;
+
+            if (enableTracePropagation) {
+              span.setAttribute('messaging.message.body.size', getBodySize(tracedMessage));
+            }
+
+            return Reflect.apply(original, target, [tracedMessage, options]);
+          });
         };
       }
 
@@ -84,17 +109,25 @@ export function instrumentQueueProducer<T extends Queue>(queue: T, bindingName:
           options?: QueueSendBatchOptions,
         ): ReturnType<Queue['sendBatch']> {
           const messageArray = Array.from(messages);
-          const totalBodySize = messageArray.reduce<number | undefined>((acc, m) => {
-            const size = getBodySize(m.body);
-            if (size === undefined) {
-              return acc;
+          const totalBodySize = getBatchBodySize(messageArray);
+
+          return startPublishSpan({ bindingName, bodySize: totalBodySize, messageCount: messageArray.length }, span => {
+            if (!enableTracePropagation) {
+              return Reflect.apply(original, target, [messageArray, options]);
             }
-            return (acc ?? 0) + size;
-          }, undefined);
 
-          return startPublishSpan({ bindingName, bodySize: totalBodySize, messageCount: messageArray.length }, () =>
-            Reflect.apply(original, target, [messageArray, options]),
-          );
+            // Every entry was published by this batch span, so they intentionally share one producer context.
+            const sentryTrace = getTraceData()['sentry-trace'];
+
+            const tracedMessages = messageArray.map(message => ({
+              ...message,
+              body: addQueueTraceContext(message.body, sentryTrace),
+            }));
+
+            span.setAttribute('messaging.message.body.size', getBatchBodySize(tracedMessages));
+
+            return Reflect.apply(original, target, [tracedMessages, options]);
+          });
         };
       }