fix(replay): Ensure handleRecordingEmit aborts when event is not added (#8938)

I noticed that in `handleRecordingEmit`, due to the async nature of
`addEvent` it could happen that an event is actually not added (because
it is discarded due to timestamp etc.). But we are still updating the
initial session timestamp (`[Replay] Updating session start time to
earliest event in buffer to...`), as we don't actually abort there.

This PR changes this to actually abort `handleRecordingEmit` in this
case. I added an `addEventSync` method for this that just returns
true/false instead of a promise, which should not change anything there
as we haven't been waiting for the result of the promise anyhow.

Author: mydea

packages/replay/src/replay.ts

@@ -38,7 +38,7 @@ import type {
   Timeouts,
 } from './types';
 import { ReplayEventTypeCustom } from './types';
-import { addEvent } from './util/addEvent';
+import { addEvent, addEventSync } from './util/addEvent';
 import { addGlobalListeners } from './util/addGlobalListeners';
 import { addMemoryEntry } from './util/addMemoryEntry';
 import { createBreadcrumb } from './util/createBreadcrumb';
@@ -666,7 +666,8 @@ export class ReplayContainer implements ReplayContainerInterface {
       });
 
       this.addUpdate(() => {
-        void addEvent(this, {
+        // Return `false` if the event _was_ added, as that means we schedule a flush
+        return !addEventSync(this, {
           type: ReplayEventTypeCustom,
           timestamp: breadcrumb.timestamp || 0,
           data: {

packages/replay/src/util/addEvent.ts

@@ -13,39 +13,46 @@ function isCustomEvent(event: RecordingEvent): event is ReplayFrameEvent {
 
 /**
  * Add an event to the event buffer.
+ * In contrast to `addEvent`, this does not return a promise & does not wait for the adding of the event to succeed/fail.
+ * Instead this returns `true` if we tried to add the event, else false.
+ * It returns `false` e.g. if we are paused, disabled, or out of the max replay duration.
+ *
  * `isCheckout` is true if this is either the very first event, or an event triggered by `checkoutEveryNms`.
  */
-export async function addEvent(
+export function addEventSync(replay: ReplayContainer, event: RecordingEvent, isCheckout?: boolean): boolean {
+  if (!shouldAddEvent(replay, event)) {
+    return false;
+  }
+
+  void _addEvent(replay, event, isCheckout);
+
+  return true;
+}
+
+/**
+ * Add an event to the event buffer.
+ * Resolves to `null` if no event was added, else to `void`.
+ *
+ * `isCheckout` is true if this is either the very first event, or an event triggered by `checkoutEveryNms`.
+ */
+export function addEvent(
   replay: ReplayContainer,
   event: RecordingEvent,
   isCheckout?: boolean,
 ): Promise<AddEventResult | null> {
-  if (!replay.eventBuffer) {
-    // This implies that `_isEnabled` is false
-    return null;
-  }
-
-  if (replay.isPaused()) {
-    // Do not add to event buffer when recording is paused
-    return null;
+  if (!shouldAddEvent(replay, event)) {
+    return Promise.resolve(null);
   }
 
-  const timestampInMs = timestampToMs(event.timestamp);
-
-  // Throw out events that happen more than 5 minutes ago. This can happen if
-  // page has been left open and idle for a long period of time and user
-  // comes back to trigger a new session. The performance entries rely on
-  // `performance.timeOrigin`, which is when the page first opened.
-  if (timestampInMs + replay.timeouts.sessionIdlePause < Date.now()) {
-    return null;
-  }
+  return _addEvent(replay, event, isCheckout);
+}
 
-  // Throw out events that are +60min from the initial timestamp
-  if (timestampInMs > replay.getContext().initialTimestamp + replay.getOptions().maxReplayDuration) {
-    logInfo(
-      `[Replay] Skipping event with timestamp ${timestampInMs} because it is after maxReplayDuration`,
-      replay.getOptions()._experiments.traceInternals,
-    );
+async function _addEvent(
+  replay: ReplayContainer,
+  event: RecordingEvent,
+  isCheckout?: boolean,
+): Promise<AddEventResult | null> {
+  if (!replay.eventBuffer) {
     return null;
   }
 
@@ -81,6 +88,34 @@ export async function addEvent(
   }
 }
 
+/** Exported only for tests. */
+export function shouldAddEvent(replay: ReplayContainer, event: RecordingEvent): boolean {
+  if (!replay.eventBuffer || replay.isPaused() || !replay.isEnabled()) {
+    return false;
+  }
+
+  const timestampInMs = timestampToMs(event.timestamp);
+
+  // Throw out events that happen more than 5 minutes ago. This can happen if
+  // page has been left open and idle for a long period of time and user
+  // comes back to trigger a new session. The performance entries rely on
+  // `performance.timeOrigin`, which is when the page first opened.
+  if (timestampInMs + replay.timeouts.sessionIdlePause < Date.now()) {
+    return false;
+  }
+
+  // Throw out events that are +60min from the initial timestamp
+  if (timestampInMs > replay.getContext().initialTimestamp + replay.getOptions().maxReplayDuration) {
+    logInfo(
+      `[Replay] Skipping event with timestamp ${timestampInMs} because it is after maxReplayDuration`,
+      replay.getOptions()._experiments.traceInternals,
+    );
+    return false;
+  }
+
+  return true;
+}
+
 function maybeApplyCallback(
   event: RecordingEvent,
   callback: ReplayPluginOptions['beforeAddRecordingEvent'],

packages/replay/src/util/handleRecordingEmit.ts

@@ -2,8 +2,8 @@ import { EventType } from '@sentry-internal/rrweb';
 import { logger } from '@sentry/utils';
 
 import { saveSession } from '../session/saveSession';
-import type { AddEventResult, RecordingEvent, ReplayContainer, ReplayOptionFrameEvent } from '../types';
-import { addEvent } from './addEvent';
+import type { RecordingEvent, ReplayContainer, ReplayOptionFrameEvent } from '../types';
+import { addEventSync } from './addEvent';
 import { logInfo } from './log';
 
 type RecordingEmitCallback = (event: RecordingEvent, isCheckout?: boolean) => void;
@@ -40,9 +40,12 @@ export function getHandleRecordingEmit(replay: ReplayContainer): RecordingEmitCa
         replay.setInitialState();
       }
 
-      // We need to clear existing events on a checkout, otherwise they are
-      // incremental event updates and should be appended
-      void addEvent(replay, event, isCheckout);
+      // If the event is not added (e.g. due to being paused, disabled, or out of the max replay duration),
+      // Skip all further steps
+      if (!addEventSync(replay, event, isCheckout)) {
+        // Return true to skip scheduling a debounced flush
+        return true;
+      }
 
       // Different behavior for full snapshots (type=2), ignore other event types
       // See https://github.com/rrweb-io/rrweb/blob/d8f9290ca496712aa1e7d472549480c4e7876594/packages/rrweb/src/types.ts#L16
@@ -56,7 +59,7 @@ export function getHandleRecordingEmit(replay: ReplayContainer): RecordingEmitCa
       //
       // `isCheckout` is always true, but want to be explicit that it should
       // only be added for checkouts
-      void addSettingsEvent(replay, isCheckout);
+      addSettingsEvent(replay, isCheckout);
 
       // If there is a previousSessionId after a full snapshot occurs, then
       // the replay session was started due to session expiration. The new session
@@ -130,11 +133,11 @@ export function createOptionsEvent(replay: ReplayContainer): ReplayOptionFrameEv
  * Add a "meta" event that contains a simplified view on current configuration
  * options. This should only be included on the first segment of a recording.
  */
-function addSettingsEvent(replay: ReplayContainer, isCheckout?: boolean): Promise<AddEventResult | null> {
+function addSettingsEvent(replay: ReplayContainer, isCheckout?: boolean): void {
   // Only need to add this event when sending the first segment
   if (!isCheckout || !replay.session || replay.session.segmentId !== 0) {
-    return Promise.resolve(null);
+    return;
   }
 
-  return addEvent(replay, createOptionsEvent(replay), false);
+  addEventSync(replay, createOptionsEvent(replay), false);
 }

packages/replay/test/integration/errorSampleRate.test.ts

@@ -810,76 +810,6 @@ describe('Integration | errorSampleRate', () => {
       expect(replay).toHaveLastSentReplay();
     });
 
-    it('does not refresh replay based on earliest event in buffer', async () => {
-      jest.setSystemTime(BASE_TIMESTAMP);
-
-      const TEST_EVENT = getTestEventIncremental({ timestamp: BASE_TIMESTAMP - 60000 });
-      mockRecord._emitter(TEST_EVENT);
-
-      expect(mockRecord.takeFullSnapshot).not.toHaveBeenCalled();
-      expect(replay).not.toHaveLastSentReplay();
-
-      jest.runAllTimers();
-      await new Promise(process.nextTick);
-
-      expect(replay).not.toHaveLastSentReplay();
-      captureException(new Error('testing'));
-
-      await waitForBufferFlush();
-
-      expect(replay).toHaveLastSentReplay();
-
-      // Flush from calling `stopRecording`
-      await waitForFlush();
-
-      // Now wait after session expires - should stop recording
-      mockRecord.takeFullSnapshot.mockClear();
-      (getCurrentHub().getClient()!.getTransport()!.send as unknown as jest.SpyInstance<any>).mockClear();
-
-      expect(replay).not.toHaveLastSentReplay();
-
-      const TICKS = 80;
-
-      // We advance time so that we are on the border of expiring, taking into
-      // account that TEST_EVENT timestamp is 60000 ms before BASE_TIMESTAMP. The
-      // 3 DEFAULT_FLUSH_MIN_DELAY is to account for the `waitForFlush` that has
-      // happened, and for the next two that will happen. The first following
-      // `waitForFlush` does not expire session, but the following one will.
-      jest.advanceTimersByTime(SESSION_IDLE_EXPIRE_DURATION - 60000 - 3 * DEFAULT_FLUSH_MIN_DELAY - TICKS);
-      await new Promise(process.nextTick);
-
-      mockRecord._emitter(TEST_EVENT);
-      expect(replay).not.toHaveLastSentReplay();
-      await waitForFlush();
-
-      expect(replay).not.toHaveLastSentReplay();
-      expect(mockRecord.takeFullSnapshot).toHaveBeenCalledTimes(0);
-      expect(replay.isEnabled()).toBe(true);
-
-      mockRecord._emitter(TEST_EVENT);
-      expect(replay).not.toHaveLastSentReplay();
-      await waitForFlush();
-
-      expect(replay).not.toHaveLastSentReplay();
-      expect(mockRecord.takeFullSnapshot).toHaveBeenCalledTimes(0);
-      expect(replay.isEnabled()).toBe(true);
-
-      const sessionId = replay.getSessionId();
-
-      // It's hard to test, but if we advance the below time less 1 ms, it should
-      // be enabled, but we can't trigger a session check via flush without
-      // incurring another DEFAULT_FLUSH_MIN_DELAY timeout.
-      jest.advanceTimersByTime(60000 - DEFAULT_FLUSH_MIN_DELAY);
-      mockRecord._emitter(TEST_EVENT);
-      expect(replay).not.toHaveLastSentReplay();
-      await waitForFlush();
-
-      expect(replay).not.toHaveLastSentReplay();
-      expect(mockRecord.takeFullSnapshot).toHaveBeenCalledTimes(0);
-      expect(replay.isEnabled()).toBe(true);
-      expect(replay.getSessionId()).not.toBe(sessionId);
-    });
-
     it('handles very long active buffer session', async () => {
       const stepDuration = 10_000;
       const steps = 5_000;

packages/replay/test/integration/flush.test.ts

@@ -403,6 +403,8 @@ describe('Integration | flush', () => {
   it('logs warning if adding event that is after maxReplayDuration', async () => {
     replay.getOptions()._experiments.traceInternals = true;
 
+    const spyLogger = jest.spyOn(SentryUtils.logger, 'info');
+
     sessionStorage.clear();
     clearSession(replay);
     replay['_initializeSessionForSampling']();
@@ -422,32 +424,18 @@ describe('Integration | flush', () => {
     // no checkout!
     await advanceTimers(DEFAULT_FLUSH_MIN_DELAY);
 
-    expect(mockFlush).toHaveBeenCalledTimes(1);
-    expect(mockSendReplay).toHaveBeenCalledTimes(1);
-
-    const replayData = mockSendReplay.mock.calls[0][0];
+    // No flush is scheduled is aborted because event is after maxReplayDuration
+    expect(mockFlush).toHaveBeenCalledTimes(0);
+    expect(mockSendReplay).toHaveBeenCalledTimes(0);
 
-    expect(JSON.parse(replayData.recordingData)).toEqual([
-      {
-        type: 5,
-        timestamp: BASE_TIMESTAMP,
-        data: {
-          tag: 'breadcrumb',
-          payload: {
-            timestamp: BASE_TIMESTAMP / 1000,
-            type: 'default',
-            category: 'console',
-            data: { logger: 'replay' },
-            level: 'info',
-            message: `[Replay] Skipping event with timestamp ${
-              BASE_TIMESTAMP + MAX_REPLAY_DURATION + 100
-            } because it is after maxReplayDuration`,
-          },
-        },
-      },
-    ]);
+    expect(spyLogger).toHaveBeenLastCalledWith(
+      `[Replay] Skipping event with timestamp ${
+        BASE_TIMESTAMP + MAX_REPLAY_DURATION + 100
+      } because it is after maxReplayDuration`,
+    );
 
     replay.getOptions()._experiments.traceInternals = false;
+    spyLogger.mockRestore();
   });
 
   /**

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

@@ -1,9 +1,9 @@
 import 'jsdom-worker';
 
 import { BASE_TIMESTAMP } from '../..';
-import { REPLAY_MAX_EVENT_BUFFER_SIZE } from '../../../src/constants';
+import { MAX_REPLAY_DURATION, REPLAY_MAX_EVENT_BUFFER_SIZE, SESSION_IDLE_PAUSE_DURATION } from '../../../src/constants';
 import type { EventBufferProxy } from '../../../src/eventBuffer/EventBufferProxy';
-import { addEvent } from '../../../src/util/addEvent';
+import { addEvent, shouldAddEvent } from '../../../src/util/addEvent';
 import { getTestEventIncremental } from '../../utils/getTestEvent';
 import { setupReplayContainer } from '../../utils/setupReplayContainer';
 import { useFakeTimers } from '../../utils/use-fake-timers';
@@ -57,4 +57,65 @@ describe('Unit | util | addEvent', () => {
 
     expect(replay.isEnabled()).toEqual(false);
   });
+
+  describe('shouldAddEvent', () => {
+    beforeEach(() => {
+      jest.setSystemTime(BASE_TIMESTAMP);
+    });
+
+    it('returns true by default', () => {
+      const replay = setupReplayContainer({});
+      const event = getTestEventIncremental({ timestamp: BASE_TIMESTAMP });
+
+      expect(shouldAddEvent(replay, event)).toEqual(true);
+    });
+
+    it('returns false when paused', () => {
+      const replay = setupReplayContainer({});
+      const event = getTestEventIncremental({ timestamp: BASE_TIMESTAMP });
+
+      replay.pause();
+
+      expect(shouldAddEvent(replay, event)).toEqual(false);
+    });
+
+    it('returns false when disabled', async () => {
+      const replay = setupReplayContainer({});
+      const event = getTestEventIncremental({ timestamp: BASE_TIMESTAMP });
+
+      await replay.stop();
+
+      expect(shouldAddEvent(replay, event)).toEqual(false);
+    });
+
+    it('returns false if there is no eventBuffer', () => {
+      const replay = setupReplayContainer({});
+      const event = getTestEventIncremental({ timestamp: BASE_TIMESTAMP });
+
+      replay.eventBuffer = null;
+
+      expect(shouldAddEvent(replay, event)).toEqual(false);
+    });
+
+    it('returns false when event is too old', () => {
+      const replay = setupReplayContainer({});
+      const event = getTestEventIncremental({ timestamp: BASE_TIMESTAMP - SESSION_IDLE_PAUSE_DURATION - 1 });
+
+      expect(shouldAddEvent(replay, event)).toEqual(false);
+    });
+
+    it('returns false if event is too long after initial timestamp', () => {
+      const replay = setupReplayContainer({});
+      const event = getTestEventIncremental({ timestamp: BASE_TIMESTAMP + MAX_REPLAY_DURATION + 1 });
+
+      expect(shouldAddEvent(replay, event)).toEqual(false);
+    });
+
+    it('returns true if event is withing max duration after after initial timestamp', () => {
+      const replay = setupReplayContainer({});
+      const event = getTestEventIncremental({ timestamp: BASE_TIMESTAMP + MAX_REPLAY_DURATION - 1 });
+
+      expect(shouldAddEvent(replay, event)).toEqual(true);
+    });
+  });
 });

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

@@ -16,8 +16,8 @@ describe('Unit | util | handleRecordingEmit', () => {
 
   beforeEach(function () {
     jest.setSystemTime(BASE_TIMESTAMP);
-    addEventMock = jest.spyOn(SentryAddEvent, 'addEvent').mockImplementation(async () => {
-      // Do nothing
+    addEventMock = jest.spyOn(SentryAddEvent, 'addEventSync').mockImplementation(() => {
+      return true;
     });
   });