diff --git a/packages/rs-platform-wallet/src/changeset/traits.rs b/packages/rs-platform-wallet/src/changeset/traits.rs index 60d98195ba9..16e8ac8d217 100644 --- a/packages/rs-platform-wallet/src/changeset/traits.rs +++ b/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, diff --git a/packages/rs-platform-wallet/src/error.rs b/packages/rs-platform-wallet/src/error.rs index d24412b3007..a4c24264a2d 100644 --- a/packages/rs-platform-wallet/src/error.rs +++ b/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), + #[error("Wallet not found: {0}")] WalletNotFound(String), diff --git a/packages/rs-platform-wallet/src/manager/load.rs b/packages/rs-platform-wallet/src/manager/load.rs index 3588aef2b54..d7b7c244ba4 100644 --- a/packages/rs-platform-wallet/src/manager/load.rs +++ b/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 PlatformWalletManager

{ /// Load the full [`ClientStartState`] from the configured persister @@ -30,6 +30,22 @@ impl PlatformWalletManager

{ /// /// [`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, @@ -37,12 +53,7 @@ impl PlatformWalletManager

{ // 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 PlatformWalletManager

{ } } } + // 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 { + 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 { + 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 = 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` 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 = 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` 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 = 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 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" + ); + } +} diff --git a/packages/rs-platform-wallet/src/manager/mod.rs b/packages/rs-platform-wallet/src/manager/mod.rs index 9192dfae148..cc484031aca 100644 --- a/packages/rs-platform-wallet/src/manager/mod.rs +++ b/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 PlatformWalletManager

{ } } +/// 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

` 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

` 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 Drop for PlatformWalletManager

