fix(platform-wallet): typed persister errors with bounded transient retry

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

@@ -266,6 +266,19 @@ pub trait PlatformWalletPersistence: Send + Sync {
     /// wallet accessor (readers and writers) for its duration. Keep the
     /// per-call work bounded; if the backend does inline I/O (see the type
     /// doc), size it accordingly.
+    ///
+    /// # Transient-failure retry contract
+    ///
+    /// An implementation that returns a [`PersistenceError`] classified
+    /// [`PersistenceErrorKind::Transient`] from `store` **MUST** have already
+    /// buffered/preserved the changeset so that a subsequent bare
+    /// [`flush`](Self::flush) — with no re-supplied changeset — completes the
+    /// write (mirroring `flush`'s own transient contract). This is what lets a
+    /// caller retry a transient `store` failure via `flush` alone; re-calling
+    /// `store` with the same changeset would double-merge it. An
+    /// implementation that cannot preserve the changeset on failure MUST
+    /// classify that failure [`PersistenceErrorKind::Fatal`] (or
+    /// [`Constraint`](PersistenceErrorKind::Constraint)), never `Transient`.
     fn store(
         &self,
         wallet_id: WalletId,

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

@@ -14,6 +14,40 @@ pub enum PlatformWalletError {
     #[error("Wallet creation failed: {0}")]
     WalletCreation(String),
 
+    /// The persister failed to load the client start state during
+    /// rehydration. Carries the typed [`PersistenceError`] so callers keep
+    /// its retry classification (`is_transient()` /
+    /// [`PersistenceErrorKind`]) instead of a flattened string — a
+    /// transient backend hiccup (e.g. `SQLITE_BUSY`) stays distinguishable
+    /// from a permanent failure and can be retried.
+    ///
+    /// [`PersistenceError`]: crate::changeset::PersistenceError
+    /// [`PersistenceErrorKind`]: crate::changeset::PersistenceErrorKind
+    #[error("failed to load persisted client state: {0}")]
+    PersisterLoad(#[from] crate::changeset::PersistenceError),
+
+    /// The persister failed to store the wallet-registration changeset.
+    /// Like [`Self::PersisterLoad`], it carries the typed
+    /// [`PersistenceError`] so the retry classification (`is_transient()`
+    /// / [`PersistenceErrorKind`]) survives the boundary — a transient
+    /// `SQLITE_BUSY` stays distinguishable from a permanent failure.
+    /// Distinct from [`Self::PersisterLoad`] so callers can tell a failed
+    /// registration write from a failed rehydration read; not `#[from]`
+    /// because that conversion is already claimed by [`Self::PersisterLoad`].
+    ///
+    /// [`PersistenceError`]: crate::changeset::PersistenceError
+    /// [`PersistenceErrorKind`]: crate::changeset::PersistenceErrorKind
+    #[error("failed to persist wallet registration changeset: {0}")]
+    PersisterStore(#[source] crate::changeset::PersistenceError),
+
+    /// Restoring the persisted platform-address state into the freshly
+    /// registered wallet failed. Wraps the underlying
+    /// [`PlatformWalletError`](Self) (boxed to break the recursive type) so
+    /// its concrete variant and `#[source]` chain survive instead of being
+    /// flattened into a string.
+    #[error("failed to restore persisted platform-address state: {0}")]
+    PersisterRestore(#[source] Box<PlatformWalletError>),
+
     #[error("Wallet not found: {0}")]
     WalletNotFound(String),
 

packages/rs-platform-wallet/src/manager/load.rs

@@ -10,7 +10,7 @@ use crate::wallet::identity::IdentityManager;
 use crate::wallet::platform_wallet::{PlatformWalletInfo, WalletId};
 use crate::wallet::PlatformWallet;
 
-use super::PlatformWalletManager;
+use super::{wallet_lifecycle::retry_transient, PlatformWalletManager};
 
 impl<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
     /// Load the full [`ClientStartState`] from the configured persister
@@ -30,19 +30,30 @@ impl<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
     ///
     /// [`WalletManager`]: key_wallet_manager::WalletManager
     pub async fn load_from_persistor(&self) -> Result<(), PlatformWalletError> {
+        let start_state = match retry_transient(|| self.persister.load()).await {
+            Ok(state) => state,
+            Err(e) => {
+                // Preserve the typed source chain (Debug carries the real
+                // cause — e.g. a bincode decode failure) instead of flattening
+                // it to a Display string, and release the wallet-event adapter
+                // so a reconstruct on the same path doesn't hit `AlreadyOpen`
+                // masking this error.
+                tracing::debug!(error = ?e, "persister load failed during rehydration");
+                let report = self.shutdown().await;
+                if !report.all_clean() {
+                    tracing::warn!(?report, "wallet workers unclean after aborting rehydration");
+                }
+                return Err(PlatformWalletError::PersisterLoad(e));
+            }
+        };
         let ClientStartState {
             mut platform_addresses,
             wallets,
             // Shielded restore happens lazily on `bind_shielded`,
             // not here — drop the snapshot at this entry point.
             #[cfg(feature = "shielded")]
                 shielded: _,
-        } = self.persister.load().map_err(|e| {
-            PlatformWalletError::WalletCreation(format!(
-                "Failed to load persisted client state: {}",
-                e
-            ))
-        })?;
+        } = start_state;
 
         // Tracked (wallet-independent) masternodes ride the same startup
         // hydration; a failure logs and starts empty rather than failing
@@ -237,6 +248,16 @@ impl<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
                     }
                 }
             }
+            // Release the wallet-event adapter so a reconstruct on the same
+            // persister path doesn't hit `AlreadyOpen` (see the early-return
+            // path above).
+            let report = self.shutdown().await;
+            if !report.all_clean() {
+                tracing::warn!(
+                    ?report,
+                    "wallet workers left unclean after rolling back a failed rehydration"
+                );
+            }
             return Err(err);
         }
 
@@ -363,3 +384,183 @@ mod idempotent_load_tests {
         );
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::sync::Arc;
+
+    use super::*;
+    use crate::changeset::{PersistenceError, PersistenceErrorKind, PlatformWalletChangeSet};
+    use crate::events::{EventHandler, PlatformEventHandler};
+
+    /// Persister whose `load()` always fails — the failure path under test.
+    struct FailingLoadPersister;
+
+    impl PlatformWalletPersistence for FailingLoadPersister {
+        fn store(
+            &self,
+            _wallet_id: WalletId,
+            _changeset: PlatformWalletChangeSet,
+        ) -> Result<(), PersistenceError> {
+            Ok(())
+        }
+
+        fn flush(&self, _wallet_id: WalletId) -> Result<(), PersistenceError> {
+            Ok(())
+        }
+
+        fn load(&self) -> Result<ClientStartState, PersistenceError> {
+            Err(PersistenceError::backend("simulated load failure"))
+        }
+    }
+
+    struct TransientOnceLoadPersister {
+        load_calls: AtomicUsize,
+    }
+
+    impl PlatformWalletPersistence for TransientOnceLoadPersister {
+        fn store(
+            &self,
+            _wallet_id: WalletId,
+            _changeset: PlatformWalletChangeSet,
+        ) -> Result<(), PersistenceError> {
+            Ok(())
+        }
+
+        fn flush(&self, _wallet_id: WalletId) -> Result<(), PersistenceError> {
+            Ok(())
+        }
+
+        fn load(&self) -> Result<ClientStartState, PersistenceError> {
+            if self.load_calls.fetch_add(1, Ordering::SeqCst) == 0 {
+                return Err(PersistenceError::backend_with_kind(
+                    PersistenceErrorKind::Transient,
+                    "simulated transient load failure",
+                ));
+            }
+            Ok(ClientStartState::default())
+        }
+    }
+
+    struct NoopEventHandler;
+    impl EventHandler for NoopEventHandler {}
+    impl PlatformEventHandler for NoopEventHandler {}
+
+    #[tokio::test]
+    async fn transient_load_failure_during_startup_rehydration_is_retried() {
+        let persister = Arc::new(TransientOnceLoadPersister {
+            load_calls: AtomicUsize::new(0),
+        });
+        let probe = Arc::clone(&persister);
+        let sdk = Arc::new(dash_sdk::SdkBuilder::new_mock().build().expect("mock sdk"));
+        let handler: Arc<dyn PlatformEventHandler> = Arc::new(NoopEventHandler);
+        let manager = PlatformWalletManager::new(sdk, persister, handler);
+
+        manager
+            .load_from_persistor()
+            .await
+            .expect("transient startup load failure must be retried");
+
+        assert_eq!(probe.load_calls.load(Ordering::SeqCst), 2);
+    }
+
+    /// A failed `load_from_persistor` must (a) surface the typed `PersisterLoad`
+    /// error preserving the source chain, and (b) release the wallet-event
+    /// adapter's `Arc<persister>` clone so a reconstruct on the same path
+    /// doesn't hit `WalletStorageError::AlreadyOpen` masking the real error
+    /// (issue #4133).
+    ///
+    /// This is a **manager-side proxy**, not an end-to-end proof: a strong count
+    /// back at 1 (the test's own probe) after a failed load + teardown is the
+    /// necessary precondition for a clean re-open, not the re-open itself. The
+    /// concrete `SqlitePersister` lives in `platform-wallet-storage`, which
+    /// depends on this crate, so only that side can drive the composed path —
+    /// and its `sqlite_second_open_guard` covers just the other half (dropping
+    /// the last handle frees the path claim), never building a
+    /// `PlatformWalletManager`.
+    // TODO: cover the composed open -> failed load -> reopen from
+    // platform-wallet-storage; neither side asserts it today.
+    // Multi-thread: dropping the manager runs upstream's `Drop`, whose
+    // `ThreadRegistry::shutdown()` asserts a multi-thread runtime.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn failed_load_releases_persister_for_reconstruct() {
+        let persister = Arc::new(FailingLoadPersister);
+        let probe = Arc::clone(&persister);
+        let sdk = Arc::new(dash_sdk::SdkBuilder::new_mock().build().expect("mock sdk"));
+        let handler: Arc<dyn PlatformEventHandler> = Arc::new(NoopEventHandler);
+
+        let manager = PlatformWalletManager::new(sdk, persister, handler);
+
+        let err = manager
+            .load_from_persistor()
+            .await
+            .expect_err("load must fail");
+        assert!(
+            matches!(err, PlatformWalletError::PersisterLoad(_)),
+            "load failure must surface as the typed PersisterLoad variant, got {err:?}"
+        );
+
+        drop(manager);
+        // Asserted directly, never polled: the failure path awaits the
+        // adapter's `JoinHandle` inside `shutdown`, so the task's clone is
+        // already released before `drop` runs. Release on THIS path is
+        // synchronous, which is the stronger guarantee — a poll loop (or a
+        // `yield_now`, which cedes nothing to another worker) would only
+        // hide a regression into eventual release.
+        assert_eq!(
+            Arc::strong_count(&probe),
+            1,
+            "after a failed load + teardown nothing may still hold the persister"
+        );
+    }
+
+    /// The `Drop` backstop alone (no `shutdown` first) must *eventually* release
+    /// the adapter's `Arc<persister>` clone. Unlike the graceful path this is
+    /// not synchronous: `Drop::drop` calls `abort()`, which only *requests*
+    /// cancellation — the runtime drops the aborted task (and its clone) at its
+    /// next poll. So the strong count is polled, not asserted immediately, which
+    /// is exactly the "eventual, not synchronous" contract the `Drop` impl's
+    /// doc-comment describes. This is the branch the graceful-path test above
+    /// never exercises (there `shutdown` has already taken the join handle, so
+    /// `Drop`'s `abort` sees `None`).
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn drop_backstop_eventually_releases_persister_without_shutdown() {
+        let persister = Arc::new(FailingLoadPersister);
+        let probe = Arc::clone(&persister);
+        let sdk = Arc::new(dash_sdk::SdkBuilder::new_mock().build().expect("mock sdk"));
+        let handler: Arc<dyn PlatformEventHandler> = Arc::new(NoopEventHandler);
+
+        let manager = PlatformWalletManager::new(sdk, persister, handler);
+        // The adapter task spawned in `new()` holds a clone, so the count is
+        // above the probe before any teardown.
+        assert!(
+            Arc::strong_count(&probe) > 1,
+            "the spawned adapter task must hold an Arc<persister> clone"
+        );
+
+        // Dirty drop: never call `shutdown`, so `Drop`'s `abort` is the only
+        // thing that can reclaim the adapter's clone.
+        drop(manager);
+
+        // Release is eventual: poll until the aborted task is dropped by the
+        // runtime rather than asserting immediately. The wait must be a timed
+        // sleep, not `yield_now`: the aborted task is reclaimed by whichever
+        // worker thread owns it, and yielding this thread never forces that
+        // one to run — the whole budget can burn in microseconds while the
+        // clone is still live. Breaks on the first observation, so the 2s
+        // ceiling is only ever paid by a genuine regression.
+        let mut count = Arc::strong_count(&probe);
+        for _ in 0..2_000 {
+            if count == 1 {
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(1)).await;
+            count = Arc::strong_count(&probe);
+        }
+        assert_eq!(
+            count, 1,
+            "the Drop backstop must eventually release the persister after aborting the adapter"
+        );
+    }
+}

