fix(wallet): reconcile late inputs and publish accounting corrections

CHANGELOG.md

@@ -6,6 +6,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## Unreleased
 
+### Fixed
+
+- Keep late wallet input attribution account-local, restore outgoing history details,
+  and emit complete accounting corrections without assigning another transaction’s InstantSend lock.
+  Preserve finalized-record retention after collecting all late input corrections.
+
 ### Changed
 
 - **Breaking:** the `bincode` feature and binary serialization dependencies now use

key-wallet-manager/src/event_tests.rs

@@ -2119,3 +2119,124 @@ async fn dropped_persistence_consumer_does_not_wedge_emission() {
         "broadcast delivery must be unaffected by a lost persistence consumer"
     );
 }
+
+#[tokio::test]
+async fn should_emit_late_input_corrections_without_borrowing_funding_lock() {
+    for locked_funding in [false, true] {
+        let (mut manager, wallet_id, addr) = setup_manager_with_wallet();
+        let funding = create_tx_paying_to(&addr, 0xa1);
+        let mut spender = create_tx_paying_to(&addr, 0xa2);
+        spender.input[0].previous_output = OutPoint {
+            txid: funding.txid(),
+            vout: 0,
+        };
+        spender.output[0].value = TX_AMOUNT - 2000;
+        spender.output.push(TxOut {
+            value: 1000,
+            script_pubkey: ScriptBuf::new_p2pkh(
+                &PublicKey::from_slice(&[2; 33]).unwrap().pubkey_hash(),
+            ),
+        });
+        manager.process_mempool_transaction(&spender, None).await;
+        let mut rx = manager.subscribe_events();
+        let lock = locked_funding.then(|| dummy_instant_lock(funding.txid()));
+        manager.process_mempool_transaction(&funding, lock.clone()).await;
+        let events = drain_events(&mut rx);
+        let corrected = events
+            .iter()
+            .find_map(|event| match event {
+                WalletEvent::TransactionDetected {
+                    wallet_id: id,
+                    record,
+                    ..
+                } if *id == wallet_id && record.txid == spender.txid() => Some(record),
+                _ => None,
+            })
+            .expect("the corrected spender must reach persistence subscribers");
+        assert_eq!(corrected.net_amount, -2000);
+        assert_eq!(
+            corrected.direction,
+            key_wallet::managed_account::transaction_record::TransactionDirection::Outgoing
+        );
+        assert_eq!(corrected.input_details.len(), 1);
+        assert_eq!(corrected.context, TransactionContext::Mempool);
+        assert!(
+            !events.iter().any(|event| matches!(event,
+                WalletEvent::TransactionInstantLocked { txid, .. } if *txid == spender.txid()
+            )),
+            "the funding lock must never be assigned to its spender"
+        );
+        manager.process_mempool_transaction(&funding, lock).await;
+        assert_no_events(&mut rx);
+    }
+}
+
+#[tokio::test]
+async fn should_preserve_attributed_inputs_when_another_funding_arrives() {
+    let (mut manager, _, addr) = setup_manager_with_wallet();
+    let first = create_tx_paying_to(&addr, 0xb1);
+    let second = create_tx_paying_to(&addr, 0xb2);
+    let mut spender = create_tx_paying_to(&addr, 0xb3);
+    spender.input[0].previous_output = OutPoint {
+        txid: first.txid(),
+        vout: 0,
+    };
+    let mut input = spender.input[0].clone();
+    input.previous_output.txid = second.txid();
+    spender.input.push(input);
+    spender.output[0].value = 2 * TX_AMOUNT - 1000;
+    manager.process_mempool_transaction(&spender, None).await;
+    manager.process_mempool_transaction(&first, None).await;
+    let mut rx = manager.subscribe_events();
+    manager.process_mempool_transaction(&second, None).await;
+    let events = drain_events(&mut rx);
+    let corrected = events
+        .iter()
+        .find_map(|event| match event {
+            WalletEvent::TransactionDetected {
+                record,
+                ..
+            } if record.txid == spender.txid() => Some(record),
+            _ => None,
+        })
+        .expect("second input correction");
+    assert_eq!(corrected.net_amount, -1000);
+    assert_eq!(corrected.input_details.len(), 2);
+    assert_eq!(corrected.input_details[0].index, 0);
+    assert_eq!(corrected.input_details[1].index, 1);
+    manager.process_mempool_transaction(&second, None).await;
+    assert_no_events(&mut rx);
+}
+
+#[tokio::test]
+async fn should_emit_one_complete_correction_for_multiple_inputs_from_one_parent() {
+    let (mut manager, _, addr) = setup_manager_with_wallet();
+    let mut funding = create_tx_paying_to(&addr, 0xc1);
+    funding.output.push(funding.output[0].clone());
+    let mut spender = create_tx_paying_to(&addr, 0xc2);
+    spender.input[0].previous_output = OutPoint {
+        txid: funding.txid(),
+        vout: 0,
+    };
+    let mut input = spender.input[0].clone();
+    input.previous_output.vout = 1;
+    spender.input.push(input);
+    spender.output[0].value = 2 * TX_AMOUNT - 1000;
+    manager.process_mempool_transaction(&spender, None).await;
+    let mut rx = manager.subscribe_events();
+    manager.process_mempool_transaction(&funding, None).await;
+    let events = drain_events(&mut rx);
+    let corrected: Vec<_> = events
+        .iter()
+        .filter_map(|event| match event {
+            WalletEvent::TransactionDetected {
+                record,
+                ..
+            } if record.txid == spender.txid() => Some(record),
+            _ => None,
+        })
+        .collect();
+    assert_eq!(corrected.len(), 1, "one final slice per account and transaction");
+    assert_eq!(corrected[0].net_amount, -1000);
+    assert_eq!(corrected[0].input_details.len(), 2);
+}

