fix(dash-spv): collect the scripts derived by every application of a block, not only the tracked one

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"
+    );
+}

dash-spv/src/sync/filters/manager.rs

@@ -1095,9 +1095,10 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
         };
 
         tracing::info!(
-            "Rescan filters ({}-{}) for new scripts across {} wallets",
+            "Rescan filters ({}-{}) for {} new scripts across {} wallets",
             batch.start_height(),
             batch.end_height(),
+            new_scripts.values().map(|s| s.len()).sum::<usize>(),
             new_scripts.len()
         );
 
@@ -1454,9 +1455,10 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage, F: FilterStorage, W: Wallet
 
         self.committed_range_sweeps += 1;
         tracing::info!(
-            "Rescan committed filters ({}-{}) for new scripts across {} wallets (sweep #{})",
+            "Rescan committed filters ({}-{}) for {} new scripts across {} wallets (sweep #{})",
             range_start,
             range_end,
+            wallet_queries.iter().map(|(_, s)| s.len()).sum::<usize>(),
             wallet_queries.len(),
             self.committed_range_sweeps
         );

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::<usize>(),
+                            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;
                 }
             }