fix(platform-wallet): data-integrity follow-ups from the #3990 sync review (#4008)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: QuantumExplorer

packages/rs-platform-wallet/src/wallet/identity/network/register_from_addresses.rs

@@ -132,21 +132,45 @@ impl IdentityWallet {
         }
         let identity = registered_identity;
 
-        // Step 3: Add the identity to the local manager (with its HD
-        // index) so subsequent operations route through it.
+        // Step 3 (best-effort): add the identity to the local manager
+        // (with its HD index) so subsequent operations route through it.
+        //
+        // Platform has ALREADY accepted the registration, so a local
+        // persistence failure must NOT propagate as `Err` — that would
+        // suppress the `(identity, address_infos)` return the caller
+        // needs to reconcile the spent funding-address balances, leaving
+        // them stale even though the identity exists on chain. A missed
+        // local add self-heals on the next identity re-sync. This mirrors
+        // the sibling `transfer_credits_to_addresses` / top-up flows,
+        // which likewise treat post-acceptance local bookkeeping as
+        // best-effort.
         {
             let mut wm = self.wallet_manager.write().await;
-            let info = wm.get_wallet_info_mut(&self.wallet_id).ok_or_else(|| {
-                crate::error::PlatformWalletError::WalletNotFound(
-                    "Wallet info not found in wallet manager".to_string(),
-                )
-            })?;
-            info.identity_manager.add_identity(
-                identity.clone(),
-                identity_index,
-                self.wallet_id,
-                &self.persister,
-            )?;
+            match wm.get_wallet_info_mut(&self.wallet_id) {
+                Some(info) => {
+                    if let Err(e) = info.identity_manager.add_identity(
+                        identity.clone(),
+                        identity_index,
+                        self.wallet_id,
+                        &self.persister,
+                    ) {
+                        tracing::warn!(
+                            error = %e,
+                            identity_id = %identity.id(),
+                            "register_from_addresses: identity registered on Platform but \
+                             local add_identity failed; returning the registered identity \
+                             anyway so address balances can reconcile"
+                        );
+                    }
+                }
+                None => {
+                    tracing::warn!(
+                        identity_id = %identity.id(),
+                        "register_from_addresses: identity registered on Platform but wallet \
+                         info was not found locally; skipping local persistence"
+                    );
+                }
+            }
         }
 
         // The spent platform-address balances are reconciled by the

packages/rs-platform-wallet/src/wallet/platform_addresses/fund_from_asset_lock.rs

@@ -307,11 +307,40 @@ impl PlatformAddressWallet {
         // forcing the next pass to full-scan-reconcile — the automated
         // equivalent of the manual Sync-tab "Clear" + "Sync Now".
         let credited_outputs = super::credited_outputs_set(addresses.keys());
-        let cs = self
-            .reconcile_address_infos(&address_infos, &credited_outputs, "fund from asset lock")
+        // Use the persistence-reporting variant: marking the lock
+        // `Consumed` below is irreversible, so it MUST be gated on the
+        // reconciled balances actually reaching disk. `persisted` is
+        // false ONLY when the in-memory balances were updated but the
+        // durable write failed — exactly the case where a Consumed lock
+        // would pair with stale rows and under-budget the next spend
+        // after a restart.
+        let (cs, persisted) = self
+            .reconcile_address_infos_with_persistence(
+                &address_infos,
+                &credited_outputs,
+                "fund from asset lock",
+            )
             .await;
 
         if let Some(out_point) = tracked_out_point {
+            if !persisted {
+                // The proof-attested balances were applied in memory but
+                // did not reach disk. Leave the lock non-Consumed: it
+                // stays in the Resumable Funding list, and a user Resume
+                // gets Platform's deterministic "lock already consumed"
+                // rejection — the same benign recovery path as a failed
+                // consume below — while the next platform-address sync
+                // repairs the stale rows. Consuming here would strand the
+                // lock as Consumed over durable balances that under-report
+                // the credit.
+                tracing::error!(
+                    outpoint = %out_point,
+                    "skipping consume_asset_lock: the reconciled balance changeset \
+                     was not durably stored; the lock stays non-Consumed (Resumable) \
+                     rather than pairing a Consumed lock with stale balance rows on disk"
+                );
+                return Ok(cs);
+            }
             // Platform DID accept the top-up — propagating an Err
             // here would misreport the protocol outcome, since the
             // caller's recipient(s) already have credits attested

packages/rs-platform-wallet/src/wallet/platform_addresses/provider.rs

@@ -1061,29 +1061,63 @@ impl PlatformPaymentAddressProvider {
                     }
                 }
             }
