From 3e8c27fbd091596111adacf0bd12075d7facfebc Mon Sep 17 00:00:00 2001 From: bfoss765 <38437574+bfoss765@users.noreply.github.com> Date: Tue, 4 Aug 2026 15:21:47 -0400 Subject: [PATCH 1/5] fix(platform-wallet): batch wallet-event persistence so the sync watermark stops freezing The wallet-event adapter issued one `persister.store(..)` per `WalletEvent`. On Android that store is a JNI hop into a Room transaction (milliseconds), while projecting an event into a `CoreChangeSet` is microseconds - so the drain rate was pinned at the store rate, a few hundred events/sec. The upstream producer publishes fire-and-forget onto a bounded broadcast ring (`DEFAULT_WALLET_EVENT_CAPACITY`, 1000), and a historical SPV catch-up outruns a consumer that slow. The ring overflows, `recv()` returns `Lagged`, and the durable-watermark guard added for dashpay/platform#4069 freezes `synced_height` for the rest of the process lifetime. That freeze is a *permanent* latch (`AdapterFaultState` has no clear path, by design). In the field it presented as a mainnet sync that climbs toward completion and then appears to "roll back" on every relaunch: the watermark froze shortly after install, so each restart resumed the filter scan from that same frozen height no matter how far the session had actually scanned. Fix the throughput mismatch rather than the guard: fold every event already buffered in the ring into one `CoreChangeSet` per wallet and issue a single store per batch. `CoreChangeSet` merging is commutative and associative and its `Merge` impl already anticipates exactly this fold ("a flush can fold multiple events together (TransactionDetected + BlockProcessed for the same wallet over a sync round)"), so this uses the existing contract rather than widening it. Drain rate becomes bounded by the ring instead of by the persister, which removes the overflow that trips the guard. The #4069 safety invariant is deliberately left intact - the durable watermark still must never outrun the rows it implies. Note the guard cannot simply be unfrozen on lag recovery: `keep-finalized-transactions` is off by default, so finalized `TransactionRecord`s are evicted from the in-memory wallet and the event channel is the *only* delivery path for that history. There is no source of truth to reconcile a dropped event against, so resuming watermark writes after a lag would reintroduce the silent fund-loss/inflation of #4069. Preventing the lag is the sound fix; the freeze remains as a fail-closed backstop. The freeze is now applied *after* the fold, so a `synced_height` that entered a changeset via `Merge` is stripped just like a standalone one - otherwise folding would smuggle the watermark past the guard. Tests: three new cases cover the fold (one store per wallet per batch, per-wallet scoping, and the post-fault strip of a merged watermark). All four pre-existing guard tests still pass unchanged; full crate suite 531/531. --- .../src/changeset/core_bridge.rs | 369 +++++++++++++++--- 1 file changed, 309 insertions(+), 60 deletions(-) diff --git a/packages/rs-platform-wallet/src/changeset/core_bridge.rs b/packages/rs-platform-wallet/src/changeset/core_bridge.rs index b700e73dcaf..40c36fd9054 100644 --- a/packages/rs-platform-wallet/src/changeset/core_bridge.rs +++ b/packages/rs-platform-wallet/src/changeset/core_bridge.rs @@ -38,15 +38,49 @@ 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::broadcast::error::{RecvError, 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. The upstream producer publishes +/// fire-and-forget onto a bounded ring (`DEFAULT_WALLET_EVENT_CAPACITY`, +/// 1000), so a consumer that slow overflows it, `recv()` returns +/// `Lagged`, and the durable-watermark guard freezes the wallet's +/// `synced_height` for the rest of the process lifetime. In the field that +/// presented as a sync that climbs toward completion and then "rolls back" +/// to near the install position on every relaunch: the watermark froze +/// after the first batch, so every later restart resumed from that same +/// frozen height. +/// +/// Folding every event *already buffered* in the ring into one changeset +/// per wallet collapses a burst of N events into a single store, so the +/// drain rate is bounded by the ring rather than by the persister. 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). /// @@ -210,67 +244,118 @@ async fn run_wallet_event_adapter

