From 853e6eec0018ed3500da984b14e03553abcf92be Mon Sep 17 00:00:00 2001 From: Austin Barrington Date: Thu, 22 Oct 2026 14:46:21 +0100 Subject: [PATCH] Promote cluster.peers learners as they catch up; no 503 after /health is 200 A cluster formed through the deprecated `cluster.peers` left every node but the first a Raft learner, Syncing, its /health on 503 for good: - The bootstrap node promoted learners only once every configured peer was reachable and caught up. With ordered startup (a StatefulSet's OrderedReady, or a script) the next node starts only when this one is Ready, and a learner never is. Each learner is now promoted as soon as it has caught up, up to five voters. - Every pass re-proposed each known peer as a learner, appending a membership entry just before checking that learners had matched the log tail, so none ever had. A peer is now proposed once. A node's first writes after its /health turned 200 could get an empty 503: - /health reports a node Raft serves as a voter as Active while its own record still says Syncing; /write refused on the record. It now follows /health. - A node that had synced but holds no data (a new, empty cluster) was treated as new by the leader's sync trigger and re-joined, which flipped it to Syncing. It now reconnects, which stays Active for a small gap. 3-node ordered start: node 2 healthy after 7 s, node 3 after 11 s, all voters (before: learners, 503, indefinitely). Write straight after /health 200 on every node: 0 503s in 9 of 10 fresh clusters (before: 4 of 6 had one). The remaining case is a node writing in the second or two before the leader adds it as a learner. --- .../src/adapters/http/peer_handlers.rs | 13 +- hyperbytedb/src/adapters/http/ping.rs | 2 +- hyperbytedb/src/adapters/http/write.rs | 9 +- .../src/application/cluster/raft_formation.rs | 298 +++++++++++------- 4 files changed, 203 insertions(+), 119 deletions(-) diff --git a/hyperbytedb/src/adapters/http/peer_handlers.rs b/hyperbytedb/src/adapters/http/peer_handlers.rs index 69c3c91..d95bf1e 100644 --- a/hyperbytedb/src/adapters/http/peer_handlers.rs +++ b/hyperbytedb/src/adapters/http/peer_handlers.rs @@ -945,7 +945,18 @@ pub async fn handle_sync_trigger(State(state): State>) -> 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; diff --git a/hyperbytedb/src/adapters/http/ping.rs b/hyperbytedb/src/adapters/http/ping.rs index 3f9de74..43650ba 100644 --- a/hyperbytedb/src/adapters/http/ping.rs +++ b/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; }; diff --git a/hyperbytedb/src/adapters/http/write.rs b/hyperbytedb/src/adapters/http/write.rs index 3bade88..3ce4fbe 100644 --- a/hyperbytedb/src/adapters/http/write.rs +++ b/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. diff --git a/hyperbytedb/src/application/cluster/raft_formation.rs b/hyperbytedb/src/application/cluster/raft_formation.rs index 0d19b7c..aaaceb3 100644 --- a/hyperbytedb/src/application/cluster/raft_formation.rs +++ b/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, + /// The voter set to change to: current voters plus every learner that + /// has caught up, up to [`MAX_VOTERS`]. + promote_to: Option>, +} + +/// 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, - peer_ids: &BTreeSet, + discovered: &BTreeSet, leader_id: u64, -) -> bool { - let target_index = metrics.last_log_index.unwrap_or(0); - let learner_ids: Vec = peer_ids +) -> FormationStep { + let membership = metrics.membership_config.membership(); + let members: BTreeSet = membership.nodes().map(|(id, _)| *id).collect(); + let voters: BTreeSet = membership + .get_joint_config() + .iter() + .flat_map(|set| set.iter()) + .copied() + .collect(); + let add_learners: Vec = 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 = 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,42 +118,18 @@ 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 = 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 = 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 { @@ -123,9 +137,9 @@ pub async fn run_raft_cluster_formation( 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 = - 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::>() + .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 { + 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 { + 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::::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() + ); + } +}