From 8fcee67c4eda0720061b9d24513a44e7518b19bd Mon Sep 17 00:00:00 2001
From: Billy Vong <billyvg@users.noreply.github.com>
Date: Tue, 10 Sep 2024 16:08:29 -0230
Subject: [PATCH] fix(replay): Fixes potential out-of-order segments (#13609)

This fixes a potential issue where segments can come in out of order due
to our flush "lock" was not being respected in two cases:

1) No current flush in progress, but flush throws an error (this should
be rare as the common errors that get thrown should stop the replay
completely)
2) Flush is in progress, which skips the code block that releases lock
and then calls debouncedFlush. This leaves the lock always set to a
resolved (or rejected) promise.

This ultimately should not change too much as the flush calls are
debounced anyway, but this cleans up the code a bit and also logs any
exceptions that may occur. However this can fix issues where segments
can come in out of order depending on how long the send request takes.
e.g.


![image](https://github.com/user-attachments/assets/ea304892-1c72-4e96-acc6-c714d263980c)


where ideally it looks like

![image](https://github.com/user-attachments/assets/8c3e706c-d3b2-43bd-a970-561b32b05458)
---
 packages/replay-internal/src/replay.ts        | 28 ++++-----
 .../test/integration/flush.test.ts            | 58 +++++++++++++++++++
 2 files changed, 73 insertions(+), 13 deletions(-)

diff --git a/packages/replay-internal/src/replay.ts b/packages/replay-internal/src/replay.ts
index b48ac787543b..cfeb26841911 100644
--- a/packages/replay-internal/src/replay.ts
+++ b/packages/replay-internal/src/replay.ts
@@ -1226,27 +1226,29 @@ export class ReplayContainer implements ReplayContainerInterface {
       // TODO FN: Evaluate if we want to stop here, or remove this again?
     }
 
-    // this._flushLock acts as a lock so that future calls to `_flush()`
-    // will be blocked until this promise resolves
+    const _flushInProgress = !!this._flushLock;
+
+    // this._flushLock acts as a lock so that future calls to `_flush()` will
+    // be blocked until current flush is finished (i.e. this promise resolves)
     if (!this._flushLock) {
       this._flushLock = this._runFlush();
-      await this._flushLock;
-      this._flushLock = undefined;
-      return;
     }
 
-    // Wait for previous flush to finish, then call the debounced `_flush()`.
-    // It's possible there are other flush requests queued and waiting for it
-    // to resolve. We want to reduce all outstanding requests (as well as any
-    // new flush requests that occur within a second of the locked flush
-    // completing) into a single flush.
-
     try {
       await this._flushLock;
     } catch (err) {
-      DEBUG_BUILD && logger.error(err);
+      this.handleException(err);
     } finally {
-      this._debouncedFlush();
+      this._flushLock = undefined;
+
+      if (_flushInProgress) {
+        // Wait for previous flush to finish, then call the debounced
+        // `_flush()`. It's possible there are other flush requests queued and
+        // waiting for it to resolve. We want to reduce all outstanding
+        // requests (as well as any new flush requests that occur within a
+        // second of the locked flush completing) into a single flush.
+        this._debouncedFlush();
+      }
     }
   };
 
diff --git a/packages/replay-internal/test/integration/flush.test.ts b/packages/replay-internal/test/integration/flush.test.ts
index ffc0a83bb141..52654fa909d3 100644
--- a/packages/replay-internal/test/integration/flush.test.ts
+++ b/packages/replay-internal/test/integration/flush.test.ts
@@ -493,6 +493,64 @@ describe('Integration | flush', () => {
     await replay.start();
   });
 
+  it('resets flush lock if runFlush rejects/throws', async () => {
+    mockRunFlush.mockImplementation(
+      () =>
+        new Promise((resolve, reject) => {
+          reject(new Error('runFlush'));
+        }),
+    );
+    try {
+      await replay['_flush']();
+    } catch {
+      // do nothing
+    }
+    expect(replay['_flushLock']).toBeUndefined();
+  });
+
+  it('resets flush lock when flush is called multiple times before it resolves', async () => {
+    let _resolve;
+    mockRunFlush.mockImplementation(
+      () =>
+        new Promise(resolve => {
+          _resolve = resolve;
+        }),
+    );
+    const mockDebouncedFlush: MockedFunction<ReplayContainer['_debouncedFlush']> = vi.spyOn(replay, '_debouncedFlush');
+    mockDebouncedFlush.mockImplementation(vi.fn);
+    mockDebouncedFlush.cancel = vi.fn();
+
+    const results = [replay['_flush'](), replay['_flush']()];
+    expect(replay['_flushLock']).not.toBeUndefined();
+
+    _resolve && _resolve();
+    await Promise.all(results);
+    expect(replay['_flushLock']).toBeUndefined();
+    mockDebouncedFlush.mockRestore();
+  });
+
+  it('resets flush lock when flush is called multiple times before it rejects', async () => {
+    let _reject;
+    mockRunFlush.mockImplementation(
+      () =>
+        new Promise((_, reject) => {
+          _reject = reject;
+        }),
+    );
+    const mockDebouncedFlush: MockedFunction<ReplayContainer['_debouncedFlush']> = vi.spyOn(replay, '_debouncedFlush');
+    mockDebouncedFlush.mockImplementation(vi.fn);
+    mockDebouncedFlush.cancel = vi.fn();
+    expect(replay['_flushLock']).toBeUndefined();
+    replay['_flush']();
+    const result = replay['_flush']();
+    expect(replay['_flushLock']).not.toBeUndefined();
+
+    _reject && _reject(new Error('Throw runFlush'));
+    await result;
+    expect(replay['_flushLock']).toBeUndefined();
+    mockDebouncedFlush.mockRestore();
+  });
+
   /**
    * Assuming the user wants to record a session
    * when calling flush() without replay being enabled