( let mut fault = AdapterFaultState::default(); 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 ring 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, + }; + + let mut batch: BTreeMap = BTreeMap::new(); + let mut missed: u64 = 0; + let mut closed = false; + + match first { + 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 core = build_core_changeset(&wallet_manager, &event).await; + batch.entry(wallet_id).or_default().merge(core); + } + Err(RecvError::Lagged(n)) => missed += n, + Err(RecvError::Closed) if cancel.is_cancelled() => break, + Err(RecvError::Closed) => { + tracing::error!("WalletEvent broadcast closed unexpectedly"); + break; + } + } + + // 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 ring 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::Lagged(n)) => { + missed += n; + folded += 1; + } + Err(TryRecvError::Empty) => break, + Err(TryRecvError::Closed) => { + closed = true; + break; } } - _ = cancel.cancelled() => break, + } + + if missed > 0 { + // The 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). + // + // Faulting before storing this batch also covers the events + // folded in *ahead* of the lag: stripping their `synced_height` + // can only hold the watermark lower, never advance it past + // uncommitted rows, so the conservative direction is the safe + // one. + fault.fault_all(&sync_fault); + tracing::error!( + missed, + "wallet-event adapter lagged on broadcast channel; {missed} persistence events dropped — freezing every wallet's sync watermark so the next scan re-persists them (dashpay/platform#4069)" + ); + } + + for (wallet_id, mut core) in batch { + // 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. Applied after the + // fold so a `synced_height` that arrived via merge is stripped + // too. + 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 batch 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)" + ); + } + } + + if closed { + if !cancel.is_cancelled() { + tracing::error!("WalletEvent broadcast closed unexpectedly"); + } + break; } } tracing::debug!("wallet-event adapter task exiting"); @@ -1300,4 +1385,168 @@ 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) = broadcast::channel::(64); + 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) = broadcast::channel::(64); + // 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 folding: once the fault latch is set, a + /// batch that merges a record-bearing event together with a watermark + /// event still persists the records but must NOT carry the merged + /// `synced_height`. + /// + /// This is the property that makes batching safe. The freeze is + /// applied after the fold, so a `synced_height` that entered the + /// changeset 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 merged_watermark_is_still_stripped_after_a_fault() { + let wallet_id = [9u8; 32]; + // Capacity 2 with 4 sends → the first recv() reports Lagged(2), + // which latches the global fault before anything is stored. + let (tx, rx) = broadcast::channel::(2); + for h in 1..=4u32 { + 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(), + )); + + // Wait until the lag has been observed and latched, so the events + // below are guaranteed to be evaluated under the fault. + while !sync_fault.load(Ordering::Relaxed) { + tokio::task::yield_now().await; + } + + // A record-bearing event and a watermark event that will fold + // into ONE changeset for this wallet. + tx.send(block_processed_event(wallet_id, 60)).unwrap(); + tx.send(sync_height_event(wallet_id, 900)).unwrap(); + + let observed = obs_rx + .recv() + .await + .expect("the record-bearing half of the batch must still persist"); + assert_eq!(observed.wallet_id, wallet_id); + assert_eq!( + observed.synced_height, None, + "a merged watermark must still be stripped while the wallet is faulted" + ); + assert_eq!( + observed.last_processed_height, + Some(60), + "record-bearing fields must survive the freeze" + ); + + cancel.cancel(); + drop(tx); + handle.await.unwrap(); + } } From e64662205b92210c47f932df928befe043c5b751 Mon Sep 17 00:00:00 2001 From: bfoss765 <38437574+bfoss765@users.noreply.github.com> Date: Tue, 4 Aug 2026 15:22:07 -0400 Subject: [PATCH 2/5] feat(ffi/jni/kotlin): expose sync_fault_detected to the host `PlatformWalletManager::sync_fault_detected()` has existed since the dashpay/platform#4069 watermark guard landed, but it stopped at the Rust boundary - nothing above Rust could see it. When the guard freezes a wallet's durable sync watermark, the only evidence was an error-level log line, so on Android a wallet whose watermark had frozen looked identical to one that was simply syncing slowly. Surface it through the existing four-layer path so the app can report "verification failed / rescan pending" instead of silently re-scanning from a stale height forever: - `platform_wallet_manager_sync_fault_detected` (C FFI, out-param + result code, mirroring `platform_wallet_manager_shielded_sync_is_syncing`) - `Java_..._WalletManagerNative_syncFaultDetected` (JNI) - `WalletManagerNative.syncFaultDetected` (Kotlin external) - `PlatformWalletManager.syncFaultDetected()` (Kotlin suspend wrapper) All four are unconditional - deliberately outside the `shielded` feature gate, since the fault is a core-persistence signal. Verified with and without default features so the symbol is emitted in both builds. No new FFI error codes (the standard null-pointer / invalid-handle macros are reused), so ERROR_CODE_REGISTRY.md is unchanged. The generated cbindgen header picks the symbol up automatically; no checked-in header or symbol list exists to update. Note the native flag latches for the process lifetime and never clears, so this is a one-shot poll rather than an observable flow - a UI that needs to react must check it at a lifecycle point. --- .../dashsdk/ffi/WalletManagerNative.kt | 8 +++++ .../dashsdk/wallet/PlatformWalletManager.kt | 15 +++++++++ .../rs-platform-wallet-ffi/src/manager.rs | 24 ++++++++++++++ .../rs-unified-sdk-jni/src/wallet_manager.rs | 33 +++++++++++++++++++ 4 files changed, 80 insertions(+) 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 205fe225ba0..faad7665039 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 940a9b79639..8bed27da94c 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 c3aba1491b1..cc1d16b0bf0 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-unified-sdk-jni/src/wallet_manager.rs b/packages/rs-unified-sdk-jni/src/wallet_manager.rs index 78c25061fa6..298d5845b83 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, From c1d5d157e37a082ff06d5237e1ac5f1bcdb71d9c Mon Sep 17 00:00:00 2001 From: bfoss765 <38437574+bfoss765@users.noreply.github.com> Date: Tue, 4 Aug 2026 23:04:11 -0400 Subject: [PATCH 3/5] fix(platform-wallet): drain the lossless persistence channel (mpsc) so the watermark can't freeze MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root-cause follow-up to the batching + sync_fault commits on this branch. Batching raised the burst threshold but a single `broadcast::Lagged` still froze a wallet's durable sync watermark permanently (dashpay/platform#4069). The producer (dashpay/rust-dashcore#924) now offers a dedicated, unbounded `mpsc` persistence channel alongside its lossy broadcast. This switches the consumer onto it: - core_bridge.rs: `spawn_/run_wallet_event_adapter` take `mpsc::UnboundedReceiver` instead of `broadcast::Receiver`. The batched `try_recv` fold is kept verbatim; the `Lagged`/`missed`/global `fault_all` path is removed because an unbounded channel can never lag. `AdapterFaultState` keeps only the per-wallet store-rejection freeze as a fail-closed backstop (never fires in a healthy run). - manager/mod.rs: take the receiver via `take_persistence_receiver()` instead of `subscribe_events()`. Unlike a broadcast receiver, the mpsc buffers events emitted before the task's first poll, so there is no subscribe-before-publish race. - Diagnostics via the `log` facade (android_logger forwards `log` to logcat; `tracing` may not — see rs-unified-sdk-jni JNI_OnLoad): one `log::info!("wallet-event batch: folded=.. wallets=.. synced_height_persisted=.. faulted=.. missed=0")` per drain, and a one-shot `log::error!("SYNC WATERMARK FROZEN ...")` if the per-wallet freeze ever latches — so the next tester logcat is unambiguous about whether the watermark is advancing. - Cargo.toml: re-pin key-wallet-manager (and the sibling rust-dashcore crates, kept consistent to avoid a duplicate-crate type mismatch) to the fork rev carrying #924. #4069-safety: the channel is lossless and in-order, so every row event reaches the persister before the `SyncHeightAdvanced` watermark that implies it — the durable watermark can never outrun its rows. The freeze guard stays as a backstop but should now never fire. Tests: broadcast-driven adapter tests ported to the mpsc; the `Lagged` test is replaced by `lossless_burst_never_freezes_and_watermark_reaches_tip` (a 3000-event burst — 3× the old ring — advances the watermark to the tip with no freeze). `cargo test -p platform-wallet` (531) and `-p platform-wallet-ffi` (224) green. Stacked on the batching + sync_fault commits (dashpay/platform#4289). Requires dashpay/rust-dashcore#924 (producer) to land. Co-Authored-By: Claude Opus 4.8 --- Cargo.lock | 24 +- Cargo.toml | 16 +- .../src/changeset/core_bridge.rs | 522 +++++++++--------- .../rs-platform-wallet/src/manager/mod.rs | 21 +- 4 files changed, 286 insertions(+), 297 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index fcbbb115465..579d20a91bc 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 86e5432b7ef..4918ad3ba72 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/rs-platform-wallet/src/changeset/core_bridge.rs b/packages/rs-platform-wallet/src/changeset/core_bridge.rs index 40c36fd9054..258ab51aeda 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,8 +45,8 @@ 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, TryRecvError}; +use tokio::sync::mpsc; +use tokio::sync::mpsc::error::TryRecvError; use tokio::sync::RwLock; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; @@ -58,19 +66,15 @@ use crate::wallet::platform_wallet::PlatformWalletInfo; /// 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. The upstream producer publishes -/// fire-and-forget onto a bounded ring (`DEFAULT_WALLET_EVENT_CAPACITY`, -/// 1000), so a consumer that slow overflows it, `recv()` returns -/// `Lagged`, and the durable-watermark guard freezes the wallet's -/// `synced_height` for the rest of the process lifetime. In the field that -/// presented as a sync that climbs toward completion and then "rolls back" -/// to near the install position on every relaunch: the watermark froze -/// after the first batch, so every later restart resumed from that same -/// frozen height. +/// 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 ring into one changeset +/// 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 rate is bounded by the ring rather than by the persister. This is +/// 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 + @@ -84,22 +88,18 @@ 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, } @@ -107,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 @@ -116,22 +116,14 @@ 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); - } } -/// 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 @@ -145,7 +137,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<()> @@ -165,76 +157,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). -/// -/// # Subscribe-before-publish +/// (drive a real `mpsc::UnboundedSender`, inject a probe persister). /// -/// 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. +/// # Lossless persistence channel (dashpay/platform#4069) /// -/// # Durable-watermark guard (dashpay/platform#4069) +/// 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. /// -/// 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. +/// 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. /// -/// 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. +/// # Durable-watermark guard (fail-closed backstop) /// -/// 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 @@ -242,42 +213,44 @@ 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 { - // Block for the first event of a batch. Everything already sitting - // in the ring 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`]). + // 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 missed: u64 = 0; let mut closed = false; - - match first { - 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 core = build_core_changeset(&wallet_manager, &event).await; - batch.entry(wallet_id).or_default().merge(core); - } - Err(RecvError::Lagged(n)) => missed += n, - Err(RecvError::Closed) if cancel.is_cancelled() => break, - Err(RecvError::Closed) => { - tracing::error!("WalletEvent broadcast closed unexpectedly"); - break; - } + { + 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 ring is empty. + // 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() { @@ -287,62 +260,61 @@ async fn run_wallet_event_adapter

