Health summary: a node is failing after two failures in a row; one is a warning, a good run ends both

The summary read degraded for 15 minutes after any single node_error, even
once the node had run well again and including one bad input from a visitor.
The controller now counts a node's failures since it last ran well and notes
when it recovered; both carry across rebuilds and an acknowledgement resets
the count. Two in a row inside the 15 minute window is "failing" (a problem),
one is listed under warnings, and nodes.failing keeps meaning "failing now".

backend/fluksio/api/routes/observability.py

@@ -180,6 +180,14 @@ def _aware(when: datetime) -> datetime:
     return when if when.tzinfo else when.replace(tzinfo=UTC)
 
 
+#: How long after its last failure a node still counts as failing, and as
+#: having failed. A node that stopped being called is not kept in the summary
+#: forever, and a failure someone read is cleared with `acknowledge_error`.
+FAILING_WINDOW_S = 900.0
+#: Failures in a row, with no good run between, that make a node "failing".
+FAILING_AFTER = 2
+
+
 @router.get("/summary", response_model=HealthSummary)
 async def read_summary(request: Request, controller: FlowControllerDep) -> Any:
     """How the engine is doing right now. Always 200, degraded or not."""
@@ -215,14 +223,17 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any:
             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.
+    # would read as healthy. It reads degraded once it has failed twice running
+    # with no good run between, and only while the last failure is recent: one
+    # bad input, such as a visitor's, is a warning, and the next good run ends it.
     now = time.time()
