From 2592b836573d153c602091ea948bc459d916bd1b Mon Sep 17 00:00:00 2001 From: Roman <51091564+jeanpierreroma@users.noreply.github.com> Date: Tue, 18 Aug 2026 22:19:04 +0300 Subject: [PATCH 1/2] fix(dash-spv): coalesce committed-range filter rescans across batch commits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The backward sweep introduced for #846 re-tested newly derived scripts against the whole committed prefix at every script-carrying batch commit. On a mainnet restore of a heavily CoinJoined wallet that meant 191 full sweeps from the same floor — 301M filter evaluations, ~14.5 minutes of a 28-minute run that jetsam then killed. Move the accumulated backward scripts from the per-batch state to the manager, and run the sweep only when the forward pipeline drains: the committing batch is the last one active and its end has reached the filter-header tip. Intermediate commits accumulate and move on, so a sync shares one walk of the stored history plus one walk per follow-up round whose recovered blocks derive genuinely new scripts. Sweep hits still attribute to the committing batch, so its pending_blocks accounting holds the commit — and with it FiltersSyncComplete — until the backward fixpoint converges, and the accumulator deliberately survives reset_for_rescan so a restarted scan still owes the committed prefix its pass. The regression test drives four batches, three of which derive new scripts at commit, through the real wallet pipeline: unmodified code runs 4 committed-range sweeps, coalesced code runs 2 (drain-time sweep plus one follow-up round), and a beyond-window block in the first committed batch must still be recovered before FiltersSyncComplete is emitted. Co-Authored-By: Claude Opus 5 --- dash-spv/src/sync/filters/batch.rs | 22 --- .../filters/coinjoin_gap_discovery_tests.rs | 165 ++++++++++++++++++ dash-spv/src/sync/filters/manager.rs | 91 +++++++--- 3 files changed, 233 insertions(+), 45 deletions(-) diff --git a/dash-spv/src/sync/filters/batch.rs b/dash-spv/src/sync/filters/batch.rs index 43f1e8384..b389c5a1b 100644 --- a/dash-spv/src/sync/filters/batch.rs +++ b/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>, - /// 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>, } 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> { 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>, - ) { - 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> { - 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) { diff --git a/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs b/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs index a2f54d192..f948aa69f 100644 --- a/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs +++ b/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,163 @@ 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 = 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); + + // Drive the production event loop to quiescence like `drive_to_quiescence` + // does, additionally watching for `FiltersSyncComplete` so the completion + // contract can be asserted at the moment it is emitted. + let (tx, _rx) = unbounded_channel(); + let requests = RequestSender::new(tx); + let mut events = manager.try_process_batch().await.unwrap(); + let mut sync_complete_seen = false; + 'rounds: for _round in 0..64 { + let mut pending: BTreeMap<(u32, BlockHash), BTreeSet> = BTreeMap::new(); + for event in events.drain(..) { + match event { + SyncEvent::BlocksNeeded { + blocks: needed, + } => { + for (key, wallets) in needed { + pending.entry((key.height(), *key.hash())).or_default().extend(wallets); + } + } + SyncEvent::FiltersSyncComplete { + .. + } => { + let (highest_used, _, _) = coinjoin_pool_state(&wallet, &wallet_id).await; + assert_eq!( + highest_used, + Some((G + 21) as u32), + "FiltersSyncComplete emitted before the deferred committed-range \ + sweep recovered the height-10 block: a script derived after its \ + range committed was never tested against the committed prefix" + ); + sync_complete_seen = true; + } + _ => {} + } + } + if pending.is_empty() { + break 'rounds; + } + for ((height, block_hash), wallets) in pending { + let block = blocks.get(&block_hash).expect("BlocksNeeded for an unknown test block"); + let result = wallet + .write() + .await + .process_block_for_wallets(block, block_hash, height, &wallets) + .await; + let confirmed_txids = result.relevant_txids().cloned().collect(); + let event = SyncEvent::BlockProcessed { + block_hash, + height, + wallets, + new_scripts: result.new_scripts, + confirmed_txids, + }; + events.extend( + manager.handle_sync_event(&event, &requests).await.expect("BlockProcessed"), + ); + } + } + + assert!(sync_complete_seen, "the run must reach FiltersSyncComplete"); + + let (highest_used, _, used_count) = coinjoin_pool_state(&wallet, &wallet_id).await; + assert_eq!( + highest_used, + Some((G + 21) as u32), + "the height-10 block's beyond-window outputs must be recovered by the \ + drain-time committed-range sweep (used_count={used_count})" + ); + assert_eq!(used_count, 90 + 12, "indices 0..=89 and G+10..=G+21 must all be marked used"); + + assert!( + manager.committed_range_sweeps >= 1, + "the deferred committed-range sweep must still run before completion" + ); + assert!( + manager.committed_range_sweeps <= 2, + "three script-carrying batch commits must share the committed-range sweep \ + (one drain-time sweep plus one follow-up round for the scripts derived from \ + the recovered block); got {} sweeps — one sweep per commit means the \ + coalescing regressed", + manager.committed_range_sweeps + ); +} diff --git a/dash-spv/src/sync/filters/manager.rs b/dash-spv/src/sync/filters/manager.rs index 5fc18c5f3..d9fc901ec 100644 --- a/dash-spv/src/sync/filters/manager.rs +++ b/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>, + /// 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 @@ -135,6 +150,8 @@ impl= 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 0 { // Forward rescan found blocks; converge the // forward direction first. @@ -604,17 +638,23 @@ impl= 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> = BTreeMap::new(); From 5aff465ee3316eab89bd9378edbab9d825fcb870 Mon Sep 17 00:00:00 2001 From: Roman <51091564+jeanpierreroma@users.noreply.github.com> Date: Thu, 20 Aug 2026 12:21:02 +0300 Subject: [PATCH 2/2] test(dash-spv): make the drain-gate test compare against real progress tips `setup()` leaves `filter_header_tip_height` and `target_height` at 0, so `end_height() >= filter_header_tip_height()` was trivially true and the test only ever exercised the `active_batches.len() == 1` half of the gate. `FiltersSyncComplete` was being checked against a zero target for the same reason. Both tips are now seeded to 399, matching the fixture. This does not make the height comparison itself covered, and I could not get it covered: with the comparison deleted the suite still passes. A fixture where one batch is in flight with more of the chain ahead does not reach the sweep branch at all in these tests, so an assertion there passes for the wrong reason rather than for the right one. I removed the attempt instead of keeping a test that proves nothing. What that half does in production: it separates "this is the last batch" from "this is the only batch right now", which is what keeps the sweep from firing per commit early in a scan. It is exercised by the end-to-end device run in the PR description, not by the unit suite. --- dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs b/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs index f948aa69f..bf064fd97 100644 --- a/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs +++ b/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs @@ -493,6 +493,14 @@ async fn committed_range_sweep_coalesces_across_batch_commits() { 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