ref(replay): Move earliest timestamp tracking to eventBuffer

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: [],