diff --git a/key-wallet-manager/src/accessors.rs b/key-wallet-manager/src/accessors.rs index db66cb763..970de1f3c 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,33 @@ impl WalletManager { &self.event_sender } + /// Install a durable persistence consumer and take its event receiver + /// (dashpay/platform#4069). + /// + /// 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> { + 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). 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..4f9f09d96 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,8 +127,56 @@ 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`]. + /// + /// 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 + /// [`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 { @@ -140,7 +188,53 @@ impl WalletManager { wallet_infos: BTreeMap::new(), structural_revision: 0, event_sender: broadcast::Sender::new(DEFAULT_WALLET_EVENT_CAPACITY), + // 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), + } + } + + /// 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. 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. + let _ = self.event_sender.send(event); } /// Increment the structural revision for wallet/account additions or removals. 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); } } }