arthurfiorette · Jul 15, 2026 · Jul 15, 2026 · Jul 15, 2026 · Jul 15, 2026 · Oct 13, 2026
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -15,6 +15,10 @@
   - The plugins use the `sentry` CLI SDK instead of `@sentry/cli`. Sentry self-hosted 26.4.2 or higher is supported. Lower versions may continue to work.
   - Deploys that the plugins create on Vercel use the value of `VERCEL_TARGET_ENV` (`production`, `preview` or a custom name) as their environment, instead of `vercel-production` or `vercel-preview`. To keep the old value, set `release.deploy.env`.
 
+### Other Changes
+
+- feat(cloudflare): Add `queueTracePropagationBindings` to link Queue producer and consumer traces ([#25306](https://github.com/getsentry/sentry-javascript/pull/25306))
+
 ## 11.7.0
 
 ### Important Changes

@@ -10,6 +10,7 @@ export default Sentry.withSentry(
   (env: Env) => ({
     dsn: env.SENTRY_DSN,
     tracesSampleRate: 1,
+    queueTracePropagationBindings: ['MY_QUEUE'],
   }),
   {
     async fetch(request, env) {
@@ -41,6 +42,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');

@@ -1,4 +1,4 @@
-import type { Envelope } from '@sentry/core';
+import type { Envelope, SerializedStreamedSpan } from '@sentry/core';
 import { expect, it } from 'vitest';
 import { createRunner } from '../../runner';
 import { getSpansFromEnvelope } from '../../spanUtils';
@@ -30,11 +30,13 @@ 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 segment span on the consumer', async ({
   signal,
 }) => {
+  let publishSpan: SerializedStreamedSpan | undefined;
+  let processSpan: SerializedStreamedSpan | undefined;
   const runner = createRunner(__dirname)
     .unordered()
     .expect((envelope: Envelope) => {
       // The producer trace carries a `queue.publish` child span.
-      const publishSpan = getSpansFromEnvelope(envelope).find(span => span.name === 'send MY_QUEUE');
+      publishSpan = getSpansFromEnvelope(envelope).find(span => span.name === 'send MY_QUEUE');
 
       expect(publishSpan).toBeDefined();
       expect(publishSpan?.attributes).toEqual(
@@ -63,20 +65,30 @@ it('emits a queue.publish span on env.MY_QUEUE.send and a queue.process segment
           'messaging.operation.name': { type: 'string', value: 'process' },
           'messaging.batch.message_count': { type: 'integer', value: 1 },
           'faas.trigger': { type: 'string', value: 'pubsub' },
+          // The handler must not see the trace context the producer injected.
+          'test.queue.message_bodies': { type: 'string', value: '[{"payload":"hello"}]' },
         }),
       );
+      processSpan = segmentSpan;
     })
     .start(signal);
 
   await runner.makeRequest('post', '/enqueue/ok');
   await runner.completed();
+
+  expect(processSpan?.trace_id).not.toBe(publishSpan?.trace_id);
+  expect(processSpan?.links).toEqual([
+    expect.objectContaining({ trace_id: publishSpan?.trace_id, span_id: publishSpan?.span_id, sampled: true }),
+  ]);
 });
 
 it('emits a queue.publish span with batch attributes on env.MY_QUEUE.sendBatch', async ({ signal }) => {
+  let publishSpan: SerializedStreamedSpan | undefined;
+  let processSpan: SerializedStreamedSpan | undefined;
   const runner = createRunner(__dirname)
     .unordered()
     .expect((envelope: Envelope) => {
-      const publishSpan = getSpansFromEnvelope(envelope).find(span => span.name === 'send MY_QUEUE');
+      publishSpan = getSpansFromEnvelope(envelope).find(span => span.name === 'send MY_QUEUE');
 
       expect(publishSpan).toBeDefined();
       expect(publishSpan?.attributes).toEqual(
@@ -96,11 +108,21 @@ it('emits a queue.publish span with batch attributes on env.MY_QUEUE.sendBatch',
 
       expect(segmentSpan?.name).toBe('process test-queue');
       expect(segmentSpan?.attributes['messaging.batch.message_count']).toEqual({ type: 'integer', value: 3 });
+      expect(segmentSpan?.attributes['test.queue.message_bodies']).toEqual({
+        type: 'string',
+        value: '[{"payload":"one"},{"payload":"two"},{"payload":"three"}]',
+      });
+      processSpan = segmentSpan;
     })
     .start(signal);
 
   await runner.makeRequest('post', '/enqueue/batch');
   await runner.completed();
+
+  // All three messages come from one publish span, so the consumer links it once.
+  expect(processSpan?.links).toEqual([
+    expect.objectContaining({ trace_id: publishSpan?.trace_id, span_id: publishSpan?.span_id, sampled: true }),
+  ]);
 });
 
 it('calls non-instrumented Queue methods like metrics() on the underlying binding', async ({ signal }) => {

@@ -370,6 +370,34 @@ interface BaseCloudflareOptions {
    */
   rpcTracePropagationBindings?: TracePropagationTargets;
 
+  /**
+   * The Queue producer bindings on `env` that outgoing messages propagate trace context to.
+   *
+   * Cloudflare Queues has no message headers, so the SDK adds the producer's `sentry-trace` to
+   * plain-object message bodies under `__sentry_queue_trace__`. Arrays, binary data and other
+   * non-plain values are sent unchanged. A Sentry-instrumented consumer removes that field before
+   * your queue handler runs and links its `queue.process` span to every producer in the batch.
+   * Anywhere else the field arrives as part of the message body, so list only the queues whose
+   * consumer you know runs Sentry.
+   *
+   * Strings match a binding name exactly, regular expressions match by pattern. Both match
+   * case-insensitively.
+   *
+   * @default []
+   * @example
+   * ```ts
+   * // Propagate to `env.BILLING_QUEUE` and every `env.JOBS_*` binding
+   * export default Sentry.withSentry(
+   *   env => ({
+   *     dsn: env.SENTRY_DSN,
+   *     queueTracePropagationBindings: ['BILLING_QUEUE', /^JOBS_/],
+   *   }),
+   *   handler,
+   * );
+   * ```
+   */
+  queueTracePropagationBindings?: TracePropagationTargets;
+
   /**
    * Table names that should stay instrumented even though they match a reserved prefix used by
    * Durable Object frameworks for their internal SQLite tables: `cf_` (`agents`, `partyserver`, ...)

@@ -11,7 +11,7 @@ import {
 } from '../../utils/isBinding';
 import { instrumentD1 } from './instrumentD1';
 import { appendRpcMeta } from '../../utils/rpcMeta';
-import { createRpcPropagationResolver } from '../../utils/rpcPropagation';
+import { createBindingPropagationResolver, createRpcPropagationResolver } from '../../utils/rpcPropagation';
 import { instrumentDurableObjectNamespace, STUB_NON_RPC_METHODS } from '../instrumentDurableObjectNamespace';
 import { instrumentFetcher } from './instrumentFetcher';
 import { instrumentQueueProducer } from './instrumentQueueProducer';
@@ -22,6 +22,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
@@ -35,14 +36,15 @@ const instrumentedBindings = new WeakMap<object, unknown>();
  * - Workers AI (via `run` + `gateway` + `toMarkdown` duck-typing)
  *
  * @param env - The Cloudflare env object to instrument
- * @param options - Optional CloudflareOptions to control RPC trace propagation
+ * @param options - Optional CloudflareOptions to control RPC and Queue trace propagation
  */
 export function instrumentEnv<Env extends Record<string, unknown>>(env: Env, options?: CloudflareOptions): Env {
   if (!env || typeof env !== 'object') {
     return env;
   }
 
   const shouldPropagateRpcTrace = createRpcPropagationResolver(options);
+  const shouldPropagateQueueTrace = createBindingPropagationResolver(options?.queueTracePropagationBindings);
 
   return new Proxy(env, {
     get(target, prop, receiver) {
@@ -52,6 +54,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 = shouldPropagateQueueTrace(bindingName);
+        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) {
@@ -64,13 +82,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);

@@ -1,8 +1,10 @@
 import type { MessageBatch } from '@cloudflare/workers-types';
 import type { AnyExportedHandler } from '../../types';
+import { TraceFlags } from '@opentelemetry/api';
 import type { env as cloudflareEnv, WorkerEntrypoint } from 'cloudflare:workers';
 import { SENTRY_SEGMENT_NAME_SOURCE, SENTRY_OP, SENTRY_ORIGIN } from '@sentry/conventions/attributes';
 import { QUEUE_PROCESS } from '@sentry/conventions/op';
+import type { SpanLink } from '@sentry/core';
 import { captureException, startSpan, withIsolationScope } from '@sentry/core';
 import type { CloudflareOptions } from '../../client';
 import { flushAndDispose } from '../../flush';
@@ -12,8 +14,39 @@ import { addCloudResourceContext } from '../../scope-utils';
 import { init } from '../../sdk';
 import { instrumentContext } from '../../utils/instrumentContext';
 import { setInvocationState } from '../../utils/invocationContext';
+import { extractQueueTraceContext } from '../../utils/queueTraceMeta';
 import { instrumentEnv } from './instrumentEnv';
 
+function extractQueueTraceLinks(batch: MessageBatch): SpanLink[] {
+  const linkedSpans = new Set<string>();
+  const links: SpanLink[] = [];
+
+  for (const message of batch.messages) {
+    const producerContext = extractQueueTraceContext(message.body);
+    if (!producerContext) {
+      continue;
+    }
+
+    // `sendBatch` stamps every message with the same publish span, so link each producer only once.
+    const spanKey = `${producerContext.traceId}-${producerContext.spanId}`;
+    if (linkedSpans.has(spanKey)) {
+      continue;
+    }
+    linkedSpans.add(spanKey);
+
+    links.push({
+      context: {
+        traceId: producerContext.traceId,
+        spanId: producerContext.spanId,
+        traceFlags: producerContext.sampled ? TraceFlags.SAMPLED : TraceFlags.NONE,
+        isRemote: true,
+      },
+    });
+  }
+
+  return links;
+}
+
 /**
  * Core queue handler logic - wraps execution with Sentry instrumentation.
  */
@@ -36,6 +69,9 @@ function wrapQueueHandler(
     return startSpan(
       {
         name: `process ${batch.queue}`,
+        // A batch can mix messages from unrelated traces, so continuing any one of them would invent
+        // a false parent. Keep queue.process as a root span and link every producer instead.
+        links: extractQueueTraceLinks(batch),
         attributes: {
           [SENTRY_OP]: QUEUE_PROCESS,
           'faas.trigger': 'pubsub',

@@ -1,7 +1,8 @@
 import type { MessageSendRequest, Queue, QueueSendBatchOptions, QueueSendOptions } from '@cloudflare/workers-types';
 import { SENTRY_OP, SENTRY_ORIGIN } from '@sentry/conventions/attributes';
 import { QUEUE_PUBLISH } from '@sentry/conventions/op';
-import { startSpan } from '@sentry/core';
+import { getTraceData, startSpan, type Span } from '@sentry/core';
+import { addQueueTraceContext } from '../../utils/queueTraceMeta';
 
 const ORIGIN = 'auto.faas.cloudflare.queue';
 
@@ -11,7 +12,7 @@ function startPublishSpan<T>(
     bodySize: number | undefined;
     messageCount?: number;
   },
-  callback: () => T,
+  callback: (span: Span) => T,
 ): T {
   const { bindingName, bodySize, messageCount } = options;
 
@@ -57,23 +58,47 @@ 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]),
-          );
+          if (!enableTracePropagation) {
+            return startPublishSpan({ bindingName, bodySize: getBodySize(message) }, () =>
+              Reflect.apply(original, target, [message, options]),
+            );
+          }
+
+          // The body size is only known once the trace context is injected, which must happen inside
+          // the span so the carrier points to queue.publish rather than its parent.
+          return startPublishSpan({ bindingName, bodySize: undefined }, span => {
+            const tracedMessage = addQueueTraceContext(message, getTraceData()['sentry-trace']);
+            span.setAttribute('messaging.message.body.size', getBodySize(tracedMessage));
+            return Reflect.apply(original, target, [tracedMessage, options]);
+          });
         };
       }
 
@@ -85,17 +110,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;
-            }
-            return (acc ?? 0) + size;
-          }, undefined);
-
-          return startPublishSpan({ bindingName, bodySize: totalBodySize, messageCount: messageArray.length }, () =>
-            Reflect.apply(original, target, [messageArray, options]),
-          );
+          const messageCount = messageArray.length;
+
+          if (!enableTracePropagation) {
+            return startPublishSpan({ bindingName, bodySize: getBatchBodySize(messageArray), messageCount }, () =>
+              Reflect.apply(original, target, [messageArray, options]),
+            );
+          }
+
+          return startPublishSpan({ bindingName, bodySize: undefined, messageCount }, span => {
+            // Every message is published by this one span, so they all share its trace 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]);
+          });
         };
       }