Wait for a deadline instead of polling for one

The timer thread promoted due work on a fixed one-second tick, so every
delayed item was 0-1000ms late whatever the load — measured on the house
at 705ms mean on a rollershutter stop, which is 2-4% of a 26-second
travel and accumulates in the position the motor node believes it is at.
It now sleeps to the soonest deadline and is woken when a nearer one is
scheduled, which measures 0.9ms end to end through Redis.

A promoted timer also went to the back of the queue. It goes into a due
lane of its own that `claim` reads first, so work that has waited out a
deadline is not held up by work that is merely queued.

Beside it, in the same code: seeding a message now bumps its version, so
a re-put flow's synchronous nodes no longer wait forever on a value that
is sitting in state; the consumer group drops the consumers of engines
that are gone (138 had accumulated on this installation); and the cast
that closes the long-standing `xclaim` mypy error.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

backend/fluksio/flow/executor.py

@@ -36,6 +36,10 @@ RECLAIM_INTERVAL_S = 30.0
 # node may run for as long as it likes, so what marks an item abandoned is this
 # stopping — which is what an engine that died does.
 TOUCH_INTERVAL_S = 20.0
+# The longest the timer thread sleeps with nothing due. It is a housekeeping
+# cadence and a backstop for a deadline written by another process, not the
+# 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
 #: How many cascades may be in flight, unless the service is given a number.
 #: Sustained throughput is this over the mean cascade time, so an installation
@@ -67,6 +71,10 @@ class ExecutionService:
         self.max_cascades = max_cascades or MAX_CASCADES
         self._pipeline: Pipeline | None = None
         self._stop = threading.Event()
+        # Set when a deadline moves closer, so the timer thread stops waiting
+        # on the one it read and goes back for the new one.
+        self._timer_wake = threading.Event()
+        queue.on_delayed = self._timer_wake.set
         self._intake = threading.Event()
         self._intake.set()
         self._inflight = 0
@@ -104,6 +112,8 @@ class ExecutionService:
 
     def stop(self) -> None:
         self._stop.set()
+        # The timer thread sleeps on this, not on _stop.
+        self._timer_wake.set()
         for thread in (self._consumer, self._timers):
             if thread is not None:
                 thread.join(timeout=5)
@@ -173,19 +183,44 @@ class ExecutionService:
             for item in items:
                 self._dispatch(item)
 
+    def _sleep_until_due(self) -> None:
+        """Wait for the soonest deadline, the housekeeping cap, or a new one.
+
+        A fixed poll here made every delayed item late by 0-1000ms whatever the
+        load — on a rollershutter driven for a measured 26 seconds, 2-4% of its
+        travel every time, accumulating in the position its node believes it is
+        at. Sleeping to the deadline instead leaves a wake-up and a promotion,
+        which is milliseconds.
+        """
+        self._timer_wake.clear()
+        try:
+            # Read after the clear: a deadline arriving in between sets the
+            # event again, so the wait below returns immediately rather than
+            # sleeping through work that landed in the gap.
+            due = self.queue.next_due()
+        except Exception as exc:
+            logger.error("Could not read the next deadline: %s", exc)
+            due = None
+        wait = DELAYED_INTERVAL_S if due is None else due - time.time()
+        self._timer_wake.wait(min(max(wait, 0.0), DELAYED_INTERVAL_S))
+
     def _tick(self) -> None:
         """Promote delayed items, and take back what a dead engine dropped."""
         last_reclaim = 0.0
         last_touch = 0.0
         last_backlog = 0.0
         while not self._stop.is_set():
-            self._stop.wait(DELAYED_INTERVAL_S)
+            self._sleep_until_due()
             if self._stop.is_set():
                 break
             try:
                 self.queue.move_due(time.time())
             except Exception as exc:
                 logger.error("Could not promote delayed work: %s", exc)
+                # The item is still due, so the wait above would be zero and
+                # this would spin on a queue that is down. Back off to what a
+                # fixed poll used to cost.
+                self._stop.wait(DELAYED_INTERVAL_S)
 
             now = time.monotonic()
             if now - last_backlog >= BACKLOG_INTERVAL_S:

backend/fluksio/flow/pipeline.py

@@ -395,13 +395,24 @@ class Pipeline:
         self.replace_flow(flow, [])
 
     def _seed(self, initial_values: dict[str, Any] | None) -> None:
-        """Give messages a starting value, without overwriting one already there."""
+        """Give messages a starting value, without overwriting one already there.
+
+        A seeded value counts as having arrived. Writing it without its version
+        left the value in state at version 0, which `_check_synchronous_ready`
+        reads as "never published" — so a flow that was dropped and recreated
+        came back with its synchronous nodes waiting on inputs that were sitting
+        right there, reporting `active` and `ok` and never running again.
+        """
         if not initial_values:
             return
+        seeded = []
         with self._state.lock():
             for name, value in initial_values.items():
                 if name not in self._state:
                     self._state[name] = value
+                    seeded.append(name)
+        if seeded:
+            self._state.increment_multi([self._version_key(name) for name in seeded])
 
     def get_node_by_id(self, nid: str) -> Node | None:
         return next((n for n in self._nodes if n.id == nid), None)

backend/fluksio/flow/queue.py

@@ -18,6 +18,7 @@ import threading
 import time
 from abc import ABC, abstractmethod
 from collections import deque
+from collections.abc import Callable
 from dataclasses import dataclass, field
 from typing import Any, cast
 
@@ -33,6 +34,13 @@ STREAM_MAXLEN = 10_000
 # can look rather than letting it loop forever.
 MAX_DELIVERIES = 3
 GROUP = "engine"
+# Entry ids from the due lane wear this, so one id still says which stream it
+# came from. Deterministic per stream, which is what keeps it usable as an
+# idempotency key across redeliveries.
+DUE_PREFIX = "due:"
+# A consumer this far past its last read belongs to an engine that is gone. A
+# live one interacts every claim, so nothing in service comes close.
+STALE_CONSUMER_IDLE_MS = 3_600_000
 
 
 @dataclass
@@ -113,6 +121,21 @@ class WorkItem:
 class WorkQueue(ABC):
     """What the execution service pulls from."""
 
+    #: Called after an item is journaled for later, so whoever is waiting on
+    #: the next deadline can find out it just moved closer. Set by the
+    #: execution service; a queue nobody is timing leaves it alone.
+    on_delayed: Callable[[], None] | None = None
+
+    def _woken(self) -> None:
+        """Say a deadline changed. Never lets a listener break the enqueue."""
+        wake = self.on_delayed
+        if wake is None:
+            return
+        try:
+            wake()
+        except Exception:
+            logger.exception("Delayed-work listener failed")
+
     @abstractmethod
     def add(self, item: WorkItem) -> None:
         """Journal an item for immediate execution."""
@@ -121,6 +144,15 @@ class WorkQueue(ABC):
     def add_delayed(self, item: WorkItem, not_before: float) -> None:
         """Journal an item that must not run before ``not_before``."""
 
+    @abstractmethod
+    def next_due(self) -> float | None:
+        """When the soonest delayed item comes due, or None if there is none.
+
+        What lets the engine sleep until a timer rather than poll for it: a
+        fixed poll makes every delayed item late by up to its interval, which
+        a motor driven for a measured number of seconds notices.
+        """
+
     @abstractmethod
     def claim(self, count: int, block_ms: int) -> list[WorkItem]:
         """Take up to ``count`` items, waiting up to ``block_ms`` for one."""
@@ -237,6 +269,13 @@ class MemoryWorkQueue(WorkQueue):
             self._counter += 1
             heapq.heappush(self._delayed, (not_before, self._counter, item))
             self._wake.notify()
+        # Outside the lock: the listener is whoever is asleep until the old
+        # deadline, and it comes straight back here to read the new one.
+        self._woken()
+
+    def next_due(self) -> float | None:
+        with self._lock:
+            return self._delayed[0][0] if self._delayed else None
 
     def claim(self, count: int, block_ms: int) -> list[WorkItem]:
         deadline = time.monotonic() + block_ms / 1000.0
