From e544cdf376f96fd0276e84549c1eb13b025dd55b Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Mon, 7 Sep 2026 10:28:14 -0700 Subject: [PATCH] fix(dash-spv): stop losing derived scripts, and close the loop on wallet state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two syncs of the same wallet against the same chain returned balances 2 000 020 sat apart, about half the time each. The money was one block, 2 429 667, which the runs that ended high never processed. Scripts derived while applying a block were handed to the batch named by the block's in-flight record. That record is keyed by block hash and consumed by the first delivery, and a block is delivered more than once — a rescan re-queues what the forward scan already handed over, and on a mainnet restore 2 904 of 3 039 relevant heights arrive twice or more. Every later delivery found no record, so its scripts reached no batch, no later batch, and no backward sweep. What made it invisible is that the derivation itself was not lost: the wallet kept the addresses. So no later block reported them as new, no rescan carried them, and the filter layer went on matching a query it did not know was incomplete. Measured: a second delivery of the block at 2 429 637 derived 27 scripts covering the mixing session that owns 2 429 667, the batch holding that block rescanned four times without them, and the block was never matched. `collect_new_scripts` now routes by height instead — to the batch whose range contains the block, which still holds that range's filters, falling back to the backward accumulator only when no active batch covers the height. That alone would still rest on a one-shot notification arriving, and 23.6% of blocks are applied out of order, so notification-shaped invariants are not worth much here. `reconcile_untested_scripts` therefore closes the loop on state: each batch records which scripts have been matched against its filters, and before committing it asks the wallet what it watches now and re-tests the difference. It also gives the lower active batches the scripts a higher one derived, which nothing did before. Ten full mainnet restores across five configurations now return the same 13 876 outputs and the same balance, where the same wallet previously split roughly 50/50 between two answers. Two of the first three runs exercised the routing path (92 and 27 scripts rescued), so the agreement is not luck. Two consequences of the new routing, both handled here. `rescan_batch` marks scripts tested before its empty-filters return, as `scan_batch` already did: otherwise a batch with no filters is handed the same set by every commit attempt and never converges. And `backward_scripts` can now be non-empty with no active batch, when a block is delivered after its batch committed — so the assertion in `try_process_batch` that it is empty no longer holds. Those scripts cannot be left to a next commit that may never come: the accumulator is in-memory only, nothing looks below the committed frontier again, and a shutdown at the tip loses them for good, since the restart resumes with `committed_height` already at the tip and no batch to reconcile them against. The completion branch therefore sweeps them itself over the whole committed range, and holds `FiltersSyncComplete` while the blocks that sweep found are still in flight. The gate is the tracker, not the commit gate: blocks a tip sweep queues are charged to no batch, so no batch can hold the completion for them. Their `BlockProcessed` re-enters the branch, and a round that derives no new scripts is the fixpoint. Processed records left by blocks applied after the last commit are pruned there too, since no commit will. Four regression tests, each failing without the fix: a `BlockProcessed` with no in-flight record whose scripts must reach the batch covering the height (and the backward accumulator when none does); a batch scanned with a query missing one address, whose commit must find that address's block without ever being told; a rescan of an empty batch, which must still record what it was handed; and a tip with scripts stranded in the accumulator, which must sweep them and withhold completion until the block it finds has been applied. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01K92DcuiKs8UdWkrhyfghqX --- dash-spv/src/sync/filters/batch.rs | 22 ++ .../src/sync/filters/block_match_tracker.rs | 5 + dash-spv/src/sync/filters/manager.rs | 342 +++++++++++++++++- dash-spv/src/sync/filters/sync_manager.rs | 17 +- 4 files changed, 364 insertions(+), 22 deletions(-) diff --git a/dash-spv/src/sync/filters/batch.rs b/dash-spv/src/sync/filters/batch.rs index b389c5a1b..3fd609b79 100644 --- a/dash-spv/src/sync/filters/batch.rs +++ b/dash-spv/src/sync/filters/batch.rs @@ -37,6 +37,8 @@ pub(super) struct FiltersBatch { /// need rescan, attributed per wallet so we can rerun matching only /// against the wallet that produced each new script. collected_scripts: HashMap>, + /// Every script already matched against this batch's filters, per wallet. + tested_scripts: HashMap>, } impl FiltersBatch { @@ -56,6 +58,7 @@ impl FiltersBatch { rescan_complete: false, scanned_wallets: BTreeMap::new(), collected_scripts: HashMap::new(), + tested_scripts: HashMap::new(), } } /// Start height of this batch (inclusive). @@ -119,6 +122,25 @@ impl FiltersBatch { ) { 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>( + &mut self, + wallet_id: WalletId, + scripts: I, + ) { + self.tested_scripts.entry(wallet_id).or_default().extend(scripts); + } + + /// The wallet's scripts that this batch has never been matched against. + pub(super) fn untested<'a>( + &'a self, + wallet_id: &WalletId, + monitored: &'a [ScriptBuf], + ) -> impl Iterator { + let tested = self.tested_scripts.get(wallet_id); + 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> { std::mem::take(&mut self.collected_scripts) diff --git a/dash-spv/src/sync/filters/block_match_tracker.rs b/dash-spv/src/sync/filters/block_match_tracker.rs index e0d911b47..eb81d5344 100644 --- a/dash-spv/src/sync/filters/block_match_tracker.rs +++ b/dash-spv/src/sync/filters/block_match_tracker.rs @@ -148,6 +148,11 @@ 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() diff --git a/dash-spv/src/sync/filters/manager.rs b/dash-spv/src/sync/filters/manager.rs index 9e7834e48..f5e0cdf05 100644 --- a/dash-spv/src/sync/filters/manager.rs +++ b/dash-spv/src/sync/filters/manager.rs @@ -566,18 +566,29 @@ impl= self.progress.filter_header_tip_height() && self.progress.committed_height() >= self.progress.target_height() { - // Every commit goes through `try_commit_batches`, where the last - // batch — examined when it alone remains and its end has reached - // the filter-header tip — either sweeps the accumulated backward - // scripts or stays uncommitted on the blocks its sweep found. A - // batch whose end is still below the tip commits without the - // sweep, but then `committed_height` is below the tip too and - // this branch is not taken. So no newly derived script is still - // waiting on its committed-range test here. - debug_assert!( - self.backward_scripts.is_empty(), - "backward scripts pending with no active batches" - ); + // A block delivered after its batch committed derives scripts with + // no active batch to route them to, so they land in the accumulator + // instead. It is in-memory only and nothing looks below the + // committed frontier again, so sweep here rather than wait for a + // next commit that may never come. + if !self.backward_scripts.is_empty() { + let backward_scripts = std::mem::take(&mut self.backward_scripts); + let sweep_start = self.progress.committed_height().saturating_add(1); + events.extend(self.rescan_committed_range(sweep_start, &backward_scripts).await?); + } + + // Blocks that sweep found are charged to no batch, so the commit + // gate cannot hold completion — gate on the tracker. Their + // `BlockProcessed` re-enters here, and a round deriving no new + // scripts is the fixpoint. + if self.tracker.has_blocks_in_flight() { + return Ok(events); + } + + // Blocks applied after the last commit leave processed records that + // no commit will prune. + self.tracker.prune_at_or_below(self.progress.committed_height()); + if self.state() == SyncState::Syncing { self.set_state(SyncState::Synced); } @@ -653,6 +664,14 @@ impl 0 { + // Reconciliation found blocks; converge before committing. + break; + } + } + // The backward sweep waits for the forward pipeline to drain: // it runs only when this batch is the last one active and no // lookahead batch can be created past it. Commits before that @@ -837,6 +856,90 @@ impl>, + ) -> usize { + let target = self + .active_batches + .range(..=height) + .next_back() + .filter(|(_, batch)| batch.end_height() >= height) + .map(|(&start, _)| start); + + let mut routed = 0; + for (wallet_id, scripts) in new_scripts { + if scripts.is_empty() { + continue; + } + routed += scripts.len(); + match target.and_then(|start| self.active_batches.get_mut(&start)) { + Some(batch) => batch.add_scripts_for_wallet(*wallet_id, scripts.iter().cloned()), + // Its range has committed: only the sweep can still test these. + None => self + .backward_scripts + .entry(*wallet_id) + .or_default() + .extend(scripts.iter().cloned()), + } + } + routed + } + + /// Re-test anything the wallet watches that this batch has never been + /// matched against, before letting it commit. Closing the loop on wallet + /// state does not depend on a `new_scripts` notification arriving; a + /// script the wallet has dropped from `scan_script_pubkeys_for` is still + /// not re-tested. + async fn reconcile_untested_scripts(&mut self, batch_start: u32) -> SyncResult> { + let Some(batch) = self.active_batches.get(&batch_start) else { + return Ok(vec![]); + }; + let wallets: Vec = batch.scanned_wallets().keys().copied().collect(); + if wallets.is_empty() { + return Ok(vec![]); + } + + let mut untested: HashMap> = HashMap::new(); + { + let wallet = self.wallet.read().await; + let Some(batch) = self.active_batches.get(&batch_start) else { + return Ok(vec![]); + }; + for wallet_id in &wallets { + let monitored = wallet.scan_script_pubkeys_for(wallet_id); + let missing: HashSet = + batch.untested(wallet_id, &monitored).cloned().collect(); + if !missing.is_empty() { + untested.insert(*wallet_id, missing); + } + } + } + if untested.is_empty() { + return Ok(vec![]); + } + + tracing::debug!( + "Reconcile batch {}: {} script(s) the wallet watches had never been matched here", + batch_start, + untested.values().map(HashSet::len).sum::(), + ); + // Same accounting as any other newly derived script. + let events = self.rescan_batch(batch_start, &untested).await?; + for (wallet_id, scripts) in untested { + self.backward_scripts.entry(wallet_id).or_default().extend(scripts); + } + Ok(events) + } + /// Rescan a specific batch for newly discovered scriptPubKeys, attributed /// per wallet so each new script is matched only against the filters /// relevant to its owning wallet. @@ -860,6 +963,17 @@ impl blocks.contains_key(&key), + _ => false, + }); + assert!(queued, "reconciliation must match the untested address before committing"); + assert!( + manager.active_batches.contains_key(&0), + "the batch must wait for the block it just found" + ); + } + + /// Proves the tip is not reported synced while scripts sit in the backward + /// accumulator with no batch left to sweep them: the sweep runs here, and + /// completion waits for the block it finds. + #[tokio::test] + async fn test_tip_sweeps_stranded_backward_scripts_before_completing() { + let wallet_id: WalletId = [3; 32]; + let watched = Address::dummy(Network::Testnet, 31); + + let mut multi = MultiMockWallet::new(); + multi.insert_wallet( + wallet_id, + MockWalletState { + addresses: vec![watched.clone()], + synced_height: 9, + last_processed_height: 9, + account_generation: 0, + }, + ); + let mut manager = create_multi_test_manager(Arc::new(RwLock::new(multi))).await; + manager.set_state(SyncState::Syncing); + + // Every height at or below the committed frontier has its header and + // filter persisted; only height 4 pays `watched`. + let paying = Block::dummy(4, vec![Transaction::dummy(&watched, 0..0, &[4])]); + let paying_filter = BlockFilter::dummy(&paying); + let key = FilterMatchKey::new(4, paying.block_hash()); + { + let mut header_storage = manager.header_storage.write().await; + let mut filter_storage = manager.filter_storage.write().await; + for height in 0..=9u32 { + let (header, bytes) = if height == 4 { + (paying.header, paying_filter.content.clone()) + } else { + let filler = Block::dummy(height, vec![]); + (filler.header, BlockFilter::dummy(&filler).content) + }; + header_storage.store_headers_at_height(&[header.into()], height).await.unwrap(); + filter_storage.store_filter(height, &bytes).await.unwrap(); + } + } + + // At the tip with nothing active: the last commit left + // `processing_height` past the tip, so no lookahead batch is created + // and the completion branch is reached. + manager.processing_height = 10; + manager.progress.update_stored_height(9); + manager.progress.update_committed_height(9); + manager.progress.update_filter_header_tip_height(9); + manager.progress.update_target_height(9); + manager.backward_scripts.entry(wallet_id).or_default().insert(watched.script_pubkey()); + + let events = manager.try_process_batch().await.unwrap(); + assert!( + events.iter().any(|e| match e { + SyncEvent::BlocksNeeded { + blocks, + } => blocks.contains_key(&key), + _ => false, + }), + "the tip sweep must find the block paying the stranded script" + ); + assert!( + !events.iter().any(|e| matches!(e, SyncEvent::FiltersSyncComplete { .. })), + "completion must wait for that block to be applied" + ); + assert!(manager.backward_scripts.is_empty(), "the sweep consumes the accumulator"); + + let (tx, _rx) = unbounded_channel(); + let requests = RequestSender::new(tx); + let processed = SyncEvent::BlockProcessed { + block_hash: paying.block_hash(), + height: 4, + wallets: BTreeSet::from([wallet_id]), + new_scripts: BTreeMap::new(), + confirmed_txids: vec![], + }; + let events = manager.handle_sync_event(&processed, &requests).await.unwrap(); + assert!( + events.iter().any(|e| matches!(e, SyncEvent::FiltersSyncComplete { .. })), + "completion follows once the sweep's block is applied" + ); + } + #[tokio::test] async fn test_start_download_waits_when_filter_headers_insufficient() { let mut manager = create_test_manager().await; diff --git a/dash-spv/src/sync/filters/sync_manager.rs b/dash-spv/src/sync/filters/sync_manager.rs index 2dbd8ad5e..8e763b8b3 100644 --- a/dash-spv/src/sync/filters/sync_manager.rs +++ b/dash-spv/src/sync/filters/sync_manager.rs @@ -183,7 +183,8 @@ impl< 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) { + let in_flight = self.tracker.finish_in_flight(block_hash); + if let Some((_, batch_start)) = in_flight { if let Some(batch) = self.active_batches.get_mut(&batch_start) { batch.decrement_pending_blocks(); tracing::debug!( @@ -194,17 +195,13 @@ impl< batch.pending_blocks() ); } + } - // Collect per-wallet new scripts for deferred rescan at commit time. - for (wallet_id, scripts) in new_scripts { - if scripts.is_empty() { - continue; - } - if let Some(batch) = self.active_batches.get_mut(&batch_start) { - batch.add_scripts_for_wallet(*wallet_id, scripts.iter().cloned()); - } - } + // Outside the in-flight arm on purpose: that record is consumed + // by the first delivery, and a block is delivered more than once. + let derived = self.collect_new_scripts(*height, new_scripts); + if in_flight.is_some() || derived > 0 { return self.try_process_batch().await; } }