feat(core,cloudflare): Add dispose to the client for proper cleanup (#19506)

closes #19475
closes
[JS-1785](https://linear.app/getsentry/issue/JS-1785/investigate-memory-leaks-in-cloudflare)

This is a way to dispose the client entirely. Every request in
Cloudflare Workers create their own client. Once the request is done the
client would stay in memory forever, unless we `dispose` it after every
request. We also have to wait until all `waitUntil`s are finished,
otherwise we would loose these traces.

The `dispose()` method got added on purpose into the core client, as the
`getCurrentClient()` would return a `Client`. The `dispose()` method
actually only has functionality inside the `ServerRuntimeClient`, as
only the server would need this functionality.

There is still a leak in one of the default integrations, but when
running load tests against [the reproduction
repo](https://github.com/JPeer264/temp-cloudflare-leak) and setting
`defaultIntegrations: false`, then no leak is happening.

FWIW there will be a separate PR for adding a MemoryProfiler as seen in
#19364, to prevent this memory leak in the future.

Author: JPeer264

packages/cloudflare/src/client.ts

@@ -16,6 +16,9 @@ export class CloudflareClient extends ServerRuntimeClient {
   private _spanCompletionPromise: Promise<void> | null = null;
   private _resolveSpanCompletion: (() => void) | null = null;
 
+  private _unsubscribeSpanStart: (() => void) | null = null;
+  private _unsubscribeSpanEnd: (() => void) | null = null;
+
   /**
    * Creates a new Cloudflare SDK instance.
    * @param options Configuration options for this SDK.
@@ -37,7 +40,7 @@ export class CloudflareClient extends ServerRuntimeClient {
     this._flushLock = flushLock;
 
     // Track span lifecycle to know when to flush
-    this.on('spanStart', span => {
+    this._unsubscribeSpanStart = this.on('spanStart', span => {
       const spanId = span.spanContext().spanId;
       DEBUG_BUILD && debug.log('[CloudflareClient] Span started:', spanId);
       this._pendingSpans.add(spanId);
@@ -49,7 +52,7 @@ export class CloudflareClient extends ServerRuntimeClient {
       }
     });
 
-    this.on('spanEnd', span => {
+    this._unsubscribeSpanEnd = this.on('spanEnd', span => {
       const spanId = span.spanContext().spanId;
       DEBUG_BUILD && debug.log('[CloudflareClient] Span ended:', spanId);
       this._pendingSpans.delete(spanId);
@@ -99,6 +102,33 @@ export class CloudflareClient extends ServerRuntimeClient {
     return super.flush(timeout);
   }
 
+  /**
+   * Disposes of the client and releases all resources.
+   *
+   * This method clears all Cloudflare-specific state in addition to the base client cleanup.
+   * It unsubscribes from span lifecycle events and clears pending span tracking.
+   *
+   * Call this method after flushing to allow the client to be garbage collected.
+   * After calling dispose(), the client should not be used anymore.
+   */
+  public override dispose(): void {
+    DEBUG_BUILD && debug.log('[CloudflareClient] Disposing client...');
+
+    super.dispose();
+
+    if (this._unsubscribeSpanStart) {
+      this._unsubscribeSpanStart();
+      this._unsubscribeSpanStart = null;
+    }
+    if (this._unsubscribeSpanEnd) {
+      this._unsubscribeSpanEnd();
+      this._unsubscribeSpanEnd = null;
+    }
+
+    this._resetSpanCompletionPromise();
+    (this as unknown as { _flushLock: ReturnType<typeof makeFlushLock> | void })._flushLock = undefined;
+  }
+
   /**
    * Resets the span completion promise and resolve function.
    */

packages/cloudflare/src/flush.ts

@@ -1,4 +1,6 @@
 import type { ExecutionContext } from '@cloudflare/workers-types';
+import type { Client } from '@sentry/core';
+import { flush } from '@sentry/core';
 
 type FlushLock = {
   readonly ready: Promise<void>;
@@ -36,3 +38,16 @@ export function makeFlushLock(context: ExecutionContext): FlushLock {
     },
   });
 }
+
+/**
+ * Flushes the client and then disposes of it to allow garbage collection.
+ * This should be called at the end of each request to prevent memory leaks.
+ *
+ * @param client - The CloudflareClient instance to flush and dispose
+ * @param timeout - Timeout in milliseconds for the flush operation
+ * @returns A promise that resolves when flush and dispose are complete
+ */
+export async function flushAndDispose(client: Client | undefined, timeout = 2000): Promise<void> {
+  await flush(timeout);
+  client?.dispose();
+}

packages/cloudflare/src/handler.ts

@@ -1,6 +1,5 @@
 import {
   captureException,
-  flush,
   SEMANTIC_ATTRIBUTE_SENTRY_OP,
   SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN,
   SEMANTIC_ATTRIBUTE_SENTRY_SOURCE,
@@ -9,6 +8,7 @@ import {
 } from '@sentry/core';
 import { setAsyncLocalStorageAsyncContextStrategy } from './async';
 import type { CloudflareOptions } from './client';
+import { flushAndDispose } from './flush';
 import { isInstrumented, markAsInstrumented } from './instrument';
 import { getHonoIntegration } from './integrations/hono';
 import { getFinalOptions } from './options';
@@ -113,7 +113,7 @@ export function withSentry<
                   captureException(e, { mechanism: { handled: false, type: 'auto.faas.cloudflare.scheduled' } });
                   throw e;
                 } finally {
-                  waitUntil(flush(2000));
+                  waitUntil(flushAndDispose(client));
                 }
               },
             );
