Alerts: cooldowns survive a restart, and a connection must stay down a minute before it alerts

The alert manager kept its cooldowns and its ceiling per hour in memory, so a
node that stayed down was announced again by every restart. Each delivered
alert is now a row in the new alert_sent table (additive migration), loaded
for the last seven days when the manager starts and pruned as new ones come
in. A ledger that fails never stops an alert. The manager's clock is wall
time for the same reason.

A node_health "down" alert also waits 60 s (HEALTH_GRACE_S) in a pending task
per node; a later event that is not "down" cancels it, as does stopping the
manager, so a broker restarting with the stack alerts nobody.

The cooldown used to be stamped before the ceiling per hour was checked, so
an alert the ceiling held back still spent its cooldown. It is stamped only
once the alert is sent now.

backend/fluksio/alembic/versions/b9e4d2a7c1f6_alert_sent.py

@@ -0,0 +1,35 @@
+"""alert_sent
+
+What the alert manager has sent, so a cooldown holds across a restart.
+
+Revision ID: b9e4d2a7c1f6
+Revises: a3c7e5d91f42
+Create Date: 2026-10-28
+
+"""
+
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "b9e4d2a7c1f6"
+down_revision = "a3c7e5d91f42"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "alert_sent",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("key", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.Column("ts", sa.Float(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(op.f("ix_alert_sent_ts"), "alert_sent", ["ts"], unique=False)
+
+
+def downgrade():
+    op.drop_index(op.f("ix_alert_sent_ts"), table_name="alert_sent")
+    op.drop_table("alert_sent")

backend/fluksio/flow/alert_ledger.py

@@ -0,0 +1,34 @@
+"""The alerts the manager has sent, in the engine's database.
+
+Cooldowns used to live in memory, so a node that stayed down was announced
+again by every restart. One row per delivered alert is enough to rebuild both
+the cooldowns and the ceiling per hour: the newest row of a key is when it was
+last sent, and the rows of the last hour are what the ceiling counts. Rows older
+than the retention are deleted as new ones come in; a cooldown longer than that
+is not supported.
+
+Blocking: the manager calls it through ``asyncio.to_thread``.
+"""
+
+from __future__ import annotations
+
+from sqlalchemy import delete
+from sqlmodel import Session, col, select
+
+from fluksio.core.db import engine
+from fluksio.models import AlertSent
+
+
+class AlertLedger:
+    def load(self, since: float) -> list[tuple[str, float]]:
+        """Every alert sent at or after *since*, as ``(key, ts)``."""
+        with Session(engine) as session:
+            rows = session.exec(select(AlertSent).where(AlertSent.ts >= since)).all()
+        return [(row.key, row.ts) for row in rows]
+
+    def add(self, key: str, ts: float, keep_since: float) -> None:
+        """Record one alert, and drop what is older than *keep_since*."""
+        with Session(engine) as session:
+            session.add(AlertSent(key=key, ts=ts))
+            session.exec(delete(AlertSent).where(col(AlertSent.ts) < keep_since))
+            session.commit()

backend/fluksio/flow/alerts.py

@@ -17,7 +17,7 @@ import logging
 import time
 from collections import deque
 from collections.abc import Callable
-from typing import Any, Literal
+from typing import Any, Literal, Protocol
 
 import httpx
 from pydantic import BaseModel, Field
@@ -47,6 +47,14 @@ FLAP_THRESHOLD = 3
 # However bad it gets, this is the most anyone is told per hour.
 RATE_LIMIT = 10
 RATE_WINDOW_S = 3600.0
+# A connection has to stay down this long before anyone is told. A broker that
+# restarts with the stack, or a device that reconnects, is back before it
+# matters. The same figure as `controller.HEALTH_FLOOR`, which the screens wait
+# out; it is not imported, because the controller imports this module.
+HEALTH_GRACE_S = 60.0
+# How far back a restart remembers what was sent. A cooldown longer than this
+# does not survive one.
+LEDGER_RETAIN_S = 7 * 24 * 3600.0
 
 
 class Channel(BaseModel):
@@ -163,6 +171,14 @@ def dedup_key(event: dict[str, Any]) -> str:
     return f"{event.get('type')}:{event.get('node') or event.get('flow') or ''}"
 
 
+class Ledger(Protocol):
+    """Where sent alerts are kept across a restart (``alert_ledger.AlertLedger``)."""
+
+    def load(self, since: float) -> list[tuple[str, float]]: ...
+
+    def add(self, key: str, ts: float, keep_since: float) -> None: ...
+
+
 class AlertManager:
     """Watches the event bus and tells someone when it matters."""
 
@@ -170,11 +186,17 @@ class AlertManager:
         self,
         events: EventBus,
         config: AlertsConfig | None = None,
-        now: Any = time.monotonic,
+        now: Any = time.time,
+        ledger: Ledger | None = None,
+        grace_s: float = HEALTH_GRACE_S,
     ) -> None:
         self._events = events
         self.config = config or AlertsConfig()
+        # Wall time, because what was sent is remembered across a restart and
+        # a monotonic clock means nothing to the next process.
         self._now = now
+        self._ledger = ledger
+        self.grace_s = grace_s
         #: How a "dashboard" channel puts its alert into the graph. Bound after
         #: construction, because the controller that publishes does not exist
         #: yet when the manager is built.
@@ -184,6 +206,8 @@ class AlertManager:
         self._health_flips: dict[str, deque[float]] = {}
         self._flapping: dict[str, float] = {}
         self._recent: deque[float] = deque()
+        #: Connections that went down and are waiting out the grace.
+        self._down: dict[str, asyncio.Task[None]] = {}
 
     # -------------------------------------------------------------------------
     # The loop
@@ -191,13 +215,41 @@ class AlertManager:
 
     async def run(self) -> None:
         """Consume the bus until cancelled."""
-        async with self._events.subscribe() as queue:
-            while True:
-                event = await queue.get()
-                try:
-                    await self.handle(event)
-                except Exception:
-                    logger.exception("Alerting failed for %s", event.get("type"))
+        try:
+            async with self._events.subscribe() as queue:
+                await self.restore()
+                while True:
+                    event = await queue.get()
+                    try:
+                        await self.handle(event)
+                    except Exception:
+                        logger.exception("Alerting failed for %s", event.get("type"))
+        finally:
+            for task in self._down.values():
+                task.cancel()
+            self._down.clear()
+
+    async def restore(self) -> None:
+        """Take what was sent before this process from the ledger.
+
+        Without it every restart starts with nothing on cooldown, and a node
+        that stays down is announced again each time.
+        """
+        if self._ledger is None:
+            return
+        now = self._now()
+        try:
+            rows = await asyncio.to_thread(self._ledger.load, now - LEDGER_RETAIN_S)
+        except Exception:
+            logger.exception("Could not read the alert ledger; cooldowns start empty")
+            return
+        for key, ts in sorted(rows, key=lambda row: row[1]):
+            # A clock that was set back would otherwise hold a key silent until
+            # it caught up.
+            ts = min(ts, now)
+            self._last_sent[key] = ts
+            if ts > now - RATE_WINDOW_S:
+                self._recent.append(ts)
 
     async def handle(self, event: dict[str, Any]) -> None:
         if not self.config.enabled:
@@ -205,9 +257,37 @@ class AlertManager:
         kind = str(event.get("type") or "")
         if kind not in ALERTING_EVENTS:
             return
+        node = str(event.get("node") or "")
+        if kind == "node_health" and event.get("health") != "down":
+            # Back before anyone was told, so there is nothing to tell.
+            self._cancel_down(node)
         if self._flaps(event):
             return
+        if kind == "node_health" and event.get("health") == "down" and self.grace_s > 0:
+            self._cancel_down(node)
+            self._down[node] = asyncio.create_task(self._down_after_grace(node, event))
+            return
+        await self._deliver(event)
 
+    def _cancel_down(self, node: str) -> None:
+        task = self._down.pop(node, None)
+        if task is not None:
+            task.cancel()
+
+    async def _down_after_grace(self, node: str, event: dict[str, Any]) -> None:
+        """Tell someone a connection is down, unless it came back first."""
+        await asyncio.sleep(self.grace_s)
+        if self._down.get(node) is asyncio.current_task():
+            del self._down[node]
+        if not self.config.enabled:
+            return
+        try:
+            await self._deliver(event)
+        except Exception:
+            logger.exception("Alerting failed for %s", event.get("type"))
+
+    async def _deliver(self, event: dict[str, Any]) -> None:
+        kind = str(event.get("type") or "")
         alert = describe(event)
         if alert is None:
             return
@@ -218,10 +298,14 @@ class AlertManager:
 
         key = dedup_key(event)
         cooldown = min(r.cooldown_s for r in rules)
-        if not self._due(key, cooldown):
+        if not self._cooled_down(key, cooldown):
             return
         if not self._within_rate_limit():
+            # Held back, not spent: the cooldown only starts once it is sent.
+            self._suppressed[key] = self._suppressed.get(key, 0) + 1
             return
+        now = self._stamp(key)
+        await self._remember(key, now)
 
         held = self._suppressed.pop(key, 0)
         if held:
@@ -238,13 +322,11 @@ class AlertManager:
     # Deciding whether to speak
     # -------------------------------------------------------------------------
 
-    def _due(self, key: str, cooldown: float) -> bool:
-        now = self._now()
+    def _cooled_down(self, key: str, cooldown: float) -> bool:
         last = self._last_sent.get(key)
-        if last is not None and now - last < cooldown:
+        if last is not None and self._now() - last < cooldown:
             self._suppressed[key] = self._suppressed.get(key, 0) + 1
             return False
-        self._last_sent[key] = now
         return True
 
     def _within_rate_limit(self) -> bool:
@@ -254,9 +336,24 @@ class AlertManager:
         if len(self._recent) >= RATE_LIMIT:
             logger.warning("Alert rate limit reached; holding back")
             return False
-        self._recent.append(now)
         return True
 
+    def _stamp(self, key: str) -> float:
+        """Count an alert as sent, before anything is awaited."""
+        now: float = self._now()
+        self._last_sent[key] = now
+        self._recent.append(now)
+        return now
+
+    async def _remember(self, key: str, ts: float) -> None:
+        """Write it to the ledger. A failure here never stops the alert."""
+        if self._ledger is None:
+            return
+        try:
+            await asyncio.to_thread(self._ledger.add, key, ts, ts - LEDGER_RETAIN_S)
+        except Exception:
+            logger.exception("Could not record the alert in the ledger")
+
     def _flaps(self, event: dict[str, Any]) -> bool:
         """Is this health event part of a connection flapping up and down?
 

backend/fluksio/main.py

@@ -30,6 +30,7 @@ from fluksio.core.config import settings
 from fluksio.core.db import engine as db_engine
 from fluksio.core.db import prepare
 from fluksio.flow import logs, modules
+from fluksio.flow.alert_ledger import AlertLedger
 from fluksio.flow.alerts import AlertManager
 from fluksio.flow.artifacts import ArtifactStore, VolatileStore
 from fluksio.flow.controller import FlowController, RebuildBusy
@@ -284,7 +285,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         # Connectors register their node types before any flow is built with them.
         load_plugins()
 
-        alerts = AlertManager(event_bus, config=read_alerts_config())
+        alerts = AlertManager(
+            event_bus, config=read_alerts_config(), ledger=AlertLedger()
+        )
         execution = ExecutionService(
             queue=_work_queue(),
             max_workers=settings.FLOW_MAX_WORKERS,

backend/fluksio/models.py

@@ -586,6 +586,27 @@ class ConnectorState(SQLModel, table=True):
     updated_at: datetime = Field(default_factory=get_datetime_utc, sa_type=UTCDateTime)
 
 
+# -----------------------------------------------------------------------------
+# Alerts sent
+#
+# What the alert manager has told someone, so its cooldowns and its ceiling per
+# hour hold across a restart. A node that stays down would otherwise be
+# announced again by every one.
+# -----------------------------------------------------------------------------
+
+
+class AlertSent(SQLModel, table=True):
+    """One alert that was delivered, by the key it deduplicates on."""
+
+    __tablename__ = "alert_sent"
+
+    id: int | None = Field(default=None, primary_key=True)
+    #: `alerts.dedup_key`: the event type and the node or flow it is about.
+    key: str = Field(max_length=255)
+    #: Seconds since the epoch, from the manager's own clock.
+    ts: float = Field(index=True)
+
+
 # -----------------------------------------------------------------------------
 # Worker tokens
 #

backend/tests/flow/test_alerts.py

@@ -9,6 +9,7 @@ import pytest
 from fluksio.flow.alerts import (
     ALERTING_EVENTS,
     FLAP_THRESHOLD,
+    LEDGER_RETAIN_S,
     RATE_LIMIT,
     Alert,
     AlertManager,
@@ -20,6 +21,20 @@ from fluksio.flow.alerts import (
 from fluksio.flow.events import EventBus
 
 
+class Ledger:
+    """What the database holds, without one."""
+
+    def __init__(self) -> None:
+        self.rows: list[tuple[str, float]] = []
+
+    def load(self, since: float) -> list[tuple[str, float]]:
+        return [row for row in self.rows if row[1] >= since]
+
+    def add(self, key: str, ts: float, keep_since: float) -> None:
+        self.rows = [row for row in self.rows if row[1] >= keep_since]
+        self.rows.append((key, ts))
+
+
 class Clock:
     """A hand-wound clock, so cooldowns take no real time."""
 
@@ -39,7 +54,9 @@ def manager(clock: Clock) -> tuple[AlertManager, list[Alert]]:
         channels=[Channel(name="phone", kind="ntfy", config={"topic": "t"})],
         rules=[Rule(events=[], channels=["phone"], cooldown_s=900)],
     )
-    alerts = AlertManager(EventBus(), config=config, now=clock)
+    # No grace: these tests are about what is sent, not how long a connection
+    # has to stay down first.
+    alerts = AlertManager(EventBus(), config=config, now=clock, grace_s=0)
 
     async def capture(channel, alert):
         sent.append(alert)
@@ -153,6 +170,174 @@ def test_a_storm_is_capped():
     assert len(sent) == RATE_LIMIT
 
 
+def test_a_cooldown_survives_a_restart():
+    """A node that stays down was announced again by every restart."""
+    clock, ledger = Clock(), Ledger()
+
+    async def scenario() -> list[Alert]:
+        first, sent_before = manager(clock)
+        first._ledger = ledger
+        await first.handle(error_event())
+        clock.advance(120)
+
+        second, sent_after = manager(clock)
+        second._ledger = ledger
+        await second.restore()
+        await second.handle(error_event())
+        assert len(sent_before) == 1
+        clock.advance(1000)
+        await second.handle(error_event())
+        return sent_after
+
+    sent_after = asyncio.run(scenario())
+
+    # Silent inside the cooldown, and speaking again once it has passed.
+    assert len(sent_after) == 1
+
+
+def test_the_ceiling_per_hour_survives_a_restart():
+    clock, ledger = Clock(), Ledger()
+    clock.now = 10_000.0
+    ledger.rows = [
+        (f"node_error:heating.n{i}", clock.now - 60) for i in range(RATE_LIMIT)
+    ]
+    alerts, sent = manager(clock)
+    alerts._ledger = ledger
+
+    async def scenario():
+        await alerts.restore()
+        await alerts.handle(error_event("heating.new"))
+
+    asyncio.run(scenario())
+
+    assert sent == []
+
+
+def test_a_clock_set_back_does_not_hold_a_key_silent():
+    clock, ledger = Clock(), Ledger()
+    clock.now = 100.0
+    ledger.rows = [("node_error:heating.pump", 50_000.0)]
+    alerts, sent = manager(clock)
+    alerts._ledger = ledger
+
+    async def scenario():
+        await alerts.restore()
+        clock.advance(1000)
+        await alerts.handle(error_event())
+
+    asyncio.run(scenario())
+
+    assert len(sent) == 1
+
+
+def test_sent_alerts_are_written_to_the_ledger_and_old_ones_dropped():
+    clock, ledger = Clock(), Ledger()
+    clock.now = LEDGER_RETAIN_S * 2
+    ledger.rows = [("node_error:heating.old", 1.0)]
+    alerts, _sent = manager(clock)
+    alerts._ledger = ledger
+
+    asyncio.run(alerts.handle(error_event()))
+
+    assert ledger.rows == [("node_error:heating.pump", clock.now)]
+
+
+def test_a_ledger_that_fails_does_not_stop_the_alert():
+    clock = Clock()
+    alerts, sent = manager(clock)
+
+    class Broken:
+        def load(self, since):
+            raise OSError("disk")
+
+        def add(self, key, ts, keep_since):
+            raise OSError("disk")
+
+    alerts._ledger = Broken()  # type: ignore[assignment]
+
+    async def scenario():
+        await alerts.restore()
+        await alerts.handle(error_event())
+
+    asyncio.run(scenario())
+
+    assert len(sent) == 1
+
+
+def test_an_alert_held_back_by_the_ceiling_does_not_spend_its_cooldown():
+    clock = Clock()
+    alerts, sent = manager(clock)
+
+    async def scenario():
+        for i in range(RATE_LIMIT):
+            await alerts.handle(error_event(f"heating.n{i}"))
+        await alerts.handle(error_event("heating.late"))
+        # An hour on, the ceiling has cleared and the same fault is still due.
+        clock.advance(3601)
+        await alerts.handle(error_event("heating.late"))
+
+    asyncio.run(scenario())
+
+    assert [a.node for a in sent][-1] == "heating.late"
+    assert "1 more since the last alert" in sent[-1].body
+
+
+def down(node: str = "heating.mqtt") -> dict:
+    return {"type": "node_health", "node": node, "health": "down", "detail": "refused"}
+
+
+def up(node: str = "heating.mqtt") -> dict:
+    return {"type": "node_health", "node": node, "health": "ok"}
+
+
+def with_grace(grace_s: float) -> tuple[AlertManager, list[Alert]]:
+    alerts, sent = manager(Clock())
+    alerts.grace_s = grace_s
+    return alerts, sent
+
+
+def test_a_connection_that_comes_back_within_the_grace_alerts_nobody():
+    alerts, sent = with_grace(0.05)
+
+    async def scenario():
+        await alerts.handle(down())
+        await alerts.handle(up())
+        await asyncio.sleep(0.1)
+
+    asyncio.run(scenario())
+
+    assert sent == []
+
+
+def test_a_connection_that_stays_down_is_told_after_the_grace():
+    alerts, sent = with_grace(0.05)
+
+    async def scenario():
+        await alerts.handle(down())
+        assert sent == []
+        await asyncio.sleep(0.1)
+
+    asyncio.run(scenario())
+
+    assert [a.title for a in sent] == ["heating.mqtt lost its connection"]
+
+
+def test_a_pending_alert_is_dropped_when_the_manager_stops():
+    alerts, sent = with_grace(0.05)
+
+    async def scenario():
+        await alerts.handle(down())
+        runner = asyncio.create_task(alerts.run())
+        await asyncio.sleep(0.01)
+        runner.cancel()
+        await asyncio.gather(runner, return_exceptions=True)
+        await asyncio.sleep(0.1)
+
+    asyncio.run(scenario())
+
+    assert sent == []
+
+
 def test_nothing_is_sent_without_a_rule():
     clock = Clock()
     alerts, sent = manager(clock)

backend/tests/test_alert_ledger.py

@@ -0,0 +1,26 @@
+"""The alert ledger: what the manager remembers across a restart."""
+
+import pytest
+from sqlalchemy import delete
+from sqlmodel import Session
+
+from fluksio.core.db import engine
+from fluksio.flow.alert_ledger import AlertLedger
+from fluksio.models import AlertSent
+
+
+@pytest.fixture(autouse=True)
+def _empty(db: Session):  # noqa: ARG001 - the schema has to exist first
+    with Session(engine) as session:
+        session.exec(delete(AlertSent))
+        session.commit()
+    yield
+
+
+def test_a_sent_alert_comes_back_and_old_ones_are_pruned():
+    ledger = AlertLedger()
+    ledger.add("node_error:house.calc", 100.0, keep_since=0.0)
+    ledger.add("node_health:house.mqtt", 5000.0, keep_since=1000.0)
+
+    assert ledger.load(since=0.0) == [("node_health:house.mqtt", 5000.0)]
+    assert ledger.load(since=6000.0) == []

docs/interface/operations.md

@@ -83,9 +83,13 @@ matches every supported alert event, including future additions.
 | `queue_unavailable` | the work queue becomes unreachable |
 | `run_finished` | a batch run ends in error |
 
-Repeated faults are deduplicated with a 15-minute default cooldown.
-Flapping connections are suppressed, and the engine caps alert dispatches at
-ten per hour. A quiet channel may be cooling down; **Test** helps distinguish
+Repeated faults are deduplicated with a 15-minute default cooldown. The
+engine remembers what it sent in its database, so a restart does not start
+every cooldown over, and the ten-per-hour cap holds across one too. Cooldowns
+longer than seven days are not kept. A `node_health` alert waits until the
+connection has been down for a minute, so a broker that restarts with the
+stack or a device that reconnects alerts nobody. Flapping connections are
+suppressed. A quiet channel may be cooling down; **Test** helps distinguish
 that from a broken configuration.
 
 ## Admin: users and clients