O(k log n) mempool transaction sampler + Fee estimation API

kaspad/src/daemon.rs

@@ -419,15 +419,16 @@ do you confirm? (answer y/n or pass --yes to the Kaspad command line to confirm
 
     let (address_manager, port_mapping_extender_svc) = AddressManager::new(config.clone(), meta_db, tick_service.clone());
 
-    let mining_monitor = Arc::new(MiningMonitor::new(mining_counters.clone(), tx_script_cache_counters.clone(), tick_service.clone()));
     let mining_manager = MiningManagerProxy::new(Arc::new(MiningManager::new_with_extended_config(
         config.target_time_per_block,
         false,
         config.max_block_mass,
         config.ram_scale,
         config.block_template_cache_lifetime,
-        mining_counters,
+        mining_counters.clone(),
     )));
+    let mining_monitor =
+        Arc::new(MiningMonitor::new(mining_manager.clone(), mining_counters, tx_script_cache_counters.clone(), tick_service.clone()));
 
     let flow_context = Arc::new(FlowContext::new(
         consensus_manager.clone(),

mining/src/feerate/mod.rs

@@ -2,27 +2,44 @@
 //! behind this fee estimator.
 
 use crate::block_template::selector::ALPHA;
+use itertools::Itertools;
+use std::fmt::Display;
+
+/// A type representing fee/mass of a transaction in `sompi/gram` units.
+/// Given a feerate value recommendation, calculate the required fee by
+/// taking the transaction mass and multiplying it by feerate: `fee = feerate * mass(tx)`
+pub type Feerate = f64;
 
 #[derive(Clone, Copy, Debug)]
 pub struct FeerateBucket {
     pub feerate: f64,
     pub estimated_seconds: f64,
 }
 
+impl Display for FeerateBucket {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "({:.4}, {:.4}s)", self.feerate, self.estimated_seconds)
+    }
+}
+
 #[derive(Clone, Debug)]
 pub struct FeerateEstimations {
     /// *Top-priority* feerate bucket. Provides an estimation of the feerate required for sub-second DAG inclusion.
+    ///
+    /// Note: for all buckets, feerate values represent fee/mass of a transaction in `sompi/gram` units.
+    /// Given a feerate value recommendation, calculate the required fee by
+    /// taking the transaction mass and multiplying it by feerate: `fee = feerate * mass(tx)`
     pub priority_bucket: FeerateBucket,
 
-    /// A vector of *normal* priority feerate values. The first value of this vector is guaranteed to
+    /// A vector of *normal* priority feerate values. The first value of this vector is guaranteed to exist and
     /// provide an estimation for sub-*minute* DAG inclusion. All other values will have shorter estimation
     /// times than all `low_bucket` values. Therefor by chaining `[priority] | normal | low` and interpolating
     /// between them, one can compose a complete feerate function on the client side. The API makes an effort
     /// to sample enough "interesting" points on the feerate-to-time curve, so that the interpolation is meaningful.
     pub normal_buckets: Vec<FeerateBucket>,
 
     /// A vector of *low* priority feerate values. The first value of this vector is guaranteed to
-    /// provide an estimation for sub-*hour* DAG inclusion.
+    /// exist and provide an estimation for sub-*hour* DAG inclusion.
     pub low_buckets: Vec<FeerateBucket>,
 }
 
@@ -35,6 +52,14 @@ impl FeerateEstimations {
     }
 }
 
