feat(dash-spv): adaptive gap-limit probe escalation for BIP44 discovery

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

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<WalletId, HashSet<ScriptBuf>>,
+    /// 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<WalletId>,
+    /// 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<WalletId, HashSet<ScriptBuf>> {
         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<Item = &'a WalletId>,
+    ) {
+        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<WalletId> {
+        &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<WalletId, u64>) {

dash-spv/src/sync/filters/sync_manager.rs

@@ -160,13 +160,57 @@ impl<
                 height,
                 wallets,
                 new_scripts,
+                confirmed_txids,
                 ..
             } => {
                 // Record per-wallet processing so a future scan can give a
                 // late-added wallet its own pass at this block via the
                 // `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::<usize>()
+                    );
+                }
+
+                // 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) {

key-wallet-manager/src/process_block.rs

@@ -234,6 +234,16 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> 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<ScriptBuf> {
+        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<Vec<u8>> {
         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<ManagedWalletInfo> = WalletManager::new(Network::Testnet);

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<ScriptBuf> {
+        Vec::new()
+    }
+
     /// Get the bare `hash160` compact-filter elements monitored by `wallet_id`
     /// that are not covered by its scriptPubKeys.
     ///

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<Vec<AddressInfo>> {
+        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()]);