From e7e48c4f13a96034632934731b9a0468ecac3483 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 16 Aug 2026 07:23:02 +0200 Subject: [PATCH] Supervise the loops a flow starts, and give up loudly MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A node's subscription, schedule or poll loop was a bare asyncio task: one that raised outside its own retry handling was simply gone, and the node went on being listed as running while nothing listened any more. Those loops now run under a supervisor that restarts them with a growing delay and quarantines a flow that burns through five restarts in five minutes — a flow crash-looping every second is worse than one that is visibly stopped, and the dashboard can now say which. The MQTT subscription loses its private five-second reconnect in the process: one backoff policy per socket, and it belongs to the supervisor. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_011LF61rxW1FG5YCD2J9YqjY --- ROADMAP.md | 4 + backend/app/api/routes/flows.py | 1 + backend/app/flow/connector.py | 21 +++-- backend/app/flow/controller.py | 15 +++ backend/app/flow/nodes.py | 120 +++++++++++++---------- backend/app/flow/schemas.py | 2 + backend/app/flow/supervision.py | 124 ++++++++++++++++++++++++ backend/tests/flow/test_supervision.py | 126 +++++++++++++++++++++++++ frontend/src/client/schemas.gen.ts | 5 + frontend/src/client/types.gen.ts | 1 + 10 files changed, 358 insertions(+), 61 deletions(-) create mode 100644 backend/app/flow/supervision.py create mode 100644 backend/tests/flow/test_supervision.py diff --git a/ROADMAP.md b/ROADMAP.md index fa92ba9..32e5796 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -76,6 +76,10 @@ external interfaces. See `docs/architecture/structure.canvas` → *Backend – M reachability and fails the container healthcheck, so a wedged engine is restarted rather than counted as up. One engine per deployment — the API image runs a single worker, because a second one would be a second engine +- [x] Supervised background tasks: a node's subscription, schedule or poll loop is + restarted with growing delay when it dies, and a flow that spends its failure + budget is quarantined and surfaced rather than left crash-looping. The loops + themselves no longer carry private retry logic - [ ] Test nodes: a small node dragged onto an existing one, smoke or unit, blocking deployment on failure - [ ] User management scoped per flow and per data set diff --git a/backend/app/api/routes/flows.py b/backend/app/api/routes/flows.py index e008108..27a9af7 100644 --- a/backend/app/api/routes/flows.py +++ b/backend/app/api/routes/flows.py @@ -172,6 +172,7 @@ def read_flows(controller: FlowControllerDep) -> Any: has_draft=controller.store.has_draft(name), enabled=controller.is_enabled(name), paused=controller.is_paused(name), + quarantined=controller.is_quarantined(name), ) ) return FlowsPublic(data=summaries, count=len(summaries)) diff --git a/backend/app/flow/connector.py b/backend/app/flow/connector.py index a570719..2bd0a05 100644 --- a/backend/app/flow/connector.py +++ b/backend/app/flow/connector.py @@ -109,21 +109,22 @@ class ConnectorNode(Node): # ------------------------------------------------------------------------- async def start(self, app: FastAPI | None = None) -> None: - if self.config.poll_interval > 0 and self._poll_task is None: + if self.config.poll_interval > 0 and self._stop_event is None: self._stop_event = asyncio.Event() - self._poll_task = asyncio.create_task(self._poll_loop()) + self._poll_task = self._run_supervised("poll", self._poll_loop) async def stop(self, app: FastAPI | None = None) -> None: - if self._poll_task is None: + if self._stop_event is None: return - if self._stop_event is not None: - self._stop_event.set() - self._poll_task.cancel() - try: - await self._poll_task - except (asyncio.CancelledError, Exception): # noqa: B014 - shutting down - pass + self._stop_event.set() + if self._poll_task is not None: + self._poll_task.cancel() + try: + await self._poll_task + except (asyncio.CancelledError, Exception): # noqa: B014 - shutting down + pass self._poll_task = None + self._stop_event = None self._last_published = {} async def _poll_loop(self) -> None: diff --git a/backend/app/flow/controller.py b/backend/app/flow/controller.py index 71e7818..64547d1 100644 --- a/backend/app/flow/controller.py +++ b/backend/app/flow/controller.py @@ -43,6 +43,7 @@ from app.flow.schemas import ( from app.flow.secrets import SecretNotFound, resolve_params from app.flow.state import MemoryState, StateBackend from app.flow.store import LIB_DIR, FlowNotFound, FlowStore, LibNotFound +from app.flow.supervision import Supervisor logger = logging.getLogger(__name__) @@ -207,6 +208,7 @@ class FlowController: self.loaded: dict[str, LoadedNode] = {} self.issues: list[ValidationIssue] = [] self.disabled: set[str] = set() + self.supervisor = Supervisor(events) self._lock = asyncio.Lock() # ------------------------------------------------------------------------- @@ -228,6 +230,9 @@ class FlowController: """Rebuild the whole pipeline from what is currently stored.""" async with self._lock: await self._teardown() + # A fresh supervisor per build, so a flow quarantined by the last + # one gets another chance once its author has changed something. + self.supervisor = Supervisor(self.events) published = self.store.read_all() self.disabled = { @@ -272,6 +277,8 @@ class FlowController: await node.stop(self.app) except Exception: logger.exception("Error stopping node '%s'", entry.id) + # After the nodes, so a loop still winding down is not restarted. + await self.supervisor.cancel_all() async def _activate(self) -> None: """Start subscriptions, schedules and webhooks of the new pipeline. @@ -287,6 +294,7 @@ class FlowController: # that is what stopping it means. if entry.flow in self.disabled: continue + node.supervisor = self.supervisor try: await node.start(self.app) except Exception as exc: @@ -474,6 +482,13 @@ class FlowController: def is_paused(self, flow: str) -> bool: return self.pipeline is not None and flow in self.pipeline.paused_flows() + def is_quarantined(self, flow: str) -> bool: + return flow in self.supervisor.quarantined + + @property + def quarantined(self) -> set[str]: + return self.supervisor.quarantined + def paused_flows(self) -> list[str]: return self.pipeline.paused_flows() if self.pipeline else [] diff --git a/backend/app/flow/nodes.py b/backend/app/flow/nodes.py index 1e03f2f..366fd90 100644 --- a/backend/app/flow/nodes.py +++ b/backend/app/flow/nodes.py @@ -5,7 +5,7 @@ import hmac import logging import threading import time -from collections.abc import Callable, Iterable +from collections.abc import Callable, Coroutine, Iterable from enum import Enum from typing import TYPE_CHECKING, Any, Literal @@ -15,6 +15,7 @@ from pydantic import BaseModel, ConfigDict, Field from app.flow import logs from app.flow.messages import MessageSpec, qualify +from app.flow.supervision import Supervisor if TYPE_CHECKING: from fastapi import FastAPI @@ -75,6 +76,7 @@ class Node: "_pipeline", "synchronous", "_on_health", + "supervisor", ) def __init__( @@ -88,6 +90,9 @@ class Node: self.f = f self._pipeline: Pipeline | None = None self._on_health: Callable[[Node, str, str | None], None] | None = None + # Set by the controller before start(); a node built on its own (tests, + # previews) runs its loops unsupervised. + self.supervisor: Supervisor | None = None self.params = dict(params) if params else {} self.synchronous = bool(self.params.get("synchronous", False)) @@ -155,6 +160,19 @@ class Node: async def stop(self, app: FastAPI | None = None) -> None: """Undo :meth:`start`. Called before a rebuild, and must be idempotent.""" + def _run_supervised( + self, name: str, factory: Callable[[], Coroutine[Any, Any, None]] + ) -> asyncio.Task | None: + """Run a background loop under supervision where there is one. + + Returns the task only when unsupervised, since that is the one case + where the caller has to cancel it itself. + """ + if self.supervisor is not None: + self.supervisor.spawn(f"{self.id}:{name}", self.flow, factory) + return None + return asyncio.create_task(factory()) + def report_health(self, status: str, detail: str | None = None) -> None: """Say how this node's connection is doing: ok, degraded or down.""" if self._on_health is not None: @@ -854,9 +872,7 @@ class MqttNode(Node): broker_host: str = "localhost" broker_port: int = 1883 username: str | None = None - password: str | None = Field( - default=None, json_schema_extra={"x-secret": True} - ) + password: str | None = Field(default=None, json_schema_extra={"x-secret": True}) client_id: str | None = None qos: int = 0 retain: bool = False @@ -1071,7 +1087,7 @@ class MqttNode(Node): return # Already running self._stop_event = asyncio.Event() - self._subscription_task = asyncio.create_task(self._subscription_loop()) + self._subscription_task = self._run_supervised("mqtt", self._subscription_loop) logger.info( "Started MQTT subscription for node '%s' to topics %s", self.name, @@ -1082,19 +1098,21 @@ class MqttNode(Node): """ Stop the MQTT subscription. - Gracefully stops the background subscription task. + Gracefully stops the background subscription task. A supervised + subscription is cancelled with the rest of them at teardown; only an + unsupervised one is this method's to cancel. """ - if self._subscription_task is None: + if self._stop_event is None: return - if self._stop_event: - self._stop_event.set() + self._stop_event.set() - self._subscription_task.cancel() - try: - await self._subscription_task - except asyncio.CancelledError: - pass + if self._subscription_task is not None: + self._subscription_task.cancel() + try: + await self._subscription_task + except asyncio.CancelledError: + pass self._subscription_task = None self._stop_event = None @@ -1105,17 +1123,21 @@ class MqttNode(Node): async def _subscription_loop(self) -> None: """ - Background loop that listens for MQTT messages and triggers the pipeline. + Listen for MQTT messages and trigger the pipeline. Subscribes to all unique topics from the ``topics`` mapping and uses the reverse lookup ``_topic_to_ports`` to route incoming payloads to the correct pipeline message names. + + One connection attempt: a dropped broker raises, and the supervisor + decides when to try again. Reconnecting here as well would mean two + backoff policies fighting over the same socket. """ import json import aiomqtt - while not (self._stop_event and self._stop_event.is_set()): + if not (self._stop_event and self._stop_event.is_set()): try: async with aiomqtt.Client( hostname=self.broker_host, @@ -1189,27 +1211,25 @@ class MqttNode(Node): exc_info=True, ) - except aiomqtt.MqttError as e: + # Falling out of the message iterator without being told to + # stop means the broker went away quietly. Raising is how the + # supervisor hears about it. + if not (self._stop_event and self._stop_event.is_set()): + self.report_health("down", "subscription ended") + raise ConnectionError( + f"MQTT subscription for '{self.name}' ended unexpectedly" + ) + + except asyncio.CancelledError: + raise + except Exception as e: logger.warning( - "MQTT connection error in node '%s': %s. Retrying in 5s...", - self.name, - e, + "MQTT subscription for node '%s' failed: %s", self.name, e ) self.report_health("down", str(e)) - if not (self._stop_event and self._stop_event.is_set()): - # Reconnect after a delay - await asyncio.sleep(5) - except asyncio.CancelledError: - break - except Exception as e: - logger.error( - "Unexpected error in MQTT subscription for node '%s': %s. Retrying in 5s...", - self.name, - e, - exc_info=True, - ) - if not (self._stop_event and self._stop_event.is_set()): - await asyncio.sleep(5) + if self._stop_event and self._stop_event.is_set(): + return + raise @property def is_subscribed(self) -> bool: @@ -1880,22 +1900,22 @@ class DelayNode(Node): return self._stop_cron = asyncio.Event() - self._cron_task = asyncio.create_task(self._cron_loop()) + self._cron_task = self._run_supervised("cron", self._cron_loop) logger.info("Started cron for node '%s': %s", self.name, self.cron_expr) async def stop_cron(self) -> None: """Stop the cron scheduler.""" - if self._cron_task is None: + if self._stop_cron is None: return - if self._stop_cron: - self._stop_cron.set() + self._stop_cron.set() - self._cron_task.cancel() - try: - await self._cron_task - except asyncio.CancelledError: - pass + if self._cron_task is not None: + self._cron_task.cancel() + try: + await self._cron_task + except asyncio.CancelledError: + pass self._cron_task = None self._stop_cron = None @@ -1955,16 +1975,14 @@ class DelayNode(Node): await self._trigger_cron() except asyncio.CancelledError: - break - except Exception as e: + raise + except Exception: + # One bad tick should not cost the schedule; the supervisor + # picks it up from here, backoff included. logger.error( - "Error in cron loop for node '%s': %s", - self.name, - e, - exc_info=True, + "Error in cron loop for node '%s'", self.name, exc_info=True ) - # Back off on error to avoid tight loops - await asyncio.sleep(60) + raise async def _trigger_cron(self): """Emit data into the pipeline on a cron tick.""" diff --git a/backend/app/flow/schemas.py b/backend/app/flow/schemas.py index 71b7aec..fae3dd3 100644 --- a/backend/app/flow/schemas.py +++ b/backend/app/flow/schemas.py @@ -126,6 +126,8 @@ class FlowSummary(BaseModel): has_draft: bool = False enabled: bool = True paused: bool = False + # Its background tasks kept crashing, so the engine stopped restarting them. + quarantined: bool = False class FlowsPublic(BaseModel): diff --git a/backend/app/flow/supervision.py b/backend/app/flow/supervision.py new file mode 100644 index 0000000..63ffd2b --- /dev/null +++ b/backend/app/flow/supervision.py @@ -0,0 +1,124 @@ +"""Supervision for the long-lived tasks a pipeline starts. + +A node's subscription, schedule or poll loop is an `asyncio.Task`, and a task +that raises is simply gone — the node stays listed as running while nothing +listens any more. The supervisor restarts those tasks with a growing delay and +gives up on a flow that keeps failing, because a flow crash-looping every +second is worse than a flow that is visibly stopped. + +Deliberately a plain task registry rather than a `TaskGroup`: a group cancels +its siblings when one member fails, which is the opposite of what supervision +means here. +""" + +from __future__ import annotations + +import asyncio +import logging +import time +from collections import deque +from collections.abc import Callable, Coroutine +from typing import Any + +from app.flow.events import EventBus + +logger = logging.getLogger(__name__) + +BACKOFF = (1.0, 5.0, 30.0, 60.0) +# A flow that burns through this many restarts in the window is not going to +# recover by being restarted again. +FAILURE_BUDGET = 5 +FAILURE_WINDOW = 300.0 + +TaskFactory = Callable[[], Coroutine[Any, Any, None]] + + +class Supervisor: + """Keeps the pipeline's background tasks alive, or admits it cannot.""" + + def __init__(self, events: EventBus | None = None) -> None: + self._events = events + self._tasks: dict[str, asyncio.Task[None]] = {} + self._failures: dict[str, deque[float]] = {} + self.quarantined: set[str] = set() + + def spawn(self, name: str, flow: str, factory: TaskFactory) -> None: + """Run `factory()` and keep running it until told to stop.""" + if name in self._tasks: + return + self._tasks[name] = asyncio.create_task( + self._supervise(name, flow, factory), name=f"supervised:{name}" + ) + + async def _supervise(self, name: str, flow: str, factory: TaskFactory) -> None: + attempt = 0 + while True: + try: + await factory() + except asyncio.CancelledError: + raise + except Exception as exc: + if not self._record_failure(name, flow, exc): + return + else: + # A clean return means the loop decided it was done. + return + + delay = BACKOFF[min(attempt, len(BACKOFF) - 1)] + attempt += 1 + await asyncio.sleep(delay) + + def _record_failure(self, name: str, flow: str, exc: Exception) -> bool: + """Note the crash; False when the flow has spent its budget.""" + logger.warning("Supervised task '%s' crashed: %s", name, exc, exc_info=True) + self._publish( + { + "type": "task_crashed", + "task": name, + "flow": flow, + "error": f"{type(exc).__name__}: {exc}", + "ts": time.time(), + } + ) + + now = time.monotonic() + window = self._failures.setdefault(flow, deque()) + window.append(now) + while window and window[0] < now - FAILURE_WINDOW: + window.popleft() + + if len(window) < FAILURE_BUDGET: + return True + + logger.error( + "Flow '%s' crashed %d times in %.0fs — quarantined", + flow, + len(window), + FAILURE_WINDOW, + ) + self.quarantined.add(flow) + self._publish( + { + "type": "flow_quarantined", + "flow": flow, + "error": f"{type(exc).__name__}: {exc}", + "ts": time.time(), + } + ) + return False + + async def cancel_all(self) -> None: + """Stop supervising. Idempotent, and safe to call mid-restart.""" + tasks = list(self._tasks.values()) + self._tasks.clear() + for task in tasks: + task.cancel() + for task in tasks: + try: + await task + except (asyncio.CancelledError, Exception): # noqa: B014 - shutting down + pass + + def _publish(self, event: dict[str, Any]) -> None: + if self._events is not None: + self._events.publish(event) diff --git a/backend/tests/flow/test_supervision.py b/backend/tests/flow/test_supervision.py new file mode 100644 index 0000000..7aa9550 --- /dev/null +++ b/backend/tests/flow/test_supervision.py @@ -0,0 +1,126 @@ +"""Supervision: a background task that dies gets restarted, until it is hopeless.""" + +import asyncio + +import pytest + +from app.flow import supervision +from app.flow.events import EventBus +from app.flow.supervision import FAILURE_BUDGET, Supervisor + + +@pytest.fixture(autouse=True) +def no_backoff(monkeypatch: pytest.MonkeyPatch): + """Real backoff would make these tests a minute long.""" + monkeypatch.setattr(supervision, "BACKOFF", (0.0,)) + + +async def _settle() -> None: + """Let the supervisor work through its restarts.""" + for _ in range(50): + await asyncio.sleep(0) + + +def test_a_crashing_loop_is_restarted(): + attempts = 0 + + async def loop() -> None: + nonlocal attempts + attempts += 1 + if attempts < 3: + raise ConnectionError("broker went away") + + async def scenario() -> None: + supervisor = Supervisor() + supervisor.spawn("mqtt", "heating", loop) + await _settle() + + assert attempts == 3 + assert supervisor.quarantined == set() + await supervisor.cancel_all() + + asyncio.run(scenario()) + + +def test_a_loop_that_returns_is_left_alone(): + """Returning is how a loop says it is finished, not that it failed.""" + attempts = 0 + + async def loop() -> None: + nonlocal attempts + attempts += 1 + + async def scenario() -> None: + supervisor = Supervisor() + supervisor.spawn("cron", "heating", loop) + await _settle() + + assert attempts == 1 + await supervisor.cancel_all() + + asyncio.run(scenario()) + + +def test_a_flow_that_keeps_crashing_is_quarantined(): + attempts = 0 + events: list[dict] = [] + bus = EventBus() + bus.publish = events.append # type: ignore[method-assign] + + async def loop() -> None: + nonlocal attempts + attempts += 1 + raise RuntimeError("nope") + + async def scenario() -> None: + supervisor = Supervisor(bus) + supervisor.spawn("mqtt", "heating", loop) + await _settle() + + # Restarting stops once the budget is gone, rather than spinning forever. + assert attempts == FAILURE_BUDGET + assert supervisor.quarantined == {"heating"} + assert [e["type"] for e in events].count("task_crashed") == FAILURE_BUDGET + assert [e["type"] for e in events][-1] == "flow_quarantined" + await supervisor.cancel_all() + + asyncio.run(scenario()) + + +def test_the_budget_is_per_flow(): + ran = 0 + + async def failing() -> None: + raise RuntimeError("nope") + + async def healthy() -> None: + nonlocal ran + ran += 1 + await asyncio.sleep(3600) + + async def scenario() -> None: + supervisor = Supervisor() + supervisor.spawn("a", "broken", failing) + supervisor.spawn("b", "fine", healthy) + await _settle() + + assert supervisor.quarantined == {"broken"} + assert ran == 1 + await supervisor.cancel_all() + + asyncio.run(scenario()) + + +def test_cancel_all_is_idempotent(): + async def loop() -> None: + await asyncio.sleep(3600) + + async def scenario() -> None: + supervisor = Supervisor() + supervisor.spawn("mqtt", "heating", loop) + await _settle() + + await supervisor.cancel_all() + await supervisor.cancel_all() + + asyncio.run(scenario()) diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 42f7399..357cdd3 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -364,6 +364,11 @@ export const FlowSummarySchema = { type: 'boolean', title: 'Paused', default: false + }, + quarantined: { + type: 'boolean', + title: 'Quarantined', + default: false } }, type: 'object', diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 19bcf6c..e2b660c 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -101,6 +101,7 @@ export type FlowSummary = { has_draft?: boolean; enabled?: boolean; paused?: boolean; + quarantined?: boolean; }; /**