From 2554488a734bf4b3230578213b58adbf6654376e Mon Sep 17 00:00:00 2001 From: stroblme Date: Mon, 17 Aug 2026 11:33:25 +0200 Subject: [PATCH] Backend: real alert test results, state cleanup on delete/rename, node trigger errors, queue and collector fixes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `AlertManager.send` swallowed every delivery failure, so the alerts screen's Test button answered 200 whatever happened — the one thing it exists for. It takes `raise_on_error` now, which only the test route passes; the per-channel loop keeps the swallow, because one dead channel must not stop the others hearing about the same fault. A refused delivery answers 502 with whatever the sender said. Renaming a flow left its values under the old name for good: the delete path already swept them, the rename path never did. It calls the same `forget_flow`, which covers the messages and the `__ts__`/`__version__`/`__history__` bookkeeping keyed by message name. Cleanup, not migration — they repopulate under the new name on the next run. Triggering a node by hand ran `Node.__call__` with nothing catching it, so a node that raised produced a 500 and a stack trace in the server log, and nothing at all on the canvas. `Pipeline.publish_error` is the reporting half of `_execute_node` lifted out; both paths go through it, so a manual failure now reads the same on the canvas and in the metrics as a queued one. The route answers 400 with the node's error. `MemoryWorkQueue.stats()` counts claimed-but-unacknowledged work rather than reporting zero, so the health tile means something without Redis. The metrics collector's held tracebacks are capped at `DETAIL_CAP` and swept on the same `RUN_STALE_S` cutoff the open runs use, instead of one untruncated traceback per node kept for the life of the process — a traceback still survives the flush between the log and the failure it belongs to. `GET /observability/events` takes `since`/`until`, the window `/runs` already took, so a failures list can cover the span the charts beside it are drawn from. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01XC2jX6Hdj7pxGGKzBTrbqB --- backend/app/api/routes/alerts.py | 22 ++++--- backend/app/api/routes/flows.py | 10 ++- backend/app/api/routes/observability.py | 12 +++- backend/app/flow/alerts.py | 15 ++++- backend/app/flow/controller.py | 20 ++++-- backend/app/flow/metrics.py | 15 +++-- backend/app/flow/pipeline.py | 46 +++++++++---- backend/app/flow/queue.py | 14 +++- backend/tests/api/routes/test_alerts.py | 66 +++++++++++++++++++ backend/tests/api/routes/test_flows.py | 66 +++++++++++++++++++ .../tests/api/routes/test_observability.py | 32 +++++++++ backend/tests/flow/test_queue.py | 14 ++++ backend/tests/test_metrics.py | 29 +++++++- 13 files changed, 322 insertions(+), 39 deletions(-) create mode 100644 backend/tests/api/routes/test_alerts.py diff --git a/backend/app/api/routes/alerts.py b/backend/app/api/routes/alerts.py index d5a5ff4..0293fc3 100644 --- a/backend/app/api/routes/alerts.py +++ b/backend/app/api/routes/alerts.py @@ -74,12 +74,18 @@ async def test_channel(channel_name: str, controller: FlowControllerDep) -> Any: if channel is None: raise HTTPException(status_code=404, detail=f"No channel '{channel_name}'") - await controller.alerts.send( - channel, - Alert( - title="Fluksio test alert", - body="If you are reading this, the channel works.", - severity="warning", - ), - ) + try: + await controller.alerts.send( + channel, + Alert( + title="Fluksio test alert", + body="If you are reading this, the channel works.", + severity="warning", + ), + raise_on_error=True, + ) + except Exception as exc: + # Whatever the sender said, verbatim: it is the only clue the operator + # has about why the channel does not work. + raise HTTPException(status_code=502, detail=str(exc) or type(exc).__name__) return Message(message=f"Sent a test alert through '{channel_name}'") diff --git a/backend/app/api/routes/flows.py b/backend/app/api/routes/flows.py index bb23521..4aa7c70 100644 --- a/backend/app/api/routes/flows.py +++ b/backend/app/api/routes/flows.py @@ -432,6 +432,10 @@ async def rename_flow( except FlowExists as exc: raise HTTPException(status_code=409, detail=str(exc)) + # The old name is nobody's namespace now; its values would sit there under a + # flow that no longer exists. They repopulate under the new name on the next + # run, so this is cleanup rather than a migration. + await run_in_threadpool(controller.forget_flow, name) await controller.reload() return _detail(controller, renamed) @@ -633,13 +637,17 @@ async def trigger_node( """Feed values into a single node.""" _require_enabled(controller, name) try: - await run_in_threadpool( + error = await run_in_threadpool( controller.trigger_node, f"{name}.{node_id}", body.values ) except KeyError: raise HTTPException( status_code=404, detail=f"No node named '{node_id}' in flow '{name}'" ) + if error: + # The canvas already has the failure from the bus; the person who + # clicked gets to read it too, instead of a stack trace in the log. + raise HTTPException(status_code=400, detail=error) return _flow_state(controller, name) diff --git a/backend/app/api/routes/observability.py b/backend/app/api/routes/observability.py index ff7e55e..c60eace 100644 --- a/backend/app/api/routes/observability.py +++ b/backend/app/api/routes/observability.py @@ -306,9 +306,15 @@ def read_events( session: SessionDep, kind: Literal["failure", "audit"] = "failure", flow: str | None = None, + since: datetime | None = None, + until: datetime | None = None, limit: int = 100, ) -> Any: - """What went wrong, or who changed what. Newest first.""" + """What went wrong, or who changed what. Newest first. + + ``since`` is inclusive and ``until`` exclusive, the same window ``/runs`` + takes, so a list can cover the span the charts beside it are drawn from. + """ statement = select(EngineEvent).order_by(col(EngineEvent.ts).desc()) if kind == "audit": statement = statement.where(col(EngineEvent.type) == "audit") @@ -316,6 +322,10 @@ def read_events( statement = statement.where(col(EngineEvent.type) != "audit") if flow: statement = statement.where(col(EngineEvent.flow) == flow) + if since: + statement = statement.where(col(EngineEvent.ts) >= _aware(since)) + if until: + statement = statement.where(col(EngineEvent.ts) < _aware(until)) return list(session.exec(statement.limit(min(limit, 500)))) diff --git a/backend/app/flow/alerts.py b/backend/app/flow/alerts.py index 11c42bc..3390168 100644 --- a/backend/app/flow/alerts.py +++ b/backend/app/flow/alerts.py @@ -271,10 +271,21 @@ class AlertManager: # Delivery # ------------------------------------------------------------------------- - async def send(self, channel: Channel, alert: Alert) -> None: + async def send( + self, channel: Channel, alert: Alert, raise_on_error: bool = False + ) -> None: + """Deliver one alert. + + A failure is logged and swallowed, because one dead channel must not + stop the others hearing about the same fault. The test button passes + ``raise_on_error``: telling a working channel from a broken one is the + only thing it exists for. + """ try: config = resolve_params(channel.config) except Exception as exc: + if raise_on_error: + raise logger.error("Channel '%s' has unusable settings: %s", channel.name, exc) return @@ -286,6 +297,8 @@ class AlertManager: else: await self._send_email(config, alert) except Exception as exc: + if raise_on_error: + raise logger.error("Could not alert through '%s': %s", channel.name, exc) async def _send_ntfy(self, config: dict[str, Any], alert: Alert) -> None: diff --git a/backend/app/flow/controller.py b/backend/app/flow/controller.py index 00ce655..24ff71b 100644 --- a/backend/app/flow/controller.py +++ b/backend/app/flow/controller.py @@ -836,19 +836,29 @@ class FlowController: ) pipeline.run(inputs or {}) - def trigger_node(self, node_id: str, values: dict[str, Any] | None = None) -> None: + def trigger_node(self, node_id: str, values: dict[str, Any] | None = None) -> str: """Feed values into one node. Blocking — call from a worker thread. Runs here rather than through the queue: the caller is a person waiting on the response, and wants the state it produced. + + Returns what the node raised, or an empty string. A node failing by hand + is reported the way a queued run reports it — on the canvas and in the + metrics — rather than as a stack trace in the server log. """ node = self.get_node(node_id) if node is None: raise KeyError(node_id) - if node.requires and values: - node.trigger(values, durable=False) - else: - node.inject(values or {}, durable=False) + try: + if node.requires and values: + node.trigger(values, durable=False) + else: + node.inject(values or {}, durable=False) + except Exception as exc: + if self.pipeline is None: + raise + return self.pipeline.publish_error(node, exc) + return "" def _publish(self, event: dict[str, Any]) -> None: if self.events is not None: diff --git a/backend/app/flow/metrics.py b/backend/app/flow/metrics.py index 035c3ff..db8a31d 100644 --- a/backend/app/flow/metrics.py +++ b/backend/app/flow/metrics.py @@ -87,8 +87,9 @@ class MetricsCollector: self._buckets: dict[tuple[str, str, datetime], dict[str, float]] = {} self._runs: dict[str, dict[str, Any]] = {} self._pending: list[EngineEvent] = [] - # The traceback arrives one event before the failure it belongs to. - self._tracebacks: dict[tuple[str, str], str] = {} + # The traceback arrives one event before the failure it belongs to, + # held with the time it arrived so an unpaired one does not stay put. + self._tracebacks: dict[tuple[str, str], tuple[float, str]] = {} self._last_prune = 0.0 # ------------------------------------------------------------------------- @@ -171,7 +172,7 @@ class MetricsCollector: # Held for the node_error that follows it from the same thread. if event.get("level") == "error": key = (str(event.get("flow") or ""), str(event.get("node") or "")) - self._tracebacks[key] = str(event.get("text") or "") + self._tracebacks[key] = (ts, str(event.get("text") or "")[:DETAIL_CAP]) return if kind == "node_error": @@ -179,7 +180,7 @@ class MetricsCollector: if run is not None: run["errors"] += 1 key = (str(event.get("flow") or ""), str(event.get("node") or "")) - traceback = self._tracebacks.pop(key, "") + traceback = self._tracebacks.pop(key, (0.0, ""))[1] error = str(event.get("error") or "") self._pending.append( EngineEvent( @@ -293,7 +294,11 @@ class MetricsCollector: del self._runs[run_id] # Held tracebacks survive the flush: the log and the failure it belongs # to are two events, and a flush can fall between them. One per node, - # each replaced by that node's next failure. + # each replaced by that node's next failure — and dropped once it is + # clear the failure it was waiting for is never coming. + for key, (held, _) in list(self._tracebacks.items()): + if held < cutoff: + del self._tracebacks[key] def _hold( self, diff --git a/backend/app/flow/pipeline.py b/backend/app/flow/pipeline.py index a38d389..d3c53dd 100644 --- a/backend/app/flow/pipeline.py +++ b/backend/app/flow/pipeline.py @@ -569,6 +569,37 @@ class Pipeline: } ) + def publish_error( + self, + node: Node, + exc: Exception, + collected: logs.Collector | None = None, + entry_id: str = "", + ) -> str: + """Report a node failure, and hand the one-line version back. + + Call it from an ``except`` block — the traceback comes from the + exception being handled. Every way of running a node reports through + here, so a manual run reads the same on the canvas and in the metrics + as a queued one. + """ + logger.exception("Node '%s' failed", node.id) + # The one-line error goes on the node; the traceback goes to the log + # panel, which is where there is room to read it. + self.publish_log(node, collected or logs.Collector(), logs.node_traceback()) + error = f"{type(exc).__name__}: {exc}" + self._publish( + { + "type": "node_error", + "flow": node.flow, + "node": node.id, + "error": error, + "run": entry_id, + "ts": time.time(), + } + ) + return error + def _already_done(self, entry_id: str, node: Node) -> bool: """Did this node's side effect already happen for this work item?""" if node.idempotent or self._queue is None: @@ -650,20 +681,7 @@ class Pipeline: return result except Exception as exc: # One failing node must not take the rest of the graph down. - logger.exception("Node '%s' failed", node.id) - # The one-line error goes on the node; the traceback goes to the - # log panel, which is where there is room to read it. - self.publish_log(node, collected, logs.node_traceback()) - self._publish( - { - "type": "node_error", - "flow": node.flow, - "node": node.id, - "error": f"{type(exc).__name__}: {exc}", - "run": entry_id, - "ts": time.time(), - } - ) + self.publish_error(node, exc, collected, entry_id) return None # ------------------------------------------------------------------------- diff --git a/backend/app/flow/queue.py b/backend/app/flow/queue.py index cb1f781..ad3c468 100644 --- a/backend/app/flow/queue.py +++ b/backend/app/flow/queue.py @@ -177,6 +177,8 @@ class MemoryWorkQueue(WorkQueue): self._delayed: list[tuple[float, int, WorkItem]] = [] self._parked: dict[str, list[WorkItem]] = {} self._done: set[tuple[str, str]] = set() + # Claimed and not yet acknowledged, which is what Redis's `pending` is. + self._in_flight = 0 self._counter = 0 self._seq = 0 self._lock = threading.Lock() @@ -210,10 +212,16 @@ class MemoryWorkQueue(WorkQueue): if remaining <= 0: return [] self._wake.wait(remaining) - return [self._items.popleft() for _ in range(min(count, len(self._items)))] + claimed = [ + self._items.popleft() for _ in range(min(count, len(self._items))) + ] + self._in_flight += len(claimed) + return claimed def ack(self, item: WorkItem) -> None: - """Nothing to acknowledge: claiming already removed it.""" + """Nothing to take off the queue — claiming did that. Only counted.""" + with self._lock: + self._in_flight = max(0, self._in_flight - 1) def reclaim_stale(self, min_idle_ms: int) -> list[WorkItem]: """Nothing to reclaim: an item lost here died with the process.""" @@ -253,7 +261,7 @@ class MemoryWorkQueue(WorkQueue): with self._lock: return { "depth": len(self._items), - "pending": 0, + "pending": self._in_flight, "delayed": len(self._delayed), "parked": sum(len(v) for v in self._parked.values()), "oldest_pending_s": 0.0, diff --git a/backend/tests/api/routes/test_alerts.py b/backend/tests/api/routes/test_alerts.py new file mode 100644 index 0000000..b41df61 --- /dev/null +++ b/backend/tests/api/routes/test_alerts.py @@ -0,0 +1,66 @@ +"""The Test button: it exists to tell a working channel from a broken one.""" + +from typing import Any + +import pytest +from fastapi.testclient import TestClient + +from app.core.config import settings +from app.flow.alerts import AlertsConfig, Channel + +PREFIX = f"{settings.API_V1_STR}/alerts" + + +def _one_channel(client: TestClient) -> Any: + alerts = client.app.state.flow_controller.alerts # type: ignore[attr-defined] + alerts.config = AlertsConfig( + channels=[Channel(name="phone", kind="ntfy", config={"topic": "t"})] + ) + return alerts + + +def test_a_channel_that_cannot_deliver_is_not_a_success( + client: TestClient, + superuser_token_headers: dict[str, str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + alerts = _one_channel(client) + + async def refuse(_config: dict[str, Any], _alert: Any) -> None: + raise RuntimeError("nobody home") + + monkeypatch.setattr(alerts, "_send_ntfy", refuse) + + response = client.post(f"{PREFIX}/test/phone", headers=superuser_token_headers) + + assert response.status_code == 502 + assert "nobody home" in response.json()["detail"] + + +def test_a_channel_that_delivers_says_so( + client: TestClient, + superuser_token_headers: dict[str, str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + alerts = _one_channel(client) + sent = [] + + async def accept(_config: dict[str, Any], alert: Any) -> None: + sent.append(alert) + + monkeypatch.setattr(alerts, "_send_ntfy", accept) + + response = client.post(f"{PREFIX}/test/phone", headers=superuser_token_headers) + + assert response.status_code == 200 + assert len(sent) == 1 + + +def test_an_unknown_channel_is_a_404( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + _one_channel(client) + + response = client.post(f"{PREFIX}/test/nowhere", headers=superuser_token_headers) + + assert response.status_code == 404 diff --git a/backend/tests/api/routes/test_flows.py b/backend/tests/api/routes/test_flows.py index eed6f74..27fbdc2 100644 --- a/backend/tests/api/routes/test_flows.py +++ b/backend/tests/api/routes/test_flows.py @@ -357,3 +357,69 @@ def test_rename_rejects_an_invalid_name( json={"new_name": "Not A Flow Name"}, ) assert response.status_code == 400 + + +def test_a_node_that_raises_answers_with_its_error( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """A manual run fails the way a scheduled one does, not as a 500.""" + saved = client.put( + f"{PREFIX}/failing", headers=superuser_token_headers, json=a_flow("failing") + ).json() + client.put( + f"{PREFIX}/failing/nodes/sensor/source", + headers=superuser_token_headers, + json={"code": BROKEN_NODE}, + ) + client.post( + f"{PREFIX}/failing/publish", + headers=superuser_token_headers, + json={"version": saved["definition"]["version"]}, + ) + + response = client.post( + f"{PREFIX}/failing/nodes/sensor/trigger", + headers=superuser_token_headers, + json={"values": {}}, + ) + + assert response.status_code == 400 + assert "boom" in response.json()["detail"] + + client.delete(f"{PREFIX}/failing", headers=superuser_token_headers) + + +def test_renaming_a_flow_leaves_nothing_under_the_old_name( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + saved = client.put( + f"{PREFIX}/movable", headers=superuser_token_headers, json=a_flow("movable") + ).json() + client.put( + f"{PREFIX}/movable/nodes/sensor/source", + headers=superuser_token_headers, + json={"code": WORKING_NODE}, + ) + client.post( + f"{PREFIX}/movable/publish", + headers=superuser_token_headers, + json={"version": saved["definition"]["version"]}, + ) + client.post( + f"{PREFIX}/movable/run", headers=superuser_token_headers, json={"inputs": {}} + ) + + state = client.app.state.flow_controller.state + assert [key for key in state.keys() if "movable." in key] + + client.post( + f"{PREFIX}/movable/rename", + headers=superuser_token_headers, + json={"new_name": "moved"}, + ) + + # The values repopulate under the new name on the next run; what the old + # name left behind would sit there for good. + assert [key for key in state.keys() if "movable." in key] == [] + + client.delete(f"{PREFIX}/moved", headers=superuser_token_headers) diff --git a/backend/tests/api/routes/test_observability.py b/backend/tests/api/routes/test_observability.py index 04f5d28..eff67d4 100644 --- a/backend/tests/api/routes/test_observability.py +++ b/backend/tests/api/routes/test_observability.py @@ -134,3 +134,35 @@ def test_runs_narrow_to_one_minute( # The upper bound is exclusive, so the run starting the next minute is not # in this one. assert [run["id"] for run in runs] == ["minute-in"] + + +def test_events_narrow_to_one_minute( + client: TestClient, superuser_token_headers: dict[str, str], db: Session +) -> None: + """The failures list reaches as far back as the charts beside it.""" + minute = datetime.now(timezone.utc).replace(second=0, microsecond=0) - timedelta( + hours=4 + ) + db.add(EngineEvent(ts=minute, type="node_error", flow=FLOW, detail="minute-in")) + db.add( + EngineEvent( + ts=minute + timedelta(minutes=1), + type="node_error", + flow=FLOW, + detail="minute-after", + ) + ) + db.commit() + + events = client.get( + f"{PREFIX}/events", + headers=superuser_token_headers, + params={ + "flow": FLOW, + "since": minute.isoformat(), + "until": (minute + timedelta(minutes=1)).isoformat(), + }, + ).json() + + # The upper bound is exclusive, so the failure a minute later is not in it. + assert [event["detail"] for event in events] == ["minute-in"] diff --git a/backend/tests/flow/test_queue.py b/backend/tests/flow/test_queue.py index b5e1460..07c8225 100644 --- a/backend/tests/flow/test_queue.py +++ b/backend/tests/flow/test_queue.py @@ -60,6 +60,20 @@ def test_a_deleted_flow_leaves_nothing_parked(): assert queue.unpark("gone") == [] +def test_claimed_work_counts_as_in_flight_until_it_is_acknowledged(): + """The health tile's "in flight" reads zero without this.""" + queue = MemoryWorkQueue() + queue.add(WorkItem(kind="cascade", node="f.n", flow="f")) + + assert queue.stats()["pending"] == 0 + + (item,) = queue.claim(1, 10) + assert queue.stats()["pending"] == 1 + + queue.ack(item) + assert queue.stats()["pending"] == 0 + + def _pipeline_with_a_consumer() -> tuple[Pipeline, Node, MemoryState, list]: """A source whose message a consumer records.""" seen: list[float] = [] diff --git a/backend/tests/test_metrics.py b/backend/tests/test_metrics.py index d65c94d..9f42f1b 100644 --- a/backend/tests/test_metrics.py +++ b/backend/tests/test_metrics.py @@ -11,7 +11,7 @@ from datetime import datetime, timezone from sqlmodel import Session, select from app.flow.events import EventBus -from app.flow.metrics import MetricsCollector +from app.flow.metrics import DETAIL_CAP, RUN_STALE_S, MetricsCollector from app.models import EngineEvent, FlowRun, MetricBucket FLOW = "metrics-test" @@ -197,3 +197,30 @@ def test_a_run_that_did_not_fail_reads_ok(db: Session) -> None: run = db.exec(select(FlowRun).where(FlowRun.id == "manual-abc")).one() assert (run.status, run.source) == ("ok", "manual") + + +def test_a_traceback_no_failure_ever_claims_is_dropped() -> None: + """One entry per node that ever errored would else last the process out.""" + node = f"{FLOW}.stale" + collector = MetricsCollector(EventBus()) + collector.handle( + { + "type": "node_log", + "flow": FLOW, + "node": node, + "level": "error", + "text": "x" * (DETAIL_CAP * 2), + "ts": time.time() - RUN_STALE_S - 1, + } + ) + + # Untruncated, one of these holds a whole run of a chatty node. + assert len(collector._tracebacks[(FLOW, node)][1]) == DETAIL_CAP + + # Something to write, so the flush does not stop at its early return. + collector.handle( + {"type": "node_executed", "flow": FLOW, "node": node, "ts": time.time()} + ) + asyncio.run(collector.flush()) + + assert collector._tracebacks == {}