Revert "feat(cloudflare): Introduce lock instrumentation for context.waitUntil to prevent multiple instrumentation (#17539)"

dev-packages/e2e-tests/test-applications/cloudflare-workers/tests/index.test.ts

@@ -40,7 +40,7 @@ test("Request processed by DurableObject's fetch is recorded", async ({ baseURL
 
 test('Websocket.webSocketMessage', async ({ baseURL }) => {
   const eventWaiter = waitForError('cloudflare-workers', event => {
-    return event.exception?.values?.[0]?.mechanism?.type === 'auto.faas.cloudflare.durable_object';
+    return !!event.exception?.values?.[0];
   });
   const url = new URL('/pass-to-object/ws', baseURL);
   url.protocol = url.protocol.replace('http', 'ws');
@@ -51,11 +51,12 @@ test('Websocket.webSocketMessage', async ({ baseURL }) => {
   const event = await eventWaiter;
   socket.close();
   expect(event.exception?.values?.[0]?.value).toBe('Should be recorded in Sentry: webSocketMessage');
+  expect(event.exception?.values?.[0]?.mechanism?.type).toBe('auto.faas.cloudflare.durable_object');
 });
 
 test('Websocket.webSocketClose', async ({ baseURL }) => {
   const eventWaiter = waitForError('cloudflare-workers', event => {
-    return event.exception?.values?.[0]?.mechanism?.type === 'auto.faas.cloudflare.durable_object';
+    return !!event.exception?.values?.[0];
   });
   const url = new URL('/pass-to-object/ws', baseURL);
   url.protocol = url.protocol.replace('http', 'ws');
@@ -66,4 +67,5 @@ test('Websocket.webSocketClose', async ({ baseURL }) => {
   });
   const event = await eventWaiter;
   expect(event.exception?.values?.[0]?.value).toBe('Should be recorded in Sentry: webSocketClose');
+  expect(event.exception?.values?.[0]?.mechanism?.type).toBe('auto.faas.cloudflare.durable_object');
 });

packages/cloudflare/src/client.ts

@@ -1,7 +1,7 @@
 import type { ClientOptions, Options, ServerRuntimeClientOptions } from '@sentry/core';
 import { applySdkMetadata, ServerRuntimeClient } from '@sentry/core';
+import type { makeFlushLock } from './flush';
 import type { CloudflareTransportOptions } from './transport';
-import type { makeFlushLock } from './utils/flushLock';
 
 /**
  * The Sentry Cloudflare SDK Client.

packages/cloudflare/src/durableobject.ts

@@ -18,7 +18,6 @@ import { isInstrumented, markAsInstrumented } from './instrument';
 import { getFinalOptions } from './options';
 import { wrapRequestHandler } from './request';
 import { init } from './sdk';
-import { copyExecutionContext } from './utils/copyExecutionContext';
 
 type MethodWrapperOptions = {
   spanName?: string;
@@ -193,11 +192,9 @@ export function instrumentDurableObjectWithSentry<
   C extends new (state: DurableObjectState, env: E) => T,
 >(optionsCallback: (env: E) => CloudflareOptions, DurableObjectClass: C): C {
   return new Proxy(DurableObjectClass, {
-    construct(target, [ctx, env]) {
+    construct(target, [context, env]) {
       setAsyncLocalStorageAsyncContextStrategy();
 
-      const context = copyExecutionContext(ctx);
-
       const options = getFinalOptions(optionsCallback(env), env);
 
       const obj = new target(context, env);

packages/cloudflare/src/flush.ts

@@ -0,0 +1,38 @@
+import type { ExecutionContext } from '@cloudflare/workers-types';
+
+type FlushLock = {
+  readonly ready: Promise<void>;
+  readonly finalize: () => Promise<void>;
+};
+
+/**
+ * Enhances the given execution context by wrapping its `waitUntil` method with a proxy
+ * to monitor pending tasks, and provides a flusher function to ensure all tasks
+ * have been completed before executing any subsequent logic.
+ *
+ * @param {ExecutionContext} context - The execution context to be enhanced. If no context is provided, the function returns undefined.
+ * @return {FlushLock} Returns a flusher function if a valid context is provided, otherwise undefined.
+ */
+export function makeFlushLock(context: ExecutionContext): FlushLock {
+  let resolveAllDone: () => void = () => undefined;
+  const allDone = new Promise<void>(res => {
+    resolveAllDone = res;
+  });
+  let pending = 0;
+  const originalWaitUntil = context.waitUntil.bind(context) as typeof context.waitUntil;
+  context.waitUntil = promise => {
+    pending++;
+    return originalWaitUntil(
+      promise.finally(() => {
+        if (--pending === 0) resolveAllDone();
+      }),
+    );
+  };
+  return Object.freeze({
+    ready: allDone,
+    finalize: () => {
+      if (pending === 0) resolveAllDone();
+      return allDone;
+    },
+  });
+}

packages/cloudflare/src/handler.ts

@@ -14,7 +14,6 @@ import { getFinalOptions } from './options';
 import { wrapRequestHandler } from './request';
 import { addCloudResourceContext } from './scope-utils';
 import { init } from './sdk';
-import { copyExecutionContext } from './utils/copyExecutionContext';
 
 /**
  * Wrapper for Cloudflare handlers.
@@ -38,11 +37,9 @@ export function withSentry<Env = unknown, QueueHandlerMessage = unknown, CfHostM
     if ('fetch' in handler && typeof handler.fetch === 'function' && !isInstrumented(handler.fetch)) {
       handler.fetch = new Proxy(handler.fetch, {
         apply(target, thisArg, args: Parameters<ExportedHandlerFetchHandler<Env, CfHostMetadata>>) {
-          const [request, env, ctx] = args;
+          const [request, env, context] = args;
 
           const options = getFinalOptions(optionsCallback(env), env);
-          const context = copyExecutionContext(ctx);
-          args[2] = context;
 
           return wrapRequestHandler({ options, request, context }, () => target.apply(thisArg, args));
         },
@@ -74,9 +71,7 @@ export function withSentry<Env = unknown, QueueHandlerMessage = unknown, CfHostM
     if ('scheduled' in handler && typeof handler.scheduled === 'function' && !isInstrumented(handler.scheduled)) {
       handler.scheduled = new Proxy(handler.scheduled, {
         apply(target, thisArg, args: Parameters<ExportedHandlerScheduledHandler<Env>>) {
-          const [event, env, ctx] = args;
-          const context = copyExecutionContext(ctx);
-          args[2] = context;
+          const [event, env, context] = args;
           return withIsolationScope(isolationScope => {
             const options = getFinalOptions(optionsCallback(env), env);
             const waitUntil = context.waitUntil.bind(context);
@@ -119,9 +114,7 @@ export function withSentry<Env = unknown, QueueHandlerMessage = unknown, CfHostM
     if ('email' in handler && typeof handler.email === 'function' && !isInstrumented(handler.email)) {
       handler.email = new Proxy(handler.email, {
         apply(target, thisArg, args: Parameters<EmailExportedHandler<Env>>) {
-          const [emailMessage, env, ctx] = args;
-          const context = copyExecutionContext(ctx);
-          args[2] = context;
+          const [emailMessage, env, context] = args;
           return withIsolationScope(isolationScope => {
             const options = getFinalOptions(optionsCallback(env), env);
             const waitUntil = context.waitUntil.bind(context);
@@ -162,9 +155,7 @@ export function withSentry<Env = unknown, QueueHandlerMessage = unknown, CfHostM
     if ('queue' in handler && typeof handler.queue === 'function' && !isInstrumented(handler.queue)) {
       handler.queue = new Proxy(handler.queue, {
         apply(target, thisArg, args: Parameters<ExportedHandlerQueueHandler<Env, QueueHandlerMessage>>) {
-          const [batch, env, ctx] = args;
-          const context = copyExecutionContext(ctx);
-          args[2] = context;
+          const [batch, env, context] = args;
 
           return withIsolationScope(isolationScope => {
             const options = getFinalOptions(optionsCallback(env), env);
@@ -214,9 +205,7 @@ export function withSentry<Env = unknown, QueueHandlerMessage = unknown, CfHostM
     if ('tail' in handler && typeof handler.tail === 'function' && !isInstrumented(handler.tail)) {
       handler.tail = new Proxy(handler.tail, {
         apply(target, thisArg, args: Parameters<ExportedHandlerTailHandler<Env>>) {
-          const [, env, ctx] = args;
-          const context = copyExecutionContext(ctx);
-          args[2] = context;
+          const [, env, context] = args;
 
           return withIsolationScope(async isolationScope => {
             const options = getFinalOptions(optionsCallback(env), env);

packages/cloudflare/src/sdk.ts

@@ -12,10 +12,10 @@ import {
 } from '@sentry/core';
 import type { CloudflareClientOptions, CloudflareOptions } from './client';
 import { CloudflareClient } from './client';
+import { makeFlushLock } from './flush';
 import { fetchIntegration } from './integrations/fetch';
 import { setupOpenTelemetryTracer } from './opentelemetry/tracer';
 import { makeCloudflareTransport } from './transport';
-import { makeFlushLock } from './utils/flushLock';
 import { defaultStackParser } from './vendor/stacktrace';
 
 /** Get the default integrations for the Cloudflare SDK. */

packages/cloudflare/test/durableobject.test.ts

@@ -1,9 +1,8 @@
 import type { ExecutionContext } from '@cloudflare/workers-types';
 import * as SentryCore from '@sentry/core';
-import { afterEach, describe, expect, it, vi } from 'vitest';
+import { afterEach, describe, expect, it, onTestFinished, vi } from 'vitest';
 import { instrumentDurableObjectWithSentry } from '../src';
 import { isInstrumented } from '../src/instrument';
-import { createPromiseResolver } from '../src/utils/makePromiseResolver';
 
 describe('instrumentDurableObjectWithSentry', () => {
   afterEach(() => {
@@ -123,13 +122,15 @@ describe('instrumentDurableObjectWithSentry', () => {
   });
 
   it('flush performs after all waitUntil promises are finished', async () => {
+    vi.useFakeTimers();
+    onTestFinished(() => {
+      vi.useRealTimers();
+    });
     const flush = vi.spyOn(SentryCore.Client.prototype, 'flush');
     const waitUntil = vi.fn();
-    const { promise, resolve } = createPromiseResolver();
-    process.nextTick(resolve);
     const testClass = vi.fn(context => ({
       fetch: () => {
-        context.waitUntil(promise);
+        context.waitUntil(new Promise(res => setTimeout(res)));
         return new Response('test');
       },
     }));
@@ -141,7 +142,8 @@ describe('instrumentDurableObjectWithSentry', () => {
     expect(() => dObject.fetch(new Request('https://example.com'))).not.toThrow();
     expect(flush).not.toBeCalled();
     expect(waitUntil).toHaveBeenCalledOnce();
-    await Promise.all(waitUntil.mock.calls.map(call => call[0]));
+    vi.advanceTimersToNextTimer();
+    await Promise.all(waitUntil.mock.calls.map(([p]) => p));
     expect(flush).toBeCalled();
   });