diff --git a/frontend/src/lib/telemetry-sink.js b/frontend/src/lib/telemetry-sink.js index 37233612..cd75d516 100644 --- a/frontend/src/lib/telemetry-sink.js +++ b/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 }; } diff --git a/frontend/src/test/telemetry-sink.test.js b/frontend/src/test/telemetry-sink.test.js index 67cf3eb7..65cefead 100644 --- a/frontend/src/test/telemetry-sink.test.js +++ b/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', () => {