( batch.entry(wallet_id).or_default().merge(core); folded += 1; } - Err(TryRecvError::Lagged(n)) => { - missed += n; - folded += 1; - } Err(TryRecvError::Empty) => break, - Err(TryRecvError::Closed) => { + Err(TryRecvError::Disconnected) => { closed = true; break; } } } - if missed > 0 { - // The 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). - // - // Faulting before storing this batch also covers the events - // folded in *ahead* of the lag: stripping their `synced_height` - // can only hold the watermark lower, never advance it past - // uncommitted rows, so the conservative direction is the safe - // one. - fault.fault_all(&sync_fault); - tracing::error!( - missed, - "wallet-event adapter lagged on broadcast channel; {missed} persistence events dropped — freezing every wallet's sync watermark so the next scan re-persists them (dashpay/platform#4069)" - ); - } - + // 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 below). + let wallets_in_batch = batch.len(); + let mut synced_height_persisted: Option = None; + let mut faulted_in_batch = 0usize; for (wallet_id, mut core) in batch { - // 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. Applied after the - // fold so a `synced_height` that arrived via merge is stripped - // too. - freeze_synced_height_if_faulted(&mut core, fault.is_faulted(&wallet_id)); + // 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. Applied after the fold so a + // `synced_height` that arrived via merge is stripped too. + let is_faulted = fault.is_faulted(&wallet_id); + if is_faulted { + faulted_in_batch += 1; + } + freeze_synced_height_if_faulted(&mut core, is_faulted); 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. + // BlockProcessed, a watermark-only batch stripped by the fault + // guard above, etc. — nothing to persist. Skip the round-trip. continue; } + if let Some(h) = core.synced_height { + synced_height_persisted = Some(synced_height_persisted.map_or(h, |cur| cur.max(h))); + } 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. + // 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); + faulted_in_batch += 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, @@ -351,9 +323,22 @@ async fn run_wallet_event_adapter

( } } + // One structured line per drain via the `log` facade so a tester + // logcat is unambiguous about whether the watermark is advancing. + // `missed` is always 0 now (the lossless channel cannot drop); it is + // kept in the line so a nonzero value would immediately stand out as a + // regression. + log::info!( + "wallet-event batch: folded={} wallets={} synced_height_persisted={:?} faulted={} missed=0", + folded, + wallets_in_batch, + synced_height_persisted, + faulted_in_batch, + ); + if closed { if !cancel.is_cancelled() { - tracing::error!("WalletEvent broadcast closed unexpectedly"); + tracing::error!("WalletEvent persistence channel closed unexpectedly"); } break; } @@ -363,15 +348,14 @@ async fn run_wallet_event_adapter

