From f39e512990469ab9a84f41aa59f93ddc3e864692 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 4 May 2026 14:13:01 -0400 Subject: [PATCH 1/8] feat(web): split Queued column out of Initializing on the run board Submitted and Queued lifecycle statuses now live in a dedicated Queued column rendered to the left of Initializing; Starting stays in Initializing. The column is omitted from the board when it has no items so day-to-day boards stay compact. Co-Authored-By: Claude Opus 4.7 (1M context) --- apps/fabro-web/app/data/runs.ts | 4 ++- apps/fabro-web/app/routes/runs.test.tsx | 1 + apps/fabro-web/app/routes/runs.tsx | 9 ++++-- docs/public/api-reference/fabro-api.yaml | 1 + lib/crates/fabro-server/src/demo/mod.rs | 30 +++++++++++++++++-- .../fabro-server/src/server/handler/runs.rs | 4 ++- lib/crates/fabro-server/src/server/tests.rs | 4 +-- .../src/models/board-column.ts | 1 + 8 files changed, 46 insertions(+), 8 deletions(-) diff --git a/apps/fabro-web/app/data/runs.ts b/apps/fabro-web/app/data/runs.ts index 3922b7cd3..77d8a4711 100644 --- a/apps/fabro-web/app/data/runs.ts +++ b/apps/fabro-web/app/data/runs.ts @@ -37,9 +37,10 @@ export interface RunItem { sourceDirectory?: string; } -export type ColumnStatus = "initializing" | "running" | "blocked" | "succeeded" | "failed"; +export type ColumnStatus = "queued" | "initializing" | "running" | "blocked" | "succeeded" | "failed"; 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 +114,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/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..90ef09405 100644 --- a/apps/fabro-web/app/routes/runs.tsx +++ b/apps/fabro-web/app/routes/runs.tsx @@ -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"] }, @@ -65,6 +66,7 @@ type Column = { }; const SKELETON_STATUSES: ColumnStatus[] = [ + "queued", "initializing", "running", "blocked", @@ -754,6 +756,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 +820,7 @@ export default function Runs() { {view === "columns" ? ( <>
- {filteredColumns.map((col) => ( + {visibleColumns.map((col) => (
@@ -838,7 +843,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 d1b8bc7f3..d10db4a47 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 diff --git a/lib/crates/fabro-server/src/demo/mod.rs b/lib/crates/fabro-server/src/demo/mod.rs index c9751cf83..787fb1ae4 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), ] }) } @@ -973,6 +974,10 @@ mod runs { pub(super) fn columns() -> Vec { vec![ + BoardColumnDefinition { + id: "queued".into(), + name: "Queued".into(), + }, BoardColumnDefinition { id: "initializing".into(), name: "Initializing".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..eb48c7c4f 100644 --- a/lib/crates/fabro-server/src/server/handler/runs.rs +++ b/lib/crates/fabro-server/src/server/handler/runs.rs @@ -84,7 +84,8 @@ impl ListRunsParams { fn board_column(status: RunStatus) -> Option<&'static str> { match status { - RunStatus::Submitted | RunStatus::Queued | RunStatus::Starting => Some("initializing"), + RunStatus::Submitted | RunStatus::Queued => Some("queued"), + RunStatus::Starting => Some("initializing"), RunStatus::Running | RunStatus::Paused { .. } => Some("running"), RunStatus::Blocked { .. } => Some("blocked"), RunStatus::Succeeded { .. } => Some("succeeded"), @@ -95,6 +96,7 @@ fn board_column(status: RunStatus) -> Option<&'static str> { pub(crate) fn board_columns() -> serde_json::Value { serde_json::json!([ + {"id": "queued", "name": "Queued"}, {"id": "initializing", "name": "Initializing"}, {"id": "running", "name": "Running"}, {"id": "blocked", "name": "Blocked"}, diff --git a/lib/crates/fabro-server/src/server/tests.rs b/lib/crates/fabro-server/src/server/tests.rs index 6078c362f..5480fef0f 100644 --- a/lib/crates/fabro-server/src/server/tests.rs +++ b/lib/crates/fabro-server/src/server/tests.rs @@ -6125,7 +6125,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")) @@ -6140,7 +6140,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")); } 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', From ade721ae65e8b7cf0bb0b93e4bb9e6f13a78397b Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 4 May 2026 14:54:39 -0400 Subject: [PATCH 2/8] feat(web): coordinate SSE subscriptions across tabs Elect a single browser tab to own the global attach stream and broadcast run events to sibling tabs. Keep the existing per-tab EventSource path as the fallback when cross-tab coordination is unavailable. --- apps/fabro-web/app/lib/board-events.test.tsx | 84 +- apps/fabro-web/app/lib/board-events.ts | 41 +- apps/fabro-web/app/lib/cross-tab-sse.test.ts | 385 ++++++ apps/fabro-web/app/lib/cross-tab-sse.ts | 1175 ++++++++++++++++++ apps/fabro-web/app/lib/run-events.test.tsx | 128 +- apps/fabro-web/app/lib/run-events.ts | 75 +- 6 files changed, 1852 insertions(+), 36 deletions(-) create mode 100644 apps/fabro-web/app/lib/cross-tab-sse.test.ts create mode 100644 apps/fabro-web/app/lib/cross-tab-sse.ts diff --git a/apps/fabro-web/app/lib/board-events.test.tsx b/apps/fabro-web/app/lib/board-events.test.tsx index 853d300b9..9b398fdc1 100644 --- a/apps/fabro-web/app/lib/board-events.test.tsx +++ b/apps/fabro-web/app/lib/board-events.test.tsx @@ -4,6 +4,10 @@ import { shouldRefreshBoardForEvent, subscribeToBoardEvents, } from "./board-events"; +import { + createCrossTabSseCoordinator, + type BroadcastChannelLike, +} from "./cross-tab-sse"; import { queryKeys } from "./query-keys"; type MessageHandler = ((event: { data: string }) => void) | null; @@ -21,6 +25,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,10 +43,11 @@ 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(); const mutate = (key: string) => { keys.push(key); return Promise.resolve(); @@ -43,10 +56,13 @@ describe("subscribeToBoardEvents", () => { const firstCleanup = subscribeToBoardEvents(mutate, (url) => { created.push(url); return source; - }, { debounceMs: 0 }); + }, { 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 +73,67 @@ 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() { + return createCrossTabSseCoordinator({ + tabId: "board-test", + channelFactory: () => new FakeBroadcastChannel(), + 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..a182d9ad2 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,37 @@ 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..32cff001c --- /dev/null +++ b/apps/fabro-web/app/lib/cross-tab-sse.test.ts @@ -0,0 +1,385 @@ +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; + + onmessage: ((event: { data: unknown }) => void) | null = null; + closed = false; + + constructor(readonly name: string) { + FakeBroadcastChannel.channels.add(this); + } + + postMessage(message: CrossTabSseMessage) { + 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 } }); + } + }); + } + + 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; + } +} + +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("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 === "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("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("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); + }); +}); + +function newHarness() { + const harness = new Harness(); + harnesses.push(harness); + return harness; +} + +function subscribeForRunEvent(coordinator: CrossTabSseCoordinator, keys: string[]) { + return subscribeToCrossTabSse({ + coordinator, + subscriptionKey: "run-feed", + mutate: ((key: string) => { + keys.push(key); + return Promise.resolve(); + }) as MutateFn, + resolveInvalidation: (payload) => ({ + keys: payload.event === "run.running" ? ["event"] : [], + }), + resyncKeys: () => ["resync"], + fallbackSubscribe: () => { + throw new Error("fallback should not be used"); + }, + debounceMs: 0, + }); +} + +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..255c4c7ce --- /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 { isRecord } 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; + eventSourceFactory?: (url: string) => EventSourceLike; + 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.prune(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 prune(now: number) { + for (const [key, seenAt] of this.seen) { + if (now - seenAt > this.ttlMs) { + 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; + private sourceFactoryLocked = false; + + 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 { + if (options.eventSourceFactory && !this.source && !this.sourceFactoryLocked) { + this.sourceFactory = options.eventSourceFactory; + this.sourceFactoryLocked = true; + } + + if (this.coordinationUnavailable) { + return options.fallbackSubscribe(); + } + + if (!this.initialized && !this.initialize()) { + this.coordinationUnavailable = true; + return options.fallbackSubscribe(); + } + + const subscription = this.addLocalSubscription(options); + 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.subscriptions.clear(); + this.leader = null; + this.initialized = false; + this.fallbackMode = false; + this.sourceFactoryLocked = false; + } + + 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) { + if (subscription.debounceTimer) { + clearTimeout(subscription.debounceTimer); + } + this.subscriptions.delete(subscriptionKey); + } + + if (this.subscriptions.size === 0) { + this.releaseLeadership({ broadcast: true, resync: false }); + this.clearCandidate(); + this.clearNoLeaderTimer(); + this.shutdownTimersAndChannel(); + this.initialized = false; + this.fallbackMode = false; + this.sourceFactoryLocked = false; + } + } + + 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); + 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); + + 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.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.now() - current.lastSeen > this.timing.leaderStaleMs) { + this.leader = null; + this.generation = Math.max(this.generation, current.generation); + 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.now() - this.leader.lastSeen <= this.timing.leaderStaleMs + ) { + 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); + this.post(candidate); + 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.now() - this.leader.lastSeen <= this.timing.leaderStaleMs + ) { + 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.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); + }; + + this.post({ + type: "leader-changed", + version: MESSAGE_VERSION, + tabId: this.tabId, + sentAt: this.now(), + leaderId: this.tabId, + generation, + visibility: this.currentVisibility(), + }); + 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); + } + this.sendHeartbeat(); + this.heartbeatTimer = setInterval(() => { + this.sendHeartbeat(); + }, this.timing.heartbeatMs); + } + + private sendHeartbeat() { + if (!this.isLeader) return; + const visibility = this.currentVisibility(); + this.leader = { + leaderId: this.tabId, + generation: this.generation, + visibility, + lastSeen: this.now(), + }; + 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) { + this.post({ + type: "release", + version: MESSAGE_VERSION, + tabId: this.tabId, + sentAt: this.now(), + leaderId: this.tabId, + generation, + }); + 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 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) 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 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"; + } +} + +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 = data.type; + const tabId = data.tabId; + const sentAt = data.sentAt; + if (typeof type !== "string") return undefined; + if (typeof tabId !== "string") return undefined; + if (typeof sentAt !== "number") return undefined; + const base = { tabId, sentAt }; + + switch (type) { + case "hello": + return baseMessage(base, "hello"); + case "heartbeat": { + const { leaderId, generation, visibility } = data; + if (typeof leaderId === "string" && typeof generation === "number" && isVisibility(visibility)) { + return { + ...baseMessage(base, "heartbeat"), + leaderId, + generation, + visibility, + }; + } + return undefined; + } + case "candidate": { + const { + candidateId, + candidateGeneration, + visibility, + observedLeaderId, + observedGeneration, + reason, + } = data; + if ( + typeof candidateId === "string" && + typeof candidateGeneration === "number" && + isVisibility(visibility) && + (typeof observedLeaderId === "string" || observedLeaderId === null) && + typeof observedGeneration === "number" && + isCandidateReason(reason) + ) { + const normalizedObservedLeaderId = + typeof observedLeaderId === "string" ? observedLeaderId : null; + return { + ...baseMessage(base, "candidate"), + candidateId, + candidateGeneration, + visibility, + observedLeaderId: normalizedObservedLeaderId, + observedGeneration, + reason, + }; + } + return undefined; + } + case "leader-changed": { + const { leaderId, generation, visibility } = data; + if (typeof leaderId === "string" && typeof generation === "number" && isVisibility(visibility)) { + return { + ...baseMessage(base, "leader-changed"), + leaderId, + generation, + visibility, + }; + } + return undefined; + } + case "release": { + const { leaderId, generation } = data; + if (typeof leaderId === "string" && typeof generation === "number") { + return { + ...baseMessage(base, "release"), + leaderId, + generation, + }; + } + return undefined; + } + case "resync": { + const { leaderId, generation, reason } = data; + if ( + (typeof leaderId === "string" || leaderId === null) && + typeof generation === "number" && + isCandidateReason(reason) + ) { + const normalizedLeaderId = typeof leaderId === "string" ? leaderId : null; + return { + ...baseMessage(base, "resync"), + leaderId: normalizedLeaderId, + generation, + reason, + }; + } + return undefined; + } + case "event": { + const { leaderId, generation, payload } = data; + if (typeof leaderId === "string" && typeof generation === "number" && isRecord(payload)) { + return { + ...baseMessage(base, "event"), + leaderId, + generation, + payload, + }; + } + return undefined; + } + default: + return undefined; + } +} + +function baseMessage( + data: { + tabId: string; + sentAt: number; + }, + type: TType, +) { + return { + type, + version: MESSAGE_VERSION, + tabId: data.tabId, + sentAt: data.sentAt, + }; +} + +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/run-events.test.tsx b/apps/fabro-web/app/lib/run-events.test.tsx index 14acf4f23..db35a757d 100644 --- a/apps/fabro-web/app/lib/run-events.test.tsx +++ b/apps/fabro-web/app/lib/run-events.test.tsx @@ -4,6 +4,10 @@ import { queryKeysForRunEvent, subscribeToRunEvents, } from "./run-events"; +import { + createCrossTabSseCoordinator, + type BroadcastChannelLike, +} from "./cross-tab-sse"; import { queryKeys } from "./query-keys"; type MessageHandler = ((event: { data: string }) => void) | null; @@ -25,6 +29,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([ @@ -39,10 +51,74 @@ 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(); + + const cleanup = subscribeToRunEvents( + "run-coordinated", + (key) => { + keys.push(key); + return Promise.resolve(); + }, + (url) => { + created.push(url); + return source; + }, + { 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(); + 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(); @@ -51,10 +127,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"]); @@ -66,11 +142,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) => { @@ -78,7 +156,7 @@ describe("subscribeToRunEvents", () => { return Promise.resolve(); }, () => source, - { debounceMs: 0 }, + { debounceMs: 0, coordinator }, ); source.emit({ event: "run.failed" }); @@ -88,13 +166,15 @@ describe("subscribeToRunEvents", () => { expect(keys).toContain(queryKeys.runs.billing("run-terminal")); 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", @@ -103,7 +183,7 @@ describe("subscribeToRunEvents", () => { return Promise.resolve(); }, () => sources.shift()!, - { debounceMs: 0 }, + { debounceMs: 0, coordinator }, ); firstSource.emitRaw("{broken"); firstCleanup(); @@ -115,12 +195,44 @@ 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() { + return createCrossTabSseCoordinator({ + tabId: "run-test", + channelFactory: () => new FakeBroadcastChannel(), + 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 924264af2..80394da48 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,10 +17,16 @@ import { interface RunEventPayload extends EventPayload { event?: string; + run_id?: string; node_id?: string; properties?: Record; } +interface RunEventOptions { + debounceMs?: number; + coordinator?: CrossTabSseCoordinator; +} + const subscriptions = new Map(); const TERMINAL_EVENTS = new Set(["run.completed", "run.failed"]); @@ -104,31 +114,64 @@ 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, { closeOnTerminal: false }); }, + fallbackSubscribe: () => + subscribeToSharedEventSource({ + subscriptions, + subscriptionKey: runId, + url: queryKeys.runs.attach(runId), + mutate, + eventSourceFactory, + debounceMs, + resolveInvalidation: (payload) => + runInvalidation(runId, payload, { closeOnTerminal: true }), + }), }); } +function runInvalidation( + runId: string, + payload: RunEventPayload, + { closeOnTerminal }: { closeOnTerminal: boolean }, +) { + 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: closeOnTerminal && terminal, + 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.node_id === "string") return payload.node_id; const nodeId = payload.properties?.node_id; From 38726666afecdb2b928d0b5f4a5a5fe5e0410319 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 4 May 2026 15:24:25 -0400 Subject: [PATCH 3/8] fix(web): harden cross-tab SSE fallback Stop coordinated election and leadership work when BroadcastChannel posting fails, so tabs degrade cleanly to per-subscriber fallback without stale resync or heartbeat side effects. Expand election coverage for the edge cases called out in the coordination plan. --- apps/fabro-web/app/lib/cross-tab-sse.test.ts | 183 ++++++++++++++++++- apps/fabro-web/app/lib/cross-tab-sse.ts | 21 ++- 2 files changed, 189 insertions(+), 15 deletions(-) diff --git a/apps/fabro-web/app/lib/cross-tab-sse.test.ts b/apps/fabro-web/app/lib/cross-tab-sse.test.ts index 32cff001c..573ef6589 100644 --- a/apps/fabro-web/app/lib/cross-tab-sse.test.ts +++ b/apps/fabro-web/app/lib/cross-tab-sse.test.ts @@ -40,6 +40,7 @@ class FakeEventSource { class FakeBroadcastChannel implements BroadcastChannelLike { static channels = new Set(); static muted = false; + static throwOnTypes = new Set(); onmessage: ((event: { data: unknown }) => void) | null = null; closed = false; @@ -49,6 +50,9 @@ class FakeBroadcastChannel implements BroadcastChannelLike { } 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, @@ -61,6 +65,15 @@ class FakeBroadcastChannel implements BroadcastChannelLike { }); } + 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); @@ -72,6 +85,7 @@ class FakeBroadcastChannel implements BroadcastChannelLike { } FakeBroadcastChannel.channels.clear(); FakeBroadcastChannel.muted = false; + FakeBroadcastChannel.throwOnTypes.clear(); } } @@ -166,6 +180,40 @@ describe("subscribeToCrossTabSse", () => { 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[] = []; @@ -218,7 +266,8 @@ describe("subscribeToCrossTabSse", () => { subscribeForRunEvent(harness.createTab("b", "visible"), []); subscribeForRunEvent(harness.createTab("a", "visible"), []); - await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "a"); + 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 () => { @@ -250,6 +299,46 @@ describe("subscribeToCrossTabSse", () => { 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("same-generation split brain converges to the higher-priority visible leader", async () => { const harness = newHarness(); FakeBroadcastChannel.muted = true; @@ -278,6 +367,29 @@ describe("subscribeToCrossTabSse", () => { 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"), []); @@ -320,6 +432,41 @@ describe("subscribeToCrossTabSse", () => { 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); + }); }); function newHarness() { @@ -329,17 +476,39 @@ function newHarness() { } function subscribeForRunEvent(coordinator: CrossTabSseCoordinator, keys: string[]) { - return subscribeToCrossTabSse({ - coordinator, + return subscribeForEvent(coordinator, { subscriptionKey: "run-feed", - mutate: ((key: string) => { - keys.push(key); - return Promise.resolve(); - }) as MutateFn, + 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"); }, diff --git a/apps/fabro-web/app/lib/cross-tab-sse.ts b/apps/fabro-web/app/lib/cross-tab-sse.ts index 255c4c7ce..0ddb6e220 100644 --- a/apps/fabro-web/app/lib/cross-tab-sse.ts +++ b/apps/fabro-web/app/lib/cross-tab-sse.ts @@ -594,7 +594,7 @@ export class CrossTabSseCoordinator { this.ownCandidate = candidate; this.candidates.set(candidateKey(candidate), candidate); - this.post(candidate); + if (!this.post(candidate)) return; this.candidateTimer = setTimeout(() => { this.completeCandidacy(candidate); }, this.timing.electionJitterMs); @@ -651,7 +651,7 @@ export class CrossTabSseCoordinator { this.handleLeaderEventSourceMessage(message.data); }; - this.post({ + const announced = this.post({ type: "leader-changed", version: MESSAGE_VERSION, tabId: this.tabId, @@ -660,6 +660,7 @@ export class CrossTabSseCoordinator { generation, visibility: this.currentVisibility(), }); + if (!announced) return; this.startHeartbeat(); this.resyncAll(); } @@ -761,14 +762,14 @@ export class CrossTabSseCoordinator { if (this.heartbeatTimer) { clearInterval(this.heartbeatTimer); } - this.sendHeartbeat(); + if (!this.sendHeartbeat()) return; this.heartbeatTimer = setInterval(() => { this.sendHeartbeat(); }, this.timing.heartbeatMs); } - private sendHeartbeat() { - if (!this.isLeader) return; + private sendHeartbeat(): boolean { + if (!this.isLeader) return false; const visibility = this.currentVisibility(); this.leader = { leaderId: this.tabId, @@ -776,7 +777,7 @@ export class CrossTabSseCoordinator { visibility, lastSeen: this.now(), }; - this.post({ + return this.post({ type: "heartbeat", version: MESSAGE_VERSION, tabId: this.tabId, @@ -806,7 +807,7 @@ export class CrossTabSseCoordinator { this.leader = null; if (broadcast) { - this.post({ + const released = this.post({ type: "release", version: MESSAGE_VERSION, tabId: this.tabId, @@ -814,6 +815,7 @@ export class CrossTabSseCoordinator { leaderId: this.tabId, generation, }); + if (!released) return; this.post({ type: "resync", version: MESSAGE_VERSION, @@ -888,7 +890,10 @@ export class CrossTabSseCoordinator { } private post(message: CrossTabSseMessage): boolean { - if (!this.channel) return false; + if (!this.channel) { + this.degradeToFallback(); + return false; + } try { this.channel.postMessage(message); return true; From 65298455545f3b43449e94e7b71e37b93535db9d Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 4 May 2026 15:29:42 -0400 Subject: [PATCH 4/8] fix(web): clean up cross-tab SSE lifecycle Prune stale election candidates as generations advance, reset coordination availability on explicit close, and keep fallback subscribers tracked so coordinator shutdown can clean them up consistently. --- apps/fabro-web/app/lib/cross-tab-sse.test.ts | 137 +++++++++++++++++++ apps/fabro-web/app/lib/cross-tab-sse.ts | 34 +++-- 2 files changed, 161 insertions(+), 10 deletions(-) diff --git a/apps/fabro-web/app/lib/cross-tab-sse.test.ts b/apps/fabro-web/app/lib/cross-tab-sse.test.ts index 573ef6589..f38c881fa 100644 --- a/apps/fabro-web/app/lib/cross-tab-sse.test.ts +++ b/apps/fabro-web/app/lib/cross-tab-sse.test.ts @@ -339,6 +339,32 @@ describe("subscribeToCrossTabSse", () => { 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; @@ -467,6 +493,89 @@ describe("subscribeToCrossTabSse", () => { 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("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() { @@ -516,6 +625,34 @@ function subscribeForEvent( }); } +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, diff --git a/apps/fabro-web/app/lib/cross-tab-sse.ts b/apps/fabro-web/app/lib/cross-tab-sse.ts index 0ddb6e220..97ffeb9f7 100644 --- a/apps/fabro-web/app/lib/cross-tab-sse.ts +++ b/apps/fabro-web/app/lib/cross-tab-sse.ts @@ -231,17 +231,16 @@ export class CrossTabSseCoordinator { this.sourceFactoryLocked = true; } - if (this.coordinationUnavailable) { - return options.fallbackSubscribe(); - } - - if (!this.initialized && !this.initialize()) { - this.coordinationUnavailable = true; - return options.fallbackSubscribe(); - } - const subscription = this.addLocalSubscription(options); - if (this.fallbackMode) { + + 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(); @@ -262,8 +261,10 @@ export class CrossTabSseCoordinator { this.shutdownTimersAndChannel(); this.closeFallbacks(); this.subscriptions.clear(); + this.candidates.clear(); this.leader = null; this.initialized = false; + this.coordinationUnavailable = false; this.fallbackMode = false; this.sourceFactoryLocked = false; } @@ -453,6 +454,7 @@ export class CrossTabSseCoordinator { 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(); } @@ -473,6 +475,7 @@ export class CrossTabSseCoordinator { private handleCandidate(message: CandidateMessage) { this.candidates.set(candidateKey(message), message); + this.pruneStaleCandidates(); if ( this.isLeader && @@ -500,6 +503,7 @@ export class CrossTabSseCoordinator { ) { this.leader = null; this.generation = Math.max(this.generation, message.generation); + this.pruneStaleCandidates(); this.enterCandidacy("release", { leaderId: message.leaderId, generation: message.generation, @@ -546,6 +550,7 @@ export class CrossTabSseCoordinator { if (this.now() - current.lastSeen > this.timing.leaderStaleMs) { this.leader = null; this.generation = Math.max(this.generation, current.generation); + this.pruneStaleCandidates(); this.resyncAll(); this.enterCandidacy("stale-leader", current); return; @@ -638,6 +643,7 @@ export class CrossTabSseCoordinator { this.closeSource(); this.isLeader = true; this.generation = generation; + this.pruneStaleCandidates(); this.leader = { leaderId: this.tabId, generation, @@ -843,6 +849,14 @@ export class CrossTabSseCoordinator { 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 || From e4e51511e08c29a196146a92862f0b1a6457a3ce Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 4 May 2026 15:38:19 -0400 Subject: [PATCH 5/8] refactor(web): simplify cross-tab SSE message parsing and helpers Use unknown.ts helpers in parseMessage, factor out parseLeaderPair/Triple and per-variant parsers to remove repeated typeof guards. Extract leaderIsFresh() for the staleness check used in three places, and make RecentEventCache amortized O(1) by walking expired entries from the oldest instead of scanning the whole map per event. Drop the closeOnTerminal parameter in run-events; the fallback path computes close at its single call site. Co-Authored-By: Claude Opus 4.7 (1M context) --- apps/fabro-web/app/lib/cross-tab-sse.ts | 216 +++++++++++------------- apps/fabro-web/app/lib/run-events.ts | 22 +-- 2 files changed, 104 insertions(+), 134 deletions(-) diff --git a/apps/fabro-web/app/lib/cross-tab-sse.ts b/apps/fabro-web/app/lib/cross-tab-sse.ts index 97ffeb9f7..99281a8f0 100644 --- a/apps/fabro-web/app/lib/cross-tab-sse.ts +++ b/apps/fabro-web/app/lib/cross-tab-sse.ts @@ -6,7 +6,7 @@ import { type EventSourceLike, type MutateFn, } from "./sse"; -import { isRecord } from "./unknown"; +import { getNumber, getString, isRecord, type UnknownRecord } from "./unknown"; export const CROSS_TAB_SSE_CHANNEL = "fabro:sse:v1"; export const HEARTBEAT_MS = 1000; @@ -156,7 +156,7 @@ class RecentEventCache { remember(key: string | undefined, now: number): boolean { if (!key) return true; - this.prune(now); + this.evictExpired(now); if (this.seen.has(key)) return false; this.seen.set(key, now); @@ -168,11 +168,10 @@ class RecentEventCache { return true; } - private prune(now: number) { + private evictExpired(now: number) { for (const [key, seenAt] of this.seen) { - if (now - seenAt > this.ttlMs) { - this.seen.delete(key); - } + if (now - seenAt <= this.ttlMs) return; + this.seen.delete(key); } } } @@ -547,7 +546,7 @@ export class CrossTabSseCoordinator { return; } - if (this.now() - current.lastSeen > this.timing.leaderStaleMs) { + if (!this.leaderIsFresh(current)) { this.leader = null; this.generation = Math.max(this.generation, current.generation); this.pruneStaleCandidates(); @@ -568,7 +567,7 @@ export class CrossTabSseCoordinator { reason === "no-leader" && this.leader && this.leader.visibility === "visible" && - this.now() - this.leader.lastSeen <= this.timing.leaderStaleMs + this.leaderIsFresh(this.leader) ) { return; } @@ -618,10 +617,7 @@ export class CrossTabSseCoordinator { } } - if ( - this.leader && - this.now() - this.leader.lastSeen <= this.timing.leaderStaleMs - ) { + if (this.leader && this.leaderIsFresh(this.leader)) { if (this.leader.generation > candidate.candidateGeneration) { this.clearCandidate(); return; @@ -968,6 +964,10 @@ export class CrossTabSseCoordinator { 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(); @@ -1017,132 +1017,108 @@ function addBrowserPagehideListener(handler: () => void): () => void { function parseMessage(data: unknown): CrossTabSseMessage | undefined { if (!isRecord(data)) return undefined; if (data.version !== MESSAGE_VERSION) return undefined; - const type = data.type; - const tabId = data.tabId; - const sentAt = data.sentAt; - if (typeof type !== "string") return undefined; - if (typeof tabId !== "string") return undefined; - if (typeof sentAt !== "number") return undefined; - const base = { tabId, sentAt }; + 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 baseMessage(base, "hello"); + return { ...base, type: "hello" }; case "heartbeat": { - const { leaderId, generation, visibility } = data; - if (typeof leaderId === "string" && typeof generation === "number" && isVisibility(visibility)) { - return { - ...baseMessage(base, "heartbeat"), - leaderId, - generation, - visibility, - }; - } - return undefined; - } - case "candidate": { - const { - candidateId, - candidateGeneration, - visibility, - observedLeaderId, - observedGeneration, - reason, - } = data; - if ( - typeof candidateId === "string" && - typeof candidateGeneration === "number" && - isVisibility(visibility) && - (typeof observedLeaderId === "string" || observedLeaderId === null) && - typeof observedGeneration === "number" && - isCandidateReason(reason) - ) { - const normalizedObservedLeaderId = - typeof observedLeaderId === "string" ? observedLeaderId : null; - return { - ...baseMessage(base, "candidate"), - candidateId, - candidateGeneration, - visibility, - observedLeaderId: normalizedObservedLeaderId, - observedGeneration, - reason, - }; - } - return undefined; + const triple = parseLeaderTriple(data); + return triple && { ...base, type: "heartbeat", ...triple }; } + case "candidate": + return parseCandidate(data, base); case "leader-changed": { - const { leaderId, generation, visibility } = data; - if (typeof leaderId === "string" && typeof generation === "number" && isVisibility(visibility)) { - return { - ...baseMessage(base, "leader-changed"), - leaderId, - generation, - visibility, - }; - } - return undefined; + const triple = parseLeaderTriple(data); + return triple && { ...base, type: "leader-changed", ...triple }; } case "release": { - const { leaderId, generation } = data; - if (typeof leaderId === "string" && typeof generation === "number") { - return { - ...baseMessage(base, "release"), - leaderId, - generation, - }; - } - return undefined; - } - case "resync": { - const { leaderId, generation, reason } = data; - if ( - (typeof leaderId === "string" || leaderId === null) && - typeof generation === "number" && - isCandidateReason(reason) - ) { - const normalizedLeaderId = typeof leaderId === "string" ? leaderId : null; - return { - ...baseMessage(base, "resync"), - leaderId: normalizedLeaderId, - generation, - reason, - }; - } - return undefined; - } - case "event": { - const { leaderId, generation, payload } = data; - if (typeof leaderId === "string" && typeof generation === "number" && isRecord(payload)) { - return { - ...baseMessage(base, "event"), - leaderId, - generation, - payload, - }; - } - return undefined; + 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; } } -function baseMessage( - data: { - tabId: string; - sentAt: number; - }, - type: TType, -) { +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 { - type, - version: MESSAGE_VERSION, - tabId: data.tabId, - sentAt: data.sentAt, + ...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"; } diff --git a/apps/fabro-web/app/lib/run-events.ts b/apps/fabro-web/app/lib/run-events.ts index 80394da48..f0a5f6a66 100644 --- a/apps/fabro-web/app/lib/run-events.ts +++ b/apps/fabro-web/app/lib/run-events.ts @@ -125,7 +125,7 @@ export function subscribeToRunEvents( resyncKeys: () => resyncKeysForRun(runId), resolveInvalidation: (payload) => { if (payload.run_id !== runId) return { keys: [] }; - return runInvalidation(runId, payload, { closeOnTerminal: false }); + return runInvalidation(runId, payload); }, fallbackSubscribe: () => subscribeToSharedEventSource({ @@ -135,28 +135,22 @@ export function subscribeToRunEvents( mutate, eventSourceFactory, debounceMs, - resolveInvalidation: (payload) => - runInvalidation(runId, payload, { closeOnTerminal: true }), + resolveInvalidation: (payload) => { + const result = runInvalidation(runId, payload); + return { ...result, close: result.immediate }; + }, }), }); } -function runInvalidation( - runId: string, - payload: RunEventPayload, - { closeOnTerminal }: { closeOnTerminal: boolean }, -) { +function runInvalidation(runId: string, payload: RunEventPayload) { const event = payload.event; - if (!event) return { keys: [] }; + if (!event) return { keys: [], immediate: false }; const stageId = stageIdFromPayload(payload); const keys = queryKeysForRunEvent(runId, event, stageId); const terminal = TERMINAL_EVENTS.has(event); - return { - keys, - close: closeOnTerminal && terminal, - immediate: terminal, - }; + return { keys, immediate: terminal }; } function resyncKeysForRun(runId: string) { From 63940fdddc1cce458f7498035e45ea3b2511ba3b Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 4 May 2026 15:52:18 -0400 Subject: [PATCH 6/8] fix(web): recover cross-tab SSE coordination after fallback Reset coordinator state when the last subscriber leaves, clear pending debounce timers on close, and keep coordinated EventSource construction owned by the coordinator while fallback subscriptions keep their local factories. --- apps/fabro-web/app/lib/board-events.test.tsx | 14 +++--- apps/fabro-web/app/lib/board-events.ts | 1 - apps/fabro-web/app/lib/cross-tab-sse.test.ts | 45 ++++++++++++++++++++ apps/fabro-web/app/lib/cross-tab-sse.ts | 41 ++++++++++-------- apps/fabro-web/app/lib/run-events.test.tsx | 16 ++++--- apps/fabro-web/app/lib/run-events.ts | 1 - 6 files changed, 87 insertions(+), 31 deletions(-) diff --git a/apps/fabro-web/app/lib/board-events.test.tsx b/apps/fabro-web/app/lib/board-events.test.tsx index 9b398fdc1..497bb6984 100644 --- a/apps/fabro-web/app/lib/board-events.test.tsx +++ b/apps/fabro-web/app/lib/board-events.test.tsx @@ -9,6 +9,7 @@ import { type BroadcastChannelLike, } from "./cross-tab-sse"; import { queryKeys } from "./query-keys"; +import type { EventSourceLike } from "./sse"; type MessageHandler = ((event: { data: string }) => void) | null; @@ -47,15 +48,17 @@ describe("subscribeToBoardEvents", () => { const source = new FakeEventSource(); const created: string[] = []; const keys: string[] = []; - const coordinator = createCoordinator(); + 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; + 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"); @@ -107,10 +110,11 @@ describe("subscribeToBoardEvents", () => { }); }); -function createCoordinator() { +function createCoordinator(eventSourceFactory: (url: string) => EventSourceLike) { return createCrossTabSseCoordinator({ tabId: "board-test", channelFactory: () => new FakeBroadcastChannel(), + eventSourceFactory, addVisibilityChangeListener: () => () => {}, addPagehideListener: () => () => {}, timing: { diff --git a/apps/fabro-web/app/lib/board-events.ts b/apps/fabro-web/app/lib/board-events.ts index a182d9ad2..dc01765ee 100644 --- a/apps/fabro-web/app/lib/board-events.ts +++ b/apps/fabro-web/app/lib/board-events.ts @@ -56,7 +56,6 @@ export function subscribeToBoardEvents( coordinator, subscriptionKey: BOARD_SUBSCRIPTION_KEY, mutate, - eventSourceFactory, debounceMs, resyncKeys: () => [queryKeys.boards.runs()], resolveInvalidation: boardInvalidation, diff --git a/apps/fabro-web/app/lib/cross-tab-sse.test.ts b/apps/fabro-web/app/lib/cross-tab-sse.test.ts index f38c881fa..76fc0b16e 100644 --- a/apps/fabro-web/app/lib/cross-tab-sse.test.ts +++ b/apps/fabro-web/app/lib/cross-tab-sse.test.ts @@ -542,6 +542,51 @@ describe("subscribeToCrossTabSse", () => { 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"); diff --git a/apps/fabro-web/app/lib/cross-tab-sse.ts b/apps/fabro-web/app/lib/cross-tab-sse.ts index 99281a8f0..6975e6948 100644 --- a/apps/fabro-web/app/lib/cross-tab-sse.ts +++ b/apps/fabro-web/app/lib/cross-tab-sse.ts @@ -113,7 +113,6 @@ interface SubscribeOptions { resolveInvalidation: (payload: TPayload) => EventInvalidation; resyncKeys: () => string[]; fallbackSubscribe: () => () => void; - eventSourceFactory?: (url: string) => EventSourceLike; debounceMs?: number; } @@ -206,7 +205,6 @@ export class CrossTabSseCoordinator { private leaderCheckTimer: ReturnType | null = null; private removeVisibilityListener: (() => void) | null = null; private removePagehideListener: (() => void) | null = null; - private sourceFactoryLocked = false; constructor(options: CrossTabSseCoordinatorOptions = {}) { this.tabId = options.tabId ?? createTabId(); @@ -225,11 +223,6 @@ export class CrossTabSseCoordinator { } subscribe(options: SubscribeOptions): () => void { - if (options.eventSourceFactory && !this.source && !this.sourceFactoryLocked) { - this.sourceFactory = options.eventSourceFactory; - this.sourceFactoryLocked = true; - } - const subscription = this.addLocalSubscription(options); if (this.coordinationUnavailable) { @@ -259,13 +252,10 @@ export class CrossTabSseCoordinator { this.clearNoLeaderTimer(); this.shutdownTimersAndChannel(); this.closeFallbacks(); + this.clearSubscriptionTimers(); this.subscriptions.clear(); this.candidates.clear(); - this.leader = null; - this.initialized = false; - this.coordinationUnavailable = false; - this.fallbackMode = false; - this.sourceFactoryLocked = false; + this.resetIdleState(); } private initialize(): boolean { @@ -356,9 +346,7 @@ export class CrossTabSseCoordinator { subscription.refcount -= 1; if (subscription.refcount <= 0) { - if (subscription.debounceTimer) { - clearTimeout(subscription.debounceTimer); - } + this.clearSubscriptionTimer(subscription); this.subscriptions.delete(subscriptionKey); } @@ -367,9 +355,7 @@ export class CrossTabSseCoordinator { this.clearCandidate(); this.clearNoLeaderTimer(); this.shutdownTimersAndChannel(); - this.initialized = false; - this.fallbackMode = false; - this.sourceFactoryLocked = false; + this.resetIdleState(); } } @@ -945,6 +931,25 @@ export class CrossTabSseCoordinator { } } + 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; diff --git a/apps/fabro-web/app/lib/run-events.test.tsx b/apps/fabro-web/app/lib/run-events.test.tsx index db35a757d..e9faa93fc 100644 --- a/apps/fabro-web/app/lib/run-events.test.tsx +++ b/apps/fabro-web/app/lib/run-events.test.tsx @@ -9,6 +9,7 @@ import { type BroadcastChannelLike, } from "./cross-tab-sse"; import { queryKeys } from "./query-keys"; +import type { EventSourceLike } from "./sse"; type MessageHandler = ((event: { data: string }) => void) | null; @@ -55,7 +56,10 @@ describe("subscribeToRunEvents", () => { const source = new FakeEventSource(); const created: string[] = []; const keys: string[] = []; - const coordinator = createCoordinator(); + const coordinator = createCoordinator((url) => { + created.push(url); + return source; + }); const cleanup = subscribeToRunEvents( "run-coordinated", @@ -63,9 +67,8 @@ describe("subscribeToRunEvents", () => { keys.push(key); return Promise.resolve(); }, - (url) => { - created.push(url); - return source; + () => { + throw new Error("source should be created by coordinator"); }, { debounceMs: 0, coordinator }, ); @@ -86,7 +89,7 @@ describe("subscribeToRunEvents", () => { test("coordinated terminal events invalidate without closing the global stream", async () => { const source = new FakeEventSource(); const keys: string[] = []; - const coordinator = createCoordinator(); + const coordinator = createCoordinator(() => source); const cleanup = subscribeToRunEvents( "run-terminal", (key) => { @@ -206,10 +209,11 @@ describe("subscribeToRunEvents", () => { }); }); -function createCoordinator() { +function createCoordinator(eventSourceFactory: (url: string) => EventSourceLike) { return createCrossTabSseCoordinator({ tabId: "run-test", channelFactory: () => new FakeBroadcastChannel(), + eventSourceFactory, addVisibilityChangeListener: () => () => {}, addPagehideListener: () => () => {}, timing: { diff --git a/apps/fabro-web/app/lib/run-events.ts b/apps/fabro-web/app/lib/run-events.ts index f0a5f6a66..1413d38d4 100644 --- a/apps/fabro-web/app/lib/run-events.ts +++ b/apps/fabro-web/app/lib/run-events.ts @@ -120,7 +120,6 @@ export function subscribeToRunEvents( coordinator, subscriptionKey: `run:${runId}`, mutate, - eventSourceFactory, debounceMs, resyncKeys: () => resyncKeysForRun(runId), resolveInvalidation: (payload) => { From b5b08e78d389a1746aa52490efe5d66f26a07eaa Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 4 May 2026 15:52:24 -0400 Subject: [PATCH 7/8] refactor(api): reuse board column contract across clients Make BoardColumnDefinition.id reference the existing BoardColumn schema and carry that typed contract through generated TypeScript, server responses, demo data, and the runs board UI. --- apps/fabro-web/app/data/runs.ts | 21 ++++++-- apps/fabro-web/app/lib/queries.ts | 9 ++-- apps/fabro-web/app/routes/runs.tsx | 17 ++---- docs/public/api-reference/fabro-api.yaml | 2 +- lib/crates/fabro-server/src/demo/mod.rs | 12 ++--- .../fabro-server/src/server/handler/runs.rs | 54 +++++++++++++------ lib/crates/fabro-server/src/server/tests.rs | 4 +- .../src/models/board-column-definition.ts | 6 ++- 8 files changed, 74 insertions(+), 51 deletions(-) diff --git a/apps/fabro-web/app/data/runs.ts b/apps/fabro-web/app/data/runs.ts index 77d8a4711..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,7 +39,16 @@ export interface RunItem { sourceDirectory?: string; } -export type ColumnStatus = "queued" | "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" }, 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/routes/runs.tsx b/apps/fabro-web/app/routes/runs.tsx index 90ef09405..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"; @@ -50,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"]; } @@ -65,17 +65,8 @@ type Column = { items: RunItem[]; }; -const SKELETON_STATUSES: ColumnStatus[] = [ - "queued", - "initializing", - "running", - "blocked", - "succeeded", - "failed", -]; - function buildSkeletonColumns(): Column[] { - return SKELETON_STATUSES.map((id) => { + return columnStatuses.map((id) => { const colors = columnStatusDisplay[id]; return { id, @@ -100,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, diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index d10db4a47..4137dfaad 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/docs/public/api-reference/fabro-api.yaml @@ -5670,7 +5670,7 @@ components: - name properties: id: - type: string + $ref: "#/components/schemas/BoardColumn" name: type: string diff --git a/lib/crates/fabro-server/src/demo/mod.rs b/lib/crates/fabro-server/src/demo/mod.rs index 787fb1ae4..08423cd55 100644 --- a/lib/crates/fabro-server/src/demo/mod.rs +++ b/lib/crates/fabro-server/src/demo/mod.rs @@ -975,27 +975,27 @@ mod runs { pub(super) fn columns() -> Vec { vec![ BoardColumnDefinition { - id: "queued".into(), + id: BoardColumn::Queued, name: "Queued".into(), }, BoardColumnDefinition { - id: "initializing".into(), + 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(), }, ] diff --git a/lib/crates/fabro-server/src/server/handler/runs.rs b/lib/crates/fabro-server/src/server/handler/runs.rs index eb48c7c4f..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,27 +84,45 @@ impl ListRunsParams { } } -fn board_column(status: RunStatus) -> Option<&'static str> { +fn board_column(status: RunStatus) -> Option { match status { - RunStatus::Submitted | RunStatus::Queued => Some("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": "queued", "name": "Queued"}, - {"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 5480fef0f..74a18fc17 100644 --- a/lib/crates/fabro-server/src/server/tests.rs +++ b/lib/crates/fabro-server/src/server/tests.rs @@ -6615,8 +6615,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; } - From 4b100d350d952e11cc5aff9c03423392408835d2 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 4 May 2026 16:09:12 -0400 Subject: [PATCH 8/8] chore: add plan --- .../2026-05-04-cross-tab-sse-coordination.md | 106 ++++++++++++++++++ 1 file changed, 106 insertions(+) create mode 100644 docs/superpowers/plans/2026-05-04-cross-tab-sse-coordination.md 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.