Q&A: FederationHealthMonitor configuration — poll interval, score threshold, circuit breaker, SSE stream, Grafana monitoring

[web3guru888]

Q&A: FederationHealthMonitor Configuration

Answers to the most common configuration and design questions for FederationHealthMonitor (Issue #315).

---

Q1: How do I choose poll_interval_ms?

A: The right value depends on your SLA for detecting failures:

poll_interval_ms	Detection latency	CPU overhead	Use case
1,000 (1 s)	≤ 1s	Higher	Dev/test or latency-sensitive prod
5,000 (5 s)	≤ 5s	Low	Default — most production deployments
10,000 (10 s)	≤ 10s	Minimal	Low-priority background federations
30,000 (30 s)	≤ 30s	Near-zero	Cost-optimised or high-peer-count clusters

The poll loop calls snapshot() on all four components. Each snapshot() should be O(1) in-memory reads — no network calls. So even 1 s intervals are fine for local federations.

---

Q2: How does score_threshold interact with trip_count?

A: The circuit breaker trips when overall_score < score_threshold for trip_count consecutive polls:

circuit_open = (consecutive_low >= trip_count)

score_threshold	trip_count	Meaning
0.5 (default)	3 (default)	Circuit opens after 3 consecutive polls below 50% overall health
0.7	2	More sensitive — trips faster on moderate degradation
0.3	5	Tolerant — only trips on severe, sustained failure

Recommendation: Start with defaults (0.5 / 3). Tune trip_count higher if you experience false positives during transient network blips.

---

Q3: What happens when the circuit breaker opens?

A: Three things happen simultaneously:

1. FederationHealthEvent.circuit_open = True is emitted on the SSE stream
2. CognitiveCycle._phase_federation() writes federation.circuit_open = True to the Blackboard
3. Any Blackboard subscriber that checks federation.circuit_open can pause federation operations

What does NOT happen automatically: FederatedTaskRouter.route() does not internally query the health monitor — you must wire the circuit breaker check into your routing middleware. Example:

async def route_with_health_guard(
    task: TaskEnvelope,
    router: FederatedTaskRouter,
    monitor: FederationHealthMonitor,
) -> RoutingResult:
    snap = await monitor.snapshot()
    if snap.circuit_open:
        return RoutingResult(decision=RoutingDecision.FALLBACK, target_peer=None, ...)
    return await router.route(task)

---

Q4: How do I reset the circuit breaker in production?

A: Call reset_circuit() — it resets consecutive_low = 0, which immediately closes the circuit:

# Via CognitiveCycle
await cognitive_cycle._health_monitor.reset_circuit()

# Or via your ops CLI
python -c "
import asyncio
from your_module import get_health_monitor
asyncio.run(get_health_monitor().reset_circuit())
"

The decision to require manual reset is intentional: in a BFT system, automatic recovery from a cluster-wide failure risks masking an ongoing fault. Add an auto_reset_after_s parameter if your use case requires automatic recovery.

---

Q5: How are the component score weights calibrated?

A: The default weights reflect failure criticality:

overall_score = 0.30 × gateway     ← peer connectivity is foundational
              + 0.25 × blackboard  ← state sync drives all consumers
              + 0.25 × router      ← routing failure drops tasks silently
              + 0.20 × consensus   ← agreement layer — most resilient

To override for your deployment, subclass InMemoryFederationHealthMonitor and override _collect():

class MyHealthMonitor(InMemoryFederationHealthMonitor):
    _WEIGHTS = {"gateway": 0.40, "blackboard": 0.20, "router": 0.20, "consensus": 0.20}

    async def _collect(self) -> FederationHealthEvent:
        event = await super()._collect()
        # recompute overall_score with custom weights
        scores = {c.name: c.score for c in event.components}
        overall = sum(self._WEIGHTS[k] * v for k, v in scores.items())
        return FederationHealthEvent(
            timestamp_ms=event.timestamp_ms,
            components=event.components,
            overall_score=overall,
            circuit_open=event.circuit_open,
            consecutive_low=event.consecutive_low,
        )

---

Q6: How do I subscribe to the SSE stream from a FastAPI endpoint?

A: Use StreamingResponse with text/event-stream:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

@app.get("/federation/health/stream")
async def health_stream(monitor: FederationHealthMonitor = Depends(get_monitor)):
    async def generate():
        async for event in monitor.health_stream():
            data = json.dumps({
                "overall_score": event.overall_score,
                "circuit_open":  event.circuit_open,
                "components":    [
                    {"name": c.name, "score": c.score, "health": c.health}
                    for c in event.components
                ],
            })
            yield f"data: {data}\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

---

Q7: How do I build a Grafana dashboard for federation health?

A: Four panels cover the essentials:

Panel	PromQL	Visualization
Overall health score	asi_federation_health_score{component="overall"}	Stat (thresholds: ≥0.8 green, ≥0.5 yellow, <0.5 red)
Per-component scores	asi_federation_health_score	Time series, 4 lines
Circuit breaker status	asi_federation_circuit_open	State timeline
Component health state	asi_federation_component_health{health="failed"}	Alert list

Alert rule (Prometheus format):

groups:
  - name: federation
    rules:
      - alert: FederationCircuitOpen
        expr: asi_federation_circuit_open == 1
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Federation circuit breaker open"
          runbook: "See Issue #315 for reset procedure"