feat(replay): Keep 30-60s instead of 0-60s of recording in replay errror mode

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/package.json

@@ -49,7 +49,7 @@
     "@sentry-internal/rrweb": "1.100.1",
     "@types/pako": "^2.0.0",
     "jsdom-worker": "^0.2.1",
-    "pako": "^2.0.4",
+    "pako": "2.1.0",
     "tslib": "^1.9.3"
   },
   "dependencies": {

packages/replay/src/constants.ts

@@ -27,8 +27,8 @@ export const MASK_ALL_TEXT_SELECTOR = 'body *:not(style), body *:not(script)';
 export const DEFAULT_FLUSH_MIN_DELAY = 5_000;
 export const DEFAULT_FLUSH_MAX_DELAY = 5_000;
 
-/* How long to wait for error checkouts */
-export const ERROR_CHECKOUT_TIME = 60_000;
+/* How often to capture a full checkout when in error mode */
+export const ERROR_CHECKOUT_TIME = 30_000;
 
 export const RETRY_BASE_INTERVAL = 5000;
 export const RETRY_MAX_COUNT = 3;

packages/replay/src/eventBuffer/EventBufferArray.ts

@@ -1,54 +1,59 @@
-import type { AddEventResult, EventBuffer, RecordingEvent } from '../types';
+import type { ReplayRecordingData } from '@sentry/types';
+
+import type { EventBuffer, RecordingEvent } from '../types';
+import { PartitionedQueue } from './PartitionedQueue';
 
 /**
  * A basic event buffer that does not do any compression.
  * Used as fallback if the compression worker cannot be loaded or is disabled.
  */
 export class EventBufferArray implements EventBuffer {
-  private _events: RecordingEvent[];
+  private _events: PartitionedQueue<RecordingEvent>;
 
   public constructor() {
-    this._events = [];
+    this._events = new PartitionedQueue<RecordingEvent>();
   }
 
   /** @inheritdoc */
   public get pendingLength(): number {
-    return this._events.length;
+    return this._events.getLength();
   }
 
-  /**
-   * Returns the raw events that are buffered. In `EventBufferArray`, this is the
-   * same as `this._events`.
-   */
+  /** @inheritdoc */
   public get pendingEvents(): RecordingEvent[] {
-    return this._events;
+    return this._events.getItems();
+  }
+
+  /** @inheritdoc */
+  public getEarliestTimestamp(): number | null {
+    return this.pendingEvents.map(event => event.timestamp).sort()[0] || null;
   }
 
   /** @inheritdoc */
   public destroy(): void {
-    this._events = [];
+    this.clear();
   }
 
   /** @inheritdoc */
-  public async addEvent(event: RecordingEvent, isCheckout?: boolean): Promise<AddEventResult> {
-    if (isCheckout) {
-      this._events = [event];
-      return;
-    }
+  public addEvent(event: RecordingEvent, isCheckout?: boolean): void {
+    this._events.add(event, isCheckout);
+  }
 
-    this._events.push(event);
-    return;
+  /** @inheritdoc */
+  public clear(keepLastCheckout?: boolean): void {
+    this._events.clear(keepLastCheckout);
   }
 
   /** @inheritdoc */
-  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));
-    });
+  public finish(): Promise<ReplayRecordingData> {
+    const { pendingEvents } = this;
+    this.clear();
+
+    return Promise.resolve(this._finishRecording(pendingEvents));
+  }
+
+  /** Finish in a sync manner. */
+  protected _finishRecording(events: RecordingEvent[]): ReplayRecordingData {
+    return JSON.stringify(events);
   }
 }

packages/replay/src/eventBuffer/EventBufferCompressionWorker.ts

@@ -1,45 +1,23 @@
 import type { ReplayRecordingData } from '@sentry/types';
 import { logger } from '@sentry/utils';
 
