feat(replay): Handle large amounts of consecutive events

packages/replay/src/constants.ts

@@ -31,3 +31,7 @@ export const ERROR_CHECKOUT_TIME = 60_000;
 
 export const RETRY_BASE_INTERVAL = 5000;
 export const RETRY_MAX_COUNT = 3;
+
+// How many events can occur in the given rolling time window before we want to pause & full checkout the replay?
+export const EVENT_ROLLING_WINDOW_TIME = 100;
+export const EVENT_ROLLING_WINDOW_MAX = 2_000;

packages/replay/src/replay.ts

@@ -35,6 +35,7 @@ import { createBreadcrumb } from './util/createBreadcrumb';
 import { createPerformanceEntries } from './util/createPerformanceEntries';
 import { createPerformanceSpans } from './util/createPerformanceSpans';
 import { debounce } from './util/debounce';
+import { EventCounter } from './util/EventCounter';
 import { isExpired } from './util/isExpired';
 import { isSessionExpired } from './util/isSessionExpired';
 import { overwriteRecordDroppedEvent, restoreRecordDroppedEvent } from './util/monkeyPatchRecordDroppedEvent';
@@ -60,6 +61,8 @@ export class ReplayContainer implements ReplayContainerInterface {
    */
   public recordingMode: ReplayRecordingMode = 'session';
 
+  public eventCounter: EventCounter;
+
   /**
    * Options to pass to `rrweb.record()`
    */
@@ -122,6 +125,8 @@ export class ReplayContainer implements ReplayContainerInterface {
     this._debouncedFlush = debounce(() => this._flush(), this._options.flushMinDelay, {
       maxWait: this._options.flushMaxDelay,
     });
+
+    this.eventCounter = new EventCounter();
   }
 
   /** Get the event context. */

packages/replay/src/types.ts

@@ -1,6 +1,7 @@
 import type { ReplayRecordingData, ReplayRecordingMode } from '@sentry/types';
 
 import type { eventWithTime, recordOptions } from './types/rrweb';
+import type { EventCounter } from './util/EventCounter';
 
 export type RecordingEvent = eventWithTime;
 export type RecordingOptions = recordOptions;
@@ -289,6 +290,7 @@ export interface ReplayContainer {
   performanceEvents: AllPerformanceEntry[];
   session: Session | undefined;
   recordingMode: ReplayRecordingMode;
+  eventCounter: EventCounter;
   isEnabled(): boolean;
   isPaused(): boolean;
   getContext(): InternalEventContext;

packages/replay/src/util/EventCounter.ts

@@ -0,0 +1,29 @@
+import { EVENT_ROLLING_WINDOW_MAX, EVENT_ROLLING_WINDOW_TIME } from '../constants';
+
+/** A simple rolling window event counter. */
+export class EventCounter {
+  // How many events happed in a rolling window of 100ms
+  private _count: number;
+  // How long the rolling window is
+  private _time: number;
+  // How many events can happen in the rolling window
+  private _max: number;
+
+  public constructor(time = EVENT_ROLLING_WINDOW_TIME, max = EVENT_ROLLING_WINDOW_MAX) {
+    this._count = 0;
+    this._time = time;
+    this._max = max;
+  }
+
+  /** An event is added. */
+  public add(): void {
+    this._count++;
+
+    setTimeout(() => this._count--, this._time);
+  }
+
+  /** If there are too many events in the rolling window. */
+  public hasExceededLimit(): boolean {
+    return this._count > this._max;
+  }
+}

packages/replay/src/util/addEvent.ts

@@ -1,8 +1,9 @@
 import { getCurrentHub } from '@sentry/core';
 import { logger } from '@sentry/utils';
 
-import { SESSION_IDLE_DURATION } from '../constants';
+import { SESSION_IDLE_DURATION, EVENT_ROLLING_WINDOW_TIME } from '../constants';
 import type { AddEventResult, RecordingEvent, ReplayContainer } from '../types';
+import { EventCounter } from './EventCounter';
 
 /**
  * Add an event to the event buffer
@@ -42,6 +43,19 @@ export async function addEvent(
     replay.getContext().earliestEvent = timestampInMs;
   }
 
+  replay.eventCounter.add();
+
+  // If we exceed the event limit, pause the recording and resume it after the rolling window time
+  // The resuming will trigger a full checkout
+  // This means the user will have a brief gap in their recording, but it's better than freezing the page due to too many events happening at the same time
+  // Afterwards, things will continue as normally
+  if (replay.eventCounter.hasExceededLimit()) {
+    replay.eventCounter = new EventCounter();
+    replay.pause();
+    setTimeout(() => replay.resume(), EVENT_ROLLING_WINDOW_TIME);
+    return;
+  }
+
   try {
     return await replay.eventBuffer.addEvent(event, isCheckout);
   } catch (error) {

packages/replay/test/mocks/mockSdk.ts

@@ -5,6 +5,7 @@ import type { ReplayContainer } from '../../src/replay';
 import type { ReplayConfiguration } from '../../src/types';
 import type { TestClientOptions } from '../utils/TestClient';
 import { getDefaultClientOptions, init } from '../utils/TestClient';
+import { EventCounter } from '../../src/util/EventCounter';
 
 export interface MockSdkParams {
   replayOptions?: ReplayConfiguration;
@@ -84,5 +85,9 @@ export async function mockSdk({ replayOptions, sentryOptions, autoStart = true }
 
   const replay = replayIntegration['_replay']!;
 
+  // In tests, we want to ignore the event counter by default
+  // As it adds timeouts that can interfere with other things
+  replay.eventCounter = new EventCounter(0);
+
   return { replay, integration: replayIntegration };
 }

packages/replay/test/unit/util/addEvent.test.ts

@@ -5,13 +5,17 @@ import type { EventBufferProxy } from '../../../src/eventBuffer/EventBufferProxy
 import { addEvent } from '../../../src/util/addEvent';
 import { setupReplayContainer } from '../../utils/setupReplayContainer';
 import { useFakeTimers } from '../../utils/use-fake-timers';
+import { EventCounter } from '../../../src/util/EventCounter';
+import { EVENT_ROLLING_WINDOW_TIME, EVENT_ROLLING_WINDOW_MAX } from '../../../src/constants';
 
 useFakeTimers();
 
 describe('Unit | util | addEvent', () => {
-  it('stops when encountering a compression error', async function () {
+  beforeEach(function () {
     jest.setSystemTime(BASE_TIMESTAMP);
+  });
 
+  it('stops when encountering a compression error', async function () {
     const replay = setupReplayContainer({
       options: {
         useCompression: true,
@@ -29,4 +33,59 @@ describe('Unit | util | addEvent', () => {
 
     expect(replay.isEnabled()).toEqual(false);
   });
+
+  describe('event count rolling window', () => {
+    it('pauses when triggering too many events', async function () {
+      const replay = setupReplayContainer({});
+      // This is overwritten by defaults for tests, we want to try it with the proper values
+      replay.eventCounter = new EventCounter(EVENT_ROLLING_WINDOW_TIME, EVENT_ROLLING_WINDOW_MAX);
+
+      // Now trigger A LOT of events
+      for (let i = 0; i < EVENT_ROLLING_WINDOW_MAX - 10; i++) {
+        addEvent(replay, { data: {}, timestamp: BASE_TIMESTAMP, type: 2 });
+      }
+      await new Promise(process.nextTick);
+
+      // Nothing should have happend, all still live
+      expect(replay.isPaused()).toEqual(false);
+
+      // now add a few more with a short delay, should still be running
+      for (let i = 0; i < 10; i++) {
+        jest.advanceTimersByTime(5);
+        addEvent(replay, { data: {}, timestamp: BASE_TIMESTAMP + i * 5, type: 2 });
+      }
+      await new Promise(process.nextTick);
+
+      expect(replay.isPaused()).toEqual(false);
+
+      // Now add one more, should trigger the pause
+      addEvent(replay, { data: {}, timestamp: BASE_TIMESTAMP + 90, type: 2 });
+      await new Promise(process.nextTick);
+
+      expect(replay.isPaused()).toEqual(true);
+
+      // Wait for the rolling window to pass, should trigger a resume
+      jest.advanceTimersByTime(EVENT_ROLLING_WINDOW_TIME);
+      await new Promise(process.nextTick);
+
+      expect(replay.isPaused()).toEqual(false);
+    });
+
+    it('throws out event count after rolling window timeout', async function () {
+      const replay = setupReplayContainer({});
+      // This is overwritten by defaults for tests, we want to try it with the proper values
+      replay.eventCounter = new EventCounter(EVENT_ROLLING_WINDOW_TIME, EVENT_ROLLING_WINDOW_MAX);
+
+      // Now trigger A LOT of events
+      for (let i = 0; i < EVENT_ROLLING_WINDOW_MAX * 2; i++) {
+        jest.advanceTimersByTime(1);
+        addEvent(replay, { data: {}, timestamp: BASE_TIMESTAMP + i * 1, type: 2 });
+      }
+      await new Promise(process.nextTick);
+
+      // Nothing should have happend, all still live,
+      // because the events continuously move out of the window
+      expect(replay.isPaused()).toEqual(false);
+    });
+  });
 });