kipuka/ha/
mod.rs

1//! High-Availability subsystem for multi-CA failover.
2//!
3//! Implements RHELBU-3536 requirements R1 through R6:
4//! - R1: Multiple CA backend support with independent health tracking
5//! - R2: Circuit-breaker pattern with configurable cooldown
6//! - R3: Pluggable failover strategies (active-passive, round-robin, weighted, latency)
7//! - R4: Health probes with state machine transitions
8//! - R5: Automatic failover on CA unavailability
9//! - R6: Graceful degradation when all CAs are unhealthy
10
11pub mod health;
12pub mod pool;
13pub mod strategy;
14
15pub use health::{HealthChecker, HealthConfig, HealthState};
16pub use pool::{CaConnection, CaId, CaPool, CaStatus};
17pub use strategy::{FailoverStrategy, FallbackBehavior, StrategySelector};
18
19use std::sync::Arc;
20use tokio::sync::watch;
21use tracing::{info, warn};
22
23/// Central coordinator for the HA subsystem.
24///
25/// Owns the [`CaPool`] and [`HealthChecker`], wiring health state updates
26/// into pool availability decisions. The pool uses the configured
27/// [`FailoverStrategy`] to select a CA for each enrollment request.
28pub struct HaManager {
29    pool: Arc<CaPool>,
30    health_checker: HealthChecker,
31    shutdown_tx: watch::Sender<bool>,
32    shutdown_rx: watch::Receiver<bool>,
33}
34
35impl HaManager {
36    /// Build a new HA manager from pool and health configuration.
37    pub fn new(pool: Arc<CaPool>, health_config: HealthConfig) -> Self {
38        let (shutdown_tx, shutdown_rx) = watch::channel(false);
39        let health_checker = HealthChecker::new(Arc::clone(&pool), health_config);
40        Self {
41            pool,
42            health_checker,
43            shutdown_tx,
44            shutdown_rx,
45        }
46    }
47
48    /// Start background health checking.
49    ///
50    /// Spawns a tokio task that periodically probes each CA backend and
51    /// updates the pool's availability map. The task runs until
52    /// [`HaManager::shutdown`] is called.
53    pub async fn start(&self) -> tokio::task::JoinHandle<()> {
54        let checker = self.health_checker.clone();
55        let mut rx = self.shutdown_rx.clone();
56
57        info!("HA manager starting health check loop");
58
59        tokio::spawn(async move {
60            loop {
61                if *rx.borrow() {
62                    break;
63                }
64                checker.run_probes().await;
65
66                tokio::select! {
67                    _ = tokio::time::sleep(checker.interval()) => {}
68                    _ = rx.changed() => {
69                        info!("HA manager received shutdown signal");
70                        break;
71                    }
72                }
73            }
74        })
75    }
76
77    /// Signal the health checker to stop.
78    pub fn shutdown(&self) {
79        let _ = self.shutdown_tx.send(true);
80        warn!("HA manager shutting down");
81    }
82
83    /// Reference to the managed CA pool.
84    pub fn pool(&self) -> &Arc<CaPool> {
85        &self.pool
86    }
87}
88
89/// Build explicitly configured HA routing and probes before server startup.
90pub fn build_manager(
91    config: Arc<crate::config::Config>,
92    cas: Arc<indexmap::IndexMap<String, Arc<crate::state::CaState>>>,
93    hsm: Option<Arc<kipuka_hsm::HsmContext>>,
94) -> Option<Arc<HaManager>> {
95    let ha = config.ha.as_ref().filter(|ha| ha.enabled)?;
96    let connections = config
97        .cas
98        .iter()
99        .enumerate()
100        .map(|(priority, ca)| CaConnection {
101            id: CaId(ca.id.clone()),
102            endpoint: ha
103                .endpoints
104                .get(&ca.id)
105                .cloned()
106                .unwrap_or_else(|| format!("local:{}", ca.id)),
107            weight: 1,
108            priority: priority as u32,
109        })
110        .collect();
111    let pool = Arc::new(CaPool::new(
112        connections,
113        pool::PoolConfig {
114            strategy: ha.strategy.clone(),
115            fallback: FallbackBehavior::Reject,
116            circuit_breaker: pool::CircuitBreakerConfig {
117                failure_threshold: ha.failure_threshold,
118                cooldown: std::time::Duration::from_secs(ha.cooldown_secs),
119            },
120        },
121    ));
122    let health_config = HealthConfig {
123        probe_interval: std::time::Duration::from_secs(ha.probe_interval_secs),
124        probe_timeout: std::time::Duration::from_secs(ha.probe_timeout_secs),
125        ..HealthConfig::default()
126    };
127    let mut manager = HaManager::new(pool, health_config);
128    manager.health_checker = manager
129        .health_checker
130        .with_local(Arc::new(health::LocalProbeContext { config, cas, hsm }));
131    Some(Arc::new(manager))
132}
kipuka/ha/mod.rs

kipuka/ha/
mod.rs
