feat(rooch-db): add repair command in database module (#2849)

* feat(rooch-db): simplify error handling in write_batch

Streamline error handling by removing unnecessary unwrapping in delete and put operations. This improves code readability and reduces potential runtime panics.

* feat(rooch-da): add repair command in database module

Introduce a new RepairCommand to the database module, allowing users to repair database issues. The command has options for execution, base data directory, and chain ID.

* refactor(rooch-store): improve data repair and validation logic

Enhanced the data repair mechanisms and added detailed validation logic.
This improves consistency checks and ensures data integrity during repairs.

* feat(rooch-db): add thorough store check and logging

Introduce a detailed check for MoveOS store state root consistency with enhanced logging for DA repair outcomes. This enhances diagnostic capability by outputting information on issues and repairs conducted.

* feat(rooch-db): integrate RepairCommand into DBCommand

Add RepairCommand execution capability to the DBCommand enum, allowing repair operations via CLI. Updated error handling to use RoochResult in repair.rs.

* refactor(rooch-store): simplify append_submitting_block function

Remove redundant parameters in append_submitting_block methods for clarity and efficiency. This change reduces the function's signature complexity and ensures that last_block_number is determined internally, streamlining the block submission process.

* fix(rooch-da): add last_block_number to check_append

crates/rooch-db/src/lib.rs

@@ -26,8 +26,9 @@ use raw_store::{rocks::RocksDB, StoreInstance};
 use rooch_config::store_config::StoreConfig;
 use rooch_indexer::store::traits::IndexerStoreTrait;
 use rooch_indexer::{indexer_reader::IndexerReader, IndexerStore};
-use rooch_store::meta_store::SEQUENCER_INFO_KEY;
+use rooch_store::meta_store::{MetaStore, SEQUENCER_INFO_KEY};
 use rooch_store::state_store::StateStore;
+use rooch_store::transaction_store::TransactionStore;
 use rooch_store::{
     RoochStore, META_SEQUENCER_INFO_COLUMN_FAMILY_NAME, STATE_CHANGE_SET_COLUMN_FAMILY_NAME,
     TRANSACTION_COLUMN_FAMILY_NAME, TX_SEQUENCE_INFO_MAPPING_COLUMN_FAMILY_NAME,
@@ -37,6 +38,7 @@ use rooch_types::indexer::state::{
     IndexerObjectStatesIndexGenerator,
 };
 use rooch_types::sequencer::SequencerInfo;
+use tracing::error;
 
 #[derive(Clone)]
 pub struct RoochDB {
@@ -245,10 +247,10 @@ impl RoochDB {
         let inner_store = &self.rooch_store.store_instance;
         let mut write_batch = WriteBatch::new();
         // remove
-        write_batch.delete(to_bytes(&tx_hash).unwrap())?; // tx_hash:tx
-        write_batch.delete(to_bytes(&tx_order).unwrap())?; // tx_order:tx_hash
-        write_batch.delete(to_bytes(&tx_hash).unwrap())?; // tx_hash:tx_execution_info
-        write_batch.delete(to_bytes(&tx_order).unwrap())?; // tx_order:tx_state_change_set
+        write_batch.delete(to_bytes(&tx_hash)?)?; // tx_hash:tx
+        write_batch.delete(to_bytes(&tx_order)?)?; // tx_order:tx_hash
+        write_batch.delete(to_bytes(&tx_hash)?)?; // tx_hash:tx_execution_info
+        write_batch.delete(to_bytes(&tx_order)?)?; // tx_order:tx_state_change_set
         let mut cf_names = vec![
             TRANSACTION_COLUMN_FAMILY_NAME,
             TX_SEQUENCE_INFO_MAPPING_COLUMN_FAMILY_NAME,
@@ -271,13 +273,10 @@ impl RoochDB {
                 previous_execution_info.size,
             );
             write_batch.put(
-                to_bytes(SEQUENCER_INFO_KEY).unwrap(),
-                to_bytes(&previous_sequencer_info).unwrap(),
-            )?;
-            write_batch.put(
-                to_bytes(STARTUP_INFO_KEY).unwrap(),
-                to_bytes(&startup_info).unwrap(),
+                to_bytes(SEQUENCER_INFO_KEY)?,
+                to_bytes(&previous_sequencer_info)?,
             )?;
+            write_batch.put(to_bytes(STARTUP_INFO_KEY)?, to_bytes(&startup_info)?)?;
             cf_names.push(META_SEQUENCER_INFO_COLUMN_FAMILY_NAME);
             cf_names.push(CONFIG_STARTUP_INFO_COLUMN_FAMILY_NAME);
         }
@@ -347,4 +346,67 @@ impl RoochDB {
         };
         Ok(())
     }
+
+    // check the moveos store:
+    // last execution info match state root
+    fn check_moveos_store_thorough(&self) -> anyhow::Result<()> {
+        let mut last_order = self
+            .rooch_store
+            .get_sequencer_info()?
+            .ok_or_else(|| anyhow::anyhow!("Sequencer info not found"))?
+            .last_order;
+        if last_order == 0 {
+            return Ok(()); // Only genesis
+        }
+        let mut state_root = H256::default();
+        for order in (1..=last_order).rev() {
+            let tx_hash = self.rooch_store.get_tx_hashes(vec![order])?.pop().flatten();
+            if let Some(tx_hash) = tx_hash {
+                let execution_info = self.moveos_store.get_tx_execution_info(tx_hash)?;
+                if let Some(execution_info) = execution_info {
+                    state_root = execution_info.state_root;
+                    last_order = order;
+                    break; // found the last execution info
+                }
+            }
+        }
+        let startup_info = self.moveos_store.config_store.get_startup_info()?;
+        let startup_state_root = startup_info
+            .map(|s| s.state_root)
+            .ok_or_else(|| anyhow::anyhow!("Startup info not found"))?;
+
+        if state_root != startup_state_root {
+            return Err(anyhow!(
+                "State root mismatch: last execution info state root {:?} for order: {}, startup state root {:?}",
+                state_root,
+                last_order,
+                startup_state_root
+            ));
+        }
+
+        Ok(())
+    }
+
+    /// repair the rooch store, return the (issues count, fixed count)
+    /// if exec is false, only report issues, otherwise repair the issues
+    pub fn repair(&self, thorough: bool, exec: bool) -> anyhow::Result<(usize, usize)> {
+        let mut issues = 0;
+        let mut fixed = 0;
+        // repair the rooch store
+        let (rooch_store_issues, rooch_store_fixed) = self.rooch_store.repair(thorough, exec)?;
+        issues += rooch_store_issues;
+        fixed += rooch_store_fixed;
+        // check moveos store
+        if thorough {
+            match self.check_moveos_store_thorough() {
+                Ok(_) => {}
+                Err(e) => {
+                    issues += 1;
+                    error!("MoveOS store check failed: {:?}", e);
+                }
+            }
+        }
+        // TODO repair the changeset sync and indexer store
+        Ok((issues, fixed))
+    }
 }

crates/rooch-rpc-server/src/lib.rs

@@ -298,7 +298,8 @@ pub async fn run_start_server(opt: RoochOpt, server_opt: ServerOpt) -> Result<Se
     let genesis_bytes = RoochGenesis::build(network.clone())?.encode();
     let genesis_namespace = derive_genesis_namespace(&genesis_bytes);
     let last_tx_order = sequencer_proxy.get_sequencer_order().await?;
-    rooch_store.try_repair_da_meta(last_tx_order)?;
+    let (da_issues, da_fixed) = rooch_store.try_repair_da_meta(last_tx_order, false)?;
+    info!("DA meta issues: {:?}, fixed: {:?}", da_issues, da_fixed);
     let da_config = opt.da_config().clone();
     let da_proxy = DAServerProxy::new(
         DAServerActor::new(

crates/rooch-store/Cargo.toml

@@ -18,6 +18,7 @@ anyhow = { workspace = true }
 once_cell = { workspace = true }
 prometheus = { workspace = true }
 tokio = { workspace = true }
+tracing = { workspace = true }
 
 raw-store = { workspace = true }
 accumulator = { workspace = true }