Merge pull request #1 from happy-v587/codex/compact-del-validation

Codex/compact del validation

Author: happy-v587

src/storage/src/data_compaction_filter.rs

@@ -48,6 +48,7 @@ const DATA_FILTER_FACTORY_NAME: &std::ffi::CStr = c"DataCompactionFilterFactory"
 struct CompactionFilterTestState {
     entered: bool,
     released: bool,
+    removed_keys: Vec<Vec<u8>>,
 }
 
 #[cfg(test)]
@@ -88,6 +89,27 @@ impl CompactionFilterTestGate {
         self.changed.notify_all();
     }
 
+    pub(crate) fn wait_until_removed(&self, timeout: std::time::Duration) -> Vec<Vec<u8>> {
+        let state = self
+            .state
+            .lock()
+            .expect("compaction filter test gate mutex should not be poisoned");
+        let (state, _) = self
+            .changed
+            .wait_timeout_while(state, timeout, |state| state.removed_keys.is_empty())
+            .expect("compaction filter test gate wait should succeed");
+        state.removed_keys.clone()
+    }
+
+    fn record_removed_key(&self, key: &[u8]) {
+        let mut state = self
+            .state
+            .lock()
+            .expect("compaction filter test gate mutex should not be poisoned");
+        state.removed_keys.push(key.to_vec());
+        self.changed.notify_all();
+    }
+
     fn enter_and_wait(&self, key: &[u8]) {
         if !key.starts_with(&self.key_prefix) {
             return;
@@ -161,6 +183,18 @@ fn block_once_for_compaction_filter_test(key: &[u8]) {
     }
 }
 
+#[cfg(test)]
+fn record_compaction_filter_remove(key: &[u8]) {
+    let gate = compaction_filter_test_gate()
+        .lock()
+        .expect("compaction filter test gate registry should not be poisoned")
+        .as_ref()
+        .and_then(Weak::upgrade);
+    if let Some(gate) = gate {
+        gate.record_removed_key(key);
+    }
+}
+
 #[derive(Debug)]
 enum MetaLookup {
     Valid,
@@ -326,7 +360,7 @@ impl CompactionFilter for DataCompactionFilter {
             return CompactionDecision::Keep;
         };
 
-        match self.ensure_meta_state(&meta_key) {
+        let decision = match self.ensure_meta_state(&meta_key) {
             MetaLookup::Unavailable => CompactionDecision::Keep,
             MetaLookup::NotFound => CompactionDecision::Remove,
             MetaLookup::Valid => {
@@ -340,7 +374,14 @@ impl CompactionFilter for DataCompactionFilter {
                     _ => CompactionDecision::Keep,
                 }
             }
+        };
+
+        #[cfg(test)]
+        if let CompactionDecision::Remove = &decision {
+            record_compaction_filter_remove(key);
         }
+
+        decision
     }
 }
 

src/storage/src/redis.rs

@@ -1010,6 +1010,7 @@ macro_rules! get_db_and_cfs {
 #[cfg(test)]
 mod lifecycle_tests {
     use std::sync::{Arc, mpsc};
+    use std::thread;
     use std::time::Duration;
 
     use super::{
@@ -1018,8 +1019,11 @@ mod lifecycle_tests {
     use crate::data_compaction_filter::{
         CompactionFilterTestGate, install_compaction_filter_test_gate,
     };
+    use crate::format_base_key::BaseMetaKey;
+    use crate::format_base_meta_value::ParsedSetsMetaValue;
     use crate::{BgTaskHandler, StorageOptions, safe_cleanup_test_db, unique_test_db_path};
     use kstd::lock_mgr::LockMgr;
+    use rocksdb::IteratorMode;
 
     fn open_compaction_test_redis(path: &std::path::Path) -> Redis {
         let mut storage_options = StorageOptions::default();
@@ -1043,6 +1047,94 @@ mod lifecycle_tests {
         redis
     }
 
+    #[test]
+    fn del_compaction_removes_old_generation_and_preserves_recreated_set() {
+        let path = unique_test_db_path();
+        safe_cleanup_test_db(&path);
+
+        let mut storage_options = StorageOptions::default();
+        storage_options.options.set_disable_auto_compactions(true);
+        let (bg_task_handler, _) = BgTaskHandler::new();
+        let mut redis = Redis::new(
+            Arc::new(storage_options),
+            0,
+            Arc::new(bg_task_handler),
+            Arc::new(kstd::lock_mgr::LockMgr::new(64)),
+        );
+        redis
+            .open(path.to_str().expect("test DB path should be valid UTF-8"))
+            .expect("compaction test Redis should open");
+        let redis = Arc::new(redis);
+
+        let key = b"del_compaction_set";
+        redis
+            .sadd(key, &[b"old-member"])
+            .expect("initial set write should succeed");
+        let db = redis.db().expect("Redis should own RocksDB");
+        db.flush().expect("initial data should be flushed");
+
+        let meta_key = BaseMetaKey::new(key).encode().unwrap();
+        let meta_cf = redis.get_cf_handle(ColumnFamilyIndex::MetaCF).unwrap();
+        let original_meta = db.get_cf(&meta_cf, &meta_key).unwrap().unwrap();
+        let original_version = ParsedSetsMetaValue::new(&original_meta[..])
+            .unwrap()
+            .version();
+        let data_cf = redis.get_cf_handle(ColumnFamilyIndex::SetsDataCF).unwrap();
+        let old_data_keys: Vec<Vec<u8>> = db
+            .iterator_cf(&data_cf, IteratorMode::Start)
+            .map(|entry| entry.unwrap().0.to_vec())
+            .collect();
+        assert_eq!(old_data_keys.len(), 1);
+
+        assert!(redis.del_key(key).unwrap());
+        db.flush().expect("tombstone should be flushed");
+        let tombstone = db.get_cf(&meta_cf, &meta_key).unwrap().unwrap();
+        let parsed_tombstone = ParsedSetsMetaValue::new(&tombstone[..]).unwrap();
+        assert_eq!(parsed_tombstone.count(), 0);
+        assert_eq!(parsed_tombstone.etime(), 0);
+        assert!(parsed_tombstone.version() > original_version);
+
+        let gate = CompactionFilterTestGate::new(&[]);
+        let _gate_guard = install_compaction_filter_test_gate(Arc::clone(&gate));
+        let compactor = Arc::clone(&redis);
+        let compaction_thread = thread::spawn(move || {
+            compactor
+                .compact_range(None, None)
+                .expect("manual compaction should succeed");
+        });
+        assert!(gate.wait_until_entered(Duration::from_secs(10)));
+        gate.release();
+        compaction_thread.join().unwrap();
+
+        let removed_keys = gate.wait_until_removed(Duration::from_secs(10));
+        assert!(
+            old_data_keys
+                .iter()
+                .any(|old_key| removed_keys.contains(old_key)),
+            "DataCompactionFilter did not remove the old generation key"
+        );
+        let remaining_data_keys: Vec<Vec<u8>> = db
+            .iterator_cf(&data_cf, IteratorMode::Start)
+            .map(|entry| entry.unwrap().0.to_vec())
+            .collect();
+        assert!(
+            old_data_keys
+                .iter()
+                .all(|old_key| !remaining_data_keys.contains(old_key)),
+            "old generation key survived compaction"
+        );
+
+        redis
+            .sadd(key, &[b"new-member"])
+            .expect("recreated set write should succeed");
+        assert_eq!(redis.smembers(key).unwrap(), vec!["new-member".to_string()]);
+
+        drop(data_cf);
+        drop(meta_cf);
+        drop(redis);
+        safe_cleanup_test_db(&path);
+    }
+
     #[test]
     fn dropping_last_owner_waits_for_active_compaction_filter_before_reopen() {
         let path = unique_test_db_path();

src/storage/src/storage.rs

@@ -402,8 +402,15 @@ impl Storage {
         Ok(())
     }
 
-    fn do_compact_range(&self, _dtype: DataType, _start: &str, _end: &str) -> Result<()> {
-        log::info!("do_compact_range {_dtype:?} {_start} {_end}");
+    fn do_compact_range(&self, dtype: DataType, start: &str, end: &str) -> Result<()> {
+        log::info!("do_compact_range {dtype:?} {start} {end}");
+
+        let begin = (!start.is_empty()).then_some(start.as_bytes());
+        let end = (!end.is_empty()).then_some(end.as_bytes());
+        for instance in &self.insts {
+            instance.compact_range(begin, end)?;
+        }
+
         Ok(())
     }
 

src/storage/tests/storage_basic_test.rs

@@ -19,8 +19,9 @@
 
 use std::sync::Arc;
 
+use rocksdb::IteratorMode;
 use storage::storage::Storage;
-use storage::{BgTask, BgTaskHandler, DataType, StorageOptions};
+use storage::{BgTask, BgTaskHandler, ColumnFamilyIndex, DataType, StorageOptions};
 
 // This test ensures:
 // - All tasks are sent successfully (no panic)
@@ -69,6 +70,62 @@ async fn test_bg_task_worker_concurrent() {
     worker_handle.await.unwrap();
 }
 
+#[tokio::test]
+async fn test_storage_compact_range_executes_manual_compaction() {
+    let temp_dir = tempfile::tempdir().unwrap();
+    let mut storage_options = StorageOptions::default();
+    storage_options.options.set_disable_auto_compactions(true);
+
+    let mut storage = Storage::new(1, 0);
+    let _receiver = storage
+        .open(Arc::new(storage_options), temp_dir.path())
+        .unwrap();
+
+    storage.sadd(b"compact_range_key", &[b"member"]).unwrap();
+    storage.insts[0].db().unwrap().flush().unwrap();
+
+    let old_data_keys: Vec<Vec<u8>> = {
+        let data_cf = storage.insts[0]
+            .get_cf_handle(ColumnFamilyIndex::SetsDataCF)
+            .unwrap();
+        storage.insts[0]
+            .db()
+            .unwrap()
+            .iterator_cf(&data_cf, IteratorMode::Start)
+            .map(|entry| entry.unwrap().0.to_vec())
+            .collect()
+    };
+    assert_eq!(old_data_keys.len(), 1);
+
+    storage.del(&[b"compact_range_key".to_vec()]).unwrap();
+    storage.insts[0].db().unwrap().flush().unwrap();
+
+    storage
+        .compact_range(DataType::All, "", "", true)
+        .await
+        .unwrap();
+
+    let remaining_data_keys: Vec<Vec<u8>> = {
+        let data_cf = storage.insts[0]
+            .get_cf_handle(ColumnFamilyIndex::SetsDataCF)
+            .unwrap();
+        storage.insts[0]
+            .db()
+            .unwrap()
+            .iterator_cf(&data_cf, IteratorMode::Start)
+            .map(|entry| entry.unwrap().0.to_vec())
+            .collect()
+    };
+    assert!(
+        old_data_keys
+            .iter()
+            .all(|old_key| !remaining_data_keys.contains(old_key)),
+        "stale set data key survived compaction"
+    );
+
+    storage.shutdown().await;
+}
+
 /// Test get_global_smallest_flushed_log_index returns minimum across all instances
 #[tokio::test]
 async fn test_get_global_smallest_flushed_log_index() {