Skip to content

feat: per-wallet filter scan and runtime wallet catch-up #118

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
xdustinface wants to merge 2 commits into from
Closed

feat: per-wallet filter scan and runtime wallet catch-up #118

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a11b1b8
feat: per-wallet filter scan and block processing
xdustinface Apr 25, 2026
5603c6a
feat: rescan filters when a wallet falls behind committed height
xdustinface Apr 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions dash-spv-ffi/src/callbacks.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand All @@ -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,
Expand Down
23 changes: 13 additions & 10 deletions dash-spv/src/sync/blocks/manager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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: {}",
Expand All @@ -96,19 +98,20 @@ 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()
);
}
Expand Down Expand Up @@ -170,7 +173,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>;
Expand Down Expand Up @@ -215,8 +218,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,
};
Expand Down
77 changes: 49 additions & 28 deletions dash-spv/src/sync/blocks/pipeline.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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 {
Expand Down Expand Up @@ -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);
}
}

Expand Down Expand Up @@ -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
Expand All @@ -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);
}

Expand Down Expand Up @@ -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());
Expand All @@ -226,9 +246,9 @@ mod tests {
let block2 = make_test_block(2);

@manki-review manki-review Bot Apr 27, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 ✨ Suggestion: queue() wallet-merge behavior is untested

The updated queue() merges wallet sets when a block hash is already tracked (hash_to_wallets.entry(hash).or_default().extend(wallets)). This merge is important for the late-wallet catch-up case, but no test exercises it. A future refactor that accidentally switches or_default().extend() to an overwrite would go undetected.

Suggested fix
Suggested change
let block2 = make_test_block(2);
Add a test that calls `queue` twice for the same `FilterMatchKey` with disjoint wallet sets, then calls `take_next_ordered_block` and asserts the returned wallet set is the union of both.
AI context 
{
  "file": "dash-spv/src/sync/blocks/pipeline.rs",
  "line": 246,
  "severity": "suggestion",
  "confidence": "medium",
  "flaggedBy": [
    "Testing & Coverage"
  ],
  "title": "queue() wallet-merge behavior is untested",
  "fix": "Add a test that calls `queue` twice for the same `FilterMatchKey` with disjoint wallet sets, then calls `take_next_ordered_block` and asserts the returned wallet set is the union of both.",
  "reachability": "reachable"
}

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);
Expand All @@ -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);
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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);

Expand All @@ -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());

@manki-review manki-review Bot Apr 27, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔴 ⚠️ Warning: Pipeline wallet propagation has zero test coverage

Every pipeline unit test passes BTreeSet::new() as the wallet set, and all call sites of take_next_ordered_block() destructure the result as (block, height, _), silently discarding the returned wallet set. The central new behavior—that a block carries only the wallets whose filters matched it through the pipeline—is never verified at the unit level. A bug that drops or corrupts the wallet set would not be caught by these tests.

Suggested fix
Suggested change
Add a test that queues a block with a non-empty wallet set, simulates download completion, calls `take_next_ordered_block`, and asserts the returned wallet set equals the one that was queued. Also add a test for the merge path in `queue()`: queue the same block hash twice with different wallet sets and verify the returned set is the union.
AI context 
{
  "file": "dash-spv/src/sync/blocks/pipeline.rs",
  "line": 362,
  "severity": "warning",
  "confidence": "high",
  "flaggedBy": [
    "Testing & Coverage"
  ],
  "title": "Pipeline wallet propagation has zero test coverage",
  "fix": "Add a test that queues a block with a non-empty wallet set, simulates download completion, calls `take_next_ordered_block`, and asserts the returned wallet set equals the one that was queued. Also add",
  "reachability": "reachable"
}

// 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);

Expand All @@ -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());
Expand All @@ -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);
}

Expand All @@ -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);
}
Expand All @@ -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
Expand All @@ -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);

Expand Down
17 changes: 11 additions & 6 deletions dash-spv/src/sync/blocks/sync_manager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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() {
Expand All @@ -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(),

@manki-review manki-review Bot Apr 27, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔴 ✨ Suggestion: Unnecessary clone of wallets BTreeSet in block-loading loop

In the for (key, wallets) in blocks loop, wallets is cloned for both the storage path (add_from_storage) and the download path (to_download.push). Because each branch is mutually exclusive (the storage path continues and the download path is the last statement), ownership of wallets can be moved in both cases, eliminating both clones. This is in the block-loading hot path called for every BlocksNeeded event.

Suggested fix
Suggested change
key.height(),
// Storage path — move wallets:
self.pipeline.add_from_storage(hashed_block.block().clone(), key.height(), wallets);
self.progress.add_from_storage(1);
continue;
// Download path — move key and wallets:
to_download.push((key, wallets));
AI context 
{
  "file": "dash-spv/src/sync/blocks/sync_manager.rs",
  "line": 141,
  "severity": "suggestion",
  "confidence": "high",
  "flaggedBy": [
    "Performance & Efficiency"
  ],
  "title": "Unnecessary clone of wallets BTreeSet in block-loading loop",
  "fix": "// Storage path — move wallets:\nself.pipeline.add_from_storage(hashed_block.block().clone(), key.height(), wallets);\nself.progress.add_from_storage(1);\ncontinue;\n\n// Download path — move key and walle",
  "reachability": "reachable"
}

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

Expand Down
Loading
Loading