[#850] Fix EventSource reconnect storm/leak in useStreamEvents

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

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<EventSource | null>(null);
   const retryDelayRef = useRef(1000);
   const reconnectTimeoutRef = useRef<ReturnType<typeof setTimeout> | 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]);