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

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
@@ -204,9 +204,46 @@ impl<T: WalletInfoInterface + Send + Sync + 'static> WalletManager<T> {
         self.event_sender.subscribe()
     }
 
-    /// Get a reference to the event sender for emitting events.
-    pub fn event_sender(&self) -> &broadcast::Sender<WalletEvent> {
-        &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;
+        }
+        // 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)
     }
 
     /// Return the total monitor revision (structural + per-wallet account revisions).

key-wallet-manager/src/event_tests.rs

@@ -1339,3 +1339,169 @@ 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}"
+    );
+}
+
+/// 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"
+    );
+}

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,65 @@ 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,
+    /// Latches `true` on the first [`emit_event`] call. Read by
+    /// [`take_persistence_receiver`] to warn when a persistence consumer
+    /// installs itself only AFTER events have already been emitted — those
+    /// events reached only the lossy broadcast and are permanently absent from
+    /// the persistence stream, with no `Lagged`-style marker to reveal the gap.
+    ///
+    /// [`emit_event`]: WalletManager::emit_event
+    /// [`take_persistence_receiver`]: WalletManager::take_persistence_receiver
+    events_emitted: std::sync::atomic::AtomicBool,
 }
 
 impl<T: WalletInfoInterface + Send + Sync + 'static> WalletManager<T> {
@@ -140,7 +197,55 @@ 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),
+            events_emitted: 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) {
+        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
+        // 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.