+            // Derivation-index conflict: `entry.address` isn't yet in the
+            // bijection, but its `address_index` already maps to a
+            // DIFFERENT address. Detected BEFORE the `found` mutation
+            // because a conflicting credit must not be half-applied.
+            let index_conflict = state.addresses.get_by_right(&entry.address).is_none()
+                && state.addresses.contains_left(&entry.address_index);
+
+            if index_conflict && !is_removal {
+                // A credit under a conflicting index: dropping it outright
+                // is the only safe response. Inserting the pairing would
+                // evict the existing one (`BiBTreeMap::insert` drops
+                // conflicting pairs, orphaning the other address's `found`
+                // entry); NOT inserting it would commit a `found` balance
+                // downstream can't pair with a derivation index, so
+                // `current_balances` couldn't round-trip the committed
+                // seed. The address stays unresolved until `initialize` /
+                // `add_provider` re-snapshots the account set.
+                tracing::error!(
+                    account_index = entry.account_index,
+                    address_index = entry.address_index,
+                    address = %entry.address,
+                    "commit_reconciliation: derivation index already maps to a \
+                     different address — dropping the credit reconciliation entry \
+                     to avoid corrupting the bijection"
+                );
+                continue;
+            }
+
             if is_removal {
                 state.found.remove(&entry.address);
             } else {
                 state.found.insert(entry.address, entry.funds);
             }
-            // Merge pool-resolved addresses into the bijection so
-            // `current_balances` can pair the fresh funds with a
-            // derivation index. Never overwrite an existing pairing —
-            // `BiBTreeMap::insert` evicts conflicting pairs, which
-            // would orphan another address's `found` entry.
-            if state.addresses.get_by_right(&entry.address).is_none() {
-                if state.addresses.contains_left(&entry.address_index) {
-                    tracing::error!(
-                        account_index = entry.account_index,
-                        address_index = entry.address_index,
-                        address = %entry.address,
-                        "commit_reconciliation: derivation index already \
-                         maps to a different address — state drift; \
-                         leaving the bijection untouched"
-                    );
-                } else {
-                    state.addresses.insert(entry.address_index, entry.address);
-                }
+
+            if index_conflict {
+                // A removal under a conflicting index: the credit case
+                // already `continue`d above, so this is a zero-out. It
+                // MUST still zero `found` (done) and be emitted (below) so
+                // the durable persister writes the zero — otherwise a
+                // stale persisted balance for this address resurrects
+                // after restart. Only the bijection merge is skipped, so
+                // the pre-existing `(index -> other address)` pairing
+                // survives.
+                tracing::warn!(
+                    account_index = entry.account_index,
+                    address_index = entry.address_index,
+                    address = %entry.address,
+                    "commit_reconciliation: derivation index already maps to a \
+                     different address — applying the removal without touching \
+                     the bijection so a stale persisted balance can't resurrect"
+                );
+            } else if state.addresses.get_by_right(&entry.address).is_none() {
+                // Merge pool-resolved addresses into the bijection so
+                // `current_balances` can pair the fresh funds with a
+                // derivation index. The conflict guard above ruled out an
+                // eviction, so this insert is always a fresh pairing.
+                state.addresses.insert(entry.address_index, entry.address);
             }
             outcome.entries.push(entry);
         }
@@ -1858,6 +1892,126 @@ mod tests {
         assert_eq!(seed[0].2, funds(700, 5));
     }
 
