Require N consecutive ticks of non-canonical head before fork alert

Propagation timing routinely produces transient 1-2 slot leads or lags
that the previous code surfaced as fork alerts (and an immediate
resolved alert a tick later). Configurable via fork_confirm_ticks
(default 3 = ~90s at the default 30s poll), persisted per-client in
the dedup state so it survives restarts.

Author: edg-l

dora_monitor/config.example.yaml

@@ -20,6 +20,12 @@ slot_scan_limit: 64
 # Sync lag threshold (slots behind the canonical head) before alerting.
 sync_lag_threshold: 16
 
+# Number of consecutive polls a matched client must be on a non-canonical
+# head before we fire a fork alert. Filters propagation-timing jitter
+# (1-2 slot leads/lags that self-resolve in the next poll). With
+# poll_interval=30, fork_confirm_ticks=3 = ~90s of confirmation.
+fork_confirm_ticks: 3
+
 # Path for persisted dedup state. Use null for in-memory only.
 state_file: "./dora_monitor_state.json"
 

dora_monitor/dora_monitor/checks.py

@@ -75,6 +75,7 @@ def check_client_head_forks(
     state.last_known_head = max(state.last_known_head, canonical_slot)
 
     current_forked: set[str] = set()
+    forked_candidates: set[str] = set()
     current_lagging: set[str] = set()
     current_offline: set[str] = set()
     matched_clients: dict[str, dict] = {}
@@ -110,7 +111,7 @@ def check_client_head_forks(
                 continue
 
             if cfg.checks.forks and not is_canonical:
-                current_forked.add(name)
+                forked_candidates.add(name)
 
             if cfg.checks.sync_lag:
                 distance = canonical_slot - client_head
@@ -131,6 +132,24 @@ def check_client_head_forks(
         state.offline_clients = current_offline
 
     if cfg.checks.forks:
+        # Bump the consecutive-tick counter for clients seen on a
+        # non-canonical fork this tick; reset for everyone else. Only treat
+        # a client as truly forked once the counter crosses fork_confirm_ticks
+        # to filter propagation-timing noise (1-2 slot jitter that resolves
+        # within a poll or two).
+        threshold = max(cfg.fork_confirm_ticks, 1)
+        seen_matched = set(matched_clients.keys())
+        for name in list(state.pending_fork_ticks.keys()):
+            if name not in seen_matched:
+                # Client vanished from the payload (lost from network view);
+                # drop the pending counter so it doesn't persist forever.
+                del state.pending_fork_ticks[name]
+        for name in forked_candidates:
+            state.pending_fork_ticks[name] = state.pending_fork_ticks.get(name, 0) + 1
+        for name in seen_matched - forked_candidates:
+            state.pending_fork_ticks.pop(name, None)
+        current_forked = {n for n, c in state.pending_fork_ticks.items() if c >= threshold}
+
         new_forked = current_forked - state.forked_clients
         resolved_forked = state.forked_clients - current_forked
         for name in sorted(new_forked):

dora_monitor/dora_monitor/config.py

@@ -22,6 +22,10 @@ class Config:
     poll_interval: int = 30
     slot_scan_limit: int = 64
     sync_lag_threshold: int = 16
+    # Number of consecutive polls a matched client must be on a non-canonical
+    # head before we fire a fork alert. Filters out propagation-timing noise
+    # where one client briefly leads or lags by a slot or two.
+    fork_confirm_ticks: int = 3
     state_file: str | None = "./dora_monitor_state.json"
     http_timeout: int = 10
     debug: bool = False

dora_monitor/dora_monitor/state.py

@@ -13,6 +13,9 @@ class State:
     offline_clients: set[str] = field(default_factory=set)
     lagging_clients: set[str] = field(default_factory=set)
     forked_clients: set[str] = field(default_factory=set)
+    # Per-client consecutive non-canonical observations. Used to require
+    # N ticks of confirmation before firing a fork alert.
+    pending_fork_ticks: dict[str, int] = field(default_factory=dict)
     last_known_head: int = 0
     last_heartbeat_ts: float = 0.0
     client_versions: dict[str, str] = field(default_factory=dict)
@@ -40,6 +43,7 @@ def from_json(cls, d: dict) -> "State":
             last_known_head=int(d.get("last_known_head", 0)),
             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()},
         )