From e2b806b8a35f57f5fe887a7aadd82232e0128df4 Mon Sep 17 00:00:00 2001 From: Roman <51091564+jeanpierreroma@users.noreply.github.com> Date: Tue, 11 Aug 2026 16:55:28 +0300 Subject: [PATCH 1/7] fix(platform-wallet): commit wallet events off the async runtime MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `run_wallet_event_adapter` called `commit_batch` — and through it `persister.store()` — inline on the tokio worker driving it. `store()` is synchronous, and for the SQLite backend commits a real transaction per call; its own trait docs say so, and warn that a slow write blocks every other wallet accessor for its duration. What they do not say, because until now it was not true, is that it also blocks the runtime those accessors run on. Field evidence from a testnet restore of a 6663-transaction wallet: - drains coalesced into ever larger, ever rarer batches — folded 1 → 47 → 164 → 512, with gaps of 42s, 144s and finally 1109s between them; - the metrics tick covering the 512-event drain reported `busy_ratio=1106 mean_poll_us=1397886` — a 1.4s mean poll on a runtime that read 24µs one second later; - `Blocks: last_activity: 549s` at the same moment, so the SPV managers sharing that runtime were starved, not idle; - the durable watermark topped out at height 2179999 against a chain tip of 2520064 and never caught up, so the home timeline — which only advances when a batch lands — showed roughly a third of the history ten minutes after core sync reported 100%. The commit now runs on `spawn_blocking`. The handle is awaited rather than raced against `cancel`: a store that has started must finish, and dropping the handle would not stop the thread in any case — shutdown is observed at the next `recv`. `AdapterFaultState` and the freeze latch move behind an `Arc>` and an `Arc` rather than being moved into the closure by value. That is deliberate: if the commit thread ever panicked, moving them would lose a wallet's frozen watermark, which would un-freeze a wallet whose verification had failed — the one outcome the fail-closed guard exists to prevent. The lock is uncontended by construction (one drain commits at a time, and this task is the only writer). A panicking commit thread is now reported and the drain skipped, rather than taking the adapter down with it. cargo test -p platform-wallet --lib # 662 passed cargo clippy --all-targets + fmt # clean --- .../src/changeset/core_bridge.rs | 74 ++++++++++++++++--- 1 file changed, 63 insertions(+), 11 deletions(-) diff --git a/packages/rs-platform-wallet/src/changeset/core_bridge.rs b/packages/rs-platform-wallet/src/changeset/core_bridge.rs index df1b4701cf9..6bae9a46383 100644 --- a/packages/rs-platform-wallet/src/changeset/core_bridge.rs +++ b/packages/rs-platform-wallet/src/changeset/core_bridge.rs @@ -34,7 +34,7 @@ use std::collections::{BTreeMap, HashMap, HashSet}; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use dashcore::blockdata::transaction::{txout::TxOut, OutPoint}; use dashcore::ScriptBuf; @@ -294,10 +294,16 @@ async fn run_wallet_event_adapter

