rootandClaude Fable 5
e7e48c4f13
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011LF61rxW1FG5YCD2J9YqjY
127 lines
3.2 KiB
Python
127 lines
3.2 KiB
Python
"""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())
|