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

.github/workflows/ci.yml

@@ -96,7 +96,7 @@ jobs:
       - name: Run Backend Tests
         run: |
           ls -la src/generated/prisma
-          npx vitest run --coverage --reporter=basic
+          npx vitest run --coverage --reporter=basic --no-file-parallelism
         working-directory: backend
         env:
           DATABASE_URL: postgresql://postgres:password@127.0.0.1:5432/flowfi_test

.github/workflows/pr-test-gate.yml

@@ -65,7 +65,7 @@ jobs:
           DATABASE_URL: postgresql://postgres:password@127.0.0.1:5432/flowfi_test
 
       - name: Run backend tests
-        run: npm test
+        run: npm test -- --no-file-parallelism
         working-directory: backend
         env:
           DATABASE_URL: postgresql://postgres:password@127.0.0.1:5432/flowfi_test

backend/tests/events-wire-format.test.ts

@@ -91,7 +91,7 @@ describe('event wire format', () => {
 
       expect(decodedKeys).toEqual([...allFields].sort());
 
-      for (const field of HANDLER_READ_FIELDS[eventName]) {
+      for (const field of HANDLER_READ_FIELDS[eventName] ?? []) {
         expect(decoded).toHaveProperty(field);
       }
     },

backend/tests/integration/stream-lifecycle.test.ts

@@ -88,7 +88,7 @@
   ];
 
   const overrideEntries: [string, xdr.ScVal][] = Object.entries(overrides).map(
-    ([k, v]) => [k, nativeToScVal(v)],
+    ([k, v]) => [k, typeof v === 'bigint' ? scvI128(v) : nativeToScVal(v)],
   );
 
   return {
@@ -306,7 +306,7 @@
       // Verify API reflects the change
       const response = await request(app)
         .get(`/v1/streams/${streamId}`)
         .expect(200);
 
       expect(response.body.depositedAmount).toBe("87400");
 
@@ -355,7 +355,7 @@
       // Verify API reflects paused state
       const response = await request(app)
         .get(`/v1/streams/${streamId}`)
         .expect(200);
 
       expect(response.body.isPaused).toBe(true);
 

backend/tests/stream.validator.test.ts

@@ -42,7 +42,9 @@ describe('Stream Validator', () => {
     };
     const result = createStreamSchema.safeParse(data);
     expect(result.success).toBe(false);
-    expect(result.error?.issues[0].message).toBe('Rate exceeds maximum allowed value');
+    expect(result.error).toBeDefined();
+    expect(result.error!.issues.length).toBeGreaterThan(0);
+    expect(result.error!.issues[0]!.message).toBe('Rate exceeds maximum allowed value');
   });
 
   it('should accept ratePerSecond at i128 max', () => {

contracts/stream_contract/src/test.rs

@@ -433,8 +433,8 @@ fn test_backdated_start_time_would_immediately_vest_full_amount() {
     // env.ledger().timestamp() — but it demonstrates the risk that would exist
     // if a caller-supplied start_time were ever added.
     let mut stream = client.get_stream(&stream_id).unwrap();
-    stream.start_time = 0;          // backdated far into the past
-    stream.last_update_time = 0;    // sync anchor to match
+    stream.start_time = 0; // backdated far into the past
+    stream.last_update_time = 0; // sync anchor to match
     env.as_contract(&client.address, || {
         env.storage()
             .persistent()
@@ -2708,11 +2708,8 @@ fn test_cancel_state_committed_before_transfers_prevents_double_cancel() {
 fn event_field_names(env: &Env, payload: &soroban_sdk::Val) -> std::vec::Vec<std::string::String> {
     let map = soroban_sdk::Map::<Symbol, soroban_sdk::Val>::try_from_val(env, payload)
         .expect("event data is not a Map");
-    let mut names: std::vec::Vec<std::string::String> = map
-        .keys()
-        .iter()
-        .map(|sym| sym.to_string())
-        .collect();
+    let mut names: std::vec::Vec<std::string::String> =
+        map.keys().iter().map(|sym| sym.to_string()).collect();
     names.sort();
     names
 }

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/components/dashboard/dashboard-view.tsx

@@ -18,7 +18,6 @@ import toast from "react-hot-toast";
 import {
   getDashboardAnalytics,
   fetchDashboardData,
-  useDashboard,
   dashboardQueryKey,
   type DashboardSnapshot,
   type Stream,

frontend/src/hooks/useIncomingStreams.test.ts → frontend/src/hooks/useIncomingStreams.test.tsx

@@ -74,21 +74,29 @@ describe("useIncomingStreams hooks", () => {
       );
 
       await expect(
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any
         result.current.mutateAsync({} as any)
       ).rejects.toThrow("Please connect your wallet first");
       expect(withdrawFromStream).not.toHaveBeenCalled();
     });
 
     it("invalidates incomingStreamsQueryKey(publicKey) on success", async () => {
+      // Return a matching stream so pollIndexerForWithdraw exits on the first
+      // attempt (1 s delay) by finding updatedStream.withdrawn > oldWithdrawn.
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
       (withdrawFromStream as any).mockResolvedValue({ status: "success" });
-      (fetchIncomingStreams as any).mockResolvedValue([]);
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      (fetchIncomingStreams as any).mockResolvedValue([
+        { streamId: 1, withdrawn: 100 },
+      ]);
 
       const { result } = renderHook(
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any
         () => useWithdrawIncomingStream({} as any, "pubkey"),
         { wrapper }
       );
 
-      const invalidateSpy = vi.spyOn(queryClient, "invalidateQueries");
+      const setQueryDataSpy = vi.spyOn(queryClient, "setQueryData");
 
       await act(async () => {
         await result.current.mutateAsync({
@@ -99,15 +107,22 @@ describe("useIncomingStreams hooks", () => {
           ratePerSecond: 1,
           isPaused: false,
           lastUpdateTime: Date.now() / 1000,
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any
         } as any);
       });
 
-      // Wait for pollIndexerForWithdraw to complete and call invalidateQueries
+      // Clear any calls made during mutation (e.g. optimistic update in
+      // onMutate) so we only assert on the poll's setQueryData call.
+      setQueryDataSpy.mockClear();
+
+      // Poll should find the updated stream (withdrawn 100 > 0) and call
+      // setQueryData after ~1 s of simulated delay.
       await waitFor(() => {
-        expect(invalidateSpy).toHaveBeenCalledWith({
-          queryKey: incomingStreamsQueryKey("pubkey"),
-        });
-      }, { timeout: 10000 });
+        expect(setQueryDataSpy).toHaveBeenCalledWith(
+          incomingStreamsQueryKey("pubkey"),
+          expect.any(Array),
+        );
+      }, { timeout: 5000 });
     });
   });
 });

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]);

frontend/src/lib/dashboard.ts

@@ -88,21 +88,19 @@ async function fetchStreams(
   for (const endpoint of endpoints) {
     try {
       const response = await fetch(`${endpoint}?${params.toString()}`, { signal });
-    if (response.ok) {
-      const payload = (await response.json()) as
-        | BackendStream[]
-        | { data?: BackendStream[] };
-      return Array.isArray(payload) ? payload : payload.data ?? [];
-    }
-
-    if (response.status === 404) {
-      lastError = new Error(`Endpoint not found: ${endpoint}`);
-      continue;
-    }
+      if (response.ok) {
+        const payload = (await response.json()) as
+          | BackendStream[]
+          | { data?: BackendStream[] };
+        return Array.isArray(payload) ? payload : payload.data ?? [];
+      }
 
-    lastError = new Error(`Failed to fetch streams (${response.status}) from ${endpoint}`);
-  }
+      if (response.status === 404) {
+        lastError = new Error(`Endpoint not found: ${endpoint}`);
+        continue;
+      }
 
+      lastError = new Error(`Failed to fetch streams (${response.status}) from ${endpoint}`);
     } catch (err) {
       if (err instanceof Error && err.name === "AbortError") {
         throw err;