fix(platform-wallet): address sync review feedback for #3990

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

@@ -132,21 +132,41 @@ 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 suppress the `(identity,
+        // address_infos)` return — the caller still needs it to
+        // reconcile the spent address balances, and the missed local
+        // add self-heals on the next identity re-sync.
         {
             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"
+                        );
+                    }
+                }
+                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

@@ -286,14 +286,33 @@ impl PlatformAddressWallet {
         // The seam persists before returning, and the persist MUST
         // happen before `consume_asset_lock` so we never have a
         // Consumed lock paired with a stale balance row on disk.
-        // Persistence errors are logged inside the seam rather than
-        // propagated: Platform already accepted the transition, and a
-        // persistence hiccup shouldn't mask that.
-        let cs = self
-            .reconcile_address_infos(&address_infos, "fund from asset lock")
+        // Persistence errors don't propagate as `Err` (Platform
+        // already accepted the transition, and a persistence hiccup
+        // shouldn't mask that), but they DO gate the consume below:
+        // marking the lock `Consumed` over stale durable rows would
+        // leave `auto_select_inputs` under-budgeting after a restart
+        // with no Resumable lock left to repair it from.
+        let (cs, persisted) = self
+            .reconcile_address_infos_with_persistence(&address_infos, "fund from asset lock")
             .await;
 
         if let Some(out_point) = tracked_out_point {
+            if !persisted {
+                // Keep the lock row non-Consumed: it stays visible 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.
+                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

@@ -999,29 +999,36 @@ impl PlatformPaymentAddressProvider {
                     }
                 }
             }
+            // Drop the entry outright when its derivation index is
+            // already paired with a DIFFERENT address — committing it
+            // would either evict that pairing (`BiBTreeMap::insert`
+            // drops conflicting pairs, orphaning the other address's
+            // `found` entry) or persist an address `current_balances`
+            // can't round-trip back out of the committed seed.
+            if state.addresses.get_by_right(&entry.address).is_none()
+                && 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; \
+                     dropping the reconciliation entry"
+                );
+                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.
+            // derivation index. The conflict guard above already
+            // ensured this never overwrites an existing pairing.
             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);
-                }
+                state.addresses.insert(entry.address_index, entry.address);
             }
             outcome.entries.push(entry);
         }

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

