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..08432b7dfe45 100644 --- a/dev-packages/cloudflare-integration-tests/suites/queue/test.ts +++ b/dev-packages/cloudflare-integration-tests/suites/queue/test.ts @@ -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 }) => { diff --git a/packages/cloudflare/src/client.ts b/packages/cloudflare/src/client.ts index ca9e75dbef64..8f3c1acea394 100644 --- a/packages/cloudflare/src/client.ts +++ b/packages/cloudflare/src/client.ts @@ -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`, ...) diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts b/packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts index 18cf562d95be..a3de0f67b24b 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts +++ b/packages/cloudflare/src/instrumentations/worker/instrumentEnv.ts @@ -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,7 +38,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') { @@ -46,6 +46,7 @@ export function instrumentEnv>(env: Env, opt } 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: 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; } diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts b/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts index 34e2d8bddd7c..f6339248726f 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts +++ b/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts @@ -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(); + 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', diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts b/packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts index 804cf3119d51..e2bfae75a4f7 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts +++ b/packages/cloudflare/src/instrumentations/worker/instrumentQueueProducer.ts @@ -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( bodySize: number | undefined; messageCount?: number; }, - callback: () => T, + callback: (span: Span) => T, ): T { const { bindingName, bodySize, messageCount } = options; @@ -66,6 +67,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. @@ -73,16 +84,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]); + }); }; } @@ -94,17 +119,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/rpcPropagation.ts b/packages/cloudflare/src/utils/bindingPropagation.ts similarity index 65% rename from packages/cloudflare/src/utils/rpcPropagation.ts rename to packages/cloudflare/src/utils/bindingPropagation.ts index d914b21032a9..5d3f38f5e264 100644 --- a/packages/cloudflare/src/utils/rpcPropagation.ts +++ b/packages/cloudflare/src/utils/bindingPropagation.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/src/utils/queueTraceMeta.ts b/packages/cloudflare/src/utils/queueTraceMeta.ts new file mode 100644 index 000000000000..c63fcfa04480 --- /dev/null +++ b/packages/cloudflare/src/utils/queueTraceMeta.ts @@ -0,0 +1,63 @@ +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; + } + + // Cloudflare rejects messages over 128 KB (https://developers.cloudflare.com/queues/platform/limits/), + // and the field adds about 80 bytes, so a payload right under the limit can start failing to send. + // Span links only need `sentry-trace`, so baggage would only cost more of that budget. 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/test/instrumentations/instrumentEnv.test.ts b/packages/cloudflare/test/instrumentations/instrumentEnv.test.ts index b5858d087008..fe1fe59ce1f4 100644 --- a/packages/cloudflare/test/instrumentations/instrumentEnv.test.ts +++ b/packages/cloudflare/test/instrumentations/instrumentEnv.test.ts @@ -223,6 +223,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('wraps Workers KV bindings in a proxy', async () => { const get = vi.fn().mockResolvedValue('value'); const put = vi.fn().mockResolvedValue(undefined); diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts index afe51790adfd..d58d1e5ff4fa 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts @@ -2,6 +2,7 @@ // Although this is not ideal, this is the best we can do until we have a better way to test cloudflare workers. import type { ExecutionContext, MessageBatch } from '@cloudflare/workers-types'; +import { SENTRY_LINK_TYPE } from '@sentry/conventions/attributes'; import type { Event } from '@sentry/core'; import * as SentryCore from '@sentry/core'; import { beforeEach, describe, expect, onTestFinished, test, vi } from 'vitest'; @@ -295,6 +296,108 @@ 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, + attributes: { [SENTRY_LINK_TYPE]: 'message_origin' }, + }, + { + trace_id: 'abcdefabcdefabcdefabcdefabcdefab', + span_id: 'fedcbafedcbafedc', + sampled: false, + attributes: { [SENTRY_LINK_TYPE]: 'message_origin' }, + }, + ]); + }); + + 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, + attributes: { [SENTRY_LINK_TYPE]: 'message_origin' }, + }, + ]); + }); }); 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..bdf85eab1b93 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentQueueProducer.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentQueueProducer.test.ts @@ -1,4 +1,5 @@ import type { Queue } from '@cloudflare/workers-types'; +import { MESSAGING_MESSAGE_BODY_SIZE } from '@sentry/conventions/attributes'; import * as SentryCore from '@sentry/core'; import { beforeEach, describe, expect, test, vi } from 'vitest'; import { instrumentQueueProducer } from '../../../src/instrumentations/worker/instrumentQueueProducer'; @@ -12,6 +13,7 @@ function createMockQueue(): Queue { describe('instrumentQueueProducer', () => { beforeEach(() => { + vi.restoreAllMocks(); vi.clearAllMocks(); }); @@ -26,6 +28,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 +203,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(); diff --git a/packages/cloudflare/test/utils/rpcPropagation.test.ts b/packages/cloudflare/test/utils/bindingPropagation.test.ts similarity index 96% rename from packages/cloudflare/test/utils/rpcPropagation.test.ts rename to packages/cloudflare/test/utils/bindingPropagation.test.ts index 047424e442ed..01840c3c2200 100644 --- a/packages/cloudflare/test/utils/rpcPropagation.test.ts +++ b/packages/cloudflare/test/utils/bindingPropagation.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from 'vitest'; -import { createRpcPropagationResolver } from '../../src/utils/rpcPropagation'; +import { createRpcPropagationResolver } from '../../src/utils/bindingPropagation'; describe('createRpcPropagationResolver', () => { it('propagates to nothing when no options are available', () => {