@@ -157,7 +157,7 @@ export function withSentry<
                   captureException(e, { mechanism: { handled: false, type: 'auto.faas.cloudflare.email' } });
                   throw e;
                 } finally {
-                  waitUntil(flush(2000));
+                  waitUntil(flushAndDispose(client));
                 }
               },
             );
@@ -209,7 +209,7 @@ export function withSentry<
                   captureException(e, { mechanism: { handled: false, type: 'auto.faas.cloudflare.queue' } });
                   throw e;
                 } finally {
-                  waitUntil(flush(2000));
+                  waitUntil(flushAndDispose(client));
                 }
               },
             );
@@ -243,7 +243,7 @@ export function withSentry<
               captureException(e, { mechanism: { handled: false, type: 'auto.faas.cloudflare.tail' } });
               throw e;
             } finally {
-              waitUntil(flush(2000));
+              waitUntil(flushAndDispose(client));
             }
           });
         },

packages/cloudflare/src/request.ts

@@ -2,7 +2,6 @@ import type { ExecutionContext, IncomingRequestCfProperties } from '@cloudflare/
 import {
   captureException,
   continueTrace,
-  flush,
   getClient,
   getHttpSpanDetailsFromUrlObject,
   httpHeadersToSpanAttributes,
@@ -14,6 +13,7 @@ import {
   withIsolationScope,
 } from '@sentry/core';
 import type { CloudflareOptions } from './client';
+import { flushAndDispose } from './flush';
 import { addCloudResourceContext, addCultureContext, addRequest } from './scope-utils';
 import { init } from './sdk';
 import { classifyResponseStreaming } from './utils/streaming';
@@ -95,7 +95,7 @@ export function wrapRequestHandler(
         }
         throw e;
       } finally {
-        waitUntil?.(flush(2000));
+        waitUntil?.(flushAndDispose(client));
       }
     }
 
@@ -122,7 +122,7 @@ export function wrapRequestHandler(
             if (captureErrors) {
               captureException(e, { mechanism: { handled: false, type: 'auto.http.cloudflare' } });
             }
-            waitUntil?.(flush(2000));
+            waitUntil?.(flushAndDispose(client));
             throw e;
           }
 
@@ -149,7 +149,7 @@ export function wrapRequestHandler(
                 } finally {
                   reader.releaseLock();
                   span.end();
-                  waitUntil?.(flush(2000));
+                  waitUntil?.(flushAndDispose(client));
                 }
               })();
 
@@ -165,14 +165,22 @@ export function wrapRequestHandler(
             } catch (e) {
               // tee() failed (e.g stream already locked) - fall back to non-streaming handling
               span.end();
-              waitUntil?.(flush(2000));
+              waitUntil?.(flushAndDispose(client));
               return res;
             }
           }
 
           // Non-streaming response - end span immediately and return original
           span.end();
