feat: add mempool fee per gram stats to transaction service (#4105)

Description
---
- adds fee per gram stats call to transaction service.
- adds `get_mempool_fee_per_gram_stats` to wallet RPC service

Motivation and Context
---
Add call for the wallet to query the base node for min/max and average mempool fee per gram stats to allow it to determine the best fee given current network conditions.

How Has This Been Tested?
---
Added new unit tests

base_layer/core/src/base_node/proto/rpc.proto

@@ -89,3 +89,18 @@ message SyncUtxosByBlockResponse  {
   uint64 height = 2;
   bytes header_hash = 3;
 }
+
+message GetMempoolFeePerGramStatsRequest {
+  uint64 count = 1;
+}
+
+message GetMempoolFeePerGramStatsResponse {
+  repeated MempoolFeePerGramStat stats = 1;
+}
+
+message MempoolFeePerGramStat {
+  uint64 order = 1;
+  uint64 max_fee_per_gram = 2;
+  uint64 avg_fee_per_gram = 4;
+  uint64 min_fee_per_gram = 5;
+}

base_layer/core/src/base_node/proto/rpc.rs

@@ -24,7 +24,7 @@ use std::convert::{TryFrom, TryInto};
 
 use tari_utilities::Hashable;
 
-use crate::{blocks::Block, chain_storage::PrunedOutput, proto::base_node as proto};
+use crate::{blocks::Block, chain_storage::PrunedOutput, mempool::FeePerGramStat, proto::base_node as proto};
 
 impl TryFrom<Block> for proto::BlockBodyResponse {
     type Error = String;
@@ -55,3 +55,22 @@ impl From<PrunedOutput> for proto::SyncUtxo {
         }
     }
 }
+
+impl From<Vec<FeePerGramStat>> for proto::GetMempoolFeePerGramStatsResponse {
+    fn from(stats: Vec<FeePerGramStat>) -> Self {
+        Self {
+            stats: stats.into_iter().map(Into::into).collect(),
+        }
+    }
+}
+
+impl From<FeePerGramStat> for proto::MempoolFeePerGramStat {
+    fn from(stat: FeePerGramStat) -> Self {
+        Self {
+            order: stat.order,
+            min_fee_per_gram: stat.min_fee_per_gram.as_u64(),
+            avg_fee_per_gram: stat.avg_fee_per_gram.as_u64(),
+            max_fee_per_gram: stat.max_fee_per_gram.as_u64(),
+        }
+    }
+}

base_layer/core/src/base_node/rpc/mod.rs

@@ -43,6 +43,8 @@ use crate::{
         base_node::{
             FetchMatchingUtxos,
             FetchUtxosResponse,
+            GetMempoolFeePerGramStatsRequest,
+            GetMempoolFeePerGramStatsResponse,
             QueryDeletedRequest,
             QueryDeletedResponse,
             Signatures,
@@ -111,6 +113,12 @@ pub trait BaseNodeWalletService: Send + Sync + 'static {
         &self,
         request: Request<SyncUtxosByBlockRequest>,
     ) -> Result<Streaming<SyncUtxosByBlockResponse>, RpcStatus>;
+
+    #[rpc(method = 12)]
+    async fn get_mempool_fee_per_gram_stats(
+        &self,
+        request: Request<GetMempoolFeePerGramStatsRequest>,
+    ) -> Result<Response<GetMempoolFeePerGramStatsResponse>, RpcStatus>;
 }
 
 #[cfg(feature = "base_node")]

base_layer/core/src/base_node/rpc/service.rs

@@ -42,6 +42,8 @@ use crate::{
         base_node::{
             FetchMatchingUtxos,
             FetchUtxosResponse,
+            GetMempoolFeePerGramStatsRequest,
+            GetMempoolFeePerGramStatsResponse,
             QueryDeletedRequest,
             QueryDeletedResponse,
             Signatures as SignaturesProto,
@@ -597,4 +599,30 @@ impl<B: BlockchainBackend + 'static> BaseNodeWalletService for BaseNodeWalletRpc
 
         Ok(Streaming::new(rx))
     }
+
+    async fn get_mempool_fee_per_gram_stats(
+        &self,
+        request: Request<GetMempoolFeePerGramStatsRequest>,
+    ) -> Result<Response<GetMempoolFeePerGramStatsResponse>, RpcStatus> {
+        let req = request.into_message();
+        let count =
+            usize::try_from(req.count).map_err(|_| RpcStatus::bad_request("count must be less than or equal to 20"))?;
+
+        if count > 20 {
+            return Err(RpcStatus::bad_request("count must be less than or equal to 20"));
+        }
+
+        let metadata = self
+            .db
+            .get_chain_metadata()
+            .await
+            .rpc_status_internal_error(LOG_TARGET)?;
+        let stats = self
+            .mempool()
+            .get_fee_per_gram_stats(count, metadata.height_of_longest_chain())
+            .await
+            .rpc_status_internal_error(LOG_TARGET)?;
+
+        Ok(Response::new(stats.into()))
+    }
 }

