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/unit/index.test.ts

@@ -1,5 +1,5 @@
 import { getCurrentHub, Hub } from '@sentry/core';
-import { Event, Scope } from '@sentry/types';
+import { Event, Scope, TransportMakeRequestResponse } from '@sentry/types';
 import { EventType } from 'rrweb';
 
 import {
@@ -874,6 +874,96 @@ describe('Replay', () => {
     await advanceTimers(DEFAULT_FLUSH_MIN_DELAY);
     expect(replay.flush).toHaveBeenCalledTimes(1);
   });
+
+  describe('rate-limiting behaviour', () => {
+    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);
+    });
+  });
 });
 
 describe('eventProcessors', () => {