Prune Merkle tree of bridge pool

.changelog/unreleased/improvements/2110-prune-bp-merkle-tree.md

@@ -0,0 +1,2 @@
+- Prune merkle tree of bridge pool
+  ([\#2110](https://github.com/anoma/namada/issues/2110))

apps/src/lib/node/ledger/storage/mod.rs

@@ -55,14 +55,18 @@ mod tests {
 
     use itertools::Itertools;
     use namada::core::ledger::masp_conversions::update_allowed_conversions;
+    use namada::eth_bridge::storage::proof::BridgePoolRootProof;
+    use namada::ledger::eth_bridge::storage::bridge_pool;
     use namada::ledger::gas::STORAGE_ACCESS_GAS_PER_BYTE;
     use namada::ledger::ibc::storage::ibc_key;
     use namada::ledger::parameters::{EpochDuration, Parameters};
     use namada::ledger::storage::write_log::WriteLog;
     use namada::ledger::storage::{types, StoreType, WlStorage};
     use namada::ledger::storage_api::{self, StorageWrite};
     use namada::types::chain::ChainId;
+    use namada::types::ethereum_events::Uint;
     use namada::types::hash::Hash;
+    use namada::types::keccak::KeccakHash;
     use namada::types::storage::{BlockHash, BlockHeight, Key};
     use namada::types::time::DurationSecs;
     use namada::types::{address, storage, token};
@@ -487,6 +491,11 @@ mod tests {
             let value_bytes = types::encode(&storage.block.height);
             storage.write(&key, value_bytes)?;
         }
+        let key = bridge_pool::get_signed_root_key();
+        let root_proof =
+            BridgePoolRootProof::new((KeccakHash::default(), Uint::default()));
+        let bytes = types::encode(&root_proof);
+        storage.write(&key, bytes)?;
 
         // Update and commit
         let hash = BlockHash::default();
@@ -581,6 +590,13 @@ mod tests {
             None,
             Some(5),
         );
+        let signed_root_key = bridge_pool::get_signed_root_key();
+        let nonce = Uint::default();
+        let root_proof =
+            BridgePoolRootProof::new((KeccakHash::default(), nonce));
+        let bytes = types::encode(&root_proof);
+        storage.write(&signed_root_key, bytes).unwrap();
+
         storage
             .begin_block(BlockHash::default(), BlockHeight(1))
             .expect("begin_block failed");
@@ -606,6 +622,12 @@ mod tests {
             .write(&key, types::encode(&value))
             .expect("write failed");
 
+        let nonce = nonce + 1;
+        let root_proof =
+            BridgePoolRootProof::new((KeccakHash::default(), nonce));
+        let bytes = types::encode(&root_proof);
+        storage.write(&signed_root_key, bytes).unwrap();
+
         storage.block.epoch = storage.block.epoch.next();
         storage.block.pred_epochs.new_epoch(BlockHeight(6));
         let batch = PersistentStorage::batch();
@@ -617,6 +639,13 @@ mod tests {
         storage
             .begin_block(BlockHash::default(), BlockHeight(11))
             .expect("begin_block failed");
+
+        let nonce = nonce + 1;
+        let root_proof =
+            BridgePoolRootProof::new((KeccakHash::default(), nonce));
+        let bytes = types::encode(&root_proof);
+        storage.write(&signed_root_key, bytes).unwrap();
+
         storage.block.epoch = storage.block.epoch.next();
         storage.block.pred_epochs.new_epoch(BlockHeight(11));
         let batch = PersistentStorage::batch();
@@ -630,7 +659,32 @@ mod tests {
             "The tree at Height 5 shouldn't be able to be restored"
         );
         let result = storage.get_merkle_tree(6.into(), Some(StoreType::Ibc));
-        assert!(result.is_ok(), "The tree should be restored");
+        assert!(result.is_ok(), "The ibc tree should be restored");
+        let result =
+            storage.get_merkle_tree(6.into(), Some(StoreType::BridgePool));
+        assert!(result.is_ok(), "The bridge pool tree should be restored");
+
+        storage
+            .begin_block(BlockHash::default(), BlockHeight(12))
+            .expect("begin_block failed");
+
+        let nonce = nonce + 1;
+        let root_proof =
+            BridgePoolRootProof::new((KeccakHash::default(), nonce));
+        let bytes = types::encode(&root_proof);
+        storage.write(&signed_root_key, bytes).unwrap();
+        storage.block.epoch = storage.block.epoch.next();
+        storage.block.pred_epochs.new_epoch(BlockHeight(12));
+        let batch = PersistentStorage::batch();
+        storage.commit_block(batch).expect("commit failed");
+
+        // ibc tree should be able to be restored
+        let result = storage.get_merkle_tree(6.into(), Some(StoreType::Ibc));
+        assert!(result.is_ok(), "The ibc tree should be restored");
+        // bridge pool tree should be pruned because of the nonce
+        let result =
+            storage.get_merkle_tree(6.into(), Some(StoreType::BridgePool));
+        assert!(result.is_err(), "The bridge pool tree should be pruned");
     }
 
     /// Test the prefix iterator with RocksDB.

apps/src/lib/node/ledger/storage/rocksdb.rs

@@ -50,6 +50,8 @@ use data_encoding::HEXLOWER;
 use itertools::Either;
 use namada::core::ledger::masp_conversions::ConversionState;
 use namada::core::types::ethereum_structs;
+use namada::eth_bridge::storage::proof::BridgePoolRootProof;
+use namada::ledger::eth_bridge::storage::bridge_pool;
 use namada::ledger::storage::merkle_tree::{
     base_tree_key_prefix, subtree_key_prefix,
 };
@@ -58,6 +60,7 @@ use namada::ledger::storage::{
     types, BlockStateRead, BlockStateWrite, DBIter, DBWriteBatch, Error,
     MerkleTreeStoresRead, Result, StoreType, DB,
 };
+use namada::types::ethereum_events::Uint;
 use namada::types::internal::TxQueue;
 use namada::types::storage::{
     BlockHeight, BlockResults, Epoch, EthEventsQueue, Header, Key, KeySeg,
@@ -1446,6 +1449,23 @@ impl DB for RocksDB {
         Ok(())
     }
 
+    fn read_bridge_pool_signed_nonce(
+        &self,
+        height: BlockHeight,
+        last_height: BlockHeight,
+    ) -> Result<Uint> {
+        let nonce_key = bridge_pool::get_signed_root_key();
+        let bytes = if height == BlockHeight(0) || height >= last_height {
+            self.read_subspace_val(&nonce_key)?
+        } else {
+            self.read_subspace_val_with_height(&nonce_key, height, last_height)?
+        };
+        let bytes = bytes.expect("Signed root should exist");
+        let bp_root_proof = BridgePoolRootProof::try_from_slice(&bytes)
+            .map_err(Error::BorshCodingError)?;
+        Ok(bp_root_proof.data.1)
+    }
+
     fn write_replay_protection_entry(
         &mut self,
         batch: &mut Self::WriteBatch,

core/src/ledger/storage/mockdb.rs

@@ -18,6 +18,7 @@ use super::{
 };
 use crate::ledger::masp_conversions::ConversionState;
 use crate::ledger::storage::types::{self, KVBytes, PrefixIterator};
+use crate::types::ethereum_events::Uint;
 use crate::types::ethereum_structs;
 use crate::types::hash::Hash;
 use crate::types::internal::TxQueue;
@@ -588,6 +589,14 @@ impl DB for MockDB {
         Ok(())
     }
 
+    fn read_bridge_pool_signed_nonce(
+        &self,
+        _height: BlockHeight,
+        _last_height: BlockHeight,
+    ) -> Result<Uint> {
+        Ok(Uint::default())
+    }
+
     fn write_replay_protection_entry(
         &mut self,
         _batch: &mut Self::WriteBatch,

core/src/ledger/storage/mod.rs

@@ -39,6 +39,7 @@ use crate::ledger::storage::merkle_tree::{
 use crate::tendermint::merkle::proof::ProofOps;
 use crate::types::address::{Address, EstablishedAddressGen, InternalAddress};
 use crate::types::chain::{ChainId, CHAIN_ID_LENGTH};
+use crate::types::ethereum_events::Uint;
 use crate::types::ethereum_structs;
 use crate::types::hash::{Error as HashError, Hash};
 use crate::types::internal::{ExpiredTxsQueue, TxQueue};
@@ -357,6 +358,13 @@ pub trait DB: std::fmt::Debug {
         pruned_epoch: Epoch,
     ) -> Result<()>;
 
+    /// Read the signed nonce of Bridge Pool
+    fn read_bridge_pool_signed_nonce(
+        &self,
+        height: BlockHeight,
+        last_height: BlockHeight,
+    ) -> Result<Uint>;
+
     /// Write a replay protection entry
     fn write_replay_protection_entry(
         &mut self,
@@ -1144,32 +1152,40 @@ where
         if self.block.epoch.0 == 0 {
             return Ok(());
         }
-        if let Some(limit) = self.storage_read_past_height_limit {
-            if self.get_last_block_height().0 <= limit {
-                return Ok(());
-            }
-
-            let min_height = (self.get_last_block_height().0 - limit).into();
-            if let Some(epoch) = self.block.pred_epochs.get_epoch(min_height) {
-                if epoch.0 == 0 {
-                    return Ok(());
-                }
-                // Remove stores at the previous epoch because the Merkle tree
-                // stores at the starting height of the epoch would be used to
-                // restore stores at a height (> min_height) in the epoch
-                for st in StoreType::iter_provable() {
-                    self.db.prune_merkle_tree_store(batch, st, epoch.prev())?;
-                }
-            }
-        }
-        // remove non-provable stores at the previous epoch
+        // Prune non-provable stores at the previous epoch
         for st in StoreType::iter_non_provable() {
             self.db.prune_merkle_tree_store(
                 batch,
                 st,
                 self.block.epoch.prev(),
             )?;
         }
+        // Prune provable stores
+        let oldest_epoch = self.get_oldest_epoch();
+        if oldest_epoch.0 > 0 {
+            // Remove stores at the previous epoch because the Merkle tree
+            // stores at the starting height of the epoch would be used to
+            // restore stores at a height (> oldest_height) in the epoch
+            for st in StoreType::iter_provable() {
+                self.db.prune_merkle_tree_store(
+                    batch,
+                    st,
+                    oldest_epoch.prev(),
+                )?;
+            }
+
+            // Prune the BridgePool subtree stores with invalid nonce
+            let mut epoch = self.get_oldest_epoch_with_valid_nonce()?;
+            while oldest_epoch < epoch {
+                epoch = epoch.prev();
+                self.db.prune_merkle_tree_store(
+                    batch,
+                    &StoreType::BridgePool,
+                    epoch,
+                )?;
+            }
+        }
+
         Ok(())
     }
 
@@ -1182,6 +1198,49 @@ where
             .unwrap_or_default()
     }
 
+    /// Get the oldest epoch where we can read a value
+    pub fn get_oldest_epoch(&self) -> Epoch {
+        let oldest_height = match self.storage_read_past_height_limit {
+            Some(limit) if limit < self.get_last_block_height().0 => {
+                (self.get_last_block_height().0 - limit).into()
+            }
+            _ => BlockHeight(1),
+        };
+        self.block
+            .pred_epochs
+            .get_epoch(oldest_height)
+            .unwrap_or_default()
+    }
+
+    /// Get oldest epoch which has the valid signed nonce of the bridge pool
+    pub fn get_oldest_epoch_with_valid_nonce(&self) -> Result<Epoch> {
+        let last_height = self.get_last_block_height();
+        let current_nonce = self
+            .db
+            .read_bridge_pool_signed_nonce(last_height, last_height)?;
+        let (mut epoch, _) = self.get_last_epoch();
+        // We don't need to check the older epochs because their Merkle tree
+        // snapshots have been already removed
+        let oldest_epoch = self.get_oldest_epoch();
+        // Look up the last valid epoch which has the previous nonce of the
+        // current one. It has the previous nonce, but it was
+        // incremented during the epoch.
+        while 0 < epoch.0 && oldest_epoch <= epoch {
+            epoch = epoch.prev();
+            let height =
+                match self.block.pred_epochs.get_start_height_of_epoch(epoch) {
+                    Some(h) => h,
+                    None => continue,
+                };
+            let nonce =
+                self.db.read_bridge_pool_signed_nonce(height, last_height)?;
+            if nonce < current_nonce {
+                break;
+            }
+        }
+        Ok(epoch)
+    }
+
     /// Check it the given transaction's hash is already present in storage
     pub fn has_replay_protection_entry(&self, hash: &Hash) -> Result<bool> {
         self.db.has_replay_protection_entry(hash)