fix(dash-spv): coalesce committed-range filter rescans across batch commits

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

@@ -37,12 +37,6 @@ 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>>,
-    /// Scripts already forward-rescanned but still awaiting the one combined
-    /// backward sweep over the committed range (#846). Accumulated across
-    /// fixpoint rounds so each script crosses the stored history exactly
-    /// once, instead of the whole history being reloaded per derivation
-    /// round.
-    backward_scripts: HashMap<WalletId, HashSet<ScriptBuf>>,
 }
 
 impl FiltersBatch {
@@ -62,7 +56,6 @@ impl FiltersBatch {
             rescan_complete: false,
             scanned_wallets: BTreeMap::new(),
             collected_scripts: HashMap::new(),
-            backward_scripts: HashMap::new(),
         }
     }
     /// Start height of this batch (inclusive).
@@ -130,21 +123,6 @@ impl FiltersBatch {
     pub(super) fn take_collected_scripts(&mut self) -> HashMap<WalletId, HashSet<ScriptBuf>> {
         std::mem::take(&mut self.collected_scripts)
     }
-    /// Queue already forward-rescanned scripts for the deferred backward
-    /// sweep over the committed range.
-    pub(super) fn accumulate_backward_scripts(
-        &mut self,
-        scripts: HashMap<WalletId, HashSet<ScriptBuf>>,
-    ) {
-        for (wallet_id, scripts) in scripts {
-            self.backward_scripts.entry(wallet_id).or_default().extend(scripts);
-        }
-    }
-    /// Take the scripts accumulated for the backward sweep, leaving the map
-    /// empty.
-    pub(super) fn take_backward_scripts(&mut self) -> HashMap<WalletId, HashSet<ScriptBuf>> {
-        std::mem::take(&mut self.backward_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<WalletId, u64>) {

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

@@ -25,6 +25,11 @@
 //!    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)
@@ -411,3 +416,171 @@ async fn coinjoin_gap_limit_stall_across_committed_batch() {
          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<BlockHash, Block> = 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<WalletId>> = 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
+    );
+}

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

@@ -91,6 +91,21 @@ 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<WalletId, HashSet<ScriptBuf>>,
+    /// 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<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: WalletInterface>
@@ -135,6 +150,8 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
             active_batches: BTreeMap::new(),
             processing_height: 0,
             tracker: BlockMatchTracker::new(),
+            backward_scripts: HashMap::new(),
+            committed_range_sweeps: 0,
         }
     }
 
@@ -160,6 +177,11 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
         self.tracker.clear();
         self.pending_batches.clear();
         self.filter_pipeline = FiltersPipeline::new();
+        // `backward_scripts` is kept: the restarted scan covers heights above
+        // its entry point with the wallet's current script set, while the
+        // accumulated scripts still owe a sweep of the committed prefix below
+        // it. They get that sweep when the restarted scan's forward pipeline
+        // drains.
     }
 
     async fn load_filters(
@@ -529,6 +551,18 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
             && self.progress.committed_height() >= self.progress.filter_header_tip_height()
             && self.progress.committed_height() >= self.progress.target_height()
         {
+            // Every commit goes through `try_commit_batches`, where the last
+            // batch — examined when it alone remains and its end has reached
+            // the filter-header tip — either sweeps the accumulated backward
+            // scripts or stays uncommitted on the blocks its sweep found. A
+            // batch whose end is still below the tip commits without the
+            // sweep, but then `committed_height` is below the tip too and
+            // this branch is not taken. So no newly derived script is still
+            // waiting on its committed-range test here.
+            debug_assert!(
+                self.backward_scripts.is_empty(),
+                "backward scripts pending with no active batches"
+            );
             if self.state() == SyncState::Syncing {
                 self.set_state(SyncState::Synced);
             }
@@ -590,12 +624,12 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
                     // 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 here
-                    // and sweep once when the forward fixpoint is quiescent,
-                    // instead of re-walking the committed range on every
-                    // derivation round.
-                    if let Some(batch) = self.active_batches.get_mut(&batch_start) {
-                        batch.accumulate_backward_scripts(scripts_by_wallet);
+                    // 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.
@@ -604,17 +638,23 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
                     }
                 }
 
-                // Forward direction quiescent: one combined backward sweep
-                // over the committed range with everything accumulated. 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 backward_scripts = self
-                    .active_batches
-                    .get_mut(&batch_start)
-                    .map(|b| b.take_backward_scripts())
-                    .unwrap_or_default();
-                if !backward_scripts.is_empty() {
+                // 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?);
 
@@ -1113,10 +1153,13 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
     /// matching blocks are re-downloaded, via the same
     /// `track_for_new_scripts` path `rescan_batch` uses.
     ///
-    /// Matched blocks attribute to the committing batch at `batch_start`:
-    /// its `pending_blocks` accounting defers the commit, and scripts their
-    /// processing derives collect into that batch, so the existing
-    /// commit-time fixpoint loop covers the backward direction too.
+    /// Called by `try_commit_batches` once the forward pipeline has drained,
+    /// with the scripts accumulated in `backward_scripts` across every
+    /// commit since the previous sweep. Matched blocks attribute to the
+    /// committing batch at `batch_start`: its `pending_blocks` accounting
+    /// defers the commit, and scripts their processing derives collect into
+    /// that batch, so the existing commit-time fixpoint loop covers the
+    /// backward direction too.
     ///
     /// Deliberately matches only `new_scripts` — not the wallets' bare
     /// filter elements — since those were already watched when the range
@@ -1151,11 +1194,13 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
             return Ok(vec![]);
         }
 
+        self.committed_range_sweeps += 1;
         tracing::info!(
-            "Rescan committed filters ({}-{}) for new scripts across {} wallets",
+            "Rescan committed filters ({}-{}) for new scripts across {} wallets (sweep #{})",
             range_start,
             range_end,
-            wallet_queries.len()
+            wallet_queries.len(),
+            self.committed_range_sweeps
         );
 
         let mut block_to_wallets: BTreeMap<FilterMatchKey, BTreeSet<WalletId>> = BTreeMap::new();