feat(replay): Track pending events in EventBuffer

* Track raw pending events in `EventBuffer`. This can be helpful in the case
where page is reloaded before the worker has a chance to close the compression
stream.
* Change `EventBuffer.length` to `EventBuffer.pendingLength` to better reflect
what it is. In the case of compression worker, it is async, so recent events
added to the buffer are not necessarily present in the workers compression
stream.

Author: billyvg

packages/replay/src/eventBuffer.ts

@@ -45,10 +45,18 @@ class EventBufferArray implements EventBuffer {
     this._events = [];
   }
 
-  public get length(): number {
+  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 destroy(): void {
     this._events = [];
   }
@@ -80,6 +88,13 @@ class EventBufferArray implements EventBuffer {
  * 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[] = [];
+
   private _worker: null | Worker;
   private _eventBufferItemLength: number = 0;
   private _id: number = 0;
@@ -89,13 +104,21 @@ export class EventBufferCompressionWorker implements EventBuffer {
   }
 
   /**
-   * Note that this may not reflect what is actually in the event buffer. This
-   * is only a local count of the buffer size since `addEvent` is async.
+   * 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 length(): number {
+  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;
+  }
+
   /**
    * Destroy the event buffer.
    */
@@ -121,6 +144,11 @@ export class EventBufferCompressionWorker implements EventBuffer {
       });
     }
 
+    // Don't store checkout events in `_pendingEvents` because they are too large
+    if (!isCheckout) {
+      this._pendingEvents.push(event);
+    }
+
     return this._sendEventToWorker(event);
   }
 
@@ -202,6 +230,9 @@ export class EventBufferCompressionWorker implements EventBuffer {
     // XXX: See note in `get length()`
     this._eventBufferItemLength = 0;
 
+    await promise
+
+    this._pendingEvents = [];
     return promise as Promise<Uint8Array>;
   }
 

packages/replay/src/types.ts

@@ -211,7 +211,15 @@ export interface Session {
 }
 
 export interface EventBuffer {
-  readonly length: number;
+  /**
+   * The number of raw events that are buffered
+   */
+  readonly pendingLength: number;
+
+  /**
+   * The raw events that are buffered.
+   */
+  readonly pendingEvents: RecordingEvent[];
 
   /**
    * Destroy the event buffer.
@@ -226,7 +234,7 @@ export interface EventBuffer {
   addEvent(event: RecordingEvent, isCheckout?: boolean): Promise<WorkerAddEventResponse>;
 
   /**
-   * Clears and returns the contents and the buffer.
+   * Clears and returns the contents of the buffer.
    */
   finish(): Promise<ReplayRecordingData>;
 }