diff --git a/Cargo.lock b/Cargo.lock index fcbbb11546..579d20a91b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1662,7 +1662,7 @@ dependencies = [ [[package]] name = "dash-network" version = "0.45.0" -source = "git+https://github.com/dashpay/rust-dashcore?rev=70d4bf8e36057c58e02d56769a6e9760f701dd06#70d4bf8e36057c58e02d56769a6e9760f701dd06" +source = "git+https://github.com/bfoss765/rust-dashcore?rev=b5dff6de05e4a354680e5b01a54bae9e642a6ad0#b5dff6de05e4a354680e5b01a54bae9e642a6ad0" dependencies = [ "bincode", "bincode_derive", @@ -1673,7 +1673,7 @@ dependencies = [ [[package]] name = "dash-network-seeds" version = "0.45.0" -source = "git+https://github.com/dashpay/rust-dashcore?rev=70d4bf8e36057c58e02d56769a6e9760f701dd06#70d4bf8e36057c58e02d56769a6e9760f701dd06" +source = "git+https://github.com/bfoss765/rust-dashcore?rev=b5dff6de05e4a354680e5b01a54bae9e642a6ad0#b5dff6de05e4a354680e5b01a54bae9e642a6ad0" dependencies = [ "dash-network", ] @@ -1750,7 +1750,7 @@ dependencies = [ [[package]] name = "dash-spv" version = "0.45.0" -source = "git+https://github.com/dashpay/rust-dashcore?rev=70d4bf8e36057c58e02d56769a6e9760f701dd06#70d4bf8e36057c58e02d56769a6e9760f701dd06" +source = "git+https://github.com/bfoss765/rust-dashcore?rev=b5dff6de05e4a354680e5b01a54bae9e642a6ad0#b5dff6de05e4a354680e5b01a54bae9e642a6ad0" dependencies = [ "async-trait", "chrono", @@ -1779,7 +1779,7 @@ dependencies = [ [[package]] name = "dashcore" version = "0.45.0" -source = "git+https://github.com/dashpay/rust-dashcore?rev=70d4bf8e36057c58e02d56769a6e9760f701dd06#70d4bf8e36057c58e02d56769a6e9760f701dd06" +source = "git+https://github.com/bfoss765/rust-dashcore?rev=b5dff6de05e4a354680e5b01a54bae9e642a6ad0#b5dff6de05e4a354680e5b01a54bae9e642a6ad0" dependencies = [ "anyhow", "base64-compat", @@ -1805,12 +1805,12 @@ dependencies = [ [[package]] name = "dashcore-private" version = "0.45.0" -source = "git+https://github.com/dashpay/rust-dashcore?rev=70d4bf8e36057c58e02d56769a6e9760f701dd06#70d4bf8e36057c58e02d56769a6e9760f701dd06" +source = "git+https://github.com/bfoss765/rust-dashcore?rev=b5dff6de05e4a354680e5b01a54bae9e642a6ad0#b5dff6de05e4a354680e5b01a54bae9e642a6ad0" [[package]] name = "dashcore-rpc" version = "0.45.0" -source = "git+https://github.com/dashpay/rust-dashcore?rev=70d4bf8e36057c58e02d56769a6e9760f701dd06#70d4bf8e36057c58e02d56769a6e9760f701dd06" +source = "git+https://github.com/bfoss765/rust-dashcore?rev=b5dff6de05e4a354680e5b01a54bae9e642a6ad0#b5dff6de05e4a354680e5b01a54bae9e642a6ad0" dependencies = [ "dashcore-rpc-json", "hex", @@ -1823,7 +1823,7 @@ dependencies = [ [[package]] name = "dashcore-rpc-json" version = "0.45.0" -source = "git+https://github.com/dashpay/rust-dashcore?rev=70d4bf8e36057c58e02d56769a6e9760f701dd06#70d4bf8e36057c58e02d56769a6e9760f701dd06" +source = "git+https://github.com/bfoss765/rust-dashcore?rev=b5dff6de05e4a354680e5b01a54bae9e642a6ad0#b5dff6de05e4a354680e5b01a54bae9e642a6ad0" dependencies = [ "bincode", "dashcore", @@ -1838,7 +1838,7 @@ dependencies = [ [[package]] name = "dashcore_hashes" version = "0.45.0" -source = "git+https://github.com/dashpay/rust-dashcore?rev=70d4bf8e36057c58e02d56769a6e9760f701dd06#70d4bf8e36057c58e02d56769a6e9760f701dd06" +source = "git+https://github.com/bfoss765/rust-dashcore?rev=b5dff6de05e4a354680e5b01a54bae9e642a6ad0#b5dff6de05e4a354680e5b01a54bae9e642a6ad0" dependencies = [ "bincode", "dashcore-private", @@ -2904,7 +2904,7 @@ dependencies = [ [[package]] name = "git-state" version = "0.45.0" -source = "git+https://github.com/dashpay/rust-dashcore?rev=70d4bf8e36057c58e02d56769a6e9760f701dd06#70d4bf8e36057c58e02d56769a6e9760f701dd06" +source = "git+https://github.com/bfoss765/rust-dashcore?rev=b5dff6de05e4a354680e5b01a54bae9e642a6ad0#b5dff6de05e4a354680e5b01a54bae9e642a6ad0" [[package]] name = "glob" @@ -4095,7 +4095,7 @@ dependencies = [ [[package]] name = "key-wallet" version = "0.45.0" -source = "git+https://github.com/dashpay/rust-dashcore?rev=70d4bf8e36057c58e02d56769a6e9760f701dd06#70d4bf8e36057c58e02d56769a6e9760f701dd06" +source = "git+https://github.com/bfoss765/rust-dashcore?rev=b5dff6de05e4a354680e5b01a54bae9e642a6ad0#b5dff6de05e4a354680e5b01a54bae9e642a6ad0" dependencies = [ "aes", "async-trait", @@ -4124,7 +4124,7 @@ dependencies = [ [[package]] name = "key-wallet-ffi" version = "0.45.0" -source = "git+https://github.com/dashpay/rust-dashcore?rev=70d4bf8e36057c58e02d56769a6e9760f701dd06#70d4bf8e36057c58e02d56769a6e9760f701dd06" +source = "git+https://github.com/bfoss765/rust-dashcore?rev=b5dff6de05e4a354680e5b01a54bae9e642a6ad0#b5dff6de05e4a354680e5b01a54bae9e642a6ad0" dependencies = [ "cbindgen 0.29.4", "dash-network", @@ -4140,7 +4140,7 @@ dependencies = [ [[package]] name = "key-wallet-manager" version = "0.45.0" -source = "git+https://github.com/dashpay/rust-dashcore?rev=70d4bf8e36057c58e02d56769a6e9760f701dd06#70d4bf8e36057c58e02d56769a6e9760f701dd06" +source = "git+https://github.com/bfoss765/rust-dashcore?rev=b5dff6de05e4a354680e5b01a54bae9e642a6ad0#b5dff6de05e4a354680e5b01a54bae9e642a6ad0" dependencies = [ "async-trait", "bincode", diff --git a/Cargo.toml b/Cargo.toml index 86e5432b7e..4918ad3ba7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -52,14 +52,14 @@ members = [ ] [workspace.dependencies] -dashcore = { git = "https://github.com/dashpay/rust-dashcore", rev = "70d4bf8e36057c58e02d56769a6e9760f701dd06" } -dash-network-seeds = { git = "https://github.com/dashpay/rust-dashcore", rev = "70d4bf8e36057c58e02d56769a6e9760f701dd06" } -dash-spv = { git = "https://github.com/dashpay/rust-dashcore", rev = "70d4bf8e36057c58e02d56769a6e9760f701dd06" } -key-wallet = { git = "https://github.com/dashpay/rust-dashcore", rev = "70d4bf8e36057c58e02d56769a6e9760f701dd06" } -key-wallet-ffi = { git = "https://github.com/dashpay/rust-dashcore", rev = "70d4bf8e36057c58e02d56769a6e9760f701dd06" } -key-wallet-manager = { git = "https://github.com/dashpay/rust-dashcore", rev = "70d4bf8e36057c58e02d56769a6e9760f701dd06" } -dash-network = { git = "https://github.com/dashpay/rust-dashcore", rev = "70d4bf8e36057c58e02d56769a6e9760f701dd06" } -dashcore-rpc = { git = "https://github.com/dashpay/rust-dashcore", rev = "70d4bf8e36057c58e02d56769a6e9760f701dd06" } +dashcore = { git = "https://github.com/bfoss765/rust-dashcore", rev = "b5dff6de05e4a354680e5b01a54bae9e642a6ad0" } +dash-network-seeds = { git = "https://github.com/bfoss765/rust-dashcore", rev = "b5dff6de05e4a354680e5b01a54bae9e642a6ad0" } +dash-spv = { git = "https://github.com/bfoss765/rust-dashcore", rev = "b5dff6de05e4a354680e5b01a54bae9e642a6ad0" } +key-wallet = { git = "https://github.com/bfoss765/rust-dashcore", rev = "b5dff6de05e4a354680e5b01a54bae9e642a6ad0" } +key-wallet-ffi = { git = "https://github.com/bfoss765/rust-dashcore", rev = "b5dff6de05e4a354680e5b01a54bae9e642a6ad0" } +key-wallet-manager = { git = "https://github.com/bfoss765/rust-dashcore", rev = "b5dff6de05e4a354680e5b01a54bae9e642a6ad0" } +dash-network = { git = "https://github.com/bfoss765/rust-dashcore", rev = "b5dff6de05e4a354680e5b01a54bae9e642a6ad0" } +dashcore-rpc = { git = "https://github.com/bfoss765/rust-dashcore", rev = "b5dff6de05e4a354680e5b01a54bae9e642a6ad0" } tokio-metrics = "0.5" 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 205fe225ba..faad766503 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 @@ -381,6 +381,14 @@ internal object WalletManagerNative { external fun identitySyncStop(managerHandle: Long) external fun identitySyncIsRunning(managerHandle: Long): Boolean + /** + * Whether the durable sync watermark has been frozen this session because + * persistence events were dropped or a store was rejected — the persisted + * `syncedHeight` is held behind the chain tip and a rescan is pending on + * the next launch. Latches for the process lifetime. + */ + external fun syncFaultDetected(managerHandle: Long): Boolean + /** Shielded loop — only present when the native library is built with shielded. */ external fun shieldedSyncStart(managerHandle: Long) external fun shieldedSyncStop(managerHandle: Long) 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 940a9b7963..8bed27da94 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 @@ -1120,6 +1120,21 @@ class PlatformWalletManager( mapNativeErrors { WalletManagerNative.platformAddressSyncIsRunning(managerHandle) } } + /** + * Whether the native manager has frozen its durable sync watermark this + * session (dashpay/platform#4069). `true` means the wallet-event adapter + * dropped record-bearing events, or a persistence `store()` was rejected, + * so the persisted `syncedHeight` is deliberately held behind the chain + * tip and a rescan is pending on the next launch. Poll this to surface a + * hard "verification failed / rescan pending" state instead of leaving + * the fault visible only in the error logs. + * + * The flag latches: once `true` it stays `true` for the process lifetime. + */ + suspend fun syncFaultDetected(): Boolean = withContext(Dispatchers.IO) { + mapNativeErrors { WalletManagerNative.syncFaultDetected(managerHandle) } + } + /** * Reset the platform-address (BLAST) sync state — the native side of the * Sync tab's "Clear" action (#3959), port of Swift diff --git a/packages/rs-platform-wallet-ffi/src/manager.rs b/packages/rs-platform-wallet-ffi/src/manager.rs index c3aba1491b..cc1d16b0bf 100644 --- a/packages/rs-platform-wallet-ffi/src/manager.rs +++ b/packages/rs-platform-wallet-ffi/src/manager.rs @@ -179,6 +179,30 @@ pub unsafe extern "C" fn platform_wallet_manager_persistence_capabilities( PlatformWalletFFIResult::ok() } +/// Whether the manager has frozen its durable sync watermark this session +/// (dashpay/platform#4069). +/// +/// `true` means the wallet-event adapter dropped record-bearing events (a +/// broadcast lag) or had a persistence `store()` rejected, so the persisted +/// `syncedHeight` is deliberately held behind the chain tip and a rescan is +/// pending on the next launch. Hosts poll this to surface a hard +/// "verification failed / rescan pending" state instead of the fault being +/// visible only in error logs. +/// +/// The flag latches: once `true` it stays `true` for the process lifetime. +#[no_mangle] +pub unsafe extern "C" fn platform_wallet_manager_sync_fault_detected( + handle: Handle, + out_detected: *mut bool, +) -> PlatformWalletFFIResult { + check_ptr!(out_detected); + + let option = + PLATFORM_WALLET_MANAGER_STORAGE.with_item(handle, |manager| manager.sync_fault_detected()); + *out_detected = unwrap_option_or_return!(option); + PlatformWalletFFIResult::ok() +} + /// Map the C `has_x: bool` + `x` companion-pair idiom to a Rust `Option`. /// /// `has == true` yields `Some(value)` — including `Some(0)`, kept distinct diff --git a/packages/rs-platform-wallet/Cargo.toml b/packages/rs-platform-wallet/Cargo.toml index 6c7e86a5b8..d1c6dda282 100644 --- a/packages/rs-platform-wallet/Cargo.toml +++ b/packages/rs-platform-wallet/Cargo.toml @@ -34,14 +34,8 @@ tokio = { version = "1", features = ["sync", "rt", "time", "macros"] } tokio-util = { version = "0.7.12" } dash-async = { path = "../rs-dash-async" } -# Logging. `log` sits alongside `tracing` for on-device (Android) -# diagnostics: the JNI layer installs `android_logger` as the global `log` -# logger (logcat tag `DashSDK`), while the only `tracing` subscriber the -# Kotlin SDK installs (`dash_sdk_enable_logging`) writes to stdout, which -# Android discards — so breadcrumbs that must be visible in logcat are -# emitted through BOTH facades. See `network/encrypted_document.rs`. +# Logging tracing = "0.1" -log = "0.4" # Encoding hex = "0.4" @@ -56,6 +50,18 @@ image = { version = "0.25", default-features = false, features = ["png", "jpeg", # Security zeroize = "1" +# `log` facade. `changeset/core_bridge.rs` emits watermark-freeze breadcrumbs +# through `log` so they reach Android logcat (the JNI layer installs +# `android_logger` as the global `log` logger, tag `DashSDK`; the Kotlin SDK's +# only `tracing` subscriber writes to stdout, which Android discards). +# Declared here — deliberately NOT inside the `tracing` block above — so this +# crate owns the dependency independently of the encrypted-txMetadata change +# (#4277) that first introduced a `log` line: that change was reverted on +# v4.2-dev (#4279), and a `log` line living in that reverted region gets +# dropped by the 3-way merge, leaving the `log::` calls in core_bridge.rs +# undeclared (E0433). Keeping it in this untouched region makes it survive. +log = "0.4" + # Shielded pool (optional, behind `shielded` feature) grovedb-commitment-tree = { git = "https://github.com/dashpay/grovedb", tag = "v5.0.1", optional = true } # Direct `rusqlite` access so `FileBackedShieldedStore::open_path` can set diff --git a/packages/rs-platform-wallet/src/changeset/core_bridge.rs b/packages/rs-platform-wallet/src/changeset/core_bridge.rs index b700e73dca..df61687c74 100644 --- a/packages/rs-platform-wallet/src/changeset/core_bridge.rs +++ b/packages/rs-platform-wallet/src/changeset/core_bridge.rs @@ -1,21 +1,29 @@ //! Adapter that turns upstream `WalletEvent`s into `PlatformWalletChangeSet`s. //! -//! Upstream `key_wallet_manager::WalletManager` exposes a -//! `broadcast::Sender` and a `subscribe_events()` accessor -//! returning a `broadcast::Receiver`; consumers attach at -//! startup and drain the stream. [`spawn_wallet_event_adapter`] is the -//! platform-wallet-side consumer: a tokio task that pulls events off -//! that broadcast, projects each one into a +//! Upstream `key_wallet_manager::WalletManager` exposes a dedicated, +//! **unbounded** `mpsc` persistence channel (drained via +//! `take_persistence_receiver()`) that carries every `WalletEvent` to this +//! single durable-persistence consumer losslessly and in order. +//! [`spawn_wallet_event_adapter`] is that consumer: a tokio task that pulls +//! events off the channel, projects each one into a //! [`CoreChangeSet`](crate::changeset::CoreChangeSet), wraps it in a //! [`PlatformWalletChangeSet`](crate::changeset::PlatformWalletChangeSet), //! and forwards to the [`PlatformWalletPersistence`] sink. //! -//! # Why a single subscriber, not per-wallet +//! The manager keeps a separate, *bounded and lossy* `broadcast` bus for its +//! incidental subscribers (dash-spv's `EventHandler` fan-out, tests). This +//! consumer deliberately does NOT use that broadcast: under a heavy SPV +//! catch-up the broadcast ring overflows (`RecvError::Lagged`) and drops the +//! record/watermark events, which let the durable sync height outrun the +//! rows it implies and freeze forever (dashpay/platform#4069). The unbounded +//! persistence channel can never `Lagged`, so that freeze cannot occur. //! -//! The broadcast channel emits every event for every wallet. Each +//! # Why a single consumer, not per-wallet +//! +//! The persistence channel carries every event for every wallet. Each //! event already carries a `wallet_id`, which the adapter forwards //! verbatim to [`PlatformWalletPersistence::store`] — no need to fan -//! out a subscriber per wallet. +//! out a consumer per wallet. //! //! # Lifetime //! @@ -37,35 +45,61 @@ use key_wallet::transaction_checking::transaction_router::AccountTypeToCheck; use key_wallet::transaction_checking::{DerivedAddressInfo, TransactionContext}; use key_wallet::Utxo; use key_wallet_manager::{WalletEvent, WalletId, WalletManager}; -use tokio::sync::broadcast; -use tokio::sync::broadcast::error::RecvError; +use tokio::sync::mpsc; +use tokio::sync::mpsc::error::TryRecvError; use tokio::sync::RwLock; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use crate::changeset::changeset::{CoreChangeSet, HighestUsedIndexes, PlatformWalletChangeSet}; +use crate::changeset::merge::Merge; use crate::changeset::traits::PlatformWalletPersistence; use crate::wallet::platform_wallet::PlatformWalletInfo; +/// Maximum number of `WalletEvent`s folded into a single +/// `persister.store(..)` round-trip by [`run_wallet_event_adapter`]. +/// +/// # Why batch at all (dashpay/platform#4069 follow-up) +/// +/// This adapter's per-event cost is overwhelmingly the persister call: on +/// Android that is a JNI hop into a Room transaction (milliseconds), while +/// projecting a `WalletEvent` into a [`CoreChangeSet`] is microseconds. +/// Storing one event per `store()` therefore pinned the drain rate at +/// roughly the *store* rate — low hundreds per second — which is far below +/// what a historical SPV catch-up emits. Draining the lossless persistence +/// channel one store at a time let its backlog grow without bound during a +/// catch-up (and, on the old bounded broadcast this consumer used to read, +/// overflowed the ring and froze the watermark — the root cause the +/// dedicated unbounded channel removes). +/// +/// Folding every event *already buffered* in the channel into one changeset +/// per wallet collapses a burst of N events into a single store, so the +/// drain keeps pace with the producer at projection speed. This is +/// exactly the fold [`Merge`] was specified for — `CoreChangeSet` merging +/// is commutative and associative, and its doc comment already anticipates +/// "a flush can fold multiple events together (TransactionDetected + +/// BlockProcessed for the same wallet over a sync round)". +/// +/// The cap bounds the worst-case size of a single merged changeset (and +/// hence one Room transaction), and keeps a saturated producer from +/// starving the cancellation branch of the select below. +const ADAPTER_STORE_BATCH_LIMIT: usize = 512; + /// Session fault state for the durable-watermark guard /// (dashpay/platform#4069). /// -/// Faults are scoped as narrowly as the triggering signal allows: +/// Now that the persistence channel is a lossless unbounded `mpsc`, the only +/// remaining fault trigger is a **`store()` rejection**, which carries a +/// `wallet_id`, so only THAT wallet's watermark freezes. A sibling wallet +/// whose rows are still landing atomically keeps advancing — freezing it too +/// would force a redundant rescan of a wallet that never lost a row. /// -/// - A **`store()` rejection** carries a `wallet_id`, so only THAT -/// wallet's watermark freezes. A sibling wallet whose rows are still -/// landing atomically keeps advancing — freezing it too would force a -/// redundant rescan of a wallet that never lost a row (the CodeRabbit / -/// reviewer suggestion made concrete). -/// - A broadcast **`Lagged`** has no `wallet_id` — the dropped events -/// could have belonged to any wallet — so it freezes EVERY wallet, -/// including wallets first seen in a later event. That is modelled as a -/// single `global` latch rather than a per-wallet entry so future -/// wallet ids are covered without the adapter having to enumerate them. +/// The old global (`broadcast::Lagged`) latch is gone: the unbounded channel +/// can never `Lagged`, so there is no more "dropped events of unknown wallet" +/// signal to freeze everything for. This per-wallet freeze remains as a +/// fail-closed backstop — in a healthy run it never fires. #[derive(Default)] struct AdapterFaultState { - /// Set by a broadcast `Lagged`: freezes every wallet's watermark. - global: bool, /// Set by a `store()` rejection: freezes only the named wallet. per_wallet: HashMap, } @@ -73,7 +107,7 @@ struct AdapterFaultState { impl AdapterFaultState { /// Whether `wallet_id`'s durable watermark must be held frozen. fn is_faulted(&self, wallet_id: &WalletId) -> bool { - self.global || self.per_wallet.get(wallet_id).copied().unwrap_or(false) + self.per_wallet.get(wallet_id).copied().unwrap_or(false) } /// Fault a single wallet after its `store()` was rejected, and raise @@ -82,22 +116,92 @@ impl AdapterFaultState { self.per_wallet.insert(wallet_id, true); hard_signal.store(true, Ordering::Relaxed); } +} - /// Fault every wallet after a dropped-event broadcast lag, and raise - /// the host-visible hard-fault signal. - fn fault_all(&mut self, hard_signal: &AtomicBool) { - self.global = true; - hard_signal.store(true, Ordering::Relaxed); +/// Per-drain accounting behind the one-line batch diagnostic. +/// +/// This line is read off a tester's logcat to answer one question — *did the +/// durable sync watermark actually advance?* — so it must only ever report +/// what the persister truly accepted. A height can meet three different fates +/// in a single drain, and they are tracked separately because conflating them +/// sends a diagnosis down the wrong path: +/// +/// * `persisted` — a changeset carrying this height was handed to +/// [`PlatformWalletPersistence::store`] and it returned `Ok`. **This is the +/// only field that means the durable watermark advanced.** +/// * `frozen` — the batch proposed this height, but the fail-closed guard +/// ([`freeze_synced_height_if_faulted`]) stripped it because that wallet had +/// already faulted this session, so it was never offered to the store. +/// Without this field a held-back watermark is indistinguishable from a +/// batch that simply carried no watermark at all. +/// * `rejected` — this height *was* offered to the store and the store +/// returned an error, so the rows and the watermark are not on disk. +/// +/// Each is the monotonic max over the wallets in the drain, so a batch +/// spanning a healthy wallet and a faulted one reports both. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +struct BatchDiagnostics { + /// Events folded into this drain. + folded: usize, + /// Distinct wallets the drain produced a changeset for. + wallets: usize, + /// Highest watermark the store accepted and committed. + persisted: Option, + /// Highest watermark withheld by the fail-closed guard. + frozen: Option, + /// Highest watermark the store rejected. + rejected: Option, + /// Wallets in this drain that are faulted, plus wallets faulted *by* it. + faulted: usize, +} + +impl BatchDiagnostics { + fn new(folded: usize, wallets: usize) -> Self { + Self { + folded, + wallets, + ..Self::default() + } + } + + /// Raise `slot` to `height` if it is higher (or set it if unset). + fn raise(slot: &mut Option, height: u32) { + *slot = Some(slot.map_or(height, |cur| cur.max(height))); + } + + /// The store returned `Ok` for a changeset carrying `height`. + fn record_persisted(&mut self, height: u32) { + Self::raise(&mut self.persisted, height); + } + + /// The fail-closed guard stripped `height` before it reached the store. + fn record_frozen(&mut self, height: u32) { + Self::raise(&mut self.frozen, height); + } + + /// The store returned an error for a changeset carrying `height`. + fn record_rejected(&mut self, height: u32) { + Self::raise(&mut self.rejected, height); + } +} + +impl std::fmt::Display for BatchDiagnostics { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "wallet-event batch: folded={} wallets={} synced_height_persisted={:?} \ + synced_height_frozen={:?} synced_height_rejected={:?} faulted={}", + self.folded, self.wallets, self.persisted, self.frozen, self.rejected, self.faulted, + ) } } -/// Spawn the wallet-event subscriber task. +/// Spawn the wallet-event persistence task. /// -/// The `receiver` MUST be subscribed by the caller *before the manager is -/// published to producers* (see [`run_wallet_event_adapter`] for why) and -/// then handed to this function. This function only moves it into the -/// spawned task; it does not subscribe. Exits when `cancel` fires or the -/// upstream broadcast channel closes. +/// The `receiver` is the manager's lossless persistence receiver, taken once +/// via `take_persistence_receiver()` before the manager is published to +/// producers, and handed to this function. Exits when `cancel` fires or the +/// persistence channel's sender (the manager) is dropped. /// /// `sync_fault` is the host-visible hard-fault latch: the task sets it /// (and never clears it) the first time it freezes a durable watermark, so @@ -111,7 +215,7 @@ impl AdapterFaultState { pub fn spawn_wallet_event_adapter

( wallet_manager: Arc>>, persister: Arc

, - receiver: broadcast::Receiver, + receiver: mpsc::UnboundedReceiver, sync_fault: Arc, cancel: CancellationToken, ) -> JoinHandle<()> @@ -131,76 +235,55 @@ where /// [`CoreChangeSet`], and forward to the persister. Split out of /// [`spawn_wallet_event_adapter`] so the loop — not just the /// [`freeze_synced_height_if_faulted`] helper — is directly testable -/// (drive a real `broadcast::Sender`, inject a probe persister). +/// (drive a real `mpsc::UnboundedSender`, inject a probe persister). /// -/// # Subscribe-before-publish +/// # Lossless persistence channel (dashpay/platform#4069) /// -/// The caller subscribes and passes the `receiver` in, rather than the -/// task subscribing itself. A `tokio::broadcast::Receiver` only sees -/// messages sent *after* its `subscribe()` returns; a message sent before -/// the receiver exists is simply invisible to it — NOT reported as -/// `Lagged`. If subscription happened inside the spawned task (as it used -/// to), every event emitted between the manager being handed to producers -/// and the task's first poll would be silently lost with no fault signal. -/// Subscribing synchronously before the manager is published closes that -/// window. +/// The upstream `WalletManager` publishes `WalletEvent`s to this consumer +/// over a dedicated, **unbounded** `mpsc` persistence channel (taken once +/// via `take_persistence_receiver()`). Because it is unbounded, a burst +/// larger than any ring cannot overflow it: the consumer never observes a +/// `Lagged`, and every `TransactionDetected` / `BlockProcessed` row event +/// reaches the persister before the `SyncHeightAdvanced` watermark that +/// implies it — in the same order the manager emitted them. There is also no +/// subscribe-before-publish race: an `mpsc::UnboundedReceiver` buffers events +/// sent before the task's first poll rather than dropping them. /// -/// # Durable-watermark guard (dashpay/platform#4069) +/// This closes the historical freeze: previously this consumer read the +/// manager's *bounded* broadcast ring, and during a historical SPV catch-up +/// the manager processed blocks far faster than this single-threaded adapter +/// could drain them through the (slow, JNI + Room) persister, so the ring +/// overflowed and `recv()` returned `Lagged` — the dropped events being +/// exactly the record/UTXO/spent-marker events, while the bare +/// `SyncHeightAdvanced` watermark kept flowing and advanced the persisted +/// `syncedHeight` past blocks whose rows never reached disk. The durable +/// watermark then outran its rows and the guard below latched it frozen +/// forever. With the lossless channel that path no longer exists. /// -/// The upstream `WalletManager` publishes `WalletEvent`s onto a *bounded* -/// `tokio::broadcast` ring (capacity `DEFAULT_WALLET_EVENT_CAPACITY`, -/// 1000) via fire-and-forget `let _ = event_sender.send(..)`. During a -/// historical SPV catch-up the manager processes blocks far faster than -/// this single-threaded adapter can drain them through the (slow, JNI + -/// Room) persister, so the ring overflows and `recv()` returns -/// `RecvError::Lagged(n)` — the `n` dropped events are gone for good. -/// Those dropped events are exactly the `TransactionDetected` / -/// `BlockProcessed` records that carry the new UTXOs and the -/// spent-outpoint markers. Meanwhile the separate `SyncHeightAdvanced` -/// event (a bare height watermark, the ONLY event whose height reaches the -/// host persister's `syncedHeight` — see -/// `WalletChangeSetFFI::from_changeset`) keeps flowing and eventually -/// lands, advancing the persisted watermark past blocks whose rows never -/// made it to disk. +/// # Durable-watermark guard (fail-closed backstop) /// -/// Symptoms (all one root cause): rows dropped entirely while the -/// watermark advances (fresh scan persists nothing yet reports "scanned"); -/// spent-markers lost so consumed outputs rehydrate as spendable (inflated -/// balance); and — because the wallet's own `synced_height` is what gates -/// a rescan — the wallet believes it is fully scanned and never -/// re-matches, so the corruption is unrecoverable without deleting + -/// recreating the wallet. -/// -/// Fix: once a wallet has faulted this session (a broadcast lag OR a -/// `store()` rejection), never advance ITS persisted sync watermark again -/// (see [`AdapterFaultState`] for the per-wallet vs. global scoping). We -/// strip `synced_height` from every subsequent changeset, freezing the +/// One fault trigger remains: a rejected `store()` (the rows for that batch +/// are not on disk). When a wallet faults this way, we never advance ITS +/// persisted sync watermark again this session — [`freeze_synced_height_if_faulted`] +/// strips `synced_height` from every subsequent changeset, holding the /// durable watermark at the last height whose rows were fully committed. +/// Records/UTXO deltas in the same changeset still persist; only the height +/// advance is suppressed. On the next launch the SPV scan resumes from that +/// (lower) watermark and the persister's idempotent upserts re-apply the +/// missing rows. This is a fail-closed safety property — the durable +/// watermark never outruns the rows it implies — and in a healthy run it +/// never fires, since the channel is lossless and a `store()` rejection means +/// a genuine backend error, not overload. /// -/// This is a **fail-closed mitigation, not lossless recovery.** On the -/// next process launch the wallet restores that (lower) watermark and the -/// SPV scan resumes from it, re-emitting the dropped `BlockProcessed` -/// records; the persister's upserts are idempotent on the outpoint key, so -/// re-applying them restores the missing rows and re-marks the lost -/// spends. But the capacity-1000 producer is still fire-and-forget: if the -/// replayed catch-up again outruns this consumer, the ring lags again and -/// the watermark re-freezes. A single restart therefore resumes from the -/// last durable watermark — it does NOT *guarantee* a one-restart -/// self-heal under repeated overload; repeated overload keeps freezing -/// until the throughput mismatch is resolved (or the host acts on the -/// hard-fault signal below). The guarantee it DOES make is the safety one: -/// the durable watermark never outruns the rows it implies, so funds are -/// never silently lost or inflated — the worst case is a repeated rescan. -/// -/// When any wallet faults, the task also raises `sync_fault` (an -/// `AtomicBool` the host can poll via -/// `PlatformWalletManager::sync_fault_detected`) and logs at error level, -/// so integrators can show a hard "verification failed / rescan pending" -/// state instead of the failure being visible only in logs. +/// When a wallet faults, the task raises `sync_fault` (an `AtomicBool` the +/// host polls via `PlatformWalletManager::sync_fault_detected`) and logs a +/// one-shot `SYNC WATERMARK FROZEN` line via the `log` facade (which +/// android_logger forwards to logcat; `tracing` may not), so integrators can +/// show a hard "verification failed / rescan pending" state. async fn run_wallet_event_adapter

( wallet_manager: Arc>>, persister: Arc

, - mut receiver: broadcast::Receiver, + mut receiver: mpsc::UnboundedReceiver, sync_fault: Arc, cancel: CancellationToken, ) where @@ -208,85 +291,199 @@ async fn run_wallet_event_adapter

( { tracing::debug!("wallet-event adapter task started"); let mut fault = AdapterFaultState::default(); + // One-shot latch so the hard "watermark frozen" line hits logcat exactly + // once per session rather than once per faulted batch. + let mut freeze_logged = false; loop { - tokio::select! { - recv = receiver.recv() => { - match recv { - Ok(event) => { - let wallet_id = event.wallet_id(); - // For events that need to consult per-wallet - // state (today only `TransactionInstantLocked`, - // which checks finality before recording the IS - // lock), grab a brief read lock on the manager. - let mut core = build_core_changeset(&wallet_manager, &event).await; - // Hold this wallet's durable watermark at the last - // fully-persisted height once it has faulted (see - // the guard doc above). Records/UTXOs in this - // changeset are still persisted — only the height - // advance is suppressed. - freeze_synced_height_if_faulted(&mut core, fault.is_faulted(&wallet_id)); - if core.is_empty_no_records() { - // SyncHeightAdvanced for an unknown wallet, - // empty BlockProcessed, a watermark-only event - // stripped by the fault guard above, etc. — - // nothing to persist. Skip the round-trip. - continue; - } - let cs = PlatformWalletChangeSet { - core: Some(core), - ..PlatformWalletChangeSet::default() - }; - if let Err(e) = persister.store(wallet_id, cs) { - // A rejected changeset means these rows are not - // on disk. Fault THIS wallet's watermark so it - // can't outrun them; the next scan re-emits and - // the idempotent upserts recover the state. - fault.fault_wallet(wallet_id, &sync_fault); - tracing::error!( - wallet_id = %hex::encode(wallet_id), - error = %e, - "Persister rejected core changeset; freezing this wallet's sync watermark so the next scan re-persists the missing rows (dashpay/platform#4069)" - ); - } - } - Err(RecvError::Closed) if cancel.is_cancelled() => break, - Err(RecvError::Closed) => { - tracing::error!("WalletEvent broadcast closed unexpectedly"); - break; - } - Err(RecvError::Lagged(n)) => { - // The `n` dropped events carried record/UTXO/spend - // rows we will never see again this session, and we - // don't know which wallet(s) they belonged to. - // Fault EVERY wallet so no watermark outruns its - // rows; the next scan re-emits the lost blocks - // (dashpay/platform#4069). - fault.fault_all(&sync_fault); - tracing::error!( - missed = n, - "wallet-event adapter lagged on broadcast channel; {n} persistence events dropped — freezing every wallet's sync watermark so the next scan re-persists them (dashpay/platform#4069)" - ); - } + // Block for the first event of a batch. Everything already sitting in + // the channel behind it is folded in below without another await, so a + // burst costs one `store()` per wallet instead of one per event (see + // [`ADAPTER_STORE_BATCH_LIMIT`]). + let first = tokio::select! { + recv = receiver.recv() => recv, + _ = cancel.cancelled() => break, + }; + + // `recv()` on an mpsc returns `None` only when every sender (the + // manager) has been dropped — the lossless channel has no `Lagged`. + let Some(event) = first else { + if !cancel.is_cancelled() { + tracing::error!("WalletEvent persistence channel closed unexpectedly"); + } + break; + }; + + let mut batch: BTreeMap = BTreeMap::new(); + let mut closed = false; + { + let wallet_id = event.wallet_id(); + // For events that need to consult per-wallet state (today only + // `TransactionInstantLocked`, which checks finality before + // recording the IS lock), `build_core_changeset` takes a brief + // read lock on the manager. + let core = build_core_changeset(&wallet_manager, &event).await; + batch.entry(wallet_id).or_default().merge(core); + } + + // Fold in whatever else is already buffered. `try_recv` never waits, + // so this drains the backlog at projection speed and stops as soon as + // the channel is empty. + let mut folded = 1usize; + while folded < ADAPTER_STORE_BATCH_LIMIT { + match receiver.try_recv() { + Ok(event) => { + let wallet_id = event.wallet_id(); + let core = build_core_changeset(&wallet_manager, &event).await; + batch.entry(wallet_id).or_default().merge(core); + folded += 1; + } + Err(TryRecvError::Empty) => break, + Err(TryRecvError::Disconnected) => { + closed = true; + break; } } - _ = cancel.cancelled() => break, + } + + // Commit the folded batch. The channel is lossless, so the only way a + // watermark is held back is a rejected `store()` (the fail-closed + // backstop inside `commit_batch`). + let diag = commit_batch( + &*persister, + batch, + folded, + &mut fault, + &sync_fault, + &mut freeze_logged, + ); + + // One structured line per drain via the `log` facade so a tester + // logcat is unambiguous about whether the watermark is advancing. + // Every field reports an observed outcome — see [`BatchDiagnostics`]. + log::info!("{}", diag); + + if closed { + if !cancel.is_cancelled() { + tracing::error!("WalletEvent persistence channel closed unexpectedly"); + } + break; } } tracing::debug!("wallet-event adapter task exiting"); } +/// Commit one folded drain to the persister and report what actually happened. +/// +/// Split out of [`run_wallet_event_adapter`] so the batch diagnostic — the line +/// we read off a tester's logcat to decide whether the durable watermark is +/// advancing — is directly unit-testable against a real `store()` rejection, +/// without the async channel plumbing. The returned [`BatchDiagnostics`] is +/// what the caller logs. +/// +/// Ordering matters and is load-bearing: +/// +/// 1. Apply [`freeze_synced_height_if_faulted`] *after* the fold, so a +/// `synced_height` that entered via `Merge` is stripped just like a +/// standalone one (otherwise folding would smuggle a watermark past the +/// guard and reintroduce dashpay/platform#4069). +/// 2. Record `frozen` from the height the batch *proposed*, captured before the +/// guard strips it. +/// 3. Record `persisted` only from the `Ok` arm of `store()`. A rejected store +/// means the rows never reached disk — the same condition that makes us +/// fault the wallet — so counting it as persisted would make the trace +/// contradict itself. +fn commit_batch

( + persister: &P, + batch: BTreeMap, + folded: usize, + fault: &mut AdapterFaultState, + sync_fault: &AtomicBool, + freeze_logged: &mut bool, +) -> BatchDiagnostics +where + P: PlatformWalletPersistence + ?Sized, +{ + let mut diag = BatchDiagnostics::new(folded, batch.len()); + for (wallet_id, mut core) in batch { + // Hold this wallet's durable watermark at the last fully persisted + // height once it has faulted. Records/UTXOs still persist — only the + // height advance is suppressed. + let is_faulted = fault.is_faulted(&wallet_id); + if is_faulted { + diag.faulted += 1; + } + // Capture what this batch PROPOSED before the guard can strip it, so a + // withheld watermark is reported as frozen instead of silently reading + // as "this batch carried no watermark". + let proposed_height = core.synced_height; + freeze_synced_height_if_faulted(&mut core, is_faulted); + if is_faulted { + if let Some(h) = proposed_height { + diag.record_frozen(h); + } + } + if core.is_empty_no_records() { + // SyncHeightAdvanced for an unknown wallet, empty BlockProcessed, a + // watermark-only batch stripped by the fault guard above, etc. — + // nothing to persist. Skip the round-trip. + continue; + } + // The height this changeset OFFERS to the store. It is counted as + // persisted only in the `Ok` arm below. + let offered_height = core.synced_height; + let cs = PlatformWalletChangeSet { + core: Some(core), + ..PlatformWalletChangeSet::default() + }; + match persister.store(wallet_id, cs) { + Ok(()) => { + if let Some(h) = offered_height { + diag.record_persisted(h); + } + } + Err(e) => { + // A rejected changeset means these rows are not on disk. Fault + // THIS wallet's watermark so it can't outrun them; the next + // scan re-emits and the idempotent upserts recover the state. + if let Some(h) = offered_height { + diag.record_rejected(h); + } + fault.fault_wallet(wallet_id, sync_fault); + diag.faulted += 1; + // One-shot, unambiguous logcat marker via the `log` facade + // (android_logger forwards `log` to logcat; `tracing` may not). + if !*freeze_logged { + *freeze_logged = true; + log::error!( + "SYNC WATERMARK FROZEN: persister rejected a changeset for wallet {} ({}); \ + its durable sync height is now held so the next scan re-persists the \ + missing rows (dashpay/platform#4069). syncFaultDetected() is latched.", + hex::encode(wallet_id), + e + ); + } + tracing::error!( + wallet_id = %hex::encode(wallet_id), + error = %e, + "Persister rejected core changeset; freezing this wallet's sync watermark so the next scan re-persists the missing rows (dashpay/platform#4069)" + ); + } + } + } + diag +} + /// Durable-watermark guard for dashpay/platform#4069. /// -/// When a wallet has faulted this session (a broadcast lag dropped -/// record-bearing events, or a `store()` was rejected), its persisted -/// `synced_height` watermark must not advance past the last height whose -/// rows were fully committed — otherwise the wallet believes it is +/// When a wallet has faulted this session (a `store()` was rejected), its +/// persisted `synced_height` watermark must not advance past the last height +/// whose rows were fully committed — otherwise the wallet believes it is /// scanned and never re-matches the blocks whose rows were lost. This /// strips ONLY `synced_height`; every other field (records, UTXO /// deltas, `last_processed_height`, chain-lock) is left intact so /// in-flight rows still persist. Factored out as a pure function so the -/// invariant is unit-testable without the async broadcast plumbing. +/// invariant is unit-testable without the async channel plumbing. fn freeze_synced_height_if_faulted(core: &mut CoreChangeSet, persistence_faulted: bool) { if persistence_faulted { core.synced_height = None; @@ -965,10 +1162,11 @@ mod tests { } /// dashpay/platform#4069 (per-wallet fault scoping): a `store()` - /// rejection freezes ONLY the named wallet; a broadcast `Lagged` - /// (no wallet id) freezes every wallet, including one first seen later. + /// rejection freezes ONLY the named wallet; a sibling keeps advancing. + /// (The old global `broadcast::Lagged` latch is gone — the lossless + /// unbounded persistence channel can never lag.) #[test] - fn fault_state_scopes_store_rejection_per_wallet_and_lag_globally() { + fn fault_state_scopes_store_rejection_per_wallet() { let a = [0xAAu8; 32]; let b = [0xBBu8; 32]; let signal = std::sync::atomic::AtomicBool::new(false); @@ -984,25 +1182,15 @@ mod tests { signal.load(std::sync::atomic::Ordering::Relaxed), "hard-fault signal must be raised on a store rejection" ); - - // A broadcast lag freezes everything, including a never-before-seen - // wallet id. - let mut fault2 = AdapterFaultState::default(); - let c = [0xCCu8; 32]; - fault2.fault_all(&signal); - assert!( - fault2.is_faulted(&c), - "lag must freeze every wallet, even future ids" - ); } // ── Adapter-loop integration tests (dashpay/platform#4069) ── // - // These drive `run_wallet_event_adapter` with a real `broadcast` - // channel and a probe persister so the LOOP — not just the - // `freeze_synced_height_if_faulted` helper — is exercised: actual - // `RecvError::Lagged`, a rejected `store()`, the per-wallet freeze, - // and subscribe-before-publish timing. + // These drive `run_wallet_event_adapter` with a real lossless + // `mpsc::UnboundedSender` and a probe persister so the LOOP — not just + // the `freeze_synced_height_if_faulted` helper — is exercised: a large + // lossless burst, a rejected `store()`, the per-wallet freeze, and + // per-wallet batch folding. use super::{run_wallet_event_adapter, AdapterFaultState}; use crate::changeset::changeset::PlatformWalletChangeSet; @@ -1014,7 +1202,6 @@ mod tests { use std::collections::{BTreeMap, HashSet}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; - use tokio::sync::broadcast; use tokio::sync::mpsc::{unbounded_channel, UnboundedSender}; use tokio::sync::RwLock; use tokio_util::sync::CancellationToken; @@ -1112,16 +1299,19 @@ mod tests { } } - /// (b) A real `RecvError::Lagged` flips the fault latch, and a - /// subsequent watermark-only event is stripped (never delivered as a - /// store), while a record-bearing event still persists. + /// (b) THE ROOT-CAUSE FIX: a burst far larger than the old bounded ring + /// (`DEFAULT_WALLET_EVENT_CAPACITY` == 1000) is delivered losslessly over + /// the unbounded persistence channel, so the fault latch never trips and + /// the durable watermark advances all the way to the tip of the + /// catch-up. On the old broadcast this burst would `Lagged` and freeze the + /// watermark forever (dashpay/platform#4069). #[tokio::test] - async fn lagged_broadcast_freezes_and_strips_subsequent_watermark() { + async fn lossless_burst_never_freezes_and_watermark_reaches_tip() { + const BURST: u32 = 3000; // >> the old broadcast ring (1000) let wallet_id = [7u8; 32]; - // Capacity 2, send 4 before the adapter drains → the receiver's - // first recv() observes Lagged(2). Ring retains heights 3,4. - let (tx, rx) = broadcast::channel::(2); - for h in 1..=4u32 { + let (tx, rx) = unbounded_channel::(); + // Pre-buffer the whole burst; an unbounded mpsc cannot drop it. + for h in 1..=BURST { tx.send(sync_height_event(wallet_id, h)).unwrap(); } @@ -1137,35 +1327,34 @@ mod tests { cancel.clone(), )); - // Post-lag watermark: must be stripped (never stored). - tx.send(sync_height_event(wallet_id, 100)).unwrap(); - // Record-bearing sentinel: proves the loop advanced past every - // watermark AND that records still persist after the freeze. - tx.send(block_processed_event(wallet_id, 50)).unwrap(); + // Close the channel so the adapter drains the entire burst (across as + // many batches as `ADAPTER_STORE_BATCH_LIMIT` requires) and exits. + drop(tx); + handle.await.unwrap(); - let observed = obs_rx.recv().await.expect("sentinel store must arrive"); - assert_eq!(observed.wallet_id, wallet_id); - assert_eq!( - observed.synced_height, None, - "watermark must be stripped after a broadcast lag" - ); - assert_eq!( - observed.last_processed_height, - Some(50), - "record-bearing field must persist after the freeze" - ); - assert!( - obs_rx.try_recv().is_err(), - "no watermark-only store should have been delivered after the lag" - ); assert!( - sync_fault.load(Ordering::Relaxed), - "hard sync-fault signal must be raised on a lag" + !sync_fault.load(Ordering::Relaxed), + "a lossless burst must never raise the freeze signal" ); - cancel.cancel(); - drop(tx); - handle.await.unwrap(); + let mut max_synced = 0u32; + let mut any_store = false; + while let Ok(observed) = obs_rx.try_recv() { + any_store = true; + assert_eq!(observed.wallet_id, wallet_id); + assert!( + !observed.rejected, + "no store should be rejected on the happy path" + ); + if let Some(h) = observed.synced_height { + max_synced = max_synced.max(h); + } + } + assert!(any_store, "the burst must have produced at least one store"); + assert_eq!( + max_synced, BURST, + "the durable watermark must advance to the tip across the whole catch-up" + ); } /// (c) A rejected `store()` faults the wallet, and the very next @@ -1173,7 +1362,7 @@ mod tests { #[tokio::test] async fn rejected_store_freezes_wallet_and_strips_watermark() { let wallet_id = [9u8; 32]; - let (tx, rx) = broadcast::channel::(16); + let (tx, rx) = unbounded_channel::(); let (obs_tx, mut obs_rx) = unbounded_channel(); let persister = Arc::new(ProbePersister::new(obs_tx)); persister.fail_next(wallet_id); // first store() for this wallet is rejected @@ -1224,7 +1413,7 @@ mod tests { #[tokio::test] async fn record_bearing_changesets_persist_after_guard_activates() { let wallet_id = [4u8; 32]; - let (tx, rx) = broadcast::channel::(16); + let (tx, rx) = unbounded_channel::(); let (obs_tx, mut obs_rx) = unbounded_channel(); let persister = Arc::new(ProbePersister::new(obs_tx)); persister.fail_next(wallet_id); // activate the guard via a rejection @@ -1261,14 +1450,16 @@ mod tests { handle.await.unwrap(); } - /// (e) Subscribe-before-publish: an event emitted BEFORE the adapter - /// task is spawned (i.e. before it polls) is still delivered, because - /// the receiver was subscribed up-front and handed to the task. This is - /// the exact invariant the manager now relies on. + /// (e) No startup race: an event emitted BEFORE the adapter task is + /// spawned (i.e. before it polls) is still delivered, because an + /// `mpsc::UnboundedReceiver` buffers messages sent before the first + /// `recv()` rather than dropping them (unlike a `broadcast::Receiver`, + /// which only sees messages sent after it subscribes). This is the exact + /// invariant the manager relies on. #[tokio::test] async fn events_emitted_before_task_poll_are_received() { let wallet_id = [3u8; 32]; - let (tx, rx) = broadcast::channel::(16); + let (tx, rx) = unbounded_channel::(); // Emit BEFORE the task is spawned / polls. tx.send(block_processed_event(wallet_id, 77)).unwrap(); @@ -1300,4 +1491,465 @@ mod tests { drop(tx); handle.await.unwrap(); } + + /// (e) Events already buffered in the ring are folded into a SINGLE + /// `store()` per wallet, with the watermark taking the monotonic max. + /// + /// This is the throughput property that keeps the ring from + /// overflowing in the first place: the adapter's cost is one + /// (JNI + Room) store per *batch*, not per event. Every event is + /// published before the task is spawned, so the first `recv()` sees + /// event 1 and the `try_recv()` drain folds in 2..=5 without ever + /// awaiting — making the batch boundary deterministic. + #[tokio::test] + async fn buffered_events_fold_into_one_store_per_wallet() { + let wallet_id = [11u8; 32]; + // Comfortably larger than the burst, so nothing is dropped. + let (tx, rx) = unbounded_channel::(); + for h in 1..=5u32 { + tx.send(sync_height_event(wallet_id, h)).unwrap(); + } + + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = Arc::new(ProbePersister::new(obs_tx)); + let sync_fault = Arc::new(AtomicBool::new(false)); + let cancel = CancellationToken::new(); + let handle = tokio::spawn(run_wallet_event_adapter( + test_manager(), + Arc::clone(&persister), + rx, + Arc::clone(&sync_fault), + cancel.clone(), + )); + + let observed = obs_rx.recv().await.expect("merged store must arrive"); + assert_eq!(observed.wallet_id, wallet_id); + assert_eq!( + observed.synced_height, + Some(5), + "folded watermark must be the monotonic max of the batch" + ); + assert!( + !sync_fault.load(Ordering::Relaxed), + "a clean batch must not raise the fault signal" + ); + + cancel.cancel(); + drop(tx); + handle.await.unwrap(); + + // Exactly one store for the whole burst — five events, one + // round-trip. Drained after the task has exited so no further + // store can still be in flight. + assert!( + obs_rx.try_recv().is_err(), + "the buffered burst must collapse into a single store" + ); + } + + /// (f) Folding is scoped per wallet: a batch carrying events for two + /// wallets produces one store each, correctly attributed. Merging + /// across wallets would mis-persist one wallet's rows under the + /// other's id. + #[tokio::test] + async fn batch_folds_per_wallet_not_across_wallets() { + let wallet_a = [1u8; 32]; + let wallet_b = [2u8; 32]; + let (tx, rx) = unbounded_channel::(); + // Interleaved on purpose. + tx.send(sync_height_event(wallet_a, 10)).unwrap(); + tx.send(sync_height_event(wallet_b, 20)).unwrap(); + tx.send(sync_height_event(wallet_a, 11)).unwrap(); + tx.send(sync_height_event(wallet_b, 21)).unwrap(); + + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = Arc::new(ProbePersister::new(obs_tx)); + let sync_fault = Arc::new(AtomicBool::new(false)); + let cancel = CancellationToken::new(); + let handle = tokio::spawn(run_wallet_event_adapter( + test_manager(), + Arc::clone(&persister), + rx, + Arc::clone(&sync_fault), + cancel.clone(), + )); + + let mut heights: BTreeMap> = BTreeMap::new(); + for _ in 0..2 { + let observed = obs_rx.recv().await.expect("both wallets must store"); + heights.insert(observed.wallet_id, observed.synced_height); + } + + assert_eq!(heights.get(&wallet_a), Some(&Some(11))); + assert_eq!(heights.get(&wallet_b), Some(&Some(21))); + + cancel.cancel(); + drop(tx); + handle.await.unwrap(); + assert!( + obs_rx.try_recv().is_err(), + "one store per wallet, not per event" + ); + } + + /// (g) SAFETY INVARIANT under a fault: once the per-wallet fault latch is + /// set (here by a rejected `store()`), no later changeset for that wallet + /// may advance the durable `synced_height` — whether the watermark arrives + /// standalone or folded together with a record. The freeze is applied + /// after the fold, so a `synced_height` that entered via `Merge` is + /// stripped just like a standalone one; otherwise folding would smuggle + /// the watermark past the guard and reintroduce dashpay/platform#4069 + /// (durable watermark outrunning the rows it implies). + #[tokio::test] + async fn watermark_is_still_stripped_after_a_fault() { + let wallet_id = [9u8; 32]; + let (tx, rx) = unbounded_channel::(); + + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = Arc::new(ProbePersister::new(obs_tx)); + // Fault the wallet via a rejected store (the only remaining trigger + // now that the lossless channel can't lag). + persister.fail_next(wallet_id); + let sync_fault = Arc::new(AtomicBool::new(false)); + let cancel = CancellationToken::new(); + let handle = tokio::spawn(run_wallet_event_adapter( + test_manager(), + Arc::clone(&persister), + rx, + Arc::clone(&sync_fault), + cancel.clone(), + )); + + // Trip the fault and wait until it has latched. + tx.send(block_processed_event(wallet_id, 10)).unwrap(); + let _ = obs_rx.recv().await.expect("first (rejected) store"); + while !sync_fault.load(Ordering::Relaxed) { + tokio::task::yield_now().await; + } + + // A record-bearing event and a watermark event, evaluated under the + // fault. Whether they fold into one changeset or arrive as two, the + // durable watermark must never be persisted while faulted, and the + // record must still land. Await the record-bearing store *before* + // cancelling so the cancel can't race ahead of processing it. + tx.send(block_processed_event(wallet_id, 60)).unwrap(); + tx.send(sync_height_event(wallet_id, 900)).unwrap(); + + let post = obs_rx + .recv() + .await + .expect("the record-bearing event must still persist while faulted"); + assert_eq!(post.wallet_id, wallet_id); + assert_eq!( + post.synced_height, None, + "no store may carry a synced_height once the wallet is faulted" + ); + assert_eq!( + post.last_processed_height, + Some(60), + "record-bearing fields must survive the freeze" + ); + + cancel.cancel(); + drop(tx); + handle.await.unwrap(); + + // Any further store (if the watermark arrived unfolded) must also have + // been stripped — never a bare advancing watermark. + while let Ok(observed) = obs_rx.try_recv() { + assert_eq!( + observed.synced_height, None, + "no store may carry a synced_height once the wallet is faulted" + ); + } + } + + // ── Batch-diagnostic reporting (dashpay/platform#4290 review) ── + // + // The per-drain `wallet-event batch: ...` line is read off a mainnet + // tester's logcat to answer "is the durable watermark advancing?", so what + // it reports is a tested property, not a comment. + // + // The regression these lock down: the diagnostic used to fold + // `core.synced_height` into `synced_height_persisted` BEFORE calling + // `persister.store(...)`. A rejected store therefore logged + // `synced_height_persisted=Some(h)` in the very same drain that faulted the + // wallet *because* height `h`'s rows were not accepted — an internally + // contradictory trace that points a diagnosis at the wrong subsystem. + // + // These drive the real `commit_batch` (the production commit path, + // including the fail-closed guard) so the assertions cover the shipped + // code, not a restatement of it. + + use super::{commit_batch, BatchDiagnostics}; + + /// A changeset that both proposes a watermark and carries a record-bearing + /// field, so it survives `is_empty_no_records()` and actually reaches + /// `store()`. + fn watermark_with_rows(synced: u32, processed: u32) -> CoreChangeSet { + CoreChangeSet { + synced_height: Some(synced), + last_processed_height: Some(processed), + ..CoreChangeSet::default() + } + } + + fn one_wallet_batch( + wallet_id: WalletId, + core: CoreChangeSet, + ) -> BTreeMap { + let mut batch = BTreeMap::new(); + batch.insert(wallet_id, core); + batch + } + + /// Baseline: a height the store ACCEPTED is the one case that may be + /// reported as persisted. + #[test] + fn accepted_store_reports_the_watermark_as_persisted() { + let wallet_id = [1u8; 32]; + let (obs_tx, _obs_rx) = unbounded_channel(); + let persister = ProbePersister::new(obs_tx); + let sync_fault = AtomicBool::new(false); + let mut fault = AdapterFaultState::default(); + let mut freeze_logged = false; + + let diag = commit_batch( + &persister, + one_wallet_batch(wallet_id, watermark_with_rows(500, 500)), + 1, + &mut fault, + &sync_fault, + &mut freeze_logged, + ); + + assert_eq!(diag.persisted, Some(500)); + assert_eq!(diag.frozen, None); + assert_eq!(diag.rejected, None); + assert_eq!(diag.faulted, 0); + assert!(!sync_fault.load(Ordering::Relaxed)); + assert!(diag + .to_string() + .contains("synced_height_persisted=Some(500)")); + } + + /// REGRESSION (PR #4290 review): a REJECTED `store()` must never be + /// reported as persisted. + #[test] + fn rejected_store_is_not_reported_as_persisted() { + let wallet_id = [9u8; 32]; + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = ProbePersister::new(obs_tx); + persister.fail_next(wallet_id); + let sync_fault = AtomicBool::new(false); + let mut fault = AdapterFaultState::default(); + let mut freeze_logged = false; + + let diag = commit_batch( + &persister, + one_wallet_batch(wallet_id, watermark_with_rows(500, 500)), + 1, + &mut fault, + &sync_fault, + &mut freeze_logged, + ); + + // The height was genuinely offered to the store... + let observed = obs_rx.try_recv().expect("the rejected store still ran"); + assert!(observed.rejected); + assert_eq!(observed.synced_height, Some(500)); + + // ...and the store rejected it, so it is NOT on disk. + assert_eq!( + diag.persisted, None, + "a rejected watermark must never be counted as persisted" + ); + assert_eq!( + diag.rejected, + Some(500), + "the rejected height belongs under its own field" + ); + assert_eq!( + diag.frozen, None, + "the guard did not strip this one — the store rejected it" + ); + assert_eq!(diag.faulted, 1); + + // The rendered logcat line must not claim the height reached disk. + let line = diag.to_string(); + assert!( + line.contains("synced_height_persisted=None"), + "logcat line must report no persisted watermark: {line}" + ); + assert!( + !line.contains("synced_height_persisted=Some(500)"), + "logcat line must not report the rejected height as persisted: {line}" + ); + assert!( + line.contains("synced_height_rejected=Some(500)"), + "logcat line must surface the rejected height: {line}" + ); + + // The fail-closed guard itself is untouched: the wallet is faulted, the + // host-visible signal is latched, and the one-shot frozen marker fired. + assert!(fault.is_faulted(&wallet_id)); + assert!(sync_fault.load(Ordering::Relaxed)); + assert!( + freeze_logged, + "the one-shot SYNC WATERMARK FROZEN marker must have been emitted" + ); + } + + /// A watermark withheld by the fail-closed guard is reported as `frozen` — + /// never as persisted, and distinguishably from a drain that simply carried + /// no watermark at all. + #[test] + fn guard_stripped_watermark_is_reported_as_frozen_not_persisted() { + let wallet_id = [9u8; 32]; + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = ProbePersister::new(obs_tx); + let sync_fault = AtomicBool::new(false); + let mut fault = AdapterFaultState::default(); + // Pre-fault the wallet, as an earlier drain's rejection would have. + fault.fault_wallet(wallet_id, &sync_fault); + let mut freeze_logged = true; // one-shot already spent + + let diag = commit_batch( + &persister, + one_wallet_batch(wallet_id, watermark_with_rows(900, 900)), + 1, + &mut fault, + &sync_fault, + &mut freeze_logged, + ); + + assert_eq!(diag.persisted, None, "a frozen watermark is not persisted"); + assert_eq!(diag.frozen, Some(900)); + assert_eq!(diag.rejected, None); + assert_eq!(diag.faulted, 1); + + // Guard behaviour unchanged: the store never saw the height, while the + // record-bearing field still persisted. + let observed = obs_rx + .try_recv() + .expect("record-bearing rows still persist while frozen"); + assert_eq!( + observed.synced_height, None, + "the guard must strip the watermark before it reaches the store" + ); + assert_eq!(observed.last_processed_height, Some(900)); + + let line = diag.to_string(); + assert!(line.contains("synced_height_persisted=None"), "{line}"); + assert!(line.contains("synced_height_frozen=Some(900)"), "{line}"); + } + + /// A watermark-ONLY changeset under the guard is stripped to empty and + /// skips the store round-trip entirely. It must still be reported as + /// frozen, otherwise that drain reads as idle. + #[test] + fn frozen_watermark_only_batch_is_still_reported() { + let wallet_id = [9u8; 32]; + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = ProbePersister::new(obs_tx); + let sync_fault = AtomicBool::new(false); + let mut fault = AdapterFaultState::default(); + fault.fault_wallet(wallet_id, &sync_fault); + let mut freeze_logged = true; + + let core = CoreChangeSet { + synced_height: Some(1234), + ..CoreChangeSet::default() + }; + let diag = commit_batch( + &persister, + one_wallet_batch(wallet_id, core), + 1, + &mut fault, + &sync_fault, + &mut freeze_logged, + ); + + assert_eq!(diag.frozen, Some(1234)); + assert_eq!(diag.persisted, None); + assert_eq!(diag.rejected, None); + assert!( + obs_rx.try_recv().is_err(), + "a stripped watermark-only batch must skip the store round-trip" + ); + } + + /// One drain can span a healthy wallet and a rejecting one. Each outcome is + /// reported under its own field instead of collapsing into a single + /// "persisted" number that would over-report the rejecting wallet. + #[test] + fn mixed_batch_reports_persisted_and_rejected_separately() { + let healthy = [1u8; 32]; + let rejecting = [2u8; 32]; + let (obs_tx, _obs_rx) = unbounded_channel(); + let persister = ProbePersister::new(obs_tx); + persister.fail_next(rejecting); + let sync_fault = AtomicBool::new(false); + let mut fault = AdapterFaultState::default(); + let mut freeze_logged = false; + + let mut batch = BTreeMap::new(); + batch.insert(healthy, watermark_with_rows(10, 10)); + batch.insert(rejecting, watermark_with_rows(20, 20)); + + let diag = commit_batch( + &persister, + batch, + 2, + &mut fault, + &sync_fault, + &mut freeze_logged, + ); + + assert_eq!( + diag.persisted, + Some(10), + "only the healthy wallet's height landed" + ); + assert_eq!( + diag.rejected, + Some(20), + "the rejecting wallet's height did not land" + ); + assert_eq!(diag.wallets, 2); + assert_eq!(diag.folded, 2); + assert!( + !fault.is_faulted(&healthy), + "a sibling wallet must not be faulted by another's rejection" + ); + assert!(fault.is_faulted(&rejecting)); + } + + /// Each field takes its monotonic max independently, so a lower height + /// later in a drain cannot pull a reported watermark backwards and the + /// three outcomes never bleed into each other. + #[test] + fn diagnostics_fields_take_independent_monotonic_max() { + let mut diag = BatchDiagnostics::new(3, 3); + diag.record_persisted(10); + diag.record_persisted(4); // lower — must not regress + diag.record_frozen(7); + diag.record_rejected(99); + assert_eq!(diag.persisted, Some(10)); + assert_eq!(diag.frozen, Some(7)); + assert_eq!(diag.rejected, Some(99)); + } + + /// The exact logcat contract a tester greps for. + #[test] + fn diagnostic_line_format_is_stable() { + let mut diag = BatchDiagnostics::new(512, 2); + diag.record_persisted(100); + diag.record_rejected(200); + assert_eq!( + diag.to_string(), + "wallet-event batch: folded=512 wallets=2 synced_height_persisted=Some(100) \ + synced_height_frozen=None synced_height_rejected=Some(200) faulted=0" + ); + } } diff --git a/packages/rs-platform-wallet/src/manager/mod.rs b/packages/rs-platform-wallet/src/manager/mod.rs index ade7ac6e0a..249766d377 100644 --- a/packages/rs-platform-wallet/src/manager/mod.rs +++ b/packages/rs-platform-wallet/src/manager/mod.rs @@ -412,15 +412,18 @@ impl PlatformWalletManager

{ persister: Arc

, app_handler: Arc, ) -> Self { - // Subscribe to the wallet-event broadcast BEFORE the manager is - // wrapped in the shared `Arc` and handed to any producer, - // so no event emitted during startup is lost without a `Lagged` - // marker (a `broadcast::Receiver` only sees messages sent after its - // `subscribe()` — see `run_wallet_event_adapter`'s - // subscribe-before-publish note). The receiver is created here, - // synchronously, and moved into the adapter task below. - let wallet_manager_inner = WalletManager::new(sdk.network); - let event_receiver = wallet_manager_inner.subscribe_events(); + // Take the manager's lossless, unbounded persistence receiver BEFORE + // the manager is wrapped in the shared `Arc` and handed to any + // producer. Unlike the old broadcast subscription, an + // `mpsc::UnboundedReceiver` buffers events emitted during startup + // rather than dropping them, so there is no subscribe-before-publish + // race and — being unbounded — it can never `Lagged` and freeze the + // durable sync watermark (dashpay/platform#4069). The receiver is + // taken here, once, and moved into the adapter task below. + let mut wallet_manager_inner = WalletManager::new(sdk.network); + let event_receiver = wallet_manager_inner + .take_persistence_receiver() + .expect("persistence receiver is available exactly once on a fresh WalletManager"); let wallet_manager = Arc::new(RwLock::new(wallet_manager_inner)); let wallets = Arc::new(RwLock::new(std::collections::BTreeMap::new())); let lock_notify = Arc::new(Notify::new()); diff --git a/packages/rs-unified-sdk-jni/src/wallet_manager.rs b/packages/rs-unified-sdk-jni/src/wallet_manager.rs index 78c25061fa..298d5845b8 100644 --- a/packages/rs-unified-sdk-jni/src/wallet_manager.rs +++ b/packages/rs-unified-sdk-jni/src/wallet_manager.rs @@ -2224,6 +2224,39 @@ sync_start_stop!( platform_wallet_ffi::platform_wallet_manager_identity_sync_is_running ); +/// Whether the manager has frozen its durable sync watermark this session +/// (dashpay/platform#4069). `true` means the wallet-event adapter dropped +/// record-bearing events, or a persistence `store()` was rejected, so the +/// persisted `syncedHeight` is deliberately held behind the chain tip and a +/// rescan is pending on the next launch — the host should surface a hard +/// "verification failed / rescan pending" state rather than leave the fault +/// in the error logs. Latches for the process lifetime. Backs +/// `PlatformWalletManager.syncFaultDetected()`. +#[no_mangle] +pub extern "system" fn Java_org_dashfoundation_dashsdk_ffi_WalletManagerNative_syncFaultDetected( + mut env: JNIEnv, + _class: JClass, + manager_handle: jlong, +) -> jboolean { + guard(&mut env, JNI_FALSE, |env| { + let mut detected = false; + let result = unsafe { + platform_wallet_ffi::platform_wallet_manager_sync_fault_detected( + manager_handle as Handle, + &mut detected as *mut bool, + ) + }; + if take_pwffi_error(env, result) { + return JNI_FALSE; + } + if detected { + JNI_TRUE + } else { + JNI_FALSE + } + }) +} + #[cfg(feature = "shielded")] sync_start_stop!( Java_org_dashfoundation_dashsdk_ffi_WalletManagerNative_shieldedSyncStart,