diff --git a/packages/cloudflare/src/client.ts b/packages/cloudflare/src/client.ts index 3b4f591d5e22..bef4f0536d47 100644 --- a/packages/cloudflare/src/client.ts +++ b/packages/cloudflare/src/client.ts @@ -1,4 +1,4 @@ -import type { ClientOptions, Options, TracePropagationTargets } from '@sentry/core'; +import type { CheckIn, ClientOptions, MonitorConfig, Options, Scope, TracePropagationTargets } from '@sentry/core'; import { _INTERNAL_clearAiProviderSkips, _INTERNAL_flushLogsBuffer, @@ -163,6 +163,20 @@ export class CloudflareClient extends ServerRuntimeClient { } } + /** + * Sends the check-in at once instead of holding it for the flush at the end of the invocation. + * An `in_progress` check-in of a run that is longer than the check-in margin would otherwise + * reach Sentry too late, and Sentry would mark the run as missed. + */ + public override captureCheckIn(checkIn: CheckIn, monitorConfig?: MonitorConfig, scope?: Scope): string { + const checkInId = super.captureCheckIn(checkIn, monitorConfig, scope); + const transport = this.getTransport(); + if (transport) { + this._registerWithInvocationWaitUntil(transport.flush(2000)); + } + return checkInId; + } + /** * Disposes of the client and releases all resources. * diff --git a/packages/cloudflare/src/instrumentations/worker/instrumentScheduled.ts b/packages/cloudflare/src/instrumentations/worker/instrumentScheduled.ts index 8bf690d05175..8285830b73c1 100644 --- a/packages/cloudflare/src/instrumentations/worker/instrumentScheduled.ts +++ b/packages/cloudflare/src/instrumentations/worker/instrumentScheduled.ts @@ -65,11 +65,6 @@ function wrapScheduledHandler( finishCheckIn = client ?.getIntegrationByName('CronTriggers') ?.startCheckIn(controller.cron); - // The transport holds envelopes until the flush after the handler, so a run that is longer than - // the check-in margin would deliver its in_progress check-in too late and show as missed. - if (finishCheckIn) { - waitUntil(Promise.resolve(client?.getTransport()?.flush(2000)).catch(() => undefined)); - } } catch (e) { DEBUG_BUILD && debug.warn('Failed to send the in_progress cron check-in', e); } diff --git a/packages/cloudflare/src/integrations/cronTriggers.ts b/packages/cloudflare/src/integrations/cronTriggers.ts index 289cfd18c849..4f49824018c1 100644 --- a/packages/cloudflare/src/integrations/cronTriggers.ts +++ b/packages/cloudflare/src/integrations/cronTriggers.ts @@ -139,8 +139,6 @@ const _cronTriggersIntegration = ((options: CronTriggersOptions = {}): CronTrigg ); } - // Check-ins are captured directly rather than through `withMonitor`, which would fork the - // isolation scope and lose the invocation state attached to it. const checkInId = captureCheckIn( { monitorSlug, status: 'in_progress' }, crontab ? { ...monitorSettings, schedule: { type: 'crontab', value: crontab } } : undefined, diff --git a/packages/cloudflare/src/transport.ts b/packages/cloudflare/src/transport.ts index b7f5a3181680..ee7bcce7730b 100644 --- a/packages/cloudflare/src/transport.ts +++ b/packages/cloudflare/src/transport.ts @@ -66,12 +66,16 @@ export class IsolatedPromiseBuffer { const drainController = new AbortController(); this.drainSignal = drainController.signal; - let tasks: PromiseLike[]; - try { - tasks = oldTaskProducers.map(taskProducer => taskProducer()); - } finally { - this.drainSignal = undefined; - } + const tasks = oldTaskProducers.map(taskProducer => { + // A request that throws synchronously, such as a custom `fetch`, must not stop the other requests + // or throw into the caller. + try { + return taskProducer(); + } catch (error) { + return Promise.reject(error); + } + }); + this.drainSignal = undefined; return new Promise(resolve => { const timer = setTimeout(() => { diff --git a/packages/cloudflare/src/utils/invocationContext.ts b/packages/cloudflare/src/utils/invocationContext.ts index 5c343c3bd88a..aa2f55482a2f 100644 --- a/packages/cloudflare/src/utils/invocationContext.ts +++ b/packages/cloudflare/src/utils/invocationContext.ts @@ -16,8 +16,9 @@ import { getOriginalWaitUntil } from '../flush'; * * The state rides on the invocation's forked isolation scope, which the async * context strategy (AsyncLocalStorage) hands back for exactly the async context that - * owns the invocation — including detached continuations created inside it. A symbol - * key keeps it off `Scope.clone()` and out of serialized event data. + * owns the invocation — including detached continuations created inside it. It lives under + * a symbol key in the scope's `refs`, so isolation scopes forked inside the invocation + * (`withMonitor`, `withIsolationScope`) share it, and it stays out of serialized event data. */ export interface InvocationState { /** @@ -63,7 +64,7 @@ export function getInvocationWaitUntil(state: InvocationState): ExecutionContext return state.waitUntil ?? undefined; } -type ScopeWithInvocationState = Scope & { +type ScopeRefsWithInvocationState = Scope['refs'] & { [INVOCATION_STATE]?: InvocationState; }; @@ -81,7 +82,7 @@ export function setInvocationState(scope: Scope, state: InvocationState): void { return; } - (scope as ScopeWithInvocationState)[INVOCATION_STATE] = state; + (scope.refs as ScopeRefsWithInvocationState)[INVOCATION_STATE] = state; } /** @@ -91,9 +92,9 @@ export function setInvocationState(scope: Scope, state: InvocationState): void { * is never attached there in the first place). */ export function getInvocationState(): InvocationState | undefined { - const isolationScope = getIsolationScope() as ScopeWithInvocationState; + const isolationScope = getIsolationScope(); if (isolationScope === getDefaultIsolationScope()) { return undefined; } - return isolationScope[INVOCATION_STATE]; + return (isolationScope.refs as ScopeRefsWithInvocationState)[INVOCATION_STATE]; } diff --git a/packages/cloudflare/test/client.test.ts b/packages/cloudflare/test/client.test.ts index 2d7828be238d..27d0eae75835 100644 --- a/packages/cloudflare/test/client.test.ts +++ b/packages/cloudflare/test/client.test.ts @@ -2,6 +2,7 @@ import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'; import { setAsyncLocalStorageAsyncContextStrategy } from '@sentry/server-utils/no-diagnostic-channels'; import { CloudflareClient, type CloudflareClientOptions } from '../src/client'; import { makeFlushLock } from '../src/flush'; +import { makeCloudflareTransport } from '../src/transport'; import { getInvocationState } from '../src/utils/invocationContext'; import { withInvocationIsolationScope } from '../src/utils/invocationScope'; @@ -797,4 +798,50 @@ describe('CloudflareClient', () => { expect(flushSpy).not.toHaveBeenCalled(); }); }); + + describe('check-in delivery', () => { + it.each([true, false])( + 'drains the transport for a check-in before the flush point and registers it with the invocation waitUntil (cacheClient: %s)', + cacheClient => { + const flushMock = vi.fn().mockResolvedValue(true); + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + cacheClient, + transport: () => ({ send: vi.fn().mockResolvedValue({}), flush: flushMock }), + }); + client.init(); + const ctx = { waitUntil: vi.fn(), passThroughOnException: vi.fn() }; + + withInvocationIsolationScope(() => { + client.captureCheckIn({ monitorSlug: 'foo', status: 'in_progress' }); + }, ctx as never); + + expect(flushMock).toHaveBeenCalledTimes(1); + expect(ctx.waitUntil).toHaveBeenCalledTimes(1); + expect(ctx.waitUntil).toHaveBeenCalledWith(expect.any(Promise)); + }, + ); + + it('does not throw when the fetch of the transport throws synchronously', async () => { + const client = new CloudflareClient({ + ...MOCK_CLIENT_OPTIONS, + transport: makeCloudflareTransport, + transportOptions: { + fetch: () => { + throw new Error('foo'); + }, + }, + }); + client.init(); + const waits: Promise[] = []; + const ctx = { waitUntil: vi.fn(promise => waits.push(promise)), passThroughOnException: vi.fn() }; + + withInvocationIsolationScope(() => { + expect(() => client.captureCheckIn({ monitorSlug: 'foo', status: 'in_progress' })).not.toThrow(); + }, ctx as never); + + expect(waits).toHaveLength(1); + await expect(Promise.all(waits)).resolves.toBeDefined(); + }); + }); }); diff --git a/packages/cloudflare/test/instrumentations/worker/instrumentScheduled.test.ts b/packages/cloudflare/test/instrumentations/worker/instrumentScheduled.test.ts index 33649a188f06..843683be8925 100644 --- a/packages/cloudflare/test/instrumentations/worker/instrumentScheduled.test.ts +++ b/packages/cloudflare/test/instrumentations/worker/instrumentScheduled.test.ts @@ -2,7 +2,7 @@ // Although this is not ideal, this is the best we can do until we have a better way to test cloudflare workers. import type { ExecutionContext, ScheduledController } from '@cloudflare/workers-types'; -import type { Event, Integration } from '@sentry/core'; +import type { Event, Integration, SerializedCheckIn } from '@sentry/core'; import * as SentryCore from '@sentry/core'; import { beforeEach, describe, expect, onTestFinished, test, vi } from 'vitest'; import { CloudflareClient } from '../../../src/client'; @@ -478,6 +478,59 @@ describe('instrumentScheduled', () => { expect(sentItemTypes.filter(type => type === 'check_in')).toHaveLength(2); }); + test('sends the in_progress check-in of withMonitor while its callback runs', async () => { + vi.useFakeTimers(); + onTestFinished(() => { + vi.useRealTimers(); + }); + // Like the Cloudflare transport, this transport holds each envelope until it is drained. + const queued: string[] = []; + const delivered: string[] = []; + const waits: Promise[] = []; + const wrappedHandler = withSentry( + env => ({ + dsn: env.SENTRY_DSN, + transport: () => ({ + send: async envelope => { + for (const [itemHeader, item] of envelope[1]) { + if (itemHeader.type === 'check_in') { + queued.push((item as SerializedCheckIn).status); + } + } + return {}; + }, + flush: async () => { + delivered.push(...queued.splice(0)); + return true; + }, + }), + }), + { + async scheduled() { + await SentryCore.withMonitor('foo', () => new Promise(resolve => setTimeout(resolve, 5 * 60_000)), { + schedule: { type: 'crontab', value: '30 9 * * *' }, + checkinMargin: 1, + }); + }, + } satisfies ExportedHandler, + ); + + const run = wrappedHandler.scheduled?.(controllerFor('30 9 * * 1-5'), MOCK_ENV, { + waitUntil: vi.fn(promise => waits.push(promise)), + passThroughOnException: vi.fn(), + } as unknown as ExecutionContext); + + await vi.advanceTimersByTimeAsync(60_000); + expect(delivered).toEqual(['in_progress']); + expect(waits).toHaveLength(1); + + await vi.advanceTimersByTimeAsync(4 * 60_000); + await run; + await vi.runAllTimersAsync(); + await Promise.all(waits); + expect(delivered).toEqual(['in_progress', 'ok']); + }); + test('drains the transport for the in_progress check-in while the handler runs', async () => { const transportFlush = vi.fn(async () => true); let finishHandler: () => void = () => undefined; diff --git a/packages/cloudflare/test/transport.test.ts b/packages/cloudflare/test/transport.test.ts index 1750176f5591..3fa9f99f157f 100644 --- a/packages/cloudflare/test/transport.test.ts +++ b/packages/cloudflare/test/transport.test.ts @@ -1,4 +1,4 @@ -import type { EventEnvelope, EventItem } from '@sentry/core'; +import type { EventEnvelope, EventItem, TransportMakeRequestResponse } from '@sentry/core'; import { createEnvelope, SENTRY_BUFFER_FULL_ERROR, serializeEnvelope } from '@sentry/core'; import { afterAll, describe, expect, it, vi } from 'vitest'; import type { CloudflareTransportOptions } from '../src/transport'; @@ -236,6 +236,22 @@ describe('IsolatedPromiseBuffer', () => { expect(task2).toHaveBeenCalled(); }); + it('runs the other tasks of a drain when one task throws synchronously', async () => { + const ipb = new IsolatedPromiseBuffer(); + + const task1 = vi.fn((): Promise => { + throw new Error('foo'); + }); + const task2 = vi.fn(() => Promise.resolve({})); + + await ipb.add(task1); + await ipb.add(task2); + + await expect(ipb.drain()).resolves.toEqual(true); + + expect(task2).toHaveBeenCalled(); + }); + it('should allow for a custom fetch function to be passed in', async () => { const customFetch = vi.fn(async () => { return { diff --git a/packages/cloudflare/test/utils/invocationContext.test.ts b/packages/cloudflare/test/utils/invocationContext.test.ts index cd05ebf6b8cd..3378f75dfdfb 100644 --- a/packages/cloudflare/test/utils/invocationContext.test.ts +++ b/packages/cloudflare/test/utils/invocationContext.test.ts @@ -48,13 +48,16 @@ describe('invocation state', () => { expect(getInvocationState()).toBeUndefined(); }); - it('is not inherited by scope clones', () => { - const scope = getDefaultIsolationScope().clone(); - setInvocationState(scope, { ctx: undefined }); + it('is shared with isolation scopes forked inside the invocation', () => { + const ctx = { waitUntil: vi.fn(), passThroughOnException: vi.fn() }; - withIsolationScope(scope.clone(), () => { - expect(getInvocationState()).toBeUndefined(); - }); + withInvocationIsolationScope(() => { + const state = getInvocationState(); + + withIsolationScope(() => { + expect(getInvocationState()).toBe(state); + }); + }, ctx); }); it('is attached at the invocation entry point and kept when reentrant', () => {