fix(storage): align DEL with generation lifecycle (#409)

* fix(storage): align DEL with generation lifecycle

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,26 +89,45 @@ 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 Ok(mut state) = self.state.lock() else {
+            return;
+        };
+        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;
         }
 
-        let mut state = self
-            .state
-            .lock()
-            .expect("compaction filter test gate mutex should not be poisoned");
+        let Ok(mut state) = self.state.lock() else {
+            return;
+        };
         if state.entered {
             return;
         }
 
         state.entered = true;
         self.changed.notify_all();
         while !state.released {
-            state = self
-                .changed
-                .wait(state)
-                .expect("compaction filter test gate wait should succeed");
+            let Ok(next_state) = self.changed.wait(state) else {
+                return;
+            };
+            state = next_state;
         }
     }
 }
@@ -118,6 +138,12 @@ fn compaction_filter_test_gate() -> &'static Mutex<Option<Weak<CompactionFilterT
     GATE.get_or_init(|| Mutex::new(None))
 }
 
+#[cfg(test)]
+pub(crate) fn compaction_filter_test_serial_mutex() -> &'static Mutex<()> {
+    static MUTEX: OnceLock<Mutex<()>> = OnceLock::new();
+    MUTEX.get_or_init(|| Mutex::new(()))
+}
+
 #[cfg(test)]
 pub(crate) struct CompactionFilterTestGateGuard {
     gate: Arc<CompactionFilterTestGate>,
@@ -127,9 +153,9 @@ pub(crate) struct CompactionFilterTestGateGuard {
 impl Drop for CompactionFilterTestGateGuard {
     fn drop(&mut self) {
         self.gate.release();
-        *compaction_filter_test_gate()
-            .lock()
-            .expect("compaction filter test gate registry should not be poisoned") = None;
+        if let Ok(mut installed) = compaction_filter_test_gate().lock() {
+            *installed = None;
+        }
     }
 }
 
@@ -139,7 +165,7 @@ pub(crate) fn install_compaction_filter_test_gate(
 ) -> CompactionFilterTestGateGuard {
     let mut installed = compaction_filter_test_gate()
         .lock()
-        .expect("compaction filter test gate registry should not be poisoned");
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
     assert!(
         installed.is_none(),
         "only one compaction test gate may be installed"
@@ -151,16 +177,26 @@ pub(crate) fn install_compaction_filter_test_gate(
 
 #[cfg(test)]
 fn block_once_for_compaction_filter_test(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);
+    let Ok(installed) = compaction_filter_test_gate().lock() else {
+        return;
+    };
+    let gate = installed.as_ref().and_then(Weak::upgrade);
     if let Some(gate) = gate {
         gate.enter_and_wait(key);
     }
 }
 
+#[cfg(test)]
+fn record_compaction_filter_remove(key: &[u8]) {
+    let Ok(installed) = compaction_filter_test_gate().lock() else {
+        return;
+    };
+    let gate = installed.as_ref().and_then(Weak::upgrade);
+    if let Some(gate) = gate {
+        gate.record_removed_key(key);
+    }
+}
+
 #[derive(Debug)]
 enum MetaLookup {
     Valid,
@@ -326,21 +362,28 @@ 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 => {
                 let cur_time = Utc::now().timestamp_micros() as u64;
                 if self.cur_meta_etime != 0 && self.cur_meta_etime < cur_time {
-                    return CompactionDecision::Remove;
-                }
-
-                match Self::extract_data_version(key) {
-                    Some(ver) if self.cur_meta_version > ver => CompactionDecision::Remove,
-                    _ => CompactionDecision::Keep,
+                    CompactionDecision::Remove
+                } else {
+                    match Self::extract_data_version(key) {
+                        Some(ver) if self.cur_meta_version > ver => CompactionDecision::Remove,
+                        _ => CompactionDecision::Keep,
+                    }
                 }
             }
+        };
+
+        #[cfg(test)]
+        if let CompactionDecision::Remove = &decision {
+            record_compaction_filter_remove(key);
         }
+
+        decision
     }
 }
 
@@ -576,6 +619,9 @@ mod tests {
 
     #[test]
     fn test_removes_data_if_meta_is_expired() {
+        let _serial = compaction_filter_test_serial_mutex()
+            .lock()
+            .expect("compaction filter gate tests must run serially");
         let path = unique_test_db_path();
         let (db_cell, db) = setup_db_for_filter_test(&path);
 
@@ -598,8 +644,15 @@ mod tests {
 
         let data_key = encode_data_key(b"mykey", 1);
 
+        let gate = CompactionFilterTestGate::new(b"mykey");
+        let _gate_guard = install_compaction_filter_test_gate(Arc::clone(&gate));
+
         let decision = filter.filter(0, &data_key, b"");
         assert!(matches!(decision, CompactionDecision::Remove));
+        assert_eq!(
+            gate.wait_until_removed(std::time::Duration::from_secs(1)),
+            vec![data_key]
+        );
     }
 
     #[test]