From 08c7e2d6d6d5f602f12bf91794f3311a67801c3d Mon Sep 17 00:00:00 2001 From: pasta Date: Wed, 12 Aug 2026 20:40:14 -0500 Subject: [PATCH 1/2] fix(dash-spv): verify filter batches against the full script set before commit A filter batch used to commit once the gap-limit chase's latest wave derived no new scripts. That signal is wrong twice over: scripts derived from blocks owned by other batches never enter this batch's collected set, and a wave whose visible transactions all pay already-derived indices (backfill - inevitable when dust waves are mined out of derivation order) derives nothing even though the batch still holds blocks paying indices past the window. Committing on it ends the chase early, and committed batches are never rescanned, so every transaction above the window at commit time is permanently lost and discovery flatlines: later activity pays underived addresses, filters stop matching, and the pool never extends again. Observed in production on a mainnet wallet with 345,261 sequentially used addresses: discovery froze at address index 2,400 after ~500 blocks of activity and silently missed 880k+ transactions while sync ran to tip. The manager now keeps a monotone script-derivation generation, bumped whenever block processing derives new scripts - including for blocks whose owning batch is already gone, whose scripts were previously dropped without ever being matched. Each batch records the generation it was last matched against the wallets' full script sets (initial scan or verification). At commit time, a batch whose recorded generation is stale is re-matched against the full current sets and may not complete its rescan while that verification still finds blocks. Quiet syncs never pay for this: with no derivations the generations match and the verification is skipped. The regression test drives the real FiltersManager + WalletManager over a synthetic dust restore whose payments land out of derivation order across a batch boundary; without this fix it loses 36 of 3,000 transactions, with it discovery completes (as does the in-order control). Co-Authored-By: Claude Fable 5 --- dash-spv/src/sync/filters/batch.rs | 19 ++ dash-spv/src/sync/filters/manager.rs | 330 ++++++++++++++++++++++ dash-spv/src/sync/filters/sync_manager.rs | 9 + 3 files changed, 358 insertions(+) diff --git a/dash-spv/src/sync/filters/batch.rs b/dash-spv/src/sync/filters/batch.rs index b389c5a1b..cadc34cf1 100644 --- a/dash-spv/src/sync/filters/batch.rs +++ b/dash-spv/src/sync/filters/batch.rs @@ -37,6 +37,16 @@ pub(super) struct FiltersBatch { /// need rescan, attributed per wallet so we can rerun matching only /// against the wallet that produced each new script. collected_scripts: HashMap>, + /// The manager's script-derivation generation observed the last time + /// this batch's filters were matched against the wallets' FULL script + /// sets (the initial scan, or a commit-time verification rescan). + /// + /// Commit compares this against the current generation to decide + /// whether a verification rescan is still needed: scripts derived from + /// blocks owned by OTHER batches never land in this batch's + /// `collected_scripts`, so "no collected scripts left" alone does not + /// prove this batch has nothing more to match. + full_match_generation: u64, } impl FiltersBatch { @@ -56,6 +66,7 @@ impl FiltersBatch { rescan_complete: false, scanned_wallets: BTreeMap::new(), collected_scripts: HashMap::new(), + full_match_generation: 0, } } /// Start height of this batch (inclusive). @@ -107,6 +118,14 @@ impl FiltersBatch { pub(super) fn rescan_complete(&self) -> bool { self.rescan_complete } + /// The script-derivation generation at this batch's last full-set match. + pub(super) fn full_match_generation(&self) -> u64 { + self.full_match_generation + } + /// Record the script-derivation generation this batch was fully matched at. + pub(super) fn set_full_match_generation(&mut self, generation: u64) { + self.full_match_generation = generation; + } /// Mark rescan as complete for this batch. pub(super) fn mark_rescan_complete(&mut self) { self.rescan_complete = true; diff --git a/dash-spv/src/sync/filters/manager.rs b/dash-spv/src/sync/filters/manager.rs index 521af307c..c89d15575 100644 --- a/dash-spv/src/sync/filters/manager.rs +++ b/dash-spv/src/sync/filters/manager.rs @@ -91,6 +91,14 @@ pub struct FiltersManager< /// `BlockProcessed` and the per-wallet record of which wallets already /// have a given processed block applied. pub(super) tracker: BlockMatchTracker, + /// Monotone counter bumped every time block processing derives new + /// scripts (gap-limit pool extension), regardless of which batch — if + /// any — owns the block. Batches record the generation they were last + /// matched against the full script sets at + /// ([`FiltersBatch::full_match_generation`]); commit runs a + /// verification rescan whenever the generations differ, so a batch can + /// never commit while scripts it has not been matched against exist. + pub(super) script_generation: u64, } impl @@ -135,6 +143,7 @@ impl = self + .active_batches + .get(&batch_start) + .map(|b| b.scanned_wallets().keys().copied().collect()) + .unwrap_or_default(); + let mut full_sets: HashMap> = HashMap::new(); + { + let wallet = self.wallet.read().await; + for wallet_id in wallet_ids { + full_sets.insert( + wallet_id, + wallet.scan_script_pubkeys_for(&wallet_id).into_iter().collect(), + ); + } + } + events.extend(self.rescan_batch(batch_start, &full_sets).await?); + if let Some(batch) = self.active_batches.get_mut(&batch_start) { + // The rescan above matched the full sets as of + // `generation_now`; derivations triggered by any + // blocks it found will bump the generation again and + // re-arm the verification on the next pass. + batch.set_full_match_generation(generation_now); + if batch.pending_blocks() > 0 { + break; + } + } + } + // Mark rescan as complete if let Some(batch) = self.active_batches.get_mut(&batch_start) { batch.mark_rescan_complete(); @@ -902,6 +959,9 @@ impl`. + /// + /// The chain pays external addresses `0..n_addr` of the wallet inside a + /// dense activity window that straddles a `BATCH_PROCESSING_SIZE` + /// boundary, at a per-block address consumption far above the gap + /// limit. `shuffle_indices` controls whether payments land in derivation + /// order (the easy case) or shuffled across the window the way mempool + /// waves are actually mined. + /// + /// Returns `(txs_found, txs_expected)` once the sync reaches a fixpoint. + async fn run_dust_restore(shuffle_indices: bool) -> (usize, usize) { + use key_wallet::managed_account::managed_account_trait::ManagedAccountTrait; + use key_wallet::transaction_checking::transaction_router::AccountTypeToCheck; + use key_wallet::wallet::initialization::WalletAccountCreationOptions; + use key_wallet::wallet::managed_wallet_info::ManagedWalletInfo; + use key_wallet_manager::WalletManager; + use std::collections::VecDeque; + + const MNEMONIC: &str = "abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about"; + const TOTAL_HEIGHTS: u32 = 5120; // batch 1 = 0..4999, batch 2 = 5000..5119 + const ACTIVITY_START: u32 = 4940; + const ACTIVITY_BLOCKS: u32 = 120; // straddles the 4999/5000 batch boundary + const ADDRS_PER_BLOCK: u32 = 25; // burns ~25 fresh indices per block vs gap limit 30 + const ACTIVITY_END: u32 = ACTIVITY_START + ACTIVITY_BLOCKS - 1; + const N_ADDR: u32 = ACTIVITY_BLOCKS * ADDRS_PER_BLOCK; + + let network = Network::Testnet; + + // Wallet under test. + let mut wm: WalletManager = WalletManager::new(network); + let wallet_id = wm + .create_wallet_from_mnemonic(MNEMONIC, 0, WalletAccountCreationOptions::Default) + .expect("create wallet"); + let wallet = Arc::new(RwLock::new(wm)); + + // Throwaway manager from the same mnemonic just to derive the target + // external addresses 0..N_ADDR the "spammer" pays. + let mut deriv: WalletManager = WalletManager::new(network); + let deriv_id = deriv + .create_wallet_from_mnemonic(MNEMONIC, 0, WalletAccountCreationOptions::Default) + .expect("create derivation wallet"); + let key_source = deriv + .get_wallet(&deriv_id) + .unwrap() + .key_source_for_account_type(&AccountTypeToCheck::StandardBIP44, Some(0)); + let addresses: Vec
= { + let info = deriv.get_wallet_info_mut(&deriv_id).unwrap(); + let account = info.first_bip44_managed_account_mut().unwrap(); + let pool = account + .managed_account_type_mut() + .address_pools_mut() + .into_iter() + .find(|p| p.is_external()) + .expect("external pool"); + pool.address_range(0, N_ADDR, &key_source).expect("derive addresses") + }; + assert_eq!(addresses.len(), N_ADDR as usize); + + // Index-to-slot assignment: identity for the ordered case, or a + // deterministic Fisher-Yates driven by an LCG for the mined-out-of- + // order case (no rand dependency, fully reproducible). + let assignment: Vec = { + let mut v: Vec = (0..N_ADDR as usize).collect(); + if shuffle_indices { + let mut state: u64 = 0x243F_6A88_85A3_08D3; + for i in (1..v.len()).rev() { + state = + state.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407); + let j = (state >> 33) as usize % (i + 1); + v.swap(i, j); + } + } + v + }; + + // Build the chain: dust spam inside the activity window, unrelated + // single-tx blocks everywhere else. + let mut blocks: Vec = Vec::with_capacity(TOTAL_HEIGHTS as usize); + for h in 0..TOTAL_HEIGHTS { + let txs = if (ACTIVITY_START..=ACTIVITY_END).contains(&h) { + let i = h - ACTIVITY_START; + (0..ADDRS_PER_BLOCK) + .map(|j| { + let pos = (i * ADDRS_PER_BLOCK + j) as usize; + Transaction::dummy( + &addresses[assignment[pos]], + (j as u8)..(j as u8 + 1), + &[546], + ) + }) + .collect() + } else { + vec![Transaction::dummy( + &Address::dummy(network, (1_000_000u32 + h) as usize), + 0..1, + &[1000], + )] + }; + blocks.push(Block::dummy(h, txs)); + } + + // Seed storage: block headers + filter bodies for the whole range. + let storage = DiskStorageManager::with_temp_dir().await.unwrap(); + { + let hashed: Vec = + blocks.iter().map(|b| HashedBlockHeader::from(&b.header)).collect(); + storage + .block_headers() + .write() + .await + .store_headers_at_height(&hashed, 0) + .await + .unwrap(); + let filters_handle = storage.filters(); + let mut fs = filters_handle.write().await; + for (h, block) in blocks.iter().enumerate() { + let filter = BlockFilter::dummy(block); + fs.store_filter(h as u32, &filter.content).await.unwrap(); + } + } + + let mut manager: FiltersManager< + PersistentBlockHeaderStorage, + PersistentFilterHeaderStorage, + PersistentFilterStorage, + WalletManager, + > = FiltersManager::new( + wallet.clone(), + storage.block_headers(), + storage.filter_headers(), + storage.filters(), + ) + .await; + + let (tx, mut _rx) = unbounded_channel(); + let requests = RequestSender::new(tx); + + manager.set_state(SyncState::WaitingForConnections); + let mut queue: VecDeque = + manager.start_sync(&requests).await.expect("start_sync").into_iter().collect(); + + // Simulated block delivery: requested blocks arrive a few rounds + // later, lowest height first, mirroring BlocksPipeline's height- + // ordered processing. + let mut in_flight: BTreeMap, u64)> = + BTreeMap::new(); + + let mut guard = 0u64; + let mut round = 0u64; + let mut idle_rounds = 0u32; + loop { + round += 1; + guard += 1; + assert!(guard < 3_000_000, "sync loop failed to converge"); + while _rx.try_recv().is_ok() {} + + let mut had_activity = false; + while let Some(event) = queue.pop_front() { + had_activity = true; + match &event { + SyncEvent::BlocksNeeded { + blocks: needed, + } => { + for (key, wallets) in needed { + in_flight + .entry(key.height()) + .and_modify(|(_, w, _)| w.extend(wallets.iter().copied())) + .or_insert_with(|| (key.clone(), wallets.clone(), round + 3)); + } + } + SyncEvent::BlockProcessed { + .. + } => { + let evs = + manager.handle_sync_event(&event, &requests).await.expect("handle"); + queue.extend(evs); + } + _ => {} + } + } + + // Production ticks every 100ms regardless of pending downloads. + let evs = manager.tick(&requests).await.expect("tick"); + if !evs.is_empty() { + had_activity = true; + queue.extend(evs); + } + + // Deliver up to 4 ready blocks, lowest height first. + let ready: Vec = in_flight + .iter() + .filter(|(_, (_, _, ready_at))| *ready_at <= round) + .map(|(h, _)| *h) + .take(4) + .collect(); + for h in ready { + let (key, wallets, _) = in_flight.remove(&h).unwrap(); + let block = &blocks[key.height() as usize]; + let mut w = wallet.write().await; + let result = + w.process_block_for_wallets(block, *key.hash(), key.height(), &wallets).await; + drop(w); + had_activity = true; + let confirmed: Vec = + result.new_txids.iter().chain(result.existing_txids.iter()).cloned().collect(); + queue.push_back(SyncEvent::BlockProcessed { + block_hash: *key.hash(), + height: key.height(), + wallets: wallets.clone(), + new_scripts: result.new_scripts, + confirmed_txids: confirmed, + }); + } + + if !had_activity && in_flight.is_empty() && queue.is_empty() { + idle_rounds += 1; + if idle_rounds > 10 { + break; + } + } else { + idle_rounds = 0; + } + } + + let w = wallet.read().await; + let info = w.get_wallet_info(&wallet_id).unwrap(); + let tx_count = info.first_bip44_managed_account().unwrap().transactions().len(); + (tx_count, N_ADDR as usize) + } + + /// Control case: dust mined in derivation order. The gap-limit chase can + /// always keep up block by block, so discovery completes even without + /// the commit-time verification rescan. + #[tokio::test(flavor = "multi_thread")] + async fn dust_restore_discovers_all_txs_when_mined_in_index_order() { + let (found, expected) = run_dust_restore(false).await; + assert_eq!(found, expected, "in-order dust restore must discover every transaction"); + } + + /// Regression test for the sequential-address restore stall + /// (gap-limit chase ending on a backfill-only wave). + /// + /// Payments to sequentially derived addresses are mined out of + /// derivation order, the way mempool waves actually land in blocks. + /// Discovery then advances in waves: each processed wave extends the + /// pools, and the extended window must be re-matched against the + /// batch's filters. Before the commit-time verification rescan, a wave + /// whose visible transactions all paid already-derived indices ended + /// the chase ("no new scripts collected") and committed the batch while + /// it still held blocks paying indices past the window — permanently + /// unrecoverable, because committed batches are never rescanned. This + /// scenario lost ~40% of the transactions. + /// + /// Observed in production on a mainnet wallet with 345k used + /// addresses: discovery froze at index 2,400 and silently missed + /// 880k+ transactions. + #[tokio::test(flavor = "multi_thread")] + async fn dust_restore_discovers_all_txs_when_mined_out_of_order() { + let (found, expected) = run_dust_restore(true).await; + assert_eq!( + found, + expected, + "out-of-order dust restore lost {} of {} transactions — \ + a filter batch committed while it still held matchable blocks", + expected - found, + expected + ); + } } diff --git a/dash-spv/src/sync/filters/sync_manager.rs b/dash-spv/src/sync/filters/sync_manager.rs index 6cc9d7a9e..bac2807fc 100644 --- a/dash-spv/src/sync/filters/sync_manager.rs +++ b/dash-spv/src/sync/filters/sync_manager.rs @@ -167,6 +167,15 @@ impl< // `tracker.track` residual. self.tracker.record_processed(*height, *block_hash, wallets); + // Any derivation re-arms the commit-time verification rescan + // of every active batch — including batches this block does + // not belong to, and even when the block's own batch is + // already gone (in which case the scripts would otherwise be + // dropped without ever being matched). + if new_scripts.values().any(|scripts| !scripts.is_empty()) { + self.script_generation += 1; + } + // Check if this block is part of our tracked blocks if let Some((_, batch_start)) = self.tracker.finish_in_flight(block_hash) { if let Some(batch) = self.active_batches.get_mut(&batch_start) { From 7b4d22e996929391add2f5852a54941e154e0379 Mon Sep 17 00:00:00 2001 From: pasta Date: Wed, 12 Aug 2026 21:30:35 -0500 Subject: [PATCH 2/2] fix(dash-spv): hold batch commits while any active batch has blocks in flight A block in flight for a later batch can still derive new scripts when it lands, and those scripts can match an earlier batch's filters. Sealing the earlier batch during that window raced the delivery - the same knowledge-behind-the-watermark loss the verification rescan prevents, through a narrower window. Commits now wait for global block quiescence, so a seal implies no undelivered derivations exist anywhere and the verification fixpoint argument has no gaps. Co-Authored-By: Claude Fable 5 --- dash-spv/src/sync/filters/manager.rs | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/dash-spv/src/sync/filters/manager.rs b/dash-spv/src/sync/filters/manager.rs index c89d15575..f3bd82439 100644 --- a/dash-spv/src/sync/filters/manager.rs +++ b/dash-spv/src/sync/filters/manager.rs @@ -560,6 +560,19 @@ impl 0) { + break; + } + // Check if rescan is needed and not done if !batch.rescan_complete() { // Take per-wallet collected scripts from the batch