ref(replay): Try to make flush fully sync

Author: mydea

packages/replay/src/eventBuffer.ts

@@ -71,17 +71,11 @@ class EventBufferArray implements EventBuffer {
     return;
   }
 
-  public finish(): Promise<string> {
-    return new Promise<string>(resolve => {
-      resolve(this._finish());
-    });
-  }
-
-  public finishImmediate(): string {
-    return this._finish();
+  public async finish(): Promise<string> {
+    return this.finishSync();
   }
 
-  private _finish(): string {
+  public finishSync(): string {
     // Make a copy of the events array reference and immediately clear the
     // events member so that we do not lose new events while uploading
     // attachment.
@@ -169,7 +163,7 @@ export class EventBufferCompressionWorker implements EventBuffer {
   /**
    * Finish the event buffer and return the pending events.
    */
-  public finishImmediate(): string {
+  public finishSync(): string {
     const events = this._pendingEvents;
 
     // Ensure worker is still in a good state and disregard the result

packages/replay/src/replay.ts

@@ -625,7 +625,7 @@ export class ReplayContainer implements ReplayContainerInterface {
     // Send replay when the page/tab becomes hidden. There is no reason to send
     // replay if it becomes visible, since no actions we care about were done
     // while it was hidden
-    this._conditionalFlush({ finishImmediate: true });
+    this._conditionalFlush({ sync: true });
   }
 
   /**
@@ -756,8 +756,8 @@ export class ReplayContainer implements ReplayContainerInterface {
      * Page is likely to unload so need to bypass debounce completely and
      * synchronously retrieve pending events from buffer and send request asap.
      */
-    if (options.finishImmediate) {
-      void this._runFlush(options);
+    if (options.sync) {
+      this._flushSync();
       return;
     }
 
@@ -804,81 +804,143 @@ export class ReplayContainer implements ReplayContainerInterface {
    *
    * Should never be called directly, only by `flush`
    */
-  private async _runFlush(options: FlushOptions = {}): Promise<void> {
-    if (!this.session || !this.eventBuffer) {
-      __DEBUG_BUILD__ && logger.error('[Replay] No session or eventBuffer found to flush.');
-      return;
-    }
-
+  private async _runFlush(): Promise<void> {
     try {
-      this._debouncedFlush.cancel();
+      const flushData = this._prepareFlush();
+
+      if (!flushData) {
+        return;
+      }
 
-      const promises: Promise<any>[] = [];
+      const { promises, replayId, segmentId, eventContext, eventBuffer, session } = flushData;
 
-      promises.push(this._addPerformanceEntries());
+      // NOTE: Be mindful that nothing after this point (the first `await`)
+      // will run after when the page is unloaded.
+      await Promise.all(promises);
 
-      // Do not continue if there are no pending events in buffer
-      if (!this.eventBuffer || !this.eventBuffer.pendingLength) {
+      // This can be empty due to blur events calling `runFlush` directly. In
+      // the case where we have a snapshot checkout and a blur event
+      // happening near the same time, the blur event can end up emptying the
+      // buffer even if snapshot happens first.
+      if (!eventBuffer.pendingLength) {
         return;
       }
 
-      // Only attach memory entry if eventBuffer is not empty
-      promises.push(addMemoryEntry(this));
+      // This empties the event buffer regardless of outcome of sending replay
+      const recordingData = await eventBuffer.finish();
 
-      // NOTE: Copy values from instance members, as it's possible they could
-      // change before the flush finishes.
-      const replayId = this.session.id;
-      const eventContext = this._popEventContext();
-      // Always increment segmentId regardless of outcome of sending replay
-      const segmentId = this.session.segmentId++;
-
-      // Save session (new segment id) after we save flush data assuming either
-      // 1) request succeeds or 2) it fails or never happens, in which case we
-      // need to retry this segment.
-      this._maybeSaveSession();
+      await sendReplay({
+        replayId,
+        recordingData,
+        segmentId,
+        includeReplayStartTimestamp: segmentId === 0,
+        eventContext,
+        session,
+        options: this.getOptions(),
+        timestamp: new Date().getTime(),
+      });
+    } catch (err) {
+      this._handleSendError(err);
+    }
+  }
 
-      let recordingData: ReplayRecordingData;
+  /**
+   * Flush event buffer synchonously.
+   * This is necessary e.g. when running flush on page unload or similar.
+   */
+  private _flushSync(): void {
+    try {
+      const flushData = this._prepareFlush();
 
-      if (options.finishImmediate && this.eventBuffer.pendingLength) {
-        recordingData = this.eventBuffer.finishImmediate();
-      } else {
-        // NOTE: Be mindful that nothing after this point (the first `await`)
-        // will run after when the page is unloaded.
-        await Promise.all(promises);
-
-        // This can be empty due to blur events calling `runFlush` directly. In
-        // the case where we have a snapshot checkout and a blur event
-        // happening near the same time, the blur event can end up emptying the
-        // buffer even if snapshot happens first.
-        if (!this.eventBuffer.pendingLength) {
-          return;
-        }
-        // This empties the event buffer regardless of outcome of sending replay
-        recordingData = await this.eventBuffer.finish();
+      if (!flushData) {
+        return;
       }
 
-      await sendReplay({
+      const { replayId, segmentId, eventContext, eventBuffer, session } = flushData;
+
+      const recordingData = eventBuffer.finishSync();
+
+      sendReplay({
         replayId,
         recordingData,
         segmentId,
         includeReplayStartTimestamp: segmentId === 0,
         eventContext,
-        session: this.session,
+        session,
         options: this.getOptions(),
         timestamp: new Date().getTime(),
+      }).catch(err => {
+        this._handleSendError(err);
       });
     } catch (err) {
-      this._handleException(err);
+      this._handleSendError(err);
+    }
+  }
 
-      if (err instanceof RateLimitError) {
-        this._handleRateLimit(err.rateLimits);
-        return;
+  /** Prepare flush data */
+  private _prepareFlush():
+    | {
+        replayId: string;
+        eventContext: PopEventContext;
+        segmentId: number;
+        promises: Promise<unknown>[];
+        eventBuffer: EventBuffer;
+        session: Session;
       }
+    | undefined {
+    if (!this.session || !this.eventBuffer) {
+      __DEBUG_BUILD__ && logger.error('[Replay] No session or eventBuffer found to flush.');
+      return;
+    }
 
-      // This means we retried 3 times, and all of them failed
-      // In this case, we want to completely stop the replay - otherwise, we may get inconsistent segments
-      this.stop();
+    this._debouncedFlush.cancel();
+
+    const promises: Promise<unknown>[] = [];
+
+    promises.push(this._addPerformanceEntries());
+
+    // Do not continue if there are no pending events in buffer
+    if (!this.eventBuffer || !this.eventBuffer.pendingLength) {
+      return;
     }
+
+    // Only attach memory entry if eventBuffer is not empty
+    promises.push(addMemoryEntry(this));
+
+    // NOTE: Copy values from instance members, as it's possible they could
+    // change before the flush finishes.
+    const replayId = this.session.id;
+    const eventContext = this._popEventContext();
+    // Always increment segmentId regardless of outcome of sending replay
+    const segmentId = this.session.segmentId++;
+
+    // Save session (new segment id) after we save flush data assuming either
+    // 1) request succeeds or 2) it fails or never happens, in which case we
+    // need to retry this segment.
+    this._maybeSaveSession();
+
+    return {
+      replayId,
+      eventContext,
+      segmentId,
+      promises,
+      eventBuffer: this.eventBuffer,
+      session: this.session,
+    };
+  }
+
+  /** Handle an error when sending a replay. */
+  private _handleSendError(error: unknown): void {
+    this._handleException(error);
+
+    if (error instanceof RateLimitError) {
+      this._handleRateLimit(error.rateLimits);
+      return;
+    }
+
+    // This means we retried 3 times, and all of them failed
+    // In this case, we want to completely stop the replay - otherwise, we may get inconsistent segments
+    this.stop();
   }
 
   /**

packages/replay/src/types.ts

@@ -13,7 +13,7 @@ export interface FlushOptions {
    * (e.g. worker calls). This is not directly related to `flushImmediate` which
    * skips the debounced flush.
    */
-  finishImmediate?: boolean;
+  sync?: boolean;
 }
 
 export interface SendReplayData {
@@ -254,7 +254,7 @@ export interface EventBuffer {
   /**
    * Clears and synchronously returns the pending contents of the buffer. This means no compression.
    */
-  finishImmediate(): string;
+  finishSync(): string;
 }
 
 export type AddUpdateCallback = () => boolean | void;

packages/replay/test/integration/flush.test.ts

@@ -108,23 +108,28 @@ describe('Integration | flush', () => {
   });
 
   it('flushes after each blur event', async () => {
+    // @ts-ignore privaye API
+    const mockFlushSync = jest.spyOn(replay, '_flushSync');
+
     // blur events cause an immediate flush that bypass the debounced flush
     // function and skip any async workers
-    expect(mockRunFlush).toHaveBeenCalledTimes(0);
+    expect(mockFlushSync).toHaveBeenCalledTimes(0);
     WINDOW.dispatchEvent(new Event('blur'));
-    expect(mockRunFlush).toHaveBeenCalledTimes(1);
+    expect(mockFlushSync).toHaveBeenCalledTimes(1);
     WINDOW.dispatchEvent(new Event('blur'));
-    expect(mockRunFlush).toHaveBeenCalledTimes(2);
+    expect(mockFlushSync).toHaveBeenCalledTimes(2);
     WINDOW.dispatchEvent(new Event('blur'));
-    expect(mockRunFlush).toHaveBeenCalledTimes(3);
+    expect(mockFlushSync).toHaveBeenCalledTimes(3);
     WINDOW.dispatchEvent(new Event('blur'));
-    expect(mockRunFlush).toHaveBeenCalledTimes(4);
+    expect(mockFlushSync).toHaveBeenCalledTimes(4);
 
+    expect(mockRunFlush).toHaveBeenCalledTimes(0);
     expect(mockFlush).toHaveBeenCalledTimes(0);
 
     jest.runAllTimers();
     await new Promise(process.nextTick);
-    expect(mockRunFlush).toHaveBeenCalledTimes(4);
+    expect(mockFlushSync).toHaveBeenCalledTimes(4);
+    expect(mockRunFlush).toHaveBeenCalledTimes(0);
   });
 
   it('long first flush enqueues following events', async () => {