Cut the round trips a message costs the engine

Measured with `make bench-engine` against a real Redis: 103.6 -> 164.4
messages a second on a five-node chain (p50 latency 2125 -> 1171 ms) and
34.8 -> 63.2 on a fan-out of twenty. Against the memory backend, which is
what a pip install runs on, 262 -> 626.

The two that bought most of it:

- `StateBackend.record` puts a published value, its timestamp, its series
  and its version counter in one round trip. They were four calls building
  four pipelines, and a value crossing an edge pays them twice. A released
  rate-limit hold rides along instead of a DEL per port.
- the readiness check reads a node's inputs and hands them to the node,
  rather than reading the triggering ones to count them and having the node
  read the same keys again a moment later.

`apply_outputs` was a second copy of `_record_outputs` and is now the same
code plus the event that distinguishes it.

The rest, each small:

- `_derive` builds a node-by-id map and a `consumes` index, so dispatching
  an item and publishing a value stop scanning every node in the
  installation.
- `read_all` is memoised against the store revision — it sits on the
  publish path, so a dashboard slider was reading and validating every
  flow file per value. Same mechanism `_wiring` already uses.
- the `message_value` source block is built once per node instead of per
  emission.
- both timer threads ask the queue to promote only when something is
  actually due, which takes an idle engine from ~4 Redis round trips a
  second to one.
- the shared httpx client is bounded (32 connections, one retry); its
  default pool is 100 with no per-host cap, so one slow endpoint could
  take it and every other sender node with it.
- the MQTT and delay nodes no longer log a line per message at INFO.

Robustness, in the same pass:

- `MemoryWorkQueue._done` was a set nothing ever removed from — one entry
  per non-idempotent node per item, for the life of the process, in the
  default configuration. Capped, the way the Redis side expires its
  markers.
- a saturated engine can claim from the due lane past the cascade limit.
  The capacity gate sits in front of the claim, so the due lane's priority
  — decided inside it — did not apply while every slot was held: a motor's
  stop was not behind the long nodes, it was unread. Only after a slot has
  genuinely failed to free for half a second, and briefly, so the backlog
  is not starved in turn.
- `reclaim_stale` dispatches through that same gate. It could return sixty
  entries and push in-flight far past the limit the gate exists to hold.
- a flow's nodes are stopped together rather than one after another. Each
  gets `NODE_STOP_TIMEOUT`, so a flow whose broker was unreachable took
  five seconds per node — long enough to outlast `REBUILD_WAIT` and 503
  the deploy.
- the worker pool and the HTTP client are closed on a thread, not on the
  event loop, and a run closes the state backend it built (on Redis, a
  client and a connection pool per run).
- the five background tasks say something when they die. Each catches
  exceptions inside its loop, so one raised anywhere else left the engine
  serving with no metrics, no alerts or no artifact sweep, silently.

`tests/flow/test_round_trips.py` counts the state operations one message
costs — four, where it was about eleven — because none of the above would
fail a behavioural test if it were undone.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01M6hPWS6YEbT1P8LxhhFb2T

backend/fluksio/flow/controller.py

@@ -825,13 +825,19 @@ class FlowController:
         return merged
 
     async def _teardown(self, flow: str | None = None) -> None:
-        """Stop everything the previous pipeline started, or one flow's share."""
-        for entry in self.loaded.values():
+        """Stop everything the previous pipeline started, or one flow's share.
+
+        Together rather than one after another: each node is given
+        `NODE_STOP_TIMEOUT`, so in sequence a flow whose broker is unreachable
+        took that many seconds *per node* — long enough for a rebuild to
+        outlast `REBUILD_WAIT` and answer 503 on every deploy. Concurrently it
+        is five seconds flat however many nodes there are.
+        """
+
+        async def stop(entry: LoadedNode) -> None:
             node = entry.node
             if node is None:
-                continue
-            if flow is not None and entry.flow != flow:
-                continue
+                return
             try:
                 await asyncio.wait_for(node.stop(self.app), NODE_STOP_TIMEOUT)
             except TimeoutError:
@@ -844,6 +850,14 @@ class FlowController:
                 )
             except Exception:
                 logger.exception("Error stopping node '%s'", entry.id)
+
+        await asyncio.gather(
+            *(
+                stop(entry)
+                for entry in list(self.loaded.values())
+                if flow is None or entry.flow == flow
+            )
+        )
         # After the nodes, so a loop still winding down is not restarted.
         if flow is None:
             await self.supervisor.cancel_all()

backend/fluksio/flow/executor.py

@@ -41,6 +41,13 @@ TOUCH_INTERVAL_S = 20.0
 # resolution of a delay: a delayed item is waited for exactly, so what a timer
 # fires late by is a wake-up and a promotion rather than up to a whole second.
 DELAYED_INTERVAL_S = 1.0
