feat(replay): Change addEvent to be async (#6695)

billyvg · web-flow · commit 770a33dba89c · 2023-01-11T13:09:49.000-05:00
`addEvent` should be async as it waits for a response from Compression worker. We don't necessarily always await `addEvent()`, but it's more correct that it is marked as `async` and can maybe save some headaches down the line.
diff --git a/packages/replay/src/eventBuffer.ts b/packages/replay/src/eventBuffer.ts
@@ -2,10 +2,9 @@
 // TODO: figure out member access types and remove the line above
 
 import { captureException } from '@sentry/core';
-import type { ReplayRecordingData } from '@sentry/types';
 import { logger } from '@sentry/utils';
 
-import type { EventBuffer, RecordingEvent, WorkerRequest, WorkerResponse } from './types';
+import type { AddEventResult, EventBuffer, RecordingEvent, WorkerRequest } from './types';
 import workerString from './worker/worker.js';
 
 interface CreateEventBufferParams {
@@ -54,13 +53,14 @@ class EventBufferArray implements EventBuffer {
     this._events = [];
   }
 
-  public addEvent(event: RecordingEvent, isCheckout?: boolean): void {
+  public async addEvent(event: RecordingEvent, isCheckout?: boolean): Promise<AddEventResult> {
     if (isCheckout) {
       this._events = [event];
       return;
     }
 
     this._events.push(event);
+    return;
   }
 
   public finish(): Promise<string> {
@@ -107,8 +107,10 @@ export class EventBufferCompressionWorker implements EventBuffer {
 
   /**
    * Add an event to the event buffer.
+   *
+   * Returns true if event was successfuly received and processed by worker.
    */
-  public async addEvent(event: RecordingEvent, isCheckout?: boolean): Promise<ReplayRecordingData> {
+  public async addEvent(event: RecordingEvent, isCheckout?: boolean): Promise<AddEventResult> {
     if (isCheckout) {
       // This event is a checkout, make sure worker buffer is cleared before
       // proceeding.
@@ -132,7 +134,7 @@ export class EventBufferCompressionWorker implements EventBuffer {
   /**
    * Post message to worker and wait for response before resolving promise.
    */
-  private _postMessage({ id, method, args }: WorkerRequest): Promise<WorkerResponse['response']> {
+  private _postMessage<T>({ id, method, args }: WorkerRequest): Promise<T> {
     return new Promise((resolve, reject) => {
       // eslint-disable-next-line @typescript-eslint/explicit-function-return-type
       const listener = ({ data }: MessageEvent) => {
@@ -178,8 +180,8 @@ export class EventBufferCompressionWorker implements EventBuffer {
   /**
    * Send the event to the worker.
    */
-  private _sendEventToWorker(event: RecordingEvent): Promise<ReplayRecordingData> {
-    const promise = this._postMessage({
+  private async _sendEventToWorker(event: RecordingEvent): Promise<AddEventResult> {
+    const promise = this._postMessage<void>({
       id: this._getAndIncrementId(),
       method: 'addEvent',
       args: [event],
@@ -195,12 +197,12 @@ export class EventBufferCompressionWorker implements EventBuffer {
    * Finish the request and return the compressed data from the worker.
    */
   private async _finishRequest(id: number): Promise<Uint8Array> {
-    const promise = this._postMessage({ id, method: 'finish', args: [] });
+    const promise = this._postMessage<Uint8Array>({ id, method: 'finish', args: [] });
 
     // XXX: See note in `get length()`
     this._eventBufferItemLength = 0;
 
-    return promise as Promise<Uint8Array>;
+    return promise;
   }
 
   /** Get the current ID and increment it for the next call. */
diff --git a/packages/replay/src/replay.ts b/packages/replay/src/replay.ts
@@ -22,6 +22,7 @@ import { createEventBuffer } from './eventBuffer';
 import { getSession } from './session/getSession';
 import { saveSession } from './session/saveSession';
 import type {
+  AddEventResult,
   AddUpdateCallback,
   AllPerformanceEntry,
   EventBuffer,
@@ -450,7 +451,7 @@ export class ReplayContainer implements ReplayContainerInterface {
 
       // We need to clear existing events on a checkout, otherwise they are
       // incremental event updates and should be appended
-      addEvent(this, event, isCheckout);
+      void addEvent(this, event, isCheckout);
 
       // Different behavior for full snapshots (type=2), ignore other event types
       // See https://github.com/rrweb-io/rrweb/blob/d8f9290ca496712aa1e7d472549480c4e7876594/packages/rrweb/src/types.ts#L16
@@ -556,7 +557,7 @@ export class ReplayContainer implements ReplayContainerInterface {
       }
 
       this.addUpdate(() => {
-        addEvent(this, {
+        void addEvent(this, {
           type: EventType.Custom,
           // TODO: We were converting from ms to seconds for breadcrumbs, spans,
           // but maybe we should just keep them as milliseconds
@@ -674,7 +675,7 @@ export class ReplayContainer implements ReplayContainerInterface {
    */
   createCustomBreadcrumb(breadcrumb: Breadcrumb): void {
     this.addUpdate(() => {
-      addEvent(this, {
+      void addEvent(this, {
         type: EventType.Custom,
         timestamp: breadcrumb.timestamp || 0,
         data: {
@@ -689,12 +690,12 @@ export class ReplayContainer implements ReplayContainerInterface {
    * Observed performance events are added to `this.performanceEvents`. These
    * are included in the replay event before it is finished and sent to Sentry.
    */
-  addPerformanceEntries(): void {
+  addPerformanceEntries(): Promise<Array<AddEventResult | null>> {
     // Copy and reset entries before processing
     const entries = [...this.performanceEvents];
     this.performanceEvents = [];
 
-    createPerformanceSpans(this, createPerformanceEntries(entries));
+    return Promise.all(createPerformanceSpans(this, createPerformanceEntries(entries)));
   }
 
   /**
diff --git a/packages/replay/src/types.ts b/packages/replay/src/types.ts
@@ -48,9 +48,11 @@ export interface WorkerResponse {
   id: number;
   method: string;
   success: boolean;
-  response: ReplayRecordingData;
+  response: unknown;
 }
 
+export type AddEventResult = void;
+
 export interface SampleRates {
   /**
    * The sample rate for session-long replays. 1.0 will record all sessions and
@@ -210,8 +212,22 @@ export interface Session {
 
 export interface EventBuffer {
   readonly length: number;
+
+  /**
+   * Destroy the event buffer.
+   */
   destroy(): void;
-  addEvent(event: RecordingEvent, isCheckout?: boolean): void;
+
+  /**
+   * Add an event to the event buffer.
+   *
+   * Returns true if event was successfully added.
+   */
+  addEvent(event: RecordingEvent, isCheckout?: boolean): Promise<AddEventResult>;
+
+  /**
+   * Clears and returns the contents and the buffer.
+   */
   finish(): Promise<ReplayRecordingData>;
 }
 
diff --git a/packages/replay/src/util/addEvent.ts b/packages/replay/src/util/addEvent.ts
@@ -1,18 +1,22 @@
 import { SESSION_IDLE_DURATION } from '../constants';
-import type { RecordingEvent, ReplayContainer } from '../types';
+import type { AddEventResult, RecordingEvent, ReplayContainer } from '../types';
 
 /**
  * Add an event to the event buffer
  */
-export function addEvent(replay: ReplayContainer, event: RecordingEvent, isCheckout?: boolean): void {
+export async function addEvent(
+  replay: ReplayContainer,
+  event: RecordingEvent,
+  isCheckout?: boolean,
+): Promise<AddEventResult | null> {
   if (!replay.eventBuffer) {
     // This implies that `_isEnabled` is false
-    return;
+    return null;
   }
 
   if (replay.isPaused()) {
     // Do not add to event buffer when recording is paused
-    return;
+    return null;
   }
 
   // TODO: sadness -- we will want to normalize timestamps to be in ms -
@@ -25,7 +29,7 @@ export function addEvent(replay: ReplayContainer, event: RecordingEvent, isCheck
   // comes back to trigger a new session. The performance entries rely on
   // `performance.timeOrigin`, which is when the page first opened.
   if (timestampInMs + SESSION_IDLE_DURATION < new Date().getTime()) {
-    return;
+    return null;
   }
 
   // Only record earliest event if a new session was created, otherwise it
@@ -35,5 +39,5 @@ export function addEvent(replay: ReplayContainer, event: RecordingEvent, isCheck
     replay.getContext().earliestEvent = timestampInMs;
   }
 
-  replay.eventBuffer.addEvent(event, isCheckout);
+  return replay.eventBuffer.addEvent(event, isCheckout);
 }
diff --git a/packages/replay/src/util/addMemoryEntry.ts b/packages/replay/src/util/addMemoryEntry.ts
@@ -1,5 +1,5 @@
 import { WINDOW } from '../constants';
-import type { ReplayContainer, ReplayPerformanceEntry } from '../types';
+import type { AddEventResult, ReplayContainer, ReplayPerformanceEntry } from '../types';
 import { createPerformanceSpans } from './createPerformanceSpans';
 
 type ReplayMemoryEntry = ReplayPerformanceEntry & { data: { memory: MemoryInfo } };
@@ -14,15 +14,18 @@ interface MemoryInfo {
  * Create a "span" for the total amount of memory being used by JS objects
  * (including v8 internal objects).
  */
-export function addMemoryEntry(replay: ReplayContainer): void {
+export async function addMemoryEntry(replay: ReplayContainer): Promise<Array<AddEventResult | null>> {
   // window.performance.memory is a non-standard API and doesn't work on all browsers, so we try-catch this
   try {
-    createPerformanceSpans(replay, [
-      // @ts-ignore memory doesn't exist on type Performance as the API is non-standard (we check that it exists above)
-      createMemoryEntry(WINDOW.performance.memory),
-    ]);
+    return Promise.all(
+      createPerformanceSpans(replay, [
+        // @ts-ignore memory doesn't exist on type Performance as the API is non-standard (we check that it exists above)
+        createMemoryEntry(WINDOW.performance.memory),
+      ]),
+    );
   } catch (error) {
     // Do nothing
+    return [];
   }
 }
 
diff --git a/packages/replay/src/util/createPerformanceSpans.ts b/packages/replay/src/util/createPerformanceSpans.ts
@@ -1,13 +1,16 @@
 import { EventType } from 'rrweb';
 
-import type { ReplayContainer, ReplayPerformanceEntry } from '../types';
+import type { AddEventResult, ReplayContainer, ReplayPerformanceEntry } from '../types';
 import { addEvent } from './addEvent';
 
 /**
  * Create a "span" for each performance entry. The parent transaction is `this.replayEvent`.
  */
-export function createPerformanceSpans(replay: ReplayContainer, entries: ReplayPerformanceEntry[]): void {
-  entries.map(({ type, start, end, name, data }) =>
+export function createPerformanceSpans(
+  replay: ReplayContainer,
+  entries: ReplayPerformanceEntry[],
+): Promise<AddEventResult | null>[] {
+  return entries.map(({ type, start, end, name, data }) =>
     addEvent(replay, {
       type: EventType.Custom,
       timestamp: start,
diff --git a/packages/replay/src/worker/worker.js b/packages/replay/src/worker/worker.js
diff --git a/packages/replay/test/integration/flush.test.ts b/packages/replay/test/integration/flush.test.ts
@@ -182,37 +182,39 @@ describe('Integration | flush', () => {
     });
 
     // Add this to test that segment ID increases
-    mockAddPerformanceEntries.mockImplementationOnce(() => {
-      createPerformanceSpans(
-        replay,
-        createPerformanceEntries([
-          {
-            name: 'https://sentry.io/foo.js',
-            entryType: 'resource',
-            startTime: 176.59999990463257,
-            duration: 5.600000023841858,
-            initiatorType: 'link',
-            nextHopProtocol: 'h2',
-            workerStart: 177.5,
-            redirectStart: 0,
-            redirectEnd: 0,
-            fetchStart: 177.69999992847443,
-            domainLookupStart: 177.69999992847443,
-            domainLookupEnd: 177.69999992847443,
-            connectStart: 177.69999992847443,
-            connectEnd: 177.69999992847443,
-            secureConnectionStart: 177.69999992847443,
-            requestStart: 177.5,
-            responseStart: 181,
-            responseEnd: 182.19999992847443,
-            transferSize: 0,
-            encodedBodySize: 0,
-            decodedBodySize: 0,
-            serverTiming: [],
-          } as unknown as PerformanceResourceTiming,
-        ]),
-      );
-    });
+    mockAddPerformanceEntries.mockImplementationOnce(() =>
+      Promise.all(
+        createPerformanceSpans(
+          replay,
+          createPerformanceEntries([
+            {
+              name: 'https://sentry.io/foo.js',
+              entryType: 'resource',
+              startTime: 176.59999990463257,
+              duration: 5.600000023841858,
+              initiatorType: 'link',
+              nextHopProtocol: 'h2',
+              workerStart: 177.5,
+              redirectStart: 0,
+              redirectEnd: 0,
+              fetchStart: 177.69999992847443,
+              domainLookupStart: 177.69999992847443,
+              domainLookupEnd: 177.69999992847443,
+              connectStart: 177.69999992847443,
+              connectEnd: 177.69999992847443,
+              secureConnectionStart: 177.69999992847443,
+              requestStart: 177.5,
+              responseStart: 181,
+              responseEnd: 182.19999992847443,
+              transferSize: 0,
+              encodedBodySize: 0,
+              decodedBodySize: 0,
+              serverTiming: [],
+            } as unknown as PerformanceResourceTiming,
+          ]),
+        ),
+      ),
+    );
     // flush #5 @ t=25s - debounced flush calls `flush`
     // 20s + `flushMinDelay` which is 5 seconds
     await advanceTimers(DEFAULT_FLUSH_MIN_DELAY);
diff --git a/packages/replay/test/unit/eventBuffer.test.ts b/packages/replay/test/unit/eventBuffer.test.ts
@@ -53,6 +53,8 @@ describe('Unit | eventBuffer', () => {
     }) as EventBufferCompressionWorker;
 
     buffer.addEvent(TEST_EVENT);
+    // @ts-ignore make sure it handles invalid data
+    buffer.addEvent(undefined);
     buffer.addEvent(TEST_EVENT);
 
     const result = await buffer.finish();
diff --git a/packages/replay/test/unit/worker/Compressor.test.ts b/packages/replay/test/unit/worker/Compressor.test.ts
@@ -26,28 +26,16 @@ describe('Unit | worker | Compressor', () => {
     expect(restored).toBe(JSON.stringify(events));
   });
 
-  it('ignores undefined events', () => {
+  it('throws on invalid/undefined events', () => {
     const compressor = new Compressor();
 
-    const events = [
-      {
-        id: 1,
-        foo: ['bar', 'baz'],
-      },
-      undefined,
-      {
-        id: 2,
-        foo: [false],
-      },
-    ] as Record<string, any>[];
-
-    events.forEach(event => compressor.addEvent(event));
+    // @ts-ignore ignoring type for test
+    expect(() => void compressor.addEvent(undefined)).toThrow();
 
     const compressed = compressor.finish();
 
     const restored = pako.inflate(compressed, { to: 'string' });
 
-    const expected = [events[0], events[2]];
-    expect(restored).toBe(JSON.stringify(expected));
+    expect(restored).toBe(JSON.stringify([]));
   });
 });
diff --git a/packages/replay/worker/src/Compressor.ts b/packages/replay/worker/src/Compressor.ts
@@ -27,18 +27,17 @@ export class Compressor {
 
   public addEvent(data: Record<string, unknown>): void {
     if (!data) {
-      return;
+      throw new Error('Adding invalid event');
     }
     // If the event is not the first event, we need to prefix it with a `,` so
     // that we end up with a list of events
     const prefix = this.added > 0 ? ',' : '';
     // TODO: We may want Z_SYNC_FLUSH or Z_FULL_FLUSH (not sure the difference)
     // Using NO_FLUSH here for now as we can create many attachments that our
     // web UI will get API rate limited.
-    this.deflate.push(prefix + JSON.stringify(data), constants.Z_NO_FLUSH);
-    this.added++;
+    this.deflate.push(prefix + JSON.stringify(data), constants.Z_SYNC_FLUSH);
 
-    return;
+    this.added++;
   }
 
   public finish(): Uint8Array {
diff --git a/packages/replay/worker/src/handleMessage.ts b/packages/replay/worker/src/handleMessage.ts
