fix: accumulated block data bitmap now contains current stxo indexes (#…

…3109)

<!--- Provide a general summary of your changes in the Title above -->
## Description
<!--- Describe your changes in detail -->
Greatly reduces the amount of data stored per block by only storing the
spent indexes at a specific height instead of the entire spent bitmap.

The block chain database now stores a single full deleted bitmap at the
tip.

Breaking changes:

- [x] Blockchain Database (database must be reynced)
- [ ] Blockchain Consensus (hard/soft fork)
- [ ] P2P Network (some protocols are no longer compatible)
- [ ] GRPC (GRPC clients need to be updated)

## Motivation and Context
<!--- Why is this change required? What problem does it solve? -->
<!--- If it fixes an open issue, please link to the issue here. -->
Bytes of storage per block reduced. At height 8033, the savings are 
130mb on the new database. As the STXO set increases, the size of 
the deleted bitmap typically increases (depends on compressibility of the data
in the bitmap), previously resulting in many megabytes of data per block.
This is no longer the case.

As a result of only having the tip bitmap, we can only calculate the output_mr for the next block 
from the tip. In practice this is all that is needed, however some unit tests needed to be updated.

## How Has This Been Tested?
<!--- Please describe in detail how you tested your changes. -->
<!--- Include details of your testing environment, and the tests you ran to -->
<!--- see how your change affects other areas of the code, etc. -->
- Some unit tests updated.
- Base node archival sync
- Currently not able to test pruned sync since it is broken (#3082)
- Mined monero and sha3 blocks 

## Checklist:
<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
* [x] I'm merging against the `development` branch.
* [x] I have squashed my commits into a single commit.

BREAKING CHANGE: Blockchain Database (database must be resynced)

base_layer/core/src/base_node/state_machine_service/states/horizon_state_sync/horizon_state_synchronization.rs

@@ -206,9 +206,9 @@ impl<'a, B: BlockchainBackend + 'static> HorizonStateSynchronization<'a, B> {
                 let kernel_pruned_set = block_data.dissolve().0;
                 let mut kernel_mmr = MerkleMountainRange::<HashDigest, _>::new(kernel_pruned_set);
 
-                let mut kernel_sum = HomomorphicCommitment::default();
+                // let mut kernel_sum = HomomorphicCommitment::default();
                 for kernel in kernels.drain(..) {
-                    kernel_sum = &kernel.excess + &kernel_sum;
+                    // kernel_sum = &kernel.excess + &kernel_sum;
                     kernel_mmr.push(kernel.hash())?;
                 }
 
@@ -323,7 +323,7 @@ impl<'a, B: BlockchainBackend + 'static> HorizonStateSynchronization<'a, B> {
         let block_data = db
             .fetch_block_accumulated_data(current_header.header().prev_hash.clone())
             .await?;
-        let (_, output_pruned_set, rp_pruned_set, mut deleted) = block_data.dissolve();
+        let (_, output_pruned_set, rp_pruned_set, mut full_bitmap) = block_data.dissolve();
 
         let mut output_mmr = MerkleMountainRange::<HashDigest, _>::new(output_pruned_set);
         let mut witness_mmr = MerkleMountainRange::<HashDigest, _>::new(rp_pruned_set);
@@ -416,13 +416,34 @@ impl<'a, B: BlockchainBackend + 'static> HorizonStateSynchronization<'a, B> {
                         witness_mmr.push(hash)?;
                     }
 
-                    // Add in the changes
-                    let bitmap = Bitmap::deserialize(&diff_bitmap);
-                    deleted.or_inplace(&bitmap);
-                    deleted.run_optimize();
+                    // Check that the difference bitmap is excessively large. Bitmap::deserialize panics if greater than
+                    // isize::MAX, however isize::MAX is still an inordinate amount of data. An
+                    // arbitrary 4 MiB limit is used.
+                    const MAX_DIFF_BITMAP_BYTE_LEN: usize = 4 * 1024 * 1024;
+                    if diff_bitmap.len() > MAX_DIFF_BITMAP_BYTE_LEN {
+                        return Err(HorizonSyncError::IncorrectResponse(format!(
+                            "Received difference bitmap (size = {}) that exceeded the maximum size limit of {} from \
+                             peer {}",
+                            diff_bitmap.len(),
+                            MAX_DIFF_BITMAP_BYTE_LEN,
+                            self.sync_peer.peer_node_id()
+                        )));
+                    }
+
+                    let diff_bitmap = Bitmap::try_deserialize(&diff_bitmap).ok_or_else(|| {
+                        HorizonSyncError::IncorrectResponse(format!(
+                            "Peer {} sent an invalid difference bitmap",
+                            self.sync_peer.peer_node_id()
+                        ))
+                    })?;
+
+                    // Merge the differences into the final bitmap so that we can commit to the entire spend state
+                    // in the output MMR
+                    full_bitmap.or_inplace(&diff_bitmap);
+                    full_bitmap.run_optimize();
 
                     let pruned_output_set = output_mmr.get_pruned_hash_set()?;
