fix(replay): Fixes potential out-of-order segments

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();
+      }
     }
   };
 

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