Skip to content

Commit 1674cd2

Browse files
ZocoLiniclaude
andcommitted
refactor(dash-spv): drop the committed-range sweep and the collected-scripts rescan
The #846 sweep re-tested every late-derived script against the whole stored filter history, from the wallet's birth height. On a mainnet restore of the bench wallet that was one 63 s walk over 200 000..tip matching 41 687 blocks, overwhelmingly compact-filter false positives, then about 17.7 min downloading and applying them, out of a 21–27 min sync. The notification-driven path (`collect_new_scripts`, `FiltersBatch::collected_scripts`, the current-and-later batch rescan at commit) duplicated `reconcile_untested_scripts`, which asks the wallet what it watches, a superset of any `new_scripts`, and re-tests whatever the batch never matched. `rescan_complete` was set immediately before the batch left `active_batches`, so it was never read as true. Trade-off: the #846 case, a CoinJoin output paying an index derived only after its batch committed, is no longer recovered. Its repro, `coinjoin_gap_limit_stall_across_committed_batch`, fails without the sweep and is removed with it, as is the sweep-cost test. The bench wallet does not hit that case. Validated on top of the spend re-application fix: 5 consecutive mainnet restores at 100 Mbit / 100 ms, then 100/50 Mbit x 100/500 ms, all ending at 7112 records, 14 114 383 sat and 13 389 addresses, in 6.0–17.2 min. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017DruChNTWXwJoWPartZwCf
1 parent 59dd41b commit 1674cd2

5 files changed

Lines changed: 13 additions & 772 deletions

File tree

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

