Stopping a live flow ends the calls it has running, on workers and in the local pool, and deleting one now ends its local calls too; what a call returns after its flow was stopped or deleted is dropped by a per-flow count the node compares, so a flow started again never takes an old call's answer. A live remote call's id carries a suffix, so two calls of one node no longer share it, and a call is registered before it queues for a slot, so ending its flow or run reaches it and it is never sent; a run's cancel goes through the call's own thread as well

backend/fluksio/flow/controller.py

@@ -699,7 +699,9 @@ class FlowController:
             else:
                 self.disabled = self.disabled | {flow}
                 self.pipeline.set_disabled(self.disabled)
+                self.pipeline.end_running(flow)
                 await self._teardown(flow)
+                await self._end_calls(flow)
         finally:
             self._lock.release()
 
@@ -862,10 +864,7 @@ class FlowController:
             if built is None:
                 if name:
                     pipeline.remove_flow(name)
-                    # A call still out on a worker would keep its slot and its
-                    # process there, for a result that is now dropped.
-                    if self.remote is not None:
-                        self.remote.cancel_flow(name)
+                    await self._end_calls(name)
                 self.loaded = {
                     node_id: entry
                     for node_id, entry in self.loaded.items()
@@ -1142,6 +1141,26 @@ class FlowController:
         else:
             await self.supervisor.cancel_flow(flow)
 
+    async def _end_calls(self, flow: str) -> None:
+        """End the calls a stopped or deleted flow has running, here and on
+        workers. A call left out there would keep its slot and its process for
+        a result that is dropped (`Pipeline.end_running`)."""
+        if self.remote is not None:
+            self.remote.cancel_flow(flow)
+        pool = self.workers
+        if pool is None:
+            return
+        nodes = [
+            node_id for node_id, entry in self.loaded.items() if entry.flow == flow
+        ]
+
+        def cancel() -> None:
+            for node_id in nodes:
+                pool.cancel(node_id)
+
+        # Off the loop: a killed worker is waited for.
+        await run_in_threadpool(cancel)
+
     async def _activate(self, flow: str | None = None) -> None:
         """Start subscriptions, schedules and webhooks of the new pipeline.
 

backend/fluksio/flow/nodes/base.py

@@ -464,11 +464,12 @@ class Node:
 
         # A source node asked to inject nothing produces its own data.
         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 {}
-            if self._pipeline is None:
-                # Its flow was deleted while it ran (`Pipeline.remove_flow`).
+            if self._pipeline is None or self._pipeline.ended(self.flow) != ended:
+                # Its flow was stopped or deleted while it ran.
                 return None
             self._pipeline.publish_log(self, collected, "")
 
@@ -490,7 +491,11 @@ class Node:
         :returns: Node outputs if successful, or pipeline execution results if bound.
         :rtype: dict | None
         """
+        ended = self._pipeline.ended(self.flow) if self._pipeline is not None else 0
         outputs = self.execute(inputs)
         if self._pipeline is None:
             return outputs
+        if self._pipeline.ended(self.flow) != ended:
+            # Its flow was stopped while it ran.
+            return None
         return self._pipeline.trigger(self, outputs, durable=durable)

backend/fluksio/flow/pipeline.py

@@ -308,6 +308,7 @@ class Pipeline:
         "emission_observer",
         "run_cache",
         "run_id",
+        "_ended",
     )
 
     def __init__(
@@ -373,6 +374,9 @@ class Pipeline:
         # How deep to keep each message's series; a chart asking for more
         # than the default puts its message in here. Swapped, never mutated.
         self.history_limits: dict[str, int] = {}
+        # How often each flow had what it was running ended (`end_running`).
+        # Swapped, never mutated.
+        self._ended: dict[str, int] = {}
 
         (
             self.produces,
@@ -493,6 +497,22 @@ class Pipeline:
         self.replace_flow(flow, [])
         for node in gone:
             node.bind(None)
+        self.end_running(flow)
+
+    def end_running(self, flow: str) -> None:
+        """Drop what this flow's nodes return from what they are running now.
+
+        For a flow that was stopped or deleted. Started again, its nodes run
+        as before, and a call from before still answering is not taken for
+        theirs.
+        """
+        with self._gate_lock:
+            self._ended = {**self._ended, flow: self._ended.get(flow, 0) + 1}
+
+    def ended(self, flow: str) -> int:
+        """How often *flow* had what it was running ended: a node compares it
+        before and after it runs."""
+        return self._ended.get(flow, 0)
 
     def _seed(self, initial_values: dict[str, Any] | None) -> None:
         """Give messages a starting value, without overwriting one already there.
@@ -1093,6 +1113,7 @@ class Pipeline:
         """
         started = time.perf_counter()
         collected = logs.Collector()
+        ended = self.ended(node.flow)
         try:
             # One MGET. The lock this used to be read under bought nothing a
             # single bulk read does not, and it was the engine's one global
@@ -1127,8 +1148,8 @@ class Pipeline:
                     result = node.execute(inputs)
             finally:
                 ARRIVED.reset(token)
-            if node._pipeline is not self:
-                # Its flow was deleted while it ran (`remove_flow`).
+            if self.ended(node.flow) != ended:
+                # Its flow was stopped or deleted while it ran (`end_running`).
                 return None
             self.publish_log(node, collected, "")
 
@@ -1191,9 +1212,9 @@ class Pipeline:
                 )
             return result or {}
         except Exception as exc:
-            if node._pipeline is not self:
-                # Deleted while it ran, and most likely failing because the
-                # call was ended for that: nobody is left to tell.
+            if self.ended(node.flow) != ended:
+                # Stopped or deleted while it ran, and most likely failing
+                # because the call was ended for that.
                 return None
             # One failing node must not take the rest of the graph down.
             error = self.publish_error(node, exc, collected, entry_id)

backend/fluksio/flow/remote.py

@@ -27,13 +27,19 @@ import queue
 import sys
 import threading
 import time
+import uuid
 from collections.abc import Callable
 from dataclasses import dataclass
 from typing import Any
 
 from fluksio.flow.nodes.base import ARRIVED, NodeOutputError
 from fluksio.flow.resources import ResourceAccountant
-from fluksio.flow.workers import NodeTimeout, RemoteError, _remote_class
+from fluksio.flow.workers import (
+    NodeCancelled,
+    NodeTimeout,
+    RemoteError,
+    _remote_class,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -189,13 +195,19 @@ class RemoteWorker:
         call_id = str(payload["call_id"])
         if inbox is None:
             inbox = queue.Queue()
-        # Blocking here is the backpressure, exactly as taking a slot is in the
-        # local pool.
-        if slot:
-            self._slots.acquire()
+        # Registered before it queues for a slot, so ending its flow or its run
+        # reaches a call still waiting for one.
         with self._lock:
             self._pending[call_id] = inbox
         try:
+            # Blocking here is the backpressure, exactly as taking a slot is in
+            # the local pool.
+            if slot:
+                self._slots.acquire()
+            if not inbox.empty():
+                # Ended, or the worker gone, while it waited: nothing went out,
+                # so there is nothing on the worker to cancel.
+                raise NodeCancelled(f"ended before '{self.name}' was sent the call")
             future = asyncio.run_coroutine_threadsafe(self._send(payload), self._loop)
             future.result(timeout=SEND_TIMEOUT_S)
 
@@ -271,12 +283,9 @@ class RemoteWorker:
         with self._lock:
             return len(self._pending)
 
-    def calls_of(self, run_id: str) -> list[str]:
-        with self._lock:
-            return [call for call in self._pending if call.startswith(f"{run_id}:")]
-
-    def end_calls(self, prefix: str) -> None:
-        """End every call whose id starts with *prefix*, each once it is sent."""
+    def end_calls(self, prefix: str) -> int:
+        """End every call whose id starts with *prefix*, each once it is sent,
+        and one still waiting for a slot before it is."""
         with self._lock:
             inboxes = [
                 inbox
@@ -285,6 +294,7 @@ class RemoteWorker:
             ]
         for inbox in inboxes:
             inbox.put(self.END)
+        return len(inboxes)
 
     def cancel(self, call_id: str) -> None:
         """Ask the worker to kill what it is running for this call."""
@@ -423,7 +433,11 @@ class RemoteWorkerHub:
         """
         payload = {
             "op": "run",
-            "call_id": f"{run_id}:{node_id}" if run_id else node_id,
+            # A live node can run in two cascades at once, so its calls are
+            # told apart; a run runs each node once.
+            "call_id": (
+                f"{run_id}:{node_id}" if run_id else f"{node_id}:{uuid.uuid4().hex[:8]}"
+            ),
             "flow": flow,
             "node": node,
             "source": source,
@@ -529,18 +543,14 @@ class RemoteWorkerHub:
 
     def cancel_run(self, run_id: str) -> int:
         """Kill whatever this run has executing on any attached worker."""
-        stopped = 0
-        for worker in self.workers():
-            for call in worker.calls_of(run_id):
-                worker.cancel(call)
-                stopped += 1
-        return stopped
+        return sum(worker.end_calls(f"{run_id}:") for worker in self.workers())
 
     def cancel_flow(self, flow: str) -> None:
         """End what a live flow has running on any attached worker.
 
-        A live call goes by its node's id, ``flow.node``. A run's carries the
-        run's id first and a compile its op, so neither is touched.
+        A live call goes by its node's id and a suffix, ``flow.node:…``. A
+        run's carries the run's id first and a compile its op, so neither is
+        touched.
         """
         for worker in self.workers():
             worker.end_calls(f"{flow}.")

backend/tests/flow/test_remote.py

@@ -244,49 +244,65 @@ def test_cancelling_a_run_reaches_only_that_run(loop):
     call_in_thread(lambda: call("run-b"))
     assert socket.arrived.wait(5)
 
+    socket.arrived.clear()
     assert hub.cancel_run("run-a") == 1
+    # Sent by the call's own thread, once its frame is out.
+    assert socket.arrived.wait(5)
     cancels = [frame for frame in socket.sent if frame.get("op") == "cancel"]
     assert [frame["call_id"] for frame in cancels] == ["run-a:flow.node"]
 
 
+class Loading(FakeSocket):
+    """A worker that loads whatever it is sent and leaves each call running."""
+
+    worker: RemoteWorker
+
+    async def send_json(self, payload: dict) -> None:
+        await super().send_json(payload)
+        if payload["op"] == "compile":
+            self.worker.deliver({"call_id": payload["call_id"], "ok": True})
+
+
+def live_voice(loop, tmp_path, pool: PythonWorkerPool, local: str = ""):
+    """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")
+    nodes = [
+        NodeDef(
+            id="gpu",
+            worker="gpu",
+            provides=[MessageSpec(name="reply", dtype=DType.STR)],
+        )
+    ]
+    if local:
+        nodes.append(
+            NodeDef(id="cpu", provides=[MessageSpec(name="local", dtype=DType.STR)])
+        )
+        store.write_node_source("voice", "cpu", local)
+    store.write_flow(FlowDef(name="voice", nodes=nodes))
+    store.write_node_source("voice", "gpu", "def process():\n    return None\n")
+    socket = Loading()
+    socket.worker = RemoteWorker(
+        name="gpu1", labels=["gpu"], send=socket.send_json, loop=loop, max_parallel=2
+    )
+    hub = RemoteWorkerHub()
+    hub.attach(socket.worker)
+    controller = FlowController(store, workers=pool, remote=hub)
+    asyncio.run(controller.reload())
+    socket.arrived.clear()
+    return controller, store, socket
+
+
 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."""
-    store = FlowStore(tmp_path / "flows")
-    store.write_flow(
-        FlowDef(
-            name="voice",
-            nodes=[
-                NodeDef(
-                    id="gpu",
-                    worker="gpu",
-                    provides=[MessageSpec(name="reply", dtype=DType.STR)],
-                )
-            ],
-        )
-    )
-    store.write_node_source("voice", "gpu", "def process():\n    return None\n")
+    pool = PythonWorkerPool(python=sys.executable, size=1)
+    controller, store, socket = live_voice(loop, tmp_path, pool)
+    worker = socket.worker
 
-    class Agent(FakeSocket):
-        """Loads whatever it is sent, and leaves a call running until told."""
-
-        async def send_json(self, payload: dict) -> None:
-            await super().send_json(payload)
-            if payload["op"] == "compile":
-                worker.deliver({"call_id": payload["call_id"], "ok": True})
-
-    socket = Agent()
-    worker = RemoteWorker(name="gpu1", labels=["gpu"], send=socket.send_json, loop=loop)
-    hub = RemoteWorkerHub()
-    hub.attach(worker)
-    controller = FlowController(
-        store, workers=PythonWorkerPool(python=sys.executable, size=1), remote=hub
-    )
-    asyncio.run(controller.reload())
-
-    socket.arrived.clear()
     running = call_in_thread(lambda: controller.run_flow("voice"))
     assert socket.arrived.wait(5)
+    call = socket.sent[-1]["call_id"]
     assert socket.sent[-1]["op"] == "run"
 
     socket.arrived.clear()
@@ -294,15 +310,62 @@ def test_deleting_a_flow_ends_its_call_and_drops_a_late_answer(loop, tmp_path):
     asyncio.run(controller.reload_flow("voice"))
     controller.forget_flow(key)
     assert socket.arrived.wait(5)
-    assert socket.sent[-1] == {"op": "cancel", "call_id": "voice.gpu"}
+    assert socket.sent[-1] == {"op": "cancel", "call_id": call}
 
-    worker.deliver({"call_id": "voice.gpu", "ok": True, "result": {"reply": "late"}})
+    worker.deliver({"call_id": call, "ok": True, "result": {"reply": "late"}})
     running.join(timeout=5)
     assert not running.is_alive()
     assert worker.in_flight == 0
     assert "voice.reply" not in controller.state
 
 
+def test_stopping_a_flow_ends_its_calls_here_and_on_workers(loop, tmp_path):
+    """A stopped flow's calls ran on, here and on the worker, and what they
+    returned was written to its state, or, once it was started again, taken
+    for a new call's: a live call's id was its node's."""
+    pool = PythonWorkerPool(python=sys.executable, size=1)
+    pool.start()
+    local = "import time\n\n\ndef process():\n    time.sleep(60)\n    return None\n"
+    controller, _, socket = live_voice(loop, tmp_path, pool, local=local)
+    worker = socket.worker
+    try:
+        running = call_in_thread(lambda: controller.run_flow("voice"))
+        assert socket.arrived.wait(5)
+        old = socket.sent[-1]["call_id"]
+        deadline = time.monotonic() + 10
+        while ("", "voice.cpu") not in pool._running:
+            assert time.monotonic() < deadline, "the local node never started"
+            time.sleep(0.01)
+
+        socket.arrived.clear()
+        asyncio.run(controller.set_enabled("voice", "voice", False))
+        assert socket.arrived.wait(5)
+        assert socket.sent[-1] == {"op": "cancel", "call_id": old}
+        deadline = time.monotonic() + 5
+        while ("", "voice.cpu") in pool._running:
+            assert time.monotonic() < deadline, "the local node was left running"
+            time.sleep(0.01)
+
+        # Started again before the worker has answered the old call.
+        asyncio.run(controller.set_enabled("voice", "voice", True))
+        socket.arrived.clear()
+        gpu = controller.get_node("voice.gpu")
+        assert gpu is not None
+        again = call_in_thread(lambda: controller.pipeline.run({}, nodes={gpu}))
+        assert socket.arrived.wait(5)
+        new = socket.sent[-1]["call_id"]
+
+        worker.deliver({"call_id": old, "ok": True, "result": {"reply": "old"}})
+        running.join(timeout=5)
+        assert not running.is_alive()
+        assert "voice.reply" not in controller.state
+        worker.deliver({"call_id": new, "ok": True, "result": {"reply": "new"}})
+        again.join(timeout=5)
+        assert controller.state.get("voice.reply") == "new"
+    finally:
+        pool.stop()
+
+
 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_service_node.py

@@ -180,10 +180,11 @@ def test_a_service_that_never_gets_healthy_reads_down(monkeypatch):
     asyncio.run(scenario())
 
 
-def test_a_stop_right_after_the_start_ends_the_serve_after_it(monkeypatch):
+def test_a_stop_right_after_the_start_leaves_no_server_running(monkeypatch):
     """The stop's cancel went out on a thread of its own, so with the serve's
     thread slow to start it reached the worker first. A worker ignores a cancel
     for a call it does not know, and ran that server until its socket dropped.
+    Ended before its thread has sent it, the serve now never goes out.
     """
     monkeypatch.setattr(remote, "SILENCE_S", 1.0)
     monkeypatch.setattr(service, "STOP_WAIT_S", 1.0)
@@ -228,4 +229,4 @@ def test_a_stop_right_after_the_start_ends_the_serve_after_it(monkeypatch):
         await node.stop()
 
     asyncio.run(scenario())
-    assert [frame["op"] for frame in agent.sent] == ["serve", "cancel"]
+    assert agent.sent == []

docs/code/workers.md

@@ -96,10 +96,10 @@ 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. 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.
+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.
 
 Every call runs in a process group of its own. The agent kills the group when

docs/concepts/flows.md

@@ -167,10 +167,11 @@ Build issues prevent the affected flow from running. `node_unhealthy` and
 `node_degraded` are runtime issues: other work continues, and they clear when
 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, or
-**pause** it to hold pending work while values still arrive. **Step** releases
-one held item. Repeated background crashes can quarantine a flow; publishing
-a change gives it another chance.
+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**
+releases one held item. Repeated background crashes can quarantine a flow;
+publishing a change gives it another chance.
 
 ## Where to next