@@ -262,14 +301,17 @@ class MemoryWorkQueue(WorkQueue):
         return []
 
     def move_due(self, now: float) -> int:
-        moved = 0
+        due: list[WorkItem] = []
         with self._wake:
             while self._delayed and self._delayed[0][0] <= now:
-                self._items.append(heapq.heappop(self._delayed)[2])
-                moved += 1
-            if moved:
+                due.append(heapq.heappop(self._delayed)[2])
+            if due:
+                # In front of the backlog, in due order: an item that has
+                # waited out a deadline is late by however long it queues
+                # here, while work merely enqueued is not waiting on a clock.
+                self._items.extendleft(reversed(due))
                 self._wake.notify()
-        return moved
+        return len(due)
 
     def dead_letter(self, item: WorkItem, reason: str) -> None:
         logger.error("Dropping work item for '%s': %s", item.node, reason)
@@ -343,6 +385,10 @@ class RedisWorkQueue(WorkQueue):
         self._ns = namespace
         self._consumer = consumer or f"engine-{int(time.time() * 1000) % 1_000_000}"
         self._stream = f"{namespace}:__queue__"
+        # Promoted timers, read before the backlog. A stream of its own rather
+        # than a flag on the entry, because a stream is ordered by id and there
+        # is no way to put a late item in front of one already in it.
+        self._due_stream = f"{namespace}:__due__"
         self._delayed_key = f"{namespace}:__delayed__"
         self._dead_key = f"{namespace}:__dead__"
         # Which flows have something parked. Maintained rather than discovered:
@@ -350,14 +396,43 @@ class RedisWorkQueue(WorkQueue):
         # key in the database, twice per health poll.
         self._parked_flows_key = f"{namespace}:__parked_flows__"
         self._ensure_group()
+        self._drop_stale_consumers()
 
     def _ensure_group(self) -> None:
+        for stream in (self._stream, self._due_stream):
             try:
-            self._redis.xgroup_create(self._stream, GROUP, id="0", mkstream=True)
+                self._redis.xgroup_create(stream, GROUP, id="0", mkstream=True)
             except redis.ResponseError as exc:
                 if "BUSYGROUP" not in str(exc):
                     raise
 
+    def _drop_stale_consumers(self) -> None:
+        """Forget the consumers of engines that are gone.
+
+        One is registered per engine start and none was ever removed, so the
+        list grew without bound — and `XPENDING`, `XAUTOCLAIM` and `XINFO` all
+        walk it. Only consumers holding nothing and untouched for an hour are
+        dropped, so a live engine blocked on a read is never one of them; a
+        consumer is recreated by its next read anyway.
+        """
+        for stream in (self._stream, self._due_stream):
+            try:
+                consumers = cast(
+                    list[dict[str, Any]], self._redis.xinfo_consumers(stream, GROUP)
+                )
+            except redis.ResponseError:
+                continue
+            for row in consumers:
+                name = str(row.get("name", ""))
+                if name == self._consumer or row.get("pending"):
+                    continue
+                if int(row.get("idle") or 0) < STALE_CONSUMER_IDLE_MS:
+                    continue
+                try:
+                    self._redis.xgroup_delconsumer(stream, GROUP, name)
+                except redis.ResponseError:
+                    pass
+
     def _parked_key(self, flow: str) -> str:
         return f"{self._ns}:__parked__:{flow}"
 
@@ -377,27 +452,56 @@ class RedisWorkQueue(WorkQueue):
         self._redis.zadd(
             self._delayed_key, {orjson.dumps(item.to_fields()): not_before}
         )
+        self._woken()
+
+    def next_due(self) -> float | None:
+        soonest = cast(
+            list[tuple[str, float]],
+            self._redis.zrange(self._delayed_key, 0, 0, withscores=True),
+        )
+        return soonest[0][1] if soonest else None
+
+    def _route(self, entry_id: str) -> tuple[str, str]:
+        """The stream an entry id belongs to, and its plain id there."""
+        if entry_id.startswith(DUE_PREFIX):
+            return self._due_stream, entry_id[len(DUE_PREFIX) :]
+        return self._stream, entry_id
 
     def claim(self, count: int, block_ms: int) -> list[WorkItem]:
+        """Take up to ``count`` items, due timers before anything queued.
+
+        One read over both streams rather than a read each: the block has to
+        cover the due lane, or an item promoted while this waits sits there
+        until the block expires — which is the second of lateness the sleeping
+        timer thread exists to remove. ``COUNT`` is per stream, so a read may
+        return up to twice what was asked for; the excess is bounded by the
+        cascade pool's own free count, and starting a due timer is worth more
+        than holding that number exactly.
+        """
         response = cast(
             list[Any],
             self._redis.xreadgroup(
                 GROUP,
                 self._consumer,
-                {self._stream: ">"},
+                {self._due_stream: ">", self._stream: ">"},
                 count=count,
                 block=block_ms,
             ),
         )
-        items: list[WorkItem] = []
-        for _stream, entries in response or []:
+        due: list[WorkItem] = []
+        backlog: list[WorkItem] = []
+        for stream, entries in response or []:
+            is_due = stream == self._due_stream
+            into = due if is_due else backlog
+            prefix = DUE_PREFIX if is_due else ""
             for entry_id, fields in entries:
-                items.append(WorkItem.from_fields(fields, entry_id))
-        return items
+                into.append(WorkItem.from_fields(fields, prefix + entry_id))
+        return due + backlog
 
     def ack(self, item: WorkItem) -> None:
         if item.entry_id:
-            self._redis.xack(self._stream, GROUP, item.entry_id)
+            stream, entry_id = self._route(item.entry_id)
+            self._redis.xack(stream, GROUP, entry_id)
 
     def touch(self, entry_ids: list[str]) -> None:
         if not entry_ids:
@@ -407,21 +511,33 @@ class RedisWorkQueue(WorkQueue):
         # delivery count where it is, so a long node neither redelivers nor
         # spends its way towards the dead-letter cap. An entry already
         # acknowledged is simply not there, and this is a no-op for it.
+        by_stream: dict[str, list[str]] = {}
+        for entry_id in entry_ids:
+            stream, plain = self._route(entry_id)
+            by_stream.setdefault(stream, []).append(plain)
+        for stream, ids in by_stream.items():
             self._redis.xclaim(
-            self._stream,
+                stream,
                 GROUP,
                 self._consumer,
                 min_idle_time=0,
-            message_ids=entry_ids,
+                # redis-py types `message_ids` as an invariant list of every id
+                # shape it accepts, which a list[str] is not assignable to.
+                message_ids=cast("list[int | bytes | str | memoryview]", ids),
                 justid=True,
             )
 
     def reclaim_stale(self, min_idle_ms: int) -> list[WorkItem]:
         """Take over entries a dead consumer never acknowledged."""
+        return self._reclaim(self._due_stream, DUE_PREFIX, min_idle_ms) + self._reclaim(
+            self._stream, "", min_idle_ms
+        )
+
+    def _reclaim(self, stream: str, prefix: str, min_idle_ms: int) -> list[WorkItem]:
         _cursor, entries, _deleted = cast(
             tuple[Any, list[Any], Any],
             self._redis.xautoclaim(
-                self._stream,
+                stream,
                 GROUP,
                 self._consumer,
                 min_idle_time=min_idle_ms,
@@ -435,12 +551,12 @@ class RedisWorkQueue(WorkQueue):
         counts: dict[str, int] = {}
         for record in cast(
             list[dict[str, Any]],
-            self._redis.xpending_range(self._stream, GROUP, min="-", max="+", count=64),
+            self._redis.xpending_range(stream, GROUP, min="-", max="+", count=64),
         ):
             counts[record["message_id"]] = record["times_delivered"]
 
         return [
-            WorkItem.from_fields(fields, entry_id, counts.get(entry_id, 1))
+            WorkItem.from_fields(fields, prefix + entry_id, counts.get(entry_id, 1))
             for entry_id, fields in entries
             if fields
         ]
@@ -464,8 +580,10 @@ class RedisWorkQueue(WorkQueue):
         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._stream,
+                    self._due_stream,
                     cast(Any, orjson.loads(raw)),
                     maxlen=STREAM_MAXLEN,
                     approximate=True,
@@ -514,24 +632,29 @@ class RedisWorkQueue(WorkQueue):
         pipe.srem(self._parked_flows_key, flow)
         pipe.execute()
 
-    def _group_info(self) -> dict[str, Any]:
+    def _group_info(self, stream: str = "") -> dict[str, Any]:
         """This consumer group's row of ``XINFO GROUPS``, or an empty one."""
         try:
-            groups = cast(list[dict[str, Any]], self._redis.xinfo_groups(self._stream))
+            groups = cast(
+                list[dict[str, Any]], self._redis.xinfo_groups(stream or self._stream)
+            )
         except redis.ResponseError:
             # No stream yet: nothing has ever been enqueued.
             return {}
         return next((g for g in groups if g.get("name") == GROUP), {})
 
     def backlog(self) -> int:
-        """Entries in the stream this group has never been handed.
+        """Entries in either stream this group has never been handed.
 
         Redis calls it the group's ``lag``. It is nil rather than zero when the
         stream has been trimmed under the group — entries that were dropped
         before anyone read them — and unknown is reported as none waiting,
         since the alternative is a health screen crying wolf after a trim.
         """
-        return int(self._group_info().get("lag") or 0)
+        return sum(
+            int(self._group_info(stream).get("lag") or 0)
+            for stream in (self._stream, self._due_stream)
+        )
 
     def backlog_flows(self, sample: int = 100) -> dict[str, int]:
         """Which flows the waiting work belongs to, from a sample of the tail.
@@ -556,18 +679,20 @@ class RedisWorkQueue(WorkQueue):
         return counts
 
     def stats(self) -> dict[str, Any]:
-        pending = cast(dict[str, Any], self._redis.xpending(self._stream, GROUP))
+        count = 0
         oldest = 0.0
-        count = pending.get("pending", 0) if isinstance(pending, dict) else 0
-        if count:
+        for stream in (self._stream, self._due_stream):
+            pending = cast(dict[str, Any], self._redis.xpending(stream, GROUP))
+            waiting = pending.get("pending", 0) if isinstance(pending, dict) else 0
+            count += waiting
+            if not waiting:
+                continue
             records = cast(
                 list[dict[str, Any]],
-                self._redis.xpending_range(
-                    self._stream, GROUP, min="-", max="+", count=1
-                ),
+                self._redis.xpending_range(stream, GROUP, min="-", max="+", count=1),
             )
             if records:
-                oldest = records[0]["time_since_delivered"] / 1000.0
+                oldest = max(oldest, records[0]["time_since_delivered"] / 1000.0)
         flows = cast(set[str], self._redis.smembers(self._parked_flows_key))
         parked = sum(cast(int, self._redis.llen(self._parked_key(f))) for f in flows)
         return {

backend/tests/flow/test_queue.py

@@ -43,6 +43,68 @@ def test_a_delayed_item_stays_put_until_it_is_due():
     assert [i.node for i in queue.claim(1, 10)] == ["f.n"]
 
 
+def test_the_queue_says_when_the_soonest_delayed_item_is_due():
+    """What lets the engine sleep to a deadline rather than poll for it."""
+    queue = MemoryWorkQueue()
+    assert queue.next_due() is None
+
+    queue.add_delayed(WorkItem(kind="cascade", node="f.late", flow="f"), 500.0)
+    queue.add_delayed(WorkItem(kind="cascade", node="f.soon", flow="f"), 100.0)
+    assert queue.next_due() == 100.0
+
+    queue.move_due(200.0)
+    assert queue.next_due() == 500.0
+
+
+def test_scheduling_says_a_deadline_moved():
+    """A delay scheduled for 200ms must cut short a sleep to the next second."""
+    queue = MemoryWorkQueue()
+    woken = threading.Event()
+    queue.on_delayed = woken.set
+
+    queue.add_delayed(WorkItem(kind="cascade", node="f.n", flow="f"), time.time() + 0.2)
+
+    assert woken.is_set()
+
+
+def test_a_due_timer_is_claimed_before_the_backlog():
+    """Lateness is what a timer is measured by; queued work waits on no clock."""
+    queue = MemoryWorkQueue()
+    for i in range(3):
+        queue.add(WorkItem(kind="cascade", node=f"f.queued{i}", flow="f"))
+    queue.add_delayed(WorkItem(kind="cascade", node="f.timer", flow="f"), 100.0)
+
+    queue.move_due(200.0)
+
+    assert [i.node for i in queue.claim(10, 10)][0] == "f.timer"
+
+
+def test_a_delay_fires_without_waiting_out_the_poll():
+    """The whole point: a 200ms delay is 200ms late, not up to a second."""
+    queue = MemoryWorkQueue()
+    service = ExecutionService(queue)
+    dispatched: list[float] = []
+    service._dispatch = lambda item: dispatched.append(time.monotonic())  # type: ignore[method-assign]
+    service.start()
+    try:
+        started = time.monotonic()
+        queue.add_delayed(
+            WorkItem(kind="cascade", node="f.n", flow="f"), time.time() + 0.2
+        )
+        # Claimed by the consumer thread, which is what calls _dispatch.
+        deadline = time.monotonic() + 2.0
+        while not dispatched and time.monotonic() < deadline:
+            time.sleep(0.01)
+    finally:
+        service.stop()
+
+    assert dispatched, "the deferred item never ran"
+    late = dispatched[0] - started - 0.2
+    # A fixed one-second poll made this up to 1.0s; the budget is generous
+    # because CI is not a real-time machine.
+    assert late < 0.3, f"fired {late:.3f}s late"
+
+
 def test_parked_work_comes_back_oldest_first():
     queue = MemoryWorkQueue()
     for i in range(3):

backend/tests/flow/test_synchronous_nodes.py

@@ -72,6 +72,37 @@ def test_synchronous_node_waits_for_all_inputs_to_be_fresh():
     assert sorted(runs) == ["eager", "sync"]
 
 
+def test_a_seeded_value_counts_as_having_arrived():
+    """A re-put flow used to come back with its synchronous nodes wedged.
+
+    Seeding wrote the value and not its version, and version 0 reads as
+    "never published" — so the node reported active and ok and never ran
+    again, with the value it was waiting for sitting right there in state.
+    """
+    runs: list[str] = []
+
+    def sync(a, b, params):
+        runs.append("sync")
+        return None
+
+    def sensor_b(params):
+        return {"b": 2.0}
+
+    b = node("b", sensor_b, provides=[spec("b")])
+    sync_node = node(
+        "sync", sync, requires=[spec("a"), spec("b")], params={"synchronous": True}
+    )
+
+    Pipeline(
+        nodes=[b, sync_node],
+        max_workers=1,
+        initial_values={"f.a": 1.0},
+    )
+
+    b.inject()
+    assert runs == ["sync"]
+
+
 def test_increment_and_multi_get():
     state = MemoryState()
 

docs/reference/node-types.md

@@ -155,6 +155,12 @@ Order of operations: rate check → delay → forward. With a `cron` and no inpu
 it emits the current time on each tick; with inputs it emits the last value it
 received.
 
+The engine sleeps until a delay is due rather than polling for it, so a `delay`
+fires within a few milliseconds of its deadline on an idle engine, and a due
+timer is taken off the queue before work that is merely waiting. It is not a
+real-time guarantee: the wait ends on a normal thread, and a node that then
+needs a busy cascade slot waits for one.
+
 !!! note "Not in a batch flow"
 
     A rate limit holds a value back for a timer to release, and a run has no