From 9504d0bbe414976106d56d861d54218fa435fef3 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Wed, 18 Feb 2026 20:56:45 -0500 Subject: [PATCH 1/2] feat(cluster): add VoteRequest/VoteGranted gossip updates for elections MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit adds two new NodeUpdate variants — VoteRequest and VoteGranted — that are piggybacked on Ping/Ack messages to coordinate automatic failover elections. also adds the matching GossipEvent variants that the server layer consumes, and queue_vote_request / queue_vote_granted helpers on GossipEngine. the new election.rs module tracks election state: epoch, votes received, and quorum calculation. a majority (n/2 + 1) of alive primaries must vote for a candidate before it can promote. --- crates/ember-cluster/src/election.rs | 132 +++++++++++++++++++++++++++ crates/ember-cluster/src/gossip.rs | 67 ++++++++++++++ crates/ember-cluster/src/lib.rs | 2 + crates/ember-cluster/src/message.rs | 95 +++++++++++++++++++ 4 files changed, 296 insertions(+) create mode 100644 crates/ember-cluster/src/election.rs diff --git a/crates/ember-cluster/src/election.rs b/crates/ember-cluster/src/election.rs new file mode 100644 index 00000000..3fe6598c --- /dev/null +++ b/crates/ember-cluster/src/election.rs @@ -0,0 +1,132 @@ +//! Automatic failover election state machine. +//! +//! When gossip confirms a primary has failed, its replicas compete to replace +//! 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`. +//! +//! # Fairness +//! +//! Replicas that are more up-to-date (higher replication offset) are +//! preferred by waiting less before broadcasting their request. This is +//! achieved by the caller applying a stagger delay before calling +//! `queue_vote_request`. + +use std::collections::HashSet; + +use crate::NodeId; + +/// State for an in-progress automatic failover election. +/// +/// Created when a replica detects its primary has failed. Discarded when +/// the election succeeds (promotion triggered) or times out. +pub struct Election { + /// Config epoch this election is contesting. + pub epoch: u64, + /// Node IDs of primaries that have voted for us. + votes: HashSet, + /// Whether promotion has already been triggered for this election. + promoted: bool, +} + +impl Election { + /// Creates a new election for the given epoch. + pub fn new(epoch: u64) -> Self { + Self { + epoch, + votes: HashSet::new(), + promoted: false, + } + } + + /// 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`. + /// + /// Once quorum is reached this returns `true` exactly once; subsequent + /// calls return `false` so the caller promotes exactly once. + pub fn record_vote(&mut self, from: NodeId, total_primaries: usize) -> bool { + if self.promoted { + return false; + } + self.votes.insert(from); + if self.votes.len() >= Self::quorum(total_primaries) { + self.promoted = true; + true + } else { + false + } + } + + /// Minimum votes required for a majority. + pub fn quorum(total_primaries: usize) -> usize { + total_primaries / 2 + 1 + } + + /// Returns `true` if this election has already succeeded. + pub fn is_promoted(&self) -> bool { + self.promoted + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn quorum_single_primary() { + assert_eq!(Election::quorum(1), 1); + } + + #[test] + fn quorum_three_primaries() { + assert_eq!(Election::quorum(3), 2); + } + + #[test] + fn quorum_five_primaries() { + assert_eq!(Election::quorum(5), 3); + } + + #[test] + fn record_vote_reaches_quorum() { + let mut e = Election::new(1); + let n1 = NodeId::new(); + let n2 = NodeId::new(); + // 3 primaries → need 2 votes + assert!(!e.record_vote(n1, 3), "first vote should not reach quorum"); + assert!(e.record_vote(n2, 3), "second vote should reach quorum"); + assert!(e.is_promoted()); + } + + #[test] + fn record_vote_deduplicates() { + let mut e = Election::new(1); + let n1 = NodeId::new(); + // 3 primaries need 2 votes; same voter counted only once + assert!(!e.record_vote(n1, 3)); + assert!(!e.record_vote(n1, 3), "duplicate vote must not count"); + assert!(!e.is_promoted()); + } + + #[test] + fn no_votes_after_promotion() { + let mut e = Election::new(1); + let n1 = NodeId::new(); + // single primary — one vote is enough + assert!(e.record_vote(n1, 1)); + // further votes return false + assert!(!e.record_vote(NodeId::new(), 1)); + } + + #[test] + fn single_node_quorum() { + let mut e = Election::new(42); + assert!(e.record_vote(NodeId::new(), 1)); + assert!(e.is_promoted()); + } +} diff --git a/crates/ember-cluster/src/gossip.rs b/crates/ember-cluster/src/gossip.rs index af661ffc..ffcd94fe 100644 --- a/crates/ember-cluster/src/gossip.rs +++ b/crates/ember-cluster/src/gossip.rs @@ -98,6 +98,19 @@ pub enum GossipEvent { SlotsChanged(NodeId, Vec), /// A node's role changed. Fields: node ID, is_primary, replicates. RoleChanged(NodeId, bool, Option), + /// A replica requested votes for a failover election. + VoteRequested { + candidate: NodeId, + epoch: u64, + /// Candidate's replication offset at the time of the request. + offset: u64, + }, + /// A primary granted its vote to a candidate. + VoteGranted { + from: NodeId, + candidate: NodeId, + epoch: u64, + }, } /// The gossip engine manages cluster membership and failure detection. @@ -245,6 +258,30 @@ impl GossipEngine { }); } + /// Queues a vote request for gossip propagation. + /// + /// Called by a replica that is starting an automatic failover election. + /// The update will be piggybacked on the next outgoing Ping or Ack. + pub fn queue_vote_request(&mut self, candidate: NodeId, epoch: u64, offset: u64) { + self.queue_update(NodeUpdate::VoteRequest { + candidate, + epoch, + offset, + }); + } + + /// Queues a vote grant for gossip propagation. + /// + /// Called by a primary that has decided to vote for the given candidate. + /// The update will be piggybacked on the next outgoing Ping or Ack. + pub fn queue_vote_granted(&mut self, from: NodeId, candidate: NodeId, epoch: u64) { + self.queue_update(NodeUpdate::VoteGranted { + from, + candidate, + epoch, + }); + } + /// Adds a seed node to bootstrap cluster discovery. pub fn add_seed(&mut self, id: NodeId, addr: SocketAddr) { if id == self.local_id { @@ -756,6 +793,36 @@ impl GossipEngine { } } } + + NodeUpdate::VoteRequest { + candidate, + epoch, + offset, + } => { + // Relay to the server layer to decide whether to grant. + // No incarnation check needed: epoch ordering is handled upstream. + if *candidate != self.local_id { + self.emit(GossipEvent::VoteRequested { + candidate: *candidate, + epoch: *epoch, + offset: *offset, + }) + .await; + } + } + + NodeUpdate::VoteGranted { + from, + candidate, + epoch, + } => { + self.emit(GossipEvent::VoteGranted { + from: *from, + candidate: *candidate, + epoch: *epoch, + }) + .await; + } } } } diff --git a/crates/ember-cluster/src/lib.rs b/crates/ember-cluster/src/lib.rs index 8dfa58b2..87234508 100644 --- a/crates/ember-cluster/src/lib.rs +++ b/crates/ember-cluster/src/lib.rs @@ -29,6 +29,7 @@ //! assert!(cluster.owns_slot(slot)); //! ``` +mod election; mod error; mod gossip; mod message; @@ -38,6 +39,7 @@ mod raft_transport; mod slots; mod topology; +pub use election::Election; pub use error::ClusterError; pub use gossip::{GossipConfig, GossipEngine, GossipEvent, MemberState, MemberStatus}; pub use message::{GossipMessage, MemberInfo, NodeUpdate}; diff --git a/crates/ember-cluster/src/message.rs b/crates/ember-cluster/src/message.rs index 5d83f2dd..e9bac705 100644 --- a/crates/ember-cluster/src/message.rs +++ b/crates/ember-cluster/src/message.rs @@ -116,6 +116,24 @@ pub enum NodeUpdate { /// The primary this node replicates from, if it is a replica. replicates: Option, }, + /// A replica is requesting votes to take over a failed primary. + VoteRequest { + /// The candidate requesting votes. + candidate: NodeId, + /// Config epoch this election is contesting. + epoch: u64, + /// Candidate's replication offset; higher value signals the most up-to-date replica. + offset: u64, + }, + /// A primary is granting its vote to a candidate replica. + VoteGranted { + /// The primary casting the vote. + from: NodeId, + /// The candidate receiving the vote. + candidate: NodeId, + /// The epoch this vote is for. + epoch: u64, + }, } /// Information about a cluster member. @@ -142,6 +160,8 @@ const UPDATE_DEAD: u8 = 3; const UPDATE_LEFT: u8 = 4; const UPDATE_SLOTS_CHANGED: u8 = 5; const UPDATE_ROLE_CHANGED: u8 = 6; +const UPDATE_VOTE_REQUEST: u8 = 7; +const UPDATE_VOTE_GRANTED: u8 = 8; impl GossipMessage { /// Serializes the message to bytes. @@ -457,6 +477,26 @@ fn encode_update(buf: &mut BytesMut, update: &NodeUpdate) { } } } + NodeUpdate::VoteRequest { + candidate, + epoch, + offset, + } => { + buf.put_u8(UPDATE_VOTE_REQUEST); + encode_node_id(buf, candidate); + buf.put_u64_le(*epoch); + buf.put_u64_le(*offset); + } + NodeUpdate::VoteGranted { + from, + candidate, + epoch, + } => { + buf.put_u8(UPDATE_VOTE_GRANTED); + encode_node_id(buf, from); + encode_node_id(buf, candidate); + buf.put_u64_le(*epoch); + } } } @@ -541,6 +581,26 @@ fn decode_update(buf: &mut &[u8]) -> io::Result { replicates, }) } + UPDATE_VOTE_REQUEST => { + let candidate = decode_node_id(buf)?; + let epoch = safe_get_u64_le(buf)?; + let offset = safe_get_u64_le(buf)?; + Ok(NodeUpdate::VoteRequest { + candidate, + epoch, + offset, + }) + } + UPDATE_VOTE_GRANTED => { + let from = decode_node_id(buf)?; + let candidate = decode_node_id(buf)?; + let epoch = safe_get_u64_le(buf)?; + Ok(NodeUpdate::VoteGranted { + from, + candidate, + epoch, + }) + } other => Err(io::Error::new( io::ErrorKind::InvalidData, format!("unknown update type: {other}"), @@ -852,6 +912,41 @@ mod tests { assert!(result.is_err(), "should reject inverted slot range"); } + #[test] + fn vote_request_roundtrip() { + let candidate = NodeId::new(); + let msg = GossipMessage::Ping { + seq: 1, + sender: candidate, + updates: vec![NodeUpdate::VoteRequest { + candidate, + epoch: 5, + offset: 1234, + }], + }; + let encoded = msg.encode(); + let decoded = GossipMessage::decode(&encoded).unwrap(); + assert_eq!(msg, decoded); + } + + #[test] + fn vote_granted_roundtrip() { + let primary = NodeId::new(); + let candidate = NodeId::new(); + let msg = GossipMessage::Ack { + seq: 2, + sender: primary, + updates: vec![NodeUpdate::VoteGranted { + from: primary, + candidate, + epoch: 5, + }], + }; + let encoded = msg.encode(); + let decoded = GossipMessage::decode(&encoded).unwrap(); + assert_eq!(msg, decoded); + } + #[test] fn out_of_range_slot_in_welcome_rejected() { // craft a Welcome message with a slot >= 16384 From f14cd17b38948ac78a18ea10a2a3ed246766ec57 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Wed, 18 Feb 2026 20:56:52 -0500 Subject: [PATCH 2/2] feat(server): automatic failover via epoch-based voting when gossip confirms a primary as dead and this node is one of its replicas, we start an election: - wait 500ms (stagger delay lets more up-to-date replicas win) - broadcast VoteRequest via gossip piggybacking - primaries that haven't voted in this epoch respond with VoteGranted - first replica to reach majority quorum promotes itself via FAILOVER FORCE VoteRequest events are handled by primaries (one vote per epoch, enforced by last_voted_epoch atomic). VoteGranted events are handled by the candidate; quorum triggers cluster_failover(force=true). post-lock actions (StartElection, HandleVoteRequest, HandleVoteGranted) are dispatched from the gossip event consumer after releasing the state write-lock to avoid deadlocks. --- crates/ember-server/src/cluster.rs | 393 ++++++++++++++++++++++++++++- 1 file changed, 391 insertions(+), 2 deletions(-) diff --git a/crates/ember-server/src/cluster.rs b/crates/ember-server/src/cluster.rs index 7af71ab9..e58d75e7 100644 --- a/crates/ember-server/src/cluster.rs +++ b/crates/ember-server/src/cluster.rs @@ -15,8 +15,9 @@ use std::sync::Arc; use bytes::Bytes; use ember_cluster::{ key_slot, raft_id_from_node_id, BasicNode, ClusterCommand, ClusterNode, ClusterState, - ClusterStateData, ConfigParseError, GossipConfig, GossipEngine, GossipEvent, GossipMessage, - MigrationManager, NodeId, NodeRole, RaftNode, RaftProposalError, SlotRange, SLOT_COUNT, + ClusterStateData, ConfigParseError, Election, GossipConfig, GossipEngine, GossipEvent, + GossipMessage, MigrationManager, NodeId, NodeRole, RaftNode, RaftProposalError, SlotRange, + SLOT_COUNT, }; use ember_core::Engine; use ember_protocol::Frame; @@ -50,6 +51,17 @@ pub struct ClusterCoordinator { /// set by the primary when a replica requests failover; prevents new /// mutations from arriving after the replica has decided to promote. writes_paused: std::sync::atomic::AtomicBool, + /// in-progress automatic failover election (we are the candidate). + election: Mutex>, + /// config epoch of the last vote we granted as a primary; enforces one vote per epoch. + last_voted_epoch: std::sync::atomic::AtomicU64, +} + +/// Tracks an in-progress automatic failover election that this node initiated. +struct ElectionAttempt { + inner: Election, + /// Number of alive primaries when the election started (used for quorum). + total_primaries: usize, } impl std::fmt::Debug for ClusterCoordinator { @@ -114,6 +126,8 @@ impl ClusterCoordinator { raft_node: std::sync::OnceLock::new(), engine: std::sync::OnceLock::new(), writes_paused: std::sync::atomic::AtomicBool::new(false), + election: Mutex::new(None), + last_voted_epoch: std::sync::atomic::AtomicU64::new(0), }; Ok((coordinator, event_rx)) @@ -172,6 +186,8 @@ impl ClusterCoordinator { raft_node: std::sync::OnceLock::new(), engine: std::sync::OnceLock::new(), writes_paused: std::sync::atomic::AtomicBool::new(false), + election: Mutex::new(None), + last_voted_epoch: std::sync::atomic::AtomicU64::new(0), }; Ok((coordinator, event_rx)) @@ -856,6 +872,163 @@ impl ClusterCoordinator { .store(false, std::sync::atomic::Ordering::Release); } + // -- automatic failover -- + + /// Starts an automatic failover election after `failed_primary` is confirmed dead. + /// + /// Applies a brief stagger delay so that the most up-to-date replica wins. + /// Broadcasts a `VoteRequest` via gossip and waits up to 5 seconds for + /// a quorum of primaries to respond with `VoteGranted`. + async fn start_election(self: &Arc, failed_primary: NodeId) { + // Brief delay so a replica that is farther behind waits longer. + // A proper implementation would compute: (max_offset - my_offset) * scale, + // but without a shared offset oracle we use a fixed delay as a tie-breaker. + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + + // Re-read cluster state — the primary may have recovered during our sleep. + let (epoch, total_primaries, still_failed) = { + let state = self.state.read().await; + let epoch = state.config_epoch + 1; + // count alive primaries excluding the failed one + let total = state + .nodes + .values() + .filter(|n| { + n.role == NodeRole::Primary && !n.flags.fail && n.id != failed_primary + }) + .count(); + let still_failed = state + .nodes + .get(&failed_primary) + .map(|n| n.flags.fail) + .unwrap_or(true); + (epoch, total, still_failed) + }; + + if !still_failed { + debug!( + "election: primary {} recovered before election started", + failed_primary + ); + return; + } + + if total_primaries == 0 { + // No other primaries to collect votes from — promote directly. + info!( + "election: no other primaries in cluster; auto-promoting self for epoch {}", + epoch + ); + let _ = self.cluster_failover(true, false).await; + return; + } + + // Initialize the election. + { + let mut guard = self.election.lock().await; + *guard = Some(ElectionAttempt { + inner: Election::new(epoch), + total_primaries, + }); + } + + info!( + "election: starting for epoch {} ({} primary voters needed)", + epoch, + Election::quorum(total_primaries) + ); + + // Broadcast vote request via gossip piggybacking. + { + let mut gossip = self.gossip.lock().await; + gossip.queue_vote_request(self.local_id, epoch, 0 /* offset */); + } + + // Wait for votes; give the cluster time to respond. + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + + // Timed out without quorum — clear election state. + let mut guard = self.election.lock().await; + if let Some(ref e) = *guard { + if e.inner.epoch == epoch && !e.inner.is_promoted() { + warn!("election: timed out for epoch {} without reaching quorum", epoch); + *guard = None; + } + } + } + + /// Handles an incoming `VoteRequest` gossip event. + /// + /// If this node is a primary and hasn't voted in the given epoch, it + /// grants its vote to the candidate and broadcasts `VoteGranted` via gossip. + async fn handle_vote_request(&self, candidate: NodeId, epoch: u64) { + // Only primaries vote. + let is_primary = { + let state = self.state.read().await; + state + .nodes + .get(&self.local_id) + .map(|n| n.role == NodeRole::Primary) + .unwrap_or(false) + }; + if !is_primary { + return; + } + + // Enforce one vote per epoch. + let prev = self + .last_voted_epoch + .fetch_max(epoch, std::sync::atomic::Ordering::AcqRel); + if prev >= epoch { + debug!( + "election: already voted in epoch {} (requested {}); ignoring", + prev, epoch + ); + return; + } + + info!( + "election: granting vote to candidate {} for epoch {}", + candidate, epoch + ); + + let mut gossip = self.gossip.lock().await; + gossip.queue_vote_granted(self.local_id, candidate, epoch); + } + + /// Handles an incoming `VoteGranted` gossip event. + /// + /// If this node is the intended candidate and has an in-progress election, + /// records the vote. Triggers promotion when quorum is reached. + async fn handle_vote_granted( + self: &Arc, + from: NodeId, + candidate: NodeId, + epoch: u64, + ) { + if candidate != self.local_id { + return; // not meant for us + } + + let should_promote = { + let mut guard = self.election.lock().await; + match guard.as_mut() { + Some(attempt) if attempt.inner.epoch == epoch => { + attempt.inner.record_vote(from, attempt.total_primaries) + } + _ => false, + } + }; + + if should_promote { + info!( + "election: quorum reached for epoch {}; promoting self", + epoch + ); + let _ = self.cluster_failover(true, false).await; + } + } + /// CLUSTER FAILOVER [FORCE|TAKEOVER] /// /// Promotes this replica to primary. Must be run on a replica node. @@ -1226,6 +1399,15 @@ impl ClusterCoordinator { let coordinator = Arc::clone(self); tokio::spawn(async move { while let Some(event) = event_rx.recv().await { + // Actions that need to run after releasing the state write-lock. + enum PostAction { + None, + StartElection(NodeId), + HandleVoteRequest { candidate: NodeId, epoch: u64 }, + HandleVoteGranted { from: NodeId, candidate: NodeId, epoch: u64 }, + } + let mut post_action = PostAction::None; + let needs_save = { let mut state = coordinator.state.write().await; match event { @@ -1340,6 +1522,16 @@ impl ClusterCoordinator { node.flags.pfail = false; } state.update_health(); + // if we are a replica of the failed node, start an election + let replicates_failed = state + .nodes + .get(&coordinator.local_id) + .and_then(|n| n.replicates) + .map(|primary_id| primary_id == id) + .unwrap_or(false); + if replicates_failed { + post_action = PostAction::StartElection(id); + } true } GossipEvent::MemberLeft(id) => { @@ -1399,9 +1591,53 @@ impl ClusterCoordinator { state.update_health(); true } + GossipEvent::VoteRequested { + candidate, + epoch, + offset: _, + } => { + // Handle outside the lock so we can call gossip. + post_action = PostAction::HandleVoteRequest { candidate, epoch }; + false + } + GossipEvent::VoteGranted { + from, + candidate, + epoch, + } => { + post_action = PostAction::HandleVoteGranted { + from, + candidate, + epoch, + }; + false + } } }; + // Handle post-lock actions outside the state write-lock. + match post_action { + PostAction::None => {} + PostAction::StartElection(primary_id) => { + let coord = Arc::clone(&coordinator); + tokio::spawn(async move { + coord.start_election(primary_id).await; + }); + } + PostAction::HandleVoteRequest { candidate, epoch } => { + coordinator.handle_vote_request(candidate, epoch).await; + } + PostAction::HandleVoteGranted { + from, + candidate, + epoch, + } => { + coordinator + .handle_vote_granted(from, candidate, epoch) + .await; + } + } + if needs_save { coordinator.save_config().await; } @@ -2050,4 +2286,157 @@ mod tests { coord.resume_writes(); assert!(!coord.is_writes_paused()); } + + // -- automatic failover -- + + #[tokio::test] + async fn primary_grants_vote_once_per_epoch() { + // A primary should grant a vote for a given epoch exactly once. + let (coord, _rx) = test_coordinator_bootstrapped(); + let candidate = NodeId::new(); + + // first request for epoch 5 should be granted (gossip queue entry added) + coord.handle_vote_request(candidate, 5).await; + assert_eq!( + coord + .last_voted_epoch + .load(std::sync::atomic::Ordering::Acquire), + 5, + "last_voted_epoch should be 5 after granting" + ); + + // second request for epoch 5 should be ignored + let prev_epoch = coord + .last_voted_epoch + .load(std::sync::atomic::Ordering::Acquire); + coord.handle_vote_request(candidate, 5).await; + assert_eq!( + coord + .last_voted_epoch + .load(std::sync::atomic::Ordering::Acquire), + prev_epoch, + "epoch should not change on duplicate request" + ); + } + + #[tokio::test] + async fn replica_does_not_grant_vote() { + // Replicas must not vote in elections. + let primary_id = NodeId::new(); + let replica_id = NodeId::new(); + let addr: SocketAddr = "127.0.0.1:6380".parse().unwrap(); + let (coord, _rx) = + ClusterCoordinator::new(replica_id, addr, GossipConfig::default(), false, None) + .unwrap(); + + // set up coord as a replica + { + let mut state = coord.state.write().await; + state.add_node(ClusterNode::new_primary(primary_id, "127.0.0.1:6379".parse().unwrap())); + if let Some(n) = state.nodes.get_mut(&replica_id) { + n.role = NodeRole::Replica; + n.replicates = Some(primary_id); + } + } + + // replica should not update last_voted_epoch + coord.handle_vote_request(NodeId::new(), 3).await; + assert_eq!( + coord + .last_voted_epoch + .load(std::sync::atomic::Ordering::Acquire), + 0, + "replica must not grant votes" + ); + } + + #[tokio::test] + async fn vote_granted_reaches_quorum_and_promotes() { + // A replica that receives enough votes should promote itself. + let primary_id = NodeId::new(); + let voter1 = NodeId::new(); + let voter2 = NodeId::new(); + let replica_id = NodeId::new(); + let addr: SocketAddr = "127.0.0.1:6381".parse().unwrap(); + let (coord, _rx) = + ClusterCoordinator::new(replica_id, addr, GossipConfig::default(), false, None) + .unwrap(); + let coord = Arc::new(coord); + + // set up coord as a replica with two peers owning all slots + { + let mut state = coord.state.write().await; + let mut primary_node = ClusterNode::new_primary(primary_id, "127.0.0.1:6379".parse().unwrap()); + primary_node.slots = vec![SlotRange::new(0, 16383)]; + primary_node.flags.fail = true; // mark as failed + state.add_node(primary_node.clone()); + state.add_node(ClusterNode::new_primary(voter1, "127.0.0.1:6382".parse().unwrap())); + state.add_node(ClusterNode::new_primary(voter2, "127.0.0.1:6383".parse().unwrap())); + // assign slots to primary + for slot in 0..16384u16 { + state.slot_map.assign(slot, primary_id); + } + // set replica state + if let Some(n) = state.nodes.get_mut(&replica_id) { + n.role = NodeRole::Replica; + n.replicates = Some(primary_id); + } + } + + // seed election: epoch=1, 2 alive primaries (voter1, voter2), need 2 votes + { + let mut guard = coord.election.lock().await; + *guard = Some(ElectionAttempt { + inner: Election::new(1), + total_primaries: 2, + }); + } + + // first vote: not yet promoted + coord.handle_vote_granted(voter1, replica_id, 1).await; + assert!(!coord.is_replica().await || { + // either still replica (not yet quorum) or promoted — check election + coord.election.lock().await.as_ref().map(|e| !e.inner.is_promoted()).unwrap_or(true) + }); + + // second vote: quorum reached + coord.handle_vote_granted(voter2, replica_id, 1).await; + + // after quorum the election entry should be promoted and the node primary + // (cluster_failover runs synchronously in tests since there's no Raft) + assert!( + !coord.is_replica().await, + "node should be primary after winning election" + ); + } + + #[tokio::test] + async fn vote_granted_wrong_candidate_ignored() { + let replica_id = NodeId::new(); + let other_candidate = NodeId::new(); + let addr: SocketAddr = "127.0.0.1:6384".parse().unwrap(); + let (coord, _rx) = + ClusterCoordinator::new(replica_id, addr, GossipConfig::default(), false, None) + .unwrap(); + let coord = Arc::new(coord); + + { + let mut guard = coord.election.lock().await; + *guard = Some(ElectionAttempt { + inner: Election::new(1), + total_primaries: 1, + }); + } + + // vote granted to a different candidate — should be ignored + coord + .handle_vote_granted(NodeId::new(), other_candidate, 1) + .await; + + let guard = coord.election.lock().await; + assert!( + !guard.as_ref().unwrap().inner.is_promoted(), + "vote for wrong candidate should not trigger promotion" + ); + } }