diff --git a/apps/web/src/hooks/use-sse.test.ts b/apps/web/src/hooks/use-sse.test.ts index 3b7000b7..f6d0c4e1 100644 --- a/apps/web/src/hooks/use-sse.test.ts +++ b/apps/web/src/hooks/use-sse.test.ts @@ -1,6 +1,8 @@ // @vitest-environment jsdom -import { QueryClient } from "@tanstack/react-query"; -import { describe, expect, it, vi } from "vitest"; +import { createElement, type ReactNode } from "react"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { act, cleanup, renderHook } from "@testing-library/react"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { agentDiffQueryKey } from "@/hooks/use-agent-diff"; import { diffStatsQueryKey } from "@/hooks/use-agent-diff-stats"; @@ -11,6 +13,7 @@ import { applyAgentUpsert, applyDiffStateChanged, applyReviewCreated, + useSSE, } from "./use-sse"; function agent( @@ -81,3 +84,228 @@ describe("review submission SSE state", () => { }); }); }); + +/** + * Minimal EventSource stand-in. Real EventSource only auto-reconnects after + * transient failures, so the tests drive `readyState` explicitly to + * distinguish "browser is retrying" (CONNECTING) from "browser gave up" + * (CLOSED) — the latter is the case the hook has to recover from itself. + */ +class FakeEventSource { + static readonly CONNECTING = 0; + static readonly OPEN = 1; + static readonly CLOSED = 2; + static instances: FakeEventSource[] = []; + + readyState = FakeEventSource.CONNECTING; + onopen: ((event: Event) => void) | null = null; + onmessage: ((event: MessageEvent) => void) | null = null; + onerror: ((event: Event) => void) | null = null; + + constructor(readonly url: string) { + FakeEventSource.instances.push(this); + } + + close(): void { + this.readyState = FakeEventSource.CLOSED; + } + + /** + * Simulate the connection establishing. Fires on the browser's own internal + * retries too, which is what separates connection age from instance age. + */ + open(): void { + this.readyState = FakeEventSource.OPEN; + this.onopen?.(new Event("open")); + } + + /** Simulate the server delivering an event (the hook's success signal). */ + emit(payload: unknown): void { + this.readyState = FakeEventSource.OPEN; + this.onmessage?.( + new MessageEvent("message", { data: JSON.stringify(payload) }) + ); + } + + /** Simulate a failure; `fatal` mirrors a non-200 / bad-content-type response. */ + fail(fatal: boolean): void { + this.readyState = fatal + ? FakeEventSource.CLOSED + : FakeEventSource.CONNECTING; + this.onerror?.(new Event("error")); + } +} + +describe("useSSE reconnect", () => { + let hiddenValue = false; + + function renderSSE() { + const queryClient = new QueryClient(); + return renderHook(() => useSSE("authenticated"), { + wrapper: ({ children }: { children: ReactNode }) => + createElement(QueryClientProvider, { client: queryClient }, children), + }); + } + + /** Flip tab visibility the way a browser does — state change then event. */ + function setHidden(value: boolean) { + hiddenValue = value; + document.dispatchEvent(new Event("visibilitychange")); + } + + beforeEach(() => { + vi.useFakeTimers(); + FakeEventSource.instances = []; + hiddenValue = false; + vi.stubGlobal("EventSource", FakeEventSource); + vi.spyOn(document, "hidden", "get").mockImplementation(() => hiddenValue); + }); + + afterEach(() => { + // This config sets neither `globals: true` nor `setupFiles`, so React + // Testing Library's auto-cleanup never registers — without an explicit + // call, every hook stays mounted (listeners and all) for the rest of the + // file and a leaked instance answers the next test's document events. + cleanup(); + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + vi.useRealTimers(); + }); + + it("reopens the stream after a fatal error the browser will not retry", () => { + renderSSE(); + expect(FakeEventSource.instances).toHaveLength(1); + + act(() => FakeEventSource.instances[0].fail(true)); + // Nothing yet — the retry is scheduled, not immediate. + expect(FakeEventSource.instances).toHaveLength(1); + + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(2); + }); + + it("leaves transient errors to the browser's own retry", () => { + renderSSE(); + + act(() => FakeEventSource.instances[0].fail(false)); + act(() => void vi.advanceTimersByTime(60_000)); + + expect(FakeEventSource.instances).toHaveLength(1); + }); + + it("backs off exponentially while reconnects keep failing", () => { + renderSSE(); + + act(() => FakeEventSource.instances[0].fail(true)); + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(2); + + act(() => FakeEventSource.instances[1].fail(true)); + act(() => void vi.advanceTimersByTime(1_000)); + // Second delay is 2s, so 1s is not enough. + expect(FakeEventSource.instances).toHaveLength(2); + + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(3); + }); + + it("resets the backoff once a connection has delivered and stayed up", () => { + renderSSE(); + + act(() => FakeEventSource.instances[0].fail(true)); + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(2); + + // The reconnect holds long enough to count as healthy, then delivers. + act(() => void vi.advanceTimersByTime(10_000)); + act(() => + FakeEventSource.instances[1].emit({ type: "snapshot", agents: [] }) + ); + + // Backoff is back at the floor, so 1s is enough for the next retry. + act(() => FakeEventSource.instances[1].fail(true)); + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(3); + }); + + it("keeps backing off when a flapping server delivers only on connect", () => { + // The server writes a snapshot the instant it accepts the connection, so a + // short-lived connect delivers an event without ever being healthy. That + // must not clear the backoff, or a flapping server pins us at the floor. + renderSSE(); + + act(() => FakeEventSource.instances[0].fail(true)); + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(2); + + // Snapshot lands immediately on connect, then the stream dies right away. + act(() => + FakeEventSource.instances[1].emit({ type: "snapshot", agents: [] }) + ); + act(() => FakeEventSource.instances[1].fail(true)); + + // The delay kept doubling: 2s, so 1s must not be enough. + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(2); + + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(3); + }); + + it("measures connection age, not instance age, across an internal retry", () => { + // The browser reconnects transiently under the *same* instance without + // re-running openSSE. A snapshot from that fresh connection must not clear + // the backoff just because the instance itself is old. + renderSSE(); + + act(() => FakeEventSource.instances[0].fail(true)); + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(2); + + const instance = FakeEventSource.instances[1]; + act(() => instance.open()); + act(() => void vi.advanceTimersByTime(10_000)); + + // Transient drop, then the browser re-establishes on its own and the + // server's connect-time snapshot lands on a seconds-old connection. + act(() => instance.fail(false)); + act(() => instance.open()); + act(() => instance.emit({ type: "snapshot", agents: [] })); + + // Backoff must still be elevated: 2s, so 1s is not enough. + act(() => instance.fail(true)); + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(2); + + act(() => void vi.advanceTimersByTime(1_000)); + expect(FakeEventSource.instances).toHaveLength(3); + }); + + it("recovers when the tab is hidden while a retry is pending", () => { + renderSSE(); + + act(() => FakeEventSource.instances[0].fail(true)); + act(() => setHidden(true)); + + // Hiding cancels the pending retry, so nothing reconnects behind the + // user's back — the timer's own `document.hidden` guard never gets to run. + act(() => void vi.advanceTimersByTime(60_000)); + expect(FakeEventSource.instances).toHaveLength(1); + + // Foregrounding is the escape hatch: reopen at once, and only once. + act(() => setHidden(false)); + expect(FakeEventSource.instances).toHaveLength(2); + act(() => void vi.advanceTimersByTime(60_000)); + expect(FakeEventSource.instances).toHaveLength(2); + }); + + it("stops retrying after unmount", () => { + const { unmount } = renderSSE(); + + act(() => FakeEventSource.instances[0].fail(true)); + unmount(); + act(() => void vi.advanceTimersByTime(60_000)); + + expect(FakeEventSource.instances).toHaveLength(1); + }); +}); diff --git a/apps/web/src/hooks/use-sse.ts b/apps/web/src/hooks/use-sse.ts index 7d558aca..44ff82d1 100644 --- a/apps/web/src/hooks/use-sse.ts +++ b/apps/web/src/hooks/use-sse.ts @@ -1,4 +1,4 @@ -import { useEffect, useRef } from "react"; +import { useEffect } from "react"; import { type QueryClient, useQueryClient } from "@tanstack/react-query"; import { useStore } from "jotai"; import { @@ -20,6 +20,28 @@ import { type ReleaseInfoSnapshot, } from "@/hooks/use-cached-release-info"; +/** Backoff bounds for self-driven reconnects after a fatal EventSource error. */ +const INITIAL_RECONNECT_DELAY_MS = 1_000; +const MAX_RECONNECT_DELAY_MS = 30_000; +/** + * How long a connection must last before a delivered event counts as proof + * that it is healthy. + * + * A delivered event on its own proves only that the *connect* succeeded: the + * server writes the snapshot the moment it accepts the connection (see + * `sendUiSnapshot` in apps/server/src/routes/agents/events-routes.ts, called + * immediately after accept). Clearing the backoff on that lets a flapping + * server pin every tab at the 1s floor forever — connect, snapshot, reset, + * drop, repeat — so the cap never engages in the case it exists for. + * + * The reset stays keyed on a delivered event rather than a bare timer, so a + * hung proxy holding the socket open without sending anything can't clear the + * backoff either. Tradeoff: a connection that stays healthy but silent for its + * whole life fails with an elevated delay (capped at MAX_RECONNECT_DELAY_MS, + * and cleared on tab foreground) — acceptable for a stream this chatty. + */ +const STABLE_CONNECTION_MS = 10_000; + type UiEvent = | { type: "snapshot"; agents: Agent[] } | { type: "agent.upsert"; agent: Agent } @@ -147,9 +169,24 @@ export function applyReviewCreated( export function useSSE(authState: AuthState): void { const queryClient = useQueryClient(); const jotaiStore = useStore(); - const eventSourceRef = useRef(null); - useEffect(() => { + // EventSource only auto-reconnects after *transient* failures. A fatal + // one (non-200 response, wrong content-type — e.g. hitting the server + // mid-restart) moves it to CLOSED permanently, and nothing here noticed: + // the dead instance stayed put, so `openSSE` early-returned forever and a + // visible tab silently lost every realtime update. We drive our own capped + // backoff for that case. + // + // The connection and its backoff are one state machine, so they share one + // lifetime: all of it is effect-scoped and torn down together. Nothing + // here is read during render, so none of it needs to be a ref. + let source: EventSource | null = null; + let reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS; + let reconnectTimer: ReturnType | null = null; + /** When the current connection last established — or, before it has + * opened, when we started attempting it. */ + let connectionAliveSince = 0; + const handleSSEMessage = (event: MessageEvent) => { try { recordSSEEvent(); @@ -350,22 +387,80 @@ export function useSSE(authState: AuthState): void { } catch {} }; + const cancelReconnect = () => { + if (reconnectTimer !== null) { + clearTimeout(reconnectTimer); + reconnectTimer = null; + } + }; + + /** A delivered event clears the backoff, but only once the connection has + * proven it can last — see STABLE_CONNECTION_MS. */ + const noteStreamActivity = () => { + if (Date.now() - connectionAliveSince >= STABLE_CONNECTION_MS) { + reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS; + } + }; + + const scheduleReconnect = () => { + if (reconnectTimer !== null) return; + const delay = reconnectDelayMs; + reconnectDelayMs = Math.min(reconnectDelayMs * 2, MAX_RECONNECT_DELAY_MS); + reconnectTimer = setTimeout(() => { + reconnectTimer = null; + if (document.hidden) return; + openSSE(); + }, delay); + }; + const openSSE = () => { - if (eventSourceRef.current) return; - const source = new EventSource("/api/v1/events", { + // A live source and a pending retry are mutually exclusive: a retry is + // only ever scheduled after the source is dropped. + if (source) return; + cancelReconnect(); + const opened = new EventSource("/api/v1/events", { withCredentials: true, }); - eventSourceRef.current = source; - source.onmessage = handleSSEMessage; - source.onerror = () => { + // No `open` yet, so measure from the attempt. Generous by the + // establishment latency, but never 0 — a zero here would make the + // staleness check trivially true and silently clear the backoff. + connectionAliveSince = Date.now(); + source = opened; + // `open` fires on every establishment, including the browser's own + // internal retries after a transient drop. Those never re-run `openSSE`, + // so without this the gate would measure the age of the *instance* + // rather than of the connection, and a connect-time snapshot delivered + // by an internal retry would clear the backoff — the exact event the + // gate exists to discount. + opened.onopen = () => { + connectionAliveSince = Date.now(); + }; + opened.onmessage = (event) => { + noteStreamActivity(); + handleSSEMessage(event); + }; + opened.onerror = () => { recordSSEReconnect(); + // CONNECTING means the browser is retrying on its own — leave it be. + // CLOSED means it has given up; the instance is dead, so drop it and + // retry ourselves. + if (opened.readyState !== EventSource.CLOSED) return; + opened.close(); + // `close()` aborts queued dispatches, so a replaced instance can't + // reach here in a conformant browser. Keep the whole branch behind the + // identity check anyway: scheduling a retry beside a live connection + // would double the delay for nothing. + if (source !== opened) return; + source = null; + scheduleReconnect(); }; }; const closeSSE = () => { - if (eventSourceRef.current) { - eventSourceRef.current.close(); - eventSourceRef.current = null; + cancelReconnect(); + if (source) { + source.close(); + source = null; } }; @@ -377,6 +472,9 @@ export function useSSE(authState: AuthState): void { if (document.hidden || authState !== "authenticated") { closeSSE(); } else { + // Foregrounding is a fresh start — don't inherit a backed-off delay + // from whatever killed the previous connection. + reconnectDelayMs = INITIAL_RECONNECT_DELAY_MS; openSSE(); } };