fix(core): Add a PromiseBuffer for incoming events on the client (#18120)

## Problem

Previously, the client would process all incoming events without any
limit, which could lead to unbounded growth of pending events/promises
in memory. This could cause performance issues and memory pressure in
high-throughput scenarios. This occurs when two conditions are met:

- when an integration with an async `processEvent` are added (e.g.
`ContextLines`, which is a defaultIntegration)
- events, e.g. `Sentry.captureException`, are called synchronously

```js
Sentry.init({ ... });

// ...

for (let i = 0; i < 5000; i++) {
  Sentry.captureException(new Error());
}
```

## Solution

This PR adds a `PromiseBuffer` to the `Client` class to limit the number
of concurrent event processing operations.

- Introduced a `_promiseBuffer` in the `Client` class that limits
concurrent event processing
- The buffer size defaults to `DEFAULT_TRANSPORT_BUFFER_SIZE` (64) but
can be configured via `transportOptions.bufferSize`
- When the buffer is full, events are rejected and properly tracked as
dropped events with the `queue_overflow` reason
  - Please tak
- Modified the `_process()` method to:
- Accept a task producer function instead of a promise directly (lazy
evaluation)
  - Use the promise buffer to manage concurrent operations
  - Track the data category for proper dropped event categorization

## Special 👀  on

- About reusing `transportOptions.bufferSize`: Not sure if this is the
best technique, but IMO both should have the same size - because if it
wouldn't it would be capped at a later stage (asking myself if the
transport still needs the promise buffer - as we have it now way earlier
in place)
- The `_process` takes now a `DataCategory`. At the time of the process
the event type is almost unknown. Not sure if I assumed the categories
correctly there, or if there is another technique of getting the type
(**edit:** a [comment by
Cursor](https://github.com/getsentry/sentry-javascript/pull/18120/files/2ee14b484d00432145d4f9a6773fbd31f92921d7#r2504259236)
helped a little and I added [a helper
function](7381a49))
- `recordDroppedEvent` is now printing it one after each other -
theoretically we can count all occurences and print the count on it. I
decided against this one, since it would delay the user feedback - this
can be challenged though

Author: JPeer264

.size-limit.js

@@ -38,7 +38,7 @@ module.exports = [
     path: 'packages/browser/build/npm/esm/prod/index.js',
     import: createImport('init', 'browserTracingIntegration'),
     gzip: true,
-    limit: '41.38 KB',
+    limit: '41.5 KB',
   },
   {
     name: '@sentry/browser (incl. Tracing, Profiling)',
@@ -127,7 +127,7 @@ module.exports = [
     import: createImport('init', 'ErrorBoundary', 'reactRouterV6BrowserTracingIntegration'),
     ignore: ['react/jsx-runtime'],
     gzip: true,
-    limit: '43.33 KB',
+    limit: '43.5 KB',
   },
   // Vue SDK (ESM)
   {
@@ -142,7 +142,7 @@ module.exports = [
     path: 'packages/vue/build/esm/index.js',
     import: createImport('init', 'browserTracingIntegration'),
     gzip: true,
-    limit: '43.2 KB',
+    limit: '43.3 KB',
   },
   // Svelte SDK (ESM)
   {
@@ -163,7 +163,7 @@ module.exports = [
     name: 'CDN Bundle (incl. Tracing)',
     path: createCDNPath('bundle.tracing.min.js'),
     gzip: true,
-    limit: '42 KB',
+    limit: '42.1 KB',
   },
   {
     name: 'CDN Bundle (incl. Tracing, Replay)',
@@ -231,7 +231,7 @@ module.exports = [
     import: createImport('init'),
     ignore: [...builtinModules, ...nodePrefixedBuiltinModules],
     gzip: true,
-    limit: '51 KB',
+    limit: '51.1 KB',
   },
   // Node SDK (ESM)
   {

packages/core/src/client.ts

@@ -11,13 +11,14 @@ import { _INTERNAL_flushMetricsBuffer } from './metrics/internal';
 import type { Scope } from './scope';
 import { updateSession } from './session';
 import { getDynamicSamplingContextFromScope } from './tracing/dynamicSamplingContext';
+import { DEFAULT_TRANSPORT_BUFFER_SIZE } from './transports/base';
 import type { Breadcrumb, BreadcrumbHint, FetchBreadcrumbHint, XhrBreadcrumbHint } from './types-hoist/breadcrumb';
 import type { CheckIn, MonitorConfig } from './types-hoist/checkin';
 import type { EventDropReason, Outcome } from './types-hoist/clientreport';
 import type { DataCategory } from './types-hoist/datacategory';
 import type { DsnComponents } from './types-hoist/dsn';
 import type { DynamicSamplingContext, Envelope } from './types-hoist/envelope';
-import type { ErrorEvent, Event, EventHint, TransactionEvent } from './types-hoist/event';
+import type { ErrorEvent, Event, EventHint, EventType, TransactionEvent } from './types-hoist/event';
 import type { EventProcessor } from './types-hoist/eventprocessor';
 import type { FeedbackEvent } from './types-hoist/feedback';
 import type { Integration } from './types-hoist/integration';
@@ -43,6 +44,7 @@ import { merge } from './utils/merge';
 import { checkOrSetAlreadyCaught, uuid4 } from './utils/misc';
 import { parseSampleRate } from './utils/parseSampleRate';
 import { prepareEvent } from './utils/prepareEvent';
+import { type PromiseBuffer, makePromiseBuffer, SENTRY_BUFFER_FULL_ERROR } from './utils/promisebuffer';
 import { reparentChildSpans, shouldIgnoreSpan } from './utils/should-ignore-span';
 import { showSpanDropWarning } from './utils/spanUtils';
 import { rejectedSyncPromise } from './utils/syncpromise';
@@ -201,6 +203,8 @@ export abstract class Client<O extends ClientOptions = ClientOptions> {
   // eslint-disable-next-line @typescript-eslint/ban-types
   private _hooks: Record<string, Set<Function>>;
 
+  private _promiseBuffer: PromiseBuffer<unknown>;
+
   /**
    * Initializes this client instance.
    *
@@ -213,6 +217,7 @@ export abstract class Client<O extends ClientOptions = ClientOptions> {
     this._outcomes = {};
     this._hooks = {};
     this._eventProcessors = [];
+    this._promiseBuffer = makePromiseBuffer(options.transportOptions?.bufferSize ?? DEFAULT_TRANSPORT_BUFFER_SIZE);
 
     if (options.dsn) {
       this._dsn = makeDsn(options.dsn);
@@ -275,9 +280,11 @@ export abstract class Client<O extends ClientOptions = ClientOptions> {
     };
 
     this._process(
-      this.eventFromException(exception, hintWithEventId).then(event =>
-        this._captureEvent(event, hintWithEventId, scope),
-      ),
+      () =>
+        this.eventFromException(exception, hintWithEventId)
+          .then(event => this._captureEvent(event, hintWithEventId, scope))
+          .then(res => res),
+      'error',
     );
 
     return hintWithEventId.event_id;
@@ -300,12 +307,15 @@ export abstract class Client<O extends ClientOptions = ClientOptions> {
     };
 
     const eventMessage = isParameterizedString(message) ? message : String(message);
-
-    const promisedEvent = isPrimitive(message)
+    const isMessage = isPrimitive(message);
+    const promisedEvent = isMessage
       ? this.eventFromMessage(eventMessage, level, hintWithEventId)
       : this.eventFromException(message, hintWithEventId);
 
-    this._process(promisedEvent.then(event => this._captureEvent(event, hintWithEventId, currentScope)));
+    this._process(
+      () => promisedEvent.then(event => this._captureEvent(event, hintWithEventId, currentScope)),
+      isMessage ? 'unknown' : 'error',
+    );
 
     return hintWithEventId.event_id;
   }
@@ -332,9 +342,11 @@ export abstract class Client<O extends ClientOptions = ClientOptions> {
     const sdkProcessingMetadata = event.sdkProcessingMetadata || {};
     const capturedSpanScope: Scope | undefined = sdkProcessingMetadata.capturedSpanScope;
     const capturedSpanIsolationScope: Scope | undefined = sdkProcessingMetadata.capturedSpanIsolationScope;
+    const dataCategory = getDataCategoryByType(event.type);
 
     this._process(
-      this._captureEvent(event, hintWithEventId, capturedSpanScope || currentScope, capturedSpanIsolationScope),
+      () => this._captureEvent(event, hintWithEventId, capturedSpanScope || currentScope, capturedSpanIsolationScope),
+      dataCategory,
     );
 
     return hintWithEventId.event_id;
@@ -1252,7 +1264,7 @@ export abstract class Client<O extends ClientOptions = ClientOptions> {
       );
     }
 
-    const dataCategory = (eventType === 'replay_event' ? 'replay' : eventType) satisfies DataCategory;
+    const dataCategory = getDataCategoryByType(event.type);
 
     return this._prepareEvent(event, hint, currentScope, isolationScope)
       .then(prepared => {
@@ -1335,15 +1347,21 @@ export abstract class Client<O extends ClientOptions = ClientOptions> {
   /**
    * Occupies the client with processing and event
    */
-  protected _process<T>(promise: PromiseLike<T>): void {
+  protected _process<T>(taskProducer: () => PromiseLike<T>, dataCategory: DataCategory): void {
     this._numProcessing++;
-    void promise.then(
+
+    void this._promiseBuffer.add(taskProducer).then(
       value => {
         this._numProcessing--;
         return value;
       },
       reason => {
         this._numProcessing--;
+
+        if (reason === SENTRY_BUFFER_FULL_ERROR) {
+          this.recordDroppedEvent('queue_overflow', dataCategory);
+        }
+
         return reason;
       },
     );
@@ -1408,6 +1426,10 @@ export abstract class Client<O extends ClientOptions = ClientOptions> {
   ): PromiseLike<Event>;
 }
 
+function getDataCategoryByType(type: EventType | 'replay_event' | undefined): DataCategory {
+  return type === 'replay_event' ? 'replay' : type || 'error';
+}
+
 /**
  * Verifies that return value of configured `beforeSend` or `beforeSendTransaction` is of expected type, and returns the value if so.
  */

packages/core/test/lib/client.test.ts

@@ -15,6 +15,7 @@ import {
 import * as integrationModule from '../../src/integration';
 import { _INTERNAL_captureLog } from '../../src/logs/internal';
 import { _INTERNAL_captureMetric } from '../../src/metrics/internal';
+import { DEFAULT_TRANSPORT_BUFFER_SIZE } from '../../src/transports/base';
 import type { Envelope } from '../../src/types-hoist/envelope';
 import type { ErrorEvent, Event, TransactionEvent } from '../../src/types-hoist/event';
 import type { SpanJSON } from '../../src/types-hoist/span';
@@ -23,7 +24,7 @@ import * as miscModule from '../../src/utils/misc';
 import * as stringModule from '../../src/utils/string';
 import * as timeModule from '../../src/utils/time';
 import { getDefaultTestClientOptions, TestClient } from '../mocks/client';
-import { AdHocIntegration, TestIntegration } from '../mocks/integration';
+import { AdHocIntegration, AsyncTestIntegration, TestIntegration } from '../mocks/integration';
 import { makeFakeTransport } from '../mocks/transport';
 import { clearGlobalScope } from '../testutils';
 
@@ -2935,4 +2936,66 @@ describe('Client', () => {
       expect(sendEnvelopeSpy).toHaveBeenCalledTimes(1);
     });
   });
+
+  describe('promise buffer usage', () => {
+    it('respects the default value of the buffer size', async () => {
+      const options = getDefaultTestClientOptions({ dsn: PUBLIC_DSN });
+      const client = new TestClient(options);
+
+      client.addIntegration(new AsyncTestIntegration());
+
+      Array.from({ length: DEFAULT_TRANSPORT_BUFFER_SIZE + 1 }).forEach(() => {
+        client.captureException(new Error('ʕノ•ᴥ•ʔノ ︵ ┻━┻'));
+      });
+
+      expect(client._clearOutcomes()).toEqual([{ reason: 'queue_overflow', category: 'error', quantity: 1 }]);
+    });
+
+    it('records queue_overflow when promise buffer is full', async () => {
+      const options = getDefaultTestClientOptions({ dsn: PUBLIC_DSN, transportOptions: { bufferSize: 1 } });
+      const client = new TestClient(options);
+
+      client.addIntegration(new AsyncTestIntegration());
+
+      client.captureException(new Error('first'));
+      client.captureException(new Error('second'));
+      client.captureException(new Error('third'));
+
+      expect(client._clearOutcomes()).toEqual([{ reason: 'queue_overflow', category: 'error', quantity: 2 }]);
+    });
+
+    it('records different types of dropped events', async () => {
+      const options = getDefaultTestClientOptions({ dsn: PUBLIC_DSN, transportOptions: { bufferSize: 1 } });
+      const client = new TestClient(options);
+
+      client.addIntegration(new AsyncTestIntegration());
+
+      client.captureException(new Error('first')); // error
+      client.captureException(new Error('second')); // error
+      client.captureMessage('third'); // unknown
+      client.captureEvent({ message: 'fourth' }); // error
+      client.captureEvent({ message: 'fifth', type: 'replay_event' }); // replay
+      client.captureEvent({ message: 'sixth', type: 'transaction' }); // transaction
+
+      expect(client._clearOutcomes()).toEqual([
+        { reason: 'queue_overflow', category: 'error', quantity: 2 },
+        { reason: 'queue_overflow', category: 'unknown', quantity: 1 },
+        { reason: 'queue_overflow', category: 'replay', quantity: 1 },
+        { reason: 'queue_overflow', category: 'transaction', quantity: 1 },
+      ]);
+    });
+
+    it('should skip the promise buffer with sync integrations', async () => {
+      const options = getDefaultTestClientOptions({ dsn: PUBLIC_DSN, transportOptions: { bufferSize: 1 } });
+      const client = new TestClient(options);
+
+      client.addIntegration(new TestIntegration());
+
+      client.captureException(new Error('first'));
+      client.captureException(new Error('second'));
+      client.captureException(new Error('third'));
+
+      expect(client._clearOutcomes()).toEqual([]);
+    });
+  });
 });

packages/core/test/mocks/integration.ts

@@ -24,6 +24,16 @@ export class TestIntegration implements Integration {
   }
 }
 
+export class AsyncTestIntegration implements Integration {
+  public static id: string = 'AsyncTestIntegration';
+
+  public name: string = 'AsyncTestIntegration';
+
+  processEvent(event: Event): Event | null | PromiseLike<Event | null> {
+    return new Promise(resolve => setTimeout(() => resolve(event), 1));
+  }
+}
+
 export class AddAttachmentTestIntegration implements Integration {
   public static id: string = 'AddAttachmentTestIntegration';