key-wallet-manager/src/events.rs

@@ -180,9 +180,8 @@ fn format_account_balances(map: &BTreeMap<AccountType, WalletCoreBalance>) -> St
 /// consumers can persist the record(s) and balance atomically.
 #[derive(Debug, Clone)]
 pub enum WalletEvent {
-    /// First time the wallet sees an off-chain wallet-relevant transaction
-    /// (mempool, or directly via an InstantSend lock — in that case
-    /// `record.context` is `InstantSend(..)`).
+    /// An off-chain transaction was detected or its accounting details were corrected.
+    /// Corrections retain the transaction's own confirmation context; consumers upsert by account and txid.
     TransactionDetected {
         /// ID of the affected wallet.
         wallet_id: WalletId,

key-wallet-manager/src/process_block.rs

@@ -272,26 +272,30 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> WalletInterface for WalletM
             per_wallet_released
         );
 
-        if let Some(lock) = instant_lock {
-            for (wallet_id, records) in per_wallet_updated_records {
-                if records.is_empty() {
-                    continue;
-                }
-                let Some(info) = self.wallet_infos.get(&wallet_id) else {
-                    continue;
-                };
-                let balance = info.balance();
-                let account_balances =
-                    per_wallet_account_diff.get(&wallet_id).cloned().unwrap_or_default();
-                for record in records {
-                    let event = WalletEvent::TransactionInstantLocked {
+        for (wallet_id, records) in per_wallet_updated_records {
+            let Some(info) = self.wallet_infos.get(&wallet_id) else {
+                continue;
+            };
+            let balance = info.balance();
+            let account_balances =
+                per_wallet_account_diff.get(&wallet_id).cloned().unwrap_or_default();
+            for record in records {
+                let txid = record.txid;
+                self.emit_event(WalletEvent::TransactionDetected {
+                    wallet_id,
+                    record: Box::new(record),
+                    balance,
+                    account_balances: account_balances.clone(),
+                    addresses_derived: Vec::new(),
+                });
+                if let Some(lock) = instant_lock.as_ref().filter(|lock| lock.txid == txid) {
+                    self.emit_event(WalletEvent::TransactionInstantLocked {
                         wallet_id,
-                        txid: record.txid,
+                        txid,
                         instant_lock: lock.clone(),
                         balance,
                         account_balances: account_balances.clone(),
-                    };
-                    self.emit_event(event);
+                    });
                 }
             }
         }

key-wallet/src/managed_account/managed_account_ref.rs

@@ -423,6 +423,34 @@ 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 late funding output to its owning account, using known spender templates.
+    /// Always empty for the [`Keys`](Self::Keys) variant.
+    pub(crate) fn attribute_spent_input(
+        &mut self,
+        outpoint: &OutPoint,
+        value: u64,
+        address: &Address,
+        spenders: &[TransactionRecord],
+    ) -> Vec<TransactionRecord> {
+        match self {
+            ManagedAccountRefMut::Funds(a) => {
+                a.attribute_spent_input(outpoint, value, address, spenders)
+            }
+            ManagedAccountRefMut::Keys(_) => Vec::new(),
+        }
+    }
 }
 
 /// Owned managed core account, either funds-bearing or keys-only.

key-wallet/src/managed_account/managed_core_funds_account.rs

@@ -71,6 +71,10 @@ pub struct ManagedCoreFundsAccount {
     /// re-establish which coins are spent.
     #[cfg_attr(feature = "serde", serde(skip))]
     reservations: ReservationSet,
+    /// Late funding outputs awaiting attribution to account-local spender records.
+    /// Drained at wallet scope before returning; never persisted.
+    #[cfg_attr(feature = "serde", serde(skip))]
+    born_spent_outputs: Vec<(OutPoint, u64, Address)>,
 }
 
 /// What [`ManagedCoreFundsAccount::apply_abandon`] removed from one account.
@@ -108,6 +112,7 @@ impl ManagedCoreFundsAccount {
             spent_outpoints: HashSet::new(),
             spent_before_funded: BTreeMap::new(),
             reservations: ReservationSet::default(),
+            born_spent_outputs: Vec::new(),
         }
     }
 
@@ -136,6 +141,7 @@ impl ManagedCoreFundsAccount {
             spent_outpoints: HashSet::new(),
             spent_before_funded: BTreeMap::new(),
             reservations: ReservationSet::default(),
+            born_spent_outputs: Vec::new(),
         }
     }
 
@@ -329,6 +335,11 @@ impl ManagedCoreFundsAccount {
                                     outpoint = %outpoint,
                                     "Skipping UTXO already spent by previously processed transaction"
                                 );
+                                self.born_spent_outputs.push((
+                                    outpoint,
+                                    output.value,
+                                    addr.clone(),
+                                ));
                                 continue;
                             }
 
