feat: per-wallet filter scan and block processing

When a wallet is added behind the current scan progress, only that wallet now goes through filter matching and block processing. Already-synced wallets are completely untouched: their addresses are not tested against any filter, no matched block is processed for them, and their per-wallet `synced_height` and `last_processed_height` stay where they were.

`WalletInterface` is restructured around per-wallet operations:

- `process_block_for_wallets(block, height, wallets)` replaces the global `process_block` and only updates the listed wallets.
- `wallets_behind(height)` returns the wallet ids that still need filter coverage at `height`.
- `monitored_addresses_for(wallet_id)` and `wallet_synced_height(wallet_id)` give per-wallet projections for filter matching.
- `update_wallet_synced_height(wallet_id, height)` and `update_wallet_last_processed_height(wallet_id, height)` advance one wallet at a time and are monotonic, so a rescan that lowers the aggregate min never regresses an already-synced wallet's per-wallet checkpoint.
- `BlockProcessingResult.new_addresses` and `CheckTransactionsResult.new_addresses` carry gap-limit discoveries with wallet attribution so re-matching is also per-wallet.

`FiltersManager.scan_batch` now reads each behind wallet's `(synced_height, addresses)`, projects the batch's filters to heights that wallet hasn't yet covered, and matches them only against that wallet's address set. The per-block result is a `BTreeMap<FilterMatchKey, BTreeSet<WalletId>>` that flows through the renamed `BlocksNeeded` payload to `BlocksManager`. `FiltersBatch` records the wallet set scanned at scan time so the commit phase advances only those wallets' `synced_height`. Block-collision in `blocks_remaining` merges wallet sets when the block is still in flight, and re-queues via `BlocksNeeded` when the block has already been processed for an earlier wallet set so the late-added wallet still gets its turn from local block storage.

`BlocksManager` queues blocks with their wallet sets, threads them through the pipeline, and calls `process_block_for_wallets` with that exact set rather than every wallet in the manager. `BlockProcessed` carries `new_addresses` so `FiltersManager` can route gap-limit re-matches per wallet via `rescan_batch`.

`process_block_for_wallets` always refreshes the cached balance after applying a block's transactions, even when the block height is below a wallet's current `last_processed_height` (a rescan path), because UTXOs may have been added or removed.

