Collapse missed-block spam into storm mode

When a single proposer crosses missed_burst_threshold misses inside
missed_burst_window_minutes, per-slot alerts are suppressed in favor of
one burst-start alert, periodic backoff updates (15/30/60/120 min by
default, last value repeats), and a burst-resolved alert once the window
clears. Burst state is persisted so restarts don't re-fire transitions.

Author: edg-l

dora_monitor/config.example.yaml

@@ -52,6 +52,16 @@ heartbeat_slot_window: 256
 #   detailed  — per-client list with status / head / distance
 heartbeat_other_clients: "detailed"
 
+# Missed-block storm mode. When a single proposer hits
+# `missed_burst_threshold` misses inside the last `missed_burst_window_minutes`,
+# we stop posting per-slot alerts and switch to one "burst started" alert
+# plus periodic "still bursting" updates whose interval backs off through
+# `missed_burst_update_schedule_minutes` (last value repeats indefinitely).
+# A "burst resolved" alert fires once the window has been clear of misses.
+missed_burst_threshold: 5
+missed_burst_window_minutes: 15
+missed_burst_update_schedule_minutes: [15, 30, 60, 120]
+
 # Which checks to enable.
 checks:
   missed_blocks: true

dora_monitor/dora_monitor/checks.py

@@ -17,12 +17,70 @@ def _matches(name: str | None, needle: str) -> bool:
     return needle.lower() in name.lower()
 
 
+def _burst_update_interval_s(schedule_min: list[int], update_count: int) -> int:
+    """Pick the next update interval (seconds) for a burst.
+
+    The schedule is consumed in order; once past the end, the final value
+    is reused indefinitely (the cap). Defaults give 15min → 30min → 1h → 2h
+    and then 2h forever.
+    """
+    if not schedule_min:
+        return 120 * 60
+    idx = min(update_count, len(schedule_min) - 1)
+    return max(int(schedule_min[idx]), 1) * 60
+
+
 def check_missed_blocks(
     dora: DoraClient,
     notifier: Notifier,
     cfg: Config,
     state: State,
 ) -> None:
+    now = time.time()
+    window_s = max(cfg.missed_burst_window_minutes, 1) * 60
+    threshold = max(cfg.missed_burst_threshold, 1)
+
+    # 1. Age out the recent-misses window for every known proposer; resolve
+    #    any burst whose window is now empty; fire backoff updates for the rest.
+    for proposer in list(state.missed_recent.keys()):
+        cutoff = now - window_s
+        kept = [pair for pair in state.missed_recent[proposer] if pair[0] >= cutoff]
+        if kept:
+            state.missed_recent[proposer] = kept
+        else:
+            del state.missed_recent[proposer]
+
+    for proposer in list(state.burst_state.keys()):
+        burst = state.burst_state[proposer]
+        if proposer not in state.missed_recent:
+            duration_min = max(int((now - float(burst.get("started_ts", now))) / 60), 0)
+            notifier.send(
+                f":white_check_mark: *Missed-block burst resolved* — `{proposer}`: "
+                f"*{int(burst.get('total_misses', 0))}* missed blocks over {duration_min}min "
+                f"(first `{int(burst.get('first_slot', 0))}`, last `{int(burst.get('last_slot', 0))}`)"
+            )
+            del state.burst_state[proposer]
+            continue
+        interval = _burst_update_interval_s(
+            cfg.missed_burst_update_schedule_minutes,
+            int(burst.get("update_count", 0)),
+        )
+        if now - float(burst.get("last_update_ts", 0.0)) >= interval:
+            burst["last_update_ts"] = now
+            burst["update_count"] = int(burst.get("update_count", 0)) + 1
+            next_interval = _burst_update_interval_s(
+                cfg.missed_burst_update_schedule_minutes,
+                int(burst["update_count"]),
+            )
+            next_min = next_interval // 60
+            notifier.send(
+                f":fire: *Missed-block burst continues* — `{proposer}`: "
+                f"*{int(burst.get('total_misses', 0))}* missed blocks since start "
+                f"(latest slot `{int(burst.get('last_slot', 0))}`). "
+                f"Next update in ~{next_min}min."
+            )
+
+    # 2. Process new missed / orphaned slots from Dora.
     slots = dora.slots(limit=cfg.slot_scan_limit, with_orphaned=1, with_missing=1)
     for s in slots:
         proposer_name = s.get("proposer_name") or ""
