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

@@ -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")
 

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:

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:

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:

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.
 

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:

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()

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")

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,

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.
 

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.