diff --git a/CHANGELOG.md b/CHANGELOG.md index 5571f47fb4eb..9bd071e3406b 100644 --- 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 diff --git a/dev-packages/cloudflare-integration-tests/suites/queue/index.ts b/dev-packages/cloudflare-integration-tests/suites/queue/index.ts index 44d20be6e939..eb47d5b981e6 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, + 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'); diff --git a/dev-packages/cloudflare-integration-tests/suites/queue/test.ts b/dev-packages/cloudflare-integration-tests/suites/queue/test.ts index 9093e6624998..a785ee9de3fc 100644 --- a/dev-packages/cloudflare-integration-tests/suites/queue/test.ts +++ b/dev-packages/cloudflare-integration-tests/suites/queue/test.ts @@ -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 }) => { diff --git a/packages/cloudflare/src/client.ts b/packages/cloudflare/src/client.ts index 3b4f591d5e22..9bb334a49cbf 100644 --- a/packages/cloudflare/src/client.ts +++ b/packages/cloudflare/src/client.ts @@ -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`, ...) diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts b/packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts index 33b02b9bd74f..a0b11eaea6a2 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts +++ b/packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts @@ -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(); +const instrumentedQueueBindings = new WeakMap>(); /** * Wraps the Cloudflare `env` object in a Proxy that detects binding types @@ -35,7 +36,7 @@ const instrumentedBindings = new WeakMap(); * - 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: Env, options?: CloudflareOptions): Env { if (!env || typeof env !== 'object') { @@ -43,6 +44,7 @@ export function instrumentEnv>(env: Env, opt } 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: 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(); + 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: 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); diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts b/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts index b12f01474d6a..84dde1e3da26 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts +++ b/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts @@ -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(); + 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', diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts b/packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts index 3460ea32c633..255f8d07d58b 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts +++ b/packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts @@ -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( 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. @@ -64,16 +75,30 @@ 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') { 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, [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(queue: T, bindingName: 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); - - 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]); + }); }; } diff --git a/packages/cloudflare/src/utils/queueTraceMeta.ts b/packages/cloudflare/src/utils/queueTraceMeta.ts new file mode 100644 index 000000000000..8e05c4b29ffa --- /dev/null +++ b/packages/cloudflare/src/utils/queueTraceMeta.ts @@ -0,0 +1,61 @@ +import { extractTraceparentData } from '@sentry/core'; + +const SENTRY_QUEUE_TRACE_KEY = '__sentry_queue_trace__'; + +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, sentryTrace: string | undefined): unknown { + if ( + !isQueueTraceCarrier(body) || + !sentryTrace || + Object.prototype.hasOwnProperty.call(body, SENTRY_QUEUE_TRACE_KEY) + ) { + return body; + } + + // Span links only need `sentry-trace`, so baggage would only cost message size. Copy instead of + // mutating, because callers may reuse a payload after enqueueing it. + return { ...body, [SENTRY_QUEUE_TRACE_KEY]: sentryTrace }; +} + +export function extractQueueTraceContext( + body: unknown, +): { traceId: string; spanId: string; sampled: boolean | undefined } | undefined { + if (!isQueueTraceCarrier(body)) { + return undefined; + } + + if (!Object.prototype.hasOwnProperty.call(body, SENTRY_QUEUE_TRACE_KEY)) { + return undefined; + } + + const sentryTrace = body[SENTRY_QUEUE_TRACE_KEY]; + + // The key is reserved transport metadata, so strip it even if its value is not a valid trace. + // Unlike `delete`, this returns false instead of throwing on frozen bodies, which keep the field. + Reflect.deleteProperty(body, SENTRY_QUEUE_TRACE_KEY); + + if (typeof sentryTrace !== 'string') { + return undefined; + } + + const traceparentData = extractTraceparentData(sentryTrace); + if (!traceparentData?.traceId || !traceparentData.parentSpanId) { + return undefined; + } + + return { + traceId: traceparentData.traceId, + spanId: traceparentData.parentSpanId, + sampled: traceparentData.parentSampled, + }; +} diff --git a/packages/cloudflare/src/utils/rpcPropagation.ts b/packages/cloudflare/src/utils/rpcPropagation.ts index d914b21032a9..5d3f38f5e264 100644 --- a/packages/cloudflare/src/utils/rpcPropagation.ts +++ b/packages/cloudflare/src/utils/rpcPropagation.ts @@ -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); +} diff --git a/packages/cloudflare/test/instrumentations/instrumentEnv.test.ts b/packages/cloudflare/test/instrumentations/instrumentEnv.test.ts index 9219091d53bd..4a0f81a9cfe9 100644 --- a/packages/cloudflare/test/instrumentations/instrumentEnv.test.ts +++ b/packages/cloudflare/test/instrumentations/instrumentEnv.test.ts @@ -222,6 +222,34 @@ describe('instrumentEnv', () => { expect(send.mock.calls[0]?.[0]).toBe('hello'); }); + it('propagates Queue trace context only to the configured bindings', async () => { + vi.spyOn(SentryCore, 'getTraceData').mockReturnValue({ + 'sentry-trace': '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + baggage: 'sentry-release=1.0.0', + }); + const billingSend = vi.fn().mockResolvedValue(undefined); + const otherSend = vi.fn().mockResolvedValue(undefined); + const instrumented = instrumentEnv( + { + BILLING_QUEUE: { send: billingSend, sendBatch: vi.fn() }, + OTHER_QUEUE: { send: otherSend, sendBatch: vi.fn() }, + }, + { queueTracePropagationBindings: ['billing_queue'] }, + ); + + await instrumented.BILLING_QUEUE.send({ invoiceId: 'inv_123' }); + await instrumented.OTHER_QUEUE.send({ invoiceId: 'inv_456' }); + + expect(billingSend).toHaveBeenCalledExactlyOnceWith( + { + invoiceId: 'inv_123', + __sentry_queue_trace__: '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + }, + undefined, + ); + expect(otherSend).toHaveBeenCalledExactlyOnceWith({ invoiceId: 'inv_456' }, undefined); + }); + it('caches the wrapped Queue binding across repeated access', () => { const queue = { send: vi.fn(), sendBatch: vi.fn() }; const env = { MY_QUEUE: queue }; @@ -230,6 +258,31 @@ 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 }, { queueTracePropagationBindings: ['MY_QUEUE'] }).MY_QUEUE; + + expect(first).not.toBe(second); + await first.send(message); + await second.send(message); + expect(send).toHaveBeenNthCalledWith(1, message, undefined); + expect(send).toHaveBeenNthCalledWith( + 2, + { + 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 afe51790adfd..a6aa941d16ff 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts @@ -295,6 +295,105 @@ 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_trace__: '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + }, + }, + { + ...originalBatch.messages[1]!, + body: { + invoiceId: 'inv_456', + __sentry_queue_trace__: 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', + }, + }, + { + ...originalBatch.messages[0]!, + id: '3', + body: { + invoiceId: 'inv_789', + __sentry_queue_trace__: '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + }, + }, + { + ...originalBatch.messages[1]!, + id: '4', + body: { + invoiceId: 'inv_invalid', + __sentry_queue_trace__: 'not-a-valid-trace', + }, + }, + ], + } 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 }), handler); + + await wrappedHandler.queue?.(batch, MOCK_ENV, createMockExecutionContext()); + + 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([ + { + trace_id: '1234567890abcdef1234567890abcdef', + span_id: '1234567890abcdef', + sampled: true, + }, + { + trace_id: 'abcdefabcdefabcdefabcdefabcdefab', + span_id: 'fedcbafedcbafedc', + sampled: false, + }, + ]); + }); + + test('does not throw when the transport context cannot be removed from a frozen body', async () => { + const originalBatch = createMockQueueBatch(); + const body = Object.freeze({ + invoiceId: 'inv_123', + __sentry_queue_trace__: '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + }); + const batch = { + ...originalBatch, + messages: [{ ...originalBatch.messages[0]!, body }], + } satisfies MessageBatch; + let processSpan: ReturnType | undefined; + const handler = { + queue(_batch, _env, _context) { + processSpan = SentryCore.spanToJSON(SentryCore.getActiveSpan()!); + }, + } satisfies ExportedHandler; + const wrappedHandler = withSentry(env => ({ dsn: env.SENTRY_DSN, tracesSampleRate: 1 }), handler); + + await wrappedHandler.queue?.(batch, MOCK_ENV, createMockExecutionContext()); + + expect(processSpan?.links).toEqual([ + { + trace_id: '1234567890abcdef1234567890abcdef', + span_id: '1234567890abcdef', + sampled: true, + }, + ]); + }); }); 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 f176d92e53db..cf8ce4a4cf29 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,110 @@ 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_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 = { + invoiceId: 'inv_123', + __sentry_queue_trace__: '1234567890abcdef1234567890abcdef-1234567890abcdef-1', + }; + expect(queue.send).toHaveBeenCalledExactlyOnceWith(sentBody, undefined); + expect(setAttribute).toHaveBeenCalledExactlyOnceWith( + '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', + }); + const queue = createMockQueue(); + const wrapped = instrumentQueueProducer(queue, 'BILLING_QUEUE', true); + const message = { + invoiceId: 'inv_123', + __sentry_queue_trace__: 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', + }; + + 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(); @@ -97,6 +202,69 @@ 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_trace__: 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', + }, + contentType: 'json', + }, + { + body: { + invoiceId: 'inv_456', + __sentry_queue_trace__: 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-0', + }, + delaySeconds: 10, + }, + ], + { delaySeconds: 5 }, + ); + }); + + 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 sentTrace = 'abcdefabcdefabcdefabcdefabcdefab-fedcbafedcbafedc-1'; + const sentMessages = [ + { body: { invoiceId: 'inv_123', __sentry_queue_trace__: sentTrace } }, + { body: { invoiceId: 'inv_456', __sentry_queue_trace__: sentTrace } }, + ]; + const expectedSize = sentMessages.reduce( + (total, message) => total + new TextEncoder().encode(JSON.stringify(message.body)).byteLength, + 0, + ); + expect(queue.sendBatch).toHaveBeenCalledExactlyOnceWith(sentMessages, undefined); + expect(setAttribute).toHaveBeenCalledExactlyOnceWith('messaging.message.body.size', expectedSize); + }); + test('starts a queue.publish span with batch attributes', async () => { const startSpanSpy = vi.spyOn(SentryCore, 'startSpan'); const queue = createMockQueue();