Closes [#114](xdustinface#114).

Author: xdustinface

dash-spv-ffi/src/callbacks.rs

@@ -347,7 +347,7 @@ impl FFISyncEventCallbacks {
             } => {
                 if let Some(cb) = self.on_blocks_needed {
                     let ffi_blocks: Vec<FFIBlockNeeded> = blocks
-                        .iter()
+                        .keys()
                         .map(|key| FFIBlockNeeded {
                             height: key.height(),
                             hash: *key.hash().as_byte_array(),
@@ -366,10 +366,11 @@ impl FFISyncEventCallbacks {
                     let hash_bytes = block_hash.as_byte_array();
                     let txid_bytes: Vec<[u8; 32]> =
                         confirmed_txids.iter().map(|txid| *txid.as_byte_array()).collect();
+                    let total_new_addresses: usize = new_addresses.values().map(|v| v.len()).sum();
                     cb(
                         *height,
                         hash_bytes as *const [u8; 32],
-                        new_addresses.len() as u32,
+                        total_new_addresses as u32,
                         txid_bytes.as_ptr(),
                         txid_bytes.len() as u32,
                         self.user_data,

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

@@ -79,15 +79,17 @@ impl<H: BlockHeaderStorage, B: BlockStorage, W: WalletInterface> BlocksManager<H
         let mut events = Vec::new();
 
         // Process blocks in height order using pipeline's ordering logic
-        while let Some((block, height)) = self.pipeline.take_next_ordered_block() {
+        while let Some((block, height, interested)) = self.pipeline.take_next_ordered_block() {
             let hash = block.block_hash();
 
-            // Process block through wallet
+            // 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(&block, height).await;
+            let result = wallet.process_block_for_wallets(&block, height, &interested).await;
             drop(wallet);
 
             let total_relevant = result.relevant_tx_count();
+            let new_addresses_total: usize = result.new_addresses.values().map(|v| v.len()).sum();
             if total_relevant > 0 {
                 tracing::info!(
                     "Found {} relevant transactions ({} new, {} existing) {} at height {}, new addresses: {}",
@@ -96,19 +98,18 @@ impl<H: BlockHeaderStorage, B: BlockStorage, W: WalletInterface> BlocksManager<H
                     result.existing_txids.len(),
                     hash,
                     height,
-                    result.new_addresses.len()
+                    new_addresses_total
                 );
             }
 
-            // Collect confirmed txids before moving new_addresses out of result
             let confirmed_txids: Vec<_> = result.relevant_txids().cloned().collect();
 
-            // Collect new addresses for gap limit rescanning
-            let new_addresses: Vec<_> = result.new_addresses.into_iter().collect();
-            if !new_addresses.is_empty() {
+            let new_addresses = result.new_addresses;
+            if new_addresses_total > 0 {
                 tracing::debug!(
-                    "Block {} generated {} new addresses for gap limit maintenance",
+                    "Block {} generated {} new addresses for gap limit maintenance across {} wallets",
                     height,
+                    new_addresses_total,
                     new_addresses.len()
                 );
             }
@@ -170,7 +171,7 @@ mod tests {
     use crate::test_utils::MockNetworkManager;
     use key_wallet_manager::test_utils::MockWallet;
     use key_wallet_manager::FilterMatchKey;
-    use std::collections::BTreeSet;
+    use std::collections::{BTreeMap, BTreeSet};
 
     type TestBlocksManager =
         BlocksManager<PersistentBlockHeaderStorage, PersistentBlockStorage, MockWallet>;
@@ -215,8 +216,8 @@ mod tests {
         let requests = network.request_sender();
 
         let block_hash = dashcore::BlockHash::dummy(0);
-        let mut blocks = BTreeSet::new();
-        blocks.insert(FilterMatchKey::new(100, block_hash));
+        let mut blocks = BTreeMap::new();
+        blocks.insert(FilterMatchKey::new(100, block_hash), BTreeSet::new());
         let event = SyncEvent::BlocksNeeded {
             blocks,
         };

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

@@ -11,7 +11,7 @@ use crate::network::RequestSender;
 use crate::sync::download_coordinator::{DownloadConfig, DownloadCoordinator};
 use dashcore::blockdata::block::Block;
 use dashcore::BlockHash;
-use key_wallet_manager::FilterMatchKey;
+use key_wallet_manager::{FilterMatchKey, WalletId};
 
 /// Maximum number of concurrent block downloads.
 const MAX_CONCURRENT_BLOCK_DOWNLOADS: usize = 20;
@@ -36,6 +36,9 @@ pub(super) struct BlocksPipeline {
     downloaded: BTreeMap<u32, Block>,
     /// Map hash -> height for looking up height when block arrives.
     hash_to_height: HashMap<BlockHash, u32>,
+    /// Per-block interested wallets, populated when the block is queued.
+    /// Only those wallets get the block processed.
+    hash_to_wallets: HashMap<BlockHash, BTreeSet<WalletId>>,
 }
 
 impl std::fmt::Debug for BlocksPipeline {
@@ -66,17 +69,26 @@ impl BlocksPipeline {
             pending_heights: BTreeSet::new(),
             downloaded: BTreeMap::new(),
             hash_to_height: HashMap::new(),
+            hash_to_wallets: HashMap::new(),
         }
     }
 
-    /// Queue blocks with their heights for download.
+    /// Queue blocks with their heights and per-block interested wallet sets.
     ///
-    /// This is the preferred method as it enables height-ordered processing.
-    pub(super) fn queue(&mut self, blocks: impl IntoIterator<Item = FilterMatchKey>) {
-        for key in blocks {
-            self.coordinator.enqueue([*key.hash()]);
+    /// Each entry's wallet set is the union of wallets whose addresses matched
+    /// the filter for that block. If the block is already queued we merge the
+    /// new wallet ids into the existing set so a late-discovered wallet still
+    /// gets the block processed when it arrives.
+    pub(super) fn queue(
+        &mut self,
+        blocks: impl IntoIterator<Item = (FilterMatchKey, BTreeSet<WalletId>)>,
+    ) {
+        for (key, wallets) in blocks {
+            let hash = *key.hash();
+            self.coordinator.enqueue([hash]);
             self.pending_heights.insert(key.height());
-            self.hash_to_height.insert(*key.hash(), key.height());
+            self.hash_to_height.insert(hash, key.height());
+            self.hash_to_wallets.entry(hash).or_default().extend(wallets);
         }
     }
 
@@ -141,12 +153,13 @@ impl BlocksPipeline {
         true
     }
 
-    /// Take the next block that's safe to process in height order.
+    /// Take the next block that's safe to process in height order, along with
+    /// the wallet set whose filters matched this block.
     ///
     /// Returns None if:
     /// - No downloaded blocks available, or
     /// - Waiting for a lower-height block still pending
-    pub(super) fn take_next_ordered_block(&mut self) -> Option<(Block, u32)> {
+    pub(super) fn take_next_ordered_block(&mut self) -> Option<(Block, u32, BTreeSet<WalletId>)> {
         let lowest_downloaded = *self.downloaded.keys().next()?;
 
         // Check if any pending blocks have lower heights
@@ -156,15 +169,22 @@ impl BlocksPipeline {
             }
         }
 
-        // Safe to return this block
         let block = self.downloaded.remove(&lowest_downloaded).unwrap();
-        Some((block, lowest_downloaded))
+        let wallets = self.hash_to_wallets.remove(&block.block_hash()).unwrap_or_default();
+        Some((block, lowest_downloaded, wallets))
     }
 
     /// Add a block that was loaded from storage (skip download).
     ///
     /// Used when blocks are already persisted from a previous sync.
-    pub(super) fn add_from_storage(&mut self, block: Block, height: u32) {
+    pub(super) fn add_from_storage(
+        &mut self,
+        block: Block,
+        height: u32,
+        wallets: BTreeSet<WalletId>,
+    ) {
+        let hash = block.block_hash();
+        self.hash_to_wallets.entry(hash).or_default().extend(wallets);
         self.downloaded.insert(height, block);
     }
 
@@ -212,7 +232,7 @@ mod tests {
     fn test_queue_block() {
         let mut pipeline = BlocksPipeline::new();
         let block = make_test_block(1);
-        pipeline.queue([FilterMatchKey::new(100, block.block_hash())]);
+        pipeline.queue([(FilterMatchKey::new(100, block.block_hash()), BTreeSet::new())]);
 
         assert_eq!(pipeline.coordinator.pending_count(), 1);
         assert!(!pipeline.is_complete());
@@ -226,9 +246,9 @@ mod tests {
         let block2 = make_test_block(2);
         let block3 = make_test_block(3);
         pipeline.queue([
-            FilterMatchKey::new(100, block1.block_hash()),
-            FilterMatchKey::new(101, block2.block_hash()),
-            FilterMatchKey::new(102, block3.block_hash()),
+            (FilterMatchKey::new(100, block1.block_hash()), BTreeSet::new()),
+            (FilterMatchKey::new(101, block2.block_hash()), BTreeSet::new()),
+            (FilterMatchKey::new(102, block3.block_hash()), BTreeSet::new()),
         ]);
 
         assert_eq!(pipeline.coordinator.pending_count(), 3);
@@ -245,7 +265,7 @@ mod tests {
         let hash = block.block_hash();
 
         // Queue with height tracking
-        pipeline.queue([FilterMatchKey::new(100, block.block_hash())]);
+        pipeline.queue([(FilterMatchKey::new(100, block.block_hash()), BTreeSet::new())]);
 
         // Simulate sending via coordinator
         let hashes = pipeline.coordinator.take_pending(1);
@@ -276,7 +296,7 @@ mod tests {
         // Queue more blocks than max concurrent
         for i in 0..=MAX_CONCURRENT_BLOCK_DOWNLOADS {
             let block = make_test_block(i as u8);
-            pipeline.queue([FilterMatchKey::new(i as u32, block.block_hash())]);
+            pipeline.queue([(FilterMatchKey::new(i as u32, block.block_hash()), BTreeSet::new())]);
         }
 
         // Take and mark as downloading up to limit
@@ -301,6 +321,7 @@ mod tests {
             pending_heights: BTreeSet::new(),
             downloaded: BTreeMap::new(),
             hash_to_height: HashMap::new(),
+            hash_to_wallets: HashMap::new(),
         };
 
         // Use coordinator directly to set up in-flight state
@@ -328,7 +349,7 @@ mod tests {
 
         // Use add_from_storage to test ordering logic without network
         // Add block 2 first (out of order)
-        pipeline.add_from_storage(block2.clone(), 101);
+        pipeline.add_from_storage(block2.clone(), 101, BTreeSet::new());
         // Also track height 100 as pending to simulate waiting
         pipeline.pending_heights.insert(100);
 
@@ -337,15 +358,15 @@ mod tests {
 
         // Add block 1
         pipeline.pending_heights.remove(&100);
-        pipeline.add_from_storage(block1.clone(), 100);
+        pipeline.add_from_storage(block1.clone(), 100, BTreeSet::new());
 
         // Now block 1 is ready (lowest height)
-        let (block, height) = pipeline.take_next_ordered_block().unwrap();
+        let (block, height, _) = pipeline.take_next_ordered_block().unwrap();
         assert_eq!(height, 100);
         assert_eq!(block.block_hash(), hash1);
 
         // Block 2 is now ready
-        let (block, height) = pipeline.take_next_ordered_block().unwrap();
+        let (block, height, _) = pipeline.take_next_ordered_block().unwrap();
         assert_eq!(height, 101);
         assert_eq!(block.block_hash(), hash2);
 
@@ -360,7 +381,7 @@ mod tests {
 
         // Add block at height 101, but height 100 is still pending
         pipeline.pending_heights.insert(100);
-        pipeline.add_from_storage(block2.clone(), 101);
+        pipeline.add_from_storage(block2.clone(), 101, BTreeSet::new());
 
         // Cannot take block 2 - block at height 100 is still pending
         assert!(pipeline.take_next_ordered_block().is_none());
@@ -369,7 +390,7 @@ mod tests {
         pipeline.pending_heights.remove(&100);
 
         // Now block 2 is ready
-        let (_, height) = pipeline.take_next_ordered_block().unwrap();
+        let (_, height, _) = pipeline.take_next_ordered_block().unwrap();
         assert_eq!(height, 101);
     }
 
@@ -379,11 +400,11 @@ mod tests {
         let block = make_test_block(1);
         let hash = block.block_hash();
 
-        pipeline.add_from_storage(block.clone(), 100);
+        pipeline.add_from_storage(block.clone(), 100, BTreeSet::new());
 
         assert_eq!(pipeline.downloaded.len(), 1);
 
-        let (taken_block, height) = pipeline.take_next_ordered_block().unwrap();
+        let (taken_block, height, _) = pipeline.take_next_ordered_block().unwrap();
         assert_eq!(height, 100);
         assert_eq!(taken_block.block_hash(), hash);
     }
@@ -395,7 +416,7 @@ mod tests {
 
         // Adding to downloaded makes it incomplete
         let block = make_test_block(1);
-        pipeline.add_from_storage(block, 100);
+        pipeline.add_from_storage(block, 100, BTreeSet::new());
         assert!(!pipeline.is_complete());
 
         // Take the block
@@ -422,7 +443,7 @@ mod tests {
         let block = make_test_block(1);
 
         // Queue and mark as sent via coordinator
-        pipeline.queue([FilterMatchKey::new(100, block.block_hash())]);
+        pipeline.queue([(FilterMatchKey::new(100, block.block_hash()), BTreeSet::new())]);
         let hashes = pipeline.coordinator.take_pending(1);
         pipeline.coordinator.mark_sent(&hashes);
 

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

@@ -10,7 +10,8 @@ use crate::types::HashedBlock;
 use crate::SyncError;
 use async_trait::async_trait;
 use dashcore::network::message::NetworkMessage;
-use key_wallet_manager::WalletInterface;
+use key_wallet_manager::{FilterMatchKey, WalletId, WalletInterface};
+use std::collections::BTreeSet;
 
 #[async_trait]
 impl<H: BlockHeaderStorage, B: BlockStorage, W: WalletInterface + 'static> SyncManager
@@ -115,10 +116,10 @@ impl<H: BlockHeaderStorage, B: BlockStorage, W: WalletInterface + 'static> SyncM
 
             tracing::debug!("Blocks needed: {} blocks", blocks.len());
 
-            let mut to_download = Vec::new();
+            let mut to_download: Vec<(FilterMatchKey, BTreeSet<WalletId>)> = Vec::new();
 
             let block_storage = self.block_storage.read().await;
-            for key in blocks {
+            for (key, wallets) in blocks {
                 // Check if block is already stored (from previous sync)
                 if let Ok(Some(hashed_block)) = block_storage.load_block(key.height()).await {
                     if hashed_block.hash() != key.hash() {
@@ -135,13 +136,17 @@ impl<H: BlockHeaderStorage, B: BlockStorage, W: WalletInterface + 'static> SyncM
                         )));
                     }
                     // Block loaded from storage, add to pipeline for processing
-                    self.pipeline.add_from_storage(hashed_block.block().clone(), key.height());
+                    self.pipeline.add_from_storage(
+                        hashed_block.block().clone(),
+                        key.height(),
+                        wallets.clone(),
+                    );
                     self.progress.add_from_storage(1);
                     continue;
                 }
 
-                // Block not in storage, queue for download with height
-                to_download.push(key.clone());
+                // Block not in storage, queue for download with height + wallets
+                to_download.push((key.clone(), wallets.clone()));
             }
             drop(block_storage);