mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-09-14 23:22:51 +00:00
Merge remote-tracking branch 'origin/main' into fabro/run/01KQT1VDVXGWN9P6MFK4R5E44D
# Conflicts: # apps/fabro-web/app/lib/run-events.test.tsx
This commit is contained in:
Bryan Helmkamp
commit
c0089013af
17 changed files with 2445 additions and 100 deletions
|
|
@ -1,8 +1,10 @@
|
|||
import { formatElapsedSecs, formatDurationSecs } from "../lib/format";
|
||||
import type {
|
||||
RunListItem,
|
||||
RunStatus as ApiRunStatus,
|
||||
RunSummary,
|
||||
import {
|
||||
BoardColumn,
|
||||
type BoardColumn as ApiBoardColumn,
|
||||
type RunListItem,
|
||||
type RunStatus as ApiRunStatus,
|
||||
type RunSummary,
|
||||
} from "@qltysh/fabro-api-client";
|
||||
|
||||
export type CiStatus = "passing" | "failing" | "pending";
|
||||
|
|
@ -37,9 +39,19 @@ export interface RunItem {
|
|||
sourceDirectory?: string;
|
||||
}
|
||||
|
||||
export type ColumnStatus = "initializing" | "running" | "blocked" | "succeeded" | "failed";
|
||||
export type ColumnStatus = ApiBoardColumn;
|
||||
|
||||
export const columnStatuses = [
|
||||
BoardColumn.QUEUED,
|
||||
BoardColumn.INITIALIZING,
|
||||
BoardColumn.RUNNING,
|
||||
BoardColumn.BLOCKED,
|
||||
BoardColumn.SUCCEEDED,
|
||||
BoardColumn.FAILED,
|
||||
] as const satisfies readonly ColumnStatus[];
|
||||
|
||||
export const columnStatusDisplay: Record<ColumnStatus, { label: string; dot: string; text: string }> = {
|
||||
queued: { label: "Queued", dot: "bg-fg-muted", text: "text-fg-muted" },
|
||||
initializing: { label: "Initializing", dot: "bg-amber", text: "text-amber" },
|
||||
running: { label: "Running", dot: "bg-teal-500", text: "text-teal-500" },
|
||||
blocked: { label: "Blocked", dot: "bg-amber", text: "text-amber" },
|
||||
|
|
@ -113,6 +125,7 @@ export function columnForStatus(status: ApiRunStatus | null | undefined): Column
|
|||
switch (status?.kind) {
|
||||
case "submitted":
|
||||
case "queued":
|
||||
return "queued";
|
||||
case "starting":
|
||||
return "initializing";
|
||||
case "running":
|
||||
|
|
|
|||
|
|
@ -4,7 +4,12 @@ import {
|
|||
shouldRefreshBoardForEvent,
|
||||
subscribeToBoardEvents,
|
||||
} from "./board-events";
|
||||
import {
|
||||
createCrossTabSseCoordinator,
|
||||
type BroadcastChannelLike,
|
||||
} from "./cross-tab-sse";
|
||||
import { queryKeys } from "./query-keys";
|
||||
import type { EventSourceLike } from "./sse";
|
||||
|
||||
type MessageHandler = ((event: { data: string }) => void) | null;
|
||||
|
||||
|
|
@ -21,6 +26,14 @@ class FakeEventSource {
|
|||
}
|
||||
}
|
||||
|
||||
class FakeBroadcastChannel implements BroadcastChannelLike {
|
||||
onmessage: ((event: { data: unknown }) => void) | null = null;
|
||||
|
||||
postMessage() {}
|
||||
|
||||
close() {}
|
||||
}
|
||||
|
||||
describe("shouldRefreshBoardForEvent", () => {
|
||||
test("refreshes board for run and interview status changes only", () => {
|
||||
expect(shouldRefreshBoardForEvent("run.running")).toBe(true);
|
||||
|
|
@ -31,22 +44,28 @@ describe("shouldRefreshBoardForEvent", () => {
|
|||
});
|
||||
|
||||
describe("subscribeToBoardEvents", () => {
|
||||
test("shares one source and invalidates the board runs key", () => {
|
||||
test("coordinated mode shares one global source and invalidates the board runs key", async () => {
|
||||
const source = new FakeEventSource();
|
||||
const created: string[] = [];
|
||||
const keys: string[] = [];
|
||||
const coordinator = createCoordinator((url) => {
|
||||
created.push(url);
|
||||
return source;
|
||||
});
|
||||
const mutate = (key: string) => {
|
||||
keys.push(key);
|
||||
return Promise.resolve();
|
||||
};
|
||||
|
||||
const firstCleanup = subscribeToBoardEvents(mutate, (url) => {
|
||||
created.push(url);
|
||||
return source;
|
||||
}, { debounceMs: 0 });
|
||||
const firstCleanup = subscribeToBoardEvents(mutate, () => {
|
||||
throw new Error("source should be created by coordinator");
|
||||
}, { debounceMs: 0, coordinator });
|
||||
const secondCleanup = subscribeToBoardEvents(mutate, () => {
|
||||
throw new Error("source should be reused");
|
||||
}, { debounceMs: 0 });
|
||||
}, { debounceMs: 0, coordinator });
|
||||
|
||||
await waitFor(() => created.length === 1);
|
||||
keys.length = 0;
|
||||
|
||||
source.emit({ event: "run.running" });
|
||||
|
||||
|
|
@ -57,5 +76,68 @@ describe("subscribeToBoardEvents", () => {
|
|||
expect(source.closed).toBe(false);
|
||||
secondCleanup();
|
||||
expect(source.closed).toBe(true);
|
||||
coordinator.close();
|
||||
});
|
||||
|
||||
test("fallback mode preserves the existing shared board EventSource", () => {
|
||||
const source = new FakeEventSource();
|
||||
const created: string[] = [];
|
||||
const keys: string[] = [];
|
||||
const coordinator = createFallbackCoordinator();
|
||||
const mutate = (key: string) => {
|
||||
keys.push(key);
|
||||
return Promise.resolve();
|
||||
};
|
||||
|
||||
const firstCleanup = subscribeToBoardEvents(mutate, (url) => {
|
||||
created.push(url);
|
||||
return source;
|
||||
}, { debounceMs: 0, coordinator });
|
||||
const secondCleanup = subscribeToBoardEvents(mutate, () => {
|
||||
throw new Error("source should be reused");
|
||||
}, { debounceMs: 0, coordinator });
|
||||
|
||||
source.emit({ event: "run.running" });
|
||||
|
||||
expect(created).toEqual(["/api/v1/attach"]);
|
||||
expect(keys).toEqual([queryKeys.boards.runs()]);
|
||||
|
||||
firstCleanup();
|
||||
expect(source.closed).toBe(false);
|
||||
secondCleanup();
|
||||
expect(source.closed).toBe(true);
|
||||
coordinator.close();
|
||||
});
|
||||
});
|
||||
|
||||
function createCoordinator(eventSourceFactory: (url: string) => EventSourceLike) {
|
||||
return createCrossTabSseCoordinator({
|
||||
tabId: "board-test",
|
||||
channelFactory: () => new FakeBroadcastChannel(),
|
||||
eventSourceFactory,
|
||||
addVisibilityChangeListener: () => () => {},
|
||||
addPagehideListener: () => () => {},
|
||||
timing: {
|
||||
heartbeatMs: 10,
|
||||
leaderStaleMs: 50,
|
||||
electionJitterMs: 0,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
function createFallbackCoordinator() {
|
||||
return createCrossTabSseCoordinator({
|
||||
channelFactory: () => {
|
||||
throw new Error("BroadcastChannel unavailable");
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
async function waitFor(condition: () => boolean, timeoutMs = 200) {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
if (condition()) return;
|
||||
await new Promise((resolve) => setTimeout(resolve, 2));
|
||||
}
|
||||
throw new Error("condition did not become true before timeout");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,10 @@
|
|||
import { useEffect } from "react";
|
||||
import { useSWRConfig } from "swr";
|
||||
|
||||
import {
|
||||
subscribeToCrossTabSse,
|
||||
type CrossTabSseCoordinator,
|
||||
} from "./cross-tab-sse";
|
||||
import { queryKeys } from "./query-keys";
|
||||
import {
|
||||
createBrowserEventSource,
|
||||
|
|
@ -11,6 +15,11 @@ import {
|
|||
type SharedEventSubscription,
|
||||
} from "./sse";
|
||||
|
||||
interface BoardEventOptions {
|
||||
debounceMs?: number;
|
||||
coordinator?: CrossTabSseCoordinator;
|
||||
}
|
||||
|
||||
const BOARD_STATUS_EVENTS = new Set([
|
||||
"run.submitted",
|
||||
"run.queued",
|
||||
|
|
@ -41,23 +50,36 @@ export function shouldRefreshBoardForEvent(event: string) {
|
|||
export function subscribeToBoardEvents(
|
||||
mutate: MutateFn,
|
||||
eventSourceFactory: (url: string) => EventSourceLike = createBrowserEventSource,
|
||||
{ debounceMs = 500 }: { debounceMs?: number } = {},
|
||||
{ debounceMs = 500, coordinator }: BoardEventOptions = {},
|
||||
): () => void {
|
||||
return subscribeToSharedEventSource<EventPayload>({
|
||||
subscriptions,
|
||||
return subscribeToCrossTabSse<EventPayload>({
|
||||
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<EventPayload>({
|
||||
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();
|
||||
|
||||
|
|
|
|||
736
apps/fabro-web/app/lib/cross-tab-sse.test.ts
Normal file
736
apps/fabro-web/app/lib/cross-tab-sse.test.ts
Normal file
|
|
@ -0,0 +1,736 @@
|
|||
import { afterEach, describe, expect, test } from "bun:test";
|
||||
|
||||
import {
|
||||
CROSS_TAB_SSE_CHANNEL,
|
||||
createCrossTabSseCoordinator,
|
||||
subscribeToCrossTabSse,
|
||||
type BroadcastChannelLike,
|
||||
type CrossTabSseCoordinator,
|
||||
type CrossTabSseMessage,
|
||||
} from "./cross-tab-sse";
|
||||
import type { EventPayload, MutateFn } from "./sse";
|
||||
|
||||
type MessageHandler = ((event: { data: string }) => void) | null;
|
||||
type TabVisibility = "visible" | "hidden";
|
||||
|
||||
const TEST_TIMING = {
|
||||
heartbeatMs: 10,
|
||||
leaderStaleMs: 35,
|
||||
electionJitterMs: 5,
|
||||
};
|
||||
|
||||
class FakeEventSource {
|
||||
onmessage: MessageHandler = null;
|
||||
closed = false;
|
||||
|
||||
constructor(
|
||||
readonly url: string,
|
||||
readonly owner: string,
|
||||
) {}
|
||||
|
||||
emit(payload: unknown) {
|
||||
this.onmessage?.({ data: JSON.stringify(payload) });
|
||||
}
|
||||
|
||||
close() {
|
||||
this.closed = true;
|
||||
}
|
||||
}
|
||||
|
||||
class FakeBroadcastChannel implements BroadcastChannelLike {
|
||||
static channels = new Set<FakeBroadcastChannel>();
|
||||
static muted = false;
|
||||
static throwOnTypes = new Set<CrossTabSseMessage["type"]>();
|
||||
|
||||
onmessage: ((event: { data: unknown }) => void) | null = null;
|
||||
closed = false;
|
||||
|
||||
constructor(readonly name: string) {
|
||||
FakeBroadcastChannel.channels.add(this);
|
||||
}
|
||||
|
||||
postMessage(message: CrossTabSseMessage) {
|
||||
if (FakeBroadcastChannel.throwOnTypes.has(message.type)) {
|
||||
throw new Error(`postMessage failed for ${message.type}`);
|
||||
}
|
||||
if (FakeBroadcastChannel.muted) return;
|
||||
const recipients = [...FakeBroadcastChannel.channels].filter(
|
||||
(channel) => channel !== this && !channel.closed && channel.name === this.name,
|
||||
);
|
||||
queueMicrotask(() => {
|
||||
for (const channel of recipients) {
|
||||
if (channel.closed) continue;
|
||||
channel.onmessage?.({ data: { ...message } });
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
static broadcastExternal(message: CrossTabSseMessage) {
|
||||
queueMicrotask(() => {
|
||||
for (const channel of FakeBroadcastChannel.channels) {
|
||||
if (channel.closed) continue;
|
||||
channel.onmessage?.({ data: { ...message } });
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
close() {
|
||||
this.closed = true;
|
||||
FakeBroadcastChannel.channels.delete(this);
|
||||
}
|
||||
|
||||
static reset() {
|
||||
for (const channel of FakeBroadcastChannel.channels) {
|
||||
channel.closed = true;
|
||||
}
|
||||
FakeBroadcastChannel.channels.clear();
|
||||
FakeBroadcastChannel.muted = false;
|
||||
FakeBroadcastChannel.throwOnTypes.clear();
|
||||
}
|
||||
}
|
||||
|
||||
class Harness {
|
||||
readonly sources: FakeEventSource[] = [];
|
||||
readonly coordinators = new Map<string, CrossTabSseCoordinator>();
|
||||
readonly visibility = new Map<string, TabVisibility>();
|
||||
readonly visibilityHandlers = new Map<string, () => 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<string, string[]>();
|
||||
|
||||
for (const tabId of ["a", "b", "c"]) {
|
||||
keysByTab.set(tabId, []);
|
||||
subscribeForRunEvent(harness.createTab(tabId), keysByTab.get(tabId)!);
|
||||
}
|
||||
|
||||
await waitFor(() => harness.openSources().length === 1);
|
||||
clearRecordedKeys(keysByTab);
|
||||
|
||||
harness.openSources()[0].emit(runEvent({ id: "evt-1", runId: "run-1", seq: 1 }));
|
||||
await waitFor(() => [...keysByTab.values()].every((keys) => keys.length === 1));
|
||||
|
||||
expect(keysByTab.get("a")).toEqual(["event"]);
|
||||
expect(keysByTab.get("b")).toEqual(["event"]);
|
||||
expect(keysByTab.get("c")).toEqual(["event"]);
|
||||
});
|
||||
|
||||
test("board and run subscriptions coexist on the same global stream", async () => {
|
||||
const harness = newHarness();
|
||||
const coordinator = harness.createTab("a");
|
||||
const boardKeys: string[] = [];
|
||||
const runKeys: string[] = [];
|
||||
|
||||
subscribeForEvent(coordinator, {
|
||||
subscriptionKey: "board",
|
||||
keys: boardKeys,
|
||||
resolveInvalidation: (payload) => ({
|
||||
keys: payload.event === "run.running" ? ["board"] : [],
|
||||
}),
|
||||
resyncKeys: () => ["board-resync"],
|
||||
});
|
||||
subscribeForEvent(coordinator, {
|
||||
subscriptionKey: "run:run-1",
|
||||
keys: runKeys,
|
||||
resolveInvalidation: (payload) => ({
|
||||
keys: payload.event === "run.running" && payload.run_id === "run-1" ? ["run"] : [],
|
||||
}),
|
||||
resyncKeys: () => ["run-resync"],
|
||||
});
|
||||
|
||||
await waitFor(() => harness.openSources().length === 1);
|
||||
boardKeys.length = 0;
|
||||
runKeys.length = 0;
|
||||
|
||||
harness.openSources()[0].emit(runEvent({ id: "evt-coexist", runId: "run-1", seq: 1 }));
|
||||
|
||||
expect(boardKeys).toEqual(["board"]);
|
||||
expect(runKeys).toEqual(["run"]);
|
||||
expect(harness.openSources().map((source) => source.url)).toEqual(["/api/v1/attach"]);
|
||||
});
|
||||
|
||||
test("dedupes duplicate event ids until TTL or max-size eviction", async () => {
|
||||
const harness = newHarness();
|
||||
const keys: string[] = [];
|
||||
subscribeForRunEvent(harness.createTab("a"), keys);
|
||||
|
||||
await waitFor(() => harness.openSources().length === 1);
|
||||
keys.length = 0;
|
||||
|
||||
const source = harness.openSources()[0];
|
||||
source.emit(runEvent({ id: "evt-dup", runId: "run-1", seq: 1 }));
|
||||
source.emit(runEvent({ id: "evt-dup", runId: "run-1", seq: 1 }));
|
||||
|
||||
expect(keys).toEqual(["event"]);
|
||||
|
||||
harness.now += 5 * 60 * 1000 + 1;
|
||||
source.emit(runEvent({ id: "evt-dup", runId: "run-1", seq: 1 }));
|
||||
expect(keys).toEqual(["event", "event"]);
|
||||
|
||||
keys.length = 0;
|
||||
for (let i = 0; i < 1001; i += 1) {
|
||||
source.emit(runEvent({ id: `evt-${i}`, runId: "run-1", seq: i + 2 }));
|
||||
}
|
||||
source.emit(runEvent({ id: "evt-0", runId: "run-1", seq: 2 }));
|
||||
expect(keys).toHaveLength(1002);
|
||||
});
|
||||
|
||||
test("visible followers take over from a fresh hidden leader and resync", async () => {
|
||||
const harness = newHarness();
|
||||
const hiddenKeys: string[] = [];
|
||||
const visibleKeys: string[] = [];
|
||||
|
||||
subscribeForRunEvent(harness.createTab("z", "hidden"), hiddenKeys);
|
||||
await waitFor(() => harness.openSources().length === 1);
|
||||
const hiddenSource = harness.openSources()[0];
|
||||
|
||||
subscribeForRunEvent(harness.createTab("a", "visible"), visibleKeys);
|
||||
|
||||
await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "a");
|
||||
|
||||
expect(hiddenSource.closed).toBe(true);
|
||||
expect(visibleKeys).toContain("resync");
|
||||
});
|
||||
|
||||
test("visible candidates racing for the same hidden leader resolve lexically", async () => {
|
||||
const harness = newHarness();
|
||||
|
||||
subscribeForRunEvent(harness.createTab("z", "hidden"), []);
|
||||
await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "z");
|
||||
|
||||
subscribeForRunEvent(harness.createTab("b", "visible"), []);
|
||||
subscribeForRunEvent(harness.createTab("a", "visible"), []);
|
||||
|
||||
await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner !== "z");
|
||||
expect(harness.openSources().map((source) => source.owner)).toEqual(["a"]);
|
||||
});
|
||||
|
||||
test("a lower lexical follower does not preempt a fresh visible leader", async () => {
|
||||
const harness = newHarness();
|
||||
|
||||
subscribeForRunEvent(harness.createTab("z", "visible"), []);
|
||||
await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "z");
|
||||
|
||||
subscribeForRunEvent(harness.createTab("a", "visible"), []);
|
||||
await sleep(TEST_TIMING.electionJitterMs * 4);
|
||||
|
||||
expect(harness.openSources().map((source) => source.owner)).toEqual(["z"]);
|
||||
});
|
||||
|
||||
test("stale leader detection opens a new leader source and resyncs followers", async () => {
|
||||
const harness = newHarness();
|
||||
const followerKeys: string[] = [];
|
||||
|
||||
subscribeForRunEvent(harness.createTab("a"), []);
|
||||
subscribeForRunEvent(harness.createTab("b"), followerKeys);
|
||||
await waitFor(() => harness.openSources().length === 1);
|
||||
|
||||
const staleLeader = harness.openSources()[0];
|
||||
harness.coordinators.get(staleLeader.owner)?.close();
|
||||
harness.now += TEST_TIMING.leaderStaleMs + TEST_TIMING.heartbeatMs + 1;
|
||||
|
||||
await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner !== staleLeader.owner);
|
||||
|
||||
expect(followerKeys).toContain("resync");
|
||||
});
|
||||
|
||||
test("simultaneous stale leader elections resolve to the lexical winner", async () => {
|
||||
const harness = newHarness();
|
||||
|
||||
subscribeForRunEvent(harness.createTab("z"), []);
|
||||
await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "z");
|
||||
subscribeForRunEvent(harness.createTab("b"), []);
|
||||
subscribeForRunEvent(harness.createTab("a"), []);
|
||||
await sleep(TEST_TIMING.heartbeatMs * 2);
|
||||
|
||||
harness.coordinators.get("z")?.close();
|
||||
harness.now += TEST_TIMING.leaderStaleMs + TEST_TIMING.heartbeatMs + 1;
|
||||
|
||||
await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "a");
|
||||
});
|
||||
|
||||
test("hidden leader ignores candidates for old observed leadership", async () => {
|
||||
const harness = newHarness();
|
||||
|
||||
subscribeForRunEvent(harness.createTab("z", "hidden"), []);
|
||||
await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "z");
|
||||
const hiddenSource = harness.openSources()[0];
|
||||
|
||||
FakeBroadcastChannel.broadcastExternal({
|
||||
type: "candidate",
|
||||
version: 1,
|
||||
tabId: "ghost",
|
||||
sentAt: harness.now,
|
||||
candidateId: "ghost",
|
||||
candidateGeneration: 1,
|
||||
visibility: "visible",
|
||||
observedLeaderId: "z",
|
||||
observedGeneration: 0,
|
||||
reason: "hidden-leader",
|
||||
});
|
||||
await sleep(TEST_TIMING.electionJitterMs * 2);
|
||||
|
||||
expect(hiddenSource.closed).toBe(false);
|
||||
expect(harness.openSources().map((source) => source.owner)).toEqual(["z"]);
|
||||
});
|
||||
|
||||
test("prunes candidate records from older generations", async () => {
|
||||
const harness = newHarness();
|
||||
subscribeForRunEvent(harness.createTab("z", "hidden"), []);
|
||||
await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "z");
|
||||
|
||||
const coordinator = harness.createTab("a", "visible");
|
||||
subscribeForRunEvent(coordinator, []);
|
||||
await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "a");
|
||||
|
||||
FakeBroadcastChannel.broadcastExternal({
|
||||
type: "candidate",
|
||||
version: 1,
|
||||
tabId: "old-candidate",
|
||||
sentAt: harness.now,
|
||||
candidateId: "old-candidate",
|
||||
candidateGeneration: 1,
|
||||
visibility: "visible",
|
||||
observedLeaderId: "previous-leader",
|
||||
observedGeneration: 0,
|
||||
reason: "stale-leader",
|
||||
});
|
||||
await sleep(TEST_TIMING.electionJitterMs * 2);
|
||||
|
||||
expect(candidateGenerations(coordinator)).not.toContain(1);
|
||||
});
|
||||
|
||||
test("same-generation split brain converges to the higher-priority visible leader", async () => {
|
||||
const harness = newHarness();
|
||||
FakeBroadcastChannel.muted = true;
|
||||
|
||||
subscribeForRunEvent(harness.createTab("b"), []);
|
||||
subscribeForRunEvent(harness.createTab("a"), []);
|
||||
await waitFor(() => harness.openSources().length === 2);
|
||||
|
||||
FakeBroadcastChannel.muted = false;
|
||||
await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "a");
|
||||
});
|
||||
|
||||
test("old leader events are ignored after takeover", async () => {
|
||||
const harness = newHarness();
|
||||
const keys: string[] = [];
|
||||
|
||||
subscribeForRunEvent(harness.createTab("z", "hidden"), []);
|
||||
await waitFor(() => harness.openSources().length === 1);
|
||||
const oldSource = harness.openSources()[0];
|
||||
|
||||
subscribeForRunEvent(harness.createTab("a", "visible"), keys);
|
||||
await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "a");
|
||||
keys.length = 0;
|
||||
|
||||
oldSource.emit(runEvent({ id: "evt-old", runId: "run-1", seq: 1 }));
|
||||
expect(keys).toEqual([]);
|
||||
});
|
||||
|
||||
test("old leader heartbeats are ignored after takeover", async () => {
|
||||
const harness = newHarness();
|
||||
|
||||
subscribeForRunEvent(harness.createTab("z", "hidden"), []);
|
||||
await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "z");
|
||||
|
||||
subscribeForRunEvent(harness.createTab("a", "visible"), []);
|
||||
await waitFor(() => harness.openSources().length === 1 && harness.openSources()[0].owner === "a");
|
||||
|
||||
FakeBroadcastChannel.broadcastExternal({
|
||||
type: "heartbeat",
|
||||
version: 1,
|
||||
tabId: "z",
|
||||
sentAt: harness.now,
|
||||
leaderId: "z",
|
||||
generation: 1,
|
||||
visibility: "hidden",
|
||||
});
|
||||
await sleep(TEST_TIMING.heartbeatMs * 2);
|
||||
|
||||
expect(harness.openSources().map((source) => source.owner)).toEqual(["a"]);
|
||||
});
|
||||
|
||||
test("last unsubscribe closes the leader source and releases leadership", async () => {
|
||||
const harness = newHarness();
|
||||
const cleanup = subscribeForRunEvent(harness.createTab("a"), []);
|
||||
|
||||
await waitFor(() => harness.openSources().length === 1);
|
||||
const source = harness.openSources()[0];
|
||||
|
||||
cleanup();
|
||||
|
||||
expect(source.closed).toBe(true);
|
||||
expect(harness.openSources()).toEqual([]);
|
||||
});
|
||||
|
||||
test("missing BroadcastChannel uses subscriber fallback", () => {
|
||||
const coordinator = createCrossTabSseCoordinator({
|
||||
channelFactory: () => {
|
||||
throw new Error("no channel");
|
||||
},
|
||||
});
|
||||
let fallbackStarted = 0;
|
||||
let fallbackStopped = 0;
|
||||
|
||||
const cleanup = subscribeToCrossTabSse<EventPayload>({
|
||||
coordinator,
|
||||
subscriptionKey: "fallback",
|
||||
mutate: (() => Promise.resolve()) as MutateFn,
|
||||
resolveInvalidation: () => ({ keys: [] }),
|
||||
resyncKeys: () => [],
|
||||
fallbackSubscribe: () => {
|
||||
fallbackStarted += 1;
|
||||
return () => {
|
||||
fallbackStopped += 1;
|
||||
};
|
||||
},
|
||||
debounceMs: 0,
|
||||
});
|
||||
|
||||
cleanup();
|
||||
|
||||
expect(fallbackStarted).toBe(1);
|
||||
expect(fallbackStopped).toBe(1);
|
||||
});
|
||||
|
||||
test("postMessage failure after initialization degrades to fallback without coordinated resync", async () => {
|
||||
const harness = newHarness();
|
||||
const coordinator = harness.createTab("a");
|
||||
const keys: string[] = [];
|
||||
let fallbackStarted = 0;
|
||||
let fallbackStopped = 0;
|
||||
|
||||
FakeBroadcastChannel.throwOnTypes.add("leader-changed");
|
||||
const cleanup = subscribeToCrossTabSse<EventPayload>({
|
||||
coordinator,
|
||||
subscriptionKey: "throwing-channel",
|
||||
mutate: ((key: string) => {
|
||||
keys.push(key);
|
||||
return Promise.resolve();
|
||||
}) as MutateFn,
|
||||
resolveInvalidation: () => ({ keys: ["event"] }),
|
||||
resyncKeys: () => ["resync"],
|
||||
fallbackSubscribe: () => {
|
||||
fallbackStarted += 1;
|
||||
return () => {
|
||||
fallbackStopped += 1;
|
||||
};
|
||||
},
|
||||
debounceMs: 0,
|
||||
});
|
||||
|
||||
await waitFor(() => fallbackStarted === 1);
|
||||
|
||||
expect(harness.openSources()).toEqual([]);
|
||||
expect(keys).toEqual([]);
|
||||
|
||||
cleanup();
|
||||
expect(fallbackStopped).toBe(1);
|
||||
});
|
||||
|
||||
test("close resets coordination availability after an initial channel failure", async () => {
|
||||
let channelUnavailable = true;
|
||||
const sources: FakeEventSource[] = [];
|
||||
const coordinator = createCrossTabSseCoordinator({
|
||||
tabId: "recovering",
|
||||
channelFactory: (name) => {
|
||||
if (channelUnavailable) throw new Error("channel unavailable");
|
||||
return new FakeBroadcastChannel(name);
|
||||
},
|
||||
eventSourceFactory: (url) => {
|
||||
const source = new FakeEventSource(url, "recovering");
|
||||
sources.push(source);
|
||||
return source;
|
||||
},
|
||||
addVisibilityChangeListener: () => () => {},
|
||||
addPagehideListener: () => () => {},
|
||||
timing: TEST_TIMING,
|
||||
});
|
||||
let firstFallbackStarted = 0;
|
||||
let secondFallbackStarted = 0;
|
||||
|
||||
const firstCleanup = subscribeWithFallback(coordinator, {
|
||||
fallbackSubscribe: () => {
|
||||
firstFallbackStarted += 1;
|
||||
return () => {};
|
||||
},
|
||||
});
|
||||
firstCleanup();
|
||||
coordinator.close();
|
||||
|
||||
channelUnavailable = false;
|
||||
const secondCleanup = subscribeWithFallback(coordinator, {
|
||||
fallbackSubscribe: () => {
|
||||
secondFallbackStarted += 1;
|
||||
return () => {};
|
||||
},
|
||||
});
|
||||
|
||||
await waitFor(() => sources.some((source) => !source.closed));
|
||||
|
||||
expect(firstFallbackStarted).toBe(1);
|
||||
expect(secondFallbackStarted).toBe(0);
|
||||
expect(sources.filter((source) => !source.closed).map((source) => source.url)).toEqual(["/api/v1/attach"]);
|
||||
|
||||
secondCleanup();
|
||||
coordinator.close();
|
||||
});
|
||||
|
||||
test("last unsubscribe retries coordination after an initial channel failure", async () => {
|
||||
let channelUnavailable = true;
|
||||
const sources: FakeEventSource[] = [];
|
||||
const coordinator = createCrossTabSseCoordinator({
|
||||
tabId: "retry-after-unsubscribe",
|
||||
channelFactory: (name) => {
|
||||
if (channelUnavailable) throw new Error("channel unavailable");
|
||||
return new FakeBroadcastChannel(name);
|
||||
},
|
||||
eventSourceFactory: (url) => {
|
||||
const source = new FakeEventSource(url, "retry-after-unsubscribe");
|
||||
sources.push(source);
|
||||
return source;
|
||||
},
|
||||
addVisibilityChangeListener: () => () => {},
|
||||
addPagehideListener: () => () => {},
|
||||
timing: TEST_TIMING,
|
||||
});
|
||||
let fallbackStarted = 0;
|
||||
|
||||
const firstCleanup = subscribeWithFallback(coordinator, {
|
||||
fallbackSubscribe: () => {
|
||||
fallbackStarted += 1;
|
||||
return () => {};
|
||||
},
|
||||
});
|
||||
firstCleanup();
|
||||
|
||||
channelUnavailable = false;
|
||||
const secondCleanup = subscribeWithFallback(coordinator, {
|
||||
fallbackSubscribe: () => {
|
||||
fallbackStarted += 1;
|
||||
return () => {};
|
||||
},
|
||||
});
|
||||
|
||||
await waitFor(() => sources.some((source) => !source.closed));
|
||||
|
||||
expect(fallbackStarted).toBe(1);
|
||||
expect(sources.filter((source) => !source.closed).map((source) => source.url)).toEqual(["/api/v1/attach"]);
|
||||
|
||||
secondCleanup();
|
||||
coordinator.close();
|
||||
});
|
||||
|
||||
test("close stops fallback subscriptions added after degradation", async () => {
|
||||
const harness = newHarness();
|
||||
const coordinator = harness.createTab("a");
|
||||
let fallbackStarted = 0;
|
||||
let fallbackStopped = 0;
|
||||
|
||||
FakeBroadcastChannel.throwOnTypes.add("leader-changed");
|
||||
subscribeWithFallback(coordinator, {
|
||||
subscriptionKey: "before-degrade",
|
||||
fallbackSubscribe: () => {
|
||||
fallbackStarted += 1;
|
||||
return () => {
|
||||
fallbackStopped += 1;
|
||||
};
|
||||
},
|
||||
});
|
||||
await waitFor(() => fallbackStarted === 1);
|
||||
|
||||
FakeBroadcastChannel.throwOnTypes.clear();
|
||||
subscribeWithFallback(coordinator, {
|
||||
subscriptionKey: "after-degrade",
|
||||
fallbackSubscribe: () => {
|
||||
fallbackStarted += 1;
|
||||
return () => {
|
||||
fallbackStopped += 1;
|
||||
};
|
||||
},
|
||||
});
|
||||
expect(fallbackStarted).toBe(2);
|
||||
|
||||
coordinator.close();
|
||||
|
||||
expect(fallbackStopped).toBe(2);
|
||||
});
|
||||
});
|
||||
|
||||
function newHarness() {
|
||||
const harness = new Harness();
|
||||
harnesses.push(harness);
|
||||
return harness;
|
||||
}
|
||||
|
||||
function subscribeForRunEvent(coordinator: CrossTabSseCoordinator, keys: string[]) {
|
||||
return subscribeForEvent(coordinator, {
|
||||
subscriptionKey: "run-feed",
|
||||
keys,
|
||||
resolveInvalidation: (payload) => ({
|
||||
keys: payload.event === "run.running" ? ["event"] : [],
|
||||
}),
|
||||
resyncKeys: () => ["resync"],
|
||||
});
|
||||
}
|
||||
|
||||
function subscribeForEvent(
|
||||
coordinator: CrossTabSseCoordinator,
|
||||
{
|
||||
subscriptionKey,
|
||||
keys,
|
||||
resolveInvalidation,
|
||||
resyncKeys,
|
||||
}: {
|
||||
subscriptionKey: string;
|
||||
keys: string[];
|
||||
resolveInvalidation: (payload: EventPayload) => { keys: string[] };
|
||||
resyncKeys: () => string[];
|
||||
},
|
||||
) {
|
||||
return subscribeToCrossTabSse<EventPayload>({
|
||||
coordinator,
|
||||
subscriptionKey,
|
||||
mutate: ((key: string) => {
|
||||
keys.push(key);
|
||||
return Promise.resolve();
|
||||
}) as MutateFn,
|
||||
resolveInvalidation,
|
||||
resyncKeys,
|
||||
fallbackSubscribe: () => {
|
||||
throw new Error("fallback should not be used");
|
||||
},
|
||||
debounceMs: 0,
|
||||
});
|
||||
}
|
||||
|
||||
function subscribeWithFallback(
|
||||
coordinator: CrossTabSseCoordinator,
|
||||
{
|
||||
subscriptionKey = "fallback-test",
|
||||
fallbackSubscribe,
|
||||
}: {
|
||||
subscriptionKey?: string;
|
||||
fallbackSubscribe: () => () => void;
|
||||
},
|
||||
) {
|
||||
return subscribeToCrossTabSse<EventPayload>({
|
||||
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<string, { candidateGeneration: number }>;
|
||||
};
|
||||
return [...inspectable.candidates.values()].map((candidate) => candidate.candidateGeneration);
|
||||
}
|
||||
|
||||
function runEvent({
|
||||
id,
|
||||
runId,
|
||||
seq,
|
||||
}: {
|
||||
id: string;
|
||||
runId: string;
|
||||
seq: number;
|
||||
}) {
|
||||
return {
|
||||
id,
|
||||
seq,
|
||||
run_id: runId,
|
||||
event: "run.running",
|
||||
ts: "2026-05-04T12:00:00.000Z",
|
||||
};
|
||||
}
|
||||
|
||||
async function waitFor(condition: () => boolean, timeoutMs = 500) {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
if (condition()) return;
|
||||
await sleep(2);
|
||||
}
|
||||
throw new Error("condition did not become true before timeout");
|
||||
}
|
||||
|
||||
function sleep(ms: number) {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
function clearRecordedKeys(keysByTab: Map<string, string[]>) {
|
||||
for (const keys of keysByTab.values()) {
|
||||
keys.length = 0;
|
||||
}
|
||||
}
|
||||
1175
apps/fabro-web/app/lib/cross-tab-sse.ts
Normal file
1175
apps/fabro-web/app/lib/cross-tab-sse.ts
Normal file
File diff suppressed because it is too large
Load diff
|
|
@ -34,6 +34,9 @@ const immutableOptions: SWRConfiguration = {
|
|||
revalidateOnReconnect: false,
|
||||
};
|
||||
|
||||
type BoardRunsEnvelope = PaginatedEnvelope<PaginatedBoardRunList["data"][number]> &
|
||||
Pick<PaginatedBoardRunList, "columns">;
|
||||
|
||||
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<PaginatedBoardRunList["data"][number]> & {
|
||||
columns: { id: string; name: string }[];
|
||||
}
|
||||
>(queryKeys.boards.runs(), apiPaginatedFetcher);
|
||||
return useSWR<BoardRunsEnvelope>(queryKeys.boards.runs(), apiPaginatedFetcher);
|
||||
}
|
||||
|
||||
export function useRun(id: string | undefined) {
|
||||
|
|
|
|||
|
|
@ -4,7 +4,12 @@ import {
|
|||
queryKeysForRunEvent,
|
||||
subscribeToRunEvents,
|
||||
} from "./run-events";
|
||||
import {
|
||||
createCrossTabSseCoordinator,
|
||||
type BroadcastChannelLike,
|
||||
} from "./cross-tab-sse";
|
||||
import { queryKeys } from "./query-keys";
|
||||
import type { EventSourceLike } from "./sse";
|
||||
|
||||
type MessageHandler = ((event: { data: string }) => void) | null;
|
||||
|
||||
|
|
@ -25,6 +30,14 @@ class FakeEventSource {
|
|||
}
|
||||
}
|
||||
|
||||
class FakeBroadcastChannel implements BroadcastChannelLike {
|
||||
onmessage: ((event: { data: unknown }) => void) | null = null;
|
||||
|
||||
postMessage() {}
|
||||
|
||||
close() {}
|
||||
}
|
||||
|
||||
describe("queryKeysForRunEvent", () => {
|
||||
test("terminal events invalidate run-scoped resources", () => {
|
||||
expect(queryKeysForRunEvent("run-1", "run.completed")).toEqual([
|
||||
|
|
@ -47,10 +60,76 @@ describe("queryKeysForRunEvent", () => {
|
|||
});
|
||||
|
||||
describe("subscribeToRunEvents", () => {
|
||||
test("refcounts shared sources and keeps mutators active until final unsubscribe", () => {
|
||||
test("coordinated mode uses the global attach stream and filters by run_id", async () => {
|
||||
const source = new FakeEventSource();
|
||||
const created: string[] = [];
|
||||
const keys: string[] = [];
|
||||
const coordinator = createCoordinator((url) => {
|
||||
created.push(url);
|
||||
return source;
|
||||
});
|
||||
|
||||
const cleanup = subscribeToRunEvents(
|
||||
"run-coordinated",
|
||||
(key) => {
|
||||
keys.push(key);
|
||||
return Promise.resolve();
|
||||
},
|
||||
() => {
|
||||
throw new Error("source should be created by coordinator");
|
||||
},
|
||||
{ debounceMs: 0, coordinator },
|
||||
);
|
||||
|
||||
await waitFor(() => created.length === 1);
|
||||
keys.length = 0;
|
||||
|
||||
source.emit({ event: "checkpoint.completed", run_id: "other-run" });
|
||||
source.emit({ event: "checkpoint.completed", run_id: "run-coordinated" });
|
||||
|
||||
expect(created).toEqual(["/api/v1/attach"]);
|
||||
expect(keys).toEqual([queryKeys.runs.files("run-coordinated")]);
|
||||
|
||||
cleanup();
|
||||
coordinator.close();
|
||||
});
|
||||
|
||||
test("coordinated terminal events invalidate without closing the global stream", async () => {
|
||||
const source = new FakeEventSource();
|
||||
const keys: string[] = [];
|
||||
const coordinator = createCoordinator(() => source);
|
||||
const cleanup = subscribeToRunEvents(
|
||||
"run-terminal",
|
||||
(key) => {
|
||||
keys.push(key);
|
||||
return Promise.resolve();
|
||||
},
|
||||
() => source,
|
||||
{ debounceMs: 0, coordinator },
|
||||
);
|
||||
|
||||
await waitFor(() => source.onmessage !== null);
|
||||
keys.length = 0;
|
||||
|
||||
source.emit({ event: "run.failed", run_id: "run-terminal" });
|
||||
expect(source.closed).toBe(false);
|
||||
expect(keys).toContain(queryKeys.runs.files("run-terminal"));
|
||||
expect(keys).toContain(queryKeys.runs.billing("run-terminal"));
|
||||
|
||||
keys.length = 0;
|
||||
source.emit({ event: "run.archived", run_id: "run-terminal" });
|
||||
expect(source.closed).toBe(false);
|
||||
expect(keys).toEqual([queryKeys.runs.detail("run-terminal")]);
|
||||
|
||||
cleanup();
|
||||
coordinator.close();
|
||||
});
|
||||
|
||||
test("fallback refcounts run-scoped sources and keeps mutators active until final unsubscribe", () => {
|
||||
const source = new FakeEventSource();
|
||||
const created: string[] = [];
|
||||
const keys: string[] = [];
|
||||
const coordinator = createFallbackCoordinator();
|
||||
const mutate = (key: string) => {
|
||||
keys.push(key);
|
||||
return Promise.resolve();
|
||||
|
|
@ -59,10 +138,10 @@ describe("subscribeToRunEvents", () => {
|
|||
const firstCleanup = subscribeToRunEvents("run-refcount", mutate, (url) => {
|
||||
created.push(url);
|
||||
return source;
|
||||
}, { debounceMs: 0 });
|
||||
}, { debounceMs: 0, coordinator });
|
||||
const secondCleanup = subscribeToRunEvents("run-refcount", mutate, () => {
|
||||
throw new Error("source should be reused");
|
||||
}, { debounceMs: 0 });
|
||||
}, { debounceMs: 0, coordinator });
|
||||
|
||||
expect(created).toEqual(["/api/v1/runs/run-refcount/attach"]);
|
||||
|
||||
|
|
@ -74,11 +153,13 @@ describe("subscribeToRunEvents", () => {
|
|||
|
||||
secondCleanup();
|
||||
expect(source.closed).toBe(true);
|
||||
coordinator.close();
|
||||
});
|
||||
|
||||
test("terminal events close the source after invalidating keys", () => {
|
||||
test("fallback terminal events close the source after invalidating keys", () => {
|
||||
const source = new FakeEventSource();
|
||||
const keys: string[] = [];
|
||||
const coordinator = createFallbackCoordinator();
|
||||
const cleanup = subscribeToRunEvents(
|
||||
"run-terminal",
|
||||
(key) => {
|
||||
|
|
@ -86,7 +167,7 @@ describe("subscribeToRunEvents", () => {
|
|||
return Promise.resolve();
|
||||
},
|
||||
() => source,
|
||||
{ debounceMs: 0 },
|
||||
{ debounceMs: 0, coordinator },
|
||||
);
|
||||
|
||||
source.emit({ event: "run.failed" });
|
||||
|
|
@ -96,11 +177,13 @@ describe("subscribeToRunEvents", () => {
|
|||
expect(keys).toContain(queryKeys.runs.billing("run-terminal"));
|
||||
|
||||
cleanup();
|
||||
coordinator.close();
|
||||
});
|
||||
|
||||
test("envelope with suffixed stage_id invalidates stageTurns(runId, stageId)", () => {
|
||||
test("envelope with suffixed stage_id invalidates stageTurns(runId, stageId)", async () => {
|
||||
const source = new FakeEventSource();
|
||||
const keys: string[] = [];
|
||||
const coordinator = createCoordinator(() => source);
|
||||
const cleanup = subscribeToRunEvents(
|
||||
"run-stage",
|
||||
(key) => {
|
||||
|
|
@ -108,10 +191,16 @@ describe("subscribeToRunEvents", () => {
|
|||
return Promise.resolve();
|
||||
},
|
||||
() => source,
|
||||
{ debounceMs: 0 },
|
||||
{ debounceMs: 0, coordinator },
|
||||
);
|
||||
|
||||
source.emit({ event: "stage.retrying", stage_id: "verify@2", node_id: "verify" });
|
||||
await waitFor(() => source.onmessage !== null);
|
||||
source.emit({
|
||||
event: "stage.retrying",
|
||||
run_id: "run-stage",
|
||||
stage_id: "verify@2",
|
||||
node_id: "verify",
|
||||
});
|
||||
|
||||
expect(keys).toContain(queryKeys.runs.stageTurns("run-stage", "verify@2"));
|
||||
expect(keys).toContain(queryKeys.runs.stages("run-stage"));
|
||||
|
|
@ -121,11 +210,13 @@ describe("subscribeToRunEvents", () => {
|
|||
expect(keys).not.toContain(queryKeys.runs.stageTurns("run-stage", "verify"));
|
||||
|
||||
cleanup();
|
||||
coordinator.close();
|
||||
});
|
||||
|
||||
test("falls back to node_id when an event has no stage_id", () => {
|
||||
test("falls back to node_id when an event has no stage_id", async () => {
|
||||
const source = new FakeEventSource();
|
||||
const keys: string[] = [];
|
||||
const coordinator = createCoordinator(() => source);
|
||||
const cleanup = subscribeToRunEvents(
|
||||
"run-stage-node",
|
||||
(key) => {
|
||||
|
|
@ -133,22 +224,25 @@ describe("subscribeToRunEvents", () => {
|
|||
return Promise.resolve();
|
||||
},
|
||||
() => source,
|
||||
{ debounceMs: 0 },
|
||||
{ debounceMs: 0, coordinator },
|
||||
);
|
||||
|
||||
source.emit({ event: "stage.started", node_id: "verify" });
|
||||
await waitFor(() => source.onmessage !== null);
|
||||
source.emit({ event: "stage.started", run_id: "run-stage-node", node_id: "verify" });
|
||||
|
||||
expect(keys).toContain(queryKeys.runs.stageTurns("run-stage-node", "verify"));
|
||||
expect(keys).toContain(queryKeys.runs.stages("run-stage-node"));
|
||||
|
||||
cleanup();
|
||||
coordinator.close();
|
||||
});
|
||||
|
||||
test("malformed events are ignored and StrictMode-style cleanup does not underflow", () => {
|
||||
test("fallback malformed events are ignored and StrictMode-style cleanup does not underflow", () => {
|
||||
const firstSource = new FakeEventSource();
|
||||
const secondSource = new FakeEventSource();
|
||||
const sources = [firstSource, secondSource];
|
||||
const keys: string[] = [];
|
||||
const coordinator = createFallbackCoordinator();
|
||||
|
||||
const firstCleanup = subscribeToRunEvents(
|
||||
"run-strict",
|
||||
|
|
@ -157,7 +251,7 @@ describe("subscribeToRunEvents", () => {
|
|||
return Promise.resolve();
|
||||
},
|
||||
() => sources.shift()!,
|
||||
{ debounceMs: 0 },
|
||||
{ debounceMs: 0, coordinator },
|
||||
);
|
||||
firstSource.emitRaw("{broken");
|
||||
firstCleanup();
|
||||
|
|
@ -169,12 +263,45 @@ describe("subscribeToRunEvents", () => {
|
|||
return Promise.resolve();
|
||||
},
|
||||
() => sources.shift()!,
|
||||
{ debounceMs: 0 },
|
||||
{ debounceMs: 0, coordinator },
|
||||
);
|
||||
secondCleanup();
|
||||
|
||||
expect(keys).toEqual([]);
|
||||
expect(firstSource.closed).toBe(true);
|
||||
expect(secondSource.closed).toBe(true);
|
||||
coordinator.close();
|
||||
});
|
||||
});
|
||||
|
||||
function createCoordinator(eventSourceFactory: (url: string) => EventSourceLike) {
|
||||
return createCrossTabSseCoordinator({
|
||||
tabId: "run-test",
|
||||
channelFactory: () => new FakeBroadcastChannel(),
|
||||
eventSourceFactory,
|
||||
addVisibilityChangeListener: () => () => {},
|
||||
addPagehideListener: () => () => {},
|
||||
timing: {
|
||||
heartbeatMs: 10,
|
||||
leaderStaleMs: 50,
|
||||
electionJitterMs: 0,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
function createFallbackCoordinator() {
|
||||
return createCrossTabSseCoordinator({
|
||||
channelFactory: () => {
|
||||
throw new Error("BroadcastChannel unavailable");
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
async function waitFor(condition: () => boolean, timeoutMs = 200) {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
if (condition()) return;
|
||||
await new Promise((resolve) => setTimeout(resolve, 2));
|
||||
}
|
||||
throw new Error("condition did not become true before timeout");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,10 @@
|
|||
import { useEffect } from "react";
|
||||
import { useSWRConfig } from "swr";
|
||||
|
||||
import {
|
||||
subscribeToCrossTabSse,
|
||||
type CrossTabSseCoordinator,
|
||||
} from "./cross-tab-sse";
|
||||
import { queryKeys } from "./query-keys";
|
||||
import {
|
||||
createBrowserEventSource,
|
||||
|
|
@ -13,11 +17,17 @@ import {
|
|||
|
||||
interface RunEventPayload extends EventPayload {
|
||||
event?: string;
|
||||
run_id?: string;
|
||||
node_id?: string;
|
||||
stage_id?: string;
|
||||
properties?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
interface RunEventOptions {
|
||||
debounceMs?: number;
|
||||
coordinator?: CrossTabSseCoordinator;
|
||||
}
|
||||
|
||||
const subscriptions = new Map<string, SharedEventSubscription>();
|
||||
|
||||
const TERMINAL_EVENTS = new Set(["run.completed", "run.failed"]);
|
||||
|
|
@ -110,31 +120,57 @@ export function subscribeToRunEvents(
|
|||
runId: string,
|
||||
mutate: MutateFn,
|
||||
eventSourceFactory: (url: string) => EventSourceLike = createBrowserEventSource,
|
||||
{ debounceMs = 300 }: { debounceMs?: number } = {},
|
||||
{ debounceMs = 300, coordinator }: RunEventOptions = {},
|
||||
): () => void {
|
||||
return subscribeToSharedEventSource<RunEventPayload>({
|
||||
subscriptions,
|
||||
subscriptionKey: runId,
|
||||
url: queryKeys.runs.attach(runId),
|
||||
return subscribeToCrossTabSse<RunEventPayload>({
|
||||
coordinator,
|
||||
subscriptionKey: `run:${runId}`,
|
||||
mutate,
|
||||
eventSourceFactory,
|
||||
debounceMs,
|
||||
resyncKeys: () => resyncKeysForRun(runId),
|
||||
resolveInvalidation: (payload) => {
|
||||
const event = payload.event;
|
||||
if (!event) return { keys: [] };
|
||||
|
||||
const stageId = stageIdFromPayload(payload);
|
||||
const keys = queryKeysForRunEvent(runId, event, stageId);
|
||||
const terminal = TERMINAL_EVENTS.has(event);
|
||||
return {
|
||||
keys,
|
||||
close: terminal,
|
||||
immediate: terminal,
|
||||
};
|
||||
if (payload.run_id !== runId) return { keys: [] };
|
||||
return runInvalidation(runId, payload);
|
||||
},
|
||||
fallbackSubscribe: () =>
|
||||
subscribeToSharedEventSource<RunEventPayload>({
|
||||
subscriptions,
|
||||
subscriptionKey: runId,
|
||||
url: queryKeys.runs.attach(runId),
|
||||
mutate,
|
||||
eventSourceFactory,
|
||||
debounceMs,
|
||||
resolveInvalidation: (payload) => {
|
||||
const result = runInvalidation(runId, payload);
|
||||
return { ...result, close: result.immediate };
|
||||
},
|
||||
}),
|
||||
});
|
||||
}
|
||||
|
||||
function runInvalidation(runId: string, payload: RunEventPayload) {
|
||||
const event = payload.event;
|
||||
if (!event) return { keys: [], immediate: false };
|
||||
|
||||
const stageId = stageIdFromPayload(payload);
|
||||
const keys = queryKeysForRunEvent(runId, event, stageId);
|
||||
const terminal = TERMINAL_EVENTS.has(event);
|
||||
return { keys, immediate: terminal };
|
||||
}
|
||||
|
||||
function resyncKeysForRun(runId: string) {
|
||||
return [
|
||||
queryKeys.runs.detail(runId),
|
||||
queryKeys.runs.files(runId),
|
||||
queryKeys.runs.billing(runId),
|
||||
queryKeys.runs.stages(runId),
|
||||
queryKeys.runs.events(runId, 1000),
|
||||
queryKeys.runs.graph(runId, "LR"),
|
||||
queryKeys.runs.graph(runId, "TB"),
|
||||
queryKeys.runs.questions(runId, 25, 0),
|
||||
];
|
||||
}
|
||||
|
||||
function stageIdFromPayload(payload: RunEventPayload): string | undefined {
|
||||
if (typeof payload.stage_id === "string") return payload.stage_id;
|
||||
if (typeof payload.node_id === "string") return payload.node_id;
|
||||
|
|
|
|||
|
|
@ -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" },
|
||||
|
|
|
|||
|
|
@ -18,7 +18,7 @@ import {
|
|||
arrayMove,
|
||||
} from "@dnd-kit/sortable";
|
||||
import { CSS } from "@dnd-kit/utilities";
|
||||
import { ciConfig, columnStatusDisplay, deriveCiStatus, mapRunListItem } from "../data/runs";
|
||||
import { ciConfig, columnStatusDisplay, columnStatuses, deriveCiStatus, mapRunListItem } from "../data/runs";
|
||||
import type { CiStatus, CheckRun, CheckStatus, RunItem, RunWithStatus, ColumnStatus } from "../data/runs";
|
||||
import { EmptyState } from "../components/state";
|
||||
import { shouldRefreshBoardForEvent, useBoardEvents } from "../lib/board-events";
|
||||
|
|
@ -38,6 +38,7 @@ interface ColumnStyle {
|
|||
}
|
||||
|
||||
const columnStyles: Record<ColumnStatus, ColumnStyle> = {
|
||||
queued: { iconType: "branch", actions: [] },
|
||||
initializing: { iconType: "branch", actions: [] },
|
||||
running: { iconType: "branch", actions: ["Watch", "Steer"] },
|
||||
blocked: { iconType: "branch", actions: ["Answer Question"] },
|
||||
|
|
@ -49,7 +50,7 @@ const defaultColumnStyle: ColumnStyle = { iconType: "branch", actions: [] };
|
|||
const defaultColumnColors = { dot: "bg-fg-muted", text: "text-fg-muted" };
|
||||
|
||||
interface BoardRunsResponse {
|
||||
columns: { id: string; name: string }[];
|
||||
columns: PaginatedBoardRunList["columns"];
|
||||
data: PaginatedBoardRunList["data"];
|
||||
meta: PaginatedBoardRunList["meta"];
|
||||
}
|
||||
|
|
@ -64,16 +65,8 @@ type Column = {
|
|||
items: RunItem[];
|
||||
};
|
||||
|
||||
const SKELETON_STATUSES: ColumnStatus[] = [
|
||||
"initializing",
|
||||
"running",
|
||||
"blocked",
|
||||
"succeeded",
|
||||
"failed",
|
||||
];
|
||||
|
||||
function buildSkeletonColumns(): Column[] {
|
||||
return SKELETON_STATUSES.map((id) => {
|
||||
return columnStatuses.map((id) => {
|
||||
const colors = columnStatusDisplay[id];
|
||||
return {
|
||||
id,
|
||||
|
|
@ -98,7 +91,7 @@ export function buildBoardColumns(response: BoardRunsResponse): Column[] {
|
|||
}
|
||||
|
||||
return response.columns.map((col) => {
|
||||
const id = col.id as ColumnStatus;
|
||||
const id = col.id;
|
||||
const colors = columnStatusDisplay[id] ?? defaultColumnColors;
|
||||
return {
|
||||
id,
|
||||
|
|
@ -754,6 +747,9 @@ export default function Runs() {
|
|||
(sum, col) => sum + col.items.length,
|
||||
0,
|
||||
);
|
||||
const visibleColumns = filteredColumns.filter(
|
||||
(col) => col.id !== "queued" || col.items.length > 0,
|
||||
);
|
||||
|
||||
return (
|
||||
<DndContext sensors={sensors} collisionDetection={closestCenter} onDragEnd={handleDragEnd}>
|
||||
|
|
@ -815,7 +811,7 @@ export default function Runs() {
|
|||
{view === "columns" ? (
|
||||
<>
|
||||
<div className="flex gap-5 overflow-x-auto pb-4">
|
||||
{filteredColumns.map((col) => (
|
||||
{visibleColumns.map((col) => (
|
||||
<div key={col.id} className="w-72 shrink-0">
|
||||
<BoardColumn column={col} />
|
||||
</div>
|
||||
|
|
@ -838,7 +834,7 @@ export default function Runs() {
|
|||
) : (
|
||||
<>
|
||||
<div className="space-y-4">
|
||||
{filteredColumns.map((col) => {
|
||||
{visibleColumns.map((col) => {
|
||||
const isCollapsed = collapsed.has(col.id);
|
||||
return (
|
||||
<div key={col.id}>
|
||||
|
|
|
|||
|
|
@ -5656,6 +5656,7 @@ components:
|
|||
description: Board column status for a run in the list view.
|
||||
type: string
|
||||
enum:
|
||||
- queued
|
||||
- initializing
|
||||
- running
|
||||
- blocked
|
||||
|
|
@ -5669,7 +5670,7 @@ components:
|
|||
- name
|
||||
properties:
|
||||
id:
|
||||
type: string
|
||||
$ref: "#/components/schemas/BoardColumn"
|
||||
name:
|
||||
type: string
|
||||
|
||||
|
|
|
|||
106
docs/superpowers/plans/2026-05-04-cross-tab-sse-coordination.md
Normal file
106
docs/superpowers/plans/2026-05-04-cross-tab-sse-coordination.md
Normal file
|
|
@ -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.
|
||||
|
|
@ -795,8 +795,8 @@ mod runs {
|
|||
.collect()
|
||||
}
|
||||
|
||||
fn demo_run_ids() -> &'static [RunId; 6] {
|
||||
static IDS: OnceLock<[RunId; 6]> = OnceLock::new();
|
||||
fn demo_run_ids() -> &'static [RunId; 7] {
|
||||
static IDS: OnceLock<[RunId; 7]> = OnceLock::new();
|
||||
IDS.get_or_init(|| {
|
||||
[
|
||||
RunId::with_timestamp(ts("2026-03-06T14:30:00Z"), 1),
|
||||
|
|
@ -805,6 +805,7 @@ mod runs {
|
|||
RunId::with_timestamp(ts("2026-03-04T10:00:00Z"), 4),
|
||||
RunId::with_timestamp(ts("2026-03-03T16:45:00Z"), 5),
|
||||
RunId::with_timestamp(ts("2026-02-28T14:00:00Z"), 6),
|
||||
RunId::with_timestamp(ts("2026-03-06T14:35:00Z"), 7),
|
||||
]
|
||||
})
|
||||
}
|
||||
|
|
@ -974,23 +975,27 @@ mod runs {
|
|||
pub(super) fn columns() -> Vec<BoardColumnDefinition> {
|
||||
vec![
|
||||
BoardColumnDefinition {
|
||||
id: "initializing".into(),
|
||||
id: BoardColumn::Queued,
|
||||
name: "Queued".into(),
|
||||
},
|
||||
BoardColumnDefinition {
|
||||
id: BoardColumn::Initializing,
|
||||
name: "Initializing".into(),
|
||||
},
|
||||
BoardColumnDefinition {
|
||||
id: "running".into(),
|
||||
id: BoardColumn::Running,
|
||||
name: "Running".into(),
|
||||
},
|
||||
BoardColumnDefinition {
|
||||
id: "blocked".into(),
|
||||
id: BoardColumn::Blocked,
|
||||
name: "Blocked".into(),
|
||||
},
|
||||
BoardColumnDefinition {
|
||||
id: "succeeded".into(),
|
||||
id: BoardColumn::Succeeded,
|
||||
name: "Succeeded".into(),
|
||||
},
|
||||
BoardColumnDefinition {
|
||||
id: "failed".into(),
|
||||
id: BoardColumn::Failed,
|
||||
name: "Failed".into(),
|
||||
},
|
||||
]
|
||||
|
|
@ -1082,6 +1087,20 @@ mod runs {
|
|||
Some(720000),
|
||||
&[("release", "preview")],
|
||||
),
|
||||
summary(
|
||||
7,
|
||||
"api-server",
|
||||
"implement",
|
||||
"Implement",
|
||||
"Add audit log retention policy",
|
||||
"queued",
|
||||
"2026-03-06T14:35:00Z",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
&[("owner", "platform")],
|
||||
),
|
||||
]
|
||||
}
|
||||
|
||||
|
|
@ -1150,6 +1169,13 @@ mod runs {
|
|||
None,
|
||||
None,
|
||||
),
|
||||
board_item(
|
||||
take_summary(&mut summaries, demo_run_id(7)),
|
||||
BoardColumn::Queued,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
),
|
||||
]
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -10,7 +10,9 @@ use axum::{Json, Router};
|
|||
use base64::Engine as _;
|
||||
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
|
||||
use bytes::Bytes;
|
||||
use fabro_api::types::{RunManifest, RunStatusResponse, SubmitAnswerRequest};
|
||||
use fabro_api::types::{
|
||||
BoardColumn, BoardColumnDefinition, RunManifest, RunStatusResponse, SubmitAnswerRequest,
|
||||
};
|
||||
use fabro_config::Storage;
|
||||
use fabro_interview::AnswerSubmission;
|
||||
use fabro_types::{
|
||||
|
|
@ -82,25 +84,45 @@ impl ListRunsParams {
|
|||
}
|
||||
}
|
||||
|
||||
fn board_column(status: RunStatus) -> Option<&'static str> {
|
||||
fn board_column(status: RunStatus) -> Option<BoardColumn> {
|
||||
match status {
|
||||
RunStatus::Submitted | RunStatus::Queued | RunStatus::Starting => Some("initializing"),
|
||||
RunStatus::Running | RunStatus::Paused { .. } => Some("running"),
|
||||
RunStatus::Blocked { .. } => Some("blocked"),
|
||||
RunStatus::Succeeded { .. } => Some("succeeded"),
|
||||
RunStatus::Failed { .. } | RunStatus::Dead => Some("failed"),
|
||||
RunStatus::Submitted | RunStatus::Queued => Some(BoardColumn::Queued),
|
||||
RunStatus::Starting => Some(BoardColumn::Initializing),
|
||||
RunStatus::Running | RunStatus::Paused { .. } => Some(BoardColumn::Running),
|
||||
RunStatus::Blocked { .. } => Some(BoardColumn::Blocked),
|
||||
RunStatus::Succeeded { .. } => Some(BoardColumn::Succeeded),
|
||||
RunStatus::Failed { .. } | RunStatus::Dead => Some(BoardColumn::Failed),
|
||||
RunStatus::Removing | RunStatus::Archived { .. } => None,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn board_columns() -> serde_json::Value {
|
||||
serde_json::json!([
|
||||
{"id": "initializing", "name": "Initializing"},
|
||||
{"id": "running", "name": "Running"},
|
||||
{"id": "blocked", "name": "Blocked"},
|
||||
{"id": "succeeded", "name": "Succeeded"},
|
||||
{"id": "failed", "name": "Failed"},
|
||||
])
|
||||
pub(crate) fn board_columns() -> Vec<BoardColumnDefinition> {
|
||||
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(
|
||||
|
|
|
|||
|
|
@ -6572,7 +6572,7 @@ async fn pause_run_sets_pending_control_on_board_response() {
|
|||
assert_eq!(body["pending_control"].as_str(), Some("pause"));
|
||||
|
||||
// Verify the run appears on the board (store has Submitted status →
|
||||
// "initializing" column)
|
||||
// "queued" column)
|
||||
let req = Request::builder()
|
||||
.method("GET")
|
||||
.uri(api("/boards/runs"))
|
||||
|
|
@ -6587,7 +6587,7 @@ async fn pause_run_sets_pending_control_on_board_response() {
|
|||
.find(|item| item["run_id"].as_str() == Some(run_id_str.as_str()))
|
||||
.expect("board item should exist");
|
||||
assert!(item["status"].is_object());
|
||||
assert_eq!(item["column"].as_str(), Some("initializing"));
|
||||
assert_eq!(item["column"].as_str(), Some("queued"));
|
||||
assert_eq!(item["pending_control"].as_str(), Some("pause"));
|
||||
}
|
||||
|
||||
|
|
@ -7062,8 +7062,8 @@ async fn queue_position_reported_for_queued_runs() {
|
|||
let first_run_id = create_and_start_run(&app, MINIMAL_DOT).await;
|
||||
let second_run_id = create_and_start_run(&app, MINIMAL_DOT).await;
|
||||
|
||||
// Queued runs are excluded from the board, so verify queue positions
|
||||
// via the in-memory state directly.
|
||||
// Queue position is tracked in memory even when queued runs are also
|
||||
// visible on the board.
|
||||
let runs = state.runs.lock().expect("runs lock poisoned");
|
||||
let positions = compute_queue_positions(&runs);
|
||||
let first_id = first_run_id.parse::<RunId>().unwrap();
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@
|
|||
*/
|
||||
|
||||
export const BoardColumn = {
|
||||
QUEUED: 'queued',
|
||||
INITIALIZING: 'initializing',
|
||||
RUNNING: 'running',
|
||||
BLOCKED: 'blocked',
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue