diff --git a/backend/fluksio/flow/executor.py b/backend/fluksio/flow/executor.py index f933a8e..8711df1 100644 --- a/backend/fluksio/flow/executor.py +++ b/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: diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index beddd25..7b0418b 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/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) diff --git a/backend/fluksio/flow/queue.py b/backend/fluksio/flow/queue.py index b9a9b12..f40581d 100644 --- a/backend/fluksio/flow/queue.py +++ b/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,13 +396,42 @@ 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: - try: - self._redis.xgroup_create(self._stream, GROUP, id="0", mkstream=True) - except redis.ResponseError as exc: - if "BUSYGROUP" not in str(exc): - raise + for stream in (self._stream, self._due_stream): + try: + 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. - self._redis.xclaim( - self._stream, - GROUP, - self._consumer, - min_idle_time=0, - message_ids=entry_ids, - justid=True, - ) + 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( + stream, + GROUP, + self._consumer, + min_idle_time=0, + # 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 { diff --git a/backend/tests/flow/test_queue.py b/backend/tests/flow/test_queue.py index 18d08ee..bde61bf 100644 --- a/backend/tests/flow/test_queue.py +++ b/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): diff --git a/backend/tests/flow/test_synchronous_nodes.py b/backend/tests/flow/test_synchronous_nodes.py index 403d126..2833f0f 100644 --- a/backend/tests/flow/test_synchronous_nodes.py +++ b/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() diff --git a/docs/reference/node-types.md b/docs/reference/node-types.md index 998edd1..8753fb3 100644 --- a/docs/reference/node-types.md +++ b/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