fix(storage): align DEL with generation lifecycle

Author: happy-v587

src/storage/src/redis_strings.rs

@@ -1993,7 +1993,20 @@ impl Redis {
             });
         };
 
-        if value.is_empty() || self.is_stale(&value)? {
+        let is_live = if value.is_empty() {
+            false
+        } else {
+            match DataType::try_from(value[0])? {
+                DataType::String => !ParsedStringsValue::new(&value[..])?.is_stale(),
+                DataType::Hash | DataType::Set | DataType::ZSet => {
+                    ParsedBaseMetaValue::new(&value[..])?.is_valid()
+                }
+                DataType::List => ParsedListsMetaValue::new(&value[..])?.is_valid(),
+                _ => false,
+            }
+        };
+
+        if !is_live {
             return Err(crate::error::Error::KeyNotFound {
                 key: String::from_utf8_lossy(key).to_string(),
                 location: snafu::location!(),
@@ -2005,9 +2018,6 @@ impl Redis {
 
     /// Delete a key (works for all data types)
     pub fn del_key(&self, key: &[u8]) -> Result<bool> {
-        use crate::storage_define::{PREFIX_RESERVE_LENGTH, encode_user_key};
-        use bytes::BufMut;
-
         let db = self.db.as_ref().context(OptionNoneSnafu {
             message: "db is not initialized".to_string(),
         })?;
@@ -2019,77 +2029,68 @@ impl Redis {
                 message: "MetaCF is not initialized".to_string(),
             })?;
 
-        // Check if key exists in MetaCF (where all metadata is stored)
-        let string_key = BaseKey::new(key);
+        let key_str = String::from_utf8_lossy(key).to_string();
+        let _lock = ScopeRecordLock::new(self.lock_mgr.as_ref(), &key_str);
         let meta_key = BaseMetaKey::new(key);
-
-        let string_existed = db
-            .get_cf_opt(&meta_cf, &string_key.encode()?, &self.read_options)
-            .context(RocksSnafu)?
-            .is_some();
-        let meta_existed = db
-            .get_cf_opt(&meta_cf, &meta_key.encode()?, &self.read_options)
+        let encoded_meta_key = meta_key.encode()?;
+        let Some(value) = db
+            .get_cf_opt(&meta_cf, &encoded_meta_key, &self.read_options)
             .context(RocksSnafu)?
-            .is_some();
-
-        if string_existed || meta_existed {
-            let encoded = string_key.encode()?;
-
-            // Build correct prefix for data CF scanning:
-            // Data keys format: | reserve1 (8B) | encoded_user_key | version (8B) | data | reserve2 |
-            // We need prefix: | reserve1 (8B) | encoded_user_key (with \x00\x00 delimiter) |
-            // Note: BaseKey.encode() includes reserve2, which would not match data keys.
-            let mut data_key_prefix =
-                bytes::BytesMut::with_capacity(PREFIX_RESERVE_LENGTH + key.len() * 2 + 2);
-            data_key_prefix.put_slice(&[0u8; PREFIX_RESERVE_LENGTH]);
-            encode_user_key(key, &mut data_key_prefix)?;
-
-            // Collect all keys to delete first (to avoid borrow conflicts)
-            let mut keys_to_delete: Vec<(ColumnFamilyIndex, Vec<u8>)> = Vec::new();
-
-            // Delete from MetaCF
-            if string_existed {
-                keys_to_delete.push((ColumnFamilyIndex::MetaCF, encoded.to_vec()));
-            }
-            if meta_existed {
-                keys_to_delete.push((ColumnFamilyIndex::MetaCF, meta_key.encode()?.to_vec()));
-            }
+        else {
+            return Ok(false);
+        };
+        if value.is_empty() {
+            return Ok(false);
+        }
 
-            // For composite data types, perform prefix scan to delete all related entries
-            for cf_index in [
-                ColumnFamilyIndex::HashesDataCF,
-                ColumnFamilyIndex::SetsDataCF,
-                ColumnFamilyIndex::ListsDataCF,
-                ColumnFamilyIndex::ZsetsDataCF,
-                ColumnFamilyIndex::ZsetsScoreCF,
-            ] {
-                let cf_handle = self.get_cf_handle(cf_index);
-                if let Some(cf) = cf_handle {
-                    // Prefix-scan data CF and delete all derived keys
-                    let iter = db.iterator_cf(
-                        &cf,
-                        rocksdb::IteratorMode::From(&data_key_prefix, rocksdb::Direction::Forward),
-                    );
-                    for item in iter {
-                        let (k, _) = item.context(RocksSnafu)?;
-                        if !k.starts_with(&data_key_prefix) {
-                            break;
-                        }
-                        keys_to_delete.push((cf_index, k.to_vec()));
-                    }
+        match DataType::try_from(value[0])? {
+            DataType::String => {
+                let parsed = ParsedStringsValue::new(&value[..])?;
+                if parsed.is_stale() {
+                    return Ok(false);
                 }
+                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)?;
             }
-
-            // Now create batch and delete all collected keys
-            let mut batch = self.create_batch()?;
-            for (cf_idx, key) in keys_to_delete {
-                batch.delete(cf_idx, &key)?;
+            DataType::Hash | DataType::Set | DataType::ZSet => {
+                let mut parsed = ParsedBaseMetaValue::new(&value[..])?;
+                if !parsed.is_valid() {
+                    return Ok(false);
+                }
+                let count = parsed.count();
+                parsed.set_count(0);
+                parsed.update_version();
+                parsed.set_etime(0);
+                let data_type = DataType::try_from(value[0])?;
+                let mut batch = self.create_batch()?;
+                batch.put(
+                    ColumnFamilyIndex::MetaCF,
+                    &encoded_meta_key,
+                    parsed.encoded(),
+                )?;
+                batch.commit()?;
+                self.update_specific_key_statistics(data_type, &key_str, count)?;
             }
-            batch.commit()?;
-            Ok(true)
-        } else {
-            Ok(false)
+            DataType::List => {
+                let mut parsed = ParsedListsMetaValue::new(&value[..])?;
+                if !parsed.is_valid() {
+                    return Ok(false);
+                }
+                let count = parsed.count();
+                parsed.set_count(0);
+                parsed.update_version();
+                parsed.set_etime(0);
+                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)?;
+            }
+            _ => return Ok(false),
         }
+
+        Ok(true)
     }
 
     /// Scan for keys matching a pattern

src/storage/tests/ttl_test.rs

@@ -20,7 +20,10 @@
 use std::sync::Arc;
 
 use storage::ZsetScoreMember;
-use storage::{StorageOptions, storage::Storage, unique_test_db_path};
+use storage::{
+    BaseMetaKey, ColumnFamilyIndex, StorageOptions, format_base_meta_value::ParsedBaseMetaValue,
+    storage::Storage, unique_test_db_path,
+};
 
 #[tokio::test]
 async fn test_ttl_basic_operations() {
@@ -337,6 +340,78 @@ async fn test_del_command() {
     storage.shutdown().await;
 }
 
+#[tokio::test]
+async fn test_del_returns_zero_for_stale_string() {
+    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"stale_del_key";
+    storage.set(key, b"value").unwrap();
+    assert!(storage.pexpire(key, 1).unwrap());
+    tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+    assert_eq!(storage.del(&[key.to_vec()]).unwrap(), 0);
+
+    storage.shutdown().await;
+}
+
+#[tokio::test]
+async fn test_del_composite_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_hash";
+    storage.hset(key, b"old-field", b"old-value").unwrap();
+    assert_eq!(storage.del(&[key.to_vec()]).unwrap(), 1);
+    assert_eq!(storage.exists(&[key.to_vec()]).unwrap(), 0);
+    assert_eq!(storage.key_type(key).unwrap(), "none");
+    assert_eq!(storage.ttl(key).unwrap(), -2);
+    assert!(storage.keys(b"del_recreate_hash").unwrap().is_empty());
+    storage.hset(key, b"new-field", b"new-value").unwrap();
+    assert_eq!(storage.hget(key, b"old-field").unwrap(), None);
+    assert_eq!(
+        storage.hget(key, b"new-field").unwrap(),
+        Some("new-value".to_string())
+    );
+
+    storage.shutdown().await;
+}
+
+#[tokio::test]
+async fn test_del_writes_composite_meta_tombstone_with_new_version() {
+    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_meta_tombstone";
+    storage.hset(key, b"field", b"value").unwrap();
+    let meta_key = BaseMetaKey::new(key).encode().unwrap();
+    let original_version = {
+        let instance = &storage.insts[0];
+        let meta_cf = instance.get_cf_handle(ColumnFamilyIndex::MetaCF).unwrap();
+        let db = instance.db().unwrap();
+        let original = db.get_cf(&meta_cf, &meta_key).unwrap().unwrap();
+        ParsedBaseMetaValue::new(&original[..]).unwrap().version()
+    };
+
+    assert_eq!(storage.del(&[key.to_vec()]).unwrap(), 1);
+    {
+        let instance = &storage.insts[0];
+        let meta_cf = instance.get_cf_handle(ColumnFamilyIndex::MetaCF).unwrap();
+        let db = instance.db().unwrap();
+        let tombstone = db.get_cf(&meta_cf, &meta_key).unwrap().unwrap();
+        let parsed = ParsedBaseMetaValue::new(&tombstone[..]).unwrap();
+        assert_eq!(parsed.count(), 0);
+        assert!(parsed.version() > original_version);
+    }
+
+    storage.shutdown().await;
+}
+
 #[tokio::test]
 async fn test_keys_command() {
     let db_path = unique_test_db_path();