feat: get_node_for_keys -> list_replids_for_keys

Author: Migorithm

duva/src/domains/cluster_actors/actor.rs

@@ -284,14 +284,21 @@ impl<T: TWriteAheadLog> ClusterActor<T> {
             return;
         };
 
-        // TODO get_node_for_keys need to be revisited as currently it takes only the first key
-        match self.hash_ring.get_node_for_keys(&req.request.all_keys()) {
-            | Ok(replid) if replid == self.replication.replid => {
+        match self.hash_ring.list_replids_for_keys(&req.request.all_keys()) {
+            | Ok(replids) if replids.keys().all(|replid| *replid == self.replication.replid) => {
                 self.req_consensus(req).await;
             },
-            | Ok(replid) => {
-                err!("Given keys {:?} moved to {}", req.request.all_keys(), replid);
-                let _ = req.callback.send(format!("MOVED {replid}").into());
+            | Ok(replids) => {
+                let moved_keys = replids
+                    .iter()
+                    .filter_map(|(replid, keys)| {
+                        if *replid != self.replication.replid { Some(keys.iter()) } else { None }
+                    })
+                    .flatten()
+                    .copied()
+                    .collect::<Vec<_>>()
+                    .join(" ");
+                let _ = req.callback.send(format!("MOVED {}", moved_keys).into());
             },
             | Err(err) => {
                 err!("{}", err);

duva/src/domains/cluster_actors/command.rs

@@ -126,7 +126,6 @@ impl FromStr for LazyOption {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct SessionRequest {
     pub(crate) request_id: u64,
-
     pub(crate) client_id: Uuid,
 }
 impl SessionRequest {

duva/src/domains/cluster_actors/hash_ring.rs

@@ -158,12 +158,23 @@ impl HashRing {
         self.vnodes.len()
     }
 
-    pub(crate) fn get_node_for_keys(&self, keys: &[&str]) -> anyhow::Result<ReplicationId> {
-        // Use the first key to determine the node
-        let hash = fnv_1a_hash(keys[0]);
-        self.find_replid(hash)
-            .cloned()
-            .ok_or_else(|| anyhow::anyhow!("No node found for keys: {:?}", keys))
+    pub(crate) fn list_replids_for_keys<'a>(
+        &self,
+        keys: &[&'a str],
+    ) -> anyhow::Result<HashMap<ReplicationId, Vec<&'a str>>> {
+        let mut map: HashMap<ReplicationId, Vec<&str>> = HashMap::new();
+
+        for key in keys {
+            let hash = fnv_1a_hash(key);
+            let replid = self
+                .find_replid(hash)
+                .cloned()
+                .ok_or_else(|| anyhow::anyhow!("No node found for keys: {:?}", keys))?;
+            let v = map.entry(replid).or_default();
+            v.push(key);
+        }
+
+        Ok(map)
     }
 
     #[cfg(test)]