+# Cascade slots a promoted timer may use past `max_cascades`. A due item was
+# already waited for, so making it queue behind whatever long node happens to
+# hold the pool is the one lateness the sleeping timer thread cannot remove.
+DUE_RESERVE = 2
+# How long the saturated engine waits on the due lane before going back to
+# check whether a cascade slot has freed.
+DUE_CLAIM_BLOCK_MS = 200
 #: How many cascades may be in flight, unless the service is given a number.
 #: Sustained throughput is this over the mean cascade time, so an installation
 #: whose nodes wait on a network rather than a CPU may want more of them —
@@ -172,11 +179,17 @@ class ExecutionService:
             if not self._intake.is_set():
                 self._intake.wait(timeout=0.5)
                 continue
-            free = self._await_capacity()
+            free, due_only = self._await_capacity()
             if not free:
                 continue
             try:
-                items = self.queue.claim(free, CLAIM_BLOCK_MS)
+                items = self.queue.claim(
+                    free,
+                    # Briefly, in the due-only case: this is the saturated
+                    # engine, and a slot freeing has to be noticed promptly.
+                    DUE_CLAIM_BLOCK_MS if due_only else CLAIM_BLOCK_MS,
+                    due_only,
+                )
                 failures = 0
             except Exception as exc:
                 failures += 1
@@ -189,7 +202,7 @@ class ExecutionService:
             for item in items:
                 self._dispatch(item)
 
-    def _sleep_until_due(self) -> None:
+    def _sleep_until_due(self) -> bool:
         """Wait for the soonest deadline, the housekeeping cap, or a new one.
 
         A fixed poll here made every delayed item late by 0-1000ms whatever the
@@ -209,6 +222,10 @@ class ExecutionService:
             due = None
         wait = DELAYED_INTERVAL_S if due is None else due - time.time()
         self._timer_wake.wait(min(max(wait, 0.0), DELAYED_INTERVAL_S))
+        # What the caller promotes for: the deadline this woke for has passed,
+        # or the read failed and it should look anyway. An idle engine reads
+        # `next_due` once a second and asks for nothing.
+        return due is None or due <= time.time()
 
     def _tick(self) -> None:
         """Promote delayed items, and take back what a dead engine dropped."""
@@ -216,11 +233,12 @@ class ExecutionService:
         last_touch = 0.0
         last_backlog = 0.0
         while not self._stop.is_set():
-            self._sleep_until_due()
+            promote = self._sleep_until_due()
             if self._stop.is_set():
                 break
             try:
-                self.queue.move_due(time.time())
+                if promote:
+                    self.queue.move_due(time.time())
             except Exception as exc:
                 logger.error("Could not promote delayed work: %s", exc)
                 # The item is still due, so the wait above would be zero and
@@ -255,6 +273,12 @@ class ExecutionService:
                         item.node,
                         item.deliveries,
                     )
+                    # Through the same gate the main loop uses: a reclaim can
+                    # return sixty-odd entries at once, and dispatching them
+                    # all would push `_inflight` far past `max_cascades` —
+                    # exactly the overcommit the gate exists to prevent.
+                    if not self._await_capacity()[0]:
+                        break
                     self._dispatch(item)
             except Exception as exc:
                 logger.error("Could not reclaim stale work: %s", exc)
@@ -291,8 +315,8 @@ class ExecutionService:
             }
         )
 
-    def _await_capacity(self) -> int:
-        """How many cascades may be claimed now. Zero means the service stops.
+    def _await_capacity(self) -> tuple[int, bool]:
+        """How many cascades may be claimed now, and whether only due ones.
 
         Claiming past what the pool can run makes nothing faster: the extra
         items queue up inside the pool, count as in flight and hold their
@@ -300,11 +324,29 @@ class ExecutionService:
         came to report hundreds busy on a healthy engine. Work left in the
         stream is work that is still anyone's to take; work that is claimed is
         work that is actually being run.
+
+        The gate sat in front of the claim, though, and the due lane's
+        priority is decided *inside* it — so with every slot held by a long
+        node, a motor's stop was not merely behind them, it was unread. Past
+        the limit this therefore keeps claiming, from the due lane alone:
+        a promoted timer is work that was already waited for, and there are
+        only ever as many of them as there are deadlines.
         """
         with self._inflight_lock:
