feat: Support backing up the shard databases

Cargo.toml

@@ -56,6 +56,11 @@ itertools = "0.13.0"
 cadence = "1.5.0"
 tempfile = "3.13.0"
 foundry-common = { git = "https://github.com/foundry-rs/foundry", rev="6b07c77eb1c1d1c4b56ffa7f79240254b73236d2" }
+chrono = "0.4.39"
+tar = "0.4.40"
+gzp = "0.11.3"
+flate2 = "1.0.28"
+
 
 [build-dependencies]
 tonic-build = "0.9.2"

Dockerfile

@@ -2,7 +2,7 @@ FROM rust:1.83 AS builder
 
 WORKDIR /usr/src/app
 
-RUN apt-get update && apt-get install -y libclang-dev libjemalloc-dev llvm-dev make protobuf-compiler
+RUN apt-get update && apt-get install -y libclang-dev libjemalloc-dev llvm-dev make protobuf-compiler cmake
 
 # Unfortunately, we can't prefetch creates without including the source code,
 # since the Cargo configuration references files in src.

src/cfg.rs

@@ -35,6 +35,7 @@ pub struct Config {
     pub mempool: mempool::mempool::Config,
     pub rpc_address: String,
     pub rocksdb_dir: String,
+    pub backup_dir: String,
     pub clear_db: bool,
     pub statsd: StatsdConfig,
     pub trie_branching_factor: u32,
@@ -52,6 +53,7 @@ impl Default for Config {
             mempool: mempool::mempool::Config::default(),
             rpc_address: "0.0.0.0:3383".to_string(),
             rocksdb_dir: ".rocks".to_string(),
+            backup_dir: "".to_string(),
             clear_db: false,
             statsd: StatsdConfig::default(),
             trie_branching_factor: 16,

src/main.rs

@@ -13,11 +13,11 @@ use snapchain::proto::hub_service_server::HubServiceServer;
 use snapchain::storage::db::RocksDB;
 use snapchain::storage::store::BlockStore;
 use snapchain::utils::statsd_wrapper::StatsdClientWrapper;
+use std::collections::HashMap;
 use std::error::Error;
 use std::net;
 use std::net::SocketAddr;
 use std::process;
-use std::sync::Arc;
 use std::time::Duration;
 use tokio::signal::ctrl_c;
 use tokio::sync::mpsc;
@@ -81,9 +81,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
     let addr = app_config.gossip.address.clone();
     let grpc_addr = app_config.rpc_address.clone();
     let grpc_socket_addr: SocketAddr = grpc_addr.parse()?;
-    let db_path = format!("{}/farcaster", app_config.rocksdb_dir);
-    let db = Arc::new(RocksDB::new(db_path.clone().as_str()));
-    db.open().unwrap();
+    let db = RocksDB::open_block_db(app_config.rocksdb_dir.as_str());
     let block_store = BlockStore::new(db);
 
     info!(addr = addr, grpc_addr = grpc_addr, "HubService listening",);
@@ -206,6 +204,29 @@ async fn main() -> Result<(), Box<dyn Error>> {
         shutdown_tx.send(()).await.ok();
     });
 
+    if app_config.backup_dir != "" {
+        let backup_dir = app_config.backup_dir.clone();
+        let shard_ids = app_config.consensus.shard_ids.clone();
+        let block_db = block_store.db.clone();
+        let mut dbs = HashMap::new();
+        dbs.insert(0, block_db.clone());
+        node.shard_stores
+            .iter()
+            .for_each(|(shard_id, shard_store)| {
+                dbs.insert(*shard_id, shard_store.shard_store.db.clone());
+            });
+        tokio::spawn(async move {
+            info!(
+                "Backing up {:?} shard databases to {:?}",
+                shard_ids, backup_dir
+            );
+            let timestamp = chrono::Utc::now().timestamp_millis();
+            dbs.iter().for_each(|(shard_id, db)| {
+                RocksDB::backup_db(db.clone(), &backup_dir, *shard_id, timestamp).unwrap();
+            });
+        });
+    }
+
     // Create a timer for block creation
     let mut block_interval = time::interval(Duration::from_secs(2));
 

src/node/snapchain_node.rs

@@ -21,7 +21,6 @@ use informalsystems_malachitebft_metrics::Metrics;
 use libp2p::identity::ed25519::Keypair;
 use ractor::ActorRef;
 use std::collections::{BTreeMap, HashMap};
-use std::sync::Arc;
 use tokio::sync::mpsc;
 use tracing::warn;
 
@@ -85,12 +84,10 @@ impl SnapchainNode {
             };
             let ctx = SnapchainValidatorContext::new(keypair.clone());
 
-            let db = RocksDB::new(format!("{}/shard{}", rocksdb_dir, shard_id).as_str());
-            db.open().unwrap();
-
+            let db = RocksDB::open_shard_db(rocksdb_dir.as_str(), shard_id);
             let trie = merkle_trie::MerkleTrie::new(trie_branching_factor).unwrap(); //TODO: don't unwrap()
             let engine = ShardEngine::new(
-                Arc::new(db),
+                db,
                 trie,
                 shard_id,
                 StoreLimits::default(),

src/storage/db/mod.rs

@@ -1,3 +1,4 @@
 pub use self::rocksdb::*;
 
+mod multi_chunk_writer;
 mod rocksdb;