bgp+monitor: Remove process_events_[tx/rx]

MaxFangX · MaxFangX · commit b670bd3df940 · 2025-03-13T13:46:42.000-07:00
It looks like LDK #2090 *did* resolve our issue - but it notifies the BGP only after *all* monitor updates are handled. See lightningdevkit/rust-lightning#3660 (comment) for more info. Also took the opportunity to: - Instrument the channel monitor persister task - Make the monitor logs easier to grep by using a consistent message - Log how long it takes to process events - Remove the unneeded `idx` field in `lexe_ln::channel_monitor` - Remove the unneeded `Error` type in `lexe_ln::channel_monitor`
diff --git a/lexe-ln/src/background_processor.rs b/lexe-ln/src/background_processor.rs
@@ -1,11 +1,8 @@
 use std::{sync::Arc, time::Duration};
 
-use common::{notify_once::NotifyOnce, task::LxTask};
-use tokio::{
-    sync::{mpsc, oneshot},
-    time::Instant,
-};
-use tracing::{debug, error, info_span, Instrument};
+use common::{notify_once::NotifyOnce, task::LxTask, time::DisplayMs};
+use tokio::time::Instant;
+use tracing::{debug, error, info, info_span, warn, Instrument};
 
 use crate::{
     alias::{LexeChainMonitorType, LexeOnionMessengerType},
@@ -49,16 +46,6 @@ pub fn start<CM, PM, PS, EH>(
     chain_monitor: Arc<LexeChainMonitorType<PS>>,
     onion_messenger: Arc<LexeOnionMessengerType<CM>>,
     event_handler: EH,
-    // TODO(max): A `process_events` notification should be sent every time
-    // an event is generated which does not also cause the future returned
-    // by `get_event_or_persistence_needed_future()` to resolve.
-    //
-    // Ideally, we can remove this channel entirely, but a manual trigger
-    // is currently still required after every channel monitor
-    // persist (which may resume monitor updating and create more
-    // events). This was supposed to be resolved by LDK#2052 and
-    // LDK#2090, but our integration tests still fail without this channel.
-    mut process_events_rx: mpsc::Receiver<oneshot::Sender<()>>,
     mut shutdown: NotifyOnce,
 ) -> LxTask<()>
 where
@@ -71,17 +58,17 @@ where
         "background processor",
         info_span!("(bgp)"),
         async move {
-            let now = Instant::now();
+            let bgp_start = Instant::now();
 
             let mk_interval = |delay: Duration, interval: Duration| {
                 // Remove the staggering in debug mode in an attempt to catch
                 // any subtle race conditions which may arise
-                let start = if cfg!(debug_assertions) {
-                    now
+                let timer_start = if cfg!(debug_assertions) {
+                    bgp_start
                 } else {
-                    now + delay
+                    bgp_start + delay
                 };
-                tokio::time::interval_at(start, interval)
+                tokio::time::interval_at(timer_start, interval)
             };
 
             let mut process_events_timer =
@@ -100,10 +87,9 @@ where
                 //
                 // - Our process events timer ticked
                 // - The channel manager got an update (event or repersist)
-                // - The chain monitor got an update
+                // - The chain monitor got an update (typically that all updates
+                //   were persisted for a channel monitor)
                 // - The onion messenger got an update
-                // - The background processor was explicitly triggered
-                let mut processed_txs = Vec::new();
                 let process_events_fut = async {
                     tokio::select! {
                         biased;
@@ -116,27 +102,18 @@ where
                             debug!("Triggered: Chain monitor update"),
                         _ = onion_messenger.get_update_future() =>
                             debug!("Triggered: Onion messenger update"),
-                        // TODO(max): If LDK has fixed the BGP waking issue,
-                        // our integration tests should pass with this branch
-                        // commented out.
-                        Some(tx) = process_events_rx.recv() => {
-                            debug!("Triggered: process_events channel");
-                            processed_txs.push(tx);
-                        }
                     };
 
                     // We're about to process events. Prevent duplicate work by
                     // resetting the process_events_timer & clearing out the
                     // process_events channel.
                     process_events_timer.reset();
-                    while let Ok(tx) = process_events_rx.try_recv() {
-                        processed_txs.push(tx);
-                    }
                 };
 
                 tokio::select! {
                     () = process_events_fut => {
                         debug!("Processing pending events");
+                        let process_start = Instant::now();
 
                         channel_manager
                             .process_pending_events_async(mk_event_handler_fut)
@@ -159,11 +136,6 @@ where
                             peer_manager.process_events();
                         }.instrument(info_span!("(process-bgp)(peer-man)")).await;
 
-                        // Notify waiters that events have been processed.
-                        for tx in processed_txs {
-                            let _ = tx.send(());
-                        }
-
                         if channel_manager.get_and_clear_needs_persistence() {
                             let try_persist = persister
                                 .persist_manager(channel_manager.deref())
@@ -177,6 +149,16 @@ where
                                 break shutdown.send();
                             }
                         }
+
+                        let elapsed = process_start.elapsed();
+                        let elapsed_ms = DisplayMs(elapsed);
+                        if elapsed > Duration::from_secs(10) {
+                            warn!("Event processing took {elapsed_ms}");
+                        } else if elapsed > Duration::from_secs(1) {
+                            info!("Event processing took {elapsed_ms}");
+                        } else {
+                            debug!("Event processing took {elapsed_ms}");
+                        }
                     }
 
                     _ = pm_timer.tick() =>
diff --git a/lexe-ln/src/channel_monitor.rs b/lexe-ln/src/channel_monitor.rs
@@ -1,24 +1,18 @@
 use std::{
     fmt::{self, Display},
     sync::Arc,
-    time::Duration,
 };
 
+use anyhow::{anyhow, Context};
 use common::{ln::channel::LxOutPoint, notify_once::NotifyOnce, task::LxTask};
 use lightning::chain::transaction::OutPoint;
-use thiserror::Error;
-use tokio::sync::{mpsc, oneshot};
-use tracing::{debug, error, info};
+use tokio::sync::mpsc;
+use tracing::{debug, error, info, info_span};
 
 use crate::{
     alias::LexeChainMonitorType, traits::LexePersister, BoxedAnyhowFuture,
 };
 
-/// How long we'll wait to receive a reply from the background processor that
-/// event processing is complete.
-// 45s because we've seen this timeout on the node.
-const PROCESS_EVENTS_TIMEOUT: Duration = Duration::from_secs(45);
-
 /// Represents a channel monitor update. See docs on each field for details.
 pub struct LxChannelMonitorUpdate {
     pub funding_txo: LxOutPoint,
@@ -63,31 +57,29 @@ impl Display for ChannelMonitorUpdateKind {
 pub fn spawn_channel_monitor_persister_task<PS>(
     chain_monitor: Arc<LexeChainMonitorType<PS>>,
     mut channel_monitor_persister_rx: mpsc::Receiver<LxChannelMonitorUpdate>,
-    process_events_tx: mpsc::Sender<oneshot::Sender<()>>,
     mut shutdown: NotifyOnce,
 ) -> LxTask<()>
 where
     PS: LexePersister,
 {
     debug!("Starting channel monitor persister task");
-    LxTask::spawn("channel monitor persister", async move {
-        let mut idx = 0;
+    const SPAN_NAME: &str = "(chan-monitor-persister)";
+    LxTask::spawn_with_span(SPAN_NAME, info_span!(SPAN_NAME), async move {
         loop {
             tokio::select! {
                 Some(update) = channel_monitor_persister_rx.recv() => {
-                    idx += 1;
 
                     let handle_result = handle_update(
                         chain_monitor.as_ref(),
                         update,
-                        idx,
-                        &process_events_tx,
                     ).await;
 
                     if let Err(e) = handle_result {
                         error!("Monitor persist error: {e:#}");
 
-                        // All errors are considered fatal.
+                        // Channel monitor persistence errors are serious;
+                        // all errors are considered fatal.
+                        // Shut down to prevent any loss of funds.
                         shutdown.send();
                         break;
                     }
@@ -101,24 +93,6 @@ where
     })
 }
 
-/// Errors that can occur when handling a channel monitor update.
-///
-/// This enum is intentionally kept private; it exists solely to prevent the
-/// caller from having to use some variant of `err_str.contains(..)`
-#[derive(Debug, Error)]
-enum Error {
-    #[error("Couldn't persist {kind} channel #{idx}: {inner:#}")]
-    PersistFailure {
-        kind: ChannelMonitorUpdateKind,
-        idx: usize,
-        inner: anyhow::Error,
-    },
-    #[error("Chain monitor returned err: {0:?}")]
-    ChainMonitor(lightning::util::errors::APIError),
-    #[error("Timed out waiting for events to be processed")]
-    EventsProcessTimeout,
-}
-
 /// A helper to prevent [`spawn_channel_monitor_persister_task`]'s control flow
 /// from getting too complex.
 ///
@@ -127,46 +101,36 @@ enum Error {
 async fn handle_update<PS: LexePersister>(
     chain_monitor: &LexeChainMonitorType<PS>,
     update: LxChannelMonitorUpdate,
-    idx: usize,
-    process_events_tx: &mpsc::Sender<oneshot::Sender<()>>,
-) -> Result<(), Error> {
-    debug!("Handling channel monitor update #{idx}");
+) -> anyhow::Result<()> {
+    let LxChannelMonitorUpdate {
+        funding_txo,
+        update_id,
+        api_call_fut,
+        kind,
+    } = update;
 
-    // Run the persist future.
-    let kind = update.kind;
-    if let Err(inner) = update.api_call_fut.await {
-        // Channel monitor persistence errors are serious;
-        // return err and shut down to prevent any loss of funds.
-        return Err(Error::PersistFailure { kind, idx, inner });
-    }
+    debug!(%kind, %funding_txo, %update_id, "Handling channel monitor update");
 
-    // Update the chain monitor with the update id and funding txo the channel
-    // monitor update.
-    let chain_monitor_update_res = chain_monitor.channel_monitor_updated(
-        OutPoint::from(update.funding_txo),
-        update.update_id,
-    );
-    if let Err(e) = chain_monitor_update_res {
-        // If the update wasn't accepted, the channel is disabled, so no
-        // transactions can be made. Just return err and shut down.
-        return Err(Error::ChainMonitor(e));
-    }
-
-    // Trigger the background processor to reprocess events, as the completed
-    // channel monitor update may have generated an event that can be handled,
-    // such as to restore monitor updating and broadcast a funding tx.
-    // Furthermore, wait for the event to be handled.
-    debug!("Triggering BGP via process_events_tx");
-    let (processed_tx, processed_rx) = oneshot::channel();
-    let _ = process_events_tx.try_send(processed_tx);
-
-    tokio::time::timeout(PROCESS_EVENTS_TIMEOUT, processed_rx)
+    // Run the persist future.
+    api_call_fut
         .await
-        .map_err(|_| Error::EventsProcessTimeout)?
-        // Channel sender dropped, probably means we're shutting down.
-        .ok();
-
-    info!("Success: persisted {kind} channel #{idx}");
+        .with_context(|| format!("{kind} {funding_txo} {update_id}"))
+        .context("Channel monitor persist API call failed")?;
+
+    // Notify the chain monitor that the monitor update has been persisted.
+    // - This should trigger a log like "Completed off-chain monitor update ..."
+    // - NOTE: After this update, there may still be more updates to persist.
+    //   The LDK log message will say "all off-chain updates complete" or "still
+    //   have pending off-chain updates" (common during payments)
+    // - NOTE: Only after *all* channel monitor updates are handled will the
+    //   channel be reenabled and the BGP woken to process events via the chain
+    //   monitor future.
+    chain_monitor
+        .channel_monitor_updated(OutPoint::from(funding_txo), update_id)
+        .map_err(|e| anyhow!("{kind} {funding_txo} {update_id}: {e:?}"))
+        .context("channel_monitor_updated returned Err")?;
+
+    info!(%kind, %funding_txo, %update_id, "Success: persisted monitor");
 
     Ok(())
 }
diff --git a/node/src/persister/mod.rs b/node/src/persister/mod.rs
@@ -780,7 +780,8 @@ impl Persist<SignerType> for NodePersister {
     ) -> ChannelMonitorUpdateStatus {
         let funding_txo = LxOutPoint::from(funding_txo);
         let update_id = monitor.get_latest_update_id();
-        info!(%funding_txo, %update_id, "Persisting new channel");
+        let kind = ChannelMonitorUpdateKind::New;
+        info!(%kind, %funding_txo, %update_id, "Persisting channel monitor");
 
         let file_id =
             VfsFileId::new(CHANNEL_MONITORS_DIR, funding_txo.to_string());
@@ -809,8 +810,6 @@ impl Persist<SignerType> for NodePersister {
             }
         });
 
-        let kind = ChannelMonitorUpdateKind::New;
-
         let update = LxChannelMonitorUpdate {
             funding_txo,
             update_id,
@@ -847,7 +846,8 @@ impl Persist<SignerType> for NodePersister {
             .as_ref()
             .map(|u| u.update_id)
             .unwrap_or_else(|| monitor.get_latest_update_id());
-        info!(%funding_txo, %update_id, "Updating persisted channel");
+        let kind = ChannelMonitorUpdateKind::Updated;
+        info!(%kind, %funding_txo, %update_id, "Persisting channel monitor");
 
         let file_id =
             VfsFileId::new(CHANNEL_MONITORS_DIR, funding_txo.to_string());
@@ -871,8 +871,6 @@ impl Persist<SignerType> for NodePersister {
             }
         });
 
-        let kind = ChannelMonitorUpdateKind::Updated;
-
         let update = LxChannelMonitorUpdate {
             funding_txo,
             update_id,
@@ -898,6 +896,8 @@ impl Persist<SignerType> for NodePersister {
     }
 
     fn archive_persisted_channel(&self, funding_txo: OutPoint) {
+        info!(%funding_txo, "Archiving channel monitor");
+
         let backend_api = self.backend_api.clone();
         let authenticator = self.authenticator.clone();
         let vfs_master_key = self.vfs_master_key.clone();
diff --git a/node/src/run.rs b/node/src/run.rs
@@ -618,12 +618,9 @@ impl UserNode {
             .push(event::spawn_event_replayer_task(event_handler.clone()));
 
         // Set up the channel monitor persistence task
-        let (process_events_tx, process_events_rx) =
-            mpsc::channel(DEFAULT_CHANNEL_SIZE);
         let task = channel_monitor::spawn_channel_monitor_persister_task(
             chain_monitor.clone(),
             channel_monitor_persister_rx,
-            process_events_tx,
             shutdown.clone(),
         );
         static_tasks.push(task);
@@ -756,7 +753,6 @@ impl UserNode {
             chain_monitor.clone(),
             onion_messenger.clone(),
             event_handler,
-            process_events_rx,
             shutdown.clone(),
         );
         static_tasks.push(bg_processor_task);
