ref(replay): Move earliest timestamp tracking to eventBuffer (#7983)

In preparation to allow us to handle this better for buffering sessions.

Author: mydea

packages/replay/src/eventBuffer/EventBufferArray.ts

@@ -1,4 +1,5 @@
 import type { AddEventResult, EventBuffer, RecordingEvent } from '../types';
+import { timestampToMs } from '../util/timestampToMs';
 
 /**
  * A basic event buffer that does not do any compression.
@@ -44,4 +45,15 @@ export class EventBufferArray implements EventBuffer {
       resolve(JSON.stringify(eventsRet));
     });
   }
+
+  /** @inheritdoc */
+  public getEarliestTimestamp(): number | null {
+    const timestamp = this.events.map(event => event.timestamp).sort()[0];
+
+    if (!timestamp) {
+      return null;
+    }
+
+    return timestampToMs(timestamp);
+  }
 }

packages/replay/src/eventBuffer/EventBufferCompressionWorker.ts

@@ -1,21 +1,25 @@
 import type { ReplayRecordingData } from '@sentry/types';
 
 import type { AddEventResult, EventBuffer, RecordingEvent } from '../types';
+import { timestampToMs } from '../util/timestampToMs';
 import { WorkerHandler } from './WorkerHandler';
 
 /**
  * Event buffer that uses a web worker to compress events.
  * Exported only for testing.
  */
 export class EventBufferCompressionWorker implements EventBuffer {
-  /** @inheritdoc */
-  public hasEvents: boolean;
-
   private _worker: WorkerHandler;
+  private _earliestTimestamp: number | null;
 
   public constructor(worker: Worker) {
     this._worker = new WorkerHandler(worker);
-    this.hasEvents = false;
+    this._earliestTimestamp = null;
+  }
+
+  /** @inheritdoc */
+  public get hasEvents(): boolean {
+    return !!this._earliestTimestamp;
   }
 
   /**
@@ -39,14 +43,17 @@ export class EventBufferCompressionWorker implements EventBuffer {
    * Returns true if event was successfuly received and processed by worker.
    */
   public async addEvent(event: RecordingEvent, isCheckout?: boolean): Promise<AddEventResult> {
-    this.hasEvents = true;
-
     if (isCheckout) {
       // This event is a checkout, make sure worker buffer is cleared before
       // proceeding.
       await this._clear();
     }
 
+    const timestamp = timestampToMs(event.timestamp);
+    if (!this._earliestTimestamp || timestamp < this._earliestTimestamp) {
+      this._earliestTimestamp = timestamp;
+    }
+
     return this._sendEventToWorker(event);
   }
 
@@ -57,6 +64,11 @@ export class EventBufferCompressionWorker implements EventBuffer {
     return this._finishRequest();
   }
 
+  /** @inheritdoc */
+  public getEarliestTimestamp(): number | null {
+    return this._earliestTimestamp;
+  }
+
   /**
    * Send the event to the worker.
    */
@@ -70,13 +82,14 @@ export class EventBufferCompressionWorker implements EventBuffer {
   private async _finishRequest(): Promise<Uint8Array> {
     const response = await this._worker.postMessage<Uint8Array>('finish');
 
-    this.hasEvents = false;
+    this._earliestTimestamp = null;
 
     return response;
   }
 
   /** Clear any pending events from the worker. */
   private _clear(): Promise<void> {
+    this._earliestTimestamp = null;
     return this._worker.postMessage('clear');
   }
 }

packages/replay/src/eventBuffer/EventBufferProxy.ts

@@ -35,6 +35,11 @@ export class EventBufferProxy implements EventBuffer {
     this._compression.destroy();
   }
 
+  /** @inheritdoc */
+  public getEarliestTimestamp(): number | null {
+    return this._used.getEarliestTimestamp();
+  }
+
   /**
    * Add an event to the event buffer.
    *

packages/replay/src/replay.ts

@@ -118,7 +118,6 @@ export class ReplayContainer implements ReplayContainerInterface {
     errorIds: new Set(),
     traceIds: new Set(),
     urls: [],
-    earliestEvent: null,
     initialTimestamp: Date.now(),
     initialUrl: '',
   };
@@ -819,22 +818,35 @@ export class ReplayContainer implements ReplayContainerInterface {
     this._context.errorIds.clear();
     this._context.traceIds.clear();
     this._context.urls = [];
-    this._context.earliestEvent = null;
+  }
+
+  /** Update the initial timestamp based on the buffer content. */
+  private _updateInitialTimestampFromEventBuffer(): void {
+    const { session, eventBuffer } = this;
+    if (!session || !eventBuffer) {
+      return;
+    }
+
+    // we only ever update this on the initial segment
+    if (session.segmentId) {
+      return;
+    }
+
+    const earliestEvent = eventBuffer.getEarliestTimestamp();
+    if (earliestEvent && earliestEvent < this._context.initialTimestamp) {
+      this._context.initialTimestamp = earliestEvent;
+    }
   }
 
   /**
    * Return and clear _context
    */
   private _popEventContext(): PopEventContext {
-    if (this._context.earliestEvent && this._context.earliestEvent < this._context.initialTimestamp) {
-      this._context.initialTimestamp = this._context.earliestEvent;
-    }
-
     const _context = {
       initialTimestamp: this._context.initialTimestamp,
       initialUrl: this._context.initialUrl,
-      errorIds: Array.from(this._context.errorIds).filter(Boolean),
-      traceIds: Array.from(this._context.traceIds).filter(Boolean),
+      errorIds: Array.from(this._context.errorIds),
+      traceIds: Array.from(this._context.traceIds),
       urls: this._context.urls,
     };
 
@@ -873,6 +885,9 @@ export class ReplayContainer implements ReplayContainerInterface {
     }
 
     try {
+      // This uses the data from the eventBuffer, so we need to call this before `finish()
+      this._updateInitialTimestampFromEventBuffer();
+
       // Note this empties the event buffer regardless of outcome of sending replay
       const recordingData = await this.eventBuffer.finish();
 

packages/replay/src/types.ts

@@ -360,7 +360,7 @@ interface CommonEventContext {
   initialUrl: string;
 
   /**
-   * The initial starting timestamp of the session
+   * The initial starting timestamp in ms of the session.
    */
   initialTimestamp: number;
 
@@ -395,11 +395,6 @@ export interface InternalEventContext extends CommonEventContext {
    * Set of Sentry trace ids that have occurred during a replay segment
    */
   traceIds: Set<string>;
-
-  /**
-   * The timestamp of the earliest event that has been added to event buffer. This can happen due to the Performance Observer which buffers events.
-   */
-  earliestEvent: number | null;
 }
 
 export type Sampled = false | 'session' | 'buffer';
@@ -408,12 +403,12 @@ export interface Session {
   id: string;
 
   /**
-   * Start time of current session
+   * Start time of current session (in ms)
    */
   started: number;
 
   /**
-   * Last known activity of the session
+   * Last known activity of the session (in ms)
    */
   lastActivity: number;
 
@@ -463,6 +458,11 @@ export interface EventBuffer {
    * Clears and returns the contents of the buffer.
    */
   finish(): Promise<ReplayRecordingData>;
+
+  /**
+   * Get the earliest timestamp in ms of any event currently in the buffer.
+   */
+  getEarliestTimestamp(): number | null;
 }
 
 export type AddUpdateCallback = () => boolean | void;

packages/replay/src/util/addEvent.ts

@@ -2,6 +2,7 @@ import { getCurrentHub } from '@sentry/core';
 import { logger } from '@sentry/utils';
 
 import type { AddEventResult, RecordingEvent, ReplayContainer } from '../types';
+import { timestampToMs } from './timestampToMs';
 
 /**
  * Add an event to the event buffer.
@@ -22,10 +23,7 @@ export async function addEvent(
     return null;
   }
 
-  // TODO: sadness -- we will want to normalize timestamps to be in ms -
-  // requires coordination with frontend
-  const isMs = event.timestamp > 9999999999;
-  const timestampInMs = isMs ? event.timestamp : event.timestamp * 1000;
+  const timestampInMs = timestampToMs(event.timestamp);
 
   // Throw out events that happen more than 5 minutes ago. This can happen if
   // page has been left open and idle for a long period of time and user
@@ -35,13 +33,6 @@ export async function addEvent(
     return null;
   }
 
-  // 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)) {
-    replay.getContext().earliestEvent = timestampInMs;
-  }
-
   try {
     return await replay.eventBuffer.addEvent(event, isCheckout);
   } catch (error) {

packages/replay/src/util/handleRecordingEmit.ts

@@ -58,10 +58,10 @@ export function getHandleRecordingEmit(replay: ReplayContainer): RecordingEmitCa
         return true;
       }
 
-      // See note above re: session start needs to reflect the most recent
-      // checkout.
-      if (replay.recordingMode === 'buffer' && replay.session) {
-        const { earliestEvent } = replay.getContext();
+      // When in buffer mode, make sure we adjust the session started date to the current earliest event of the buffer
+      // this should usually be the timestamp of the checkout event, but to be safe...
+      if (replay.recordingMode === 'buffer' && replay.session && replay.eventBuffer) {
+        const earliestEvent = replay.eventBuffer.getEarliestTimestamp();
         if (earliestEvent) {
           replay.session.started = earliestEvent;
 

packages/replay/src/util/timestampToMs.ts

@@ -0,0 +1,7 @@
+/**
+ * Converts a timestamp to ms, if it was in s, or keeps it as ms.
+ */
+export function timestampToMs(timestamp: number): number {
+  const isMs = timestamp > 9999999999;
+  return isMs ? timestamp : timestamp * 1000;
+}

packages/replay/test/integration/events.test.ts

@@ -156,7 +156,7 @@ describe('Integration | events', () => {
     );
 
     // This should be null because `addEvent` has not been called yet
-    expect(replay.getContext().earliestEvent).toBe(null);
+    expect(replay.eventBuffer?.getEarliestTimestamp()).toBe(null);
     expect(mockTransportSend).toHaveBeenCalledTimes(0);
 
     // A new checkout occurs (i.e. a new session was started)
@@ -196,6 +196,6 @@ describe('Integration | events', () => {
     });
 
     // This gets reset after sending replay
-    expect(replay.getContext().earliestEvent).toBe(null);
+    expect(replay.eventBuffer?.getEarliestTimestamp()).toBe(null);
   });
 });

packages/replay/test/integration/sampling.test.ts

@@ -24,13 +24,13 @@ describe('Integration | sampling', () => {
     jest.runAllTimers();
 
     expect(replay.session).toBe(undefined);
+    expect(replay.eventBuffer).toBeNull();
 
     // This is what the `_context` member is initialized with
     expect(replay.getContext()).toEqual({
       errorIds: new Set(),
       traceIds: new Set(),
       urls: [],
-      earliestEvent: null,
       initialTimestamp: expect.any(Number),
       initialUrl: '',
     });
@@ -63,11 +63,12 @@ describe('Integration | sampling', () => {
     jest.runAllTimers();
 
     expect(replay.session?.id).toBeDefined();
+    expect(replay.eventBuffer).toBeDefined();
+    expect(replay.eventBuffer?.getEarliestTimestamp()).toEqual(expect.any(Number));
 
     // This is what the `_context` member is initialized with
     expect(replay.getContext()).toEqual({
       errorIds: new Set(),
-      earliestEvent: expect.any(Number),
       initialTimestamp: expect.any(Number),
       initialUrl: 'http://localhost/',
       traceIds: new Set(),

packages/replay/test/integration/session.test.ts

@@ -221,9 +221,11 @@ describe('Integration | session', () => {
       ]),
     });
 
+    // Earliest event is reset
+    expect(replay.eventBuffer?.getEarliestTimestamp()).toBeNull();
+
     // `_context` should be reset when a new session is created
     expect(replay.getContext()).toEqual({
-      earliestEvent: null,
       initialUrl: 'http://dummy/',
       initialTimestamp: newTimestamp,
       urls: [],