Lines changed: 0 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,6 @@ pub(super) struct FiltersBatch {
2222
scanned: bool,
2323
/// Number of blocks still being downloaded for this batch.
2424
pending_blocks: u32,
25-
/// Whether rescan has been completed for this batch.
26-
rescan_complete: bool,
2725
/// Wallets that were behind for this batch's height range at scan time —
2826
/// and therefore need their `synced_height` advanced when the batch
2927
/// commits — each mapped to the wallet's `account_generation` at scan
@@ -33,10 +31,6 @@ pub(super) struct FiltersBatch {
3331
/// current account set (dashpay/rust-dashcore#649). Already-synced wallets
3432
/// must not be touched.
3533
scanned_wallets: BTreeMap<WalletId, u64>,
36-
/// Cached scriptPubKeys discovered during block processing that still
37-
/// need rescan, attributed per wallet so we can rerun matching only
38-
/// against the wallet that produced each new script.
39-
collected_scripts: HashMap<WalletId, HashSet<ScriptBuf>>,
4034
/// Every script already matched against this batch's filters, per wallet.
4135
tested_scripts: HashMap<WalletId, HashSet<ScriptBuf>>,
4236
}
@@ -55,9 +49,7 @@ impl FiltersBatch {
5549
verified: false,
5650
scanned: false,
5751
pending_blocks: 0,
58-
rescan_complete: false,
5952
scanned_wallets: BTreeMap::new(),
60-
collected_scripts: HashMap::new(),
6153
tested_scripts: HashMap::new(),
6254
}
6355
}
@@ -106,22 +98,6 @@ impl FiltersBatch {
10698
self.pending_blocks = self.pending_blocks.saturating_sub(1);
10799
self.pending_blocks
108100
}
109-
/// Returns whether rescan has been completed for this batch.
110-
pub(super) fn rescan_complete(&self) -> bool {
111-
self.rescan_complete
112-
}
113-
/// Mark rescan as complete for this batch.
114-
pub(super) fn mark_rescan_complete(&mut self) {
115-
self.rescan_complete = true;
116-
}
117-
/// Add scriptPubKeys discovered during block processing for later rescan.
118-
pub(super) fn add_scripts_for_wallet(
119-
&mut self,
120-
wallet_id: WalletId,
121-
scripts: impl IntoIterator<Item = ScriptBuf>,
122-
) {
123-
self.collected_scripts.entry(wallet_id).or_default().extend(scripts);
124-
}
125101
/// Record that `scripts` have been matched against this batch's filters.
126102
pub(super) fn mark_tested<I: IntoIterator<Item = ScriptBuf>>(
127103
&mut self,
@@ -141,10 +117,6 @@ impl FiltersBatch {
141117
monitored.iter().filter(move |script| tested.is_none_or(|t| !t.contains(*script)))
142118
}
143119

144-
/// Take collected per-wallet scripts for rescan, leaving the map empty.
145-
pub(super) fn take_collected_scripts(&mut self) -> HashMap<WalletId, HashSet<ScriptBuf>> {
146-
std::mem::take(&mut self.collected_scripts)
147-
}
148120
/// Record the wallets that were behind for this batch at scan time, each
149121
/// with its `account_generation` snapshot.
150122
pub(super) fn set_scanned_wallets(&mut self, wallets: BTreeMap<WalletId, u64>) {

‎dash-spv/src/sync/filters/block_match_tracker.rs‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -148,11 +148,6 @@ impl BlockMatchTracker {
148148
self.processed_blocks_per_wallet.split_off(&(height + 1));
149149
}
150150

151-
/// True while matched blocks are still awaiting their `BlockProcessed`.
152-
pub(super) fn has_blocks_in_flight(&self) -> bool {
153-
!self.blocks_remaining.is_empty()
154-
}
155-
156151
/// True when there is no in-flight or processed-record state.
157152
pub(super) fn is_empty(&self) -> bool {
158153
self.blocks_remaining.is_empty() && self.processed_blocks_per_wallet.is_empty()

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

Lines changed: 2 additions & 273 deletions
Original file line numberDiff line numberDiff line change
@@ -12,24 +12,14 @@
1212
//! `rescan_batch` now re-queues already-processed blocks of ACTIVE batches
1313
//! against newly derived scripts, to a fixpoint.
1414
//!
15-
//! These tests pin both the fixed behaviour and the remaining hole:
15+
//! These tests pin the fixed behaviour:
1616
//!
1717
//! 1. [`coinjoin_gap_limit_dense_same_batch_recovers`] — the empirical
1818
//! stall-at-59 shape (two dense blocks in one batch). GREEN since #820;
1919
//! kept as a regression guard.
2020
//! 2. [`coinjoin_gap_limit_inversion_within_batch_recovers`] — a gap-window
2121
//! output in an EARLIER block of the SAME (still-active) batch is
2222
//! recovered by the commit-time rescan. GREEN since #820.
23-
//! 3. [`coinjoin_gap_limit_stall_across_committed_batch`] — the SAME shape
24-
//! with the earlier block in an already-COMMITTED batch (#846): rescans
25-
//! only reach `active_batches`, and committed batches are gone. GREEN
26-
//! since `rescan_committed_range` re-tests newly derived scripts against
27-
//! the STORED filters below the committing batch.
28-
//! 4. [`committed_range_sweep_coalesces_across_batch_commits`] — the cost
29-
//! side of #846's fix: N script-carrying commits share one drain-time
30-
//! committed-range sweep instead of walking the stored history once per
31-
//! commit, while a late-derived script still finds its
32-
//! committed-prefix block before `FiltersSyncComplete`.
3323
//!
3424
//! Each test drives the manager exactly the way the production event loop
3525
//! does: `try_process_batch` → `BlocksNeeded` → (blocks-manager stand-in)
@@ -296,8 +286,7 @@ async fn coinjoin_gap_limit_dense_same_batch_recovers() {
296286
/// only at height 20. The initial scan cannot see block A (nothing watched
297287
/// matches it), but once block B's processing extends the window past G+21
298288
/// the commit-time rescan re-tests block A's filter and recovers it. GREEN
299-
/// since #820 — contrast for the cross-batch test below, isolating the
300-
/// commit boundary as the broken seam.
289+
/// since #820.
301290
#[tokio::test]
302291
async fn coinjoin_gap_limit_inversion_within_batch_recovers() {
303292
let (mut manager, wallet, wallet_id) = setup().await;
@@ -324,263 +313,3 @@ async fn coinjoin_gap_limit_inversion_within_batch_recovers() {
324313
recovered by the commit-time rescan (PR #820). used_count={used_count}"
325314
);
326315
}
327-
328-
/// Gap-window outputs in an already-COMMITTED batch (#846).
329-
///
330-
/// Same funding shape as the within-batch inversion test, but the early
331-
/// block (indices G+10..=G+21, height 10) sits in batch 0..=99 while the
332-
/// in-window block (indices 0..=29) sits at height 110 in batch 100..=199.
333-
/// Batch 0 scans clean (nothing watched matches) and commits. Processing the
334-
/// height-110 block extends the window past G+21, and those scripts DO match
335-
/// block 10's filter — but `rescan_batch` only reaches `active_batches`, and
336-
/// committed batches are gone (`try_commit_batches` removes them; the
337-
/// tracker prunes at-or-below the committed height). Indices G+10..=G+21 —
338-
/// squarely inside the BIP-44/CoinJoin gap-limit recovery contract
339-
/// (G+21 < 29 + 1 + G) — used to stay invisible forever, along with their
340-
/// funds; a fresh re-sync from genesis hit the same wall deterministically.
341-
///
342-
/// GREEN since `rescan_committed_range`: newly derived scripts are re-tested
343-
/// against the persisted filters below the committing batch (BIP-158 filters
344-
/// are address-independent, so re-matching needs no re-download), and hits
345-
/// flow through the `track_for_new_scripts` re-download path to the same
346-
/// commit-time fixpoint. `highest_used` reaches G+21.
347-
#[tokio::test]
348-
async fn coinjoin_gap_limit_stall_across_committed_batch() {
349-
let (mut manager, wallet, wallet_id) = setup().await;
350-
let addresses = coinjoin_external_addresses(&wallet, &wallet_id, (G + 22) as u32).await;
351-
352-
let (block_a, filter_a, key_a) = block_paying(10, &addresses[(G + 10)..=(G + 21)]);
353-
let (block_b, filter_b, key_b) = block_paying(110, &addresses[0..=29]);
354-
355-
// Uphold the production invariant the injected batches imply: every
356-
// height at or below `stored_height` has its header and filter
357-
// persisted (store_and_match_batches stores a batch's filters before
358-
// stored_height advances past it). The committed-range recovery path
359-
// re-tests exactly this stored data, so the invariant is load-bearing
360-
// here: batch 0 commits before block B's processing derives the missing
361-
// scripts, and by then its in-memory filters are gone.
362-
{
363-
let mut header_storage = manager.header_storage.write().await;
364-
let mut filter_storage = manager.filter_storage.write().await;
365-
for height in 0..=99u32 {
366-
let (header, filter_bytes) = if height == 10 {
367-
(block_a.header, filter_a.content.clone())
368-
} else {
369-
let filler = Block::dummy(height, vec![]);
370-
let filter = BlockFilter::dummy(&filler);
371-
(filler.header, filter.content)
372-
};
373-
header_storage
374-
.store_headers_at_height(&[header.into()], height)
375-
.await
376-
.expect("seed header");
377-
filter_storage.store_filter(height, &filter_bytes).await.expect("seed filter");
378-
}
379-
}
380-
381-
let blocks: HashMap<BlockHash, Block> =
382-
HashMap::from([(block_a.block_hash(), block_a), (block_b.block_hash(), block_b)]);
383-
384-
let mut batch_0 = FiltersBatch::new(0, 99, HashMap::from([(key_a, filter_a)]));
385-
batch_0.mark_verified();
386-
manager.active_batches.insert(0, batch_0);
387-
let mut batch_1 = FiltersBatch::new(100, 199, HashMap::from([(key_b, filter_b)]));
388-
batch_1.mark_verified();
389-
manager.active_batches.insert(100, batch_1);
390-
manager.progress.update_stored_height(199);
391-
392-
let initial_events = manager.try_process_batch().await.unwrap();
393-
drive_to_quiescence(&mut manager, &wallet, &blocks, initial_events).await;
394-
395-
let (highest_used, highest_generated, used_count) =
396-
coinjoin_pool_state(&wallet, &wallet_id).await;
397-
// Sanity: the in-window block was found and the gap window extended past
398-
// index G+21, so the missed indices ARE inside the watched range by now.
399-
assert!(
400-
highest_generated >= Some((G + 21) as u32),
401-
"gap maintenance must have extended the watch window past index G+21 \
402-
(got {highest_generated:?})"
403-
);
404-
assert_eq!(
405-
highest_used,
406-
Some((G + 21) as u32),
407-
"CoinJoin External indices G+10..=G+21 were funded at height 10 in a batch that \
408-
committed before their scripts were derived, and the new-script rescan never \
409-
looks below the committed boundary (rescan_batch only reaches active_batches; \
410-
BlockMatchTracker/commit pruning drops the range). The addresses are within \
411-
the gap-limit recovery contract and are watched now (highest_generated = \
412-
{highest_generated:?}), yet their outputs stay invisible: highest_used stalls \
413-
at {highest_used:?}, used_count={used_count}. Fix direction: key re-scan \
414-
suppression by (wallet, address/script) instead of block/commit progress, or \
415-
trigger a below-committed-height rescan for a wallet whose gap maintenance \
416-
derives scripts mid-sync."
417-
);
418-
}
419-
420-
/// Committed-range sweeps coalesce across batch commits.
421-
///
422-
/// Four batches; the last three each contain one in-window block whose
423-
/// processing derives new scripts (each funds the next run of CoinJoin
424-
/// indices, so gap maintenance extends the watch window at every commit).
425-
/// Per-commit sweeping walks the entire committed prefix once per
426-
/// script-carrying commit — on a real mainnet restore that shape produced
427-
/// 191 full-prefix sweeps totalling ~14.5 minutes. Coalesced, the
428-
/// accumulated scripts cross the stored history when the forward pipeline
429-
/// drains: one sweep, plus one follow-up round for the scripts derived from
430-
/// the block that sweep recovers.
431-
///
432-
/// The early block (height 10, beyond-window indices G+10..=G+21, unwatched
433-
/// when its batch scans and commits) pins the #846 correctness contract at
434-
/// the same time: the deferred sweep must still find it, and
435-
/// `FiltersSyncComplete` must not be emitted before it has been found and
436-
/// applied. `committed_range_sweeps` counts sweeps that reach the chunk walk
437-
/// in `rescan_committed_range`.
438-
#[tokio::test]
439-
async fn committed_range_sweep_coalesces_across_batch_commits() {
440-
let (mut manager, wallet, wallet_id) = setup().await;
441-
let addresses = coinjoin_external_addresses(&wallet, &wallet_id, (G + 22) as u32).await;
442-
443-
// Beyond-window block in the range that commits first (#846 shape).
444-
let (block_early, filter_early, key_early) = block_paying(10, &addresses[(G + 10)..=(G + 21)]);
445-
// One in-window block per later batch. Each extends `highest_used` by 30,
446-
// so every one of these batches carries newly derived scripts into its
447-
// commit. After block 110 the generated window reaches 29 + G >= G + 21,
448-
// so the early block's scripts are among the first commit's derivations.
449-
let (block_1, filter_1, key_1) = block_paying(110, &addresses[0..=29]);
450-
let (block_2, filter_2, key_2) = block_paying(210, &addresses[30..=59]);
451-
let (block_3, filter_3, key_3) = block_paying(310, &addresses[60..=89]);
452-
453-
// Persist headers and filters for the committed prefix 0..=299 — the
454-
// range the drain-time sweep reloads from storage. Real data at the
455-
// three block heights, filler elsewhere.
456-
{
457-
let mut header_storage = manager.header_storage.write().await;
458-
let mut filter_storage = manager.filter_storage.write().await;
459-
for height in 0..=299u32 {
460-
let (header, filter_bytes) = match height {
461-
10 => (block_early.header, filter_early.content.clone()),
462-
110 => (block_1.header, filter_1.content.clone()),
463-
210 => (block_2.header, filter_2.content.clone()),
464-
_ => {
465-
let filler = Block::dummy(height, vec![]);
466-
let filter = BlockFilter::dummy(&filler);
467-
(filler.header, filter.content)
468-
}
469-
};
470-
header_storage
471-
.store_headers_at_height(&[header.into()], height)
472-
.await
473-
.expect("seed header");
474-
filter_storage.store_filter(height, &filter_bytes).await.expect("seed filter");
475-
}
476-
}
477-
478-
let blocks: HashMap<BlockHash, Block> = HashMap::from([
479-
(block_early.block_hash(), block_early),
480-
(block_1.block_hash(), block_1),
481-
(block_2.block_hash(), block_2),
482-
(block_3.block_hash(), block_3),
483-
]);
484-
485-
for (start, filters) in [
486-
(0u32, HashMap::from([(key_early, filter_early)])),
487-
(100, HashMap::from([(key_1, filter_1)])),
488-
(200, HashMap::from([(key_2, filter_2)])),
489-
(300, HashMap::from([(key_3, filter_3)])),
490-
] {
491-
let mut batch = FiltersBatch::new(start, start + 99, filters);
492-
batch.mark_verified();
493-
manager.active_batches.insert(start, batch);
494-
}
495-
manager.progress.update_stored_height(399);
496-
// Both halves of the drain gate have to be live, or the test only proves
497-
// the `active_batches.len() == 1` half: with the tips left at their
498-
// default 0, `end_height() >= filter_header_tip_height()` is trivially
499-
// true and the comparison that keeps the sweep from firing while more
500-
// batches are still to come is never exercised. Same for the target
501-
// height, which `FiltersSyncComplete` is checked against below.
502-
manager.progress.update_filter_header_tip_height(399);
503-
manager.progress.update_target_height(399);
504-
505-
// Drive the production event loop to quiescence like `drive_to_quiescence`
506-
// does, additionally watching for `FiltersSyncComplete` so the completion
507-
// contract can be asserted at the moment it is emitted.
508-
let (tx, _rx) = unbounded_channel();
509-
let requests = RequestSender::new(tx);
510-
let mut events = manager.try_process_batch().await.unwrap();
511-
let mut sync_complete_seen = false;
512-
'rounds: for _round in 0..64 {
513-
let mut pending: BTreeMap<(u32, BlockHash), BTreeSet<WalletId>> = BTreeMap::new();
514-
for event in events.drain(..) {
515-
match event {
516-
SyncEvent::BlocksNeeded {
517-
blocks: needed,
518-
} => {
519-
for (key, wallets) in needed {
520-
pending.entry((key.height(), *key.hash())).or_default().extend(wallets);
521-
}
522-
}
523-
SyncEvent::FiltersSyncComplete {
524-
..
525-
} => {
526-
let (highest_used, _, _) = coinjoin_pool_state(&wallet, &wallet_id).await;
527-
assert_eq!(
528-
highest_used,
529-
Some((G + 21) as u32),
530-
"FiltersSyncComplete emitted before the deferred committed-range \
531-
sweep recovered the height-10 block: a script derived after its \
532-
range committed was never tested against the committed prefix"
533-
);
534-
sync_complete_seen = true;
535-
}
536-
_ => {}
537-
}
538-
}
539-
if pending.is_empty() {
540-
break 'rounds;
541-
}
542-
for ((height, block_hash), wallets) in pending {
543-
let block = blocks.get(&block_hash).expect("BlocksNeeded for an unknown test block");
544-
let result = wallet
545-
.write()
546-
.await
547-
.process_block_for_wallets(block, block_hash, height, &wallets)
548-
.await;
549-
let confirmed_txids = result.relevant_txids().cloned().collect();
550-
let event = SyncEvent::BlockProcessed {
551-
block_hash,
552-
height,
553-
wallets,
554-
new_scripts: result.new_scripts,
555-
confirmed_txids,
556-
};
557-
events.extend(
558-
manager.handle_sync_event(&event, &requests).await.expect("BlockProcessed"),
559-
);
560-
}
561-
}
562-
563-
assert!(sync_complete_seen, "the run must reach FiltersSyncComplete");
564-
565-
let (highest_used, _, used_count) = coinjoin_pool_state(&wallet, &wallet_id).await;
566-
assert_eq!(
567-
highest_used,
568-
Some((G + 21) as u32),
569-
"the height-10 block's beyond-window outputs must be recovered by the \
570-
drain-time committed-range sweep (used_count={used_count})"
571-
);
572-
assert_eq!(used_count, 90 + 12, "indices 0..=89 and G+10..=G+21 must all be marked used");
573-
574-
assert!(
575-
manager.committed_range_sweeps >= 1,
576-
"the deferred committed-range sweep must still run before completion"
577-
);
578-
assert!(
579-
manager.committed_range_sweeps <= 2,
580-
"three script-carrying batch commits must share the committed-range sweep \
581-
(one drain-time sweep plus one follow-up round for the scripts derived from \
582-
the recovered block); got {} sweeps — one sweep per commit means the \
583-
coalescing regressed",
584-
manager.committed_range_sweeps
585-
);
586-
}

0 commit comments

Comments
 (0)