From 643d7511c20c537d667b43db1e7365229c280774 Mon Sep 17 00:00:00 2001 From: Matthew Jones Date: Thu, 17 Sep 2026 15:56:47 -0600 Subject: [PATCH 1/6] fix(cloudflare): Enforce flush timeout across Workflow lifecycle Cloudflare's flush deadline could expire while transport fetches remained active, Workflow steps could inherit the previous step's flush point and schedule redundant eager drains, and flush-lock finalization could wait outside the configured timeout. Abort requests when a transport drain expires, reset the flush point at each Workflow step boundary, and share one deadline across flush-lock, pending-span, and transport phases. Add focused regression coverage for all three paths and update timer-based tests to await the user task before asserting teardown. Fixes #24482 Co-authored-by: OpenAI Codex --- packages/cloudflare/src/client.ts | 42 ++++++++++- packages/cloudflare/src/transport.ts | 61 ++++++++++++++-- packages/cloudflare/src/workflows.ts | 9 +++ packages/cloudflare/test/client.test.ts | 24 +++++- .../worker/instrumentEmail.test.ts | 2 +- .../worker/instrumentFetch.test.ts | 2 +- .../worker/instrumentQueue.test.ts | 2 +- .../worker/instrumentScheduled.test.ts | 2 +- .../worker/instrumentTail.test.ts | 2 +- packages/cloudflare/test/request.test.ts | 5 +- packages/cloudflare/test/transport.test.ts | 73 +++++++++++++++++++ packages/cloudflare/test/workflow.test.ts | 29 +++++++- 12 files changed, 233 insertions(+), 20 deletions(-) diff --git a/packages/cloudflare/src/client.ts b/packages/cloudflare/src/client.ts index a71445b03f92..928fcbcdb19f 100644 --- a/packages/cloudflare/src/client.ts +++ b/packages/cloudflare/src/client.ts @@ -15,6 +15,22 @@ import type { makeFlushLock } from './flush'; import type { CloudflareTransportOptions } from './transport'; import { getInvocationState, getInvocationWaitUntil } from './utils/invocationContext'; +async function waitForPromise(promise: PromiseLike, timeout: number): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + Promise.resolve(promise).then(() => true), + new Promise(resolve => { + timer = setTimeout(() => resolve(false), timeout); + }), + ]); + } finally { + if (timer) { + clearTimeout(timer); + } + } +} + /** * The Sentry Cloudflare SDK Client. * @@ -115,18 +131,34 @@ 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 { + const deadline = timeout && timeout > 0 ? Date.now() + timeout : undefined; + const remainingTimeout = (): number | undefined => + deadline === undefined ? timeout : Math.max(0, deadline - Date.now()); + // 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(); + const lockTimeout = remainingTimeout(); + if (lockTimeout && lockTimeout > 0) { + if (!(await waitForPromise(this._flushLock.finalize(), lockTimeout))) { + return false; + } + } else if (deadline !== undefined) { + return false; + } else { + await this._flushLock.finalize(); + } } if (this._pendingSpans.size > 0 && this._spanCompletionPromise) { DEBUG_BUILD && debug.log('[CloudflareClient] Waiting for', this._pendingSpans.size, 'pending spans to complete...'); - const timeoutMs = timeout ?? 5000; + const timeoutMs = remainingTimeout() ?? 5000; + if (deadline !== undefined && timeoutMs <= 0) { + return false; + } const spanCompletionRace = Promise.race([ this._spanCompletionPromise, new Promise(resolve => @@ -146,7 +178,11 @@ export class CloudflareClient extends ServerRuntimeClient { // of them would also start an eager drain and a `waitUntil` registration. this._inBoundaryFlush = true; try { - return await super.flush(timeout); + const transportTimeout = remainingTimeout(); + if (deadline !== undefined && (!transportTimeout || transportTimeout <= 0)) { + return false; + } + return await super.flush(transportTimeout); } finally { this._inBoundaryFlush = false; } diff --git a/packages/cloudflare/src/transport.ts b/packages/cloudflare/src/transport.ts index 25d9e05572b9..c9e95b30fc14 100644 --- a/packages/cloudflare/src/transport.ts +++ b/packages/cloudflare/src/transport.ts @@ -16,6 +16,9 @@ export interface CloudflareTransportOptions extends BaseTransportOptions { */ const DEFAULT_TRANSPORT_BUFFER_SIZE = 256; +type TaskProducer = () => PromiseLike; +type RunTask = (taskProducer: TaskProducer, signal: AbortSignal) => PromiseLike; + /** * This is a modified promise buffer that collects tasks until drain is called. * We need this in the edge runtime because edge function invocations may not share I/O objects, like fetch requests @@ -29,20 +32,23 @@ export class IsolatedPromiseBuffer { // If we ever remove it from the interface we should also remove it here. public $: Array>; - private _taskProducers: (() => PromiseLike)[]; + private _taskProducers: TaskProducer[]; private readonly _bufferSize: number; - public constructor(_bufferSize = DEFAULT_TRANSPORT_BUFFER_SIZE) { + private readonly _runTask: RunTask; + + public constructor(_bufferSize = DEFAULT_TRANSPORT_BUFFER_SIZE, _runTask: RunTask = taskProducer => taskProducer()) { this.$ = []; this._taskProducers = []; this._bufferSize = _bufferSize; + this._runTask = _runTask; } /** * @inheritdoc */ - public add(taskProducer: () => PromiseLike): PromiseLike { + public add(taskProducer: TaskProducer): PromiseLike { if (this._taskProducers.length >= this._bufferSize) { return Promise.reject(SENTRY_BUFFER_FULL_ERROR); } @@ -57,10 +63,13 @@ export class IsolatedPromiseBuffer { public drain(timeout?: number): PromiseLike { const oldTaskProducers = [...this._taskProducers]; this._taskProducers = []; + const drainController = new AbortController(); + const tasks = oldTaskProducers.map(taskProducer => this._runTask(taskProducer, drainController.signal)); return new Promise(resolve => { const timer = setTimeout(() => { if (timeout && timeout > 0) { + drainController.abort(); resolve(false); } }, timeout); @@ -68,8 +77,8 @@ export class IsolatedPromiseBuffer { // This cannot reject // eslint-disable-next-line @typescript-eslint/no-floating-promises Promise.all( - oldTaskProducers.map(taskProducer => - taskProducer().then(null, () => { + tasks.map(task => + task.then(null, () => { // catch all failed requests }), ), @@ -86,15 +95,36 @@ export class IsolatedPromiseBuffer { * Creates a Transport that uses the native fetch API to send events to Sentry. */ export function makeCloudflareTransport(options: CloudflareTransportOptions): Transport { + let activeDrainSignal: AbortSignal | undefined; + function makeRequest(request: TransportRequest): PromiseLike { + const controller = new AbortController(); + const callerSignal = options.fetchOptions?.signal; + const drainSignal = activeDrainSignal; + const abortFromCallerSignal = (): void => controller.abort(); + const abortFromDrainSignal = (): void => controller.abort(); + + if (callerSignal?.aborted) { + controller.abort(); + } else { + callerSignal?.addEventListener('abort', abortFromCallerSignal, { once: true }); + } + + if (drainSignal?.aborted) { + controller.abort(); + } else { + drainSignal?.addEventListener('abort', abortFromDrainSignal, { once: true }); + } + const requestOptions: RequestInit = { body: request.body as BodyInit, method: 'POST', headers: options.headers, ...options.fetchOptions, + signal: controller.signal, }; - return suppressTracing(() => { + const requestPromise = suppressTracing(() => { return (options.fetch ?? fetch)(options.url, requestOptions).then(async response => { // Consume the response body to satisfy Cloudflare Workers' fetch requirements. // The runtime requires all fetch response bodies to be read or explicitly canceled @@ -116,7 +146,24 @@ export function makeCloudflareTransport(options: CloudflareTransportOptions): Tr }; }); }); + + return Promise.resolve(requestPromise).finally(() => { + callerSignal?.removeEventListener('abort', abortFromCallerSignal); + drainSignal?.removeEventListener('abort', abortFromDrainSignal); + }); + } + + function runTaskWithinDrain( + taskProducer: TaskProducer, + signal: AbortSignal, + ): PromiseLike { + activeDrainSignal = signal; + try { + return taskProducer(); + } finally { + activeDrainSignal = undefined; + } } - return createTransport(options, makeRequest, new IsolatedPromiseBuffer(options.bufferSize)); + return createTransport(options, makeRequest, new IsolatedPromiseBuffer(options.bufferSize, runTaskWithinDrain)); } diff --git a/packages/cloudflare/src/workflows.ts b/packages/cloudflare/src/workflows.ts index c46470c54355..c5ce89614992 100644 --- a/packages/cloudflare/src/workflows.ts +++ b/packages/cloudflare/src/workflows.ts @@ -30,6 +30,7 @@ import { instrumentEnv } from './instrumentations/worker/instrumentEnv'; import { addCloudResourceContext } from './scope-utils'; import { init } from './sdk'; import { instrumentContext } from './utils/instrumentContext'; +import { getInvocationState } from './utils/invocationContext'; import type { DefaultEnv, ResolveEnv, StrictCloudflareOptions } from './types'; import { withInvocationIsolationScope } from './utils/invocationScope'; @@ -124,6 +125,14 @@ class WrappedWorkflowStep implements WorkflowStep { // run's isolation scope (and with it the invocation state that ties eager sends // to this invocation's `waitUntil`) has to be restored explicitly. return withIsolationScope(this._isolationScope, () => { + // Each Workflow step is its own RPC invocation with its own boundary flush. + // The isolation scope is shared across steps, so clear the previous step's + // flush point before capturing anything for this one. + const invocationState = getInvocationState(); + if (invocationState) { + invocationState.flushPointReached = false; + } + const stepResult = startSpan( { name, diff --git a/packages/cloudflare/test/client.test.ts b/packages/cloudflare/test/client.test.ts index 09bff574e479..a28cd735493d 100644 --- a/packages/cloudflare/test/client.test.ts +++ b/packages/cloudflare/test/client.test.ts @@ -266,7 +266,29 @@ describe('CloudflareClient', () => { releaseLock(); await flushPromise; - expect(privateClient._transport.flush).toHaveBeenCalledWith(1000); + expect(privateClient._transport.flush).toHaveBeenCalledWith(expect.any(Number)); + const transportTimeout = privateClient._transport.flush.mock.calls[0]?.[0]; + expect(transportTimeout).toBeGreaterThan(0); + expect(transportTimeout).toBeLessThanOrEqual(1000); + }); + + it('includes the flush lock in the timeout', async () => { + const finalize = vi.fn(() => new Promise(() => undefined)); + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + flushLock: { ready: Promise.resolve(), finalize }, + }); + + const privateClient = client as unknown as { + _transport: { flush: ReturnType }; + }; + const result = await Promise.race([ + client.flush(10), + new Promise<'did-not-settle'>(resolve => setTimeout(() => resolve('did-not-settle'), 30)), + ]); + + expect(result).toBe(false); + expect(privateClient._transport.flush).not.toHaveBeenCalled(); }); }); diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentEmail.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentEmail.test.ts index 2c15f5a39966..a12b29f6a3bb 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentEmail.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentEmail.test.ts @@ -302,7 +302,7 @@ describe('instrumentEmail', () => { } as unknown as ExecutionContext); expect(flush).not.toBeCalled(); expect(waitUntil).toBeCalled(); - vi.advanceTimersToNextTimer().runAllTimers(); + await vi.advanceTimersToNextTimerAsync(); await Promise.all(waits); expect(flush).toHaveBeenCalledOnce(); }); diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentFetch.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentFetch.test.ts index 1a0e94093444..d57687b8b2d8 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentFetch.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentFetch.test.ts @@ -172,7 +172,7 @@ describe('instrumentFetch', () => { .then(response => response.text()); expect(flush).not.toBeCalled(); expect(waitUntil).toBeCalled(); - vi.advanceTimersToNextTimer().runAllTimers(); + await vi.advanceTimersToNextTimerAsync(); await Promise.all(waits); expect(flush).toHaveBeenCalledOnce(); }); diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts index afe51790adfd..623d8825cea0 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts @@ -318,7 +318,7 @@ describe('instrumentQueue', () => { } as unknown as ExecutionContext); expect(flush).not.toBeCalled(); expect(waitUntil).toBeCalled(); - vi.advanceTimersToNextTimer().runAllTimers(); + await vi.advanceTimersToNextTimerAsync(); await Promise.all(waits); expect(flush).toHaveBeenCalledOnce(); }); diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentScheduled.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentScheduled.test.ts index 46090684daea..814d814583f4 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentScheduled.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentScheduled.test.ts @@ -323,7 +323,7 @@ describe('instrumentScheduled', () => { } as unknown as ExecutionContext); expect(flush).not.toBeCalled(); expect(waitUntil).toBeCalled(); - vi.advanceTimersToNextTimer().runAllTimers(); + await vi.advanceTimersToNextTimerAsync(); await Promise.all(waits); expect(flush).toHaveBeenCalledOnce(); }); diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentTail.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentTail.test.ts index 014916e6f158..7d16d2578362 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentTail.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentTail.test.ts @@ -270,7 +270,7 @@ describe('instrumentTail', () => { } as unknown as ExecutionContext); expect(flush).not.toBeCalled(); expect(waitUntil).toBeCalled(); - vi.advanceTimersToNextTimer().runAllTimers(); + await vi.advanceTimersToNextTimerAsync(); await Promise.all(waits); expect(flush).toHaveBeenCalledOnce(); }); diff --git a/packages/cloudflare/test/request.test.ts b/packages/cloudflare/test/request.test.ts index 052147f84bc4..885a48f4cfb0 100644 --- a/packages/cloudflare/test/request.test.ts +++ b/packages/cloudflare/test/request.test.ts @@ -150,7 +150,7 @@ describe('withSentry', () => { return new Response('test'); }).then(response => response.text()); expect(waitUntil).toBeCalled(); - vi.advanceTimersToNextTimer().runAllTimers(); + await vi.advanceTimersToNextTimerAsync(); await Promise.all(waits); const after = flushSpy.mock.calls.length; @@ -1051,6 +1051,7 @@ describe('Durable Object (DO) context', () => { // Teardown is registered via waitUntil on error too expect(waitUntilSpy).toHaveBeenCalled(); + await Promise.all(waitUntilSpy.mock.calls.map(([promise]) => promise)); // And flush runs as part of that teardown expect(flushSpy).toHaveBeenCalled(); @@ -1072,6 +1073,7 @@ describe('Durable Object (DO) context', () => { ); expect(waitUntilSpy).toHaveBeenCalled(); + await Promise.all(waitUntilSpy.mock.calls.map(([promise]) => promise)); expect(flushSpy).toHaveBeenCalled(); flushSpy.mockRestore(); @@ -1092,6 +1094,7 @@ describe('Durable Object (DO) context', () => { ); expect(waitUntilSpy).toHaveBeenCalled(); + await Promise.all(waitUntilSpy.mock.calls.map(([promise]) => promise)); expect(flushSpy).toHaveBeenCalled(); flushSpy.mockRestore(); diff --git a/packages/cloudflare/test/transport.test.ts b/packages/cloudflare/test/transport.test.ts index fdb9fbc5e30f..3044c1dffbef 100644 --- a/packages/cloudflare/test/transport.test.ts +++ b/packages/cloudflare/test/transport.test.ts @@ -52,6 +52,7 @@ describe('Edge Transport', () => { expect(mockFetch).toHaveBeenLastCalledWith(DEFAULT_EDGE_TRANSPORT_OPTIONS.url, { body: serializeEnvelope(ERROR_ENVELOPE), method: 'POST', + signal: expect.any(AbortSignal), }); }); @@ -104,6 +105,7 @@ describe('Edge Transport', () => { body: serializeEnvelope(ERROR_ENVELOPE), method: 'POST', ...REQUEST_OPTIONS, + signal: expect.any(AbortSignal), }); }); @@ -249,4 +251,75 @@ describe('IsolatedPromiseBuffer', () => { await transport.flush(); expect(customFetch).toHaveBeenCalledTimes(1); }); + + it('aborts fetch requests when their drain times out', async () => { + let signal: AbortSignal | undefined; + const customFetch = vi.fn( + (_input: RequestInfo | URL, init?: RequestInit): Promise => + new Promise((_resolve, reject) => { + signal = init?.signal ?? undefined; + signal?.addEventListener('abort', () => reject(new DOMException('Aborted', 'AbortError')), { once: true }); + }), + ); + const transport = makeCloudflareTransport({ ...DEFAULT_EDGE_TRANSPORT_OPTIONS, fetch: customFetch }); + + await transport.send(ERROR_ENVELOPE); + await expect(transport.flush(1)).resolves.toBe(false); + + expect(signal?.aborted).toBe(true); + }); + + it('preserves a caller-provided abort signal', async () => { + let signal: AbortSignal | undefined; + const callerController = new AbortController(); + const customFetch = vi.fn( + (_input: RequestInfo | URL, init?: RequestInit): Promise => + new Promise((_resolve, reject) => { + signal = init?.signal ?? undefined; + signal?.addEventListener('abort', () => reject(new DOMException('Aborted', 'AbortError')), { once: true }); + }), + ); + const transport = makeCloudflareTransport({ + ...DEFAULT_EDGE_TRANSPORT_OPTIONS, + fetch: customFetch, + fetchOptions: { signal: callerController.signal }, + }); + + await transport.send(ERROR_ENVELOPE); + const flush = transport.flush(); + callerController.abort(); + + await expect(flush).resolves.toBe(true); + expect(signal?.aborted).toBe(true); + }); + + it('does not abort requests belonging to another drain', async () => { + const signals: AbortSignal[] = []; + const resolveRequests: ((response: Response) => void)[] = []; + const customFetch = vi.fn( + (_input: RequestInfo | URL, init?: RequestInit): Promise => + new Promise((resolve, reject) => { + const signal = init?.signal as AbortSignal; + signals.push(signal); + resolveRequests.push(resolve); + signal.addEventListener('abort', () => reject(new DOMException('Aborted', 'AbortError')), { once: true }); + }), + ); + const transport = makeCloudflareTransport({ ...DEFAULT_EDGE_TRANSPORT_OPTIONS, fetch: customFetch }); + + await transport.send(ERROR_ENVELOPE); + const firstFlush = transport.flush(1000); + await transport.send(ERROR_ENVELOPE); + await expect(transport.flush(1)).resolves.toBe(false); + + expect(signals[0]?.aborted).toBe(false); + expect(signals[1]?.aborted).toBe(true); + + resolveRequests[0]?.({ + headers: new Headers(), + status: 200, + text: () => Promise.resolve('OK'), + } as unknown as Response); + await expect(firstFlush).resolves.toBe(true); + }); }); diff --git a/packages/cloudflare/test/workflow.test.ts b/packages/cloudflare/test/workflow.test.ts index f2ddfae9d8f0..ef9ebbd94941 100644 --- a/packages/cloudflare/test/workflow.test.ts +++ b/packages/cloudflare/test/workflow.test.ts @@ -458,9 +458,9 @@ describe.skipIf(NODE_MAJOR_VERSION < 20)('workflows', () => { expect(mockStep.do).toHaveBeenCalledTimes(1); expect(mockStep.do).toHaveBeenCalledWith('sometimes error step', expect.any(Function)); - // One flush per attempt (failed and retried, past the span end) and one at end of - // run, plus one eager registration for the envelope of the error captured mid-run - expect(mockContext.waitUntil).toHaveBeenCalledTimes(4); + // One boundary flush per attempt (failed and retried) and one at the end of the run. + // The retry starts before its flush point, so its envelope does not add an eager flush. + expect(mockContext.waitUntil).toHaveBeenCalledTimes(3); expect(mockContext.waitUntil).toHaveBeenCalledWith(expect.any(Promise)); // No error event (not final attempt), only failed transaction + successful retry transaction expect(mockTransport.send).toHaveBeenCalledTimes(2); @@ -784,4 +784,27 @@ describe.skipIf(NODE_MAJOR_VERSION < 20)('workflows', () => { expect(tagInsideStep).toBe('marker'); expect(hasInvocationStateInsideStep).toBe(true); }); + + test('each workflow step starts before the invocation flush point', async () => { + const flushPointsAtStepStart: Array = []; + + class MultipleStepWorkflow { + constructor(_ctx: ExecutionContext, _env: unknown) {} + + async run(_event: Readonly>, step: WorkflowStep): Promise { + await step.do('first step', async () => { + flushPointsAtStepStart.push(getInvocationState()?.flushPointReached); + }); + await step.do('second step', async () => { + flushPointsAtStepStart.push(getInvocationState()?.flushPointReached); + }); + } + } + + const TestWorkflowInstrumented = instrumentWorkflowWithSentry(getSentryOptions, MultipleStepWorkflow as any); + const workflow = new TestWorkflowInstrumented(mockContext, {}) as MultipleStepWorkflow; + await workflow.run({ payload: {}, timestamp: new Date(), instanceId: INSTANCE_ID }, mockStep); + + expect(flushPointsAtStepStart).toEqual([false, false]); + }); }); From acd76c49a637a0b98dfad67f563cf1e1e0c4feac Mon Sep 17 00:00:00 2001 From: Matthew Jones Date: Thu, 17 Sep 2026 16:32:54 -0600 Subject: [PATCH 2/6] fix(cloudflare): Preserve transport budget during flush Share one deadline across the flush lock, pending spans, client processing, and transport drain so pre-drain timeouts cannot strand buffered envelopes. Add regression coverage for lock and span starvation and for the complete flush deadline. Co-authored-by: OpenAI Codex --- packages/cloudflare/src/client.ts | 49 +++++++++++---------- packages/cloudflare/test/client.test.ts | 58 +++++++++++++++++++++++-- 2 files changed, 82 insertions(+), 25 deletions(-) diff --git a/packages/cloudflare/src/client.ts b/packages/cloudflare/src/client.ts index 928fcbcdb19f..ceddb36516a0 100644 --- a/packages/cloudflare/src/client.ts +++ b/packages/cloudflare/src/client.ts @@ -134,43 +134,46 @@ export class CloudflareClient extends ServerRuntimeClient { const deadline = timeout && timeout > 0 ? Date.now() + timeout : undefined; const remainingTimeout = (): number | undefined => deadline === undefined ? timeout : Math.max(0, deadline - Date.now()); + const hasPendingSpans = this._pendingSpans.size > 0 && this._spanCompletionPromise; + let remainingStages = 2 + (this._flushLock ? 1 : 0) + (hasPendingSpans ? 1 : 0); + let preDrainWorkCompleted = true; + const timeoutForNextStage = (): number | undefined => { + const remaining = remainingTimeout(); + return remaining === undefined ? undefined : Math.floor(remaining / remainingStages); + }; // 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) { - const lockTimeout = remainingTimeout(); + const lockTimeout = timeoutForNextStage(); if (lockTimeout && lockTimeout > 0) { if (!(await waitForPromise(this._flushLock.finalize(), lockTimeout))) { - return false; + preDrainWorkCompleted = false; } } else if (deadline !== undefined) { - return false; + preDrainWorkCompleted = false; } else { await this._flushLock.finalize(); } + remainingStages--; } - if (this._pendingSpans.size > 0 && this._spanCompletionPromise) { + if (hasPendingSpans) { DEBUG_BUILD && debug.log('[CloudflareClient] Waiting for', this._pendingSpans.size, 'pending spans to complete...'); - const timeoutMs = remainingTimeout() ?? 5000; - if (deadline !== undefined && timeoutMs <= 0) { - return false; + const spanTimeout = timeoutForNextStage() ?? 5000; + if (spanTimeout > 0) { + if (!(await waitForPromise(hasPendingSpans, spanTimeout))) { + DEBUG_BUILD && + debug.log('[CloudflareClient] Span completion timeout after', spanTimeout, 'ms, flushing anyway'); + preDrainWorkCompleted = false; + } + } else { + preDrainWorkCompleted = false; } - const spanCompletionRace = Promise.race([ - this._spanCompletionPromise, - new Promise(resolve => - setTimeout(() => { - DEBUG_BUILD && - debug.log('[CloudflareClient] Span completion timeout after', timeoutMs, 'ms, flushing anyway'); - resolve(undefined); - }, timeoutMs), - ), - ]); - - await spanCompletionRace; + remainingStages--; } // Envelopes created while this flush drains (log/metric/span buffers turning into @@ -178,11 +181,13 @@ export class CloudflareClient extends ServerRuntimeClient { // of them would also start an eager drain and a `waitUntil` registration. this._inBoundaryFlush = true; try { - const transportTimeout = remainingTimeout(); - if (deadline !== undefined && (!transportTimeout || transportTimeout <= 0)) { + // BaseClient.flush uses its timeout once for client processing and once for + // transport draining, so give it one stage's share for each internal wait. + const baseFlushTimeout = timeoutForNextStage(); + if (deadline !== undefined && (!baseFlushTimeout || baseFlushTimeout <= 0)) { return false; } - return await super.flush(transportTimeout); + return (await super.flush(baseFlushTimeout)) && preDrainWorkCompleted; } finally { this._inBoundaryFlush = false; } diff --git a/packages/cloudflare/test/client.test.ts b/packages/cloudflare/test/client.test.ts index a28cd735493d..2e1e2e6d5734 100644 --- a/packages/cloudflare/test/client.test.ts +++ b/packages/cloudflare/test/client.test.ts @@ -225,7 +225,7 @@ describe('CloudflareClient', () => { }); describe('flush()', () => { - it('calls transport flush with the given timeout', async () => { + it('shares the timeout between client processing and transport flushing', async () => { const client = new CloudflareClient(MOCK_CLIENT_OPTIONS); const privateClient = client as unknown as { @@ -234,7 +234,7 @@ describe('CloudflareClient', () => { await client.flush(3000); - expect(privateClient._transport.flush).toHaveBeenCalledWith(3000); + expect(privateClient._transport.flush).toHaveBeenCalledWith(1500); }); it('resolves with the transport flush result', async () => { @@ -288,7 +288,59 @@ describe('CloudflareClient', () => { ]); expect(result).toBe(false); - expect(privateClient._transport.flush).not.toHaveBeenCalled(); + expect(privateClient._transport.flush).toHaveBeenCalledOnce(); + }); + + it('still drains the transport when pending spans consume their wait budget', async () => { + const client = new CloudflareClient(MOCK_CLIENT_OPTIONS); + const privateClient = client as unknown as { + _transport: { flush: ReturnType }; + }; + const pendingSpan = { + spanContext: () => ({ spanId: 'pending-span', traceFlags: TRACE_FLAG_SAMPLED }), + }; + + client.emit('spanStart', pendingSpan as any); + + const result = await client.flush(10); + + expect(result).toBe(false); + expect(privateClient._transport.flush).toHaveBeenCalledOnce(); + }); + + it('keeps all flush stages within one timeout', async () => { + vi.useFakeTimers(); + try { + vi.setSystemTime(0); + const transportFlush = vi.fn( + (timeout?: number) => new Promise(resolve => setTimeout(() => resolve(false), timeout)), + ); + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + flushLock: { ready: Promise.resolve(), finalize: () => new Promise(() => undefined) }, + transport: () => ({ send: vi.fn().mockResolvedValue({}), flush: transportFlush }), + }); + const privateClient = client as unknown as { _numProcessing: number }; + privateClient._numProcessing = 1; + client.emit('spanStart', { + spanContext: () => ({ spanId: 'pending-span', traceFlags: TRACE_FLAG_SAMPLED }), + } as any); + let settled = false; + + const flushPromise = client.flush(100).then(result => { + settled = true; + return result; + }); + + await vi.advanceTimersByTimeAsync(99); + expect(settled).toBe(false); + await vi.advanceTimersByTimeAsync(1); + expect(await flushPromise).toBe(false); + expect(transportFlush).toHaveBeenCalledOnce(); + expect(Date.now()).toBe(100); + } finally { + vi.useRealTimers(); + } }); }); From 6dad74dcc9e4d5d13001239d8adda21dde388732 Mon Sep 17 00:00:00 2001 From: Matthew Jones Date: Thu, 17 Sep 2026 16:47:27 -0600 Subject: [PATCH 3/6] test(cloudflare): Stabilize flush timeout coverage Use the Cloudflare test suite's established fake-timer cleanup and explicit timer advancement patterns for flush deadline assertions. Co-authored-by: OpenAI Codex --- packages/cloudflare/test/client.test.ts | 101 +++++++++++++++--------- 1 file changed, 62 insertions(+), 39 deletions(-) diff --git a/packages/cloudflare/test/client.test.ts b/packages/cloudflare/test/client.test.ts index 2e1e2e6d5734..cfed58bf18d5 100644 --- a/packages/cloudflare/test/client.test.ts +++ b/packages/cloudflare/test/client.test.ts @@ -1,4 +1,4 @@ -import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'; +import { beforeAll, beforeEach, describe, expect, it, onTestFinished, vi } from 'vitest'; import { setAsyncLocalStorageAsyncContextStrategy } from '@sentry/server-utils/no-diagnostic-channels'; import { CloudflareClient, type CloudflareClientOptions } from '../src/client'; import { makeFlushLock } from '../src/flush'; @@ -226,13 +226,20 @@ describe('CloudflareClient', () => { describe('flush()', () => { it('shares the timeout between client processing and transport flushing', async () => { + vi.useFakeTimers(); + onTestFinished(() => { + vi.useRealTimers(); + }); + vi.setSystemTime(0); const client = new CloudflareClient(MOCK_CLIENT_OPTIONS); const privateClient = client as unknown as { _transport: { flush: ReturnType }; }; - await client.flush(3000); + const flushPromise = client.flush(3000); + await vi.advanceTimersToNextTimerAsync(); + await flushPromise; expect(privateClient._transport.flush).toHaveBeenCalledWith(1500); }); @@ -273,6 +280,11 @@ describe('CloudflareClient', () => { }); it('includes the flush lock in the timeout', async () => { + vi.useFakeTimers(); + onTestFinished(() => { + vi.useRealTimers(); + }); + vi.setSystemTime(0); const finalize = vi.fn(() => new Promise(() => undefined)); const client = new CloudflareClient({ ...MOCK_CLIENT_OPTIONS, @@ -282,16 +294,23 @@ describe('CloudflareClient', () => { const privateClient = client as unknown as { _transport: { flush: ReturnType }; }; - const result = await Promise.race([ - client.flush(10), - new Promise<'did-not-settle'>(resolve => setTimeout(() => resolve('did-not-settle'), 30)), - ]); + const flushPromise = client.flush(30); + + await vi.advanceTimersByTimeAsync(9); + expect(privateClient._transport.flush).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + await vi.advanceTimersToNextTimerAsync(); - expect(result).toBe(false); + await expect(flushPromise).resolves.toBe(false); expect(privateClient._transport.flush).toHaveBeenCalledOnce(); }); it('still drains the transport when pending spans consume their wait budget', async () => { + vi.useFakeTimers(); + onTestFinished(() => { + vi.useRealTimers(); + }); + vi.setSystemTime(0); const client = new CloudflareClient(MOCK_CLIENT_OPTIONS); const privateClient = client as unknown as { _transport: { flush: ReturnType }; @@ -302,45 +321,49 @@ describe('CloudflareClient', () => { client.emit('spanStart', pendingSpan as any); - const result = await client.flush(10); + const flushPromise = client.flush(30); + + await vi.advanceTimersByTimeAsync(9); + expect(privateClient._transport.flush).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + await vi.advanceTimersToNextTimerAsync(); - expect(result).toBe(false); + await expect(flushPromise).resolves.toBe(false); expect(privateClient._transport.flush).toHaveBeenCalledOnce(); }); it('keeps all flush stages within one timeout', async () => { vi.useFakeTimers(); - try { - vi.setSystemTime(0); - const transportFlush = vi.fn( - (timeout?: number) => new Promise(resolve => setTimeout(() => resolve(false), timeout)), - ); - const client = new CloudflareClient({ - ...MOCK_CLIENT_OPTIONS, - flushLock: { ready: Promise.resolve(), finalize: () => new Promise(() => undefined) }, - transport: () => ({ send: vi.fn().mockResolvedValue({}), flush: transportFlush }), - }); - const privateClient = client as unknown as { _numProcessing: number }; - privateClient._numProcessing = 1; - client.emit('spanStart', { - spanContext: () => ({ spanId: 'pending-span', traceFlags: TRACE_FLAG_SAMPLED }), - } as any); - let settled = false; - - const flushPromise = client.flush(100).then(result => { - settled = true; - return result; - }); - - await vi.advanceTimersByTimeAsync(99); - expect(settled).toBe(false); - await vi.advanceTimersByTimeAsync(1); - expect(await flushPromise).toBe(false); - expect(transportFlush).toHaveBeenCalledOnce(); - expect(Date.now()).toBe(100); - } finally { + onTestFinished(() => { vi.useRealTimers(); - } + }); + vi.setSystemTime(0); + const transportFlush = vi.fn( + (timeout?: number) => new Promise(resolve => setTimeout(() => resolve(false), timeout)), + ); + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + flushLock: { ready: Promise.resolve(), finalize: () => new Promise(() => undefined) }, + transport: () => ({ send: vi.fn().mockResolvedValue({}), flush: transportFlush }), + }); + const privateClient = client as unknown as { _numProcessing: number }; + privateClient._numProcessing = 1; + client.emit('spanStart', { + spanContext: () => ({ spanId: 'pending-span', traceFlags: TRACE_FLAG_SAMPLED }), + } as any); + let settled = false; + + const flushPromise = client.flush(100).then(result => { + settled = true; + return result; + }); + + await vi.advanceTimersByTimeAsync(99); + expect(settled).toBe(false); + await vi.advanceTimersByTimeAsync(1); + expect(await flushPromise).toBe(false); + expect(transportFlush).toHaveBeenCalledOnce(); + expect(Date.now()).toBe(100); }); }); From 48f22a8ecb1abaa24487edc9cb3b3576ad968c1e Mon Sep 17 00:00:00 2001 From: JPeer264 Date: Fri, 18 Sep 2026 14:23:28 +0200 Subject: [PATCH 4/6] Revert to make the diff easier for me --- packages/cloudflare/src/client.ts | 73 +++--------- packages/cloudflare/src/transport.ts | 61 ++-------- packages/cloudflare/src/workflows.ts | 9 -- packages/cloudflare/test/client.test.ts | 107 +----------------- .../worker/instrumentEmail.test.ts | 2 +- .../worker/instrumentFetch.test.ts | 2 +- .../worker/instrumentQueue.test.ts | 2 +- .../worker/instrumentScheduled.test.ts | 2 +- .../worker/instrumentTail.test.ts | 2 +- packages/cloudflare/test/request.test.ts | 5 +- packages/cloudflare/test/workflow.test.ts | 29 +---- 11 files changed, 37 insertions(+), 257 deletions(-) diff --git a/packages/cloudflare/src/client.ts b/packages/cloudflare/src/client.ts index ceddb36516a0..a71445b03f92 100644 --- a/packages/cloudflare/src/client.ts +++ b/packages/cloudflare/src/client.ts @@ -15,22 +15,6 @@ import type { makeFlushLock } from './flush'; import type { CloudflareTransportOptions } from './transport'; import { getInvocationState, getInvocationWaitUntil } from './utils/invocationContext'; -async function waitForPromise(promise: PromiseLike, timeout: number): Promise { - let timer: ReturnType | undefined; - try { - return await Promise.race([ - Promise.resolve(promise).then(() => true), - new Promise(resolve => { - timer = setTimeout(() => resolve(false), timeout); - }), - ]); - } finally { - if (timer) { - clearTimeout(timer); - } - } -} - /** * The Sentry Cloudflare SDK Client. * @@ -131,49 +115,30 @@ 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 { - const deadline = timeout && timeout > 0 ? Date.now() + timeout : undefined; - const remainingTimeout = (): number | undefined => - deadline === undefined ? timeout : Math.max(0, deadline - Date.now()); - const hasPendingSpans = this._pendingSpans.size > 0 && this._spanCompletionPromise; - let remainingStages = 2 + (this._flushLock ? 1 : 0) + (hasPendingSpans ? 1 : 0); - let preDrainWorkCompleted = true; - const timeoutForNextStage = (): number | undefined => { - const remaining = remainingTimeout(); - return remaining === undefined ? undefined : Math.floor(remaining / remainingStages); - }; - // 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) { - const lockTimeout = timeoutForNextStage(); - if (lockTimeout && lockTimeout > 0) { - if (!(await waitForPromise(this._flushLock.finalize(), lockTimeout))) { - preDrainWorkCompleted = false; - } - } else if (deadline !== undefined) { - preDrainWorkCompleted = false; - } else { - await this._flushLock.finalize(); - } - remainingStages--; + await this._flushLock.finalize(); } - if (hasPendingSpans) { + if (this._pendingSpans.size > 0 && this._spanCompletionPromise) { DEBUG_BUILD && debug.log('[CloudflareClient] Waiting for', this._pendingSpans.size, 'pending spans to complete...'); - const spanTimeout = timeoutForNextStage() ?? 5000; - if (spanTimeout > 0) { - if (!(await waitForPromise(hasPendingSpans, spanTimeout))) { - DEBUG_BUILD && - debug.log('[CloudflareClient] Span completion timeout after', spanTimeout, 'ms, flushing anyway'); - preDrainWorkCompleted = false; - } - } else { - preDrainWorkCompleted = false; - } - remainingStages--; + const timeoutMs = timeout ?? 5000; + const spanCompletionRace = Promise.race([ + this._spanCompletionPromise, + new Promise(resolve => + setTimeout(() => { + DEBUG_BUILD && + debug.log('[CloudflareClient] Span completion timeout after', timeoutMs, 'ms, flushing anyway'); + resolve(undefined); + }, timeoutMs), + ), + ]); + + await spanCompletionRace; } // Envelopes created while this flush drains (log/metric/span buffers turning into @@ -181,13 +146,7 @@ export class CloudflareClient extends ServerRuntimeClient { // of them would also start an eager drain and a `waitUntil` registration. this._inBoundaryFlush = true; try { - // BaseClient.flush uses its timeout once for client processing and once for - // transport draining, so give it one stage's share for each internal wait. - const baseFlushTimeout = timeoutForNextStage(); - if (deadline !== undefined && (!baseFlushTimeout || baseFlushTimeout <= 0)) { - return false; - } - return (await super.flush(baseFlushTimeout)) && preDrainWorkCompleted; + return await super.flush(timeout); } finally { this._inBoundaryFlush = false; } diff --git a/packages/cloudflare/src/transport.ts b/packages/cloudflare/src/transport.ts index c9e95b30fc14..25d9e05572b9 100644 --- a/packages/cloudflare/src/transport.ts +++ b/packages/cloudflare/src/transport.ts @@ -16,9 +16,6 @@ export interface CloudflareTransportOptions extends BaseTransportOptions { */ const DEFAULT_TRANSPORT_BUFFER_SIZE = 256; -type TaskProducer = () => PromiseLike; -type RunTask = (taskProducer: TaskProducer, signal: AbortSignal) => PromiseLike; - /** * This is a modified promise buffer that collects tasks until drain is called. * We need this in the edge runtime because edge function invocations may not share I/O objects, like fetch requests @@ -32,23 +29,20 @@ export class IsolatedPromiseBuffer { // If we ever remove it from the interface we should also remove it here. public $: Array>; - private _taskProducers: TaskProducer[]; + private _taskProducers: (() => PromiseLike)[]; private readonly _bufferSize: number; - private readonly _runTask: RunTask; - - public constructor(_bufferSize = DEFAULT_TRANSPORT_BUFFER_SIZE, _runTask: RunTask = taskProducer => taskProducer()) { + public constructor(_bufferSize = DEFAULT_TRANSPORT_BUFFER_SIZE) { this.$ = []; this._taskProducers = []; this._bufferSize = _bufferSize; - this._runTask = _runTask; } /** * @inheritdoc */ - public add(taskProducer: TaskProducer): PromiseLike { + public add(taskProducer: () => PromiseLike): PromiseLike { if (this._taskProducers.length >= this._bufferSize) { return Promise.reject(SENTRY_BUFFER_FULL_ERROR); } @@ -63,13 +57,10 @@ export class IsolatedPromiseBuffer { public drain(timeout?: number): PromiseLike { const oldTaskProducers = [...this._taskProducers]; this._taskProducers = []; - const drainController = new AbortController(); - const tasks = oldTaskProducers.map(taskProducer => this._runTask(taskProducer, drainController.signal)); return new Promise(resolve => { const timer = setTimeout(() => { if (timeout && timeout > 0) { - drainController.abort(); resolve(false); } }, timeout); @@ -77,8 +68,8 @@ export class IsolatedPromiseBuffer { // This cannot reject // eslint-disable-next-line @typescript-eslint/no-floating-promises Promise.all( - tasks.map(task => - task.then(null, () => { + oldTaskProducers.map(taskProducer => + taskProducer().then(null, () => { // catch all failed requests }), ), @@ -95,36 +86,15 @@ export class IsolatedPromiseBuffer { * Creates a Transport that uses the native fetch API to send events to Sentry. */ export function makeCloudflareTransport(options: CloudflareTransportOptions): Transport { - let activeDrainSignal: AbortSignal | undefined; - function makeRequest(request: TransportRequest): PromiseLike { - const controller = new AbortController(); - const callerSignal = options.fetchOptions?.signal; - const drainSignal = activeDrainSignal; - const abortFromCallerSignal = (): void => controller.abort(); - const abortFromDrainSignal = (): void => controller.abort(); - - if (callerSignal?.aborted) { - controller.abort(); - } else { - callerSignal?.addEventListener('abort', abortFromCallerSignal, { once: true }); - } - - if (drainSignal?.aborted) { - controller.abort(); - } else { - drainSignal?.addEventListener('abort', abortFromDrainSignal, { once: true }); - } - const requestOptions: RequestInit = { body: request.body as BodyInit, method: 'POST', headers: options.headers, ...options.fetchOptions, - signal: controller.signal, }; - const requestPromise = suppressTracing(() => { + return suppressTracing(() => { return (options.fetch ?? fetch)(options.url, requestOptions).then(async response => { // Consume the response body to satisfy Cloudflare Workers' fetch requirements. // The runtime requires all fetch response bodies to be read or explicitly canceled @@ -146,24 +116,7 @@ export function makeCloudflareTransport(options: CloudflareTransportOptions): Tr }; }); }); - - return Promise.resolve(requestPromise).finally(() => { - callerSignal?.removeEventListener('abort', abortFromCallerSignal); - drainSignal?.removeEventListener('abort', abortFromDrainSignal); - }); - } - - function runTaskWithinDrain( - taskProducer: TaskProducer, - signal: AbortSignal, - ): PromiseLike { - activeDrainSignal = signal; - try { - return taskProducer(); - } finally { - activeDrainSignal = undefined; - } } - return createTransport(options, makeRequest, new IsolatedPromiseBuffer(options.bufferSize, runTaskWithinDrain)); + return createTransport(options, makeRequest, new IsolatedPromiseBuffer(options.bufferSize)); } diff --git a/packages/cloudflare/src/workflows.ts b/packages/cloudflare/src/workflows.ts index c5ce89614992..c46470c54355 100644 --- a/packages/cloudflare/src/workflows.ts +++ b/packages/cloudflare/src/workflows.ts @@ -30,7 +30,6 @@ import { instrumentEnv } from './instrumentations/worker/instrumentEnv'; import { addCloudResourceContext } from './scope-utils'; import { init } from './sdk'; import { instrumentContext } from './utils/instrumentContext'; -import { getInvocationState } from './utils/invocationContext'; import type { DefaultEnv, ResolveEnv, StrictCloudflareOptions } from './types'; import { withInvocationIsolationScope } from './utils/invocationScope'; @@ -125,14 +124,6 @@ class WrappedWorkflowStep implements WorkflowStep { // run's isolation scope (and with it the invocation state that ties eager sends // to this invocation's `waitUntil`) has to be restored explicitly. return withIsolationScope(this._isolationScope, () => { - // Each Workflow step is its own RPC invocation with its own boundary flush. - // The isolation scope is shared across steps, so clear the previous step's - // flush point before capturing anything for this one. - const invocationState = getInvocationState(); - if (invocationState) { - invocationState.flushPointReached = false; - } - const stepResult = startSpan( { name, diff --git a/packages/cloudflare/test/client.test.ts b/packages/cloudflare/test/client.test.ts index cfed58bf18d5..09bff574e479 100644 --- a/packages/cloudflare/test/client.test.ts +++ b/packages/cloudflare/test/client.test.ts @@ -1,4 +1,4 @@ -import { beforeAll, beforeEach, describe, expect, it, onTestFinished, vi } from 'vitest'; +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'; @@ -225,23 +225,16 @@ describe('CloudflareClient', () => { }); describe('flush()', () => { - it('shares the timeout between client processing and transport flushing', async () => { - vi.useFakeTimers(); - onTestFinished(() => { - vi.useRealTimers(); - }); - vi.setSystemTime(0); + 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 }; }; - const flushPromise = client.flush(3000); - await vi.advanceTimersToNextTimerAsync(); - await flushPromise; + await client.flush(3000); - expect(privateClient._transport.flush).toHaveBeenCalledWith(1500); + expect(privateClient._transport.flush).toHaveBeenCalledWith(3000); }); it('resolves with the transport flush result', async () => { @@ -273,97 +266,7 @@ describe('CloudflareClient', () => { releaseLock(); await flushPromise; - expect(privateClient._transport.flush).toHaveBeenCalledWith(expect.any(Number)); - const transportTimeout = privateClient._transport.flush.mock.calls[0]?.[0]; - expect(transportTimeout).toBeGreaterThan(0); - expect(transportTimeout).toBeLessThanOrEqual(1000); - }); - - it('includes the flush lock in the timeout', async () => { - vi.useFakeTimers(); - onTestFinished(() => { - vi.useRealTimers(); - }); - vi.setSystemTime(0); - const finalize = vi.fn(() => new Promise(() => undefined)); - 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(30); - - await vi.advanceTimersByTimeAsync(9); - expect(privateClient._transport.flush).not.toHaveBeenCalled(); - await vi.advanceTimersByTimeAsync(1); - await vi.advanceTimersToNextTimerAsync(); - - await expect(flushPromise).resolves.toBe(false); - expect(privateClient._transport.flush).toHaveBeenCalledOnce(); - }); - - it('still drains the transport when pending spans consume their wait budget', async () => { - vi.useFakeTimers(); - onTestFinished(() => { - vi.useRealTimers(); - }); - vi.setSystemTime(0); - const client = new CloudflareClient(MOCK_CLIENT_OPTIONS); - const privateClient = client as unknown as { - _transport: { flush: ReturnType }; - }; - const pendingSpan = { - spanContext: () => ({ spanId: 'pending-span', traceFlags: TRACE_FLAG_SAMPLED }), - }; - - client.emit('spanStart', pendingSpan as any); - - const flushPromise = client.flush(30); - - await vi.advanceTimersByTimeAsync(9); - expect(privateClient._transport.flush).not.toHaveBeenCalled(); - await vi.advanceTimersByTimeAsync(1); - await vi.advanceTimersToNextTimerAsync(); - - await expect(flushPromise).resolves.toBe(false); - expect(privateClient._transport.flush).toHaveBeenCalledOnce(); - }); - - it('keeps all flush stages within one timeout', async () => { - vi.useFakeTimers(); - onTestFinished(() => { - vi.useRealTimers(); - }); - vi.setSystemTime(0); - const transportFlush = vi.fn( - (timeout?: number) => new Promise(resolve => setTimeout(() => resolve(false), timeout)), - ); - const client = new CloudflareClient({ - ...MOCK_CLIENT_OPTIONS, - flushLock: { ready: Promise.resolve(), finalize: () => new Promise(() => undefined) }, - transport: () => ({ send: vi.fn().mockResolvedValue({}), flush: transportFlush }), - }); - const privateClient = client as unknown as { _numProcessing: number }; - privateClient._numProcessing = 1; - client.emit('spanStart', { - spanContext: () => ({ spanId: 'pending-span', traceFlags: TRACE_FLAG_SAMPLED }), - } as any); - let settled = false; - - const flushPromise = client.flush(100).then(result => { - settled = true; - return result; - }); - - await vi.advanceTimersByTimeAsync(99); - expect(settled).toBe(false); - await vi.advanceTimersByTimeAsync(1); - expect(await flushPromise).toBe(false); - expect(transportFlush).toHaveBeenCalledOnce(); - expect(Date.now()).toBe(100); + expect(privateClient._transport.flush).toHaveBeenCalledWith(1000); }); }); diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentEmail.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentEmail.test.ts index a12b29f6a3bb..2c15f5a39966 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentEmail.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentEmail.test.ts @@ -302,7 +302,7 @@ describe('instrumentEmail', () => { } as unknown as ExecutionContext); expect(flush).not.toBeCalled(); expect(waitUntil).toBeCalled(); - await vi.advanceTimersToNextTimerAsync(); + vi.advanceTimersToNextTimer().runAllTimers(); await Promise.all(waits); expect(flush).toHaveBeenCalledOnce(); }); diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentFetch.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentFetch.test.ts index d57687b8b2d8..1a0e94093444 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentFetch.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentFetch.test.ts @@ -172,7 +172,7 @@ describe('instrumentFetch', () => { .then(response => response.text()); expect(flush).not.toBeCalled(); expect(waitUntil).toBeCalled(); - await vi.advanceTimersToNextTimerAsync(); + vi.advanceTimersToNextTimer().runAllTimers(); await Promise.all(waits); expect(flush).toHaveBeenCalledOnce(); }); diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts index 623d8825cea0..afe51790adfd 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentQueue.test.ts @@ -318,7 +318,7 @@ describe('instrumentQueue', () => { } as unknown as ExecutionContext); expect(flush).not.toBeCalled(); expect(waitUntil).toBeCalled(); - await vi.advanceTimersToNextTimerAsync(); + vi.advanceTimersToNextTimer().runAllTimers(); await Promise.all(waits); expect(flush).toHaveBeenCalledOnce(); }); diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentScheduled.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentScheduled.test.ts index 814d814583f4..46090684daea 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentScheduled.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentScheduled.test.ts @@ -323,7 +323,7 @@ describe('instrumentScheduled', () => { } as unknown as ExecutionContext); expect(flush).not.toBeCalled(); expect(waitUntil).toBeCalled(); - await vi.advanceTimersToNextTimerAsync(); + vi.advanceTimersToNextTimer().runAllTimers(); await Promise.all(waits); expect(flush).toHaveBeenCalledOnce(); }); diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentTail.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentTail.test.ts index 7d16d2578362..014916e6f158 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentTail.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentTail.test.ts @@ -270,7 +270,7 @@ describe('instrumentTail', () => { } as unknown as ExecutionContext); expect(flush).not.toBeCalled(); expect(waitUntil).toBeCalled(); - await vi.advanceTimersToNextTimerAsync(); + vi.advanceTimersToNextTimer().runAllTimers(); await Promise.all(waits); expect(flush).toHaveBeenCalledOnce(); }); diff --git a/packages/cloudflare/test/request.test.ts b/packages/cloudflare/test/request.test.ts index 885a48f4cfb0..052147f84bc4 100644 --- a/packages/cloudflare/test/request.test.ts +++ b/packages/cloudflare/test/request.test.ts @@ -150,7 +150,7 @@ describe('withSentry', () => { return new Response('test'); }).then(response => response.text()); expect(waitUntil).toBeCalled(); - await vi.advanceTimersToNextTimerAsync(); + vi.advanceTimersToNextTimer().runAllTimers(); await Promise.all(waits); const after = flushSpy.mock.calls.length; @@ -1051,7 +1051,6 @@ describe('Durable Object (DO) context', () => { // Teardown is registered via waitUntil on error too expect(waitUntilSpy).toHaveBeenCalled(); - await Promise.all(waitUntilSpy.mock.calls.map(([promise]) => promise)); // And flush runs as part of that teardown expect(flushSpy).toHaveBeenCalled(); @@ -1073,7 +1072,6 @@ describe('Durable Object (DO) context', () => { ); expect(waitUntilSpy).toHaveBeenCalled(); - await Promise.all(waitUntilSpy.mock.calls.map(([promise]) => promise)); expect(flushSpy).toHaveBeenCalled(); flushSpy.mockRestore(); @@ -1094,7 +1092,6 @@ describe('Durable Object (DO) context', () => { ); expect(waitUntilSpy).toHaveBeenCalled(); - await Promise.all(waitUntilSpy.mock.calls.map(([promise]) => promise)); expect(flushSpy).toHaveBeenCalled(); flushSpy.mockRestore(); diff --git a/packages/cloudflare/test/workflow.test.ts b/packages/cloudflare/test/workflow.test.ts index ef9ebbd94941..f2ddfae9d8f0 100644 --- a/packages/cloudflare/test/workflow.test.ts +++ b/packages/cloudflare/test/workflow.test.ts @@ -458,9 +458,9 @@ describe.skipIf(NODE_MAJOR_VERSION < 20)('workflows', () => { expect(mockStep.do).toHaveBeenCalledTimes(1); expect(mockStep.do).toHaveBeenCalledWith('sometimes error step', expect.any(Function)); - // One boundary flush per attempt (failed and retried) and one at the end of the run. - // The retry starts before its flush point, so its envelope does not add an eager flush. - expect(mockContext.waitUntil).toHaveBeenCalledTimes(3); + // One flush per attempt (failed and retried, past the span end) and one at end of + // run, plus one eager registration for the envelope of the error captured mid-run + expect(mockContext.waitUntil).toHaveBeenCalledTimes(4); expect(mockContext.waitUntil).toHaveBeenCalledWith(expect.any(Promise)); // No error event (not final attempt), only failed transaction + successful retry transaction expect(mockTransport.send).toHaveBeenCalledTimes(2); @@ -784,27 +784,4 @@ describe.skipIf(NODE_MAJOR_VERSION < 20)('workflows', () => { expect(tagInsideStep).toBe('marker'); expect(hasInvocationStateInsideStep).toBe(true); }); - - test('each workflow step starts before the invocation flush point', async () => { - const flushPointsAtStepStart: Array = []; - - class MultipleStepWorkflow { - constructor(_ctx: ExecutionContext, _env: unknown) {} - - async run(_event: Readonly>, step: WorkflowStep): Promise { - await step.do('first step', async () => { - flushPointsAtStepStart.push(getInvocationState()?.flushPointReached); - }); - await step.do('second step', async () => { - flushPointsAtStepStart.push(getInvocationState()?.flushPointReached); - }); - } - } - - const TestWorkflowInstrumented = instrumentWorkflowWithSentry(getSentryOptions, MultipleStepWorkflow as any); - const workflow = new TestWorkflowInstrumented(mockContext, {}) as MultipleStepWorkflow; - await workflow.run({ payload: {}, timestamp: new Date(), instanceId: INSTANCE_ID }, mockStep); - - expect(flushPointsAtStepStart).toEqual([false, false]); - }); }); From fb6382019093a4dde7b15e1ea9b405c4f19b7ad7 Mon Sep 17 00:00:00 2001 From: JPeer264 Date: Fri, 18 Sep 2026 18:48:21 +0200 Subject: [PATCH 5/6] fix(cloudflare): Abort pending sends when a flush times out --- .../suites/flush-timeout/index.ts | 52 +++++++++ .../suites/flush-timeout/instrument.server.ts | 25 +++++ .../suites/flush-timeout/lastSend.ts | 2 + .../suites/flush-timeout/test.ts | 76 +++++++++++++ .../suites/flush-timeout/vite.config.mts | 7 ++ .../suites/flush-timeout/wrangler.jsonc | 14 +++ packages/cloudflare/src/client.ts | 11 +- packages/cloudflare/src/transport.ts | 32 +++++- packages/cloudflare/test/client.test.ts | 25 +++++ packages/cloudflare/test/request.test.ts | 3 + packages/cloudflare/test/transport.test.ts | 106 ++++++++++++------ 11 files changed, 312 insertions(+), 41 deletions(-) create mode 100644 dev-packages/cloudflare-integration-tests/suites/flush-timeout/index.ts create mode 100644 dev-packages/cloudflare-integration-tests/suites/flush-timeout/instrument.server.ts create mode 100644 dev-packages/cloudflare-integration-tests/suites/flush-timeout/lastSend.ts create mode 100644 dev-packages/cloudflare-integration-tests/suites/flush-timeout/test.ts create mode 100644 dev-packages/cloudflare-integration-tests/suites/flush-timeout/vite.config.mts create mode 100644 dev-packages/cloudflare-integration-tests/suites/flush-timeout/wrangler.jsonc diff --git a/dev-packages/cloudflare-integration-tests/suites/flush-timeout/index.ts b/dev-packages/cloudflare-integration-tests/suites/flush-timeout/index.ts new file mode 100644 index 000000000000..4e00dbb4109c --- /dev/null +++ b/dev-packages/cloudflare-integration-tests/suites/flush-timeout/index.ts @@ -0,0 +1,52 @@ +import * as Sentry from '@sentry/cloudflare'; +import { WorkflowEntrypoint } from 'cloudflare:workers'; +import type { WorkflowEvent, WorkflowStep } from 'cloudflare:workers'; +import { lastSend } from './lastSend'; + +interface Env { + SERVER_URL: string; + ISSUE_WORKFLOW: Workflow; +} + +// The Workflow from https://github.com/getsentry/sentry-javascript/issues/24482. After its steps it flushes +// with the same timeout the SDK uses after each step and reports to SERVER_URL how the pending send ended. +export class IssueWorkflow extends WorkflowEntrypoint { + async run(_event: WorkflowEvent, step: WorkflowStep): Promise { + for (let index = 0; index < 100; index++) { + await step.do(`step-${index}`, async () => index); + } + + // Locally the step spans are not sent before `run()` returns, so flush here with the timeout the SDK uses + // after each step, while the Workflow can still observe whether the pending send gets aborted. + lastSend.aborted = false; + const flushed = await Sentry.flush(2000); + const send = lastSend.aborted ? 'aborted' : 'not aborted'; + await fetch(`${this.env.SERVER_URL}/result`, { method: 'POST', body: JSON.stringify({ flushed, send }) }); + } +} + +export default { + async fetch(request, env, ctx) { + const url = new URL(request.url); + + if (url.pathname === '/workflow/trigger') { + const instance = await env.ISSUE_WORKFLOW.create(); + return Response.json({ id: instance.id }); + } + + // The flush runs inside the invocation, so the send is still pending when its drain times out. + if (url.pathname === '/flush-with-timeout') { + Sentry.captureException(new Error('Captured on /flush-with-timeout')); + lastSend.aborted = false; + const flushed = await Sentry.flush(500); + return Response.json({ flushed, send: lastSend.aborted ? 'aborted' : 'not aborted' }); + } + + if (url.pathname === '/pending-wait-until') { + ctx.waitUntil(new Promise(resolve => setTimeout(resolve, 120_000))); + Sentry.captureException(new Error('Captured on /pending-wait-until')); + } + + return new Response('ok'); + }, +} satisfies ExportedHandler; diff --git a/dev-packages/cloudflare-integration-tests/suites/flush-timeout/instrument.server.ts b/dev-packages/cloudflare-integration-tests/suites/flush-timeout/instrument.server.ts new file mode 100644 index 000000000000..13a9e52db068 --- /dev/null +++ b/dev-packages/cloudflare-integration-tests/suites/flush-timeout/instrument.server.ts @@ -0,0 +1,25 @@ +import { defineCloudflareOptions } from '@sentry/cloudflare'; +import { lastSend } from './lastSend'; + +interface Env { + SENTRY_DSN: string; + SERVER_URL: string; + // "true" sends envelopes to SERVER_URL, a server that never answers + SLOW_INGEST?: string; + // "false" creates one client per invocation, which waits for the invocation's flush lock + CACHE_CLIENT?: string; + // "true" samples every trace, so the Workflow steps create spans to send + TRACING?: string; +} + +export default defineCloudflareOptions((env: Env) => ({ + dsn: env.SLOW_INGEST === 'true' ? `${env.SERVER_URL.replace('://', '://public@')}/1337` : env.SENTRY_DSN, + cacheClient: env.CACHE_CLIENT !== 'false', + tracesSampleRate: env.TRACING === 'true' ? 1 : undefined, + transportOptions: { + fetch: (input, init) => { + init?.signal?.addEventListener('abort', () => (lastSend.aborted = true)); + return fetch(input, init); + }, + }, +})); diff --git a/dev-packages/cloudflare-integration-tests/suites/flush-timeout/lastSend.ts b/dev-packages/cloudflare-integration-tests/suites/flush-timeout/lastSend.ts new file mode 100644 index 000000000000..d9a7b995a130 --- /dev/null +++ b/dev-packages/cloudflare-integration-tests/suites/flush-timeout/lastSend.ts @@ -0,0 +1,2 @@ +// Whether the transport aborted an envelope fetch since `aborted` was last reset. +export const lastSend: { aborted: boolean } = { aborted: false }; diff --git a/dev-packages/cloudflare-integration-tests/suites/flush-timeout/test.ts b/dev-packages/cloudflare-integration-tests/suites/flush-timeout/test.ts new file mode 100644 index 000000000000..4eee8b06a45c --- /dev/null +++ b/dev-packages/cloudflare-integration-tests/suites/flush-timeout/test.ts @@ -0,0 +1,76 @@ +import type { Envelope, Event } from '@sentry/core'; +import { createServer } from 'node:http'; +import type { AddressInfo } from 'node:net'; +import { expect, it, onTestFinished } from 'vitest'; +import { createRunner } from '../../runner'; + +// Starts an ingest server that never answers envelope requests, so every send stays pending. The Workflow +// posts its result to `/result`, which resolves the returned promise with the posted body. +async function startSilentIngest(): Promise<{ url: string; result: Promise }> { + let resolveResult!: (body: unknown) => void; + const result = new Promise(resolve => (resolveResult = resolve)); + + const server = createServer((req, res) => { + if (req.url !== '/result') { + return; + } + let body = ''; + req.on('data', chunk => (body += chunk)); + req.on('end', () => { + res.end(); + resolveResult(JSON.parse(body)); + }); + }); + await new Promise(resolve => server.listen(0, resolve)); + onTestFinished(() => { + server.closeAllConnections(); + server.close(); + }); + + return { url: `http://localhost:${(server.address() as AddressInfo).port}`, result }; +} + +it.for([true, false])( + 'cacheClient: %s - aborts a send that is still pending when the flush times out', + async (cacheClient, { signal }) => { + const ingest = await startSilentIngest(); + + const runner = createRunner(__dirname) + .withServerUrl(ingest.url) + .withWranglerArgs('--var', 'SLOW_INGEST:true', '--var', `CACHE_CLIENT:${cacheClient}`) + .start(signal); + + const result = await runner.makeRequest('get', '/flush-with-timeout'); + expect(result).toEqual({ flushed: false, send: 'aborted' }); + }, +); + +// A cached client sends the step spans in eager drains the Workflow cannot wait for, so this runs with one +// client per invocation. The transport abort itself is covered for both modes by the test above. +it('cacheClient: false - the Workflow from #24482 aborts its pending send when the flush times out', async ({ + signal, +}) => { + const ingest = await startSilentIngest(); + + const runner = createRunner(__dirname) + .withServerUrl(ingest.url) + .withWranglerArgs('--var', 'SLOW_INGEST:true', '--var', 'CACHE_CLIENT:false', '--var', 'TRACING:true') + .start(signal); + + await runner.makeRequest('get', '/workflow/trigger'); + expect(await ingest.result).toEqual({ flushed: false, send: 'aborted' }); +}); + +it('cacheClient: false - delivers events while a user waitUntil task is still running', async ({ signal }) => { + const runner = createRunner(__dirname) + .withWranglerArgs('--var', 'CACHE_CLIENT:false') + .expect((envelope: Envelope) => { + const event = envelope[1]?.[0]?.[1] as Event; + expect(event.exception?.values?.[0]?.value).toBe('Captured on /pending-wait-until'); + }) + .unordered() + .start(signal); + + await runner.makeRequest('get', '/pending-wait-until'); + await runner.completed(); +}); diff --git a/dev-packages/cloudflare-integration-tests/suites/flush-timeout/vite.config.mts b/dev-packages/cloudflare-integration-tests/suites/flush-timeout/vite.config.mts new file mode 100644 index 000000000000..005f4448f6cb --- /dev/null +++ b/dev-packages/cloudflare-integration-tests/suites/flush-timeout/vite.config.mts @@ -0,0 +1,7 @@ +import { cloudflare } from '@cloudflare/vite-plugin'; +import { sentryCloudflareVitePlugin } from '@sentry/cloudflare/vite'; +import { defineConfig } from 'vite'; + +export default defineConfig({ + plugins: [cloudflare(), sentryCloudflareVitePlugin()], +}); diff --git a/dev-packages/cloudflare-integration-tests/suites/flush-timeout/wrangler.jsonc b/dev-packages/cloudflare-integration-tests/suites/flush-timeout/wrangler.jsonc new file mode 100644 index 000000000000..a74a1ff6059c --- /dev/null +++ b/dev-packages/cloudflare-integration-tests/suites/flush-timeout/wrangler.jsonc @@ -0,0 +1,14 @@ +{ + "$schema": "../../node_modules/wrangler/config-schema.json", + "name": "cloudflare-flush-timeout", + "main": "index.ts", + "compatibility_date": "2025-06-17", + "compatibility_flags": ["nodejs_compat"], + "workflows": [ + { + "name": "issue-workflow", + "binding": "ISSUE_WORKFLOW", + "class_name": "IssueWorkflow", + }, + ], +} diff --git a/packages/cloudflare/src/client.ts b/packages/cloudflare/src/client.ts index a71445b03f92..f99163328fb1 100644 --- a/packages/cloudflare/src/client.ts +++ b/packages/cloudflare/src/client.ts @@ -118,8 +118,17 @@ export class CloudflareClient extends ServerRuntimeClient { // 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. + // + // Only per-invocation clients (`cacheClient: false`) have a flush lock; remove this with them in v12. + // The wait is bounded by `timeout` because a user `waitUntil` task that outlives the invocation + // would otherwise keep the flush from draining until the runtime cancels it. if (this._flushLock) { - await this._flushLock.finalize(); + let timer: ReturnType | undefined; + await Promise.race([ + this._flushLock.finalize(), + ...(timeout ? [new Promise(resolve => (timer = setTimeout(resolve, timeout)))] : []), + ]); + clearTimeout(timer); } if (this._pendingSpans.size > 0 && this._spanCompletionPromise) { diff --git a/packages/cloudflare/src/transport.ts b/packages/cloudflare/src/transport.ts index 25d9e05572b9..b7f5a3181680 100644 --- a/packages/cloudflare/src/transport.ts +++ b/packages/cloudflare/src/transport.ts @@ -29,6 +29,12 @@ export class IsolatedPromiseBuffer { // If we ever remove it from the interface we should also remove it here. public $: Array>; + /** + * Abort signal of the drain that is starting its requests. It is set only while `drain()` runs the task + * producers, so a request reads the signal of the drain that sends it. + */ + public drainSignal: AbortSignal | undefined; + private _taskProducers: (() => PromiseLike)[]; private readonly _bufferSize: number; @@ -58,9 +64,21 @@ export class IsolatedPromiseBuffer { const oldTaskProducers = [...this._taskProducers]; this._taskProducers = []; + const drainController = new AbortController(); + this.drainSignal = drainController.signal; + let tasks: PromiseLike[]; + try { + tasks = oldTaskProducers.map(taskProducer => taskProducer()); + } finally { + this.drainSignal = undefined; + } + return new Promise(resolve => { const timer = setTimeout(() => { if (timeout && timeout > 0) { + // Requests still pending when the drain times out are aborted. Otherwise Cloudflare keeps them + // until it cancels the invocation's `waitUntil` work and logs a warning. + drainController.abort(); resolve(false); } }, timeout); @@ -68,8 +86,8 @@ export class IsolatedPromiseBuffer { // This cannot reject // eslint-disable-next-line @typescript-eslint/no-floating-promises Promise.all( - oldTaskProducers.map(taskProducer => - taskProducer().then(null, () => { + tasks.map(task => + task.then(null, () => { // catch all failed requests }), ), @@ -86,12 +104,20 @@ export class IsolatedPromiseBuffer { * Creates a Transport that uses the native fetch API to send events to Sentry. */ export function makeCloudflareTransport(options: CloudflareTransportOptions): Transport { + const buffer = new IsolatedPromiseBuffer(options.bufferSize); + function makeRequest(request: TransportRequest): PromiseLike { + const drainSignal = buffer.drainSignal; + const callerSignal = options.fetchOptions?.signal ?? undefined; + const signal = + drainSignal && callerSignal ? AbortSignal.any([drainSignal, callerSignal]) : (drainSignal ?? callerSignal); + const requestOptions: RequestInit = { body: request.body as BodyInit, method: 'POST', headers: options.headers, ...options.fetchOptions, + ...(signal ? { signal } : {}), }; return suppressTracing(() => { @@ -118,5 +144,5 @@ export function makeCloudflareTransport(options: CloudflareTransportOptions): Tr }); } - return createTransport(options, makeRequest, new IsolatedPromiseBuffer(options.bufferSize)); + return createTransport(options, makeRequest, buffer); } diff --git a/packages/cloudflare/test/client.test.ts b/packages/cloudflare/test/client.test.ts index 09bff574e479..2d7828be238d 100644 --- a/packages/cloudflare/test/client.test.ts +++ b/packages/cloudflare/test/client.test.ts @@ -268,6 +268,31 @@ describe('CloudflareClient', () => { await flushPromise; expect(privateClient._transport.flush).toHaveBeenCalledWith(1000); }); + + it('drains the transport when the flush lock does not settle within the timeout', async () => { + vi.useFakeTimers(); + try { + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + flushLock: { ready: Promise.resolve(), finalize: () => new Promise(() => undefined) }, + }); + + const privateClient = client as unknown as { + _transport: { flush: ReturnType }; + }; + + void client.flush(1000); + + await vi.advanceTimersByTimeAsync(999); + expect(privateClient._transport.flush).not.toHaveBeenCalled(); + + // The lock wait ends at 1000 ms; the client processing check after it also runs on timers. + await vi.advanceTimersByTimeAsync(100); + expect(privateClient._transport.flush).toHaveBeenCalledWith(1000); + } finally { + vi.useRealTimers(); + } + }); }); describe('span lifecycle tracking', () => { diff --git a/packages/cloudflare/test/request.test.ts b/packages/cloudflare/test/request.test.ts index 052147f84bc4..baf9282f9e15 100644 --- a/packages/cloudflare/test/request.test.ts +++ b/packages/cloudflare/test/request.test.ts @@ -1051,6 +1051,7 @@ describe('Durable Object (DO) context', () => { // Teardown is registered via waitUntil on error too expect(waitUntilSpy).toHaveBeenCalled(); + await Promise.all(waitUntilSpy.mock.calls.map(([promise]) => promise)); // And flush runs as part of that teardown expect(flushSpy).toHaveBeenCalled(); @@ -1072,6 +1073,7 @@ describe('Durable Object (DO) context', () => { ); expect(waitUntilSpy).toHaveBeenCalled(); + await Promise.all(waitUntilSpy.mock.calls.map(([promise]) => promise)); expect(flushSpy).toHaveBeenCalled(); flushSpy.mockRestore(); @@ -1092,6 +1094,7 @@ describe('Durable Object (DO) context', () => { ); expect(waitUntilSpy).toHaveBeenCalled(); + await Promise.all(waitUntilSpy.mock.calls.map(([promise]) => promise)); expect(flushSpy).toHaveBeenCalled(); flushSpy.mockRestore(); diff --git a/packages/cloudflare/test/transport.test.ts b/packages/cloudflare/test/transport.test.ts index 3044c1dffbef..1750176f5591 100644 --- a/packages/cloudflare/test/transport.test.ts +++ b/packages/cloudflare/test/transport.test.ts @@ -252,21 +252,53 @@ describe('IsolatedPromiseBuffer', () => { expect(customFetch).toHaveBeenCalledTimes(1); }); - it('aborts fetch requests when their drain times out', async () => { - let signal: AbortSignal | undefined; - const customFetch = vi.fn( - (_input: RequestInfo | URL, init?: RequestInit): Promise => - new Promise((_resolve, reject) => { - signal = init?.signal ?? undefined; - signal?.addEventListener('abort', () => reject(new DOMException('Aborted', 'AbortError')), { once: true }); - }), - ); - const transport = makeCloudflareTransport({ ...DEFAULT_EDGE_TRANSPORT_OPTIONS, fetch: customFetch }); + it('aborts a request that is still pending when its drain times out', async () => { + vi.useFakeTimers(); + try { + let signal: AbortSignal | undefined; + const customFetch = vi.fn( + (_input: RequestInfo | URL, init?: RequestInit): Promise => + new Promise((_resolve, reject) => { + signal = init?.signal ?? undefined; + signal?.addEventListener('abort', () => reject(new DOMException('Aborted', 'AbortError')), { once: true }); + }), + ); + const transport = makeCloudflareTransport({ ...DEFAULT_EDGE_TRANSPORT_OPTIONS, fetch: customFetch }); - await transport.send(ERROR_ENVELOPE); - await expect(transport.flush(1)).resolves.toBe(false); + await transport.send(ERROR_ENVELOPE); + const flush = transport.flush(1000); - expect(signal?.aborted).toBe(true); + await vi.advanceTimersByTimeAsync(999); + expect(signal?.aborted).toBe(false); + + await vi.advanceTimersByTimeAsync(1); + await expect(flush).resolves.toBe(false); + expect(signal?.aborted).toBe(true); + } finally { + vi.useRealTimers(); + } + }); + + it('does not abort requests of a drain without a timeout', async () => { + vi.useFakeTimers(); + try { + let signal: AbortSignal | undefined; + const customFetch = vi.fn( + (_input: RequestInfo | URL, init?: RequestInit): Promise => + new Promise(() => { + signal = init?.signal ?? undefined; + }), + ); + const transport = makeCloudflareTransport({ ...DEFAULT_EDGE_TRANSPORT_OPTIONS, fetch: customFetch }); + + await transport.send(ERROR_ENVELOPE); + void transport.flush(); + + await vi.advanceTimersByTimeAsync(60_000); + expect(signal?.aborted).toBe(false); + } finally { + vi.useRealTimers(); + } }); it('preserves a caller-provided abort signal', async () => { @@ -294,32 +326,32 @@ describe('IsolatedPromiseBuffer', () => { }); it('does not abort requests belonging to another drain', async () => { - const signals: AbortSignal[] = []; - const resolveRequests: ((response: Response) => void)[] = []; - const customFetch = vi.fn( - (_input: RequestInfo | URL, init?: RequestInit): Promise => - new Promise((resolve, reject) => { - const signal = init?.signal as AbortSignal; - signals.push(signal); - resolveRequests.push(resolve); - signal.addEventListener('abort', () => reject(new DOMException('Aborted', 'AbortError')), { once: true }); - }), - ); - const transport = makeCloudflareTransport({ ...DEFAULT_EDGE_TRANSPORT_OPTIONS, fetch: customFetch }); + vi.useFakeTimers(); + try { + const signals: AbortSignal[] = []; + const customFetch = vi.fn( + (_input: RequestInfo | URL, init?: RequestInit): Promise => + new Promise((_resolve, reject) => { + const signal = init?.signal as AbortSignal; + signals.push(signal); + signal.addEventListener('abort', () => reject(new DOMException('Aborted', 'AbortError')), { once: true }); + }), + ); + const transport = makeCloudflareTransport({ ...DEFAULT_EDGE_TRANSPORT_OPTIONS, fetch: customFetch }); - await transport.send(ERROR_ENVELOPE); - const firstFlush = transport.flush(1000); - await transport.send(ERROR_ENVELOPE); - await expect(transport.flush(1)).resolves.toBe(false); + await transport.send(ERROR_ENVELOPE); + void transport.flush(2000); + await transport.send(ERROR_ENVELOPE); + void transport.flush(1000); - expect(signals[0]?.aborted).toBe(false); - expect(signals[1]?.aborted).toBe(true); + await vi.advanceTimersByTimeAsync(1000); + expect(signals[0]?.aborted).toBe(false); + expect(signals[1]?.aborted).toBe(true); - resolveRequests[0]?.({ - headers: new Headers(), - status: 200, - text: () => Promise.resolve('OK'), - } as unknown as Response); - await expect(firstFlush).resolves.toBe(true); + await vi.advanceTimersByTimeAsync(1000); + expect(signals[0]?.aborted).toBe(true); + } finally { + vi.useRealTimers(); + } }); }); From f0fc6587af37c107fb61aa00427f9e8ab890542b Mon Sep 17 00:00:00 2001 From: JPeer264 Date: Fri, 18 Sep 2026 19:01:01 +0200 Subject: [PATCH 6/6] fixup! fix(cloudflare): Abort pending sends when a flush times out --- packages/cloudflare/src/client.ts | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/packages/cloudflare/src/client.ts b/packages/cloudflare/src/client.ts index f99163328fb1..1a968cfe66b2 100644 --- a/packages/cloudflare/src/client.ts +++ b/packages/cloudflare/src/client.ts @@ -107,11 +107,13 @@ export class CloudflareClient extends ServerRuntimeClient { /** * Flushes pending operations and ensures all data is processed. - * If a timeout is provided, the operation will be completed within the specified time limit. * - * It will wait for all pending spans to complete before flushing. + * Each phase waits at most `timeout`: the flush lock of a per-invocation client, pending spans, event + * processing and the transport drain. So a flush can take a small multiple of `timeout`, which stays well + * below Cloudflare's 30 second `waitUntil` limit for the timeouts the SDK uses. Sends still pending when + * the drain times out are aborted. * - * @param {number} [timeout] - Optional timeout in milliseconds to force the completion of the flush operation. + * @param {number} [timeout] - Maximum time in milliseconds for each phase of the flush. * @return {Promise} A promise that resolves to a boolean indicating whether the flush operation was successful. */ public async flush(timeout?: number): Promise {