fix(wallet-storage)!: reconcile historical AssetUnlock identities

packages/rs-platform-wallet-storage/src/sqlite/error.rs

@@ -418,6 +418,16 @@ pub enum WalletStorageError {
         blob_height: Option<u32>,
     },
 
+    /// Aliases of one canonical transaction disagree about their confirmed block.
+    #[error(
+        "history aliases for {txid} in wallet {} have conflicting contexts",
+        hex::encode(wallet_id)
+    )]
+    HistoryAliasContextConflict {
+        wallet_id: [u8; 32],
+        txid: dashcore::Txid,
+    },
+
     /// An incoming transaction record reuses a stored txid with a body whose
     /// txid-committed content differs (witness-only differences are not a
     /// conflict); neither copy is trusted to replace the other.
@@ -786,6 +796,7 @@ impl WalletStorageError {
             | Self::AssetLockEntryMismatch { .. }
             | Self::AssetLockStatusMismatch { .. }
             | Self::CoreTransactionEntryMismatch { .. }
+            | Self::HistoryAliasContextConflict { .. }
             | Self::TransactionBodyConflict { .. }
             | Self::UnknownWalletNetwork { .. }
             | Self::NetAmountOverflow { .. }
@@ -836,9 +847,9 @@ impl WalletStorageError {
             Self::IdentityKeyWalletMismatch { .. } => PersistenceErrorKind::Constraint,
             // History invariants checked in Rust on the write path: the incoming
             // record contradicts stored history, so the data is wrong, not the engine.
-            Self::TransactionBodyConflict { .. } | Self::NetAmountOverflow { .. } => {
-                PersistenceErrorKind::Constraint
-            }
+            Self::HistoryAliasContextConflict { .. }
+            | Self::TransactionBodyConflict { .. }
+            | Self::NetAmountOverflow { .. } => PersistenceErrorKind::Constraint,
             // Refinery surfaces FK / constraint problems through rusqlite;
             // if that path leaks through here the typed variant lives in
             // `Self::Migration`, which we leave as `Fatal` since a
@@ -980,6 +991,7 @@ impl WalletStorageError {
             Self::AssetLockEntryMismatch { .. } => "asset_lock_entry_mismatch",
             Self::AssetLockStatusMismatch { .. } => "asset_lock_status_mismatch",
             Self::CoreTransactionEntryMismatch { .. } => "core_transaction_entry_mismatch",
+            Self::HistoryAliasContextConflict { .. } => "history_alias_context_conflict",
             Self::TransactionBodyConflict { .. } => "transaction_body_conflict",
             Self::UnknownWalletNetwork { .. } => "unknown_wallet_network",
             Self::NetAmountOverflow { .. } => "net_amount_overflow",

packages/rs-platform-wallet-storage/src/sqlite/load_ctx.rs

@@ -329,6 +329,12 @@ impl LoadCtx {
         }
     }
 
+    /// Recheck an already-counted snapshot under the same policy without
+    /// adding its inconsistencies to the load's original tally again.
+    pub(crate) fn for_recheck(&self) -> Self {
+        Self::new(self.policy)
+    }
+
     /// Context that aborts the load on any inconsistency. A production
     /// load takes its policy from the config; this is the shorthand for a
     /// caller driving a reader directly.

packages/rs-platform-wallet-storage/src/sqlite/persister.rs

@@ -1620,13 +1620,17 @@ impl PlatformWalletPersistence for SqlitePersister {
         Option<key_wallet::managed_account::transaction_record::TransactionRecord>,
         PersistenceError,
     > {
-        let conn = self.conn().map_err(PersistenceError::from)?;
+        let mut conn = self.conn().map_err(PersistenceError::from)?;
         // Tally deliberately dropped — see `last_load_degradation`. The
         // policy still decides whether drift is fatal, and `tolerate` still
         // logs it.
         let ctx = LoadCtx::new(self.config.load_policy);
-        schema::core_state::get_tx_record(&conn, &wallet_id, txid, &ctx)
-            .map_err(PersistenceError::from)
+        // Alias evidence and its active canonical record must belong to one
+        // SQLite snapshot, including when another connection writes the file.
+        let tx = conn.transaction().map_err(WalletStorageError::from)?;
+        let record = schema::core_history_aliases::get_record(&tx, &wallet_id, txid, &ctx)?;
+        tx.commit().map_err(WalletStorageError::from)?;
+        Ok(record)
     }
 
     /// Served from the `dpns_name_states` table this persister already
@@ -1720,9 +1724,11 @@ fn load_wallet_snapshot(
 
     let account_manifest =
         schema::accounts::load_state(conn, &wallet_id, ctx).map_err(PersistenceError::from)?;
-    let (core_state, utxo_accounts) =
+    let (mut core_state, utxo_accounts) =
         schema::core_state::load_state(conn, &wallet_id, network, ctx)
             .map_err(PersistenceError::from)?;
+    schema::core_history_aliases::project_records(conn, &wallet_id, &mut core_state.records, ctx)
+        .map_err(PersistenceError::from)?;
     // Pre-keyed rehydration: each `ManagedIdentity` leaves the loader
     // already carrying its own public keys + contact state (matching
     // the FFI persister), so signing works immediately post-load

packages/rs-platform-wallet-storage/src/sqlite/schema/core_history.rs

@@ -12,7 +12,7 @@ use platform_wallet::wallet::platform_wallet::WalletId;
 use rusqlite::{params, Connection, OptionalExtension, Transaction};
 
 use super::accounts::DASHPAY_EXTERNAL_LABEL;
-use super::{blob, core_state, wallets};
+use super::{blob, core_history_aliases, core_state, wallets};
 use crate::sqlite::error::WalletStorageError;
 use crate::sqlite::load_ctx::LoadCtx;
 use crate::sqlite::util::safe_cast::i64_to_u64;
@@ -23,10 +23,20 @@ pub(super) fn preserve_known_details(
     wallet_id: &WalletId,
     incoming: &TransactionRecord,
 ) -> Result<TransactionRecord, WalletStorageError> {
-    let mut merged = incoming.clone();
+    let merged = incoming.clone();
     let Some(previous) = prior_record(tx, wallet_id, &incoming.txid)? else {
         return Ok(merged);
     };
+    merge_known_details(wallet_id, &previous, incoming)
+}
+
+/// Merge accounting evidence without assigning identity or finality policy.
+pub(super) fn merge_known_details(
+    wallet_id: &WalletId,
+    previous: &TransactionRecord,
+    incoming: &TransactionRecord,
+) -> Result<TransactionRecord, WalletStorageError> {
+    let mut merged = incoming.clone();
     // Compare only txid-committed content: a peer may attach BIP144 witnesses
     // to a known txid, and that must neither conflict nor replace the stored body.
     if previous.transaction.txid() != incoming.transaction.txid() {
@@ -35,18 +45,20 @@ pub(super) fn preserve_known_details(
             txid: incoming.txid,
         });
     }
-    merged.transaction = previous.transaction;
+    merged.transaction = previous.transaction.clone();
     let mut inputs: BTreeMap<_, _> = previous
         .input_details
-        .into_iter()
+        .iter()
+        .cloned()
         .map(|d| (d.index, d))
         .collect();
     for detail in &incoming.input_details {
         inputs.insert(detail.index, detail.clone());
     }
     let mut outputs: BTreeMap<_, _> = previous
         .output_details
-        .into_iter()
+        .iter()
+        .cloned()
         .map(|d| (d.index, d))
         .collect();
     for detail in &incoming.output_details {
@@ -102,15 +114,39 @@ pub(crate) fn apply(
     let mut consumers = tx.prepare_cached(
         "SELECT txid FROM core_transaction_inputs WHERE wallet_id = ?1 AND outpoint = ?2",
     )?;
-    for utxo in cs.new_utxos.iter().chain(&cs.spent_utxos) {
-        affected.insert(utxo.outpoint.txid);
-        let mut rows = consumers.query(params![
-            wallet_id.as_slice(),
-            blob::encode_outpoint(&utxo.outpoint)?
-        ])?;
-        while let Some(row) = rows.next()? {
-            let bytes: Vec<u8> = row.get(0)?;
-            affected.insert(Txid::from_slice(&bytes)?);
+    let mut funding: Vec<_> = cs
+        .new_utxos
+        .iter()
+        .chain(&cs.spent_utxos)
+        .map(|utxo| utxo.outpoint)
+        .collect();
+    // Adoption can connect already-stored coins to consumers without a new
+    // coin delivery. Ordinary records keep their existing scheduling policy.
+    for record in cs
+        .records
+        .iter()
+        .filter(|record| core_history_aliases::is_unlock(record))
+    {
+        funding.extend(
+            record
+                .transaction
+                .output
+                .iter()
+                .enumerate()
+                .map(|(index, _)| OutPoint::new(record.txid, index as u32)),
+        );
+    }
+    for changed in funding {
+        for outpoint in core_history_aliases::outpoints(tx, wallet_id, &changed)? {
+            affected.insert(outpoint.txid);
+            let mut rows = consumers.query(params![
+                wallet_id.as_slice(),
+                blob::encode_outpoint(&outpoint)?
+            ])?;
+            while let Some(row) = rows.next()? {
+                let bytes: Vec<u8> = row.get(0)?;
+                affected.insert(Txid::from_slice(&bytes)?);
+            }
         }
     }
     if affected.is_empty() {
@@ -143,6 +179,46 @@ fn owned_output(
     wallet_id: &WalletId,
     outpoint: &OutPoint,
     network: dashcore::Network,
+) -> Result<Option<(u64, Address)>, WalletStorageError> {
+    // Exact physical evidence keeps its existing semantics. Only a missing
+    // exact coin can use a validated active history family's equivalent key.
+    if let Some(output) = physical_owned_output(tx, wallet_id, outpoint, network)? {
+        return Ok(Some(output));
+    }
+    let equivalents = core_history_aliases::outpoints(tx, wallet_id, outpoint)?;
+    for alternative in equivalents.into_iter().filter(|other| other != outpoint) {
+        if let Some((value, address)) = physical_owned_output(tx, wallet_id, &alternative, network)?
+        {
+            let record = core_history_aliases::get_record(
+                tx,
+                wallet_id,
+                &outpoint.txid,
+                &LoadCtx::strict(),
+            )?
+            .ok_or_else(|| {
+                WalletStorageError::blob_decode(
+                    "active history alias disappeared during accounting",
+                )
+            })?;
+            let output = record.transaction.output.get(outpoint.vout as usize);
+            if !output.is_some_and(|output| {
+                output.value == value && output.script_pubkey == address.script_pubkey()
+            }) {
+                return Err(WalletStorageError::blob_decode(
+                    "alias accounting output disagrees with canonical transaction body",
+                ));
+            }
+            return Ok(Some((value, address)));
+        }
+    }
+    Ok(None)
+}
+
+fn physical_owned_output(
+    tx: &Transaction<'_>,
+    wallet_id: &WalletId,
+    outpoint: &OutPoint,
+    network: dashcore::Network,
 ) -> Result<Option<(u64, Address)>, WalletStorageError> {
     let encoded = blob::encode_outpoint(outpoint)?;
     let key = params![wallet_id.as_slice(), encoded];
@@ -264,23 +340,9 @@ fn repair_record(
     }
     record.input_details = inputs.into_values().collect();
     record.output_details = outputs.into_values().collect();
-    // The live projection computes the same accounting, so repair never
-    // flips a row it just wrote; only the overflow policy differs.
-    let (net, direction) = wallet_accounting(&record);
-    record.net_amount = match i64::try_from(net) {
-        Ok(net) => net,
-        // Nothing authenticates a mempool transaction: this is peer-supplied data,
-        // not corruption. Keep its stored accounting until representable evidence arrives.
-        Err(_) if matches!(record.context, TransactionContext::Mempool) => return Ok(()),
-        Err(_) => {
-            return Err(WalletStorageError::NetAmountOverflow {
-                wallet_id: *wallet_id,
-                txid: *txid,
-                value: net,
-            })
-        }
-    };
-    record.direction = direction;
+    if !recompute_accounting(wallet_id, &mut record)? {
+        return Ok(());
+    }
     let repaired = blob::encode(&record)?;
     if repaired != original {
         // Append-only: the first pre-repair blob is kept verbatim and never
@@ -303,6 +365,35 @@ fn repair_record(
     Ok(())
 }
 
+/// Apply the existing history accounting policy to a merged snapshot. Returns
+/// false when absent or unrepresentable mempool evidence cannot replace it.
+pub(super) fn recompute_accounting(
+    wallet_id: &WalletId,
+    record: &mut TransactionRecord,
+) -> Result<bool, WalletStorageError> {
+    if record.input_details.is_empty() && record.output_details.is_empty() {
+        return Ok(false);
+    }
+    // The live projection computes the same accounting, so repair never
+    // flips a row it just wrote; only the overflow policy differs.
+    let (net, direction) = wallet_accounting(record);
+    record.net_amount = match i64::try_from(net) {
+        Ok(net) => net,
+        // Nothing authenticates a mempool transaction: this is peer-supplied data,
+        // not corruption. Keep its stored accounting until representable evidence arrives.
+        Err(_) if matches!(record.context, TransactionContext::Mempool) => return Ok(false),
+        Err(_) => {
+            return Err(WalletStorageError::NetAmountOverflow {
+                wallet_id: *wallet_id,
+                txid: record.txid,
+                value: net,
+            })
+        }
+    };
+    record.direction = direction;
+    Ok(true)
+}
+
 #[cfg(test)]
 mod tests {
     use dashcore::TxOut;