Warm a worker before the node's clock starts, and refuse to delete a live flow

A node's timeout now covers its body only: the pool loads the source into the
worker it picked, off the node's budget, so imports that outlast the timeout no
longer make a node impossible to run. Draft checks compile without caching, so
saving does not evict what a busy node is serving calls from. Requests carry an
id the worker echoes and the pool checks, a reply is encoded once, and the
remote-exception cache is bounded.

DELETE /flows/{name} answers 409 while the flow has a running or queued run,
which is what was letting run_node rows outlive their run.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01StpRc2C6au1WJ1EUU7fsfu

backend/fluksio/api/routes/flows.py

@@ -454,6 +454,19 @@ async def delete_flow(
     name: str, controller: FlowControllerDep, user: CurrentUser, session: SessionDep
 ) -> Any:
     """Delete a flow and everything in it."""
+    live = session.exec(
+        select(Run.id).where(
+            col(Run.flow) == name, col(Run.status).in_(("running", "queued"))
+        )
+    ).first()
+    if live is not None:
+        raise HTTPException(
+            status_code=409,
+            detail=(
+                f"Flow '{name}' has a run in progress ({live}). Cancel it, or "
+                "wait for it to finish, before deleting the flow."
+            ),
+        )
     try:
         await run_in_threadpool(controller.store.delete_flow, name)
     except FlowNotFound:

backend/fluksio/flow/controller.py

@@ -793,11 +793,15 @@ class FlowController:
         Loaded on the machine that will run it: a node bound to a device is
         checked against that worker's packages, because a missing import here
         says nothing about whether it is missing there.
+
+        The source is usually a draft, and the node is usually still running
+        the published one — so ``keep=False``: compile it to answer the
+        question, and leave the worker holding what it is serving calls from.
         """
         if device and self.remote is not None:
             return self.remote.compile(device, flow, node_id, code)
         if self.workers is not None:
-            return self.workers.compile(flow, node_id, code)
+            return self.workers.compile(flow, node_id, code, keep=False)
         try:
             load_function(flow, node_id, code)
         except Exception as exc:

backend/fluksio/flow/workers.py

@@ -12,6 +12,8 @@ message contract already guarantees for everything a node consumes or provides.
 
 from __future__ import annotations
 
+import hashlib
+import itertools
 import json
 import logging
 import os
@@ -89,6 +91,11 @@ class NodeCancelled(RemoteError):
     """Someone asked for this node to stop while it was running."""
 
 
+#: How many distinct exception names to keep classes for. The key is a name
+#: node code chose, so an unbounded cache is a node raising a class named after
+#: its loop counter.
+MAX_REMOTE_TYPES = 256
+
 _remote_types: dict[str, type[RemoteError]] = {}
 
 
@@ -100,8 +107,13 @@ def _remote_class(name: str) -> type[RemoteError]:
     the transport that carried it.
     """
     cls = _remote_types.get(name)
-    if cls is None:
-        cls = type(name, (RemoteError,), {})
+    if cls is not None:
+        return cls
+    cls = type(name, (RemoteError,), {})
+    # Past the cap the class is still built, just not kept: the names already
+    # here are the ones a deployment actually raises, and evicting them to make
+    # room for generated ones would be the wrong way round.
+    if len(_remote_types) < MAX_REMOTE_TYPES:
         _remote_types[name] = cls
     return cls
 
@@ -114,6 +126,10 @@ class _Worker:
     ) -> None:
         self.generation = generation
         self.cancelled = False
+        # Mirrors the worker's own module cache: (flow, node) -> source digest.
+        # What it is for is knowing whether a call has to pay for imports
+        # before the node's clock starts; see ``PythonWorkerPool._warm``.
+        self.loaded: dict[tuple[str, str], str] = {}
         # What a read took past the end of a line. A node reporting quickly
         # puts several lines in one chunk, and the next one is the caller's to
         # read — dropping it would lose a metric, returning it with the first
@@ -207,6 +223,15 @@ class PythonWorkerPool:
         self._running: dict[tuple[str, str], _Worker] = {}
         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.
         self._lock = threading.Lock()
 
     # -------------------------------------------------------------------------
@@ -319,42 +344,11 @@ class PythonWorkerPool:
                 }
             )
         try:
