feat(replay): Keep min. 30 seconds of data in error mode

Author: mydea

packages/replay/src/constants.ts

@@ -24,8 +24,8 @@ export const MAX_SESSION_LIFE = 3_600_000; // 60 minutes
 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,47 +1,59 @@
+import type { ReplayRecordingData } from '@sentry/types';
+
 import type { AddEventResult, 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 {
   /** All the events that are buffered to be sent. */
-  public events: RecordingEvent[];
+  protected _events: PartitionedQueue<RecordingEvent>;
 
   public constructor() {
-    this.events = [];
+    this._events = new PartitionedQueue<RecordingEvent>();
+  }
+
+  /** @inheritdoc */
+  public get events(): RecordingEvent[] {
+    return this._events.getItems();
   }
 
   /** @inheritdoc */
   public get hasEvents(): boolean {
-    return this.events.length > 0;
+    return this._events.getLength() > 0;
   }
 
   /** @inheritdoc */
   public destroy(): void {
-    this.events = [];
+    this._events.clear();
   }
 
   /** @inheritdoc */
-  public async addEvent(event: RecordingEvent, isCheckout?: boolean): Promise<AddEventResult> {
-    if (isCheckout) {
-      this.events = [event];
-      return;
-    }
+  public async clear(keepLastCheckout?: boolean): Promise<void> {
+    this._events.clear(keepLastCheckout);
+  }
 
-    this.events.push(event);
-    return;
+  /** @inheritdoc */
+  public async addEvent(event: RecordingEvent, isCheckout?: boolean): Promise<AddEventResult> {
+    this._events.add(event, isCheckout);
   }
 
   /** @inheritdoc */
-  public finish(): Promise<string> {
+  public finish(): Promise<ReplayRecordingData> {
     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 = [];
+      this._events.clear();
       resolve(JSON.stringify(eventsRet));
     });
   }
+
+  /** @inheritdoc */
+  public getEarliestTimestamp(): number | null {
+    return this.events.map(event => event.timestamp).sort()[0] || null;
+  }
 }

packages/replay/src/eventBuffer/EventBufferCompressionWorker.ts

@@ -57,6 +57,18 @@ export class EventBufferCompressionWorker implements EventBuffer {
     return this._finishRequest();
   }
 
+  /** @inheritdoc */
+  public clear(): Promise<void> {
+    // NOTE: We do not support keeping the last checkout in this implementation.
+    return this._clear();
+  }
+
+  /** @inheritdoc */
+  public getEarliestTimestamp(): number | null {
+    // Not supported in this mode
+    return null;
+  }
+
   /**
    * Send the event to the worker.
    */

packages/replay/src/eventBuffer/EventBufferPartitionedCompressionWorker.ts

@@ -0,0 +1,46 @@
+import type { ReplayRecordingData } from '@sentry/types';
+
+import type { RecordingEvent } from '../types';
+import { EventBufferArray } from './EventBufferArray';
+import { WorkerHandler } from './WorkerHandler';
+
+/**
+ * Event buffer that uses a web worker to compress events.
+ * Exported only for testing.
+ */
+export class EventBufferPartitionedCompressionWorker extends EventBufferArray {
+  private _worker: WorkerHandler;
+
+  public constructor(worker: Worker) {
+    super();
+    this._worker = new WorkerHandler(worker);
+  }
+  /**
+   * 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> {
+    return this._worker.ensureReady();
+  }
+
+  /** @inheritdoc */
+  public destroy(): void {
+    this._worker.destroy();
+    super.destroy();
+  }
+
+  /**
+   * Finish the event buffer and return the compressed data.
+   */
+  public finish(): Promise<ReplayRecordingData> {
+    const { events } = this;
+    this._events.clear();
+
+    return this._compressEvents(events);
+  }
+
+  /** Compress a given array of events at once. */
+  private _compressEvents(events: RecordingEvent[]): Promise<Uint8Array> {
+    return this._worker.postMessage<Uint8Array>('compress', JSON.stringify(events));
+  }
+}

packages/replay/src/eventBuffer/EventBufferProxy.ts

@@ -4,6 +4,7 @@ import { logger } from '@sentry/utils';
 import type { AddEventResult, EventBuffer, RecordingEvent } from '../types';
 import { EventBufferArray } from './EventBufferArray';
 import { EventBufferCompressionWorker } from './EventBufferCompressionWorker';
+import { EventBufferPartitionedCompressionWorker } from './EventBufferPartitionedCompressionWorker';
 
 /**
  * This proxy will try to use the compression worker, and fall back to use the simple buffer if an error occurs there.
@@ -12,15 +13,21 @@ import { EventBufferCompressionWorker } from './EventBufferCompressionWorker';
  */
 export class EventBufferProxy implements EventBuffer {
   private _fallback: EventBufferArray;
-  private _compression: EventBufferCompressionWorker;
+  private _compression: EventBufferCompressionWorker | EventBufferPartitionedCompressionWorker;
   private _used: EventBuffer;
   private _ensureWorkerIsLoadedPromise: Promise<void>;
 
-  public constructor(worker: Worker) {
+  public constructor(worker: Worker, keepLastCheckout: boolean) {
     this._fallback = new EventBufferArray();
-    this._compression = new EventBufferCompressionWorker(worker);
-    this._used = this._fallback;
 
+    // In error mode, we use the partitioned compression worker, which does not use compression streaming
+    // Instead, all events are sent at finish-time, as we need to continuously modify the queued events
+    // In session mode, we use a streaming compression implementation, which is more performant
+    this._compression = keepLastCheckout
+      ? new EventBufferPartitionedCompressionWorker(worker)
+      : new EventBufferCompressionWorker(worker);
+
+    this._used = this._fallback;
     this._ensureWorkerIsLoadedPromise = this._ensureWorkerIsLoaded();
   }
 
@@ -52,11 +59,21 @@ export class EventBufferProxy implements EventBuffer {
     return this._used.finish();
   }
 
+  /** @inheritdoc */
+  public clear(keepLastCheckout?: boolean): Promise<void> {
+    return this._used.clear(keepLastCheckout);
+  }
+
   /** Ensure the worker has loaded. */
   public ensureWorkerIsLoaded(): Promise<void> {
     return this._ensureWorkerIsLoadedPromise;
   }
 
+  /** @inheritdoc */
+  public getEarliestTimestamp(): number | null {
+    return this._used.getEarliestTimestamp();
+  }
+
   /** Actually check if the worker has been loaded. */
   private async _ensureWorkerIsLoaded(): Promise<void> {
     try {

packages/replay/src/eventBuffer/PartitionedQueue.ts

@@ -0,0 +1,49 @@
+/**
+ * A queue with partitions for each checkout.
+ */
+export class PartitionedQueue<T> {
+  private _items: T[];
+  private _lastCheckoutPos?: number;
+
+  public constructor() {
+    this._items = [];
+  }
+
+  /** Add an item to the queue. */
+  public add(record: T, isCheckout?: boolean): void {
+    this._items.push(record);
+
+    if (isCheckout) {
+      this._lastCheckoutPos = this._items.length - 1;
+    }
+  }
+
+  /**
+   * Clear items from the queue.
+   * If `keepLastCheckout` is given, all items after the last checkout will be kept.
+   */
+  public clear(keepLastCheckout?: boolean): void {
+    if (!keepLastCheckout) {
+      this._items = [];
+      this._lastCheckoutPos = undefined;
+      return;
+    }
+
+    if (this._lastCheckoutPos) {
+      this._items = this._items.splice(this._lastCheckoutPos);
+      this._lastCheckoutPos = undefined;
+    }
+
+    // Else, there is only a single checkout recorded yet, which we don't want to clear out
+  }
+
+  /** Get all items */
+  public getItems(): T[] {
+    return this._items;
+  }
+
+  /** Get the number of items that are queued. */
+  public getLength(): number {
+    return this._items.length;
+  }
+}

packages/replay/src/eventBuffer/index.ts

@@ -7,20 +7,21 @@ import { EventBufferProxy } from './EventBufferProxy';
 
 interface CreateEventBufferParams {
   useCompression: boolean;
+  keepLastCheckout: boolean;
 }
 
 /**
  * Create an event buffer for replays.
  */
-export function createEventBuffer({ useCompression }: CreateEventBufferParams): EventBuffer {
+export function createEventBuffer({ useCompression, keepLastCheckout }: CreateEventBufferParams): EventBuffer {
   // eslint-disable-next-line no-restricted-globals
   if (useCompression && window.Worker) {
     try {
       const workerUrl = getWorkerURL();
 
       __DEBUG_BUILD__ && logger.log('[Replay] Using compression worker');
       const worker = new Worker(workerUrl);
-      return new EventBufferProxy(worker);
+      return new EventBufferProxy(worker, keepLastCheckout);
     } catch (error) {
       __DEBUG_BUILD__ && logger.log('[Replay] Failed to create compression worker');
       // Fall back to use simple event buffer array

packages/replay/src/replay.ts

@@ -180,6 +180,7 @@ export class ReplayContainer implements ReplayContainerInterface {
 
     this.eventBuffer = createEventBuffer({
       useCompression: this._options.useCompression,
+      keepLastCheckout: this.recordingMode === 'error',
     });
 
     this._addListeners();

packages/replay/src/types.ts

@@ -23,7 +23,7 @@ export interface SendReplayData {
  */
 export interface WorkerRequest {
   id: number;
-  method: 'clear' | 'addEvent' | 'finish';
+  method: 'clear' | 'addEvent' | 'finish' | 'compress';
   arg?: string;
 }
 
@@ -280,6 +280,12 @@ export interface EventBuffer {
    * Clears and returns the contents of the buffer.
    */
   finish(): Promise<ReplayRecordingData>;
+
+  /** Clear the buffer, and optional ensure we keep the last checkout. */
+  clear(keepLastCheckout?: boolean): Promise<void>;
+
+  /** Get the earliest timestamp of a pending event. */
+  getEarliestTimestamp(): number | null;
 }
 
 export type AddUpdateCallback = () => boolean | void;

packages/replay/src/util/addEvent.ts

@@ -12,7 +12,9 @@ export async function addEvent(
   event: RecordingEvent,
   isCheckout?: boolean,
 ): Promise<AddEventResult | null> {
-  if (!replay.eventBuffer) {
+  const { eventBuffer, session } = replay;
+
+  if (!eventBuffer || !session) {
     // This implies that `_isEnabled` is false
     return null;
   }
@@ -38,12 +40,27 @@ export async function addEvent(
   // Only record earliest event if a new session was created, otherwise it
   // shouldn't be relevant
   const earliestEvent = replay.getContext().earliestEvent;
-  if (replay.session && replay.session.segmentId === 0 && (!earliestEvent || timestampInMs < earliestEvent)) {
+  if (session.segmentId === 0 && (!earliestEvent || timestampInMs < earliestEvent)) {
     replay.getContext().earliestEvent = timestampInMs;
   }
 
   try {
-    return await replay.eventBuffer.addEvent(event, isCheckout);
+    if (isCheckout) {
+      if (replay.recordingMode === 'error') {
+        // Do not wait on it, just do it
+        // We know in this mode this is actually "sync"
+        void eventBuffer.clear(true);
+
+        // Ensure we have the correct first checkout timestamp when an error occurs
+        if (!session.segmentId) {
+          replay.getContext().earliestEvent = eventBuffer.getEarliestTimestamp();
+        }
+      } else {
+        await eventBuffer.clear();
+      }
+    }
+
+    return await eventBuffer.addEvent(event, isCheckout);
   } catch (error) {
     __DEBUG_BUILD__ && logger.error(error);
     replay.stop();