-                    let output_mmr = MutableMmr::<HashDigest, _>::new(pruned_output_set.clone(), deleted.clone())?;
+                    let output_mmr = MutableMmr::<HashDigest, _>::new(pruned_output_set.clone(), full_bitmap.clone())?;
 
                     let mmr_root = output_mmr.get_merkle_root()?;
                     if mmr_root != current_header.header().output_mr {
@@ -450,13 +471,14 @@ impl<'a, B: BlockchainBackend + 'static> HorizonStateSynchronization<'a, B> {
                             .map_err(|err| HorizonSyncError::InvalidRangeProof(o.hash().to_hex(), err.to_string()))?;
                     }
 
+                    txn.update_deleted_bitmap(diff_bitmap.clone());
                     txn.update_pruned_hash_set(MmrTree::Utxo, current_header.hash().clone(), pruned_output_set);
                     txn.update_pruned_hash_set(
                         MmrTree::Witness,
                         current_header.hash().clone(),
                         witness_mmr.get_pruned_hash_set()?,
                     );
-                    txn.update_deleted_with_diff(current_header.hash().clone(), output_mmr.deleted().clone());
+                    txn.update_block_accumulated_data_with_deleted_diff(current_header.hash().clone(), diff_bitmap);
 
                     txn.commit().await?;
 

base_layer/core/src/chain_storage/accumulated_data.rs

@@ -52,12 +52,6 @@ use tari_mmr::{pruned_hashset::PrunedHashSet, ArrayLike};
 
 const LOG_TARGET: &str = "c::bn::acc_data";
 
-#[derive(Debug)]
-// Helper struct to serialize and deserialize Bitmap
-pub struct DeletedBitmap {
-    pub(super) deleted: Bitmap,
-}
-
 #[derive(Debug, Serialize, Deserialize)]
 pub struct BlockAccumulatedData {
     pub(super) kernels: PrunedHashSet,
@@ -84,7 +78,6 @@ impl BlockAccumulatedData {
         }
     }
 
-    #[inline(always)]
     pub fn deleted(&self) -> &Bitmap {
         &self.deleted.deleted
     }
@@ -116,7 +109,7 @@ impl Display for BlockAccumulatedData {
     fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), fmt::Error> {
         write!(
             f,
-            "{} output(s), {} spent, {} kernel(s), {} rangeproof(s)",
+            "{} output(s), {} spends this block, {} kernel(s), {} rangeproof(s)",
             self.outputs.len().unwrap_or(0),
             self.deleted.deleted.cardinality(),
             self.kernels.len().unwrap_or(0),
@@ -125,6 +118,32 @@ impl Display for BlockAccumulatedData {
     }
 }
 
+/// Wrapper struct to serialize and deserialize Bitmap
+#[derive(Debug, Clone)]
+pub struct DeletedBitmap {
+    deleted: Bitmap,
+}
+
+impl DeletedBitmap {
+    pub fn into_bitmap(self) -> Bitmap {
+        self.deleted
+    }
+
+    pub fn bitmap(&self) -> &Bitmap {
+        &self.deleted
+    }
+
+    pub(super) fn bitmap_mut(&mut self) -> &mut Bitmap {
+        &mut self.deleted
+    }
+}
+
+impl From<Bitmap> for DeletedBitmap {
+    fn from(deleted: Bitmap) -> Self {
+        Self { deleted }
+    }
+}
+
 impl Serialize for DeletedBitmap {
     fn serialize<S>(&self, serializer: S) -> Result<<S as Serializer>::Ok, <S as Serializer>::Error>
     where S: Serializer {
@@ -341,32 +360,26 @@ impl ChainHeader {
         })
     }
 
-    #[inline]
     pub fn height(&self) -> u64 {
         self.header.height
     }
 
-    #[inline]
     pub fn hash(&self) -> &HashOutput {
         &self.accumulated_data.hash
     }
 
-    #[inline]
     pub fn header(&self) -> &BlockHeader {
         &self.header
     }
 
-    #[inline]
     pub fn accumulated_data(&self) -> &BlockHeaderAccumulatedData {
         &self.accumulated_data
     }
 
-    #[inline]
     pub fn into_parts(self) -> (BlockHeader, BlockHeaderAccumulatedData) {
         (self.header, self.accumulated_data)
     }
 
-    #[inline]
     pub fn into_header(self) -> BlockHeader {
         self.header
     }
@@ -407,36 +420,30 @@ impl ChainBlock {
         })
     }
 
