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;