fix: fix try pin condition, refactor version manager

exhauster/src/lib.rs

@@ -56,7 +56,7 @@ pub async fn build_exhauster_with_object_store(
     let sstable_store = build_sstable_store(config, object_store, lsm_tree_metrics)?;
 
     let options = ExhausterOptions {
-        node_id: config.id,
+        node: config.id,
         sstable_store,
         // TODO: Restore from persistent store.
         sstable_sequential_id: 1,

exhauster/src/service.rs

@@ -25,29 +25,39 @@ fn internal(e: impl Into<Box<dyn std::error::Error>>) -> Status {
 }
 
 pub struct ExhausterOptions {
-    pub node_id: u64,
+    pub node: u64,
     pub sstable_store: SstableStoreRef,
     pub sstable_sequential_id: u64,
 }
 
 pub struct Exhauster {
-    options: ExhausterOptions,
+    node: u64,
     sstable_store: SstableStoreRef,
     sstable_sequential_id: AtomicU64,
 }
 
+impl std::fmt::Debug for Exhauster {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Exhauster")
+            .field("node", &self.node)
+            .finish()
+    }
+}
+
 impl Exhauster {
     pub fn new(options: ExhausterOptions) -> Self {
         Self {
+            node: options.node,
+
             sstable_store: options.sstable_store.clone(),
             sstable_sequential_id: AtomicU64::new(options.sstable_sequential_id),
-            options,
         }
     }
 }
 
 #[async_trait]
 impl ExhausterService for Exhauster {
+    #[tracing::instrument(level = "trace")]
     async fn compaction(
         &self,
         request: Request<CompactionRequest>,
@@ -145,8 +155,7 @@ impl ExhausterService for Exhauster {
 impl Exhauster {
     fn gen_sstable_id(&self) -> u64 {
         let sequential_id = self.sstable_sequential_id.fetch_add(1, Ordering::SeqCst);
-        let node_id = self.options.node_id;
-        (node_id << 32) | sequential_id
+        (self.node << 32) | sequential_id
     }
 
     async fn build_and_upload_sst(
@@ -158,7 +167,7 @@ impl Exhauster {
         let (meta, data) = builder.build()?;
         let data_size = meta.data_size as u64;
         let sst = Sstable::new(sst_id, Arc::new(meta));
-        trace!("build sst: {:#?}", sst);
+        trace!("build sst: {:?}", sst);
         self.sstable_store
             .put(&sst, data, CachePolicy::Fill)
             .await?;

rudder/src/lib.rs

@@ -1,3 +1,5 @@
+#![feature(let_chains)]
+
 pub mod config;
 pub mod error;
 pub mod meta;
@@ -123,6 +125,7 @@ async fn build_version_manager(
     sstable_store: SstableStoreRef,
 ) -> Result<VersionManager> {
     let version_manager_options = VersionManagerOptions {
+        node: config.id,
         levels_options: config.lsm_tree.levels_options.clone(),
         // TODO: Recover from meta or scanning.
         levels: vec![vec![]; config.lsm_tree.levels_options.len()],
@@ -144,6 +147,7 @@ async fn build_version_manager(
 fn build_meta_store(config: &RudderConfig) -> Result<MetaStoreRef> {
     // TODO: Build with storage.
     let meta_store = MemoryMetaStore::new(
+        config.id,
         config
             .lsm_tree
             .compaction_pin_ttl

rudder/src/meta/mem.rs

@@ -46,19 +46,32 @@ pub struct MemoryMetaStoreCore {
     raft_states: HashMap<u64, (RaftState, SystemTime)>,
 
     pinned_sstables: BTreeMap<u64, SystemTime>,
-    sstable_pin_ttl: Duration,
 }
 
 pub struct MemoryMetaStore {
+    node: u64,
+    sstable_pin_ttl: Duration,
+
     core: RwLock<MemoryMetaStoreCore>,
     timestamp: AtomicU32,
 }
 
+impl std::fmt::Debug for MemoryMetaStore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("MemoryMetaStore")
+            .field("node", &self.node)
+            .field("sstable_pin_ttl", &self.sstable_pin_ttl)
+            .finish()
+    }
+}
+
 impl MemoryMetaStore {
-    pub fn new(sstable_pin_ttl: Duration) -> Self {
+    pub fn new(node: u64, sstable_pin_ttl: Duration) -> Self {
         Self {
+            node,
+            sstable_pin_ttl,
+
             core: RwLock::new(MemoryMetaStoreCore {
-                sstable_pin_ttl,
                 ..Default::default()
             }),
             timestamp: AtomicU32::new(1),
@@ -212,12 +225,11 @@ impl MetaStore for MemoryMetaStore {
 
     async fn pin_sstables(&self, sst_ids: &[u64], time: SystemTime) -> Result<bool> {
         let mut guard = self.core.write();
+        trace!("pin - try to pin ssts: {:?}", sst_ids);
         let mut pin = true;
         for sst_id in sst_ids.iter() {
-            if guard.pinned_sstables.get(sst_id).map_or_else(
-                || false,
-                |last_pin_time| *last_pin_time + guard.sstable_pin_ttl < time,
-            ) {
+            // If target sst is already pinned, and doesn't timeout.
+            if let Some(last_pinned_time) = guard.pinned_sstables.get(sst_id) && *last_pinned_time + self.sstable_pin_ttl > time{
                 pin = false;
                 break;
             }
@@ -235,8 +247,10 @@ impl MetaStore for MemoryMetaStore {
         Ok(true)
     }
 
+    #[tracing::instrument(level = "trace")]
     async fn unpin_sstables(&self, sst_ids: &[u64]) -> Result<()> {
         let mut guard = self.core.write();
+        trace!("unpin - try to unpin ssts: {:?}", sst_ids);
         for sst_id in sst_ids.iter() {
             guard.pinned_sstables.remove(sst_id);
         }
@@ -247,14 +261,16 @@ impl MetaStore for MemoryMetaStore {
         Ok(())
     }
 
+    #[tracing::instrument(level = "trace", ret, err)]
     async fn is_sstables_pinned(&self, sst_ids: &[u64], time: SystemTime) -> Result<Vec<bool>> {
         let mut pinned = Vec::with_capacity(sst_ids.len());
         let guard = self.core.read();
         for sst_id in sst_ids {
-            pinned.push(guard.pinned_sstables.get(sst_id).map_or_else(
-                || false,
-                |last_pin_time| *last_pin_time + guard.sstable_pin_ttl >= time,
-            ));
+            if let Some(last_pinned_time) = guard.pinned_sstables.get(sst_id) && *last_pinned_time + self.sstable_pin_ttl > time{
+                pinned.push(true);
+            }else {
+                pinned.push(false);
+            }
         }
         Ok(pinned)
     }

rudder/src/worker/compaction_detector.rs

@@ -422,7 +422,7 @@ async fn sub_compaction(
         id: 0,
         sstable_diffs,
     };
-    trace!("compaction version diff:\n{:#?}", version_diff);
+    trace!("compaction version diff: {:?}", version_diff);
     ctx.version_manager.update(version_diff, false).await?;
 
     Ok(())