diff --git a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/errors/DashSdkError.kt b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/errors/DashSdkError.kt index 3227c3d606b..e1120e16765 100644 --- a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/errors/DashSdkError.kt +++ b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/errors/DashSdkError.kt @@ -240,6 +240,17 @@ sealed class DashSdkError( class AssetLockTooManyInputs(message: String, cause: Throwable? = null) : PlatformWallet(message, cause) + /** + * `ErrorShieldedTreeInconsistent` (native code 67). The local shielded + * commitment tree holds more leaves than a fresh sync proves. Not + * retryable: resyncing cannot repair it. Call + * [rebuildShieldedTree][org.dashfoundation.dashsdk.wallet.PlatformWalletManager.rebuildShieldedTree], + * which keeps notes and spent status. Also delivered per wallet on + * [org.dashfoundation.dashsdk.wallet.WalletSyncEvent.ShieldedResult.error]. + */ + class ShieldedTreeInconsistent(message: String, cause: Throwable? = null) : + PlatformWallet(message, cause) + /** * `ErrorShieldedNoRecordedAnchor` (native code 19). A shielded spend * could not be built against a Platform-recorded anchor because the @@ -672,6 +683,13 @@ sealed class DashSdkError( */ const val PLATFORM_WALLET_NOT_FOUND_CODE = 98 + /** + * Map a de-offset `PlatformWalletFFIResultCode` delivered outside an + * exception (a sync-completed callback) into the public hierarchy. + */ + fun fromPlatformWalletCode(code: Int, message: String): DashSdkError = + fromPlatformWalletNative(code, message, null) + /** Map a native error code + message into the public hierarchy. */ fun fromNative(e: DashSDKException): DashSdkError { val message = e.message ?: "Unknown SDK error" @@ -768,6 +786,7 @@ sealed class DashSdkError( 27 -> PlatformWallet.ShutdownIncomplete(message, cause) // ErrorShutdownIncomplete 29 -> PlatformWallet.AssetLockInsufficientFunds(message, cause) // ErrorAssetLockInsufficientFunds 60 -> PlatformWallet.AssetLockTooManyInputs(message, cause) // ErrorAssetLockTooManyInputs + 67 -> PlatformWallet.ShieldedTreeInconsistent(message, cause) // ErrorShieldedTreeInconsistent // The deferred-token trio sits at the contiguous block 34-36 because // 27-33 are claimed elsewhere: 27 ErrorShutdownIncomplete // (dashpay/platform#4268, merged), 29 ErrorAssetLockInsufficientFunds diff --git a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/NativeWalletEventBridge.kt b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/NativeWalletEventBridge.kt index 4d7a69993e1..afae383e190 100644 --- a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/NativeWalletEventBridge.kt +++ b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/NativeWalletEventBridge.kt @@ -98,7 +98,7 @@ abstract class NativeWalletEventBridge { /** * `on_shielded_sync_completed_fn`, once per wallet result — descriptor - * `([BZZZIJIJLjava/lang/String;)V`. + * `([BZZZIJIJLjava/lang/String;I)V`. * * @param walletId 32-byte wallet id. * @param success sync succeeded; numeric fields are meaningful. @@ -111,6 +111,8 @@ abstract class NativeWalletEventBridge { * @param newlySpent notes newly detected as spent this pass. * @param balance current unspent shielded balance after the pass. * @param errorMessage failure message, or null on success / skipped. + * @param errorCode de-offset `PlatformWalletFFIResultCode` classifying the + * failure; 0 on success / skipped. */ open fun onShieldedSyncCompleted( walletId: ByteArray, @@ -122,6 +124,7 @@ abstract class NativeWalletEventBridge { newlySpent: Int, balance: Long, errorMessage: String?, + errorCode: Int, ) { } diff --git a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/WalletManagerNative.kt b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/WalletManagerNative.kt index 58351c59fbb..e4eaa20b610 100644 --- a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/WalletManagerNative.kt +++ b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/WalletManagerNative.kt @@ -550,6 +550,14 @@ internal object WalletManagerNative { */ external fun shieldedClear(managerHandle: Long) + /** + * Repair an inconsistent shielded commitment tree: empty it and rewind + * scan cursors, keeping notes, spent status and registrations. Background + * shielded sync restarts by itself if it was running. Throws on failure. + * Only present when the native library is built with shielded. + */ + external fun shieldedRebuildTree(managerHandle: Long) + /** * Start the Core SPV client — flattened form of * `platform_wallet_manager_spv_start` (11 discrete params, no config diff --git a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/wallet/PlatformWalletManager.kt b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/wallet/PlatformWalletManager.kt index e552cdfcdfc..c8b5e604d08 100644 --- a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/wallet/PlatformWalletManager.kt +++ b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/wallet/PlatformWalletManager.kt @@ -366,6 +366,7 @@ class PlatformWalletManager( newlySpent: Int, balance: Long, errorMessage: String?, + errorCode: Int, ) { _syncEvents.tryEmit( WalletSyncEvent.ShieldedResult( @@ -378,6 +379,7 @@ class PlatformWalletManager( newlySpent = newlySpent, balance = balance, errorMessage = errorMessage, + errorCode = errorCode, ), ) } @@ -1610,6 +1612,23 @@ class PlatformWalletManager( mapNativeErrors { WalletManagerNative.shieldedClear(managerHandle) } } + /** + * Repair an inconsistent shielded commitment tree — port of Swift's + * `PlatformWalletManager.rebuildShieldedTree()`. Call it when a sync + * reports [DashSdkError.PlatformWallet.ShieldedTreeInconsistent]. Empties + * the tree and rewinds scan cursors so the next pass rescans the whole + * pool; notes, spent status and registrations are kept, so the host keeps + * its Room rows. Background shielded sync restarts by itself if it was + * running, except after a quiesce timeout on top of an earlier stop's: + * check that it is running after a successful retry. Concurrent rebuilds + * run one at a time, so a second caller also waits for the first to + * finish. Throws on failure. Only meaningful on a shielded build + * ([Sdk.hasShielded]); the native entry point is absent otherwise. + */ + suspend fun rebuildShieldedTree() = withContext(Dispatchers.IO) { + mapNativeErrors { WalletManagerNative.shieldedRebuildTree(managerHandle) } + } + // ── Shielded funding submits ────────────────────────────────────── // // These are manager-handle FFI calls (they resolve the wallet + its diff --git a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/wallet/WalletSyncEvent.kt b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/wallet/WalletSyncEvent.kt index 41e82ce83f2..64659173add 100644 --- a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/wallet/WalletSyncEvent.kt +++ b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/wallet/WalletSyncEvent.kt @@ -1,5 +1,7 @@ package org.dashfoundation.dashsdk.wallet +import org.dashfoundation.dashsdk.errors.DashSdkError + /** * Typed sync events fanned out by [PlatformWalletManager] from the native * `EventHandlerCallbacks` vtable (bridged in `rs-unified-sdk-jni/src/events.rs`). @@ -111,7 +113,24 @@ sealed interface WalletSyncEvent { val newlySpent: Int, val balance: Long, val errorMessage: String?, + /** De-offset `PlatformWalletFFIResultCode`; 0 on success / skipped. */ + val errorCode: Int = 0, ) : WalletSyncEvent { + /** + * The failure as a typed error, or null on success / skipped. + * [DashSdkError.PlatformWallet.ShieldedTreeInconsistent] means only + * [PlatformWalletManager.rebuildShieldedTree] repairs it. + */ + val error: DashSdkError? + get() = if (success || skipped) { + null + } else { + DashSdkError.fromPlatformWalletCode( + errorCode, + errorMessage ?: "shielded sync failed", + ) + } + override fun equals(other: Any?): Boolean = other is ShieldedResult && walletId.contentEquals(other.walletId) && @@ -122,7 +141,8 @@ sealed interface WalletSyncEvent { totalScanned == other.totalScanned && newlySpent == other.newlySpent && balance == other.balance && - errorMessage == other.errorMessage + errorMessage == other.errorMessage && + errorCode == other.errorCode override fun hashCode(): Int { var result = walletId.contentHashCode() diff --git a/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/errors/DashSdkErrorTest.kt b/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/errors/DashSdkErrorTest.kt index fd09ae4051a..f13f86214b1 100644 --- a/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/errors/DashSdkErrorTest.kt +++ b/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/errors/DashSdkErrorTest.kt @@ -19,6 +19,16 @@ class DashSdkErrorTest { assertFalse(mapped.isRetryable) } + @Test + fun shouldMapAnInconsistentShieldedTreeToItsTypedError() { + val native = DashSDKException(DashSdkError.PLATFORM_WALLET_CODE_OFFSET + 67, "3 leaves, 2 proven") + val mapped = DashSdkError.fromNative(native) + assertTrue(mapped is DashSdkError.PlatformWallet.ShieldedTreeInconsistent) + assertEquals(native.message, mapped.message) + assertEquals(native, mapped.cause) + assertFalse(mapped.isRetryable) + } + @Test fun mapsNativeCodesToHierarchy() { // Codes mirror DashSDKErrorCode in rs-sdk-ffi/src/error.rs. diff --git a/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/wallet/WalletEventFanOutTest.kt b/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/wallet/WalletEventFanOutTest.kt index e11f58621f5..e626b4139e6 100644 --- a/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/wallet/WalletEventFanOutTest.kt +++ b/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/wallet/WalletEventFanOutTest.kt @@ -7,7 +7,10 @@ import kotlinx.coroutines.launch import kotlinx.coroutines.test.UnconfinedTestDispatcher import kotlinx.coroutines.test.runTest import org.dashfoundation.dashsdk.ffi.NativeWalletEventBridge +import org.dashfoundation.dashsdk.errors.DashSdkError import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull import org.junit.Assert.assertTrue import org.junit.Test @@ -88,11 +91,12 @@ class WalletEventFanOutTest { newlySpent: Int, balance: Long, errorMessage: String?, + errorCode: Int, ) { sink.tryEmit( WalletSyncEvent.ShieldedResult( walletId, success, skipped, cooldownSkip, newNotes, - totalScanned, newlySpent, balance, errorMessage, + totalScanned, newlySpent, balance, errorMessage, errorCode, ), ) } @@ -130,7 +134,7 @@ class WalletEventFanOutTest { bridge.onPlatformAddressSyncPassCompleted(1_000, 1) bridge.onDpnsMarketplaceSyncCompleted(walletId, true, 7, 8, 9, 10, null) bridge.onDpnsMarketplaceSyncPassCompleted(1_500, 1) - bridge.onShieldedSyncCompleted(walletId, true, false, false, 7, 8, 9, 10, null) + bridge.onShieldedSyncCompleted(walletId, true, false, false, 7, 8, 9, 10, null, 0) bridge.onShieldedSyncPassCompleted(2_000, 1) bridge.onShieldedSyncProgress(11, 12) bridge.onShieldedTreeProgress(13, 14) @@ -161,6 +165,7 @@ class WalletEventFanOutTest { assertEquals(7, sr.newNotes) assertEquals(8L, sr.totalScanned) assertEquals(10L, sr.balance) + assertNull(sr.error) assertTrue(received[7] is WalletSyncEvent.ShieldedPassCompleted) val sp = received[8] as WalletSyncEvent.ShieldedProgress @@ -168,4 +173,24 @@ class WalletEventFanOutTest { val tp = received[9] as WalletSyncEvent.ShieldedTreeProgress assertEquals(13L, tp.leavesCommitted) } + + @Test + fun shouldTypeAnInconsistentShieldedTreeFromItsResultCode() { + val walletId = ByteArray(32) { 6 } + val result = WalletSyncEvent.ShieldedResult( + walletId, false, false, false, 0, 0, 0, 0, + "tree has 3 leaves, chain reports 2", 67, + ) + + val error = result.error + assertTrue(error is DashSdkError.PlatformWallet.ShieldedTreeInconsistent) + assertEquals("tree has 3 leaves, chain reports 2", error!!.message) + assertFalse("a resync cannot repair the tree; rebuild it", error.isRetryable) + + val generic = WalletSyncEvent.ShieldedResult( + walletId, false, false, false, 0, 0, 0, 0, "transport failed", 6, + ).error + assertFalse(generic is DashSdkError.PlatformWallet.ShieldedTreeInconsistent) + assertEquals("transport failed", generic!!.message) + } } diff --git a/packages/rs-platform-wallet-ffi/ERROR_CODE_REGISTRY.md b/packages/rs-platform-wallet-ffi/ERROR_CODE_REGISTRY.md index a916a4eb433..4a172f5175c 100644 --- a/packages/rs-platform-wallet-ffi/ERROR_CODE_REGISTRY.md +++ b/packages/rs-platform-wallet-ffi/ERROR_CODE_REGISTRY.md @@ -114,7 +114,9 @@ These are shipped ABI. Do not renumber. | 98 | `NotFound` | Sentinel — `Option` returned as an error | | 99 | `ErrorUnknown` | Sentinel — unmapped/flattened errors | -**Next allocatable integer: 61** — 27–60 are all claimed (27, 29, 31, 34–42 +**Next allocatable integer: 68** — 27–67 are all claimed (67 proposed by +#5565 for an inconsistent shielded commitment tree; 61 is held by open #5395, +62 by open #5487 and 63–66 by open #5536, which record their own rows; 27, 29, 31, 34–42 and 46 merged; 43–45 proposed by active #4313 at head `0302b188ab`; 47 and 48 proposed by active #4356 (47 renumbered from 42, 48 from 43 — see their rows below); 49–54 proposed by active #4586 (the persister @@ -177,6 +179,7 @@ Fork-era numbers remain in the collision history, which is immutable record. | 58 | `ErrorIdentityBalanceUnavailable` | #4799 | Proposed — Platform returned no balance for a managed identity. Retrying the read is safe; this is distinct from missing wallet ownership and must not trigger registration or funding | | 59 | `ErrorSpvProcessRestartRequired` | #5322 | Proposed — SPV startup or teardown panicked; restart the host process. Not retryable, unlike 27, which SPV start, stop and storage clear now return while an SPV teardown is still running. Typed in Rust, Swift and Kotlin (the PR also types 27 on Kotlin) | | 60 | `ErrorAssetLockTooManyInputs` | Core-to-shielded pooled Max | Proposed — the full drain needs more inputs than an asset lock may carry (100 on Dash Core relay policy and Platform consensus); no partial drain was broadcast. Rust, Swift and Kotlin retain the typed refusal. Renumbered from 59 after #5322 claimed it | +| 67 | `ErrorShieldedTreeInconsistent` | #5565 | Proposed — the local shielded commitment tree holds more leaves than a sufficiently fresh sync proves; resyncing cannot repair it, and `platform_wallet_manager_shielded_rebuild_tree` does, keeping notes and spent status. Not retryable. Returned by `platform_wallet_manager_shielded_sync_wallet` and carried per wallet in `ShieldedSyncWalletResultFFI.error_code` on the sync-completed callback. Skips 61–66, held by open #5395, #5487 and #5536. Rust value + pin test, Swift raw case + `init(ffi:)` arm + typed `shieldedTreeInconsistent` + `ErrorHandlingTests` pin, Kotlin typed `PlatformWallet.ShieldedTreeInconsistent` + `fromPlatformWalletNative` arm + `DashSdkErrorTest` pin | **Code 31 left this table on 2026-08-04.** `ErrorSigningKeyUnavailable` sat here as #4183's proposal until #4183 merged (`189a3abb1c`); it is now in the merged diff --git a/packages/rs-platform-wallet-ffi/src/error.rs b/packages/rs-platform-wallet-ffi/src/error.rs index d14d197ddb2..c73b36a1fcb 100644 --- a/packages/rs-platform-wallet-ffi/src/error.rs +++ b/packages/rs-platform-wallet-ffi/src/error.rs @@ -619,6 +619,14 @@ pub enum PlatformWalletFFIResultCode { /// host must not silently substitute a partial drain. ErrorAssetLockTooManyInputs = 60, + /// Maps `PlatformWalletError::ShieldedTreeInconsistent`: the local shielded + /// commitment tree holds more leaves than a sufficiently fresh sync proves. + /// Resyncing cannot repair it; call + /// `platform_wallet_manager_shielded_rebuild_tree`, which keeps notes and + /// spent status. Also carried per wallet by the shielded sync-completed + /// callback. 61-66 are claimed by open PRs; see `ERROR_CODE_REGISTRY.md`. + ErrorShieldedTreeInconsistent = 67, + /// The named thing does not exist. /// /// Originally (and still mostly) the code for every `Option` returned as an @@ -928,6 +936,9 @@ impl From for PlatformWalletFFIResult { PlatformWalletError::ShieldedNoRecordedAnchor(..) => { PlatformWalletFFIResultCode::ErrorShieldedNoRecordedAnchor } + PlatformWalletError::ShieldedTreeInconsistent { .. } => { + PlatformWalletFFIResultCode::ErrorShieldedTreeInconsistent + } PlatformWalletError::PlatformShieldCapacityExceeded { .. } => { PlatformWalletFFIResultCode::ErrorShieldedInsufficientBalance } @@ -2214,6 +2225,25 @@ mod tests { ); } + /// Hosts offer the tree rebuild by code alone, so the inconsistent-tree + /// verdict must keep its own pinned code rather than flatten. + #[test] + fn should_map_shielded_tree_inconsistent_to_pinned_code_67() { + assert_eq!( + PlatformWalletFFIResultCode::ErrorShieldedTreeInconsistent as i32, + 67 + ); + let result: PlatformWalletFFIResult = PlatformWalletError::ShieldedTreeInconsistent { + local_leaf_count: 3, + chain_leaf_count: 2, + } + .into(); + assert_eq!( + result.code, + PlatformWalletFFIResultCode::ErrorShieldedTreeInconsistent + ); + } + #[test] fn shielded_insufficient_balance_code_is_pinned_at_41() { assert_eq!( diff --git a/packages/rs-platform-wallet-ffi/src/event_handler.rs b/packages/rs-platform-wallet-ffi/src/event_handler.rs index 0db7649a7c6..ad1cbd6405d 100644 --- a/packages/rs-platform-wallet-ffi/src/event_handler.rs +++ b/packages/rs-platform-wallet-ffi/src/event_handler.rs @@ -1,5 +1,7 @@ //! FFI callback-based implementation of PlatformEventHandler. +#[cfg(feature = "shielded")] +use crate::error::PlatformWalletFFIResultCode; use crate::platform_address_sync::{ PlatformAddressSyncMetricsFFI, PlatformAddressSyncWalletResultFFI, }; @@ -8,6 +10,8 @@ use platform_wallet::events::{EventHandler, PlatformEventHandler, WalletEvent}; use platform_wallet::manager::dpns_sync::{DpnsSyncPassSummary, WalletDpnsSyncOutcome}; #[cfg(feature = "shielded")] use platform_wallet::manager::shielded_sync::{ShieldedSyncPassSummary, WalletShieldedOutcome}; +#[cfg(feature = "shielded")] +use platform_wallet::PlatformWalletError; use platform_wallet::{PlatformAddressSyncSummary, WalletSyncOutcome}; use std::os::raw::{c_char, c_void}; @@ -398,16 +402,35 @@ impl PlatformEventHandler for FFIEventHandler { // and retain its message for the entire callback invocation. let result = match outcome { WalletShieldedOutcome::Ok(result) => { - ShieldedSyncWalletResultFFI::ok(wallet_id, result).map_err(|e| e.to_string()) + ShieldedSyncWalletResultFFI::ok(wallet_id, result).map_err(|e| { + ( + PlatformWalletFFIResultCode::ErrorWalletOperation, + e.to_string(), + ) + }) } WalletShieldedOutcome::Skipped => { Ok(ShieldedSyncWalletResultFFI::skipped(wallet_id)) } - WalletShieldedOutcome::Err(error) => Err(error.clone()), + WalletShieldedOutcome::Err(error) => Err(( + PlatformWalletFFIResultCode::ErrorWalletOperation, + error.clone(), + )), + WalletShieldedOutcome::TreeInconsistent { + local_leaf_count, + chain_leaf_count, + } => Err(( + PlatformWalletFFIResultCode::ErrorShieldedTreeInconsistent, + PlatformWalletError::ShieldedTreeInconsistent { + local_leaf_count: *local_leaf_count, + chain_leaf_count: *chain_leaf_count, + } + .to_string(), + )), }; match result { Ok(result) => results.push(result), - Err(error) => { + Err((error_code, error)) => { let error_message = std::ffi::CString::new(error).ok(); let error_ptr = error_message .as_ref() @@ -415,7 +438,9 @@ impl PlatformEventHandler for FFIEventHandler { if let Some(error_message) = error_message { owned_errors.push(error_message); } - results.push(ShieldedSyncWalletResultFFI::err(wallet_id, error_ptr)); + results.push(ShieldedSyncWalletResultFFI::err( + wallet_id, error_code, error_ptr, + )); } } } @@ -539,6 +564,94 @@ mod release_tests { ); } + /// Background sync is the path hosts actually run, so an inconsistent + /// tree must reach them as a result code, not only as message text. + #[cfg(feature = "shielded")] + #[test] + fn should_report_typed_error_codes_through_the_shielded_sync_callback() { + use platform_wallet::wallet::shielded::ShieldedSyncSummary; + use std::collections::BTreeMap; + use std::ffi::CStr; + + type Received = Vec<([u8; 32], PlatformWalletFFIResultCode, Option)>; + unsafe extern "C" fn capture( + context: *mut c_void, + results: *const ShieldedSyncWalletResultFFI, + count: usize, + _: u64, + ) { + let received = &mut *(context as *mut Received); + for result in std::slice::from_raw_parts(results, count) { + received.push(( + result.wallet_id, + result.error_code, + (!result.error_message.is_null()).then(|| { + CStr::from_ptr(result.error_message) + .to_string_lossy() + .into_owned() + }), + )); + } + } + let mut received = Received::new(); + let handler = FFIEventHandler::new( + EventHandlerCallbacks { + context: &mut received as *mut Received as *mut c_void, + on_wallet_event_fn: None, + on_error_fn: None, + on_platform_address_sync_completed_fn: None, + on_shielded_sync_completed_fn: Some(capture), + on_shielded_sync_progress_fn: None, + on_shielded_tree_progress_fn: None, + release_fn: None, + }, + None, + ); + handler.on_shielded_sync_completed(&ShieldedSyncPassSummary { + wallet_results: BTreeMap::from([ + ( + [1; 32], + WalletShieldedOutcome::Ok(ShieldedSyncSummary::default()), + ), + ([2; 32], WalletShieldedOutcome::Skipped), + ( + [3; 32], + WalletShieldedOutcome::Err("transport failed".to_string()), + ), + ( + [4; 32], + WalletShieldedOutcome::TreeInconsistent { + local_leaf_count: 3, + chain_leaf_count: 2, + }, + ), + ]), + sync_unix_seconds: 100, + }); + let inconsistent = PlatformWalletError::ShieldedTreeInconsistent { + local_leaf_count: 3, + chain_leaf_count: 2, + } + .to_string(); + assert_eq!( + received, + vec![ + ([1; 32], PlatformWalletFFIResultCode::Success, None), + ([2; 32], PlatformWalletFFIResultCode::Success, None), + ( + [3; 32], + PlatformWalletFFIResultCode::ErrorWalletOperation, + Some("transport failed".to_string()) + ), + ( + [4; 32], + PlatformWalletFFIResultCode::ErrorShieldedTreeInconsistent, + Some(inconsistent) + ), + ] + ); + } + /// Mirror of the persister-level test: the event context is released /// exactly once, only when the last `Arc` /// clone drops — so a straggling event dispatcher keeps the host diff --git a/packages/rs-platform-wallet-ffi/src/shielded_sync.rs b/packages/rs-platform-wallet-ffi/src/shielded_sync.rs index 95f4b94f037..0bd8dd2c156 100644 --- a/packages/rs-platform-wallet-ffi/src/shielded_sync.rs +++ b/packages/rs-platform-wallet-ffi/src/shielded_sync.rs @@ -49,6 +49,7 @@ impl ShieldedSyncWalletResultFFI { newly_spent, balance, error_message: std::ptr::null(), + error_code: PlatformWalletFFIResultCode::Success, }) } } @@ -218,7 +219,9 @@ pub unsafe extern "C" fn platform_wallet_manager_local_shielded_balance_snapshot // Shielded sync coordinator FFI // --------------------------------------------------------------------------- -/// Start the shielded sync manager in the background. +/// Start the shielded sync manager in the background. After a stop whose +/// drain timed out, the loop reports running while every pass bails, until a +/// later drain (a stop or a rebuild) succeeds. #[no_mangle] pub unsafe extern "C" fn platform_wallet_manager_shielded_sync_start( handle: Handle, @@ -278,7 +281,8 @@ pub unsafe extern "C" fn platform_wallet_manager_shielded_sync_stop( PlatformWalletFFIResult::ok() } -/// Whether the shielded sync background loop is running. +/// Whether the shielded sync background loop is running. A running loop can +/// still be doing no work: see `platform_wallet_manager_shielded_sync_start`. #[no_mangle] pub unsafe extern "C" fn platform_wallet_manager_shielded_sync_is_running( handle: Handle, @@ -574,7 +578,9 @@ pub unsafe extern "C" fn platform_wallet_manager_configure_shielded( /// persistence: a silent failure would leave the shared tree /// populated while the host drops its rows, and the next cold /// resync would gate-skip every re-downloaded position against the -/// stale tree size. +/// stale tree size. An error does not prove the tree is unchanged: a reset +/// whose post-commit durability checkpoint failed or was busy has taken effect +/// and the tree is empty. Keep the rows and retry; a retry just resets again. /// /// Errors with `ErrorWalletOperation` when no shielded coordinator is /// installed on this manager (the sync-loop stop still runs unconditionally @@ -616,6 +622,43 @@ pub unsafe extern "C" fn platform_wallet_manager_shielded_clear( PlatformWalletFFIResult::ok() } +// --------------------------------------------------------------------------- +// Rebuild the shielded commitment tree +// --------------------------------------------------------------------------- + +/// Repair an inconsistent shielded commitment tree +/// (`ErrorShieldedTreeInconsistent`): empty the tree and rewind scan +/// cursors so the next sync rescans the whole pool. Notes, spent status +/// and wallet registrations are kept, so the host keeps its own rows. +/// +/// Background shielded sync is paused for the reset and restarted if it +/// was running; on success the host needs no follow-up call. The rescan +/// downloads and trial-decrypts the entire pool. Concurrent rebuilds run one +/// at a time, so a second caller also waits for the first to finish. +/// +/// Returns `ErrorShutdownIncomplete` if an in-flight pass did not drain +/// (nothing was reset; retry once sync is idle). If an earlier stop's drain +/// had also timed out, the rebuild does not restart the loop: start it after +/// the retry succeeds. +/// +/// Returns `ErrorWalletOperation` if no shielded coordinator is configured, +/// or the reset fails, or the restart task dies after a successful reset. A +/// reset whose post-commit durability checkpoint failed or was busy has taken +/// effect; a retry just resets again. +#[no_mangle] +pub unsafe extern "C" fn platform_wallet_manager_shielded_rebuild_tree( + handle: Handle, +) -> PlatformWalletFFIResult { + let option = PLATFORM_WALLET_MANAGER_STORAGE.with_item(handle, |manager| { + runtime().block_on(manager.rebuild_shielded_tree()) + }); + match unwrap_option_or_return!(option) { + Ok(()) => PlatformWalletFFIResult::ok(), + Err(e @ PlatformWalletError::ShutdownIncomplete(_)) => e.into(), + Err(e) => map_shielded_error(e, "rebuild_shielded_tree"), + } +} + // --------------------------------------------------------------------------- // Default Orchard payment address // --------------------------------------------------------------------------- @@ -727,11 +770,13 @@ pub unsafe extern "C" fn platform_wallet_manager_shielded_sync_wallet( } /// Recovery errors must remain typed when opening, binding or syncing durable state, -/// just as they do when submitting or explicitly abandoning a payment. +/// just as they do when submitting or explicitly abandoning a payment. So must an +/// inconsistent tree, which only a rebuild repairs. fn map_shielded_error(error: PlatformWalletError, operation: &str) -> PlatformWalletFFIResult { match error { error @ PlatformWalletError::ShieldedRecoveryCorrupted { .. } - | error @ PlatformWalletError::ShieldedRecoveryKeysRequired { .. } => error.into(), + | error @ PlatformWalletError::ShieldedRecoveryKeysRequired { .. } + | error @ PlatformWalletError::ShieldedTreeInconsistent { .. } => error.into(), other => PlatformWalletFFIResult::err( PlatformWalletFFIResultCode::ErrorWalletOperation, format!("{operation} failed: {other}"), @@ -760,6 +805,13 @@ mod recovery_error_tests { }, PlatformWalletFFIResultCode::ErrorShieldedRecoveryKeysRequired, ), + ( + PlatformWalletError::ShieldedTreeInconsistent { + local_leaf_count: 3, + chain_leaf_count: 2, + }, + PlatformWalletFFIResultCode::ErrorShieldedTreeInconsistent, + ), ] { let message = error.to_string(); let result = map_shielded_error(error, "configure or bind"); @@ -787,7 +839,7 @@ mod local_balance_tests { use std::task::Poll; use std::time::Instant; - fn bound_manager() -> ( + pub(super) fn bound_manager() -> ( Handle, [u8; 32], Arc, @@ -1136,3 +1188,70 @@ mod local_balance_tests { unsafe { crate::platform_wallet_ffi_result_free(&mut result) }; } } + +#[cfg(test)] +mod rebuild_tree_tests { + use super::local_balance_tests::bound_manager; + use super::*; + use platform_wallet::wallet::shielded::ShieldedStore; + + #[test] + fn should_reject_rebuild_on_an_invalid_handle() { + let mut result = unsafe { platform_wallet_manager_shielded_rebuild_tree(NULL_HANDLE) }; + assert_eq!(result.code, PlatformWalletFFIResultCode::NotFound); + unsafe { crate::platform_wallet_ffi_result_free(&mut result) }; + } + + /// The export reaches the manager's rebuild: the tree empties, the + /// registration survives, and background sync is running again. + #[test] + fn should_rebuild_tree_keep_registration_and_restore_background_sync() { + let (handle, wallet_id, coordinator, path) = bound_manager(); + runtime().block_on(async { + coordinator + .store() + .write() + .await + .append_commitment(&[7; 32], true) + .expect("append leaf"); + }); + let mut start = unsafe { platform_wallet_manager_shielded_sync_start(handle) }; + assert_eq!(start.code, PlatformWalletFFIResultCode::Success); + + let mut result = unsafe { platform_wallet_manager_shielded_rebuild_tree(handle) }; + let code = result.code; + + let mut running = false; + let mut is_running = + unsafe { platform_wallet_manager_shielded_sync_is_running(handle, &mut running) }; + let (tree_size, registered) = runtime().block_on(async { + ( + coordinator + .store() + .read() + .await + .tree_size() + .expect("tree size"), + coordinator + .registered_subwallets() + .await + .iter() + .any(|id| id.wallet_id == wallet_id), + ) + }); + unsafe { + crate::platform_wallet_ffi_result_free(&mut start); + crate::platform_wallet_ffi_result_free(&mut result); + crate::platform_wallet_ffi_result_free(&mut is_running); + let mut destroy = crate::manager::platform_wallet_manager_destroy(handle); + crate::platform_wallet_ffi_result_free(&mut destroy); + } + drop(coordinator); + std::fs::remove_file(path).expect("remove test store"); + + assert_eq!(code, PlatformWalletFFIResultCode::Success); + assert_eq!(tree_size, 0, "rebuild must empty the commitment tree"); + assert!(registered, "rebuild must keep wallet registrations"); + assert!(running, "rebuild must restore the running background sync"); + } +} diff --git a/packages/rs-platform-wallet-ffi/src/shielded_types.rs b/packages/rs-platform-wallet-ffi/src/shielded_types.rs index 156d25b352a..2a5c756c8f7 100644 --- a/packages/rs-platform-wallet-ffi/src/shielded_types.rs +++ b/packages/rs-platform-wallet-ffi/src/shielded_types.rs @@ -10,6 +10,8 @@ use std::os::raw::c_char; +use crate::error::PlatformWalletFFIResultCode; + #[cfg(feature = "shielded")] use platform_wallet::wallet::shielded::{IdentityDebitRecoveryRecord, IdentityDebitRecoveryStatus}; @@ -99,7 +101,7 @@ pub struct ShieldedShieldPreflightFFI { /// so the pass passed it over; `success` is false and /// `error_message` is NULL. /// - both flags `false` and `error_message != NULL`: the sync -/// itself failed. +/// itself failed; `error_code` classifies the failure. #[repr(C)] #[derive(Debug, Clone, Copy)] pub struct ShieldedSyncWalletResultFFI { @@ -128,6 +130,11 @@ pub struct ShieldedSyncWalletResultFFI { /// NUL-terminated UTF-8 error message; valid until the callback /// returns. NULL on success and skipped wallets. pub error_message: *const c_char, + /// `Success` on success and skipped wallets. On failure, + /// `ErrorShieldedTreeInconsistent` means only + /// `platform_wallet_manager_shielded_rebuild_tree` repairs it; + /// any other failure is `ErrorWalletOperation`. + pub error_code: PlatformWalletFFIResultCode, } impl Default for ShieldedSyncWalletResultFFI { @@ -142,6 +149,7 @@ impl Default for ShieldedSyncWalletResultFFI { newly_spent: 0, balance: 0, error_message: std::ptr::null(), + error_code: PlatformWalletFFIResultCode::Success, } } } @@ -160,10 +168,15 @@ impl ShieldedSyncWalletResultFFI { } } - pub(crate) fn err(wallet_id: [u8; 32], error_ptr: *const c_char) -> Self { + pub(crate) fn err( + wallet_id: [u8; 32], + error_code: PlatformWalletFFIResultCode, + error_ptr: *const c_char, + ) -> Self { Self { wallet_id, error_message: error_ptr, + error_code, ..Self::default() } } diff --git a/packages/rs-platform-wallet/src/error.rs b/packages/rs-platform-wallet/src/error.rs index 85b86828e64..fcefca9a7aa 100644 --- a/packages/rs-platform-wallet/src/error.rs +++ b/packages/rs-platform-wallet/src/error.rs @@ -1028,9 +1028,27 @@ pub enum PlatformWalletError { #[error("Background sync did not quiesce: {0}")] ShutdownIncomplete(String), + /// The shielded tree reset took effect, but the task restarting background + /// shielded sync panicked or was cancelled. The loop may be stopped; check + /// `is_running()` and start it if needed. Do not rebuild again for this. + #[error("Background shielded sync did not restart after the tree reset: {0}")] + ShieldedSyncRestartFailed(String), + #[error("Shielded commitment tree update failed: {0}")] ShieldedTreeUpdateFailed(String), + /// The local tree exceeds the maximum proven count from a sync containing + /// a response at least as fresh as the tree's recorded height bound. + /// Stale responses and unknown tree provenance cannot trigger this error. + /// No automatic repair is performed. + #[error("Shielded commitment tree has {local_leaf_count} leaves, but a sufficiently fresh sync reports only {chain_leaf_count}; explicitly rebuild the shielded tree and resync")] + ShieldedTreeInconsistent { + /// Number of leaves in the local tree after consuming the stream. + local_leaf_count: u64, + /// Largest commitment count reported by the stream's proofs. + chain_leaf_count: u64, + }, + #[error("Shielded store error: {0}")] ShieldedStoreError(String), diff --git a/packages/rs-platform-wallet/src/manager/mod.rs b/packages/rs-platform-wallet/src/manager/mod.rs index 6ea8ba9b5fd..a9eeedd2058 100644 --- a/packages/rs-platform-wallet/src/manager/mod.rs +++ b/packages/rs-platform-wallet/src/manager/mod.rs @@ -229,6 +229,13 @@ impl QuiesceGate { QuiesceGuard(self) } + /// Whether the gate is closed with no holder whose drop could reopen it: + /// a loop started now would report running while every pass bails. + pub(crate) fn is_stuck_closed(&self) -> bool { + let bookkeeping = self.bookkeeping(); + bookkeeping.latched || bookkeeping.sealed + } + /// A drain observed the pass fully drained while holding the gate: /// clear any latch left by a previously timed-out drain. The caller /// still holds its guard, so the gate stays closed until that drops. @@ -301,15 +308,41 @@ pub(crate) async fn drain_pass<'a>( is_syncing: &std::sync::atomic::AtomicBool, stop: impl FnOnce(), budget: Duration, +) -> Option> { + drain_pass_with(gate, is_syncing, stop, budget, DrainTimeout::Latch).await +} + +/// What a [`drain_pass_with`] that times out leaves on the gate. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum DrainTimeout { + /// Latch the gate closed (see [`drain_pass`]). + Latch, + /// Drop only the drain's own hold. For a caller that, on timeout, + /// changes nothing and resumes sync: a latch would leave its restarted + /// loop reporting running while every pass bails. The wedged pass + /// still holds `is_syncing`, so no pass overlaps it; the next one runs + /// once it ends. A latch set by an earlier drain stays. + Release, +} + +/// [`drain_pass`] with an explicit timeout policy. +pub(crate) async fn drain_pass_with<'a>( + gate: &'a QuiesceGate, + is_syncing: &std::sync::atomic::AtomicBool, + stop: impl FnOnce(), + budget: Duration, + on_timeout: DrainTimeout, ) -> Option> { let guard = gate.hold(); stop(); let deadline = tokio::time::Instant::now() + budget; while is_syncing.load(std::sync::atomic::Ordering::Acquire) { if tokio::time::Instant::now() >= deadline { - // Latch BEFORE the guard drops so there is no instant in - // which the gate is open on the timeout path. - gate.latch_closed(); + // Latch BEFORE the guard drops so a latching drain leaves no + // instant in which the gate is open on the timeout path. + if on_timeout == DrainTimeout::Latch { + gate.latch_closed(); + } drop(guard); return None; } @@ -410,6 +443,12 @@ pub struct PlatformWalletManager { #[cfg(feature = "shielded")] pub(super) shielded_coordinator: Arc>>>, + /// Serializes tree rebuilds. Without it, a second rebuild reads the loop + /// as stopped because the first quiesced it, and its clearing latch then + /// refuses the first's restart. Outermost lock of a rebuild: stop, Clear + /// and shutdown never take it, so they cannot wait on a rebuild here. + #[cfg(feature = "shielded")] + shielded_rebuild_lock: tokio::sync::Mutex<()>, /// Shared `PlatformEventManager` — held on the manager so /// `configure_shielded` can install a per-chunk progress handler /// onto the freshly-created `NetworkShieldedCoordinator` that @@ -648,6 +687,8 @@ impl PlatformWalletManager

{ shielded_sync_manager: shielded_sync, #[cfg(feature = "shielded")] shielded_coordinator, + #[cfg(feature = "shielded")] + shielded_rebuild_lock: tokio::sync::Mutex::new(()), #[cfg(any(test, feature = "shielded"))] event_manager, persister, @@ -994,6 +1035,8 @@ impl PlatformWalletManager

{ // the store `coord.clear()` is about to reset. The guard's Drop // releases the latch on every exit path (including `?` and panic). let _clearing = self.registry.hold_clearing(WalletWorker::ShieldedSync); + // Clear stops the loop for good: an in-flight rebuild must not restart it. + self.shielded_sync_manager.record_stop_intent(); // Hold sync admission shut for the WHOLE quiesce -> wipe as well. // The clearing latch only bars registry (re)starts; a direct // `sync_now` / `sync_wallet` on a host thread does not consult it, @@ -1028,6 +1071,96 @@ impl PlatformWalletManager

{ } } + /// Rebuild the network's shielded commitment tree on the next sync. + /// Preserves notes, nullifiers, spent status, and wallet registrations while + /// clearing the tree and rewinding scan cursors. The full rescan from zero + /// costs bandwidth and trial decryption proportional to the entire pool. + /// Background shielded sync is paused for the reset and, if it was + /// running, restarted afterwards on every outcome unless a stop or Clear + /// was requested meanwhile; its first pass begins the rescan at once. + /// Restarting after a failure is safe: a reset that rolled back leaves a + /// non-empty tree and its cursors untouched, so the next pass reports the + /// same inconsistency; a reset whose durability checkpoint failed or was + /// busy has taken effect, so the next pass rescans; and a quiesce timeout + /// leaves sync admission open for the passes after the wedged one. + /// Concurrent rebuilds run one at a time. + /// + /// The rebuild does not restart the loop if an earlier stop's timed-out + /// drain still latches the gate, because no pass could run. The call then + /// returns [`PlatformWalletError::ShutdownIncomplete`]; once a retry + /// drains, the latch is clear and the host starts the loop again. This + /// covers the rebuild's own restart only: a host + /// [`start`](ShieldedSyncManager::start) from that state still reports + /// running while every pass bails, until a later drain clears the latch. + /// + /// The restart runs on the blocking pool, so awaiting this call does not + /// occupy an async worker. If it dies there, a reset that succeeded + /// returns [`PlatformWalletError::ShieldedSyncRestartFailed`]; a failed + /// reset keeps its own error. Never invoked automatically. + #[cfg(feature = "shielded")] + pub async fn rebuild_shielded_tree(&self) -> Result<(), PlatformWalletError> { + self.rebuild_shielded_tree_within(COORDINATOR_DRAIN_BUDGET) + .await + } + + /// [`Self::rebuild_shielded_tree`] with an explicit quiesce budget. + #[cfg(feature = "shielded")] + async fn rebuild_shielded_tree_within( + &self, + budget: Duration, + ) -> Result<(), PlatformWalletError> { + let _serialized = self.shielded_rebuild_lock.lock().await; + // Read before `is_running`, so a stop racing this entry still wins. + let stop_intents = self.shielded_sync_manager.stop_intents(); + let was_running = self.shielded_sync_manager.is_running(); + let result = self.rebuild_shielded_tree_quiesced(budget).await; + // The clearing latch and quiesce guard dropped with the call above; + // while either is held, `start` is a no-op. + if was_running { + let restart = self + .shielded_sync_manager + .restart_off_worker_unless_stopped_since(stop_intents) + .await; + if let Err(restart) = restart { + if result.is_ok() { + return Err(restart); + } + // The reset's own error tells the host whether to retry. + tracing::error!(error = %restart, "shielded sync did not restart after a failed rebuild"); + } + } + result + } + + /// Reset half of [`Self::rebuild_shielded_tree`], with background sync + /// held stopped and admission closed for its whole duration. + #[cfg(feature = "shielded")] + async fn rebuild_shielded_tree_quiesced( + &self, + budget: Duration, + ) -> Result<(), PlatformWalletError> { + let _clearing = self.registry.hold_clearing(WalletWorker::ShieldedSync); + // A timeout here resets nothing, so it must not latch the gate: the + // loop restarted afterwards would report running while every pass + // bails. + let Some(_quiesced) = self + .shielded_sync_manager + .quiesce_held_releasing_within(budget) + .await + else { + return Err(PlatformWalletError::ShutdownIncomplete( + "shielded sync pass did not drain within the quiesce budget; rebuild aborted" + .to_string(), + )); + }; + let coordinator = self.shielded_coordinator().await.ok_or_else(|| { + PlatformWalletError::ShieldedStoreError( + "shielded rebuild requested but no coordinator is configured".to_string(), + ) + })?; + coordinator.rebuild_shielded_tree().await + } + /// Reset the platform-address (BLAST/DIP-17) incremental-sync /// watermark and drop every cached balance across **all** /// registered wallets, forcing a full rescan on the next sync. @@ -1831,4 +1964,387 @@ mod tests { .expect("the parked adapter task never resumed after the manager was dropped") .expect("the adapter task was aborted on drop: it never finished"); } + + /// The rebuild quiesces background shielded sync, but the host must not + /// have to restart it: a loop that was running before keeps running after. + #[cfg(feature = "shielded")] + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn should_restart_background_shielded_sync_after_rebuild_when_it_was_running() { + let mgr = make_manager(); + let dir = tempfile::tempdir().expect("temp dir"); + mgr.configure_shielded(dir.path().join("tree.sqlite")) + .await + .expect("configure shielded"); + Arc::clone(&mgr.shielded_sync_manager).start(); + assert!(mgr.shielded_sync_manager.is_running()); + + mgr.rebuild_shielded_tree().await.expect("rebuild"); + + assert!( + mgr.shielded_sync_manager.is_running(), + "rebuild must restart the background loop it stopped" + ); + assert!(mgr.shutdown().await.all_clean()); + } + + /// A host that drives shielded sync manually keeps doing so: the rebuild + /// does not start a background loop that was not running. + #[cfg(feature = "shielded")] + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn should_leave_background_shielded_sync_stopped_after_rebuild_when_it_was_not_running() { + let mgr = make_manager(); + let dir = tempfile::tempdir().expect("temp dir"); + mgr.configure_shielded(dir.path().join("tree.sqlite")) + .await + .expect("configure shielded"); + + mgr.rebuild_shielded_tree().await.expect("rebuild"); + + assert!(!mgr.shielded_sync_manager.is_running()); + assert!(mgr.shutdown().await.all_clean()); + } + + /// A host stop issued while a rebuild is in flight wins: the rebuild must + /// not revive the loop the host just stopped. + #[cfg(feature = "shielded")] + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn should_not_restart_background_shielded_sync_stopped_during_rebuild() { + let mgr = make_manager(); + let dir = tempfile::tempdir().expect("temp dir"); + mgr.configure_shielded(dir.path().join("tree.sqlite")) + .await + .expect("configure shielded"); + let coordinator = mgr.shielded_coordinator().await.expect("coordinator"); + Arc::clone(&mgr.shielded_sync_manager).start(); + + // Park the rebuild on the store lock, after it has quiesced the loop. + let store_guard = coordinator.store().write().await; + let rebuild = tokio::spawn({ + let mgr = Arc::clone(&mgr); + async move { mgr.rebuild_shielded_tree().await } + }); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while mgr.shielded_sync_manager.is_running() { + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + }) + .await + .expect("rebuild never quiesced the background loop"); + + mgr.shielded_sync_manager.stop(); + drop(store_guard); + rebuild.await.expect("rebuild task").expect("rebuild"); + + assert!( + !mgr.shielded_sync_manager.is_running(), + "a stop issued during the rebuild must keep background sync stopped" + ); + assert!(mgr.shutdown().await.all_clean()); + } + + /// Fails if the gate keeps barring passes. The loop's own first pass may + /// hold the slot, making `sync_now` skip, so this waits for either to stamp. + #[cfg(feature = "shielded")] + async fn assert_a_shielded_pass_completes(sync: &ShieldedSyncManager) { + tokio::time::timeout(Duration::from_secs(5), async { + while sync.last_sync_unix_seconds().is_none() { + sync.sync_now(true).await; + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("no shielded pass completed: the gate still bars them"); + } + + /// A rebuild whose quiesce times out restarts the loop with admission + /// open: once the wedged pass ends, passes run again, so `is_running()` + /// reporting true is not a loop whose every pass bails. + #[cfg(feature = "shielded")] + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn should_admit_passes_after_restart_when_rebuild_quiesce_times_out() { + let mgr = make_manager(); + let dir = tempfile::tempdir().expect("temp dir"); + mgr.configure_shielded(dir.path().join("tree.sqlite")) + .await + .expect("configure shielded"); + let wedged = mgr.shielded_sync_manager.hold_sync_slot(); + Arc::clone(&mgr.shielded_sync_manager).start(); + + let result = mgr + .rebuild_shielded_tree_within(Duration::from_millis(50)) + .await; + + assert!( + matches!(result, Err(PlatformWalletError::ShutdownIncomplete(_))), + "the wedged pass must time the quiesce out: {result:?}" + ); + assert!(mgr.shielded_sync_manager.is_running()); + drop(wedged); + assert_a_shielded_pass_completes(&mgr.shielded_sync_manager).await; + assert!(mgr.shutdown().await.all_clean()); + } + + /// A failed rebuild restores the loop too, so a retrying host still finds + /// background sync in the state it left it. + #[cfg(feature = "shielded")] + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn should_restart_background_shielded_sync_when_rebuild_fails() { + let mgr = make_manager(); + Arc::clone(&mgr.shielded_sync_manager).start(); + + let result = mgr.rebuild_shielded_tree().await; + + assert!( + matches!(result, Err(PlatformWalletError::ShieldedStoreError(_))), + "no coordinator is configured: {result:?}" + ); + assert!( + mgr.shielded_sync_manager.is_running(), + "a failed rebuild must not leave background sync stopped" + ); + assert!(mgr.shutdown().await.all_clean()); + } + + /// An earlier stop's timed-out drain latched the gate and the host then + /// started the loop again. A rebuild that also times out cannot clear the + /// latch, so it must not restart a loop whose every pass would bail. The + /// retry that drains clears the latch, and only then does `start` run passes. + #[cfg(feature = "shielded")] + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn should_not_restart_background_shielded_sync_into_a_latched_gate() { + let mgr = make_manager(); + let sync = &mgr.shielded_sync_manager; + let wedged = sync.hold_sync_slot(); + assert!(!sync.quiesce_within(Duration::from_millis(50)).await); + Arc::clone(sync).start(); + assert!(sync.is_running(), "the latch does not stop a host start"); + + let result = mgr + .rebuild_shielded_tree_within(Duration::from_millis(50)) + .await; + + assert!( + matches!(result, Err(PlatformWalletError::ShutdownIncomplete(_))), + "the wedged pass must time the quiesce out: {result:?}" + ); + assert!( + !sync.is_running(), + "the latched gate bars every pass, so the loop must not report running" + ); + + drop(wedged); + let retry = mgr + .rebuild_shielded_tree_within(Duration::from_millis(50)) + .await; + assert!( + matches!(retry, Err(PlatformWalletError::ShieldedStoreError(_))), + "the retry drains and then finds no coordinator: {retry:?}" + ); + assert!(!sync.is_running(), "only the host restarts a skipped loop"); + Arc::clone(sync).start(); + assert_a_shielded_pass_completes(sync).await; + assert!(mgr.shutdown().await.all_clean()); + } + + /// The reset took effect but the restart panicked on the blocking pool: + /// the host gets the typed error, not a phantom success, and the loop + /// reports stopped. `start_thread` panics on a current-thread runtime, + /// which makes the panic real rather than injected. + #[cfg(feature = "shielded")] + #[test] + fn should_report_a_typed_error_when_the_restart_panics_after_a_reset() { + let multi_thread = tokio::runtime::Runtime::new().expect("multi-thread runtime"); + let (mgr, _dir) = multi_thread.block_on(async { + let mgr = make_manager(); + let dir = tempfile::tempdir().expect("temp dir"); + mgr.configure_shielded(dir.path().join("tree.sqlite")) + .await + .expect("configure shielded"); + Arc::clone(&mgr.shielded_sync_manager).start(); + (mgr, dir) + }); + let current_thread = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("current-thread runtime"); + + let result = current_thread.block_on(mgr.rebuild_shielded_tree()); + + assert!( + matches!( + result, + Err(PlatformWalletError::ShieldedSyncRestartFailed(_)) + ), + "the restart panicked: {result:?}" + ); + assert!(!mgr.shielded_sync_manager.is_running()); + assert!(multi_thread.block_on(mgr.shutdown()).all_clean()); + } + + /// A sealed gate never reopens, so a rebuild outcome racing shutdown must + /// not restart the loop. + #[cfg(feature = "shielded")] + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn should_not_restart_background_shielded_sync_into_a_sealed_gate() { + let mgr = make_manager(); + let sync = &mgr.shielded_sync_manager; + let stop_intents = sync.stop_intents(); + assert!(sync.quiesce_sealed_within(Duration::from_secs(1)).await); + + Arc::clone(sync).restart_unless_stopped_since(stop_intents); + + assert!(!sync.is_running()); + assert!(mgr.shutdown().await.all_clean()); + } + + /// Without serialization, a second rebuild reads the loop as stopped + /// because the first has quiesced it, and then its clearing latch refuses + /// the first's restart: both return `Ok` with background sync stopped. + #[cfg(feature = "shielded")] + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn should_keep_background_shielded_sync_running_across_concurrent_rebuilds() { + let mgr = make_manager(); + let dir = tempfile::tempdir().expect("temp dir"); + mgr.configure_shielded(dir.path().join("tree.sqlite")) + .await + .expect("configure shielded"); + let coordinator = mgr.shielded_coordinator().await.expect("coordinator"); + Arc::clone(&mgr.shielded_sync_manager).start(); + + // Park the first rebuild on the store lock, after it has quiesced the loop. + let store_guard = coordinator.store().write().await; + let first = tokio::spawn({ + let mgr = Arc::clone(&mgr); + async move { mgr.rebuild_shielded_tree().await } + }); + tokio::time::timeout(Duration::from_secs(5), async { + while mgr.shielded_sync_manager.is_running() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("the first rebuild never quiesced the background loop"); + let second = tokio::spawn({ + let mgr = Arc::clone(&mgr); + async move { mgr.rebuild_shielded_tree().await } + }); + // Lets the second rebuild enter; a serialized one parks at its entry. + tokio::time::sleep(Duration::from_millis(100)).await; + + drop(store_guard); + first.await.expect("first task").expect("first rebuild"); + second.await.expect("second task").expect("second rebuild"); + + assert!( + mgr.shielded_sync_manager.is_running(), + "both rebuilds returned Ok, so the loop that was running must still run" + ); + assert!(mgr.shutdown().await.all_clean()); + } + + /// Parks the first shielded pass in its completion callback, on the loop + /// thread, so that thread cannot exit until [`Self::release`] is called. + #[cfg(feature = "shielded")] + #[derive(Default)] + struct WedgedCompletion { + entered: std::sync::atomic::AtomicBool, + released: std::sync::atomic::AtomicBool, + } + + #[cfg(feature = "shielded")] + impl WedgedCompletion { + fn release(&self) { + self.released + .store(true, std::sync::atomic::Ordering::SeqCst); + } + } + + #[cfg(feature = "shielded")] + impl EventHandler for WedgedCompletion {} + + #[cfg(feature = "shielded")] + impl PlatformEventHandler for WedgedCompletion { + fn on_shielded_sync_completed( + &self, + _summary: &crate::manager::shielded_sync::ShieldedSyncPassSummary, + ) { + self.entered + .store(true, std::sync::atomic::Ordering::SeqCst); + while !self.released.load(std::sync::atomic::Ordering::SeqCst) { + std::thread::sleep(Duration::from_millis(1)); + } + } + } + + /// The restart reaps the prior loop thread, spinning for up to the registry + /// backstop while that thread is still finishing. That must not occupy an + /// async worker. With one worker, a task can observe the restarted loop + /// before the rebuild completes only if the reap left the worker free: a + /// reap on the worker runs through to the rebuild's completion in one poll. + #[cfg(feature = "shielded")] + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + async fn should_not_block_an_async_worker_while_the_restart_reaps_the_prior_loop() { + use std::sync::atomic::{AtomicBool, Ordering}; + + let wedge = Arc::new(WedgedCompletion::default()); + let sdk = Arc::new(dash_sdk::SdkBuilder::new_mock().build().expect("mock sdk")); + let mgr = Arc::new(PlatformWalletManager::new( + sdk, + Arc::new(NoopPersister), + Arc::clone(&wedge) as Arc, + )); + Arc::clone(&mgr.shielded_sync_manager).start(); + tokio::time::timeout(Duration::from_secs(5), async { + while !wedge.entered.load(Ordering::SeqCst) { + tokio::time::sleep(Duration::from_millis(1)).await; + } + }) + .await + .expect("the loop's first pass never reached its completion callback"); + + let rebuild_done = Arc::new(AtomicBool::new(false)); + let observer = tokio::spawn({ + let mgr = Arc::clone(&mgr); + let wedge = Arc::clone(&wedge); + let rebuild_done = Arc::clone(&rebuild_done); + async move { + let mut saw_stopped = false; + loop { + let done = rebuild_done.load(Ordering::SeqCst); + let running = mgr.shielded_sync_manager.is_running(); + saw_stopped |= !running; + if done || (saw_stopped && running) { + wedge.release(); + return done; + } + tokio::time::sleep(Duration::from_millis(1)).await; + } + } + }); + let rebuild = tokio::spawn({ + let mgr = Arc::clone(&mgr); + let rebuild_done = Arc::clone(&rebuild_done); + async move { + let result = mgr + .rebuild_shielded_tree_within(Duration::from_millis(50)) + .await; + rebuild_done.store(true, Ordering::SeqCst); + result + } + }); + + let done_before_observed = observer.await.expect("observer task"); + let result = rebuild.await.expect("rebuild task"); + + assert!( + !done_before_observed, + "the restart's reap occupied the only async worker for its whole backstop" + ); + assert!( + matches!(result, Err(PlatformWalletError::ShutdownIncomplete(_))), + "the wedged pass must time the quiesce out: {result:?}" + ); + assert!(mgr.shielded_sync_manager.is_running()); + assert!(mgr.shutdown().await.all_clean()); + } } diff --git a/packages/rs-platform-wallet/src/manager/shielded_sync.rs b/packages/rs-platform-wallet/src/manager/shielded_sync.rs index dac831cb809..402aa555821 100644 --- a/packages/rs-platform-wallet/src/manager/shielded_sync.rs +++ b/packages/rs-platform-wallet/src/manager/shielded_sync.rs @@ -28,18 +28,20 @@ use std::collections::BTreeMap; use std::sync::{ atomic::{AtomicBool, AtomicU64, Ordering}, - Arc, + Arc, Mutex, MutexGuard, PoisonError, }; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::sync::RwLock; +use tokio::task::JoinError; use dash_async::ThreadRegistry; +use crate::error::PlatformWalletError; use crate::events::PlatformEventManager; use crate::manager::{ - coordinator_worker_config, drain_pass, QuiesceGate, QuiesceGuard, SyncSlotGuard, WalletWorker, - COORDINATOR_DRAIN_BUDGET, + coordinator_worker_config, drain_pass, drain_pass_with, DrainTimeout, QuiesceGate, + QuiesceGuard, SyncSlotGuard, WalletWorker, COORDINATOR_DRAIN_BUDGET, }; use crate::wallet::platform_wallet::WalletId; use crate::wallet::shielded::{NetworkShieldedCoordinator, ShieldedSyncSummary}; @@ -65,6 +67,16 @@ pub enum WalletShieldedOutcome { Skipped, /// Error message from a failed sync. Err(String), + /// The pass found the local commitment tree inconsistent with the chain + /// (see [`PlatformWalletError::ShieldedTreeInconsistent`]). Resyncing + /// cannot repair it; call + /// [`PlatformWalletManager::rebuild_shielded_tree`](crate::PlatformWalletManager::rebuild_shielded_tree). + TreeInconsistent { + /// Number of leaves in the local tree after consuming the stream. + local_leaf_count: u64, + /// Largest commitment count reported by the stream's proofs. + chain_leaf_count: u64, + }, } impl WalletShieldedOutcome { @@ -113,6 +125,19 @@ impl ShieldedSyncPassSummary { } } +/// Typed outcome of the blocking-pool task behind +/// [`ShieldedSyncManager::restart_off_worker_unless_stopped_since`]. +fn restart_outcome(joined: Result<(), JoinError>) -> Result<(), PlatformWalletError> { + joined.map_err(|join_err| { + let cause = if join_err.is_panic() { + "the restart task panicked" + } else { + "the restart task was cancelled" + }; + PlatformWalletError::ShieldedSyncRestartFailed(cause.to_string()) + }) +} + /// Periodic shielded sync coordinator. /// /// Holds a handle to the same `coordinator_slot` owned by @@ -160,6 +185,11 @@ pub struct ShieldedSyncManager { /// giving Clear / stop a real "no more host-visible mutations" /// barrier that cancel-only [`stop`](Self::stop) does not provide. quiescing: QuiesceGate, + /// Count of stop requests ([`stop`](Self::stop), [`quiesce`](Self::quiesce), + /// Clear). A pausing operation restarts the loop only if this is unchanged, + /// so a stop issued meanwhile wins. The lock also spans that check and the + /// restart, so no stop can slip between them. + stop_intents: Mutex, /// Unix seconds of the last completed pass. `0` = never. last_sync_unix: AtomicU64, } @@ -177,6 +207,7 @@ impl ShieldedSyncManager { interval_secs: AtomicU64::new(DEFAULT_SYNC_INTERVAL_SECS), is_syncing: AtomicBool::new(false), quiescing: QuiesceGate::default(), + stop_intents: Mutex::new(0), last_sync_unix: AtomicU64::new(0), } } @@ -194,7 +225,8 @@ impl ShieldedSyncManager { Duration::from_secs(self.interval_secs.load(Ordering::Acquire)) } - /// Whether the background loop is currently running. + /// Whether the background loop is currently running. A running loop can + /// still be doing no work: see [`start`](Self::start) on a latched gate. pub fn is_running(&self) -> bool { self.registry.is_running(WalletWorker::ShieldedSync) } @@ -225,6 +257,10 @@ impl ShieldedSyncManager { /// reaps a still-draining prior-generation thread, spinning up to the /// registry reap backstop (default 1 s) before returning. Call it from /// the FFI host thread, not an async task. + /// + /// Does not consult the [quiesce gate](QuiesceGate): after a drain timed + /// out and latched it, this still starts a loop that reports running while + /// every pass bails, until a later drain clears the latch. pub fn start(self: Arc) { let handle = tokio::runtime::Handle::current(); let registry = Arc::clone(&self.registry); @@ -274,9 +310,67 @@ impl ShieldedSyncManager { /// nothing more will be persisted" barrier — required by Clear, /// unregister, and rebind — use [`quiesce`](Self::quiesce). pub fn stop(&self) { + self.record_stop_intent(); + self.cancel_loop(); + } + + fn cancel_loop(&self) { self.registry.cancel(WalletWorker::ShieldedSync); } + fn lock_stop_intents(&self) -> MutexGuard<'_, u64> { + self.stop_intents + .lock() + .unwrap_or_else(PoisonError::into_inner) + } + + /// Record that the host wants background sync stopped, so a pausing + /// operation in flight does not restart it. + pub(crate) fn record_stop_intent(&self) { + let mut intents = self.lock_stop_intents(); + *intents = intents.wrapping_add(1); + } + + /// Current stop-request count, for [`restart_unless_stopped_since`](Self::restart_unless_stopped_since). + pub(crate) fn stop_intents(&self) -> u64 { + *self.lock_stop_intents() + } + + /// [`start`](Self::start) the loop unless a stop was requested after + /// `stop_intents` was read, or the gate is stuck closed: a loop started + /// then would report running while every pass bails. Blocks briefly like + /// `start`. + /// + /// Lock order: `stop_intents`, then the gate's bookkeeping. Nothing takes + /// them the other way round. A latching drain that begins after the check + /// cancels the loop this starts. + pub(crate) fn restart_unless_stopped_since(self: &Arc, stop_intents: u64) { + let intents = self.lock_stop_intents(); + if *intents != stop_intents { + return; + } + if self.quiescing.is_stuck_closed() { + tracing::warn!( + "shielded sync gate is latched or sealed; leaving the background loop stopped" + ); + return; + } + Arc::clone(self).start(); + } + + /// [`restart_unless_stopped_since`](Self::restart_unless_stopped_since) on + /// the blocking pool, because `start` can spin for the registry's reap + /// backstop and must not occupy an async worker. + pub(crate) async fn restart_off_worker_unless_stopped_since( + self: &Arc, + stop_intents: u64, + ) -> Result<(), PlatformWalletError> { + let this = Arc::clone(self); + let restart = + tokio::task::spawn_blocking(move || this.restart_unless_stopped_since(stop_intents)); + restart_outcome(restart.await) + } + /// Cancel the background loop **and wait for any in-flight sync pass /// to fully drain** before returning — a real quiescence barrier, /// unlike cancel-only [`stop`](Self::stop). @@ -302,6 +396,7 @@ impl ShieldedSyncManager { /// `quiesce_within` for the timeout contract. #[must_use = "a false return means the pass did NOT drain; the caller must fail closed"] pub async fn quiesce(&self) -> bool { + self.record_stop_intent(); self.quiesce_within(COORDINATOR_DRAIN_BUDGET).await } @@ -339,7 +434,42 @@ impl ShieldedSyncManager { /// [`quiesce_held`](Self::quiesce_held) with an explicit drain budget. pub(crate) async fn quiesce_held_within(&self, budget: Duration) -> Option> { - drain_pass(&self.quiescing, &self.is_syncing, || self.stop(), budget).await + drain_pass( + &self.quiescing, + &self.is_syncing, + || self.cancel_loop(), + budget, + ) + .await + } + + /// [`quiesce_held_within`](Self::quiesce_held_within) whose timeout + /// leaves no latch ([`DrainTimeout::Release`]), for a caller that changes + /// nothing on timeout and then resumes background sync. + pub(crate) async fn quiesce_held_releasing_within( + &self, + budget: Duration, + ) -> Option> { + drain_pass_with( + &self.quiescing, + &self.is_syncing, + || self.cancel_loop(), + budget, + DrainTimeout::Release, + ) + .await + } + + /// Occupy the pass slot as a wedged pass would, until the guard drops. + #[cfg(test)] + pub(crate) fn hold_sync_slot(&self) -> SyncSlotGuard<'_> { + assert!( + self.is_syncing + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_ok(), + "a pass already holds the slot" + ); + SyncSlotGuard(&self.is_syncing) } /// [`quiesce_within`](Self::quiesce_within) that **seals** the gate: @@ -485,7 +615,15 @@ impl ShieldedSyncManager { return Ok(None); } - let pass = coordinator.sync(force).await; + if !coordinator + .registered_subwallets() + .await + .iter() + .any(|id| &id.wallet_id == wallet_id) + { + return Ok(None); + } + let pass = coordinator.try_sync(force).await?; // Extract this wallet's slice from the network-wide pass // summary. If the wallet is registered, we'll get back an @@ -498,8 +636,18 @@ impl ShieldedSyncManager { Some((_, WalletShieldedOutcome::Ok(summary))) => Ok(Some(summary)), Some((_, WalletShieldedOutcome::Skipped)) => Ok(None), Some((_, WalletShieldedOutcome::Err(e))) => { - Err(crate::error::PlatformWalletError::ShieldedSyncFailed(e)) + Err(PlatformWalletError::ShieldedSyncFailed(e)) } + Some(( + _, + WalletShieldedOutcome::TreeInconsistent { + local_leaf_count, + chain_leaf_count, + }, + )) => Err(PlatformWalletError::ShieldedTreeInconsistent { + local_leaf_count, + chain_leaf_count, + }), None => Ok(None), } } @@ -515,3 +663,46 @@ impl std::fmt::Debug for ShieldedSyncManager { .finish() } } + +#[cfg(test)] +mod restart_outcome_tests { + use super::*; + + #[tokio::test] + async fn should_report_a_typed_error_when_the_restart_task_panics() { + let joined = tokio::task::spawn_blocking(|| panic!("restart")).await; + + let outcome = restart_outcome(joined); + + assert!( + matches!( + &outcome, + Err(PlatformWalletError::ShieldedSyncRestartFailed(cause)) if cause.contains("panicked") + ), + "{outcome:?}" + ); + } + + #[tokio::test] + async fn should_report_a_typed_error_when_the_restart_task_is_cancelled() { + let task = tokio::spawn(std::future::pending::<()>()); + task.abort(); + + let outcome = restart_outcome(task.await); + + assert!( + matches!( + &outcome, + Err(PlatformWalletError::ShieldedSyncRestartFailed(cause)) if cause.contains("cancelled") + ), + "{outcome:?}" + ); + } + + #[tokio::test] + async fn should_report_success_when_the_restart_task_completes() { + let joined = tokio::task::spawn_blocking(|| {}).await; + + assert!(restart_outcome(joined).is_ok()); + } +} diff --git a/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs b/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs index 731b35e51ad..0134dfd629d 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/coordinator.rs @@ -374,6 +374,12 @@ impl ShieldedInstall<'_> { } } +/// A failed sync pass, with the subwallet snapshot it ran for. +struct PassFailure { + subwallets: Vec<(SubwalletId, AccountViewingKeys)>, + error: PlatformWalletError, +} + impl NetworkShieldedCoordinator { /// Build a new coordinator. Called by /// `PlatformWalletManager::configure_shielded` (Phase 1) on @@ -1186,7 +1192,10 @@ impl NetworkShieldedCoordinator { /// SwiftData rows) after `clear()` succeeds. If a reset fails /// silently the host could drop its rows while the shared tree /// stays populated, and the next cold resync would gate-skip every - /// re-downloaded position against the stale `tree_size`. + /// re-downloaded position against the stale `tree_size`. An error does not + /// prove the tree is unchanged: a reset whose post-commit durability + /// checkpoint failed or was busy has taken effect and the tree is empty. + /// Keep the rows and retry; a retry just resets again. pub async fn clear(&self) -> Result<(), crate::error::PlatformWalletError> { // Held across the whole Clear, store reset included, so no bind // install can be halfway through its transaction here: one that @@ -1221,9 +1230,9 @@ impl NetworkShieldedCoordinator { } // Verify the reset actually landed to 0 leaves under the SAME // write guard. `reset_commitment_tree` can theoretically report - // Ok while leaving the on-disk tree populated (e.g. a checkpoint - // that didn't take, or a shardtree reopen that re-materialized a - // cached frontier). Without this check `clear()` would return Ok, + // Ok while leaving the on-disk tree populated (e.g. a shardtree + // reopen that re-materialized a cached frontier). Without this + // check `clear()` would return Ok, // the host would wipe its own Room/SwiftData rows, and the next // cold resync would gate-skip every re-downloaded position against // the still-full tree — the exact "Clear did nothing, tree frozen @@ -1291,6 +1300,42 @@ impl NetworkShieldedCoordinator { Ok(()) } + /// Empty only the commitment tree and rewind all subwallet scan cursors. + /// Notes, nullifiers, spent status, pending spends, and registrations survive. + /// The next sync performs a full rescan from zero, costing bandwidth and + /// trial decryption proportional to the entire pool. Never called automatically. + /// The reset itself is atomic: if it rolls back, the tree stays as it was, + /// and so do the cursors unless the tree was already empty, when they are + /// rewound. A durability checkpoint that fails or is busy after COMMIT also + /// returns an error, but the reset has then taken effect and the cursors + /// are rewound; a retry just resets again. + pub async fn rebuild_shielded_tree(&self) -> Result<(), PlatformWalletError> { + let _lifecycle = self.lifecycle.lock().await; + let mut store = self.store.write().await; + // The reset is atomic, so the cursors follow the live tree: rewound + // whenever it is empty (including a committed reset whose durability + // checkpoint failed, and a rollback on a tree that was already empty, + // which is harmless), untouched when a rollback leaves leaves behind, + // so the next pass reports the same inconsistency again. + let reset = store.reset_commitment_tree(); + let remaining = store.tree_size(); + if matches!(remaining, Ok(0)) { + store.rewind_scan_watermarks(); + } + // Either way the next pass runs at once: a rescan, or a fresh verdict. + if let Ok(mut caught_up) = self.last_caught_up_at.lock() { + *caught_up = None; + } + reset?; + let remaining = remaining?; + if remaining != 0 { + return Err(PlatformWalletError::ShieldedTreeUpdateFailed(format!( + "commitment tree still has {remaining} leaves after rebuild reset" + ))); + } + Ok(()) + } + /// Run one shielded sync pass for every registered wallet on /// this coordinator's network. Returns a per-wallet outcome /// summary suitable for emission to UI / persistence layers @@ -1315,6 +1360,26 @@ impl NetworkShieldedCoordinator { /// [`PlatformEventManager::on_shielded_sync_completed`]: /// crate::events::PlatformEventManager::on_shielded_sync_completed pub async fn sync(&self, force: bool) -> ShieldedSyncPassSummary { + match self.run_pass(force).await { + Ok(summary) => summary, + // Attribute the failure to the wallets the pass ran for, not to + // whatever the registry holds after a concurrent (un)register. + Err(PassFailure { subwallets, error }) => self.fail_all_wallets(&subwallets, &error), + } + } + + /// Run the same pass as [`Self::sync`], preserving typed failures for callers + /// that need to distinguish an inconsistent tree from a transient sync error. + pub async fn try_sync( + &self, + force: bool, + ) -> Result { + self.run_pass(force).await.map_err(|failure| failure.error) + } + + /// Body of [`Self::sync`] / [`Self::try_sync`]; a failure carries the + /// pass's own subwallet snapshot. + async fn run_pass(&self, force: bool) -> Result { // Network-wide cooldown gate. Snapshot the remaining // window into a local before any await — `std::sync::Mutex` // is `!Send` across await points (clippy's @@ -1336,7 +1401,7 @@ impl NetworkShieldedCoordinator { cooldown_total_secs = CAUGHT_UP_COOLDOWN.as_secs(), "Coordinator sync skipped — within caught-up cooldown" ); - return Self::cooldown_skip_summary(self).await; + return Ok(Self::cooldown_skip_summary(self).await); } // Snapshot the flat subwallet registry. This Vec is both @@ -1347,12 +1412,23 @@ impl NetworkShieldedCoordinator { accounts.iter().map(|(id, v)| (*id, v.clone())).collect() }; - let mut summary = ShieldedSyncPassSummary::default(); if subwallets.is_empty() { - summary.sync_unix_seconds = Self::now_unix(); - return summary; + return Ok(ShieldedSyncPassSummary { + sync_unix_seconds: Self::now_unix(), + ..Default::default() + }); } + self.sync_subwallets(&subwallets) + .await + .map_err(|error| PassFailure { subwallets, error }) + } + /// Scan, reconcile, and persist one pass over a non-empty `subwallets` + /// snapshot. + async fn sync_subwallets( + &self, + subwallets: &[(SubwalletId, AccountViewingKeys)], + ) -> Result { // 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 @@ -1366,7 +1442,7 @@ impl NetworkShieldedCoordinator { // 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; + 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 @@ -1378,18 +1454,14 @@ impl NetworkShieldedCoordinator { // Second, distinct signal: commitments committed to the local // tree as the interleaved consumer drains the SDK stream. let on_tree_progress = self.tree_progress_handler(); - let notes = match super::sync::sync_notes_across( + let notes = super::sync::sync_notes_across( &self.sdk, &self.store, - &subwallets, + subwallets, on_progress.as_ref(), on_tree_progress.as_ref(), ) - .await - { - Ok(r) => r, - Err(e) => return self.fail_all_wallets(&subwallets, &e), - }; + .await?; // Scan-based spend detection now happens INSIDE // `sync_notes_across`: every scanned action's nullifier is // replayed against each subwallet's store as part of the note @@ -1441,10 +1513,7 @@ impl NetworkShieldedCoordinator { 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), - }; + let balances_per_sub = super::sync::balances_across(&self.store, subwallets).await?; // Restore-path activity derivation: reconstruct per-operation // activity entries best-effort from the notes / outgoing notes @@ -1456,7 +1525,7 @@ impl NetworkShieldedCoordinator { // swallowed: a derivation miss must never fail a sync pass. let mut notes = notes; if let Err(e) = self - .derive_activity_into_changeset(&subwallets, &mut notes.changeset) + .derive_activity_into_changeset(subwallets, &mut notes.changeset) .await { tracing::warn!(error = %e, "Shielded activity derivation failed; skipping this pass"); @@ -1521,16 +1590,16 @@ impl NetworkShieldedCoordinator { // `ShieldedSyncSummary` shape that // `PlatformEventManager::on_shielded_sync_completed` // already speaks. Same emission shape as Phase 2a. - summary = - build_per_wallet_summary(&subwallets, ¬es, &newly_spent_per_sub, &balances_per_sub); + let mut summary = + build_per_wallet_summary(subwallets, ¬es, &newly_spent_per_sub, &balances_per_sub); summary.sync_unix_seconds = Self::now_unix(); - summary + Ok(summary) } /// Build a `ShieldedSyncPassSummary` where every registered - /// wallet's outcome is the supplied error string. Used when - /// the network-wide SDK note scan (sync_notes_across) errors - /// before any per-wallet result can be produced. + /// wallet's outcome is the supplied error: typed for an + /// inconsistent tree, its message otherwise. Used when the + /// network-wide pass errors before any per-wallet result exists. fn fail_all_wallets( &self, subwallets: &[(SubwalletId, AccountViewingKeys)], @@ -1546,9 +1615,17 @@ impl NetworkShieldedCoordinator { error = %e, "Network-wide shielded sync failed" ); - summary - .wallet_results - .insert(wallet_id, WalletShieldedOutcome::Err(e.to_string())); + let outcome = match *e { + PlatformWalletError::ShieldedTreeInconsistent { + local_leaf_count, + chain_leaf_count, + } => WalletShieldedOutcome::TreeInconsistent { + local_leaf_count, + chain_leaf_count, + }, + _ => WalletShieldedOutcome::Err(e.to_string()), + }; + summary.wallet_results.insert(wallet_id, outcome); } summary.sync_unix_seconds = Self::now_unix(); summary 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 0970e75d48a..09c31095385 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/file_store.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/file_store.rs @@ -15,7 +15,7 @@ use std::collections::BTreeMap; use std::error::Error as StdError; use std::fmt; use std::path::{Path, PathBuf}; -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; use dpp::serialization::{PlatformDeserializableTrusted, PlatformSerializable}; use dpp::state_transition::StateTransition; @@ -148,6 +148,20 @@ impl FileBackedShieldedStore { } } + /// Rewind all scan cursors without discarding receipts or spend guards. + /// Scan coverage is dropped with the tree; retained notes count as restored. + pub(super) fn rewind_scan_watermarks(&mut self) { + for state in self.subwallets.values_mut() { + state.last_synced_index = 0; + state.last_scanned_index = None; + state.balance_source = if state.notes.is_empty() { + ShieldedBalanceSource::NoHistory + } else { + ShieldedBalanceSource::Restored + }; + } + } + /// Open or create a shielded store at `path`. /// /// SQLite is opened with **WAL journal + synchronous=NORMAL + temp_store=MEMORY** @@ -194,6 +208,19 @@ impl FileBackedShieldedStore { .map_err(|e| { FileShieldedStoreError::Storage(format!("create pending_spends table: {e}")) })?; + // The bound shares the tree's database/WAL: committing it before an + // append prevents a restart from pairing new leaves with an old bound. + pending_conn + .execute( + "CREATE TABLE IF NOT EXISTS shielded_tree_sync_height ( + id INTEGER PRIMARY KEY CHECK (id = 0), + height INTEGER NOT NULL CHECK (height >= 0) + )", + [], + ) + .map_err(|e| { + FileShieldedStoreError::Storage(format!("create tree sync height: {e}")) + })?; // Additive, idempotent migration: old guards remain active. let columns = { let mut statement = pending_conn @@ -542,6 +569,17 @@ impl ShieldedStore for FileBackedShieldedStore { .unwrap_or_default()) } + fn note_spent_status( + &self, + id: SubwalletId, + nullifier: &[u8; 32], + ) -> Result, Self::Error> { + Ok(self + .subwallets + .get(&id) + .and_then(|state| state.note_spent_status(nullifier))) + } + fn mark_spent(&mut self, id: SubwalletId, nullifier: &[u8; 32]) -> Result { let Some(sw) = self.subwallets.get_mut(&id) else { return Ok(false); @@ -857,6 +895,43 @@ impl ShieldedStore for FileBackedShieldedStore { Ok(size) } + fn tree_sync_height(&self) -> Result, Self::Error> { + let conn = self + .pending_conn + .lock() + .map_err(|e| FileShieldedStoreError::Storage(format!("tree sync height mutex: {e}")))?; + let height: Option = conn + .query_row( + "SELECT MAX(height) FROM shielded_tree_sync_height", + [], + |row| row.get(0), + ) + .map_err(|e| FileShieldedStoreError::Storage(format!("read tree sync height: {e}")))?; + height + .map(u64::try_from) + .transpose() + .map_err(|e| FileShieldedStoreError::Storage(format!("invalid tree sync height: {e}"))) + } + + fn set_tree_sync_height(&mut self, height: u64) -> Result<(), Self::Error> { + let height = i64::try_from(height).map_err(|e| { + FileShieldedStoreError::Storage(format!( + "tree sync height exceeds SQLite integer range: {e}" + )) + })?; + let conn = self + .pending_conn + .lock() + .map_err(|e| FileShieldedStoreError::Storage(format!("tree sync height mutex: {e}")))?; + conn.execute( + "INSERT INTO shielded_tree_sync_height (id, height) VALUES (0, ?1) + ON CONFLICT(id) DO UPDATE SET height = MAX(height, excluded.height)", + [height], + ) + .map_err(|e| FileShieldedStoreError::Storage(format!("write tree sync height: {e}")))?; + Ok(()) + } + fn last_synced_note_index(&self, id: SubwalletId) -> Result { Ok(self .subwallets @@ -953,60 +1028,113 @@ impl ShieldedStore for FileBackedShieldedStore { // The `ClientPersistentCommitmentTree` wrapper owns its // `Connection` and exposes no public truncate (only the inner // `SqliteShardStore` has `truncate_shards`). A full reset - // therefore (1) wipes the four `commitment_tree_*` tables on a - // fresh connection, then (2) rebuilds the wrapper over the now - // empty DB so the in-memory shardtree frontier/cap reflect the - // empty state. Reopening — rather than mutating the live tree — - // is what guarantees `tree_size()` reads back 0: the wrapper - // caches frontier nodes that a bare `DELETE` wouldn't clear. + // therefore wipes the tree tables on a fresh connection and + // rebuilds the wrapper over the now empty DB, so the in-memory + // shardtree frontier/cap reflect the empty state. Reopening — + // rather than mutating the live tree — is what guarantees + // `tree_size()` reads back 0: the wrapper caches frontier nodes + // that a bare `DELETE` wouldn't clear. + // + // The reset is atomic: the deletes and the replacement wrapper are + // staged in ONE transaction and the live wrapper is swapped only + // after it commits. Any failure before that rolls back, leaving the + // disk and the live tree exactly as they were — a half-emptied tree + // under a stale frontier would let the next sync append onto + // leaves the disk no longer holds. The replacement opens on the + // staging connection (shared mode) so it reads the uncommitted + // empty state; this function drops its own handle before + // returning, so the wrapper is that connection's only user. let mut tree = self .tree .lock() .map_err(|e| FileShieldedStoreError::Storage(format!("tree mutex poisoned: {e}")))?; + let conn = Arc::new(Mutex::new(Self::open_tuned_connection(&self.path)?)); + // Locks only per statement: the wrapper takes the same mutex. + let run = |sql: &str, what: &str| -> Result<(), FileShieldedStoreError> { + conn.lock() + .map_err(|e| { + FileShieldedStoreError::Storage(format!("reset connection poisoned: {e}")) + })? + .execute_batch(sql) + .map_err(|e| FileShieldedStoreError::Storage(format!("{what}: {e}"))) + }; - { - let conn = Self::open_tuned_connection(&self.path)?; - // `commitment_tree_cap` is included alongside the three - // shard/checkpoint tables: it caches upper-level tree nodes, - // so leaving it populated while the shards are empty would - // reopen into an inconsistent (non-empty) tree state. - conn.execute_batch( - "DELETE FROM commitment_tree_checkpoint_marks_removed; - DELETE FROM commitment_tree_checkpoints; - DELETE FROM commitment_tree_shards; - DELETE FROM commitment_tree_cap;", + run("BEGIN IMMEDIATE;", "begin commitment tree reset")?; + // `commitment_tree_cap` is included alongside the three + // shard/checkpoint tables: it caches upper-level tree nodes, + // so leaving it populated while the shards are empty would + // reopen into an inconsistent (non-empty) tree state. + let staged = run( + "DELETE FROM commitment_tree_checkpoint_marks_removed; + DELETE FROM commitment_tree_checkpoints; + DELETE FROM commitment_tree_shards; + DELETE FROM commitment_tree_cap; + DELETE FROM shielded_tree_sync_height;", + "reset commitment tree tables", + ) + .and_then(|()| { + ClientPersistentCommitmentTree::open_on_shared_connection( + Arc::clone(&conn), + self.max_checkpoints, ) + .map_err(|e| FileShieldedStoreError::Storage(format!("reopen commitment tree: {e}"))) + }) + .and_then(|replacement| { + run("COMMIT;", "commit commitment tree reset")?; + Ok(replacement) + }); + let replacement = match staged { + Ok(replacement) => replacement, + Err(e) => { + // A failed COMMIT may already have rolled back on its own. + if let Err(rollback) = run("ROLLBACK;", "roll back commitment tree reset") { + tracing::debug!(error = %rollback, "commitment tree reset rollback"); + } + return Err(e); + } + }; + *tree = replacement; + + // Durably flush the DELETEs into the main database file with a + // TRUNCATE checkpoint. This runs after COMMIT, outside the atomic + // reset: a failed or busy checkpoint leaves the reset committed and + // the live tree consistent with it, so the caller sees an error for a + // reset that took effect. Only durability against a power loss or OS + // crash is at stake, and a retry just resets again. + // + // The store runs `synchronous=NORMAL` in WAL mode (see + // `open_tuned_connection`), so a committed transaction lands in the + // `-wal` file but is NOT fsync'd until a checkpoint. Other connections + // (`pending_conn`) stay open on the same file, so SQLite's + // last-connection-close auto-checkpoint never fires — the emptied + // tables live only in the WAL. A process kill keeps those frames, + // because the OS still holds them and the next open recovers them; a + // power loss or OS crash can lose them and bring back the OLD full + // tree. A completed TRUNCATE checkpoint rewrites and syncs the main db + // and resets the WAL, closing that window. SQLite reports one blocked + // by another connection as a `busy=1` row, not as an error, so that + // row must be read; `execute_batch` would discard it. + // + // This checkpoint was added for a "Clear did nothing" report (tree + // frozen at 771/771 after a Clear and an Android force-stop). A kill + // alone does not lose committed frames, so that report's cause is not + // established here. + let busy: i64 = conn + .lock() + .map_err(|e| { + FileShieldedStoreError::Storage(format!("reset connection poisoned: {e}")) + })? + .query_row("PRAGMA wal_checkpoint(TRUNCATE);", [], |row| row.get(0)) .map_err(|e| { - FileShieldedStoreError::Storage(format!("reset commitment tree tables: {e}")) + FileShieldedStoreError::Storage(format!("checkpoint after tree reset: {e}")) })?; - - // Durably flush the DELETEs into the main database file with a - // TRUNCATE checkpoint before this connection is dropped. - // - // Without this, Clear is a no-op across a hard kill. The store - // runs `synchronous=NORMAL` in WAL mode (see `open_tuned_connection`), - // so a committed transaction lands in the `-wal` file but is NOT - // fsync'd until a checkpoint. Two other connections (`tree` and - // `pending_conn`) stay open on the same file, so SQLite's - // last-connection-close auto-checkpoint never fires when this - // transient connection drops — the emptied tables live only in the - // WAL. On Android the "Clear" button is routinely followed by a - // force-stop (non-graceful SIGKILL, no checkpoint), so the WAL - // frames are discarded and the next launch reopens the OLD full - // tree (the 771/771 "Clear did nothing" symptom). A TRUNCATE - // checkpoint rewrites the main db and resets the WAL, making the - // emptied state durable regardless of how the process later dies. - // The prior graceful `drop(store)` in the unit test masked this — - // a clean close checkpoints, a SIGKILL does not. - conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);") - .map_err(|e| { - FileShieldedStoreError::Storage(format!("checkpoint after tree reset: {e}")) - })?; + if busy != 0 { + return Err(FileShieldedStoreError::Storage( + "checkpoint after tree reset was busy: the reset is committed but may not \ + survive a power loss or OS crash; retry to reset again" + .to_string(), + )); } - - let conn = Self::open_tuned_connection(&self.path)?; - *tree = ClientPersistentCommitmentTree::open(conn, self.max_checkpoints) - .map_err(|e| FileShieldedStoreError::Storage(format!("reopen commitment tree: {e}")))?; Ok(()) } } @@ -1814,27 +1942,56 @@ mod tests { ); } - /// Durability regression guard for the "Clear did nothing after a - /// force-stop" bug: `reset_commitment_tree` must land the emptied - /// tables in the MAIN database file, not merely in the `-wal` file of - /// the store's own connections. + /// SQLite reports a checkpoint blocked by a reader as a `busy=1` row, not + /// as an error, so a reset must read that row: otherwise it reports success + /// with the emptied state still only in the WAL. Waits out the store's + /// 5 s busy timeout. + #[test] + fn reset_commitment_tree_reports_a_busy_checkpoint_and_stays_applied() { + let path = temp_tree_path("reset_busy_checkpoint"); + let mut store = FileBackedShieldedStore::open_path(&path, 100).expect("open"); + store.append_commitment(&[1u8; 32], false).expect("append"); + store.checkpoint_tree(1).expect("checkpoint"); + + let reader = rusqlite::Connection::open(&path).expect("reader"); + reader.execute_batch("BEGIN;").expect("begin read"); + let _: i64 = reader + .query_row("SELECT count(*) FROM sqlite_master", [], |row| row.get(0)) + .expect("hold a read snapshot"); + + let result = store.reset_commitment_tree(); + + assert!( + matches!(&result, Err(FileShieldedStoreError::Storage(reason)) if reason.contains("busy")), + "{result:?}" + ); + assert_eq!(store.tree_size().expect("tree size"), 0, "COMMIT stands"); + + reader.execute_batch("COMMIT;").expect("end read"); + store + .reset_commitment_tree() + .expect("the retry resets again"); + drop(store); + std::fs::remove_file(&path).ok(); + } + + /// Durability regression guard: `reset_commitment_tree` must land the + /// emptied tables in the MAIN database file, not merely in the `-wal` file + /// of the store's own connections. /// /// The store runs `synchronous=NORMAL` in WAL mode and keeps multiple /// connections open, so SQLite's last-connection-close auto-checkpoint /// never fires on the transient reset connection — without an explicit - /// checkpoint the DELETEs live only in the WAL. On Android the Clear - /// button is routinely followed by a force-stop (SIGKILL, no graceful - /// close, no checkpoint), which discards those WAL frames and reopens - /// the OLD full tree (the 771/771 symptom). + /// checkpoint the DELETEs live only in the WAL, which a power loss or OS + /// crash can lose. A process kill alone keeps them. /// - /// This test proves the fix WITHOUT dropping the store (a graceful drop - /// would checkpoint and mask the bug — exactly what the prior test's - /// `drop(store)` did): it opens an INDEPENDENT read-only connection that - /// deliberately does NOT attach the `-wal` (`?immutable=1`), so it can - /// only see rows already written to the main db file. If the reset left - /// the shard rows in the WAL, this connection would still see the old - /// leaves; seeing zero proves the checkpoint flushed them to the main db, - /// where they survive any later process death. + /// This test proves the checkpoint WITHOUT dropping the store (a graceful + /// drop would checkpoint and mask a missing one): it opens an INDEPENDENT + /// read-only connection that deliberately does NOT attach the `-wal` + /// (`?immutable=1`), so it can only see rows already written to the main db + /// file. If the reset left the shard rows in the WAL, this connection would + /// still see the old leaves; seeing zero proves the checkpoint flushed them + /// to the main db. #[test] fn reset_commitment_tree_flushes_to_main_db_file_not_just_wal() { let path = temp_tree_path("reset_durable"); @@ -1852,7 +2009,7 @@ mod tests { assert_eq!(store.tree_size().unwrap(), N); // Reset. The store's connections stay open (mirroring a live - // process that hasn't been force-stopped yet). + // process). store.reset_commitment_tree().unwrap(); assert_eq!(store.tree_size().unwrap(), 0); @@ -1878,7 +2035,7 @@ mod tests { shard_rows, 0, "reset must checkpoint the emptied tables into the MAIN db file; \ a non-zero count means the DELETEs sit only in the WAL and a \ - force-stop before checkpoint would resurrect the old tree" + power loss or OS crash before a checkpoint could resurrect the old tree" ); } diff --git a/packages/rs-platform-wallet/src/wallet/shielded/store.rs b/packages/rs-platform-wallet/src/wallet/shielded/store.rs index 6e20a0ae1fc..9bff865ef18 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/store.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/store.rs @@ -239,6 +239,20 @@ pub trait ShieldedStore: Send + Sync { /// Return all notes (spent and unspent) for `id`. fn get_all_notes(&self, id: SubwalletId) -> Result, Self::Error>; + /// Look up one receipt's spent status; `None` means the nullifier is unknown. + /// Stores with a nullifier index should override this compatibility fallback. + fn note_spent_status( + &self, + id: SubwalletId, + nullifier: &[u8; 32], + ) -> Result, Self::Error> { + Ok(self + .get_all_notes(id)? + .iter() + .find(|note| ¬e.nullifier == nullifier) + .map(|note| note.is_spent)) + } + /// Mark `id`'s note with `nullifier` as spent. Returns `true` /// if a matching unspent note was found. fn mark_spent(&mut self, id: SubwalletId, nullifier: &[u8; 32]) -> Result; @@ -460,6 +474,19 @@ pub trait ShieldedStore: Send + Sync { /// found in the recorded anchors tree" at spend time). fn tree_size(&self) -> Result; + /// An upper bound on the response heights needed to supply all local leaves. + /// `None` means provenance is unknown: a lower chain count cannot prove corruption. + fn tree_sync_height(&self) -> Result, Self::Error> { + Ok(None) + } + + /// Raise the tree's height bound before appending data from that response. + /// Persistent stores must durably order this write before the corresponding + /// appends. Stores without provenance retain the default unknown bound. + fn set_tree_sync_height(&mut self, _height: u64) -> Result<(), Self::Error> { + Ok(()) + } + // ── Sync state (per-subwallet) ───────────────────────────────────── /// Sync watermark for `id`: the count of note positions already @@ -628,6 +655,13 @@ impl SubwalletState { .try_fold(0u64, |total, note| total.checked_add(note.value)) } + pub(super) fn note_spent_status(&self, nullifier: &[u8; 32]) -> Option { + self.nullifier_index + .get(nullifier) + .and_then(|index| self.notes.get(*index)) + .map(|note| note.is_spent) + } + pub(super) fn all_notes(&self) -> Vec { self.notes.clone() } @@ -853,6 +887,7 @@ pub struct InMemoryShieldedStore { checkpoints: Vec, /// Placeholder anchor; production stores compute the real Sinsemilla root. anchor: [u8; 32], + tree_sync_height: Option, } impl InMemoryShieldedStore { @@ -893,6 +928,17 @@ impl ShieldedStore for InMemoryShieldedStore { .unwrap_or_default()) } + fn note_spent_status( + &self, + id: SubwalletId, + nullifier: &[u8; 32], + ) -> Result, Self::Error> { + Ok(self + .subwallets + .get(&id) + .and_then(|state| state.note_spent_status(nullifier))) + } + fn mark_spent(&mut self, id: SubwalletId, nullifier: &[u8; 32]) -> Result { // In-memory store: the redrive map lives in the same // `SubwalletState`, so `mark_spent` already dropped any resolved @@ -1091,6 +1137,15 @@ impl ShieldedStore for InMemoryShieldedStore { Ok(self.commitments.len() as u64) } + fn tree_sync_height(&self) -> Result, Self::Error> { + Ok(self.tree_sync_height) + } + + fn set_tree_sync_height(&mut self, height: u64) -> Result<(), Self::Error> { + self.tree_sync_height = Some(self.tree_sync_height.unwrap_or(0).max(height)); + Ok(()) + } + fn last_synced_note_index(&self, id: SubwalletId) -> Result { Ok(self .subwallets @@ -1142,6 +1197,7 @@ impl ShieldedStore for InMemoryShieldedStore { // checkpoint vectors drops `tree_size()` to 0 and makes the // next append start at position 0, matching the file-backed // store's reset contract. + self.tree_sync_height = None; self.commitments.clear(); self.marked_positions.clear(); self.checkpoints.clear(); diff --git a/packages/rs-platform-wallet/src/wallet/shielded/sync.rs b/packages/rs-platform-wallet/src/wallet/shielded/sync.rs index cddaed6718d..187ccfa4c41 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/sync.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/sync.rs @@ -226,20 +226,27 @@ pub(super) async fn sync_notes_across( // it already stored, while a lagging one still saves from its // own start. let (watermarks, already_have, tree_size) = { + let tree_size = store + .tree_size() + .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))?; let mut watermarks: BTreeMap = BTreeMap::new(); let mut min_idx: Option = None; for (id, _) in subwallets { let idx = store .last_synced_note_index(*id) .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))?; + // Host persistence may restore a pre-rebuild watermark after a + // restart. Never skip commitments absent from the rebuilt tree. + let idx = idx.min(tree_size); watermarks.insert(*id, idx); min_idx = Some(min_idx.map_or(idx, |m| m.min(idx))); } - let tree_size = store - .tree_size() - .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))?; (watermarks, min_idx.unwrap_or(0), tree_size) }; + let mut tree_height = store + .tree_sync_height() + .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))? + .or(if tree_size == 0 { Some(0) } else { None }); let aligned_start = (already_have / CHUNK_SIZE) * CHUNK_SIZE; info!( @@ -267,6 +274,10 @@ pub(super) async fn sync_notes_across( // `tree_size == 0`, so this preserves the host's indeterminate handling // (total 0) until the first batch lands. let mut total_target: u64 = tree_size; + // Keep evidence separate from the progress denominator, which is floored + // at the local size and would hide excess local leaves. + let mut chain_leaf_count: Option = None; + let mut response_height = 0; // Drive the FIRST subwallet's IVK as the streaming driver. Its hits // come back per batch as `batch.decrypted`; every other subwallet's @@ -408,7 +419,27 @@ pub(super) async fn sync_notes_across( // so a late chunk proven at a slightly higher block never lowers // it. Stays 0 (indeterminate) only if no batch is ever produced. total_target = total_target.max(batch.total_count); + chain_leaf_count = Some(chain_leaf_count.unwrap_or(0).max(batch.total_count)); + response_height = response_height.max(batch.block_height); total_notes_scanned += batch.notes.len() as u64; + let batch_end = batch.start_index + batch.notes.len() as u64; + let mut next_tree_height = tree_height; + // A legacy tree's last existing position must actually be present in + // this chunk. total_count may come from a newer speculative response, + // so pairing that count with this chunk's height would be unsound. + if next_tree_height.is_none() && batch.start_index < tree_size && batch_end >= tree_size { + next_tree_height = Some(batch.block_height); + } + if let Some(height) = next_tree_height { + if batch_end > tree_size || tree_height != next_tree_height { + let height = height.max(batch.block_height); + store + .set_tree_sync_height(height) + .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))?; + next_tree_height = Some(height); + } + } + tree_height = next_tree_height; if !batch.notes.is_empty() { last_nonempty = Some((batch.start_index, batch.is_partial)); } @@ -521,6 +552,29 @@ pub(super) async fn sync_notes_across( } } + if let Some(chain_leaf_count) = chain_leaf_count { + let local_leaf_count = store + .tree_size() + .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))?; + if local_leaf_count > chain_leaf_count { + // The stream's maximum count includes the proof at response_height. + // Only that height (or newer) can contradict our tree's bound. + if tree_height.is_some_and(|height| response_height >= height) { + return Err(PlatformWalletError::ShieldedTreeInconsistent { + local_leaf_count, + chain_leaf_count, + }); + } + debug!( + local_leaf_count, + chain_leaf_count, + ?tree_height, + response_height, + "Skipping shielded tree count check: source is stale or tree provenance is unknown" + ); + } + } + // Diagnostic mirror of the SDK's one-shot `next_start_index` // (rewinds to the last partial chunk's start). NOT the watermark: // the streaming path persists `aligned_start + total_notes_scanned` @@ -600,6 +654,9 @@ pub(super) async fn sync_notes_across( continue; } let nullifier = d.note.nullifier(&views.full_viewing_key); + let known_spent = store + .note_spent_status(*id, &nullifier.to_bytes()) + .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))?; let value = d.note.value().inner(); debug!( wallet_id = %hex::encode(id.wallet_id), @@ -621,14 +678,18 @@ pub(super) async fn sync_notes_across( cmx: d.cmx, nullifier: nullifier.to_bytes(), block_height: d.block_height, - is_spent: false, + // Rebuild rescans receipts; a known spend must never become + // selectable while its spending action is being re-scanned. + is_spent: known_spent.unwrap_or(false), value, }; store .save_note(*id, &shielded_note) .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))?; changeset.record_note(*id, shielded_note); - *per_subwallet_new_notes.entry(*id).or_default() += 1; + if known_spent.is_none() { + *per_subwallet_new_notes.entry(*id).or_default() += 1; + } } } @@ -683,13 +744,14 @@ pub(super) async fn sync_notes_across( ) .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))?; - // Advance every subwallet's watermark to the same global - // tree position so the next sync resumes coherently across - // the union. `total_notes_scanned` is accumulated from every + // Advance each subwallet to the scanned position without undoing prior + // coverage when this pass was served by a lagging node. `total_notes_scanned` is accumulated from every // streamed batch's note count — identical to the one-shot path's // `result.total_notes_scanned`. let new_index = aligned_start + total_notes_scanned; for (id, _) in subwallets { + // A lagging source must not rewind a previously scanned subwallet. + let new_index = new_index.max(watermarks.get(id).copied().unwrap_or(0)); store .set_last_synced_note_index(*id, new_index) .map_err(|e| PlatformWalletError::ShieldedStoreError(e.to_string()))?; @@ -1075,3 +1137,6 @@ mod memo_roundtrip_tests; #[cfg(test)] mod concurrent_sync_tests; + +#[cfg(test)] +mod tree_consistency_tests; diff --git a/packages/rs-platform-wallet/src/wallet/shielded/sync/concurrent_sync_tests.rs b/packages/rs-platform-wallet/src/wallet/shielded/sync/concurrent_sync_tests.rs index ace5a8c37e6..07221a8ecd4 100644 --- a/packages/rs-platform-wallet/src/wallet/shielded/sync/concurrent_sync_tests.rs +++ b/packages/rs-platform-wallet/src/wallet/shielded/sync/concurrent_sync_tests.rs @@ -8,10 +8,15 @@ use drive_proof_verifier::types::{ use tokio::sync::RwLock; use super::{sync_notes_across, MAX_CONCURRENT_CHUNK_FETCHES}; +use crate::error::PlatformWalletError; use crate::wallet::shielded::keys::{AccountViewingKeys, OrchardKeySet}; use crate::wallet::shielded::store::{InMemoryShieldedStore, ShieldedStore, SubwalletId}; async fn pool_sdk(note_count: u8) -> Arc { + pool_sdk_with_total(note_count, u64::from(note_count)).await +} + +async fn pool_sdk_with_total(note_count: u8, total_count: u64) -> Arc { let mut sdk = Sdk::new_mock(); let count = 2048 * u32::from( @@ -41,10 +46,7 @@ async fn pool_sdk(note_count: u8) -> Arc { start_index: u64::from(count) * chunk, count, }, - Some(ShieldedEncryptedNotes { - notes, - total_count: u64::from(note_count), - }), + Some(ShieldedEncryptedNotes { notes, total_count }), ) .await .unwrap(); @@ -138,3 +140,117 @@ async fn should_append_commitments_once_across_concurrent_resumed_syncs() { // One subwallet is caught up and the other still needs the existing leaves. assert_concurrent_sync_matches_sequential(2).await; } + +#[tokio::test] +async fn should_reject_duplicated_commitments_after_sync() { + let keys = OrchardKeySet::from_seed(&[1; 32], Network::Testnet, 0).unwrap(); + let id = SubwalletId::new([1; 32], 0); + let subwallets = [( + id, + AccountViewingKeys::from_full_viewing_key(keys.full_viewing_key), + )]; + let store = seeded_store(4, id); + sync_notes_across(&pool_sdk(4).await, &store, &subwallets, None, None) + .await + .unwrap(); + assert_eq!(store.read().await.tree_sync_height().unwrap(), Some(0)); + store + .write() + .await + .append_commitment(&[4; 32], true) + .unwrap(); + let result = sync_notes_across(&pool_sdk(4).await, &store, &subwallets, None, None).await; + assert!( + matches!( + result, + Err(PlatformWalletError::ShieldedTreeInconsistent { + local_leaf_count: 5, + chain_leaf_count: 4, + }) + ), + "a duplicated tree must fail sync with evidence, got {result:?}" + ); + assert_eq!( + store.read().await.tree_size().unwrap(), + 5, + "detection must not rebuild automatically" + ); +} + +#[tokio::test] +async fn should_accept_healthy_and_behind_trees() { + let keys = OrchardKeySet::from_seed(&[1; 32], Network::Testnet, 0).unwrap(); + let id = SubwalletId::new([1; 32], 0); + let subwallets = [( + id, + AccountViewingKeys::from_full_viewing_key(keys.full_viewing_key), + )]; + for initial_size in [0, 2, 4] { + let store = seeded_store(initial_size, id); + sync_notes_across(&pool_sdk(4).await, &store, &subwallets, None, None) + .await + .unwrap(); + assert_eq!(store.read().await.tree_size().unwrap(), 4); + } +} + +#[tokio::test] +async fn should_accept_tree_behind_streams_maximum_count() { + let keys = OrchardKeySet::from_seed(&[1; 32], Network::Testnet, 0).unwrap(); + let id = SubwalletId::new([1; 32], 0); + let store = seeded_store(0, id); + // The stream's max-seen total can come from a newer speculative response + // than its final partial chunk. The local tree remains legitimately behind. + sync_notes_across( + &pool_sdk_with_total(4, 6).await, + &store, + &[(id, keys.viewing_keys())], + None, + None, + ) + .await + .unwrap(); + assert_eq!(store.read().await.tree_size().unwrap(), 4); +} + +#[tokio::test] +async fn should_skip_lower_count_for_legacy_tree_with_unknown_height() { + let keys = OrchardKeySet::from_seed(&[1; 32], Network::Testnet, 0).unwrap(); + let id = SubwalletId::new([1; 32], 0); + // A restored healthy tree may have been built at height 1; SDK mock + // responses have height 0. Without provenance a lower count proves nothing. + let store = seeded_store(5, id); + sync_notes_across( + &pool_sdk(4).await, + &store, + &[(id, keys.viewing_keys())], + None, + None, + ) + .await + .expect("an older source must not diagnose a healthy tree as corrupt"); + assert_eq!(store.read().await.tree_size().unwrap(), 5); +} + +#[tokio::test] +async fn should_not_infer_tree_height_from_speculative_total() { + let keys = OrchardKeySet::from_seed(&[1; 32], Network::Testnet, 0).unwrap(); + let id = SubwalletId::new([1; 32], 0); + let store = seeded_store(5, id); + let subwallets = [(id, keys.viewing_keys())]; + // total_count may describe a newer speculative response, not this chunk. + sync_notes_across( + &pool_sdk_with_total(4, 6).await, + &store, + &subwallets, + None, + None, + ) + .await + .unwrap(); + assert_eq!(store.read().await.tree_sync_height().unwrap(), None); + sync_notes_across(&pool_sdk(4).await, &store, &subwallets, None, None) + .await + .unwrap(); + assert_eq!(store.read().await.last_synced_note_index(id).unwrap(), 5); +} diff --git a/packages/rs-platform-wallet/src/wallet/shielded/sync/tree_consistency_tests.rs b/packages/rs-platform-wallet/src/wallet/shielded/sync/tree_consistency_tests.rs new file mode 100644 index 00000000000..58453b15232 --- /dev/null +++ b/packages/rs-platform-wallet/src/wallet/shielded/sync/tree_consistency_tests.rs @@ -0,0 +1,516 @@ +use std::collections::BTreeMap; +use std::sync::Arc; + +use dash_async::ThreadRegistry; +use dash_sdk::platform::fetch_current_no_parameters::FetchCurrent; +use dash_sdk::Sdk; +use dashcore::Network; +use drive_proof_verifier::types::{ + NoParamQuery, ShieldedAnchors, ShieldedEncryptedNote, ShieldedEncryptedNotes, + ShieldedEncryptedNotesQuery, +}; +use grovedb_commitment_tree::{ + DashMemo, Domain, ExtractedNoteCommitment, Note, NoteValue, OrchardDomain, + OrchardNoteEncryption, RandomSeed, Rho, +}; +use rusqlite::Connection; +use tokio::sync::RwLock; + +use super::{serialize_note, sync_notes_across, MAX_CONCURRENT_CHUNK_FETCHES}; +use crate::error::PlatformWalletError; +use crate::events::PlatformEventManager; +use crate::manager::shielded_sync::{ShieldedSyncManager, WalletShieldedOutcome}; +use crate::wallet::persister::{NoPlatformPersistence, WalletPersister}; +use crate::wallet::shielded::balance::ShieldedBalanceSource; +use crate::wallet::shielded::keys::OrchardKeySet; +use crate::wallet::shielded::store::{ShieldedNote, ShieldedStore, SubwalletId}; +use crate::wallet::shielded::{FileBackedShieldedStore, NetworkShieldedCoordinator}; + +fn owned_note(keys: &OrchardKeySet, position: u64) -> (ShieldedNote, ShieldedEncryptedNote) { + let mut rho_bytes = [0; 32]; + rho_bytes[0] = position as u8 + 1; + let rho = Rho::from_bytes(&rho_bytes).unwrap(); + let rseed = (1u16..=u16::MAX) + .find_map(|n| { + let mut bytes = [0; 32]; + bytes[2..4].copy_from_slice(&n.to_le_bytes()); + RandomSeed::from_bytes(bytes, &rho).into_option() + }) + .unwrap(); + let note = + Note::from_parts(keys.default_address, NoteValue::from_raw(100), rho, rseed).unwrap(); + let cmx = ExtractedNoteCommitment::from(note.commitment()).to_bytes(); + let encryption = OrchardNoteEncryption::::new(None, note, [0; 36]); + let mut ciphertext = OrchardDomain::::epk_bytes(encryption.epk()) + .0 + .to_vec(); + ciphertext.extend_from_slice(encryption.encrypt_note_plaintext().as_ref()); + ciphertext.extend_from_slice(&[0; 80]); + ( + ShieldedNote { + note_data: serialize_note(¬e), + position, + cmx, + nullifier: note.nullifier(&keys.full_viewing_key).to_bytes(), + block_height: 1, + is_spent: false, + value: 100, + }, + ShieldedEncryptedNote { + cmx: cmx.to_vec(), + nullifier: rho_bytes.to_vec(), + cv_net: vec![0; 32], + encrypted_note: ciphertext, + }, + ) +} + +async fn sdk_with_notes(notes: Vec) -> Arc { + let mut sdk = Sdk::new_mock(); + let count = 2048 + * u32::from( + sdk.version() + .drive_abci + .query + .shielded_queries + .max_query_chunks, + ); + for chunk in 0..MAX_CONCURRENT_CHUNK_FETCHES as u64 { + sdk.mock() + .expect_fetch( + ShieldedEncryptedNotesQuery { + start_index: u64::from(count) * chunk, + count, + }, + Some(ShieldedEncryptedNotes { + notes: if chunk == 0 { + notes.clone() + } else { + Vec::new() + }, + total_count: notes.len() as u64, + }), + ) + .await + .unwrap(); + } + Arc::new(sdk) +} + +#[tokio::test] +async fn should_rebuild_tree_preserving_notes_and_spends_then_match_recorded_anchor() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("tree.sqlite"); + let keys = OrchardKeySet::from_seed(&[42; 32], Network::Testnet, 0).unwrap(); + let id = SubwalletId::new([42; 32], 0); + let (first, first_wire) = owned_note(&keys, 0); + let (mut spent, spent_wire) = owned_note(&keys, 1); + spent.is_spent = true; + let sdk = sdk_with_notes(vec![first_wire, spent_wire]).await; + let mut store = FileBackedShieldedStore::open_path(&path, 100).unwrap(); + for note in [&first, &spent] { + store.append_commitment(¬e.cmx, true).unwrap(); + store.save_note(id, note).unwrap(); + } + store.set_tree_sync_height(0).unwrap(); + store.checkpoint_tree(2).unwrap(); + let recorded_anchor = store.tree_anchor().unwrap(); + store.append_commitment(&spent.cmx, true).unwrap(); + store.checkpoint_tree(3).unwrap(); + store.set_last_synced_note_index(id, 2).unwrap(); + let coordinator = Arc::new(NetworkShieldedCoordinator::new( + sdk, + Network::Testnet, + path.clone(), + store, + )); + coordinator + .register_wallet( + id.wallet_id, + BTreeMap::from([(0, keys.viewing_keys())]), + WalletPersister::new(id.wallet_id, Arc::new(NoPlatformPersistence)), + ) + .await + .unwrap(); + + assert!(matches!( + coordinator.try_sync(true).await, + Err(PlatformWalletError::ShieldedTreeInconsistent { + local_leaf_count: 3, + chain_leaf_count: 2, + }) + )); + let sync_manager = ShieldedSyncManager::new( + Arc::new(PlatformEventManager::new(Vec::new())), + Arc::new(RwLock::new(Some(Arc::clone(&coordinator)))), + ThreadRegistry::new(), + ); + assert!(sync_manager + .sync_wallet(&[99; 32], true) + .await + .unwrap() + .is_none()); + assert!(matches!( + sync_manager.sync_wallet(&id.wallet_id, true).await, + Err(PlatformWalletError::ShieldedTreeInconsistent { + local_leaf_count: 3, + chain_leaf_count: 2, + }) + )); + // The background pass reports the same verdict per wallet as a typed + // outcome, so hosts need not parse the error text to offer a rebuild. + let summary = coordinator.sync(false).await; + assert_eq!(summary.error_count(), 1); + assert_eq!(summary.success_count(), 0); + assert!( + matches!( + summary.wallet_results.get(&id.wallet_id), + Some(WalletShieldedOutcome::TreeInconsistent { + local_leaf_count: 3, + chain_leaf_count: 2, + }) + ), + "background sync must report a typed inconsistency, got {:?}", + summary.wallet_results.get(&id.wallet_id) + ); + coordinator.rebuild_shielded_tree().await.unwrap(); + { + let store = coordinator.store().read().await; + assert_eq!(store.tree_size().unwrap(), 0); + assert_eq!(store.last_synced_note_index(id).unwrap(), 0); + let notes = store.get_all_notes(id).unwrap(); + assert_eq!(notes.len(), 2); + assert!(notes + .iter() + .any(|n| n.nullifier == spent.nullifier && n.is_spent)); + } + assert_eq!( + FileBackedShieldedStore::open_path(&path, 100) + .unwrap() + .tree_size() + .unwrap(), + 0 + ); + let summary = coordinator.sync(false).await; + assert_eq!(summary.error_count(), 0); + assert_eq!( + summary.success_count(), + 1, + "registration must survive rebuild" + ); + let store = coordinator.store().read().await; + assert_eq!(store.tree_size().unwrap(), 2); + assert_eq!(store.last_synced_note_index(id).unwrap(), 2); + let notes = store.get_all_notes(id).unwrap(); + assert_eq!(notes.len(), 2); + assert!(notes + .iter() + .any(|n| n.nullifier == spent.nullifier && n.is_spent)); + assert_eq!(store.get_unspent_notes(id).unwrap().len(), 1); + let witness = store.witness_at_depth(0, 0).unwrap().unwrap(); + let root = witness + .root(ExtractedNoteCommitment::from_bytes(&first.cmx).unwrap()) + .to_bytes(); + drop(store); + let mut anchors_sdk = Sdk::new_mock(); + anchors_sdk + .mock() + .expect_fetch( + NoParamQuery {}, + Some(ShieldedAnchors(vec![recorded_anchor])), + ) + .await + .unwrap(); + let anchors = ShieldedAnchors::fetch_current(&anchors_sdk).await.unwrap(); + assert!(anchors.0.contains(&root)); + // Enter the caught-up cooldown, then prove explicit rebuild bypasses it. + coordinator.sync(false).await; + coordinator.rebuild_shielded_tree().await.unwrap(); + let summary = coordinator.sync(false).await; + assert_eq!(summary.error_count(), 0); + assert_eq!(coordinator.store().read().await.tree_size().unwrap(), 2); +} + +#[tokio::test] +async fn should_rescan_empty_tree_despite_restored_watermark() { + let dir = tempfile::tempdir().unwrap(); + let mut store = + FileBackedShieldedStore::open_path(dir.path().join("tree.sqlite"), 100).unwrap(); + let keys = OrchardKeySet::from_seed(&[42; 32], Network::Testnet, 0).unwrap(); + let id = SubwalletId::new([42; 32], 0); + let (note, wire) = owned_note(&keys, 0); + store.save_note(id, ¬e).unwrap(); + // A host can restore its old watermark after a rebuild and process restart. + store.set_last_synced_note_index(id, 4096).unwrap(); + let store = Arc::new(RwLock::new(store)); + sync_notes_across( + &sdk_with_notes(vec![wire]).await, + &store, + &[(id, keys.viewing_keys())], + None, + None, + ) + .await + .unwrap(); + assert_eq!(store.read().await.tree_size().unwrap(), 1); + assert_eq!(store.read().await.get_all_notes(id).unwrap().len(), 1); +} + +#[tokio::test] +async fn should_preserve_tree_height_across_reopen_and_ignore_one_block_behind_source() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("tree.sqlite"); + let keys = OrchardKeySet::from_seed(&[42; 32], Network::Testnet, 0).unwrap(); + let id = SubwalletId::new([42; 32], 0); + let (first, wire) = owned_note(&keys, 0); + let (second, _) = owned_note(&keys, 1); + { + let mut store = FileBackedShieldedStore::open_path(&path, 100).unwrap(); + // The tree was built from height 1; SDK mock responses are at height 0. + store.set_tree_sync_height(1).unwrap(); + store.append_commitment(&first.cmx, true).unwrap(); + store.append_commitment(&second.cmx, true).unwrap(); + store.checkpoint_tree(2).unwrap(); + } + let mut store = FileBackedShieldedStore::open_path(&path, 100).unwrap(); + assert_eq!(store.tree_sync_height().unwrap(), Some(1)); + store.set_last_synced_note_index(id, 2).unwrap(); + let store = Arc::new(RwLock::new(store)); + sync_notes_across( + &sdk_with_notes(vec![wire]).await, + &store, + &[(id, keys.viewing_keys())], + None, + None, + ) + .await + .expect("one-block-behind source must not trigger inconsistency"); + let mut store = store.write().await; + assert_eq!(store.tree_size().unwrap(), 2); + assert_eq!(store.tree_sync_height().unwrap(), Some(1)); + assert_eq!(store.last_synced_note_index(id).unwrap(), 2); + store.reset_commitment_tree().unwrap(); + assert_eq!(store.tree_sync_height().unwrap(), None); +} + +#[test] +fn should_reset_scan_metadata_when_rewinding_for_rebuild() { + let dir = tempfile::tempdir().unwrap(); + let mut store = + FileBackedShieldedStore::open_path(dir.path().join("tree.sqlite"), 100).unwrap(); + let keys = OrchardKeySet::from_seed(&[42; 32], Network::Testnet, 0).unwrap(); + let with_notes = SubwalletId::new([42; 32], 0); + let empty = SubwalletId::new([42; 32], 1); + let (note, _) = owned_note(&keys, 0); + store.save_note(with_notes, ¬e).unwrap(); + store.set_last_synced_note_index(with_notes, 2).unwrap(); + store.set_last_synced_note_index(empty, 2).unwrap(); + + store.rewind_scan_watermarks(); + + // The emptied tree no longer backs the old scan coverage; only the + // retained note ledger remains as evidence. + let balance = store.local_account_balance(with_notes).unwrap(); + assert_eq!(balance.last_scanned_index, None); + assert_eq!(balance.source, ShieldedBalanceSource::Restored); + assert_eq!(balance.spendable_credits, 100); + let balance = store.local_account_balance(empty).unwrap(); + assert_eq!(balance.last_scanned_index, None); + assert_eq!(balance.source, ShieldedBalanceSource::NoHistory); +} + +#[test] +fn should_lookup_only_the_discovered_nullifier_spent_status() { + let dir = tempfile::tempdir().unwrap(); + let mut store = + FileBackedShieldedStore::open_path(dir.path().join("tree.sqlite"), 100).unwrap(); + let keys = OrchardKeySet::from_seed(&[42; 32], Network::Testnet, 0).unwrap(); + let id = SubwalletId::new([42; 32], 0); + let (note, _) = owned_note(&keys, 0); + assert_eq!(store.note_spent_status(id, ¬e.nullifier).unwrap(), None); + store.save_note(id, ¬e).unwrap(); + assert_eq!( + store.note_spent_status(id, ¬e.nullifier).unwrap(), + Some(false) + ); + store.mark_spent(id, ¬e.nullifier).unwrap(); + assert_eq!( + store.note_spent_status(id, ¬e.nullifier).unwrap(), + Some(true) + ); +} + +#[tokio::test] +async fn should_record_height_before_appending_commitments() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("tree.sqlite"); + let store = Arc::new(RwLock::new( + FileBackedShieldedStore::open_path(&path, 100).unwrap(), + )); + let keys = OrchardKeySet::from_seed(&[42; 32], Network::Testnet, 0).unwrap(); + let id = SubwalletId::new([42; 32], 0); + let (_, wire) = owned_note(&keys, 0); + let conn = Connection::open(&path).unwrap(); + conn.execute_batch( + "CREATE TRIGGER reject_height BEFORE INSERT ON shielded_tree_sync_height + BEGIN SELECT RAISE(FAIL, 'injected metadata failure'); END;", + ) + .unwrap(); + let result = sync_notes_across( + &sdk_with_notes(vec![wire]).await, + &store, + &[(id, keys.viewing_keys())], + None, + None, + ) + .await; + assert!(matches!( + result, + Err(PlatformWalletError::ShieldedStoreError(_)) + )); + assert_eq!( + store.read().await.tree_size().unwrap(), + 0, + "height persistence must succeed before any corresponding leaf is appended" + ); +} + +/// A coordinator over a 3-leaf local tree that the 2-note chain contradicts, +/// with the subwallet's cursor at 2 and the tree's recorded height present. +async fn inconsistent_coordinator( + path: &std::path::Path, +) -> (Arc, SubwalletId) { + let keys = OrchardKeySet::from_seed(&[42; 32], Network::Testnet, 0).unwrap(); + let id = SubwalletId::new([42; 32], 0); + let (first, first_wire) = owned_note(&keys, 0); + let (second, second_wire) = owned_note(&keys, 1); + let mut store = FileBackedShieldedStore::open_path(path, 100).unwrap(); + for note in [&first, &second, &second] { + store.append_commitment(¬e.cmx, true).unwrap(); + } + store.save_note(id, &first).unwrap(); + store.set_tree_sync_height(0).unwrap(); + store.checkpoint_tree(3).unwrap(); + store.set_last_synced_note_index(id, 2).unwrap(); + let coordinator = Arc::new(NetworkShieldedCoordinator::new( + sdk_with_notes(vec![first_wire, second_wire]).await, + Network::Testnet, + path.to_path_buf(), + store, + )); + coordinator + .register_wallet( + id.wallet_id, + BTreeMap::from([(0, keys.viewing_keys())]), + WalletPersister::new(id.wallet_id, Arc::new(NoPlatformPersistence)), + ) + .await + .unwrap(); + (coordinator, id) +} + +#[tokio::test] +async fn should_leave_tree_and_cursors_untouched_when_rebuild_reset_fails() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("tree.sqlite"); + let (coordinator, id) = inconsistent_coordinator(&path).await; + // Fail the LAST statement of the reset, after every tree table was + // already emptied: only a single transaction can undo those deletes. + let conn = Connection::open(&path).unwrap(); + conn.execute_batch( + "CREATE TRIGGER reject_height_reset BEFORE DELETE ON shielded_tree_sync_height + BEGIN SELECT RAISE(FAIL, 'injected reset failure'); END;", + ) + .unwrap(); + + let result = coordinator.rebuild_shielded_tree().await; + + assert!( + matches!(result, Err(PlatformWalletError::ShieldedStoreError(_))), + "the injected failure must surface: {result:?}" + ); + { + let store = coordinator.store().read().await; + assert_eq!(store.tree_size().unwrap(), 3, "live tree must be untouched"); + assert_eq!( + store.last_synced_note_index(id).unwrap(), + 2, + "the cursor must not be rewound over an intact tree" + ); + } + assert_eq!( + FileBackedShieldedStore::open_path(&path, 100) + .unwrap() + .tree_size() + .unwrap(), + 3, + "on-disk tree must be untouched" + ); + conn.execute_batch("DROP TRIGGER reject_height_reset;") + .unwrap(); + // The next pass reports the same verdict, so the host can retry. + let summary = coordinator.sync(false).await; + assert!( + matches!( + summary.wallet_results.get(&id.wallet_id), + Some(WalletShieldedOutcome::TreeInconsistent { + local_leaf_count: 3, + chain_leaf_count: 2, + }) + ), + "a failed rebuild must leave the inconsistency detectable, got {:?}", + summary.wallet_results.get(&id.wallet_id) + ); + coordinator.rebuild_shielded_tree().await.unwrap(); + assert_eq!(coordinator.store().read().await.tree_size().unwrap(), 0); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn should_attribute_a_failed_pass_to_the_wallets_it_snapshotted() { + let dir = tempfile::tempdir().unwrap(); + let (coordinator, id) = inconsistent_coordinator(&dir.path().join("tree.sqlite")).await; + // Park the pass on the store write lock, after its account snapshot. + let store_guard = coordinator.store().read().await; + let pass = tokio::spawn({ + let coordinator = Arc::clone(&coordinator); + async move { coordinator.sync(true).await } + }); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + // Tokio's RwLock is write-preferring: once the pass queues for the + // write guard, further readers are refused. + while coordinator.store().try_read().is_ok() { + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + }) + .await + .expect("the pass never queued for the store write lock"); + // Unregister mid-pass: the registry drops the wallet at once, and the + // store purge queues behind the pass. + let unregister = tokio::spawn({ + let coordinator = Arc::clone(&coordinator); + async move { coordinator.unregister_wallet(id.wallet_id).await } + }); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while !coordinator.registered_subwallets().await.is_empty() { + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + }) + .await + .expect("unregister never dropped the wallet from the registry"); + drop(store_guard); + + let summary = pass.await.unwrap(); + unregister.await.unwrap(); + + assert!( + matches!( + summary.wallet_results.get(&id.wallet_id), + Some(WalletShieldedOutcome::TreeInconsistent { + local_leaf_count: 3, + chain_leaf_count: 2, + }) + ), + "the failure belongs to the wallets the pass ran for, got {:?}", + summary.wallet_results + ); +} diff --git a/packages/rs-unified-sdk-jni/src/events.rs b/packages/rs-unified-sdk-jni/src/events.rs index 47639f89a60..ffe627b567f 100644 --- a/packages/rs-unified-sdk-jni/src/events.rs +++ b/packages/rs-unified-sdk-jni/src/events.rs @@ -272,7 +272,7 @@ unsafe extern "C" fn tramp_shielded_sync_completed( env.call_method( bridge, "onShieldedSyncCompleted", - "([BZZZIJIJLjava/lang/String;)V", + "([BZZZIJIJLjava/lang/String;I)V", &[ (&wid).into(), JValue::Bool(r.success as u8), @@ -283,6 +283,7 @@ unsafe extern "C" fn tramp_shielded_sync_completed( JValue::Int(r.newly_spent as i32), JValue::Long(r.balance as i64), (&err).into(), + JValue::Int(r.error_code as i32), ], )?; Ok(()) diff --git a/packages/rs-unified-sdk-jni/src/wallet_manager.rs b/packages/rs-unified-sdk-jni/src/wallet_manager.rs index 99eba41dbd0..ed2a951504f 100644 --- a/packages/rs-unified-sdk-jni/src/wallet_manager.rs +++ b/packages/rs-unified-sdk-jni/src/wallet_manager.rs @@ -2863,6 +2863,30 @@ pub extern "system" fn Java_org_dashfoundation_dashsdk_ffi_WalletManagerNative_s }) } +/// Repair an inconsistent shielded commitment tree: empty the tree and +/// rewind scan cursors, keeping notes, spent status and registrations. +/// Background shielded sync restarts by itself if it was running, except +/// after a quiesce timeout on top of an earlier stop's (see +/// `platform_wallet_manager_shielded_rebuild_tree`). Throws +/// `DashSDKException` on failure. Mirrors Swift's +/// `PlatformWalletManager.rebuildShieldedTree()`. +#[cfg(feature = "shielded")] +#[no_mangle] +pub extern "system" fn Java_org_dashfoundation_dashsdk_ffi_WalletManagerNative_shieldedRebuildTree( + mut env: JNIEnv, + _class: JClass, + manager_handle: jlong, +) { + guard(&mut env, (), |env| { + let result = unsafe { + platform_wallet_ffi::platform_wallet_manager_shielded_rebuild_tree( + manager_handle as Handle, + ) + }; + let _ = take_pwffi_error(env, result); + }) +} + /// Run one forced shielded sync pass across all registered wallets — the /// user-initiated "Sync Now" entry point (`force=true` on the Rust side /// bypasses the caught-up cooldown). Blocks the calling thread for the diff --git a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManagerShieldedSync.swift b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManagerShieldedSync.swift index 97d86c8dfb9..18d12a8cc81 100644 --- a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManagerShieldedSync.swift +++ b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletManagerShieldedSync.swift @@ -37,7 +37,7 @@ public struct ShieldedIdentityCreateUnconfirmedError: LocalizedError { /// so the pass passed it over; both `success` and `errorMessage` /// are vacuous. /// - both flags `false` and `errorMessage != nil`: the sync itself -/// failed. +/// failed; `error` carries it typed. public struct ShieldedWalletSyncResult: Sendable { public let walletId: Data public let success: Bool @@ -56,6 +56,16 @@ public struct ShieldedWalletSyncResult: Sendable { public let newlySpent: UInt32 public let balance: UInt64 public let errorMessage: String? + /// Rust's classification of the failure; `.success` on success and + /// skipped wallets. + public let errorCode: PlatformWalletResultCode + + /// The failure as a typed error, or `nil` on success and skipped wallets. + /// `.shieldedTreeInconsistent` means only `rebuildShieldedTree()` repairs it. + public var error: PlatformWalletError? { + guard !success, !skipped else { return nil } + return PlatformWalletError(code: errorCode, message: errorMessage) + } init(ffi: ShieldedSyncWalletResultFFI) { var walletId = ffi.wallet_id @@ -68,6 +78,7 @@ public struct ShieldedWalletSyncResult: Sendable { self.newlySpent = ffi.newly_spent self.balance = ffi.balance self.errorMessage = ffi.error_message.map { String(cString: $0) } + self.errorCode = PlatformWalletResultCode(ffi: ffi.error_code) } } @@ -469,6 +480,30 @@ extension PlatformWalletManager { resetCurrentShieldedProgress() } + /// Repair an inconsistent shielded commitment tree + /// (`PlatformWalletError.shieldedTreeInconsistent`): empty the tree and + /// rewind scan cursors so the next pass rescans the whole pool. Notes, + /// spent status and registrations are kept, so the host keeps its + /// SwiftData rows. Background shielded sync restarts by itself if it was + /// running, except after a quiesce timeout on top of an earlier stop's: + /// check `isShieldedSyncRunning()` after a successful retry. Blocks the + /// calling thread while an in-flight pass drains. Concurrent rebuilds run + /// one at a time, so a second caller also waits for the first to finish. + /// + /// Throws `walletOperation` while an async [`stopShieldedSync()`] is in + /// flight, for the same reason as [`clearShielded()`]. + public func rebuildShieldedTree() throws { + try ensureShieldedLifecycleAllowed("rebuildShieldedTree") + try ensureNoShieldedStopInFlight(before: "rebuilding the shielded tree") + try withShieldedLocalBalanceMutation { + try platform_wallet_manager_shielded_rebuild_tree(handle).check() + } + // Drop a trailing completion from the pass that found the tree + // inconsistent, so it cannot re-surface the error after the repair. + shieldedSyncGeneration.bump() + resetCurrentShieldedProgress() + } + public func isShieldedSyncRunning() throws -> Bool { guard isConfigured, handle != NULL_HANDLE else { throw PlatformWalletError.invalidHandle( diff --git a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletResult.swift b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletResult.swift index e7e746105e6..6ca1485aaae 100644 --- a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletResult.swift +++ b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletResult.swift @@ -264,6 +264,9 @@ public enum PlatformWalletResultCode: Int32, Sendable { case errorSpvProcessRestartRequired = 59 /// A whole-wallet asset lock exceeds the input cap; nothing was broadcast. case errorAssetLockTooManyInputs = 60 + /// The local shielded commitment tree is inconsistent with the chain. + /// Not repaired by resyncing: call `rebuildShieldedTree()`. + case errorShieldedTreeInconsistent = 67 /// The named thing does not exist. Besides the handle/lookup failures this /// has always covered, BOTH deferred-send paths report the /// wallet-was-REMOVED case here. @@ -395,6 +398,8 @@ public enum PlatformWalletResultCode: Int32, Sendable { self = .errorSpvProcessRestartRequired case PLATFORM_WALLET_FFI_RESULT_CODE_ERROR_ASSET_LOCK_TOO_MANY_INPUTS: self = .errorAssetLockTooManyInputs + case PLATFORM_WALLET_FFI_RESULT_CODE_ERROR_SHIELDED_TREE_INCONSISTENT: + self = .errorShieldedTreeInconsistent case PLATFORM_WALLET_FFI_RESULT_CODE_NOT_FOUND: self = .notFound case PLATFORM_WALLET_FFI_RESULT_CODE_ERROR_UNKNOWN: @@ -756,6 +761,11 @@ public enum PlatformWalletError: LocalizedError { case shieldedRecoveryCorrupted(String) /// Recovery needs its account and compatible keys; damaged ciphertext can look the same. case shieldedRecoveryKeysRequired(String) + /// The local shielded commitment tree holds more leaves than a fresh sync + /// proves. Not retryable: call `rebuildShieldedTree()`, which keeps notes + /// and spent status. Kotlin parity: + /// `DashSdkError.PlatformWallet.ShieldedTreeInconsistent`. + case shieldedTreeInconsistent(String) /// The named thing does not exist. For the deferred payment calls this is /// the wallet-was-REMOVED case: the token's wallet (or the wallet a payment /// was just signed against) is no longer registered in the manager, so there @@ -811,6 +821,7 @@ public enum PlatformWalletError: LocalizedError { .shieldedNoRecordedAnchor(let m), .shieldedInsufficientBalance(let m), .shieldedIdentityDebitPending(let m), .shieldedRecoveryCorrupted(let m), .shieldedRecoveryKeysRequired(let m), + .shieldedTreeInconsistent(let m), .transactionBroadcastUnconfirmed(let m), .masternodeWithdrawalUnconfirmed(let m), .masternodeListUnavailable(let m), @@ -1021,6 +1032,8 @@ public enum PlatformWalletError: LocalizedError { self = .shieldedRecoveryCorrupted(detail) case .errorShieldedRecoveryKeysRequired: self = .shieldedRecoveryKeysRequired(detail) + case .errorShieldedTreeInconsistent: + self = .shieldedTreeInconsistent(detail) case .notFound: self = .notFound(detail) // The catch-all, which is where a consensus rejection with no // dedicated code lands. When Rust stamped its verdict on the result, diff --git a/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/ErrorHandlingTests.swift b/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/ErrorHandlingTests.swift index f18b44a9fce..7423c80057e 100644 --- a/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/ErrorHandlingTests.swift +++ b/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/ErrorHandlingTests.swift @@ -439,6 +439,41 @@ final class ErrorHandlingTests: XCTestCase { XCTAssertEqual(keyMessage, detail) } + func testShouldPreserveShieldedTreeInconsistentFromFFI() { + let code = PlatformWalletResultCode( + ffi: PLATFORM_WALLET_FFI_RESULT_CODE_ERROR_SHIELDED_TREE_INCONSISTENT + ) + XCTAssertEqual(code, .errorShieldedTreeInconsistent) + XCTAssertEqual(code.rawValue, 67) + let detail = "Shielded commitment tree has 3 leaves, but a sufficiently fresh sync reports only 2" + guard case .shieldedTreeInconsistent(let message) = PlatformWalletError( + code: code, message: detail + ) else { return XCTFail("lost typed tree-inconsistent error") } + XCTAssertEqual(message, detail) + } + + /// The background sync callback carries the same code, so a host can + /// offer the rebuild without parsing the message. + func testShouldTypeShieldedSyncCallbackResultFromItsErrorCode() { + let detail = "Shielded commitment tree has 3 leaves, but a sufficiently fresh sync reports only 2" + detail.withCString { cString in + var ffi = ShieldedSyncWalletResultFFI() + ffi.error_message = cString + ffi.error_code = PLATFORM_WALLET_FFI_RESULT_CODE_ERROR_SHIELDED_TREE_INCONSISTENT + let result = ShieldedWalletSyncResult(ffi: ffi) + XCTAssertEqual(result.errorCode, .errorShieldedTreeInconsistent) + guard case .shieldedTreeInconsistent(let message)? = result.error else { + return XCTFail("callback result lost the typed tree-inconsistent error") + } + XCTAssertEqual(message, detail) + + var ok = ShieldedSyncWalletResultFFI() + ok.success = true + ok.error_code = PLATFORM_WALLET_FFI_RESULT_CODE_SUCCESS + XCTAssertNil(ShieldedWalletSyncResult(ffi: ok).error) + } + } + func testShouldPreserveSpvProcessRestartRequiredFromFFI() { let code = PlatformWalletResultCode( ffi: PLATFORM_WALLET_FFI_RESULT_CODE_ERROR_SPV_PROCESS_RESTART_REQUIRED