feat(cloudflare): Introduce lock instrumentation for context.waitUntil to prevent multiple instrumentation

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,6 +18,7 @@ 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;
@@ -192,9 +193,11 @@ export function instrumentDurableObjectWithSentry<
   C extends new (state: DurableObjectState, env: E) => T,
 >(optionsCallback: (env: E) => CloudflareOptions, DurableObjectClass: C): C {
   return new Proxy(DurableObjectClass, {
-    construct(target, [context, env]) {
+    construct(target, [ctx, env]) {
       setAsyncLocalStorageAsyncContextStrategy();
 
+      const context = copyExecutionContext(ctx);
+
       const options = getFinalOptions(optionsCallback(env), env);
 
       const obj = new target(context, env);

packages/cloudflare/src/handler.ts

@@ -14,6 +14,7 @@ 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.
@@ -37,9 +38,11 @@ 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, context] = args;
+          const [request, env, ctx] = args;
 
           const options = getFinalOptions(optionsCallback(env), env);
+          const context = copyExecutionContext(ctx);
+          args[2] = context;
 
           return wrapRequestHandler({ options, request, context }, () => target.apply(thisArg, args));
         },
@@ -71,7 +74,9 @@ 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, context] = args;
+          const [event, env, ctx] = args;
+          const context = copyExecutionContext(ctx);
+          args[2] = context;
           return withIsolationScope(isolationScope => {
             const options = getFinalOptions(optionsCallback(env), env);
             const waitUntil = context.waitUntil.bind(context);
@@ -114,7 +119,9 @@ 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, context] = args;
+          const [emailMessage, env, ctx] = args;
+          const context = copyExecutionContext(ctx);
+          args[2] = context;
           return withIsolationScope(isolationScope => {
             const options = getFinalOptions(optionsCallback(env), env);
             const waitUntil = context.waitUntil.bind(context);
@@ -155,7 +162,9 @@ 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, context] = args;
+          const [batch, env, ctx] = args;
+          const context = copyExecutionContext(ctx);
+          args[2] = context;
 
           return withIsolationScope(isolationScope => {
             const options = getFinalOptions(optionsCallback(env), env);
@@ -205,7 +214,9 @@ 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, context] = args;
+          const [, env, ctx] = args;
+          const context = copyExecutionContext(ctx);
+          args[2] = context;
 
           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/src/utils/copyExecutionContext.ts

@@ -0,0 +1,42 @@
+import { type DurableObjectState, type ExecutionContext } from '@cloudflare/workers-types';
+
+const kBound = Symbol.for('kBound');
+
+const defaultPropertyOptions: PropertyDescriptor = {
+  enumerable: true,
+  configurable: true,
+  writable: true,
+};
+
+/**
+ * Clones the given execution context by creating a shallow copy while ensuring the binding of specific methods.
+ *
+ * @param {ExecutionContext|DurableObjectState|void} ctx - The execution context to clone. Can be void.
+ * @return {ExecutionContext|DurableObjectState|void} A cloned execution context with bound methods, or the original void value if no context was provided.
+ */
+export function copyExecutionContext<T extends ExecutionContext | DurableObjectState>(ctx: T): T {
+  if (!ctx) return ctx;
+  return Object.create(ctx, {
+    waitUntil: { ...defaultPropertyOptions, value: copyBound(ctx, 'waitUntil') },
+    ...('passThroughOnException' in ctx && {
+      passThroughOnException: { ...defaultPropertyOptions, value: copyBound(ctx, 'passThroughOnException') },
+    }),
+  });
+}
+
+function copyBound<T, K extends keyof T>(obj: T, method: K): T[K] {
+  const method_impl = obj[method];
+  if (typeof method_impl !== 'function') return method_impl;
+  if ((method_impl as T[K] & { [kBound]?: boolean })[kBound]) return method_impl;
+
+  return new Proxy(method_impl.bind(obj), {
+    get: (target, key, receiver) => {
+      if ('bind' === key) {
+        return () => receiver;
+      } else if (kBound === key) {
+        return true;
+      }
+      return Reflect.get(target, key, receiver);
+    },
+  });
+}

packages/cloudflare/src/utils/flushLock.ts

@@ -0,0 +1,57 @@
+import type { ExecutionContext } from '@cloudflare/workers-types';
+import { createPromiseResolver } from './makePromiseResolver';
+
+type FlushLock = {
+  readonly ready: Promise<void>;
+  readonly finalize: () => Promise<void>;
+};
+type MaybeLockable<T extends object> = T & { [kFlushLock]?: FlushLock };
+
+const kFlushLock = Symbol.for('kFlushLock');
+
+function getInstrumentedLock<T extends object>(o: MaybeLockable<T>): FlushLock | undefined {
+  return o[kFlushLock];
+}
+
+function storeInstrumentedLock<T extends object>(o: MaybeLockable<T>, lock: FlushLock): void {
+  o[kFlushLock] = lock;
+}
+
+/**
+ * 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 {
+  // eslint-disable-next-line @typescript-eslint/unbound-method
+  let lock = getInstrumentedLock(context.waitUntil);
+  if (lock) {
+    // It is fine to return the same lock multiple times because this means the context has already been instrumented.
+    return lock;
+  }
+  let pending = 0;
+  const originalWaitUntil = context.waitUntil.bind(context) as typeof context.waitUntil;
+  const { promise, resolve } = createPromiseResolver();
+  const hijackedWaitUntil: typeof originalWaitUntil = promise => {
+    pending++;
+    return originalWaitUntil(
+      promise.finally(() => {
+        if (--pending === 0) resolve();
+      }),
+    );
+  };
+  lock = Object.freeze({
+    ready: promise,
+    finalize: () => {
+      if (pending === 0) resolve();
+      return promise;
+    },
+  }) as FlushLock;
+  storeInstrumentedLock(hijackedWaitUntil, lock);
+  context.waitUntil = hijackedWaitUntil;
+
+  return lock;
+}

packages/cloudflare/src/utils/makePromiseResolver.ts

@@ -0,0 +1,26 @@
+type PromiseWithResolvers<T, E = unknown> = {
+  readonly promise: Promise<T>;
+  readonly resolve: (value?: T | PromiseLike<T>) => void;
+  readonly reject: (reason?: E) => void;
+};
+/**
+ * Creates an object containing a promise, along with its corresponding resolve and reject functions.
+ *
+ * This method provides a convenient way to create a promise and access its resolvers externally.
+ *
+ * @template T - The type of the resolved value of the promise.
+ * @template E - The type of the rejected value of the promise. Defaults to `unknown`.
+ * @return {PromiseWithResolvers<T, E>} An object containing the promise and its resolve and reject functions.
+ */
+export function createPromiseResolver<T, E = unknown>(): PromiseWithResolvers<T, E> {
+  if ('withResolvers' in Promise && typeof Promise.withResolvers === 'function') {
+    return Promise.withResolvers();
+  }
+  let resolve;
+  let reject;
+  const promise = new Promise<T>((res, rej) => {
+    resolve = res;
+    reject = rej;
+  });
+  return { promise, resolve, reject } as unknown as PromiseWithResolvers<T, E>;
+}

packages/cloudflare/test/copy-execution-context.test.ts

@@ -0,0 +1,47 @@
+import { type ExecutionContext } from '@cloudflare/workers-types';
+import { type Mocked, describe, expect, it, vi } from 'vitest';
+import { copyExecutionContext } from '../src/utils/copyExecutionContext';
+
+describe('Copy of the execution context', () => {
+  describe.for<keyof ExecutionContext>(['waitUntil', 'passThroughOnException'])('%s', method => {
+    it('Was not bound more than once', async () => {
+      const context = makeExecutionContextMock();
+      const copy = copyExecutionContext(context);
+      const copy_of_copy = copyExecutionContext(copy);
+
+      expect(copy[method]).toBe(copy_of_copy[method]);
+    });
+    it('Copied method is bound to the original', async () => {
+      const context = makeExecutionContextMock();
+      const copy = copyExecutionContext(context);
+
+      expect(copy[method]()).toBe(context);
+    });
+    it('Copied method "rebind" prevention', async () => {
+      const context = makeExecutionContextMock();
+      const copy = copyExecutionContext(context);
+      expect(copy[method].bind('test')).toBe(copy[method]);
+    });
+  });
+
+  it('No side effects', async () => {
+    const context = makeExecutionContextMock();
+    expect(() => copyExecutionContext(Object.freeze(context))).not.toThrow(
+      /Cannot define property \w+, object is not extensible/,
+    );
+  });
+  it('Respects symbols', async () => {
+    const s = Symbol('test');
+    const context = makeExecutionContextMock<ExecutionContext & { [s]: unknown }>();
+    context[s] = {};
+    const copy = copyExecutionContext(context);
+    expect(copy[s]).toBe(context[s]);
+  });
+});
+
+function makeExecutionContextMock<T extends ExecutionContext>() {
+  return {
+    waitUntil: vi.fn().mockReturnThis(),
+    passThroughOnException: vi.fn().mockReturnThis(),
+  } as unknown as Mocked<T>;
+}

packages/cloudflare/test/durableobject.test.ts

@@ -1,8 +1,9 @@
 import type { ExecutionContext } from '@cloudflare/workers-types';
 import * as SentryCore from '@sentry/core';
-import { afterEach, describe, expect, it, onTestFinished, vi } from 'vitest';
+import { afterEach, describe, expect, it, vi } from 'vitest';
 import { instrumentDurableObjectWithSentry } from '../src';
 import { isInstrumented } from '../src/instrument';
+import { createPromiseResolver } from '../src/utils/makePromiseResolver';
 
 describe('instrumentDurableObjectWithSentry', () => {
   afterEach(() => {
@@ -122,15 +123,13 @@ 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(new Promise(res => setTimeout(res)));
+        context.waitUntil(promise);
         return new Response('test');
       },
     }));
@@ -142,8 +141,7 @@ describe('instrumentDurableObjectWithSentry', () => {
     expect(() => dObject.fetch(new Request('https://example.com'))).not.toThrow();
     expect(flush).not.toBeCalled();
     expect(waitUntil).toHaveBeenCalledOnce();
-    vi.advanceTimersToNextTimer();
-    await Promise.all(waitUntil.mock.calls.map(([p]) => p));
+    await Promise.all(waitUntil.mock.calls.map(call => call[0]));
     expect(flush).toBeCalled();
   });
 

packages/cloudflare/test/flush-lock.test.ts

@@ -0,0 +1,28 @@
+import { type ExecutionContext } from '@cloudflare/workers-types';
+import { describe, expect, it, vi } from 'vitest';
+import { makeFlushLock } from '../src/utils/flushLock';
+import { createPromiseResolver } from '../src/utils/makePromiseResolver';
+
+describe('Flush buffer test', () => {
+  const mockExecutionContext: ExecutionContext = {
+    waitUntil: vi.fn(),
+    passThroughOnException: vi.fn(),
+    props: null,
+  };
+  it('should flush buffer immediately if no waitUntil were called', async () => {
+    const { finalize } = makeFlushLock(mockExecutionContext);
+    await expect(finalize()).resolves.toBeUndefined();
+  });
+  it('waitUntil should not be wrapped mose than once', () => {
+    expect(makeFlushLock(mockExecutionContext), 'Execution context wrapped twice').toBe(
+      makeFlushLock(mockExecutionContext),
+    );
+  });
+  it('should flush buffer only after all waitUntil were finished', async () => {
+    const { promise, resolve } = createPromiseResolver();
+    const lock = makeFlushLock(mockExecutionContext);
+    mockExecutionContext.waitUntil(promise);
+    process.nextTick(resolve);
+    await expect(lock.finalize()).resolves.toBeUndefined();
+  });
+});