From 85453370a42f4ccacaadca7da4dee255c3fad8d7 Mon Sep 17 00:00:00 2001 From: bfoss765 <38437574+bfoss765@users.noreply.github.com> Date: Tue, 4 Aug 2026 22:44:48 -0400 Subject: [PATCH 1/5] fix(key-wallet-manager): lossless persistence event channel to stop the sync-watermark freeze MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The wallet event bus is a single bounded `tokio::broadcast` ring (`DEFAULT_WALLET_EVENT_CAPACITY` = 1000), fanned out fire-and-forget from `process_block.rs`. It feeds two very different consumers: 1. incidental subscribers reached via `subscribe_events()` — dash-spv's `EventHandler` dispatch and the unit tests — for which a dropped event under overload is harmless; and 2. the platform durable-persistence consumer, for which a dropped event is NOT harmless: the dropped `TransactionDetected`/`BlockProcessed` rows never reach disk while the surviving `SyncHeightAdvanced` watermark keeps advancing, so the persisted sync height outruns the rows it implies. The platform `#4069` guard then latches that wallet's watermark into a *permanent* freeze for the rest of the process. In the field this is the mainnet sync that climbs, then "rolls back" to a stale height on every relaunch. Under a heavy historical SPV catch-up a burst larger than the ring between consumer drains overflows it (`RecvError::Lagged`) and trips exactly that freeze. Batching the persistence stores (shipped earlier) raised the burst threshold but a single lag still freezes forever. Fix: give the persistence consumer its own dedicated, unbounded `tokio::mpsc` channel that carries the same event stream, in the same order, with no drops — so it can never observe `Lagged` and its freeze guard never fires. The existing broadcast is kept unchanged for the incidental subscribers (no behavioural change for dash-spv or tests). Why unbounded rather than a bounded back-pressuring channel: several emit sites run inside this manager's `RwLock` write guard (SPV block processing holds `wallet.write().await` across `process_block_for_wallets`), while the persistence consumer needs a `read()` lock on the *same* manager to project each event. A bounded `send().await`/`blocking_send` that parked under the write guard would deadlock the very consumer that must drain it. A non-blocking unbounded enqueue keeps every emit path lock-safe while still guaranteeing losslessness. All emission now flows through a single `emit_event` choke point that pushes to both channels in order. `#4069`-safety: unchanged ordering + zero drops means the persistence consumer sees every row event before the watermark event that implies it, exactly as the non-lagged broadcast path already guaranteed; the durable watermark can never outrun its rows. - lib.rs: add `persistence_sender`/`persistence_receiver` + `emit_event`. - accessors.rs: add `take_persistence_receiver()` (handed to the consumer once). - process_block.rs: route all six emit sites through `emit_event`. - event_tests.rs: prove a 5000-event burst is delivered losslessly and in order on the persistence channel while the bounded broadcast lags. Co-Authored-By: Claude Opus 4.8 --- key-wallet-manager/src/accessors.rs | 16 ++++- key-wallet-manager/src/event_tests.rs | 82 +++++++++++++++++++++++++ key-wallet-manager/src/lib.rs | 71 ++++++++++++++++++++- key-wallet-manager/src/process_block.rs | 21 +++++-- 4 files changed, 181 insertions(+), 9 deletions(-) diff --git a/key-wallet-manager/src/accessors.rs b/key-wallet-manager/src/accessors.rs index db66cb763..4d6fedbc4 100644 --- a/key-wallet-manager/src/accessors.rs +++ b/key-wallet-manager/src/accessors.rs @@ -7,7 +7,7 @@ use key_wallet::wallet::managed_wallet_info::wallet_info_interface::WalletInfoIn use key_wallet::wallet::managed_wallet_info::TransactionRecord; use key_wallet::{Account, Address, Network, Utxo, Wallet}; use std::collections::{BTreeMap, BTreeSet}; -use tokio::sync::broadcast; +use tokio::sync::{broadcast, mpsc}; impl WalletManager { /// Get a wallet by ID @@ -209,6 +209,20 @@ impl WalletManager { &self.event_sender } + /// Take the lossless persistence-event receiver (dashpay/platform#4069). + /// + /// Returns the receive half of the unbounded persistence channel exactly + /// once; every subsequent call returns `None`. The platform durable + /// consumer calls this before the manager is shared with any producer, + /// then drains the stream losslessly (see the `persistence_sender` field + /// docs). Because it is an `mpsc::UnboundedReceiver`, events emitted before + /// the consumer starts draining are buffered rather than lost, so there is + /// no subscribe-before-publish startup race as there is with + /// [`subscribe_events`](Self::subscribe_events). + pub fn take_persistence_receiver(&mut self) -> Option> { + self.persistence_receiver.take() + } + /// Return the total monitor revision (structural + per-wallet account revisions). pub fn monitor_revision(&self) -> u64 { self.structural_revision diff --git a/key-wallet-manager/src/event_tests.rs b/key-wallet-manager/src/event_tests.rs index 05d89c2f0..e6ea061e7 100644 --- a/key-wallet-manager/src/event_tests.rs +++ b/key-wallet-manager/src/event_tests.rs @@ -1339,3 +1339,85 @@ async fn test_block_processing_stamps_in_block_position() { "second tx in block.txdata must be stamped position 1" ); } + +// --------------------------------------------------------------------------- +// Lossless persistence channel (dashpay/platform#4069) +// --------------------------------------------------------------------------- + +/// A large burst of watermark events that the durable-persistence consumer +/// does not drain until the very end is delivered **losslessly and in order** +/// over the unbounded persistence channel — the exact property the platform +/// consumer's durable sync watermark relies on. The bounded broadcast, by +/// contrast, drops events (`Lagged`) under the same burst; that drop is the +/// freeze root cause this dedicated channel removes. +/// +/// This models a stalled/slow persistence consumer during a heavy SPV +/// catch-up: neither receiver is drained while `BURST` monotonically +/// increasing `SyncHeightAdvanced` watermarks are emitted. On the unbounded +/// channel every watermark survives, so the watermark can keep advancing to +/// the tip after the stall clears; on the broadcast it lags and the watermark +/// would freeze. +#[tokio::test] +async fn persistence_channel_is_lossless_under_a_large_burst() { + // `BURST` far exceeds the broadcast ring (`DEFAULT_WALLET_EVENT_CAPACITY` + // == 1000), so the bounded broadcast is guaranteed to lag. + const BURST: u32 = 5000; + + let (mut manager, wallet_id, _addr) = setup_manager_with_wallet(); + // Take the lossless persistence receiver AND subscribe a bounded broadcast + // receiver before emitting; neither is drained during the burst. + let mut persistence_rx = + manager.take_persistence_receiver().expect("persistence receiver available once"); + let mut broadcast_rx = manager.subscribe_events(); + + for h in 1..=BURST { + manager.update_wallet_synced_height(&wallet_id, h); + } + + // Persistence channel: every watermark arrived, in order, no gaps. + let mut received: Vec = Vec::with_capacity(BURST as usize); + while let Ok(event) = persistence_rx.try_recv() { + match event { + WalletEvent::SyncHeightAdvanced { + wallet_id: w, + height, + } => { + assert_eq!(w, wallet_id, "watermark for the wrong wallet"); + received.push(height); + } + other => panic!("unexpected event on persistence channel: {other:?}"), + } + } + assert_eq!( + received.len(), + BURST as usize, + "persistence channel dropped events: got {} of {BURST}", + received.len() + ); + assert!( + received.windows(2).all(|w| w[0] + 1 == w[1]), + "persistence channel reordered or gapped the watermark stream" + ); + assert_eq!(received.first().copied(), Some(1)); + assert_eq!(received.last().copied(), Some(BURST), "final watermark must reach the tip"); + + // Broadcast channel: the same burst overflows the bounded ring and drops + // events — demonstrating why the persistence consumer must NOT use it. + let mut broadcast_lagged = false; + let mut broadcast_delivered = 0usize; + loop { + match broadcast_rx.try_recv() { + Ok(_) => broadcast_delivered += 1, + Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) => broadcast_lagged = true, + Err(_) => break, + } + } + assert!( + broadcast_lagged, + "the bounded broadcast should have lagged under a {BURST}-event burst" + ); + assert!( + broadcast_delivered < BURST as usize, + "broadcast should have dropped events but delivered all {BURST}" + ); +} diff --git a/key-wallet-manager/src/lib.rs b/key-wallet-manager/src/lib.rs index b3337e7a3..e9b772141 100644 --- a/key-wallet-manager/src/lib.rs +++ b/key-wallet-manager/src/lib.rs @@ -40,7 +40,7 @@ use std::str::FromStr; use dashcore::address::NetworkUnchecked; use key_wallet::wallet::managed_wallet_info::fee::FeeRate; -use tokio::sync::broadcast; +use tokio::sync::{broadcast, mpsc}; /// Default capacity for the wallet event bus. const DEFAULT_WALLET_EVENT_CAPACITY: usize = 1000; @@ -127,22 +127,89 @@ pub struct WalletManager, + /// Lossless, unbounded persistence event channel (dashpay/platform#4069). + /// + /// Carries the *same* event stream, in the *same* order, to the single + /// durable-persistence consumer with **no drops**. The platform consumer + /// projects each event into a persisted changeset; a dropped + /// record/watermark event there lets the durable sync watermark advance + /// past rows that never reached disk, which its `#4069` guard then latches + /// into a permanent freeze. An unbounded `mpsc` can never `Lagged`, so that + /// guard never fires. + /// + /// It is deliberately unbounded rather than a bounded back-pressuring + /// channel: several emit sites run *inside this manager's `RwLock` write + /// guard* (SPV block processing holds `wallet.write().await` across + /// `process_block_for_wallets`), while the consumer needs a `read()` lock + /// on the same manager to project each event. A bounded + /// `send().await`/`blocking_send` that parked under the write guard would + /// therefore deadlock the very consumer that must drain it. A non-blocking + /// unbounded enqueue keeps the emit paths lock-safe. See [`emit_event`]. + /// + /// [`emit_event`]: WalletManager::emit_event + persistence_sender: mpsc::UnboundedSender, + /// Receive half of `persistence_sender`, handed to the persistence + /// consumer exactly once via [`take_persistence_receiver`]; `None` + /// afterwards. Held here rather than returned from [`new`] so the + /// `WalletManager::new(network)` signature is unchanged. Unlike a + /// `broadcast::Receiver`, this buffers events emitted before the consumer + /// starts draining, so there is no subscribe-before-publish race. + /// + /// [`take_persistence_receiver`]: WalletManager::take_persistence_receiver + /// [`new`]: WalletManager::new + persistence_receiver: Option>, } impl WalletManager { /// Create a new wallet manager pub fn new(network: Network) -> Self { + let (persistence_sender, persistence_receiver) = mpsc::unbounded_channel(); Self { network, wallets: BTreeMap::new(), wallet_infos: BTreeMap::new(), structural_revision: 0, event_sender: broadcast::Sender::new(DEFAULT_WALLET_EVENT_CAPACITY), + persistence_sender, + persistence_receiver: Some(persistence_receiver), } } + /// Emit a wallet event to BOTH the incidental broadcast fan-out and the + /// lossless persistence channel (dashpay/platform#4069). + /// + /// This is the single emit choke point, so both channels observe events in + /// the same order (the manager is the only producer). The persistence send + /// is a non-blocking, non-dropping `mpsc` enqueue: it is safe to call from + /// the synchronous emit paths that run while this manager's `RwLock` write + /// guard is held, whereas a bounded blocking send there would deadlock the + /// persistence consumer (which needs a `read()` lock to drain). See the + /// `persistence_sender` field docs for the full rationale. + fn emit_event(&self, event: WalletEvent) { + // Lossless path: the persistence consumer must never miss a + // row-bearing or watermark event, or its durable sync height freezes. + // `send` only fails once the consumer half has been dropped (manager + // shutdown) — at which point there is nothing left to persist to. + let _ = self.persistence_sender.send(event.clone()); + // Lossy-tolerant fan-out for incidental subscribers (dash-spv + // `EventHandler` dispatch, tests). A `Lagged` drop here never affects + // the durable watermark. + let _ = self.event_sender.send(event); + } + /// Increment the structural revision for wallet/account additions or removals. fn bump_structural_revision(&mut self) { self.structural_revision += 1; diff --git a/key-wallet-manager/src/process_block.rs b/key-wallet-manager/src/process_block.rs index 11db8aeef..b751ada64 100644 --- a/key-wallet-manager/src/process_block.rs +++ b/key-wallet-manager/src/process_block.rs @@ -170,7 +170,7 @@ impl WalletInterface for WalletM account_balances: account_balances.clone(), addresses_derived: project_derived_addresses(for_record), }; - let _ = self.event_sender.send(event); + self.emit_event(event); } // If any derivations were left unattributed (records vector // didn't cover every account that derived), log so the @@ -204,7 +204,7 @@ impl WalletInterface for WalletM balance, account_balances: account_balances.clone(), }; - let _ = self.event_sender.send(event); + self.emit_event(event); } } } @@ -282,7 +282,7 @@ impl WalletInterface for WalletM if let Some(info) = self.wallet_infos.get_mut(wallet_id) { if height > info.synced_height() { info.update_synced_height(height); - let _ = self.event_sender.send(WalletEvent::SyncHeightAdvanced { + self.emit_event(WalletEvent::SyncHeightAdvanced { wallet_id: *wallet_id, height, }); @@ -314,6 +314,12 @@ impl WalletInterface for WalletM } fn apply_chain_lock(&mut self, chain_lock: ChainLock) { + // Collect the events under the `iter_mut` borrow, then emit them once + // the mutable borrow of `self.wallet_infos` has ended. `emit_event` + // takes `&self`, which cannot overlap the live `iter_mut` borrow; the + // BTreeMap iteration order is preserved, so the emit order is + // unchanged. + let mut events = Vec::new(); for (wallet_id, info) in self.wallet_infos.iter_mut() { let outcome = info.apply_chain_lock(chain_lock.clone()); @@ -323,13 +329,16 @@ impl WalletInterface for WalletM // promoted nothing). Replays of the same chainlock (no // metadata advance) are silent. if outcome.metadata_advanced { - let _ = self.event_sender.send(WalletEvent::ChainLockProcessed { + events.push(WalletEvent::ChainLockProcessed { wallet_id: *wallet_id, chain_lock: chain_lock.clone(), locked_transactions: outcome.locked_transactions, }); } } + for event in events { + self.emit_event(event); + } } fn process_instant_send_lock(&mut self, instant_lock: InstantLock) { @@ -361,7 +370,7 @@ impl WalletInterface for WalletM }; let prior = prior_account_balances.remove(&wallet_id).unwrap_or_default(); let account_balances = diff_account_balances(&prior, &info.account_balances()); - let _ = self.event_sender().send(WalletEvent::TransactionInstantLocked { + self.emit_event(WalletEvent::TransactionInstantLocked { wallet_id, txid, instant_lock: instant_lock.clone(), @@ -501,7 +510,7 @@ impl WalletManager { account_balances, addresses_derived, }; - let _ = self.event_sender.send(event); + self.emit_event(event); } } } From c76d17238589f6c38727600dd3e678e17b920486 Mon Sep 17 00:00:00 2001 From: bfoss765 <38437574+bfoss765@users.noreply.github.com> Date: Wed, 5 Aug 2026 07:16:02 -0400 Subject: [PATCH 2/5] fix(key-wallet-manager): make the persistence channel opt-in and surface a lost consumer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses two CodeRabbit findings on the lossless persistence mpsc: - Stability & Availability: `WalletManager::new` unconditionally created and retained `mpsc::unbounded_channel()`, so a manager whose consumer is never installed (only tests call `take_persistence_receiver`) would accumulate every emitted event without bound. Make persistence delivery opt-in: the channel is now created lazily by `take_persistence_receiver` (the send half is `None` until a consumer installs itself), so `emit_event` enqueues nothing when no consumer exists. Unbounded + non-blocking is retained for installed consumers, which is required: several emit sites run inside the manager's RwLock write guard while the consumer needs a read lock to drain, so a bounded blocking send would deadlock, and a bounded try_send that dropped would reintroduce the watermark freeze (#4069). The documented take-before-emit contract means no pre-consumer events are lost. - Data Integrity: `emit_event` ignored `SendError`. If the installed consumer drops its receiver while the manager is still running, durable persistence has silently stopped. Surface it: log once (latched) that the consumer was lost. We deliberately do not halt in-memory advancement — the manager does not own durable state, and on restart the wallet re-scans from the last persisted height, so a lost consumer causes no durable corruption. All key-wallet-manager tests pass, including the lossless-burst persistence test. Co-Authored-By: Claude Opus 4.8 --- key-wallet-manager/src/accessors.rs | 33 ++++++++++----- key-wallet-manager/src/lib.rs | 65 +++++++++++++++++++++-------- 2 files changed, 70 insertions(+), 28 deletions(-) diff --git a/key-wallet-manager/src/accessors.rs b/key-wallet-manager/src/accessors.rs index 4d6fedbc4..970de1f3c 100644 --- a/key-wallet-manager/src/accessors.rs +++ b/key-wallet-manager/src/accessors.rs @@ -209,18 +209,31 @@ impl WalletManager { &self.event_sender } - /// Take the lossless persistence-event receiver (dashpay/platform#4069). + /// Install a durable persistence consumer and take its event receiver + /// (dashpay/platform#4069). /// - /// Returns the receive half of the unbounded persistence channel exactly - /// once; every subsequent call returns `None`. The platform durable - /// consumer calls this before the manager is shared with any producer, - /// then drains the stream losslessly (see the `persistence_sender` field - /// docs). Because it is an `mpsc::UnboundedReceiver`, events emitted before - /// the consumer starts draining are buffered rather than lost, so there is - /// no subscribe-before-publish startup race as there is with - /// [`subscribe_events`](Self::subscribe_events). + /// Persistence delivery is opt-in: the first call **creates** the lossless, + /// unbounded persistence channel, installs the send half on the manager, and + /// returns the receive half; every subsequent call returns `None`. Creating + /// the channel only when a consumer asks for it means a manager that never + /// installs a consumer never accumulates an undrained backlog of events (an + /// unbounded `mpsc` with no reader would otherwise grow without limit). + /// + /// The platform durable consumer calls this before the manager is shared + /// with any producer, then drains the stream losslessly (see the + /// `persistence_sender` field docs). Because installation precedes emission, + /// no events are emitted before the consumer is in place; unlike + /// [`subscribe_events`](Self::subscribe_events) there is no + /// subscribe-before-publish race, and unlike a `broadcast::Receiver` the + /// unbounded `mpsc` never `Lagged`s a row-bearing or watermark event. pub fn take_persistence_receiver(&mut self) -> Option> { - self.persistence_receiver.take() + if self.persistence_sender.is_some() { + // A consumer has already been installed; the receiver is taken once. + return None; + } + let (sender, receiver) = mpsc::unbounded_channel(); + self.persistence_sender = Some(sender); + Some(receiver) } /// Return the total monitor revision (structural + per-wallet account revisions). diff --git a/key-wallet-manager/src/lib.rs b/key-wallet-manager/src/lib.rs index e9b772141..6efe950a0 100644 --- a/key-wallet-manager/src/lib.rs +++ b/key-wallet-manager/src/lib.rs @@ -159,32 +159,39 @@ pub struct WalletManager, - /// Receive half of `persistence_sender`, handed to the persistence - /// consumer exactly once via [`take_persistence_receiver`]; `None` - /// afterwards. Held here rather than returned from [`new`] so the - /// `WalletManager::new(network)` signature is unchanged. Unlike a - /// `broadcast::Receiver`, this buffers events emitted before the consumer - /// starts draining, so there is no subscribe-before-publish race. + /// Persistence delivery is **opt-in**: this is `None` until a durable + /// consumer installs itself by calling [`take_persistence_receiver`], which + /// creates the channel and returns the receive half. Until then + /// [`emit_event`] enqueues nothing, so a `WalletManager` used without a + /// persistence consumer (e.g. a pure wallet-management embedding that never + /// drives block processing, or one that simply never installs a consumer) + /// cannot accumulate an unbounded backlog of undrained events. The + /// documented contract is that the consumer takes the receiver *before* the + /// manager is shared with any producer, so no pre-consumer events are lost. /// /// [`take_persistence_receiver`]: WalletManager::take_persistence_receiver - /// [`new`]: WalletManager::new - persistence_receiver: Option>, + /// [`emit_event`]: WalletManager::emit_event + persistence_sender: Option>, + /// Latches `true` the first time a persistence send fails because the + /// installed consumer dropped its receiver while the manager is still + /// running, so that anomaly is surfaced (logged) exactly once instead of on + /// every subsequent emit. + persistence_consumer_lost: std::sync::atomic::AtomicBool, } impl WalletManager { /// Create a new wallet manager pub fn new(network: Network) -> Self { - let (persistence_sender, persistence_receiver) = mpsc::unbounded_channel(); Self { network, wallets: BTreeMap::new(), wallet_infos: BTreeMap::new(), structural_revision: 0, event_sender: broadcast::Sender::new(DEFAULT_WALLET_EVENT_CAPACITY), - persistence_sender, - persistence_receiver: Some(persistence_receiver), + // Persistence delivery is opt-in; the channel is created lazily when + // a consumer calls `take_persistence_receiver`. See that field's docs. + persistence_sender: None, + persistence_consumer_lost: std::sync::atomic::AtomicBool::new(false), } } @@ -199,11 +206,33 @@ impl WalletManager { /// persistence consumer (which needs a `read()` lock to drain). See the /// `persistence_sender` field docs for the full rationale. fn emit_event(&self, event: WalletEvent) { - // Lossless path: the persistence consumer must never miss a - // row-bearing or watermark event, or its durable sync height freezes. - // `send` only fails once the consumer half has been dropped (manager - // shutdown) — at which point there is nothing left to persist to. - let _ = self.persistence_sender.send(event.clone()); + // Lossless path: the persistence consumer must never miss a row-bearing + // or watermark event, or its durable sync height freezes. Only enqueue + // once a consumer has installed itself (opt-in), so a manager with no + // persistence consumer never accumulates an undrained backlog. + if let Some(sender) = &self.persistence_sender { + // `send` fails only after the consumer dropped its receiver. Under + // the documented contract that happens at manager shutdown, when + // there is nothing left to persist to. If it happens *while the + // manager is still running* (a consumer task that exited early), + // durable persistence has silently stopped: surface it loudly, once. + // We do not halt in-memory state advancement — the manager does not + // own durable state, and on restart the wallet re-scans from the + // last persisted height, so a lost consumer causes no durable + // corruption (dashpay/platform#4069). + if sender.send(event.clone()).is_err() + && !self + .persistence_consumer_lost + .swap(true, std::sync::atomic::Ordering::Relaxed) + { + tracing::error!( + "wallet-manager persistence consumer dropped its receiver while the manager is \ + still running; durable persistence has stopped. In-memory state keeps \ + advancing and is recovered by a re-scan from the last persisted height on \ + restart, so there is no durable corruption (dashpay/platform#4069)." + ); + } + } // Lossy-tolerant fan-out for incidental subscribers (dash-spv // `EventHandler` dispatch, tests). A `Lagged` drop here never affects // the durable watermark. From f3fdb9ca72b0bc411b72e62d87d6d16612f5ab25 Mon Sep 17 00:00:00 2001 From: bfoss765 <38437574+bfoss765@users.noreply.github.com> Date: Wed, 5 Aug 2026 07:21:38 -0400 Subject: [PATCH 3/5] style(key-wallet-manager): rustfmt the persistence-lost guard in emit_event Collapse the hand-wrapped `&&` condition onto one line to satisfy cargo fmt --check (the pre-commit gate). Formatting only; no behavior change. Co-Authored-By: Claude Opus 4.8 --- key-wallet-manager/src/lib.rs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/key-wallet-manager/src/lib.rs b/key-wallet-manager/src/lib.rs index 6efe950a0..4f9f09d96 100644 --- a/key-wallet-manager/src/lib.rs +++ b/key-wallet-manager/src/lib.rs @@ -221,9 +221,7 @@ impl WalletManager { // last persisted height, so a lost consumer causes no durable // corruption (dashpay/platform#4069). if sender.send(event.clone()).is_err() - && !self - .persistence_consumer_lost - .swap(true, std::sync::atomic::Ordering::Relaxed) + && !self.persistence_consumer_lost.swap(true, std::sync::atomic::Ordering::Relaxed) { tracing::error!( "wallet-manager persistence consumer dropped its receiver while the manager is \ From 85bb3fd2dab2e1e61d82c537b42f069638ab176f Mon Sep 17 00:00:00 2001 From: Quantum Explorer Date: Thu, 6 Aug 2026 16:30:00 +0700 Subject: [PATCH 4/5] fix(key-wallet-manager): warn when a persistence consumer installs after emission has begun MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Installing the persistence consumer after events were already emitted leaves those events permanently absent from the persistence stream with no Lagged-style marker — the exact invisible-loss failure mode the channel exists to eliminate. Latch events_emitted in emit_event and warn from take_persistence_receiver on a late install (the receiver is still returned; delivery is lossless from that point on). Also pin the opt-in contract with tests: the receiver is taken exactly once, a late install delivers only post-install events (no invented replay), and a dropped consumer neither wedges nor panics emission. Co-Authored-By: Claude Fable 5 --- key-wallet-manager/src/accessors.rs | 15 +++++ key-wallet-manager/src/event_tests.rs | 84 +++++++++++++++++++++++++++ key-wallet-manager/src/lib.rs | 11 ++++ 3 files changed, 110 insertions(+) diff --git a/key-wallet-manager/src/accessors.rs b/key-wallet-manager/src/accessors.rs index 970de1f3c..cc9f0a510 100644 --- a/key-wallet-manager/src/accessors.rs +++ b/key-wallet-manager/src/accessors.rs @@ -231,6 +231,21 @@ impl WalletManager { // A consumer has already been installed; the receiver is taken once. return None; } + // Installing after emission has begun violates the documented contract: + // everything emitted so far reached only the lossy broadcast and is + // permanently absent from the persistence stream, with no + // `Lagged`-style marker to reveal the gap. The receiver is still + // returned (delivery is lossless from this point on), but the gap must + // not be silent — it is exactly the invisible-loss failure mode this + // channel exists to eliminate. + if self.events_emitted.load(std::sync::atomic::Ordering::Relaxed) { + tracing::warn!( + "take_persistence_receiver called after wallet events were already emitted; \ + earlier events are absent from the persistence stream. Install the persistence \ + consumer before the manager is shared with any producer \ + (dashpay/platform#4069)." + ); + } let (sender, receiver) = mpsc::unbounded_channel(); self.persistence_sender = Some(sender); Some(receiver) diff --git a/key-wallet-manager/src/event_tests.rs b/key-wallet-manager/src/event_tests.rs index e6ea061e7..3f8eb4052 100644 --- a/key-wallet-manager/src/event_tests.rs +++ b/key-wallet-manager/src/event_tests.rs @@ -1421,3 +1421,87 @@ async fn persistence_channel_is_lossless_under_a_large_burst() { "broadcast should have dropped events but delivered all {BURST}" ); } + +/// The receiver is taken exactly once: the second call returns `None` and the +/// first receiver keeps working. Platform's manager construction relies on +/// exactly this (`take_persistence_receiver().expect(..)` on a fresh manager). +#[tokio::test] +async fn persistence_receiver_is_taken_exactly_once() { + let (mut manager, wallet_id, _addr) = setup_manager_with_wallet(); + let mut first = manager.take_persistence_receiver().expect("first take must succeed"); + assert!(manager.take_persistence_receiver().is_none(), "second take must return None"); + manager.update_wallet_synced_height(&wallet_id, 1); + assert!( + matches!( + first.try_recv(), + Ok(WalletEvent::SyncHeightAdvanced { + height: 1, + .. + }) + ), + "the first receiver must keep receiving after a second take attempt" + ); +} + +/// Late install (contract violation): events emitted before +/// `take_persistence_receiver` are permanently absent from the persistence +/// stream — the channel is created empty and only delivers from installation +/// onward. The accessor warns (see its docs); this pins the behavioural half: +/// no silent replay is invented. +#[tokio::test] +async fn late_install_delivers_only_post_install_events() { + let (mut manager, wallet_id, _addr) = setup_manager_with_wallet(); + // Emitted BEFORE any consumer exists: reaches only the lossy broadcast. + manager.update_wallet_synced_height(&wallet_id, 7); + + let mut rx = + manager.take_persistence_receiver().expect("late install still returns the receiver"); + manager.update_wallet_synced_height(&wallet_id, 8); + + match rx.try_recv() { + Ok(WalletEvent::SyncHeightAdvanced { + height, + .. + }) => { + assert_eq!(height, 8, "only post-install events may be delivered"); + } + other => panic!("expected the post-install watermark, got {other:?}"), + } + assert!( + rx.try_recv().is_err(), + "the pre-install event must not be replayed into the persistence stream" + ); +} + +/// A consumer that drops its receiver while the manager is running must not +/// wedge or panic the emit paths: in-memory processing continues and the +/// broadcast fan-out still delivers (the lost-consumer anomaly is logged once +/// inside `emit_event`). +#[tokio::test] +async fn dropped_persistence_consumer_does_not_wedge_emission() { + let (mut manager, wallet_id, _addr) = setup_manager_with_wallet(); + let rx = manager.take_persistence_receiver().expect("receiver available once"); + drop(rx); + + let mut broadcast_rx = manager.subscribe_events(); + // Two emissions: the first trips the log-once latch, the second proves + // emission keeps flowing afterwards. + manager.update_wallet_synced_height(&wallet_id, 21); + manager.update_wallet_synced_height(&wallet_id, 22); + + let mut heights = Vec::new(); + while let Ok(event) = broadcast_rx.try_recv() { + if let WalletEvent::SyncHeightAdvanced { + height, + .. + } = event + { + heights.push(height); + } + } + assert_eq!( + heights, + vec![21, 22], + "broadcast delivery must be unaffected by a lost persistence consumer" + ); +} diff --git a/key-wallet-manager/src/lib.rs b/key-wallet-manager/src/lib.rs index 4f9f09d96..3ef0a9398 100644 --- a/key-wallet-manager/src/lib.rs +++ b/key-wallet-manager/src/lib.rs @@ -177,6 +177,15 @@ pub struct WalletManager WalletManager { @@ -192,6 +201,7 @@ impl WalletManager { // a consumer calls `take_persistence_receiver`. See that field's docs. persistence_sender: None, persistence_consumer_lost: std::sync::atomic::AtomicBool::new(false), + events_emitted: std::sync::atomic::AtomicBool::new(false), } } @@ -206,6 +216,7 @@ impl WalletManager { /// persistence consumer (which needs a `read()` lock to drain). See the /// `persistence_sender` field docs for the full rationale. fn emit_event(&self, event: WalletEvent) { + self.events_emitted.store(true, std::sync::atomic::Ordering::Relaxed); // Lossless path: the persistence consumer must never miss a row-bearing // or watermark event, or its durable sync height freezes. Only enqueue // once a consumer has installed itself (opt-in), so a manager with no From c51fa834b05718fb0edbf5600ee102df4c792869 Mon Sep 17 00:00:00 2001 From: Quantum Explorer Date: Thu, 6 Aug 2026 17:02:49 +0700 Subject: [PATCH 5/5] refactor(key-wallet-manager): remove the unused public event_sender accessor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit event_sender() exposed the raw broadcast::Sender, letting downstream code publish a WalletEvent to the lossy broadcast while bypassing the lossless persistence channel — the exact split-brain emit_event exists to prevent. It has zero callers anywhere (this workspace including dash-spv, and the dashpay/platform consumer); subscription goes through subscribe_events(). Removing it makes emit_event the only emission surface structurally, not just by convention. Co-Authored-By: Claude Fable 5 --- key-wallet-manager/src/accessors.rs | 5 ----- 1 file changed, 5 deletions(-) diff --git a/key-wallet-manager/src/accessors.rs b/key-wallet-manager/src/accessors.rs index cc9f0a510..edee935b5 100644 --- a/key-wallet-manager/src/accessors.rs +++ b/key-wallet-manager/src/accessors.rs @@ -204,11 +204,6 @@ impl WalletManager { self.event_sender.subscribe() } - /// Get a reference to the event sender for emitting events. - pub fn event_sender(&self) -> &broadcast::Sender { - &self.event_sender - } - /// Install a durable persistence consumer and take its event receiver /// (dashpay/platform#4069). ///