fix(replay): Handle compression failures more robustly

We now stop the recording when either `addEvent` or `finish` fail.

Author: mydea

packages/replay/.eslintrc.js

@@ -13,6 +13,10 @@ module.exports = {
         // TODO: figure out if we need a worker-specific tsconfig
         project: ['tsconfig.worker.json'],
       },
+      rules: {
+        // We cannot use backticks, as that conflicts with the stringified worker
+        'prefer-template': 'off',
+      },
     },
     {
       files: ['src/worker/**/*.js'],

packages/replay/src/eventBuffer/EventBufferArray.ts

@@ -5,38 +5,31 @@ import type { AddEventResult, EventBuffer, RecordingEvent } from '../types';
  * Used as fallback if the compression worker cannot be loaded or is disabled.
  */
 export class EventBufferArray implements EventBuffer {
-  private _events: RecordingEvent[];
+  /** All the events that are buffered to be sent. */
+  public events: RecordingEvent[];
 
   public constructor() {
-    this._events = [];
+    this.events = [];
   }
 
   /** @inheritdoc */
-  public get pendingLength(): number {
-    return this._events.length;
-  }
-
-  /**
-   * Returns the raw events that are buffered. In `EventBufferArray`, this is the
-   * same as `this._events`.
-   */
-  public get pendingEvents(): RecordingEvent[] {
-    return this._events;
+  public get hasEvents(): boolean {
+    return this.events.length > 0;
   }
 
   /** @inheritdoc */
   public destroy(): void {
-    this._events = [];
+    this.events = [];
   }
 
   /** @inheritdoc */
   public async addEvent(event: RecordingEvent, isCheckout?: boolean): Promise<AddEventResult> {
     if (isCheckout) {
-      this._events = [event];
+      this.events = [event];
       return;
     }
 
-    this._events.push(event);
+    this.events.push(event);
     return;
   }
 
@@ -46,8 +39,8 @@ export class EventBufferArray implements EventBuffer {
       // 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 = [];
+      const eventsRet = this.events;
+      this.events = [];
       resolve(JSON.stringify(eventsRet));
     });
   }

packages/replay/src/eventBuffer/EventBufferCompressionWorker.ts

@@ -8,36 +8,17 @@ import type { AddEventResult, EventBuffer, RecordingEvent, WorkerRequest, Worker
  * Exported only for testing.
  */
 export class EventBufferCompressionWorker implements EventBuffer {
-  /**
-   * Keeps track of the list of events since the last flush that have not been compressed.
-   * For example, page is reloaded and a flush attempt is made, but
-   * `finish()` (and thus the flush), does not complete.
-   */
-  public _pendingEvents: RecordingEvent[] = [];
+  /** @inheritdoc */
+  public hasEvents: boolean;
 
   private _worker: Worker;
-  private _eventBufferItemLength: number = 0;
-  private _id: number = 0;
+  private _id: number;
   private _ensureReadyPromise?: Promise<void>;
 
   public constructor(worker: Worker) {
     this._worker = worker;
-  }
-
-  /**
-   * The number of raw events that are buffered. This may not be the same as
-   * the number of events that have been compresed in the worker because
-   * `addEvent` is async.
-   */
-  public get pendingLength(): number {
-    return this._eventBufferItemLength;
-  }
-
-  /**
-   * Returns a list of the raw recording events that are being compressed.
-   */
-  public get pendingEvents(): RecordingEvent[] {
-    return this._pendingEvents;
+    this.hasEvents = false;
+    this._id = 0;
   }
 
   /**
@@ -89,42 +70,31 @@ export class EventBufferCompressionWorker implements EventBuffer {
    * Returns true if event was successfuly received and processed by worker.
    */
   public async addEvent(event: RecordingEvent, isCheckout?: boolean): Promise<AddEventResult> {
+    this.hasEvents = true;
+
     if (isCheckout) {
       // This event is a checkout, make sure worker buffer is cleared before
       // proceeding.
       await this._postMessage({
         id: this._getAndIncrementId(),
-        method: 'init',
-        args: [],
+        method: 'clear',
       });
     }
 
-    // Don't store checkout events in `_pendingEvents` because they are too large
-    if (!isCheckout) {
-      this._pendingEvents.push(event);
-    }
-
     return this._sendEventToWorker(event);
   }
 
   /**
    * Finish the event buffer and return the compressed data.
    */
-  public async finish(): Promise<ReplayRecordingData> {
-    try {
-      return await this._finishRequest(this._getAndIncrementId());
-    } catch (error) {
-      __DEBUG_BUILD__ && logger.error('[Replay] Error when trying to compress events', error);
-      // fall back to uncompressed
-      const events = this.pendingEvents;
-      return JSON.stringify(events);
-    }
+  public finish(): Promise<ReplayRecordingData> {
+    return this._finishRequest(this._getAndIncrementId());
   }
 
   /**
    * Post message to worker and wait for response before resolving promise.
    */
-  private _postMessage<T>({ id, method, args }: WorkerRequest): Promise<T> {
+  private _postMessage<T>({ id, method, arg }: WorkerRequest): Promise<T> {
     return new Promise((resolve, reject) => {
       const listener = ({ data }: MessageEvent): void => {
         const response = data as WorkerResponse;
@@ -152,51 +122,33 @@ export class EventBufferCompressionWorker implements EventBuffer {
         resolve(response.response as T);
       };
 
-      let stringifiedArgs;
-      try {
-        stringifiedArgs = JSON.stringify(args);
-      } catch (err) {
-        __DEBUG_BUILD__ && logger.error('[Replay] Error when trying to stringify args', err);
-        stringifiedArgs = '[]';
-      }
-
       // Note: we can't use `once` option because it's possible it needs to
       // listen to multiple messages
       this._worker.addEventListener('message', listener);
-      this._worker.postMessage({ id, method, args: stringifiedArgs });
+      this._worker.postMessage({ id, method, arg });
     });
   }
 
   /**
    * Send the event to the worker.
    */
-  private async _sendEventToWorker(event: RecordingEvent): Promise<AddEventResult> {
-    const promise = this._postMessage<void>({
+  private _sendEventToWorker(event: RecordingEvent): Promise<AddEventResult> {
+    return this._postMessage<void>({
       id: this._getAndIncrementId(),
       method: 'addEvent',
-      args: [event],
+      arg: JSON.stringify(event),
     });
-
-    // XXX: See note in `get length()`
-    this._eventBufferItemLength++;
-
-    return promise;
   }
 
   /**
    * Finish the request and return the compressed data from the worker.
    */
   private async _finishRequest(id: number): Promise<Uint8Array> {
-    const promise = this._postMessage<Uint8Array>({ id, method: 'finish', args: [] });
-
-    // XXX: See note in `get length()`
-    this._eventBufferItemLength = 0;
-
-    await promise;
+    const response = await this._postMessage<Uint8Array>({ id, method: 'finish' });
 
-    this._pendingEvents = [];
+    this.hasEvents = false;
 
-    return promise;
+    return response;
   }
 
   /** Get the current ID and increment it for the next call. */

packages/replay/src/eventBuffer/EventBufferProxy.ts

@@ -21,19 +21,12 @@ export class EventBufferProxy implements EventBuffer {
     this._compression = new EventBufferCompressionWorker(worker);
     this._used = this._fallback;
 
-    this._ensureWorkerIsLoadedPromise = this._ensureWorkerIsLoaded().catch(() => {
-      // Ignore errors here
-    });
+    this._ensureWorkerIsLoadedPromise = this._ensureWorkerIsLoaded();
   }
 
   /** @inheritDoc */
-  public get pendingLength(): number {
-    return this._used.pendingLength;
-  }
-
-  /** @inheritDoc */
-  public get pendingEvents(): RecordingEvent[] {
-    return this._used.pendingEvents;
+  public get hasEvents(): boolean {
+    return this._used.hasEvents;
   }
 
   /** @inheritDoc */
@@ -75,18 +68,28 @@ export class EventBufferProxy implements EventBuffer {
       return;
     }
 
-    // Compression worker is ready, we can use it
     // Now we need to switch over the array buffer to the compression worker
+    await this._switchToCompressionWorker();
+  }
+
+  /** Switch the used buffer to the compression worker. */
+  private async _switchToCompressionWorker(): Promise<void> {
+    const { events } = this._fallback;
+
     const addEventPromises: Promise<void>[] = [];
-    for (const event of this._fallback.pendingEvents) {
+    for (const event of events) {
       addEventPromises.push(this._compression.addEvent(event));
     }
 
-    // We switch over to the compression buffer immediately - any further events will be added
+    // We switch over to the new buffer immediately - any further events will be added
     // after the previously buffered ones
     this._used = this._compression;
 
     // Wait for original events to be re-added before resolving
-    await Promise.all(addEventPromises);
+    try {
+      await Promise.all(addEventPromises);
+    } catch (error) {
+      __DEBUG_BUILD__ && logger.warn('[Replay] Failed to add events when switching buffers.', error);
+    }
   }
 }

packages/replay/src/replay.ts

@@ -772,7 +772,7 @@ export class ReplayContainer implements ReplayContainerInterface {
     await this._addPerformanceEntries();
 
     // Check eventBuffer again, as it could have been stopped in the meanwhile
-    if (!this.eventBuffer || !this.eventBuffer.pendingLength) {
+    if (!this.eventBuffer || !this.eventBuffer.hasEvents) {
       return;
     }
 

packages/replay/src/types.ts

@@ -23,8 +23,8 @@ export interface SendReplayData {
  */
 export interface WorkerRequest {
   id: number;
-  method: string;
-  args: unknown[];
+  method: 'clear' | 'addEvent' | 'finish';
+  arg?: string;
 }
 
 // PerformancePaintTiming and PerformanceNavigationTiming are only available with TS 4.4 and newer
@@ -261,14 +261,9 @@ export interface Session {
 
 export interface EventBuffer {
   /**
-   * The number of raw events that are buffered
-   */
-  readonly pendingLength: number;
-
-  /**
-   * The raw events that are buffered.
+   * If any events have been added to the buffer.
    */
-  readonly pendingEvents: RecordingEvent[];
+  readonly hasEvents: boolean;
 
   /**
    * Destroy the event buffer.

packages/replay/src/util/addEvent.ts

@@ -1,3 +1,5 @@
+import { logger } from '@sentry/utils';
+
 import { SESSION_IDLE_DURATION } from '../constants';
 import type { AddEventResult, RecordingEvent, ReplayContainer } from '../types';
 
@@ -39,5 +41,10 @@ export async function addEvent(
     replay.getContext().earliestEvent = timestampInMs;
   }
 
-  return replay.eventBuffer.addEvent(event, isCheckout);
+  try {
+    return await replay.eventBuffer.addEvent(event, isCheckout);
+  } catch (error) {
+    __DEBUG_BUILD__ && logger.error(error);
+    replay.stop();
+  }
 }

packages/replay/src/worker/worker.js

@@ -168,7 +168,7 @@ describe('Integration | rate-limiting behaviour', () => {
       expect(replay).toHaveLastSentReplay({ recordingData: JSON.stringify([TEST_EVENT3]) });
 
       // events array should be empty
-      expect(replay.eventBuffer?.pendingLength).toBe(0);
+      expect(replay.eventBuffer?.hasEvents).toBe(false);
     },
   );
 
@@ -253,7 +253,7 @@ describe('Integration | rate-limiting behaviour', () => {
     expect(replay).toHaveLastSentReplay({ recordingData: JSON.stringify([TEST_EVENT3]) });
 
     // events array should be empty
-    expect(replay.eventBuffer?.pendingLength).toBe(0);
+    expect(replay.eventBuffer?.hasEvents).toBe(false);
   });
 
   it("doesn't do anything, if a rate limit is hit and recording is already paused", async () => {

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

@@ -106,7 +106,7 @@ describe('Integration | sendReplayEvent', () => {
     expect(replay.session?.segmentId).toBe(1);
 
     // events array should be empty
-    expect(replay.eventBuffer?.pendingLength).toBe(0);
+    expect(replay.eventBuffer?.hasEvents).toBe(false);
   });
 
   it('update last activity when user clicks mouse', async () => {
@@ -145,7 +145,7 @@ describe('Integration | sendReplayEvent', () => {
     expect(replay.session?.segmentId).toBe(1);
 
     // events array should be empty
-    expect(replay.eventBuffer?.pendingLength).toBe(0);
+    expect(replay.eventBuffer?.hasEvents).toBe(false);
   });
 
   it('uploads a replay event if maxFlushDelay is set 15 seconds have elapsed since the last replay upload', async () => {
@@ -173,7 +173,7 @@ describe('Integration | sendReplayEvent', () => {
     expect(replay.session?.lastActivity).toBe(BASE_TIMESTAMP);
     expect(replay.session?.segmentId).toBe(1);
     // events array should be empty
-    expect(replay.eventBuffer?.pendingLength).toBe(0);
+    expect(replay.eventBuffer?.hasEvents).toBe(false);
 
     // Let's make sure it continues to work
     mockTransportSend.mockClear();
@@ -218,7 +218,7 @@ describe('Integration | sendReplayEvent', () => {
     // Session's last activity should not be updated
     expect(replay.session?.lastActivity).toBe(BASE_TIMESTAMP);
     // events array should be empty
-    expect(replay.eventBuffer?.pendingLength).toBe(0);
+    expect(replay.eventBuffer?.hasEvents).toBe(false);
   });
 
   it('uploads a replay event when document becomes hidden', async () => {
@@ -246,7 +246,7 @@ describe('Integration | sendReplayEvent', () => {
     // visibilitystate as user being active
     expect(replay.session?.lastActivity).toBe(BASE_TIMESTAMP);
     // events array should be empty
-    expect(replay.eventBuffer?.pendingLength).toBe(0);
+    expect(replay.eventBuffer?.hasEvents).toBe(false);
   });
 
   it('uploads a dom breadcrumb 5 seconds after listener receives an event', async () => {