Merge pull request #2742 from ljedrz/feat/db_checkpoints

[Feat] Database checkpoints

Author: vicsn

Cargo.lock

@@ -241,6 +241,15 @@ impl<N: Network, C: ConsensusStorage<N>> Ledger<N, C> {
         Ok(ledger)
     }
 
+    /// Creates a rocksdb checkpoint in the specified directory, which needs to not exist at the
+    /// moment of calling. The checkpoints are based on hard links, which means they can both be
+    /// incremental (i.e. they aren't full physical copies), and used as full rollback points
+    /// (a checkpoint can be used to completely replace the original ledger).
+    #[cfg(feature = "rocks")]
+    pub fn backup_database<P: AsRef<std::path::Path>>(&self, path: P) -> Result<()> {
+        self.vm.block_store().backup_database(path).map_err(|err| anyhow!(err))
+    }
+
     /// Loads the provers and the number of solutions they have submitted for the current epoch.
     pub fn load_epoch_provers(&self) -> IndexMap<Address<N>, u32> {
         // Fetch the block heights that belong to the current epoch.

ledger/src/lib.rs

@@ -1021,6 +1021,9 @@ pub trait BlockStorage<N: Network>: 'static + Clone + Send + Sync {
             aborted_transaction_ids,
         )?))
     }
+
+    #[cfg(feature = "rocks")]
+    fn backup_database<P: AsRef<std::path::Path>>(&self, path: P) -> Result<(), String>;
 }
 
 /// The block store.
@@ -1193,6 +1196,11 @@ impl<N: Network, B: BlockStorage<N>> BlockStore<N, B> {
     pub fn unpause_atomic_writes<const DISCARD_BATCH: bool>(&self) -> Result<()> {
         self.storage.unpause_atomic_writes::<DISCARD_BATCH>()
     }
+
+    #[cfg(feature = "rocks")]
+    pub fn backup_database<P: AsRef<std::path::Path>>(&self, path: P) -> Result<(), String> {
+        self.storage.backup_database(path)
+    }
 }
 
 impl<N: Network, B: BlockStorage<N>> BlockStore<N, B> {

ledger/store/src/block/mod.rs

@@ -211,4 +211,9 @@ impl<N: Network> BlockStorage<N> for BlockMemory<N> {
     fn transaction_store(&self) -> &TransactionStore<N, Self::TransactionStorage> {
         &self.transaction_store
     }
+
+    #[cfg(feature = "rocks")]
+    fn backup_database<P: AsRef<std::path::Path>>(&self, _path: P) -> Result<(), String> {
+        Err("Unavailable in memory-only mode".to_owned())
+    }
 }

ledger/store/src/helpers/memory/block.rs

@@ -218,4 +218,10 @@ impl<N: Network> BlockStorage<N> for BlockDB<N> {
     fn transaction_store(&self) -> &TransactionStore<N, Self::TransactionStorage> {
         &self.transaction_store
     }
+
+    #[cfg(feature = "rocks")]
+    fn backup_database<P: AsRef<std::path::Path>>(&self, path: P) -> Result<(), String> {
+        // Any map can be used to retrieve the common RocksDB instance.
+        self.id_map().backup_database(path)
+    }
 }

ledger/store/src/helpers/rocksdb/block.rs

@@ -21,7 +21,7 @@ use crate::helpers::{Map, MapRead};
 use core::{fmt, fmt::Debug, hash::Hash, mem};
 use indexmap::IndexMap;
 use smallvec::SmallVec;
-use std::{borrow::Cow, ops::Deref, sync::atomic::Ordering};
+use std::{borrow::Cow, ops::Deref, path::Path, sync::atomic::Ordering};
 use tracing::error;
 
 #[derive(Clone)]
@@ -48,6 +48,13 @@ pub struct InnerDataMap<K: Serialize + DeserializeOwned, V: Serialize + Deserial
     pub(super) checkpoints: Mutex<Vec<usize>>,
 }
 
+impl<K: Serialize + DeserializeOwned, V: Serialize + DeserializeOwned> InnerDataMap<K, V> {
+    pub fn backup_database<P: AsRef<Path>>(&self, path: P) -> Result<(), String> {
+        let checkpoint = rocksdb::checkpoint::Checkpoint::new(&self.database)?;
+        checkpoint.create_checkpoint(path).map_err(|e| e.into_string())
+    }
+}
+
 impl<
     'a,
     K: 'a + Copy + Clone + Debug + PartialEq + Eq + Hash + Serialize + DeserializeOwned + Send + Sync,