From 75b3a930948f2da4e5e7d300f3804a9fe3a016f0 Mon Sep 17 00:00:00 2001
From: stroblme <stroblme@posteo.de>
Date: Thu, 29 Oct 2026 14:24:17 +0100
Subject: [PATCH] Fix work recovery and reduce engine and chart overhead

---
 backend/fluksio/api/routes/flows.py           |  34 +++-
 backend/fluksio/api/routes/messages.py        |   8 +-
 backend/fluksio/cloud/connector.py            |  74 ++++---
 backend/fluksio/flow/controller.py            |   3 +
 backend/fluksio/flow/events.py                | 138 ++++++++++---
 backend/fluksio/flow/executor.py              | 147 +++++++++++---
 backend/fluksio/flow/metrics.py               |  36 ++--
 backend/fluksio/flow/pipeline.py              | 186 +++++++++---------
 backend/fluksio/flow/queue.py                 |  54 +++--
 backend/fluksio/spa.py                        |  31 ++-
 backend/tests/flow/test_event_bus.py          |  93 ++++++++-
 backend/tests/flow/test_queue.py              |  42 ++++
 backend/tests/flow/test_redis_queue.py        |  68 +++++++
 backend/tests/flow/test_round_trips.py        |  21 ++
 backend/tests/flow/test_scheduling.py         |  84 ++++++++
 backend/tests/flow/test_socket_catchup.py     | 129 ++++++++++++
 backend/tests/flow/test_timer_capacity.py     | 112 +++++++++++
 backend/tests/test_metrics.py                 |  55 ++++++
 backend/tests/test_spa.py                     |  52 +++++
 development.md                                |   4 +
 docs/reference/configuration.md               |   4 +
 frontend/src/components/Common/UplotChart.tsx |  75 ++++---
 .../src/components/Common/chart-data.check.ts |  68 +++++++
 .../src/components/Dashboard/ChartWidget.tsx  |  68 ++++---
 frontend/vite.config.ts                       |  31 ++-
 25 files changed, 1316 insertions(+), 301 deletions(-)
 create mode 100644 backend/tests/flow/test_redis_queue.py
 create mode 100644 backend/tests/flow/test_scheduling.py
 create mode 100644 backend/tests/flow/test_socket_catchup.py
 create mode 100644 backend/tests/flow/test_timer_capacity.py
 create mode 100644 frontend/src/components/Common/chart-data.check.ts

diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py
index e3f79ba..978b2be 100644
--- a/backend/fluksio/api/routes/flows.py
+++ b/backend/fluksio/api/routes/flows.py
@@ -1700,8 +1700,8 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
         await websocket.send_text(orjson.dumps(payload).decode())
         return list(payload.get("demand", []))
 
-    async def catch_up() -> bool:
-        """Rescope a panel and send the snapshot again, demand included.
+    async def rescope() -> bool:
+        """Recheck a panel and its demand before forwarding retained events.
 
         False once a device whose credential stopped standing is closed with
         1008, which sends it to check: otherwise its frames would still be
@@ -1722,12 +1722,18 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
             device_def = _device_def(websocket.app, device)
             wanted &= only
             demand.want(viewer, wanted)
+        return True
+
+    async def catch_up() -> bool:
+        if not await rescope():
+            return False
         await send_snapshot()
         return True
 
-    told = await send_snapshot()
-
     async with event_bus.subscribe() as queue:
+        # Listen before reading state: a value changing while the snapshot is
+        # sent must follow it on the socket, even if it never changes again.
+        told = await send_snapshot()
         receiver = asyncio.create_task(_receive(websocket))
         # Where a live view this socket is one end of reaches it. Its own
         # outbox rather than the bus's queue, which drops when full.
@@ -1741,6 +1747,7 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
                 await websocket.send_text(demand_frame(ports))
             live_view.sessions.attach(device, endpoint)
         outbox = asyncio.create_task(endpoint.next())
+        catching_up = False
         try:
             while True:
                 sender = asyncio.create_task(queue.get())
@@ -1845,10 +1852,13 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
                     except asyncio.QueueEmpty:
                         break
 
-                # A panel this far behind lost events the bus dropped for it:
-                # its demand, a dashboard published or its own unpairing. It
-                # catches up once this batch is out, as after a dashboard.
-                resync = only is not None and event_bus.dropped(queue)
+                # Every client catches up after overflow. A lost event may be
+                # a value that never changes again, or a device's unpairing.
+                lost = event_bus.dropped(queue)
+                catching_up = catching_up or lost
+                if lost and not await rescope():
+                    return
+                resync = False
                 if only is None:
                     fresh = hidden_from(user, websocket.app)
                     if fresh != hidden:
@@ -1883,8 +1893,12 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
                     for frame in media_frames(out, wanted, store):
                         await websocket.send_bytes(frame)
                     await _send(websocket, out)
-                if resync and not await catch_up():
-                    return
+                # Send retained events before the recovery snapshot: values
+                # left in the queue would otherwise overwrite its newer state.
+                if resync or (catching_up and queue.empty()):
+                    if not await catch_up():
+                        return
+                    catching_up = catching_up and not queue.empty()
         except (WebSocketDisconnect, RuntimeError):
             # A peer that goes away mid-send takes the RuntimeError route
             # ("websocket.send after websocket.close") rather than the clean
diff --git a/backend/fluksio/api/routes/messages.py b/backend/fluksio/api/routes/messages.py
index 4a2e4e5..79d52a6 100644
--- a/backend/fluksio/api/routes/messages.py
+++ b/backend/fluksio/api/routes/messages.py
@@ -19,7 +19,6 @@ from fluksio.api.deps import (
     get_current_user,
 )
 from fluksio.flow.devices import DeviceDef, DeviceStore
-from fluksio.flow.messages import flow_of
 from fluksio.flow.pipeline import ValueSource
 from fluksio.flow.state import as_number
 
@@ -118,7 +117,7 @@ def _port_source(request: Request, name: str) -> ValueSource | None:
 
 
 @router.post("/{name}", response_model=MessageValue)
