Refactor/drop committed range sweep

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

@@ -3,6 +3,7 @@
 //! Downloads blocks that matched wallet filters and processes them in height order.
 //! Subscribes to BlockNeeded events and emits BlockProcessed events.
 
+use std::collections::{BTreeMap, BTreeSet};
 use std::sync::Arc;
 
 use tokio::sync::RwLock;
@@ -12,7 +13,8 @@ use crate::error::SyncResult;
 use crate::network::RequestSender;
 use crate::storage::{BlockHeaderStorage, BlockStorage};
 use crate::sync::{BlocksProgress, SyncEvent, SyncManager, SyncState};
-use key_wallet_manager::WalletInterface;
+use crate::types::HashedBlock;
+use key_wallet_manager::{BlockProcessingResult, WalletId, WalletInterface};
 
 /// Blocks manager for downloading and processing matching blocks.
 ///
@@ -87,55 +89,22 @@ impl<H: BlockHeaderStorage, B: BlockStorage, W: WalletInterface> BlocksManager<H
 
             // Process the block only for the wallets whose filter matched it.
             // Already-synced wallets that did not match are not touched.
-            let mut wallet = self.wallet.write().await;
-            let result =
-                wallet.process_block_for_wallets(block.block(), hash, height, &interested).await;
-            drop(wallet);
-
-            let total_relevant = result.relevant_tx_count();
-            let new_scripts_total: usize = result.new_scripts.values().map(|v| v.len()).sum();
-            if total_relevant > 0 {
-                tracing::info!(
-                    "Found {} relevant transactions ({} new, {} existing) {} at height {}, new scripts: {}",
-                    total_relevant,
-                    result.new_txids.len(),
-                    result.existing_txids.len(),
-                    hash,
-                    height,
-                    new_scripts_total
-                );
-            }
+            let result = self.apply_block(&block, height, &interested).await;
 
             // Collect confirmed txids before moving new_scripts out of result
             let confirmed_txids: Vec<_> = result.relevant_txids().cloned().collect();
-
-            // Collect new scripts for gap limit rescanning
-            let new_scripts = result.new_scripts;
-            if new_scripts_total > 0 {
-                tracing::debug!(
-                    "Block {} generated {} new scripts for gap limit maintenance across {} wallets",
-                    height,
-                    new_scripts_total,
-                    new_scripts.len()
-                );
-            }
-
-            self.progress.add_processed(1);
-            if total_relevant > 0 {
-                self.progress.add_relevant(1);
-            }
-            // Only count new transactions to avoid double-counting during rescans
-            self.progress.add_transactions(result.new_txids.len() as u32);
             self.progress.update_last_processed(height);
             last_applied = Some(height);
 
             events.push(SyncEvent::BlockProcessed {
                 block_hash: hash,
                 height,
                 wallets: interested,
-                new_scripts,
+                new_scripts: result.new_scripts,
                 confirmed_txids,
             });
+
+            self.reapply_blocks(result.reapply_heights).await?;
         }
 
         // Blocks are drained in strict height order, so `last_applied` is the
@@ -169,6 +138,71 @@ impl<H: BlockHeaderStorage, B: BlockStorage, W: WalletInterface> BlocksManager<H
 
         Ok(events)
     }
+
+    async fn apply_block(
+        &mut self,
+        block: &HashedBlock,
+        height: u32,
+        wallets: &BTreeSet<WalletId>,
+    ) -> BlockProcessingResult {
+        let hash = *block.hash();
+        let mut wallet = self.wallet.write().await;
+        let result = wallet.process_block_for_wallets(block.block(), hash, height, wallets).await;
+        drop(wallet);
+
+        let total_relevant = result.relevant_tx_count();
+        let new_scripts_total: usize = result.new_scripts.values().map(|v| v.len()).sum();
+        if total_relevant > 0 {
+            tracing::info!(
+                "Found {} relevant transactions ({} new, {} existing) {} at height {}, new scripts: {}",
+                total_relevant,
+                result.new_txids.len(),
+                result.existing_txids.len(),
+                hash,
+                height,
+                new_scripts_total
+            );
+        }
+        if new_scripts_total > 0 {
+            tracing::debug!(
+                "Block {} generated {} new scripts for gap limit maintenance across {} wallets",
+                height,
+                new_scripts_total,
+                result.new_scripts.len()
+            );
+        }
+
+        self.progress.add_processed(1);
+        if total_relevant > 0 {
+            self.progress.add_relevant(1);
+        }
+        // Only count new transactions to avoid double-counting during rescans
+        self.progress.add_transactions(result.new_txids.len() as u32);
+        result
+    }
+
+    async fn reapply_blocks(
+        &mut self,
+        reapply: BTreeMap<WalletId, BTreeSet<u32>>,
+    ) -> SyncResult<()> {
+        let mut queue: BTreeSet<(u32, WalletId)> = reapply
+            .into_iter()
+            .flat_map(|(wallet_id, heights)| heights.into_iter().map(move |h| (h, wallet_id)))
+            .collect();
+        while let Some((height, wallet_id)) = queue.pop_first() {
+            let Some(block) = self.block_storage.read().await.load_block(height).await? else {
+                tracing::warn!("Cannot re-apply block at height {}: not in storage", height);
+                continue;
+            };
+            let result = self.apply_block(&block, height, &BTreeSet::from([wallet_id])).await;
+            queue.extend(
+                result.reapply_heights.into_iter().flat_map(|(wallet_id, heights)| {
+                    heights.into_iter().map(move |h| (h, wallet_id))
+                }),
+            );
+        }
+        Ok(())
+    }
 }
 
 impl<H: BlockHeaderStorage, B: BlockStorage, W: WalletInterface> std::fmt::Debug
@@ -353,6 +387,40 @@ mod tests {
         assert_eq!(processed[0].1, 100);
     }
 
+    #[tokio::test]
+    async fn test_process_buffered_blocks_reapplies_requested_stored_block() {
+        let storage = DiskStorageManager::with_temp_dir().await.unwrap();
+        let mut wallet = MockWallet::new();
+        wallet.set_reapply_heights(100, BTreeSet::from([200]));
+        let wallet = Arc::new(RwLock::new(wallet));
+        let mut manager: TestBlocksManager =
+            BlocksManager::new(wallet.clone(), storage.block_headers(), storage.blocks()).await;
+        manager.progress.set_state(SyncState::Syncing);
+
+        manager
+            .block_storage
+            .write()
+            .await
+            .store_block(200, HashedBlock::dummy(200, vec![]))
+            .await
+            .unwrap();
+        manager.pipeline.add_from_storage(
+            HashedBlock::dummy(100, vec![]),
+            100,
+            BTreeSet::from([MOCK_WALLET_ID]),
+        );
+
+        let events = manager.process_buffered_blocks().await.unwrap();
+        assert_eq!(
+            events.iter().filter(|e| matches!(e, SyncEvent::BlockProcessed { .. })).count(),
+            1
+        );
+
+        let processed = wallet.read().await.processed_blocks();
+        let heights: Vec<u32> = processed.lock().await.iter().map(|(_, h)| *h).collect();
+        assert_eq!(heights, vec![100, 200]);
+    }
+
     /// A wallet that is NOT in the pipeline's interested set must not be
     /// routed the block. Two wallets are registered, but only `wallet_in`
     /// appears in the routed set; the other wallet's processed log must

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

@@ -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
@@ -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>>,
 }
@@ -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(),
         }
     }
@@ -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,
@@ -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>) {

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

@@ -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()