Merge pull request #740 from Migorithm/feat/739

feat: lindex

Author: Migorithm

duva-client/src/broker/mod.rs

@@ -130,7 +130,7 @@ impl Broker {
     // pull-based leader discovery
 
     async fn discover_leader(&mut self) -> Result<(), IoError> {
-        for node in self.topology.node_infos.iter().map(|n| n.peer_id.clone()).into_iter() {
+        for node in self.topology.node_infos.iter().map(|n| n.peer_id.clone()) {
             tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
             println!("Trying to connect to node: {node}...");
 

duva-client/src/controller.rs

@@ -42,6 +42,7 @@ impl<T> ClientController<T> {
         match kind {
             | Ping
             | Get { .. }
+            | LIndex { .. }
             | IndexGet { .. }
             | Echo { .. }
             | Config { .. }

duva/src/domains/caches/actor.rs

@@ -206,6 +206,20 @@ impl CacheActor {
             | _ => Err(anyhow::anyhow!(WRONG_TYPE_ERR_MSG)),
         }
     }
+
+    pub(crate) fn lindex(&mut self, key: String, index: isize) -> anyhow::Result<CacheValue> {
+        let Some(CacheValue { value, .. }) = self.cache.get_mut(&key) else {
+            return Ok(CacheValue::new(TypedValue::Null));
+        };
+        match value {
+            | TypedValue::List(list) => Ok(list
+                .lindex(index)
+                .map(|v| CacheValue::new(TypedValue::String(v)))
+                .unwrap_or_default()),
+
+            | _ => Err(anyhow::anyhow!(WRONG_TYPE_ERR_MSG)),
+        }
+    }
 }
 
 #[derive(Clone, Debug)]

duva/src/domains/caches/cache_manager.rs

@@ -413,6 +413,15 @@ impl CacheManager {
 
         Ok(IndexedValueCodec::encode("".to_string(), current_idx))
     }
+
+    pub(crate) async fn route_lindex(&self, key: String, index: isize) -> Result<CacheValue> {
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        self.select_shard(&key)
+            .send(CacheCommand::LIndex { key, index, callback: tx.into() })
+            .await?;
+        let value = rx.await??;
+        Ok(value)
+    }
 }
 
 pub struct IndexedValueCodec;

duva/src/domains/caches/cache_objects/types/quicklist.rs

@@ -294,11 +294,11 @@ impl QuickList {
         }
 
         let max_bytes = kb * 1024;
+        let merge_threshold = max_bytes / 4;
 
         let should_merge = {
             let node = &self.nodes[index];
             let next = &self.nodes[index + 1];
-            let merge_threshold = max_bytes / 4;
 
             (node.entry_count > 0 && next.entry_count > 0)
                 && (node.byte_size() < merge_threshold || next.byte_size() < merge_threshold)
@@ -408,10 +408,13 @@ impl QuickList {
     pub fn compress(&mut self) {
         let node_count = self.nodes.len();
         if self.compress_depth > 0 && node_count > self.compress_depth * 2 {
-            for i in self.compress_depth..(node_count - self.compress_depth) {
-                if let Some(node) = self.nodes.get_mut(i) {
-                    node.try_compress();
-                }
+            for node in self
+                .nodes
+                .iter_mut()
+                .skip(self.compress_depth)
+                .take(node_count - self.compress_depth * 2)
+            {
+                node.try_compress();
             }
         }
     }
@@ -502,6 +505,32 @@ impl QuickList {
             self.rpush(element);
         }
     }
+
+    pub(crate) fn lindex(&mut self, index: isize) -> Option<Bytes> {
+        if self.len == 0 {
+            return None;
+        }
+
+        // Calculate absolute index
+        let len = self.len as isize;
+        let index = if index < 0 { (len + index).max(0) } else { index } as usize;
+
+        if index >= self.len {
+            return None; // Out of bounds
+        }
+
+        let mut current_index = 0;
+        for node in &mut self.nodes {
+            if current_index + node.entry_count > index {
+                node.ensure_decompressed(&self.fill_factor);
+                if let NodeData::Uncompressed(ziplist) = &node.data {
+                    return ziplist.to_vec().get(index - current_index).cloned();
+                }
+            }
+            current_index += node.entry_count;
+        }
+        None
+    }
 }
 
 #[cfg(test)]

duva/src/domains/caches/command.rs

@@ -85,4 +85,9 @@ pub(crate) enum CacheCommand {
         end: isize,
         callback: Callback<anyhow::Result<()>>,
     },
+    LIndex {
+        key: String,
+        index: isize,
+        callback: Callback<anyhow::Result<CacheValue>>,
+    },
 }

duva/src/domains/caches/service.rs

@@ -91,6 +91,9 @@ impl CacheActor {
                 | CacheCommand::LTrim { key, start, end, callback } => {
                     let _ = callback.send(self.ltrim(key, start, end));
                 },
+                | CacheCommand::LIndex { key, index, callback } => {
+                    let _ = callback.send(self.lindex(key, index));
+                },
             }
         }
         Ok(self)

duva/src/domains/cluster_actors/topology.rs

@@ -33,7 +33,7 @@ impl NodeReplInfo {
         }
     }
 
-    pub fn from_replication_state(replication_state: &ReplicationState) -> Self {
+    pub(crate) fn from_replication_state(replication_state: &ReplicationState) -> Self {
         Self {
             peer_id: replication_state.self_identifier(),
             repl_id: replication_state.replid.clone(),

duva/src/domains/query_io.rs

@@ -467,9 +467,7 @@ mod test {
     use super::*;
     use crate::domains::caches::cache_objects::CacheEntry;
     use crate::domains::cluster_actors::hash_ring::{BatchId, HashRing};
-    use crate::domains::cluster_actors::replication::{
-        ReplicationId, ReplicationRole, ReplicationState,
-    };
+    use crate::domains::cluster_actors::replication::{ReplicationId, ReplicationRole};
     use crate::domains::cluster_actors::topology::NodeReplInfo;
     use crate::domains::operation_logs::WriteRequest;
     use crate::domains::peers::command::BannedPeer;

duva/src/presentation/clients/controller.rs

@@ -184,8 +184,6 @@ impl ClientController {
             | ClientAction::RPush { key, value } => QueryIO::SimpleString(
                 self.cache_manager.route_rpush(key, value, current_index.unwrap()).await?.into(),
             ),
-
-            //TODO
             | ClientAction::RPushX { key, value } => todo!(),
             | ClientAction::RPop { key, count } => {
                 let values = self.cache_manager.route_rpop(key, count).await?;
@@ -208,6 +206,9 @@ impl ClientController {
                 let values = self.cache_manager.route_lrange(key, start, end).await?;
                 QueryIO::Array(values.into_iter().map(|v| QueryIO::BulkString(v.into())).collect())
             },
+            | ClientAction::LIndex { key, index } => {
+                self.cache_manager.route_lindex(key, index).await?.into()
+            },
         };
 
         Ok(response)