fix(cloudflare): Enforce flush timeout across workflow lifecycle (#24482)

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 <codex@openai.com>

Author: matthewbjones

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<unknown>, timeout: number): Promise<boolean> {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  try {
+    return await Promise.race([
+      Promise.resolve(promise).then(() => true),
+      new Promise<false>(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<boolean>} A promise that resolves to a boolean indicating whether the flush operation was successful.
    */
   public async flush(timeout?: number): Promise<boolean> {
+    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;
     }

packages/cloudflare/src/transport.ts

@@ -16,6 +16,9 @@ export interface CloudflareTransportOptions extends BaseTransportOptions {
  */
 const DEFAULT_TRANSPORT_BUFFER_SIZE = 256;
 
+type TaskProducer = () => PromiseLike<TransportMakeRequestResponse>;
+type RunTask = (taskProducer: TaskProducer, signal: AbortSignal) => PromiseLike<TransportMakeRequestResponse>;
+
 /**
  * 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<PromiseLike<TransportMakeRequestResponse>>;
 
-  private _taskProducers: (() => PromiseLike<TransportMakeRequestResponse>)[];
+  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<TransportMakeRequestResponse>): PromiseLike<TransportMakeRequestResponse> {
+  public add(taskProducer: TaskProducer): PromiseLike<TransportMakeRequestResponse> {
     if (this._taskProducers.length >= this._bufferSize) {
       return Promise.reject(SENTRY_BUFFER_FULL_ERROR);
     }
@@ -57,19 +63,22 @@ export class IsolatedPromiseBuffer {
   public drain(timeout?: number): PromiseLike<boolean> {
     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);
 
       // 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<TransportMakeRequestResponse> {
+    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<TransportMakeRequestResponse> {
+    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));
 }

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,

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<void>(() => undefined));
+      const client = new CloudflareClient({
+        ...MOCK_CLIENT_OPTIONS,
+        flushLock: { ready: Promise.resolve(), finalize },
+      });
+
+      const privateClient = client as unknown as {
+        _transport: { flush: ReturnType<typeof vi.fn> };
+      };
+      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();
     });
   });
 

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();
   });

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();
   });

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();
   });

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();
   });

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();
   });

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();