From e3b086e01f2fe38b5e902f757f4219e3524923f5 Mon Sep 17 00:00:00 2001 From: JPeer264 Date: Fri, 7 Aug 2026 10:37:03 +0200 Subject: [PATCH 1/3] feat(cloudflare): Add cacheClient to reuse the client across invocations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Building and disposing a client per invocation costs real time on every request, and in a Durable Object it also loses data: there is no `waitUntil` boundary that dependably extends execution, so anything captured after the handler returned went to a client that had already been disposed. Enabled by default, this caches one client per isolate. The first initialization wins for the isolate's lifetime: a later init with different options reuses that client, and a new deployment always starts fresh isolates, so clients are always built from the current version's options. A cached client is flushed but not disposed at an invocation boundary, and it is re-bound to the current scope on every invocation — otherwise `initialScope` would apply only to an isolate's first invocation, and a client disposed by a competing init would keep being handed out. A cached client whose transport is gone is evicted rather than returned. Because a reused client never reaches an end-of-invocation flush, delivery is eager: the new `afterEnvelope` hook on the core client drains the transport buffer as soon as an envelope has been accepted, and logs and metrics drain on a debounced hook so they are batched rather than sent one at a time. Spans that end after the invocation's flush point are delivered through core's `flushTraceSpans` hook, which flushes only that trace's bucket from the span streaming buffer. The per-invocation flush lock and span tracking are skipped, since binding a client that outlives the invocation to one invocation's lock would make later flushes wait on that invocation's work forever. A shared client also shares integration state, so dedupe works across invocations: the same error raised by two separate requests is reported only once. Uncached behavior is unchanged; pass `cacheClient: false` to restore it. Co-authored-by: Cursor --- .../suites/cache-client/index.ts | 257 +++++++++ .../suites/cache-client/test.ts | 321 +++++++++++ .../suites/cache-client/wrangler.jsonc | 18 + packages/cloudflare/src/baseSdk.ts | 8 +- packages/cloudflare/src/client.ts | 303 +++++++++-- packages/cloudflare/src/clientCache.ts | 23 + packages/cloudflare/src/flush.ts | 14 +- .../worker/instrumentEmail.ts | 3 + .../worker/instrumentQueue.ts | 3 + .../worker/instrumentScheduled.ts | 3 + .../instrumentations/worker/instrumentTail.ts | 3 + packages/cloudflare/src/request.ts | 2 +- packages/cloudflare/src/sdk.ts | 46 +- packages/cloudflare/src/transport.ts | 2 +- .../cloudflare/src/utils/invocationContext.ts | 72 +++ .../cloudflare/src/utils/invocationScope.ts | 11 +- .../cloudflare/src/wrapMethodWithSentry.ts | 5 +- packages/cloudflare/test/client.test.ts | 514 ++++++++++++++++++ packages/cloudflare/test/flush.test.ts | 13 + .../worker/instrumentEmail.test.ts | 4 +- .../worker/instrumentFetch.test.ts | 4 +- .../worker/instrumentQueue.test.ts | 4 +- .../worker/instrumentScheduled.test.ts | 4 +- .../worker/instrumentTail.test.ts | 4 +- packages/cloudflare/test/request.test.ts | 193 +++++++ packages/cloudflare/test/sdk.test.ts | 207 ++++++- packages/cloudflare/test/testUtils.ts | 2 + .../test/utils/invocationContext.test.ts | 76 +++ packages/cloudflare/test/workflow.test.ts | 26 +- 29 files changed, 2089 insertions(+), 56 deletions(-) create mode 100644 dev-packages/cloudflare-integration-tests/suites/cache-client/index.ts create mode 100644 dev-packages/cloudflare-integration-tests/suites/cache-client/test.ts create mode 100644 dev-packages/cloudflare-integration-tests/suites/cache-client/wrangler.jsonc create mode 100644 packages/cloudflare/src/clientCache.ts create mode 100644 packages/cloudflare/src/utils/invocationContext.ts create mode 100644 packages/cloudflare/test/utils/invocationContext.test.ts diff --git a/dev-packages/cloudflare-integration-tests/suites/cache-client/index.ts b/dev-packages/cloudflare-integration-tests/suites/cache-client/index.ts new file mode 100644 index 000000000000..1ef5f389fcd3 --- /dev/null +++ b/dev-packages/cloudflare-integration-tests/suites/cache-client/index.ts @@ -0,0 +1,257 @@ +import * as Sentry from '@sentry/cloudflare'; +import { DurableObject } from 'cloudflare:workers'; + +interface Env { + SENTRY_DSN: string; + CACHE_DO: DurableObjectNamespace; + NO_CACHE_DO: DurableObjectNamespace; +} + +/** + * Sync KV and SQL work against the DO's own storage, which the SDK instruments into `db` spans. + * Used to check that those spans still reach the transport from inside a Durable Object, where a + * cached client never hits an invocation-boundary flush and has to rely on the eager drain. + */ +function runStorageOps(ctx: DurableObjectState): { listSize: number; rows: number } { + ctx.storage.kv.put('cache-key', { hello: 'sync' }); + ctx.storage.kv.get('cache-key'); + const entries = [...ctx.storage.kv.list()]; + ctx.storage.kv.delete('cache-key'); + + ctx.storage.sql.exec('CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)'); + ctx.storage.sql.exec('INSERT INTO users (name) VALUES (?)', 'Alice'); + const rows = ctx.storage.sql.exec('SELECT * FROM users').toArray(); + + return { listSize: entries.length, rows: rows.length }; +} + +function startDetachedWork(message: string): string { + void (async () => { + await new Promise(r => setTimeout(r, 3000)); + await Sentry.startSpan({ name: 'do.detached-task', op: 'task' }, async () => { + Sentry.logger.info(`Detached log: ${message}`); + Sentry.metrics.count('do.detached', 1); + Sentry.captureException(new Error(message)); + }); + })(); + return `Detached work started: ${message}`; +} + +// DO with cacheClient: true (the default) — detached work events SHOULD be captured +class CacheDurableObjectBase extends DurableObject { + async echo(n: number): Promise { + return n; + } + + async handlerError(instanceId: string): Promise { + throw new Error(`Cache DO handler error from ${instanceId}`); + } + + async dedupe(): Promise { + Sentry.captureException(new Error('Same error')); + return 'dedupe test'; + } + + async scopeCheck(seed: boolean): Promise { + if (seed) { + Sentry.setTag('seeded_tag', 'from-seeding-call'); + Sentry.setUser({ id: 'user-from-seeding-call' }); + } + Sentry.captureException(new Error(seed ? 'Cache scope seed' : 'Cache scope probe')); + return 'ok'; + } + + async storage(): Promise { + const { listSize, rows } = runStorageOps(this.ctx); + return `cache storage ${listSize}/${rows}`; + } + + async fetch(request: Request): Promise { + const url = new URL(request.url); + if (url.pathname === '/detached') { + return new Response(startDetachedWork(`Detached work from cache DO ${url.searchParams.get('id')}`)); + } + if (url.pathname === '/streaming') { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('chunk1')); + controller.enqueue(new TextEncoder().encode('chunk2')); + controller.close(); + }, + }); + return new Response(stream, { headers: { 'content-type': 'text/event-stream' } }); + } + return new Response('Cache DO'); + } +} + +// DO with cacheClient: false — detached work events should NOT be captured +class NoCacheDurableObjectBase extends DurableObject { + async handlerError(instanceId: string): Promise { + throw new Error(`No-cache DO handler error from ${instanceId}`); + } + + async dedupe(): Promise { + Sentry.captureException(new Error('Same error')); + return 'dedupe test'; + } + + async storage(): Promise { + const { listSize, rows } = runStorageOps(this.ctx); + return `no-cache storage ${listSize}/${rows}`; + } + + async fetch(request: Request): Promise { + const url = new URL(request.url); + if (url.pathname === '/detached') { + return new Response(startDetachedWork(`Detached work from no-cache DO ${url.searchParams.get('id')}`)); + } + return new Response('No-cache DO'); + } +} + +export const CacheDurableObject = Sentry.instrumentDurableObjectWithSentry( + (env: Env) => ({ + dsn: env.SENTRY_DSN, + tracesSampleRate: 1, + enableLogs: true, + enableRpcTracePropagation: true, + }), + CacheDurableObjectBase, +); + +export const NoCacheDurableObject = Sentry.instrumentDurableObjectWithSentry( + (env: Env) => ({ + dsn: env.SENTRY_DSN, + tracesSampleRate: 1, + cacheClient: false, + enableRpcTracePropagation: true, + }), + NoCacheDurableObjectBase, +); + +export default Sentry.withSentry( + (env: Env) => ({ + dsn: env.SENTRY_DSN, + tracesSampleRate: 1, + enableLogs: true, + enableRpcTracePropagation: true, + }), + { + async fetch(request, env, ctx) { + const url = new URL(request.url); + const instanceId = url.searchParams.get('id') || 'default'; + + // Work that finishes AFTER the response: a post-response span tree plus a + // log, metric and error, all registered via waitUntil. This is the worker-side + // half of the #22545 lifecycle (the DO-side half is /detached). + if (url.pathname === '/post-response') { + ctx.waitUntil( + Sentry.startSpan({ name: 'checkout.post-response', op: 'task' }, async () => { + Sentry.logger.info('checkout post-response log'); + Sentry.metrics.count('checkout.processed', 1); + await new Promise(r => setTimeout(r, 50)); + await Sentry.startSpan({ name: 'checkout.notify-webhook', op: 'http.client' }, async () => { + await new Promise(r => setTimeout(r, 25)); + Sentry.captureException(new Error('Webhook delivery failed')); + }); + }), + ); + return new Response('checkout accepted'); + } + + // Fan a single request out into N sequential DO RPC calls — every RPC span must + // land in this request's trace when RPC trace propagation is on. + if (url.pathname === '/burst') { + const n = Math.min(Number(url.searchParams.get('n')) || 1, 20); + const stub = env.CACHE_DO.get( + env.CACHE_DO.idFromName(`burst-${instanceId}`), + ) as DurableObjectStub; + + let sum = 0; + for (let i = 0; i < n; i++) { + sum += (await stub.echo(i)) as number; + } + return Response.json({ calls: n, sum }); + } + + // Cache DO RPC calls + if (url.pathname === '/cache/handler-error') { + const stub = env.CACHE_DO.get( + env.CACHE_DO.idFromName(`cache-do-${instanceId}`), + ) as DurableObjectStub; + await stub.handlerError(instanceId); + } + + if (url.pathname === '/cache/dedupe') { + const stub = env.CACHE_DO.get( + env.CACHE_DO.idFromName(`cache-do-${instanceId}`), + ) as DurableObjectStub; + const result = await stub.dedupe(); + return new Response(String(result)); + } + + if (url.pathname === '/cache/scope') { + const stub = env.CACHE_DO.get( + env.CACHE_DO.idFromName(`cache-do-${instanceId}`), + ) as DurableObjectStub; + return new Response(await stub.scopeCheck(url.searchParams.get('seed') === '1')); + } + + // Cache DO fetch calls — detached work goes through fetch (matching the #22545 repro), + // since the DO fetch handler always initializes the DO's own client + if (url.pathname === '/cache/detached') { + const stub = env.CACHE_DO.get( + env.CACHE_DO.idFromName(`cache-do-${instanceId}`), + ) as DurableObjectStub; + return stub.fetch(new Request(`http://do/detached?id=${instanceId}`)); + } + + if (url.pathname === '/cache/storage') { + const stub = env.CACHE_DO.get( + env.CACHE_DO.idFromName(`cache-do-${instanceId}`), + ) as DurableObjectStub; + return new Response(await stub.storage()); + } + + if (url.pathname === '/cache/streaming') { + const stub = env.CACHE_DO.get( + env.CACHE_DO.idFromName(`cache-do-${instanceId}`), + ) as DurableObjectStub; + return stub.fetch(new Request('http://do/streaming')); + } + + // No-cache DO calls + if (url.pathname === '/no-cache/handler-error') { + const stub = env.NO_CACHE_DO.get( + env.NO_CACHE_DO.idFromName(`no-cache-do-${instanceId}`), + ) as DurableObjectStub; + await stub.handlerError(instanceId); + } + + if (url.pathname === '/no-cache/dedupe') { + const stub = env.NO_CACHE_DO.get( + env.NO_CACHE_DO.idFromName(`no-cache-do-${instanceId}`), + ) as DurableObjectStub; + const result = await stub.dedupe(); + return new Response(String(result)); + } + + if (url.pathname === '/no-cache/storage') { + const stub = env.NO_CACHE_DO.get( + env.NO_CACHE_DO.idFromName(`no-cache-do-${instanceId}`), + ) as DurableObjectStub; + return new Response(await stub.storage()); + } + + if (url.pathname === '/no-cache/detached') { + const stub = env.NO_CACHE_DO.get( + env.NO_CACHE_DO.idFromName(`no-cache-do-${instanceId}`), + ) as DurableObjectStub; + return stub.fetch(new Request(`http://do/detached?id=${instanceId}`)); + } + + return new Response('Hello World!'); + }, + } satisfies ExportedHandler, +); diff --git a/dev-packages/cloudflare-integration-tests/suites/cache-client/test.ts b/dev-packages/cloudflare-integration-tests/suites/cache-client/test.ts new file mode 100644 index 000000000000..dde6864391c6 --- /dev/null +++ b/dev-packages/cloudflare-integration-tests/suites/cache-client/test.ts @@ -0,0 +1,321 @@ +import type { Envelope, Event } from '@sentry/core'; +import { describe, expect, it } from 'vitest'; +import { createRunner } from '../../runner'; + +type Mechanism = { type: string; handled: boolean }; + +/** + * Matches an error event by exception value and capture mechanism. + * + * Callback-style (instead of exact `eventEnvelope` matching) because Durable Object + * RPC events carry no `request` and their trace context varies with propagation — + * only the exception payload is stable. Non-matching envelopes (worker-side duplicate + * captures, transactions) are dropped by the runner's unordered mode. + */ +function errorEventExpectation(value: string, mechanism: Mechanism) { + return (envelope: Envelope) => { + const event = envelope[1]?.[0]?.[1] as Event; + expect(event).toEqual( + expect.objectContaining({ + level: 'error', + exception: { + values: [ + expect.objectContaining({ + type: 'Error', + value, + stacktrace: { frames: expect.any(Array) }, + mechanism, + }), + ], + }, + }), + ); + }; +} + +const DO_MECHANISM: Mechanism = { type: 'auto.faas.cloudflare.durable_object', handled: false }; +// Direct `captureException` calls (not routed through a wrapped handler) always get this mechanism +const CAPTURE_MECHANISM: Mechanism = { type: 'generic', handled: true }; + +/** span-v2 streamed envelope payload. */ +type SpanV2Payload = { + items?: Array<{ + name?: string; + trace_id?: string; + attributes?: Record; + }>; +}; + +/** + * Matches a span-v2 envelope containing at least one span with the given name. + * Spans are matched loosely because batching can coalesce multiple spans of one + * trace into a single envelope. + */ +function spanEnvelopeExpectation(spanName: string) { + return (envelope: Envelope) => { + const payload = envelope[1]?.[0]?.[1] as SpanV2Payload; + expect(payload.items?.some(span => span.name === spanName)).toBe(true); + }; +} + +it('cacheClient: false - DO handler error is captured', async ({ signal }) => { + const runner = createRunner(__dirname) + .expect(errorEventExpectation('No-cache DO handler error from instance-1', DO_MECHANISM)) + .expect(errorEventExpectation('No-cache DO handler error from instance-2', DO_MECHANISM)) + .unordered() + .start(signal); + + await runner.makeRequest('get', '/no-cache/handler-error?id=instance-1', { expectError: true }); + await runner.makeRequest('get', '/no-cache/handler-error?id=instance-2', { expectError: true }); + await runner.completed(); +}); + +it('cacheClient: true - DO handler error is captured', async ({ signal }) => { + const runner = createRunner(__dirname) + .expect(errorEventExpectation('Cache DO handler error from instance-1', DO_MECHANISM)) + .expect(errorEventExpectation('Cache DO handler error from instance-2', DO_MECHANISM)) + .unordered() + .start(signal); + + await runner.makeRequest('get', '/cache/handler-error?id=instance-1', { expectError: true }); + await runner.makeRequest('get', '/cache/handler-error?id=instance-2', { expectError: true }); + await runner.completed(); +}); + +it('cacheClient: true - detached work events ARE captured', async ({ signal }) => { + const runner = createRunner(__dirname) + .expect(errorEventExpectation('Detached work from cache DO instance-1', CAPTURE_MECHANISM)) + .expect(errorEventExpectation('Detached work from cache DO instance-2', CAPTURE_MECHANISM)) + // Logs batch client-side and the idle drain timer is disabled for this runtime, so a + // log only ever becomes an envelope if the cached client drains its log buffer on + // capture. Without that, detached logs are silently dropped while errors still arrive. + .expect((envelope: Envelope) => { + const payload = envelope[1]?.[0]?.[1] as { items?: Array<{ body?: string }> }; + expect(payload.items?.some(log => log.body?.startsWith('Detached log: Detached work from cache DO'))).toBe(true); + }) + // The detached span itself: captured in work that starts after the RPC invocation + // settled, so it only survives because the cached client delivers it eagerly. + .expect(spanEnvelopeExpectation('do.detached-task')) + // The detached metric, delivered via the same eager drain as the span and log. + .expect((envelope: Envelope) => { + const payload = envelope[1]?.[0]?.[1] as { items?: Array<{ name?: string }> }; + expect(payload.items?.some(metric => metric.name === 'do.detached')).toBe(true); + }) + .unordered() + .start(signal); + + await runner.makeRequest('get', '/cache/detached?id=instance-1'); + await runner.makeRequest('get', '/cache/detached?id=instance-2'); + await runner.completed(); +}); + +it('cacheClient: false - repro #22545: detached work events are silently dropped', async ({ signal }) => { + const runner = createRunner(__dirname).ignore('transaction', 'span').start(signal); + + // Make the request that spawns detached work + await runner.makeRequest('get', '/no-cache/detached?id=repro-1'); + + // With cacheClient: false, the client is disposed after the handler returns, + // so the detached work's captureException (3s later) is silently dropped. + // We verify by waiting for the event with a timeout — if it doesn't arrive, + // the event was silently dropped as expected. + const result = await Promise.race([ + runner.makeRequestAndWaitForEnvelope('get', '/no-cache/detached?id=repro-2', () => { + throw new Error('Received an event that should have been dropped with cacheClient: false'); + }), + // Timeout: resolve with 'timeout' if no event arrives within 5s + new Promise(resolve => setTimeout(() => resolve('timeout'), 5000)), + ]); + + // The event should NOT have been received (timeout should win the race) + expect(result).toBe('timeout'); +}); + +it('cacheClient: true - dedupe drops the same error across invocations', async ({ signal }) => { + // A shared client shares its dedupe state, so the same error captured by two separate + // invocations is reported only once — the second is dropped as a duplicate. + const runner = createRunner(__dirname).ignore('transaction', 'span').start(signal); + + await runner.makeRequestAndWaitForEnvelope( + 'get', + '/cache/dedupe?id=dedupe-shared', + errorEventExpectation('Same error', CAPTURE_MECHANISM), + ); + + // Second and third invocations capture the same error, but dedupe drops them. + await runner.makeRequest('get', '/cache/dedupe?id=dedupe-shared'); + await runner.makeRequest('get', '/cache/dedupe?id=dedupe-shared'); +}); + +it('cacheClient: false - dedupe does not persist across invocations', async ({ signal }) => { + // A fresh client per invocation means fresh dedupe state, so each invocation reports + // the same error independently. + const runner = createRunner(__dirname).ignore('transaction', 'span').start(signal); + + for (let i = 0; i < 3; i++) { + await runner.makeRequestAndWaitForEnvelope( + 'get', + '/no-cache/dedupe?id=dedupe-fresh', + errorEventExpectation('Same error', CAPTURE_MECHANISM), + ); + } +}); + +// A cached client outlives the invocation that created it, so this checks that reusing it does not +// also start reusing the isolation scope `setTag`/`setUser` write to. The uncached counterpart of +// this test lives in the `durable-object-scope` suite. +it('cacheClient: true - two consecutive invocations get different isolation scopes', async ({ signal }) => { + const runner = createRunner(__dirname).ignore('transaction', 'span').start(signal); + + await runner.makeRequestAndWaitForEnvelope('get', '/cache/scope?id=scope-shared&seed=1', (envelope: Envelope) => { + const event = envelope[1]?.[0]?.[1] as Event; + expect(event.exception?.values?.[0]?.value).toBe('Cache scope seed'); + // Guards the probe assertions below against passing vacuously. + expect(event.tags).toEqual(expect.objectContaining({ seeded_tag: 'from-seeding-call' })); + expect(event.user).toEqual({ id: 'user-from-seeding-call' }); + }); + + await runner.makeRequestAndWaitForEnvelope('get', '/cache/scope?id=scope-shared&seed=0', (envelope: Envelope) => { + const event = envelope[1]?.[0]?.[1] as Event; + expect(event.exception?.values?.[0]?.value).toBe('Cache scope probe'); + expect(event.tags?.seeded_tag).toBeUndefined(); + expect(event.user).toBeUndefined(); + }); +}); + +it('cacheClient: true - streaming response works with shared client', async ({ signal }) => { + // A streamed request produces two span envelopes — the DO's own `GET /streaming` and the + // outer worker's `GET /cache/streaming` — and their arrival order is not guaranteed. Accept + // either, since the point is that spans still reach the transport at all. + const streamingSpanExpectation = (envelope: Envelope) => { + const payload = envelope[1]?.[0]?.[1] as { items?: Array<{ name?: string }> }; + expect(payload.items?.map(span => span.name)).toEqual( + expect.arrayContaining([expect.stringMatching(/^GET \/(cache\/)?streaming$/)]), + ); + }; + + const runner = createRunner(__dirname).start(signal); + + // Waiting per request keeps the runner alive for the second one: with both + // expectations queued up front it completes on the first request's spans and + // tears down before the second is sent. + for (let i = 0; i < 2; i++) { + const text = await runner.makeRequestAndWaitForEnvelope( + 'get', + '/cache/streaming', + streamingSpanExpectation, + ); + expect(text).toBe('chunk1chunk2'); + } +}); + +// Sync KV and SQL instrumentation produces child `db` spans inside the Durable Object. A cached +// client never reaches an invocation-boundary flush, so these only arrive if the eager drain +// covers spans too — the uncached mode is the control that the routes themselves are sound. +describe('durable object storage spans', () => { + // span-v2 wraps every attribute value as `{ value, type }`. + type SpanV2 = { name?: string; attributes?: Record }; + + const dbSpanNames = (envelope: Envelope): string[] => { + const payload = envelope[1]?.[0]?.[1] as { items?: SpanV2[] }; + return (payload.items ?? []) + .filter(span => span.attributes?.['db.system.name']?.value === 'cloudflare-durable-object-sql') + .map(span => span.name ?? ''); + }; + + for (const mode of ['cache', 'no-cache'] as const) { + it(`cacheClient: ${mode === 'cache'} - db spans are delivered`, async ({ signal }) => { + // The DO's span envelope and the outer worker's arrive in either order, so match + // unordered rather than asserting on whichever comes first. + const runner = createRunner(__dirname) + .expect((envelope: Envelope) => { + expect(dbSpanNames(envelope)).toEqual([ + 'durable_object_storage_kv_put', + 'durable_object_storage_kv_get', + 'durable_object_storage_kv_list', + 'durable_object_storage_kv_delete', + 'CREATE TABLE users', + 'INSERT users', + 'SELECT users', + ]); + }) + .unordered() + .start(signal); + + await runner.makeRequest('get', `/${mode}/storage?id=storage-${mode}`); + await runner.completed(); + }); + } +}); + +it('cacheClient: true - multiple DO instances share the same client', async ({ signal }) => { + const runner = createRunner(__dirname) + .expect(errorEventExpectation('Cache DO handler error from instance-1', DO_MECHANISM)) + .expect(errorEventExpectation('Cache DO handler error from instance-2', DO_MECHANISM)) + .unordered() + .start(signal); + + // Two different DO instances — both should capture errors + await runner.makeRequest('get', '/cache/handler-error?id=instance-1', { expectError: true }); + await runner.makeRequest('get', '/cache/handler-error?id=instance-2', { expectError: true }); + await runner.completed(); +}); + +// The worker-side half of #22545: work registered via ctx.waitUntil finishes after +// the response and after the invocation's flush point, so the spans/log/metric/error +// are only delivered because the cached client drains them eagerly. +it('cacheClient: true - post-response waitUntil work delivers spans, log, metric and error', async ({ signal }) => { + const runner = createRunner(__dirname) + .expect(spanEnvelopeExpectation('checkout.post-response')) + .expect(spanEnvelopeExpectation('checkout.notify-webhook')) + .expect(errorEventExpectation('Webhook delivery failed', CAPTURE_MECHANISM)) + .expect((envelope: Envelope) => { + const payload = envelope[1]?.[0]?.[1] as { items?: Array<{ body?: string }> }; + expect(payload.items?.some(log => log.body === 'checkout post-response log')).toBe(true); + }) + .expect((envelope: Envelope) => { + const payload = envelope[1]?.[0]?.[1] as { items?: Array<{ name?: string }> }; + expect(payload.items?.some(metric => metric.name === 'checkout.processed')).toBe(true); + }) + .unordered() + .start(signal); + + const text = await runner.makeRequest('get', '/post-response?id=checkout-1'); + expect(text).toBe('checkout accepted'); + await runner.completed(); +}); + +// One request fans out into N sequential DO RPC calls. Each RPC span must be +// delivered and must belong to the worker request's trace (RPC trace propagation). +it('cacheClient: true - burst DO RPC span shares the worker request trace', async ({ signal }) => { + let workerTraceId: string | undefined; + const echoTraceIds = new Set(); + + const runner = createRunner(__dirname) + .expect((envelope: Envelope) => { + const payload = envelope[1]?.[0]?.[1] as SpanV2Payload; + const echoSpans = (payload.items ?? []).filter(span => span.name === 'echo'); + expect(echoSpans.length).toBeGreaterThan(0); + for (const span of echoSpans) { + expect(span.attributes?.['sentry.op']?.value).toBe('rpc'); + expect(span.trace_id).toBeDefined(); + echoTraceIds.add(span.trace_id!); + } + }) + .expect((envelope: Envelope) => { + const payload = envelope[1]?.[0]?.[1] as SpanV2Payload; + const root = payload.items?.find(span => span.name === 'GET /burst'); + expect(root).toBeDefined(); + expect(root?.attributes?.['sentry.op']?.value).toBe('http.server'); + workerTraceId = root?.trace_id; + }) + .unordered() + .start(signal); + + await runner.makeRequest('get', '/burst?n=1&id=fanout'); + await runner.completed(); + + expect(workerTraceId).toBeDefined(); + expect(echoTraceIds.size).toBe(1); + expect([...echoTraceIds][0]).toBe(workerTraceId); +}); diff --git a/dev-packages/cloudflare-integration-tests/suites/cache-client/wrangler.jsonc b/dev-packages/cloudflare-integration-tests/suites/cache-client/wrangler.jsonc new file mode 100644 index 000000000000..55491bcfc34d --- /dev/null +++ b/dev-packages/cloudflare-integration-tests/suites/cache-client/wrangler.jsonc @@ -0,0 +1,18 @@ +{ + "name": "cache-client-test", + "compatibility_date": "2025-06-17", + "main": "index.ts", + "compatibility_flags": ["nodejs_compat"], + "durable_objects": { + "bindings": [ + { "name": "CACHE_DO", "class_name": "CacheDurableObject" }, + { "name": "NO_CACHE_DO", "class_name": "NoCacheDurableObject" }, + ], + }, + "migrations": [ + { + "tag": "v1", + "new_sqlite_classes": ["CacheDurableObject", "NoCacheDurableObject"], + }, + ], +} diff --git a/packages/cloudflare/src/baseSdk.ts b/packages/cloudflare/src/baseSdk.ts index 3fdf6958f112..6443ec67ce70 100644 --- a/packages/cloudflare/src/baseSdk.ts +++ b/packages/cloudflare/src/baseSdk.ts @@ -81,12 +81,17 @@ export function getBaseDefaultIntegrations(options: CloudflareOptions): Integrat export function initWithDefaultIntegrations( options: CloudflareOptions, getDefaultIntegrationsImpl: (options: CloudflareOptions) => Integration[], + { skipFlushLock = false }: { skipFlushLock?: boolean } = {}, ): CloudflareClient | undefined { if (options.defaultIntegrations === undefined) { options.defaultIntegrations = getDefaultIntegrationsImpl(options); } - const flushLock = options.ctx ? makeFlushLock(options.ctx) : undefined; + // A cached client outlives any single invocation, so binding it to one + // invocation's flush lock would make later flushes wait on that invocation's + // waitUntil work forever. Eager delivery replaces the flush lock's purpose. + const invocationContext = options.ctx; + const flushLock = !skipFlushLock && invocationContext ? makeFlushLock(invocationContext) : undefined; delete options.ctx; const clientOptions: CloudflareClientOptions = { @@ -98,6 +103,7 @@ export function initWithDefaultIntegrations( // provider. Scope isolation is handled by the entrypoint wrappers' AsyncLocalStorage strategy. skipOpenTelemetrySetup: options.skipOpenTelemetrySetup ?? true, flushLock, + invocationContext, }; /*! rollup-include-development-only */ diff --git a/packages/cloudflare/src/client.ts b/packages/cloudflare/src/client.ts index 55fbe1a605cc..71a37406ba00 100644 --- a/packages/cloudflare/src/client.ts +++ b/packages/cloudflare/src/client.ts @@ -1,6 +1,8 @@ -import type { ClientOptions, Options, ServerRuntimeClientOptions } from '@sentry/core'; +import type { Client, ClientOptions, Options, ServerRuntimeClientOptions } from '@sentry/core'; import { _INTERNAL_clearAiProviderSkips, + _INTERNAL_flushLogsBuffer, + _INTERNAL_flushMetricsBuffer, applySdkMetadata, debug, ServerRuntimeClient, @@ -9,7 +11,9 @@ import { import { DEBUG_BUILD } from './debug-build'; import type { ExecutionContextCompat } from './executionContext'; import type { makeFlushLock } from './flush'; +import { getOriginalWaitUntil } from './flush'; import type { CloudflareTransportOptions } from './transport'; +import { getInvocationState } from './utils/invocationContext'; /** * The Sentry Cloudflare SDK Client. @@ -26,6 +30,26 @@ export class CloudflareClient extends ServerRuntimeClient { private _unsubscribeSpanStart: (() => void) | null = null; private _unsubscribeSpanEnd: (() => void) | null = null; + private _invocationContext: ExecutionContextCompat | undefined; + + /** + * Whether this client is a cached, cross-invocation client (`cacheClient`). + * Cached clients are never disposed at an invocation boundary, so their spans/events + * are delivered eagerly instead of waiting for a per-invocation flush. + */ + public readonly isCachedClient: boolean; + + /** + * Points the client at the execution context of the invocation currently being + * served. Called on every invocation for cached clients, since they outlive any + * single invocation. Only a fallback: under concurrency the correct context is + * resolved from the invocation's async context instead (see + * `getInvocationState`), which this field cannot disambiguate. + */ + public setExecutionContext(ctx: ExecutionContextCompat | undefined): void { + this._invocationContext = ctx; + } + /** * Creates a new Cloudflare SDK instance. * @param options Configuration options for this SDK. @@ -33,7 +57,7 @@ export class CloudflareClient extends ServerRuntimeClient { public constructor(options: CloudflareClientOptions) { applySdkMetadata(options, 'cloudflare'); options._metadata = options._metadata || {}; - const { flushLock, ...serverOptions } = options; + const { flushLock, invocationContext, ...serverOptions } = options; const clientOptions: ServerRuntimeClientOptions = { ...serverOptions, @@ -46,41 +70,55 @@ export class CloudflareClient extends ServerRuntimeClient { super(clientOptions); this._flushLock = flushLock; + this._invocationContext = invocationContext; + this.isCachedClient = options.cacheClient === true; - // Track span lifecycle to know when to flush - this._unsubscribeSpanStart = this.on('spanStart', span => { - const spanId = span.spanContext().spanId; - DEBUG_BUILD && debug.log('[CloudflareClient] Span started:', spanId); + if (this.isCachedClient) { + this._setupEagerEnvelopeDelivery(); + this._setupEagerSpanDelivery(); + this._setupEagerLogAndMetricDelivery(); + } - // Negatively sampled spans never emit spanEnd, - // so tracking them would cause _pendingSpans to grow unboundedly. - // We should fix the inconsistent behavior for NonRecordingSpans in the future but - // for now, we just ignore them. - if (!spanIsSampled(span)) { - return; - } + // Track span lifecycle to know when to flush. Skipped for cached clients + // (`cacheClient`): they are never disposed, so spans that end after + // a flush are still delivered. Per-invocation clients are disposed right after + // the boundary flush, so the flush must wait for open spans to end otherwise + // their transaction never gets emitted. + if (!this.isCachedClient) { + this._unsubscribeSpanStart = this.on('spanStart', span => { + const spanId = span.spanContext().spanId; + DEBUG_BUILD && debug.log('[CloudflareClient] Span started:', spanId); - this._pendingSpans.add(spanId); + // Negatively sampled spans never emit spanEnd, + // so tracking them would cause _pendingSpans to grow unboundedly. + // We should fix the inconsistent behavior for NonRecordingSpans in the future but + // for now, we just ignore them. + if (!spanIsSampled(span)) { + return; + } - if (!this._spanCompletionPromise) { - this._spanCompletionPromise = new Promise(resolve => { - this._resolveSpanCompletion = resolve; - }); - } - }); + this._pendingSpans.add(spanId); - this._unsubscribeSpanEnd = this.on('spanEnd', span => { - const spanId = span.spanContext().spanId; - DEBUG_BUILD && debug.log('[CloudflareClient] Span ended:', spanId); - this._pendingSpans.delete(spanId); + if (!this._spanCompletionPromise) { + this._spanCompletionPromise = new Promise(resolve => { + this._resolveSpanCompletion = resolve; + }); + } + }); - // If no more pending spans, resolve the completion promise - if (this._pendingSpans.size === 0 && this._resolveSpanCompletion) { - DEBUG_BUILD && debug.log('[CloudflareClient] All spans completed, resolving promise'); - this._resolveSpanCompletion(); - this._resetSpanCompletionPromise(); - } - }); + this._unsubscribeSpanEnd = this.on('spanEnd', span => { + const spanId = span.spanContext().spanId; + DEBUG_BUILD && debug.log('[CloudflareClient] Span ended:', spanId); + this._pendingSpans.delete(spanId); + + // If no more pending spans, resolve the completion promise + if (this._pendingSpans.size === 0 && this._resolveSpanCompletion) { + DEBUG_BUILD && debug.log('[CloudflareClient] All spans completed, resolving promise'); + this._resolveSpanCompletion(); + this._resetSpanCompletionPromise(); + } + }); + } } /** @@ -93,10 +131,36 @@ export class CloudflareClient extends ServerRuntimeClient { * @return {Promise} A promise that resolves to a boolean indicating whether the flush operation was successful. */ public async flush(timeout?: number): Promise { + // Mark this invocation as past its natural flush point: anything captured from + // now on (post-response waitUntil work, detached continuations) has no later + // flush to ride, so it is delivered eagerly (see _setupEagerSpanDelivery). + const invocationState = getInvocationState(); + if (invocationState) { + invocationState.flushPointReached = true; + } + + // Wait for user waitUntil-registered work to settle before draining, so events + // captured in that work are still in the buffer. Without this the final flush + // can drain (and the client be disposed) before background captures land. if (this._flushLock) { await this._flushLock.finalize(); } + // The eager log/metric drain is debounced to a microtask, so captured logs and + // metrics may not be envelopes yet. Draining only the transport would resolve + // while they are still buffer entries — and a resolving boundary flush lets the + // invocation end before their envelopes are ever created. + if (this.isCachedClient) { + _INTERNAL_flushLogsBuffer(this); + _INTERNAL_flushMetricsBuffer(this); + } + + // Await only drains owned by this invocation. Concurrent invocations keep + // independent chains on their own isolation scopes. + if (invocationState?.eagerFlushPromise) { + await invocationState.eagerFlushPromise; + } + if (this._pendingSpans.size > 0 && this._spanCompletionPromise) { DEBUG_BUILD && debug.log('[CloudflareClient] Waiting for', this._pendingSpans.size, 'pending spans to complete...'); @@ -164,6 +228,164 @@ export class CloudflareClient extends ServerRuntimeClient { this._spanCompletionPromise = null; this._resolveSpanCompletion = null; } + + /** + * Drains the transport after an envelope has been accepted. + * + * The Cloudflare transport queues request producers until `flush()` is called. Cached + * clients cannot rely on a later invocation boundary, so each accepted envelope starts + * an eager drain. Drains are serialized per invocation and registered with that + * invocation's `waitUntil`, ensuring the runtime keeps their fetches alive after the + * response is returned. + */ + private _setupEagerEnvelopeDelivery(): void { + this.on('afterEnvelope', () => { + const transport = this.getTransport(); + if (!transport) { + return; + } + const invocationState = getInvocationState(); + const flushTransport = (): PromiseLike => transport.flush(2000); + const flushPromise = invocationState?.eagerFlushPromise + ? Promise.resolve(invocationState.eagerFlushPromise).then(flushTransport, flushTransport) + : flushTransport(); + + if (invocationState) { + invocationState.eagerFlushPromise = flushPromise; + void Promise.resolve(flushPromise).finally(() => { + if (invocationState.eagerFlushPromise === flushPromise) { + invocationState.eagerFlushPromise = undefined; + } + }); + } + + this._registerWithInvocationWaitUntil(flushPromise); + }); + } + + /** + * Delivers spans that end after the invocation's flush point. + * + * Spans ending while the invocation is in flight batch in the span buffer and + * are drained by the boundary `flush()` — nothing to do here. Spans ending + * after it (in `waitUntil` work or detached continuations) have no later + * natural flush point, and the buffer's own 5s flush timer would fire outside + * any invocation, where the send can only be registered with a stale execution + * context (or none), and the runtime suspends it. Those traces are flushed + * directly — only their own bucket, never the whole buffer, so a fan-out of + * concurrent traces stays one envelope per trace. + * + * The flush point is per invocation. In Durable Objects it lands at RPC-method + * settle, so RPC spans (which end before it) keep batching one envelope per + * trace — flushing them per call would turn a fan-out trace into one envelope + * per RPC — while detached continuations inheriting that invocation's state + * flush eagerly. + */ + private _setupEagerSpanDelivery(): void { + this.on('afterSpanEnd', span => { + const invocationState = getInvocationState(); + // Only deliver spans that end after the invocation's flush point — spans + // ending before it batch in the buffer and are drained by the boundary + // flush. RPC sub-invocations in Durable Objects never reach a flush point, + // so their spans batch one envelope per trace here. + if (!invocationState?.flushPointReached || invocationState.spanFlushScheduled) { + return; + } + invocationState.spanFlushScheduled = true; + // The trace id must come from the span, not the current scope: `continueTrace` + // writes the propagation context to the *current* scope, so the forked + // isolation scope's propagation context carries a different trace id and + // flushing by it silently no-ops (measured: ~40% of post-flush traces lost). + const traceId = span.spanContext().traceId; + // Defer to a microtask: a synchronous flush here runs before the span + // streaming integration's own `afterSpanEnd` handler has added the + // triggering span to the buffer (it is registered after this one), so the + // tail span of the invocation would be left behind. The microtask still + // runs in the same async context, so the send stays attributed to this + // invocation. + queueMicrotask(() => { + invocationState.spanFlushScheduled = false; + this.emit('flushTraceSpans', traceId); + }); + }); + } + + /** + * Turns log and metric captures into envelopes without waiting for a flush. + * + * Unlike events, logs and metrics batch client-side and only become an envelope when + * their buffer is drained. The idle drain timer is disabled for this runtime + * (`_flushInterval: 0`), and a cached client never reaches an invocation-boundary + * `flush()`, so without this a captured log or metric is never delivered at all. + * + * The buffers are drained directly rather than via `emit('flush')`, which would also + * flush an opt-in span buffer mid-invocation and fragment span segments. Draining is + * debounced to a microtask so a synchronous burst (e.g. a loop of `logger` calls) + * still produces a single envelope. + */ + private _setupEagerLogAndMetricDelivery(): void { + let scheduled = false; + const scheduleDrain = (): void => { + if (scheduled) { + return; + } + scheduled = true; + queueMicrotask(() => { + scheduled = false; + _INTERNAL_flushLogsBuffer(this); + _INTERNAL_flushMetricsBuffer(this); + }); + }; + + this.on('afterCaptureLog', scheduleDrain); + this.on('afterCaptureMetric', scheduleDrain); + } + + /** + * Registers every envelope send as tracked I/O with the capturing invocation's + * `waitUntil`. + * + * The SDK never awaits `sendEnvelope()` promises, so an envelope's fetch can be + * pending-but-untracked when the invocation's tracked work settles the runtime + * suspends it and the envelope is lost even though the send started while the + * invocation was still open. This is the dominant loss path for the last captures + * of an invocation (the root span, post-response `waitUntil` work). + */ + public override sendEnvelope(envelope: Parameters[0]): ReturnType { + const sendPromise = super.sendEnvelope(envelope); + if (this.isCachedClient) { + this._registerWithInvocationWaitUntil(sendPromise); + } + return sendPromise; + } + + /** + * Attaches a promise to the `waitUntil` of the invocation that owns the current + * async context. The invocation state identifies that invocation even under + * concurrency the fallback field would point at whichever invocation last + * called `init()`, which is the wrong one when invocations overlap. In Durable + * Objects `waitUntil` is a no-op, so this degrades to the same fire-and-forget + * behavior as before there. + */ + private _registerWithInvocationWaitUntil(promise: PromiseLike): void { + const ctx = getInvocationState()?.ctx ?? this._invocationContext; + + if (!ctx) { + return; + } + + try { + getOriginalWaitUntil(ctx)?.call( + ctx, + Promise.resolve(promise).then( + () => undefined, + () => undefined, + ), + ); + } catch { + // The owning invocation already ended; the send races isolate teardown either way. + } + } } interface BaseCloudflareOptions { @@ -278,6 +500,24 @@ interface BaseCloudflareOptions { * IMPORTANT: Only set this option to `true` while developing, not in production! */ spotlight?: boolean | string; + + /** + * Cache the client and reuse it across invocations within the same isolate. + * + * The SDK creates one client per isolate and reuses it for all requests/DO + * handlers in that isolate. This avoids the per-invocation cost of + * constructing a new client. + * + * Since a cached client outlives any single invocation, delivery cannot rely + * on end-of-invocation flushes: captured events are flushed eagerly as they are + * captured, so data captured in detached/background work is still delivered. + * + * When disabled, a new client is created per invocation and disposed after the + * handler completes. + * + * @default true + */ + cacheClient?: boolean; } /** @@ -296,4 +536,5 @@ export interface CloudflareOptions extends Options, */ export interface CloudflareClientOptions extends ClientOptions, BaseCloudflareOptions { flushLock?: ReturnType; + invocationContext?: ExecutionContextCompat; } diff --git a/packages/cloudflare/src/clientCache.ts b/packages/cloudflare/src/clientCache.ts new file mode 100644 index 000000000000..2a604f36f0f0 --- /dev/null +++ b/packages/cloudflare/src/clientCache.ts @@ -0,0 +1,23 @@ +import { GLOBAL_OBJ } from '@sentry/core'; +import type { CloudflareClient } from './client'; + +const GLOBAL_CLIENT_KEY = '__SENTRY_CLOUDFLARE_CLIENT__' as const; + +type GlobalWithCloudflareClient = typeof GLOBAL_OBJ & { + [GLOBAL_CLIENT_KEY]?: CloudflareClient; +}; + +/** Returns the one cached Cloudflare client for this isolate. */ +export function getCachedClient(): CloudflareClient | undefined { + return (GLOBAL_OBJ as GlobalWithCloudflareClient)[GLOBAL_CLIENT_KEY]; +} + +/** Stores the one Cloudflare client reused by every invocation in this isolate. */ +export function cacheClient(client: CloudflareClient): void { + (GLOBAL_OBJ as GlobalWithCloudflareClient)[GLOBAL_CLIENT_KEY] = client; +} + +/** @hidden Only for testing - clears the isolate's cached Cloudflare client. */ +export function _clearGlobalClientCache(): void { + (GLOBAL_OBJ as GlobalWithCloudflareClient)[GLOBAL_CLIENT_KEY] = undefined; +} diff --git a/packages/cloudflare/src/flush.ts b/packages/cloudflare/src/flush.ts index fe86e21dbd62..8273f1c318da 100644 --- a/packages/cloudflare/src/flush.ts +++ b/packages/cloudflare/src/flush.ts @@ -119,6 +119,8 @@ function getOrCreateFlushLockRegistry(context: ExecutionContextCompat): FlushLoc /** * Flushes the client and then disposes of it to allow garbage collection. * This should be called at the end of each request to prevent memory leaks. + * Cached clients (`cacheClient`) are reused across invocations, so + * they are flushed but not disposed. * * This function never rejects. On Workers, a rejected promise passed to * `ctx.waitUntil` marks the whole invocation as `outcome: exception` even when @@ -139,10 +141,14 @@ export async function flushAndDispose(client: Client | undefined, timeout = 2000 } catch (e) { DEBUG_BUILD && debug.warn('Failed to flush client', e); } finally { - try { - client?.dispose(); - } catch (e) { - DEBUG_BUILD && debug.warn('Failed to dispose client', e); + // Only dispose per-invocation clients. Cached clients (`cacheClient`) + // are reused across invocations and must not be disposed at an invocation boundary. + if (!(client as { isCachedClient?: boolean } | undefined)?.isCachedClient) { + try { + client?.dispose(); + } catch (e) { + DEBUG_BUILD && debug.warn('Failed to dispose client', e); + } } } } diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentEmail.ts b/packages/cloudflare/src/instrumentations/worker/instrumentEmail.ts index ed2d9a05f7f5..4fcc477a5082 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentEmail.ts +++ b/packages/cloudflare/src/instrumentations/worker/instrumentEmail.ts @@ -17,6 +17,7 @@ import { getFinalOptions } from '../../options'; import { addCloudResourceContext } from '../../scope-utils'; import { init } from '../../sdk'; import { instrumentContext } from '../../utils/instrumentContext'; +import { setInvocationState } from '../../utils/invocationContext'; import { instrumentEnv } from './instrumentEnv'; /** @@ -31,6 +32,8 @@ function wrapEmailHandler( return withIsolationScope(isolationScope => { const waitUntil = context.waitUntil.bind(context); + setInvocationState(isolationScope, { ctx: context }); + const client = init({ ...options, ctx: context }); isolationScope.setClient(client); diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts b/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts index ac609e241a55..975bc1a986dc 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts +++ b/packages/cloudflare/src/instrumentations/worker/instrumentQueue.ts @@ -17,6 +17,7 @@ import { getFinalOptions } from '../../options'; import { addCloudResourceContext } from '../../scope-utils'; import { init } from '../../sdk'; import { instrumentContext } from '../../utils/instrumentContext'; +import { setInvocationState } from '../../utils/invocationContext'; import { instrumentEnv } from './instrumentEnv'; /** @@ -31,6 +32,8 @@ function wrapQueueHandler( return withIsolationScope(isolationScope => { const waitUntil = context.waitUntil.bind(context); + setInvocationState(isolationScope, { ctx: context }); + const client = init({ ...options, ctx: context }); isolationScope.setClient(client); diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentScheduled.ts b/packages/cloudflare/src/instrumentations/worker/instrumentScheduled.ts index 018dd8b56ee1..dd53c41c1b99 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentScheduled.ts +++ b/packages/cloudflare/src/instrumentations/worker/instrumentScheduled.ts @@ -17,6 +17,7 @@ import { getFinalOptions } from '../../options'; import { addCloudResourceContext } from '../../scope-utils'; import { init } from '../../sdk'; import { instrumentContext } from '../../utils/instrumentContext'; +import { setInvocationState } from '../../utils/invocationContext'; import { instrumentEnv } from './instrumentEnv'; function wrapScheduledHandler( @@ -28,6 +29,8 @@ function wrapScheduledHandler( return withIsolationScope(isolationScope => { const waitUntil = context.waitUntil.bind(context); + setInvocationState(isolationScope, { ctx: context }); + const client = init({ ...options, ctx: context }); isolationScope.setClient(client); diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentTail.ts b/packages/cloudflare/src/instrumentations/worker/instrumentTail.ts index 925f2b504605..31ecd6674120 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentTail.ts +++ b/packages/cloudflare/src/instrumentations/worker/instrumentTail.ts @@ -9,6 +9,7 @@ import { getFinalOptions } from '../../options'; import { addCloudResourceContext } from '../../scope-utils'; import { init } from '../../sdk'; import { instrumentContext } from '../../utils/instrumentContext'; +import { setInvocationState } from '../../utils/invocationContext'; import { instrumentEnv } from './instrumentEnv'; /** @@ -19,6 +20,8 @@ function wrapTailHandler(options: CloudflareOptions, context: ExecutionContext, return withIsolationScope(async isolationScope => { const waitUntil = context.waitUntil.bind(context); + setInvocationState(isolationScope, { ctx: context }); + const client = init({ ...options, ctx: context }); isolationScope.setClient(client); diff --git a/packages/cloudflare/src/request.ts b/packages/cloudflare/src/request.ts index 4f7e2fbbe916..356bdb69761d 100644 --- a/packages/cloudflare/src/request.ts +++ b/packages/cloudflare/src/request.ts @@ -225,5 +225,5 @@ export function wrapRequestHandlerWithInit( }); }, ); - }); + }, wrapperOptions.context); } diff --git a/packages/cloudflare/src/sdk.ts b/packages/cloudflare/src/sdk.ts index f96cbdc0626f..cf09560f945a 100644 --- a/packages/cloudflare/src/sdk.ts +++ b/packages/cloudflare/src/sdk.ts @@ -1,7 +1,12 @@ import type { Integration } from '@sentry/core'; +import { getCurrentScope, setCurrentClient } from '@sentry/core'; import { vercelAIIntegration } from './integrations/tracing/vercelai'; import { getBaseDefaultIntegrations, initWithDefaultIntegrations } from './baseSdk'; import type { CloudflareClient, CloudflareOptions } from './client'; +import { cacheClient, getCachedClient } from './clientCache'; + +// Test-only helper, re-exported here so tests can reset the global client cache. +export { _clearGlobalClientCache } from './clientCache'; /** * Get the default integrations for the Cloudflare SDK. @@ -20,7 +25,46 @@ export function getDefaultIntegrations(options: CloudflareOptions): Integration[ /** * Initializes the cloudflare SDK. + * + * The client is cached and reused across invocations within the same isolate, + * unless `cacheClient: false` is passed. This avoids the + * per-invocation cost of constructing a new client, and it is what makes + * Durable Object telemetry reliable: a per-invocation client is disposed at + * the end of the handler, and in a Durable Object there is no `waitUntil` + * boundary that reliably extends execution, so spans/events that end after + * disposal would otherwise be lost. */ export function init(options: CloudflareOptions): CloudflareClient | undefined { - return initWithDefaultIntegrations(options, getDefaultIntegrations); + const cacheEnabled = options.cacheClient !== false; + + if (cacheEnabled) { + // Normalize the flag so the client marks itself as cached. + options.cacheClient = true; + } + + if (cacheEnabled && options.dsn) { + const cached = getCachedClient(); + // A cached client that has lost its transport was disposed. Replace it rather + // than returning a dead client for the rest of the isolate's lifetime. + if (cached?.getTransport()) { + // Mirror the two scope side effects of `initAndBind`, which only runs on first + // creation. Without the re-bind the scope keeps whatever client a previous init + // left behind — which may have been disposed since — and without the update + // `initialScope` would apply only to an isolate's very first invocation. + getCurrentScope().update(options.initialScope); + setCurrentClient(cached); + // The cached client outlives the invocation that created it, so its eager + // sends must be registered with the current invocation's waitUntil. + cached.setExecutionContext(options.ctx); + return cached; + } + } + + const client = initWithDefaultIntegrations(options, getDefaultIntegrations, { skipFlushLock: cacheEnabled }); + + if (cacheEnabled && client && options.dsn) { + cacheClient(client); + } + + return client; } diff --git a/packages/cloudflare/src/transport.ts b/packages/cloudflare/src/transport.ts index 6069ec631189..685cb7fbccd5 100644 --- a/packages/cloudflare/src/transport.ts +++ b/packages/cloudflare/src/transport.ts @@ -8,7 +8,7 @@ export interface CloudflareTransportOptions extends BaseTransportOptions { fetchOptions?: RequestInit; } -const DEFAULT_TRANSPORT_BUFFER_SIZE = 30; +const DEFAULT_TRANSPORT_BUFFER_SIZE = 256; /** * This is a modified promise buffer that collects tasks until drain is called. diff --git a/packages/cloudflare/src/utils/invocationContext.ts b/packages/cloudflare/src/utils/invocationContext.ts new file mode 100644 index 000000000000..ec1f450bb885 --- /dev/null +++ b/packages/cloudflare/src/utils/invocationContext.ts @@ -0,0 +1,72 @@ +import type { Scope } from '@sentry/core'; +import { getDefaultIsolationScope, getIsolationScope } from '@sentry/core'; +import type { ExecutionContextCompat } from '../executionContext'; + +/** + * State owned by a single invocation (request, RPC call, cron, ...). + * + * A cached client (`cacheClient`) is shared by all invocations running in the same + * isolate, so anything invocation-owned must not live on the client: two overlapping + * requests would otherwise overwrite each other's state (e.g. the execution context + * used to register eager flushes) and the earlier invocation's envelopes would be + * suspended when the later invocation ends. + * + * The state rides on the invocation's forked isolation scope, which the async + * context strategy (AsyncLocalStorage) hands back for exactly the async context that + * owns the invocation — including detached continuations created inside it. A symbol + * key keeps it off `Scope.clone()` and out of serialized event data. + */ +export interface InvocationState { + /** + * The execution context of the invocation that owns this scope. Eager envelope + * sends are registered with this context's `waitUntil`, so they are attributed to + * the invocation that captured the data — even when a concurrent invocation has + * since pointed the shared client at its own context. + */ + readonly ctx: ExecutionContextCompat | undefined; + /** + * Set by `CloudflareClient.flush()` — the invocation's natural flush point. + * Spans ending before it are drained by that flush; spans ending after it (in + * `waitUntil` work or detached continuations) have no later flush to ride and + * are delivered eagerly. + */ + flushPointReached?: boolean; + /** + * Set while an eager span flush is scheduled for this invocation. Kept per + * invocation (not on the shared client) so two concurrent invocations past + * their flush point each schedule their own flush in their own async context — + * the flush and its envelope send stay attributed to the owning invocation. + */ + spanFlushScheduled?: boolean; + /** Eager transport drains owned by this invocation, serialized in capture order. */ + eagerFlushPromise?: PromiseLike; +} + +const INVOCATION_STATE: unique symbol = Symbol('sentryInvocationState'); + +type ScopeWithInvocationState = Scope & { + [INVOCATION_STATE]?: InvocationState; +}; + +/** + * Attaches invocation state to a forked isolation scope. Only meaningful on a scope + * that outlives nothing but this invocation — never attach to the default isolation + * scope, which is shared by every invocation in the isolate. + */ +export function setInvocationState(scope: Scope, state: InvocationState): void { + (scope as ScopeWithInvocationState)[INVOCATION_STATE] = state; +} + +/** + * Returns the state of the invocation that owns the current async context, or + * `undefined` outside any instrumented invocation (the default isolation scope is + * shared, so state read from it could not be attributed to one invocation — and it + * is never attached there in the first place). + */ +export function getInvocationState(): InvocationState | undefined { + const isolationScope = getIsolationScope() as ScopeWithInvocationState; + if (isolationScope === getDefaultIsolationScope()) { + return undefined; + } + return isolationScope[INVOCATION_STATE]; +} diff --git a/packages/cloudflare/src/utils/invocationScope.ts b/packages/cloudflare/src/utils/invocationScope.ts index 3591bac64eb2..aaaf98fdaa2a 100644 --- a/packages/cloudflare/src/utils/invocationScope.ts +++ b/packages/cloudflare/src/utils/invocationScope.ts @@ -1,4 +1,6 @@ import { getDefaultIsolationScope, getIsolationScope, type Scope, withIsolationScope } from '@sentry/core'; +import type { ExecutionContextCompat } from '../executionContext'; +import { setInvocationState } from './invocationContext'; /** * Runs `callback` on the isolation scope for the current invocation. @@ -21,10 +23,15 @@ import { getDefaultIsolationScope, getIsolationScope, type Scope, withIsolationS * default scope even inside an invocation; there the fork degrades to a no-op, which the stack strategy * tolerates. This matches the approach used by `patchEventHandler` in Nuxt. */ -export function withInvocationIsolationScope(callback: (scope: Scope) => T): T { +export function withInvocationIsolationScope(callback: (scope: Scope) => T, context?: ExecutionContextCompat): T { const isolationScope = getIsolationScope(); - const newIsolationScope = isolationScope === getDefaultIsolationScope() ? isolationScope.clone() : isolationScope; + const isEntryPoint = isolationScope === getDefaultIsolationScope(); + const newIsolationScope = isEntryPoint ? isolationScope.clone() : isolationScope; + + if (isEntryPoint) { + setInvocationState(newIsolationScope, { ctx: context }); + } return withIsolationScope(newIsolationScope, () => callback(newIsolationScope)); } diff --git a/packages/cloudflare/src/wrapMethodWithSentry.ts b/packages/cloudflare/src/wrapMethodWithSentry.ts index 290c9947b5e0..e15d4cd7d5b6 100644 --- a/packages/cloudflare/src/wrapMethodWithSentry.ts +++ b/packages/cloudflare/src/wrapMethodWithSentry.ts @@ -234,7 +234,10 @@ export function wrapMethodWithSentry( return executeSpan(); }; - return withInvocationIsolationScope(wrappedFunction); + return withInvocationIsolationScope( + wrappedFunction, + wrapperOptions.context as ExecutionContextCompat | undefined, + ); }, }), noMark, diff --git a/packages/cloudflare/test/client.test.ts b/packages/cloudflare/test/client.test.ts index 7e305f6aec65..3bb3c542453c 100644 --- a/packages/cloudflare/test/client.test.ts +++ b/packages/cloudflare/test/client.test.ts @@ -2,6 +2,8 @@ import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'; import { setAsyncLocalStorageAsyncContextStrategy } from '@sentry/server-utils/no-diagnostic-channels'; import { CloudflareClient, type CloudflareClientOptions } from '../src/client'; import { makeFlushLock } from '../src/flush'; +import { getInvocationState } from '../src/utils/invocationContext'; +import { withInvocationIsolationScope } from '../src/utils/invocationScope'; const TRACE_FLAG_SAMPLED = 0x1; @@ -9,6 +11,8 @@ const MOCK_CLIENT_OPTIONS: CloudflareClientOptions = { dsn: 'https://public@dsn.ingest.sentry.io/1337', stackParser: () => [], integrations: [], + // These tests exercise the per-invocation client behavior + cacheClient: false, transport: () => ({ send: vi.fn().mockResolvedValue({}), flush: vi.fn().mockResolvedValue(true), @@ -220,6 +224,52 @@ describe('CloudflareClient', () => { }); }); + describe('flush()', () => { + it('calls transport flush with the given timeout', async () => { + const client = new CloudflareClient(MOCK_CLIENT_OPTIONS); + + const privateClient = client as unknown as { + _transport: { flush: ReturnType }; + }; + + await client.flush(3000); + + expect(privateClient._transport.flush).toHaveBeenCalledWith(3000); + }); + + it('resolves with the transport flush result', async () => { + const client = new CloudflareClient(MOCK_CLIENT_OPTIONS); + + const result = await client.flush(1000); + + expect(result).toBe(true); + }); + + it('waits for the flush lock before draining the transport', async () => { + let releaseLock!: () => void; + const finalize = vi.fn(() => new Promise(resolve => (releaseLock = resolve))); + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + flushLock: { ready: Promise.resolve(), finalize }, + }); + + const privateClient = client as unknown as { + _transport: { flush: ReturnType }; + }; + + const flushPromise = client.flush(1000); + + // The transport must not drain while the lock is pending + await Promise.resolve(); + expect(finalize).toHaveBeenCalled(); + expect(privateClient._transport.flush).not.toHaveBeenCalled(); + + releaseLock(); + await flushPromise; + expect(privateClient._transport.flush).toHaveBeenCalledWith(1000); + }); + }); + describe('span lifecycle tracking', () => { it('tracks pending spans when spanStart is emitted', () => { const client = new CloudflareClient(MOCK_CLIENT_OPTIONS); @@ -332,5 +382,469 @@ describe('CloudflareClient', () => { client.emit('spanStart', mockSpan as any); expect(privateClient._pendingSpans.has('test-span-id')).toBe(false); }); + + it('does not track spans when cacheClient is enabled', async () => { + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + cacheClient: true, + }); + + const privateClient = client as unknown as { + _pendingSpans: Set; + _unsubscribeSpanStart: (() => void) | null; + _unsubscribeSpanEnd: (() => void) | null; + }; + + // Span tracking is disabled for cached clients — flush must not wait + expect(privateClient._unsubscribeSpanStart).toBeNull(); + expect(privateClient._unsubscribeSpanEnd).toBeNull(); + + const mockSpan = { + spanContext: () => ({ spanId: 'test-span-id', traceFlags: TRACE_FLAG_SAMPLED }), + }; + client.emit('spanStart', mockSpan as any); + + expect(privateClient._pendingSpans.size).toBe(0); + await expect(client.flush(10)).resolves.toBe(true); + }); + }); + + describe('cached client eager flush tracking', () => { + function makeEagerFlushClient(flushMock: ReturnType): CloudflareClient { + return new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + cacheClient: true, + transport: () => ({ + send: vi.fn().mockResolvedValue({}), + flush: flushMock, + }), + }); + } + + it('flush() awaits in-flight eager envelope flushes', async () => { + let resolveEagerFlush: (value: boolean) => void = () => undefined; + let call = 0; + const flushMock = vi.fn().mockImplementation(() => { + call++; + if (call === 1) { + return new Promise(res => { + resolveEagerFlush = res; + }); + } + return Promise.resolve(true); + }); + const client = makeEagerFlushClient(flushMock); + const ctx = { waitUntil: vi.fn(), passThroughOnException: vi.fn() }; + + await withInvocationIsolationScope(async () => { + // An emitted envelope starts this invocation's eager transport drain. + client.emit('afterEnvelope', {}); + expect(flushMock).toHaveBeenCalledTimes(1); + + let flushResolved = false; + const flushPromise = client.flush(10).then(result => { + flushResolved = true; + return result; + }); + + // The boundary flush waits for the drain owned by this invocation. + await new Promise(resolve => setTimeout(resolve, 20)); + expect(flushResolved).toBe(false); + + resolveEagerFlush(true); + await expect(flushPromise).resolves.toBe(true); + expect(flushResolved).toBe(true); + }, ctx as never); + }); + + it('flush() resolves immediately once eager flushes have settled', async () => { + const flushMock = vi.fn().mockResolvedValue(true); + const client = makeEagerFlushClient(flushMock); + + client.emit('afterEnvelope', {}); + expect(flushMock).toHaveBeenCalledTimes(1); + + await expect(client.flush(10)).resolves.toBe(true); + }); + + it('tracks eager drains independently for concurrent invocations', async () => { + let resolveA: (value: boolean) => void = () => undefined; + let resolveB: (value: boolean) => void = () => undefined; + let call = 0; + const flushMock = vi.fn().mockImplementation(() => { + call++; + if (call === 1) { + return new Promise(resolve => { + resolveA = resolve; + }); + } + if (call === 2) { + return new Promise(resolve => { + resolveB = resolve; + }); + } + return Promise.resolve(true); + }); + const client = makeEagerFlushClient(flushMock); + const ctxA = { waitUntil: vi.fn(), passThroughOnException: vi.fn() }; + const ctxB = { waitUntil: vi.fn(), passThroughOnException: vi.fn() }; + + let flushAResolved = false; + let flushBResolved = false; + const flushA = withInvocationIsolationScope(async () => { + client.emit('afterEnvelope', {}); + return client.flush(1000).then(result => { + flushAResolved = true; + return result; + }); + }, ctxA as never); + const flushB = withInvocationIsolationScope(async () => { + client.emit('afterEnvelope', {}); + return client.flush(1000).then(result => { + flushBResolved = true; + return result; + }); + }, ctxB as never); + + resolveA(true); + await expect(flushA).resolves.toBe(true); + expect(flushAResolved).toBe(true); + expect(flushBResolved).toBe(false); + + resolveB(true); + await expect(flushB).resolves.toBe(true); + expect(flushBResolved).toBe(true); + }); + + it('does not flush eagerly per envelope when cacheClient is disabled', () => { + const flushMock = vi.fn().mockResolvedValue(true); + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + cacheClient: false, + transport: () => ({ + send: vi.fn().mockResolvedValue({}), + flush: flushMock, + }), + }); + + client.emit('afterEnvelope', {}); + expect(flushMock).not.toHaveBeenCalled(); + }); + + it('registers the eager flush with the invocation context waitUntil', () => { + const flushMock = vi.fn().mockResolvedValue(true); + const waitUntil = vi.fn(); + const ctx = { waitUntil, passThroughOnException: vi.fn() }; + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + cacheClient: true, + // oxlint-disable-next-line typescript/no-explicit-any + invocationContext: ctx as any, + transport: () => ({ + send: vi.fn().mockResolvedValue({}), + flush: flushMock, + }), + }); + + client.emit('afterEnvelope', {}); + expect(flushMock).toHaveBeenCalledTimes(1); + expect(waitUntil).toHaveBeenCalledTimes(1); + }); + + it('uses the context from setExecutionContext for eager flush registration', () => { + const flushMock = vi.fn().mockResolvedValue(true); + const waitUntil = vi.fn(); + const ctx = { waitUntil, passThroughOnException: vi.fn() }; + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + cacheClient: true, + transport: () => ({ + send: vi.fn().mockResolvedValue({}), + flush: flushMock, + }), + }); + + // oxlint-disable-next-line typescript/no-explicit-any + client.setExecutionContext(ctx as any); + client.emit('afterEnvelope', {}); + expect(waitUntil).toHaveBeenCalledTimes(1); + }); + + it('registers the eager flush with the capturing invocation, not the latest one', () => { + const flushMock = vi.fn().mockResolvedValue(true); + const waitUntilA = vi.fn(); + const waitUntilB = vi.fn(); + const ctxA = { waitUntil: waitUntilA, passThroughOnException: vi.fn() }; + const ctxB = { waitUntil: waitUntilB, passThroughOnException: vi.fn() }; + const client = makeEagerFlushClient(flushMock); + + // The shared client's fallback points at the latest invocation (B). An + // envelope captured by the still-running invocation A must register its + // flush on A's waitUntil — otherwise it is suspended when B's invocation + // ends first. + // oxlint-disable-next-line typescript/no-explicit-any + client.setExecutionContext(ctxB as any); + + withInvocationIsolationScope(() => { + client.emit('afterEnvelope', {}); + }, ctxA as never); + expect(waitUntilA).toHaveBeenCalledTimes(1); + expect(waitUntilB).not.toHaveBeenCalled(); + + withInvocationIsolationScope(() => { + client.emit('afterEnvelope', {}); + }, ctxB as never); + expect(waitUntilB).toHaveBeenCalledTimes(1); + }); + + it('registers envelope sends with the capturing invocation waitUntil', () => { + const waitUntil = vi.fn(); + const ctx = { waitUntil, passThroughOnException: vi.fn() }; + const sendMock = vi.fn().mockResolvedValue({}); + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + cacheClient: true, + transport: () => ({ + send: sendMock, + flush: vi.fn().mockResolvedValue(true), + }), + }); + + withInvocationIsolationScope(() => { + void client.sendEnvelope([{}, []] as never); + }, ctx as never); + + expect(sendMock).toHaveBeenCalledTimes(1); + expect(waitUntil).toHaveBeenCalledTimes(1); + }); + + it('does not register sends with waitUntil when cacheClient is disabled', () => { + const waitUntil = vi.fn(); + const ctx = { waitUntil, passThroughOnException: vi.fn() }; + const sendMock = vi.fn().mockResolvedValue({}); + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + cacheClient: false, + transport: () => ({ + send: sendMock, + flush: vi.fn().mockResolvedValue(true), + }), + }); + + withInvocationIsolationScope(() => { + void client.sendEnvelope([{}, []] as never); + }, ctx as never); + + expect(sendMock).toHaveBeenCalledTimes(1); + expect(waitUntil).not.toHaveBeenCalled(); + }); + + it('never lets a failing send reject the waitUntil registration', async () => { + let registered: Promise | undefined; + const waitUntil = vi.fn((promise: Promise) => { + registered = promise; + }); + const ctx = { waitUntil, passThroughOnException: vi.fn() }; + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + cacheClient: true, + transport: () => ({ + send: vi.fn().mockRejectedValue(new Error('ingest down')), + flush: vi.fn().mockResolvedValue(true), + }), + }); + + withInvocationIsolationScope(() => { + void client.sendEnvelope([{}, []] as never); + }, ctx as never); + + expect(waitUntil).toHaveBeenCalledTimes(1); + // The promise handed to the runtime must resolve — a rejected waitUntil + // promise would mark the invocation's outcome as an exception. + await expect(registered).resolves.toBeUndefined(); + }); + + it('does not register a waitUntil when no invocation context is set', () => { + const flushMock = vi.fn().mockResolvedValue(true); + const waitUntil = vi.fn(); + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + cacheClient: true, + transport: () => ({ + send: vi.fn().mockResolvedValue({}), + flush: flushMock, + }), + }); + + client.emit('afterEnvelope', {}); + expect(flushMock).toHaveBeenCalledTimes(1); + expect(waitUntil).not.toHaveBeenCalled(); + }); + }); + + describe('cached client eager span delivery', () => { + function makeCachedClient(): { client: CloudflareClient; flushSpy: ReturnType } { + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + cacheClient: true, + traceLifecycle: 'stream', + } as never); + const flushSpy = vi.fn(); + client.on('flushTraceSpans', flushSpy); + return { client, flushSpy }; + } + + // The handler only reads the span's trace id — it is forwarded to the flushTraceSpans hook. + function makeSpan(traceId: string) { + return { spanContext: () => ({ traceId }) }; + } + + const tick = (): Promise => new Promise(resolve => setTimeout(resolve, 0)); + const ctx = { waitUntil: vi.fn(), passThroughOnException: vi.fn() }; + + it('does not flush spans ending before the invocation flush point', async () => { + const { client, flushSpy } = makeCachedClient(); + + await withInvocationIsolationScope(async () => { + client.emit('afterSpanEnd', makeSpan('trace-b') as never); + await tick(); + }, ctx as never); + + expect(flushSpy).not.toHaveBeenCalled(); + }); + + it('flushes the trace of a span ending after the invocation flush point', async () => { + const { client, flushSpy } = makeCachedClient(); + + await withInvocationIsolationScope(async () => { + await client.flush(0); + client.emit('afterSpanEnd', makeSpan('trace-1') as never); + await tick(); + }, ctx as never); + + expect(flushSpy).toHaveBeenCalledTimes(1); + expect(flushSpy).toHaveBeenCalledWith('trace-1'); + }); + + it('does not flush spans of an invocation whose flush point has not been reached', async () => { + const { client, flushSpy } = makeCachedClient(); + + // Invocation A passes its flush point … + await withInvocationIsolationScope(async () => { + await client.flush(0); + }, ctx as never); + + // … while invocation B is still in flight — its spans keep batching + const ctxB = { waitUntil: vi.fn(), passThroughOnException: vi.fn() }; + await withInvocationIsolationScope(async () => { + client.emit('afterSpanEnd', makeSpan('trace-1') as never); + await tick(); + }, ctxB as never); + + expect(flushSpy).not.toHaveBeenCalled(); + }); + + it('schedules a flush per invocation when concurrent invocations are past their flush point', async () => { + const { client, flushSpy } = makeCachedClient(); + + const ctxA = { waitUntil: vi.fn(), passThroughOnException: vi.fn() }; + const ctxB = { waitUntil: vi.fn(), passThroughOnException: vi.fn() }; + + // Both invocations end a span past their flush point in the same tick — each + // must schedule its own flush in its own async context, so the listener + // derives the right trace (and waitUntil) per invocation. + await Promise.all([ + withInvocationIsolationScope(async () => { + await client.flush(0); + client.emit('afterSpanEnd', makeSpan('trace-a') as never); + }, ctxA as never), + withInvocationIsolationScope(async () => { + await client.flush(0); + client.emit('afterSpanEnd', makeSpan('trace-b') as never); + }, ctxB as never), + ]); + await tick(); + + expect(flushSpy).toHaveBeenCalledTimes(2); + expect(flushSpy).toHaveBeenCalledWith('trace-a'); + expect(flushSpy).toHaveBeenCalledWith('trace-b'); + }); + + it("flushes each invocation's trace in its own async context", async () => { + const { client } = makeCachedClient(); + const ctxA = { waitUntil: vi.fn(), passThroughOnException: vi.fn() }; + const ctxB = { waitUntil: vi.fn(), passThroughOnException: vi.fn() }; + const flushContext = new Map(); + + client.on('flushTraceSpans', traceId => { + flushContext.set(String(traceId), getInvocationState()?.ctx); + }); + + await Promise.all([ + withInvocationIsolationScope(async () => { + await client.flush(0); + client.emit('afterSpanEnd', makeSpan('trace-a') as never); + }, ctxA as never), + withInvocationIsolationScope(async () => { + await client.flush(0); + client.emit('afterSpanEnd', makeSpan('trace-b') as never); + }, ctxB as never), + ]); + await tick(); + + expect(flushContext.get('trace-a')).toBe(ctxA); + expect(flushContext.get('trace-b')).toBe(ctxB); + }); + + it('delivers spans of detached continuations eagerly once the owning invocation flushed', async () => { + const { client, flushSpy } = makeCachedClient(); + + let releaseDetached!: () => void; + const detachedGate = new Promise(resolve => { + releaseDetached = resolve; + }); + // A detached continuation is created inside the invocation but settles after it + const continuation = withInvocationIsolationScope(async () => { + await client.flush(0); + return (async () => { + await detachedGate; + client.emit('afterSpanEnd', makeSpan('trace-1') as never); + })(); + }, ctx as never); + + releaseDetached(); + await continuation; + await tick(); + + expect(flushSpy).toHaveBeenCalledTimes(1); + expect(flushSpy).toHaveBeenCalledWith('trace-1'); + }); + + it('does not flush spans ending outside any invocation', async () => { + const { client, flushSpy } = makeCachedClient(); + + client.emit('afterSpanEnd', makeSpan('trace-1') as never); + await tick(); + + expect(flushSpy).not.toHaveBeenCalled(); + }); + + it('does not flush for span ends when cacheClient is disabled', async () => { + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + cacheClient: false, + traceLifecycle: 'stream', + } as never); + const flushSpy = vi.fn(); + client.on('flushTraceSpans', flushSpy); + + await withInvocationIsolationScope(async () => { + await client.flush(0); + client.emit('afterSpanEnd', makeSpan('trace-1') as never); + await tick(); + }, ctx as never); + + expect(flushSpy).not.toHaveBeenCalled(); + }); }); }); diff --git a/packages/cloudflare/test/flush.test.ts b/packages/cloudflare/test/flush.test.ts index bcef56a8c101..9686d3e7e61b 100644 --- a/packages/cloudflare/test/flush.test.ts +++ b/packages/cloudflare/test/flush.test.ts @@ -138,6 +138,19 @@ describe('flushAndDispose', () => { await expect(flushAndDispose(undefined)).resolves.toBeUndefined(); flushSpy.mockRestore(); }); + + it('should not dispose the client when it is cached (cacheClient: true)', async () => { + const mockClient = { + flush: vi.fn().mockResolvedValue(true), + dispose: vi.fn(), + isCachedClient: true, + } as unknown as Client; + + await flushAndDispose(mockClient); + + expect(mockClient.flush).toHaveBeenCalled(); + expect(mockClient.dispose).not.toHaveBeenCalled(); + }); }); describe('getOriginalWaitUntil', () => { diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentEmail.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentEmail.test.ts index ff524a76fa8f..d5bcd5e9f95c 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentEmail.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentEmail.test.ts @@ -7,6 +7,7 @@ import * as SentryCore from '@sentry/core'; import { beforeEach, describe, expect, onTestFinished, test, vi } from 'vitest'; import { CloudflareClient } from '../../../src/client'; import { withSentry } from '../../../src/withSentry'; +import { resetSdk } from '../../testUtils'; const MOCK_ENV = { SENTRY_DSN: 'https://public@dsn.ingest.sentry.io/1337', @@ -44,6 +45,7 @@ function addDelayedWaitUntil(context: ExecutionContext) { describe('instrumentEmail', () => { beforeEach(() => { vi.clearAllMocks(); + resetSdk(); }); test('does not double-wrap when withSentry is called twice', async () => { @@ -289,7 +291,7 @@ describe('instrumentEmail', () => { }, } satisfies ExportedHandler; - const wrappedHandler = withSentry(vi.fn(), handler); + const wrappedHandler = withSentry(() => ({ cacheClient: false }), handler); const waits: Promise[] = []; const waitUntil = vi.fn(promise => waits.push(promise)); await wrappedHandler.email?.(createMockEmailMessage(), MOCK_ENV_WITHOUT_DSN, { diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentFetch.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentFetch.test.ts index 29ce3f4c5948..1a0e94093444 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentFetch.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentFetch.test.ts @@ -6,6 +6,7 @@ import type { Event } from '@sentry/core'; import * as SentryCore from '@sentry/core'; import { beforeEach, describe, expect, onTestFinished, test, vi } from 'vitest'; import { withSentry } from '../../../src/withSentry'; +import { resetSdk } from '../../testUtils'; const MOCK_ENV = { SENTRY_DSN: 'https://public@dsn.ingest.sentry.io/1337', @@ -30,6 +31,7 @@ function addDelayedWaitUntil(context: ExecutionContext) { describe('instrumentFetch', () => { beforeEach(() => { vi.clearAllMocks(); + resetSdk(); }); test('does not double-wrap when withSentry is called twice', async () => { @@ -160,7 +162,7 @@ describe('instrumentFetch', () => { }, } satisfies ExportedHandler; - const wrappedHandler = withSentry(vi.fn(), handler); + const wrappedHandler = withSentry(() => ({ cacheClient: false }), handler); const waits: Promise[] = []; const waitUntil = vi.fn(promise => waits.push(promise)); await wrappedHandler diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts index 66bf3077c39e..7aa89513dcad 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts @@ -7,6 +7,7 @@ import * as SentryCore from '@sentry/core'; import { beforeEach, describe, expect, onTestFinished, test, vi } from 'vitest'; import { CloudflareClient } from '../../../src/client'; import { withSentry } from '../../../src/withSentry'; +import { resetSdk } from '../../testUtils'; const MOCK_ENV = { SENTRY_DSN: 'https://public@dsn.ingest.sentry.io/1337', @@ -57,6 +58,7 @@ function addDelayedWaitUntil(context: ExecutionContext) { describe('instrumentQueue', () => { beforeEach(() => { vi.clearAllMocks(); + resetSdk(); }); test('does not double-wrap when withSentry is called twice', async () => { @@ -308,7 +310,7 @@ describe('instrumentQueue', () => { }, } satisfies ExportedHandler; - const wrappedHandler = withSentry(vi.fn(), handler); + const wrappedHandler = withSentry(() => ({ cacheClient: false }), handler); const waits: Promise[] = []; const waitUntil = vi.fn(promise => waits.push(promise)); await wrappedHandler.queue?.(createMockQueueBatch(), MOCK_ENV_WITHOUT_DSN, { diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentScheduled.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentScheduled.test.ts index 2597441d249e..cd541ca1ae51 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentScheduled.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentScheduled.test.ts @@ -7,6 +7,7 @@ import * as SentryCore from '@sentry/core'; import { beforeEach, describe, expect, onTestFinished, test, vi } from 'vitest'; import { CloudflareClient } from '../../../src/client'; import { withSentry } from '../../../src/withSentry'; +import { resetSdk } from '../../testUtils'; const MOCK_ENV = { SENTRY_DSN: 'https://public@dsn.ingest.sentry.io/1337', @@ -39,6 +40,7 @@ function addDelayedWaitUntil(context: ExecutionContext) { describe('instrumentScheduled', () => { beforeEach(() => { vi.clearAllMocks(); + resetSdk(); }); test('does not double-wrap when withSentry is called twice', async () => { @@ -285,7 +287,7 @@ describe('instrumentScheduled', () => { }, } satisfies ExportedHandler; - const wrappedHandler = withSentry(vi.fn(), handler); + const wrappedHandler = withSentry(() => ({ cacheClient: false }), handler); const waits: Promise[] = []; const waitUntil = vi.fn(promise => waits.push(promise)); await wrappedHandler.scheduled?.(createMockScheduledController(), MOCK_ENV_WITHOUT_DSN, { diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentTail.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentTail.test.ts index 4f47dc3c62c7..014916e6f158 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentTail.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentTail.test.ts @@ -7,6 +7,7 @@ import * as SentryCore from '@sentry/core'; import { beforeEach, describe, expect, onTestFinished, test, vi } from 'vitest'; import { CloudflareClient } from '../../../src/client'; import { withSentry } from '../../../src/withSentry'; +import { resetSdk } from '../../testUtils'; const MOCK_ENV = { SENTRY_DSN: 'https://public@dsn.ingest.sentry.io/1337', @@ -58,6 +59,7 @@ function addDelayedWaitUntil(context: ExecutionContext) { describe('instrumentTail', () => { beforeEach(() => { vi.clearAllMocks(); + resetSdk(); }); test('does not double-wrap when withSentry is called twice', async () => { @@ -260,7 +262,7 @@ describe('instrumentTail', () => { }, } satisfies ExportedHandler; - const wrappedHandler = withSentry(vi.fn(), handler); + const wrappedHandler = withSentry(() => ({ cacheClient: false }), handler); const waits: Promise[] = []; const waitUntil = vi.fn(promise => waits.push(promise)); await wrappedHandler.tail?.(createMockTailEvent(), MOCK_ENV_WITHOUT_DSN, { diff --git a/packages/cloudflare/test/request.test.ts b/packages/cloudflare/test/request.test.ts index 61f43e7c91e7..7baba20188e3 100644 --- a/packages/cloudflare/test/request.test.ts +++ b/packages/cloudflare/test/request.test.ts @@ -10,6 +10,7 @@ import type { CloudflareOptions } from '../src/client'; import { CloudflareClient } from '../src/client'; import { httpServerIntegration } from '../src/integrations/httpServer'; import { wrapRequestHandler } from '../src/request'; +import { _clearGlobalClientCache, init } from '../src/sdk'; const MOCK_OPTIONS: CloudflareOptions = { dsn: 'https://public@dsn.ingest.sentry.io/1337', @@ -976,3 +977,195 @@ describe('flushAndDispose', () => { disposeSpy.mockRestore(); }); }); + +function createMockDOContext(): ExecutionContext { + return { + waitUntil: vi.fn(), + passThroughOnException: vi.fn(), + storage: {}, + } as unknown as ExecutionContext; +} + +describe('Durable Object (DO) context', () => { + test('DO handler registers teardown via waitUntil', async () => { + const context = createMockDOContext(); + const waitUntilSpy = vi.spyOn(context, 'waitUntil'); + + // Send a body with a content-length so the response is treated as non-streaming + // and teardown runs at the handler boundary rather than on stream completion. + const result = await wrapRequestHandler( + { options: MOCK_OPTIONS, request: new Request('https://example.com'), context }, + () => new Response('test', { headers: { 'content-type': 'application/json' } }), + ); + + expect(result.status).toBe(200); + // Teardown is registered via waitUntil (a DurableObjectState.waitUntil exists + // for API compatibility and still runs the passed promise) + expect(waitUntilSpy).toHaveBeenCalled(); + }); + + test('DO handler error path registers teardown via waitUntil', async () => { + const context = createMockDOContext(); + const waitUntilSpy = vi.spyOn(context, 'waitUntil'); + const flushSpy = vi.spyOn(SentryCore.Client.prototype, 'flush').mockResolvedValue(true); + + try { + await wrapRequestHandler({ options: MOCK_OPTIONS, request: new Request('https://example.com'), context }, () => { + throw new Error('test error'); + }); + } catch { + // Expected + } + + // Teardown is registered via waitUntil on error too + expect(waitUntilSpy).toHaveBeenCalled(); + // And flush runs as part of that teardown + expect(flushSpy).toHaveBeenCalled(); + + flushSpy.mockRestore(); + }); + + test('DO handler for OPTIONS registers teardown via waitUntil', async () => { + const context = createMockDOContext(); + const waitUntilSpy = vi.spyOn(context, 'waitUntil'); + const flushSpy = vi.spyOn(SentryCore.Client.prototype, 'flush').mockResolvedValue(true); + + await wrapRequestHandler( + { + options: MOCK_OPTIONS, + request: new Request('https://example.com', { method: 'OPTIONS' }), + context, + }, + () => new Response('', { status: 200 }), + ); + + expect(waitUntilSpy).toHaveBeenCalled(); + expect(flushSpy).toHaveBeenCalled(); + + flushSpy.mockRestore(); + }); + + test('DO handler for HEAD registers teardown via waitUntil', async () => { + const context = createMockDOContext(); + const waitUntilSpy = vi.spyOn(context, 'waitUntil'); + const flushSpy = vi.spyOn(SentryCore.Client.prototype, 'flush').mockResolvedValue(true); + + await wrapRequestHandler( + { + options: MOCK_OPTIONS, + request: new Request('https://example.com', { method: 'HEAD' }), + context, + }, + () => new Response('', { status: 200 }), + ); + + expect(waitUntilSpy).toHaveBeenCalled(); + expect(flushSpy).toHaveBeenCalled(); + + flushSpy.mockRestore(); + }); + + test('DO handler for streaming response registers teardown via waitUntil', async () => { + const context = createMockDOContext(); + const waitUntilSpy = vi.spyOn(context, 'waitUntil'); + const flushSpy = vi.spyOn(SentryCore.Client.prototype, 'flush').mockResolvedValue(true); + + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('chunk1')); + controller.enqueue(new TextEncoder().encode('chunk2')); + controller.close(); + }, + }); + + const result = await wrapRequestHandler( + { options: MOCK_OPTIONS, request: new Request('https://example.com'), context }, + () => new Response(stream), + ); + + await result.text(); + + // Teardown is registered via waitUntil + expect(waitUntilSpy).toHaveBeenCalled(); + // And flush runs as part of that teardown + expect(flushSpy).toHaveBeenCalled(); + + flushSpy.mockRestore(); + }); + + test('DO handler for protocol upgrade (101) registers teardown via waitUntil', async () => { + const context = createMockDOContext(); + const waitUntilSpy = vi.spyOn(context, 'waitUntil'); + const flushSpy = vi.spyOn(CloudflareClient.prototype, 'flush').mockResolvedValue(true); + const disposeSpy = vi.spyOn(CloudflareClient.prototype, 'dispose'); + + const mockWebSocketResponse = { + status: 101, + statusText: 'Switching Protocols', + headers: new Headers(), + body: null, + ok: false, + redirected: false, + type: 'basic' as ResponseType, + url: '', + clone: () => mockWebSocketResponse, + arrayBuffer: () => Promise.resolve(new ArrayBuffer(0)), + blob: () => Promise.resolve(new Blob()), + formData: () => Promise.resolve(new FormData()), + json: () => Promise.resolve({}), + text: () => Promise.resolve(''), + bodyUsed: false, + bytes: () => Promise.resolve(new Uint8Array()), + } as Response; + + await wrapRequestHandler( + { options: MOCK_OPTIONS, request: new Request('https://example.com'), context }, + () => mockWebSocketResponse, + ); + + // Teardown is registered via waitUntil + expect(waitUntilSpy).toHaveBeenCalled(); + // Flush runs as part of that teardown + expect(flushSpy).toHaveBeenCalled(); + // Dispose should NOT be called for 101 + expect(disposeSpy).not.toHaveBeenCalled(); + + flushSpy.mockRestore(); + disposeSpy.mockRestore(); + }); +}); + +describe('cached client (cacheClient)', () => { + beforeEach(() => { + _clearGlobalClientCache(); + }); + + // `init()` resolves defaults into the options object it is given, so each call + // needs a fresh object to fingerprint identically — exactly like real callers, + // which build their options per invocation. + const makeOptions = (dsn?: string): CloudflareOptions => ({ + dsn: dsn ?? MOCK_OPTIONS.dsn, + beforeSend() { + return null; + }, + }); + + test('returns the same cached client for the same options', async () => { + const client1 = init(makeOptions()); + const client2 = init(makeOptions()); + expect(client2).toBe(client1); + }); + + test('returns the isolate client even when a later init uses a different DSN', async () => { + const client1 = init(makeOptions()); + const client2 = init(makeOptions('https://other@dsn.ingest.sentry.io/9999')); + expect(client2).toBe(client1); + }); + + test('clears cache with _clearGlobalClientCache', async () => { + const client1 = init(makeOptions()); + _clearGlobalClientCache(); + const client2 = init(makeOptions()); + expect(client2).not.toBe(client1); + }); +}); diff --git a/packages/cloudflare/test/sdk.test.ts b/packages/cloudflare/test/sdk.test.ts index 23f057d4eee3..bc27aa9202d7 100644 --- a/packages/cloudflare/test/sdk.test.ts +++ b/packages/cloudflare/test/sdk.test.ts @@ -1,7 +1,8 @@ import * as SentryCore from '@sentry/core'; -import type { Integration } from '@sentry/core'; +import type { Envelope, Integration } from '@sentry/core'; import { getClient } from '@sentry/core'; import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; +import type { CloudflareOptions } from '../src/client'; import { CloudflareClient } from '../src/client'; import { getDefaultIntegrations, init } from '../src/sdk'; import { resetSdk } from './testUtils'; @@ -60,6 +61,210 @@ describe('init', () => { }); }); +describe('cacheClient', () => { + beforeEach(() => { + resetSdk(); + }); + + afterEach(() => { + vi.unstubAllGlobals(); + }); + + const TEST_ENVELOPE = [ + { event_id: 'aa3ff046696b4bc6b609ce6d28fde9e2', sent_at: '2023-05-31T12:00:00.000Z' }, + [[{ type: 'event' }, { event_id: 'aa3ff046696b4bc6b609ce6d28fde9e2' }]], + ] as Envelope; + + test('returns the same client for repeated init with identical options', () => { + const options = { + dsn: 'https://public@dsn.ingest.sentry.io/1337', + } as const; + + const first = init({ ...options }); + const second = init({ ...options }); + + expect(second).toBe(first); + }); + + test('returns the isolate client when later init options differ', () => { + const first = init({ + dsn: 'https://public@dsn.ingest.sentry.io/1337', + tracesSampleRate: 0.5, + }); + const second = init({ + dsn: 'https://public@dsn.ingest.sentry.io/1337', + tracesSampleRate: 1, + }); + + expect(second).toBe(first); + }); + + test('re-binds the cached client to the current scope on repeated init', () => { + const options = { + dsn: 'https://public@dsn.ingest.sentry.io/1337', + } as const; + + const cached = init({ ...options }); + + // Simulate a competing init leaving a different client bound to the scope + SentryCore.getCurrentScope().setClient(undefined); + expect(getClient()).toBeUndefined(); + + const again = init({ ...options }); + expect(again).toBe(cached); + expect(getClient()).toBe(cached); + }); + + test('creates a fresh client when the cached one was disposed', () => { + const options = { + dsn: 'https://public@dsn.ingest.sentry.io/1337', + } as const; + + const cached = init({ ...options }); + cached?.dispose(); + + const again = init({ ...options }); + expect(again).toBeDefined(); + expect(again).not.toBe(cached); + expect(again?.getTransport()).toBeDefined(); + }); + + test('flushes eagerly when an envelope is sent on a cached client', async () => { + // The eager drain fires the buffered fetch, so stub out the network + vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response('ok'))); + + const client = init({ + dsn: 'https://public@dsn.ingest.sentry.io/1337', + }); + + const transport = client?.getTransport(); + expect(transport).toBeDefined(); + + const flushSpy = vi.spyOn(transport!, 'flush'); + await client!.sendEnvelope(TEST_ENVELOPE); + + expect(flushSpy).toHaveBeenCalled(); + }); + + test('does not flush eagerly when cacheClient is disabled', async () => { + const client = init({ dsn: 'https://public@dsn.ingest.sentry.io/1337', cacheClient: false }); + + const transport = client?.getTransport(); + expect(transport).toBeDefined(); + + const flushSpy = vi.spyOn(transport!, 'flush'); + await client!.sendEnvelope(TEST_ENVELOPE); + + expect(flushSpy).not.toHaveBeenCalled(); + }); + + // Logs and metrics batch client-side and the idle drain timer is disabled for this + // runtime, so unlike an event a capture alone never produces an envelope. A cached + // client never reaches an invocation-boundary flush, so without an eager drain these + // are dropped entirely — and silently, since errors keep working. + describe('log and metric delivery', () => { + function initWithCapturingTransport(options: Partial = {}) { + const envelopes: Envelope[] = []; + const client = init({ + dsn: 'https://public@dsn.ingest.sentry.io/1337', + enableLogs: true, + transport: () => ({ + send: (envelope: Envelope) => { + envelopes.push(envelope); + return Promise.resolve({}); + }, + flush: () => Promise.resolve(true), + }), + ...options, + })!; + + return { client, envelopes }; + } + + const itemTypes = (envelopes: Envelope[]): string[] => + envelopes.map(envelope => (envelope[1]?.[0]?.[0] as { type: string })?.type); + + test('delivers a log captured on a cached client without an explicit flush', async () => { + const { envelopes } = initWithCapturingTransport(); + + SentryCore.logger.info('detached log'); + await vi.waitFor(() => expect(itemTypes(envelopes)).toContain('log')); + }); + + test('delivers a metric captured on a cached client without an explicit flush', async () => { + const { envelopes } = initWithCapturingTransport(); + + SentryCore.metrics.count('detached_metric', 1); + await vi.waitFor(() => expect(itemTypes(envelopes)).toContain('trace_metric')); + }); + + test('coalesces a synchronous burst of logs into a single envelope', async () => { + const { envelopes } = initWithCapturingTransport(); + + for (let i = 0; i < 5; i++) { + SentryCore.logger.info(`burst ${i}`); + } + + await vi.waitFor(() => expect(itemTypes(envelopes)).toContain('log')); + expect(itemTypes(envelopes).filter(type => type === 'log')).toHaveLength(1); + }); + + test('keeps batching logs until flush for a non-cached client', async () => { + const { client, envelopes } = initWithCapturingTransport({ cacheClient: false }); + + SentryCore.logger.info('batched log'); + await new Promise(resolve => setTimeout(resolve, 10)); + expect(envelopes).toHaveLength(0); + + await client.flush(0); + expect(itemTypes(envelopes)).toContain('log'); + }); + + test('flush() delivers buffered logs on a cached client', async () => { + const { client, envelopes } = initWithCapturingTransport(); + + SentryCore.logger.info('tail log'); + await client.flush(0); + + expect(itemTypes(envelopes)).toContain('log'); + }); + }); + + test('applies initialScope on every cached init, not just the first', () => { + const options = { + dsn: 'https://public@dsn.ingest.sentry.io/1337', + } as const; + + init({ ...options }); + SentryCore.getCurrentScope().clear(); + + init({ ...options, initialScope: { tags: { from: 'initialScope' } } }); + + expect(SentryCore.getCurrentScope().getScopeData().tags).toEqual({ from: 'initialScope' }); + }); + + test('does not instrument ctx.waitUntil with the flush lock for cached clients', () => { + const waitUntil = vi.fn(); + const context = { waitUntil, passThroughOnException: vi.fn() }; + + init({ + dsn: 'https://public@dsn.ingest.sentry.io/1337', + ctx: context, + }); + + expect(context.waitUntil).toBe(waitUntil); + }); + + test('instruments ctx.waitUntil with the flush lock for non-cached clients', () => { + const waitUntil = vi.fn(); + const context = { waitUntil, passThroughOnException: vi.fn() }; + + init({ dsn: 'https://public@dsn.ingest.sentry.io/1337', cacheClient: false, ctx: context }); + + expect(context.waitUntil).not.toBe(waitUntil); + }); +}); + describe('getDefaultIntegrations', () => { afterEach(() => { delete globalThis.__SENTRY_ORCHESTRION__; diff --git a/packages/cloudflare/test/testUtils.ts b/packages/cloudflare/test/testUtils.ts index 8dcd3d43a4d9..dfe612f911ac 100644 --- a/packages/cloudflare/test/testUtils.ts +++ b/packages/cloudflare/test/testUtils.ts @@ -1,5 +1,6 @@ import { context, propagation, trace } from '@opentelemetry/api'; import { getCurrentScope, getGlobalScope, getIsolationScope } from '@sentry/core'; +import { _clearGlobalClientCache } from '../src/clientCache'; function resetGlobals(): void { getCurrentScope().clear(); @@ -18,4 +19,5 @@ function cleanupOtel(): void { export function resetSdk(): void { resetGlobals(); cleanupOtel(); + _clearGlobalClientCache(); } diff --git a/packages/cloudflare/test/utils/invocationContext.test.ts b/packages/cloudflare/test/utils/invocationContext.test.ts new file mode 100644 index 000000000000..d560b21e9707 --- /dev/null +++ b/packages/cloudflare/test/utils/invocationContext.test.ts @@ -0,0 +1,76 @@ +import { getDefaultIsolationScope, getIsolationScope, GLOBAL_OBJ, withIsolationScope } from '@sentry/core'; +import { AsyncLocalStorage } from 'async_hooks'; +import { setAsyncLocalStorageAsyncContextStrategy } from '@sentry/server-utils/no-diagnostic-channels'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { getInvocationState, setInvocationState } from '../../src/utils/invocationContext'; +import { withInvocationIsolationScope } from '../../src/utils/invocationScope'; + +describe('invocation state', () => { + beforeEach(() => { + (GLOBAL_OBJ as never).AsyncLocalStorage = AsyncLocalStorage; + setAsyncLocalStorageAsyncContextStrategy(); + }); + + it('returns undefined outside any invocation', () => { + expect(getInvocationState()).toBeUndefined(); + }); + + it('returns undefined for a forked scope that carries no state', () => { + withIsolationScope(getDefaultIsolationScope().clone(), () => { + expect(getInvocationState()).toBeUndefined(); + }); + }); + + it('exposes state attached to the active isolation scope', () => { + const ctx = { waitUntil: vi.fn(), passThroughOnException: vi.fn() }; + const scope = getDefaultIsolationScope().clone(); + setInvocationState(scope, { ctx }); + + withIsolationScope(scope, () => { + expect(getInvocationState()?.ctx).toBe(ctx); + }); + + expect(getInvocationState()).toBeUndefined(); + }); + + it('is not inherited by scope clones', () => { + const scope = getDefaultIsolationScope().clone(); + setInvocationState(scope, { ctx: undefined }); + + withIsolationScope(scope.clone(), () => { + expect(getInvocationState()).toBeUndefined(); + }); + }); + + it('is attached at the invocation entry point and kept when reentrant', () => { + const outerCtx = { waitUntil: vi.fn(), passThroughOnException: vi.fn() }; + const innerCtx = { waitUntil: vi.fn(), passThroughOnException: vi.fn() }; + + withInvocationIsolationScope(() => { + expect(getInvocationState()?.ctx).toBe(outerCtx); + + withInvocationIsolationScope(() => { + expect(getInvocationState()?.ctx).toBe(outerCtx); + }, innerCtx); + }, outerCtx); + }); + + it('isolates state between concurrent invocations', async () => { + const ctxA = { waitUntil: vi.fn(), passThroughOnException: vi.fn() }; + const ctxB = { waitUntil: vi.fn(), passThroughOnException: vi.fn() }; + + await Promise.all([ + withInvocationIsolationScope(async () => { + await new Promise(resolve => setTimeout(resolve, 10)); + expect(getInvocationState()?.ctx).toBe(ctxA); + }, ctxA), + withInvocationIsolationScope(async () => { + await new Promise(resolve => setTimeout(resolve, 5)); + expect(getInvocationState()?.ctx).toBe(ctxB); + }, ctxB), + ]); + + expect(getIsolationScope()).toBe(getDefaultIsolationScope()); + expect(getInvocationState()).toBeUndefined(); + }); +}); diff --git a/packages/cloudflare/test/workflow.test.ts b/packages/cloudflare/test/workflow.test.ts index 2578c1e0343d..0e36e15ab5d7 100644 --- a/packages/cloudflare/test/workflow.test.ts +++ b/packages/cloudflare/test/workflow.test.ts @@ -3,6 +3,7 @@ import { startSpan } from '@sentry/core'; import type { WorkflowEvent, WorkflowStep, WorkflowStepConfig } from 'cloudflare:workers'; import { beforeEach, describe, expect, test, vi } from 'vitest'; import { deterministicTraceIdFromInstanceId, instrumentWorkflowWithSentry } from '../src/workflows'; +import { resetSdk } from './testUtils'; vi.mock('../src/instrumentations/worker/instrumentEnv', () => ({ instrumentEnv: vi.fn((env: unknown) => env), @@ -104,6 +105,7 @@ async function drainWaitUntilLikeCloudflareVitestPool( describe.skipIf(NODE_MAJOR_VERSION < 20)('workflows', () => { beforeEach(() => { + resetSdk(); vi.clearAllMocks(); }); @@ -133,8 +135,10 @@ describe.skipIf(NODE_MAJOR_VERSION < 20)('workflows', () => { expect(mockStep.do).toHaveBeenCalledTimes(1); expect(mockStep.do).toHaveBeenCalledWith('first step', expect.any(Function)); - // We flush after the step.do and at the end of the run - expect(mockContext.waitUntil).toHaveBeenCalledTimes(2); + // We flush after the step.do and at the end of the run, plus one + // waitUntil registration for the eagerly delivered envelope + // and one for the envelope send itself + expect(mockContext.waitUntil).toHaveBeenCalledTimes(4); expect(mockContext.waitUntil).toHaveBeenCalledWith(expect.any(Promise)); expect(mockTransport.send).toHaveBeenCalledTimes(1); expect(mockTransport.send).toHaveBeenCalledWith([ @@ -379,8 +383,10 @@ describe.skipIf(NODE_MAJOR_VERSION < 20)('workflows', () => { expect(mockStep.do).toHaveBeenCalledTimes(1); expect(mockStep.do).toHaveBeenCalledWith('first step', expect.any(Function)); - // We flush after the step.do and at the end of the run - expect(mockContext.waitUntil).toHaveBeenCalledTimes(2); + // We flush after the step.do and at the end of the run, plus one + // waitUntil registration for the eagerly delivered envelope + // and one for the envelope send itself + expect(mockContext.waitUntil).toHaveBeenCalledTimes(4); expect(mockContext.waitUntil).toHaveBeenCalledWith(expect.any(Promise)); expect(mockTransport.send).toHaveBeenCalledTimes(1); expect(mockTransport.send).toHaveBeenCalledWith([ @@ -453,8 +459,10 @@ describe.skipIf(NODE_MAJOR_VERSION < 20)('workflows', () => { expect(mockStep.do).toHaveBeenCalledTimes(1); expect(mockStep.do).toHaveBeenCalledWith('sometimes error step', expect.any(Function)); - // One flush for the failed attempt, one for the retry success, one at end of run - expect(mockContext.waitUntil).toHaveBeenCalledTimes(3); + // One flush for the failed attempt, one for the retry success, one at end of run, + // plus one waitUntil registration per eagerly delivered envelope + // and one per envelope send + expect(mockContext.waitUntil).toHaveBeenCalledTimes(7); expect(mockContext.waitUntil).toHaveBeenCalledWith(expect.any(Promise)); // No error event (not final attempt), only failed transaction + successful retry transaction expect(mockTransport.send).toHaveBeenCalledTimes(2); @@ -724,8 +732,10 @@ describe.skipIf(NODE_MAJOR_VERSION < 20)('workflows', () => { const event = { payload: {}, timestamp: new Date(), instanceId: INSTANCE_ID }; await workflow.run(event, mockStep); - // Flush after step.do and at end of run - expect(mockContext.waitUntil).toHaveBeenCalledTimes(2); + // Flush after step.do and at end of run, plus one + // waitUntil registration for the eagerly delivered envelope + // and one for the envelope send itself + expect(mockContext.waitUntil).toHaveBeenCalledTimes(4); expect(mockTransport.send).toHaveBeenCalledTimes(1); const sendArg = mockTransport.send.mock.calls[0]![0]; From ae63f9c571c0257c3312ebed606ab3cad46c3908 Mon Sep 17 00:00:00 2001 From: JPeer264 Date: Fri, 7 Aug 2026 15:43:39 +0200 Subject: [PATCH 2/3] fixup! feat(cloudflare): Add cacheClient to reuse the client across invocations --- .../cloudflare-integration-tests/runner.ts | 15 ++++++- .../suites/cache-client/test.ts | 18 ++++---- packages/cloudflare/src/baseSdk.ts | 30 ++++++++++++- packages/cloudflare/src/sdk.ts | 43 +------------------ packages/cloudflare/test/request.test.ts | 21 +++++++++ 5 files changed, 74 insertions(+), 53 deletions(-) diff --git a/dev-packages/cloudflare-integration-tests/runner.ts b/dev-packages/cloudflare-integration-tests/runner.ts index 40e864d0a817..19873330d82e 100644 --- a/dev-packages/cloudflare-integration-tests/runner.ts +++ b/dev-packages/cloudflare-integration-tests/runner.ts @@ -161,6 +161,7 @@ export function createRunner(...paths: string[]) { // controls whether envelopes are expected in predefined order or not let unordered = false; + let failOnUnexpected = false; if (!existsSync(testPath)) { throw new Error(`Test scenario not found: ${testPath}`); @@ -195,6 +196,10 @@ export function createRunner(...paths: string[]) { unordered = true; return this; }, + failOnUnexpected: function () { + failOnUnexpected = true; + return this; + }, ignore: function (...types: EnvelopeItemType[]) { types.forEach(t => ignored.add(t)); return this; @@ -222,6 +227,7 @@ export function createRunner(...paths: string[]) { const expectedEnvelopeCount = expectedEnvelopes.length; let envelopeCount = 0; + let unexpectedEnvelopeError: Error | undefined; const envelopeWaiters: { expected: Expected; resolve: () => void; reject: (e: unknown) => void }[] = []; const { resolve: setWorkerPort, @@ -297,6 +303,10 @@ export function createRunner(...paths: string[]) { // no match found if (matchIndex < 0) { + if (failOnUnexpected) { + unexpectedEnvelopeError ??= new Error('Received an unexpected envelope'); + reject(unexpectedEnvelopeError); + } return; } @@ -429,7 +439,10 @@ export function createRunner(...paths: string[]) { return { completed: async function (): Promise { - return isComplete; + await isComplete; + if (unexpectedEnvelopeError) { + throw unexpectedEnvelopeError; + } }, makeRequest: async function ( method: 'get' | 'post', diff --git a/dev-packages/cloudflare-integration-tests/suites/cache-client/test.ts b/dev-packages/cloudflare-integration-tests/suites/cache-client/test.ts index dde6864391c6..38fcdce95913 100644 --- a/dev-packages/cloudflare-integration-tests/suites/cache-client/test.ts +++ b/dev-packages/cloudflare-integration-tests/suites/cache-client/test.ts @@ -134,17 +134,19 @@ it('cacheClient: false - repro #22545: detached work events are silently dropped it('cacheClient: true - dedupe drops the same error across invocations', async ({ signal }) => { // A shared client shares its dedupe state, so the same error captured by two separate // invocations is reported only once — the second is dropped as a duplicate. - const runner = createRunner(__dirname).ignore('transaction', 'span').start(signal); - - await runner.makeRequestAndWaitForEnvelope( - 'get', - '/cache/dedupe?id=dedupe-shared', - errorEventExpectation('Same error', CAPTURE_MECHANISM), - ); + const runner = createRunner(__dirname) + .ignore('transaction', 'span') + .unordered() + .failOnUnexpected() + .expect(errorEventExpectation('Same error', CAPTURE_MECHANISM)) + .start(signal); - // Second and third invocations capture the same error, but dedupe drops them. + // All three invocations are made without per-request waiters, while the runner requires the + // single expected error and rejects if either duplicate is delivered unexpectedly. + await runner.makeRequest('get', '/cache/dedupe?id=dedupe-shared'); await runner.makeRequest('get', '/cache/dedupe?id=dedupe-shared'); await runner.makeRequest('get', '/cache/dedupe?id=dedupe-shared'); + await runner.completed(); }); it('cacheClient: false - dedupe does not persist across invocations', async ({ signal }) => { diff --git a/packages/cloudflare/src/baseSdk.ts b/packages/cloudflare/src/baseSdk.ts index 6443ec67ce70..7e82be99c4de 100644 --- a/packages/cloudflare/src/baseSdk.ts +++ b/packages/cloudflare/src/baseSdk.ts @@ -1,4 +1,5 @@ import type { Integration } from '@sentry/core'; +import { getCurrentScope, setCurrentClient } from '@sentry/core'; import { consoleIntegration, conversationIdIntegration, @@ -15,6 +16,7 @@ import { import type { CloudflareClientOptions, CloudflareOptions } from './client'; import { CloudflareClient } from './client'; import { makeFlushLock } from './flush'; +import { cacheClient, getCachedClient } from './clientCache'; import { fetchIntegration } from './integrations/fetch'; import { httpServerIntegration } from './integrations/httpServer'; import { INTEGRATION_NAME as SPOTLIGHT_INTEGRATION_NAME, spotlightIntegration } from './integrations/spotlight'; @@ -77,12 +79,31 @@ export function getBaseDefaultIntegrations(options: CloudflareOptions): Integrat * Node.js-only code. `request.ts` — which backs both `wrapRequestHandler` and the * `@sentry/cloudflare/request` entry point, and therefore has to work on runtimes without the * `nodejs_compat` compatibility flag — creates its client from here instead of from `sdk.ts`. + * + * The client is cached and reused across invocations within the same isolate, + * unless `cacheClient: false` is passed. This avoids the + * per-invocation cost of constructing a new client, and it is what makes + * Durable Object telemetry reliable: a per-invocation client is disposed at + * the end of the handler, and in a Durable Object there is no `waitUntil` + * boundary that reliably extends execution, so spans/events that end after + * disposal would otherwise be lost. */ export function initWithDefaultIntegrations( options: CloudflareOptions, getDefaultIntegrationsImpl: (options: CloudflareOptions) => Integration[], - { skipFlushLock = false }: { skipFlushLock?: boolean } = {}, ): CloudflareClient | undefined { + const cacheEnabled = options.cacheClient !== false && Boolean(options.dsn); + + if (cacheEnabled) { + const cached = getCachedClient(); + if (cached?.getTransport()) { + getCurrentScope().update(options.initialScope); + setCurrentClient(cached); + cached.setExecutionContext(options.ctx); + return cached; + } + } + if (options.defaultIntegrations === undefined) { options.defaultIntegrations = getDefaultIntegrationsImpl(options); } @@ -91,11 +112,12 @@ export function initWithDefaultIntegrations( // invocation's flush lock would make later flushes wait on that invocation's // waitUntil work forever. Eager delivery replaces the flush lock's purpose. const invocationContext = options.ctx; - const flushLock = !skipFlushLock && invocationContext ? makeFlushLock(invocationContext) : undefined; + const flushLock = !cacheEnabled && invocationContext ? makeFlushLock(invocationContext) : undefined; delete options.ctx; const clientOptions: CloudflareClientOptions = { ...options, + cacheClient: cacheEnabled, stackParser: stackParserFromStackParserOptions(options.stackParser || defaultStackParser), integrations: getIntegrationsToSetup(options), transport: options.transport || makeCloudflareTransport, @@ -124,6 +146,10 @@ export function initWithDefaultIntegrations( const client = initAndBind(CloudflareClient, clientOptions) as CloudflareClient; + if (cacheEnabled && client && options.dsn) { + cacheClient(client); + } + // An instrumented module that first evaluates AFTER this init (e.g. a driver // lazily required on first use) stores its subscriber factory on the global // marker too late for the default-integrations snapshot above. Its injected diff --git a/packages/cloudflare/src/sdk.ts b/packages/cloudflare/src/sdk.ts index cf09560f945a..cf08fa8e717d 100644 --- a/packages/cloudflare/src/sdk.ts +++ b/packages/cloudflare/src/sdk.ts @@ -1,9 +1,7 @@ import type { Integration } from '@sentry/core'; -import { getCurrentScope, setCurrentClient } from '@sentry/core'; import { vercelAIIntegration } from './integrations/tracing/vercelai'; import { getBaseDefaultIntegrations, initWithDefaultIntegrations } from './baseSdk'; import type { CloudflareClient, CloudflareOptions } from './client'; -import { cacheClient, getCachedClient } from './clientCache'; // Test-only helper, re-exported here so tests can reset the global client cache. export { _clearGlobalClientCache } from './clientCache'; @@ -25,46 +23,7 @@ export function getDefaultIntegrations(options: CloudflareOptions): Integration[ /** * Initializes the cloudflare SDK. - * - * The client is cached and reused across invocations within the same isolate, - * unless `cacheClient: false` is passed. This avoids the - * per-invocation cost of constructing a new client, and it is what makes - * Durable Object telemetry reliable: a per-invocation client is disposed at - * the end of the handler, and in a Durable Object there is no `waitUntil` - * boundary that reliably extends execution, so spans/events that end after - * disposal would otherwise be lost. */ export function init(options: CloudflareOptions): CloudflareClient | undefined { - const cacheEnabled = options.cacheClient !== false; - - if (cacheEnabled) { - // Normalize the flag so the client marks itself as cached. - options.cacheClient = true; - } - - if (cacheEnabled && options.dsn) { - const cached = getCachedClient(); - // A cached client that has lost its transport was disposed. Replace it rather - // than returning a dead client for the rest of the isolate's lifetime. - if (cached?.getTransport()) { - // Mirror the two scope side effects of `initAndBind`, which only runs on first - // creation. Without the re-bind the scope keeps whatever client a previous init - // left behind — which may have been disposed since — and without the update - // `initialScope` would apply only to an isolate's very first invocation. - getCurrentScope().update(options.initialScope); - setCurrentClient(cached); - // The cached client outlives the invocation that created it, so its eager - // sends must be registered with the current invocation's waitUntil. - cached.setExecutionContext(options.ctx); - return cached; - } - } - - const client = initWithDefaultIntegrations(options, getDefaultIntegrations, { skipFlushLock: cacheEnabled }); - - if (cacheEnabled && client && options.dsn) { - cacheClient(client); - } - - return client; + return initWithDefaultIntegrations(options, getDefaultIntegrations); } diff --git a/packages/cloudflare/test/request.test.ts b/packages/cloudflare/test/request.test.ts index 7baba20188e3..77ca82eae945 100644 --- a/packages/cloudflare/test/request.test.ts +++ b/packages/cloudflare/test/request.test.ts @@ -15,6 +15,7 @@ import { _clearGlobalClientCache, init } from '../src/sdk'; const MOCK_OPTIONS: CloudflareOptions = { dsn: 'https://public@dsn.ingest.sentry.io/1337', traceLifecycle: 'static', + cacheClient: false, }; const NODE_MAJOR_VERSION = parseInt(process.versions.node.split('.')[0]!); @@ -1150,6 +1151,26 @@ describe('cached client (cacheClient)', () => { }, }); + test('wrapRequestHandler reuses a client when cacheClient is enabled', async () => { + const initAndBindSpy = vi.spyOn(SentryCore, 'initAndBind'); + const options = { ...MOCK_OPTIONS, cacheClient: true }; + + await wrapRequestHandler( + { options, request: new Request('https://example.com/first'), context: createMockExecutionContext() }, + () => new Response('first'), + ); + await wrapRequestHandler( + { + options: { ...options }, + request: new Request('https://example.com/second'), + context: createMockExecutionContext(), + }, + () => new Response('second'), + ); + + expect(initAndBindSpy).toHaveBeenCalledTimes(1); + }); + test('returns the same cached client for the same options', async () => { const client1 = init(makeOptions()); const client2 = init(makeOptions()); From 0ba7c001ae7dbf36a643e765b6d2934b89c2e411 Mon Sep 17 00:00:00 2001 From: JPeer264 Date: Fri, 7 Aug 2026 15:50:51 +0200 Subject: [PATCH 3/3] chore: Update size-limit --- .size-limit.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.size-limit.js b/.size-limit.js index 973fb3570c37..0bd2152d6a08 100644 --- a/.size-limit.js +++ b/.size-limit.js @@ -477,7 +477,7 @@ module.exports = [ ignore: [...builtinModules, ...nodePrefixedBuiltinModules], gzip: false, brotli: false, - limit: '524 KiB', + limit: '530 KiB', disablePlugins: ['@size-limit/webpack'], webpack: false, modifyEsbuildConfig: function (config) {