From e31d68f2c111b0c9836ed14c2d11c2229a816f3f Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Fri, 11 Sep 2026 16:36:42 +0000 Subject: [PATCH 1/2] fix(key-wallet): re-apply a spend whose coin was funded after it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two mainnet restores of the same wallet ended at the same balance with 7112 and 7111 wallet records. Replaying each run's logged block applications offline, against the blocks it stored, reproduced both results exactly, with no divergence from the logs. The missing record was e66553f3…8c9e at height 2 185 057: it pays change to the BIP44 account and spends a CoinJoin coin funded at 2 182 877. When the spend is applied before its funding, the CoinJoin account cannot recognise it. The funding then parks the coin in `spent_before_funded` (#1001), and that only attributes the spend if its block is delivered again. In one run it was (funding at step 1494, spend at 1516); in the other it was not (spend at 1608, funding at 1627, no redelivery), so the result depended on delivery order. `WalletInfoInterface::unrecorded_spend_heights` reports, for a transaction, the heights of the blocks that spent its outputs before it arrived and that the owning account has not recorded yet. `process_block_for_wallets` returns them per wallet in `BlockProcessingResult::reapply_heights`, only heights above the block being applied, so re-applying cannot loop. `BlocksManager` re-applies those blocks from block storage straight away; every downloaded block is stored on arrival. Re-applications emit no `SyncEvent::BlockProcessed` and so never touch a batch's pending-block accounting. Offline, both orderings now end with identical per-account records (7112), with about 105 blocks re-applied from disk per restore. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017DruChNTWXwJoWPartZwCf --- dash-spv/src/sync/blocks/manager.rs | 144 +++++++++++++----- key-wallet-manager/src/process_block.rs | 46 ++++++ .../src/test_utils/mock_wallet.rs | 13 +- key-wallet-manager/src/wallet_interface.rs | 1 + .../tests/observed_spent_outpoints_tests.rs | 22 ++- .../wallet_info_interface.rs | 19 ++- 6 files changed, 200 insertions(+), 45 deletions(-) diff --git a/dash-spv/src/sync/blocks/manager.rs b/dash-spv/src/sync/blocks/manager.rs index e4a1464d8..851fc318a 100644 --- a/dash-spv/src/sync/blocks/manager.rs +++ b/dash-spv/src/sync/blocks/manager.rs @@ -3,6 +3,7 @@ //! Downloads blocks that matched wallet filters and processes them in height order. //! Subscribes to BlockNeeded events and emits BlockProcessed events. +use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; use tokio::sync::RwLock; @@ -12,7 +13,8 @@ use crate::error::SyncResult; use crate::network::RequestSender; use crate::storage::{BlockHeaderStorage, BlockStorage}; use crate::sync::{BlocksProgress, SyncEvent, SyncManager, SyncState}; -use key_wallet_manager::WalletInterface; +use crate::types::HashedBlock; +use key_wallet_manager::{BlockProcessingResult, WalletId, WalletInterface}; /// Blocks manager for downloading and processing matching blocks. /// @@ -87,45 +89,10 @@ impl BlocksManager 0 { - tracing::info!( - "Found {} relevant transactions ({} new, {} existing) {} at height {}, new scripts: {}", - total_relevant, - result.new_txids.len(), - result.existing_txids.len(), - hash, - height, - new_scripts_total - ); - } + let result = self.apply_block(&block, height, &interested).await; // Collect confirmed txids before moving new_scripts out of result let confirmed_txids: Vec<_> = result.relevant_txids().cloned().collect(); - - // Collect new scripts for gap limit rescanning - let new_scripts = result.new_scripts; - if new_scripts_total > 0 { - tracing::debug!( - "Block {} generated {} new scripts for gap limit maintenance across {} wallets", - height, - new_scripts_total, - new_scripts.len() - ); - } - - self.progress.add_processed(1); - if total_relevant > 0 { - self.progress.add_relevant(1); - } - // Only count new transactions to avoid double-counting during rescans - self.progress.add_transactions(result.new_txids.len() as u32); self.progress.update_last_processed(height); last_applied = Some(height); @@ -133,9 +100,11 @@ impl BlocksManager BlocksManager, + ) -> BlockProcessingResult { + let hash = *block.hash(); + let mut wallet = self.wallet.write().await; + let result = wallet.process_block_for_wallets(block.block(), hash, height, wallets).await; + drop(wallet); + + let total_relevant = result.relevant_tx_count(); + let new_scripts_total: usize = result.new_scripts.values().map(|v| v.len()).sum(); + if total_relevant > 0 { + tracing::info!( + "Found {} relevant transactions ({} new, {} existing) {} at height {}, new scripts: {}", + total_relevant, + result.new_txids.len(), + result.existing_txids.len(), + hash, + height, + new_scripts_total + ); + } + if new_scripts_total > 0 { + tracing::debug!( + "Block {} generated {} new scripts for gap limit maintenance across {} wallets", + height, + new_scripts_total, + result.new_scripts.len() + ); + } + + self.progress.add_processed(1); + if total_relevant > 0 { + self.progress.add_relevant(1); + } + // Only count new transactions to avoid double-counting during rescans + self.progress.add_transactions(result.new_txids.len() as u32); + result + } + + async fn reapply_blocks( + &mut self, + reapply: BTreeMap>, + ) -> SyncResult<()> { + let mut queue: BTreeSet<(u32, WalletId)> = reapply + .into_iter() + .flat_map(|(wallet_id, heights)| heights.into_iter().map(move |h| (h, wallet_id))) + .collect(); + while let Some((height, wallet_id)) = queue.pop_first() { + let Some(block) = self.block_storage.read().await.load_block(height).await? else { + tracing::warn!("Cannot re-apply block at height {}: not in storage", height); + continue; + }; + let result = self.apply_block(&block, height, &BTreeSet::from([wallet_id])).await; + queue.extend( + result.reapply_heights.into_iter().flat_map(|(wallet_id, heights)| { + heights.into_iter().map(move |h| (h, wallet_id)) + }), + ); + } + Ok(()) + } } impl std::fmt::Debug @@ -353,6 +387,40 @@ mod tests { assert_eq!(processed[0].1, 100); } + #[tokio::test] + async fn test_process_buffered_blocks_reapplies_requested_stored_block() { + let storage = DiskStorageManager::with_temp_dir().await.unwrap(); + let mut wallet = MockWallet::new(); + wallet.set_reapply_heights(100, BTreeSet::from([200])); + let wallet = Arc::new(RwLock::new(wallet)); + let mut manager: TestBlocksManager = + BlocksManager::new(wallet.clone(), storage.block_headers(), storage.blocks()).await; + manager.progress.set_state(SyncState::Syncing); + + manager + .block_storage + .write() + .await + .store_block(200, HashedBlock::dummy(200, vec![])) + .await + .unwrap(); + manager.pipeline.add_from_storage( + HashedBlock::dummy(100, vec![]), + 100, + BTreeSet::from([MOCK_WALLET_ID]), + ); + + let events = manager.process_buffered_blocks().await.unwrap(); + assert_eq!( + events.iter().filter(|e| matches!(e, SyncEvent::BlockProcessed { .. })).count(), + 1 + ); + + let processed = wallet.read().await.processed_blocks(); + let heights: Vec = processed.lock().await.iter().map(|(_, h)| *h).collect(); + assert_eq!(heights, vec![100, 200]); + } + /// A wallet that is NOT in the pipeline's interested set must not be /// routed the block. Two wallets are registered, but only `wallet_in` /// appears in the routed set; the other wallet's processed log must diff --git a/key-wallet-manager/src/process_block.rs b/key-wallet-manager/src/process_block.rs index cc442df16..13f5d4b8f 100644 --- a/key-wallet-manager/src/process_block.rs +++ b/key-wallet-manager/src/process_block.rs @@ -50,6 +50,7 @@ impl WalletInterface for WalletM let mut per_wallet_inserted: BTreeMap> = BTreeMap::new(); let mut per_wallet_updated: BTreeMap> = BTreeMap::new(); let mut per_wallet_derived: BTreeMap> = BTreeMap::new(); + let mut relevant_positions: BTreeMap> = BTreeMap::new(); for (position, tx) in block.txdata.iter().enumerate() { // Stamp each record with its `block.vtx` index so consumers @@ -71,6 +72,9 @@ impl WalletInterface for WalletM result.existing_txids.push(tx.txid()); } } + for wallet_id in &check_result.affected_wallets { + relevant_positions.entry(*wallet_id).or_default().push(position); + } for (wallet_id, derived) in check_result.new_addresses { let scripts = @@ -118,6 +122,20 @@ impl WalletInterface for WalletM ); } + for (wallet_id, positions) in relevant_positions { + let Some(info) = self.wallet_infos.get(&wallet_id) else { + continue; + }; + let heights: BTreeSet = positions + .into_iter() + .flat_map(|position| info.unrecorded_spend_heights(&block.txdata[position])) + .filter(|spend_height| *spend_height > height) + .collect(); + if !heights.is_empty() { + result.reapply_heights.insert(wallet_id, heights); + } + } + self.finalize_block_advance( height, wallets, @@ -669,6 +687,34 @@ mod tests { assert_eq!(manager.last_processed_height(), 0); } + #[tokio::test] + async fn test_funding_after_its_spend_asks_to_reapply_the_spend_block() { + let (mut manager, wallet_id, addr) = setup_manager_with_wallet(); + let funding = create_tx_paying_to(&addr, 0xaa); + let spend = spend_first_output_of(&funding); + let wallets = BTreeSet::from([wallet_id]); + + let mut spend_block = make_block(vec![spend]); + spend_block.header.nonce = 1; + let funding_block = make_block(vec![funding]); + + manager + .process_block_for_wallets(&spend_block, spend_block.block_hash(), 200, &wallets) + .await; + let result = manager + .process_block_for_wallets(&funding_block, funding_block.block_hash(), 100, &wallets) + .await; + assert_eq!(result.reapply_heights, BTreeMap::from([(wallet_id, BTreeSet::from([200]))])); + + manager + .process_block_for_wallets(&spend_block, spend_block.block_hash(), 200, &wallets) + .await; + let again = manager + .process_block_for_wallets(&funding_block, funding_block.block_hash(), 100, &wallets) + .await; + assert!(again.reapply_heights.is_empty()); + } + #[tokio::test] async fn test_sweep_expired_reservations_fans_out_over_wallets() { let (mut manager, wallet_id, _addr) = setup_manager_with_wallet(); diff --git a/key-wallet-manager/src/test_utils/mock_wallet.rs b/key-wallet-manager/src/test_utils/mock_wallet.rs index 63e02b487..dfbcf6e29 100644 --- a/key-wallet-manager/src/test_utils/mock_wallet.rs +++ b/key-wallet-manager/src/test_utils/mock_wallet.rs @@ -6,7 +6,7 @@ use dashcore::ephemerealdata::instant_lock::InstantLock; use dashcore::prelude::CoreBlockHeight; use dashcore::{Address, Block, OutPoint, ScriptBuf, Transaction, Txid}; use key_wallet::transaction_checking::TransactionContext; -use std::collections::BTreeSet; +use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; use tokio::sync::{broadcast, Mutex}; @@ -43,6 +43,7 @@ pub struct MockWallet { pub processed_instant_locks: InstantLockCaptures, /// Monitor revision counter for staleness detection. monitor_revision: u64, + reapply_heights: BTreeMap>, } impl Default for MockWallet { @@ -70,9 +71,14 @@ impl MockWallet { status_changes: Arc::new(Mutex::new(Vec::new())), processed_instant_locks: Arc::new(Mutex::new(Vec::new())), monitor_revision: 0, + reapply_heights: BTreeMap::new(), } } + pub fn set_reapply_heights(&mut self, height: u32, heights: BTreeSet) { + self.reapply_heights.insert(height, heights); + } + /// Override the wallet id used for per-wallet API surfaces. pub fn set_wallet_id(&mut self, wallet_id: WalletId) { self.wallet_id = wallet_id; @@ -145,6 +151,11 @@ impl WalletInterface for MockWallet { new_txids: block.txdata.iter().map(|tx| tx.txid()).collect(), existing_txids: Vec::new(), new_scripts: Default::default(), + reapply_heights: self + .reapply_heights + .get(&height) + .map(|heights| BTreeMap::from([(self.wallet_id, heights.clone())])) + .unwrap_or_default(), } } diff --git a/key-wallet-manager/src/wallet_interface.rs b/key-wallet-manager/src/wallet_interface.rs index 6f43d99de..c65025f98 100644 --- a/key-wallet-manager/src/wallet_interface.rs +++ b/key-wallet-manager/src/wallet_interface.rs @@ -21,6 +21,7 @@ pub struct BlockProcessingResult { /// Cached scriptPubKeys of addresses freshly generated per wallet during /// gap-limit maintenance. pub new_scripts: BTreeMap>, + pub reapply_heights: BTreeMap>, } /// Result of processing a mempool transaction through the wallet diff --git a/key-wallet/src/tests/observed_spent_outpoints_tests.rs b/key-wallet/src/tests/observed_spent_outpoints_tests.rs index 70a7e68ed..759ff5b2d 100644 --- a/key-wallet/src/tests/observed_spent_outpoints_tests.rs +++ b/key-wallet/src/tests/observed_spent_outpoints_tests.rs @@ -193,9 +193,9 @@ fn adding_account_from_xpub_rewinds_sync_checkpoint() { /// dropped. The balance and UTXO set are unaffected (the output is genuinely /// spent on-chain), so this is purely about not losing the history record. /// -/// This is the spend-first / out-of-order ordering produced by the committed- -/// range rescan (`track_for_new_scripts`): the forward scan already processed -/// the spend, then the old funding block is re-applied. #649's fix keeps the +/// This is the spend-first / out-of-order ordering produced by a rescan +/// (`track_for_new_scripts`): the scan already processed the spend, then the +/// older funding block is re-applied. #649's fix keeps the /// record: `check_transaction_for_match` classifies relevance by address /// membership (never gated on spent-status, so the fully-spent funding is still /// relevant), and `ManagedCoreFundsAccount::record_transaction` unconditionally @@ -241,8 +241,8 @@ async fn born_fully_spent_funding_tx_is_recorded_in_history() { }; // Spend-first: the spend's block (height 200) is applied before the - // funding's block (height 100), exactly as the committed-range rescan - // re-applies old funding blocks after their spends. + // funding's block (height 100), exactly as a rescan re-applies older + // funding blocks after their spends. let spend_ctx = TransactionContext::InBlock(BlockInfo::new( 200, BlockHash::from_slice(&[2u8; 32]).expect("hash"), @@ -428,6 +428,18 @@ async fn spend_seen_before_its_funding_is_recorded_on_redelivery() { ); } +#[tokio::test] +async fn funding_after_its_spend_reports_the_spend_height() { + use crate::wallet::managed_wallet_info::wallet_info_interface::WalletInfoInterface; + use std::collections::BTreeSet; + + let (mut ctx, funding, spend) = spend_first_context(in_block(100, 1)).await; + assert_eq!(ctx.managed_wallet.unrecorded_spend_heights(&funding), BTreeSet::from([200])); + + ctx.check_transaction(&spend, in_block(200, 2)).await; + assert!(ctx.managed_wallet.unrecorded_spend_heights(&funding).is_empty()); +} + /// Abandoning the funding transaction takes its held output with it: the coin /// was never ours, so a spend of it must stop being recognisable. #[tokio::test] diff --git a/key-wallet/src/wallet/managed_wallet_info/wallet_info_interface.rs b/key-wallet/src/wallet/managed_wallet_info/wallet_info_interface.rs index 585effb53..419781a8a 100644 --- a/key-wallet/src/wallet/managed_wallet_info/wallet_info_interface.rs +++ b/key-wallet/src/wallet/managed_wallet_info/wallet_info_interface.rs @@ -20,7 +20,7 @@ use dashcore::address::Payload; use dashcore::ephemerealdata::chain_lock::ChainLock; use dashcore::ephemerealdata::instant_lock::InstantLock; use dashcore::prelude::CoreBlockHeight; -use dashcore::{Address as DashAddress, ScriptBuf, Transaction, Txid}; +use dashcore::{Address as DashAddress, OutPoint, ScriptBuf, Transaction, Txid}; /// Outcome of [`WalletInfoInterface::apply_chain_lock`]. /// @@ -278,6 +278,8 @@ pub trait WalletInfoInterface: Sized + WalletTransactionChecker + ManagedAccount /// sweep removing a loser, so this is broader than "a UTXO was marked". fn mark_instant_send_utxos(&mut self, txid: &Txid, lock: &InstantLock) -> bool; + fn unrecorded_spend_heights(&self, tx: &Transaction) -> BTreeSet; + /// Return the aggregated monitor revision across all accounts. /// Increments whenever the monitored address set changes. fn monitor_revision(&self) -> u64 { @@ -601,6 +603,21 @@ impl WalletInfoInterface for ManagedWalletInfo { fn monitor_revision(&self) -> u64 { self.accounts.all_accounts().iter().map(|a| a.monitor_revision()).sum() } + + fn unrecorded_spend_heights(&self, tx: &Transaction) -> BTreeSet { + let txid = tx.txid(); + (0..tx.output.len() as u32) + .map(|vout| OutPoint::new(txid, vout)) + .filter(|outpoint| { + self.accounts + .all_accounts() + .into_iter() + .filter_map(|account| account.as_funds()) + .any(|account| account.spent_before_funded.contains_key(outpoint)) + }) + .filter_map(|outpoint| self.observed_spent_outpoints.get(&outpoint).copied()) + .collect() + } } #[cfg(test)] From 5156e071e9a73144866d4a0ccae1a4dfa8149f7b Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Fri, 11 Sep 2026 16:36:43 +0000 Subject: [PATCH 2/2] refactor(dash-spv): drop the committed-range sweep and the collected-scripts rescan MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The #846 sweep re-tested every late-derived script against the whole stored filter history, from the wallet's birth height. On a mainnet restore of the bench wallet that was one 63 s walk over 200 000..tip matching 41 687 blocks, overwhelmingly compact-filter false positives, then about 17.7 min downloading and applying them, out of a 21–27 min sync. The notification-driven path (`collect_new_scripts`, `FiltersBatch::collected_scripts`, the current-and-later batch rescan at commit) duplicated `reconcile_untested_scripts`, which asks the wallet what it watches, a superset of any `new_scripts`, and re-tests whatever the batch never matched. `rescan_complete` was set immediately before the batch left `active_batches`, so it was never read as true. Trade-off: the #846 case, a CoinJoin output paying an index derived only after its batch committed, is no longer recovered. Its repro, `coinjoin_gap_limit_stall_across_committed_batch`, fails without the sweep and is removed with it, as is the sweep-cost test. The bench wallet does not hit that case. Validated on top of the spend re-application fix: 5 consecutive mainnet restores at 100 Mbit / 100 ms, then 100/50 Mbit x 100/500 ms, all ending at 7112 records, 14 114 383 sat and 13 389 addresses, in 6.0–17.2 min. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017DruChNTWXwJoWPartZwCf --- dash-spv/src/sync/filters/batch.rs | 28 -- .../src/sync/filters/block_match_tracker.rs | 5 - .../filters/coinjoin_gap_discovery_tests.rs | 275 +---------- dash-spv/src/sync/filters/manager.rs | 466 +----------------- dash-spv/src/sync/filters/sync_manager.rs | 11 +- 5 files changed, 13 insertions(+), 772 deletions(-) diff --git a/dash-spv/src/sync/filters/batch.rs b/dash-spv/src/sync/filters/batch.rs index 3fd609b79..67263ef3b 100644 --- a/dash-spv/src/sync/filters/batch.rs +++ b/dash-spv/src/sync/filters/batch.rs @@ -22,8 +22,6 @@ pub(super) struct FiltersBatch { scanned: bool, /// Number of blocks still being downloaded for this batch. pending_blocks: u32, - /// Whether rescan has been completed for this batch. - rescan_complete: bool, /// Wallets that were behind for this batch's height range at scan time — /// and therefore need their `synced_height` advanced when the batch /// commits — each mapped to the wallet's `account_generation` at scan @@ -33,10 +31,6 @@ pub(super) struct FiltersBatch { /// current account set (dashpay/rust-dashcore#649). Already-synced wallets /// must not be touched. scanned_wallets: BTreeMap, - /// Cached scriptPubKeys discovered during block processing that still - /// need rescan, attributed per wallet so we can rerun matching only - /// against the wallet that produced each new script. - collected_scripts: HashMap>, /// Every script already matched against this batch's filters, per wallet. tested_scripts: HashMap>, } @@ -55,9 +49,7 @@ impl FiltersBatch { verified: false, scanned: false, pending_blocks: 0, - rescan_complete: false, scanned_wallets: BTreeMap::new(), - collected_scripts: HashMap::new(), tested_scripts: HashMap::new(), } } @@ -106,22 +98,6 @@ impl FiltersBatch { self.pending_blocks = self.pending_blocks.saturating_sub(1); self.pending_blocks } - /// Returns whether rescan has been completed for this batch. - pub(super) fn rescan_complete(&self) -> bool { - self.rescan_complete - } - /// Mark rescan as complete for this batch. - pub(super) fn mark_rescan_complete(&mut self) { - self.rescan_complete = true; - } - /// Add scriptPubKeys discovered during block processing for later rescan. - pub(super) fn add_scripts_for_wallet( - &mut self, - wallet_id: WalletId, - scripts: impl IntoIterator, - ) { - self.collected_scripts.entry(wallet_id).or_default().extend(scripts); - } /// Record that `scripts` have been matched against this batch's filters. pub(super) fn mark_tested>( &mut self, @@ -141,10 +117,6 @@ impl FiltersBatch { monitored.iter().filter(move |script| tested.is_none_or(|t| !t.contains(*script))) } - /// Take collected per-wallet scripts for rescan, leaving the map empty. - pub(super) fn take_collected_scripts(&mut self) -> HashMap> { - std::mem::take(&mut self.collected_scripts) - } /// Record the wallets that were behind for this batch at scan time, each /// with its `account_generation` snapshot. pub(super) fn set_scanned_wallets(&mut self, wallets: BTreeMap) { diff --git a/dash-spv/src/sync/filters/block_match_tracker.rs b/dash-spv/src/sync/filters/block_match_tracker.rs index eb81d5344..e0d911b47 100644 --- a/dash-spv/src/sync/filters/block_match_tracker.rs +++ b/dash-spv/src/sync/filters/block_match_tracker.rs @@ -148,11 +148,6 @@ impl BlockMatchTracker { self.processed_blocks_per_wallet.split_off(&(height + 1)); } - /// True while matched blocks are still awaiting their `BlockProcessed`. - pub(super) fn has_blocks_in_flight(&self) -> bool { - !self.blocks_remaining.is_empty() - } - /// True when there is no in-flight or processed-record state. pub(super) fn is_empty(&self) -> bool { self.blocks_remaining.is_empty() && self.processed_blocks_per_wallet.is_empty() diff --git a/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs b/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs index bf064fd97..235645464 100644 --- a/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs +++ b/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs @@ -12,7 +12,7 @@ //! `rescan_batch` now re-queues already-processed blocks of ACTIVE batches //! against newly derived scripts, to a fixpoint. //! -//! These tests pin both the fixed behaviour and the remaining hole: +//! These tests pin the fixed behaviour: //! //! 1. [`coinjoin_gap_limit_dense_same_batch_recovers`] — the empirical //! stall-at-59 shape (two dense blocks in one batch). GREEN since #820; @@ -20,16 +20,6 @@ //! 2. [`coinjoin_gap_limit_inversion_within_batch_recovers`] — a gap-window //! output in an EARLIER block of the SAME (still-active) batch is //! recovered by the commit-time rescan. GREEN since #820. -//! 3. [`coinjoin_gap_limit_stall_across_committed_batch`] — the SAME shape -//! with the earlier block in an already-COMMITTED batch (#846): rescans -//! only reach `active_batches`, and committed batches are gone. GREEN -//! since `rescan_committed_range` re-tests newly derived scripts against -//! the STORED filters below the committing batch. -//! 4. [`committed_range_sweep_coalesces_across_batch_commits`] — the cost -//! side of #846's fix: N script-carrying commits share one drain-time -//! committed-range sweep instead of walking the stored history once per -//! commit, while a late-derived script still finds its -//! committed-prefix block before `FiltersSyncComplete`. //! //! Each test drives the manager exactly the way the production event loop //! does: `try_process_batch` → `BlocksNeeded` → (blocks-manager stand-in) @@ -296,8 +286,7 @@ async fn coinjoin_gap_limit_dense_same_batch_recovers() { /// only at height 20. The initial scan cannot see block A (nothing watched /// matches it), but once block B's processing extends the window past G+21 /// the commit-time rescan re-tests block A's filter and recovers it. GREEN -/// since #820 — contrast for the cross-batch test below, isolating the -/// commit boundary as the broken seam. +/// since #820. #[tokio::test] async fn coinjoin_gap_limit_inversion_within_batch_recovers() { let (mut manager, wallet, wallet_id) = setup().await; @@ -324,263 +313,3 @@ async fn coinjoin_gap_limit_inversion_within_batch_recovers() { recovered by the commit-time rescan (PR #820). used_count={used_count}" ); } - -/// Gap-window outputs in an already-COMMITTED batch (#846). -/// -/// Same funding shape as the within-batch inversion test, but the early -/// block (indices G+10..=G+21, height 10) sits in batch 0..=99 while the -/// in-window block (indices 0..=29) sits at height 110 in batch 100..=199. -/// Batch 0 scans clean (nothing watched matches) and commits. Processing the -/// height-110 block extends the window past G+21, and those scripts DO match -/// block 10's filter — but `rescan_batch` only reaches `active_batches`, and -/// committed batches are gone (`try_commit_batches` removes them; the -/// tracker prunes at-or-below the committed height). Indices G+10..=G+21 — -/// squarely inside the BIP-44/CoinJoin gap-limit recovery contract -/// (G+21 < 29 + 1 + G) — used to stay invisible forever, along with their -/// funds; a fresh re-sync from genesis hit the same wall deterministically. -/// -/// GREEN since `rescan_committed_range`: newly derived scripts are re-tested -/// against the persisted filters below the committing batch (BIP-158 filters -/// are address-independent, so re-matching needs no re-download), and hits -/// flow through the `track_for_new_scripts` re-download path to the same -/// commit-time fixpoint. `highest_used` reaches G+21. -#[tokio::test] -async fn coinjoin_gap_limit_stall_across_committed_batch() { - let (mut manager, wallet, wallet_id) = setup().await; - let addresses = coinjoin_external_addresses(&wallet, &wallet_id, (G + 22) as u32).await; - - let (block_a, filter_a, key_a) = block_paying(10, &addresses[(G + 10)..=(G + 21)]); - let (block_b, filter_b, key_b) = block_paying(110, &addresses[0..=29]); - - // Uphold the production invariant the injected batches imply: every - // height at or below `stored_height` has its header and filter - // persisted (store_and_match_batches stores a batch's filters before - // stored_height advances past it). The committed-range recovery path - // re-tests exactly this stored data, so the invariant is load-bearing - // here: batch 0 commits before block B's processing derives the missing - // scripts, and by then its in-memory filters are gone. - { - let mut header_storage = manager.header_storage.write().await; - let mut filter_storage = manager.filter_storage.write().await; - for height in 0..=99u32 { - let (header, filter_bytes) = if height == 10 { - (block_a.header, filter_a.content.clone()) - } else { - let filler = Block::dummy(height, vec![]); - let filter = BlockFilter::dummy(&filler); - (filler.header, filter.content) - }; - header_storage - .store_headers_at_height(&[header.into()], height) - .await - .expect("seed header"); - filter_storage.store_filter(height, &filter_bytes).await.expect("seed filter"); - } - } - - let blocks: HashMap = - HashMap::from([(block_a.block_hash(), block_a), (block_b.block_hash(), block_b)]); - - let mut batch_0 = FiltersBatch::new(0, 99, HashMap::from([(key_a, filter_a)])); - batch_0.mark_verified(); - manager.active_batches.insert(0, batch_0); - let mut batch_1 = FiltersBatch::new(100, 199, HashMap::from([(key_b, filter_b)])); - batch_1.mark_verified(); - manager.active_batches.insert(100, batch_1); - manager.progress.update_stored_height(199); - - let initial_events = manager.try_process_batch().await.unwrap(); - drive_to_quiescence(&mut manager, &wallet, &blocks, initial_events).await; - - let (highest_used, highest_generated, used_count) = - coinjoin_pool_state(&wallet, &wallet_id).await; - // Sanity: the in-window block was found and the gap window extended past - // index G+21, so the missed indices ARE inside the watched range by now. - assert!( - highest_generated >= Some((G + 21) as u32), - "gap maintenance must have extended the watch window past index G+21 \ - (got {highest_generated:?})" - ); - assert_eq!( - highest_used, - Some((G + 21) as u32), - "CoinJoin External indices G+10..=G+21 were funded at height 10 in a batch that \ - committed before their scripts were derived, and the new-script rescan never \ - looks below the committed boundary (rescan_batch only reaches active_batches; \ - BlockMatchTracker/commit pruning drops the range). The addresses are within \ - the gap-limit recovery contract and are watched now (highest_generated = \ - {highest_generated:?}), yet their outputs stay invisible: highest_used stalls \ - at {highest_used:?}, used_count={used_count}. Fix direction: key re-scan \ - suppression by (wallet, address/script) instead of block/commit progress, or \ - trigger a below-committed-height rescan for a wallet whose gap maintenance \ - derives scripts mid-sync." - ); -} - -/// Committed-range sweeps coalesce across batch commits. -/// -/// Four batches; the last three each contain one in-window block whose -/// processing derives new scripts (each funds the next run of CoinJoin -/// indices, so gap maintenance extends the watch window at every commit). -/// Per-commit sweeping walks the entire committed prefix once per -/// script-carrying commit — on a real mainnet restore that shape produced -/// 191 full-prefix sweeps totalling ~14.5 minutes. Coalesced, the -/// accumulated scripts cross the stored history when the forward pipeline -/// drains: one sweep, plus one follow-up round for the scripts derived from -/// the block that sweep recovers. -/// -/// The early block (height 10, beyond-window indices G+10..=G+21, unwatched -/// when its batch scans and commits) pins the #846 correctness contract at -/// the same time: the deferred sweep must still find it, and -/// `FiltersSyncComplete` must not be emitted before it has been found and -/// applied. `committed_range_sweeps` counts sweeps that reach the chunk walk -/// in `rescan_committed_range`. -#[tokio::test] -async fn committed_range_sweep_coalesces_across_batch_commits() { - let (mut manager, wallet, wallet_id) = setup().await; - let addresses = coinjoin_external_addresses(&wallet, &wallet_id, (G + 22) as u32).await; - - // Beyond-window block in the range that commits first (#846 shape). - let (block_early, filter_early, key_early) = block_paying(10, &addresses[(G + 10)..=(G + 21)]); - // One in-window block per later batch. Each extends `highest_used` by 30, - // so every one of these batches carries newly derived scripts into its - // commit. After block 110 the generated window reaches 29 + G >= G + 21, - // so the early block's scripts are among the first commit's derivations. - let (block_1, filter_1, key_1) = block_paying(110, &addresses[0..=29]); - let (block_2, filter_2, key_2) = block_paying(210, &addresses[30..=59]); - let (block_3, filter_3, key_3) = block_paying(310, &addresses[60..=89]); - - // Persist headers and filters for the committed prefix 0..=299 — the - // range the drain-time sweep reloads from storage. Real data at the - // three block heights, filler elsewhere. - { - let mut header_storage = manager.header_storage.write().await; - let mut filter_storage = manager.filter_storage.write().await; - for height in 0..=299u32 { - let (header, filter_bytes) = match height { - 10 => (block_early.header, filter_early.content.clone()), - 110 => (block_1.header, filter_1.content.clone()), - 210 => (block_2.header, filter_2.content.clone()), - _ => { - let filler = Block::dummy(height, vec![]); - let filter = BlockFilter::dummy(&filler); - (filler.header, filter.content) - } - }; - header_storage - .store_headers_at_height(&[header.into()], height) - .await - .expect("seed header"); - filter_storage.store_filter(height, &filter_bytes).await.expect("seed filter"); - } - } - - let blocks: HashMap = HashMap::from([ - (block_early.block_hash(), block_early), - (block_1.block_hash(), block_1), - (block_2.block_hash(), block_2), - (block_3.block_hash(), block_3), - ]); - - for (start, filters) in [ - (0u32, HashMap::from([(key_early, filter_early)])), - (100, HashMap::from([(key_1, filter_1)])), - (200, HashMap::from([(key_2, filter_2)])), - (300, HashMap::from([(key_3, filter_3)])), - ] { - let mut batch = FiltersBatch::new(start, start + 99, filters); - batch.mark_verified(); - manager.active_batches.insert(start, batch); - } - manager.progress.update_stored_height(399); - // Both halves of the drain gate have to be live, or the test only proves - // the `active_batches.len() == 1` half: with the tips left at their - // default 0, `end_height() >= filter_header_tip_height()` is trivially - // true and the comparison that keeps the sweep from firing while more - // batches are still to come is never exercised. Same for the target - // height, which `FiltersSyncComplete` is checked against below. - manager.progress.update_filter_header_tip_height(399); - manager.progress.update_target_height(399); - - // Drive the production event loop to quiescence like `drive_to_quiescence` - // does, additionally watching for `FiltersSyncComplete` so the completion - // contract can be asserted at the moment it is emitted. - let (tx, _rx) = unbounded_channel(); - let requests = RequestSender::new(tx); - let mut events = manager.try_process_batch().await.unwrap(); - let mut sync_complete_seen = false; - 'rounds: for _round in 0..64 { - let mut pending: BTreeMap<(u32, BlockHash), BTreeSet> = BTreeMap::new(); - for event in events.drain(..) { - match event { - SyncEvent::BlocksNeeded { - blocks: needed, - } => { - for (key, wallets) in needed { - pending.entry((key.height(), *key.hash())).or_default().extend(wallets); - } - } - SyncEvent::FiltersSyncComplete { - .. - } => { - let (highest_used, _, _) = coinjoin_pool_state(&wallet, &wallet_id).await; - assert_eq!( - highest_used, - Some((G + 21) as u32), - "FiltersSyncComplete emitted before the deferred committed-range \ - sweep recovered the height-10 block: a script derived after its \ - range committed was never tested against the committed prefix" - ); - sync_complete_seen = true; - } - _ => {} - } - } - if pending.is_empty() { - break 'rounds; - } - for ((height, block_hash), wallets) in pending { - let block = blocks.get(&block_hash).expect("BlocksNeeded for an unknown test block"); - let result = wallet - .write() - .await - .process_block_for_wallets(block, block_hash, height, &wallets) - .await; - let confirmed_txids = result.relevant_txids().cloned().collect(); - let event = SyncEvent::BlockProcessed { - block_hash, - height, - wallets, - new_scripts: result.new_scripts, - confirmed_txids, - }; - events.extend( - manager.handle_sync_event(&event, &requests).await.expect("BlockProcessed"), - ); - } - } - - assert!(sync_complete_seen, "the run must reach FiltersSyncComplete"); - - let (highest_used, _, used_count) = coinjoin_pool_state(&wallet, &wallet_id).await; - assert_eq!( - highest_used, - Some((G + 21) as u32), - "the height-10 block's beyond-window outputs must be recovered by the \ - drain-time committed-range sweep (used_count={used_count})" - ); - assert_eq!(used_count, 90 + 12, "indices 0..=89 and G+10..=G+21 must all be marked used"); - - assert!( - manager.committed_range_sweeps >= 1, - "the deferred committed-range sweep must still run before completion" - ); - assert!( - manager.committed_range_sweeps <= 2, - "three script-carrying batch commits must share the committed-range sweep \ - (one drain-time sweep plus one follow-up round for the scripts derived from \ - the recovered block); got {} sweeps — one sweep per commit means the \ - coalescing regressed", - manager.committed_range_sweeps - ); -} diff --git a/dash-spv/src/sync/filters/manager.rs b/dash-spv/src/sync/filters/manager.rs index f5e0cdf05..be77c0668 100644 --- a/dash-spv/src/sync/filters/manager.rs +++ b/dash-spv/src/sync/filters/manager.rs @@ -91,21 +91,6 @@ pub struct FiltersManager< /// `BlockProcessed` and the per-wallet record of which wallets already /// have a given processed block applied. pub(super) tracker: BlockMatchTracker, - /// Scripts already forward-rescanned but still awaiting the combined - /// backward sweep over the committed range (#846). Held at the manager - /// level and carried across batch commits: intermediate commits only - /// accumulate here, and `try_commit_batches` runs the sweep when the - /// forward pipeline drains — the committing batch is the last one active - /// and no lookahead can extend past it — so script-carrying commits share - /// one walk of the stored history instead of walking it once per commit. - /// Deliberately survives `reset_for_rescan`: the restarted scan covers - /// heights above its entry point, while these scripts still owe a pass - /// over the committed prefix below it. - backward_scripts: HashMap>, - /// Number of committed-range sweeps that reached the chunk walk in - /// `rescan_committed_range`. Diagnostic counter; the sweep-coalescing - /// regression test asserts on it. - pub(super) committed_range_sweeps: u64, } impl @@ -150,8 +135,6 @@ impl= self.progress.filter_header_tip_height() && self.progress.committed_height() >= self.progress.target_height() { - // A block delivered after its batch committed derives scripts with - // no active batch to route them to, so they land in the accumulator - // instead. It is in-memory only and nothing looks below the - // committed frontier again, so sweep here rather than wait for a - // next commit that may never come. - if !self.backward_scripts.is_empty() { - let backward_scripts = std::mem::take(&mut self.backward_scripts); - let sweep_start = self.progress.committed_height().saturating_add(1); - events.extend(self.rescan_committed_range(sweep_start, &backward_scripts).await?); - } - - // Blocks that sweep found are charged to no batch, so the commit - // gate cannot hold completion — gate on the tracker. Their - // `BlockProcessed` re-enters here, and a round deriving no new - // scripts is the fixpoint. - if self.tracker.has_blocks_in_flight() { - return Ok(events); - } - // Blocks applied after the last commit leave processed records that // no commit will prune. self.tracker.prune_at_or_below(self.progress.committed_height()); @@ -620,89 +579,11 @@ impl = self - .active_batches - .iter() - .filter(|(&start, batch)| start > batch_start && batch.scanned()) - .map(|(&start, _)| start) - .collect(); - - for later_start in later_batches { - events.extend(self.rescan_batch(later_start, &scripts_by_wallet).await?); - } - - // Newly derived scripts also have to reach ranges that - // already committed: those blocks were matched against a - // watch set that predates these scripts, and nothing else - // ever looks below `committed_height` again (#846). That - // backward sweep walks stored history — the expensive - // direction — so defer it: accumulate the scripts at the - // manager level, across batch commits, and sweep below. - for (wallet_id, scripts) in scripts_by_wallet { - self.backward_scripts.entry(wallet_id).or_default().extend(scripts); - } - if let Some(batch) = self.active_batches.get(&batch_start) { - if batch.pending_blocks() > 0 { - // Forward rescan found blocks; converge the - // forward direction first. - break; - } - } - } - - events.extend(self.reconcile_untested_scripts(batch_start).await?); - if let Some(batch) = self.active_batches.get(&batch_start) { - if batch.pending_blocks() > 0 { - // Reconciliation found blocks; converge before committing. - break; - } - } - - // The backward sweep waits for the forward pipeline to drain: - // it runs only when this batch is the last one active and no - // lookahead batch can be created past it. Commits before that - // point leave the accumulated scripts in place, so a sync's - // script-carrying commits share one walk of the stored - // history — plus one walk per follow-up round whose block - // processing derives genuinely new scripts — instead of - // walking it once per commit. Hits attribute to this batch, - // so scripts their processing derives re-enter through - // `collected_scripts` above and only genuinely new scripts - // get a follow-up sweep. - let forward_drained = self.active_batches.len() == 1 - && self.active_batches.get(&batch_start).is_some_and(|b| { - b.end_height() >= self.progress.filter_header_tip_height() - }); - if forward_drained && !self.backward_scripts.is_empty() { - let backward_scripts = std::mem::take(&mut self.backward_scripts); - events - .extend(self.rescan_committed_range(batch_start, &backward_scripts).await?); - - // Check if the backward sweep found more blocks - if let Some(batch) = self.active_batches.get(&batch_start) { - if batch.pending_blocks() > 0 { - // Found more blocks, can't commit yet - break; - } - } - } - // Mark rescan as complete - if let Some(batch) = self.active_batches.get_mut(&batch_start) { - batch.mark_rescan_complete(); + events.extend(self.reconcile_untested_scripts(batch_start).await?); + if let Some(batch) = self.active_batches.get(&batch_start) { + if batch.pending_blocks() > 0 { + // Reconciliation found blocks; converge before committing. + break; } } @@ -856,49 +737,9 @@ impl>, - ) -> usize { - let target = self - .active_batches - .range(..=height) - .next_back() - .filter(|(_, batch)| batch.end_height() >= height) - .map(|(&start, _)| start); - - let mut routed = 0; - for (wallet_id, scripts) in new_scripts { - if scripts.is_empty() { - continue; - } - routed += scripts.len(); - match target.and_then(|start| self.active_batches.get_mut(&start)) { - Some(batch) => batch.add_scripts_for_wallet(*wallet_id, scripts.iter().cloned()), - // Its range has committed: only the sweep can still test these. - None => self - .backward_scripts - .entry(*wallet_id) - .or_default() - .extend(scripts.iter().cloned()), - } - } - routed - } - /// Re-test anything the wallet watches that this batch has never been - /// matched against, before letting it commit. Closing the loop on wallet - /// state does not depend on a `new_scripts` notification arriving; a - /// script the wallet has dropped from `scan_script_pubkeys_for` is still - /// not re-tested. + /// matched against, before letting it commit. A script the wallet has + /// dropped from `scan_script_pubkeys_for` is not re-tested. async fn reconcile_untested_scripts(&mut self, batch_start: u32) -> SyncResult> { let Some(batch) = self.active_batches.get(&batch_start) else { return Ok(vec![]); @@ -933,11 +774,7 @@ impl(), ); // Same accounting as any other newly derived script. - let events = self.rescan_batch(batch_start, &untested).await?; - for (wallet_id, scripts) in untested { - self.backward_scripts.entry(wallet_id).or_default().extend(scripts); - } - Ok(events) + self.rescan_batch(batch_start, &untested).await } /// Rescan a specific batch for newly discovered scriptPubKeys, attributed @@ -1012,26 +849,6 @@ impl>, - context: &str, - ) -> Vec { let mut events = Vec::new(); let mut blocks_needed: BTreeMap> = BTreeMap::new(); let mut new_blocks_count = 0; @@ -1062,7 +879,7 @@ impl>, - ) -> SyncResult> { - let Some(range_end) = batch_start.checked_sub(1) else { - return Ok(vec![]); - }; - - let wallet_queries: Vec<(WalletId, Vec)> = new_scripts - .iter() - .filter(|(_, scripts)| !scripts.is_empty()) - .map(|(id, scripts)| (*id, scripts.iter().cloned().collect())) - .collect(); - if wallet_queries.is_empty() { - return Ok(vec![]); - } - - // Nothing relevant can precede the earliest wallet birth height, and - // nothing is loadable below the first stored filter. - let wallet_base = self.wallet.read().await.earliest_required_height().await; - let Some(filter_base) = self.filter_storage.read().await.filter_start_height().await else { - return Ok(vec![]); - }; - let range_start = wallet_base.max(filter_base); - if range_start > range_end { - return Ok(vec![]); - } - - self.committed_range_sweeps += 1; - tracing::info!( - "Rescan committed filters ({}-{}) for new scripts across {} wallets (sweep #{})", - range_start, - range_end, - wallet_queries.len(), - self.committed_range_sweeps - ); - - let mut block_to_wallets: BTreeMap> = BTreeMap::new(); - let mut chunk_start = range_start; - while chunk_start <= range_end { - let chunk_end = (chunk_start + BATCH_PROCESSING_SIZE - 1).min(range_end); - // A chunk the storage cannot serve (e.g. filters pruned or never - // stored for a sub-range) is skipped rather than failing the - // commit: the sweep is best-effort recovery over whatever - // history is locally available. - let filters = match self.load_filters(chunk_start, chunk_end).await { - Ok(filters) => filters, - Err(e) => { - tracing::warn!( - "Committed-range rescan skipping {}-{}: {}", - chunk_start, - chunk_end, - e - ); - chunk_start = chunk_end + 1; - continue; - } - }; - for (wallet_id, scripts) in &wallet_queries { - let matches = check_compact_filters_for_elements(&filters, scripts, &[], 0); - for key in matches { - block_to_wallets.entry(key).or_default().insert(*wallet_id); - } - } - chunk_start = chunk_end + 1; - } - - Ok(self.queue_new_script_matches(batch_start, block_to_wallets, "Committed-range rescan")) - } - /// Handle notification that new filter headers are available. /// Used by both FilterHeadersSyncComplete and FilterHeadersStored events. pub(super) async fn handle_new_filter_headers( @@ -1831,7 +1552,6 @@ mod tests { let mut batch1 = FiltersBatch::new(0, 4999, HashMap::new()); batch1.set_pending_blocks(0); batch1.mark_scanned(); - batch1.mark_rescan_complete(); manager.active_batches.insert(0, batch1); @@ -1858,7 +1578,6 @@ mod tests { let mut batch1 = FiltersBatch::new(0, 4999, HashMap::new()); batch1.set_pending_blocks(0); batch1.mark_scanned(); - batch1.mark_rescan_complete(); batch1.set_scanned_wallets(BTreeMap::from([(MOCK_WALLET_ID, 0)])); manager.active_batches.insert(0, batch1); @@ -1873,7 +1592,6 @@ mod tests { let mut batch2 = FiltersBatch::new(5000, 9999, HashMap::new()); batch2.set_pending_blocks(0); batch2.mark_scanned(); - batch2.mark_rescan_complete(); manager.active_batches.insert(5000, batch2); manager.try_commit_batches().await.unwrap(); @@ -1902,7 +1620,6 @@ mod tests { let mut batch = FiltersBatch::new(0, 4999, HashMap::new()); batch.set_pending_blocks(0); batch.mark_scanned(); - batch.mark_rescan_complete(); batch.set_scanned_wallets(BTreeMap::from([(wallet_a, 0)])); manager.active_batches.insert(0, batch); @@ -1940,7 +1657,6 @@ mod tests { let mut batch = FiltersBatch::new(5000, 9999, HashMap::new()); batch.set_pending_blocks(0); batch.mark_scanned(); - batch.mark_rescan_complete(); batch.set_scanned_wallets(BTreeMap::from([(wallet_a, 0)])); manager.active_batches.insert(5000, batch); @@ -1981,7 +1697,6 @@ mod tests { let mut batch = FiltersBatch::new(205_000, 209_999, HashMap::new()); batch.set_pending_blocks(0); batch.mark_scanned(); - batch.mark_rescan_complete(); batch.set_scanned_wallets(BTreeMap::from([(wallet_b, 0)])); manager.active_batches.insert(205_000, batch); @@ -2010,7 +1725,6 @@ mod tests { let mut batch1 = FiltersBatch::new(0, 4999, HashMap::new()); batch1.set_pending_blocks(0); batch1.mark_scanned(); - batch1.mark_rescan_complete(); batch1.set_scanned_wallets(BTreeMap::from([(wallet_a, 0)])); manager.active_batches.insert(0, batch1); @@ -2022,7 +1736,6 @@ mod tests { let mut batch2 = FiltersBatch::new(5000, 9999, HashMap::new()); batch2.set_pending_blocks(0); batch2.mark_scanned(); - batch2.mark_rescan_complete(); batch2.set_scanned_wallets(BTreeMap::from([(wallet_a, 0)])); manager.active_batches.insert(5000, batch2); @@ -2068,7 +1781,6 @@ mod tests { let mut batch = FiltersBatch::new(5000, 9999, HashMap::new()); batch.set_pending_blocks(0); batch.mark_scanned(); - batch.mark_rescan_complete(); batch.set_scanned_wallets(BTreeMap::from([(wallet_a, 0), (wallet_b, 0)])); manager.active_batches.insert(5000, batch); @@ -2118,7 +1830,6 @@ mod tests { let mut batch = FiltersBatch::new(5000, 9999, HashMap::new()); batch.set_pending_blocks(0); batch.mark_scanned(); - batch.mark_rescan_complete(); batch.set_scanned_wallets(BTreeMap::from([(wallet_a, 0)])); manager.active_batches.insert(5000, batch); @@ -2175,7 +1886,6 @@ mod tests { let mut batch = FiltersBatch::new(1000, 5999, HashMap::new()); batch.set_pending_blocks(0); batch.mark_scanned(); - batch.mark_rescan_complete(); batch.set_scanned_wallets(BTreeMap::from([(wallet_a, 0)])); manager.active_batches.insert(1000, batch); @@ -2205,7 +1915,6 @@ mod tests { let mut batch = FiltersBatch::new(200_000, 204_999, HashMap::new()); batch.set_pending_blocks(0); batch.mark_scanned(); - batch.mark_rescan_complete(); batch.set_scanned_wallets(BTreeMap::from([(wallet_b, 0)])); manager.active_batches.insert(200_000, batch); @@ -2243,7 +1952,6 @@ mod tests { let mut batch = FiltersBatch::new(0, 4999, HashMap::new()); batch.set_pending_blocks(0); batch.mark_scanned(); - batch.mark_rescan_complete(); batch.set_scanned_wallets(BTreeMap::from([(wallet_a, 3)])); manager.active_batches.insert(0, batch); @@ -2612,7 +2320,6 @@ mod tests { // Mark batch ready so commit can run, then commit. if let Some(b) = manager.active_batches.get_mut(&0) { b.set_pending_blocks(0); - b.mark_rescan_complete(); } manager.try_commit_batches().await.unwrap(); @@ -2806,7 +2513,6 @@ mod tests { let mut batch = FiltersBatch::new(0, 4999, HashMap::new()); batch.set_pending_blocks(0); batch.mark_scanned(); - batch.mark_rescan_complete(); manager.active_batches.insert(0, batch); manager.try_commit_batches().await.unwrap(); @@ -2932,12 +2638,10 @@ mod tests { let mut batch1 = FiltersBatch::new(0, 4999, HashMap::new()); batch1.set_pending_blocks(0); batch1.mark_scanned(); - batch1.mark_rescan_complete(); let mut batch2 = FiltersBatch::new(5000, 9999, HashMap::new()); batch2.set_pending_blocks(0); batch2.mark_scanned(); - batch2.mark_rescan_complete(); manager.active_batches.insert(5000, batch2); // Insert higher one first manager.active_batches.insert(0, batch1); @@ -3063,73 +2767,6 @@ mod tests { assert!(manager.is_idle()); } - #[tokio::test] - async fn test_batch_collects_scripts() { - use crate::sync::filters::batch::FiltersBatch; - use dashcore::Network; - - let mut batch = FiltersBatch::new(0, 4999, HashMap::new()); - - // Initially empty - assert!(batch.take_collected_scripts().is_empty()); - - // Add scripts using test utility - let script1 = dashcore::Address::dummy(Network::Testnet, 1).script_pubkey(); - let script2 = dashcore::Address::dummy(Network::Testnet, 2).script_pubkey(); - let wallet_id: WalletId = [7; 32]; - - batch.add_scripts_for_wallet(wallet_id, [script1.clone(), script2.clone()]); - - let collected = batch.take_collected_scripts(); - let for_wallet = collected.get(&wallet_id).expect("wallet entry"); - assert_eq!(for_wallet.len(), 2); - assert!(for_wallet.contains(&script1)); - assert!(for_wallet.contains(&script2)); - - // After take, should be empty - assert!(batch.take_collected_scripts().is_empty()); - } - - /// Proves that scripts from a block with no in-flight record — every - /// delivery after the first — reach the batch covering their height, and - /// the backward accumulator when no active batch does. - #[tokio::test] - async fn test_new_scripts_route_by_height_with_no_in_flight_record() { - use crate::sync::filters::batch::FiltersBatch; - use dashcore::Network; - - let mut manager = create_test_manager().await; - manager.active_batches.insert(0, FiltersBatch::new(0, 4999, HashMap::new())); - manager.active_batches.insert(5000, FiltersBatch::new(5000, 9999, HashMap::new())); - - let wallet_id: WalletId = [7; 32]; - let covered = Address::dummy(Network::Testnet, 1).script_pubkey(); - let below = Address::dummy(Network::Testnet, 2).script_pubkey(); - - let (tx, _rx) = unbounded_channel(); - let requests = RequestSender::new(tx); - let block_processed = |height: u32, script: &ScriptBuf| SyncEvent::BlockProcessed { - block_hash: Header::dummy(height).block_hash(), - height, - wallets: BTreeSet::from([wallet_id]), - new_scripts: BTreeMap::from([(wallet_id, vec![script.clone()])]), - confirmed_txids: vec![], - }; - - manager.handle_sync_event(&block_processed(6000, &covered), &requests).await.unwrap(); - - let lower = manager.active_batches.get_mut(&0).unwrap().take_collected_scripts(); - assert!(lower.is_empty(), "scripts must not land in a batch that cannot match them"); - let covering = manager.active_batches.get_mut(&5000).unwrap().take_collected_scripts(); - assert!(covering.get(&wallet_id).is_some_and(|s| s.contains(&covered))); - assert!(manager.backward_scripts.is_empty()); - - manager.active_batches.remove(&0); - manager.handle_sync_event(&block_processed(120, &below), &requests).await.unwrap(); - assert!(manager.active_batches.get_mut(&5000).unwrap().take_collected_scripts().is_empty()); - assert!(manager.backward_scripts.get(&wallet_id).is_some_and(|s| s.contains(&below))); - } - /// Proves a batch with no filters still marks the scripts tested, so /// reconciliation does not hand it the same set on every commit attempt. #[tokio::test] @@ -3208,89 +2845,6 @@ mod tests { ); } - /// Proves the tip is not reported synced while scripts sit in the backward - /// accumulator with no batch left to sweep them: the sweep runs here, and - /// completion waits for the block it finds. - #[tokio::test] - async fn test_tip_sweeps_stranded_backward_scripts_before_completing() { - let wallet_id: WalletId = [3; 32]; - let watched = Address::dummy(Network::Testnet, 31); - - let mut multi = MultiMockWallet::new(); - multi.insert_wallet( - wallet_id, - MockWalletState { - addresses: vec![watched.clone()], - synced_height: 9, - last_processed_height: 9, - account_generation: 0, - }, - ); - let mut manager = create_multi_test_manager(Arc::new(RwLock::new(multi))).await; - manager.set_state(SyncState::Syncing); - - // Every height at or below the committed frontier has its header and - // filter persisted; only height 4 pays `watched`. - let paying = Block::dummy(4, vec![Transaction::dummy(&watched, 0..0, &[4])]); - let paying_filter = BlockFilter::dummy(&paying); - let key = FilterMatchKey::new(4, paying.block_hash()); - { - let mut header_storage = manager.header_storage.write().await; - let mut filter_storage = manager.filter_storage.write().await; - for height in 0..=9u32 { - let (header, bytes) = if height == 4 { - (paying.header, paying_filter.content.clone()) - } else { - let filler = Block::dummy(height, vec![]); - (filler.header, BlockFilter::dummy(&filler).content) - }; - header_storage.store_headers_at_height(&[header.into()], height).await.unwrap(); - filter_storage.store_filter(height, &bytes).await.unwrap(); - } - } - - // At the tip with nothing active: the last commit left - // `processing_height` past the tip, so no lookahead batch is created - // and the completion branch is reached. - manager.processing_height = 10; - manager.progress.update_stored_height(9); - manager.progress.update_committed_height(9); - manager.progress.update_filter_header_tip_height(9); - manager.progress.update_target_height(9); - manager.backward_scripts.entry(wallet_id).or_default().insert(watched.script_pubkey()); - - let events = manager.try_process_batch().await.unwrap(); - assert!( - events.iter().any(|e| match e { - SyncEvent::BlocksNeeded { - blocks, - } => blocks.contains_key(&key), - _ => false, - }), - "the tip sweep must find the block paying the stranded script" - ); - assert!( - !events.iter().any(|e| matches!(e, SyncEvent::FiltersSyncComplete { .. })), - "completion must wait for that block to be applied" - ); - assert!(manager.backward_scripts.is_empty(), "the sweep consumes the accumulator"); - - let (tx, _rx) = unbounded_channel(); - let requests = RequestSender::new(tx); - let processed = SyncEvent::BlockProcessed { - block_hash: paying.block_hash(), - height: 4, - wallets: BTreeSet::from([wallet_id]), - new_scripts: BTreeMap::new(), - confirmed_txids: vec![], - }; - let events = manager.handle_sync_event(&processed, &requests).await.unwrap(); - assert!( - events.iter().any(|e| matches!(e, SyncEvent::FiltersSyncComplete { .. })), - "completion follows once the sweep's block is applied" - ); - } - #[tokio::test] async fn test_start_download_waits_when_filter_headers_insufficient() { let mut manager = create_test_manager().await; diff --git a/dash-spv/src/sync/filters/sync_manager.rs b/dash-spv/src/sync/filters/sync_manager.rs index 8e763b8b3..3a3b9cc69 100644 --- a/dash-spv/src/sync/filters/sync_manager.rs +++ b/dash-spv/src/sync/filters/sync_manager.rs @@ -174,7 +174,6 @@ impl< block_hash, height, wallets, - new_scripts, .. } => { // Record per-wallet processing so a future scan can give a @@ -183,8 +182,7 @@ impl< self.tracker.record_processed(*height, *block_hash, wallets); // Check if this block is part of our tracked blocks - let in_flight = self.tracker.finish_in_flight(block_hash); - if let Some((_, batch_start)) = in_flight { + if let Some((_, batch_start)) = self.tracker.finish_in_flight(block_hash) { if let Some(batch) = self.active_batches.get_mut(&batch_start) { batch.decrement_pending_blocks(); tracing::debug!( @@ -195,13 +193,6 @@ impl< batch.pending_blocks() ); } - } - - // Outside the in-flight arm on purpose: that record is consumed - // by the first delivery, and a block is delivered more than once. - let derived = self.collect_new_scripts(*height, new_scripts); - - if in_flight.is_some() || derived > 0 { return self.try_process_batch().await; } }