From 22104eb6c41ebaf12521b442782e9e2b389640d5 Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Mon, 26 Oct 2026 09:30:29 -0700 Subject: [PATCH 1/2] feat: persist masternode list engine state --- dash-spv/Cargo.toml | 4 +- dash-spv/src/client/core.rs | 5 - dash-spv/src/client/lifecycle.rs | 8 +- dash-spv/src/client/queries.rs | 64 +- dash-spv/src/sml_engine.rs | 24 + dash-spv/src/storage/masternode.rs | 784 ++++-------------- dash-spv/src/storage/migrator/mod.rs | 7 +- dash-spv/src/storage/migrator/v3.rs | 23 + dash-spv/src/storage/mod.rs | 35 +- dash-spv/src/sync/masternodes/manager.rs | 40 +- dash-spv/src/sync/masternodes/sync_manager.rs | 67 +- dash-spv/tests/dashd_masternode/helpers.rs | 2 +- dash-spv/tests/dashd_masternode/tests_sync.rs | 14 +- 13 files changed, 272 insertions(+), 805 deletions(-) create mode 100644 dash-spv/src/storage/migrator/v3.rs diff --git a/dash-spv/Cargo.toml b/dash-spv/Cargo.toml index 5890b9fe9..3527e3022 100644 --- a/dash-spv/Cargo.toml +++ b/dash-spv/Cargo.toml @@ -47,6 +47,9 @@ hex-conservative = { workspace = true } # Parallelization rayon = { workspace = true } +# Masternode engine snapshots +bincode = { workspace = true, features = ["serde"] } + # Test utilities (optional) tempfile = { version = "3.0", optional = true } dashcore-rpc = { path = "../rpc-client", optional = true } @@ -56,7 +59,6 @@ serde = { workspace = true, features = ["std", "derive"] } serde_json = "1.0" [dev-dependencies] -bincode = { workspace = true } dash-spv = { path = ".", features = ["test-utils"] } dashcore = { path = "../dash", features = ["test-utils"] } key-wallet = { path = "../key-wallet", features = ["test-utils"] } diff --git a/dash-spv/src/client/core.rs b/dash-spv/src/client/core.rs index dba91211c..b4ec634e4 100644 --- a/dash-spv/src/client/core.rs +++ b/dash-spv/src/client/core.rs @@ -181,11 +181,6 @@ impl DashSpvClient { self.storage.lock().await.clear().await?; self.sync_coordinator.lock().await.reset_progress(); - if let Some(engine) = &self.masternode_engine { - let network = self.config.read().await.network; - let headers = self.storage.lock().await.block_headers(); - *engine.write().await = MasternodeListEngine::new(network, headers); - } Ok(()) } diff --git a/dash-spv/src/client/lifecycle.rs b/dash-spv/src/client/lifecycle.rs index 587d9088e..8999c4c58 100644 --- a/dash-spv/src/client/lifecycle.rs +++ b/dash-spv/src/client/lifecycle.rs @@ -53,8 +53,7 @@ impl DashSpvClient { let masternode_engine = { if config.enable_masternodes { - let engine = storage.masternodes().read().await.load_engine().await; - Some(Arc::new(RwLock::new(engine))) + Some(storage.masternodes().read().await.engine()) } else { None } @@ -284,13 +283,10 @@ impl DashSpvClient { height: u32, ) -> std::result::Result<(), Box> { self.wallet.write().await.truncate_above(height)?; + let truncated = StorageManager::truncate_above(&mut *self.storage.lock().await, height).await; - if let Some(engine) = &self.masternode_engine { - engine.write().await.truncate_above(height).await; - } - Ok(truncated?) } diff --git a/dash-spv/src/client/queries.rs b/dash-spv/src/client/queries.rs index a56ee8150..d4d186324 100644 --- a/dash-spv/src/client/queries.rs +++ b/dash-spv/src/client/queries.rs @@ -9,12 +9,12 @@ use crate::error::{Result, SpvError}; use crate::network::NetworkManager; use crate::sml_engine::MasternodeListEngine; -use crate::storage::{BlockHeaderStorage, PersistentBlockHeaderStorage, StorageManager}; +use crate::storage::PersistentBlockHeaderStorage; use dashcore::hashes::Hash; use dashcore::sml::llmq_type::LLMQType; use dashcore::sml::masternode_list::MasternodeList; use dashcore::sml::quorum_entry::qualified_quorum_entry::QualifiedQuorumEntry; -use dashcore::{BlockHash, QuorumHash}; +use dashcore::QuorumHash; use key_wallet_manager::WalletInterface; use std::sync::Arc; use tokio::sync::RwLock; @@ -67,8 +67,8 @@ impl DashSpvClient { engine.blocking_read().latest_masternode_list().cloned() } - /// Get a quorum entry by type and hash at a specific block height. A height - /// whose lists the engine no longer holds is rebuilt from storage. + /// Get a quorum entry by type and hash at a specific block height. Only + /// heights within the lists the engine holds resolve. /// Returns `SpvError::QuorumLookupError` if the quorum is not found. pub async fn get_quorum_at_height( &self, @@ -76,32 +76,12 @@ impl DashSpvClient { quorum_type: LLMQType, quorum_hash: QuorumHash, ) -> Result { - let (in_memory, oldest_list) = { - let engine = self.masternode_list_engine()?; - let engine = engine.read().await; - let quorum = engine - .quorum_entry_for_hash_at_or_before_height(quorum_type, quorum_hash, height) - .map(|(_, quorum)| quorum.clone()); - (quorum, engine.masternode_lists.keys().next().copied()) - }; - - let quorum = match in_memory { - Some(quorum) => Some(quorum), - None => { - let (headers, masternodes) = { - let storage = self.storage.lock().await; - (storage.block_headers(), storage.masternodes()) - }; - let quorum_block = BlockHash::from_byte_array(quorum_hash.to_byte_array()); - let mined = headers.read().await.get_header_height_by_hash(&quorum_block).await?; - if storage_may_hold(mined, height, oldest_list) { - let log = masternodes.read().await.message_log(); - log.quorum_entry_at_or_before(quorum_type, quorum_hash, height).await - } else { - None - } - } - }; + let engine = self.masternode_list_engine()?; + let quorum = engine + .read() + .await + .quorum_entry_for_hash_at_or_before_height(quorum_type, quorum_hash, height) + .map(|(_, quorum)| quorum.clone()); quorum.ok_or_else(|| { let message = format!( @@ -115,27 +95,3 @@ impl DashSpvClient { }) } } - -/// Whether the stored messages can resolve a quorum the engine missed. A -/// quorum's hash is the block it was mined in, so a hash outside the header -/// chain names no quorum, one mined above `height` was not active there, and -/// one mined at or above the engine's oldest list was held by every list the -/// engine kept since. Only a quorum mined below that list may sit in lists the -/// engine has pruned. -fn storage_may_hold(mined: Option, height: u32, oldest_list: Option) -> bool { - mined.is_some_and(|mined| mined <= height && oldest_list.is_none_or(|oldest| mined < oldest)) -} - -#[cfg(test)] -mod tests { - use super::storage_may_hold; - - #[test] - fn only_a_quorum_mined_below_the_oldest_list_goes_to_storage() { - assert!(!storage_may_hold(None, 1_000, Some(500)), "not a block in the header chain"); - assert!(!storage_may_hold(Some(1_001), 1_000, Some(500)), "mined above the lookup"); - assert!(!storage_may_hold(Some(600), 1_000, Some(500)), "the engine held all its lists"); - assert!(storage_may_hold(Some(400), 1_000, Some(500)), "its lists may be pruned"); - assert!(storage_may_hold(Some(400), 1_000, None), "the engine holds no list yet"); - } -} diff --git a/dash-spv/src/sml_engine.rs b/dash-spv/src/sml_engine.rs index 169f3d2c7..d4114473b 100644 --- a/dash-spv/src/sml_engine.rs +++ b/dash-spv/src/sml_engine.rs @@ -244,6 +244,30 @@ impl MasternodeListEngine { } } + /// The engine's lists and rotated cycles as bytes, see [`Self::from_snapshot`]. + pub fn snapshot(&self) -> Result, bincode::error::EncodeError> { + bincode::serde::encode_to_vec( + (&self.masternode_lists, &self.rotated_quorums_per_cycle), + bincode::config::standard(), + ) + } + + /// An engine holding the lists and rotated cycles [`Self::snapshot`] encoded. + pub fn from_snapshot( + network: Network, + headers: Arc>, + bytes: &[u8], + ) -> Result { + let ((masternode_lists, rotated_quorums_per_cycle), _) = + bincode::serde::decode_from_slice(bytes, bincode::config::standard())?; + Ok(Self { + headers, + masternode_lists, + rotated_quorums_per_cycle, + network, + }) + } + /// 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 diff --git a/dash-spv/src/storage/masternode.rs b/dash-spv/src/storage/masternode.rs index 4614d9bd8..0abbfc7d8 100644 --- a/dash-spv/src/storage/masternode.rs +++ b/dash-spv/src/storage/masternode.rs @@ -1,49 +1,20 @@ -use std::collections::{BTreeMap, HashSet}; use std::path::{Path, PathBuf}; use std::sync::Arc; use async_trait::async_trait; use tokio::sync::RwLock; -use crate::sml_engine::{qr_info_diffs, MasternodeListEngine, WORK_DIFF_DEPTH}; -use dashcore::consensus::{deserialize, serialize, Decodable, Encodable}; -use dashcore::network::message_qrinfo::QRInfo; -use dashcore::network::message_sml::MnListDiff; +use crate::sml_engine::MasternodeListEngine; use dashcore::prelude::CoreBlockHeight; -use dashcore::sml::llmq_type::network::NetworkLLMQExt; -use dashcore::sml::llmq_type::LLMQType; -use dashcore::sml::quorum_entry::qualified_quorum_entry::QualifiedQuorumEntry; -use dashcore::{BlockHash, Network, QuorumHash}; +use dashcore::Network; use crate::error::{StorageError, StorageResult}; use crate::storage::{io::atomic_write, BlockHeaderStorage}; -type IndexMap = BTreeMap; - -enum Message { - Diff(Box), - QrInfo(Box), -} - -/// At one height a diff replays before a QRInfo. -#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] -enum Kind { - Diff, - QrInfo, -} - #[async_trait] pub trait MasternodeStorage: Send + Sync + 'static { - async fn store_diff(&mut self, height: CoreBlockHeight, diff: &MnListDiff) - -> StorageResult<()>; - - async fn store_qr_info( - &mut self, - height: CoreBlockHeight, - qr_info: &QRInfo, - ) -> StorageResult<()>; - - /// Drops the messages stored above `height`. + /// Drops what the engine holds above `height` and stores it again, so the + /// snapshot never holds lists of an abandoned branch. async fn truncate_above(&mut self, height: CoreBlockHeight) -> StorageResult<()>; } @@ -51,699 +22,284 @@ pub struct PersistentMasternodeStorage { storage_path: PathBuf, headers: Arc>, network: Network, - diffs: IndexMap, - qr_infos: IndexMap, -} - -/// The stored messages as of when it was taken. A replay runs on this, so the -/// caller can release the storage lock before it starts. -pub struct MessageLog { - headers: Arc>, - network: Network, - diffs: IndexMap, - qr_infos: IndexMap, + engine: Arc>>, } impl PersistentMasternodeStorage { const FOLDER_NAME: &str = "masternodes"; - const DIFF_PREFIX: &str = "diff_"; - const QRINFO_PREFIX: &str = "qrinfo_"; - const EXTENSION: &str = "dat"; + const FILE_NAME: &str = "engine.dat"; + /// Opens the storage and restores the engine from the snapshot when its + /// newest list is still in the header chain at its height. The engine + /// starts empty otherwise. pub async fn open( storage_path: impl Into + Send, headers: Arc>, network: Network, - ) -> StorageResult { + ) -> Self { let storage_path = storage_path.into(); - let (diffs, qr_infos) = Self::index_folder(&storage_path.join(Self::FOLDER_NAME)).await?; - Ok(PersistentMasternodeStorage { - storage_path, - headers, - network, - diffs, - qr_infos, - }) - } + let path = storage_path.join(Self::FOLDER_NAME).join(Self::FILE_NAME); - fn folder(&self) -> PathBuf { - self.storage_path.join(Self::FOLDER_NAME) - } - - fn file_name(prefix: &str, height: CoreBlockHeight) -> String { - format!("{prefix}{height}.{}", Self::EXTENSION) - } - - fn height_from_file_name(name: &str, prefix: &str) -> Option { - name.strip_prefix(prefix)?.strip_suffix(&format!(".{}", Self::EXTENSION))?.parse().ok() - } + let engine = match Self::read_snapshot(network, &headers, &path).await { + Ok(Some(engine)) => Some(engine), + Ok(None) => None, + Err(e) => { + if path.exists() { + tokio::fs::remove_file(&path).await.ok(); + } - async fn index_folder(folder: &Path) -> StorageResult<(IndexMap, IndexMap)> { - let mut diffs = BTreeMap::new(); - let mut qr_infos = BTreeMap::new(); + tracing::warn!("Not restoring the masternode engine snapshot: {e}"); + None + } + }; - if !folder.exists() { - return Ok((diffs, qr_infos)); - } + let engine = + engine.unwrap_or_else(|| MasternodeListEngine::new(network, Arc::clone(&headers))); - let mut entries = tokio::fs::read_dir(folder).await?; - while let Some(entry) = entries.next_entry().await? { - let path = entry.path(); - let Some(name) = path.file_name().and_then(|n| n.to_str()) else { - continue; - }; - if let Some(height) = Self::height_from_file_name(name, Self::DIFF_PREFIX) { - diffs.insert(height, path); - } else if let Some(height) = Self::height_from_file_name(name, Self::QRINFO_PREFIX) { - qr_infos.insert(height, path); - } + PersistentMasternodeStorage { + storage_path, + headers, + network, + engine: Arc::new(RwLock::new(engine)), } - - Ok((diffs, qr_infos)) } - async fn store_message( - folder: &Path, - index: &mut IndexMap, - prefix: &str, - height: CoreBlockHeight, - message: &T, - ) -> StorageResult<()> { - tokio::fs::create_dir_all(folder).await?; - let path = folder.join(Self::file_name(prefix, height)); - atomic_write(&path, &serialize(message)).await?; - index.insert(height, path); - Ok(()) - } + async fn read_snapshot( + network: Network, + headers: &Arc>, + path: &Path, + ) -> StorageResult>> { + if !path.exists() { + return Ok(None); + } - async fn read_message(path: &Path) -> StorageResult { let bytes = tokio::fs::read(path).await?; - deserialize(&bytes).map_err(|e| { - StorageError::Corruption(format!("Failed to decode {}: {e}", path.display())) - }) - } - async fn read_entry(path: &Path, kind: Kind) -> StorageResult { - Ok(match kind { - Kind::Diff => Message::Diff(Box::new(Self::read_message(path).await?)), - Kind::QrInfo => Message::QrInfo(Box::new(Self::read_message(path).await?)), - }) - } + let engine = MasternodeListEngine::from_snapshot(network, Arc::clone(headers), &bytes) + .map_err(|e| StorageError::ReadFailed(e.to_string()))?; - pub fn message_log(&self) -> MessageLog { - MessageLog { - headers: Arc::clone(&self.headers), - network: self.network, - diffs: self.diffs.clone(), - qr_infos: self.qr_infos.clone(), - } - } + let Some(engine_tip) = engine.latest_masternode_list() else { + tracing::warn!("The masternode engine snapshot has no lists, ignoring it"); + return Ok(None); + }; - pub(crate) async fn load_engine(&self) -> MasternodeListEngine { - self.message_log().replay(CoreBlockHeight::MAX, |_| {}).await - } -} + let in_chain = + headers.read().await.get_header_height_by_hash(&engine_tip.block_hash).await?; -impl MessageLog { - /// Rebuilds the engine from the messages stored up to `until`, calling - /// `visit` after each one. Messages apply in the order of their newest - /// base, so each comes after the one that built the list it extends. They - /// are read once to plan that order and again to apply, so only one is held - /// in memory at a time, and the header storage is locked per message. - async fn replay( - &self, - until: CoreBlockHeight, - mut visit: impl FnMut(&MasternodeListEngine), - ) -> MasternodeListEngine { - let mut engine = MasternodeListEngine::new(self.network, Arc::clone(&self.headers)); - - let entries = - self.diffs.range(..=until).map(|(height, path)| (*height, Kind::Diff, path)).chain( - self.qr_infos.range(..=until).map(|(height, path)| (*height, Kind::QrInfo, path)), + if in_chain != Some(engine_tip.known_height) { + tracing::warn!( + "The masternode engine snapshot's tip {} is not in the header chain, ignoring it", + engine_tip.known_height ); - - let mut plan = Vec::new(); - for (height, kind, path) in entries { - let Some(message) = self.read_entry(height, kind, path).await else { - continue; - }; - let mut newest_base = None; - for base in message.base_hashes() { - newest_base = newest_base.max(engine.height_if_known(&base).await); - } - plan.push((newest_base.unwrap_or(height), height, kind, path)); - } - plan.sort_by_key(|(newest_base, height, kind, _)| (*newest_base, *height, *kind)); - - let total = plan.len(); - let mut applied = 0; - for (_, height, kind, path) in plan { - if let Some(message) = self.read_entry(height, kind, path).await { - match message.apply(&mut engine).await { - Ok(()) => applied += 1, - Err(e) => tracing::warn!("Masternode message at {height} does not apply: {e}"), - } - } - visit(&engine); + return Ok(None); } - tracing::debug!( - "Replayed {applied}/{total} masternode messages into {} masternode lists", - engine.masternode_lists.len() - ); - - engine + Ok(Some(engine)) } - /// Reads the message stored at `height` while its block is still in the - /// header chain there, or logs why it is skipped. - async fn read_entry( - &self, - height: CoreBlockHeight, - kind: Kind, - path: &Path, - ) -> Option { - let message = match PersistentMasternodeStorage::::read_entry(path, kind).await { - Ok(message) => message, - Err(e) => { - tracing::warn!("Skipping unreadable masternode message at {height}: {e}"); - return None; - } - }; - let block_hash = message.block_hash(); - let stored_height = - self.headers.read().await.get_header_height_by_hash(&block_hash).await.ok().flatten(); - if stored_height != Some(height) { - tracing::warn!( - "Skipping masternode message at {height}: {block_hash} is no longer in the \ - header chain" - ); - return None; - } - Some(message) + /// Shared access to the engine this storage persists. + pub fn engine(&self) -> Arc>> { + Arc::clone(&self.engine) } - /// The quorum as the engine would have resolved it at `height` before its - /// retention window dropped the lists holding it. A QRInfo builds lists - /// back to its h-4c work block, so messages stored up to six rotation - /// cycles above `height` still count. - pub async fn quorum_entry_at_or_before( - &self, - llmq_type: LLMQType, - quorum_hash: QuorumHash, - height: CoreBlockHeight, - ) -> Option { - let cycle = self.network.isd_llmq_type().params().dkg_params.interval; - let until = height.saturating_add(cycle.saturating_mul(6)).saturating_add(WORK_DIFF_DEPTH); - let mut found: Option<(CoreBlockHeight, QualifiedQuorumEntry)> = None; - self.replay(until, |engine| { - let hit = - engine.quorum_entry_for_hash_at_or_before_height(llmq_type, quorum_hash, height); - if let Some((list_height, quorum)) = hit { - if found.as_ref().is_none_or(|(best, _)| list_height >= *best) { - found = Some((list_height, quorum.clone())); - } - } - }) - .await; - found.map(|(_, quorum)| quorum) + fn folder(&self) -> PathBuf { + self.storage_path.join(Self::FOLDER_NAME) } -} -impl Message { - /// 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.map(drop).map_err(|e| e.to_string()) - } - Message::Diff(diff) => engine.apply_diff(*diff).await.map_err(|e| e.to_string()), - } + fn path(&self) -> PathBuf { + self.folder().join(Self::FILE_NAME) } - fn block_hash(&self) -> BlockHash { - match self { - Message::Diff(diff) => diff.block_hash, - Message::QrInfo(qr_info) => qr_info.mn_list_diff_tip.block_hash, - } - } + /// Replaces the snapshot with the engine. An engine without lists is not + /// stored, and drops the snapshot instead, so it never outlives the lists + /// it held. + pub async fn persist(&mut self) -> StorageResult<()> { + let engine = self.engine.read().await; - /// Block hashes of the lists this message is applied on top of. A QRInfo's - /// diffs chain from its oldest one, which Core sends as an empty diff from - /// the base to itself, so only bases no other diff of it produces count. - fn base_hashes(&self) -> Vec { - match self { - Message::Diff(diff) => vec![diff.base_block_hash], - Message::QrInfo(qr_info) => qr_info_bases(qr_info).into_iter().collect(), + if engine.latest_masternode_list().is_none() { + drop(engine); + let _ = tokio::fs::remove_file(self.path()).await; + return Ok(()); } - } -} -/// Block hashes of the lists a QRInfo's diffs build. -fn qr_info_built(qr_info: &QRInfo) -> HashSet { - qr_info_diffs(qr_info) - .iter() - .filter(|d| d.block_hash != d.base_block_hash) - .map(|d| d.block_hash) - .collect() -} + let bytes = engine.snapshot().map_err(|e| StorageError::WriteFailed(e.to_string()))?; -/// Block hashes of the lists a QRInfo is applied on top of. -fn qr_info_bases(qr_info: &QRInfo) -> HashSet { - let built = qr_info_built(qr_info); - qr_info_diffs(qr_info) - .iter() - .map(|d| d.base_block_hash) - .filter(|base| !built.contains(base)) - .collect() -} + drop(engine); -#[async_trait] -impl MasternodeStorage for PersistentMasternodeStorage { - async fn store_diff( - &mut self, - height: CoreBlockHeight, - diff: &MnListDiff, - ) -> StorageResult<()> { - let folder = self.folder(); - Self::store_message(&folder, &mut self.diffs, Self::DIFF_PREFIX, height, diff).await + tokio::fs::create_dir_all(self.folder()).await?; + atomic_write(&self.path(), &bytes).await?; + + Ok(()) } - async fn store_qr_info( - &mut self, - height: CoreBlockHeight, - qr_info: &QRInfo, - ) -> StorageResult<()> { - // A QRInfo for the tip already stored, requested from a list the stored - // one built, would replace the message it is applied on top of. - if let Some(path) = self.qr_infos.get(&height) { - if let Ok(stored) = Self::read_message::(path).await { - if stored.mn_list_diff_tip.block_hash == qr_info.mn_list_diff_tip.block_hash - && !qr_info_bases(qr_info).is_disjoint(&qr_info_built(&stored)) - { - return Ok(()); - } - } - } - let folder = self.folder(); - Self::store_message(&folder, &mut self.qr_infos, Self::QRINFO_PREFIX, height, qr_info).await + pub async fn clear(&mut self, headers: Arc>) { + self.headers = Arc::clone(&headers); + *self.engine.write().await = MasternodeListEngine::new(self.network, headers); } +} +#[async_trait] +impl MasternodeStorage for PersistentMasternodeStorage { async fn truncate_above(&mut self, height: CoreBlockHeight) -> StorageResult<()> { - // A message leaves its index only once its file is gone, so a retry - // finds what a failed deletion left. - for index in [&mut self.diffs, &mut self.qr_infos] { - while let Some(entry) = index.last_entry().filter(|entry| *entry.key() > height) { - match tokio::fs::remove_file(entry.get()).await { - Err(e) if e.kind() != std::io::ErrorKind::NotFound => return Err(e.into()), - _ => entry.remove(), - }; - } - } + self.engine.write().await.truncate_above(height).await; - Ok(()) + self.persist().await } } #[cfg(test)] mod tests { use super::*; + use crate::sml_engine::test_support::TestEngine; use crate::test_utils::MockHeaderStorage; + use dashcore::sml::masternode_list::MasternodeList; + use dashcore::BlockHash; use dashcore_hashes::Hash; use tempfile::TempDir; use test_case::test_case; + type TestStorage = PersistentMasternodeStorage; + fn hash(byte: u8) -> BlockHash { BlockHash::from_slice(&[byte; 32]).unwrap() } - async fn open_storage( - dir: &TempDir, - heights: &[(u8, u32)], - ) -> PersistentMasternodeStorage { - let map = heights.iter().map(|(b, h)| (hash(*b), *h)).collect(); - PersistentMasternodeStorage::open( - dir.path(), - Arc::new(RwLock::new(MockHeaderStorage(map))), - Network::Regtest, - ) - .await - .expect("open") + async fn open_storage(dir: &TempDir, headers: Arc>) -> TestStorage { + PersistentMasternodeStorage::open(dir.path(), headers, Network::Mainnet).await } - #[tokio::test] - async fn truncating_drops_the_messages_stored_above() { - let dir = TempDir::new().unwrap(); - let heights = [(0x00, 0), (0xAA, 100), (0xBB, 200)]; - let mut storage = open_storage(&dir, &heights).await; + /// An engine whose header chain holds `heights` and whose lists stand at + /// each of them. + fn engine_with_lists(heights: &[(u8, u32)]) -> TestEngine { + let mut engine = TestEngine::knowing( + Network::Mainnet, + heights.iter().map(|(b, h)| (hash(*b), *h)).collect(), + ); + for (byte, height) in heights { + engine.masternode_lists.insert(*height, MasternodeList::empty(hash(*byte), *height)); + } + engine + } - storage.store_diff(100, &MnListDiff::dummy(0x00, 0xAA)).await.unwrap(); - storage.store_diff(200, &MnListDiff::dummy(0xAA, 0xBB)).await.unwrap(); - storage.store_qr_info(200, &QRInfo::dummy(0xBB)).await.unwrap(); + /// Makes `engine` the storage's engine and persists it. + async fn persist(storage: &mut TestStorage, engine: TestEngine) { + *storage.engine().write().await = engine; + storage.persist().await.unwrap(); + } - storage.truncate_above(100).await.unwrap(); + async fn list_heights(storage: &TestStorage) -> Vec { + storage.engine().read().await.masternode_lists.keys().copied().collect() + } - let reopened = open_storage(&dir, &heights).await; - assert_eq!(reopened.diffs.keys().copied().collect::>(), vec![100]); - assert!(reopened.qr_infos.is_empty()); + fn snapshot_file(dir: &TempDir) -> PathBuf { + dir.path().join("masternodes").join("engine.dat") } #[tokio::test] - async fn a_reopened_storage_replays_what_was_stored() { + async fn a_reopened_storage_restores_the_persisted_engine() { let dir = TempDir::new().unwrap(); - let heights = [(0x00, 0), (0xAA, 100), (0xBB, 200)]; - { - let mut storage = open_storage(&dir, &heights).await; - storage.store_diff(100, &MnListDiff::dummy(0x00, 0xAA)).await.unwrap(); - storage.store_diff(200, &MnListDiff::dummy(0xAA, 0xBB)).await.unwrap(); - } - - let engine = open_storage(&dir, &heights).await.load_engine().await; - - assert_eq!(engine.masternode_lists.keys().copied().collect::>(), vec![100, 200]); - assert_eq!(engine.masternode_lists[&200].block_hash, hash(0xBB)); - assert_eq!(engine.masternode_lists[&200].masternodes.len(), 2); - } + let engine = TestEngine::mainnet_fixture(); + let headers = engine.headers(); + let expected = engine.snapshot().unwrap(); + persist(&mut open_storage(&dir, Arc::clone(&headers)).await, engine).await; - #[tokio::test] - async fn replay_verifies_the_newest_lists_non_rotating_quorums() { - use dashcore::bls_sig_utils::{BLSPublicKey, BLSSignature}; - use dashcore::hash_types::QuorumVVecHash; - use dashcore::network::message_sml::QuorumCLSigObject; - use dashcore::sml::llmq_entry_verification::{ - LLMQEntryVerificationSkipStatus, LLMQEntryVerificationStatus, - }; - use dashcore::sml::llmq_type::LLMQType; - use dashcore::transaction::special_transaction::quorum_commitment::QuorumEntry; - use dashcore::QuorumHash; + let reopened = open_storage(&dir, headers).await; - let dir = TempDir::new().unwrap(); - let quorum_hash = QuorumHash::from_byte_array([0xAB; 32]); - let diff = MnListDiff { - new_quorums: vec![QuorumEntry { - version: 1, - llmq_type: LLMQType::LlmqtypeTest, - quorum_hash, - quorum_index: None, - signers: vec![true; 3], - valid_members: vec![true; 3], - quorum_public_key: BLSPublicKey::from([7; 48]), - quorum_vvec_hash: QuorumVVecHash::all_zeros(), - threshold_sig: BLSSignature::from([1; 96]), - all_commitment_aggregated_signature: BLSSignature::from([1; 96]), - }], - quorums_chainlock_signatures: vec![QuorumCLSigObject { - signature: BLSSignature::from([1; 96]), - index_set: vec![0], - }], - ..MnListDiff::dummy(0x00, 0xAA) - }; - let mut storage = open_storage(&dir, &[(0x00, 0), (0xAA, 100)]).await; - storage.store_diff(100, &diff).await.unwrap(); - - let engine = storage.load_engine().await; - - let status = - &engine.masternode_lists[&100].quorums[&LLMQType::LlmqtypeTest][&quorum_hash].verified; - assert_ne!( - *status, - LLMQEntryVerificationStatus::Skipped( - LLMQEntryVerificationSkipStatus::NotMarkedForVerification - ), - "the replay never tried to verify it" - ); + let restored = reopened.engine(); + let restored = restored.read().await; + assert!(!restored.rotated_quorums_per_cycle().is_empty(), "the fixture holds cycles"); + assert_eq!(restored.snapshot().unwrap(), expected); } - /// A reorg truncates the headers above 100, but the diff stored for the - /// orphaned block at 200 stays on disk until a new one replaces it. #[tokio::test] - async fn replay_skips_a_message_whose_block_left_the_chain() { + async fn an_empty_storage_opens_with_an_empty_engine() { let dir = TempDir::new().unwrap(); - let mut storage = open_storage(&dir, &[(0x00, 0), (0xAA, 100)]).await; - storage.store_diff(100, &MnListDiff::dummy(0x00, 0xAA)).await.unwrap(); - storage.store_diff(200, &MnListDiff::dummy(0xAA, 0xBB)).await.unwrap(); - - let engine = storage.load_engine().await; + let storage = open_storage(&dir, engine_with_lists(&[]).headers()).await; - assert_eq!(engine.masternode_lists.keys().copied().collect::>(), vec![100]); + assert!(list_heights(&storage).await.is_empty()); } #[tokio::test] - async fn replay_skips_an_orphan_and_keeps_going() { + async fn persisting_replaces_the_previous_snapshot() { let dir = TempDir::new().unwrap(); - let mut storage = - open_storage(&dir, &[(0x00, 0), (0xAA, 100), (0xBB, 150), (0xDD, 120)]).await; + let heights = [(0xAA, 100), (0xBB, 200)]; + let headers = engine_with_lists(&heights).headers(); + let mut storage = open_storage(&dir, Arc::clone(&headers)).await; - storage.store_diff(100, &MnListDiff::dummy(0x00, 0xAA)).await.unwrap(); - storage.store_diff(120, &MnListDiff::dummy(0xEE, 0xDD)).await.unwrap(); - storage.store_diff(150, &MnListDiff::dummy(0xAA, 0xBB)).await.unwrap(); + persist(&mut storage, engine_with_lists(&heights[..1])).await; + persist(&mut storage, engine_with_lists(&heights)).await; - let engine = storage.load_engine().await; - - assert!( - !engine.masternode_lists.contains_key(&120), - "the orphan has no base to apply on and is left to the network" - ); - assert!(engine.masternode_lists.contains_key(&150), "the diff after it still applies"); + let files = std::fs::read_dir(dir.path().join("masternodes")).unwrap().count(); + assert_eq!(files, 1); + assert_eq!(list_heights(&open_storage(&dir, headers).await).await, vec![100, 200]); } #[tokio::test] - async fn storing_the_same_height_twice_keeps_the_newer_message() { + async fn an_engine_without_lists_is_not_persisted() { let dir = TempDir::new().unwrap(); - let mut storage = open_storage(&dir, &[]).await; + let mut storage = open_storage(&dir, engine_with_lists(&[]).headers()).await; - storage.store_diff(100, &MnListDiff::dummy(0x00, 0xAA)).await.unwrap(); - storage.store_diff(100, &MnListDiff::dummy(0x00, 0xBB)).await.unwrap(); + storage.persist().await.unwrap(); - assert_eq!(storage.diffs.len(), 1, "one file per height"); - let stored: MnListDiff = - PersistentMasternodeStorage::::read_message(&storage.diffs[&100]) - .await - .expect("read back"); - assert_eq!(stored.block_hash, hash(0xBB), "the second write wins"); + assert!(!snapshot_file(&dir).exists()); } - /// A QRInfo chaining `start`, `start + 1`, `start + 2`, `start + 3` and - /// `tip`, applied on top of the list at `start`. - fn chained_qr_info(start: u8, tip: u8) -> QRInfo { - QRInfo { - mn_list_diff_at_h_minus_3c: MnListDiff::dummy_empty(start, start), - mn_list_diff_at_h_minus_2c: MnListDiff::dummy_empty(start, start + 1), - mn_list_diff_at_h_minus_c: MnListDiff::dummy_empty(start + 1, start + 2), - mn_list_diff_h: MnListDiff::dummy_empty(start + 2, start + 3), - mn_list_diff_tip: MnListDiff::dummy_empty(start + 3, tip), - ..QRInfo::dummy(tip) - } - } - - /// The QRInfo stored at `height`, `None` when the file does not decode. - async fn stored_qr_info( - storage: &PersistentMasternodeStorage, - height: CoreBlockHeight, - ) -> Option { - PersistentMasternodeStorage::::read_message(&storage.qr_infos[&height]) - .await - .ok() - } - - #[test_case(chained_qr_info(0xA2, 0xE0), false, false; "keeps the stored one it builds on")] - #[test_case(chained_qr_info(0x10, 0xE0), false, true; "replaces it when unrelated")] - #[test_case(chained_qr_info(0xA2, 0xEF), false, true; "replaces it for another tip after a reorg")] - #[test_case(chained_qr_info(0xA2, 0xE0), true, true; "replaces a stored file that does not decode")] + #[test_case(200, &[100, 200]; "at the newest list keeps it")] + #[test_case(150, &[100]; "between the lists drops the newer one")] + #[test_case(50, &[]; "below every list drops the snapshot")] #[tokio::test] - async fn a_same_tip_qr_info_replaces_the_stored_one_unless_built_on_it( - incoming: QRInfo, - corrupt_stored: bool, - replaces: bool, - ) { - let height = 100; - let stored = chained_qr_info(0xA0, 0xE0); + async fn truncating_stores_what_the_engine_keeps(height: u32, kept: &[u32]) { let dir = TempDir::new().unwrap(); - let mut storage = open_storage(&dir, &[]).await; - storage.store_qr_info(height, &stored).await.unwrap(); - if corrupt_stored { - tokio::fs::write(&storage.qr_infos[&height], b"not a qrinfo").await.unwrap(); - } + let heights = [(0xAA, 100), (0xBB, 200)]; + let headers = engine_with_lists(&heights).headers(); + let mut storage = open_storage(&dir, Arc::clone(&headers)).await; + persist(&mut storage, engine_with_lists(&heights)).await; - storage.store_qr_info(height, &incoming).await.unwrap(); + storage.truncate_above(height).await.unwrap(); - let expected = if replaces { - incoming - } else { - stored - }; - assert_eq!(stored_qr_info(&storage, height).await, Some(expected)); + assert_eq!(list_heights(&storage).await, kept); + assert_eq!(snapshot_file(&dir).exists(), !kept.is_empty()); + assert_eq!(list_heights(&open_storage(&dir, headers).await).await, kept); } #[tokio::test] - async fn replay_survives_unreadable_files_and_ignores_foreign_names() { + async fn a_snapshot_whose_block_left_the_chain_is_not_restored() { let dir = TempDir::new().unwrap(); - let folder = dir.path().join(PersistentMasternodeStorage::::FOLDER_NAME); - tokio::fs::create_dir_all(&folder).await.unwrap(); + let heights = [(0xAA, 100), (0xBB, 200)]; + let mut chain = engine_with_lists(&heights); + persist(&mut open_storage(&dir, chain.headers()).await, engine_with_lists(&heights)).await; - { - let mut storage = open_storage(&dir, &[(0x00, 0), (0xAA, 100)]).await; - storage.store_diff(100, &MnListDiff::dummy(0x00, 0xAA)).await.unwrap(); - } + chain.forget_block(&hash(0xBB)); - tokio::fs::write(folder.join("diff_50.dat"), b"not a diff").await.unwrap(); - tokio::fs::write(folder.join("diff_abc.dat"), b"x").await.unwrap(); - - let storage = open_storage(&dir, &[(0x00, 0), (0xAA, 100)]).await; - assert_eq!( - storage.diffs.keys().copied().collect::>(), - vec![50, 100], - "only well-formed names are indexed" - ); - assert!(storage.qr_infos.is_empty()); - - let engine = storage.load_engine().await; - assert!( - engine.masternode_lists.contains_key(&100), - "the readable message still rebuilds its list" - ); + assert!(list_heights(&open_storage(&dir, chain.headers()).await).await.is_empty()); } - /// A Platform quorum mined at 1 and retired at 2, then a diff per block up to - /// past the retention window, written without fsync to keep the setup fast. + #[test_case(&[]; "empty")] + #[test_case(&[0xFF, 0xFF]; "undecodable")] #[tokio::test] - async fn a_lookup_rebuilds_a_quorum_only_lists_out_of_the_window_held() { - use dashcore::bls_sig_utils::{BLSPublicKey, BLSSignature}; - use dashcore::hash_types::QuorumVVecHash; - use dashcore::network::message_sml::{DeletedQuorum, QuorumCLSigObject}; - use dashcore::transaction::special_transaction::quorum_commitment::QuorumEntry; - type S = PersistentMasternodeStorage; - - let tip = 2_500; - let llmq_type = Network::Regtest.platform_type(); - let quorum_hash = QuorumHash::from_byte_array([0xAB; 32]); - let mined = MnListDiff { - block_hash: BlockHash::dummy(1), - new_quorums: vec![QuorumEntry { - version: 1, - llmq_type, - quorum_hash, - quorum_index: None, - signers: vec![true; 3], - valid_members: vec![true; 3], - quorum_public_key: BLSPublicKey::from([7; 48]), - quorum_vvec_hash: QuorumVVecHash::all_zeros(), - threshold_sig: BLSSignature::from([1; 96]), - all_commitment_aggregated_signature: BLSSignature::from([1; 96]), - }], - quorums_chainlock_signatures: vec![QuorumCLSigObject { - signature: BLSSignature::from([1; 96]), - index_set: vec![0], - }], - ..MnListDiff::dummy(0x00, 0x01) - }; - let retired = MnListDiff { - deleted_quorums: vec![DeletedQuorum { - llmq_type, - quorum_hash, - }], - ..MnListDiff::dummy_between(1, 2) - }; - + async fn an_unreadable_snapshot_is_not_restored(bytes: &[u8]) { let dir = TempDir::new().unwrap(); - let folder = dir.path().join(S::FOLDER_NAME); - std::fs::create_dir_all(&folder).unwrap(); - let write = |height, diff: &MnListDiff| { - std::fs::write(folder.join(S::file_name(S::DIFF_PREFIX, height)), serialize(diff)) - .unwrap() - }; - write(1, &mined); - write(2, &retired); - for height in 3..=tip { - write(height, &MnListDiff::dummy_between(height - 1, height)); - } - let heights = (1..=tip).map(|height| (BlockHash::dummy(height), height)).collect(); - let storage = S::open( - dir.path(), - Arc::new(RwLock::new(MockHeaderStorage(heights))), - Network::Regtest, - ) - .await - .unwrap(); - - let engine = storage.load_engine().await; - assert!(engine - .quorum_entry_for_hash_at_or_before_height(llmq_type, quorum_hash, 100) - .is_none()); - - let log = storage.message_log(); - let quorum = log.quorum_entry_at_or_before(llmq_type, quorum_hash, 100).await; - assert_eq!(quorum.map(|quorum| quorum.quorum_entry.quorum_hash), Some(quorum_hash)); - let unknown = QuorumHash::from_byte_array([0xCD; 32]); - assert!(log.quorum_entry_at_or_before(llmq_type, unknown, 100).await.is_none()); - } + std::fs::create_dir_all(dir.path().join("masternodes")).unwrap(); + std::fs::write(snapshot_file(&dir), bytes).unwrap(); - #[test] - fn qr_info_bases_are_what_its_chain_starts_from() { - let mut qr_info = QRInfo::dummy(0x00); - qr_info.mn_list_diff_at_h_minus_3c = MnListDiff::dummy_empty(0xA0, 0xA0); - qr_info.mn_list_diff_at_h_minus_2c = MnListDiff::dummy_empty(0xA0, 0xB0); - qr_info.mn_list_diff_at_h_minus_c = MnListDiff::dummy_empty(0xB0, 0xC0); - qr_info.mn_list_diff_h = MnListDiff::dummy_empty(0xC0, 0xD0); - qr_info.mn_list_diff_tip = MnListDiff::dummy_empty(0xD0, 0xE0); - - assert_eq!( - Message::QrInfo(Box::new(qr_info)).base_hashes(), - vec![hash(0xA0)], - "the empty diff at the start of the chain needs its list to exist already" - ); + let storage = open_storage(&dir, engine_with_lists(&[]).headers()).await; + + assert!(list_heights(&storage).await.is_empty()); } - /// Mainnet headers start at a checkpoint, so the genesis a first QRInfo - /// is built from has no height. A work-block diff requested from a list - /// that QRInfo built still has to replay after it. #[tokio::test] - async fn replay_applies_a_diff_on_a_list_of_a_qr_info_built_from_genesis() { - use crate::sml_engine::test_support::fixture_heights; - use dashcore::network::constants::NetworkExt; - - let qr_info: QRInfo = bincode::decode_from_slice( - include_bytes!("../../../dash/tests/data/test_DML_diffs/qrinfo_2518986.dat"), - bincode::config::standard(), - ) - .unwrap() - .0; - - let mut heights = fixture_heights(include_bytes!( - "../../../dash/tests/data/test_DML_diffs/block_container_2518986.dat" - )); - heights.remove(&Network::Mainnet.known_genesis_block_hash().unwrap()); - let tip_height = heights[&qr_info.mn_list_diff_tip.block_hash]; - let base = qr_info.mn_list_diff_at_h_minus_c.block_hash; - let work_block_height = heights[&base] + 24; - heights.insert(hash(0xEE), work_block_height); - + async fn clearing_empties_the_shared_engine_in_place() { let dir = TempDir::new().unwrap(); - let mut storage = PersistentMasternodeStorage::open( - dir.path(), - Arc::new(RwLock::new(MockHeaderStorage(heights))), - Network::Mainnet, - ) - .await - .unwrap(); - storage.store_qr_info(tip_height, &qr_info).await.unwrap(); - let diff = MnListDiff { - base_block_hash: base, - ..MnListDiff::dummy(0x00, 0xEE) - }; - storage.store_diff(work_block_height, &diff).await.unwrap(); + let heights = [(0xAA, 100)]; + let headers = engine_with_lists(&heights).headers(); + let mut storage = open_storage(&dir, Arc::clone(&headers)).await; + persist(&mut storage, engine_with_lists(&heights)).await; + let shared = storage.engine(); - let engine = storage.load_engine().await; + storage.clear(headers).await; - assert!(engine.masternode_lists.contains_key(&tip_height), "the QRInfo replays"); - assert!( - engine.masternode_lists.contains_key(&work_block_height), - "the diff on the QRInfo's h-c list replays after the QRInfo" - ); + assert!(shared.read().await.masternode_lists.is_empty()); } } diff --git a/dash-spv/src/storage/migrator/mod.rs b/dash-spv/src/storage/migrator/mod.rs index fc10ecd67..9ef8659dc 100644 --- a/dash-spv/src/storage/migrator/mod.rs +++ b/dash-spv/src/storage/migrator/mod.rs @@ -1,5 +1,6 @@ mod v1; mod v2; +mod v3; use std::io; use std::path::{Path, PathBuf}; @@ -11,8 +12,9 @@ use crate::error::{StorageError, StorageResult}; use crate::storage::io::atomic_write; use v1::V1Migrator; use v2::V2Migrator; +use v3::V3Migrator; -pub const CURRENT_VERSION: u32 = 2; +pub const CURRENT_VERSION: u32 = 3; #[async_trait] trait Migrator { @@ -118,7 +120,8 @@ async fn migration_loop(storage_path: &Path) -> Result<(), MigratorError> { while version < CURRENT_VERSION { tracing::info!("Migrating storage from version {} to {}", version, version + 1); - const MIGRATORS: &[&dyn Migrator; CURRENT_VERSION as usize] = &[&V1Migrator, &V2Migrator]; + const MIGRATORS: &[&dyn Migrator; CURRENT_VERSION as usize] = + &[&V1Migrator, &V2Migrator, &V3Migrator]; MIGRATORS[version as usize].apply_migration(storage_path).await?; diff --git a/dash-spv/src/storage/migrator/v3.rs b/dash-spv/src/storage/migrator/v3.rs new file mode 100644 index 000000000..f650e4690 --- /dev/null +++ b/dash-spv/src/storage/migrator/v3.rs @@ -0,0 +1,23 @@ +use std::io; +use std::path::Path; + +use async_trait::async_trait; + +use super::{Migrator, MigratorError}; + +const MASTERNODES_DIR: &str = "masternodes"; + +/// Drops the masternode messages earlier versions stored. The storage now +/// keeps snapshots of the masternode list engine instead, so the first start +/// after the migration syncs the masternode lists from the network again. +pub struct V3Migrator; + +#[async_trait] +impl Migrator for V3Migrator { + async fn apply_migration(&self, storage_path: &Path) -> Result<(), MigratorError> { + match tokio::fs::remove_dir_all(storage_path.join(MASTERNODES_DIR)).await { + Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()), + result => Ok(result?), + } + } +} diff --git a/dash-spv/src/storage/mod.rs b/dash-spv/src/storage/mod.rs index f9cbbf7af..fdd1ddede 100644 --- a/dash-spv/src/storage/mod.rs +++ b/dash-spv/src/storage/mod.rs @@ -20,10 +20,7 @@ use crate::types::{HashedBlock, HashedBlockHeader}; use crate::ClientConfig; use async_trait::async_trait; use dashcore::hash_types::FilterHeader; -use dashcore::network::message_qrinfo::QRInfo; -use dashcore::network::message_sml::MnListDiff; use dashcore::prelude::CoreBlockHeight; -use dashcore::Network; use serde::de::DeserializeOwned; use std::ops::Range; use std::path::{Path, PathBuf}; @@ -66,7 +63,7 @@ pub trait StorageManager: async fn clear(&mut self) -> StorageResult<()>; /// Drops the block headers, filter headers, filters, blocks and masternode - /// messages above `height`. + /// engine snapshots above `height`. async fn truncate_above(&mut self, height: u32) -> StorageResult<()>; /// Starts the background tasks, restarting them after a [`Self::stop`]. @@ -100,7 +97,6 @@ pub trait StorageManager: /// until [`StorageManager::stop`] or drop. pub struct DiskStorageManager { storage_path: PathBuf, - network: Network, block_headers: Arc>, filter_headers: Arc>, @@ -155,7 +151,6 @@ impl DiskStorageManager { Ok(Self { storage_path: storage_path.clone(), - network: config.network, filter_headers: Arc::new(RwLock::new( PersistentFilterHeaderStorage::open(&storage_path).await?, @@ -169,7 +164,7 @@ impl DiskStorageManager { Arc::clone(&block_headers), config.network, ) - .await?, + .await, )), block_headers, @@ -233,6 +228,7 @@ impl DiskStorageManager { let _ = self.filters.write().await.persist(storage_path).await; let _ = self.blocks.write().await.persist(storage_path).await; let _ = self.metadata.write().await.persist(storage_path).await; + let _ = self.masternodes.write().await.persist().await; } } @@ -287,14 +283,7 @@ impl StorageManager for DiskStorageManager { self.filters = Arc::new(RwLock::new(PersistentFilterStorage::open(storage_path).await?)); self.blocks = Arc::new(RwLock::new(PersistentBlockStorage::open(storage_path).await?)); self.metadata = Arc::new(RwLock::new(PersistentMetadataStorage::open(storage_path).await?)); - self.masternodes = Arc::new(RwLock::new( - PersistentMasternodeStorage::open( - storage_path, - Arc::clone(&self.block_headers), - self.network, - ) - .await?, - )); + self.masternodes.write().await.clear(Arc::clone(&self.block_headers)).await; if worker_was_running { self.start_worker().await; @@ -483,22 +472,6 @@ impl metadata::MetadataStorage for DiskStorageManager { #[async_trait] impl masternode::MasternodeStorage for DiskStorageManager { - async fn store_diff( - &mut self, - height: CoreBlockHeight, - diff: &MnListDiff, - ) -> StorageResult<()> { - self.masternodes.write().await.store_diff(height, diff).await - } - - async fn store_qr_info( - &mut self, - height: CoreBlockHeight, - qr_info: &QRInfo, - ) -> StorageResult<()> { - self.masternodes.write().await.store_qr_info(height, qr_info).await - } - async fn truncate_above(&mut self, height: CoreBlockHeight) -> StorageResult<()> { self.masternodes.write().await.truncate_above(height).await } diff --git a/dash-spv/src/sync/masternodes/manager.rs b/dash-spv/src/sync/masternodes/manager.rs index 74392a594..aa4415217 100644 --- a/dash-spv/src/sync/masternodes/manager.rs +++ b/dash-spv/src/sync/masternodes/manager.rs @@ -12,12 +12,11 @@ use dashcore::sml::llmq_type::network::NetworkLLMQExt; use tokio::sync::RwLock; use super::pipeline::MnListDiffPipeline; -use crate::error::{StorageResult, SyncError, SyncResult}; +use crate::error::{SyncError, SyncResult}; use crate::network::RequestSender; -use crate::storage::{BlockHeaderStorage, MasternodeStorage, PersistentMasternodeStorage}; +use crate::storage::{BlockHeaderStorage, PersistentMasternodeStorage}; use crate::sync::{MasternodesProgress, SyncEvent, SyncManager, SyncState}; use dashcore::network::message_qrinfo::QRInfo; -use dashcore::network::message_sml::MnListDiff; use dashcore::BlockHash; use std::collections::BTreeSet; @@ -260,8 +259,8 @@ pub struct MasternodesManager { network: dashcore::Network, /// Sync state tracking. pub(super) sync_state: MasternodeSyncState, - /// Where the messages that built the engine are kept for the next start. - message_storage: Option>>>, + /// Where the engine is kept for the next start. + engine_storage: Option>>>, } impl MasternodesManager { @@ -270,7 +269,7 @@ impl MasternodesManager { header_storage: Arc>, engine: Arc>>, network: dashcore::Network, - message_storage: Option>>>, + engine_storage: Option>>>, ) -> Self { // Recover sync state from the engine's stored masternode lists so that a // restart can resume from where the previous run left off. @@ -295,28 +294,20 @@ impl MasternodesManager { engine, network, sync_state, - message_storage, + engine_storage, } } - /// Messages are stored only until the sync reaches the tip. Once synced, - /// nothing is, and the next start catches up from the last stored message. - fn storage_while_syncing(&self) -> Option<&Arc>>> { - self.message_storage.as_ref().filter(|_| self.state() != SyncState::Synced) - } - - pub(super) async fn store_diff(&self, height: u32, diff: &MnListDiff) -> StorageResult<()> { - let Some(storage) = self.storage_while_syncing() else { - return Ok(()); + /// Stores the engine for the next start. A failure is logged and only + /// costs a sync from the network on the next start. + async fn store_engine(&self) { + let Some(storage) = &self.engine_storage else { + return; }; - storage.write().await.store_diff(height, diff).await - } - pub(super) async fn store_qr_info(&self, height: u32, qr_info: &QRInfo) -> StorageResult<()> { - let Some(storage) = self.storage_while_syncing() else { - return Ok(()); - }; - storage.write().await.store_qr_info(height, qr_info).await + if let Err(e) = storage.write().await.persist().await { + tracing::warn!("Could not store the masternode engine, a restart syncs it again: {e}"); + } } /// Decide which [`PipelineMode`] to use when a new header lands at `tip_height` @@ -614,6 +605,9 @@ impl MasternodesManager { drop(engine); + // Once per QRInfo, so at each start and rotation cycle, not per block. + self.store_engine().await; + if is_initial_sync { self.set_state(SyncState::Synced); tracing::info!("Masternode sync complete at height {}", self.progress.current_height()); diff --git a/dash-spv/src/sync/masternodes/sync_manager.rs b/dash-spv/src/sync/masternodes/sync_manager.rs index 8272c1e94..1677781ae 100644 --- a/dash-spv/src/sync/masternodes/sync_manager.rs +++ b/dash-spv/src/sync/masternodes/sync_manager.rs @@ -120,17 +120,8 @@ impl SyncManager for MasternodesManager { }; 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 { - if let Err(e) = self.store_qr_info(height, qr_info).await { - tracing::warn!( - "Could not store QRInfo at {height}, a restart syncs it again: {e}" - ); - } - } - tracing::info!( "QRInfo processed: stored_cycle_height={:?}, rotated_quorum_count={}/{}, fully_verified_count={}", qr_info_result.stored_cycle_height, @@ -233,15 +224,6 @@ impl SyncManager for MasternodesManager { }; drop(engine); - if apply_ok { - if let Err(e) = self.store_diff(target_height, diff).await { - tracing::warn!( - "Could not store MnListDiff at {target_height}, a restart syncs it \ - again: {e}" - ); - } - } - self.progress.add_diffs_processed(1); self.sync_state.mnlistdiff_pipeline.receive(diff); if !work_block_lists.is_empty() { @@ -574,7 +556,7 @@ mod tests { }; use crate::sync::{MasternodesManager, SyncEvent, SyncManager, SyncState}; use crate::types::HashedBlockHeader; - use crate::{ClientConfig, SyncError}; + use crate::SyncError; use dashcore::block::Header; use dashcore::bls_sig_utils::{BLSPublicKey, BLSSignature}; use dashcore::network::message::NetworkMessage; @@ -592,7 +574,6 @@ mod tests { use std::sync::Arc; use std::time::Duration; use std::time::Instant; - use test_case::test_case; use tokio::sync::{mpsc, RwLock}; /// The QRInfo retry budget escalates: a tight first timeout fails over @@ -756,44 +737,12 @@ mod tests { (manager, requests, rx, tip_hash) } - /// Diffs are stored only until the sync reaches the tip. Once synced, - /// nothing is, and the next start catches up from the last stored message. - #[test_case(SyncState::Syncing, true; "while syncing")] - #[test_case(SyncState::Synced, false; "once synced")] - #[tokio::test] - async fn a_diff_is_stored_only_until_the_sync_reaches_the_tip(state: SyncState, stored: bool) { - let dir = tempfile::TempDir::new().unwrap(); - let storage = - DiskStorageManager::new(&ClientConfig::testnet().with_storage_path(dir.path())) - .await - .unwrap(); - let (mut manager, engine, requests, _rx, hashes) = - manager_after_a_tip_update_request(&storage, true, |hashes| { - vec![MasternodeList::empty(hashes[TIP as usize], TIP)] - }) - .await; - manager.set_state(state); - - let diff = MnListDiff { - base_block_hash: hashes[TIP as usize], - block_hash: hashes[TIP as usize + 1], - ..MnListDiff::dummy_between(0, 0) - }; - deliver(&mut manager, &requests, diff).await; - - assert!(engine.read().await.masternode_lists.contains_key(&(TIP + 1))); - let file = dir.path().join("masternodes").join(format!("diff_{}.dat", TIP + 1)); - assert_eq!(file.exists(), stored); - } - /// A synced manager over `storage`, whose header chain runs to `TIP + 1` /// and whose engine holds the lists `lists` builds from the block hashes, - /// once it sent the tip update from `TIP` to `TIP + 1`. With - /// `store_diffs`, it stores diffs in `storage`. Returns the manager, its + /// once it sent the tip update from `TIP` to `TIP + 1`. Returns the manager, its /// engine, the request channel, and the block hash at each height. async fn manager_after_a_tip_update_request( storage: &DiskStorageManager, - store_diffs: bool, lists: impl FnOnce(&[BlockHash]) -> Vec, ) -> ( MasternodesManager, @@ -816,13 +765,9 @@ mod tests { engine.masternode_lists.insert(list.known_height, list); } let engine = Arc::new(RwLock::new(engine)); - let mut manager = MasternodesManager::new( - block_headers, - Arc::clone(&engine), - Network::Regtest, - store_diffs.then(|| storage.masternodes()), - ) - .await; + let mut manager = + MasternodesManager::new(block_headers, Arc::clone(&engine), Network::Regtest, None) + .await; manager.set_state(SyncState::Synced); manager.progress.update_block_header_tip_height(TIP + 1); let (tx, mut rx) = mpsc::unbounded_channel(); @@ -848,7 +793,7 @@ mod tests { ) { let storage = DiskStorageManager::with_temp_dir().await.unwrap(); let (manager, engine, requests, rx, hashes) = - manager_after_a_tip_update_request(&storage, false, |hashes| { + manager_after_a_tip_update_request(&storage, |hashes| { let old = Arc::new(test_quorum(hashes[OLD_MINED as usize]).into()); let quorums = BTreeMap::from([( LLMQType::LlmqtypeTest, diff --git a/dash-spv/tests/dashd_masternode/helpers.rs b/dash-spv/tests/dashd_masternode/helpers.rs index d01a43bf7..3b78158d6 100644 --- a/dash-spv/tests/dashd_masternode/helpers.rs +++ b/dash-spv/tests/dashd_masternode/helpers.rs @@ -56,7 +56,7 @@ pub(super) const EXPECTED_STORAGE: &[(&str, &str)] = &[ ("metadata", "sync checkpoints"), ("version.json", "the storage version system"), ("peers", "peer set and reputations"), - ("masternodes", "the masternode messages this session stored"), + ("masternodes", "the masternode engine snapshot this session stored"), ]; pub(super) fn assert_storage_persisted(snapshot: &BTreeMap, what: &str) { diff --git a/dash-spv/tests/dashd_masternode/tests_sync.rs b/dash-spv/tests/dashd_masternode/tests_sync.rs index ea7bdc07d..c005d3642 100644 --- a/dash-spv/tests/dashd_masternode/tests_sync.rs +++ b/dash-spv/tests/dashd_masternode/tests_sync.rs @@ -156,8 +156,8 @@ async fn test_masternode_list_sync_with_restart() { let mut client_handle = create_client(&config, Arc::clone(&wallet)).await; // Read before the run loop is spawned, so nothing has come off the network yet: - // this is what replaying the stored messages produced on its own. - let replayed_tip = { + // this is what the stored engine restored on its own. + let restored_tip = { let engine = client_handle.engine.read().await; engine .masternode_lists @@ -166,7 +166,7 @@ async fn test_masternode_list_sync_with_restart() { .map(|(height, list)| (*height, list.block_hash, list.masternodes.len())) }; assert_eq!( - replayed_tip, first_tip, + restored_tip, first_tip, "startup must rebuild the first session's masternode list from storage, \ not default and let a fresh dashd sync cover for it" ); @@ -192,10 +192,10 @@ async fn test_masternode_list_sync_with_restart() { client_handle.stop().await; drop(client_handle); - // The second session requested a QRInfo for the same tip. Storing it in place - // of the first one, which built the list it is based on, broke the next replay. + // The second session stored the engine again at the same tip, in place of + // the first session's snapshot. let client_handle = create_client(&config, Arc::clone(&wallet)).await; - let replayed_tip = { + let restored_tip = { let engine = client_handle.engine.read().await; engine .masternode_lists @@ -203,7 +203,7 @@ async fn test_masternode_list_sync_with_restart() { .next_back() .map(|(height, list)| (*height, list.block_hash, list.masternodes.len())) }; - assert_eq!(replayed_tip, first_tip, "a second restart must still replay the stored list"); + assert_eq!(restored_tip, first_tip, "a second restart must still restore the stored list"); } /// Sync to pre-generated height, generate new blocks, verify incremental update. From ebe1fa3cba03b3fb787f74de7ab9b4c332a9efc4 Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Mon, 26 Oct 2026 10:11:43 -0700 Subject: [PATCH 2/2] fix(dash-spv): truncate the engine even when storage truncation fails, reject trailing snapshot bytes Co-Authored-By: Claude Opus 5.5 --- dash-spv/src/client/lifecycle.rs | 5 ++++- dash-spv/src/sml_engine.rs | 7 ++++++- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/dash-spv/src/client/lifecycle.rs b/dash-spv/src/client/lifecycle.rs index 8999c4c58..eed1be13c 100644 --- a/dash-spv/src/client/lifecycle.rs +++ b/dash-spv/src/client/lifecycle.rs @@ -283,10 +283,13 @@ impl DashSpvClient { height: u32, ) -> std::result::Result<(), Box> { self.wallet.write().await.truncate_above(height)?; - let truncated = StorageManager::truncate_above(&mut *self.storage.lock().await, height).await; + if let Some(engine) = &self.masternode_engine { + engine.write().await.truncate_above(height).await; + } + Ok(truncated?) } diff --git a/dash-spv/src/sml_engine.rs b/dash-spv/src/sml_engine.rs index d4114473b..3bf7a6231 100644 --- a/dash-spv/src/sml_engine.rs +++ b/dash-spv/src/sml_engine.rs @@ -258,8 +258,13 @@ impl MasternodeListEngine { headers: Arc>, bytes: &[u8], ) -> Result { - let ((masternode_lists, rotated_quorums_per_cycle), _) = + let ((masternode_lists, rotated_quorums_per_cycle), read) = bincode::serde::decode_from_slice(bytes, bincode::config::standard())?; + + if read != bytes.len() { + return Err(bincode::error::DecodeError::Other("trailing bytes after the snapshot")); + } + Ok(Self { headers, masternode_lists,