fix(cloudflare): Enforce flush timeout across Workflow lifecycle

dev-packages/cloudflare-integration-tests/suites/flush-timeout/index.ts

@@ -0,0 +1,50 @@
+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. Each step flushes its span to an
+// ingest that never answers. The run reports to SERVER_URL once the SDK has aborted one of those pending sends.
+export class IssueWorkflow extends WorkflowEntrypoint<Env> {
+  async run(_event: WorkflowEvent<unknown>, step: WorkflowStep): Promise<void> {
+    const stepSendAborted = new Promise<void>(resolve => (lastSend.onAbort = resolve));
+
+    for (let index = 0; index < 100; index++) {
+      await step.do(`step-${index}`, async () => index);
+    }
+
+    await stepSendAborted;
+    await fetch(`${this.env.SERVER_URL}/result`, { method: 'POST', body: JSON.stringify({ send: 'aborted' }) });
+  }
+}
+
+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<Env>;

dev-packages/cloudflare-integration-tests/suites/flush-timeout/instrument.server.ts

@@ -0,0 +1,28 @@
+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;
+        lastSend.onAbort?.();
+      });
+      return fetch(input, init);
+    },
+  },
+}));

dev-packages/cloudflare-integration-tests/suites/flush-timeout/lastSend.ts

@@ -0,0 +1,3 @@
+// `aborted`: whether the transport aborted an envelope fetch since it was last reset.
+// `onAbort`: called each time the transport aborts an envelope fetch.
+export const lastSend: { aborted: boolean; onAbort?: () => void } = { aborted: false };

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<unknown> }> {
+  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<void>(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' });
+  },
+);
+
+// The local runtime does not settle a cached client's step drains while the run waits, 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 the pending send of a step when its 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({ 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();
+});

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

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",
+    },
+  ],
+}

dev-packages/cloudflare-integration-tests/suites/workflows/step-flush/index.ts

@@ -0,0 +1,28 @@
+import { WorkflowEntrypoint } from 'cloudflare:workers';
+import type { WorkflowEvent, WorkflowStep } from 'cloudflare:workers';
+
+interface Env {
+  SENTRY_DSN: string;
+  SLEEP_WORKFLOW: Workflow;
+}
+
+export class SleepWorkflow extends WorkflowEntrypoint<Env> {
+  async run(_event: WorkflowEvent<unknown>, step: WorkflowStep): Promise<void> {
+    await step.do('before-sleep', async () => 'done');
+    await step.sleep('pause', '1 hour');
+    await step.do('after-sleep', async () => 'done');
+  }
+}
+
+export default {
+  async fetch(request: Request, env: Env): Promise<Response> {
+    const url = new URL(request.url);
+
+    if (url.pathname === '/workflow/trigger') {
+      const instance = await env.SLEEP_WORKFLOW.create();
+      return Response.json({ id: instance.id });
+    }
+
+    return new Response('Not found', { status: 404 });
+  },
+} satisfies ExportedHandler<Env>;

dev-packages/cloudflare-integration-tests/suites/workflows/step-flush/instrument.server.ts

@@ -0,0 +1,6 @@
+import { defineCloudflareOptions } from '@sentry/cloudflare';
+
+export default defineCloudflareOptions((env: { SENTRY_DSN: string }) => ({
+  dsn: env.SENTRY_DSN,
+  tracesSampleRate: 1.0,
+}));

dev-packages/cloudflare-integration-tests/suites/workflows/step-flush/test.ts

@@ -0,0 +1,27 @@
+import type { SerializedStreamedSpanContainer } from '@sentry/core';
+import { expect, it } from 'vitest';
+import { createRunner } from '../../../runner';
+
+it('sends the span of a step before the Workflow goes to sleep', async ({ signal }) => {
+  const runner = createRunner(__dirname)
+    .expect(envelope => {
+      const container = envelope[1].find(item => item[0].type === 'span')?.[1] as SerializedStreamedSpanContainer;
+
+      expect(container.items).toHaveLength(1);
+      expect(container.items[0]!.name).toBe('before-sleep');
+      expect(envelope[0].trace).toEqual({
+        environment: 'production',
+        public_key: 'public',
+        trace_id: container.items[0]!.trace_id,
+        transaction: 'before-sleep',
+        sampled: 'true',
+        sample_rand: expect.any(String),
+        sample_rate: '1',
+      });
+    })
+    .unordered()
+    .start(signal);
+
+  await runner.makeRequest('get', '/workflow/trigger');
+  await runner.completed();
+});

dev-packages/cloudflare-integration-tests/suites/workflows/step-flush/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()],
+});

dev-packages/cloudflare-integration-tests/suites/workflows/step-flush/wrangler.jsonc

@@ -0,0 +1,14 @@
+{
+  "$schema": "../../../node_modules/wrangler/config-schema.json",
+  "name": "cloudflare-workflow-step-flush",
+  "main": "index.ts",
+  "compatibility_date": "2025-06-17",
+  "compatibility_flags": ["nodejs_compat"],
+  "workflows": [
+    {
+      "name": "sleep-workflow",
+      "binding": "SLEEP_WORKFLOW",
+      "class_name": "SleepWorkflow",
+    },
+  ],
+}

dev-packages/e2e-tests/test-applications/cloudflare-workers-send-to-sentry/deployed-worker.mjs

@@ -1,5 +1,5 @@
 import { execFileSync } from 'node:child_process';
-import { mkdtempSync, readFileSync, rmSync } from 'node:fs';
+import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
 import { tmpdir } from 'node:os';
 import { dirname, join } from 'node:path';
 import { fileURLToPath } from 'node:url';
