fix(FN-2048): prevent stale SSE reconnects after channel teardown

- Add a closed flag to sse-bus channel state so teardown permanently disables reconnect scheduling
- Guard useTasks onReconnect callbacks against stale effect instances when toggling views
- Add regression tests for sse-bus and useTasks to ensure closed/unmounted channels do not reconnect
- Document the leak root cause and closed-flag pattern in .fusion/memory.md
This commit is contained in:
Fusion
2026-04-18 02:53:49 -07:00
committed by gsxdsm
parent 17fcbfe6c8
commit 76123c9538
5 changed files with 179 additions and 2 deletions

View File

@@ -1233,3 +1233,61 @@ When a mission feature's implementation task completes, `MissionExecutionLoop.pr
- **VALID_TRANSITIONS**: `"in-progress"``"done"` was added specifically for validation tasks (note in `types.ts`)
The error status from `parseValidationResult()` (empty response, invalid JSON, invalid status) is now handled in `processTaskOutcome()` alongside pass/fail/blocked.
## FN-2048: SSE Connection Leak — `closed` Flag Pattern in sse-bus
Rapid view transitions (e.g., Board↔Missions) caused zombie EventSource connections to accumulate, exhausting the browser's HTTP/1.1 connection pool (6 per origin) and blocking subsequent `fetchMissions` calls. Two fixes prevent this:
### 1. `closed` flag in sse-bus `Channel`
The `closeChannel` function sets `channel.closed = true` **before** closing the EventSource. This prevents `forceReconnect()` (triggered by error events) from scheduling a reconnect timer after the channel has already been torn down. Additionally, the reconnect timer callback itself checks `channel.closed` before calling `openChannel`.
```typescript
interface Channel {
// ...
closed: boolean;
}
function closeChannel(channel: Channel): void {
channel.closed = true; // Set BEFORE es.close() to block synchronous reconnect
if (channel.es) channel.es.close();
// ...
}
function forceReconnect(channel: Channel): void {
// ...
if (channel.closed) return; // Guard: do not schedule reconnect on teardown
// ...
channel.reconnectTimer = setTimeout(() => {
if (channel.closed) return; // Guard: do not reopen after teardown
// ...
}, RECONNECT_DELAY_MS);
}
```
**Key insight**: The `closed` check at the top of `forceReconnect` prevents the timer from being set when `closeChannel` runs synchronously after an error event. Without this, the reconnect timer could be set after the channel was already marked for teardown, creating a zombie connection.
### 2. `active` flag in useTasks SSE effect
The SSE effect in `useTasks.ts` uses an `active` boolean to prevent stale `onReconnect` callbacks from firing after the effect has cleaned up (e.g., when `sseEnabled` flips to `false`):
```typescript
useEffect(() => {
let active = true;
// ...
return subscribeSse(url, {
onReconnect: () => {
if (!active) return; // Block stale callbacks after effect cleanup
// ...
},
});
return () => { active = false; }; // Runs after subscribeSse's cleanup
}, [projectId, sseEnabled]);
```
**Important**: The cleanup function that sets `active = false` must be returned **after** the `subscribeSse()` return value — not as a second `return` (which would be unreachable). The unsubscribe from `subscribeSse` is called first, then `active` is set to false.
### Files affected
- `packages/dashboard/app/sse-bus.ts``closed` flag on Channel
- `packages/dashboard/app/hooks/useTasks.ts``active` flag in SSE effect
- Tests: `packages/dashboard/app/__tests__/sse-bus.test.ts`, `packages/dashboard/app/hooks/__tests__/useTasks.test.ts`

View File

@@ -1,4 +1,4 @@
import { describe, it, expect, afterEach } from "vitest";
import { describe, it, expect, afterEach, vi, beforeEach } from "vitest";
import { MockEventSource } from "../../vitest.setup";
import { subscribeSse, __resetSseBus, __sseBusChannelCount } from "../sse-bus";
@@ -129,4 +129,43 @@ describe("sse-bus", () => {
expect(reconnects).toBe(1);
unsub();
});
it("does not set a reconnect timer after closeChannel is called", () => {
vi.useFakeTimers();
const url = "/api/events";
const unsub = subscribeSse(url, { events: { "task:created": () => {} } });
const es = MockEventSource.instances[0];
// Simulate an error which triggers forceReconnect (and schedules reconnect timer)
es._emit("error");
// Immediately unsubscribe (triggers closeChannel which sets closed=true)
unsub();
// Advance timers past RECONNECT_DELAY_MS (3 seconds)
vi.advanceTimersByTime(4_000);
// No new EventSource should be created — the closed flag prevented reconnect
expect(MockEventSource.instances).toHaveLength(1);
vi.useRealTimers();
});
it("does not leak channels on rapid subscribe/unsubscribe cycles", () => {
const url = "/api/events";
for (let i = 0; i < 5; i++) {
const unsub = subscribeSse(url, { events: { "task:created": () => {} } });
unsub();
}
expect(__sseBusChannelCount()).toBe(0);
// After 5 subscribe/unsubscribe cycles the channel has been opened and closed 5
// times, creating 5 EventSource instances (channel is deleted from the map on
// every unsubscribe). The key leak concern is zombie reconnect timers — verify
// no additional instances are created when reconnect delay fires after teardown.
const countBeforeTimers = MockEventSource.instances.length;
vi.useFakeTimers();
vi.advanceTimersByTime(4_000);
vi.useRealTimers();
// No new instances should be created by the (blocked) reconnect timer
expect(MockEventSource.instances.length).toBe(countBeforeTimers);
});
});

