diff --git a/dash-spv-ffi/src/callbacks.rs b/dash-spv-ffi/src/callbacks.rs index cad8c20d0..e62ec454b 100644 --- a/dash-spv-ffi/src/callbacks.rs +++ b/dash-spv-ffi/src/callbacks.rs @@ -383,6 +383,7 @@ impl FFISyncEventCallbacks { } SyncEvent::MasternodeStateUpdated { height, + .. } => { if let Some(cb) = self.on_masternode_state_updated { cb(*height, self.user_data); diff --git a/dash-spv/src/sync/events.rs b/dash-spv/src/sync/events.rs index 86fe6ec76..dad4de4c0 100644 --- a/dash-spv/src/sync/events.rs +++ b/dash-spv/src/sync/events.rs @@ -1,6 +1,7 @@ use crate::sync::ManagerIdentifier; use dashcore::ephemerealdata::chain_lock::ChainLock; use dashcore::ephemerealdata::instant_lock::InstantLock; +use dashcore::sml::masternode_list_engine::QRInfoFeedResult; use dashcore::{Address, BlockHash, Txid}; use key_wallet_manager::{FilterMatchKey, WalletId}; use std::collections::{BTreeMap, BTreeSet}; @@ -117,6 +118,14 @@ pub enum SyncEvent { MasternodeStateUpdated { /// New masternode state height height: u32, + /// QRInfo processing result when this update came through the + /// QuorumValidation pipeline. `None` for Incremental (MnListDiff-only) + /// updates. Consumers that care about rotation cycle storage (e.g. + /// IS lock verification across rotation) can gate on + /// `result.all_fully_verified()` together with + /// `result.stored_cycle_height` to know which cycle was fully + /// verified and stored in `rotated_quorums_per_cycle` by this update. + qr_info_result: Option, }, /// A manager encountered a recoverable error. @@ -225,9 +234,18 @@ impl SyncEvent { } SyncEvent::MasternodeStateUpdated { height, - } => { - format!("MasternodeStateUpdated(height={})", height) - } + qr_info_result, + } => match qr_info_result { + Some(s) => format!( + "MasternodeStateUpdated(height={}, qr_info={{stored_cycle_height={:?}, verified={}/{}, newly_qualified={}}})", + height, + s.stored_cycle_height, + s.fully_verified_count, + s.rotated_quorum_count, + s.newly_qualified_count, + ), + None => format!("MasternodeStateUpdated(height={})", height), + }, SyncEvent::ManagerError { manager, error, diff --git a/dash-spv/src/sync/masternodes/manager.rs b/dash-spv/src/sync/masternodes/manager.rs index 02fa3e0d9..5f3c58833 100644 --- a/dash-spv/src/sync/masternodes/manager.rs +++ b/dash-spv/src/sync/masternodes/manager.rs @@ -8,7 +8,7 @@ use std::sync::Arc; use std::time::Instant; use dashcore::sml::llmq_type::network::NetworkLLMQExt; -use dashcore::sml::masternode_list_engine::MasternodeListEngine; +use dashcore::sml::masternode_list_engine::{MasternodeListEngine, QRInfoFeedResult}; use tokio::sync::RwLock; use super::pipeline::MnListDiffPipeline; @@ -16,6 +16,7 @@ use crate::error::{SyncError, SyncResult}; use crate::network::RequestSender; use crate::storage::BlockHeaderStorage; use crate::sync::{MasternodesProgress, SyncEvent, SyncManager, SyncState}; +use dashcore::network::message_qrinfo::QRInfo; use dashcore::BlockHash; use std::collections::BTreeSet; @@ -26,6 +27,54 @@ use std::collections::BTreeSet; /// `false` at the call site. const QRINFO_ANCHOR_CYCLES_BEHIND: u32 = 4; +/// Single enum that serves two roles in the masternode-sync flow: +/// +/// - **Decision** — returned from [`MasternodesManager::next_pipeline_mode`] to pick +/// which request to fire when a new header lands while sync is `Synced`. +/// - **State** — stored on [`MasternodeSyncState::pipeline_mode`] to record what the +/// mnlistdiff pipeline is currently running, so [`MasternodesManager::complete_pipeline`] +/// can dispatch the right completion flow when the pipeline drains. +/// +/// The two variants map 1:1 between the two roles: +/// +/// | Variant | Decision action | Completion flow | +/// |---------------------|----------------------------------------------|------------------------------------------| +/// | `QuorumValidation` | Fire `getqrinfo` (which queues historical diffs for non-rotating quorum verification). | Full `verify_and_complete`: hard-fails into `Error` on verification failure, transitions initial sync to `Synced` on success. | +/// | `Incremental` | Fire a targeted `GetMnListDiff` from the latest known masternode list tip to the new header tip. | Lightweight verification at the latest height. On failure, log warn and stay in `Synced`. A single failed tip refresh should not kill the whole sync state. | +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) enum PipelineMode { + /// Full `getqrinfo` request / post-QRInfo historical cycle diffs. See enum docs. + /// + /// `qr_info_result` is set by the QRInfo message handler when a response is + /// successfully fed to the engine, and is consumed by `complete_pipeline` when + /// the mnlistdiff pipeline drains. `None` while the pipeline is being set up + /// or between cycles. + QuorumValidation { + qr_info_result: Option, + }, + /// Targeted single-diff tip refresh. See enum docs. + Incremental, +} + +impl Default for PipelineMode { + fn default() -> Self { + Self::QuorumValidation { + qr_info_result: None, + } + } +} + +/// In-flight QRInfo request: tip hash that was requested and when the request fired. +/// Held inside `MasternodeSyncState::qrinfo_in_flight` while a QRInfo is outstanding. +#[derive(Debug, Clone, Copy)] +pub(super) struct QRInfoInFlight { + /// Tip block hash of the request. Used to reject late responses for a previously + /// requested tip after a timeout retry has rotated the active tip. + pub(super) tip: BlockHash, + /// When the request was fired. Used by the timeout check. + pub(super) wait_start: Instant, +} + /// Sync state for masternode list synchronization. #[derive(Debug, Default)] pub(super) struct MasternodeSyncState { @@ -33,15 +82,40 @@ pub(super) struct MasternodeSyncState { pub(super) known_mn_list_heights: BTreeSet, /// Pipeline for MnListDiff requests. pub(super) mnlistdiff_pipeline: MnListDiffPipeline, - /// Whether we are waiting for a QRInfo response. - pub(super) waiting_for_qrinfo: bool, - /// When we started waiting for QRInfo response. - pub(super) qrinfo_wait_start: Option, + /// What the pipeline is currently being used for. See [`PipelineMode`]. + pub(super) pipeline_mode: PipelineMode, + /// Active QRInfo request, if any. `Some` between firing the request and either + /// processing the response or timing out. Carries the requested tip so a late + /// straggler from a previously requested tip can be rejected after a retry. + pub(super) qrinfo_in_flight: Option, /// Current retry count for QRInfo. pub(super) qrinfo_retry_count: u8, - /// When to retry after a ChainLock unavailability error. - /// The QRInfo response includes the current tip which may not have ChainLock yet. - pub(super) chainlock_retry_after: Option, + /// Block hash of the latest masternode list the engine holds. Initialized from + /// engine state on startup (so it survives restarts) and refreshed after every + /// successful pipeline completion. + pub(super) last_synced_block_hash: Option, + /// Rotation cycle boundary heights we have successfully freshly-validated. Used + /// to stop firing QRInfo for a cycle once its rotated quorums are verified. + /// Subsequent tip updates within the same cycle take the `Incremental` path. + pub(super) validated_cycle_heights: BTreeSet, + /// Current cycle boundary height the in-cycle tracking is for. Resets on cycle + /// change. + pub(super) current_cycle_height: Option, + /// Number of QRInfo attempts fired for `current_cycle_height`. Used for the + /// one-shot degraded-cycle log message. There is no hard cap. QRInfo is fired + /// on every new block inside the mining window until one succeeds. + pub(super) current_cycle_attempts: u8, + /// Highest tip height a QRInfo has already been fired for inside the current + /// cycle's mining window. Gates `next_pipeline_mode` so that unrelated ticks + /// (peer events, response receipt, timers) cannot re-fire QRInfo for the same + /// tip when validation fails deterministically. Reset on cycle rollover. + pub(super) last_window_qrinfo_tip: Option, + /// Block hash of the most recently successfully processed QRInfo's `mn_list_diff_tip`. + /// A response carrying the same tip hash as the last successful processing is dropped + /// at handler entry. This defends against the case where `qrinfo_in_flight` is set + /// (because a new request was already fired for a newer tip) but a late straggler from + /// a previous tip's request still arrives. + pub(super) last_processed_qrinfo_tip: Option, } /// Pick the QRInfo base anchor for a request at `tip_height`: the highest stored @@ -77,23 +151,69 @@ impl MasternodeSyncState { } pub(super) fn has_pending_requests(&self) -> bool { - !self.mnlistdiff_pipeline.is_complete() || self.waiting_for_qrinfo + !self.mnlistdiff_pipeline.is_complete() || self.qrinfo_in_flight.is_some() } pub(super) fn clear_pending(&mut self) { self.mnlistdiff_pipeline.clear(); - self.waiting_for_qrinfo = false; - self.qrinfo_wait_start = None; + self.qrinfo_in_flight = None; + self.pipeline_mode = PipelineMode::default(); } - fn start_waiting_for_qrinfo(&mut self) { - self.waiting_for_qrinfo = true; - self.qrinfo_wait_start = Some(Instant::now()); + /// Record that a QRInfo request was actually fired for `tip_height`. Bumps + /// the per-cycle attempt counter and sets the per-tip gate so subsequent + /// calls to [`MasternodesManager::next_pipeline_mode`] at the same tip + /// return `Incremental` instead of refiring. + pub(super) fn record_qrinfo_attempt(&mut self, tip_height: u32) { + self.last_window_qrinfo_tip = Some(tip_height); + self.current_cycle_attempts = self.current_cycle_attempts.saturating_add(1); + } + + fn start_waiting_for_qrinfo(&mut self, expected_tip: BlockHash) { + self.qrinfo_in_flight = Some(QRInfoInFlight { + tip: expected_tip, + wait_start: Instant::now(), + }); } pub(super) fn qrinfo_received(&mut self) { - self.waiting_for_qrinfo = false; - self.qrinfo_wait_start = None; + self.qrinfo_in_flight = None; + } + + /// Decide whether an incoming QRInfo should be processed by the handler. + /// + /// Drops: + /// - Duplicates of the last successfully processed tip (late straggler from a + /// previous request whose response already won). + /// - Unsolicited responses (no QRInfo request currently in flight). + /// - Responses whose tip does not match the active in-flight request tip + /// (late straggler from a previous tip whose request was rotated by a + /// timeout retry). + pub(super) fn should_process_qrinfo(&self, qr_info: &QRInfo) -> bool { + let tip = qr_info.mn_list_diff_tip.block_hash; + if self.last_processed_qrinfo_tip == Some(tip) { + tracing::debug!( + tip = %tip, + "Dropping duplicate QRInfo (same tip already processed)" + ); + return false; + } + let Some(in_flight) = self.qrinfo_in_flight else { + tracing::debug!( + tip = %tip, + "Ignoring unsolicited/late QRInfo" + ); + return false; + }; + if in_flight.tip != tip { + tracing::debug!( + tip = %tip, + expected = %in_flight.tip, + "Dropping QRInfo for non-active request tip" + ); + return false; + } + true } } @@ -126,9 +246,15 @@ impl MasternodesManager { engine: Arc>, network: dashcore::Network, ) -> Self { - // Load current height from engine's masternode lists - let current_height = - engine.read().await.masternode_lists.keys().last().copied().unwrap_or(0); + // Recover sync state from the engine's stored masternode lists so that a + // restart can resume from where the previous run left off. + let (current_height, last_synced_block_hash) = { + let engine_guard = engine.read().await; + match engine_guard.masternode_lists.iter().next_back() { + Some((&height, list)) => (height, Some(list.block_hash)), + None => (0, None), + } + }; // Load block header tip for progress display let header_tip = @@ -140,15 +266,201 @@ impl MasternodesManager { initial_progress.update_block_header_tip_height(header_tip); initial_progress.set_state(SyncState::WaitingForConnections); + let mut sync_state = MasternodeSyncState::new(); + sync_state.last_synced_block_hash = last_synced_block_hash; + Self { progress: initial_progress, header_storage, engine, network, - sync_state: MasternodeSyncState::new(), + sync_state, + } + } + + /// Decide which [`PipelineMode`] to use when a new header lands at `tip_height` + /// and masternode sync needs to catch up. The rule is: + /// + /// - Before `cycle_start + dkgMiningWindowStart`: the rotated commitment for this + /// cycle cannot possibly have been mined yet, so a QRInfo would fail at the `tip` slot. + /// Return `Incremental` to fire a targeted `GetMnListDiff` that keeps the tip + /// list fresh. + /// - Inside `[cycle_start + dkgMiningWindowStart, cycle_start + dkgMiningWindowEnd]` + /// and the cycle is not yet validated: return `QuorumValidation` so a full + /// QRInfo fires on every new header. Any block in this window can be the one + /// that contains the commit, and firing on every block gives the earliest + /// success path to fresh rotated quorum validation. The mining window is short + /// (e.g. 9 blocks for `llmq_60_75`), so the per-cycle request volume is + /// naturally bounded by the window length. + /// - Once `feed_qr_info` returns a result where every rotated quorum was freshly + /// validated, `mark_cycle_validated` records the cycle done and every + /// subsequent header in that cycle falls through to `Incremental`. + /// - After `cycle_start + dkgMiningWindowEnd` without a successful validation: + /// the cycle is degraded (DKG likely failed or commits were never mined). Log + /// the condition and fall through to `Incremental` for the remainder of the + /// cycle. + /// + /// This applies only to the incremental-update path while state is `Synced`. + /// Initial sync and explicit retry paths (timeout) bypass it. + pub(super) fn next_pipeline_mode(&mut self, tip_height: u32) -> PipelineMode { + let params = self.network.isd_llmq_type().params(); + let dkg_interval = params.dkg_params.interval; + if dkg_interval == 0 { + return PipelineMode::QuorumValidation { + qr_info_result: None, + }; + } + let mining_start = params.dkg_params.mining_window_start; + let mining_end = params.dkg_params.mining_window_end; + let cycle_height = tip_height - (tip_height % dkg_interval); + + // Reset per-cycle tracking when the tip enters a new cycle. + if self.sync_state.current_cycle_height != Some(cycle_height) { + self.sync_state.current_cycle_height = Some(cycle_height); + self.sync_state.current_cycle_attempts = 0; + self.sync_state.last_window_qrinfo_tip = None; + self.progress.add_rotation_cycles(1); + } + + // Already validated this cycle? Keep the tip list fresh but don't touch QRInfo. + if self.sync_state.validated_cycle_heights.contains(&cycle_height) { + return PipelineMode::Incremental; + } + // Before mining window opens: QRInfo would fail at the `tip` slot. Keep tip list fresh. + if tip_height < cycle_height + mining_start { + return PipelineMode::Incremental; + } + // Past mining window without success. + if tip_height > cycle_height + mining_end { + // If we never attempted QRInfo for this cycle (all blocks arrived + // in a batch that overshot the window), fire ONE QRInfo now so the + // cycle's rotated quorums get stored. Without this, IS locks from + // the new cycle can't be verified. + if self.sync_state.current_cycle_attempts == 0 { + tracing::info!( + cycle_height, + tip_height, + "Mining window missed (blocks batched); firing catch-up QRInfo" + ); + return PipelineMode::QuorumValidation { + qr_info_result: None, + }; + } + tracing::warn!( + cycle_height, + mining_window_start = cycle_height + mining_start, + mining_window_end = cycle_height + mining_end, + attempts = self.sync_state.current_cycle_attempts, + "Rotated quorum fresh validation failed for cycle: mining window \ + closed without a successful QRInfo response. Falling back to \ + mnlistdiff-only tip updates for the remainder of this cycle." + ); + return PipelineMode::Incremental; + } + + // Inside the mining window and not yet validated: pick QRInfo once per + // new tip. The per-tip gate is set when the caller actually fires via + // `record_qrinfo_attempt`, so unrelated ticks at the same tip fall + // through to `Incremental` only after a real fire. + if self.sync_state.last_window_qrinfo_tip == Some(tip_height) { + tracing::trace!( + tip_height, + cycle_height, + attempts = self.sync_state.current_cycle_attempts, + "next_pipeline_mode: QRInfo already fired for this tip, picking Incremental" + ); + return PipelineMode::Incremental; + } + PipelineMode::QuorumValidation { + qr_info_result: None, + } + } + + /// Mark a cycle boundary height as freshly validated, so `next_pipeline_mode` + /// will return `Incremental` for any future tip update in this cycle. Called + /// after a successful `feed_qr_info` where every rotated quorum was freshly + /// validated. + pub(super) fn mark_cycle_validated(&mut self, cycle_height: u32) { + if self.sync_state.validated_cycle_heights.insert(cycle_height) { + self.progress.add_validated_cycles(1); } } + /// Fire a targeted `GetMnListDiff` from the latest known masternode list tip to + /// the current header tip, to keep the tip list fresh without running a full + /// QRInfo. Sets `pipeline_mode = Incremental` so `complete_pipeline()` takes the + /// lightweight completion path when the response drains the pipeline. + pub(super) async fn send_tip_mnlistdiff_update( + &mut self, + requests: &RequestSender, + ) -> SyncResult> { + let new_tip_hash = { + let storage = self.header_storage.read().await; + match storage.get_tip().await { + Some(tip) => *tip.hash(), + None => return Ok(vec![]), + } + }; + + let Some(base_hash) = self.sync_state.last_synced_block_hash else { + // No stored masternode list at all, so a targeted diff is not possible. + // This should only happen transiently before the first successful sync. + return Ok(vec![]); + }; + + if base_hash == new_tip_hash { + return Ok(vec![]); + } + + self.sync_state.pipeline_mode = PipelineMode::Incremental; + self.sync_state.mnlistdiff_pipeline.queue_requests(vec![(base_hash, new_tip_hash)]); + self.sync_state.mnlistdiff_pipeline.send_pending(requests)?; + Ok(vec![]) + } + + /// Dispatch pipeline completion based on the current `PipelineMode`. Called when + /// the mnlistdiff pipeline drains, from either the message handler or the tick + /// handler's timeout-cleanup path. + pub(super) async fn complete_pipeline(&mut self) -> SyncResult> { + match std::mem::take(&mut self.sync_state.pipeline_mode) { + PipelineMode::QuorumValidation { + qr_info_result, + } => self.verify_and_complete(qr_info_result).await, + PipelineMode::Incremental => self.complete_incremental_pipeline().await, + } + } + + /// Complete the Incremental pipeline: verify non-rotating quorums at the latest + /// engine height and update progress on success. On verification failure, log at + /// warn level and return `Ok(vec![])` without changing state. A single failed + /// tip refresh should not bounce the whole sync into Error. + async fn complete_incremental_pipeline(&mut self) -> SyncResult> { + let mut engine = self.engine.write().await; + let Some((&height, list)) = engine.masternode_lists.iter().next_back() else { + return Ok(vec![]); + }; + let latest_block_hash = list.block_hash; + + if let Err(e) = engine.verify_non_rotating_masternode_list_quorums(height, &[]) { + tracing::warn!( + height, + "Incremental quorum verification failed, keeping previous state: {}", + e + ); + drop(engine); + return Ok(vec![]); + } + drop(engine); + + self.sync_state.last_synced_block_hash = Some(latest_block_hash); + self.progress.update_current_height(height); + tracing::debug!("Incremental MnListDiff complete at height {}", height); + Ok(vec![SyncEvent::MasternodeStateUpdated { + height, + qr_info_result: None, + }]) + } + /// Send QRInfo request for the current tip. /// /// Called when BlockHeaderSyncComplete is received, ensuring we have all headers. @@ -192,8 +504,10 @@ impl MasternodesManager { base_hashes.len() ); requests.request_qr_info(base_hashes, tip_block_hash, true)?; + self.progress.add_qr_infos_requested(1); + self.sync_state.record_qrinfo_attempt(tip_height); - self.sync_state.start_waiting_for_qrinfo(); + self.sync_state.start_waiting_for_qrinfo(tip_block_hash); Ok(vec![]) } @@ -202,14 +516,18 @@ impl MasternodesManager { /// /// For initial sync (state == Syncing), emits MasternodeStateUpdated and logs completion. /// For incremental updates (state == Synced), updates quietly without events. - pub(super) async fn verify_and_complete(&mut self) -> SyncResult> { + pub(super) async fn verify_and_complete( + &mut self, + qr_info_result: Option, + ) -> SyncResult> { let mut events = Vec::new(); let is_initial_sync = self.state() == SyncState::Syncing; let mut engine = self.engine.write().await; // Get the latest height from the engine and verify at that height - if let Some(&height) = engine.masternode_lists.keys().last() { + if let Some((&height, list)) = engine.masternode_lists.iter().next_back() { + let latest_block_hash = list.block_hash; if let Err(e) = engine.verify_non_rotating_masternode_list_quorums(height, &[]) { drop(engine); self.set_state(SyncState::Error); @@ -221,10 +539,12 @@ impl MasternodesManager { tracing::info!("Non-rotating quorum verification completed at height {}", height); + self.sync_state.last_synced_block_hash = Some(latest_block_hash); self.progress.update_current_height(height); events.push(SyncEvent::MasternodeStateUpdated { height, + qr_info_result, }); } else if is_initial_sync { drop(engine); @@ -261,12 +581,14 @@ mod tests { type TestMasternodesManager = MasternodesManager; - async fn create_test_manager() -> TestMasternodesManager { + async fn create_test_manager_for(network: dashcore::Network) -> TestMasternodesManager { let storage = DiskStorageManager::with_temp_dir().await.unwrap(); - let engine = Arc::new(RwLock::new(MasternodeListEngine::default_for_network( - dashcore::Network::Testnet, - ))); - MasternodesManager::new(storage.block_headers(), engine, dashcore::Network::Testnet).await + let engine = Arc::new(RwLock::new(MasternodeListEngine::default_for_network(network))); + MasternodesManager::new(storage.block_headers(), engine, network).await + } + + async fn create_test_manager() -> TestMasternodesManager { + create_test_manager_for(dashcore::Network::Testnet).await } #[tokio::test] @@ -286,12 +608,18 @@ mod tests { manager.progress.update_current_height(500); manager.progress.update_target_height(1000); manager.progress.add_diffs_processed(10); + manager.progress.add_qr_infos_requested(3); + manager.progress.add_validated_cycles(2); + manager.progress.add_rotation_cycles(4); let progress = manager.progress(); if let SyncManagerProgress::Masternodes(progress) = progress { assert_eq!(progress.current_height(), 500); assert_eq!(progress.target_height(), 1000); assert_eq!(progress.diffs_processed(), 10); + assert_eq!(progress.qr_infos_requested(), 3); + assert_eq!(progress.validated_cycles(), 2); + assert_eq!(progress.rotation_cycles(), 4); assert!(progress.last_activity().elapsed().as_secs() < 1); } else { panic!("Expected SyncManagerProgress::Masternodes"); @@ -366,4 +694,148 @@ mod tests { assert_eq!(got, case.expect.map(anchor_hash), "case: {}", case.name); } } + + // Regtest `isd_llmq_type` uses `DKG_TEST_DIP0024` (`interval=24`, + // `mining_window=[12, 20]`). Cycle 48 → window `[60, 68]`; cycle 72 → + // window `[84, 92]`. + #[tokio::test] + async fn test_next_pipeline_mode_fires_qrinfo_once_per_tip() { + let mut manager = create_test_manager_for(dashcore::Network::Regtest).await; + + // First call inside cycle 48's window picks QuorumValidation. The + // cycle rollover from None into cycle 48 also bumps `rotation_cycles` + // to 1. Per-tip state is not yet bumped because `next_pipeline_mode` + // is purely a decider. The caller bumps state via `record_qrinfo_attempt` + // when it actually fires. + assert!(matches!(manager.next_pipeline_mode(60), PipelineMode::QuorumValidation { .. })); + assert_eq!(manager.sync_state.current_cycle_attempts, 0); + assert_eq!(manager.sync_state.last_window_qrinfo_tip, None); + assert_eq!(manager.progress.rotation_cycles(), 1); + + // Simulate the caller firing the QRInfo. + manager.sync_state.record_qrinfo_attempt(60); + assert_eq!(manager.sync_state.current_cycle_attempts, 1); + assert_eq!(manager.sync_state.last_window_qrinfo_tip, Some(60)); + + // Re-entering with the same tip after a fire falls through to + // Incremental. The per-tip gate prevents refiring and the cycle is + // unchanged. + assert!(matches!(manager.next_pipeline_mode(60), PipelineMode::Incremental)); + assert_eq!(manager.sync_state.current_cycle_attempts, 1); + assert_eq!(manager.progress.rotation_cycles(), 1); + + // A new tip inside the same window picks QuorumValidation again. + assert!(matches!(manager.next_pipeline_mode(61), PipelineMode::QuorumValidation { .. })); + manager.sync_state.record_qrinfo_attempt(61); + assert_eq!(manager.sync_state.current_cycle_attempts, 2); + assert_eq!(manager.sync_state.last_window_qrinfo_tip, Some(61)); + + // Same tip again: still Incremental. + assert!(matches!(manager.next_pipeline_mode(61), PipelineMode::Incremental)); + assert_eq!(manager.sync_state.current_cycle_attempts, 2); + + // Cycle rollover to cycle 72 resets the per-tip gate, so the first tip + // inside the new window picks QuorumValidation and `rotation_cycles` + // bumps to 2. + assert!(matches!(manager.next_pipeline_mode(84), PipelineMode::QuorumValidation { .. })); + assert_eq!(manager.sync_state.current_cycle_height, Some(72)); + assert_eq!(manager.sync_state.current_cycle_attempts, 0); + assert_eq!(manager.sync_state.last_window_qrinfo_tip, None); + assert_eq!(manager.progress.rotation_cycles(), 2); + manager.sync_state.record_qrinfo_attempt(84); + assert_eq!(manager.sync_state.current_cycle_attempts, 1); + assert_eq!(manager.sync_state.last_window_qrinfo_tip, Some(84)); + assert!(matches!(manager.next_pipeline_mode(84), PipelineMode::Incremental)); + + // If the caller decides not to fire (e.g. another QRInfo is already in + // flight), the per-tip gate stays open so the next call re-picks + // QuorumValidation. The decider must not eagerly burn the gate. + let mut manager = create_test_manager_for(dashcore::Network::Regtest).await; + assert!(matches!(manager.next_pipeline_mode(60), PipelineMode::QuorumValidation { .. })); + assert!(matches!(manager.next_pipeline_mode(60), PipelineMode::QuorumValidation { .. })); + assert_eq!(manager.sync_state.current_cycle_attempts, 0); + assert_eq!(manager.sync_state.last_window_qrinfo_tip, None); + + // Tip before the mining window opens stays Incremental and does not + // touch per-cycle state. + let mut manager = create_test_manager_for(dashcore::Network::Regtest).await; + assert!(matches!(manager.next_pipeline_mode(50), PipelineMode::Incremental)); + assert_eq!(manager.sync_state.current_cycle_height, Some(48)); + assert_eq!(manager.sync_state.current_cycle_attempts, 0); + assert_eq!(manager.sync_state.last_window_qrinfo_tip, None); + + // Tip past the mining window with no prior attempts picks the catch-up + // QuorumValidation. After the caller fires, subsequent calls at any + // tip past the window fall through to Incremental for the rest of the + // cycle. `rotation_cycles` bumps once for entering the cycle and again + // when the tip crosses into the next cycle. + let mut manager = create_test_manager_for(dashcore::Network::Regtest).await; + assert!(matches!(manager.next_pipeline_mode(70), PipelineMode::QuorumValidation { .. })); + assert_eq!(manager.sync_state.current_cycle_height, Some(48)); + assert_eq!(manager.sync_state.current_cycle_attempts, 0); + assert_eq!(manager.progress.rotation_cycles(), 1); + manager.sync_state.record_qrinfo_attempt(70); + assert!(matches!(manager.next_pipeline_mode(70), PipelineMode::Incremental)); + assert!(matches!(manager.next_pipeline_mode(71), PipelineMode::Incremental)); + assert_eq!(manager.progress.rotation_cycles(), 1); + assert!(matches!(manager.next_pipeline_mode(96), PipelineMode::Incremental)); + assert_eq!(manager.sync_state.current_cycle_height, Some(96)); + assert_eq!(manager.progress.rotation_cycles(), 2); + + // `mark_cycle_validated` short-circuits any subsequent tip in that + // cycle to Incremental, even tips inside the mining window. Calling + // it twice for the same cycle bumps `validated_cycles` only once. + let mut manager = create_test_manager_for(dashcore::Network::Regtest).await; + manager.mark_cycle_validated(48); + manager.mark_cycle_validated(48); + assert_eq!(manager.progress.validated_cycles(), 1); + assert!(matches!(manager.next_pipeline_mode(60), PipelineMode::Incremental)); + assert!(matches!(manager.next_pipeline_mode(65), PipelineMode::Incremental)); + assert!(matches!(manager.next_pipeline_mode(50), PipelineMode::Incremental)); + } + + /// On restart, `MasternodesManager::new` must recover + /// `last_synced_block_hash` from the engine's stored masternode lists so + /// the next pipeline run can target the correct base. Without recovery, + /// `send_tip_mnlistdiff_update` would early-return for lack of a base + /// hash and the SPV would re-run the full QRInfo flow on every restart + /// instead of resuming. + #[tokio::test] + async fn test_masternode_manager_recovers_last_synced_hash_from_engine() { + let storage = DiskStorageManager::with_temp_dir().await.unwrap(); + let mut engine = MasternodeListEngine::default_for_network(dashcore::Network::Testnet); + let tip_hash = BlockHash::from_byte_array([0xAB; 32]); + let mid_hash = BlockHash::from_byte_array([0xCD; 32]); + engine.masternode_lists.insert(100, MasternodeList::empty(mid_hash, 100)); + engine.masternode_lists.insert(200, MasternodeList::empty(tip_hash, 200)); + + let manager = MasternodesManager::new( + storage.block_headers(), + Arc::new(RwLock::new(engine)), + dashcore::Network::Testnet, + ) + .await; + + assert_eq!( + manager.sync_state.last_synced_block_hash, + Some(tip_hash), + "new() must recover last_synced_block_hash from the engine's tip list" + ); + assert_eq!( + manager.progress.current_height(), + 200, + "new() must seed progress.current_height from the engine's tip list height" + ); + } + + /// Counterpart to the recovery test: when the engine has no stored + /// masternode lists, `new()` must leave `last_synced_block_hash` as None + /// so the QRInfo path knows it must run from scratch instead of trying + /// to issue a targeted GetMnListDiff against a bogus base. + #[tokio::test] + async fn test_masternode_manager_starts_clean_with_empty_engine() { + let manager = create_test_manager_for(dashcore::Network::Testnet).await; + assert_eq!(manager.sync_state.last_synced_block_hash, None); + assert_eq!(manager.progress.current_height(), 0); + } } diff --git a/dash-spv/src/sync/masternodes/progress.rs b/dash-spv/src/sync/masternodes/progress.rs index aab9b8abf..a2e00c396 100644 --- a/dash-spv/src/sync/masternodes/progress.rs +++ b/dash-spv/src/sync/masternodes/progress.rs @@ -16,6 +16,14 @@ pub struct MasternodesProgress { block_header_tip_height: u32, /// Number of mnlistdiffs processed in the current sync session. diffs_processed: u32, + /// Number of QRInfo requests sent in the current sync session. + qr_infos_requested: u32, + /// Number of rotation cycles that completed full fresh validation + /// (every rotated quorum verified) this session. + validated_cycles: u32, + /// Number of distinct rotation cycles the tip has entered this session, + /// counted on every cycle boundary crossing including the initial one. + rotation_cycles: u32, /// The last time a mnlistdiff was stored/processed or the last manager state change. last_activity: Instant, } @@ -28,6 +36,9 @@ impl Default for MasternodesProgress { target_height: 0, block_header_tip_height: 0, diffs_processed: 0, + qr_infos_requested: 0, + validated_cycles: 0, + rotation_cycles: 0, last_activity: Instant::now(), } } @@ -57,6 +68,21 @@ impl MasternodesProgress { self.diffs_processed } + /// Number of QRInfo requests sent in the current sync session. + pub fn qr_infos_requested(&self) -> u32 { + self.qr_infos_requested + } + + /// Number of rotation cycles that completed full fresh validation this session. + pub fn validated_cycles(&self) -> u32 { + self.validated_cycles + } + + /// Number of distinct rotation cycles the tip has entered this session. + pub fn rotation_cycles(&self) -> u32 { + self.rotation_cycles + } + /// The last time a mnlistdiff was stored/processed or the last manager state change. pub fn last_activity(&self) -> Instant { self.last_activity @@ -94,6 +120,21 @@ impl MasternodesProgress { self.bump_last_activity(); } + pub fn add_qr_infos_requested(&mut self, count: u32) { + self.qr_infos_requested += count; + self.bump_last_activity(); + } + + pub fn add_validated_cycles(&mut self, count: u32) { + self.validated_cycles += count; + self.bump_last_activity(); + } + + pub fn add_rotation_cycles(&mut self, count: u32) { + self.rotation_cycles += count; + self.bump_last_activity(); + } + pub fn bump_last_activity(&mut self) { self.last_activity = Instant::now(); } @@ -103,11 +144,14 @@ impl fmt::Display for MasternodesProgress { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!( f, - "{:?} {}/{} | diffs_processed: {}, last_activity: {}s", + "{:?} {}/{} | diffs_processed: {}, qr_infos_requested: {}, validated_cycles: {}, rotation_cycles: {}, last_activity: {}s", self.state, self.current_height, self.target_height, self.diffs_processed, + self.qr_infos_requested, + self.validated_cycles, + self.rotation_cycles, self.last_activity.elapsed().as_secs() ) } diff --git a/dash-spv/src/sync/masternodes/sync_manager.rs b/dash-spv/src/sync/masternodes/sync_manager.rs index ffc15e7b3..79c8116c7 100644 --- a/dash-spv/src/sync/masternodes/sync_manager.rs +++ b/dash-spv/src/sync/masternodes/sync_manager.rs @@ -1,3 +1,4 @@ +use super::manager::PipelineMode; use crate::error::SyncResult; use crate::network::{Message, MessageType, RequestSender}; use crate::storage::BlockHeaderStorage; @@ -9,21 +10,34 @@ use async_trait::async_trait; use dashcore::network::message::NetworkMessage; use dashcore::network::message_qrinfo::QRInfo; use dashcore::sml::masternode_list_engine::{MasternodeListEngine, WORK_DIFF_DEPTH}; -use dashcore::sml::quorum_validation_error::QuorumValidationError; use dashcore::{BlockHash, QuorumHash}; use dashcore_hashes::Hash; use std::collections::{BTreeSet, HashSet}; -use std::time::{Duration, Instant}; +use std::time::Duration; -/// Timeout duration for waiting for QRInfo response. -const QRINFO_TIMEOUT_SECS: u64 = 15; +/// Per-attempt timeout schedule for QRInfo, indexed by the in-flight attempt's +/// retry count (0 = initial send, N = N-th retry). Round-robin peer selection in +/// the network manager rotates peers naturally on every send, so a short first +/// timeout fails over fast when one peer drops the request silently while the +/// later, longer timeouts give a slow but responsive network room to answer. +/// +/// Total worst-case wall clock if every attempt times out: +/// `sum(QRINFO_TIMEOUT_SCHEDULE_SECS) = 100s`. +const QRINFO_TIMEOUT_SCHEDULE_SECS: [u64; 3] = [10, 30, 60]; -/// Maximum number of retry attempts before giving up. -const MAX_RETRY_ATTEMPTS: u8 = 3; +/// Maximum number of in-flight attempts (initial send plus retries) before +/// giving up. Equal to `QRINFO_TIMEOUT_SCHEDULE_SECS.len()`. +const MAX_RETRY_ATTEMPTS: u8 = QRINFO_TIMEOUT_SCHEDULE_SECS.len() as u8; -/// Delay between retries when ChainLock is not yet available for the tip. -/// ChainLocks typically propagate within a few seconds after a block is mined. -const CHAINLOCK_RETRY_DELAY_SECS: u64 = 5; +/// Returns the timeout for the in-flight QRInfo attempt with the given retry count. +/// +/// `retry_count == 0` is the initial send; values past the last entry clamp to +/// the slowest schedule slot. `MAX_RETRY_ATTEMPTS` already prevents going past +/// the array, but the clamp is defensive in case the constants drift. +fn qrinfo_timeout_for(retry_count: u8) -> Duration { + let idx = (retry_count as usize).min(QRINFO_TIMEOUT_SCHEDULE_SECS.len() - 1); + Duration::from_secs(QRINFO_TIMEOUT_SCHEDULE_SECS[idx]) +} /// Build MnListDiff request pairs (base_hash, target_hash) for quorum validation. /// @@ -214,7 +228,7 @@ impl SyncManager for MasternodesManager { fn clear_in_flight_state(&mut self) { self.sync_state.clear_pending(); self.sync_state.qrinfo_retry_count = 0; - self.sync_state.chainlock_retry_after = None; + self.sync_state.last_processed_qrinfo_tip = None; } async fn handle_message( @@ -224,6 +238,9 @@ impl SyncManager for MasternodesManager { ) -> SyncResult> { match msg.inner() { NetworkMessage::QRInfo(qr_info) => { + if !self.sync_state.should_process_qrinfo(qr_info) { + return Ok(vec![]); + } tracing::info!("Processing QRInfo message"); self.sync_state.qrinfo_received(); @@ -235,43 +252,18 @@ impl SyncManager for MasternodesManager { tracing::info!("Fed {} block heights to engine", fed); // Feed QRInfo to engine first to populate masternode lists - if let Err(e) = engine.feed_qr_info(qr_info.clone(), true, true) { - // Check if this is a tip ChainLock error (h - 0 means the tip block) - // The QRInfo response always includes `mn_list_diff_tip` which is the current - // chain tip. If the tip was just mined, the ChainLock hasn't propagated yet. - let is_tip_chainlock_error = matches!( - e, - QuorumValidationError::RequiredRotatedChainLockSigNotPresent(0, _) - ); - - if is_tip_chainlock_error { - self.sync_state.qrinfo_retry_count += 1; - - if self.sync_state.qrinfo_retry_count <= MAX_RETRY_ATTEMPTS { - tracing::info!( - "ChainLock not yet available for tip, scheduling retry {}/{} in {}s", - self.sync_state.qrinfo_retry_count, - MAX_RETRY_ATTEMPTS, - CHAINLOCK_RETRY_DELAY_SECS - ); - // Schedule a delayed retry - the tick handler will trigger it - self.sync_state.chainlock_retry_after = Some( - Instant::now() + Duration::from_secs(CHAINLOCK_RETRY_DELAY_SECS), - ); - drop(engine); - self.set_state(SyncState::Syncing); - return Ok(vec![]); - } + let qr_info_result = match engine.feed_qr_info(qr_info.clone(), true, true) { + Ok(qr_info_result) => qr_info_result, + Err(e) => { + tracing::error!("QRInfo feed into engine failed: {}", e); + return Err(SyncError::MasternodeSyncFailed(e.to_string())); } + }; - // For other errors or max retries reached, fail - tracing::error!( - "QRInfo failed after {} retries: {}", - self.sync_state.qrinfo_retry_count, - e - ); - return Err(SyncError::MasternodeSyncFailed(e.to_string())); - } + // Record the successfully processed tip so a late straggler carrying + // the same tip hash is dropped by `should_process_qrinfo`. + self.sync_state.last_processed_qrinfo_tip = + Some(qr_info.mn_list_diff_tip.block_hash); // Populate known_mn_list_heights from engine after QRInfo processing self.sync_state.known_mn_list_heights = @@ -296,7 +288,32 @@ impl SyncManager for MasternodesManager { drop(engine); drop(storage); - // Queue and send MnListDiff requests via pipeline + if let Some(ref qr_info_result) = qr_info_result { + tracing::info!( + "QRInfo processed: stored_cycle_height={:?}, rotated_quorum_count={}, fully_verified_count={}, newly_qualified_count={}", + qr_info_result.stored_cycle_height, + qr_info_result.rotated_quorum_count, + qr_info_result.fully_verified_count, + qr_info_result.newly_qualified_count, + ); + // If every rotated quorum in this QRInfo ended up Verified, + // mark the cycle validated so `next_pipeline_mode` will + // return `Incremental` for every subsequent header in this + // cycle. No more QRInfo requests for this cycle until the + // next boundary. + if qr_info_result.all_fully_verified() { + if let Some(ref stored_cycle_height) = qr_info_result.stored_cycle_height { + self.mark_cycle_validated(*stored_cycle_height); + } + } + } + + // The historical diffs that follow a QRInfo run under QuorumValidation. + // Carry the result on the mode so `complete_pipeline` can pass it + // into the resulting `MasternodeStateUpdated` event. + self.sync_state.pipeline_mode = PipelineMode::QuorumValidation { + qr_info_result, + }; self.sync_state.mnlistdiff_pipeline.queue_requests(request_pairs); self.sync_state.mnlistdiff_pipeline.send_pending(requests)?; @@ -304,7 +321,7 @@ impl SyncManager for MasternodesManager { // If no pending requests, complete if !self.sync_state.has_pending_requests() { - return self.verify_and_complete().await; + return self.complete_pipeline().await; } } @@ -348,19 +365,22 @@ impl SyncManager for MasternodesManager { let mut engine = self.engine.write().await; engine.feed_block_height(target_height, diff.block_hash); - match engine.apply_diff(diff.clone(), Some(target_height), false, None) { - Ok(_) => { - self.sync_state.known_mn_list_heights.insert(target_height); - tracing::debug!("Applied MnListDiff at height {}", target_height); - } - Err(e) => { - tracing::warn!( - "Failed to apply MnListDiff at height {}: {}", - target_height, - e - ); - } - } + let apply_ok = + match engine.apply_diff(diff.clone(), Some(target_height), false, None) { + Ok(_) => { + self.sync_state.known_mn_list_heights.insert(target_height); + tracing::debug!("Applied MnListDiff at height {}", target_height); + true + } + Err(e) => { + tracing::warn!( + "Failed to apply MnListDiff at height {}: {}", + target_height, + e + ); + false + } + }; drop(engine); self.progress.add_diffs_processed(1); @@ -369,8 +389,17 @@ impl SyncManager for MasternodesManager { // Check if all responses received if self.sync_state.mnlistdiff_pipeline.is_complete() { + // In `Incremental` mode, a failed `apply_diff` means the engine + // state is unchanged. Skip completion to avoid emitting a + // spurious `MasternodeStateUpdated` for stale state. The next + // `BlockHeadersStored` event will re-drive an incremental update. + if !apply_ok + && matches!(self.sync_state.pipeline_mode, PipelineMode::Incremental) + { + return Ok(vec![]); + } tracing::info!("All MnListDiff responses received"); - return self.verify_and_complete().await; + return self.complete_pipeline().await; } } @@ -396,18 +425,54 @@ impl SyncManager for MasternodesManager { self.progress.update_target_height(*tip_height); } - // If Synced but behind, trigger incremental update to catch up with new blocks + // If Synced but behind, pick the pipeline mode for this tip update. The + // mode selector fires a full QRInfo only inside the current cycle's DKG + // mining window (and only while the cycle has not been freshly validated), + // and uses a targeted `GetMnListDiff` for tip updates in every other case - + // keeping the masternode list fresh on every new block without re-running + // rotated quorum validation. if self.state() == SyncState::Synced && self.progress.current_height() < self.progress.block_header_tip_height() { - tracing::debug!( - "New headers stored (tip: {}), updating masternode list from {}", - tip_height, - self.progress.current_height() - ); - self.sync_state.qrinfo_retry_count = 0; - self.sync_state.clear_pending(); - return self.send_qrinfo_for_tip(requests).await; + // A previous pipeline (QRInfo + draining historical MnListDiffs, or an + // earlier incremental update) may still be in flight. Starting a new + // pipeline here would overwrite `pipeline_mode` and append into the same + // queue, so when the shared pipeline completes the wrong completion path + // runs and any pending `qr_info_result` is discarded. The `tick` handler + // re-fires once the pipeline drains. + if self.sync_state.has_pending_requests() { + return Ok(vec![]); + } + + match self.next_pipeline_mode(*tip_height) { + PipelineMode::QuorumValidation { + .. + } => { + if self.sync_state.qrinfo_in_flight.is_some() { + tracing::debug!( + "New headers stored (tip: {}), QRInfo already in flight", + tip_height, + ); + return Ok(vec![]); + } + tracing::debug!( + "New headers stored (tip: {}), firing QRInfo from {}", + tip_height, + self.progress.current_height() + ); + self.sync_state.qrinfo_retry_count = 0; + self.sync_state.clear_pending(); + return self.send_qrinfo_for_tip(requests).await; + } + PipelineMode::Incremental => { + tracing::debug!( + "New headers stored (tip: {}), firing targeted MnListDiff from {}", + tip_height, + self.progress.current_height() + ); + return self.send_tip_mnlistdiff_update(requests).await; + } + } } } @@ -439,18 +504,57 @@ impl SyncManager for MasternodesManager { }; if should_restart { - // Use debug for incremental updates (when already Synced) + // A `BlockHeaderSyncComplete` event fires whenever the header pipeline + // catches up to the latest known tip, including after brief lags during + // normal runtime, so this branch runs both for initial sync and for + // catch-ups from `Synced`. The two cases need different dispatch: + // + // - Initial sync (`WaitingForConnections` / `WaitForEvents` / stuck + // `Syncing`): fire a full QRInfo unconditionally to seed the + // masternode list engine from scratch. + // - Catch-up from `Synced`: route through `next_pipeline_mode` so that + // the gate picks QRInfo vs targeted `GetMnListDiff` based on where + // the tip sits relative to the current cycle's DKG mining window, + // matching the `BlockHeadersStored` per-block path. Bypassing the + // gate here would cause a full QRInfo on every batch catch-up, + // which fires several times per cycle even when the cycle is + // already freshly-validated and the tip should just be refreshed + // with a targeted mnlistdiff. if self.state() == SyncState::Synced { + // Same guard as the `BlockHeadersStored` Synced arm above: never + // start a new pipeline while an earlier one is still draining, + // otherwise the shared queue and `pipeline_mode` get clobbered. + if self.sync_state.has_pending_requests() { + return Ok(vec![]); + } tracing::debug!( "Headers sync complete at {}, updating masternode list", self.progress.block_header_tip_height() ); - } else { - tracing::info!( - "Headers sync complete at {}, starting masternode sync", - self.progress.block_header_tip_height() - ); + match self.next_pipeline_mode(*tip_height) { + PipelineMode::QuorumValidation { + .. + } => { + if self.sync_state.qrinfo_in_flight.is_some() { + tracing::debug!( + "Headers sync complete at {}, QRInfo already in flight", + self.progress.block_header_tip_height() + ); + return Ok(vec![]); + } + self.sync_state.qrinfo_retry_count = 0; + self.sync_state.clear_pending(); + return self.send_qrinfo_for_tip(requests).await; + } + PipelineMode::Incremental => { + return self.send_tip_mnlistdiff_update(requests).await; + } + } } + tracing::info!( + "Headers sync complete at {}, starting masternode sync", + self.progress.block_header_tip_height() + ); self.sync_state.qrinfo_retry_count = 0; self.sync_state.clear_pending(); return self.send_qrinfo_for_tip(requests).await; @@ -466,40 +570,52 @@ impl SyncManager for MasternodesManager { return Ok(vec![]); } - // If Synced with no pending requests, nothing to do + // If Synced with no pending requests, check whether new headers arrived + // while the initial sync was in progress. BlockHeadersStored events that + // landed during Syncing state updated block_header_tip_height but couldn't + // trigger an incremental update (the handler requires Synced). The tick + // catches this gap and fires the appropriate pipeline. if self.state() == SyncState::Synced && !self.sync_state.has_pending_requests() { - return Ok(vec![]); - } - - // Check for ChainLock retry (tip didn't have ChainLock yet) - if let Some(retry_after) = self.sync_state.chainlock_retry_after { - if Instant::now() >= retry_after { - tracing::info!("Retrying QRInfo after ChainLock delay"); - self.sync_state.chainlock_retry_after = None; - return self.send_qrinfo_for_tip(requests).await; + if self.progress.current_height() < self.progress.block_header_tip_height() { + let tip = self.progress.block_header_tip_height(); + match self.next_pipeline_mode(tip) { + PipelineMode::QuorumValidation { + .. + } => { + if self.sync_state.qrinfo_in_flight.is_none() { + self.sync_state.qrinfo_retry_count = 0; + self.sync_state.clear_pending(); + return self.send_qrinfo_for_tip(requests).await; + } + } + PipelineMode::Incremental => { + return self.send_tip_mnlistdiff_update(requests).await; + } + } } - // Still waiting for retry delay return Ok(vec![]); } // Check for QRInfo timeout - if self.sync_state.waiting_for_qrinfo { - if let Some(wait_start) = self.sync_state.qrinfo_wait_start { - let timeout = Duration::from_secs(QRINFO_TIMEOUT_SECS); - if wait_start.elapsed() > timeout { - if self.sync_state.qrinfo_retry_count < MAX_RETRY_ATTEMPTS { - tracing::warn!("Timeout waiting for QRInfo response, retrying..."); - self.sync_state.qrinfo_retry_count += 1; - self.sync_state.clear_pending(); - return self.send_qrinfo_for_tip(requests).await; - } else { - tracing::warn!( - "QRInfo timeout after {} retries, skipping masternode sync", - MAX_RETRY_ATTEMPTS - ); - self.sync_state.clear_pending(); - return self.verify_and_complete().await; - } + if let Some(in_flight) = self.sync_state.qrinfo_in_flight { + let timeout = qrinfo_timeout_for(self.sync_state.qrinfo_retry_count); + if in_flight.wait_start.elapsed() > timeout { + if self.sync_state.qrinfo_retry_count < MAX_RETRY_ATTEMPTS - 1 { + tracing::warn!( + timeout_secs = timeout.as_secs(), + retry_count = self.sync_state.qrinfo_retry_count, + "Timeout waiting for QRInfo response, retrying..." + ); + self.sync_state.qrinfo_retry_count += 1; + self.sync_state.clear_pending(); + return self.send_qrinfo_for_tip(requests).await; + } else { + tracing::warn!( + "QRInfo timeout after {} retries, skipping masternode sync", + MAX_RETRY_ATTEMPTS + ); + self.sync_state.clear_pending(); + return self.complete_pipeline().await; } } return Ok(vec![]); @@ -515,7 +631,7 @@ impl SyncManager for MasternodesManager { // Check if complete after handling timeouts if self.sync_state.mnlistdiff_pipeline.is_complete() { tracing::info!("MnListDiff pipeline complete"); - return self.verify_and_complete().await; + return self.complete_pipeline().await; } } @@ -529,7 +645,11 @@ impl SyncManager for MasternodesManager { #[cfg(test)] mod tests { - use super::feed_qrinfo_heights_to_engine; + use super::super::manager::{MasternodeSyncState, QRInfoInFlight}; + use super::{ + feed_qrinfo_heights_to_engine, qrinfo_timeout_for, MAX_RETRY_ATTEMPTS, + QRINFO_TIMEOUT_SCHEDULE_SECS, + }; use crate::error::StorageResult; use crate::storage::{BlockHeaderStorage, BlockHeaderTip}; use crate::types::HashedBlockHeader; @@ -546,6 +666,7 @@ mod tests { use dashcore_hashes::Hash; use std::collections::HashMap; use std::ops::Range; + use std::time::Instant; struct MockHeaderStorage(HashMap); @@ -708,4 +829,129 @@ mod tests { ); } } + + /// The QRInfo retry budget escalates: a tight first timeout fails over + /// fast when one peer drops the request silently, while later attempts + /// get progressively more headroom so a genuinely slow but responsive + /// network still gets to answer. + #[test] + fn test_qrinfo_timeout_schedule() { + assert_eq!(QRINFO_TIMEOUT_SCHEDULE_SECS.len(), MAX_RETRY_ATTEMPTS as usize); + + // First attempt fails over fast so a single bad peer does not block sync. + assert_eq!(qrinfo_timeout_for(0).as_secs(), 10); + + // Schedule escalates monotonically so a slow but responsive network + // still gets enough time to answer. + let schedule: Vec = + (0..MAX_RETRY_ATTEMPTS).map(|n| qrinfo_timeout_for(n).as_secs()).collect(); + assert!( + schedule.windows(2).all(|w| w[0] <= w[1]), + "timeout schedule must be non-decreasing, got {:?}", + schedule + ); + + // Out-of-range retry counts must clamp to the slowest slot rather than + // panic, in case the constants drift relative to MAX_RETRY_ATTEMPTS. + let last = *QRINFO_TIMEOUT_SCHEDULE_SECS.last().unwrap(); + assert_eq!(qrinfo_timeout_for(MAX_RETRY_ATTEMPTS).as_secs(), last); + assert_eq!(qrinfo_timeout_for(u8::MAX).as_secs(), last); + } + + /// Build a minimal `QRInfo` whose `mn_list_diff_tip.block_hash` is `[tip_byte; 32]`. + /// Only the tip hash is read by `should_process_qrinfo`; every other field is filler. + fn qrinfo_with_tip(tip_byte: u8) -> QRInfo { + QRInfo { + quorum_snapshot_at_h_minus_c: make_snapshot(), + quorum_snapshot_at_h_minus_2c: make_snapshot(), + quorum_snapshot_at_h_minus_3c: make_snapshot(), + mn_list_diff_tip: make_diff(0x00, tip_byte), + mn_list_diff_h: make_diff(0x00, 0x00), + mn_list_diff_at_h_minus_c: make_diff(0x00, 0x00), + mn_list_diff_at_h_minus_2c: make_diff(0x00, 0x00), + mn_list_diff_at_h_minus_3c: make_diff(0x00, 0x00), + quorum_snapshot_and_mn_list_diff_at_h_minus_4c: None, + mn_list_diff_list: vec![], + last_commitment_per_index: vec![], + quorum_snapshot_list: vec![], + } + } + + /// `should_process_qrinfo` is the dedup gate at the QRInfo handler entry. It + /// must: + /// 1. Drop a response carrying the same `mn_list_diff_tip.block_hash` as the + /// last successfully processed one (defends against a late straggler from + /// a previous request whose response already won, even when the in-flight + /// gate is open for a newer request). + /// 2. Drop an unsolicited response (no QRInfo currently in flight). + /// 3. Allow a fresh response that matches the active in-flight request tip. + /// 4. Drop a response whose tip does not match the active in-flight request + /// tip (late straggler from a previous tip whose request was rotated by a + /// timeout retry). + #[test] + fn test_should_process_qrinfo_dedup_gate() { + let tip_a = BlockHash::from_slice(&[0xAA; 32]).unwrap(); + let tip_b = BlockHash::from_slice(&[0xBB; 32]).unwrap(); + let in_flight_b = QRInfoInFlight { + tip: tip_b, + wait_start: Instant::now(), + }; + + // Same-tip duplicate is dropped even when a request is in flight. + let state = MasternodeSyncState { + qrinfo_in_flight: Some(in_flight_b), + last_processed_qrinfo_tip: Some(tip_a), + ..Default::default() + }; + assert!( + !state.should_process_qrinfo(&qrinfo_with_tip(0xAA)), + "duplicate of last processed tip must be dropped" + ); + + // Unsolicited response (no request in flight) is dropped, even for a + // fresh tip. + let state = MasternodeSyncState::default(); + assert!(state.qrinfo_in_flight.is_none()); + assert!( + !state.should_process_qrinfo(&qrinfo_with_tip(0xBB)), + "unsolicited response must be dropped" + ); + + // Response matching the active in-flight tip is accepted. + let state = MasternodeSyncState { + qrinfo_in_flight: Some(in_flight_b), + last_processed_qrinfo_tip: Some(tip_a), + ..Default::default() + }; + assert!( + state.should_process_qrinfo(&qrinfo_with_tip(0xBB)), + "response matching the active request tip must be accepted" + ); + + // Same-tip dedup wins over the in-flight check: even if the flag has + // already been cleared (e.g. a sibling response just flipped it), the + // straggler must not be processed twice. + let state = MasternodeSyncState { + qrinfo_in_flight: None, + last_processed_qrinfo_tip: Some(tip_a), + ..Default::default() + }; + assert!( + !state.should_process_qrinfo(&qrinfo_with_tip(0xAA)), + "duplicate must be dropped even when no request is in flight" + ); + + // Late straggler from a previous tip whose request was rotated by a + // timeout retry: the in-flight gate is open for tip B, but tip C + // arrives. Dropped because the tip does not match the active request. + let state = MasternodeSyncState { + qrinfo_in_flight: Some(in_flight_b), + last_processed_qrinfo_tip: Some(tip_a), + ..Default::default() + }; + assert!( + !state.should_process_qrinfo(&qrinfo_with_tip(0xCC)), + "response for non-active request tip must be dropped" + ); + } }