From 33ad4e267145dfed697de669043702d13907d11a Mon Sep 17 00:00:00 2001 From: Bartosz Rozwarski Date: Wed, 9 Sep 2026 19:34:05 +0200 Subject: [PATCH] fix(dash-spv): collect the scripts derived by every application of a block, not only the tracked one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A block is applied more than once during a scan: re-applied by a batch rescan for scripts derived after its first application, or delivered again for an in-flight re-request. Each application can recognise outputs the previous one could not, and derive further scripts through gap widening. The BlockProcessed handler collected new scripts only when the tracker still held the block's in-flight entry — which the first delivery consumes — so the scripts derived by every later application were dropped: never collected for the batch rescan and the committed-range sweep. On a CoinJoin-heavy mainnet wallet (fresh import, release build) 297 of the 3 645 gap-widened scripts were derived by re-applied blocks and lost. The blocks that spend the coins paid to those addresses match only on the coins' prevout scripts, so they were never found; the wallet ended the scan with 11 spent coins credited (dashpay/rust-dashcore#1006). Settle the in-flight entry and the pending count from the tracked delivery only, but collect new scripts from every delivery: into the tracked batch while it is active, else the batch covering the block, else the lowest active batch — whose commit rescans it, every later batch and the committed prefix. The rescan and sweep log lines now carry the script count. Regression test: a second BlockProcessed for an already-settled block must leave its scripts in the batch's collected scripts. Co-Authored-By: Claude Fable 5.1 --- .../filters/coinjoin_gap_discovery_tests.rs | 87 +++++++++++++++++++ dash-spv/src/sync/filters/manager.rs | 6 +- dash-spv/src/sync/filters/sync_manager.rs | 55 ++++++++++-- 3 files changed, 141 insertions(+), 7 deletions(-) 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 bf064fd97..9a85e9d4c 100644 --- a/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs +++ b/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs @@ -584,3 +584,90 @@ async fn committed_range_sweep_coalesces_across_batch_commits() { manager.committed_range_sweeps ); } + +/// A block is applied more than once during a scan — re-applied by a +/// rescan for scripts derived after its first application, or delivered +/// again for an in-flight re-request — and each application can recognise +/// outputs the previous one could not, deriving further scripts. Only the +/// first delivery carries the tracker's in-flight entry; the scripts derived +/// by the later ones used to be dropped on the floor (never collected into +/// the batch), so blocks that match only on those scripts — typically the +/// blocks that spend the coins paid to them — were never found. On a +/// CoinJoin-heavy mainnet wallet this left 297 of 3 645 gap-widened scripts +/// unswept and 11 spent coins credited (dashpay/rust-dashcore#1006). +#[tokio::test] +async fn scripts_derived_by_a_redelivered_block_enter_the_cascade() { + let (mut manager, wallet, wallet_id) = setup().await; + let addresses = coinjoin_external_addresses(&wallet, &wallet_id, 40).await; + let (block, filter, key) = block_paying(10, &addresses[0..=5]); + { + 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.header, filter.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 block_hash = block.block_hash(); + let mut batch_0 = FiltersBatch::new(0, 99, HashMap::from([(key, filter)])); + batch_0.mark_verified(); + manager.active_batches.insert(0, batch_0); + manager.progress.update_stored_height(99); + + // Forward pass: the block matches on the initial window and is requested. + let events = manager.try_process_batch().await.unwrap(); + let needed: Vec<_> = + events.iter().filter(|e| matches!(e, SyncEvent::BlocksNeeded { .. })).collect(); + assert_eq!(needed.len(), 1, "the forward pass must request the paying block"); + // The batch is still waiting on another block when the re-delivery + // arrives — the shape observed on mainnet, where the re-applied blocks + // landed while their batch had other blocks outstanding. + manager.active_batches.get_mut(&0).expect("batch 0 active").set_pending_blocks(2); + + let (tx, _rx) = unbounded_channel(); + let requests = RequestSender::new(tx); + let wallets = BTreeSet::from([wallet_id]); + + // First (tracked) delivery: applied for real, settles the in-flight entry. + let result = + wallet.write().await.process_block_for_wallets(&block, block_hash, 10, &wallets).await; + let confirmed_txids: Vec<_> = result.relevant_txids().cloned().collect(); + let first = SyncEvent::BlockProcessed { + block_hash, + height: 10, + wallets: wallets.clone(), + new_scripts: result.new_scripts, + confirmed_txids, + }; + manager.handle_sync_event(&first, &requests).await.expect("first delivery"); + + // Second delivery of the same block (a rescan re-application): it derives + // a script the first one did not. Nothing tracks this delivery. + let late_script = addresses[35].script_pubkey(); + let second = SyncEvent::BlockProcessed { + block_hash, + height: 10, + wallets: wallets.clone(), + new_scripts: BTreeMap::from([(wallet_id, vec![late_script.clone()])]), + confirmed_txids: vec![], + }; + manager.handle_sync_event(&second, &requests).await.expect("second delivery"); + + let collected = + manager.active_batches.get_mut(&0).map(|b| b.take_collected_scripts()).unwrap_or_default(); + assert!( + collected.get(&wallet_id).is_some_and(|s| s.contains(&late_script)), + "a script derived by a re-applied block must be collected for the batch's rescan and the \ + committed-range sweep, not dropped because the tracker had already settled the block" + ); +} diff --git a/dash-spv/src/sync/filters/manager.rs b/dash-spv/src/sync/filters/manager.rs index 9e7834e48..277ef153c 100644 --- a/dash-spv/src/sync/filters/manager.rs +++ b/dash-spv/src/sync/filters/manager.rs @@ -854,9 +854,10 @@ impl(), new_scripts.len() ); @@ -1213,9 +1214,10 @@ impl(), wallet_queries.len(), self.committed_range_sweeps ); diff --git a/dash-spv/src/sync/filters/sync_manager.rs b/dash-spv/src/sync/filters/sync_manager.rs index 2dbd8ad5e..e569f3e9c 100644 --- a/dash-spv/src/sync/filters/sync_manager.rs +++ b/dash-spv/src/sync/filters/sync_manager.rs @@ -182,8 +182,10 @@ impl< // `tracker.track` residual. self.tracker.record_processed(*height, *block_hash, wallets); - // Check if this block is part of our tracked blocks - if let Some((_, batch_start)) = self.tracker.finish_in_flight(block_hash) { + // Settle the in-flight entry, if this delivery is the tracked + // one, and its batch's pending count. + let tracked_batch = self.tracker.finish_in_flight(block_hash).map(|(_, b)| b); + if let Some(batch_start) = tracked_batch { if let Some(batch) = self.active_batches.get_mut(&batch_start) { batch.decrement_pending_blocks(); tracing::debug!( @@ -194,17 +196,60 @@ impl< batch.pending_blocks() ); } + } - // Collect per-wallet new scripts for deferred rescan at commit time. + // Every application of a block can extend the pools: a block + // re-applied by a rescan, or delivered again for an in-flight + // re-request, recognises outputs its first application could + // not, and the scripts it derives must enter the cascade + // whether or not this delivery was the tracked one. Only the + // tracked delivery settles the pending count above; scripts + // are collected from all of them. They are charged to the + // tracked batch while it is active, else to the batch covering + // the block, else to the lowest active batch, whose commit + // rescans it, every later batch and the committed prefix. + let has_new_scripts = new_scripts.values().any(|s| !s.is_empty()); + if has_new_scripts { + let target = tracked_batch + .filter(|b| self.active_batches.contains_key(b)) + .or_else(|| { + self.active_batches + .iter() + .find(|(_, b)| { + b.start_height() <= *height && *height <= b.end_height() + }) + .map(|(start, _)| *start) + }) + .or_else(|| self.active_batches.keys().next().copied()); + if tracked_batch.is_none() { + tracing::debug!( + "Block {} at height {} re-applied outside its tracked delivery; {} new scripts collected into batch {:?}", + block_hash, + height, + new_scripts.values().map(|s| s.len()).sum::(), + target + ); + } for (wallet_id, scripts) in new_scripts { if scripts.is_empty() { continue; } - if let Some(batch) = self.active_batches.get_mut(&batch_start) { - batch.add_scripts_for_wallet(*wallet_id, scripts.iter().cloned()); + match target.and_then(|b| self.active_batches.get_mut(&b)) { + Some(batch) => { + batch.add_scripts_for_wallet(*wallet_id, scripts.iter().cloned()) + } + None => tracing::warn!( + "No active batch to carry {} new scripts derived from block {} at height {}; \ + they will only be re-tested by a later rescan", + scripts.len(), + block_hash, + height + ), } } + } + if tracked_batch.is_some() || has_new_scripts { return self.try_process_batch().await; } }