Skip to content

refactor(dash-spv): drop committed range sweep #1016

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ZocoLini merged 1 commit into from
Sep 23, 2026
Merged

refactor(dash-spv): drop committed range sweep #1016

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 0 additions & 28 deletions dash-spv/src/sync/filters/batch.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,6 @@ pub(super) struct FiltersBatch {
scanned: bool,
/// Number of blocks still being downloaded for this batch.
pending_blocks: u32,
/// Whether rescan has been completed for this batch.
rescan_complete: bool,
/// Wallets that were behind for this batch's height range at scan time —
/// and therefore need their `synced_height` advanced when the batch
/// commits — each mapped to the wallet's `account_generation` at scan
Expand All @@ -33,10 +31,6 @@ pub(super) struct FiltersBatch {
/// current account set (dashpay/rust-dashcore#649). Already-synced wallets
/// must not be touched.
scanned_wallets: BTreeMap<WalletId, u64>,
/// Cached scriptPubKeys discovered during block processing that still
/// 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>>,
/// Every script already matched against this batch's filters, per wallet.
tested_scripts: HashMap<WalletId, HashSet<ScriptBuf>>,
}
Expand All @@ -55,9 +49,7 @@ impl FiltersBatch {
verified: false,
scanned: false,
pending_blocks: 0,
rescan_complete: false,
scanned_wallets: BTreeMap::new(),
collected_scripts: HashMap::new(),
tested_scripts: HashMap::new(),
}
}
Expand Down Expand Up @@ -106,22 +98,6 @@ impl FiltersBatch {
self.pending_blocks = self.pending_blocks.saturating_sub(1);
self.pending_blocks
}
/// Returns whether rescan has been completed for this batch.
pub(super) fn rescan_complete(&self) -> bool {
self.rescan_complete
}
/// Mark rescan as complete for this batch.
pub(super) fn mark_rescan_complete(&mut self) {
self.rescan_complete = true;
}
/// Add scriptPubKeys discovered during block processing for later rescan.
pub(super) fn add_scripts_for_wallet(
&mut self,
wallet_id: WalletId,
scripts: impl IntoIterator<Item = ScriptBuf>,
) {
self.collected_scripts.entry(wallet_id).or_default().extend(scripts);
}
/// Record that `scripts` have been matched against this batch's filters.
pub(super) fn mark_tested<I: IntoIterator<Item = ScriptBuf>>(
&mut self,
Expand All @@ -141,10 +117,6 @@ impl FiltersBatch {
monitored.iter().filter(move |script| tested.is_none_or(|t| !t.contains(*script)))
}

/// Take collected per-wallet scripts for rescan, leaving the map empty.
pub(super) fn take_collected_scripts(&mut self) -> HashMap<WalletId, HashSet<ScriptBuf>> {
std::mem::take(&mut self.collected_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>) {
Expand Down
5 changes: 0 additions & 5 deletions dash-spv/src/sync/filters/block_match_tracker.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,11 +148,6 @@ impl BlockMatchTracker {
self.processed_blocks_per_wallet.split_off(&(height + 1));
}

/// True while matched blocks are still awaiting their `BlockProcessed`.
pub(super) fn has_blocks_in_flight(&self) -> bool {
!self.blocks_remaining.is_empty()
}

/// True when there is no in-flight or processed-record state.
pub(super) fn is_empty(&self) -> bool {
self.blocks_remaining.is_empty() && self.processed_blocks_per_wallet.is_empty()
Expand Down
275 changes: 2 additions & 273 deletions dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,24 +12,14 @@
//! `rescan_batch` now re-queues already-processed blocks of ACTIVE batches
//! against newly derived scripts, to a fixpoint.
//!
//! These tests pin both the fixed behaviour and the remaining hole:
//! These tests pin the fixed behaviour:
//!
//! 1. [`coinjoin_gap_limit_dense_same_batch_recovers`] — the empirical
//! stall-at-59 shape (two dense blocks in one batch). GREEN since #820;
//! kept as a regression guard.
//! 2. [`coinjoin_gap_limit_inversion_within_batch_recovers`] — a gap-window
//! output in an EARLIER block of the SAME (still-active) batch is
//! recovered by the commit-time rescan. GREEN since #820.
//! 3. [`coinjoin_gap_limit_stall_across_committed_batch`] — the SAME shape
//! with the earlier block in an already-COMMITTED batch (#846): rescans
//! 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)
Expand Down Expand Up @@ -296,8 +286,7 @@ async fn coinjoin_gap_limit_dense_same_batch_recovers() {
/// only at height 20. The initial scan cannot see block A (nothing watched
/// matches it), but once block B's processing extends the window past G+21
/// the commit-time rescan re-tests block A's filter and recovers it. GREEN
/// since #820 — contrast for the cross-batch test below, isolating the
/// commit boundary as the broken seam.
/// since #820.
#[tokio::test]
async fn coinjoin_gap_limit_inversion_within_batch_recovers() {
let (mut manager, wallet, wallet_id) = setup().await;
Expand All @@ -324,263 +313,3 @@ async fn coinjoin_gap_limit_inversion_within_batch_recovers() {
recovered by the commit-time rescan (PR #820). used_count={used_count}"
);
}

/// Gap-window outputs in an already-COMMITTED batch (#846).
///
/// Same funding shape as the within-batch inversion test, but the early
/// block (indices G+10..=G+21, height 10) sits in batch 0..=99 while the
/// in-window block (indices 0..=29) sits at height 110 in batch 100..=199.
/// Batch 0 scans clean (nothing watched matches) and commits. Processing the
/// height-110 block extends the window past G+21, and those scripts DO match
/// block 10's filter — but `rescan_batch` only reaches `active_batches`, and
/// committed batches are gone (`try_commit_batches` removes them; the
/// tracker prunes at-or-below the committed height). Indices G+10..=G+21 —
/// squarely inside the BIP-44/CoinJoin gap-limit recovery contract
/// (G+21 < 29 + 1 + G) — used to stay invisible forever, along with their
/// funds; a fresh re-sync from genesis hit the same wall deterministically.
///
/// GREEN since `rescan_committed_range`: newly derived scripts are re-tested
/// against the persisted filters below the committing batch (BIP-158 filters
/// are address-independent, so re-matching needs no re-download), and hits
/// flow through the `track_for_new_scripts` re-download path to the same
/// commit-time fixpoint. `highest_used` reaches G+21.
#[tokio::test]
async fn coinjoin_gap_limit_stall_across_committed_batch() {
let (mut manager, wallet, wallet_id) = setup().await;
let addresses = coinjoin_external_addresses(&wallet, &wallet_id, (G + 22) as u32).await;

let (block_a, filter_a, key_a) = block_paying(10, &addresses[(G + 10)..=(G + 21)]);
let (block_b, filter_b, key_b) = block_paying(110, &addresses[0..=29]);

// Uphold the production invariant the injected batches imply: every
// height at or below `stored_height` has its header and filter
// persisted (store_and_match_batches stores a batch's filters before
// stored_height advances past it). The committed-range recovery path
// re-tests exactly this stored data, so the invariant is load-bearing
// here: batch 0 commits before block B's processing derives the missing
// scripts, and by then its in-memory filters are gone.
{
let mut header_storage = manager.header_storage.write().await;
let mut filter_storage = manager.filter_storage.write().await;
for height in 0..=99u32 {
let (header, filter_bytes) = if height == 10 {
(block_a.header, filter_a.content.clone())
} else {
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_a.block_hash(), block_a), (block_b.block_hash(), block_b)]);

let mut batch_0 = FiltersBatch::new(0, 99, HashMap::from([(key_a, filter_a)]));
batch_0.mark_verified();
manager.active_batches.insert(0, batch_0);
let mut batch_1 = FiltersBatch::new(100, 199, HashMap::from([(key_b, filter_b)]));
batch_1.mark_verified();
manager.active_batches.insert(100, batch_1);
manager.progress.update_stored_height(199);

let initial_events = manager.try_process_batch().await.unwrap();
drive_to_quiescence(&mut manager, &wallet, &blocks, initial_events).await;

let (highest_used, highest_generated, used_count) =
coinjoin_pool_state(&wallet, &wallet_id).await;
// Sanity: the in-window block was found and the gap window extended past
// index G+21, so the missed indices ARE inside the watched range by now.
assert!(
highest_generated >= Some((G + 21) as u32),
"gap maintenance must have extended the watch window past index G+21 \
(got {highest_generated:?})"
);
assert_eq!(
highest_used,
Some((G + 21) as u32),
"CoinJoin External indices G+10..=G+21 were funded at height 10 in a batch that \
committed before their scripts were derived, and the new-script rescan never \
looks below the committed boundary (rescan_batch only reaches active_batches; \
BlockMatchTracker/commit pruning drops the range). The addresses are within \
the gap-limit recovery contract and are watched now (highest_generated = \
{highest_generated:?}), yet their outputs stay invisible: highest_used stalls \
at {highest_used:?}, used_count={used_count}. Fix direction: key re-scan \
suppression by (wallet, address/script) instead of block/commit progress, or \
trigger a below-committed-height rescan for a wallet whose gap maintenance \
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
);
}
Loading
Loading