fix(platform-wallet): finalize reconstructed asset locks as RecoveredFromChain, in-session

.codecov.yml

@@ -108,6 +108,11 @@ ignore:
   - "packages/rs-dpp/src/state_transition/state_transitions/document/batch_transition/**/v1_methods.rs"
   # Platform wallet — requires Core wallet integration, not unit-testable
   - "packages/rs-platform-wallet/src/**"
+  # Platform wallet storage — its tests run in the wallet fast-path
+  # workflow (tests-rs-wallet.yml), which intentionally omits coverage
+  # upload, so codecov never receives data for this crate on
+  # wallet-scoped PRs and patch status would fail spuriously
+  - "packages/rs-platform-wallet-storage/**"
   # Proof-verifier response types and unproved handling
   - "packages/rs-drive-proof-verifier/src/types.rs"
   - "packages/rs-drive-proof-verifier/src/unproved.rs"

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

@@ -27,6 +27,20 @@ pub fn apply(
     cs: &AssetLockChangeSet,
 ) -> Result<(), WalletStorageError> {
     if !cs.asset_locks.is_empty() {
+        // The upsert's WHERE clause enforces the one terminal lifecycle
+        // rule: a stored `consumed` row is never overwritten by a
+        // non-consumed snapshot. Racing writers persist through
+        // different paths (the wallet-event adapter's batched drain vs
+        // the live flows' synchronous changeset queue), so a stale
+        // reconstruction/enrichment snapshot can land AFTER the
+        // consumption write — this guard makes that arrival order
+        // immaterial. Every other transition is deliberately
+        // last-write-wins: non-terminal statuses move both ways (live
+        // advances overwrite `recovered_from_chain`, defensive resumes
+        // re-enter `broadcast`), so terminality is the only ordering
+        // the store can enforce without vetoing legitimate writes.
+        // `AssetLockChangeSet::merge` applies the same rule when
+        // batches fold before reaching the store.
         let mut stmt = tx.prepare_cached(
             "INSERT INTO asset_locks \
                 (wallet_id, outpoint, status, account_index, identity_index, amount_duffs, lifecycle_blob) \
@@ -36,7 +50,8 @@ pub fn apply(
                 account_index = excluded.account_index, \
                 identity_index = excluded.identity_index, \
                 amount_duffs = excluded.amount_duffs, \
-                lifecycle_blob = excluded.lifecycle_blob",
+                lifecycle_blob = excluded.lifecycle_blob \
+             WHERE asset_locks.status != 'consumed' OR excluded.status = 'consumed'",
         )?;
         for (op, entry) in &cs.asset_locks {
             let op_bytes = blob::encode_outpoint(op)?;

packages/rs-platform-wallet-storage/tests/sqlite_persist_roundtrip.rs

@@ -522,6 +522,106 @@ fn tc010b_recovered_from_chain_lock_roundtrip() {
     drop(tmp);
 }
 
+/// TC-010c: the store-order race between the wallet-event adapter and
+/// the live flows, applied in the exact adversarial order. A stale
+/// reconstruction/enrichment snapshot (`RecoveredFromChain`) that the
+/// adapter's batched drain persists AFTER the live flow's synchronous
+/// `Consumed` write must NOT regress the durable row — `Consumed` is
+/// terminal and the upsert's WHERE guard rejects the late arrival.
+/// Every other direction stays last-write-wins, including `Consumed`
+/// landing over `RecoveredFromChain`.
+#[test]
+fn tc010c_stale_recovery_snapshot_cannot_regress_consumed_row() {
+    use dashcore::hashes::Hash;
+    use dashcore::{OutPoint, Transaction, Txid};
+    use dpp::identity::state_transition::asset_lock_proof::chain::ChainAssetLockProof;
+    use key_wallet::wallet::managed_wallet_info::asset_lock_builder::AssetLockFundingType;
+    use platform_wallet::changeset::{AssetLockChangeSet, AssetLockEntry};
+    use platform_wallet::wallet::asset_lock::tracked::AssetLockStatus;
+
+    let entry_with = |outpoint: OutPoint, status: AssetLockStatus| AssetLockEntry {
+        out_point: outpoint,
+        transaction: Transaction {
+            version: 3,
+            lock_time: 0,
+            input: vec![],
+            output: vec![],
+            special_transaction_payload: None,
+        },
+        account_index: 0,
+        funding_type: AssetLockFundingType::IdentityRegistration,
+        identity_index: 0,
+        amount_duffs: 1_000_000,
+        status: status.clone(),
+        proof: match status {
+            AssetLockStatus::RecoveredFromChain => {
+                Some(dpp::prelude::AssetLockProof::Chain(ChainAssetLockProof {
+                    core_chain_locked_height: 900,
+                    out_point: outpoint,
+                }))
+            }
+            _ => None,
+        },
+    };
+    let store_one = |persister: &SqlitePersister, w, outpoint, status| {
+        let mut locks = AssetLockChangeSet::default();
+        locks
+            .asset_locks
+            .insert(outpoint, entry_with(outpoint, status));
+        persister
+            .store(
+                w,
+                PlatformWalletChangeSet {
+                    asset_locks: Some(locks),
+                    ..Default::default()
+                },
+            )
+            .unwrap();
+    };
+
+    let (persister, tmp, path) = fresh_persister();
+    let w = wid(0xFA);
+    ensure_wallet_meta(&persister, &w);
+
+    // Outpoint A: live lock consumed, THEN the stale recovery snapshot
+    // arrives (the adapter drained its batch after the live write).
+    let a = OutPoint {
+        txid: Txid::from_byte_array([0x51; 32]),
+        vout: 0,
+    };
+    store_one(&persister, w, a, AssetLockStatus::Broadcast);
+    store_one(&persister, w, a, AssetLockStatus::Consumed);
+    store_one(&persister, w, a, AssetLockStatus::RecoveredFromChain);
+
+    // Outpoint B: the legitimate direction — a recovered lock is
+    // explicitly resumed and consumed; the terminal write must land.
+    let b = OutPoint {
+        txid: Txid::from_byte_array([0x52; 32]),
+        vout: 0,
+    };
+    store_one(&persister, w, b, AssetLockStatus::RecoveredFromChain);
+    store_one(&persister, w, b, AssetLockStatus::Consumed);
+
+    drop(persister);
+    let p2 = SqlitePersister::open(SqlitePersisterConfig::new(&path)).unwrap();
+    let bucketed = platform_wallet_storage::sqlite::schema::asset_locks::load_state(
+        &p2.lock_conn_for_test(),
+        &w,
+    )
+    .unwrap();
+    assert_eq!(
+        bucketed[&0][&a].status,
+        AssetLockStatus::Consumed,
+        "a stale RecoveredFromChain snapshot landing after Consumed must be rejected"
+    );
+    assert_eq!(
+        bucketed[&0][&b].status,
+        AssetLockStatus::Consumed,
+        "Consumed must still land over RecoveredFromChain"
+    );
+    drop(tmp);
+}
+
 /// TC-012: DashPay profile + payment overlay round-trip through the
 /// dashpay_* tables via bincode-serde blobs.
 #[test]

packages/rs-platform-wallet/src/changeset/changeset.rs

@@ -953,8 +953,29 @@ pub struct AssetLockEntry {
 
 impl Merge for AssetLockChangeSet {
     fn merge(&mut self, other: Self) {
-        // Last write wins — later status is higher finality.
-        self.asset_locks.extend(other.asset_locks);
+        // Last write wins, with ONE lifecycle exception: `Consumed` is
+        // the terminal state, so a non-Consumed snapshot never replaces
+        // a Consumed one. Writers race here — the wallet-event
+        // adapter's batched drain can fold (or persist) a stale
+        // reconstruction/enrichment snapshot AFTER the live flow's
+        // synchronous consumption write — and every non-terminal
+        // transition is legitimately bidirectional (a live advance
+        // overwrites `RecoveredFromChain`, a defensive resume
+        // re-enters `Broadcast`), so terminality is the only ordering
+        // the merge can enforce without vetoing real transitions. The
+        // durable stores apply the same rule (sqlite upsert guard,
+        // swift-sdk `persistAssetLocks`), making the store order of
+        // racing snapshots immaterial.
+        for (out_point, entry) in other.asset_locks {
+            if entry.status != AssetLockStatus::Consumed {
+                if let Some(existing) = self.asset_locks.get(&out_point) {
+                    if existing.status == AssetLockStatus::Consumed {
+                        continue;
+                    }
+                }
+            }
+            self.asset_locks.insert(out_point, entry);
+        }
         self.removed.extend(other.removed);
     }
 
@@ -1663,6 +1684,76 @@ mod tests {
         assert!(cs.is_empty());
     }
 
+    /// Asset-lock merge is last-write-wins EXCEPT for the Consumed
+    /// terminal: when the wallet-event adapter's batched drain folds a
+    /// stale reconstruction/enrichment snapshot after (or before) the
+    /// live flow's consumption write, the fold must never regress
+    /// Consumed — while Consumed itself must still land over anything.
+    #[test]
+    fn asset_lock_merge_never_regresses_consumed() {
+        use dashcore::hashes::Hash;
+        use key_wallet::wallet::managed_wallet_info::asset_lock_builder::AssetLockFundingType;
+
+        let outpoint = OutPoint {
+            txid: Txid::from_byte_array([0x61; 32]),
+            vout: 0,
+        };
+        let entry_with = |status: AssetLockStatus| AssetLockEntry {
+            out_point: outpoint,
+            transaction: Transaction {
+                version: 3,
+                lock_time: 0,
+                input: vec![],
+                output: vec![],
+                special_transaction_payload: None,
+            },
+            account_index: 0,
+            funding_type: AssetLockFundingType::IdentityRegistration,
+            identity_index: 0,
+            amount_duffs: 1,
+            status,
+            proof: None,
+        };
+        let cs_with = |status: AssetLockStatus| {
+            let mut cs = AssetLockChangeSet::default();
+            cs.asset_locks.insert(outpoint, entry_with(status));
+            cs
+        };
+
+        // Stale recovery snapshot folded AFTER the consumption write.
+        let mut folded = cs_with(AssetLockStatus::Consumed);
+        folded.merge(cs_with(AssetLockStatus::RecoveredFromChain));
+        assert_eq!(
+            folded.asset_locks[&outpoint].status,
+            AssetLockStatus::Consumed,
+            "a non-Consumed snapshot must not replace the Consumed terminal"
+        );
+
+        // The legitimate direction still lands.
+        let mut folded = cs_with(AssetLockStatus::RecoveredFromChain);
+        folded.merge(cs_with(AssetLockStatus::Consumed));
+        assert_eq!(
+            folded.asset_locks[&outpoint].status,
+            AssetLockStatus::Consumed
+        );
+
+        // Non-terminal transitions stay last-write-wins in both
+        // directions (live advances overwrite RecoveredFromChain, and
+        // enrichment overwrites Broadcast).
+        let mut folded = cs_with(AssetLockStatus::RecoveredFromChain);
+        folded.merge(cs_with(AssetLockStatus::ChainLocked));
+        assert_eq!(
+            folded.asset_locks[&outpoint].status,
+            AssetLockStatus::ChainLocked
+        );
+        let mut folded = cs_with(AssetLockStatus::Broadcast);
+        folded.merge(cs_with(AssetLockStatus::RecoveredFromChain));
+        assert_eq!(
+            folded.asset_locks[&outpoint].status,
+            AssetLockStatus::RecoveredFromChain
+        );
+    }
+
     #[test]
     fn contested_dpns_merge_replaces_canonical_snapshot_and_allows_empty() {
         let id = Identifier::from([0x51; 32]);