Merge pull request #143 from vacp2p/schouhy/node-state-reconstruction

Node state reconstruction from block storage

sequencer_core/src/block_store.rs

@@ -7,7 +7,7 @@ use storage::RocksDBIO;
 pub struct SequecerBlockStore {
     dbio: RocksDBIO,
     // TODO: Consider adding the hashmap to the database for faster recovery.
-    tx_hash_to_block_map: HashMap<HashType, u64>,
+    pub tx_hash_to_block_map: HashMap<HashType, u64>,
     pub genesis_id: u64,
     pub signing_key: nssa::PrivateKey,
 }
@@ -28,7 +28,7 @@ impl SequecerBlockStore {
             HashMap::new()
         };
 
-        let dbio = RocksDBIO::new(location, genesis_block)?;
+        let dbio = RocksDBIO::open_or_create(location, genesis_block)?;
 
         let genesis_id = dbio.get_meta_first_block_in_db()?;
 
@@ -71,7 +71,7 @@ impl SequecerBlockStore {
     }
 }
 
-fn block_to_transactions_map(block: &Block) -> HashMap<HashType, u64> {
+pub(crate) fn block_to_transactions_map(block: &Block) -> HashMap<HashType, u64> {
     block
         .body
         .transactions

sequencer_core/src/lib.rs

@@ -1,6 +1,8 @@
 use std::{fmt::Display, time::Instant};
 
 use anyhow::Result;
+#[cfg(feature = "testnet")]
+use common::PINATA_BASE58;
 use common::{
     HashType,
     block::HashableBlockData,
@@ -9,14 +11,16 @@ use common::{
 use config::SequencerConfig;
 use log::warn;
 use mempool::MemPool;
-use sequencer_store::SequecerChainStore;
 use serde::{Deserialize, Serialize};
 
+use crate::block_store::SequecerBlockStore;
+
+pub mod block_store;
 pub mod config;
-pub mod sequencer_store;
 
 pub struct SequencerCore {
-    pub store: SequecerChainStore,
+    pub state: nssa::V02State,
+    pub block_store: SequecerBlockStore,
     pub mempool: MemPool<EncodedTransaction>,
     pub sequencer_config: SequencerConfig,
     pub chain_height: u64,
@@ -39,6 +43,24 @@ impl std::error::Error for TransactionMalformationError {}
 
 impl SequencerCore {
     pub fn start_from_config(config: SequencerConfig) -> Self {
+        let hashable_data = HashableBlockData {
+            block_id: config.genesis_id,
+            transactions: vec![],
+            prev_block_hash: [0; 32],
+            timestamp: 0,
+        };
+
+        let signing_key = nssa::PrivateKey::try_new(config.signing_key).unwrap();
+        let genesis_block = hashable_data.into_block(&signing_key);
+
+        //Sequencer should panic if unable to open db,
+        //as fixing this issue may require actions non-native to program scope
+        let block_store = SequecerBlockStore::open_db_with_genesis(
+            &config.home.join("rocksdb"),
+            Some(genesis_block),
+            signing_key,
+        )
+        .unwrap();
         let mut initial_commitments = vec![];
 
         for init_comm_data in config.initial_commitments.clone() {
@@ -53,18 +75,47 @@ impl SequencerCore {
             initial_commitments.push(comm);
         }
 
-        Self {
+        let init_accs: Vec<(nssa::Address, u128)> = config
-            store: SequecerChainStore::new_with_genesis(
+            .initial_accounts
-                &config.home,
+            .iter()
-                config.genesis_id,
+            .map(|acc_data| (acc_data.addr.parse().unwrap(), acc_data.balance))
-                config.is_genesis_random,
+            .collect();
-                &config.initial_accounts,
+
-                &initial_commitments,
+        let mut state = nssa::V02State::new_with_genesis_accounts(&init_accs, &initial_commitments);
-                nssa::PrivateKey::try_new(config.signing_key).unwrap(),
+
-            ),
+        #[cfg(feature = "testnet")]
+        state.add_pinata_program(PINATA_BASE58.parse().unwrap());
+
+        let mut this = Self {
+            state,
+            block_store,
             mempool: MemPool::default(),
             chain_height: config.genesis_id,
             sequencer_config: config,
+        };
+
+        this.sync_state_with_stored_blocks();
+
+        this
+    }
+
+    /// If there are stored blocks ahead of the current height, this method will load and process all transaction
+    /// in them in the order they are stored. The NSSA state will be updated accordingly.
+    fn sync_state_with_stored_blocks(&mut self) {
+        let mut next_block_id = self.sequencer_config.genesis_id + 1;
+        while let Ok(block) = self.block_store.get_block_at_id(next_block_id) {
+            for encoded_transaction in block.body.transactions {
+                let transaction = NSSATransaction::try_from(&encoded_transaction).unwrap();
+                // Process transaction and update state
+                self.execute_check_transaction_on_state(transaction)
+                    .unwrap();
+                // Update the tx hash to block id map.
+                self.block_store
+                    .tx_hash_to_block_map
+                    .insert(encoded_transaction.hash(), next_block_id);
+            }
+            self.chain_height = next_block_id;
+            next_block_id += 1;
         }
     }
 
@@ -122,20 +173,17 @@ impl SequencerCore {
     ) -> Result<NSSATransaction, nssa::error::NssaError> {
         match &tx {
             NSSATransaction::Public(tx) => {
-                self.store
+                self.state
-                    .state
                     .transition_from_public_transaction(tx)
                     .inspect_err(|err| warn!("Error at transition {err:#?}"))?;
             }
             NSSATransaction::PrivacyPreserving(tx) => {
-                self.store
+                self.state
-                    .state
                     .transition_from_privacy_preserving_transaction(tx)
                     .inspect_err(|err| warn!("Error at transition {err:#?}"))?;
             }
             NSSATransaction::ProgramDeployment(tx) => {
-                self.store
+                self.state
-                    .state
                     .transition_from_program_deployment_transaction(tx)
                     .inspect_err(|err| warn!("Error at transition {err:#?}"))?;
             }
@@ -168,7 +216,6 @@ impl SequencerCore {
         }
 
         let prev_block_hash = self
-            .store
             .block_store
             .get_block_at_id(self.chain_height)?
             .header
@@ -185,9 +232,9 @@ impl SequencerCore {
             timestamp: curr_time,
         };
 
-        let block = hashable_data.into_block(&self.store.block_store.signing_key);
+        let block = hashable_data.into_block(&self.block_store.signing_key);
 
-        self.store.block_store.put_block_at_id(block)?;
+        self.block_store.put_block_at_id(block)?;
 
         self.chain_height = new_block_height;
 
@@ -205,6 +252,7 @@ impl SequencerCore {
 mod tests {
     use base58::{FromBase58, ToBase58};
     use common::test_utils::sequencer_sign_key_for_testing;
+    use nssa::PrivateKey;
 
     use crate::config::AccountInitialData;
 
@@ -305,12 +353,10 @@ mod tests {
             .unwrap();
 
         let balance_acc_1 = sequencer
-            .store
             .state
             .get_account_by_address(&nssa::Address::new(acc1_addr))
             .balance;
         let balance_acc_2 = sequencer
-            .store
             .state
             .get_account_by_address(&nssa::Address::new(acc2_addr))
             .balance;
@@ -364,7 +410,6 @@ mod tests {
         assert_eq!(
             10000,
             sequencer
-                .store
                 .state
                 .get_account_by_address(&nssa::Address::new(acc1_addr))
                 .balance
@@ -372,7 +417,6 @@ mod tests {
         assert_eq!(
             20000,
             sequencer
-                .store
                 .state
                 .get_account_by_address(&nssa::Address::new(acc2_addr))
                 .balance
@@ -541,12 +585,10 @@ mod tests {
             .unwrap();
 
         let bal_from = sequencer
-            .store
             .state
             .get_account_by_address(&nssa::Address::new(acc1))
             .balance;
         let bal_to = sequencer
-            .store
             .state
             .get_account_by_address(&nssa::Address::new(acc2))
             .balance;
@@ -645,7 +687,6 @@ mod tests {
             .produce_new_block_with_mempool_transactions()
             .unwrap();
         let block = sequencer
-            .store
             .block_store
             .get_block_at_id(current_height)
             .unwrap();
@@ -688,7 +729,6 @@ mod tests {
             .produce_new_block_with_mempool_transactions()
             .unwrap();
         let block = sequencer
-            .store
             .block_store
             .get_block_at_id(current_height)
             .unwrap();
@@ -700,10 +740,59 @@ mod tests {
             .produce_new_block_with_mempool_transactions()
             .unwrap();
         let block = sequencer
-            .store
             .block_store
             .get_block_at_id(current_height)
             .unwrap();
         assert!(block.body.transactions.is_empty());
     }
+
+    #[test]
+    fn test_restart_from_storage() {
+        let config = setup_sequencer_config();
+        let acc1_addr: nssa::Address = config.initial_accounts[0].addr.parse().unwrap();
+        let acc2_addr: nssa::Address = config.initial_accounts[1].addr.parse().unwrap();
+        let balance_to_move = 13;
+
+        // In the following code block a transaction will be processed that moves `balance_to_move`
+        // from `acc_1` to `acc_2`. The block created with that transaction will be kept stored in
+        // the temporary directory for the block storage of this test.
+        {
+            let mut sequencer = SequencerCore::start_from_config(config.clone());
+            let signing_key = PrivateKey::try_new([1; 32]).unwrap();
+
+            let tx = common::test_utils::create_transaction_native_token_transfer(
+                *acc1_addr.value(),
+                0,
+                *acc2_addr.value(),
+                balance_to_move,
+                signing_key,
+            );
+
+            sequencer.mempool.push_item(tx.clone());
+            let current_height = sequencer
+                .produce_new_block_with_mempool_transactions()
+                .unwrap();
+            let block = sequencer
+                .block_store
+                .get_block_at_id(current_height)
+                .unwrap();
+            assert_eq!(block.body.transactions, vec![tx.clone()]);
+        }
+
+        // Instantiating a new sequencer from the same config. This should load the existing block
+        // with the above transaction and update the state to reflect that.
+        let sequencer = SequencerCore::start_from_config(config.clone());
+        let balance_acc_1 = sequencer.state.get_account_by_address(&acc1_addr).balance;
+        let balance_acc_2 = sequencer.state.get_account_by_address(&acc2_addr).balance;
+
+        // Balances should be consistent with the stored block
+        assert_eq!(
+            balance_acc_1,
+            config.initial_accounts[0].balance - balance_to_move
+        );
+        assert_eq!(
+            balance_acc_2,
+            config.initial_accounts[1].balance + balance_to_move
+        );
+    }
 }

sequencer_rpc/src/process.rs

@@ -104,10 +104,7 @@ impl JsonHandler {
         let block = {
             let state = self.sequencer_state.lock().await;
 
-            state
+            state.block_store.get_block_at_id(get_block_req.block_id)?
-                .store
-                .block_store
-                .get_block_at_id(get_block_req.block_id)?
         };
 
         let helperstruct = GetBlockDataResponse {
@@ -123,7 +120,7 @@ impl JsonHandler {
         let genesis_id = {
             let state = self.sequencer_state.lock().await;
 
-            state.store.block_store.genesis_id
+            state.block_store.genesis_id
         };
 
         let helperstruct = GetGenesisIdResponse { genesis_id };
@@ -176,7 +173,7 @@ impl JsonHandler {
 
         let balance = {
             let state = self.sequencer_state.lock().await;
-            let account = state.store.state.get_account_by_address(&address);
+            let account = state.state.get_account_by_address(&address);
             account.balance
         };
 
@@ -203,7 +200,7 @@ impl JsonHandler {
 
             addresses
                 .into_iter()
-                .map(|addr| state.store.state.get_account_by_address(&addr).nonce)
+                .map(|addr| state.state.get_account_by_address(&addr).nonce)
                 .collect()
         };
 
@@ -225,7 +222,7 @@ impl JsonHandler {
         let account = {
             let state = self.sequencer_state.lock().await;
 
-            state.store.state.get_account_by_address(&address)
+            state.state.get_account_by_address(&address)
         };
 
         let helperstruct = GetAccountResponse { account };
@@ -246,7 +243,6 @@ impl JsonHandler {
         let transaction = {
             let state = self.sequencer_state.lock().await;
             state
-                .store
                 .block_store
                 .get_transaction_by_hash(hash)
                 .map(|tx| borsh::to_vec(&tx).unwrap())
@@ -265,7 +261,6 @@ impl JsonHandler {
         let membership_proof = {
             let state = self.sequencer_state.lock().await;
             state
-                .store
                 .state
                 .get_proof_for_commitment(&get_proof_req.commitment)
         };

sequencer_runner/Cargo.toml

@@ -22,6 +22,7 @@ path = "../sequencer_rpc"
 
 [dependencies.sequencer_core]
 path = "../sequencer_core"
+features = ["testnet"]
 
 [dependencies.common]
 path = "../common"

storage/src/lib.rs

@@ -44,7 +44,7 @@ pub struct RocksDBIO {
 }
 
 impl RocksDBIO {
-    pub fn new(path: &Path, start_block: Option<Block>) -> DbResult<Self> {
+    pub fn open_or_create(path: &Path, start_block: Option<Block>) -> DbResult<Self> {
         let mut cf_opts = Options::default();
         cf_opts.set_max_write_buffer_number(16);
         //ToDo: Add more column families for different data
@@ -74,7 +74,6 @@ impl RocksDBIO {
             let block_id = block.header.block_id;
             dbio.put_meta_first_block_in_db(block)?;
             dbio.put_meta_is_first_block_set()?;
-
             dbio.put_meta_last_block_in_db(block_id)?;
 
             Ok(dbio)