ref: Use partitionedqueue for event buffer

Author: mydea

packages/replay/src/eventBuffer/EventBufferArray.ts

@@ -1,39 +1,32 @@
 import type { ReplayRecordingData } from '@sentry/types';
 
 import type { EventBuffer, RecordingEvent } from '../types';
-
-interface EventsGroup {
-  checkoutTimestamp: number;
-  events: RecordingEvent[];
-}
+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: EventsGroup[];
-  private _eventsFlat: RecordingEvent[];
+  private _events: PartitionedQueue<RecordingEvent>;
 
   public constructor() {
-    this._events = [];
-    this._eventsFlat = [];
+    this._events = new PartitionedQueue<RecordingEvent>();
   }
 
   /** @inheritdoc */
   public get pendingLength(): number {
-    return this.pendingEvents.length;
+    return this._events.getLength();
   }
 
   /** @inheritdoc */
   public get pendingEvents(): RecordingEvent[] {
-    return this._eventsFlat;
-    // return this._events.reduce((acc, { events }) => [...events, ...acc], [] as RecordingEvent[]);
+    return this._events.getItems();
   }
 
   /** @inheritdoc */
-  public getFirstCheckoutTimestamp(): number | null {
-    return (this._events[0] && this._events[0].checkoutTimestamp) || null;
+  public getEarliestTimestamp(): number | null {
+    return this.pendingEvents.map(event => event.timestamp).sort()[0] || null;
   }
 
   /** @inheritdoc */
@@ -43,40 +36,17 @@ export class EventBufferArray implements EventBuffer {
 
   /** @inheritdoc */
   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);
-    }
-
-    this._eventsFlat.push(event);
+    this._events.add(event, isCheckout);
   }
 
   /** @inheritdoc */
   public clear(keepLastCheckout?: boolean): void {
-    if (keepLastCheckout) {
-      this._events.splice(1);
-
-      if (this._events.length === 0) {
-        this._eventsFlat = [];
-      } else {
-        // Remove all events from the flat array that are not in the first group
-        const firstGroup = this._events[0];
-        this._eventsFlat = this._eventsFlat.filter(event => firstGroup.events.includes(event));
-      }
-    } else {
-      this._events = [];
-      this._eventsFlat = [];
-    }
+    this._events.clear(keepLastCheckout);
   }
 
   /** @inheritdoc */
   public finish(): Promise<ReplayRecordingData> {
-    const pendingEvents = this.pendingEvents.slice();
+    const { pendingEvents } = this;
     this.clear();
 
     return Promise.resolve(this._finishRecording(pendingEvents));

packages/replay/src/eventBuffer/EventBufferProxy.ts

@@ -59,8 +59,8 @@ export class EventBufferProxy implements EventBuffer {
   }
 
   /** @inheritdoc */
-  public getFirstCheckoutTimestamp(): number | null {
-    return this._used.getFirstCheckoutTimestamp();
+  public getEarliestTimestamp(): number | null {
+    return this._used.getEarliestTimestamp();
   }
 
   /** Ensure the worker has loaded. */

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/types.ts

@@ -292,7 +292,7 @@ export interface EventBuffer {
   finish(): Promise<ReplayRecordingData>;
 
   /** Get the timestamp of the first checkout that is pending. */
-  getFirstCheckoutTimestamp(): number | null;
+  getEarliestTimestamp(): number | null;
 }
 
 export type AddUpdateCallback = () => boolean | void;

packages/replay/src/util/addEvent.ts

@@ -43,7 +43,7 @@ export function addEvent(replay: ReplayContainer, event: RecordingEvent, isCheck
 
       // Ensure we have the correct first checkout timestamp when an error occurs
       if (!session.segmentId) {
-        replay.getContext().earliestEvent = eventBuffer.getFirstCheckoutTimestamp();
+        replay.getContext().earliestEvent = eventBuffer.getEarliestTimestamp();
       }
     } else {
       eventBuffer.clear();

packages/replay/test/unit/eventBuffer/PartitionedQueue.test.ts

@@ -0,0 +1,58 @@
+import { PartitionedQueue } from '../../../src/eventBuffer/PartitionedQueue';
+
+describe('Unit | eventBuffer | PartitionedQueue', () => {
+  it('works with empty queue', () => {
+    const queue = new PartitionedQueue<string>();
+
+    expect(queue.getItems()).toEqual([]);
+    expect(queue.getLength()).toEqual(0);
+
+    queue.clear();
+
+    expect(queue.getItems()).toEqual([]);
+    expect(queue.getLength()).toEqual(0);
+
+    queue.clear(true);
+
+    expect(queue.getItems()).toEqual([]);
+    expect(queue.getLength()).toEqual(0);
+  });
+
+  it('allows to add records', () => {
+    const queue = new PartitionedQueue<string>();
+
+    queue.add('one');
+    queue.add('two');
+    queue.add('three');
+
+    expect(queue.getItems()).toEqual(['one', 'two', 'three']);
+    expect(queue.getLength()).toEqual(3);
+
+    queue.clear();
+
+    expect(queue.getItems()).toEqual([]);
+    expect(queue.getLength()).toEqual(0);
+  });
+
+  it('allows to add records with checkouts', () => {
+    const queue = new PartitionedQueue<string>();
+
+    queue.add('one');
+    queue.add('two');
+    queue.add('three', true);
+    queue.add('four');
+
+    expect(queue.getItems()).toEqual(['one', 'two', 'three', 'four']);
+    expect(queue.getLength()).toEqual(4);
+
+    queue.clear(true);
+
+    expect(queue.getItems()).toEqual(['three', 'four']);
+    expect(queue.getLength()).toEqual(2);
+
+    queue.clear(true);
+
+    expect(queue.getItems()).toEqual(['three', 'four']);
+    expect(queue.getLength()).toEqual(2);
+  });
+});

packages/replay/test/unit/util/addEvent.test.ts

@@ -6,33 +6,33 @@ import { useFakeTimers } from '../../utils/use-fake-timers';
 useFakeTimers();
 
 describe('Unit | util | addEvent', () => {
-  it('clears queue after two checkouts in error mode', async function () {
+  it('clears queue after two checkouts in error mode xxx', async function () {
     jest.setSystemTime(BASE_TIMESTAMP);
 
     const replay = setupReplayContainer();
     replay.recordingMode = 'error';
 
-    await addEvent(replay, { data: {}, timestamp: BASE_TIMESTAMP, type: 2 }, false);
-    await addEvent(replay, { data: {}, timestamp: BASE_TIMESTAMP + 10, type: 3 });
-    await addEvent(replay, { data: {}, timestamp: BASE_TIMESTAMP + 100, type: 2 }, true);
+    addEvent(replay, { data: {}, timestamp: BASE_TIMESTAMP + 10, type: 2 }, false);
+    addEvent(replay, { data: {}, timestamp: BASE_TIMESTAMP + 0, type: 3 });
+    addEvent(replay, { data: {}, timestamp: BASE_TIMESTAMP + 100, type: 2 }, true);
 
     expect(replay.getContext().earliestEvent).toEqual(BASE_TIMESTAMP);
     expect(replay.eventBuffer?.pendingEvents).toEqual([
-      { data: {}, timestamp: BASE_TIMESTAMP, type: 2 },
-      { data: {}, timestamp: BASE_TIMESTAMP + 10, type: 3 },
+      { data: {}, timestamp: BASE_TIMESTAMP + 10, type: 2 },
+      { data: {}, timestamp: BASE_TIMESTAMP, type: 3 },
       { data: {}, timestamp: BASE_TIMESTAMP + 100, type: 2 },
     ]);
 
-    await addEvent(replay, { data: {}, timestamp: BASE_TIMESTAMP + 200, type: 2 }, true);
+    addEvent(replay, { data: {}, timestamp: BASE_TIMESTAMP + 200, type: 2 }, true);
 
     expect(replay.getContext().earliestEvent).toEqual(BASE_TIMESTAMP + 100);
     expect(replay.eventBuffer?.pendingEvents).toEqual([
       { data: {}, timestamp: BASE_TIMESTAMP + 100, type: 2 },
       { data: {}, timestamp: BASE_TIMESTAMP + 200, type: 2 },
     ]);
 
-    await addEvent(replay, { data: {}, timestamp: BASE_TIMESTAMP + 250, type: 3 }, false);
-    await addEvent(replay, { data: {}, timestamp: BASE_TIMESTAMP + 300, type: 2 }, true);
+    addEvent(replay, { data: {}, timestamp: BASE_TIMESTAMP + 250, type: 3 }, false);
+    addEvent(replay, { data: {}, timestamp: BASE_TIMESTAMP + 300, type: 2 }, true);
 
     expect(replay.getContext().earliestEvent).toEqual(BASE_TIMESTAMP + 200);
     expect(replay.eventBuffer?.pendingEvents).toEqual([