From 8dcceecd95dd7bbb3c2230bd5de99afcd9a84dd2 Mon Sep 17 00:00:00 2001 From: stroblme Date: Wed, 11 Nov 2026 12:27:33 +0100 Subject: [PATCH] A local call waiting for a slot is reached by a stop, a delete or a cancel and ends without running; what a call emits after its flow was stopped is dropped as its result is; a node run by hand whose flow is stopped or deleted tells its caller but publishes no node_error; cancelling a node also ends its calls on workers --- backend/fluksio/api/routes/flows.py | 13 +++-- backend/fluksio/flow/controller.py | 11 +++- backend/fluksio/flow/nodes/base.py | 19 +++++-- backend/fluksio/flow/pipeline.py | 9 +++- backend/fluksio/flow/remote.py | 5 ++ backend/fluksio/flow/workers.py | 63 +++++++++++++++++------ backend/tests/flow/test_remote.py | 80 ++++++++++++++++++++++++++++- backend/tests/flow/test_workers.py | 34 ++++++++++++ docs/code/workers.md | 11 ++-- docs/concepts/flows.md | 6 ++- frontend/src/client/sdk.gen.ts | 3 +- 11 files changed, 219 insertions(+), 35 deletions(-) diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index a7fb662..b6f9531 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -1141,14 +1141,19 @@ async def trigger_node( @router.post("/{flow_id}/nodes/{node_id}/cancel", response_model=Message) def cancel_node(flow_id: str, node_id: str, controller: FlowControllerDep) -> Any: - """Stop a node that is running right now, by killing the worker running it. + """Stop a node that is running right now, here or on an attached worker, + by killing the process running it. Idempotent on purpose: by the time a click reaches here the node may well have finished, and that is the outcome that was asked for. """ - name = _read_flow(controller, flow_id).name - pool = controller.workers - if pool is not None and pool.cancel(f"{name}.{node_id}"): + node = f"{_read_flow(controller, flow_id).name}.{node_id}" + pool, remote = controller.workers, controller.remote + stopped = pool is not None and pool.cancel(node) + # Not ``or``: a node may have calls in both places. + if remote is not None and remote.cancel_node(node): + stopped = True + if stopped: return Message(message=f"Stopped '{node_id}'") return Message(message=f"'{node_id}' was not running") diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 30c706c..2914a57 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -2113,6 +2113,8 @@ class FlowController: node = self.get_node(node_id) if node is None: raise KeyError(node_id) + pipeline = self.pipeline + ended = pipeline.ended(node.flow) if pipeline is not None else 0 try: if node.requires and values: node.trigger(values, durable=False) @@ -2121,9 +2123,14 @@ class FlowController: except EngineHeld: raise except Exception as exc: - if self.pipeline is None: + if pipeline is None: raise - return self.pipeline.publish_error(node, exc) + if pipeline.ended(node.flow) != ended: + # Its flow was stopped or deleted while it ran, most likely + # ending the call. The person who clicked is told, alert rules + # are not, as for a cascade. + return f"{type(exc).__name__}: {exc}" + return pipeline.publish_error(node, exc) return "" def _publish(self, event: dict[str, Any]) -> None: diff --git a/backend/fluksio/flow/nodes/base.py b/backend/fluksio/flow/nodes/base.py index 42a0972..f4df421 100644 --- a/backend/fluksio/flow/nodes/base.py +++ b/backend/fluksio/flow/nodes/base.py @@ -43,6 +43,11 @@ RESERVED_SETTINGS = frozenset({"synchronous"}) #: worker request carries it to ``fluksio.arrived()`` inside the node. ARRIVED: ContextVar[list[str] | None] = ContextVar("arrived", default=None) +#: What ``Pipeline.ended`` read for the running node's flow when its call +#: began. What the call emits once that has moved is dropped, as is what it +#: returns. +STARTED: ContextVar[int | None] = ContextVar("started", default=None) + class Node: """ @@ -466,8 +471,12 @@ class Node: if not outputs and not self.requires: ended = self._pipeline.ended(self.flow) collected = logs.Collector() - with logs.capture(collected): - outputs = self.f(params=self.params) or {} + token = STARTED.set(ended) + try: + with logs.capture(collected): + outputs = self.f(params=self.params) or {} + finally: + STARTED.reset(token) if self._pipeline is None or self._pipeline.ended(self.flow) != ended: # Its flow was stopped or deleted while it ran. return None @@ -492,7 +501,11 @@ class Node: :rtype: dict | None """ ended = self._pipeline.ended(self.flow) if self._pipeline is not None else 0 - outputs = self.execute(inputs) + token = STARTED.set(ended) + try: + outputs = self.execute(inputs) + finally: + STARTED.reset(token) if self._pipeline is None: return outputs if self._pipeline.ended(self.flow) != ended: diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 15978f6..8eaa30a 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -44,7 +44,7 @@ from fluksio.flow.namespace import ( VERSION, ) from fluksio.flow.nodes import Node -from fluksio.flow.nodes.base import ARRIVED +from fluksio.flow.nodes.base import ARRIVED, STARTED from fluksio.flow.queue import WorkQueue from fluksio.flow.state import MemoryState, StateBackend @@ -1143,10 +1143,12 @@ class Pipeline: return restored or {} token = ARRIVED.set(arrived) + began = STARTED.set(ended) try: with logs.capture(collected): result = node.execute(inputs) finally: + STARTED.reset(began) ARRIVED.reset(token) if self.ended(node.flow) != ended: # Its flow was stopped or deleted while it ran (`end_running`). @@ -1289,6 +1291,11 @@ class Pipeline: three thousand mid-node cascades would leave "the run has finished" with no meaning. """ + started = STARTED.get() + if started is not None and self.ended(node.flow) != started: + # Its flow was stopped or deleted after the call began + # (`end_running`), so this is dropped as its result will be. + return self._observe_emission(node, outputs) passed, stale = self._throttled(node, outputs) if not passed: diff --git a/backend/fluksio/flow/remote.py b/backend/fluksio/flow/remote.py index df6bdce..8246fd0 100644 --- a/backend/fluksio/flow/remote.py +++ b/backend/fluksio/flow/remote.py @@ -545,6 +545,11 @@ class RemoteWorkerHub: """Kill whatever this run has executing on any attached worker.""" return sum(worker.end_calls(f"{run_id}:") for worker in self.workers()) + def cancel_node(self, node_id: str) -> int: + """End what a live node has running on any attached worker, its calls + being ``flow.node:…``.""" + return sum(worker.end_calls(f"{node_id}:") for worker in self.workers()) + def cancel_flow(self, flow: str) -> None: """End what a live flow has running on any attached worker. diff --git a/backend/fluksio/flow/workers.py b/backend/fluksio/flow/workers.py index 65baada..b18418c 100644 --- a/backend/fluksio/flow/workers.py +++ b/backend/fluksio/flow/workers.py @@ -278,17 +278,21 @@ class PythonWorkerPool: # runs at once, and cancelling one of them must not kill the others. # The live pipeline's own executions carry an empty run. self._running: dict[tuple[str, str], _Worker] = {} + # Calls still waiting for a slot, each by its (run, node). A cancel + # marks one, and it ends once it has a slot instead of running. + self._queued: dict[threading.Event, tuple[str, str]] = {} self._generation = 0 self._stopped = False self._requests = itertools.count(1) # Not a lock over the pool — the slot queue is what serialises calls. - # It covers the two places where bookkeeping must not interleave: a - # cancel taking a worker out of ``_running`` while ``_request`` is - # handing that same worker back to a slot, and two respawns draining - # the idle queue at once. Registering into ``_running``, the generation - # counter and the ``_stopped`` flag are deliberately outside it — each - # is one atomic operation, and a cancel that misses a node by a - # microsecond is a cancel that arrived a microsecond early. + # It covers the places where bookkeeping must not interleave: a cancel + # taking a worker out of ``_running`` while ``_request`` is handing + # that same worker back to a slot, a cancel looking for a call while it + # moves from ``_queued`` to ``_running``, and two respawns draining the + # idle queue at once. The generation counter and the ``_stopped`` flag + # are deliberately outside it — each is one atomic operation, and a + # cancel that misses a node by a microsecond is a cancel that arrived a + # microsecond early. self._lock = threading.Lock() # ------------------------------------------------------------------------- @@ -444,10 +448,27 @@ class PythonWorkerPool: run_id: str = "", on_event: Callable[[dict[str, Any]], None] | None = None, ) -> dict[str, Any]: - worker = self._acquire() key = (run_id, node_id) + # Registered before it waits for a slot, so a cancel reaches a call + # queued behind a busy one, and it ends there without running. + queued = threading.Event() + if node_id: + with self._lock: + self._queued[queued] = key + try: + worker = self._acquire() + except BaseException: + with self._lock: + self._queued.pop(queued, None) + raise + with self._lock: + self._queued.pop(queued, None) + if queued.is_set(): + self._release(worker) + raise NodeCancelled("cancelled while it waited for a worker") + if node_id: + self._running[key] = worker if node_id: - self._running[key] = worker self._publish( { "type": "node_started", @@ -679,20 +700,25 @@ class PythonWorkerPool: return call def cancel(self, node_id: str, run_id: str = "") -> bool: - """Stop a node that is running now. False when there was nothing to stop.""" + """Stop a node that is running now, and its calls waiting for a slot. + False when there was nothing to stop.""" + key = (run_id, node_id) with self._lock: - worker = self._running.get((run_id, node_id)) + worker = self._running.get(key) + queued = [call for call, owner in self._queued.items() if owner == key] + for call in queued: + call.set() children = list(self._children.values()) if worker is not None: worker.cancelled = True worker.kill() - return True # A node with declared resources runs in a pool of its own, and it is # the same node to whoever pressed cancel. - return any(child.cancel(node_id, run_id) for child in children) + stopped = [child.cancel(node_id, run_id) for child in children] + return worker is not None or bool(queued) or any(stopped) def cancel_run(self, run_id: str) -> int: - """Stop every node this run has in a worker right now.""" + """Stop every node this run has in a worker right now or waiting for one.""" with self._lock: workers = [ worker @@ -702,8 +728,15 @@ class PythonWorkerPool: for worker in workers: worker.cancelled = True worker.kill() + queued = [call for call, key in self._queued.items() if key[0] == run_id] + for call in queued: + call.set() children = list(self._children.values()) - return len(workers) + sum(child.cancel_run(run_id) for child in children) + return ( + len(workers) + + len(queued) + + sum(child.cancel_run(run_id) for child in children) + ) def _publish(self, event: dict[str, Any]) -> None: if self.events is not None: diff --git a/backend/tests/flow/test_remote.py b/backend/tests/flow/test_remote.py index 05d9a8e..7a867fa 100644 --- a/backend/tests/flow/test_remote.py +++ b/backend/tests/flow/test_remote.py @@ -17,6 +17,7 @@ from collections.abc import Iterator import pytest +from fluksio.api.routes.flows import cancel_node from fluksio.flow import remote from fluksio.flow.controller import FlowController from fluksio.flow.messages import DType, MessageSpec @@ -263,7 +264,7 @@ class Loading(FakeSocket): self.worker.deliver({"call_id": payload["call_id"], "ok": True}) -def live_voice(loop, tmp_path, pool: PythonWorkerPool, local: str = ""): +def live_voice(loop, tmp_path, pool: PythonWorkerPool, local: str = "", events=None): """A live flow 'voice' whose node 'gpu' runs on a worker and, given *local*, a node 'cpu' running that source in *pool*. Built and started.""" store = FlowStore(tmp_path / "flows") @@ -287,12 +288,23 @@ def live_voice(loop, tmp_path, pool: PythonWorkerPool, local: str = ""): ) hub = RemoteWorkerHub() hub.attach(socket.worker) - controller = FlowController(store, workers=pool, remote=hub) + controller = FlowController(store, workers=pool, remote=hub, events=events) asyncio.run(controller.reload()) socket.arrived.clear() return controller, store, socket +class Events(list): + """Stands in for the event bus: keeps what was published.""" + + def publish(self, event: dict) -> None: + self.append(event) + + +#: What a worker answers for a call it was told to cancel. +CANCELLED = {"ok": False, "error": {"type": "NodeCancelled", "message": "cancelled"}} + + def test_deleting_a_flow_ends_its_call_and_drops_a_late_answer(loop, tmp_path): """A node that never returns kept its worker slot past its flow's delete, and an answer that came anyway put the deleted flow's value back.""" @@ -366,6 +378,70 @@ def test_stopping_a_flow_ends_its_calls_here_and_on_workers(loop, tmp_path): pool.stop() +def test_a_node_run_by_hand_and_stopped_tells_the_caller_alone(loop, tmp_path): + """Stopping a flow failed the node a person had started by hand as an + error, which the alert rules fire on.""" + pool = PythonWorkerPool(python=sys.executable, size=1) + events = Events() + controller, _, socket = live_voice(loop, tmp_path, pool, events=events) + + answer: list[str] = [] + running = call_in_thread( + lambda: answer.append(controller.trigger_node("voice.gpu")) + ) + assert socket.arrived.wait(5) + call = socket.sent[-1]["call_id"] + socket.arrived.clear() + asyncio.run(controller.set_enabled("voice", "voice", False)) + assert socket.arrived.wait(5) + socket.worker.deliver({"call_id": call, **CANCELLED}) + running.join(timeout=5) + + assert answer == ["NodeCancelled: cancelled"] + assert [event for event in events if event["type"] == "node_error"] == [] + + +def test_what_a_call_emits_after_its_flow_stopped_is_dropped(loop, tmp_path): + """An emission between a flow's stop and its call's end was still written, + though what the call returned no longer was.""" + pool = PythonWorkerPool(python=sys.executable, size=1) + controller, _, socket = live_voice(loop, tmp_path, pool) + + running = call_in_thread(lambda: controller.run_flow("voice")) + assert socket.arrived.wait(5) + call = socket.sent[-1]["call_id"] + socket.arrived.clear() + asyncio.run(controller.set_enabled("voice", "voice", False)) + assert socket.arrived.wait(5) + socket.worker.deliver( + {"call_id": call, "event": "emit", "outputs": {"reply": "late"}} + ) + socket.worker.deliver({"call_id": call, **CANCELLED}) + running.join(timeout=5) + + assert not running.is_alive() + assert "voice.reply" not in controller.state + + +def test_stopping_a_node_reaches_its_call_on_a_worker(loop, tmp_path): + """The canvas's stop ended only a call in the local pool.""" + pool = PythonWorkerPool(python=sys.executable, size=1) + controller, _, socket = live_voice(loop, tmp_path, pool) + + running = call_in_thread(lambda: controller.run_flow("voice")) + assert socket.arrived.wait(5) + call = socket.sent[-1]["call_id"] + socket.arrived.clear() + assert cancel_node("voice", "gpu", controller).message == "Stopped 'gpu'" + assert socket.arrived.wait(5) + assert socket.sent[-1] == {"op": "cancel", "call_id": call} + + socket.worker.deliver({"call_id": call, **CANCELLED}) + running.join(timeout=5) + assert not running.is_alive() + assert socket.worker.in_flight == 0 + + def test_a_service_and_a_compile_leave_the_only_slot_to_calls(loop): """A service holding the slot for good would leave the worker blocked.""" hub = RemoteWorkerHub() diff --git a/backend/tests/flow/test_workers.py b/backend/tests/flow/test_workers.py index af98175..a66c96f 100644 --- a/backend/tests/flow/test_workers.py +++ b/backend/tests/flow/test_workers.py @@ -176,6 +176,40 @@ def test_a_running_node_can_be_cancelled(pool): stopper.join() +def test_a_call_waiting_for_a_slot_is_cancelled_before_it_runs(pool): + """A cancel missed a call queued for a slot, so it took the slot that + cancelling the node ahead of it freed, and ran.""" + outcome: dict[str, object] = {} + + def call(node: str, code: str) -> None: + try: + outcome[node] = pool.run("demo", node, code, {}, f"demo.{node}", 30) + except NodeCancelled as exc: + outcome[node] = exc + + sleep = "import time\n\n\ndef process():\n time.sleep(30)\n" + ahead = threading.Thread(target=call, args=("ahead", sleep), daemon=True) + ahead.start() + deadline = time.monotonic() + 10 + while ("", "demo.ahead") not in pool._running: + assert time.monotonic() < deadline, "the first call never started" + time.sleep(0.01) + queued = threading.Thread( + target=call, args=("queued", "def process():\n return 1\n"), daemon=True + ) + queued.start() + deadline = time.monotonic() + 5 + while not pool.cancel("demo.queued"): + assert time.monotonic() < deadline, "the queued call was not found" + time.sleep(0.01) + + assert pool.cancel("demo.ahead") + ahead.join(timeout=10) + queued.join(timeout=10) + assert isinstance(outcome["queued"], NodeCancelled) + assert "waited" in str(outcome["queued"]) + + def test_a_result_that_is_not_json_is_refused(pool): with pytest.raises(Exception, match="cannot be sent back as JSON"): run(pool, "def process():\n return {'out': {1, 2}}\n") diff --git a/docs/code/workers.md b/docs/code/workers.md index 3149c89..0ea25a8 100644 --- a/docs/code/workers.md +++ b/docs/code/workers.md @@ -96,11 +96,12 @@ Source code travels with each call and compiles on the worker, so remote-only dependencies do not need to be installed on the engine. Calls to `emit`, `save_artifact` and `load_artifact` work there too. -Cancelling a run stops its active node processes. Stopping or deleting a flow -does the same for the calls it has out, and drops an answer that still comes -back. A disconnected worker fails in-flight calls. While executing, the agent -sends heartbeats every ten seconds; 90 seconds without contact fails the call. -Heartbeats do not reset the node's own silence timeout. +Cancelling a run or a single node stops its active node processes. Stopping +or deleting a flow does the same for the calls it has out, and drops an answer +or emission that still comes back. A disconnected worker fails in-flight +calls. While executing, the agent sends heartbeats every ten seconds; 90 +seconds without contact fails the call. Heartbeats do not reset the node's own +silence timeout. Every call runs in a process group of its own. The agent kills the group when the engine ends the call, when the connection drops and when the agent stops, diff --git a/docs/concepts/flows.md b/docs/concepts/flows.md index f08f620..ae6c009 100644 --- a/docs/concepts/flows.md +++ b/docs/concepts/flows.md @@ -168,8 +168,10 @@ Build issues prevent the affected flow from running. `node_unhealthy` and the node recovers. A degraded node is a warning on Home rather than a fault. You can **stop** a live flow to remove its subscriptions and schedules and end -the node calls it has running, or **pause** it to hold pending work while -values still arrive. What a stopped call still returns is dropped. **Step** +the node calls it has running or queued, or **pause** it to hold pending work +while values still arrive. What a stopped call still emits or returns is +dropped. A node you started by hand reports the cancel to you and is not +marked as failed. **Step** releases one held item. Repeated background crashes can quarantine a flow; publishing a change gives it another chance. diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index fcfa1e9..8344273 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -1964,7 +1964,8 @@ export class FlowsService { /** * Cancel Node - * Stop a node that is running right now, by killing the worker running it. + * Stop a node that is running right now, here or on an attached worker, + * by killing the process running it. * * Idempotent on purpose: by the time a click reaches here the node may well * have finished, and that is the outcome that was asked for.