feat(replay): Add non-async flush for page unloads

Add a method of using a non-async flush so that we are able to send a segment when the user unloads the page.

Author: billyvg

packages/replay/src/eventBuffer.ts

@@ -73,14 +73,23 @@ class EventBufferArray implements EventBuffer {
 
   public finish(): Promise<string> {
     return new Promise<string>(resolve => {
-      // Make a copy of the events array reference and immediately clear the
-      // events member so that we do not lose new events while uploading
-      // attachment.
-      const eventsRet = this._events;
-      this._events = [];
-      resolve(JSON.stringify(eventsRet));
+      resolve(
+        this._finish());
     });
   }
+
+  public finishImmediate(): string {
+    return this._finish();
+  }
+
+  private _finish(): string {
+    // Make a copy of the events array reference and immediately clear the
+    // events member so that we do not lose new events while uploading
+    // attachment.
+    const events = this._events;
+    this._events = [];
+    return JSON.stringify(events);
+  }
 }
 
 /**
@@ -158,6 +167,18 @@ export class EventBufferCompressionWorker implements EventBuffer {
     return this._finishRequest(this._getAndIncrementId());
   }
 
+  /**
+   * Finish the event buffer and return the pending events.
+   */
+  public finishImmediate(): string {
+    const events = this._pendingEvents;
+
+    // Ensure worker is still in a good state and disregard the result
+    void this._finishRequest(this._getAndIncrementId());
+
+    return JSON.stringify(events);
+  }
+
   /**
    * Post message to worker and wait for response before resolving promise.
    */

packages/replay/src/integration.ts

