From 70fbfb304de9b70980fef3086b43fb596948525e Mon Sep 17 00:00:00 2001 From: xdustinface Date: Thu, 9 Jul 2026 10:56:07 +1000 Subject: [PATCH 1/4] feat(dash-spv): add `force_resync` to re-anchor a running client Add `force_resync`, which wipes all chain storage and re-syncs from the checkpoint at or below the current minimum wallet birth height, plus `resync_needed`, which reports when a wallet sits below the anchored header start and a resync is required to scan its history. Wallet state (per-wallet synced heights and discovered outputs) lives outside chain storage and is preserved, so only chain data is refetched: an added wallet with a lower birth height drags the anchor down and gets its range scanned, while wallets already ahead keep their progress. Making an already-running client resync in place required teardown and rebuild across layers that were previously single-use: - The network manager is now self-arming across a reconnect. The request processor returns its receiver to the shared slot when it shuts down, so the request channel and the sender clones managers hold are never rebuilt and stay valid, and `connect` re-arms the shutdown token that `disconnect` cancelled. A second `connect` therefore behaves like a fresh start with no separate reset step. - `SyncCoordinator::reset` rebuilds the managers in place while keeping the progress and sync-event senders, so existing subscribers stay connected. Replacing the whole coordinator would drop those senders and silently freeze every progress and event stream. - `force_resync` tears down and restarts the coordinator and network directly instead of via `stop`/`start`, leaving the `running` flag set. Flipping it would make the `run` monitoring loop exit for good, freezing all outward progress even though sync continues internally. The block-headers manager also polls its tip once on the sync-completion edge: the peer only announces a new block when it connects to the tip it believes we already have, so a block that lands while we were still syncing is never pushed to us, and without the poll a freshly resynced client can miss it until the next block arrives. Covered by a node-free unit test for the re-anchor and detection logic and a dashd integration test that resyncs a running client, mines a boundary block, and asserts the client keeps tracking the chain with wallet state preserved. --- dash-spv/src/client/lifecycle.rs | 174 ++++++++++++++---- dash-spv/src/client/mod.rs | 59 +++++- dash-spv/src/network/manager.rs | 16 ++ dash-spv/src/network/mod.rs | 3 +- dash-spv/src/sync/block_headers/manager.rs | 6 + dash-spv/src/sync/sync_coordinator.rs | 28 ++- .../tests/dashd_sync/tests_multi_wallet.rs | 74 ++++++++ 7 files changed, 317 insertions(+), 43 deletions(-) diff --git a/dash-spv/src/client/lifecycle.rs b/dash-spv/src/client/lifecycle.rs index 2f789bdf8..3904474be 100644 --- a/dash-spv/src/client/lifecycle.rs +++ b/dash-spv/src/client/lifecycle.rs @@ -46,9 +46,69 @@ impl DashSpvClient + let storage = Arc::new(Mutex::new(storage)); + + let client = Self { + config: Arc::new(RwLock::new(config)), + network: Arc::new(Mutex::new(network)), + storage, + wallet, + masternode_engine, + sync_coordinator: Arc::new(Mutex::new(sync_coordinator)), + running: Arc::new(watch::Sender::new(false)), + event_handlers: Arc::new(event_handlers), + }; + + // Load wallet data from storage + client.load_wallet_data().await?; + + // Emit initial progress so callers get immediate feedback + let initial_progress = client.sync_coordinator.lock().await.progress().clone(); + for event_handler in client.event_handlers.iter() { + event_handler.on_progress(&initial_progress); + } + + Ok(client) + } + + /// Resolve the anchor height, initialize the genesis or checkpoint header in storage, + /// and construct every sync manager against storage. + /// + /// The caller owns `storage`. When it has just been cleared this re-anchors from scratch, + /// otherwise `initialize_genesis_block` sees the existing tip and leaves the chain in place. + async fn build_managers( + config: &ClientConfig, + storage: &mut S, + wallet: &Arc>, + masternode_engine: &Option>>, + ) -> Result< + Managers< + PersistentBlockHeaderStorage, + PersistentFilterHeaderStorage, + PersistentFilterStorage, + PersistentBlockStorage, + PersistentMetadataStorage, + W, + >, + > { + // An explicit `start_from_height` always wins. Otherwise fall back to the wallet + // birth height so we don't sync headers and filter headers from genesis when the + // wallet only cares about recent blocks. let start_from_height = match config.start_from_height { Some(height) => Some(height), None => { @@ -59,17 +119,7 @@ impl DashSpvClient DashSpvClient - let storage = Arc::new(Mutex::new(storage)); + /// Whether the current wallet set needs a resync: some wallet's birth height is below + /// the anchored header start, so its earlier history cannot be scanned without + /// re-anchoring lower. Returns false when nothing is anchored yet or every wallet is + /// already within the synced range. + pub async fn resync_needed(&self) -> bool { + let required = self.wallet.read().await.earliest_required_height().await; + if required == 0 { + return false; + } - let client = Self { - config: Arc::new(RwLock::new(config)), - network: Arc::new(Mutex::new(network)), - storage, - wallet, - masternode_engine, - sync_coordinator: Arc::new(Mutex::new(sync_coordinator)), - running: Arc::new(watch::Sender::new(false)), - event_handlers: Arc::new(event_handlers), - }; + match self.storage.lock().await.get_start_height().await { + Some(header_start) => required < header_start, + None => false, + } + } - // Load wallet data from storage - client.load_wallet_data().await?; + /// Force a full resync from the checkpoint at or below the current minimum wallet birth + /// height. Tears down sync, wipes all chain storage, and rebuilds the sync managers so the + /// next sync re-anchors from scratch. + /// + /// Wallet state (per-wallet synced heights and discovered outputs) lives outside chain + /// storage and is preserved, so only chain data is refetched: an added wallet with a + /// lower birth height drags the anchor down and gets its range scanned, while wallets + /// that were already ahead keep their progress and are not rescanned below it. + /// + /// Restores the running state it was called in. A running client keeps running across the + /// resync. A stopped client stays stopped with its managers rebuilt against the cleared + /// storage. + /// + /// The teardown and restart are done directly on the coordinator and network rather than + /// through `stop`/`start`, and the coordinator is reset in place rather than replaced. This + /// keeps the `running` flag set and the coordinator's event and progress senders alive, so + /// the `run` monitoring loop and every progress and event subscriber stay connected across + /// the resync instead of tearing down. Flipping `running` would make `run` exit and never + /// come back, silently freezing all outward progress. + pub async fn force_resync(&self) -> Result<()> { + let was_running = self.is_running(); + + // Tear down the running sync without touching `running`, so the `run` loop keeps + // monitoring. Drain the coordinator's manager tasks and drop the network connection. + if was_running { + self.sync_coordinator.lock().await.shutdown().await.map_err(SpvError::Sync)?; + self.network.lock().await.disconnect().await?; + } - // Emit initial progress so callers get immediate feedback - let initial_progress = client.sync_coordinator.lock().await.progress().clone(); - for event_handler in client.event_handlers.iter() { - event_handler.on_progress(&initial_progress); + // Masternode lists are chain-derived and wiped below, so reset the shared engine so it + // doesn't retain state from before the resync. + if let Some(engine) = &self.masternode_engine { + let network = self.config.read().await.network; + *engine.write().await = MasternodeListEngine::default_for_network(network); } - Ok(client) + // Rebuild the managers while holding only the storage and config locks, then reset the + // coordinator with them, so we never hold the storage lock and the coordinator lock at + // once. + let managers = { + let mut storage = self.storage.lock().await; + storage.clear().await.map_err(SpvError::Storage)?; + let config = self.config.read().await; + Self::build_managers(&config, &mut storage, &self.wallet, &self.masternode_engine) + .await? + }; + self.sync_coordinator.lock().await.reset(managers); + + if was_running { + // Respawn the coordinator's tasks and reconnect, mirroring `start` but without + // flipping `running`. `connect` re-arms the network for a fresh start on its own, + // so the rebuilt managers pick up a working request path. + let mut network = self.network.lock().await; + self.sync_coordinator + .lock() + .await + .start(&mut *network) + .await + .map_err(SpvError::Sync)?; + network.connect().await?; + } + + Ok(()) } /// Start the SPV client: spawn sync tasks and connect to the network. diff --git a/dash-spv/src/client/mod.rs b/dash-spv/src/client/mod.rs index 541a71794..0e8cca718 100644 --- a/dash-spv/src/client/mod.rs +++ b/dash-spv/src/client/mod.rs @@ -53,13 +53,21 @@ mod tests { const TEST_MNEMONIC: &str = "abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about"; - /// Construct a mainnet client with the given wallet birth height and optional - /// explicit `start_from_height`, then return the height and hash its chain is - /// anchored at. - async fn anchored_tip( + const SECOND_MNEMONIC: &str = + "legal winner thank year wave sausage worth useful legal winner thank yellow"; + + /// Construct a mainnet client seeded with a single wallet at `birth_height` and an + /// optional explicit `start_from_height`. Returns the client, its shared wallet manager, + /// and the storage `TempDir` guard, which the caller must keep alive for the client's + /// storage to remain valid. + async fn build_test_client( birth_height: u32, start_from_height: Option, - ) -> (u32, Option) { + ) -> ( + DashSpvClient, MockNetworkManager, DiskStorageManager>, + Arc>>, + TempDir, + ) { let mut wallet_manager = WalletManager::::new(Network::Mainnet); wallet_manager .create_wallet_from_mnemonic( @@ -82,11 +90,21 @@ mod tests { config, MockNetworkManager::new(), storage, - wallet, + wallet.clone(), vec![Arc::new(())], ) .await .expect("client construction must succeed"); + (client, wallet, temp_dir) + } + + /// Return the height and hash a client's chain is anchored at for the given wallet birth + /// height and optional explicit `start_from_height`. + async fn anchored_tip( + birth_height: u32, + start_from_height: Option, + ) -> (u32, Option) { + let (client, _wallet, _temp_dir) = build_test_client(birth_height, start_from_height).await; (client.tip_height().await, client.tip_hash().await) } @@ -109,6 +127,35 @@ mod tests { assert_eq!(anchored_tip(120_000, Some(60_000)).await.0, 50_000); } + #[tokio::test] + async fn force_resync_reanchors_to_lower_birth_wallet() { + // A client whose only wallet is born at 120_000 anchors at the 100_000 checkpoint. + let (client, wallet, _temp_dir) = build_test_client(120_000, None).await; + + assert_eq!(client.tip_height().await, 100_000); + assert!(!client.resync_needed().await); + + // Add a wallet born below the current anchor. Nothing re-anchors on its own, so the + // stored chain still starts at 100_000 and a resync is now required. + wallet + .write() + .await + .create_wallet_from_mnemonic( + SECOND_MNEMONIC, + 60_000, + WalletAccountCreationOptions::Default, + ) + .expect("second wallet creation must succeed"); + assert!(client.resync_needed().await); + assert_eq!(client.tip_height().await, 100_000); + + // Forcing a resync wipes the old anchor and re-anchors at the 50_000 checkpoint, the + // nearest at or below the new minimum birth height. + client.force_resync().await.expect("force resync must succeed"); + assert_eq!(client.tip_height().await, 50_000); + assert!(!client.resync_needed().await); + } + #[tokio::test] async fn client_exposes_shared_wallet_manager() { let config = ClientConfig::mainnet() diff --git a/dash-spv/src/network/manager.rs b/dash-spv/src/network/manager.rs index 680eedf4d..6d2a911f3 100644 --- a/dash-spv/src/network/manager.rs +++ b/dash-spv/src/network/manager.rs @@ -878,6 +878,12 @@ impl PeerNetworkManager { } } } + + // Return the receiver to the shared slot so a later `connect` can take it again. + // The channel and the `request_tx` clones managers hold are never rebuilt, so a + // restart reuses the same channel and every previously handed-out sender stays + // valid. + *this.request_rx.lock().await = Some(request_rx); }); } @@ -1338,6 +1344,10 @@ impl PeerNetworkManager { for addr in self.pool.get_connected_addresses().await { self.pool.remove_peer(&addr).await; } + + // `remove_peer` above bypasses the counter maintained on the notify path, so zero it + // explicitly here to leave a clean count for any later reconnect. + self.connected_peer_count.store(0, Ordering::Relaxed); } async fn record_capability_rejection(&self, addr: SocketAddr) { @@ -1414,6 +1424,12 @@ impl NetworkManager for PeerNetworkManager { } async fn connect(&mut self) -> NetworkResult<()> { + // Re-arm the shutdown token so a reconnect after `disconnect` (which cancelled it) + // spawns its tasks against a live token. The request channel is not rebuilt here: the + // request processor returns its receiver to the shared slot on shutdown, so `start` + // simply takes it again, keeping every manager's request sender valid across + // reconnects. + self.shutdown_token = CancellationToken::new(); self.start().await.map_err(|e| NetworkError::ConnectionFailed(e.to_string())) } diff --git a/dash-spv/src/network/mod.rs b/dash-spv/src/network/mod.rs index 017af47d9..d224f9fe3 100644 --- a/dash-spv/src/network/mod.rs +++ b/dash-spv/src/network/mod.rs @@ -201,7 +201,8 @@ pub trait NetworkManager: Send + Sync + 'static { /// Connect to the network. async fn connect(&mut self) -> NetworkResult<()>; - /// Disconnect from the network. + /// Disconnect from the network. Implementations must leave the manager in a startable + /// state, so a later `connect` runs as if freshly constructed. async fn disconnect(&mut self) -> NetworkResult<()>; /// Send a message to a peer. diff --git a/dash-spv/src/sync/block_headers/manager.rs b/dash-spv/src/sync/block_headers/manager.rs index b3699e979..1f58c08f7 100644 --- a/dash-spv/src/sync/block_headers/manager.rs +++ b/dash-spv/src/sync/block_headers/manager.rs @@ -209,6 +209,12 @@ impl BlockHeadersManager { events.push(SyncEvent::BlockHeaderSyncComplete { tip_height: tip.height(), }); + + // A block that lands while we're still catching up can be missed: the peer + // only announces a new block when it connects to the tip it believes we + // already have, so one mined before we finished syncing is never pushed to + // us. Poll our tip once on the sync-completion edge to pull it in directly. + requests.request_block_headers(*tip.hash())?; } } diff --git a/dash-spv/src/sync/sync_coordinator.rs b/dash-spv/src/sync/sync_coordinator.rs index 7f9bc245d..476efb79d 100644 --- a/dash-spv/src/sync/sync_coordinator.rs +++ b/dash-spv/src/sync/sync_coordinator.rs @@ -147,8 +147,8 @@ where M: MetadataStorage, W: WalletInterface + 'static, { - /// Create a new coordinator with the given config. - pub(crate) async fn new(managers: Managers) -> Self { + /// Aggregate the starting progress across every present manager. + fn seed_initial_progress(managers: &Managers) -> SyncProgress { let mut initial_progress = SyncProgress::default(); try_update_progress(managers.block_headers.as_ref(), &mut initial_progress); @@ -160,6 +160,13 @@ where try_update_progress(managers.instantsend.as_ref(), &mut initial_progress); try_update_progress(managers.mempool.as_ref(), &mut initial_progress); + initial_progress + } + + /// Create a new coordinator with the given config. + pub(crate) async fn new(managers: Managers) -> Self { + let initial_progress = Self::seed_initial_progress(&managers); + tracing::info!("Initial sync progress {}", initial_progress.clone()); let (progress_sender, progress_receiver) = watch::channel(initial_progress); @@ -177,6 +184,23 @@ where } } + /// Reset the coordinator for a fresh sync against rebuilt managers, keeping the progress + /// and sync-event senders so existing subscribers stay connected across the restart. + /// + /// Call after `shutdown` has drained the previous tasks, then `start` respawns against the + /// new managers. Replacing the whole coordinator instead would drop these senders and + /// silently orphan every subscriber (progress and event streams would freeze). + pub(crate) fn reset(&mut self, managers: Managers) { + let initial_progress = Self::seed_initial_progress(&managers); + + self.managers = managers; + self.progress_receivers.clear(); + self.shutdown = CancellationToken::new(); + self.sync_start_time = None; + self.progress_task = None; + self.progress_sender.send_replace(initial_progress); + } + /// Subscribe to progress updates. pub fn subscribe_progress(&self) -> watch::Receiver { self.progress_sender.subscribe() diff --git a/dash-spv/tests/dashd_sync/tests_multi_wallet.rs b/dash-spv/tests/dashd_sync/tests_multi_wallet.rs index 276f43cbf..ed2e286c1 100644 --- a/dash-spv/tests/dashd_sync/tests_multi_wallet.rs +++ b/dash-spv/tests/dashd_sync/tests_multi_wallet.rs @@ -653,3 +653,77 @@ async fn test_runtime_add_with_tip_advance_during_rescan() { client_handle.stop().await; } + +/// `force_resync` while running tears down sync, wipes chain storage, and rebuilds the +/// coordinator so the next sync re-anchors from scratch (genesis on regtest, which has no +/// checkpoints). The client keeps running and re-syncs the full chain, and the +/// already-synced wallet keeps its per-wallet progress and discovered outputs since wallet +/// state lives outside chain storage. +#[tokio::test] +async fn test_force_resync_rebuilds_and_keeps_syncing() { + let Some(ctx) = TestContext::new(TestChain::Minimal).await else { + return; + }; + if !ctx.dashd.supports_mining { + eprintln!("Skipping test (dashd RPC miner not available)"); + return; + } + + let wallet = Arc::new(RwLock::new(WalletManager::::new(Network::Regtest))); + let mut client_handle = create_and_start_client(&ctx.client_config, Arc::clone(&wallet)).await; + wait_for_sync(&mut client_handle.progress_receiver, ctx.dashd.initial_height).await; + + let initial_height = ctx.dashd.initial_height; + + // Add the pre-funded wallet and let it catch up so it has discovered state. + let w1_id = { + let mut wallet_guard = client_handle.client.wallet().write().await; + wallet_guard + .create_wallet_from_mnemonic( + &ctx.dashd.wallet.mnemonic, + 0, + default_test_account_options(), + ) + .expect("add pre-funded wallet at runtime") + }; + wait_for_wallet_synced(client_handle.client.wallet(), &w1_id, initial_height).await; + + let balance_before = get_spendable_balance(client_handle.client.wallet(), &w1_id).await; + let tx_count_before = count_wallet_transactions(client_handle.client.wallet(), &w1_id).await; + assert!(tx_count_before > 0, "pre-funded wallet should have transactions before the resync"); + + // Wipe chain storage and rebuild the sync managers while running. + client_handle.client.force_resync().await.expect("force resync must succeed"); + + // Wait on the external progress receiver for the re-sync back to the tip. force_resync + // resets the coordinator in place: it re-seeds progress to the genesis anchor (regtest has + // no checkpoints) and re-anchors from scratch, so the receiver observably drops from the + // pre-resync tip down to genesis and climbs back. Reaching the tip on this receiver proves + // both that the re-sync ran and that the progress stream survived the resync rather than + // freezing at its pre-resync value. + wait_for_sync(&mut client_handle.progress_receiver, initial_height).await; + + // Once fully re-synced and steady, a boundary block mined at the tip must be picked up, + // proving the rebuilt client keeps tracking the chain after a resync. Regtest replays the + // whole re-sync in milliseconds, so without a pause the block can be mined before dashd + // has registered the just-reconnected peer's tip and it never gets announced. Give dashd + // that moment, matching the settling a real network provides between blocks. + tokio::time::sleep(Duration::from_secs(3)).await; + let miner_address = ctx.dashd.node.get_new_address_from_wallet("default"); + ctx.dashd.node.generate_blocks(1, &miner_address); + wait_for_sync(&mut client_handle.progress_receiver, initial_height + 1).await; + + // Wallet state lives outside chain storage, so the resync preserves it. + assert_eq!( + get_spendable_balance(client_handle.client.wallet(), &w1_id).await, + balance_before, + "wallet balance must be preserved across force_resync", + ); + assert_eq!( + count_wallet_transactions(client_handle.client.wallet(), &w1_id).await, + tx_count_before, + "wallet transaction count must be preserved across force_resync", + ); + + client_handle.stop().await; +} From 1e7319dbb7009d75a6cb96c5b172a5fdfd0ba13a Mon Sep 17 00:00:00 2001 From: xdustinface Date: Thu, 9 Jul 2026 10:56:19 +1000 Subject: [PATCH 2/4] feat(dash-spv): detect and signal when a resync is needed Add `resync_if_needed`, which runs `force_resync` only when `resync_needed` reports a wallet now sits below the anchored header start, returning whether a resync actually ran. A caller that just added wallets can do the whole "rescan the older history if I need to" step in one call instead of pairing the check and the action itself. Add `EventHandler::on_resync_needed`, fired once by the run loop on the rising edge of `resync_needed` so a wallet added below the anchor is reported without the caller polling. The check runs on its own low-frequency interval, kept well above the sync tick so the extra storage lock it takes does not contend with active sync. The client only notifies: reacting is a policy decision (`force_resync` wipes and refetches chain data), so the app calls `force_resync` or `resync_if_needed` when it fits, or defers. --- dash-spv/src/client/event_handler.rs | 7 ++ dash-spv/src/client/lifecycle.rs | 13 +++ dash-spv/src/client/mod.rs | 111 +++++++++++++++++++++++- dash-spv/src/client/sync_coordinator.rs | 21 +++++ 4 files changed, 150 insertions(+), 2 deletions(-) diff --git a/dash-spv/src/client/event_handler.rs b/dash-spv/src/client/event_handler.rs index a6ef1aace..563f6d832 100644 --- a/dash-spv/src/client/event_handler.rs +++ b/dash-spv/src/client/event_handler.rs @@ -29,6 +29,13 @@ pub trait EventHandler: Send + Sync + 'static { fn on_progress(&self, _progress: &SyncProgress) {} /// Called for wallet events (transaction received, balance updated). fn on_wallet_event(&self, _event: &WalletEvent) {} + /// Called once when a wallet's history first falls below the anchored header start, so its + /// earlier blocks cannot be scanned without re-anchoring lower. Fires on the rising edge of + /// [`crate::client::DashSpvClient::resync_needed`]. The client does not act on it: react by + /// calling [`crate::client::DashSpvClient::force_resync`] or + /// [`crate::client::DashSpvClient::resync_if_needed`] when the policy fits (it wipes and + /// refetches chain data), or defer. + fn on_resync_needed(&self) {} /// Called on fatal errors (start failure, monitor channel failure, sync loop error). fn on_error(&self, _error: &str) {} } diff --git a/dash-spv/src/client/lifecycle.rs b/dash-spv/src/client/lifecycle.rs index 3904474be..5347abdf2 100644 --- a/dash-spv/src/client/lifecycle.rs +++ b/dash-spv/src/client/lifecycle.rs @@ -278,6 +278,19 @@ impl DashSpvClient Result { + if self.resync_needed().await { + self.force_resync().await?; + Ok(true) + } else { + Ok(false) + } + } + /// Start the SPV client: spawn sync tasks and connect to the network. pub(super) async fn start(&self) -> Result<()> { if self.is_running() { diff --git a/dash-spv/src/client/mod.rs b/dash-spv/src/client/mod.rs index 0e8cca718..4ebd4915d 100644 --- a/dash-spv/src/client/mod.rs +++ b/dash-spv/src/client/mod.rs @@ -38,7 +38,7 @@ mod config_test; #[cfg(test)] mod tests { - use super::{ClientConfig, DashSpvClient}; + use super::{ClientConfig, DashSpvClient, EventHandler}; use crate::client::config::MempoolStrategy; use crate::storage::DiskStorageManager; use crate::test_utils::MockNetworkManager; @@ -46,7 +46,9 @@ mod tests { use key_wallet::wallet::initialization::WalletAccountCreationOptions; use key_wallet::wallet::managed_wallet_info::ManagedWalletInfo; use key_wallet_manager::WalletManager; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; + use std::time::Duration; use tempfile::TempDir; use tokio::sync::RwLock; @@ -67,6 +69,20 @@ mod tests { DashSpvClient, MockNetworkManager, DiskStorageManager>, Arc>>, TempDir, + ) { + build_test_client_with_handlers(birth_height, start_from_height, vec![Arc::new(())]).await + } + + /// Like [`build_test_client`] but installs the given event handlers, so a test can observe + /// the client's push notifications. + async fn build_test_client_with_handlers( + birth_height: u32, + start_from_height: Option, + event_handlers: Vec>, + ) -> ( + DashSpvClient, MockNetworkManager, DiskStorageManager>, + Arc>>, + TempDir, ) { let mut wallet_manager = WalletManager::::new(Network::Mainnet); wallet_manager @@ -91,7 +107,7 @@ mod tests { MockNetworkManager::new(), storage, wallet.clone(), - vec![Arc::new(())], + event_handlers, ) .await .expect("client construction must succeed"); @@ -156,6 +172,97 @@ mod tests { assert!(!client.resync_needed().await); } + #[tokio::test] + async fn resync_if_needed_runs_only_when_needed() { + // Only wallet born at 120_000, anchored at 100_000: no resync is needed, so + // `resync_if_needed` is a no-op that reports it did nothing and leaves the anchor. + let (client, wallet, _temp_dir) = build_test_client(120_000, None).await; + assert_eq!(client.tip_height().await, 100_000); + assert!(!client.resync_if_needed().await.expect("resync_if_needed must succeed")); + assert_eq!(client.tip_height().await, 100_000); + + // A wallet added below the anchor makes a resync necessary, so now it re-anchors and + // reports that it ran. + wallet + .write() + .await + .create_wallet_from_mnemonic( + SECOND_MNEMONIC, + 60_000, + WalletAccountCreationOptions::Default, + ) + .expect("second wallet creation must succeed"); + assert!(client.resync_if_needed().await.expect("resync_if_needed must succeed")); + assert_eq!(client.tip_height().await, 50_000); + + // Once re-anchored the condition is cleared, so a second call is again a no-op. + assert!(!client.resync_if_needed().await.expect("resync_if_needed must succeed")); + assert_eq!(client.tip_height().await, 50_000); + } + + #[tokio::test] + async fn run_loop_signals_resync_needed_on_wallet_add_below_anchor() { + // Records how many times the client pushed `on_resync_needed`. + struct ResyncRecorder { + count: Arc, + } + impl EventHandler for ResyncRecorder { + fn on_resync_needed(&self) { + self.count.fetch_add(1, Ordering::SeqCst); + } + } + + let count = Arc::new(AtomicUsize::new(0)); + let (client, wallet, _temp_dir) = build_test_client_with_handlers( + 120_000, + None, + vec![Arc::new(ResyncRecorder { + count: count.clone(), + })], + ) + .await; + let client = Arc::new(client); + + // Drive the run loop so its periodic resync check runs. The mock network connects + // without a real peer, so sync just idles. + let run_client = Arc::clone(&client); + let run_handle = tokio::spawn(async move { run_client.run().await }); + + // Let the loop start and seed its baseline (resync not yet needed) before we change + // the wallet set, so adding the wallet is observed as a rising edge. + while !client.is_running() { + tokio::time::sleep(Duration::from_millis(20)).await; + } + tokio::time::sleep(Duration::from_millis(300)).await; + assert_eq!(count.load(Ordering::SeqCst), 0); + + // Add a wallet below the anchor: the next periodic check must see the rising edge and + // fire the notification exactly once. + wallet + .write() + .await + .create_wallet_from_mnemonic( + SECOND_MNEMONIC, + 60_000, + WalletAccountCreationOptions::Default, + ) + .expect("second wallet creation must succeed"); + + let mut waited = Duration::ZERO; + while count.load(Ordering::SeqCst) == 0 && waited < Duration::from_secs(5) { + tokio::time::sleep(Duration::from_millis(50)).await; + waited += Duration::from_millis(50); + } + assert_eq!(count.load(Ordering::SeqCst), 1, "resync-needed must fire once on the edge"); + + // The condition stays true but must not re-fire on later checks. + tokio::time::sleep(Duration::from_millis(1200)).await; + assert_eq!(count.load(Ordering::SeqCst), 1, "resync-needed must not repeat"); + + client.stop().await.expect("stop must succeed"); + let _ = run_handle.await; + } + #[tokio::test] async fn client_exposes_shared_wallet_manager() { let config = ClientConfig::mainnet() diff --git a/dash-spv/src/client/sync_coordinator.rs b/dash-spv/src/client/sync_coordinator.rs index e428bcb80..2b88c77b1 100644 --- a/dash-spv/src/client/sync_coordinator.rs +++ b/dash-spv/src/client/sync_coordinator.rs @@ -18,6 +18,11 @@ use key_wallet_manager::WalletInterface; const SYNC_COORDINATOR_TICK_MS: Duration = Duration::from_millis(100); +/// How often the run loop re-checks whether a resync is needed. Kept well above the sync tick +/// so the extra storage lock this takes doesn't contend with active sync: a resync becomes +/// needed only when a wallet is added below the anchor, which is rare and not latency-sensitive. +const RESYNC_CHECK_INTERVAL: Duration = Duration::from_secs(1); + impl DashSpvClient { /// Get current sync progress. pub async fn sync_progress(&self) -> SyncProgress { @@ -113,6 +118,12 @@ impl DashSpvClient = loop { if !self.is_running() { @@ -124,6 +135,16 @@ impl DashSpvClient { self.sync_coordinator.lock().await.tick().await.err().map(Into::into) } + _ = resync_check_interval.tick() => { + let needed = self.resync_needed().await; + if needed && !resync_was_needed { + for handler in handlers.iter() { + handler.on_resync_needed(); + } + } + resync_was_needed = needed; + None + } _ = stop_rx.changed() => { tracing::debug!("DashSpvClient run loop stop requested"); break None From ffa8322b406cd8dc4a1371bd5b7a2a146f19d1b6 Mon Sep 17 00:00:00 2001 From: xdustinface Date: Thu, 9 Jul 2026 17:41:10 +1000 Subject: [PATCH 3/4] feat(dash-spv): integration-test checkpoint anchoring and re-anchor against a real node Add `ClientConfig::with_checkpoints` to override the network's bundled checkpoints (regtest and devnet ship none), a `DashSpvClient::start_height` accessor, and a `DashCoreNode::checkpoint_at` test helper that builds a real checkpoint from the running chain. Regtest ships no checkpoints, so the anchor was always genesis and no integration test could exercise checkpoint anchoring against a real node, only mock unit tests. `test_checkpoint_anchoring_and_reanchor_to_lower_checkpoint` now injects checkpoints built from the running chain, anchors a client at a mid-chain checkpoint and syncs forward from it, then adds a wallet below the anchor and asserts `force_resync` re-anchors at the next checkpoint down and re-syncs. This also closes the pre-existing gap that checkpoint anchoring at startup had no real-node coverage at all. Production networks keep their bundled checkpoints, so the config override is a test-only affordance with no behavior change. --- dash-spv/src/client/config.rs | 30 +++++++++ dash-spv/src/client/core.rs | 6 ++ dash-spv/src/client/lifecycle.rs | 5 +- dash-spv/src/test_utils/node.rs | 26 ++++++++ .../tests/dashd_sync/tests_multi_wallet.rs | 63 +++++++++++++++++++ 5 files changed, 127 insertions(+), 3 deletions(-) diff --git a/dash-spv/src/client/config.rs b/dash-spv/src/client/config.rs index d14aa095f..3b856e8a9 100644 --- a/dash-spv/src/client/config.rs +++ b/dash-spv/src/client/config.rs @@ -7,6 +7,9 @@ use std::path::PathBuf; use dashcore::Network; // Serialization removed due to complex Address types +#[cfg(feature = "test-utils")] +use crate::chain::Checkpoint; +use crate::chain::CheckpointManager; use crate::client::devnet::DevnetConfig; use crate::types::ValidationMode; @@ -72,6 +75,12 @@ pub struct ClientConfig { /// The client will use the nearest checkpoint at or before this height. pub start_from_height: Option, + /// Override the bundled checkpoints for a network that ships none (regtest, devnet), so tests + /// can exercise checkpoint anchoring against a real node. Compiled only under the + /// `test-utils` feature, so it is not part of the production config surface. + #[cfg(feature = "test-utils")] + pub checkpoints: Option>, + /// Devnet-only configuration. Must be `Some` iff `network == Network::Devnet`. pub devnet: Option, } @@ -94,6 +103,8 @@ impl Default for ClientConfig { max_mempool_transactions: 1000, fetch_mempool_transactions: true, start_from_height: None, + #[cfg(feature = "test-utils")] + checkpoints: None, devnet: None, } } @@ -186,6 +197,25 @@ impl ClientConfig { self } + /// Override the bundled checkpoints with a custom set, for a network that ships none (e.g. + /// regtest). Test-only: compiled under the `test-utils` feature. + #[cfg(feature = "test-utils")] + pub fn with_checkpoints(mut self, checkpoints: Vec) -> Self { + self.checkpoints = Some(checkpoints); + self + } + + /// Build the checkpoint manager for this config: the network's bundled checkpoints, or a + /// `test-utils`-only custom override when set (for networks that ship none). Production builds + /// have no override field, so they always use the bundled set. + pub(crate) fn checkpoint_manager(&self) -> CheckpointManager { + #[cfg(feature = "test-utils")] + if let Some(checkpoints) = &self.checkpoints { + return CheckpointManager::new(checkpoints.clone()); + } + CheckpointManager::for_network(self.network) + } + /// Attach a [`DevnetConfig`]. The network must be `Network::Devnet`. /// [`validate`](Self::validate) enforces the biconditional. pub fn with_devnet(mut self, devnet: DevnetConfig) -> Self { diff --git a/dash-spv/src/client/core.rs b/dash-spv/src/client/core.rs index 09e8bb712..2f7c9924b 100644 --- a/dash-spv/src/client/core.rs +++ b/dash-spv/src/client/core.rs @@ -168,6 +168,12 @@ impl DashSpvClient Option { + self.storage.lock().await.get_start_height().await + } + // ============ Storage Operations ============ /// Clear all persisted storage (headers, filters, state, sync state) and reset in-memory state. diff --git a/dash-spv/src/client/lifecycle.rs b/dash-spv/src/client/lifecycle.rs index 5347abdf2..9a48b9cfb 100644 --- a/dash-spv/src/client/lifecycle.rs +++ b/dash-spv/src/client/lifecycle.rs @@ -9,7 +9,6 @@ //! - Wallet data loading use super::{ClientConfig, DashSpvClient, EventHandler}; -use crate::chain::checkpoints::CheckpointManager; use crate::error::{Result, SpvError}; use crate::network::NetworkManager; use crate::storage::{ @@ -130,7 +129,7 @@ impl DashSpvClient = Managers::default(); - let checkpoint_manager = Arc::new(CheckpointManager::for_network(config.network)); + let checkpoint_manager = Arc::new(config.checkpoint_manager()); managers.block_headers = Some( BlockHeadersManager::new( storage.block_headers(), @@ -373,7 +372,7 @@ impl DashSpvClient Checkpoint { + let client = self.rpc_client(); + let block_hash = client.get_block_hash(height).expect("getblockhash failed"); + let header = client.get_block_header(&block_hash).expect("getblockheader failed"); + Checkpoint { + height, + block_hash, + prev_blockhash: header.prev_blockhash, + timestamp: header.time, + target: header.target(), + merkle_root: Some(BlockHash::from_byte_array(header.merkle_root.to_byte_array())), + chain_work: format!("0x{:064x}", height.wrapping_mul(1000)), + masternode_list_name: None, + protocol_version: None, + nonce: header.nonce, + } + } + /// Call getblocktemplate to trigger CreateNewBlock (includes quorum commitments). pub fn get_block_template(&self) { let client = self.rpc_client(); diff --git a/dash-spv/tests/dashd_sync/tests_multi_wallet.rs b/dash-spv/tests/dashd_sync/tests_multi_wallet.rs index ed2e286c1..84ce86abc 100644 --- a/dash-spv/tests/dashd_sync/tests_multi_wallet.rs +++ b/dash-spv/tests/dashd_sync/tests_multi_wallet.rs @@ -727,3 +727,66 @@ async fn test_force_resync_rebuilds_and_keeps_syncing() { client_handle.stop().await; } + +/// Checkpoint anchoring and re-anchoring against a real node. Regtest ships no checkpoints, so +/// the test injects two built from the running chain. A wallet born above the higher checkpoint +/// anchors the client there and syncs forward from mid-chain (the path mock unit tests can't +/// cover). Adding a wallet below that anchor makes a resync necessary, and `force_resync` +/// re-anchors at the next checkpoint down and re-syncs from there. +#[tokio::test] +async fn test_checkpoint_anchoring_and_reanchor_to_lower_checkpoint() { + let Some(ctx) = TestContext::new(TestChain::Minimal).await else { + return; + }; + + let initial_height = ctx.dashd.initial_height; + + // Two checkpoints below the tip, built from the real chain so forward sync connects to them. + let checkpoints = vec![ctx.dashd.node.checkpoint_at(50), ctx.dashd.node.checkpoint_at(100)]; + let config = ctx.client_config.clone().with_checkpoints(checkpoints); + + // A single wallet born above the higher checkpoint, so the client anchors at height 100. + let wallet = Arc::new(RwLock::new(WalletManager::::new(Network::Regtest))); + wallet + .write() + .await + .create_wallet_from_mnemonic(EMPTY_MNEMONIC, 150, default_test_account_options()) + .expect("first wallet creation must succeed"); + + let mut client_handle = create_and_start_client(&config, Arc::clone(&wallet)).await; + wait_for_sync(&mut client_handle.progress_receiver, initial_height).await; + + // Anchored at the 100 checkpoint rather than genesis, proving checkpoint sync works against a + // real node, and no resync is needed while the only wallet is above the anchor. + assert_eq!( + client_handle.client.start_height().await, + Some(100), + "client should anchor at the 100 checkpoint", + ); + assert!(!client_handle.client.resync_needed().await); + + // A wallet born below the anchor cannot have its earlier history scanned without + // re-anchoring lower, so a resync becomes necessary. + wallet + .write() + .await + .create_wallet_from_mnemonic(SECONDARY_MNEMONIC, 60, default_test_account_options()) + .expect("second wallet creation must succeed"); + assert!( + client_handle.client.resync_needed().await, + "a wallet born at 60 sits below the 100 anchor", + ); + + // Re-anchor to the next checkpoint down (50, the nearest at or below birth 60) and re-sync + // from there against the node. + client_handle.client.force_resync().await.expect("force resync must succeed"); + wait_for_sync(&mut client_handle.progress_receiver, initial_height).await; + assert_eq!( + client_handle.client.start_height().await, + Some(50), + "force_resync should re-anchor at the 50 checkpoint", + ); + assert!(!client_handle.client.resync_needed().await); + + client_handle.stop().await; +} From 894e0e5a792d973118807a9dfa8e9fb6e7955e49 Mon Sep 17 00:00:00 2001 From: xdustinface Date: Fri, 10 Jul 2026 00:44:12 +1000 Subject: [PATCH 4/4] fix(dash-spv): mark client stopped when `force_resync` fails mid-teardown If `storage.clear()` or `build_managers()` fails after `force_resync` has already torn down the coordinator and disconnected the network, the client was left with `running` set but a drained coordinator, so `run`'s `tick` reported no error and spun forever without surfacing the failure. Factor the body into `force_resync_body` and, on failure while `was_running`, drain the coordinator (also cleaning up any tasks a failed `start`/`connect` restart spawned) and flip `running` to false so `run` exits and the error surfaces. A later `force_resync` retries the rebuild cleanly. Addresses CodeRabbit review comment on PR #856 https://github.com/dashpay/rust-dashcore/pull/856#discussion_r3551975396 --- dash-spv/src/client/lifecycle.rs | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/dash-spv/src/client/lifecycle.rs b/dash-spv/src/client/lifecycle.rs index 9a48b9cfb..9fa8f848f 100644 --- a/dash-spv/src/client/lifecycle.rs +++ b/dash-spv/src/client/lifecycle.rs @@ -231,9 +231,26 @@ impl DashSpvClient Result<()> { let was_running = self.is_running(); + let result = self.force_resync_body(was_running).await; + + if was_running && result.is_err() { + let _ = self.sync_coordinator.lock().await.shutdown().await; + self.running.send_replace(false); + } + + result + } + + async fn force_resync_body(&self, was_running: bool) -> Result<()> { // Tear down the running sync without touching `running`, so the `run` loop keeps // monitoring. Drain the coordinator's manager tasks and drop the network connection. if was_running {