From 3397739c1468bf45d6528ebd8dfefb557175e8e5 Mon Sep 17 00:00:00 2001 From: stroblme Date: Wed, 2 Sep 2026 22:37:44 +0200 Subject: [PATCH] Refuse a port the function cannot take, and read a failing node as degraded MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A python node's ports and settings arrive as keyword arguments, so a declared name its `process` does not take was a TypeError on every call — and a node that loads fine and fails every time it runs is the quiet kind of broken: the hosted demo did it 720 times an hour for two days and the health badge read ok throughout. `_build_node` now reads a written body with `ast` and refuses the mismatch at load, so the node is an error on the canvas and an issue on publish. Skipped for `**kwargs`, a decorated or absent `process`, and the template a new node opens with. The SDK's generated shim always takes `**settings`, so synced flows are untouched. `/observability/summary` names a node that has failed in the last fifteen minutes and reads degraded while it does, which is what would have made the badge amber. `nodes.failing` carries the count. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01SkgNtaR6JspnHBFFP6crZj --- backend/fluksio/api/routes/observability.py | 16 ++++ backend/fluksio/flow/controller.py | 50 +++++++++++ .../tests/api/routes/test_observability.py | 35 ++++++++ backend/tests/flow/test_node_settings.py | 87 +++++++++++++++++++ docs/index.md | 5 +- 5 files changed, 191 insertions(+), 2 deletions(-) diff --git a/backend/fluksio/api/routes/observability.py b/backend/fluksio/api/routes/observability.py index 75bcd86..27bf0c7 100644 --- a/backend/fluksio/api/routes/observability.py +++ b/backend/fluksio/api/routes/observability.py @@ -7,6 +7,7 @@ which the generated SDK turns into a thrown error — and a health page that cannot render while the engine is degraded is the wrong way round. """ +import time from datetime import UTC, datetime, timedelta from typing import Annotated, Any, Literal @@ -191,6 +192,20 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any: f"{len(unhealthy)} node(s) down: " f"{', '.join(sorted(e.id for e in unhealthy))}" ) + # A node that loads and then fails on every call is not in `errored`, and + # until this it read as healthy. ponytail: one failure reads degraded for + # 15 minutes; a counter with decay if that proves noisy. + now = time.time() + failing = [ + e + for e in entries + if e.last_error_ts is not None and now - e.last_error_ts < 900 + ] + if failing: + problems.append( + f"{len(failing)} node(s) failing: " + f"{', '.join(sorted(e.id for e in failing))}" + ) # What the canvas flags on a flow — a dependency loop, an input nothing # feeds — stops that flow running just as surely as a node that will not @@ -234,6 +249,7 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any: "total": len(entries), "error": len(errored), "unhealthy": len(unhealthy), + "failing": len(failing), }, queue=queue, loop_lag=( diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index fe9533a..e87584a 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -14,6 +14,7 @@ nodes are built like any other's, so `set_enabled` only starts or stops them. from __future__ import annotations +import ast import asyncio import hashlib import json @@ -1027,6 +1028,19 @@ class FlowController: flow, node_def.id ) + # Only a body somebody wrote: the template a new node opens + # with takes nothing, and a node declared before it is written + # is reported as missing its source rather than as wrong. + written = bool(node_def.source_ref) or self.store.has_node_source( + flow, node_def.id, draft=draft + ) + if written: + gap = _signature_gap( + code, node_def.id, ports, set(params) - RESERVED_SETTINGS + ) + if gap: + raise ValueError(gap) + # What a node produces before it returns comes back as frames; # this puts them through the node's own ports. emissions = EmitSink() @@ -1716,6 +1730,42 @@ def _bound(specs: list[MessageSpec]) -> list[MessageSpec]: return [spec for spec in specs if spec.name] +def _signature_gap( + code: str, node_id: str, ports: set[str], settings: set[str] +) -> str | None: + """A declared port or setting ``process`` cannot accept, or None. + + Ports and settings both arrive as keyword arguments, so a name the function + does not take is a ``TypeError`` on every call — and a node that loads fine + and fails every time it runs is the quiet kind of broken. Read statically: + with a worker pool the code is never imported here. Anything this cannot + read for certain (no plain ``def process``, a decorator, ``**kwargs``, a + syntax error the compile step words better) is left to the call. + """ + try: + tree = ast.parse(code) + except SyntaxError: + return None + defs = [ + node + for node in tree.body + if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef) + and node.name == "process" + ] + if not defs or defs[-1].decorator_list or defs[-1].args.kwarg is not None: + return None + args = defs[-1].args + names = [a.arg for a in [*args.posonlyargs, *args.args, *args.kwonlyargs]] + missing = sorted((ports | settings) - set(names)) + if not missing: + return None + kind = "an input" if missing[0] in ports else "a setting" + return ( + f"'{missing[0]}' is {kind} of '{node_id}' but " + f"process({', '.join(names)}) takes no such argument" + ) + + def with_settings( function: Callable[..., Any], params: dict[str, Any] ) -> Callable[..., Any]: diff --git a/backend/tests/api/routes/test_observability.py b/backend/tests/api/routes/test_observability.py index c2834f3..6e8f963 100644 --- a/backend/tests/api/routes/test_observability.py +++ b/backend/tests/api/routes/test_observability.py @@ -447,3 +447,38 @@ def test_a_timeseries_window_can_be_named( assert response.status_code == 200 # Two of the four minutes: `since` inclusive, `until` exclusive. assert sum(point["executions"] for point in response.json()) == 2 + + +def test_a_node_failing_on_every_call_makes_the_summary_degraded( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """A node that loads and then raises each time it runs is not in `error`. + + It is named while its last failure is recent, and drops out once it is not. + """ + import time + + from fluksio.flow.controller import LoadedNode + + controller = client.app.state.flow_controller + before = controller.loaded + controller.loaded = { + "house.calc": LoadedNode( + id="house.calc", flow="house", last_error="boom", last_error_ts=time.time() + ), + "house.old": LoadedNode( + id="house.old", + flow="house", + last_error="boom", + last_error_ts=time.time() - 3600, + ), + } + try: + body = client.get(f"{PREFIX}/summary", headers=superuser_token_headers).json() + finally: + controller.loaded = before + + assert body["status"] == "degraded" + assert body["nodes"]["failing"] == 1 + (problem,) = [p for p in body["problems"] if "failing" in p] + assert "house.calc" in problem and "house.old" not in problem diff --git a/backend/tests/flow/test_node_settings.py b/backend/tests/flow/test_node_settings.py index 2dabc91..6fabb30 100644 --- a/backend/tests/flow/test_node_settings.py +++ b/backend/tests/flow/test_node_settings.py @@ -61,3 +61,90 @@ def test_a_setting_named_after_a_port_is_refused(store: FlowStore): assert [node.status for node in preview.nodes] == ["error"] assert "both an input and a setting" in (preview.nodes[0].error or "") + + +def _flow(requires: list[MessageSpec], **params: object) -> FlowDef: + return FlowDef( + name="house", + nodes=[ + NodeDef( + id="scale", + params=dict(params), + requires=requires, + provides=[MessageSpec(name="scaled", dtype=DType.FLOAT)], + ) + ], + ) + + +READING = [MessageSpec(name="reading", dtype=DType.FLOAT)] + + +def _preview(store: FlowStore, flow: FlowDef, source: str | None): + store.write_draft(flow, 0) + if source is not None: + store.write_node_source("house", "scale", source, draft=True) + return FlowController(store).preview("house").nodes[0] + + +def test_a_port_the_function_cannot_take_is_refused_at_load(store: FlowStore): + node = _preview(store, _flow(READING), "def process(value):\n return {}\n") + + assert node.status == "error" + assert "'reading' is an input of 'scale'" in (node.error or "") + assert "process(value)" in (node.error or "") + + +def test_a_setting_the_function_cannot_take_is_refused_at_load(store: FlowStore): + node = _preview(store, _flow(READING, gain=2), SOURCE) + + assert node.status == "error" + assert "'gain' is a setting of 'scale'" in (node.error or "") + + +def test_a_function_taking_keywords_accepts_anything(store: FlowStore): + node = _preview( + store, _flow(READING, gain=2), "def process(**kw):\n return {}\n" + ) + + assert node.status == "active" + + +def test_a_node_never_written_is_not_wrong_yet(store: FlowStore): + # The template a new node opens with takes nothing; that is missing, not a + # mismatch. + node = _preview(store, _flow(READING), None) + + assert node.status == "active" + + +def test_a_port_arrives_under_its_own_name(store: FlowStore): + other = [MessageSpec(name="other.reading", dtype=DType.FLOAT)] + node = _preview(store, _flow(other, factor=3), SOURCE) + + assert node.status == "active" + + +def test_a_shared_body_is_checked_against_each_user(store: FlowStore): + store.write_flow(_flow(READING, factor=3)) + store.write_node_source("house", "scale", SOURCE) + store.share_node("house", "scale", "scale_it") + store.write_flow( + FlowDef( + name="garden", + nodes=[ + NodeDef( + id="scale", + source_ref="scale_it", + requires=[MessageSpec(name="level", dtype=DType.FLOAT)], + provides=[MessageSpec(name="scaled", dtype=DType.FLOAT)], + ) + ], + ) + ) + controller = FlowController(store) + + nodes = controller.preview("garden").nodes + + assert [n.status for n in nodes] == ["error"] + assert "'level' is an input of 'scale'" in (nodes[0].error or "") diff --git a/docs/index.md b/docs/index.md index ace9211..3f82d33 100644 --- a/docs/index.md +++ b/docs/index.md @@ -35,8 +35,9 @@ API. package, the other stands up a server. Everything after that is shared. To look before installing, the hosted demo at -[fluksio.com](https://fluksio.com) runs a real instance with a small-house panel -and a training pipeline on it. +[fluksio.com](https://fluksio.com) runs a real instance: a small-house panel, a +media screen, and a lab where a training pipeline runs and its runs are +compared, four screens on one panel. ## Where things are