-            try:
-                worker.send(payload)
-            except OSError as exc:
-                raise RemoteError(f"worker died: {exc}") from exc
-
-            while True:
-                # Reset per line, so the timeout measures how long the node has
-                # been silent rather than how long it has been working. A node
-                # that reports nothing is still held to it, which is what keeps
-                # the deadline meaningful for the ones that never report.
-                line = worker.read_line(time.monotonic() + timeout)
-                if line:
-                    try:
-                        message = dict(json.loads(line))
-                    except (TypeError, ValueError) as exc:
-                        # A reply we cannot read leaves this worker out of step:
-                        # whatever is still in its pipe would be taken by the
-                        # next request on this slot as its own answer. Retire it.
-                        worker.cancelled = True
-                        raise RemoteError(
-                            f"unreadable reply from the worker: {exc}"
-                        ) from exc
-                    if message.get("event"):
-                        if on_event is not None:
-                            try:
-                                on_event(message)
-                            except Exception:
-                                logger.exception("Could not record a worker event")
-                        continue
-                    return message
-                if worker.cancelled:
-                    raise NodeCancelled("cancelled while it was running")
-                if line is None:
-                    worker.kill()
-                    raise NodeTimeout(f"was silent for {timeout}s and was killed")
-                raise RemoteError("worker died")
+            if payload["op"] == "run":
+                refused = self._warm(worker, payload)
+                if refused is not None:
+                    return refused
+            return self._exchange(worker, payload, timeout, on_event)
         finally:
             # Under the lock, so a cancel that has already read this worker out
             # of _running cannot kill it after another node has taken the slot.
@@ -363,11 +357,103 @@ class PythonWorkerPool:
                     self._running.pop(key, None)
                 self._release(worker)
 
-    def compile(self, flow: str, node: str, source: str) -> str | None:
-        """Load this source in a worker. Returns what to show the author, or None."""
+    def _warm(self, worker: _Worker, payload: dict[str, Any]) -> dict[str, Any] | None:
+        """Load the node's source in this worker before its clock starts.
+
+        A node's timeout is what its *body* may take. Importing torch on a cold
+        worker is not the body, and charging it to the same budget cannot be
+        won: the timeout kills the worker, so the retry is cold again and pays
+        the imports from the start. ``compile`` warms only the one worker it
+        happened to land on, and a pool has several.
+
+        Returns the failed reply when the source does not load, so the caller
+        raises the author's own error rather than a timeout.
+        """
+        key = (str(payload["flow"]), str(payload["node"]))
+        digest = hashlib.md5(str(payload.get("source") or "").encode()).hexdigest()
+        if worker.loaded.get(key) == digest:
+            return None
+        response = self._exchange(worker, {**payload, "op": "compile"}, COMPILE_TIMEOUT)
+        if not response.get("ok"):
+            return response
+        worker.loaded[key] = digest
+        return None
+
+    def _exchange(
+        self,
+        worker: _Worker,
+        payload: dict[str, Any],
+        timeout: float,
+        on_event: Callable[[dict[str, Any]], None] | None = None,
+    ) -> dict[str, Any]:
+        """One request down the pipe, and the reply that belongs to it."""
+        request_id = next(self._requests)
+        try:
+            worker.send({**payload, "id": request_id})
+        except OSError as exc:
+            raise RemoteError(f"worker died: {exc}") from exc
+
+        while True:
+            # Reset per line, so the timeout measures how long the node has
+            # been silent rather than how long it has been working. A node
+            # that reports nothing is still held to it, which is what keeps
+            # the deadline meaningful for the ones that never report.
+            line = worker.read_line(time.monotonic() + timeout)
+            if line:
+                try:
+                    message = dict(json.loads(line))
+                except (TypeError, ValueError) as exc:
+                    # A reply we cannot read leaves this worker out of step:
+                    # whatever is still in its pipe would be taken by the
+                    # next request on this slot as its own answer. Retire it.
+                    worker.cancelled = True
+                    raise RemoteError(
+                        f"unreadable reply from the worker: {exc}"
+                    ) from exc
+                if message.get("event"):
+                    if on_event is not None:
+                        try:
+                            on_event(message)
+                        except Exception:
+                            logger.exception("Could not record a worker event")
+                    continue
+                if message.get("id") != request_id:
+                    # The pipe is a call behind. Returning this would hand one
+                    # node another node's result, which is worse than any
+                    # failure — so say so and retire the worker rather than
+                    # log a warning nobody reads.
+                    worker.cancelled = True
+                    raise RemoteError(
+                        f"the worker answered request {message.get('id')!r} "
+                        f"while {request_id} was outstanding"
+                    )
+                return message
+            if worker.cancelled:
+                raise NodeCancelled("cancelled while it was running")
+            if line is None:
+                worker.kill()
+                raise NodeTimeout(f"was silent for {timeout}s and was killed")
+            raise RemoteError("worker died")
+
+    def compile(
+        self, flow: str, node: str, source: str, keep: bool = True
+    ) -> str | None:
+        """Load this source in a worker. Returns what to show the author, or None.
+
+        ``keep`` off is a question rather than a load: the source is compiled
+        to see whether it would run and then dropped. That is what a draft
+        wants — caching it would evict the published source the node's next
+        call is about to need.
+        """
         try:
             response = self._request(
-                {"op": "compile", "flow": flow, "node": node, "source": source},
+                {
+                    "op": "compile",
+                    "flow": flow,
+                    "node": node,
+                    "source": source,
+                    "keep": keep,
+                },
                 timeout=COMPILE_TIMEOUT,
             )
         except RemoteError as exc:

