From 037569ecc25430669922b3195300cd2a9e01b4ac Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Thu, 12 Nov 2026 18:47:33 -0500 Subject: [PATCH 1/4] fix: send replica roles to joining nodes and never lower the incarnation --- crates/ember-cluster/README.md | 2 +- crates/ember-cluster/src/election.rs | 11 +-- crates/ember-cluster/src/gossip.rs | 130 +++++++++++++++++++++++---- crates/ember-cluster/src/message.rs | 42 +++++---- 4 files changed, 148 insertions(+), 37 deletions(-) diff --git a/crates/ember-cluster/README.md b/crates/ember-cluster/README.md index 7dbd68cb..d42b0119 100644 --- a/crates/ember-cluster/README.md +++ b/crates/ember-cluster/README.md @@ -10,7 +10,7 @@ distributed coordination for [ember](https://github.com/kacy/ember). provides cl - **config epochs** — slot claims spread over gossip, and the higher epoch wins a conflict - **live migration** — slot resharding without downtime, MOVED/ASK redirects - **authentication** — HMAC-SHA256 signing and constant-time verification for cluster transport messages -- **automatic failover** — replica election state machine; a replica needs votes from a majority of primaries, then claims the slots at a new epoch +- **automatic failover** — replica election state machine; a replica needs votes from a majority of all primaries, the failed one included, then claims the slots at a new epoch ## modules diff --git a/crates/ember-cluster/src/election.rs b/crates/ember-cluster/src/election.rs index a6d9aa84..1586b2cd 100644 --- a/crates/ember-cluster/src/election.rs +++ b/crates/ember-cluster/src/election.rs @@ -4,8 +4,9 @@ //! it. Each replica starts a timed election: it broadcasts a `VoteRequest` //! via gossip and waits for primary nodes to respond with `VoteGranted`. //! -//! The first replica to collect votes from a majority of the alive primaries -//! wins the election and promotes itself via `CLUSTER FAILOVER FORCE`. +//! The first replica to collect votes from a majority of all primaries, the +//! failed one included, wins the election and promotes itself via +//! `CLUSTER FAILOVER FORCE`. //! //! # Fairness //! @@ -43,9 +44,9 @@ impl Election { /// Records a vote from `from`. Returns `true` if quorum is newly reached. /// - /// `total_primaries` is the number of alive primaries (excluding the - /// failed one) at the time the election started. Quorum is a simple - /// majority: `total_primaries / 2 + 1`. + /// `total_primaries` is the number of primaries when the election + /// started, the failed one included. Quorum is a simple majority: + /// `total_primaries / 2 + 1`. /// /// Once quorum is reached this returns `true` exactly once; subsequent /// calls return `false` so the caller promotes exactly once. diff --git a/crates/ember-cluster/src/gossip.rs b/crates/ember-cluster/src/gossip.rs index b860ecce..e935cc90 100644 --- a/crates/ember-cluster/src/gossip.rs +++ b/crates/ember-cluster/src/gossip.rs @@ -160,6 +160,9 @@ pub struct GossipEngine { local_slots: Vec, /// The local node's config epoch, sent with its slot claims. local_config_epoch: u64, + /// The primary the local node replicates, if it is a replica. Sent in + /// Welcome replies so a joining node learns this node's role. + local_replicates: Option, /// Active PingReq relays waiting for an Ack from the target. relay_pending: HashMap, } @@ -228,6 +231,7 @@ impl GossipEngine { pending_events: Vec::new(), local_slots: Vec::new(), local_config_epoch: 0, + local_replicates: None, relay_pending: HashMap::new(), } } @@ -237,6 +241,11 @@ impl GossipEngine { self.local_id } + /// How long a node stays unanswered before it is declared dead. + pub fn failure_timeout(&self) -> Duration { + self.config.protocol_period * self.config.suspicion_mult + } + /// Returns the local node's incarnation number. pub fn local_incarnation(&self) -> u64 { self.incarnation @@ -260,6 +269,12 @@ impl GossipEngine { self.incarnation = n; } + /// Sets the local node's role for Welcome replies: the primary it + /// replicates, or `None` for a primary. + pub fn set_local_role(&mut self, replicates: Option) { + self.local_replicates = replicates; + } + /// Returns all known members. pub fn members(&self) -> impl Iterator { self.members.values() @@ -322,6 +337,9 @@ impl GossipEngine { is_primary: bool, replicates: Option, ) { + if node == self.local_id { + self.local_replicates = replicates; + } self.queue_update(NodeUpdate::RoleChanged { node, incarnation, @@ -537,17 +555,20 @@ impl GossipEngine { addr: m.addr, incarnation: m.incarnation, is_primary: m.is_primary, + replicates: m.replicates, config_epoch: m.config_epoch, slots: m.slots.clone(), }) .collect(); - // include our own slot info so the joiner learns the full map + // include ourselves so the joiner learns the full map and + // our role members.push(MemberInfo { id: self.local_id, addr: self.local_addr, incarnation: self.incarnation, - is_primary: true, + is_primary: self.local_replicates.is_none(), + replicates: self.local_replicates, config_epoch: self.local_config_epoch, slots: self.local_slots.clone(), }); @@ -567,17 +588,8 @@ impl GossipEngine { sender, members.len() ); - let sender_is_new = !self.members.contains_key(&sender); - self.ensure_member(sender, from); - if sender_is_new { - let (sender_slots, epoch) = self - .members - .get(&sender) - .map(|m| (m.slots.clone(), m.config_epoch)) - .unwrap_or_default(); - self.emit(GossipEvent::MemberJoined(sender, from, sender_slots, epoch)); - } - + // the sender lists itself among the members, with its role + // and slots for member in members { if member.id == self.local_id { continue; @@ -593,7 +605,7 @@ impl GossipEngine { state: MemberStatus::Alive, state_change: Instant::now(), is_primary: member.is_primary, - replicates: None, + replicates: member.replicates, slots: slots.clone(), config_epoch: member.config_epoch, }); @@ -603,8 +615,22 @@ impl GossipEngine { slots, member.config_epoch, )); + // a joining node starts out as a primary; tell the + // cluster state about a replica, or it counts it + // among the primaries that vote in elections + if !member.is_primary { + self.emit(GossipEvent::RoleChanged( + member.id, + false, + member.replicates, + )); + } } } + if !self.members.contains_key(&sender) { + self.ensure_member(sender, from); + self.emit(GossipEvent::MemberJoined(sender, from, Vec::new(), 0)); + } vec![] } @@ -856,8 +882,9 @@ impl GossipEngine { NodeUpdate::Dead { node, incarnation } => { if *node == self.local_id { - // Refute death claim - self.incarnation = incarnation.saturating_add(1); + // refute the death claim with an incarnation newer + // than both the claim's and our own, never older + self.incarnation = self.incarnation.max(incarnation.saturating_add(1)); self.queue_update(NodeUpdate::Alive { node: self.local_id, addr: self.local_addr, @@ -1553,6 +1580,76 @@ mod tests { } } + #[tokio::test] + async fn welcome_reports_the_senders_replica_role() { + let (tx, _rx) = mpsc::channel(16); + let local = NodeId::new(); + let mut engine = GossipEngine::new(local, test_addr(6379), GossipConfig::default(), tx); + let primary = NodeId::new(); + engine.queue_role_update(local, 1, false, Some(primary)); + + let msg = GossipMessage::Join { + sender: NodeId::new(), + sender_addr: test_addr(6380), + }; + let responses = engine.handle_message(msg, test_addr(6380)); + let GossipMessage::Welcome { members, .. } = &responses[0].1 else { + panic!("expected Welcome"); + }; + let me = members.iter().find(|m| m.id == local).unwrap(); + assert!(!me.is_primary); + assert_eq!(me.replicates, Some(primary)); + } + + #[tokio::test] + async fn a_joiner_learns_replica_roles_from_the_welcome() { + let (tx, mut rx) = mpsc::channel(16); + let mut engine = + GossipEngine::new(NodeId::new(), test_addr(6379), GossipConfig::default(), tx); + let (sender, primary) = (NodeId::new(), NodeId::new()); + let msg = GossipMessage::Welcome { + sender, + members: vec![MemberInfo { + id: sender, + addr: test_addr(6380), + incarnation: 1, + is_primary: false, + replicates: Some(primary), + config_epoch: 0, + slots: vec![], + }], + }; + engine.handle_message(msg, test_addr(6380)); + assert_eq!(engine.members[&sender].replicates, Some(primary)); + + engine.take_events().send().await; + let mut role = None; + while let Ok(event) = rx.try_recv() { + if let GossipEvent::RoleChanged(id, is_primary, replicates) = event { + role = Some((id, is_primary, replicates)); + } + } + assert_eq!(role, Some((sender, false, Some(primary)))); + } + + #[tokio::test] + async fn refuting_a_stale_death_claim_never_lowers_the_incarnation() { + let (tx, _rx) = mpsc::channel(16); + let local = NodeId::new(); + let mut engine = GossipEngine::new(local, test_addr(6379), GossipConfig::default(), tx); + engine.set_incarnation(10); + let msg = GossipMessage::Ping { + seq: 1, + sender: NodeId::new(), + updates: vec![NodeUpdate::Dead { + node: local, + incarnation: 3, + }], + }; + engine.handle_message(msg, test_addr(6380)); + assert!(engine.local_incarnation() >= 10); + } + #[tokio::test] async fn welcome_propagates_member_slots() { let (tx, mut rx) = mpsc::channel(16); @@ -1570,6 +1667,7 @@ mod tests { addr: test_addr(6381), incarnation: 1, is_primary: true, + replicates: None, config_epoch: 1, slots: slots.clone(), }], diff --git a/crates/ember-cluster/src/message.rs b/crates/ember-cluster/src/message.rs index db69071c..0b67660d 100644 --- a/crates/ember-cluster/src/message.rs +++ b/crates/ember-cluster/src/message.rs @@ -160,6 +160,8 @@ pub struct MemberInfo { pub addr: SocketAddr, pub incarnation: u64, pub is_primary: bool, + /// The primary this member replicates, if it is a replica. + pub replicates: Option, pub config_epoch: u64, pub slots: Vec, } @@ -552,15 +554,7 @@ fn encode_update(buf: &mut BytesMut, update: &NodeUpdate) { encode_node_id(buf, node); buf.put_u64_le(*incarnation); buf.put_u8(if *is_primary { 1 } else { 0 }); - match replicates { - Some(primary_id) => { - buf.put_u8(1); - encode_node_id(buf, primary_id); - } - None => { - buf.put_u8(0); - } - } + encode_optional_node_id(buf, replicates); } NodeUpdate::VoteRequest { candidate, @@ -645,12 +639,7 @@ fn decode_update(buf: &mut &[u8]) -> io::Result { let node = decode_node_id(buf)?; let incarnation = safe_get_u64_le(buf)?; let is_primary = safe_get_u8(buf)? != 0; - let has_replicates = safe_get_u8(buf)? != 0; - let replicates = if has_replicates { - Some(decode_node_id(buf)?) - } else { - None - }; + let replicates = decode_optional_node_id(buf)?; Ok(NodeUpdate::RoleChanged { node, incarnation, @@ -687,11 +676,30 @@ fn decode_update(buf: &mut &[u8]) -> io::Result { } } +fn encode_optional_node_id(buf: &mut BytesMut, id: &Option) { + match id { + Some(id) => { + buf.put_u8(1); + encode_node_id(buf, id); + } + None => buf.put_u8(0), + } +} + +fn decode_optional_node_id(buf: &mut &[u8]) -> io::Result> { + if safe_get_u8(buf)? != 0 { + Ok(Some(decode_node_id(buf)?)) + } else { + Ok(None) + } +} + fn encode_member_info(buf: &mut BytesMut, member: &MemberInfo) { encode_node_id(buf, &member.id); encode_socket_addr(buf, &member.addr); buf.put_u64_le(member.incarnation); buf.put_u8(if member.is_primary { 1 } else { 0 }); + encode_optional_node_id(buf, &member.replicates); buf.put_u64_le(member.config_epoch); encode_slot_ranges(buf, &member.slots); } @@ -701,6 +709,7 @@ fn decode_member_info(buf: &mut &[u8]) -> io::Result { let addr = decode_socket_addr(buf)?; let incarnation = safe_get_u64_le(buf)?; let is_primary = safe_get_u8(buf)? != 0; + let replicates = decode_optional_node_id(buf)?; let config_epoch = safe_get_u64_le(buf)?; let slots = decode_slot_ranges(buf)?; Ok(MemberInfo { @@ -708,6 +717,7 @@ fn decode_member_info(buf: &mut &[u8]) -> io::Result { addr, incarnation, is_primary, + replicates, config_epoch, slots, }) @@ -829,6 +839,7 @@ mod tests { addr: test_addr(), incarnation: 1, is_primary: true, + replicates: None, config_epoch: 1, slots: vec![SlotRange::new(0, 5460)], }, @@ -837,6 +848,7 @@ mod tests { addr: test_addr(), incarnation: 2, is_primary: false, + replicates: Some(NodeId::new()), config_epoch: 1, slots: vec![], }, From a1dd03f54568c328079f65939bbb0adf186a3e89 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Thu, 12 Nov 2026 18:47:33 -0500 Subject: [PATCH 2/4] fix: count the failed primary in the quorum and fence a cut-off primary --- crates/ember-server/src/cluster/commands.rs | 23 +++++++ crates/ember-server/src/cluster/failover.rs | 55 ++++++++++++----- crates/ember-server/src/cluster/mod.rs | 11 +++- crates/ember-server/src/cluster/tests.rs | 59 ++++++++++++++++++ .../ember-server/src/connection/dispatch.rs | 7 +++ tests/integration/src/cluster.rs | 60 +++++++++++++++---- 6 files changed, 186 insertions(+), 29 deletions(-) diff --git a/crates/ember-server/src/cluster/commands.rs b/crates/ember-server/src/cluster/commands.rs index ff02a7a3..56ca8987 100644 --- a/crates/ember-server/src/cluster/commands.rs +++ b/crates/ember-server/src/cluster/commands.rs @@ -337,6 +337,29 @@ impl ClusterCoordinator { Frame::Simple("OK".into()) } + /// Whether this node can reach a majority of the cluster's primaries, + /// counting itself. + /// + /// A primary that can't is on the minority side of a partition, where + /// the majority may already be promoting one of its replicas. It stops + /// taking writes then, as Redis Cluster does, so writes aren't + /// acknowledged on both sides and lost when the partition heals. + pub async fn reaches_majority_of_primaries(&self) -> bool { + let state = self.state.read().await; + let primaries = state + .nodes + .values() + .filter(|n| n.role == NodeRole::Primary && !n.flags.handshake); + let (mut total, mut reachable) = (0, 0); + for node in primaries { + total += 1; + if node.id == self.local_id || node.is_healthy() { + reachable += 1; + } + } + reachable * 2 > total + } + /// Checks slot ownership with migration-aware routing. /// /// During a migration, as in Redis Cluster: diff --git a/crates/ember-server/src/cluster/failover.rs b/crates/ember-server/src/cluster/failover.rs index a7938a01..18557184 100644 --- a/crates/ember-server/src/cluster/failover.rs +++ b/crates/ember-server/src/cluster/failover.rs @@ -53,7 +53,8 @@ impl ClusterCoordinator { // this requires a shared offset oracle (e.g., gossip-advertised offset). tokio::time::sleep(std::time::Duration::from_millis(ELECTION_STAGGER_MS)).await; - // Re-read cluster state — the primary may have recovered during our sleep. + // Re-read cluster state — the primary may have recovered during our + // sleep, or another replica may already have taken over its slots. let (epoch, voters, still_failed) = { let state = self.state.read().await; let epoch = state.config_epoch + round; @@ -72,29 +73,33 @@ impl ClusterCoordinator { .nodes .get(&failed_primary) .map(|n| n.flags.fail) - .unwrap_or(true); + .unwrap_or(true) + && !state.slot_map.slots_for_node(failed_primary).is_empty(); (epoch, voters, still_failed) }; if !still_failed { debug!( - "election: primary {} recovered before election started", + "election: primary {} recovered or was replaced before the election", failed_primary ); return false; } - if voters.is_empty() { - // nobody can confirm the primary is really gone, so promoting - // could leave two primaries serving the same slots + // the quorum is a majority of all primaries, the failed one + // included, as in Redis Cluster. a replica cut off with fewer than + // half of them can't win, so the failed primary, which may still be + // serving on the other side of a partition, keeps its slots. + let total_primaries = voters.len() + 1; + let quorum = Election::quorum(total_primaries); + if voters.len() < quorum { warn!( - "election: no other primaries to vote; automatic failover of {} needs at \ - least one more primary in the cluster", - failed_primary + "election: automatic failover of {failed_primary} needs {quorum} votes but \ + only {} other primaries exist; add primaries for automatic failover", + voters.len() ); return false; } - let total_primaries = voters.len(); // ask every voter directly let requests: Vec<_> = { @@ -114,11 +119,7 @@ impl ClusterCoordinator { }); } - info!( - "election: starting for epoch {} ({} primary voters needed)", - epoch, - Election::quorum(total_primaries) - ); + info!("election: starting for epoch {epoch} ({quorum} votes needed)"); self.send_direct(requests).await; @@ -156,6 +157,24 @@ impl ClusterCoordinator { // The candidate names that primary in the request, since news of // who replicates whom may not have reached this node yet; if it has, // the two must agree. + // + // As in Redis Cluster, once a primary votes for a replica to + // replace a failed primary, it votes for no other replica of it for + // two failure timeouts. Otherwise a second replica of the same + // primary could win a later round and both would serve its slots. + // By then the winner has claimed the slots, and the other replicas + // stop their elections. The same replica can ask again, since a + // round can fall short of a quorum. + let cooldown = 2 * self.gossip.lock().await.failure_timeout(); + let mut votes = self.failover_votes.lock().await; + if let Some((chosen, at)) = votes.get(&failed_primary) { + if *chosen != candidate && at.elapsed() < cooldown { + debug!( + "election: voted for {chosen} to replace {failed_primary}; not voting for {candidate}" + ); + return; + } + } { let mut state = self.state.write().await; let is_primary = state @@ -191,6 +210,8 @@ impl ClusterCoordinator { } state.last_vote_epoch = epoch; } + votes.insert(failed_primary, (candidate, std::time::Instant::now())); + drop(votes); // the vote goes to disk before it is granted, so a restart can't // lead to a second vote in the same epoch self.save_config().await; @@ -222,7 +243,9 @@ impl ClusterCoordinator { let mut guard = self.election.lock().await; match guard.as_mut() { Some(attempt) if attempt.inner.epoch == epoch && attempt.voters.contains(&from) => { - attempt.inner.record_vote(from, attempt.voters.len()) + // the failed primary counts toward the total, though it + // can't vote + attempt.inner.record_vote(from, attempt.voters.len() + 1) } _ => false, } diff --git a/crates/ember-server/src/cluster/mod.rs b/crates/ember-server/src/cluster/mod.rs index 87b9eacc..f1bd9388 100644 --- a/crates/ember-server/src/cluster/mod.rs +++ b/crates/ember-server/src/cluster/mod.rs @@ -9,7 +9,7 @@ //! nodes claim a slot, the higher epoch wins. Failover elections run over //! gossip as well, and a promoted replica claims its slots at a new epoch. -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::net::SocketAddr; use std::path::PathBuf; use std::sync::Arc; @@ -77,6 +77,10 @@ pub struct ClusterCoordinator { writes_paused: std::sync::atomic::AtomicBool, /// in-progress automatic failover election (we are the candidate). election: Mutex>, + /// the replica this node last voted for to replace each failed + /// primary, and when. it votes for one replica per failed primary at a + /// time, so two replicas of the same primary can't both win. + failover_votes: Mutex>, /// optional shared secret for authenticating cluster transport messages. secret: Option>, /// the task pulling data from our primary, while this node is a replica @@ -158,6 +162,7 @@ impl ClusterCoordinator { engine: std::sync::OnceLock::new(), writes_paused: std::sync::atomic::AtomicBool::new(false), election: Mutex::new(None), + failover_votes: Mutex::new(HashMap::new()), save_lock: Mutex::new(()), replication_task: Mutex::new(None), secret, @@ -204,9 +209,10 @@ impl ClusterCoordinator { } } - // set local slots in gossip engine + // set local slots and role in gossip engine let local_slots = state.slot_map.slots_for_node(local_id); gossip.set_local_slots(local_slots, state.local_epoch()); + gossip.set_local_role(state.nodes.get(&local_id).and_then(|n| n.replicates)); let coordinator = Self { state: RwLock::new(state), @@ -220,6 +226,7 @@ impl ClusterCoordinator { engine: std::sync::OnceLock::new(), writes_paused: std::sync::atomic::AtomicBool::new(false), election: Mutex::new(None), + failover_votes: Mutex::new(HashMap::new()), save_lock: Mutex::new(()), replication_task: Mutex::new(None), secret, diff --git a/crates/ember-server/src/cluster/tests.rs b/crates/ember-server/src/cluster/tests.rs index e37e80ce..b59a84b9 100644 --- a/crates/ember-server/src/cluster/tests.rs +++ b/crates/ember-server/src/cluster/tests.rs @@ -585,6 +585,65 @@ async fn add_primary(coord: &ClusterCoordinator, failed: bool) -> NodeId { id } +#[tokio::test] +async fn a_primary_takes_writes_only_while_it_reaches_a_majority() { + let (coord, _rx) = test_coordinator_bootstrapped(); + assert!(coord.reaches_majority_of_primaries().await, "alone"); + + let a = add_primary(&coord, false).await; + let b = add_primary(&coord, false).await; + assert!(coord.reaches_majority_of_primaries().await); + + // cut off from both others: 1 of 3 is no majority + for id in [a, b] { + coord + .state + .write() + .await + .nodes + .get_mut(&id) + .unwrap() + .flags + .pfail = true; + } + assert!(!coord.reaches_majority_of_primaries().await); + + // one back: 2 of 3 + coord + .state + .write() + .await + .nodes + .get_mut(&a) + .unwrap() + .flags + .pfail = false; + assert!(coord.reaches_majority_of_primaries().await); +} + +#[tokio::test] +async fn a_primary_votes_for_one_replica_of_a_failed_primary() { + // two replicas of the same failed primary ask in different epochs. the + // second must not win too, or both would serve the slots + let (coord, _rx) = test_coordinator_bootstrapped(); + let failed = add_primary(&coord, true).await; + let (first, second) = (NodeId::new(), NodeId::new()); + + coord.handle_vote_request(first, failed, 1).await; + assert_eq!(coord.state.read().await.last_vote_epoch, 1); + coord.handle_vote_request(second, failed, 2).await; + assert_eq!(coord.state.read().await.last_vote_epoch, 1); + + // the first replica can ask again after a round without a quorum + coord.handle_vote_request(first, failed, 3).await; + assert_eq!(coord.state.read().await.last_vote_epoch, 3); + + // another failed primary is a separate failover + let other = add_primary(&coord, true).await; + coord.handle_vote_request(NodeId::new(), other, 4).await; + assert_eq!(coord.state.read().await.last_vote_epoch, 4); +} + #[tokio::test] async fn primary_grants_vote_once_per_epoch() { // A primary should grant a vote for a given epoch exactly once. diff --git a/crates/ember-server/src/connection/dispatch.rs b/crates/ember-server/src/connection/dispatch.rs index 0df28259..29aea2bb 100644 --- a/crates/ember-server/src/connection/dispatch.rs +++ b/crates/ember-server/src/connection/dispatch.rs @@ -244,6 +244,8 @@ pub(super) async fn prepare_command( /// error to send: /// - READONLY for a write while failover has paused writes, or MOVED to the /// primary (READONLY if it is unknown) for a write on a replica +/// - CLUSTERDOWN for a write on a primary that can't reach a majority of +/// the primaries /// - CROSSSLOT when the command's keys hash to different slots /// - MOVED, ASK or CLUSTERDOWN when this node does not serve the key's slot /// - while the slot migrates out: ASK when none of the keys are still here, @@ -279,6 +281,11 @@ pub(super) async fn cluster_slot_check( "READONLY You can't write against a read only replica.".into(), )); } + if !cluster.reaches_majority_of_primaries().await { + return Some(Frame::Error( + "CLUSTERDOWN this primary can't reach a majority of the primaries".into(), + )); + } } // replicas serve every read locally diff --git a/tests/integration/src/cluster.rs b/tests/integration/src/cluster.rs index 681b5259..c18de63f 100644 --- a/tests/integration/src/cluster.rs +++ b/tests/integration/src/cluster.rs @@ -747,12 +747,44 @@ async fn migrate_copy_keeps_the_local_key() { assert_eq!(target.get_bulk(&["GET", &key]).await, Some("v".into())); } +/// A primary cut off from most of the primaries stops taking writes, so +/// the majority can replace it without writes landing on both sides. +#[tokio::test] +async fn a_primary_cut_off_from_the_majority_rejects_writes() { + use std::time::{Duration, Instant}; + + let mut cluster = TestCluster::start(); + cluster.init().await; + let (tag, _) = tag_on_node_0(&cluster).await; + let key = format!("{tag}k"); + let mut c = cluster.connect(0).await; + c.ok(&["SET", &key, "before"]).await; + + cluster.nodes[1].kill(); + cluster.nodes[2].kill(); + + let deadline = Instant::now() + Duration::from_secs(30); + loop { + match c.cmd(&["SET", &key, "after"]).await { + Frame::Error(e) if e.starts_with("CLUSTERDOWN") => break, + reply => assert!( + Instant::now() <= deadline, + "writes still accepted after 30s: {reply:?}" + ), + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + // reads still work + assert!(c.get_bulk(&["GET", &key]).await.is_some()); +} + // -- automatic failover -- -/// End-to-end automatic failover: a 3-node cluster where node 0 owns all -/// slots, node 2 replicates it, and node 1 is a slotless primary whose vote -/// forms the election quorum. Killing node 0 must promote node 2 and restore -/// write availability. +/// End-to-end automatic failover: a 4-node cluster where one primary owns +/// all slots, a replica follows it, and two slotless primaries vote. With +/// the failed primary there are 3 primaries, so the quorum is 2 votes. +/// Killing the primary must promote the replica and restore write +/// availability. /// /// Generously timed: failure confirmation + election + promotion involve /// gossip rounds, so every step polls under a deadline instead of sleeping @@ -772,6 +804,7 @@ async fn cluster_automatic_failover_promotes_replica() { ..opts() }); let voter = TestServer::start_with(opts()); + let voter2 = TestServer::start_with(opts()); let replica = TestServer::start_with(opts()); let mut c0 = primary.connect().await; @@ -788,11 +821,16 @@ async fn cluster_automatic_failover_promotes_replica() { // replica needs the voter in its primary count for the election quorum, // and the voter must know the candidate to grant it a vote. let mut cv = voter.connect().await; - c0.ok(&["CLUSTER", "MEET", "127.0.0.1", &voter.port.to_string()]) - .await; - c0.ok(&["CLUSTER", "MEET", "127.0.0.1", &replica.port.to_string()]) - .await; - cv.ok(&["CLUSTER", "MEET", "127.0.0.1", &replica.port.to_string()]) + let mut cv2 = voter2.connect().await; + for port in [voter.port, voter2.port, replica.port] { + c0.ok(&["CLUSTER", "MEET", "127.0.0.1", &port.to_string()]) + .await; + } + for port in [voter2.port, replica.port] { + cv.ok(&["CLUSTER", "MEET", "127.0.0.1", &port.to_string()]) + .await; + } + cv2.ok(&["CLUSTER", "MEET", "127.0.0.1", &replica.port.to_string()]) .await; // wait until gossip has delivered the primary's identity to the @@ -810,7 +848,7 @@ async fn cluster_automatic_failover_promotes_replica() { } // wait until gossip has fully converged from the replica's point of - // view: it must know all 3 nodes and the primary's ownership of all + // view: it must know all 4 nodes and the primary's ownership of all // 16384 slots. killing the primary earlier leaves the replica without // a slot table to inherit and without the voter as a failure-detection // relay, so no election would ever start. @@ -819,7 +857,7 @@ async fn cluster_automatic_failover_promotes_replica() { let resp = c2.cmd(&["CLUSTER", "INFO"]).await; if let Frame::Bulk(ref data) = resp { let text = String::from_utf8_lossy(data); - if text.contains("cluster_known_nodes:3") + if text.contains("cluster_known_nodes:4") && text.contains("cluster_slots_assigned:16384") { break; From 815afe204eef53d499003d8b32e36a04e8a0808a Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Thu, 12 Nov 2026 18:47:33 -0500 Subject: [PATCH 3/4] docs: describe failover quorum and write fencing --- CHANGELOG.md | 6 ++++++ docs/production-checklist.md | 4 +++- docs/troubleshooting.md | 2 +- 3 files changed, 10 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6ff1c796..0b3493e8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,12 @@ - raft and the `openraft` dependency. raft only ever ran on the bootstrap node, and gossip with config epochs already decided slot ownership and failover, as in Redis Cluster. `raft-port-offset` and `--cluster-raft-port-offset` are still accepted and ignored, so existing configs keep working (#443) ### fixes +- failover safety: + - an election needs a majority of all primaries, the failed one included, as in Redis Cluster. it counted only the other primaries, so with two primaries a replica cut off with one of them won with a single vote while its primary kept serving + - a primary votes for one replica per failed primary for two failure timeouts, so two replicas of the same primary can't both win. a replica stops its election once another node owns the failed primary's slots + - a primary that can't reach a majority of the primaries rejects writes with CLUSTERDOWN, so a cut-off primary stops accepting writes the majority is about to replace + - a node that joins learns which nodes are replicas from the Welcome reply. it saw them as primaries before, which put them in the election quorum, and the node sending the reply always described itself as a primary + - refuting a Dead claim no longer lowers the node's incarnation - slot migration (#451): - MIGRATE sends ASKING before RESTORE, so the target accepts the key for a slot it is importing. every MIGRATE during a reshard failed with MOVED before - MIGRATE takes `AUTH password` and `AUTH2 username password` for a target with a password, honors `COPY`, and replies `NOKEY` for a missing key, as in Redis diff --git a/docs/production-checklist.md b/docs/production-checklist.md index d62c5391..c46c22a0 100644 --- a/docs/production-checklist.md +++ b/docs/production-checklist.md @@ -319,7 +319,9 @@ CLUSTER FAILOVER run this on a replica to promote it to primary. the outgoing primary will demote itself cleanly if it is reachable. -automatic failover triggers when a primary is unreachable. replicas hold an epoch-based election; the replica with the most up-to-date replication offset wins. the cluster should recover within a few seconds of failure detection. +automatic failover triggers when a primary is unreachable. its replicas hold an epoch-based election, and the first to collect votes from a majority of all primaries, the failed one included, takes over. a primary votes for one replica per failed primary, so two replicas can't both take over. the cluster should recover within a few seconds of failure detection. + +automatic failover needs at least three primaries: with two, the one left can't form a majority of two on its own. a primary that can't reach a majority of the primaries stops taking writes, so the same rule keeps a cut-off primary from accepting writes the majority is about to replace. **resharding** diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index bc8955f5..b114bf48 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -211,7 +211,7 @@ when bringing up a new cluster: **split-brain** -if a network partition separates nodes, each side keeps its own view of the cluster. a replica is promoted only when a majority of primaries vote for it, so the minority side cannot elect new primaries. a primary cut off on the minority side still accepts writes for its slots until the partition heals. when it rejoins and a higher-epoch claim takes its slots, the writes it accepted during the partition can be lost. +if a network partition separates nodes, each side keeps its own view of the cluster. a replica is promoted only when a majority of all primaries, its failed primary included, vote for it, so the minority side cannot elect new primaries. a primary cut off on the minority side rejects writes with `CLUSTERDOWN this primary can't reach a majority of the primaries` once it has marked the other primaries as unreachable, and serves reads only. writes it accepted before then, during the failure detection timeout, can still be lost when the majority promotes one of its replicas. to recover: 1. restore network connectivity between partitions From 4b3dd37811e4b7bda269096e0363399047fd77fa Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Thu, 12 Nov 2026 18:47:50 -0500 Subject: [PATCH 4/4] docs: link the changelog entry to its PR --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0b3493e8..ca78ce01 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,7 +11,7 @@ - raft and the `openraft` dependency. raft only ever ran on the bootstrap node, and gossip with config epochs already decided slot ownership and failover, as in Redis Cluster. `raft-port-offset` and `--cluster-raft-port-offset` are still accepted and ignored, so existing configs keep working (#443) ### fixes -- failover safety: +- failover safety (#452): - an election needs a majority of all primaries, the failed one included, as in Redis Cluster. it counted only the other primaries, so with two primaries a replica cut off with one of them won with a single vote while its primary kept serving - a primary votes for one replica per failed primary for two failure timeouts, so two replicas of the same primary can't both win. a replica stops its election once another node owns the failed primary's slots - a primary that can't reach a majority of the primaries rejects writes with CLUSTERDOWN, so a cut-off primary stops accepting writes the majority is about to replace