diff --git a/dash-spv/src/client/lifecycle.rs b/dash-spv/src/client/lifecycle.rs index 46e26f71c..683df8f7d 100644 --- a/dash-spv/src/client/lifecycle.rs +++ b/dash-spv/src/client/lifecycle.rs @@ -106,6 +106,12 @@ impl DashSpvClient>, + /// 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>, } 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> { - 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> { 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> { + &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) { diff --git a/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs b/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs index a2f54d192..c7e8299c9 100644 --- a/dash-spv/src/sync/filters/coinjoin_gap_discovery_tests.rs +++ b/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::::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 = + 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)> = 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 = 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 = 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, + ); +} diff --git a/dash-spv/src/sync/filters/manager.rs b/dash-spv/src/sync/filters/manager.rs index 5fc18c5f3..9587d989f 100644 --- a/dash-spv/src/sync/filters/manager.rs +++ b/dash-spv/src/sync/filters/manager.rs @@ -91,6 +91,77 @@ pub struct FiltersManager< /// `BlockProcessed` and the per-wallet record of which wallets already /// have a given processed block applied. pub(super) tracker: BlockMatchTracker, + + // === Durable pending-sweep state (interrupted-sync recovery) === + /// Scripts derived during block processing whose rescan cascade has not + /// yet COMMITTED. The in-memory cascade (`collected_scripts` → + /// `backward_scripts` → sweep → commit) evaporates on process death, and + /// nothing ever re-tests already-scanned heights against scripts derived + /// after them — outputs paying those scripts stay invisible to the + /// engine forever (the interrupted-restore fund loss, 2026-08-19). + /// Mirrored to [`Self::metadata`] when present: added when scripts enter + /// the manager, cleared per batch commit via the batch's retired-scripts + /// receipt, and re-seeded into the lowest active batch after a restart. + pending_sweep: HashMap>, + /// The batch start the current `pending_sweep` content has been seeded + /// into, so seeding is idempotent per batch. A tick-triggered rescan + /// wipes `active_batches`, after which the new lowest batch gets its own + /// seeding pass. + pending_seeded_into: Option, + /// Durable store for `pending_sweep`. `None` (tests without storage, + /// callers that never opt in) degrades to the previous in-memory-only + /// behavior. + metadata: Option>>, +} + +/// Metadata key holding the encoded pending-sweep set. +const PENDING_SWEEP_KEY: &str = "filters_pending_sweep"; + +/// Encode the pending-sweep set: `[u32 wallet_count]`, then per wallet +/// `[32-byte wallet id][u32 script_count]` followed by `[u32 len][bytes]` +/// per script. All integers little-endian. Sorted iteration so identical +/// sets encode identically. +fn encode_pending_sweep(pending: &HashMap>) -> Vec { + let mut out = Vec::new(); + let wallets: BTreeMap<&WalletId, &HashSet> = pending.iter().collect(); + out.extend((wallets.len() as u32).to_le_bytes()); + for (wallet_id, scripts) in wallets { + out.extend(*wallet_id); + let mut sorted: Vec<&ScriptBuf> = scripts.iter().collect(); + sorted.sort(); + out.extend((sorted.len() as u32).to_le_bytes()); + for script in sorted { + out.extend((script.len() as u32).to_le_bytes()); + out.extend(script.as_bytes()); + } + } + out +} + +/// Decode [`encode_pending_sweep`]'s format. Any structural inconsistency +/// yields `None` — the caller treats a corrupt blob as "no pending sweep", +/// which fails toward a missed recovery rather than a crash loop. +fn decode_pending_sweep(bytes: &[u8]) -> Option>> { + let mut cursor = 0usize; + let mut read = |n: usize| -> Option<&[u8]> { + let end = cursor.checked_add(n)?; + let slice = bytes.get(cursor..end)?; + cursor = end; + Some(slice) + }; + let wallet_count = u32::from_le_bytes(read(4)?.try_into().ok()?) as usize; + let mut pending = HashMap::new(); + for _ in 0..wallet_count { + let wallet_id: WalletId = read(32)?.try_into().ok()?; + let script_count = u32::from_le_bytes(read(4)?.try_into().ok()?) as usize; + let mut scripts = HashSet::new(); + for _ in 0..script_count { + let len = u32::from_le_bytes(read(4)?.try_into().ok()?) as usize; + scripts.insert(ScriptBuf::from(read(len)?.to_vec())); + } + pending.insert(wallet_id, scripts); + } + Some(pending) } impl @@ -135,6 +206,77 @@ impl>, + ) -> Self { + use crate::storage::MetadataStorage; + match metadata.read().await.load_metadata(PENDING_SWEEP_KEY).await { + Ok(Some(bytes)) => match decode_pending_sweep(&bytes) { + Some(pending) => { + if !pending.is_empty() { + tracing::info!( + wallets = pending.len(), + scripts = pending.values().map(|s| s.len()).sum::(), + "Recovered pending script sweep from a previous session; \ + already-scanned heights will be re-tested for these scripts" + ); + } + self.pending_sweep = pending; + } + None => { + tracing::warn!( + "Corrupt pending-sweep metadata; discarding (a wallet whose \ + discovery was interrupted may need a manual rescan)" + ); + } + }, + Ok(None) => {} + Err(e) => { + tracing::warn!(error = %e, "Failed to load pending-sweep metadata"); + } + } + self.metadata = Some(metadata); + self + } + + /// Record scripts entering the rescan cascade into the durable + /// pending-sweep set. Persisted BEFORE any sweep work runs, so a crash + /// anywhere in the cascade replays them next session. + pub(super) async fn note_pending_sweep( + &mut self, + wallet_id: WalletId, + scripts: impl IntoIterator, + ) { + let entry = self.pending_sweep.entry(wallet_id).or_default(); + let before = entry.len(); + entry.extend(scripts); + if entry.len() != before { + self.persist_pending_sweep().await; + } + } + + /// Write the current pending-sweep set through to metadata storage (a + /// no-op without [`Self::metadata`]). Failures are logged, not fatal: + /// the in-memory cascade still runs; only crash durability degrades. + async fn persist_pending_sweep(&self) { + let Some(metadata) = &self.metadata else { + return; + }; + use crate::storage::MetadataStorage; + let bytes = encode_pending_sweep(&self.pending_sweep); + if let Err(e) = metadata.write().await.store_metadata(PENDING_SWEEP_KEY, &bytes).await { + tracing::warn!(error = %e, "Failed to persist pending-sweep metadata"); } } @@ -158,6 +300,13 @@ impl SyncResult> { let mut events = Vec::new(); + // Phase 0: Seed a recovered pending sweep into the lowest active + // batch. Feeding the scripts through `add_scripts_for_wallet` hands + // them to the existing commit-time cascade — forward rescan of this + // and later batches, then the backward sweep over the committed + // range — so an interrupted session's obligation is honored by the + // same fixpoint that handles freshly derived scripts. Idempotent per + // batch via `pending_seeded_into`; the durable set is only cleared + // when the seeded batch COMMITS (see `try_commit_batches`). + if !self.pending_sweep.is_empty() { + if let Some((&lowest_start, _)) = self.active_batches.first_key_value() { + if self.pending_seeded_into != Some(lowest_start) { + tracing::info!( + wallets = self.pending_sweep.len(), + scripts = self.pending_sweep.values().map(|s| s.len()).sum::(), + batch_start = lowest_start, + "Seeding recovered pending script sweep into the lowest active batch" + ); + let pending = self.pending_sweep.clone(); + if let Some(batch) = self.active_batches.get_mut(&lowest_start) { + for (wallet_id, scripts) in pending { + batch.add_scripts_for_wallet(wallet_id, scripts); + } + } + self.pending_seeded_into = Some(lowest_start); + } + } + } + // Phase 1: Commit completed batches in order events.extend(self.try_commit_batches().await?); @@ -636,6 +813,31 @@ impl self.progress.committed_height() { self.progress.update_committed_height(end); diff --git a/dash-spv/src/sync/filters/sync_manager.rs b/dash-spv/src/sync/filters/sync_manager.rs index 2dbd8ad5e..9b4855ca8 100644 --- a/dash-spv/src/sync/filters/sync_manager.rs +++ b/dash-spv/src/sync/filters/sync_manager.rs @@ -200,6 +200,12 @@ impl< if scripts.is_empty() { continue; } + // Durable first: persist the sweep obligation before + // the in-memory cascade takes it, so a crash anywhere + // between here and the batch's COMMIT replays these + // scripts next session instead of orphaning heights + // scanned before the scripts existed. + self.note_pending_sweep(*wallet_id, scripts.iter().cloned()).await; if let Some(batch) = self.active_batches.get_mut(&batch_start) { batch.add_scripts_for_wallet(*wallet_id, scripts.iter().cloned()); } diff --git a/key-wallet/src/managed_account/address_pool.rs b/key-wallet/src/managed_account/address_pool.rs index 57c021bed..def05bd48 100644 --- a/key-wallet/src/managed_account/address_pool.rs +++ b/key-wallet/src/managed_account/address_pool.rs @@ -448,6 +448,41 @@ impl AddressPool { self.pool_type == AddressPoolType::External } + /// Derive every missing index in `0..=index`, leaving existing entries + /// (and their used flags) untouched. Returns how many holes were filled. + /// + /// A pool restored from a persistence mirror can be SPARSE: mirrors have + /// been observed dropping individual address rows, and a restore that + /// ingests the surviving rows as-is both inherits the holes and (by + /// advancing `highest_generated` to the max persisted index) suppresses + /// the gap-limit maintenance that would otherwise re-derive them. An + /// address missing from the pool makes every output paying it + /// permanently unrecognizable — funds invisible on every rescan. Since + /// derivation is pure arithmetic over the key source, holes are always + /// repairable; restore paths call this after ingesting persisted rows. + pub fn ensure_contiguous_to(&mut self, index: u32, key_source: &KeySource) -> Result { + // A restored watermark is untrusted input: a corrupt row could name + // an index in the billions and turn the repair into an unbounded + // derivation stall during wallet load. Real pools top out in the + // low thousands (heaviest observed field wallet: 8,281); anything + // past this bound is corruption, and refusing restores the pool + // exactly as persisted — the pre-repair behavior. + const MAX_REPAIR_INDEX: u32 = 1_000_000; + if index > MAX_REPAIR_INDEX { + return Err(Error::InvalidParameter(format!( + "refusing address-pool hole repair to index {index}: exceeds the {MAX_REPAIR_INDEX} repair bound (corrupt watermark?)" + ))); + } + let mut filled = 0u32; + for idx in 0..=index { + if !self.addresses.contains_key(&idx) { + self.generate_address_at_index(idx, key_source, true)?; + filled += 1; + } + } + Ok(filled) + } + /// Generate addresses up to the specified count pub fn generate_addresses( &mut self, @@ -1367,6 +1402,86 @@ mod tests { assert_eq!(pool.addresses.len(), 10); } + /// A sparse pool (as restored from a lossy persistence mirror) must be + /// repairable to a contiguous one without disturbing surviving entries: + /// holes are re-derived, existing entries and their used flags are kept, + /// and the filled addresses match what direct derivation produces (so + /// recognition of outputs paying formerly-holed addresses works). + #[test] + fn test_ensure_contiguous_fills_holes_and_preserves_used() { + let base_path = DerivationPath::from(vec![ChildNumber::from_normal_idx(0).unwrap()]); + let key_source = test_key_source(); + + // The reference: a contiguous pool 0..=9. + let mut reference = AddressPool::new_without_generation( + base_path.clone(), + AddressPoolType::External, + 20, + Network::Testnet, + ); + reference.generate_addresses(10, &key_source, true).unwrap(); + + // The victim: same pool, then indices 3, 4, 7 dropped (mirror holes) + // with `highest_generated` still claiming full coverage — the exact + // post-restore shape. + let mut sparse = AddressPool::new_without_generation( + base_path, + AddressPoolType::External, + 20, + Network::Testnet, + ); + sparse.generate_addresses(10, &key_source, true).unwrap(); + let used_addr = sparse.addresses[&5].address.clone(); + sparse.mark_used(&used_addr); + for idx in [3u32, 4, 7] { + let info = sparse.addresses.remove(&idx).unwrap(); + sparse.address_index.remove(&info.address); + sparse.script_pubkey_index.remove(&info.script_pubkey); + } + assert_eq!(sparse.addresses.len(), 7); + assert_eq!(sparse.highest_generated, Some(9), "watermark still claims coverage"); + + let filled = sparse.ensure_contiguous_to(9, &key_source).unwrap(); + assert_eq!(filled, 3, "exactly the three holes are re-derived"); + assert_eq!(sparse.addresses.len(), 10); + for idx in 0..=9u32 { + assert_eq!( + sparse.addresses.get(&idx).map(|i| &i.address), + reference.addresses.get(&idx).map(|i| &i.address), + "index {idx} must match direct derivation" + ); + assert!( + sparse.address_index.contains_key(&reference.addresses[&idx].address), + "reverse lookup must cover index {idx}" + ); + } + assert!(sparse.used_indices.contains(&5), "used flag survives the repair"); + assert_eq!( + sparse.addresses[&5].state, + AddressState::Used, + "the surviving entry's state is untouched" + ); + for idx in [3u32, 4, 7] { + let info = &sparse.addresses[&idx]; + assert_eq!( + sparse.address_index.get(&info.address), + Some(&idx), + "repaired index {idx} maps back through the address index" + ); + assert_eq!( + sparse.script_pubkey_index.get(&info.script_pubkey), + Some(&idx), + "repaired index {idx} maps back through the script index" + ); + assert_eq!(info.state, AddressState::Available, "a repaired hole starts Available"); + } + // The bound refuses implausible watermarks instead of stalling. + assert!(sparse.ensure_contiguous_to(u32::MAX, &key_source).is_err()); + + // Idempotent: a second pass finds nothing to fill. + assert_eq!(sparse.ensure_contiguous_to(9, &key_source).unwrap(), 0); + } + #[test] fn test_address_usage() { let base_path = DerivationPath::from(vec![ChildNumber::from_normal_idx(0).unwrap()]); diff --git a/key-wallet/src/managed_account/managed_account_ref.rs b/key-wallet/src/managed_account/managed_account_ref.rs index 172d40327..692a540f2 100644 --- a/key-wallet/src/managed_account/managed_account_ref.rs +++ b/key-wallet/src/managed_account/managed_account_ref.rs @@ -415,6 +415,32 @@ impl<'a> ManagedAccountRefMut<'a> { ManagedAccountRefMut::Keys(_) => false, } } + + /// Drain the born-spent funding outputs staged during the last + /// `record_transaction` / `confirm_transaction` call (out-of-order + /// funding — see `ManagedCoreFundsAccount::take_born_spent_outputs`). + /// Always empty for the [`Keys`](Self::Keys) variant. + pub(crate) fn take_born_spent_outputs(&mut self) -> Vec<(OutPoint, u64, Address)> { + match self { + ManagedAccountRefMut::Funds(a) => a.take_born_spent_outputs(), + ManagedAccountRefMut::Keys(_) => Vec::new(), + } + } + + /// Attribute a born-spent funding output onto every recorded spender in + /// this account — the wallet-scope half of the out-of-order correction. + /// Always empty for the [`Keys`](Self::Keys) variant. + pub(crate) fn attribute_spent_input( + &mut self, + outpoint: &OutPoint, + value: u64, + address: &Address, + ) -> Vec { + match self { + ManagedAccountRefMut::Funds(a) => a.attribute_spent_input(outpoint, value, address), + ManagedAccountRefMut::Keys(_) => Vec::new(), + } + } } /// Owned managed core account, either funds-bearing or keys-only. diff --git a/key-wallet/src/managed_account/managed_core_funds_account.rs b/key-wallet/src/managed_account/managed_core_funds_account.rs index ca3210bb6..611d55310 100644 --- a/key-wallet/src/managed_account/managed_core_funds_account.rs +++ b/key-wallet/src/managed_account/managed_core_funds_account.rs @@ -68,6 +68,20 @@ pub struct ManagedCoreFundsAccount { /// re-establish which coins are spent. #[cfg_attr(feature = "serde", serde(skip))] reservations: ReservationSet, + /// Born-spent funding outputs staged by [`Self::update_utxos`]: this + /// account recognized an output of a FUNDING transaction that an + /// earlier-processed transaction already spent (out-of-order block + /// processing during rescan/discovery). The spender was recorded with + /// that input unattributed (`net_amount` = income only — the + /// inflated-history shape, 2026-08-20 field wallet: 49 one-sided rows, + /// +4.63 DASH of phantom net) — and the spender's record may live in a + /// SIBLING account (it matched wherever its own outputs landed), so the + /// correction must run at wallet scope: `wallet_checker` drains this + /// staging and calls [`Self::attribute_spent_input`] on every fund + /// account, emitting each corrected record. Transient working state, + /// never persisted. + #[cfg_attr(feature = "serde", serde(skip))] + born_spent_outputs: Vec<(OutPoint, u64, Address)>, } /// What [`ManagedCoreFundsAccount::apply_abandon`] removed from one account. @@ -104,6 +118,7 @@ impl ManagedCoreFundsAccount { utxos: BTreeMap::new(), spent_outpoints: HashSet::new(), reservations: ReservationSet::default(), + born_spent_outputs: Vec::new(), } } @@ -131,6 +146,7 @@ impl ManagedCoreFundsAccount { utxos: BTreeMap::new(), spent_outpoints: HashSet::new(), reservations: ReservationSet::default(), + born_spent_outputs: Vec::new(), } } @@ -189,6 +205,23 @@ impl ManagedCoreFundsAccount { self.spent_outpoints.contains(outpoint) } + /// The outpoints this account knows were spent by recorded transactions. + /// Read-only view for store reconciliation. POSITIVE membership is the + /// only safe signal: a mirror row still marked unspent whose outpoint + /// appears here lost its spend update (the dashpay/platform#4425 + /// stale-TXO class) and can be flipped. + /// + /// ABSENCE proves nothing. Under the default feature configuration, + /// finalized (chain-locked) transactions drop their full records and + /// this set is rebuilt from the records that remain, so a genuinely + /// spent outpoint can be missing from BOTH this set and [`Self::utxos`] + /// after a reload. A reconciler must treat "in neither inventory" as + /// ambiguous — swept/abandoned residue (pre-rust-dashcore#971 stores) + /// OR a finalized spend whose record is gone — and never mutate on it. + pub fn spent_outpoints(&self) -> &HashSet { + &self.spent_outpoints + } + /// Collect the outpoints among `tx`'s inputs that this account holds as a /// final UTXO — confirmed, InstantSend-locked, or trusted. /// @@ -237,6 +270,14 @@ impl ManagedCoreFundsAccount { /// outpoints that a *sibling* account of the same wallet holds as final. /// See [`Self::record_transaction`] for why a per-account view is not /// enough. An empty set degrades this to the account-local check. + /// Returns an [`OutputDetail`] for every output of `tx` this call + /// recognized as paying this account — including outputs whose UTXO + /// insertion was skipped because an earlier-processed block already + /// spent them (they are still ours and still belong on the record). + /// `record_transaction` ignores the return value (its details come from + /// the same match); `confirm_transaction` uses it to detect ownership + /// learned only on re-processing — the gap-limit rescan shape — and fold + /// the correction back into the already-stored record. fn update_utxos( &mut self, tx: &Transaction, @@ -244,7 +285,8 @@ impl ManagedCoreFundsAccount { context: TransactionContext, observed_spent: &BTreeMap, external_final_parents: &BTreeSet, - ) { + ) -> Vec { + let mut recognized: Vec = Vec::new(); // Update UTXOs only for spendable account types match self.keys.managed_account_type() { ManagedAccountType::Standard { @@ -320,7 +362,9 @@ impl ManagedCoreFundsAccount { %txid, "Not crediting a transaction whose input a block already spent" ); - return; + // Nothing recognized either: a doomed transaction's + // outputs must not be folded into any record. + return recognized; } let network = self.keys.network(); @@ -334,6 +378,21 @@ impl ManagedCoreFundsAccount { vout: vout as u32, }; + // Report recognition before the spent-skips below: + // an output of ours that is already spent on-chain + // still belongs on the transaction's record and in + // its net_amount. + recognized.push(OutputDetail { + index: vout as u32, + role: if change_addrs.contains(&addr) { + OutputRole::Change + } else { + OutputRole::Received + }, + address: Some(addr.clone()), + value: output.value, + }); + // Check if this outpoint was already spent by a transaction we've seen. // This handles out-of-order block processing during rescan where a // spending transaction at a higher height may be processed before @@ -346,6 +405,19 @@ impl ManagedCoreFundsAccount { outpoint = %outpoint, "Skipping UTXO already spent by previously processed transaction" ); + // Born-spent: the spender ran BEFORE this + // funding transaction, so its record was + // built with this input unattributed + // (income-only net — the inflated-history + // shape). This is the one moment the missing + // attribution is provable; patch the + // spender's record and stage it for + // re-emission. + self.born_spent_outputs.push(( + outpoint, + output.value, + addr.clone(), + )); continue; } @@ -358,6 +430,15 @@ impl ManagedCoreFundsAccount { outpoint = %outpoint, "Skipping UTXO already observed spent in an earlier-processed block (#649)" ); + // Same born-spent shape via the wallet-level + // observed-spends view: if this account also + // holds the spender's record, its spent side + // is missing this input too. + self.born_spent_outputs.push(( + outpoint, + output.value, + addr.clone(), + )); continue; } @@ -422,6 +503,68 @@ impl ManagedCoreFundsAccount { } _ => {} } + recognized + } + + /// Attribute a born-spent funding output onto EVERY recorded + /// transaction in THIS account that spends `outpoint` (a conflicting + /// double-spend can leave more than one). Called at wallet scope by + /// `wallet_checker` for each staged born-spent output, across all fund + /// accounts — the spender's record lives wherever its own outputs + /// matched, which need not be the account that owns the funding output. + /// Patches the input detail (index, value, address), recomputes the + /// record's derived fields, and returns the corrected copies for + /// re-emission. No-op for records already attributed and for accounts + /// holding no matching record (a finalized record already dropped + /// cannot be patched). + pub(crate) fn attribute_spent_input( + &mut self, + outpoint: &OutPoint, + value: u64, + address: &Address, + ) -> Vec { + let spenders: Vec<(Txid, u32)> = self + .keys + .transactions() + .iter() + .filter_map(|(txid, rec)| { + rec.transaction + .input + .iter() + .position(|i| &i.previous_output == outpoint) + .map(|pos| (*txid, pos as u32)) + }) + .collect(); + let mut corrected = Vec::new(); + for (spender_txid, input_index) in spenders { + let Some(record) = self.keys.transactions_mut().get_mut(&spender_txid) else { + continue; + }; + if record.input_details.iter().any(|d| d.index == input_index) { + continue; + } + record.input_details.push(InputDetail { + index: input_index, + value, + address: address.clone(), + }); + record.input_details.sort_by_key(|d| d.index); + record.recompute_net_and_direction(); + tracing::info!( + outpoint = %outpoint, + spender = %spender_txid, + corrected_net = record.net_amount, + "Attributed born-spent funding output onto its spender's record" + ); + corrected.push(record.clone()); + } + corrected + } + + /// Drain the born-spent outputs staged by [`Self::update_utxos`] since + /// the last drain, for the wallet-scope attribution sweep. + pub(crate) fn take_born_spent_outputs(&mut self) -> Vec<(OutPoint, u64, Address)> { + std::mem::take(&mut self.born_spent_outputs) } /// Drop the spent-marks that `freed` contributed, keeping every mark a @@ -767,14 +910,6 @@ impl ManagedCoreFundsAccount { } } - // Capture the (possibly updated) record before any pruning so the - // caller can still emit it in an event. - let record_after = if changed { - self.keys.transactions().get(&txid).cloned() - } else { - None - }; - // The chainlock is the trigger for dropping the full record under // the default feature configuration; an IS-lock alone is *not* // enough — we keep the record so the surrounding block @@ -782,7 +917,58 @@ impl ManagedCoreFundsAccount { // chainlock catches up. #[cfg(not(feature = "keep-finalized-transactions"))] let drop_now = context.is_chain_locked(); - self.update_utxos(tx, account_match, context, observed_spent, external_final_parents); + let recognized = + self.update_utxos(tx, account_match, context, observed_spent, external_final_parents); + + // Gap-limit rescan correction: a re-processed block can recognize + // outputs the FIRST processing could not see (their addresses were + // beyond the watch window then, derived since). `update_utxos` above + // already healed the account's UTXO set; without the block below the + // stored record — and every persistence mirror built from emitted + // records — keeps the born-wrong shape forever: `net_amount` equal to + // the full input value and no `output_details` entry for the output. + // A reload from such a mirror is exactly the field fund-loss this + // corrects (kotlin-sdk TXO-store bug, 2026-08-19). Merge the newly + // recognized outputs into the record, recompute the derived fields, + // and signal the caller so a corrective event is emitted. + if let Some(tx_record) = self.keys.transactions_mut().get_mut(&txid) { + // An output the first processing could not attribute is not + // absent from the record — `record_transaction` classified it as + // `Sent` (counterparty). Recognition therefore corrects roles in + // place as well as filling genuine gaps. + let mut corrected = false; + for ours in recognized { + match tx_record.output_details.iter_mut().find(|o| o.index == ours.index) { + Some(existing) => { + if existing.role != ours.role { + *existing = ours; + corrected = true; + } + } + None => { + tx_record.output_details.push(ours); + corrected = true; + } + } + } + if corrected { + tx_record.output_details.sort_by_key(|o| o.index); + // Shared derivation over the now-complete details. + // `input_details` was built at first processing while the + // spent parents were still live, so the spend side needs no + // recomputation here. + tx_record.recompute_net_and_direction(); + changed = true; + } + } + + // Capture the (possibly updated) record before any pruning so the + // caller can still emit it in an event. + let record_after = if changed { + self.keys.transactions().get(&txid).cloned() + } else { + None + }; #[cfg(not(feature = "keep-finalized-transactions"))] if drop_now { self.keys.drop_finalized_transaction(&txid); @@ -1326,6 +1512,7 @@ impl<'de> Deserialize<'de> for ManagedCoreFundsAccount { utxos: helper.utxos, spent_outpoints, reservations: ReservationSet::default(), + born_spent_outputs: Vec::new(), }) } } diff --git a/key-wallet/src/managed_account/transaction_record.rs b/key-wallet/src/managed_account/transaction_record.rs index b51aee6f1..536220ae5 100644 --- a/key-wallet/src/managed_account/transaction_record.rs +++ b/key-wallet/src/managed_account/transaction_record.rs @@ -131,6 +131,39 @@ impl TransactionRecord { } } + /// Recompute `net_amount` and `direction` from the CURRENT + /// `input_details` / `output_details` — the single derivation both + /// late-attribution paths share: output-side corrections (a rescan + /// newly recognizing an output the first processing recorded as + /// `Sent`) and input-side corrections (a funding transaction processed + /// AFTER its spender attributing the spent input onto the spender's + /// record). `net_amount` is Σ(owned outputs) − Σ(attributed inputs); + /// `direction` follows `record_transaction`'s rules, except a + /// `CoinJoin` direction is never overwritten (it is keyed to the + /// transaction type, not the flow shape). + pub fn recompute_net_and_direction(&mut self) { + let owned: i64 = self + .output_details + .iter() + .filter(|o| matches!(o.role, OutputRole::Received | OutputRole::Change)) + .map(|o| o.value as i64) + .sum(); + let spent: i64 = self.input_details.iter().map(|i| i.value as i64).sum(); + self.net_amount = owned - spent; + if self.direction != TransactionDirection::CoinJoin { + let has_inputs = !self.input_details.is_empty(); + let has_sent = self.output_details.iter().any(|d| d.role == OutputRole::Sent); + let has_our_outputs = owned > 0; + self.direction = if !has_sent && has_inputs && has_our_outputs { + TransactionDirection::Internal + } else if has_inputs { + TransactionDirection::Outgoing + } else { + TransactionDirection::Incoming + }; + } + } + /// Calculate the number of confirmations based on current chain height pub fn confirmations(&self, current_height: u32) -> u32 { match self.context.block_info() { diff --git a/key-wallet/src/transaction_checking/wallet_checker.rs b/key-wallet/src/transaction_checking/wallet_checker.rs index efea71a5f..8a8b230d3 100644 --- a/key-wallet/src/transaction_checking/wallet_checker.rs +++ b/key-wallet/src/transaction_checking/wallet_checker.rs @@ -195,6 +195,7 @@ impl WalletTransactionChecker for ManagedWalletInfo { } // Process each affected account + let mut born_spent: Vec<(dashcore::OutPoint, u64, dashcore::Address)> = Vec::new(); for account_match in result.affected_accounts.clone() { let Some(mut account) = self.accounts.get_by_account_type_match_mut(&account_match.account_type_match) @@ -232,6 +233,14 @@ impl WalletTransactionChecker for ManagedWalletInfo { } } + // Born-spent staging: processing THIS transaction may have + // revealed outputs an earlier-processed transaction already + // spent (out-of-order funding during rescan/discovery). Collect + // them here; the attribution sweep runs at wallet scope after + // this loop, because the spender's record lives wherever ITS + // outputs matched — not necessarily in this account. + born_spent.extend(account.take_born_spent_outputs()); + for address_info in account_match.account_type_match.all_involved_addresses() { account.mark_address_used(&address_info.address); } @@ -270,6 +279,26 @@ impl WalletTransactionChecker for ManagedWalletInfo { } } + // Wallet-scope born-spent attribution (out-of-order funding): each + // staged output's spender may be recorded in ANY fund account, and a + // conflicting double-spend can leave several records. Runs only when + // something was staged — the common path never touches the full + // account list. The corrections surface as updated records so an + // event re-emits them to the persistence mirrors; without this the + // engine's records are right but every store keeps the income-only + // net (the inflated-history shape). + if !born_spent.is_empty() { + for (outpoint, value, address) in &born_spent { + for mut account in self.accounts.all_accounts_mut() { + let corrected = account.attribute_spent_input(outpoint, *value, address); + if !corrected.is_empty() { + result.state_modified = true; + result.updated_records.extend(corrected); + } + } + } + } + if is_new { // Populate dedup sets when a tx arrives with an initial IS status if context.is_instant_send() { @@ -3592,4 +3621,216 @@ mod tests { assert_eq!(ctx.managed_wallet.balance.confirmed(), 0); assert_eq!(ctx.managed_wallet.balance.unconfirmed(), payment_value); } + + /// Cross-account variant of the born-spent correction: the funding + /// output belongs to the BIP44 account, but the spender's record lives + /// in the COINJOIN account (it matched there via its own outputs when + /// processed first, its input unknown). The attribution must run at + /// wallet scope — an account-local lookup finds no spender and the + /// CoinJoin record keeps its income-only net forever. + #[tokio::test] + async fn born_spent_attribution_reaches_sibling_account_spenders() { + let network = Network::Testnet; + let mut wallet = + Wallet::new_random(network, WalletAccountCreationOptions::Default).expect("wallet"); + let mut managed_wallet = + ManagedWalletInfo::from_wallet_with_name(&wallet, "Test".to_string(), 0); + + // Funding pays the BIP44 receive address (account A). + let bip44_xpub = + wallet.accounts.standard_bip44_accounts.get(&0).expect("bip44").account_xpub; + let bip44_address = managed_wallet + .first_bip44_managed_account_mut() + .expect("bip44 managed") + .next_receive_address(Some(&bip44_xpub), true) + .expect("bip44 address"); + // The spender pays the CoinJoin account (account B) — so processed + // first, it is recorded in B only. The pool is pre-generated to the + // gap limit at construction; index 0 is already watched. + let cj_address = managed_wallet + .coinjoin_managed_account_at_index(0) + .expect("coinjoin managed") + .managed_account_type() + .address_pools() + .into_iter() + .find(|pool| { + pool.pool_type == crate::managed_account::address_pool::AddressPoolType::External + }) + .expect("coinjoin external pool") + .address_at_index(0) + .expect("pre-generated coinjoin address") + .clone(); + + const FUND: u64 = 1_000_000; + const BACK: u64 = 900_000; + let funding_tx = Transaction::dummy(&bip44_address, 0..1, &[FUND]); + let funded_outpoint = OutPoint { + txid: funding_tx.txid(), + vout: 0, + }; + let spender_tx = Transaction { + version: 1, + lock_time: 0, + input: vec![TxIn { + previous_output: funded_outpoint, + script_sig: ScriptBuf::new(), + sequence: 0xffffffff, + witness: dashcore::Witness::new(), + }], + output: vec![TxOut { + value: BACK, + script_pubkey: cj_address.script_pubkey(), + }], + special_transaction_payload: None, + }; + let spender_txid = spender_tx.txid(); + + // Spender first (height 2), landing in the CoinJoin account. + let result = managed_wallet + .check_core_transaction( + &spender_tx, + TransactionContext::InBlock(BlockInfo::new( + 2, + BlockHash::from_slice(&[3u8; 32]).expect("hash"), + 1_650_000_100, + )), + &mut wallet, + true, + true, + ) + .await; + assert!(result.is_relevant && result.is_new_transaction); + { + let cj = managed_wallet.coinjoin_managed_account_at_index(0).expect("cj"); + let record = cj.transactions().get(&spender_txid).expect("spender in CoinJoin acct"); + assert_eq!(record.net_amount, BACK as i64, "born income-only in the sibling account"); + } + + // Funding second (height 1) — recognized by BIP44, whose account- + // local records contain no spender. The wallet-scope sweep must + // reach the CoinJoin record. + let result = managed_wallet + .check_core_transaction( + &funding_tx, + TransactionContext::InBlock(BlockInfo::new( + 1, + BlockHash::from_slice(&[2u8; 32]).expect("hash"), + 1_650_000_000, + )), + &mut wallet, + true, + true, + ) + .await; + assert!(result.is_relevant); + + let cj = managed_wallet.coinjoin_managed_account_at_index(0).expect("cj"); + let record = cj.transactions().get(&spender_txid).expect("spender record"); + assert_eq!( + record.net_amount, + BACK as i64 - FUND as i64, + "the sibling account's record gains the spent side" + ); + assert_eq!(record.input_details.len(), 1); + let corrected = result + .updated_records + .iter() + .find(|r| r.txid == spender_txid) + .expect("cross-account correction surfaces in updated_records"); + assert_eq!(corrected.net_amount, BACK as i64 - FUND as i64); + } + + /// Out-of-order funding (rescan block-download order): the SPENDER is + /// processed before the transaction that funded it. At spender-process + /// time its input is unknown, so the record is born income-only + /// (`net_amount = +received` — the inflated-history shape: 2026-08-20 + /// field wallet showed 49 such rows summing +4.63 DASH of phantom net). + /// When the funding transaction is finally processed, its born-spent + /// output must be attributed onto the spender's record — corrected net, + /// input_details filled — and the corrected record must surface in + /// `updated_records` so an event re-emits it to persistence mirrors. + #[tokio::test] + async fn born_spent_attribution_corrects_out_of_order_spender() { + let mut ctx = TestWalletContext::new_random(); + const FUND: u64 = 1_000_000; + const BACK: u64 = 900_000; + + let funding_tx = Transaction::dummy(&ctx.receive_address, 0..1, &[FUND]); + let funded_outpoint = OutPoint { + txid: funding_tx.txid(), + vout: 0, + }; + let spender_tx = Transaction { + version: 1, + lock_time: 0, + input: vec![TxIn { + previous_output: funded_outpoint, + script_sig: ScriptBuf::new(), + sequence: 0xffffffff, + witness: dashcore::Witness::new(), + }], + output: vec![TxOut { + value: BACK, + script_pubkey: ctx.receive_address.script_pubkey(), + }], + special_transaction_payload: None, + }; + let spender_txid = spender_tx.txid(); + + // Spender first (height 2 processed before height 1). + let spender_context = TransactionContext::InBlock(BlockInfo::new( + 2, + BlockHash::from_slice(&[3u8; 32]).expect("block hash"), + 1_650_000_100, + )); + let result = ctx.check_transaction(&spender_tx, spender_context).await; + assert!(result.is_relevant && result.is_new_transaction); + { + let account = ctx.managed_wallet.first_bip44_managed_account().expect("bip44 account"); + let record = account.transactions().get(&spender_txid).expect("spender recorded"); + assert_eq!( + record.net_amount, BACK as i64, + "pin the born-wrong shape: income-only net before the funding tx arrives" + ); + assert!(record.input_details.is_empty()); + } + + // Funding second — out of order. + let funding_context = TransactionContext::InBlock(BlockInfo::new( + 1, + BlockHash::from_slice(&[2u8; 32]).expect("block hash"), + 1_650_000_000, + )); + let result = ctx.check_transaction(&funding_tx, funding_context).await; + assert!(result.is_relevant); + + // The engine record is corrected... + { + let account = ctx.managed_wallet.first_bip44_managed_account().expect("bip44 account"); + let record = account.transactions().get(&spender_txid).expect("spender record"); + assert_eq!( + record.net_amount, + BACK as i64 - FUND as i64, + "spent side attributed: net = received - spent" + ); + assert_eq!(record.input_details.len(), 1); + assert_eq!(record.input_details[0].value, FUND); + // The born-spent funding output must NOT enter the UTXO set. + assert!( + !account.utxos.contains_key(&funded_outpoint), + "a born-spent output never becomes spendable" + ); + } + + // ...AND the correction surfaces as an updated record for the event + // pipeline — without this every persistence mirror keeps the + // income-only net forever. + let corrected = result + .updated_records + .iter() + .find(|r| r.txid == spender_txid) + .expect("corrected spender record must surface in updated_records"); + assert_eq!(corrected.net_amount, BACK as i64 - FUND as i64); + assert_eq!(corrected.input_details.len(), 1); + } }