stroblme
71628abcbc
The worker script is handed to the interpreter by path, so app/flow was sys.path[0] for every node: `import queue` got the engine's. It now drops its own directory before anything else imports, and runs with the deployment's credentials scrubbed out of its environment. Also: reload builds off the event loop, the pool wakes what is blocked on it when it stops, a refused metrics flush is kept for the next one rather than dropped, and the cascade events are paired through failures.
325 lines
8.9 KiB
TypeScript
325 lines
8.9 KiB
TypeScript
import { useCallback, useSyncExternalStore } from "react"
|
|
|
|
/**
|
|
* Live engine state, deliberately outside React Query.
|
|
*
|
|
* Values can arrive many times a second. Keeping them here, with one subscriber
|
|
* set per key, means a changing value re-renders the chip showing it and
|
|
* nothing else.
|
|
*/
|
|
|
|
/** Who caused a value. The canvas needs it to pulse the right edge. */
|
|
export type ValueSource = {
|
|
kind: "node" | "dashboard" | "flow" | "agent" | "api"
|
|
id: string
|
|
label: string
|
|
detail?: string
|
|
}
|
|
export type LiveValue = {
|
|
value: unknown
|
|
ts: number | null
|
|
source?: ValueSource
|
|
}
|
|
export type LiveStatus = {
|
|
status: "active" | "error" | "running" | "success"
|
|
error?: string | null
|
|
}
|
|
/** How a node's connection is doing, which is not how its last run went. */
|
|
export type NodeHealth = {
|
|
health: "ok" | "down" | "unknown"
|
|
detail?: string | null
|
|
}
|
|
/** Something the engine reported about itself, for the health page. */
|
|
export type EngineEvent = {
|
|
type: string
|
|
flow?: string
|
|
node?: string
|
|
detail?: string
|
|
ts: number
|
|
}
|
|
/** One node execution's output, as the log panel shows it. */
|
|
export type LogLine = {
|
|
flow: string
|
|
node: string
|
|
text: string
|
|
level: "info" | "error"
|
|
truncated?: boolean
|
|
ts: number
|
|
}
|
|
|
|
type Listener = () => void
|
|
|
|
/** Enough to see what a flow has been doing, not a log store. */
|
|
const LOG_LIMIT = 500
|
|
/** The health page reads these to know when to refetch; it is not a history. */
|
|
const ENGINE_EVENT_LIMIT = 100
|
|
|
|
const values = new Map<string, LiveValue>()
|
|
const statuses = new Map<string, LiveStatus>()
|
|
const health = new Map<string, NodeHealth>()
|
|
let engineEvents: EngineEvent[] = []
|
|
// How many times a node has emitted. The number itself means nothing; a change
|
|
// is what restarts the pulse.
|
|
const emits = new Map<string, number>()
|
|
let logLines: LogLine[] = []
|
|
const paused = new Set<string>()
|
|
const listeners = new Map<string, Set<Listener>>()
|
|
|
|
let connected = false
|
|
const connectionListeners = new Set<Listener>()
|
|
|
|
function notify(key: string) {
|
|
for (const listener of listeners.get(key) ?? []) listener()
|
|
}
|
|
|
|
function subscribeKey(key: string, listener: Listener) {
|
|
let set = listeners.get(key)
|
|
if (!set) {
|
|
set = new Set()
|
|
listeners.set(key, set)
|
|
}
|
|
set.add(listener)
|
|
return () => {
|
|
set.delete(listener)
|
|
if (set.size === 0) listeners.delete(key)
|
|
}
|
|
}
|
|
|
|
export const liveStore = {
|
|
setValue(name: string, value: LiveValue) {
|
|
values.set(name, value)
|
|
notify(`value:${name}`)
|
|
},
|
|
setValues(entries: Record<string, LiveValue>) {
|
|
for (const [name, value] of Object.entries(entries)) {
|
|
values.set(name, value)
|
|
notify(`value:${name}`)
|
|
}
|
|
},
|
|
getValue(name: string) {
|
|
return values.get(name)
|
|
},
|
|
setStatus(nodeId: string, status: LiveStatus) {
|
|
statuses.set(nodeId, status)
|
|
notify(`status:${nodeId}`)
|
|
},
|
|
setStatuses(
|
|
entries: { id: string; status: string; error?: string | null }[],
|
|
) {
|
|
for (const entry of entries) {
|
|
statuses.set(entry.id, {
|
|
status: entry.status as LiveStatus["status"],
|
|
error: entry.error,
|
|
})
|
|
notify(`status:${entry.id}`)
|
|
}
|
|
},
|
|
getStatus(nodeId: string) {
|
|
return statuses.get(nodeId)
|
|
},
|
|
setHealth(nodeId: string, entry: NodeHealth) {
|
|
health.set(nodeId, entry)
|
|
notify(`health:${nodeId}`)
|
|
},
|
|
getHealth(nodeId: string) {
|
|
return health.get(nodeId)
|
|
},
|
|
recordEngineEvent(event: EngineEvent) {
|
|
// A new array each time, so the hook's snapshot comparison sees the change.
|
|
engineEvents = [...engineEvents, event].slice(-ENGINE_EVENT_LIMIT)
|
|
notify("engine")
|
|
},
|
|
recordEmit(nodeId: string) {
|
|
emits.set(nodeId, (emits.get(nodeId) ?? 0) + 1)
|
|
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)
|
|
notify("logs")
|
|
},
|
|
setLogs(lines: LogLine[]) {
|
|
logLines = lines.slice(-LOG_LIMIT)
|
|
notify("logs")
|
|
},
|
|
clearLogs() {
|
|
logLines = []
|
|
notify("logs")
|
|
},
|
|
setPaused(flow: string, isPaused: boolean) {
|
|
if (isPaused) paused.add(flow)
|
|
else paused.delete(flow)
|
|
notify(`paused:${flow}`)
|
|
},
|
|
setPausedFlows(flows: string[]) {
|
|
const next = new Set(flows)
|
|
for (const flow of new Set([...paused, ...next])) {
|
|
if (paused.has(flow) === next.has(flow)) continue
|
|
if (next.has(flow)) paused.add(flow)
|
|
else paused.delete(flow)
|
|
notify(`paused:${flow}`)
|
|
}
|
|
},
|
|
setConnected(next: boolean) {
|
|
if (connected === next) return
|
|
connected = next
|
|
for (const listener of connectionListeners) listener()
|
|
},
|
|
isConnected() {
|
|
return connected
|
|
},
|
|
reset() {
|
|
for (const key of values.keys()) notify(`value:${key}`)
|
|
values.clear()
|
|
for (const key of statuses.keys()) notify(`status:${key}`)
|
|
statuses.clear()
|
|
for (const key of emits.keys()) notify(`emit:${key}`)
|
|
emits.clear()
|
|
for (const key of health.keys()) notify(`health:${key}`)
|
|
health.clear()
|
|
engineEvents = []
|
|
notify("engine")
|
|
logLines = []
|
|
notify("logs")
|
|
for (const flow of paused) notify(`paused:${flow}`)
|
|
paused.clear()
|
|
},
|
|
}
|
|
|
|
export function useLiveValue(name: string | undefined): LiveValue | undefined {
|
|
return useSyncExternalStore(
|
|
(listener) => (name ? subscribeKey(`value:${name}`, listener) : () => {}),
|
|
() => (name ? values.get(name) : undefined),
|
|
)
|
|
}
|
|
|
|
export function useNodeStatus(nodeId: string): LiveStatus | undefined {
|
|
return useSyncExternalStore(
|
|
(listener) => subscribeKey(`status:${nodeId}`, listener),
|
|
() => statuses.get(nodeId),
|
|
)
|
|
}
|
|
|
|
/** How the node's connection is doing, once it has said anything about it. */
|
|
export function useNodeHealth(nodeId: string): NodeHealth | undefined {
|
|
return useSyncExternalStore(
|
|
(listener) => subscribeKey(`health:${nodeId}`, listener),
|
|
() => health.get(nodeId),
|
|
)
|
|
}
|
|
|
|
/** The last hundred things the engine said about itself, oldest first. */
|
|
export function useEngineEvents(): EngineEvent[] {
|
|
return useSyncExternalStore(
|
|
(listener) => subscribeKey("engine", listener),
|
|
() => engineEvents,
|
|
)
|
|
}
|
|
|
|
/** Increments each time the node publishes something. */
|
|
export function useNodeEmits(nodeId: string): number {
|
|
return useSyncExternalStore(
|
|
(listener) => subscribeKey(`emit:${nodeId}`, listener),
|
|
() => emits.get(nodeId) ?? 0,
|
|
)
|
|
}
|
|
|
|
/** Every captured line, newest last. Filtered by the panel that shows it. */
|
|
export function useLiveLogs(): LogLine[] {
|
|
return useSyncExternalStore(
|
|
(listener) => subscribeKey("logs", listener),
|
|
() => logLines,
|
|
)
|
|
}
|
|
|
|
export function useFlowPaused(flow: string): boolean {
|
|
return useSyncExternalStore(
|
|
(listener) => subscribeKey(`paused:${flow}`, listener),
|
|
() => paused.has(flow),
|
|
)
|
|
}
|
|
|
|
/*
|
|
* Group hooks, for the brain graph: one neuron stands for several nodes, and
|
|
* one edge for several messages.
|
|
*
|
|
* Each returns a primitive. `useSyncExternalStore` compares snapshots by
|
|
* identity, so a fresh object or array every render would re-render forever.
|
|
* The names arrive as an array, which is also new every render, so they are
|
|
* joined into one string before anything hangs off them.
|
|
*/
|
|
|
|
const SEP = "\u0000"
|
|
|
|
function subscribeAll(keys: string[], listener: Listener) {
|
|
const unsubscribes = keys.map((key) => subscribeKey(key, listener))
|
|
return () => {
|
|
for (const unsubscribe of unsubscribes) unsubscribe()
|
|
}
|
|
}
|
|
|
|
/** How many times any of these nodes has published. A change is a pulse. */
|
|
export function useGroupEmits(ids: string[]): number {
|
|
const joined = ids.join(SEP)
|
|
return useSyncExternalStore(
|
|
useCallback(
|
|
(listener: Listener) =>
|
|
subscribeAll(
|
|
joined.split(SEP).map((id) => `emit:${id}`),
|
|
listener,
|
|
),
|
|
[joined],
|
|
),
|
|
() =>
|
|
joined.split(SEP).reduce((total, id) => total + (emits.get(id) ?? 0), 0),
|
|
)
|
|
}
|
|
|
|
/** Whether any of these nodes is currently failing. */
|
|
export function useGroupError(ids: string[]): boolean {
|
|
const joined = ids.join(SEP)
|
|
return useSyncExternalStore(
|
|
useCallback(
|
|
(listener: Listener) =>
|
|
subscribeAll(
|
|
joined.split(SEP).map((id) => `status:${id}`),
|
|
listener,
|
|
),
|
|
[joined],
|
|
),
|
|
() => joined.split(SEP).some((id) => statuses.get(id)?.status === "error"),
|
|
)
|
|
}
|
|
|
|
/** When any of these messages last carried a value. */
|
|
export function useLatestTs(names: string[]): number {
|
|
const joined = names.join(SEP)
|
|
return useSyncExternalStore(
|
|
useCallback(
|
|
(listener: Listener) =>
|
|
subscribeAll(
|
|
joined.split(SEP).map((name) => `value:${name}`),
|
|
listener,
|
|
),
|
|
[joined],
|
|
),
|
|
() =>
|
|
joined
|
|
.split(SEP)
|
|
.reduce(
|
|
(latest, name) => Math.max(latest, values.get(name)?.ts ?? 0),
|
|
0,
|
|
),
|
|
)
|
|
}
|
|
|
|
export function useLiveConnection(): boolean {
|
|
return useSyncExternalStore(
|
|
(listener) => {
|
|
connectionListeners.add(listener)
|
|
return () => connectionListeners.delete(listener)
|
|
},
|
|
() => connected,
|
|
)
|
|
}
|