( /// 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; @@ -1050,10 +1034,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); @@ -1069,25 +1054,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; @@ -1099,7 +1074,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; @@ -1197,16 +1171,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(); } @@ -1222,35 +1199,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 @@ -1258,7 +1234,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 @@ -1309,7 +1285,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 @@ -1346,14 +1322,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(); @@ -1399,7 +1377,7 @@ mod tests { 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) = broadcast::channel::(64); + let (tx, rx) = unbounded_channel::(); for h in 1..=5u32 { tx.send(sync_height_event(wallet_id, h)).unwrap(); } @@ -1449,7 +1427,7 @@ mod tests { async fn batch_folds_per_wallet_not_across_wallets() { let wallet_a = [1u8; 32]; let wallet_b = [2u8; 32]; - let (tx, rx) = broadcast::channel::(64); + 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(); @@ -1486,29 +1464,24 @@ mod tests { ); } - /// (g) SAFETY INVARIANT under folding: once the fault latch is set, a - /// batch that merges a record-bearing event together with a watermark - /// event still persists the records but must NOT carry the merged - /// `synced_height`. - /// - /// This is the property that makes batching safe. The freeze is - /// applied after the fold, so a `synced_height` that entered the - /// changeset 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). + /// (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 merged_watermark_is_still_stripped_after_a_fault() { + async fn watermark_is_still_stripped_after_a_fault() { let wallet_id = [9u8; 32]; - // Capacity 2 with 4 sends → the first recv() reports Lagged(2), - // which latches the global fault before anything is stored. - let (tx, rx) = broadcast::channel::(2); - for h in 1..=4u32 { - tx.send(sync_height_event(wallet_id, h)).unwrap(); - } + 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( @@ -1519,28 +1492,32 @@ mod tests { cancel.clone(), )); - // Wait until the lag has been observed and latched, so the events - // below are guaranteed to be evaluated under the fault. + // 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 that will fold - // into ONE changeset for this wallet. + // 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 observed = obs_rx + let post = obs_rx .recv() .await - .expect("the record-bearing half of the batch must still persist"); - assert_eq!(observed.wallet_id, wallet_id); + .expect("the record-bearing event must still persist while faulted"); + assert_eq!(post.wallet_id, wallet_id); assert_eq!( - observed.synced_height, None, - "a merged watermark must still be stripped while the wallet is faulted" + post.synced_height, None, + "no store may carry a synced_height once the wallet is faulted" ); assert_eq!( - observed.last_processed_height, + post.last_processed_height, Some(60), "record-bearing fields must survive the freeze" ); @@ -1548,5 +1525,14 @@ mod tests { 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" + ); + } } } diff --git a/packages/rs-platform-wallet/src/manager/mod.rs b/packages/rs-platform-wallet/src/manager/mod.rs index ade7ac6e0a3..249766d3772 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()); From 3c57cb1a706133bd6d672f664117b1e102522a5e Mon Sep 17 00:00:00 2001 From: bfoss765 <38437574+bfoss765@users.noreply.github.com> Date: Wed, 5 Aug 2026 07:03:16 -0400 Subject: [PATCH 4/5] fix(platform-wallet): declare log outside the reverted txMetadata hunk so the merge keeps it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The log dependency was first added by the encrypted-txMetadata change (#4277), then reverted on v4.2-dev (#4279). This branch carries the log line only passively (unchanged from the merge-base), so GitHub's 3-way PR merge applies the base-side deletion and the merged Cargo.toml loses the declaration — while the log:: breadcrumb calls this branch adds in changeset/core_bridge.rs remain, producing error[E0433]: unresolved crate log in the Kotlin SDK CI build. Relocate log = "0.4" out of the reverted Logging hunk into the untouched Security region so it is a branch-owned insertion that survives the merge. Co-Authored-By: Claude Opus 4.8 --- packages/rs-platform-wallet/Cargo.toml | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/packages/rs-platform-wallet/Cargo.toml b/packages/rs-platform-wallet/Cargo.toml index 6c7e86a5b8e..d1c6dda282b 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 From ce9cc1acc03085abad3fd4294917141efeb00791 Mon Sep 17 00:00:00 2001 From: bfoss765 <38437574+bfoss765@users.noreply.github.com> Date: Wed, 5 Aug 2026 13:44:02 -0400 Subject: [PATCH 5/5] fix(platform-wallet): report only store-accepted heights as persisted MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review finding on dashpay/platform#4290: "the batch diagnostic still counts a rejected watermark as persisted". The per-drain batch line folded `core.synced_height` into `synced_height_persisted` BEFORE calling `persister.store(...)`, so a rejected changeset was logged as `synced_height_persisted=Some(h)` in the very drain that faulted the wallet *because* height h's rows were not accepted. We read these lines off a mainnet tester's logcat to decide whether the durable watermark is advancing, so an internally contradictory trace points the diagnosis at the wrong subsystem. This is a reporting bug, not a cosmetic nit. Split the commit path out of `run_wallet_event_adapter` into `commit_batch`, which returns a `BatchDiagnostics` distinguishing the three fates a height can meet within one drain: - `synced_height_persisted` — `store()` returned Ok. The ONLY field that means the durable watermark advanced. - `synced_height_frozen` — the fail-closed guard stripped it before it ever reached the store. Previously this collapsed to `persisted=None`, which is indistinguishable from a drain that simply carried no watermark. - `synced_height_rejected` — offered to the store, which 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 rather than over-reporting one number. The fail-closed guard (dashpay/platform#4069) is deliberately untouched — this changes REPORTING only. `freeze_synced_height_if_faulted` still strips `synced_height` after the fold, the per-wallet fault scoping is unchanged, and the one-shot `SYNC WATERMARK FROZEN` `log::error!` plus the `sync_fault` latch behave exactly as before (both asserted in the new tests). Also drops the hardcoded `missed=0` field: it reported a number the code never measured (the lossless mpsc has no drop counter), which is the same defect class as the finding above. Nothing in the repo parses this line, and the `wallet-event batch:` prefix testers grep for is unchanged. Tests: 7 new cases driving the real `commit_batch` (production commit path, guard included), covering accepted / rejected / guard-stripped / watermark-only-stripped / mixed-batch / monotonic-max / exact line format. `rejected_store_is_not_reported_as_persisted` was mutation-verified: with the pre-fix ordering reintroduced it fails with `left: Some(500), right: None`. `cargo test -p platform-wallet` green (538 + 9); rustfmt clean; clippy introduces no new findings in the touched file (the 3 pre-existing `-D warnings` errors in asset_lock/sync/recovery.rs and identity/network/withdrawal.rs are unchanged from this branch's head). Co-Authored-By: Claude Opus 4.8 --- .../src/changeset/core_bridge.rs | 517 ++++++++++++++++-- 1 file changed, 467 insertions(+), 50 deletions(-) diff --git a/packages/rs-platform-wallet/src/changeset/core_bridge.rs b/packages/rs-platform-wallet/src/changeset/core_bridge.rs index 258ab51aeda..df61687c74e 100644 --- a/packages/rs-platform-wallet/src/changeset/core_bridge.rs +++ b/packages/rs-platform-wallet/src/changeset/core_bridge.rs @@ -118,6 +118,84 @@ impl AdapterFaultState { } } +/// 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 persistence task. /// /// The `receiver` is the manager's lossless persistence receiver, taken once @@ -270,43 +348,113 @@ async fn run_wallet_event_adapter

( // 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 below). - let wallets_in_batch = batch.len(); - let mut synced_height_persisted: Option = None; - let mut faulted_in_batch = 0usize; - 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. Applied after the fold so a - // `synced_height` that arrived via merge is stripped too. - let is_faulted = fault.is_faulted(&wallet_id); - if is_faulted { - faulted_in_batch += 1; + // 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"); } - freeze_synced_height_if_faulted(&mut core, is_faulted); - 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; + 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 let Some(h) = core.synced_height { - synced_height_persisted = Some(synced_height_persisted.map_or(h, |cur| cur.max(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); + } } - let cs = PlatformWalletChangeSet { - core: Some(core), - ..PlatformWalletChangeSet::default() - }; - if let Err(e) = persister.store(wallet_id, cs) { + 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. - fault.fault_wallet(wallet_id, &sync_fault); - faulted_in_batch += 1; + 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; + 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 \ @@ -322,28 +470,8 @@ async fn run_wallet_event_adapter

( ); } } - - // One structured line per drain via the `log` facade so a tester - // logcat is unambiguous about whether the watermark is advancing. - // `missed` is always 0 now (the lossless channel cannot drop); it is - // kept in the line so a nonzero value would immediately stand out as a - // regression. - log::info!( - "wallet-event batch: folded={} wallets={} synced_height_persisted={:?} faulted={} missed=0", - folded, - wallets_in_batch, - synced_height_persisted, - faulted_in_batch, - ); - - if closed { - if !cancel.is_cancelled() { - tracing::error!("WalletEvent persistence channel closed unexpectedly"); - } - break; - } } - tracing::debug!("wallet-event adapter task exiting"); + diag } /// Durable-watermark guard for dashpay/platform#4069. @@ -1535,4 +1663,293 @@ mod tests { ); } } + + // ── 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" + ); + } }