Promote cluster.peers learners as they catch up; no 503 after /health is 200

hyperbytedb/src/adapters/http/peer_handlers.rs

@@ -945,7 +945,18 @@ pub async fn handle_sync_trigger(State(state): State<Arc<AppState>>) -> impl Int
             .unwrap_or(false);
         let wal_seq = wal.last_sequence().await.unwrap_or(0);
 
-        let result = if !has_data && wal_seq == 0 {
+        // A node that already synced (Active, nothing flagged) holds no data
+        // only because the cluster has none yet. Re-joining it flipped it to
+        // Syncing, so the first writes after `/health` turned 200 got 503s;
+        // a reconnect stays Active for a small gap. A node the operator
+        // brought up knowing no peers starts flagged `needs_sync` and still
+        // joins.
+        let synced_before = {
+            let m = membership_clone.read().await;
+            m.get_node(node_id)
+                .is_some_and(|n| n.state == NodeState::Active && !n.needs_sync)
+        };
+        let result = if !has_data && wal_seq == 0 && !synced_before {
             tracing::info!("sync trigger: new node, running join_and_sync");
             {
                 let mut m = membership_clone.write().await;

hyperbytedb/src/adapters/http/ping.rs

@@ -66,7 +66,7 @@ fn json_response(code: StatusCode, body: String) -> Response {
 /// were still a learner) must not keep `/health` on 503 — that deadlocks
 /// StatefulSet OrderedReady and, via self-echoed `/health` bodies, the
 /// peer heartbeat that would otherwise promote us.
-fn raft_is_serving(state: &AppState) -> bool {
+pub(crate) fn raft_is_serving(state: &AppState) -> bool {
     let Some(raft) = state.raft.as_ref() else {
         return false;
     };

hyperbytedb/src/adapters/http/write.rs

@@ -71,9 +71,16 @@ pub async fn handle_write(
                     }
                     return Ok(resp);
                 }
-                NodeState::Syncing | NodeState::Joining => {
+                // As `/health`: a node Raft already serves as a voter is
+                // Active even while its own record still says Syncing (it
+                // converges a few seconds later). Refusing here made the
+                // first writes after `/health` turned 200 fail with 503.
+                NodeState::Syncing | NodeState::Joining
+                    if !super::ping::raft_is_serving(&state) =>
+                {
                     return Ok(StatusCode::SERVICE_UNAVAILABLE.into_response());
                 }
+                NodeState::Syncing | NodeState::Joining => {}
                 // Listed explicitly rather than `_`: a catch-all here silently
                 // admits client writes for any state added later, which is how
                 // a node being removed kept accepting them.

hyperbytedb/src/application/cluster/raft_formation.rs

@@ -8,35 +8,73 @@ use crate::ports::peer_http::PeerHttp;
 use openraft::BasicNode;
 use openraft::RaftMetrics;
 
-/// Returns true when every non-leader peer in `peer_ids` has replicated the
-/// leader's current log tail (per Raft replication metrics).
-fn learners_caught_up(
+/// Voters the formation keeps: the bootstrap node and up to four others;
+/// the rest stay learners.
+const MAX_VOTERS: usize = 5;
+
+/// The bootstrap node's next step for the peers it has discovered.
+#[derive(Debug, Default, PartialEq, Eq)]
+struct FormationStep {
+    /// Discovered peers not yet in the Raft membership.
+    add_learners: Vec<u64>,
+    /// The voter set to change to: current voters plus every learner that
+    /// has caught up, up to [`MAX_VOTERS`].
+    promote_to: Option<BTreeSet<u64>>,
+}
+
+/// Plan one pass of cluster formation from the leader's Raft metrics.
+///
+/// Each learner is promoted as soon as it has caught up, not once every
+/// configured peer has: with ordered startup (a StatefulSet's OrderedReady)
+/// the next pod starts only when this one is Ready, and a learner is not
+/// Ready, so waiting for all of them never finished. A peer is proposed as
+/// a learner only once: re-proposing every pass appended a membership entry
+/// just before the catch-up check, so no learner was ever caught up.
+fn plan_formation(
     metrics: &RaftMetrics<u64, BasicNode>,
-    peer_ids: &BTreeSet<u64>,
+    discovered: &BTreeSet<u64>,
     leader_id: u64,
-) -> bool {
-    let target_index = metrics.last_log_index.unwrap_or(0);
-    let learner_ids: Vec<u64> = peer_ids
+) -> FormationStep {
+    let membership = metrics.membership_config.membership();
+    let members: BTreeSet<u64> = membership.nodes().map(|(id, _)| *id).collect();
+    let voters: BTreeSet<u64> = membership
+        .get_joint_config()
+        .iter()
+        .flat_map(|set| set.iter())
+        .copied()
+        .collect();
+    let add_learners: Vec<u64> = discovered
         .iter()
         .copied()
-        .filter(|id| *id != leader_id)
+        .filter(|id| *id != leader_id && !members.contains(id))
         .collect();
-    if learner_ids.is_empty() {
-        return true;
+    if !add_learners.is_empty() || voters.len() >= MAX_VOTERS {
+        return FormationStep {
+            add_learners,
+            promote_to: None,
+        };
     }
-
-    let replication = match &metrics.replication {
-        Some(r) => r,
-        None => return false,
+    let target_index = metrics.last_log_index.unwrap_or(0);
+    let caught_up = |id: &u64| match metrics.replication.as_ref().and_then(|r| r.get(id)) {
+        Some(Some(matched)) => matched.index >= target_index,
+        Some(None) => target_index == 0,
+        None => false,
     };
-
-    learner_ids
+    let ready: Vec<u64> = members
         .iter()
-        .all(|learner_id| match replication.get(learner_id) {
-            Some(Some(matched)) => matched.index >= target_index,
-            Some(None) => target_index == 0,
-            None => false,
-        })
+        .copied()
+        .filter(|id| !voters.contains(id) && discovered.contains(id) && caught_up(id))
+        .take(MAX_VOTERS - voters.len())
+        .collect();
+    if ready.is_empty() {
+        return FormationStep::default();
+    }
+    let mut promote_to = voters;
+    promote_to.extend(ready);
+    FormationStep {
+        add_learners: Vec::new(),
+        promote_to: Some(promote_to),
+    }
 }
 
 /// Background task that forms the Raft cluster from the static peer list.
@@ -80,52 +118,28 @@ pub async fn run_raft_cluster_formation(
 
     let expected_size = 1 + peer_addrs.len();
 
-    // Discover peers, add as Raft learners, then promote to voters.
-    //
-    // Phase 1: Discover all peers and propose them as Raft learners.
-    //   Uses blocking=false so we don't hang on unreachable learners
-    //   (their HTTP servers may still be starting up).
-    // Phase 2: Wait for all learners to catch up via Raft metrics.
-    // Phase 3: Promote all learners to voters.
-    for attempt in 1u32..=120 {
+    // Discover peers, add each as a Raft learner, and promote each learner
+    // once it has caught up ([`plan_formation`]). Learners are proposed
+    // with blocking=false so an unreachable one (its HTTP server may still
+    // be starting) does not hang the loop.
+    for attempt in 1u32.. {
         if *shutdown_rx.borrow() {
             return;
         }
 
-        let metrics = raft.metrics().borrow().clone();
-
-        // Count current voters — if already fully formed, we're done
-        let voter_ids: BTreeSet<u64> = metrics
-            .membership_config
-            .membership()
-            .get_joint_config()
-            .iter()
-            .flat_map(|set| set.iter())
-            .copied()
-            .collect();
-
-        if voter_ids.len() >= expected_size {
-            tracing::info!(
-                voters = ?voter_ids,
-                "raft cluster fully formed"
-            );
-            break;
-        }
-
-        // ── Phase 1: Discover peers and add as learners ──────────────
         let mut discovered_ids: BTreeSet<u64> = BTreeSet::new();
         discovered_ids.insert(node_id);
-
+        let mut discovered_addrs = std::collections::HashMap::new();
         for peer_addr in &peer_addrs {
             let url = format!("http://{}/cluster/metrics", peer_addr);
             let probe = async {
                 let resp = client.get(&url).send().await.ok()?;
                 let v: serde_json::Value = resp.json().await.ok()?;
                 v["node_id"].as_u64()
             };
-
             if let Some(peer_id) = probe.await {
                 discovered_ids.insert(peer_id);
+                discovered_addrs.insert(peer_id, peer_addr.clone());
 
                 // Update SharedMembership so the data plane can see this peer
                 if let Some(ref m) = state.membership {
@@ -143,78 +157,130 @@ pub async fn run_raft_cluster_formation(
                         });
                     }
                 }
-
-                // Non-blocking: proposes the learner without waiting for
-                // it to catch up. Avoids deadlocking when the learner's
-                // HTTP server isn't up yet.
-                let node = openraft::BasicNode::new(peer_addr.clone());
-                match raft.add_learner(peer_id, node, false).await {
-                    Ok(_) => {
-                        tracing::debug!(
-                            peer_id = peer_id,
-                            peer_addr = %peer_addr,
-                            attempt = attempt,
-                            "raft learner added"
-                        );
-                    }
-                    Err(e) => {
-                        tracing::warn!(
-                            peer_id = peer_id,
-                            error = %e,
-                            attempt = attempt,
-                            "add_learner failed (may already exist or change in progress)"
-                        );
-                    }
-                }
             }
         }
 
-        // ── Phase 2 & 3: Promote when all peers discovered and caught up ──
-        if discovered_ids.len() >= expected_size {
-            let metrics = raft.metrics().borrow().clone();
-            if learners_caught_up(&metrics, &discovered_ids, node_id) {
-                // Phase 2: keep a fixed voter set (bootstrap node + up to four
-                // others) and leave the rest as learners.
-                const MAX_VOTERS: usize = 5;
-                let mut voters: BTreeSet<u64> =
-                    discovered_ids.iter().take(MAX_VOTERS).copied().collect();
-                voters.insert(node_id);
-                match raft.change_membership(voters.clone(), false).await {
-                    Ok(_) => {
-                        tracing::info!(
-                            voters = ?voters,
-                            "raft membership promotion succeeded"
-                        );
-                        break;
-                    }
-                    Err(e) => {
-                        tracing::debug!(
-                            error = %e,
-                            attempt = attempt,
-                            discovered = ?discovered_ids,
-                            "membership promotion not ready yet"
-                        );
-                    }
+        let metrics = raft.metrics().borrow().clone();
+        let step = plan_formation(&metrics, &discovered_ids, node_id);
+        for peer_id in &step.add_learners {
+            let Some(peer_addr) = discovered_addrs.get(peer_id) else {
+                continue;
+            };
+            let node = openraft::BasicNode::new(peer_addr.clone());
+            match raft.add_learner(*peer_id, node, false).await {
+                Ok(_) => tracing::info!(peer_id, peer_addr = %peer_addr, "raft learner added"),
+                Err(e) => tracing::warn!(
+                    peer_id,
+                    error = %e,
+                    attempt,
+                    "add_learner failed (change in progress?)"
+                ),
+            }
+        }
+        if let Some(voters) = step.promote_to {
+            match raft.change_membership(voters.clone(), false).await {
+                Ok(_) => tracing::info!(voters = ?voters, "raft learners promoted to voters"),
+                Err(e) => {
+                    tracing::debug!(error = %e, attempt, "membership promotion not ready yet")
                 }
-            } else {
-                tracing::debug!(
-                    attempt = attempt,
-                    discovered = ?discovered_ids,
-                    last_log_index = ?metrics.last_log_index,
-                    replication = ?metrics.replication,
-                    "waiting for learners to catch up before promotion"
-                );
             }
-        } else {
-            tracing::debug!(
-                attempt = attempt,
-                discovered = discovered_ids.len(),
-                expected = expected_size,
-                "waiting for peers to become reachable"
-            );
+        }
+
+        let metrics = raft.metrics().borrow().clone();
+        let membership = metrics.membership_config.membership();
+        let members = membership.nodes().count();
+        let voters = membership
+            .get_joint_config()
+            .iter()
+            .flat_map(|set| set.iter())
+            .collect::<BTreeSet<_>>()
+            .len();
+        if members >= expected_size && voters >= expected_size.min(MAX_VOTERS) {
+            tracing::info!(voters, members, "raft cluster fully formed");
+            break;
         }
 
         let backoff = Duration::from_secs(if attempt < 10 { 2 } else { 5 });
         tokio::time::sleep(backoff).await;
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use openraft::{CommittedLeaderId, LogId, Membership, StoredMembership};
+    use std::collections::{BTreeMap, BTreeSet};
+
+    fn ids(v: &[u64]) -> BTreeSet<u64> {
+        v.iter().copied().collect()
+    }
+
+    /// Leader 1's metrics: `voters`, `learners`, a log ending at `last`, and
+    /// each follower's matched index.
+    fn metrics(
+        voters: &[u64],
+        learners: &[u64],
+        last: u64,
+        matched: &[(u64, u64)],
+    ) -> RaftMetrics<u64, BasicNode> {
+        let mut nodes = BTreeMap::new();
+        for id in voters.iter().chain(learners) {
+            nodes.insert(*id, BasicNode::new(format!("n{id}")));
+        }
+        let mut m = RaftMetrics::new_initial(1);
+        m.membership_config = std::sync::Arc::new(StoredMembership::new(
+            None,
+            Membership::new(vec![ids(voters)], nodes),
+        ));
+        m.last_log_index = Some(last);
+        let log_id = |index| LogId::new(CommittedLeaderId::new(1, 1), index);
+        m.replication = Some(
+            matched
+                .iter()
+                .map(|(id, index)| (*id, Some(log_id(*index))))
+                .collect(),
+        );
+        m
+    }
+
+    /// Node 2 joined alone (node 3 not started, as OrderedReady does): it is
+    /// promoted once caught up, without waiting for node 3.
+    #[test]
+    fn a_caught_up_learner_is_promoted_without_waiting_for_every_peer() {
+        let m = metrics(&[1], &[2], 7, &[(2, 7)]);
+        let step = plan_formation(&m, &ids(&[1, 2]), 1);
+        assert_eq!(step.add_learners, Vec::<u64>::new());
+        assert_eq!(step.promote_to, Some(ids(&[1, 2])));
+    }
+
+    #[test]
+    fn a_lagging_learner_waits() {
+        let m = metrics(&[1], &[2], 7, &[(2, 6)]);
+        assert_eq!(
+            plan_formation(&m, &ids(&[1, 2]), 1),
+            FormationStep::default()
+        );
+    }
+
+    /// A learner already in the membership is not proposed again: that
+    /// appended an entry each pass, one past what any learner had matched.
+    #[test]
+    fn a_known_learner_is_not_added_again() {
+        let m = metrics(&[1], &[2], 7, &[(2, 7)]);
+        let step = plan_formation(&m, &ids(&[1, 2, 3]), 1);
+        assert_eq!(step.add_learners, vec![3], "only the new peer");
+        assert_eq!(step.promote_to, None, "promotion waits for the add");
+    }
+
+    #[test]
+    fn voters_stop_at_five() {
+        let m = metrics(&[1, 2, 3, 4], &[5, 6], 9, &[(5, 9), (6, 9)]);
+        let step = plan_formation(&m, &ids(&[1, 2, 3, 4, 5, 6]), 1);
+        assert_eq!(step.promote_to, Some(ids(&[1, 2, 3, 4, 5])));
+        let m = metrics(&[1, 2, 3, 4, 5], &[6], 9, &[(6, 9)]);
+        assert_eq!(
+            plan_formation(&m, &ids(&[1, 2, 3, 4, 5, 6]), 1),
+            FormationStep::default()
+        );
+    }
+}