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; } }