-          waitUntil?.(flush(2000));
+
+          // Don't dispose for protocol upgrades (101 Switching Protocols) - the connection stays alive.
+          // This includes WebSocket upgrades where webSocketMessage/webSocketClose handlers
+          // will still be called and may need the client to capture errors.
+          if (res.status === 101) {
+            waitUntil?.(client?.flush(2000));
+          } else {
+            waitUntil?.(flushAndDispose(client));
+          }
           return res;
         });
       },

packages/cloudflare/src/workflows.ts

@@ -20,6 +20,7 @@ import type {
 } from 'cloudflare:workers';
 import { setAsyncLocalStorageAsyncContextStrategy } from './async';
 import type { CloudflareOptions } from './client';
+import { flushAndDispose } from './flush';
 import { addCloudResourceContext } from './scope-utils';
 import { init } from './sdk';
 import { instrumentContext } from './utils/instrumentContext';
@@ -186,7 +187,7 @@ export function instrumentWorkflowWithSentry<
                       new WrappedWorkflowStep(event.instanceId, context, options, step),
                     );
                   } finally {
-                    context.waitUntil(flush(2000));
+                    context.waitUntil(flushAndDispose(client));
                   }
                 });
               });

packages/cloudflare/src/wrapMethodWithSentry.ts

@@ -1,7 +1,6 @@
 import type { DurableObjectStorage } from '@cloudflare/workers-types';
 import {
   captureException,
-  flush,
   getClient,
   isThenable,
   type Scope,
@@ -12,6 +11,7 @@ import {
   withScope,
 } from '@sentry/core';
 import type { CloudflareOptions } from './client';
+import { flushAndDispose } from './flush';
 import { isInstrumented, markAsInstrumented } from './instrument';
 import { init } from './sdk';
 
@@ -74,6 +74,8 @@ export function wrapMethodWithSentry<T extends OriginalMethod>(
           scope.setClient(client);
         }
 
+        const clientToDispose = currentClient || scope.getClient();
+
         if (!wrapperOptions.spanName) {
           try {
             if (callback) {
@@ -84,7 +86,7 @@ export function wrapMethodWithSentry<T extends OriginalMethod>(
             if (isThenable(result)) {
               return result.then(
                 (res: unknown) => {
-                  waitUntil?.(flush(2000));
+                  waitUntil?.(flushAndDispose(clientToDispose));
                   return res;
                 },
                 (e: unknown) => {
@@ -94,12 +96,12 @@ export function wrapMethodWithSentry<T extends OriginalMethod>(
                       handled: false,
                     },
                   });
-                  waitUntil?.(flush(2000));
+                  waitUntil?.(flushAndDispose(clientToDispose));
                   throw e;
                 },
               );
             } else {
-              waitUntil?.(flush(2000));
+              waitUntil?.(flushAndDispose(clientToDispose));
               return result;
             }
           } catch (e) {
@@ -109,7 +111,7 @@ export function wrapMethodWithSentry<T extends OriginalMethod>(
                 handled: false,
               },
             });
-            waitUntil?.(flush(2000));
+            waitUntil?.(flushAndDispose(clientToDispose));
             throw e;
           }
         }
@@ -128,7 +130,7 @@ export function wrapMethodWithSentry<T extends OriginalMethod>(
             if (isThenable(result)) {
               return result.then(
                 (res: unknown) => {
-                  waitUntil?.(flush(2000));
+                  waitUntil?.(flushAndDispose(clientToDispose));
                   return res;
                 },
                 (e: unknown) => {
@@ -138,12 +140,12 @@ export function wrapMethodWithSentry<T extends OriginalMethod>(
                       handled: false,
                     },
                   });
-                  waitUntil?.(flush(2000));
+                  waitUntil?.(flushAndDispose(clientToDispose));
                   throw e;
                 },
               );
             } else {
-              waitUntil?.(flush(2000));
+              waitUntil?.(flushAndDispose(clientToDispose));
               return result;
             }
           } catch (e) {
@@ -153,7 +155,7 @@ export function wrapMethodWithSentry<T extends OriginalMethod>(
                 handled: false,
               },
             });
-            waitUntil?.(flush(2000));
+            waitUntil?.(flushAndDispose(clientToDispose));
             throw e;
           }
         });