From 0b05f24e072e2b3baf3b9497cf488255f5d1acce Mon Sep 17 00:00:00 2001 From: Arthur Fiorette Date: Wed, 15 Jul 2026 15:25:21 -0300 Subject: [PATCH 1/3] code --- packages/cloudflare/src/client.ts | 13 ++ .../instrumentations/worker/instrumentEnv.ts | 2 +- .../worker/instrumentQueue.ts | 30 ++++ .../worker/instrumentQueueProducer.ts | 31 +++- .../cloudflare/src/utils/queueTraceMeta.ts | 60 ++++++++ .../instrumentations/instrumentEnv.test.ts | 24 ++++ .../worker/instrumentQueue.test.ts | 62 ++++++++ .../worker/instrumentQueueProducer.test.ts | 132 ++++++++++++++++++ 8 files changed, 347 insertions(+), 7 deletions(-) create mode 100644 packages/cloudflare/src/utils/queueTraceMeta.ts diff --git a/packages/cloudflare/src/client.ts b/packages/cloudflare/src/client.ts index 21be86bf4c59..0631b9b86873 100644 --- a/packages/cloudflare/src/client.ts +++ b/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_meta__`. 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. * diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts b/packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts index 5899ee66ada5..f90bf8dea048 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts +++ b/packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts @@ -57,7 +57,7 @@ export function instrumentEnv>(env: Env, opt if (isQueue(item)) { const bindingName = typeof prop === 'string' ? prop : String(prop); - const instrumented = instrumentQueueProducer(item, bindingName); + const instrumented = instrumentQueueProducer(item, bindingName, options?.enableQueueTracePropagation === true); instrumentedBindings.set(item, instrumented); return instrumented; } diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts b/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts index a99b81c6c341..9ccecd4014df 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts +++ b/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,30 @@ 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[] { + return batch.messages.flatMap(message => { + const producerContext = extractQueueTraceContext(message.body); + if (!producerContext) { + return []; + } + + 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 +59,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, diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts b/packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts index 40b9f404e3bc..a55777bc8d1b 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts +++ b/packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts @@ -1,5 +1,6 @@ import type { MessageSendRequest, Queue, QueueSendBatchOptions, QueueSendOptions } from '@cloudflare/workers-types'; -import { SEMANTIC_ATTRIBUTE_SENTRY_OP, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, startSpan } 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'; @@ -63,7 +64,11 @@ function getBodySize(body: unknown): number | undefined { * 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(queue: T, bindingName: string): T { +export function instrumentQueueProducer( + queue: T, + bindingName: string, + enableTracePropagation = false, +): T { return new Proxy(queue, { get(target, prop, receiver) { if (prop === 'send') { @@ -71,7 +76,11 @@ export function instrumentQueueProducer(queue: T, bindingName: return function (this: unknown, message: unknown, options?: QueueSendOptions): ReturnType { return startPublishSpan({ bindingName, bodySize: getBodySize(message) }, () => - Reflect.apply(original, target, [message, options]), + Reflect.apply(original, target, [ + // Read trace data here so the carrier points to queue.publish, not its parent span. + enableTracePropagation ? addQueueTraceContext(message, getTraceData()) : message, + options, + ]), ); }; } @@ -92,9 +101,19 @@ export function instrumentQueueProducer(queue: T, bindingName: return (acc ?? 0) + size; }, undefined); - return startPublishSpan({ bindingName, bodySize: totalBodySize, messageCount: messageArray.length }, () => - Reflect.apply(original, target, [messageArray, options]), - ); + return startPublishSpan({ bindingName, bodySize: totalBodySize, messageCount: messageArray.length }, () => { + if (!enableTracePropagation) { + return Reflect.apply(original, target, [messageArray, options]); + } + + // Every entry was published by this batch span, so they intentionally share one producer context. + const traceData = getTraceData(); + const tracedMessages = messageArray.map(message => ({ + ...message, + body: addQueueTraceContext(message.body, traceData), + })); + return Reflect.apply(original, target, [tracedMessages, options]); + }); }; } diff --git a/packages/cloudflare/src/utils/queueTraceMeta.ts b/packages/cloudflare/src/utils/queueTraceMeta.ts new file mode 100644 index 000000000000..84480c792c2b --- /dev/null +++ b/packages/cloudflare/src/utils/queueTraceMeta.ts @@ -0,0 +1,60 @@ +import type { SerializedTraceData } from '@sentry/core'; +import { extractTraceparentData } from '@sentry/core'; + +const SENTRY_QUEUE_TRACE_KEY = '__sentry_queue_meta__'; + +function isQueueTraceCarrier(body: unknown): body is Record { + if (typeof body !== 'object' || body === null || Array.isArray(body)) { + return false; + } + + // Cloudflare Queues has no message-header API. Only record-like bodies can carry a named field; binary data and + // other structured-clone values must pass through untouched. + const prototype = Object.getPrototypeOf(body); + return prototype === Object.prototype || prototype === null; +} + +export function addQueueTraceContext(body: unknown, traceData: SerializedTraceData): unknown { + if ( + !isQueueTraceCarrier(body) || + !traceData['sentry-trace'] || + Object.prototype.hasOwnProperty.call(body, SENTRY_QUEUE_TRACE_KEY) + ) { + return body; + } + + // Avoid mutating application-owned values; callers may reuse a payload after enqueueing it. + return { ...body, [SENTRY_QUEUE_TRACE_KEY]: traceData }; +} + +export function extractQueueTraceContext( + body: unknown, +): { traceId: string; spanId: string; sampled: boolean | undefined } | undefined { + if (!isQueueTraceCarrier(body)) { + return undefined; + } + + const traceData = body[SENTRY_QUEUE_TRACE_KEY]; + if (typeof traceData !== 'object' || traceData === null) { + return undefined; + } + + const sentryTrace = Reflect.get(traceData, 'sentry-trace'); + if (typeof sentryTrace !== 'string') { + return undefined; + } + + const traceparentData = extractTraceparentData(sentryTrace); + if (!traceparentData?.traceId || !traceparentData.parentSpanId) { + return undefined; + } + + // Transport metadata must not leak into domain validation or application contracts. + delete body.__sentry_queue_meta__; + + return { + traceId: traceparentData.traceId, + spanId: traceparentData.parentSpanId, + sampled: traceparentData.parentSampled, + }; +} diff --git a/packages/cloudflare/test/instrumentations/instrumentEnv.test.ts b/packages/cloudflare/test/instrumentations/instrumentEnv.test.ts index 9309284abb41..76f81874e236 100644 --- a/packages/cloudflare/test/instrumentations/instrumentEnv.test.ts +++ b/packages/cloudflare/test/instrumentations/instrumentEnv.test.ts @@ -231,6 +231,30 @@ describe('instrumentEnv', () => { expect(send.mock.calls[0]?.[0]).toBe('hello'); }); + it('enables Queue trace propagation only when configured', async () => { + vi.spyOn(SentryCore, 'getTraceData').mockReturnValue({ + 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + baggage: 'sentry-release=1.0.0', + }); + const send = vi.fn().mockResolvedValue(undefined); + const queue = { send, sendBatch: vi.fn().mockResolvedValue(undefined) }; + const instrumented = instrumentEnv({ BILLING_QUEUE: queue }, { enableQueueTracePropagation: true }); + + await instrumented.BILLING_QUEUE.send({ invoiceId: 'inv_123' }); + + expect(send).toHaveBeenCalledTimes(1); + expect(send).toHaveBeenLastCalledWith( + { + invoiceId: 'inv_123', + __sentry_queue_meta__: { + 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + baggage: 'sentry-release=1.0.0', + }, + }, + undefined, + ); + }); + it('caches the wrapped Queue binding across repeated access', () => { const queue = { send: vi.fn(), sendBatch: vi.fn() }; const env = { MY_QUEUE: queue }; diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts index 999073012b92..876ae1616a19 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts @@ -291,6 +291,68 @@ describe('instrumentQueue', () => { trace_id: expect.stringMatching(/[a-f0-9]{32}/), }); }); + + test('links producer traces and removes transport context before invoking the handler', async () => { + const originalBatch = createMockQueueBatch(); + const batch = { + ...originalBatch, + messages: [ + { + ...originalBatch.messages[0]!, + body: { + invoiceId: 'inv_123', + __sentry_queue_meta__: { + 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + baggage: 'sentry-release=1.0.0', + }, + }, + }, + { + ...originalBatch.messages[1]!, + body: { + invoiceId: 'inv_456', + __sentry_queue_meta__: { + 'sentry-trace': 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', + baggage: 'sentry-environment=production', + }, + }, + }, + ], + } satisfies MessageBatch; + let receivedBodies: unknown[] = []; + let processSpan: ReturnType | undefined; + const handler = { + queue(receivedBatch, _env, _context) { + receivedBodies = receivedBatch.messages.map(message => message.body); + processSpan = SentryCore.spanToJSON(SentryCore.getActiveSpan()!); + }, + } satisfies ExportedHandler; + const wrappedHandler = withSentry( + env => ({ dsn: env.SENTRY_DSN, tracesSampleRate: 1, enableQueueTracePropagation: true }), + handler, + ); + + await wrappedHandler.queue?.(batch, MOCK_ENV, createMockExecutionContext()); + + expect(receivedBodies).toEqual([{ invoiceId: 'inv_123' }, { invoiceId: 'inv_456' }]); + expect(processSpan?.parent_span_id).toBeUndefined(); + expect(processSpan?.links).toEqual([ + { + trace_id: '1234567890abcdef1234567890abcdef', + span_id: '1234567890abcdef', + sampled: true, + isRemote: true, + attributes: { 'sentry.link.type': 'previous_trace' }, + }, + { + trace_id: 'abcdefabcdefabcdefabcdefabcdefab', + span_id: 'fedcbafedcbafedc', + sampled: false, + isRemote: true, + attributes: { 'sentry.link.type': 'previous_trace' }, + }, + ]); + }); }); test('flush must be called when all waitUntil are done', async () => { diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentQueueProducer.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentQueueProducer.test.ts index b094641984ce..40df3f33f314 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentQueueProducer.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentQueueProducer.test.ts @@ -12,6 +12,7 @@ function createMockQueue(): Queue { describe('instrumentQueueProducer', () => { beforeEach(() => { + vi.restoreAllMocks(); vi.clearAllMocks(); }); @@ -26,6 +27,93 @@ describe('instrumentQueueProducer', () => { expect(queue.send).toHaveBeenLastCalledWith({ hello: 'world' }, { contentType: 'json' }); }); + test('does not add trace context by default', async () => { + vi.spyOn(SentryCore, 'getTraceData').mockReturnValue({ + 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + }); + const queue = createMockQueue(); + const wrapped = instrumentQueueProducer(queue, 'BILLING_QUEUE'); + const message = { invoiceId: 'inv_123' }; + + await wrapped.send(message); + + expect(queue.send).toHaveBeenCalledTimes(1); + expect(queue.send).toHaveBeenLastCalledWith(message, undefined); + }); + + test('adds trace context to object payloads', async () => { + vi.spyOn(SentryCore, 'getTraceData').mockReturnValue({ + 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + baggage: 'sentry-release=1.0.0', + }); + const queue = createMockQueue(); + const wrapped = instrumentQueueProducer(queue, 'BILLING_QUEUE', true); + const message = { invoiceId: 'inv_123' }; + + await wrapped.send(message); + + expect(queue.send).toHaveBeenCalledTimes(1); + expect(queue.send).toHaveBeenLastCalledWith( + { + invoiceId: 'inv_123', + __sentry_queue_meta__: { + 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + baggage: 'sentry-release=1.0.0', + }, + }, + undefined, + ); + expect(message).toEqual({ invoiceId: 'inv_123' }); + }); + + test('preserves an existing metadata field', async () => { + vi.spyOn(SentryCore, 'getTraceData').mockReturnValue({ + 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + }); + const queue = createMockQueue(); + const wrapped = instrumentQueueProducer(queue, 'BILLING_QUEUE', true); + const message = { + invoiceId: 'inv_123', + __sentry_queue_meta__: { + 'sentry-trace': 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', + baggage: 'billing-field=value', + }, + }; + + await wrapped.send(message); + + expect(queue.send).toHaveBeenCalledTimes(1); + expect(queue.send).toHaveBeenLastCalledWith(message, undefined); + }); + + test('does not add trace context to array payloads', async () => { + vi.spyOn(SentryCore, 'getTraceData').mockReturnValue({ + 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + }); + const queue = createMockQueue(); + const wrapped = instrumentQueueProducer(queue, 'BILLING_QUEUE', true); + const message = [{ invoiceId: 'inv_123' }]; + + await wrapped.send(message); + + expect(queue.send).toHaveBeenCalledTimes(1); + expect(queue.send).toHaveBeenLastCalledWith(message, undefined); + }); + + test('does not add trace context to binary payloads', async () => { + vi.spyOn(SentryCore, 'getTraceData').mockReturnValue({ + 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + }); + const queue = createMockQueue(); + const wrapped = instrumentQueueProducer(queue, 'BILLING_QUEUE', true); + const message = new ArrayBuffer(42); + + await wrapped.send(message); + + expect(queue.send).toHaveBeenCalledTimes(1); + expect(queue.send).toHaveBeenLastCalledWith(message, undefined); + }); + test('starts a queue.publish span with messaging attributes', async () => { const startSpanSpy = vi.spyOn(SentryCore, 'startSpan'); const queue = createMockQueue(); @@ -98,6 +186,50 @@ describe('instrumentQueueProducer', () => { expect(queue.sendBatch).toHaveBeenCalledTimes(1); }); + test('adds the publish span context to every object body', async () => { + vi.spyOn(SentryCore, 'getTraceData').mockReturnValue({ + 'sentry-trace': 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', + baggage: 'sentry-environment=production', + }); + const queue = createMockQueue(); + const wrapped = instrumentQueueProducer(queue, 'BILLING_QUEUE', true); + + await wrapped.sendBatch( + [ + { body: { invoiceId: 'inv_123' }, contentType: 'json' }, + { body: { invoiceId: 'inv_456' }, delaySeconds: 10 }, + ], + { delaySeconds: 5 }, + ); + + expect(queue.sendBatch).toHaveBeenCalledTimes(1); + expect(queue.sendBatch).toHaveBeenLastCalledWith( + [ + { + body: { + invoiceId: 'inv_123', + __sentry_queue_meta__: { + 'sentry-trace': 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', + baggage: 'sentry-environment=production', + }, + }, + contentType: 'json', + }, + { + body: { + invoiceId: 'inv_456', + __sentry_queue_meta__: { + 'sentry-trace': 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', + baggage: 'sentry-environment=production', + }, + }, + delaySeconds: 10, + }, + ], + { delaySeconds: 5 }, + ); + }); + test('starts a queue.publish span with batch attributes', async () => { const startSpanSpy = vi.spyOn(SentryCore, 'startSpan'); const queue = createMockQueue(); From 0bd582c466a709e52889b7f9bd89f15439ddc9d4 Mon Sep 17 00:00:00 2001 From: Arthur Fiorette Date: Wed, 15 Jul 2026 15:41:41 -0300 Subject: [PATCH 2/3] fix review comments --- .../suites/queue/index.ts | 6 +++ .../suites/queue/test.ts | 46 ++++++++++++++++--- .../worker/instrumentQueue.ts | 8 ++++ .../worker/instrumentQueue.test.ts | 13 +++++- 4 files changed, 66 insertions(+), 7 deletions(-) diff --git a/dev-packages/cloudflare-integration-tests/suites/queue/index.ts b/dev-packages/cloudflare-integration-tests/suites/queue/index.ts index e06560ccf690..e672997b798b 100644 --- a/dev-packages/cloudflare-integration-tests/suites/queue/index.ts +++ b/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'); diff --git a/dev-packages/cloudflare-integration-tests/suites/queue/test.ts b/dev-packages/cloudflare-integration-tests/suites/queue/test.ts index f0886ba3f37c..5dc93433766c 100644 --- a/dev-packages/cloudflare-integration-tests/suites/queue/test.ts +++ b/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 | undefined; + let consumerTrace: Record | 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 | undefined; + let consumerTrace: Record | 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' }, + }, + ]); }); diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts b/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts index 9ccecd4014df..8ccb3fe0867d 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts +++ b/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts @@ -22,12 +22,20 @@ import { extractQueueTraceContext } from '../../utils/queueTraceMeta'; import { instrumentEnv } from './instrumentEnv'; function extractQueueTraceLinks(batch: MessageBatch): SpanLink[] { + const linkedSpans = new Set(); + 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: { diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts index 876ae1616a19..542dc7d6d055 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts @@ -317,6 +317,17 @@ describe('instrumentQueue', () => { }, }, }, + { + ...originalBatch.messages[0]!, + id: '3', + body: { + invoiceId: 'inv_789', + __sentry_queue_meta__: { + 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + baggage: 'sentry-release=1.0.0', + }, + }, + }, ], } satisfies MessageBatch; let receivedBodies: unknown[] = []; @@ -334,7 +345,7 @@ describe('instrumentQueue', () => { await wrappedHandler.queue?.(batch, MOCK_ENV, createMockExecutionContext()); - expect(receivedBodies).toEqual([{ invoiceId: 'inv_123' }, { invoiceId: 'inv_456' }]); + expect(receivedBodies).toEqual([{ invoiceId: 'inv_123' }, { invoiceId: 'inv_456' }, { invoiceId: 'inv_789' }]); expect(processSpan?.parent_span_id).toBeUndefined(); expect(processSpan?.links).toEqual([ { From fe0b73000f7ef224a5c6e3a53a47a9dfd7635d8a Mon Sep 17 00:00:00 2001 From: Arthur Fiorette Date: Wed, 15 Jul 2026 16:17:51 -0300 Subject: [PATCH 3/3] fix reviews --- packages/cloudflare/src/client.ts | 2 +- .../instrumentations/worker/instrumentEnv.ts | 24 ++++--- .../worker/instrumentQueueProducer.ts | 50 +++++++++------ .../cloudflare/src/utils/queueTraceMeta.ts | 23 +++---- .../instrumentations/instrumentEnv.test.ts | 31 +++++++-- .../worker/instrumentQueue.test.ts | 30 +++++---- .../worker/instrumentQueueProducer.test.ts | 63 ++++++++++++++----- 7 files changed, 152 insertions(+), 71 deletions(-) diff --git a/packages/cloudflare/src/client.ts b/packages/cloudflare/src/client.ts index 0631b9b86873..61c98f83c735 100644 --- a/packages/cloudflare/src/client.ts +++ b/packages/cloudflare/src/client.ts @@ -214,7 +214,7 @@ interface BaseCloudflareOptions { * 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_meta__`. The consumer removes this field before invoking the queue handler. + * 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. diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts b/packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts index f90bf8dea048..6e9456bc4815 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts +++ b/packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts @@ -14,6 +14,7 @@ function isProxyable(item: unknown): item is object { } const instrumentedBindings = new WeakMap(); +const instrumentedQueueBindings = new WeakMap>(); /** * Wraps the Cloudflare `env` object in a Proxy that detects binding types @@ -43,6 +44,22 @@ export function instrumentEnv>(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(); + 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: Env, opt return instrumented; } - if (isQueue(item)) { - const bindingName = typeof prop === 'string' ? prop : String(prop); - const instrumented = instrumentQueueProducer(item, bindingName, options?.enableQueueTracePropagation === true); - instrumentedBindings.set(item, instrumented); - return instrumented; - } - if (isR2Bucket(item)) { const bindingName = typeof prop === 'string' ? prop : String(prop); const instrumented = instrumentR2Bucket(item, bindingName); diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts b/packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts index a55777bc8d1b..fd86bc39278e 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts +++ b/packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts @@ -1,4 +1,5 @@ import type { MessageSendRequest, Queue, QueueSendBatchOptions, QueueSendOptions } from '@cloudflare/workers-types'; +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'; @@ -10,7 +11,7 @@ function startPublishSpan( bodySize: number | undefined; messageCount?: number; }, - callback: () => T, + callback: (span: Span) => T, ): T { const { bindingName, bodySize, messageCount } = options; @@ -57,6 +58,16 @@ function getBodySize(body: unknown): number | undefined { } } +function getBatchBodySize(messages: MessageSendRequest[]): number | undefined { + return messages.reduce((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. @@ -75,13 +86,18 @@ export function instrumentQueueProducer( const original = Reflect.get(target, prop, receiver) as Queue['send']; return function (this: unknown, message: unknown, options?: QueueSendOptions): ReturnType { - return startPublishSpan({ bindingName, bodySize: getBodySize(message) }, () => - Reflect.apply(original, target, [ - // Read trace data here so the carrier points to queue.publish, not its parent span. - enableTracePropagation ? addQueueTraceContext(message, getTraceData()) : 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]); + }); }; } @@ -93,25 +109,23 @@ export function instrumentQueueProducer( options?: QueueSendBatchOptions, ): ReturnType { const messageArray = Array.from(messages); - const totalBodySize = messageArray.reduce((acc, m) => { - const size = getBodySize(m.body); - if (size === undefined) { - return acc; - } - return (acc ?? 0) + size; - }, undefined); + const totalBodySize = getBatchBodySize(messageArray); - return startPublishSpan({ bindingName, bodySize: totalBodySize, messageCount: messageArray.length }, () => { + return startPublishSpan({ bindingName, bodySize: totalBodySize, messageCount: messageArray.length }, span => { if (!enableTracePropagation) { return Reflect.apply(original, target, [messageArray, options]); } // Every entry was published by this batch span, so they intentionally share one producer context. - const traceData = getTraceData(); + const sentryTrace = getTraceData()['sentry-trace']; + const tracedMessages = messageArray.map(message => ({ ...message, - body: addQueueTraceContext(message.body, traceData), + body: addQueueTraceContext(message.body, sentryTrace), })); + + span.setAttribute('messaging.message.body.size', getBatchBodySize(tracedMessages)); + return Reflect.apply(original, target, [tracedMessages, options]); }); }; diff --git a/packages/cloudflare/src/utils/queueTraceMeta.ts b/packages/cloudflare/src/utils/queueTraceMeta.ts index 84480c792c2b..73435515f6a2 100644 --- a/packages/cloudflare/src/utils/queueTraceMeta.ts +++ b/packages/cloudflare/src/utils/queueTraceMeta.ts @@ -1,7 +1,6 @@ -import type { SerializedTraceData } from '@sentry/core'; import { extractTraceparentData } from '@sentry/core'; -const SENTRY_QUEUE_TRACE_KEY = '__sentry_queue_meta__'; +const SENTRY_QUEUE_TRACE_KEY = '__sentry_queue_trace__'; function isQueueTraceCarrier(body: unknown): body is Record { if (typeof body !== 'object' || body === null || Array.isArray(body)) { @@ -14,17 +13,19 @@ function isQueueTraceCarrier(body: unknown): body is Record { return prototype === Object.prototype || prototype === null; } -export function addQueueTraceContext(body: unknown, traceData: SerializedTraceData): unknown { +export function addQueueTraceContext(body: unknown, sentryTrace: string | undefined): unknown { if ( !isQueueTraceCarrier(body) || - !traceData['sentry-trace'] || + !sentryTrace || Object.prototype.hasOwnProperty.call(body, SENTRY_QUEUE_TRACE_KEY) ) { return body; } + // Queue messages are capped at 128 KB, but size is not the main reason to keep this minimal: span links only need + // sentry-trace, so baggage would be unused even when typical JSON payloads have ample headroom. // Avoid mutating application-owned values; callers may reuse a payload after enqueueing it. - return { ...body, [SENTRY_QUEUE_TRACE_KEY]: traceData }; + return { ...body, [SENTRY_QUEUE_TRACE_KEY]: sentryTrace }; } export function extractQueueTraceContext( @@ -34,12 +35,15 @@ export function extractQueueTraceContext( return undefined; } - const traceData = body[SENTRY_QUEUE_TRACE_KEY]; - if (typeof traceData !== 'object' || traceData === null) { + if (!Object.prototype.hasOwnProperty.call(body, SENTRY_QUEUE_TRACE_KEY)) { return undefined; } - const sentryTrace = Reflect.get(traceData, 'sentry-trace'); + const sentryTrace = body[SENTRY_QUEUE_TRACE_KEY]; + + // Once propagation is enabled this key is reserved transport metadata, even if its value was corrupted in transit. + delete body.__sentry_queue_trace__; + if (typeof sentryTrace !== 'string') { return undefined; } @@ -49,9 +53,6 @@ export function extractQueueTraceContext( return undefined; } - // Transport metadata must not leak into domain validation or application contracts. - delete body.__sentry_queue_meta__; - return { traceId: traceparentData.traceId, spanId: traceparentData.parentSpanId, diff --git a/packages/cloudflare/test/instrumentations/instrumentEnv.test.ts b/packages/cloudflare/test/instrumentations/instrumentEnv.test.ts index 76f81874e236..8b0abb1c645e 100644 --- a/packages/cloudflare/test/instrumentations/instrumentEnv.test.ts +++ b/packages/cloudflare/test/instrumentations/instrumentEnv.test.ts @@ -246,10 +246,7 @@ describe('instrumentEnv', () => { expect(send).toHaveBeenLastCalledWith( { invoiceId: 'inv_123', - __sentry_queue_meta__: { - 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', - baggage: 'sentry-release=1.0.0', - }, + __sentry_queue_trace__: '1234567890abcdef1234567890abcdef-1234567890abcdef-1', }, undefined, ); @@ -263,6 +260,32 @@ describe('instrumentEnv', () => { expect(instrumented.MY_QUEUE).toBe(instrumented.MY_QUEUE); }); + it('caches Queue bindings separately by propagation setting', async () => { + vi.spyOn(SentryCore, 'getTraceData').mockReturnValue({ + 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + }); + const send = vi.fn().mockResolvedValue(undefined); + const queue = { send, sendBatch: vi.fn() }; + const message = { invoiceId: 'inv_123' }; + + const first = instrumentEnv({ MY_QUEUE: queue }).MY_QUEUE; + const second = instrumentEnv({ MY_QUEUE: queue }, { enableQueueTracePropagation: true }).MY_QUEUE; + + expect(first).not.toBe(second); + await first.send(message); + await second.send(message); + expect(send.mock.calls).toEqual([ + [message, undefined], + [ + { + invoiceId: 'inv_123', + __sentry_queue_trace__: '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + }, + undefined, + ], + ]); + }); + it('wraps Queue bindings independently from DO bindings', () => { const queue = { send: vi.fn(), sendBatch: vi.fn() }; const doNamespace = { diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts index 542dc7d6d055..4c22a618f742 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts @@ -301,20 +301,14 @@ describe('instrumentQueue', () => { ...originalBatch.messages[0]!, body: { invoiceId: 'inv_123', - __sentry_queue_meta__: { - 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', - baggage: 'sentry-release=1.0.0', - }, + __sentry_queue_trace__: '1234567890abcdef1234567890abcdef-1234567890abcdef-1', }, }, { ...originalBatch.messages[1]!, body: { invoiceId: 'inv_456', - __sentry_queue_meta__: { - 'sentry-trace': 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', - baggage: 'sentry-environment=production', - }, + __sentry_queue_trace__: 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', }, }, { @@ -322,10 +316,15 @@ describe('instrumentQueue', () => { id: '3', body: { invoiceId: 'inv_789', - __sentry_queue_meta__: { - 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', - baggage: 'sentry-release=1.0.0', - }, + __sentry_queue_trace__: '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + }, + }, + { + ...originalBatch.messages[1]!, + id: '4', + body: { + invoiceId: 'inv_invalid', + __sentry_queue_trace__: 'not-a-valid-trace', }, }, ], @@ -345,7 +344,12 @@ describe('instrumentQueue', () => { await wrappedHandler.queue?.(batch, MOCK_ENV, createMockExecutionContext()); - expect(receivedBodies).toEqual([{ invoiceId: 'inv_123' }, { invoiceId: 'inv_456' }, { invoiceId: 'inv_789' }]); + expect(receivedBodies).toEqual([ + { invoiceId: 'inv_123' }, + { invoiceId: 'inv_456' }, + { invoiceId: 'inv_789' }, + { invoiceId: 'inv_invalid' }, + ]); expect(processSpan?.parent_span_id).toBeUndefined(); expect(processSpan?.links).toEqual([ { diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentQueueProducer.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentQueueProducer.test.ts index 40df3f33f314..a01baaffdf3c 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentQueueProducer.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentQueueProducer.test.ts @@ -1,4 +1,4 @@ -import type { Queue } from '@cloudflare/workers-types'; +import type { MessageSendRequest, Queue } from '@cloudflare/workers-types'; import * as SentryCore from '@sentry/core'; import { beforeEach, describe, expect, test, vi } from 'vitest'; import { instrumentQueueProducer } from '../../../src/instrumentations/worker/instrumentQueueProducer'; @@ -56,16 +56,33 @@ describe('instrumentQueueProducer', () => { expect(queue.send).toHaveBeenLastCalledWith( { invoiceId: 'inv_123', - __sentry_queue_meta__: { - 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', - baggage: 'sentry-release=1.0.0', - }, + __sentry_queue_trace__: '1234567890abcdef1234567890abcdef-1234567890abcdef-1', }, undefined, ); expect(message).toEqual({ invoiceId: 'inv_123' }); }); + test('includes injected trace context in body size', async () => { + const setAttribute = vi.fn(); + vi.spyOn(SentryCore, 'startSpan').mockImplementation(((_options, callback) => + callback({ setAttribute } as unknown as SentryCore.Span)) as typeof SentryCore.startSpan); + vi.spyOn(SentryCore, 'getTraceData').mockReturnValue({ + 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + }); + const queue = createMockQueue(); + const wrapped = instrumentQueueProducer(queue, 'BILLING_QUEUE', true); + + await wrapped.send({ invoiceId: 'inv_123' }); + + const sentBody = vi.mocked(queue.send).mock.calls[0]![0]; + expect(setAttribute).toHaveBeenCalledTimes(1); + expect(setAttribute).toHaveBeenCalledWith( + 'messaging.message.body.size', + new TextEncoder().encode(JSON.stringify(sentBody)).byteLength, + ); + }); + test('preserves an existing metadata field', async () => { vi.spyOn(SentryCore, 'getTraceData').mockReturnValue({ 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', @@ -74,10 +91,7 @@ describe('instrumentQueueProducer', () => { const wrapped = instrumentQueueProducer(queue, 'BILLING_QUEUE', true); const message = { invoiceId: 'inv_123', - __sentry_queue_meta__: { - 'sentry-trace': 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', - baggage: 'billing-field=value', - }, + __sentry_queue_trace__: 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', }; await wrapped.send(message); @@ -208,20 +222,14 @@ describe('instrumentQueueProducer', () => { { body: { invoiceId: 'inv_123', - __sentry_queue_meta__: { - 'sentry-trace': 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', - baggage: 'sentry-environment=production', - }, + __sentry_queue_trace__: 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', }, contentType: 'json', }, { body: { invoiceId: 'inv_456', - __sentry_queue_meta__: { - 'sentry-trace': 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', - baggage: 'sentry-environment=production', - }, + __sentry_queue_trace__: 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', }, delaySeconds: 10, }, @@ -230,6 +238,27 @@ describe('instrumentQueueProducer', () => { ); }); + test('includes injected trace context in batch body size', async () => { + const setAttribute = vi.fn(); + vi.spyOn(SentryCore, 'startSpan').mockImplementation(((_options, callback) => + callback({ setAttribute } as unknown as SentryCore.Span)) as typeof SentryCore.startSpan); + vi.spyOn(SentryCore, 'getTraceData').mockReturnValue({ + 'sentry-trace': 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-1', + }); + const queue = createMockQueue(); + const wrapped = instrumentQueueProducer(queue, 'BILLING_QUEUE', true); + + await wrapped.sendBatch([{ body: { invoiceId: 'inv_123' } }, { body: { invoiceId: 'inv_456' } }]); + + const sentMessages = vi.mocked(queue.sendBatch).mock.calls[0]![0] as MessageSendRequest[]; + const expectedSize = sentMessages.reduce( + (total, message) => total + new TextEncoder().encode(JSON.stringify(message.body)).byteLength, + 0, + ); + expect(setAttribute).toHaveBeenCalledTimes(1); + expect(setAttribute).toHaveBeenCalledWith('messaging.message.body.size', expectedSize); + }); + test('starts a queue.publish span with batch attributes', async () => { const startSpanSpy = vi.spyOn(SentryCore, 'startSpan'); const queue = createMockQueue();