fix(replay): Handle compression failures more robustly (#6988)

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

@@ -1,86 +1,36 @@
 import type { ReplayRecordingData } from '@sentry/types';
-import { logger } from '@sentry/utils';
 
-import type { AddEventResult, EventBuffer, RecordingEvent, WorkerRequest, WorkerResponse } from '../types';
+import type { AddEventResult, EventBuffer, RecordingEvent } from '../types';
+import { WorkerHandler } from './WorkerHandler';
 
 /**
  * Event buffer that uses a web worker to compress events.
  * 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 _ensureReadyPromise?: Promise<void>;
+  private _worker: WorkerHandler;
 
   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._worker = new WorkerHandler(worker);
+    this.hasEvents = false;
   }
 
   /**
    * Ensure the worker is ready (or not).
    * This will either resolve when the worker is ready, or reject if an error occured.
    */
   public ensureReady(): Promise<void> {
-    // Ensure we only check once
-    if (this._ensureReadyPromise) {
-      return this._ensureReadyPromise;
-    }
-
-    this._ensureReadyPromise = new Promise((resolve, reject) => {
-      this._worker.addEventListener(
-        'message',
-        ({ data }: MessageEvent) => {
-          if ((data as WorkerResponse).success) {
-            resolve();
-          } else {
-            reject();
-          }
-        },
-        { once: true },
-      );
-
-      this._worker.addEventListener(
-        'error',
-        error => {
-          reject(error);
-        },
-        { once: true },
-      );
-    });
-
-    return this._ensureReadyPromise;
+    return this._worker.ensureReady();
   }
 
   /**
    * Destroy the event buffer.
    */
   public destroy(): void {
-    __DEBUG_BUILD__ && logger.log('[Replay] Destroying compression worker');
-    this._worker.terminate();
+    this._worker.destroy();
   }
 
   /**
@@ -89,19 +39,12 @@ 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: [],
-      });
-    }
-
-    // Don't store checkout events in `_pendingEvents` because they are too large
-    if (!isCheckout) {
-      this._pendingEvents.push(event);
+      await this._clear();
     }
 
     return this._sendEventToWorker(event);
@@ -110,97 +53,30 @@ export class EventBufferCompressionWorker implements EventBuffer {
   /**
    * 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);
-    }
-  }
-
-  /**
-   * Post message to worker and wait for response before resolving promise.
-   */
-  private _postMessage<T>({ id, method, args }: WorkerRequest): Promise<T> {
-    return new Promise((resolve, reject) => {
-      const listener = ({ data }: MessageEvent): void => {
-        const response = data as WorkerResponse;
-        if (response.method !== method) {
-          return;
-        }
-
-        // There can be multiple listeners for a single method, the id ensures
-        // that the response matches the caller.
-        if (response.id !== id) {
-          return;
-        }
-
-        // At this point, we'll always want to remove listener regardless of result status
-        this._worker.removeEventListener('message', listener);
-
-        if (!response.success) {
-          // TODO: Do some error handling, not sure what
-          __DEBUG_BUILD__ && logger.error('[Replay]', response.response);
-
-          reject(new Error('Error in compression worker'));
-          return;
-        }
-
-        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 });
-    });
+  public finish(): Promise<ReplayRecordingData> {
+    return this._finishRequest();
   }
 
   /**
    * Send the event to the worker.
    */
-  private async _sendEventToWorker(event: RecordingEvent): Promise<AddEventResult> {
-    const promise = this._postMessage<void>({
-      id: this._getAndIncrementId(),
-      method: 'addEvent',
-      args: [event],
-    });
-
-    // XXX: See note in `get length()`
-    this._eventBufferItemLength++;
-
-    return promise;
+  private _sendEventToWorker(event: RecordingEvent): Promise<AddEventResult> {
+    return this._worker.postMessage<void>('addEvent', JSON.stringify(event));
   }
 
   /**
    * 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;
+  private async _finishRequest(): Promise<Uint8Array> {
+    const response = await this._worker.postMessage<Uint8Array>('finish');
 
-    this._pendingEvents = [];
+    this.hasEvents = false;
 
-    return promise;
+    return response;
   }
 
-  /** Get the current ID and increment it for the next call. */
-  private _getAndIncrementId(): number {
-    return this._id++;
+  /** Clear any pending events from the worker. */
+  private _clear(): Promise<void> {
+    return this._worker.postMessage('clear');
   }
 }

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);
+    }
   }
 }