fix: make failover safe under partitions

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 (#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
+  - 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

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
 

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.

crates/ember-cluster/src/gossip.rs

@@ -160,6 +160,9 @@ pub struct GossipEngine {
     local_slots: Vec<SlotRange>,
     /// 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<NodeId>,
     /// Active PingReq relays waiting for an Ack from the target.
     relay_pending: HashMap<u64, RelayEntry>,
 }
@@ -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<NodeId>) {
+        self.local_replicates = replicates;
+    }
+
     /// Returns all known members.
     pub fn members(&self) -> impl Iterator<Item = &MemberState> {
         self.members.values()
@@ -322,6 +337,9 @@ impl GossipEngine {
         is_primary: bool,
         replicates: Option<NodeId>,
     ) {
+        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(),
             }],

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<NodeId>,
     pub config_epoch: u64,
     pub slots: Vec<SlotRange>,
 }
@@ -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<NodeUpdate> {
             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<NodeUpdate> {
     }
 }
 
+fn encode_optional_node_id(buf: &mut BytesMut, id: &Option<NodeId>) {
+    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<Option<NodeId>> {
+    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,13 +709,15 @@ fn decode_member_info(buf: &mut &[u8]) -> io::Result<MemberInfo> {
     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 {
         id,
         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![],
                 },