diff --git a/packages/rs-platform-wallet-ffi/src/shielded_send.rs b/packages/rs-platform-wallet-ffi/src/shielded_send.rs index bd6fe5f2247..54f6edaa711 100644 --- a/packages/rs-platform-wallet-ffi/src/shielded_send.rs +++ b/packages/rs-platform-wallet-ffi/src/shielded_send.rs @@ -311,9 +311,11 @@ pub unsafe extern "C" fn platform_wallet_manager_shielded_transfer( // calling thread. let result = block_on_worker(async move { let prover = CachedOrchardProver::new(); - wallet + let r = wallet .shielded_transfer_to(&coordinator, account, &recipient, amount, memo, &prover) - .await + .await; + poke_sync_on_unconfirmed(&r, handle); + r }); map_spend_result(result, "shielded transfer") } @@ -363,9 +365,11 @@ pub unsafe extern "C" fn platform_wallet_manager_shielded_unshield( let result = block_on_worker(async move { let prover = CachedOrchardProver::new(); - wallet + let r = wallet .shielded_unshield_to(&coordinator, account, &to_addr_str, amount, &prover) - .await + .await; + poke_sync_on_unconfirmed(&r, handle); + r }); map_spend_result(result, "shielded unshield") } @@ -412,7 +416,7 @@ pub unsafe extern "C" fn platform_wallet_manager_shielded_withdraw( let result = block_on_worker(async move { let prover = CachedOrchardProver::new(); - wallet + let r = wallet .shielded_withdraw_to( &coordinator, account, @@ -421,11 +425,61 @@ pub unsafe extern "C" fn platform_wallet_manager_shielded_withdraw( core_fee_per_byte, &prover, ) - .await + .await; + poke_sync_on_unconfirmed(&r, handle); + r }); map_spend_result(result, "shielded withdraw") } +/// On the AMBIGUOUS outcome (broadcast accepted, result unconfirmed), +/// kick an immediate forced shielded sync so the first re-drive check — +/// nullifier re-check, then re-broadcast of the persisted transition — +/// happens now instead of at the next background tick. +/// +/// Routed through the manager's [`ShieldedSyncManager::sync_now`] so the +/// pass respects the same `is_syncing` CAS + `quiescing` drain barrier as +/// the periodic loop and the host's Sync Now button — a raw +/// `coordinator.sync(...)` here would race both. `force = true` bypasses +/// only the caught-up cooldown, never the serialization gate. If a pass +/// is already in flight the poke no-ops (empty summary) and the next +/// tick's pass picks the redrive up — that pass's pre-scan snapshot may +/// predate this arm, which is fine. +/// +/// Fire-and-forget: the spend's own result is already decided and the +/// sync pass owns resolution from here (`redrive_pending_spends` + the +/// prune backstop); the pass outcome is logged, not surfaced. +/// +/// [`ShieldedSyncManager::sync_now`]: platform_wallet::manager::shielded_sync::ShieldedSyncManager::sync_now +fn poke_sync_on_unconfirmed(result: &Result, handle: Handle) { + let ambiguous = matches!( + result, + Err(PlatformWalletError::ShieldedSpendUnconfirmed { .. }) + | Err(PlatformWalletError::ShieldedBroadcastUnconfirmed { .. }) + ); + if !ambiguous { + return; + } + let Some(sync_manager) = + PLATFORM_WALLET_MANAGER_STORAGE.with_item(handle, |manager| manager.shielded_sync_arc()) + else { + return; + }; + runtime().spawn(async move { + let summary = sync_manager.sync_now(true).await; + if summary.sync_unix_seconds == 0 { + tracing::debug!( + "post-unconfirmed shielded sync poke skipped (a pass was already in flight or shielded is unconfigured); the next pass owns the re-drive" + ); + } else { + tracing::debug!( + wallets = summary.wallet_results.len(), + "post-unconfirmed shielded sync pass completed" + ); + } + }); +} + /// Map a shielded operation outcome (shield / unshield / transfer / /// withdraw) to a typed FFI result, mirroring the identity-create sibling's /// code split so hosts can tell "definitively failed, safe to retry" from @@ -600,7 +654,7 @@ pub unsafe extern "C" fn platform_wallet_manager_shielded_identity_create_from_p // `Signer`. let identity_signer: &VTableSigner = &*(signer_identity_addr as *const VTableSigner); let prover = CachedOrchardProver::new(); - wallet + let r = wallet .shielded_identity_create_from_pool( &coordinator, account, @@ -611,7 +665,9 @@ pub unsafe extern "C" fn platform_wallet_manager_shielded_identity_create_from_p identity_signer, &prover, ) - .await + .await; + poke_sync_on_unconfirmed(&r, handle); + r }); match result { diff --git a/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs b/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs index 097efd32d32..29213990fa1 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs @@ -638,13 +638,46 @@ impl NetworkShieldedCoordinator { // `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. + // Residual-spend resolution: `sync_notes_across` above marked every + // landed spend (clearing its reservation and dropping its redrive + // record via the store hook). Two passes over what's left, both + // judged against the PRE-scan recorded-anchor set: + // + // 1. Re-drive — for each armed unconfirmed spend whose anchor is + // still recorded, re-broadcast the stored byte-identical + // transition (bounded by MAX_REDRIVE_ATTEMPTS) to actively + // resolve the ambiguity instead of waiting out the retention + // window. + // 2. Prune backstop — release any still-pending pre-scan + // reservation whose anchor was already pruned (the spend can + // never execute). + // + // Runs before the balance read so freed notes are reflected in + // this pass's balances. if let Some((snapshot, recorded)) = stranded_release { + // Snapshot the per-wallet persisters BEFORE the loop and drop + // the read guard: `redrive_pending_spends` performs network + // broadcasts, and holding the persisters lock across those + // awaits would block wallet register/unregister for the + // duration of the round trips. + let subwallet_persisters: Vec<(SubwalletId, Option)> = { + let persisters = self.persisters.read().await; + subwallets + .iter() + .map(|(id, _)| (*id, persisters.get(&id.wallet_id).cloned())) + .collect() + }; + for (id, persister) in &subwallet_persisters { + super::operations::redrive_pending_spends( + &self.sdk, + &self.store, + persister.as_ref(), + id.wallet_id, + *id, + &recorded, + ) + .await; + } self.release_stranded_spends(snapshot, &recorded).await; } 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 a540b338b0e..1ea30ea9104 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/file_store.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/file_store.rs @@ -20,8 +20,8 @@ use std::sync::Mutex; use grovedb_commitment_tree::{ClientPersistentCommitmentTree, Position, Retention}; use super::store::{ - ShieldedNote, ShieldedOutgoingNote, ShieldedStore, StalePendingSpend, SubwalletId, - SubwalletState, + PendingRedrive, ShieldedNote, ShieldedOutgoingNote, ShieldedStore, StalePendingSpend, + SubwalletId, SubwalletState, }; use crate::wallet::platform_wallet::WalletId; @@ -66,6 +66,13 @@ pub struct FileBackedShieldedStore { /// Per-subwallet notes + sync state, keyed by `(wallet_id, /// account_index)`. Lazily populated on first use of an id. subwallets: BTreeMap, + /// Second connection on the same SQLite file, owning the + /// `shielded_pending_spends` table (armed [`PendingRedrive`] + /// records). Separate from `tree` because the commitment-tree + /// wrapper takes its `Connection` by value; WAL mode makes the + /// two-connection setup safe. `Mutex` for the same `Sync`-shim + /// reason as `tree`. + pending_conn: Mutex, } impl FileBackedShieldedStore { @@ -95,12 +102,94 @@ impl FileBackedShieldedStore { let conn = Self::open_tuned_connection(&path)?; let tree = ClientPersistentCommitmentTree::open(conn, max_checkpoints) .map_err(|e| FileShieldedStoreError(format!("open commitment tree: {e}")))?; - Ok(Self { + let pending_conn = Self::open_tuned_connection(&path)?; + pending_conn + .execute( + "CREATE TABLE IF NOT EXISTS shielded_pending_spends ( + wallet_id BLOB NOT NULL, + account_index INTEGER NOT NULL, + activity_id BLOB NOT NULL, + anchor BLOB NOT NULL, + nullifiers BLOB NOT NULL, + st_bytes BLOB NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (wallet_id, account_index, activity_id) + )", + [], + ) + .map_err(|e| FileShieldedStoreError(format!("create pending_spends table: {e}")))?; + let mut store = Self { tree: Mutex::new(tree), path, max_checkpoints, subwallets: BTreeMap::new(), - }) + pending_conn: Mutex::new(pending_conn), + }; + store.rehydrate_pending_spends()?; + Ok(store) + } + + /// Reload every persisted [`PendingRedrive`] into the in-memory + /// per-subwallet state, re-arming both the redrive record and the + /// note reservations its nullifiers carry — an unconfirmed + /// broadcast therefore keeps its notes reserved (and its re-drive + /// alive) across restarts. Corrupt rows are dropped with a warning + /// rather than failing the open. + fn rehydrate_pending_spends(&mut self) -> Result<(), FileShieldedStoreError> { + let conn = self.pending_conn.lock().expect("pending_conn mutex"); + let mut stmt = conn + .prepare( + "SELECT wallet_id, account_index, activity_id, anchor, nullifiers, st_bytes, \ + attempts FROM shielded_pending_spends", + ) + .map_err(|e| FileShieldedStoreError(format!("prepare rehydrate: {e}")))?; + let rows = stmt + .query_map([], |row| { + Ok(( + row.get::<_, Vec>(0)?, + row.get::<_, u32>(1)?, + row.get::<_, Vec>(2)?, + row.get::<_, Vec>(3)?, + row.get::<_, Vec>(4)?, + row.get::<_, Vec>(5)?, + row.get::<_, u32>(6)?, + )) + }) + .map_err(|e| FileShieldedStoreError(format!("query rehydrate: {e}")))?; + for row in rows { + let (wallet_id, account_index, activity_id, anchor, nullifiers, st_bytes, attempts) = + row.map_err(|e| FileShieldedStoreError(format!("read rehydrate row: {e}")))?; + let (Ok(wallet_id), Ok(activity_id), Ok(anchor)) = ( + <[u8; 32]>::try_from(wallet_id.as_slice()), + <[u8; 32]>::try_from(activity_id.as_slice()), + <[u8; 32]>::try_from(anchor.as_slice()), + ) else { + tracing::warn!("dropping corrupt shielded_pending_spends row (bad key widths)"); + continue; + }; + if nullifiers.is_empty() || nullifiers.len() % 32 != 0 { + tracing::warn!("dropping corrupt shielded_pending_spends row (bad nullifiers)"); + continue; + } + let nullifiers: Vec<[u8; 32]> = nullifiers + .chunks_exact(32) + .map(|c| <[u8; 32]>::try_from(c).expect("chunks_exact(32)")) + .collect(); + let id = SubwalletId::new(wallet_id, account_index); + let sw = self.subwallets.entry(id).or_default(); + for n in &nullifiers { + sw.mark_pending(n); + sw.set_pending_spend(n, anchor, activity_id); + } + sw.arm_redrive(PendingRedrive { + activity_id, + anchor, + nullifiers, + st_bytes, + attempts, + }); + } + Ok(()) } /// Open a `rusqlite::Connection` on `path` with the same WAL / @@ -127,8 +216,55 @@ impl FileBackedShieldedStore { conn.pragma_update(None, k, v) .map_err(|e| FileShieldedStoreError(format!("PRAGMA {k}={v}: {e}")))?; } + // Two writer connections share this file (the commitment tree's and + // `pending_conn`). WAL allows one writer at a time; without a busy + // timeout a write colliding with the other connection's write txn + // fails immediately with SQLITE_BUSY instead of briefly waiting. + conn.busy_timeout(std::time::Duration::from_secs(5)) + .map_err(|e| FileShieldedStoreError(format!("busy_timeout: {e}")))?; Ok(conn) } + + /// Mirror to SQLite the redrive deletions [`SubwalletState`] performs + /// in memory when a nullifier resolves (`mark_spent` / + /// `clear_pending`): delete every persisted row for `id` whose + /// nullifier blob contains `nullifier`. + fn delete_redrive_rows_containing( + &self, + id: SubwalletId, + nullifier: &[u8; 32], + ) -> Result<(), FileShieldedStoreError> { + let conn = self.pending_conn.lock().expect("pending_conn mutex"); + let mut stmt = conn + .prepare( + "SELECT activity_id, nullifiers FROM shielded_pending_spends \ + WHERE wallet_id = ?1 AND account_index = ?2", + ) + .map_err(|e| FileShieldedStoreError(format!("prepare redrive lookup: {e}")))?; + let rows: Vec<(Vec, Vec)> = stmt + .query_map( + rusqlite::params![id.wallet_id.as_slice(), id.account_index], + |row| Ok((row.get::<_, Vec>(0)?, row.get::<_, Vec>(1)?)), + ) + .map_err(|e| FileShieldedStoreError(format!("query redrive lookup: {e}")))? + .collect::>() + .map_err(|e| FileShieldedStoreError(format!("read redrive lookup: {e}")))?; + drop(stmt); + for (activity_id, nullifiers) in rows { + if nullifiers + .chunks_exact(32) + .any(|c| c == nullifier.as_slice()) + { + conn.execute( + "DELETE FROM shielded_pending_spends \ + WHERE wallet_id = ?1 AND account_index = ?2 AND activity_id = ?3", + rusqlite::params![id.wallet_id.as_slice(), id.account_index, activity_id], + ) + .map_err(|e| FileShieldedStoreError(format!("delete redrive row: {e}")))?; + } + } + Ok(()) + } } impl ShieldedStore for FileBackedShieldedStore { @@ -156,11 +292,27 @@ impl ShieldedStore for FileBackedShieldedStore { } fn mark_spent(&mut self, id: SubwalletId, nullifier: &[u8; 32]) -> Result { - Ok(self - .subwallets - .get_mut(&id) - .map(|sw| sw.mark_spent(nullifier)) - .unwrap_or(false)) + let Some(sw) = self.subwallets.get_mut(&id) else { + return Ok(false); + }; + let marked = sw.mark_spent(nullifier); + if marked { + // `SubwalletState::mark_spent` dropped any redrive carrying + // this nullifier from memory; mirror the deletions. The + // in-memory transition already happened, so a SQLite failure + // must not abort the call — log it and keep the trait + // behavior consistent. A surviving stale row rehydrates a + // reservation on the next open, which the reconcile / prune + // passes then clear. + if let Err(e) = self.delete_redrive_rows_containing(id, nullifier) { + tracing::warn!( + error = %e, + "redrive row deletion failed after mark_spent; a stale row may \ + rehydrate on the next open (self-heals via reconcile/prune)" + ); + } + } + Ok(marked) } fn mark_pending(&mut self, id: SubwalletId, nullifier: &[u8; 32]) -> Result { @@ -176,11 +328,21 @@ impl ShieldedStore for FileBackedShieldedStore { id: SubwalletId, nullifier: &[u8; 32], ) -> Result { - Ok(self - .subwallets - .get_mut(&id) - .map(|sw| sw.clear_pending(nullifier)) - .unwrap_or(false)) + let Some(sw) = self.subwallets.get_mut(&id) else { + return Ok(false); + }; + let removed = sw.clear_pending(nullifier); + if removed { + // Same log-don't-abort rationale as `mark_spent` above. + if let Err(e) = self.delete_redrive_rows_containing(id, nullifier) { + tracing::warn!( + error = %e, + "redrive row deletion failed after clear_pending; a stale row may \ + rehydrate on the next open (self-heals via reconcile/prune)" + ); + } + } + Ok(removed) } fn set_pending_spend( @@ -204,6 +366,98 @@ impl ShieldedStore for FileBackedShieldedStore { .unwrap_or_default()) } + fn arm_redrive(&mut self, id: SubwalletId, redrive: PendingRedrive) -> Result<(), Self::Error> { + { + let conn = self.pending_conn.lock().expect("pending_conn mutex"); + let nullifier_blob: Vec = redrive.nullifiers.iter().flatten().copied().collect(); + conn.execute( + "INSERT OR REPLACE INTO shielded_pending_spends \ + (wallet_id, account_index, activity_id, anchor, nullifiers, st_bytes, attempts) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + rusqlite::params![ + id.wallet_id.as_slice(), + id.account_index, + redrive.activity_id.as_slice(), + redrive.anchor.as_slice(), + nullifier_blob, + redrive.st_bytes, + redrive.attempts, + ], + ) + .map_err(|e| FileShieldedStoreError(format!("persist redrive: {e}")))?; + } + self.subwallets.entry(id).or_default().arm_redrive(redrive); + Ok(()) + } + + fn pending_redrives(&self, id: SubwalletId) -> Result, Self::Error> { + Ok(self + .subwallets + .get(&id) + .map(SubwalletState::pending_redrives) + .unwrap_or_default()) + } + + fn bump_redrive_attempts( + &mut self, + id: SubwalletId, + activity_id: &[u8; 32], + ) -> Result { + // Persist FIRST, mutate memory only on success: the reverse order + // would leave the in-memory counter ahead of the durable row on a + // SQLite failure, and a restart would rewind the attempt budget. + let Some(next) = self + .subwallets + .get(&id) + .and_then(|sw| sw.redrive_attempts(activity_id)) + .map(|attempts| attempts + 1) + else { + return Ok(0); + }; + { + let conn = self.pending_conn.lock().expect("pending_conn mutex"); + conn.execute( + "UPDATE shielded_pending_spends SET attempts = ?4 \ + WHERE wallet_id = ?1 AND account_index = ?2 AND activity_id = ?3", + rusqlite::params![ + id.wallet_id.as_slice(), + id.account_index, + activity_id.as_slice(), + next, + ], + ) + .map_err(|e| FileShieldedStoreError(format!("bump redrive attempts: {e}")))?; + } + let attempts = self + .subwallets + .get_mut(&id) + .map(|sw| sw.bump_redrive_attempts(activity_id)) + .unwrap_or(0); + Ok(attempts) + } + + fn clear_redrive( + &mut self, + id: SubwalletId, + activity_id: &[u8; 32], + ) -> Result<(), Self::Error> { + if let Some(sw) = self.subwallets.get_mut(&id) { + sw.clear_redrive(activity_id); + } + let conn = self.pending_conn.lock().expect("pending_conn mutex"); + conn.execute( + "DELETE FROM shielded_pending_spends \ + WHERE wallet_id = ?1 AND account_index = ?2 AND activity_id = ?3", + rusqlite::params![ + id.wallet_id.as_slice(), + id.account_index, + activity_id.as_slice(), + ], + ) + .map_err(|e| FileShieldedStoreError(format!("clear redrive: {e}")))?; + Ok(()) + } + fn record_outgoing_note( &mut self, id: SubwalletId, @@ -417,6 +671,63 @@ mod tests { std::env::temp_dir().join(format!("shielded_tree_test_{tag}_{nanos}.sqlite")) } + /// A [`PendingRedrive`] survives a store reopen — record, attempt + /// counter, AND the note reservations its nullifiers carry — and is + /// deleted (durably) when one of its nullifiers resolves. + #[test] + fn redrive_roundtrip_rehydration_and_resolution() { + let path = temp_tree_path("redrive_roundtrip"); + let id = SubwalletId::new([7u8; 32], 0); + let redrive = PendingRedrive { + activity_id: [1u8; 32], + anchor: [2u8; 32], + nullifiers: vec![[3u8; 32], [4u8; 32]], + st_bytes: vec![0xAB; 96], + attempts: 0, + }; + { + let mut store = FileBackedShieldedStore::open_path(&path, 100).expect("open"); + store.arm_redrive(id, redrive.clone()).expect("arm"); + assert_eq!( + store + .bump_redrive_attempts(id, &redrive.activity_id) + .expect("bump"), + 1 + ); + } + { + // Reopen: record + attempts + reservations all rehydrated. + let store = FileBackedShieldedStore::open_path(&path, 100).expect("reopen"); + let got = store.pending_redrives(id).expect("pending_redrives"); + assert_eq!(got.len(), 1, "record survives reopen"); + assert_eq!(got[0].attempts, 1, "attempt counter persists"); + assert_eq!(got[0].st_bytes, redrive.st_bytes, "transition bytes intact"); + assert_eq!( + store.stale_pending_spends(id).expect("stale").len(), + 2, + "both nullifier reservations rehydrated from the record" + ); + } + { + // Resolving one nullifier (release path) durably drops the row. + let mut store = FileBackedShieldedStore::open_path(&path, 100).expect("reopen 2"); + assert!(store.clear_pending(id, &[3u8; 32]).expect("clear")); + assert!(store.pending_redrives(id).expect("redrives").is_empty()); + } + { + let store = FileBackedShieldedStore::open_path(&path, 100).expect("reopen 3"); + assert!( + store.pending_redrives(id).expect("redrives").is_empty(), + "deletion persisted across reopen" + ); + assert!( + store.stale_pending_spends(id).expect("stale").is_empty(), + "no reservations rehydrate once the record is gone" + ); + } + let _ = std::fs::remove_file(&path); + } + /// Regression test for the "Shielded Merkle witness /// unavailable" spend failure (multi-wallet shared-tree bug). /// diff --git a/packages/rs-platform-wallet/src/wallet/shielded/operations.rs b/packages/rs-platform-wallet/src/wallet/shielded/operations.rs index 20c8556ddeb..9fe357f431e 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/operations.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/operations.rs @@ -27,7 +27,7 @@ use super::keys::OrchardKeySet; use super::note_selection::{ select_notes_for_denomination, select_notes_with_fee, ShieldedFeeKind, }; -use super::store::{ShieldedNote, ShieldedStore, SubwalletId}; +use super::store::{PendingRedrive, ShieldedNote, ShieldedStore, SubwalletId}; use crate::changeset::{PlatformWalletChangeSet, ShieldedChangeSet}; use crate::error::PlatformWalletError; use crate::wallet::persister::WalletPersister; @@ -61,7 +61,7 @@ use dpp::state_transition::StateTransition; use dpp::withdrawal::Pooling; use grovedb_commitment_tree::{Anchor, PaymentAddress}; use tokio::sync::RwLock; -use tracing::{info, trace, warn}; +use tracing::{debug, info, trace, warn}; /// Number of Orchard actions in a `Shield` (Type 15) bundle. /// @@ -742,7 +742,17 @@ pub async fn unshield( 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 + broadcast_shielded_spend_with_redrive( + sdk, + store, + id, + &pending_entry, + anchor_bytes, + &selected_notes, + &state_transition, + "unshield", + ) + .await } .await; @@ -909,7 +919,17 @@ pub async fn transfer( 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 + broadcast_shielded_spend_with_redrive( + sdk, + store, + id, + &pending_entry, + anchor_bytes, + &selected_notes, + &state_transition, + "transfer", + ) + .await } .await; @@ -1066,7 +1086,17 @@ pub async fn withdraw( 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 + broadcast_shielded_spend_with_redrive( + sdk, + store, + id, + &pending_entry, + anchor_bytes, + &selected_notes, + &state_transition, + "withdraw", + ) + .await } .await; @@ -1355,6 +1385,21 @@ where )); } None => { + // Arm the persisted re-drive before surfacing the + // ambiguity: the sync-time pass re-checks the + // nullifiers, then re-broadcasts this + // byte-identical transition up to + // MAX_REDRIVE_ATTEMPTS times. + arm_redrive_record( + store, + id, + &pending_entry, + anchor_bytes, + &selected_notes, + &st, + "identity_create", + ) + .await; return Err(PlatformWalletError::ShieldedBroadcastUnconfirmed { identity_id, reason: wait_err.to_string(), @@ -1936,6 +1981,297 @@ async fn arm_pending_release( } } +/// Maximum sync-time re-broadcast attempts for a +/// broadcast-accepted-but-unconfirmed spend before the re-drive stops +/// and the anchor-prune release backstop owns the reservation. +pub(super) const MAX_REDRIVE_ATTEMPTS: u32 = 3; + +/// Broadcast a built shielded spend and, on the AMBIGUOUS outcome only +/// (`ShieldedSpendUnconfirmed` — accepted broadcast, failed result +/// wait), persist a [`PendingRedrive`] so the sync-time re-drive can +/// resolve the ambiguity actively: the next scan detects a landing via +/// the nullifiers; otherwise the byte-identical transition is +/// re-broadcast up to [`MAX_REDRIVE_ATTEMPTS`] times (fund-safe — +/// identical nullifiers cannot double-spend); only if every attempt +/// stays silent does the anchor-prune release backstop take over. +#[allow(clippy::too_many_arguments)] +async fn broadcast_shielded_spend_with_redrive( + sdk: &Arc, + store: &Arc>, + id: SubwalletId, + pending_entry: &Option, + anchor: [u8; 32], + notes: &[ShieldedNote], + state_transition: &StateTransition, + operation: &'static str, +) -> Result<(), PlatformWalletError> { + let result = broadcast_shielded_spend(sdk, state_transition, operation).await; + if matches!( + &result, + Err(PlatformWalletError::ShieldedSpendUnconfirmed { .. }) + ) { + arm_redrive_record( + store, + id, + pending_entry, + anchor, + notes, + state_transition, + operation, + ) + .await; + } + result +} + +/// Persist the re-drivable record for an ambiguous spend. Best-effort: +/// a failure here only demotes the resolution path to the anchor-prune +/// backstop (plus restart-loss of the reservation), never fails the +/// spend call itself — the ambiguity already happened. +async fn arm_redrive_record( + store: &Arc>, + id: SubwalletId, + pending_entry: &Option, + anchor: [u8; 32], + notes: &[ShieldedNote], + state_transition: &StateTransition, + operation: &'static str, +) { + use dpp::serialization::PlatformSerializable; + + let Some(entry) = pending_entry else { + return; + }; + let st_bytes = match state_transition.serialize_to_bytes() { + Ok(b) => b, + Err(e) => { + warn!( + operation, + error = %e, + "failed to serialize the unconfirmed transition; re-drive disabled for this \ + spend (prune backstop still applies)" + ); + return; + } + }; + let redrive = PendingRedrive { + activity_id: entry.id, + anchor, + nullifiers: notes.iter().map(|n| n.nullifier).collect(), + st_bytes, + attempts: 0, + }; + if let Err(e) = store.write().await.arm_redrive(id, redrive) { + warn!( + operation, + error = %e, + "failed to persist the redrive record; re-drive disabled for this spend (prune \ + backstop still applies)" + ); + } +} + +/// Whether an SDK error is Platform's `NullifierAlreadySpentError` — +/// on a RE-broadcast of our own byte-identical transition this means +/// the ORIGINAL broadcast executed: the nullifiers are consumed by the +/// very spend being re-driven, so it is a success signal (the next scan +/// confirms the notes spent), never a failure. +fn is_nullifier_already_spent(e: &dash_sdk::Error) -> bool { + use dpp::consensus::state::state_error::StateError; + use dpp::consensus::ConsensusError; + + let consensus: Option<&ConsensusError> = match e { + dash_sdk::Error::Protocol(dpp::ProtocolError::ConsensusError(c)) => Some(c), + dash_sdk::Error::StateTransitionBroadcastError(b) => b.cause.as_ref(), + _ => None, + }; + matches!( + consensus, + Some(ConsensusError::StateError( + StateError::NullifierAlreadySpentError(_) + )) + ) +} + +/// Pure outcome classification for one re-broadcast attempt. Extracted +/// from the redrive loop so the arm ORDER — `AlreadyExecuted` must win +/// over the generic consensus-rejection check, since +/// `NullifierAlreadySpent` is itself a consensus error — is pinned by +/// unit tests without needing a broadcast-mockable network seam. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum RedriveBroadcastOutcome { + /// Relay accepted the re-broadcast; the next scan detects a landing. + Accepted, + /// `NullifierAlreadySpent`: the ORIGINAL broadcast executed — a + /// success signal, never a failure. + AlreadyExecuted, + /// Any other consensus verdict: the transition can never execute. + DefinitiveRejection, + /// Transport noise / `AlreadyExists` / anything non-definitive. + Inconclusive, +} + +fn classify_redrive_broadcast(result: &Result<(), dash_sdk::Error>) -> RedriveBroadcastOutcome { + match result { + Ok(()) => RedriveBroadcastOutcome::Accepted, + Err(e) if is_nullifier_already_spent(e) => RedriveBroadcastOutcome::AlreadyExecuted, + Err(e) if carries_consensus_rejection(e) => RedriveBroadcastOutcome::DefinitiveRejection, + Err(_) => RedriveBroadcastOutcome::Inconclusive, + } +} + +/// Bump a redrive attempt counter, logging (rather than discarding) a +/// persistence failure. On `Err` the durable counter did not advance and +/// the file store's persist-first ordering leaves memory untouched, so +/// the same attempt slot is retried on the next pass — the log line is +/// what makes that visible. +async fn bump_redrive_attempts_logged( + store: &Arc>, + id: SubwalletId, + activity_id: &[u8; 32], +) -> u32 { + match store.write().await.bump_redrive_attempts(id, activity_id) { + Ok(attempts) => attempts, + Err(e) => { + warn!( + error = %e, + "redrive: failed to persist the attempt counter; the attempt will be \ + retried on the next pass" + ); + 0 + } + } +} + +/// Sync-time re-drive for `id`'s armed unconfirmed spends: for each +/// [`PendingRedrive`] whose anchor is still in Platform's `recorded` +/// set and whose attempt budget remains, re-broadcast the stored +/// byte-identical transition (relay-ACK only — the landing itself is +/// detected by the NEXT scan's nullifier reconcile) and classify: +/// +/// - accepted / inconclusive → count the attempt; wait for the next scan; +/// - `NullifierAlreadySpent` → the original executed; the next scan +/// confirms — touch nothing; +/// - any other consensus verdict → provably dead NOW: release the +/// reservation and flip the activity row to Failed, hours before the +/// prune backstop would; +/// - pruned anchor / exhausted attempts → leave it to the +/// prune-backstop release pass. +/// +/// Runs after the scan's spent-note reconcile (a landed spend's record +/// was already dropped by the `mark_spent` hook, so anything still +/// armed here is genuinely unresolved). +pub(super) async fn redrive_pending_spends( + sdk: &Arc, + store: &Arc>, + persister: Option<&WalletPersister>, + wallet_id: WalletId, + id: SubwalletId, + recorded: &std::collections::HashSet<[u8; 32]>, +) { + use dpp::serialization::PlatformDeserializable; + + let redrives = match store.read().await.pending_redrives(id) { + Ok(r) => r, + Err(e) => { + warn!( + error = %e, + "redrive: pending_redrives failed; skipping subwallet" + ); + return; + } + }; + for redrive in redrives { + // A pruned anchor is the release pass's call, not ours; an + // exhausted budget means we've said our three pieces. + if !recorded.contains(&redrive.anchor) || redrive.attempts >= MAX_REDRIVE_ATTEMPTS { + continue; + } + let st = match StateTransition::deserialize_from_bytes(&redrive.st_bytes) { + Ok(st) => st, + Err(e) => { + warn!( + error = %e, + "redrive: stored transition failed to deserialize; dropping the record \ + (the prune backstop still frees the notes)" + ); + if let Err(e) = store.write().await.clear_redrive(id, &redrive.activity_id) { + warn!(error = %e, "redrive: clear_redrive failed"); + } + continue; + } + }; + let broadcast_result = st.broadcast(sdk, None).await; + let err_display = broadcast_result + .as_ref() + .err() + .map(|e| e.to_string()) + .unwrap_or_default(); + match classify_redrive_broadcast(&broadcast_result) { + RedriveBroadcastOutcome::Accepted => { + let attempts = bump_redrive_attempts_logged(store, id, &redrive.activity_id).await; + info!( + attempts, + max = MAX_REDRIVE_ATTEMPTS, + "redrive: re-broadcast accepted; the next scan detects the landing" + ); + } + RedriveBroadcastOutcome::AlreadyExecuted => { + // Success signal — but still consume an attempt: the scan + // normally confirms the landing and drops the record, and + // if it lags, this arm must not re-broadcast unboundedly + // on every pass. The cap parks the record for the scan / + // prune passes to settle. + let attempts = bump_redrive_attempts_logged(store, id, &redrive.activity_id).await; + info!( + attempts, + max = MAX_REDRIVE_ATTEMPTS, + "redrive: transition already executed on-chain; the next scan confirms \ + the notes spent" + ); + } + RedriveBroadcastOutcome::DefinitiveRejection => { + warn!( + error = %err_display, + "redrive: definitive consensus rejection; the spend can never execute — \ + releasing the reservation" + ); + { + let mut guard = store.write().await; + for n in &redrive.nullifiers { + // Also drops the redrive record via the + // clear_pending hook. + if let Err(e) = guard.clear_pending(id, n) { + warn!(error = %e, "redrive: clear_pending failed"); + } + } + } + record_activity_status_by_id( + store, + persister, + wallet_id, + id, + &redrive.activity_id, + ShieldedActivityStatus::Failed, + ) + .await; + } + RedriveBroadcastOutcome::Inconclusive => { + // `AlreadyExists` (still in a mempool after a lost-ACK + // retry) or transport noise: inconclusive; counts toward + // the cap. + let attempts = bump_redrive_attempts_logged(store, id, &redrive.activity_id).await; + debug!( + attempts, + max = MAX_REDRIVE_ATTEMPTS, + error = %err_display, + "redrive: re-broadcast inconclusive" + ); + } + } + } +} + /// Whether an SDK error carries Platform's own consensus verdict on the /// transition. Two shapes qualify: /// @@ -2175,6 +2511,146 @@ fn deserialize_note(data: &[u8]) -> Option { Note::from_parts(recipient, value, rho, rseed).into_option() } +#[cfg(test)] +mod redrive_tests { + use super::*; + use crate::wallet::shielded::store::InMemoryShieldedStore; + use dash_sdk::error::StateTransitionBroadcastError; + use dpp::consensus::state::shielded::nullifier_already_spent_error::NullifierAlreadySpentError; + use dpp::consensus::state::state_error::StateError; + use dpp::consensus::ConsensusError; + + /// On a re-broadcast of our own byte-identical transition, + /// `NullifierAlreadySpent` means the ORIGINAL executed — the + /// classification must treat it as a success signal, distinct from + /// every other consensus verdict. + #[test] + fn nullifier_already_spent_is_a_success_signal() { + let cause = ConsensusError::StateError(StateError::NullifierAlreadySpentError( + NullifierAlreadySpentError::new([1u8; 32]), + )); + let err = dash_sdk::Error::StateTransitionBroadcastError(StateTransitionBroadcastError { + code: 1, + message: "state error".to_string(), + cause: Some(cause), + }); + assert!(is_nullifier_already_spent(&err)); + // ...and it still counts as a consensus-rejection shape, so arm + // ordering matters: the already-spent check must run first. + assert!(carries_consensus_rejection(&err)); + + let other = dash_sdk::Error::TimeoutReached( + std::time::Duration::from_secs(1), + "waiting".to_string(), + ); + assert!(!is_nullifier_already_spent(&other)); + } + + /// The re-broadcast outcome classifier, arm order included: + /// `NullifierAlreadySpent` is itself a consensus error, so the + /// `AlreadyExecuted` arm must win over `DefinitiveRejection`. + #[test] + fn redrive_broadcast_classification_matrix() { + assert_eq!( + classify_redrive_broadcast(&Ok(())), + RedriveBroadcastOutcome::Accepted + ); + + let already_spent = ConsensusError::StateError(StateError::NullifierAlreadySpentError( + NullifierAlreadySpentError::new([1u8; 32]), + )); + let already_spent_err = + dash_sdk::Error::StateTransitionBroadcastError(StateTransitionBroadcastError { + code: 1, + message: "state error".to_string(), + cause: Some(already_spent), + }); + assert_eq!( + classify_redrive_broadcast(&Err(already_spent_err)), + RedriveBroadcastOutcome::AlreadyExecuted, + "already-spent must classify as success BEFORE the generic rejection arm" + ); + + let other_rejection = ConsensusError::BasicError( + dpp::consensus::basic::BasicError::ProtocolVersionParsingError( + dpp::consensus::basic::decode::ProtocolVersionParsingError::new( + "bad version".to_string(), + ), + ), + ); + let rejection_err = + dash_sdk::Error::StateTransitionBroadcastError(StateTransitionBroadcastError { + code: 1, + message: "state error".to_string(), + cause: Some(other_rejection), + }); + assert_eq!( + classify_redrive_broadcast(&Err(rejection_err)), + RedriveBroadcastOutcome::DefinitiveRejection + ); + + let timeout = dash_sdk::Error::TimeoutReached( + std::time::Duration::from_secs(1), + "waiting".to_string(), + ); + assert_eq!( + classify_redrive_broadcast(&Err(timeout)), + RedriveBroadcastOutcome::Inconclusive + ); + } + + /// Decision paths that must NOT touch the network (the mock SDK has + /// no broadcast expectation, so any attempt would error into the + /// inconclusive arm and bump the counter): a pruned anchor belongs + /// to the release backstop, an exhausted budget stays parked, and a + /// corrupt stored transition is dropped so it can't wedge the pass. + #[tokio::test] + async fn redrive_skips_pruned_and_exhausted_and_drops_garbage() { + let sdk = Arc::new(dash_sdk::SdkBuilder::new_mock().build().expect("mock sdk")); + let store = Arc::new(RwLock::new(InMemoryShieldedStore::new())); + let wallet_id = [5u8; 32]; + let id = SubwalletId::new(wallet_id, 0); + + let rec = |activity: u8, anchor: u8, attempts: u32| PendingRedrive { + activity_id: [activity; 32], + anchor: [anchor; 32], + nullifiers: vec![[activity ^ 0xFF; 32]], + st_bytes: vec![0xDE, 0xAD], // never deserializes + attempts, + }; + { + let mut guard = store.write().await; + // Pruned anchor (10 not in recorded set) → untouched. + guard.arm_redrive(id, rec(1, 10, 0)).unwrap(); + // Recorded anchor but attempts exhausted → untouched. + guard + .arm_redrive(id, rec(2, 11, MAX_REDRIVE_ATTEMPTS)) + .unwrap(); + // Recorded anchor, budget left, garbage bytes → dropped + // before any broadcast. + guard.arm_redrive(id, rec(3, 12, 0)).unwrap(); + } + let recorded: std::collections::HashSet<[u8; 32]> = + [[11u8; 32], [12u8; 32]].into_iter().collect(); + + redrive_pending_spends(&sdk, &store, None, wallet_id, id, &recorded).await; + + let left = store.read().await.pending_redrives(id).unwrap(); + let ids: Vec<[u8; 32]> = left.iter().map(|r| r.activity_id).collect(); + assert!(ids.contains(&[1u8; 32]), "pruned-anchor record left alone"); + assert!(ids.contains(&[2u8; 32]), "exhausted record left alone"); + assert!( + !ids.contains(&[3u8; 32]), + "corrupt record dropped without a broadcast attempt" + ); + assert_eq!( + left.iter().map(|r| r.attempts).max(), + Some(MAX_REDRIVE_ATTEMPTS), + "no attempt counters were bumped — nothing touched the network" + ); + } +} + #[cfg(test)] mod classify_spend_wait_failure_tests { use super::*; diff --git a/packages/rs-platform-wallet/src/wallet/shielded/store.rs b/packages/rs-platform-wallet/src/wallet/shielded/store.rs index f8e2e82fe35..ac47ddc423f 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/store.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/store.rs @@ -123,6 +123,35 @@ pub struct ShieldedOutgoingNote { /// Platform's recorded set (the spend can then never execute). pub type StalePendingSpend = ([u8; 32], [u8; 32], Option<[u8; 32]>); +/// A re-drivable broadcast-accepted-but-unconfirmed spend: the signed +/// transition bytes plus everything the sync-time re-drive needs to +/// resolve the ambiguity actively — re-broadcast the transition +/// ([`nullifiers`](Self::nullifiers) detect a landing, `anchor` feeds +/// the prune backstop, `activity_id` links the UI row, `attempts` +/// bounds the retries. +/// +/// Armed only on the ambiguous outcome (`ShieldedSpendUnconfirmed`): +/// the broadcast was accepted but the result wait failed, so the spend +/// may or may not have executed. Re-broadcasting the byte-identical +/// transition is fund-safe — identical nullifiers cannot double-spend — +/// and converts silence into either a confirmation (next scan sees the +/// nullifiers spent) or a definitive consensus verdict. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PendingRedrive { + /// Activity-entry id of the spend (sha256 of visible output cmxs); + /// the spend-level key — every nullifier of one spend shares it. + pub activity_id: [u8; 32], + /// Platform-recorded anchor the spend was built against. + pub anchor: [u8; 32], + /// Nullifiers of every note the spend consumes. + pub nullifiers: Vec<[u8; 32]>, + /// The signed state transition, platform-serialized byte-exact as + /// originally broadcast. + pub st_bytes: Vec, + /// Re-broadcast attempts made so far. + pub attempts: u32, +} + /// Storage abstraction for shielded wallet state. /// /// Consumers implement this for their persistence layer. The @@ -199,6 +228,39 @@ pub trait ShieldedStore: Send + Sync { /// anchor is pruned — the spend can then never execute. fn stale_pending_spends(&self, id: SubwalletId) -> Result, Self::Error>; + // ── Re-drivable unconfirmed spends (per-subwallet) ───────────────── + + /// Persist a re-drivable record for a broadcast-accepted spend whose + /// result wait failed ambiguously. Keyed by `redrive.activity_id`; + /// re-arming the same id overwrites. Unlike the bare `mark_pending` + /// reservations, redrive records survive a restart where the backend + /// persists them (the file store does): on reopen both the record + /// and its note reservations are rehydrated, so the re-drive — and + /// the linked activity row's eventual Confirmed/Failed flip — + /// continue across relaunches. + fn arm_redrive(&mut self, id: SubwalletId, redrive: PendingRedrive) -> Result<(), Self::Error>; + + /// Every armed redrive record for `id`. + fn pending_redrives(&self, id: SubwalletId) -> Result, Self::Error>; + + /// Increment the attempt counter on `id`'s redrive keyed by + /// `activity_id`, returning the new count (`0` when no such record + /// exists). + fn bump_redrive_attempts( + &mut self, + id: SubwalletId, + activity_id: &[u8; 32], + ) -> Result; + + /// Drop the redrive record keyed by `activity_id` — the spend + /// resolved (landed, definitively rejected, or released by the + /// anchor-prune backstop). Implementations also drop the record + /// implicitly when [`Self::mark_spent`] or [`Self::clear_pending`] + /// resolves one of its nullifiers, since a transition lands or dies + /// atomically for all of its nullifiers. + fn clear_redrive(&mut self, id: SubwalletId, activity_id: &[u8; 32]) + -> Result<(), Self::Error>; + // ── Outgoing history (per-subwallet) ─────────────────────────────── /// Record an outgoing (sent) note recovered via OVK for `id`. @@ -418,9 +480,19 @@ pub(super) struct SubwalletState { /// Nullifiers of notes currently being spent in an in-flight /// 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. + /// concurrent callers can't double-select. Held in memory; a + /// pre-broadcast reservation dies with the process, but a + /// reservation belonging to an armed [`PendingRedrive`] is + /// rehydrated on file-store open (the redrive row carries its + /// nullifiers), so an unconfirmed broadcast keeps its notes + /// reserved across restarts. pub pending_nullifiers: BTreeMap<[u8; 32], PendingSpend>, + /// Armed re-drivable unconfirmed spends, keyed by activity id. + /// See [`PendingRedrive`]. Kept consistent with + /// `pending_nullifiers`: resolving any of a redrive's nullifiers + /// (mark_spent / clear_pending) drops the whole record — a + /// transition lands or dies atomically for all its nullifiers. + pub redrives: BTreeMap<[u8; 32], PendingRedrive>, /// Notes this subwallet SENT, recovered via OVK during the scan. /// Append-only send history in recording order. pub outgoing_notes: Vec, @@ -471,12 +543,32 @@ impl SubwalletState { // common path already cleared pending in the // spend-flow finalizer. self.pending_nullifiers.remove(nullifier); + // The spend landed — its redrive record (if armed) is + // resolved for every nullifier it carries. + self.drop_redrives_containing(nullifier); return true; } } false } + /// Drop every redrive record that carries `nullifier`, returning + /// the dropped activity ids (the file store mirrors the deletions + /// to SQLite). A transition lands or dies atomically for all of + /// its nullifiers, so resolving one resolves the record. + pub(super) fn drop_redrives_containing(&mut self, nullifier: &[u8; 32]) -> Vec<[u8; 32]> { + let dropped: Vec<[u8; 32]> = self + .redrives + .iter() + .filter(|(_, r)| r.nullifiers.contains(nullifier)) + .map(|(id, _)| *id) + .collect(); + for id in &dropped { + self.redrives.remove(id); + } + dropped + } + /// Reserve `nullifier` against an in-flight spend. Returns /// `true` if newly added, `false` if it was already reserved. /// Re-reserving is a true no-op: an already-armed entry keeps its @@ -531,7 +623,43 @@ impl SubwalletState { /// 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).is_some() + let removed = self.pending_nullifiers.remove(nullifier).is_some(); + if removed { + // Releasing a reservation resolves its spend for good + // (definitive rejection or prune backstop) — the redrive + // record goes with it. + self.drop_redrives_containing(nullifier); + } + removed + } + + /// Arm (or overwrite by activity id) a re-drivable record. + pub(super) fn arm_redrive(&mut self, redrive: PendingRedrive) { + self.redrives.insert(redrive.activity_id, redrive); + } + + pub(super) fn pending_redrives(&self) -> Vec { + self.redrives.values().cloned().collect() + } + + /// Current attempt count for `activity_id`'s redrive, if armed. + pub(super) fn redrive_attempts(&self, activity_id: &[u8; 32]) -> Option { + self.redrives.get(activity_id).map(|r| r.attempts) + } + + /// Bump the attempt counter; `0` when no such record exists. + pub(super) fn bump_redrive_attempts(&mut self, activity_id: &[u8; 32]) -> u32 { + self.redrives + .get_mut(activity_id) + .map(|r| { + r.attempts += 1; + r.attempts + }) + .unwrap_or(0) + } + + pub(super) fn clear_redrive(&mut self, activity_id: &[u8; 32]) { + self.redrives.remove(activity_id); } /// Record an outgoing (sent) note. Idempotent by `cmx`: returns @@ -694,6 +822,42 @@ impl ShieldedStore for InMemoryShieldedStore { .unwrap_or_default()) } + fn arm_redrive(&mut self, id: SubwalletId, redrive: PendingRedrive) -> Result<(), Self::Error> { + self.subwallets.entry(id).or_default().arm_redrive(redrive); + Ok(()) + } + + fn pending_redrives(&self, id: SubwalletId) -> Result, Self::Error> { + Ok(self + .subwallets + .get(&id) + .map(SubwalletState::pending_redrives) + .unwrap_or_default()) + } + + fn bump_redrive_attempts( + &mut self, + id: SubwalletId, + activity_id: &[u8; 32], + ) -> Result { + Ok(self + .subwallets + .get_mut(&id) + .map(|sw| sw.bump_redrive_attempts(activity_id)) + .unwrap_or(0)) + } + + fn clear_redrive( + &mut self, + id: SubwalletId, + activity_id: &[u8; 32], + ) -> Result<(), Self::Error> { + if let Some(sw) = self.subwallets.get_mut(&id) { + sw.clear_redrive(activity_id); + } + Ok(()) + } + fn record_outgoing_note( &mut self, id: SubwalletId, @@ -885,6 +1049,53 @@ mod tests { assert!(!store.mark_spent(id, &nullifier).unwrap()); } + /// Resolving any of a redrive's nullifiers — landing (`mark_spent`) + /// or release (`clear_pending`) — drops the whole record: a + /// transition lands or dies atomically for all its nullifiers. + #[test] + fn resolving_a_nullifier_drops_the_redrive_record() { + let mut store = InMemoryShieldedStore::new(); + let id = test_id(0); + let n1 = [3u8; 32]; + let n2 = [4u8; 32]; + let note = ShieldedNote { + position: 0, + cmx: [1u8; 32], + nullifier: n1, + block_height: 50, + is_spent: false, + value: 500, + note_data: vec![0u8; 115], + }; + store.save_note(id, ¬e).unwrap(); + let redrive = PendingRedrive { + activity_id: [9u8; 32], + anchor: [8u8; 32], + nullifiers: vec![n1, n2], + st_bytes: vec![1, 2, 3], + attempts: 0, + }; + + // Landing path: mark_spent on one nullifier drops the record. + store.mark_pending(id, &n1).unwrap(); + store.arm_redrive(id, redrive.clone()).unwrap(); + assert_eq!(store.pending_redrives(id).unwrap().len(), 1); + assert!(store.mark_spent(id, &n1).unwrap()); + assert!( + store.pending_redrives(id).unwrap().is_empty(), + "landed spend drops its redrive record" + ); + + // Release path: clear_pending on a nullifier drops the record. + store.mark_pending(id, &n2).unwrap(); + store.arm_redrive(id, redrive).unwrap(); + assert!(store.clear_pending(id, &n2).unwrap()); + assert!( + store.pending_redrives(id).unwrap().is_empty(), + "released reservation drops its redrive record" + ); + } + #[test] fn test_sync_state_per_subwallet() { let mut store = InMemoryShieldedStore::new();