From 944507da9aafa9c2d7fa282b09d1565a14776ba7 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Wed, 18 Feb 2026 20:35:56 -0500 Subject: [PATCH] feat(cluster): implement CLUSTER FAILOVER state machine adds CLUSTER FAILOVER [FORCE|TAKEOVER] to promote a replica to primary. topology.rs: new ClusterState::promote_replica(replica_id) method - transfers all slots from the old primary to the promoted node - demotes the old primary to replica status (replicates the new primary) - updates replica lists on both nodes - bumps the global config epoch for conflict resolution cluster.rs: ClusterCoordinator::cluster_failover(force, takeover) - default: waits 500ms for replication to catch up, then proposes PromoteReplica + slot transfer via Raft for cluster-wide agreement - force: same as default but skips the grace period - takeover: bypasses Raft entirely; updates local state and announces the new role via gossip (for when Raft quorum is unavailable) - adds writes_paused AtomicBool + pause_writes/resume_writes for future primary-side coordination during default failover connection.rs: - replaces the FAILOVER stub with a call to cluster_failover - adds is_writes_paused check to the write rejection gate --- crates/ember-cluster/src/topology.rs | 126 +++++++++++++ crates/ember-server/src/cluster.rs | 246 ++++++++++++++++++++++++++ crates/ember-server/src/connection.rs | 15 +- 3 files changed, 383 insertions(+), 4 deletions(-) diff --git a/crates/ember-cluster/src/topology.rs b/crates/ember-cluster/src/topology.rs index 978d70af..6e540667 100644 --- a/crates/ember-cluster/src/topology.rs +++ b/crates/ember-cluster/src/topology.rs @@ -446,6 +446,66 @@ impl ClusterState { self.state = ClusterHealth::Ok; } + /// Promotes a replica to primary, transferring slots from its current primary. + /// + /// Performs the full state transition for a failover: + /// - Transfers all slots from the old primary to the promoted replica. + /// - Demotes the old primary to a replica of the new primary. + /// - Updates replica lists on both nodes. + /// - Bumps the global config epoch. + /// + /// Returns an error if the target is not a replica with a configured primary. + pub fn promote_replica(&mut self, replica_id: NodeId) -> Result<(), String> { + // locate the replica and find its current primary + let primary_id = { + let replica = self + .nodes + .get(&replica_id) + .ok_or_else(|| format!("node {replica_id} not found in cluster state"))?; + if replica.role != NodeRole::Replica { + return Err(format!("node {replica_id} is not a replica")); + } + replica + .replicates + .ok_or_else(|| format!("replica {replica_id} has no primary configured"))? + }; + + // transfer every slot currently owned by the old primary + for slot in 0..SLOT_COUNT { + if self.slot_map.owner(slot) == Some(primary_id) { + self.slot_map.assign(slot, replica_id); + } + } + let new_primary_slots = self.slot_map.slots_for_node(replica_id); + + // bump epoch before touching node state + self.config_epoch += 1; + let new_epoch = self.config_epoch; + + // demote old primary → it now replicates the new primary + if let Some(old_primary) = self.nodes.get_mut(&primary_id) { + old_primary.role = NodeRole::Replica; + old_primary.replicates = Some(replica_id); + old_primary.replicas.retain(|&id| id != replica_id); + old_primary.slots.clear(); + old_primary.config_epoch = new_epoch; + } + + // promote the replica → it becomes the new primary + if let Some(new_primary) = self.nodes.get_mut(&replica_id) { + new_primary.role = NodeRole::Primary; + new_primary.replicates = None; + if !new_primary.replicas.contains(&primary_id) { + new_primary.replicas.push(primary_id); + } + new_primary.slots = new_primary_slots; + new_primary.config_epoch = new_epoch; + } + + self.update_health(); + Ok(()) + } + /// Generates the response for CLUSTER INFO command. pub fn cluster_info(&self) -> String { let assigned_slots = (SLOT_COUNT as usize - self.slot_map.unassigned_count()) as u16; @@ -965,4 +1025,70 @@ mod tests { assert_eq!(state.replicas().count(), 1); assert_eq!(state.replicas_of(primary_id).count(), 1); } + + #[test] + fn promote_replica_transfers_slots() { + let primary_id = NodeId::new(); + let replica_id = NodeId::new(); + + let mut primary = ClusterNode::new_primary(primary_id, test_addr(6379)); + primary.set_myself(); + + let mut state = ClusterState::single_node(primary); + let replica = ClusterNode::new_replica(replica_id, test_addr(6380), primary_id); + state.add_node(replica); + + // register the replica in the primary's replica list + state.nodes.get_mut(&primary_id).unwrap().replicas.push(replica_id); + + let initial_epoch = state.config_epoch; + state.promote_replica(replica_id).unwrap(); + + // epoch should have been bumped + assert_eq!(state.config_epoch, initial_epoch + 1); + + // the promoted node is now a primary with all slots + let new_primary = state.nodes.get(&replica_id).unwrap(); + assert_eq!(new_primary.role, NodeRole::Primary); + assert_eq!(new_primary.replicates, None); + assert!(new_primary.replicas.contains(&primary_id)); + assert!(!new_primary.slots.is_empty()); + + // the old primary is now a replica + let old_primary = state.nodes.get(&primary_id).unwrap(); + assert_eq!(old_primary.role, NodeRole::Replica); + assert_eq!(old_primary.replicates, Some(replica_id)); + assert!(old_primary.slots.is_empty()); + + // slot ownership must be transferred + for slot in 0..SLOT_COUNT { + assert_eq!(state.slot_map.owner(slot), Some(replica_id)); + } + + // cluster should still be healthy + assert_eq!(state.state, ClusterHealth::Ok); + } + + #[test] + fn promote_replica_rejects_non_replica() { + let id = NodeId::new(); + let mut node = ClusterNode::new_primary(id, test_addr(6379)); + node.set_myself(); + let mut state = ClusterState::single_node(node); + + let err = state.promote_replica(id).unwrap_err(); + assert!(err.contains("not a replica"), "unexpected error: {err}"); + } + + #[test] + fn promote_replica_rejects_unknown_node() { + let id = NodeId::new(); + let mut node = ClusterNode::new_primary(id, test_addr(6379)); + node.set_myself(); + let mut state = ClusterState::single_node(node); + + let missing = NodeId::new(); + let err = state.promote_replica(missing).unwrap_err(); + assert!(err.contains("not found"), "unexpected error: {err}"); + } } diff --git a/crates/ember-server/src/cluster.rs b/crates/ember-server/src/cluster.rs index c2907d10..7af71ab9 100644 --- a/crates/ember-server/src/cluster.rs +++ b/crates/ember-server/src/cluster.rs @@ -46,6 +46,10 @@ pub struct ClusterCoordinator { raft_node: std::sync::OnceLock>, /// engine handle for replication; set once during startup engine: std::sync::OnceLock>, + /// temporarily pauses writes on this node during failover coordination. + /// 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, } impl std::fmt::Debug for ClusterCoordinator { @@ -109,6 +113,7 @@ impl ClusterCoordinator { data_dir, raft_node: std::sync::OnceLock::new(), engine: std::sync::OnceLock::new(), + writes_paused: std::sync::atomic::AtomicBool::new(false), }; Ok((coordinator, event_rx)) @@ -166,6 +171,7 @@ impl ClusterCoordinator { data_dir: Some(data_dir), raft_node: std::sync::OnceLock::new(), engine: std::sync::OnceLock::new(), + writes_paused: std::sync::atomic::AtomicBool::new(false), }; Ok((coordinator, event_rx)) @@ -824,6 +830,157 @@ impl ClusterCoordinator { } } + /// Returns `true` if writes are temporarily paused on this node. + /// + /// Set by the primary during failover coordination to prevent new mutations + /// from arriving after the replica has committed to promoting. + pub fn is_writes_paused(&self) -> bool { + self.writes_paused + .load(std::sync::atomic::Ordering::Acquire) + } + + /// Pauses write commands on this node. + /// + /// Called by the primary when a replica requests a coordinated failover, + /// ensuring no new writes arrive after the replica decides to promote. + #[allow(dead_code)] + pub fn pause_writes(&self) { + self.writes_paused + .store(true, std::sync::atomic::Ordering::Release); + } + + /// Resumes write commands after a failover pause. + #[allow(dead_code)] + pub fn resume_writes(&self) { + self.writes_paused + .store(false, std::sync::atomic::Ordering::Release); + } + + /// CLUSTER FAILOVER [FORCE|TAKEOVER] + /// + /// Promotes this replica to primary. Must be run on a replica node. + /// + /// Three modes: + /// - **Default**: waits 500ms for replication to catch up, then promotes + /// via Raft so all nodes agree on the new topology. + /// - **FORCE**: skips the grace period; promotes via Raft immediately. + /// Use when the primary is unreachable and you accept possible data loss. + /// - **TAKEOVER**: bypasses Raft entirely. Updates local state and + /// announces the new role via gossip. Use when Raft quorum is lost. + pub async fn cluster_failover(&self, force: bool, takeover: bool) -> Frame { + // verify we are a replica with a configured primary + let primary_id = { + let state = self.state.read().await; + let local = match state.nodes.get(&self.local_id) { + Some(n) => n, + None => { + return Frame::Error( + "ERR local node not found in cluster state".into(), + ) + } + }; + if local.role != NodeRole::Replica { + return Frame::Error( + "ERR You should send CLUSTER FAILOVER to a replica".into(), + ); + } + match local.replicates { + Some(id) => id, + None => { + return Frame::Error( + "ERR No primary configured for this replica".into(), + ) + } + } + }; + + if takeover { + // TAKEOVER: immediate local promotion, no Raft or primary coordination. + // The replica asserts itself as primary and gossips the change; + // the rest of the cluster learns via gossip convergence. + { + let mut state = self.state.write().await; + if let Err(e) = state.promote_replica(self.local_id) { + return Frame::Error(format!("ERR {e}")); + } + } + self.announce_promotion().await; + self.save_config().await; + info!(local_id = %self.local_id, %primary_id, "TAKEOVER: promoted to primary"); + return Frame::Simple("OK".into()); + } + + // Default / FORCE: use Raft for cluster-wide agreement. + if !force { + // give the replication stream a brief window to deliver + // any in-flight records before we cut over. a future improvement + // could track the exact offset and wait for full catchup. + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + + // get the primary's current slot ranges so we can hand them over + let primary_slots = { + let state = self.state.read().await; + state + .nodes + .get(&primary_id) + .map(|n| n.slots.clone()) + .unwrap_or_default() + }; + + if let Some(raft) = self.raft_node.get() { + // promote this replica in the Raft state machine + let promote_cmd = ClusterCommand::PromoteReplica { + replica_id: self.local_id, + }; + if let Err(e) = raft.propose(promote_cmd).await { + return Self::raft_error_frame(e); + } + + // transfer the primary's slots to this node via Raft + if !primary_slots.is_empty() { + let assign_cmd = ClusterCommand::AssignSlots { + node_id: self.local_id, + slots: primary_slots.clone(), + }; + if let Err(e) = raft.propose(assign_cmd).await { + return Self::raft_error_frame(e); + } + + // remove those slots from the old primary + let remove_cmd = ClusterCommand::RemoveSlots { + node_id: primary_id, + slots: primary_slots, + }; + // best-effort: old primary may already be unreachable + let _ = raft.propose(remove_cmd).await; + } + } + + // apply locally right away so this node can start accepting writes + // without waiting for the async Raft reconciliation to complete + { + let mut state = self.state.write().await; + if let Err(e) = state.promote_replica(self.local_id) { + warn!(%e, "local promote_replica after Raft proposal failed"); + } + } + + self.announce_promotion().await; + self.save_config().await; + + let mode = if force { "FORCE" } else { "default" }; + info!(local_id = %self.local_id, %primary_id, mode, "promoted to primary"); + Frame::Simple("OK".into()) + } + + /// Queues a role-change gossip announcement marking this node as primary. + async fn announce_promotion(&self) { + let mut gossip = self.gossip.lock().await; + let inc = gossip.local_incarnation(); + gossip.queue_role_update(self.local_id, inc, true, None); + } + /// Attaches the engine so replication can start on demand. /// /// Must be called once after the engine is built, before any `CLUSTER REPLICATE` @@ -1804,4 +1961,93 @@ mod tests { let (coord, _rx) = test_coordinator(); assert!(!coord.is_replica().await, "new coordinator should be a primary"); } + + #[tokio::test] + async fn failover_rejected_on_primary() { + let (coord, _rx) = test_coordinator_bootstrapped(); + let result = coord.cluster_failover(false, false).await; + match result { + Frame::Error(msg) => assert!( + msg.contains("replica"), + "expected replica error, got: {msg}" + ), + other => panic!("expected error frame, got {other:?}"), + } + } + + #[tokio::test] + async fn failover_takeover_promotes_replica() { + // set up: primary owns all slots, replica replicates from it + let (coord, _rx) = test_coordinator_bootstrapped(); + let primary_id = coord.local_id; + + let replica_id = NodeId::new(); + let replica_addr: SocketAddr = "127.0.0.1:6380".parse().unwrap(); + + // add the replica to the primary's state + { + let mut state = coord.state.write().await; + let mut replica = ClusterNode::new_replica(replica_id, replica_addr, primary_id); + replica.set_myself(); // pretend we're running on the replica + state.add_node(replica); + // register replica in primary's list + state + .nodes + .get_mut(&primary_id) + .unwrap() + .replicas + .push(replica_id); + // update local_id to the replica + // (simulate running on the replica node) + } + + // build a replica coordinator with the same state + let replica_addr_sa: SocketAddr = "127.0.0.1:6380".parse().unwrap(); + let (replica_coord, _rx2) = + ClusterCoordinator::new(replica_id, replica_addr_sa, GossipConfig::default(), false, None) + .unwrap(); + + // manually set up the replica's state + { + let mut state = replica_coord.state.write().await; + let primary_node = coord.state.read().await.nodes.get(&primary_id).unwrap().clone(); + state.add_node(primary_node); + // set this node as a replica of primary + if let Some(local) = state.nodes.get_mut(&replica_id) { + local.role = NodeRole::Replica; + local.replicates = Some(primary_id); + } + // assign all slots to primary in the slot map + for slot in 0..16384u16 { + state.slot_map.assign(slot, primary_id); + } + } + + // TAKEOVER: should succeed since no Raft is needed + let result = replica_coord.cluster_failover(false, true).await; + assert!(matches!(result, Frame::Simple(_)), "expected OK, got {result:?}"); + + // verify the replica is now a primary + assert!( + !replica_coord.is_replica().await, + "after TAKEOVER, node should be primary" + ); + + // verify it owns all slots + let state = replica_coord.state.read().await; + for slot in 0..16384u16 { + assert_eq!(state.slot_map.owner(slot), Some(replica_id)); + } + } + + #[tokio::test] + async fn writes_paused_blocks_and_resumes() { + let (coord, _rx) = test_coordinator_bootstrapped(); + + assert!(!coord.is_writes_paused()); + coord.pause_writes(); + assert!(coord.is_writes_paused()); + coord.resume_writes(); + assert!(!coord.is_writes_paused()); + } } diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs index c75311c2..f624ebf4 100644 --- a/crates/ember-server/src/connection.rs +++ b/crates/ember-server/src/connection.rs @@ -1762,8 +1762,14 @@ async fn execute( pubsub: &Arc, asking: bool, ) -> Frame { - // Replica write rejection: redirect mutations to the primary. + // Write gating: reject mutations when the node is a replica or when + // writes are temporarily paused (e.g. during failover coordination). if let Some(ref cluster) = ctx.cluster { + if cluster.is_writes_paused() && cmd.is_write() { + return Frame::Error( + "READONLY Failover in progress; writes are temporarily paused.".into(), + ); + } if cluster.is_replica().await && cmd.is_write() { if let Some(key) = cmd.primary_key() { use ember_cluster::key_slot; @@ -2711,9 +2717,10 @@ async fn execute( None => Frame::Error("ERR This instance has cluster support disabled".into()), }, - Command::ClusterFailover { .. } => { - Frame::Error("ERR FAILOVER not yet implemented".into()) - } + Command::ClusterFailover { force, takeover } => match &ctx.cluster { + Some(c) => c.cluster_failover(force, takeover).await, + None => Frame::Error("ERR This instance has cluster support disabled".into()), + }, Command::Migrate { host,