[WIP] point in time recovery POC

Cargo.toml

@@ -95,7 +95,7 @@ version = "0.7.0"
 license = "Apache-2.0"
 
 [workspace.dependencies]
-rocksdb = "0.22"
+rocksdb = { git="https://github.com/rust-rocksdb/rust-rocksdb.git", branch="master" }
 librocksdb-sys = "0.17.0"
 
 # Bonsai trie dependencies
@@ -242,8 +242,8 @@ tracing-test = "0.2.5"
 tracing-opentelemetry = "0.26.0"
 
 [patch.crates-io]
-rocksdb = { git = "https://github.com/madara-alliance/rust-rocksdb", branch = "read-options-set-raw-snapshot" }
-librocksdb-sys = { git = "https://github.com/madara-alliance/rust-rocksdb", branch = "read-options-set-raw-snapshot" }
+rocksdb = { git="https://github.com/rust-rocksdb/rust-rocksdb.git", branch="master" }
+librocksdb-sys = { git="https://github.com/rust-rocksdb/rust-rocksdb.git", branch="master" }
 
 starknet-types-core = { git = "https://github.com/kasarlabs/types-rs.git", branch = "feat-deserialize-v0.1.7" }
 starknet-types-rpc = { git = "https://github.com/jbcaron/types-rs.git", branch = "fork" }

configs/presets/devnet.yaml

@@ -5,7 +5,7 @@ gateway_url: "http://localhost:8080/gateway/"
 native_fee_token_address: "0x04718f5a0fc34cc1af16a1cdee98ffb20c31f5cd61d6ab07201858f4287c938d"
 parent_fee_token_address: "0x049d36570d4e46f48e99674bd3fcc84644ddd6b96f7c741b1562b82f9e004dc7"
 latest_protocol_version: "0.13.2"
-block_time: "10s"
+block_time: "1s"
 pending_block_update_time: "2s"
 execution_batch_size: 16
 bouncer_config:

crates/client/db/src/block_db.rs

@@ -248,7 +248,7 @@ impl MadaraBackend {
             tx.put_cf(&col, ROW_PENDING_BOUNCER_WEIGHTS, bincode::serialize(&bouncer_weights)?);
         }
         let mut writeopts = WriteOptions::new();
-        writeopts.disable_wal(true);
+        //writeopts.disable_wal(true);
         self.db.write_opt(tx, &writeopts)?;
         Ok(())
     }
@@ -263,7 +263,7 @@ impl MadaraBackend {
         tx.delete_cf(&col, ROW_PENDING_SEGMENTS);
         tx.delete_cf(&col, ROW_PENDING_BOUNCER_WEIGHTS);
         let mut writeopts = WriteOptions::new();
-        writeopts.disable_wal(true);
+        //writeopts.disable_wal(true);
         self.db.write_opt(tx, &writeopts)?;
         Ok(())
     }
