Delete inactive entries after staging commit

components/consensusmanager/src/lib.rs

@@ -37,6 +37,8 @@ pub trait ConsensusFactory: Sync + Send {
 
     /// Close the factory and cleanup any shared resources used by it
     fn close(&self);
+
+    fn clean_non_active_consensus_entries(&self);
 }
 
 /// Test-only mock factory
@@ -54,6 +56,10 @@ impl ConsensusFactory for MockFactory {
     fn close(&self) {
         unimplemented!()
     }
+
+    fn clean_non_active_consensus_entries(&self) {
+        unimplemented!()
+    }
 }
 
 /// Defines a trait which handles consensus resets for external parts of the system. We avoid using
@@ -146,6 +152,10 @@ impl ConsensusManager {
         debug!("[Consensus manager] all consensus threads exited");
         self.factory.close();
     }
+
+    pub fn clean_non_active_consensus_entries(&self) {
+        self.factory.clean_non_active_consensus_entries();
+    }
 }
 
 impl Service for ConsensusManager {
@@ -185,7 +195,7 @@ impl StagingConsensus {
         for handler in handlers {
             handler.handle_consensus_reset();
         }
-        // TODO: Delete non active consensus entries
+        self.manager.clean_non_active_consensus_entries();
     }
 
     pub fn cancel(self) {

consensus/src/consensus/factory.rs

@@ -6,7 +6,7 @@ use itertools::Itertools;
 use kaspa_consensus_core::config::Config;
 use kaspa_consensus_notify::root::ConsensusNotificationRoot;
 use kaspa_consensusmanager::{ConsensusFactory, ConsensusInstance, DynConsensusCtl, SessionLock};
-use kaspa_core::{debug, time::unix_now};
+use kaspa_core::{debug, info, time::unix_now};
 use kaspa_database::{
     prelude::{BatchDbWriter, CachedDbAccess, CachedDbItem, DirectDbWriter, StoreError, StoreResult, StoreResultExtensions, DB},
     registry::DatabaseStorePrefixes,
@@ -216,26 +216,6 @@ impl Factory {
         factory.clean_non_active_consensus_entries();
         factory
     }
-
-    pub fn clean_non_active_consensus_entries(&self) {
-        if self.config.is_archival {
-            return;
-        }
-
-        let mut write_guard = self.management_store.write();
-        let entries_to_delete = write_guard.iterate_non_active().collect_vec();
-        for entry_result in entries_to_delete.iter() {
-            let entry = entry_result.as_ref().unwrap();
-            let dir = self.db_root_dir.join(entry.directory_name.clone());
-            if dir.exists() {
-                fs::remove_dir_all(dir).unwrap();
-            }
-        }
-
-        for entry_result in entries_to_delete {
-            write_guard.delete_entry(entry_result.unwrap()).unwrap();
-        }
-    }
 }
 
 impl ConsensusFactory for Factory {
@@ -307,4 +287,34 @@ impl ConsensusFactory for Factory {
         debug!("Consensus factory: closing");
         self.notification_root.close();
     }
+
+    fn clean_non_active_consensus_entries(&self) {
+        if self.config.is_archival {
+            return;
+        }
+
+        let mut write_guard = self.management_store.write();
+        let entries_to_delete = write_guard
+            .iterate_non_active()
+            .filter_map(|entry_result| {
+                let entry = entry_result.unwrap();
+                let dir = self.db_root_dir.join(entry.directory_name.clone());
+                if dir.exists() {
+                    match fs::remove_dir_all(dir) {
+                        Ok(_) => Some(entry),
+                        Err(e) => {
+                            info!("Couldn't delete consensus entry {}: {}", entry.key, e);
+                            None
+                        }
+                    }
+                } else {
+                    Some(entry)
+                }
+            })
+            .collect_vec();
+
+        for entry in entries_to_delete {
+            write_guard.delete_entry(entry).unwrap();
+        }
+    }
 }

consensus/src/consensus/test_consensus.rs

@@ -282,4 +282,8 @@ impl ConsensusFactory for TestConsensusFactory {
     fn close(&self) {
         self.tc.notification_root().close();
     }
+
+    fn clean_non_active_consensus_entries(&self) {
+        unimplemented!()
+    }
 }