-async def publish_message(
+def publish_message(
     name: str, body: PublishRequest, controller: FlowControllerDep, request: Request
 ) -> Any:
     """Put a value into the graph, as a dashboard control does.
@@ -133,7 +132,7 @@ async def publish_message(
         detail=body.source_detail,
     )
     try:
-        await run_in_threadpool(controller.publish_message, name, body.value, source)
+        controller.publish_message(name, body.value, source)
     except KeyError:
         raise HTTPException(
             status_code=404, detail=f"No flow declares a message named '{name}'"
@@ -141,8 +140,7 @@ async def publish_message(
     except TypeError as exc:
         raise HTTPException(status_code=422, detail=str(exc))
 
-    values = controller.values(flow_of(name))
-    current = values.get(name, {})
+    current = controller.value(name)
     return MessageValue(name=name, value=current.get("value"), ts=current.get("ts"))
 
 
diff --git a/backend/fluksio/cloud/connector.py b/backend/fluksio/cloud/connector.py
index cfa3cf3..a243b2e 100644
--- a/backend/fluksio/cloud/connector.py
+++ b/backend/fluksio/cloud/connector.py
@@ -722,16 +722,6 @@ class CloudConnector:
         # The cameras a person may not see, looked at again per event.
         hidden = hidden_from(user, self._app, device=only is not None)
         controller = getattr(self._app.state, "flow_controller", None)
-        told: list[str] = []
-        if controller is not None:
-            # A device here sends its frames over HTTP: the tunnel carries
-            # text only.
-            snapshot = snapshot_payload(controller, only, device, "http", hidden)
-            told = list(snapshot.get("demand", []))
-            await socket.send(
-                _dump({"op": "ws_msg", "id": stream_id, "text": _dump(snapshot)})
-            )
-
         # The media this client's tiles are drawing count as demand on the
         # devices sending them, as a direct socket's do. The bytes themselves
         # are fetched one by one through the portal.
@@ -743,6 +733,20 @@ class CloudConnector:
         outbox: asyncio.Task[str] | None = None
         sender: asyncio.Task[dict[str, Any]] | None = None
 
+        async def rescope() -> bool:
+            nonlocal only, wanted
+            if only is not None:
+                if device and revoked(token, self._app):
+                    await socket.send(
+                        _dump({"op": "ws_close", "id": stream_id, "code": 1008})
+                    )
+                    return False
+                # A deleted panel scopes to nothing, never to every message.
+                only = panel_scope(token, self._app) or set()
+                wanted &= only
+                demand.want(viewer, wanted)
+            return True
+
         def heard(text: str) -> None:
             nonlocal wanted
             asked = wanted_names(text, only, hidden)
@@ -768,6 +772,19 @@ class CloudConnector:
             live_view.sessions.attach(device, endpoint)
         try:
             async with event_bus.subscribe() as queue:
+                # Subscribe before the snapshot so changes during its send
+                # cannot fall between the snapshot and the live stream.
+                told: list[str] = []
+                if controller is not None:
+                    snapshot = snapshot_payload(
+                        controller, only, device, "http", hidden
+                    )
+                    told = list(snapshot.get("demand", []))
+                    await socket.send(
+                        _dump(
+                            {"op": "ws_msg", "id": stream_id, "text": _dump(snapshot)}
+                        )
+                    )
                 outbox = asyncio.create_task(endpoint.next())
                 if device and (ports := demand.ports(device)) != told:
                     await socket.send(
@@ -779,6 +796,7 @@ class CloudConnector:
                             }
                         )
                     )
+                catching_up = False
                 while True:
                     sender = asyncio.create_task(queue.get())
                     done, _ = await asyncio.wait(
@@ -796,11 +814,13 @@ class CloudConnector:
                         if sender not in done:
                             continue
                     event = sender.result()
-                    # Behind far enough to lose events, a device catches up
-                    # as after a dashboard: the lost one may have been its
-                    # demand or its unpairing. A person whose cameras opened
-                    # or closed is sent what may be seen now.
-                    resync = only is not None and event_bus.dropped(queue)
+                    # A browser needs the same recovery as a device after
+                    # overflow: the lost value may never be published again.
+                    lost = event_bus.dropped(queue)
+                    catching_up = catching_up or lost
+                    if lost and not await rescope():
+                        return
+                    resync = False
                     if only is None:
                         fresh = hidden_from(user, self._app)
                         if fresh != hidden:
@@ -824,24 +844,10 @@ class CloudConnector:
                                 {"op": "ws_msg", "id": stream_id, "text": _dump(event)}
                             )
                         )
-                    if not resync:
+                    if not resync and not (catching_up and queue.empty()):
                         continue
-                    if only is not None:
-                        if device and revoked(token, self._app):
-                            # Unpaired or deleted: 1008 sends it to check.
-                            await socket.send(
-                                _dump({"op": "ws_close", "id": stream_id, "code": 1008})
-                            )
-                            return
-                        # What this screen shows may have just changed under
-                        # it. Rescope, then resend the snapshot so a dashboard
-                        # it has only now been assigned draws values instead of
-                        # blanks. ``or set()`` is the point: a panel that was
-                        # deleted scopes to nothing, and None would widen this
-                        # socket to everything on the bus.
-                        only = panel_scope(token, self._app) or set()
-                        wanted &= only
-                        demand.want(viewer, wanted)
+                    if not await rescope():
+                        return
                     if controller is not None:
                         catch_up = snapshot_payload(
                             controller, only, device, "http", hidden
@@ -855,6 +861,10 @@ class CloudConnector:
                                 }
                             )
                         )
+                    # The snapshot follows the retained backlog. If sending
+                    # it queued more events, finish those before recovering
+                    # again so an older event cannot leave the client stale.
+                    catching_up = catching_up and not queue.empty()
                     if kind == "dashboard_changed" and only is not None:
                         # After the snapshot, as the device socket sends it.
                         await socket.send(
diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py
index 1d16dbb..8aaf95c 100644
--- a/backend/fluksio/flow/controller.py
+++ b/backend/fluksio/flow/controller.py
@@ -1545,6 +1545,9 @@ class FlowController:
             return _short_error(exc)
         return None
 
+    def value(self, name: str) -> dict[str, Any]:
+        return self.pipeline.value(name) if self.pipeline else {}
+
     def values(self, flow: str | None = None) -> dict[str, dict[str, Any]]:
         return self.pipeline.values(flow) if self.pipeline else {}
 
diff --git a/backend/fluksio/flow/events.py b/backend/fluksio/flow/events.py
index 8f00ba3..314caec 100644
--- a/backend/fluksio/flow/events.py
+++ b/backend/fluksio/flow/events.py
@@ -1,9 +1,10 @@
 """Event bus bridging the engine's worker threads to async subscribers.
 
 Nodes execute in a thread pool; websocket clients live on the event loop.
-Publishers are therefore thread-safe and never block: a subscriber that cannot
-keep up loses its oldest queued events rather than stalling the engine, and
-can ask whether it did (``dropped``) to catch up another way.
+Publishers are therefore thread-safe and never wait for subscribers. Both
+staging and subscriber queues are bounded, retaining errors and control events
+ahead of routine traffic. Subscribers can ask whether they lost events
+(``dropped``) to catch up another way.
 """
 
 from __future__ import annotations
@@ -19,24 +20,88 @@ from typing import Any
 logger = logging.getLogger(__name__)
 
 QUEUE_SIZE = 256
+DISPATCH_BUDGET = 1024
 LOG_HISTORY = 400
 #: How many of a message's last artifact digests are remembered.
 RECENT_DIGESTS = 8
 
 
+def _critical(event: dict[str, Any]) -> bool:
+    """Keep failures, lifecycle changes and unknown events ahead of telemetry."""
+    kind = event.get("type")
+    if kind == "node_log":
+        return event.get("level") == "error"
+    return kind not in {
+        "message_value",
+        "node_started",
+        "node_executed",
+        "work_latency",
+        "run_metric",
+    }
+
+
+class _EventBuffer:
+    """Two FIFO lanes allow constant-time eviction without reordering events."""
+
+    def __init__(self) -> None:
+        self._ordinary: deque[tuple[int, dict[str, Any]]] = deque()
+        self._critical: deque[tuple[int, dict[str, Any]]] = deque()
+        self._sequence = 0
+
+    def __len__(self) -> int:
+        return len(self._ordinary) + len(self._critical)
+
+    def append(self, event: dict[str, Any]) -> None:
+        lane = self._critical if _critical(event) else self._ordinary
+        lane.append((self._sequence, event))
+        self._sequence += 1
+
+    def popleft(self) -> dict[str, Any]:
+        lane = self._critical
+        if self._ordinary and (not lane or self._ordinary[0][0] < lane[0][0]):
+            lane = self._ordinary
+        return lane.popleft()[1]
+
+    def make_room(self, event: dict[str, Any]) -> bool:
+        """Evict the oldest ordinary event, or keep a full critical backlog."""
+        if self._ordinary:
+            self._ordinary.popleft()
+        elif _critical(event):
+            self._critical.popleft()
+        else:
+            return False
+        return True
+
+
+class _EventQueue(asyncio.Queue[dict[str, Any]]):
+    def _init(self, maxsize: int) -> None:
+        self._queue = _EventBuffer()
+
+    def offer(self, event: dict[str, Any]) -> bool:
+        """Enqueue without waiting and report any loss, including rejection."""
+        lost = self.full()
+        if not lost or self._queue.make_room(event):
+            self.put_nowait(event)
+        return lost
+
+
 class EventBus:
     """Fan-out of engine events to any number of async subscribers."""
 
     def __init__(self) -> None:
         self._loop: asyncio.AbstractEventLoop | None = None
-        self._subscribers: set[asyncio.Queue[dict[str, Any]]] = set()
+        self._subscribers: set[_EventQueue] = set()
         # Events published since the last flush was scheduled. A cascade
         # publishes 13-16 of them and each used to cost its own
         # `call_soon_threadsafe`; they now cost one between two turns of the
         # loop, which is what a wave is.
-        self._pending: list[dict[str, Any]] = []
+        self._pending = _EventBuffer()
+        self._pending_dropped = False
         self._pending_lock = threading.Lock()
         self._flush_scheduled = False
+        self._batch: list[dict[str, Any]] = []
+        self._batch_dropped = False
+        self._targets: deque[_EventQueue] = deque()
         # Kept whether or not anyone is listening, so opening the log panel
         # shows what just happened rather than an empty box.
         self.recent_logs: deque[dict[str, Any]] = deque(maxlen=LOG_HISTORY)
@@ -82,7 +147,12 @@ class EventBus:
         if loop is None or not self._subscribers:
             return
         with self._pending_lock:
-            self._pending.append(event)
+            room = True
+            if len(self._pending) == QUEUE_SIZE:
+                self._pending_dropped = True
+                room = self._pending.make_room(event)
+            if room:
+                self._pending.append(event)
             if self._flush_scheduled:
                 # A flush is already on its way and has not run yet, so it
                 # will find this event too.
@@ -93,33 +163,49 @@ class EventBus:
         except RuntimeError:
             # Loop already closed — shutting down.
             with self._pending_lock:
-                self._pending.clear()
+                self._pending = _EventBuffer()
+                self._pending_dropped = False
                 self._flush_scheduled = False
 
     def _dispatch(self) -> None:
-        """Hand everything published since the last turn to every subscriber."""
-        with self._pending_lock:
-            batch = self._pending
-            self._pending = []
-            self._flush_scheduled = False
-        if not batch:
-            return
-        for queue in self._subscribers:
-            for event in batch:
-                if queue.full():
-                    # Drop the oldest so a slow client never blocks the engine.
-                    try:
-                        queue.get_nowait()
-                    except asyncio.QueueEmpty:
-                        pass
+        """Deliver bounded work, yielding between batches or subscriber groups."""
+        if not self._targets:
+            with self._pending_lock:
+                self._batch = [
+                    self._pending.popleft() for _ in range(len(self._pending))
+                ]
+                self._batch_dropped = self._pending_dropped
+                self._pending_dropped = False
+            self._targets = deque(self._subscribers)
+        if self._batch:
+            # A subscriber receives the whole bounded batch before it resumes,
+            # so its catch-up snapshot cannot precede the rest of that batch.
+            count = max(1, DISPATCH_BUDGET // len(self._batch))
+            for _ in range(min(count, len(self._targets))):
+                queue = self._targets.popleft()
+                if queue not in self._subscribers:
+                    continue
+                if self._batch_dropped:
                     self._dropped.add(queue)
-                queue.put_nowait(event)
+                for event in self._batch:
+                    if queue.offer(event):
+                        self._dropped.add(queue)
+        else:
+            self._targets.clear()
+        if not self._targets:
+            self._batch = []
+            with self._pending_lock:
+                if not self._pending:
+                    self._flush_scheduled = False
+                    return
+        if self._loop is not None:
+            self._loop.call_soon(self._dispatch)
 
     def dropped(self, queue: asyncio.Queue[dict[str, Any]]) -> bool:
         """Whether this subscriber lost events since it last asked.
 
-        A device's socket that did catches up with a snapshot: what it lost
-        may have been its demand, or the event that tells it it was unpaired.
+        This includes staging overflow before fan-out. A socket catches up
+        with a snapshot, since a lost value might never change again.
         """
         if queue in self._dropped:
             self._dropped.discard(queue)
@@ -129,7 +215,7 @@ class EventBus:
     @asynccontextmanager
     async def subscribe(self) -> AsyncIterator[asyncio.Queue[dict[str, Any]]]:
         """Yield a queue receiving every event published while subscribed."""
-        queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=QUEUE_SIZE)
+        queue = _EventQueue(maxsize=QUEUE_SIZE)
         self._subscribers.add(queue)
         try:
             yield queue
diff --git a/backend/fluksio/flow/executor.py b/backend/fluksio/flow/executor.py
index 3f62990..992fee5 100644
--- a/backend/fluksio/flow/executor.py
+++ b/backend/fluksio/flow/executor.py
@@ -17,8 +17,10 @@ from __future__ import annotations
 import logging
 import threading
 import time
-from concurrent.futures import ThreadPoolExecutor
-from typing import TYPE_CHECKING, Any
+from collections import deque
+from collections.abc import Callable
+from concurrent.futures import Executor, Future, ThreadPoolExecutor
+from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar
 
 from fluksio.flow.queue import MAX_DELIVERIES, WorkItem, WorkQueue
 
@@ -41,9 +43,9 @@ 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
+# Extra cascade and node workers reserved for promoted timers. 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.
+# hold either pool is 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.
@@ -67,6 +69,73 @@ BACKLOG_INTERVAL_S = 5.0
 BACKLOG_DEGRADED = 50
 BACKLOG_STRIKES = 3
 
+_P = ParamSpec("_P")
+_T = TypeVar("_T")
+
+
+class _PoolLane(Executor):
+    def __init__(self, pool: _ReservedPool, due: bool) -> None:
+        self._pool = pool
+        self._due = due
+
+    def submit(
+        self, fn: Callable[_P, _T], /, *args: _P.args, **kwargs: _P.kwargs
+    ) -> Future[_T]:
+        return self._pool.submit(self._due, fn, *args, **kwargs)
+
+
+class _ReservedPool:
+    """Due work uses idle normal workers, or its own bounded reserve."""
+
+    def __init__(
+        self, workers: int, prefix: str, on_done: Callable[[], None] | None = None
+    ) -> None:
+        self._limits = (workers, DUE_RESERVE)
+        self._pools = (
+            ThreadPoolExecutor(max_workers=workers, thread_name_prefix=prefix),
+            ThreadPoolExecutor(
+                max_workers=DUE_RESERVE, thread_name_prefix=f"{prefix}-due"
+            ),
+        )
+        self._pending = [0, 0]
+        self._lock = threading.Lock()
+        self._on_done = on_done
+        self.normal = _PoolLane(self, False)
+        self.due = _PoolLane(self, True)
+
+    def available(self) -> tuple[int, int]:
+        with self._lock:
+            return (
+                max(0, self._limits[0] - self._pending[0]),
+                max(0, self._limits[1] - self._pending[1]),
+            )
+
+    def _done(self, lane: int) -> None:
+        with self._lock:
+            self._pending[lane] -= 1
+        if self._on_done is not None:
+            self._on_done()
+
+    def submit(
+        self, due: bool, fn: Callable[_P, _T], /, *args: _P.args, **kwargs: _P.kwargs
+    ) -> Future[_T]:
+        with self._lock:
+            # Count queued work too: an idle thread with ordinary work already
+            # waiting is not capacity a timer can use without waiting behind it.
+            lane = int(due and self._pending[0] >= self._limits[0])
+            self._pending[lane] += 1
+        try:
+            future = self._pools[lane].submit(fn, *args, **kwargs)
+        except RuntimeError:
+            self._done(lane)
+            raise
+        future.add_done_callback(lambda _: self._done(lane))
+        return future
+
+    def shutdown(self) -> None:
+        for pool in self._pools:
+            pool.shutdown(wait=False)
+
 
 class ExecutionService:
     """Owns the engine's worker threads and the queue they read from."""
@@ -93,12 +162,15 @@ class ExecutionService:
         self._inflight_lock = threading.Condition()
         # Entry ids claimed and still running, under _inflight_lock.
         self._active: set[str] = set()
-        self.node_pool = ThreadPoolExecutor(
-            max_workers=MAX_WORKERS if max_workers is None else max_workers,
-            thread_name_prefix="node",
+        # At most one reclaim batch waits here. The timer thread must keep
+        # promoting and touching work while the consumer waits for capacity.
+        self._reclaimed: deque[WorkItem] = deque()
+        self._node_pools = _ReservedPool(
+            MAX_WORKERS if max_workers is None else max_workers, "node"
         )
-        self._cascade_pool = ThreadPoolExecutor(
-            max_workers=self.max_cascades, thread_name_prefix="cascade"
+        self.node_pool = self._node_pools.normal
+        self._cascade_pools = _ReservedPool(
+            self.max_cascades, "cascade", self._capacity_changed
         )
         self._consumer: threading.Thread | None = None
         self._timers: threading.Thread | None = None
@@ -132,14 +204,18 @@ class ExecutionService:
                 thread.join(timeout=5)
         self._consumer = None
         self._timers = None
-        self._cascade_pool.shutdown(wait=False)
-        self.node_pool.shutdown(wait=False)
+        self._cascade_pools.shutdown()
+        self._node_pools.shutdown()
         self.queue.close()
 
     def bind(self, pipeline: Pipeline) -> None:
         """Point the service at the pipeline it should execute against."""
         self._pipeline = pipeline
 
+    def _capacity_changed(self) -> None:
+        with self._inflight_lock:
+            self._inflight_lock.notify_all()
+
     def pause_intake(self) -> None:
         """Stop claiming, and wait for what is already claimed to finish.
 
@@ -182,6 +258,18 @@ class ExecutionService:
             free, due_only = self._await_capacity()
             if not free:
                 continue
+            with self._inflight_lock:
+                recovered = [
+                    item
+                    for item in self._reclaimed
+                    if not due_only or item.not_before > 0
+                ][:free]
+                for item in recovered:
+                    self._reclaimed.remove(item)
+            if recovered:
+                for item in recovered:
+                    self._dispatch(item)
+                continue
             try:
                 items = self.queue.claim(
                     free,
@@ -262,21 +350,23 @@ class ExecutionService:
 
             if now - last_reclaim < RECLAIM_INTERVAL_S:
                 continue
+            with self._inflight_lock:
+                if self._reclaimed or not self._intake.is_set():
+                    continue
             last_reclaim = now
             try:
-                for item in self.queue.reclaim_stale(RECLAIM_IDLE_MS):
+                recovered = self.queue.reclaim_stale(RECLAIM_IDLE_MS)
+                for item in recovered:
                     logger.info(
                         "Reclaimed work for '%s' (delivery %d)",
                         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)
+                with self._inflight_lock:
+                    self._reclaimed.extend(recovered)
+                    self._active.update(
+                        item.entry_id for item in recovered if item.entry_id
+                    )
             except Exception as exc:
                 logger.error("Could not reclaim stale work: %s", exc)
 
@@ -331,7 +421,9 @@ class ExecutionService:
         """
         with self._inflight_lock:
             while not self._stop.is_set():
-                free = self.max_cascades - self._inflight
+                if not self._intake.is_set():
+                    return 0, False
+                free, _ = self._cascade_pools.available()
                 if free > 0:
                     return free, False
                 # Only once a slot has genuinely failed to free: the due lane
@@ -340,7 +432,7 @@ class ExecutionService:
                 # read every time the pool filled up.
                 if self._inflight_lock.wait(0.5):
                     continue
-                reserve = self.max_cascades + DUE_RESERVE - self._inflight
+                _, reserve = self._cascade_pools.available()
                 if reserve > 0:
                     return reserve, True
             return 0, False
@@ -351,7 +443,12 @@ class ExecutionService:
             if item.entry_id:
                 self._active.add(item.entry_id)
         try:
-            self._cascade_pool.submit(self._handle, item)
+            pool = (
+                self._cascade_pools.due
+                if item.not_before > 0
+                else self._cascade_pools.normal
+            )
+            pool.submit(self._handle, item)
         except RuntimeError:
             # Pool already shutting down.
             self._done(item)
@@ -386,7 +483,7 @@ class ExecutionService:
         return item.node
 
     def _handle(self, item: WorkItem) -> None:
-        handled = True
+        handled = False
         try:
             handled = self._run_item(item)
         except Exception:
@@ -473,11 +570,12 @@ class ExecutionService:
 
         replay = item.deliveries > 1
         emission = item.kind == "emission"
+        node_pool = self._node_pools.due if item.not_before > 0 else self.node_pool
         try:
             if item.kind == "flush":
                 # A rate-limit window ended; nothing to replay, only to let out.
                 # A run of its own, so what it sets off is recorded somewhere.
-                pipeline.flush(node, item.entry_id)
+                pipeline.flush(node, item.entry_id, node_pool=node_pool)
                 return True
             # An emission's values went into state when the node produced them;
             # this item carries them so its readers get the chunk that caused
@@ -498,6 +596,7 @@ class ExecutionService:
                 # Neither can say what changed, so neither filters on it.
                 changed=None if replay or not item.outputs else published,
                 overrides=item.outputs if emission else None,
+                node_pool=node_pool,
             )
         finally:
             # Paired, or a cascade that raised — state backend gone, say — is a
diff --git a/backend/fluksio/flow/metrics.py b/backend/fluksio/flow/metrics.py
index b2f7d3f..88b8c61 100644
--- a/backend/fluksio/flow/metrics.py
+++ b/backend/fluksio/flow/metrics.py
@@ -394,13 +394,8 @@ class MetricsCollector:
         prune: bool,
     ) -> None:
         with Session(engine) as session:
-            for (flow, node, minute), agg in buckets.items():
-                statement = insert(MetricBucket).values(
-                    flow=flow,
-                    node=node,
-                    bucket=datetime.fromtimestamp(minute, UTC),
-                    **agg,
-                )
+            if buckets:
+                statement = insert(MetricBucket)
                 # The same minute is written several times, so the counters add
                 # and the maxima take whichever is larger. Columns are read by
                 # subscript: `excluded.items` is the collection's own method.
@@ -418,21 +413,36 @@ class MetricsCollector:
                             name: func.max(col(getattr(MetricBucket, name)), new[name])
                             for name in MAXIMA
                         },
