root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(FN-4921): merge fusion/fn-4921

Browse Source
This commit is contained in:
gsxdsm
2026-05-17 19:11:59 -07:00
parent bf0e9a3e37
commit f9cf3f1bde
9 changed files with 644 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
packages/dashboard/app/__tests__/sse-bus.test.ts
View File

@@ -1,6 +1,7 @@
import { describe, it, expect, afterEach, vi, beforeEach } from "vitest";
import { MockEventSource } from "../../vitest.setup";
import { subscribeSse, __resetSseBus, __sseBusChannelCount } from "../sse-bus";
import { clearTraces, getTraces } from "../utils/dashboardTraceBuffer";
function expectEventsUrl(url: string, projectId?: string): void {
const parsed = new URL(url, "http://localhost");
@@ -11,6 +12,7 @@ function expectEventsUrl(url: string, projectId?: string): void {
beforeEach(() => {
window.sessionStorage.clear();
clearTraces();
});
afterEach(() => {
@@ -173,15 +175,76 @@ describe("sse-bus", () => {
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);
});
it("reopens subscribed channel on pageshow even when event.persisted is false", () => {
subscribeSse("/api/events?projectId=p1", {});
expect(MockEventSource.instances).toHaveLength(1);
window.dispatchEvent(new Event("pagehide"));
window.dispatchEvent(new PageTransitionEvent("pageshow", { persisted: false }));
expect(MockEventSource.instances).toHaveLength(2);
});
it("does not reopen on pageshow when there are no subscribers", () => {
const unsub = subscribeSse("/api/events", {});
unsub();
const count = MockEventSource.instances.length;
window.dispatchEvent(new PageTransitionEvent("pageshow", { persisted: false }));
expect(MockEventSource.instances).toHaveLength(count);
});
it("reopens null channels on visibilitychange visible", () => {
subscribeSse("/api/events", {});
window.dispatchEvent(new Event("pagehide"));
Object.defineProperty(document, "visibilityState", { value: "visible", configurable: true });
document.dispatchEvent(new Event("visibilitychange"));
expect(MockEventSource.instances).toHaveLength(2);
});
it("reopens CLOSED channels on visibilitychange visible", () => {
subscribeSse("/api/events", {});
const first = MockEventSource.instances[0]!;
first.readyState = MockEventSource.CLOSED;
Object.defineProperty(document, "visibilityState", { value: "visible", configurable: true });
document.dispatchEvent(new Event("visibilitychange"));
expect(MockEventSource.instances).toHaveLength(2);
});
it("does not reopen on visibilitychange hidden", () => {
subscribeSse("/api/events", {});
const count = MockEventSource.instances.length;
Object.defineProperty(document, "visibilityState", { value: "hidden", configurable: true });
document.dispatchEvent(new Event("visibilitychange"));
expect(MockEventSource.instances).toHaveLength(count);
});
it("pushes traces for pageshow, visibilitychange, and forceReconnect", () => {
subscribeSse("/api/events", {});
const first = MockEventSource.instances[0]!;
window.dispatchEvent(new PageTransitionEvent("pageshow", { persisted: true }));
Object.defineProperty(document, "visibilityState", { value: "visible", configurable: true });
document.dispatchEvent(new Event("visibilitychange"));
first._emit("error");
const events = getTraces().map((entry) => entry.event);
expect(events).toContain("pageshow");
expect(events).toContain("visibilitychange");
expect(events).toContain("forceReconnect");
});
});

82
packages/dashboard/app/__tests__/versionCheck.test.ts
View File

@@ -13,7 +13,9 @@ import {
setAutoReloadEnabled,
_isAutoReloadEnabled,
MIN_CHECK_INTERVAL_MS,
_resetMismatchState,
} from "../versionCheck";
import { clearTraces, getTraces } from "../utils/dashboardTraceBuffer";
// Mock __BUILD_VERSION__ (declared as const in the module)
vi.stubGlobal("__BUILD_VERSION__", "test-build-abc123");
@@ -100,7 +102,7 @@ describe("consumeVersionUpdateFlag", () => {
});
});
describe("checkVersion cooldown", () => {
describe("checkVersion cooldown + mismatch gating", () => {
const reloadSpy = vi.fn();
beforeEach(() => {
@@ -108,6 +110,8 @@ describe("checkVersion cooldown", () => {
window.sessionStorage.clear();
reloadSpy.mockClear();
_resetCheckState();
_resetMismatchState();
clearTraces();
// Ensure tab is visible
Object.defineProperty(document, "visibilityState", { value: "visible", configurable: true });
});
@@ -176,6 +180,82 @@ describe("checkVersion cooldown", () => {
await checkVersion();
expect(reloadSpy).not.toHaveBeenCalled();
expect(getTraces().some((t) => t.event === "remote-unavailable")).toBe(true);
});
it("single mismatch pushes trace and does not reload", async () => {
const fetchSpy = vi.fn().mockResolvedValue({
ok: true,
headers: new Headers({ "content-type": "application/json" }),
json: () => Promise.resolve({ version: "different-version" }),
});
vi.stubGlobal("fetch", fetchSpy);
await checkVersion("focus");
expect(reloadSpy).not.toHaveBeenCalled();
const mismatchTrace = getTraces().find((t) => t.event === "mismatch");
expect(mismatchTrace?.detail).toMatchObject({ trigger: "focus", remote: "different-version" });
expect(getTraces().some((t) => t.event === "mismatch-pending")).toBe(true);
});
it("reloads once after two consecutive identical mismatches", async () => {
vi.useFakeTimers();
const fetchSpy = vi.fn().mockResolvedValue({
ok: true,
headers: new Headers({ "content-type": "application/json" }),
json: () => Promise.resolve({ version: "different-version" }),
});
vi.stubGlobal("fetch", fetchSpy);
await checkVersion("initial");
vi.advanceTimersByTime(MIN_CHECK_INTERVAL_MS + 1);
await checkVersion("visibilitychange");
expect(reloadSpy).toHaveBeenCalledTimes(1);
vi.useRealTimers();
});
it("mismatch then match resets gating", async () => {
vi.useFakeTimers();
const fetchSpy = vi.fn()
.mockResolvedValueOnce({ ok: true, headers: new Headers({ "content-type": "application/json" }), json: () => Promise.resolve({ version: "different-version" }) })
.mockResolvedValueOnce({ ok: true, headers: new Headers({ "content-type": "application/json" }), json: () => Promise.resolve({ version: "test-build-abc123" }) })
.mockResolvedValueOnce({ ok: true, headers: new Headers({ "content-type": "application/json" }), json: () => Promise.resolve({ version: "different-version" }) })
.mockResolvedValueOnce({ ok: true, headers: new Headers({ "content-type": "application/json" }), json: () => Promise.resolve({ version: "different-version" }) });
vi.stubGlobal("fetch", fetchSpy);
await checkVersion("initial");
vi.advanceTimersByTime(MIN_CHECK_INTERVAL_MS + 1);
await checkVersion("focus");
vi.advanceTimersByTime(MIN_CHECK_INTERVAL_MS + 1);
await checkVersion("focus");
expect(reloadSpy).not.toHaveBeenCalled();
vi.advanceTimersByTime(MIN_CHECK_INTERVAL_MS + 1);
await checkVersion("focus");
expect(reloadSpy).toHaveBeenCalledTimes(1);
vi.useRealTimers();
});
it("captures trigger source in mismatch traces", async () => {
vi.useFakeTimers();
const fetchSpy = vi.fn()
.mockResolvedValue({ ok: true, headers: new Headers({ "content-type": "application/json" }), json: () => Promise.resolve({ version: "different-version" }) });
vi.stubGlobal("fetch", fetchSpy);
await checkVersion("initial");
vi.advanceTimersByTime(MIN_CHECK_INTERVAL_MS + 1);
await checkVersion("visibilitychange");
vi.advanceTimersByTime(MIN_CHECK_INTERVAL_MS + 1);
_resetMismatchState();
await checkVersion("focus");
const mismatchTriggers = getTraces()
.filter((entry) => entry.event === "mismatch")
.map((entry) => entry.detail.trigger);
expect(mismatchTriggers).toEqual(expect.arrayContaining(["initial", "visibilitychange", "focus"]));
vi.useRealTimers();
});
});

69
packages/dashboard/app/hooks/__tests__/useTasks.stale-trace.test.ts Normal file
View File

@@ -0,0 +1,69 @@
import { describe, it, expect, vi, beforeEach } from "vitest";
import { renderHook, act, waitFor } from "@testing-library/react";
const subscribeCalls: Array<{
handlers: {
events?: Record<string, (event: MessageEvent) => void>;
onReconnect?: () => void;
};
}> = [];
vi.mock("../../sse-bus", () => ({
subscribeSse: (_url: string, handlers: { events?: Record<string, (event: MessageEvent) => void>; onReconnect?: () => void }) => {
subscribeCalls.push({ handlers });
return () => {};
},
}));
vi.mock("../../api", async (importOriginal) => {
const { createDashboardApiMock } = await import("../../test/mockApi");
return createDashboardApiMock(() => importOriginal<typeof import("../../api")>(), {
fetchTasks: vi.fn().mockResolvedValue([]),
});
});
describe("useTasks stale trace instrumentation", () => {
beforeEach(() => {
subscribeCalls.length = 0;
vi.resetModules();
});
it("emits dropped-stale-event trace when stale subscription handler fires after project switch", async () => {
const traceBuffer = await import("../../utils/dashboardTraceBuffer");
traceBuffer.clearTraces();
const { useTasks } = await import("../useTasks");
const { rerender } = renderHook(
({ projectId }: { projectId: string }) => useTasks({ projectId }),
{ initialProps: { projectId: "project-a" } },
);
await waitFor(() => {
expect(subscribeCalls.length).toBeGreaterThanOrEqual(1);
});
const staleCreated = subscribeCalls[0]?.handlers.events?.["task:created"];
expect(staleCreated).toBeTypeOf("function");
await act(async () => {
rerender({ projectId: "project-b" });
});
act(() => {
staleCreated?.({ data: JSON.stringify({ id: "FN-STALE", dependencies: [], steps: [], log: [] }) } as MessageEvent);
});
const staleTrace = traceBuffer
.getTraces()
.find((entry) => entry.source === "useTasks" && entry.event === "dropped-stale-event");
expect(staleTrace).toBeDefined();
expect(staleTrace?.detail).toMatchObject({
count: 1,
projectId: "project-a",
contextVersionAtStart: 0,
currentContextVersion: 1,
});
});
});

115
packages/dashboard/app/hooks/__tests__/useTasks.test.ts
View File

@@ -20,6 +20,7 @@ import { renderHook, act, waitFor } from "@testing-library/react";
import { useTasks } from "../useTasks";
import * as api from "../../api";
import * as swrCache from "../../utils/swrCache";
import { clearTraces, getTraces } from "../../utils/dashboardTraceBuffer";
import type { Task, Column } from "@fusion/core";
// Mock the api module
@@ -102,6 +103,7 @@ beforeEach(() => {
// Ensure we start with real timers for every test
vi.useRealTimers();
clearTraces();
});
afterEach(() => {
@@ -639,6 +641,77 @@ describe("useTasks", () => {
expect(result.current.tasks[0]?.title).toBe("Fresh title");
});
it("applies post-reconnect events without stale-drop trace when context matches", async () => {
vi.useFakeTimers();
mockFetchTasks.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
const { result } = renderHook(() => useTasks({ projectId: "project-a" }));
await act(async () => {
await flushPromises();
});
const first = MockEventSource.instances[0];
act(() => {
first._emit("error");
});
await act(async () => {
vi.advanceTimersByTime(3000);
await flushPromises();
});
const second = MockEventSource.instances[1];
act(() => {
second._emit("task:created", createMockTask({ id: "FN-POST" }));
});
expect(result.current.tasks.find((task) => task.id === "FN-POST")).toBeDefined();
expect(getTraces().some((entry) => entry.event === "dropped-stale-event")).toBe(false);
vi.useRealTimers();
});
it("refreshes immediately when project context changes while tab is hidden", async () => {
vi.useFakeTimers();
const visibilityState = { value: "visible" as VisibilityState };
Object.defineProperty(document, "visibilityState", {
configurable: true,
get: () => visibilityState.value,
});
mockFetchTasks.mockResolvedValue([]);
const { rerender } = renderHook(
({ projectId }: { projectId: string }) => useTasks({ projectId }),
{ initialProps: { projectId: "project-a" } },
);
await act(async () => {
await flushPromises();
});
mockFetchTasks.mockClear();
visibilityState.value = "hidden";
act(() => {
document.dispatchEvent(new Event("visibilitychange"));
});
await act(async () => {
rerender({ projectId: "project-b" });
await flushPromises();
});
mockFetchTasks.mockClear();
visibilityState.value = "visible";
act(() => {
document.dispatchEvent(new Event("visibilitychange"));
});
expect(mockFetchTasks).toHaveBeenCalledTimes(1);
expect(getTraces().some((entry) => entry.event === "visibility-context-version-changed")).toBe(true);
vi.useRealTimers();
});
describe("SSE event: task:updated", () => {
it("updates task fields", async () => {
@@ -1672,7 +1745,7 @@ describe("useTasks", () => {
mockFetchTasks.mockResolvedValueOnce([initialTask]).mockResolvedValueOnce([refreshedTask]);
const { result } = renderHook(() => useTasks());
const { result } = renderHook(() => useTasks({ sseEnabled: false }));
await act(async () => {
await Promise.resolve();
@@ -1702,7 +1775,7 @@ describe("useTasks", () => {
const initialTask = createMockTask({ id: "FN-001" });
mockFetchTasks.mockResolvedValueOnce([initialTask]);
renderHook(() => useTasks());
renderHook(() => useTasks({ sseEnabled: false }));
await act(async () => {
await Promise.resolve();
@@ -1723,7 +1796,7 @@ describe("useTasks", () => {
const initialTask = createMockTask({ id: "FN-001" });
mockFetchTasks.mockResolvedValue([initialTask]);
renderHook(() => useTasks());
renderHook(() => useTasks({ sseEnabled: false }));
await act(async () => {
await Promise.resolve();
@@ -1760,12 +1833,46 @@ describe("useTasks", () => {
expect(mockFetchTasks).toHaveBeenCalledTimes(2);
});
it("forces immediate refresh on visible when project context changed while hidden", async () => {
vi.useFakeTimers();
vi.setSystemTime(new Date("2026-01-01T00:00:00Z"));
mockFetchTasks
.mockResolvedValueOnce([createMockTask({ id: "FN-A", title: "A" })])
.mockResolvedValueOnce([createMockTask({ id: "FN-B", title: "B" })]);
const { rerender } = renderHook(
({ projectId }: { projectId?: string }) => useTasks({ projectId, sseEnabled: false }),
{ initialProps: { projectId: "project-a" } },
);
await act(async () => {
await Promise.resolve();
});
setVisibilityState("hidden");
await dispatchVisibilityChange();
await act(async () => {
rerender({ projectId: "project-b" });
});
mockFetchTasks.mockClear();
vi.setSystemTime(new Date("2026-01-01T00:00:00.200Z"));
setVisibilityState("visible");
await dispatchVisibilityChange();
expect(mockFetchTasks).toHaveBeenCalledTimes(1);
expect(getTraces().some((entry) => entry.event === "visibility-context-version-changed")).toBe(true);
vi.useRealTimers();
});
it("cleans up visibility change listener on unmount", async () => {
mockFetchTasks.mockResolvedValueOnce([]);
const removeEventListenerSpy = vi.spyOn(document, "removeEventListener");
const { unmount } = renderHook(() => useTasks());
const { unmount } = renderHook(() => useTasks({ sseEnabled: false }));
await waitFor(() => {
expect(mockFetchTasks).toHaveBeenCalledTimes(1);

67
packages/dashboard/app/hooks/useTasks.ts
View File

@@ -4,6 +4,7 @@ import { normalizeColumn } from "@fusion/core";
import * as api from "../api";
import { subscribeSse } from "../sse-bus";
import { clearCache, readCache, SWR_CACHE_KEYS, SWR_TASKS_MAX_AGE_MS, writeCache } from "../utils/swrCache";
import { pushTrace } from "../utils/dashboardTraceBuffer";
const loggedTaskCacheHitProjects = new Set<string>();
@@ -115,15 +116,17 @@ export function useTasks(options?: UseTasksOptions) {
includeArchivedRef.current = includeArchived;
const tasksRef = useRef(tasks);
const fetchVersionRef = useRef(0);
// Tracks the project context version to detect stale SSE events after project switches.
// Incremented whenever projectId changes, invalidating any in-flight SSE handlers.
const projectContextVersionRef = useRef(0);
const lastVisibilityRefreshRef = useRef<number>(0);
const contextVersionAtLastVisibilityRef = useRef(projectContextVersionRef.current);
const droppedStaleEventsRef = useRef(0);
const searchQueryRef = useRef(searchQuery);
const refreshTasksRef = useRef<typeof refreshTasks>(null!);
// Tracks when task data was last confirmed fresh by the server.
// Used to prevent false positives in stuck detection when tab has been in background.
const lastFetchTimeMs = useRef<number | undefined>(undefined);
// Tracks the project context version to detect stale SSE events after project switches.
// Incremented whenever projectId changes, invalidating any in-flight SSE handlers.
const projectContextVersionRef = useRef(0);
// Track previous projectId to detect changes
const previousProjectIdRef = useRef<string | undefined>(projectId);
tasksRef.current = tasks;
@@ -226,6 +229,22 @@ export function useTasks(options?: UseTasksOptions) {
const handleVisibilityChange = () => {
if (document.visibilityState !== "visible") {
contextVersionAtLastVisibilityRef.current = projectContextVersionRef.current;
return;
}
const previousContextVersion = contextVersionAtLastVisibilityRef.current;
const contextChangedWhileHidden = previousContextVersion !== projectContextVersionRef.current;
contextVersionAtLastVisibilityRef.current = projectContextVersionRef.current;
if (contextChangedWhileHidden) {
lastVisibilityRefreshRef.current = Date.now();
pushTrace("useTasks", "visibility-context-version-changed", {
projectId,
previousContextVersion,
currentContextVersion: projectContextVersionRef.current,
});
void refreshTasks();
return;
}
@@ -254,10 +273,19 @@ export function useTasks(options?: UseTasksOptions) {
useEffect(() => {
if (sseEnabled === false) return;
const contextVersionAtStart = projectContextVersionRef.current;
let contextVersionAtStart = projectContextVersionRef.current;
const query = projectId ? `?projectId=${encodeURIComponent(projectId)}` : "";
const isStale = () => projectContextVersionRef.current !== contextVersionAtStart;
const traceDroppedStaleEvent = () => {
droppedStaleEventsRef.current += 1;
pushTrace("useTasks", "dropped-stale-event", {
count: droppedStaleEventsRef.current,
contextVersionAtStart,
currentContextVersion: projectContextVersionRef.current,
projectId,
});
};
// 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;
@@ -266,7 +294,10 @@ export function useTasks(options?: UseTasksOptions) {
// effect unmounts, these handlers must not fire refreshTasks into a
// missions-only view where the SSE should be inactive.
const handleCreated = (e: MessageEvent) => {
if (isStale()) return;
if (isStale()) {
traceDroppedStaleEvent();
return;
}
const task = normalizeTask(JSON.parse(e.data) as Task);
if (searchQueryRef.current) {
void refreshTasksRef.current({ searchQueryOverride: searchQueryRef.current });
@@ -292,7 +323,10 @@ export function useTasks(options?: UseTasksOptions) {
};
const handleMoved = (e: MessageEvent) => {
if (isStale()) return;
if (isStale()) {
traceDroppedStaleEvent();
return;
}
if (searchQueryRef.current) {
void refreshTasksRef.current({ searchQueryOverride: searchQueryRef.current });
return;
@@ -315,7 +349,10 @@ export function useTasks(options?: UseTasksOptions) {
};
const handleUpdated = (e: MessageEvent) => {
if (isStale()) return;
if (isStale()) {
traceDroppedStaleEvent();
return;
}
if (searchQueryRef.current) {
void refreshTasksRef.current({ searchQueryOverride: searchQueryRef.current });
return;
@@ -337,7 +374,10 @@ export function useTasks(options?: UseTasksOptions) {
};
const handleDeleted = (e: MessageEvent) => {
if (isStale()) return;
if (isStale()) {
traceDroppedStaleEvent();
return;
}
if (searchQueryRef.current) {
void refreshTasksRef.current({ searchQueryOverride: searchQueryRef.current });
return;
@@ -347,7 +387,10 @@ export function useTasks(options?: UseTasksOptions) {
};
const handleMerged = (e: MessageEvent) => {
if (isStale()) return;
if (isStale()) {
traceDroppedStaleEvent();
return;
}
if (searchQueryRef.current) {
void refreshTasksRef.current({ searchQueryOverride: searchQueryRef.current });
return;
@@ -377,8 +420,12 @@ export function useTasks(options?: UseTasksOptions) {
// Guard onReconnect against stale SSE callbacks: do not call refreshTasks
// if the SSE was disabled or the effect unmounted while reconnect was pending.
onReconnect: () => {
contextVersionAtStart = projectContextVersionRef.current;
if (!active) return;
if (isStale()) return;
if (isStale()) {
traceDroppedStaleEvent();
return;
}
void refreshTasksRef.current();
},
});

120
packages/dashboard/app/sse-bus.ts
View File

@@ -1,4 +1,5 @@
import { appendTokenQuery } from "./auth";
import { pushTrace } from "./utils/dashboardTraceBuffer";
// Shared EventSource multiplexer.
//
@@ -16,6 +17,7 @@ const HEARTBEAT_TIMEOUT_MS = 45_000;
const RECONNECT_DELAY_MS = 3_000;
const CLIENT_KEEPALIVE_INTERVAL_MS = 2_000;
const CLIENT_KEEPALIVE_TIMEOUT_MS = 1_500;
const VISIBILITY_REOPEN_DEDUPE_MS = 1_000;
const CLIENT_ID_STORAGE_KEY = "fusion:sse-client-id";
let memoryClientId: string | null = null;
@@ -41,6 +43,7 @@ interface Channel {
}
const channels = new Map<string, Channel>();
let lastVisibilityReopenAt = 0;
function createClientId(): string {
const cryptoApi = typeof globalThis !== "undefined" ? globalThis.crypto : undefined;
@@ -140,29 +143,36 @@ function stopClientKeepalive(channel: Channel): void {
}
}
function sendClientKeepalive(channel: Channel): void {
if (typeof window === "undefined" || typeof window.fetch !== "function") return;
async function probeClientKeepalive(channel: Channel): Promise<"ok" | "definite-dead" | "inconclusive"> {
if (typeof window === "undefined" || typeof window.fetch !== "function") return "inconclusive";
const url = createControlUrl(channel.url, "keepalive");
if (!url) return;
if (!url) return "inconclusive";
const controller = typeof AbortController !== "undefined" ? new AbortController() : null;
const timeout = controller
? window.setTimeout(() => controller.abort(), CLIENT_KEEPALIVE_TIMEOUT_MS)
: null;
void window.fetch(url, {
method: "POST",
cache: "no-store",
signal: controller?.signal,
}).catch(() => {
// If this page is suspended or the network drops, the server-side stale
// timer will reap the stream and EventSource will reconnect later.
}).finally(() => {
try {
const res = await window.fetch(url, {
method: "POST",
cache: "no-store",
signal: controller?.signal,
});
return res.ok ? "ok" : "definite-dead";
} catch (error) {
const aborted = error instanceof DOMException && error.name === "AbortError";
return aborted ? "inconclusive" : "definite-dead";
} finally {
if (timeout !== null) {
window.clearTimeout(timeout);
}
});
}
}
function sendClientKeepalive(channel: Channel): void {
void probeClientKeepalive(channel);
}
function startClientKeepalive(channel: Channel): void {
@@ -182,8 +192,10 @@ function startClientKeepalive(channel: Channel): void {
// few refreshes the browser hits its 6-connection-per-origin limit
// and every subsequent fetch stalls. Using `pagehide` (fires reliably
// on bfcache navigations too) plus `beforeunload` as a fallback.
if (typeof window !== "undefined") {
if (typeof window !== "undefined" && typeof document !== "undefined") {
const closeAllChannels = () => {
console.info("[sse-bus] pagehide", { channelCount: channels.size });
pushTrace("sse-bus", "pagehide", { channelCount: channels.size });
for (const channel of Array.from(channels.values())) {
if (channel.closed) continue;
stopClientKeepalive(channel);
@@ -199,27 +211,72 @@ if (typeof window !== "undefined") {
channel.closed = true;
}
};
const reopenPersistedChannels = (event: PageTransitionEvent) => {
if (!event.persisted) return;
const reopenSubscribedChannels = (event: PageTransitionEvent) => {
console.info("[sse-bus] pageshow", { persisted: event.persisted, channelCount: channels.size });
pushTrace("sse-bus", "pageshow", { persisted: event.persisted, channelCount: channels.size });
for (const channel of Array.from(channels.values())) {
if (channel.subscribers.size === 0) continue;
if (channel.es !== null && !channel.closed) continue;
channel.closed = false;
openChannel(channel);
}
};
const reopenVisibleChannels = () => {
if (document.visibilityState !== "visible") return;
const now = Date.now();
if (now - lastVisibilityReopenAt < VISIBILITY_REOPEN_DEDUPE_MS) return;
lastVisibilityReopenAt = now;
console.info("[sse-bus] visibilitychange", { visibilityState: document.visibilityState, channelCount: channels.size });
pushTrace("sse-bus", "visibilitychange", { visibilityState: document.visibilityState, channelCount: channels.size });
for (const channel of Array.from(channels.values())) {
if (channel.subscribers.size === 0) continue;
if (channel.es === null || channel.es.readyState === EventSource.CLOSED) {
channel.closed = false;
if (channel.es && channel.es.readyState === EventSource.CLOSED) {
channel.es = null;
}
openChannel(channel);
continue;
}
void probeClientKeepalive(channel).then((status) => {
if (status === "definite-dead") {
forceReconnect(channel, "external");
}
});
}
};
window.addEventListener("pagehide", closeAllChannels);
window.addEventListener("beforeunload", closeAllChannels);
window.addEventListener("pageshow", reopenPersistedChannels);
window.addEventListener("pageshow", reopenSubscribedChannels);
document.addEventListener("visibilitychange", reopenVisibleChannels);
}
function resetHeartbeat(channel: Channel): void {
if (channel.heartbeatTimer) clearTimeout(channel.heartbeatTimer);
channel.heartbeatTimer = setTimeout(() => {
forceReconnect(channel);
forceReconnect(channel, "heartbeat-timeout");
}, HEARTBEAT_TIMEOUT_MS);
}
function forceReconnect(channel: Channel): void {
function forceReconnect(channel: Channel, cause: "heartbeat-timeout" | "error" | "external" = "external"): void {
console.warn("[sse-bus] forceReconnect", {
cause,
url: channel.url,
subscriberCount: channel.subscribers.size,
hasOpenedOnce: channel.hasOpenedOnce,
});
pushTrace("sse-bus", "forceReconnect", {
cause,
url: channel.url,
subscriberCount: channel.subscribers.size,
hasOpenedOnce: channel.hasOpenedOnce,
});
if (channel.heartbeatTimer) {
clearTimeout(channel.heartbeatTimer);
channel.heartbeatTimer = null;
@@ -258,6 +315,19 @@ function forceReconnect(channel: Channel): void {
}
function openChannel(channel: Channel): void {
pushTrace("sse-bus", "openChannel", {
url: channel.url,
subscriberCount: channel.subscribers.size,
hasOpenedOnce: channel.hasOpenedOnce,
closed: channel.closed,
hasEventSource: channel.es !== null,
});
console.info("[sse-bus] openChannel", {
url: channel.url,
subscriberCount: channel.subscribers.size,
hasOpenedOnce: channel.hasOpenedOnce,
closed: channel.closed,
});
if (channel.es) return;
if (channel.closed) return;
if (channel.reconnectTimer) {
@@ -287,8 +357,7 @@ function openChannel(channel: Channel): void {
// Any error triggers a forced reconnect cycle — matches the pre-bus
// behavior in useTasks and ensures the stream recovers even when
// EventSource's own retry has stalled.
forceReconnect(channel);
});
forceReconnect(channel, "error"); });
// Unnamed `message` events and server "heartbeat" events both count as
// liveness signals, regardless of whether a subscriber registered them.
@@ -322,6 +391,16 @@ function reattachNativeListeners(channel: Channel): void {
}
function closeChannel(channel: Channel): void {
pushTrace("sse-bus", "closeChannel", {
url: channel.url,
subscriberCount: channel.subscribers.size,
hasOpenedOnce: channel.hasOpenedOnce,
});
console.info("[sse-bus] closeChannel", {
url: channel.url,
subscriberCount: channel.subscribers.size,
hasOpenedOnce: channel.hasOpenedOnce,
});
channel.closed = true;
if (channel.heartbeatTimer) clearTimeout(channel.heartbeatTimer);
stopClientKeepalive(channel);
@@ -414,6 +493,7 @@ export function subscribeSse(url: string, sub: SseSubscription = {}): () => void
export function __resetSseBus(): void {
for (const channel of Array.from(channels.values())) closeChannel(channel);
memoryClientId = null;
lastVisibilityReopenAt = 0;
}
/** Test-only: inspect the number of live channels. */

47
packages/dashboard/app/utils/__tests__/dashboardTraceBuffer.test.ts Normal file
View File

@@ -0,0 +1,47 @@
/**
* @vitest-environment jsdom
*/
import { beforeEach, describe, expect, it } from "vitest";
import { clearTraces, getTraces, pushTrace } from "../dashboardTraceBuffer";
describe("dashboardTraceBuffer", () => {
beforeEach(() => {
clearTraces();
});
it("appends trace entries", () => {
pushTrace("versionCheck", "mismatch", { local: "a", remote: "b" });
const traces = getTraces();
expect(traces).toHaveLength(1);
expect(traces[0]).toMatchObject({
source: "versionCheck",
event: "mismatch",
detail: { local: "a", remote: "b" },
});
expect(typeof traces[0].ts).toBe("string");
});
it("caps entries at 200 and drops oldest", () => {
for (let i = 0; i < 205; i += 1) {
pushTrace("sse-bus", "event", { idx: i });
}
const traces = getTraces();
expect(traces).toHaveLength(200);
expect(traces[0]?.detail).toEqual({ idx: 5 });
expect(traces[199]?.detail).toEqual({ idx: 204 });
});
it("exposes traces via window.__fusionDebug.dashboardTraces.get", () => {
pushTrace("useTasks", "dropped-stale-event", { count: 1 });
const debugApi = window.__fusionDebug?.dashboardTraces;
expect(debugApi).toBeDefined();
expect(debugApi?.get()).toHaveLength(1);
expect(debugApi?.get()[0]).toMatchObject({
source: "useTasks",
event: "dropped-stale-event",
});
});
});

49
packages/dashboard/app/utils/dashboardTraceBuffer.ts Normal file
View File

@@ -0,0 +1,49 @@
export type TraceEntry = {
ts: string;
source: string;
event: string;
detail: Record<string, unknown>;
};
const TRACE_CAP = 200;
const traces: TraceEntry[] = [];
export function pushTrace(source: string, event: string, detail: Record<string, unknown>): void {
traces.push({
ts: new Date().toISOString(),
source,
event,
detail,
});
if (traces.length > TRACE_CAP) {
traces.splice(0, traces.length - TRACE_CAP);
}
}
export function getTraces(): TraceEntry[] {
return [...traces];
}
export function clearTraces(): void {
traces.length = 0;
}
declare global {
interface Window {
__fusionDebug?: {
dashboardTraces?: {
get: typeof getTraces;
clear: typeof clearTraces;
};
};
}
}
if (typeof window !== "undefined") {
window.__fusionDebug ??= {};
window.__fusionDebug.dashboardTraces = {
get: getTraces,
clear: clearTraces,
};
}

74
packages/dashboard/app/versionCheck.ts
View File

@@ -1,3 +1,5 @@
import { pushTrace } from "./utils/dashboardTraceBuffer";
declare const __BUILD_VERSION__: string;
const RELOAD_FLAG = "fusion:version-reload";
@@ -28,6 +30,7 @@ export function _resetState(): void {
lastCheckTime = 0;
checkInFlight = false;
autoReloadEnabled = true;
_resetMismatchState();
}
export function consumeVersionUpdateFlag(): boolean {
@@ -43,7 +46,9 @@ export function consumeVersionUpdateFlag(): boolean {
}
export function reloadOnce(reason: string): void {
if (sessionStorage.getItem(RELOAD_FLAG)) {
const alreadySet = Boolean(sessionStorage.getItem(RELOAD_FLAG));
pushTrace("versionCheck", "reload-attempt", { reason, alreadySet });
if (alreadySet) {
console.warn("[versionCheck] reload already attempted, suppressing", reason);
return;
}
@@ -118,28 +123,73 @@ async function bootstrapAutoReloadSetting(): Promise<void> {
export const MIN_CHECK_INTERVAL_MS = 60_000; // 1 minute
let lastCheckTime = 0;
let checkInFlight = false;
let lastMismatchedRemote: string | null = null;
let lastMismatchAt = 0;
/** Exported for testing — resets internal cooldown state */
export function _resetCheckState(): void {
lastCheckTime = 0;
checkInFlight = false;
_resetMismatchState();
}
export async function checkVersion(): Promise<void> {
export function _resetMismatchState(): void {
lastMismatchedRemote = null;
lastMismatchAt = 0;
}
export async function checkVersion(trigger: "visibilitychange" | "focus" | "initial" = "initial"): Promise<void> {
if (checkInFlight || document.visibilityState !== "visible") return;
if (Date.now() - lastCheckTime < MIN_CHECK_INTERVAL_MS) return;
lastCheckTime = Date.now();
checkInFlight = true;
try {
const remote = await fetchRemoteVersion();
if (remote && remote !== __BUILD_VERSION__) {
try {
sessionStorage.setItem(VERSION_UPDATE_FLAG, "1");
} catch {
// ignore
}
reloadOnce(`build version changed: ${__BUILD_VERSION__} -> ${remote}`);
if (remote === null) {
pushTrace("versionCheck", "remote-unavailable", {
trigger,
visibilityState: document.visibilityState,
});
lastMismatchedRemote = null;
return;
}
if (remote === __BUILD_VERSION__) {
lastMismatchedRemote = null;
return;
}
pushTrace("versionCheck", "mismatch", {
local: __BUILD_VERSION__,
remote,
trigger,
visibilityState: document.visibilityState,
});
console.info("[versionCheck] mismatch", { local: __BUILD_VERSION__, remote, trigger });
if (lastMismatchedRemote !== remote) {
lastMismatchedRemote = remote;
lastMismatchAt = Date.now();
pushTrace("versionCheck", "mismatch-pending", {
local: __BUILD_VERSION__,
remote,
trigger,
visibilityState: document.visibilityState,
});
return;
}
pushTrace("versionCheck", "mismatch-confirmed", {
remote,
trigger,
elapsedMs: Date.now() - lastMismatchAt,
});
try {
sessionStorage.setItem(VERSION_UPDATE_FLAG, "1");
} catch {
// ignore
}
reloadOnce(`build version changed: ${__BUILD_VERSION__} -> ${remote}`);
} finally {
checkInFlight = false;
}
@@ -152,11 +202,11 @@ export function installVersionCheck(): void {
// Clear stale flag once a fresh page has rendered successfully.
window.setTimeout(() => sessionStorage.removeItem(RELOAD_FLAG), 5_000);
document.addEventListener("visibilitychange", () => {
void checkVersion();
void checkVersion("visibilitychange");
});
window.addEventListener("focus", () => {
void checkVersion();
void checkVersion("focus");
});
// Initial check after load to catch tabs restored from bfcache.
window.setTimeout(() => void checkVersion(), 2_000);
window.setTimeout(() => void checkVersion("initial"), 2_000);
}