backend/tests/api/routes/test_flows.py

@@ -281,7 +281,9 @@ def test_delete_flow(
     client: TestClient, superuser_token_headers: dict[str, str], db: Session
 ) -> None:
     client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
-    db.add(Run(id="run-1", flow="demo", created_at=datetime.now(UTC)))
+    # Finished: a flow with a live run refuses to be deleted, which is its own
+    # test below.
+    db.add(Run(id="run-1", flow="demo", status="ok", created_at=datetime.now(UTC)))
     db.add(RunNode(run_id="run-1", node="sensor"))
     db.add(RunMetric(run_id="run-1", name="loss", step=-1))
     db.add(RunArtifact(run_id="run-1", name="model.pt"))
@@ -301,6 +303,24 @@ def test_delete_flow(
         assert db.exec(select(func.count()).select_from(model)).one() == 0
 
 
+def test_delete_flow_is_refused_while_a_run_is_live(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
+    db.add(
+        Run(id="run-live", flow="demo", status="running", created_at=datetime.now(UTC))
+    )
+    db.commit()
+
+    response = client.delete(f"{PREFIX}/demo", headers=superuser_token_headers)
+    assert response.status_code == 409
+    assert "run-live" in response.json()["detail"]
+    # Its driver is still writing node rows against that id, so nothing may go.
+    assert (
+        client.get(f"{PREFIX}/demo", headers=superuser_token_headers).status_code == 200
+    )
+
+
 def test_node_types_are_listed(
     client: TestClient, superuser_token_headers: dict[str, str]
 ) -> None:

backend/tests/flow/test_workers.py

@@ -76,6 +76,20 @@ def test_a_node_that_runs_too_long_is_killed_and_the_pool_recovers(pool):
     assert run(pool, "def process():\n    return {'out': 2}\n") == {"out": 2}
 
 
+def test_a_slow_import_is_not_charged_to_the_nodes_timeout(pool):
+    # Module-level work is what a node's imports are, and it happens once, on a
+    # cold worker. Paying for it out of the per-call budget makes a node with
+    # heavy imports impossible to run at all: the timeout kills the worker, so
+    # the next attempt is cold again and starts over.
+    code = "import time\n\ntime.sleep(1.5)\n\n\ndef process():\n    return {'out': 1}\n"
+    assert pool.run("demo", "heavy", code, {}, "demo.heavy", timeout=0.5) == {"out": 1}
+
+    # And the same worker does not pay for them a second time.
+    started = time.monotonic()
+    assert pool.run("demo", "heavy", code, {}, "demo.heavy", timeout=0.5) == {"out": 1}
+    assert time.monotonic() - started < 1.0
+
+
 def test_a_running_node_can_be_cancelled(pool):
     def stop_it() -> None:
         for _ in range(100):