diff --git a/dash-spv/src/sml_engine.rs b/dash-spv/src/sml_engine.rs index 0434c0480..169f3d2c7 100644 --- a/dash-spv/src/sml_engine.rs +++ b/dash-spv/src/sml_engine.rs @@ -5,6 +5,7 @@ mod validation; use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; +use crate::error::{StorageError, StorageResult}; use crate::storage::BlockHeaderStorage; use dashcore::bls_sig_utils::BLSSignature; use dashcore::hashes::Hash; @@ -54,6 +55,41 @@ impl QRInfoFeedResult { } } +/// Why the engine did not take a diff or a QRInfo. +#[derive(Debug, thiserror::Error)] +pub enum FeedError { + /// The header chain could not be read, so the message was not judged. + /// The fault is ours: the message may well be taken once it reads again. + #[error("Could not read the header chain: {0}")] + HeaderStorage(StorageError), + /// The engine refused the message for the reason given. A reason can be + /// a gap of ours rather than a fault of the message: a block our header + /// chain does not hold, or a base list the engine lacks. + #[error(transparent)] + Refused(#[from] E), +} + +impl From for FeedError { + fn from(error: StorageError) -> Self { + FeedError::HeaderStorage(error) + } +} + +impl From for FeedError { + fn from(error: StorageError) -> Self { + FeedError::HeaderStorage(error) + } +} + +impl From> for FeedError { + fn from(error: FeedError) -> Self { + match error { + FeedError::HeaderStorage(e) => FeedError::HeaderStorage(e), + FeedError::Refused(e) => FeedError::Refused(e.into()), + } + } +} + pub struct MasternodeListEngine { headers: Arc>, pub masternode_lists: BTreeMap, @@ -158,6 +194,45 @@ fn cycle_quarter_work_heights( .map(|cycles_back: u32| cycle_base.checked_sub(cycles_back * interval + WORK_DIFF_DEPTH)) } +/// An engine's lists and rotated cycles. +type ListsAndCycles = ( + BTreeMap, + BTreeMap>, +); + +/// The lists and rotated cycles of an engine as a QRInfo feed found them, put +/// back when this drops unless the feed's changes are kept: a feed that fails +/// or is dropped part way leaves the engine as it was. The copy shares the +/// lists' maps; the few stored cycles are copied. +struct QRInfoRollback<'a, H: BlockHeaderStorage> { + engine: &'a mut MasternodeListEngine, + saved: Option, +} + +impl<'a, H: BlockHeaderStorage> QRInfoRollback<'a, H> { + fn new(engine: &'a mut MasternodeListEngine) -> Self { + let saved = (engine.masternode_lists.clone(), engine.rotated_quorums_per_cycle.clone()); + Self { + engine, + saved: Some(saved), + } + } + + /// Keeps what the feed changed. + fn keep(mut self) { + self.saved = None; + } +} + +impl Drop for QRInfoRollback<'_, H> { + fn drop(&mut self) { + if let Some((masternode_lists, rotated_quorums_per_cycle)) = self.saved.take() { + self.engine.masternode_lists = masternode_lists; + self.engine.rotated_quorums_per_cycle = rotated_quorums_per_cycle; + } + } +} + impl MasternodeListEngine { /// An empty engine for `network` that looks block heights up in `headers`. pub fn new(network: Network, headers: Arc>) -> Self { @@ -169,33 +244,50 @@ impl MasternodeListEngine { } } - /// The height of `block_hash` in the header chain. The genesis block, and - /// the all-zero hash standing for it, is at 0 even when the chain starts - /// at a checkpoint. - pub async fn height_of(&self, block_hash: &BlockHash) -> Option { + /// The height of `block_hash` in the header chain, `None` when the chain + /// does not hold it. The genesis block, and the all-zero hash standing for + /// it, is at 0 even when the chain starts at a checkpoint. Fails when the + /// header storage cannot be read, which says nothing about the block. + pub async fn height_of( + &self, + block_hash: &BlockHash, + ) -> StorageResult> { if *block_hash == BlockHash::all_zeros() || self.network.known_genesis_block_hash() == Some(*block_hash) { - return Some(0); - } - match self.headers.read().await.get_header_height_by_hash(block_hash).await { - Ok(height) => height, - Err(e) => { - tracing::warn!("Could not look up the height of {block_hash}: {e}"); - None - } + return Ok(Some(0)); } + self.headers.read().await.get_header_height_by_hash(block_hash).await } - /// The hash of the block at `height` in the header chain. - async fn hash_at(&self, height: CoreBlockHeight) -> Option { - match self.headers.read().await.get_header(height).await { - Ok(header) => header.map(|header| *header.hash()), - Err(e) => { - tracing::warn!("Could not look up the block at {height}: {e}"); - None - } - } + /// [`Self::height_of`] for callers that treat a block whose height could + /// not be read like one the header chain does not hold. The storage error + /// is logged. + pub(crate) async fn height_if_known(&self, block_hash: &BlockHash) -> Option { + self.height_of(block_hash).await.unwrap_or_else(|e| { + tracing::warn!("Could not look up the height of {block_hash}: {e}"); + None + }) + } + + /// [`Self::height_of`] for a block a diff names and cannot be applied + /// without: refuses the diff when the header chain does not hold the + /// block, and fails with [`FeedError::HeaderStorage`] when the chain + /// cannot be read. + async fn required_height_of( + &self, + block_hash: &BlockHash, + ) -> Result> { + self.height_of(block_hash) + .await + .inspect_err(|e| tracing::warn!("Could not look up the height of {block_hash}: {e}"))? + .ok_or(FeedError::Refused(SmlError::BlockHashLookupFailed(*block_hash))) + } + + /// The hash of the block at `height` in the header chain, `None` when the + /// chain holds none there. Fails when the header storage cannot be read. + async fn hash_at(&self, height: CoreBlockHeight) -> StorageResult> { + Ok(self.headers.read().await.get_header(height).await?.map(|header| *header.hash())) } pub fn network(&self) -> Network { @@ -234,10 +326,35 @@ impl MasternodeListEngine { /// The `(base, target)` diffs that bring the work-block list of every /// non-rotating quorum of the newest list still to validate, oldest first. - /// A retired quorum type is never validated, so it needs none. - pub async fn missing_work_block_list_requests(&self) -> Vec<(BlockHash, BlockHash)> { - let work_heights = self.work_heights_awaiting_validation(|_, _| true).await; - self.work_block_list_requests(work_heights.into_keys()).await + /// A retired quorum type is never validated, so it needs none. Fails when + /// the header storage cannot read a block it looks up. + pub async fn missing_work_block_list_requests( + &self, + ) -> StorageResult> { + match self.scan_missing_work_block_lists().await { + (_, Some(e)) => Err(e), + (requests, None) => Ok(requests), + } + } + + /// [`Self::missing_work_block_list_requests`] leaving out the quorums whose + /// blocks the header storage cannot read. The failures are logged. + pub(crate) async fn readable_missing_work_block_list_requests( + &self, + ) -> Vec<(BlockHash, BlockHash)> { + self.scan_missing_work_block_lists().await.0 + } + + /// [`Self::readable_missing_work_block_list_requests`] with the first + /// header storage failure met, if any. + async fn scan_missing_work_block_lists( + &self, + ) -> (Vec<(BlockHash, BlockHash)>, Option) { + let (work_heights, unread_heights) = + self.work_heights_awaiting_validation(|_, _| true).await; + let (requests, unread_blocks) = + self.work_block_list_requests(work_heights.into_keys()).await; + (requests, unread_heights.or(unread_blocks)) } /// [`Self::missing_work_block_list_requests`] for only the quorums `diff` @@ -252,14 +369,14 @@ impl MasternodeListEngine { ) -> Vec<(BlockHash, BlockHash)> { let added: BTreeSet<(LLMQType, QuorumHash)> = diff.new_quorums.iter().map(|quorum| (quorum.llmq_type, quorum.quorum_hash)).collect(); - let mut work_heights = self + let (mut work_heights, _) = self .work_heights_awaiting_validation(|llmq_type, quorum_hash| { added.contains(&(*llmq_type, *quorum_hash)) }) .await; let blocks = match ( - self.height_of(&diff.base_block_hash).await, - self.height_of(&diff.block_hash).await, + self.height_if_known(&diff.base_block_hash).await, + self.height_if_known(&diff.block_hash).await, ) { (Some(base), Some(target)) => target.saturating_sub(base), _ => 0, @@ -285,18 +402,21 @@ impl MasternodeListEngine { } work_heights.retain(|_, quorums| !quorums.is_empty()); - self.work_block_list_requests(work_heights.into_keys()).await + self.work_block_list_requests(work_heights.into_keys()).await.0 } /// The work heights of the newest list's non-rotating quorums still to - /// validate that `wanted` selects, with the quorums waiting on each. + /// validate that `wanted` selects, with the quorums waiting on each, and + /// the first header storage failure met. A quorum whose block cannot be + /// read is left out. async fn work_heights_awaiting_validation( &self, wanted: impl Fn(&LLMQType, &QuorumHash) -> bool, - ) -> BTreeMap> { + ) -> (BTreeMap>, Option) { let mut work_heights: BTreeMap<_, BTreeSet<_>> = BTreeMap::new(); + let mut unread = None; let Some(newest) = self.latest_masternode_list() else { - return work_heights; + return (work_heights, unread); }; for (llmq_type, quorums) in newest.quorums.iter() { if llmq_type.is_rotating_quorum_type() @@ -313,34 +433,48 @@ impl MasternodeListEngine { continue; } match self.height_of(quorum_hash).await { - Some(height) => { + Ok(Some(height)) => { work_heights .entry(height.saturating_sub(QUORUM_MEMBER_LIST_OFFSET)) .or_default() .insert((*llmq_type, *quorum_hash)); } - None => tracing::warn!("No height for quorum {quorum_hash}, skipping it"), + Ok(None) => tracing::warn!("No height for quorum {quorum_hash}, skipping it"), + Err(e) => { + tracing::warn!("Could not look up quorum {quorum_hash}, skipping it: {e}"); + unread.get_or_insert(e); + } } } } - work_heights + (work_heights, unread) } /// The `(base, target)` diffs that bring the lists at /// `work_heights` the engine lacks, each from the nearest lower list it - /// holds. + /// holds, and the first header storage failure met. A work height whose + /// block cannot be read is left out. async fn work_block_list_requests( &self, work_heights: impl IntoIterator, - ) -> Vec<(BlockHash, BlockHash)> { + ) -> (Vec<(BlockHash, BlockHash)>, Option) { let mut requests = Vec::new(); + let mut unread = None; for height in work_heights { if self.masternode_lists.contains_key(&height) { continue; } - let Some(target) = self.hash_at(height).await else { - tracing::warn!("No block at work height {height}, skipping it"); - continue; + let target = match self.hash_at(height).await { + Ok(Some(target)) => target, + Ok(None) => { + tracing::warn!("No block at work height {height}, skipping it"); + continue; + } + Err(e) => { + tracing::warn!("Could not read the block at work height {height}: {e}"); + unread.get_or_insert(e); + continue; + } }; let base = self .masternode_lists @@ -349,7 +483,7 @@ impl MasternodeListEngine { .map_or(BlockHash::all_zeros(), |(_, list)| list.block_hash); requests.push((base, target)); } - requests + (requests, unread) } /// The stored rotated quorum `quorum_entry` commits to, if any. @@ -371,37 +505,38 @@ impl MasternodeListEngine { /// Stores the previous cycle, the active set of the list at the h work /// block, so IS locks of that cycle verify after a fresh sync. Best effort: /// a quorum that does not verify is left out on its own, and bad data is - /// only rejected on the current-cycle path. + /// only rejected on the current-cycle path. Fails only when the header + /// storage cannot be read. async fn validate_and_store_previous_cycle_quorums( &mut self, work_height: CoreBlockHeight, sigs_by_work_height: &BTreeMap, inferred_work_heights: &BTreeSet, snapshots: &BTreeMap, - ) { + ) -> StorageResult<()> { let isd_type = self.network.isd_llmq_type(); let Some(quorums_of_type) = self.masternode_lists.get(&work_height).and_then(|list| list.quorums.get(&isd_type)) else { - return; + return Ok(()); }; let Some(cycle_hash) = - self.active_set_cycle_hash(quorums_of_type.values().map(|q| &**q).collect()).await + self.active_set_cycle_hash(quorums_of_type.values().map(|q| &**q).collect()).await? else { - return; + return Ok(()); }; if self.is_cycle_fully_verified(&cycle_hash) { - return; + return Ok(()); } let mut entries = Vec::with_capacity(quorums_of_type.len()); for quorum in quorums_of_type.values() { let mut entry = (**quorum).clone(); entry.verifying_chain_lock_signature = - self.quarter_sigs(sigs_by_work_height, &entry.quorum_entry).await; + self.quarter_sigs(sigs_by_work_height, &entry.quorum_entry).await?; entries.push(entry); } - self.settle_rotated_quorums(&mut entries, inferred_work_heights, snapshots).await; + self.settle_rotated_quorums(&mut entries, inferred_work_heights, snapshots).await?; entries.retain(|entry| { let verified = entry.verified == LLMQEntryVerificationStatus::Verified; if !verified { @@ -414,14 +549,19 @@ impl MasternodeListEngine { verified }); if entries.is_empty() { - return; + return Ok(()); } for entry in &entries { self.set_quorum_status(isd_type, entry.quorum_entry.quorum_hash, &entry.verified); } - if let Err(e) = self.store_cycle_if_fully_verified(cycle_hash, entries, isd_type).await { - tracing::warn!("Previous cycle {cycle_hash} could not be stored: {e}"); + match self.store_cycle_if_fully_verified(cycle_hash, entries, isd_type).await { + Err(FeedError::HeaderStorage(e)) => return Err(e), + Err(FeedError::Refused(e)) => { + tracing::warn!("Previous cycle {cycle_hash} could not be stored: {e}") + } + Ok(_) => {} } + Ok(()) } /// `true` when the stored cycle has a `Verified` quorum for every active @@ -442,17 +582,18 @@ impl MasternodeListEngine { cycle_key: BlockHash, qualified_last_commitment_per_index: Vec, rotation_quorum_type: LLMQType, - ) -> Result, QuorumValidationError> { + ) -> Result, FeedError> { let all_entries_verified = qualified_last_commitment_per_index .iter() .all(|q| q.verified == LLMQEntryVerificationStatus::Verified); if !all_entries_verified || self.is_cycle_fully_verified(&cycle_key) { return Ok(None); } + let height = self.height_of(&cycle_key).await?; let cycle_map = build_cycle_quorum_map(qualified_last_commitment_per_index, rotation_quorum_type)?; self.rotated_quorums_per_cycle.entry(cycle_key).or_default().extend(cycle_map); - Ok(self.height_of(&cycle_key).await) + Ok(height) } /// The ChainLock signature of each rotation work block, from the @@ -467,10 +608,10 @@ impl MasternodeListEngine { async fn rotation_cl_sigs_by_work_height( &self, diffs: Vec<&MnListDiff>, - ) -> (BTreeMap, BTreeSet) { + ) -> StorageResult<(BTreeMap, BTreeSet)> { let mut diffs_by_end_height = Vec::new(); for diff in diffs { - diffs_by_end_height.push((self.height_of(&diff.block_hash).await, diff)); + diffs_by_end_height.push((self.height_of(&diff.block_hash).await?, diff)); } diffs_by_end_height.sort_by_key(|(end_height, _)| *end_height); @@ -480,7 +621,7 @@ impl MasternodeListEngine { for quorum in signed_rotated_quorums(diff, sig_obj) { if let Some(work_height) = self .rotated_quorum_cycle_base(quorum) - .await + .await? .and_then(|cycle_base| cycle_base.checked_sub(WORK_DIFF_DEPTH)) { sigs_by_work_height.insert(work_height, sig_obj.signature); @@ -505,7 +646,7 @@ impl MasternodeListEngine { let mut group_has_unresolved_rotated = false; let mut hypothesis_mismatch = false; for quorum in signed_rotated_quorums(diff, sig_obj) { - if self.rotated_quorum_cycle_base(quorum).await.is_some() { + if self.rotated_quorum_cycle_base(quorum).await?.is_some() { group_resolved = true; continue; } @@ -525,7 +666,7 @@ impl MasternodeListEngine { }; if self .hash_at(presumed_height) - .await + .await? .is_some_and(|block_hash| block_hash != quorum.quorum_hash) { hypothesis_mismatch = true; @@ -548,7 +689,7 @@ impl MasternodeListEngine { inferred_work_heights.insert(newest_work_height); } } - (sigs_by_work_height, inferred_work_heights) + Ok((sigs_by_work_height, inferred_work_heights)) } /// For a diff ending at a work block, the work block of the newest cycle @@ -560,40 +701,54 @@ impl MasternodeListEngine { .flatten() } - async fn rotated_quorum_cycle_base(&self, quorum: &QuorumEntry) -> Option { - rotated_cycle_base_height(quorum, self.height_of(&quorum.quorum_hash).await?) + async fn rotated_quorum_cycle_base( + &self, + quorum: &QuorumEntry, + ) -> StorageResult> { + let height = self.height_of(&quorum.quorum_hash).await?; + Ok(height.and_then(|height| rotated_cycle_base_height(quorum, height))) } - async fn quarter_work_heights(&self, quorum: &QuorumEntry) -> Option<[CoreBlockHeight; 4]> { - let cycle_base = self.rotated_quorum_cycle_base(quorum).await?; + async fn quarter_work_heights( + &self, + quorum: &QuorumEntry, + ) -> StorageResult> { + let Some(cycle_base) = self.rotated_quorum_cycle_base(quorum).await? else { + return Ok(None); + }; let interval = quorum.llmq_type.params().dkg_params.interval; - let [h3, h2, h1, h0] = cycle_quarter_work_heights(cycle_base, interval); - Some([h3?, h2?, h1?, h0?]) + let [Some(h3), Some(h2), Some(h1), Some(h0)] = + cycle_quarter_work_heights(cycle_base, interval) + else { + return Ok(None); + }; + Ok(Some([h3, h2, h1, h0])) } async fn quarter_sigs( &self, sigs_by_work_height: &BTreeMap, quorum: &QuorumEntry, - ) -> Option { - let [h3, h2, h1, h0] = self.quarter_work_heights(quorum).await?; - let sig_at = |work_height: CoreBlockHeight| sigs_by_work_height.get(&work_height).copied(); - Some(VerifyingChainLockSignaturesType::Rotating([ - sig_at(h3)?, - sig_at(h2)?, - sig_at(h1)?, - sig_at(h0)?, - ])) + ) -> StorageResult> { + let Some(work_heights) = self.quarter_work_heights(quorum).await? else { + return Ok(None); + }; + let [Some(h3), Some(h2), Some(h1), Some(h0)] = + work_heights.map(|work_height| sigs_by_work_height.get(&work_height).copied()) + else { + return Ok(None); + }; + Ok(Some(VerifyingChainLockSignaturesType::Rotating([h3, h2, h1, h0]))) } async fn quarter_sigs_are_inferred( &self, inferred_work_heights: &BTreeSet, quorum: &QuorumEntry, - ) -> bool { - self.quarter_work_heights(quorum).await.is_some_and(|work_heights| { + ) -> StorageResult { + Ok(self.quarter_work_heights(quorum).await?.is_some_and(|work_heights| { work_heights.iter().any(|work_height| inferred_work_heights.contains(work_height)) - }) + })) } /// The configured rotation type, or on a devnet without an override, the @@ -616,31 +771,37 @@ impl MasternodeListEngine { async fn active_set_cycle_hash( &self, entries: Vec<&QualifiedQuorumEntry>, - ) -> Option { + ) -> StorageResult> { let mut newest_cycle_base = None; for entry in entries { - let cycle_base = self.rotated_quorum_cycle_base(&entry.quorum_entry).await; + let cycle_base = self.rotated_quorum_cycle_base(&entry.quorum_entry).await?; newest_cycle_base = newest_cycle_base.max(cycle_base); } - self.hash_at(newest_cycle_base?).await + match newest_cycle_base { + Some(cycle_base) => self.hash_at(cycle_base).await, + None => Ok(None), + } } /// Sets each rotated quorum's status. A quarter signature keyed by /// elimination may be another cycle's, so a quorum failing under one is - /// skipped rather than invalid. + /// skipped rather than invalid. Fails when the header storage cannot be + /// read, which would otherwise leave such a quorum invalid. async fn settle_rotated_quorums( &self, quorums: &mut [QualifiedQuorumEntry], inferred_work_heights: &BTreeSet, snapshots: &BTreeMap, - ) { + ) -> StorageResult<()> { let statuses = self.rotated_quorum_statuses(&quorums.iter().collect::>(), snapshots).await; for quorum in quorums.iter_mut() { let quorum_hash = quorum.quorum_entry.quorum_hash; quorum.verified = statuses.get(&quorum_hash).cloned().unwrap_or_default(); if matches!(quorum.verified, LLMQEntryVerificationStatus::Invalid(_)) - && self.quarter_sigs_are_inferred(inferred_work_heights, &quorum.quorum_entry).await + && self + .quarter_sigs_are_inferred(inferred_work_heights, &quorum.quorum_entry) + .await? { tracing::warn!( "Rotated quorum {quorum_hash} failed validation ({}) under inferred quarter signatures, skipping it", @@ -651,16 +812,34 @@ impl MasternodeListEngine { ); } } + Ok(()) } /// Applies a QRInfo's diffs, then validates and stores its active rotation /// set and the previous cycle's. An invalid current-cycle quorum fails it. + /// + /// A QRInfo that fails, or whose feed is dropped part way as an aborted + /// task drops it, leaves the engine as it was: its retry starts from the + /// same lists, and no later message builds on lists of a QRInfo that was + /// not taken, and so never stored for the next start. pub async fn feed_qr_info( &mut self, qr_info: QRInfo, - ) -> Result { + ) -> Result> { + let rollback = QRInfoRollback::new(self); + let result = rollback.engine.apply_qr_info(qr_info).await; + if result.is_ok() { + rollback.keep(); + } + result + } + + async fn apply_qr_info( + &mut self, + qr_info: QRInfo, + ) -> Result> { let (rotation_sigs_by_work_height, inferred_work_heights) = - self.rotation_cl_sigs_by_work_height(qr_info_diffs(&qr_info)).await; + self.rotation_cl_sigs_by_work_height(qr_info_diffs(&qr_info)).await?; let QRInfo { quorum_snapshot_at_h_minus_c, @@ -677,6 +856,14 @@ impl MasternodeListEngine { mn_list_diff_list, } = qr_info; + // Core sends a snapshot with every diff of the list. Pairing them with + // `zip` would drop a diff without one unnoticed. + if let Some(unpaired) = mn_list_diff_list.get(quorum_snapshot_list.len()) { + return Err( + QuorumValidationError::RequiredSnapshotNotPresent(unpaired.block_hash).into() + ); + } + let can_verify_previous = quorum_snapshot_and_mn_list_diff_at_h_minus_4c.is_some(); let h_block_hash = mn_list_diff_h.block_hash; let rotation_quorum_type = self.rotation_quorum_type(&last_commitment_per_index); @@ -700,15 +887,16 @@ impl MasternodeListEngine { self.apply_and_prune(mn_list_diff_tip).await?; // Only the h-4c share brings the oldest quarters of the list at h. - if let Some(h_height) = self.height_of(&h_block_hash).await.filter(|_| can_verify_previous) - { - self.validate_and_store_previous_cycle_quorums( - h_height, - &rotation_sigs_by_work_height, - &inferred_work_heights, - &snapshots, - ) - .await; + if can_verify_previous { + if let Some(h_height) = self.height_of(&h_block_hash).await? { + self.validate_and_store_previous_cycle_quorums( + h_height, + &rotation_sigs_by_work_height, + &inferred_work_heights, + &snapshots, + ) + .await?; + } } let mut qualified_last_commitment_per_index = @@ -719,14 +907,15 @@ impl MasternodeListEngine { continue; } let quarter_sigs = - self.quarter_sigs(&rotation_sigs_by_work_height, &quorum_entry).await; + self.quarter_sigs(&rotation_sigs_by_work_height, &quorum_entry).await?; let mut qualified_quorum_entry: QualifiedQuorumEntry = quorum_entry.into(); qualified_quorum_entry.verifying_chain_lock_signature = quarter_sigs; qualified_last_commitment_per_index.push(qualified_quorum_entry); } - let cycle_key = - self.active_set_cycle_hash(qualified_last_commitment_per_index.iter().collect()).await; + let cycle_key = self + .active_set_cycle_hash(qualified_last_commitment_per_index.iter().collect()) + .await?; if cycle_key.is_none() && rotated_quorum_count > 0 { tracing::warn!( "None of the {} rotated quorums served resolves a cycle base, so the cycle cannot be stored under any key", @@ -739,13 +928,13 @@ impl MasternodeListEngine { &inferred_work_heights, &snapshots, ) - .await; + .await?; if let Some(LLMQEntryVerificationStatus::Invalid(e)) = qualified_last_commitment_per_index .iter() .map(|quorum| &quorum.verified) .find(|status| matches!(status, LLMQEntryVerificationStatus::Invalid(_))) { - return Err(e.clone()); + return Err(e.clone().into()); } for quorum in &qualified_last_commitment_per_index { @@ -782,7 +971,10 @@ impl MasternodeListEngine { /// Applies a diff, then verifies the newest list's non-rotating quorums: the /// diff may have brought one, or the work-block list one needs. - pub async fn apply_diff(&mut self, masternode_list_diff: MnListDiff) -> Result<(), SmlError> { + pub async fn apply_diff( + &mut self, + masternode_list_diff: MnListDiff, + ) -> Result<(), FeedError> { self.apply_and_prune(masternode_list_diff).await?; self.verify_newest_quorums().await; Ok(()) @@ -790,7 +982,10 @@ impl MasternodeListEngine { /// A diff on the newest list moves the tip and prunes, see /// [`Self::prune_after_extending`]. Others may share an older base. - async fn apply_and_prune(&mut self, masternode_list_diff: MnListDiff) -> Result<(), SmlError> { + async fn apply_and_prune( + &mut self, + masternode_list_diff: MnListDiff, + ) -> Result<(), FeedError> { let extended_tip = self .latest_masternode_list() .filter(|list| list.block_hash == masternode_list_diff.base_block_hash) @@ -805,15 +1000,11 @@ impl MasternodeListEngine { async fn apply_diff_to_base( &mut self, masternode_list_diff: MnListDiff, - ) -> Result<(), SmlError> { + ) -> Result<(), FeedError> { let block_hash = masternode_list_diff.block_hash; let base_block_hash = masternode_list_diff.base_block_hash; - let base_height = self - .height_of(&base_block_hash) - .await - .ok_or(SmlError::BlockHashLookupFailed(base_block_hash))?; - let diff_end_height = - self.height_of(&block_hash).await.ok_or(SmlError::BlockHashLookupFailed(block_hash))?; + let base_height = self.required_height_of(&base_block_hash).await?; + let diff_end_height = self.required_height_of(&block_hash).await?; if base_height == 0 { let masternode_list = MasternodeList::from_diff(masternode_list_diff, diff_end_height, self.network)?; @@ -822,7 +1013,7 @@ impl MasternodeListEngine { } let Some(base_masternode_list) = self.masternode_lists.get(&base_height) else { - return Err(SmlError::MissingStartMasternodeList(base_block_hash)); + return Err(SmlError::MissingStartMasternodeList(base_block_hash).into()); }; let masternode_list = base_masternode_list.apply_diff(masternode_list_diff, diff_end_height, self.network)?; @@ -1023,7 +1214,13 @@ pub(crate) mod test_support { #[cfg(test)] mod tests { use super::test_support::{fixture_heights, quorum_entry, TestEngine}; - use super::{build_cycle_quorum_map, WORK_DIFF_DEPTH}; + use super::{ + build_cycle_quorum_map, FeedError, MasternodeListEngine, QRInfoFeedResult, WORK_DIFF_DEPTH, + }; + use crate::error::{StorageError, StorageResult}; + use crate::storage::{BlockHeaderStorage, BlockHeaderTip}; + use crate::test_utils::MockHeaderStorage; + use crate::types::HashedBlockHeader; use bincode::{config, decode_from_slice, Decode}; use dashcore::bls_sig_utils::{BLSPublicKey, BLSSignature}; use dashcore::consensus::deserialize; @@ -1031,6 +1228,7 @@ mod tests { use dashcore::network::message_qrinfo::QRInfo; use dashcore::network::message_sml::{MnListDiff, QuorumCLSigObject}; use dashcore::prelude::CoreBlockHeight; + use dashcore::sml::error::SmlError; use dashcore::sml::llmq_entry_verification::{ LLMQEntryVerificationSkipStatus, LLMQEntryVerificationStatus, }; @@ -1045,7 +1243,10 @@ mod tests { use dashcore::transaction::special_transaction::quorum_commitment::QuorumEntry; use dashcore::{BlockHash, Network, QuorumHash}; use std::collections::{BTreeMap, BTreeSet}; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; + use std::time::Duration; + use tokio::sync::RwLock; fn make_qualified_quorum_entry( llmq_type: LLMQType, @@ -1116,7 +1317,8 @@ mod tests { // A known quorum height keys the signature exactly, no inference. let engine = engine_knowing_blocks(&[(diff_end, end_hash), (cycle_base + 3, quorum_hash)]); - let (sigs, inferred) = engine.rotation_cl_sigs_by_work_height(vec![&diff]).await; + let (sigs, inferred) = + engine.rotation_cl_sigs_by_work_height(vec![&diff]).await.expect("the headers read"); assert_eq!(sigs, BTreeMap::from([(work_height, sig)])); assert!(inferred.is_empty(), "a known quorum height must key exactly"); @@ -1131,7 +1333,10 @@ mod tests { vec![quorum_entry(isd_type, quorum_hash, Some(shifted_index))], vec![(sig, vec![0])], ); - let (sigs, inferred) = engine.rotation_cl_sigs_by_work_height(vec![&shifted]).await; + let (sigs, inferred) = engine + .rotation_cl_sigs_by_work_height(vec![&shifted]) + .await + .expect("the headers read"); assert_eq!( sigs, BTreeMap::from([(work_height, sig)]), @@ -1148,7 +1353,8 @@ mod tests { // Without that height the only remaining group is keyed by // elimination, and the caller is told the key was inferred. let engine = engine_knowing_blocks(&[(diff_end, end_hash)]); - let (sigs, inferred) = engine.rotation_cl_sigs_by_work_height(vec![&diff]).await; + let (sigs, inferred) = + engine.rotation_cl_sigs_by_work_height(vec![&diff]).await.expect("the headers read"); assert_eq!(sigs, BTreeMap::from([(work_height, sig)])); assert_eq!(inferred, BTreeSet::from([work_height])); @@ -1161,7 +1367,10 @@ mod tests { ], vec![(sig, vec![0]), (other_sig, vec![1])], ); - let (sigs, inferred) = engine.rotation_cl_sigs_by_work_height(vec![&ambiguous]).await; + let (sigs, inferred) = engine + .rotation_cl_sigs_by_work_height(vec![&ambiguous]) + .await + .expect("the headers read"); assert!( sigs.is_empty() && inferred.is_empty(), "two unresolved groups must resolve to nothing" @@ -1183,8 +1392,10 @@ mod tests { vec![quorum_entry(isd_type, quorum_hash, Some(3))], vec![(zero_sig, vec![0])], ); - let (sigs, inferred) = - engine.rotation_cl_sigs_by_work_height(vec![&older_zeroed, &newer_zeroed]).await; + let (sigs, inferred) = engine + .rotation_cl_sigs_by_work_height(vec![&older_zeroed, &newer_zeroed]) + .await + .expect("the headers read"); assert_eq!( sigs, BTreeMap::from([(older_work_height, zero_sig), (work_height, zero_sig)]), @@ -1207,7 +1418,10 @@ mod tests { ], vec![(sig, vec![0]), (other_sig, vec![1])], ); - let (sigs, _) = engine.rotation_cl_sigs_by_work_height(vec![&older, &newer]).await; + let (sigs, _) = engine + .rotation_cl_sigs_by_work_height(vec![&older, &newer]) + .await + .expect("the headers read"); assert_eq!( sigs, BTreeMap::from([(older_work_height, sig), (work_height, other_sig)]), @@ -1317,6 +1531,7 @@ mod tests { engine .active_set_cycle_hash(entries.iter().collect()) .await + .expect("the headers read") .expect("fixture must resolve the cycle key of its active set") } @@ -1376,6 +1591,62 @@ mod tests { (engine, qr_info) } + /// A QRInfo the engine refuses leaves it as it was, so a retry against + /// another peer starts from the same lists, and no later message builds on + /// lists of a QRInfo that was never stored. + #[tokio::test] + async fn a_refused_qr_info_leaves_the_engine_as_it_was() { + let (mut engine, mut qr_info) = load_qrinfo_2518986_fixture(); + assert!( + !qr_info.mn_list_diff_h.new_quorums.is_empty(), + "diff h must bring quorums for their signatures to be required" + ); + qr_info.mn_list_diff_h.quorums_chainlock_signatures.clear(); + let lists = engine.masternode_lists.clone(); + let cycles = engine.rotated_quorums_per_cycle.clone(); + + let result = engine.feed_qr_info(qr_info).await; + + assert!( + matches!( + result, + Err(FeedError::Refused(QuorumValidationError::SMLError( + SmlError::IncompleteSignatureSet + ))) + ), + "diff h must be refused, got {result:?}" + ); + assert!( + engine.masternode_lists == lists, + "a refused QRInfo must leave the lists as they were" + ); + assert!(engine.rotated_quorums_per_cycle == cycles, "and the stored cycles"); + } + + /// Core sends a snapshot for every diff of `mn_list_diff_list`. A diff + /// without one fails the QRInfo instead of being skipped unnoticed. + #[tokio::test] + async fn a_qr_info_with_a_diff_but_no_snapshot_for_it_is_refused() { + let (mut engine, mut qr_info) = load_qrinfo_2518986_fixture(); + assert_eq!(qr_info.quorum_snapshot_list.len(), qr_info.mn_list_diff_list.len()); + let unpaired = MnListDiff { + block_hash: BlockHash::dummy(1), + ..qr_info.mn_list_diff_at_h_minus_3c.clone() + }; + qr_info.mn_list_diff_list.push(unpaired); + + let result = engine.feed_qr_info(qr_info).await; + + assert!( + matches!( + result, + Err(FeedError::Refused(QuorumValidationError::RequiredSnapshotNotPresent(block_hash))) + if block_hash == BlockHash::dummy(1) + ), + "got {result:?}" + ); + } + #[tokio::test] async fn validate_first_qr_info_on_fresh_engine_with_mixed_cycle_quorums() { let (mut engine, qr_info) = load_qrinfo_2518986_fixture(); @@ -1387,7 +1658,8 @@ mod tests { let current_cycle_base = 2518848; let mut mixed_cycle_bases = BTreeSet::new(); for quorum in &qr_info.last_commitment_per_index { - mixed_cycle_bases.extend(engine.rotated_quorum_cycle_base(quorum).await); + mixed_cycle_bases + .extend(engine.rotated_quorum_cycle_base(quorum).await.expect("the headers read")); } assert!( mixed_cycle_bases.len() > 1 && mixed_cycle_bases.contains(¤t_cycle_base), @@ -1446,8 +1718,11 @@ mod tests { // The list at h holds the same previous-cycle quorums and must agree // with the stored cycle instead of reading them as skipped. - let h_height = - engine.height_of(&h_block_hash).await.expect("expected the height of the h block"); + let h_height = engine + .height_of(&h_block_hash) + .await + .expect("the headers read") + .expect("expected the height of the h block"); let h_quorums = &engine.masternode_lists[&h_height].quorums[&engine.network.isd_llmq_type()]; for quorum in previous_cycle.values() { @@ -1627,7 +1902,12 @@ mod tests { let err = engine.feed_qr_info(qr_info).await.expect_err("corrupt signature must reject"); assert!( - matches!(err, QuorumValidationError::AllCommitmentAggregatedSignatureNotValid(_)), + matches!( + err, + FeedError::Refused( + QuorumValidationError::AllCommitmentAggregatedSignatureNotValid(_) + ) + ), "expected aggregate-signature rejection, got {:?}", err ); @@ -1637,6 +1917,37 @@ mod tests { ); } + /// A QRInfo refused after its diffs applied, here for a current-cycle + /// quorum whose signature does not verify, leaves the engine as it was + /// too: neither the lists its diffs built nor the previous cycle it stored + /// on the way stay behind. + #[tokio::test] + async fn a_qr_info_refused_after_its_diffs_applied_leaves_the_engine_as_it_was() { + let (mut engine, mut qr_info) = load_qrinfo_2240504_fixture().await; + let target = qr_info.last_commitment_per_index[0].quorum_hash; + qr_info.last_commitment_per_index[0].all_commitment_aggregated_signature = + foreign_aggregate_signature(&qr_info.last_commitment_per_index, target); + let lists = engine.masternode_lists.clone(); + let cycles = engine.rotated_quorums_per_cycle.clone(); + + let result = engine.feed_qr_info(qr_info).await; + + assert!( + matches!( + result, + Err(FeedError::Refused( + QuorumValidationError::AllCommitmentAggregatedSignatureNotValid(_) + )) + ), + "the current-cycle quorum must be refused, got {result:?}" + ); + assert!( + engine.masternode_lists == lists, + "a refused QRInfo must leave the lists as they were" + ); + assert!(engine.rotated_quorums_per_cycle == cycles, "and the stored cycles"); + } + /// A quarter signature keyed by elimination can belong to another cycle, /// so a current-cycle quorum resting on one must settle as `Skipped` when /// it fails to verify. Aborting the feed there would turn a gap in the @@ -1686,6 +1997,218 @@ mod tests { ); } + /// Header storage answering from `inner`, except that reading the block + /// `unreadable` fails once it has been read `readable` times, as it does + /// when its header segment stops reading partway through a feed. When + /// `hangs`, that read never completes instead. + struct FailingFrom { + inner: MockHeaderStorage, + unreadable: BlockHash, + readable: usize, + reads: AtomicUsize, + hangs: bool, + } + + impl FailingFrom { + async fn read( + &self, + block_hash: &BlockHash, + read: StorageResult, + ) -> StorageResult { + if *block_hash == self.unreadable + && self.reads.fetch_add(1, Ordering::Relaxed) >= self.readable + { + if self.hangs { + std::future::pending::<()>().await; + } + return Err(StorageError::ReadFailed("the header segment is unreadable".into())); + } + read + } + } + + #[async_trait::async_trait] + impl BlockHeaderStorage for FailingFrom { + async fn store_headers(&mut self, headers: &[HashedBlockHeader]) -> StorageResult<()> { + self.inner.store_headers(headers).await + } + async fn store_headers_at_height( + &mut self, + headers: &[HashedBlockHeader], + height: u32, + ) -> StorageResult<()> { + self.inner.store_headers_at_height(headers, height).await + } + async fn load_headers( + &self, + range: std::ops::Range, + ) -> StorageResult> { + self.inner.load_headers(range).await + } + async fn get_header(&self, height: u32) -> StorageResult> { + let header = self.inner.get_header(height).await; + match header + .as_ref() + .ok() + .and_then(|header| header.as_ref()) + .map(|header| *header.hash()) + { + Some(block_hash) => self.read(&block_hash, header).await, + None => header, + } + } + async fn get_tip_height(&self) -> Option { + self.inner.get_tip_height().await + } + async fn get_tip(&self) -> Option { + self.inner.get_tip().await + } + async fn get_start_height(&self) -> Option { + self.inner.get_start_height().await + } + async fn get_stored_headers_len(&self) -> u32 { + self.inner.get_stored_headers_len().await + } + async fn get_header_height_by_hash(&self, hash: &BlockHash) -> StorageResult> { + self.read(hash, self.inner.get_header_height_by_hash(hash).await).await + } + async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> { + self.inner.truncate_above(target_height).await + } + } + + /// An engine holding `engine`'s lists and cycles over its header chain, + /// whose reads of `unreadable` fail once it was read `readable` times. + fn with_reads_failing( + engine: &TestEngine, + unreadable: BlockHash, + readable: usize, + ) -> MasternodeListEngine { + let headers = Arc::new(RwLock::new(FailingFrom { + inner: MockHeaderStorage(engine.headers().try_read().expect("unlocked").0.clone()), + unreadable, + readable, + reads: AtomicUsize::new(0), + hangs: false, + })); + let mut failing = MasternodeListEngine::new(engine.network, headers); + failing.masternode_lists = engine.masternode_lists.clone(); + failing.rotated_quorums_per_cycle = engine.rotated_quorums_per_cycle.clone(); + failing + } + + /// Feeds `qr_info` to [`with_reads_failing`]'s engine. Returns the result + /// and how often `unreadable` was read. + async fn feed_with_reads_failing( + engine: &TestEngine, + qr_info: &QRInfo, + unreadable: BlockHash, + readable: usize, + ) -> (Result>, usize) { + let mut failing = with_reads_failing(engine, unreadable, readable); + let result = failing.feed_qr_info(qr_info.clone()).await; + let reads = failing.headers.read().await.reads.load(Ordering::Relaxed); + (result, reads) + } + + /// Our header storage failing, at whichever read of a QRInfo's feed, never + /// refuses the QRInfo, which would pass our failure off as the message's: + /// the feed fails with `HeaderStorage`, or the read only informed + /// and the feed goes on without it. That holds on the path a rotated + /// quorum takes when it fails under a quarter signature keyed by + /// elimination, which is skipped rather than invalid, and for the blocks + /// a diff is applied from and to. + #[tokio::test] + async fn a_header_storage_failure_never_refuses_a_qr_info() { + let (mut engine, mut qr_info) = load_qrinfo_2240504_fixture().await; + for height in 2239488..2239488 + 32 { + if let Some(block_hash) = engine.block_hash_at(height) { + engine.forget_block(&block_hash); + } + } + let target = qr_info.last_commitment_per_index[0].quorum_hash; + qr_info.last_commitment_per_index[0].all_commitment_aggregated_signature = + foreign_aggregate_signature(&qr_info.last_commitment_per_index, target); + let blocks = [ + BlockHash::from_byte_array(target.to_byte_array()), + qr_info.mn_list_diff_h.block_hash, + qr_info.mn_list_diff_h.base_block_hash, + ]; + + for unreadable in blocks { + let (result, reads) = + feed_with_reads_failing(&engine, &qr_info, unreadable, usize::MAX).await; + assert!(result.is_ok(), "the QRInfo feeds while the storage reads, got {result:?}"); + assert!(reads > 0, "the feed reads {unreadable}"); + for readable in 0..reads { + let (result, _) = + feed_with_reads_failing(&engine, &qr_info, unreadable, readable).await; + assert!( + !matches!(result, Err(FeedError::Refused(_))), + "reading {unreadable} failing from its read {readable} on refused the \ + QRInfo: {result:?}" + ); + } + } + } + + /// Stopping the client, as following a fork does, aborts a manager task + /// still running after a grace period, dropping a QRInfo feed part way. + /// That leaves the engine as it was, as a refusal does, so a run started + /// again in the same process does not build on a QRInfo never taken. + #[tokio::test(start_paused = true)] + async fn a_qr_info_feed_dropped_part_way_leaves_the_engine_as_it_was() { + let (engine, qr_info) = load_qrinfo_2240504_fixture().await; + let quorum_hash = qr_info.last_commitment_per_index[0].quorum_hash; + let unreadable = BlockHash::from_byte_array(quorum_hash.to_byte_array()); + let mut fed = with_reads_failing(&engine, unreadable, usize::MAX); + let result = fed.feed_qr_info(qr_info.clone()).await; + assert!(result.is_ok(), "the QRInfo feeds while the storage reads, got {result:?}"); + assert!(fed.masternode_lists != engine.masternode_lists, "a whole feed changes the lists"); + let reads = fed.headers.read().await.reads.load(Ordering::Relaxed); + assert!(reads > 0, "the feed reads the quorum's block"); + let mut dropped = with_reads_failing(&engine, unreadable, reads - 1); + dropped.headers.try_write().expect("unlocked").hangs = true; + + let feed = tokio::time::timeout(Duration::from_secs(5), dropped.feed_qr_info(qr_info)); + assert!(feed.await.is_err(), "the feed waits on its last read of the quorum's block"); + + assert_eq!(dropped.headers.read().await.reads.load(Ordering::Relaxed), reads); + assert!( + dropped.masternode_lists == engine.masternode_lists, + "a dropped feed must leave the lists as they were" + ); + assert!( + dropped.rotated_quorums_per_cycle == engine.rotated_quorums_per_cycle, + "and the stored cycles" + ); + } + + /// Following a fork keeps a rotated cycle only when its base shows it was + /// mined below the fork. Our header storage failing to read the base shows + /// nothing, so the cycle goes rather than answer for blocks that may have + /// left the chain. + #[tokio::test] + async fn a_fork_drops_a_rotated_cycle_whose_base_cannot_be_read() { + let fork = 1_000_000; + let window_end = Network::Mainnet.isd_llmq_type().params().dkg_params.mining_window_end; + let mut engine = TestEngine::dummy_with_lists(&[fork]); + let readable = fork - window_end; + let unreadable = readable - 24; + for base in [readable, unreadable] { + engine.feed_block_height(base, BlockHash::dummy(base)); + engine.rotated_quorums_per_cycle.insert(BlockHash::dummy(base), BTreeMap::new()); + } + let mut failing = with_reads_failing(&engine, BlockHash::dummy(unreadable), 0); + + failing.truncate_above(fork).await; + + assert_eq!( + failing.rotated_quorums_per_cycle.keys().collect::>(), + vec![&BlockHash::dummy(readable)] + ); + } + /// The previous-cycle path is best-effort enrichment, so a corrupt /// aggregated signature there must only leave that quorum out of the /// stored cycle, never abort the whole feed. The rest of the cycle and @@ -1705,6 +2228,7 @@ mod tests { let cycle_base = engine .rotated_quorum_cycle_base(&qr_info.mn_list_diff_h.new_quorums[corrupt_position]) .await + .expect("the headers read") .expect("fixture must carry the corrupted quorum's cycle base"); let previous_cycle_key = engine.block_hash_at(cycle_base).expect("expected cycle base hash"); @@ -1891,7 +2415,10 @@ mod tests { .map(|height| (height, BlockHash::dummy(height))) .collect(); let mut engine = engine_knowing_blocks(&blocks); - assert!(engine.missing_work_block_list_requests().await.is_empty(), "no list, no request"); + assert!( + engine.missing_work_block_list_requests().await.expect("the headers read").is_empty(), + "no list, no request" + ); let quorum = |llmq_type, block_hash, verified| { let mut entry: QualifiedQuorumEntry = quorum_entry(llmq_type, block_hash, None).into(); @@ -1936,7 +2463,7 @@ mod tests { ); assert_eq!( - engine.missing_work_block_list_requests().await, + engine.missing_work_block_list_requests().await.expect("the headers read"), vec![ (BlockHash::all_zeros(), BlockHash::dummy(work(first))), (BlockHash::dummy(work(held)), BlockHash::dummy(work(second))), @@ -1945,6 +2472,37 @@ mod tests { ); } + /// A quorum whose block our header storage cannot read is left out of the + /// work-block lists a scan asks for, while the others still are. Only a + /// scan that has to tell what is no longer needed fails instead. + #[tokio::test] + async fn a_quorum_whose_block_does_not_read_leaves_the_others_asked_for() { + let work = |mined: CoreBlockHeight| mined - QUORUM_MEMBER_LIST_OFFSET; + let (tip, unreadable, readable) = (2_000_000, 1_999_008, 1_999_208); + let blocks: Vec<_> = [unreadable, work(unreadable), readable, work(readable), tip] + .into_iter() + .map(|height| (height, BlockHash::dummy(height))) + .collect(); + let mut engine = engine_knowing_blocks(&blocks); + let mut quorums: BTreeMap> = BTreeMap::new(); + for mined in [unreadable, readable] { + let entry = quorum_entry(LLMQType::Llmqtype400_60, BlockHash::dummy(mined), None); + quorums.insert(entry.quorum_hash, Arc::new(entry.into())); + } + let quorums = BTreeMap::from([(LLMQType::Llmqtype400_60, quorums)]); + engine.masternode_lists.insert( + tip, + MasternodeList::build(BTreeMap::new(), quorums, BlockHash::dummy(tip), tip).build(), + ); + let failing = with_reads_failing(&engine, BlockHash::dummy(unreadable), 0); + + assert_eq!( + failing.readable_missing_work_block_list_requests().await, + vec![(BlockHash::all_zeros(), BlockHash::dummy(work(readable)))] + ); + assert!(failing.missing_work_block_list_requests().await.is_err()); + } + /// The tip list of an engine holding `quorums` (type, mined height, /// status) at `tip`, with an older list at `older_list`. Every quorum's /// block and work block are known, and so are `extra_blocks`. diff --git a/dash-spv/src/sml_engine/helpers.rs b/dash-spv/src/sml_engine/helpers.rs index 83f387aeb..791cd4df2 100644 --- a/dash-spv/src/sml_engine/helpers.rs +++ b/dash-spv/src/sml_engine/helpers.rs @@ -74,7 +74,7 @@ impl MasternodeListEngine { let mut below_floor = BTreeSet::new(); for cycle_hash in self.rotated_quorums_per_cycle.keys() { - if self.height_of(cycle_hash).await.is_some_and(|height| height < floor) { + if self.height_if_known(cycle_hash).await.is_some_and(|height| height < floor) { below_floor.insert(*cycle_hash); } } @@ -85,7 +85,8 @@ impl MasternodeListEngine { /// chain has dropped them for another branch: the lists above it, and the /// rotated cycles whose base block left the chain or whose mining window /// ends above it, since their commitments may have been mined on the - /// dropped branch. + /// dropped branch. A cycle whose base block cannot be read goes too: + /// nothing shows it was mined below `height`. pub(crate) async fn truncate_above(&mut self, height: CoreBlockHeight) { if let Some(above) = height.checked_add(1) { self.masternode_lists.split_off(&above); @@ -95,7 +96,7 @@ impl MasternodeListEngine { let mut dropped = BTreeSet::new(); for cycle_hash in self.rotated_quorums_per_cycle.keys() { let mined_below = self - .height_of(cycle_hash) + .height_if_known(cycle_hash) .await .is_some_and(|base| base.saturating_add(mining_window_end) <= height); if !mined_below { diff --git a/dash-spv/src/sml_engine/rotated_quorum_construction.rs b/dash-spv/src/sml_engine/rotated_quorum_construction.rs index 0302d8a83..d8d77a6bf 100644 --- a/dash-spv/src/sml_engine/rotated_quorum_construction.rs +++ b/dash-spv/src/sml_engine/rotated_quorum_construction.rs @@ -39,7 +39,7 @@ impl MasternodeListEngine { quorum_hash, index, }; - let Some(height) = self.height_of(&quorum_hash).await else { + let Some(height) = self.height_if_known(&quorum_hash).await else { members.push(Err(QuorumValidationError::RequiredBlockNotPresent( quorum_hash, "rotated quorum height".to_string(), diff --git a/dash-spv/src/sml_engine/validation.rs b/dash-spv/src/sml_engine/validation.rs index ff5057dcb..2d27f30c7 100644 --- a/dash-spv/src/sml_engine/validation.rs +++ b/dash-spv/src/sml_engine/validation.rs @@ -23,7 +23,7 @@ impl MasternodeListEngine { ) -> Result<(), QuorumValidationError> { quorum.quorum_entry.validate_structure()?; let quorum_hash = quorum.quorum_entry.quorum_hash; - let height = self.height_of(&quorum_hash).await.ok_or_else(|| { + let height = self.height_if_known(&quorum_hash).await.ok_or_else(|| { QuorumValidationError::RequiredBlockNotPresent(quorum_hash, "quorum height".to_string()) })?; let work_height = height.saturating_sub(QUORUM_MEMBER_LIST_OFFSET); diff --git a/dash-spv/src/storage/masternode.rs b/dash-spv/src/storage/masternode.rs index f2a1b8cd7..4614d9bd8 100644 --- a/dash-spv/src/storage/masternode.rs +++ b/dash-spv/src/storage/masternode.rs @@ -190,7 +190,7 @@ impl MessageLog { }; let mut newest_base = None; for base in message.base_hashes() { - newest_base = newest_base.max(engine.height_of(&base).await); + newest_base = newest_base.max(engine.height_if_known(&base).await); } plan.push((newest_base.unwrap_or(height), height, kind, path)); } @@ -201,8 +201,8 @@ impl MessageLog { for (_, height, kind, path) in plan { if let Some(message) = self.read_entry(height, kind, path).await { match message.apply(&mut engine).await { - true => applied += 1, - false => tracing::warn!("Masternode message at {height} does not apply"), + Ok(()) => applied += 1, + Err(e) => tracing::warn!("Masternode message at {height} does not apply: {e}"), } } visit(&engine); @@ -272,10 +272,16 @@ impl MessageLog { } impl Message { - async fn apply(self, engine: &mut MasternodeListEngine) -> bool { + /// Feeds the message to `engine`, failing with why it was not taken. + async fn apply( + self, + engine: &mut MasternodeListEngine, + ) -> Result<(), String> { match self { - Message::QrInfo(qr_info) => engine.feed_qr_info(*qr_info).await.is_ok(), - Message::Diff(diff) => engine.apply_diff(*diff).await.is_ok(), + Message::QrInfo(qr_info) => { + engine.feed_qr_info(*qr_info).await.map(drop).map_err(|e| e.to_string()) + } + Message::Diff(diff) => engine.apply_diff(*diff).await.map_err(|e| e.to_string()), } } diff --git a/dash-spv/src/sync/masternodes/sync_manager.rs b/dash-spv/src/sync/masternodes/sync_manager.rs index 9108923d8..8272c1e94 100644 --- a/dash-spv/src/sync/masternodes/sync_manager.rs +++ b/dash-spv/src/sync/masternodes/sync_manager.rs @@ -119,8 +119,8 @@ impl SyncManager for MasternodesManager { } }; - let request_pairs = engine.missing_work_block_list_requests().await; - let tip_height = engine.height_of(&qr_info.mn_list_diff_tip.block_hash).await; + let request_pairs = engine.readable_missing_work_block_list_requests().await; + let tip_height = engine.height_if_known(&qr_info.mn_list_diff_tip.block_hash).await; drop(engine); if let Some(height) = tip_height { @@ -191,7 +191,7 @@ impl SyncManager for MasternodesManager { tracing::debug!("Processing MnListDiff message for {}", diff.block_hash); let mut engine = self.engine.write().await; - let Some(target_height) = engine.height_of(&diff.block_hash).await else { + let Some(target_height) = engine.height_if_known(&diff.block_hash).await else { drop(engine); tracing::warn!( "Height not found for MnListDiff block {}, requeuing for retry", diff --git a/dash-spv/src/test_utils/mod.rs b/dash-spv/src/test_utils/mod.rs index 77a3fd7cd..d767b055f 100644 --- a/dash-spv/src/test_utils/mod.rs +++ b/dash-spv/src/test_utils/mod.rs @@ -31,7 +31,7 @@ pub use wallet::{ pub(crate) use node::DashCoreConfig; -pub use crate::sml_engine::{MasternodeListEngine, WORK_DIFF_DEPTH}; +pub use crate::sml_engine::{FeedError, MasternodeListEngine, WORK_DIFF_DEPTH}; /// Storage types for tests and benches that open the client's storage directly. pub use crate::storage::{ diff --git a/dash-spv/tests/dashd_masternode/tests_sync.rs b/dash-spv/tests/dashd_masternode/tests_sync.rs index 75a3a8486..ea7bdc07d 100644 --- a/dash-spv/tests/dashd_masternode/tests_sync.rs +++ b/dash-spv/tests/dashd_masternode/tests_sync.rs @@ -312,7 +312,8 @@ async fn test_masternode_list_sync_with_quorum_rotation() { let engine = client_handle.engine.read().await; let mut max_cycle = 0; for cycle_hash in engine.rotated_quorums_per_cycle().keys() { - max_cycle = max_cycle.max(engine.height_of(cycle_hash).await.unwrap_or(0)); + let height = engine.height_of(cycle_hash).await.expect("the headers read"); + max_cycle = max_cycle.max(height.unwrap_or(0)); } max_cycle }; @@ -443,7 +444,7 @@ async fn test_rotated_quorums_stored_when_sync_starts_post_dkg() { assert_all_rotated_quorums_verified(&engine); let mut heights: Vec = Vec::new(); for cycle_hash in engine.rotated_quorums_per_cycle().keys() { - heights.extend(engine.height_of(cycle_hash).await); + heights.extend(engine.height_of(cycle_hash).await.expect("the headers read")); } heights.sort_unstable(); heights.dedup();