diff --git a/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs b/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs index 7459fe575d7..097efd32d32 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs @@ -51,7 +51,7 @@ //! //! [`sync_notes_across`]: super::sync::sync_notes_across -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashSet}; use std::path::PathBuf; use std::sync::Arc; @@ -80,7 +80,7 @@ use tokio::sync::RwLock; use super::file_store::FileBackedShieldedStore; use super::keys::AccountViewingKeys; -use super::store::{ShieldedStore, SubwalletId}; +use super::store::{ShieldedStore, StalePendingSpend, SubwalletId}; use super::CAUGHT_UP_COOLDOWN; use crate::manager::shielded_sync::{ShieldedSyncPassSummary, WalletShieldedOutcome}; use crate::wallet::persister::WalletPersister; @@ -593,6 +593,21 @@ impl NetworkShieldedCoordinator { return summary; } + // Snapshot armed reservations and prefetch Platform's recorded + // anchor set BEFORE the note scan. The release pass after the scan + // may only act on this pre-scan pair: an anchor already absent from + // a PRE-scan set was pruned before the scan began, so a spend that + // did execute did so at a height the scan covers — the scan's + // spent-note reconcile clears its reservation, which the release + // pass honors via `clear_pending`'s return value. A set fetched + // AFTER the scan would leave a window (spend executes past the + // scan's coverage, anchor pruned before the fetch) where a landed + // spend could be wrongly released. Reservations armed DURING the + // scan are deliberately absent from the snapshot — their anchors + // may be newer than this set — and get checked next pass. Skips the + // network call entirely when nothing is armed (the common case). + let stranded_release = self.prefetch_stranded_release(&subwallets).await; + // ONE SDK call covers every registered IVK on the network. // Snapshot the optional progress handler installed by the // manager; sync_notes_across feeds it into the SDK's chunk @@ -622,6 +637,17 @@ impl NetworkShieldedCoordinator { // newly-spent counts and the spend records ride the same // `notes` result and `notes.changeset` the receipts do. let newly_spent_per_sub = notes.per_subwallet_newly_spent.clone(); + + // Residual-spend reconcile: `sync_notes_across` above marked every + // landed spend (clearing its reservation). Now release any still- + // pending pre-scan reservation whose recorded anchor Platform had + // already pruned before the scan — a spend broadcast-accepted but + // never landed, otherwise stranded for the session. Runs before the + // balance read so freed notes are reflected in this pass's balances. + if let Some((snapshot, recorded)) = stranded_release { + self.release_stranded_spends(snapshot, &recorded).await; + } + let balances_per_sub = match super::sync::balances_across(&self.store, &subwallets).await { Ok(r) => r, Err(e) => return self.fail_all_wallets(&subwallets, &e), @@ -730,6 +756,193 @@ impl NetworkShieldedCoordinator { .unwrap_or(0) } + /// Pre-scan half of the stranded-reservation release: snapshot the + /// anchored reservations armed right now and fetch Platform's recorded + /// anchor set, or `None` when there is nothing to do. + /// + /// MUST run before the pass's note scan — the release's fund-safety + /// argument ([`Self::release_stranded_spends`]) rests on both the + /// snapshot and the set predating the scan's spent-note reconcile. + /// + /// Skips the network round-trip when no anchored reservation exists (the + /// common case), and treats a recorded-anchor fetch failure as + /// non-fatal — a sync must not fail because that query hiccupped; the + /// release simply waits for the next pass. + async fn prefetch_stranded_release( + &self, + subwallets: &[(SubwalletId, AccountViewingKeys)], + ) -> Option<(Vec<(SubwalletId, StalePendingSpend)>, HashSet<[u8; 32]>)> { + // Gather anchored reservations across every synced subwallet. The + // common case is none — then the network round-trip is skipped. + let stale: Vec<(SubwalletId, StalePendingSpend)> = { + let store = self.store.read().await; + let mut acc = Vec::new(); + for (id, _) in subwallets { + match store.stale_pending_spends(*id) { + Ok(entries) => acc.extend(entries.into_iter().map(|entry| (*id, entry))), + Err(e) => tracing::warn!( + wallet_id = %hex::encode(id.wallet_id), + account = id.account_index, + error = %e, + "shielded reservation release: stale_pending_spends failed; skipping subwallet" + ), + } + } + acc + }; + 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())) + } + Err(e) => { + tracing::warn!( + error = %e, + "shielded reservation release: failed to fetch the recorded anchor set; \ + skipping this pass" + ); + None + } + } + } + + /// Release any stranded shielded-spend reservation whose recorded + /// anchor Platform has pruned. + /// + /// A spend that returns broadcast-accepted-but-unconfirmed keeps its + /// note reservation (so a retry can't double-spend a note that may in + /// fact have landed), released only when the tx lands or the app + /// restarts. A spend accepted but that never lands would otherwise + /// strand its notes for the whole session. Here, after the pass's + /// spent-note reconcile, any reservation from the PRE-scan `snapshot` + /// whose anchor is absent from the PRE-scan `recorded` set is provably + /// dead: `validate_anchor_exists` accepts a spend only while its anchor + /// is retained (`shielded_anchor_retention_blocks = 1000`), so once the + /// anchor is pruned the transition can never execute. The notes are + /// freed and the linked activity row flipped to Failed so the UI shows + /// a clear, retryable failure instead of "Pending" forever. The + /// on-chain nullifier set stays the authoritative double-spend guard. + /// + /// Fund-safe by construction, resting on two ordering facts: + /// + /// 1. Both inputs predate the scan ([`Self::prefetch_stranded_release`]), + /// and the scan's spent-note reconcile ran to completion in between. + /// An anchor absent from the pre-scan set was pruned before the scan + /// began (a pruned root can never re-enter the set — the tree only + /// grows), so if that spend nevertheless executed, it executed at a + /// height the scan covered — the reconcile already cleared its + /// reservation, which fact 2 catches. + /// 2. Each release is a check-and-clear under a single store write + /// guard: the reservation is cleared only while the nullifier is + /// still armed with the snapshot's exact anchor + activity. A + /// reservation resolved concurrently (the reconcile above, or the + /// spend's own result-wait confirming mid-pass) — or cleared and + /// re-armed by a retry with a fresh anchor — is skipped, and its + /// activity row is left alone. + /// + /// A still-recorded (slow-but-landing) spend is never released, and an + /// anchor-less reservation (reserved but not yet built) is never in the + /// snapshot (`stale_pending_spends` returns only anchored entries). + async fn release_stranded_spends( + &self, + snapshot: Vec<(SubwalletId, StalePendingSpend)>, + recorded: &HashSet<[u8; 32]>, + ) { + // Persister map is only written on wallet register/unregister — + // one read acquisition covers the whole loop. + let persisters = self.persisters.read().await; + // Every note of a multi-note spend carries the same activity id: + // flip each linked row once, not once per nullifier. + let mut flipped: HashSet<[u8; 32]> = HashSet::new(); + for (id, (nullifier, anchor, activity_id)) in snapshot { + // Fund-safety invariant: release ONLY when the anchor is absent + // from the recorded set. A still-recorded anchor may yet land. + if recorded.contains(&anchor) { + continue; + } + // Check-and-clear under ONE write acquisition: the snapshot + // entry is released only while the nullifier is still armed + // with the SAME anchor + activity. A reservation cleared + // mid-scan and re-armed by a retry (same note, fresh anchor + // and activity) must not be judged against the pre-scan set — + // its anchor may postdate the fetch; it gets checked next pass. + { + let mut store = self.store.write().await; + let still_same = match store.stale_pending_spends(id) { + Ok(current) => current + .iter() + .any(|(n, a, act)| *n == nullifier && *a == anchor && *act == activity_id), + Err(e) => { + tracing::warn!( + wallet_id = %hex::encode(id.wallet_id), + account = id.account_index, + error = %e, + "shielded reservation release: stale_pending_spends failed; skipping entry" + ); + continue; + } + }; + if !still_same { + // Resolved (landed and reconciled, result-wait + // confirmed) or re-armed by a retry while this pass + // was scanning. Nothing to release — leave the + // activity row alone. + tracing::debug!( + wallet_id = %hex::encode(id.wallet_id), + account = id.account_index, + nullifier = %hex::encode(nullifier), + "shielded reservation release: reservation resolved or re-armed; skipping" + ); + continue; + } + match store.clear_pending(id, &nullifier) { + // `still_same` held under this same guard, so the clear + // removes exactly the snapshot's reservation. + Ok(true) => {} + Ok(false) => continue, + Err(e) => { + tracing::warn!( + wallet_id = %hex::encode(id.wallet_id), + account = id.account_index, + error = %e, + "shielded reservation release: clear_pending failed" + ); + continue; + } + } + } + tracing::info!( + wallet_id = %hex::encode(id.wallet_id), + account = id.account_index, + nullifier = %hex::encode(nullifier), + anchor = %hex::encode(anchor), + "shielded reservation released: its recorded anchor was pruned, so the stranded \ + spend can never execute; freeing the notes" + ); + // Flip the linked activity row to Failed. Only queue to the + // wallet's own persister (cloned out — `WalletPersister: Clone`). + if let Some(entry_id) = activity_id { + if !flipped.insert(entry_id) { + continue; + } + let persister = persisters.get(&id.wallet_id).cloned(); + super::operations::record_activity_status_by_id( + &self.store, + persister.as_ref(), + id.wallet_id, + id, + &entry_id, + super::activity::ShieldedActivityStatus::Failed, + ) + .await; + } + } + } + /// Derive best-effort activity entries from each subwallet's /// persisted scan data and add the new ones to `changeset`. /// @@ -1125,4 +1338,91 @@ mod tests { "persisters must survive a failed clear" ); } + + /// Common case: with no anchored reservation in the store, the pre-scan + /// prefetch short-circuits before any network round-trip — it never + /// touches the mock SDK (which has no anchor-query expectation), so this + /// both pins the fast-path skip and proves the pass is wired without + /// panic (sync() skips the release entirely on `None`). + #[tokio::test] + async fn release_stranded_spends_no_op_without_anchored_reservation() { + let dir = temp_dir("release_noop"); + let coordinator = coordinator_with_one_wallet(&dir).await; + + let subwallets = vec![( + SubwalletId::new([0x11; 32], 0), + OrchardKeySet::from_seed(&[0x42u8; 64], dashcore::Network::Testnet, 0) + .expect("derive viewing keys") + .viewing_keys(), + )]; + + // No reservation was armed, so `stale_pending_spends` is empty and + // the prefetch returns None before fetching the recorded anchor set. + let prefetched = coordinator.prefetch_stranded_release(&subwallets).await; + assert!( + prefetched.is_none(), + "nothing armed ⇒ no anchor fetch, no release pass" + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + /// Full release-path coverage through `release_stranded_spends` itself + /// (not just the store predicate): of two armed reservations, the one + /// whose anchor is absent from the recorded set is released while the + /// one whose anchor is still recorded survives untouched. + #[tokio::test] + async fn release_stranded_spends_releases_pruned_and_retains_recorded() { + let dir = temp_dir("release_paths"); + let coordinator = coordinator_with_one_wallet(&dir).await; + let id = SubwalletId::new([0x11; 32], 0); + + let pruned_nf = [0xAAu8; 32]; + let pruned_anchor = [0xABu8; 32]; + let pruned_act = [0xACu8; 32]; + let live_nf = [0xBAu8; 32]; + let live_anchor = [0xBBu8; 32]; + let live_act = [0xBCu8; 32]; + { + let mut store = coordinator.store.write().await; + store.mark_pending(id, &pruned_nf).expect("mark pruned"); + store + .set_pending_spend(id, &pruned_nf, pruned_anchor, pruned_act) + .expect("arm pruned"); + store.mark_pending(id, &live_nf).expect("mark live"); + store + .set_pending_spend(id, &live_nf, live_anchor, live_act) + .expect("arm live"); + } + + // Snapshot as the pre-scan prefetch would have produced it; the + // recorded set holds only the live anchor (the other was "pruned"). + let snapshot = vec![ + (id, (pruned_nf, pruned_anchor, Some(pruned_act))), + (id, (live_nf, live_anchor, Some(live_act))), + ]; + let recorded: HashSet<[u8; 32]> = [live_anchor].into_iter().collect(); + + coordinator + .release_stranded_spends(snapshot, &recorded) + .await; + + let remaining = coordinator + .store + .read() + .await + .stale_pending_spends(id) + .expect("stale_pending_spends"); + assert_eq!( + remaining.len(), + 1, + "pruned-anchor reservation released; recorded-anchor one retained" + ); + assert_eq!( + remaining[0].0, live_nf, + "the still-recorded reservation must survive" + ); + + let _ = std::fs::remove_dir_all(&dir); + } } diff --git a/packages/rs-platform-wallet/src/wallet/shielded/file_store.rs b/packages/rs-platform-wallet/src/wallet/shielded/file_store.rs index 436b0513a57..a540b338b0e 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/file_store.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/file_store.rs @@ -20,7 +20,8 @@ use std::sync::Mutex; use grovedb_commitment_tree::{ClientPersistentCommitmentTree, Position, Retention}; use super::store::{ - ShieldedNote, ShieldedOutgoingNote, ShieldedStore, SubwalletId, SubwalletState, + ShieldedNote, ShieldedOutgoingNote, ShieldedStore, StalePendingSpend, SubwalletId, + SubwalletState, }; use crate::wallet::platform_wallet::WalletId; @@ -182,6 +183,27 @@ impl ShieldedStore for FileBackedShieldedStore { .unwrap_or(false)) } + fn set_pending_spend( + &mut self, + id: SubwalletId, + nullifier: &[u8; 32], + anchor: [u8; 32], + activity_id: [u8; 32], + ) -> Result<(), Self::Error> { + if let Some(sw) = self.subwallets.get_mut(&id) { + sw.set_pending_spend(nullifier, anchor, activity_id); + } + Ok(()) + } + + fn stale_pending_spends(&self, id: SubwalletId) -> Result, Self::Error> { + Ok(self + .subwallets + .get(&id) + .map(SubwalletState::stale_pending_spends) + .unwrap_or_default()) + } + fn record_outgoing_note( &mut self, id: SubwalletId, diff --git a/packages/rs-platform-wallet/src/wallet/shielded/operations.rs b/packages/rs-platform-wallet/src/wallet/shielded/operations.rs index ea212045916..20c8556ddeb 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/operations.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/operations.rs @@ -330,6 +330,33 @@ async fn record_activity_status( queue_shielded_changeset(persister, wallet_id, changeset_for_entry(id, next)); } +/// Flip the activity row identified by `entry_id` (if present) to +/// `status`, reusing [`record_activity_status`]'s semantics — it re-reads +/// the CURRENT stored row and leaves a scan-`Confirmed`-with-height row +/// untouched. Used by the sync reconcile, which knows only the reservation's +/// stored `activity_id`, not the whole entry. No-op if no such row exists. +pub(super) async fn record_activity_status_by_id( + store: &Arc>, + persister: Option<&WalletPersister>, + wallet_id: WalletId, + id: SubwalletId, + entry_id: &[u8; 32], + status: ShieldedActivityStatus, +) { + let entry = match store.read().await.get_activity_by_entry_id(id, entry_id) { + Ok(entry) => entry, + Err(e) => { + warn!( + entry_id = %hex::encode(entry_id), + error = %e, + "activity status flip by id: lookup failed; skipping" + ); + return; + } + }; + record_activity_status(store, persister, wallet_id, id, &entry, status, None).await; +} + // ------------------------------------------------------------------------- // Shield: platform addresses -> shielded pool (Type 15) // ------------------------------------------------------------------------- @@ -664,6 +691,10 @@ pub async fn unshield( let mut pending_entry = None; let result = async { let (spends, anchor) = extract_spends_and_anchor(sdk, store, &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. + let anchor_bytes = anchor.to_bytes(); // The builder computes and returns the fee authoritatively; `exact_fee` (== the // minimum) was already used above for note reservation. @@ -708,6 +739,7 @@ pub async fn unshield( }, ) .await; + arm_pending_release(store, id, anchor_bytes, &pending_entry, &selected_notes).await; trace!("Unshield: state transition built, broadcasting..."); broadcast_shielded_spend(sdk, &state_transition, "unshield").await @@ -826,6 +858,10 @@ pub async fn transfer( let mut pending_entry = None; let result = async { let (spends, anchor) = extract_spends_and_anchor(sdk, store, &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. + let anchor_bytes = anchor.to_bytes(); // The builder computes and returns the fee authoritatively; `exact_fee` (== the // minimum) was already used above for note reservation. @@ -870,6 +906,7 @@ pub async fn transfer( }, ) .await; + arm_pending_release(store, id, anchor_bytes, &pending_entry, &selected_notes).await; trace!("Shielded transfer: state transition built, broadcasting..."); broadcast_shielded_spend(sdk, &state_transition, "transfer").await @@ -976,6 +1013,10 @@ pub async fn withdraw( let mut pending_entry = None; let result = async { let (spends, anchor) = extract_spends_and_anchor(sdk, store, &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. + let anchor_bytes = anchor.to_bytes(); // The builder computes and returns the fee authoritatively; `exact_fee` (== the // minimum) was already used above for note reservation. @@ -1022,6 +1063,7 @@ pub async fn withdraw( }, ) .await; + arm_pending_release(store, id, anchor_bytes, &pending_entry, &selected_notes).await; trace!("Shielded withdrawal: state transition built, broadcasting..."); broadcast_shielded_spend(sdk, &state_transition, "withdraw").await @@ -1165,6 +1207,10 @@ where let mut pending_entry = None; let result = async { let (spends, anchor) = extract_spends_and_anchor(sdk, store, &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. + let anchor_bytes = anchor.to_bytes(); let build = build_identity_create_from_shielded_pool_transition( public_keys, @@ -1211,6 +1257,7 @@ where }, ) .await; + arm_pending_release(store, id, anchor_bytes, &pending_entry, &selected_notes).await; trace!("IdentityCreateFromShieldedPool: built, broadcasting via SDK..."); // Stage the broadcast and the result-wait SEPARATELY (instead of one `broadcast_and_wait`) @@ -1853,6 +1900,42 @@ async fn cancel_pending( } } +/// Record the recorded `anchor` the spend was built against and the +/// linked activity entry on every selected note's reservation, so a +/// spend that ends broadcast-accepted-but-unconfirmed can be released +/// on a later sync once that anchor is pruned from Platform's recorded +/// set (see `NetworkShieldedCoordinator::release_stranded_spends`). +/// +/// No-op when no activity entry was recorded (an output-less bundle — +/// unreachable for our own builders, which always carry a visible +/// output). Best-effort: a store write failure only means the +/// reservation won't self-release on a pruned anchor (it still frees +/// on the next restart), so it must never abort a spend about to +/// broadcast. A success (`finalize_pending`) or a definite failure +/// (`cancel_pending`) removes the entry, so only an ambiguous +/// unconfirmed outcome leaves it carrying the anchor. +async fn arm_pending_release( + store: &Arc>, + id: SubwalletId, + anchor: [u8; 32], + pending_entry: &Option, + notes: &[ShieldedNote], +) { + let Some(entry) = pending_entry else { + return; + }; + let mut store = store.write().await; + for note in notes { + if let Err(e) = store.set_pending_spend(id, ¬e.nullifier, anchor, entry.id) { + warn!( + error = %e, + "set_pending_spend failed; this reservation won't self-release on a pruned \ + anchor (it still frees on the next restart)" + ); + } + } +} + /// Whether an SDK error carries Platform's own consensus verdict on the /// transition. Two shapes qualify: /// @@ -2434,6 +2517,58 @@ mod record_activity_status_tests { assert_eq!(stored.status, ShieldedActivityStatus::Confirmed); assert_eq!(stored.block_height, Some(900)); } + + /// The by-id flip the sync reconcile uses: it knows only the + /// reservation's stored `activity_id`, so it looks the row up and flips + /// it — a released stranded spend moves Pending → Failed. + #[tokio::test] + async fn status_flip_by_id_flips_pending_to_failed() { + let store = Arc::new(RwLock::new(InMemoryShieldedStore::new())); + let id = sub(); + let pending = captured_pending(); + store.write().await.save_activity(id, &pending).unwrap(); + + record_activity_status_by_id( + &store, + None, + id.wallet_id, + id, + &pending.id, + ShieldedActivityStatus::Failed, + ) + .await; + + let stored = store + .read() + .await + .get_activity_by_entry_id(id, &pending.id) + .unwrap() + .expect("row must exist"); + assert_eq!(stored.status, ShieldedActivityStatus::Failed); + } + + /// A by-id flip for an entry that doesn't exist is a silent no-op + /// (nothing to flip), never a panic. + #[tokio::test] + async fn status_flip_by_id_missing_entry_is_noop() { + let store = Arc::new(RwLock::new(InMemoryShieldedStore::new())); + let id = sub(); + record_activity_status_by_id( + &store, + None, + id.wallet_id, + id, + &[0xDE; 32], + ShieldedActivityStatus::Failed, + ) + .await; + assert!(store + .read() + .await + .get_activity_by_entry_id(id, &[0xDE; 32]) + .unwrap() + .is_none()); + } } /// Unit tests for the pure anchor-selection probe ([`select_recorded_spends`]) diff --git a/packages/rs-platform-wallet/src/wallet/shielded/store.rs b/packages/rs-platform-wallet/src/wallet/shielded/store.rs index d6f6d1801cb..f8e2e82fe35 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/store.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/store.rs @@ -116,6 +116,13 @@ pub struct ShieldedOutgoingNote { pub block_height: u64, } +/// A pending reservation that carries a recorded anchor: +/// `(nullifier, anchor, activity_id)`. Yielded by +/// [`ShieldedStore::stale_pending_spends`] for the sync reconcile, +/// which releases the reservation once `anchor` is no longer in +/// Platform's recorded set (the spend can then never execute). +pub type StalePendingSpend = ([u8; 32], [u8; 32], Option<[u8; 32]>); + /// Storage abstraction for shielded wallet state. /// /// Consumers implement this for their persistence layer. The @@ -169,6 +176,29 @@ pub trait ShieldedStore: Send + Sync { fn clear_pending(&mut self, id: SubwalletId, nullifier: &[u8; 32]) -> Result; + /// Attach the recorded `anchor` the spend was built against, and + /// the linked `activity_id`, to `id`'s pending reservation for + /// `nullifier` (taken earlier by [`Self::mark_pending`]). No-op + /// if the nullifier is no longer pending. The sync reconcile uses + /// the stored anchor to detect a stranded spend whose anchor + /// Platform has pruned (see [`Self::stale_pending_spends`]). + fn set_pending_spend( + &mut self, + id: SubwalletId, + nullifier: &[u8; 32], + anchor: [u8; 32], + activity_id: [u8; 32], + ) -> Result<(), Self::Error>; + + /// Pending reservations for `id` that carry a recorded anchor — + /// `(nullifier, anchor, activity_id)` per built-but-unconfirmed + /// spend. Empty when `id` has no anchored reservations (the common + /// case, which lets the sync reconcile skip its network round-trip). + /// The reconcile checks each anchor against Platform's recorded set + /// and releases the reservation via [`Self::clear_pending`] when the + /// anchor is pruned — the spend can then never execute. + fn stale_pending_spends(&self, id: SubwalletId) -> Result, Self::Error>; + // ── Outgoing history (per-subwallet) ─────────────────────────────── /// Record an outgoing (sent) note recovered via OVK for `id`. @@ -350,6 +380,28 @@ pub trait ShieldedStore: Send + Sync { // ── Per-subwallet bookkeeping ────────────────────────────────────────── +/// In-flight-spend bookkeeping for a single reserved nullifier. +/// +/// A reservation starts life bare (both fields `None`) the moment +/// [`SubwalletState::mark_pending`] excludes the note from selection. +/// Once the spend is actually built, [`SubwalletState::set_pending_spend`] +/// records the Platform-**recorded** anchor it was built against and the +/// linked activity-log entry, so the sync reconcile can later detect a +/// stranded (broadcast-accepted-but-never-landed) spend: once that anchor +/// is pruned from Platform's recorded set the spend can never execute, and +/// with the nullifier still unspent the reservation is provably dead and +/// its note can be freed. All in-memory only — a restart drops every +/// reservation regardless. +#[derive(Debug, Clone, Default)] +pub(super) struct PendingSpend { + /// The recorded anchor the spend was built against, once known. + /// `None` while the note is reserved but the spend isn't built yet. + pub anchor: Option<[u8; 32]>, + /// The linked activity-log entry id, so a released reservation can + /// flip its "Pending" row to "Failed" instead of stranding it. + pub activity_id: Option<[u8; 32]>, +} + /// Per-subwallet note + sync state used by both the in-memory and /// file-backed stores. Kept in this module so both share the /// exact same shape and the persister callback can serialize it @@ -364,10 +416,11 @@ pub(super) struct SubwalletState { /// global index to scan (exclusive). `0` = nothing scanned yet. pub last_synced_index: u64, /// Nullifiers of notes currently being spent in an in-flight - /// transition. Excluded from `unspent_notes()` so concurrent - /// callers can't double-select. In-memory only — never - /// persisted; the next sync after a crash reconciles state. - pub pending_nullifiers: BTreeSet<[u8; 32]>, + /// transition, mapped to the [`PendingSpend`] bookkeeping the + /// sync reconcile needs. Excluded from `unspent_notes()` so + /// concurrent callers can't double-select. In-memory only — + /// never persisted; the next sync after a crash reconciles state. + pub pending_nullifiers: BTreeMap<[u8; 32], PendingSpend>, /// Notes this subwallet SENT, recovered via OVK during the scan. /// Append-only send history in recording order. pub outgoing_notes: Vec, @@ -400,7 +453,7 @@ impl SubwalletState { pub(super) fn unspent_notes(&self) -> Vec { self.notes .iter() - .filter(|n| !n.is_spent && !self.pending_nullifiers.contains(&n.nullifier)) + .filter(|n| !n.is_spent && !self.pending_nullifiers.contains_key(&n.nullifier)) .cloned() .collect() } @@ -425,16 +478,60 @@ impl SubwalletState { } /// Reserve `nullifier` against an in-flight spend. Returns - /// `true` if newly added. + /// `true` if newly added, `false` if it was already reserved. + /// Re-reserving is a true no-op: an already-armed entry keeps its + /// anchor/activity link (selection excludes pending nullifiers, so + /// this shouldn't happen — but a plain `insert` would silently + /// wipe the release pass's only handle on a stranded spend if it + /// ever did). pub(super) fn mark_pending(&mut self, nullifier: &[u8; 32]) -> bool { - self.pending_nullifiers.insert(*nullifier) + match self.pending_nullifiers.entry(*nullifier) { + std::collections::btree_map::Entry::Vacant(e) => { + e.insert(PendingSpend::default()); + true + } + std::collections::btree_map::Entry::Occupied(_) => false, + } + } + + /// Attach the built spend's recorded `anchor` and linked + /// `activity_id` to an existing reservation for `nullifier`. + /// No-op if the nullifier is no longer pending (e.g. the + /// reservation was already cleared by a concurrent finalize). + pub(super) fn set_pending_spend( + &mut self, + nullifier: &[u8; 32], + anchor: [u8; 32], + activity_id: [u8; 32], + ) { + if let Some(pending) = self.pending_nullifiers.get_mut(nullifier) { + pending.anchor = Some(anchor); + pending.activity_id = Some(activity_id); + } + } + + /// Reservations that carry a recorded anchor — i.e. built, + /// broadcast-but-unconfirmed spends. Returns `(nullifier, + /// anchor, activity_id)` per such entry; reservations still + /// awaiting a built spend (`anchor: None`) are skipped. The + /// sync reconcile checks each anchor against Platform's recorded + /// set and releases the reservation when the anchor is pruned. + pub(super) fn stale_pending_spends(&self) -> Vec { + self.pending_nullifiers + .iter() + .filter_map(|(nullifier, spend)| { + spend + .anchor + .map(|anchor| (*nullifier, anchor, spend.activity_id)) + }) + .collect() } /// Release a reservation previously taken via `mark_pending`. /// Returns `true` if a matching reservation was actually /// removed. pub(super) fn clear_pending(&mut self, nullifier: &[u8; 32]) -> bool { - self.pending_nullifiers.remove(nullifier) + self.pending_nullifiers.remove(nullifier).is_some() } /// Record an outgoing (sent) note. Idempotent by `cmx`: returns @@ -576,6 +673,27 @@ impl ShieldedStore for InMemoryShieldedStore { .unwrap_or(false)) } + fn set_pending_spend( + &mut self, + id: SubwalletId, + nullifier: &[u8; 32], + anchor: [u8; 32], + activity_id: [u8; 32], + ) -> Result<(), Self::Error> { + if let Some(sw) = self.subwallets.get_mut(&id) { + sw.set_pending_spend(nullifier, anchor, activity_id); + } + Ok(()) + } + + fn stale_pending_spends(&self, id: SubwalletId) -> Result, Self::Error> { + Ok(self + .subwallets + .get(&id) + .map(SubwalletState::stale_pending_spends) + .unwrap_or_default()) + } + fn record_outgoing_note( &mut self, id: SubwalletId, @@ -888,4 +1006,151 @@ mod tests { store.append_commitment(&[3u8; 32], true).unwrap(); assert_eq!(store.tree_size().unwrap(), 1); } + + /// Build a minimal unspent note carrying `nullifier`. + fn note_with_nullifier(nullifier: [u8; 32]) -> ShieldedNote { + ShieldedNote { + position: 0, + cmx: [1u8; 32], + nullifier, + block_height: 1, + is_spent: false, + value: 1_000, + note_data: vec![0u8; 115], + } + } + + /// A reservation carrying a recorded anchor surfaces via + /// `stale_pending_spends`, and the sync reconcile's release decision + /// (release iff the anchor is absent from the recorded set) must RETAIN + /// it while the anchor is still recorded — a slow-but-landing spend is + /// never freed, so its note stays out of the unspent set. + #[test] + fn pending_spend_with_recorded_anchor_is_retained() { + use std::collections::HashSet; + + let mut store = InMemoryShieldedStore::new(); + let id = test_id(0); + let nullifier = [0x11; 32]; + let anchor = [0x22; 32]; + let activity_id = [0x33; 32]; + + store + .save_note(id, ¬e_with_nullifier(nullifier)) + .unwrap(); + assert!( + store.mark_pending(id, &nullifier).unwrap(), + "newly reserved" + ); + store + .set_pending_spend(id, &nullifier, anchor, activity_id) + .unwrap(); + + assert_eq!( + store.stale_pending_spends(id).unwrap(), + vec![(nullifier, anchor, Some(activity_id))], + "an armed reservation surfaces as an anchored (stale-checkable) spend" + ); + + // Anchor IS still recorded → retain. + let recorded: HashSet<[u8; 32]> = [anchor].into_iter().collect(); + for (n, a, _) in store.stale_pending_spends(id).unwrap() { + if !recorded.contains(&a) { + store.clear_pending(id, &n).unwrap(); + } + } + assert_eq!( + store.stale_pending_spends(id).unwrap().len(), + 1, + "a still-recorded anchor must not be released" + ); + assert!( + store.get_unspent_notes(id).unwrap().is_empty(), + "the retained note stays excluded from selection candidates" + ); + } + + /// The fund-safe release: once the anchor is pruned (absent from the + /// recorded set) the spend can never execute, so the reservation is + /// released and the freed note becomes spendable again. + #[test] + fn pending_spend_with_pruned_anchor_is_released() { + use std::collections::HashSet; + + let mut store = InMemoryShieldedStore::new(); + let id = test_id(0); + let nullifier = [0x11; 32]; + let anchor = [0x22; 32]; + let activity_id = [0x33; 32]; + + store + .save_note(id, ¬e_with_nullifier(nullifier)) + .unwrap(); + store.mark_pending(id, &nullifier).unwrap(); + store + .set_pending_spend(id, &nullifier, anchor, activity_id) + .unwrap(); + assert!(store.get_unspent_notes(id).unwrap().is_empty()); + + // Anchor is ABSENT from the recorded set (pruned) → release. + let recorded: HashSet<[u8; 32]> = [[0xEE; 32]].into_iter().collect(); + for (n, a, _) in store.stale_pending_spends(id).unwrap() { + if !recorded.contains(&a) { + assert!(store.clear_pending(id, &n).unwrap()); + } + } + assert!( + store.stale_pending_spends(id).unwrap().is_empty(), + "a pruned-anchor reservation must be released" + ); + let unspent = store.get_unspent_notes(id).unwrap(); + assert_eq!(unspent.len(), 1, "the freed note is spendable again"); + assert_eq!(unspent[0].nullifier, nullifier); + } + + /// A reserved note is excluded from `unspent_notes` (so a concurrent + /// spend can't re-select it) and re-included once the reservation is + /// cleared. + #[test] + fn unspent_notes_excludes_pending_and_reincludes_after_clear() { + let mut store = InMemoryShieldedStore::new(); + let id = test_id(0); + let nullifier = [0x55; 32]; + store + .save_note(id, ¬e_with_nullifier(nullifier)) + .unwrap(); + assert_eq!(store.get_unspent_notes(id).unwrap().len(), 1); + + store.mark_pending(id, &nullifier).unwrap(); + assert!( + store.get_unspent_notes(id).unwrap().is_empty(), + "a reserved note is excluded from selection candidates" + ); + + assert!(store.clear_pending(id, &nullifier).unwrap()); + assert_eq!( + store.get_unspent_notes(id).unwrap().len(), + 1, + "releasing the reservation re-includes the note" + ); + } + + /// `stale_pending_spends` ignores a reservation that was `mark_pending`ed + /// but never armed with an anchor (a just-reserved, not-yet-built spend): + /// the reconcile must never release such a transient entry. + #[test] + fn unarmed_reservation_is_not_stale() { + let mut store = InMemoryShieldedStore::new(); + let id = test_id(0); + let nullifier = [0x66; 32]; + store + .save_note(id, ¬e_with_nullifier(nullifier)) + .unwrap(); + store.mark_pending(id, &nullifier).unwrap(); + + assert!( + store.stale_pending_spends(id).unwrap().is_empty(), + "a reservation with no recorded anchor must not surface as stale" + ); + } }