Own the live socket outside React so leaving Home cannot orphan it
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) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HTsT1isxUjw5gtkJk8WhuA
This commit is contained in:
Claude Opus 5
@@ -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<string, number>
|
||||
}
|
||||
| {
|
||||
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<typeof setTimeout> | 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<WebSocket | null>(null)
|
||||
const retry = useRef(RECONNECT_MIN)
|
||||
const timer = useRef<ReturnType<typeof setTimeout> | 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])
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user