Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
40eaf026c078a8b489bfbd490f9ea27f7b14c99c
app/backend/fluksio/flow/controller.py
T

2493 lines
101 KiB
Python
Raw Blame History

"""Turns stored flows into a running pipeline.
The controller is the only thing that builds nodes: it reads flow definitions
from the store, instantiates each node from its type, and puts them in the
shared pipeline. A node that fails to load is reported and skipped — one
broken node never stops the rest.
Two scopes: `reload` builds every flow, and `reload_flow` swaps one flow's
nodes into the graph and leaves every other flow connected. Deploying takes
the second — the cost of a rebuild is each node opening its socket again, and
one flow's are all that changed. Starting one is neither: a stopped flow's
nodes are built like any other's, so `set_enabled` only starts or stops them.
"""
from __future__ import annotations
import ast
import asyncio
import hashlib
import json
import logging
import threading
import time
import traceback
from collections import Counter
from collections.abc import Callable
from contextlib import suppress
from dataclasses import dataclass, field, replace
from enum import Enum
from typing import Any, TypeGuard, cast
from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool
from fluksio_worker.worker_main import load_function
from fluksio.core.config import settings
from fluksio.flow.alerts import AlertManager
from fluksio.flow.demand import demand
from fluksio.flow.events import EventBus
from fluksio.flow.executor import ExecutionService
from fluksio.flow.messages import MessageSpec, flow_of, qualify
from fluksio.flow.namespace import Rename, touches
from fluksio.flow.nodes import (
RESERVED_SETTINGS,
AssistantNode,
ChangeNode,
DebounceNode,
DelayNode,
DeviceNode,
ExecNode,
FileNode,
HttpNode,
InfluxDbNode,
InjectNode,
JoinNode,
MLPNode,
MqttNode,
Node,
NtfyNode,
RbeNode,
ServiceNode,
SttNode,
SwitchNode,
TtsNode,
WebSocketNode,
WrenNode,
)
from fluksio.flow.pipeline import (
EngineHeld,
NodeOutcome,
Pipeline,
RunCacheLookup,
ValidationIssue,
ValueSource,
)
from fluksio.flow.placement import Placer
from fluksio.flow.queue import WorkItem, owned_by
from fluksio.flow.remote import RemoteWorkerHub
from fluksio.flow.resources import (
ResourceAccountant,
UnknownFlavor,
derive_env,
resolve_flavor,
)
from fluksio.flow.schemas import (
BrainEdge,
BrainGraph,
BrainNode,
FlowDef,
Health,
NodeDef,
NodeStatusPublic,
NodeTypeInfo,
Resources,
)
from fluksio.flow.secrets import SecretNotFound, resolve_params
from fluksio.flow.state import MemoryState, StateBackend
from fluksio.flow.store import LIB_DIR, FlowNotFound, FlowStore, LibNotFound
from fluksio.flow.supervision import Supervisor
from fluksio.flow.workers import PythonWorkerPool
logger = logging.getLogger(__name__)
HOOK_PREFIX = "/hooks"
def _is_hook(node: Node | None) -> TypeGuard[HttpNode | WebSocketNode]:
"""Mounted under HOOK_PREFIX: a webhook, or a socket the engine serves."""
return (isinstance(node, HttpNode) and node.mode is HttpNode.Mode.LISTENER) or (
isinstance(node, WebSocketNode) and node.serves
)
# How long a node gets to close what it opened before the rebuild moves on.
# A node's `stop` talks to whatever it connected to, and a broker that has gone
# away can leave it waiting for an acknowledgement that never arrives — which
# used to hold the rebuild, and everything queued behind it, forever.
NODE_STOP_TIMEOUT = settings.FLOW_NODE_STOP_TIMEOUT
# How long a rebuild asked for by a request waits for one already running.
# Generous on purpose: a rebuild of a populated instance reconnects every
# node and takes the better part of ten seconds, and a caller queued behind a
# healthy one of those should not be turned away. Past that the controller is
# wedged rather than busy, and an error the caller can act on beats a request
# that never ends.
REBUILD_WAIT = settings.FLOW_PIPELINE_REBUILD_WAIT
# How long a caller waits for a *flow* rebuild that is already running. A
# per-flow rebuild reconnects one flow's nodes rather than the instance's,
# so this is a queueing budget — several of them back to back, which is what
# seeding does — not the room a single one needs. Fifteen seconds was sized for
# the whole-pipeline rebuild and would let a wedge sit unreported.
FLOW_REBUILD_WAIT = settings.FLOW_REBUILD_WAIT
# How long a node has to have been down before it is shown as down. A device
# that misses one poll and comes back is a flake, not a fault, and without this
# it turned the health badge amber and the node on the canvas red for a couple
# of seconds. Only going *down* is held back: recovery clears at once, since
# calling a working thing broken is the worse of the two mistakes to leave on
# screen. The stored health is never delayed — see `unhealthy_nodes`.
HEALTH_FLOOR = 60.0
class RebuildBusy(RuntimeError):
"""A rebuild could not start because the one before it has not finished.
Answered as a 503: nothing is wrong with the request, the engine is busy.
"""
class NodesNotLoaded(RuntimeError):
"""A run's flow has nodes that could not be built, so none of it runs."""
def __init__(self, errors: dict[str, str]) -> None:
#: The load error of each node, by its qualified id.
self.errors = errors
node, error = next(iter(errors.items()))
more = f" and {len(errors) - 1} more" if len(errors) > 1 else ""
super().__init__(f"Node '{node}'{more} could not be loaded: {error}")
class NodeStatus(str, Enum):
ACTIVE = "active"
ERROR = "error"
@dataclass
class LoadedNode:
"""A node as the editor sees it: built, or failed with a reason.
``status`` says whether it loaded; ``health`` says how its connection is
doing once it is running. A node can be perfectly loaded and unable to
reach its broker.
"""
id: str
#: The flow's key, which ``id`` is qualified with.
flow: str
#: The flow's id, which its store paths and API routes use.
flow_id: str = ""
status: NodeStatus = NodeStatus.ACTIVE
node: Node | None = None
error: str | None = None
health: Health = "ok"
health_detail: str | None = None
#: When `health` last became what it is, for the floor `unhealthy_nodes`
#: applies. A detail rotating under an unchanged status does not move it.
#: Zero reads as "for as long as anyone knows", so a node that starts out
#: down is shown straight away rather than waiting out a floor from a
#: transition that never happened.
health_since: float = 0.0
#: The last time this node raised while running, and what it said. Kept
#: after it has run again — a failure nobody saw is the one worth keeping —
#: so only an acknowledgement clears it, not a good run and not a rebuild.
last_error: str = ""
last_error_ts: float | None = None
#: Failures since it last ran well, so one bad input reads differently from
#: a node that fails every time. The summary goes by this, not by the age
#: of `last_error_ts`.
failures_in_a_row: int = 0
#: When it last ran well after failing: the recovery the counter above
#: was reset by.
last_ok_ts: float | None = None
#: Published, and with no body of its own — so what runs is the new-node
#: template, which returns nothing at all.
missing_source: bool = False
#: The worker this node names although its type never runs on one.
ignored_worker: str = ""
def inherit_failure(self, previous: LoadedNode) -> None:
"""Keep what the node this one replaces last failed with."""
self.last_error = previous.last_error
self.last_error_ts = previous.last_error_ts
self.failures_in_a_row = previous.failures_in_a_row
self.last_ok_ts = previous.last_ok_ts
@dataclass
class RunContext:
"""The run a set of nodes is being built for.
A run builds nodes of its own, so which run a node's worker call belongs to
is bound when the proxy is made rather than looked up while it runs. That
is what lets one run's node be cancelled without touching the same node in
another.
"""
run_id: str
#: What this run records as the code it ran: its nodes' digests, hashed.
code_digest: str = ""
#: Per node, what the modules that node's function reaches hashed to when
#: the run was claimed. This is what a fingerprint is keyed on, so editing
#: a function the node calls into invalidates the cache the way editing
#: the node itself does — and editing something it does not reach leaves
#: the hit standing. Empty for a node synced before this was recorded,
#: which falls back to `code_digest`.
node_digests: dict[str, str] = field(default_factory=dict)
@dataclass(frozen=True, slots=True)
class Wiring:
"""One published node and the qualified messages it is bound to."""
#: The flow's key, which the messages are qualified with, and its id.
flow: str
node: str
type: str
provides: list[str]
requires: list[str]
flow_id: str = ""
#: Nothing held, as distinct from a node that yielded ``None``.
_NOTHING = object()
class EmitSink:
"""Turns a worker's mid-call frames back into the node's own outputs.
The proxy has to exist before the node that runs it does, so this stands
between them: built empty, handed to the proxy, and pointed at the node as
soon as there is one. What arrives is a dict keyed by output port, which is
the same thing a return value is — so it goes through the node, gets
checked against the ports it declared, and is published from there.
A generator's yields arrive here one at a time, and the last of them is the
node's *result* rather than an emission when the generator returns nothing
of its own. Which one is last is only known once the call ends, so a yield
is checked against the ports on arrival and published one behind;
:meth:`wrap` is what decides afterwards which the held one was.
"""
__slots__ = ("node", "_held")
def __init__(self) -> None:
self.node: Node | None = None
# One thread serves one call, and the same node may be running on
# another thread for another run.
self._held = threading.local()
def _take(self) -> Any:
"""What is held for this thread, and nothing held afterwards."""
held = getattr(self._held, "value", _NOTHING)
self._held.value = _NOTHING
return held
def handle(self, event: dict[str, Any]) -> None:
if self.node is None:
return
kind = event.get("event")
if kind == "emit":
# ``fluksio.emit()``, which is a publication and nothing else — and
# every yield of a worker old enough to have drained its own.
self.node.emit(event.get("outputs") or {})
return
if kind != "yield":
return
outputs = event.get("outputs")
held = self._take()
if held is not _NOTHING:
# A yield with one after it was an emission, whatever the node
# goes on to return.
self.node.emit(held)
# Checked here rather than where it is published: an undeclared port
# fails the call at the yield that produced it, not at the next one.
self.node._to_messages(outputs)
self._held.value = outputs
def wrap(self, call: Callable[..., Any]) -> Callable[..., Any]:
"""A worker-backed function, with its generator's last yield accounted for.
The worker sends every yield as it happens and returns whatever its
generator returned, which is ``None`` for a node that only yields. So
the last yield is still here when the call ends, and this is where it
becomes either the node's result or one more emission.
"""
def run(*args: Any, **kwargs: Any) -> Any:
# A call that died mid-generator cannot leave its last yield on
# this thread for the next one to adopt.
self._take()
try:
result = call(*args, **kwargs)
except BaseException:
self._take()
raise
held = self._take()
if held is _NOTHING:
return result
if result is None:
return held
if self.node is not None:
self.node.emit(held)
return result
return run
@dataclass
class Preview:
"""A draft as it would run, without running it."""
nodes: list[NodeStatusPublic]
issues: list[ValidationIssue]
@dataclass
class NodeType:
"""A node type the editor can place on a canvas."""
title: str
description: str
# Constructor signatures differ per node type.
cls: Any
has_source: bool = False
#: Runs code its definition supplies — a source file, or a command among
#: its settings — with everything the engine can reach.
runs_code: bool = False
free_params: bool = False
params_schema: dict[str, Any] = field(default_factory=dict)
#: Which installed package supplied this type, for the ones that are not
#: built in.
plugin: str | None = None
def _schema_of(cls: Any) -> dict[str, Any]:
params = getattr(cls, "Params", None)
return params.model_json_schema() if params is not None else {}
NODE_TYPES: dict[str, NodeType] = {
"python": NodeType(
title="Function",
description="Your own Python code, run on every incoming message.",
cls=Node,
has_source=True,
runs_code=True,
free_params=True,
),
"mqtt": NodeType(
title="MQTT",
description="Listen on topics, or send to them.",
cls=MqttNode,
params_schema=_schema_of(MqttNode),
),
"http": NodeType(
title="HTTP",
description="Listen on a webhook, or send to a URL and pass the reply on.",
cls=HttpNode,
params_schema=_schema_of(HttpNode),
),
"websocket": NodeType(
title="WebSocket",
description="Listen on a socket, or send to it, as a client or a server.",
cls=WebSocketNode,
params_schema=_schema_of(WebSocketNode),
),
"influxdb": NodeType(
title="InfluxDB",
description="Write measurements to a bucket, or read them back.",
cls=InfluxDbNode,
params_schema=_schema_of(InfluxDbNode),
),
"delay": NodeType(
title="Delay & schedule",
description="Hold messages back, limit their rate, or emit on a schedule.",
cls=DelayNode,
params_schema=_schema_of(DelayNode),
),
"mlp": NodeType(
title="Perceptron",
description="A small neural layer over its numeric inputs.",
cls=MLPNode,
params_schema=_schema_of(MLPNode),
),
"wren": NodeType(
title="Wren",
description="Turn a typed command into values for the devices wired to it.",
cls=WrenNode,
params_schema=_schema_of(WrenNode),
),
"assistant": NodeType(
title="Assistant",
description="Answer a request with the AI provider, acting on what is wired to it.",
cls=AssistantNode,
params_schema=_schema_of(AssistantNode),
),
"stt": NodeType(
title="Speech to text",
description="Turn a microphone's audio into text, through a speech server.",
cls=SttNode,
params_schema=_schema_of(SttNode),
),
"tts": NodeType(
title="Text to speech",
description="Speak text as audio for a speaker, through a speech server.",
cls=TtsNode,
params_schema=_schema_of(TtsNode),
),
"inject": NodeType(
title="Inject",
description="Emit a value on request, on a timer, or when the flow starts.",
cls=InjectNode,
params_schema=_schema_of(InjectNode),
),
"switch": NodeType(
title="Switch",
description="Send a value down one branch or another, by rule.",
cls=SwitchNode,
params_schema=_schema_of(SwitchNode),
),
"change": NodeType(
title="Change",
description="Scale, offset, round or map a value on its way past.",
cls=ChangeNode,
params_schema=_schema_of(ChangeNode),
),
"rbe": NodeType(
title="Filter unchanged",
description="Pass a value on only when it has actually changed.",
cls=RbeNode,
params_schema=_schema_of(RbeNode),
),
"join": NodeType(
title="Join",
description="Gather several inputs into one object or list.",
cls=JoinNode,
params_schema=_schema_of(JoinNode),
),
"debounce": NodeType(
title="Debounce",
description="Send one value now and another once things go quiet.",
cls=DebounceNode,
params_schema=_schema_of(DebounceNode),
),
"exec": NodeType(
title="Command",
description="Run a command in the engine's container and read its output.",
cls=ExecNode,
runs_code=True,
params_schema=_schema_of(ExecNode),
),
"service": NodeType(
title="Service",
description="Keep a server running on a worker, and publish its url.",
cls=ServiceNode,
runs_code=True,
params_schema=_schema_of(ServiceNode),
),
"file": NodeType(
title="File",
description="Read a file into the flow, or write one out of it.",
cls=FileNode,
params_schema=_schema_of(FileNode),
),
"ntfy": NodeType(
title="Notification",
description="Push an incoming value to a phone through ntfy.",
cls=NtfyNode,
params_schema=_schema_of(NtfyNode),
),
"device": NodeType(
title="Device",
description="A paired device's own ports: what it publishes, what it follows.",
cls=DeviceNode,
params_schema=_schema_of(DeviceNode),
),
}
def node_type_info() -> list[NodeTypeInfo]:
return [
NodeTypeInfo(
type=key,
title=spec.title,
description=spec.description,
params_schema=spec.params_schema,
has_source=spec.has_source,
runs_code=spec.runs_code,
free_params=spec.free_params,
plugin=spec.plugin,
)
for key, spec in NODE_TYPES.items()
]
class FlowController:
"""Owns the running pipeline and keeps it in step with the store."""
def __init__(
self,
store: FlowStore,
state: StateBackend | None = None,
events: EventBus | None = None,
max_workers: int | None = None,
fastapi_app: FastAPI | None = None,
execution: ExecutionService | None = None,
alerts: AlertManager | None = None,
workers: PythonWorkerPool | None = None,
remote: RemoteWorkerHub | None = None,
resources: ResourceAccountant | None = None,
placer: Placer | None = None,
) -> None:
self.store = store
# Without a pool, python nodes are compiled and run in this process —
# which is what the tests do, and what a bare `Pipeline` has always done.
self.workers = workers
# Workers on other hosts. A node naming no worker never touches it.
self.remote = remote
# This machine's own books. The pool sizes its fair share off them.
self.resources = resources
# Every machine there is, for the nodes that say what they need.
# Without one, a declaration is recorded and nothing is held against it.
self.placer = placer
self.state = state if state is not None else MemoryState()
self.events = events
self.max_workers = max_workers
self.app = fastapi_app
# Without one, every trigger runs inline where it was raised.
self.execution = execution
self.alerts = alerts
# Set by the lifespan once the store exists; the canvas asks the
# controller which dashboards are wired into a flow.
self.dashboards: Any = None
#: The device store, for the same reason: validation checks a device
#: node against the device it names.
self.devices: Any = None
#: Every device's bound ports as of the last build, so a rebuild that
#: moves them can tell the devices to read theirs again.
self._device_wiring: Any = None
self.pipeline: Pipeline | None = None
self.loaded: dict[str, LoadedNode] = {}
self.issues: list[ValidationIssue] = []
self.disabled: set[str] = set()
#: Flows that only run when a run asks them to.
self.batch: set[str] = set()
#: The key each built flow runs under, by its id. What a rebuild of a
#: flow that is gone takes out, since its document can no longer say.
self._flow_keys: dict[str, str] = {}
#: Each flow's declared inputs, kept so rebuilding one flow can
#: validate the whole graph without re-reading every flow off the disk.
self._flow_inputs: dict[str, dict[str, bool]] = {}
self.supervisor = Supervisor(events)
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
# -------------------------------------------------------------------------
async def start(self) -> None:
# Subscribed before anything runs, so no failure falls between the
# first build and someone watching for one.
self._failures = asyncio.create_task(
self._watch_failures(), name="node-failures"
)
# Build first: the consumer must have a pipeline to execute against
# before it claims anything, or work waiting from the last run would be
# taken and dropped — which is the very case the queue exists for.
# Nothing is running to queue behind here, and a lifespan has nobody to
# report a timeout to, so this one build waits however long it needs.
await self.reload(wait=None)
if self.execution is not None:
self.execution.start()
async def stop(self) -> None:
watcher, self._failures = self._failures, None
if watcher is not None:
watcher.cancel()
with suppress(asyncio.CancelledError):
await watcher
await self._teardown()
if self.execution is not None:
await run_in_threadpool(self.execution.stop)
async def _watch_failures(self) -> None:
"""Record every node failure the engine reports, from the event bus.
A node can be run by the live pipeline, by a run's own, or by hand, and
each builds its own graph — but they all report through
``Pipeline.publish_error``, so the bus is the one place they meet.
Which is the point: what is recorded here is the node's, not any
particular pipeline's, and it outlives both the good run that follows
it and the rebuild after that.
"""
if self.events is None:
return
async with self.events.subscribe() as queue:
while True:
event = await queue.get()
kind = event.get("type")
if kind not in ("node_error", "node_executed"):
continue
entry = self.loaded.get(str(event.get("node") or ""))
if entry is None:
continue
ts = float(event.get("ts") or time.time())
if kind == "node_executed":
# Almost every event here, so only a node with a failure
# standing is touched: running well is how it recovers.
if entry.failures_in_a_row:
entry.failures_in_a_row = 0
entry.last_ok_ts = ts
continue
entry.last_error = str(event.get("error") or "")
entry.last_error_ts = ts
entry.failures_in_a_row += 1
def acknowledge_error(self, node_id: str) -> None:
"""Forget what a node's last failure was. The only thing that clears it.
Raises ``KeyError`` for a node the engine does not have, so a route can
answer that the way every other node call does.
"""
entry = self.loaded.get(node_id)
if entry is None:
raise KeyError(node_id)
entry.last_error = ""
entry.last_error_ts = None
entry.failures_in_a_row = 0
async def set_enabled(self, flow_id: str, flow: str, enabled: bool) -> None:
"""Stop or start one flow, without building anything.
Its nodes are built whether or not it runs — being stopped means
having no subscriptions, schedules or webhooks, not being absent — so
all this has to do is the lifecycle call and the gate that goes with
it. Nothing is compiled and no other flow is touched. *flow_id* is
where that is stored, *flow* the key the gate goes by.
"""
await run_in_threadpool(self.store.write_enabled, flow_id, enabled)
if self.pipeline is None:
# Nothing built yet, so there are no nodes to start or stop.
await self.reload()
return
await self._acquire(FLOW_REBUILD_WAIT)
try:
if enabled:
running = flow not in self.disabled
# The gate first: a subscription that fires the instant it
# starts must not be turned away by a flag on its way out.
self.disabled = self.disabled - {flow}
self.pipeline.set_disabled(self.disabled)
# Lifts any quarantine, which is what a rebuild used to do by
# throwing the whole supervisor away. A running flow's nodes
# are stopped first: a node already started returns from
# start() at once, which would leave its cancelled loop dead.
if running:
await self._teardown(flow)
else:
await self.supervisor.cancel_flow(flow)
await self._activate(flow)
else:
self.disabled = self.disabled | {flow}
self.pipeline.set_disabled(self.disabled)
await self._teardown(flow)
finally:
self._lock.release()
# A stopped flow wires no device ports.
self._rewired()
self._publish(
{
"type": "pipeline_rebuilt",
"issues": [issue.model_dump() for issue in self.issues],
"nodes": [status.model_dump() for status in self.node_statuses()],
# Every flow's, not this one's: the canvas replaces its paused
# set from this, so a narrowed list would clear the markers of
# flows this never touched.
"paused": self.paused_flows(),
}
)
async def _acquire(self, wait: float | None) -> None:
"""Take the rebuild lock, or say the engine is busy.
Only one rebuild runs at a time whatever its scope. A caller waits
*wait* seconds for the one in front of it and then gives up with
``RebuildBusy`` — hanging on a rebuild that is stuck is worse than
saying so. ``None`` waits.
"""
if wait is None:
await self._lock.acquire()
return
try:
await asyncio.wait_for(self._lock.acquire(), wait)
except TimeoutError:
raise RebuildBusy(
f"A pipeline rebuild is still running after {wait:.0f}s"
) from None
async def reload(self, wait: float | None = REBUILD_WAIT) -> None:
"""Rebuild the whole pipeline from what is currently stored."""
await self._acquire(wait)
try:
# Work already claimed belongs to the pipeline it was claimed
# against; let it finish there before swapping the graph out.
# Work that does not finish in time keeps the old graph, which
# nothing has touched yet.
if self.execution is not None and not await run_in_threadpool(
self.execution.pause_intake
):
self.execution.resume_intake()
raise RebuildBusy("Claimed work is still running on the pipeline")
await self._teardown()
# A fresh supervisor per build, so a flow quarantined by the last
# one gets another chance once its author has changed something.
self.supervisor = Supervisor(self.events)
published = self.store.read_all()
self._flow_keys = {flow.id: flow.name for flow in published}
self.disabled = {
flow.name for flow in published if not self.store.read_enabled(flow.id)
}
# A batch flow is built — the canvas draws it and validation covers
# it — but never activated: it runs when a run asks it to, and a
# subscription of its own would be a second way in.
self.batch = {flow.name for flow in published if flow.mode == "batch"}
# Off the loop: building a python node asks its worker to compile,
# which waits for a free slot — and a busy node holds one for as
# long as its timeout. On the loop that stalls every request, the
# websocket and the health check the container is judged by.
nodes, loaded, initial_values, flow_inputs = await run_in_threadpool(
self._build_flows, [(flow, False) for flow in published]
)
self._flow_inputs = {
flow.name: _declared_inputs(flow)[0] for flow in published
}
# A rebuild is a fresh set of nodes, but not a fresh history:
# dropping the failures here would wipe them on every rebuild,
# and they are the operator's to dismiss.
for node_id, entry in loaded.items():
previous = self.loaded.get(node_id)
if previous is not None and previous.last_error:
entry.inherit_failure(previous)
self.loaded = loaded
self.pipeline = Pipeline(
nodes=nodes,
state=self.state,
events=self.events,
max_workers=self.max_workers,
initial_values=initial_values,
disabled_flows=self.disabled,
work_queue=self.execution.queue if self.execution else None,
node_pool=self.execution.node_pool if self.execution else None,
)
self.pipeline.history_limits = self.history_limits
if self.execution is not None:
self.execution.bind(self.pipeline)
self.issues = _collect_issues(
loaded, self.pipeline, flow_inputs, self.devices
)
await self._activate()
if self.execution is not None:
self.execution.resume_intake()
# A rebuild clears the pause, so no resume will ever come for
# what the old pipeline parked. Release it here or it is lost.
for flow in published:
self._release_parked(flow.name)
finally:
self._lock.release()
self._publish(
{
"type": "pipeline_rebuilt",
"issues": [issue.model_dump() for issue in self.issues],
"nodes": [status.model_dump() for status in self.node_statuses()],
# A rebuild is a fresh pipeline, so nothing is paused any more.
"paused": self.paused_flows(),
}
)
self._rewired()
async def reload_flow(
self, flow_id: str, wait: float | None = FLOW_REBUILD_WAIT
) -> None:
"""Rebuild one flow, leaving every other flow's nodes connected.
A flow is not a subgraph — its nodes can read and write messages
another flow owns — but the wiring is derived from message names, so
swapping one flow's nodes into the graph and deriving the edges again
is enough. What that saves is the reconnecting: the cost of a rebuild
on a populated instance is every node opening its socket again,
and only one flow's have changed.
A flow the store no longer has is taken out instead of replaced. The
flow is named by its id; everything about running it goes by its key.
The failure shape is the whole-pipeline one narrowed to a flow, and
deliberately no better: a build that raises changes nothing, and
anything failing after the teardown leaves that flow's nodes stopped
in the graph — which is exactly what a failed ``reload`` leaves behind
for all of them.
"""
pipeline = self.pipeline
if pipeline is None:
# Nothing built yet, so there is nothing to splice one flow into.
await self.reload(wait=wait)
return
await self._acquire(wait)
try:
# Built before anything is stopped, so the window in which the
# flow is not running is its own teardown and nothing more. Off
# the loop for the same reason the full build is: compiling a
# python node waits for a worker slot.
built = await run_in_threadpool(self._build_one, flow_id)
# A flow never built has nothing to take out, so no key either.
name = built[0].name if built else self._flow_keys.pop(flow_id, "")
await self._teardown(name)
if built is None:
if name:
pipeline.remove_flow(name)
self.loaded = {
node_id: entry
for node_id, entry in self.loaded.items()
if entry.flow != name
}
# Or the flow would go on declaring inputs nothing provides
# any more, and the consumers it left behind would look fine.
self._flow_inputs.pop(name, None)
self.disabled = self.disabled - {name}
self.batch = self.batch - {name}
else:
flow, nodes, loaded, initial_values, enabled = built
self._flow_keys[flow_id] = name
pipeline.replace_flow(name, nodes, initial_values)
# A rebuild is a fresh set of nodes, but not a fresh history:
# a failure nobody has dismissed is the operator's to keep.
for node_id, entry in loaded.items():
previous = self.loaded.get(node_id)
if previous is not None and previous.last_error:
entry.inherit_failure(previous)
# Rebound rather than mutated: request threads and the failure
# watcher read these without holding anything.
self.loaded = {
**{
node_id: entry
for node_id, entry in self.loaded.items()
if entry.flow != name
},
**loaded,
}
self._flow_inputs[name] = _declared_inputs(flow)[0]
self.disabled = (
self.disabled - {name} if enabled else self.disabled | {name}
)
self.batch = (
self.batch | {name} if flow.mode == "batch" else self.batch - {name}
)
pipeline.set_disabled(self.disabled)
# Over the whole graph, because it has to be: taking a producer out
# of one flow is what leaves another flow's input unconnected.
self.issues = _collect_issues(
self.loaded, pipeline, self._all_flow_inputs(), self.devices
)
await self._activate(name)
# This flow's pause is cleared by the replace, so no resume will
# ever come for what it parked. Release it here or it is lost.
self._release_parked(name)
finally:
self._lock.release()
self._publish(
{
"type": "pipeline_rebuilt",
"issues": [issue.model_dump() for issue in self.issues],
"nodes": [status.model_dump() for status in self.node_statuses()],
# Every flow's, not this one's: the canvas replaces its paused
# set from this, so a narrowed list would clear the markers of
# flows the rebuild never touched.
"paused": self.paused_flows(),
}
)
self._rewired()
async def splice_renamed(self, rename: Rename, rebuilt: list[str]) -> None:
"""Put the flows a key rename rewrote back into the graph, built from
their new documents and named as they are now.
Under the controller lock, which the rename holds, with the engine
held and the flows *rebuilt* (ids) torn down before their documents
changed. Not `reload_flow`, which would take the lock again, nor
`reload`, which clears every pause. Stopped flows stay stopped, which
is stored by id, pauses carry over under the new keys and parked work
stays parked. A rebuilt flow's quarantine is gone with its teardown,
as on a publish. What the documents feed is refreshed before anything
starts: history limits, device wiring and what devices are asked to
send.
"""
pipeline = self.pipeline
if pipeline is None:
return
paused = pipeline.paused_flows()
builds = await run_in_threadpool(
lambda: [(flow_id, self._build_one(flow_id)) for flow_id in rebuilt]
)
# By the name each node has now, so a failure carries over.
previous = {rename.name(node_id): e for node_id, e in self.loaded.items()}
loaded = dict(self.loaded)
keys = []
for flow_id, built in builds:
if built is None:
continue
flow, nodes, entries, initial_values, enabled = built
old, key = self._flow_keys.get(flow_id, flow.name), flow.name
if old != key:
pipeline.replace_flow(old, [])
pipeline.replace_flow(key, nodes, initial_values)
self._flow_keys[flow_id] = key
for node_id, entry in entries.items():
before = previous.get(node_id)
if before is not None and before.last_error:
entry.inherit_failure(before)
loaded = {
node_id: entry for node_id, entry in loaded.items() if entry.flow != old
} | entries
self._flow_inputs.pop(old, None)
self._flow_inputs[key] = _declared_inputs(flow)[0]
self.disabled = (self.disabled - {old}) | (set() if enabled else {key})
self.batch = (self.batch - {old}) | (
{key} if flow.mode == "batch" else set()
)
keys.append(key)
self.loaded = loaded
pipeline.set_disabled(self.disabled)
# The replace cleared the pause of every flow it touched.
for was in map(rename.flow, paused):
if not pipeline.is_paused(was):
pipeline.pause(was)
self.issues = _collect_issues(
self.loaded, pipeline, self._all_flow_inputs(), self.devices
)
if self.dashboards is not None:
self.set_history_limits(
await run_in_threadpool(self.dashboards.history_requirements)
)
if self.devices is not None:
# Taken note of without telling the devices: the rename does,
# once it has let go of the engine.
await run_in_threadpool(demand.refresh)
self._device_wiring = await run_in_threadpool(self.devices.wiring)
for key in keys:
await self._activate(key)
async def recheck(self, wait: float | None = FLOW_REBUILD_WAIT) -> None:
"""Validate the built graph again without rebuilding anything.
For what validation reads from outside the flows: a device created or
deleted changes whether a device node names one that exists.
"""
if self.pipeline is None:
return
await self._acquire(wait)
try:
self.issues = _collect_issues(
self.loaded, self.pipeline, self._all_flow_inputs(), self.devices
)
finally:
self._lock.release()
# A device created, deleted or reporting changes whose ports send on
# demand.
demand.refresh()
self._publish(
{
"type": "pipeline_rebuilt",
"issues": [issue.model_dump() for issue in self.issues],
"nodes": [status.model_dump() for status in self.node_statuses()],
"paused": self.paused_flows(),
}
)
def _rewired(self) -> None:
"""Tell the devices when a rebuild moved what their ports are wired to.
An empty dashboard name is what a device refetches its own document
on, and what its socket rescopes on. The first build only takes note:
a device connecting later reads the document as it is.
What a running flow reads may have moved with it, so demand on the
devices' ports is worked out again too.
"""
if self.devices is None:
return
demand.refresh()
wiring = self.devices.wiring()
known, self._device_wiring = self._device_wiring, wiring
if known is not None and known != wiring:
self._publish(
{"type": "dashboard_changed", "dashboard": "", "ts": time.time()}
)
async def reload_failed_flows(self) -> list[str]:
"""Rebuild only the flows holding a node that would not load.
What installing a module wants: a node red because its import was
missing is built again against the packages just installed, and every
other flow keeps the connections it already has.
"""
broken = sorted(
{
entry.flow_id
for entry in self.loaded.values()
if entry.status is NodeStatus.ERROR
}
)
for flow in broken:
await self.reload_flow(flow)
return broken
async def reload_lib_users(self, ref: str) -> list[str]:
"""Rebuild the flows running a shared node whose source has changed.
Returns their ids.
"""
flows = self.store.lib_users(ref)
for flow in flows:
await self.reload_flow(flow)
return flows
def _build_one(
self, flow_id: str
) -> tuple[FlowDef, list[Node], dict[str, LoadedNode], dict[str, Any], bool] | None:
"""Read one published flow and build its nodes. None when it is gone.
Blocking: this is the part of a rebuild that talks to the worker pool.
"""
try:
flow = self.store.read_flow(flow_id)
except FlowNotFound:
return None
nodes, loaded, initial_values, _ = self._build_flows([(flow, False)])
return flow, nodes, loaded, initial_values, self.store.read_enabled(flow_id)
def _all_flow_inputs(self) -> dict[str, bool]:
"""Every flow's declared inputs, merged the way one build would see them.
In name order, which is the order ``read_all`` returns flows in, so a
name two flows both declare resolves to the same one either way.
"""
merged: dict[str, bool] = {}
for name in sorted(self._flow_inputs):
merged.update(self._flow_inputs[name])
return merged
async def _teardown(self, flow: str | None = None) -> None:
"""Stop everything the previous pipeline started, or one flow's share.
Together rather than one after another: each node is given
`NODE_STOP_TIMEOUT`, so in sequence a flow whose broker is unreachable
took that many seconds *per node* — long enough for a rebuild to
outlast `REBUILD_WAIT` and answer 503 on every deploy. Concurrently it
is five seconds flat however many nodes there are.
"""
async def stop(entry: LoadedNode) -> None:
node = entry.node
if node is None:
return
try:
await asyncio.wait_for(node.stop(self.app), NODE_STOP_TIMEOUT)
except TimeoutError:
# Abandoned rather than waited on: the next node still gets to
# close, and the rebuild still happens.
logger.warning(
"Node '%s' did not stop within %.0fs — carrying on without it",
entry.id,
NODE_STOP_TIMEOUT,
)
except Exception:
logger.exception("Error stopping node '%s'", entry.id)
await asyncio.gather(
*(
stop(entry)
for entry in list(self.loaded.values())
if flow is None or entry.flow == flow
)
)
# After the nodes, so a loop still winding down is not restarted.
if flow is None:
await self.supervisor.cancel_all()
else:
await self.supervisor.cancel_flow(flow)
async def _activate(self, flow: str | None = None) -> None:
"""Start subscriptions, schedules and webhooks of the new pipeline.
What that means per node type is the node's own business — a connector
written against the contract starts here the same way the built-ins do.
"""
for entry in self.loaded.values():
node = entry.node
if node is None:
continue
if flow is not None and entry.flow != flow:
continue
# A stopped flow gets no subscriptions, schedules or webhooks —
# that is what stopping it means. Nor does a batch flow, which has
# no outside to listen to.
if entry.flow in self.disabled or entry.flow in self.batch:
continue
node.supervisor = self.supervisor
try:
await node.start(self.app)
except Exception as exc:
logger.exception("Error starting node '%s'", entry.id)
entry.status = NodeStatus.ERROR
entry.error = f"{type(exc).__name__}: {exc}"
# -------------------------------------------------------------------------
# Building
# -------------------------------------------------------------------------
def _runs_elsewhere(self, wanted: Resources | None) -> bool:
"""Whether this node asks for more than this machine could ever give."""
if wanted is None or self.placer is None:
return False
return not self.placer.local.fits(wanted.cpus, wanted.gpus, wanted.ram or 0)
def _placed(
self,
wanted: Resources,
label: str | None,
policy: str,
owner: str,
local: str,
code: str,
*,
node_id: str,
timeout: float,
run_id: str,
on_event: Callable[[dict[str, Any]], None],
) -> Callable[..., Any]:
"""A call that picks a machine and holds its share while it runs.
Which machine is decided per call rather than when the flow was built,
so a worker that attaches later is used without anything being rebuilt.
The order is load-bearing: the resources are claimed first, and only
then is a worker slot taken. The other way round, a node holding a slot
could sit waiting for cores that a node holding the cores cannot get a
slot to release.
What the allocation implies is handed to the process the node runs in,
here or on the worker, because a library reads those variables when it
is imported and never again.
"""
placer, pool = self.placer, self.workers
def call(**kwargs: Any) -> Any:
with placer.claim( # type: ignore[union-attr]
wanted, label=label, policy=policy, node=node_id, run=run_id
) as (target, allocation):
env = derive_env(wanted, allocation, local=target.worker is None)
if target.worker is None:
return pool.for_env(env).run( # type: ignore[union-attr]
owner,
local,
code,
kwargs,
node_id,
timeout,
run_id=run_id,
on_event=on_event,
)
return self.remote.run_on( # type: ignore[union-attr]
target.worker,
owner,
local,
code,
kwargs,
node_id,
timeout,
run_id=run_id,
on_event=on_event,
env=env,
)
return call
def _build_flows(
self,
flows: list[tuple[FlowDef, bool]],
run: RunContext | None = None,
) -> tuple[list[Node], dict[str, LoadedNode], dict[str, Any], dict[str, bool]]:
"""Instantiate the nodes of several flows, each published or draft."""
nodes: list[Node] = []
loaded: dict[str, LoadedNode] = {}
initial_values: dict[str, Any] = {}
# Declared flow inputs, mapped to whether a value will be there.
flow_inputs: dict[str, bool] = {}
for flow, draft in flows:
for node_def in flow.nodes:
entry = self._build_node(flow, node_def, draft=draft, run=run)
loaded[entry.id] = entry
if entry.node is not None:
nodes.append(entry.node)
declared, initial = _declared_inputs(flow)
flow_inputs.update(declared)
initial_values.update(initial)
return nodes, loaded, initial_values, flow_inputs
def _build_node(
self,
flow_def: FlowDef,
node_def: NodeDef,
draft: bool = False,
run: RunContext | None = None,
) -> LoadedNode:
"""One node of *flow_def*: its code read by the flow's id, and run,
named and wired by its key."""
flow, flow_id = flow_def.name, flow_def.id
node_id = f"{flow}.{node_def.id}"
entry = LoadedNode(id=node_id, flow=flow, flow_id=flow_id)
try:
node_type = NODE_TYPES.get(node_def.type)
if node_type is None:
raise ValueError(f"Unknown node type '{node_def.type}'")
if node_def.worker and not node_type.has_source:
entry.ignored_worker = node_def.worker
params = resolve_params(node_def.params)
if node_type.has_source:
# Settings and ports are both keyword arguments of the same
# function, so one name cannot mean both.
ports = {spec.port for spec in _bound(node_def.requires)}
clash = sorted((set(params) - RESERVED_SETTINGS) & ports)
if clash:
raise ValueError(
f"'{clash[0]}' is both an input and a setting of "
f"'{node_def.id}'. A setting is an argument like a "
"port, so rename one of them."
)
# A shared node runs the library's copy, compiled once under
# the library's own name so every flow using it agrees.
if node_def.source_ref:
owner, local = LIB_DIR, node_def.source_ref
code = self.store.read_lib_source(node_def.source_ref)
else:
owner, local = flow, node_def.id
code = self.store.read_node_source(
flow_id, node_def.id, draft=draft
)
# The store answers the new-node template when nothing was
# written, which is what an editor should open with and not
# something to run. A draft legitimately has none yet.
entry.missing_source = not draft and not self.store.has_node_source(
flow_id, node_def.id
)
# Only a body somebody wrote: the template a new node opens
# with takes nothing, and a node declared before it is written
# is reported as missing its source rather than as wrong.
written = bool(node_def.source_ref) or self.store.has_node_source(
flow_id, node_def.id, draft=draft
)
if written:
gap = _signature_gap(
code, node_def.id, ports, set(params) - RESERVED_SETTINGS
)
if gap:
raise ValueError(gap)
# What a node produces before it returns comes back as frames;
# this puts them through the node's own ports.
emissions = EmitSink()
if self.workers is None:
function = load_function(owner, local, code)
else:
# The code never runs here: it is loaded in a worker, and
# the node calls that worker instead of a local function.
# A node bound to a worker is loaded on *that* machine —
# one importing torch is correct on the GPU box and a
# missing module here, so checking it here would fail a
# node that is fine.
try:
# A named size is read now rather than stored on the
# node, so editing the flavor edits the next run.
wanted = (
resolve_flavor(node_def.resources)
if node_def.resources is not None
else None
)
except UnknownFlavor as exc:
entry.status = NodeStatus.ERROR
entry.error = str(exc)
return entry
remote_only = (
node_def.worker
and node_def.worker_policy == "require"
and self.remote is not None
)
if remote_only and self.remote is not None:
problem = self.remote.compile(
node_def.worker or "", owner, local, code
)
elif self._runs_elsewhere(wanted):
# Asks for more than this machine has, so it will run on
# one that has it. Same reason as a worker label: checking the
# import here would fail a node that is fine there.
problem = None
else:
problem = self.workers.compile(owner, local, code)
if problem:
entry.status = NodeStatus.ERROR
entry.error = problem
return entry
# Not ``or``: 0 is a node saying it has no limit, which is
# exactly the value that would fall through to the default.
timeout = (
node_def.timeout
if node_def.timeout is not None
else settings.FLOW_NODE_TIMEOUT
)
if wanted is not None and self.placer is not None:
# Says how much of a machine it takes, so which machine
# and how much of it are one decision — including when
# it also names a worker, which used to mean the two
# answers disagreed and nothing was accounted at all.
function = self._placed(
wanted,
node_def.worker,
node_def.worker_policy,
owner,
local,
code,
node_id=node_id,
timeout=timeout,
run_id=run.run_id if run else "",
on_event=emissions.handle,
)
elif node_def.worker and self.remote is not None:
# A worker and nothing about size: the label alone
# decides, least busy first, as it always has.
function = self.remote.proxy(
node_def.worker,
owner,
local,
code,
node_id=node_id,
timeout=timeout,
run_id=run.run_id if run else "",
on_event=emissions.handle,
fallback=(
self.workers.proxy(
owner,
local,
code,
node_id=node_id,
timeout=timeout,
run_id=run.run_id if run else "",
on_event=emissions.handle,
)
if node_def.worker_policy == "prefer"
else None
),
)
else:
# Declares nothing: the shared pool, at no extra cost.
function = self.workers.proxy(
owner,
local,
code,
node_id=node_id,
timeout=timeout,
run_id=run.run_id if run else "",
on_event=emissions.handle,
)
# Outermost, so it sees the result whichever of the three
# above answered the call.
function = emissions.wrap(function)
node = Node(
f=with_settings(function, params),
requires=_bound(node_def.requires),
provides=_bound(node_def.provides),
params=params,
name=node_def.id,
)
if run is not None and node_def.cache:
# The raw params, not the resolved ones: a secret's value
# must not end up in a key, and its name is what changes
# when the node is reconfigured anyway.
#
# `source` is the stored body, which for a code-defined
# flow is a shim that imports the real function — it does
# not move when that function does, and it says nothing
# about what the function calls into. `code` is what
# covers both: what the modules this node's function
# reaches hash to, which `sync` worked out by following
# its imports. The ports are here because declaring a new
# one changes what the node answers with, which a hit
# would otherwise restore without. Changing what goes in
# here changes every key: move `CACHE_SHAPE_SINCE` with it.
node.fingerprint = hashlib.sha256(
json.dumps(
{
"source": code,
"params": node_def.params,
"requires": [
spec.model_dump(mode="json")
for spec in node_def.requires
],
"provides": [
spec.model_dump(mode="json")
for spec in node_def.provides
],
"code": run.node_digests.get(node_def.id)
or run.code_digest,
},
sort_keys=True,
separators=(",", ":"),
).encode()
).hexdigest()
emissions.node = node
else:
node = node_type.cls(
requires=_bound(node_def.requires),
provides=_bound(node_def.provides),
params=params,
name=node_def.id,
)
node.assign_flow(flow, node_def.id)
node._on_health = self._health_changed
if _is_hook(node):
# Hooks live under their flow, away from the JSON API: under
# its id, so a URL handed out keeps working when the key
# changes.
node.url = f"{HOOK_PREFIX}/{flow_id}/{node.url.lstrip('/')}"
entry.node = node
except (SecretNotFound, LibNotFound) as exc:
entry.status = NodeStatus.ERROR
entry.error = str(exc)
except Exception as exc:
logger.warning("Node '%s' failed to load: %s", node_id, exc)
entry.status = NodeStatus.ERROR
entry.error = _short_error(exc)
return entry
# -------------------------------------------------------------------------
# Queries
# -------------------------------------------------------------------------
def node_statuses(self, flow: str | None = None) -> list[NodeStatusPublic]:
return [
NodeStatusPublic(
id=entry.id,
status=entry.status.value,
error=entry.error,
health=entry.health,
health_detail=entry.health_detail,
last_error=entry.last_error,
last_error_ts=entry.last_error_ts,
)
for entry in self.loaded.values()
if flow is None or entry.flow == flow
]
def _health_changed(self, node: Node, status: str, detail: str | None) -> None:
"""A node saying how its connection is doing, from any thread."""
entry = self.loaded.get(node.id)
if entry is None or (entry.health == status and entry.health_detail == detail):
return
if entry.health != status:
# Only a real transition restarts the clock. A connector retrying
# with a different errno each poll rotates the detail while staying
# down, and moving the mark for that would keep the node forever
# under `HEALTH_FLOOR` — debouncing nothing at all.
entry.health_since = time.time()
entry.health = cast(Health, status)
entry.health_detail = detail
self._publish(
{
"type": "node_health",
"flow": node.flow,
"node": node.id,
"health": status,
"detail": detail,
}
)
def unhealthy_nodes(
self, flow: str | None = None, health: Health = "down"
) -> list[LoadedNode]:
"""Nodes that have been down long enough to be worth telling anyone.
Or degraded, with ``health``: working, but not as well as they should,
which is a warning rather than a fault.
The floor is on the surfacing, not on the transition: `entry.health` is
the truth the moment the node reports it, so a flow reading a node's
health is never told a stale story — only the screens wait. And only
for the way down; a node reading "ok" leaves this list immediately,
whenever it recovered.
"""
now = time.time()
return [
entry
for entry in self.loaded.values()
if entry.health == health
and now - entry.health_since >= HEALTH_FLOOR
and (flow is None or entry.flow == flow)
]
def _health_issues(self, flow: str | None = None) -> list[ValidationIssue]:
"""Nodes that are running but not working, as issues on their flow.
A degraded node is an advisory one: it works, if not as it should, such
as a device node whose device is switched off.
Not part of `self.issues`: that list is what a build found, and this is
what is happening now. Derived on read, so a node reporting itself well
again clears it with nothing to remember.
"""
return [
ValidationIssue(
code="node_unhealthy",
message=(
f"Node '{entry.id.rpartition('.')[2]}' is down: "
f"{entry.health_detail or 'no detail given'}"
),
flow=entry.flow,
node=entry.id,
)
for entry in self.unhealthy_nodes(flow)
] + [
ValidationIssue(
code="node_degraded",
message=(
f"Node '{entry.id.rpartition('.')[2]}' is degraded: "
f"{entry.health_detail or 'no detail given'}"
),
flow=entry.flow,
node=entry.id,
)
for entry in self.unhealthy_nodes(flow, "degraded")
]
def flow_issues(self, flow: str) -> list[ValidationIssue]:
return [
issue for issue in self.issues if not issue.flow or issue.flow == flow
] + self._health_issues(flow)
def preview(self, flow_id: str) -> Preview:
"""Build a flow's unpublished draft without deploying it.
The draft stands in for its published self among the other flows, so
cross-flow messages resolve the way they would after publishing. The
graph is thrown away afterwards and seeded into a throwaway state, so
nothing here touches what is running.
"""
try:
draft = self.store.read_flow(flow_id, draft=True)
except FlowNotFound:
return Preview(nodes=[], issues=[])
name = draft.name
flows: list[tuple[FlowDef, bool]] = [
(flow, False) for flow in self.store.read_all() if flow.id != flow_id
]
flows.append((draft, True))
nodes, loaded, initial_values, flow_inputs = self._build_flows(flows)
pipeline = Pipeline(
nodes=nodes,
state=MemoryState(),
max_workers=self.max_workers,
initial_values=initial_values,
)
issues = _collect_issues(loaded, pipeline, flow_inputs, self.devices)
return Preview(
nodes=[
NodeStatusPublic(
id=entry.id, status=entry.status.value, error=entry.error
)
for entry in loaded.values()
if entry.flow == name
],
issues=[issue for issue in issues if not issue.flow or issue.flow == name],
)
def compile_check(
self, flow: str, node_id: str, code: str, worker: str | None = None
) -> str | None:
"""Does this source load? Returns what to show the author, or None.
Loaded on the machine that will run it: a node bound to a worker is
checked against that worker's packages, because a missing import here
says nothing about whether it is missing there.
The source is usually a draft, and the node is usually still running
the published one — so ``keep=False``: compile it to answer the
question, and leave the worker holding what it is serving calls from.
"""
if worker and self.remote is not None:
return self.remote.compile(worker, flow, node_id, code)
if self.workers is not None:
return self.workers.compile(flow, node_id, code, keep=False)
try:
load_function(flow, node_id, code)
except Exception as exc:
return _short_error(exc)
return None
def value(self, name: str) -> dict[str, Any]:
return self.pipeline.value(name) if self.pipeline else {}
def values(self, flow: str | None = None) -> dict[str, dict[str, Any]]:
return self.pipeline.values(flow) if self.pipeline else {}
def consumed_messages(self) -> set[str]:
"""The messages a node of a running flow reads.
What keeps a device's on-demand port sending whether or not anyone is
watching: a node that reads a camera wants every frame. A stopped or
batch flow reads nothing until it runs, and a device node is passive.
"""
pipeline = self.pipeline
if pipeline is None:
return set()
idle = self.disabled | self.batch
return {
name
for name, nodes in pipeline.consumes.items()
if any(node.flow not in idle for node in nodes)
}
def get_node(self, node_id: str) -> Node | None:
entry = self.loaded.get(node_id)
return entry.node if entry else None
# -------------------------------------------------------------------------
# Execution
# -------------------------------------------------------------------------
def is_enabled(self, flow: str) -> bool:
return flow not in self.disabled
def is_paused(self, flow: str) -> bool:
return self.pipeline is not None and flow in self.pipeline.paused_flows()
def is_quarantined(self, flow: str) -> bool:
return flow in self.supervisor.quarantined
@property
def quarantined(self) -> set[str]:
return self.supervisor.quarantined
def paused_flows(self) -> list[str]:
return self.pipeline.paused_flows() if self.pipeline else []
def pause_flow(self, flow: str) -> None:
if self.pipeline is not None:
self.pipeline.pause(flow)
def resume_flow(self, flow: str) -> None:
"""Blocking — call from a worker thread: held-back nodes run on resume."""
if self.pipeline is None:
return
self.pipeline.resume(flow)
self._release_parked(flow)
def _release_parked(self, flow: str) -> None:
"""Queue what a pause held back again, oldest first, so it is not lost."""
if self.execution is None:
return
for item in self.execution.queue.unpark(flow):
self.execution.queue.add(item)
def step_flow(self, flow: str) -> str | None:
"""Run one held-back item, leaving the flow paused. Blocking.
Returns the node it came from, or None when nothing is held back.
"""
return self.execution.step(flow) if self.execution is not None else None
def set_history_limits(self, limits: dict[str, int]) -> None:
"""How deep to keep each charted message's series. Applies at once."""
self.history_limits = limits
if self.pipeline is not None:
self.pipeline.history_limits = limits
def message_catalog(self) -> list[Any]:
"""Every message the published flows declare, with its last value.
What a dashboard picks from, so it spans flows rather than sitting
inside one.
"""
from fluksio.api.routes.messages import MessageInfo
specs: dict[str, MessageSpec] = {}
providers: dict[str, list[str]] = {}
for flow in self.store.read_all():
for node in flow.nodes:
for spec in _bound(node.provides):
name = qualify(flow.name, spec.name)
specs.setdefault(name, spec)
providers.setdefault(name, []).append(f"{flow.name}.{node.id}")
for spec in _bound(node.requires):
specs.setdefault(qualify(flow.name, spec.name), spec)
for declared in flow.inputs:
if declared.spec.name:
specs.setdefault(
qualify(flow.name, declared.spec.name), declared.spec
)
values = self.values()
infos = []
for name, spec in sorted(specs.items()):
current = values.get(name, {})
infos.append(
MessageInfo(
name=name,
flow=flow_of(name),
dtype=spec.dtype.value,
providers=sorted(providers.get(name, [])),
writable=True,
numeric=spec.dtype.value in ("float", "int"),
value=current.get("value"),
ts=current.get("ts"),
)
)
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, other.id
)
)
self._wiring_cache = (revision, index)
return index
def cross_flow_nodes(self, flow: str) -> list[Wiring]:
"""Nodes in other flows on the far side of a dotted message name.
Returns the far node's `Wiring`, narrowed to the messages of `flow` it
provides and requires. This is what link-in and link-out mean: the
wiring is already real — the canvas simply had no way to draw the
other end. *flow* is a key.
"""
prefix = f"{flow}."
found = []
for entry in self._wiring():
if entry.flow == flow:
continue
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(replace(entry, provides=provides, requires=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.
Two flows on the same broker topic are one neuron here: the topic is a
single thing, and the path that runs through it is real wiring that no
one flow's canvas can show. Read from the store rather than from the
compiled pipeline, so a node that failed to load still appears — a
broken neuron is exactly what someone comes to this view to find.
"""
# What validation found, by the node it names, plus what is down right
# now — a neuron someone comes here to find is as often a node that
# stopped working as one that never built. A cycle names every node in
# it, so all of them are marked rather than an arbitrary one.
troubled: dict[str, str] = {}
for issue in self.issues + self._health_issues():
if issue.advisory:
continue
for member in (*issue.nodes, *filter(None, [issue.node])):
troubled.setdefault(member, issue.message)
groups: dict[str, BrainNode] = {}
# Which group each `flow.node_id` ended up in.
gid_of: dict[str, str] = {}
producers: dict[str, list[str]] = {}
# (member, message) for every input that actually wakes a node.
consumers: list[tuple[str, str]] = []
for flow in self.store.read_all():
for node_def in flow.nodes:
member = f"{flow.name}.{node_def.id}"
node_type = NODE_TYPES.get(node_def.type)
key = None
if node_type is not None:
try:
key = node_type.cls.target_key(node_def.params)
except Exception:
# A plugin's own grouping is not worth the whole graph;
# this node just stands on its own.
logger.exception(
"target_key failed for node type '%s'", node_def.type
)
gid = f"{node_def.type}:{key}" if key else member
gid_of[member] = gid
group = groups.get(gid)
if group is None:
# The first member names the neuron. What it points at is
# in the id, which can be a whole topic map and is far too
# long to draw on a circle.
group = groups[gid] = BrainNode(
id=gid,
label=node_def.title or node_def.id,
kind=node_def.type,
)
group.members.append(member)
if flow.name not in group.flows:
group.flows.append(flow.name)
group.flow_ids.append(flow.id)
# A merged neuron stands for several nodes, so one of them being
# unable to run is enough to mark it; the first reason wins.
if group.issue is None:
group.issue = troubled.get(member)
for spec in _bound(node_def.provides):
producers.setdefault(qualify(flow.name, spec.name), []).append(
member
)
for spec in _bound(node_def.requires):
if spec.trigger:
consumers.append((member, qualify(flow.name, spec.name)))
messages: dict[tuple[str, str], set[str]] = {}
for member, message in consumers:
target = gid_of[member]
for producer in producers.get(message, ()):
source = gid_of[producer]
# Everything a merged neuron says to itself is inside it.
if source != target:
messages.setdefault((source, target), set()).add(message)
return BrainGraph(
nodes=sorted(groups.values(), key=lambda node: node.id),
edges=[
BrainEdge(source=source, target=target, messages=sorted(names))
for (source, target), names in sorted(messages.items())
],
)
def publish_message(
self, name: str, value: Any, source: ValueSource | None = None
) -> None:
"""Put a value into the graph from outside. Blocking.
Refuses a name no flow declares: the flows own the namespace, and a
message nothing reads or writes would just be a key nobody sees.
:raises EngineHeld: While a rename holds the engine, for the caller to
answer: a 503, a device's ``busy``, a log line.
"""
if self.pipeline is None:
raise KeyError(name)
spec = None
for flow in self.store.read_all():
for node in flow.nodes:
for candidate in [*_bound(node.provides), *_bound(node.requires)]:
if qualify(flow.name, candidate.name) == name:
spec = candidate
break
for declared in flow.inputs:
if qualify(flow.name, declared.spec.name) == name:
spec = declared.spec
if spec is None:
raise KeyError(name)
# The same check a node's output gets; a dashboard is not looser.
spec.check(value)
self.pipeline.publish({name: value}, source)
def queue_stats(self) -> dict[str, Any]:
return self.execution.stats() if self.execution is not None else {}
def forget_flow(self, flow: str) -> None:
"""Drop what a deleted flow left behind, by its key. Blocking.
After its nodes are torn down, or they put some of it straight back.
"""
if self.execution is not None:
self.execution.queue.clear_flow(flow)
with self.state.lock():
for key in [k for k in self.state.keys() if touches(flow, k)]:
self.state.delete(key)
def key_residue(self, flow: str) -> list[str]:
"""What still lives under the flow key *flow*. Blocking.
State and live queue work in its namespace, judged as a rename maps
it (`namespace.touches`): left by a flow that held the key before, or
put there by anything still writing to it. A new flow taking the key
would inherit it, so creating one is refused while this is not empty,
and so is renaming a flow onto it. Work that was acknowledged,
dead-lettered or marked done is history and not counted. Names what it
found, state keys first.
"""
found = [key for key in self.state.keys() if touches(flow, key)]
if self.execution is not None:
found += [
f"queued work of '{item.node}'"
for item in self.execution.queue.live_items()
if _touches(item, flow)
]
return found
def run_flow(self, flow: str, inputs: dict[str, Any] | None = None) -> None:
"""Run every node of one flow. Blocking — call from a worker thread."""
if self.pipeline is None:
return
self.pipeline.run(inputs or {}, nodes=self.pipeline.flow_nodes(flow))
def build_run_pipeline(
self,
flow: FlowDef,
state: StateBackend,
draft: bool = False,
observer: Callable[[NodeOutcome], None] | None = None,
emission_observer: Callable[[str, dict[str, Any]], None] | None = None,
run: RunContext | None = None,
run_cache: RunCacheLookup | None = None,
) -> Pipeline:
"""Build one flow as a pipeline of its own, for a single run.
Nothing here is activated: a run executes the graph from its inputs
rather than waiting to be told something, so subscriptions, schedules
and webhooks would only be a second copy of what the live pipeline
already holds. The state is the run's, which is what keeps two runs of
one flow from overwriting each other's messages.
:raises NodesNotLoaded: When a node could not be built. The live engine
leaves such a node out, but a run without it would finish as if
the flow had done its work.
"""
nodes, loaded, initial_values, _inputs = self._build_flows(
[(flow, draft)], run=run
)
errors = {
entry.id: entry.error or "unknown error"
for entry in loaded.values()
if entry.status is NodeStatus.ERROR
}
if errors:
raise NodesNotLoaded(errors)
pipeline = Pipeline(
nodes=nodes,
state=state,
events=self.events,
max_workers=self.max_workers,
initial_values=initial_values,
observer=observer,
emission_observer=emission_observer,
run_cache=run_cache,
run_id=run.run_id if run is not None else "",
)
pipeline.history_limits = self.history_limits
return pipeline
def run_preview(self, flow_id: str, inputs: dict[str, Any] | None = None) -> None:
"""Run a flow's unpublished draft once, as the editor shows it.
The nodes are built and run but never activated, so a draft never
acquires subscriptions, schedules or webhooks — only publishing does
that. State and events are the live ones, so values land on the canvas
exactly as they do for a published run.
"""
flow = self.store.read_flow(flow_id, draft=True)
nodes, _, initial_values, _ = self._build_flows([(flow, True)])
pipeline = Pipeline(
nodes=nodes,
state=self.state,
events=self.events,
max_workers=self.max_workers,
initial_values=initial_values,
)
pipeline.run(inputs or {})
def trigger_node(self, node_id: str, values: dict[str, Any] | None = None) -> str:
"""Feed values into one node. Blocking — call from a worker thread.
Runs here rather than through the queue: the caller is a person waiting
on the response, and wants the state it produced.
Returns what the node raised, or an empty string. A node failing by hand
is reported the way a queued run reports it — on the canvas and in the
metrics — rather than as a stack trace in the server log.
:raises EngineHeld: While a rename holds the engine, which is no
failure of the node's.
"""
node = self.get_node(node_id)
if node is None:
raise KeyError(node_id)
try:
if node.requires and values:
node.trigger(values, durable=False)
else:
node.inject(values or {}, durable=False)
except EngineHeld:
raise
except Exception as exc:
if self.pipeline is None:
raise
return self.pipeline.publish_error(node, exc)
return ""
def _publish(self, event: dict[str, Any]) -> None:
if self.events is not None:
self.events.publish(event)
def _touches(item: WorkItem, flow: str) -> bool:
"""Whether a queued item names anything of the flow key *flow*: as its
flow, its node or one of the messages it carries."""
return owned_by(item, flow) or any(flow_of(name) == flow for name in item.outputs)
def _bound(specs: list[MessageSpec]) -> list[MessageSpec]:
"""Ports without a message name are not wired into the graph."""
return [spec for spec in specs if spec.name]
def _signature_gap(
code: str, node_id: str, ports: set[str], settings: set[str]
) -> str | None:
"""A declared port or setting ``process`` cannot accept, or None.
Ports and settings both arrive as keyword arguments, so a name the function
does not take is a ``TypeError`` on every call — and a node that loads fine
and fails every time it runs is the quiet kind of broken. Read statically:
with a worker pool the code is never imported here. Anything this cannot
read for certain (no plain ``def process``, a decorator, ``**kwargs``, a
syntax error the compile step words better) is left to the call.
"""
try:
tree = ast.parse(code)
except SyntaxError:
return None
defs = [
node
for node in tree.body
if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef)
and node.name == "process"
]
if not defs or defs[-1].decorator_list or defs[-1].args.kwarg is not None:
return None
args = defs[-1].args
names = [a.arg for a in [*args.posonlyargs, *args.args, *args.kwonlyargs]]
missing = sorted((ports | settings) - set(names))
if not missing:
return None
kind = "an input" if missing[0] in ports else "a setting"
gap = (
f"'{missing[0]}' is {kind} of '{node_id}' but "
f"process({', '.join(names)}) takes no such argument"
)
if kind == "a setting" and "params" in names:
# Code written for the `params` dict settings used to arrive in.
gap += f". Settings arrive by name: process(..., {missing[0]})"
return gap
def follow_header(node: NodeDef, code: str, drop: bool = True) -> NodeDef:
"""The node with its input ports and settings read off ``def process(...)``.
The inverse of the canvas's scaffold, which writes the input ports first,
in their order, and the settings after them. A parameter already declared
keeps its kind. A new one is a setting when it has a default, which becomes
its value, or when it follows a setting, and an input port otherwise. One
name swapped for one other is a rename, so the port keeps its message and
the setting its value. A declared name the header no longer has is dropped,
unless ``drop`` is false: code saved while it is still being typed keeps
those, so a pause mid-edit does not cost a port its message.
A header this cannot read for certain leaves the node as it is: a syntax
error, no top-level ``def process``, a decorator, ``*args``, ``**kwargs``,
a bare ``*`` or ``/``.
"""
try:
tree = ast.parse(code)
except SyntaxError:
return node
defs = [
entry
for entry in tree.body
if isinstance(entry, ast.FunctionDef) and entry.name == "process"
]
if not defs:
return node
args = defs[-1].args
if (
defs[-1].decorator_list
or args.vararg
or args.kwarg
or args.posonlyargs
or args.kwonlyargs
):
return node
# Defaults belong to the last parameters, so they line up from the end.
defaults: list[ast.expr | None] = [None] * (
len(args.args) - len(args.defaults)
) + list(args.defaults)
header = [
(arg.arg, default)
for arg, default in zip(args.args, defaults, strict=True)
if arg.arg not in RESERVED_SETTINGS
]
def written(name: str) -> bool:
return name.isascii() and name.isidentifier()
ports = {spec.port: spec for spec in node.requires if written(spec.port)}
settings = {
key: value
for key, value in node.params.items()
if key not in RESERVED_SETTINGS and written(key)
}
names = [name for name, _ in header]
gone = [name for name in [*ports, *settings] if name not in names]
new = [name for name in names if name not in ports and name not in settings]
renamed = {new[0]: gone[0]} if len(gone) == len(new) == 1 else {}
requires: list[MessageSpec] = []
params = {key: value for key, value in node.params.items() if key not in settings}
after_setting = False
for name, default in header:
was = renamed.get(name, name)
if was in ports:
requires.append(ports[was].model_copy(update={"port": name}))
elif was in settings or default is not None or after_setting:
params[name] = settings[was] if was in settings else _literal(default)
after_setting = True
else:
requires.append(MessageSpec(port=name))
if not drop:
kept = set(gone) - set(renamed.values())
# Where they were, so that typing the name back changes nothing.
for index, spec in enumerate(node.requires):
if spec.port in kept:
requires.insert(index, spec)
params |= {key: settings[key] for key in kept if key in settings}
# Ports the header cannot name, such as a row not named yet, stay.
requires += [spec for spec in node.requires if not written(spec.port)]
return node.model_copy(update={"requires": requires, "params": params})
def _literal(default: ast.expr | None) -> Any:
"""A parameter's default as a setting's value, or empty when it is not JSON."""
if default is None:
return ""
try:
value = ast.literal_eval(default)
json.dumps(value)
except (ValueError, TypeError, SyntaxError, RecursionError):
return ""
return value
def with_settings(
function: Callable[..., Any], params: dict[str, Any]
) -> Callable[..., Any]:
"""A node's function with its settings bound as keyword arguments.
A setting is a constant of one node's function, so it is passed the way a
port is: by name. The engine's own settings never reach the code, and the
``params`` the pipeline offers is dropped here rather than travelling to a
worker that has nothing to do with it.
"""
settings = {k: v for k, v in params.items() if k not in RESERVED_SETTINGS}
def call(
params: dict[str, Any] | None = None, # noqa: ARG001 - absorbed here
**ports: Any,
) -> Any:
return function(**ports, **settings)
return call
def _declared_inputs(flow: FlowDef) -> tuple[dict[str, bool], dict[str, Any]]:
"""A flow's declared inputs, and the ones that start with a value.
An input maps to whether a value will be there when the flow starts, which
for a batch flow is not the same as carrying an initial: it is started
explicitly and its inputs arrive with the run, so one without an initial is
a parameter the caller fills rather than a message nothing ever sets.
Reporting it as the latter had the health summary count an experiment as a
flow that cannot run while the Runs screen showed it running.
"""
declared: dict[str, bool] = {}
initial: dict[str, Any] = {}
for flow_input in flow.inputs:
name = qualify(flow.name, flow_input.spec.name)
if not name:
continue
declared[name] = flow_input.initial is not None or flow.mode == "batch"
if flow_input.initial is not None:
initial[name] = flow_input.initial
return declared, initial
def _collect_issues(
loaded: dict[str, LoadedNode],
pipeline: Pipeline,
flow_inputs: dict[str, bool],
devices: Any = None,
) -> list[ValidationIssue]:
"""Everything wrong with a built graph, whether it is running or not."""
issues = pipeline.validate(flow_inputs) + _device_issues(loaded, devices)
issues += [
ValidationIssue(
code="node_error",
message=entry.error or "This node could not be loaded.",
flow=entry.flow,
node=entry.id,
)
for entry in loaded.values()
if entry.status is NodeStatus.ERROR
]
# Hooks are mounted without authentication, so a hook without a secret is
# open to anyone who guesses its URL. Flows written before the parameter
# existed keep running, but say so.
issues += [
ValidationIssue(
code="unauthenticated_hook",
message=(
f"Hook '{entry.node.local_id}' has no secret. "
"Anyone who knows its URL can reach it."
),
flow=entry.flow,
node=entry.id,
)
for entry in loaded.values()
if _is_hook(entry.node) and not entry.node.secret
]
# A node whose body was never stored runs the new-node template, which
# returns nothing — quietly, and looking healthy the whole time.
issues += [
ValidationIssue(
code="missing_source",
message=(
f"Node '{entry.id.rpartition('.')[2]}' has no stored code. It "
"runs as an empty node and publishes nothing."
),
flow=entry.flow,
node=entry.id,
)
for entry in loaded.values()
if entry.missing_source
]
# Only code travels to a worker, so a worker on any other node is a field
# nothing reads. Saying so beats a setting that looks like it took.
issues += [
ValidationIssue(
code="ignored_worker",
message=(
f"Node '{entry.id.rpartition('.')[2]}' names worker "
f"'{entry.ignored_worker}', but only a Function node runs on a "
"worker. The setting is ignored."
),
flow=entry.flow,
node=entry.id,
)
for entry in loaded.values()
if entry.ignored_worker
]
return issues
#: The dtypes a device port reads as a plain number.
_NUMBERS = frozenset({"int", "float"})
def _ways(said: dict[str, Any]) -> list[str]:
"""The directions a reported port may be wired in, none when it does not
say: a direction this version does not know is no reason to refuse one."""
if said.get("modes"):
return list(said["modes"])
return [said["direction"]] if said.get("direction") in ("in", "out") else []
def _device_issues(
loaded: dict[str, LoadedNode], devices: Any
) -> list[ValidationIssue]:
"""Device nodes naming no device, and ports that cannot be what they say.
A port is wired to one message in one place, so one bound twice across
the graph is an error on each. Against a report, a port in the wrong
direction or of another dtype is an error too, and one the device does not
list is a warning: the device may know it all the same. Without a report
a port is the device's business and nothing is said about it.
A port reporting ``modes`` may be wired in any of them, as a GPIO pin is
either way. ``int`` and ``float`` pass for each other: the device reads a
number either way.
"""
if devices is None:
return []
nodes = [
(entry, entry.node)
for entry in loaded.values()
if isinstance(entry.node, DeviceNode)
]
if not nodes:
return []
known = {device.id for device in devices.list()}
bound = Counter(
(node.cfg.device, spec.port)
for _, node in nodes
for spec in (*node.input_ports, *node.output_ports)
if spec.name
)
issues: list[ValidationIssue] = []
for entry, node in nodes:
device = node.cfg.device
if device not in known:
issues.append(
ValidationIssue(
code="unknown_device",
message=(
f"'{node.local_id}' names device '{device}', which does "
"not exist."
if device
else f"Pick the device '{node.local_id}' stands for."
),
flow=entry.flow,
node=entry.id,
)
)
continue
reported = devices.reported_ports(device)
for direction, specs in (("in", node.output_ports), ("out", node.input_ports)):
for spec in specs:
if not spec.name:
continue
problem, advisory = "", False
said = reported.get(spec.port) if reported is not None else None
if bound[(device, spec.port)] > 1:
problem = (
f"Port '{spec.port}' of '{device}' is wired more than "
"once. A port carries one message, in one flow."
)
elif reported is None:
pass
elif said is None:
problem = f"'{device}' does not report a port '{spec.port}'."
advisory = True
elif (ways := _ways(said)) and direction not in ways:
problem = (
f"'{spec.port}' of '{device}' is an {said['direction']} "
f"port, wired here as {direction}."
)
elif said.get("dtype") and not (
said["dtype"] == spec.dtype.value
or {said["dtype"], spec.dtype.value} <= _NUMBERS
):
problem = (
f"'{spec.port}' of '{device}' carries {said['dtype']}, "
f"wired here to a {spec.dtype.value} message."
)
if problem:
issues.append(
ValidationIssue(
code="device_port",
message=problem,
flow=entry.flow,
node=entry.id,
port=spec.port,
message_name=spec.name,
advisory=advisory,
)
)
return issues
def _short_error(exc: Exception) -> str:
"""One line a node author can act on: what broke, and where in their code.
The frames in between belong to the loader rather than to the node, so only
the line in the node's own source is kept. The full traceback stays in the
server log.
"""
if isinstance(exc, SyntaxError):
# Its own message already names the compiled file, which is noise here.
return f"{type(exc).__name__}: {exc.msg} (line {exc.lineno})"
frames = [
frame
for frame in traceback.extract_tb(exc.__traceback__)
if frame.filename.startswith("<node ")
]
where = f" (line {frames[-1].lineno})" if frames else ""
return f"{type(exc).__name__}: {exc}{where}"
Reference in New Issue View Git Blame Copy Permalink