diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index 325ec52..3060626 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -30,7 +30,7 @@ from fluksio.core.db import engine from fluksio.flow.controller import FlowController from fluksio.flow.dashboards import DashboardStore from fluksio.flow.events import event_bus -from fluksio.flow.messages import qualify +from fluksio.flow.messages import flow_of, qualify from fluksio.flow.panels import messages_for from fluksio.flow.pipeline import ValidationIssue from fluksio.flow.runs import RunRejected @@ -130,36 +130,65 @@ class TriggerRequest(BaseModel): values: dict[str, Any] = {} -def _endpoints(controller: FlowController, flow: str) -> list[Endpoint]: - """Everything wired into ``flow`` from outside it.""" - found: list[Endpoint] = [] +def _endpoints(controller: FlowController, definition: FlowDef) -> list[Endpoint]: + """Everything wired into this flow from outside it.""" + flow = definition.name + found: dict[str, Endpoint] = {} dashboards: DashboardStore | None = getattr(controller, "dashboards", None) if dashboards is not None: for binding in dashboards.bindings_for(flow): - found.append( - Endpoint( - kind="dashboard", - id=f"dashboard:{binding['dashboard']}:{binding['widget']}", - label=binding["title"], - detail=binding["type"], - provides=[binding["provides"]] if binding["provides"] else [], - requires=binding["requires"], - ) + found[f"dashboard:{binding['dashboard']}:{binding['widget']}"] = Endpoint( + kind="dashboard", + id=f"dashboard:{binding['dashboard']}:{binding['widget']}", + label=binding["title"], + detail=binding["type"], + provides=[binding["provides"]] if binding["provides"] else [], + requires=binding["requires"], ) - for other, node_id, provides, requires in controller.cross_flow_nodes(flow): - found.append( - Endpoint( - kind="flow", - id=f"flow:{other}.{node_id}", - label=f"{other}.{node_id}", - detail="flow", - provides=provides, - requires=requires, - ) + def _flow_endpoint(key: str, detail: str) -> Endpoint: + """One label per node on the far side, however many names reach it.""" + return found.setdefault( + f"flow:{key}", + Endpoint(kind="flow", id=f"flow:{key}", label=key, detail=detail), ) - return found + + # An outsider reaching into this flow. + for other, node_id, provides, requires in controller.cross_flow_nodes(flow): + endpoint = _flow_endpoint(f"{other}.{node_id}", "flow") + endpoint.provides += provides + endpoint.requires += requires + + # And this flow reaching out: its own ports bound to a message of another + # flow. Read from the working document rather than from the store, so a + # name just typed is drawn before it has been published — the same reason + # the canvas draws its own boundary from the document. + for node in definition.nodes: + for spec, ours_publishes in [ + *((spec, False) for spec in node.requires), + *((spec, True) for spec in node.provides), + ]: + message = qualify(flow, spec.name or "") + if not message or flow_of(message) == flow: + continue + # The node at the other end, so the label reads like a dashboard's: + # what it is on the first line, what sort of thing it is on the + # second. A message no published flow declares yet has no other end + # to name, so it is drawn as the message it is. + far = controller.message_node( + message, published=not ours_publishes, exclude=flow + ) + endpoint = _flow_endpoint( + f"{far.flow}.{far.node}" if far else message, + far.type if far else "flow", + ) + # An endpoint publishing into this flow is what this flow reads. + side = endpoint.requires if ours_publishes else endpoint.provides + if message not in side: + side.append(message) + + return list(found.values()) def _detail(controller: FlowController, definition: FlowDef) -> FlowDetail: @@ -168,7 +197,7 @@ def _detail(controller: FlowController, definition: FlowDef) -> FlowDetail: "enabled": controller.is_enabled(name), "paused": controller.is_paused(name), } - endpoints = _endpoints(controller, name) + endpoints = _endpoints(controller, definition) if controller.store.has_draft(name): # Report the draft the editor is showing, not the version running # underneath it — otherwise a node the author just broke looks fine. diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index d8d8b56..1eddce6 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -113,6 +113,17 @@ class RunContext: run_id: str +@dataclass(frozen=True, slots=True) +class Wiring: + """One published node and the qualified messages it is bound to.""" + + flow: str + node: str + type: str + provides: list[str] + requires: list[str] + + class EmitSink: """Turns a worker's mid-call frames back into the node's own outputs. @@ -315,6 +326,9 @@ class FlowController: self.history_limits: dict[str, int] = {} self._lock = asyncio.Lock() self._failures: asyncio.Task[None] | None = None + #: Every published node's ports, with the store revision it was read + #: at — see `_wiring`. + self._wiring_cache: tuple[int, list[Wiring]] | None = None # ------------------------------------------------------------------------- # Lifecycle @@ -843,6 +857,34 @@ class FlowController: ) return infos + def _wiring(self) -> list[Wiring]: + """Every published node and the qualified messages it is bound to. + + Opening one flow asks what every other flow is wired to, in both + directions, which otherwise reads and parses every flow on the disk per + request. Rebuilt when the store commits — see `FlowStore.revision` — + rather than cached with a lifetime, so a save is visible immediately. + """ + revision = self.store.revision + if self._wiring_cache is not None and self._wiring_cache[0] == revision: + return self._wiring_cache[1] + + index: list[Wiring] = [] + for other in self.store.read_all(): + for node in other.nodes: + provides = sorted( + qualify(other.name, spec.name) for spec in _bound(node.provides) + ) + requires = sorted( + qualify(other.name, spec.name) for spec in _bound(node.requires) + ) + if provides or requires: + index.append( + Wiring(other.name, node.id, node.type, provides, requires) + ) + self._wiring_cache = (revision, index) + return index + def cross_flow_nodes( self, flow: str ) -> list[tuple[str, str, list[str], list[str]]]: @@ -854,24 +896,33 @@ class FlowController: """ prefix = f"{flow}." found = [] - for other in self.store.read_all(): - if other.name == flow: + for entry in self._wiring(): + if entry.flow == flow: continue - for node in other.nodes: - provides = sorted( - qualify(other.name, spec.name) - for spec in _bound(node.provides) - if qualify(other.name, spec.name).startswith(prefix) - ) - requires = sorted( - qualify(other.name, spec.name) - for spec in _bound(node.requires) - if qualify(other.name, spec.name).startswith(prefix) - ) - if provides or requires: - found.append((other.name, node.id, provides, requires)) + provides = [name for name in entry.provides if name.startswith(prefix)] + requires = [name for name in entry.requires if name.startswith(prefix)] + if provides or requires: + found.append((entry.flow, entry.node, provides, requires)) return found + def message_node( + self, message: str, published: bool, exclude: str = "" + ) -> Wiring | None: + """The published node on the far side of ``message``. + + ``published`` asks for the one that publishes it, otherwise for one + that reads it, and ``exclude`` names the flow doing the asking — whose + own nodes are never the far side of anything. A message may have + several readers; the first stands for them, which is all a label naming + the other end has to say. + """ + for entry in self._wiring(): + if entry.flow == exclude: + continue + if message in (entry.provides if published else entry.requires): + return entry + return None + def brain_graph(self) -> BrainGraph: """Every published flow as one graph, merged on what its nodes talk to. diff --git a/backend/fluksio/flow/store.py b/backend/fluksio/flow/store.py index fe2ad3f..3d6cfa2 100644 --- a/backend/fluksio/flow/store.py +++ b/backend/fluksio/flow/store.py @@ -103,6 +103,12 @@ class FlowStore: # Draft writes are check-and-set, so two clients saving at once must not # interleave between reading the current version and writing the next. self._write_lock = threading.Lock() + #: Bumped on every commit, so something derived from every flow at once + #: can tell whether it is still current without reading them all again. + #: In memory rather than `head()`, which is a git subprocess per call — + #: it therefore counts from zero per process and misses an edit made on + #: disk behind the API, which no writer here does. + self.revision = 0 if not (self.root / ".git").exists(): self._git("init", "-q") self._commit("Initialise flow store", allow_empty=True) @@ -135,6 +141,10 @@ class FlowStore: ) def _commit(self, message: str, allow_empty: bool = False) -> None: + # Before the commit rather than after it: what a reader has to notice + # is that the files changed, which is already true whether or not git + # is there to record it. + self.revision += 1 self._git("add", "-A") result = self._git( "-c", diff --git a/backend/tests/flow/test_endpoints.py b/backend/tests/flow/test_endpoints.py new file mode 100644 index 0000000..f448e81 --- /dev/null +++ b/backend/tests/flow/test_endpoints.py @@ -0,0 +1,111 @@ +"""What a flow's canvas draws for the wiring that leaves it. + +Both directions matter and only one used to be answered: an outsider reaching +into this flow was drawn, and this flow reaching out was not — so a port bound +to another flow's message had nothing on the canvas to account for it. +""" + +from pathlib import Path + +import pytest + +from fluksio.api.routes.flows import _endpoints +from fluksio.flow.controller import FlowController +from fluksio.flow.messages import DType, MessageSpec +from fluksio.flow.schemas import FlowDef, NodeDef +from fluksio.flow.store import FlowStore + + +def spec(name: str) -> MessageSpec: + return MessageSpec(name=name, dtype=DType.FLOAT) + + +@pytest.fixture +def controller(tmp_path: Path) -> FlowController: + store = FlowStore(tmp_path / "flows") + store.write_flow( + FlowDef( + name="house", + nodes=[ + NodeDef(id="sensor", type="mqtt", provides=[spec("temp")]), + # Reaches into `garage`, which is the direction that was drawn. + NodeDef(id="relay", type="change", provides=[spec("garage.open")]), + ], + ) + ) + store.write_flow( + FlowDef( + name="garage", + nodes=[ + NodeDef(id="door", type="python", requires=[spec("garage.open")]), + # Reaches out into `house`, which was drawn nowhere. + NodeDef(id="watch", type="python", requires=[spec("house.temp")]), + ], + ) + ) + return FlowController(store) + + +def labels( + controller: FlowController, definition: FlowDef +) -> dict[str, tuple[str, str]]: + return { + endpoint.id: (endpoint.label, endpoint.detail) + for endpoint in _endpoints(controller, definition) + } + + +def test_a_port_bound_to_another_flow_names_the_node_at_the_far_end(controller): + endpoints = labels(controller, controller.store.read_flow("garage")) + + # `house.relay` publishes `garage.open`, so it reaches in — drawn before. + # `house.sensor` publishes `house.temp`, which `garage.watch` reads, so + # this flow reaches out — and the label names the node and its type, the + # way a dashboard's label names the widget and its type. + assert endpoints == { + "flow:house.relay": ("house.relay", "flow"), + "flow:house.sensor": ("house.sensor", "mqtt"), + } + + +def test_a_node_reaching_into_and_out_of_one_flow_is_one_label(controller): + endpoints = _endpoints(controller, controller.store.read_flow("house")) + + # `garage.door` reads what `house.relay` publishes, and `garage.watch` + # reads `house.temp` — two directions, two nodes, two labels, no collision. + assert {endpoint.id for endpoint in endpoints} == { + "flow:garage.watch", + "flow:garage.door", + } + door = next(one for one in endpoints if one.id == "flow:garage.door") + assert door.requires == ["garage.open"] + + +def test_a_name_no_published_flow_declares_is_drawn_as_the_name(controller): + definition = controller.store.read_flow("house") + definition.nodes.append( + NodeDef(id="guess", type="python", requires=[spec("shed.humidity")]) + ) + + endpoints = labels(controller, definition) + + assert endpoints["flow:shed.humidity"] == ("shed.humidity", "flow") + + +def test_the_wiring_index_is_read_once_per_store_revision(controller): + reads = 0 + original = controller.store.read_all + + def counted(): + nonlocal reads + reads += 1 + return original() + + controller.store.read_all = counted # type: ignore[method-assign] + _endpoints(controller, controller.store.read_flow("house")) + _endpoints(controller, controller.store.read_flow("garage")) + assert reads == 1 + + controller.store.write_flow(FlowDef(name="shed", nodes=[])) + _endpoints(controller, controller.store.read_flow("house")) + assert reads == 2 diff --git a/frontend/src/components/Flow/FlowEditor.tsx b/frontend/src/components/Flow/FlowEditor.tsx index 5cf9544..28d0f69 100644 --- a/frontend/src/components/Flow/FlowEditor.tsx +++ b/frontend/src/components/Flow/FlowEditor.tsx @@ -61,7 +61,7 @@ import { FIT_VIEW, FIT_VIEW_PANEL, FlowDock } from "./FlowDock" import { FlowNode, type FlowNodeData } from "./FlowNode" import { FlowPanel } from "./FlowPanel" import { LiveEdge } from "./LiveEdge" -import { type Direction, layoutGraph } from "./layout" +import { type Direction, layoutGraph, nodeHeight } from "./layout" import { NodePanel } from "./NodePanel" import { RunDialog } from "./RunDialog" import "./flow.css" @@ -512,9 +512,27 @@ function FlowEditorInner({ ...external.nodes.map((node) => node.id), ] const shapeKey = `${direction}|${key}|${ids.join(",")}` + // How tall each node's ports make it, which `FlowNode` draws to the same + // number. A function of the document — the bindings key above already covers + // every port, so this changes exactly when the layout has to run again, and + // nothing measured is ever fed back into it. + const heights = + direction === "LR" + ? new Map( + definitions.map((node) => [ + node.id, + nodeHeight( + Math.max( + (node.requires ?? []).length, + (node.provides ?? []).length, + ), + ), + ]), + ) + : new Map() // biome-ignore lint/correctness/useExhaustiveDependencies: the shape key is the dependency; the arrays are rebuilt every render. const positions = useMemo( - () => layoutGraph(ids, edges, direction), + () => layoutGraph(ids, edges, direction, heights), [shapeKey, edges], ) diff --git a/frontend/src/components/Flow/FlowNode.tsx b/frontend/src/components/Flow/FlowNode.tsx index 576a31f..c87029c 100644 --- a/frontend/src/components/Flow/FlowNode.tsx +++ b/frontend/src/components/Flow/FlowNode.tsx @@ -32,6 +32,7 @@ import { useIsMobile } from "@/hooks/useMobile" import { duration } from "@/lib/motion" import { cn } from "@/lib/utils" import { portOf } from "./deriveEdges" +import { nodeHeight, PORT_SPAN } from "./layout" import { liveStore, useNodeEmits, @@ -79,7 +80,9 @@ export type FlowNodeData = { /** Spread handles along the node's edge so several ports stay reachable. */ function handleOffset(index: number, total: number): string { if (total <= 1) return "50%" - const span = 60 + // The same fraction `nodeHeight` sizes the node for, so a node is always + // tall enough for the ports this spreads down it. + const span = PORT_SPAN * 100 return `${50 - span / 2 + (span / (total - 1)) * index}%` } @@ -128,6 +131,14 @@ function FlowNodeComponent({ data, selected }: NodeProps) { // The graph runs top to bottom on a phone, so the ports have to face that // way too — see DESIGN-GUIDELINES.md → Responsive. const vertical = useIsMobile() + // Ports run down the sides only while the graph runs across, and that is the + // only direction in which their number decides the height: running downwards + // they spread along the node's width, which is fixed. Read from the document + // rather than measured, so `layoutGraph` can reserve exactly this much. + const ports = Math.max( + (definition.requires ?? []).length, + (definition.provides ?? []).length, + ) // Whether a pulse is playing right now; the ring is mounted only while it is. const [firing, setFiring] = useState(false) // The count outlives this component: the store is module-level, and a @@ -186,8 +197,9 @@ function FlowNodeComponent({ data, selected }: NodeProps) { /> ) : null}
- {running ? ( - - - - - Stop this node - - ) : null} + {/* Each control sits in a slot that is there whether the control + is or not. They come and go with what the node is doing — and a + node running many times a second comes and goes that often — so + in the flex row itself they would resize the card on every + execution, which reads as a flickering shape. */} + + {running ? ( + + + + + Stop this node + + ) : null} + - {(status === "error" || failedEarlier) && onShowLogs ? ( - - - - - - {failedEarlier - ? `Failed at ${failedAt} — show the traceback` - : "Show the traceback"} - - - ) : null} + + {(status === "error" || failedEarlier) && onShowLogs ? ( + + + + + + {failedEarlier + ? `Failed at ${failedAt} — show the traceback` + : "Show the traceback"} + + + ) : null} + - {style ? ( - - - - - - {problem || style.label} - - - ) : null} + + {style ? ( + + + + + + {problem || style.label} + + + ) : null} +
56) + +for (let ports = 2; ports <= 12; ports += 1) { + const height = nodeHeight(ports) + // `handleOffset` spreads the ports over `PORT_SPAN` of the edge, so that is + // the room they actually get. Two handles must not touch. + const between = (PORT_SPAN * height) / (ports - 1) + assert.ok( + between > HANDLE, + `${ports} ports on a ${height}px node leave ${between}px between handles`, + ) + // Growing, never shrinking: one more port can only need more room. + assert.ok(height >= nodeHeight(ports - 1)) +} + +// The layout places top-left corners, so a taller node has to be lifted by its +// own half-height rather than by the default box's. +const tall = nodeHeight(8) +const placed = layoutGraph( + ["a", "b"], + [{ source: "a", target: "b" }], + "LR", + new Map([["b", tall]]), +) +const a = placed.get("a") +const b = placed.get("b") +assert.ok(a && b) +// Dagre centres the two on one rank line, so the taller one starts higher up +// by exactly the difference in half-heights. +assert.equal(Math.round(a.y - b.y), Math.round((tall - 56) / 2)) + +console.log("layout: ok") diff --git a/frontend/src/components/Flow/layout.ts b/frontend/src/components/Flow/layout.ts index ecce134..b3681ac 100644 --- a/frontend/src/components/Flow/layout.ts +++ b/frontend/src/components/Flow/layout.ts @@ -15,8 +15,33 @@ export type Direction = "LR" | "TB" /** `FlowNode` is `min-w-[168px] max-w-[220px]`; an endpoint is narrower. */ const NODE_W = 220 -/** Icon row plus two text lines, as measured. */ +/** Icon row plus two text lines, as measured. The floor, not the height. */ const NODE_H = 56 +/** + * How much of a node's edge the ports are spread over, as a fraction. + * + * Shared with `FlowNode`'s `handleOffset`, which does the spreading: the + * height below is chosen so that span has room for the ports, so the two + * cannot be allowed to drift apart. + */ +export const PORT_SPAN = 0.6 +/** + * Centre to centre between two handles. They are 12px across (`flow.css`), so + * this leaves 6px of rim between them. + */ +const PORT_PITCH = 18 + +/** + * How tall a node with this many ports on one side has to be. + * + * A pure function of the document — the ports are declared, so the height is + * known before anything is drawn. That is what keeps it out of the measuring + * problem below: nothing is fed back, so there is nothing to oscillate. + */ +export function nodeHeight(ports: number): number { + return Math.max(NODE_H, Math.ceil(((ports - 1) * PORT_PITCH) / PORT_SPAN)) +} + /** * Room for the live value an edge carries (`LiveEdge`'s chip is * `max-w-[140px]`). Reserved on the edge itself, so dagre routes nodes around @@ -58,6 +83,7 @@ function build( edges: { source: string; target: string }[], direction: Direction, wrap: Wrap[], + heights: Map, ) { const graph = new dagre.graphlib.Graph() graph.setDefaultEdgeLabel(() => ({})) @@ -74,7 +100,7 @@ function build( // Insertion order is what makes the result deterministic, so it follows the // document rather than whatever order the edges happen to mention nodes in. for (const id of ids) { - graph.setNode(id, { width: NODE_W, height: NODE_H }) + graph.setNode(id, { width: NODE_W, height: heights.get(id) ?? NODE_H }) } for (const edge of edges) { if (!graph.hasNode(edge.source) || !graph.hasNode(edge.target)) continue @@ -120,16 +146,21 @@ function wrapWideRanks(graph: ReturnType, ids: string[]): Wrap[] { /** * Lay the graph out and return each node's top-left corner. * - * ponytail: every node is treated as 220×56 rather than measured. Measuring - * would feed the result back into the layout and oscillate; if nodes ever grow - * past that box, take the sizes from `node.measured` once they have settled. + * `heights` is what a node's ports make it, from {@link nodeHeight}; anything + * left out is the plain box. It is read from the document rather than from the + * canvas on purpose — see that function. + * + * ponytail: every node is still treated as 220 wide rather than measured. + * Measuring would feed the result back into the layout and oscillate; if nodes + * ever grow past that, take the width from `node.measured` once it has settled. */ export function layoutGraph( ids: string[], edges: { source: string; target: string }[], direction: Direction, + heights: Map = new Map(), ): Map { - let graph = build(ids, edges, direction, []) + let graph = build(ids, edges, direction, [], heights) // Running downwards, a rank wider than the screen is the one thing the // layout can still do something about. Wrapping one rank pushes whatever was @@ -142,7 +173,7 @@ export function layoutGraph( const more = wrapWideRanks(graph, ids) if (!more.length) break wrap.push(...more) - graph = build(ids, edges, direction, wrap) + graph = build(ids, edges, direction, wrap, heights) } } @@ -153,7 +184,7 @@ export function layoutGraph( return [ id, node - ? { x: node.x - NODE_W / 2, y: node.y - NODE_H / 2 } + ? { x: node.x - NODE_W / 2, y: node.y - node.height / 2 } : { x: 0, y: 0 }, ] }),