Merge pull request #743 from Migorithm/feat/742

Migorithm · web-flow · commit 64222b16f7ff · 2025-07-15T23:37:57.000+04:00
feat: lset
diff --git a/duva-client/src/controller.rs b/duva-client/src/controller.rs
@@ -96,7 +96,7 @@ impl<T> ClientController<T> {
                 };
                 Response::Null
             },
-            | Set { .. } | SetWithExpiry { .. } | LTrim { .. } => match query_io {
+            | Set { .. } | SetWithExpiry { .. } | LTrim { .. } | LSet { .. } => match query_io {
                 | QueryIO::SimpleString(_) => Response::String("OK".into()),
                 | QueryIO::Err(value) => Response::Error(value),
                 | _ => Response::FormatError,
diff --git a/duva/src/domains/caches/actor.rs b/duva/src/domains/caches/actor.rs
@@ -220,6 +220,22 @@ impl CacheActor {
             | _ => Err(anyhow::anyhow!(WRONG_TYPE_ERR_MSG)),
         }
     }
+
+    pub(crate) fn lset(
+        &mut self,
+        key: String,
+        index: isize,
+        val: String,
+    ) -> Result<(), anyhow::Error> {
+        let Some(var) = self.cache.get_mut(&key) else {
+            return Err(anyhow::anyhow!("ERR no such key"));
+        };
+        let CacheValue { value: TypedValue::List(list), .. } = var else {
+            return Err(anyhow::anyhow!(WRONG_TYPE_ERR_MSG));
+        };
+
+        list.lset(index, val)
+    }
 }
 
 #[derive(Clone, Debug)]
diff --git a/duva/src/domains/caches/cache_manager.rs b/duva/src/domains/caches/cache_manager.rs
@@ -198,6 +198,9 @@ impl CacheManager {
             | WriteRequest::LPushX { key, value } => {
                 self.route_lpushx(key, value, log_index).await?;
             },
+            | WriteRequest::LSet { key, index, value } => {
+                self.route_lset(key, index, value, log_index).await?;
+            },
         };
 
         // * This is to wake up the cache actors to process the pending read requests
@@ -422,6 +425,21 @@ impl CacheManager {
         let value = rx.await??;
         Ok(value)
     }
+
+    pub(crate) async fn route_lset(
+        &self,
+        key: String,
+        index: isize,
+        value: String,
+        current_idx: u64,
+    ) -> Result<String> {
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        self.select_shard(key.as_str())
+            .send(CacheCommand::LSet { key, index, value, callback: tx.into() })
+            .await?;
+        rx.await??;
+        Ok(IndexedValueCodec::encode("", current_idx))
+    }
 }
 
 pub struct IndexedValueCodec;
