From 108ee9c514697aef46fe939238dc0f8d65b39c40 Mon Sep 17 00:00:00 2001 From: Bartosz Rozwarski Date: Wed, 9 Sep 2026 18:46:56 +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: neither noted in the durable pending-sweep set nor 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. With no active batch the scripts stay in the durable pending-sweep set and are replayed by the next scan, as before. 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 both in the pending-sweep set and in the batch's collected scripts. Co-Authored-By: Claude Fable 5.1 --- .../filters/coinjoin_gap_discovery_tests.rs | 92 +++++++++++++++++++ dash-spv/src/sync/filters/manager.rs | 6 +- dash-spv/src/sync/filters/sync_manager.rs | 55 ++++++++++- 3 files changed, 146 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 16810c436..2a0fdff73 100644 --- a/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs +++ b/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs @@ -912,3 +912,95 @@ async fn born_wrong_record_is_corrected_by_gap_rescan() { PAY as i64 - FUND as i64, ); } + +/// 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 (neither collected into +/// the batch nor recorded in the durable pending-sweep set), 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"); + + assert!( + manager.pending_sweep.get(&wallet_id).is_some_and(|s| s.contains(&late_script)), + "a script derived by a re-applied block must be recorded in the durable pending-sweep set" + ); + 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 79940ee9e..5029687c0 100644 --- a/dash-spv/src/sync/filters/manager.rs +++ b/dash-spv/src/sync/filters/manager.rs @@ -1095,9 +1095,10 @@ impl(), new_scripts.len() ); @@ -1454,9 +1455,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 9b4855ca8..3861038ef 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,8 +196,40 @@ 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; @@ -206,11 +240,22 @@ impl< // scripts next session instead of orphaning heights // scanned before the scripts existed. self.note_pending_sweep(*wallet_id, scripts.iter().cloned()).await; - 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 stay in the durable pending-sweep set until the next scan", + scripts.len(), + block_hash, + height + ), } } + } + if tracked_batch.is_some() || has_new_scripts { return self.try_process_batch().await; } }