feat(cluster): automatic failover via epoch-based voting

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<NodeId>,
+    /// 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());
+    }
+}

crates/ember-cluster/src/gossip.rs

@@ -98,6 +98,19 @@ pub enum GossipEvent {
     SlotsChanged(NodeId, Vec<SlotRange>),
     /// A node's role changed. Fields: node ID, is_primary, replicates.
     RoleChanged(NodeId, bool, Option<NodeId>),
+    /// 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;
+                }
             }
         }
     }

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};

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<NodeId>,
     },
+    /// 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<NodeUpdate> {
                 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