@@ -14,8 +14,24 @@ function wrangler(args, env = {}) {
   });
 }
 
+/**
+ * Workflow names are unique per Cloudflare account, so every worker gets its own. The Vite build writes the
+ * config wrangler deploys from, and `.wrangler/deploy/config.json` points to it.
+ */
+function nameWorkflowsAfterWorker(name) {
+  const redirect = JSON.parse(readFileSync(join(__dirname, '.wrangler/deploy/config.json'), 'utf8'));
+  const configPath = join(__dirname, '.wrangler/deploy', redirect.configPath);
+  const config = JSON.parse(readFileSync(configPath, 'utf8'));
+
+  for (const workflow of config.workflows ?? []) {
+    workflow.name = name;
+  }
+  writeFileSync(configPath, JSON.stringify(config, null, 2));
+}
+
 /** Deploys the worker under `name` and returns its workers.dev URL. */
 export function deployWorker(name, dsn) {
+  nameWorkflowsAfterWorker(name);
   const outputDir = mkdtempSync(join(tmpdir(), 'wrangler-output-'));
   const outputFile = join(outputDir, 'output.ndjson');
 

dev-packages/e2e-tests/test-applications/cloudflare-workers-send-to-sentry/src/env.d.ts

@@ -1,3 +1,4 @@
 interface Env {
   E2E_TEST_DSN: string;
+  SLEEP_WORKFLOW: Workflow;
 }

dev-packages/e2e-tests/test-applications/cloudflare-workers-send-to-sentry/src/index.ts

@@ -1,7 +1,20 @@
 import * as Sentry from '@sentry/cloudflare';
+import { WorkflowEntrypoint } from 'cloudflare:workers';
+import type { WorkflowEvent, WorkflowStep } from 'cloudflare:workers';
+
+export class SleepWorkflow extends WorkflowEntrypoint<Env> {
+  async run(_event: WorkflowEvent<unknown>, step: WorkflowStep): Promise<void> {
+    for (let index = 0; index < 3; index++) {
+      await step.do(`before-sleep-${index}`, async () => index);
+    }
+
+    await step.sleep('pause', '10 minutes');
+    await step.do('after-sleep', async () => 'done');
+  }
+}
 
 export default {
-  async fetch(request) {
+  async fetch(request, env) {
     const url = new URL(request.url);
     // The handler runs inside the request span the Vite plugin's `withSentry` wrapper starts, so
     // this is the `http.server` span.
@@ -16,6 +29,14 @@ export default {
         throw new Error('E2E test unhandled error');
       case '/test-span':
         return Response.json({ spanId: spanContext?.spanId, traceId: spanContext?.traceId });
+      case '/test-workflow-sleep': {
+        const instance = await env.SLEEP_WORKFLOW.create({ id: crypto.randomUUID() });
+        return Response.json({ instanceId: instance.id, traceId: instance.id.replace(/-/g, '') });
+      }
+      case '/test-workflow-status': {
+        const instance = await env.SLEEP_WORKFLOW.get(url.searchParams.get('id') ?? '');
+        return Response.json(await instance.status());
+      }
       default:
         return new Response('Hello World!');
     }

dev-packages/e2e-tests/test-applications/cloudflare-workers-send-to-sentry/tests/send-to-sentry.test.ts

@@ -1,6 +1,13 @@
 import { randomBytes } from 'node:crypto';
 import { expect, test } from '@playwright/test';
-import { EVENT_POLLING_OPTIONS, findErrorInTrace, findSpanInTrace, traceTarget } from '@sentry-internal/test-utils/cli';
+import {
+  EVENT_POLLING_OPTIONS,
+  fetchTrace,
+  findErrorInTrace,
+  findSpanInTrace,
+  flattenTrace,
+  traceTarget,
+} from '@sentry-internal/test-utils/cli';
 
 // Set by global-setup.mjs once the worker for this run is deployed.
 const workerUrl = process.env.E2E_TEST_WORKER_URL;
@@ -43,3 +50,24 @@ test('Sends a request span to Sentry', async () => {
     .poll(() => findSpanInTrace(traceId, 'http.server'), EVENT_POLLING_OPTIONS)
     .toMatchObject({ event_id: spanId });
 });
+
+test('Sends the spans of Workflow steps before the Workflow goes to sleep', async () => {
+  const response = await fetch(`${workerUrl}/test-workflow-sleep`);
+  expect(response.status).toBe(200);
+  const { instanceId, traceId } = await response.json();
+
+  console.log(`Polling for the Workflow step spans: sentry trace view ${traceTarget(traceId)}`);
+
+  await expect
+    .poll(
+      () =>
+        flattenTrace(fetchTrace(traceId)).filter(
+          item => item.event_type === 'span' && item.op === 'function' && item.description?.startsWith('before-sleep-'),
+        ).length,
+      EVENT_POLLING_OPTIONS,
+    )
+    .toBe(3);
+
+  const { status } = await fetch(`${workerUrl}/test-workflow-status?id=${instanceId}`).then(res => res.json());
+  expect(['running', 'waiting']).toContain(status);
+});

dev-packages/e2e-tests/test-applications/cloudflare-workers-send-to-sentry/wrangler.jsonc

@@ -8,4 +8,8 @@
   "workers_dev": true,
   // Workers Logs keep the invocations of the last 7 days, so a failed CI run can still be inspected.
   "observability": { "enabled": true },
+  // Workflow names are unique per account, so deployWorker() renames it to the worker name.
+  "workflows": [
+    { "name": "cloudflare-workers-send-to-sentry", "binding": "SLEEP_WORKFLOW", "class_name": "SleepWorkflow" },
+  ],
 }