Implement Sync Rate Rule

Author: coderofstuff

protocol/mining/src/rule_engine.rs

@@ -1,14 +1,11 @@
 use std::{
-    sync::{
-        atomic::{AtomicBool, Ordering},
-        Arc,
-    },
+    sync::{atomic::AtomicBool, Arc},
     time::{Duration, Instant},
 };
 
 use kaspa_consensus_core::{
     api::counters::ProcessingCounters,
-    config::{params::NEW_DIFFICULTY_WINDOW_DURATION, Config},
+    config::Config,
     daa_score_timestamp::DaaScoreTimestamp,
     mining_rules::MiningRules,
     network::NetworkType::{Mainnet, Testnet},
@@ -20,14 +17,14 @@ use kaspa_core::{
         tick::{TickReason, TickService},
     },
     time::unix_now,
-    trace, warn,
+    trace,
 };
 use kaspa_p2p_lib::Hub;
 
-use crate::rules::mining_rule::MiningRule;
+use crate::rules::{mining_rule::MiningRule, sync_rate_rule::SyncRateRule, ExtraData};
 
 const RULE_ENGINE: &str = "mining-rule-engine";
-const SYNC_RATE_THRESHOLD: f64 = 0.10;
+pub const SNAPSHOT_INTERVAL: u64 = 10;
 
 #[derive(Clone)]
 pub struct MiningRuleEngine {
@@ -44,13 +41,11 @@ pub struct MiningRuleEngine {
 
 impl MiningRuleEngine {
     pub async fn worker(self: &Arc<MiningRuleEngine>) {
-        println!(module_path!());
-        let snapshot_interval = 10;
         let mut last_snapshot = self.processing_counters.snapshot();
         let mut last_log_time = Instant::now();
         loop {
             // START: Sync monitor
-            if let TickReason::Shutdown = self.tick_service.tick(Duration::from_secs(snapshot_interval)).await {
+            if let TickReason::Shutdown = self.tick_service.tick(Duration::from_secs(SNAPSHOT_INTERVAL)).await {
                 // Let the system print final logs before exiting
                 tokio::time::sleep(Duration::from_millis(500)).await;
                 break;
@@ -70,60 +65,24 @@ impl MiningRuleEngine {
             if elapsed_time.as_secs() > 0 {
                 let session = self.consensus_manager.consensus().unguarded_session();
                 let sink_daa_timestamp = session.async_get_sink_daa_score_timestamp().await;
-                let expected_blocks =
-                    (elapsed_time.as_millis() as u64) / self.config.target_time_per_block().get(sink_daa_timestamp.daa_score);
-                let received_blocks = delta.body_counts.max(delta.header_counts);
-                let rate: f64 = (received_blocks as f64) / (expected_blocks as f64);
 
                 let finality_point = session.async_finality_point().await;
                 let finality_point_timestamp = session.async_get_header(finality_point).await.unwrap().timestamp;
-                // Finality point is considered "recent" if it is within 3 finality durations from the current time
-                let is_finality_recent = finality_point_timestamp
-                    >= unix_now()
-                        .saturating_sub(self.config.finality_duration_in_milliseconds().get(sink_daa_timestamp.daa_score) * 3);
-
-                trace!(
-                    "Sync rate: {:.2} | Finality point recent: {} | Elapsed time: {}s | Connected: {} | Found/Expected blocks: {}/{}",
-                    rate,
-                    is_finality_recent,
-                    elapsed_time.as_secs(),
-                    delta.body_counts,
-                    self.has_sufficient_peer_connectivity(),
-                    expected_blocks,
-                );
-
-                if is_finality_recent && rate < SYNC_RATE_THRESHOLD {
-                    // if sync rate rule conditions are met:
-                    if let Ok(false) = self.use_sync_rate_rule.compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed) {
-                        warn!("Sync rate {:.2} is below threshold: {}", rate, SYNC_RATE_THRESHOLD);
-                    }
-                } else {
-                    // else when sync rate conditions are not met:
-                    if let Ok(true) = self.use_sync_rate_rule.compare_exchange(true, false, Ordering::Relaxed, Ordering::Relaxed) {
-                        if !is_finality_recent {
-                            warn!("Sync rate {:.2} recovered: {} by entering IBD", rate, SYNC_RATE_THRESHOLD);
-                        } else {
-                            warn!("Sync rate {:.2} recovered: {}", rate, SYNC_RATE_THRESHOLD);
-                        }
-                    } else if !is_finality_recent {
-                        trace!("Finality period is old. Timestamp: {}. Sync rate: {:.2}", finality_point_timestamp, rate);
-                    }
-                }
 
-                // END - Sync monitor
+                let extra_data = ExtraData {
+                    finality_point_timestamp,
+                    target_time_per_block: self.config.target_time_per_block().get(sink_daa_timestamp.daa_score),
+                    has_sufficient_peer_connectivity: self.has_sufficient_peer_connectivity(),
+                    finality_duration: self.config.finality_duration_in_milliseconds().get(sink_daa_timestamp.daa_score),
+                    elapsed_time,
+                };
 
-                // START - Rule Engine
                 trace!("Current Mining Rule: {:?}", self.mining_rules);
 
-                // Blue Parents Only Check:
+                // Check for all the rules
                 for rule in &self.rules {
-                    rule.check_rule(&snapshot);
+                    rule.check_rule(&delta, &extra_data);
                 }
-
-                // No Transactions Check:
-                // TODO: implement this part
-
-                // End - Rule Engine
             }
 
             last_snapshot = snapshot;
@@ -139,18 +98,10 @@ impl MiningRuleEngine {
         hub: Hub,
         mining_rules: Arc<MiningRules>,
     ) -> Self {
-        let rules: Vec<Arc<(dyn MiningRule + 'static)>> = vec![];
-
-        Self {
-            consensus_manager,
-            config,
-            processing_counters,
-            tick_service,
-            hub,
-            use_sync_rate_rule: Arc::new(AtomicBool::new(false)),
-            mining_rules,
-            rules,
-        }
+        let use_sync_rate_rule = Arc::new(AtomicBool::new(false));
+        let rules: Vec<Arc<(dyn MiningRule + 'static)>> = vec![Arc::new(SyncRateRule::new(use_sync_rate_rule.clone()))];
+
+        Self { consensus_manager, config, processing_counters, tick_service, hub, use_sync_rate_rule, mining_rules, rules }
     }
 
     pub fn should_mine(&self, sink_daa_score_timestamp: DaaScoreTimestamp) -> bool {
@@ -169,27 +120,16 @@ impl MiningRuleEngine {
     pub fn is_nearly_synced(&self, sink_daa_score_timestamp: DaaScoreTimestamp) -> bool {
         let sink_timestamp = sink_daa_score_timestamp.timestamp;
 
-        if self.config.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.config.expected_difficulty_window_duration_in_milliseconds().get(sink_daa_score_timestamp.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 10BPS testnets)
-            let max_expected_duration_without_blocks_in_milliseconds =
-                self.config.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
-        }
+        // We consider the node close to being synced if the sink (virtual selected parent) block
+        // timestamp is within a quarter of the 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
+        let synced_threshold =
+            self.config.expected_difficulty_window_duration_in_milliseconds().get(sink_daa_score_timestamp.daa_score) / 4;
+
+        // Roughly 10mins in all networks
+        unix_now() < sink_timestamp + synced_threshold
     }
 
     fn has_sufficient_peer_connectivity(&self) -> bool {

protocol/mining/src/rules/mining_rule.rs

@@ -1,5 +1,7 @@
 use kaspa_consensus_core::api::counters::ProcessingCountersSnapshot;
 
+use super::ExtraData;
+
 pub trait MiningRule: Send + Sync + 'static {
-    fn check_rule(&self, delta: &ProcessingCountersSnapshot);
+    fn check_rule(&self, delta: &ProcessingCountersSnapshot, extra_data: &ExtraData);
 }

protocol/mining/src/rules/mod.rs

@@ -1 +1,13 @@
+use std::time::Duration;
+
+pub mod sync_rate_rule;
+
 pub mod mining_rule;
+
+pub struct ExtraData {
+    pub finality_point_timestamp: u64,
+    pub target_time_per_block: u64,
+    pub has_sufficient_peer_connectivity: bool,
+    pub finality_duration: u64,
+    pub elapsed_time: Duration,
+}

protocol/mining/src/rules/sync_rate_rule.rs

@@ -0,0 +1,114 @@
+use std::{
+    collections::VecDeque,
+    sync::{
+        atomic::{AtomicBool, AtomicU64, Ordering},
+        Arc, RwLock,
+    },
+};
+
+use kaspa_consensus_core::api::counters::ProcessingCountersSnapshot;
+use kaspa_core::{time::unix_now, trace, warn};
+
+use crate::rule_engine::SNAPSHOT_INTERVAL;
+
+use super::{mining_rule::MiningRule, ExtraData};
+
+// within a 5 minute period, we expect sync rate
+const SYNC_RATE_THRESHOLD: f64 = 0.90;
+// number of samples you expect in a 5 minute interval, sampled every 10s
+const SYNC_RATE_WINDOW_MAX_SIZE: usize = 5 * 60 / (SNAPSHOT_INTERVAL as usize);
+// number of samples required before considering this rule. This allows using the sync rate rule
+// even before the full window size is reached. Represents the number of samples in 1 minute
+const SYNC_RATE_WINDOW_MIN_THRESHOLD: usize = 60 / (SNAPSHOT_INTERVAL as usize);
+
+pub struct SyncRateRule {
+    pub use_sync_rate_rule: Arc<AtomicBool>,
+    sync_rate_samples: RwLock<VecDeque<(u64, u64)>>,
+    total_expected_blocks: AtomicU64,
+    total_received_blocks: AtomicU64,
+}
+
+impl SyncRateRule {
+    pub fn new(use_sync_rate_rule: Arc<AtomicBool>) -> Self {
+        Self {
+            use_sync_rate_rule,
+            sync_rate_samples: RwLock::new(VecDeque::new()),
+            total_expected_blocks: AtomicU64::new(0),
+            total_received_blocks: AtomicU64::new(0),
+        }
+    }
+
+    /// Adds current observation of received and expected blocks to the sample window, and removes
+    /// old samples. Returns true if there are enough samples in the window to start triggering the
+    /// sync rate rule.
+    fn update_sync_rate_window(&self, received_blocks: u64, expected_blocks: u64) -> bool {
+        self.total_received_blocks.fetch_add(received_blocks, Ordering::SeqCst);
+        self.total_expected_blocks.fetch_add(expected_blocks, Ordering::SeqCst);
+
+        let mut samples = self.sync_rate_samples.write().unwrap();
+
+        samples.push_back((received_blocks, expected_blocks));
+
+        // Remove old samples. Usually is a single op after the window is full per 10s:
+        while samples.len() > SYNC_RATE_WINDOW_MAX_SIZE {
+            let (old_received_blocks, old_expected_blocks) = samples.pop_front().unwrap();
+            self.total_received_blocks.fetch_sub(old_received_blocks, Ordering::SeqCst);
+            self.total_expected_blocks.fetch_sub(old_expected_blocks, Ordering::SeqCst);
+        }
+
+        samples.len() >= SYNC_RATE_WINDOW_MIN_THRESHOLD
+    }
+}
+
+/// SyncRateRule
+/// Allow mining even if the node is "not nearly synced" if the sync rate is below threshold
+/// and the finality point is recent. This is to prevent the network from undermining and to allow
+/// the network to automatically recover from any short-term mining halt.
+///
+/// Trigger: Sync rate is below threshold and finality point is recent
+/// Recovery: Sync rate is back above threshold
+impl MiningRule for SyncRateRule {
+    fn check_rule(&self, delta: &ProcessingCountersSnapshot, extra_data: &ExtraData) {
+        let expected_blocks = (extra_data.elapsed_time.as_millis() as u64) / extra_data.target_time_per_block;
+        let received_blocks = delta.body_counts.max(delta.header_counts);
+
+        if !self.update_sync_rate_window(received_blocks, expected_blocks) {
+            // Don't process the sync rule if the window doesn't have enough samples to filter out noise
+            return;
+        }
+
+        let rate: f64 =
+            (self.total_received_blocks.load(Ordering::SeqCst) as f64) / (self.total_expected_blocks.load(Ordering::SeqCst) as f64);
+
+        // Finality point is considered "recent" if it is within 3 finality durations from the current time
+        let is_finality_recent = extra_data.finality_point_timestamp >= unix_now().saturating_sub(extra_data.finality_duration * 3);
+
+        trace!(
+            "Sync rate: {:.2} | Finality point recent: {} | Elapsed time: {}s | Connected: {} | Found/Expected blocks: {}/{}",
+            rate,
+            is_finality_recent,
+            extra_data.elapsed_time.as_secs(),
+            extra_data.has_sufficient_peer_connectivity,
+            delta.body_counts,
+            expected_blocks,
+        );
+
+        if is_finality_recent && rate < SYNC_RATE_THRESHOLD {
+            // if sync rate rule conditions are met:
+            if let Ok(false) = self.use_sync_rate_rule.compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed) {
+                warn!("Sync rate {:.2} is below threshold: {}", rate, SYNC_RATE_THRESHOLD);
+            }
+        } else {
+            // else when sync rate conditions are not met:
+            if let Ok(true) = self.use_sync_rate_rule.compare_exchange(true, false, Ordering::Relaxed, Ordering::Relaxed) {
+                if !is_finality_recent {
+                    warn!("Sync rate {:.2} recovered: {} by entering IBD", rate, SYNC_RATE_THRESHOLD);
+                } else {
+                    warn!("Sync rate {:.2} recovered: {}", rate, SYNC_RATE_THRESHOLD);
+                }
+            } else if !is_finality_recent {
+                trace!("Finality period is old. Timestamp: {}. Sync rate: {:.2}", extra_data.finality_point_timestamp, rate);
+            }
+        }
+    }
+}