From badcf944915ba3410e8a7d8a06d66737f4b5f852 Mon Sep 17 00:00:00 2001 From: stroblme Date: Mon, 2 Nov 2026 20:21:17 +0100 Subject: [PATCH] Engine holds for a rename: the pipeline refuses publish and inline cascades with EngineHeld once those running have finished, the execution service stops claiming and its timers once claimed cascades have finished; both wait without a limit, report what is busy and can be aborted. REST answers 503 with Retry-After, a device frame busy. Redis adds no longer wait for each other, only a queue hold waits for them --- backend/fluksio/api/routes/flows.py | 7 +- backend/fluksio/flow/controller.py | 3 + backend/fluksio/flow/executor.py | 258 +++++++++++++------- backend/fluksio/flow/nodes/device.py | 6 +- backend/fluksio/flow/pipeline.py | 201 ++++++++++------ backend/fluksio/flow/queue.py | 21 +- backend/fluksio/main.py | 10 +- backend/tests/api/routes/test_flows.py | 34 +++ backend/tests/flow/test_engine_hold.py | 272 ++++++++++++++++++++++ backend/tests/flow/test_rename_runtime.py | 31 +++ 10 files changed, 687 insertions(+), 156 deletions(-) create mode 100644 backend/tests/flow/test_engine_hold.py diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index 082d1f4..93a3be6 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -65,7 +65,7 @@ from fluksio.flow.devices import ( from fluksio.flow.events import event_bus from fluksio.flow.messages import MEDIA_FAMILIES, DType, flow_of, qualify from fluksio.flow.nodes.wren import TRAIN_FLOW, WrenNode, pack_meta -from fluksio.flow.pipeline import ValidationIssue +from fluksio.flow.pipeline import EngineHeld, ValidationIssue from fluksio.flow.runs import RunRejected from fluksio.flow.schemas import ( NAME_PATTERN, @@ -1558,6 +1558,7 @@ FRAME_REFUSALS = { "not_scalar": "the port it names is not wired as a bool, int, float or str", "undeclared": "no flow declares its message", "refused": "its message refused the value", + "busy": "a rename holds the engine", } @@ -1604,6 +1605,8 @@ def ingest_frame(app: Any, device: DeviceDef, data: bytes) -> str: return "undeclared" except TypeError: return "refused" + except EngineHeld: + return "busy" return "" @@ -1657,6 +1660,8 @@ def ingest_publish(app: Any, device: DeviceDef, payload: dict[str, Any]) -> str: return "undeclared" except TypeError: return "refused" + except EngineHeld: + return "busy" return "" diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index bf1018d..7ca2e48 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -1858,6 +1858,9 @@ class FlowController: Refuses a name no flow declares: the flows own the namespace, and a message nothing reads or writes would just be a key nobody sees. + + :raises EngineHeld: While a rename holds the engine, for the caller to + answer: a 503, a device's ``busy``, a log line. """ if self.pipeline is None: raise KeyError(name) diff --git a/backend/fluksio/flow/executor.py b/backend/fluksio/flow/executor.py index 992fee5..89f15d0 100644 --- a/backend/fluksio/flow/executor.py +++ b/backend/fluksio/flow/executor.py @@ -17,7 +17,7 @@ from __future__ import annotations import logging import threading import time -from collections import deque +from collections import Counter, deque from collections.abc import Callable from concurrent.futures import Executor, Future, ThreadPoolExecutor from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar @@ -26,6 +26,7 @@ from fluksio.flow.queue import MAX_DELIVERIES, WorkItem, WorkQueue if TYPE_CHECKING: from fluksio.flow.events import EventBus + from fluksio.flow.namespace import Rename from fluksio.flow.pipeline import Pipeline logger = logging.getLogger(__name__) @@ -69,10 +70,41 @@ BACKLOG_INTERVAL_S = 5.0 BACKLOG_DEGRADED = 50 BACKLOG_STRIKES = 3 +#: How often an engine hold looks at its abort while it waits. +HOLD_POLL_S = 0.5 + _P = ParamSpec("_P") _T = TypeVar("_T") +def wait_out( + lock: threading.Condition, + busy: Callable[[], list[str]], + abort: threading.Event, + report: Callable[[list[str]], None] | None = None, +) -> bool: + """Wait until ``busy``, read under ``lock``, names nothing. + + Without a limit, because an engine hold waits for as long as the work + takes. ``report`` hears each new set of names, so whoever asked can show + what it waits for, and setting ``abort`` gives up with False. + """ + said: list[str] = [] + while True: + with lock: + names = busy() + while names and names == said and not abort.is_set(): + lock.wait(HOLD_POLL_S) + names = busy() + if abort.is_set(): + return False + if not names: + return True + said = names + if report is not None: + report(names) + + class _PoolLane(Executor): def __init__(self, pool: _ReservedPool, due: bool) -> None: self._pool = pool @@ -158,7 +190,8 @@ class ExecutionService: queue.on_delayed = self._timer_wake.set self._intake = threading.Event() self._intake.set() - self._inflight = 0 + # Cascades claimed and still running, by the node each came from. + self._running: Counter[str] = Counter() self._inflight_lock = threading.Condition() # Entry ids claimed and still running, under _inflight_lock. self._active: set[str] = set() @@ -174,6 +207,12 @@ class ExecutionService: ) self._consumer: threading.Thread | None = None self._timers: threading.Thread | None = None + # Held by the consumer for one pass of capacity check, claim and + # dispatch, and by the timer thread for one pass of promotion, touch + # and reclaim. `hold` takes both. Plain locks, because `release` may + # come from another thread than the one that held. + self._consumer_lock = threading.Lock() + self._timer_lock = threading.Lock() # Consecutive backlog readings over the threshold, and whether the last # of them said so out loud. self._backlog_strikes = 0 @@ -225,11 +264,11 @@ class ExecutionService: self._intake.clear() deadline = time.monotonic() + DRAIN_TIMEOUT_S with self._inflight_lock: - while self._inflight: + while self._running: remaining = deadline - time.monotonic() if remaining <= 0: logger.warning( - "Rebuild did not wait out %d cascades", self._inflight + "Rebuild did not wait out %d cascades", self._running.total() ) return self._inflight_lock.wait(remaining) @@ -237,13 +276,59 @@ class ExecutionService: def resume_intake(self) -> None: self._intake.set() + def hold( + self, + abort: threading.Event, + report: Callable[[list[str]], None] | None = None, + ) -> bool: + """Stop claiming, promoting, touching and reclaiming, once every + claimed cascade has finished. For a rename, which rewrites the queue. + + Unlike `pause_intake` this also waits out a claim already blocking + when intake clears, and a timer pass. It waits without a limit and + tells ``report`` the nodes whose cascades are still running each time + they change. Setting ``abort`` gives up: claiming resumes and it + returns False. True means held until `release`. + """ + self._intake.clear() + # So a consumer waiting for capacity sees the intake closed. + self._capacity_changed() + self._consumer_lock.acquire() + # The timer lock only after the drain: while a cascade runs, the + # timer thread keeps touching it, so it is not reclaimed as dropped. + if not wait_out(self._inflight_lock, self._busy, abort, report): + self._consumer_lock.release() + self._intake.set() + return False + self._timer_lock.acquire() + return True + + def release(self) -> None: + """End a `hold`.""" + self._timer_lock.release() + self._consumer_lock.release() + self._intake.set() + + def rename_reclaimed(self, rename: Rename) -> None: + """Move the names in reclaimed work still waiting to be run, which + the queue's own `rename` cannot reach. Under a `hold`.""" + with self._inflight_lock: + self._reclaimed = deque( + rename.item(item) or item for item in self._reclaimed + ) + + def _busy(self) -> list[str]: + """The nodes whose cascades are running. Under _inflight_lock.""" + return sorted(self._running) + def alive(self) -> bool: return self._consumer is not None and self._consumer.is_alive() @property def inflight(self) -> int: """Cascades claimed and still running.""" - return self._inflight + with self._inflight_lock: + return self._running.total() # ------------------------------------------------------------------------- # Threads @@ -255,40 +340,45 @@ class ExecutionService: if not self._intake.is_set(): self._intake.wait(timeout=0.5) continue - 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, - # Briefly, in the due-only case: this is the saturated - # engine, and a slot freeing has to be noticed promptly. - DUE_CLAIM_BLOCK_MS if due_only else CLAIM_BLOCK_MS, - due_only, - ) - failures = 0 - except Exception as exc: - failures += 1 - logger.error("Could not claim work: %s", exc) - self._publish_unavailable(exc) - # Backing off hard: a queue that is down stays down for a while. - self._stop.wait(min(30.0, 2.0**failures)) - continue + backoff = 0.0 + with self._consumer_lock: + 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, + # Briefly, in the due-only case: this is the saturated + # engine, and a slot freeing has to be noticed promptly. + DUE_CLAIM_BLOCK_MS if due_only else CLAIM_BLOCK_MS, + due_only, + ) + failures = 0 + except Exception as exc: + failures += 1 + logger.error("Could not claim work: %s", exc) + self._publish_unavailable(exc) + # Backing off hard: a queue that is down stays down for a + # while. Outside the lock, so a hold does not wait it out. + backoff = min(30.0, 2.0**failures) + items = [] - for item in items: - self._dispatch(item) + for item in items: + self._dispatch(item) + if backoff: + self._stop.wait(backoff) def _sleep_until_due(self) -> bool: """Wait for the soonest deadline, the housekeeping cap, or a new one. @@ -321,54 +411,56 @@ class ExecutionService: promote = self._sleep_until_due() if self._stop.is_set(): break - try: - if promote: - self.queue.move_due(time.time()) - except Exception as exc: - logger.error("Could not promote delayed work: %s", exc) - # The item is still due, so the wait above would be zero and - # 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: - last_backlog = now + # One pass, not the sleep above: a hold waits for it to end. + with self._timer_lock: try: - self._check_backlog() + if promote: + self.queue.move_due(time.time()) except Exception as exc: - logger.error("Could not read the queue backlog: %s", 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) - if now - last_touch >= TOUCH_INTERVAL_S: - last_touch = now - with self._inflight_lock: - running = list(self._active) - try: - self.queue.touch(running) - except Exception as exc: - logger.error("Could not touch claimed work: %s", exc) + now = time.monotonic() + if now - last_backlog >= BACKLOG_INTERVAL_S: + last_backlog = now + try: + self._check_backlog() + except Exception as exc: + logger.error("Could not read the queue backlog: %s", exc) - if now - last_reclaim < RECLAIM_INTERVAL_S: - continue - with self._inflight_lock: - if self._reclaimed or not self._intake.is_set(): + if now - last_touch >= TOUCH_INTERVAL_S: + last_touch = now + with self._inflight_lock: + running = list(self._active) + try: + self.queue.touch(running) + except Exception as exc: + logger.error("Could not touch claimed work: %s", exc) + + if now - last_reclaim < RECLAIM_INTERVAL_S: continue - last_reclaim = now - try: - recovered = self.queue.reclaim_stale(RECLAIM_IDLE_MS) - for item in recovered: - logger.info( - "Reclaimed work for '%s' (delivery %d)", - item.node, - item.deliveries, - ) 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) + if self._reclaimed or not self._intake.is_set(): + continue + last_reclaim = now + try: + recovered = self.queue.reclaim_stale(RECLAIM_IDLE_MS) + for item in recovered: + logger.info( + "Reclaimed work for '%s' (delivery %d)", + item.node, + item.deliveries, + ) + 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) def _check_backlog(self) -> None: """Say so when work has been waiting in the stream for a while. @@ -439,7 +531,7 @@ class ExecutionService: def _dispatch(self, item: WorkItem) -> None: with self._inflight_lock: - self._inflight += 1 + self._running[item.node] += 1 if item.entry_id: self._active.add(item.entry_id) try: @@ -455,7 +547,7 @@ class ExecutionService: def _done(self, item: WorkItem) -> None: with self._inflight_lock: - self._inflight -= 1 + self._running -= Counter((item.node,)) self._active.discard(item.entry_id) self._inflight_lock.notify_all() @@ -621,7 +713,7 @@ class ExecutionService: except Exception as exc: return {"error": str(exc), "consumer_alive": self.alive()} stats["consumer_alive"] = self.alive() - stats["cascades_busy"] = self._inflight + stats["cascades_busy"] = self.inflight stats["behind"] = self.behind return stats diff --git a/backend/fluksio/flow/nodes/device.py b/backend/fluksio/flow/nodes/device.py index 937d9d2..e2e7f23 100644 --- a/backend/fluksio/flow/nodes/device.py +++ b/backend/fluksio/flow/nodes/device.py @@ -23,7 +23,7 @@ from pydantic import BaseModel, Field from fluksio.flow.devices import PORT_PATTERN, presence from fluksio.flow.messages import MessageSpec from fluksio.flow.nodes.base import Node -from fluksio.flow.pipeline import ValueSource +from fluksio.flow.pipeline import EngineHeld, ValueSource if TYPE_CHECKING: from fastapi import FastAPI @@ -135,5 +135,9 @@ class DeviceNode(Node): ) try: pipeline.publish(flags, source) + except EngineHeld: + logger.warning( + "Kept the flags of '%s': a rename holds the engine", self.cfg.device + ) except Exception: logger.exception("Could not reset the flags of '%s'", self.cfg.device) diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 8eb2bfa..8cd66bd 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -19,7 +19,7 @@ import logging import threading import time import uuid -from collections import deque +from collections import Counter, deque from collections.abc import Callable, Iterator from concurrent.futures import Executor, Future, ThreadPoolExecutor from contextlib import contextmanager @@ -33,6 +33,7 @@ from pydantic import BaseModel, model_validator from fluksio.flow import logs from fluksio.flow.artifacts import is_reference from fluksio.flow.events import EventBus +from fluksio.flow.executor import wait_out from fluksio.flow.messages import flow_of, requalify from fluksio.flow.namespace import ( DELIVERED, @@ -265,6 +266,14 @@ def _derive( return produces, dependencies, consumes, by_id +class EngineHeld(RuntimeError): + """A value arrived while a rename holds the engine. + + Answered as a 503 with ``Retry-After``: nothing is wrong with the value, + and the same publish works once the rename is done. + """ + + class Pipeline: """Directed graph of nodes with automatic dependency resolution.""" @@ -286,6 +295,9 @@ class Pipeline: "_gate_lock", "_gate_revision", "_graph_lock", + "_holding", + "_inline", + "_inline_lock", "_queue", "_node_pool", "history_limits", @@ -324,6 +336,12 @@ class Pipeline: # `_gate_lock` or `state.lock()` in either order, which is what keeps # three locks from needing an ordering rule. self._graph_lock = threading.RLock() + # Set while a rename holds the engine (`hold`). What runs outside the + # queue, by node id or message name, is counted so a hold can wait + # it out. + self._holding = False + self._inline: Counter[str] = Counter() + self._inline_lock = threading.Condition() # An empty state backend is falsy, so this cannot be ``state or ...``: # that would quietly hand the pipeline a second, private state and # leave everyone reading the shared one seeing nothing. @@ -1325,6 +1343,49 @@ class Pipeline: with self._gate_lock: return flow in self._stepping + def hold( + self, + abort: threading.Event, + report: Callable[[list[str]], None] | None = None, + ) -> bool: + """Refuse `publish` and every cascade outside the queue, once those + already running have finished. For a rename, which nothing may write + past. + + Waits without a limit and tells ``report`` the nodes and messages + still running inline each time they change. Setting ``abort`` gives + up and lets them through again: False. True means held until + `release`. + """ + with self._inline_lock: + self._holding = True + if wait_out(self._inline_lock, lambda: sorted(self._inline), abort, report): + return True + self.release() + return False + + def release(self) -> None: + """End a `hold`.""" + with self._inline_lock: + self._holding = False + + @contextmanager + def _inline_run(self, names: list[str]) -> Iterator[None]: + """Count a run outside the queue, which a hold waits out. + + :raises EngineHeld: While held. + """ + with self._inline_lock: + if self._holding: + raise EngineHeld("A rename holds the engine; try again shortly") + self._inline.update(names) + try: + yield + finally: + with self._inline_lock: + self._inline -= Counter(names) + self._inline_lock.notify_all() + def _execute_parallel( self, nodes_subset: set[Node] | None, @@ -1625,44 +1686,45 @@ class Pipeline: already in state, so they are handed to their readers rather than written again. """ - run_id = f"{MANUAL_RUN_PREFIX}{uuid.uuid4().hex[:12]}" - self._publish( - { - "type": "cascade_started", - "run": run_id, - "flow": node.flow, - "node": node.id, - "cause": cause, - "deliveries": 1, - "ts": time.time(), - } - ) - try: - published = ( - set(outputs or {}) - if delivered - else self.apply_outputs(node, outputs, run_id) - ) - state = self.run_downstream( - node, - entry_id=run_id, - # No payload means the value is already in state and this is a - # wake-up, which has nothing to name as changed. - changed=published if outputs else None, - overrides=outputs if delivered else None, - ) - finally: - # Paired, or a cascade that raised leaves the run open until the - # abandoned sweep ten minutes later. + with self._inline_run([node.id]): + run_id = f"{MANUAL_RUN_PREFIX}{uuid.uuid4().hex[:12]}" self._publish( { - "type": "cascade_finished", + "type": "cascade_started", "run": run_id, "flow": node.flow, + "node": node.id, + "cause": cause, + "deliveries": 1, "ts": time.time(), } ) - return state + try: + published = ( + set(outputs or {}) + if delivered + else self.apply_outputs(node, outputs, run_id) + ) + state = self.run_downstream( + node, + entry_id=run_id, + # No payload means the value is already in state and this is a + # wake-up, which has nothing to name as changed. + changed=published if outputs else None, + overrides=outputs if delivered else None, + ) + finally: + # Paired, or a cascade that raised leaves the run open until the + # abandoned sweep ten minutes later. + self._publish( + { + "type": "cascade_finished", + "run": run_id, + "flow": node.flow, + "ts": time.time(), + } + ) + return state def publish( self, values: dict[str, Any], source: ValueSource | None = None @@ -1676,45 +1738,48 @@ class Pipeline: ``source`` says what did, so the canvas can show the value arriving from outside instead of blaming whichever node happens to be drawn as a producer of that message. + + :raises EngineHeld: While a rename holds the engine. """ if not values: return origin = source or ValueSource(kind="api", label="API") - ts = time.time() - self._state.record( - values, - {self._timestamp_key(name): ts for name in values}, - [self._version_key(name) for name in values], - ts, - self.history_limits, - ) - source_dump = origin.model_dump() - for name, value in values.items(): - self._publish( - { - "type": "message_value", - "flow": flow_of(name), - "name": name, - "value": value, - "ts": ts, - "source": source_dump, - } + with self._inline_run(list(values)): + ts = time.time() + self._state.record( + values, + {self._timestamp_key(name): ts for name in values}, + [self._version_key(name) for name in values], + ts, + self.history_limits, ) + source_dump = origin.model_dump() + for name, value in values.items(): + self._publish( + { + "type": "message_value", + "flow": flow_of(name), + "name": name, + "value": value, + "ts": ts, + "source": source_dump, + } + ) - targets: set[Node] = set() - for name in values: - for consumer in self.consumes.get(name, ()): - targets.add(consumer) - targets.update(self._get_downstream(consumer)) - if targets: - # Only what these values reach, as in a cascade: a node further - # down whose producer published nothing would otherwise run again - # on what it already read, once per microphone chunk. - self._execute_parallel( - targets, self._state, check_ready=True, changed=set(values) - ) + targets: set[Node] = set() + for name in values: + for consumer in self.consumes.get(name, ()): + targets.add(consumer) + targets.update(self._get_downstream(consumer)) + if targets: + # Only what these values reach, as in a cascade: a node further + # down whose producer published nothing would otherwise run again + # on what it already read, once per microphone chunk. + self._execute_parallel( + targets, self._state, check_ready=True, changed=set(values) + ) def defer( self, @@ -1785,8 +1850,14 @@ class Pipeline: "ts": time.time(), } ) - # Losing the value outright would be worse than running it here. - self._run_here(node, outputs, cause="external", delivered=kind == "emission") + # Losing the value outright would be worse than running it here, + # unless a rename holds the engine, which nothing runs past. + try: + self._run_here( + node, outputs, cause="external", delivered=kind == "emission" + ) + except EngineHeld: + logger.warning("Dropped work for '%s' while the engine is held", node.id) def value(self, name: str) -> dict[str, Any]: """One message and its timestamp, read together without listing state.""" diff --git a/backend/fluksio/flow/queue.py b/backend/fluksio/flow/queue.py index 19c02f4..6ad519c 100644 --- a/backend/fluksio/flow/queue.py +++ b/backend/fluksio/flow/queue.py @@ -756,12 +756,15 @@ class RedisWorkQueue(WorkQueue): self._held_key = f"{namespace}:__held__" self._move_due = self._redis.register_script(_MOVE_DUE) self._release = self._redis.register_script(_RELEASE) - # `add` reads the flag and writes under this, and `release` empties the - # held list and clears the flag under it. So nothing added after `hold` - # lands in a stream a rename is rebuilding, and nothing added while - # held overtakes held work. - self._hold_lock = threading.Lock() + # `add` reads the flag under this, and counts an `XADD` in `_adding` + # until it is done, so adds from parallel producers do not wait for + # each other. `hold` sets the flag and waits for `_adding` to reach 0, + # and `release` empties the held list and clears the flag under it. So + # nothing added after `hold` lands in a stream a rename is rebuilding, + # and nothing added while held overtakes held work. + self._hold_lock = threading.Condition() self._holding = False + self._adding = 0 self._ensure_group() self._drop_stale_consumers() @@ -811,12 +814,18 @@ class RedisWorkQueue(WorkQueue): if self._holding: self._redis.rpush(self._held_key, orjson.dumps(item.to_fields())) return + self._adding += 1 + try: self._redis.xadd( self._stream, cast(Any, item.to_fields()), maxlen=STREAM_MAXLEN, approximate=True, ) + finally: + with self._hold_lock: + self._adding -= 1 + self._hold_lock.notify_all() def add_delayed(self, item: WorkItem, not_before: float) -> None: item.not_before = not_before @@ -1154,6 +1163,8 @@ class RedisWorkQueue(WorkQueue): def hold(self) -> None: with self._hold_lock: self._holding = True + while self._adding: + self._hold_lock.wait() def release(self) -> None: with self._hold_lock: diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index 95e4af0..078e8b4 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -43,7 +43,7 @@ from fluksio.flow.executor import ExecutionService from fluksio.flow.identity import assign_ids from fluksio.flow.metrics import MetricsCollector from fluksio.flow.nodes.http import close_shared_client -from fluksio.flow.pipeline import ValueSource +from fluksio.flow.pipeline import EngineHeld, ValueSource from fluksio.flow.placement import Placer from fluksio.flow.plugins import load_plugins from fluksio.flow.provision import load_provisioners @@ -609,6 +609,14 @@ async def rebuild_busy(request: Request, exc: RebuildBusy) -> JSONResponse: # n return JSONResponse(status_code=503, content={"detail": str(exc)}) +@app.exception_handler(EngineHeld) +async def engine_held(request: Request, exc: EngineHeld) -> JSONResponse: # noqa: ARG001 + """A value published while a rename holds the engine, which ends shortly.""" + return JSONResponse( + status_code=503, content={"detail": str(exc)}, headers={"Retry-After": "5"} + ) + + # Tagged because the operation-id builder reads the first tag; the route # itself stays out of the schema. @app.get( diff --git a/backend/tests/api/routes/test_flows.py b/backend/tests/api/routes/test_flows.py index 9256ad7..e531f57 100644 --- a/backend/tests/api/routes/test_flows.py +++ b/backend/tests/api/routes/test_flows.py @@ -9,7 +9,9 @@ from sqlmodel import Session, select from fluksio.api.routes import flows as flows_route from fluksio.core.config import settings +from fluksio.flow.devices import Binding from fluksio.flow.events import event_bus +from fluksio.flow.pipeline import EngineHeld from fluksio.models import Run, RunArtifact, RunMetric, RunNode from tests.utils.flows import put_flow @@ -600,3 +602,35 @@ def test_a_shared_source_write_is_audited( headers=superuser_token_headers, ) client.delete(f"{PREFIX}/audited", headers=superuser_token_headers) + + +def test_a_publish_while_a_rename_holds_the_engine_is_refused( + client: TestClient, + superuser_token_headers: dict[str, str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Over REST as a 503 to retry, from a device's socket as ``busy``.""" + controller = client.app.state.flow_controller # type: ignore[attr-defined] + + def held(*_: object) -> None: + raise EngineHeld("A rename holds the engine; try again shortly") + + monkeypatch.setattr(controller, "publish_message", held) + response = client.post( + f"{settings.API_V1_STR}/messages/porch.doorbell", + headers=superuser_token_headers, + json={"value": True}, + ) + assert response.status_code == 503 + assert response.headers["Retry-After"] == "5" + + bound = Binding("panel", "porch", "panel", "gpio.4", "porch.doorbell", "in", "bool") + app = SimpleNamespace( + state=SimpleNamespace( + device_store=SimpleNamespace(bound=lambda _: [bound]), + flow_controller=controller, + ) + ) + device = SimpleNamespace(id="panel", title="Panel") + value = {"type": "publish", "name": "porch.doorbell", "value": True} + assert flows_route.ingest_publish(app, device, value) == "busy" # type: ignore[arg-type] diff --git a/backend/tests/flow/test_engine_hold.py b/backend/tests/flow/test_engine_hold.py new file mode 100644 index 0000000..2f6761f --- /dev/null +++ b/backend/tests/flow/test_engine_hold.py @@ -0,0 +1,272 @@ +"""Engine holds for a rename (section 2.2 of docs/private/id_refac_rename.md). + +The pipeline refuses what it would run outside the queue, and the execution +service stops claiming and its timers once claimed work has finished. Both +wait without a limit and can be aborted. +""" + +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.namespace import Rename +from fluksio.flow.nodes import Node +from fluksio.flow.pipeline import EngineHeld, Pipeline +from fluksio.flow.queue import MemoryWorkQueue, WorkItem +from fluksio.flow.state import MemoryState + + +def _engine(consume, queue=None) -> tuple[Pipeline, ExecutionService, Node]: + """A source whose readings ``consume`` gets, and a service draining the + queue one cascade at a time, so order shows.""" + + def consumer_body(reading, params): + consume(reading) + + source = Node( + f=lambda params: None, + provides=[MessageSpec(name="reading", dtype=DType.FLOAT)], + name="source", + ) + consumer = Node( + f=consumer_body, + requires=[MessageSpec(name="reading", dtype=DType.FLOAT)], + name="consumer", + ) + source.assign_flow("f", "source") + consumer.assign_flow("f", "consumer") + queue = queue or MemoryWorkQueue() + service = ExecutionService(queue, max_cascades=1) + pipeline = Pipeline( + nodes=[source, consumer], + state=MemoryState(), + work_queue=queue, + node_pool=service.node_pool, + ) + service.bind(pipeline) + return pipeline, service, source + + +def _in_thread(fn, *args) -> tuple[threading.Thread, list]: + result: list = [] + thread = threading.Thread(target=lambda: result.append(fn(*args)), daemon=True) + thread.start() + return thread, result + + +def _until(check, timeout: float = 5.0) -> None: + deadline = time.monotonic() + timeout + while not check(): + assert time.monotonic() < deadline, "timed out" + time.sleep(0.02) + + +def _blocking(seen: list) -> tuple[threading.Event, threading.Event, object]: + """A consumer that records and then waits until told to finish.""" + running, finish = threading.Event(), threading.Event() + + def consume(reading): + seen.append(reading) + running.set() + finish.wait(5) + + return running, finish, consume + + +def test_a_hold_waits_out_a_running_cascade_without_a_limit(monkeypatch): + # What a rebuild's drain gives up after, which a hold does not. + monkeypatch.setattr(executor, "DRAIN_TIMEOUT_S", 0.05) + seen: list = [] + running, finish, consume = _blocking(seen) + _, service, source = _engine(consume) + service.start() + try: + source.inject({"reading": 1.0}) + assert running.wait(5) + busy: list = [] + thread, held = _in_thread(service.hold, threading.Event(), busy.append) + thread.join(0.3) + assert thread.is_alive() + assert busy == [["f.source"]] + + source.inject({"reading": 2.0}) + finish.set() + thread.join(5) + assert held == [True] + time.sleep(0.2) + assert seen == [1.0] + + service.release() + _until(lambda: seen == [1.0, 2.0]) + finally: + finish.set() + service.stop() + + +class SlowClaim(MemoryWorkQueue): + """A queue whose claim is still under way when a hold begins.""" + + def __init__(self) -> None: + super().__init__() + self.claiming = threading.Event() + self.go = threading.Event() + + def claim(self, count, block_ms, due_only=False): + items = super().claim(count, block_ms, due_only) + if items: + self.claiming.set() + self.go.wait(5) + return items + + +def test_a_hold_waits_out_a_claim_already_under_way(): + """`pause_intake` returns here and the item runs after it.""" + seen: list = [] + queue = SlowClaim() + _, service, source = _engine(seen.append, queue) + service.start() + try: + source.inject({"reading": 1.0}) + assert queue.claiming.wait(5) + thread, held = _in_thread(service.hold, threading.Event()) + thread.join(0.2) + assert thread.is_alive() + + queue.go.set() + thread.join(5) + assert held == [True] + assert seen == [1.0] + assert service.inflight == 0 + service.release() + finally: + queue.go.set() + service.stop() + + +def test_an_aborted_hold_leaves_the_engine_as_it_was(): + seen: list = [] + running, finish, consume = _blocking(seen) + pipeline, service, source = _engine(consume) + service.start() + try: + source.inject({"reading": 1.0}) + assert running.wait(5) + abort = threading.Event() + thread, held = _in_thread(service.hold, abort) + thread.join(0.2) + assert thread.is_alive() + + abort.set() + thread.join(5) + assert held == [False] + finish.set() + + # Claiming and promotion carry on. + assert pipeline.defer(source, {"f.reading": 2.0}, 0.05) + source.inject({"reading": 3.0}) + _until(lambda: sorted(seen) == [1.0, 2.0, 3.0]) + assert not service.queue.held + finally: + finish.set() + service.stop() + + +def test_an_aborted_pipeline_hold_lets_publishes_through_again(): + seen: list = [] + running, finish, consume = _blocking(seen) + pipeline, _, _ = _engine(consume) + publishing, _ = _in_thread(pipeline.publish, {"f.reading": 1.0}) + try: + assert running.wait(5) + busy: list = [] + abort = threading.Event() + thread, held = _in_thread(pipeline.hold, abort, busy.append) + thread.join(0.2) + assert thread.is_alive() + assert busy == [["f.reading"]] + + abort.set() + thread.join(5) + assert held == [False] + finish.set() + publishing.join(5) + pipeline.publish({"f.reading": 2.0}) + assert seen == [1.0, 2.0] + finally: + finish.set() + + +def test_publish_is_refused_while_held(): + seen: list = [] + pipeline, _, _ = _engine(seen.append) + assert pipeline.hold(threading.Event()) + + with pytest.raises(EngineHeld): + pipeline.publish({"f.reading": 1.0}) + assert "f.reading" not in pipeline.state + + pipeline.release() + pipeline.publish({"f.reading": 2.0}) + assert seen == [2.0] + + +def test_work_arriving_during_a_hold_runs_after_it_in_order(): + seen: list = [] + pipeline, service, source = _engine(seen.append) + service.start() + try: + abort = threading.Event() + assert pipeline.hold(abort) + assert service.hold(abort) + source.inject({"reading": 1.0}) + service.queue.hold() + source.inject({"reading": 2.0}) + source.inject({"reading": 3.0}) + time.sleep(0.2) + assert seen == [] + + service.queue.release() + service.release() + pipeline.release() + _until(lambda: seen == [1.0, 2.0, 3.0]) + finally: + service.stop() + + +def test_a_queue_failure_while_held_drops_the_value_rather_than_run_it( + monkeypatch, +): + seen: list = [] + pipeline, service, source = _engine(seen.append) + + def down(item): + raise ConnectionError("queue down") + + monkeypatch.setattr(service.queue, "add", down) + assert pipeline.hold(threading.Event()) + source.inject({"reading": 1.0}) + assert seen == [] + + pipeline.release() + source.inject({"reading": 2.0}) + assert seen == [2.0] + + +def test_reclaimed_work_waiting_to_run_follows_a_rename(): + service = ExecutionService(MemoryWorkQueue()) + service._reclaimed.extend( + [ + WorkItem(kind="cascade", node="heat.valve", flow="heat", entry_id="1-0"), + WorkItem(kind="cascade", node="heating.pump", flow="heating"), + ] + ) + service.rename_reclaimed(Rename("heat", "warm")) + assert [(i.node, i.flow) for i in service._reclaimed] == [ + ("warm.valve", "warm"), + ("heating.pump", "heating"), + ] + assert service._reclaimed[0].entry_id == "1-0" diff --git a/backend/tests/flow/test_rename_runtime.py b/backend/tests/flow/test_rename_runtime.py index 6ad5b87..aa5e281 100644 --- a/backend/tests/flow/test_rename_runtime.py +++ b/backend/tests/flow/test_rename_runtime.py @@ -3,6 +3,7 @@ and 3 of docs/private/id_refac_rename.md). Memory always, Redis with FLUKSIO_TEST_REDIS_HOST (and optional PORT).""" import os +import threading import uuid import pytest @@ -218,6 +219,36 @@ def test_held_work_is_released_in_order_behind_older_work(queue): ] +def test_a_hold_waits_for_an_add_already_writing_to_the_stream( + redis_queue, monkeypatch +): + """Adds do not wait for each other, but a hold waits for them, so no + entry lands in a stream after the rebuild has read it.""" + writing, finish = threading.Event(), threading.Event() + xadd = redis_queue._redis.xadd + + def slow(*args, **kwargs): + writing.set() + finish.wait(5) + return xadd(*args, **kwargs) + + monkeypatch.setattr(redis_queue._redis, "xadd", slow) + adding = threading.Thread(target=redis_queue.add, args=(item("heat.a"),)) + adding.start() + assert writing.wait(5) + holding = threading.Thread(target=redis_queue.hold) + holding.start() + holding.join(0.2) + assert holding.is_alive() + + finish.set() + holding.join(5) + adding.join(5) + assert redis_queue.count(HEAT)["queue_items"] == 1 + assert redis_queue.held + redis_queue.release() + + def test_a_guarded_timer_still_finds_its_guard(queue): state = MemoryState() state["__node__:heat.limit:generation"] = 3