feat: add check to verify mempool state (#6316)

Description
---
Added a check to verify that the mempool last processed block is in sync
with the blockchain state whenever a new block template is requested, as
adding a block to the db and processing the same block in the mempool
happens asynchronously. Miners can get block templates with double
spending if this is not done.

Motivation and Context
---
In a recent stress test the following behaviour was observed multiple
times:
1. [mempool] New transaction `4776...2001` received with output
`000e9...6b57c`
1. [mempool] Transaction `4776d...82001` inserted into the mempool
1. [grpc] Miner mined block `af0c8...c76c0` containing output
`000e9...6b57c`
1. [lmdb_db] Output `000e9...6b57c` inserted into the db
1. [grpc] New block template requested (`GetNewBlockTemplate)
1. [mempool] New block template prepared containing output
`000e9...6b57c`
1. [mempool] Transaction `4776d...82001` with output `000e9...6b57c`
removed from the mempool
1. [mempool] Processed new block `af0c8...c76c0` containing output
`000e9...6b57c`
1. [grpc] New block requested (`GetNewBlock`)
1. [grpc] Using the latest block template, `pub fn
prepare_new_block(&self, template: NewBlockTemplate)` fails at `let
roots = match calculate_mmr_roots(&*db, self.rules(), &block, &mut smt)`
with `ERROR Output
commitment(000e9e1fd41f672e0b88f89b3d24390beb37d50e8be2967980e24a13d796b57c)
already in SMT`


How Has This Been Tested?
---
System-level stress test to be done

What process can a PR reviewer use to test or verify this change?
---
Review code change

<!-- Checklist -->
<!-- 1. Is the title of your PR in the form that would make nice release
notes? The title, excluding the conventional commit
tag, will be included exactly as is in the CHANGELOG, so please think
about it carefully. -->


Breaking Changes
---

- [x] None
- [ ] Requires data directory on base node to be deleted
- [ ] Requires hard fork
- [ ] Other - Please specify

<!-- Does this include a breaking change? If so, include this line as a
footer -->
<!-- BREAKING CHANGE: Description what the user should do, e.g. delete a
database, resync the chain -->

base_layer/core/src/base_node/comms_interface/inbound_handlers.rs

@@ -257,6 +257,15 @@ where B: BlockchainBackend + 'static
             },
             NodeCommsRequest::GetNewBlockTemplate(request) => {
                 let best_block_header = self.blockchain_db.fetch_tip_header().await?;
+                let last_seen_hash = self.mempool.get_last_seen_hash().await?;
+                if last_seen_hash != FixedHash::default() && best_block_header.hash() != &last_seen_hash {
+                    warn!(
+                        target: LOG_TARGET,
+                        "Mempool out of sync - last seen hash '{}' does not match the tip hash '{}'.",
+                        last_seen_hash, best_block_header.hash()
+                    );
+                    return Err(CommsInterfaceError::InternalError("Mempool out of sync".to_string()));
+                }
                 let mut header = BlockHeader::from_previous(best_block_header.header());
                 let constants = self.consensus_manager.consensus_constants(header.height);
                 header.version = constants.blockchain_version();
@@ -989,6 +998,10 @@ where B: BlockchainBackend + 'static
         debug!(target: LOG_TARGET, "Target difficulty {} for PoW {}", target, pow_algo);
         Ok(target)
     }
+
+    pub async fn get_last_seen_hash(&self) -> Result<FixedHash, CommsInterfaceError> {
+        self.mempool.get_last_seen_hash().await.map_err(|e| e.into())
+    }
 }
 
 impl<B> Clone for InboundNodeCommsHandlers<B> {

base_layer/core/src/mempool/mempool.rs

@@ -23,7 +23,7 @@
 use std::sync::{Arc, RwLock};
 
 use log::debug;
-use tari_common_types::types::{PrivateKey, Signature};
+use tari_common_types::types::{FixedHash, PrivateKey, Signature};
 use tokio::task;
 
 use crate::{
@@ -213,4 +213,8 @@ impl Mempool {
         })
         .await?
     }
+
+    pub async fn get_last_seen_hash(&self) -> Result<FixedHash, MempoolError> {
+        self.with_read_access(|storage| Ok(storage.last_seen_hash)).await
+    }
 }

base_layer/core/src/mempool/mempool_storage.rs

@@ -23,7 +23,7 @@
 use std::{sync::Arc, time::Instant};
 
 use log::*;
-use tari_common_types::types::{PrivateKey, Signature};
+use tari_common_types::types::{FixedHash, PrivateKey, Signature};
 use tari_utilities::hex::Hex;
 
 use crate::{
@@ -57,6 +57,7 @@ pub struct MempoolStorage {
     validator: Box<dyn TransactionValidator>,
     rules: ConsensusManager,
     last_seen_height: u64,
+    pub(crate) last_seen_hash: FixedHash,
 }
 
 impl MempoolStorage {
@@ -68,6 +69,7 @@ impl MempoolStorage {
             validator,
             rules,
             last_seen_height: 0,
+            last_seen_hash: Default::default(),
         }
     }
 
@@ -220,6 +222,7 @@ impl MempoolStorage {
         self.reorg_pool.compact();
 
         self.last_seen_height = published_block.header.height;
+        self.last_seen_hash = published_block.header.hash();
         debug!(target: LOG_TARGET, "Compaction took {:.2?}", timer.elapsed());
         match self.stats() {
             Ok(stats) => debug!(target: LOG_TARGET, "{}", stats),
@@ -269,12 +272,13 @@ impl MempoolStorage {
             .remove_reorged_txs_and_discard_double_spends(removed_blocks, new_blocks);
         self.insert_txs(removed_txs)
             .map_err(|e| MempoolError::InternalError(e.to_string()))?;
-        if let Some(height) = new_blocks
+        if let Some((height, hash)) = new_blocks
             .last()
             .or_else(|| removed_blocks.first())
-            .map(|block| block.header.height)
+            .map(|block| (block.header.height, block.header.hash()))
         {
             self.last_seen_height = height;
+            self.last_seen_hash = hash;
         }
         Ok(())
     }