From 70e542ec3cb53c988ecec212dfb3b5e803943e13 Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 28 Aug 2026 12:22:55 +0200 Subject: [PATCH] Surface a failing connector poll as node health and a flow issue The poll loop remembered what it read rather than what it published, so a value the node could not publish counted as said: the next poll skipped it, succeeded, and health went back to ok with the port still dark. Remember it only after inject returns, and report ok last. A node reporting itself down is now derived into its flow's issues on read and counted on the health summary, so the canvas marks it and Home says so. Being down does not stop the flow, and the issue clears by itself when the node reports well again. The repeating poll warning is logged once per outage rather than once per tick. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01K1moruzue2kTJd3uVisgNk --- backend/fluksio/api/routes/observability.py | 12 +++++- backend/fluksio/flow/connector.py | 23 +++++++--- backend/fluksio/flow/controller.py | 25 ++++++++++- backend/fluksio/flow/pipeline.py | 1 + .../tests/api/routes/test_observability.py | 31 ++++++++++++++ backend/tests/flow/test_connector.py | 34 +++++++++++++++ .../tests/flow/test_unhealthy_node_issue.py | 42 +++++++++++++++++++ docs/concepts/flows.md | 9 ++-- docs/interface/flow-editor.md | 7 +++- docs/reference/connector-contract.md | 11 +++-- frontend/src/client/schemas.gen.ts | 2 +- frontend/src/client/types.gen.ts | 4 +- 12 files changed, 182 insertions(+), 19 deletions(-) create mode 100644 backend/tests/flow/test_unhealthy_node_issue.py diff --git a/backend/fluksio/api/routes/observability.py b/backend/fluksio/api/routes/observability.py index f578563..54b6176 100644 --- a/backend/fluksio/api/routes/observability.py +++ b/backend/fluksio/api/routes/observability.py @@ -176,10 +176,16 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any: paused = set(controller.paused_flows()) entries = list(controller.loaded.values()) errored = [e for e in entries if e.status is NodeStatus.ERROR] + unhealthy = [e for e in entries if e.health == "down"] if quarantined: problems.append(f"{len(quarantined)} flow(s) quarantined") if errored: problems.append(f"{len(errored)} node(s) failed to load") + if unhealthy: + problems.append( + f"{len(unhealthy)} node(s) down: " + f"{', '.join(sorted(e.id for e in unhealthy))}" + ) # 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 @@ -219,7 +225,11 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any: "quarantined": len(quarantined), "invalid": len(invalid), }, - nodes={"total": len(entries), "error": len(errored)}, + nodes={ + "total": len(entries), + "error": len(errored), + "unhealthy": len(unhealthy), + }, queue=queue, loop_lag=( watchdog.snapshot() diff --git a/backend/fluksio/flow/connector.py b/backend/fluksio/flow/connector.py index 18f2e77..1cfe897 100644 --- a/backend/fluksio/flow/connector.py +++ b/backend/fluksio/flow/connector.py @@ -86,7 +86,14 @@ class ConnectorNode(Node): description="Seconds between polls; 0 polls never.", ) - __slots__ = ("config", "_poll_task", "_stop_event", "_last_published", "_artifacts") + __slots__ = ( + "config", + "_poll_task", + "_stop_event", + "_last_published", + "_artifacts", + "_down", + ) def __init__(self, **kwargs: Any) -> None: super().__init__(f=self._dispatch, **kwargs) @@ -95,6 +102,7 @@ class ConnectorNode(Node): self._stop_event: asyncio.Event | None = None self._last_published: dict[str, Any] = {} self._artifacts: ArtifactStore | None = None + self._down = False def _dispatch(self, params: dict[str, Any], **ports: Any) -> dict[str, Any] | None: """The scheduler's entry point. Settings are already on ``self.config``.""" @@ -171,24 +179,29 @@ class ConnectorNode(Node): Only changed ports are published: a device polled every few seconds is usually saying the same thing, and every publication wakes everything - downstream of it. + downstream of it. What is remembered is what was *published*, not what + the poll returned — a publication that raised is retried next tick + rather than counting as said. """ while not (self._stop_event and self._stop_event.is_set()): try: values = await self.poll() - self.report_health("ok") changed = { port: value for port, value in (values or {}).items() if self._last_published.get(port, object()) != value } if changed: - self._last_published.update(changed) # inject runs the graph, which is blocking work. await asyncio.to_thread(self.inject, changed) + self._last_published.update(changed) + self.report_health("ok") + self._down = False except asyncio.CancelledError: break except Exception as exc: - logger.warning("Connector '%s' failed to poll: %s", self.id, exc) + if not self._down: + logger.warning("Connector '%s' failed to poll: %s", self.id, exc) + self._down = True self.report_health("down", f"{type(exc).__name__}: {exc}") await asyncio.sleep(self.config.poll_interval) diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index d883416..e05d8d9 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -1224,8 +1224,31 @@ class FlowController: } ) + def _health_issues(self, flow: str | None = None) -> list[ValidationIssue]: + """Nodes that are running but not working, as issues on their flow. + + Not part of `self.issues`: that list is what a build found, and this is + what is happening now. Derived on read, so a node reporting itself well + again clears it with nothing to remember. + """ + return [ + ValidationIssue( + code="node_unhealthy", + message=( + f"Node '{entry.id.rpartition('.')[2]}' is down: " + f"{entry.health_detail or 'no detail given'}" + ), + flow=entry.flow, + node=entry.id, + ) + for entry in self.loaded.values() + if entry.health == "down" and (flow is None or entry.flow == flow) + ] + def flow_issues(self, flow: str) -> list[ValidationIssue]: - return [issue for issue in self.issues if not issue.flow or issue.flow == flow] + return [ + issue for issue in self.issues if not issue.flow or issue.flow == flow + ] + self._health_issues(flow) def preview(self, name: str) -> Preview: """Build a flow's unpublished draft without deploying it. diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index be80ea0..a2c82d8 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -59,6 +59,7 @@ class ValidationIssue(BaseModel): "unauthenticated_hook", "self_loop_needs_initial", "missing_source", + "node_unhealthy", ] message: str flow: str = "" diff --git a/backend/tests/api/routes/test_observability.py b/backend/tests/api/routes/test_observability.py index 010714f..088b7c6 100644 --- a/backend/tests/api/routes/test_observability.py +++ b/backend/tests/api/routes/test_observability.py @@ -127,6 +127,37 @@ def test_a_flow_that_cannot_run_makes_the_summary_degraded( assert not any("hooky" in problem for problem in body["problems"]) +def test_a_down_node_makes_the_summary_degraded( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """A connector that cannot reach its device is not a flow that cannot run. + + It is counted on its own, so the flow keeps running and "invalid" stays + about validation. + """ + from fluksio.flow.controller import LoadedNode + + controller = client.app.state.flow_controller + before = controller.loaded + controller.loaded = { + "house.owm": LoadedNode( + id="house.owm", + flow="house", + health="down", + health_detail="ConnectionError: name resolution failed", + ) + } + try: + body = client.get(f"{PREFIX}/summary", headers=superuser_token_headers).json() + finally: + controller.loaded = before + + assert body["status"] == "degraded" + assert body["nodes"]["unhealthy"] == 1 + assert any("down" in problem for problem in body["problems"]) + assert body["flows"]["invalid"] == 0 + + def test_the_history_reads_back( client: TestClient, superuser_token_headers: dict[str, str], db: Session ) -> None: diff --git a/backend/tests/flow/test_connector.py b/backend/tests/flow/test_connector.py index ed44191..4abde58 100644 --- a/backend/tests/flow/test_connector.py +++ b/backend/tests/flow/test_connector.py @@ -104,6 +104,40 @@ def test_a_failing_poll_reports_down_and_keeps_going(): assert health[-1][0] == "ok" +def test_an_undeclared_port_keeps_failing_until_the_node_declares_it(): + """A publication that raised is retried, not remembered as published. + + The loop remembers what it published. If it remembered what it read, a + value the node cannot publish would be skipped on the next poll, the poll + would succeed, and the node would go back to reporting itself healthy with + its port still dark. + """ + + class Chatty(Sensor): + """Reads a port it never declared.""" + + async def poll(self) -> dict[str, Any]: + self.polls += 1 + return {"reading": 21.5, "lat": 48.1} + + health: list[tuple[str, str | None]] = [] + node = Chatty( + provides=[MessageSpec(name="reading", dtype=DType.FLOAT)], + params={"poll_interval": 0.01}, + ) + node.assign_flow("demo", "sensor") + node._on_health = lambda _node, status, detail: health.append((status, detail)) + pipeline = Pipeline(nodes=[node]) + + run_briefly(node) + + assert pipeline.state.get("demo.reading") is None + assert health[-1][0] == "down" + assert "NodeOutputError" in (health[-1][1] or "") + # Still failing on the last poll, not just the first. + assert len([entry for entry in health if entry[0] == "down"]) > 1 + + class Actuator(ConnectorNode): """A connector that commands something instead of reading it.""" diff --git a/backend/tests/flow/test_unhealthy_node_issue.py b/backend/tests/flow/test_unhealthy_node_issue.py new file mode 100644 index 0000000..45e44fa --- /dev/null +++ b/backend/tests/flow/test_unhealthy_node_issue.py @@ -0,0 +1,42 @@ +"""A node that loaded but is not working shows up as an issue on its flow. + +Health used to go nowhere: the connector reported it, the controller stored it, +and no screen ever asked. These cover the derivation that closes that gap. +""" + +from pathlib import Path + +from fluksio.flow.controller import FlowController, LoadedNode +from fluksio.flow.store import FlowStore + + +def a_controller(tmp_path: Path) -> FlowController: + controller = FlowController(FlowStore(tmp_path / "flows")) + controller.loaded["house.owm"] = LoadedNode( + id="house.owm", + flow="house", + health="down", + health_detail="ConnectionError: name resolution failed", + ) + return controller + + +def test_a_down_node_is_an_issue_on_its_flow(tmp_path: Path) -> None: + controller = a_controller(tmp_path) + + issues = controller.flow_issues("house") + + assert [issue.code for issue in issues] == ["node_unhealthy"] + assert issues[0].node == "house.owm" + assert "name resolution failed" in issues[0].message + # Not advisory: the canvas has to mark the node. + assert not issues[0].advisory + assert controller.flow_issues("other") == [] + + +def test_the_issue_clears_when_the_node_reports_itself_well(tmp_path: Path) -> None: + controller = a_controller(tmp_path) + + controller.loaded["house.owm"].health = "ok" + + assert controller.flow_issues("house") == [] diff --git a/docs/concepts/flows.md b/docs/concepts/flows.md index a73db87..9aae75f 100644 --- a/docs/concepts/flows.md +++ b/docs/concepts/flows.md @@ -163,10 +163,13 @@ to: | `self_loop_needs_initial` | a node reads a message it also writes, with no starting value | | `node_error` | the node's code did not load: a syntax error, a missing import | | `unauthenticated_hook` | advisory — a webhook with no shared secret is open to anyone | +| `node_unhealthy` | the node loaded but is not working: a connector that cannot reach its device, or whose last publication failed | -A flow with any of these except the advisory one does not run. The health -summary on Home counts them, so "why is nothing happening?" has an answer that -does not involve reading logs. +A flow with any of these except the advisory one and `node_unhealthy` does not +run — a node reporting itself down is a live condition, not a build error, so +the rest of the flow keeps going and the issue clears by itself once the node +reports well again. The health summary on Home counts them, so "why is nothing +happening?" has an answer that does not involve reading logs. ## What happens at runtime diff --git a/docs/interface/flow-editor.md b/docs/interface/flow-editor.md index 675a091..1536341 100644 --- a/docs/interface/flow-editor.md +++ b/docs/interface/flow-editor.md @@ -146,9 +146,12 @@ The canvas validates as you edit and marks the node each issue belongs to: - a node reading a message it also writes, with nothing to start it from - code that did not load - a webhook with no shared secret (advisory — it does not stop the flow) +- a node that loaded but reports itself down, such as a connector that cannot + reach its device -A flow with any of these except the last does not run, and the health summary -on Home counts it. +A flow with any of these except the last two does not run, and the health +summary on Home counts it. The last one clears on its own once the node reports +itself well again. ## See also diff --git a/docs/reference/connector-contract.md b/docs/reference/connector-contract.md index 93cfec4..288062a 100644 --- a/docs/reference/connector-contract.md +++ b/docs/reference/connector-contract.md @@ -131,7 +131,9 @@ async def poll(self) -> dict[str, Any] | None: - Return `None` when there is nothing new. - **Only changed values are published.** A device polled every few seconds usually says the same thing, and every publication wakes everything - downstream, so the loop compares against what it last published. + downstream, so the loop compares against what it last published — what it + actually published, so a publication that failed is retried next tick rather + than counting as said. - Raising is not fatal: it is reported as a health problem and retried on the next tick. - The loop calls `inject`, which runs the graph, on a worker thread. `poll()` @@ -213,9 +215,10 @@ self.report_health("degraded", "3 of 5 registers timed out") self.report_health("down", str(exc)) ``` -Three values, `ok`, `degraded` and `down`, plus an optional detail string. The -engine forwards changes to the editor, which shows them on the node. Reporting -the same status twice is free — only changes are published. The polling loop +Three values, `ok`, `degraded` and `down`, plus an optional detail string. +Reporting the same status twice is free — only changes are published. A node +reporting `down` is named among its flow's issues and counted on the health +summary on Home; `degraded` means still working, and is not. The polling loop already reports around `poll()`; a connector managing its own connection should report when it connects and when it loses the connection. diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 93974ba..f2181a6 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -3561,7 +3561,7 @@ export const ValidationIssueSchema = { properties: { code: { type: 'string', - enum: ['cycle', 'unconnected_input', 'missing_initial_value', 'node_error', 'unauthenticated_hook', 'self_loop_needs_initial', 'missing_source'], + enum: ['cycle', 'unconnected_input', 'missing_initial_value', 'node_error', 'unauthenticated_hook', 'self_loop_needs_initial', 'missing_source', 'node_unhealthy'], title: 'Code' }, message: { diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 2069fc4..05b0a4d 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -1207,7 +1207,7 @@ export type ValidationError = { * Something wrong with a flow — a fault, or merely advisory. */ export type ValidationIssue = { - code: 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source'; + code: 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source' | 'node_unhealthy'; message: string; flow?: string; nodes?: Array<(string)>; @@ -1220,7 +1220,7 @@ export type ValidationIssue = { readonly advisory: boolean; }; -export type code = 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source'; +export type code = 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source' | 'node_unhealthy'; export type ValidationResult = { issues?: Array;