diff --git a/duva/src/domains/caches/cache_objects/types/quicklist.rs b/duva/src/domains/caches/cache_objects/types/quicklist.rs
@@ -123,7 +123,7 @@ struct QuickListNode {
 
 impl QuickListNode {
     /// Ensures the node is decompressed. Must be called before any read/write.
-    fn ensure_decompressed(&mut self, fill_factor: &FillFactor) {
+    fn ensure_decompressed(&mut self, fill_factor: &FillFactor) -> &mut Ziplist {
         let current_data =
             std::mem::replace(&mut self.data, NodeData::Uncompressed(Ziplist::default()));
         self.data = match current_data {
@@ -138,6 +138,9 @@ impl QuickListNode {
             },
             | uncompressed => uncompressed,
         };
+
+        let NodeData::Uncompressed(ziplist) = &mut self.data else { panic!() };
+        ziplist
     }
 
     /// Attempts to compress the node.
@@ -308,21 +311,13 @@ impl QuickList {
         if should_merge {
             // First, remove the next node from the list.
             let mut removed_node = self.nodes.remove(index + 1).unwrap();
-            removed_node.ensure_decompressed(&self.fill_factor);
+            let removed_ziplist = removed_node.ensure_decompressed(&self.fill_factor);
 
             // Now, get mutable access to the current node to merge into.
             let current_node = &mut self.nodes[index];
-            current_node.ensure_decompressed(&self.fill_factor);
 
-            // Use pattern matching to get the ziplists and perform the merge.
-            if let (
-                NodeData::Uncompressed(current_ziplist),
-                NodeData::Uncompressed(removed_ziplist),
-            ) = (&mut current_node.data, &mut removed_node.data)
-            {
-                current_ziplist.extend_from_slice(removed_ziplist);
-                current_node.entry_count += removed_node.entry_count;
-            }
+            current_node.ensure_decompressed(&self.fill_factor).extend_from_slice(removed_ziplist);
+            current_node.entry_count += removed_node.entry_count;
 
             self.return_node(removed_node);
         }
@@ -451,21 +446,21 @@ impl QuickList {
             }
 
             // 3. Decompress the node and read its entries
-            node.ensure_decompressed(&self.fill_factor);
-            if let NodeData::Uncompressed(ziplist) = &node.data {
-                let entries = ziplist.to_vec();
-                for (i, entry) in entries.iter().enumerate() {
-                    let overall_index = current_index + i;
-                    if overall_index >= start && overall_index <= stop {
-                        // Clone the Bytes object and add it to our result.
-                        // Cloning is cheap (reference counted).
-                        result.push(entry.clone());
-                    }
-                    if overall_index >= stop {
-                        break;
-                    }
+            let ziplist = node.ensure_decompressed(&self.fill_factor);
+
+            let entries = ziplist.to_vec();
+            for (i, entry) in entries.iter().enumerate() {
+                let overall_index = current_index + i;
+                if overall_index >= start && overall_index <= stop {
+                    // Clone the Bytes object and add it to our result.
+                    // Cloning is cheap (reference counted).
+                    result.push(entry.clone());
+                }
+                if overall_index >= stop {
+                    break;
                 }
             }
+
             current_index += node_len;
         }
         result
@@ -522,15 +517,71 @@ impl QuickList {
         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();
-                }
+                let ziplist = node.ensure_decompressed(&self.fill_factor);
+                return ziplist.to_vec().get(index - current_index).cloned();
             }
             current_index += node.entry_count;
         }
         None
     }
+
+    pub(crate) fn lset(&mut self, index: isize, value: String) -> anyhow::Result<()> {
+        if self.len == 0 {
+            return Err(anyhow::anyhow!("List is empty"));
+        }
+
+        // Convert to absolute index
+        let len = self.len as isize;
+        let abs_index = if index < 0 { (len + index).max(0) } else { index } as usize;
+
+        if abs_index >= self.len {
+            return Err(anyhow::anyhow!("Index out of bounds"));
+        }
+
+        // Find the node containing this index
+        let mut current_index = 0;
+
+        for node in &mut self.nodes {
+            if current_index + node.entry_count <= abs_index {
+                current_index += node.entry_count;
+                continue;
+            }
+
+            let local_index = abs_index - current_index;
+
+            let ziplist = node.ensure_decompressed(&self.fill_factor);
+            let new_bytes = Bytes::from(value);
+            let mut cursor = 0;
+
+            // Skip to the target entry
+            for _ in 0..local_index {
+                let entry_len =
+                    u32::from_le_bytes(ziplist[cursor..cursor + 4].try_into().unwrap()) as usize;
+                cursor += 4 + entry_len;
+            }
+
+            let old_entry_len =
+                u32::from_le_bytes(ziplist[cursor..cursor + 4].try_into().unwrap()) as usize;
+
+            if new_bytes.len() == old_entry_len {
+                // Same size - overwrite in place
+                ziplist[cursor + 4..cursor + 4 + old_entry_len].copy_from_slice(&new_bytes);
+            } else {
+                // Different size - rebuild ziplist
+                let mut new_ziplist =
+                    Vec::with_capacity(ziplist.len() - old_entry_len + new_bytes.len());
+                new_ziplist.extend_from_slice(&ziplist[0..cursor]);
+                new_ziplist.extend_from_slice(&(new_bytes.len() as u32).to_le_bytes());
+                new_ziplist.extend_from_slice(&new_bytes);
+                new_ziplist.extend_from_slice(&ziplist[cursor + 4 + old_entry_len..]);
+                ziplist.0 = new_ziplist;
+            }
+
+            return Ok(());
+        }
+
+        Err(anyhow::anyhow!("Index not found"))
+    }
 }
 
 #[cfg(test)]
diff --git a/duva/src/domains/caches/command.rs b/duva/src/domains/caches/command.rs
@@ -90,4 +90,10 @@ pub(crate) enum CacheCommand {
         index: isize,
         callback: Callback<anyhow::Result<CacheValue>>,
     },
+    LSet {
+        key: String,
+        index: isize,
+        value: String,
+        callback: Callback<anyhow::Result<()>>,
+    },
 }
