From d589a44f4f4290fee47db5b7605006f68f6cfed2 Mon Sep 17 00:00:00 2001 From: Hollujay Date: Wed, 5 Aug 2026 09:04:04 +0100 Subject: [PATCH] fix(#850): eventsource reconnect storm + backoff timing - Memoize subscription key so identical streamIds/token don't trigger spurious reconnects on every render. - Clear any pending reconnect timeout at the start of connect() to prevent overlapping connections. - Cap reconnect attempts at 20 to stop infinite reconnect loops. - Combined with an independent backoff-timing fix already on main (precompute delay before scheduling, not after). --- .../src/__tests__/useStreamEvents.test.tsx | 132 ++++++++++++++++++ frontend/src/hooks/useStreamEvents.ts | 60 +++++--- 2 files changed, 174 insertions(+), 18 deletions(-) diff --git a/frontend/src/__tests__/useStreamEvents.test.tsx b/frontend/src/__tests__/useStreamEvents.test.tsx index b28a3c9b..06a6dcc8 100644 --- a/frontend/src/__tests__/useStreamEvents.test.tsx +++ b/frontend/src/__tests__/useStreamEvents.test.tsx @@ -9,6 +9,7 @@ type ErrorHandler = () => void; class MockEventSource { static instance: MockEventSource | null = null; + static instanceCount = 0; url: string; onopen: (() => void) | null = null; @@ -21,6 +22,7 @@ class MockEventSource { constructor(url: string) { this.url = url; MockEventSource.instance = this; + MockEventSource.instanceCount += 1; } addEventListener(type: string, handler: EventHandler) { @@ -66,6 +68,7 @@ class MockEventSource { describe('useStreamEvents', () => { beforeEach(() => { MockEventSource.instance = null; + MockEventSource.instanceCount = 0; vi.useFakeTimers(); }); @@ -152,6 +155,86 @@ describe('useStreamEvents', () => { expect(MockEventSource.instance).not.toBeNull(); }); + it('grows the reconnect delay exponentially across consecutive failures', () => { + // Keep `streamIds` referentially stable across re-renders (an inline + // array literal would change identity on every render and cause the + // hook to reconnect on its own, independent of the backoff timer). + const streamIds = ['1']; + renderHook(() => + useStreamEvents({ streamIds, autoReconnect: true, maxRetryDelay: 30000 }), + ); + + const first = MockEventSource.instance; + act(() => { first?.triggerError(); }); + + // First retry is scheduled after the initial 1000ms delay. + act(() => { vi.advanceTimersByTime(999); }); + expect(MockEventSource.instance).toBe(first); + act(() => { vi.advanceTimersByTime(1); }); + const second = MockEventSource.instance; + expect(second).not.toBe(first); + expect(second).not.toBeNull(); + + act(() => { second?.triggerError(); }); + + // Second retry should wait ~2000ms (doubled), not repeat the 1000ms delay. + act(() => { vi.advanceTimersByTime(1999); }); + expect(MockEventSource.instance).toBe(second); + act(() => { vi.advanceTimersByTime(1); }); + const third = MockEventSource.instance; + expect(third).not.toBe(second); + expect(third).not.toBeNull(); + + act(() => { third?.triggerError(); }); + + // Third retry should wait ~4000ms. + act(() => { vi.advanceTimersByTime(3999); }); + expect(MockEventSource.instance).toBe(third); + act(() => { vi.advanceTimersByTime(1); }); + expect(MockEventSource.instance).not.toBe(third); + }); + + it('caps the reconnect delay at maxRetryDelay', () => { + const streamIds = ['1']; + renderHook(() => + useStreamEvents({ streamIds, autoReconnect: true, maxRetryDelay: 1500 }), + ); + + const first = MockEventSource.instance; + act(() => { first?.triggerError(); }); + act(() => { vi.advanceTimersByTime(1000); }); // capped delay is min(1000, 1500) = 1000 + const second = MockEventSource.instance; + expect(second).not.toBe(first); + + act(() => { second?.triggerError(); }); + // Next delay would be 2000 uncapped, but maxRetryDelay caps it at 1500. + act(() => { vi.advanceTimersByTime(1499); }); + expect(MockEventSource.instance).toBe(second); + act(() => { vi.advanceTimersByTime(1); }); + expect(MockEventSource.instance).not.toBe(second); + }); + + it('resets the backoff delay to the initial value after a successful connection', () => { + const streamIds = ['1']; + renderHook(() => + useStreamEvents({ streamIds, autoReconnect: true, maxRetryDelay: 30000 }), + ); + + const first = MockEventSource.instance; + act(() => { first?.triggerError(); }); + act(() => { vi.advanceTimersByTime(1000); }); // reconnect after initial 1000ms + + const second = MockEventSource.instance; + act(() => { second?.open(); }); // successful connection resets the counter + act(() => { second?.triggerError(); }); + + // Backoff should restart at 1000ms, not continue growing to 2000ms. + act(() => { vi.advanceTimersByTime(999); }); + expect(MockEventSource.instance).toBe(second); + act(() => { vi.advanceTimersByTime(1); }); + expect(MockEventSource.instance).not.toBe(second); + }); + it('clearEvents empties the events array', () => { const { result } = renderHook(() => useStreamEvents({ streamIds: ['1'], autoReconnect: false }), @@ -200,4 +283,53 @@ describe('useStreamEvents', () => { expect(result.current.events).toHaveLength(types.length); }); + + it('creates only one EventSource across multiple re-renders and incoming events', () => { + const { result, rerender } = renderHook( + (opts: { streamIds: string[] } = { streamIds: ['1'] }) => + useStreamEvents({ ...opts, autoReconnect: false }), + ); + + const firstInstance = MockEventSource.instance; + + act(() => { firstInstance?.open(); }); + + // Simulate multiple re-renders with the same subscription (inline array) + rerender({ streamIds: ['1'] }); + rerender({ streamIds: ['1'] }); + rerender({ streamIds: ['1'] }); + + // Simulate incoming events causing re-renders of the consumer + act(() => { + MockEventSource.instance?.emit('stream.created', { i: 1 }); + MockEventSource.instance?.emit('stream.created', { i: 2 }); + MockEventSource.instance?.emit('stream.created', { i: 3 }); + }); + + expect(result.current.events).toHaveLength(3); + + // Re-render again after events + rerender({ streamIds: ['1'] }); + rerender({ streamIds: ['1'] }); + + expect(MockEventSource.instanceCount).toBe(1); + expect(MockEventSource.instance).toBe(firstInstance); + }); + + it('stops reconnecting after reaching the cap', () => { + renderHook(() => + useStreamEvents({ streamIds: ['1'], autoReconnect: true, maxRetryDelay: 1000 }), + ); + + // Trigger errors repeatedly to consume reconnect attempts. + // The reconnect delay stays at 1000ms (capped by maxRetryDelay). + for (let i = 0; i < 25; i++) { + act(() => { MockEventSource.instance?.triggerError(); }); + act(() => { vi.advanceTimersByTime(2000); }); + } + + // 1 initial + 20 reconnect attempts = 21 instances max. + // After the 20th reconnect attempt, no more timers should fire. + expect(MockEventSource.instanceCount).toBeLessThanOrEqual(21); + }); }); diff --git a/frontend/src/hooks/useStreamEvents.ts b/frontend/src/hooks/useStreamEvents.ts index f782c158..b7dfdde1 100644 --- a/frontend/src/hooks/useStreamEvents.ts +++ b/frontend/src/hooks/useStreamEvents.ts @@ -1,4 +1,4 @@ -import { useEffect, useState, useCallback, useRef } from 'react'; +import { useEffect, useState, useCallback, useRef, useMemo } from 'react'; interface StreamEvent { type: 'created' | 'topped_up' | 'withdrawn' | 'cancelled' | 'completed' | 'paused' | 'resumed'; @@ -23,12 +23,13 @@ interface UseStreamEventsReturn { clearEvents: () => void; } +const MAX_RECONNECT_ATTEMPTS = 20; + export function useStreamEvents( options: UseStreamEventsOptions = {} ): UseStreamEventsReturn { const { - streamIds = [], - // userPublicKeys = [], + streamIds: rawStreamIds = [], subscribeToAll = false, autoReconnect = true, maxRetryDelay = 30000, @@ -43,32 +44,43 @@ export function useStreamEvents( const eventSourceRef = useRef(null); const retryDelayRef = useRef(1000); const reconnectTimeoutRef = useRef | null>(null); + const reconnectAttemptsRef = useRef(0); const connectRef = useRef<() => void>(() => undefined); + const subscriptionKey = useMemo(() => { + const streams = [...rawStreamIds].sort().join(','); + return `${subscribeToAll ? 'all' : streams}|${jwtToken || ''}`; + }, [rawStreamIds, subscribeToAll, jwtToken]); + const buildUrl = useCallback(() => { const params = new URLSearchParams(); if (subscribeToAll) { params.append('all', 'true'); } else { - streamIds.forEach(id => params.append('streams', id)); + rawStreamIds.forEach(id => params.append('streams', id)); } - // Add JWT token to query string for authentication - // (EventSource doesn't support custom headers in browser) if (jwtToken) { params.append('token', jwtToken); } const baseUrl = process.env.NEXT_PUBLIC_API_URL || 'http://localhost:3001'; return `${baseUrl}/v1/events/subscribe?${params}`; - }, [streamIds, subscribeToAll, jwtToken]); + // subscriptionKey captures all subscription parameters as a stable string + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [subscriptionKey]); const clearEvents = useCallback(() => { setEvents([]); }, []); const connect = useCallback(() => { + if (reconnectTimeoutRef.current !== null) { + clearTimeout(reconnectTimeoutRef.current); + reconnectTimeoutRef.current = null; + } + const url = buildUrl(); const eventSource = new EventSource(url); eventSourceRef.current = eventSource; @@ -77,16 +89,16 @@ export function useStreamEvents( setConnected(true); setReconnecting(false); setError(null); - retryDelayRef.current = 1000; // Reset retry delay + retryDelayRef.current = 1000; + reconnectAttemptsRef.current = 0; }; - const handleEvent = (type: StreamEvent['type']) => (e: MessageEvent) => { try { const data = JSON.parse(e.data); setEvents((prev: StreamEvent[]) => [ { type, data, timestamp: Date.now() }, - ...prev.slice(0, 99), // Keep last 100 events + ...prev.slice(0, 99), ]); } catch { // Silently ignore malformed event messages @@ -108,13 +120,24 @@ export function useStreamEvents( if (autoReconnect) { setReconnecting(true); - reconnectTimeoutRef.current = setTimeout(() => { - connectRef.current(); - retryDelayRef.current = Math.min( - retryDelayRef.current * 2, - maxRetryDelay - ); - }, retryDelayRef.current); + + if (reconnectTimeoutRef.current !== null) { + clearTimeout(reconnectTimeoutRef.current); + reconnectTimeoutRef.current = null; + } + + reconnectAttemptsRef.current += 1; + + if (reconnectAttemptsRef.current <= MAX_RECONNECT_ATTEMPTS) { + // Cap the delay we're about to wait on, and precompute the next + // (doubled, capped) delay up front so consecutive failures keep + // growing the backoff even if the next attempt fails immediately. + const delay = Math.min(retryDelayRef.current, maxRetryDelay); + retryDelayRef.current = Math.min(retryDelayRef.current * 2, maxRetryDelay); + reconnectTimeoutRef.current = setTimeout(() => { + connectRef.current(); + }, delay); + } } }; }, [buildUrl, autoReconnect, maxRetryDelay]); @@ -131,8 +154,9 @@ export function useStreamEvents( eventSourceRef.current.close(); eventSourceRef.current = null; } - if (reconnectTimeoutRef.current) { + if (reconnectTimeoutRef.current !== null) { clearTimeout(reconnectTimeoutRef.current); + reconnectTimeoutRef.current = null; } }; }, [connect]);