ref(replay): Pause recording when replay requests are rate-limited

Author: Lms24

packages/replay/src/replay.ts

@@ -1,7 +1,14 @@
 /* eslint-disable max-lines */ // TODO: We might want to split this file up
 import { addGlobalEventProcessor, captureException, getCurrentHub, setContext } from '@sentry/core';
 import type { Breadcrumb, ReplayEvent, ReplayRecordingMode, TransportMakeRequestResponse } from '@sentry/types';
-import { addInstrumentationHandler, logger } from '@sentry/utils';
+import {
+  addInstrumentationHandler,
+  disabledUntil,
+  isRateLimited,
+  logger,
+  RateLimits,
+  updateRateLimits,
+} from '@sentry/utils';
 import { EventType, record } from 'rrweb';
 
 import {
@@ -127,6 +134,11 @@ export class ReplayContainer implements ReplayContainerInterface {
     initialUrl: '',
   };
 
+  /**
+   * A RateLimits object holding the rate-limit durations in case a sent replay event was rate-limited.
+   */
+  private _rateLimits: RateLimits = {};
+
   constructor({ options, recordingOptions }: { options: ReplayPluginOptions; recordingOptions: RecordingOptions }) {
     this._recordingOptions = recordingOptions;
     this._options = options;
@@ -984,7 +996,15 @@ export class ReplayContainer implements ReplayContainerInterface {
     const envelope = createReplayEnvelope(replayEvent, recordingData, dsn, client.getOptions().tunnel);
 
     try {
-      return await transport.send(envelope);
+      const response = await transport.send(envelope);
+      // TODO (v8): we can remove this guard once transport.end's type signature doesn't include void anymore
+      if (response) {
+        this._rateLimits = updateRateLimits(this._rateLimits, response);
+        if (isRateLimited(this._rateLimits, 'replay_event') || isRateLimited(this._rateLimits, 'replay_recording')) {
+          this._handleRateLimit();
+        }
+      }
+      return response;
     } catch {
       throw new Error(UNABLE_TO_SEND_REPLAY);
     }
@@ -1036,9 +1056,8 @@ export class ReplayContainer implements ReplayContainerInterface {
         throw new Error(`${UNABLE_TO_SEND_REPLAY} - max retries exceeded`);
       }
 
-      this._retryCount = this._retryCount + 1;
       // will retry in intervals of 5, 10, 30
-      this._retryInterval = this._retryCount * this._retryInterval;
+      this._retryInterval = ++this._retryCount * this._retryInterval;
 
       return await new Promise((resolve, reject) => {
         setTimeout(async () => {
@@ -1065,4 +1084,28 @@ export class ReplayContainer implements ReplayContainerInterface {
       saveSession(this.session);
     }
   }
+
+  /**
+   * Pauses the replay and resumes it after the rate-limit duration is over.
+   */
+  private _handleRateLimit(): void {
+    const rateLimitEnd = Math.max(
+      disabledUntil(this._rateLimits, 'replay_event'),
+      disabledUntil(this._rateLimits, 'replay_recording'),
+    );
+
+    const rateLimitDuration = rateLimitEnd - Date.now();
+
+    if (rateLimitDuration > 0) {
+      __DEBUG_BUILD__ && logger.warn('[Replay]', `Rate limit hit, pausing replay for ${rateLimitDuration}ms`);
+      this.pause();
+      this._debouncedFlush && this._debouncedFlush.cancel();
+
+      setTimeout(() => {
+        __DEBUG_BUILD__ && logger.info('[Replay]', 'Resuming replay after rate limit');
+        this.resume();
+        this._debouncedFlush();
+      }, rateLimitDuration);
+    }
+  }
 }

packages/replay/test/integration/rateLimiting.test.ts

@@ -0,0 +1,154 @@
+import { Transport, TransportMakeRequestResponse } from '@sentry/types';
+import { ReplayContainer } from '../../src/replay';
+import { BASE_TIMESTAMP, mockSdk } from '../index';
+import { mockRrweb } from '../mocks/mockRrweb';
+import { useFakeTimers } from '../utils/use-fake-timers';
+import { getCurrentHub } from '@sentry/core';
+import { clearSession } from '../utils/clearSession';
+import { DEFAULT_FLUSH_MIN_DELAY, SESSION_IDLE_DURATION } from '../../src/constants';
+
+useFakeTimers();
+
+async function advanceTimers(time: number) {
+  jest.advanceTimersByTime(time);
+  await new Promise(process.nextTick);
+}
+
+type MockTransportSend = jest.MockedFunction<Transport['send']>;
+type MockSendReplayRequest = jest.MockedFunction<ReplayContainer['sendReplayRequest']>;
+
+describe('Integration | rate-limiting behaviour', () => {
+  let replay: ReplayContainer;
+  let mockTransportSend: MockTransportSend;
+  let mockSendReplayRequest: MockSendReplayRequest;
+  const { record: mockRecord } = mockRrweb();
+
+  beforeAll(async () => {
+    jest.setSystemTime(new Date(BASE_TIMESTAMP));
+
+    ({ replay } = await mockSdk({
+      replayOptions: {
+        stickySession: false,
+      },
+    }));
+
+    jest.spyOn(replay, 'sendReplayRequest');
+
+    jest.runAllTimers();
+    mockTransportSend = getCurrentHub()?.getClient()?.getTransport()?.send as MockTransportSend;
+    mockSendReplayRequest = replay.sendReplayRequest as MockSendReplayRequest;
+  });
+
+  beforeEach(() => {
+    jest.setSystemTime(new Date(BASE_TIMESTAMP));
+    mockRecord.takeFullSnapshot.mockClear();
+    mockTransportSend.mockClear();
+
+    // Create a new session and clear mocks because a segment (from initial
+    // checkout) will have already been uploaded by the time the tests run
+    clearSession(replay);
+    replay.loadSession({ expiry: 0 });
+
+    mockSendReplayRequest.mockClear();
+  });
+
+  afterEach(async () => {
+    jest.runAllTimers();
+    await new Promise(process.nextTick);
+    jest.setSystemTime(new Date(BASE_TIMESTAMP));
+    clearSession(replay);
+    replay.loadSession({ expiry: SESSION_IDLE_DURATION });
+  });
+
+  afterAll(() => {
+    replay && replay.stop();
+  });
+
+  it('pauses recording and flushing a rate limit is hit and resumes both after the rate limit duration is over', async () => {
+    expect(replay.session?.segmentId).toBe(0);
+    jest.spyOn(replay, 'sendReplay');
+    jest.spyOn(replay, 'pause');
+    jest.spyOn(replay, 'resume');
+    // @ts-ignore private API
+    jest.spyOn(replay, '_handleRateLimit');
+
+    const TEST_EVENT = { data: {}, timestamp: BASE_TIMESTAMP, type: 2 };
+
+    mockTransportSend.mockImplementationOnce(() => {
+      return Promise.resolve({
+        statusCode: 429,
+        headers: {
+          'x-sentry-rate-limits': null,
+          'retry-after': `30`,
+        },
+      } as TransportMakeRequestResponse);
+    });
+
+    mockRecord._emitter(TEST_EVENT);
+
+    // T = base + 5
+    await advanceTimers(DEFAULT_FLUSH_MIN_DELAY);
+
+    expect(mockRecord.takeFullSnapshot).not.toHaveBeenCalled();
+    expect(mockTransportSend).toHaveBeenCalledTimes(1);
+    expect(replay).toHaveLastSentReplay({ events: JSON.stringify([TEST_EVENT]) });
+
+    expect(replay['_handleRateLimit']).toHaveBeenCalledTimes(1);
+    // resume() was called once before we even started
+    expect(replay.resume).not.toHaveBeenCalled();
+    expect(replay.pause).toHaveBeenCalledTimes(1);
+
+    // No user activity to trigger an update
+    expect(replay.session?.lastActivity).toBe(BASE_TIMESTAMP);
+    expect(replay.session?.segmentId).toBe(1);
+
+    // let's simulate the rate-limit time of inactivity (30secs) and check that we don't do anything in the meantime
+    const TEST_EVENT2 = { data: {}, timestamp: BASE_TIMESTAMP + DEFAULT_FLUSH_MIN_DELAY, type: 3 };
+    for (let i = 0; i < 5; i++) {
+      const ev = {
+        ...TEST_EVENT2,
+        timestamp: BASE_TIMESTAMP + DEFAULT_FLUSH_MIN_DELAY * (i + 1),
+      };
+      mockRecord._emitter(ev);
+      await advanceTimers(DEFAULT_FLUSH_MIN_DELAY);
+      expect(replay.isPaused()).toBe(true);
+      expect(replay.sendReplay).toHaveBeenCalledTimes(1);
+      expect(mockTransportSend).toHaveBeenCalledTimes(1);
+    }
+
+    // T = base + 35
+    await advanceTimers(DEFAULT_FLUSH_MIN_DELAY);
+
+    // now, recording should resume and first, we expect a checkout event to be sent, as resume()
+    // should trigger a full snapshot
+    expect(replay.resume).toHaveBeenCalledTimes(1);
+    expect(replay.isPaused()).toBe(false);
+
+    expect(replay.sendReplay).toHaveBeenCalledTimes(2);
+    expect(replay).toHaveLastSentReplay({
+      events: '[{"data":{"isCheckout":true},"timestamp":1580598035000,"type":2}]',
+    });
+
+    // and let's also emit a new event and check that it is recorded
+    const TEST_EVENT3 = {
+      data: {},
+      timestamp: BASE_TIMESTAMP + 7 * DEFAULT_FLUSH_MIN_DELAY,
+      type: 3,
+    };
+    mockRecord._emitter(TEST_EVENT3);
+
+    // T = base + 40
+    await advanceTimers(DEFAULT_FLUSH_MIN_DELAY);
+    expect(replay.sendReplay).toHaveBeenCalledTimes(3);
+    expect(replay).toHaveLastSentReplay({ events: JSON.stringify([TEST_EVENT3]) });
+
+    // nothing should happen afterwards
+    // T = base + 60
+    await advanceTimers(20_000);
+    expect(replay.sendReplay).toHaveBeenCalledTimes(3);
+    expect(replay).toHaveLastSentReplay({ events: JSON.stringify([TEST_EVENT3]) });
+
+    // events array should be empty
+    expect(replay.eventBuffer?.length).toBe(0);
+  });
+});