feat(replay): Change stop() to flush and remove current session (#7741)

`stop()` will now flush the eventBuffer before clearing it, as well as removing the session from Session Storage. Due to the flushing, `stop()` is now async.

Closes: #7738

Author: billyvg

packages/browser-integration-tests/suites/replay/dsc/test.ts

@@ -54,11 +54,19 @@ sentryTest(
       sentryTest.skip();
     }
 
+    await page.route('https://dsn.ingest.sentry.io/**/*', route => {
+      return route.fulfill({
+        status: 200,
+        contentType: 'application/json',
+        body: JSON.stringify({ id: 'test-id' }),
+      });
+    });
+
     const url = await getLocalTestPath({ testDir: __dirname });
     await page.goto(url);
 
-    await page.evaluate(() => {
-      (window as unknown as TestWindow).Replay.stop();
+    await page.evaluate(async () => {
+      await (window as unknown as TestWindow).Replay.stop();
 
       (window as unknown as TestWindow).Sentry.configureScope(scope => {
         scope.setUser({ id: 'user123', segment: 'segmentB' });

packages/replay/src/integration.ts

@@ -219,12 +219,12 @@ Sentry.init({ replaysOnErrorSampleRate: ${errorSampleRate} })`,
    * Currently, this needs to be manually called (e.g. for tests). Sentry SDK
    * does not support a teardown
    */
-  public stop(): void {
+  public stop(): Promise<void> {
     if (!this._replay) {
-      return;
+      return Promise.resolve();
     }
 
-    this._replay.stop();
+    return this._replay.stop();
   }
 
   /**
@@ -234,9 +234,9 @@ Sentry.init({ replaysOnErrorSampleRate: ${errorSampleRate} })`,
    * Unless `continueRecording` is false, the replay will continue to record and
    * behave as a "session"-based replay.
    */
-  public flush(options?: SendBufferedReplayOptions): Promise<void> | void {
+  public flush(options?: SendBufferedReplayOptions): Promise<void> {
     if (!this._replay || !this._replay.isEnabled()) {
-      return;
+      return Promise.resolve();
     }
 
     return this._replay.sendBufferedReplayOrFlush(options);

packages/replay/src/replay.ts

@@ -7,6 +7,7 @@ import { logger } from '@sentry/utils';
 import { ERROR_CHECKOUT_TIME, MAX_SESSION_LIFE, SESSION_IDLE_DURATION, WINDOW } from './constants';
 import { setupPerformanceObserver } from './coreHandlers/performanceObserver';
 import { createEventBuffer } from './eventBuffer';
+import { clearSession } from './session/clearSession';
 import { getSession } from './session/getSession';
 import { saveSession } from './session/saveSession';
 import type {
@@ -239,7 +240,7 @@ export class ReplayContainer implements ReplayContainerInterface {
    * Currently, this needs to be manually called (e.g. for tests). Sentry SDK
    * does not support a teardown
    */
-  public stop(reason?: string): void {
+  public async stop(reason?: string): Promise<void> {
     if (!this._isEnabled) {
       return;
     }
@@ -255,12 +256,24 @@ export class ReplayContainer implements ReplayContainerInterface {
         log(msg);
       }
 
+      // We can't move `_isEnabled` after awaiting a flush, otherwise we can
+      // enter into an infinite loop when `stop()` is called while flushing.
       this._isEnabled = false;
       this._removeListeners();
       this.stopRecording();
+
+      this._debouncedFlush.cancel();
+      // See comment above re: `_isEnabled`, we "force" a flush, ignoring the
+      // `_isEnabled` state of the plugin since it was disabled above.
+      await this._flush({ force: true });
+
+      // After flush, destroy event buffer
       this.eventBuffer && this.eventBuffer.destroy();
       this.eventBuffer = null;
-      this._debouncedFlush.cancel();
+
+      // Clear session from session storage, note this means if a new session
+      // is started after, it will not have `previousSessionId`
+      clearSession(this);
     } catch (err) {
       this._handleException(err);
     }
@@ -500,7 +513,7 @@ export class ReplayContainer implements ReplayContainerInterface {
     this.session = session;
 
     if (!this.session.sampled) {
-      this.stop('session unsampled');
+      void this.stop('session unsampled');
       return false;
     }
 
@@ -793,7 +806,7 @@ export class ReplayContainer implements ReplayContainerInterface {
       // This means we retried 3 times and all of them failed,
       // or we ran into a problem we don't want to retry, like rate limiting.
       // In this case, we want to completely stop the replay - otherwise, we may get inconsistent segments
-      this.stop('sendReplay');
+      void this.stop('sendReplay');
 
       const client = getCurrentHub().getClient();
 
@@ -807,8 +820,17 @@ export class ReplayContainer implements ReplayContainerInterface {
    * Flush recording data to Sentry. Creates a lock so that only a single flush
    * can be active at a time. Do not call this directly.
    */
-  private _flush: () => Promise<void> = async () => {
-    if (!this._isEnabled) {
+  private _flush = async ({
+    force = false,
+  }: {
+    /**
+     * If true, flush while ignoring the `_isEnabled` state of
+     * Replay integration. (By default, flush is noop if integration
+     * is stopped).
+     */
+    force?: boolean;
+  } = {}): Promise<void> => {
+    if (!this._isEnabled && !force) {
       // This can happen if e.g. the replay was stopped because of exceeding the retry limit
       return;
     }

packages/replay/test/utils/clearSession.ts renamed to packages/replay/src/session/clearSession.ts

@@ -1,7 +1,10 @@
 import { REPLAY_SESSION_KEY, WINDOW } from '../../src/constants';
 import type { ReplayContainer } from '../../src/types';
 
-export function clearSession(replay: ReplayContainer) {
+/**
+ * Removes the session from Session Storage and unsets session in replay instance
+ */
+export function clearSession(replay: ReplayContainer): void {
   deleteSession();
   replay.session = undefined;
 }

packages/replay/src/types.ts

@@ -472,7 +472,7 @@ export interface ReplayContainer {
   isPaused(): boolean;
   getContext(): InternalEventContext;
   start(): void;
-  stop(reason?: string): void;
+  stop(reason?: string): Promise<void>;
   pause(): void;
   resume(): void;
   startRecording(): void;

packages/replay/src/util/addEvent.ts

@@ -46,7 +46,7 @@ export async function addEvent(
     return await replay.eventBuffer.addEvent(event, isCheckout);
   } catch (error) {
     __DEBUG_BUILD__ && logger.error(error);
-    replay.stop('addEvent');
+    await replay.stop('addEvent');
 
     const client = getCurrentHub().getClient();
 

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

@@ -9,13 +9,13 @@ import {
   WINDOW,
 } from '../../src/constants';
 import type { ReplayContainer } from '../../src/replay';
+import { clearSession } from '../../src/session/clearSession';
 import { addEvent } from '../../src/util/addEvent';
 import { PerformanceEntryResource } from '../fixtures/performanceEntry/resource';
 import type { RecordMock } from '../index';
 import { BASE_TIMESTAMP } from '../index';
 import { resetSdkMock } from '../mocks/resetSdkMock';
 import type { DomHandler } from '../types';
-import { clearSession } from '../utils/clearSession';
 import { useFakeTimers } from '../utils/use-fake-timers';
 
 useFakeTimers();

packages/replay/test/integration/events.test.ts

@@ -2,12 +2,12 @@ import { getCurrentHub } from '@sentry/core';
 
 import { WINDOW } from '../../src/constants';
 import type { ReplayContainer } from '../../src/replay';
+import { clearSession } from '../../src/session/clearSession';
 import { addEvent } from '../../src/util/addEvent';
 import { PerformanceEntryResource } from '../fixtures/performanceEntry/resource';
 import type { RecordMock } from '../index';
 import { BASE_TIMESTAMP } from '../index';
 import { resetSdkMock } from '../mocks/resetSdkMock';
-import { clearSession } from '../utils/clearSession';
 import { useFakeTimers } from '../utils/use-fake-timers';
 
 useFakeTimers();

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

@@ -2,13 +2,13 @@ import * as SentryUtils from '@sentry/utils';
 
 import { DEFAULT_FLUSH_MIN_DELAY, WINDOW } from '../../src/constants';
 import type { ReplayContainer } from '../../src/replay';
+import { clearSession } from '../../src/session/clearSession';
 import type { EventBuffer } from '../../src/types';
 import * as AddMemoryEntry from '../../src/util/addMemoryEntry';
 import { createPerformanceEntries } from '../../src/util/createPerformanceEntries';
 import { createPerformanceSpans } from '../../src/util/createPerformanceSpans';
 import * as SendReplay from '../../src/util/sendReplay';
 import { BASE_TIMESTAMP, mockRrweb, mockSdk } from '../index';
-import { clearSession } from '../utils/clearSession';
 import { useFakeTimers } from '../utils/use-fake-timers';
 
 useFakeTimers();

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

@@ -3,10 +3,10 @@ import type { Transport } from '@sentry/types';
 
 import { DEFAULT_FLUSH_MIN_DELAY } from '../../src/constants';
 import type { ReplayContainer } from '../../src/replay';
+import { clearSession } from '../../src/session/clearSession';
 import * as SendReplayRequest from '../../src/util/sendReplayRequest';
 import { BASE_TIMESTAMP, mockSdk } from '../index';
 import { mockRrweb } from '../mocks/mockRrweb';
-import { clearSession } from '../utils/clearSession';
 import { useFakeTimers } from '../utils/use-fake-timers';
 
 useFakeTimers();
@@ -86,8 +86,7 @@ describe('Integration | rate-limiting behaviour', () => {
     expect(replay.stop).toHaveBeenCalledTimes(1);
 
     // No user activity to trigger an update
-    expect(replay.session?.lastActivity).toBe(BASE_TIMESTAMP);
-    expect(replay.session?.segmentId).toBe(1);
+    expect(replay.session).toBe(undefined);
 
     // let's simulate the default rate-limit time of inactivity (60secs) and check that we
     // don't do anything in the meantime or after the time has passed

packages/replay/test/integration/sendReplayEvent.test.ts

@@ -4,10 +4,10 @@ import * as SentryUtils from '@sentry/utils';
 
 import { DEFAULT_FLUSH_MIN_DELAY, WINDOW } from '../../src/constants';
 import type { ReplayContainer } from '../../src/replay';
+import { clearSession } from '../../src/session/clearSession';
 import { addEvent } from '../../src/util/addEvent';
 import * as SendReplayRequest from '../../src/util/sendReplayRequest';
 import { BASE_TIMESTAMP, mockRrweb, mockSdk } from '../index';
-import { clearSession } from '../utils/clearSession';
 import { useFakeTimers } from '../utils/use-fake-timers';
 
 useFakeTimers();
@@ -396,13 +396,8 @@ describe('Integration | sendReplayEvent', () => {
       'Something bad happened',
     );
 
-    // No activity has occurred, session's last activity should remain the same
-    expect(replay.session?.lastActivity).toBe(BASE_TIMESTAMP);
-
-    // segmentId increases despite error
-    expect(replay.session?.segmentId).toBe(1);
-
-    // Replay should be completely stopped now
+    // Replay has stopped, no session should exist
+    expect(replay.session).toBe(undefined);
     expect(replay.isEnabled()).toBe(false);
 
     // Events are ignored now, because we stopped

packages/replay/test/integration/session.test.ts

@@ -9,13 +9,13 @@ import {
   WINDOW,
 } from '../../src/constants';
 import type { ReplayContainer } from '../../src/replay';
+import { clearSession } from '../../src/session/clearSession';
 import type { Session } from '../../src/types';
 import { addEvent } from '../../src/util/addEvent';
 import { createPerformanceSpans } from '../../src/util/createPerformanceSpans';
 import { BASE_TIMESTAMP } from '../index';
 import type { RecordMock } from '../mocks/mockRrweb';
 import { resetSdkMock } from '../mocks/resetSdkMock';
-import { clearSession } from '../utils/clearSession';
 import { useFakeTimers } from '../utils/use-fake-timers';
 
 useFakeTimers();

packages/replay/test/integration/stop.test.ts

@@ -3,14 +3,16 @@ import * as SentryUtils from '@sentry/utils';
 import type { Replay } from '../../src';
 import { WINDOW } from '../../src/constants';
 import type { ReplayContainer } from '../../src/replay';
+import { clearSession } from '../../src/session/clearSession';
 import { addEvent } from '../../src/util/addEvent';
 // mock functions need to be imported first
 import { BASE_TIMESTAMP, mockRrweb, mockSdk } from '../index';
-import { clearSession } from '../utils/clearSession';
 import { useFakeTimers } from '../utils/use-fake-timers';
 
 useFakeTimers();
 
+type MockRunFlush = jest.MockedFunction<ReplayContainer['_runFlush']>;
+
 describe('Integration | stop', () => {
   let replay: ReplayContainer;
   let integration: Replay;
@@ -20,6 +22,7 @@ describe('Integration | stop', () => {
   const { record: mockRecord } = mockRrweb();
 
   let mockAddInstrumentationHandler: MockAddInstrumentationHandler;
+  let mockRunFlush: MockRunFlush;
 
   beforeAll(async () => {
     jest.setSystemTime(new Date(BASE_TIMESTAMP));
@@ -29,6 +32,10 @@ describe('Integration | stop', () => {
     ) as MockAddInstrumentationHandler;
 
     ({ replay, integration } = await mockSdk());
+
+    // @ts-ignore private API
+    mockRunFlush = jest.spyOn(replay, '_runFlush');
+
     jest.runAllTimers();
   });
 
@@ -68,9 +75,10 @@ describe('Integration | stop', () => {
     // Not sure where the 20ms comes from tbh
     const EXTRA_TICKS = 20;
     const TEST_EVENT = { data: {}, timestamp: BASE_TIMESTAMP, type: 3 };
+    const previousSessionId = replay.session?.id;
 
     // stop replays
-    integration.stop();
+    await integration.stop();
 
     // Pretend 5 seconds have passed
     jest.advanceTimersByTime(ELAPSED);
@@ -80,14 +88,17 @@ describe('Integration | stop', () => {
     await new Promise(process.nextTick);
     expect(mockRecord.takeFullSnapshot).not.toHaveBeenCalled();
     expect(replay).not.toHaveLastSentReplay();
-    // Session's last activity should not be updated
-    expect(replay.session?.lastActivity).toEqual(BASE_TIMESTAMP);
+    // Session's does not exist
+    expect(replay.session).toEqual(undefined);
     // eventBuffer is destroyed
     expect(replay.eventBuffer).toBe(null);
 
     // re-enable replay
     integration.start();
 
+    // will be different session
+    expect(replay.session?.id).not.toEqual(previousSessionId);
+
     jest.advanceTimersByTime(ELAPSED);
 
     const timestamp = +new Date(BASE_TIMESTAMP + ELAPSED + ELAPSED + EXTRA_TICKS) / 1000;
@@ -126,27 +137,33 @@ describe('Integration | stop', () => {
     expect(replay.session?.lastActivity).toBe(BASE_TIMESTAMP + ELAPSED + 20);
   });
 
-  it('does not buffer events when stopped', async function () {
-    WINDOW.dispatchEvent(new Event('blur'));
+  it('does not buffer new events after being stopped', async function () {
+    const TEST_EVENT = { data: {}, timestamp: BASE_TIMESTAMP, type: 3 };
+    addEvent(replay, TEST_EVENT);
     expect(replay.eventBuffer?.hasEvents).toBe(true);
+    expect(mockRunFlush).toHaveBeenCalledTimes(0);
 
     // stop replays
-    integration.stop();
+    await integration.stop();
+
+    expect(mockRunFlush).toHaveBeenCalledTimes(1);
 
     expect(replay.eventBuffer).toBe(null);
 
     WINDOW.dispatchEvent(new Event('blur'));
     await new Promise(process.nextTick);
 
     expect(replay.eventBuffer).toBe(null);
-    expect(replay).not.toHaveLastSentReplay();
+    expect(replay).toHaveLastSentReplay({
+      recordingData: JSON.stringify([TEST_EVENT]),
+    });
   });
 
   it('does not call core SDK `addInstrumentationHandler` after initial setup', async function () {
     // NOTE: We clear addInstrumentationHandler mock after every test
-    integration.stop();
+    await integration.stop();
     integration.start();
-    integration.stop();
+    await integration.stop();
     integration.start();
 
     expect(mockAddInstrumentationHandler).not.toHaveBeenCalled();

packages/replay/test/utils/setupReplayContainer.ts

@@ -1,7 +1,7 @@
 import { createEventBuffer } from '../../src/eventBuffer';
 import { ReplayContainer } from '../../src/replay';
+import { clearSession } from '../../src/session/clearSession';
 import type { RecordingOptions, ReplayPluginOptions } from '../../src/types';
-import { clearSession } from './clearSession';
 
 export function setupReplayContainer({
   options,