Fix/telemetry overlapping flushes

frontend/src/lib/telemetry-sink.js

@@ -14,6 +14,7 @@ const DEFAULT_CONFIG = {
 let sinkConfig = { ...DEFAULT_CONFIG };
 let eventQueue = [];
 let flushTimer = null;
+let activeFlushPromise = null;
 
 export function configureSink(config) {
   sinkConfig = {
@@ -111,12 +112,29 @@ async function sendBatch(events, attempt = 1) {
 }
 
 export async function flush() {
+  if (activeFlushPromise) {
+    return activeFlushPromise.then(() => flush()).catch(() => flush());
+  }
+
   if (!isSinkEnabled() || eventQueue.length === 0) {
     return { success: true, count: 0 };
   }
 
   const batch = eventQueue.splice(0, sinkConfig.batchSize);
-  return sendBatch(batch);
+  activeFlushPromise = sendBatch(batch).then((result) => {
+    activeFlushPromise = null;
+    if (!result.success && isSinkEnabled()) {
+      eventQueue.unshift(...batch);
+    }
+    return result;
+  }).catch((error) => {
+    activeFlushPromise = null;
+    if (isSinkEnabled()) {
+      eventQueue.unshift(...batch);
+    }
+    throw error;
+  });
+  return activeFlushPromise;
 }
 
 export async function sendSnapshot() {
@@ -165,5 +183,6 @@ export function clearQueue() {
 export function resetSink() {
   stopFlushTimer();
   eventQueue = [];
+  activeFlushPromise = null;
   sinkConfig = { ...DEFAULT_CONFIG };
 }

frontend/src/test/telemetry-sink.test.js

@@ -196,10 +196,107 @@ describe('telemetry-sink', () => {
       await vi.advanceTimersByTimeAsync(30);
 
       const result = await flushPromise;
-
       expect(result.success).toBe(false);
       expect(result.error).toBe('Network error');
     });
+
+    it('serializes overlapping flush operations', async () => {
+      let resolveFirstFetch;
+      const firstFetchPromise = new Promise((resolve) => {
+        resolveFirstFetch = () => resolve({ ok: true, status: 200 });
+      });
+
+      global.fetch
+        .mockReturnValueOnce(firstFetchPromise)
+        .mockResolvedValueOnce({ ok: true, status: 200 });
+
+      configureSink({
+        enabled: true,
+        endpoint: 'https://test.com',
+        batchSize: 1,
+      });
+
+      queueEvent('event1', {});
+      queueEvent('event2', {});
+
+      const promise1 = flush();
+      const promise2 = flush();
+
+      expect(global.fetch).toHaveBeenCalledTimes(1);
+
+      resolveFirstFetch();
+      await promise1;
+      await promise2;
+
+      expect(global.fetch).toHaveBeenCalledTimes(2);
+    });
+
+    it('preserves queue integrity and order when flush fails', async () => {
+      global.fetch
+        .mockRejectedValueOnce(new Error('Network error 1'))
+        .mockRejectedValueOnce(new Error('Network error 2'))
+        .mockResolvedValueOnce({ ok: true, status: 200 });
+
+      configureSink({
+        enabled: true,
+        endpoint: 'https://test.com',
+        batchSize: 10,
+        retryAttempts: 2,
+        retryDelayMs: 10,
+      });
+
+      queueEvent('event1', {});
+
+      const promise1 = flush();
+      queueEvent('event2', {});
+      const promise2 = flush();
+
+      await vi.advanceTimersByTimeAsync(30);
+
+      const result1 = await promise1;
+      expect(result1.success).toBe(false);
+
+      const result2 = await promise2;
+      expect(result2.success).toBe(true);
+      expect(result2.count).toBe(2);
+
+      expect(getQueueLength()).toBe(0);
+    });
+
+    it('handles sink reset/disable during failed flush to not prepend', async () => {
+      let rejectFirstFetch;
+      const firstFetchPromise = new Promise((_, reject) => {
+        rejectFirstFetch = () => reject(new Error('Fetch failed'));
+      });
+
+      global.fetch.mockReturnValueOnce(firstFetchPromise);
+
+      configureSink({
+        enabled: true,
+        endpoint: 'https://test.com',
+        batchSize: 10,
+        retryAttempts: 1,
+      });
+
+      queueEvent('event1', {});
+      queueEvent('event2', {});
+
+      const promise1 = flush();
+      const promise2 = flush();
+
+      resetSink();
+
+      rejectFirstFetch();
+
+      const result1 = await promise1;
+      expect(result1.success).toBe(false);
+
+      const result2 = await promise2;
+      expect(result2.success).toBe(true);
+      expect(result2.count).toBe(0);
+
+      expect(getQueueLength()).toBe(0);
+    });
   });
 
   describe('clearQueue', () => {