diff --git a/frontend/src/components/Flow/BrainEdge.tsx b/frontend/src/components/Flow/BrainEdge.tsx index 554a4b5..1283e55 100644 --- a/frontend/src/components/Flow/BrainEdge.tsx +++ b/frontend/src/components/Flow/BrainEdge.tsx @@ -47,7 +47,10 @@ function BrainEdgeComponent({ id, source, target, data }: EdgeProps) { const targetNode = useInternalNode(target) const ts = useLatestTs(messages) const [pulsing, setPulsing] = useState(false) - const lastTs = useRef(0) + // Seeded with whatever the store already holds rather than with 0: the store + // outlives this component, so a value that arrived before it mounted is + // history and must not be replayed as a pulse. + const lastTs = useRef(ts) // The two centres, taken from the layout rather than from the `sourceX`/ // `targetX` React Flow hands an edge. Those come off the handles, which it diff --git a/frontend/src/components/Flow/liveStore.ts b/frontend/src/components/Flow/liveStore.ts index cb54766..a561d44 100644 --- a/frontend/src/components/Flow/liveStore.ts +++ b/frontend/src/components/Flow/liveStore.ts @@ -58,9 +58,14 @@ const values = new Map() const statuses = new Map() const health = new Map() let engineEvents: EngineEvent[] = [] -// How many times a node has emitted. The number itself means nothing; a change -// is what restarts the pulse. +// How many times this page has seen a node emit. The number itself means +// nothing; a change is what restarts the pulse. const emits = new Map() +// What the engine had already counted when we connected, from the snapshot. +// Kept apart from the live count on purpose: it says the node has been busy, +// which is worth drawing, but it happened before anyone was watching, and a +// pulse claims something is happening now. +const priorEmits = new Map() let logLines: LogLine[] = [] const paused = new Set() const listeners = new Map>() @@ -133,6 +138,21 @@ export const liveStore = { emits.set(nodeId, (emits.get(nodeId) ?? 0) + 1) notify(`emit:${nodeId}`) }, + /** + * The counts a snapshot carries, so a page that connects into an engine + * already at work draws what has been busy rather than a graph of idle + * neurons. + * + * Absent keys are left alone rather than cleared: a restarted engine counts + * from zero again, which is no reason to forget what this page saw. + */ + setEmits(entries: Record) { + for (const [nodeId, count] of Object.entries(entries)) { + if (priorEmits.get(nodeId) === count) continue + priorEmits.set(nodeId, count) + notify(`emit:${nodeId}`) + } + }, appendLog(line: LogLine) { // A new array each time, so the hook's snapshot comparison sees the change. logLines = [...logLines, line].slice(-LOG_LIMIT) @@ -173,8 +193,10 @@ export const liveStore = { values.clear() for (const key of statuses.keys()) notify(`status:${key}`) statuses.clear() - for (const key of emits.keys()) notify(`emit:${key}`) + for (const key of new Set([...emits.keys(), ...priorEmits.keys()])) + notify(`emit:${key}`) emits.clear() + priorEmits.clear() for (const key of health.keys()) notify(`health:${key}`) health.clear() engineEvents = [] @@ -251,6 +273,9 @@ export function useFlowPaused(flow: string): boolean { const SEP = "\u0000" +/** Back to the array. `"".split(SEP)` is `[""]`, which is one empty name. */ +const parts = (joined: string) => (joined ? joined.split(SEP) : []) + function subscribeAll(keys: string[], listener: Listener) { const unsubscribes = keys.map((key) => subscribeKey(key, listener)) return () => { @@ -265,13 +290,37 @@ export function useGroupEmits(ids: string[]): number { useCallback( (listener: Listener) => subscribeAll( - joined.split(SEP).map((id) => `emit:${id}`), + parts(joined).map((id) => `emit:${id}`), + listener, + ), + [joined], + ), + () => parts(joined).reduce((total, id) => total + (emits.get(id) ?? 0), 0), + ) +} + +/** + * Whether any of these nodes has ever published, this page or before it. + * + * Separate from the count above because it answers a different question: the + * count is what pulses, and this is what tells a neuron nobody has ever seen + * run from one that is merely quiet just now. + */ +export function useGroupActive(ids: string[]): boolean { + const joined = ids.join(SEP) + return useSyncExternalStore( + useCallback( + (listener: Listener) => + subscribeAll( + parts(joined).map((id) => `emit:${id}`), listener, ), [joined], ), () => - joined.split(SEP).reduce((total, id) => total + (emits.get(id) ?? 0), 0), + parts(joined).some( + (id) => (emits.get(id) ?? 0) + (priorEmits.get(id) ?? 0) > 0, + ), ) } @@ -282,12 +331,12 @@ export function useGroupError(ids: string[]): boolean { useCallback( (listener: Listener) => subscribeAll( - joined.split(SEP).map((id) => `status:${id}`), + parts(joined).map((id) => `status:${id}`), listener, ), [joined], ), - () => joined.split(SEP).some((id) => statuses.get(id)?.status === "error"), + () => parts(joined).some((id) => statuses.get(id)?.status === "error"), ) } @@ -298,18 +347,16 @@ export function useLatestTs(names: string[]): number { useCallback( (listener: Listener) => subscribeAll( - joined.split(SEP).map((name) => `value:${name}`), + parts(joined).map((name) => `value:${name}`), listener, ), [joined], ), () => - joined - .split(SEP) - .reduce( - (latest, name) => Math.max(latest, values.get(name)?.ts ?? 0), - 0, - ), + parts(joined).reduce( + (latest, name) => Math.max(latest, values.get(name)?.ts ?? 0), + 0, + ), ) } diff --git a/frontend/src/components/Flow/useFlowSocket.ts b/frontend/src/components/Flow/useFlowSocket.ts index 095d80e..f1f2d16 100644 --- a/frontend/src/components/Flow/useFlowSocket.ts +++ b/frontend/src/components/Flow/useFlowSocket.ts @@ -1,5 +1,6 @@ +import type { QueryClient } from "@tanstack/react-query" import { useQueryClient } from "@tanstack/react-query" -import { useEffect, useRef } from "react" +import { useEffect } from "react" import { OpenAPI } from "@/client" import { connectionStore } from "@/lib/connectionStore" @@ -8,10 +9,18 @@ import { type LogLine, liveStore, type ValueSource } from "./liveStore" import { flowKeys } from "./queries" const RECONNECT_MIN = 1000 - -/** How many components want the socket open. */ -let mounted = 0 const RECONNECT_MAX = 30000 +/** + * Consecutive token rejections before the session is given up on. + * + * A 1008 means the server would not take the token. Under a portal that is + * routine — the handoff token lasts twelve hours against a local session's + * eight days — so it has to be recoverable rather than fatal: the token is read + * again for every attempt, so a refreshed one is picked up. A session that has + * genuinely been revoked fails all three and falls through to the caller's auth + * handler instead of retrying forever. + */ +const AUTH_ATTEMPTS = 3 type FlowEvent = | { @@ -20,6 +29,8 @@ type FlowEvent = nodes: { id: string; status: string; error?: string | null }[] paused?: string[] logs?: LogLine[] + /** Emission counts per qualified node. Absent on older installations. */ + emits?: Record } | { type: "message_value" @@ -85,163 +96,202 @@ function socketUrl(): string { const url = new URL(`${base.replace(/\/$/, "")}/api/v1/flows/ws`) url.protocol = url.protocol === "https:" ? "wss:" : "ws:" // Browsers cannot set headers on a websocket handshake, so the token rides - // in the query string. + // in the query string. Read per attempt, so a reconnect after the old one + // expired carries whatever is current rather than what we started with. url.searchParams.set("token", apiToken()) return url.toString() } -/** - * Keeps one socket open for the editor, feeding the live store. +/* + * One socket for the page, owned here rather than by a component. * - * @param onAuthFailure called when the server rejects the token, so the caller - * can send the user back to the login screen. + * Several places want live events at once — the app shell, the brain graph on + * Home, the editor — and React mounts and unmounts them in an order none of + * them controls: passive effects run children before parents, and StrictMode + * runs mount → unmount → mount. Keeping the socket inside whichever hook + * instance ran first made that one its owner, so leaving Home closed the socket + * while the shell kept the count above zero and nothing ever reopened it. The + * connection lives out here now with a plain refcount; a component only says + * whether it is still watching. + */ +let socket: WebSocket | null = null +let timer: ReturnType | null = null +let retry = RECONNECT_MIN +/** How many mounted components want the socket open. */ +let watchers = 0 +/** Consecutive 1008s, cleared by a handshake the server accepts. */ +let rejected = 0 +let client: QueryClient | null = null +const authHandlers = new Set<() => void>() + +function schedule() { + if (timer) return + timer = setTimeout(() => { + timer = null + connect() + }, retry) + retry = Math.min(retry * 2, RECONNECT_MAX) +} + +function connect() { + if (watchers === 0 || socket || timer) return + const ws = new WebSocket(socketUrl()) + socket = ws + + ws.onopen = () => { + retry = RECONNECT_MIN + rejected = 0 + liveStore.setConnected(true) + connectionStore.setOnline() + // Whatever happened while the socket was down was missed, so nothing + // held in cache can be trusted to still be current. + client?.invalidateQueries() + } + + ws.onmessage = (event) => { + const message: FlowEvent = JSON.parse(event.data) + switch (message.type) { + case "snapshot": + liveStore.setValues(message.values) + liveStore.setStatuses(message.nodes) + liveStore.setPausedFlows(message.paused ?? []) + liveStore.setLogs(message.logs ?? []) + // Missing against an installation older than this bundle; the graph + // then starts from zero the way it always did. + liveStore.setEmits(message.emits ?? {}) + break + case "message_value": + liveStore.setValue(message.name, { + value: message.value, + ts: message.ts, + source: message.source, + }) + break + case "node_started": + liveStore.setStatus(message.node, { status: "running" }) + break + case "node_executed": + liveStore.setStatus(message.node, { status: "success" }) + if (message.outputs > 0) liveStore.recordEmit(message.node) + break + case "node_error": + liveStore.setStatus(message.node, { + status: "error", + error: message.error, + }) + liveStore.recordEngineEvent({ + type: message.type, + flow: message.flow, + node: message.node, + detail: message.error, + ts: message.ts ?? Date.now() / 1000, + }) + break + case "node_health": + liveStore.setHealth(message.node, { + health: message.health, + detail: message.detail, + }) + if (message.health === "down") { + liveStore.recordEngineEvent({ + type: message.type, + flow: message.flow, + node: message.node, + detail: message.detail ?? "Reported itself down.", + ts: message.ts ?? Date.now() / 1000, + }) + } + break + case "flow_quarantined": + case "engine_degraded": + case "engine_fatal": + case "cascade_dropped": + case "queue_unavailable": + liveStore.recordEngineEvent({ + type: message.type, + flow: "flow" in message ? message.flow : undefined, + node: "node" in message ? message.node : undefined, + detail: + ("error" in message ? message.error : undefined) ?? + ("reason" in message ? message.reason : undefined) ?? + "", + ts: message.ts ?? Date.now() / 1000, + }) + break + case "node_status": + liveStore.setStatus(message.node, { + status: message.status as "active" | "error", + error: message.error, + }) + break + case "node_log": + liveStore.appendLog(message) + break + case "flow_paused": + liveStore.setPaused(message.flow, message.paused) + break + case "pipeline_rebuilt": + liveStore.setStatuses(message.nodes) + liveStore.setPausedFlows(message.paused ?? []) + // Someone published or started a flow, here or in another tab: the + // markers on the flow chips are stale until the list is refetched. + client?.invalidateQueries({ queryKey: flowKeys.all }) + break + } + } + + ws.onclose = (event) => { + // A socket we already dropped: its close says nothing about the connection + // we want now. + if (socket !== ws) return + socket = null + liveStore.setConnected(false) + if (watchers === 0) return + if (event.code === 1008) { + rejected += 1 + if (rejected >= AUTH_ATTEMPTS) { + for (const handler of authHandlers) handler() + return + } + } + // 1013 is the portal saying the installation is not attached — the one + // close code that means "offline" rather than "the socket dropped". + if (event.code === 1013) { + connectionStore.setOffline(null) + } + schedule() + } +} + +function release(onAuthFailure?: () => void) { + watchers -= 1 + if (onAuthFailure) authHandlers.delete(onAuthFailure) + if (watchers > 0) return + if (timer) { + clearTimeout(timer) + timer = null + } + const ws = socket + socket = null + ws?.close() + liveStore.setConnected(false) +} + +/** + * Keeps one socket open while anything on the page wants live events. + * + * @param onAuthFailure called once the server has refused the token often + * enough to mean the session is gone, so the caller can send the user back to + * the login screen. */ export function useFlowSocket(onAuthFailure?: () => void): void { - const socket = useRef(null) - const retry = useRef(RECONNECT_MIN) - const timer = useRef | null>(null) - const closed = useRef(false) const queryClient = useQueryClient() useEffect(() => { - // The editor and a dashboard can both be mounted; one socket serves both, - // and the second caller just rides along. - mounted += 1 - if (mounted > 1) { - return () => { - mounted -= 1 - } - } - closed.current = false - - const connect = () => { - if (closed.current) return - const ws = new WebSocket(socketUrl()) - socket.current = ws - - ws.onopen = () => { - retry.current = RECONNECT_MIN - liveStore.setConnected(true) - connectionStore.setOnline() - // Whatever happened while the socket was down was missed, so nothing - // held in cache can be trusted to still be current. - queryClient.invalidateQueries() - } - - ws.onmessage = (event) => { - const message: FlowEvent = JSON.parse(event.data) - switch (message.type) { - case "snapshot": - liveStore.setValues(message.values) - liveStore.setStatuses(message.nodes) - liveStore.setPausedFlows(message.paused ?? []) - liveStore.setLogs(message.logs ?? []) - break - case "message_value": - liveStore.setValue(message.name, { - value: message.value, - ts: message.ts, - source: message.source, - }) - break - case "node_started": - liveStore.setStatus(message.node, { status: "running" }) - break - case "node_executed": - liveStore.setStatus(message.node, { status: "success" }) - if (message.outputs > 0) liveStore.recordEmit(message.node) - break - case "node_error": - liveStore.setStatus(message.node, { - status: "error", - error: message.error, - }) - liveStore.recordEngineEvent({ - type: message.type, - flow: message.flow, - node: message.node, - detail: message.error, - ts: message.ts ?? Date.now() / 1000, - }) - break - case "node_health": - liveStore.setHealth(message.node, { - health: message.health, - detail: message.detail, - }) - if (message.health === "down") { - liveStore.recordEngineEvent({ - type: message.type, - flow: message.flow, - node: message.node, - detail: message.detail ?? "Reported itself down.", - ts: message.ts ?? Date.now() / 1000, - }) - } - break - case "flow_quarantined": - case "engine_degraded": - case "engine_fatal": - case "cascade_dropped": - case "queue_unavailable": - liveStore.recordEngineEvent({ - type: message.type, - flow: "flow" in message ? message.flow : undefined, - node: "node" in message ? message.node : undefined, - detail: - ("error" in message ? message.error : undefined) ?? - ("reason" in message ? message.reason : undefined) ?? - "", - ts: message.ts ?? Date.now() / 1000, - }) - break - case "node_status": - liveStore.setStatus(message.node, { - status: message.status as "active" | "error", - error: message.error, - }) - break - case "node_log": - liveStore.appendLog(message) - break - case "flow_paused": - liveStore.setPaused(message.flow, message.paused) - break - case "pipeline_rebuilt": - liveStore.setStatuses(message.nodes) - liveStore.setPausedFlows(message.paused ?? []) - // Someone published or started a flow, here or in another tab: the - // markers on the flow chips are stale until the list is refetched. - queryClient.invalidateQueries({ queryKey: flowKeys.all }) - break - } - } - - ws.onclose = (event) => { - liveStore.setConnected(false) - if (closed.current) return - if (event.code === 1008) { - onAuthFailure?.() - return - } - // 1013 is the portal saying the installation is not attached — the one - // close code that means "offline" rather than "the socket dropped". - if (event.code === 1013) { - connectionStore.setOffline(null) - } - timer.current = setTimeout(connect, retry.current) - retry.current = Math.min(retry.current * 2, RECONNECT_MAX) - } - } - + client = queryClient + if (onAuthFailure) authHandlers.add(onAuthFailure) + watchers += 1 connect() - - return () => { - mounted -= 1 - closed.current = true - if (timer.current) clearTimeout(timer.current) - socket.current?.close() - liveStore.setConnected(false) - } + return () => release(onAuthFailure) }, [onAuthFailure, queryClient]) } diff --git a/frontend/src/components/Health/LiveIndicator.tsx b/frontend/src/components/Health/LiveIndicator.tsx new file mode 100644 index 0000000..cfbb491 --- /dev/null +++ b/frontend/src/components/Health/LiveIndicator.tsx @@ -0,0 +1,73 @@ +import { Activity } from "lucide-react" +import { AnimatePresence, motion } from "motion/react" +import { useEffect, useState, useSyncExternalStore } from "react" + +import { useLiveConnection } from "@/components/Flow/liveStore" +import { connectionStore } from "@/lib/connectionStore" +import { fadeIn } from "@/lib/motion" + +/** + * How long the socket may be down before it is worth saying so. + * + * The first reconnect attempt is a second away, so a dropped frame is back + * before this elapses and nothing appears. Long enough to cover the opening + * handshake too: a page that has just loaded is connecting, not disconnected. + */ +const GRACE = 3000 + +/** + * Says when live values have stopped arriving. + * + * Everything else on Home is polled over HTTP, so a dead socket leaves the page + * looking perfectly current while nothing on it moves any more — the neurons + * stop pulsing and the values stop changing, with nothing to say why. Nothing + * at all while the socket is up: a page that works needs no chip saying so. + * + * Quiet too while `ConnectionBanner` is up, since an installation that cannot + * be reached has no socket either and one explanation of that is enough. Named + * in words rather than coloured, and deliberately not terracotta: the brain + * graph above it already owns that accent for a flow that cannot run. + */ +export function LiveIndicator() { + const connected = useLiveConnection() + const { offline } = useSyncExternalStore( + connectionStore.subscribe, + connectionStore.snapshot, + connectionStore.snapshot, + ) + const [down, setDown] = useState(false) + + useEffect(() => { + if (connected) { + setDown(false) + return + } + const timer = setTimeout(() => setDown(true), GRACE) + return () => clearTimeout(timer) + }, [connected]) + + return ( + + {down && !offline ? ( + + + Live updates disconnected + + values and activity are frozen — reconnecting… + + + ) : null} + + ) +} diff --git a/frontend/src/routes/_layout/index.tsx b/frontend/src/routes/_layout/index.tsx index d4932de..e4de0ed 100644 --- a/frontend/src/routes/_layout/index.tsx +++ b/frontend/src/routes/_layout/index.tsx @@ -9,6 +9,7 @@ import { BrainView } from "@/components/Flow/BrainView" import { flowKeys, flowsQueryOptions } from "@/components/Flow/queries" import { HealthActivity } from "@/components/Health/HealthActivity" import { HealthOverview } from "@/components/Health/HealthOverview" +import { LiveIndicator } from "@/components/Health/LiveIndicator" import { Badge } from "@/components/ui/badge" import { Card } from "@/components/ui/card" import { Skeleton } from "@/components/ui/skeleton" @@ -116,6 +117,10 @@ function Dashboard() { // Every section here is free to shrink instead. See DESIGN-GUIDELINES.md // → Responsive.
+ {/* Nothing at all while the socket is up. Everything below is polled, so + a dead socket would otherwise look like an engine that went quiet. */} + + {/* Nothing wired up yet means nothing to draw, and the band would still hold a screenful of empty space above the flows card. Node count rather than flow count: a flow made a minute ago has none. */} diff --git a/frontend/tests/live.spec.ts b/frontend/tests/live.spec.ts new file mode 100644 index 0000000..99600a7 --- /dev/null +++ b/frontend/tests/live.spec.ts @@ -0,0 +1,95 @@ +import { expect, test } from "@playwright/test" +import { api, apiPage, deleteAll } from "./utils/api" + +/** + * One socket serves the whole shell, and it has to survive the routes under it + * coming and going. + * + * It used to belong to whichever component's effect ran first, which on Home is + * the brain graph rather than the shell around it. Leaving Home closed the + * socket while the shell held the reference count above zero, and the page was + * deaf for the rest of its life: no live values, no pulses, and nothing that + * would ever reconnect. + */ + +const flowName = `test_live_${Date.now().toString(36)}` + +test.use({ storageState: "playwright/.auth/user.json" }) + +test.beforeAll(async ({ browser }) => { + const page = await apiPage(browser) + await api(page, `/flows/${flowName}`, { + method: "PUT", + data: { + name: flowName, + title: "Live", + nodes: [ + { + id: "source", + type: "python", + provides: [{ name: "reading", dtype: "float" }], + }, + { + id: "sink", + type: "python", + requires: [{ name: "reading", dtype: "float" }], + }, + ], + }, + }) + const detail = await (await api(page, `/flows/${flowName}`)).json() + await api(page, `/flows/${flowName}/publish`, { + method: "POST", + data: { version: detail.definition.version }, + }) + await page.close() +}) + +test.afterAll(async ({ browser }) => { + await deleteAll(browser, [`/flows/${flowName}`]) +}) + +test("the live socket survives leaving Home and coming back", async ({ + page, +}) => { + await page.addInitScript(() => { + const Native = window.WebSocket + const opened: WebSocket[] = [] + ;(window as unknown as { __sockets: WebSocket[] }).__sockets = opened + window.WebSocket = class extends Native { + constructor(url: string | URL, protocols?: string | string[]) { + super(url, protocols) + opened.push(this) + } + } + }) + + await page.goto("/") + const neuron = page.locator(".brain-cell").first() + await neuron.waitFor() + + // The full-bleed canvas is its own shell, so a trip through it takes the + // sidebar shell down to nothing: coming back mounts the shell and the brain + // graph in the same commit, which is the order that used to pick an owner. + await page.goto(`/flows/${flowName}`) + const bar = page.locator('[data-sidebar="sidebar"]') + await bar.getByRole("link", { name: "Home", exact: true }).click() + await neuron.waitFor() + + // And this is the navigation that used to end it: a sibling route in the + // same shell, which unmounts the graph but not the shell. + await bar.getByRole("link", { name: "Flows", exact: true }).click() + await page.waitForURL(/\/flows$/) + await bar.getByRole("link", { name: "Home", exact: true }).click() + await neuron.waitFor() + + await expect + .poll(() => + page.evaluate(() => + (window as unknown as { __sockets: WebSocket[] }).__sockets.some( + (socket) => socket.readyState === 1, + ), + ), + ) + .toBe(true) +})