From 084194f77b3dc411883b86ecc249743ac5c3df86 Mon Sep 17 00:00:00 2001 From: stroblme Date: Sun, 23 Aug 2026 17:11:55 +0200 Subject: [PATCH] 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) Claude-Session: https://claude.ai/code/session_01StpRc2C6au1WJ1EUU7fsfu --- NOTEPAD.md | 19 +-- backend/fluksio/api/routes/flows.py | 13 ++ backend/fluksio/flow/controller.py | 6 +- backend/fluksio/flow/workers.py | 168 +++++++++++++++++++------ backend/tests/api/routes/test_flows.py | 22 +++- backend/tests/flow/test_workers.py | 14 +++ worker/fluksio_worker/worker_main.py | 44 +++++-- 7 files changed, 224 insertions(+), 62 deletions(-) diff --git a/NOTEPAD.md b/NOTEPAD.md index 56a4067..e782cc2 100644 --- a/NOTEPAD.md +++ b/NOTEPAD.md @@ -178,12 +178,15 @@ Deferring because out of scope is fine, but don't mention deferring than. insurance for a future widget, but nothing would notice if they regressed. - CHORE/API: artifact blobs are content-addressed and have no GC, so deleting a flow drops its `run_artifact` rows and leaves the bytes on the data volume. -- CHORE/API: `DELETE /flows/{name}` does not refuse while the flow has a `running` or - `queued` Run. `RunService._record_node` can then insert `run_node` rows for a run that no - longer exists; `_finish` is an UPDATE, so it degrades to a harmless 0-row no-op. +- CHORE/API: no run table carries a foreign key to `run.id` — the baseline migration declares + none for `run_node`, `run_metric` or `run_artifact` — so nothing at the database level stops + a late write from orphaning rows. `DELETE /flows/{name}` now refuses while a run is + `running` or `queued`, which closes the path that was known; a constraint would need a + migration and would turn `_record_node`'s late insert into a hard error rather than the + no-op it is today. +- CHORE/INFRA: `backend/htmlcov` can be left root-owned by a container run, and `make test-backend` then fails at the coverage HTML step *after* every test has passed — - which reads like a test failure and is not one. - (tests/flows.spec.ts, tests/admin.spec.ts)". There are nine. + which reads like a test failure and is not one. `make clean` is the fix. - CHORE/UI: the edge popover shows the same value twice — `MessageSparkline` falls through to a collapsed `ValuePreview` for a non-numeric value, and `EdgeInspector` then renders its own `ValuePreview defaultOpen` below it. Cosmetic; one of the two is redundant. - FEAT/UI: a settings-and-inputs overview page, so what every node of an installation is configured with can be read and searched in one place rather than one panel at a time. - FEAT/UI: an input endpoint opens the flow panel, which is right for editing but not for reading one value. A panel of its own — the declaration, the current value, its history — is what clicking a label wants to give. @@ -279,11 +282,11 @@ does not settle. ### Out-of-process nodes and modules -- CHORE/FLOW: `compile_check` sends the *draft* source under the running node's cache key, so the worker recompiles the published source on its next call. Correct, but one wasted compile per save on a busy node. - FEAT/API: `POST /modules/apply` rebuilds the whole pipeline so a node that could not import its package stops being red. That resubscribes every MQTT node in the deployment; a targeted rebuild of the flows that actually failed to load would be gentler. - CHORE/FLOW: a node's return value now round-trips through JSON, so tuples arrive downstream as lists and anything non-JSON is an explicit error. That is the message contract, but flows written before this may notice. -- BUG/FLOW: a node whose cold-start imports plus body exceed its timeout can never succeed. The timeout covers the first call's imports, a timeout kills the worker so the next attempt is cold again, and `compile()` only ever warms one of the N workers. Broadcasting `compile` to every worker is the candidate fix, at the cost of N module executions per reload. -- CHORE/FLOW: worker protocol loose ends — the request `id` is echoed but never checked, `json.dumps` runs twice per result (once to prove it is JSON, once to send it), `_remote_types` is an unbounded cache keyed on class names that user code chooses, and `PythonWorkerPool._lock` guards less than its name suggests. +- CHORE/FLOW: a remote worker's agent spawns a fresh subprocess per call, so a node on a device pays for its imports on *every* call and pays them out of the node's own timeout — the cold-start problem local workers no longer have. Masked above ten seconds, because the agent's heartbeat resets the hub's silence deadline; a node with a shorter timeout and slow imports cannot run there at all. The fix is a persistent process on the agent, which is where its module cache would live. +- CHORE/FLOW: `worker_main.load_function` keeps every source it ever compiled in `sys.modules`, keyed by digest, and never drops one. A worker that has served many edits of one node holds all of them. Bounded in practice by `respawn_all` on a module change. +- CHORE/FLOW: `_remote_class` caps how many exception classes it caches but not how long a name may be, and for a remote worker that name comes off the wire. Local workers can only send a name their own code defined. ### Engine history diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index 3060626..a4c61ed 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/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: diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 29192c8..dead426 100644 --- a/backend/fluksio/flow/controller.py +++ b/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: diff --git a/backend/fluksio/flow/workers.py b/backend/fluksio/flow/workers.py index bf3b46b..12346bd 100644 --- a/backend/fluksio/flow/workers.py +++ b/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: diff --git a/backend/tests/api/routes/test_flows.py b/backend/tests/api/routes/test_flows.py index 179ed09..baa4fea 100644 --- a/backend/tests/api/routes/test_flows.py +++ b/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: diff --git a/backend/tests/flow/test_workers.py b/backend/tests/flow/test_workers.py index 4a62fcc..a6c4be2 100644 --- a/backend/tests/flow/test_workers.py +++ b/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): diff --git a/worker/fluksio_worker/worker_main.py b/worker/fluksio_worker/worker_main.py index 1a8a1ee..e50b59a 100644 --- a/worker/fluksio_worker/worker_main.py +++ b/worker/fluksio_worker/worker_main.py @@ -294,6 +294,13 @@ def _handle(request: dict[str, Any], cache: dict[tuple[str, str], Any]) -> Any: source = request.get("source") or "" digest = hashlib.md5(source.encode()).hexdigest() + if request["op"] == "compile" and not request.get("keep", True): + # A question about a draft, not a load: does this source compile? The + # answer is all the editor wants, and keeping it would throw away the + # published source this worker is still serving calls from. + load_function(flow, node, source) + return None + cached = cache.get((flow, node)) if cached is None or cached[0] != digest: cache[(flow, node)] = (digest, load_function(flow, node, source)) @@ -305,15 +312,6 @@ def _handle(request: dict[str, Any], cache: dict[tuple[str, str], Any]) -> Any: result = function(**(request.get("kwargs") or {})) if inspect.isgenerator(result): result = _drain(result) - try: - json.dumps(result) - except (TypeError, ValueError): - # The typed-message contract only carries JSON, and a node running out - # of process is where that stops being a formality. - raise ValueError( - f"returned {type(result).__name__}, which cannot be sent back as " - "JSON — return numbers, strings, booleans, lists or dicts." - ) from None return result @@ -360,7 +358,10 @@ def main() -> None: request = json.loads(line) _CALL_ID = str(request.get("call_id") or "") captured = _Capped() - response: dict[str, Any] = {"call_id": _CALL_ID} + # ``id`` back untouched: the engine matches it against what it sent, so + # a pipe that has slipped a call is caught rather than handing one node + # another node's answer. + response: dict[str, Any] = {"id": request.get("id"), "call_id": _CALL_ID} try: with ( contextlib.redirect_stdout(captured), @@ -377,7 +378,28 @@ def main() -> None: "traceback": _node_traceback(exc), } response["logs"] = captured.getvalue() - _RPC.write(json.dumps(response) + "\n") + try: + reply = json.dumps(response) + except (TypeError, ValueError, RecursionError): + # Encoding the reply is also the check that the node returned + # something the typed-message contract can carry — the result is + # the only part of this dict a node controls, so it is the only + # part that can fail. Doing it once is why there is no separate + # dumps of the result above. + kind = type(response.pop("result", None)).__name__ + trouble = ( + f"returned {kind}, which cannot be sent back as JSON — " + "return numbers, strings, booleans, lists or dicts." + ) + response["ok"] = False + response["error"] = { + "type": "ValueError", + "message": trouble, + "short": f"ValueError: {trouble}", + "traceback": "", + } + reply = json.dumps(response) + _RPC.write(reply + "\n") _RPC.flush()