fix(key-wallet, dash-spv): re-emit late knowledge — record corrections, durable rescans, address-pool repair

dash-spv/src/client/lifecycle.rs

@@ -106,6 +106,12 @@ impl<W: WalletInterface, N: NetworkManager, S: StorageManager> DashSpvClient<W,
                     storage.filter_headers(),
                     storage.filters(),
                 )
+                .await
+                // Durable pending-sweep set: scripts derived mid-sync whose
+                // rescan cascade a crash interrupted are replayed next start
+                // instead of leaving their funded heights permanently
+                // untested (interrupted-restore fund loss, 2026-08-19).
+                .with_metadata(storage.metadata())
                 .await,
             );
             managers.blocks = Some(

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

@@ -43,6 +43,13 @@ pub(super) struct FiltersBatch {
     /// once, instead of the whole history being reloaded per derivation
     /// round.
     backward_scripts: HashMap<WalletId, HashSet<ScriptBuf>>,
+    /// Every script this batch has taken through its rescan cascade
+    /// (forward + backward), retained until commit. This is the receipt the
+    /// durable pending-sweep set is cleared against: only a COMMIT proves
+    /// the whole fixpoint (including blocks the sweep re-downloaded)
+    /// completed, so scripts must stay persisted until then — a crash
+    /// mid-cascade replays them on the next start.
+    retired_scripts: HashMap<WalletId, HashSet<ScriptBuf>>,
 }
 
 impl FiltersBatch {
@@ -63,6 +70,7 @@ impl FiltersBatch {
             scanned_wallets: BTreeMap::new(),
             collected_scripts: HashMap::new(),
             backward_scripts: HashMap::new(),
+            retired_scripts: HashMap::new(),
         }
     }
     /// Start height of this batch (inclusive).
@@ -127,8 +135,14 @@ impl FiltersBatch {
         self.collected_scripts.entry(wallet_id).or_default().extend(scripts);
     }
     /// Take collected per-wallet scripts for rescan, leaving the map empty.
+    /// The taken scripts are also recorded in [`Self::retired_scripts`] so
+    /// the commit can clear them from the durable pending-sweep set.
     pub(super) fn take_collected_scripts(&mut self) -> HashMap<WalletId, HashSet<ScriptBuf>> {
-        std::mem::take(&mut self.collected_scripts)
+        let taken = std::mem::take(&mut self.collected_scripts);
+        for (wallet_id, scripts) in &taken {
+            self.retired_scripts.entry(*wallet_id).or_default().extend(scripts.iter().cloned());
+        }
+        taken
     }
     /// Queue already forward-rescanned scripts for the deferred backward
     /// sweep over the committed range.
@@ -145,6 +159,11 @@ impl FiltersBatch {
     pub(super) fn take_backward_scripts(&mut self) -> HashMap<WalletId, HashSet<ScriptBuf>> {
         std::mem::take(&mut self.backward_scripts)
     }
+    /// Every script this batch carried through its rescan cascade — see the
+    /// field docs; read at commit to clear the durable pending-sweep set.
+    pub(super) fn retired_scripts(&self) -> &HashMap<WalletId, HashSet<ScriptBuf>> {
+        &self.retired_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/coinjoin_gap_discovery_tests.rs

@@ -411,3 +411,334 @@ async fn coinjoin_gap_limit_stall_across_committed_batch() {
          derives scripts mid-sync."
     );
 }
+
+/// Cross-committed-batch discovery survives a process death between script
+/// derivation and the backward sweep's completion (the durable pending-sweep
+/// set; interrupted-restore fund loss, 2026-08-19).
+///
+/// Same funding shape as [`coinjoin_gap_limit_stall_across_committed_batch`]:
+/// block A (height 10, batch 0..=99) pays beyond-window indices G+10..=G+21;
+/// block B (height 110, batch 100..=199) pays in-window indices 0..=29.
+/// Batch 0 scans clean and COMMITS. Processing block B derives the missing
+/// scripts — at which point the session "crashes": the manager is dropped
+/// before the backward sweep's re-downloaded block is processed. The
+/// in-memory cascade is gone, batch 0 is committed, and nothing in a
+/// restarted session would ever look below the committed boundary again.
+///
+/// The durable pending-sweep set closes this: the scripts were persisted to
+/// metadata storage the moment they entered the manager, a restarted manager
+/// (same storage, `with_metadata`) reloads them, seeds them into its lowest
+/// active batch, and the ordinary commit-time cascade re-tests the stored
+/// filters below the committed boundary — recovering block A's outputs.
+#[tokio::test]
+async fn interrupted_sweep_is_replayed_after_restart() {
+    // Shared across the "restart": wallet (its pools/synced heights persist
+    // in the real system via the SDK store) and disk storage (headers,
+    // filters, metadata).
+    let mut wm = WalletManager::<ManagedWalletInfo>::new(Network::Regtest);
+    let wallet_id = wm
+        .create_wallet_from_mnemonic(TEST_MNEMONIC, 0, WalletAccountCreationOptions::Default)
+        .expect("create deterministic test wallet");
+    let wallet = Arc::new(RwLock::new(wm));
+    let storage = DiskStorageManager::with_temp_dir().await.unwrap();
+
+    let addresses = coinjoin_external_addresses(&wallet, &wallet_id, (G + 22) as u32).await;
+    let (block_a, filter_a, key_a) = block_paying(10, &addresses[(G + 10)..=(G + 21)]);
+    let (block_b, filter_b, key_b) = block_paying(110, &addresses[0..=29]);
+    let blocks: HashMap<BlockHash, Block> =
+        HashMap::from([(block_a.block_hash(), block_a.clone()), (block_b.block_hash(), block_b)]);
+
+    // Persist headers+filters for the to-be-committed range (see the
+    // invariant note in the cross-committed-batch test).
+    {
+        let headers_arc = storage.block_headers();
+        let filters_arc = storage.filters();
+        let mut header_storage = headers_arc.write().await;
+        let mut filter_storage = filters_arc.write().await;
+        for height in 0..=99u32 {
+            let (header, filter_bytes) = if height == 10 {
+                (block_a.header, filter_a.content.clone())
+            } else {
+                let filler = Block::dummy(height, vec![]);
+                let filter = BlockFilter::dummy(&filler);
+                (filler.header, filter.content)
+            };
+            header_storage
+                .store_headers_at_height(&[header.into()], height)
+                .await
+                .expect("seed header");
+            filter_storage.store_filter(height, &filter_bytes).await.expect("seed filter");
+        }
+    }
+
+    // ── Session 1: batch 0 commits clean, block B derives scripts, CRASH ──
+    {
+        let mut manager = FiltersManager::new(
+            Arc::clone(&wallet),
+            storage.block_headers(),
+            storage.filter_headers(),
+            storage.filters(),
+        )
+        .await
+        .with_metadata(storage.metadata())
+        .await;
+        manager.set_state(SyncState::Syncing);
+
+        let mut batch_0 = FiltersBatch::new(0, 99, HashMap::from([(key_a, filter_a.clone())]));
+        batch_0.mark_verified();
+        manager.active_batches.insert(0, batch_0);
+        let mut batch_1 =
+            FiltersBatch::new(100, 199, HashMap::from([(key_b.clone(), filter_b.clone())]));
+        batch_1.mark_verified();
+        manager.active_batches.insert(100, batch_1);
+        manager.progress.update_stored_height(199);
+
+        // Initial pass: batch 0 matches nothing and commits; batch 1
+        // requests block B.
+        let initial_events = manager.try_process_batch().await.unwrap();
+        let needed: Vec<(u32, BlockHash, BTreeSet<WalletId>)> = initial_events
+            .iter()
+            .filter_map(|e| match e {
+                SyncEvent::BlocksNeeded {
+                    blocks: needed,
+                } => Some(needed.iter().map(|(k, w)| (k.height(), *k.hash(), w.clone()))),
+                _ => None,
+            })
+            .flatten()
+            .collect();
+        assert_eq!(needed.len(), 1, "only block B should be requested initially");
+
+        // Process block B once — derives the beyond-window scripts, which
+        // the manager must persist durably at this exact moment.
+        let (tx, _rx) = unbounded_channel();
+        let requests = RequestSender::new(tx);
+        for (height, block_hash, wallets) in needed {
+            let block = blocks.get(&block_hash).expect("known block");
+            let result = wallet
+                .write()
+                .await
+                .process_block_for_wallets(block, block_hash, height, &wallets)
+                .await;
+            let confirmed_txids = result.relevant_txids().cloned().collect();
+            let event = SyncEvent::BlockProcessed {
+                block_hash,
+                height,
+                wallets,
+                new_scripts: result.new_scripts,
+                confirmed_txids,
+            };
+            // The returned events (the sweep's own BlocksNeeded for block A
+            // among them) are deliberately DROPPED: the process dies here.
+            let _ = manager.handle_sync_event(&event, &requests).await.expect("BlockProcessed");
+        }
+        // manager dropped — in-memory cascade gone.
+    }
+
+    // ── Session 2: fresh manager over the same storage ──
+    let mut manager = FiltersManager::new(
+        Arc::clone(&wallet),
+        storage.block_headers(),
+        storage.filter_headers(),
+        storage.filters(),
+    )
+    .await
+    .with_metadata(storage.metadata())
+    .await;
+    manager.set_state(SyncState::Syncing);
+
+    // Production resume recreates batches from the committed frontier up;
+    // batch 0 is committed (synced_height advanced), so only batch 1 exists.
+    let mut batch_1 = FiltersBatch::new(100, 199, HashMap::from([(key_b, filter_b)]));
+    batch_1.mark_verified();
+    manager.active_batches.insert(100, batch_1);
+    manager.progress.update_stored_height(199);
+
+    let initial_events = manager.try_process_batch().await.unwrap();
+    drive_to_quiescence(&mut manager, &wallet, &blocks, initial_events).await;
+
+    let (highest_used, highest_generated, used_count) =
+        coinjoin_pool_state(&wallet, &wallet_id).await;
+    assert_eq!(
+        highest_used,
+        Some((G + 21) as u32),
+        "block A's outputs sit in a batch that committed before their scripts were \
+         derived, and the session died before the backward sweep finished — only the \
+         durable pending-sweep set can bring them back after the restart. \
+         highest_generated={highest_generated:?}, used_count={used_count}"
+    );
+}
+
+/// A born-wrong `TransactionRecord` IS corrected by the gap rescan — both
+/// in the account and in the emitted event stream. Pins the fix for the
+/// kotlin-sdk "TXO-store reconcile" field bug (2026-08-19), where the
+/// correction never happened:
+///
+/// One block carries a funding tx paying in-window index 0 and a self-send
+/// spending it, paying index G-1 (in-window) and index G+10 (beyond the
+/// initial watch window). First processing records the self-send with the
+/// beyond-window output invisible: `net_amount = PAY - FUND` instead of
+/// `-fee`, and no `output_details` entry for vout 1 — a record born wrong,
+/// projected as-is into every persistence mirror by the emitted events.
+/// Marking G-1 used extends the window past G+10, the commit-time rescan
+/// (#820) re-matches the block, and re-processing runs `update_utxos`
+/// unconditionally — the account's UTXO set self-heals. But
+/// `confirm_transaction` only re-emits (and only mutates) the record when
+/// its *context* changed, so neither the in-memory record nor any event
+/// carries the correction. On-device this is the CoinJoin-funded-send shape:
+/// the store keeps `netAmount` = full input value with the change TXO row
+/// missing, and an engine reload from that store makes the funds vanish.
+#[tokio::test]
+async fn born_wrong_record_is_corrected_by_gap_rescan() {
+    use dashcore::ScriptBuf;
+    use key_wallet::managed_account::transaction_record::{OutputRole, TransactionRecord};
+    use key_wallet_manager::WalletEvent;
+
+    let (mut manager, wallet, wallet_id) = setup().await;
+    let addresses = coinjoin_external_addresses(&wallet, &wallet_id, (G + 11) as u32).await;
+    let mut events_rx = wallet.read().await.subscribe_events();
+
+    const FUND: u64 = 1_000_100_000; // 10.001 into in-window index 0
+    const PAY: u64 = 50_000_000; // 0.5 back to in-window index G-1
+    const HIDDEN: u64 = 950_000_000; // 9.5 to beyond-window index G+10
+    const FEE: u64 = FUND - PAY - HIDDEN; // 100_000
+
+    let funding = Transaction {
+        version: 1,
+        lock_time: 0,
+        input: vec![TxIn {
+            previous_output: OutPoint::new(Txid::from([0xABu8; 32]), 0),
+            script_sig: ScriptBuf::new(),
+            sequence: 0xffffffff,
+            witness: Witness::new(),
+        }],
+        output: vec![TxOut {
+            value: FUND,
+            script_pubkey: addresses[0].script_pubkey(),
+        }],
+        special_transaction_payload: None,
+    };
+    let send = Transaction {
+        version: 1,
+        lock_time: 0,
+        input: vec![TxIn {
+            previous_output: OutPoint::new(funding.txid(), 0),
+            script_sig: ScriptBuf::new(),
+            sequence: 0xffffffff,
+            witness: Witness::new(),
+        }],
+        output: vec![
+            TxOut {
+                value: PAY,
+                script_pubkey: addresses[G - 1].script_pubkey(),
+            },
+            TxOut {
+                value: HIDDEN,
+                script_pubkey: addresses[G + 10].script_pubkey(),
+            },
+        ],
+        special_transaction_payload: None,
+    };
+    let send_txid = send.txid();
+    let hidden_outpoint = OutPoint::new(send_txid, 1);
+
+    let block = Block::dummy(10, vec![funding, send]);
+    let filter = BlockFilter::dummy(&block);
+    let key = FilterMatchKey::new(10, block.block_hash());
+    let blocks: HashMap<BlockHash, Block> = HashMap::from([(block.block_hash(), block.clone())]);
+
+    let mut batch = FiltersBatch::new(0, 99, HashMap::from([(key, filter)]));
+    batch.mark_verified();
+    manager.active_batches.insert(0, batch);
+    manager.progress.update_stored_height(99);
+
+    let initial_events = manager.try_process_batch().await.unwrap();
+    drive_to_quiescence(&mut manager, &wallet, &blocks, initial_events).await;
+
+    {
+        let wm = wallet.read().await;
+        let info = wm.get_wallet_info(&wallet_id).expect("wallet info present");
+        let account =
+            info.coinjoin_managed_account_at_index(0).expect("CoinJoin account 0 present");
+
+        // Engine UTXO self-heal — already correct in the field. If this
+        // fails, the rescan re-processing itself regressed, which is a
+        // different bug than the one this test pins.
+        assert!(
+            account.utxos.contains_key(&hidden_outpoint),
+            "rescan re-processing must insert the beyond-window output into the \
+             account UTXO set (update_utxos runs unconditionally in confirm_transaction)"
+        );
+
+        // The engine's own record must be corrected by the same re-processing.
+        // Note the born-wrong shape is not a MISSING entry: the first
+        // processing classified the unattributable output as `Sent`
+        // (counterparty), so the correction is a role flip.
+        let record =
+            account.transactions().get(&send_txid).expect("send record present in account");
+        assert!(
+            record.output_details.iter().any(
+                |o| o.index == 1 && matches!(o.role, OutputRole::Received | OutputRole::Change)
+            ),
+            "the re-processed record must classify the beyond-window output (vout 1) as \
+             ours (Received/Change); a lingering Sent role means every store projection \
+             derived from this record drops the TXO. got: {:?}",
+            record.output_details,
+        );
+        assert_eq!(
+            record.net_amount,
+            -(FEE as i64),
+            "the re-processed record's net_amount must be recomputed from the now-\
+             complete ownership view (self-send nets -fee); the born-wrong value \
+             PAY-FUND = {} is what makes restored balances collapse on reload",
+            PAY as i64 - FUND as i64,
+        );
+    }
+
+    // The persistence mirror is built exclusively from the emitted events:
+    // whatever the last record-bearing event said about this txid is what
+    // every store on every platform now holds.
+    let mut last_record: Option<TransactionRecord> = None;
+    while let Ok(event) = events_rx.try_recv() {
+        match event {
+            WalletEvent::TransactionDetected {
+                record,
+                ..
+            } if record.txid == send_txid => {
+                last_record = Some(*record);
+            }
+            WalletEvent::BlockProcessed {
+                inserted,
+                updated,
+                ..
+            } => {
+                for r in inserted.into_iter().chain(updated) {
+                    if r.txid == send_txid {
+                        last_record = Some(r);
+                    }
+                }
+            }
+            _ => {}
+        }
+    }
+    let projected = last_record.expect("an event must have carried the send record");
+    assert!(
+        projected
+            .output_details
+            .iter()
+            .any(|o| o.index == 1 && matches!(o.role, OutputRole::Received | OutputRole::Change)),
+        "the LAST emitted record for the send must classify the beyond-window output \
+         (vout 1) as ours — the store mirrors are built from these events only, and \
+         without a corrective emission (confirm_transaction returning None when only \
+         ownership knowledge changed) every mirror keeps the born-wrong Sent role. \
+         got: {:?}",
+        projected.output_details,
+    );
+    assert_eq!(
+        projected.net_amount,
+        -(FEE as i64),
+        "the LAST emitted record's net_amount must be the corrected value; the field \
+         stores show the born-wrong {} shape (full input value) because no corrective \
+         event ever fires",
+        PAY as i64 - FUND as i64,
+    );
+}