-import type { AddEventResult, EventBuffer, RecordingEvent, WorkerRequest, WorkerResponse } from '../types';
+import type { RecordingEvent, WorkerRequest, WorkerResponse } from '../types';
+import { EventBufferArray } from './EventBufferArray';
 
 /**
  * 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[] = [];
-
+export class EventBufferCompressionWorker extends EventBufferArray {
   private _worker: Worker;
-  private _eventBufferItemLength: number = 0;
+
   private _id: number = 0;
   private _ensureReadyPromise?: Promise<void>;
 
   public constructor(worker: Worker) {
+    super();
     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;
-  }
-
   /**
    * Ensure the worker is ready (or not).
    * This will either resolve when the worker is ready, or reject if an error occured.
@@ -75,56 +53,34 @@ export class EventBufferCompressionWorker implements EventBuffer {
     return this._ensureReadyPromise;
   }
 
-  /**
-   * Destroy the event buffer.
-   */
+  /** @inheritdoc */
   public destroy(): void {
     __DEBUG_BUILD__ && logger.log('[Replay] Destroying compression worker');
     this._worker.terminate();
-  }
-
-  /**
-   * Add an event to the event buffer.
-   *
-   * Returns true if event was successfuly received and processed by worker.
-   */
-  public async addEvent(event: RecordingEvent, isCheckout?: boolean): Promise<AddEventResult> {
-    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);
-    }
-
-    return this._sendEventToWorker(event);
+    super.destroy();
   }
 
   /**
    * Finish the event buffer and return the compressed data.
    */
   public async finish(): Promise<ReplayRecordingData> {
+    const pendingEvents = this.pendingEvents.slice();
+
+    this.clear();
+
     try {
-      return await this._finishRequest(this._getAndIncrementId());
+      return await this._compressEvents(this._getAndIncrementId(), pendingEvents);
     } 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);
+      return this._finishRecording(pendingEvents);
     }
   }
 
   /**
    * 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 +108,18 @@ 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 });
-    });
-  }
-
-  /**
-   * 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],
+      this._worker.postMessage({ id, method, arg });
     });
-
-    // 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;
-
-    this._pendingEvents = [];
-
-    return promise;
+  private async _compressEvents(id: number, events: RecordingEvent[]): Promise<Uint8Array> {
+    return this._postMessage<Uint8Array>({ id, method: 'compress', arg: JSON.stringify(events) });
   }
 
   /** Get the current ID and increment it for the next call. */

packages/replay/src/eventBuffer/EventBufferProxy.ts

@@ -1,7 +1,7 @@
 import type { ReplayRecordingData } from '@sentry/types';
 import { logger } from '@sentry/utils';
 
-import type { AddEventResult, EventBuffer, RecordingEvent } from '../types';
+import type { EventBuffer, RecordingEvent } from '../types';
 import { EventBufferArray } from './EventBufferArray';
 import { EventBufferCompressionWorker } from './EventBufferCompressionWorker';
 
@@ -21,9 +21,7 @@ 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 */
@@ -42,15 +40,16 @@ export class EventBufferProxy implements EventBuffer {
     this._compression.destroy();
   }
 
-  /**
-   * Add an event to the event buffer.
-   *
-   * Returns true if event was successfully added.
-   */
-  public addEvent(event: RecordingEvent, isCheckout?: boolean): Promise<AddEventResult> {
+  /** @inheritdoc */
+  public addEvent(event: RecordingEvent, isCheckout?: boolean): void {
     return this._used.addEvent(event, isCheckout);
   }
 
+  /** @inheritdoc */
+  public clear(keepLastCheckout?: boolean): void {
+    return this._used.clear(keepLastCheckout);
+  }
+
   /** @inheritDoc */
   public async finish(): Promise<ReplayRecordingData> {
     // Ensure the worker is loaded, so the sent event is compressed
@@ -59,6 +58,11 @@ export class EventBufferProxy implements EventBuffer {
     return this._used.finish();
   }
 
+  /** @inheritdoc */
+  public getEarliestTimestamp(): number | null {
+    return this._used.getEarliestTimestamp();
+  }
+
   /** Ensure the worker has loaded. */
   public ensureWorkerIsLoaded(): Promise<void> {
     return this._ensureWorkerIsLoadedPromise;
@@ -77,16 +81,14 @@ export class EventBufferProxy implements EventBuffer {
 
     // Compression worker is ready, we can use it
     // Now we need to switch over the array buffer to the compression worker
-    const addEventPromises: Promise<void>[] = [];
     for (const event of this._fallback.pendingEvents) {
-      addEventPromises.push(this._compression.addEvent(event));
+      this._compression.addEvent(event);
     }
 
     // We switch over to the compression 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);
+    this._fallback.clear();
   }
 }