diff --git a/apps/fabro-web/app/data/runs.ts b/apps/fabro-web/app/data/runs.ts index 3922b7cd3..b397c5ecc 100644 --- a/apps/fabro-web/app/data/runs.ts +++ b/apps/fabro-web/app/data/runs.ts @@ -1,8 +1,10 @@ import { formatElapsedSecs, formatDurationSecs } from "../lib/format"; -import type { - RunListItem, - RunStatus as ApiRunStatus, - RunSummary, +import { + BoardColumn, + type BoardColumn as ApiBoardColumn, + type RunListItem, + type RunStatus as ApiRunStatus, + type RunSummary, } from "@qltysh/fabro-api-client"; export type CiStatus = "passing" | "failing" | "pending"; @@ -37,9 +39,19 @@ export interface RunItem { sourceDirectory?: string; } -export type ColumnStatus = "initializing" | "running" | "blocked" | "succeeded" | "failed"; +export type ColumnStatus = ApiBoardColumn; + +export const columnStatuses = [ + BoardColumn.QUEUED, + BoardColumn.INITIALIZING, + BoardColumn.RUNNING, + BoardColumn.BLOCKED, + BoardColumn.SUCCEEDED, + BoardColumn.FAILED, +] as const satisfies readonly ColumnStatus[]; export const columnStatusDisplay: Record = { + queued: { label: "Queued", dot: "bg-fg-muted", text: "text-fg-muted" }, initializing: { label: "Initializing", dot: "bg-amber", text: "text-amber" }, running: { label: "Running", dot: "bg-teal-500", text: "text-teal-500" }, blocked: { label: "Blocked", dot: "bg-amber", text: "text-amber" }, @@ -113,6 +125,7 @@ export function columnForStatus(status: ApiRunStatus | null | undefined): Column switch (status?.kind) { case "submitted": case "queued": + return "queued"; case "starting": return "initializing"; case "running": diff --git a/apps/fabro-web/app/lib/board-events.test.tsx b/apps/fabro-web/app/lib/board-events.test.tsx index 853d300b9..497bb6984 100644 --- a/apps/fabro-web/app/lib/board-events.test.tsx +++ b/apps/fabro-web/app/lib/board-events.test.tsx @@ -4,7 +4,12 @@ import { shouldRefreshBoardForEvent, subscribeToBoardEvents, } from "./board-events"; +import { + createCrossTabSseCoordinator, + type BroadcastChannelLike, +} from "./cross-tab-sse"; import { queryKeys } from "./query-keys"; +import type { EventSourceLike } from "./sse"; type MessageHandler = ((event: { data: string }) => void) | null; @@ -21,6 +26,14 @@ class FakeEventSource { } } +class FakeBroadcastChannel implements BroadcastChannelLike { + onmessage: ((event: { data: unknown }) => void) | null = null; + + postMessage() {} + + close() {} +} + describe("shouldRefreshBoardForEvent", () => { test("refreshes board for run and interview status changes only", () => { expect(shouldRefreshBoardForEvent("run.running")).toBe(true); @@ -31,22 +44,28 @@ describe("shouldRefreshBoardForEvent", () => { }); describe("subscribeToBoardEvents", () => { - test("shares one source and invalidates the board runs key", () => { + test("coordinated mode shares one global source and invalidates the board runs key", async () => { const source = new FakeEventSource(); const created: string[] = []; const keys: string[] = []; + const coordinator = createCoordinator((url) => { + created.push(url); + return source; + }); const mutate = (key: string) => { keys.push(key); return Promise.resolve(); }; - const firstCleanup = subscribeToBoardEvents(mutate, (url) => { - created.push(url); - return source; - }, { debounceMs: 0 }); + const firstCleanup = subscribeToBoardEvents(mutate, () => { + throw new Error("source should be created by coordinator"); + }, { debounceMs: 0, coordinator }); const secondCleanup = subscribeToBoardEvents(mutate, () => { throw new Error("source should be reused"); - }, { debounceMs: 0 }); + }, { debounceMs: 0, coordinator }); + + await waitFor(() => created.length === 1); + keys.length = 0; source.emit({ event: "run.running" }); @@ -57,5 +76,68 @@ describe("subscribeToBoardEvents", () => { expect(source.closed).toBe(false); secondCleanup(); expect(source.closed).toBe(true); + coordinator.close(); + }); + + test("fallback mode preserves the existing shared board EventSource", () => { + const source = new FakeEventSource(); + const created: string[] = []; + const keys: string[] = []; + const coordinator = createFallbackCoordinator(); + const mutate = (key: string) => { + keys.push(key); + return Promise.resolve(); + }; + + const firstCleanup = subscribeToBoardEvents(mutate, (url) => { + created.push(url); + return source; + }, { debounceMs: 0, coordinator }); + const secondCleanup = subscribeToBoardEvents(mutate, () => { + throw new Error("source should be reused"); + }, { debounceMs: 0, coordinator }); + + source.emit({ event: "run.running" }); + + expect(created).toEqual(["/api/v1/attach"]); + expect(keys).toEqual([queryKeys.boards.runs()]); + + firstCleanup(); + expect(source.closed).toBe(false); + secondCleanup(); + expect(source.closed).toBe(true); + coordinator.close(); }); }); + +function createCoordinator(eventSourceFactory: (url: string) => EventSourceLike) { + return createCrossTabSseCoordinator({ + tabId: "board-test", + channelFactory: () => new FakeBroadcastChannel(), + eventSourceFactory, + addVisibilityChangeListener: () => () => {}, + addPagehideListener: () => () => {}, + timing: { + heartbeatMs: 10, + leaderStaleMs: 50, + electionJitterMs: 0, + }, + }); +} + +function createFallbackCoordinator() { + return createCrossTabSseCoordinator({ + channelFactory: () => { + throw new Error("BroadcastChannel unavailable"); + }, + }); +} + +async function waitFor(condition: () => boolean, timeoutMs = 200) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (condition()) return; + await new Promise((resolve) => setTimeout(resolve, 2)); + } + throw new Error("condition did not become true before timeout"); +} diff --git a/apps/fabro-web/app/lib/board-events.ts b/apps/fabro-web/app/lib/board-events.ts index ef30009b1..dc01765ee 100644 --- a/apps/fabro-web/app/lib/board-events.ts +++ b/apps/fabro-web/app/lib/board-events.ts @@ -1,6 +1,10 @@ import { useEffect } from "react"; import { useSWRConfig } from "swr"; +import { + subscribeToCrossTabSse, + type CrossTabSseCoordinator, +} from "./cross-tab-sse"; import { queryKeys } from "./query-keys"; import { createBrowserEventSource, @@ -11,6 +15,11 @@ import { type SharedEventSubscription, } from "./sse"; +interface BoardEventOptions { + debounceMs?: number; + coordinator?: CrossTabSseCoordinator; +} + const BOARD_STATUS_EVENTS = new Set([ "run.submitted", "run.queued", @@ -41,23 +50,36 @@ export function shouldRefreshBoardForEvent(event: string) { export function subscribeToBoardEvents( mutate: MutateFn, eventSourceFactory: (url: string) => EventSourceLike = createBrowserEventSource, - { debounceMs = 500 }: { debounceMs?: number } = {}, + { debounceMs = 500, coordinator }: BoardEventOptions = {}, ): () => void { - return subscribeToSharedEventSource({ - subscriptions, + return subscribeToCrossTabSse({ + coordinator, subscriptionKey: BOARD_SUBSCRIPTION_KEY, - url: queryKeys.system.attach(), mutate, - eventSourceFactory, debounceMs, - resolveInvalidation: (payload) => ({ - keys: payload.event && shouldRefreshBoardForEvent(payload.event) - ? [queryKeys.boards.runs()] - : [], - }), + resyncKeys: () => [queryKeys.boards.runs()], + resolveInvalidation: boardInvalidation, + fallbackSubscribe: () => + subscribeToSharedEventSource({ + subscriptions, + subscriptionKey: BOARD_SUBSCRIPTION_KEY, + url: queryKeys.system.attach(), + mutate, + eventSourceFactory, + debounceMs, + resolveInvalidation: boardInvalidation, + }), }); } +function boardInvalidation(payload: EventPayload) { + return { + keys: payload.event && shouldRefreshBoardForEvent(payload.event) + ? [queryKeys.boards.runs()] + : [], + }; +} + export function useBoardEvents() { const { mutate } = useSWRConfig(); diff --git a/apps/fabro-web/app/lib/cross-tab-sse.test.ts b/apps/fabro-web/app/lib/cross-tab-sse.test.ts new file mode 100644 index 000000000..76fc0b16e --- /dev/null +++ b/apps/fabro-web/app/lib/cross-tab-sse.test.ts @@ -0,0 +1,736 @@ +import { afterEach, describe, expect, test } from "bun:test"; + +import { + CROSS_TAB_SSE_CHANNEL, + createCrossTabSseCoordinator, + subscribeToCrossTabSse, + type BroadcastChannelLike, + type CrossTabSseCoordinator, + type CrossTabSseMessage, +} from "./cross-tab-sse"; +import type { EventPayload, MutateFn } from "./sse"; + +type MessageHandler = ((event: { data: string }) => void) | null; +type TabVisibility = "visible" | "hidden"; + +const TEST_TIMING = { + heartbeatMs: 10, + leaderStaleMs: 35, + electionJitterMs: 5, +}; + +class FakeEventSource { + onmessage: MessageHandler = null; + closed = false; + + constructor( + readonly url: string, + readonly owner: string, + ) {} + + emit(payload: unknown) { + this.onmessage?.({ data: JSON.stringify(payload) }); + } + + close() { + this.closed = true; + } +} + +class FakeBroadcastChannel implements BroadcastChannelLike { + static channels = new Set(); + static muted = false; + static throwOnTypes = new Set(); + + onmessage: ((event: { data: unknown }) => void) | null = null; + closed = false; + + constructor(readonly name: string) { + FakeBroadcastChannel.channels.add(this); + } + + postMessage(message: CrossTabSseMessage) { + if (FakeBroadcastChannel.throwOnTypes.has(message.type)) { + throw new Error(`postMessage failed for ${message.type}`); + } + if (FakeBroadcastChannel.muted) return; + const recipients = [...FakeBroadcastChannel.channels].filter( + (channel) => channel !== this && !channel.closed && channel.name === this.name, + ); + queueMicrotask(() => { + for (const channel of recipients) { + if (channel.closed) continue; + channel.onmessage?.({ data: { ...message } }); + } + }); + } + + static broadcastExternal(message: CrossTabSseMessage) { + queueMicrotask(() => { + for (const channel of FakeBroadcastChannel.channels) { + if (channel.closed) continue; + channel.onmessage?.({ data: { ...message } }); + } + }); + } + + close() { + this.closed = true; + FakeBroadcastChannel.channels.delete(this); + } + + static reset() { + for (const channel of FakeBroadcastChannel.channels) { + channel.closed = true; + } + FakeBroadcastChannel.channels.clear(); + FakeBroadcastChannel.muted = false; + FakeBroadcastChannel.throwOnTypes.clear(); + } +} + +class Harness { + readonly sources: FakeEventSource[] = []; + readonly coordinators = new Map(); + readonly visibility = new Map(); + readonly visibilityHandlers = new Map void>(); + now = 1000; + + createTab(tabId: string, visibility: TabVisibility = "visible") { + this.visibility.set(tabId, visibility); + const coordinator = createCrossTabSseCoordinator({ + tabId, + channelFactory: (name) => new FakeBroadcastChannel(name), + eventSourceFactory: (url) => { + const source = new FakeEventSource(url, tabId); + this.sources.push(source); + return source; + }, + getVisibility: () => this.visibility.get(tabId) ?? "visible", + addVisibilityChangeListener: (handler) => { + this.visibilityHandlers.set(tabId, handler); + return () => this.visibilityHandlers.delete(tabId); + }, + addPagehideListener: () => () => {}, + now: () => this.now, + timing: TEST_TIMING, + }); + this.coordinators.set(tabId, coordinator); + return coordinator; + } + + setVisibility(tabId: string, visibility: TabVisibility) { + this.visibility.set(tabId, visibility); + this.visibilityHandlers.get(tabId)?.(); + } + + openSources() { + return this.sources.filter((source) => !source.closed); + } + + close() { + for (const coordinator of this.coordinators.values()) { + coordinator.close(); + } + } +} + +const harnesses: Harness[] = []; + +afterEach(() => { + for (const harness of harnesses.splice(0)) { + harness.close(); + } + FakeBroadcastChannel.reset(); +}); + +describe("subscribeToCrossTabSse", () => { + test("opens one leader-owned global EventSource and keeps followers passive", async () => { + const harness = newHarness(); + const cleanups = ["a", "b", "c"].map((tabId) => { + const coordinator = harness.createTab(tabId); + return subscribeForRunEvent(coordinator, []); + }); + + await waitFor(() => harness.openSources().length === 1); + + expect(harness.openSources().map((source) => source.url)).toEqual(["/api/v1/attach"]); + expect([...FakeBroadcastChannel.channels].every((channel) => channel.name === CROSS_TAB_SSE_CHANNEL)).toBe(true); + + cleanups.forEach((cleanup) => cleanup()); + }); + + test("leader broadcasts events to all local subscribers", async () => { + const harness = newHarness(); + const keysByTab = new Map(); + + for (const tabId of ["a", "b", "c"]) { + keysByTab.set(tabId, []); + subscribeForRunEvent(harness.createTab(tabId), keysByTab.get(tabId)!); + } + + await waitFor(() => harness.openSources().length === 1); + clearRecordedKeys(keysByTab); + + harness.openSources()[0].emit(runEvent({ id: "evt-1", runId: "run-1", seq: 1 })); + await waitFor(() => [...keysByTab.values()].every((keys) => keys.length === 1)); + + expect(keysByTab.get("a")).toEqual(["event"]); + expect(keysByTab.get("b")).toEqual(["event"]); + expect(keysByTab.get("c")).toEqual(["event"]); + }); + + test("board and run subscriptions coexist on the same global stream", async () => { + const harness = newHarness(); + const coordinator = harness.createTab("a"); + const boardKeys: string[] = []; + const runKeys: string[] = []; + + subscribeForEvent(coordinator, { + subscriptionKey: "board", + keys: boardKeys, + resolveInvalidation: (payload) => ({ + keys: payload.event === "run.running" ? ["board"] : [], + }), + resyncKeys: () => ["board-resync"], + }); + subscribeForEvent(coordinator, { + subscriptionKey: "run:run-1", + keys: runKeys, + resolveInvalidation: (payload) => ({ + keys: payload.event === "run.running" && payload.run_id === "run-1" ? ["run"] : [], + }), + resyncKeys: () => ["run-resync"], + }); + + await waitFor(() => harness.openSources().length === 1); + boardKeys.length = 0; + runKeys.length = 0; + + harness.openSources()[0].emit(runEvent({ id: "evt-coexist", runId: "run-1", seq: 1 })); + + expect(boardKeys).toEqual(["board"]); + expect(runKeys).toEqual(["run"]); + expect(harness.openSources().map((source) => source.url)).toEqual(["/api/v1/attach"]); + }); + + test("dedupes duplicate event ids until TTL or max-size eviction", async () => { + const harness = newHarness(); + const keys: string[] = []; + subscribeForRunEvent(harness.createTab("a"), keys); + + await waitFor(() => harness.openSources().length === 1); + keys.length = 0; + + const source = harness.openSources()[0]; + source.emit(runEvent({ id: "evt-dup", runId: "run-1", seq: 1 })); + source.emit(runEvent({ id: "evt-dup", runId: "run-1", seq: 1 })); + + expect(keys).toEqual(["event"]); + + harness.now += 5 * 60 * 1000 + 1; + source.emit(runEvent({ id: "evt-dup", runId: "run-1", seq: 1 })); + expect(keys).toEqual(["event", "event"]); + + keys.length = 0; + for (let i = 0; i < 1001; i += 1) { + source.emit(runEvent({ id: `evt-${i}`, runId: "run-1", seq: i + 2 })); + } + source.emit(runEvent({ id: "evt-0", runId: "run-1", seq: 2 })); + expect(keys).toHaveLength(1002); + }); + + test("visible followers take over from a fresh hidden leader and resync", async () => { + const harness = newHarness(); + const hiddenKeys: string[] = []; + const visibleKeys: string[] = []; + + subscribeForRunEvent(harness.createTab("z", "hidden"), hiddenKeys); + await waitFor(() => harness.openSources().length === 1); + const hiddenSource = harness.openSources()[0]; + + subscribeForRunEvent(harness.createTab("a", "visible"), visibleKeys); + + await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "a"); + + expect(hiddenSource.closed).toBe(true); + expect(visibleKeys).toContain("resync"); + }); + + test("visible candidates racing for the same hidden leader resolve lexically", async () => { + const harness = newHarness(); + + subscribeForRunEvent(harness.createTab("z", "hidden"), []); + await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "z"); + + subscribeForRunEvent(harness.createTab("b", "visible"), []); + subscribeForRunEvent(harness.createTab("a", "visible"), []); + + await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner !== "z"); + expect(harness.openSources().map((source) => source.owner)).toEqual(["a"]); + }); + + test("a lower lexical follower does not preempt a fresh visible leader", async () => { + const harness = newHarness(); + + subscribeForRunEvent(harness.createTab("z", "visible"), []); + await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "z"); + + subscribeForRunEvent(harness.createTab("a", "visible"), []); + await sleep(TEST_TIMING.electionJitterMs * 4); + + expect(harness.openSources().map((source) => source.owner)).toEqual(["z"]); + }); + + test("stale leader detection opens a new leader source and resyncs followers", async () => { + const harness = newHarness(); + const followerKeys: string[] = []; + + subscribeForRunEvent(harness.createTab("a"), []); + subscribeForRunEvent(harness.createTab("b"), followerKeys); + await waitFor(() => harness.openSources().length === 1); + + const staleLeader = harness.openSources()[0]; + harness.coordinators.get(staleLeader.owner)?.close(); + harness.now += TEST_TIMING.leaderStaleMs + TEST_TIMING.heartbeatMs + 1; + + await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner !== staleLeader.owner); + + expect(followerKeys).toContain("resync"); + }); + + test("simultaneous stale leader elections resolve to the lexical winner", async () => { + const harness = newHarness(); + + subscribeForRunEvent(harness.createTab("z"), []); + await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "z"); + subscribeForRunEvent(harness.createTab("b"), []); + subscribeForRunEvent(harness.createTab("a"), []); + await sleep(TEST_TIMING.heartbeatMs * 2); + + harness.coordinators.get("z")?.close(); + harness.now += TEST_TIMING.leaderStaleMs + TEST_TIMING.heartbeatMs + 1; + + await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "a"); + }); + + test("hidden leader ignores candidates for old observed leadership", async () => { + const harness = newHarness(); + + subscribeForRunEvent(harness.createTab("z", "hidden"), []); + await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "z"); + const hiddenSource = harness.openSources()[0]; + + FakeBroadcastChannel.broadcastExternal({ + type: "candidate", + version: 1, + tabId: "ghost", + sentAt: harness.now, + candidateId: "ghost", + candidateGeneration: 1, + visibility: "visible", + observedLeaderId: "z", + observedGeneration: 0, + reason: "hidden-leader", + }); + await sleep(TEST_TIMING.electionJitterMs * 2); + + expect(hiddenSource.closed).toBe(false); + expect(harness.openSources().map((source) => source.owner)).toEqual(["z"]); + }); + + test("prunes candidate records from older generations", async () => { + const harness = newHarness(); + subscribeForRunEvent(harness.createTab("z", "hidden"), []); + await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "z"); + + const coordinator = harness.createTab("a", "visible"); + subscribeForRunEvent(coordinator, []); + await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "a"); + + FakeBroadcastChannel.broadcastExternal({ + type: "candidate", + version: 1, + tabId: "old-candidate", + sentAt: harness.now, + candidateId: "old-candidate", + candidateGeneration: 1, + visibility: "visible", + observedLeaderId: "previous-leader", + observedGeneration: 0, + reason: "stale-leader", + }); + await sleep(TEST_TIMING.electionJitterMs * 2); + + expect(candidateGenerations(coordinator)).not.toContain(1); + }); + + test("same-generation split brain converges to the higher-priority visible leader", async () => { + const harness = newHarness(); + FakeBroadcastChannel.muted = true; + + subscribeForRunEvent(harness.createTab("b"), []); + subscribeForRunEvent(harness.createTab("a"), []); + await waitFor(() => harness.openSources().length === 2); + + FakeBroadcastChannel.muted = false; + await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "a"); + }); + + test("old leader events are ignored after takeover", async () => { + const harness = newHarness(); + const keys: string[] = []; + + subscribeForRunEvent(harness.createTab("z", "hidden"), []); + await waitFor(() => harness.openSources().length === 1); + const oldSource = harness.openSources()[0]; + + subscribeForRunEvent(harness.createTab("a", "visible"), keys); + await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "a"); + keys.length = 0; + + oldSource.emit(runEvent({ id: "evt-old", runId: "run-1", seq: 1 })); + expect(keys).toEqual([]); + }); + + test("old leader heartbeats are ignored after takeover", async () => { + const harness = newHarness(); + + subscribeForRunEvent(harness.createTab("z", "hidden"), []); + await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "z"); + + subscribeForRunEvent(harness.createTab("a", "visible"), []); + await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "a"); + + FakeBroadcastChannel.broadcastExternal({ + type: "heartbeat", + version: 1, + tabId: "z", + sentAt: harness.now, + leaderId: "z", + generation: 1, + visibility: "hidden", + }); + await sleep(TEST_TIMING.heartbeatMs * 2); + + expect(harness.openSources().map((source) => source.owner)).toEqual(["a"]); + }); + + test("last unsubscribe closes the leader source and releases leadership", async () => { + const harness = newHarness(); + const cleanup = subscribeForRunEvent(harness.createTab("a"), []); + + await waitFor(() => harness.openSources().length === 1); + const source = harness.openSources()[0]; + + cleanup(); + + expect(source.closed).toBe(true); + expect(harness.openSources()).toEqual([]); + }); + + test("missing BroadcastChannel uses subscriber fallback", () => { + const coordinator = createCrossTabSseCoordinator({ + channelFactory: () => { + throw new Error("no channel"); + }, + }); + let fallbackStarted = 0; + let fallbackStopped = 0; + + const cleanup = subscribeToCrossTabSse({ + coordinator, + subscriptionKey: "fallback", + mutate: (() => Promise.resolve()) as MutateFn, + resolveInvalidation: () => ({ keys: [] }), + resyncKeys: () => [], + fallbackSubscribe: () => { + fallbackStarted += 1; + return () => { + fallbackStopped += 1; + }; + }, + debounceMs: 0, + }); + + cleanup(); + + expect(fallbackStarted).toBe(1); + expect(fallbackStopped).toBe(1); + }); + + test("postMessage failure after initialization degrades to fallback without coordinated resync", async () => { + const harness = newHarness(); + const coordinator = harness.createTab("a"); + const keys: string[] = []; + let fallbackStarted = 0; + let fallbackStopped = 0; + + FakeBroadcastChannel.throwOnTypes.add("leader-changed"); + const cleanup = subscribeToCrossTabSse({ + coordinator, + subscriptionKey: "throwing-channel", + mutate: ((key: string) => { + keys.push(key); + return Promise.resolve(); + }) as MutateFn, + resolveInvalidation: () => ({ keys: ["event"] }), + resyncKeys: () => ["resync"], + fallbackSubscribe: () => { + fallbackStarted += 1; + return () => { + fallbackStopped += 1; + }; + }, + debounceMs: 0, + }); + + await waitFor(() => fallbackStarted === 1); + + expect(harness.openSources()).toEqual([]); + expect(keys).toEqual([]); + + cleanup(); + expect(fallbackStopped).toBe(1); + }); + + test("close resets coordination availability after an initial channel failure", async () => { + let channelUnavailable = true; + const sources: FakeEventSource[] = []; + const coordinator = createCrossTabSseCoordinator({ + tabId: "recovering", + channelFactory: (name) => { + if (channelUnavailable) throw new Error("channel unavailable"); + return new FakeBroadcastChannel(name); + }, + eventSourceFactory: (url) => { + const source = new FakeEventSource(url, "recovering"); + sources.push(source); + return source; + }, + addVisibilityChangeListener: () => () => {}, + addPagehideListener: () => () => {}, + timing: TEST_TIMING, + }); + let firstFallbackStarted = 0; + let secondFallbackStarted = 0; + + const firstCleanup = subscribeWithFallback(coordinator, { + fallbackSubscribe: () => { + firstFallbackStarted += 1; + return () => {}; + }, + }); + firstCleanup(); + coordinator.close(); + + channelUnavailable = false; + const secondCleanup = subscribeWithFallback(coordinator, { + fallbackSubscribe: () => { + secondFallbackStarted += 1; + return () => {}; + }, + }); + + await waitFor(() => sources.some((source) => !source.closed)); + + expect(firstFallbackStarted).toBe(1); + expect(secondFallbackStarted).toBe(0); + expect(sources.filter((source) => !source.closed).map((source) => source.url)).toEqual(["/api/v1/attach"]); + + secondCleanup(); + coordinator.close(); + }); + + test("last unsubscribe retries coordination after an initial channel failure", async () => { + let channelUnavailable = true; + const sources: FakeEventSource[] = []; + const coordinator = createCrossTabSseCoordinator({ + tabId: "retry-after-unsubscribe", + channelFactory: (name) => { + if (channelUnavailable) throw new Error("channel unavailable"); + return new FakeBroadcastChannel(name); + }, + eventSourceFactory: (url) => { + const source = new FakeEventSource(url, "retry-after-unsubscribe"); + sources.push(source); + return source; + }, + addVisibilityChangeListener: () => () => {}, + addPagehideListener: () => () => {}, + timing: TEST_TIMING, + }); + let fallbackStarted = 0; + + const firstCleanup = subscribeWithFallback(coordinator, { + fallbackSubscribe: () => { + fallbackStarted += 1; + return () => {}; + }, + }); + firstCleanup(); + + channelUnavailable = false; + const secondCleanup = subscribeWithFallback(coordinator, { + fallbackSubscribe: () => { + fallbackStarted += 1; + return () => {}; + }, + }); + + await waitFor(() => sources.some((source) => !source.closed)); + + expect(fallbackStarted).toBe(1); + expect(sources.filter((source) => !source.closed).map((source) => source.url)).toEqual(["/api/v1/attach"]); + + secondCleanup(); + coordinator.close(); + }); + + test("close stops fallback subscriptions added after degradation", async () => { + const harness = newHarness(); + const coordinator = harness.createTab("a"); + let fallbackStarted = 0; + let fallbackStopped = 0; + + FakeBroadcastChannel.throwOnTypes.add("leader-changed"); + subscribeWithFallback(coordinator, { + subscriptionKey: "before-degrade", + fallbackSubscribe: () => { + fallbackStarted += 1; + return () => { + fallbackStopped += 1; + }; + }, + }); + await waitFor(() => fallbackStarted === 1); + + FakeBroadcastChannel.throwOnTypes.clear(); + subscribeWithFallback(coordinator, { + subscriptionKey: "after-degrade", + fallbackSubscribe: () => { + fallbackStarted += 1; + return () => { + fallbackStopped += 1; + }; + }, + }); + expect(fallbackStarted).toBe(2); + + coordinator.close(); + + expect(fallbackStopped).toBe(2); + }); +}); + +function newHarness() { + const harness = new Harness(); + harnesses.push(harness); + return harness; +} + +function subscribeForRunEvent(coordinator: CrossTabSseCoordinator, keys: string[]) { + return subscribeForEvent(coordinator, { + subscriptionKey: "run-feed", + keys, + resolveInvalidation: (payload) => ({ + keys: payload.event === "run.running" ? ["event"] : [], + }), + resyncKeys: () => ["resync"], + }); +} + +function subscribeForEvent( + coordinator: CrossTabSseCoordinator, + { + subscriptionKey, + keys, + resolveInvalidation, + resyncKeys, + }: { + subscriptionKey: string; + keys: string[]; + resolveInvalidation: (payload: EventPayload) => { keys: string[] }; + resyncKeys: () => string[]; + }, +) { + return subscribeToCrossTabSse({ + coordinator, + subscriptionKey, + mutate: ((key: string) => { + keys.push(key); + return Promise.resolve(); + }) as MutateFn, + resolveInvalidation, + resyncKeys, + fallbackSubscribe: () => { + throw new Error("fallback should not be used"); + }, + debounceMs: 0, + }); +} + +function subscribeWithFallback( + coordinator: CrossTabSseCoordinator, + { + subscriptionKey = "fallback-test", + fallbackSubscribe, + }: { + subscriptionKey?: string; + fallbackSubscribe: () => () => void; + }, +) { + return subscribeToCrossTabSse({ + coordinator, + subscriptionKey, + mutate: (() => Promise.resolve()) as MutateFn, + resolveInvalidation: () => ({ keys: [] }), + resyncKeys: () => [], + fallbackSubscribe, + debounceMs: 0, + }); +} + +function candidateGenerations(coordinator: CrossTabSseCoordinator): number[] { + const inspectable = coordinator as unknown as { + candidates: Map; + }; + return [...inspectable.candidates.values()].map((candidate) => candidate.candidateGeneration); +} + +function runEvent({ + id, + runId, + seq, +}: { + id: string; + runId: string; + seq: number; +}) { + return { + id, + seq, + run_id: runId, + event: "run.running", + ts: "2026-05-04T12:00:00.000Z", + }; +} + +async function waitFor(condition: () => boolean, timeoutMs = 500) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (condition()) return; + await sleep(2); + } + throw new Error("condition did not become true before timeout"); +} + +function sleep(ms: number) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function clearRecordedKeys(keysByTab: Map) { + for (const keys of keysByTab.values()) { + keys.length = 0; + } +} diff --git a/apps/fabro-web/app/lib/cross-tab-sse.ts b/apps/fabro-web/app/lib/cross-tab-sse.ts new file mode 100644 index 000000000..6975e6948 --- /dev/null +++ b/apps/fabro-web/app/lib/cross-tab-sse.ts @@ -0,0 +1,1175 @@ +import { queryKeys } from "./query-keys"; +import { + createBrowserEventSource, + type EventInvalidation, + type EventPayload, + type EventSourceLike, + type MutateFn, +} from "./sse"; +import { getNumber, getString, isRecord, type UnknownRecord } from "./unknown"; + +export const CROSS_TAB_SSE_CHANNEL = "fabro:sse:v1"; +export const HEARTBEAT_MS = 1000; +export const LEADER_STALE_MS = 4000; +export const ELECTION_JITTER_MS = 150; + +const MESSAGE_VERSION = 1 as const; +const EVENT_DEDUPE_TTL_MS = 5 * 60 * 1000; +const EVENT_DEDUPE_MAX = 1000; + +type TabVisibility = "visible" | "hidden"; +type CandidateReason = "hidden-leader" | "stale-leader" | "release" | "no-leader"; + +interface BaseMessage { + type: string; + version: typeof MESSAGE_VERSION; + tabId: string; + sentAt: number; +} + +interface HelloMessage extends BaseMessage { + type: "hello"; +} + +interface HeartbeatMessage extends BaseMessage { + type: "heartbeat"; + leaderId: string; + generation: number; + visibility: TabVisibility; +} + +interface CandidateMessage extends BaseMessage { + type: "candidate"; + candidateId: string; + candidateGeneration: number; + visibility: TabVisibility; + observedLeaderId: string | null; + observedGeneration: number; + reason: CandidateReason; +} + +interface LeaderChangedMessage extends BaseMessage { + type: "leader-changed"; + leaderId: string; + generation: number; + visibility: TabVisibility; +} + +interface ReleaseMessage extends BaseMessage { + type: "release"; + leaderId: string; + generation: number; +} + +interface ResyncMessage extends BaseMessage { + type: "resync"; + leaderId: string | null; + generation: number; + reason: CandidateReason; +} + +interface EventMessage extends BaseMessage { + type: "event"; + leaderId: string; + generation: number; + payload: EventPayload; +} + +export type CrossTabSseMessage = + | HelloMessage + | HeartbeatMessage + | CandidateMessage + | LeaderChangedMessage + | ReleaseMessage + | ResyncMessage + | EventMessage; + +export interface BroadcastChannelLike { + onmessage: ((event: { data: unknown }) => void) | null; + postMessage(message: CrossTabSseMessage): void; + close(): void; +} + +interface TimingOptions { + heartbeatMs: number; + leaderStaleMs: number; + electionJitterMs: number; +} + +export interface CrossTabSseCoordinatorOptions { + tabId?: string; + channelFactory?: (name: string) => BroadcastChannelLike; + eventSourceFactory?: (url: string) => EventSourceLike; + getVisibility?: () => TabVisibility; + addVisibilityChangeListener?: (handler: () => void) => () => void; + addPagehideListener?: (handler: () => void) => () => void; + now?: () => number; + timing?: Partial; +} + +interface SubscribeOptions { + subscriptionKey: string; + mutate: MutateFn; + resolveInvalidation: (payload: TPayload) => EventInvalidation; + resyncKeys: () => string[]; + fallbackSubscribe: () => () => void; + debounceMs?: number; +} + +export interface SubscribeToCrossTabSseOptions + extends SubscribeOptions { + coordinator?: CrossTabSseCoordinator; +} + +interface FallbackEntry { + count: number; + subscribe: () => () => void; + cleanup?: () => void; +} + +interface LocalSubscription { + refcount: number; + mutators: Map; + fallbacks: Map; + pendingKeys: Set; + debounceTimer: ReturnType | null; + debounceMs: number; + resolveInvalidation: (payload: EventPayload) => EventInvalidation; + resyncKeys: () => string[]; +} + +interface LeaderState { + leaderId: string; + generation: number; + visibility: TabVisibility; + lastSeen: number; +} + +class RecentEventCache { + private readonly seen = new Map(); + + constructor( + private readonly maxSize: number, + private readonly ttlMs: number, + ) {} + + remember(key: string | undefined, now: number): boolean { + if (!key) return true; + this.evictExpired(now); + if (this.seen.has(key)) return false; + + this.seen.set(key, now); + while (this.seen.size > this.maxSize) { + const oldest = this.seen.keys().next().value; + if (oldest === undefined) break; + this.seen.delete(oldest); + } + return true; + } + + private evictExpired(now: number) { + for (const [key, seenAt] of this.seen) { + if (now - seenAt <= this.ttlMs) return; + this.seen.delete(key); + } + } +} + +export class CrossTabSseCoordinator { + readonly tabId: string; + + private readonly channelFactory: (name: string) => BroadcastChannelLike; + private readonly getVisibility: () => TabVisibility; + private readonly addVisibilityChangeListener: (handler: () => void) => () => void; + private readonly addPagehideListener: (handler: () => void) => () => void; + private readonly now: () => number; + private readonly timing: TimingOptions; + private readonly recentEvents = new RecentEventCache(EVENT_DEDUPE_MAX, EVENT_DEDUPE_TTL_MS); + private readonly subscriptions = new Map(); + private readonly candidates = new Map(); + + private sourceFactory: (url: string) => EventSourceLike; + private channel: BroadcastChannelLike | null = null; + private source: EventSourceLike | null = null; + private initialized = false; + private coordinationUnavailable = false; + private fallbackMode = false; + private degradingToFallback = false; + private isLeader = false; + private leader: LeaderState | null = null; + private generation = 0; + private ownCandidate: CandidateMessage | null = null; + private candidateTimer: ReturnType | null = null; + private noLeaderTimer: ReturnType | null = null; + private heartbeatTimer: ReturnType | null = null; + private leaderCheckTimer: ReturnType | null = null; + private removeVisibilityListener: (() => void) | null = null; + private removePagehideListener: (() => void) | null = null; + + constructor(options: CrossTabSseCoordinatorOptions = {}) { + this.tabId = options.tabId ?? createTabId(); + this.channelFactory = options.channelFactory ?? createBrowserBroadcastChannel; + this.sourceFactory = options.eventSourceFactory ?? createBrowserEventSource; + this.getVisibility = options.getVisibility ?? getBrowserVisibility; + this.addVisibilityChangeListener = + options.addVisibilityChangeListener ?? addBrowserVisibilityChangeListener; + this.addPagehideListener = options.addPagehideListener ?? addBrowserPagehideListener; + this.now = options.now ?? Date.now; + this.timing = { + heartbeatMs: options.timing?.heartbeatMs ?? HEARTBEAT_MS, + leaderStaleMs: options.timing?.leaderStaleMs ?? LEADER_STALE_MS, + electionJitterMs: options.timing?.electionJitterMs ?? ELECTION_JITTER_MS, + }; + } + + subscribe(options: SubscribeOptions): () => void { + const subscription = this.addLocalSubscription(options); + + if (this.coordinationUnavailable) { + this.fallbackMode = true; + this.startFallbacksFor(subscription); + } else if (!this.initialized && !this.initialize()) { + this.coordinationUnavailable = true; + this.fallbackMode = true; + this.startFallbacksFor(subscription); + } else if (this.fallbackMode) { + this.startFallbacksFor(subscription); + } else { + this.ensureLeadershipProgress(); + } + + let active = true; + return () => { + if (!active) return; + active = false; + this.removeLocalSubscription(options.subscriptionKey, options.mutate); + }; + } + + close() { + this.releaseLeadership({ broadcast: false, resync: false }); + this.clearCandidate(); + this.clearNoLeaderTimer(); + this.shutdownTimersAndChannel(); + this.closeFallbacks(); + this.clearSubscriptionTimers(); + this.subscriptions.clear(); + this.candidates.clear(); + this.resetIdleState(); + } + + private initialize(): boolean { + try { + this.channel = this.channelFactory(CROSS_TAB_SSE_CHANNEL); + } catch { + this.channel = null; + return false; + } + + this.channel.onmessage = (event) => this.handleMessage(event.data); + this.initialized = true; + this.removeVisibilityListener = this.addVisibilityChangeListener(() => { + this.handleVisibilityChange(); + }); + this.removePagehideListener = this.addPagehideListener(() => { + this.handlePagehide(); + }); + this.leaderCheckTimer = setInterval(() => { + this.checkLeaderFreshness(); + }, this.timing.heartbeatMs); + + return this.post({ type: "hello", version: MESSAGE_VERSION, tabId: this.tabId, sentAt: this.now() }); + } + + private addLocalSubscription( + options: SubscribeOptions, + ): LocalSubscription { + let subscription = this.subscriptions.get(options.subscriptionKey); + if (!subscription) { + subscription = { + refcount: 0, + mutators: new Map(), + fallbacks: new Map(), + pendingKeys: new Set(), + debounceTimer: null, + debounceMs: options.debounceMs ?? 300, + resolveInvalidation: options.resolveInvalidation as (payload: EventPayload) => EventInvalidation, + resyncKeys: options.resyncKeys, + }; + this.subscriptions.set(options.subscriptionKey, subscription); + } else { + subscription.resolveInvalidation = + options.resolveInvalidation as (payload: EventPayload) => EventInvalidation; + subscription.resyncKeys = options.resyncKeys; + subscription.debounceMs = options.debounceMs ?? subscription.debounceMs; + } + + subscription.refcount += 1; + subscription.mutators.set( + options.mutate, + (subscription.mutators.get(options.mutate) ?? 0) + 1, + ); + + const fallback = subscription.fallbacks.get(options.mutate); + if (fallback) { + fallback.count += 1; + fallback.subscribe = options.fallbackSubscribe; + } else { + subscription.fallbacks.set(options.mutate, { + count: 1, + subscribe: options.fallbackSubscribe, + }); + } + + return subscription; + } + + private removeLocalSubscription(subscriptionKey: string, mutate: MutateFn) { + const subscription = this.subscriptions.get(subscriptionKey); + if (!subscription) return; + + const mutateCount = subscription.mutators.get(mutate) ?? 0; + if (mutateCount <= 1) { + subscription.mutators.delete(mutate); + } else { + subscription.mutators.set(mutate, mutateCount - 1); + } + + const fallback = subscription.fallbacks.get(mutate); + if (fallback) { + fallback.count -= 1; + if (fallback.count <= 0) { + fallback.cleanup?.(); + subscription.fallbacks.delete(mutate); + } + } + + subscription.refcount -= 1; + if (subscription.refcount <= 0) { + this.clearSubscriptionTimer(subscription); + this.subscriptions.delete(subscriptionKey); + } + + if (this.subscriptions.size === 0) { + this.releaseLeadership({ broadcast: true, resync: false }); + this.clearCandidate(); + this.clearNoLeaderTimer(); + this.shutdownTimersAndChannel(); + this.resetIdleState(); + } + } + + private handleMessage(data: unknown) { + const message = parseMessage(data); + if (!message || message.tabId === this.tabId) return; + + switch (message.type) { + case "hello": + if (this.isLeader) this.sendHeartbeat(); + break; + case "heartbeat": + this.handleLeaderAnnouncement(message, { resyncOnChange: true }); + break; + case "candidate": + this.handleCandidate(message); + break; + case "leader-changed": + this.handleLeaderAnnouncement(message, { resyncOnChange: true }); + break; + case "release": + this.handleRelease(message); + break; + case "resync": + this.handleResync(message); + break; + case "event": + this.handleBroadcastEvent(message); + break; + } + } + + private handleLeaderAnnouncement( + message: HeartbeatMessage | LeaderChangedMessage, + { resyncOnChange }: { resyncOnChange: boolean }, + ) { + const incoming: LeaderState = { + leaderId: message.leaderId, + generation: message.generation, + visibility: message.visibility, + lastSeen: this.now(), + }; + + if (this.isLeader && incoming.leaderId !== this.tabId) { + const own: LeaderState = { + leaderId: this.tabId, + generation: this.generation, + visibility: this.currentVisibility(), + lastSeen: this.now(), + }; + if ( + incoming.generation > own.generation || + (incoming.generation === own.generation && leaderHasHigherPriority(incoming, own)) + ) { + this.releaseLeadership({ broadcast: false, resync: true }); + } else { + return; + } + } + + const previous = this.leader; + if (this.ownCandidate && incoming.generation === this.ownCandidate.candidateGeneration) { + const sawIncomingCandidate = this.candidates.has( + `${incoming.generation}:${incoming.leaderId}`, + ); + const shouldAcceptFreshVisibleLeader = + this.ownCandidate.reason === "no-leader" && + incoming.visibility === "visible" && + !sawIncomingCandidate; + + if ( + !shouldAcceptFreshVisibleLeader && + !leaderHasHigherPriority(incoming, leaderStateForCandidate(this.ownCandidate, this.now())) + ) { + return; + } + } + + if (!this.shouldAcceptLeader(incoming)) return; + + this.leader = incoming; + this.clearNoLeaderTimer(); + this.generation = Math.max(this.generation, incoming.generation); + this.pruneStaleCandidates(); + if (this.ownCandidate && incoming.generation >= this.ownCandidate.candidateGeneration) { + this.clearCandidate(); + } + + const changed = + !previous || + previous.leaderId !== incoming.leaderId || + previous.generation !== incoming.generation; + + if (changed && resyncOnChange) { + this.resyncAll(); + } + + if (incoming.visibility === "hidden" && this.currentVisibility() === "visible") { + this.enterCandidacy("hidden-leader", incoming); + } + } + + private handleCandidate(message: CandidateMessage) { + this.candidates.set(candidateKey(message), message); + this.pruneStaleCandidates(); + + if ( + this.isLeader && + message.observedLeaderId === this.tabId && + message.observedGeneration >= this.generation + ) { + this.releaseLeadership({ broadcast: true, resync: true }); + } + + if ( + this.ownCandidate && + message.candidateGeneration === this.ownCandidate.candidateGeneration && + candidateHasHigherPriority(message, this.ownCandidate) + ) { + this.clearCandidate(); + } + } + + private handleRelease(message: ReleaseMessage) { + const current = this.leader; + if ( + current && + message.leaderId === current.leaderId && + message.generation >= current.generation + ) { + this.leader = null; + this.generation = Math.max(this.generation, message.generation); + this.pruneStaleCandidates(); + this.enterCandidacy("release", { + leaderId: message.leaderId, + generation: message.generation, + visibility: "hidden", + lastSeen: this.now(), + }); + } + } + + private handleResync(message: ResyncMessage) { + if (this.leader && message.generation < this.leader.generation) return; + this.resyncAll(); + } + + private handleBroadcastEvent(message: EventMessage) { + if (!this.isCurrentLeader(message.leaderId, message.generation)) return; + if (!this.recentEvents.remember(eventDedupeKey(message.payload), this.now())) return; + this.dispatchPayload(message.payload); + } + + private ensureLeadershipProgress() { + if (this.subscriptions.size === 0 || this.isLeader || this.ownCandidate) return; + + if (!this.leader) { + this.scheduleNoLeaderCandidacy(); + return; + } + + if (this.leader.visibility === "hidden" && this.currentVisibility() === "visible") { + this.enterCandidacy("hidden-leader", this.leader); + } + } + + private checkLeaderFreshness() { + if (this.subscriptions.size === 0 || this.fallbackMode) return; + if (this.isLeader) return; + + const current = this.leader; + if (!current) { + this.scheduleNoLeaderCandidacy(); + return; + } + + if (!this.leaderIsFresh(current)) { + this.leader = null; + this.generation = Math.max(this.generation, current.generation); + this.pruneStaleCandidates(); + this.resyncAll(); + this.enterCandidacy("stale-leader", current); + return; + } + + if (current.visibility === "hidden" && this.currentVisibility() === "visible") { + this.enterCandidacy("hidden-leader", current); + } + } + + private enterCandidacy(reason: CandidateReason, observedLeader: LeaderState | null = this.leader) { + if (this.subscriptions.size === 0 || this.fallbackMode) return; + + if ( + reason === "no-leader" && + this.leader && + this.leader.visibility === "visible" && + this.leaderIsFresh(this.leader) + ) { + return; + } + + const observedGeneration = observedLeader?.generation ?? this.generation; + const candidateGeneration = observedGeneration + 1; + if ( + this.ownCandidate && + this.ownCandidate.candidateGeneration >= candidateGeneration + ) { + return; + } + + this.clearCandidate(); + this.clearNoLeaderTimer(); + const candidate: CandidateMessage = { + type: "candidate", + version: MESSAGE_VERSION, + tabId: this.tabId, + sentAt: this.now(), + candidateId: this.tabId, + candidateGeneration, + visibility: this.currentVisibility(), + observedLeaderId: observedLeader?.leaderId ?? null, + observedGeneration, + reason, + }; + + this.ownCandidate = candidate; + this.candidates.set(candidateKey(candidate), candidate); + if (!this.post(candidate)) return; + this.candidateTimer = setTimeout(() => { + this.completeCandidacy(candidate); + }, this.timing.electionJitterMs); + } + + private completeCandidacy(candidate: CandidateMessage) { + if (this.ownCandidate !== candidate || this.fallbackMode) return; + + for (const other of this.candidates.values()) { + if ( + other.candidateGeneration === candidate.candidateGeneration && + candidateHasHigherPriority(other, candidate) + ) { + this.clearCandidate(); + return; + } + } + + if (this.leader && this.leaderIsFresh(this.leader)) { + if (this.leader.generation > candidate.candidateGeneration) { + this.clearCandidate(); + return; + } + if ( + this.leader.generation === candidate.candidateGeneration && + leaderHasHigherPriority(this.leader, leaderStateForCandidate(candidate, this.now())) + ) { + this.clearCandidate(); + return; + } + } + + this.becomeLeader(candidate.candidateGeneration); + } + + private becomeLeader(generation: number) { + this.clearCandidate(); + this.closeSource(); + this.isLeader = true; + this.generation = generation; + this.pruneStaleCandidates(); + this.leader = { + leaderId: this.tabId, + generation, + visibility: this.currentVisibility(), + lastSeen: this.now(), + }; + + const source = this.sourceFactory(queryKeys.system.attach()); + this.source = source; + source.onmessage = (message) => { + this.handleLeaderEventSourceMessage(message.data); + }; + + const announced = this.post({ + type: "leader-changed", + version: MESSAGE_VERSION, + tabId: this.tabId, + sentAt: this.now(), + leaderId: this.tabId, + generation, + visibility: this.currentVisibility(), + }); + if (!announced) return; + this.startHeartbeat(); + this.resyncAll(); + } + + private handleLeaderEventSourceMessage(data: string) { + if (!this.isLeader) return; + + let payload: EventPayload; + try { + payload = JSON.parse(data) as EventPayload; + } catch { + return; + } + + if (!this.recentEvents.remember(eventDedupeKey(payload), this.now())) return; + this.dispatchPayload(payload); + this.post({ + type: "event", + version: MESSAGE_VERSION, + tabId: this.tabId, + sentAt: this.now(), + leaderId: this.tabId, + generation: this.generation, + payload, + }); + } + + private dispatchPayload(payload: EventPayload) { + for (const subscription of this.subscriptions.values()) { + const invalidation = subscription.resolveInvalidation(payload); + this.queueInvalidations(subscription, invalidation.keys, { + immediate: invalidation.immediate, + }); + } + } + + private queueInvalidations( + subscription: LocalSubscription, + keys: string[], + { immediate = false }: { immediate?: boolean } = {}, + ) { + if (keys.length === 0) return; + for (const key of keys) { + subscription.pendingKeys.add(key); + } + + if (immediate || subscription.debounceMs <= 0) { + this.flushInvalidations(subscription); + return; + } + + if (subscription.debounceTimer) { + clearTimeout(subscription.debounceTimer); + } + subscription.debounceTimer = setTimeout(() => { + subscription.debounceTimer = null; + this.flushInvalidations(subscription); + }, subscription.debounceMs); + } + + private flushInvalidations(subscription: LocalSubscription) { + if (subscription.pendingKeys.size === 0) return; + const keys = [...subscription.pendingKeys]; + subscription.pendingKeys.clear(); + + for (const mutator of subscription.mutators.keys()) { + for (const key of keys) { + void mutator(key); + } + } + } + + private resyncAll() { + for (const subscription of this.subscriptions.values()) { + this.queueInvalidations(subscription, subscription.resyncKeys(), { immediate: true }); + } + } + + private handleVisibilityChange() { + if (this.isLeader) { + this.sendHeartbeat(); + } + + if (this.currentVisibility() === "visible") { + this.resyncAll(); + if (this.leader?.visibility === "hidden") { + this.enterCandidacy("hidden-leader", this.leader); + } + } + } + + private handlePagehide() { + this.releaseLeadership({ broadcast: true, resync: false }); + this.clearCandidate(); + this.clearNoLeaderTimer(); + } + + private startHeartbeat() { + if (this.heartbeatTimer) { + clearInterval(this.heartbeatTimer); + } + if (!this.sendHeartbeat()) return; + this.heartbeatTimer = setInterval(() => { + this.sendHeartbeat(); + }, this.timing.heartbeatMs); + } + + private sendHeartbeat(): boolean { + if (!this.isLeader) return false; + const visibility = this.currentVisibility(); + this.leader = { + leaderId: this.tabId, + generation: this.generation, + visibility, + lastSeen: this.now(), + }; + return this.post({ + type: "heartbeat", + version: MESSAGE_VERSION, + tabId: this.tabId, + sentAt: this.now(), + leaderId: this.tabId, + generation: this.generation, + visibility, + }); + } + + private releaseLeadership({ + broadcast, + resync, + }: { + broadcast: boolean; + resync: boolean; + }) { + if (!this.isLeader && !this.source) return; + + const generation = this.generation; + this.closeSource(); + this.isLeader = false; + if (this.heartbeatTimer) { + clearInterval(this.heartbeatTimer); + this.heartbeatTimer = null; + } + this.leader = null; + + if (broadcast) { + const released = this.post({ + type: "release", + version: MESSAGE_VERSION, + tabId: this.tabId, + sentAt: this.now(), + leaderId: this.tabId, + generation, + }); + if (!released) return; + this.post({ + type: "resync", + version: MESSAGE_VERSION, + tabId: this.tabId, + sentAt: this.now(), + leaderId: this.tabId, + generation, + reason: "release", + }); + } + if (resync) this.resyncAll(); + } + + private closeSource() { + if (!this.source) return; + this.source.close(); + this.source = null; + } + + private clearCandidate() { + if (this.candidateTimer) { + clearTimeout(this.candidateTimer); + this.candidateTimer = null; + } + this.ownCandidate = null; + } + + private pruneStaleCandidates() { + for (const [key, candidate] of this.candidates) { + if (candidate.candidateGeneration < this.generation) { + this.candidates.delete(key); + } + } + } + + private scheduleNoLeaderCandidacy() { + if ( + this.noLeaderTimer || + this.ownCandidate || + this.isLeader || + this.leader || + this.subscriptions.size === 0 || + this.fallbackMode + ) { + return; + } + + this.noLeaderTimer = setTimeout(() => { + this.noLeaderTimer = null; + if (!this.leader && !this.isLeader) { + this.enterCandidacy("no-leader"); + } + }, this.timing.electionJitterMs); + } + + private clearNoLeaderTimer() { + if (!this.noLeaderTimer) return; + clearTimeout(this.noLeaderTimer); + this.noLeaderTimer = null; + } + + private shutdownTimersAndChannel() { + this.clearNoLeaderTimer(); + if (this.heartbeatTimer) { + clearInterval(this.heartbeatTimer); + this.heartbeatTimer = null; + } + if (this.leaderCheckTimer) { + clearInterval(this.leaderCheckTimer); + this.leaderCheckTimer = null; + } + this.removeVisibilityListener?.(); + this.removeVisibilityListener = null; + this.removePagehideListener?.(); + this.removePagehideListener = null; + if (this.channel) { + this.channel.close(); + this.channel = null; + } + } + + private post(message: CrossTabSseMessage): boolean { + if (!this.channel) { + this.degradeToFallback(); + return false; + } + try { + this.channel.postMessage(message); + return true; + } catch { + this.degradeToFallback(); + return false; + } + } + + private degradeToFallback() { + if (this.degradingToFallback || this.fallbackMode) return; + this.degradingToFallback = true; + this.releaseLeadership({ broadcast: false, resync: false }); + this.clearCandidate(); + this.shutdownTimersAndChannel(); + this.initialized = false; + this.fallbackMode = true; + this.coordinationUnavailable = true; + for (const subscription of this.subscriptions.values()) { + this.startFallbacksFor(subscription); + } + this.degradingToFallback = false; + } + + private startFallbacksFor(subscription: LocalSubscription) { + for (const fallback of subscription.fallbacks.values()) { + if (!fallback.cleanup) { + fallback.cleanup = fallback.subscribe(); + } + } + } + + private closeFallbacks() { + for (const subscription of this.subscriptions.values()) { + for (const fallback of subscription.fallbacks.values()) { + fallback.cleanup?.(); + fallback.cleanup = undefined; + } + } + } + + private clearSubscriptionTimers() { + for (const subscription of this.subscriptions.values()) { + this.clearSubscriptionTimer(subscription); + } + } + + private clearSubscriptionTimer(subscription: LocalSubscription) { + if (!subscription.debounceTimer) return; + clearTimeout(subscription.debounceTimer); + subscription.debounceTimer = null; + } + + private resetIdleState() { + this.leader = null; + this.initialized = false; + this.coordinationUnavailable = false; + this.fallbackMode = false; + } + + private shouldAcceptLeader(incoming: LeaderState): boolean { + if (!this.leader) return true; + if (incoming.generation > this.leader.generation) return true; + if (incoming.generation < this.leader.generation) return false; + if (incoming.leaderId === this.leader.leaderId) return true; + return leaderHasHigherPriority(incoming, this.leader); + } + + private isCurrentLeader(leaderId: string, generation: number): boolean { + return Boolean( + this.leader && + this.leader.leaderId === leaderId && + this.leader.generation === generation, + ); + } + + private currentVisibility(): TabVisibility { + return this.getVisibility() === "hidden" ? "hidden" : "visible"; + } + + private leaderIsFresh(leader: LeaderState): boolean { + return this.now() - leader.lastSeen <= this.timing.leaderStaleMs; + } +} + +const defaultCoordinator = new CrossTabSseCoordinator(); + +export function createCrossTabSseCoordinator(options: CrossTabSseCoordinatorOptions = {}) { + return new CrossTabSseCoordinator(options); +} + +export function subscribeToCrossTabSse({ + coordinator = defaultCoordinator, + ...options +}: SubscribeToCrossTabSseOptions): () => void { + return coordinator.subscribe(options); +} + +function createBrowserBroadcastChannel(name: string): BroadcastChannelLike { + if (typeof BroadcastChannel === "undefined") { + throw new Error("BroadcastChannel is unavailable"); + } + return new BroadcastChannel(name); +} + +function createTabId(): string { + if (typeof crypto !== "undefined" && typeof crypto.randomUUID === "function") { + return crypto.randomUUID(); + } + return `tab-${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}`; +} + +function getBrowserVisibility(): TabVisibility { + if (typeof document === "undefined") return "visible"; + return document.visibilityState === "visible" ? "visible" : "hidden"; +} + +function addBrowserVisibilityChangeListener(handler: () => void): () => void { + if (typeof document === "undefined") return () => {}; + document.addEventListener("visibilitychange", handler); + return () => document.removeEventListener("visibilitychange", handler); +} + +function addBrowserPagehideListener(handler: () => void): () => void { + if (typeof window === "undefined") return () => {}; + window.addEventListener("pagehide", handler); + return () => window.removeEventListener("pagehide", handler); +} + +function parseMessage(data: unknown): CrossTabSseMessage | undefined { + if (!isRecord(data)) return undefined; + if (data.version !== MESSAGE_VERSION) return undefined; + const type = getString(data, "type"); + const tabId = getString(data, "tabId"); + const sentAt = getNumber(data, "sentAt"); + if (!type || !tabId || sentAt === undefined) return undefined; + const base = { type, version: MESSAGE_VERSION, tabId, sentAt }; + + switch (type) { + case "hello": + return { ...base, type: "hello" }; + case "heartbeat": { + const triple = parseLeaderTriple(data); + return triple && { ...base, type: "heartbeat", ...triple }; + } + case "candidate": + return parseCandidate(data, base); + case "leader-changed": { + const triple = parseLeaderTriple(data); + return triple && { ...base, type: "leader-changed", ...triple }; + } + case "release": { + const pair = parseLeaderPair(data); + return pair && { ...base, type: "release", ...pair }; + } + case "resync": + return parseResync(data, base); + case "event": + return parseEvent(data, base); + default: + return undefined; + } +} + +interface ParsedBase { + version: typeof MESSAGE_VERSION; + tabId: string; + sentAt: number; +} + +function parseLeaderPair(data: unknown): { leaderId: string; generation: number } | undefined { + const leaderId = getString(data, "leaderId"); + const generation = getNumber(data, "generation"); + if (!leaderId || generation === undefined) return undefined; + return { leaderId, generation }; +} + +function parseLeaderTriple( + data: unknown, +): { leaderId: string; generation: number; visibility: TabVisibility } | undefined { + const pair = parseLeaderPair(data); + const visibility = getString(data, "visibility"); + if (!pair || !isVisibility(visibility)) return undefined; + return { ...pair, visibility }; +} + +function parseCandidate(data: UnknownRecord, base: ParsedBase): CandidateMessage | undefined { + const candidateId = getString(data, "candidateId"); + const candidateGeneration = getNumber(data, "candidateGeneration"); + const visibility = getString(data, "visibility"); + const observedGeneration = getNumber(data, "observedGeneration"); + const reason = getString(data, "reason"); + const observedLeaderRaw = data.observedLeaderId; + const observedLeaderId = + typeof observedLeaderRaw === "string" ? observedLeaderRaw : observedLeaderRaw === null ? null : undefined; + if ( + !candidateId || + candidateGeneration === undefined || + !isVisibility(visibility) || + observedLeaderId === undefined || + observedGeneration === undefined || + !isCandidateReason(reason) + ) { + return undefined; + } + return { + ...base, + type: "candidate", + candidateId, + candidateGeneration, + visibility, + observedLeaderId, + observedGeneration, + reason, + }; +} + +function parseResync(data: UnknownRecord, base: ParsedBase): ResyncMessage | undefined { + const generation = getNumber(data, "generation"); + const reason = getString(data, "reason"); + const leaderRaw = data.leaderId; + const leaderId = typeof leaderRaw === "string" ? leaderRaw : leaderRaw === null ? null : undefined; + if (generation === undefined || !isCandidateReason(reason) || leaderId === undefined) { + return undefined; + } + return { ...base, type: "resync", leaderId, generation, reason }; +} + +function parseEvent(data: UnknownRecord, base: ParsedBase): EventMessage | undefined { + const pair = parseLeaderPair(data); + if (!pair || !isRecord(data.payload)) return undefined; + return { ...base, type: "event", ...pair, payload: data.payload }; +} + +function isVisibility(value: unknown): value is TabVisibility { + return value === "visible" || value === "hidden"; +} + +function isCandidateReason(value: unknown): value is CandidateReason { + return ( + value === "hidden-leader" || + value === "stale-leader" || + value === "release" || + value === "no-leader" + ); +} + +function candidateHasHigherPriority(candidate: CandidateMessage, other: CandidateMessage): boolean { + if (candidate.visibility !== other.visibility) return candidate.visibility === "visible"; + return candidate.candidateId < other.candidateId; +} + +function leaderHasHigherPriority(candidate: LeaderState, other: LeaderState): boolean { + if (candidate.visibility !== other.visibility) return candidate.visibility === "visible"; + return candidate.leaderId < other.leaderId; +} + +function leaderStateForCandidate(candidate: CandidateMessage, lastSeen: number): LeaderState { + return { + leaderId: candidate.candidateId, + generation: candidate.candidateGeneration, + visibility: candidate.visibility, + lastSeen, + }; +} + +function candidateKey(candidate: CandidateMessage): string { + return `${candidate.candidateGeneration}:${candidate.candidateId}`; +} + +function eventDedupeKey(payload: EventPayload): string | undefined { + if (typeof payload.id === "string" && payload.id.length > 0) { + return payload.id; + } + + const runId = typeof payload.run_id === "string" ? payload.run_id : undefined; + const seq = typeof payload.seq === "number" ? payload.seq : undefined; + const event = typeof payload.event === "string" ? payload.event : undefined; + if (runId && seq != null && event) { + return `${runId}:${seq}:${event}`; + } + return undefined; +} diff --git a/apps/fabro-web/app/lib/queries.ts b/apps/fabro-web/app/lib/queries.ts index 4de23a73e..0da17785e 100644 --- a/apps/fabro-web/app/lib/queries.ts +++ b/apps/fabro-web/app/lib/queries.ts @@ -34,6 +34,9 @@ const immutableOptions: SWRConfiguration = { revalidateOnReconnect: false, }; +type BoardRunsEnvelope = PaginatedEnvelope & + Pick; + export function useAuthConfig() { return useSWR<{ methods: string[] }>(queryKeys.auth.config(), apiFetcher, immutableOptions); } @@ -61,11 +64,7 @@ export function useSystemInfo() { } export function useBoardsRuns() { - return useSWR< - PaginatedEnvelope & { - columns: { id: string; name: string }[]; - } - >(queryKeys.boards.runs(), apiPaginatedFetcher); + return useSWR(queryKeys.boards.runs(), apiPaginatedFetcher); } export function useRun(id: string | undefined) { diff --git a/apps/fabro-web/app/lib/run-events.test.tsx b/apps/fabro-web/app/lib/run-events.test.tsx index 6e323e751..c38ea1dd2 100644 --- a/apps/fabro-web/app/lib/run-events.test.tsx +++ b/apps/fabro-web/app/lib/run-events.test.tsx @@ -4,7 +4,12 @@ import { queryKeysForRunEvent, subscribeToRunEvents, } from "./run-events"; +import { + createCrossTabSseCoordinator, + type BroadcastChannelLike, +} from "./cross-tab-sse"; import { queryKeys } from "./query-keys"; +import type { EventSourceLike } from "./sse"; type MessageHandler = ((event: { data: string }) => void) | null; @@ -25,6 +30,14 @@ class FakeEventSource { } } +class FakeBroadcastChannel implements BroadcastChannelLike { + onmessage: ((event: { data: unknown }) => void) | null = null; + + postMessage() {} + + close() {} +} + describe("queryKeysForRunEvent", () => { test("terminal events invalidate run-scoped resources", () => { expect(queryKeysForRunEvent("run-1", "run.completed")).toEqual([ @@ -47,10 +60,76 @@ describe("queryKeysForRunEvent", () => { }); describe("subscribeToRunEvents", () => { - test("refcounts shared sources and keeps mutators active until final unsubscribe", () => { + test("coordinated mode uses the global attach stream and filters by run_id", async () => { const source = new FakeEventSource(); const created: string[] = []; const keys: string[] = []; + const coordinator = createCoordinator((url) => { + created.push(url); + return source; + }); + + const cleanup = subscribeToRunEvents( + "run-coordinated", + (key) => { + keys.push(key); + return Promise.resolve(); + }, + () => { + throw new Error("source should be created by coordinator"); + }, + { debounceMs: 0, coordinator }, + ); + + await waitFor(() => created.length === 1); + keys.length = 0; + + source.emit({ event: "checkpoint.completed", run_id: "other-run" }); + source.emit({ event: "checkpoint.completed", run_id: "run-coordinated" }); + + expect(created).toEqual(["/api/v1/attach"]); + expect(keys).toEqual([queryKeys.runs.files("run-coordinated")]); + + cleanup(); + coordinator.close(); + }); + + test("coordinated terminal events invalidate without closing the global stream", async () => { + const source = new FakeEventSource(); + const keys: string[] = []; + const coordinator = createCoordinator(() => source); + const cleanup = subscribeToRunEvents( + "run-terminal", + (key) => { + keys.push(key); + return Promise.resolve(); + }, + () => source, + { debounceMs: 0, coordinator }, + ); + + await waitFor(() => source.onmessage !== null); + keys.length = 0; + + source.emit({ event: "run.failed", run_id: "run-terminal" }); + expect(source.closed).toBe(false); + expect(keys).toContain(queryKeys.runs.files("run-terminal")); + expect(keys).toContain(queryKeys.runs.billing("run-terminal")); + + keys.length = 0; + source.emit({ event: "run.archived", run_id: "run-terminal" }); + expect(source.closed).toBe(false); + expect(keys).toEqual([queryKeys.runs.detail("run-terminal")]); + + cleanup(); + coordinator.close(); + }); + + test("fallback refcounts run-scoped sources and keeps mutators active until final unsubscribe", () => { + const source = new FakeEventSource(); + const created: string[] = []; + const keys: string[] = []; + const coordinator = createFallbackCoordinator(); const mutate = (key: string) => { keys.push(key); return Promise.resolve(); @@ -59,10 +138,10 @@ describe("subscribeToRunEvents", () => { const firstCleanup = subscribeToRunEvents("run-refcount", mutate, (url) => { created.push(url); return source; - }, { debounceMs: 0 }); + }, { debounceMs: 0, coordinator }); const secondCleanup = subscribeToRunEvents("run-refcount", mutate, () => { throw new Error("source should be reused"); - }, { debounceMs: 0 }); + }, { debounceMs: 0, coordinator }); expect(created).toEqual(["/api/v1/runs/run-refcount/attach"]); @@ -74,11 +153,13 @@ describe("subscribeToRunEvents", () => { secondCleanup(); expect(source.closed).toBe(true); + coordinator.close(); }); - test("terminal events close the source after invalidating keys", () => { + test("fallback terminal events close the source after invalidating keys", () => { const source = new FakeEventSource(); const keys: string[] = []; + const coordinator = createFallbackCoordinator(); const cleanup = subscribeToRunEvents( "run-terminal", (key) => { @@ -86,7 +167,7 @@ describe("subscribeToRunEvents", () => { return Promise.resolve(); }, () => source, - { debounceMs: 0 }, + { debounceMs: 0, coordinator }, ); source.emit({ event: "run.failed" }); @@ -96,11 +177,13 @@ describe("subscribeToRunEvents", () => { expect(keys).toContain(queryKeys.runs.billing("run-terminal")); cleanup(); + coordinator.close(); }); - test("envelope with suffixed stage_id invalidates stageTurns(runId, stageId)", () => { + test("envelope with suffixed stage_id invalidates stageTurns(runId, stageId)", async () => { const source = new FakeEventSource(); const keys: string[] = []; + const coordinator = createCoordinator(() => source); const cleanup = subscribeToRunEvents( "run-stage", (key) => { @@ -108,10 +191,16 @@ describe("subscribeToRunEvents", () => { return Promise.resolve(); }, () => source, - { debounceMs: 0 }, + { debounceMs: 0, coordinator }, ); - source.emit({ event: "stage.retrying", stage_id: "verify@2", node_id: "verify" }); + await waitFor(() => source.onmessage !== null); + source.emit({ + event: "stage.retrying", + run_id: "run-stage", + stage_id: "verify@2", + node_id: "verify", + }); expect(keys).toContain(queryKeys.runs.stageTurns("run-stage", "verify@2")); expect(keys).toContain(queryKeys.runs.stages("run-stage")); @@ -121,11 +210,13 @@ describe("subscribeToRunEvents", () => { expect(keys).not.toContain(queryKeys.runs.stageTurns("run-stage", "verify")); cleanup(); + coordinator.close(); }); - test("falls back to node_id when an event has no stage_id", () => { + test("falls back to node_id when an event has no stage_id", async () => { const source = new FakeEventSource(); const keys: string[] = []; + const coordinator = createCoordinator(() => source); const cleanup = subscribeToRunEvents( "run-stage-node", (key) => { @@ -133,22 +224,25 @@ describe("subscribeToRunEvents", () => { return Promise.resolve(); }, () => source, - { debounceMs: 0 }, + { debounceMs: 0, coordinator }, ); - source.emit({ event: "stage.started", node_id: "verify" }); + await waitFor(() => source.onmessage !== null); + source.emit({ event: "stage.started", run_id: "run-stage-node", node_id: "verify" }); expect(keys).toContain(queryKeys.runs.stageTurns("run-stage-node", "verify")); expect(keys).toContain(queryKeys.runs.stages("run-stage-node")); cleanup(); + coordinator.close(); }); - test("malformed events are ignored and StrictMode-style cleanup does not underflow", () => { + test("fallback malformed events are ignored and StrictMode-style cleanup does not underflow", () => { const firstSource = new FakeEventSource(); const secondSource = new FakeEventSource(); const sources = [firstSource, secondSource]; const keys: string[] = []; + const coordinator = createFallbackCoordinator(); const firstCleanup = subscribeToRunEvents( "run-strict", @@ -157,7 +251,7 @@ describe("subscribeToRunEvents", () => { return Promise.resolve(); }, () => sources.shift()!, - { debounceMs: 0 }, + { debounceMs: 0, coordinator }, ); firstSource.emitRaw("{broken"); firstCleanup(); @@ -169,12 +263,45 @@ describe("subscribeToRunEvents", () => { return Promise.resolve(); }, () => sources.shift()!, - { debounceMs: 0 }, + { debounceMs: 0, coordinator }, ); secondCleanup(); expect(keys).toEqual([]); expect(firstSource.closed).toBe(true); expect(secondSource.closed).toBe(true); + coordinator.close(); }); }); + +function createCoordinator(eventSourceFactory: (url: string) => EventSourceLike) { + return createCrossTabSseCoordinator({ + tabId: "run-test", + channelFactory: () => new FakeBroadcastChannel(), + eventSourceFactory, + addVisibilityChangeListener: () => () => {}, + addPagehideListener: () => () => {}, + timing: { + heartbeatMs: 10, + leaderStaleMs: 50, + electionJitterMs: 0, + }, + }); +} + +function createFallbackCoordinator() { + return createCrossTabSseCoordinator({ + channelFactory: () => { + throw new Error("BroadcastChannel unavailable"); + }, + }); +} + +async function waitFor(condition: () => boolean, timeoutMs = 200) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (condition()) return; + await new Promise((resolve) => setTimeout(resolve, 2)); + } + throw new Error("condition did not become true before timeout"); +} diff --git a/apps/fabro-web/app/lib/run-events.ts b/apps/fabro-web/app/lib/run-events.ts index 00bd93e4f..a81a11c2b 100644 --- a/apps/fabro-web/app/lib/run-events.ts +++ b/apps/fabro-web/app/lib/run-events.ts @@ -1,6 +1,10 @@ import { useEffect } from "react"; import { useSWRConfig } from "swr"; +import { + subscribeToCrossTabSse, + type CrossTabSseCoordinator, +} from "./cross-tab-sse"; import { queryKeys } from "./query-keys"; import { createBrowserEventSource, @@ -13,11 +17,17 @@ import { interface RunEventPayload extends EventPayload { event?: string; + run_id?: string; node_id?: string; stage_id?: string; properties?: Record; } +interface RunEventOptions { + debounceMs?: number; + coordinator?: CrossTabSseCoordinator; +} + const subscriptions = new Map(); const TERMINAL_EVENTS = new Set(["run.completed", "run.failed"]); @@ -110,31 +120,57 @@ export function subscribeToRunEvents( runId: string, mutate: MutateFn, eventSourceFactory: (url: string) => EventSourceLike = createBrowserEventSource, - { debounceMs = 300 }: { debounceMs?: number } = {}, + { debounceMs = 300, coordinator }: RunEventOptions = {}, ): () => void { - return subscribeToSharedEventSource({ - subscriptions, - subscriptionKey: runId, - url: queryKeys.runs.attach(runId), + return subscribeToCrossTabSse({ + coordinator, + subscriptionKey: `run:${runId}`, mutate, - eventSourceFactory, debounceMs, + resyncKeys: () => resyncKeysForRun(runId), resolveInvalidation: (payload) => { - const event = payload.event; - if (!event) return { keys: [] }; - - const stageId = stageIdFromPayload(payload); - const keys = queryKeysForRunEvent(runId, event, stageId); - const terminal = TERMINAL_EVENTS.has(event); - return { - keys, - close: terminal, - immediate: terminal, - }; + if (payload.run_id !== runId) return { keys: [] }; + return runInvalidation(runId, payload); }, + fallbackSubscribe: () => + subscribeToSharedEventSource({ + subscriptions, + subscriptionKey: runId, + url: queryKeys.runs.attach(runId), + mutate, + eventSourceFactory, + debounceMs, + resolveInvalidation: (payload) => { + const result = runInvalidation(runId, payload); + return { ...result, close: result.immediate }; + }, + }), }); } +function runInvalidation(runId: string, payload: RunEventPayload) { + const event = payload.event; + if (!event) return { keys: [], immediate: false }; + + const stageId = stageIdFromPayload(payload); + const keys = queryKeysForRunEvent(runId, event, stageId); + const terminal = TERMINAL_EVENTS.has(event); + return { keys, immediate: terminal }; +} + +function resyncKeysForRun(runId: string) { + return [ + queryKeys.runs.detail(runId), + queryKeys.runs.files(runId), + queryKeys.runs.billing(runId), + queryKeys.runs.stages(runId), + queryKeys.runs.events(runId, 1000), + queryKeys.runs.graph(runId, "LR"), + queryKeys.runs.graph(runId, "TB"), + queryKeys.runs.questions(runId, 25, 0), + ]; +} + function stageIdFromPayload(payload: RunEventPayload): string | undefined { if (typeof payload.stage_id === "string") return payload.stage_id; if (typeof payload.node_id === "string") return payload.node_id; diff --git a/apps/fabro-web/app/routes/runs.test.tsx b/apps/fabro-web/app/routes/runs.test.tsx index 948900769..3cd9dec0a 100644 --- a/apps/fabro-web/app/routes/runs.test.tsx +++ b/apps/fabro-web/app/routes/runs.test.tsx @@ -25,6 +25,7 @@ describe("runs route board mapping", () => { test("keeps blocked runs in the blocked lane and preserves question text", () => { const columns = buildBoardColumns({ columns: [ + { id: "queued", name: "Queued" }, { id: "initializing", name: "Initializing" }, { id: "running", name: "Running" }, { id: "blocked", name: "Blocked" }, diff --git a/apps/fabro-web/app/routes/runs.tsx b/apps/fabro-web/app/routes/runs.tsx index f46c19396..78db03496 100644 --- a/apps/fabro-web/app/routes/runs.tsx +++ b/apps/fabro-web/app/routes/runs.tsx @@ -18,7 +18,7 @@ import { arrayMove, } from "@dnd-kit/sortable"; import { CSS } from "@dnd-kit/utilities"; -import { ciConfig, columnStatusDisplay, deriveCiStatus, mapRunListItem } from "../data/runs"; +import { ciConfig, columnStatusDisplay, columnStatuses, deriveCiStatus, mapRunListItem } from "../data/runs"; import type { CiStatus, CheckRun, CheckStatus, RunItem, RunWithStatus, ColumnStatus } from "../data/runs"; import { EmptyState } from "../components/state"; import { shouldRefreshBoardForEvent, useBoardEvents } from "../lib/board-events"; @@ -38,6 +38,7 @@ interface ColumnStyle { } const columnStyles: Record = { + queued: { iconType: "branch", actions: [] }, initializing: { iconType: "branch", actions: [] }, running: { iconType: "branch", actions: ["Watch", "Steer"] }, blocked: { iconType: "branch", actions: ["Answer Question"] }, @@ -49,7 +50,7 @@ const defaultColumnStyle: ColumnStyle = { iconType: "branch", actions: [] }; const defaultColumnColors = { dot: "bg-fg-muted", text: "text-fg-muted" }; interface BoardRunsResponse { - columns: { id: string; name: string }[]; + columns: PaginatedBoardRunList["columns"]; data: PaginatedBoardRunList["data"]; meta: PaginatedBoardRunList["meta"]; } @@ -64,16 +65,8 @@ type Column = { items: RunItem[]; }; -const SKELETON_STATUSES: ColumnStatus[] = [ - "initializing", - "running", - "blocked", - "succeeded", - "failed", -]; - function buildSkeletonColumns(): Column[] { - return SKELETON_STATUSES.map((id) => { + return columnStatuses.map((id) => { const colors = columnStatusDisplay[id]; return { id, @@ -98,7 +91,7 @@ export function buildBoardColumns(response: BoardRunsResponse): Column[] { } return response.columns.map((col) => { - const id = col.id as ColumnStatus; + const id = col.id; const colors = columnStatusDisplay[id] ?? defaultColumnColors; return { id, @@ -754,6 +747,9 @@ export default function Runs() { (sum, col) => sum + col.items.length, 0, ); + const visibleColumns = filteredColumns.filter( + (col) => col.id !== "queued" || col.items.length > 0, + ); return ( @@ -815,7 +811,7 @@ export default function Runs() { {view === "columns" ? ( <>
- {filteredColumns.map((col) => ( + {visibleColumns.map((col) => (
@@ -838,7 +834,7 @@ export default function Runs() { ) : ( <>
- {filteredColumns.map((col) => { + {visibleColumns.map((col) => { const isCollapsed = collapsed.has(col.id); return (
diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index bea733407..46b1b81cf 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/docs/public/api-reference/fabro-api.yaml @@ -5656,6 +5656,7 @@ components: description: Board column status for a run in the list view. type: string enum: + - queued - initializing - running - blocked @@ -5669,7 +5670,7 @@ components: - name properties: id: - type: string + $ref: "#/components/schemas/BoardColumn" name: type: string diff --git a/docs/superpowers/plans/2026-05-04-cross-tab-sse-coordination.md b/docs/superpowers/plans/2026-05-04-cross-tab-sse-coordination.md new file mode 100644 index 000000000..115ff9326 --- /dev/null +++ b/docs/superpowers/plans/2026-05-04-cross-tab-sse-coordination.md @@ -0,0 +1,106 @@ +# Cross-Tab SSE Coordination Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Coordinate Fabro web SSE subscriptions across tabs so one browser profile/origin opens at most one UI-owned `/api/v1/attach` EventSource in steady state when `BroadcastChannel` is available; brief overlap during election/takeover is tolerated and deduped. + +**Architecture:** Add a browser-side SSE coordinator that elects one tab as leader, has that leader own the global EventSource, and broadcasts parsed run events to follower tabs over `BroadcastChannel`. Existing board and run-detail invalidation logic becomes a consumer of that global event feed, with the current per-tab SSE behavior preserved as a compatibility fallback. + +**Tech Stack:** React, SWR, browser `BroadcastChannel`, browser `EventSource`, Bun tests, existing Fabro web API query keys. + +--- + +## Summary + +Build a browser-side SSE coordinator so Fabro web opens at most one `/api/v1/attach` EventSource per origin/browser profile in steady state when `BroadcastChannel` is available. Temporary duplicate leaders may exist during election/takeover, but event dedupe and generation checks make the overlap harmless and short-lived. The global stream becomes a shared cache-invalidation feed for both the runs board and run detail pages. No server API, OpenAPI, or Rust streaming contract changes are part of v1. + +This supersedes the earlier web SSE limitation documented in `docs/plans/2026-04-19-002-feat-web-ui-lifecycle-actions-plan.md`: the old shared hook was code reuse only; this plan adds actual socket deduplication. + +## Implementation Changes + +- Add `apps/fabro-web/app/lib/cross-tab-sse.ts`. + - Export `subscribeToCrossTabSse(...)` with the same invalidation style as `subscribeToSharedEventSource`, a `resyncKeys` callback for gap recovery, and a `fallbackSubscribe` callback used when cross-tab coordination is unavailable. + - Use `BroadcastChannel` name `fabro:sse:v1`. + - Generate `tabId` with `crypto.randomUUID()` and a safe random fallback. + - Open one leader-owned `EventSource` to `queryKeys.system.attach()` (`/api/v1/attach`) in steady state. + - Leader dispatches each parsed `EventEnvelope` locally and broadcasts it to follower tabs. + - Followers do not open EventSource while a valid visible leader heartbeat exists; lower lexical `tabId` does not preempt a healthy visible leader. + +- Implement leader election in the cross-tab module. + - Constants: `HEARTBEAT_MS = 1000`, `LEADER_STALE_MS = 4000`, `ELECTION_JITTER_MS = 150`. + - Messages: `hello`, `heartbeat`, `candidate`, `leader-changed`, `release`, `resync`, `event`. + - Define a typed message union. Every message includes `type`, `version: 1`, `tabId`, and `sentAt`. + - `heartbeat`: `{ type, version, tabId, sentAt, leaderId, generation, visibility }`. + - `candidate`: `{ type, version, tabId, sentAt, candidateId: tabId, candidateGeneration, visibility, observedLeaderId, observedGeneration, reason }`, where `reason` is `"hidden-leader" | "stale-leader" | "release" | "no-leader"`. + - `leader-changed`: `{ type, version, tabId, sentAt, leaderId, generation, visibility }`. + - `release`: `{ type, version, tabId, sentAt, leaderId, generation }`. + - `resync`: `{ type, version, tabId, sentAt, leaderId, generation, reason }`. + - `event`: `{ type, version, tabId, sentAt, leaderId, generation, payload }`. + - Use the candidacy phase for all leadership changes: hidden-leader takeover, stale-leader recovery, leader release, and no-leader startup. + - A candidate sets `candidateGeneration = observedGeneration + 1`, broadcasts `candidate`, waits jitter, and opens EventSource only if no higher-priority candidate for the same `candidateGeneration` appears. + - Candidate priority is election-scoped: visible candidates outrank hidden candidates; for equal visibility, lower lexical `candidateId` wins. This priority resolves elections and same-generation split brain only; it is not a reason to preempt a fresh visible leader. + - When a visible follower observes a fresh hidden leader heartbeat, it enters candidacy with `reason: "hidden-leader"`. + - When tabs detect a stale leader, leader release, or no known leader, they enter the same candidacy flow with the matching `reason`. + - If two visible candidates race for the same observed leader/generation, the lower lexical `candidateId` wins. + - Current leaders release when they observe a candidate whose `observedLeaderId` matches their `leaderId` and whose `observedGeneration` is current or newer. + - If same-generation split brain still occurs, lower-priority leaders release when they observe a same-generation higher-priority leader heartbeat or `leader-changed`. + - Each new leader uses `candidateGeneration`, broadcasts `leader-changed`, and followers ignore heartbeats/events from non-current leaders or stale generations. + - Hidden leader keeps the stream only when no visible candidate takes over. + - On `pagehide`/last local unsubscribe, a leader closes EventSource and broadcasts `release`. + - Brief split brain is tolerated; dedupe events by `payload.id`, falling back to `${run_id}:${seq}:${event}`. + - Keep dedupe bounded with a recent-event cache: max 1000 IDs and 5 minute TTL. Evict oldest entries when the max is exceeded and prune expired entries during event handling. Duplicate invalidations after eviction are acceptable; unbounded growth is not. + +- Migrate consumers. + - `apps/fabro-web/app/lib/board-events.ts`: subscribe through the cross-tab global stream; keep existing board event allowlist. + - `apps/fabro-web/app/lib/run-events.ts`: subscribe through the same global stream, filter by `payload.run_id === runId`, and reuse `queryKeysForRunEvent`. + - In coordinated mode, run detail pages stay subscribed while mounted, including terminal runs, so post-terminal archive/unarchive changes can reconcile live. + - Do not close the global stream on `run.completed` / `run.failed`; terminal events only invalidate run-scoped keys. + - Keep `subscribeToSharedEventSource` in `apps/fabro-web/app/lib/sse.ts` for fallback and existing local sharing behavior. + +- Gap and fallback behavior. + - If `BroadcastChannel` is unavailable or throws, call each subscriber's `fallbackSubscribe`. + - Board fallback uses the existing global `/api/v1/attach` path. + - Run detail fallback preserves the existing run-scoped `/api/v1/runs/:id/attach` path, so the old terminal-tab stale limitation remains only in fallback mode. + - Do not add replay to `/api/v1/attach`. + - On leader takeover, stale leader timeout, leader release, and new leader generation, broadcast `resync` or `leader-changed` so every tab with active local subscriptions runs its own `resyncKeys`. + - On `visibilitychange` back to visible without leadership change, run only that tab's local `resyncKeys`; do not broadcast cross-tab resync. + - Board `resyncKeys`: `queryKeys.boards.runs()`. + - Run `resyncKeys`: detail, files, billing, stages, events, LR graph, TB graph, and questions for that run. + +## Tests + +- Add `apps/fabro-web/app/lib/cross-tab-sse.test.ts` with fake `BroadcastChannel`, fake `EventSource`, and fake timers. + - One leader opens `/api/v1/attach`; followers open no EventSource. + - Leader broadcasts an event and all local subscribers receive invalidations. + - Run subscribers ignore events for other `run_id` values. + - Board and run subscriptions can coexist on the same global stream. + - Temporary duplicate leaders are allowed only during election/takeover and converge back to one leader. + - With fresh hidden-leader heartbeats, a visible tab broadcasts candidacy, hidden leader closes, visible tab opens `/api/v1/attach`, and followers resync. + - Two visible candidates racing for the same hidden leader resolve to the lower lexical `candidateId`. + - Two tabs detect the same stale leader simultaneously; only the winning candidate opens `/api/v1/attach` after jitter. + - Same-generation split brain converges to one leader by visibility, then lexical `tabId`. + - A fresh visible leader exists; a new visible follower with a lower lexical `tabId` joins and does not take leadership. + - A hidden leader does not release for a candidate that references an old `observedLeaderId` or stale `observedGeneration`. + - Stale heartbeat triggers takeover and every active tab calls its own `resyncKeys`. + - A follower tab calls its own board/run `resyncKeys` after another tab becomes leader. + - Duplicate event IDs are ignored. + - The recent-event dedupe cache evicts by TTL and max-size bound; duplicate invalidation may recur only after eviction. + - Stale heartbeat/event messages from an old leader/generation are ignored after takeover. + - Last unsubscribe closes leader EventSource and sends `release`. + - Missing/broken BroadcastChannel uses the per-tab fallback. + +- Update existing tests. + - `apps/fabro-web/app/lib/board-events.test.tsx`: assert coordinated mode uses `/api/v1/attach` once and fallback preserves current behavior. + - `apps/fabro-web/app/lib/run-events.test.tsx`: assert coordinated mode filters by `run_id`; fallback preserves current `/runs/:id/attach` behavior; terminal run events do not close the global coordinator; terminal detail tabs still receive archive/unarchive invalidations while mounted in coordinated mode. + +- Verification commands. + - `cd apps/fabro-web && bun test app/lib/cross-tab-sse.test.ts app/lib/board-events.test.tsx app/lib/run-events.test.tsx` + - `cd apps/fabro-web && bun run typecheck` + - Optional manual check: open 8 run-detail tabs for active runs; after election settles and with BroadcastChannel available, Chrome should show one active `/api/v1/attach` EventStream across the participating tabs and no UI-created `/api/v1/runs/:id/attach` streams. + +## Assumptions + +- Treat SSE as live cache invalidation, not an exact gapless event log. +- Preserve current run-specific attach endpoint for CLI, API clients, and fallback only. +- Do not touch unrelated dirty worktree files; intentionally replace or adapt the earlier interrupted SSE test edits as part of the new test suite. +- No docs or public API updates are required for v1 because this is an internal web transport change. diff --git a/lib/crates/fabro-server/src/demo/mod.rs b/lib/crates/fabro-server/src/demo/mod.rs index f338cdc04..127589a0e 100644 --- a/lib/crates/fabro-server/src/demo/mod.rs +++ b/lib/crates/fabro-server/src/demo/mod.rs @@ -795,8 +795,8 @@ mod runs { .collect() } - fn demo_run_ids() -> &'static [RunId; 6] { - static IDS: OnceLock<[RunId; 6]> = OnceLock::new(); + fn demo_run_ids() -> &'static [RunId; 7] { + static IDS: OnceLock<[RunId; 7]> = OnceLock::new(); IDS.get_or_init(|| { [ RunId::with_timestamp(ts("2026-03-06T14:30:00Z"), 1), @@ -805,6 +805,7 @@ mod runs { RunId::with_timestamp(ts("2026-03-04T10:00:00Z"), 4), RunId::with_timestamp(ts("2026-03-03T16:45:00Z"), 5), RunId::with_timestamp(ts("2026-02-28T14:00:00Z"), 6), + RunId::with_timestamp(ts("2026-03-06T14:35:00Z"), 7), ] }) } @@ -974,23 +975,27 @@ mod runs { pub(super) fn columns() -> Vec { vec![ BoardColumnDefinition { - id: "initializing".into(), + id: BoardColumn::Queued, + name: "Queued".into(), + }, + BoardColumnDefinition { + id: BoardColumn::Initializing, name: "Initializing".into(), }, BoardColumnDefinition { - id: "running".into(), + id: BoardColumn::Running, name: "Running".into(), }, BoardColumnDefinition { - id: "blocked".into(), + id: BoardColumn::Blocked, name: "Blocked".into(), }, BoardColumnDefinition { - id: "succeeded".into(), + id: BoardColumn::Succeeded, name: "Succeeded".into(), }, BoardColumnDefinition { - id: "failed".into(), + id: BoardColumn::Failed, name: "Failed".into(), }, ] @@ -1082,6 +1087,20 @@ mod runs { Some(720000), &[("release", "preview")], ), + summary( + 7, + "api-server", + "implement", + "Implement", + "Add audit log retention policy", + "queued", + "2026-03-06T14:35:00Z", + None, + None, + None, + None, + &[("owner", "platform")], + ), ] } @@ -1150,6 +1169,13 @@ mod runs { None, None, ), + board_item( + take_summary(&mut summaries, demo_run_id(7)), + BoardColumn::Queued, + None, + None, + None, + ), ] } diff --git a/lib/crates/fabro-server/src/server/handler/runs.rs b/lib/crates/fabro-server/src/server/handler/runs.rs index ff03b55e5..9f31597ad 100644 --- a/lib/crates/fabro-server/src/server/handler/runs.rs +++ b/lib/crates/fabro-server/src/server/handler/runs.rs @@ -10,7 +10,9 @@ use axum::{Json, Router}; use base64::Engine as _; use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; use bytes::Bytes; -use fabro_api::types::{RunManifest, RunStatusResponse, SubmitAnswerRequest}; +use fabro_api::types::{ + BoardColumn, BoardColumnDefinition, RunManifest, RunStatusResponse, SubmitAnswerRequest, +}; use fabro_config::Storage; use fabro_interview::AnswerSubmission; use fabro_types::{ @@ -82,25 +84,45 @@ impl ListRunsParams { } } -fn board_column(status: RunStatus) -> Option<&'static str> { +fn board_column(status: RunStatus) -> Option { match status { - RunStatus::Submitted | RunStatus::Queued | RunStatus::Starting => Some("initializing"), - RunStatus::Running | RunStatus::Paused { .. } => Some("running"), - RunStatus::Blocked { .. } => Some("blocked"), - RunStatus::Succeeded { .. } => Some("succeeded"), - RunStatus::Failed { .. } | RunStatus::Dead => Some("failed"), + RunStatus::Submitted | RunStatus::Queued => Some(BoardColumn::Queued), + RunStatus::Starting => Some(BoardColumn::Initializing), + RunStatus::Running | RunStatus::Paused { .. } => Some(BoardColumn::Running), + RunStatus::Blocked { .. } => Some(BoardColumn::Blocked), + RunStatus::Succeeded { .. } => Some(BoardColumn::Succeeded), + RunStatus::Failed { .. } | RunStatus::Dead => Some(BoardColumn::Failed), RunStatus::Removing | RunStatus::Archived { .. } => None, } } -pub(crate) fn board_columns() -> serde_json::Value { - serde_json::json!([ - {"id": "initializing", "name": "Initializing"}, - {"id": "running", "name": "Running"}, - {"id": "blocked", "name": "Blocked"}, - {"id": "succeeded", "name": "Succeeded"}, - {"id": "failed", "name": "Failed"}, - ]) +pub(crate) fn board_columns() -> Vec { + vec![ + BoardColumnDefinition { + id: BoardColumn::Queued, + name: "Queued".into(), + }, + BoardColumnDefinition { + id: BoardColumn::Initializing, + name: "Initializing".into(), + }, + BoardColumnDefinition { + id: BoardColumn::Running, + name: "Running".into(), + }, + BoardColumnDefinition { + id: BoardColumn::Blocked, + name: "Blocked".into(), + }, + BoardColumnDefinition { + id: BoardColumn::Succeeded, + name: "Succeeded".into(), + }, + BoardColumnDefinition { + id: BoardColumn::Failed, + name: "Failed".into(), + }, + ] } async fn board_run_metadata( diff --git a/lib/crates/fabro-server/src/server/tests.rs b/lib/crates/fabro-server/src/server/tests.rs index d984d30bd..89f347870 100644 --- a/lib/crates/fabro-server/src/server/tests.rs +++ b/lib/crates/fabro-server/src/server/tests.rs @@ -6572,7 +6572,7 @@ async fn pause_run_sets_pending_control_on_board_response() { assert_eq!(body["pending_control"].as_str(), Some("pause")); // Verify the run appears on the board (store has Submitted status → - // "initializing" column) + // "queued" column) let req = Request::builder() .method("GET") .uri(api("/boards/runs")) @@ -6587,7 +6587,7 @@ async fn pause_run_sets_pending_control_on_board_response() { .find(|item| item["run_id"].as_str() == Some(run_id_str.as_str())) .expect("board item should exist"); assert!(item["status"].is_object()); - assert_eq!(item["column"].as_str(), Some("initializing")); + assert_eq!(item["column"].as_str(), Some("queued")); assert_eq!(item["pending_control"].as_str(), Some("pause")); } @@ -7062,8 +7062,8 @@ async fn queue_position_reported_for_queued_runs() { let first_run_id = create_and_start_run(&app, MINIMAL_DOT).await; let second_run_id = create_and_start_run(&app, MINIMAL_DOT).await; - // Queued runs are excluded from the board, so verify queue positions - // via the in-memory state directly. + // Queue position is tracked in memory even when queued runs are also + // visible on the board. let runs = state.runs.lock().expect("runs lock poisoned"); let positions = compute_queue_positions(&runs); let first_id = first_run_id.parse::().unwrap(); diff --git a/lib/packages/fabro-api-client/src/models/board-column-definition.ts b/lib/packages/fabro-api-client/src/models/board-column-definition.ts index f1730a16c..20deba1f5 100644 --- a/lib/packages/fabro-api-client/src/models/board-column-definition.ts +++ b/lib/packages/fabro-api-client/src/models/board-column-definition.ts @@ -13,9 +13,11 @@ */ +// May contain unused imports in some cases +// @ts-ignore +import type { BoardColumn } from './board-column'; export interface BoardColumnDefinition { - 'id': string; + 'id': BoardColumn; 'name': string; } - diff --git a/lib/packages/fabro-api-client/src/models/board-column.ts b/lib/packages/fabro-api-client/src/models/board-column.ts index 9d97dc120..66c24c107 100644 --- a/lib/packages/fabro-api-client/src/models/board-column.ts +++ b/lib/packages/fabro-api-client/src/models/board-column.ts @@ -19,6 +19,7 @@ */ export const BoardColumn = { + QUEUED: 'queued', INITIALIZING: 'initializing', RUNNING: 'running', BLOCKED: 'blocked',