Light History Store for full nodes

Use the light history store for full node clients
Necessary changes to use the history interface
Several changes to usage of the history store

blockchain/Cargo.toml

@@ -22,6 +22,7 @@ workspace = true
 [dependencies]
 futures = { package = "futures-util", version = "0.3" }
 hex = "0.4"
+itertools = "0.10.3"
 log = { package = "tracing", version = "0.1", features = ["log"] }
 parking_lot = "0.12"
 prometheus-client = { version = "0.22.2", optional = true }

blockchain/src/block_production/mod.rs

@@ -150,7 +150,7 @@ impl BlockProducer {
 
         let (history_root, _) = blockchain
             .history_store
-            .add_to_history(&mut txn, Policy::epoch_at(block_number), &hist_txs)
+            .add_to_history(&mut txn, block_number, &hist_txs)
             .expect("Failed to compute history root during block production.");
 
         // Not strictly necessary to drop the lock here, but sign as well as compress might be somewhat expensive
@@ -327,7 +327,7 @@ impl BlockProducer {
 
         macro_block.header.history_root = blockchain
             .history_store
-            .add_to_history(&mut txn, Policy::epoch_at(block_number), &hist_txs)
+            .add_to_history(&mut txn, block_number, &hist_txs)
             .expect("Failed to compute history root during block production.")
             .0;
 

blockchain/src/blockchain/accounts.rs

@@ -10,7 +10,6 @@ use nimiq_primitives::{
     trie::{error::IncompleteTrie, trie_diff::TrieDiff, trie_proof::TrieProof},
 };
 use nimiq_serde::Deserialize;
-use nimiq_transaction::historic_transaction::HistoricTransaction;
 use nimiq_trie::WriteTransactionProxy;
 
 use crate::Blockchain;
@@ -58,18 +57,9 @@ impl Blockchain {
                 // as rebranching across this block is not possible.
                 self.chain_store.clear_revert_infos(txn.raw());
 
-                // Store the transactions and the inherents into the History tree.
-                let hist_txs = HistoricTransaction::from(
-                    self.network_id,
-                    macro_block.header.block_number,
-                    macro_block.header.timestamp,
-                    vec![],
-                    inherents,
-                    vec![],
-                );
                 let total_tx_size = self
                     .history_store
-                    .add_to_history(txn.raw(), macro_block.epoch_number(), &hist_txs)
+                    .add_block(txn.raw(), block, inherents)
                     .expect("Failed to store history")
                     .1;
 
@@ -136,21 +126,9 @@ impl Blockchain {
                     &revert_info,
                 );
 
-                // Store the transactions and the inherents into the History tree.
-                let hist_txs = HistoricTransaction::from(
-                    self.network_id,
-                    micro_block.header.block_number,
-                    micro_block.header.timestamp,
-                    body.transactions.clone(),
-                    inherents,
-                    body.equivocation_proofs
-                        .iter()
-                        .map(|proof| proof.locator())
-                        .collect(),
-                );
                 let total_tx_size = self
                     .history_store
-                    .add_to_history(txn.raw(), micro_block.epoch_number(), &hist_txs)
+                    .add_block(txn.raw(), block, inherents)
                     .expect("Failed to store history")
                     .1;
 
@@ -221,20 +199,10 @@ impl Blockchain {
             panic!("Failed to revert {block} - {e:?}");
         }
 
-        // Remove the transactions from the History tree. For this you only need to calculate the
-        // number of transactions that you want to remove.
-        let num_txs = HistoricTransaction::count(
-            body.transactions.len(),
-            &inherents,
-            body.equivocation_proofs
-                .iter()
-                .map(|proof| proof.locator())
-                .collect(),
-        );
-        let (_, total_size) = self
+        let total_size = self
             .history_store
-            .remove_partial_history(txn.raw(), block.epoch_number(), num_txs)
-            .expect("Failed to remove partial history");
+            .remove_block(txn.raw(), block, inherents)
+            .expect("Failed to remove block from history");
 
         Ok(total_size)
     }

blockchain/src/blockchain/blockchain.rs

@@ -19,7 +19,8 @@ use tokio::sync::broadcast::{channel as broadcast, Sender as BroadcastSender};
 use crate::chain_metrics::BlockchainMetrics;
 use crate::{
     blockchain_state::BlockchainState, chain_store::ChainStore, history::HistoryStore,
-    interface::HistoryInterface, reward::genesis_parameters,
+    interface::HistoryInterface, light_history_store::LightHistoryStore,
+    reward::genesis_parameters,
 };
 
 const BROADCAST_MAX_CAPACITY: usize = 256;
@@ -67,13 +68,17 @@ pub struct BlockchainConfig {
     /// Maximum number of epochs (other than the current one) that the ChainStore will store fully.
     /// Epochs older than this number will be pruned.
     pub max_epochs_stored: u32,
+    /// The history store that is used by the full blockchain.
+    /// If this is set to true, the light history store is used.
+    pub light_history_store: bool,
 }
 
 impl Default for BlockchainConfig {
     fn default() -> Self {
         Self {
             keep_history: true,
             max_epochs_stored: Policy::MIN_EPOCHS_STORED,
+            light_history_store: false,
         }
     }
 }
@@ -131,14 +136,12 @@ impl Blockchain {
         }
 
         let chain_store = ChainStore::new(env.clone());
-        let history_store = HistoryStore::new(env.clone());
 
         Ok(match chain_store.get_head(None) {
             Some(head_hash) => Blockchain::load(
                 env,
                 config,
                 chain_store,
-                history_store,
                 time,
                 network_id,
                 genesis_block,
@@ -148,7 +151,6 @@ impl Blockchain {
                 env,
                 config,
                 chain_store,
-                history_store,
                 time,
                 network_id,
                 genesis_block,
@@ -162,7 +164,7 @@ impl Blockchain {
         env: DatabaseProxy,
         config: BlockchainConfig,
         chain_store: ChainStore,
-        history_store: HistoryStore,
+
         time: Arc<OffsetTime>,
         network_id: NetworkId,
         genesis_block: Block,
@@ -262,6 +264,14 @@ impl Blockchain {
         let (tx_fork, _rx_fork) = broadcast(BROADCAST_MAX_CAPACITY);
         let (tx_log, _rx_log) = broadcast(BROADCAST_MAX_CAPACITY);
 
+        let history_store = if config.light_history_store {
+            Box::new(LightHistoryStore::new(env.clone(), network_id))
+                as Box<dyn HistoryInterface + Sync + Send>
+        } else {
+            Box::new(HistoryStore::new(env.clone(), network_id))
+                as Box<dyn HistoryInterface + Sync + Send>
+        };
+
         Ok(Blockchain {
             env,
             config,
@@ -271,7 +281,7 @@ impl Blockchain {
             fork_notifier: tx_fork,
             log_notifier: tx_log,
             chain_store,
-            history_store: Box::new(history_store) as Box<dyn HistoryInterface + Sync + Send>,
+            history_store,
             state: BlockchainState {
                 accounts,
                 main_chain,
@@ -297,7 +307,7 @@ impl Blockchain {
         env: DatabaseProxy,
         config: BlockchainConfig,
         chain_store: ChainStore,
-        history_store: HistoryStore,
+
         time: Arc<OffsetTime>,
         network_id: NetworkId,
         genesis_block: Block,
@@ -327,6 +337,14 @@ impl Blockchain {
         let (tx_fork, _rx_fork) = broadcast(BROADCAST_MAX_CAPACITY);
         let (tx_log, _rx_log) = broadcast(BROADCAST_MAX_CAPACITY);
 
+        let history_store = if config.light_history_store {
+            Box::new(LightHistoryStore::new(env.clone(), network_id))
+                as Box<dyn HistoryInterface + Sync + Send>
+        } else {
+            Box::new(HistoryStore::new(env.clone(), network_id))
+                as Box<dyn HistoryInterface + Sync + Send>
+        };
+
         Ok(Blockchain {
             env,
             config,
@@ -336,7 +354,7 @@ impl Blockchain {
             fork_notifier: tx_fork,
             log_notifier: tx_log,
             chain_store,
-            history_store: Box::new(history_store) as Box<dyn HistoryInterface + Sync + Send>,
+            history_store,
             state: BlockchainState {
                 accounts,
                 macro_info: main_chain.clone(),

blockchain/src/blockchain/history_sync.rs

@@ -1,5 +1,6 @@
-use std::error::Error;
+use std::{collections::HashMap, error::Error};
 
+use itertools::Itertools;
 use nimiq_account::{BlockLogger, BlockState};
 use nimiq_block::{Block, BlockError};
 use nimiq_blockchain_interface::{
@@ -96,24 +97,41 @@ impl Blockchain {
     /// Extends the current chain with transactions from the validity sync process
     pub fn extend_validity_sync(
         this: RwLockUpgradableReadGuard<Blockchain>,
-        epoch: u32,
         history: &[HistoricTransaction],
     ) -> Option<Blake2bHash> {
         let mut txn = this.write_transaction();
 
-        // Store the new historic transactions into the History tree.
-        if let Some((root, _)) = this.history_store.add_to_history(&mut txn, epoch, history) {
-            txn.commit();
+        let mut txns_per_block: HashMap<u32, Vec<HistoricTransaction>> = HashMap::new();
+
+        for txn in history {
+            let bn = txn.block_number;
+
+            if let Some(txns) = txns_per_block.get_mut(&bn) {
+                txns.push(txn.clone());
+            } else {
+                let new_txns = vec![txn.clone()];
+                txns_per_block.insert(bn, new_txns);
+            }
+        }
+
+        let mut root = Blake2bHash::default();
+        for bn in txns_per_block.keys().sorted() {
+            let hist_txs = txns_per_block.get(bn).unwrap();
+
+            root = this
+                .history_store
+                .add_to_history(&mut txn, *bn, hist_txs)
+                .expect("Pushing txns to the history store failed")
+                .0;
 
             debug!(
-                num_transactions = history.len(),
+                num_transactions = hist_txs.len(),
+                block = bn,
                 "Pushed txns to the history store during validity sync"
             );
-            Some(root)
-        } else {
-            txn.abort();
-            None
         }
+        txn.commit();
+        Some(root)
     }
 
     /// Extends the current chain with a macro block (election or checkpoint) during history sync.
@@ -347,7 +365,7 @@ impl Blockchain {
         // Store the new historic transactions into the History tree.
         this.history_store.add_to_history(
             &mut txn,
-            block.epoch_number(),
+            block.block_number(),
             &history[first_new_hist_tx..],
         );
 
@@ -370,7 +388,7 @@ impl Blockchain {
 
         let wanted_history_root = this
             .history_store
-            .get_history_tree_root(block.epoch_number(), Some(&txn))
+            .get_history_tree_root(block.block_number(), Some(&txn))
             .ok_or(PushError::InvalidBlock(BlockError::InvalidHistoryRoot))?;
 
         if *block.history_root() != wanted_history_root {

blockchain/src/blockchain/verify.rs

@@ -169,7 +169,7 @@ impl Blockchain {
         // Verify the history root when we have the full transaction history of the epoch.
         let real_history_root = self
             .history_store
-            .get_history_tree_root(block.epoch_number(), Some(txn))
+            .get_history_tree_root(block.block_number(), Some(txn))
             .ok_or_else(|| {
                 error!(
                     %block,

blockchain/src/blockchain/wrappers.rs

@@ -192,32 +192,6 @@ impl Blockchain {
             .release_balance(account, transaction, reserved_balance, None)
     }
 
-    /// Checks if we have seen some transaction with this hash inside the a validity window.
-    pub fn tx_in_validity_window(
-        &self,
-        tx_hash: &Blake2bHash,
-        validity_window_start: u32,
-        txn_opt: Option<&DBTransaction>,
-    ) -> bool {
-        // Get a vector with all transactions corresponding to the given hash.
-        let ext_hash_vec = self.history_store.get_hist_tx_by_hash(tx_hash, txn_opt);
-
-        // If the vector is empty then we have never seen a transaction with this hash.
-        if ext_hash_vec.is_empty() {
-            return false;
-        }
-
-        for hist_tx in ext_hash_vec {
-            // If the transaction is inside the validity window, return true.
-            if hist_tx.block_number >= validity_window_start {
-                return true;
-            }
-        }
-
-        // If we didn't see any transaction inside the validity window then we can return false.
-        false
-    }
-
     /// Checks if we have seen some transaction with this hash inside the validity window. This is
     /// used to prevent replay attacks.
     pub fn contains_tx_in_validity_window(
@@ -228,7 +202,8 @@ impl Blockchain {
         let max_block_number = self
             .block_number()
             .saturating_sub(Policy::transaction_validity_window_blocks());
-        self.tx_in_validity_window(tx_hash, max_block_number, txn_opt)
+        self.history_store
+            .tx_in_validity_window(tx_hash, max_block_number, txn_opt)
     }
 
     pub fn staking_contract_address(&self) -> Address {
@@ -265,10 +240,11 @@ impl Blockchain {
             true
         } else {
             // We can enforce the validity window when our history store root equals our head one.
+
             *self.head().history_root()
                 == self
                     .history_store
-                    .get_history_tree_root(Policy::epoch_at(self.block_number()), None)
+                    .get_history_tree_root(self.block_number(), None)
                     .unwrap()
         }
     }