From 5d737ef0d11be5a5dee454385dcc02461f00483f Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Thu, 27 Aug 2026 14:27:57 +0000 Subject: [PATCH 1/7] test(dash-spv): assert the restart test's storage, not just its progress MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `test_masternode_list_sync_with_restart` compared masternode sync progress either side of a restart. A from-scratch network re-sync produces the same progress as a restored one, so the test passed while the list was being rebuilt from nothing every time (dashpay/rust-dashcore#988). It now looks at the disk. After the first session's clean shutdown every directory that session earned must hold a file, and across the restart no directory may disappear or lose files. Fails as written: the first session builds four masternodes and writes no `masternodestate/`, while `block_headers/`, `filter_headers/`, `metadata/` and `peers/` all persist through the same shutdown to the same directory — so the storage layer and the shutdown are ruled out as causes. `filters/` and `blocks/` are left out of the must-hold set on purpose: the client stops as soon as the masternode phase reports `Synced`, which is before the filter phase leaves `WaitForEvents`, so they are legitimately empty here. The no-shrink check still covers them. The engine is read before the shutdown and the count carried into the failure message, so the assertion cannot be satisfied by a session that synced nothing — which is the shape dashpay/rust-dashcore#954 produces. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_015NhHBDGiKfiGpy7FwooyfS --- dash-spv/tests/dashd_masternode/helpers.rs | 91 +++++++++++++++++++ dash-spv/tests/dashd_masternode/tests_sync.rs | 30 +++++- 2 files changed, 119 insertions(+), 2 deletions(-) diff --git a/dash-spv/tests/dashd_masternode/helpers.rs b/dash-spv/tests/dashd_masternode/helpers.rs index aa27b7a06..8166df9bb 100644 --- a/dash-spv/tests/dashd_masternode/helpers.rs +++ b/dash-spv/tests/dashd_masternode/helpers.rs @@ -1,3 +1,6 @@ +use std::collections::BTreeMap; +use std::path::Path; + use dash_spv::sync::{MasternodesProgress, SyncEvent, SyncProgress, SyncState}; use dashcore::ephemerealdata::instant_lock::InstantLock; use dashcore::sml::llmq_entry_verification::LLMQEntryVerificationStatus; @@ -14,6 +17,94 @@ use super::setup::{TestContext, SYNC_TIMEOUT}; /// Mine a DKG cycle and wait for the SPV to surface a `MasternodeStateUpdated` /// event above `baseline_height`. +/// Files held under each immediate subdirectory of the storage root, keyed by +/// directory name. +/// +/// A sync writes into these and never removes a whole class of state, so across +/// a restart every directory must still be there and hold at least as much — +/// see [`assert_storage_did_not_shrink`]. +pub(super) fn storage_snapshot(root: &Path) -> BTreeMap { + let mut counts = BTreeMap::new(); + let Ok(entries) = std::fs::read_dir(root) else { + return counts; + }; + for entry in entries.flatten() { + if !entry.path().is_dir() { + continue; + } + let files = walkdir_count(&entry.path()); + counts.insert(entry.file_name().to_string_lossy().into_owned(), files); + } + counts +} + +fn walkdir_count(dir: &Path) -> usize { + let Ok(entries) = std::fs::read_dir(dir) else { + return 0; + }; + entries + .flatten() + .map(|e| { + let path = e.path(); + if path.is_dir() { + walkdir_count(&path) + } else { + 1 + } + }) + .sum() +} + +/// Directories that must hold state once this test's first session has run, and +/// why. `filters` and `blocks` are deliberately absent: the client is stopped +/// as soon as the masternode phase reports `Synced`, which is before the filter +/// phase leaves `WaitForEvents`, so those stay legitimately empty here. +pub(super) const EXPECTED_STORAGE: &[(&str, &str)] = &[ + ("block_headers", "headers synced to the tip"), + ("filter_headers", "filter headers synced to the tip"), + ("metadata", "sync checkpoints"), + ("peers", "peer set and reputations"), + ("masternodestate", "the masternode list this session built"), +]; + +/// Assert every directory in [`EXPECTED_STORAGE`] exists and holds at least one +/// file, reporting all of them at once rather than the first to fail. +pub(super) fn assert_storage_persisted(snapshot: &BTreeMap, what: &str) { + let missing: Vec = EXPECTED_STORAGE + .iter() + .filter(|(dir, _)| snapshot.get(*dir).is_none_or(|files| *files == 0)) + .map(|(dir, why)| format!(" {dir}/ — {why}")) + .collect(); + assert!( + missing.is_empty(), + "{what}: {} storage director{} empty or absent after a clean shutdown:\n{}\n\nstorage holds {snapshot:?}", + missing.len(), + if missing.len() == 1 { "y is" } else { "ies are" }, + missing.join("\n"), + ); +} + +/// Every directory present before a restart must still be present after, with +/// at least as many files. A directory that vanishes or shrinks means a restart +/// threw away state that the previous session had already earned. +pub(super) fn assert_storage_did_not_shrink( + before: &BTreeMap, + after: &BTreeMap, + what: &str, +) { + for (dir, before_count) in before { + match after.get(dir) { + None => panic!( + "{what}: storage directory {dir:?} disappeared across the restart\n before: {before:?}\n after: {after:?}" + ), + Some(after_count) if after_count < before_count => panic!( + "{what}: storage directory {dir:?} shrank across the restart, {before_count} -> {after_count}\n before: {before:?}\n after: {after:?}" + ), + Some(_) => {} + } + } +} + pub(super) async fn mine_dkg_cycle_and_wait( ctx: &mut TestContext, sync_event_receiver: &mut broadcast::Receiver, diff --git a/dash-spv/tests/dashd_masternode/tests_sync.rs b/dash-spv/tests/dashd_masternode/tests_sync.rs index 805e469ad..b38d28783 100644 --- a/dash-spv/tests/dashd_masternode/tests_sync.rs +++ b/dash-spv/tests/dashd_masternode/tests_sync.rs @@ -10,8 +10,9 @@ use dashcore::sml::llmq_entry_verification::LLMQEntryVerificationStatus; use dashcore::sml::llmq_type::LLMQType; use super::helpers::{ - assert_all_rotated_quorums_verified, wait_for_chainlock_height_at_least, - wait_for_masternode_sync, wait_for_mn_state_event, wait_for_mn_state_event_above, + assert_all_rotated_quorums_verified, assert_storage_did_not_shrink, assert_storage_persisted, + storage_snapshot, wait_for_chainlock_height_at_least, wait_for_masternode_sync, + wait_for_mn_state_event, wait_for_mn_state_event_above, wait_for_mn_state_with_stored_cycle_above, }; use super::setup::{ @@ -103,9 +104,29 @@ async fn test_masternode_list_sync_with_restart() { let first_mn_progress = wait_for_masternode_sync(&mut client_handle.progress_receiver, SYNC_TIMEOUT).await; let first_height = first_mn_progress.current_height(); + + // Control: the first session really built a list, so the persistence + // assertion below cannot be satisfied by a client that synced nothing. + let first_masternodes = { + let engine = client_handle.engine.read().await; + engine.masternode_lists.values().map(|list| list.masternodes.len()).max().unwrap_or(0) + }; + assert!( + first_masternodes > 0, + "the first session must have a masternode list before its persistence can be tested" + ); + client_handle.stop().await; drop(client_handle); + // What the first session earned and wrote down. A clean shutdown of a + // fully-synced client must leave every sync phase's state on disk. + let after_first = storage_snapshot(ctx.storage_path()); + assert_storage_persisted( + &after_first, + &format!("after a first session that built {first_masternodes} masternode(s)"), + ); + // Restart with same storage tracing::info!("=== Restarting with same storage ==="); let mut client_handle = create_and_start_client(&config, Arc::clone(&wallet)).await; @@ -123,6 +144,11 @@ async fn test_masternode_list_sync_with_restart() { "Should reach Synced state after restart" ); + // A restart re-syncs on top of what it restored; it never discards a whole + // class of state it already had. + let after_second = storage_snapshot(ctx.storage_path()); + assert_storage_did_not_shrink(&after_first, &after_second, "masternode restart"); + tracing::info!( "Restart verified: first_height={}, second_height={}", first_height, From f35c03cfebfc0b7218dc5f45b9fffa09d3094bcc Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Thu, 27 Aug 2026 14:43:43 +0000 Subject: [PATCH 2/7] fix(dash-spv): persist the masternode list and restore it at startup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `storage/masternode.rs` has had no callers outside `storage/` since the legacy sync engine was deleted, and `DashSpvClient::new` always built a fresh `MasternodeListEngine`. Every start therefore rebuilt the whole list from the network — a full QRInfo plus every MnListDiff — while headers, filters and ChainLocks resumed from disk. On mobile, where the host app restarts the client every minute or two, the rebuild rarely finishes, so a client can run with no masternode list at all despite having synced one in a previous session (dashpay/rust-dashcore#988). Both halves are wired here. `MasternodesManager` takes the state store and writes the engine wherever it reports `MasternodeStateUpdated` — the same condition that makes the new state worth keeping. `DashSpvClient::new` loads the state and seeds the engine, before the managers are built: `MasternodesManager::new` already recovers its resume point from the engine's stored lists, so a restore landing after it would be ignored. Both directions fail soft. An unwritten list costs a rebuild next start; a failed sync costs the list now. Likewise state that cannot be read is logged and rebuilt, which is exactly the old behaviour. `test_masternode_list_sync_with_restart` now passes, and the log shows why: `0 base hash(es)` on the first sync, `Restored masternode state from height 406`, then `1 base hash(es)` on the second — the delta, not a rebuild. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_015NhHBDGiKfiGpy7FwooyfS --- dash-spv/src/client/lifecycle.rs | 32 +++++++++++-- dash-spv/src/storage/mod.rs | 6 +++ dash-spv/src/sync/masternodes/manager.rs | 48 ++++++++++++++++++- dash-spv/src/sync/masternodes/sync_manager.rs | 10 ++-- 4 files changed, 86 insertions(+), 10 deletions(-) diff --git a/dash-spv/src/client/lifecycle.rs b/dash-spv/src/client/lifecycle.rs index 46e26f71c..d3b17d715 100644 --- a/dash-spv/src/client/lifecycle.rs +++ b/dash-spv/src/client/lifecycle.rs @@ -13,8 +13,9 @@ use crate::chain::checkpoints::CheckpointManager; use crate::error::{Result, SpvError}; use crate::network::NetworkManager; use crate::storage::{ - PersistentBlockHeaderStorage, PersistentBlockStorage, PersistentFilterHeaderStorage, - PersistentFilterStorage, PersistentMetadataStorage, StorageManager, + MasternodeStateStorage, PersistentBlockHeaderStorage, PersistentBlockStorage, + PersistentFilterHeaderStorage, PersistentFilterStorage, PersistentMetadataStorage, + StorageManager, }; use crate::sync::{ BlockHeadersManager, BlocksManager, ChainLockManager, FilterHeadersManager, FiltersManager, @@ -65,11 +66,31 @@ impl DashSpvClient match serde_json::from_slice(&state.engine_state) { + Ok(restored) => { + engine = restored; + tracing::info!( + "Restored masternode state from height {}", + state.last_height + ); + } + Err(e) => tracing::warn!( + "Could not read persisted masternode state, rebuilding: {}", + e + ), + }, + Ok(None) => tracing::debug!("No persisted masternode state"), + Err(e) => { + tracing::warn!("Could not load masternode state, rebuilding: {}", e) + } + } + Some(Arc::new(RwLock::new(engine))) } else { None } @@ -123,6 +144,7 @@ impl DashSpvClient Arc>; + + fn masternodestate(&self) -> Arc>; } /// Disk-based storage manager with segmented files and async background saving. @@ -282,6 +284,10 @@ impl StorageManager for DiskStorageManager { fn metadata(&self) -> Arc> { Arc::clone(&self.metadata) } + + fn masternodestate(&self) -> Arc> { + Arc::clone(&self.masternodestate) + } } #[async_trait] diff --git a/dash-spv/src/sync/masternodes/manager.rs b/dash-spv/src/sync/masternodes/manager.rs index 428673535..868a96c90 100644 --- a/dash-spv/src/sync/masternodes/manager.rs +++ b/dash-spv/src/sync/masternodes/manager.rs @@ -14,7 +14,9 @@ use tokio::sync::RwLock; use super::pipeline::MnListDiffPipeline; use crate::error::{SyncError, SyncResult}; use crate::network::RequestSender; -use crate::storage::BlockHeaderStorage; +use crate::storage::{ + BlockHeaderStorage, MasternodeState, MasternodeStateStorage, PersistentMasternodeStateStorage, +}; use crate::sync::{MasternodesProgress, SyncEvent, SyncManager, SyncState}; use dashcore::network::message_qrinfo::QRInfo; use dashcore::BlockHash; @@ -299,6 +301,8 @@ pub struct MasternodesManager { network: dashcore::Network, /// Sync state tracking. pub(super) sync_state: MasternodeSyncState, + /// `None` leaves the list in memory only. + pub(super) state_storage: Option>>, } impl MasternodesManager { @@ -307,6 +311,7 @@ impl MasternodesManager { header_storage: Arc>, engine: Arc>, network: dashcore::Network, + state_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. @@ -337,6 +342,38 @@ impl MasternodesManager { engine, network, sync_state, + state_storage, + } + } + + /// Best effort: an unwritten list costs a rebuild next start, a failed sync + /// costs the list now. + pub(super) async fn persist_engine(&self, height: u32) { + let Some(storage) = &self.state_storage else { + return; + }; + let engine_state = { + let engine = self.engine.read().await; + match serde_json::to_vec(&*engine) { + Ok(bytes) => bytes, + Err(e) => { + tracing::warn!("Could not serialize masternode engine at {height}: {e}"); + return; + } + } + }; + let state = MasternodeState { + last_height: height, + engine_state, + last_update: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0), + }; + if let Err(e) = storage.write().await.store_masternode_state(&state).await { + tracing::warn!("Could not persist masternode state at {height}: {e}"); + } else { + tracing::debug!("Persisted masternode state at height {height}"); } } @@ -559,6 +596,7 @@ impl MasternodesManager { self.sync_state.last_synced_block_hash = Some(latest_block_hash); self.progress.update_current_height(height); + self.persist_engine(height).await; tracing::debug!("Incremental MnListDiff complete at height {}", height); Ok(vec![SyncEvent::MasternodeStateUpdated { height, @@ -662,6 +700,10 @@ impl MasternodesManager { drop(engine); + if !events.is_empty() { + self.persist_engine(self.progress.current_height()).await; + } + if is_initial_sync { self.set_state(SyncState::Synced); tracing::info!("Masternode sync complete at height {}", self.progress.current_height()); @@ -696,7 +738,7 @@ mod tests { async fn create_test_manager_for(network: dashcore::Network) -> TestMasternodesManager { let storage = DiskStorageManager::with_temp_dir().await.unwrap(); let engine = Arc::new(RwLock::new(MasternodeListEngine::default_for_network(network))); - MasternodesManager::new(storage.block_headers(), engine, network).await + MasternodesManager::new(storage.block_headers(), engine, network, None).await } async fn create_test_manager() -> TestMasternodesManager { @@ -733,6 +775,7 @@ mod tests { block_headers, Arc::new(RwLock::new(engine)), dashcore::Network::Regtest, + None, ) .await; manager.set_state(SyncState::Synced); @@ -964,6 +1007,7 @@ mod tests { storage.block_headers(), Arc::new(RwLock::new(engine)), dashcore::Network::Testnet, + None, ) .await; diff --git a/dash-spv/src/sync/masternodes/sync_manager.rs b/dash-spv/src/sync/masternodes/sync_manager.rs index 1a077a8a2..d59b2b00b 100644 --- a/dash-spv/src/sync/masternodes/sync_manager.rs +++ b/dash-spv/src/sync/masternodes/sync_manager.rs @@ -1097,9 +1097,13 @@ mod tests { .await .unwrap(); let engine = MasternodeListEngine::default_for_network(Network::Regtest); - let mut manager = - MasternodesManager::new(block_headers, Arc::new(RwLock::new(engine)), Network::Regtest) - .await; + let mut manager = MasternodesManager::new( + block_headers, + Arc::new(RwLock::new(engine)), + Network::Regtest, + None, + ) + .await; manager.progress.update_block_header_tip_height(tip); let (tx, mut rx) = mpsc::unbounded_channel(); From d251592daaa24eeee0243edf0741674673d27151 Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Thu, 27 Aug 2026 15:09:55 +0000 Subject: [PATCH 3/7] refactor(dash-spv): let the storage own the masternode state format MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `MasternodeStateStorage` took and returned `MasternodeState`, the on-disk shape, so both callers had to build it: the manager serialized the engine, stamped a timestamp and assembled the struct, and the client took it apart again. Two places knew the encoding, and neither was the one that owns it. The trait now takes and returns the engine. `MasternodeState` stays as the file format and is built and read inside `masternode.rs` alone — it is no longer named outside `storage/`. Changing how the engine is encoded, which the current JSON-array-of-bytes shape will want, is now an edit to one file rather than three. `load_engine` also absorbs the case that is not an error: nothing persisted yet yields the network's default, which is where a first run starts anyway, so the caller loses an `Option` it only ever mapped one way. A file that exists and cannot be read stays an `Err`, because that one is worth seeing — the client logs it and rebuilds from the network. The masternode manager's persistence path goes from 24 lines to 5, the client's restore from 22 to 8. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_015NhHBDGiKfiGpy7FwooyfS --- dash-spv/src/client/lifecycle.rs | 26 +++-------- dash-spv/src/storage/masternode.rs | 58 ++++++++++++++++++++---- dash-spv/src/storage/mod.rs | 14 ++++-- dash-spv/src/sync/masternodes/manager.rs | 23 ++-------- 4 files changed, 68 insertions(+), 53 deletions(-) diff --git a/dash-spv/src/client/lifecycle.rs b/dash-spv/src/client/lifecycle.rs index d3b17d715..11452e20f 100644 --- a/dash-spv/src/client/lifecycle.rs +++ b/dash-spv/src/client/lifecycle.rs @@ -70,26 +70,12 @@ impl DashSpvClient match serde_json::from_slice(&state.engine_state) { - Ok(restored) => { - engine = restored; - tracing::info!( - "Restored masternode state from height {}", - state.last_height - ); - } - Err(e) => tracing::warn!( - "Could not read persisted masternode state, rebuilding: {}", - e - ), - }, - Ok(None) => tracing::debug!("No persisted masternode state"), - Err(e) => { - tracing::warn!("Could not load masternode state, rebuilding: {}", e) - } - } + let loader = storage.masternodestate(); + let engine = loader.read().await.load_engine(config.network).await; + let engine = engine.unwrap_or_else(|e| { + tracing::warn!("Could not load masternode state, rebuilding: {}", e); + MasternodeListEngine::default_for_network(config.network) + }); Some(Arc::new(RwLock::new(engine))) } else { None diff --git a/dash-spv/src/storage/masternode.rs b/dash-spv/src/storage/masternode.rs index d7ec1dd9f..6ec1a7217 100644 --- a/dash-spv/src/storage/masternode.rs +++ b/dash-spv/src/storage/masternode.rs @@ -2,16 +2,30 @@ use std::path::PathBuf; use async_trait::async_trait; +use dashcore::sml::masternode_list_engine::MasternodeListEngine; +use dashcore::Network; + use crate::{ error::StorageResult, storage::{io::atomic_write, MasternodeState, PersistentStorage}, }; +/// Persistence for the masternode list engine. +/// +/// Takes and returns the engine itself: the on-disk shape is +/// [`MasternodeState`] and stays here, so a caller neither builds it nor knows +/// how it is encoded. #[async_trait] pub trait MasternodeStateStorage { - async fn store_masternode_state(&mut self, state: &MasternodeState) -> StorageResult<()>; - - async fn load_masternode_state(&self) -> StorageResult>; + async fn store_engine( + &mut self, + engine: &MasternodeListEngine, + height: u32, + ) -> StorageResult<()>; + + /// Always yields an engine: with nothing persisted yet, the network's + /// default, which is what a first run starts from anyway. + async fn load_engine(&self, network: Network) -> StorageResult; } pub struct PersistentMasternodeStateStorage { @@ -39,13 +53,31 @@ impl PersistentStorage for PersistentMasternodeStateStorage { #[async_trait] impl MasternodeStateStorage for PersistentMasternodeStateStorage { - async fn store_masternode_state(&mut self, state: &MasternodeState) -> StorageResult<()> { + async fn store_engine( + &mut self, + engine: &MasternodeListEngine, + height: u32, + ) -> StorageResult<()> { let masternodestate_folder = self.storage_path.join(Self::FOLDER_NAME); let path = masternodestate_folder.join(Self::MASTERNODE_FILE_NAME); tokio::fs::create_dir_all(masternodestate_folder).await?; - let json = serde_json::to_string_pretty(state).map_err(|e| { + let state = MasternodeState { + last_height: height, + engine_state: serde_json::to_vec(engine).map_err(|e| { + crate::error::StorageError::Serialization(format!( + "Failed to serialize masternode engine: {}", + e + )) + })?, + last_update: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0), + }; + + let json = serde_json::to_string_pretty(&state).map_err(|e| { crate::error::StorageError::Serialization(format!( "Failed to serialize masternode state: {}", e @@ -56,21 +88,29 @@ impl MasternodeStateStorage for PersistentMasternodeStateStorage { Ok(()) } - async fn load_masternode_state(&self) -> StorageResult> { + async fn load_engine(&self, network: Network) -> StorageResult { let path = self.storage_path.join(Self::FOLDER_NAME).join(Self::MASTERNODE_FILE_NAME); if !path.exists() { - return Ok(None); + tracing::debug!("No persisted masternode state, starting from the network default"); + return Ok(MasternodeListEngine::default_for_network(network)); } let content = tokio::fs::read_to_string(path).await?; - let state = serde_json::from_str(&content).map_err(|e| { + let state: MasternodeState = serde_json::from_str(&content).map_err(|e| { crate::error::StorageError::Serialization(format!( "Failed to deserialize masternode state: {}", e )) })?; + let engine = serde_json::from_slice(&state.engine_state).map_err(|e| { + crate::error::StorageError::Serialization(format!( + "Failed to deserialize masternode engine: {}", + e + )) + })?; - Ok(Some(state)) + tracing::debug!("Loaded masternode engine from height {}", state.last_height); + Ok(engine) } } diff --git a/dash-spv/src/storage/mod.rs b/dash-spv/src/storage/mod.rs index fb43891ea..cfe974b77 100644 --- a/dash-spv/src/storage/mod.rs +++ b/dash-spv/src/storage/mod.rs @@ -19,6 +19,8 @@ use crate::ClientConfig; use async_trait::async_trait; use dashcore::hash_types::FilterHeader; use dashcore::prelude::CoreBlockHeight; +use dashcore::sml::masternode_list_engine::MasternodeListEngine; +use dashcore::Network; use std::ops::Range; use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -438,12 +440,16 @@ impl metadata::MetadataStorage for DiskStorageManager { #[async_trait] impl masternode::MasternodeStateStorage for DiskStorageManager { - async fn store_masternode_state(&mut self, state: &MasternodeState) -> StorageResult<()> { - self.masternodestate.write().await.store_masternode_state(state).await + async fn store_engine( + &mut self, + engine: &MasternodeListEngine, + height: u32, + ) -> StorageResult<()> { + self.masternodestate.write().await.store_engine(engine, height).await } - async fn load_masternode_state(&self) -> StorageResult> { - self.masternodestate.read().await.load_masternode_state().await + async fn load_engine(&self, network: Network) -> StorageResult { + self.masternodestate.read().await.load_engine(network).await } } diff --git a/dash-spv/src/sync/masternodes/manager.rs b/dash-spv/src/sync/masternodes/manager.rs index 868a96c90..0571c1d7c 100644 --- a/dash-spv/src/sync/masternodes/manager.rs +++ b/dash-spv/src/sync/masternodes/manager.rs @@ -15,7 +15,7 @@ use super::pipeline::MnListDiffPipeline; use crate::error::{SyncError, SyncResult}; use crate::network::RequestSender; use crate::storage::{ - BlockHeaderStorage, MasternodeState, MasternodeStateStorage, PersistentMasternodeStateStorage, + BlockHeaderStorage, MasternodeStateStorage, PersistentMasternodeStateStorage, }; use crate::sync::{MasternodesProgress, SyncEvent, SyncManager, SyncState}; use dashcore::network::message_qrinfo::QRInfo; @@ -352,25 +352,8 @@ impl MasternodesManager { let Some(storage) = &self.state_storage else { return; }; - let engine_state = { - let engine = self.engine.read().await; - match serde_json::to_vec(&*engine) { - Ok(bytes) => bytes, - Err(e) => { - tracing::warn!("Could not serialize masternode engine at {height}: {e}"); - return; - } - } - }; - let state = MasternodeState { - last_height: height, - engine_state, - last_update: std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_secs()) - .unwrap_or(0), - }; - if let Err(e) = storage.write().await.store_masternode_state(&state).await { + let engine = self.engine.read().await; + if let Err(e) = storage.write().await.store_engine(&engine, height).await { tracing::warn!("Could not persist masternode state at {height}: {e}"); } else { tracing::debug!("Persisted masternode state at height {height}"); From 7f5f7abf42cb66c402f92ecefe72432ba5b38604 Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Fri, 28 Aug 2026 12:33:09 +0000 Subject: [PATCH 4/7] refactor(dash-spv): improved and optimized masternode storage prototype --- dash-spv/Cargo.toml | 1 + dash-spv/src/client/lifecycle.rs | 10 +- dash-spv/src/storage/masternode.rs | 312 +++++++++++++----- dash-spv/src/storage/mod.rs | 65 ++-- dash-spv/src/storage/types.rs | 16 - dash-spv/src/sync/chainlock/manager.rs | 77 ++++- dash-spv/src/sync/masternodes/manager.rs | 53 ++- dash-spv/src/sync/masternodes/sync_manager.rs | 15 + dash-spv/tests/dashd_masternode/helpers.rs | 2 +- .../src/sml/masternode_list_engine/helpers.rs | 23 +- .../message_request_verification.rs | 2 +- 11 files changed, 433 insertions(+), 143 deletions(-) delete mode 100644 dash-spv/src/storage/types.rs diff --git a/dash-spv/Cargo.toml b/dash-spv/Cargo.toml index 0c80f5b36..d419d9978 100644 --- a/dash-spv/Cargo.toml +++ b/dash-spv/Cargo.toml @@ -33,6 +33,7 @@ thiserror = "1.0" # Serialization serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" +bincode = "2.0.1" # Logging tracing = "0.1" diff --git a/dash-spv/src/client/lifecycle.rs b/dash-spv/src/client/lifecycle.rs index 11452e20f..7dd9968e1 100644 --- a/dash-spv/src/client/lifecycle.rs +++ b/dash-spv/src/client/lifecycle.rs @@ -13,7 +13,7 @@ use crate::chain::checkpoints::CheckpointManager; use crate::error::{Result, SpvError}; use crate::network::NetworkManager; use crate::storage::{ - MasternodeStateStorage, PersistentBlockHeaderStorage, PersistentBlockStorage, + MasternodeStorage, PersistentBlockHeaderStorage, PersistentBlockStorage, PersistentFilterHeaderStorage, PersistentFilterStorage, PersistentMetadataStorage, StorageManager, }; @@ -70,10 +70,10 @@ impl DashSpvClient DashSpvClient DashSpvClient, BLSPublicKey, LLMQEntryVerificationStatus)>, +>; + +type EngineContext = (MasternodeListEngineBlockContainer, QuorumStatuses); -/// Persistence for the masternode list engine. -/// -/// Takes and returns the engine itself: the on-disk shape is -/// [`MasternodeState`] and stays here, so a caller neither builds it nor knows -/// how it is encoded. #[async_trait] -pub trait MasternodeStateStorage { - async fn store_engine( +pub trait MasternodeStorage: Send + Sync + 'static { + async fn store_diff(&mut self, height: CoreBlockHeight, diff: &MnListDiff) + -> StorageResult<()>; + + async fn store_qr_info( &mut self, - engine: &MasternodeListEngine, - height: u32, + height: CoreBlockHeight, + qr_info: &QRInfo, ) -> StorageResult<()>; - /// Always yields an engine: with nothing persisted yet, the network's - /// default, which is what a first run starts from anyway. + async fn store_context(&mut self, engine: &MasternodeListEngine) -> StorageResult<()>; + async fn load_engine(&self, network: Network) -> StorageResult; + + async fn masternode_list_at_or_before( + &self, + network: Network, + height: CoreBlockHeight, + ) -> StorageResult>; } -pub struct PersistentMasternodeStateStorage { +pub struct PersistentMasternodeStorage { storage_path: PathBuf, + diffs: BTreeMap, + qr_infos: BTreeMap, } -impl PersistentMasternodeStateStorage { - const FOLDER_NAME: &str = "masternodestate"; - const MASTERNODE_FILE_NAME: &str = "masternodestate.json"; +impl PersistentMasternodeStorage { + const FOLDER_NAME: &str = "masternodes"; + const DIFF_PREFIX: &str = "diff_"; + const QRINFO_PREFIX: &str = "qrinfo_"; + const EXTENSION: &str = "dat"; + const CONTEXT_FILE_NAME: &str = "context.dat"; + + 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() + } + + async fn index_folder(folder: &Path) -> StorageResult<(IndexMap, IndexMap)> { + let mut diffs = BTreeMap::new(); + let mut qr_infos = BTreeMap::new(); + + if !folder.exists() { + return Ok((diffs, qr_infos)); + } + + 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); + } + } + + Ok((diffs, qr_infos)) + } + + async fn write_message( + &mut self, + prefix: &str, + height: CoreBlockHeight, + message: &T, + ) -> StorageResult { + let folder = self.folder(); + tokio::fs::create_dir_all(&folder).await?; + let path = folder.join(Self::file_name(prefix, height)); + atomic_write(&path, &serialize(message)).await?; + Ok(path) + } + + 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 load_context(&self) -> StorageResult> { + let path = self.folder().join(Self::CONTEXT_FILE_NAME); + if !path.exists() { + return Ok(None); + } + let bytes = tokio::fs::read(&path).await?; + let (context, _) = bincode::decode_from_slice(&bytes, bincode::config::standard()) + .map_err(|e| { + StorageError::Corruption(format!("Failed to decode masternode context: {e}")) + })?; + Ok(Some(context)) + } + + async fn replay(&self, network: Network) -> StorageResult { + let mut engine = MasternodeListEngine::default_for_network(network); + + if let Some((block_container, quorum_statuses)) = self.load_context().await? { + engine.block_container = block_container; + engine.quorum_statuses = quorum_statuses; + } + + let mut pending_qr_infos = Vec::new(); + for (height, path) in &self.qr_infos { + match Self::read_message::(path).await { + Ok(qr_info) => pending_qr_infos.push((*height, qr_info)), + Err(e) => tracing::warn!("Skipping unreadable QRInfo at {height}: {e}"), + } + } + + let mut pending_diffs = Vec::new(); + for (height, path) in &self.diffs { + match Self::read_message::(path).await { + Ok(diff) => pending_diffs.push((*height, diff)), + Err(e) => tracing::warn!("Skipping unreadable MnListDiff at {height}: {e}"), + } + } + + let qr_info_count = pending_qr_infos.len(); + let diff_count = pending_diffs.len(); + + loop { + let remaining = pending_qr_infos.len() + pending_diffs.len(); + + pending_qr_infos + .retain(|(_, qr_info)| engine.feed_qr_info(qr_info.clone(), true, true).is_err()); + + pending_diffs.retain(|(height, diff)| { + engine.feed_block_height(*height, diff.block_hash); + engine.apply_diff(diff.clone(), Some(*height), false, None).is_err() + }); + + if pending_qr_infos.len() + pending_diffs.len() == remaining { + break; + } + } + + for (height, _) in &pending_qr_infos { + tracing::warn!("QRInfo at {height} has no reachable base, leaving it to the network"); + } + + for (height, _) in &pending_diffs { + tracing::warn!( + "MnListDiff at {height} has no reachable base, leaving it to the network" + ); + } + + tracing::debug!( + "Replayed {}/{} QRInfo and {}/{} MnListDiff messages into {} masternode lists", + qr_info_count - pending_qr_infos.len(), + qr_info_count, + diff_count - pending_diffs.len(), + diff_count, + engine.masternode_lists.len() + ); + + Ok(engine) + } } +type IndexMap = BTreeMap; + #[async_trait] -impl PersistentStorage for PersistentMasternodeStateStorage { +impl PersistentStorage for PersistentMasternodeStorage { async fn open(storage_path: impl Into + Send) -> StorageResult { - Ok(PersistentMasternodeStateStorage { - storage_path: storage_path.into(), + let storage_path = storage_path.into(); + let (diffs, qr_infos) = Self::index_folder(&storage_path.join(Self::FOLDER_NAME)).await?; + + Ok(PersistentMasternodeStorage { + storage_path, + diffs, + qr_infos, }) } async fn persist(&mut self, _storage_path: impl Into + Send) -> StorageResult<()> { - // Current implementation persists data everytime data is stored Ok(()) } } #[async_trait] -impl MasternodeStateStorage for PersistentMasternodeStateStorage { - async fn store_engine( +impl MasternodeStorage for PersistentMasternodeStorage { + async fn store_diff( &mut self, - engine: &MasternodeListEngine, - height: u32, + height: CoreBlockHeight, + diff: &MnListDiff, ) -> StorageResult<()> { - let masternodestate_folder = self.storage_path.join(Self::FOLDER_NAME); - let path = masternodestate_folder.join(Self::MASTERNODE_FILE_NAME); - - tokio::fs::create_dir_all(masternodestate_folder).await?; - - let state = MasternodeState { - last_height: height, - engine_state: serde_json::to_vec(engine).map_err(|e| { - crate::error::StorageError::Serialization(format!( - "Failed to serialize masternode engine: {}", - e - )) - })?, - last_update: std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_secs()) - .unwrap_or(0), - }; - - let json = serde_json::to_string_pretty(&state).map_err(|e| { - crate::error::StorageError::Serialization(format!( - "Failed to serialize masternode state: {}", - e - )) - })?; - - atomic_write(&path, json.as_bytes()).await?; + if self.diffs.contains_key(&height) { + return Ok(()); + } + let path = self.write_message(Self::DIFF_PREFIX, height, diff).await?; + self.diffs.insert(height, path); Ok(()) } - async fn load_engine(&self, network: Network) -> StorageResult { - let path = self.storage_path.join(Self::FOLDER_NAME).join(Self::MASTERNODE_FILE_NAME); - - if !path.exists() { - tracing::debug!("No persisted masternode state, starting from the network default"); - return Ok(MasternodeListEngine::default_for_network(network)); + async fn store_qr_info( + &mut self, + height: CoreBlockHeight, + qr_info: &QRInfo, + ) -> StorageResult<()> { + if self.qr_infos.contains_key(&height) { + return Ok(()); } + let path = self.write_message(Self::QRINFO_PREFIX, height, qr_info).await?; + self.qr_infos.insert(height, path); + Ok(()) + } - let content = tokio::fs::read_to_string(path).await?; - let state: MasternodeState = serde_json::from_str(&content).map_err(|e| { - crate::error::StorageError::Serialization(format!( - "Failed to deserialize masternode state: {}", - e - )) - })?; - let engine = serde_json::from_slice(&state.engine_state).map_err(|e| { - crate::error::StorageError::Serialization(format!( - "Failed to deserialize masternode engine: {}", - e - )) + async fn store_context(&mut self, engine: &MasternodeListEngine) -> StorageResult<()> { + let folder = self.folder(); + tokio::fs::create_dir_all(&folder).await?; + + let bytes = bincode::encode_to_vec( + (&engine.block_container, &engine.quorum_statuses), + bincode::config::standard(), + ) + .map_err(|e| { + StorageError::Serialization(format!("Failed to encode masternode context: {e}")) })?; - tracing::debug!("Loaded masternode engine from height {}", state.last_height); - Ok(engine) + atomic_write(&folder.join(Self::CONTEXT_FILE_NAME), &bytes).await + } + + async fn load_engine(&self, network: Network) -> StorageResult { + self.replay(network).await + } + + async fn masternode_list_at_or_before( + &self, + network: Network, + height: CoreBlockHeight, + ) -> StorageResult> { + let engine = self.replay(network).await?; + Ok(engine.masternode_lists_around_height(height).0.cloned()) } } diff --git a/dash-spv/src/storage/mod.rs b/dash-spv/src/storage/mod.rs index cfe974b77..f4b5c5d40 100644 --- a/dash-spv/src/storage/mod.rs +++ b/dash-spv/src/storage/mod.rs @@ -1,7 +1,5 @@ //! Storage abstraction for the Dash SPV client. -pub mod types; - mod block_headers; mod blocks; mod filter_headers; @@ -18,7 +16,10 @@ 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::sml::masternode_list::MasternodeList; use dashcore::sml::masternode_list_engine::MasternodeListEngine; use dashcore::Network; use std::ops::Range; @@ -33,12 +34,10 @@ pub use crate::storage::block_headers::{ pub use crate::storage::blocks::{BlockStorage, PersistentBlockStorage}; pub use crate::storage::filter_headers::{FilterHeaderStorage, PersistentFilterHeaderStorage}; pub use crate::storage::filters::{FilterStorage, PersistentFilterStorage}; -pub use crate::storage::masternode::{MasternodeStateStorage, PersistentMasternodeStateStorage}; +pub use crate::storage::masternode::{MasternodeStorage, PersistentMasternodeStorage}; pub use crate::storage::metadata::{MetadataStorage, PersistentMetadataStorage}; pub use crate::storage::peers::{PeerStorage, PersistentPeerStorage}; -pub use types::*; - #[async_trait] pub trait PersistentStorage: Sized { /// If the storage_path contains persisted data the storage will use it, if not, @@ -55,7 +54,7 @@ pub trait StorageManager: + FilterStorage + BlockStorage + MetadataStorage - + MasternodeStateStorage + + MasternodeStorage + Send + Sync + 'static @@ -81,7 +80,7 @@ pub trait StorageManager: /// Returns shared access to the metadata storage. fn metadata(&self) -> Arc>; - fn masternodestate(&self) -> Arc>; + fn masternodes(&self) -> Arc>; } /// Disk-based storage manager with segmented files and async background saving. @@ -95,7 +94,7 @@ pub struct DiskStorageManager { filters: Arc>, blocks: Arc>, metadata: Arc>, - masternodestate: Arc>, + masternodes: Arc>, // Background worker worker_handle: Option>, @@ -148,8 +147,8 @@ impl DiskStorageManager { filters: Arc::new(RwLock::new(PersistentFilterStorage::open(&storage_path).await?)), blocks: Arc::new(RwLock::new(PersistentBlockStorage::open(&storage_path).await?)), metadata: Arc::new(RwLock::new(PersistentMetadataStorage::open(&storage_path).await?)), - masternodestate: Arc::new(RwLock::new( - PersistentMasternodeStateStorage::open(&storage_path).await?, + masternodes: Arc::new(RwLock::new( + PersistentMasternodeStorage::open(&storage_path).await?, )), worker_handle: None, @@ -177,7 +176,7 @@ impl DiskStorageManager { let filters = Arc::clone(&self.filters); let blocks = Arc::clone(&self.blocks); let metadata = Arc::clone(&self.metadata); - let masternodestate = Arc::clone(&self.masternodestate); + let masternodes = Arc::clone(&self.masternodes); let storage_path = self.storage_path.clone(); @@ -192,7 +191,7 @@ impl DiskStorageManager { let _ = filters.write().await.persist(&storage_path).await; let _ = blocks.write().await.persist(&storage_path).await; let _ = metadata.write().await.persist(&storage_path).await; - let _ = masternodestate.write().await.persist(&storage_path).await; + let _ = masternodes.write().await.persist(&storage_path).await; } }); @@ -214,7 +213,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.masternodestate.write().await.persist(storage_path).await; + let _ = self.masternodes.write().await.persist(storage_path).await; } } @@ -251,8 +250,8 @@ 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.masternodestate = - Arc::new(RwLock::new(PersistentMasternodeStateStorage::open(storage_path).await?)); + self.masternodes = + Arc::new(RwLock::new(PersistentMasternodeStorage::open(storage_path).await?)); // Restart the background worker for future operations self.start_worker().await; @@ -287,8 +286,8 @@ impl StorageManager for DiskStorageManager { Arc::clone(&self.metadata) } - fn masternodestate(&self) -> Arc> { - Arc::clone(&self.masternodestate) + fn masternodes(&self) -> Arc> { + Arc::clone(&self.masternodes) } } @@ -439,17 +438,37 @@ impl metadata::MetadataStorage for DiskStorageManager { } #[async_trait] -impl masternode::MasternodeStateStorage for DiskStorageManager { - async fn store_engine( +impl masternode::MasternodeStorage for DiskStorageManager { + async fn store_diff( &mut self, - engine: &MasternodeListEngine, - height: u32, + 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.masternodestate.write().await.store_engine(engine, height).await + self.masternodes.write().await.store_qr_info(height, qr_info).await + } + + async fn store_context(&mut self, engine: &MasternodeListEngine) -> StorageResult<()> { + self.masternodes.write().await.store_context(engine).await } async fn load_engine(&self, network: Network) -> StorageResult { - self.masternodestate.read().await.load_engine(network).await + self.masternodes.read().await.load_engine(network).await + } + + async fn masternode_list_at_or_before( + &self, + network: Network, + height: CoreBlockHeight, + ) -> StorageResult> { + self.masternodes.read().await.masternode_list_at_or_before(network, height).await } } diff --git a/dash-spv/src/storage/types.rs b/dash-spv/src/storage/types.rs deleted file mode 100644 index 678553caa..000000000 --- a/dash-spv/src/storage/types.rs +++ /dev/null @@ -1,16 +0,0 @@ -//! Storage-related types and structures. - -use serde::{Deserialize, Serialize}; - -/// Masternode state for storage. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct MasternodeState { - /// Last processed height. - pub last_height: u32, - - /// Serialized masternode list engine state. - pub engine_state: Vec, - - /// Last update timestamp. - pub last_update: u64, -} diff --git a/dash-spv/src/sync/chainlock/manager.rs b/dash-spv/src/sync/chainlock/manager.rs index c211919df..40a2d8d93 100644 --- a/dash-spv/src/sync/chainlock/manager.rs +++ b/dash-spv/src/sync/chainlock/manager.rs @@ -10,11 +10,14 @@ use std::sync::Arc; use dashcore::ephemerealdata::chain_lock::ChainLock; use dashcore::hash_types::ChainLockHash; use dashcore::sml::masternode_list_engine::MasternodeListEngine; +use dashcore::Network; use std::collections::HashSet; use tokio::sync::RwLock; use crate::error::SyncResult; -use crate::storage::{BlockHeaderStorage, MetadataStorage}; +use crate::storage::{ + BlockHeaderStorage, MasternodeStorage, MetadataStorage, PersistentMasternodeStorage, +}; use crate::sync::{ChainLockProgress, SyncEvent}; /// Metadata key for persisting the best validated ChainLock. @@ -36,6 +39,9 @@ pub struct ChainLockManager { metadata_storage: Arc>, /// Masternode engine for BLS signature validation. masternode_engine: Arc>, + /// Rebuilds a masternode list the engine no longer retains. + masternode_storage: Option>>, + network: Network, /// The best (highest height) validated ChainLock. best_chainlock: Option, /// ChainLock hashes that have been requested (to avoid duplicate requests). @@ -56,12 +62,16 @@ impl ChainLockManager { header_storage: Arc>, metadata_storage: Arc>, masternode_engine: Arc>, + masternode_storage: Option>>, + network: Network, ) -> Self { let mut manager = Self { progress: ChainLockProgress::default(), header_storage, metadata_storage, masternode_engine, + masternode_storage, + network, best_chainlock: None, requested_chainlocks: HashSet::new(), masternode_ready: false, @@ -254,6 +264,53 @@ impl ChainLockManager { "ChainLock signature verified for height {}", chainlock.block_height ); + return true; + } + Err(e) => tracing::debug!( + "ChainLock at height {} not verifiable against the retained lists: {}", + chainlock.block_height, + e + ), + } + drop(engine); + + self.validate_signature_from_storage(chainlock).await + } + + async fn validate_signature_from_storage(&self, chainlock: &ChainLock) -> bool { + let Some(storage) = &self.masternode_storage else { + return false; + }; + + let signing_height = chainlock.block_height.saturating_sub(8); + let list = + storage.read().await.masternode_list_at_or_before(self.network, signing_height).await; + + let list = match list { + Ok(Some(list)) => list, + Ok(None) => return false, + Err(e) => { + tracing::warn!( + "Could not rebuild the masternode list for height {}: {}", + signing_height, + e + ); + return false; + } + }; + + let engine = self.masternode_engine.read().await; + let Ok(request_id) = chainlock.request_id() else { + return false; + }; + + match engine.verify_chain_lock_with_masternode_list(chainlock, &list, &request_id) { + Ok(()) => { + tracing::info!( + "ChainLock signature verified for height {} from a rebuilt list at {}", + chainlock.block_height, + list.known_height + ); true } Err(e) => { @@ -309,7 +366,14 @@ mod tests { let storage = DiskStorageManager::with_temp_dir().await.unwrap(); let engine = Arc::new(RwLock::new(MasternodeListEngine::default_for_network(Network::Testnet))); - ChainLockManager::new(storage.block_headers(), storage.metadata(), engine).await + ChainLockManager::new( + storage.block_headers(), + storage.metadata(), + engine, + None, + Network::Testnet, + ) + .await } async fn create_test_manager_with_storage( @@ -317,7 +381,14 @@ mod tests { ) -> TestChainLockManager { let engine = Arc::new(RwLock::new(MasternodeListEngine::default_for_network(Network::Testnet))); - ChainLockManager::new(storage.block_headers(), storage.metadata(), engine).await + ChainLockManager::new( + storage.block_headers(), + storage.metadata(), + engine, + None, + Network::Testnet, + ) + .await } fn create_test_chainlock(height: u32) -> ChainLock { diff --git a/dash-spv/src/sync/masternodes/manager.rs b/dash-spv/src/sync/masternodes/manager.rs index 0571c1d7c..96b012f3e 100644 --- a/dash-spv/src/sync/masternodes/manager.rs +++ b/dash-spv/src/sync/masternodes/manager.rs @@ -14,11 +14,10 @@ use tokio::sync::RwLock; use super::pipeline::MnListDiffPipeline; use crate::error::{SyncError, SyncResult}; use crate::network::RequestSender; -use crate::storage::{ - BlockHeaderStorage, MasternodeStateStorage, PersistentMasternodeStateStorage, -}; +use crate::storage::{BlockHeaderStorage, MasternodeStorage, 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; @@ -302,7 +301,7 @@ pub struct MasternodesManager { /// Sync state tracking. pub(super) sync_state: MasternodeSyncState, /// `None` leaves the list in memory only. - pub(super) state_storage: Option>>, + pub(super) message_storage: Option>>, } impl MasternodesManager { @@ -311,7 +310,7 @@ impl MasternodesManager { header_storage: Arc>, engine: Arc>, network: dashcore::Network, - state_storage: Option>>, + message_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. @@ -342,22 +341,42 @@ impl MasternodesManager { engine, network, sync_state, - state_storage, + message_storage, } } - /// Best effort: an unwritten list costs a rebuild next start, a failed sync - /// costs the list now. - pub(super) async fn persist_engine(&self, height: u32) { - let Some(storage) = &self.state_storage else { + pub(super) async fn store_diff(&self, height: u32, diff: &MnListDiff) { + let Some(storage) = &self.message_storage else { return; }; - let engine = self.engine.read().await; - if let Err(e) = storage.write().await.store_engine(&engine, height).await { - tracing::warn!("Could not persist masternode state at {height}: {e}"); - } else { - tracing::debug!("Persisted masternode state at height {height}"); + if let Err(e) = storage.write().await.store_diff(height, diff).await { + tracing::warn!("Could not store MnListDiff at {height}: {e}"); + } + } + + pub(super) async fn store_qr_info(&self, height: u32, qr_info: &QRInfo) { + let Some(storage) = &self.message_storage else { + return; + }; + if let Err(e) = storage.write().await.store_qr_info(height, qr_info).await { + tracing::warn!("Could not store QRInfo at {height}: {e}"); + } + } + + pub(super) async fn persist_context(&self, tip: u32) { + let Some(storage) = &self.message_storage else { + return; + }; + + let mut engine = self.engine.write().await; + let pruned = engine.prune_masternode_lists(tip); + + if let Err(e) = storage.write().await.store_context(&engine).await { + tracing::warn!("Could not persist masternode context at {tip}: {e}"); + return; } + + tracing::debug!("Persisted masternode context at {tip}, pruned {pruned} in-memory lists"); } /// Decide which [`PipelineMode`] to use when a new header lands at `tip_height` @@ -579,7 +598,7 @@ impl MasternodesManager { self.sync_state.last_synced_block_hash = Some(latest_block_hash); self.progress.update_current_height(height); - self.persist_engine(height).await; + self.persist_context(height).await; tracing::debug!("Incremental MnListDiff complete at height {}", height); Ok(vec![SyncEvent::MasternodeStateUpdated { height, @@ -684,7 +703,7 @@ impl MasternodesManager { drop(engine); if !events.is_empty() { - self.persist_engine(self.progress.current_height()).await; + self.persist_context(self.progress.current_height()).await; } if is_initial_sync { diff --git a/dash-spv/src/sync/masternodes/sync_manager.rs b/dash-spv/src/sync/masternodes/sync_manager.rs index d59b2b00b..ae16cf6bf 100644 --- a/dash-spv/src/sync/masternodes/sync_manager.rs +++ b/dash-spv/src/sync/masternodes/sync_manager.rs @@ -295,6 +295,9 @@ impl SyncManager for MasternodesManager { } }; + let qr_info_height = + engine.block_container.get_height(&qr_info.mn_list_diff_tip.block_hash); + // Populate known_mn_list_heights from engine after QRInfo processing self.sync_state.known_mn_list_heights = engine.masternode_lists.keys().copied().collect(); @@ -318,6 +321,14 @@ impl SyncManager for MasternodesManager { drop(engine); drop(storage); + match qr_info_height { + Some(height) => self.store_qr_info(height, qr_info).await, + None => tracing::warn!( + "QRInfo tip {} has no known height, rotated quorums will not survive a restart", + qr_info.mn_list_diff_tip.block_hash + ), + } + if let Some(ref qr_info_result) = qr_info_result { tracing::info!( "QRInfo processed: stored_cycle_height={:?}, rotated_quorum_count={}/{}, fully_verified_count={}, newly_qualified_count={}, cycle_key_unresolved={}, previous_cycle_invalid_count={}", @@ -431,6 +442,10 @@ impl SyncManager for MasternodesManager { }; drop(engine); + if apply_ok { + self.store_diff(target_height, diff).await; + } + self.progress.add_diffs_processed(1); self.sync_state.mnlistdiff_pipeline.receive(diff); self.sync_state.mnlistdiff_pipeline.send_pending(requests)?; diff --git a/dash-spv/tests/dashd_masternode/helpers.rs b/dash-spv/tests/dashd_masternode/helpers.rs index 8166df9bb..d981af7a1 100644 --- a/dash-spv/tests/dashd_masternode/helpers.rs +++ b/dash-spv/tests/dashd_masternode/helpers.rs @@ -64,7 +64,7 @@ pub(super) const EXPECTED_STORAGE: &[(&str, &str)] = &[ ("filter_headers", "filter headers synced to the tip"), ("metadata", "sync checkpoints"), ("peers", "peer set and reputations"), - ("masternodestate", "the masternode list this session built"), + ("masternodes", "the masternode messages this session stored"), ]; /// Assert every directory in [`EXPECTED_STORAGE`] exists and holds at least one diff --git a/dash/src/sml/masternode_list_engine/helpers.rs b/dash/src/sml/masternode_list_engine/helpers.rs index 9dfbaeccc..be72bc3e4 100644 --- a/dash/src/sml/masternode_list_engine/helpers.rs +++ b/dash/src/sml/masternode_list_engine/helpers.rs @@ -2,6 +2,8 @@ use crate::QuorumHash; use crate::prelude::CoreBlockHeight; use crate::sml::llmq_entry_verification::LLMQEntryVerificationStatus; use crate::sml::llmq_type::LLMQType; +#[cfg(feature = "quorum_validation")] +use crate::sml::llmq_type::network::NetworkLLMQExt; use crate::sml::masternode_list::MasternodeList; use crate::sml::masternode_list_engine::MasternodeListEngine; use crate::sml::quorum_entry::qualified_quorum_entry::QualifiedQuorumEntry; @@ -11,9 +13,28 @@ use crate::sml::quorum_entry::qualified_quorum_entry::QualifiedQuorumEntry; /// height that can exceed one active window (Platform selects roughly 4.5 DKG intervals back), so a /// single window is too tight. Four windows covers that lag with wide margin while still bounding a /// miss to a fixed span of lists rather than every list the engine has accumulated. -const QUORUM_WALK_BACK_ACTIVE_WINDOWS: u32 = 4; +pub const QUORUM_WALK_BACK_ACTIVE_WINDOWS: u32 = 4; impl MasternodeListEngine { + #[cfg(feature = "quorum_validation")] + pub fn retained_list_floor(&self, tip: CoreBlockHeight) -> CoreBlockHeight { + let params = self.network.chain_locks_type().params(); + tip.saturating_sub( + params + .signing_active_quorum_count + .saturating_mul(params.dkg_params.interval) + .saturating_mul(QUORUM_WALK_BACK_ACTIVE_WINDOWS), + ) + } + + #[cfg(feature = "quorum_validation")] + pub fn prune_masternode_lists(&mut self, tip: CoreBlockHeight) -> usize { + let floor = self.retained_list_floor(tip); + let before = self.masternode_lists.len(); + self.masternode_lists.retain(|height, _| *height >= floor); + before - self.masternode_lists.len() + } + /// Retrieves the closest masternode lists before and after a given core block height. /// /// This function searches the `masternode_lists` map to find the nearest masternode lists diff --git a/dash/src/sml/masternode_list_engine/message_request_verification.rs b/dash/src/sml/masternode_list_engine/message_request_verification.rs index 662626ace..303a06827 100644 --- a/dash/src/sml/masternode_list_engine/message_request_verification.rs +++ b/dash/src/sml/masternode_list_engine/message_request_verification.rs @@ -383,7 +383,7 @@ impl MasternodeListEngine { } /// Helper function to verify a ChainLock using a specific masternode list. - fn verify_chain_lock_with_masternode_list( + pub fn verify_chain_lock_with_masternode_list( &self, chain_lock: &ChainLock, masternode_list: &MasternodeList, From efc8dbd711f99c0e150ce4580ffb91ee26dcf066 Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Sun, 30 Aug 2026 14:52:14 +0000 Subject: [PATCH 5/7] refactor(dash-spv): rebuild masternode state from headers, not a stored copy The masternode storage persisted the engine's block container alongside the quorum statuses, so every hash/height pair the replay needed was written a second time next to the header storage that already holds it. Two copies of the same mapping can only diverge: a header reorg rewrites one of them. The replay now works exactly like the live sync path, only reading its messages from disk instead of waiting for peers. The header storage is injected at construction, `PersistentMasternodeStorage` keeps the shared handle, and heights are resolved against it: QRInfo through the same `feed_qrinfo_heights_to_engine` the sync manager uses (moved to the storage module, where both callers reach it), MnListDiff through its file name plus a lookup of the base hash it extends. What is left of the old context file is the quorum statuses, so it is named for them: `quorum_statuses.dat`, written by `store_quorum_statuses`. One that fails to decode is now ignored with a warning instead of failing the whole load, since the statuses are a verification cache the replay re-derives. Tying the storage to `H` also ties it to the same header storage its manager already carries, instead of hardcoding the concrete type. `PersistentStorage` is gone from this storage: its `persist` was a no-op, the messages are written as they arrive, so the background worker no longer wakes it up. Verified against dashd regtest: the whole `dash-spv` suite passes (569 unit, 10 dashd_masternode, 30 dashd_sync). The restart test's log shows the replay rebuilding the lists purely from headers plus stored messages: "Replayed 1/1 QRInfo and 1/1 MnListDiff messages into 7 masternode lists". Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Uep5dcAKWZ3GA2Rm7GqUbr --- dash-spv/src/storage/masternode.rs | 171 +++++++++++++----- dash-spv/src/storage/mod.rs | 33 ++-- dash-spv/src/sync/chainlock/manager.rs | 4 +- dash-spv/src/sync/masternodes/manager.rs | 18 +- dash-spv/src/sync/masternodes/sync_manager.rs | 74 +------- 5 files changed, 160 insertions(+), 140 deletions(-) diff --git a/dash-spv/src/storage/masternode.rs b/dash-spv/src/storage/masternode.rs index 0df623336..9d9b97791 100644 --- a/dash-spv/src/storage/masternode.rs +++ b/dash-spv/src/storage/masternode.rs @@ -1,7 +1,9 @@ use std::collections::{BTreeMap, BTreeSet}; use std::path::{Path, PathBuf}; +use std::sync::Arc; use async_trait::async_trait; +use tokio::sync::RwLock; use dashcore::bls_sig_utils::BLSPublicKey; use dashcore::consensus::{deserialize, serialize, Decodable, Encodable}; @@ -11,21 +13,17 @@ use dashcore::prelude::CoreBlockHeight; use dashcore::sml::llmq_entry_verification::LLMQEntryVerificationStatus; use dashcore::sml::llmq_type::LLMQType; use dashcore::sml::masternode_list::MasternodeList; -use dashcore::sml::masternode_list_engine::{ - MasternodeListEngine, MasternodeListEngineBlockContainer, -}; +use dashcore::sml::masternode_list_engine::{MasternodeListEngine, WORK_DIFF_DEPTH}; use dashcore::{Network, QuorumHash}; use crate::error::{StorageError, StorageResult}; -use crate::storage::{io::atomic_write, PersistentStorage}; +use crate::storage::{io::atomic_write, BlockHeaderStorage}; type QuorumStatuses = BTreeMap< LLMQType, BTreeMap, BLSPublicKey, LLMQEntryVerificationStatus)>, >; -type EngineContext = (MasternodeListEngineBlockContainer, QuorumStatuses); - #[async_trait] pub trait MasternodeStorage: Send + Sync + 'static { async fn store_diff(&mut self, height: CoreBlockHeight, diff: &MnListDiff) @@ -37,7 +35,7 @@ pub trait MasternodeStorage: Send + Sync + 'static { qr_info: &QRInfo, ) -> StorageResult<()>; - async fn store_context(&mut self, engine: &MasternodeListEngine) -> StorageResult<()>; + async fn store_quorum_statuses(&mut self, engine: &MasternodeListEngine) -> StorageResult<()>; async fn load_engine(&self, network: Network) -> StorageResult; @@ -48,18 +46,40 @@ pub trait MasternodeStorage: Send + Sync + 'static { ) -> StorageResult>; } -pub struct PersistentMasternodeStorage { +/// Stores the raw messages a masternode list is rebuilt from. The heights the +/// replay needs come from the header storage rather than from a persisted copy +/// of the engine's block container: the headers already hold every hash/height +/// pair, and a second copy can only go stale against them. +pub struct PersistentMasternodeStorage { storage_path: PathBuf, + /// Shared with whoever writes the headers, so a replay reads them as they + /// stand rather than as they were when this storage was opened. + headers: Arc>, diffs: BTreeMap, qr_infos: BTreeMap, } -impl PersistentMasternodeStorage { +impl PersistentMasternodeStorage { const FOLDER_NAME: &str = "masternodes"; const DIFF_PREFIX: &str = "diff_"; const QRINFO_PREFIX: &str = "qrinfo_"; const EXTENSION: &str = "dat"; - const CONTEXT_FILE_NAME: &str = "context.dat"; + const QUORUM_STATUSES_FILE_NAME: &str = "quorum_statuses.dat"; + + pub async fn open( + storage_path: impl Into + Send, + headers: Arc>, + ) -> StorageResult { + 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, + diffs, + qr_infos, + }) + } fn folder(&self) -> PathBuf { self.storage_path.join(Self::FOLDER_NAME) @@ -117,24 +137,27 @@ impl PersistentMasternodeStorage { }) } - async fn load_context(&self) -> StorageResult> { - let path = self.folder().join(Self::CONTEXT_FILE_NAME); + async fn load_quorum_statuses(&self) -> StorageResult> { + let path = self.folder().join(Self::QUORUM_STATUSES_FILE_NAME); if !path.exists() { return Ok(None); } let bytes = tokio::fs::read(&path).await?; - let (context, _) = bincode::decode_from_slice(&bytes, bincode::config::standard()) - .map_err(|e| { - StorageError::Corruption(format!("Failed to decode masternode context: {e}")) - })?; - Ok(Some(context)) + match bincode::decode_from_slice(&bytes, bincode::config::standard()) { + Ok((statuses, _)) => Ok(Some(statuses)), + // The statuses are a cache of what the replay re-derives, so a file + // written by an older format costs verification work, not the sync. + Err(e) => { + tracing::warn!("Ignoring undecodable masternode quorum statuses: {e}"); + Ok(None) + } + } } async fn replay(&self, network: Network) -> StorageResult { let mut engine = MasternodeListEngine::default_for_network(network); - if let Some((block_container, quorum_statuses)) = self.load_context().await? { - engine.block_container = block_container; + if let Some(quorum_statuses) = self.load_quorum_statuses().await? { engine.quorum_statuses = quorum_statuses; } @@ -154,6 +177,23 @@ impl PersistentMasternodeStorage { } } + { + let headers = self.headers.read().await; + for (_, qr_info) in &pending_qr_infos { + feed_qrinfo_heights_to_engine(&mut engine, qr_info, &*headers).await; + } + // A diff is keyed by its own height, but applying it also needs the + // height of the list it extends. + for (height, diff) in &pending_diffs { + engine.feed_block_height(*height, diff.block_hash); + if let Ok(Some(base_height)) = + headers.get_header_height_by_hash(&diff.base_block_hash).await + { + engine.feed_block_height(base_height, diff.base_block_hash); + } + } + } + let qr_info_count = pending_qr_infos.len(); let diff_count = pending_diffs.len(); @@ -164,7 +204,6 @@ impl PersistentMasternodeStorage { .retain(|(_, qr_info)| engine.feed_qr_info(qr_info.clone(), true, true).is_err()); pending_diffs.retain(|(height, diff)| { - engine.feed_block_height(*height, diff.block_hash); engine.apply_diff(diff.clone(), Some(*height), false, None).is_err() }); @@ -199,25 +238,7 @@ impl PersistentMasternodeStorage { type IndexMap = BTreeMap; #[async_trait] -impl PersistentStorage for PersistentMasternodeStorage { - async fn open(storage_path: impl Into + Send) -> StorageResult { - let storage_path = storage_path.into(); - let (diffs, qr_infos) = Self::index_folder(&storage_path.join(Self::FOLDER_NAME)).await?; - - Ok(PersistentMasternodeStorage { - storage_path, - diffs, - qr_infos, - }) - } - - async fn persist(&mut self, _storage_path: impl Into + Send) -> StorageResult<()> { - Ok(()) - } -} - -#[async_trait] -impl MasternodeStorage for PersistentMasternodeStorage { +impl MasternodeStorage for PersistentMasternodeStorage { async fn store_diff( &mut self, height: CoreBlockHeight, @@ -244,19 +265,16 @@ impl MasternodeStorage for PersistentMasternodeStorage { Ok(()) } - async fn store_context(&mut self, engine: &MasternodeListEngine) -> StorageResult<()> { + async fn store_quorum_statuses(&mut self, engine: &MasternodeListEngine) -> StorageResult<()> { let folder = self.folder(); tokio::fs::create_dir_all(&folder).await?; - let bytes = bincode::encode_to_vec( - (&engine.block_container, &engine.quorum_statuses), - bincode::config::standard(), - ) - .map_err(|e| { - StorageError::Serialization(format!("Failed to encode masternode context: {e}")) - })?; + let bytes = bincode::encode_to_vec(&engine.quorum_statuses, bincode::config::standard()) + .map_err(|e| { + StorageError::Serialization(format!("Failed to encode quorum statuses: {e}")) + })?; - atomic_write(&folder.join(Self::CONTEXT_FILE_NAME), &bytes).await + atomic_write(&folder.join(Self::QUORUM_STATUSES_FILE_NAME), &bytes).await } async fn load_engine(&self, network: Network) -> StorageResult { @@ -272,3 +290,60 @@ impl MasternodeStorage for PersistentMasternodeStorage { Ok(engine.masternode_lists_around_height(height).0.cloned()) } } + +/// Feed QRInfo block heights to the engine from the header storage. +/// +/// Resolves heights for every hash enumerated by +/// [`MasternodeListEngine::qr_info_referenced_block_hashes`], plus the cycle boundary +/// block for each work-block diff (`work_height + WORK_DIFF_DEPTH`), which is needed +/// for rotated quorum storage key calculation. +pub(crate) async fn feed_qrinfo_heights_to_engine( + engine: &mut MasternodeListEngine, + qr_info: &QRInfo, + storage: &S, +) -> usize { + let mut fed_count = 0; + for block_hash in MasternodeListEngine::qr_info_referenced_block_hashes(qr_info) { + if let Ok(Some(height)) = storage.get_header_height_by_hash(&block_hash).await { + engine.feed_block_height(height, block_hash); + fed_count += 1; + tracing::trace!("Fed height {} for block {}", height, block_hash); + } + } + + // Feed cycle boundary heights for all diffs (current and historical cycles). + // Each diff's block_hash is at the "work block" height; the cycle boundary is + // WORK_DIFF_DEPTH higher. + let mut work_block_hashes = vec![ + qr_info.mn_list_diff_h.block_hash, + qr_info.mn_list_diff_at_h_minus_c.block_hash, + qr_info.mn_list_diff_at_h_minus_2c.block_hash, + qr_info.mn_list_diff_at_h_minus_3c.block_hash, + ]; + + if let Some((_, diff)) = &qr_info.quorum_snapshot_and_mn_list_diff_at_h_minus_4c { + work_block_hashes.push(diff.block_hash); + } + + for work_block_hash in work_block_hashes { + if let Ok(Some(work_block_height)) = + storage.get_header_height_by_hash(&work_block_hash).await + { + let cycle_boundary_height = work_block_height + WORK_DIFF_DEPTH; + if let Ok(Some(cycle_boundary_header)) = storage.get_header(cycle_boundary_height).await + { + let cycle_boundary_hash = *cycle_boundary_header.hash(); + engine.feed_block_height(cycle_boundary_height, cycle_boundary_hash); + fed_count += 1; + tracing::debug!( + "Fed cycle boundary height {} for block {}", + cycle_boundary_height, + cycle_boundary_hash + ); + } + } + } + + tracing::info!("Fed {} block heights to engine", fed_count); + fed_count +} diff --git a/dash-spv/src/storage/mod.rs b/dash-spv/src/storage/mod.rs index f4b5c5d40..42ed36b97 100644 --- a/dash-spv/src/storage/mod.rs +++ b/dash-spv/src/storage/mod.rs @@ -34,6 +34,7 @@ pub use crate::storage::block_headers::{ pub use crate::storage::blocks::{BlockStorage, PersistentBlockStorage}; pub use crate::storage::filter_headers::{FilterHeaderStorage, PersistentFilterHeaderStorage}; pub use crate::storage::filters::{FilterStorage, PersistentFilterStorage}; +pub(crate) use crate::storage::masternode::feed_qrinfo_heights_to_engine; pub use crate::storage::masternode::{MasternodeStorage, PersistentMasternodeStorage}; pub use crate::storage::metadata::{MetadataStorage, PersistentMetadataStorage}; pub use crate::storage::peers::{PeerStorage, PersistentPeerStorage}; @@ -80,7 +81,8 @@ pub trait StorageManager: /// Returns shared access to the metadata storage. fn metadata(&self) -> Arc>; - fn masternodes(&self) -> Arc>; + fn masternodes(&self) + -> Arc>>; } /// Disk-based storage manager with segmented files and async background saving. @@ -94,7 +96,7 @@ pub struct DiskStorageManager { filters: Arc>, blocks: Arc>, metadata: Arc>, - masternodes: Arc>, + masternodes: Arc>>, // Background worker worker_handle: Option>, @@ -135,12 +137,12 @@ impl DiskStorageManager { let lock_file = LockFile::new(lock_file)?; + let block_headers = + Arc::new(RwLock::new(PersistentBlockHeaderStorage::open(&storage_path).await?)); + let mut storage = Self { storage_path: storage_path.clone(), - block_headers: Arc::new(RwLock::new( - PersistentBlockHeaderStorage::open(&storage_path).await?, - )), filter_headers: Arc::new(RwLock::new( PersistentFilterHeaderStorage::open(&storage_path).await?, )), @@ -148,8 +150,10 @@ impl DiskStorageManager { blocks: Arc::new(RwLock::new(PersistentBlockStorage::open(&storage_path).await?)), metadata: Arc::new(RwLock::new(PersistentMetadataStorage::open(&storage_path).await?)), masternodes: Arc::new(RwLock::new( - PersistentMasternodeStorage::open(&storage_path).await?, + PersistentMasternodeStorage::open(&storage_path, Arc::clone(&block_headers)) + .await?, )), + block_headers, worker_handle: None, @@ -176,7 +180,6 @@ impl DiskStorageManager { let filters = Arc::clone(&self.filters); let blocks = Arc::clone(&self.blocks); let metadata = Arc::clone(&self.metadata); - let masternodes = Arc::clone(&self.masternodes); let storage_path = self.storage_path.clone(); @@ -191,7 +194,6 @@ impl DiskStorageManager { let _ = filters.write().await.persist(&storage_path).await; let _ = blocks.write().await.persist(&storage_path).await; let _ = metadata.write().await.persist(&storage_path).await; - let _ = masternodes.write().await.persist(&storage_path).await; } }); @@ -213,7 +215,6 @@ 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(storage_path).await; } } @@ -250,8 +251,10 @@ 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).await?)); + self.masternodes = Arc::new(RwLock::new( + PersistentMasternodeStorage::open(storage_path, Arc::clone(&self.block_headers)) + .await?, + )); // Restart the background worker for future operations self.start_worker().await; @@ -286,7 +289,9 @@ impl StorageManager for DiskStorageManager { Arc::clone(&self.metadata) } - fn masternodes(&self) -> Arc> { + fn masternodes( + &self, + ) -> Arc>> { Arc::clone(&self.masternodes) } } @@ -455,8 +460,8 @@ impl masternode::MasternodeStorage for DiskStorageManager { self.masternodes.write().await.store_qr_info(height, qr_info).await } - async fn store_context(&mut self, engine: &MasternodeListEngine) -> StorageResult<()> { - self.masternodes.write().await.store_context(engine).await + async fn store_quorum_statuses(&mut self, engine: &MasternodeListEngine) -> StorageResult<()> { + self.masternodes.write().await.store_quorum_statuses(engine).await } async fn load_engine(&self, network: Network) -> StorageResult { diff --git a/dash-spv/src/sync/chainlock/manager.rs b/dash-spv/src/sync/chainlock/manager.rs index 40a2d8d93..229d99342 100644 --- a/dash-spv/src/sync/chainlock/manager.rs +++ b/dash-spv/src/sync/chainlock/manager.rs @@ -40,7 +40,7 @@ pub struct ChainLockManager { /// Masternode engine for BLS signature validation. masternode_engine: Arc>, /// Rebuilds a masternode list the engine no longer retains. - masternode_storage: Option>>, + masternode_storage: Option>>>, network: Network, /// The best (highest height) validated ChainLock. best_chainlock: Option, @@ -62,7 +62,7 @@ impl ChainLockManager { header_storage: Arc>, metadata_storage: Arc>, masternode_engine: Arc>, - masternode_storage: Option>>, + masternode_storage: Option>>>, network: Network, ) -> Self { let mut manager = Self { diff --git a/dash-spv/src/sync/masternodes/manager.rs b/dash-spv/src/sync/masternodes/manager.rs index 96b012f3e..c3ac259b3 100644 --- a/dash-spv/src/sync/masternodes/manager.rs +++ b/dash-spv/src/sync/masternodes/manager.rs @@ -301,7 +301,7 @@ pub struct MasternodesManager { /// Sync state tracking. pub(super) sync_state: MasternodeSyncState, /// `None` leaves the list in memory only. - pub(super) message_storage: Option>>, + pub(super) message_storage: Option>>>, } impl MasternodesManager { @@ -310,7 +310,7 @@ impl MasternodesManager { header_storage: Arc>, engine: Arc>, network: dashcore::Network, - message_storage: Option>>, + message_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. @@ -363,7 +363,7 @@ impl MasternodesManager { } } - pub(super) async fn persist_context(&self, tip: u32) { + pub(super) async fn prune_and_persist_statuses(&self, tip: u32) { let Some(storage) = &self.message_storage else { return; }; @@ -371,12 +371,14 @@ impl MasternodesManager { let mut engine = self.engine.write().await; let pruned = engine.prune_masternode_lists(tip); - if let Err(e) = storage.write().await.store_context(&engine).await { - tracing::warn!("Could not persist masternode context at {tip}: {e}"); + if let Err(e) = storage.write().await.store_quorum_statuses(&engine).await { + tracing::warn!("Could not persist masternode quorum statuses at {tip}: {e}"); return; } - tracing::debug!("Persisted masternode context at {tip}, pruned {pruned} in-memory lists"); + tracing::debug!( + "Persisted masternode quorum statuses at {tip}, pruned {pruned} in-memory lists" + ); } /// Decide which [`PipelineMode`] to use when a new header lands at `tip_height` @@ -598,7 +600,7 @@ impl MasternodesManager { self.sync_state.last_synced_block_hash = Some(latest_block_hash); self.progress.update_current_height(height); - self.persist_context(height).await; + self.prune_and_persist_statuses(height).await; tracing::debug!("Incremental MnListDiff complete at height {}", height); Ok(vec![SyncEvent::MasternodeStateUpdated { height, @@ -703,7 +705,7 @@ impl MasternodesManager { drop(engine); if !events.is_empty() { - self.persist_context(self.progress.current_height()).await; + self.prune_and_persist_statuses(self.progress.current_height()).await; } if is_initial_sync { diff --git a/dash-spv/src/sync/masternodes/sync_manager.rs b/dash-spv/src/sync/masternodes/sync_manager.rs index ae16cf6bf..eaddb698e 100644 --- a/dash-spv/src/sync/masternodes/sync_manager.rs +++ b/dash-spv/src/sync/masternodes/sync_manager.rs @@ -1,15 +1,13 @@ use super::manager::PipelineMode; use crate::error::SyncResult; use crate::network::{Message, MessageType, RequestSender}; -use crate::storage::BlockHeaderStorage; +use crate::storage::{feed_qrinfo_heights_to_engine, BlockHeaderStorage}; use crate::sync::{ ManagerIdentifier, MasternodesManager, SyncEvent, SyncManager, SyncManagerProgress, SyncState, }; use crate::SyncError; use async_trait::async_trait; use dashcore::network::message::NetworkMessage; -use dashcore::network::message_qrinfo::QRInfo; -use dashcore::sml::masternode_list_engine::{MasternodeListEngine, WORK_DIFF_DEPTH}; use dashcore::{BlockHash, QuorumHash}; use dashcore_hashes::Hash; use std::collections::{BTreeSet, HashSet}; @@ -159,63 +157,6 @@ pub(super) async fn build_mnlistdiff_request_pairs( Ok(pairs_with_height.into_iter().map(|(_, base, target)| (base, target)).collect()) } -/// Feed QRInfo block heights to the engine from storage. -/// -/// Resolves heights for every hash enumerated by -/// [`MasternodeListEngine::qr_info_referenced_block_hashes`], plus the cycle boundary -/// block for each work-block diff (`work_height + WORK_DIFF_DEPTH`), which is needed -/// for rotated quorum storage key calculation. -pub(super) async fn feed_qrinfo_heights_to_engine( - engine: &mut MasternodeListEngine, - qr_info: &QRInfo, - storage: &S, -) -> SyncResult { - let mut fed_count = 0; - for block_hash in MasternodeListEngine::qr_info_referenced_block_hashes(qr_info) { - if let Ok(Some(height)) = storage.get_header_height_by_hash(&block_hash).await { - engine.feed_block_height(height, block_hash); - fed_count += 1; - tracing::trace!("Fed height {} for block {}", height, block_hash); - } - } - - // Feed cycle boundary heights for all diffs (current and historical cycles). - // Each diff's block_hash is at the "work block" height; the cycle boundary is - // WORK_DIFF_DEPTH higher. - let mut work_block_hashes = vec![ - qr_info.mn_list_diff_h.block_hash, - qr_info.mn_list_diff_at_h_minus_c.block_hash, - qr_info.mn_list_diff_at_h_minus_2c.block_hash, - qr_info.mn_list_diff_at_h_minus_3c.block_hash, - ]; - - if let Some((_, diff)) = &qr_info.quorum_snapshot_and_mn_list_diff_at_h_minus_4c { - work_block_hashes.push(diff.block_hash); - } - - for work_block_hash in work_block_hashes { - if let Ok(Some(work_block_height)) = - storage.get_header_height_by_hash(&work_block_hash).await - { - let cycle_boundary_height = work_block_height + WORK_DIFF_DEPTH; - if let Ok(Some(cycle_boundary_header)) = storage.get_header(cycle_boundary_height).await - { - let cycle_boundary_hash = *cycle_boundary_header.hash(); - engine.feed_block_height(cycle_boundary_height, cycle_boundary_hash); - fed_count += 1; - tracing::debug!( - "Fed cycle boundary height {} for block {}", - cycle_boundary_height, - cycle_boundary_hash - ); - } - } - } - - tracing::info!("Fed {} block heights to engine", fed_count); - Ok(fed_count) -} - #[async_trait] impl SyncManager for MasternodesManager { fn identifier(&self) -> ManagerIdentifier { @@ -269,7 +210,7 @@ impl SyncManager for MasternodesManager { // Feed block heights to engine using internal storage let storage = self.header_storage.read().await; let mut engine = self.engine.write().await; - let fed = feed_qrinfo_heights_to_engine(&mut engine, qr_info, &*storage).await?; + let fed = feed_qrinfo_heights_to_engine(&mut engine, qr_info, &*storage).await; drop(storage); tracing::info!("Fed {} block heights to engine", fed); @@ -762,14 +703,13 @@ impl SyncManager for MasternodesManager { mod tests { use super::super::manager::{MasternodeSyncState, QRInfoInFlight}; use super::{ - feed_qrinfo_heights_to_engine, qrinfo_timeout_for, MAX_RETRY_ATTEMPTS, - QRINFO_STALL_WATCHDOG, QRINFO_TIMEOUT_SCHEDULE_SECS, + qrinfo_timeout_for, MAX_RETRY_ATTEMPTS, QRINFO_STALL_WATCHDOG, QRINFO_TIMEOUT_SCHEDULE_SECS, }; use crate::error::StorageResult; use crate::network::{Message, NetworkRequest, RequestSender}; use crate::storage::{ - BlockHeaderStorage, BlockHeaderTip, DiskStorageManager, PersistentBlockHeaderStorage, - StorageManager, + feed_qrinfo_heights_to_engine, BlockHeaderStorage, BlockHeaderTip, DiskStorageManager, + PersistentBlockHeaderStorage, StorageManager, }; use crate::sync::{MasternodesManager, SyncManager, SyncState}; use crate::types::HashedBlockHeader; @@ -935,9 +875,7 @@ mod tests { network: Network::Testnet, ..Default::default() }; - feed_qrinfo_heights_to_engine(&mut engine, &qr_info, &MockHeaderStorage(height_map)) - .await - .unwrap(); + feed_qrinfo_heights_to_engine(&mut engine, &qr_info, &MockHeaderStorage(height_map)).await; for &b in expected_hashes { let hash = BlockHash::from_slice(&[b; 32]).unwrap(); From 875ce3269e9fde7b1bb451961a49f4f49e630d16 Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Tue, 1 Sep 2026 09:39:32 +0000 Subject: [PATCH 6/7] api cleanup --- dash-spv/src/client/lifecycle.rs | 2 - dash-spv/src/storage/masternode.rs | 124 +++++++++++------- dash-spv/src/sync/chainlock/manager.rs | 11 +- dash-spv/src/sync/masternodes/manager.rs | 1 - dash-spv/src/sync/masternodes/sync_manager.rs | 3 +- dash-spv/tests/dashd_masternode/helpers.rs | 15 --- dash-spv/tests/dashd_masternode/tests_sync.rs | 6 - .../src/sml/masternode_list_engine/helpers.rs | 13 +- 8 files changed, 84 insertions(+), 91 deletions(-) diff --git a/dash-spv/src/client/lifecycle.rs b/dash-spv/src/client/lifecycle.rs index 7dd9968e1..37904d9c3 100644 --- a/dash-spv/src/client/lifecycle.rs +++ b/dash-spv/src/client/lifecycle.rs @@ -66,8 +66,6 @@ impl DashSpvClient; + +struct CachedList { + from: CoreBlockHeight, + until: Option, + list: Option, +} + type QuorumStatuses = BTreeMap< LLMQType, BTreeMap, BLSPublicKey, LLMQEntryVerificationStatus)>, @@ -46,17 +54,12 @@ pub trait MasternodeStorage: Send + Sync + 'static { ) -> StorageResult>; } -/// Stores the raw messages a masternode list is rebuilt from. The heights the -/// replay needs come from the header storage rather than from a persisted copy -/// of the engine's block container: the headers already hold every hash/height -/// pair, and a second copy can only go stale against them. pub struct PersistentMasternodeStorage { storage_path: PathBuf, - /// Shared with whoever writes the headers, so a replay reads them as they - /// stand rather than as they were when this storage was opened. headers: Arc>, - diffs: BTreeMap, - qr_infos: BTreeMap, + diffs: IndexMap, + qr_infos: IndexMap, + cached_list: Mutex>, } impl PersistentMasternodeStorage { @@ -78,6 +81,7 @@ impl PersistentMasternodeStorage { headers, diffs, qr_infos, + cached_list: Mutex::new(None), }) } @@ -117,17 +121,21 @@ impl PersistentMasternodeStorage { Ok((diffs, qr_infos)) } - async fn write_message( - &mut self, + async fn store_message( + folder: &Path, + index: &mut IndexMap, prefix: &str, height: CoreBlockHeight, message: &T, - ) -> StorageResult { - let folder = self.folder(); - tokio::fs::create_dir_all(&folder).await?; + ) -> StorageResult<()> { + if index.contains_key(&height) { + return Ok(()); + } + tokio::fs::create_dir_all(folder).await?; let path = folder.join(Self::file_name(prefix, height)); atomic_write(&path, &serialize(message)).await?; - Ok(path) + index.insert(height, path); + Ok(()) } async fn read_message(path: &Path) -> StorageResult { @@ -137,6 +145,20 @@ impl PersistentMasternodeStorage { }) } + async fn load_pending( + index: &IndexMap, + label: &str, + ) -> Vec<(CoreBlockHeight, T)> { + let mut pending = Vec::new(); + for (height, path) in index { + match Self::read_message::(path).await { + Ok(message) => pending.push((*height, message)), + Err(e) => tracing::warn!("Skipping unreadable {label} at {height}: {e}"), + } + } + pending + } + async fn load_quorum_statuses(&self) -> StorageResult> { let path = self.folder().join(Self::QUORUM_STATUSES_FILE_NAME); if !path.exists() { @@ -145,8 +167,6 @@ impl PersistentMasternodeStorage { let bytes = tokio::fs::read(&path).await?; match bincode::decode_from_slice(&bytes, bincode::config::standard()) { Ok((statuses, _)) => Ok(Some(statuses)), - // The statuses are a cache of what the replay re-derives, so a file - // written by an older format costs verification work, not the sync. Err(e) => { tracing::warn!("Ignoring undecodable masternode quorum statuses: {e}"); Ok(None) @@ -154,6 +174,17 @@ impl PersistentMasternodeStorage { } } + async fn cached_list_at(&self, height: CoreBlockHeight) -> Option> { + let cached = self.cached_list.lock().await; + let cached = cached.as_ref()?; + (height >= cached.from && cached.until.is_none_or(|until| height < until)) + .then(|| cached.list.clone()) + } + + fn invalidate_cached_list(&mut self) { + *self.cached_list.get_mut() = None; + } + async fn replay(&self, network: Network) -> StorageResult { let mut engine = MasternodeListEngine::default_for_network(network); @@ -161,29 +192,16 @@ impl PersistentMasternodeStorage { engine.quorum_statuses = quorum_statuses; } - let mut pending_qr_infos = Vec::new(); - for (height, path) in &self.qr_infos { - match Self::read_message::(path).await { - Ok(qr_info) => pending_qr_infos.push((*height, qr_info)), - Err(e) => tracing::warn!("Skipping unreadable QRInfo at {height}: {e}"), - } - } - - let mut pending_diffs = Vec::new(); - for (height, path) in &self.diffs { - match Self::read_message::(path).await { - Ok(diff) => pending_diffs.push((*height, diff)), - Err(e) => tracing::warn!("Skipping unreadable MnListDiff at {height}: {e}"), - } - } + let mut pending_qr_infos: Vec<(CoreBlockHeight, QRInfo)> = + Self::load_pending(&self.qr_infos, "QRInfo").await; + let mut pending_diffs: Vec<(CoreBlockHeight, MnListDiff)> = + Self::load_pending(&self.diffs, "MnListDiff").await; { let headers = self.headers.read().await; for (_, qr_info) in &pending_qr_infos { feed_qrinfo_heights_to_engine(&mut engine, qr_info, &*headers).await; } - // A diff is keyed by its own height, but applying it also needs the - // height of the list it extends. for (height, diff) in &pending_diffs { engine.feed_block_height(*height, diff.block_hash); if let Ok(Some(base_height)) = @@ -235,8 +253,6 @@ impl PersistentMasternodeStorage { } } -type IndexMap = BTreeMap; - #[async_trait] impl MasternodeStorage for PersistentMasternodeStorage { async fn store_diff( @@ -244,12 +260,9 @@ impl MasternodeStorage for PersistentMasternodeStorage height: CoreBlockHeight, diff: &MnListDiff, ) -> StorageResult<()> { - if self.diffs.contains_key(&height) { - return Ok(()); - } - let path = self.write_message(Self::DIFF_PREFIX, height, diff).await?; - self.diffs.insert(height, path); - Ok(()) + let folder = self.folder(); + self.invalidate_cached_list(); + Self::store_message(&folder, &mut self.diffs, Self::DIFF_PREFIX, height, diff).await } async fn store_qr_info( @@ -257,12 +270,9 @@ impl MasternodeStorage for PersistentMasternodeStorage height: CoreBlockHeight, qr_info: &QRInfo, ) -> StorageResult<()> { - if self.qr_infos.contains_key(&height) { - return Ok(()); - } - let path = self.write_message(Self::QRINFO_PREFIX, height, qr_info).await?; - self.qr_infos.insert(height, path); - Ok(()) + let folder = self.folder(); + self.invalidate_cached_list(); + Self::store_message(&folder, &mut self.qr_infos, Self::QRINFO_PREFIX, height, qr_info).await } async fn store_quorum_statuses(&mut self, engine: &MasternodeListEngine) -> StorageResult<()> { @@ -286,8 +296,21 @@ impl MasternodeStorage for PersistentMasternodeStorage network: Network, height: CoreBlockHeight, ) -> StorageResult> { + if let Some(hit) = self.cached_list_at(height).await { + return Ok(hit); + } + let engine = self.replay(network).await?; - Ok(engine.masternode_lists_around_height(height).0.cloned()) + let (before, after) = engine.masternode_lists_around_height(height); + let list = before.cloned(); + + *self.cached_list.lock().await = Some(CachedList { + from: before.map_or(0, |list| list.known_height), + until: after.map(|next| next.known_height), + list: list.clone(), + }); + + Ok(list) } } @@ -301,7 +324,7 @@ pub(crate) async fn feed_qrinfo_heights_to_engine( engine: &mut MasternodeListEngine, qr_info: &QRInfo, storage: &S, -) -> usize { +) { let mut fed_count = 0; for block_hash in MasternodeListEngine::qr_info_referenced_block_hashes(qr_info) { if let Ok(Some(height)) = storage.get_header_height_by_hash(&block_hash).await { @@ -345,5 +368,4 @@ pub(crate) async fn feed_qrinfo_heights_to_engine( } tracing::info!("Fed {} block heights to engine", fed_count); - fed_count } diff --git a/dash-spv/src/sync/chainlock/manager.rs b/dash-spv/src/sync/chainlock/manager.rs index 229d99342..0cd868c24 100644 --- a/dash-spv/src/sync/chainlock/manager.rs +++ b/dash-spv/src/sync/chainlock/manager.rs @@ -39,7 +39,6 @@ pub struct ChainLockManager { metadata_storage: Arc>, /// Masternode engine for BLS signature validation. masternode_engine: Arc>, - /// Rebuilds a masternode list the engine no longer retains. masternode_storage: Option>>>, network: Network, /// The best (highest height) validated ChainLock. @@ -283,10 +282,12 @@ impl ChainLockManager { }; let signing_height = chainlock.block_height.saturating_sub(8); - let list = - storage.read().await.masternode_list_at_or_before(self.network, signing_height).await; - - let list = match list { + let list = match storage + .read() + .await + .masternode_list_at_or_before(self.network, signing_height) + .await + { Ok(Some(list)) => list, Ok(None) => return false, Err(e) => { diff --git a/dash-spv/src/sync/masternodes/manager.rs b/dash-spv/src/sync/masternodes/manager.rs index c3ac259b3..12aca7eac 100644 --- a/dash-spv/src/sync/masternodes/manager.rs +++ b/dash-spv/src/sync/masternodes/manager.rs @@ -300,7 +300,6 @@ pub struct MasternodesManager { network: dashcore::Network, /// Sync state tracking. pub(super) sync_state: MasternodeSyncState, - /// `None` leaves the list in memory only. pub(super) message_storage: Option>>>, } diff --git a/dash-spv/src/sync/masternodes/sync_manager.rs b/dash-spv/src/sync/masternodes/sync_manager.rs index eaddb698e..3d81236ac 100644 --- a/dash-spv/src/sync/masternodes/sync_manager.rs +++ b/dash-spv/src/sync/masternodes/sync_manager.rs @@ -210,9 +210,8 @@ impl SyncManager for MasternodesManager { // Feed block heights to engine using internal storage let storage = self.header_storage.read().await; let mut engine = self.engine.write().await; - let fed = feed_qrinfo_heights_to_engine(&mut engine, qr_info, &*storage).await; + feed_qrinfo_heights_to_engine(&mut engine, qr_info, &*storage).await; drop(storage); - tracing::info!("Fed {} block heights to engine", fed); // Feed QRInfo to engine first to populate masternode lists let qr_info_result = match engine.feed_qr_info(qr_info.clone(), true, true) { diff --git a/dash-spv/tests/dashd_masternode/helpers.rs b/dash-spv/tests/dashd_masternode/helpers.rs index d981af7a1..d4b6d0269 100644 --- a/dash-spv/tests/dashd_masternode/helpers.rs +++ b/dash-spv/tests/dashd_masternode/helpers.rs @@ -17,12 +17,6 @@ use super::setup::{TestContext, SYNC_TIMEOUT}; /// Mine a DKG cycle and wait for the SPV to surface a `MasternodeStateUpdated` /// event above `baseline_height`. -/// Files held under each immediate subdirectory of the storage root, keyed by -/// directory name. -/// -/// A sync writes into these and never removes a whole class of state, so across -/// a restart every directory must still be there and hold at least as much — -/// see [`assert_storage_did_not_shrink`]. pub(super) fn storage_snapshot(root: &Path) -> BTreeMap { let mut counts = BTreeMap::new(); let Ok(entries) = std::fs::read_dir(root) else { @@ -55,10 +49,6 @@ fn walkdir_count(dir: &Path) -> usize { .sum() } -/// Directories that must hold state once this test's first session has run, and -/// why. `filters` and `blocks` are deliberately absent: the client is stopped -/// as soon as the masternode phase reports `Synced`, which is before the filter -/// phase leaves `WaitForEvents`, so those stay legitimately empty here. pub(super) const EXPECTED_STORAGE: &[(&str, &str)] = &[ ("block_headers", "headers synced to the tip"), ("filter_headers", "filter headers synced to the tip"), @@ -67,8 +57,6 @@ pub(super) const EXPECTED_STORAGE: &[(&str, &str)] = &[ ("masternodes", "the masternode messages this session stored"), ]; -/// Assert every directory in [`EXPECTED_STORAGE`] exists and holds at least one -/// file, reporting all of them at once rather than the first to fail. pub(super) fn assert_storage_persisted(snapshot: &BTreeMap, what: &str) { let missing: Vec = EXPECTED_STORAGE .iter() @@ -84,9 +72,6 @@ pub(super) fn assert_storage_persisted(snapshot: &BTreeMap, what: ); } -/// Every directory present before a restart must still be present after, with -/// at least as many files. A directory that vanishes or shrinks means a restart -/// threw away state that the previous session had already earned. pub(super) fn assert_storage_did_not_shrink( before: &BTreeMap, after: &BTreeMap, diff --git a/dash-spv/tests/dashd_masternode/tests_sync.rs b/dash-spv/tests/dashd_masternode/tests_sync.rs index b38d28783..1647b9193 100644 --- a/dash-spv/tests/dashd_masternode/tests_sync.rs +++ b/dash-spv/tests/dashd_masternode/tests_sync.rs @@ -105,8 +105,6 @@ async fn test_masternode_list_sync_with_restart() { wait_for_masternode_sync(&mut client_handle.progress_receiver, SYNC_TIMEOUT).await; let first_height = first_mn_progress.current_height(); - // Control: the first session really built a list, so the persistence - // assertion below cannot be satisfied by a client that synced nothing. let first_masternodes = { let engine = client_handle.engine.read().await; engine.masternode_lists.values().map(|list| list.masternodes.len()).max().unwrap_or(0) @@ -119,8 +117,6 @@ async fn test_masternode_list_sync_with_restart() { client_handle.stop().await; drop(client_handle); - // What the first session earned and wrote down. A clean shutdown of a - // fully-synced client must leave every sync phase's state on disk. let after_first = storage_snapshot(ctx.storage_path()); assert_storage_persisted( &after_first, @@ -144,8 +140,6 @@ async fn test_masternode_list_sync_with_restart() { "Should reach Synced state after restart" ); - // A restart re-syncs on top of what it restored; it never discards a whole - // class of state it already had. let after_second = storage_snapshot(ctx.storage_path()); assert_storage_did_not_shrink(&after_first, &after_second, "masternode restart"); diff --git a/dash/src/sml/masternode_list_engine/helpers.rs b/dash/src/sml/masternode_list_engine/helpers.rs index be72bc3e4..b226bea80 100644 --- a/dash/src/sml/masternode_list_engine/helpers.rs +++ b/dash/src/sml/masternode_list_engine/helpers.rs @@ -13,23 +13,18 @@ use crate::sml::quorum_entry::qualified_quorum_entry::QualifiedQuorumEntry; /// height that can exceed one active window (Platform selects roughly 4.5 DKG intervals back), so a /// single window is too tight. Four windows covers that lag with wide margin while still bounding a /// miss to a fixed span of lists rather than every list the engine has accumulated. -pub const QUORUM_WALK_BACK_ACTIVE_WINDOWS: u32 = 4; +const QUORUM_WALK_BACK_ACTIVE_WINDOWS: u32 = 4; impl MasternodeListEngine { #[cfg(feature = "quorum_validation")] - pub fn retained_list_floor(&self, tip: CoreBlockHeight) -> CoreBlockHeight { + pub fn prune_masternode_lists(&mut self, tip: CoreBlockHeight) -> usize { let params = self.network.chain_locks_type().params(); - tip.saturating_sub( + let floor = tip.saturating_sub( params .signing_active_quorum_count .saturating_mul(params.dkg_params.interval) .saturating_mul(QUORUM_WALK_BACK_ACTIVE_WINDOWS), - ) - } - - #[cfg(feature = "quorum_validation")] - pub fn prune_masternode_lists(&mut self, tip: CoreBlockHeight) -> usize { - let floor = self.retained_list_floor(tip); + ); let before = self.masternode_lists.len(); self.masternode_lists.retain(|height, _| *height >= floor); before - self.masternode_lists.len() From 5d1aebff29b5c8a78e937e5869efbb6ad4e75ef7 Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Tue, 1 Sep 2026 11:01:37 +0000 Subject: [PATCH 7/7] removed quorum statuses persistence --- dash-spv/Cargo.toml | 1 - dash-spv/src/storage/masternode.rs | 46 ++---------------------- dash-spv/src/storage/mod.rs | 4 --- dash-spv/src/sync/masternodes/manager.rs | 21 ++++------- 4 files changed, 8 insertions(+), 64 deletions(-) diff --git a/dash-spv/Cargo.toml b/dash-spv/Cargo.toml index d419d9978..0c80f5b36 100644 --- a/dash-spv/Cargo.toml +++ b/dash-spv/Cargo.toml @@ -33,7 +33,6 @@ thiserror = "1.0" # Serialization serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" -bincode = "2.0.1" # Logging tracing = "0.1" diff --git a/dash-spv/src/storage/masternode.rs b/dash-spv/src/storage/masternode.rs index ea8cd87ec..2e6e144aa 100644 --- a/dash-spv/src/storage/masternode.rs +++ b/dash-spv/src/storage/masternode.rs @@ -1,20 +1,17 @@ -use std::collections::{BTreeMap, BTreeSet}; +use std::collections::BTreeMap; use std::path::{Path, PathBuf}; use std::sync::Arc; use async_trait::async_trait; use tokio::sync::{Mutex, RwLock}; -use dashcore::bls_sig_utils::BLSPublicKey; use dashcore::consensus::{deserialize, serialize, Decodable, Encodable}; use dashcore::network::message_qrinfo::QRInfo; use dashcore::network::message_sml::MnListDiff; use dashcore::prelude::CoreBlockHeight; -use dashcore::sml::llmq_entry_verification::LLMQEntryVerificationStatus; -use dashcore::sml::llmq_type::LLMQType; use dashcore::sml::masternode_list::MasternodeList; use dashcore::sml::masternode_list_engine::{MasternodeListEngine, WORK_DIFF_DEPTH}; -use dashcore::{Network, QuorumHash}; +use dashcore::Network; use crate::error::{StorageError, StorageResult}; use crate::storage::{io::atomic_write, BlockHeaderStorage}; @@ -27,11 +24,6 @@ struct CachedList { list: Option, } -type QuorumStatuses = BTreeMap< - LLMQType, - BTreeMap, BLSPublicKey, LLMQEntryVerificationStatus)>, ->; - #[async_trait] pub trait MasternodeStorage: Send + Sync + 'static { async fn store_diff(&mut self, height: CoreBlockHeight, diff: &MnListDiff) @@ -43,8 +35,6 @@ pub trait MasternodeStorage: Send + Sync + 'static { qr_info: &QRInfo, ) -> StorageResult<()>; - async fn store_quorum_statuses(&mut self, engine: &MasternodeListEngine) -> StorageResult<()>; - async fn load_engine(&self, network: Network) -> StorageResult; async fn masternode_list_at_or_before( @@ -67,7 +57,6 @@ impl PersistentMasternodeStorage { const DIFF_PREFIX: &str = "diff_"; const QRINFO_PREFIX: &str = "qrinfo_"; const EXTENSION: &str = "dat"; - const QUORUM_STATUSES_FILE_NAME: &str = "quorum_statuses.dat"; pub async fn open( storage_path: impl Into + Send, @@ -159,21 +148,6 @@ impl PersistentMasternodeStorage { pending } - async fn load_quorum_statuses(&self) -> StorageResult> { - let path = self.folder().join(Self::QUORUM_STATUSES_FILE_NAME); - if !path.exists() { - return Ok(None); - } - let bytes = tokio::fs::read(&path).await?; - match bincode::decode_from_slice(&bytes, bincode::config::standard()) { - Ok((statuses, _)) => Ok(Some(statuses)), - Err(e) => { - tracing::warn!("Ignoring undecodable masternode quorum statuses: {e}"); - Ok(None) - } - } - } - async fn cached_list_at(&self, height: CoreBlockHeight) -> Option> { let cached = self.cached_list.lock().await; let cached = cached.as_ref()?; @@ -188,10 +162,6 @@ impl PersistentMasternodeStorage { async fn replay(&self, network: Network) -> StorageResult { let mut engine = MasternodeListEngine::default_for_network(network); - if let Some(quorum_statuses) = self.load_quorum_statuses().await? { - engine.quorum_statuses = quorum_statuses; - } - let mut pending_qr_infos: Vec<(CoreBlockHeight, QRInfo)> = Self::load_pending(&self.qr_infos, "QRInfo").await; let mut pending_diffs: Vec<(CoreBlockHeight, MnListDiff)> = @@ -275,18 +245,6 @@ impl MasternodeStorage for PersistentMasternodeStorage Self::store_message(&folder, &mut self.qr_infos, Self::QRINFO_PREFIX, height, qr_info).await } - async fn store_quorum_statuses(&mut self, engine: &MasternodeListEngine) -> StorageResult<()> { - let folder = self.folder(); - tokio::fs::create_dir_all(&folder).await?; - - let bytes = bincode::encode_to_vec(&engine.quorum_statuses, bincode::config::standard()) - .map_err(|e| { - StorageError::Serialization(format!("Failed to encode quorum statuses: {e}")) - })?; - - atomic_write(&folder.join(Self::QUORUM_STATUSES_FILE_NAME), &bytes).await - } - async fn load_engine(&self, network: Network) -> StorageResult { self.replay(network).await } diff --git a/dash-spv/src/storage/mod.rs b/dash-spv/src/storage/mod.rs index 42ed36b97..8a76c463c 100644 --- a/dash-spv/src/storage/mod.rs +++ b/dash-spv/src/storage/mod.rs @@ -460,10 +460,6 @@ impl masternode::MasternodeStorage for DiskStorageManager { self.masternodes.write().await.store_qr_info(height, qr_info).await } - async fn store_quorum_statuses(&mut self, engine: &MasternodeListEngine) -> StorageResult<()> { - self.masternodes.write().await.store_quorum_statuses(engine).await - } - async fn load_engine(&self, network: Network) -> StorageResult { self.masternodes.read().await.load_engine(network).await } diff --git a/dash-spv/src/sync/masternodes/manager.rs b/dash-spv/src/sync/masternodes/manager.rs index 12aca7eac..234cf5ca7 100644 --- a/dash-spv/src/sync/masternodes/manager.rs +++ b/dash-spv/src/sync/masternodes/manager.rs @@ -362,22 +362,13 @@ impl MasternodesManager { } } - pub(super) async fn prune_and_persist_statuses(&self, tip: u32) { - let Some(storage) = &self.message_storage else { - return; - }; - - let mut engine = self.engine.write().await; - let pruned = engine.prune_masternode_lists(tip); - - if let Err(e) = storage.write().await.store_quorum_statuses(&engine).await { - tracing::warn!("Could not persist masternode quorum statuses at {tip}: {e}"); + pub(super) async fn prune_retained_lists(&self, tip: u32) { + if self.message_storage.is_none() { return; } - tracing::debug!( - "Persisted masternode quorum statuses at {tip}, pruned {pruned} in-memory lists" - ); + let pruned = self.engine.write().await.prune_masternode_lists(tip); + tracing::debug!("Pruned {pruned} in-memory masternode lists at {tip}"); } /// Decide which [`PipelineMode`] to use when a new header lands at `tip_height` @@ -599,7 +590,7 @@ impl MasternodesManager { self.sync_state.last_synced_block_hash = Some(latest_block_hash); self.progress.update_current_height(height); - self.prune_and_persist_statuses(height).await; + self.prune_retained_lists(height).await; tracing::debug!("Incremental MnListDiff complete at height {}", height); Ok(vec![SyncEvent::MasternodeStateUpdated { height, @@ -704,7 +695,7 @@ impl MasternodesManager { drop(engine); if !events.is_empty() { - self.prune_and_persist_statuses(self.progress.current_height()).await; + self.prune_retained_lists(self.progress.current_height()).await; } if is_initial_sync {