-    #[inline]
     pub fn height(&self) -> u64 {
         self.block.header.height
     }
 
-    #[inline]
     pub fn hash(&self) -> &HashOutput {
         &self.accumulated_data.hash
     }
 
     /// Returns a reference to the inner block
-    #[inline]
     pub fn block(&self) -> &Block {
         &self.block
     }
 
     /// Returns a reference to the inner block's header
-    #[inline]
     pub fn header(&self) -> &BlockHeader {
         &self.block.header
     }
 
     /// Returns the inner block wrapped in an atomically reference counted (ARC) pointer. This call is cheap and does
     /// not copy the block in memory.
-    #[inline]
     pub fn to_arc_block(&self) -> Arc<Block> {
         self.block.clone()
     }
 
-    #[inline]
     pub fn accumulated_data(&self) -> &BlockHeaderAccumulatedData {
         &self.accumulated_data
     }

base_layer/core/src/chain_storage/async_db.rs

@@ -305,11 +305,21 @@ impl<'a, B: BlockchainBackend + 'static> AsyncDbTransaction<'a, B> {
         self
     }
 
-    pub fn update_deleted_with_diff(&mut self, header_hash: HashOutput, deleted: Bitmap) -> &mut Self {
+    pub fn update_block_accumulated_data_with_deleted_diff(
+        &mut self,
+        header_hash: HashOutput,
+        deleted: Bitmap,
+    ) -> &mut Self {
         self.transaction.update_deleted_with_diff(header_hash, deleted);
         self
     }
 
+    /// Updates the deleted tip bitmap with the indexes of the given bitmap.
+    pub fn update_deleted_bitmap(&mut self, deleted: Bitmap) -> &mut Self {
+        self.transaction.update_deleted_bitmap(deleted);
+        self
+    }
+
     pub fn insert_chain_header(&mut self, chain_header: ChainHeader) -> &mut Self {
         self.transaction.insert_chain_header(chain_header);
         self

base_layer/core/src/chain_storage/blockchain_backend.rs

@@ -1,6 +1,7 @@
 use crate::{
     blocks::{Block, BlockHeader},
     chain_storage::{
+        accumulated_data::DeletedBitmap,
         pruned_output::PrunedOutput,
         BlockAccumulatedData,
         BlockHeaderAccumulatedData,
@@ -140,6 +141,9 @@ pub trait BlockchainBackend: Send + Sync {
 
     fn fetch_orphan_chain_block(&self, hash: HashOutput) -> Result<Option<ChainBlock>, ChainStorageError>;
 
+    /// Returns the full deleted bitmap at the current blockchain tip
+    fn fetch_deleted_bitmap(&self) -> Result<DeletedBitmap, ChainStorageError>;
+
     /// Delete orphans according to age. Used to keep the orphan pool at a certain capacity
     fn delete_oldest_orphans(
         &mut self,

base_layer/core/src/chain_storage/blockchain_database.rs

@@ -910,11 +910,25 @@ pub fn calculate_mmr_roots<T: BlockchainBackend>(db: &T, block: &Block) -> Resul
     let header = &block.header;
     let body = &block.body;
 
+    let metadata = db.fetch_chain_metadata()?;
+    if header.prev_hash != *metadata.best_block() {
+        return Err(ChainStorageError::InvalidOperation(format!(
+            "Cannot calculate MMR roots for block that does not form a chain with the current tip. Block (#{}) \
+             previous hash is {} but the current tip is #{} {}",
+            header.height,
+            header.prev_hash.to_hex(),
+            metadata.height_of_longest_chain(),
+            metadata.best_block().to_hex()
+        )));
+    }
+
+    let deleted = db.fetch_deleted_bitmap()?;
+    let deleted = deleted.into_bitmap();
+
     let BlockAccumulatedData {
         kernels,
         outputs,
         range_proofs,
-        deleted,
         ..
     } = db
         .fetch_block_accumulated_data(&header.prev_hash)?
@@ -924,7 +938,6 @@ pub fn calculate_mmr_roots<T: BlockchainBackend>(db: &T, block: &Block) -> Resul
             value: header.prev_hash.to_hex(),
         })?;
 
-    let deleted = deleted.deleted;
     let mut kernel_mmr = MerkleMountainRange::<HashDigest, _>::new(kernels);
     let mut output_mmr = MutableMmr::<HashDigest, _>::new(outputs, deleted)?;
     let mut witness_mmr = MerkleMountainRange::<HashDigest, _>::new(range_proofs);
@@ -971,8 +984,7 @@ pub fn calculate_mmr_roots<T: BlockchainBackend>(db: &T, block: &Block) -> Resul
         kernel_mmr_size: kernel_mmr.get_leaf_count()? as u64,
         input_mr: input_mmr.get_merkle_root()?,
         output_mr: output_mmr.get_merkle_root()?,
-        // witness mmr size and output mmr size should be the same size
-        output_mmr_size: witness_mmr.get_leaf_count()? as u64,
+        output_mmr_size: output_mmr.get_leaf_count() as u64,
         witness_mr: witness_mmr.get_merkle_root()?,
     };
     Ok(mmr_roots)
@@ -1871,7 +1883,7 @@ fn prune_database_if_needed<T: BlockchainBackend>(
             )?;
             // Note, this could actually be done in one step instead of each block, since deleted is
             // accumulated
-            let inputs_to_prune = curr_block.deleted.deleted.clone() - last_block.deleted.deleted;
+            let inputs_to_prune = curr_block.deleted.bitmap().clone() - last_block.deleted.bitmap();
             last_block = curr_block;
 
             txn.prune_outputs_and_update_horizon(inputs_to_prune.to_vec(), block_to_prune);

base_layer/core/src/chain_storage/db_transaction.rs

@@ -182,6 +182,12 @@ impl DbTransaction {
         self
     }
 
+    /// Updates the deleted tip bitmap with the indexes of the given bitmap.
+    pub fn update_deleted_bitmap(&mut self, deleted: Bitmap) -> &mut Self {
+        self.operations.push(WriteOperation::UpdateDeletedBitmap { deleted });
+        self
+    }
+
     /// Add the BlockHeader and contents of a `Block` (i.e. inputs, outputs and kernels) to the database.
     /// If the `BlockHeader` already exists, then just the contents are updated along with the relevant accumulated
     /// data.
@@ -315,6 +321,9 @@ pub enum WriteOperation {
         header_hash: HashOutput,
         deleted: Bitmap,
     },
+    UpdateDeletedBitmap {
+        deleted: Bitmap,
+    },
     PruneOutputsAndUpdateHorizon {
         output_positions: Vec<u32>,
         horizon: u64,
@@ -417,6 +426,9 @@ impl fmt::Display for WriteOperation {
                 header_hash: _,
                 deleted: _,
             } => write!(f, "Add deleted data for block"),
+            UpdateDeletedBitmap { deleted } => {
+                write!(f, "Merge deleted bitmap at tip ({} new indexes)", deleted.cardinality())
+            },
             PruneOutputsAndUpdateHorizon {
                 output_positions,
                 horizon,

base_layer/core/src/chain_storage/lmdb_db/lmdb.rs

@@ -114,6 +114,7 @@ where
     })
 }
 
+/// Inserts or replaces the item at the given key
 pub fn lmdb_replace<K, V>(txn: &WriteTransaction<'_>, db: &Database, key: &K, val: &V) -> Result<(), ChainStorageError>
 where
     K: AsLmdbBytes + ?Sized,