@@ -32,9 +90,8 @@ def check_missed_blocks(
         status = (s.get("status") or "").lower()
         if status == "missing" and slot_num not in state.reported_missed_slots:
             state.reported_missed_slots.add(slot_num)
-            notifier.send(
-                f":warning: *Missed block* — slot `{slot_num}` "
-                f"(epoch {s.get('epoch')}) proposer `{proposer_name}` (idx {s.get('proposer')})"
+            _handle_missed_slot(
+                notifier, state, cfg, proposer_name, slot_num, s, now, window_s, threshold
             )
         elif status == "orphaned" and slot_num not in state.reported_orphan_slots:
             state.reported_orphan_slots.add(slot_num)
@@ -44,6 +101,57 @@ def check_missed_blocks(
             )
 
 
+def _handle_missed_slot(
+    notifier: Notifier,
+    state: State,
+    cfg: Config,
+    proposer: str,
+    slot: int,
+    raw: dict,
+    now: float,
+    window_s: int,
+    threshold: int,
+) -> None:
+    recent = state.missed_recent.setdefault(proposer, [])
+    recent.append([now, slot])
+    cutoff = now - window_s
+    state.missed_recent[proposer] = [pair for pair in recent if pair[0] >= cutoff]
+    recent = state.missed_recent[proposer]
+
+    burst = state.burst_state.get(proposer)
+    if burst:
+        # Already in storm mode: accumulate counters, suppress per-slot alert.
+        burst["total_misses"] = int(burst.get("total_misses", 0)) + 1
+        burst["last_slot"] = slot
+        return
+
+    if len(recent) >= threshold:
+        slots_csv = ", ".join(f"`{int(s)}`" for _, s in sorted(recent, key=lambda p: p[1]))
+        first_slot = min(int(s) for _, s in recent)
+        first_interval_min = _burst_update_interval_s(
+            cfg.missed_burst_update_schedule_minutes, 0
+        ) // 60
+        state.burst_state[proposer] = {
+            "started_ts": now,
+            "last_update_ts": now,
+            "first_slot": first_slot,
+            "last_slot": slot,
+            "total_misses": len(recent),
+            "update_count": 0,
+        }
+        notifier.send(
+            f":fire: *Missed-block burst* — `{proposer}`: *{len(recent)}* missed blocks "
+            f"in last {cfg.missed_burst_window_minutes}min — slots {slots_csv}. "
+            f"Further per-miss alerts suppressed; next update in ~{first_interval_min}min."
+        )
+        return
+
+    notifier.send(
+        f":warning: *Missed block* — slot `{slot}` "
+        f"(epoch {raw.get('epoch')}) proposer `{proposer}` (idx {raw.get('proposer')})"
+    )
+
+
 def check_client_head_forks(
     dora: DoraClient,
     notifier: Notifier,

dora_monitor/dora_monitor/config.py

@@ -34,6 +34,14 @@ class Config:
     heartbeat_slot_window: int = 256
     # "off" (skip), "summary" (one-line aggregate), "detailed" (per-client list)
     heartbeat_other_clients: str = "detailed"
+    # Storm mode for missed-block alerts: once a proposer hits
+    # `missed_burst_threshold` misses inside `missed_burst_window_minutes`,
+    # per-slot alerts are suppressed and replaced by one burst-start alert
+    # plus periodic "still bursting" updates whose interval backs off
+    # through `missed_burst_update_schedule_minutes` (last value repeats).
+    missed_burst_threshold: int = 5
+    missed_burst_window_minutes: int = 15
+    missed_burst_update_schedule_minutes: list[int] = field(default_factory=lambda: [15, 30, 60, 120])
     checks: Checks = field(default_factory=Checks)
 
 

dora_monitor/dora_monitor/discord.py

@@ -26,6 +26,7 @@
     ":turtle:": "\U0001f422",
     ":zap:": "⚡",
     ":package:": "\U0001f4e6",
+    ":fire:": "\U0001f525",
     ":rocket:": "\U0001f680",
     ":mag:": "\U0001f50d",
     ":desktop_computer:": "\U0001f5a5️",

dora_monitor/dora_monitor/state.py

@@ -19,6 +19,13 @@ class State:
     last_known_head: int = 0
     last_heartbeat_ts: float = 0.0
     client_versions: dict[str, str] = field(default_factory=dict)
+    # Sliding window of recent missed-block timestamps + slot numbers per
+    # proposer. Used to decide when a client's miss rate crosses the
+    # storm-mode threshold.
+    missed_recent: dict[str, list[list[float]]] = field(default_factory=dict)
+    # Active missed-block burst per proposer:
+    #   {"started_ts", "last_update_ts", "first_slot", "last_slot", "total_misses"}.
+    burst_state: dict[str, dict] = field(default_factory=dict)
 
     def to_json(self) -> dict:
         d = asdict(self)
@@ -44,6 +51,11 @@ def from_json(cls, d: dict) -> "State":
             last_heartbeat_ts=float(d.get("last_heartbeat_ts", 0.0)),
             client_versions=dict(d.get("client_versions", {})),
             pending_fork_ticks={k: int(v) for k, v in (d.get("pending_fork_ticks") or {}).items()},
+            missed_recent={
+                k: [[float(t), int(s)] for t, s in v]
+                for k, v in (d.get("missed_recent") or {}).items()
+            },
+            burst_state={k: dict(v) for k, v in (d.get("burst_state") or {}).items()},
         )