fix(key-wallet-manager): lossless persistence event channel to stop the sync-watermark freeze

dash-spv/src/sync/filters/batch.rs

@@ -1,7 +1,7 @@
 use dashcore::bip158::BlockFilter;
 use dashcore::ScriptBuf;
 use key_wallet_manager::{FilterMatchKey, WalletId};
-use std::collections::{BTreeSet, HashMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet};
 
 /// A completed batch of compact block filters ready for verification.
 ///
@@ -24,10 +24,15 @@ pub(super) struct FiltersBatch {
     pending_blocks: u32,
     /// Whether rescan has been completed for this batch.
     rescan_complete: bool,
-    /// Wallets that were behind for this batch's height range at scan time and
-    /// therefore need their `synced_height` advanced when the batch commits.
-    /// Already-synced wallets must not be touched.
-    scanned_wallets: BTreeSet<WalletId>,
+    /// Wallets that were behind for this batch's height range at scan time —
+    /// and therefore need their `synced_height` advanced when the batch
+    /// commits — each mapped to the wallet's `account_generation` at scan
+    /// time. Commit refuses to advance a wallet whose generation has changed
+    /// since the scan: an account added mid-flight means the scan did not test
+    /// the new account's scripts, so the batch cannot certify coverage for the
+    /// current account set (dashpay/rust-dashcore#649). Already-synced wallets
+    /// must not be touched.
+    scanned_wallets: BTreeMap<WalletId, u64>,
     /// Cached scriptPubKeys discovered during block processing that still
     /// need rescan, attributed per wallet so we can rerun matching only
     /// against the wallet that produced each new script.
@@ -49,7 +54,7 @@ impl FiltersBatch {
             scanned: false,
             pending_blocks: 0,
             rescan_complete: false,
-            scanned_wallets: BTreeSet::new(),
+            scanned_wallets: BTreeMap::new(),
             collected_scripts: HashMap::new(),
         }
     }
@@ -118,13 +123,14 @@ impl FiltersBatch {
     pub(super) fn take_collected_scripts(&mut self) -> HashMap<WalletId, HashSet<ScriptBuf>> {
         std::mem::take(&mut self.collected_scripts)
     }
-    /// Record the set of wallets that were behind for this batch at scan time.
-    pub(super) fn set_scanned_wallets(&mut self, wallets: BTreeSet<WalletId>) {
+    /// Record the wallets that were behind for this batch at scan time, each
+    /// with its `account_generation` snapshot.
+    pub(super) fn set_scanned_wallets(&mut self, wallets: BTreeMap<WalletId, u64>) {
         self.scanned_wallets = wallets;
     }
-    /// Wallets that were behind at scan time and must have their synced_height
-    /// advanced when this batch commits.
-    pub(super) fn scanned_wallets(&self) -> &BTreeSet<WalletId> {
+    /// Wallets that were behind at scan time (with their generation snapshot)
+    /// and must have their synced_height advanced when this batch commits.
+    pub(super) fn scanned_wallets(&self) -> &BTreeMap<WalletId, u64> {
         &self.scanned_wallets
     }
 }

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<T: WalletInfoInterface + Send + Sync + 'static> WalletManager<T> {
     /// Get a wallet by ID
@@ -209,6 +209,33 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> WalletManager<T> {
         &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<mpsc::UnboundedReceiver<WalletEvent>> {
+        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

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<u32> = 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}"
+    );
+}

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<T: WalletInfoInterface + Send + Sync + 'static = Manage
     /// added/removed. Combined with per-wallet account-level revisions to
     /// produce the total monitor revision.
     structural_revision: u64,
-    /// Event sender for wallet events
+    /// Event sender for wallet events.
+    ///
+    /// This is a *bounded, lossy* fan-out (capacity
+    /// [`DEFAULT_WALLET_EVENT_CAPACITY`]): under a heavy SPV catch-up a
+    /// subscriber that drains slower than blocks are processed overflows the
+    /// ring and receives `RecvError::Lagged`, silently dropping every
+    /// record/UTXO/watermark event in between. That is acceptable for the
+    /// *incidental* subscribers reached via [`subscribe_events`] (dash-spv's
+    /// `EventHandler` fan-out, unit tests) but NOT for the durable-persistence
+    /// consumer — see `persistence_sender`.
+    ///
+    /// [`subscribe_events`]: WalletManager::subscribe_events
     event_sender: broadcast::Sender<WalletEvent>,
+    /// 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<mpsc::UnboundedSender<WalletEvent>>,
+    /// 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<T: WalletInfoInterface + Send + Sync + 'static> WalletManager<T> {
@@ -140,7 +188,53 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> WalletManager<T> {
             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.

key-wallet-manager/src/process_block.rs

@@ -170,7 +170,7 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> 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<T: WalletInfoInterface + Send + Sync + 'static> WalletInterface for WalletM
                         balance,
                         account_balances: account_balances.clone(),
                     };
-                    let _ = self.event_sender.send(event);
+                    self.emit_event(event);
                 }
             }
         }
@@ -274,11 +274,15 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> WalletInterface for WalletM
         self.wallet_infos.get(wallet_id).map(|info| info.synced_height()).unwrap_or(0)
     }
 
+    fn wallet_account_generation(&self, wallet_id: &WalletId) -> u64 {
+        self.wallet_infos.get(wallet_id).map(|info| info.account_generation()).unwrap_or(0)
+    }
+
     fn update_wallet_synced_height(&mut self, wallet_id: &WalletId, height: CoreBlockHeight) {
         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,
                 });
@@ -310,6 +314,12 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> 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());
 
@@ -319,13 +329,16 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> 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) {
@@ -357,7 +370,7 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> 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(),
@@ -497,7 +510,7 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> WalletManager<T> {
                     account_balances,
                     addresses_derived,
                 };
-                let _ = self.event_sender.send(event);
+                self.emit_event(event);
             }
         }
     }