fix(dash-spv): stop losing derived scripts, and close the loop on wallet state

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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K92DcuiKs8UdWkrhyfghqX

Author: 2 people

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<WalletId, HashSet<ScriptBuf>>,
+    /// Every script already matched against this batch's filters, per wallet.
+    tested_scripts: HashMap<WalletId, HashSet<ScriptBuf>>,
 }
 
 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<I: IntoIterator<Item = ScriptBuf>>(
+        &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<Item = &'a ScriptBuf> {
+        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<WalletId, HashSet<ScriptBuf>> {
         std::mem::take(&mut self.collected_scripts)

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