feat!: compact block propagation (#3704)

Description
---

- Changes the propagated new block message to contain `<block_header, coinbase, excess_sigs>`
- Reconciles incoming new block with known mempool transactions by excess signature
- Requests, if required, missing transactions from peer
- Failing that, request the full block from the peer
- Reduce (deallocate) mempool memory when transactions are removed 
- Replace real time-based reorg pool expiry with block height-based expiry
- Clean up deprecated mempool messaging handlers etc
- Ensure DHT connectivity keeps hold of connection handles 

Motivation and Context
---
In the majority of cases, the inputs, outputs and kernels to be inserted in a block are already known and exist in the mempool. Previously, a propagated new block message would just contain the block hash, requiring the node request the full block from the peer. This implementation provides the minimum block data required to assemble a valid block and propagates that message (typically less than 100kb).

How Has This Been Tested?
---
Updated tests, existing block propagation tests
Manually: 3 base nodes, 1 wallet and 4 blocks worth of transactions in the mempool on dibbler

applications/tari_base_node/src/bootstrap.rs

@@ -140,7 +140,6 @@ where B: BlockchainBackend + 'static
                 node_config,
             ))
             .add_initializer(MempoolServiceInitializer::new(
-                mempool_config,
                 self.mempool.clone(),
                 peer_message_subscriptions.clone(),
             ))

applications/tari_base_node/src/command_handler.rs