( P: PlatformWalletPersistence + 'static, { tracing::debug!("wallet-event adapter task started"); - let mut fault = AdapterFaultState::default(); + // Both live behind handles rather than as locals because the commit runs + // on a blocking thread (see the `spawn_blocking` below) and has to be able + // to carry its state across drains. Moving them into the closure by value + // would lose a wallet's frozen watermark if that thread ever panicked — + // un-freezing a wallet that failed verification is the one outcome the + // fail-closed guard exists to prevent. + let fault = Arc::new(Mutex::new(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; + let freeze_logged = Arc::new(AtomicBool::new(false)); loop { // Block for the first event of a batch. Everything already sitting in @@ -360,14 +366,60 @@ 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 inside `commit_batch`). - let diag = commit_batch( - &*persister, - batch, - folded, - &mut fault, - &sync_fault, - &mut freeze_logged, - ); + // Commit on a blocking thread, never on the async worker. + // + // `store()` is synchronous and, for the SQLite backend, commits a real + // transaction per call — its own docs warn that a slow write blocks + // every other wallet accessor for its duration. Called inline here it + // blocked a tokio worker instead: a field restore showed one drain of + // 512 folded events park the runtime long enough for the metrics tick + // covering it to report a 1.4s mean poll, with the whole sync stalled + // for minutes at a time and the durable watermark left hundreds of + // thousands of blocks behind the chain tip. + // + // The handle is awaited rather than raced against `cancel`: a store + // that has started must be allowed to finish, and dropping the handle + // would not stop the thread anyway. Shutdown is observed at the next + // `recv` instead. + let persister_for_commit = Arc::clone(&persister); + let sync_fault_for_commit = Arc::clone(&sync_fault); + let fault_for_commit = Arc::clone(&fault); + let freeze_for_commit = Arc::clone(&freeze_logged); + let committed = tokio::task::spawn_blocking(move || { + // The lock is uncontended by construction — this task is the only + // writer, and one drain commits at a time — so it never blocks; + // it exists to carry the state, not to arbitrate. + let mut fault = fault_for_commit + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let mut freeze_logged = freeze_for_commit.load(Ordering::Relaxed); + let diag = commit_batch( + &*persister_for_commit, + batch, + folded, + &mut fault, + &sync_fault_for_commit, + &mut freeze_logged, + ); + freeze_for_commit.store(freeze_logged, Ordering::Relaxed); + diag + }) + .await; + + let diag = match committed { + Ok(diag) => diag, + // The commit thread panicked. The fault state survives (it lives + // behind the handle above), but this batch's outcome is unknown, + // so it is reported rather than silently folded into the next one. + Err(join_error) => { + tracing::error!( + error = %join_error, + folded, + "wallet-event commit thread failed; batch outcome unknown" + ); + continue; + } + }; // One structured line per drain via the `log` facade so a tester // logcat is unambiguous about whether the watermark is advancing. From 0499b9c99fdd40f5ac908b042b61f94ae33ae51b Mon Sep 17 00:00:00 2001 From: Roman <51091564+jeanpierreroma@users.noreply.github.com> Date: Thu, 13 Aug 2026 13:33:25 +0300 Subject: [PATCH 2/7] fix(platform-wallet): freeze a batch's wallets when its commit thread panics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Moving the commit to `spawn_blocking` quietly weakened the fail-closed rule, and both reviewers caught it. Before the move, a panic inside `store()` unwound the adapter task itself. That was violent, but it was safe in one specific way: the writer was gone, so no later batch could persist a higher `synced_height` for a wallet whose rows had just been lost. `spawn_blocking` turns the same panic into a recoverable `JoinError`, and the branch I wrote logged it and carried on — leaving the fault state untouched, so the very next batch could advance the watermark past rows of unknown fate. That is exactly the hole #4069 closed. The wallet ids are now captured before the batch moves into the closure, and a `JoinError` faults every one of them. Per-wallet rather than stopping the adapter, matching what a rejected `store()` already does: a wallet whose commit is in doubt freezes, its siblings keep syncing, and the process stays alive — which is the point of moving the commit off the runtime in the first place. `ProbePersister` gained a `panic_next` mode, and the new test asserts all three halves of the contract: the hard-fault signal is raised, the adapter survives, and no later store for that wallet carries a `synced_height`. The wait for the signal is bounded. An unbounded spin would have wedged CI with no diagnosis on a regression rather than failing it — verified by removing the fault path, where the test now fails in 5s with "a panicked commit must raise the hard-fault signal" instead of hanging. cargo test -p platform-wallet --lib # 663 passed cargo clippy --all-targets + cargo fmt --check # clean --- .../src/changeset/core_bridge.rs | 112 +++++++++++++++++- 1 file changed, 108 insertions(+), 4 deletions(-) diff --git a/packages/rs-platform-wallet/src/changeset/core_bridge.rs b/packages/rs-platform-wallet/src/changeset/core_bridge.rs index 6bae9a46383..b21f49c5a5e 100644 --- a/packages/rs-platform-wallet/src/changeset/core_bridge.rs +++ b/packages/rs-platform-wallet/src/changeset/core_bridge.rs @@ -381,6 +381,10 @@ async fn run_wallet_event_adapter

( // that has started must be allowed to finish, and dropping the handle // would not stop the thread anyway. Shutdown is observed at the next // `recv` instead. + // Captured before the batch moves into the closure: if the commit + // thread panics, these are the wallets whose rows have an unknown fate + // and whose watermark must therefore be frozen. + let batch_wallet_ids: Vec = batch.keys().copied().collect(); let persister_for_commit = Arc::clone(&persister); let sync_fault_for_commit = Arc::clone(&sync_fault); let fault_for_commit = Arc::clone(&fault); @@ -408,14 +412,34 @@ async fn run_wallet_event_adapter

( let diag = match committed { Ok(diag) => diag, - // The commit thread panicked. The fault state survives (it lives - // behind the handle above), but this batch's outcome is unknown, - // so it is reported rather than silently folded into the next one. + // The commit thread panicked, so `commit_batch` never reached the + // `store()` rejection arm that would have frozen the affected + // wallets. Freeze them here instead. + // + // Before this call moved off the runtime a panic unwound the whole + // adapter task, which stopped every later watermark advance by + // killing the writer. `spawn_blocking` turns that into a recoverable + // `JoinError`, and simply continuing would let the NEXT batch + // persist a higher `synced_height` for a wallet whose rows from this + // batch may never have landed — the exact hole the fail-closed rule + // exists to prevent. Faulting per wallet rather than stopping the + // adapter keeps the existing design: a wallet whose commit is in + // doubt freezes, its siblings keep syncing. Err(join_error) => { + { + let mut fault = fault + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + for wallet_id in &batch_wallet_ids { + fault.fault_wallet(*wallet_id, &sync_fault); + } + } tracing::error!( error = %join_error, folded, - "wallet-event commit thread failed; batch outcome unknown" + wallets = batch_wallet_ids.len(), + "wallet-event commit thread failed; freezing the batch's \ + wallets because their rows have an unknown outcome" ); continue; } @@ -1932,6 +1956,10 @@ mod tests { struct ProbePersister { obs: UnboundedSender, fail_once: Mutex>, + /// Wallets whose NEXT `store()` panics instead of returning. Models a + /// backend that dies mid-write — the case that used to unwind the whole + /// adapter task and now surfaces as a `JoinError`. + panic_once: Mutex>, } impl ProbePersister { @@ -1939,11 +1967,15 @@ mod tests { Self { obs, fail_once: Mutex::new(HashSet::new()), + panic_once: Mutex::new(HashSet::new()), } } fn fail_next(&self, wallet_id: WalletId) { self.fail_once.lock().unwrap().insert(wallet_id); } + fn panic_next(&self, wallet_id: WalletId) { + self.panic_once.lock().unwrap().insert(wallet_id); + } } impl PlatformWalletPersistence for ProbePersister { @@ -1953,6 +1985,9 @@ mod tests { changeset: PlatformWalletChangeSet, ) -> Result<(), PersistenceError> { let core = changeset.core.as_ref(); + if self.panic_once.lock().unwrap().remove(&wallet_id) { + panic!("probe persister: store panicked for {wallet_id:?}"); + } let rejected = self.fail_once.lock().unwrap().remove(&wallet_id); let _ = self.obs.send(StoreObserved { wallet_id, @@ -2305,6 +2340,75 @@ mod tests { ); } + /// (h) SAFETY INVARIANT under a commit-thread PANIC: a wallet whose + /// `store()` panicked must be frozen just as if the store had been + /// rejected, because its rows have an unknown fate. + /// + /// This is a regression guard on the move to `spawn_blocking`. Before it, + /// a panic unwound the adapter task itself, which stopped every later + /// watermark advance by killing the writer outright. `spawn_blocking` + /// turns that into a recoverable `JoinError` — and merely logging it would + /// let the NEXT batch persist a higher `synced_height` for a wallet whose + /// earlier rows may never have landed, which is exactly the hole + /// dashpay/platform#4069 closed. + #[tokio::test] + async fn a_panicking_commit_freezes_the_batch_wallets() { + let wallet_id = [0xEEu8; 32]; + let (tx, rx) = unbounded_channel::(); + + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = Arc::new(ProbePersister::new(obs_tx)); + persister.panic_next(wallet_id); + let sync_fault = Arc::new(AtomicBool::new(false)); + let cancel = CancellationToken::new(); + let handle = tokio::spawn(run_wallet_event_adapter( + test_manager(), + Arc::clone(&persister), + rx, + Arc::clone(&sync_fault), + cancel.clone(), + )); + + // The store for this batch panics: no observation is emitted, and the + // adapter must fault the wallet rather than carry on unaffected. + tx.send(block_processed_event(wallet_id, 10)).unwrap(); + // Bounded, so a regression fails the test instead of hanging it: with + // the fault-on-panic path removed, `sync_fault` is simply never raised + // and an unbounded spin would wedge CI with no diagnosis. + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while !sync_fault.load(Ordering::Relaxed) { + tokio::task::yield_now().await; + } + }) + .await + .expect("a panicked commit must raise the hard-fault signal"); + + // The adapter must still be alive — the point of moving the commit off + // the runtime is that one bad batch does not take the writer with it. + assert!( + !handle.is_finished(), + "a panicked commit must not kill the adapter" + ); + + // A later watermark for the same wallet must not reach the store. + tx.send(block_processed_event(wallet_id, 60)).unwrap(); + tx.send(sync_height_event(wallet_id, 900)).unwrap(); + + let post = obs_rx + .recv() + .await + .expect("the record-bearing event must still persist while faulted"); + assert_eq!(post.wallet_id, wallet_id); + assert_eq!( + post.synced_height, None, + "a wallet whose commit panicked must not advance its durable watermark" + ); + + cancel.cancel(); + drop(tx); + handle.await.unwrap(); + } + /// (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 From 4e5a1939ba27ccd14cf22258630cdc4a911d06c2 Mon Sep 17 00:00:00 2001 From: Roman <51091564+jeanpierreroma@users.noreply.github.com> Date: Thu, 13 Aug 2026 22:39:20 +0300 Subject: [PATCH 3/7] fix(platform-wallet): fault only the wallets whose commit outcome is unknown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review findings, and the first is a correction to my own fix. **The panic handler over-corrected.** `commit_batch` walks the batch serially, so a panic partitions it: wallets whose `store()` already returned are settled — accepted or rejected, and a rejection faulted them from inside. Faulting those too stripped a healthy wallet's watermark for the rest of the session over a sibling's bad batch. `commit_batch` now records each wallet as its `store()` RETURNS — after the call, so a panicking wallet never reaches the line, and a wallet the loop never got to is never recorded. What is missing from that list is exactly the set whose outcome nobody can reason about, and only those freeze. The list lives behind an `Arc>` outside the closure so it survives the panic that makes it interesting. **The off-runtime boundary now has a test that actually guards it.** Every other test here passes with `commit_batch` moved back inline, because they only assert persistence outcomes. `a_blocked_store_does_not_ park_the_runtime` runs the adapter on a single worker, parks a `store()` with a controllable gate, and requires a spawned task to still be scheduled. It is built out of `std::mpsc::recv_timeout` and `std::thread::sleep` rather than `tokio::time`, and that is not stylistic: the regression parks the runtime's only worker, and a tokio timer needs that runtime to fire. My first two attempts used async timeouts and HUNG on the regression instead of failing it, which is worse than the bug — CI burns the wall clock and reports nothing. Verified by reverting to an inline commit: the test now fails in 5s with "a blocked store must not hold the runtime's only worker: Timeout". The cross-wallet test is likewise bounded, for the same reason: a frozen wallet's watermark-only changeset collapses to nothing, so an unbounded `recv` waits forever on a regression. cargo test -p platform-wallet --lib # 665 passed cargo clippy --all-targets + cargo fmt --check # clean --- .../src/changeset/core_bridge.rs | 256 +++++++++++++++++- 1 file changed, 249 insertions(+), 7 deletions(-) diff --git a/packages/rs-platform-wallet/src/changeset/core_bridge.rs b/packages/rs-platform-wallet/src/changeset/core_bridge.rs index b21f49c5a5e..a3269e2319d 100644 --- a/packages/rs-platform-wallet/src/changeset/core_bridge.rs +++ b/packages/rs-platform-wallet/src/changeset/core_bridge.rs @@ -385,6 +385,12 @@ async fn run_wallet_event_adapter

( // thread panics, these are the wallets whose rows have an unknown fate // and whose watermark must therefore be frozen. let batch_wallet_ids: Vec = batch.keys().copied().collect(); + // Filled by `commit_batch` as each wallet's `store()` returns. Lives + // out here so a panicking commit thread cannot take it down with it: + // what it holds is the difference between "this wallet's rows are + // accounted for" and "nobody knows". + let settled: Arc>> = Arc::new(Mutex::new(Vec::new())); + let settled_for_commit = Arc::clone(&settled); let persister_for_commit = Arc::clone(&persister); let sync_fault_for_commit = Arc::clone(&sync_fault); let fault_for_commit = Arc::clone(&fault); @@ -397,6 +403,9 @@ async fn run_wallet_event_adapter

( .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); let mut freeze_logged = freeze_for_commit.load(Ordering::Relaxed); + let mut settled = settled_for_commit + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); let diag = commit_batch( &*persister_for_commit, batch, @@ -404,6 +413,7 @@ async fn run_wallet_event_adapter

( &mut fault, &sync_fault_for_commit, &mut freeze_logged, + &mut settled, ); freeze_for_commit.store(freeze_logged, Ordering::Relaxed); diag @@ -426,20 +436,44 @@ async fn run_wallet_event_adapter

( // adapter keeps the existing design: a wallet whose commit is in // doubt freezes, its siblings keep syncing. Err(join_error) => { + // `commit_batch` walks the batch serially, so a panic partitions + // it: wallets whose `store()` already returned are settled — their + // rows were accepted or rejected, and a rejection already faulted + // them from inside. Freezing those too would strip a healthy + // wallet's watermark for the rest of the session over a sibling's + // bad batch. + // + // What is left — the wallet that panicked, plus every wallet the + // loop never reached — has no known outcome, and its events are + // gone from the lossless channel. Those must freeze, or a later + // batch advances their watermark past rows that may never have + // landed. + let settled_ids = settled + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .iter() + .copied() + .collect::>(); + let unsettled: Vec = batch_wallet_ids + .iter() + .copied() + .filter(|id| !settled_ids.contains(id)) + .collect(); { let mut fault = fault .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); - for wallet_id in &batch_wallet_ids { + for wallet_id in &unsettled { fault.fault_wallet(*wallet_id, &sync_fault); } } tracing::error!( error = %join_error, folded, - wallets = batch_wallet_ids.len(), - "wallet-event commit thread failed; freezing the batch's \ - wallets because their rows have an unknown outcome" + settled = settled_ids.len(), + frozen = unsettled.len(), + "wallet-event commit thread failed; freezing the wallets whose \ + rows have an unknown outcome" ); continue; } @@ -487,6 +521,7 @@ fn commit_batch

( fault: &mut AdapterFaultState, sync_fault: &AtomicBool, freeze_logged: &mut bool, + settled: &mut Vec, ) -> BatchDiagnostics where P: PlatformWalletPersistence + ?Sized, @@ -535,7 +570,15 @@ where asset_locks: (!Merge::is_empty(&asset_locks)).then_some(asset_locks), ..PlatformWalletChangeSet::default() }; - match persister.store(wallet_id, cs) { + let store_result = persister.store(wallet_id, cs); + // Recorded only once the store has RETURNED, and whether it accepted + // or rejected — both are answers. A wallet whose store panicked never + // reaches this line, and one the loop never got to is never pushed at + // all, so what is missing from `settled` is exactly the set whose + // outcome the caller cannot reason about. See the `JoinError` arm in + // `run_wallet_event_adapter`. + settled.push(wallet_id); + match store_result { Ok(()) => { if let Some(h) = offered_height { diag.record_persisted(h); @@ -1960,6 +2003,13 @@ mod tests { /// backend that dies mid-write — the case that used to unwind the whole /// adapter task and now surfaces as a `JoinError`. panic_once: Mutex>, + /// Held closed to keep a `store()` call parked. The SQLite backend + /// commits a real transaction per call, so a slow disk parks the caller + /// for real; this makes that duration controllable. + block_until: Mutex>>, + /// Raised as soon as a blocked `store()` is entered, so a test can wait + /// for the block to be in effect rather than sleeping and hoping. + blocked: Arc, } impl ProbePersister { @@ -1968,8 +2018,17 @@ mod tests { obs, fail_once: Mutex::new(HashSet::new()), panic_once: Mutex::new(HashSet::new()), + block_until: Mutex::new(None), + blocked: Arc::new(AtomicBool::new(false)), } } + /// Park the next `store()` until the returned sender is dropped or + /// signalled. `blocked` reports when the park is actually in effect. + fn block_next(&self) -> (std::sync::mpsc::Sender<()>, Arc) { + let (tx, rx) = std::sync::mpsc::channel(); + *self.block_until.lock().unwrap() = Some(rx); + (tx, Arc::clone(&self.blocked)) + } fn fail_next(&self, wallet_id: WalletId) { self.fail_once.lock().unwrap().insert(wallet_id); } @@ -1985,6 +2044,14 @@ mod tests { changeset: PlatformWalletChangeSet, ) -> Result<(), PersistenceError> { let core = changeset.core.as_ref(); + if let Some(gate) = self.block_until.lock().unwrap().take() { + self.blocked.store(true, Ordering::Relaxed); + // Blocks the calling thread outright — the whole point is to + // model a synchronous backend, so an async wait would prove + // nothing. + let _ = gate.recv(); + self.blocked.store(false, Ordering::Relaxed); + } if self.panic_once.lock().unwrap().remove(&wallet_id) { panic!("probe persister: store panicked for {wallet_id:?}"); } @@ -2394,9 +2461,9 @@ mod tests { tx.send(block_processed_event(wallet_id, 60)).unwrap(); tx.send(sync_height_event(wallet_id, 900)).unwrap(); - let post = obs_rx - .recv() + let post = tokio::time::timeout(std::time::Duration::from_secs(5), obs_rx.recv()) .await + .expect("a faulted wallet must still persist its rows") .expect("the record-bearing event must still persist while faulted"); assert_eq!(post.wallet_id, wallet_id); assert_eq!( @@ -2409,6 +2476,174 @@ mod tests { handle.await.unwrap(); } + /// (j) THE PRIMARY BEHAVIOUR OF THIS PR: a blocked `store()` must not park + /// the async runtime. + /// + /// `store()` is synchronous and, for the SQLite backend, commits a real + /// transaction per call. Called inline on a tokio worker it held that + /// worker for the duration — a field restore showed one drain park the + /// runtime long enough for the metrics tick covering it to report a 1.4s + /// mean poll, with the durable watermark left hundreds of thousands of + /// blocks behind the chain tip. + /// + /// Every other test in this module would still pass with `commit_batch` + /// moved back inline, because they only check persistence outcomes. This + /// one runs the adapter on a SINGLE worker, parks a `store()`, and requires + /// a spawned task to still get scheduled. + /// + /// Deliberately built out of `std` primitives — a `std::mpsc` handoff and + /// `std::thread::sleep` on the test thread — rather than `tokio::time`. + /// The regression parks the runtime's only worker, and a tokio timer needs + /// that runtime to fire: an async timeout here hangs instead of failing, + /// which is worse than the bug it is meant to catch. + #[test] + fn a_blocked_store_does_not_park_the_runtime() { + use std::sync::mpsc as std_mpsc; + use std::time::{Duration, Instant}; + + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .build() + .unwrap(); + + let wallet_id = [0x33u8; 32]; + let (tx, rx) = unbounded_channel::(); + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = Arc::new(ProbePersister::new(obs_tx)); + let (release, blocked) = persister.block_next(); + let sync_fault = Arc::new(AtomicBool::new(false)); + let cancel = CancellationToken::new(); + + let handle = runtime.spawn(run_wallet_event_adapter( + test_manager(), + Arc::clone(&persister), + rx, + Arc::clone(&sync_fault), + cancel.clone(), + )); + + // Park the commit inside a synchronous `store()`. + tx.send(block_processed_event(wallet_id, 10)).unwrap(); + let deadline = Instant::now() + Duration::from_secs(5); + while !blocked.load(Ordering::Relaxed) { + assert!( + Instant::now() < deadline, + "the store must actually park before the assertion below means anything" + ); + std::thread::sleep(Duration::from_millis(10)); + } + + // The discriminator: a SPAWNED task has to be scheduled on the + // runtime's single worker. With the commit on `spawn_blocking` the + // worker is free and this arrives at once; with it inline the worker is + // sitting inside `store()` and this times out. + let (sentinel_tx, sentinel_rx) = std_mpsc::channel(); + runtime.spawn(async move { + let _ = sentinel_tx.send(()); + }); + sentinel_rx + .recv_timeout(Duration::from_secs(5)) + .expect("a blocked store must not hold the runtime's only worker"); + + // Release, then let the drain finish so the adapter shuts down cleanly. + drop(release); + runtime.block_on(async { + let observed = tokio::time::timeout(Duration::from_secs(5), obs_rx.recv()) + .await + .expect("the released store must complete") + .expect("store observed"); + assert_eq!(observed.wallet_id, wallet_id); + + cancel.cancel(); + drop(tx); + handle.await.unwrap(); + }); + } + + /// (i) A commit panic must not punish the wallets it did not reach. + /// + /// `commit_batch` walks the batch serially (a `BTreeMap`, so in wallet-id + /// order). If an earlier wallet's `store()` returned and a later one + /// panics, the earlier wallet's rows are on disk and its watermark is + /// safe — freezing it would strip its `synced_height` for the rest of the + /// session over a sibling's bad batch. + /// + /// Guards the fix for the first version of the panic handler, which + /// faulted every wallet in the drain. + #[tokio::test] + async fn a_panicking_commit_spares_the_wallets_it_already_stored() { + // `BTreeMap` order decides who is committed first, so the ids are + // chosen to put the healthy wallet ahead of the panicking one. + let healthy = [0x11u8; 32]; + let doomed = [0x22u8; 32]; + let (tx, rx) = unbounded_channel::(); + + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = Arc::new(ProbePersister::new(obs_tx)); + persister.panic_next(doomed); + 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(), + )); + + // Both wallets in one drain: `healthy` stores, then `doomed` panics. + // Sent before either is observed so they fold into a single batch. + tx.send(block_processed_event(healthy, 10)).unwrap(); + tx.send(block_processed_event(doomed, 10)).unwrap(); + + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while !sync_fault.load(Ordering::Relaxed) { + tokio::task::yield_now().await; + } + }) + .await + .expect("the panicked commit must raise the hard-fault signal"); + + // Drain what the panicking batch managed to observe. + while obs_rx.try_recv().is_ok() {} + + // The healthy wallet's watermark must still advance: its store + // returned, so its rows are accounted for. + tx.send(sync_height_event(healthy, 900)).unwrap(); + // Bounded: on a regression the healthy wallet is frozen, its + // watermark-only changeset collapses to nothing, and no store is + // observed at all — an unbounded `recv` would hang CI instead of + // reporting which invariant broke. + let after_healthy = tokio::time::timeout(std::time::Duration::from_secs(5), obs_rx.recv()) + .await + .expect("a wallet whose store completed must still be storable") + .expect("healthy wallet still stores"); + assert_eq!(after_healthy.wallet_id, healthy); + assert_eq!( + after_healthy.synced_height, + Some(900), + "a wallet whose store completed must not be frozen by a sibling's panic" + ); + + // The wallet whose store panicked must be frozen. + tx.send(block_processed_event(doomed, 60)).unwrap(); + tx.send(sync_height_event(doomed, 900)).unwrap(); + let after_doomed = tokio::time::timeout(std::time::Duration::from_secs(5), obs_rx.recv()) + .await + .expect("a faulted wallet must still persist its rows") + .expect("doomed wallet still persists rows"); + assert_eq!(after_doomed.wallet_id, doomed); + assert_eq!( + after_doomed.synced_height, None, + "the wallet whose commit panicked must not advance its watermark" + ); + + cancel.cancel(); + drop(tx); + handle.await.unwrap(); + } + /// (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 @@ -2827,6 +3062,7 @@ mod tests { &mut fault, &sync_fault, &mut freeze_logged, + &mut Vec::new(), ); let observed = obs_rx @@ -2899,6 +3135,7 @@ mod tests { &mut fault, &sync_fault, &mut freeze_logged, + &mut Vec::new(), ); assert_eq!(diag.persisted, Some(500)); @@ -2930,6 +3167,7 @@ mod tests { &mut fault, &sync_fault, &mut freeze_logged, + &mut Vec::new(), ); // The height was genuinely offered to the store... @@ -2999,6 +3237,7 @@ mod tests { &mut fault, &sync_fault, &mut freeze_logged, + &mut Vec::new(), ); assert_eq!(diag.persisted, None, "a frozen watermark is not persisted"); @@ -3046,6 +3285,7 @@ mod tests { &mut fault, &sync_fault, &mut freeze_logged, + &mut Vec::new(), ); assert_eq!(diag.frozen, Some(1234)); @@ -3082,6 +3322,7 @@ mod tests { &mut fault, &sync_fault, &mut freeze_logged, + &mut Vec::new(), ); assert_eq!( @@ -3125,6 +3366,7 @@ mod tests { &mut fault, &sync_fault, &mut freeze_logged, + &mut Vec::new(), ); assert_eq!(diag.wallets, 1); From 4354b16b5b20ba2eb9172e8d4ff6fd91f7cbe44c Mon Sep 17 00:00:00 2001 From: Roman <51091564+jeanpierreroma@users.noreply.github.com> Date: Tue, 18 Aug 2026 12:49:23 +0300 Subject: [PATCH 4/7] fix(platform-wallet): spare no-op wallets, and mark panic freezes like any other MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three review findings on the off-runtime commit move. **Empty batches were being frozen.** `batch_wallet_ids` took every wallet that contributed an event, but a wallet reaches the batch even when its events project to nothing — a `TransactionInstantLocked` ignored because the transaction is already chain-locked, a `SyncHeightAdvanced` for an unknown wallet. `commit_batch` skips those at the `is_empty_no_records()` gate and never calls `store()` for them, so a sibling's panic says nothing about their rows; freezing them stripped a healthy wallet's watermark for the rest of the session over someone else's bad batch. The candidate list is now built from batches that carry something to persist. The filter is deliberately conservative in one direction: `commit_batch` re-tests emptiness after `freeze_synced_height_if_faulted` has stripped a faulted wallet's watermark, so a batch that looks non-empty here can still be skipped there. That wallet is already faulted, so freezing it again costs nothing, while the opposite error — omitting a wallet whose store did run — is the hole this whole path exists to close. **A panic-induced freeze was invisible where it matters.** The branch raised the latch but reported only through `tracing`, while the documented contract is a one-shot `SYNC WATERMARK FROZEN` line through the `log` facade, because android_logger forwards `log` to logcat and `tracing` may not. It also left `freeze_logged` clear, so a later rejected store would have emitted that supposedly one-shot line as though it were the session's first fault. Both paths now go through the same marker, set with `swap` so a store panicking inside the same blocking task cannot produce it twice. **The boundary had no test.** Every existing assertion checks persistence outcomes, and all of them still pass with `commit_batch` called inline — they never touch the property this move is for. `a_blocking_store_does_not_ park_the_runtime` pins the runtime to one worker and asserts an unrelated task still runs while a store is blocked; inlined, it fails in 0.22s. Its release and verdict are driven from a plain `std::thread` on purpose: a regression parks the only worker, so neither the test body nor a `tokio::time::timeout` can run — an in-runtime deadline would hang CI rather than fail it, which is what the bounded waits elsewhere in this file exist to avoid. --- .../src/changeset/core_bridge.rs | 141 +++++++++++++++++- 1 file changed, 140 insertions(+), 1 deletion(-) diff --git a/packages/rs-platform-wallet/src/changeset/core_bridge.rs b/packages/rs-platform-wallet/src/changeset/core_bridge.rs index a3269e2319d..013209da1dc 100644 --- a/packages/rs-platform-wallet/src/changeset/core_bridge.rs +++ b/packages/rs-platform-wallet/src/changeset/core_bridge.rs @@ -384,7 +384,32 @@ async fn run_wallet_event_adapter

( // Captured before the batch moves into the closure: if the commit // thread panics, these are the wallets whose rows have an unknown fate // and whose watermark must therefore be frozen. - let batch_wallet_ids: Vec = batch.keys().copied().collect(); + // + // Only wallets `commit_batch` can actually call `store()` for. A wallet + // contributes to the batch whenever it produced an event, including + // events that project to nothing — a `TransactionInstantLocked` that is + // ignored because the transaction is already chain-locked, a + // `SyncHeightAdvanced` for an unknown wallet. Those hit the + // `is_empty_no_records()` skip and never reach a store, so a sibling's + // panic says nothing about them; freezing them would strip a healthy + // wallet's watermark for the rest of the session over someone else's + // bad batch. + // + // Deliberately conservative in one direction: `commit_batch` re-tests + // emptiness AFTER `freeze_synced_height_if_faulted` has stripped a + // faulted wallet's watermark, so a batch that looks non-empty here can + // still be skipped there. That wallet is already faulted, so freezing + // it again costs nothing — whereas the reverse error, omitting a wallet + // whose store did run, would leave a watermark free to advance past + // rows nobody can account for. + let batch_wallet_ids: Vec = batch + .iter() + .filter(|(_, wallet_batch)| { + !wallet_batch.core.is_empty_no_records() + || !Merge::is_empty(&wallet_batch.asset_locks) + }) + .map(|(wallet_id, _)| *wallet_id) + .collect(); // Filled by `commit_batch` as each wallet's `store()` returns. Lives // out here so a panicking commit thread cannot take it down with it: // what it holds is the difference between "this wallet's rows are @@ -467,6 +492,29 @@ async fn run_wallet_event_adapter

( fault.fault_wallet(*wallet_id, &sync_fault); } } + // Same one-shot marker the rejection arm emits, and via the same + // `log` facade, because this freeze is indistinguishable from + // that one as far as the host is concerned: the latch is up and + // a rescan is pending. Leaving it to `tracing` alone would hide + // a panic-induced freeze from logcat entirely, and — worse — + // leave `freeze_logged` clear, so a later rejected store would + // emit the supposedly one-shot line as though it were the first + // fault of the session. + // + // `swap` rather than load-then-store: a store panicking inside + // the same blocking task can race the flag the task wrote back + // on its way out, and emitting this line twice is a worse + // outcome than the branch reading its own write. + if !unsettled.is_empty() && !freeze_logged.swap(true, Ordering::Relaxed) { + log::error!( + "SYNC WATERMARK FROZEN: the wallet-event commit thread panicked ({}); \ + {} wallet(s) whose rows have an unknown outcome are now held so the \ + next scan re-persists them (dashpay/platform#4370). \ + syncFaultDetected() is latched.", + join_error, + unsettled.len() + ); + } tracing::error!( error = %join_error, folded, @@ -2418,6 +2466,97 @@ mod tests { /// let the NEXT batch persist a higher `synced_height` for a wallet whose /// earlier rows may never have landed, which is exactly the hole /// dashpay/platform#4069 closed. + /// The commit must not run on a Tokio worker. + /// + /// This is the behaviour the `spawn_blocking` move exists for, and no + /// other test covers it: the outcome-only assertions elsewhere all still + /// pass with `commit_batch` called inline, because a blocking store on a + /// multi-worker runtime merely steals one worker and the rest of the test + /// proceeds. Pinning the runtime to a single worker makes the difference + /// observable — inline, that one worker sits inside `store()` and nothing + /// else on the runtime can advance. + /// + /// The release is driven from a plain `std::thread`, and the verdict is + /// read there too, because a regression parks the only worker: the test + /// body cannot run, and neither can a `tokio::time::timeout` — the timer + /// needs the same worker to fire. An in-runtime deadline would therefore + /// hang CI instead of failing it, which is exactly what the bounded waits + /// elsewhere in this file exist to avoid. + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + async fn a_blocking_store_does_not_park_the_runtime() { + let wallet_id = [0xB1u8; 32]; + let (tx, rx) = unbounded_channel::(); + + let (obs_tx, mut obs_rx) = unbounded_channel(); + let persister = Arc::new(ProbePersister::new(obs_tx)); + let (unblock, blocked) = persister.block_next(); + 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(), + )); + + // Set before the store blocks, so it is waiting on the runtime rather + // than racing to be scheduled: whether it ran is then purely a question + // of the worker being free. + let progressed = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&progressed); + tokio::spawn(async move { + flag.store(true, Ordering::Relaxed); + }); + + tx.send(block_processed_event(wallet_id, 10)).unwrap(); + + // Outside the runtime entirely: waits for the store to park, records + // whether the unrelated task got to run while it was parked, then + // releases it so the test always terminates either way. + let observer_progressed = Arc::clone(&progressed); + let observer_blocked = Arc::clone(&blocked); + let watcher = std::thread::spawn(move || { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); + while !observer_blocked.load(Ordering::Relaxed) { + if std::time::Instant::now() > deadline { + return Err("store never entered its block"); + } + std::thread::sleep(std::time::Duration::from_millis(10)); + } + // Give the runtime a moment it does not need when it is healthy; + // parked, no amount of waiting would help. + std::thread::sleep(std::time::Duration::from_millis(200)); + let ran_while_blocked = observer_progressed.load(Ordering::Relaxed); + drop(unblock); + Ok(ran_while_blocked) + }); + + let observed = obs_rx + .recv() + .await + .expect("the released store should report"); + assert_eq!(observed.wallet_id, wallet_id); + + let ran_while_blocked = watcher + .join() + .expect("watcher thread should not panic") + .expect("store never entered its block"); + assert!( + ran_while_blocked, + "an unrelated task must make progress while the commit blocks — \ + the commit is running on a runtime worker instead of a blocking thread" + ); + assert!( + !sync_fault.load(Ordering::Relaxed), + "a slow store is not a fault" + ); + + cancel.cancel(); + drop(tx); + let _ = handle.await; + } + #[tokio::test] async fn a_panicking_commit_freezes_the_batch_wallets() { let wallet_id = [0xEEu8; 32]; From b8cb0231e1555cebe099224af17d7a591b16ef08 Mon Sep 17 00:00:00 2001 From: Roman <51091564+jeanpierreroma@users.noreply.github.com> Date: Tue, 18 Aug 2026 13:57:22 +0300 Subject: [PATCH 5/7] fix(platform-wallet): share the one-shot freeze flag, and settle the blocking test's race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to two review points. **The one-shot marker could still fire twice.** `commit_batch` took the flag as a `&mut bool` copied out of the atomic on the way in and written back on the way out. That write-back never happens when a store panics mid-batch, so a marker already emitted for an earlier rejection in the same batch was lost, and the panic branch then announced the same freeze a second time. Both paths now take the `AtomicBool` itself and claim it with `swap`, so the flag is updated at the moment it is spent rather than when the task unwinds. **The blocking test could pass on a regression.** It spawned its sentinel before the store was observed blocked, so the runtime was free to poll that task first, set the flag while nothing was blocking, and let the test pass with `commit_batch` called inline. The sentinel is now spawned from the watcher thread, strictly after the block is observed, through a `Handle` captured beforehand; whether it runs is then only a question of the worker being free. The fixed 200 ms observation window is gone too — both waits are deadline polls, so a loaded runner costs time rather than a false verdict. Checked by inlining `commit_batch` three times in a row: fails every run, in 0.03s, against 666 passing with the change in place. Note for the reviewer: the suggestion to delete this test as a duplicate of `a_blocked_store_does_not_park_the_runtime` at lines 2638-2701 could not be applied — there is no such test on this branch, before or after these commits. This is the only blocking test in the file. --- .../src/changeset/core_bridge.rs | 109 +++++++++--------- 1 file changed, 57 insertions(+), 52 deletions(-) diff --git a/packages/rs-platform-wallet/src/changeset/core_bridge.rs b/packages/rs-platform-wallet/src/changeset/core_bridge.rs index 013209da1dc..d2eb7c5ccbc 100644 --- a/packages/rs-platform-wallet/src/changeset/core_bridge.rs +++ b/packages/rs-platform-wallet/src/changeset/core_bridge.rs @@ -427,21 +427,22 @@ async fn run_wallet_event_adapter

( let mut fault = fault_for_commit .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); - let mut freeze_logged = freeze_for_commit.load(Ordering::Relaxed); let mut settled = settled_for_commit .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); - let diag = commit_batch( + // The flag itself, not a copy: a local `bool` written back after + // `commit_batch` returns is lost when a later store in the same + // batch panics, and the panic branch would then emit the one-shot + // marker a second time for a freeze already announced. + commit_batch( &*persister_for_commit, batch, folded, &mut fault, &sync_fault_for_commit, - &mut freeze_logged, + &freeze_for_commit, &mut settled, - ); - freeze_for_commit.store(freeze_logged, Ordering::Relaxed); - diag + ) }) .await; @@ -497,7 +498,7 @@ async fn run_wallet_event_adapter

( // that one as far as the host is concerned: the latch is up and // a rescan is pending. Leaving it to `tracing` alone would hide // a panic-induced freeze from logcat entirely, and — worse — - // leave `freeze_logged` clear, so a later rejected store would + // leave the flag clear, so a later rejected store would // emit the supposedly one-shot line as though it were the first // fault of the session. // @@ -568,7 +569,7 @@ fn commit_batch

( folded: usize, fault: &mut AdapterFaultState, sync_fault: &AtomicBool, - freeze_logged: &mut bool, + freeze_logged: &AtomicBool, settled: &mut Vec, ) -> BatchDiagnostics where @@ -648,8 +649,7 @@ where } // 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.swap(true, Ordering::Relaxed) { 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 \ @@ -2476,12 +2476,14 @@ mod tests { /// observable — inline, that one worker sits inside `store()` and nothing /// else on the runtime can advance. /// - /// The release is driven from a plain `std::thread`, and the verdict is - /// read there too, because a regression parks the only worker: the test - /// body cannot run, and neither can a `tokio::time::timeout` — the timer - /// needs the same worker to fire. An in-runtime deadline would therefore - /// hang CI instead of failing it, which is exactly what the bounded waits - /// elsewhere in this file exist to avoid. + /// Everything that decides the verdict happens on a plain `std::thread`, + /// for two reasons. The sentinel is spawned only after the store is + /// observed blocked: scheduled beforehand, the runtime could poll it first + /// and set the flag while nothing was blocking yet, so the test would pass + /// inline too. And a regression parks the only worker, so neither the test + /// body nor a `tokio::time::timeout` can run — the timer needs that same + /// worker — which would hang CI instead of failing it, the outcome the + /// bounded waits elsewhere in this file exist to avoid. #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn a_blocking_store_does_not_park_the_runtime() { let wallet_id = [0xB1u8; 32]; @@ -2500,34 +2502,37 @@ mod tests { cancel.clone(), )); - // Set before the store blocks, so it is waiting on the runtime rather - // than racing to be scheduled: whether it ran is then purely a question - // of the worker being free. - let progressed = Arc::new(AtomicBool::new(false)); - let flag = Arc::clone(&progressed); - tokio::spawn(async move { - flag.store(true, Ordering::Relaxed); - }); - tx.send(block_processed_event(wallet_id, 10)).unwrap(); - // Outside the runtime entirely: waits for the store to park, records - // whether the unrelated task got to run while it was parked, then - // releases it so the test always terminates either way. + let runtime = tokio::runtime::Handle::current(); + let progressed = Arc::new(AtomicBool::new(false)); let observer_progressed = Arc::clone(&progressed); let observer_blocked = Arc::clone(&blocked); let watcher = std::thread::spawn(move || { - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); - while !observer_blocked.load(Ordering::Relaxed) { - if std::time::Instant::now() > deadline { - return Err("store never entered its block"); + let wait_for = |flag: &AtomicBool, secs: u64| { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(secs); + while !flag.load(Ordering::Relaxed) { + if std::time::Instant::now() > deadline { + return false; + } + std::thread::sleep(std::time::Duration::from_millis(10)); } - std::thread::sleep(std::time::Duration::from_millis(10)); + true + }; + + if !wait_for(&observer_blocked, 10) { + drop(unblock); + return Err("store never entered its block"); } - // Give the runtime a moment it does not need when it is healthy; - // parked, no amount of waiting would help. - std::thread::sleep(std::time::Duration::from_millis(200)); - let ran_while_blocked = observer_progressed.load(Ordering::Relaxed); + + // Scheduled from outside, with the store already parked: whether it + // runs is now purely a question of the worker being free. + let flag = Arc::clone(&observer_progressed); + runtime.spawn(async move { + flag.store(true, Ordering::Relaxed); + }); + + let ran_while_blocked = wait_for(&observer_progressed, 5); drop(unblock); Ok(ran_while_blocked) }); @@ -3184,7 +3189,7 @@ mod tests { let persister = ProbePersister::new(obs_tx); let sync_fault = AtomicBool::new(false); let mut fault = AdapterFaultState::default(); - let mut freeze_logged = false; + let freeze_logged = AtomicBool::new(false); let mut batch = BTreeMap::new(); batch.insert( @@ -3200,7 +3205,7 @@ mod tests { 1, &mut fault, &sync_fault, - &mut freeze_logged, + &freeze_logged, &mut Vec::new(), ); @@ -3265,7 +3270,7 @@ mod tests { let persister = ProbePersister::new(obs_tx); let sync_fault = AtomicBool::new(false); let mut fault = AdapterFaultState::default(); - let mut freeze_logged = false; + let freeze_logged = AtomicBool::new(false); let diag = commit_batch( &persister, @@ -3273,7 +3278,7 @@ mod tests { 1, &mut fault, &sync_fault, - &mut freeze_logged, + &freeze_logged, &mut Vec::new(), ); @@ -3297,7 +3302,7 @@ mod tests { persister.fail_next(wallet_id); let sync_fault = AtomicBool::new(false); let mut fault = AdapterFaultState::default(); - let mut freeze_logged = false; + let freeze_logged = AtomicBool::new(false); let diag = commit_batch( &persister, @@ -3305,7 +3310,7 @@ mod tests { 1, &mut fault, &sync_fault, - &mut freeze_logged, + &freeze_logged, &mut Vec::new(), ); @@ -3350,7 +3355,7 @@ mod tests { assert!(fault.is_faulted(&wallet_id)); assert!(sync_fault.load(Ordering::Relaxed)); assert!( - freeze_logged, + freeze_logged.load(Ordering::Relaxed), "the one-shot SYNC WATERMARK FROZEN marker must have been emitted" ); } @@ -3367,7 +3372,7 @@ mod tests { 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 freeze_logged = AtomicBool::new(true); // one-shot already spent let diag = commit_batch( &persister, @@ -3375,7 +3380,7 @@ mod tests { 1, &mut fault, &sync_fault, - &mut freeze_logged, + &freeze_logged, &mut Vec::new(), ); @@ -3411,7 +3416,7 @@ mod tests { let sync_fault = AtomicBool::new(false); let mut fault = AdapterFaultState::default(); fault.fault_wallet(wallet_id, &sync_fault); - let mut freeze_logged = true; + let freeze_logged = AtomicBool::new(true); let core = CoreChangeSet { synced_height: Some(1234), @@ -3423,7 +3428,7 @@ mod tests { 1, &mut fault, &sync_fault, - &mut freeze_logged, + &freeze_logged, &mut Vec::new(), ); @@ -3448,7 +3453,7 @@ mod tests { persister.fail_next(rejecting); let sync_fault = AtomicBool::new(false); let mut fault = AdapterFaultState::default(); - let mut freeze_logged = false; + let freeze_logged = AtomicBool::new(false); let mut batch = BTreeMap::new(); batch.extend(one_wallet_batch(healthy, watermark_with_rows(10, 10))); @@ -3460,7 +3465,7 @@ mod tests { 2, &mut fault, &sync_fault, - &mut freeze_logged, + &freeze_logged, &mut Vec::new(), ); @@ -3496,7 +3501,7 @@ mod tests { 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; + let freeze_logged = AtomicBool::new(true); let diag = commit_batch( &persister, @@ -3504,7 +3509,7 @@ mod tests { 1, &mut fault, &sync_fault, - &mut freeze_logged, + &freeze_logged, &mut Vec::new(), ); From e04aeb3f37a4aff75531ec63b53559bec0addd7b Mon Sep 17 00:00:00 2001 From: Roman <51091564+jeanpierreroma@users.noreply.github.com> Date: Tue, 18 Aug 2026 14:50:31 +0300 Subject: [PATCH 6/7] docs(platform-wallet): name the commit panic as a fault trigger too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The invariant comments all still said a rejected `store()` was the only thing left that faults a wallet — true before this branch, and stale since the panic-recovery branch started faulting every persistable wallet whose outcome the commit never settled. A comment that describes an invariant the code no longer holds is worse than none: the next reader reasons from it. Updated at all five places the reviewer found — the `AdapterFaultState` docs and its two field/method comments, the adapter's fail-closed contract, the commit-loop comment, the watermark guard, and the fault test — and each now distinguishes the two cases rather than lumping them: a returned error faults a wallet whose rows are known not to be on disk, a panic faults wallets whose rows have an unknown fate. Comments only; no behaviour change. 666 tests pass. --- .../src/changeset/core_bridge.rs | 58 ++++++++++++------- 1 file changed, 36 insertions(+), 22 deletions(-) diff --git a/packages/rs-platform-wallet/src/changeset/core_bridge.rs b/packages/rs-platform-wallet/src/changeset/core_bridge.rs index d2eb7c5ccbc..56c096d4d03 100644 --- a/packages/rs-platform-wallet/src/changeset/core_bridge.rs +++ b/packages/rs-platform-wallet/src/changeset/core_bridge.rs @@ -91,11 +91,17 @@ const ADAPTER_STORE_BATCH_LIMIT: usize = 512; /// Session fault state for the durable-watermark guard /// (dashpay/platform#4069). /// -/// 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. +/// Now that the persistence channel is a lossless unbounded `mpsc`, two things +/// fault a wallet, and both name the wallets they hit, so a sibling 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**, which carries a `wallet_id`: that wallet's rows +/// are known not to be on disk. +/// - A **panic in the blocking commit thread**, which faults every wallet with +/// something to persist that is absent from `settled` — the one that panicked +/// plus every wallet the loop never reached. Their outcome is unknown rather +/// than known-bad, and unknown must fail closed the same way. /// /// The old global (`broadcast::Lagged`) latch is gone: the unbounded channel /// can never `Lagged`, so there is no more "dropped events of unknown wallet" @@ -103,7 +109,8 @@ const ADAPTER_STORE_BATCH_LIMIT: usize = 512; /// fail-closed backstop — in a healthy run it never fires. #[derive(Default)] struct AdapterFaultState { - /// Set by a `store()` rejection: freezes only the named wallet. + /// Set by a `store()` rejection, or by the panic-recovery branch for a + /// wallet whose commit outcome is unknown: freezes only the named wallets. per_wallet: HashMap, } @@ -113,8 +120,9 @@ impl AdapterFaultState { self.per_wallet.get(wallet_id).copied().unwrap_or(false) } - /// Fault a single wallet after its `store()` was rejected, and raise - /// the host-visible hard-fault signal. + /// Fault a single wallet — after its `store()` was rejected, or after a + /// commit panic left its outcome unknown — and raise the host-visible + /// hard-fault signal. fn fault_wallet(&mut self, wallet_id: WalletId, hard_signal: &AtomicBool) { self.per_wallet.insert(wallet_id, true); hard_signal.store(true, Ordering::Relaxed); @@ -266,9 +274,11 @@ where /// /// # Durable-watermark guard (fail-closed backstop) /// -/// 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`] +/// Two things fault a wallet: a rejected `store()` (the rows for that batch +/// are not on disk) and a panic in the blocking commit thread (the rows for +/// every persistable wallet the commit did not settle have an unknown fate, +/// which fails closed the same way). When a wallet faults either 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 @@ -276,8 +286,9 @@ where /// (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. +/// never fires, since the channel is lossless, and both triggers mean a +/// genuine backend error rather than overload — a rejection is one the store +/// reported, a panic one it could not. /// /// When a wallet faults, the task raises `sync_fault` (an `AtomicBool` the /// host polls via `PlatformWalletManager::sync_fault_detected`) and logs a @@ -363,9 +374,10 @@ 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 inside `commit_batch`). + // Commit the folded batch. The channel is lossless, so a watermark is + // held back only by the fail-closed backstops around this call: a + // rejected `store()` inside `commit_batch`, or a panic in the commit + // thread, handled in the `Err` arm below. // Commit on a blocking thread, never on the async worker. // // `store()` is synchronous and, for the SQLite backend, commits a real @@ -671,10 +683,11 @@ where /// Durable-watermark guard for dashpay/platform#4069. /// -/// 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 +/// When a wallet has faulted this session — its `store()` was rejected, or a +/// commit panic left the batch's outcome unknown — 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 @@ -2803,8 +2816,9 @@ mod tests { 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). + // Fault the wallet via a rejected store — the trigger with a known + // outcome. The other one, a commit panic, is covered by + // `a_panicking_commit_freezes_the_batch_wallets`. persister.fail_next(wallet_id); let sync_fault = Arc::new(AtomicBool::new(false)); let cancel = CancellationToken::new(); From 75ac76c7b441cadd720164646418cc14491bbf44 Mon Sep 17 00:00:00 2001 From: Roman <51091564+jeanpierreroma@users.noreply.github.com> Date: Tue, 18 Aug 2026 16:04:36 +0300 Subject: [PATCH 7/7] test(platform-wallet): cover the wallet a panicking commit never reached MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review points, and the first one corrects a claim I made in an earlier commit message on this branch. **The duplicate off-runtime test was real.** I said no such test existed and skipped the suggestion; the check behind that was `grep "async fn a_block"`, and `a_blocked_store_does_not_park_the_runtime` is a plain `fn` that owns its runtime, so it never matched. It has been here since `4e5a1939b`, and it is the better of the two — explicit runtime, `std` primitives, and it states the discriminator directly. Mine is removed. **The panic tests had a hole.** They covered the wallet whose `store()` panicked and a wallet settled before it, but not a persistable wallet ordered after the panic. That is the other half of what `batch_wallet_ids - settled` computes: unwinding drops such a wallet's consumed changes before `store()` is attempted, so nothing knows whether its rows landed, and it must freeze. A handler faulting only the direct casualty passed every existing assertion. The test now drains three wallets in one batch — settled, panicking, and never-reached — and requires the third to be frozen too. Checked by truncating the unsettled set to one entry: fails on exactly that assertion. The opening comment claimed the test was about wallets the panic "did not reach" being spared, which is backwards for that half, and now describes both sides. 665 tests pass. --- .../src/changeset/core_bridge.rs | 141 +++++------------- 1 file changed, 38 insertions(+), 103 deletions(-) diff --git a/packages/rs-platform-wallet/src/changeset/core_bridge.rs b/packages/rs-platform-wallet/src/changeset/core_bridge.rs index 56c096d4d03..cebbd72113a 100644 --- a/packages/rs-platform-wallet/src/changeset/core_bridge.rs +++ b/packages/rs-platform-wallet/src/changeset/core_bridge.rs @@ -2479,102 +2479,6 @@ mod tests { /// let the NEXT batch persist a higher `synced_height` for a wallet whose /// earlier rows may never have landed, which is exactly the hole /// dashpay/platform#4069 closed. - /// The commit must not run on a Tokio worker. - /// - /// This is the behaviour the `spawn_blocking` move exists for, and no - /// other test covers it: the outcome-only assertions elsewhere all still - /// pass with `commit_batch` called inline, because a blocking store on a - /// multi-worker runtime merely steals one worker and the rest of the test - /// proceeds. Pinning the runtime to a single worker makes the difference - /// observable — inline, that one worker sits inside `store()` and nothing - /// else on the runtime can advance. - /// - /// Everything that decides the verdict happens on a plain `std::thread`, - /// for two reasons. The sentinel is spawned only after the store is - /// observed blocked: scheduled beforehand, the runtime could poll it first - /// and set the flag while nothing was blocking yet, so the test would pass - /// inline too. And a regression parks the only worker, so neither the test - /// body nor a `tokio::time::timeout` can run — the timer needs that same - /// worker — which would hang CI instead of failing it, the outcome the - /// bounded waits elsewhere in this file exist to avoid. - #[tokio::test(flavor = "multi_thread", worker_threads = 1)] - async fn a_blocking_store_does_not_park_the_runtime() { - let wallet_id = [0xB1u8; 32]; - let (tx, rx) = unbounded_channel::(); - - let (obs_tx, mut obs_rx) = unbounded_channel(); - let persister = Arc::new(ProbePersister::new(obs_tx)); - let (unblock, blocked) = persister.block_next(); - 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(), - )); - - tx.send(block_processed_event(wallet_id, 10)).unwrap(); - - let runtime = tokio::runtime::Handle::current(); - let progressed = Arc::new(AtomicBool::new(false)); - let observer_progressed = Arc::clone(&progressed); - let observer_blocked = Arc::clone(&blocked); - let watcher = std::thread::spawn(move || { - let wait_for = |flag: &AtomicBool, secs: u64| { - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(secs); - while !flag.load(Ordering::Relaxed) { - if std::time::Instant::now() > deadline { - return false; - } - std::thread::sleep(std::time::Duration::from_millis(10)); - } - true - }; - - if !wait_for(&observer_blocked, 10) { - drop(unblock); - return Err("store never entered its block"); - } - - // Scheduled from outside, with the store already parked: whether it - // runs is now purely a question of the worker being free. - let flag = Arc::clone(&observer_progressed); - runtime.spawn(async move { - flag.store(true, Ordering::Relaxed); - }); - - let ran_while_blocked = wait_for(&observer_progressed, 5); - drop(unblock); - Ok(ran_while_blocked) - }); - - let observed = obs_rx - .recv() - .await - .expect("the released store should report"); - assert_eq!(observed.wallet_id, wallet_id); - - let ran_while_blocked = watcher - .join() - .expect("watcher thread should not panic") - .expect("store never entered its block"); - assert!( - ran_while_blocked, - "an unrelated task must make progress while the commit blocks — \ - the commit is running on a runtime worker instead of a blocking thread" - ); - assert!( - !sync_fault.load(Ordering::Relaxed), - "a slow store is not a fault" - ); - - cancel.cancel(); - drop(tx); - let _ = handle.await; - } - #[tokio::test] async fn a_panicking_commit_freezes_the_batch_wallets() { let wallet_id = [0xEEu8; 32]; @@ -2718,13 +2622,22 @@ mod tests { }); } - /// (i) A commit panic must not punish the wallets it did not reach. + /// (i) A commit panic must punish exactly the wallets whose outcome it + /// left unknown — no more, no less. /// /// `commit_batch` walks the batch serially (a `BTreeMap`, so in wallet-id - /// order). If an earlier wallet's `store()` returned and a later one - /// panics, the earlier wallet's rows are on disk and its watermark is - /// safe — freezing it would strip its `synced_height` for the rest of the - /// session over a sibling's bad batch. + /// order), and a panic cuts it in two. A wallet whose `store()` already + /// returned is settled: its rows are on disk and its watermark is safe, + /// so freezing it would strip its `synced_height` for the rest of the + /// session over a sibling's bad batch. A wallet ordered AFTER the panic + /// is the opposite case — unwinding dropped its consumed changes before + /// `store()` was ever attempted, so nothing knows whether its rows + /// landed, and it must freeze or a later watermark advances past rows + /// that never existed. + /// + /// Both halves are checked here, because they are guarded by the same + /// `batch_wallet_ids - settled` expression and a regression that faults + /// only the wallet that actually panicked satisfies neither. /// /// Guards the fix for the first version of the panic handler, which /// faulted every wallet in the drain. @@ -2734,6 +2647,8 @@ mod tests { // chosen to put the healthy wallet ahead of the panicking one. let healthy = [0x11u8; 32]; let doomed = [0x22u8; 32]; + // Sorts after `doomed`, so the commit unwinds before it is reached. + let unreached = [0x33u8; 32]; let (tx, rx) = unbounded_channel::(); let (obs_tx, mut obs_rx) = unbounded_channel(); @@ -2749,10 +2664,12 @@ mod tests { cancel.clone(), )); - // Both wallets in one drain: `healthy` stores, then `doomed` panics. - // Sent before either is observed so they fold into a single batch. + // All three in one drain: `healthy` stores, `doomed` panics, and + // `unreached` never gets its turn. Sent before any is observed so they + // fold into a single batch. tx.send(block_processed_event(healthy, 10)).unwrap(); tx.send(block_processed_event(doomed, 10)).unwrap(); + tx.send(block_processed_event(unreached, 10)).unwrap(); tokio::time::timeout(std::time::Duration::from_secs(5), async { while !sync_fault.load(Ordering::Relaxed) { @@ -2796,6 +2713,24 @@ mod tests { "the wallet whose commit panicked must not advance its watermark" ); + // The wallet the commit never reached must be frozen too: its changes + // went down with the unwind without a `store()` ever being attempted, + // so its rows are exactly as unaccounted-for as the panicking + // wallet's. A handler that faults only the direct casualty leaves this + // one free to advance past rows that never landed. + tx.send(block_processed_event(unreached, 60)).unwrap(); + tx.send(sync_height_event(unreached, 900)).unwrap(); + let after_unreached = + tokio::time::timeout(std::time::Duration::from_secs(5), obs_rx.recv()) + .await + .expect("a faulted wallet must still persist its rows") + .expect("unreached wallet still persists rows"); + assert_eq!(after_unreached.wallet_id, unreached); + assert_eq!( + after_unreached.synced_height, None, + "a wallet the panicking commit never reached must not advance its watermark" + ); + cancel.cancel(); drop(tx); handle.await.unwrap();