From 6dba483a4d344d2531ccc6154f8d39bf25b8dae6 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Thu, 12 Nov 2026 18:18:51 -0500 Subject: [PATCH 1/5] feat: parse the AUTH and AUTH2 options of MIGRATE --- crates/ember-protocol/src/command/mod.rs | 13 +++++++- .../src/command/parse/cluster.rs | 23 +++++++++++++- .../ember-protocol/src/command/parse/mod.rs | 4 ++- crates/ember-protocol/src/command/tests.rs | 31 +++++++++++++++++++ crates/ember-protocol/src/lib.rs | 2 +- 5 files changed, 69 insertions(+), 4 deletions(-) diff --git a/crates/ember-protocol/src/command/mod.rs b/crates/ember-protocol/src/command/mod.rs index 3b55a02..efe2198 100644 --- a/crates/ember-protocol/src/command/mod.rs +++ b/crates/ember-protocol/src/command/mod.rs @@ -19,6 +19,14 @@ pub enum SetExpire { PxAt(u64), } +/// Credentials MIGRATE sends to the target with AUTH. +#[derive(Debug, Clone, PartialEq)] +pub struct MigrateAuth { + /// Set by AUTH2. AUTH uses the default user. + pub username: Option, + pub password: String, +} + /// A parsed client command, ready for execution. #[derive(Debug, Clone, PartialEq)] pub enum Command { @@ -629,7 +637,8 @@ pub enum Command { /// CLUSTER GETKEYSINSLOT `slot` `count`. Return keys in a slot. ClusterGetKeysInSlot { slot: u16, count: u32 }, - /// MIGRATE `host` `port` `key` `db` `timeout` \[COPY\] \[REPLACE\] \[KEYS key...\]. + /// MIGRATE `host` `port` `key` `db` `timeout` \[COPY\] \[REPLACE\] + /// \[AUTH password | AUTH2 username password\]. /// Migrate a key to another node. Migrate { host: String, @@ -639,6 +648,8 @@ pub enum Command { timeout_ms: u64, copy: bool, replace: bool, + /// Credentials for the target, sent with AUTH before the key. + auth: Option, }, /// RESTORE `key` `ttl` `serialized-value` \[REPLACE\]. diff --git a/crates/ember-protocol/src/command/parse/cluster.rs b/crates/ember-protocol/src/command/parse/cluster.rs index 8d663c6..d447171 100644 --- a/crates/ember-protocol/src/command/parse/cluster.rs +++ b/crates/ember-protocol/src/command/parse/cluster.rs @@ -230,6 +230,7 @@ pub(super) fn parse_cluster_setslot(args: &[Frame]) -> Result Result { // MIGRATE host port key db timeout [COPY] [REPLACE] + // [AUTH password | AUTH2 username password] if args.len() < 5 { return Err(wrong_arity("MIGRATE")); } @@ -246,13 +247,32 @@ pub(super) fn parse_migrate(args: &[Frame]) -> Result { let mut copy = false; let mut replace = false; + let mut auth = None; - for arg in &args[5..] { + let mut rest = args[5..].iter(); + while let Some(arg) = rest.next() { let mut kw = [0u8; MAX_KEYWORD_LEN]; let opt = uppercase_arg(arg, &mut kw)?; + let mut next = || { + rest.next() + .ok_or_else(|| wrong_arity("MIGRATE")) + .and_then(extract_string) + }; match opt { "COPY" => copy = true, "REPLACE" => replace = true, + "AUTH" => { + auth = Some(MigrateAuth { + username: None, + password: next()?, + }) + } + "AUTH2" => { + auth = Some(MigrateAuth { + username: Some(next()?), + password: next()?, + }) + } _ => { return Err(ProtocolError::InvalidCommandFrame(format!( "unknown MIGRATE option '{opt}'" @@ -269,5 +289,6 @@ pub(super) fn parse_migrate(args: &[Frame]) -> Result { timeout_ms, copy, replace, + auth, }) } diff --git a/crates/ember-protocol/src/command/parse/mod.rs b/crates/ember-protocol/src/command/parse/mod.rs index 28b6660..46627a2 100644 --- a/crates/ember-protocol/src/command/parse/mod.rs +++ b/crates/ember-protocol/src/command/parse/mod.rs @@ -9,7 +9,9 @@ use bytes::Bytes; use crate::error::ProtocolError; use crate::types::Frame; -use super::{BitOpKind, BitRange, BitRangeUnit, Command, ScoreBound, SetExpire, ZAddFlags}; +use super::{ + BitOpKind, BitRange, BitRangeUnit, Command, MigrateAuth, ScoreBound, SetExpire, ZAddFlags, +}; mod cluster; mod hashes; diff --git a/crates/ember-protocol/src/command/tests.rs b/crates/ember-protocol/src/command/tests.rs index d23ddbe..d49d26c 100644 --- a/crates/ember-protocol/src/command/tests.rs +++ b/crates/ember-protocol/src/command/tests.rs @@ -1830,6 +1830,7 @@ fn migrate_basic() { timeout_ms: 5000, copy: false, replace: false, + auth: None, }, ); } @@ -1856,10 +1857,40 @@ fn migrate_with_options() { timeout_ms: 10000, copy: true, replace: true, + auth: None, }, ); } +#[test] +fn migrate_with_auth() { + let parse = |extra: &[&str]| { + let mut args = vec!["MIGRATE", "h", "1", "k", "0", "10"]; + args.extend_from_slice(extra); + match Command::from_frame(cmd(&args)) { + Ok(Command::Migrate { auth, .. }) => Ok(auth), + Ok(other) => panic!("{other:?}"), + Err(e) => Err(e), + } + }; + assert_eq!( + parse(&["AUTH", "pw"]).unwrap(), + Some(MigrateAuth { + username: None, + password: "pw".into() + }) + ); + assert_eq!( + parse(&["REPLACE", "AUTH2", "u", "pw"]).unwrap(), + Some(MigrateAuth { + username: Some("u".into()), + password: "pw".into() + }) + ); + assert!(parse(&["AUTH"]).is_err()); + assert!(parse(&["AUTH2", "u"]).is_err()); +} + #[test] fn migrate_wrong_arity() { let err = Command::from_frame(cmd(&["MIGRATE", "host", "port", "key"])).unwrap_err(); diff --git a/crates/ember-protocol/src/lib.rs b/crates/ember-protocol/src/lib.rs index d3830e4..e68f149 100644 --- a/crates/ember-protocol/src/lib.rs +++ b/crates/ember-protocol/src/lib.rs @@ -27,7 +27,7 @@ mod serialize; pub mod slots; pub mod types; -pub use command::{Command, SetExpire, ZAddFlags}; +pub use command::{Command, MigrateAuth, SetExpire, ZAddFlags}; pub use error::ProtocolError; pub use parse::{parse_frame, parse_frame_bytes, parse_request}; pub use types::wire; From 110d57c4927cb8f5ae10e18f1c83ec25ac41fc56 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Thu, 12 Nov 2026 18:18:51 -0500 Subject: [PATCH 2/5] fix: send ASKING from MIGRATE and ASK for keys a migrating slot no longer has --- crates/ember-cluster/README.md | 2 +- crates/ember-cluster/src/migration.rs | 39 +-- crates/ember-server/src/cluster/commands.rs | 57 ++-- crates/ember-server/src/cluster/mod.rs | 12 + crates/ember-server/src/cluster/tests.rs | 79 ++---- .../ember-server/src/connection/dispatch.rs | 35 ++- .../src/connection/exec/cluster.rs | 190 +++++++------- crates/ember-server/src/connection/execute.rs | 19 +- crates/ember-server/src/connection/handler.rs | 3 +- tests/integration/src/cluster.rs | 245 +++++++++++++++++- 10 files changed, 450 insertions(+), 231 deletions(-) diff --git a/crates/ember-cluster/README.md b/crates/ember-cluster/README.md index fa57b5a..7dbd68c 100644 --- a/crates/ember-cluster/README.md +++ b/crates/ember-cluster/README.md @@ -61,7 +61,7 @@ the following CLUSTER commands are supported at the protocol layer: | `CLUSTER COUNTKEYSINSLOT ` | key count in slot | | `CLUSTER GETKEYSINSLOT ` | get keys in slot | | `ASKING` | follow ASK redirect | -| `MIGRATE ` | migrate a key | +| `MIGRATE [COPY] [REPLACE] [AUTH pw \| AUTH2 user pw]` | migrate a key | ## design notes diff --git a/crates/ember-cluster/src/migration.rs b/crates/ember-cluster/src/migration.rs index 38244b3..d5c653f 100644 --- a/crates/ember-cluster/src/migration.rs +++ b/crates/ember-cluster/src/migration.rs @@ -180,8 +180,6 @@ pub struct MigrationManager { outgoing: HashMap, /// Migrations where this node is the target (slot is importing in). incoming: HashMap, - /// Keys that have been migrated but not yet confirmed. - pending_keys: HashMap>>, /// Keys a MIGRATE is copying to another node right now. in_flight: HashSet>, } @@ -230,7 +228,6 @@ impl MigrationManager { let migration = Migration::new_importing(slot, source, local_id); self.incoming.insert(slot, migration); - self.pending_keys.insert(slot, HashSet::new()); self.incoming .get(&slot) .ok_or(MigrationError::NoMigrationInProgress { slot }) @@ -254,17 +251,13 @@ impl MigrationManager { let migration = Migration::new_migrating(slot, local_id, target); self.outgoing.insert(slot, migration); - self.pending_keys.insert(slot, HashSet::new()); self.outgoing .get(&slot) .ok_or(MigrationError::NoMigrationInProgress { slot }) } - /// Record that a key has been migrated. - pub fn key_migrated(&mut self, slot: u16, key: Vec) { - if let Some(keys) = self.pending_keys.get_mut(&slot) { - keys.insert(key); - } + /// Counts a key that MIGRATE moved out of `slot`. + pub fn key_migrated(&mut self, slot: u16) { if let Some(migration) = self.outgoing.get_mut(&slot) { migration.record_migrated(1); } @@ -286,16 +279,8 @@ impl MigrationManager { self.in_flight.contains(key) } - /// Check if a specific key has been migrated. - pub fn is_key_migrated(&self, slot: u16, key: &[u8]) -> bool { - self.pending_keys - .get(&slot) - .is_some_and(|keys| keys.contains(key)) - } - /// Complete a migration and clean up state. pub fn complete_migration(&mut self, slot: u16) -> Option { - self.pending_keys.remove(&slot); // Try outgoing first, then incoming self.outgoing .remove(&slot) @@ -308,7 +293,6 @@ impl MigrationManager { /// Abort a migration and clean up state. pub fn abort_migration(&mut self, slot: u16) -> Option { - self.pending_keys.remove(&slot); self.outgoing .remove(&slot) .or_else(|| self.incoming.remove(&slot)) @@ -615,18 +599,12 @@ mod tests { } #[test] - fn manager_key_tracking() { + fn manager_counts_migrated_keys() { let mut manager = MigrationManager::new(); - let local = node_id(); - let target = node_id(); - - manager.start_migrate(100, local, target).unwrap(); - - assert!(!manager.is_key_migrated(100, b"key1")); - - manager.key_migrated(100, b"key1".to_vec()); - assert!(manager.is_key_migrated(100, b"key1")); - assert!(!manager.is_key_migrated(100, b"key2")); + manager.start_migrate(100, node_id(), node_id()).unwrap(); + manager.key_migrated(100); + manager.key_migrated(100); + assert_eq!(manager.get_outgoing(100).unwrap().keys_migrated, 2); } #[test] @@ -636,7 +614,7 @@ mod tests { let target = node_id(); manager.start_migrate(100, local, target).unwrap(); - manager.key_migrated(100, b"key1".to_vec()); + manager.key_migrated(100); let completed = manager.complete_migration(100); assert!(completed.is_some()); @@ -644,7 +622,6 @@ mod tests { // State should be cleaned up assert!(!manager.is_migrating(100)); - assert!(!manager.is_key_migrated(100, b"key1")); } #[test] diff --git a/crates/ember-server/src/cluster/commands.rs b/crates/ember-server/src/cluster/commands.rs index d1f8399..ff02a7a 100644 --- a/crates/ember-server/src/cluster/commands.rs +++ b/crates/ember-server/src/cluster/commands.rs @@ -339,14 +339,13 @@ impl ClusterCoordinator { /// Checks slot ownership with migration-aware routing. /// - /// Returns `None` if the command should be handled locally. - /// Returns `Some(Frame)` with MOVED, ASK, or CLUSTERDOWN if not. - /// - /// During migration: - /// - Source node returns ASK for keys already transferred to the target - /// - Target node allows access when the client sent ASKING - /// - Keys not yet migrated are served locally by the source - pub async fn check_slot_with_migration(&self, key: &[u8], asking: bool) -> Option { + /// During a migration, as in Redis Cluster: + /// - the source serves the keys it still has, and sends a client whose + /// keys are gone to the target with ASK ([`SlotCheck::AskIfMissing`]). + /// Keys written during the move so land on the target too + /// - the target serves the slot to a client that sent ASKING + /// - a key MIGRATE is copying right now gets TRYAGAIN + pub async fn check_slot_with_migration(&self, key: &[u8], asking: bool) -> SlotCheck { let slot = key_slot(key); let state = self.state.read().await; let migration = self.migration.lock().await; @@ -355,40 +354,38 @@ impl ClusterCoordinator { // MIGRATE is copying this key: a write now would be lost when the // local copy is deleted, so the client retries shortly if migration.is_key_in_flight(key) { - return Some(Frame::Error( + return SlotCheck::Reply(Frame::Error( "TRYAGAIN the key is being migrated, retry shortly".into(), )); } - // slot is migrating out — if key already moved, ASK redirect - if migration.is_migrating(slot) && migration.is_key_migrated(slot, key) { - if let Some(m) = migration.get_outgoing(slot) { - if let Some(target) = state.nodes.get(&m.target) { - return Some(Frame::Error(format!("ASK {} {}", slot, target.addr))); - } - } - } - return None; // handle locally + let target = migration + .get_outgoing(slot) + .and_then(|m| state.nodes.get(&m.target)); + return match target { + Some(target) => SlotCheck::AskIfMissing { + slot, + addr: target.addr.to_string(), + }, + None => SlotCheck::Local, + }; } // not our slot — but are we importing it and client sent ASKING? if migration.is_importing(slot) && asking { - return None; // allow access + return SlotCheck::Local; } // standard redirect - match state.slot_owner(slot) { - Some(owner) => Some(Frame::Error(format!("MOVED {} {}", slot, owner.addr))), - None => Some(Frame::Error("CLUSTERDOWN Hash slot not served".into())), - } + SlotCheck::Reply(match state.slot_owner(slot) { + Some(owner) => Frame::Error(format!("MOVED {} {}", slot, owner.addr)), + None => Frame::Error("CLUSTERDOWN Hash slot not served".into()), + }) } - /// Marks a key as migrated during slot migration. - /// - /// Called after MIGRATE successfully transfers a key to the target node. - /// Subsequent accesses for this key on the source will return ASK redirects. - pub async fn mark_key_migrated(&self, slot: u16, key: &[u8]) { - let mut migration = self.migration.lock().await; - migration.key_migrated(slot, key.to_vec()); + /// Counts a key MIGRATE moved out of `slot`, for the migration's + /// progress. + pub async fn mark_key_migrated(&self, slot: u16) { + self.migration.lock().await.key_migrated(slot); } /// Marks a key as being copied by MIGRATE, so commands for it get diff --git a/crates/ember-server/src/cluster/mod.rs b/crates/ember-server/src/cluster/mod.rs index 012c29e..87b9eac 100644 --- a/crates/ember-server/src/cluster/mod.rs +++ b/crates/ember-server/src/cluster/mod.rs @@ -26,6 +26,18 @@ use tokio::net::UdpSocket; use tokio::sync::{mpsc, Mutex, RwLock}; use tracing::{debug, error, info, warn}; +/// Where a command's keys are served, from [`ClusterCoordinator::check_slot_with_migration`]. +#[derive(Debug, PartialEq)] +pub enum SlotCheck { + /// Run the command here. + Local, + /// Reply with this error instead: MOVED, CLUSTERDOWN or TRYAGAIN. + Reply(Frame), + /// The slot is moving to `addr`. Serve the command here if its keys + /// are still here, and send the client there with ASK if they are gone. + AskIfMissing { slot: u16, addr: String }, +} + mod commands; mod failover; mod gossip; diff --git a/crates/ember-server/src/cluster/tests.rs b/crates/ember-server/src/cluster/tests.rs index bfee674..e37e80c 100644 --- a/crates/ember-server/src/cluster/tests.rs +++ b/crates/ember-server/src/cluster/tests.rs @@ -214,11 +214,11 @@ async fn check_slot_owned_no_migration() { let (coord, _rx) = test_coordinator_bootstrapped(); // "foo" hashes to some slot — we own all slots let result = coord.check_slot_with_migration(b"foo", false).await; - assert!(result.is_none(), "should handle locally"); + assert_eq!(result, SlotCheck::Local); } #[tokio::test] -async fn check_slot_ask_when_key_migrated() { +async fn a_migrating_slot_asks_for_keys_that_are_gone() { let (coord, _rx) = test_coordinator_bootstrapped(); let target = NodeId::new(); @@ -237,25 +237,20 @@ async fn check_slot_ask_when_key_migrated() { .cluster_setslot_migrating(slot, &target.0.to_string()) .await; - // mark "foo" as migrated via the migration manager directly - { - let mut migration = coord.migration.lock().await; - migration.key_migrated(slot, b"foo".to_vec()); - } - - // should return ASK redirect + // the connection serves the keys still here and sends the client to + // the target for the rest let result = coord.check_slot_with_migration(b"foo", false).await; - match result { - Some(Frame::Error(msg)) => { - assert!(msg.starts_with("ASK"), "expected ASK, got: {msg}"); - assert!(msg.contains("127.0.0.1:6380")); + assert_eq!( + result, + SlotCheck::AskIfMissing { + slot, + addr: "127.0.0.1:6380".into() } - other => panic!("expected ASK error, got {other:?}"), - } + ); } #[tokio::test] -async fn check_slot_local_when_key_not_migrated() { +async fn a_migrating_slot_with_an_unknown_target_is_served_here() { let (coord, _rx) = test_coordinator_bootstrapped(); let target = NodeId::new(); @@ -264,9 +259,9 @@ async fn check_slot_local_when_key_not_migrated() { .cluster_setslot_migrating(slot, &target.0.to_string()) .await; - // "foo" NOT migrated yet — should serve locally + // there is nowhere to send the client yet let result = coord.check_slot_with_migration(b"foo", false).await; - assert!(result.is_none(), "should handle locally"); + assert_eq!(result, SlotCheck::Local); } #[tokio::test] @@ -281,7 +276,7 @@ async fn check_slot_importing_with_asking() { // with asking=true, should allow local access let result = coord.check_slot_with_migration(b"foo", true).await; - assert!(result.is_none(), "should allow with ASKING"); + assert_eq!(result, SlotCheck::Local); } #[tokio::test] @@ -311,7 +306,7 @@ async fn check_slot_importing_without_asking() { // without asking, should return MOVED to the owner let result = coord.check_slot_with_migration(b"foo", false).await; match result { - Some(Frame::Error(msg)) => { + SlotCheck::Reply(Frame::Error(msg)) => { assert!(msg.starts_with("MOVED"), "expected MOVED, got: {msg}"); } other => panic!("expected MOVED error, got {other:?}"), @@ -323,7 +318,7 @@ async fn check_slot_unassigned_returns_clusterdown() { let (coord, _rx) = test_coordinator(); // no slots assigned let result = coord.check_slot_with_migration(b"foo", false).await; match result { - Some(Frame::Error(msg)) => { + SlotCheck::Reply(Frame::Error(msg)) => { assert!( msg.contains("CLUSTERDOWN"), "expected CLUSTERDOWN, got: {msg}" @@ -391,48 +386,6 @@ async fn from_config_restores_coordinator() { assert!(state.owns_slot(16383)); } -#[tokio::test] -async fn mark_key_migrated_triggers_ask_redirect() { - let (coord, _rx) = test_coordinator_bootstrapped(); - - // set up a migration: slot 12182 (hash of "foo") migrating to a target - let target = NodeId::new(); - let target_addr: SocketAddr = "127.0.0.1:6380".parse().unwrap(); - - // add target node to state - { - let mut state = coord.state.write().await; - state.add_node(ClusterNode::new_primary(target, target_addr)); - } - - // start migrating slot 12182 - coord - .cluster_setslot_migrating(12182, &target.0.to_string()) - .await; - - // before marking the key, it should be served locally - let result = coord.check_slot_with_migration(b"foo", false).await; - assert!( - result.is_none(), - "key should be served locally before migration" - ); - - // mark the key as migrated - coord.mark_key_migrated(12182, b"foo").await; - - // now it should return ASK - let result = coord.check_slot_with_migration(b"foo", false).await; - match result { - Some(Frame::Error(msg)) => { - assert!( - msg.starts_with("ASK 12182"), - "expected ASK redirect, got: {msg}" - ); - } - other => panic!("expected ASK redirect, got {other:?}"), - } -} - // -- cluster_replicate -- #[tokio::test] diff --git a/crates/ember-server/src/connection/dispatch.rs b/crates/ember-server/src/connection/dispatch.rs index ceae4b4..0df2825 100644 --- a/crates/ember-server/src/connection/dispatch.rs +++ b/crates/ember-server/src/connection/dispatch.rs @@ -4,6 +4,7 @@ use std::sync::atomic::Ordering; use std::sync::Arc; use std::time::{Instant, SystemTime, UNIX_EPOCH}; +use crate::cluster::SlotCheck; use crate::connection_common::{ frame_to_monitor_args, validate_command_sizes, MonitorEvent, Session, }; @@ -11,7 +12,7 @@ use crate::pubsub::PubSubManager; use crate::server::ServerContext; use crate::slowlog::SlowLog; use bytes::Bytes; -use ember_core::Engine; +use ember_core::{Engine, ShardRequest, ShardResponse}; use ember_protocol::{Command, Frame}; use super::route::Routing; @@ -215,7 +216,7 @@ pub(super) async fn prepare_command( }; // cluster slot validation (migration-aware when cluster is enabled) - if let Some(redirect) = cluster_slot_check(ctx, &cmd, was_asking).await { + if let Some(redirect) = cluster_slot_check(ctx, engine, &cmd, was_asking).await { return PreparedDispatch::Immediate(PendingResponse::Immediate(redirect)); } @@ -245,6 +246,8 @@ pub(super) async fn prepare_command( /// primary (READONLY if it is unknown) for a write on a replica /// - CROSSSLOT when the command's keys hash to different slots /// - MOVED, ASK or CLUSTERDOWN when this node does not serve the key's slot +/// - while the slot migrates out: ASK when none of the keys are still here, +/// and TRYAGAIN when only some are, as in Redis Cluster /// /// Every command path calls this, so the write checks cover pipelined /// commands as well as those that go through `execute`. The `asking` flag @@ -252,6 +255,7 @@ pub(super) async fn prepare_command( /// to importing slots during migration. pub(super) async fn cluster_slot_check( ctx: &ServerContext, + engine: &Engine, cmd: &Command, asking: bool, ) -> Option { @@ -290,7 +294,32 @@ pub(super) async fn cluster_slot_check( return Some(err); } } - cluster + match cluster .check_slot_with_migration(first.as_bytes(), asking) .await + { + SlotCheck::Local => None, + SlotCheck::Reply(frame) => Some(frame), + SlotCheck::AskIfMissing { slot, addr } => { + let mut here = 0; + for key in keys.iter() { + let exists = ShardRequest::Exists { + key: key.to_owned(), + }; + let idx = engine.shard_for_key(key); + if let Ok(ShardResponse::Bool(true)) = engine.send_to_shard(idx, exists).await { + here += 1; + } + } + if here == keys.iter().count() { + None + } else if here == 0 { + Some(Frame::Error(format!("ASK {slot} {addr}"))) + } else { + Some(Frame::Error( + "TRYAGAIN Multiple keys request during rehashing of slot".into(), + )) + } + } + } } diff --git a/crates/ember-server/src/connection/exec/cluster.rs b/crates/ember-server/src/connection/exec/cluster.rs index 067436f..3782a01 100644 --- a/crates/ember-server/src/connection/exec/cluster.rs +++ b/crates/ember-server/src/connection/exec/cluster.rs @@ -5,7 +5,7 @@ use std::time::Duration; use bytes::{Bytes, BytesMut}; use ember_core::{ShardRequest, ShardResponse}; -use ember_protocol::{parse_frame, Frame}; +use ember_protocol::{parse_frame, Frame, MigrateAuth}; use super::ExecCtx; @@ -194,124 +194,128 @@ pub(in crate::connection) async fn cluster_failover( } } -pub(in crate::connection) async fn migrate( - host: String, - port: u16, - key: String, - timeout_ms: u64, - replace: bool, - cx: &ExecCtx<'_>, -) -> Frame { +/// What MIGRATE was asked to do with one key. +pub(in crate::connection) struct MigrateRequest { + pub host: String, + pub port: u16, + pub key: String, + pub timeout_ms: u64, + /// Keep the local key. + pub copy: bool, + /// Overwrite the key on the target. + pub replace: bool, + pub auth: Option, +} + +pub(in crate::connection) async fn migrate(req: MigrateRequest, cx: &ExecCtx<'_>) -> Frame { // other clients must not change the key between the dump and the local // delete, or their write is lost. hold it until the move is complete. let cluster = cx.ctx.cluster.as_ref(); if let Some(c) = cluster { - if !c.begin_key_transfer(key.as_bytes()).await { + if !c.begin_key_transfer(req.key.as_bytes()).await { return Frame::Error("ERR the key is already being migrated".into()); } } - let reply = transfer_key(host, port, &key, timeout_ms, replace, cx).await; + let reply = transfer_key(&req, cx).await; if let Some(c) = cluster { - c.end_key_transfer(key.as_bytes()).await; + c.end_key_transfer(req.key.as_bytes()).await; } reply } -/// Copies `key` to the target with RESTORE, then deletes it locally and -/// marks it migrated so later commands for it are sent there with ASK. -async fn transfer_key( - host: String, - port: u16, - key: &str, - timeout_ms: u64, - replace: bool, - cx: &ExecCtx<'_>, -) -> Frame { - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - - // dump the key from the local shard +/// Copies the key to the target with RESTORE, then, unless COPY was given, +/// deletes it locally, so later commands for it are sent there with ASK. +async fn transfer_key(req: &MigrateRequest, cx: &ExecCtx<'_>) -> Frame { + let key = &req.key; let idx = cx.engine.shard_for_key(key); - let dump_req = ShardRequest::DumpKey { - key: key.to_owned(), - }; - let dump_resp = match cx.engine.send_to_shard(idx, dump_req).await { - Ok(r) => r, + let dump_req = ShardRequest::DumpKey { key: key.clone() }; + let (data, ttl_ms) = match cx.engine.send_to_shard(idx, dump_req).await { + Ok(ShardResponse::KeyDump { data, ttl_ms }) => (data, ttl_ms), + Ok(ShardResponse::Value(None)) => return Frame::Simple("NOKEY".into()), + Ok(_) => return Frame::Error("ERR internal error".into()), Err(e) => return Frame::Error(format!("ERR {e}")), }; - let (data, ttl_ms) = match dump_resp { - ShardResponse::KeyDump { data, ttl_ms } => (data, ttl_ms), - ShardResponse::Value(None) => { - return Frame::Error("ERR no such key".into()); - } - _ => return Frame::Error("ERR internal error".into()), + // the target is importing the slot, so it serves the key only after + // ASKING. a target with a password needs AUTH first, which MIGRATE + // takes as an option, as in Redis. + let bulk = |s: &str| Frame::Bulk(Bytes::copy_from_slice(s.as_bytes())); + let mut commands = Vec::new(); + if let Some(auth) = &req.auth { + let mut parts = vec![bulk("AUTH")]; + parts.extend(auth.username.as_deref().map(bulk)); + parts.push(bulk(&auth.password)); + commands.push(Frame::Array(parts)); + } + commands.push(Frame::Array(vec![bulk("ASKING")])); + let mut restore = vec![ + bulk("RESTORE"), + bulk(key), + bulk(&ttl_ms.max(0).to_string()), + Frame::Bulk(Bytes::from(data)), + ]; + if req.replace { + restore.push(bulk("REPLACE")); + } + commands.push(Frame::Array(restore)); + + let addr = format!("{}:{}", req.host, req.port); + let timeout = Duration::from_millis(req.timeout_ms.max(1000)); + let replies = match tokio::time::timeout(timeout, send_commands(&addr, &commands)).await { + Ok(Ok(replies)) => replies, + Ok(Err(e)) => return Frame::Error(format!("IOERR error talking to the target: {e}")), + Err(_) => return Frame::Error("IOERR timeout talking to the target".into()), }; + if let Some(Frame::Error(e)) = replies.iter().find(|r| matches!(r, Frame::Error(_))) { + return Frame::Error(format!("ERR target error: {e}")); + } - // send RESTORE to the target node - let ttl_arg = if ttl_ms < 0 { 0u64 } else { ttl_ms as u64 }; - let timeout = Duration::from_millis(timeout_ms.max(1000)); - let addr = format!("{host}:{port}"); - - let result = tokio::time::timeout(timeout, async { - let mut stream = tokio::net::TcpStream::connect(&addr).await?; - - // build RESTORE command as RESP3 array - let mut parts = vec![ - Frame::Bulk(Bytes::from("RESTORE")), - Frame::Bulk(Bytes::from(key.to_owned())), - Frame::Bulk(Bytes::from(ttl_arg.to_string())), - Frame::Bulk(Bytes::from(data)), - ]; - if replace { - parts.push(Frame::Bulk(Bytes::from("REPLACE"))); + if !req.copy { + // the key is still marked in flight, so no write lands between the + // RESTORE and this delete + let _ = cx + .engine + .send_to_shard(idx, ShardRequest::Del { key: key.clone() }) + .await; + if let Some(c) = &cx.ctx.cluster { + c.mark_key_migrated(ember_cluster::key_slot(key.as_bytes())) + .await; } - let cmd_frame = Frame::Array(parts); + } + Frame::Simple("OK".into()) +} - let mut buf = BytesMut::new(); - cmd_frame.serialize(&mut buf); - stream.write_all(&buf).await?; +/// Sends `commands` to `addr` in one write and reads one reply for each. +async fn send_commands(addr: &str, commands: &[Frame]) -> io::Result> { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; - // read response - let mut read_buf = BytesMut::with_capacity(256); - loop { - let n = stream.read_buf(&mut read_buf).await?; - if n == 0 { - return Err(io::Error::new( - io::ErrorKind::UnexpectedEof, - "connection closed by target", - )); + let mut stream = tokio::net::TcpStream::connect(addr).await?; + let mut buf = BytesMut::new(); + for command in commands { + command.serialize(&mut buf); + } + stream.write_all(&buf).await?; + + let mut replies = Vec::with_capacity(commands.len()); + let mut read_buf = BytesMut::with_capacity(256); + while replies.len() < commands.len() { + match parse_frame(&read_buf) { + Ok(Some((frame, used))) => { + let _ = read_buf.split_to(used); + replies.push(frame); } - match parse_frame(&read_buf) { - Ok(Some((frame, _))) => return Ok(frame), - Ok(None) => {} // need more data - Err(e) => { - return Err(io::Error::new(io::ErrorKind::InvalidData, e.to_string())); + Ok(None) => { + if stream.read_buf(&mut read_buf).await? == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "connection closed by target", + )); } } + Err(e) => return Err(io::Error::new(io::ErrorKind::InvalidData, e.to_string())), } - }) - .await; - - match result { - Ok(Ok(Frame::Simple(_))) => { - // the target has it: send later commands there, then drop the - // local copy. the key is still marked in flight, so no write can - // land in between. - if let Some(c) = &cx.ctx.cluster { - let slot = ember_cluster::key_slot(key.as_bytes()); - c.mark_key_migrated(slot, key.as_bytes()).await; - } - let del_req = ShardRequest::Del { - key: key.to_owned(), - }; - let _ = cx.engine.send_to_shard(idx, del_req).await; - Frame::Simple("OK".into()) - } - Ok(Ok(Frame::Error(e))) => Frame::Error(format!("ERR target error: {e}")), - Ok(Ok(_)) => Frame::Error("ERR unexpected response from target".into()), - Ok(Err(e)) => Frame::Error(format!("ERR {e}")), - Err(_) => Frame::Error("ERR timeout connecting to target".into()), } + Ok(replies) } pub(in crate::connection) async fn restore( diff --git a/crates/ember-server/src/connection/execute.rs b/crates/ember-server/src/connection/execute.rs index a57898e..5d4185a 100644 --- a/crates/ember-server/src/connection/execute.rs +++ b/crates/ember-server/src/connection/execute.rs @@ -33,7 +33,7 @@ pub(super) async fn execute( ) -> Frame { // cluster checks: replica writes, slot ownership, cross-slot keys. // when `asking` is true, importing slots are allowed through. - if let Some(redirect) = super::dispatch::cluster_slot_check(ctx, &cmd, asking).await { + if let Some(redirect) = super::dispatch::cluster_slot_check(ctx, engine, &cmd, asking).await { return redirect; } @@ -274,9 +274,22 @@ pub(super) async fn execute( port, key, timeout_ms, + copy, replace, - .. - } => exec::cluster::migrate(host, port, key, timeout_ms, replace, &cx).await, + auth, + db: _, + } => { + let req = exec::cluster::MigrateRequest { + host, + port, + key, + timeout_ms, + copy, + replace, + auth, + }; + exec::cluster::migrate(req, &cx).await + } Command::Restore { key, ttl_ms, diff --git a/crates/ember-server/src/connection/handler.rs b/crates/ember-server/src/connection/handler.rs index 4d5344b..7356007 100644 --- a/crates/ember-server/src/connection/handler.rs +++ b/crates/ember-server/src/connection/handler.rs @@ -221,7 +221,8 @@ pub(super) async fn handle_blocking_pop_cmd( } let was_asking = std::mem::take(asking); - if let Some(redirect) = super::dispatch::cluster_slot_check(ctx, &cmd, was_asking).await { + if let Some(redirect) = super::dispatch::cluster_slot_check(ctx, engine, &cmd, was_asking).await + { return redirect; } diff --git a/tests/integration/src/cluster.rs b/tests/integration/src/cluster.rs index eaaace7..681b525 100644 --- a/tests/integration/src/cluster.rs +++ b/tests/integration/src/cluster.rs @@ -298,13 +298,11 @@ async fn cluster_migrate_missing_key() { let server = cluster_server(); let mut c = server.connect().await; - let err = c - .err(&["MIGRATE", "127.0.0.1", "6380", "key", "0", "1000"]) + // as in Redis, a missing key isn't an error + let reply = c + .cmd(&["MIGRATE", "127.0.0.1", "6380", "key", "0", "1000"]) .await; - assert!( - err.contains("no such key"), - "expected missing key error, got: {err}" - ); + assert_eq!(reply, Frame::Simple("NOKEY".into())); } #[tokio::test] @@ -514,6 +512,241 @@ async fn cluster_redirect_followthrough() { assert_eq!(val, Some("value".into())); } +// -- slot migration -- + +/// Finds a hash tag whose slot node 0 owns (slots 0–5460). +async fn tag_on_node_0(cluster: &TestCluster) -> (String, i64) { + let mut c = cluster.connect(0).await; + for i in 0.. { + let tag = format!("{{t{i}}}"); + let slot = c.get_int(&["CLUSTER", "KEYSLOT", &tag]).await; + if slot <= 5460 { + return (tag, slot); + } + } + unreachable!() +} + +/// Moves one slot from node 0 to node 1 the way a reshard does, with keys +/// in it, a key written during the move, and clients following ASK. +#[tokio::test] +async fn migrating_a_slot_moves_its_keys_and_routes_new_ones_to_the_target() { + let cluster = TestCluster::start(); + cluster.init().await; + let (tag, slot) = tag_on_node_0(&cluster).await; + let (old, moved_later, created) = ( + format!("{tag}old"), + format!("{tag}later"), + format!("{tag}new"), + ); + let slot = slot.to_string(); + + let mut source = cluster.connect(0).await; + let mut target = cluster.connect(1).await; + let source_id = source.get_bulk(&["CLUSTER", "MYID"]).await.unwrap(); + let target_id = target.get_bulk(&["CLUSTER", "MYID"]).await.unwrap(); + source.ok(&["SET", &old, "1"]).await; + source.ok(&["SET", &moved_later, "2"]).await; + + target + .ok(&["CLUSTER", "SETSLOT", &slot, "IMPORTING", &source_id]) + .await; + source + .ok(&["CLUSTER", "SETSLOT", &slot, "MIGRATING", &target_id]) + .await; + + // MIGRATE must get past the target, which doesn't own the slot yet + let target_port = cluster.port(1).to_string(); + source + .ok(&["MIGRATE", "127.0.0.1", &target_port, &old, "0", "5000"]) + .await; + + // the moved key is asked for at the target; the one still here is + // served here + let ask = source.cmd(&["GET", &old]).await; + assert!( + matches!(&ask, Frame::Error(e) if e.starts_with(&format!("ASK {slot} "))), + "{ask:?}" + ); + assert_eq!( + source.get_bulk(&["GET", &moved_later]).await, + Some("2".into()) + ); + + // a key created during the move must land on the target, or it would + // be left behind on the source + let ask = source.cmd(&["SET", &created, "3"]).await; + assert!( + matches!(&ask, Frame::Error(e) if e.starts_with("ASK")), + "{ask:?}" + ); + target.ok(&["ASKING"]).await; + target.ok(&["SET", &created, "3"]).await; + + // keys on both sides: a multi-key command waits for the move + let busy = source.cmd(&["MGET", &old, &moved_later]).await; + assert!( + matches!(&busy, Frame::Error(e) if e.starts_with("TRYAGAIN")), + "{busy:?}" + ); + + source + .ok(&[ + "MIGRATE", + "127.0.0.1", + &target_port, + &moved_later, + "0", + "5000", + ]) + .await; + for c in [&mut source, &mut target] { + c.ok(&["CLUSTER", "SETSLOT", &slot, "NODE", &target_id]) + .await; + } + + for (key, value) in [(&old, "1"), (&moved_later, "2"), (&created, "3")] { + assert_eq!(target.get_bulk(&["GET", key]).await, Some(value.into())); + } + assert_eq!( + source.get_int(&["CLUSTER", "COUNTKEYSINSLOT", &slot]).await, + 0 + ); +} + +/// `ember-cli cluster reshard` moves slots with their keys. +#[tokio::test] +async fn cli_reshard_moves_slots_with_their_keys() { + let cluster = TestCluster::start(); + cluster.init().await; + let mut source = cluster.connect(0).await; + let mut target = cluster.connect(1).await; + let source_id = source.get_bulk(&["CLUSTER", "MYID"]).await.unwrap(); + let target_id = target.get_bulk(&["CLUSTER", "MYID"]).await.unwrap(); + + // a few keys in each of node 0's first five slots, which reshard + // moves first + let mut keys = Vec::new(); + for i in 0.. { + let key = format!("k{i}"); + if source.get_int(&["CLUSTER", "KEYSLOT", &key]).await < 5 { + source.ok(&["SET", &key, &i.to_string()]).await; + keys.push((key, i.to_string())); + if keys.len() == 10 { + break; + } + } + } + + let node = format!("127.0.0.1:{}", cluster.port(0)); + let out = crate::helpers::run_cli( + cluster.port(0), + &[ + "cluster", "reshard", &node, "--slots", "5", "--from", &source_id, "--to", &target_id, + ], + ); + assert!( + out.status.success(), + "reshard failed: {}{}", + String::from_utf8_lossy(&out.stdout), + String::from_utf8_lossy(&out.stderr) + ); + + for (key, value) in &keys { + assert_eq!(target.get_bulk(&["GET", key]).await, Some(value.clone())); + } + for slot in 0..5 { + let count = source + .get_int(&["CLUSTER", "COUNTKEYSINSLOT", &slot.to_string()]) + .await; + assert_eq!(count, 0, "slot {slot} left keys behind"); + } +} + +/// MIGRATE AUTH logs in to a target that has a password. +#[tokio::test] +async fn migrate_authenticates_with_the_target() { + let start = || { + TestServer::start_with(ServerOptions { + cluster_enabled: true, + requirepass: Some("pw".into()), + ..Default::default() + }) + }; + let (source_node, target_node) = (start(), start()); + let mut source = source_node.connect().await; + let mut target = target_node.connect().await; + for c in [&mut source, &mut target] { + c.ok(&["AUTH", "pw"]).await; + } + let target_port = target_node.port.to_string(); + source + .ok(&["CLUSTER", "MEET", "127.0.0.1", &target_port]) + .await; + source.ok(&["CLUSTER", "ADDSLOTSRANGE", "0", "16383"]).await; + tokio::time::sleep(std::time::Duration::from_millis(2500)).await; + + let source_id = source.get_bulk(&["CLUSTER", "MYID"]).await.unwrap(); + let slot = source + .get_int(&["CLUSTER", "KEYSLOT", "k"]) + .await + .to_string(); + target + .ok(&["CLUSTER", "SETSLOT", &slot, "IMPORTING", &source_id]) + .await; + source.ok(&["SET", "k", "v"]).await; + + let without = source + .cmd(&["MIGRATE", "127.0.0.1", &target_port, "k", "0", "5000"]) + .await; + assert!( + matches!(&without, Frame::Error(e) if e.contains("NOAUTH")), + "{without:?}" + ); + assert_eq!(source.get_bulk(&["GET", "k"]).await, Some("v".into())); + + source + .ok(&[ + "MIGRATE", + "127.0.0.1", + &target_port, + "k", + "0", + "5000", + "AUTH", + "pw", + ]) + .await; + target.ok(&["ASKING"]).await; + assert_eq!(target.get_bulk(&["GET", "k"]).await, Some("v".into())); +} + +/// MIGRATE COPY leaves the key on the source. +#[tokio::test] +async fn migrate_copy_keeps_the_local_key() { + let cluster = TestCluster::start(); + cluster.init().await; + let (tag, slot) = tag_on_node_0(&cluster).await; + let key = format!("{tag}k"); + let slot = slot.to_string(); + + let mut source = cluster.connect(0).await; + let mut target = cluster.connect(1).await; + let source_id = source.get_bulk(&["CLUSTER", "MYID"]).await.unwrap(); + target + .ok(&["CLUSTER", "SETSLOT", &slot, "IMPORTING", &source_id]) + .await; + source.ok(&["SET", &key, "v"]).await; + + let port = cluster.port(1).to_string(); + source + .ok(&["MIGRATE", "127.0.0.1", &port, &key, "0", "5000", "COPY"]) + .await; + assert_eq!(source.get_bulk(&["GET", &key]).await, Some("v".into())); + target.ok(&["ASKING"]).await; + assert_eq!(target.get_bulk(&["GET", &key]).await, Some("v".into())); +} + // -- automatic failover -- /// End-to-end automatic failover: a 3-node cluster where node 0 owns all From eb6677fe11e291c9a4462182c574d8c7f3d3aff4 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Thu, 12 Nov 2026 18:18:51 -0500 Subject: [PATCH 3/5] fix: stop cluster reshard on a key that fails to move --- crates/ember-cli/src/cluster.rs | 51 +++++++++++++++++++++++---------- 1 file changed, 36 insertions(+), 15 deletions(-) diff --git a/crates/ember-cli/src/cluster.rs b/crates/ember-cli/src/cluster.rs index c45919d..d4a3b5d 100644 --- a/crates/ember-cli/src/cluster.rs +++ b/crates/ember-cli/src/cluster.rs @@ -1083,8 +1083,12 @@ async fn run_cluster_reshard( break; } - // 3. migrate keys - loop { + // 3. migrate keys. a key that fails to move stays in the slot, so + // the next batch would list it again: stop instead of looping, and + // leave the slot with the source so the key isn't orphaned + let target_port_s = target_port.to_string(); + let mut stuck = false; + while !stuck { let keys_frame = match source_conn .send(&["CLUSTER", "GETKEYSINSLOT", &slot.to_string(), "100"]) .await @@ -1095,6 +1099,7 @@ async fn run_cluster_reshard( "{}", format!("GETKEYSINSLOT failed for slot {slot}: {e}").red() ); + stuck = true; break; } }; @@ -1105,18 +1110,25 @@ async fn run_cluster_reshard( } for key in &keys { - let r = source_conn - .send(&[ - "MIGRATE", - &target_host, - &target_port.to_string(), - key, - "0", - "5000", - "REPLACE", - ]) - .await; - if !matches!(&r, Ok(f) if is_ok(f)) { + let mut args = vec![ + "MIGRATE", + &target_host, + &target_port_s, + key, + "0", + "5000", + "REPLACE", + ]; + if let Some(pw) = password { + args.extend(["AUTH", pw]); + } + let r = source_conn.send(&args).await; + // NOKEY: the key expired or was deleted since it was listed + let moved = matches!( + &r, + Ok(ember_protocol::types::Frame::Simple(s)) if s == "OK" || s == "NOKEY" + ); + if !moved { eprintln!( "{}", format!( @@ -1125,10 +1137,19 @@ async fn run_cluster_reshard( ) .red() ); - // continue with other keys + stuck = true; + break; } } } + if stuck { + eprintln!( + "{}", + format!("slot {slot} is still migrating; fix the error and run reshard again") + .red() + ); + break; + } // 4. SETSLOT NODE on both source and target (and the node we're connected to) let slot_s = slot.to_string(); From 7f0fed620b744a9432a1344a9346974c7a76c50f Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Thu, 12 Nov 2026 18:18:51 -0500 Subject: [PATCH 4/5] docs: describe migration routing and MIGRATE AUTH --- CHANGELOG.md | 5 +++++ README.md | 2 +- docs/production-checklist.md | 2 +- 3 files changed, 7 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index dd2f6ba..2615f46 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,11 @@ - 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 +- slot migration: + - 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 + - while a slot migrates, the source sends ASK for any key it no longer has, not only for keys MIGRATE moved, so keys written during the move land on the target instead of being left behind. a command with keys on both sides gets TRYAGAIN + - `ember-cli cluster reshard` passes its password to MIGRATE, and stops on a failed key instead of listing it forever and then reassigning a slot that still held keys - AOF durability (#449): - under `appendfsync everysec` and `no`, writes now reach the OS before their replies, so a crash of the process loses no acknowledged write. before, they sat in an 8 KB buffer, and `no` never flushed it - a write that fails part way, as on a full disk, no longer leaves a partial record that made the rest of the file unreadable diff --git a/README.md b/README.md index de24d4f..5c531ec 100644 --- a/README.md +++ b/README.md @@ -293,7 +293,7 @@ make cluster-clean # stop + delete ./data/cluster/ | `CLUSTER SETSLOT slot IMPORTING\|MIGRATING\|NODE\|STABLE ...` | manage slot migration state | | `CLUSTER COUNTKEYSINSLOT slot` | count keys owned by a slot on this node | | `CLUSTER GETKEYSINSLOT slot count` | list keys in a slot (used during migration) | -| `MIGRATE host port key db timeout [COPY] [REPLACE]` | move a key to another node | +| `MIGRATE host port key db timeout [COPY] [REPLACE] [AUTH pw \| AUTH2 user pw]` | move a key to another node | | `RESTORE key ttl payload` | receive a migrated key | | `ASKING` | tell the server to honor an ASK redirect for the next command | diff --git a/docs/production-checklist.md b/docs/production-checklist.md index d062d63..d62c539 100644 --- a/docs/production-checklist.md +++ b/docs/production-checklist.md @@ -330,7 +330,7 @@ ember-cli cluster reshard ember-cli cluster rebalance ``` -resharding migrates keys live without downtime. clients receive `MOVED` or `ASK` redirects during migration and should follow them automatically. +resharding migrates keys live without downtime. clients receive `MOVED` or `ASK` redirects during migration and should follow them automatically. while a slot moves, the old owner serves the keys it still has and sends clients to the new owner for the rest, so keys written during the move land on the new owner. a command whose keys are split between the two gets `TRYAGAIN`. if the nodes have a password, `ember-cli cluster reshard -a ` passes it to `MIGRATE`. --- From 6c4d80d90b998b01638164691719d9e3fd3b1acb Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Thu, 12 Nov 2026 18:19:08 -0500 Subject: [PATCH 5/5] docs: link the changelog entry to its PR --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2615f46..6ff1c79 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,7 +11,7 @@ - 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 -- slot migration: +- 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 - while a slot migrates, the source sends ASK for any key it no longer has, not only for keys MIGRATE moved, so keys written during the move land on the target instead of being left behind. a command with keys on both sides gets TRYAGAIN