Show & Tell: FederationHealthMonitor — weighted health score, SSE stream, and cluster circuit breaker

[web3guru888]

Overview

FederationHealthMonitor is the final sub-phase of Phase 9 — the observability layer that ties together all four federation components into a single health signal, with SSE streaming and a cluster-level circuit breaker.

Filed as: Issue #315

---

Component Map

FederationGateway ──────────┐
FederatedBlackboard ─────── ├──► InMemoryFederationHealthMonitor
FederatedTaskRouter ─────── │        │
FederatedConsensus ─────────┘        │
                                      ├──► FederationHealthEvent (SSE stream)
                                      ├──► HealthMonitorSnapshot
                                      ├──► Prometheus metrics (5)
                                      └──► Cluster circuit breaker

---

Health Score Computation

On every poll tick (default: 5 seconds), the monitor calls snapshot() on all four components and computes:

overall_score = 0.30 × gateway_score
              + 0.25 × blackboard_score
              + 0.25 × router_score
              + 0.20 × consensus_score

Component scoring logic

Component	Score = 1.0 when…	Score = 0.0 when…
Gateway	100% peers healthy	0 peers known
Blackboard	clock_drift ≤ 100ms + subscribers active	drift > 500ms
Router	0% fallback routes	≥ 50% fallback
Consensus	0% aborted proposals	≥ 33% aborted

---

Circuit Breaker

CLOSED ─[overall_score < 0.5 for ≥ 3 consecutive polls]─► OPEN
OPEN   ─[reset_circuit() called]─────────────────────────► CLOSED

• When OPEN: FederatedTaskRouter.route() returns RoutingDecision.FALLBACK immediately
• CognitiveCycle._phase_federation() writes federation.circuit_open = True to Blackboard
• Manual reset only — automatic recovery in a BFT system requires human confirmation

---

SSE Health Stream

Each subscriber gets a dedicted asyncio.Queue(maxsize=100). Slow consumers are evicted (not blocked):

async def health_stream(self) -> AsyncIterator[FederationHealthEvent]:
    queue: asyncio.Queue[FederationHealthEvent] = asyncio.Queue(maxsize=100)
    self._subscribers.add(queue)
    try:
        while True:
            yield await queue.get()
    finally:
        self._subscribers.discard(queue)

This pattern prevents the poll loop from being blocked by a slow HTTP client.

---

FederationHealthEvent Data Model

@dataclass(frozen=True)
class FederationHealthEvent:
    timestamp_ms:    int
    components:      tuple[ComponentScore, ...]   # always 4
    overall_score:   float                        # 0.0 – 1.0
    circuit_open:    bool
    consecutive_low: int

Example event payload (JSON)

{
  "timestamp_ms": 1744497000000,
  "components": [
    {"name": "gateway",    "health": "healthy",  "score": 0.92, "details": "11/12 healthy peers"},
    {"name": "blackboard", "health": "healthy",  "score": 0.95, "details": "drift=43ms"},
    {"name": "router",     "health": "degraded", "score": 0.72, "details": "fallback=14.0%"},
    {"name": "consensus",  "health": "healthy",  "score": 0.88, "details": "abort_ratio=4.0%"}
  ],
  "overall_score": 0.868,
  "circuit_open": false,
  "consecutive_low": 0
}

---

Prometheus Metrics

Metric	Type	Labels	Description
asi_federation_health_score	Gauge	component	Health score per component + overall
asi_federation_health_polls_total	Counter	—	Total poll iterations
asi_federation_circuit_trips_total	Counter	—	Total circuit breaker trips
asi_federation_circuit_open	Gauge	—	1 if circuit is currently open
asi_federation_component_health	Gauge	component, health	1 for current health state

PromQL Alerts

# Alert: federation circuit breaker open
ALERT FederationCircuitOpen
  IF asi_federation_circuit_open == 1
  FOR 1m
  LABELS { severity="critical" }
  ANNOTATIONS { summary="Federation circuit breaker tripped — check #315 runbook" }

# Alert: gateway degraded for 2 minutes
ALERT GatewayDegraded
  IF asi_federation_health_score{component="gateway"} < 0.5
  FOR 2m
  LABELS { severity="warning" }

---

12 Test Targets

#	Test	Validates
1	test_score_gateway_all_healthy	12/12 peers → score=1.0
2	test_score_gateway_no_peers	0 peers → FAILED, score=0.0
3	test_score_gateway_degraded	6/12 peers → DEGRADED
4	test_score_blackboard_high_drift	drift=600ms → score≤0.6
5	test_score_router_high_fallback	60% fallback → FAILED
6	test_score_consensus_high_abort	40% abort_ratio → DEGRADED
7	test_poll_loop_emits_events	3 ticks → 3 events
8	test_circuit_breaker_trips	3 low polls → circuit_open=True
9	test_circuit_breaker_reset	reset_circuit() → circuit_open=False
10	test_health_stream_subscriber	subscriber receives events
11	test_snapshot_reflects_state	snapshot() matches poll state
12	test_build_health_monitor_factory	factory returns correct type

---

Open Questions

1. Should the circuit breaker auto-close? Auto-reset after N minutes of good polls (e.g. auto_reset_after_s=300) vs. always manual?
2. Health stream backpressure: Should slow consumers be evicted or should we use a deque with maxlen instead of a Queue?
3. Sepolia integration: Phase 8.5 sepolia-exporter — should Phase 9.5 add asi_federation_commits_onchain_total counter anchored to CommitCertificate hashes?
4. Score weights: Are the weights (0.30/0.25/0.25/0.20) correct for your use case? Gateway is weighted highest because connectivity loss is the most catastrophic failure mode.