diff --git a/Cargo.toml b/Cargo.toml index 63049fbea..73d4ff8b9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,3 +21,12 @@ debug = 1 [profile.dev] # Default to unwinding for most crates + +[profile.dev.package.dashcore_hashes] +# Debug-build siphash is ~50x slower than optimized, and BIP158 compact-filter +# matching hashes every query element against every filter — the dust-restore +# and gap-probe tests in dash-spv spend most of their wall time there. +# dashcore_hashes is a rarely-edited leaf crate, so optimizing it in dev +# costs one slightly longer cold build and speeds every filter/hash-heavy +# test severalfold. +opt-level = 2 diff --git a/dash-spv/src/sync/filters/batch.rs b/dash-spv/src/sync/filters/batch.rs index b389c5a1b..bddd9a3f2 100644 --- a/dash-spv/src/sync/filters/batch.rs +++ b/dash-spv/src/sync/filters/batch.rs @@ -1,7 +1,7 @@ use dashcore::bip158::BlockFilter; use dashcore::ScriptBuf; use key_wallet_manager::{FilterMatchKey, WalletId}; -use std::collections::{BTreeMap, HashMap, HashSet}; +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; /// A completed batch of compact block filters ready for verification. /// @@ -37,6 +37,23 @@ 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>, + /// Wallets for which at least one of this batch's filters matched during + /// any scan or rescan — the "activity batch" marker for adaptive + /// gap-probe escalation. Commit runs the probe ladder only for wallets in + /// this set, so batches a wallet had no activity in never pay for a + /// probe. Filter false positives can land here; that only costs a probe + /// pass, never correctness. + matched_wallets: BTreeSet, + /// 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 +73,8 @@ impl FiltersBatch { rescan_complete: false, scanned_wallets: BTreeMap::new(), collected_scripts: HashMap::new(), + matched_wallets: BTreeSet::new(), + full_match_generation: 0, } } /// Start height of this batch (inclusive). @@ -107,6 +126,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; @@ -123,6 +150,19 @@ impl FiltersBatch { pub(super) fn take_collected_scripts(&mut self) -> HashMap> { std::mem::take(&mut self.collected_scripts) } + /// Record wallets whose queries matched at least one of this batch's + /// filters during a scan or rescan (the "activity batch" marker). + pub(super) fn note_matched_wallets<'a>( + &mut self, + wallets: impl IntoIterator, + ) { + self.matched_wallets.extend(wallets.into_iter().copied()); + } + /// Wallets with at least one filter match in this batch across all scan + /// and rescan passes. + pub(super) fn matched_wallets(&self) -> &BTreeSet { + &self.matched_wallets + } /// 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/manager.rs b/dash-spv/src/sync/filters/manager.rs index 521af307c..7aa5151f9 100644 --- a/dash-spv/src/sync/filters/manager.rs +++ b/dash-spv/src/sync/filters/manager.rs @@ -42,6 +42,31 @@ struct WalletScanState { elements: Vec>, } +/// Escalating gap-probe widths for adaptive BIP44 discovery. +/// +/// Steady-state discovery derives only `gap_limit` (typically 30) addresses +/// past the highest used index, so a run of unused indices longer than the +/// gap limit stalls discovery silently — real mainnet wallets have hundreds +/// of such runs (the worst observed is exactly 100). When a batch with wallet +/// activity reaches its commit fixpoint, the manager probes each rung in +/// order against the chain's own filters: derive the widened window, re-match +/// this batch and every later scanned batch, and resume the normal chase on +/// any hit. Only when the maximum rung finds nothing is the stall considered +/// genuine wallet-end-of-history and the batch allowed to commit. +/// +/// The ladder is finite and per-wallet progress through it is monotone +/// between resets (a reset requires new usage or a new derivation — real +/// progress), so probing can never loop unboundedly. The last rung equals +/// [`key_wallet::gap_limit::MAX_GAP_LIMIT`]: pools cannot derive wider, so it +/// is also the hard cap on how large an unused-index run discovery can cross. +const PROBE_GAP_LADDER: [u32; 3] = [100, 300, 1000]; + +/// The top rung must be exactly the widest window a pool can derive. +const _: () = assert!( + PROBE_GAP_LADDER[PROBE_GAP_LADDER.len() - 1] == key_wallet::gap_limit::MAX_GAP_LIMIT, + "probe ladder must end at MAX_GAP_LIMIT" +); + /// Maximum number of batches to scan ahead while waiting for blocks. /// /// Raising this does not buy more throughput while a batch is stalled on a block. @@ -91,6 +116,29 @@ 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, + /// Per-wallet progress through [`PROBE_GAP_LADDER`]: the index of the + /// next rung to try when a commit-time probe runs for the wallet. An + /// index past the end means the full ladder found nothing — probing is + /// terminal for that wallet until real progress resets the entry + /// (a relevant transaction confirmed or new scripts derived for the + /// wallet, both signalled through `BlockProcessed`), because a moving + /// usage frontier makes the next stall a new stall. Keeping the level + /// across passes lets escalation resume where it left off instead of + /// restarting at the lowest rung every time. + /// + /// In-memory only, like the rest of the manager's scan state: a restart + /// re-probes from the lowest rung, which at worst repeats a few + /// derive-nothing rungs (each a cheap no-op against the already-derived + /// pool tail) before reaching the recorded depth again. + pub(super) probe_levels: HashMap, } impl @@ -135,6 +183,8 @@ impl 0) { + break; + } + // Check if rescan is needed and not done if !batch.rescan_complete() { - // Take per-wallet collected scripts from the batch - let scripts_by_wallet = self - .active_batches - .get_mut(&batch_start) - .map(|b| b.take_collected_scripts()) - .unwrap_or_default(); + // Take per-wallet collected scripts from EVERY scanned active + // batch, not just the committing one. A later batch's blocks + // collect their derivations into that batch, but its own + // collected-scripts rescan used to run only once it became + // the lowest batch — so while an earlier batch was being + // considered for commit, a chase running inside a later + // batch stalled with fresh scripts sitting unrescanned. The + // quiescence gate above then saw a lull that wasn't a wallet + // fixpoint at all. Draining every batch's collected scripts + // here keeps the whole chase moving no matter which batch + // owns the deriving blocks; each script set is rescanned + // against every scanned batch, a superset of the old + // "donor batch and everything after it" coverage. + let mut scanned_starts: Vec = Vec::new(); + let mut scripts_by_wallet: HashMap> = HashMap::new(); + for (&start, donor) in self.active_batches.iter_mut() { + if !donor.scanned() { + continue; + } + scanned_starts.push(start); + for (wallet_id, scripts) in donor.take_collected_scripts() { + scripts_by_wallet.entry(wallet_id).or_default().extend(scripts); + } + } if !scripts_by_wallet.is_empty() { - // Rescan current batch - events.extend(self.rescan_batch(batch_start, &scripts_by_wallet).await?); - - // Also rescan later batches that are already scanned - let later_batches: Vec = self - .active_batches - .iter() - .filter(|(&start, batch)| start > batch_start && batch.scanned()) - .map(|(&start, _)| start) - .collect(); + for start in &scanned_starts { + events.extend(self.rescan_batch(*start, &scripts_by_wallet).await?); + } - for later_start in later_batches { - events.extend(self.rescan_batch(later_start, &scripts_by_wallet).await?); + // If the rescan found blocks for ANY batch the lull is + // over: the chase resumes and no batch may seal until the + // quiescence gate above is satisfied again. + if self.active_batches.values().any(|batch| batch.pending_blocks() > 0) { + break; } + } - // Check if rescan found more blocks - if let Some(batch) = self.active_batches.get(&batch_start) { + // Verification rescan: an empty `collected_scripts` only means + // no scripts were derived from THIS batch's blocks since the + // last wave — scripts derived from blocks owned by other + // batches (or whose batch is already gone) were never matched + // against this batch's filters. Committing on that empty-wave + // signal alone is what permanently lost transactions during + // sequential-address restores: one backfill-only wave ended + // the gap-limit chase while the batch still held matchable + // blocks, and committed batches are never rescanned. Re-match + // with the wallets' full current script sets whenever any + // derivation happened since the batch's last full-set match, + // and only mark the rescan complete once that turns up + // nothing. + let last_full_match = + self.active_batches.get(&batch_start).map(|b| b.full_match_generation()); + let needs_verification = + last_full_match.is_some_and(|generation| generation != self.script_generation); + if needs_verification { + let generation_now = self.script_generation; + tracing::debug!( + "Verification rescan for batch {}: full-set match generation {} -> {}", + batch_start, + last_full_match.unwrap_or(0), + generation_now + ); + let wallet_ids: Vec = 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 { - // Found more blocks, can't commit yet break; } } } + + // Adaptive gap-probe escalation. The gates above prove this + // batch is at its commit fixpoint: verification against the + // full current script sets found nothing and no blocks are in + // flight anywhere. But "full current script sets" only reach + // `gap_limit` past each pool's usage frontier, so a run of + // unused indices longer than the gap limit stalls discovery + // right here with everything looking clean. Before sealing + // the batch, probe successively wider windows against its + // filters; any hit resumes the chase, and only a clean pass + // at the maximum width lets the batch commit. + let (probe_events, probe_found_blocks) = self.run_probe_ladder(batch_start).await?; + events.extend(probe_events); + if probe_found_blocks { + // The chase resumes: the found blocks must be processed + // (possibly deriving further scripts) before this batch + // may seal. + break; + } + // Mark rescan as complete if let Some(batch) = self.active_batches.get_mut(&batch_start) { batch.mark_rescan_complete(); @@ -656,6 +801,114 @@ impl SyncResult<(Vec, bool)> { + let candidates: Vec = self + .active_batches + .get(&batch_start) + .map(|batch| { + batch + .matched_wallets() + .iter() + .filter(|id| batch.scanned_wallets().contains_key(*id)) + .copied() + .collect() + }) + .unwrap_or_default(); + if candidates.is_empty() { + return Ok((Vec::new(), false)); + } + + // Every probe rescan covers this batch plus each later batch that has + // already been scanned. + let mut rescan_targets = vec![batch_start]; + rescan_targets.extend( + self.active_batches + .iter() + .filter(|(&start, batch)| start > batch_start && batch.scanned()) + .map(|(&start, _)| start), + ); + + let mut events = Vec::new(); + for wallet_id in candidates { + let mut level = self.probe_levels.get(&wallet_id).copied().unwrap_or(0); + let mut found_blocks = false; + while level < PROBE_GAP_LADDER.len() { + let probe_gap = PROBE_GAP_LADDER[level]; + let new_scripts = self.wallet.write().await.probe_extend_gap(&wallet_id, probe_gap); + if new_scripts.is_empty() { + // Already derived at least this deep past the current + // frontier (an earlier pass probed this rung and nothing + // moved since) — escalate. + level += 1; + continue; + } + + let scripts_by_wallet: HashMap> = + HashMap::from([(wallet_id, new_scripts.iter().cloned().collect())]); + let mut probe_events = Vec::new(); + for start in &rescan_targets { + probe_events.extend(self.rescan_batch(*start, &scripts_by_wallet).await?); + } + let blocks_found: usize = probe_events + .iter() + .filter_map(|event| match event { + SyncEvent::BlocksNeeded { + blocks, + } => Some(blocks.len()), + _ => None, + }) + .sum(); + tracing::info!( + "Gap probe: wallet {} batch {} probe_gap {} derived {} scripts, found {} blocks", + hex::encode(&wallet_id[..4]), + batch_start, + probe_gap, + new_scripts.len(), + blocks_found, + ); + events.extend(probe_events); + + if blocks_found > 0 { + // Resume at this rung: the same width re-derives from the + // new frontier once the found blocks advance it. + found_blocks = true; + break; + } + level += 1; + } + // Past the last rung means the ladder is exhausted with no finds: + // terminal for this wallet until a confirmed transaction or new + // derivation resets its level. + self.probe_levels.insert(wallet_id, level); + if found_blocks { + return Ok((events, true)); + } + } + Ok((events, false)) + } + /// Scan any active batches where filters are available but not yet scanned. async fn scan_ready_batches(&mut self) -> SyncResult> { let mut events = Vec::new(); @@ -794,6 +1047,11 @@ impl, + /// External pool's total generated address count at the fixpoint. + total_generated: u32, + } + + /// Drive a full filter-sync + gap-limit discovery loop over a synthetic + /// dust-spam restore with the real `WalletManager`. + /// + /// The chain pays `N_ADDR` external addresses 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. + /// + /// `holes` carves runs of permanently unused indices into the usage + /// space: each `(threshold, gap)` entry shifts every used index at or + /// above `threshold` up by `gap`, so the used-index sequence jumps + /// straight from `threshold - 1` to `threshold + gap` — a run of `gap` + /// consecutive unused indices, the shape that stalls plain gap-limit + /// discovery whenever `gap` exceeds the gap limit. Real restored + /// wallets are full of these (the production wallet that motivated the + /// probe ladder has 273 runs of 30+, the longest exactly 100). + /// + /// Returns a [`DustRestoreOutcome`] once the sync reaches a fixpoint. + async fn run_dust_restore(shuffle_indices: bool, holes: &[(u32, u32)]) -> DustRestoreOutcome { + 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; + + // Used-index mapping: slot `k` (the k-th payment in usage order) + // lands on derivation index `k` shifted up by every hole at or below + // it. With no holes this is the identity. + let index_of = |slot: u32| -> u32 { + slot + holes + .iter() + .filter(|(threshold, _)| slot >= *threshold) + .map(|(_, g)| g) + .sum::() + }; + let max_used_index = index_of(N_ADDR - 1); + + // 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, max_used_index + 1, &key_source).expect("derive addresses") + }; + assert_eq!(addresses.len(), (max_used_index + 1) 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[index_of(assignment[pos] as u32) as usize], + (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 account = info.first_bip44_managed_account().unwrap(); + let tx_count = account.transactions().len(); + let external_stats = account + .managed_account_type() + .address_pools() + .into_iter() + .find(|p| p.is_external()) + .expect("external pool") + .stats(); + DustRestoreOutcome { + found: tx_count, + expected: N_ADDR as usize, + max_used_index, + highest_used: external_stats.highest_used, + total_generated: external_stats.total_generated, + } + } + + /// 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 outcome = run_dust_restore(false, &[]).await; + assert_eq!( + outcome.found, outcome.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 outcome = run_dust_restore(true, &[]).await; + assert_eq!( + outcome.found, + outcome.expected, + "out-of-order dust restore lost {} of {} transactions — \ + a filter batch committed while it still held matchable blocks", + outcome.expected - outcome.found, + outcome.expected + ); + } + + /// Shared assertions for the gap-hole scenarios: full discovery, the + /// frontier resting exactly on the true last used index, and bounded pool + /// growth (the probed tail may reach `MAX_GAP_LIMIT` past the frontier, + /// plus slack for a steady-state window derived on top — pools only + /// grow, never past that). + fn assert_hole_outcome(outcome: &DustRestoreOutcome, scenario: &str) { + assert_eq!( + outcome.found, + outcome.expected, + "{}: lost {} of {} transactions — discovery stalled on an unused-index run", + scenario, + outcome.expected - outcome.found, + outcome.expected + ); + assert_eq!( + outcome.highest_used, + Some(outcome.max_used_index), + "{}: final usage frontier must equal the true max used index", + scenario + ); + let growth_cap = outcome.max_used_index + 1 + key_wallet::gap_limit::MAX_GAP_LIMIT + 64; + assert!( + outcome.total_generated <= growth_cap, + "{}: runaway pool growth — generated {} addresses, cap {}", + scenario, + outcome.total_generated, + growth_cap + ); + } + + /// A run of 54 consecutive unused indices — the first stall point of the + /// production wallet that motivated the probe ladder. Plain gap-30 + /// discovery freezes at its lower edge; the first probe rung (100) must + /// carry the chase across it so every transaction is discovered. + #[tokio::test(flavor = "multi_thread")] + async fn dust_restore_crosses_gap_hole_of_54() { + let outcome = run_dust_restore(true, &[(1000, 54)]).await; + assert_hole_outcome(&outcome, "hole-54 restore"); + } + + /// A run of 150 consecutive unused indices: past the first probe rung + /// (100), so discovery must escalate to the second (300) to cross it. + #[tokio::test(flavor = "multi_thread")] + async fn dust_restore_crosses_gap_hole_of_150_via_escalation() { + let outcome = run_dust_restore(false, &[(1000, 150)]).await; + assert_hole_outcome(&outcome, "hole-150 restore"); + } + + /// Terminal termination: after the last real payment the probe ladder + /// must exhaust (nothing past the frontier through the maximum probe), + /// the sync must reach its fixpoint rather than hang, the frontier must + /// rest exactly on the true last used index, and the pool must not have + /// grown unboundedly. Two holes force the ladder to fire mid-chase more + /// than once before the terminal pass. + #[tokio::test(flavor = "multi_thread")] + async fn dust_restore_terminates_after_last_payment() { + let outcome = run_dust_restore(true, &[(800, 54), (2000, 100)]).await; + assert_hole_outcome(&outcome, "terminal-termination restore"); + } } diff --git a/dash-spv/src/sync/filters/sync_manager.rs b/dash-spv/src/sync/filters/sync_manager.rs index 6cc9d7a9e..2de7fd2a9 100644 --- a/dash-spv/src/sync/filters/sync_manager.rs +++ b/dash-spv/src/sync/filters/sync_manager.rs @@ -160,6 +160,7 @@ impl< height, wallets, new_scripts, + confirmed_txids, .. } => { // Record per-wallet processing so a future scan can give a @@ -167,6 +168,49 @@ 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; + tracing::debug!( + "Script generation bumped to {} at height {} (+{} scripts)", + self.script_generation, + height, + new_scripts.values().map(|s| s.len()).sum::() + ); + } + + // Progress re-arms the gap-probe ladder: a wallet whose usage + // frontier may have moved gets its next stall treated as a + // new stall, restarting escalation from the lowest rung. + // Derivations alone are NOT a faithful movement signal here — + // usage discovered deep inside an already-derived probe tail + // advances the frontier without deriving anything (the + // steady-state window is already covered by the tail), so a + // confirmed relevant transaction must also reset the level or + // discovery would stay probe-terminal while the frontier + // walks through the tail. + // + // The confirmed-tx reset is deliberately coarse: the event + // does not attribute `confirmed_txids` per wallet, so every + // wallet the block was processed for is reset. Over-resetting + // only re-arms probing that would otherwise have stayed + // terminal — safe, at worst a few extra probe rungs — while + // under-resetting could permanently stall discovery. + for (wallet_id, scripts) in new_scripts { + if !scripts.is_empty() { + self.probe_levels.remove(wallet_id); + } + } + if !confirmed_txids.is_empty() { + for wallet_id in wallets { + self.probe_levels.remove(wallet_id); + } + } + // 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) { diff --git a/key-wallet-manager/src/process_block.rs b/key-wallet-manager/src/process_block.rs index 28e0f986f..3f719d746 100644 --- a/key-wallet-manager/src/process_block.rs +++ b/key-wallet-manager/src/process_block.rs @@ -234,6 +234,16 @@ impl WalletInterface for WalletM self.wallet_infos.get(wallet_id).map(|info| info.scan_script_pubkeys()).unwrap_or_default() } + fn probe_extend_gap(&mut self, wallet_id: &WalletId, probe_gap: u32) -> Vec { + let Some((wallet, info)) = self.get_wallet_and_info_mut(wallet_id) else { + return Vec::new(); + }; + let derived = info.accounts_mut().probe_extend_gap_with(probe_gap, |to_check, index| { + wallet.key_source_for_account_type(to_check, index) + }); + derived.into_iter().map(|address_info| address_info.script_pubkey).collect() + } + fn monitored_filter_elements_for(&self, wallet_id: &WalletId) -> Vec> { self.wallet_infos .get(wallet_id) @@ -779,6 +789,49 @@ mod tests { assert!(manager.scan_script_pubkeys_for(&[0xff; 32]).is_empty()); } + #[tokio::test] + async fn test_probe_extend_gap_derives_tail_and_keeps_steady_policy() { + let (mut manager, wallet_id, _addr) = setup_manager_with_wallet(); + + let monitored_before = manager.monitored_script_pubkeys_for(&wallet_id).len(); + let revision_before = manager.monitor_revision(); + + // First probe widens every funds pool to 100 past its frontier. + let derived = manager.probe_extend_gap(&wallet_id, 100); + assert!(!derived.is_empty(), "fresh wallet pools must derive up to the probe width"); + + // The derived tail joins the monitored set and bumps the revision. + let monitored_after = manager.monitored_script_pubkeys_for(&wallet_id); + assert_eq!(monitored_after.len(), monitored_before + derived.len()); + for script in &derived { + assert!(monitored_after.contains(script), "derived script must be monitored"); + } + assert!(manager.monitor_revision() > revision_before); + + // Re-probing the same width with an unmoved frontier derives nothing; + // escalation derives only the delta. + assert!(manager.probe_extend_gap(&wallet_id, 100).is_empty()); + let escalated = manager.probe_extend_gap(&wallet_id, 300); + assert!(!escalated.is_empty()); + assert!(manager.probe_extend_gap(&wallet_id, 300).is_empty()); + + // The steady-state policy is restored: ordinary gap maintenance after + // marking an address used must not derive out to the probe width + // again (the tail is already there). + let scripts_before_use = manager.monitored_script_pubkeys_for(&wallet_id).len(); + let addr = manager.monitored_addresses()[0].clone(); + let tx = create_tx_paying_to(&addr, 0xee); + manager.process_mempool_transaction(&tx, None).await; + assert_eq!( + manager.monitored_script_pubkeys_for(&wallet_id).len(), + scripts_before_use, + "steady-state maintenance must find the probed tail already derived" + ); + + // Unknown wallet id probes nothing. + assert!(manager.probe_extend_gap(&[0xff; 32], 100).is_empty()); + } + #[tokio::test] async fn test_monitor_revision_bumps_and_stability() { let mut manager: WalletManager = WalletManager::new(Network::Testnet); diff --git a/key-wallet-manager/src/wallet_interface.rs b/key-wallet-manager/src/wallet_interface.rs index 484fb2c84..6a433dcd0 100644 --- a/key-wallet-manager/src/wallet_interface.rs +++ b/key-wallet-manager/src/wallet_interface.rs @@ -103,6 +103,28 @@ pub trait WalletInterface: Send + Sync + 'static { self.monitored_script_pubkeys_for(wallet_id) } + /// Probe-widen every derivable address pool of `wallet_id` so each is + /// generated at least `probe_gap` indices past its usage frontier, and + /// return the scriptPubKeys of the addresses that derivation freshly + /// created (empty when everything was already derived that deep). + /// + /// This is the wallet half of adaptive gap-probe escalation: BIP44 + /// discovery derives only `gap_limit` addresses past the highest used + /// index, so a run of unused indices longer than the gap limit stalls + /// discovery silently. When filter sync suspects such a stall it calls + /// this with an escalating `probe_gap` and re-matches the returned + /// scripts against the chain's compact filters; any hit resumes the + /// normal chase past the hole. + /// + /// Implementations must restore the steady-state gap limit before + /// returning — the probe is a one-shot widened derivation, not a policy + /// change — but must NOT un-derive: pools only grow, and the probed tail + /// stays monitored. The default is a no-op returning empty, for + /// implementations without derivable pools. + fn probe_extend_gap(&mut self, _wallet_id: &WalletId, _probe_gap: u32) -> Vec { + Vec::new() + } + /// Get the bare `hash160` compact-filter elements monitored by `wallet_id` /// that are not covered by its scriptPubKeys. /// diff --git a/key-wallet/src/managed_account/address_pool.rs b/key-wallet/src/managed_account/address_pool.rs index 57c021bed..55bcf668e 100644 --- a/key-wallet/src/managed_account/address_pool.rs +++ b/key-wallet/src/managed_account/address_pool.rs @@ -1066,6 +1066,41 @@ impl AddressPool { self.maintain_gap_limit(key_source) } + /// Derive the window a temporarily widened gap limit of `probe_gap` would + /// require, then restore the steady-state gap limit. + /// + /// This is the primitive behind adaptive gap-probe escalation: BIP44 + /// discovery with the steady-state gap limit stalls on any run of unused + /// indices longer than that limit, so a caller that suspects a stall can + /// probe a deeper window against external evidence (e.g. compact block + /// filters) without changing the pool's ongoing derivation policy. + /// + /// The restore deliberately does NOT un-derive: pools only grow, so the + /// probed tail stays derived (and monitored) even though future + /// [`Self::maintain_gap_limit`] calls anchor at `highest_used + + /// steady_gap` again. Returns the freshly derived [`AddressInfo`] entries; + /// empty when the pool is already generated at least `probe_gap` past its + /// usage frontier (or `probe_gap` does not widen the steady-state limit). + /// + /// `probe_gap` is capped at [`crate::gap_limit::MAX_GAP_LIMIT`] like any + /// other gap limit. + pub fn probe_gap_limit( + &mut self, + probe_gap: u32, + key_source: &KeySource, + ) -> Result> { + let steady = self.gap_limit; + if probe_gap <= steady { + return Ok(Vec::new()); + } + let result = self.set_gap_limit(probe_gap, key_source); + // Restore the steady-state policy without un-deriving. Even if the + // widened derivation failed partway, whatever was derived stays and + // the ongoing policy must not remain at the probe width. + self.gap_limit = steady; + result + } + /// Generate addresses to maintain the gap limit. /// /// Returns the freshly generated [`AddressInfo`] entries (in derivation @@ -1440,6 +1475,61 @@ mod tests { assert_eq!(pool.addresses.len(), crate::gap_limit::MAX_GAP_LIMIT as usize); } + #[test] + fn test_probe_gap_limit_widens_once_and_restores_policy() { + let base_path = DerivationPath::from(vec![ChildNumber::from_normal_idx(0).unwrap()]); + let key_source = test_key_source(); + + let mut pool = AddressPool::new( + base_path, + AddressPoolType::External, + 5, + Network::Testnet, + &key_source, + ) + .unwrap(); + assert_eq!(pool.addresses.len(), 5); + + // A probe not wider than the steady-state gap derives nothing. + assert!(pool.probe_gap_limit(5, &key_source).unwrap().is_empty()); + assert!(pool.probe_gap_limit(3, &key_source).unwrap().is_empty()); + assert_eq!(pool.addresses.len(), 5); + + // Probing to 20 derives the widened window but leaves the steady + // policy at 5, and the derived tail stays. + let derived = pool.probe_gap_limit(20, &key_source).unwrap(); + assert_eq!(derived.len(), 15); + assert_eq!(pool.gap_limit, 5); + assert_eq!(pool.addresses.len(), 20); + assert_eq!(pool.highest_generated, Some(19)); + + // Re-probing the same width with an unmoved frontier is a no-op. + assert!(pool.probe_gap_limit(20, &key_source).unwrap().is_empty()); + + // Usage moves the frontier; the same probe width re-derives relative + // to it while still restoring the steady gap limit. + assert!(pool.mark_index_used(7)); + let derived = pool.probe_gap_limit(20, &key_source).unwrap(); + assert_eq!(derived.len(), 8); // indices 20..=27 (7 + 20) + assert_eq!(pool.gap_limit, 5); + assert_eq!(pool.highest_generated, Some(27)); + + // Ordinary maintenance keeps anchoring at the steady gap and finds + // the probed tail already derived. + assert!(pool.maintain_gap_limit(&key_source).unwrap().is_empty()); + + // The probe width is capped at MAX_GAP_LIMIT. + let derived = + pool.probe_gap_limit(crate::gap_limit::MAX_GAP_LIMIT + 500, &key_source).unwrap(); + assert_eq!( + pool.highest_generated, + Some(7 + crate::gap_limit::MAX_GAP_LIMIT), + "probe must clamp at MAX_GAP_LIMIT past the frontier" + ); + assert_eq!(derived.len(), (crate::gap_limit::MAX_GAP_LIMIT - 20) as usize); + assert_eq!(pool.gap_limit, 5); + } + #[test] fn test_reserve_returns_distinct_addresses() { let base_path = DerivationPath::from(vec![ChildNumber::from_normal_idx(0).unwrap()]); diff --git a/key-wallet/src/managed_account/managed_account_collection.rs b/key-wallet/src/managed_account/managed_account_collection.rs index cf60f086b..8e9c5613b 100644 --- a/key-wallet/src/managed_account/managed_account_collection.rs +++ b/key-wallet/src/managed_account/managed_account_collection.rs @@ -7,7 +7,7 @@ use std::collections::BTreeMap; use crate::account::account_collection::{DashpayAccountKey, PlatformPaymentAccountKey}; use crate::gap_limit::DIP17_GAP_LIMIT; -use crate::managed_account::address_pool::AddressPoolType; +use crate::managed_account::address_pool::{AddressInfo, AddressPoolType}; use crate::managed_account::managed_account_ref::{ ManagedAccountRef, ManagedAccountRefMut, OwnedManagedCoreAccount, }; @@ -16,6 +16,7 @@ use crate::managed_account::managed_account_type::ManagedAccountType; use crate::managed_account::managed_platform_account::ManagedPlatformAccount; use crate::managed_account::{ManagedCoreFundsAccount, ManagedCoreKeysAccount}; use crate::transaction_checking::account_checker::CoreAccountTypeMatch; +use crate::transaction_checking::transaction_router::AccountTypeToCheck; use crate::KeySource; use crate::{Account, AccountCollection, AccountType}; #[cfg(feature = "serde")] @@ -980,6 +981,74 @@ impl ManagedAccountCollection { accounts } + /// Probe-widen the gap window of every funds-bearing account's address + /// pools to `probe_gap`, returning the freshly derived [`AddressInfo`] + /// entries across all visited accounts and pools. + /// + /// This is the wallet-level seam for adaptive gap-probe escalation (see + /// [`AddressPool::probe_gap_limit`]): each pool derives the window a gap + /// limit of `probe_gap` would require and then restores its steady-state + /// gap limit, so the ongoing derivation policy is unchanged while the + /// probed tail stays derived and monitored (pools only grow). + /// + /// Visits [`Self::all_funding_accounts_mut`] — Standard BIP44/32, + /// CoinJoin, and DashPay receival accounts. Sequential funds discovery is + /// where long unused-index runs occur in practice (dust storms, batched + /// payouts) and where a stall silently severs the rest of the wallet's + /// history; keys-only pools (identity, asset-lock, provider) have tiny, + /// wallet-driven usage, and probing every one of them to the maximum gap + /// would multiply the probe's derivation and filter-matching cost for no + /// observed failure mode. Broaden the visited set here if that ever + /// changes. + /// + /// `key_source_for` maps an account's [`AccountTypeToCheck`] and index to + /// the [`KeySource`] able to derive for it — the same lookup the wallet + /// checker uses for gap-limit maintenance (typically + /// `Wallet::key_source_for_account_type`). Accounts whose lookup yields + /// [`KeySource::NoKeySource`] are skipped. Accounts that derived anything + /// get their monitor revision bumped. + /// + /// [`AddressPool::probe_gap_limit`]: crate::managed_account::address_pool::AddressPool::probe_gap_limit + pub fn probe_extend_gap_with( + &mut self, + probe_gap: u32, + key_source_for: F, + ) -> Vec + where + F: Fn(&AccountTypeToCheck, Option) -> KeySource, + { + let mut derived = Vec::new(); + for account in self.all_funding_accounts_mut() { + let account_type = account.managed_account_type().to_account_type(); + let Ok(to_check) = AccountTypeToCheck::try_from(account_type) else { + continue; + }; + let key_source = key_source_for(&to_check, account_type.index()); + if matches!(key_source, KeySource::NoKeySource) { + continue; + } + let derived_before = derived.len(); + for pool in account.managed_account_type_mut().address_pools_mut() { + let pool_type = pool.pool_type; + match pool.probe_gap_limit(probe_gap, &key_source) { + Ok(infos) => derived.extend(infos), + Err(e) => { + tracing::error!( + account_type = ?to_check, + pool_type = ?pool_type, + error = %e, + "Failed to probe-widen gap limit for address pool" + ); + } + } + } + if derived.len() > derived_before { + account.bump_monitor_revision(); + } + } + derived + } + /// Get the accounts whose funds belong to **this** wallet (Standard /// BIP44/32, CoinJoin, DashPay receival). ///