@@ -170,7 +170,7 @@ Sentry.init({ replaysOnErrorSampleRate: ${errorSampleRate} })`,
       return;
     }
 
-    this._replay.start();
+    void this._replay.start();
   }
 
   /**

packages/replay/src/replay.ts

@@ -1,6 +1,6 @@
 /* eslint-disable max-lines */ // TODO: We might want to split this file up
 import { addGlobalEventProcessor, captureException, getCurrentHub } from '@sentry/core';
-import type { Breadcrumb, ReplayRecordingMode } from '@sentry/types';
+import type { Breadcrumb, ReplayRecordingMode, ReplayRecordingData } from '@sentry/types';
 import type { RateLimits } from '@sentry/utils';
 import { addInstrumentationHandler, disabledUntil, logger } from '@sentry/utils';
 import { EventType, record } from 'rrweb';
@@ -28,6 +28,7 @@ import type {
   ReplayContainer as ReplayContainerInterface,
   ReplayPluginOptions,
   Session,
+  FlushOptions,
 } from './types';
 import { addEvent } from './util/addEvent';
 import { addMemoryEntry } from './util/addMemoryEntry';
@@ -151,7 +152,7 @@ export class ReplayContainer implements ReplayContainerInterface {
    * Creates or loads a session, attaches listeners to varying events (DOM,
    * _performanceObserver, Recording, Sentry SDK, etc)
    */
-  public start(): void {
+  public async start(): Promise<void> {
     this._setInitialState();
 
     this._loadSession({ expiry: SESSION_IDLE_DURATION });
@@ -324,7 +325,6 @@ export class ReplayContainer implements ReplayContainerInterface {
   }
 
   /**
-   *
    * Always flush via `_debouncedFlush` so that we do not have flushes triggered
    * from calling both `flush` and `_debouncedFlush`. Otherwise, there could be
    * cases of mulitple flushes happening closely together.
@@ -335,7 +335,7 @@ export class ReplayContainer implements ReplayContainerInterface {
     return this._debouncedFlush.flush() as Promise<void>;
   }
 
-  /** Get the current sesion (=replay) ID */
+  /** Get the current session (=replay) ID */
   public getSessionId(): string | undefined {
     return this.session && this.session.id;
   }
@@ -625,7 +625,7 @@ export class ReplayContainer implements ReplayContainerInterface {
     // Send replay when the page/tab becomes hidden. There is no reason to send
     // replay if it becomes visible, since no actions we care about were done
     // while it was hidden
-    this._conditionalFlush();
+    this._conditionalFlush({finishImmediate: true});
   }
 
   /**
@@ -747,11 +747,20 @@ export class ReplayContainer implements ReplayContainerInterface {
   /**
    * Only flush if `this.recordingMode === 'session'`
    */
-  private _conditionalFlush(): void {
+  private _conditionalFlush(options: FlushOptions = {}): void {
     if (this.recordingMode === 'error') {
       return;
     }
 
+    /**
+     * Page is likely to unload so need to bypass debounce completely and
+     * synchronously retrieve pending events from buffer and send request asap.
+     */
+    if (options.finishImmediate) {
+      void this._runFlush(options);
+      return;
+    }
+
     void this.flushImmediate();
   }
 
@@ -795,40 +804,60 @@ export class ReplayContainer implements ReplayContainerInterface {
    *
    * Should never be called directly, only by `flush`
    */
-  private async _runFlush(): Promise<void> {
+  private async _runFlush(options: FlushOptions = {}): Promise<void> {
     if (!this.session || !this.eventBuffer) {
       __DEBUG_BUILD__ && logger.error('[Replay] No session or eventBuffer found to flush.');
       return;
     }
 
-    await this._addPerformanceEntries();
+    try {
+      this._debouncedFlush.cancel();
 
-    // Check eventBuffer again, as it could have been stopped in the meanwhile
-    if (!this.eventBuffer || !this.eventBuffer.pendingLength) {
-      return;
-    }
+      const promises: Promise<any>[] = [];
 
-    // Only attach memory event if eventBuffer is not empty
-    await addMemoryEntry(this);
+      promises.push(this._addPerformanceEntries());
 
-    // Check eventBuffer again, as it could have been stopped in the meanwhile
-    if (!this.eventBuffer) {
-      return;
-    }
+      // Do not continue if there are no pending events in buffer
+      if (!this.eventBuffer?.pendingLength) {
+        return;
+      }
 
-    try {
-      // Note this empties the event buffer regardless of outcome of sending replay
-      const recordingData = await this.eventBuffer.finish();
+      // Only attach memory entry if eventBuffer is not empty
+      promises.push(addMemoryEntry(this));
 
       // NOTE: Copy values from instance members, as it's possible they could
       // change before the flush finishes.
       const replayId = this.session.id;
       const eventContext = this._popEventContext();
       // Always increment segmentId regardless of outcome of sending replay
       const segmentId = this.session.segmentId++;
+
+      // Save session (new segment id) after we save flush data assuming either
+      // 1) request succeeds or 2) it fails or never happens, in which case we
+      // need to retry this segment.
       this._maybeSaveSession();
 
-      await sendReplay({
+      let recordingData: ReplayRecordingData;
+
+      if (options.finishImmediate && this.eventBuffer.pendingLength) {
+        recordingData = this.eventBuffer.finishImmediate();
+      } else {
+        // NOTE: Be mindful that nothing after this point (the first `await`)
+        // will run after when the page is unloaded.
+        await Promise.all(promises);
+
+        // This can be empty due to blur events calling `runFlush` directly. In
+        // the case where we have a snapshot checkout and a blur event
+        // happening near the same time, the blur event can end up emptying the
+        // buffer even if snapshot happens first.
+        if (!this.eventBuffer.pendingLength) {
+          return;
+        }
+        // This empties the event buffer regardless of outcome of sending replay
+        recordingData = await this.eventBuffer.finish();
+      }
+
+      const sendReplayPromise = sendReplay({
         replayId,
         recordingData,
         segmentId,
@@ -838,6 +867,10 @@ export class ReplayContainer implements ReplayContainerInterface {
         options: this.getOptions(),
         timestamp: new Date().getTime(),
       });
+
+      await sendReplayPromise;
+
+      return;
     } catch (err) {
       this._handleException(err);
 

packages/replay/src/types.ts

@@ -7,6 +7,15 @@ export type RecordingOptions = recordOptions;
 
 export type AllPerformanceEntry = PerformancePaintTiming | PerformanceResourceTiming | PerformanceNavigationTiming;
 
+export interface FlushOptions {
+  /**
+   * Attempt to finish the flush immediately without any asynchronous operations
+   * (e.g. worker calls). This is not directly related to `flushImmediate` which
+   * skips the debounced flush.
+   */
+  finishImmediate?: boolean;
+}
+
 export interface SendReplayData {
   recordingData: ReplayRecordingData;
   replayId: string;
@@ -18,6 +27,10 @@ export interface SendReplayData {
   options: ReplayPluginOptions;
 }
 
+export type PendingReplayData = Omit<SendReplayData, 'recordingData'|'session'|'options'> & {
+  recordingData: RecordingEvent[];
+};
+
 export type InstrumentationTypeBreadcrumb = 'dom' | 'scope';
 
 /**
@@ -237,6 +250,11 @@ export interface EventBuffer {
    * Clears and returns the contents of the buffer.
    */
   finish(): Promise<ReplayRecordingData>;
+
+  /**
+* Clears and synchronously returns the pending contents of the buffer. This means no compression.
+*/
+  finishImmediate(): string;
 }
 
 export type AddUpdateCallback = () => boolean | void;

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

@@ -107,30 +107,24 @@ describe('Integration | flush', () => {
     replay && replay.stop();
   });
 
-  it('flushes twice after multiple flush() calls)', async () => {
-    // blur events cause an immediate flush (as well as a flush due to adding a
-    // breadcrumb) -- this means that the first blur event will be flushed and
-    // the following blur events will all call a debounced flush function, which
-    // should end up queueing a second flush
-
+  it('flushes after each blur event', async () => {
+    // blur events cause an immediate flush that bypass the debounced flush
+    // function and skip any async workers
+    expect(mockRunFlush).toHaveBeenCalledTimes(0);
     WINDOW.dispatchEvent(new Event('blur'));
+    expect(mockRunFlush).toHaveBeenCalledTimes(1);
     WINDOW.dispatchEvent(new Event('blur'));
+    expect(mockRunFlush).toHaveBeenCalledTimes(2);
     WINDOW.dispatchEvent(new Event('blur'));
+    expect(mockRunFlush).toHaveBeenCalledTimes(3);
     WINDOW.dispatchEvent(new Event('blur'));
+    expect(mockRunFlush).toHaveBeenCalledTimes(4);
 
-    expect(mockFlush).toHaveBeenCalledTimes(4);
+    expect(mockFlush).toHaveBeenCalledTimes(0);
 
     jest.runAllTimers();
     await new Promise(process.nextTick);
-    expect(mockRunFlush).toHaveBeenCalledTimes(1);
-
-    jest.runAllTimers();
-    await new Promise(process.nextTick);
-    expect(mockRunFlush).toHaveBeenCalledTimes(2);
-
-    jest.runAllTimers();
-    await new Promise(process.nextTick);
-    expect(mockRunFlush).toHaveBeenCalledTimes(2);
+    expect(mockRunFlush).toHaveBeenCalledTimes(4);
   });
 
   it('long first flush enqueues following events', async () => {
@@ -141,8 +135,11 @@ describe('Integration | flush', () => {
 
     expect(mockAddPerformanceEntries).not.toHaveBeenCalled();
 
-    // flush #1 @ t=0s - due to blur
-    WINDOW.dispatchEvent(new Event('blur'));
+    // flush #1 @ t=0s - (blur bypasses debounce, so manually call `flushImmedate`)
+    domHandler({
+      name: 'click',
+    });
+    replay.flushImmediate();
     expect(mockFlush).toHaveBeenCalledTimes(1);
     expect(mockRunFlush).toHaveBeenCalledTimes(1);
 
@@ -155,17 +152,23 @@ describe('Integration | flush', () => {
     expect(mockFlush).toHaveBeenCalledTimes(2);
 
     await advanceTimers(1000);
-    // flush #3 @ t=6s - due to blur
-    WINDOW.dispatchEvent(new Event('blur'));
+    domHandler({
+      name: 'click',
+    });
+    // flush #3 @ t=6s
+    replay.flushImmediate();
     expect(mockFlush).toHaveBeenCalledTimes(3);
 
     // NOTE: Blur also adds a breadcrumb which calls `addUpdate`, meaning it will
     // flush after `flushMinDelay`, but this gets cancelled by the blur
     await advanceTimers(8000);
     expect(mockFlush).toHaveBeenCalledTimes(3);
 
-    // flush #4 @ t=14s - due to blur
-    WINDOW.dispatchEvent(new Event('blur'));
+    // flush #4 @ t=14s
+    domHandler({
+      name: 'click',
+    });
+    replay.flushImmediate();
     expect(mockFlush).toHaveBeenCalledTimes(4);
 
     expect(mockRunFlush).toHaveBeenCalledTimes(1);

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

@@ -106,6 +106,9 @@ describe('Integration | stop', () => {
     };
 
     addEvent(replay, TEST_EVENT);
+    // This is an interesting test case because `start()` causes a checkout and a `flushImmediate`, and
+    // blur causes a direct `runFlush` call to ensure if window unloads, it is able to send out an uncompressed segment.
+    // This means that the non-async blur call can empty out the buffer before `flushImmediate` finishes.
     WINDOW.dispatchEvent(new Event('blur'));
     jest.runAllTimers();
     await new Promise(process.nextTick);
@@ -127,15 +130,15 @@ describe('Integration | stop', () => {
   });
 
   it('does not buffer events when stopped', async function () {
-    WINDOW.dispatchEvent(new Event('blur'));
+    WINDOW.dispatchEvent(new Event('focus'));
     expect(replay.eventBuffer?.pendingLength).toBe(1);
 
     // stop replays
     integration.stop();
 
     expect(replay.eventBuffer?.pendingLength).toBe(undefined);
 
-    WINDOW.dispatchEvent(new Event('blur'));
+    WINDOW.dispatchEvent(new Event('focus'));
     await new Promise(process.nextTick);
 
     expect(replay.eventBuffer?.pendingLength).toBe(undefined);