-            while self._inflight >= self.max_cascades and not self._stop.is_set():
-                self._inflight_lock.wait(0.5)
-            return 0 if self._stop.is_set() else self.max_cascades - self._inflight
+            while not self._stop.is_set():
+                free = self.max_cascades - self._inflight
+                if free > 0:
+                    return free, False
+                # Only once a slot has genuinely failed to free: the due lane
+                # is usually empty, and going to look at it ahead of waiting
+                # would leave the backlog unclaimed for the length of that
+                # read every time the pool filled up.
+                if self._inflight_lock.wait(0.5):
+                    continue
+                reserve = self.max_cascades + DUE_RESERVE - self._inflight
+                if reserve > 0:
+                    return reserve, True
+            return 0, False
 
     def _dispatch(self, item: WorkItem) -> None:
         with self._inflight_lock:

backend/fluksio/flow/nodes/delay.py

@@ -132,7 +132,7 @@ class DelayNode(Node):
     def _f(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None:
         """Forward messages with optional delay, rate-limiting, and alarm."""
 
-        logger.info("[%s] Received %s", self.name, kwargs)
+        logger.debug("[%s] Received %s", self.name, kwargs)
 
         # Store last input for cron use
         if kwargs:
@@ -142,7 +142,7 @@ class DelayNode(Node):
         if self.interval > 0:
             ts = time.time()
             if ts <= self.ts + self.interval:
-                logger.info("[%s] Stashing %s", self.name, kwargs)
+                logger.debug("[%s] Stashing %s", self.name, kwargs)
                 return None
             self.ts = ts
 
@@ -162,11 +162,13 @@ class DelayNode(Node):
             if self._pipeline is not None and self._pipeline.defer(
                 self, self._to_messages(output) or {}, self.delay
             ):
-                logger.info("[%s] Sending %s in %ss", self.name, output, self.delay)
+                logger.debug(
+                    "[%s] Sending %s in %ss", self.name, output, self.delay
+                )
                 return None
             time.sleep(self.delay)
 
-        logger.info("[%s] Sending %s", self.name, output)
+        logger.debug("[%s] Sending %s", self.name, output)
         return output
 
     # -----------------------------------------------------------------

backend/fluksio/flow/nodes/http.py

@@ -24,16 +24,35 @@ logger = logging.getLogger(__name__)
 
 # One pooled client for every sender node: connections are the expensive part
 # of an HTTP request, and a node that fires every second should keep its own.
+# The ceiling is process-wide rather than per node, which is why it is a
+# constant here rather than a `Params` field on the node.
+HTTP_POOL_LIMIT = 32
 _client: httpx.Client | None = None
 _client_lock = threading.Lock()
 
 
 def shared_client() -> httpx.Client:
-    """The process-wide HTTP client, built on first use."""
+    """The process-wide HTTP client, built on first use.
+
+    Bounded on purpose: httpx's default pool is 100 connections with no
+    per-host cap, so one endpoint that stops answering could take the whole
+    pool and every other sender node with it. Read outside the lock once it
+    exists — this is on the per-request path, and rebinding the global is
+    what the lock is for.
+    """
     global _client
+    if _client is not None:
+        return _client
     with _client_lock:
         if _client is None:
-            _client = httpx.Client()
+            _client = httpx.Client(
+                limits=httpx.Limits(
+                    max_connections=HTTP_POOL_LIMIT,
+                    max_keepalive_connections=HTTP_POOL_LIMIT // 2,
+                    keepalive_expiry=30.0,
+                ),
+                transport=httpx.HTTPTransport(retries=1),
+            )
         return _client
 
 

backend/fluksio/flow/nodes/mqtt.py

@@ -613,7 +613,10 @@ class MqttNode(Node):
                             payload = message.payload.decode("utf-8")
                             incoming_topic = str(message.topic)
 
-                            logger.info(
+                            # Debug, not info: this is one formatted line and
+                            # one write per message the broker sends, and the
+                            # payload goes to the server log verbatim.
+                            logger.debug(
                                 "[%s] Received on %s: %s",
                                 self.name,
                                 incoming_topic,

backend/fluksio/flow/pipeline.py

@@ -23,6 +23,7 @@ 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
@@ -98,6 +99,17 @@ 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.
 
@@ -184,19 +196,33 @@ def run_cache_key(fingerprint: str, inputs: dict[str, Any], flow: str = "") -> s
 
 def _derive(
     nodes: list[Node],
-) -> tuple[dict[str, list[Node]], dict[Node, frozenset[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
+    installation, 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
@@ -211,7 +237,7 @@ def _derive(
         )
         for node in nodes
     }
-    return produces, dependencies
+    return produces, dependencies, consumes, by_id
 
 
 class Pipeline:
@@ -224,6 +250,8 @@ class Pipeline:
         "_max_workers",
         "produces",
         "dependencies",
+        "consumes",
+        "_by_id",
         "_edges",
         "_execution_order",
         "_downstream_cache",
@@ -298,7 +326,12 @@ class Pipeline:
         # than the default puts its message in here. Swapped, never mutated.
         self.history_limits: dict[str, int] = {}
 
-        self.produces, self.dependencies = _derive(self._nodes)
+        (
+            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
@@ -376,7 +409,7 @@ class Pipeline:
             if not placed:
                 spliced.extend(nodes)
 
-            produces, dependencies = _derive(spliced)
+            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
@@ -385,6 +418,8 @@ class Pipeline:
             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 = {}
@@ -423,7 +458,7 @@ class Pipeline:
             self._state.increment_multi([self._version_key(name) for name in seeded])
 
     def get_node_by_id(self, nid: str) -> Node | None:
-        return next((n for n in self._nodes if n.id == nid), 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}
@@ -634,16 +669,23 @@ class Pipeline:
                 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]) -> dict[str, Any]:
+    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
@@ -652,13 +694,13 @@ class Pipeline:
                 spec = node.provides.get(name)
                 if spec is not None and spec.interval > 0:
                     # A fresh publish makes anything held for that port stale.
-                    self._state.delete(self._held_key(name))
+                    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
+        return passed, stale
 
     def _schedule_flush(
         self, node: Node, at: float, now: float, pending: float | None
@@ -791,25 +833,34 @@ class Pipeline:
         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) -> bool:
-        # 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.
-        waited_on = [msg for msg, spec in node.requires.items() if spec.trigger]
+    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.
-        if len(state.get_present(waited_on)) != len(waited_on):
-            return False
+        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
+            return False, values
 
         if not node.synchronous:
-            return True
+            return True, values
 
         is_ready, current_versions = self._check_synchronous_ready(node)
         if not is_ready:
-            return False
-        return self._try_acquire_synchronous_execution(node, current_versions)
+            return False, values
+        return self._try_acquire_synchronous_execution(node, current_versions), values
 
     # -------------------------------------------------------------------------
     # Execution
@@ -952,15 +1003,25 @@ class Pipeline:
         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."""
+        """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.
-            inputs = state.get_present(list(node.requires))
+            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
@@ -999,11 +1060,12 @@ class Pipeline:
                 except Exception as exc:
                     logger.warning("Could not mark '%s' done: %s", node.id, exc)
 
+            stale: list[str] = []
             if result:
-                result = self._throttled(node, result)
+                result, stale = self._throttled(node, result)
 
             if result:
-                self._record_outputs(node, result, state)
+                self._record_outputs(node, result, state, stale)
 
             duration_ms = round((time.perf_counter() - started) * 1000, 2)
             self._publish(
@@ -1058,24 +1120,34 @@ class Pipeline:
             return None
 
     def _record_outputs(
-        self, node: Node, outputs: dict[str, Any], state: StateBackend
-    ) -> None:
+        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()
-        # Value and timestamp in one write, which is what the lock around two
-        # of them was for — a pipeline is a transaction, so they still land
-        # together and nobody waits on a mutex to do it.
-        state.update({**outputs, **{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).model_dump()
+        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(
                 {
@@ -1087,6 +1159,7 @@ class Pipeline:
                     "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.
@@ -1103,10 +1176,10 @@ class Pipeline:
         with no meaning.
         """
         self._observe_emission(node, outputs)
-        passed = self._throttled(node, outputs)
+        passed, stale = self._throttled(node, outputs)
         if not passed:
             return
-        self._record_outputs(node, passed, self._state)
+        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
@@ -1277,11 +1350,15 @@ class Pipeline:
                         complete(n)
                         progressed = True
                         continue
-                    if check_ready and not self._is_node_ready(n, state):
-                        if n.synchronous:
-                            # Not ready now; a later trigger may make it ready.
-                            skipped.add(n)
-                        continue
+                    ready_inputs: dict[str, Any] | None = None
+                    if check_ready:
+                        ready, ready_inputs = self._is_node_ready(n, state)
+                        if not ready:
+                            if n.synchronous:
+                                # Not ready now; a later trigger may make it
+                                # ready.
+                                skipped.add(n)
+                            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.
@@ -1290,7 +1367,7 @@ class Pipeline:
                         continue
                     submitted.add(n)
                     node_futures[n] = executor.submit(
-                        self._execute_node, n, state, entry_id, overrides
+                        self._execute_node, n, state, entry_id, overrides, ready_inputs
                     )
 
         def drain(executor: ThreadPoolExecutor) -> None:
@@ -1345,31 +1422,16 @@ class Pipeline:
         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 = self._throttled(node, outputs)
+            outputs, stale = self._throttled(node, outputs)
 
         if not outputs:
             return set()
 
-        state = self._state
-        ts = time.time()
-        state.update({**outputs, **{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).model_dump()
-        for name, value in outputs.items():
-            self._publish(
-                {
-                    "type": "message_value",
-                    "flow": flow_of(name),
-                    "name": name,
-                    "value": value,
-                    "ts": ts,
-                    "source": origin,
-                }
-            )
+        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(
@@ -1559,11 +1621,13 @@ class Pipeline:
         origin = source or ValueSource(kind="api", label="API")
 
         ts = time.time()
-        self._state.update(
-            {**values, **{self._timestamp_key(name): ts for name in values}}
+        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,
         )
-        self._state.append_history(values, ts, self.history_limits)
-        self._increment_message_versions(values)
         source_dump = origin.model_dump()
         for name, value in values.items():
             self._publish(
@@ -1579,10 +1643,9 @@ class Pipeline:
 
         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))
+            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)
 

backend/fluksio/flow/queue.py

@@ -17,7 +17,7 @@ import logging
 import threading
 import time
 from abc import ABC, abstractmethod
-from collections import deque
+from collections import OrderedDict, deque
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from typing import Any, cast
@@ -41,6 +41,10 @@ DUE_PREFIX = "due:"
 # A consumer this far past its last read belongs to an engine that is gone. A
 # live one interacts every claim, so nothing in service comes close.
 STALE_CONSUMER_IDLE_MS = 3_600_000
+# How many "this side effect already happened" markers the memory queue keeps.
+# The Redis one expires each after an hour; this is the same idea sized by
+# count, since redelivery happens seconds after the claim and never later.
+DONE_MARKERS = 10_000
 
 
 @dataclass
@@ -154,8 +158,15 @@ class WorkQueue(ABC):
         """
 
     @abstractmethod
-    def claim(self, count: int, block_ms: int) -> list[WorkItem]:
-        """Take up to ``count`` items, waiting up to ``block_ms`` for one."""
+    def claim(
+        self, count: int, block_ms: int, due_only: bool = False
+    ) -> list[WorkItem]:
+        """Take up to ``count`` items, waiting up to ``block_ms`` for one.
+
+        ``due_only`` takes nothing but promoted timers. A saturated engine
+        uses it to let a motor's stop past the cascade limit while every slot
+        is held by a long node.
+        """
 
     @abstractmethod
     def ack(self, item: WorkItem) -> None:
@@ -240,9 +251,17 @@ class MemoryWorkQueue(WorkQueue):
 
     def __init__(self) -> None:
         self._items: deque[WorkItem] = deque()
+        # Promoted timers, kept apart from the backlog the way Redis keeps
+        # two streams — so they can be claimed on their own.
+        self._due: deque[WorkItem] = deque()
         self._delayed: list[tuple[float, int, WorkItem]] = []
         self._parked: dict[str, list[WorkItem]] = {}
-        self._done: set[tuple[str, str]] = set()
+        # Insertion-ordered and capped, because this grows one entry per
+        # non-idempotent node per item and nothing ever removed one — the
+        # Redis side expires its markers after an hour, this one leaked for
+        # the life of the process. Redelivery is what the marker guards, and
+        # that happens within seconds of the claim.
+        self._done: OrderedDict[tuple[str, str], None] = OrderedDict()
         # Claimed and not yet acknowledged, which is what Redis's `pending` is.
         self._in_flight = 0
         self._counter = 0
@@ -277,17 +296,25 @@ class MemoryWorkQueue(WorkQueue):
         with self._lock:
             return self._delayed[0][0] if self._delayed else None
 
-    def claim(self, count: int, block_ms: int) -> list[WorkItem]:
+    def claim(
+        self, count: int, block_ms: int, due_only: bool = False
+    ) -> list[WorkItem]:
         deadline = time.monotonic() + block_ms / 1000.0
         with self._wake:
-            while not self._items:
+            while not self._due and not (self._items and not due_only):
                 remaining = deadline - time.monotonic()
                 if remaining <= 0:
                     return []
                 self._wake.wait(remaining)
-            claimed = [
-                self._items.popleft() for _ in range(min(count, len(self._items)))
-            ]
+            # Due first: an item that has waited out a deadline is late by
+            # however long it queues here, while work merely enqueued is not
+            # waiting on a clock.
+            claimed = [self._due.popleft() for _ in range(min(count, len(self._due)))]
+            if not due_only:
+                claimed += [
+                    self._items.popleft()
+                    for _ in range(min(count - len(claimed), len(self._items)))
+                ]
             self._in_flight += len(claimed)
             return claimed
 
@@ -306,10 +333,7 @@ class MemoryWorkQueue(WorkQueue):
             while self._delayed and self._delayed[0][0] <= now:
                 due.append(heapq.heappop(self._delayed)[2])
             if due:
-                # In front of the backlog, in due order: an item that has
-                # waited out a deadline is late by however long it queues
-                # here, while work merely enqueued is not waiting on a clock.
-                self._items.extendleft(reversed(due))
+                self._due.extend(due)
                 self._wake.notify()
         return len(due)
 
@@ -335,12 +359,12 @@ class MemoryWorkQueue(WorkQueue):
 
     def backlog(self) -> int:
         with self._lock:
-            return len(self._items)
+            return len(self._items) + len(self._due)
 
     def stats(self) -> dict[str, Any]:
         with self._lock:
             return {
-                "backlog": len(self._items),
+                "backlog": len(self._items) + len(self._due),
                 "pending": self._in_flight,
                 "delayed": len(self._delayed),
                 "parked": sum(len(v) for v in self._parked.values()),
@@ -354,7 +378,9 @@ class MemoryWorkQueue(WorkQueue):
 
     def mark_done(self, entry_id: str, node: str) -> None:
         with self._lock:
-            self._done.add((entry_id, node))
+            self._done[(entry_id, node)] = None
+            while len(self._done) > DONE_MARKERS:
+                self._done.popitem(last=False)
 
     def was_done(self, entry_id: str, node: str) -> bool:
         with self._lock:
@@ -467,7 +493,9 @@ class RedisWorkQueue(WorkQueue):
             return self._due_stream, entry_id[len(DUE_PREFIX) :]
         return self._stream, entry_id
 
-    def claim(self, count: int, block_ms: int) -> list[WorkItem]:
+    def claim(
+        self, count: int, block_ms: int, due_only: bool = False
+    ) -> list[WorkItem]:
         """Take up to ``count`` items, due timers before anything queued.
 
         One read over both streams rather than a read each: the block has to
@@ -483,7 +511,11 @@ class RedisWorkQueue(WorkQueue):
             self._redis.xreadgroup(
                 GROUP,
                 self._consumer,
-                {self._due_stream: ">", self._stream: ">"},
+                (
+                    {self._due_stream: ">"}
+                    if due_only
+                    else {self._due_stream: ">", self._stream: ">"}
+                ),
                 count=count,
                 block=block_ms,
             ),

backend/fluksio/flow/runs.py

@@ -968,8 +968,12 @@ class RunService:
         while not self._stop.is_set():
             try:
                 # Runs put back to wait for a worker come due here. The claim
-                # below blocks for a second, so this is about once a second.
-                self.queue.move_due(time.time())
+                # below blocks for a second, so this is about once a second —
+                # and asks the queue for nothing at all while nothing is
+                # waiting, which is the common case.
+                due = self.queue.next_due()
+                if due is not None and due <= time.time():
+                    self.queue.move_due(time.time())
                 items = self.queue.claim(
                     max(CLAIM_COUNT, self.parallel), CLAIM_BLOCK_MS
                 )
@@ -1199,11 +1203,18 @@ class RunService:
             self._release_cards(run)
             # Its values were only ever this run's; nothing reads them once it
             # has a result. On Redis the namespace would expire anyway.
-            if state is not None and status != "error":
+            if state is not None:
+                if status != "error":
+                    try:
+                        state.clear()
+                    except Exception:
+                        logger.warning("Could not clear state of run %s", run_id)
+                # Even when the values are kept: on Redis this backend is a
+                # client and a connection pool of its own, built per run.
                 try:
-                    state.clear()
+                    state.close()
                 except Exception:
-                    logger.warning("Could not clear state of run %s", run_id)
+                    logger.warning("Could not close state of run %s", run_id)
 
     def _release_cards(self, run: Run) -> None:
         """Hand a GPU run's device memory back when the run is over.

backend/fluksio/flow/state.py

@@ -98,6 +98,15 @@ class StateBackend(ABC):
         """Clear all keys in the state."""
         ...
 
+    def close(self) -> None:  # noqa: B027
+        """Release whatever the backend holds outside this process.
+
+        A run builds a state backend of its own and drops it when it
+        finishes; on Redis that is a client and its connection pool, which
+        nothing was giving back. Concrete rather than abstract: a backend that
+        holds nothing has nothing to answer here.
+        """
+
     @abstractmethod
     def keys(self) -> list[str]:
         """
@@ -221,6 +230,36 @@ class StateBackend(ABC):
         """
         ...
 
+    def record(
+        self,
+        values: dict[str, Any],
+        stamps: dict[str, float],
+        counters: list[str],
+        ts: float,
+        limits: dict[str, int] | None = None,
+        drop: list[str] | None = None,
+    ) -> None:
+        """Everything a published value owes state, in one round trip.
+
+        The value itself, the timestamp beside it, its series and its version
+        counter used to be four calls and four round trips; a backend that can
+        batch them does so here. This default is those four calls, so a
+        backend that cannot batch gains nothing and breaks nothing.
+
+        :param values: Message names mapped to the value just published.
+        :param stamps: The timestamp key of each of those messages.
+        :param counters: The version keys to bump.
+        :param ts: When they were published.
+        :param limits: How many points to keep per message.
+        :param drop: Keys to forget in the same trip — a released rate-limit
+            hold, which is written and cleared on this same path.
+        """
+        for key in drop or ():
+            self.delete(key)
+        self.update({**values, **stamps})
+        self.append_history(values, ts, limits)
+        self.increment_multi(counters)
+
     @abstractmethod
     def history(self, key: str) -> list[tuple[float, float]]:
         """
@@ -369,6 +408,35 @@ class MemoryState(StateBackend):
                     self._history[key] = series
                 series.append((ts, number))
 
+    def record(
+        self,
+        values: dict[str, Any],
+        stamps: dict[str, float],
+        counters: list[str],
+        ts: float,
+        limits: dict[str, int] | None = None,
+        drop: list[str] | None = None,
+    ) -> None:
+        """The four writes under one acquisition of the one lock."""
+        with self._lock:
+            for key in drop or ():
+                self._data.pop(key, None)
+                self._history.pop(key, None)
+            self._data.update(values)
+            self._data.update(stamps)
+            for key, value in values.items():
+                number = as_number(value)
+                if number is None:
+                    continue
+                cap = (limits or {}).get(key, HISTORY_LIMIT)
+                series = self._history.get(key)
+                if series is None or series.maxlen != cap:
+                    series = deque(series or (), maxlen=cap)
+                    self._history[key] = series
+                series.append((ts, number))
+            for key in counters:
+                self._data[key] = self._data.get(key, 0) + 1
+
     def history(self, key: str) -> list[tuple[float, float]]:
         """The recorded points of one message, oldest first."""
         with self._lock:
@@ -648,6 +716,10 @@ class RedisState(StateBackend):
             # Another client modified one of the watched keys
             return False
 
+    def close(self) -> None:
+        """Give the connection pool back."""
+        self._client.close()
+
     def _history_key(self, key: str) -> str:
         return self._key(f"__history__:{key}")
 
@@ -671,6 +743,45 @@ class RedisState(StateBackend):
         if queued:
             pipe.execute()
 
+    def record(
+        self,
+        values: dict[str, Any],
+        stamps: dict[str, float],
+        counters: list[str],
+        ts: float,
+        limits: dict[str, int] | None = None,
+        drop: list[str] | None = None,
+    ) -> None:
+        """Value, timestamp, series and version counter in one round trip.
+
+        These were four pipelines — four round trips — for one published
+        value, and a value crossing an edge pays them twice. Redis executes a
+        pipeline as one transaction, so batching them changes nothing about
+        what lands together and removes three of the four trips.
+        """
+        pipe = self._client.pipeline()
+        for key in drop or ():
+            pipe.delete(self._key(key), self._history_key(key))
+        for key, value in {**values, **stamps}.items():
+            data = self._serialize(value)
+            if self._ttl:
+                pipe.setex(self._key(key), self._ttl, data)
+            else:
+                pipe.set(self._key(key), data)
+        for key, value in values.items():
+            number = as_number(value)
+            if number is None:
+                continue
+            history_key = self._history_key(key)
+            cap = (limits or {}).get(key, HISTORY_LIMIT)
+            pipe.lpush(history_key, orjson.dumps([ts, number]))
+            pipe.ltrim(history_key, 0, cap - 1)
+            if self._ttl:
+                pipe.expire(history_key, self._ttl)
+        for key in counters:
+            pipe.incr(self._key(key))
+        pipe.execute()
+
     def history(self, key: str) -> list[tuple[float, float]]:
         """The recorded points of one message, oldest first."""
         entries = cast(list[bytes], self._client.lrange(self._history_key(key), 0, -1))

backend/fluksio/flow/store.py

@@ -116,6 +116,10 @@ class FlowStore:
         #: it therefore counts from zero per process and misses an edit made on
         #: disk behind the API, which no writer here does.
         self.revision = 0
+        #: `read_all` memoised against that revision — see the method.
+        self._read_all: list[FlowDef] = []
+        self._read_all_at = -1
+        self._read_all_lock = threading.Lock()
         if not (self.root / ".git").exists():
             self._git("init", "-q")
             self._commit("Initialise flow store", allow_empty=True)
@@ -394,15 +398,25 @@ class FlowStore:
         return FlowDef.model_validate_json(path.read_text())
 
     def read_all(self) -> list[FlowDef]:
-        """Every published flow — what the engine runs."""
-        flows = []
-        for path in sorted(self.root.glob("*/flow.json")):
-            name = path.parent.name
-            try:
-                flows.append(self.read_flow(name))
-            except Exception:
-                logger.exception("Skipping unreadable flow '%s'", name)
-        return flows
+        """Every published flow — what the engine runs.
+
+        Cached against `revision`, because this sits on the publish path: a
+        dashboard slider moving asked every flow's file to be read and
+        validated again, per value. The list is rebuilt on the next commit,
+        and returned as a copy so a caller sorting it cannot disturb the next.
+        """
+        with self._read_all_lock:
+            if self._read_all_at != self.revision:
+                flows = []
+                for path in sorted(self.root.glob("*/flow.json")):
+                    name = path.parent.name
+                    try:
+                        flows.append(self.read_flow(name))
+                    except Exception:
+                        logger.exception("Skipping unreadable flow '%s'", name)
+                self._read_all = flows
+                self._read_all_at = self.revision
+            return list(self._read_all)
 
     def write_flow(self, flow: FlowDef) -> bool:
         """Publish a flow directly. Returns False when nothing actually changed."""

backend/fluksio/main.py

@@ -1,8 +1,9 @@
 import asyncio
 import contextlib
 import logging
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Coroutine
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
+from typing import Any
 
 from fastapi import FastAPI, Request
 from fastapi.concurrency import run_in_threadpool
@@ -223,14 +224,36 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     )
     watchdog = LoopWatchdog(event_bus)
     app.state.watchdog = watchdog
-    watchdog_task = asyncio.create_task(watchdog.run(), name="loop-watchdog")
-    alerts_task = asyncio.create_task(alerts.run(), name="alert-manager")
-    metrics_task = asyncio.create_task(
-        MetricsCollector(event_bus).run(), name="metrics-collector"
-    )
-    gc_task = asyncio.create_task(
-        _sweep_artifacts(artifacts, controller), name="artifact-gc"
-    )
+
+    def _background(coro: Coroutine[Any, Any, None], name: str) -> asyncio.Task[None]:
+        """Start a long-lived task that says something if it ever stops.
+
+        Each of these loops catches its own exceptions *inside* the loop, so
+        one raised anywhere else simply ended the task — an engine that went
+        on serving with no metrics, no alerts or no artifact sweep and nothing
+        anywhere saying so.
+        """
+        task = asyncio.create_task(coro, name=name)
+
+        def _finished(done: asyncio.Task[None]) -> None:
+            if done.cancelled():
+                return
+            exc = done.exception()
+            if exc is None:
+                logger.warning("Background task '%s' stopped on its own", name)
+                return
+            logger.error("Background task '%s' died: %s", name, exc, exc_info=exc)
+            event_bus.publish(
+                {"type": "engine_degraded", "detail": f"{name} stopped: {exc}"}
+            )
+
+        task.add_done_callback(_finished)
+        return task
+
+    watchdog_task = _background(watchdog.run(), "loop-watchdog")
+    alerts_task = _background(alerts.run(), "alert-manager")
+    metrics_task = _background(MetricsCollector(event_bus).run(), "metrics-collector")
+    gc_task = _background(_sweep_artifacts(artifacts, controller), "artifact-gc")
     await controller.start()
     run_service.start()
     # Optional, and off unless someone enrolled this installation: the
@@ -246,8 +269,8 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     # Watched whether or not one exists now: enrolling from the CLI writes the
     # config from another process entirely, and an engine already serving
     # should pick it up rather than need restarting.
-    enrol_task = asyncio.create_task(
-        cloud_connector.watch_enrolment(app), name="cloud-enrolment-watch"
+    enrol_task = _background(
+        cloud_connector.watch_enrolment(app), "cloud-enrolment-watch"
     )
     try:
         # A mounted sub-app gets no lifespan of its own, so the MCP session
@@ -266,12 +289,15 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
             running_cloud.cancel()
         await run_in_threadpool(run_service.stop)
         await controller.stop()
-        pool.stop()
+        # On a thread, like the two above: stopping a worker waits up to five
+        # seconds on each child, and with a pool per declared environment that
+        # is a shutdown the event loop should not be holding.
+        await run_in_threadpool(pool.stop)
         # A machine asked for and not yet arrived would hold an allocation
         # nobody is going to use.
         for provisioner in placer.provisioners:
             await run_in_threadpool(provisioner.shutdown)
-        close_shared_client()
+        await run_in_threadpool(close_shared_client)
         if settings.MCP_ENABLED:
             from fluksio.mcp.http import aclose