Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion packages/cloudflare/src/client.ts
Original file line number Diff line number Diff line change
@@ -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,
Expand Down Expand Up @@ -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 {
Comment thread
JPeer264 marked this conversation as resolved.
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.
*
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -65,11 +65,6 @@ function wrapScheduledHandler(
finishCheckIn = client
?.getIntegrationByName<CronTriggersIntegration>('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);
}
Expand Down
2 changes: 0 additions & 2 deletions packages/cloudflare/src/integrations/cronTriggers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
16 changes: 10 additions & 6 deletions packages/cloudflare/src/transport.ts
Original file line number Diff line number Diff line change
Expand Up @@ -66,12 +66,16 @@ export class IsolatedPromiseBuffer {

const drainController = new AbortController();
this.drainSignal = drainController.signal;
let tasks: PromiseLike<TransportMakeRequestResponse>[];
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(() => {
Expand Down
13 changes: 7 additions & 6 deletions packages/cloudflare/src/utils/invocationContext.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
/**
Expand Down Expand Up @@ -63,7 +64,7 @@ export function getInvocationWaitUntil(state: InvocationState): ExecutionContext
return state.waitUntil ?? undefined;
}

type ScopeWithInvocationState = Scope & {
type ScopeRefsWithInvocationState = Scope['refs'] & {
[INVOCATION_STATE]?: InvocationState;
};

Expand All @@ -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;
}

/**
Expand All @@ -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];
}
47 changes: 47 additions & 0 deletions packages/cloudflare/test/client.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';

Expand Down Expand Up @@ -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<unknown>[] = [];
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();
});
});
});
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -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<unknown>[] = [];
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<typeof MOCK_ENV>,
);

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;
Expand Down
18 changes: 17 additions & 1 deletion packages/cloudflare/test/transport.test.ts
Original file line number Diff line number Diff line change
@@ -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';
Expand Down Expand Up @@ -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<TransportMakeRequestResponse> => {
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 {
Expand Down
15 changes: 9 additions & 6 deletions packages/cloudflare/test/utils/invocationContext.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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', () => {
Expand Down
Loading