{ + 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::*; diff --git a/packages/rs-platform-wallet/src/manager/startup.rs b/packages/rs-platform-wallet/src/manager/startup.rs index 168512f4d40..187d269d433 100644 --- a/packages/rs-platform-wallet/src/manager/startup.rs +++ b/packages/rs-platform-wallet/src/manager/startup.rs @@ -969,8 +969,10 @@ impl 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 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" ); } } diff --git a/packages/rs-platform-wallet/src/manager/wallet_lifecycle.rs b/packages/rs-platform-wallet/src/manager/wallet_lifecycle.rs index 71bef57d723..4cc2dce78da 100644 --- a/packages/rs-platform-wallet/src/manager/wallet_lifecycle.rs +++ b/packages/rs-platform-wallet/src/manager/wallet_lifecycle.rs @@ -12,7 +12,7 @@ use key_wallet::Network; #[cfg(any(feature = "bls", feature = "eddsa"))] use crate::changeset::ProviderKeyExtendedPubKey; use crate::changeset::{ - AccountAddressPoolEntry, AccountRegistrationEntry, PlatformWalletChangeSet, + AccountAddressPoolEntry, AccountRegistrationEntry, PersistenceError, PlatformWalletChangeSet, PlatformWalletPersistence, ProviderKeyAccountEntry, WalletMetadataEntry, }; use crate::error::PlatformWalletError; @@ -51,6 +51,56 @@ fn parse_mnemonic_any_language(phrase: &str) -> Result { Err("phrase does not match any supported BIP-39 wordlist") } +/// Total attempts (initial + retries) for a transient-classified persister +/// operation on the wallet-registration path. Small on purpose: this runs +/// inline while creating a wallet, not as a background job — a lock blip +/// should be ridden out in well under a second, and a genuinely stuck +/// backend must still surface promptly. +const PERSIST_RETRY_MAX_ATTEMPTS: u32 = 4; + +/// Backoff before the first retry; doubles on each subsequent attempt. +const PERSIST_RETRY_INITIAL_BACKOFF: std::time::Duration = std::time::Duration::from_millis(20); + +/// Ceiling for the doubling backoff so registration latency stays bounded +/// (worst case with the constants above: 20 + 40 + 80 ≈ 140 ms). +const PERSIST_RETRY_MAX_BACKOFF: std::time::Duration = std::time::Duration::from_millis(200); + +/// Retry a synchronous persister operation while it fails *transiently*, +/// using bounded exponential backoff. +/// +/// `op` runs once, then re-runs after a backoff sleep for as long as it +/// returns a [`PersistenceError`] whose +/// [`is_transient()`](PersistenceError::is_transient) is true, up to +/// [`PERSIST_RETRY_MAX_ATTEMPTS`]. A fatal error (or success) returns +/// immediately — a fatal failure never retries. The sleep is async so it +/// yields the Tokio worker instead of spinning the CPU, which is exactly +/// what the storage layer's `FlushRetryable` contract asks callers to do. +pub(crate) async fn retry_transient(mut op: F) -> Result +where + F: FnMut() -> Result, +{ + let mut backoff = PERSIST_RETRY_INITIAL_BACKOFF; + let mut attempt: u32 = 1; + loop { + match op() { + Ok(value) => return Ok(value), + Err(e) if e.is_transient() && attempt < PERSIST_RETRY_MAX_ATTEMPTS => { + tracing::debug!( + attempt, + max_attempts = PERSIST_RETRY_MAX_ATTEMPTS, + backoff_ms = backoff.as_millis() as u64, + error = %e, + "transient persister failure — backing off before retry" + ); + tokio::time::sleep(backoff).await; + backoff = backoff.saturating_mul(2).min(PERSIST_RETRY_MAX_BACKOFF); + attempt += 1; + } + Err(e) => return Err(e), + } + } +} + /// Test-only rendezvous fired inside [`PlatformWalletManager::remove_wallet_with_teardown`], /// between the inner-manager removal and the public-map removal. /// @@ -484,24 +534,38 @@ impl PlatformWalletManager

{ } } - if let Err(e) = self.persister.store(wallet_id, registration_changeset) { + // Persist the registration changeset, riding out a *transient* + // backend blip (e.g. `SQLITE_BUSY`) with bounded exponential backoff + // before giving up. On a transient `store` failure the persister + // restores the buffered changeset (its documented contract), so the + // retries re-drive that same write via `flush` — no re-merge, no + // double-count: the first attempt hands the changeset over, later + // attempts flush what the buffer preserved. A fatal error is not + // retried and fails fast. Either way the typed `PersistenceError` + // (and its transient/fatal classification) is preserved for the + // caller instead of being flattened to a string. + let mut changeset_slot = Some(registration_changeset); + let store_result = 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) = store_result { tracing::error!( wallet_id = %hex::encode(wallet_id), + transient = e.is_transient(), error = %e, - "failed to persist wallet registration changeset" + "failed to persist wallet registration changeset after retries" ); let mut wm = self.wallet_manager.write().await; - if let Err(e) = wm.remove_wallet(&wallet_id) { + if let Err(remove_err) = wm.remove_wallet(&wallet_id) { tracing::warn!( wallet_id = %hex::encode(wallet_id), - error = %e, + error = %remove_err, "rollback: remove_wallet failed while unwinding a failed wallet registration" ); } - return Err(PlatformWalletError::WalletCreation(format!( - "Failed to persist wallet registration changeset: {}", - e - ))); + return Err(PlatformWalletError::PersisterStore(e)); } // Build the PlatformWallet handle. @@ -531,26 +595,36 @@ impl PlatformWalletManager

{ // earlier `insert_wallet`, absent from `self.wallets`), // poisoning every retry on `WalletAlreadyExists`. Roll back // before bailing — same shape as `manager::load`. + // Retry a transient load blip the same way as the store above; a + // load is an idempotent read, so re-reading after a lock blip is + // safe. `load_persisted()` returns the typed `PersistenceError` this + // rehydration boundary is built around, routed through the + // dedicated `PersisterLoad` variant so its retry classification + // survives to the caller. + let load_result = retry_transient(|| platform_wallet.load_persisted()).await; let crate::changeset::ClientStartState { mut platform_addresses, wallets: _, #[cfg(feature = "shielded")] shielded: _, - } = match platform_wallet.load_persisted() { + } = match load_result { Ok(state) => state, Err(e) => { + tracing::error!( + wallet_id = %hex::encode(wallet_id), + transient = e.is_transient(), + error = %e, + "failed to load persisted wallet state after retries" + ); let mut wm = self.wallet_manager.write().await; - if let Err(e) = wm.remove_wallet(&wallet_id) { + if let Err(remove_err) = wm.remove_wallet(&wallet_id) { tracing::warn!( wallet_id = %hex::encode(wallet_id), - error = %e, + error = %remove_err, "rollback: remove_wallet failed while unwinding a failed wallet setup" ); } - return Err(PlatformWalletError::WalletCreation(format!( - "Failed to load persisted wallet state: {}", - e - ))); + return Err(PlatformWalletError::PersisterLoad(e)); } }; @@ -560,18 +634,23 @@ impl PlatformWalletManager

{ .initialize_from_persisted(persisted) .await { + tracing::error!( + wallet_id = %hex::encode(wallet_id), + error = %e, + "failed to restore persisted platform-address state" + ); let mut wm = self.wallet_manager.write().await; - if let Err(e) = wm.remove_wallet(&wallet_id) { + if let Err(remove_err) = wm.remove_wallet(&wallet_id) { tracing::warn!( wallet_id = %hex::encode(wallet_id), - error = %e, + error = %remove_err, "rollback: remove_wallet failed while unwinding a failed wallet setup" ); } - return Err(PlatformWalletError::WalletCreation(format!( - "Failed to restore persisted platform address state: {}", - e - ))); + // `initialize_from_persisted` already returns a typed + // `PlatformWalletError`; wrap (boxed) rather than stringify so + // its concrete variant and source chain survive. + return Err(PlatformWalletError::PersisterRestore(Box::new(e))); } } else { platform_wallet.platform().initialize().await; @@ -1275,6 +1354,406 @@ mod register_wallet_duplicate_tests { } } +#[cfg(test)] +mod persist_retry_tests { + //! Registration-path persistence: transient-error retry and typed + //! error classification across the boundary. + + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + use key_wallet::mnemonic::{Language, Mnemonic}; + use key_wallet::wallet::initialization::WalletAccountCreationOptions; + use key_wallet::Network; + + use crate::changeset::{ + ClientStartState, PersistenceError, PersistenceErrorKind, PlatformWalletChangeSet, + PlatformWalletPersistence, + }; + use crate::error::PlatformWalletError; + use crate::events::{EventHandler, PlatformEventHandler}; + use crate::wallet::platform_wallet::WalletId; + use crate::PlatformWalletManager; + + // Canonical all-`abandon` BIP-39 test vector. + const TEST_MNEMONIC: &str = "abandon abandon abandon abandon abandon abandon \ + abandon abandon abandon abandon abandon about"; + + fn transient() -> PersistenceError { + PersistenceError::backend_with_kind( + PersistenceErrorKind::Transient, + "simulated SQLITE_BUSY", + ) + } + + fn fatal() -> PersistenceError { + PersistenceError::backend_with_kind(PersistenceErrorKind::Fatal, "simulated corruption") + } + + /// Persister whose `store` / `flush` / `load` outcomes are scripted so + /// the registration retry path can be driven deterministically. Models + /// the real contract: a transient `store` failure preserves the + /// changeset in the buffer, so the retry re-drives the write through + /// `flush`. + /// + /// `store` counts registration and identity-scan-verdict writes + /// separately. Registration ends with a best-effort `identity().sync()`, + /// so a successful registration issues a SECOND `store` carrying the scan + /// verdict; a single counter would make every assertion about the + /// registration write depend on unrelated discovery behaviour. The + /// changeset itself is the discriminator. + #[derive(Default)] + struct FaultyPersister { + /// Stores of the registration changeset. + registration_store_calls: AtomicUsize, + /// Stores of the identity-scan verdict published by `identity().sync()`. + scan_verdict_store_calls: AtomicUsize, + flush_calls: AtomicUsize, + load_calls: AtomicUsize, + /// The first registration `store` fails transiently (buffer preserved + /// for retry). + store_transient_first: bool, + /// Every registration `store` fails fatally (must NOT retry). + store_fatal: bool, + /// Number of leading scan-verdict `store` calls that fail transiently. + scan_verdict_store_transient_failures: usize, + /// Number of leading `flush` calls that fail transiently before Ok. + flush_transient_failures: usize, + /// Number of leading `load` calls that fail transiently before Ok. + load_transient_failures: usize, + /// Every `load` fails fatally (must NOT retry). + load_fatal: bool, + } + + impl PlatformWalletPersistence for FaultyPersister { + fn store( + &self, + _wallet_id: WalletId, + changeset: PlatformWalletChangeSet, + ) -> Result<(), PersistenceError> { + // One changeset can carry both: `merge` folds a buffered + // registration write and a scan verdict into a single round. Each + // counter answers only its own question — "was this changeset + // handed over?" — so both increment. Letting the first match win + // would make an assertion about the registration write depend on + // whether discovery happened to be batched with it, which is the + // coupling these separate counters exist to remove. + let registration = changeset + .wallet_metadata + .is_some() + .then(|| self.registration_store_calls.fetch_add(1, Ordering::SeqCst)); + let verdict = changeset + .identity_scan_state + .is_some() + .then(|| self.scan_verdict_store_calls.fetch_add(1, Ordering::SeqCst)); + + // The registration half decides a combined round's outcome: its + // failure aborts the whole registration, while a verdict's is + // swallowed. + if let Some(n) = registration { + if self.store_fatal { + return Err(fatal()); + } + if self.store_transient_first && n == 0 { + return Err(transient()); + } + } + if let Some(n) = verdict { + if n < self.scan_verdict_store_transient_failures { + return Err(transient()); + } + } + Ok(()) + } + + fn flush(&self, _wallet_id: WalletId) -> Result<(), PersistenceError> { + let n = self.flush_calls.fetch_add(1, Ordering::SeqCst); + if n < self.flush_transient_failures { + Err(transient()) + } else { + Ok(()) + } + } + + fn load(&self) -> Result { + let n = self.load_calls.fetch_add(1, Ordering::SeqCst); + if self.load_fatal { + return Err(fatal()); + } + if n < self.load_transient_failures { + return Err(transient()); + } + Ok(ClientStartState::default()) + } + } + + struct NoopEventHandler; + impl EventHandler for NoopEventHandler {} + impl PlatformEventHandler for NoopEventHandler {} + + fn make_manager( + persister: Arc, + ) -> Arc> { + let sdk = Arc::new(dash_sdk::SdkBuilder::new_mock().build().expect("mock sdk")); + let event_handler: Arc = Arc::new(NoopEventHandler); + Arc::new(PlatformWalletManager::new(sdk, persister, event_handler)) + } + + fn seed_bytes() -> [u8; 64] { + Mnemonic::from_phrase(TEST_MNEMONIC, Language::English) + .expect("valid test mnemonic") + .to_seed("") + } + + /// `Some(0)` skips the SPV-tip birth-height lookup so the test never + /// consults SPV. + async fn register( + manager: &PlatformWalletManager, + ) -> Result<(), PlatformWalletError> { + manager + .create_wallet_from_seed_bytes( + Network::Testnet, + &seed_bytes(), + WalletAccountCreationOptions::Default, + Some(0), + ) + .await + .map(|_| ()) + } + + /// A transient `store` failure is ridden out — the persister + /// buffers the changeset, the retry re-drives it via `flush`, and + /// registration succeeds instead of aborting. + #[tokio::test] + async fn transient_store_failure_is_retried_and_succeeds() { + let persister = Arc::new(FaultyPersister { + store_transient_first: true, + flush_transient_failures: 1, // one transient flush, then Ok + ..Default::default() + }); + let manager = make_manager(Arc::clone(&persister)); + + register(&manager) + .await + .expect("registration must succeed after retrying the transient store"); + + // store attempted once; flush retried twice (fail, then succeed). + assert_eq!(persister.registration_store_calls.load(Ordering::SeqCst), 1); + assert_eq!(persister.flush_calls.load(Ordering::SeqCst), 2); + // Registration ends in `identity().sync()`, whose scan publishes its + // verdict — the write that makes a partial scan survive a restart. + assert_eq!( + persister.scan_verdict_store_calls.load(Ordering::SeqCst), + 1, + "a completed registration must publish the identity-scan verdict" + ); + } + + /// A fatal `store` failure fails fast — no retry — and + /// surfaces as the typed `PersisterStore` whose inner classification is + /// non-transient. + #[tokio::test] + async fn fatal_store_failure_fails_fast_without_retry() { + let persister = Arc::new(FaultyPersister { + store_fatal: true, + ..Default::default() + }); + let manager = make_manager(Arc::clone(&persister)); + + let err = register(&manager) + .await + .expect_err("a fatal store must abort registration"); + + match err { + PlatformWalletError::PersisterStore(pe) => assert!( + !pe.is_transient(), + "a fatal store must carry non-transient classification" + ), + other => panic!("expected PersisterStore, got {other:?}"), + } + assert_eq!(persister.registration_store_calls.load(Ordering::SeqCst), 1); + assert_eq!( + persister.flush_calls.load(Ordering::SeqCst), + 0, + "a fatal store must not be retried via flush" + ); + assert_eq!( + persister.scan_verdict_store_calls.load(Ordering::SeqCst), + 0, + "an aborted registration never reaches the discovery scan" + ); + } + + /// A store that stays transient exhausts the + /// bounded retry budget and returns the typed `PersisterStore` still + /// carrying transient classification (distinguishable from the fatal + /// case above). + #[tokio::test] + async fn persistently_transient_store_exhausts_bounded_retries() { + let persister = Arc::new(FaultyPersister { + store_transient_first: true, + flush_transient_failures: usize::MAX, // never recovers + ..Default::default() + }); + let manager = make_manager(Arc::clone(&persister)); + + let err = register(&manager) + .await + .expect_err("registration must fail once the retry budget is spent"); + + match err { + PlatformWalletError::PersisterStore(pe) => assert!( + pe.is_transient(), + "an exhausted-but-transient store must stay classified transient" + ), + other => panic!("expected PersisterStore, got {other:?}"), + } + // 1 store + 3 flush retries == 4 total attempts (the budget). + assert_eq!(persister.registration_store_calls.load(Ordering::SeqCst), 1); + assert_eq!(persister.flush_calls.load(Ordering::SeqCst), 3); + assert_eq!( + persister.scan_verdict_store_calls.load(Ordering::SeqCst), + 0, + "an aborted registration never reaches the discovery scan" + ); + } + + /// A transient `load` blip during rehydration is retried (an + /// idempotent read), so registration succeeds. + #[tokio::test] + async fn transient_load_failure_is_retried_and_succeeds() { + let persister = Arc::new(FaultyPersister { + load_transient_failures: 1, + ..Default::default() + }); + let manager = make_manager(Arc::clone(&persister)); + + register(&manager) + .await + .expect("registration must succeed after retrying the transient load"); + + assert_eq!(persister.registration_store_calls.load(Ordering::SeqCst), 1); + assert_eq!(persister.flush_calls.load(Ordering::SeqCst), 0); + assert_eq!(persister.load_calls.load(Ordering::SeqCst), 2); + assert_eq!( + persister.scan_verdict_store_calls.load(Ordering::SeqCst), + 1, + "a completed registration must publish the identity-scan verdict" + ); + } + + /// A fatal `load` fails fast and surfaces as the typed + /// `PersisterLoad` — never the flattened `WalletCreation(String)`. + #[tokio::test] + async fn fatal_load_failure_surfaces_as_persister_load() { + let persister = Arc::new(FaultyPersister { + load_fatal: true, + ..Default::default() + }); + let manager = make_manager(Arc::clone(&persister)); + + let err = register(&manager) + .await + .expect_err("a fatal load must abort registration"); + + match err { + PlatformWalletError::PersisterLoad(pe) => assert!(!pe.is_transient()), + other => panic!("expected PersisterLoad, got {other:?}"), + } + assert_eq!( + persister.load_calls.load(Ordering::SeqCst), + 1, + "a fatal load must not be retried" + ); + } + + /// A transient failure persisting the identity-scan verdict is ridden out + /// on the same bounded policy the registration write uses, so a merely + /// busy backend does not cost the verdict its survival across a restart + /// (dashpay/platform#4365). + #[tokio::test] + async fn should_retry_a_transient_scan_verdict_store() { + let persister = Arc::new(FaultyPersister { + scan_verdict_store_transient_failures: 1, + ..Default::default() + }); + let manager = make_manager(Arc::clone(&persister)); + + register(&manager) + .await + .expect("a retried scan-verdict store must not disturb registration"); + + assert_eq!( + persister.scan_verdict_store_calls.load(Ordering::SeqCst), + 1, + "the verdict is handed over once; the retry re-drives it via flush" + ); + assert_eq!( + persister.flush_calls.load(Ordering::SeqCst), + 1, + "the transient verdict store must be retried through flush" + ); + } + + /// Retrying the verdict never escalates into failing the scan that just + /// succeeded: once the budget is spent the outcome is logged and dropped, + /// and registration still returns Ok. + #[tokio::test] + async fn should_not_fail_registration_when_the_scan_verdict_never_persists() { + let persister = Arc::new(FaultyPersister { + scan_verdict_store_transient_failures: usize::MAX, + flush_transient_failures: usize::MAX, + ..Default::default() + }); + let manager = make_manager(Arc::clone(&persister)); + + register(&manager) + .await + .expect("an unpersistable verdict must never fail wallet registration"); + + // 1 store + 3 flush retries == the shared 4-attempt budget. + assert_eq!(persister.scan_verdict_store_calls.load(Ordering::SeqCst), 1); + assert_eq!(persister.flush_calls.load(Ordering::SeqCst), 3); + } + + /// The typed persister-phase variants preserve retry + /// classification, enable structural matching, and keep the `#[source]` + /// chain instead of flattening to a string. + #[test] + fn typed_variants_preserve_classification_matching_and_source() { + use std::error::Error; + + let store_err = PlatformWalletError::PersisterStore(transient()); + match &store_err { + PlatformWalletError::PersisterStore(pe) => assert!(pe.is_transient()), + other => panic!("expected PersisterStore, got {other:?}"), + } + assert!( + store_err.source().is_some(), + "PersisterStore must expose its PersistenceError source" + ); + + let load_err = PlatformWalletError::PersisterLoad(fatal()); + match &load_err { + PlatformWalletError::PersisterLoad(pe) => assert!(!pe.is_transient()), + other => panic!("expected PersisterLoad, got {other:?}"), + } + assert!(load_err.source().is_some()); + + // The restore variant wraps a typed inner error; structural matching + // must recover the concrete inner variant, not an opaque string. + let restore_err = + PlatformWalletError::PersisterRestore(Box::new(PlatformWalletError::WalletLocked)); + assert!(restore_err.source().is_some()); + match restore_err { + PlatformWalletError::PersisterRestore(inner) => { + assert!(matches!(*inner, PlatformWalletError::WalletLocked)); + } + other => panic!("expected PersisterRestore, got {other:?}"), + } + } +} + /// Removal versus a same-id re-registration that lands *during* the removal /// (`dashpay/platform#4185` review). /// diff --git a/packages/rs-platform-wallet/src/wallet/asset_lock/sync/proof.rs b/packages/rs-platform-wallet/src/wallet/asset_lock/sync/proof.rs index 6d4b674d965..49f34530e22 100644 --- a/packages/rs-platform-wallet/src/wallet/asset_lock/sync/proof.rs +++ b/packages/rs-platform-wallet/src/wallet/asset_lock/sync/proof.rs @@ -27,9 +27,9 @@ use super::super::manager::AssetLockManager; /// Persister errors are surfaced as `Err(PersistenceError)` so call /// sites can choose their own policy: /// -/// - **Poll loops** (`wait_for_chain_lock`, `wait_for_proof`) typically -/// downgrade to `None` for the current iteration so the next tick -/// retries — see [`record_or_persister_or_log`] for that policy. +/// - **Poll loops** (`wait_for_chain_lock`, `wait_for_proof`) downgrade +/// transient failures to `None` for the current iteration and surface +/// permanent failures — see [`record_or_persister_or_log`]. /// - **One-shot recovery / fast-fail call sites** want the error /// visible so a transient backend failure isn't silently classified /// as "tx not found" — they handle the `Err` arm explicitly. @@ -143,26 +143,27 @@ pub(in crate::wallet::asset_lock) fn record_holds_local_finality( } } -/// Variant of [`record_or_persister`] that swallows persister errors -/// as `None` after a `warn`-level log. Use this from poll loops where -/// the next iteration retries — a hard error from a single tick would -/// abort the whole poll prematurely. +/// Variant of [`record_or_persister`] that retries transient failures as a miss. +/// +/// Use this from poll loops where the next iteration retries. Permanent +/// failures remain errors so an unbounded poll cannot hide them. pub(super) fn record_or_persister_or_log( in_memory: Option, persister: &crate::wallet::persister::WalletPersister, txid: &Txid, -) -> Option { +) -> Result, crate::changeset::PersistenceError> { match record_or_persister(in_memory, persister, txid) { - Ok(opt) => opt, - Err(e) => { + Ok(opt) => Ok(opt), + Err(e) if e.is_transient() => { tracing::warn!( txid = %txid, error = %e, - "Persister fallback for core tx record failed; \ + "Transient persister fallback for core tx record failed; \ treating as miss for this poll iteration" ); - None + Ok(None) } + Err(e) => Err(e), } } @@ -393,7 +394,7 @@ impl AssetLockManager { }) }; if let Some(record) = - record_or_persister_or_log(in_memory, &self.persister, &out_point.txid) + record_or_persister_or_log(in_memory, &self.persister, &out_point.txid)? { if matches!(record.context, TransactionContext::InChainLockedBlock(_)) { if let Some(h) = record.height() { @@ -520,7 +521,7 @@ impl AssetLockManager { }) }; if let Some(record) = - record_or_persister_or_log(in_memory, &self.persister, &out_point.txid) + record_or_persister_or_log(in_memory, &self.persister, &out_point.txid)? { match &record.context { TransactionContext::InstantSend(instant_lock) => { @@ -969,8 +970,7 @@ mod tests { } } - /// Test persister that always errors out on `get_core_tx_record`, - /// to exercise the error-swallowing branch in `record_or_persister`. + /// Test persister that returns a permanent `get_core_tx_record` error. struct ErroringStore; impl PlatformWalletPersistence for ErroringStore { @@ -996,6 +996,34 @@ mod tests { } } + struct TransientErroringStore; + + impl PlatformWalletPersistence for TransientErroringStore { + fn store( + &self, + _wallet_id: WalletId, + _changeset: PlatformWalletChangeSet, + ) -> Result<(), PersistenceError> { + Ok(()) + } + fn flush(&self, _wallet_id: WalletId) -> Result<(), PersistenceError> { + Ok(()) + } + fn load(&self) -> Result { + Ok(ClientStartState::default()) + } + fn get_core_tx_record( + &self, + _wallet_id: WalletId, + _txid: &Txid, + ) -> Result, PersistenceError> { + Err(PersistenceError::backend_with_kind( + crate::changeset::PersistenceErrorKind::Transient, + "simulated transient backend failure", + )) + } + } + fn wallet_persister(inner: Arc) -> WalletPersister { WalletPersister::new([0u8; 32], inner) } @@ -1060,9 +1088,7 @@ mod tests { #[test] fn record_or_persister_propagates_backend_errors() { // Backend errors surface as `Err` so call sites can choose - // their own policy (one-shot recovery logs at error and - // degrades; poll loops downgrade to None for one tick via - // `record_or_persister_or_log`). + // their own policy; poll loops only downgrade transient errors. let unknown_txid = Txid::from([0xFF; 32]); let persister = wallet_persister(Arc::new(ErroringStore)); @@ -1071,14 +1097,21 @@ mod tests { } #[test] - fn record_or_persister_or_log_swallows_backend_errors_as_none() { - // The poll-loop variant downgrades errors to `None` (after a - // `warn` log) so a transient backend failure on one tick - // doesn't abort the whole poll. + fn record_or_persister_or_log_surfaces_permanent_backend_errors() { let unknown_txid = Txid::from([0xFF; 32]); let persister = wallet_persister(Arc::new(ErroringStore)); let resolved = record_or_persister_or_log(None, &persister, &unknown_txid); + assert!(resolved.is_err()); + } + + #[test] + fn record_or_persister_or_log_retries_transient_backend_errors() { + let unknown_txid = Txid::from([0xFF; 32]); + let persister = wallet_persister(Arc::new(TransientErroringStore)); + + let resolved = record_or_persister_or_log(None, &persister, &unknown_txid) + .expect("transient poll error must be downgraded for retry"); assert!(resolved.is_none()); } } diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/discovery.rs b/packages/rs-platform-wallet/src/wallet/identity/network/discovery.rs index 93dbfe1ff5b..50731a63871 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/network/discovery.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/network/discovery.rs @@ -640,9 +640,14 @@ impl IdentityWallet { /// Best-effort by design, and on the persist half only: the in-memory /// record always lands, so a second bring-up in this process already sees /// an incomplete scan and rescans. A failed persist costs the verdict its - /// survival across a restart, which is the same exposure a host that has - /// no slot for the field already has — it must not be allowed to fail the - /// scan that just succeeded. + /// survival across a restart, and it must not be allowed to fail the scan + /// that just succeeded. + /// + /// Best-effort is not one-shot, though. A backend that is merely busy + /// would otherwise cost the verdict its durability outright, which is the + /// gap the verdict exists to close (dashpay/platform#4365), so a transient + /// failure is ridden out on the same bounded policy the registration path + /// uses before the outcome is swallowed. async fn publish_scan_verdict( &self, wallet_id: crate::wallet::platform_wallet::WalletId, @@ -672,12 +677,23 @@ impl IdentityWallet { identity_scan_state: Some(recorded), ..Default::default() }; - if let Err(e) = self.persister.store(changeset) { - tracing::warn!( + // On a transient `store` failure the persister keeps the changeset + // buffered (its documented contract), so the retries re-drive that + // same write through `flush` rather than handing it over twice. + let mut changeset_slot = Some(changeset); + let outcome = crate::manager::retry_transient(|| match changeset_slot.take() { + Some(cs) => self.persister.store(cs), + None => self.persister.flush(), + }) + .await; + if let Err(e) = outcome { + tracing::error!( wallet_id = %hex::encode(wallet_id), + transient = e.is_transient(), error = %e, - "failed to persist the identity-scan verdict; a partial scan may not be \ - retried after a restart" + "identity-scan verdict could not be persisted after retries; a partial scan \ + will not be retried after a restart, so an identity at an unanswered index \ + stays hidden until a later scan publishes a verdict that lands" ); } } diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/payments.rs b/packages/rs-platform-wallet/src/wallet/identity/network/payments.rs index d59c3250390..659a430ffe3 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/network/payments.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/network/payments.rs @@ -670,6 +670,11 @@ impl DashPayView<'_, B> { /// retried on the next sweep. /// /// Returns the number of entries confirmed this pass. + /// + /// # Errors + /// + /// Transient persistence read failures are deferred to the next sweep; + /// permanent failures return [`PlatformWalletError::PersisterLoad`]. pub async fn reconcile_sent_payments(&self) -> Result { use crate::wallet::identity::types::dashpay::payment::{PaymentDirection, PaymentStatus}; @@ -704,14 +709,16 @@ impl DashPayView<'_, B> { let record = match self.persister.get_core_tx_record(&txid) { Ok(Some(record)) => record, Ok(None) => continue, - Err(e) => { + Err(e) if e.is_transient() => { tracing::warn!( error = %e, txid = %txid_str, - "reconcile_sent_payments: tx-record read failed; will retry next sweep" + "reconcile_sent_payments: transient tx-record read failed; \ + will retry next sweep" ); continue; } + Err(e) => return Err(PlatformWalletError::PersisterLoad(e)), }; // An InstantSend lock is final for DashPay display, same as a // mined block — one definition of "final", shared with the @@ -1648,7 +1655,8 @@ mod tests { use key_wallet::Network; use crate::changeset::{ - ClientStartState, PersistenceError, PlatformWalletChangeSet, PlatformWalletPersistence, + ClientStartState, PersistenceError, PersistenceErrorKind, PlatformWalletChangeSet, + PlatformWalletPersistence, }; use crate::error::PlatformWalletError; use crate::events::{EventHandler, PlatformEventHandler}; @@ -1698,6 +1706,9 @@ mod tests { key_wallet::managed_account::transaction_record::TransactionRecord, >, >, + /// `Some(kind)` makes every `get_core_tx_record` fail with that + /// error class instead of answering from `records`. + read_error_kind: Mutex>, /// Txids the enumeration lists but `get_core_tx_record` answers /// `Ok(None)` for — the FFI shape for "row exists, record not /// available yet" (missing bytes, undecodable, pending InstantSend). @@ -1746,6 +1757,12 @@ mod tests { PersistenceError, > { *self.get_core_tx_record_calls.lock().unwrap() += 1; + if let Some(kind) = *self.read_error_kind.lock().unwrap() { + return Err(PersistenceError::backend_with_kind( + kind, + "simulated tx-record read failure", + )); + } if self.listed_but_unavailable.lock().unwrap().contains(txid) { return Ok(None); } @@ -3584,6 +3601,26 @@ mod tests { 0, "reconcile must be idempotent" ); + + *persister.read_error_kind.lock().unwrap() = Some(PersistenceErrorKind::Transient); + assert_eq!( + iw.dashpay() + .reconcile_sent_payments() + .await + .expect("transient read failure must wait for the next sweep"), + 0 + ); + + *persister.read_error_kind.lock().unwrap() = Some(PersistenceErrorKind::Fatal); + let err = iw + .dashpay() + .reconcile_sent_payments() + .await + .expect_err("permanent read failure must abort the reconcile sweep"); + assert!(matches!( + err, + PlatformWalletError::PersisterLoad(ref source) if !source.is_transient() + )); } #[tokio::test]