feat(replay): Keep the last checkout when in error mode

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

@@ -48,7 +48,7 @@
     "@babel/core": "^7.17.5",
     "@types/pako": "^2.0.0",
     "jsdom-worker": "^0.2.1",
-    "pako": "^2.0.4",
+    "pako": "2.1.0",
     "rrweb": "1.1.3",
     "tslib": "^1.9.3"
   },

packages/replay/src/eventBuffer/EventBufferArray.ts

@@ -1,63 +1,75 @@
-import type { AddEventResult, EventBuffer, RecordingEvent } from '../types';
+import type { ReplayRecordingData } from '@sentry/types';
+
+import type { EventBuffer, RecordingEvent } from '../types';
+
+interface EventsGroup {
+  checkoutTimestamp: number;
+  events: RecordingEvent[];
+}
 
 /**
  * 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: EventsGroup[];
 
   public constructor() {
     this._events = [];
   }
 
   /** @inheritdoc */
   public get pendingLength(): number {
-    return this._events.length;
+    return this.pendingEvents.length;
   }
 
-  /**
-   * 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.reduce((acc, { events }) => [...events, ...acc], [] as RecordingEvent[]);
   }
 
   /** @inheritdoc */
-  public destroy(): void {
-    this._clear();
+  public getFirstCheckoutTimestamp(): number | null {
+    return (this._events[0] && this._events[0].checkoutTimestamp) || null;
   }
 
   /** @inheritdoc */
-  public async addEvent(event: RecordingEvent): Promise<AddEventResult> {
-    this._events.push(event);
-    return;
+  public destroy(): void {
+    this.clear();
   }
 
   /** @inheritdoc */
-  public async clear(untilPos?: number): Promise<void> {
-    this._clear(untilPos);
+  public addEvent(event: RecordingEvent, isCheckout?: boolean): void {
+    if (isCheckout || this._events.length === 0) {
+      const group: EventsGroup = {
+        checkoutTimestamp: event.timestamp,
+        events: [event],
+      };
+      this._events.unshift(group);
+    } else {
+      this._events[0].events.push(event);
+    }
   }
 
   /** @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;
-      resolve(JSON.stringify(eventsRet));
-      this._clear();
-    });
-  }
-
-  /** Clear all events. */
-  private _clear(untilPos?: number): void {
-    if (untilPos) {
-      this._events.splice(0, untilPos);
+  public clear(keepLastCheckout?: boolean): void {
+    if (keepLastCheckout) {
+      this._events.splice(1);
     } else {
       this._events = [];
     }
   }
+
+  /** @inheritdoc */
+  public finish(): Promise<ReplayRecordingData> {
+    const pendingEvents = this.pendingEvents.slice();
+    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,44 +1,22 @@
+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;
 
   public constructor(worker: Worker) {
+    super();
     this._worker = worker;
-    this._pendingEvents = [];
-  }
-
-  /**
-   * 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.
@@ -67,55 +45,27 @@ export class EventBufferCompressionWorker implements EventBuffer {
     });
   }
 
-  /**
-   * 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 addEvent(event: RecordingEvent): Promise<AddEventResult> {
-    this.pendingEvents.push(event);
-    return this._sendEventToWorker(event);
-  }
-
-  /** @inheritdoc */
-  public clear(untilPos?: number): Promise<void> {
-    this._clear(untilPos);
-
-    // TODO FN: Clear up to pos
-
-    // This will clear the queue of events that are waiting to be compressed
-    return this._postMessage({
-      id: this._getAndIncrementId(),
-      method: 'init',
-      args: [],
-    });
+    super.destroy();
   }
 
   /**
    * Finish the event buffer and return the compressed data.
    */
-  public finish(): Promise<Uint8Array> {
-    this._clear();
+  public async finish(): Promise<ReplayRecordingData> {
+    const pendingEvents = this.pendingEvents.slice();
 
-    return this._finishRequest(this._getAndIncrementId());
-  }
+    this.clear();
 
-  /**
-   * Clear all pending events up to the given event pos.
-   */
-  private _clear(untilPos?: number): void {
-    if (untilPos) {
-      this._pendingEvents.splice(0, untilPos);
-    } else {
-      this._pendingEvents = [];
+    try {
+      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
+      return this._finishRecording(pendingEvents);
     }
   }
 
@@ -165,36 +115,11 @@ export class EventBufferCompressionWorker implements EventBuffer {
     });
   }
 
-  /**
-   * 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;
-  }
-
   /**
    * 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', args: [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';
 
@@ -40,20 +40,25 @@ export class EventBufferProxy implements EventBuffer {
   }
 
   /** @inheritdoc */
-  public addEvent(event: RecordingEvent): Promise<AddEventResult> {
-    return this._used.addEvent(event);
+  public addEvent(event: RecordingEvent, isCheckout?: boolean): void {
+    return this._used.addEvent(event, isCheckout);
   }
 
   /** @inheritdoc */
-  public clear(untilPos?: number): Promise<void> {
-    return this._used.clear(untilPos);
+  public clear(keepLastCheckout?: boolean): void {
+    return this._used.clear(keepLastCheckout);
   }
 
   /** @inheritDoc */
   public finish(): Promise<ReplayRecordingData> {
     return this._used.finish();
   }
 
+  /** @inheritdoc */
+  public getFirstCheckoutTimestamp(): number | null {
+    return this._used.getFirstCheckoutTimestamp();
+  }
+
   /** Ensure the worker has loaded. */
   private async _ensureWorkerIsLoaded(): Promise<void> {
     try {
@@ -67,16 +72,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();
   }
 }

packages/replay/src/replay.ts

@@ -11,7 +11,6 @@ import { createEventBuffer } from './eventBuffer';
 import { getSession } from './session/getSession';
 import { saveSession } from './session/saveSession';
 import type {
-  AddEventResult,
   AddUpdateCallback,
   AllPerformanceEntry,
   EventBuffer,
@@ -678,12 +677,12 @@ export class ReplayContainer implements ReplayContainerInterface {
    * Observed performance events are added to `this.performanceEvents`. These
    * are included in the replay event before it is finished and sent to Sentry.
    */
-  private _addPerformanceEntries(): Promise<Array<AddEventResult | null>> {
+  private _addPerformanceEntries(): void {
     // Copy and reset entries before processing
     const entries = [...this.performanceEvents];
     this.performanceEvents = [];
 
-    return Promise.all(createPerformanceSpans(this, createPerformanceEntries(entries)));
+    createPerformanceSpans(this, createPerformanceEntries(entries));
   }
 
   /**