+impl Display for FeerateEstimations {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "(fee/mass, secs) priority: {}, ", self.priority_bucket)?;
+        write!(f, "normal: {}, ", self.normal_buckets.iter().format(", "))?;
+        write!(f, "low: {}", self.low_buckets.iter().format(", "))
+    }
+}
+
 pub struct FeerateEstimatorArgs {
     pub network_blocks_per_second: u64,
     pub maximum_mass_per_block: u64,
@@ -111,16 +136,21 @@ impl FeerateEstimator {
         let high = self.time_to_feerate(1f64).max(min);
         // Choose `low` feerate such that it provides sub-hour waiting time AND it covers (at least) the 0.25 quantile
         let low = self.time_to_feerate(3600f64).max(self.quantile(min, high, 0.25));
-        // Choose `mid` feerate such that it provides sub-minute waiting time AND it covers (at least) the 0.5 quantile between low and high.
-        let mid = self.time_to_feerate(60f64).max(self.quantile(low, high, 0.5));
+        // Choose `normal` feerate such that it provides sub-minute waiting time AND it covers (at least) the 0.66 quantile between low and high.
+        let normal = self.time_to_feerate(60f64).max(self.quantile(low, high, 0.66));
+        // Choose an additional point between normal and low
+        let mid = self.time_to_feerate(1800f64).max(self.quantile(min, high, 0.5));
         /* Intuition for the above:
                1. The quantile calculations make sure that we return interesting points on the `feerate_to_time` curve.
                2. They also ensure that the times don't diminish too high if small increments to feerate would suffice
                   to cover large fractions of the integral area (reflecting the position within the waiting-time distribution)
         */
         FeerateEstimations {
             priority_bucket: FeerateBucket { feerate: high, estimated_seconds: self.feerate_to_time(high) },
-            normal_buckets: vec![FeerateBucket { feerate: mid, estimated_seconds: self.feerate_to_time(mid) }],
+            normal_buckets: vec![
+                FeerateBucket { feerate: normal, estimated_seconds: self.feerate_to_time(normal) },
+                FeerateBucket { feerate: mid, estimated_seconds: self.feerate_to_time(mid) },
+            ],
             low_buckets: vec![FeerateBucket { feerate: low, estimated_seconds: self.feerate_to_time(low) }],
         }
     }
@@ -160,6 +190,7 @@ mod tests {
         let estimator = FeerateEstimator { total_weight: 0.00659, inclusion_interval: 0.004f64 };
         let minimum_feerate = 0.755;
         let estimations = estimator.calc_estimations(minimum_feerate);
+        println!("{estimations}");
         let buckets = estimations.ordered_buckets();
         assert!(buckets.last().unwrap().feerate >= minimum_feerate);
         for (i, j) in buckets.into_iter().tuple_windows() {

mining/src/mempool/model/frontier.rs

@@ -89,10 +89,11 @@ impl Frontier {
     ///     2. Indeed, if the weight distribution is not too spread (i.e., `max(weights) = O(min(weights))`), `k << n` means
     ///        that the probability of collisions is low enough and the sampling process will converge in `O(k log(n))` w.h.p.
     ///     3. It remains to deal with the case where the weight distribution is highly biased. The process implemented below
-    ///        keeps track of the top-weight element. If the distribution is highly biased, this element will be sampled twice
-    ///        with sufficient probability (in constant time), in which case we narrow the sampling space to exclude it. We do
-    ///        this by computing the prefix weight up to this top item (exclusive) and then continue the sampling process over
-    ///        the narrowed space. This process is repeated until acquiring the desired mass.  
+    ///        keeps track of the top-weight element. If the distribution is highly biased, this element will be sampled with
+    ///        sufficient probability (in constant time). Following each sampling collision we search for a consecutive range of
+    ///        top elements which were already sampled and narrow the sampling space to exclude them all . We do this by computing
+    ///        the prefix weight up to the top most item which wasn't sampled yet (inclusive) and then continue the sampling process
+    ///        over the narrowed space. This process is repeated until acquiring the desired mass.  
     ///     4. Numerical stability. Naively, one would simply subtract `total_weight -= top.weight` in order to narrow the sampling
     ///        space. However, if `top.weight` is much larger than the remaining weight, the above f64 subtraction will yield a number
     ///        close or equal to zero. We fix this by implementing a `log(n)` prefix weight operation.

mining/src/monitor.rs

@@ -1,4 +1,5 @@
 use super::MiningCounters;
+use crate::manager::MiningManagerProxy;
 use kaspa_core::{
     debug, info,
     task::{
@@ -13,6 +14,8 @@ use std::{sync::Arc, time::Duration};
 const MONITOR: &str = "mempool-monitor";
 
 pub struct MiningMonitor {
+    mining_manager: MiningManagerProxy,
+
     // Counters
     counters: Arc<MiningCounters>,
 
@@ -24,11 +27,12 @@ pub struct MiningMonitor {
 
 impl MiningMonitor {
     pub fn new(
+        mining_manager: MiningManagerProxy,
         counters: Arc<MiningCounters>,
         tx_script_cache_counters: Arc<TxScriptCacheCounters>,
         tick_service: Arc<TickService>,
     ) -> MiningMonitor {
-        MiningMonitor { counters, tx_script_cache_counters, tick_service }
+        MiningMonitor { mining_manager, counters, tx_script_cache_counters, tick_service }
     }
 
     pub async fn worker(self: &Arc<MiningMonitor>) {
@@ -62,6 +66,8 @@ impl MiningMonitor {
                     delta.low_priority_tx_counts,
                     delta.tx_accepted_counts,
                 );
+                let feerate_estimations = self.mining_manager.clone().get_realtime_feerate_estimations().await;
+                debug!("Realtime feerate estimations: {}", feerate_estimations);
             }
             if tx_script_cache_snapshot != last_tx_script_cache_snapshot {
                 debug!(

rpc/core/src/api/rpc.rs

@@ -318,7 +318,7 @@ pub trait RpcApi: Sync + Send + AnySync {
     // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
     // Fee estimation API
 
-    async fn get_fee_estimate(&self) -> RpcResult<FeeEstimate> {
+    async fn get_fee_estimate(&self) -> RpcResult<RpcFeeEstimate> {
         Ok(self.get_fee_estimate_call(GetFeeEstimateRequest {}).await?.estimate)
     }
     async fn get_fee_estimate_call(&self, request: GetFeeEstimateRequest) -> RpcResult<GetFeeEstimateResponse>;

rpc/core/src/model/feerate_estimate.rs

@@ -3,7 +3,7 @@ use serde::{Deserialize, Serialize};
 
 #[derive(Clone, Copy, Debug, Serialize, Deserialize, BorshSerialize, BorshDeserialize)]
 #[serde(rename_all = "camelCase")]
-pub struct FeerateBucket {
+pub struct RpcFeerateBucket {
     /// The fee/mass ratio estimated to be required for inclusion time <= estimated_seconds
     pub feerate: f64,
 
@@ -13,24 +13,28 @@ pub struct FeerateBucket {
 
 #[derive(Clone, Debug, Serialize, Deserialize, BorshSerialize, BorshDeserialize)]
 #[serde(rename_all = "camelCase")]
-pub struct FeeEstimate {
+pub struct RpcFeeEstimate {
     /// *Top-priority* feerate bucket. Provides an estimation of the feerate required for sub-second DAG inclusion.
-    pub priority_bucket: FeerateBucket,
+    ///
+    /// Note: for all buckets, feerate values represent fee/mass of a transaction in `sompi/gram` units.
+    /// Given a feerate value recommendation, calculate the required fee by
+    /// taking the transaction mass and multiplying it by feerate: `fee = feerate * mass(tx)`
+    pub priority_bucket: RpcFeerateBucket,
 
     /// A vector of *normal* priority feerate values. The first value of this vector is guaranteed to exist and
     /// provide an estimation for sub-*minute* DAG inclusion. All other values will have shorter estimation
     /// times than all `low_bucket` values. Therefor by chaining `[priority] | normal | low` and interpolating
     /// between them, one can compose a complete feerate function on the client side. The API makes an effort
     /// to sample enough "interesting" points on the feerate-to-time curve, so that the interpolation is meaningful.
-    pub normal_buckets: Vec<FeerateBucket>,
+    pub normal_buckets: Vec<RpcFeerateBucket>,
 
     /// A vector of *low* priority feerate values. The first value of this vector is guaranteed to
     /// exist and provide an estimation for sub-*hour* DAG inclusion.
-    pub low_buckets: Vec<FeerateBucket>,
+    pub low_buckets: Vec<RpcFeerateBucket>,
 }
 
-impl FeeEstimate {
-    pub fn ordered_buckets(&self) -> Vec<FeerateBucket> {
+impl RpcFeeEstimate {
+    pub fn ordered_buckets(&self) -> Vec<RpcFeerateBucket> {
         std::iter::once(self.priority_bucket)
             .chain(self.normal_buckets.iter().copied())
             .chain(self.low_buckets.iter().copied())
@@ -40,7 +44,7 @@ impl FeeEstimate {
 
 #[derive(Clone, Debug, Serialize, Deserialize, BorshSerialize, BorshDeserialize)]
 #[serde(rename_all = "camelCase")]
-pub struct FeeEstimateVerboseExperimentalData {
+pub struct RpcFeeEstimateVerboseExperimentalData {
     pub mempool_ready_transactions_count: u64,
     pub mempool_ready_transactions_total_mass: u64,
     pub network_mass_per_second: u64,

rpc/core/src/model/message.rs

@@ -860,7 +860,7 @@ pub struct GetFeeEstimateRequest {}
 #[derive(Clone, Debug, Serialize, Deserialize, BorshSerialize, BorshDeserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct GetFeeEstimateResponse {
-    pub estimate: FeeEstimate,
+    pub estimate: RpcFeeEstimate,
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize, BorshSerialize, BorshDeserialize)]
@@ -873,10 +873,10 @@ pub struct GetFeeEstimateExperimentalRequest {
 #[serde(rename_all = "camelCase")]
 pub struct GetFeeEstimateExperimentalResponse {
     /// The usual feerate estimate response
-    pub estimate: FeeEstimate,
+    pub estimate: RpcFeeEstimate,
 
     /// Experimental verbose data
-    pub verbose: Option<FeeEstimateVerboseExperimentalData>,
+    pub verbose: Option<RpcFeeEstimateVerboseExperimentalData>,
 }
 
 // ----------------------------------------------------------------------------

rpc/grpc/core/proto/rpc.proto

@@ -868,13 +868,29 @@ message GetDaaScoreTimestampEstimateResponseMessage{
 }
 
 message RpcFeerateBucket {
+  // Fee/mass of a transaction in `sompi/gram` units
   double feerate = 1;
   double estimated_seconds = 2;
 }
 
+// Data required for making fee estimates.
+// 
+// Feerate values represent fee/mass of a transaction in `sompi/gram` units.
+// Given a feerate value recommendation, calculate the required fee by
+// taking the transaction mass and multiplying it by feerate: `fee = feerate * mass(tx)`
 message RpcFeeEstimate {
+  // Top-priority feerate bucket. Provides an estimation of the feerate required for sub-second DAG inclusion.
   RpcFeerateBucket priority_bucket = 1;
+
+  // A vector of *normal* priority feerate values. The first value of this vector is guaranteed to exist and
+  // provide an estimation for sub-*minute* DAG inclusion. All other values will have shorter estimation
+  // times than all `low_bucket` values. Therefor by chaining `[priority] | normal | low` and interpolating
+  // between them, one can compose a complete feerate function on the client side. The API makes an effort
+  // to sample enough "interesting" points on the feerate-to-time curve, so that the interpolation is meaningful.
   repeated RpcFeerateBucket normal_buckets = 2;
+
+  // A vector of *low* priority feerate values. The first value of this vector is guaranteed to
+  // exist and provide an estimation for sub-*hour* DAG inclusion.
   repeated RpcFeerateBucket low_buckets = 3;
 }
 
@@ -883,9 +899,9 @@ message RpcFeeEstimateVerboseExperimentalData {
   uint64 mempool_ready_transactions_total_mass = 2;
   uint64 network_mass_per_second = 3;
 
-  double next_block_template_feerate_min = 4;
-  double next_block_template_feerate_median = 5;
-  double next_block_template_feerate_max = 6;
+  double next_block_template_feerate_min = 11;
+  double next_block_template_feerate_median = 12;
+  double next_block_template_feerate_max = 13;
 }
 
 message GetFeeEstimateRequestMessage {

rpc/grpc/core/src/convert/feerate_estimate.rs

@@ -6,22 +6,22 @@ use kaspa_rpc_core::RpcError;
 // rpc_core to protowire
 // ----------------------------------------------------------------------------
 
-from!(item: &kaspa_rpc_core::FeerateBucket, protowire::RpcFeerateBucket, {
+from!(item: &kaspa_rpc_core::RpcFeerateBucket, protowire::RpcFeerateBucket, {
     Self {
         feerate: item.feerate,
         estimated_seconds: item.estimated_seconds,
     }
 });
 
-from!(item: &kaspa_rpc_core::FeeEstimate, protowire::RpcFeeEstimate, {
+from!(item: &kaspa_rpc_core::RpcFeeEstimate, protowire::RpcFeeEstimate, {
     Self {
         priority_bucket: Some((&item.priority_bucket).into()),
         normal_buckets: item.normal_buckets.iter().map(|b| b.into()).collect(),
         low_buckets: item.low_buckets.iter().map(|b| b.into()).collect(),
     }
 });
 
-from!(item: &kaspa_rpc_core::FeeEstimateVerboseExperimentalData, protowire::RpcFeeEstimateVerboseExperimentalData, {
+from!(item: &kaspa_rpc_core::RpcFeeEstimateVerboseExperimentalData, protowire::RpcFeeEstimateVerboseExperimentalData, {
     Self {
         network_mass_per_second: item.network_mass_per_second,
         mempool_ready_transactions_count: item.mempool_ready_transactions_count,
@@ -36,14 +36,14 @@ from!(item: &kaspa_rpc_core::FeeEstimateVerboseExperimentalData, protowire::RpcF
 // protowire to rpc_core
 // ----------------------------------------------------------------------------
 
-try_from!(item: &protowire::RpcFeerateBucket, kaspa_rpc_core::FeerateBucket, {
+try_from!(item: &protowire::RpcFeerateBucket, kaspa_rpc_core::RpcFeerateBucket, {
     Self {
         feerate: item.feerate,
         estimated_seconds: item.estimated_seconds,
     }
 });
 
-try_from!(item: &protowire::RpcFeeEstimate, kaspa_rpc_core::FeeEstimate, {
+try_from!(item: &protowire::RpcFeeEstimate, kaspa_rpc_core::RpcFeeEstimate, {
     Self {
         priority_bucket: item.priority_bucket
             .as_ref()
@@ -54,7 +54,7 @@ try_from!(item: &protowire::RpcFeeEstimate, kaspa_rpc_core::FeeEstimate, {
     }
 });
 
-try_from!(item: &protowire::RpcFeeEstimateVerboseExperimentalData, kaspa_rpc_core::FeeEstimateVerboseExperimentalData, {
+try_from!(item: &protowire::RpcFeeEstimateVerboseExperimentalData, kaspa_rpc_core::RpcFeeEstimateVerboseExperimentalData, {
     Self {
         network_mass_per_second: item.network_mass_per_second,
         mempool_ready_transactions_count: item.mempool_ready_transactions_count,

rpc/service/src/converter/feerate_estimate.rs

@@ -1,7 +1,7 @@
 use kaspa_mining::feerate::{FeeEstimateVerbose, FeerateBucket, FeerateEstimations};
 use kaspa_rpc_core::{
-    message::GetFeeEstimateExperimentalResponse as RpcFeeEstimateVerboseResponse, FeeEstimate as RpcFeeEstimate,
-    FeeEstimateVerboseExperimentalData as RpcFeeEstimateVerbose, FeerateBucket as RpcFeerateBucket,
+    message::GetFeeEstimateExperimentalResponse as RpcFeeEstimateVerboseResponse, RpcFeeEstimate,
+    RpcFeeEstimateVerboseExperimentalData as RpcFeeEstimateVerbose, RpcFeerateBucket,
 };
 
 pub trait FeerateBucketConverter {

rpc/service/src/service.rs

@@ -112,7 +112,7 @@ pub struct RpcCoreService {
     perf_monitor: Arc<PerfMonitor<Arc<TickService>>>,
     p2p_tower_counters: Arc<TowerConnectionCounters>,
     grpc_tower_counters: Arc<TowerConnectionCounters>,
-    fee_estimate_cache: ExpiringCache<FeeEstimate>,
+    fee_estimate_cache: ExpiringCache<RpcFeeEstimate>,
     fee_estimate_verbose_cache: ExpiringCache<GetFeeEstimateExperimentalResponse>,
 }
 

rpc/wrpc/server/src/router.rs

@@ -44,9 +44,11 @@ impl Router {
                 GetBlockTemplate,
                 GetCoinSupply,
                 GetConnectedPeerInfo,
+                GetCurrentNetwork,
                 GetDaaScoreTimestampEstimate,
                 GetServerInfo,
-                GetCurrentNetwork,
+                GetFeeEstimate,
+                GetFeeEstimateExperimental,
                 GetHeaders,
                 GetInfo,
                 GetInfo,
@@ -68,8 +70,6 @@ impl Router {
                 SubmitTransaction,
                 SubmitTransactionReplacement,
                 Unban,
-                GetFeeEstimate,
-                GetFeeEstimateExperimental,
             ]
         );