feat(cluster): distributed cluster infrastructure

Cargo.toml

@@ -22,7 +22,7 @@ members = [
 resolver = "2"
 
 [workspace.package]
-version = "0.0.3"
+version = "0.0.4"
 edition = "2024"
 rust-version = "1.94"
 license = "BUSL-1.1"

nodedb-cluster/src/bootstrap/bootstrap_fn.rs

@@ -35,7 +35,8 @@ pub(super) fn bootstrap(config: &ClusterConfig, catalog: &ClusterCatalog) -> Res
     );
 
     // Create MultiRaft with all groups (single-node, no peers).
-    let mut multi_raft = MultiRaft::new(config.node_id, routing.clone(), config.data_dir.clone());
+    let mut multi_raft = MultiRaft::new(config.node_id, routing.clone(), config.data_dir.clone())
+        .with_election_timeout(config.election_timeout_min, config.election_timeout_max);
     for group_id in routing.group_ids() {
         multi_raft.add_group(group_id, vec![])?;
     }
@@ -81,6 +82,7 @@ fn generate_cluster_id() -> u64 {
 mod tests {
     use super::*;
     use crate::catalog::ClusterCatalog;
+    use std::time::Duration;
 
     fn temp_catalog() -> (tempfile::TempDir, ClusterCatalog) {
         let dir = tempfile::tempdir().unwrap();
@@ -102,6 +104,8 @@ mod tests {
             force_bootstrap: false,
             join_retry: Default::default(),
             swim_udp_addr: None,
+            election_timeout_min: Duration::from_millis(150),
+            election_timeout_max: Duration::from_millis(300),
         };
 
         let state = bootstrap(&config, &catalog).unwrap();

nodedb-cluster/src/bootstrap/config.rs

@@ -91,6 +91,10 @@ pub struct ClusterConfig {
     /// [`crate::spawn_swim`] after the cluster is up and feed the
     /// seed list from `seed_nodes`.
     pub swim_udp_addr: Option<SocketAddr>,
+    /// Raft election timeout range. Controls how long a follower waits
+    /// before starting an election after losing contact with the leader.
+    pub election_timeout_min: Duration,
+    pub election_timeout_max: Duration,
 }
 
 /// Result of cluster startup — everything needed to run the Raft loop.

nodedb-cluster/src/bootstrap/join.rs

@@ -288,7 +288,8 @@ fn apply_join_response(
     //    learners). A learner-started group boots in the `Learner`
     //    role and will not run an election until a subsequent
     //    `PromoteLearner` conf change is applied.
-    let mut multi_raft = MultiRaft::new(config.node_id, routing.clone(), config.data_dir.clone());
+    let mut multi_raft = MultiRaft::new(config.node_id, routing.clone(), config.data_dir.clone())
+        .with_election_timeout(config.election_timeout_min, config.election_timeout_max);
     for g in &resp.groups {
         let is_voter = g.members.contains(&config.node_id);
         let is_learner = g.learners.contains(&config.node_id);
@@ -450,6 +451,8 @@ mod tests {
             force_bootstrap: false,
             join_retry: Default::default(),
             swim_udp_addr: None,
+            election_timeout_min: Duration::from_millis(150),
+            election_timeout_max: Duration::from_millis(300),
         };
         let state1 = bootstrap(&config1, &catalog1).unwrap();
 
@@ -499,6 +502,8 @@ mod tests {
             force_bootstrap: false,
             join_retry: Default::default(),
             swim_udp_addr: None,
+            election_timeout_min: Duration::from_millis(150),
+            election_timeout_max: Duration::from_millis(300),
         };
 
         let lifecycle = ClusterLifecycleTracker::new();

nodedb-cluster/src/bootstrap/probe.rs

@@ -223,6 +223,8 @@ mod tests {
             force_bootstrap: false,
             join_retry: Default::default(),
             swim_udp_addr: None,
+            election_timeout_min: Duration::from_millis(150),
+            election_timeout_max: Duration::from_millis(300),
         }
     }
 

nodedb-cluster/src/bootstrap/restart.rs

@@ -35,7 +35,8 @@ pub(super) fn restart(
     // as a learner on restart; dropping the group entirely would
     // leave the node permanently without any copy of it and
     // silently broken.
-    let mut multi_raft = MultiRaft::new(config.node_id, routing.clone(), config.data_dir.clone());
+    let mut multi_raft = MultiRaft::new(config.node_id, routing.clone(), config.data_dir.clone())
+        .with_election_timeout(config.election_timeout_min, config.election_timeout_max);
     for (group_id, info) in routing.group_members() {
         let is_voter = info.members.contains(&config.node_id);
         let is_learner = info.learners.contains(&config.node_id);
@@ -91,6 +92,7 @@ mod tests {
     use super::super::bootstrap_fn::bootstrap;
     use super::*;
     use crate::catalog::ClusterCatalog;
+    use std::time::Duration;
 
     fn temp_catalog() -> (tempfile::TempDir, ClusterCatalog) {
         let dir = tempfile::tempdir().unwrap();
@@ -112,6 +114,8 @@ mod tests {
             force_bootstrap: false,
             join_retry: Default::default(),
             swim_udp_addr: None,
+            election_timeout_min: Duration::from_millis(150),
+            election_timeout_max: Duration::from_millis(300),
         };
 
         // Bootstrap first.

nodedb-cluster/src/circuit_breaker.rs

@@ -145,6 +145,26 @@ impl CircuitBreaker {
             .unwrap_or(CircuitState::Closed)
     }
 
+    /// Return the ids of every peer whose breaker is currently Open.
+    ///
+    /// Used by the reachability driver to find peers that need an
+    /// active probe — without a periodic poke these peers never
+    /// transition back to HalfOpen (no traffic → no `check()` call
+    /// → no cooldown re-evaluation).
+    pub fn open_peers(&self) -> Vec<u64> {
+        let peers = self.peers.read().unwrap_or_else(|p| p.into_inner());
+        peers
+            .iter()
+            .filter_map(|(id, b)| {
+                if b.state == CircuitState::Open {
+                    Some(*id)
+                } else {
+                    None
+                }
+            })
+            .collect()
+    }
+
     /// Get consecutive failure count for a peer.
     pub fn failure_count(&self, peer: u64) -> u32 {
         let peers = self.peers.read().unwrap_or_else(|p| p.into_inner());