@@ -201,16 +201,39 @@ impl PlatformAddressWallet {
     ///
     /// Persistence errors are logged rather than propagated — Platform
     /// already accepted the transition, and a later sync reconciles.
+    /// Callers that must observe persistence (e.g. asset-lock funding,
+    /// which must not consume the lock over stale durable rows) use
+    /// [`reconcile_address_infos_with_persistence`] instead.
     ///
     /// [`PlatformPaymentAddressProvider::commit_reconciliation`]:
     /// super::provider::PlatformPaymentAddressProvider::commit_reconciliation
+    /// [`reconcile_address_infos_with_persistence`]:
+    /// Self::reconcile_address_infos_with_persistence
     pub async fn reconcile_address_infos(
         &self,
         address_infos: &AddressInfos,
         context: &'static str,
     ) -> crate::PlatformAddressChangeSet {
+        self.reconcile_address_infos_with_persistence(address_infos, context)
+            .await
+            .0
+    }
+
+    /// [`reconcile_address_infos`](Self::reconcile_address_infos), but
+    /// also reporting whether the durable rows now reflect the
+    /// transition: `true` when the changeset was stored (or nothing
+    /// new needed storing), `false` when the balances could not be
+    /// applied at all (no provider / no provider state / nothing
+    /// resolved to a wallet-owned slot) or the persist itself failed.
+    /// Callers with a persist-before-cleanup ordering invariant branch
+    /// on the flag; everyone else uses the plain variant.
+    pub(crate) async fn reconcile_address_infos_with_persistence(
+        &self,
+        address_infos: &AddressInfos,
+        context: &'static str,
+    ) -> (crate::PlatformAddressChangeSet, bool) {
         if address_infos.is_empty() {
-            return crate::PlatformAddressChangeSet::default();
+            return (crate::PlatformAddressChangeSet::default(), true);
         }
 
         let mut guard = self.provider.write().await;
@@ -222,7 +245,7 @@ impl PlatformAddressWallet {
                  provider for this wallet; local balances stay stale \
                  until the next platform-address sync"
             );
-            return crate::PlatformAddressChangeSet::default();
+            return (crate::PlatformAddressChangeSet::default(), false);
         };
         if provider.per_wallet_state(&self.wallet_id).is_none() {
             tracing::warn!(
@@ -232,7 +255,7 @@ impl PlatformAddressWallet {
                  provider state for this wallet; local balances stay \
                  stale until the next platform-address sync"
             );
-            return crate::PlatformAddressChangeSet::default();
+            return (crate::PlatformAddressChangeSet::default(), false);
         }
 
         // Live-pool fallback indexes for addresses derived since the last
@@ -273,7 +296,7 @@ impl PlatformAddressWallet {
                  address belongs to a third party; otherwise local \
                  balances stay stale until the next platform-address sync"
             );
-            return crate::PlatformAddressChangeSet::default();
+            return (crate::PlatformAddressChangeSet::default(), false);
         }
         if outcome.stale_skipped > 0 || outcome.unchanged_skipped > 0 {
             tracing::debug!(
@@ -286,7 +309,9 @@ impl PlatformAddressWallet {
             );
         }
         if outcome.entries.is_empty() {
-            return crate::PlatformAddressChangeSet::default();
+            // Everything the proof attested was already committed (or
+            // fresher) in the durable rows — nothing new to store.
+            return (crate::PlatformAddressChangeSet::default(), true);
         }
 
         // Apply the proof-attested balances to the managed accounts while
@@ -336,7 +361,9 @@ impl PlatformAddressWallet {
             addresses: outcome.entries,
             ..Default::default()
         };
+        let mut persisted = true;
         if let Err(e) = self.persister.store(cs.clone().into()) {
+            persisted = false;
             tracing::error!(
                 context,
                 error = %e,
@@ -346,7 +373,7 @@ impl PlatformAddressWallet {
             );
         }
         drop(guard);
-        cs
+        (cs, persisted)
     }
 
     /// Get the network from the SDK.
@@ -433,23 +460,23 @@ impl PlatformAddressWallet {
     ///
     /// Does NOT route through [`apply_sync_state`] — that helper's
     /// all-None early-return guard is meant for persisted-state replay
-    /// and is irrelevant here. The two locks are taken sequentially
-    /// (one released before the next is acquired), so there is no
-    /// nested-lock hazard; this mirrors the ordering rationale in
-    /// [`initialize_from_persisted`].
+    /// and is irrelevant here. Both clears run inside one provider
+    /// write critical section (provider → wallet-manager, the same
+    /// nesting order [`sync_balances`](Self::sync_balances) and the
+    /// reconciliation seam use), so a concurrent sync/reconciliation
+    /// pass can never interleave between the two steps and repopulate
+    /// the managed balances after only one store was cleared.
     pub async fn reset_sync_state(&self) {
-        {
-            let mut wm = self.wallet_manager.write().await;
-            if let Some(info) = wm.get_wallet_info_mut(&self.wallet_id) {
-                for account in info.core_wallet.all_platform_payment_managed_accounts_mut() {
-                    account.clear_balances();
-                }
-            }
-        }
         let mut guard = self.provider.write().await;
         if let Some(provider) = guard.as_mut() {
             provider.reset_sync_state();
         }
+        let mut wm = self.wallet_manager.write().await;
+        if let Some(info) = wm.get_wallet_info_mut(&self.wallet_id) {
+            for account in info.core_wallet.all_platform_payment_managed_accounts_mut() {
+                account.clear_balances();
+            }
+        }
     }
 
     /// Internal accessor for the diagnostic snapshot path on

packages/rs-platform-wallet/src/wallet/shielded/file_store.rs

@@ -295,11 +295,14 @@ impl ShieldedStore for FileBackedShieldedStore {
         let Some(sw) = self.subwallets.get_mut(&id) else {
             return Ok(false);
         };
+        let known = sw.nullifier_index.contains_key(nullifier);
         let marked = sw.mark_spent(nullifier);
-        if marked {
+        if known {
             // `SubwalletState::mark_spent` dropped any redrive carrying
-            // this nullifier from memory; mirror the deletions. The
-            // in-memory transition already happened, so a SQLite failure
+            // this nullifier from memory — even when the note was
+            // already spent (a redrive row can rehydrate after the
+            // promotion); mirror the deletions. The in-memory
+            // transition already happened, so a SQLite failure
             // must not abort the call — log it and keep the trait
             // behavior consistent. A surviving stale row rehydrates a
             // reservation on the next open, which the reconcile / prune
@@ -611,13 +614,24 @@ impl ShieldedStore for FileBackedShieldedStore {
     fn purge_wallet(&mut self, wallet_id: WalletId) -> Result<(), Self::Error> {
         // Per-subwallet note / watermark / checkpoint state is
         // in-memory only (`subwallets`); the commitment tree in
-        // SQLite is chain-wide and intentionally left intact.
+        // SQLite is chain-wide and intentionally left intact. The
+        // wallet's persisted redrive rows must go too, or they would
+        // rehydrate reservations for a wallet that no longer exists.
         self.subwallets.retain(|id, _| id.wallet_id != wallet_id);
+        let conn = self.pending_conn.lock().expect("pending_conn mutex");
+        conn.execute(
+            "DELETE FROM shielded_pending_spends WHERE wallet_id = ?1",
+            rusqlite::params![wallet_id.as_slice()],
+        )
+        .map_err(|e| FileShieldedStoreError(format!("purge wallet redrive rows: {e}")))?;
         Ok(())
     }
 
     fn purge_all_subwallets(&mut self) -> Result<(), Self::Error> {
         self.subwallets.clear();
+        let conn = self.pending_conn.lock().expect("pending_conn mutex");
+        conn.execute("DELETE FROM shielded_pending_spends", [])
+            .map_err(|e| FileShieldedStoreError(format!("purge redrive rows: {e}")))?;
         Ok(())
     }