+    /// A zero-funds removal that pool-resolves to an `address_index`
+    /// already paired to a DIFFERENT address in the bijection must still
+    /// zero the in-memory `found` row AND be emitted downstream —
+    /// otherwise a durable persister row for the removed address would
+    /// resurrect after restart. The bijection stays untouched so the
+    /// pre-existing `(index -> other addr)` pairing isn't evicted.
+    #[test]
+    fn commit_reconciliation_index_conflict_still_emits_removal() {
+        let owned = p2pkh(0x11);
+        let conflicting = p2pkh(0x77);
+        let mut provider = provider_with_one_funded_address(owned, funds(700, 3));
+        {
+            let state = provider
+                .per_wallet
+                .get_mut(&WALLET)
+                .and_then(|s| s.get_mut(&ACCOUNT))
+                .expect("account state present");
+            // Pin `conflicting` at index 5 with a balance we must protect.
+            state.insert_persisted_entry(5, conflicting, funds(200, 1));
+            // Stale `found` row for the address that will be removed,
+            // seeded at the SAME index 5 to force the conflict.
+            state.found.insert(p2pkh(0x22), funds(999, 4));
+        }
+
+        // The removed address pool-resolves to index 5 — a DIFFERENT
+        // address than the bijection holds there.
+        let removed = p2pkh(0x22);
+        let removed_addr = PlatformAddress::P2pkh([0x22; 20]);
+        let mut pool_indexes = BTreeMap::new();
+        pool_indexes.insert(removed, (ACCOUNT, 5u32));
+
+        let mut address_infos = AddressInfos::new();
+        // Fully-consumed input: Drive elides the info → removal entry.
+        address_infos.insert(removed_addr, None);
+
+        let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &pool_indexes);
+
+        // The removal is emitted so the durable persister writes the zero.
+        assert_eq!(outcome.entries.len(), 1, "removal survives the guard");
+        assert_eq!(outcome.entries[0].address, removed);
+        assert_eq!(outcome.entries[0].funds, funds(0, 0));
+
+        let state = provider
+            .per_wallet
+            .get(&WALLET)
+            .and_then(|s| s.get(&ACCOUNT))
+            .expect("account state present");
+        // In-memory `found` for the removed address is dropped.
+        assert!(!state.found.contains_key(&removed));
+        // The bijection is unchanged — pre-existing pairing survives.
+        assert_eq!(
+            state.addresses.get_by_left(&5u32).copied(),
+            Some(conflicting)
+        );
+        assert!(state.addresses.get_by_right(&removed).is_none());
+        // The protected address's balance is untouched.
+        assert_eq!(state.found.get(&conflicting).copied(), Some(funds(200, 1)));
+    }
+
+    /// A CREDIT (non-zero funds) that pool-resolves to an already-taken
+    /// derivation index must be dropped outright: neither applied to
+    /// `found` nor emitted, and the bijection untouched. Committing it
+    /// would either evict the existing pairing or persist a seed
+    /// `current_balances` can't round-trip.
+    #[test]
+    fn commit_reconciliation_index_conflict_drops_credit() {
+        use dash_sdk::query_types::AddressInfo;
+
+        let owned = p2pkh(0x11);
+        let conflicting = p2pkh(0x77);
+        let mut provider = provider_with_one_funded_address(owned, funds(700, 3));
+        {
+            let state = provider
+                .per_wallet
+                .get_mut(&WALLET)
+                .and_then(|s| s.get_mut(&ACCOUNT))
+                .expect("account state present");
+            state.insert_persisted_entry(5, conflicting, funds(200, 1));
+        }
+
+        // A credit for a fresh address that pool-resolves to the taken
+        // index 5.
+        let credited = p2pkh(0x33);
+        let credited_addr = PlatformAddress::P2pkh([0x33; 20]);
+        let mut pool_indexes = BTreeMap::new();
+        pool_indexes.insert(credited, (ACCOUNT, 5u32));
+
+        let mut address_infos = AddressInfos::new();
+        address_infos.insert(
+            credited_addr,
+            Some(AddressInfo {
+                address: credited_addr,
+                nonce: 2,
+                balance: 5_000,
+            }),
+        );
+
+        let outcome = provider.commit_reconciliation(&WALLET, &address_infos, &pool_indexes);
+
+        assert!(
+            outcome.entries.is_empty(),
+            "a credit under a conflicting index is dropped, not emitted"
+        );
+        let state = provider
+            .per_wallet
+            .get(&WALLET)
+            .and_then(|s| s.get(&ACCOUNT))
+            .expect("account state present");
+        // `found` never gained the conflicting credit.
+        assert!(!state.found.contains_key(&credited));
+        // Bijection untouched: index 5 still → conflicting, and the
+        // credited address was not inserted.
+        assert_eq!(
+            state.addresses.get_by_left(&5u32).copied(),
+            Some(conflicting)
+        );
+        assert!(state.addresses.get_by_right(&credited).is_none());
+        assert_eq!(state.found.get(&conflicting).copied(), Some(funds(200, 1)));
+    }
+
     /// An entry identical to the committed seed is a no-op and is dropped
     /// to avoid persister churn.
     #[test]