diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py
index 1ffce99..6f43120 100644
--- a/backend/fluksio/flow/controller.py
+++ b/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()
diff --git a/backend/fluksio/flow/executor.py b/backend/fluksio/flow/executor.py
index cbbef7a..ab9a2dd 100644
--- a/backend/fluksio/flow/executor.py
+++ b/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:
diff --git a/backend/fluksio/flow/nodes/delay.py b/backend/fluksio/flow/nodes/delay.py
index 0cc1cf3..42871ec 100644
--- a/backend/fluksio/flow/nodes/delay.py
+++ b/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
 
     # -----------------------------------------------------------------
diff --git a/backend/fluksio/flow/nodes/http.py b/backend/fluksio/flow/nodes/http.py
index b5879b2..7a4e19c 100644
--- a/backend/fluksio/flow/nodes/http.py
+++ b/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
 
 
diff --git a/backend/fluksio/flow/nodes/mqtt.py b/backend/fluksio/flow/nodes/mqtt.py
index 80ffaef..b6763ea 100644
--- a/backend/fluksio/flow/nodes/mqtt.py
+++ b/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,
diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py
index a2c82d8..b356816 100644
--- a/backend/fluksio/flow/pipeline.py
+++ b/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)
 
diff --git a/backend/fluksio/flow/queue.py b/backend/fluksio/flow/queue.py
index f40581d..417d9e5 100644
--- a/backend/fluksio/flow/queue.py
+++ b/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,
             ),
diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py
index e8a18f1..f6ea1f6 100644
--- a/backend/fluksio/flow/runs.py
+++ b/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.
diff --git a/backend/fluksio/flow/state.py b/backend/fluksio/flow/state.py
index a3b646d..9875536 100644
--- a/backend/fluksio/flow/state.py
+++ b/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))
diff --git a/backend/fluksio/flow/store.py b/backend/fluksio/flow/store.py
index 4f857d7..3d3e56d 100644
--- a/backend/fluksio/flow/store.py
+++ b/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."""
diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py
index 8b271c7..dbd4819 100644
--- a/backend/fluksio/main.py
+++ b/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
 
diff --git a/backend/tests/flow/test_queue.py b/backend/tests/flow/test_queue.py
index bde61bf..a3a8bc2 100644
--- a/backend/tests/flow/test_queue.py
+++ b/backend/tests/flow/test_queue.py
@@ -8,6 +8,7 @@ from fluksio.flow.executor import ExecutionService
 from fluksio.flow.messages import DType, MessageSpec
 from fluksio.flow.nodes import Node
 from fluksio.flow.pipeline import Pipeline
+from fluksio.flow import queue as queue_module
 from fluksio.flow.queue import MemoryWorkQueue, WorkItem
 from fluksio.flow.state import MemoryState
 
@@ -584,3 +585,44 @@ def test_a_throttled_emission_wakes_nothing():
     # The first is let through; the second is held by the interval, and a value
     # nothing published is nothing to wake on.
     assert seen == [1.0]
+
+
+def test_a_due_timer_is_claimable_while_the_backlog_is_not():
+    """What a saturated engine uses to get a motor's stop out.
+
+    Every cascade slot held by a long node used to mean the due lane was not
+    merely behind the backlog but unread, because the capacity gate sits in
+    front of the claim.
+    """
+    queue = MemoryWorkQueue()
+    queue.add(WorkItem(kind="cascade", node="f.backlog", flow="f"))
+    queue.add_delayed(
+        WorkItem(kind="cascade", node="f.timer", flow="f"), time.time() - 1
+    )
+    assert queue.move_due(time.time()) == 1
+
+    claimed = queue.claim(10, 10, due_only=True)
+    assert [item.node for item in claimed] == ["f.timer"]
+
+    # And the backlog is still there for whoever comes next.
+    assert [item.node for item in queue.claim(10, 10)] == ["f.backlog"]
+
+
+def test_a_due_only_claim_of_nothing_does_not_take_the_backlog():
+    queue = MemoryWorkQueue()
+    queue.add(WorkItem(kind="cascade", node="f.backlog", flow="f"))
+
+    assert queue.claim(10, 10, due_only=True) == []
+    assert len(queue.claim(10, 10)) == 1
+
+
+def test_done_markers_do_not_grow_without_bound():
+    """The Redis side expires them after an hour; this one is capped by count."""
+    queue = MemoryWorkQueue()
+    for i in range(queue_module.DONE_MARKERS + 500):
+        queue.mark_done(f"entry-{i}", "f.node")
+
+    assert len(queue._done) == queue_module.DONE_MARKERS
+    # The newest are what redelivery would ask about.
+    assert queue.was_done(f"entry-{queue_module.DONE_MARKERS + 499}", "f.node")
+    assert not queue.was_done("entry-0", "f.node")
diff --git a/backend/tests/flow/test_round_trips.py b/backend/tests/flow/test_round_trips.py
new file mode 100644
index 0000000..b36ba8a
--- /dev/null
+++ b/backend/tests/flow/test_round_trips.py
@@ -0,0 +1,126 @@
+"""What one message costs the state backend, in operations.
+
+The engine's time goes on round trips, not on Python: a profile against a
+real Redis put its own code at 8.5% of self-time and Redis I/O at 61%. So the
+thing worth a regression test is the *count* — a change that quietly turns one
+pipelined write back into four would not fail any behavioural test and would
+cost a fifth of the throughput.
+
+Counted against the memory backend, because the number is a property of the
+call sites rather than of the transport.
+"""
+
+from collections import Counter
+from typing import Any
+
+from fluksio.flow.messages import DType, MessageSpec
+from fluksio.flow.nodes import Node
+from fluksio.flow.pipeline import Pipeline
+from fluksio.flow.state import MemoryState
+
+COUNTED = (
+    "get",
+    "set",
+    "get_multi",
+    "get_present",
+    "update",
+    "record",
+    "append_history",
+    "increment_multi",
+    "delete",
+)
+
+
+class CountingState(MemoryState):
+    """A state backend that remembers how often it was asked for something."""
+
+    # MemoryState uses __slots__; this one needs an attribute of its own.
+    __slots__ = ("calls",)
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.calls: Counter[str] = Counter()
+
+    def __getattribute__(self, name: str) -> Any:
+        if name in COUNTED:
+            object.__getattribute__(self, "calls")[name] += 1
+        return object.__getattribute__(self, name)
+
+
+def spec(name: str) -> MessageSpec:
+    return MessageSpec(name=name, dtype=DType.FLOAT)
+
+
+def make_node(node_id: str, flow: str, f, requires=(), provides=()) -> Node:
+    node = Node(f=f, requires=list(requires), provides=list(provides), name=node_id)
+    node.assign_flow(flow, node_id)
+    return node
+
+
+def build() -> tuple[Pipeline, Node, CountingState]:
+    source = make_node(
+        "source", "chain", lambda params: {"a": 1.0}, provides=[spec("a")]
+    )
+    relay = make_node(
+        "relay",
+        "chain",
+        lambda a, params: {"b": a + 1},
+        requires=[spec("a")],
+        provides=[spec("b")],
+    )
+    sink = make_node(
+        "sink", "chain", lambda b, params: None, requires=[spec("b")]
+    )
+    state = CountingState()
+    pipeline = Pipeline([source, relay, sink], state=state)
+    return pipeline, source, state
+
+
+def test_recording_a_value_is_one_write() -> None:
+    """Value, timestamp, history and version counter go in one round trip.
+
+    They were four calls — `update`, `append_history`, `increment_multi` and a
+    `delete` per rate-limited port — and a value crossing an edge pays them
+    twice.
+    """
+    pipeline, source, state = build()
+    state.calls.clear()
+
+    pipeline.apply_outputs(source, {"chain.a": 1.0})
+
+    assert state.calls["record"] == 1
+    for method in ("update", "append_history", "increment_multi"):
+        assert state.calls[method] == 0, f"{method} should be folded into record()"
+
+
+def test_a_node_reads_its_inputs_once() -> None:
+    """Readiness and execution share one read of the same keys.
+
+    The readiness check used to read the triggering inputs and throw the
+    values away, and the node then read the same keys again to run on them.
+    """
+    pipeline, source, state = build()
+    state.calls.clear()
+
+    changed = pipeline.apply_outputs(source, {"chain.a": 1.0})
+    pipeline.run_downstream(source, changed=changed)
+
+    # Two nodes ran downstream (relay, sink); one bulk read each.
+    assert state.calls["get_present"] == 2
+
+
+def test_a_hop_costs_a_bounded_number_of_state_operations() -> None:
+    """The whole of one value crossing one edge, counted.
+
+    A ceiling rather than an exact number, so an unrelated change does not
+    fail it — but low enough that reintroducing a per-value read or a
+    split-up write does.
+    """
+    pipeline, source, state = build()
+    state.calls.clear()
+
+    changed = pipeline.apply_outputs(source, {"chain.a": 1.0})
+    pipeline.run_downstream(source, changed=changed)
+
+    total = sum(state.calls[name] for name in COUNTED)
+    assert total <= 8, f"state operations per hop grew: {dict(state.calls)}"