diff --git a/backend/fluksio/flow/alerts.py b/backend/fluksio/flow/alerts.py index e614d8c..6213215 100644 --- a/backend/fluksio/flow/alerts.py +++ b/backend/fluksio/flow/alerts.py @@ -114,10 +114,12 @@ def describe(event: dict[str, Any]) -> Alert | None: node=node, ) if kind == "flow_quarantined": + retry = event.get("retry_in_s") + again = f" Trying again in {round(float(retry) / 60)} min." if retry else "" return Alert( title=f"Flow '{flow}' was quarantined", body=( - f"It kept crashing, so the engine stopped restarting it. {error}" + f"It kept crashing, so the engine stood it down. {error}{again}" ).strip(), flow=flow, ) diff --git a/backend/fluksio/flow/supervision.py b/backend/fluksio/flow/supervision.py index 2e72678..ff2d624 100644 --- a/backend/fluksio/flow/supervision.py +++ b/backend/fluksio/flow/supervision.py @@ -3,9 +3,15 @@ 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 +stands a flow down when one keeps failing, because a flow crash-looping every second is worse than a flow that is visibly stopped. +Standing down is a rest, not a verdict. It used to be the end of the task: the +flow stayed dead until somebody published it or the engine restarted. On a +house that runs unattended that turns a broker rebooting at four in the morning +into a hole in the record until someone notices — so the rest expires, the +budget comes back, and the task tries again on a much longer clock. + 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. @@ -25,10 +31,20 @@ from fluksio.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. +# A task that burns through this many restarts in the window is not going to +# recover by being restarted again *now*. +# +# Per task rather than per flow: a flow whose subscriber and its keepalive +# publisher both go down with the same broker spends one shared budget twice as +# fast as it was sized for, and gives up before the backoff above has reached +# its longest step. That is what took the house's inverter readings out for +# nearly five hours — six crashes across two tasks inside 41 seconds. FAILURE_BUDGET = 5 FAILURE_WINDOW = 300.0 +# How long a task sits out before its budget is handed back. Minutes rather +# than seconds because a device that is really gone should not be hammered, and +# capped in the hour because nothing here is worth losing a day of readings to. +QUARANTINE_BACKOFF = (300.0, 900.0, 3600.0) # How long a cancelled task gets to notice. A loop that is still waiting after # this is not going to stop on its own — a client closing a socket the broker # no longer answers on is the case seen in the wild — and the rebuild asking @@ -48,8 +64,15 @@ class Supervisor: #: task's name, because names are a flow and a node joined by a dot and #: matching on that prefix would let 'hea' cancel 'heating'. self._flows: dict[str, str] = {} + #: Crash times per *task*, keyed by task name. self._failures: dict[str, deque[float]] = {} - self.quarantined: set[str] = set() + #: The tasks sitting out a quarantine, and the flow each belongs to. + self._resting: dict[str, str] = {} + + @property + def quarantined(self) -> set[str]: + """The flows with at least one task currently standing down.""" + return set(self._resting.values()) def spawn(self, name: str, flow: str, factory: TaskFactory) -> None: """Run `factory()` and keep running it until told to stop.""" @@ -62,6 +85,7 @@ class Supervisor: async def _supervise(self, name: str, flow: str, factory: TaskFactory) -> None: attempt = 0 + rests = 0 while True: try: await factory() @@ -69,7 +93,10 @@ class Supervisor: raise except Exception as exc: if not self._record_failure(name, flow, exc): - return + await self._rest(name, flow, exc, rests) + rests += 1 + attempt = 0 + continue else: # A clean return means the loop decided it was done. return @@ -78,8 +105,37 @@ class Supervisor: attempt += 1 await asyncio.sleep(delay) + async def _rest(self, name: str, flow: str, exc: Exception, rests: int) -> None: + """Stand the task down, then hand its budget back and let it retry.""" + delay = QUARANTINE_BACKOFF[min(rests, len(QUARANTINE_BACKOFF) - 1)] + logger.error( + "Flow '%s' crashed %d times in %.0fs — quarantined, retrying in %.0fs", + flow, + FAILURE_BUDGET, + FAILURE_WINDOW, + delay, + ) + self._resting[name] = flow + self._publish( + { + "type": "flow_quarantined", + "flow": flow, + "task": name, + "error": f"{type(exc).__name__}: {exc}", + "retry_in_s": delay, + "ts": time.time(), + } + ) + try: + await asyncio.sleep(delay) + finally: + # Both, and in this order: a task woken by a cancellation is being + # torn down, and must not leave the flow looking quarantined. + self._resting.pop(name, None) + self._failures.pop(name, None) + def _record_failure(self, name: str, flow: str, exc: Exception) -> bool: - """Note the crash; False when the flow has spent its budget.""" + """Note the crash; False when this task has spent its budget.""" logger.warning("Supervised task '%s' crashed: %s", name, exc, exc_info=True) self._publish( { @@ -92,36 +148,20 @@ class Supervisor: ) now = time.monotonic() - window = self._failures.setdefault(flow, deque()) + window = self._failures.setdefault(name, 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 + return len(window) < FAILURE_BUDGET async def cancel_all(self) -> None: """Stop supervising. Idempotent, and safe to call mid-restart.""" tasks = list(self._tasks.values()) self._tasks.clear() self._flows.clear() + self._resting.clear() + self._failures.clear() await self._cancel(tasks) async def cancel_flow(self, flow: str) -> None: @@ -136,11 +176,12 @@ class Supervisor: tasks = [self._tasks.pop(name) for name in names if name in self._tasks] for name in names: del self._flows[name] + # The clean slate: whatever these tasks spent before, the build + # that follows starts their budgets again — and a rebuild is not + # made to wait out a rest somebody has just fixed the cause of. + self._failures.pop(name, None) + self._resting.pop(name, None) await self._cancel(tasks) - # The clean slate: whatever this flow spent before, the build that - # follows starts its budget again. - self.quarantined.discard(flow) - self._failures.pop(flow, None) async def _cancel(self, tasks: list[asyncio.Task[None]]) -> None: """Ask these tasks to stop, and wait no longer than the grace period.""" diff --git a/backend/tests/flow/test_supervision.py b/backend/tests/flow/test_supervision.py index 4424c52..63be469 100644 --- a/backend/tests/flow/test_supervision.py +++ b/backend/tests/flow/test_supervision.py @@ -13,6 +13,9 @@ from fluksio.flow.supervision import FAILURE_BUDGET, Supervisor def no_backoff(monkeypatch: pytest.MonkeyPatch): """Real backoff would make these tests a minute long.""" monkeypatch.setattr(supervision, "BACKOFF", (0.0,)) + # Long enough that a quarantine holds for the length of an assertion, short + # enough that the test asking it to expire does not wait five minutes. + monkeypatch.setattr(supervision, "QUARANTINE_BACKOFF", (0.05,)) async def _settle() -> None: @@ -61,7 +64,7 @@ def test_a_loop_that_returns_is_left_alone(): asyncio.run(scenario()) -def test_a_flow_that_keeps_crashing_is_quarantined(): +def test_a_flow_that_keeps_crashing_is_stood_down(): attempts = 0 events: list[dict] = [] bus = EventBus() @@ -81,13 +84,14 @@ def test_a_flow_that_keeps_crashing_is_quarantined(): 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" + quarantine = [e for e in events if e["type"] == "flow_quarantined"][-1] + assert quarantine["retry_in_s"] == 0.05 await supervisor.cancel_all() asyncio.run(scenario()) -def test_the_budget_is_per_flow(): +def test_one_flows_failure_does_not_stand_another_down(): ran = 0 async def failing() -> None: @@ -111,6 +115,63 @@ def test_the_budget_is_per_flow(): asyncio.run(scenario()) +def test_two_tasks_on_one_flow_do_not_spend_each_others_budget(): + """The house's power flow: an MQTT subscriber and its keepalive publisher. + + Both die with the same broker. Sharing one budget meant the flow was stood + down after half the restarts either task was allowed on its own. + """ + tries = {"sub": 0, "pub": 0} + + def crasher(which: str): + async def loop() -> None: + tries[which] += 1 + raise ConnectionError("broker went away") + + return loop + + async def scenario() -> None: + supervisor = Supervisor() + supervisor.spawn("power.cerbo:mqtt", "power", crasher("sub")) + supervisor.spawn("power.keepalive:mqtt-out", "power", crasher("pub")) + await _settle() + + assert tries == {"sub": FAILURE_BUDGET, "pub": FAILURE_BUDGET} + await supervisor.cancel_all() + + asyncio.run(scenario()) + + +def test_a_quarantine_expires_and_the_task_tries_again(): + """The whole point: a broker that comes back is picked up without a human.""" + attempts = 0 + + async def loop() -> None: + nonlocal attempts + attempts += 1 + if attempts <= FAILURE_BUDGET: + raise ConnectionError("broker went away") + await asyncio.sleep(3600) + + async def scenario() -> None: + supervisor = Supervisor() + supervisor.spawn("mqtt", "power", loop) + await _settle() + + assert attempts == FAILURE_BUDGET + assert supervisor.quarantined == {"power"} + + # Sit out the rest, which the fixture has shortened to 50ms. + await asyncio.sleep(0.1) + await _settle() + + assert attempts == FAILURE_BUDGET + 1 + assert supervisor.quarantined == set() + await supervisor.cancel_all() + + asyncio.run(scenario()) + + def test_cancel_all_is_idempotent(): async def loop() -> None: await asyncio.sleep(3600) @@ -165,8 +226,8 @@ def test_cancelling_one_flow_lifts_only_its_quarantine(): assert supervisor.quarantined == {"also-broken"} # And its budget, or the flow would be quarantined again on the first # crash after the rebuild that was meant to fix it. - assert "broken" not in supervisor._failures - assert "also-broken" in supervisor._failures + assert "a" not in supervisor._failures + assert "b" in supervisor._failures await supervisor.cancel_all() asyncio.run(scenario())