@@ -272,7 +272,7 @@ impl MadaraBackend {
     pub fn write_last_confirmed_block(&self, l1_last: u64) -> Result<()> {
         let col = self.db.get_column(Column::BlockStorageMeta);
         let mut writeopts = WriteOptions::default(); // todo move that in db
-        writeopts.disable_wal(true);
+        //writeopts.disable_wal(true);
         self.db.put_cf_opt(&col, ROW_L1_LAST_CONFIRMED_BLOCK, bincode::serialize(&l1_last)?, &writeopts)?;
         Ok(())
     }
@@ -320,7 +320,7 @@ impl MadaraBackend {
         tx.delete_cf(&meta, ROW_PENDING_STATE_UPDATE);
 
         let mut writeopts = WriteOptions::new();
-        writeopts.disable_wal(true);
+        //writeopts.disable_wal(true);
         self.db.write_opt(tx, &writeopts)?;
         Ok(())
     }

crates/client/db/src/bonsai_db.rs

@@ -38,7 +38,7 @@ pub struct BonsaiDb {
 impl BonsaiDb {
     pub(crate) fn new(db: Arc<DB>, snapshots: Arc<Snapshots>, column_mapping: DatabaseKeyMapping) -> Self {
         let mut write_opt = WriteOptions::default();
-        write_opt.disable_wal(true);
+        // write_opt.disable_wal(true);
         Self { db, column_mapping, write_opt, snapshots }
     }
 }

crates/client/db/src/class_db.rs

@@ -120,7 +120,7 @@ impl MadaraBackend {
         col_compiled: Column,
     ) -> Result<(), MadaraStorageError> {
         let mut writeopts = WriteOptions::new();
-        writeopts.disable_wal(true);
+        //writeopts.disable_wal(true);
 
         converted_classes.par_chunks(DB_UPDATES_BATCH_SIZE).try_for_each_init(
             || self.db.get_column(col_info),
@@ -200,7 +200,7 @@ impl MadaraBackend {
     #[tracing::instrument(fields(module = "ClassDB"))]
     pub(crate) fn class_db_clear_pending(&self) -> Result<(), MadaraStorageError> {
         let mut writeopts = WriteOptions::new();
-        writeopts.disable_wal(true);
+        //writeopts.disable_wal(true);
 
         self.db.delete_range_cf_opt(&self.db.get_column(Column::PendingClassInfo), &[] as _, LAST_KEY, &writeopts)?;
         self.db.delete_range_cf_opt(

crates/client/db/src/contract_db.rs

@@ -150,7 +150,7 @@ impl MadaraBackend {
         let block_number = u32::try_from(block_number).map_err(|_| MadaraStorageError::InvalidBlockNumber)?;
 
         let mut writeopts = WriteOptions::new();
-        writeopts.disable_wal(true);
+        //writeopts.disable_wal(true);
 
         fn write_chunk(
             db: &DB,
@@ -214,7 +214,7 @@ impl MadaraBackend {
         contract_kv_updates: &[((Felt, Felt), Felt)],
     ) -> Result<(), MadaraStorageError> {
         let mut writeopts = WriteOptions::new();
-        writeopts.disable_wal(true);
+        //writeopts.disable_wal(true);
 
         // Note: pending has keys in bincode, not bytes
 
@@ -252,7 +252,7 @@ impl MadaraBackend {
     #[tracing::instrument(fields(module = "ContractDB"))]
     pub(crate) fn contract_db_clear_pending(&self) -> Result<(), MadaraStorageError> {
         let mut writeopts = WriteOptions::new();
-        writeopts.disable_wal(true);
+        //writeopts.disable_wal(true);
 
         self.db.delete_range_cf_opt(
             &self.db.get_column(Column::PendingContractToNonces),

crates/client/db/src/devnet_db.rs

@@ -36,7 +36,7 @@ impl MadaraBackend {
     pub fn set_devnet_predeployed_keys(&self, devnet_keys: DevnetPredeployedKeys) -> Result<()> {
         let nonce_column = self.db.get_column(Column::Devnet);
         let mut writeopts = WriteOptions::default();
-        writeopts.disable_wal(true);
+        //writeopts.disable_wal(true);
         self.db.put_cf_opt(&nonce_column, DEVNET_KEYS, bincode::serialize(&devnet_keys)?, &writeopts)?;
         Ok(())
     }

crates/client/db/src/l1_db.rs

@@ -104,7 +104,7 @@ impl MadaraBackend {
     ) -> Result<(), DbError> {
         let messaging_column = self.db.get_column(Column::L1Messaging);
         let mut writeopts = WriteOptions::default(); // todo move that in db
-        writeopts.disable_wal(true);
+        //writeopts.disable_wal(true);
         self.db.put_cf_opt(
             &messaging_column,
             LAST_SYNCED_L1_EVENT_BLOCK,
@@ -124,7 +124,7 @@ impl MadaraBackend {
     pub fn set_l1_messaging_nonce(&self, nonce: Nonce) -> Result<(), DbError> {
         let nonce_column = self.db.get_column(Column::L1MessagingNonce);
         let mut writeopts = WriteOptions::default();
-        writeopts.disable_wal(true);
+        //writeopts.disable_wal(true);
         self.db.put_cf_opt(&nonce_column, bincode::serialize(&nonce)?, /* empty value */ [], &writeopts)?;
         Ok(())
     }

crates/client/db/src/lib.rs

@@ -254,7 +254,7 @@ impl DatabaseExt for DB {
 
 fn make_write_opt_no_wal() -> WriteOptions {
     let mut opts = WriteOptions::new();
-    opts.disable_wal(true);
+    // opts.disable_wal(true);
     opts
 }
 

crates/client/db/src/rocksdb_options.rs

@@ -16,11 +16,15 @@ pub fn rocksdb_global_options() -> Result<Options> {
     let cores = std::thread::available_parallelism().map(|e| e.get() as i32).unwrap_or(1);
     options.increase_parallelism(cores);
     options.set_max_background_jobs(cores);
+    options.set_track_and_verify_wals_in_manifest(true);
 
     options.set_atomic_flush(true);
     options.set_max_subcompactions(cores as _);
+    options.set_wal_recovery_mode(rocksdb::DBRecoveryMode::PointInTime);
+    options.set_wal_ttl_seconds(60 * 60 * 24 * 7); // 1 week
+    options.set_wal_dir("/Users/anukkrit/Desktop/karnot.xyz/codes/madara-alliance/wal_log");
 
-    options.set_max_log_file_size(10 * MiB);
+    options.set_max_log_file_size(10 * GiB);
     options.set_max_open_files(2048);
     options.set_keep_log_file_num(3);
     options.set_log_level(rocksdb::LogLevel::Warn);

crates/client/db/src/rocksdb_snapshot.rs

@@ -22,7 +22,7 @@ impl<D: DBAccess> SnapshotWithDBArc<D> {
     fn readopts_with_raw_snapshot<R>(&self, mut readopts: ReadOptions, f: impl FnOnce(&ReadOptions) -> R) -> R {
         // Safety: the snapshot originates from the `db`, and it is not dropped during the lifetime of the `readopts` variable.
         unsafe {
-            readopts.set_raw_snapshot(self.inner);
+            // readopts.set_raw_snapshot(self.inner);
         }
 
         f(&readopts)

crates/client/sync/src/l2.rs

@@ -112,6 +112,7 @@ async fn l2_verify_and_apply_task(
 
         if backup_every_n_blocks.is_some_and(|backup_every_n_blocks| header.block_number % backup_every_n_blocks == 0) {
             tracing::info!("⏳ Backing up database at block {}...", header.block_number);
+            tracing::info!("[backup] Current Block Number: {}", header.block_number);
             let sw = PerfStopwatch::new();
             backend.backup().await.context("backing up database")?;
             tracing::info!("✅ Database backup is done ({:?})", sw.elapsed());

crates/node/src/cli/sync.rs

@@ -80,7 +80,6 @@ pub struct SyncParams {
     #[clap(env = "MADARA_STOP_ON_SYNC", long, default_value_t = false)]
     pub stop_on_sync: bool,
 
-    /// Periodically create a backup, for debugging purposes. Use it with `--backup-dir <PATH>`.
     #[clap(env = "MADARA_BACKUP_EVERY_N_BLOCKS", long, value_name = "NUMBER OF BLOCKS")]
     pub backup_every_n_blocks: Option<u64>,