From bbff78766e755d2c4391ae9c2bfd24af50b12f68 Mon Sep 17 00:00:00 2001 From: stroblme Date: Thu, 20 Aug 2026 11:58:49 +0200 Subject: [PATCH] Own the live socket outside React so leaving Home cannot orphan it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The socket belonged to whichever hook instance ran its effect first. Passive effects run children before parents, so on Home that was the brain graph rather than the shell: navigating to a sibling route unmounted the graph, which closed the socket, while the shell kept the reference count above zero. From there the page was deaf for the rest of its life, with nothing left to reconnect it. A module-level connection with a real refcount replaces it — connect on the first subscriber, disconnect on the last — and the hook is a thin subscription with the same signature, correct under StrictMode's mount/unmount/mount. A 1008 now reconnects instead of returning silently: the token is read afresh per attempt, and three consecutive rejections fall through to the caller's auth handler so a revoked session surfaces rather than spins. The snapshot's emit counts are read into a store of their own, apart from the live count, so a graph that connects into a busy engine is drawn as busy without every neuron claiming it just fired. The neuron and edge pulses now key off a change seen while they were mounted, so returning to Home no longer replays every emission of the session. Home gets a live indicator for the case none of this can fix: quiet while the socket is up, and named in words when it is down, since HTTP polling keeps the rest of the page looking current. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01HTsT1isxUjw5gtkJk8WhuA --- frontend/src/components/Flow/BrainEdge.tsx | 5 +- frontend/src/components/Flow/liveStore.ts | 75 +++- frontend/src/components/Flow/useFlowSocket.ts | 352 ++++++++++-------- .../src/components/Health/LiveIndicator.tsx | 73 ++++ frontend/src/routes/_layout/index.tsx | 5 + frontend/tests/live.spec.ts | 95 +++++ 6 files changed, 439 insertions(+), 166 deletions(-) create mode 100644 frontend/src/components/Health/LiveIndicator.tsx create mode 100644 frontend/tests/live.spec.ts 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) +})