View File

@@ -1621,5 +1621,70 @@ describe("useTasks", () => {
expect(MockEventSource.instances.length).toBe(0);
});
it("does not grow EventSource instances on repeated sseEnabled toggles", async () => {
vi.useFakeTimers({ shouldAdvanceTime: true });
const { rerender } = renderHook(
({ sseEnabled }: { sseEnabled?: boolean }) => useTasks({ projectId: "test-project", sseEnabled }),
{ initialProps: { sseEnabled: true } }
);
await waitFor(() => {
expect(MockEventSource.instances.length).toBeGreaterThanOrEqual(1);
});
// Toggle false → true multiple times
for (let i = 0; i < 3; i++) {
await act(async () => {
rerender({ sseEnabled: false });
});
await act(async () => {
rerender({ sseEnabled: true });
});
}
const countAfterToggles = MockEventSource.instances.length;
// Advance fake timers — no pending reconnect timers should fire after teardown
vi.advanceTimersByTime(4_000);
// Count must not grow after timer advancement (the closed flag in sse-bus prevents
// reconnect timers from creating zombie connections after channel teardown).
expect(MockEventSource.instances.length).toBe(countAfterToggles);
vi.useRealTimers();
});
it("does not trigger onReconnect refetch after sseEnabled flips to false", async () => {
vi.useFakeTimers({ shouldAdvanceTime: true });
const { rerender } = renderHook(
({ sseEnabled }: { sseEnabled?: boolean }) => useTasks({ projectId: "test-project", sseEnabled }),
{ initialProps: { sseEnabled: true } }
);
await waitFor(() => {
expect(MockEventSource.instances.length).toBeGreaterThanOrEqual(1);
});
const es = MockEventSource.instances[0];
mockFetchTasks.mockClear();
// Simulate an error on the EventSource (triggers reconnect flow)
act(() => {
es._emit("error");
});
expect(mockFetchTasks).toHaveBeenCalledTimes(1); // onReconnect fires once during error
// Before the reconnect timer fires, flip sseEnabled to false
await act(async () => {
rerender({ sseEnabled: false });
});
// Advance timers past RECONNECT_DELAY_MS (3 seconds)
vi.advanceTimersByTime(4_000);
// No additional fetchTasks should have been called — active flag blocked it
expect(mockFetchTasks).toHaveBeenCalledTimes(1);
vi.useRealTimers();
});
});
});

View File

@@ -166,6 +166,9 @@ export function useTasks(options?: UseTasksOptions) {
const query = projectId ? `?projectId=${encodeURIComponent(projectId)}` : "";
const isStale = () => projectContextVersionRef.current !== contextVersionAtStart;
// Guards against reconnect callbacks firing after the effect has cleaned up
// (e.g., sseEnabled flipped to false during a pending reconnect timer in sse-bus).
let active = true;
const handleCreated = (e: MessageEvent) => {
if (isStale()) return;
@@ -256,7 +259,7 @@ export function useTasks(options?: UseTasksOptions) {
);
};
return subscribeSse(`/api/events${query}`, {
const unsubscribe = subscribeSse(`/api/events${query}`, {
events: {
"task:created": handleCreated,
"task:moved": handleMoved,
@@ -265,10 +268,15 @@ export function useTasks(options?: UseTasksOptions) {
"task:merged": handleMerged,
},
onReconnect: () => {
if (!active) return;
if (isStale()) return;
void refreshTasksRef.current();
},
});
return () => {
active = false;
unsubscribe();
};
}, [projectId, sseEnabled]);
const createTask = useCallback(async (input: TaskCreateInput): Promise<Task> => {

View File

@@ -28,6 +28,8 @@ interface Channel {
heartbeatTimer: ReturnType<typeof setTimeout> | null;
reconnectTimer: ReturnType<typeof setTimeout> | null;
hasOpenedOnce: boolean;
/** Set true at the start of closeChannel to prevent reconnect after teardown. */
closed: boolean;
}
const channels = new Map<string, Channel>();
@@ -50,6 +52,7 @@ function forceReconnect(channel: Channel): void {
}
channel.nativeListeners.clear();
if (channel.closed) return;
if (channel.subscribers.size === 0 || channel.reconnectTimer) return;
// A teardown means events may have been missed while the stream was
@@ -59,12 +62,14 @@ function forceReconnect(channel: Channel): void {
channel.reconnectTimer = setTimeout(() => {
channel.reconnectTimer = null;
if (channel.closed) return;
if (channel.subscribers.size > 0) openChannel(channel);
}, RECONNECT_DELAY_MS);
}
function openChannel(channel: Channel): void {
if (channel.es) return;
if (channel.closed) return;
if (channel.reconnectTimer) {
clearTimeout(channel.reconnectTimer);
channel.reconnectTimer = null;
@@ -123,6 +128,7 @@ function reattachNativeListeners(channel: Channel): void {
}
function closeChannel(channel: Channel): void {
channel.closed = true;
if (channel.heartbeatTimer) clearTimeout(channel.heartbeatTimer);
if (channel.reconnectTimer) clearTimeout(channel.reconnectTimer);
if (channel.es) channel.es.close();
@@ -158,6 +164,7 @@ export function subscribeSse(url: string, sub: SseSubscription = {}): () => void
heartbeatTimer: null,
reconnectTimer: null,
hasOpenedOnce: false,
closed: false,
};
channels.set(url, channel);
}