arthurfiorette · Jul 15, 2026 · Jul 15, 2026 · Jul 15, 2026 · Jul 15, 2026 · Oct 13, 2026
@@ -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,5 @@
-import type { Envelope } from '@sentry/core';
+import { SENTRY_LINK_TYPE } from '@sentry/conventions/attributes';
+import type { Envelope, SerializedStreamedSpan } from '@sentry/core';
 import { expect, it } from 'vitest';
 import { createRunner } from '../../runner';
 import { getSpansFromEnvelope } from '../../spanUtils';
@@ -30,11 +31,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 +66,35 @@ 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([
+    {
+      trace_id: publishSpan?.trace_id,
+      span_id: publishSpan?.span_id,
+      sampled: true,
+      attributes: { [SENTRY_LINK_TYPE]: { type: 'string', value: 'message_origin' } },
+    },
+  ]);
 });
 
 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 +114,26 @@ 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([
+    {
+      trace_id: publishSpan?.trace_id,
+      span_id: publishSpan?.span_id,
+      sampled: true,
+      attributes: { [SENTRY_LINK_TYPE]: { type: 'string', value: 'message_origin' } },
+    },
+  ]);
 });
 
 it('calls non-instrumented Queue methods like metrics() on the underlying binding', async ({ signal }) => {

@@ -410,6 +410,37 @@ 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.
+   *
+   * The field adds about 80 bytes to each message. Cloudflare rejects messages over 128 KB, so a
+   * message close to that limit can fail to send once propagation is enabled.
+   *
+   * 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`, ...)

@@ -12,7 +12,7 @@ import {
 } from '../../utils/isBinding';
 import { instrumentD1 } from './instrumentD1';
 import { appendRpcMeta } from '../../utils/rpcMeta';
-import { createRpcPropagationResolver } from '../../utils/rpcPropagation';
+import { createBindingPropagationResolver, createRpcPropagationResolver } from '../../utils/bindingPropagation';
 import { instrumentDurableObjectNamespace, STUB_NON_RPC_METHODS } from '../instrumentDurableObjectNamespace';
 import { instrumentFetcher } from './instrumentFetcher';
 import { instrumentKV } from './instrumentKV';
@@ -38,14 +38,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) {
@@ -69,7 +70,7 @@ export function instrumentEnv<Env extends Record<string, unknown>>(env: Env, opt
 
       if (isQueue(item)) {
         const bindingName = typeof prop === 'string' ? prop : String(prop);
-        const instrumented = instrumentQueueProducer(item, bindingName);
+        const instrumented = instrumentQueueProducer(item, bindingName, shouldPropagateQueueTrace(bindingName));
         instrumentedBindings.set(item, instrumented);
         return instrumented;
       }

@@ -1,7 +1,9 @@
 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_LINK_TYPE,
   SENTRY_SEGMENT_NAME_SOURCE,
   SENTRY_OP,
   SENTRY_ORIGIN,
@@ -14,6 +16,7 @@ import {
   MESSAGING_SYSTEM,
 } 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';
@@ -23,8 +26,41 @@ 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,
+      },
+      attributes: {
+        [SENTRY_LINK_TYPE]: 'message_origin',
+      },
+    });
+  }
+
+  return links;
+}
+
 /**
  * Core queue handler logic - wraps execution with Sentry instrumentation.
  */
@@ -47,6 +83,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',

@@ -10,7 +10,8 @@ import {
   MESSAGING_SYSTEM,
 } 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';
 
@@ -20,7 +21,7 @@ function startPublishSpan<T>(
     bodySize: number | undefined;
     messageCount?: number;
   },
-  callback: () => T,
+  callback: (span: Span) => T,
 ): T {
   const { bindingName, bodySize, messageCount } = options;
 
@@ -66,23 +67,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]);
+          });
         };
       }
 
@@ -94,17 +119,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]);
+          });
         };
       }
 

@@ -1,18 +1,14 @@
-import { matchesTracePropagationTargets } from '@sentry/core';
+import { matchesTracePropagationTargets, type TracePropagationTargets } from '@sentry/core';
 import type { CloudflareOptions } from '../client';
 
 const PROPAGATE_TO_NONE = () => false;
 
 /**
- * Builds the per-binding predicate that decides whether a binding takes part in RPC trace
- * propagation.
- *
- * Callers only. Receivers continue an incoming trace whenever one arrives, so they have nothing to
- * match against.
+ * Builds the per-binding predicate that decides whether a binding takes part in trace propagation.
  */
-export function createRpcPropagationResolver(options: CloudflareOptions | undefined): (bindingName: string) => boolean {
-  const bindings = options?.rpcTracePropagationBindings;
-
+export function createBindingPropagationResolver(
+  bindings: TracePropagationTargets | undefined,
+): (bindingName: string) => boolean {
   if (!bindings?.length) {
     return PROPAGATE_TO_NONE;
   }
@@ -21,3 +17,14 @@ export function createRpcPropagationResolver(options: CloudflareOptions | undefi
   // propagation for a binding named `MY_DB`. Regular expressions still give pattern matching.
   return (bindingName: string) => matchesTracePropagationTargets(bindingName, bindings, true);
 }
+
+/**
+ * Builds the per-binding predicate that decides whether a binding takes part in RPC trace
+ * propagation.
+ *
+ * Callers only. Receivers continue an incoming trace whenever one arrives, so they have nothing to
+ * match against.
+ */
+export function createRpcPropagationResolver(options: CloudflareOptions | undefined): (bindingName: string) => boolean {
+  return createBindingPropagationResolver(options?.rpcTracePropagationBindings);
+}