-                    )
+                    ),
+                    [
+                        {
+                            "flow": flow,
+                            "node": node,
+                            "bucket": datetime.fromtimestamp(minute, UTC),
+                            **agg,
+                        }
+                        for (flow, node, minute), agg in buckets.items()
+                    ],
                 )
 
-            for run in runs:
-                values = {k: v for k, v in run.items() if k != "started_ts"}
-                statement = insert(FlowRun).values(**values)
+            if runs:
+                values = [
+                    {k: v for k, v in run.items() if k != "started_ts"} for run in runs
+                ]
+                statement = insert(FlowRun)
                 session.execute(
                     statement.on_conflict_do_update(
                         index_elements=["id"],
                         set_={
                             key: statement.excluded[key]
-                            for key in values
+                            for key in values[0]
                             if key != "id"
                         },
-                    )
+                    ),
+                    values,
+                    # A redelivery clears finished_at. Include nulls in the
+                    # batch so open and finished runs share the same statement.
+                    execution_options={"render_nulls": True},
                 )
 
             session.add_all(pending)
diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py
index 0eaf098..ab20a36 100644
--- a/backend/fluksio/flow/pipeline.py
+++ b/backend/fluksio/flow/pipeline.py
@@ -21,10 +21,11 @@ import time
 import uuid
 from collections import deque
 from collections.abc import Callable, Iterator
-from concurrent.futures import Future, ThreadPoolExecutor, wait
+from concurrent.futures import Executor, Future, ThreadPoolExecutor
 from contextlib import contextmanager
 from datetime import UTC, datetime
 from functools import lru_cache
+from queue import SimpleQueue
 from typing import Any, Literal, Protocol
 
 from pydantic import BaseModel, model_validator
@@ -275,6 +276,7 @@ class Pipeline:
         "_paused",
         "_stepping",
         "_gate_lock",
+        "_gate_revision",
         "_graph_lock",
         "_queue",
         "_node_pool",
@@ -294,7 +296,7 @@ class Pipeline:
         initial_values: dict[str, Any] | None = None,
         disabled_flows: set[str] | None = None,
         work_queue: WorkQueue | None = None,
-        node_pool: ThreadPoolExecutor | None = None,
+        node_pool: Executor | None = None,
         observer: Callable[[NodeOutcome], None] | None = None,
         emission_observer: Callable[[str, dict[str, Any]], None] | None = None,
         run_cache: RunCacheLookup | None = None,
@@ -307,6 +309,7 @@ class Pipeline:
         self._paused: set[str] = set()
         self._stepping: set[str] = set()
         self._gate_lock = threading.Lock()
+        self._gate_revision = 0
         # Held only while the graph maps are read or swapped together, never
         # across an await, a node call or any I/O. Reentrant because the lazy
         # `edges` build re-enters through `_graph`. It is never nested with
@@ -448,6 +451,7 @@ class Pipeline:
         with self._gate_lock:
             self._paused.discard(flow)
             self._stepping.discard(flow)
+            self._gate_revision += 1
 
     def remove_flow(self, flow: str) -> None:
         """Take a deleted flow out of the graph."""
@@ -742,7 +746,9 @@ class Pipeline:
         self._state[self._flush_key(node.id)] = at
         self.defer(node, {}, at - now, kind="flush", parent=entry_id)
 
-    def flush(self, node: Node, entry_id: str = "") -> None:
+    def flush(
+        self, node: Node, entry_id: str = "", *, node_pool: Executor | None = None
+    ) -> None:
         """Let through what this node's rate limits held back.
 
         Runs when a window ends: the last value a limited output kept back is
@@ -766,7 +772,9 @@ class Pipeline:
             # nothing came out and there is nothing downstream to wake.
             published = self.apply_outputs(node, held, entry_id)
             if published:
-                self.run_downstream(node, entry_id=entry_id, changed=published)
+                self.run_downstream(
+                    node, entry_id=entry_id, changed=published, node_pool=node_pool
+                )
         if any(spec.interval > 0 for spec in node.requires.values()):
             # A wake-up that was held back, not a value that just changed —
             # so this one deliberately walks everything reachable.
@@ -775,6 +783,7 @@ class Pipeline:
                 self._state,
                 check_ready=True,
                 entry_id=entry_id,
+                node_pool=node_pool,
             )
 
     def _increment_message_versions(self, outputs: dict[str, Any]) -> None:
@@ -1252,7 +1261,9 @@ class Pipeline:
 
     def set_disabled(self, flows: set[str]) -> None:
         """Which flows are stopped. Swapped whole, so a reader never sees half."""
-        self._disabled = frozenset(flows)
+        with self._gate_lock:
+            self._disabled = frozenset(flows)
+            self._gate_revision += 1
 
     def paused_flows(self) -> list[str]:
         with self._gate_lock:
@@ -1262,11 +1273,13 @@ class Pipeline:
         """Hold this flow's nodes. Values still arrive; nothing acts on them."""
         with self._gate_lock:
             self._paused.add(flow)
+            self._gate_revision += 1
         self._publish({"type": "flow_paused", "flow": flow, "paused": True})
 
     def resume(self, flow: str) -> None:
         with self._gate_lock:
             self._paused.discard(flow)
