Ensure EVM database integrity and consistency with DVM database

lib/ain-evm/src/core.rs

@@ -31,7 +31,10 @@ use crate::{
     gas::check_tx_intrinsic_gas,
     precompiles::MetachainPrecompiles,
     receipt::ReceiptService,
-    storage::{traits::BlockStorage, Storage},
+    storage::{
+        traits::{BlockStorage, FlushableStorage},
+        Storage,
+    },
     transaction::{
         cache::{TransactionCache, ValidateTxInfo},
         SignedTx,
@@ -143,6 +146,8 @@ impl EVMCoreService {
         );
         storage.put_latest_block(Some(&block))?;
         storage.put_block(&block)?;
+        handler.flush()?;
+        storage.flush()?;
 
         Ok(handler)
     }

lib/ain-evm/src/evm.rs

@@ -29,7 +29,10 @@ use crate::{
     filters::FilterService,
     log::{LogService, Notification},
     receipt::ReceiptService,
-    storage::{traits::BlockStorage, Storage},
+    storage::{
+        traits::{BlockStorage, FlushableStorage},
+        Storage,
+    },
     transaction::{cache::TransactionCache, SignedTx},
     trie::GENESIS_STATE_ROOT,
     Result,
@@ -252,27 +255,27 @@ impl EVMServices {
     /// across all usages. Note: To be replaced with a proper lock flow later.
     ///
     pub unsafe fn commit_block(&self, template: &BlockTemplate) -> Result<()> {
-        {
-            let Some(BlockData { block, receipts }) = template.block_data.clone() else {
-                return Err(format_err!("no constructed EVM block exist in template id").into());
-            };
-
-            debug!(
-                "[finalize_block] Finalizing block number {:#x}, state_root {:#x}",
-                block.header.number, block.header.state_root
-            );
+        let Some(BlockData { block, receipts }) = template.block_data.clone() else {
+            return Err(format_err!("no constructed EVM block exist in template id").into());
+        };
 
-            self.block.connect_block(&block)?;
-            self.logs
-                .generate_logs_from_receipts(&receipts, block.header.number)?;
-            self.receipt.put_receipts(receipts)?;
-            self.channel
-                .sender
-                .send(Notification::Block(block.header.hash()))
-                .map_err(|e| format_err!(e.to_string()))?;
-        }
-        self.core.clear_account_nonce();
+        debug!(
+            "[finalize_block] Finalizing block number {:#x}, state_root {:#x}",
+            block.header.number, block.header.state_root
+        );
 
+        self.block.connect_block(&block)?;
+        self.logs
+            .generate_logs_from_receipts(&receipts, block.header.number)?;
+        self.receipt.put_receipts(receipts)?;
+        self.channel
+            .sender
+            .send(Notification::Block(block.header.hash()))
+            .map_err(|e| format_err!(e.to_string()))?;
+
+        self.core.clear_account_nonce();
+        self.core.flush()?;
+        self.storage.flush()?;
         Ok(())
     }
 

lib/ain-evm/src/lib.rs

@@ -54,6 +54,8 @@ pub enum EVMError {
     StorageError(String),
     #[error("EVM: serde_json error")]
     JsonError(#[from] serde_json::Error),
+    #[error("EVM: serde_json error")]
+    JsonRpcError(#[from] jsonrpsee_core::Error),
     #[error("EVM: rocksdb error")]
     RocksDBError(#[from] rocksdb::Error),
     #[error("EVM: ethabi error")]

lib/ain-evm/src/services.rs

@@ -6,7 +6,7 @@ use std::{
     thread::{self, JoinHandle},
 };
 
-use anyhow::Result;
+use anyhow::{format_err, Result};
 use jsonrpsee_server::ServerHandle;
 use parking_lot::Mutex;
 use tokio::{
@@ -74,33 +74,34 @@ impl Services {
         {
             let json_rpc_handles = self.json_rpc_handles.lock();
             for server in &*json_rpc_handles {
-                server.stop().unwrap();
+                server.stop()?;
             }
         }
 
         {
             let websocket_handles = self.websocket_handles.lock();
             for server in &*websocket_handles {
-                server.stop().unwrap();
+                server.stop()?;
             }
         }
-
-        // TODO: Propogate error
         Ok(())
     }
 
-    pub fn stop(&self) {
+    pub fn stop(&self) -> Result<()> {
         let _ = self.tokio_runtime_channel_tx.blocking_send(());
 
         self.tokio_worker
             .lock()
             .take()
-            .expect("runtime terminated?")
+            .ok_or(format_err!(
+                "failed to stop tokio runtime, early termination"
+            ))?
             .join()
-            .unwrap();
+            .map_err(|_| format_err!("failed to stop tokio runtime"))?;
 
         // Persist EVM State to disk
-        self.evm.core.flush().expect("Could not flush evm state");
-        self.evm.storage.flush().expect("Could not flush storage");
+        self.evm.core.flush()?;
+        self.evm.storage.flush()?;
+        Ok(())
     }
 }

lib/ain-evm/src/storage/block_store.rs

@@ -140,7 +140,7 @@ impl BlockStorage for BlockStore {
         if let Some(block) = block {
             let latest_block_cf = self.column::<columns::LatestBlockNumber>();
             let block_number = block.header.number;
-            latest_block_cf.put(&"latest_block", &block_number)?;
+            latest_block_cf.put(&"", &block_number)?;
         }
         Ok(())
     }
@@ -228,7 +228,7 @@ impl Rollback for BlockStore {
 
             if let Some(block) = self.get_block_by_hash(&block.header.parent_hash)? {
                 let latest_block_cf = self.column::<columns::LatestBlockNumber>();
-                latest_block_cf.put(&"latest_block", &block.header.number)?;
+                latest_block_cf.put(&"", &block.header.number)?;
             }
 
             let logs_cf = self.column::<columns::AddressLogsMap>();

lib/ain-evm/src/storage/mod.rs

@@ -227,6 +227,7 @@ impl Storage {
 impl Rollback for Storage {
     fn disconnect_latest_block(&self) -> Result<()> {
         self.cache.disconnect_latest_block()?;
-        self.blockstore.disconnect_latest_block()
+        self.blockstore.disconnect_latest_block()?;
+        self.flush()
     }
 }

lib/ain-grpc/src/lib.rs

@@ -152,11 +152,12 @@ fn is_services_init_called() -> bool {
     IS_SERVICES_INIT_CALL.load(Ordering::SeqCst)
 }
 
-pub fn stop_services() {
+pub fn stop_services() -> Result<()> {
     if is_services_init_called() {
         info!("Shutdown rs services");
-        SERVICES.stop();
+        SERVICES.stop()?;
     }
+    Ok(())
 }
 
 pub fn wipe_evm_folder() -> Result<()> {

lib/ain-rs-exports/src/core.rs

@@ -1,56 +1,57 @@
-use crate::{ffi::CrossBoundaryResult, prelude::*};
+use crate::{ffi, prelude::*};
+use ain_macros::ffi_fallible;
+use anyhow::Result;
 
-pub fn ain_rs_preinit(result: &mut CrossBoundaryResult) {
+#[ffi_fallible]
+pub fn ain_rs_preinit() -> Result<()> {
     ain_grpc::preinit();
-    cross_boundary_success(result);
+    Ok(())
 }
 
-pub fn ain_rs_init_logging(result: &mut CrossBoundaryResult) {
+#[ffi_fallible]
+pub fn ain_rs_init_logging() -> Result<()> {
     ain_grpc::init_logging();
-    cross_boundary_success(result);
+    Ok(())
 }
 
-pub fn ain_rs_init_core_services(result: &mut CrossBoundaryResult) {
+#[ffi_fallible]
+pub fn ain_rs_init_core_services() -> Result<()> {
     ain_grpc::init_services();
-    result.ok = true;
+    Ok(())
 }
 
-pub fn ain_rs_stop_core_services(result: &mut CrossBoundaryResult) {
-    ain_grpc::stop_services();
-    cross_boundary_success(result);
+#[ffi_fallible]
+pub fn ain_rs_stop_core_services() -> Result<()> {
+    ain_grpc::stop_services()?;
+    Ok(())
 }
 
-pub fn ain_rs_init_network_json_rpc_service(result: &mut CrossBoundaryResult, addr: String) {
-    match ain_grpc::init_network_json_rpc_service(addr) {
-        Ok(()) => cross_boundary_success(result),
-        Err(e) => cross_boundary_error_return(result, e.to_string()),
-    }
+#[ffi_fallible]
+pub fn ain_rs_init_network_json_rpc_service(addr: String) -> Result<()> {
+    ain_grpc::init_network_json_rpc_service(addr)?;
+    Ok(())
 }
 
-pub fn ain_rs_init_network_grpc_service(result: &mut CrossBoundaryResult, addr: String) {
-    match ain_grpc::init_network_grpc_service(addr) {
-        Ok(()) => cross_boundary_success(result),
-        Err(e) => cross_boundary_error_return(result, e.to_string()),
-    }
+#[ffi_fallible]
+pub fn ain_rs_init_network_grpc_service(addr: String) -> Result<()> {
+    ain_grpc::init_network_grpc_service(addr)?;
+    Ok(())
 }
 
-pub fn ain_rs_init_network_subscriptions_service(result: &mut CrossBoundaryResult, addr: String) {
-    match ain_grpc::init_network_subscriptions_service(addr) {
-        Ok(()) => cross_boundary_success(result),
-        Err(e) => cross_boundary_error_return(result, e.to_string()),
-    }
+#[ffi_fallible]
+pub fn ain_rs_init_network_subscriptions_service(addr: String) -> Result<()> {
+    ain_grpc::init_network_subscriptions_service(addr)?;
+    Ok(())
 }
 
-pub fn ain_rs_stop_network_services(result: &mut CrossBoundaryResult) {
-    match ain_grpc::stop_network_services() {
-        Ok(()) => cross_boundary_success(result),
-        Err(e) => cross_boundary_error_return(result, e.to_string()),
-    }
+#[ffi_fallible]
+pub fn ain_rs_stop_network_services() -> Result<()> {
+    ain_grpc::stop_network_services()?;
+    Ok(())
 }
 
-pub fn ain_rs_wipe_evm_folder(result: &mut CrossBoundaryResult) {
-    match ain_grpc::wipe_evm_folder() {
-        Ok(()) => cross_boundary_success(result),
-        Err(e) => cross_boundary_error_return(result, e.to_string()),
-    }
+#[ffi_fallible]
+pub fn ain_rs_wipe_evm_folder() -> Result<()> {
+    ain_grpc::wipe_evm_folder()?;
+    Ok(())
 }

lib/ain-rs-exports/src/evm.rs

@@ -606,6 +606,15 @@ fn evm_try_get_block_number_by_hash(hash: XHash) -> Result<u64> {
     Ok(block_number)
 }
 
+/// Return the block header for a given blockhash.
+///
+/// # Arguments
+///
+/// * `hash` - The hash of the block we want to get the block header.
+///
+/// # Returns
+///
+/// Returns the block header associated with the given blockhash.
 #[ffi_fallible]
 fn evm_try_get_block_header_by_hash(hash: XHash) -> Result<ffi::EVMBlockHeader> {
     let hash = H256::from(hash);
@@ -637,6 +646,21 @@ fn evm_try_get_block_header_by_hash(hash: XHash) -> Result<ffi::EVMBlockHeader>
     Ok(out)
 }
 
+/// Return the latest block header from storage.
+///
+/// # Returns
+///
+/// Returns the latest block header.
+#[ffi_fallible]
+fn evm_try_get_latest_block_hash() -> Result<[u8; 32]> {
+    let block = SERVICES
+        .evm
+        .storage
+        .get_latest_block()?
+        .ok_or(format_err!("latest EVM block not found"))?;
+    Ok(block.header.hash().to_fixed_bytes())
+}
+
 #[ffi_fallible]
 fn evm_try_get_tx_by_hash(tx_hash: XHash) -> Result<ffi::EVMTransaction> {
     let tx_hash = H256::from(tx_hash);

lib/ain-rs-exports/src/lib.rs

@@ -292,6 +292,8 @@ pub mod ffi {
             hash: [u8; 32],
         ) -> EVMBlockHeader;
 
+        fn evm_try_get_latest_block_hash(result: &mut CrossBoundaryResult) -> [u8; 32];
+
         fn evm_try_get_tx_by_hash(
             result: &mut CrossBoundaryResult,
             tx_hash: [u8; 32],

src/init.cpp

@@ -2044,6 +2044,32 @@ bool AppInitMain(InitInterfaces& interfaces)
                 break;
             }
 
+            // State consistency check is skipped for regtest (EVM state can be initialized with state input)
+            if (Params().NetworkIDString() != CBaseChainParams::REGTEST)  {
+                // Check that EVM db and DVM db states are consistent
+                auto res = XResultValueLogged(evm_try_get_latest_block_hash(result));
+                if (res) {
+                    // After EVM activation
+                    auto evmBlockHash = uint256::FromByteArray(*res).GetHex();
+                    auto dvmBlockHash = pcustomcsview->GetVMDomainBlockEdge(VMDomainEdge::EVMToDVM, evmBlockHash);
+                    if (!dvmBlockHash.val.has_value()) {
+                        strLoadError = _("Unable to get DVM block hash from latest EVM block hash, inconsistent chainstate detected. "
+                                        "This may be due to corrupted block databases between DVM and EVM, and you will need to "
+                                        "rebuild the database using -reindex.").translated;
+                        break;
+                    }
+                    CBlockIndex *pindex = LookupBlockIndex(uint256S(*dvmBlockHash.val));
+                    auto dvmBlockHeight = pindex->nHeight;
+
+                    if (dvmBlockHeight != ::ChainActive().Tip()->nHeight) {
+                        strLoadError = _("Inconsistent chainstate detected between DVM block database and EVM block database. "
+                                        "This may be due to corrupted block databases between DVM and EVM, and you will need to "
+                                        "rebuild the database using -reindex.").translated;
+                        break;
+                    }
+                }
+            }
+
             fLoaded = true;
             LogPrintf(" block index %15dms\n", GetTimeMillis() - load_block_index_start_time);
         } while(false);