From 83c30aa1c73b08471357da38703a78c35190fefe Mon Sep 17 00:00:00 2001 From: stroblme Date: Sun, 16 Aug 2026 23:46:08 +0200 Subject: [PATCH] Take the engine off the path node code imports from, and let it stop The worker script is handed to the interpreter by path, so app/flow was sys.path[0] for every node: `import queue` got the engine's. It now drops its own directory before anything else imports, and runs with the deployment's credentials scrubbed out of its environment. Also: reload builds off the event loop, the pool wakes what is blocked on it when it stops, a refused metrics flush is kept for the next one rather than dropped, and the cascade events are paired through failures. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017MeiWk3Yq12n2pTvnQWYvt --- NOTEPAD.md | 16 +++-- backend/app/api/routes/observability.py | 6 +- backend/app/flow/controller.py | 23 +++++--- backend/app/flow/executor.py | 28 +++++---- backend/app/flow/metrics.py | 44 +++++++++++++- backend/app/flow/modules.py | 22 +++++-- backend/app/flow/nodes/http.py | 11 +++- backend/app/flow/pipeline.py | 24 ++++---- backend/app/flow/schemas.py | 6 +- backend/app/flow/worker_main.py | 13 ++++- backend/app/flow/workers.py | 68 ++++++++++++++++++---- backend/pyproject.toml | 3 + backend/tests/flow/test_modules.py | 15 +++++ backend/tests/flow/test_workers.py | 68 ++++++++++++++++++++++ backend/tests/test_metrics.py | 50 ++++++++++++++++ frontend/src/client/schemas.gen.ts | 4 +- frontend/src/client/types.gen.ts | 4 +- frontend/src/components/Flow/liveStore.ts | Bin 9108 -> 9113 bytes frontend/src/routes/_layout/health.tsx | 6 +- 19 files changed, 346 insertions(+), 65 deletions(-) diff --git a/NOTEPAD.md b/NOTEPAD.md index 24627d5..f1a3aca 100644 --- a/NOTEPAD.md +++ b/NOTEPAD.md @@ -14,10 +14,8 @@ should reopen it. ### To be sorted - INFRA: merge the philosophy statement at the beginning of vision.md into the rest of the document. Dissolve the decision dates and fold the decisions into a clean structure -- BUG/UI mobile friendly support is degraded: 1) toolbar in the "Flows" viewport extend mobile viewport width 2) position of nodes should never be static; always adjust such that there are as few as possible overlaps and direction is left to right (desktop) or top to bottom (mobile) 3) Dashboard view is not mobile friendly at all; as dashboard design is infeasible on mobile, render all widgets in a vertically stacked order. This allows to inspect each widget and make changes. Layout changes are not a feature on mobile 4) the home view is not responsive; all items shown there should re-order on mobile such that no scrollbars appear - -- Move the "brain" into the "Home" at the top; flat without a box and the same background, such the brain renders just on the surface of the home screen -- Move the "Health" page into "Home" as well underneath the flow control toggles (replaces the running / failing flows row in the current home view) +- BUG/UI mobile friendly support is degraded: 1) toolbar in the "Flows" viewport extend mobile viewport width 2) position of nodes should never be static; always adjust such that there are as few as possible overlaps (of nodes and edge labels) and direction is left to right (desktop) or top to bottom (mobile) with a minimal (but clean) overall edge length 3) Dashboard view is not mobile friendly at all; as dashboard design is infeasible on mobile, render all widgets in a vertically stacked order. This allows to inspect each widget and make changes. Layout changes are not a feature on mobile 4) the home view is not responsive; all items shown there should re-order on mobile such that no scrollbars appear +- BUG/UI remove the up/down counters from integer fields (global) ### Connector write paths @@ -44,6 +42,8 @@ is what M4 still waits on, together with porting the flows. - 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. ### Engine history @@ -53,8 +53,13 @@ is what M4 still waits on, together with porting the flows. - CHORE/API: `/observability/summary` reports the work queue's `depth` as the Redis stream length, which is the journal size (capped at `STREAM_MAXLEN`) rather than a backlog. The health screen shows `pending` instead; the field name still invites the wrong reading. - FEAT/UI: the health screen's window is fixed at 24 hours and the charts fold minute buckets in Python. A range picker (and `date_bin()` behind it) is the next step if anyone wants a week. - CHORE/FLOW: run records for a deleted flow stay until the retention window passes, so a flow that no longer exists keeps appearing in the history. Deliberate — it is a record of what ran — but `forget_flow` could offer to clear it. -- BUG/API: `MetricsCollector.flush` takes its buckets and pending events out of the collector before writing them, so a failed write loses everything accumulated since the last flush rather than retrying it. Unobserved so far — the soak run never took Postgres away — but it is a silent hole whenever the database blips. Putting the batch back, merged with whatever arrived meanwhile, closes it. - CHORE/API: nothing can ask the collector to flush now, so anything needing the tables to be current has to wait out `FLUSH_INTERVAL_S` — which is what the soak harness does before clearing its own rows. +- BUG/UI: `MemoryWorkQueue.stats()` hard-codes `pending: 0`, so the health tile always reads zero on a stack without Redis. +- CHORE/API: `MetricsCollector._start_run`'s `existing is not None` branch is unreachable: a redelivery only arrives after the record it would update has been dropped. +- CHORE/API: the collector's `_tracebacks` stash is never pruned for nodes that stop failing — one entry per node that ever errored, kept for the life of the process. +- CHORE/API: audit rows ride the same drop-oldest bus as telemetry, so a storm can lose one. Writing a node's source is not audited either; publishing is. +- PERF/API: two unmarked ceilings — `/observability/flows` scans every `metric_minute` row in the window on each 30 s poll, and `queue.stats()` does a keyspace `scan_iter` on every call while two endpoints poll it. +- CHORE/INFRA: dev only — memory-queue ids (`mem-{seq}`) restart at 0 each boot and `FlowRun.id` is the primary key, so a restart without Redis upserts over the previous boot's run rows. ### Dashboard follow-ups @@ -79,6 +84,7 @@ is what M4 still waits on, together with porting the flows. - PERF/UI: `useParamSuggestions` fetches every flow's detail to build the suggestion list. An aggregate endpoint if an installation ever has many flows. - CHORE/UX: the derived-cron chip also appears on the delay node, where `interval` is a rate limit rather than a schedule. May want it inject-only. - CHORE/UX: free-form params (python nodes) get no suggestions, since there is no schema to key them off. +- PERF/UI: `BrainView` runs 300 force-layout ticks synchronously inside a `useMemo`, so the graph is laid out on the render thread. - FEAT/UI: the brain graph draws connections as plain lines, so which way a value travels is not shown. An arrowhead needs the path shortened to the target circle's rim, which needs the node radius on the edge. - CHORE/UI: fixing the yellow message pulse gave `edge-pulse` an explicit `to`, so a *selected* edge now decays to `--muted-foreground` and snaps back to its blue at the end instead of decaying into it. Barely visible, and only while an edge is selected. diff --git a/backend/app/api/routes/observability.py b/backend/app/api/routes/observability.py index 7222bc3..b9ae34a 100644 --- a/backend/app/api/routes/observability.py +++ b/backend/app/api/routes/observability.py @@ -123,11 +123,13 @@ async def read_summary( if errored: problems.append(f"{len(errored)} node(s) failed to load") - failures = session.exec( + statement = ( select(func.count()) .select_from(EngineEvent) .where(col(EngineEvent.ts) >= _since(24), col(EngineEvent.type) != "audit") - ).one() + ) + # The health page polls this every ten seconds; the driver is synchronous. + failures = await run_in_threadpool(lambda: session.exec(statement).one()) return HealthSummary( status="degraded" if problems else "ok", diff --git a/backend/app/flow/controller.py b/backend/app/flow/controller.py index 1aef065..00ce655 100644 --- a/backend/app/flow/controller.py +++ b/backend/app/flow/controller.py @@ -302,8 +302,12 @@ class FlowController: for flow in published if not self.store.read_enabled(flow.name) } - nodes, loaded, initial_values, flow_inputs = self._build_flows( - [(flow, False) for flow in published] + # Off the loop: building a python node asks its worker to compile, + # which waits for a free slot — and a busy node holds one for as + # long as its timeout. On the loop that stalls every request, the + # websocket and the health check the container is judged by. + nodes, loaded, initial_values, flow_inputs = await run_in_threadpool( + self._build_flows, [(flow, False) for flow in published] ) self.loaded = loaded @@ -708,11 +712,16 @@ class FlowController: for node_def in flow.nodes: member = f"{flow.name}.{node_def.id}" node_type = NODE_TYPES.get(node_def.type) - key = ( - node_type.cls.instance_key(node_def.params) - if node_type is not None - else None - ) + key = None + if node_type is not None: + try: + key = node_type.cls.instance_key(node_def.params) + except Exception: + # A plugin's own grouping is not worth the whole graph; + # this node just stands on its own. + logger.exception( + "instance_key failed for node type '%s'", node_def.type + ) gid = f"{node_def.type}:{key}" if key else member gid_of[member] = gid diff --git a/backend/app/flow/executor.py b/backend/app/flow/executor.py index 1a14ca9..2909658 100644 --- a/backend/app/flow/executor.py +++ b/backend/app/flow/executor.py @@ -298,18 +298,22 @@ class ExecutionService: } ) - pipeline.apply_outputs(node, item.outputs or None) - pipeline.run_downstream( - node, entry_id=item.entry_id, replay=item.deliveries > 1 - ) - self._publish( - { - "type": "cascade_finished", - "run": item.entry_id, - "flow": item.flow, - "ts": time.time(), - } - ) + try: + pipeline.apply_outputs(node, item.outputs or None) + pipeline.run_downstream( + node, entry_id=item.entry_id, replay=item.deliveries > 1 + ) + finally: + # Paired, or a cascade that raised — state backend gone, say — is a + # run left open until the abandoned sweep ten minutes later. + self._publish( + { + "type": "cascade_finished", + "run": item.entry_id, + "flow": item.flow, + "ts": time.time(), + } + ) return True # ------------------------------------------------------------------------- diff --git a/backend/app/flow/metrics.py b/backend/app/flow/metrics.py index 8d438dd..035c3ff 100644 --- a/backend/app/flow/metrics.py +++ b/backend/app/flow/metrics.py @@ -35,6 +35,13 @@ FLUSH_INTERVAL_S = 15.0 DETAIL_CAP = 8000 #: A cascade still open this long after it started is never finishing. RUN_STALE_S = 600.0 +#: ``metric_minute.flow``/``node`` and ``flow_run.flow`` are this wide, and a +#: node id has no length of its own. One row over it aborts the transaction, +#: which would take every other row in the flush with it. +NAME_MAX = 255 +#: How much unwritten history is held while the database is unreachable. Enough +#: for a long outage, bounded so the outage cannot become a memory leak. +HOLD_MAX = 10_000 #: Retention is checked this often, not on every flush. PRUNE_INTERVAL_S = 3600.0 @@ -117,8 +124,8 @@ class MetricsCollector: # ponytail: one collector, one row per node per minute; coarsen the # bucket if the node count ever reaches thousands. key = ( - str(event.get("flow") or ""), - str(event.get("node") or ""), + str(event.get("flow") or "")[:NAME_MAX], + str(event.get("node") or "")[:NAME_MAX], _minute(float(event.get("ts") or time.time())), ) return self._buckets.setdefault( @@ -247,7 +254,7 @@ class MetricsCollector: return self._runs[run_id] = { "id": run_id, - "flow": str(event.get("flow") or ""), + "flow": str(event.get("flow") or "")[:NAME_MAX], "source": str(event.get("cause") or ""), "started_ts": ts, "started_at": datetime.fromtimestamp(ts, timezone.utc), @@ -276,6 +283,7 @@ class MetricsCollector: await asyncio.to_thread(self._write, buckets, pending, runs, prune) except Exception: logger.exception("Could not write engine metrics") + self._hold(buckets, pending) return if prune: self._last_prune = time.monotonic() @@ -287,6 +295,36 @@ class MetricsCollector: # to are two events, and a flush can fall between them. One per node, # each replaced by that node's next failure. + def _hold( + self, + buckets: dict[tuple[str, str, datetime], dict[str, float]], + pending: list[EngineEvent], + ) -> None: + """Take a batch the database refused back, rather than losing it. + + These are the audit trail and the failures with their tracebacks — the + rows nobody can reconstruct afterwards. The batch goes in front of what + arrived since, so what a long outage drops is the oldest. + """ + for key, agg in self._buckets.items(): + held = buckets.get(key) + if held is None: + buckets[key] = agg + continue + for name in SUMMED: + held[name] += agg[name] + for name in MAXIMA: + held[name] = max(held[name], agg[name]) + self._buckets = buckets + # A failed commit puts the rows back to unsaved. One that got as far as + # being handed an id keeps it, and the sequence has already moved past + # it, so adding them to the next session inserts them cleanly. + self._pending[:0] = pending + + for key in list(self._buckets)[: max(0, len(self._buckets) - HOLD_MAX)]: + del self._buckets[key] + del self._pending[: max(0, len(self._pending) - HOLD_MAX)] + def _write( self, buckets: dict[tuple[str, str, datetime], dict[str, float]], diff --git a/backend/app/flow/modules.py b/backend/app/flow/modules.py index 4522dce..3a54526 100644 --- a/backend/app/flow/modules.py +++ b/backend/app/flow/modules.py @@ -79,12 +79,18 @@ def ensure_venv() -> None: def sync(requirements: str) -> tuple[bool, str]: - """Make the venv hold exactly these packages. Returns success and uv's output.""" - ensure_venv() - with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as handle: - handle.write(requirements) - manifest = handle.name + """Make the venv hold exactly these packages. Returns success and uv's output. + + ``uv`` missing, or refusing to make the venv, is a failed apply like any + other — the caller answers 400 with what came back, which is the only thing + a person can act on. + """ + manifest = "" try: + ensure_venv() + with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as handle: + handle.write(requirements) + manifest = handle.name result = subprocess.run( # Empty means empty: without the flag uv refuses to clear a venv, # so deleting the last line would leave the package installed. @@ -101,8 +107,12 @@ def sync(requirements: str) -> tuple[bool, str]: text=True, timeout=SYNC_TIMEOUT, ) + except (OSError, subprocess.SubprocessError) as exc: + detail = str(getattr(exc, "stderr", None) or exc).strip() + return False, f"uv could not run: {detail}" finally: - Path(manifest).unlink(missing_ok=True) + if manifest: + Path(manifest).unlink(missing_ok=True) output = (result.stdout + result.stderr).strip() if result.returncode == 0: diff --git a/backend/app/flow/nodes/http.py b/backend/app/flow/nodes/http.py index 41a4a4e..b4a39a8 100644 --- a/backend/app/flow/nodes/http.py +++ b/backend/app/flow/nodes/http.py @@ -8,6 +8,7 @@ import threading from collections.abc import Iterable from enum import Enum from typing import TYPE_CHECKING, Any, Literal +from urllib.parse import urlsplit, urlunsplit import httpx from pydantic import BaseModel, ConfigDict, Field @@ -156,7 +157,15 @@ class HttpNode(Node): # prefixed onto it, so two flows both receiving on "/tick" merge into # one neuron. Key on mode as well if that ever misleads. url = params.get("url") - return str(url) if url else None + if not url: + return None + # A URL may legally carry credentials, and this key becomes a group id + # in the brain graph — rendered into the response and into the DOM. + parts = urlsplit(str(url)) + if parts.username or parts.password: + host = parts.netloc.rsplit("@", 1)[-1] + return urlunsplit(parts._replace(netloc=host)) + return str(url) def __init__( self, diff --git a/backend/app/flow/pipeline.py b/backend/app/flow/pipeline.py index cb50c3e..a38d389 100644 --- a/backend/app/flow/pipeline.py +++ b/backend/app/flow/pipeline.py @@ -932,16 +932,20 @@ class Pipeline: "ts": time.time(), } ) - self.apply_outputs(node, outputs) - state = self.run_downstream(node, entry_id=run_id) - self._publish( - { - "type": "cascade_finished", - "run": run_id, - "flow": node.flow, - "ts": time.time(), - } - ) + try: + self.apply_outputs(node, outputs) + state = self.run_downstream(node, entry_id=run_id) + finally: + # Paired, or a cascade that raised leaves the run open until the + # abandoned sweep ten minutes later. + self._publish( + { + "type": "cascade_finished", + "run": run_id, + "flow": node.flow, + "ts": time.time(), + } + ) return state def publish( diff --git a/backend/app/flow/schemas.py b/backend/app/flow/schemas.py index 9932a20..dc4b80b 100644 --- a/backend/app/flow/schemas.py +++ b/backend/app/flow/schemas.py @@ -48,8 +48,10 @@ class NodeDef(BaseModel): default=None, gt=0, description=( - "Seconds this node's code may run before it is stopped. Above 60 " - "the engine may deliver its work again while it is still running." + "Seconds this node's code may run before it is stopped. This " + "covers the first call's imports, which can be much slower than " + "the body. Above 60 the engine may deliver its work again while " + "it is still running." ), ) diff --git a/backend/app/flow/worker_main.py b/backend/app/flow/worker_main.py index 3b344f3..2dce883 100644 --- a/backend/app/flow/worker_main.py +++ b/backend/app/flow/worker_main.py @@ -14,12 +14,21 @@ instead of corrupting the reply stream. from __future__ import annotations +import os +import sys + +# Handed to the interpreter by path, so CPython puts this file's own directory +# — the engine's ``app/flow`` — on ``sys.path[0]``. A node doing ``import +# queue`` would then get the engine's queue module rather than the standard +# library's. Drop it before anything else can import. (``-P`` and +# ``PYTHONSAFEPATH`` do this at startup, but both are 3.11+.) +_HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path[:] = [p for p in sys.path if os.path.abspath(p or ".") != _HERE] + import contextlib import hashlib import io import json -import os -import sys import traceback from collections.abc import Callable from types import ModuleType diff --git a/backend/app/flow/workers.py b/backend/app/flow/workers.py index 1d04ea2..47e11f7 100644 --- a/backend/app/flow/workers.py +++ b/backend/app/flow/workers.py @@ -36,6 +36,28 @@ WORKER_MAIN = Path(__file__).with_name("worker_main.py") #: something a person is watching a spinner for. COMPILE_TIMEOUT = 60.0 +#: Environment the worker is not given. ``SECRET_KEY`` decrypts every stored +#: secret, not only the ones bound to the node asking. +ENV_DENY_PREFIXES = ("POSTGRES_", "FIRST_SUPERUSER", "SENTRY_DSN") +ENV_DENY_WORDS = ("PASSWORD", "SECRET", "TOKEN") + + +def worker_env() -> dict[str, str]: + """The engine's environment with its credentials taken out. + + A denylist, so ``PATH``, ``HOME``, the locale and whatever else a venv + needs still reach the worker. This is not a sandbox: the worker is the same + user in the same container, with the same filesystem and the same network. + It only means node code cannot read the deployment's secrets straight out + of its own environment. + """ + return { + key: value + for key, value in os.environ.items() + if not key.startswith(ENV_DENY_PREFIXES) + and not any(word in key.upper() for word in ENV_DENY_WORDS) + } + class RemoteError(Exception): """Something that went wrong inside a worker, re-raised on this side. @@ -85,6 +107,7 @@ class _Worker: stdin=subprocess.PIPE, stdout=subprocess.PIPE, close_fds=True, + env=worker_env(), ) def alive(self) -> bool: @@ -155,6 +178,7 @@ class PythonWorkerPool: # cancel kills the newest. Key by run id if that ever matters. self._running: dict[str, _Worker] = {} self._generation = 0 + self._stopped = False self._lock = threading.Lock() # ------------------------------------------------------------------------- @@ -167,12 +191,18 @@ class PythonWorkerPool: self._idle.put(None) def stop(self) -> None: + self._stopped = True self._generation += 1 for worker in list(self._running.values()): worker.kill() for slot in self._drain(): if slot is not None: slot.kill() + # A node thread can be blocked on a slot right now, and the threads are + # not daemons — without a sentinel each it waits for a worker that is + # never coming back, and the interpreter cannot exit. + for _ in range(self.size): + self._idle.put(None) def respawn_all(self) -> None: """Retire every worker, so the next call picks up a changed venv. @@ -201,7 +231,12 @@ class PythonWorkerPool: def _acquire(self) -> _Worker: """Take a slot, blocking while every worker is busy — that is the backpressure.""" + if self._stopped: + raise RemoteError("the worker pool is shutting down") slot = self._idle.get() + if self._stopped: + # The sentinel stop() put back, so this is a wake-up rather than a slot. + raise RemoteError("the worker pool is shutting down") if ( slot is not None and slot.alive() @@ -256,7 +291,16 @@ class PythonWorkerPool: line = worker.read_line(time.monotonic() + timeout) if line: - return dict(json.loads(line)) + try: + return 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 worker.cancelled: raise NodeCancelled("cancelled while it was running") if line is None: @@ -264,9 +308,12 @@ class PythonWorkerPool: raise NodeTimeout(f"exceeded {timeout}s and was killed") raise RemoteError("worker died") finally: - if node_id: - self._running.pop(node_id, None) - self._release(worker) + # 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. + with self._lock: + if node_id: + self._running.pop(node_id, 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.""" @@ -330,12 +377,13 @@ class PythonWorkerPool: def cancel(self, node_id: str) -> bool: """Stop a node that is running now. False when there was nothing to stop.""" - worker = self._running.get(node_id) - if worker is None: - return False - worker.cancelled = True - worker.kill() - return True + with self._lock: + worker = self._running.get(node_id) + if worker is None: + return False + worker.cancelled = True + worker.kill() + return True def _publish(self, event: dict[str, Any]) -> None: if self.events is not None: diff --git a/backend/pyproject.toml b/backend/pyproject.toml index d1914b4..57052b1 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -89,6 +89,9 @@ ignore = [ # Printing is what this one is about: node code is user code, and `print` is # how it says things. "tests/flow/test_logs.py" = ["ARG001", "T201"] +# This one takes its own directory off sys.path before the rest of its imports +# run, which is the whole point of doing it there. +"app/flow/worker_main.py" = ["E402"] [tool.ruff.lint.pyupgrade] # Preserve types, even if a file imports `from __future__ import annotations`. diff --git a/backend/tests/flow/test_modules.py b/backend/tests/flow/test_modules.py index 8983fe5..1b0a139 100644 --- a/backend/tests/flow/test_modules.py +++ b/backend/tests/flow/test_modules.py @@ -52,6 +52,21 @@ def test_a_manifest_already_applied_is_not_installed_again( assert (venv / ".applied").read_text() == marker +def test_a_missing_uv_is_a_failed_apply_rather_than_a_crash( + venv: Path, monkeypatch: pytest.MonkeyPatch +): + def no_uv(*args: object, **kwargs: object) -> None: + raise FileNotFoundError(2, "No such file or directory: 'uv'") + + monkeypatch.setattr(subprocess, "run", no_uv) + ok, output = modules.sync("") + + # The route turns this into a 400 with the output; an exception would be a + # 500 with nothing in it. + assert ok is False + assert "uv could not run" in output + + def test_a_manifest_that_does_not_resolve_leaves_the_venv_alone(venv: Path): modules.sync("") ok, output = modules.sync("fluksio-no-such-package-anywhere==9.9.9") diff --git a/backend/tests/flow/test_workers.py b/backend/tests/flow/test_workers.py index c6886ce..f82c066 100644 --- a/backend/tests/flow/test_workers.py +++ b/backend/tests/flow/test_workers.py @@ -107,3 +107,71 @@ def test_a_result_that_is_not_json_is_refused(pool): def test_compiling_reports_where_the_source_is_wrong(pool): assert pool.compile("demo", "broken", "def process(params)\n return {}\n") assert pool.compile("demo", "fine", "def process(params):\n return {}\n") is None + + +def test_a_node_imports_the_standard_library_not_the_engines_own_modules(pool): + # The worker script lives in app/flow, which holds queue.py, secrets.py and + # more; the interpreter would put that directory first on sys.path. + result = run( + pool, + "import queue\nimport secrets\n\n\n" + "def process(params):\n" + " return {'out': [queue.Queue().qsize(), len(secrets.token_hex(4))]}\n", + ) + assert result == {"out": [0, 8]} + + +def test_the_engines_secrets_are_not_in_a_workers_environment(pool, monkeypatch): + monkeypatch.setenv("SECRET_KEY", "not-for-nodes") + monkeypatch.setenv("POSTGRES_PASSWORD", "not-for-nodes") + monkeypatch.setenv("FLUKSIO_HARMLESS", "fine") + # A fresh process, so it is built from the environment set just now. + pool.respawn_all() + + result = run( + pool, + "import os\n\n\n" + "def process(params):\n" + " return {'out': [k for k in ('SECRET_KEY', 'POSTGRES_PASSWORD',\n" + " 'FLUKSIO_HARMLESS') if k in os.environ]}\n", + ) + assert result == {"out": ["FLUKSIO_HARMLESS"]} + + +def test_a_pool_can_stop_while_a_node_is_running(pool): + # One slot, taken by a node that will not finish on its own, and a second + # call queued behind it. The engine's node threads are not daemons, so a + # wait here is a shutdown that never completes. + outcomes: list[str] = [] + + def call(node: str) -> None: + try: + pool.run( + "demo", + node, + "import time\n\n\ndef process(params):\n time.sleep(60)\n", + {}, + {}, + f"demo.{node}", + timeout=60, + ) + outcomes.append("returned") + except Exception as exc: + outcomes.append(type(exc).__name__) + + busy = threading.Thread(target=call, args=("busy",)) + busy.start() + # Let the first one take the slot, so the second is blocked acquiring it. + time.sleep(1) + waiting = threading.Thread(target=call, args=("waiting",)) + waiting.start() + time.sleep(0.2) + + pool.stop() + for thread in (busy, waiting): + thread.join(timeout=10) + assert not thread.is_alive() + assert len(outcomes) == 2 + + with pytest.raises(Exception, match="shutting down"): + run(pool, "def process(params):\n return {'out': 1}\n") diff --git a/backend/tests/test_metrics.py b/backend/tests/test_metrics.py index 425af0f..d65c94d 100644 --- a/backend/tests/test_metrics.py +++ b/backend/tests/test_metrics.py @@ -126,6 +126,56 @@ def test_events_become_rollups_failures_runs_and_audit(db: Session) -> None: assert open_run.status == "running" +def test_a_flush_the_database_refused_is_written_by_the_next_one( + db: Session, monkeypatch +) -> None: + collector = MetricsCollector(EventBus()) + ts = datetime.now(timezone.utc).replace(second=0, microsecond=0).timestamp() + collector.handle( + { + "type": "audit", + "action": "held back", + "flow": FLOW, + "user": "held@example.com", + "ts": ts, + } + ) + + def refuse(*_args: object) -> None: + raise RuntimeError("the database is gone") + + monkeypatch.setattr(collector, "_write", refuse) + asyncio.run(collector.flush()) + monkeypatch.undo() + asyncio.run(collector.flush()) + + audit = db.exec( + select(EngineEvent).where(EngineEvent.actor == "held@example.com") + ).one() + assert audit.detail == "held back" + + +def test_a_node_id_wider_than_the_column_still_records(db: Session) -> None: + collector = MetricsCollector(EventBus()) + ts = datetime.now(timezone.utc).replace(second=0, microsecond=0).timestamp() + long_node = f"{FLOW}.{'w' * 400}" + collector.handle( + { + "type": "node_executed", + "flow": FLOW, + "node": long_node, + "duration_ms": 1.0, + "ts": ts, + } + ) + asyncio.run(collector.flush()) + + bucket = db.exec( + select(MetricBucket).where(MetricBucket.node == long_node[:255]) + ).one() + assert bucket.executions == 1 + + def test_a_run_that_did_not_fail_reads_ok(db: Session) -> None: collector = MetricsCollector(EventBus()) ts = time.time() diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index aafd885..dde6823 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -1308,7 +1308,7 @@ export const NodeDef_InputSchema = { } ], title: 'Timeout', - description: "Seconds this node's code may run before it is stopped. Above 60 the engine may deliver its work again while it is still running." + description: "Seconds this node's code may run before it is stopped. This covers the first call's imports, which can be much slower than the body. Above 60 the engine may deliver its work again while it is still running." } }, type: 'object', @@ -1381,7 +1381,7 @@ export const NodeDef_OutputSchema = { } ], title: 'Timeout', - description: "Seconds this node's code may run before it is stopped. Above 60 the engine may deliver its work again while it is still running." + description: "Seconds this node's code may run before it is stopped. This covers the first call's imports, which can be much slower than the body. Above 60 the engine may deliver its work again while it is still running." } }, type: 'object', diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index dde669b..aa4db08 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -440,7 +440,7 @@ export type NodeDef_Input = { provides?: Array; source_ref?: (string | null); /** - * Seconds this node's code may run before it is stopped. Above 60 the engine may deliver its work again while it is still running. + * Seconds this node's code may run before it is stopped. This covers the first call's imports, which can be much slower than the body. Above 60 the engine may deliver its work again while it is still running. */ timeout?: (number | null); }; @@ -460,7 +460,7 @@ export type NodeDef_Output = { provides?: Array; source_ref?: (string | null); /** - * Seconds this node's code may run before it is stopped. Above 60 the engine may deliver its work again while it is still running. + * Seconds this node's code may run before it is stopped. This covers the first call's imports, which can be much slower than the body. Above 60 the engine may deliver its work again while it is still running. */ timeout?: (number | null); }; diff --git a/frontend/src/components/Flow/liveStore.ts b/frontend/src/components/Flow/liveStore.ts index 3fa714cb0586aaf9986bb92912b1a53190bf8cf5..cb547663686ebd9d28e4dbe494c38fd095579262 100644 GIT binary patch delta 19 ZcmbQ@KGS`JgA7|tsR0mdc9aQb2LM6T1<3#a delta 14 VcmbQ~KE-{5gA60XW;dB&b^s-#1ULWy diff --git a/frontend/src/routes/_layout/health.tsx b/frontend/src/routes/_layout/health.tsx index 45a1941..48ef983 100644 --- a/frontend/src/routes/_layout/health.tsx +++ b/frontend/src/routes/_layout/health.tsx @@ -185,7 +185,11 @@ function Health() { // Something just went wrong on the socket. The row for it is written on the // collector's next flush, so the refetch waits that out rather than asking // for a failure the database does not have yet. - const seen = live.length + // + // The newest event's minute, not the count: the count stops changing once the + // ring is full, and a per-event key would let a flapping node restart the + // timer forever without it ever firing. + const seen = live.length ? Math.floor(live[live.length - 1].ts / 60) : 0 useEffect(() => { if (!seen) return const timer = setTimeout(() => {