+            self._gate_revision += 1
         self._publish({"type": "flow_paused", "flow": flow, "paused": False})
         # Whatever was held back is free to run now.
         self._execute_parallel(self.flow_nodes(flow), self._state, check_ready=True)
@@ -1280,11 +1293,13 @@ class Pipeline:
         """
         with self._gate_lock:
             self._stepping.add(flow)
+            self._gate_revision += 1
         try:
             yield
         finally:
             with self._gate_lock:
                 self._stepping.discard(flow)
+                self._gate_revision += 1
 
     def _gate_blocks(self, node: Node) -> bool:
         """Is this node's flow held back from executing?"""
@@ -1311,6 +1326,7 @@ class Pipeline:
         replay: bool = False,
         changed: set[str] | None = None,
         overrides: dict[str, Any] | None = None,
+        node_pool: Executor | None = None,
     ) -> StateBackend:
         """Execute nodes concurrently, scheduling each as its inputs arrive.
 
@@ -1344,9 +1360,11 @@ class Pipeline:
         # Copied: the caller's set is not this wave's to grow.
         fresh = set(changed) if changed is not None else None
 
-        submitted: set[Node] = set()
-        skipped: set[Node] = set()
-        node_futures: dict[Node, Future[dict[str, Any] | None]] = {}
+        ready_nodes = deque(n for n, count in in_degree.items() if count == 0)
+        blocked: set[Node] = set()
+        gate_revision = self._gate_revision
+        node_futures: dict[Future[dict[str, Any] | None], Node] = {}
+        finished: SimpleQueue[Future[dict[str, Any] | None]] = SimpleQueue()
 
         def untouched(n: Node) -> bool:
             """Did this wave publish nothing this node waits on?"""
@@ -1356,108 +1374,72 @@ class Pipeline:
             )
 
         def complete(n: Node) -> None:
-            """Count a node as done without running it, freeing its consumers."""
-            skipped.add(n)
+            """Release consumers when their last dependency finishes or skips."""
             for consumer in edges[n]:
                 if consumer in target_nodes:
                     in_degree[consumer] -= 1
+                    if in_degree[consumer] == 0:
+                        ready_nodes.append(consumer)
 
-        def submit_ready(executor: ThreadPoolExecutor) -> None:
-            # To a fixpoint: completing a node without running it frees its
-            # consumers, and one this pass has already walked past would
-            # otherwise be stranded — there is no future outstanding to bring
-            # the drain loop back round for it.
-            progressed = True
-            while progressed:
-                progressed = False
-                for n in target_nodes:
-                    if n in submitted or n in skipped:
-                        continue
-                    # Gate before the readiness check, so a held-back node does
-                    # not spend the synchronous claim it needs once it may run.
-                    if self._gate_blocks(n) or in_degree[n] != 0:
-                        continue
-                    if n.passive:
-                        # Something outside publishes for it; a run is not it.
+        def submit_ready(executor: Executor) -> None:
+            nonlocal gate_revision
+            with self._gate_lock:
+                revision = self._gate_revision
+            if blocked and revision != gate_revision:
+                # A flow can resume while another branch of this wave runs.
+                # Revisit only its held candidates, and only if a gate moved.
+                ready_nodes.extend(blocked)
+                blocked.clear()
+            gate_revision = revision
+            while ready_nodes:
+                n = ready_nodes.popleft()
+                # Gate before readiness, so a pause spends no synchronous claim.
+                if self._gate_blocks(n):
+                    blocked.add(n)
+                    continue
+                if n.passive:
+                    complete(n)
+                    continue
+                if fresh is not None and untouched(n):
+                    complete(n)
+                    continue
+                ready_inputs: dict[str, Any] | None = None
+                if check_ready:
+                    ready, ready_inputs = self._is_node_ready(n, state, entry_id)
+                    if not ready:
+                        # Its old output remains usable by a consumer whose
+                        # other producer did publish during this wave.
                         complete(n)
-                        progressed = True
                         continue
-                    if fresh is not None and untouched(n):
-                        # Ahead of the readiness check, so an unchanged node
-                        # spends neither a delivery stamp nor a synchronous
-                        # claim on a run it is not going to make.
-                        complete(n)
-                        progressed = True
-                        continue
-                    ready_inputs: dict[str, Any] | None = None
-                    if check_ready:
-                        # The values are kept whether or not the answer is yes:
-                        # a node that may run reads them below instead of
-                        # asking state for the same keys again.
-                        ready, ready_inputs = self._is_node_ready(n, state, entry_id)
-                        if not ready:
-                            # Completed, not merely passed over: a later trigger
-                            # may still make it ready, but not running is not
-                            # the same as blocking. Whatever it last published
-                            # is in state, and a consumer reading that alongside
-                            # something this wave *did* publish has every right
-                            # to run on the pair.
-                            #
-                            # Passing over it held that consumer for the whole
-                            # wave, and a node that can never be ready — an
-                            # input nothing has ever published — took its
-                            # consumers out for good. On this house
-                            # `plugs.pump_run` waits on a watering pulse that
-                            # only exists at 02:00, and it sits upstream of
-                            # `dmx.switches`: every boiler, plug and appliance
-                            # command reached its own message and stopped there,
-                            # reaching the DMX universe only in the occasional
-                            # wave the pump node happened not to be part of.
-                            complete(n)
-                            progressed = True
-                            continue
-                    if replay and entry_id and self._already_done(entry_id, n):
-                        # Its side effect happened on an earlier delivery; its
-                        # outputs are still in state, so downstream carries on.
-                        complete(n)
-                        progressed = True
-                        continue
-                    submitted.add(n)
-                    node_futures[n] = executor.submit(
-                        self._execute_node, n, state, entry_id, overrides, ready_inputs
-                    )
+                if replay and entry_id and self._already_done(entry_id, n):
+                    complete(n)
+                    continue
+                future = executor.submit(
+                    self._execute_node, n, state, entry_id, overrides, ready_inputs
+                )
+                node_futures[future] = n
+                future.add_done_callback(finished.put)
 
-        def drain(executor: ThreadPoolExecutor) -> None:
+        def drain(executor: Executor) -> None:
             submit_ready(executor)
-
             while node_futures:
-                done, _ = wait(node_futures.values(), return_when="FIRST_COMPLETED")
-                completed = [n for n, f in node_futures.items() if f in done]
+                # A completion queue also avoids scanning every outstanding
+                # future whenever one node finishes.
+                future = finished.get()
+                n = node_futures.pop(future)
+                result = future.result()
+                if result is not None and fresh is not None:
+                    fresh.update(result)
+                # Silent, failed and throttled producers still release their
+                # consumers. Fresh inputs decide whether those consumers run.
+                complete(n)
+                submit_ready(executor)
 
-                for n in completed:
-                    result = node_futures.pop(n).result()
-                    # Before the decrement: a consumer freed by this node is
-                    # judged on what it just published.
-                    if result is not None and fresh is not None:
-                        fresh.update(result)
-                    # Decremented whatever it returned. A node that published
-                    # nothing — rate limited, unchanged, or failed — has not
-                    # *blocked* its consumers; it has merely given them nothing
-                    # to read, and `untouched` above is the accurate test of
-                    # that. Holding the count instead made a quiet producer
-                    # veto a noisy sibling: `dmx.switches` reads both boilers
-                    # through report-by-exception nodes, so the kitchen one
-                    # staying silent — which is almost always — kept the main
-                    # boiler's command from ever reaching the DMX universe.
-                    for consumer in edges[n]:
-                        if consumer in target_nodes:
-                            in_degree[consumer] -= 1
-                    submit_ready(executor)
-
-        if self._node_pool is not None:
+        pool = node_pool if node_pool is not None else self._node_pool
+        if pool is not None:
             # The execution service owns a long-lived pool; building one per
             # wave is what used to spawn threads without bound under load.
-            drain(self._node_pool)
+            drain(pool)
         else:
             with ThreadPoolExecutor(max_workers=self._max_workers) as executor:
                 drain(executor)
@@ -1519,6 +1501,7 @@ class Pipeline:
         replay: bool = False,
         changed: set[str] | None = None,
         overrides: dict[str, Any] | None = None,
+        node_pool: Executor | None = None,
     ) -> StateBackend:
         """Run everything downstream of a node that has just published.
 
@@ -1552,6 +1535,7 @@ class Pipeline:
             replay=replay,
             changed=changed,
             overrides=overrides,
