stroblmeandClaude Opus 5
f4cdf71795
Redis runs with appendfsync everysec, so an unclean stop loses up to a second of journaled entries. Name that where the docstrings promised a crash costs nothing, and state the setting and its trade-off on the configuration page. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CL9zvnnvcp1mvA8o7impxk
1769 lines
71 KiB
Python
1769 lines
71 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.
|
|
|
|
Because the wiring is derived rather than declared, one flow's nodes can be
|
|
swapped in place: `replace_flow` splices them into the list and derives the
|
|
whole map again, which gets the edges crossing into other flows right by
|
|
construction. A deploy does that rather than building a second pipeline.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
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 functools import lru_cache
|
|
from typing import Any, Literal, Protocol
|
|
|
|
from pydantic import BaseModel, computed_field
|
|
|
|
from fluksio.flow import logs
|
|
from fluksio.flow.artifacts import is_reference
|
|
from fluksio.flow.events import EventBus
|
|
from fluksio.flow.messages import flow_of, requalify
|
|
from fluksio.flow.nodes import Node
|
|
from fluksio.flow.queue import WorkQueue
|
|
from fluksio.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-"
|
|
|
|
# Validation codes that are worth saying but do not stop a flow running, so
|
|
# neither the brain graph nor the health summary treats them as a fault.
|
|
# Lives here rather than beside the controller so `ValidationIssue` can carry
|
|
# the distinction itself, and every reader gets it for free.
|
|
ADVISORY_ISSUES = frozenset({"unauthenticated_hook"})
|
|
|
|
|
|
class ValidationIssue(BaseModel):
|
|
"""Something wrong with a flow — a fault, or merely advisory."""
|
|
|
|
code: Literal[
|
|
"cycle",
|
|
"unconnected_input",
|
|
"missing_initial_value",
|
|
"node_error",
|
|
"unauthenticated_hook",
|
|
"self_loop_needs_initial",
|
|
"missing_source",
|
|
"node_unhealthy",
|
|
]
|
|
message: str
|
|
flow: str = ""
|
|
nodes: list[str] = []
|
|
node: str | None = None
|
|
port: str | None = None
|
|
message_name: str | None = None
|
|
|
|
@computed_field # type: ignore[prop-decorator]
|
|
@property
|
|
def advisory(self) -> bool:
|
|
"""Worth saying, but not a fault — the UI says so in a softer tone."""
|
|
return self.code in ADVISORY_ISSUES
|
|
|
|
|
|
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)
|
|
|
|
|
|
@lru_cache(maxsize=2048)
|
|
def _node_source_dump(node_id: str, local_id: str) -> dict[str, Any]:
|
|
"""The `source` block of a `message_value`, built once per node.
|
|
|
|
It is a pure function of the two names, and it was a pydantic model
|
|
constructed and dumped per emission — the same cost the live path already
|
|
refuses to pay for `NodeOutcome`.
|
|
"""
|
|
return ValueSource(kind="node", id=node_id, label=local_id).model_dump()
|
|
|
|
|
|
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]] = {}
|
|
#: Restored from an earlier run rather than executed.
|
|
cached: bool = False
|
|
#: Which run the restored values came from — and, because a restored row
|
|
#: holds no series of its own, which run this node's metrics live in.
|
|
cached_from: str = ""
|
|
#: What an equal execution of this node would be looked up by. Empty when
|
|
#: the node is not cacheable at all.
|
|
cache_key: str = ""
|
|
#: What it returned, for whoever stores the cache. None when it published
|
|
#: nothing, which is a result a later run has to be able to restore too.
|
|
output_values: dict[str, Any] | None = None
|
|
|
|
|
|
class CacheHit(BaseModel):
|
|
"""An earlier run of a node, as the thing restoring it needs to see."""
|
|
|
|
#: The flow whose namespace the stored outputs are named in. A node reached
|
|
#: through two flows publishes the same values under two names.
|
|
flow: str = ""
|
|
#: What it published. None means it published nothing, which is a result
|
|
#: worth restoring and has to be distinguishable from a miss.
|
|
outputs: dict[str, Any] | None = None
|
|
#: The run holding this node's series. Not necessarily the run the outputs
|
|
#: were read from: a row that was itself restored has no series of its own.
|
|
metrics_run: str = ""
|
|
|
|
|
|
class RunCacheLookup(Protocol):
|
|
"""Where a pipeline asks whether a node has already been run.
|
|
|
|
Kept to one method so the pipeline never learns there is a database: a run
|
|
hands it one of these, a test hands it a dict.
|
|
"""
|
|
|
|
def lookup(self, key: str) -> CacheHit | None:
|
|
"""What an equal execution produced, or None when there is no entry."""
|
|
|
|
|
|
def run_cache_key(fingerprint: str, inputs: dict[str, Any], flow: str = "") -> str:
|
|
"""What this node, with these inputs, is known by.
|
|
|
|
An artifact input counts as its digest: the reference carries a name and a
|
|
size beside it, and the same bytes under another name are the same input.
|
|
A value JSON cannot carry cannot be part of a key, and a node reading one
|
|
is simply not cacheable.
|
|
|
|
Inputs are keyed by the *node's* name for them, not the flow's: the same
|
|
node reading the same values through ``study`` and through ``quick`` did
|
|
the same work, and the fingerprint beside it already says what the node is.
|
|
A name belonging to some other flow keeps its prefix — reading
|
|
``other.metric`` is part of what makes this execution what it is.
|
|
"""
|
|
reduced = {
|
|
(name.removeprefix(f"{flow}.") if flow else name): (
|
|
value["digest"] if is_reference(value) else value
|
|
)
|
|
for name, value in inputs.items()
|
|
}
|
|
try:
|
|
canonical = json.dumps(
|
|
{"fp": fingerprint, "in": reduced}, sort_keys=True, separators=(",", ":")
|
|
)
|
|
except (TypeError, ValueError):
|
|
return ""
|
|
return hashlib.sha256(canonical.encode()).hexdigest()
|
|
|
|
|
|
def _derive(
|
|
nodes: list[Node],
|
|
) -> tuple[
|
|
dict[str, list[Node]],
|
|
dict[Node, frozenset[Node]],
|
|
dict[str, list[Node]],
|
|
dict[str, Node],
|
|
]:
|
|
"""Work out the wiring the node list implies: producers, then dependencies.
|
|
|
|
Done over the whole list rather than one flow's share of it, because a
|
|
flow is not a subgraph — its nodes can read and write messages another
|
|
flow owns — so there is no deriving one flow's edges on their own.
|
|
|
|
The two lookups come out of the same walk: who consumes a message, and
|
|
which node an id names. Both were linear scans over every node in the
|
|
instance, on the per-message path.
|
|
"""
|
|
# A message may have several producers; every one of them is upstream
|
|
# of the nodes consuming it.
|
|
produces: dict[str, list[Node]] = {}
|
|
consumes: dict[str, list[Node]] = {}
|
|
by_id: dict[str, Node] = {}
|
|
for node in nodes:
|
|
for msg in node.provides:
|
|
produces.setdefault(msg, []).append(node)
|
|
for msg in node.requires:
|
|
consumes.setdefault(msg, []).append(node)
|
|
by_id[node.id] = 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.
|
|
dependencies: dict[Node, frozenset[Node]] = {
|
|
node: frozenset(
|
|
producer
|
|
for msg, spec in node.requires.items()
|
|
if spec.trigger
|
|
for producer in produces.get(msg, ())
|
|
if producer is not node
|
|
)
|
|
for node in nodes
|
|
}
|
|
return produces, dependencies, consumes, by_id
|
|
|
|
|
|
class Pipeline:
|
|
"""Directed graph of nodes with automatic dependency resolution."""
|
|
|
|
__slots__ = (
|
|
"_nodes",
|
|
"_state",
|
|
"_events",
|
|
"_max_workers",
|
|
"produces",
|
|
"dependencies",
|
|
"consumes",
|
|
"_by_id",
|
|
"_edges",
|
|
"_execution_order",
|
|
"_downstream_cache",
|
|
"_disabled",
|
|
"_paused",
|
|
"_stepping",
|
|
"_gate_lock",
|
|
"_graph_lock",
|
|
"_queue",
|
|
"_node_pool",
|
|
"history_limits",
|
|
"observer",
|
|
"emission_observer",
|
|
"run_cache",
|
|
"run_id",
|
|
)
|
|
|
|
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,
|
|
run_cache: RunCacheLookup | None = None,
|
|
run_id: str = "",
|
|
) -> 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()
|
|
# Held only while the graph maps are read or swapped together, never
|
|
# across an await, a node call or any I/O. Reentrant because the lazy
|
|
# `edges` build re-enters through `_graph`. It is never nested with
|
|
# `_gate_lock` or `state.lock()` in either order, which is what keeps
|
|
# three locks from needing an ordering rule.
|
|
self._graph_lock = threading.RLock()
|
|
# 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
|
|
# Set by a run that may reuse earlier results. A live pipeline has
|
|
# none: a cascade is about what just happened, not about what a node
|
|
# once returned for the same inputs.
|
|
self.run_cache = run_cache
|
|
# The batch run this pipeline belongs to, if any. A live cascade names
|
|
# the journaled item it came from instead, which is what the events
|
|
# below carry; a run has no such item, so without this its failures
|
|
# would be recorded belonging to nothing.
|
|
self.run_id = run_id
|
|
# 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] = {}
|
|
|
|
(
|
|
self.produces,
|
|
self.dependencies,
|
|
self.consumes,
|
|
self._by_id,
|
|
) = _derive(self._nodes)
|
|
|
|
self._edges: dict[Node, set[Node]] | None = None
|
|
self._execution_order: list[Node] | None = None
|
|
self._downstream_cache: dict[Node, list[Node]] = {}
|
|
|
|
self._seed(initial_values)
|
|
|
|
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."""
|
|
edges = self._edges
|
|
if edges is None:
|
|
# Under the lock the node list and the dependencies are the same
|
|
# generation, so every producer has an entry to add a consumer to.
|
|
with self._graph_lock:
|
|
edges = self._edges
|
|
if edges is None:
|
|
edges = {node: set() for node in self._nodes}
|
|
for consumer, producers in self.dependencies.items():
|
|
for producer in producers:
|
|
edges[producer].add(consumer)
|
|
self._edges = edges
|
|
return edges
|
|
|
|
def _graph(self) -> tuple[dict[Node, frozenset[Node]], dict[Node, set[Node]]]:
|
|
"""One matching view of the graph, for the length of a wave.
|
|
|
|
A replace swaps these two together; reading them a moment apart is how
|
|
a wave ends up asking the new dependencies about a node the old edges
|
|
still know, which is a KeyError rather than a wrong answer.
|
|
"""
|
|
with self._graph_lock:
|
|
return self.dependencies, self.edges
|
|
|
|
def replace_flow(
|
|
self,
|
|
flow: str,
|
|
nodes: list[Node],
|
|
initial_values: dict[str, Any] | None = None,
|
|
) -> None:
|
|
"""Swap one flow's nodes into the graph, leaving every other flow's be.
|
|
|
|
On return the graph is what a build over the same nodes would have
|
|
produced, and the flows that were not named still hold the very node
|
|
objects they held before — started, connected, and never asked to
|
|
reconnect. An empty node list removes the flow.
|
|
"""
|
|
with self._graph_lock:
|
|
# Spliced where the old ones were, so the node list stays grouped
|
|
# by flow the way a full build lays it out.
|
|
spliced: list[Node] = []
|
|
placed = False
|
|
for node in self._nodes:
|
|
if node.flow == flow:
|
|
if not placed:
|
|
spliced.extend(nodes)
|
|
placed = True
|
|
continue
|
|
spliced.append(node)
|
|
if not placed:
|
|
spliced.extend(nodes)
|
|
|
|
produces, dependencies, consumes, by_id = _derive(spliced)
|
|
|
|
# Assigned only once everything above has succeeded, and rebound
|
|
# rather than mutated: a cascade already walking the graph holds
|
|
# the old maps and finishes on them, which is the atomicity
|
|
# building a second pipeline used to give for free.
|
|
self._nodes = spliced
|
|
self.produces = produces
|
|
self.dependencies = dependencies
|
|
self.consumes = consumes
|
|
self._by_id = by_id
|
|
self._edges = None
|
|
self._execution_order = None
|
|
self._downstream_cache = {}
|
|
|
|
for node in nodes:
|
|
node.bind(self)
|
|
self._seed(initial_values)
|
|
# A rebuild clears the debugging pause of what it rebuilt, and of
|
|
# nothing else.
|
|
with self._gate_lock:
|
|
self._paused.discard(flow)
|
|
self._stepping.discard(flow)
|
|
|
|
def remove_flow(self, flow: str) -> None:
|
|
"""Take a deleted flow out of the graph."""
|
|
self.replace_flow(flow, [])
|
|
|
|
def _seed(self, initial_values: dict[str, Any] | None) -> None:
|
|
"""Give messages a starting value, without overwriting one already there.
|
|
|
|
A seeded value counts as having arrived. Writing it without its version
|
|
left the value in state at version 0, which `_check_synchronous_ready`
|
|
reads as "never published" — so a flow that was dropped and recreated
|
|
came back with its synchronous nodes waiting on inputs that were sitting
|
|
right there, reporting `active` and `ok` and never running again.
|
|
"""
|
|
if not initial_values:
|
|
return
|
|
seeded = []
|
|
with self._state.lock():
|
|
for name, value in initial_values.items():
|
|
if name not in self._state:
|
|
self._state[name] = value
|
|
seeded.append(name)
|
|
if seeded:
|
|
self._state.increment_multi([self._version_key(name) for name in seeded])
|
|
|
|
def get_node_by_id(self, nid: str) -> Node | None:
|
|
return self._by_id.get(nid)
|
|
|
|
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."""
|
|
order = self._execution_order
|
|
if order is not None:
|
|
return order
|
|
|
|
with self._graph_lock:
|
|
deps, edges = self.dependencies, self.edges
|
|
in_degree = {node: len(d) for node, d in deps.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 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]:
|
|
# Bound once: a replace rebinds the cache, and writing the memo into
|
|
# the one this call started with loses it rather than corrupting it.
|
|
cache = self._downstream_cache
|
|
if start in cache:
|
|
return cache[start]
|
|
|
|
_, edges = self._graph()
|
|
if start not in edges:
|
|
# Its flow was replaced while this cascade was on its way here.
|
|
return []
|
|
|
|
reachable: set[Node] = set()
|
|
queue = deque([start])
|
|
while queue:
|
|
for consumer in 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]
|
|
cache[start] = ordered
|
|
return ordered
|
|
|
|
# -------------------------------------------------------------------------
|
|
# 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 a value will be there when the
|
|
flow starts — from an initial, or from the run that starts it.
|
|
"""
|
|
declared = flow_inputs or {}
|
|
issues: list[ValidationIssue] = []
|
|
# Both callers hold the controller's rebuild lock, so nothing swaps
|
|
# these underneath; bound once so the report is of one graph either way.
|
|
nodes = self._nodes
|
|
produces = self.produces
|
|
|
|
ordered = set(self._topological_sort())
|
|
if len(ordered) != len(nodes):
|
|
cyclic = sorted(n.id for n in 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 nodes:
|
|
for msg_name, spec in node.requires.items():
|
|
if msg_name in 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 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 _window_split(
|
|
self, node: Node, result: dict[str, Any], now: float
|
|
) -> tuple[dict[str, Any], dict[str, Any], float, float | None]:
|
|
"""What may publish now, what its window holds back, when it ends, and
|
|
whether a flush of this node is already booked.
|
|
|
|
Reads state and writes nothing, so the same question can be asked when
|
|
a value arrives and again when the cascade carrying it is claimed.
|
|
"""
|
|
limited = {
|
|
name: spec.interval
|
|
for name, spec in node.provides.items()
|
|
if spec.interval > 0 and name in result
|
|
}
|
|
if not limited:
|
|
return result, {}, 0.0, None
|
|
|
|
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)
|
|
return passed, held, due_at, stamps.get(flush_key)
|
|
|
|
def _throttled(
|
|
self, node: Node, result: dict[str, Any]
|
|
) -> tuple[dict[str, Any], list[str]]:
|
|
"""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.
|
|
|
|
Returns what passed and the held keys the publish makes stale — the
|
|
caller drops them in the same round trip it writes the values with,
|
|
rather than one DEL per port here.
|
|
"""
|
|
now = time.time()
|
|
passed, held, due_at, pending = self._window_split(node, result, now)
|
|
stale: list[str] = []
|
|
|
|
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 passed:
|
|
spec = node.provides.get(name)
|
|
if spec is not None and spec.interval > 0:
|
|
# A fresh publish makes anything held for that port stale.
|
|
stale.append(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, pending)
|
|
return passed, stale
|
|
|
|
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. In that case
|
|
# nothing came out and there is nothing downstream to wake.
|
|
published = self.apply_outputs(node, held)
|
|
if published:
|
|
self.run_downstream(node, changed=published)
|
|
if any(spec.interval > 0 for spec in node.requires.values()):
|
|
# A wake-up that was held back, not a value that just changed —
|
|
# so this one deliberately walks everything reachable.
|
|
self._execute_parallel(
|
|
{node, *self._get_downstream(node)}, self._state, check_ready=True
|
|
)
|
|
|
|
def _increment_message_versions(self, outputs: dict[str, Any]) -> None:
|
|
self._state.increment_multi([self._version_key(name) for name in outputs])
|
|
|
|
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
|
|
|
|
self._state.update({self._delivered_key(node.id, name): now for name in due})
|
|
return True
|
|
|
|
def _is_node_ready(
|
|
self, node: Node, state: StateBackend
|
|
) -> tuple[bool, dict[str, Any]]:
|
|
"""Whether the node may run, and the inputs the answer was read from.
|
|
|
|
The read covers every input rather than only the triggering ones, so
|
|
the node itself does not have to read the same keys again a moment
|
|
later — one MGET per node per message instead of two. 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.
|
|
"""
|
|
# One MGET rather than an EXISTS per input under the global state lock.
|
|
values = state.get_present(list(node.requires))
|
|
waited_on = [msg for msg, spec in node.requires.items() if spec.trigger]
|
|
if any(msg not in values for msg in waited_on):
|
|
return False, values
|
|
|
|
if not self._input_is_due(node):
|
|
return False, values
|
|
|
|
if not node.synchronous:
|
|
return True, values
|
|
|
|
is_ready, current_versions = self._check_synchronous_ready(node)
|
|
if not is_ready:
|
|
return False, values
|
|
return self._try_acquire_synchronous_execution(node, current_versions), values
|
|
|
|
# -------------------------------------------------------------------------
|
|
# 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 or self.run_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 _from_cache(
|
|
self, node: Node, key: str, state: StateBackend, entry_id: str
|
|
) -> tuple[bool, dict[str, Any] | None]:
|
|
"""Restore an earlier run of this node: (hit, what it published).
|
|
|
|
Both halves are needed, because a node that published nothing is a
|
|
result worth restoring and looks exactly like a miss otherwise.
|
|
|
|
The outputs go into state as if the node had just returned them, which
|
|
is what everything downstream reads — a run's state namespace is its
|
|
own, so a skipped node leaves nothing behind for the next one to find.
|
|
Named for *this* node's flow rather than the one that produced them: the
|
|
same node reached through two flows publishes the same values under two
|
|
names, and downstream here looks for the ones this flow owns.
|
|
|
|
What it emitted on the way is not restored either — those values were
|
|
the story of an execution that is not happening. The run it came from is
|
|
recorded instead, and that is where its series is read from.
|
|
"""
|
|
assert self.run_cache is not None
|
|
try:
|
|
hit = self.run_cache.lookup(key)
|
|
except Exception:
|
|
# A cache that cannot answer is a cache miss, never a failed node.
|
|
logger.exception("Cache lookup failed for '%s'", node.id)
|
|
return False, None
|
|
if hit is None:
|
|
return False, None
|
|
|
|
outputs = (
|
|
{
|
|
requalify(name, hit.flow, node.flow): value
|
|
for name, value in hit.outputs.items()
|
|
}
|
|
if hit.outputs
|
|
else hit.outputs
|
|
)
|
|
if outputs:
|
|
self._record_outputs(node, outputs, state)
|
|
self._publish(
|
|
{
|
|
"type": "node_executed",
|
|
"flow": node.flow,
|
|
"node": node.id,
|
|
"outputs": len(outputs or {}),
|
|
"duration_ms": 0.0,
|
|
"run": entry_id or self.run_id,
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
self._observe(
|
|
NodeOutcome(
|
|
node=node.id,
|
|
ok=True,
|
|
cached=True,
|
|
cached_from=hit.metrics_run,
|
|
cache_key=key,
|
|
outputs=len(outputs or {}),
|
|
output_values=outputs,
|
|
artifacts={
|
|
name: value
|
|
for name, value in (outputs or {}).items()
|
|
if is_reference(value)
|
|
},
|
|
)
|
|
)
|
|
return True, outputs
|
|
|
|
def _execute_node(
|
|
self,
|
|
node: Node,
|
|
state: StateBackend,
|
|
entry_id: str = "",
|
|
overrides: dict[str, Any] | None = None,
|
|
inputs: dict[str, Any] | None = None,
|
|
) -> dict[str, Any] | None:
|
|
"""Run one node and record its outputs. Never raises.
|
|
|
|
``inputs`` is what the readiness check already read — the same keys, a
|
|
moment earlier. Reading them again here was the second of two MGETs
|
|
per node per message; running on the values that made the node ready
|
|
is also what ``overrides`` already assumes.
|
|
"""
|
|
started = time.perf_counter()
|
|
collected = logs.Collector()
|
|
try:
|
|
# One MGET. The lock this used to be read under bought nothing a
|
|
# single bulk read does not, and it was the engine's one global
|
|
# mutex — every node of every cascade queued behind it.
|
|
if inputs is None:
|
|
inputs = state.get_present(list(node.requires))
|
|
else:
|
|
inputs = dict(inputs)
|
|
if overrides:
|
|
# The value this wave is delivering wins over whatever state
|
|
# holds by now. Not written back: the newest value is still the
|
|
# one everything else reads.
|
|
inputs.update(
|
|
{
|
|
name: value
|
|
for name, value in overrides.items()
|
|
if name in node.requires
|
|
}
|
|
)
|
|
|
|
key = ""
|
|
if self.run_cache is not None and node.fingerprint:
|
|
key = run_cache_key(node.fingerprint, inputs, node.flow)
|
|
if key:
|
|
hit, restored = self._from_cache(node, key, state, entry_id)
|
|
if hit:
|
|
return restored
|
|
|
|
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)
|
|
|
|
stale: list[str] = []
|
|
if result:
|
|
result, stale = self._throttled(node, result)
|
|
|
|
if result:
|
|
self._record_outputs(node, result, state, stale)
|
|
|
|
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 or self.run_id,
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
# Guarded rather than left to `_observe`: a live cascade has no
|
|
# observer, and building this model to drop it was one pydantic
|
|
# validation per node per message on the path that runs most.
|
|
if self.observer is not None:
|
|
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)
|
|
},
|
|
cache_key=key,
|
|
# Post-throttle: what went into state is what a later
|
|
# run restoring this node has to find.
|
|
output_values=result,
|
|
)
|
|
)
|
|
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)
|
|
if self.observer is not None:
|
|
self._observe(
|
|
NodeOutcome(
|
|
node=node.id,
|
|
ok=False,
|
|
duration_ms=round((time.perf_counter() - started) * 1000, 2),
|
|
error=error,
|
|
# The run record is the only place a failure is read
|
|
# back from after the fact, so it carries the traceback
|
|
# the log panel got rather than the one line alone.
|
|
logs=collected.text + logs.node_traceback(),
|
|
)
|
|
)
|
|
return None
|
|
|
|
def _record_outputs(
|
|
self,
|
|
node: Node,
|
|
outputs: dict[str, Any],
|
|
state: StateBackend,
|
|
drop: list[str] | None = None,
|
|
) -> float:
|
|
"""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.
|
|
|
|
The four writes go in one round trip: a pipeline is a transaction, so
|
|
they land together exactly as they did when they were four calls.
|
|
Returns the timestamp they were recorded under, which the caller
|
|
stamps its own events with so they agree.
|
|
"""
|
|
ts = time.time()
|
|
state.record(
|
|
outputs,
|
|
{self._timestamp_key(name): ts for name in outputs},
|
|
[self._version_key(name) for name in outputs],
|
|
ts,
|
|
self.history_limits,
|
|
drop,
|
|
)
|
|
origin = _node_source_dump(node.id, node.local_id)
|
|
for name, value in outputs.items():
|
|
self._publish(
|
|
{
|
|
"type": "message_value",
|
|
"flow": flow_of(name),
|
|
"name": name,
|
|
"value": value,
|
|
"ts": ts,
|
|
"source": origin,
|
|
}
|
|
)
|
|
return ts
|
|
|
|
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, stale = self._throttled(node, outputs)
|
|
if not passed:
|
|
return
|
|
self._record_outputs(node, passed, self._state, stale)
|
|
if self._queue is not None:
|
|
# Journalled carrying the emitted values, as an ``emission`` item:
|
|
# the executor hands them to the nodes reading them instead of
|
|
# writing them to state a second time. That distinction is the
|
|
# whole of it — re-applying would let a mid-node emission overwrite
|
|
# the value the node returned at the end, while reading state
|
|
# instead means a consumer slower than its producer sees only the
|
|
# newest chunk and the ones between are lost. A frame of video or a
|
|
# second of speech is worth delivering; the value in state stays
|
|
# the latest, which is what everything else reads.
|
|
self._enqueue_cascade(node, passed, kind="emission")
|
|
|
|
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 set_disabled(self, flows: set[str]) -> None:
|
|
"""Which flows are stopped. Swapped whole, so a reader never sees half."""
|
|
self._disabled = frozenset(flows)
|
|
|
|
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,
|
|
changed: set[str] | None = None,
|
|
overrides: dict[str, Any] | None = None,
|
|
) -> StateBackend:
|
|
"""Execute nodes concurrently, scheduling each as its inputs arrive.
|
|
|
|
``changed`` restricts the wave to nodes something published to: one
|
|
whose triggering inputs are all untouched is completed without being
|
|
run, so what is downstream of *it* is judged on the same footing. None
|
|
runs everything the subset holds, which is what a manual run means.
|
|
|
|
``overrides`` reaches the nodes that read those names directly, and no
|
|
further: a value carried by this wave is what its readers should see,
|
|
while everything past them reads what those readers produced.
|
|
"""
|
|
# One view of the graph for the whole wave: a flow replaced halfway
|
|
# through must not have this wave asking the new dependencies about a
|
|
# node the old edges knew.
|
|
deps, edges = self._graph()
|
|
target_nodes = nodes_subset if nodes_subset is not None else set(deps)
|
|
if not target_nodes:
|
|
return state
|
|
# A wave that set out before a flow was replaced can be carrying nodes
|
|
# that are no longer in the graph. They have been stopped; running them
|
|
# is the one thing a per-flow rebuild must not let happen.
|
|
target_nodes = {n for n in target_nodes if n in deps}
|
|
if not target_nodes:
|
|
return state
|
|
|
|
in_degree = {
|
|
n: sum(1 for dep in deps[n] if dep in target_nodes) for n in target_nodes
|
|
}
|
|
|
|
# Copied: the caller's set is not this wave's to grow.
|
|
fresh = set(changed) if changed is not None else None
|
|
|
|
submitted: set[Node] = set()
|
|
skipped: set[Node] = set()
|
|
node_futures: dict[Node, Future[dict[str, Any] | None]] = {}
|
|
|
|
def untouched(n: Node) -> bool:
|
|
"""Did this wave publish nothing this node waits on?"""
|
|
assert fresh is not None
|
|
return not any(
|
|
spec.trigger and name in fresh for name, spec in n.requires.items()
|
|
)
|
|
|
|
def complete(n: Node) -> None:
|
|
"""Count a node as done without running it, freeing its consumers."""
|
|
skipped.add(n)
|
|
for consumer in edges[n]:
|
|
if consumer in target_nodes:
|
|
in_degree[consumer] -= 1
|
|
|
|
def submit_ready(executor: ThreadPoolExecutor) -> None:
|
|
# To a fixpoint: completing a node without running it frees its
|
|
# consumers, and one this pass has already walked past would
|
|
# otherwise be stranded — there is no future outstanding to bring
|
|
# the drain loop back round for it.
|
|
progressed = True
|
|
while progressed:
|
|
progressed = False
|
|
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 needs once it may run.
|
|
if self._gate_blocks(n) or in_degree[n] != 0:
|
|
continue
|
|
if fresh is not None and untouched(n):
|
|
# Ahead of the readiness check, so an unchanged node
|
|
# spends neither a delivery stamp nor a synchronous
|
|
# claim on a run it is not going to make.
|
|
complete(n)
|
|
progressed = True
|
|
continue
|
|
ready_inputs: dict[str, Any] | None = None
|
|
if check_ready:
|
|
# The values are kept whether or not the answer is yes:
|
|
# a node that may run reads them below instead of
|
|
# asking state for the same keys again.
|
|
ready, ready_inputs = self._is_node_ready(n, state)
|
|
if not ready:
|
|
# Completed, not merely passed over: a later trigger
|
|
# may still make it ready, but not running is not
|
|
# the same as blocking. Whatever it last published
|
|
# is in state, and a consumer reading that alongside
|
|
# something this wave *did* publish has every right
|
|
# to run on the pair.
|
|
#
|
|
# Passing over it held that consumer for the whole
|
|
# wave, and a node that can never be ready — an
|
|
# input nothing has ever published — took its
|
|
# consumers out for good. On this house
|
|
# `plugs.pump_run` waits on a watering pulse that
|
|
# only exists at 02:00, and it sits upstream of
|
|
# `dmx.switches`: every boiler, plug and appliance
|
|
# command reached its own message and stopped there,
|
|
# reaching the DMX universe only in the occasional
|
|
# wave the pump node happened not to be part of.
|
|
complete(n)
|
|
progressed = True
|
|
continue
|
|
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.
|
|
complete(n)
|
|
progressed = True
|
|
continue
|
|
submitted.add(n)
|
|
node_futures[n] = executor.submit(
|
|
self._execute_node, n, state, entry_id, overrides, ready_inputs
|
|
)
|
|
|
|
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()
|
|
# Before the decrement: a consumer freed by this node is
|
|
# judged on what it just published.
|
|
if result is not None and fresh is not None:
|
|
fresh.update(result)
|
|
# Decremented whatever it returned. A node that published
|
|
# nothing — rate limited, unchanged, or failed — has not
|
|
# *blocked* its consumers; it has merely given them nothing
|
|
# to read, and `untouched` above is the accurate test of
|
|
# that. Holding the count instead made a quiet producer
|
|
# veto a noisy sibling: `dmx.switches` reads both boilers
|
|
# through report-by-exception nodes, so the kitchen one
|
|
# staying silent — which is almost always — kept the main
|
|
# boiler's command from ever reaching the DMX universe.
|
|
for consumer in 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:
|
|
# `update` is already one transaction; the lock around it was not
|
|
# holding anything else still.
|
|
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) -> set[str]:
|
|
"""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.
|
|
|
|
Returns the message names that actually reached state — post rate
|
|
limiting — which is what the cascade behind it has to walk from.
|
|
"""
|
|
stale: list[str] = []
|
|
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, stale = self._throttled(node, outputs)
|
|
|
|
if not outputs:
|
|
return set()
|
|
|
|
ts = self._record_outputs(node, outputs, self._state, stale)
|
|
# 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,
|
|
}
|
|
)
|
|
return set(outputs)
|
|
|
|
def run_downstream(
|
|
self,
|
|
node: Node,
|
|
entry_id: str = "",
|
|
replay: bool = False,
|
|
changed: set[str] | None = None,
|
|
overrides: dict[str, Any] | None = None,
|
|
) -> 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.
|
|
|
|
``changed`` names the messages this wave actually published. A node
|
|
none of whose triggering inputs are in it would re-run on values it has
|
|
already read, which is how one node came to publish 619 messages a
|
|
minute off inputs that changed six times. ``None`` means walk
|
|
everything reachable, which is what a manual run wants.
|
|
|
|
``overrides`` carries values to hand to whoever reads them instead of
|
|
what state holds — an emission delivering the chunk that caused this
|
|
wave rather than whichever one is newest by the time it runs.
|
|
"""
|
|
if changed is not None and not changed:
|
|
# Everything the cascade carried was held back by a rate limit, so
|
|
# there is nothing new for anything downstream to read.
|
|
return self._state
|
|
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,
|
|
changed=changed,
|
|
overrides=overrides,
|
|
)
|
|
|
|
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 costs at most the last second of journaled work
|
|
rather than the whole wave. 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 parks the
|
|
item, so nothing of it publishes until the flow is resumed or stepped.
|
|
|
|
A value whose every port is inside its rate-limit window is kept here
|
|
and never journaled at all: the limit is about how often the value
|
|
goes out, and enqueueing a cascade to discover that costs a run record
|
|
and a walk of everything downstream per message.
|
|
"""
|
|
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:
|
|
if not self._held_at_source(node, outputs):
|
|
self._enqueue_cascade(node, outputs)
|
|
return state
|
|
|
|
return self._run_here(node, outputs)
|
|
|
|
def _held_at_source(self, node: Node, outputs: dict[str, Any] | None) -> bool:
|
|
"""Keep a value back before it costs a cascade, or say it has to run.
|
|
|
|
Only when *every* port is inside its window: anything that would
|
|
publish leaves the exact split to ``apply_outputs`` at claim time,
|
|
which is the one place it has always been made.
|
|
"""
|
|
if not outputs:
|
|
# A cascade carrying nothing exists to walk, not to publish.
|
|
return False
|
|
try:
|
|
now = time.time()
|
|
passed, held, due_at, pending = self._window_split(node, outputs, now)
|
|
if passed or not held:
|
|
return False
|
|
self._state.update(
|
|
{self._held_key(name): value for name, value in held.items()}
|
|
)
|
|
self._schedule_flush(node, due_at, now, pending)
|
|
return True
|
|
except Exception:
|
|
# Journalling it is what this was avoiding, not what it depends on.
|
|
logger.exception("Could not hold '%s' back at the source", node.id)
|
|
return False
|
|
|
|
def _run_here(
|
|
self,
|
|
node: Node,
|
|
outputs: dict[str, Any] | None,
|
|
cause: str = "manual",
|
|
delivered: bool = False,
|
|
) -> 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.
|
|
|
|
``delivered`` marks an emission the queue could not take: its values are
|
|
already in state, so they are handed to their readers rather than
|
|
written again.
|
|
"""
|
|
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:
|
|
published = (
|
|
set(outputs or {}) if delivered else self.apply_outputs(node, outputs)
|
|
)
|
|
state = self.run_downstream(
|
|
node,
|
|
entry_id=run_id,
|
|
# No payload means the value is already in state and this is a
|
|
# wake-up, which has nothing to name as changed.
|
|
changed=published if outputs else None,
|
|
overrides=outputs if delivered else None,
|
|
)
|
|
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()
|
|
self._state.record(
|
|
values,
|
|
{self._timestamp_key(name): ts for name in values},
|
|
[self._version_key(name) for name in values],
|
|
ts,
|
|
self.history_limits,
|
|
)
|
|
source_dump = origin.model_dump()
|
|
for name, value in values.items():
|
|
self._publish(
|
|
{
|
|
"type": "message_value",
|
|
"flow": flow_of(name),
|
|
"name": name,
|
|
"value": value,
|
|
"ts": ts,
|
|
"source": source_dump,
|
|
}
|
|
)
|
|
|
|
targets: set[Node] = set()
|
|
for name in values:
|
|
for consumer in self.consumes.get(name, ()):
|
|
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 fluksio.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, kind: str = "cascade"
|
|
) -> None:
|
|
"""Journal a trigger, or fall back to running it here if that fails."""
|
|
from fluksio.flow.queue import WorkItem
|
|
|
|
item = WorkItem(
|
|
kind=kind,
|
|
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", delivered=kind == "emission")
|
|
|
|
def values(self, flow: str | None = None) -> dict[str, dict[str, Any]]:
|
|
"""Last value and timestamp of every message, optionally one flow's.
|
|
|
|
Two reads whatever the message count: this is what every websocket
|
|
snapshot calls, and it used to be a round trip per value and another
|
|
per timestamp, one at a time under the global state lock.
|
|
"""
|
|
keys = [
|
|
k for k in self._state.message_names() if not flow or flow_of(k) == flow
|
|
]
|
|
if not keys:
|
|
return {}
|
|
stamps = [self._timestamp_key(k) for k in keys]
|
|
found = self._state.get_multi(keys + stamps)
|
|
return {
|
|
key: {"value": found.get(key), "ts": found.get(ts)}
|
|
for key, ts in zip(keys, stamps, strict=True)
|
|
}
|
|
|
|
def reset(self) -> None:
|
|
self._state.clear()
|