diff --git a/packages/rs-platform-wallet-ffi/src/handle.rs b/packages/rs-platform-wallet-ffi/src/handle.rs index d0034fe9fc7..84b6337b4d9 100644 --- a/packages/rs-platform-wallet-ffi/src/handle.rs +++ b/packages/rs-platform-wallet-ffi/src/handle.rs @@ -80,6 +80,25 @@ impl HandleStorage { guard.get(&handle).map(f) } + /// [`with_item`](Self::with_item) for calls that can run while another + /// thread's read guard waits on them, such as an export called from an + /// event callback that another export dispatched while holding the + /// registry. A plain read parks behind any queued writer (parking_lot + /// prefers writers), and that writer waits for the outer guard, so the + /// three deadlock. This read only waits for a writer that holds the + /// registry, so it is only valid on a storage written solely through + /// `insert` and `remove`, which hold it just for the map update, as + /// [`PLATFORM_WALLET_MANAGER_STORAGE`] is. Writers that run other code + /// under the lock (`with_item_mut`, `get_mut`, `remove_matching`) void + /// that. + pub fn with_item_recursive(&self, handle: Handle, f: F) -> Option + where + F: FnOnce(&T) -> R, + { + let guard = self.items.read_recursive(); + guard.get(&handle).map(f) + } + pub fn with_item_mut(&self, handle: Handle, f: F) -> Option where F: FnOnce(&mut T) -> R, diff --git a/packages/rs-platform-wallet-ffi/src/lib.rs b/packages/rs-platform-wallet-ffi/src/lib.rs index 4734b414279..88d16f2d428 100644 --- a/packages/rs-platform-wallet-ffi/src/lib.rs +++ b/packages/rs-platform-wallet-ffi/src/lib.rs @@ -79,6 +79,8 @@ pub mod shielded_persistence; #[cfg(feature = "shielded")] pub mod shielded_send; #[cfg(feature = "shielded")] +pub mod shielded_spend_prefetch; +#[cfg(feature = "shielded")] pub mod shielded_sync; pub mod shielded_types; #[cfg(feature = "shielded")] @@ -161,6 +163,8 @@ pub use secp256k1_primitives::*; #[cfg(feature = "shielded")] pub use shielded_send::*; #[cfg(feature = "shielded")] +pub use shielded_spend_prefetch::*; +#[cfg(feature = "shielded")] pub use shielded_sync::*; pub use shielded_types::*; #[cfg(feature = "shielded")] diff --git a/packages/rs-platform-wallet-ffi/src/shielded_spend_prefetch.rs b/packages/rs-platform-wallet-ffi/src/shielded_spend_prefetch.rs new file mode 100644 index 00000000000..7c812340df1 --- /dev/null +++ b/packages/rs-platform-wallet-ffi/src/shielded_spend_prefetch.rs @@ -0,0 +1,234 @@ +//! FFI entry point that prepares an upcoming shielded spend. +//! +//! Every shielded spend (transfer, unshield, withdraw, identity top-up / +//! create from the pool) must pick an anchor Platform has recorded, which +//! takes a round trip that downloads and proof-verifies Platform's recorded +//! anchor set. Hosts call +//! [`platform_wallet_manager_shielded_prefetch_spend_anchors`] when a send +//! screen opens so that round trip overlaps the user filling in the form +//! instead of delaying the send. Feature-gated behind `shielded`. + +use crate::error::*; +use crate::handle::*; +use crate::runtime::{run_on_big_stack_thread, runtime}; + +/// Prefetch Platform's recorded shielded-anchor set for the next spend on +/// `handle`'s network, in the background. +/// +/// Returns as soon as the fetch is queued; it runs on the FFI runtime. A +/// shielded send started within the next ~30 s, before any shielded sync +/// grows the commitment tree, reuses the prefetched set and skips that round +/// trip; otherwise the send fetches for itself exactly as without this call. +/// Cheap to call repeatedly (a no-op while a usable set is cached), and a +/// failed fetch is only logged — it never affects a later send. Safe to call +/// from any thread, including from inside an event callback. +/// +/// Returns `ErrorInvalidHandle` for an unknown `handle` and +/// `ErrorWalletOperation` when shielded support isn't configured on the +/// manager. +#[no_mangle] +pub extern "C" fn platform_wallet_manager_shielded_prefetch_spend_anchors( + handle: Handle, +) -> PlatformWalletFFIResult { + // Called from an event callback, this can run while the export that + // dispatched the event holds the registry, hence the recursive read. + let lookup = PLATFORM_WALLET_MANAGER_STORAGE.with_item_recursive(handle, |manager| { + let resolve = || runtime().block_on(manager.shielded_coordinator()); + // Event callbacks fire inside a runtime context (the sync loop + // drives its passes with `Handle::block_on`), where a nested + // `block_on` panics; resolve on a thread outside it instead. + if tokio::runtime::Handle::try_current().is_ok() { + run_on_big_stack_thread(resolve) + } else { + Ok(resolve()) + } + }); + let coordinator = match lookup { + None => { + return PlatformWalletFFIResult::err( + PlatformWalletFFIResultCode::ErrorInvalidHandle, + format!("invalid manager handle: {handle}"), + ); + } + Some(Err(e)) => { + return PlatformWalletFFIResult::err( + PlatformWalletFFIResultCode::ErrorWalletOperation, + format!("could not start the shielded spend-anchor prefetch: {e}"), + ); + } + Some(Ok(None)) => { + return PlatformWalletFFIResult::err( + PlatformWalletFFIResultCode::ErrorWalletOperation, + "shielded support not configured — call platform_wallet_manager_configure_shielded first", + ); + } + Some(Ok(Some(coordinator))) => coordinator, + }; + runtime().spawn(async move { + if let Err(e) = coordinator.prefetch_recorded_anchors().await { + tracing::debug!( + error = %e, + "shielded spend-anchor prefetch failed; the next spend fetches for itself" + ); + } + }); + PlatformWalletFFIResult::ok() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::event_handler::EventHandlerCallbacks; + use crate::shielded_sync::platform_wallet_manager_shielded_sync_sync_now; + use crate::shielded_types::ShieldedSyncWalletResultFFI; + use crate::test_fixtures::{ + assert_scenario_passes_in_child, mock_manager, mock_manager_with_events, + }; + use std::sync::atomic::{AtomicU64, Ordering}; + use std::sync::{mpsc, Mutex}; + use std::time::Duration; + + const WAIT: Duration = Duration::from_secs(10); + + fn code_of(mut result: PlatformWalletFFIResult) -> PlatformWalletFFIResultCode { + let code = result.code; + unsafe { crate::platform_wallet_ffi_result_free(&mut result) }; + code + } + + #[test] + fn should_reject_an_unknown_manager_handle() { + assert_eq!( + code_of(platform_wallet_manager_shielded_prefetch_spend_anchors( + NULL_HANDLE + )), + PlatformWalletFFIResultCode::ErrorInvalidHandle + ); + } + + #[test] + fn should_reject_a_manager_without_shielded_support() { + let handle = PLATFORM_WALLET_MANAGER_STORAGE.insert(mock_manager()); + assert_eq!( + code_of(platform_wallet_manager_shielded_prefetch_spend_anchors( + handle + )), + PlatformWalletFFIResultCode::ErrorWalletOperation + ); + drop(PLATFORM_WALLET_MANAGER_STORAGE.remove(handle)); + } + + /// Event callbacks run inside a runtime context; the lookup must not + /// nest `block_on` there. + #[test] + fn should_resolve_the_coordinator_from_inside_a_runtime_context() { + let handle = PLATFORM_WALLET_MANAGER_STORAGE.insert(mock_manager()); + let code = runtime().block_on(async { + code_of(platform_wallet_manager_shielded_prefetch_spend_anchors( + handle, + )) + }); + assert_eq!(code, PlatformWalletFFIResultCode::ErrorWalletOperation); + drop(PLATFORM_WALLET_MANAGER_STORAGE.remove(handle)); + } + + /// Set in the child process that runs [`completion_callback_scenario`]. + const COMPLETION_CALLBACK_CHILD: &str = + "PLATFORM_WALLET_FFI_PREFETCH_FROM_COMPLETION_CALLBACK_CHILD"; + + /// The lock cycle a call from a sync-completion callback must not form: + /// `platform_wallet_manager_shielded_sync_sync_now` holds the registry's + /// read guard until the pass, completion callback included, returns. A + /// registry writer queued behind that guard makes parking_lot park every + /// new plain reader, so a lookup that read the registry normally would + /// wait on the writer, the writer on the sync, the sync on the callback. + /// + /// The scenario depends on the sync export holding that guard across the + /// pass; if that changes, it no longer reproduces the cycle. It runs in a + /// child process, as a regression is a real deadlock (see + /// [`assert_scenario_passes_in_child`]). + #[test] + fn should_not_wedge_a_prefetch_from_a_sync_completion_callback_behind_a_queued_registry_writer() + { + if std::env::var_os(COMPLETION_CALLBACK_CHILD).is_some() { + completion_callback_scenario(); + return; + } + assert_scenario_passes_in_child( + module_path!(), + "should_not_wedge_a_prefetch_from_a_sync_completion_callback_behind_a_queued_registry_writer", + COMPLETION_CALLBACK_CHILD, + WAIT, + ); + } + + static SYNCING_HANDLE: AtomicU64 = AtomicU64::new(NULL_HANDLE); + type CallbackReport = (PlatformWalletFFIResultCode, std::thread::JoinHandle); + static CALLBACK_REPORT: Mutex>> = Mutex::new(None); + + /// Queues a registry writer behind the dispatching sync's read guard, + /// then calls the prefetch, as a host's completion handler would. + unsafe extern "C" fn prefetch_on_sync_completed( + _context: *mut std::ffi::c_void, + _results: *const ShieldedSyncWalletResultFFI, + _count: usize, + _sync_unix_seconds: u64, + ) { + let other = mock_manager(); + let writer = std::thread::spawn(move || PLATFORM_WALLET_MANAGER_STORAGE.insert(other)); + // Well inside the parent's budget, so a writer that never queued is + // reported as that rather than as a deadlock. + let deadline = std::time::Instant::now() + WAIT / 2; + while !PLATFORM_WALLET_MANAGER_STORAGE.try_read_refused() { + assert!( + std::time::Instant::now() < deadline, + "the writer never queued behind the sync's guard" + ); + std::thread::sleep(Duration::from_millis(5)); + } + + let code = code_of(platform_wallet_manager_shielded_prefetch_spend_anchors( + SYNCING_HANDLE.load(Ordering::SeqCst), + )); + if let Some(report) = CALLBACK_REPORT.lock().unwrap().as_ref() { + report + .send((code, writer)) + .expect("test observes the callback"); + } + } + + fn completion_callback_scenario() { + let (report_tx, report_rx) = mpsc::channel(); + *CALLBACK_REPORT.lock().unwrap() = Some(report_tx); + let handle = PLATFORM_WALLET_MANAGER_STORAGE.insert(mock_manager_with_events( + EventHandlerCallbacks { + context: std::ptr::null_mut(), + on_wallet_event_fn: None, + on_error_fn: None, + on_platform_address_sync_completed_fn: None, + on_shielded_sync_completed_fn: Some(prefetch_on_sync_completed), + on_shielded_sync_progress_fn: None, + on_shielded_tree_progress_fn: None, + release_fn: None, + }, + )); + SYNCING_HANDLE.store(handle, Ordering::SeqCst); + + assert_eq!( + code_of(unsafe { platform_wallet_manager_shielded_sync_sync_now(handle) }), + PlatformWalletFFIResultCode::Success + ); + let (code, writer) = report_rx + .recv_timeout(WAIT) + .expect("the pass dispatched its completion callback"); + assert_eq!( + code, + PlatformWalletFFIResultCode::ErrorWalletOperation, + "the prefetch looked the manager up and found no shielded support" + ); + let other = writer.join().expect("writer thread"); + + drop(PLATFORM_WALLET_MANAGER_STORAGE.remove(other)); + drop(PLATFORM_WALLET_MANAGER_STORAGE.remove(handle)); + } +} diff --git a/packages/rs-platform-wallet-ffi/src/spv.rs b/packages/rs-platform-wallet-ffi/src/spv.rs index 0e6c9c49640..1c3b28036fb 100644 --- a/packages/rs-platform-wallet-ffi/src/spv.rs +++ b/packages/rs-platform-wallet-ffi/src/spv.rs @@ -615,7 +615,7 @@ where #[cfg(test)] mod spv_start_registry_tests { use super::*; - use crate::test_fixtures::mock_manager; + use crate::test_fixtures::{assert_scenario_passes_in_child, mock_manager}; use std::sync::mpsc; use std::time::Duration; @@ -771,47 +771,12 @@ mod spv_start_registry_tests { return; } - let test_name = format!( - "{}::should_not_wedge_a_start_overlapping_a_native_stop_and_a_queued_registry_writer", - module_path!() - .split_once("::") - .map(|(_, path)| path) - .expect("test module path below the crate root") + assert_scenario_passes_in_child( + module_path!(), + "should_not_wedge_a_start_overlapping_a_native_stop_and_a_queued_registry_writer", + REGISTRY_DEADLOCK_CHILD, + WAIT, ); - let mut child = std::process::Command::new(std::env::current_exe().expect("test binary")) - .args(["--exact", &test_name, "--nocapture"]) - .env(REGISTRY_DEADLOCK_CHILD, "1") - .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()) - .spawn() - .expect("spawn the scenario child"); - - let deadline = std::time::Instant::now() + WAIT; - let status = loop { - if let Some(status) = child.try_wait().expect("child status") { - break Some(status); - } - if std::time::Instant::now() >= deadline { - let _ = child.kill(); - break None; - } - std::thread::sleep(Duration::from_millis(20)); - }; - let output = child.wait_with_output().expect("reap the scenario child"); - let transcript = format!( - "--- child stdout ---\n{}\n--- child stderr ---\n{}", - String::from_utf8_lossy(&output.stdout), - String::from_utf8_lossy(&output.stderr) - ); - match status { - Some(status) => assert!( - status.success(), - "the scenario failed in the child process ({status})\n{transcript}" - ), - None => { - panic!("the scenario deadlocked: the child was killed after {WAIT:?}\n{transcript}") - } - } } /// Set in the child process that runs [`registry_deadlock_scenario`]. diff --git a/packages/rs-platform-wallet-ffi/src/test_fixtures.rs b/packages/rs-platform-wallet-ffi/src/test_fixtures.rs index d41459b6586..4aeac5365e0 100644 --- a/packages/rs-platform-wallet-ffi/src/test_fixtures.rs +++ b/packages/rs-platform-wallet-ffi/src/test_fixtures.rs @@ -16,6 +16,22 @@ use crate::xpub_render::platform_wallet_free_string; /// A manager over a mock SDK with no-op persistence and event callbacks: /// registrable in the handle storage, never touching the network. pub(crate) fn mock_manager() -> PlatformWalletManager { + mock_manager_with_events(EventHandlerCallbacks { + context: std::ptr::null_mut(), + on_wallet_event_fn: None, + on_error_fn: None, + on_platform_address_sync_completed_fn: None, + on_shielded_sync_completed_fn: None, + on_shielded_sync_progress_fn: None, + on_shielded_tree_progress_fn: None, + release_fn: None, + }) +} + +/// [`mock_manager`] delivering its events to `callbacks`. +pub(crate) fn mock_manager_with_events( + callbacks: EventHandlerCallbacks, +) -> PlatformWalletManager { unsafe extern "C" fn begin(_: *mut std::ffi::c_void, _: *const u8) -> i32 { 0 } @@ -27,19 +43,7 @@ pub(crate) fn mock_manager() -> PlatformWalletManager { on_changeset_end_fn: Some(end), ..Default::default() }); - let events = FFIEventHandler::new( - EventHandlerCallbacks { - context: std::ptr::null_mut(), - on_wallet_event_fn: None, - on_error_fn: None, - on_platform_address_sync_completed_fn: None, - on_shielded_sync_completed_fn: None, - on_shielded_sync_progress_fn: None, - on_shielded_tree_progress_fn: None, - release_fn: None, - }, - None, - ); + let events = FFIEventHandler::new(callbacks, None); let _runtime_guard = runtime().enter(); PlatformWalletManager::new( Arc::new( @@ -72,3 +76,59 @@ pub(crate) fn out_string( unsafe { platform_wallet_free_string(out) }; Ok(Some(string)) } + +/// Runs the test `test_fn` of the module `module` (its `module_path!()`) alone +/// in a child process of this test binary, with `child_env` set so the test +/// runs its scenario instead of re-spawning, and fails unless the child passes +/// within `wait`. +/// +/// For scenarios whose regression is a real deadlock: no timeout can unwind +/// its threads, and in the test process they would hold the process-wide +/// handle registries and wedge every other registry-backed test. The child +/// is killed instead. +pub(crate) fn assert_scenario_passes_in_child( + module: &str, + test_fn: &str, + child_env: &str, + wait: std::time::Duration, +) { + let module_below_crate = module + .split_once("::") + .map(|(_, path)| path) + .expect("test module path below the crate root"); + let test_path = format!("{module_below_crate}::{test_fn}"); + let mut child = std::process::Command::new(std::env::current_exe().expect("test binary")) + .args(["--exact", &test_path, "--nocapture"]) + .env(child_env, "1") + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .spawn() + .expect("spawn the scenario child"); + + let deadline = std::time::Instant::now() + wait; + let status = loop { + if let Some(status) = child.try_wait().expect("child status") { + break Some(status); + } + if std::time::Instant::now() >= deadline { + let _ = child.kill(); + break None; + } + std::thread::sleep(std::time::Duration::from_millis(20)); + }; + let output = child.wait_with_output().expect("reap the scenario child"); + let transcript = format!( + "--- child stdout ---\n{}\n--- child stderr ---\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + match status { + Some(status) => assert!( + status.success(), + "the scenario failed in the child process ({status})\n{transcript}" + ), + None => { + panic!("the scenario deadlocked: the child was killed after {wait:?}\n{transcript}") + } + } +} diff --git a/packages/rs-platform-wallet/src/wallet/platform_wallet.rs b/packages/rs-platform-wallet/src/wallet/platform_wallet.rs index 50a564b04df..89bdd1d78c7 100644 --- a/packages/rs-platform-wallet/src/wallet/platform_wallet.rs +++ b/packages/rs-platform-wallet/src/wallet/platform_wallet.rs @@ -1415,6 +1415,7 @@ impl PlatformWallet { super::shielded::operations::transfer( &self.sdk, coordinator.store(), + coordinator.anchor_cache(), Some(&self.persister), self.wallet_id, &keyset, @@ -1455,6 +1456,7 @@ impl PlatformWallet { super::shielded::operations::unshield( &self.sdk, coordinator.store(), + coordinator.anchor_cache(), Some(&self.persister), self.wallet_id, &keyset, @@ -1498,6 +1500,7 @@ impl PlatformWallet { super::shielded::operations::identity_top_up_from_pool_with_metadata( &self.sdk, coordinator.store(), + coordinator.anchor_cache(), Some(&self.persister), self.wallet_id, &keyset, @@ -1562,6 +1565,7 @@ impl PlatformWallet { super::shielded::operations::withdraw( &self.sdk, coordinator.store(), + coordinator.anchor_cache(), Some(&self.persister), self.wallet_id, &keyset, @@ -1621,6 +1625,7 @@ impl PlatformWallet { super::shielded::operations::identity_create_from_shielded_pool( &self.sdk, coordinator.store(), + coordinator.anchor_cache(), Some(&self.persister), self.wallet_id, &keyset, diff --git a/packages/rs-platform-wallet/src/wallet/shielded/anchor_cache.rs b/packages/rs-platform-wallet/src/wallet/shielded/anchor_cache.rs new file mode 100644 index 00000000000..d10e4aa1332 --- /dev/null +++ b/packages/rs-platform-wallet/src/wallet/shielded/anchor_cache.rs @@ -0,0 +1,345 @@ +//! Short-lived cache of Platform's recorded shielded-anchor set. +//! +//! Every shielded spend must be built against a commitment-tree root that +//! Platform has recorded. Finding one means downloading and proof-verifying +//! the whole recorded set (`ShieldedAnchors`), a round trip that otherwise +//! sits serially in front of proving on every send. [`RecordedAnchorCache`] +//! keeps the most recent set so a send can skip it when a send-screen +//! prefetch, a previous send, or the reservation-release pass just fetched one. +//! +//! # Why a cached set is safe to spend against +//! +//! Platform records at most one anchor per block, in the block-end +//! transaction that appends the block's notes, and prunes only by age (every +//! `shielded_anchor_pruning_interval` blocks it drops anchors older than +//! `shielded_anchor_retention_blocks`, always keeping the newest). So every +//! anchor in a recent set is still accepted unless it was already within the +//! TTL of the retention edge, a risk a fresh fetch shares because proving +//! takes comparable time. Two guards keep the cached set from selecting a +//! different anchor than a fresh fetch would: +//! +//! - **TTL** ([`RECORDED_ANCHOR_CACHE_TTL`]), measured from when the request +//! was sent, on both the monotonic and the wall clock (the monotonic clock +//! stops while a phone sleeps). +//! - **Tree-size pin**: a set is used only while the local tree still has the +//! leaf count `N` read before its request was sent. Holding leaf `N - 1` +//! means its block was committed, so every root of a tree of size `<= N` +//! that Platform will ever record already was. Any tree growth may add a +//! newer checkpoint the set lacks, so it retires the set. +//! +//! Both guards compare against a fresh fetch from an equally current node. A +//! node the SDK still accepts as up to date may lag a block behind and omit +//! the newest checkpoint; the cache then keeps that older, still recorded +//! anchor for up to the TTL, where a fresh fetch would retry the node's luck. +//! +//! The spend path also refetches when the cached set covers none of the +//! wallet's checkpoints, and drops the cache after any failed spend. The +//! stranded-reservation release pass never reads the cache (its fund-safety +//! argument needs a set fetched right before its note scan); it only fills it. + +use std::collections::HashSet; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, PoisonError}; +use std::time::{Duration, Instant, SystemTime}; + +use dash_sdk::platform::fetch_current_no_parameters::FetchCurrent; + +/// How long a fetched recorded-anchor set may be reused by the spend path. +/// +/// At most a few percent of the 1000-block retention window even at one block +/// per second, and the same order as a spend's own proving time, so it does +/// not change which anchors can expire mid-send. +pub const RECORDED_ANCHOR_CACHE_TTL: Duration = Duration::from_secs(30); + +/// Recorded-anchor set shared by the spend path. See the module docs. +#[derive(Debug)] +pub struct RecordedAnchorCache { + ttl: Duration, + state: Mutex, + prefetch_in_flight: AtomicBool, +} + +/// Held by the one background prefetch in flight; released on drop, so also +/// on error and cancellation. +pub(crate) struct PrefetchGuard<'a>(&'a AtomicBool); + +impl Drop for PrefetchGuard<'_> { + fn drop(&mut self) { + self.0.store(false, Ordering::Release); + } +} + +#[derive(Debug, Default)] +struct CacheState { + entry: Option, + /// When [`RecordedAnchorCache::invalidate`] last ran. Requests sent + /// before it may carry the set it meant to discard, so they are dropped. + invalidated_at: Option, +} + +#[derive(Debug, Clone)] +struct CachedAnchorSet { + anchors: Arc>, + /// When the request that produced `anchors` was sent. The set reflects + /// Platform's state at some later instant, so ageing from here is + /// conservative. + fetch_started: FetchStamp, + /// Local commitment-tree leaf count read before the request was sent. + tree_size: u64, +} + +/// When a recorded-anchor request was sent, on two clocks. +/// +/// `Instant` alone is not enough: on iOS / macOS it is an uptime clock and on +/// Android / Linux `CLOCK_MONOTONIC`, neither of which advances while the +/// device sleeps, so a set fetched before the phone was locked for hours +/// would still look seconds old. The wall clock keeps running through sleep +/// but can be set backwards. A set's age is the larger of the two readings, +/// and a wall clock that moved before the stamp counts as expired. +#[derive(Debug, Clone, Copy)] +pub(crate) struct FetchStamp { + monotonic: Instant, + wall: SystemTime, +} + +impl FetchStamp { + /// Stamp a request about to be sent. + pub(crate) fn now() -> Self { + Self { + monotonic: Instant::now(), + wall: SystemTime::now(), + } + } + + /// A stamp `age` in the past on both clocks. + #[cfg(test)] + pub(crate) fn ago(age: Duration) -> Self { + Self { + monotonic: Instant::now() + .checked_sub(age) + .expect("monotonic clock is past the requested age"), + wall: SystemTime::now() + .checked_sub(age) + .expect("wall clock is past the requested age"), + } + } + + /// Whether this stamp is younger than `ttl` on both clocks. + fn is_younger_than(&self, ttl: Duration) -> bool { + self.monotonic.elapsed() < ttl && self.wall.elapsed().is_ok_and(|wall_age| wall_age < ttl) + } +} + +impl Default for RecordedAnchorCache { + fn default() -> Self { + Self::new() + } +} + +impl RecordedAnchorCache { + /// An empty cache with the default [`RECORDED_ANCHOR_CACHE_TTL`]. + pub fn new() -> Self { + Self::with_ttl(RECORDED_ANCHOR_CACHE_TTL) + } + + /// An empty cache with a custom TTL. A zero TTL disables reuse: every + /// spend fetches the set, as it did before this cache existed. + pub fn with_ttl(ttl: Duration) -> Self { + Self { + ttl, + state: Mutex::new(CacheState::default()), + prefetch_in_flight: AtomicBool::new(false), + } + } + + /// The cached set, if one is younger than the TTL and was fetched while + /// the local tree had exactly `tree_size` leaves. + pub(crate) fn get(&self, tree_size: u64) -> Option>> { + let state = self.state.lock().unwrap_or_else(PoisonError::into_inner); + state + .entry + .as_ref() + .filter(|e| e.tree_size == tree_size && e.fetch_started.is_younger_than(self.ttl)) + .map(|e| Arc::clone(&e.anchors)) + } + + /// Record a freshly fetched set. `fetch_started` is when its request was + /// sent and `tree_size` the local tree's leaf count read before that. + /// Concurrent fetches resolve to the one sent last, and a fetch sent + /// before the last [`invalidate`](Self::invalidate) is dropped. + pub(crate) fn insert( + &self, + anchors: Arc>, + fetch_started: FetchStamp, + tree_size: u64, + ) { + let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner); + let sent = fetch_started.monotonic; + if state.invalidated_at.is_some_and(|at| sent <= at) + || state + .entry + .as_ref() + .is_some_and(|current| current.fetch_started.monotonic > sent) + { + return; + } + state.entry = Some(CachedAnchorSet { + anchors, + fetch_started, + tree_size, + }); + } + + /// Claim the background prefetch slot, or `None` while another prefetch + /// is in flight. Spends and the release pass never take it. + pub(crate) fn try_begin_prefetch(&self) -> Option> { + self.prefetch_in_flight + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .ok() + .map(|_| PrefetchGuard(&self.prefetch_in_flight)) + } + + /// Drop any cached set, and any still in flight; the next spend fetches + /// a fresh one. + pub fn invalidate(&self) { + let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner); + state.entry = None; + state.invalidated_at = Some(Instant::now()); + } +} + +/// Fetch and proof-verify Platform's current recorded anchor set for the +/// credit shielded pool. +pub(crate) async fn fetch_recorded_anchor_set( + sdk: &dash_sdk::Sdk, +) -> Result, dash_sdk::Error> { + let dash_sdk::query_types::ShieldedAnchors(anchors) = + dash_sdk::query_types::ShieldedAnchors::fetch_current(sdk).await?; + Ok(anchors.into_iter().collect()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn set(bytes: &[u8]) -> Arc> { + Arc::new(bytes.iter().map(|b| [*b; 32]).collect()) + } + + #[test] + fn should_return_a_fresh_set_for_the_same_tree_size() { + let cache = RecordedAnchorCache::new(); + assert!(cache.get(10).is_none(), "an empty cache has nothing"); + + cache.insert(set(&[1, 2]), FetchStamp::now(), 10); + + let hit = cache.get(10).expect("fresh set for the pinned tree size"); + assert!(hit.contains(&[1; 32]) && hit.contains(&[2; 32])); + } + + #[test] + fn should_miss_when_the_tree_grew_since_the_fetch() { + let cache = RecordedAnchorCache::new(); + cache.insert(set(&[1]), FetchStamp::now(), 10); + + assert!( + cache.get(11).is_none(), + "a sync appended after the fetch: the set may lack the newest checkpoint" + ); + } + + #[test] + fn should_expire_after_the_ttl() { + let cache = RecordedAnchorCache::with_ttl(Duration::from_secs(30)); + cache.insert(set(&[1]), FetchStamp::ago(Duration::from_secs(31)), 10); + + assert!(cache.get(10).is_none(), "never used past the TTL"); + } + + /// The monotonic clock stops while a phone sleeps: a set fetched before + /// hours of sleep reads as seconds old on it. The wall clock must still + /// expire it. + #[test] + fn should_expire_across_device_sleep_by_the_wall_clock() { + let cache = RecordedAnchorCache::with_ttl(Duration::from_secs(30)); + let slept = FetchStamp { + monotonic: Instant::now(), + wall: SystemTime::now() - Duration::from_secs(3 * 60 * 60), + }; + cache.insert(set(&[1]), slept, 10); + + assert!(cache.get(10).is_none(), "hours of wall time have passed"); + } + + /// A wall clock set backwards past the stamp gives no trustworthy age: + /// treat the set as expired rather than as fresh. + #[test] + fn should_expire_when_the_wall_clock_moved_backwards() { + let cache = RecordedAnchorCache::with_ttl(Duration::from_secs(30)); + let future_wall = FetchStamp { + monotonic: Instant::now(), + wall: SystemTime::now() + Duration::from_secs(60 * 60), + }; + cache.insert(set(&[1]), future_wall, 10); + + assert!(cache.get(10).is_none()); + } + + #[test] + fn should_never_hit_with_a_zero_ttl() { + let cache = RecordedAnchorCache::with_ttl(Duration::ZERO); + cache.insert(set(&[1]), FetchStamp::now(), 10); + + assert!(cache.get(10).is_none()); + } + + #[test] + fn should_keep_the_most_recently_sent_fetch() { + let cache = RecordedAnchorCache::new(); + let now = FetchStamp::now(); + let earlier = FetchStamp::ago(Duration::from_secs(1)); + + cache.insert(set(&[2]), now, 10); + // A slower request sent earlier completes afterwards: it must not + // overwrite the newer snapshot. + cache.insert(set(&[1]), earlier, 10); + + let hit = cache.get(10).expect("newer set retained"); + assert!(hit.contains(&[2; 32]) && !hit.contains(&[1; 32])); + } + + #[test] + fn should_drop_the_set_on_invalidate() { + let cache = RecordedAnchorCache::new(); + cache.insert(set(&[1]), FetchStamp::now(), 10); + + cache.invalidate(); + + assert!(cache.get(10).is_none()); + } + + #[test] + fn should_allow_one_prefetch_at_a_time() { + let cache = RecordedAnchorCache::new(); + let first = cache.try_begin_prefetch().expect("slot free"); + assert!(cache.try_begin_prefetch().is_none(), "overlapping prefetch"); + drop(first); + assert!(cache.try_begin_prefetch().is_some(), "released on drop"); + } + + /// A fetch sent before a failed spend invalidated the cache may return + /// the very anchor that spend had rejected: it must not refill the cache. + #[test] + fn should_drop_a_fetch_sent_before_invalidate() { + let cache = RecordedAnchorCache::new(); + let in_flight = FetchStamp::now(); + + cache.invalidate(); + cache.insert(set(&[1]), in_flight, 10); + assert!(cache.get(10).is_none(), "stale in-flight fetch dropped"); + + std::thread::sleep(Duration::from_millis(1)); + cache.insert(set(&[2]), FetchStamp::now(), 10); + assert!(cache.get(10).is_some(), "a fetch sent afterwards is kept"); + } +} diff --git a/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs b/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs index 884f44d8dda..3959705ae51 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs @@ -84,6 +84,7 @@ use tokio::sync::RwLock; use super::activity::ShieldedActivityStatus; use super::activity_recorder::{identity_redrive_output_cmxs, IdentityRecoveryError}; +use super::anchor_cache::{fetch_recorded_anchor_set, FetchStamp, RecordedAnchorCache}; use super::balance::{ ShieldedBalanceSource, ShieldedLocalBalanceSnapshot, ShieldedLocalBalanceState, }; @@ -140,6 +141,14 @@ pub struct NetworkShieldedCoordinator { /// delete-while-open race the prior architecture had. store: Arc>, + /// Most recently fetched Platform recorded-anchor set, reused by the + /// spend path while fresh so a send can skip that round trip. Filled by + /// every fetch: sends, [`prefetch_recorded_anchors`], and the + /// stranded-reservation release pass. See [`RecordedAnchorCache`]. + /// + /// [`prefetch_recorded_anchors`]: Self::prefetch_recorded_anchors + anchor_cache: RecordedAnchorCache, + /// Flat registry of every bound `(walletId, accountIndex)` /// pair's viewing keys, populated by /// [`register_wallet`](Self::register_wallet). The sync loop @@ -397,6 +406,7 @@ impl NetworkShieldedCoordinator { network, db_path, store: Arc::new(RwLock::new(store)), + anchor_cache: RecordedAnchorCache::new(), accounts: Arc::new(RwLock::new(BTreeMap::new())), persisters: Arc::new(RwLock::new(BTreeMap::new())), last_caught_up_at: std::sync::Mutex::new(None), @@ -477,6 +487,44 @@ impl NetworkShieldedCoordinator { &self.store } + /// The recorded-anchor cache the spend operations take alongside + /// [`store`](Self::store). + pub fn anchor_cache(&self) -> &RecordedAnchorCache { + &self.anchor_cache + } + + /// Warm the recorded-anchor cache ahead of a spend — e.g. when a send + /// screen opens — so the send itself can skip the round trip that + /// downloads and proof-verifies Platform's recorded anchor set. + /// + /// No-op when the cache already holds a set the next spend could use, or + /// while another prefetch is in flight. + /// The set stays usable for [`RECORDED_ANCHOR_CACHE_TTL`] and only while + /// no sync appends to the commitment tree; a send after that fetches as + /// usual. Purely an optimization — an error here just means the send + /// fetches for itself. + /// + /// [`RECORDED_ANCHOR_CACHE_TTL`]: super::anchor_cache::RECORDED_ANCHOR_CACHE_TTL + pub async fn prefetch_recorded_anchors(&self) -> Result<(), PlatformWalletError> { + let tree_size = self + .store + .read() + .await + .tree_size() + .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))?; + if self.anchor_cache.get(tree_size).is_some() { + return Ok(()); + } + let Some(_in_flight) = self.anchor_cache.try_begin_prefetch() else { + return Ok(()); + }; + let fetch_started = FetchStamp::now(); + let recorded = fetch_recorded_anchor_set(&self.sdk).await?; + self.anchor_cache + .insert(Arc::new(recorded), fetch_started, tree_size); + Ok(()) + } + /// Register every account of a newly-bound shielded wallet so /// future [`sync`](Self::sync) passes iterate it, and attach /// the wallet's [`WalletPersister`] so the coordinator can @@ -1273,6 +1321,7 @@ impl NetworkShieldedCoordinator { // half-reset store. The generation bump also prevents an in-flight // bind from restoring a pre-Clear snapshot afterward. self.hydrated.write().await.clear(); + self.anchor_cache.invalidate(); self.clear_generation .fetch_add(1, std::sync::atomic::Ordering::AcqRel); @@ -1669,10 +1718,15 @@ impl NetworkShieldedCoordinator { async fn prefetch_stranded_release( &self, subwallets: &[(SubwalletId, AccountViewingKeys)], - ) -> Option<(Vec<(SubwalletId, StalePendingSpend)>, HashSet<[u8; 32]>)> { + ) -> Option<( + Vec<(SubwalletId, StalePendingSpend)>, + Arc>, + )> { // Gather anchored reservations across every synced subwallet. The // common case is none — then the network round-trip is skipped. - let stale: Vec<(SubwalletId, StalePendingSpend)> = { + // The tree size, read under the same guard before the fetch, pins + // the set for the spend path's cache. + let (stale, tree_size): (Vec<(SubwalletId, StalePendingSpend)>, Option) = { let store = self.store.read().await; let mut acc = Vec::new(); for (id, _) in subwallets { @@ -1686,16 +1740,25 @@ impl NetworkShieldedCoordinator { ), } } - acc + (acc, store.tree_size().ok()) }; if stale.is_empty() { return None; } - use dash_sdk::platform::fetch_current_no_parameters::FetchCurrent; - match dash_sdk::query_types::ShieldedAnchors::fetch_current(&self.sdk).await { - Ok(dash_sdk::query_types::ShieldedAnchors(anchors)) => { - Some((stale, anchors.into_iter().collect())) + let fetch_started = FetchStamp::now(); + match fetch_recorded_anchor_set(&self.sdk).await { + Ok(anchors) => { + // Populate only: the release itself must judge against THIS + // pre-scan set, never a cached older one (see + // `RecordedAnchorCache`). The scan below usually grows the + // tree, which retires this entry for spends anyway. + let anchors = Arc::new(anchors); + if let Some(tree_size) = tree_size { + self.anchor_cache + .insert(Arc::clone(&anchors), fetch_started, tree_size); + } + Some((stale, anchors)) } Err(e) => { tracing::warn!( @@ -2784,11 +2847,19 @@ mod tests { /// Build a coordinator backed by a fresh file store under `dir`, with one /// wallet registered so `accounts` / `persisters` are both non-empty. async fn coordinator_with_one_wallet(dir: &std::path::Path) -> NetworkShieldedCoordinator { + coordinator_with_one_wallet_on(dir, dash_sdk::Sdk::new_mock()).await + } + + /// [`coordinator_with_one_wallet`] over a caller-configured mock SDK. + async fn coordinator_with_one_wallet_on( + dir: &std::path::Path, + sdk: dash_sdk::Sdk, + ) -> NetworkShieldedCoordinator { std::fs::create_dir_all(dir).expect("create temp dir"); let db_path = dir.join("tree.sqlite"); let store = FileBackedShieldedStore::open_path(&db_path, 100).expect("open file store"); let coordinator = NetworkShieldedCoordinator::new( - Arc::new(dash_sdk::Sdk::new_mock()), + Arc::new(sdk), dashcore::Network::Testnet, db_path, store, @@ -3822,6 +3893,55 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// A mock SDK that answers the recorded-anchor query with `anchors`. + async fn sdk_with_recorded_anchors(anchors: &[[u8; 32]]) -> dash_sdk::Sdk { + let mut sdk = dash_sdk::Sdk::new_mock(); + sdk.mock() + .expect_fetch( + dash_sdk::query_types::NoParamQuery {}, + Some(dash_sdk::query_types::ShieldedAnchors(anchors.to_vec())), + ) + .await + .expect("set recorded-anchor expectation"); + sdk + } + + /// The send-screen prefetch leaves a set the spend path can use at the + /// tree's current size, and a second call is served from it. + #[tokio::test] + async fn prefetch_recorded_anchors_populates_the_spend_cache() { + let dir = temp_dir("prefetch_anchors"); + let anchor = [0x5Au8; 32]; + let coordinator = + coordinator_with_one_wallet_on(&dir, sdk_with_recorded_anchors(&[anchor]).await).await; + let tree_size = coordinator.store.read().await.tree_size().unwrap(); + + coordinator + .prefetch_recorded_anchors() + .await + .expect("prefetch against the mocked anchor query"); + + let cached = coordinator + .anchor_cache() + .get(tree_size) + .expect("prefetched set is usable by the next spend"); + assert!(cached.contains(&anchor)); + + // Already warm: no second fetch (the mock would serve one, so check + // the entry is the same allocation rather than a replacement). + coordinator + .prefetch_recorded_anchors() + .await + .expect("warm prefetch is a no-op"); + let again = coordinator.anchor_cache().get(tree_size).unwrap(); + assert!( + Arc::ptr_eq(&cached, &again), + "warm prefetch must not refetch" + ); + + let _ = std::fs::remove_dir_all(&dir); + } + /// Cancelled while inside the caught-up cooldown: the cooldown's /// skip summary would report one caught-up wallet, which the /// manager would stamp and announce as a completed pass. The @@ -3843,4 +3963,112 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + + /// A prefetch that overlaps one already in flight doesn't fetch again. + #[tokio::test] + async fn prefetch_recorded_anchors_skips_while_another_is_in_flight() { + let dir = temp_dir("prefetch_in_flight"); + let coordinator = + coordinator_with_one_wallet_on(&dir, sdk_with_recorded_anchors(&[[0x5D; 32]]).await) + .await; + let tree_size = coordinator.store.read().await.tree_size().unwrap(); + + let in_flight = coordinator.anchor_cache().try_begin_prefetch().unwrap(); + coordinator.prefetch_recorded_anchors().await.unwrap(); + assert!(coordinator.anchor_cache().get(tree_size).is_none()); + + drop(in_flight); + coordinator.prefetch_recorded_anchors().await.unwrap(); + assert!(coordinator.anchor_cache().get(tree_size).is_some()); + + let _ = std::fs::remove_dir_all(&dir); + } + + /// The stranded-reservation release pass already fetches the recorded + /// set; that fetch also warms the spend path's cache. The release itself + /// must judge against the set it just fetched, never a cached one: a + /// fresh cached set that still lists the reservation's (since pruned) + /// anchor must not keep it alive. + #[tokio::test] + async fn stranded_release_prefetch_populates_the_spend_cache() { + let dir = temp_dir("release_populates_cache"); + let anchor = [0x5Bu8; 32]; + let coordinator = + coordinator_with_one_wallet_on(&dir, sdk_with_recorded_anchors(&[anchor]).await).await; + let id = SubwalletId::new([0x11; 32], 0); + { + let mut store = coordinator.store.write().await; + store.mark_pending(id, &[0xAA; 32]).expect("mark"); + store + .set_pending_spend(id, &[0xAA; 32], [0xAB; 32], [0xAC; 32]) + .expect("arm"); + } + let tree_size = coordinator.store.read().await.tree_size().unwrap(); + let subwallets = vec![( + id, + OrchardKeySet::from_seed(&[0x42u8; 64], dashcore::Network::Testnet, 0) + .expect("derive viewing keys") + .viewing_keys(), + )]; + // A fresh cached set from an earlier send still lists the + // reservation's anchor; Platform's current set (the mock) does not. + coordinator.anchor_cache().insert( + Arc::new([[0xAB; 32]].into_iter().collect()), + FetchStamp::now(), + tree_size, + ); + + let (snapshot, recorded) = coordinator + .prefetch_stranded_release(&subwallets) + .await + .expect("an armed reservation triggers the anchor fetch"); + + assert_eq!(snapshot.len(), 1); + assert!(recorded.contains(&anchor)); + assert!( + !recorded.contains(&[0xAB; 32]), + "the release judges against the fresh fetch, not the cache" + ); + let cached = coordinator + .anchor_cache() + .get(tree_size) + .expect("release-pass fetch populated the spend cache"); + assert!( + Arc::ptr_eq(&cached, &recorded), + "the fresh fetch replaced the cached set" + ); + + coordinator + .release_stranded_spends(snapshot, &recorded) + .await; + assert!( + coordinator + .store + .read() + .await + .stale_pending_spends(id) + .unwrap() + .is_empty(), + "the pruned-anchor reservation is released" + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + /// Clear drops the cached set along with the tree it was pinned to. + #[tokio::test] + async fn clear_invalidates_the_recorded_anchor_cache() { + let dir = temp_dir("clear_anchor_cache"); + let coordinator = coordinator_with_one_wallet(&dir).await; + coordinator.anchor_cache().insert( + Arc::new([[0x5C; 32]].into_iter().collect()), + FetchStamp::now(), + 0, + ); + + coordinator.clear().await.expect("clear"); + + assert!(coordinator.anchor_cache().get(0).is_none()); + let _ = std::fs::remove_dir_all(&dir); + } } diff --git a/packages/rs-platform-wallet/src/wallet/shielded/mod.rs b/packages/rs-platform-wallet/src/wallet/shielded/mod.rs index e5c92c28be6..7df0e2a3212 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/mod.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/mod.rs @@ -35,6 +35,7 @@ pub mod wallet_drain; pub use wallet_drain::{WalletDrainPreview, WalletDrainPreviewStatus}; pub mod activity; pub mod activity_recorder; +pub mod anchor_cache; pub mod balance; #[cfg(test)] mod balance_tests; @@ -60,6 +61,7 @@ pub use activity::{ ScanDeriveInput, ShieldedActivityEntry, ShieldedActivityKind, ShieldedActivityStatus, ShieldedDirection, }; +pub use anchor_cache::{RecordedAnchorCache, RECORDED_ANCHOR_CACHE_TTL}; pub use balance::{ ShieldedBalanceSource, ShieldedLocalAccountBalance, ShieldedLocalBalanceSnapshot, ShieldedLocalBalanceState, diff --git a/packages/rs-platform-wallet/src/wallet/shielded/operations.rs b/packages/rs-platform-wallet/src/wallet/shielded/operations.rs index 8a8f8a4b39b..4700765f2cd 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/operations.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/operations.rs @@ -23,6 +23,7 @@ use super::activity::{ShieldedActivityKind, ShieldedActivityStatus, ShieldedDire use super::activity_recorder::{ build_pending_entry, changeset_for_entry, non_zero_memo, with_status, LiveEntryParams, }; +use super::anchor_cache::{fetch_recorded_anchor_set, FetchStamp, RecordedAnchorCache}; use super::keys::{AccountViewingKeys, OrchardKeySet}; use super::note_selection::{ ensure_allowed_denomination, ensure_positive_amount, select_notes_for_denomination, @@ -39,7 +40,6 @@ use std::collections::{BTreeMap, HashSet}; use std::sync::Arc; use dash_sdk::dapi_grpc::platform::v0::ResponseMetadata; -use dash_sdk::platform::fetch_current_no_parameters::FetchCurrent; use dash_sdk::platform::transition::broadcast::BroadcastStateTransition; use dash_sdk::platform::transition::identity_create_from_shielded_pool::IdentityCreateFromShieldedPool; use dash_sdk::platform::Fetch; @@ -1357,6 +1357,7 @@ mod shield_from_identity_build_error_tests { pub async fn identity_top_up_from_pool( sdk: &Arc, store: &Arc>, + anchor_cache: &RecordedAnchorCache, persister: Option<&WalletPersister>, wallet_id: WalletId, keys: &OrchardKeySet, @@ -1368,6 +1369,7 @@ pub async fn identity_top_up_from_pool( identity_top_up_from_pool_with_metadata( sdk, store, + anchor_cache, persister, wallet_id, keys, @@ -1387,6 +1389,7 @@ pub(in crate::wallet) async fn identity_top_up_from_pool_with_metadata< >( sdk: &Arc, store: &Arc>, + anchor_cache: &RecordedAnchorCache, persister: Option<&WalletPersister>, wallet_id: WalletId, keys: &OrchardKeySet, @@ -1414,7 +1417,8 @@ pub(in crate::wallet) async fn identity_top_up_from_pool_with_metadata< let mut pending_entry = None; let result = async { - let (spends, anchor) = extract_spends_and_anchor(sdk, store, &selected_notes).await?; + let (spends, anchor) = + extract_spends_and_anchor(sdk, store, anchor_cache, &selected_notes).await?; let anchor_bytes = anchor.to_bytes(); let (state_transition, fee_used) = build_identity_top_up_from_shielded_pool_transition( @@ -1478,6 +1482,10 @@ pub(in crate::wallet) async fn identity_top_up_from_pool_with_metadata< .await } .await; + // The anchor may have been rejected: make the retry fetch a fresh set. + if result.is_err() { + anchor_cache.invalidate(); + } match result { Ok((proof, metadata)) => { @@ -1568,6 +1576,7 @@ pub(in crate::wallet) async fn identity_top_up_from_pool_with_metadata< pub async fn unshield( sdk: &Arc, store: &Arc>, + anchor_cache: &RecordedAnchorCache, persister: Option<&WalletPersister>, wallet_id: WalletId, keys: &OrchardKeySet, @@ -1611,7 +1620,8 @@ pub async fn unshield( // it. A build failure leaves it `None` and records nothing. let mut pending_entry = None; let result = async { - let (spends, anchor) = extract_spends_and_anchor(sdk, store, &selected_notes).await?; + let (spends, anchor) = + extract_spends_and_anchor(sdk, store, anchor_cache, &selected_notes).await?; // Capture the recorded anchor before the builder consumes it, so a // broadcast-accepted-but-unconfirmed spend can be auto-released once // this anchor is pruned from Platform's recorded set. @@ -1677,6 +1687,10 @@ pub async fn unshield( .await } .await; + // The anchor may have been rejected: make the retry fetch a fresh set. + if result.is_err() { + anchor_cache.invalidate(); + } match result { Ok(_) => { @@ -1760,6 +1774,7 @@ pub async fn unshield( pub async fn transfer( sdk: &Arc, store: &Arc>, + anchor_cache: &RecordedAnchorCache, persister: Option<&WalletPersister>, wallet_id: WalletId, keys: &OrchardKeySet, @@ -1790,7 +1805,8 @@ pub async fn transfer( let mut pending_entry = None; let result = async { - let (spends, anchor) = extract_spends_and_anchor(sdk, store, &selected_notes).await?; + let (spends, anchor) = + extract_spends_and_anchor(sdk, store, anchor_cache, &selected_notes).await?; // Capture the recorded anchor before the builder consumes it, so a // broadcast-accepted-but-unconfirmed spend can be auto-released once // this anchor is pruned from Platform's recorded set. @@ -1856,6 +1872,10 @@ pub async fn transfer( .await } .await; + // The anchor may have been rejected: make the retry fetch a fresh set. + if result.is_err() { + anchor_cache.invalidate(); + } match result { Ok(_) => { @@ -1916,6 +1936,7 @@ pub async fn transfer( pub async fn withdraw( sdk: &Arc, store: &Arc>, + anchor_cache: &RecordedAnchorCache, persister: Option<&WalletPersister>, wallet_id: WalletId, keys: &OrchardKeySet, @@ -1957,7 +1978,8 @@ pub async fn withdraw( let mut pending_entry = None; let result = async { - let (spends, anchor) = extract_spends_and_anchor(sdk, store, &selected_notes).await?; + let (spends, anchor) = + extract_spends_and_anchor(sdk, store, anchor_cache, &selected_notes).await?; // Capture the recorded anchor before the builder consumes it, so a // broadcast-accepted-but-unconfirmed spend can be auto-released once // this anchor is pruned from Platform's recorded set. @@ -2025,6 +2047,10 @@ pub async fn withdraw( .await } .await; + // The anchor may have been rejected: make the retry fetch a fresh set. + if result.is_err() { + anchor_cache.invalidate(); + } match result { Ok(_) => { @@ -2102,6 +2128,7 @@ pub async fn withdraw( pub async fn identity_create_from_shielded_pool( sdk: &Arc, store: &Arc>, + anchor_cache: &RecordedAnchorCache, persister: Option<&WalletPersister>, wallet_id: WalletId, keys: &OrchardKeySet, @@ -2163,7 +2190,8 @@ where // outer match; it lives here so the flip can see it. let mut pending_entry = None; let result = async { - let (spends, anchor) = extract_spends_and_anchor(sdk, store, &selected_notes).await?; + let (spends, anchor) = + extract_spends_and_anchor(sdk, store, anchor_cache, &selected_notes).await?; // Capture the recorded anchor before the builder consumes it, so a // broadcast-accepted-but-unconfirmed create can be auto-released once // this anchor is pruned from Platform's recorded set. @@ -2387,6 +2415,10 @@ where Ok::<(Identifier, Identity), PlatformWalletError>((identity.id(), identity)) } .await; + // The anchor may have been rejected: make the retry fetch a fresh set. + if result.is_err() { + anchor_cache.invalidate(); + } match result { Ok((identity_id, identity)) => { @@ -2549,7 +2581,8 @@ const MAX_ANCHOR_PROBE_DEPTH: usize = 100; /// frequently a value Platform never recorded — building against it /// unconditionally is what made such spends fail and never land. /// -/// This fetches Platform's recorded anchor set (outside the store lock), then +/// This obtains Platform's recorded anchor set — from `anchor_cache` when it +/// holds a usable one, otherwise fetched (outside the store lock) — then /// selects the shallowest checkpoint depth whose root is in that set — depth 0 /// being the fully-synced fast path — witnessing every note at that same depth /// so the anchor and the authentication paths agree (the builder derives the @@ -2561,6 +2594,7 @@ const MAX_ANCHOR_PROBE_DEPTH: usize = 100; async fn extract_spends_and_anchor( sdk: &Arc, store: &Arc>, + anchor_cache: &RecordedAnchorCache, notes: &[ShieldedNote], ) -> Result<(Vec, Anchor), PlatformWalletError> { // Nothing selected — fail before the network round-trip. @@ -2570,12 +2604,73 @@ async fn extract_spends_and_anchor( )); } + select_spends_with_anchor_cache(store, anchor_cache, notes, || async { + Ok(fetch_recorded_anchor_set(sdk).await?) + }) + .await +} + +/// [`extract_spends_and_anchor`]'s anchor-set sourcing, with the network fetch +/// injected so the cache policy is testable without an SDK. +/// +/// 1. If the store lock is free and `anchor_cache` holds a set that is younger +/// than its TTL and was fetched at the tree's current size, probe against +/// it under that one read guard. A selection, or any error other than +/// `ShieldedNoRecordedAnchor`, is final — no network round trip. +/// 2. Otherwise (no usable set, the lock is contended, or the cached set +/// covers none of the checkpoints) `fetch` a fresh set OUTSIDE the store +/// lock, record it in the cache when the pre-fetch tree size was read, +/// and probe against it. +/// +/// See [`RecordedAnchorCache`] for why a cached set selects the same anchor a +/// fresh fetch from an equally current node would. +async fn select_spends_with_anchor_cache( + store: &Arc>, + anchor_cache: &RecordedAnchorCache, + notes: &[ShieldedNote], + fetch: F, +) -> Result<(Vec, Anchor), PlatformWalletError> +where + S: ShieldedStore, + F: FnOnce() -> Fut, + Fut: std::future::Future, PlatformWalletError>>, +{ + // `try_read`, not `read().await`: when a sync holds the store, waiting + // here would serialize the lock wait in front of the fetch below, which + // otherwise overlaps it. A contended lock simply takes the fetch path + // and leaves the cache untouched (no pre-fetch tree size to pin). + let tree_size_before_fetch = match store.try_read() { + Ok(guard) => match guard.tree_size() { + Ok(tree_size) => { + if let Some(recorded) = anchor_cache.get(tree_size) { + match select_recorded_spends(&*guard, notes, &recorded) { + Err(PlatformWalletError::ShieldedNoRecordedAnchor(_)) => { + debug!( + tree_size, + "cached recorded-anchor set covers no checkpoint; refetching" + ); + } + selected => return selected, + } + } + Some(tree_size) + } + Err(e) => { + debug!(error = %e, "tree_size unavailable; skipping the recorded-anchor cache"); + None + } + }, + Err(_) => None, + }; + // Fetch the recorded anchor set OUTSIDE the store lock so the network // round-trip doesn't serialize with other store users, and so the lock is // held only for the mutually-consistent depth/witness probe below. - let dash_sdk::query_types::ShieldedAnchors(recorded_anchors) = - dash_sdk::query_types::ShieldedAnchors::fetch_current(sdk).await?; - let recorded: HashSet<[u8; 32]> = recorded_anchors.into_iter().collect(); + let fetch_started = FetchStamp::now(); + let recorded = Arc::new(fetch().await?); + if let Some(tree_size) = tree_size_before_fetch { + anchor_cache.insert(Arc::clone(&recorded), fetch_started, tree_size); + } // Hold a single read lock across the whole probe so the checkpoint depths // and the per-note witnesses stay mutually consistent: a concurrent sync @@ -4824,4 +4919,354 @@ mod select_recorded_spends_tests { Ok(_) => panic!("expected ShieldedNoRecordedAnchor, got Ok"), } } + + // --------------------------------------------------------------------- + // Recorded-anchor cache on the spend path (`select_spends_with_anchor_cache`) + // --------------------------------------------------------------------- + + use crate::wallet::shielded::anchor_cache::RecordedAnchorCache; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::Duration; + + /// A wallet tree one block-boundary checkpoint deep, then left mid-block by + /// an index-chunk sync: the spend must select the depth-1 root. Returns the + /// store, the owned note, and that recorded block-boundary root. + fn mid_block_store( + tag: &str, + ) -> ( + std::path::PathBuf, + Arc>, + ShieldedNote, + [u8; 32], + ) { + let path = temp_tree_path(tag); + let mut store = FileBackedShieldedStore::open_path(&path, 100).unwrap(); + let note = real_note(0); + store.append_commitment(¬e.cmx, true).unwrap(); + store.append_commitment(&filler_cmx(0xA1), true).unwrap(); + store.append_commitment(&filler_cmx(0xA2), true).unwrap(); + store.checkpoint_tree(3).unwrap(); + let recorded_root = store.tree_anchor().unwrap(); + store.append_commitment(&filler_cmx(0xC1), true).unwrap(); + store.checkpoint_tree(4).unwrap(); + (path, Arc::new(RwLock::new(store)), note, recorded_root) + } + + /// A stand-in for the network fetch that counts how often it is called. + fn counting_fetch<'a>( + calls: &'a AtomicUsize, + recorded: &HashSet<[u8; 32]>, + ) -> impl FnOnce() -> std::future::Ready, PlatformWalletError>> + 'a + { + let recorded = recorded.clone(); + move || { + calls.fetch_add(1, Ordering::SeqCst); + std::future::ready(Ok(recorded)) + } + } + + async fn spend_anchor( + store: &Arc>, + cache: &RecordedAnchorCache, + note: &ShieldedNote, + calls: &AtomicUsize, + platform_set: &HashSet<[u8; 32]>, + ) -> Result<[u8; 32], PlatformWalletError> { + select_spends_with_anchor_cache( + store, + cache, + std::slice::from_ref(note), + counting_fetch(calls, platform_set), + ) + .await + .map(|(_, anchor)| anchor.to_bytes()) + } + + async fn tree_size(store: &Arc>) -> u64 { + store.read().await.tree_size().unwrap() + } + + /// The measurement behind the change: three back-to-back sends cost three + /// anchor-set round trips without the cache (a zero TTL reproduces the + /// old always-fetch behavior) and one with it — and every send still + /// selects the same recorded anchor. + #[tokio::test] + async fn should_fetch_the_anchor_set_once_for_back_to_back_sends() { + let (path, store, note, recorded_root) = mid_block_store("cache_count"); + let platform_set: HashSet<[u8; 32]> = [recorded_root].into_iter().collect(); + + let uncached = RecordedAnchorCache::with_ttl(Duration::ZERO); + let uncached_calls = AtomicUsize::new(0); + for _ in 0..3 { + let anchor = spend_anchor(&store, &uncached, ¬e, &uncached_calls, &platform_set) + .await + .expect("recorded checkpoint covers the note"); + assert_eq!(anchor, recorded_root); + } + + let cached = RecordedAnchorCache::new(); + let cached_calls = AtomicUsize::new(0); + for _ in 0..3 { + let anchor = spend_anchor(&store, &cached, ¬e, &cached_calls, &platform_set) + .await + .expect("recorded checkpoint covers the note"); + assert_eq!( + anchor, recorded_root, + "the cache never changes the selection" + ); + } + + let _ = std::fs::remove_file(&path); + assert_eq!( + uncached_calls.load(Ordering::SeqCst), + 3, + "before: one fetch per send" + ); + assert_eq!( + cached_calls.load(Ordering::SeqCst), + 1, + "after: one fetch for all three" + ); + } + + /// A set already in the cache — from a send-screen prefetch or the + /// coordinator's release pass — takes the fetch off the send entirely. + #[tokio::test] + async fn should_not_fetch_when_a_prefetched_set_is_fresh() { + let (path, store, note, recorded_root) = mid_block_store("cache_prefetched"); + let platform_set: HashSet<[u8; 32]> = [recorded_root].into_iter().collect(); + let cache = RecordedAnchorCache::new(); + cache.insert( + Arc::new(platform_set.clone()), + FetchStamp::now(), + tree_size(&store).await, + ); + + let calls = AtomicUsize::new(0); + let anchor = spend_anchor(&store, &cache, ¬e, &calls, &platform_set) + .await + .expect("prefetched set covers the note"); + + let _ = std::fs::remove_file(&path); + assert_eq!(anchor, recorded_root); + assert_eq!(calls.load(Ordering::SeqCst), 0, "no round trip on the send"); + } + + /// A cached set that covers none of the wallet's checkpoints (e.g. served + /// by a node lagging behind the one the wallet synced from) is not the + /// last word: the send refetches, succeeds, and refreshes the cache. + #[tokio::test] + async fn should_refetch_and_refresh_when_the_cached_set_covers_no_checkpoint() { + let (path, store, note, recorded_root) = mid_block_store("cache_norecorded"); + let size = tree_size(&store).await; + let cache = RecordedAnchorCache::new(); + cache.insert( + Arc::new([[0xEE; 32]].into_iter().collect()), + FetchStamp::now(), + size, + ); + let platform_set: HashSet<[u8; 32]> = [recorded_root].into_iter().collect(); + + let calls = AtomicUsize::new(0); + let anchor = spend_anchor(&store, &cache, ¬e, &calls, &platform_set) + .await + .expect("the fresh set covers the note"); + assert_eq!(anchor, recorded_root); + assert_eq!(calls.load(Ordering::SeqCst), 1, "refetched once"); + + let refreshed = cache + .get(size) + .expect("the fresh set replaced the stale one"); + assert!(refreshed.contains(&recorded_root)); + spend_anchor(&store, &cache, ¬e, &calls, &platform_set) + .await + .expect("served from the refreshed cache"); + + let _ = std::fs::remove_file(&path); + assert_eq!( + calls.load(Ordering::SeqCst), + 1, + "the next send reuses the refresh" + ); + } + + /// A set older than the TTL is never used, even when it would still + /// select a valid anchor. + #[tokio::test] + async fn should_refetch_once_the_cached_set_is_past_its_ttl() { + let (path, store, note, recorded_root) = mid_block_store("cache_ttl"); + let platform_set: HashSet<[u8; 32]> = [recorded_root].into_iter().collect(); + let cache = RecordedAnchorCache::with_ttl(Duration::from_secs(30)); + let expired = FetchStamp::ago(Duration::from_secs(31)); + cache.insert( + Arc::new(platform_set.clone()), + expired, + tree_size(&store).await, + ); + + let calls = AtomicUsize::new(0); + spend_anchor(&store, &cache, ¬e, &calls, &platform_set) + .await + .expect("the fresh set covers the note"); + + let _ = std::fs::remove_file(&path); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + + /// A sync that grows the tree after the set was fetched can add a + /// checkpoint Platform recorded later than that snapshot. The cached set + /// would still find the OLDER checkpoint — so it must not be used: the + /// send refetches and selects the newest recorded root, exactly as it + /// would without the cache. + #[tokio::test] + async fn should_refetch_when_a_sync_grew_the_tree_after_the_fetch() { + let path = temp_tree_path("cache_grown"); + let mut inner = FileBackedShieldedStore::open_path(&path, 100).unwrap(); + let note = real_note(0); + inner.append_commitment(¬e.cmx, true).unwrap(); + inner.append_commitment(&filler_cmx(0xA1), true).unwrap(); + inner.checkpoint_tree(2).unwrap(); + let root_block1 = inner.tree_anchor().unwrap(); + let store = Arc::new(RwLock::new(inner)); + + // Prefetched while the wallet was at block 1. + let cache = RecordedAnchorCache::new(); + cache.insert( + Arc::new([root_block1].into_iter().collect()), + FetchStamp::now(), + tree_size(&store).await, + ); + + // A sync then appends block 2 and checkpoints on its boundary. + let root_block2 = { + let mut s = store.write().await; + s.append_commitment(&filler_cmx(0xB1), true).unwrap(); + s.append_commitment(&filler_cmx(0xB2), true).unwrap(); + s.checkpoint_tree(4).unwrap(); + s.tree_anchor().unwrap() + }; + let platform_set: HashSet<[u8; 32]> = [root_block1, root_block2].into_iter().collect(); + + let calls = AtomicUsize::new(0); + let anchor = spend_anchor(&store, &cache, ¬e, &calls, &platform_set) + .await + .expect("the fresh set covers the note"); + + let _ = std::fs::remove_file(&path); + assert_eq!( + calls.load(Ordering::SeqCst), + 1, + "the grown tree retired the cached set" + ); + assert_eq!( + anchor, root_block2, + "selects the newest recorded root, not the older one the cached set knew" + ); + } + + /// While a sync holds the store, the send does not queue for the lock + /// just to consult the cache: it fetches (overlapping the lock wait, as + /// before the cache existed) and leaves the cache alone, having no + /// pre-fetch tree size to pin a new set to. + #[tokio::test] + async fn should_fetch_without_waiting_while_the_store_is_write_locked() { + let (path, store, note, recorded_root) = mid_block_store("cache_contended"); + let size = tree_size(&store).await; + let marker = [0xEE; 32]; + let cache = RecordedAnchorCache::new(); + cache.insert( + Arc::new([recorded_root, marker].into_iter().collect()), + FetchStamp::now(), + size, + ); + let platform_set: HashSet<[u8; 32]> = [recorded_root].into_iter().collect(); + + let sync_guard = store.write().await; + let calls = AtomicUsize::new(0); + let fetch = { + let calls = &calls; + let platform_set = platform_set.clone(); + move || { + // The "sync" finishes while the fetch is in flight. + drop(sync_guard); + calls.fetch_add(1, Ordering::SeqCst); + std::future::ready(Ok(platform_set)) + } + }; + // Bounded: waiting on the held write lock (instead of `try_read`) + // would deadlock here, since only the fetch releases it. + let anchor = tokio::time::timeout( + Duration::from_secs(5), + select_spends_with_anchor_cache(&store, &cache, std::slice::from_ref(¬e), fetch), + ) + .await + .expect("the cache check must not wait for the store lock") + .map(|(_, anchor)| anchor.to_bytes()) + .expect("the fresh set covers the note"); + + let _ = std::fs::remove_file(&path); + assert_eq!(anchor, recorded_root); + assert_eq!( + calls.load(Ordering::SeqCst), + 1, + "contended lock takes the fetch path" + ); + assert!( + cache.get(size).expect("entry kept").contains(&marker), + "a fetch without a pre-fetch tree size must not replace the cached set" + ); + } + + /// A spend that fails after its anchor was chosen leaves no cached set + /// behind, so the retry selects against a fresh one. Driven through the + /// real `unshield` op: the cached set is used (tree size matches), and + /// the selected note has no witness in the tree, so the op fails at + /// anchor selection — before any proving or network I/O. + #[tokio::test] + async fn should_invalidate_the_cache_when_a_spend_fails() { + let (path, store, _, recorded_root) = mid_block_store("cache_failed_spend"); + let keys = OrchardKeySet::from_seed(&[0x42; 32], Network::Testnet, 0) + .expect("ZIP-32 derivation from a fixed seed"); + let wallet_id: WalletId = [0x31; 32]; + let id = SubwalletId::new(wallet_id, 0); + + // An unspent note at a position the tree doesn't hold: selectable, + // but never witnessable. Its value covers any shielded fee. + let mut unwitnessable = real_note(50); + unwitnessable.value = 1_000_000_000_000_000; + store + .write() + .await + .save_note(id, &unwitnessable) + .expect("save note"); + + let size = tree_size(&store).await; + let cache = RecordedAnchorCache::new(); + cache.insert( + Arc::new([recorded_root].into_iter().collect()), + FetchStamp::now(), + size, + ); + + let sdk = Arc::new(dash_sdk::Sdk::new_mock()); + let to = PlatformAddress::P2pkh([0x22; 20]); + let prover = crate::wallet::shielded::CachedOrchardProver::new(); + let result = unshield( + &sdk, &store, &cache, None, wallet_id, &keys, 0, &to, 1_000, &&prover, + ) + .await; + + let _ = std::fs::remove_file(&path); + assert!( + matches!( + result, + Err(PlatformWalletError::ShieldedMerkleWitnessUnavailable(_)) + ), + "expected the anchor-selection failure, got {result:?}" + ); + assert!( + cache.get(size).is_none(), + "a failed spend must drop the cached anchor set" + ); + } } diff --git a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManagerShieldedSpendPrefetch.swift b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManagerShieldedSpendPrefetch.swift new file mode 100644 index 00000000000..bc49d908ec4 --- /dev/null +++ b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManagerShieldedSpendPrefetch.swift @@ -0,0 +1,23 @@ +import Foundation +import DashSDKFFI + +extension PlatformWalletManager { + /// Start fetching Platform's recorded shielded-anchor set for an + /// upcoming spend, in the background. Call it when a shielded send + /// screen (transfer, unshield, withdraw, identity top-up/create from + /// the pool) opens: a send confirmed within ~30 s — before the next + /// shielded sync grows the commitment tree — then skips that network + /// round trip. Returns immediately and is cheap to repeat; a failed + /// prefetch only means the send fetches for itself. + /// + /// Throws when the manager isn't configured or has no shielded + /// support configured. + public func prefetchShieldedSpendAnchors() throws { + guard isConfigured, handle != NULL_HANDLE else { + throw PlatformWalletError.invalidHandle( + "PlatformWalletManager not configured" + ) + } + try platform_wallet_manager_shielded_prefetch_spend_anchors(handle).check() + } +} diff --git a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Core/Views/SendTransactionView.swift b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Core/Views/SendTransactionView.swift index de48fdf33bb..7d0caea2dea 100644 --- a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Core/Views/SendTransactionView.swift +++ b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Core/Views/SendTransactionView.swift @@ -360,6 +360,20 @@ struct SendTransactionView: View { // network-tracked platform version), which the view model // deliberately doesn't hold. resolveShieldedFees() + // With shielded funds to spend, fetch Platform's recorded + // shielded-anchor set while the form is filled in, so a + // shielded send confirmed soon after skips that round trip. + // Best-effort: throws only when shielded isn't configured. + if shieldedBalance > 0 { + try? walletManager.prefetchShieldedSpendAnchors() + } + } + .onChange(of: shieldedBalance) { oldBalance, newBalance in + // Notes that land after `.onAppear` (sync still persisting) + // would otherwise leave this screen without a prefetch. + if oldBalance == 0, newBalance > 0 { + try? walletManager.prefetchShieldedSpendAnchors() + } } .onChange(of: viewModel.detectedAddressType) { _, _ in autoSelectSource()