From 93b4a9a0b1b7a00c331d4c6d8d432d28df121c44 Mon Sep 17 00:00:00 2001 From: stroblme Date: Sun, 16 Aug 2026 16:44:17 +0200 Subject: [PATCH] Step a paused flow, and deliver what a rate limit held back Three things a pause and an interval were quietly losing: - A rebuild builds a fresh pipeline, so nothing is paused any more and no resume ever comes for what the old one parked. Release it on rebuild. - POST /flows/{name}/step takes the oldest parked item and runs that one wave while the flow stays paused, so a held-back cascade can be walked through. Nothing parked answers plainly rather than failing. - A per-port interval was leading-edge only: a producer going quiet inside the window left the consumer on the value before it. The held value is kept and a flush item scheduled on the queue's existing timer, so the window ends with a delivery. One timer in flight per node, and none without a queue to run it. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01H7LwYgJfpkbLCTeiAf8U4A --- backend/app/api/routes/flows.py | 10 ++ backend/app/flow/controller.py | 24 +++- backend/app/flow/executor.py | 26 +++- backend/app/flow/pipeline.py | 139 +++++++++++++++++++-- backend/app/flow/queue.py | 15 ++- backend/tests/flow/test_discretization.py | 66 ++++++++++ backend/tests/flow/test_queue.py | 30 +++++ backend/tests/flow/test_runtime_control.py | 64 ++++++++++ 8 files changed, 354 insertions(+), 20 deletions(-) diff --git a/backend/app/api/routes/flows.py b/backend/app/api/routes/flows.py index e17fb34..4922bf6 100644 --- a/backend/app/api/routes/flows.py +++ b/backend/app/api/routes/flows.py @@ -541,6 +541,16 @@ async def resume_flow(name: str, controller: FlowControllerDep) -> Any: return Message(message=f"Resumed flow '{name}'") +@router.post("/{name}/step", response_model=Message) +async def step_flow(name: str, controller: FlowControllerDep) -> Any: + """Run one message a pause is holding back, leaving the flow paused.""" + _read_flow(controller, name) + node = await run_in_threadpool(controller.step_flow, name) + if node is None: + return Message(message=f"Nothing held back in flow '{name}'") + return Message(message=f"Stepped '{node}'") + + # ----------------------------------------------------------------------------- # Validation and execution # ----------------------------------------------------------------------------- diff --git a/backend/app/flow/controller.py b/backend/app/flow/controller.py index 4c162b7..4242f43 100644 --- a/backend/app/flow/controller.py +++ b/backend/app/flow/controller.py @@ -352,6 +352,10 @@ class FlowController: if self.execution is not None: self.execution.resume_intake() + # A rebuild clears the pause, so no resume will ever come for + # what the old pipeline parked. Release it here or it is lost. + for flow in published: + self._release_parked(flow.name) self._publish( { @@ -597,11 +601,21 @@ class FlowController: if self.pipeline is None: return self.pipeline.resume(flow) - if self.execution is not None: - # Whatever arrived while the flow was held is queued again, oldest - # first, so a pause loses nothing. - for item in self.execution.queue.unpark(flow): - self.execution.queue.add(item) + self._release_parked(flow) + + def _release_parked(self, flow: str) -> None: + """Queue what a pause held back again, oldest first, so it is not lost.""" + if self.execution is None: + return + for item in self.execution.queue.unpark(flow): + self.execution.queue.add(item) + + def step_flow(self, flow: str) -> str | None: + """Run one held-back item, leaving the flow paused. Blocking. + + Returns the node it came from, or None when nothing is held back. + """ + return self.execution.step(flow) if self.execution is not None else None def set_history_limits(self, limits: dict[str, int]) -> None: """How deep to keep each charted message's series. Applies at once.""" diff --git a/backend/app/flow/executor.py b/backend/app/flow/executor.py index df9a88a..ece465c 100644 --- a/backend/app/flow/executor.py +++ b/backend/app/flow/executor.py @@ -187,6 +187,25 @@ class ExecutionService: # Handling one item # ------------------------------------------------------------------------- + def step(self, flow: str) -> str | None: + """Run one item a pause is holding, and hold everything else still. + + Blocking, so the caller sees the wave finish. Returns the node the item + came from, or None when nothing is parked for this flow. + """ + pipeline = self._pipeline + if pipeline is None: + return None + item = self.queue.unpark_one(flow) + if item is None: + return None + with pipeline.stepping(flow): + try: + self._run_item(item) + except Exception: + logger.exception("Step of '%s' failed", item.node) + return item.node + def _handle(self, item: WorkItem) -> None: handled = True try: @@ -235,10 +254,15 @@ class ExecutionService: if pipeline.is_disabled(node.flow): return True - if pipeline.is_paused(node.flow): + if pipeline.is_paused(node.flow) and not pipeline.is_stepping(node.flow): self.queue.park(node.flow, item) return True + if item.kind == "flush": + # A rate-limit window ended; nothing to replay, only to let out. + pipeline.flush(node) + return True + if item.guard_key and str(node.recall(item.guard_key, "")) != item.guard_value: # The node moved on while this waited — a restarted timer, say. logger.debug("Guard no longer holds for '%s', dropped", item.node) diff --git a/backend/app/flow/pipeline.py b/backend/app/flow/pipeline.py index 5460c3a..6465b02 100644 --- a/backend/app/flow/pipeline.py +++ b/backend/app/flow/pipeline.py @@ -12,7 +12,9 @@ import logging import threading import time from collections import deque +from collections.abc import Iterator from concurrent.futures import Future, ThreadPoolExecutor, wait +from contextlib import contextmanager from typing import Any, Literal from pydantic import BaseModel @@ -84,6 +86,7 @@ class Pipeline: "_downstream_cache", "_disabled", "_paused", + "_stepping", "_gate_lock", "_queue", "_node_pool", @@ -106,6 +109,7 @@ class Pipeline: # debugging state that a rebuild is meant to clear. self._disabled = frozenset(disabled_flows or ()) self._paused: set[str] = set() + self._stepping: set[str] = set() self._gate_lock = threading.Lock() # An empty state backend is falsy, so this cannot be ``state or ...``: # that would quietly hand the pipeline a second, private state and @@ -327,11 +331,21 @@ class Pipeline: """When a rate-limited input last woke this node.""" return f"__in_ts__:{node_name}:{msg_name}" - def _throttled(self, node: Node, result: dict[str, Any]) -> dict[str, Any]: - """Drop the outputs whose port is not due to publish yet. + def _held_key(self, msg_name: str) -> str: + """The value a rate-limited output kept back, waiting for its window.""" + return f"__held__:{msg_name}" - The value is not lost — the port publishes the current one next time it - is due. Nothing declaring an interval means nothing to look up. + def _flush_key(self, node_name: str) -> str: + """When a rate-limit window of this node is due to be let through.""" + return f"__flush__:{node_name}" + + def _throttled(self, node: Node, result: dict[str, Any]) -> dict[str, Any]: + """Hold back the outputs whose port is not due to publish yet. + + The value is not lost: it is kept and published when the window ends, + so a producer that goes quiet still delivers its last reading rather + than leaving the consumer on the one before it. Nothing declaring an + interval means nothing to look up. """ limited = { name: spec.interval @@ -342,13 +356,79 @@ class Pipeline: return result now = time.time() - stamps = self._state.get_multi([self._timestamp_key(name) for name in limited]) - return { - name: value - for name, value in result.items() - if name not in limited - or now - (stamps.get(self._timestamp_key(name)) or 0) >= limited[name] + flush_key = self._flush_key(node.id) + stamps = self._state.get_multi( + [self._timestamp_key(name) for name in limited] + [flush_key] + ) + + passed: dict[str, Any] = {} + held: dict[str, Any] = {} + due_at = 0.0 + for name, value in result.items(): + if name not in limited: + passed[name] = value + continue + window_ends = (stamps.get(self._timestamp_key(name)) or 0) + limited[name] + if now >= window_ends: + passed[name] = value + else: + held[name] = value + due_at = min(due_at or window_ends, window_ends) + + if self._queue is not None: + # Without a queue there is no timer to let the value out later, so + # holding it would only mean losing it more slowly. + for name in limited: + if name in passed: + # A fresh publish makes anything held for that port stale. + self._state.delete(self._held_key(name)) + if held: + self._state.update( + {self._held_key(name): value for name, value in held.items()} + ) + self._schedule_flush(node, due_at, now, stamps.get(flush_key)) + return passed + + def _schedule_flush( + self, node: Node, at: float, now: float, pending: float | None + ) -> None: + """Come back to this node when its rate-limit window ends. + + One timer in flight per node: a pending one that is early enough is + left alone, and one that turns out to be too late simply finds nothing + to do when it fires. + """ + if self._queue is None or (pending and now < pending <= at): + return + self._state[self._flush_key(node.id)] = at + self.defer(node, {}, at - now, kind="flush") + + def flush(self, node: Node) -> 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 + published, and a node whose inputs were all held back gets its run. + """ + limited_out = [name for name, s in node.provides.items() if s.interval > 0] + stored = ( + self._state.get_multi([self._held_key(name) for name in limited_out]) + if limited_out + else {} + ) + held = { + name: stored[self._held_key(name)] + for name in limited_out + if stored.get(self._held_key(name)) is not None } + if held: + # Publishing runs the limit again, which is what clears the hold — + # or puts the timer back, if this fired a hair early. + self.apply_outputs(node, held) + self.run_downstream(node) + if any(spec.interval > 0 for spec in node.requires.values()): + self._execute_parallel( + {node, *self._get_downstream(node)}, self._state, check_ready=True + ) def _increment_message_versions(self, outputs: dict[str, Any]) -> None: for msg_name in outputs: @@ -411,8 +491,9 @@ class Pipeline: return True now = time.time() + flush_key = self._flush_key(node.id) keys = [self._delivered_key(node.id, name) for name in limited] - stamps = self._state.get_multi(keys) + stamps = self._state.get_multi(keys + [flush_key]) due = [ name for name, interval in limited.items() @@ -420,6 +501,17 @@ class Pipeline: ] # An unlimited input alongside a held-back one still wakes the node. if not due and len(limited) == len(node.requires): + # The value is in state, only the wake-up is held back. Come back + # for it, so a producer going quiet does not strand it there. + self._schedule_flush( + node, + min( + (stamps.get(self._delivered_key(node.id, name)) or 0) + interval + for name, interval in limited.items() + ), + now, + stamps.get(flush_key), + ) return False with self._state.lock(): @@ -585,17 +677,37 @@ class Pipeline: # Whatever was held back is free to run now. self._execute_parallel(self.flow_nodes(flow), self._state, check_ready=True) + @contextmanager + def stepping(self, flow: str) -> Iterator[None]: + """Let one held-back wave through without ending the pause. + + The flow stays paused throughout, so everything else arriving for it + keeps piling up where the next step can find it. + """ + with self._gate_lock: + self._stepping.add(flow) + try: + yield + finally: + with self._gate_lock: + self._stepping.discard(flow) + def _gate_blocks(self, node: Node) -> bool: """Is this node's flow held back from executing?""" if node.flow in self._disabled: return True with self._gate_lock: - return node.flow in self._paused + return node.flow in self._paused and node.flow not in self._stepping def is_paused(self, flow: str) -> bool: with self._gate_lock: return flow in self._paused + def is_stepping(self, flow: str) -> bool: + """Is a single step of this paused flow running right now?""" + with self._gate_lock: + return flow in self._stepping + def _execute_parallel( self, nodes_subset: set[Node] | None, @@ -838,6 +950,7 @@ class Pipeline: outputs: dict[str, Any], seconds: float, guard: tuple[str, Any] | None = None, + kind: str = "cascade", ) -> bool: """Publish a node's outputs later, without holding a worker thread. @@ -853,7 +966,7 @@ class Pipeline: if self._queue is None or seconds <= 0: return False item = WorkItem( - kind="cascade", + kind=kind, node=node.id, flow=node.flow, outputs=outputs, diff --git a/backend/app/flow/queue.py b/backend/app/flow/queue.py index 84c6a38..c245864 100644 --- a/backend/app/flow/queue.py +++ b/backend/app/flow/queue.py @@ -40,7 +40,7 @@ class WorkItem: """One unit of journaled work. :param kind: ``cascade`` replays a node's outputs and runs what is - downstream; ``node`` executes exactly one node. + downstream; ``flush`` lets out what a node's rate limits held back. :param node: The node the item is about — the source for a cascade, the target for a node item. :param flow: The flow that node belongs to, so gating needs no lookup. @@ -135,6 +135,10 @@ class WorkQueue(ABC): def unpark(self, flow: str) -> list[WorkItem]: """Return a paused flow's held items to the queue, oldest first.""" + @abstractmethod + def unpark_one(self, flow: str) -> WorkItem | None: + """Take the oldest held item, leaving the rest parked. That is a step.""" + @abstractmethod def clear_flow(self, flow: str) -> None: """Forget anything held for a flow that no longer exists.""" @@ -227,6 +231,11 @@ class MemoryWorkQueue(WorkQueue): with self._lock: return self._parked.pop(flow, []) + def unpark_one(self, flow: str) -> WorkItem | None: + with self._lock: + held = self._parked.get(flow) + return held.pop(0) if held else None + def clear_flow(self, flow: str) -> None: with self._lock: self._parked.pop(flow, None) @@ -387,6 +396,10 @@ class RedisWorkQueue(WorkQueue): self._redis.delete(key) return [WorkItem.from_fields(json.loads(r), "") for r in raw] + def unpark_one(self, flow: str) -> WorkItem | None: + raw = cast("str | None", self._redis.lpop(self._parked_key(flow))) + return WorkItem.from_fields(json.loads(raw), "") if raw else None + def clear_flow(self, flow: str) -> None: self._redis.delete(self._parked_key(flow)) diff --git a/backend/tests/flow/test_discretization.py b/backend/tests/flow/test_discretization.py index 86e98c6..d61366c 100644 --- a/backend/tests/flow/test_discretization.py +++ b/backend/tests/flow/test_discretization.py @@ -1,8 +1,15 @@ """Per-port intervals: deliver at most every x seconds.""" +import time + +from app.flow.executor import ExecutionService from app.flow.messages import DType, MessageSpec from app.flow.nodes import Node from app.flow.pipeline import Pipeline +from app.flow.queue import MemoryWorkQueue + +# Short enough to wait out in a test, long enough not to race the engine. +WINDOW = 0.05 def spec(name: str, interval: float = 0) -> MessageSpec: @@ -82,6 +89,65 @@ def test_an_unthrottled_input_still_wakes_a_node_beside_a_throttled_one(): assert [quick for quick, _ in seen] == [1.0, 2.0] +def _running(nodes: list[Node]) -> tuple[Pipeline, MemoryWorkQueue, ExecutionService]: + """A pipeline with the timer the engine uses for held-back values.""" + queue = MemoryWorkQueue() + pipeline = Pipeline(nodes=nodes, work_queue=queue) + service = ExecutionService(queue) + service.bind(pipeline) + return pipeline, queue, service + + +def _run_due(queue: MemoryWorkQueue, service: ExecutionService) -> int: + """What the engine's timer thread does once the window has passed.""" + moved = queue.move_due(time.time()) + for item in queue.claim(10, 10): + service._run_item(item) + return moved + + +def test_a_limited_output_publishes_its_last_value_when_the_window_ends(): + """A producer going quiet must not strand the reading it held back.""" + readings = iter([1.0, 2.0]) + source = make_node( + "source", + lambda params: {"temp": next(readings)}, + provides=[spec("temp", interval=WINDOW)], + ) + pipeline, queue, service = _running([source]) + + pipeline.run({}) + pipeline.run({}) + # Inside the window, so the second reading is held rather than published. + assert pipeline.state["demo.temp"] == 1.0 + + time.sleep(WINDOW * 2) + assert _run_due(queue, service) == 1 + + assert pipeline.state["demo.temp"] == 2.0 + + +def test_a_limited_input_wakes_its_node_when_the_window_ends(): + seen: list[float] = [] + source = make_node("source", lambda params: None, provides=[spec("temp")]) + consumer = make_node( + "consumer", + lambda temp, params: seen.append(temp), + requires=[spec("temp", interval=WINDOW)], + ) + _pipeline, queue, service = _running([source, consumer]) + + source.inject({"temp": 20.0}, durable=False) + source.inject({"temp": 21.0}, durable=False) + assert seen == [20.0] + + time.sleep(WINDOW * 2) + assert _run_due(queue, service) == 1 + + # The value that arrived inside the window is delivered at the end of it. + assert seen == [20.0, 21.0] + + def test_a_manual_run_is_never_throttled_on_its_inputs(): seen: list[float] = [] source = make_node("source", lambda params: {"temp": 20.0}, provides=[spec("temp")]) diff --git a/backend/tests/flow/test_queue.py b/backend/tests/flow/test_queue.py index e33df6a..b5e1460 100644 --- a/backend/tests/flow/test_queue.py +++ b/backend/tests/flow/test_queue.py @@ -127,6 +127,36 @@ def test_work_for_a_paused_flow_is_held_and_released_on_resume(): assert seen == [1.0] +def test_a_step_runs_one_held_item_and_leaves_the_flow_paused(): + pipeline, source, _state, seen = _pipeline_with_a_consumer() + service = ExecutionService(pipeline._queue) + service.bind(pipeline) + + pipeline.pause("f") + source.inject({"reading": 1.0}) + source.inject({"reading": 2.0}) + for item in pipeline._queue.claim(10, 10): + service._run_item(item) + assert seen == [] + + assert service.step("f") == "f.source" + + assert seen == [1.0] + # Still paused, and the second value is still waiting for the next step. + assert pipeline.is_paused("f") + assert [i.outputs for i in pipeline._queue.unpark("f")] == [{"f.reading": 2.0}] + + +def test_stepping_a_flow_with_nothing_held_says_so_rather_than_failing(): + pipeline, _source, _state, _seen = _pipeline_with_a_consumer() + service = ExecutionService(pipeline._queue) + service.bind(pipeline) + + pipeline.pause("f") + + assert service.step("f") is None + + def test_work_for_a_stopped_flow_is_dropped(): pipeline, source, state, seen = _pipeline_with_a_consumer() stopped = Pipeline( diff --git a/backend/tests/flow/test_runtime_control.py b/backend/tests/flow/test_runtime_control.py index 4bb40e0..6014509 100644 --- a/backend/tests/flow/test_runtime_control.py +++ b/backend/tests/flow/test_runtime_control.py @@ -1,10 +1,17 @@ """Stopping a flow takes it off the engine; pausing holds its nodes.""" from pathlib import Path +from typing import Any +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from app.api.deps import get_current_user +from app.api.routes.flows import router from app.flow.messages import DType, MessageSpec from app.flow.nodes import Node from app.flow.pipeline import Pipeline +from app.flow.schemas import FlowDef from app.flow.store import FlowStore @@ -101,3 +108,60 @@ def test_stopped_survives_a_restart(tmp_path: Path): # A second store over the same directory is what a restart looks like. assert FlowStore(tmp_path / "flows").read_enabled("heating") is False + + +class _StubController: + """Only what the step route touches: the store, and the step itself.""" + + def __init__(self, store: FlowStore, stepped: str | None) -> None: + self.store = store + self.stepped = stepped + self.calls: list[str] = [] + + def step_flow(self, flow: str) -> str | None: + self.calls.append(flow) + return self.stepped + + +def _stub(tmp_path: Path, stepped: str | None) -> _StubController: + """A controller over a store holding one flow, called 'heating'.""" + store = FlowStore(tmp_path / "flows") + store.write_flow(FlowDef(name="heating")) + return _StubController(store, stepped) + + +def _client(controller: Any) -> TestClient: + app = FastAPI() + app.include_router(router, prefix="/api/v1") + app.state.flow_controller = controller + app.dependency_overrides[get_current_user] = lambda: None + return TestClient(app) + + +def test_the_step_endpoint_reports_what_it_ran(tmp_path: Path): + controller = _stub(tmp_path, "heating.sensor") + + response = _client(controller).post("/api/v1/flows/heating/step") + + assert response.status_code == 200 + assert response.json() == {"message": "Stepped 'heating.sensor'"} + assert controller.calls == ["heating"] + + +def test_stepping_a_flow_with_nothing_parked_answers_plainly(tmp_path: Path): + """A button pressed once too often is not an error.""" + controller = _stub(tmp_path, None) + + response = _client(controller).post("/api/v1/flows/heating/step") + + assert response.status_code == 200 + assert response.json() == {"message": "Nothing held back in flow 'heating'"} + + +def test_stepping_an_unknown_flow_is_a_404(tmp_path: Path): + controller = _stub(tmp_path, None) + + response = _client(controller).post("/api/v1/flows/nosuchflow/step") + + assert response.status_code == 404 + assert controller.calls == []