-    failing = [
+    recent = [
         e
         for e in entries
-        if e.last_error_ts is not None and now - e.last_error_ts < 900
+        if e.last_error_ts is not None and now - e.last_error_ts < FAILING_WINDOW_S
     ]
+    failing = [e for e in recent if e.failures_in_a_row >= FAILING_AFTER]
+    failed_once = [e for e in recent if 0 < e.failures_in_a_row < FAILING_AFTER]
     if failing:
         problems.append(
             f"{len(failing)} node(s) failing: "
@@ -256,6 +267,11 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any:
         if degraded
         else []
     )
+    if failed_once:
+        warnings.append(
+            f"{len(failed_once)} node(s) failed: "
+            f"{', '.join(sorted(e.id for e in failed_once))}"
+        )
 
     return HealthSummary(
         status="degraded" if problems else "ok",

backend/fluksio/flow/controller.py

@@ -177,12 +177,26 @@ class LoadedNode:
     #: so only an acknowledgement clears it, not a good run and not a rebuild.
     last_error: str = ""
     last_error_ts: float | None = None
+    #: Failures since it last ran well, so one bad input reads differently from
+    #: a node that fails every time. The summary goes by this, not by the age
+    #: of `last_error_ts`.
+    failures_in_a_row: int = 0
+    #: When it last ran well after failing: the recovery the counter above
+    #: was reset by.
+    last_ok_ts: float | None = None
     #: Published, and with no body of its own — so what runs is the new-node
     #: template, which returns nothing at all.
     missing_source: bool = False
     #: The worker this node names although its type never runs on one.
     ignored_worker: str = ""
 
+    def inherit_failure(self, previous: LoadedNode) -> None:
+        """Keep what the node this one replaces last failed with."""
+        self.last_error = previous.last_error
+        self.last_error_ts = previous.last_error_ts
+        self.failures_in_a_row = previous.failures_in_a_row
+        self.last_ok_ts = previous.last_ok_ts
+
 
 @dataclass
 class RunContext:
@@ -597,13 +611,23 @@ class FlowController:
         async with self.events.subscribe() as queue:
             while True:
                 event = await queue.get()
-                if event.get("type") != "node_error":
+                kind = event.get("type")
+                if kind not in ("node_error", "node_executed"):
                     continue
                 entry = self.loaded.get(str(event.get("node") or ""))
                 if entry is None:
                     continue
+                ts = float(event.get("ts") or time.time())
+                if kind == "node_executed":
+                    # Almost every event here, so only a node with a failure
+                    # standing is touched: running well is how it recovers.
+                    if entry.failures_in_a_row:
+                        entry.failures_in_a_row = 0
+                        entry.last_ok_ts = ts
+                    continue
                 entry.last_error = str(event.get("error") or "")
-                entry.last_error_ts = float(event.get("ts") or time.time())
+                entry.last_error_ts = ts
+                entry.failures_in_a_row += 1
 
     def acknowledge_error(self, node_id: str) -> None:
         """Forget what a node's last failure was. The only thing that clears it.
@@ -616,6 +640,7 @@ class FlowController:
             raise KeyError(node_id)
         entry.last_error = ""
         entry.last_error_ts = None
+        entry.failures_in_a_row = 0
 
     async def set_enabled(self, flow: str, enabled: bool) -> None:
         """Stop or start one flow, without building anything.
@@ -721,8 +746,7 @@ class FlowController:
             for node_id, entry in loaded.items():
                 previous = self.loaded.get(node_id)
                 if previous is not None and previous.last_error:
-                    entry.last_error = previous.last_error
-                    entry.last_error_ts = previous.last_error_ts
+                    entry.inherit_failure(previous)
 
             self.loaded = loaded
             self.pipeline = Pipeline(
@@ -820,8 +844,7 @@ class FlowController:
                 for node_id, entry in loaded.items():
                     previous = self.loaded.get(node_id)
                     if previous is not None and previous.last_error:
-                        entry.last_error = previous.last_error
-                        entry.last_error_ts = previous.last_error_ts
+                        entry.inherit_failure(previous)
 
                 # Rebound rather than mutated: request threads and the failure
                 # watcher read these without holding anything.

backend/tests/api/routes/test_observability.py

@@ -456,7 +456,8 @@ def test_a_node_failing_on_every_call_makes_the_summary_degraded(
 ) -> 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.
+    Failing twice running makes the instance degraded, once is a warning, and
+    a failure that is no longer recent counts for neither.
     """
     import time
 
@@ -464,15 +465,28 @@ def test_a_node_failing_on_every_call_makes_the_summary_degraded(
 
     controller = client.app.state.flow_controller
     before = controller.loaded
+    now = time.time()
     controller.loaded = {
         "house.calc": LoadedNode(
-            id="house.calc", flow="house", last_error="boom", last_error_ts=time.time()
+            id="house.calc",
+            flow="house",
+            last_error="boom",
+            last_error_ts=now,
+            failures_in_a_row=2,
+        ),
+        "house.once": LoadedNode(
+            id="house.once",
+            flow="house",
+            last_error="boom",
+            last_error_ts=now,
+            failures_in_a_row=1,
         ),
         "house.old": LoadedNode(
             id="house.old",
             flow="house",
             last_error="boom",
-            last_error_ts=time.time() - 3600,
+            last_error_ts=now - 3600,
+            failures_in_a_row=5,
         ),
     }
     try:
@@ -483,7 +497,38 @@ def test_a_node_failing_on_every_call_makes_the_summary_degraded(
     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
+    assert "house.calc" in problem
+    assert "house.once" not in problem and "house.old" not in problem
+    (warning,) = [w for w in body["warnings"] if "failed" in w]
+    assert "house.once" in warning and "house.calc" not in warning
+
+
+def test_one_failure_is_a_warning_and_leaves_the_status_alone(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    import time
+
+    from fluksio.flow.controller import LoadedNode
+
+    controller = client.app.state.flow_controller
+    before = controller.loaded
+    controller.loaded = {
+        "house.once": LoadedNode(
+            id="house.once",
+            flow="house",
+            last_error="boom",
+            last_error_ts=time.time(),
+            failures_in_a_row=1,
+        )
+    }
+    try:
+        body = client.get(f"{PREFIX}/summary", headers=superuser_token_headers).json()
+    finally:
+        controller.loaded = before
+
+    assert body["status"] == "ok"
+    assert body["nodes"]["failing"] == 0
+    assert body["warnings"] == ["1 node(s) failed: house.once"]
 
 
 def test_a_superuser_can_flush_the_collector_now(

backend/tests/flow/test_node_failures.py

@@ -0,0 +1,58 @@
+"""A node's failures in a row, which the health summary reads."""
+
+import asyncio
+from pathlib import Path
+
+from fluksio.flow.controller import FlowController, LoadedNode
+from fluksio.flow.events import EventBus
+from fluksio.flow.store import FlowStore
+
+NODE = "house.calc"
+
+
+def _failure(ts: float) -> dict:
+    return {"type": "node_error", "node": NODE, "error": "boom", "ts": ts}
+
+
+def test_failures_count_until_the_node_runs_well(tmp_path: Path) -> None:
+    async def scenario() -> LoadedNode:
+        events = EventBus()
+        events.bind(asyncio.get_running_loop())
+        controller = FlowController(FlowStore(tmp_path / "flows"), events=events)
+        controller.loaded[NODE] = LoadedNode(id=NODE, flow="house")
+        watcher = asyncio.create_task(controller._watch_failures())
+        await asyncio.sleep(0)
+
+        events.publish(_failure(1.0))
+        events.publish(_failure(2.0))
+        await asyncio.sleep(0.05)
+        assert controller.loaded[NODE].failures_in_a_row == 2
+
+        events.publish({"type": "node_executed", "node": NODE, "ts": 3.0})
+        await asyncio.sleep(0.05)
+        watcher.cancel()
+        return controller.loaded[NODE]
+
+    entry = asyncio.run(scenario())
+
+    assert entry.failures_in_a_row == 0
+    assert entry.last_ok_ts == 3.0
+    # What it failed with stays for the canvas until someone acknowledges it.
+    assert entry.last_error == "boom"
+
+
+def test_a_rebuilt_node_keeps_its_failures() -> None:
+    old = LoadedNode(
+        id=NODE,
+        flow="house",
+        last_error="boom",
+        last_error_ts=5.0,
+        failures_in_a_row=3,
+        last_ok_ts=1.0,
+    )
+    new = LoadedNode(id=NODE, flow="house")
+
+    new.inherit_failure(old)
+
+    assert (new.last_error, new.last_error_ts) == ("boom", 5.0)
+    assert (new.failures_in_a_row, new.last_ok_ts) == (3, 1.0)

docs/interface/index.md

@@ -70,6 +70,11 @@ After you have used batch runs, it also shows running and queued runs.
 A degraded instance lists the problems that need attention. Warnings, such as
 a device that is offline, are listed in amber and leave the status as it is.
 
+A node that raises is a warning after its first failure and a problem when it
+fails twice in a row with no good run between. A good run ends either, and so
+does acknowledging the failure on the canvas. Without one, a failure counts for
+15 minutes after the last time it happened.
+
 ### Execution time per node type
 
 One bar per node type at its average time per run, slowest first, on a log