stroblmeandClaude Fable 5
774b03953a
The first cut had node code call fluksio.log_metric, which was a second, undeclared way for data to leave a node: invisible to validation, absent from the canvas, and stored where the graph could not see it. That is precisely the MLflow discrepancy this framework exists to avoid, so it is gone. A node that produces values over time is a generator. Every yield is a dict keyed by output port, published the instant it happens — same port, same type check, same place on the canvas as any other value — and what it returns is its result. A port doing this declares stream: true, and a run keeps every number one takes, so experiment tracking is a consequence of the graph rather than an API beside it: a chart binds to a training curve the way it binds to a temperature. fluksio.emit writes the same ports imperatively, for where a yield cannot reach — inside a training framework's callback. In a live flow an emission also wakes what is downstream, as a subscriber publishing does; in a run it does not, because a run's graph is scheduled once and mid-node cascades would leave 'finished' with nothing to mean. The enqueued item carries no payload: the value is already in state, and one carrying it would re-apply an old emission after the node returned. Verified on the stack: 30 loss values arrived live on the flow socket during a run, attributed to the node that produced them, and the same node run on the remote worker streamed its curve back across the socket. Also caches remote compile results per worker, so attaching a GPU box does not put a network round trip in every rebuild. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AD8SfVhzXBG2nAfFcVh3iD
1204 lines
46 KiB
Python
1204 lines
46 KiB
Python
"""Pipeline: the executable graph.
|
|
|
|
One pipeline holds the nodes of every loaded flow. Edges are not declared —
|
|
they follow from message names, so a node consuming ``heating.setpoint`` is
|
|
downstream of every node providing it. Several producers of one message are
|
|
allowed: each publication triggers the consumers, and the latest value wins.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from collections import deque
|
|
from collections.abc import Callable, Iterator
|
|
from concurrent.futures import Future, ThreadPoolExecutor, wait
|
|
from contextlib import contextmanager
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from app.flow import logs
|
|
from app.flow.artifacts import is_reference
|
|
from app.flow.events import EventBus
|
|
from app.flow.messages import flow_of
|
|
from app.flow.nodes import Node
|
|
from app.flow.queue import WorkQueue
|
|
from app.flow.state import MemoryState, StateBackend
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
#: A run that never went through the queue. It still gets a run record, so a
|
|
#: manual run shows up in the history — but it is no one's idempotency key.
|
|
MANUAL_RUN_PREFIX = "manual-"
|
|
|
|
|
|
class ValidationIssue(BaseModel):
|
|
"""A problem that keeps a flow from running correctly."""
|
|
|
|
code: Literal[
|
|
"cycle",
|
|
"unconnected_input",
|
|
"missing_initial_value",
|
|
"node_error",
|
|
"unauthenticated_hook",
|
|
"self_loop_needs_initial",
|
|
]
|
|
message: str
|
|
flow: str = ""
|
|
nodes: list[str] = []
|
|
node: str | None = None
|
|
port: str | None = None
|
|
message_name: str | None = None
|
|
|
|
|
|
class ValueSource(BaseModel):
|
|
"""Who caused a message to take its current value.
|
|
|
|
The canvas draws an edge per producer, so without this it pulses every one
|
|
of them and claims a node published something it did not. It is also what
|
|
lets a value arriving from outside the flow — a dashboard control, another
|
|
flow, an agent — be shown at all, since none of those is a node here.
|
|
"""
|
|
|
|
#: node, dashboard, flow, agent or api.
|
|
kind: str = "node"
|
|
#: Node id, dashboard name, or whatever identifies the caller.
|
|
id: str = ""
|
|
#: What to call it on screen.
|
|
label: str = ""
|
|
#: The widget, for a dashboard.
|
|
detail: str = ""
|
|
|
|
|
|
def node_source(node: Node) -> ValueSource:
|
|
return ValueSource(kind="node", id=node.id, label=node.local_id)
|
|
|
|
|
|
class NodeOutcome(BaseModel):
|
|
"""How one node execution went.
|
|
|
|
Handed to whoever is watching a particular pipeline rather than published:
|
|
a run has to record every node it ran, and the event bus drops what it
|
|
cannot keep up with.
|
|
"""
|
|
|
|
node: str
|
|
ok: bool
|
|
duration_ms: float = 0.0
|
|
outputs: int = 0
|
|
error: str = ""
|
|
logs: str = ""
|
|
#: Artifact references this node emitted, keyed by the message carrying
|
|
#: them — what a run records so a result can be opened later.
|
|
artifacts: dict[str, dict[str, Any]] = {}
|
|
|
|
|
|
class Pipeline:
|
|
"""Directed graph of nodes with automatic dependency resolution."""
|
|
|
|
__slots__ = (
|
|
"_nodes",
|
|
"_state",
|
|
"_events",
|
|
"_max_workers",
|
|
"produces",
|
|
"dependencies",
|
|
"_edges",
|
|
"_execution_order",
|
|
"_downstream_cache",
|
|
"_disabled",
|
|
"_paused",
|
|
"_stepping",
|
|
"_gate_lock",
|
|
"_queue",
|
|
"_node_pool",
|
|
"history_limits",
|
|
"observer",
|
|
"emission_observer",
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
nodes: list[Node] | None = None,
|
|
state: StateBackend | None = None,
|
|
events: EventBus | None = None,
|
|
max_workers: int | None = None,
|
|
initial_values: dict[str, Any] | None = None,
|
|
disabled_flows: set[str] | None = None,
|
|
work_queue: WorkQueue | None = None,
|
|
node_pool: ThreadPoolExecutor | None = None,
|
|
observer: Callable[[NodeOutcome], None] | None = None,
|
|
emission_observer: Callable[[str, dict[str, Any]], None] | None = None,
|
|
) -> None:
|
|
self._nodes = nodes or []
|
|
# Stopped flows are stored and survive a restart; paused ones are a
|
|
# debugging state that a rebuild is meant to clear.
|
|
self._disabled = frozenset(disabled_flows or ())
|
|
self._paused: set[str] = set()
|
|
self._stepping: set[str] = set()
|
|
self._gate_lock = threading.Lock()
|
|
# An empty state backend is falsy, so this cannot be ``state or ...``:
|
|
# that would quietly hand the pipeline a second, private state and
|
|
# leave everyone reading the shared one seeing nothing.
|
|
self._state: StateBackend = state if state is not None else MemoryState()
|
|
self._events = events
|
|
self._max_workers = max_workers
|
|
# Without a queue the pipeline runs everything inline, which is what
|
|
# tests, previews and manual runs want.
|
|
self._queue = work_queue
|
|
# A pool owned by the execution service, so a wave does not build one.
|
|
self._node_pool = node_pool
|
|
# Set by a run, which needs every node it executed written down.
|
|
self.observer = observer
|
|
# And every value a node produced on the way, which is what a
|
|
# training curve is once it goes out a port rather than into a log.
|
|
self.emission_observer = emission_observer
|
|
# How deep to keep each message's series; a chart asking for more
|
|
# than the default puts its message in here. Swapped, never mutated.
|
|
self.history_limits: dict[str, int] = {}
|
|
|
|
# A message may have several producers; every one of them is upstream
|
|
# of the nodes consuming it.
|
|
self.produces: dict[str, list[Node]] = {}
|
|
for node in self._nodes:
|
|
for msg in node.provides:
|
|
self.produces.setdefault(msg, []).append(node)
|
|
|
|
# A node never depends on itself: reading a message it also provides is
|
|
# how state is carried between runs, not a cycle. An input marked
|
|
# non-triggering is the same idea across two nodes.
|
|
self.dependencies: dict[Node, frozenset[Node]] = {
|
|
node: frozenset(
|
|
producer
|
|
for msg, spec in node.requires.items()
|
|
if spec.trigger
|
|
for producer in self.produces.get(msg, ())
|
|
if producer is not node
|
|
)
|
|
for node in self._nodes
|
|
}
|
|
|
|
self._edges: dict[Node, set[Node]] | None = None
|
|
self._execution_order: list[Node] | None = None
|
|
self._downstream_cache: dict[Node, list[Node]] = {}
|
|
|
|
if initial_values:
|
|
with self._state.lock():
|
|
for name, value in initial_values.items():
|
|
if name not in self._state:
|
|
self._state[name] = value
|
|
|
|
for node in self._nodes:
|
|
node.bind(self)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Graph
|
|
# -------------------------------------------------------------------------
|
|
|
|
@property
|
|
def nodes(self) -> list[Node]:
|
|
return self._nodes
|
|
|
|
@property
|
|
def state(self) -> StateBackend:
|
|
return self._state
|
|
|
|
@property
|
|
def edges(self) -> dict[Node, set[Node]]:
|
|
"""Producers mapped to their consumers, built on first access."""
|
|
if self._edges is None:
|
|
self._edges = {node: set() for node in self._nodes}
|
|
for consumer, producers in self.dependencies.items():
|
|
for producer in producers:
|
|
self._edges[producer].add(consumer)
|
|
return self._edges
|
|
|
|
def get_node_by_id(self, nid: str) -> Node | None:
|
|
return next((n for n in self._nodes if n.id == nid), None)
|
|
|
|
def flow_nodes(self, flow: str) -> set[Node]:
|
|
return {n for n in self._nodes if n.flow == flow}
|
|
|
|
def _topological_sort(self) -> list[Node]:
|
|
"""Kahn's algorithm; nodes left over are part of a cycle."""
|
|
if self._execution_order is not None:
|
|
return self._execution_order
|
|
|
|
in_degree = {node: len(deps) for node, deps in self.dependencies.items()}
|
|
queue = deque(n for n, deg in in_degree.items() if deg == 0)
|
|
result: list[Node] = []
|
|
|
|
while queue:
|
|
node = queue.popleft()
|
|
result.append(node)
|
|
for consumer in self.edges[node]:
|
|
in_degree[consumer] -= 1
|
|
if in_degree[consumer] == 0:
|
|
queue.append(consumer)
|
|
|
|
self._execution_order = result
|
|
return result
|
|
|
|
def _get_downstream(self, start: Node) -> list[Node]:
|
|
if start not in self._downstream_cache:
|
|
reachable: set[Node] = set()
|
|
queue = deque([start])
|
|
while queue:
|
|
for consumer in self.edges[queue.popleft()]:
|
|
if consumer not in reachable:
|
|
reachable.add(consumer)
|
|
queue.append(consumer)
|
|
order = self._topological_sort()
|
|
ordered = [n for n in order if n in reachable]
|
|
# Nodes inside a cycle never make it into the topological order.
|
|
ordered += [n for n in reachable if n not in order]
|
|
self._downstream_cache[start] = ordered
|
|
return self._downstream_cache[start]
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Validation
|
|
# -------------------------------------------------------------------------
|
|
|
|
def validate(
|
|
self, flow_inputs: dict[str, bool] | None = None
|
|
) -> list[ValidationIssue]:
|
|
"""Report everything that would keep this graph from running.
|
|
|
|
:param flow_inputs: Messages declared as inputs of a flow rather than
|
|
computed by it, mapped to whether they carry an initial value.
|
|
"""
|
|
declared = flow_inputs or {}
|
|
issues: list[ValidationIssue] = []
|
|
|
|
ordered = set(self._topological_sort())
|
|
if len(ordered) != len(self._nodes):
|
|
cyclic = sorted(n.id for n in self._nodes if n not in ordered)
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="cycle",
|
|
message=(
|
|
"These nodes depend on each other in a loop, so none of "
|
|
"them can run: " + ", ".join(cyclic)
|
|
),
|
|
flow=flow_of(cyclic[0]) if cyclic else "",
|
|
nodes=cyclic,
|
|
)
|
|
)
|
|
|
|
for node in self._nodes:
|
|
for msg_name, spec in node.requires.items():
|
|
if msg_name in self.produces:
|
|
# A message a node both reads and writes carries state
|
|
# between its runs. If the node is the only one writing it,
|
|
# the first run has nothing to read unless the flow declares
|
|
# a starting value.
|
|
if (
|
|
spec.trigger
|
|
and self.produces[msg_name] == [node]
|
|
and not declared.get(msg_name, False)
|
|
):
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="self_loop_needs_initial",
|
|
message=(
|
|
f"'{node.local_id}' reads '{msg_name}' and is "
|
|
"the only node writing it, so it needs a "
|
|
"starting value to ever run."
|
|
),
|
|
flow=node.flow,
|
|
node=node.id,
|
|
port=spec.port,
|
|
message_name=msg_name,
|
|
)
|
|
)
|
|
continue
|
|
|
|
if msg_name not in declared:
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="unconnected_input",
|
|
message=(
|
|
f"'{node.local_id}' waits for '{msg_name}', "
|
|
"which nothing provides."
|
|
),
|
|
flow=node.flow,
|
|
node=node.id,
|
|
port=spec.port,
|
|
message_name=msg_name,
|
|
)
|
|
)
|
|
elif not declared[msg_name]:
|
|
# Declared as a flow input, but nothing ever sets it, so
|
|
# the node waits forever.
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="missing_initial_value",
|
|
message=(
|
|
f"'{msg_name}' has no starting value, so "
|
|
f"'{node.local_id}' never runs."
|
|
),
|
|
flow=node.flow,
|
|
node=node.id,
|
|
port=spec.port,
|
|
message_name=msg_name,
|
|
)
|
|
)
|
|
return issues
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Readiness (synchronous nodes)
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _version_key(self, msg_name: str) -> str:
|
|
return f"__version__:{msg_name}"
|
|
|
|
def _last_seen_key(self, node_name: str, msg_name: str) -> str:
|
|
return f"__last_seen__:{node_name}:{msg_name}"
|
|
|
|
def _timestamp_key(self, msg_name: str) -> str:
|
|
return f"__ts__:{msg_name}"
|
|
|
|
def _delivered_key(self, node_name: str, msg_name: str) -> str:
|
|
"""When a rate-limited input last woke this node."""
|
|
return f"__in_ts__:{node_name}:{msg_name}"
|
|
|
|
def _held_key(self, msg_name: str) -> str:
|
|
"""The value a rate-limited output kept back, waiting for its window."""
|
|
return f"__held__:{msg_name}"
|
|
|
|
def _flush_key(self, node_name: str) -> str:
|
|
"""When a rate-limit window of this node is due to be let through."""
|
|
return f"__flush__:{node_name}"
|
|
|
|
def _throttled(self, node: Node, result: dict[str, Any]) -> dict[str, Any]:
|
|
"""Hold back the outputs whose port is not due to publish yet.
|
|
|
|
The value is not lost: it is kept and published when the window ends,
|
|
so a producer that goes quiet still delivers its last reading rather
|
|
than leaving the consumer on the one before it. Nothing declaring an
|
|
interval means nothing to look up.
|
|
"""
|
|
limited = {
|
|
name: spec.interval
|
|
for name, spec in node.provides.items()
|
|
if spec.interval > 0 and name in result
|
|
}
|
|
if not limited:
|
|
return result
|
|
|
|
now = time.time()
|
|
flush_key = self._flush_key(node.id)
|
|
stamps = self._state.get_multi(
|
|
[self._timestamp_key(name) for name in limited] + [flush_key]
|
|
)
|
|
|
|
passed: dict[str, Any] = {}
|
|
held: dict[str, Any] = {}
|
|
due_at = 0.0
|
|
for name, value in result.items():
|
|
if name not in limited:
|
|
passed[name] = value
|
|
continue
|
|
window_ends = (stamps.get(self._timestamp_key(name)) or 0) + limited[name]
|
|
if now >= window_ends:
|
|
passed[name] = value
|
|
else:
|
|
held[name] = value
|
|
due_at = min(due_at or window_ends, window_ends)
|
|
|
|
if self._queue is not None:
|
|
# Without a queue there is no timer to let the value out later, so
|
|
# holding it would only mean losing it more slowly.
|
|
for name in limited:
|
|
if name in passed:
|
|
# A fresh publish makes anything held for that port stale.
|
|
self._state.delete(self._held_key(name))
|
|
if held:
|
|
self._state.update(
|
|
{self._held_key(name): value for name, value in held.items()}
|
|
)
|
|
self._schedule_flush(node, due_at, now, stamps.get(flush_key))
|
|
return passed
|
|
|
|
def _schedule_flush(
|
|
self, node: Node, at: float, now: float, pending: float | None
|
|
) -> None:
|
|
"""Come back to this node when its rate-limit window ends.
|
|
|
|
One timer in flight per node: a pending one that is early enough is
|
|
left alone, and one that turns out to be too late simply finds nothing
|
|
to do when it fires.
|
|
"""
|
|
if self._queue is None or (pending and now < pending <= at):
|
|
return
|
|
self._state[self._flush_key(node.id)] = at
|
|
self.defer(node, {}, at - now, kind="flush")
|
|
|
|
def flush(self, node: Node) -> None:
|
|
"""Let through what this node's rate limits held back.
|
|
|
|
Runs when a window ends: the last value a limited output kept back is
|
|
published, and a node whose inputs were all held back gets its run.
|
|
"""
|
|
limited_out = [name for name, s in node.provides.items() if s.interval > 0]
|
|
stored = (
|
|
self._state.get_multi([self._held_key(name) for name in limited_out])
|
|
if limited_out
|
|
else {}
|
|
)
|
|
held = {
|
|
name: stored[self._held_key(name)]
|
|
for name in limited_out
|
|
if stored.get(self._held_key(name)) is not None
|
|
}
|
|
if held:
|
|
# Publishing runs the limit again, which is what clears the hold —
|
|
# or puts the timer back, if this fired a hair early.
|
|
self.apply_outputs(node, held)
|
|
self.run_downstream(node)
|
|
if any(spec.interval > 0 for spec in node.requires.values()):
|
|
self._execute_parallel(
|
|
{node, *self._get_downstream(node)}, self._state, check_ready=True
|
|
)
|
|
|
|
def _increment_message_versions(self, outputs: dict[str, Any]) -> None:
|
|
for msg_name in outputs:
|
|
self._state.increment(self._version_key(msg_name))
|
|
|
|
def _check_synchronous_ready(self, node: Node) -> tuple[bool, dict[str, int]]:
|
|
"""A synchronous node runs once every input is newer than last time.
|
|
|
|
Only triggering inputs count: waiting for a value the node itself
|
|
writes would mean waiting for a run that can never start.
|
|
"""
|
|
waited_on = [msg for msg, spec in node.requires.items() if spec.trigger]
|
|
if not waited_on:
|
|
return True, {}
|
|
|
|
version_keys = [self._version_key(msg) for msg in waited_on]
|
|
last_seen_keys = [self._last_seen_key(node.id, msg) for msg in waited_on]
|
|
values = self._state.get_multi(version_keys + last_seen_keys)
|
|
|
|
current_versions = {}
|
|
all_newer = True
|
|
|
|
for msg_name in waited_on:
|
|
current = values.get(self._version_key(msg_name)) or 0
|
|
last_seen = values.get(self._last_seen_key(node.id, msg_name)) or 0
|
|
current_versions[msg_name] = current
|
|
if current == 0 or current <= last_seen:
|
|
all_newer = False
|
|
|
|
return all_newer, current_versions
|
|
|
|
def _try_acquire_synchronous_execution(
|
|
self, node: Node, current_versions: dict[str, int]
|
|
) -> bool:
|
|
"""Claim the right to execute, so concurrent triggers run a node once."""
|
|
if not current_versions:
|
|
return True
|
|
|
|
expected = {}
|
|
updates = {}
|
|
for msg_name, version in current_versions.items():
|
|
expected[self._version_key(msg_name)] = version
|
|
updates[self._last_seen_key(node.id, msg_name)] = version
|
|
|
|
return self._state.compare_and_swap_multi(expected, updates)
|
|
|
|
def _input_is_due(self, node: Node) -> bool:
|
|
"""Has any rate-limited input waited out its interval?
|
|
|
|
A node runs when *any* of its inputs is due, so a slow port next to a
|
|
fast one throttles only itself. Nodes without a limited input never get
|
|
here.
|
|
"""
|
|
limited = {
|
|
name: spec.interval
|
|
for name, spec in node.requires.items()
|
|
if spec.interval > 0
|
|
}
|
|
if not limited:
|
|
return True
|
|
|
|
now = time.time()
|
|
flush_key = self._flush_key(node.id)
|
|
keys = [self._delivered_key(node.id, name) for name in limited]
|
|
stamps = self._state.get_multi(keys + [flush_key])
|
|
due = [
|
|
name
|
|
for name, interval in limited.items()
|
|
if now - (stamps.get(self._delivered_key(node.id, name)) or 0) >= interval
|
|
]
|
|
# An unlimited input alongside a held-back one still wakes the node.
|
|
if not due and len(limited) == len(node.requires):
|
|
# The value is in state, only the wake-up is held back. Come back
|
|
# for it, so a producer going quiet does not strand it there.
|
|
self._schedule_flush(
|
|
node,
|
|
min(
|
|
(stamps.get(self._delivered_key(node.id, name)) or 0) + interval
|
|
for name, interval in limited.items()
|
|
),
|
|
now,
|
|
stamps.get(flush_key),
|
|
)
|
|
return False
|
|
|
|
with self._state.lock():
|
|
for name in due:
|
|
self._state[self._delivered_key(node.id, name)] = now
|
|
return True
|
|
|
|
def _is_node_ready(self, node: Node, state: StateBackend) -> bool:
|
|
with state.lock():
|
|
for msg_name, spec in node.requires.items():
|
|
# A non-triggering input is read if it happens to be there;
|
|
# waiting for it would make an accumulator's first run
|
|
# impossible, since it is what the node is about to write.
|
|
if spec.trigger and msg_name not in state:
|
|
return False
|
|
|
|
if not self._input_is_due(node):
|
|
return False
|
|
|
|
if not node.synchronous:
|
|
return True
|
|
|
|
is_ready, current_versions = self._check_synchronous_ready(node)
|
|
if not is_ready:
|
|
return False
|
|
return self._try_acquire_synchronous_execution(node, current_versions)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Execution
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _publish(self, event: dict[str, Any]) -> None:
|
|
if self._events is not None:
|
|
self._events.publish(event)
|
|
|
|
def publish_log(self, node: Node, collected: logs.Collector, error: str) -> None:
|
|
"""One event per execution, so a chatty node cannot outrun the stream."""
|
|
text = collected.text + error
|
|
if not text:
|
|
return
|
|
self._publish(
|
|
{
|
|
"type": "node_log",
|
|
"flow": node.flow,
|
|
"node": node.id,
|
|
"text": text,
|
|
"level": "error" if error else "info",
|
|
"truncated": collected.truncated,
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
|
|
def publish_error(
|
|
self,
|
|
node: Node,
|
|
exc: Exception,
|
|
collected: logs.Collector | None = None,
|
|
entry_id: str = "",
|
|
) -> str:
|
|
"""Report a node failure, and hand the one-line version back.
|
|
|
|
Call it from an ``except`` block — the traceback comes from the
|
|
exception being handled. Every way of running a node reports through
|
|
here, so a manual run reads the same on the canvas and in the metrics
|
|
as a queued one.
|
|
"""
|
|
logger.exception("Node '%s' failed", node.id)
|
|
# The one-line error goes on the node; the traceback goes to the log
|
|
# panel, which is where there is room to read it.
|
|
self.publish_log(node, collected or logs.Collector(), logs.node_traceback())
|
|
error = f"{type(exc).__name__}: {exc}"
|
|
self._publish(
|
|
{
|
|
"type": "node_error",
|
|
"flow": node.flow,
|
|
"node": node.id,
|
|
"error": error,
|
|
"run": entry_id,
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
return error
|
|
|
|
def _already_done(self, entry_id: str, node: Node) -> bool:
|
|
"""Did this node's side effect already happen for this work item?"""
|
|
if node.idempotent or self._queue is None:
|
|
return False
|
|
try:
|
|
return self._queue.was_done(entry_id, node.id)
|
|
except Exception:
|
|
# Not knowing means running it again, which is the safer default
|
|
# for a value that may never have been delivered at all.
|
|
return False
|
|
|
|
def _execute_node(
|
|
self, node: Node, state: StateBackend, entry_id: str = ""
|
|
) -> dict[str, Any] | None:
|
|
"""Run one node and record its outputs. Never raises."""
|
|
started = time.perf_counter()
|
|
collected = logs.Collector()
|
|
try:
|
|
with state.lock():
|
|
inputs = {k: state[k] for k in node.requires if k in state}
|
|
|
|
with logs.capture(collected):
|
|
result = node.execute(inputs)
|
|
self.publish_log(node, collected, "")
|
|
|
|
if (
|
|
entry_id
|
|
and not entry_id.startswith(MANUAL_RUN_PREFIX)
|
|
and not node.idempotent
|
|
and self._queue is not None
|
|
):
|
|
# Written after the fact: a crash between the side effect and
|
|
# this marker is the one window at-least-once cannot close. A
|
|
# manual run has nothing to be redelivered, so it writes none.
|
|
try:
|
|
self._queue.mark_done(entry_id, node.id)
|
|
except Exception as exc:
|
|
logger.warning("Could not mark '%s' done: %s", node.id, exc)
|
|
|
|
if result:
|
|
result = self._throttled(node, result)
|
|
|
|
if result:
|
|
self._record_outputs(node, result, state)
|
|
|
|
duration_ms = round((time.perf_counter() - started) * 1000, 2)
|
|
self._publish(
|
|
{
|
|
"type": "node_executed",
|
|
"flow": node.flow,
|
|
"node": node.id,
|
|
# A node that returns nothing ran but published nothing,
|
|
# which is a different thing to show than one that emitted.
|
|
"outputs": len(result or {}),
|
|
"duration_ms": duration_ms,
|
|
"run": entry_id,
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
self._observe(
|
|
NodeOutcome(
|
|
node=node.id,
|
|
ok=True,
|
|
duration_ms=duration_ms,
|
|
outputs=len(result or {}),
|
|
logs=collected.text,
|
|
artifacts={
|
|
name: value
|
|
for name, value in (result or {}).items()
|
|
if is_reference(value)
|
|
},
|
|
)
|
|
)
|
|
return result
|
|
except Exception as exc:
|
|
# One failing node must not take the rest of the graph down.
|
|
error = self.publish_error(node, exc, collected, entry_id)
|
|
self._observe(
|
|
NodeOutcome(
|
|
node=node.id,
|
|
ok=False,
|
|
duration_ms=round((time.perf_counter() - started) * 1000, 2),
|
|
error=error,
|
|
logs=collected.text,
|
|
)
|
|
)
|
|
return None
|
|
|
|
def _record_outputs(
|
|
self, node: Node, outputs: dict[str, Any], state: StateBackend
|
|
) -> None:
|
|
"""Put a node's outputs where everything downstream of them looks.
|
|
|
|
State, the timestamp beside it, the series, the version counter and the
|
|
event the canvas draws from. Shared by a node returning and a node
|
|
emitting mid-execution, because those are the same act: a value the
|
|
node produced, leaving through a port it declared.
|
|
"""
|
|
ts = time.time()
|
|
with state.lock():
|
|
state.update(outputs)
|
|
state.update({self._timestamp_key(name): ts for name in outputs})
|
|
# Append-only, so it needs no lock of its own.
|
|
state.append_history(outputs, ts, self.history_limits)
|
|
self._increment_message_versions(outputs)
|
|
origin = node_source(node)
|
|
for name, value in outputs.items():
|
|
self._publish(
|
|
{
|
|
"type": "message_value",
|
|
"flow": flow_of(name),
|
|
"name": name,
|
|
"value": value,
|
|
"ts": ts,
|
|
"source": origin.model_dump(),
|
|
}
|
|
)
|
|
|
|
def publish_emission(self, node: Node, outputs: dict[str, Any]) -> None:
|
|
"""Publish what a node produced while it is still running.
|
|
|
|
Recorded before it is throttled, and throttled before it is published:
|
|
the run's history is the whole series, and a port declaring an interval
|
|
is asking for the *canvas* not to be flooded, not for its curve to have
|
|
holes in it.
|
|
|
|
Where there is a work queue — the live engine — an emission also wakes
|
|
what is downstream of it, exactly as a subscriber publishing does. A
|
|
run has no queue, and deliberately: its graph is scheduled once, and
|
|
three thousand mid-node cascades would leave "the run has finished"
|
|
with no meaning.
|
|
"""
|
|
self._observe_emission(node, outputs)
|
|
passed = self._throttled(node, outputs)
|
|
if not passed:
|
|
return
|
|
self._record_outputs(node, passed, self._state)
|
|
if self._queue is not None:
|
|
# Journalled with no payload of its own: the value is already in
|
|
# state, published in the order it was produced. An item carrying
|
|
# it would re-apply that value whenever it happened to be claimed,
|
|
# which is how an emission from the middle of a node overwrites the
|
|
# one it returned at the end. Downstream reads what is current,
|
|
# which is what "the latest value wins" has always meant here.
|
|
self._enqueue_cascade(node, None)
|
|
|
|
def _observe(self, outcome: NodeOutcome) -> None:
|
|
"""Tell the run watching this pipeline, if there is one."""
|
|
if self.observer is None:
|
|
return
|
|
try:
|
|
self.observer(outcome)
|
|
except Exception:
|
|
logger.exception("Run observer failed for '%s'", outcome.node)
|
|
|
|
def _observe_emission(self, node: Node, outputs: dict[str, Any]) -> None:
|
|
if self.emission_observer is None:
|
|
return
|
|
try:
|
|
self.emission_observer(node.id, outputs)
|
|
except Exception:
|
|
logger.exception("Run observer failed for an emission of '%s'", node.id)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Running, stopped, paused
|
|
# -------------------------------------------------------------------------
|
|
|
|
def is_disabled(self, flow: str) -> bool:
|
|
return flow in self._disabled
|
|
|
|
def paused_flows(self) -> list[str]:
|
|
with self._gate_lock:
|
|
return sorted(self._paused)
|
|
|
|
def pause(self, flow: str) -> None:
|
|
"""Hold this flow's nodes. Values still arrive; nothing acts on them."""
|
|
with self._gate_lock:
|
|
self._paused.add(flow)
|
|
self._publish({"type": "flow_paused", "flow": flow, "paused": True})
|
|
|
|
def resume(self, flow: str) -> None:
|
|
with self._gate_lock:
|
|
self._paused.discard(flow)
|
|
self._publish({"type": "flow_paused", "flow": flow, "paused": False})
|
|
# Whatever was held back is free to run now.
|
|
self._execute_parallel(self.flow_nodes(flow), self._state, check_ready=True)
|
|
|
|
@contextmanager
|
|
def stepping(self, flow: str) -> Iterator[None]:
|
|
"""Let one held-back wave through without ending the pause.
|
|
|
|
The flow stays paused throughout, so everything else arriving for it
|
|
keeps piling up where the next step can find it.
|
|
"""
|
|
with self._gate_lock:
|
|
self._stepping.add(flow)
|
|
try:
|
|
yield
|
|
finally:
|
|
with self._gate_lock:
|
|
self._stepping.discard(flow)
|
|
|
|
def _gate_blocks(self, node: Node) -> bool:
|
|
"""Is this node's flow held back from executing?"""
|
|
if node.flow in self._disabled:
|
|
return True
|
|
with self._gate_lock:
|
|
return node.flow in self._paused and node.flow not in self._stepping
|
|
|
|
def is_paused(self, flow: str) -> bool:
|
|
with self._gate_lock:
|
|
return flow in self._paused
|
|
|
|
def is_stepping(self, flow: str) -> bool:
|
|
"""Is a single step of this paused flow running right now?"""
|
|
with self._gate_lock:
|
|
return flow in self._stepping
|
|
|
|
def _execute_parallel(
|
|
self,
|
|
nodes_subset: set[Node] | None,
|
|
state: StateBackend,
|
|
check_ready: bool = False,
|
|
entry_id: str = "",
|
|
replay: bool = False,
|
|
) -> StateBackend:
|
|
"""Execute nodes concurrently, scheduling each as its inputs arrive."""
|
|
target_nodes = nodes_subset if nodes_subset is not None else set(self._nodes)
|
|
if not target_nodes:
|
|
return state
|
|
|
|
in_degree = {
|
|
n: sum(1 for dep in self.dependencies[n] if dep in target_nodes)
|
|
for n in target_nodes
|
|
}
|
|
|
|
submitted: set[Node] = set()
|
|
skipped: set[Node] = set()
|
|
node_futures: dict[Node, Future[dict[str, Any] | None]] = {}
|
|
|
|
def is_ready(n: Node) -> bool:
|
|
if in_degree[n] != 0:
|
|
return False
|
|
if check_ready:
|
|
return self._is_node_ready(n, state)
|
|
return True
|
|
|
|
def submit_ready(executor: ThreadPoolExecutor) -> None:
|
|
for n in target_nodes:
|
|
if n in submitted or n in skipped:
|
|
continue
|
|
# Gate before the readiness check, so a held-back node does not
|
|
# spend the synchronous claim it would need once it may run.
|
|
if self._gate_blocks(n):
|
|
continue
|
|
if is_ready(n):
|
|
submitted.add(n)
|
|
if replay and entry_id and self._already_done(entry_id, n):
|
|
# Its side effect happened on an earlier delivery; its
|
|
# outputs are still in state, so downstream carries on.
|
|
skipped.add(n)
|
|
submitted.discard(n)
|
|
for consumer in self.edges[n]:
|
|
if consumer in target_nodes:
|
|
in_degree[consumer] -= 1
|
|
continue
|
|
node_futures[n] = executor.submit(
|
|
self._execute_node, n, state, entry_id
|
|
)
|
|
elif n.synchronous and in_degree[n] == 0:
|
|
# Not ready now; a later trigger may make it ready.
|
|
skipped.add(n)
|
|
|
|
def drain(executor: ThreadPoolExecutor) -> None:
|
|
submit_ready(executor)
|
|
|
|
while node_futures:
|
|
done, _ = wait(node_futures.values(), return_when="FIRST_COMPLETED")
|
|
completed = [n for n, f in node_futures.items() if f in done]
|
|
|
|
for n in completed:
|
|
result = node_futures.pop(n).result()
|
|
# A node returning nothing (rate limiting, an error) stops
|
|
# propagation along its branch.
|
|
if result is not None:
|
|
for consumer in self.edges[n]:
|
|
if consumer in target_nodes:
|
|
in_degree[consumer] -= 1
|
|
submit_ready(executor)
|
|
|
|
if self._node_pool is not None:
|
|
# The execution service owns a long-lived pool; building one per
|
|
# wave is what used to spawn threads without bound under load.
|
|
drain(self._node_pool)
|
|
else:
|
|
with ThreadPoolExecutor(max_workers=self._max_workers) as executor:
|
|
drain(executor)
|
|
|
|
return state
|
|
|
|
def run(
|
|
self, inputs: dict[str, Any] | None = None, nodes: set[Node] | None = None
|
|
) -> StateBackend:
|
|
"""Execute the graph (or one flow's nodes) against the shared state."""
|
|
if inputs:
|
|
with self._state.lock():
|
|
self._state.update(inputs)
|
|
self._increment_message_versions(inputs)
|
|
|
|
return self._execute_parallel(nodes, self._state, check_ready=False)
|
|
|
|
def apply_outputs(self, node: Node, outputs: dict[str, Any] | None) -> None:
|
|
"""Record what a node emitted: state, history, versions and events.
|
|
|
|
Shared by the direct path and by the execution service replaying a
|
|
journaled item, so a value looks the same on the canvas either way.
|
|
"""
|
|
if outputs:
|
|
# This is where a chatty subscriber gets thinned out, so a port set
|
|
# to publish every 60s does so whatever the broker sends.
|
|
outputs = self._throttled(node, outputs)
|
|
|
|
if not outputs:
|
|
return
|
|
|
|
state = self._state
|
|
ts = time.time()
|
|
with state.lock():
|
|
state.update(outputs)
|
|
state.update({self._timestamp_key(name): ts for name in outputs})
|
|
state.append_history(outputs, ts, self.history_limits)
|
|
self._increment_message_versions(outputs)
|
|
origin = node_source(node)
|
|
for name, value in outputs.items():
|
|
self._publish(
|
|
{
|
|
"type": "message_value",
|
|
"flow": flow_of(name),
|
|
"name": name,
|
|
"value": value,
|
|
"ts": ts,
|
|
"source": origin.model_dump(),
|
|
}
|
|
)
|
|
# An injecting node — an MQTT subscriber, a webhook — publishes
|
|
# without going through the executor, but it did emit.
|
|
self._publish(
|
|
{
|
|
"type": "node_executed",
|
|
"flow": node.flow,
|
|
"node": node.id,
|
|
"outputs": len(outputs),
|
|
"duration_ms": 0,
|
|
"ts": ts,
|
|
}
|
|
)
|
|
|
|
def run_downstream(
|
|
self, node: Node, entry_id: str = "", replay: bool = False
|
|
) -> StateBackend:
|
|
"""Run everything downstream of a node that has just published.
|
|
|
|
``entry_id`` identifies the journaled item this run belongs to, so a
|
|
node with outside side effects can record that it ran. On a ``replay``
|
|
— the same item handed back after a crash — that record is checked
|
|
first: at-least-once delivery must not mean two of the same request.
|
|
"""
|
|
downstream = set(self._get_downstream(node))
|
|
if not downstream:
|
|
return self._state
|
|
return self._execute_parallel(
|
|
downstream,
|
|
self._state,
|
|
check_ready=True,
|
|
entry_id=entry_id,
|
|
replay=replay,
|
|
)
|
|
|
|
def trigger(
|
|
self, node: Node, outputs: dict[str, Any] | None, durable: bool | None = None
|
|
) -> StateBackend:
|
|
"""Publish a node's outputs and run everything downstream of it.
|
|
|
|
With a work queue attached the event is journaled and the caller
|
|
returns immediately — that is the path every external trigger takes, so
|
|
a crash mid-cascade loses nothing. Interactive callers (a manual run, a
|
|
draft preview) pass ``durable=False`` and get the old synchronous
|
|
behaviour, because they are waiting for the result.
|
|
|
|
A stopped flow drops the event: its subscriptions and schedules are torn
|
|
down anyway, and anything still arriving from another thread would be
|
|
work the flow was explicitly told not to do. A *paused* flow still
|
|
publishes, so the incoming value is visible on the canvas, and holds
|
|
the nodes downstream of it.
|
|
"""
|
|
state = self._state
|
|
|
|
if node.flow in self._disabled:
|
|
return state
|
|
|
|
if durable is None:
|
|
durable = self._queue is not None
|
|
if durable and self._queue is not None:
|
|
self._enqueue_cascade(node, outputs)
|
|
return state
|
|
|
|
return self._run_here(node, outputs)
|
|
|
|
def _run_here(
|
|
self, node: Node, outputs: dict[str, Any] | None, cause: str = "manual"
|
|
) -> StateBackend:
|
|
"""Run a cascade in this thread, under a run id of its own.
|
|
|
|
The queued path gets its run id from the journal entry. A run that never
|
|
went through the queue still belongs in the history, so it makes one —
|
|
marked as such, because it is no one's idempotency key.
|
|
"""
|
|
run_id = f"{MANUAL_RUN_PREFIX}{uuid.uuid4().hex[:12]}"
|
|
self._publish(
|
|
{
|
|
"type": "cascade_started",
|
|
"run": run_id,
|
|
"flow": node.flow,
|
|
"node": node.id,
|
|
"cause": cause,
|
|
"deliveries": 1,
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
try:
|
|
self.apply_outputs(node, outputs)
|
|
state = self.run_downstream(node, entry_id=run_id)
|
|
finally:
|
|
# Paired, or a cascade that raised leaves the run open until the
|
|
# abandoned sweep ten minutes later.
|
|
self._publish(
|
|
{
|
|
"type": "cascade_finished",
|
|
"run": run_id,
|
|
"flow": node.flow,
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
return state
|
|
|
|
def publish(
|
|
self, values: dict[str, Any], source: ValueSource | None = None
|
|
) -> None:
|
|
"""Put values into the graph without a node having produced them.
|
|
|
|
This is what a dashboard control does: the value is real, it just came
|
|
from a person rather than a sensor. Everything consuming those names
|
|
runs, the same as if a node had published them.
|
|
|
|
``source`` says what did, so the canvas can show the value arriving
|
|
from outside instead of blaming whichever node happens to be drawn as
|
|
a producer of that message.
|
|
"""
|
|
if not values:
|
|
return
|
|
|
|
origin = source or ValueSource(kind="api", label="API")
|
|
|
|
ts = time.time()
|
|
with self._state.lock():
|
|
self._state.update(values)
|
|
self._state.update({self._timestamp_key(name): ts for name in values})
|
|
self._state.append_history(values, ts, self.history_limits)
|
|
self._increment_message_versions(values)
|
|
for name, value in values.items():
|
|
self._publish(
|
|
{
|
|
"type": "message_value",
|
|
"flow": flow_of(name),
|
|
"name": name,
|
|
"value": value,
|
|
"ts": ts,
|
|
"source": origin.model_dump(),
|
|
}
|
|
)
|
|
|
|
targets: set[Node] = set()
|
|
for name in values:
|
|
for consumer in self._nodes:
|
|
if name in consumer.requires:
|
|
targets.add(consumer)
|
|
targets.update(self._get_downstream(consumer))
|
|
if targets:
|
|
self._execute_parallel(targets, self._state, check_ready=True)
|
|
|
|
def defer(
|
|
self,
|
|
node: Node,
|
|
outputs: dict[str, Any],
|
|
seconds: float,
|
|
guard: tuple[str, Any] | None = None,
|
|
kind: str = "cascade",
|
|
) -> bool:
|
|
"""Publish a node's outputs later, without holding a worker thread.
|
|
|
|
``guard`` names something the node must still remember when the wait is
|
|
over; if it has moved on, the item is dropped. That is how a wait which
|
|
gets restarted cancels the one it replaced.
|
|
|
|
Returns False when there is no queue to hold the item, in which case
|
|
the caller has to wait however it waited before.
|
|
"""
|
|
from app.flow.queue import WorkItem
|
|
|
|
if self._queue is None or seconds <= 0:
|
|
return False
|
|
item = WorkItem(
|
|
kind=kind,
|
|
node=node.id,
|
|
flow=node.flow,
|
|
outputs=outputs,
|
|
cause="delay",
|
|
guard_key=guard[0] if guard else "",
|
|
guard_value=str(guard[1]) if guard else "",
|
|
)
|
|
try:
|
|
self._queue.add_delayed(item, time.time() + seconds)
|
|
return True
|
|
except Exception as exc:
|
|
logger.error("Could not defer work for '%s': %s", node.id, exc)
|
|
return False
|
|
|
|
def _enqueue_cascade(self, node: Node, outputs: dict[str, Any] | None) -> None:
|
|
"""Journal a trigger, or fall back to running it here if that fails."""
|
|
from app.flow.queue import WorkItem
|
|
|
|
item = WorkItem(
|
|
kind="cascade",
|
|
node=node.id,
|
|
flow=node.flow,
|
|
outputs=outputs or {},
|
|
cause="external",
|
|
)
|
|
assert self._queue is not None
|
|
try:
|
|
self._queue.add(item)
|
|
return
|
|
except Exception as exc:
|
|
logger.error("Could not journal work for '%s': %s", node.id, exc)
|
|
self._publish(
|
|
{
|
|
"type": "queue_unavailable",
|
|
"flow": node.flow,
|
|
"node": node.id,
|
|
"error": f"{type(exc).__name__}: {exc}",
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
# Losing the value outright would be worse than running it here.
|
|
self._run_here(node, outputs, cause="external")
|
|
|
|
def values(self, flow: str | None = None) -> dict[str, dict[str, Any]]:
|
|
"""Last value and timestamp of every message, optionally one flow's."""
|
|
out: dict[str, dict[str, Any]] = {}
|
|
with self._state.lock():
|
|
keys = [k for k in self._state.keys() if not k.startswith("__")]
|
|
for key in keys:
|
|
if flow and flow_of(key) != flow:
|
|
continue
|
|
out[key] = {
|
|
"value": self._state.get(key),
|
|
"ts": self._state.get(self._timestamp_key(key)),
|
|
}
|
|
return out
|
|
|
|
def reset(self) -> None:
|
|
self._state.clear()
|