base_layer/core/src/mempool/mempool.rs

@@ -31,6 +31,7 @@ use crate::{
     mempool::{
         error::MempoolError,
         mempool_storage::MempoolStorage,
+        FeePerGramStat,
         MempoolConfig,
         StateResponse,
         StatsResponse,
@@ -135,6 +136,15 @@ impl Mempool {
         self.with_read_access(|storage| Ok(storage.state())).await
     }
 
+    pub async fn get_fee_per_gram_stats(
+        &self,
+        count: usize,
+        tip_height: u64,
+    ) -> Result<Vec<FeePerGramStat>, MempoolError> {
+        self.with_read_access(move |storage| storage.get_fee_per_gram_stats(count, tip_height))
+            .await
+    }
+
     async fn with_read_access<F, T>(&self, callback: F) -> Result<T, MempoolError>
     where
         F: FnOnce(&MempoolStorage) -> Result<T, MempoolError> + Send + 'static,

base_layer/core/src/mempool/mempool_storage.rs

@@ -33,6 +33,7 @@ use crate::{
         error::MempoolError,
         reorg_pool::ReorgPool,
         unconfirmed_pool::UnconfirmedPool,
+        FeePerGramStat,
         MempoolConfig,
         StateResponse,
         StatsResponse,
@@ -300,4 +301,13 @@ impl MempoolStorage {
             reorg_pool,
         }
     }
+
+    pub fn get_fee_per_gram_stats(&self, count: usize, tip_height: u64) -> Result<Vec<FeePerGramStat>, MempoolError> {
+        let target_weight = self
+            .rules
+            .consensus_constants(tip_height)
+            .get_max_block_weight_excluding_coinbase();
+        let stats = self.unconfirmed_pool.get_fee_per_gram_stats(count, target_weight)?;
+        Ok(stats)
+    }
 }

base_layer/core/src/mempool/mod.rs

@@ -74,7 +74,10 @@ use serde::{Deserialize, Serialize};
 pub use sync_protocol::MempoolSyncInitializer;
 use tari_common_types::types::Signature;
 
-use crate::transactions::transaction_components::Transaction;
+use crate::{
+    proto::base_node as base_node_proto,
+    transactions::{tari_amount::MicroTari, transaction_components::Transaction},
+};
 
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub struct StatsResponse {
@@ -131,3 +134,22 @@ impl Display for TxStorageResponse {
         fmt.write_str(storage)
     }
 }
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct FeePerGramStat {
+    pub order: u64,
+    pub min_fee_per_gram: MicroTari,
+    pub avg_fee_per_gram: MicroTari,
+    pub max_fee_per_gram: MicroTari,
+}
+
+impl From<base_node_proto::MempoolFeePerGramStat> for FeePerGramStat {
+    fn from(value: base_node_proto::MempoolFeePerGramStat) -> Self {
+        Self {
+            order: value.order,
+            min_fee_per_gram: value.min_fee_per_gram.into(),
+            avg_fee_per_gram: value.avg_fee_per_gram.into(),
+            max_fee_per_gram: value.max_fee_per_gram.into(),
+        }
+    }
+}

base_layer/core/src/mempool/service/handle.rs

@@ -26,6 +26,7 @@ use tari_service_framework::{reply_channel::TrySenderService, Service};
 use crate::{
     mempool::{
         service::{MempoolRequest, MempoolResponse},
+        FeePerGramStat,
         MempoolServiceError,
         StateResponse,
         StatsResponse,
@@ -81,4 +82,19 @@ impl MempoolHandle {
             _ => panic!("Incorrect response"),
         }
     }
+
+    pub async fn get_fee_per_gram_stats(
+        &mut self,
+        count: usize,
+        tip_height: u64,
+    ) -> Result<Vec<FeePerGramStat>, MempoolServiceError> {
+        match self
+            .inner
+            .call(MempoolRequest::GetFeePerGramStats { count, tip_height })
+            .await??
+        {
+            MempoolResponse::FeePerGramStats { response } => Ok(response),
+            _ => panic!("Incorrect response"),
+        }
+    }
 }

base_layer/core/src/mempool/service/inbound_handlers.rs

@@ -57,7 +57,7 @@ impl MempoolInboundHandlers {
     /// Handle inbound Mempool service requests from remote nodes and local services.
     pub async fn handle_request(&mut self, request: MempoolRequest) -> Result<MempoolResponse, MempoolServiceError> {
         debug!(target: LOG_TARGET, "Handling remote request: {}", request);
-        use MempoolRequest::{GetState, GetStats, GetTxStateByExcessSig, SubmitTransaction};
+        use MempoolRequest::{GetFeePerGramStats, GetState, GetStats, GetTxStateByExcessSig, SubmitTransaction};
         match request {
             GetStats => Ok(MempoolResponse::Stats(self.mempool.stats().await?)),
             GetState => Ok(MempoolResponse::State(self.mempool.state().await?)),
@@ -72,6 +72,10 @@ impl MempoolInboundHandlers {
                 );
                 Ok(MempoolResponse::TxStorage(self.submit_transaction(tx, None).await?))
             },
+            GetFeePerGramStats { count, tip_height } => {
+                let stats = self.mempool.get_fee_per_gram_stats(count, tip_height).await?;
+                Ok(MempoolResponse::FeePerGramStats { response: stats })
+            },
         }
     }
 

base_layer/core/src/mempool/service/request.rs

@@ -36,20 +36,25 @@ pub enum MempoolRequest {
     GetState,
     GetTxStateByExcessSig(Signature),
     SubmitTransaction(Transaction),
+    GetFeePerGramStats { count: usize, tip_height: u64 },
 }
 
 impl Display for MempoolRequest {
     fn fmt(&self, f: &mut Formatter) -> Result<(), Error> {
         match self {
-            MempoolRequest::GetStats => f.write_str("GetStats"),
-            MempoolRequest::GetState => f.write_str("GetState"),
+            MempoolRequest::GetStats => write!(f, "GetStats"),
+            MempoolRequest::GetState => write!(f, "GetState"),
             MempoolRequest::GetTxStateByExcessSig(sig) => {
-                f.write_str(&format!("GetTxStateByExcessSig ({})", sig.get_signature().to_hex()))
+                write!(f, "GetTxStateByExcessSig ({})", sig.get_signature().to_hex())
             },
-            MempoolRequest::SubmitTransaction(tx) => f.write_str(&format!(
+            MempoolRequest::SubmitTransaction(tx) => write!(
+                f,
                 "SubmitTransaction ({})",
                 tx.body.kernels()[0].excess_sig.get_signature().to_hex()
-            )),
+            ),
+            MempoolRequest::GetFeePerGramStats { count, tip_height } => {
+                write!(f, "GetFeePerGramStats(count: {}, tip_height: {})", *count, *tip_height)
+            },
         }
     }
 }

base_layer/core/src/mempool/service/response.rs

@@ -24,23 +24,25 @@ use std::{fmt, fmt::Formatter};
 
 use tari_common_types::waiting_requests::RequestKey;
 
-use crate::mempool::{StateResponse, StatsResponse, TxStorageResponse};
+use crate::mempool::{FeePerGramStat, StateResponse, StatsResponse, TxStorageResponse};
 
 /// API Response enum for Mempool responses.
 #[derive(Clone, Debug)]
 pub enum MempoolResponse {
     Stats(StatsResponse),
     State(StateResponse),
     TxStorage(TxStorageResponse),
+    FeePerGramStats { response: Vec<FeePerGramStat> },
 }
 
 impl fmt::Display for MempoolResponse {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
-        use MempoolResponse::{State, Stats, TxStorage};
+        use MempoolResponse::{FeePerGramStats, State, Stats, TxStorage};
         match &self {
             Stats(_) => write!(f, "Stats"),
             State(_) => write!(f, "State"),
             TxStorage(_) => write!(f, "TxStorage"),
+            FeePerGramStats { response } => write!(f, "FeePerGramStats({} item(s))", response.len()),
         }
     }
 }

base_layer/core/src/mempool/test_utils/mock.rs

@@ -125,7 +125,7 @@ impl MempoolServiceMock {
     }
 
     async fn handle_request(&self, req: MempoolRequest) -> Result<MempoolResponse, MempoolServiceError> {
-        use MempoolRequest::{GetState, GetStats, GetTxStateByExcessSig, SubmitTransaction};
+        use MempoolRequest::{GetFeePerGramStats, GetState, GetStats, GetTxStateByExcessSig, SubmitTransaction};
 
         self.state.inc_call_count();
         match req {
@@ -137,6 +137,9 @@ impl MempoolServiceMock {
             SubmitTransaction(_) => Ok(MempoolResponse::TxStorage(
                 self.state.submit_transaction.lock().await.clone(),
             )),
+            GetFeePerGramStats { .. } => {
+                unimplemented!()
+            },
         }
     }
 }