Deleting a live flow ends the calls it has out on workers and drops what a removed node still returns, so a node that never returns gives its slot and process back and a late answer no longer recreates the flow's values; a cancel goes through the call's own inbox and is sent after the call's frame, so a service stopped right after starting no longer cancels before its serve

backend/fluksio/flow/controller.py

@@ -862,6 +862,10 @@ 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)
                 self.loaded = {
                     node_id: entry
                     for node_id, entry in self.loaded.items()

backend/fluksio/flow/nodes/base.py

@@ -264,15 +264,16 @@ class Node:
         if self._on_health is not None:
             self._on_health(self, status, detail)
 
-    def bind(self, pipeline: Pipeline) -> None:
+    def bind(self, pipeline: Pipeline | None) -> None:
         """
         Bind this node to a pipeline for external triggering.
 
         Once bound, the node can trigger downstream execution when called.
         This is typically done automatically during pipeline construction.
 
-        :param pipeline: The pipeline to bind this node to.
-        :type pipeline: Pipeline
+        :param pipeline: The pipeline to bind this node to, or None once its
+            flow has been taken out of the graph.
+        :type pipeline: Pipeline | None
         """
         self._pipeline = pipeline
 
@@ -466,6 +467,9 @@ class Node:
             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`).
+                return None
             self._pipeline.publish_log(self, collected, "")
 
         return self._pipeline.trigger(self, self._to_messages(outputs), durable=durable)

backend/fluksio/flow/nodes/service.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import queue
 import shlex
 import uuid
 from collections.abc import Iterable
@@ -169,9 +170,12 @@ class ServiceNode(Node):
             "env": self.cfg.env,
             "weights": self.cfg.weights,
         }
-        # On a thread, as every call is, and holding no slot.
+        # On a thread, as every call is, and holding no slot. Its inbox is made
+        # here, so a stop that comes before the thread has sent the serve can
+        # still end it, after the serve rather than ahead of it.
+        inbox: queue.Queue[dict[str, Any] | None] = queue.Queue()
         call = asyncio.ensure_future(
-            asyncio.to_thread(worker.request, payload, 0, slot=False)
+            asyncio.to_thread(worker.request, payload, 0, slot=False, inbox=inbox)
         )
         # Retrieved however it ends, so one that outlives a stop is not
         # reported at exit.
@@ -197,7 +201,7 @@ class ServiceNode(Node):
                 # The flow stopped, or this side failed. Ending the call kills
                 # the server's process group on the worker, and its answer says
                 # that is done, so the next start finds the port free.
-                await asyncio.to_thread(worker.cancel, call_id)
+                inbox.put(worker.END)
                 await asyncio.wait({call}, timeout=STOP_WAIT_S)
             await asyncio.to_thread(self._withdraw)
 

backend/fluksio/flow/pipeline.py

@@ -483,8 +483,16 @@ class Pipeline:
             self._gate_revision += 1
 
     def remove_flow(self, flow: str) -> None:
-        """Take a deleted flow out of the graph."""
+        """Take a deleted flow out of the graph.
+
+        Its nodes are unbound as well. One still running has nowhere left to
+        put what it returns, and writing it anyway would bring the deleted
+        flow's values back.
+        """
+        gone = self.flow_nodes(flow)
         self.replace_flow(flow, [])
+        for node in gone:
+            node.bind(None)
 
     def _seed(self, initial_values: dict[str, Any] | None) -> None:
         """Give messages a starting value, without overwriting one already there.
@@ -1119,6 +1127,9 @@ 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`).
+                return None
             self.publish_log(node, collected, "")
 
             if (
@@ -1180,6 +1191,10 @@ 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.
+                return None
             # One failing node must not take the rest of the graph down.
             error = self.publish_error(node, exc, collected, entry_id)
             if self.observer is not None:

backend/fluksio/flow/remote.py

@@ -94,6 +94,11 @@ class WorkerInventory:
 class RemoteWorker:
     """One attached worker, and the calls it has in flight."""
 
+    #: Put in a call's inbox to end the call. The thread waiting there sends
+    #: the cancel, and only once the call's own frame is out, so a cancel never
+    #: reaches the worker ahead of the call it ends, which it would ignore.
+    END: dict[str, Any] = {}
+
     def __init__(
         self,
         name: str,
@@ -169,16 +174,21 @@ class RemoteWorker:
         timeout: float,
         on_event: Callable[[dict[str, Any]], None] | None = None,
         slot: bool = True,
+        inbox: queue.Queue[dict[str, Any] | None] | None = None,
     ) -> dict[str, Any]:
         """Send a call and wait for its answer, on a thread rather than the loop.
 
         ``slot`` off is for what must not queue behind running calls, nor hold
         one: a compile, and a service, which would hold one for good.
+
+        ``inbox`` is the caller's own, for one that may end the call before
+        this thread has even sent it: `END` put there is read after the frame.
         """
         if self._gone:
             raise RemoteError(f"worker '{self.name}' is no longer attached")
         call_id = str(payload["call_id"])
-        inbox: queue.Queue[dict[str, Any] | None] = queue.Queue()
+        if inbox is None:
+            inbox = queue.Queue()
         # Blocking here is the backpressure, exactly as taking a slot is in the
         # local pool.
         if slot:
@@ -213,6 +223,11 @@ class RemoteWorker:
                     ) from None
                 if message is None:
                     raise RemoteError(f"worker '{self.name}' went away mid-call")
