fix(storage): address DEL review feedback

Author: happy-v587

src/storage/src/redis_strings.rs

@@ -2052,7 +2052,11 @@ impl Redis {
                 let mut batch = self.create_batch()?;
                 batch.delete(ColumnFamilyIndex::MetaCF, &encoded_meta_key)?;
                 batch.commit()?;
-                self.update_specific_key_statistics(DataType::String, &key_str, 1)?;
+                if let Err(error) =
+                    self.update_specific_key_statistics(DataType::String, &key_str, 1)
+                {
+                    log::warn!("failed to update key statistics for {key_str}: {error:?}");
+                }
             }
             DataType::Hash | DataType::Set | DataType::ZSet => {
                 let mut parsed = ParsedBaseMetaValue::new(&value[..])?;
@@ -2071,7 +2075,10 @@ impl Redis {
                     parsed.encoded(),
                 )?;
                 batch.commit()?;
-                self.update_specific_key_statistics(data_type, &key_str, count)?;
+                if let Err(error) = self.update_specific_key_statistics(data_type, &key_str, count)
+                {
+                    log::warn!("failed to update key statistics for {key_str}: {error:?}");
+                }
             }
             DataType::List => {
                 let mut parsed = ParsedListsMetaValue::new(&value[..])?;
@@ -2085,7 +2092,11 @@ impl Redis {
                 let mut batch = self.create_batch()?;
                 batch.put(ColumnFamilyIndex::MetaCF, &encoded_meta_key, parsed.value())?;
                 batch.commit()?;
-                self.update_specific_key_statistics(DataType::List, &key_str, count)?;
+                if let Err(error) =
+                    self.update_specific_key_statistics(DataType::List, &key_str, count)
+                {
+                    log::warn!("failed to update key statistics for {key_str}: {error:?}");
+                }
             }
             _ => return Ok(false),
         }

src/storage/tests/ttl_test.rs

@@ -380,6 +380,47 @@ async fn test_del_composite_recreate_hides_old_members() {
     storage.shutdown().await;
 }
 
+#[tokio::test]
+async fn test_del_set_recreate_hides_old_members() {
+    let db_path = unique_test_db_path();
+    let mut storage = Storage::new(1, 0);
+    let options = Arc::new(StorageOptions::default());
+    let _receiver = storage.open(options, &db_path).unwrap();
+
+    let key = b"del_recreate_set";
+    assert_eq!(storage.sadd(key, &[b"old-member"]).unwrap(), 1);
+    assert_eq!(storage.del(&[key.to_vec()]).unwrap(), 1);
+    assert_eq!(storage.sadd(key, &[b"new-member"]).unwrap(), 1);
+
+    assert_eq!(
+        storage.smembers(key).unwrap(),
+        vec!["new-member".to_string()]
+    );
+    assert!(!storage.sismember(key, b"old-member").unwrap());
+
+    storage.shutdown().await;
+}
+
+#[tokio::test]
+async fn test_del_list_recreate_hides_old_members() {
+    let db_path = unique_test_db_path();
+    let mut storage = Storage::new(1, 0);
+    let options = Arc::new(StorageOptions::default());
+    let _receiver = storage.open(options, &db_path).unwrap();
+
+    let key = b"del_recreate_list";
+    assert_eq!(storage.rpush(key, &[b"old-member".to_vec()]).unwrap(), 1);
+    assert_eq!(storage.del(&[key.to_vec()]).unwrap(), 1);
+    assert_eq!(storage.rpush(key, &[b"new-member".to_vec()]).unwrap(), 1);
+
+    assert_eq!(
+        storage.lrange(key, 0, -1).unwrap(),
+        vec![b"new-member".to_vec()]
+    );
+
+    storage.shutdown().await;
+}
+
 #[tokio::test]
 async fn test_del_writes_composite_meta_tombstone_with_new_version() {
     let db_path = unique_test_db_path();
@@ -406,6 +447,7 @@ async fn test_del_writes_composite_meta_tombstone_with_new_version() {
         let tombstone = db.get_cf(&meta_cf, &meta_key).unwrap().unwrap();
         let parsed = ParsedBaseMetaValue::new(&tombstone[..]).unwrap();
         assert_eq!(parsed.count(), 0);
+        assert_eq!(parsed.etime(), 0);
         assert!(parsed.version() > original_version);
     }