@@ -158,7 +158,7 @@ impl CommandHandler {
                 "Mempool",
                 format!(
                     "{}tx ({}g, +/- {}blks)",
-                    mempool_stats.total_txs,
+                    mempool_stats.unconfirmed_txs,
                     mempool_stats.total_weight,
                     if mempool_stats.total_weight == 0 {
                         0
@@ -1118,7 +1118,9 @@ impl CommandHandler {
         let local_node_comms_interface = self.node_service.clone();
         self.executor.spawn(async move {
             let blocks = try_or_print!(db.rewind_to_height(new_height).await);
-            local_node_comms_interface.publish_block_event(BlockEvent::BlockSyncRewind(blocks));
+            if !blocks.is_empty() {
+                local_node_comms_interface.publish_block_event(BlockEvent::BlockSyncRewind(blocks));
+            }
         });
     }
 

base_layer/core/Cargo.toml

@@ -65,7 +65,6 @@ thiserror = "1.0.26"
 tokio = { version = "1.11", features = ["time", "sync", "macros"] }
 tracing = "0.1.26"
 tracing-attributes = "*"
-ttl_cache = "0.5.1"
 uint = { version = "0.9", default-features = false }
 
 [dev-dependencies]

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

@@ -26,7 +26,7 @@ use std::{
 };
 
 use serde::{Deserialize, Serialize};
-use tari_common_types::types::{Commitment, HashOutput, PublicKey, Signature};
+use tari_common_types::types::{Commitment, HashOutput, PrivateKey, PublicKey, Signature};
 use tari_crypto::tari_utilities::hex::Hex;
 
 use crate::{blocks::NewBlockTemplate, chain_storage::MmrTree, proof_of_work::PowAlgorithm};
@@ -44,14 +44,14 @@ pub struct MmrStateRequest {
 pub enum NodeCommsRequest {
     GetChainMetadata,
     FetchHeaders(RangeInclusive<u64>),
-    FetchHeadersWithHashes(Vec<HashOutput>),
+    FetchHeadersByHashes(Vec<HashOutput>),
     FetchHeadersAfter(Vec<HashOutput>, HashOutput),
     FetchMatchingUtxos(Vec<HashOutput>),
     FetchMatchingTxos(Vec<HashOutput>),
     FetchMatchingBlocks(RangeInclusive<u64>),
     FetchBlocksByHash(Vec<HashOutput>),
-    FetchBlocksWithKernels(Vec<Signature>),
-    FetchBlocksWithUtxos(Vec<Commitment>),
+    FetchBlocksByKernelExcessSigs(Vec<Signature>),
+    FetchBlocksByUtxos(Vec<Commitment>),
     GetHeaderByHash(HashOutput),
     GetBlockByHash(HashOutput),
     GetNewBlockTemplate(GetNewBlockTemplateRequest),
@@ -67,6 +67,9 @@ pub enum NodeCommsRequest {
     FetchAssetMetadata {
         asset_public_key: PublicKey,
     },
+    FetchMempoolTransactionsByExcessSigs {
+        excess_sigs: Vec<PrivateKey>,
+    },
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -83,16 +86,16 @@ impl Display for NodeCommsRequest {
             FetchHeaders(range) => {
                 write!(f, "FetchHeaders ({:?})", range)
             },
-            FetchHeadersWithHashes(v) => write!(f, "FetchHeadersWithHashes (n={})", v.len()),
+            FetchHeadersByHashes(v) => write!(f, "FetchHeadersByHashes (n={})", v.len()),
             FetchHeadersAfter(v, _hash) => write!(f, "FetchHeadersAfter (n={})", v.len()),
             FetchMatchingUtxos(v) => write!(f, "FetchMatchingUtxos (n={})", v.len()),
             FetchMatchingTxos(v) => write!(f, "FetchMatchingTxos (n={})", v.len()),
             FetchMatchingBlocks(range) => {
                 write!(f, "FetchMatchingBlocks ({:?})", range)
             },
             FetchBlocksByHash(v) => write!(f, "FetchBlocksByHash (n={})", v.len()),
-            FetchBlocksWithKernels(v) => write!(f, "FetchBlocksWithKernels (n={})", v.len()),
-            FetchBlocksWithUtxos(v) => write!(f, "FetchBlocksWithUtxos (n={})", v.len()),
+            FetchBlocksByKernelExcessSigs(v) => write!(f, "FetchBlocksByKernelExcessSigs (n={})", v.len()),
+            FetchBlocksByUtxos(v) => write!(f, "FetchBlocksByUtxos (n={})", v.len()),
             GetHeaderByHash(v) => write!(f, "GetHeaderByHash({})", v.to_hex()),
             GetBlockByHash(v) => write!(f, "GetBlockByHash({})", v.to_hex()),
             GetNewBlockTemplate(v) => write!(f, "GetNewBlockTemplate ({}) with weight {}", v.algo, v.max_weight),
@@ -112,6 +115,9 @@ impl Display for NodeCommsRequest {
             FetchAssetMetadata { .. } => {
                 write!(f, "FetchAssetMetadata")
             },
+            FetchMempoolTransactionsByExcessSigs { .. } => {
+                write!(f, "FetchMempoolTransactionsByExcessSigs")
+            },
         }
     }
 }

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

@@ -20,20 +20,25 @@
 // WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-use std::fmt::{self, Display, Formatter};
+use std::{
+    fmt::{self, Display, Formatter},
+    sync::Arc,
+};
 
-use serde::{Deserialize, Serialize};
-use tari_common_types::{chain_metadata::ChainMetadata, types::HashOutput};
+use tari_common_types::{
+    chain_metadata::ChainMetadata,
+    types::{HashOutput, PrivateKey},
+};
 
 use crate::{
     blocks::{Block, BlockHeader, ChainHeader, HistoricalBlock, NewBlockTemplate},
     chain_storage::UtxoMinedInfo,
     proof_of_work::Difficulty,
-    transactions::transaction::{TransactionKernel, TransactionOutput},
+    transactions::transaction::{Transaction, TransactionKernel, TransactionOutput},
 };
 
 /// API Response enum
-#[derive(Debug, Serialize, Deserialize, Clone)]
+#[derive(Debug, Clone)]
 pub enum NodeCommsResponse {
     ChainMetadata(ChainMetadata),
     TransactionKernels(Vec<TransactionKernel>),
@@ -60,6 +65,7 @@ pub enum NodeCommsResponse {
     FetchAssetMetadataResponse {
         output: Box<Option<UtxoMinedInfo>>,
     },
+    FetchMempoolTransactionsByExcessSigsResponse(FetchMempoolTransactionsResponse),
 }
 
 impl Display for NodeCommsResponse {
@@ -90,6 +96,19 @@ impl Display for NodeCommsResponse {
             FetchTokensResponse { .. } => write!(f, "FetchTokensResponse"),
             FetchAssetRegistrationsResponse { .. } => write!(f, "FetchAssetRegistrationsResponse"),
             FetchAssetMetadataResponse { .. } => write!(f, "FetchAssetMetadataResponse"),
+            FetchMempoolTransactionsByExcessSigsResponse(resp) => write!(
+                f,
+                "FetchMempoolTransactionsByExcessSigsResponse({} transaction(s), {} not found)",
+                resp.transactions.len(),
+                resp.not_found.len()
+            ),
         }
     }
 }
+
+/// Container struct for mempool transaction responses
+#[derive(Debug, Clone)]
+pub struct FetchMempoolTransactionsResponse {
+    pub transactions: Vec<Arc<Transaction>>,
+    pub not_found: Vec<PrivateKey>,
+}

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

@@ -65,4 +65,6 @@ pub enum CommsInterfaceError {
     BlockHeaderNotFound(u64),
     #[error("Block error: {0}")]
     BlockError(#[from] BlockError),
+    #[error("Invalid request for {request}: {details}")]
+    InvalidRequest { request: &'static str, details: String },
 }