diff --git a/duva/src/domains/caches/service.rs b/duva/src/domains/caches/service.rs
@@ -94,6 +94,9 @@ impl CacheActor {
                 | CacheCommand::LIndex { key, index, callback } => {
                     let _ = callback.send(self.lindex(key, index));
                 },
+                | CacheCommand::LSet { key, index, value, callback } => {
+                    let _ = callback.send(self.lset(key, index, value));
+                },
             }
         }
         Ok(self)
diff --git a/duva/src/domains/operation_logs/operation.rs b/duva/src/domains/operation_logs/operation.rs
@@ -26,6 +26,7 @@ pub enum WriteRequest {
     RPush { key: String, value: Vec<String> },
     LTrim { key: String, start: isize, end: isize },
     LPushX { key: String, value: Vec<String> },
+    LSet { key: String, index: isize, value: String },
 }
 
 impl WriteOperation {
@@ -66,6 +67,7 @@ impl WriteRequest {
             | WriteRequest::RPush { key, .. } => vec![key],
             | WriteRequest::LTrim { key, .. } => vec![key],
             | WriteRequest::LPushX { key, .. } => vec![key],
+            | WriteRequest::LSet { key, .. } => vec![key],
         }
     }
 }
diff --git a/duva/src/presentation/clients/controller.rs b/duva/src/presentation/clients/controller.rs
@@ -209,6 +209,12 @@ impl ClientController {
             | ClientAction::LIndex { key, index } => {
                 self.cache_manager.route_lindex(key, index).await?.into()
             },
+            | ClientAction::LSet { key, index, value } => QueryIO::SimpleString(
+                self.cache_manager
+                    .route_lset(key, index, value, current_index.unwrap())
+                    .await?
+                    .into(),
+            ),
         };
 
         Ok(response)
