From f69954e3af1e4b9683e47cd8923d512e5e273bd2 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Wed, 18 Feb 2026 19:29:50 -0500 Subject: [PATCH] feat(cluster): implement CLUSTER REPLICATE with role propagation via gossip MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - add `NodeUpdate::RoleChanged` gossip message variant (tag 6) with binary encode/decode; carries node id, incarnation, is_primary, and optional replicates field - add `GossipEvent::RoleChanged(NodeId, bool, Option)` emitted by apply_updates() when a peer's role changes - add `GossipEngine::queue_role_update()` to piggyback role changes on outgoing pings and acks - add `replicates: Option` to `MemberState` to track which primary each replica mirrors - add `ClusterCoordinator::cluster_replicate()` — validates target, updates local state, queues gossip announcement, saves nodes.conf - add `ClusterCoordinator::is_replica()` and `primary_addr_for_slot()` for replica-aware routing - handle `GossipEvent::RoleChanged` in spawn_gossip to update ClusterState - add `Command::is_write()` to identify mutation commands - add `Command::primary_key()` to extract the first key for slot routing - replace REPLICATE and FAILOVER stubs in connection.rs; REPLICATE now dispatches to cluster_replicate(), FAILOVER returns a clear message - add replica write rejection in execute(): writes on replicas are redirected to the primary via MOVED; reads are served locally - update cluster_slot_check() to skip slot routing on replicas - update integration tests: replicate_stub → targeted error-path tests; failover_stub updated to check new message text --- crates/ember-cluster/src/gossip.rs | 147 ++++++++++++++++++++++ crates/ember-cluster/src/message.rs | 83 ++++++++++++ crates/ember-protocol/src/command.rs | 173 +++++++++++++++++++++++++ crates/ember-server/src/cluster.rs | 175 +++++++++++++++++++++++++- crates/ember-server/src/connection.rs | 31 ++++- tests/integration/src/cluster.rs | 37 +++++- 6 files changed, 638 insertions(+), 8 deletions(-) diff --git a/crates/ember-cluster/src/gossip.rs b/crates/ember-cluster/src/gossip.rs index 165fe445..af661ffc 100644 --- a/crates/ember-cluster/src/gossip.rs +++ b/crates/ember-cluster/src/gossip.rs @@ -67,6 +67,8 @@ pub struct MemberState { pub state: MemberStatus, pub state_change: Instant, pub is_primary: bool, + /// The primary this member replicates from, if it is a replica. + pub replicates: Option, pub slots: Vec, } @@ -94,6 +96,8 @@ pub enum GossipEvent { MemberAlive(NodeId), /// A node's slot ownership changed. SlotsChanged(NodeId, Vec), + /// A node's role changed. Fields: node ID, is_primary, replicates. + RoleChanged(NodeId, bool, Option), } /// The gossip engine manages cluster membership and failure detection. @@ -222,6 +226,25 @@ impl GossipEngine { }); } + /// Queues a role change for gossip propagation. + /// + /// Called after this node changes from primary to replica (or vice versa). + /// The update will be piggybacked on the next outgoing Ping or Ack. + pub fn queue_role_update( + &mut self, + node: NodeId, + incarnation: u64, + is_primary: bool, + replicates: Option, + ) { + self.queue_update(NodeUpdate::RoleChanged { + node, + incarnation, + is_primary, + replicates, + }); + } + /// Adds a seed node to bootstrap cluster discovery. pub fn add_seed(&mut self, id: NodeId, addr: SocketAddr) { if id == self.local_id { @@ -234,6 +257,7 @@ impl GossipEngine { state: MemberStatus::Alive, state_change: Instant::now(), is_primary: false, + replicates: None, slots: Vec::new(), }); } @@ -428,6 +452,7 @@ impl GossipEngine { state: MemberStatus::Alive, state_change: Instant::now(), is_primary: member.is_primary, + replicates: None, slots: slots.clone(), }); self.emit(GossipEvent::MemberJoined(member.id, member.addr, slots)) @@ -544,6 +569,7 @@ impl GossipEngine { state: MemberStatus::Alive, state_change: Instant::now(), is_primary: false, + replicates: None, slots: Vec::new(), }); } @@ -598,6 +624,7 @@ impl GossipEngine { state: MemberStatus::Alive, state_change: Instant::now(), is_primary: false, + replicates: None, slots: Vec::new(), }, ); @@ -701,6 +728,34 @@ impl GossipEngine { } } } + + NodeUpdate::RoleChanged { + node, + incarnation, + is_primary, + replicates, + } => { + if *incarnation > MAX_INCARNATION { + warn!( + "rejecting role update for {} with excessive incarnation {}", + node, incarnation + ); + continue; + } + if *node == self.local_id { + // we know our own role + continue; + } + if let Some(member) = self.members.get_mut(node) { + if *incarnation > member.incarnation { + member.incarnation = *incarnation; + member.is_primary = *is_primary; + member.replicates = *replicates; + self.emit(GossipEvent::RoleChanged(*node, *is_primary, *replicates)) + .await; + } + } + } } } } @@ -1054,6 +1109,7 @@ mod tests { state: MemberStatus::Alive, state_change: Instant::now(), is_primary: true, + replicates: None, slots: vec![SlotRange::new(0, 5460)], }, ); @@ -1314,4 +1370,95 @@ mod tests { // stale entry should be cleaned up assert!(engine.relay_pending.is_empty()); } + + #[tokio::test] + async fn apply_role_changed_updates_member() { + let (tx, mut rx) = mpsc::channel(16); + let mut engine = + GossipEngine::new(NodeId::new(), test_addr(6379), GossipConfig::default(), tx); + + let remote = NodeId::new(); + let primary = NodeId::new(); + engine.add_seed(remote, test_addr(6380)); + + // send a role change: remote becomes a replica of primary + let msg = GossipMessage::Ping { + seq: 1, + sender: remote, + updates: vec![NodeUpdate::RoleChanged { + node: remote, + incarnation: 2, + is_primary: false, + replicates: Some(primary), + }], + }; + engine.handle_message(msg, test_addr(6380)).await; + + let member = engine.members.get(&remote).unwrap(); + assert!(!member.is_primary); + assert_eq!(member.replicates, Some(primary)); + assert_eq!(member.incarnation, 2); + + // should have emitted RoleChanged + let mut found = false; + while let Ok(event) = rx.try_recv() { + if let GossipEvent::RoleChanged(id, is_primary, replicates) = event { + if id == remote { + assert!(!is_primary); + assert_eq!(replicates, Some(primary)); + found = true; + break; + } + } + } + assert!(found, "expected RoleChanged event for remote"); + } + + #[tokio::test] + async fn stale_role_changed_ignored() { + let (tx, mut rx) = mpsc::channel(16); + let mut engine = + GossipEngine::new(NodeId::new(), test_addr(6379), GossipConfig::default(), tx); + + let remote = NodeId::new(); + engine.members.insert( + remote, + MemberState { + id: remote, + addr: test_addr(6380), + incarnation: 10, + state: MemberStatus::Alive, + state_change: Instant::now(), + is_primary: true, + replicates: None, + slots: vec![], + }, + ); + + // send role change with old incarnation + let msg = GossipMessage::Ping { + seq: 1, + sender: remote, + updates: vec![NodeUpdate::RoleChanged { + node: remote, + incarnation: 5, // stale + is_primary: false, + replicates: None, + }], + }; + engine.handle_message(msg, test_addr(6380)).await; + + // member should still be primary + let member = engine.members.get(&remote).unwrap(); + assert!(member.is_primary, "stale update should not change role"); + + // drain events (MemberAlive from the Ping sender, but no RoleChanged) + let mut role_changed = false; + while let Ok(event) = rx.try_recv() { + if matches!(event, GossipEvent::RoleChanged(..)) { + role_changed = true; + } + } + assert!(!role_changed, "stale role update should not emit RoleChanged"); + } } diff --git a/crates/ember-cluster/src/message.rs b/crates/ember-cluster/src/message.rs index 23526f15..5d83f2dd 100644 --- a/crates/ember-cluster/src/message.rs +++ b/crates/ember-cluster/src/message.rs @@ -107,6 +107,15 @@ pub enum NodeUpdate { incarnation: u64, slots: Vec, }, + /// Node's role changed (primary ↔ replica). + RoleChanged { + node: NodeId, + incarnation: u64, + /// `true` if the node is now a primary, `false` if replica. + is_primary: bool, + /// The primary this node replicates from, if it is a replica. + replicates: Option, + }, } /// Information about a cluster member. @@ -132,6 +141,7 @@ const UPDATE_SUSPECT: u8 = 2; const UPDATE_DEAD: u8 = 3; const UPDATE_LEFT: u8 = 4; const UPDATE_SLOTS_CHANGED: u8 = 5; +const UPDATE_ROLE_CHANGED: u8 = 6; impl GossipMessage { /// Serializes the message to bytes. @@ -427,6 +437,26 @@ fn encode_update(buf: &mut BytesMut, update: &NodeUpdate) { buf.put_u16_le(slot.end); } } + NodeUpdate::RoleChanged { + node, + incarnation, + is_primary, + replicates, + } => { + buf.put_u8(UPDATE_ROLE_CHANGED); + 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); + } + } + } } } @@ -494,6 +524,23 @@ fn decode_update(buf: &mut &[u8]) -> io::Result { slots, }) } + UPDATE_ROLE_CHANGED => { + 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 + }; + Ok(NodeUpdate::RoleChanged { + node, + incarnation, + is_primary, + replicates, + }) + } other => Err(io::Error::new( io::ErrorKind::InvalidData, format!("unknown update type: {other}"), @@ -666,6 +713,42 @@ mod tests { assert_eq!(msg, decoded); } + #[test] + fn role_changed_roundtrip() { + let node = NodeId::new(); + let primary = NodeId::new(); + + // replica variant + let msg = GossipMessage::Ping { + seq: 1, + sender: node, + updates: vec![NodeUpdate::RoleChanged { + node, + incarnation: 7, + is_primary: false, + replicates: Some(primary), + }], + }; + let encoded = msg.encode(); + let decoded = GossipMessage::decode(&encoded).unwrap(); + assert_eq!(msg, decoded); + + // primary variant (no replicates field) + let msg2 = GossipMessage::Ping { + seq: 2, + sender: node, + updates: vec![NodeUpdate::RoleChanged { + node, + incarnation: 8, + is_primary: true, + replicates: None, + }], + }; + let encoded2 = msg2.encode(); + let decoded2 = GossipMessage::decode(&encoded2).unwrap(); + assert_eq!(msg2, decoded2); + } + #[test] fn all_update_types() { let node = NodeId::new(); diff --git a/crates/ember-protocol/src/command.rs b/crates/ember-protocol/src/command.rs index 8cd6000d..05e62b17 100644 --- a/crates/ember-protocol/src/command.rs +++ b/crates/ember-protocol/src/command.rs @@ -617,6 +617,126 @@ impl Command { } } + /// Returns `true` if this command mutates state. + /// + /// Used by the replica write-rejection layer: any command matching this + /// predicate is redirected to the primary via MOVED rather than executed + /// locally on a read-only replica. + pub fn is_write(&self) -> bool { + matches!( + self, + Command::Set { .. } + | Command::Del { .. } + | Command::Unlink { .. } + | Command::Incr { .. } + | Command::Decr { .. } + | Command::IncrBy { .. } + | Command::DecrBy { .. } + | Command::IncrByFloat { .. } + | Command::Append { .. } + | Command::Rename { .. } + | Command::Expire { .. } + | Command::Persist { .. } + | Command::Pexpire { .. } + | Command::MSet { .. } + | Command::LPush { .. } + | Command::LPop { .. } + | Command::RPush { .. } + | Command::RPop { .. } + | Command::ZAdd { .. } + | Command::ZRem { .. } + | Command::HSet { .. } + | Command::HDel { .. } + | Command::HIncrBy { .. } + | Command::SAdd { .. } + | Command::SRem { .. } + | Command::FlushDb { .. } + | Command::BgRewriteAof + | Command::BgSave + | Command::Restore { .. } + | Command::VAdd { .. } + | Command::VAddBatch { .. } + | Command::VRem { .. } + | Command::ProtoRegister { .. } + | Command::ProtoSet { .. } + | Command::ProtoSetField { .. } + | Command::ProtoDelField { .. } + ) + } + + /// Returns the primary key for this command, if there is one. + /// + /// Used to calculate the hash slot for MOVED redirects on replicas. + /// For multi-key commands, returns the first key. + pub fn primary_key(&self) -> Option<&str> { + match self { + Command::Get { key } + | Command::Set { key, .. } + | Command::Incr { key } + | Command::Decr { key } + | Command::IncrBy { key, .. } + | Command::DecrBy { key, .. } + | Command::IncrByFloat { key, .. } + | Command::Append { key, .. } + | Command::Strlen { key } + | Command::Persist { key } + | Command::Expire { key, .. } + | Command::Pexpire { key, .. } + | Command::Ttl { key } + | Command::Pttl { key } + | Command::Type { key } + | Command::Rename { key, .. } + | Command::LPush { key, .. } + | Command::RPush { key, .. } + | Command::LPop { key } + | Command::RPop { key } + | Command::LRange { key, .. } + | Command::LLen { key } + | Command::ZAdd { key, .. } + | Command::ZRem { key, .. } + | Command::ZScore { key, .. } + | Command::ZRank { key, .. } + | Command::ZRange { key, .. } + | Command::ZCard { key } + | Command::HSet { key, .. } + | Command::HGet { key, .. } + | Command::HGetAll { key } + | Command::HDel { key, .. } + | Command::HExists { key, .. } + | Command::HLen { key } + | Command::HIncrBy { key, .. } + | Command::HKeys { key } + | Command::HVals { key } + | Command::HMGet { key, .. } + | Command::SAdd { key, .. } + | Command::SRem { key, .. } + | Command::SMembers { key } + | Command::SIsMember { key, .. } + | Command::SCard { key } + | Command::VAdd { key, .. } + | Command::VAddBatch { key, .. } + | Command::VSim { key, .. } + | Command::VRem { key, .. } + | Command::VGet { key, .. } + | Command::VCard { key } + | Command::VDim { key } + | Command::VInfo { key } + | Command::ProtoSet { key, .. } + | Command::ProtoGet { key } + | Command::ProtoType { key } + | Command::ProtoGetField { key, .. } + | Command::ProtoSetField { key, .. } + | Command::ProtoDelField { key, .. } + | Command::Restore { key, .. } => Some(key), + Command::Del { keys } + | Command::Unlink { keys } + | Command::Exists { keys } + | Command::MGet { keys } => keys.first().map(String::as_str), + Command::MSet { pairs } => pairs.first().map(|(k, _)| k.as_str()), + _ => None, + } + } + /// Parses a [`Frame`] into a [`Command`]. /// /// Expects an array frame where the first element is the command name @@ -5496,4 +5616,57 @@ mod tests { let err = Command::from_frame(cmd(&["RESTORE", "key"])).unwrap_err(); assert!(matches!(err, ProtocolError::WrongArity(_))); } + + // -- is_write / primary_key -- + + #[test] + fn is_write_returns_true_for_mutations() { + assert!(Command::Set { + key: "k".into(), + value: bytes::Bytes::new(), + expire: None, + nx: false, + xx: false, + } + .is_write()); + assert!(Command::Del { keys: vec!["k".into()] }.is_write()); + assert!(Command::Incr { key: "k".into() }.is_write()); + assert!(Command::HSet { + key: "k".into(), + fields: vec![], + } + .is_write()); + assert!(Command::LPush { key: "k".into(), values: vec![] }.is_write()); + assert!(Command::ZAdd { + key: "k".into(), + flags: crate::command::ZAddFlags::default(), + members: vec![], + } + .is_write()); + assert!(Command::SAdd { key: "k".into(), members: vec![] }.is_write()); + assert!(Command::FlushDb { async_mode: false }.is_write()); + } + + #[test] + fn is_write_returns_false_for_reads() { + assert!(!Command::Get { key: "k".into() }.is_write()); + assert!(!Command::HGet { key: "k".into(), field: "f".into() }.is_write()); + assert!(!Command::Ping(None).is_write()); + assert!(!Command::ClusterInfo.is_write()); + assert!(!Command::DbSize.is_write()); + } + + #[test] + fn primary_key_returns_first_key() { + assert_eq!( + Command::Get { key: "hello".into() }.primary_key(), + Some("hello") + ); + assert_eq!( + Command::Del { keys: vec!["a".into(), "b".into()] }.primary_key(), + Some("a") + ); + assert_eq!(Command::Ping(None).primary_key(), None); + assert_eq!(Command::DbSize.primary_key(), None); + } } diff --git a/crates/ember-server/src/cluster.rs b/crates/ember-server/src/cluster.rs index 15eda5c6..f798fe1d 100644 --- a/crates/ember-server/src/cluster.rs +++ b/crates/ember-server/src/cluster.rs @@ -16,7 +16,7 @@ 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, RaftNode, RaftProposalError, SlotRange, SLOT_COUNT, + MigrationManager, NodeId, NodeRole, RaftNode, RaftProposalError, SlotRange, SLOT_COUNT, }; use ember_protocol::Frame; use tokio::net::UdpSocket; @@ -729,6 +729,88 @@ impl ClusterCoordinator { Ok(()) } + // -- replication -- + + /// CLUSTER REPLICATE node-id + /// + /// Makes this node a replica of the given primary. Updates local topology, + /// queues a role-change gossip announcement, and persists nodes.conf. + pub async fn cluster_replicate(&self, primary_id_str: &str) -> Frame { + let primary_id = match NodeId::parse(primary_id_str) { + Ok(id) => id, + Err(_) => return Frame::Error("ERR Invalid node ID".into()), + }; + + if primary_id == self.local_id { + return Frame::Error("ERR Cannot replicate self".into()); + } + + // verify the target exists and is a primary + { + let state = self.state.read().await; + match state.nodes.get(&primary_id) { + None => return Frame::Error("ERR Unknown node ID".into()), + Some(node) if node.role != NodeRole::Primary => { + return Frame::Error("ERR Target node is not a primary".into()) + } + _ => {} + } + } + + // update local cluster state + { + let mut state = self.state.write().await; + if let Some(node) = state.nodes.get_mut(&self.local_id) { + node.role = NodeRole::Replica; + node.replicates = Some(primary_id); + } + // register ourselves in the primary's replica list + if let Some(primary) = state.nodes.get_mut(&primary_id) { + if !primary.replicas.contains(&self.local_id) { + primary.replicas.push(self.local_id); + } + } + state.update_health(); + } + + // queue role change for epidemic dissemination + let incarnation = { + let mut gossip = self.gossip.lock().await; + let inc = gossip.local_incarnation(); + gossip.queue_role_update(self.local_id, inc, false, Some(primary_id)); + inc + }; + let _ = incarnation; // used only to hold the lock briefly + + self.save_config().await; + Frame::Simple("OK".into()) + } + + /// Returns `true` if this node is currently configured as a replica. + pub async fn is_replica(&self) -> bool { + let state = self.state.read().await; + state + .nodes + .get(&self.local_id) + .map(|n| n.role == NodeRole::Replica) + .unwrap_or(false) + } + + /// Returns the address of the primary that owns the given slot. + /// + /// Used when redirecting write commands on replicas via MOVED. + pub async fn primary_addr_for_slot(&self, slot: u16) -> Option { + let state = self.state.read().await; + let owner_id = state.slot_map.owner(slot)?; + let node = state.nodes.get(&owner_id)?; + // Only redirect to primaries; if the owner is somehow a replica, skip. + if node.role == NodeRole::Primary { + Some(node.addr) + } else { + None + } + } + /// Pushes the local node's current slot ownership into the gossip engine /// so it propagates to the rest of the cluster. async fn broadcast_local_slots(&self, slots: Vec) { @@ -1025,6 +1107,23 @@ impl ClusterCoordinator { state.update_health(); true } + GossipEvent::RoleChanged(id, is_primary, replicates) => { + debug!( + "cluster: node {} role changed to {}", + id, + if is_primary { "primary" } else { "replica" } + ); + if let Some(node) = state.nodes.get_mut(&id) { + node.role = if is_primary { + NodeRole::Primary + } else { + NodeRole::Replica + }; + node.replicates = replicates; + } + state.update_health(); + true + } } }; @@ -1503,4 +1602,78 @@ mod tests { other => panic!("expected ASK redirect, got {other:?}"), } } + + // -- cluster_replicate -- + + #[tokio::test] + async fn cluster_replicate_self_rejected() { + let (coord, _rx) = test_coordinator(); + let resp = coord.cluster_replicate(&coord.local_id.0.to_string()).await; + match resp { + Frame::Error(msg) => assert!(msg.contains("Cannot replicate self")), + other => panic!("expected error, got {other:?}"), + } + } + + #[tokio::test] + async fn cluster_replicate_invalid_id_rejected() { + let (coord, _rx) = test_coordinator(); + let resp = coord.cluster_replicate("not-a-uuid").await; + match resp { + Frame::Error(msg) => assert!(msg.contains("Invalid node ID")), + other => panic!("expected error, got {other:?}"), + } + } + + #[tokio::test] + async fn cluster_replicate_unknown_node_rejected() { + let (coord, _rx) = test_coordinator(); + let unknown = NodeId::new(); + let resp = coord.cluster_replicate(&unknown.0.to_string()).await; + match resp { + Frame::Error(msg) => assert!(msg.contains("Unknown node ID")), + other => panic!("expected error, got {other:?}"), + } + } + + #[tokio::test] + async fn cluster_replicate_updates_state() { + let (coord, _rx) = test_coordinator(); + let primary_id = NodeId::new(); + + // add a primary node to replicate from + { + let mut state = coord.state.write().await; + let primary = ClusterNode::new_primary(primary_id, "127.0.0.1:6380".parse().unwrap()); + state.add_node(primary); + } + + let resp = coord.cluster_replicate(&primary_id.0.to_string()).await; + assert!(matches!(resp, Frame::Simple(_)), "expected OK, got {resp:?}"); + + // local node should now be a replica + assert!(coord.is_replica().await, "node should be a replica after REPLICATE"); + + // local node's replicates field should point to primary + let state = coord.state.read().await; + let local = state.nodes.get(&coord.local_id).unwrap(); + assert_eq!(local.replicates, Some(primary_id)); + assert!(state.nodes[&primary_id].replicas.contains(&coord.local_id)); + } + + #[tokio::test] + async fn primary_addr_for_slot_returns_primary_addr() { + let (coord, _rx) = test_coordinator_bootstrapped(); + + // the bootstrap coordinator owns all slots — find any one + let addr = coord.primary_addr_for_slot(0).await; + assert!(addr.is_some(), "should find primary for slot 0"); + assert_eq!(addr.unwrap().port(), 6379); + } + + #[tokio::test] + async fn is_replica_returns_false_initially() { + let (coord, _rx) = test_coordinator(); + assert!(!coord.is_replica().await, "new coordinator should be a primary"); + } } diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs index a546e124..31104379 100644 --- a/crates/ember-server/src/connection.rs +++ b/crates/ember-server/src/connection.rs @@ -1630,6 +1630,12 @@ fn resolve_shard_response(resp: ShardResponse, tag: ResponseTag) -> Frame { async fn cluster_slot_check(ctx: &ServerContext, cmd: &Command, asking: bool) -> Option { let cluster = ctx.cluster.as_ref()?; + // Replicas serve all reads locally — skip slot routing. + // Write rejection is handled separately in execute() before this call. + if cluster.is_replica().await { + return None; + } + match cmd { // single-key commands — check slot ownership for the key Command::Get { ref key } @@ -1756,6 +1762,22 @@ async fn execute( pubsub: &Arc, asking: bool, ) -> Frame { + // Replica write rejection: redirect mutations to the primary. + if let Some(ref cluster) = ctx.cluster { + if cluster.is_replica().await && cmd.is_write() { + if let Some(key) = cmd.primary_key() { + use ember_cluster::key_slot; + let slot = key_slot(key.as_bytes()); + if let Some(addr) = cluster.primary_addr_for_slot(slot).await { + return Frame::Error(format!("MOVED {slot} {addr}")); + } + } + return Frame::Error( + "READONLY You can't write against a read only replica.".into(), + ); + } + } + // cluster slot validation — check whether we own the slot for this key. // when `asking` is true, importing slots are allowed through. if let Some(redirect) = cluster_slot_check(ctx, &cmd, asking).await { @@ -2684,9 +2706,14 @@ async fn execute( } } - Command::ClusterReplicate { .. } => Frame::Error("ERR REPLICATE not yet supported".into()), + Command::ClusterReplicate { node_id } => match &ctx.cluster { + Some(c) => c.cluster_replicate(&node_id).await, + None => Frame::Error("ERR This instance has cluster support disabled".into()), + }, - Command::ClusterFailover { .. } => Frame::Error("ERR FAILOVER not yet supported".into()), + Command::ClusterFailover { .. } => { + Frame::Error("ERR FAILOVER not yet implemented".into()) + } Command::Migrate { host, diff --git a/tests/integration/src/cluster.rs b/tests/integration/src/cluster.rs index f6b2abdd..73ed2a4f 100644 --- a/tests/integration/src/cluster.rs +++ b/tests/integration/src/cluster.rs @@ -230,13 +230,14 @@ async fn cluster_getkeysinslot() { } } -// -- stubs / error responses -- +// -- cluster replicate / failover -- #[tokio::test] -async fn cluster_replicate_stub() { +async fn cluster_replicate_unknown_node() { let server = cluster_server(); let mut c = server.connect().await; + // using a well-formed UUID that isn't in the cluster let err = c .err(&[ "CLUSTER", @@ -245,8 +246,34 @@ async fn cluster_replicate_stub() { ]) .await; assert!( - err.contains("not yet supported"), - "expected stub error, got: {err}" + err.contains("Unknown node ID"), + "expected unknown node error, got: {err}" + ); +} + +#[tokio::test] +async fn cluster_replicate_self_rejected() { + let server = cluster_server(); + let mut c = server.connect().await; + + // fetch our own node ID first + let my_id = c.get_bulk(&["CLUSTER", "MYID"]).await.expect("MYID returned nil"); + let err = c.err(&["CLUSTER", "REPLICATE", &my_id]).await; + assert!( + err.contains("Cannot replicate self"), + "expected self-replicate error, got: {err}" + ); +} + +#[tokio::test] +async fn cluster_replicate_invalid_id() { + let server = cluster_server(); + let mut c = server.connect().await; + + let err = c.err(&["CLUSTER", "REPLICATE", "not-a-uuid"]).await; + assert!( + err.contains("Invalid node ID"), + "expected invalid node ID error, got: {err}" ); } @@ -257,7 +284,7 @@ async fn cluster_failover_stub() { let err = c.err(&["CLUSTER", "FAILOVER"]).await; assert!( - err.contains("not yet supported"), + err.contains("not yet implemented"), "expected stub error, got: {err}" ); }