Mining Rule Engine Scaffolding and Sync Rate Rule implementation (#654)

* Initial implementation of Mining Rule Engine

- monitor sync state and update anything that cares about is_sycned to
  use the updated rules
- sync rate rule: allow mining if sync rate is below threshold and
  finality point is recent

Move most is_nearly_synced logic to mining_rule_engine

- instantiate Hub in daemon
- Reverse the dependency of flow context and MRE
- remove the async_is_nearly_synced consensus api
- add get_sink_daa_score_timestamp consensus API

Move is_nearly_synced logic from params to rule engine

* Add MiningRules and wire it up

* Implement MiningRule trait in a submodule

* Implement Sync Rate Rule

* Update is_synced for non-mining context

- Make it not depend on sync rate rule

* comment fixes

* Process relay blocks if sync rate rule is triggered

Author: coderofstuff

Cargo.lock

@@ -110,6 +110,7 @@ kaspa-muhash = { version = "0.17.1", path = "crypto/muhash" }
 kaspa-notify = { version = "0.17.1", path = "notify" }
 kaspa-p2p-flows = { version = "0.17.1", path = "protocol/flows" }
 kaspa-p2p-lib = { version = "0.17.1", path = "protocol/p2p" }
+kaspa-p2p-mining = { version = "0.17.1", path = "protocol/mining" }
 kaspa-perf-monitor = { version = "0.17.1", path = "metrics/perf_monitor" }
 kaspa-pow = { version = "0.17.1", path = "consensus/pow" }
 kaspa-rpc-core = { version = "0.17.1", path = "rpc/core" }

Cargo.toml

@@ -250,6 +250,10 @@ impl ConsensusSessionOwned {
         self.clone().spawn_blocking(|c| c.get_sink_timestamp()).await
     }
 
+    pub async fn async_get_sink_daa_score_timestamp(&self) -> DaaScoreTimestamp {
+        self.clone().spawn_blocking(|c| c.get_sink_daa_score_timestamp()).await
+    }
+
     pub async fn async_get_current_block_color(&self, hash: Hash) -> Option<bool> {
         self.clone().spawn_blocking(move |c| c.get_current_block_color(hash)).await
     }
@@ -263,13 +267,6 @@ impl ConsensusSessionOwned {
         self.clone().spawn_blocking(|c| c.estimate_block_count()).await
     }
 
-    /// Returns whether this consensus is considered synced or close to being synced.
-    ///
-    /// This info is used to determine if it's ok to use a block template from this node for mining purposes.
-    pub async fn async_is_nearly_synced(&self) -> bool {
-        self.clone().spawn_blocking(|c| c.is_nearly_synced()).await
-    }
-
     pub async fn async_get_virtual_chain_from_block(
         &self,
         low: Hash,

components/consensusmanager/src/session.rs

@@ -135,6 +135,10 @@ pub trait ConsensusApi: Send + Sync {
         unimplemented!()
     }
 
+    fn get_sink_daa_score_timestamp(&self) -> DaaScoreTimestamp {
+        unimplemented!()
+    }
+
     fn get_current_block_color(&self, hash: Hash) -> Option<bool> {
         unimplemented!()
     }
@@ -152,13 +156,6 @@ pub trait ConsensusApi: Send + Sync {
         unimplemented!()
     }
 
-    /// Returns whether this consensus is considered synced or close to being synced.
-    ///
-    /// This info is used to determine if it's ok to use a block template from this node for mining purposes.
-    fn is_nearly_synced(&self) -> bool {
-        unimplemented!()
-    }
-
     /// Gets the virtual chain paths from `low` to the `sink` hash, or until `chain_path_added_limit` is reached
     ///
     /// Note:   

consensus/core/src/api/mod.rs

@@ -10,10 +10,7 @@ use crate::{
 };
 use kaspa_addresses::Prefix;
 use kaspa_math::Uint256;
-use std::{
-    cmp::min,
-    time::{SystemTime, UNIX_EPOCH},
-};
+use std::cmp::min;
 
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub struct ForkActivation(u64);
@@ -260,10 +257,6 @@ pub struct Params {
     pub crescendo_activation: ForkActivation,
 }
 
-fn unix_now() -> u64 {
-    SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis() as u64
-}
-
 impl Params {
     /// Returns the size of the full blocks window that is inspected to calculate the past median time (legacy)
     #[inline]
@@ -377,7 +370,7 @@ impl Params {
         )
     }
 
-    fn expected_difficulty_window_duration_in_milliseconds(&self) -> ForkedParam<u64> {
+    pub fn expected_difficulty_window_duration_in_milliseconds(&self) -> ForkedParam<u64> {
         ForkedParam::new(
             self.prior_target_time_per_block * self.prior_difficulty_window_size as u64,
             self.crescendo.target_time_per_block
@@ -431,29 +424,6 @@ impl Params {
         ForkedParam::new(self.prior_max_script_public_key_len, self.crescendo.max_script_public_key_len, self.crescendo_activation)
     }
 
-    /// Returns whether the sink timestamp is recent enough and the node is considered synced or nearly synced.
-    pub fn is_nearly_synced(&self, sink_timestamp: u64, sink_daa_score: u64) -> bool {
-        if self.net.is_mainnet() {
-            // We consider the node close to being synced if the sink (virtual selected parent) block
-            // timestamp is within DAA window duration far in the past. Blocks mined over such DAG state would
-            // enter the DAA window of fully-synced nodes and thus contribute to overall network difficulty
-            //
-            // [Crescendo]: both durations are nearly equal so this decision is negligible
-            unix_now() < sink_timestamp + self.expected_difficulty_window_duration_in_milliseconds().get(sink_daa_score)
-        } else {
-            // For testnets we consider the node to be synced if the sink timestamp is within a time range which
-            // is overwhelmingly unlikely to pass without mined blocks even if net hashrate decreased dramatically.
-            //
-            // This period is smaller than the above mainnet calculation in order to ensure that an IBDing miner
-            // with significant testnet hashrate does not overwhelm the network with deep side-DAGs.
-            //
-            // We use DAA duration as baseline and scale it down with BPS (and divide by 3 for mining only when very close to current time on TN11)
-            let max_expected_duration_without_blocks_in_milliseconds =
-                self.prior_target_time_per_block * NEW_DIFFICULTY_WINDOW_DURATION / 3; // = DAA duration in milliseconds / bps / 3
-            unix_now() < sink_timestamp + max_expected_duration_without_blocks_in_milliseconds
-        }
-    }
-
     pub fn network_name(&self) -> String {
         self.net.to_prefixed()
     }

consensus/core/src/config/params.rs

@@ -27,6 +27,7 @@ pub mod hashing;
 pub mod header;
 pub mod mass;
 pub mod merkle;
+pub mod mining_rules;
 pub mod muhash;
 pub mod network;
 pub mod pruning;

consensus/core/src/lib.rs

@@ -0,0 +1,14 @@
+#[derive(Debug)]
+pub struct MiningRules {}
+
+impl MiningRules {
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
+impl Default for MiningRules {
+    fn default() -> Self {
+        Self::new()
+    }
+}

consensus/core/src/mining_rules.rs

@@ -3,7 +3,7 @@ use super::utxo_set_override::{set_genesis_utxo_commitment_from_config, set_init
 use super::{ctl::Ctl, Consensus};
 use crate::{model::stores::U64Key, pipeline::ProcessingCounters};
 use itertools::Itertools;
-use kaspa_consensus_core::config::Config;
+use kaspa_consensus_core::{config::Config, mining_rules::MiningRules};
 use kaspa_consensus_notify::root::ConsensusNotificationRoot;
 use kaspa_consensusmanager::{ConsensusFactory, ConsensusInstance, DynConsensusCtl, SessionLock};
 use kaspa_core::{debug, time::unix_now, warn};
@@ -254,6 +254,7 @@ pub struct Factory {
     counters: Arc<ProcessingCounters>,
     tx_script_cache_counters: Arc<TxScriptCacheCounters>,
     fd_budget: i32,
+    mining_rules: Arc<MiningRules>,
 }
 
 impl Factory {
@@ -266,6 +267,7 @@ impl Factory {
         counters: Arc<ProcessingCounters>,
         tx_script_cache_counters: Arc<TxScriptCacheCounters>,
         fd_budget: i32,
+        mining_rules: Arc<MiningRules>,
     ) -> Self {
         assert!(fd_budget > 0, "fd_budget has to be positive");
         let mut config = config.clone();
@@ -283,6 +285,7 @@ impl Factory {
             counters,
             tx_script_cache_counters,
             fd_budget,
+            mining_rules,
         };
         factory.delete_inactive_consensus_entries();
         factory
@@ -325,6 +328,7 @@ impl ConsensusFactory for Factory {
             self.counters.clone(),
             self.tx_script_cache_counters.clone(),
             entry.creation_timestamp,
+            self.mining_rules.clone(),
         ));
 
         // We write the new active entry only once the instance was created successfully.
@@ -359,6 +363,7 @@ impl ConsensusFactory for Factory {
             self.counters.clone(),
             self.tx_script_cache_counters.clone(),
             entry.creation_timestamp,
+            self.mining_rules.clone(),
         ));
 
         (ConsensusInstance::new(session_lock, consensus.clone()), Arc::new(Ctl::new(self.management_store.clone(), db, consensus)))

consensus/src/consensus/factory.rs

@@ -65,6 +65,7 @@ use kaspa_consensus_core::{
     header::Header,
     mass::{ContextualMasses, NonContextualMasses},
     merkle::calc_hash_merkle_root,
+    mining_rules::MiningRules,
     muhash::MuHashExtensions,
     network::NetworkType,
     pruning::{PruningPointProof, PruningPointTrustedData, PruningPointsList, PruningProofMetadata},
@@ -162,6 +163,7 @@ impl Consensus {
         counters: Arc<ProcessingCounters>,
         tx_script_cache_counters: Arc<TxScriptCacheCounters>,
         creation_timestamp: u64,
+        mining_rules: Arc<MiningRules>,
     ) -> Self {
         let params = &config.params;
         let perf_params = &config.perf;
@@ -266,6 +268,7 @@ impl Consensus {
             pruning_lock.clone(),
             notification_root.clone(),
             counters.clone(),
+            mining_rules,
         ));
 
         let pruning_processor = Arc::new(PruningProcessor::new(
@@ -585,6 +588,12 @@ impl ConsensusApi for Consensus {
         self.headers_store.get_timestamp(self.get_sink()).unwrap()
     }
 
+    fn get_sink_daa_score_timestamp(&self) -> DaaScoreTimestamp {
+        let sink = self.get_sink();
+        let compact = self.headers_store.get_compact_header_data(sink).unwrap();
+        DaaScoreTimestamp { daa_score: compact.daa_score, timestamp: compact.timestamp }
+    }
+
     fn get_current_block_color(&self, hash: Hash) -> Option<bool> {
         let _guard = self.pruning_lock.blocking_read();
 
@@ -670,13 +679,6 @@ impl ConsensusApi for Consensus {
         BlockCount { header_count, block_count }
     }
 
-    fn is_nearly_synced(&self) -> bool {
-        // See comment within `config.is_nearly_synced`
-        let sink = self.get_sink();
-        let compact = self.headers_store.get_compact_header_data(sink).unwrap();
-        self.config.is_nearly_synced(compact.timestamp, compact.daa_score)
-    }
-
     fn get_virtual_chain_from_block(&self, low: Hash, chain_path_added_limit: Option<usize>) -> ConsensusResult<ChainPath> {
         // Calculate chain changes between the given `low` and the current sink hash (up to `limit` amount of block hashes).
         // Note:

consensus/src/consensus/mod.rs

@@ -1,5 +1,6 @@
 use async_channel::Sender;
 use kaspa_consensus_core::coinbase::MinerData;
+use kaspa_consensus_core::mining_rules::MiningRules;
 use kaspa_consensus_core::tx::ScriptPublicKey;
 use kaspa_consensus_core::{
     api::ConsensusApi, block::MutableBlock, blockstatus::BlockStatus, header::Header, merkle::calc_hash_merkle_root,
@@ -58,6 +59,7 @@ impl TestConsensus {
             counters,
             tx_script_cache_counters,
             0,
+            Arc::new(MiningRules::default()),
         ));
         let block_builder = TestBlockBuilder::new(consensus.virtual_processor.clone());
 
@@ -78,6 +80,7 @@ impl TestConsensus {
             counters,
             tx_script_cache_counters,
             0,
+            Arc::new(MiningRules::default()),
         ));
         let block_builder = TestBlockBuilder::new(consensus.virtual_processor.clone());
 
@@ -99,6 +102,7 @@ impl TestConsensus {
             counters,
             tx_script_cache_counters,
             0,
+            Arc::new(MiningRules::default()),
         ));
         let block_builder = TestBlockBuilder::new(consensus.virtual_processor.clone());
 

consensus/src/consensus/test_consensus.rs

@@ -60,6 +60,7 @@ use kaspa_consensus_core::{
     },
     header::Header,
     merkle::calc_hash_merkle_root,
+    mining_rules::MiningRules,
     pruning::PruningPointsList,
     tx::{MutableTransaction, Transaction},
     utxo::{
@@ -175,6 +176,9 @@ pub struct VirtualStateProcessor {
 
     // Crescendo hardfork activation score (used here for activating KIPs 9,10)
     pub(crate) crescendo_activation: ForkActivation,
+
+    // Mining Rule
+    mining_rules: Arc<MiningRules>,
 }
 
 impl VirtualStateProcessor {
@@ -191,6 +195,7 @@ impl VirtualStateProcessor {
         pruning_lock: SessionLock,
         notification_root: Arc<ConsensusNotificationRoot>,
         counters: Arc<ProcessingCounters>,
+        mining_rules: Arc<MiningRules>,
     ) -> Self {
         Self {
             receiver,
@@ -240,6 +245,7 @@ impl VirtualStateProcessor {
             counters,
             crescendo_logger: CrescendoLogger::new(),
             crescendo_activation: params.crescendo_activation,
+            mining_rules,
         }
     }
 

consensus/src/pipeline/virtual_processor/processor.rs

@@ -31,6 +31,8 @@ kaspa-index-processor.workspace = true
 kaspa-mining.workspace = true
 kaspa-notify.workspace = true
 kaspa-p2p-flows.workspace = true
+kaspa-p2p-lib.workspace = true
+kaspa-p2p-mining.workspace = true
 kaspa-perf-monitor.workspace = true
 kaspa-rpc-core.workspace = true
 kaspa-rpc-service.workspace = true