diff --git a/duva/src/presentation/clients/request.rs b/duva/src/presentation/clients/request.rs
@@ -43,6 +43,7 @@ pub enum ClientAction {
     RPush { key: String, value: Vec<String> },
     RPushX { key: String, value: Vec<String> },
     RPop { key: String, count: usize },
+    LSet { key: String, index: isize, value: String },
 
     LTrim { key: String, start: isize, end: isize },
     LLen { key: String },
@@ -78,7 +79,7 @@ impl ClientAction {
             | ClientAction::RPushX { key, value } => WriteRequest::RPush { key, value },
             | ClientAction::RPop { key, count } => WriteRequest::LPop { key, count },
             | ClientAction::LTrim { key, start, end } => WriteRequest::LTrim { key, start, end },
-
+            | ClientAction::LSet { key, index, value } => WriteRequest::LSet { key, index, value },
             | _ => {
                 debug_assert!(false, "to_write_request called on non-write action: {self:?}");
                 unreachable!(
@@ -107,6 +108,7 @@ impl ClientAction {
                 | ClientAction::RPushX { .. }
                 | ClientAction::RPop { .. }
                 | ClientAction::LTrim { .. }
+                | ClientAction::LSet { .. }
         )
     }
 }
@@ -385,6 +387,13 @@ pub fn extract_action(action: &str, args: &[&str]) -> anyhow::Result<ClientActio
             let index = args[1].parse::<isize>().context("Invalid index")?;
             Ok(ClientAction::LIndex { key, index })
         },
+        | "LSET" => {
+            require_exact_args(3)?;
+            let key = args[0].to_string();
+            let index = args[1].parse::<isize>().context("Invalid index")?;
+            let value = args[2].to_string();
+            Ok(ClientAction::LSet { key, index, value })
+        },
 
         // Add other commands as needed
         | unknown_cmd => Err(anyhow::anyhow!(
diff --git a/duva/tests/client_ops/mod.rs b/duva/tests/client_ops/mod.rs
@@ -14,6 +14,7 @@ mod test_lpop;
 mod test_lpush;
 mod test_lpushx;
 mod test_lrange;
+mod test_lset;
 mod test_ltrim;
 mod test_replication_info;
 mod test_rpop;
diff --git a/duva/tests/client_ops/test_lset.rs b/duva/tests/client_ops/test_lset.rs
@@ -0,0 +1,40 @@
+use crate::common::{Client, ServerEnv, spawn_server_process};
+
+fn run_lset(env: ServerEnv) -> anyhow::Result<()> {
+    // GIVEN
+    let process = spawn_server_process(&env)?;
+
+    let mut h = Client::new(process.port);
+    let res = h.send_and_get(format!("RPUSH test one two three"));
+    assert_eq!(res, "(integer) 3");
+
+    //WHEN & ASSERT
+    assert_eq!(h.send_and_get(format!("LSET test 0 four")), "OK");
+    assert_eq!(h.send_and_get(format!("LSET test -2 five")), "OK");
+    assert_eq!(
+        h.send_and_get_vec(format!("LRANGE test 0 -1"), 3),
+        vec!["1) \"four\"", "2) \"five\"", "3) \"three\""]
+    );
+
+    // ERROR CASE
+
+    assert_eq!(h.send_and_get(format!("LSET test 10 error")), "(error) Index out of bounds");
+    assert_eq!(h.send_and_get(format!("LSET x 1 2")), "(error) ERR no such key");
+
+    assert_eq!(h.send_and_get(format!("SET x 1")), "OK");
+    assert_eq!(
+        h.send_and_get(format!("LSET x 0 2")),
+        "(error) WRONGTYPE Operation against a key holding the wrong kind of value"
+    );
+
+    Ok(())
+}
+
+#[test]
+fn test_lset() -> anyhow::Result<()> {
+    for env in [ServerEnv::default(), ServerEnv::default().with_append_only(true)] {
+        run_lset(env)?;
+    }
+
+    Ok(())
+}

Original file line number	Diff line number	Diff line change
`@@ -94,6 +94,9 @@ impl CacheActor {`
`94`	`94`	`\| CacheCommand::LIndex { key, index, callback } => {`
`95`	`95`	`let _ = callback.send(self.lindex(key, index));`
`96`	`96`	`},`
	`97`	`+ \| CacheCommand::LSet { key, index, value, callback } => {`
	`98`	`+ let _ = callback.send(self.lset(key, index, value));`
	`99`	`+ },`
`97`	`100`	`}`
`98`	`101`	`}`
`99`	`102`	`Ok(self)`
Original file line number	Diff line number	Diff line change
`@@ -26,6 +26,7 @@ pub enum WriteRequest {`
`26`	`26`	`RPush { key: String, value: Vec<String> },`
`27`	`27`	`LTrim { key: String, start: isize, end: isize },`
`28`	`28`	`LPushX { key: String, value: Vec<String> },`
	`29`	`+ LSet { key: String, index: isize, value: String },`
`29`	`30`	`}`
`30`	`31`
`31`	`32`	`impl WriteOperation {`
`@@ -66,6 +67,7 @@ impl WriteRequest {`
`66`	`67`	`\| WriteRequest::RPush { key, .. } => vec![key],`
`67`	`68`	`\| WriteRequest::LTrim { key, .. } => vec![key],`
`68`	`69`	`\| WriteRequest::LPushX { key, .. } => vec![key],`
	`70`	`+ \| WriteRequest::LSet { key, .. } => vec![key],`
`69`	`71`	`}`
`70`	`72`	`}`
`71`	`73`	`}`