@@ -343,6 +354,11 @@ impl ManagedCoreFundsAccount {
                                     outpoint = %outpoint,
                                     "Skipping UTXO already observed spent in an earlier-processed block (#649)"
                                 );
+                                self.born_spent_outputs.push((
+                                    outpoint,
+                                    output.value,
+                                    addr.clone(),
+                                ));
                                 self.spent_before_funded.insert(
                                     outpoint,
                                     Utxo::new(
@@ -420,6 +436,92 @@ impl ManagedCoreFundsAccount {
         }
     }
 
+    /// Attribute a late funding output only to its owning account's spender slices.
+    pub(crate) fn attribute_spent_input(
+        &mut self,
+        outpoint: &OutPoint,
+        value: u64,
+        address: &Address,
+        spenders: &[TransactionRecord],
+    ) -> Vec<TransactionRecord> {
+        if !self.contains_address(address) {
+            return Vec::new();
+        }
+        let mut corrected = Vec::new();
+        for template in spenders {
+            #[cfg(not(feature = "keep-finalized-transactions"))]
+            if self.keys.transaction_is_finalized(&template.txid) {
+                continue;
+            }
+            let Some(input_index) = template
+                .transaction
+                .input
+                .iter()
+                .position(|input| &input.previous_output == outpoint)
+            else {
+                continue;
+            };
+            let mut record =
+                self.keys.transactions().get(&template.txid).cloned().unwrap_or_else(|| {
+                    let mut record = template.clone();
+                    record.account_type = self.keys.managed_account_type().to_account_type();
+                    record.input_details.clear();
+                    record.output_details.clear();
+                    record
+                });
+            if record.input_details.iter().any(|d| d.index == input_index as u32) {
+                continue;
+            }
+            record.input_details.push(InputDetail {
+                index: input_index as u32,
+                value,
+                address: address.clone(),
+            });
+            record.input_details.sort_by_key(|d| d.index);
+            for (index, output) in record.transaction.output.iter().enumerate() {
+                if record.output_details.iter().any(|detail| detail.index == index as u32) {
+                    continue;
+                }
+                let output_address =
+                    Address::from_script(&output.script_pubkey, self.keys.network()).ok();
+                let pool = output_address.as_ref().and_then(|addr| {
+                    self.managed_account_type()
+                        .address_pools()
+                        .into_iter()
+                        .find(|pool| pool.address_index(addr).is_some())
+                });
+                let role = match pool {
+                    Some(pool) if pool.pool_type == address_pool::AddressPoolType::Internal => {
+                        OutputRole::Change
+                    }
+                    Some(_) => OutputRole::Received,
+                    None if output.script_pubkey.is_provably_unspendable() => {
+                        OutputRole::Unspendable
+                    }
+                    None => OutputRole::Sent,
+                };
+                record.output_details.push(OutputDetail {
+                    index: index as u32,
+                    value: output.value,
+                    address: output_address,
+                    role,
+                });
+            }
+            record.output_details.sort_by_key(|detail| detail.index);
+            record.recompute_net_and_direction();
+            self.keys.transactions_mut().insert(record.txid, record.clone());
+            self.spent_outpoints.insert(*outpoint);
+            corrected.push(record);
+        }
+        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
     /// surviving record still claims.
     ///
@@ -1337,6 +1439,7 @@ impl<'de> Deserialize<'de> for ManagedCoreFundsAccount {
             spent_outpoints,
             spent_before_funded: helper.spent_before_funded,
             reservations: ReservationSet::default(),
+            born_spent_outputs: Vec::new(),
         })
     }
 }