packages/rs-platform-wallet/src/manager/mod.rs

@@ -11,6 +11,11 @@ pub mod shielded_sync;
 pub mod startup;
 mod wallet_lifecycle;
 
+/// Re-exported so the identity-scan verdict publishers under `wallet::`
+/// retry on the same policy the registration path uses. The module itself
+/// stays private — this is the only item it owes the rest of the crate.
+pub(crate) use wallet_lifecycle::retry_transient;
+
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -1038,6 +1043,33 @@ impl<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
     }
 }
 
+/// Drop backstop for the wallet-event adapter task.
+///
+/// The graceful teardown is [`shutdown`](PlatformWalletManager::shutdown)
+/// (cancel + await the join). A dirty drop that skips it would otherwise merely
+/// detach the `JoinHandle`, leaving the adapter task running and holding its
+/// `Arc<P>` clone — which keeps the persister "open" and turns a later re-open
+/// on the same path into a spurious `WalletStorageError::AlreadyOpen` that
+/// masks the real error (issue #4133). Cancelling the token and aborting the
+/// task here starts that release — but note it is *eventual*, not synchronous:
+/// `abort()` only requests cancellation, so the runtime drops the task (and its
+/// `Arc<P>` clone) at the task's next poll, not inside this `drop`. In practice
+/// the adapter loop parks on an `.await` almost every iteration, so the clone is
+/// reclaimed promptly. Only the graceful
+/// [`shutdown`](PlatformWalletManager::shutdown) path *guarantees* the reference
+/// is gone before it returns (it awaits the join); this backstop guarantees
+/// eventual reclamation, not synchronous.
+impl<P: PlatformWalletPersistence + 'static> Drop for PlatformWalletManager<P> {
+    fn drop(&mut self) {
+        self.event_adapter_cancel.cancel();
+        // `get_mut` needs no runtime (we hold `&mut self`); `abort` is
+        // non-blocking. `None` when `shutdown` already took the handle.
+        if let Some(handle) = self.event_adapter_join.get_mut().take() {
+            handle.abort();
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

packages/rs-platform-wallet/src/manager/startup.rs

@@ -969,8 +969,10 @@ impl<P: PlatformWalletPersistence + Send + Sync + 'static> PlatformWalletManager
 
     /// Record that a scan was abandoned before it could answer every index.
     ///
-    /// Mirrors what `discover` publishes for itself; needed separately because
-    /// a scan dropped mid-await never reaches its own bookkeeping.
+    /// Mirrors what `discover` publishes for itself, retry policy included;
+    /// needed separately because a scan dropped mid-await never reaches its own
+    /// bookkeeping. This is the verdict least affordable to lose — it is the
+    /// one that re-opens the identity question on the next launch.
     async fn record_identity_scan_cut_off(&self, wallet_id: &WalletId) {
         // Coverage of nothing: the scan was dropped mid-await, so it answered
         // no index and may not clear one an earlier scan left open.
@@ -988,12 +990,24 @@ impl<P: PlatformWalletPersistence + Send + Sync + 'static> PlatformWalletManager
             identity_scan_state: Some(recorded),
             ..Default::default()
         };
-        if let Err(e) = self.persister.store(*wallet_id, changeset) {
-            tracing::warn!(
+        // Transient failures are ridden out on the registration path's bounded
+        // policy; the buffer preserves the changeset, so the retries re-drive
+        // it through `flush`. The final outcome is still swallowed — an
+        // abandoned scan must not turn a shutdown into an error.
+        let mut changeset_slot = Some(changeset);
+        let outcome = crate::manager::retry_transient(|| match changeset_slot.take() {
+            Some(cs) => self.persister.store(*wallet_id, cs),
+            None => self.persister.flush(*wallet_id),
+        })
+        .await;
+        if let Err(e) = outcome {
+            tracing::error!(
                 wallet_id = %hex::encode(wallet_id),
+                transient = e.is_transient(),
                 error = %e,
-                "failed to persist an abandoned scan's verdict; the next launch may take the \
-                 warm shortcut over an incomplete identity set"
+                "abandoned scan's verdict could not be persisted after retries; the next \
+                 launch will take the warm shortcut over an identity set nothing proved \
+                 complete"
             );
         }
     }