+                if message is self.END:
+                    # The worker answers once the call is killed, and only
+                    # that answer gives the slot back.
+                    self.cancel(call_id)
+                    continue
                 kind = message.get("event")
                 if kind == "heartbeat":
                     continue
@@ -260,6 +275,17 @@ class RemoteWorker:
         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."""
+        with self._lock:
+            inboxes = [
+                inbox
+                for call, inbox in self._pending.items()
+                if call.startswith(prefix)
+            ]
+        for inbox in inboxes:
+            inbox.put(self.END)
+
     def cancel(self, call_id: str) -> None:
         """Ask the worker to kill what it is running for this call."""
         # A closed loop refuses the coroutine and leaves it never awaited, so it
@@ -509,3 +535,12 @@ class RemoteWorkerHub:
                 worker.cancel(call)
                 stopped += 1
         return stopped
+
+    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.
+        """
+        for worker in self.workers():
+            worker.end_calls(f"{flow}.")

backend/tests/flow/test_remote.py

@@ -9,6 +9,7 @@ what happened rather than a wait that never ends.
 
 import asyncio
 import gc
+import sys
 import threading
 import time
 import warnings
@@ -17,15 +18,20 @@ from collections.abc import Iterator
 import pytest
 
 from fluksio.flow import remote
+from fluksio.flow.controller import FlowController
+from fluksio.flow.messages import DType, MessageSpec
 from fluksio.flow.remote import (
     NoWorker,
     RemoteWorker,
     RemoteWorkerHub,
     WorkerInventory,
 )
+from fluksio.flow.schemas import FlowDef, NodeDef
+from fluksio.flow.store import FlowStore
 from fluksio.flow.workers import (
     MAX_REMOTE_NAME,
     NodeTimeout,
+    PythonWorkerPool,
     RemoteError,
     _remote_class,
 )
@@ -243,6 +249,60 @@ def test_cancelling_a_run_reaches_only_that_run(loop):
     assert [frame["call_id"] for frame in cancels] == ["run-a:flow.node"]
 
 
+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")
+
+    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)
+    assert socket.sent[-1]["op"] == "run"
+
+    socket.arrived.clear()
+    key = store.delete_flow("voice")
+    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"}
+
+    worker.deliver({"call_id": "voice.gpu", "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_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

@@ -3,6 +3,7 @@
 import asyncio
 import time
 from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor
 
 from fluksio.flow import remote
 from fluksio.flow.messages import DType, MessageSpec
@@ -177,3 +178,54 @@ def test_a_service_that_never_gets_healthy_reads_down(monkeypatch):
         await node.stop()
 
     asyncio.run(scenario())
+
+
+def test_a_stop_right_after_the_start_ends_the_serve_after_it(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.
+    """
+    monkeypatch.setattr(remote, "SILENCE_S", 1.0)
+    monkeypatch.setattr(service, "STOP_WAIT_S", 1.0)
+
+    async def refused(client, url):
+        return "is not answering"
+
+    monkeypatch.setattr(ServiceNode, "_probe", staticmethod(refused))
+
+    class Busy(ThreadPoolExecutor):
+        """Starts what it is given late, as a pool with every thread taken."""
+
+        busy = True
+        late = 0
+
+        def submit(self, fn, /, *args, **kwargs):
+            if not self.busy:
+                return super().submit(fn, *args, **kwargs)
+            self.late += 1
+            return super().submit(lambda: time.sleep(0.2) or fn(*args, **kwargs))
+
+    agent = Agent()
+
+    async def scenario() -> None:
+        pool = Busy()
+        asyncio.get_running_loop().set_default_executor(pool)
+        agent.worker = RemoteWorker(
+            name="gpu-worker",
+            labels=["gpu"],
+            send=agent.send_json,
+            loop=asyncio.get_running_loop(),
+        )
+        hub = RemoteWorkerHub()
+        hub.attach(agent.worker)
+        node = ServiceNode(params={"command": "serve", "worker": "gpu"})
+        node.assign_flow("voice", "parrot")
+        node.hub = hub
+
+        await node.start()
+        await until(lambda: pool.late == 1)
+        pool.busy = False
+        await node.stop()
+
+    asyncio.run(scenario())
+    assert [frame["op"] for frame in agent.sent] == ["serve", "cancel"]

docs/code/workers.md

@@ -96,10 +96,11 @@ 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. 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 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.
+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,