From ea5e4b9d95a4def08a5f22662b15c1a4be255bad Mon Sep 17 00:00:00 2001 From: Hollujay <165713167+Hollujay@users.noreply.github.com> Date: Mon, 27 Jul 2026 08:41:17 +0000 Subject: [PATCH] fix: prevent EventSource reconnect storm in useStreamEvents MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Use a stable subscription key derived from sorted streamIds, subscribeToAll, and jwtToken so buildUrl/connect only change when the subscription actually changes — not on every render due to inline array literals or default [] identity. - Derive subscriptionKey as a stable memoized string - Clear pending reconnect timer before scheduling a new one - Cap reconnect attempts at 20 (configurable via constant) Fixes #850 --- .../src/__tests__/useStreamEvents.test.tsx | 52 ++++++++++++++++ frontend/src/hooks/useStreamEvents.ts | 59 +++++++++++++------ 2 files changed, 93 insertions(+), 18 deletions(-) diff --git a/frontend/src/__tests__/useStreamEvents.test.tsx b/frontend/src/__tests__/useStreamEvents.test.tsx index b28a3c9b..c62e0edb 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(); }); @@ -200,4 +203,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..66438f59 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,23 @@ 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) { + reconnectTimeoutRef.current = setTimeout(() => { + connectRef.current(); + retryDelayRef.current = Math.min( + retryDelayRef.current * 2, + maxRetryDelay + ); + }, retryDelayRef.current); + } } }; }, [buildUrl, autoReconnect, maxRetryDelay]); @@ -131,8 +153,9 @@ export function useStreamEvents( eventSourceRef.current.close(); eventSourceRef.current = null; } - if (reconnectTimeoutRef.current) { + if (reconnectTimeoutRef.current !== null) { clearTimeout(reconnectTimeoutRef.current); + reconnectTimeoutRef.current = null; } }; }, [connect]);