+            node_pool=node_pool,
         )
 
     def trigger(
@@ -1796,6 +1780,12 @@ class Pipeline:
         # Losing the value outright would be worse than running it here.
         self._run_here(node, outputs, cause="external", delivered=kind == "emission")
 
+    def value(self, name: str) -> dict[str, Any]:
+        """One message and its timestamp, read together without listing state."""
+        stamp = self._timestamp_key(name)
+        found = self._state.get_multi([name, stamp])
+        return {"value": found.get(name), "ts": found.get(stamp)}
+
     def values(self, flow: str | None = None) -> dict[str, dict[str, Any]]:
         """Last value and timestamp of every message, optionally one flow's.
 
diff --git a/backend/fluksio/flow/queue.py b/backend/fluksio/flow/queue.py
index 3d46bad..1b3a605 100644
--- a/backend/fluksio/flow/queue.py
+++ b/backend/fluksio/flow/queue.py
@@ -53,6 +53,23 @@ STALE_CONSUMER_IDLE_MS = 3_600_000
 # count, since redelivery happens seconds after the claim and never later.
 DONE_MARKERS = 10_000
 
+# A promotion is one Redis operation, including when the caller loses its
+# connection before hearing the result. Write first: Lua does not roll back a
+# command error, so a refused XADD must leave its timer in the delayed set.
+_MOVE_DUE = """
+local due = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', ARGV[1], 'LIMIT', 0, 100)
+for _, raw in ipairs(due) do
+    local fields = {}
+    for key, value in pairs(cjson.decode(raw)) do
+        fields[#fields + 1] = key
+        fields[#fields + 1] = value
+    end
+    redis.call('XADD', KEYS[2], 'MAXLEN', '~', ARGV[2], '*', unpack(fields))
+    redis.call('ZREM', KEYS[1], raw)
+end
+return #due
+"""
+
 
 @dataclass
 class WorkItem:
@@ -446,6 +463,7 @@ class RedisWorkQueue(WorkQueue):
         # finding them with a keyspace scan walked every state and idempotency
         # key in the database, twice per health poll.
         self._parked_flows_key = f"{namespace}:__parked_flows__"
+        self._move_due = self._redis.register_script(_MOVE_DUE)
         self._ensure_group()
         self._drop_stale_consumers()
 
@@ -619,36 +637,14 @@ class RedisWorkQueue(WorkQueue):
         ]
 
     def move_due(self, now: float) -> int:
-        due = cast(
-            list[str],
-            self._redis.zrangebyscore(self._delayed_key, "-inf", now, start=0, num=100),
+        """Move up to 100 timers into the due lane atomically."""
+        return cast(
+            int,
+            self._move_due(
+                keys=[self._delayed_key, self._due_stream],
+                args=[now, STREAM_MAXLEN],
+            ),
         )
-        if not due:
-            return 0
-        # Whoever removes it owns it: a second engine gets 0 for that member and
-        # leaves it alone. One round trip for the batch rather than one each,
-        # which matters because this runs every second.
-        pipe = self._redis.pipeline()
-        for raw in due:
-            pipe.zrem(self._delayed_key, raw)
-        claimed = cast(list[int], pipe.execute())
-
-        pipe = self._redis.pipeline()
-        moved = 0
-        for raw, owned in zip(due, claimed, strict=True):
-            if owned:
-                # Into the due lane, which `claim` reads first: what has waited
-                # out a deadline should not then queue behind a backlog.
-                pipe.xadd(
-                    self._due_stream,
-                    cast(Any, orjson.loads(raw)),
-                    maxlen=STREAM_MAXLEN,
-                    approximate=True,
-                )
-                moved += 1
-        if moved:
-            pipe.execute()
-        return moved
 
     def dead_letter(self, item: WorkItem, reason: str) -> None:
         fields = item.to_fields()
diff --git a/backend/fluksio/spa.py b/backend/fluksio/spa.py
index 91af90b..a95f6af 100644
--- a/backend/fluksio/spa.py
+++ b/backend/fluksio/spa.py
@@ -15,7 +15,7 @@ import mimetypes
 import re
 from pathlib import Path
 
-from starlette.middleware.gzip import GZipMiddleware
+from starlette.datastructures import Headers
 from starlette.responses import FileResponse
 from starlette.types import ASGIApp, Receive, Scope, Send
 
@@ -26,6 +26,24 @@ API_PREFIXES = ("/api/", "/hooks/", "/mcp", "/.well-known/", "/docs", "/redoc")
 mimetypes.add_type("application/manifest+json", ".webmanifest")
 
 
+def _accepts_gzip(header: str) -> bool:
+    """An explicit refusal overrides a wildcard, including on a cached asset."""
+    qualities: dict[str, float] = {}
+    for entry in header.lower().split(","):
+        encoding, *parameters = entry.strip().split(";")
+        quality = 1.0
+        for parameter in parameters:
+            key, _, value = parameter.strip().partition("=")
+            if key == "q":
+                try:
+                    quality = float(value)
+                except ValueError:
+                    quality = 0.0
+        qualities[encoding.strip()] = quality if 0.0 <= quality <= 1.0 else 0.0
+    gzip = qualities.get("gzip", qualities.get("*", 0.0))
+    return gzip > 0 and gzip >= qualities.get("identity", 0.0)
+
+
 def content_security_policy(index_html: str) -> str:
     """The policy for the dashboard, admitting its inline scripts by hash.
 
@@ -84,5 +102,12 @@ class Dashboard:
             # assets, so a stale copy pins the browser to the previous build.
             file = self.index
             headers = {**self.headers, "Cache-Control": "no-cache"}
-        response = FileResponse(file, headers=headers)
-        await GZipMiddleware(response, minimum_size=1024)(scope, receive, send)
+        media_type = mimetypes.guess_type(file.name)[0] or "text/plain"
+        compressed = file.with_name(file.name + ".gz").resolve()
+        if compressed.is_relative_to(self.dist) and compressed.is_file():
+            headers["Vary"] = "Accept-Encoding"
+            if _accepts_gzip(Headers(scope=scope).get("accept-encoding", "")):
+                file = compressed
+                headers["Content-Encoding"] = "gzip"
+        response = FileResponse(file, headers=headers, media_type=media_type)
+        await response(scope, receive, send)
diff --git a/backend/tests/flow/test_event_bus.py b/backend/tests/flow/test_event_bus.py
index 9838b87..2a5e368 100644
--- a/backend/tests/flow/test_event_bus.py
+++ b/backend/tests/flow/test_event_bus.py
@@ -2,8 +2,11 @@
 
 import asyncio
 import threading
+from contextlib import AsyncExitStack
 
-from fluksio.flow.events import QUEUE_SIZE, EventBus
+import pytest
+
+from fluksio.flow.events import DISPATCH_BUDGET, QUEUE_SIZE, EventBus
 
 
 def _event(i: int) -> dict[str, object]:
@@ -100,3 +103,91 @@ def test_nothing_is_buffered_with_nobody_listening() -> None:
         return len(bus._pending)
 
     assert asyncio.run(run()) == 0
+
+
+@pytest.mark.parametrize("flush", [False, True])
+def test_overflow_keeps_errors_and_control_events_in_order(flush: bool) -> None:
+    """The staging bound and a slow subscriber use the same eviction policy."""
+
+    async def run() -> None:
+        bus = EventBus()
+        bus.bind(asyncio.get_running_loop())
+        critical = [
+            {"type": "node_log", "level": "error"},
+            {"type": "node_error"},
+            {"type": "run_started"},
+            {"type": "audit"},
+            {"type": "dashboard_changed"},
+            {"type": "future_control_event"},
+        ]
+        async with bus.subscribe() as first, bus.subscribe() as second:
+            sent = [*critical, *(_event(i) for i in range(QUEUE_SIZE + 50))]
+            sent.append({"type": "run_finished"})
+            for sequence, event in enumerate(sent):
+                event["sequence"] = sequence
+                bus.publish(event)
+                assert len(bus._pending) <= QUEUE_SIZE
+                if flush:
+                    # Queue overflow, with no staging overflow.
+                    await asyncio.sleep(0)
+            await asyncio.sleep(0)
+            for queue in (first, second):
+                received = [queue.get_nowait() for _ in range(queue.qsize())]
+                assert len(received) == QUEUE_SIZE
+                assert received[: len(critical)] == critical
+                assert received[-1]["type"] == "run_finished"
+                assert received == sorted(received, key=lambda e: e["sequence"])
+                assert bus.dropped(queue)
+                assert not bus.dropped(queue)
+
+    asyncio.run(run())
+
+
+def test_a_full_critical_backlog_rejects_telemetry_and_reports_loss() -> None:
+    async def run() -> None:
+        bus = EventBus()
+        bus.bind(asyncio.get_running_loop())
+        async with bus.subscribe() as queue:
+            for i in range(QUEUE_SIZE + 1):
+                bus.publish({"type": "node_error", "sequence": i})
+            bus.publish(_event(0))
+            assert len(bus._pending) == QUEUE_SIZE
+            await asyncio.sleep(0)
+            assert bus.dropped(queue)
+            assert (await queue.get())["sequence"] == 1
+            bus.publish({"type": "audit"})
+            await asyncio.sleep(0)
+            bus.publish(_event(1))
+            await asyncio.sleep(0)
+            assert bus.dropped(queue)
+            assert queue.qsize() == QUEUE_SIZE
+            assert all(
+                queue.get_nowait()["type"] != "node_executed" for _ in range(QUEUE_SIZE)
+            )
+
+    asyncio.run(run())
+
+
+def test_fanout_yields_before_all_subscribers_are_served() -> None:
+    async def run() -> None:
+        bus = EventBus()
+        loop = asyncio.get_running_loop()
+        bus.bind(loop)
+        async with AsyncExitStack() as stack:
+            queues = [
+                await stack.enter_async_context(bus.subscribe())
+                for _ in range(DISPATCH_BUDGET // QUEUE_SIZE + 1)
+            ]
+            for i in range(QUEUE_SIZE):
+                bus.publish(_event(i))
+            heartbeat: asyncio.Future[int] = loop.create_future()
+            loop.call_soon(
+                lambda: heartbeat.set_result(sum(queue.qsize() for queue in queues))
+            )
+            assert await heartbeat == DISPATCH_BUDGET
+            await asyncio.sleep(0)
+            assert all(queue.qsize() == QUEUE_SIZE for queue in queues)
+            assert not any(bus.dropped(queue) for queue in queues)
+            assert not bus._flush_scheduled
+
+    asyncio.run(run())
diff --git a/backend/tests/flow/test_queue.py b/backend/tests/flow/test_queue.py
index 6922a01..cc2a2f6 100644
--- a/backend/tests/flow/test_queue.py
+++ b/backend/tests/flow/test_queue.py
@@ -242,6 +242,48 @@ def test_a_trigger_is_journaled_rather_than_run_on_the_spot():
     assert state["f.doubled"] == 6.0
 
 
+def test_a_state_failure_leaves_the_delivery_unacknowledged(monkeypatch):
+    pipeline, source, state, seen = _pipeline_with_a_consumer()
+    service = ExecutionService(pipeline._queue)
+    service.bind(pipeline)
+
+    def unavailable(*args, **kwargs):
+        raise ConnectionError("State unavailable")
+
+    monkeypatch.setattr(state, "record", unavailable)
+    source.inject({"reading": 3.0})
+    item = pipeline._queue.claim(1, 10)[0]
+    try:
+        service._dispatch(item)
+        service.pause_intake()
+        assert seen == []
+        assert pipeline._queue.stats()["pending"] == 1
+        assert service.inflight == 0
+    finally:
+        service.stop()
+
+
+def test_a_node_failure_still_acknowledges_its_delivery(monkeypatch):
+    pipeline, source, _state, seen = _pipeline_with_a_consumer()
+    service = ExecutionService(pipeline._queue)
+    service.bind(pipeline)
+
+    def broken(*args, **kwargs):
+        raise ValueError("Node failed")
+
+    monkeypatch.setattr(pipeline.get_node_by_id("f.consumer"), "f", broken)
+    source.inject({"reading": 3.0})
+    item = pipeline._queue.claim(1, 10)[0]
+    try:
+        service._dispatch(item)
+        service.pause_intake()
+        assert seen == []
+        assert pipeline._queue.stats()["pending"] == 0
+        assert service.inflight == 0
+    finally:
+        service.stop()
+
+
 def test_work_for_a_paused_flow_is_held_and_released_on_resume():
     pipeline, source, state, seen = _pipeline_with_a_consumer()
     service = ExecutionService(pipeline._queue)
diff --git a/backend/tests/flow/test_redis_queue.py b/backend/tests/flow/test_redis_queue.py
new file mode 100644
index 0000000..a6fd184
--- /dev/null
+++ b/backend/tests/flow/test_redis_queue.py
@@ -0,0 +1,68 @@
+"""Redis delivery checks, enabled by FLUKSIO_TEST_REDIS_HOST (and optional PORT)."""
+
+import os
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+import redis
+
+from fluksio.flow.queue import RedisWorkQueue, WorkItem
+
+
+@pytest.fixture
+def queue():
+    host = os.environ.get("FLUKSIO_TEST_REDIS_HOST")
+    if not host:
+        pytest.skip("Set FLUKSIO_TEST_REDIS_HOST to run Redis integration checks")
+    queue = RedisWorkQueue(
+        host=host,
+        port=int(os.environ.get("FLUKSIO_TEST_REDIS_PORT", "6379")),
+        namespace=f"test-queue-{uuid.uuid4().hex}",
+    )
+    try:
+        yield queue
+    finally:
+        keys = list(queue._redis.scan_iter(match=f"{queue._ns}:*"))
+        if keys:
+            queue._redis.delete(*keys)
+        queue.close()
+
+
+def test_concurrent_promotions_deliver_each_due_item_once(queue):
+    for i in range(150):
+        queue.add_delayed(
+            WorkItem(kind="cascade", node=f"f.n{i}", flow="f", outputs={"f.x": i}),
+            100.0,
+        )
+    queue.add_delayed(WorkItem(kind="cascade", node="f.later", flow="f"), 300.0)
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        moved = list(pool.map(queue.move_due, [200.0, 200.0]))
+
+    assert sorted(moved) == [50, 100]
+    assert queue.move_due(200.0) == 0
+    assert queue.next_due() == 300.0
+    claimed = queue.claim(200, 10, due_only=True)
+    assert len(claimed) == 150
+    assert {item.outputs["f.x"] for item in claimed} == set(range(150))
+    assert all(item.not_before == 100.0 for item in claimed)
+    assert all(item.entry_id.startswith("due:") for item in claimed)
+
+
+def test_a_refused_stream_write_keeps_the_timer_for_retry(queue):
+    queue.add_delayed(WorkItem(kind="cascade", node="f.stop", flow="f"), 100.0)
+    # A Redis command error does not roll a Lua script back. The stream write
+    # must succeed before the delayed entry is removed.
+    queue._redis.delete(queue._due_stream)
+    queue._redis.set(queue._due_stream, "not a stream")
+
+    with pytest.raises(redis.ResponseError, match="WRONGTYPE"):
+        queue.move_due(200.0)
+
+    assert queue.next_due() == 100.0
+    queue._redis.delete(queue._due_stream)
+    queue._ensure_group()
+    assert queue.move_due(200.0) == 1
+    assert queue.move_due(200.0) == 0
+    assert [item.node for item in queue.claim(10, 10, due_only=True)] == ["f.stop"]
diff --git a/backend/tests/flow/test_round_trips.py b/backend/tests/flow/test_round_trips.py
index e299ce5..28bae87 100644
--- a/backend/tests/flow/test_round_trips.py
+++ b/backend/tests/flow/test_round_trips.py
@@ -122,3 +122,24 @@ def test_a_hop_costs_a_bounded_number_of_state_operations() -> None:
 
     total = sum(state.calls[name] for name in COUNTED)
     assert total <= 8, f"state operations per hop grew: {dict(state.calls)}"
+
+
+def test_reading_one_value_does_not_list_or_read_other_messages(monkeypatch) -> None:
+    state = MemoryState()
+    state.update({f"f.m{i}": i for i in range(1000)})
+    state.update({"f.m0": None, "__ts__:f.m0": 123.0})
+    pipeline = Pipeline([], state=state)
+    reads = []
+    get_multi = state.get_multi
+
+    def read(keys):
+        reads.append(keys)
+        return get_multi(keys)
+
+    def no_listing():
+        raise AssertionError("A single-message read must not enumerate state")
+
+    monkeypatch.setattr(MemoryState, "get_multi", lambda self, keys: read(keys))
+    monkeypatch.setattr(MemoryState, "message_names", lambda self: no_listing())
+    assert pipeline.value("f.m0") == {"value": None, "ts": 123.0}
+    assert reads == [["f.m0", "__ts__:f.m0"]]
diff --git a/backend/tests/flow/test_scheduling.py b/backend/tests/flow/test_scheduling.py
new file mode 100644
index 0000000..afd978b
--- /dev/null
+++ b/backend/tests/flow/test_scheduling.py
@@ -0,0 +1,84 @@
+"""Scheduling work scales with the graph and observes runtime gates."""
+
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import nullcontext
+
+import pytest
+
+from fluksio.flow.messages import DType, MessageSpec
+from fluksio.flow.nodes import Node
+from fluksio.flow.pipeline import Pipeline
+
+
+def _node(name, function, requires=(), provides=(), flow="f"):
+    node = Node(
+        f=function,
+        requires=[MessageSpec(name=key, dtype=DType.FLOAT) for key in requires],
+        provides=[MessageSpec(name=key, dtype=DType.FLOAT) for key in provides],
+        name=name,
+    )
+    node.assign_flow(flow, name)
+    return node
+
+
+def test_a_long_chain_checks_each_candidate_once():
+    class CountingPipeline(Pipeline):
+        checks = 0
+
+        def _gate_blocks(self, node):
+            self.checks += 1
+            return super()._gate_blocks(node)
+
+    nodes = []
+    for i in range(100):
+
+        def relay(params, _index=i, **inputs):
+            return {f"v{_index + 1}": inputs[f"v{_index}"] + 1.0}
+
+        nodes.append(_node(f"n{i}", relay, requires=[f"v{i}"], provides=[f"v{i + 1}"]))
+    pipeline = CountingPipeline(nodes)
+
+    pipeline.run({"f.v0": 0.0})
+
+    assert pipeline.values()["f.v100"]["value"] == 100.0
+    assert pipeline.checks == len(nodes)
+
+
+@pytest.mark.parametrize("disabled", [False, True])
+def test_a_gate_opened_mid_wave_releases_its_waiting_candidate(disabled):
+    running, release, held, acted = (threading.Event() for _ in range(4))
+
+    class WatchingPipeline(Pipeline):
+        def _gate_blocks(self, node):
+            blocked = super()._gate_blocks(node)
+            if blocked:
+                held.set()
+            return blocked
+
+    def slow(params):
+        running.set()
+        assert release.wait(3)
+
+    pipeline = WatchingPipeline(
+        [
+            _node("slow", slow, flow="running"),
+            _node("action", lambda params: acted.set(), flow="paused"),
+        ]
+    )
+    if disabled:
+        pipeline.set_disabled({"paused"})
+    else:
+        pipeline.pause("paused")
+    with ThreadPoolExecutor(max_workers=1) as driver:
+        future = driver.submit(pipeline.run)
+        try:
+            assert running.wait(1) and held.wait(1)
+            if disabled:
+                pipeline.set_disabled(set())
+            with nullcontext() if disabled else pipeline.stepping("paused"):
+                release.set()
+                future.result(2)
+            assert acted.is_set()
+        finally:
+            release.set()
diff --git a/backend/tests/flow/test_socket_catchup.py b/backend/tests/flow/test_socket_catchup.py
new file mode 100644
index 0000000..7014976
--- /dev/null
+++ b/backend/tests/flow/test_socket_catchup.py
@@ -0,0 +1,129 @@
+"""Direct and portal browsers recover state without widening camera access."""
+
+import asyncio
+import json
+from contextlib import suppress
+from types import SimpleNamespace
+
+import pytest
+
+from fluksio.api import deps
+from fluksio.api.routes import flows
+from fluksio.cloud.connector import CloudConnector
+from fluksio.flow import events
+from fluksio.flow.events import QUEUE_SIZE, EventBus
+
+
+@pytest.mark.parametrize("tunnel", [False, True])
+@pytest.mark.parametrize("burst", [0, QUEUE_SIZE + 1])
+def test_a_browser_catches_changes_during_its_snapshot(monkeypatch, tunnel, burst):
+    async def run():
+        bus = EventBus()
+        bus.bind(asyncio.get_running_loop())
+        monkeypatch.setattr(events, "event_bus", bus)
+        monkeypatch.setattr(flows, "event_bus", bus)
+        user = SimpleNamespace(id="viewer", is_superuser=False)
+        for module in (flows, deps):
+            monkeypatch.setattr(module, "user_from_token", lambda *a, **kw: user)
+            monkeypatch.setattr(
+                module, "hidden_from", lambda *a, **kw: frozenset({"f.camera"})
+            )
+        monkeypatch.setattr(flows, "panel_scope", lambda *a: None)
+        monkeypatch.setattr(flows, "socket_token", lambda *a: "test-token")
+
+        async def immediate(function, *args):
+            return function(*args)
+
+        monkeypatch.setattr(flows, "run_in_threadpool", immediate)
+        current = {"f.flag": {"value": 0}, "f.camera": {"value": "private"}}
+        controller = SimpleNamespace(
+            values=lambda: dict(current),
+            node_statuses=lambda: [],
+            issues=[],
+            paused_flows=lambda: [],
+            placer=None,
+        )
+        app = SimpleNamespace(state=SimpleNamespace(flow_controller=controller))
+        received = []
+        client_values = {}
+        caught_up = asyncio.Event()
+        drained = asyncio.Event()
+
+        class Socket:
+            headers = {}
+
+            async def accept(self):
+                pass
+
+            async def receive(self):
+                await asyncio.Future()
+
+            async def send(self, text):
+                frame = json.loads(text)
+                if frame["op"] == "ws_msg":
+                    await self.send_text(frame["text"])
+
+            async def send_text(self, text):
+                payload = json.loads(text)
+                received.append(payload)
+                if payload["type"] == "snapshot":
+                    assert "f.camera" not in payload["values"]
+                    client_values.update(payload["values"])
+                    if payload["values"]["f.flag"]["value"] == 1:
+                        caught_up.set()
+                    elif len(received) == 1:
+                        # The state was read, but its frame is still sending.
+                        # This value changes only once, then a burst loses its
+                        # event and leaves only a snapshot able to recover it.
+                        current["f.flag"] = {"value": 1}
+                        bus.publish(
+                            {"type": "message_value", "name": "f.flag", "value": 1}
+                        )
+                        for i in range(burst):
+                            current["f.noise"] = {"value": i}
+                            bus.publish(
+                                {"type": "message_value", "name": "f.noise", "value": i}
+                            )
+                        # Recovery must forward retained critical events too.
+                        bus.publish({"type": "node_error", "node": "f.example"})
+                        await asyncio.sleep(0)
+                for event in payload.get("events", [payload]):
+                    if event["type"] == "message_value":
+                        client_values[event["name"]] = {"value": event["value"]}
+                    if event.get("name") == "f.flag" and event.get("value") == 1:
+                        caught_up.set()
+                    if event["type"] == "node_error":
+                        drained.set()
+
+        socket = Socket()
+        socket.app = app
+        if tunnel:
+            action = CloudConnector(app)._stream(
+                socket,
+                {"id": "review", "path": "/api/v1/flows/ws", "query": "token=test"},
+            )
+        else:
+            action = flows.flow_events(socket)
+        task = asyncio.create_task(action)
+        try:
+            await asyncio.wait_for(caught_up.wait(), timeout=2)
+            await asyncio.wait_for(drained.wait(), timeout=2)
+        finally:
+            task.cancel()
+            with suppress(asyncio.CancelledError):
+                await task
+        assert sum(item["type"] == "snapshot" for item in received) == (
+            2 if burst else 1
+        )
+        flattened = [
+            event for payload in received for event in payload.get("events", [payload])
+        ]
+        assert sum(event["type"] == "node_error" for event in flattened) == 1
+        if burst:
+            # The client replaces values in arrival order. A snapshot ahead
+            # of the retained backlog temporarily rolls values backwards.
+            assert flattened[-1]["type"] == "snapshot"
+        assert client_values == {k: v for k, v in current.items() if k != "f.camera"}
+        assert not bus._subscribers
+
+    asyncio.run(run())
diff --git a/backend/tests/flow/test_timer_capacity.py b/backend/tests/flow/test_timer_capacity.py
new file mode 100644
index 0000000..2da9d91
--- /dev/null
+++ b/backend/tests/flow/test_timer_capacity.py
@@ -0,0 +1,112 @@
+"""Deadlines retain worker capacity when ordinary work is blocked."""
+
+import threading
+import time
+
+import pytest
+
+from fluksio.flow import executor
+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.queue import MemoryWorkQueue, WorkItem
+
+
+def _node(name, function, requires=(), provides=()):
+    node = Node(
+        f=function,
+        requires=[MessageSpec(name=key, dtype=DType.FLOAT) for key in requires],
+        provides=[MessageSpec(name=key, dtype=DType.FLOAT) for key in provides],
+        name=name,
+    )
+    node.assign_flow("f", name)
+    return node
+
+
+@pytest.mark.parametrize("reclaimed", [False, True])
+def test_a_due_node_runs_while_ordinary_nodes_and_cascades_are_full(
+    monkeypatch, reclaimed
+):
+    started, release, fired, recovered, touched = (threading.Event() for _ in range(5))
+    calls = []
+
+    class Queue(MemoryWorkQueue):
+        def reclaim_stale(self, min_idle_ms):
+            if not reclaimed or not started.is_set() or recovered.is_set():
+                return []
+            recovered.set()
+            return [
+                WorkItem(
+                    kind="cascade",
+                    node="f.source",
+                    flow="f",
+                    outputs={"f.raw": float(i)},
+                    entry_id=f"recovered-{i}",
+                )
+                for i in range(3)
+            ]
+
+        def touch(self, entry_ids):
+            if "recovered-0" in entry_ids:
+                touched.set()
+
+    def slow(raw, params):
+        calls.append(raw)
+        started.set()
+        assert release.wait(5)
+
+    source = _node("source", lambda params: None, provides=["raw"])
+    slow_node = _node("slow", slow, requires=["raw"])
+    timer = _node("timer", lambda params: None, provides=["stop"])
+    stop = _node("stop", lambda stop, params: fired.set(), requires=["stop"])
+    queue = Queue()
+    service = ExecutionService(queue, max_workers=1, max_cascades=1)
+    pipeline = Pipeline(
+        [source, slow_node, timer, stop], work_queue=queue, node_pool=service.node_pool
+    )
+    service.bind(pipeline)
+    monkeypatch.setattr(executor, "DELAYED_INTERVAL_S", 0.02)
+    monkeypatch.setattr(executor, "RECLAIM_INTERVAL_S", 0.02)
+    monkeypatch.setattr(executor, "TOUCH_INTERVAL_S", 0.02)
+    service.start()
+    try:
+        source.inject({"raw": 1.0})
+        assert started.wait(1)
+        if reclaimed:
+            assert touched.wait(1), "reclaimed work blocked the timer's heartbeat"
+        else:
+            for i in range(3):
+                source.inject({"raw": float(i)})
+        queue.add_delayed(
+            WorkItem(kind="cascade", node=timer.id, flow="f", outputs={"f.stop": 1.0}),
+            time.time(),
+        )
+        assert fired.wait(2), "the due node waited for an ordinary worker"
+        assert len(calls) == 1, "ordinary work used the timer reserve"
+    finally:
+        release.set()
+        service.stop()
+
+
+def test_timers_can_use_all_idle_normal_workers():
+    """A reserve must not cap an otherwise idle engine at two workers."""
+    service = ExecutionService(MemoryWorkQueue(), max_workers=3, max_cascades=3)
+    started = [threading.Event() for _ in range(3)]
+    release = threading.Event()
+
+    def block(index):
+        started[index].set()
+        assert release.wait(3)
+
+    futures = []
+    try:
+        for i in range(3):
+            futures.append(service._node_pools.due.submit(block, i))
+        assert all(event.wait(1) for event in started)
+        assert service._node_pools.available() == (0, executor.DUE_RESERVE)
+    finally:
+        release.set()
+        for future in futures:
+            future.result(1)
+        service.stop()
diff --git a/backend/tests/test_metrics.py b/backend/tests/test_metrics.py
index 7d8d03a..d1e5fe9 100644
--- a/backend/tests/test_metrics.py
+++ b/backend/tests/test_metrics.py
@@ -160,6 +160,61 @@ def test_a_flush_the_database_refused_is_written_by_the_next_one(
     assert audit.detail == "held back"
 
 
+def test_batched_rollups_add_and_redelivery_clears_a_finished_run(db: Session) -> None:
+    collector = MetricsCollector(EventBus())
+    ts = time.time()
+    for i in range(2):
+        collector.handle(
+            {"type": "cascade_started", "run": f"bulk-{i}", "flow": "bulk", "ts": ts}
+        )
+        collector.handle(
+            {
+                "type": "node_executed",
+                "flow": "bulk",
+                "node": f"bulk.n{i}",
+                "run": f"bulk-{i}",
+                "outputs": 1,
+                "duration_ms": 10.0,
+                "ts": ts,
+            }
+        )
+    collector.handle({"type": "cascade_finished", "run": "bulk-0", "ts": ts + 1})
+    # Write the same open/finished batch twice, as a retained run is upserted.
+    collector._write(collector._buckets, [], list(collector._runs.values()), False)
+    collector.handle(
+        {
+            "type": "cascade_started",
+            "run": "bulk-0",
+            "flow": "bulk",
+            "deliveries": 2,
+            "ts": ts + 2,
+        }
+    )
+    collector.handle({"type": "cascade_finished", "run": "bulk-1", "ts": ts + 3})
+    for bucket in collector._buckets.values():
+        bucket["duration_sum_ms"] = 4.0
+        bucket["duration_max_ms"] = 4.0
+    collector._write(collector._buckets, [], list(collector._runs.values()), False)
+
+    buckets = db.exec(select(MetricBucket).where(MetricBucket.flow == "bulk")).all()
+    assert len(buckets) == 2
+    assert all(
+        (b.executions, b.messages, b.duration_sum_ms, b.duration_max_ms)
+        == (2, 2, 14.0, 10.0)
+        for b in buckets
+    )
+    reopened = db.get(FlowRun, "bulk-0")
+    assert reopened is not None
+    assert (reopened.status, reopened.finished_at, reopened.deliveries) == (
+        "running",
+        None,
+        2,
+    )
+    finished = db.get(FlowRun, "bulk-1")
+    assert finished is not None and finished.finished_at is not None
+    assert finished.status == "ok"
+
+
 def test_a_node_id_wider_than_the_column_still_records(db: Session) -> None:
     collector = MetricsCollector(EventBus())
     ts = datetime.now(UTC).replace(second=0, microsecond=0).timestamp()
diff --git a/backend/tests/test_spa.py b/backend/tests/test_spa.py
index f5ddde6..a60a78f 100644
--- a/backend/tests/test_spa.py
+++ b/backend/tests/test_spa.py
@@ -1,7 +1,9 @@
 import base64
+import gzip
 import hashlib
 from pathlib import Path
 
+import pytest
 from starlette.applications import Starlette
 from starlette.responses import PlainTextResponse
 from starlette.routing import Route
@@ -37,3 +39,53 @@ def test_the_dashboard_and_the_api_share_one_origin(tmp_path: Path) -> None:
     assert client.get("/api/v1/ping").text == "api"
     # Nothing outside the build directory is served, only the shell.
     assert THEME in client.get("/assets/../../etc/passwd").text
+
+
+@pytest.mark.parametrize(
+    ("accepted", "compressed"),
+    [
+        ("gzip", True),
+        ("br, gzip;q=0.8", True),
+        ("gzip ; q=1, identity;q=0", True),
+        ("*", True),
+        ("gzip;q=0, *;q=1", False),
+        ("gzip;q=0.5, identity;q=1", False),
+        ("br", False),
+        ("identity", False),
+    ],
+)
+def test_precompressed_assets_keep_their_type_and_negotiate_encoding(
+    tmp_path: Path, accepted: str, compressed: bool
+) -> None:
+    client = _client(tmp_path)
+    content = b"console.log('built once');\n" * 200
+    asset = tmp_path / "assets" / "app-1a2b.js"
+    asset.write_bytes(content)
+    packed = gzip.compress(content)
+    asset.with_name(asset.name + ".gz").write_bytes(packed)
+
+    response = client.get("/assets/app-1a2b.js", headers={"Accept-Encoding": accepted})
+
+    assert response.content == content
+    assert (response.headers.get("content-encoding") == "gzip") is compressed
+    assert "javascript" in response.headers["content-type"]
+    assert response.headers["vary"] == "Accept-Encoding"
+    assert "immutable" in response.headers["cache-control"]
+    expected = len(packed) if compressed else len(content)
+    assert int(response.headers["content-length"]) == expected
+    head = client.head("/assets/app-1a2b.js", headers={"Accept-Encoding": accepted})
+    assert head.content == b""
+    assert head.headers["content-length"] == str(expected)
+
+
+def test_a_build_without_compressed_assets_still_serves_plain_files(
+    tmp_path: Path,
+) -> None:
+    client = _client(tmp_path)
+    content = b"console.log('plain build');\n" * 200
+    (tmp_path / "assets" / "app-1a2b.js").write_bytes(content)
+
+    response = client.get("/assets/app-1a2b.js", headers={"Accept-Encoding": "gzip"})
+
+    assert response.content == content
+    assert "content-encoding" not in response.headers
diff --git a/development.md b/development.md
index 205327d..1de4bba 100644
--- a/development.md
+++ b/development.md
@@ -80,6 +80,10 @@ make rebuild-frontend
 
 This preserves the running stack's domain, overlays and LAN port.
 
+The frontend build writes gzip sidecars for compressible text assets. Keep
+them beside the original files when deploying `dist/`. The API serves the
+sidecar when the client accepts gzip and the original otherwise.
+
 ## Development services
 
 The development Compose file includes services for trying integrations:
diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md
index 3cc3fa0..1cdd9a2 100644
--- a/docs/reference/configuration.md
+++ b/docs/reference/configuration.md
@@ -151,6 +151,10 @@ The three concurrency limits must be at least `1`. They also have CLI flags:
 engine logs a warning at startup and uses the value anyway. Resource flags are
 `--gpus`, `--ram` and `--worker-max-rss`.
 
+Due timers can use up to two additional cascade threads and two additional
+node-execution threads when the ordinary pools are busy. Declared resource
+budgets still apply.
+
 Nodes with declared resources are scheduled against the available budget.
 Undeclared nodes use the shared pool and receive thread caps. Attached workers
 report their own inventory; the engine's resource settings describe only its
diff --git a/frontend/src/components/Common/UplotChart.tsx b/frontend/src/components/Common/UplotChart.tsx
index 6be1676..044ed5d 100644
--- a/frontend/src/components/Common/UplotChart.tsx
+++ b/frontend/src/components/Common/UplotChart.tsx
@@ -125,8 +125,8 @@ export const CURSOR: uPlot.Cursor = {
   } as unknown as uPlot.Cursor.Bind,
   drag: {
     // Drag across the plot to read a stretch of it closer. `setScale` stays
-    // off because the chart owns its x range itself: `setData` runs on every
-    // render and would re-range the scales from the data, so a held window is
+    // off because the chart owns its x range itself: `setData` would re-range
+    // the scales from the data, so a held window is
     // what tells it to leave them alone. Without that this only ever flashed a
     // selection box over a live chart, which is why it used to be off.
     x: true,
@@ -190,17 +190,35 @@ function tickLabels(ticks: number[]): string[] {
   return ticks.map((value) => value.toFixed(decimals))
 }
 
-/** The series joined onto one x axis, which is what uPlot draws. */
-function table(plots: HistoryPoint[][]): uPlot.AlignedData {
-  return uPlot.join(
-    plots.map(
-      (plot) =>
-        [
-          plot.map((point) => point.ts),
-          plot.map((point) => point.value),
-        ] as uPlot.AlignedData,
-    ),
-  )
+/** Numeric snapshots catch both rebuilt arrays and mutable caller buffers. */
+export class ChartData {
+  private columns: uPlot.AlignedData[] = []
+  private data: uPlot.AlignedData = [[]]
+
+  update(plots: HistoryPoint[][]): uPlot.AlignedData {
+    let changed = plots.length !== this.columns.length
+    this.columns.length = plots.length
+    plots.forEach((plot, index) => {
+      const previous = this.columns[index]
+      if (
+        previous?.[0].length === plot.length &&
+        plot.every(
+          (point, i) =>
+            Object.is(point.ts, previous[0][i]) &&
+            Object.is(point.value, previous[1][i]),
+        )
+      ) {
+        return
+      }
+      this.columns[index] = [
+        plot.map((point) => point.ts),
+        plot.map((point) => point.value),
+      ]
+      changed = true
+    })
+    if (changed) this.data = plots.length ? uPlot.join(this.columns) : [[]]
+    return this.data
+  }
 }
 
 /**
@@ -264,12 +282,16 @@ export function UplotChart({
   const host = useRef<HTMLDivElement>(null)
   const legend = useRef<HTMLDivElement>(null)
   const chart = useRef<uPlot | null>(null)
+  const prepared = useRef<ChartData | null>(null)
+  if (prepared.current === null) prepared.current = new ChartData()
   // A dragged x window, held so the next reading does not wash it away. The
   // ref is what the data effect reads; the state is only what draws the way
   // back out, and the two are set together.
   const zoomed = useRef(false)
+  const resetRange = useRef(false)
   const [showReset, setShowReset] = useState(false)
   const clearZoom = () => {
+    if (zoomed.current) resetRange.current = true
     zoomed.current = false
     setShowReset(false)
   }
@@ -432,10 +454,11 @@ export function UplotChart({
       },
       // Built with the readings it already has: uPlot's axes are only half
       // initialised while its scales have no range.
-      table(plots),
+      prepared.current!.update(plots),
       element,
     )
     chart.current = plot
+    resetRange.current = false
 
     const observer = new ResizeObserver(() => {
       plot.setSize({
@@ -455,27 +478,19 @@ export function UplotChart({
     }
   }, [key, ready])
 
-  // Every render, deliberately. `plots` is rebuilt on every render in both
-  // consumers, so no memo can hit, and any signature short of hashing shares
-  // the blind spot a point count has: once a rolling window is full, a refetch
-  // carrying different readings leaves the count where it was and never fires.
-  // Safe to run this often because `setData` is idempotent and re-ranges the
-  // scales *from the data* — the opposite of the `redraw(false)` below. That
-  // re-ranging is exactly what a dragged window has to be spared, so while one
-  // is held the data goes in and the scales stay where they were put. Clearing
-  // the window renders, which brings the next pass through here with the reset
-  // back on: that is what puts the whole range back.
-  //
-  // A live chart keeps a count of points rather than a span, so a held window
-  // scrolls off the left as readings arrive. Once its last point has gone it
-  // is let go, or the chart would sit empty until somebody reset it.
+  // Check contents every render: callers may rebuild arrays or mutate buffers.
+  // Only changed readings need joining and drawing. A zoom reset still ranges
+  // the current data, and a held window expires once all its points roll off.
   useEffect(() => {
     const plot = chart.current
     if (!plot || plots.length === 0) return
-    const data = table(plots)
+    const data = prepared.current!.update(plots)
     const held = plot.scales.x.max
     if (zoomed.current && held != null && held < data[0][0]) clearZoom()
-    plot.setData(data, !zoomed.current)
+    if (data !== plot.data || resetRange.current) {
+      plot.setData(data, !zoomed.current)
+      resetRange.current = false
+    }
   })
 
   // The canvas cannot follow a CSS variable, so a theme swap is a redraw. The
diff --git a/frontend/src/components/Common/chart-data.check.ts b/frontend/src/components/Common/chart-data.check.ts
new file mode 100644
index 0000000..770c1e9
--- /dev/null
+++ b/frontend/src/components/Common/chart-data.check.ts
@@ -0,0 +1,68 @@
+/** Chart data stays current without joining unchanged series on every render. */
+import assert from "node:assert/strict"
+import uPlot from "uplot"
+
+import { chartTails } from "../Dashboard/ChartWidget"
+import { ChartData } from "./UplotChart"
+
+const cache = new ChartData()
+const plots = Array.from({ length: 5 }, (_, series) =>
+  Array.from({ length: 500 }, (_, i) => ({
+    ts: i + series / 10,
+    value: i * (series + 1),
+  })),
+)
+const join = uPlot.join
+let joins = 0
+uPlot.join = (...args) => {
+  joins++
+  return join(...args)
+}
+try {
+  const initial = cache.update(plots)
+  for (let render = 0; render < 100; render++) {
+    const rebuilt = plots.map((plot) => plot.map((point) => ({ ...point })))
+    assert.equal(cache.update(rebuilt), initial)
+  }
+  assert.equal(joins, 1)
+
+  // A correction in the middle of an existing mutable buffer is still data.
+  plots[0][200].value = -7
+  const corrected = cache.update(plots)
+  assert.notEqual(corrected, initial)
+  assert.equal(corrected[1][corrected[0].indexOf(200)], -7)
+  assert.equal(initial[1][initial[0].indexOf(200)], 200)
+
+  // A full window rolls forward without changing its point count.
+  plots[0].shift()
+  plots[0].push({ ts: 500, value: 12 })
+  const rolled = cache.update(plots)
+  assert.notEqual(rolled, corrected)
+  assert.equal(rolled[0][0], 0.1)
+  assert.equal(rolled[1][rolled[0].indexOf(500)], 12)
+
+  const reordered = cache.update([plots[1], plots[0]])
+  assert.equal(reordered.length, 3)
+  assert.equal(reordered[1][reordered[0].indexOf(200.1)], 400)
+  assert.deepEqual(cache.update([]), [[]])
+} finally {
+  uPlot.join = join
+}
+
+let tails = chartTails({}, ["a"], [{ ts: 1, value: 10 }], 3)
+assert.equal(chartTails(tails, ["a"], [{ ts: 1, value: 10 }], 3), tails)
+tails = chartTails(tails, ["a"], [{ ts: 1, value: 11 }], 3)
+assert.deepEqual(tails.a, [{ ts: 1, value: 11 }])
+tails = chartTails(tails, ["a"], [{ ts: 2, value: 12 }], 3)
+tails = chartTails(tails, ["a"], [{ ts: 3, value: 13 }], 3)
+tails = chartTails(tails, ["a"], [{ ts: 3, value: 13 }], 2)
+assert.deepEqual(tails.a, [
+  { ts: 2, value: 12 },
+  { ts: 3, value: 13 },
+])
+tails = chartTails(tails, ["b"], [undefined], 2)
+assert.deepEqual(tails, { b: [] })
+assert.equal(chartTails(tails, ["b"], [undefined], 2), tails)
+assert.equal(chartTails(tails, ["b", "b"], [undefined, undefined], 2), tails)
+
+console.log("chart data: ok (101 unchanged renders, one join)")
diff --git a/frontend/src/components/Dashboard/ChartWidget.tsx b/frontend/src/components/Dashboard/ChartWidget.tsx
index f2af7f7..a8ee6ca 100644
--- a/frontend/src/components/Dashboard/ChartWidget.tsx
+++ b/frontend/src/components/Dashboard/ChartWidget.tsx
@@ -8,7 +8,11 @@ import {
   type Range,
 } from "@/components/Common/RangePicker"
 import { MAX_SERIES, UplotChart } from "@/components/Common/UplotChart"
-import { useLiveValue, useLiveValues } from "@/components/Flow/liveStore"
+import {
+  type LiveValue,
+  useLiveValue,
+  useLiveValues,
+} from "@/components/Flow/liveStore"
 import {
   compareQueryOptions,
   NO_CURVE,
@@ -91,6 +95,36 @@ function merge(fetched: HistoryPoint[], tail: HistoryPoint[], cap: number) {
   return [...fetched, ...tail.filter((point) => point.ts > since)].slice(-cap)
 }
 
+/** Advance the bounded live tails without scheduling a stale chart commit. */
+export function chartTails(
+  previous: Record<string, HistoryPoint[]>,
+  names: string[],
+  live: (LiveValue | undefined)[],
+  cap: number,
+): Record<string, HistoryPoint[]> {
+  const next: Record<string, HistoryPoint[]> = {}
+  let changed = false
+  names.forEach((name, index) => {
+    const kept = previous[name] ?? []
+    const point = live[index]
+    const last = kept[kept.length - 1]
+    let tail = kept
+    if (typeof point?.value === "number" && point.ts) {
+      if (last?.ts !== point.ts) {
+        tail = [...kept, { ts: point.ts, value: point.value }]
+      } else if (!Object.is(last.value, point.value)) {
+        tail = [...kept.slice(0, -1), { ts: point.ts, value: point.value }]
+      }
+    }
+    if (tail.length > cap) tail = tail.slice(-cap)
+    next[name] = tail
+    if (tail !== previous[name]) changed = true
+  })
+  return changed || Object.keys(previous).length !== Object.keys(next).length
+    ? next
+    : previous
+}
+
 /**
  * A chart, drawn from whichever source it was pointed at.
  *
@@ -275,34 +309,10 @@ function LiveChart({ widget }: WidgetProps) {
   // What arrived over the socket since the history was fetched. Appending beats
   // refetching the whole series every time a value lands.
   const [tails, setTails] = useState<Record<string, HistoryPoint[]>>({})
-  const key = names.join(" ")
-  const stamps = live.map((value) => value?.ts ?? 0).join(",")
-  // biome-ignore lint/correctness/useExhaustiveDependencies: the timestamps are what makes a tail longer; `live` is a fresh array every render.
-  useEffect(() => {
-    setTails((previous) => {
-      const next: Record<string, HistoryPoint[]> = {}
-      let changed = false
-      names.forEach((name, index) => {
-        const point = live[index]
-        const kept = previous[name] ?? []
-        const ts = point?.ts
-        if (
-          typeof point?.value !== "number" ||
-          !ts ||
-          kept[kept.length - 1]?.ts === ts
-        ) {
-          next[name] = kept
-          return
-        }
-        next[name] = [...kept, { ts, value: point.value }].slice(-cap)
-        changed = true
-      })
-      // Dropping a series is a change too, even without a new reading.
-      return changed || Object.keys(previous).length !== names.length
-        ? next
-        : previous
-    })
-  }, [key, stamps, cap])
+  const nextTails = chartTails(tails, names, live, cap)
+  // React retries this component before rendering its children. An effect
+  // would commit the old chart first, then render and draw it a second time.
+  if (nextTails !== tails) setTails(nextTails)
 
   const plots = names.map((name, index) =>
     merge(
diff --git a/frontend/vite.config.ts b/frontend/vite.config.ts
index 0b7ae83..7ee155e 100644
--- a/frontend/vite.config.ts
+++ b/frontend/vite.config.ts
@@ -1,8 +1,36 @@
+import { readdir, readFile, writeFile } from "node:fs/promises"
 import path from "node:path"
+import { gzipSync } from "node:zlib"
 import tailwindcss from "@tailwindcss/vite"
 import { tanstackRouter } from "@tanstack/router-plugin/vite"
 import react from "@vitejs/plugin-react-swc"
-import { defineConfig } from "vite"
+import { defineConfig, type Plugin } from "vite"
+
+/** Compress once at build time, including text copied from public/. */
+function precompress(): Plugin {
+  async function compress(directory: string): Promise<void> {
+    const entries = await readdir(directory, { withFileTypes: true })
+    await Promise.all(
+      entries.map(async (entry) => {
+        const file = path.join(directory, entry.name)
+        if (entry.isDirectory()) return compress(file)
+        if (!/\.(?:js|css|html|svg|json|webmanifest)$/.test(entry.name)) return
+        const original = await readFile(file)
+        if (original.length < 1024) return
+        const compressed = gzipSync(original, { level: 9 })
+        if (compressed.length < original.length)
+          await writeFile(`${file}.gz`, compressed)
+      }),
+    )
+  }
+  return {
+    name: "precompressed-assets",
+    apply: "build",
+    async writeBundle({ dir }) {
+      if (dir) await compress(dir)
+    },
+  }
+}
 
 // https://vitejs.dev/config/
 export default defineConfig({
@@ -39,5 +67,6 @@ export default defineConfig({
     }),
     react(),
     tailwindcss(),
+    precompress(),
   ],
 })