monitor+: Remove api_call_fut hack

- Adds `LexeInnerPersister::persist_monitor`
- Uses this method instead of `api_call_fut`
- Misc other improvements

Author: MaxFangX

common/src/constants.rs

@@ -76,8 +76,10 @@ pub const IMPORTANT_PERSIST_RETRIES: usize = 5;
 
 /// The vfs directory name used by singleton objects.
 pub const SINGLETON_DIRECTORY: &str = ".";
-/// The vfs directory name used to persist events.
+
 pub const EVENTS_DIR: &str = "events";
+pub const CHANNEL_MONITORS_DIR: &str = "channel_monitors";
+pub const CHANNEL_MONITORS_ARCHIVE_DIR: &str = "channel_monitors_archive";
 
 pub const CHANNEL_MANAGER_FILENAME: &str = "channel_manager";
 pub const PW_ENC_ROOT_SEED_FILENAME: &str = "password_encrypted_root_seed";

lexe-ln/src/channel_monitor.rs

@@ -10,23 +10,13 @@ use tokio::sync::mpsc;
 use tracing::{debug, error, info, info_span, Instrument};
 
 use crate::{
-    alias::LexeChainMonitorType, traits::LexePersister, BoxedAnyhowFuture,
+    alias::LexeChainMonitorType,
+    traits::{LexeInnerPersister, LexePersister},
 };
 
-// `api_call_fut` is a future which makes an api call (typically with
-// retries) to the backend to persist the channel monitor state, returning
-// an `anyhow::Result<()>` once either (1) persistence succeeds or (2)
-// there were too many failures to keep trying. We take this future as
-// input (instead of e.g. a `VfsFile`) because it is the cleanest and
-// easiest way to abstract over the user node and LSP's differing api
-// clients, vfs structures, and expected error types.
-//
-// TODO(max): Add a required `upsert_monitor` method to the `LexePersister`
-// trait to avoid this.
-pub type MonitorChannelItem = (LxChannelMonitorUpdate, BoxedAnyhowFuture);
-
 /// Represents a channel monitor update. See docs on each field for details.
 pub struct LxChannelMonitorUpdate {
+    #[allow(dead_code)] // Conceptually part of the update.
     kind: ChannelMonitorUpdateKind,
     funding_txo: LxOutPoint,
     /// The ID of the channel monitor update, given by
@@ -85,27 +75,24 @@ impl Display for ChannelMonitorUpdateKind {
 /// This prevent a race conditions where two monitor updates come in quick
 /// succession and the newer channel monitor state is overwritten by the older
 /// channel monitor state.
-pub fn spawn_channel_monitor_persister_task<PS>(
+pub fn spawn_channel_monitor_persister_task<PS: LexePersister>(
+    persister: PS,
     chain_monitor: Arc<LexeChainMonitorType<PS>>,
-    mut channel_monitor_persister_rx: mpsc::Receiver<MonitorChannelItem>,
+    mut channel_monitor_persister_rx: mpsc::Receiver<LxChannelMonitorUpdate>,
     mut shutdown: NotifyOnce,
-) -> LxTask<()>
-where
-    PS: LexePersister,
-{
+) -> LxTask<()> {
     debug!("Starting channel monitor persister task");
     const SPAN_NAME: &str = "(chan-monitor-persister)";
     LxTask::spawn_with_span(SPAN_NAME, info_span!(SPAN_NAME), async move {
         loop {
             tokio::select! {
-                Some((update, api_call_fut))
-                    = channel_monitor_persister_rx.recv() => {
+                Some(update) = channel_monitor_persister_rx.recv() => {
                     let update_span = update.span();
 
                     let handle_result = handle_update(
+                        &persister,
                         chain_monitor.as_ref(),
                         update,
-                        api_call_fut,
                     )
                         .instrument(update_span.clone())
                         .await;
@@ -137,23 +124,18 @@ where
 /// Since channel monitor persistence is very important, all [`Err`]s are
 /// considered fatal; the caller should send a shutdown signal and exit.
 async fn handle_update<PS: LexePersister>(
+    persister: &PS,
     chain_monitor: &LexeChainMonitorType<PS>,
     update: LxChannelMonitorUpdate,
-    api_call_fut: BoxedAnyhowFuture,
 ) -> anyhow::Result<()> {
-    let LxChannelMonitorUpdate {
-        funding_txo,
-        update_id,
-        kind: _,
-        span: _,
-    } = update;
-
     debug!("Handling channel monitor update");
 
-    // Run the persist future.
-    api_call_fut
+    // Persist the monitor.
+    let funding_txo = OutPoint::from(update.funding_txo);
+    persister
+        .persist_monitor(chain_monitor, &update.funding_txo)
         .await
-        .context("Channel monitor persist API call failed")?;
+        .context("persist_monitor failed")?;
 
     // Notify the chain monitor that the monitor update has been persisted.
     // - This should trigger a log like "Completed off-chain monitor update ..."
@@ -164,7 +146,7 @@ async fn handle_update<PS: LexePersister>(
     //   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)
+        .channel_monitor_updated(funding_txo, update.update_id)
         .map_err(|e| anyhow!("channel_monitor_updated returned Err: {e:?}"))?;
 
     info!("Success: persisted monitor");

lexe-ln/src/traits.rs

@@ -1,14 +1,17 @@
 use std::{future::Future, ops::Deref, str::FromStr};
 
-use anyhow::Context;
+use anyhow::{anyhow, Context};
 use async_trait::async_trait;
 use common::{
     api::{
         user::NodePk,
         vfs::{Vfs, VfsDirectory, VfsFileId},
     },
     constants,
-    ln::payments::{LxPaymentId, PaymentIndex},
+    ln::{
+        channel::LxOutPoint,
+        payments::{LxPaymentId, PaymentIndex},
+    },
     notify_once::NotifyOnce,
 };
 use lightning::{
@@ -75,6 +78,31 @@ pub trait LexeInnerPersister: Vfs + Persist<SignerType> {
             .await
     }
 
+    async fn persist_monitor<PS: LexePersister>(
+        &self,
+        chain_monitor: &LexeChainMonitorType<PS>,
+        funding_txo: &LxOutPoint,
+    ) -> anyhow::Result<()> {
+        let file = {
+            let locked_monitor =
+                chain_monitor.get_monitor((*funding_txo).into()).map_err(
+                    |e| anyhow!("No monitor for this funding_txo: {e:?}"),
+                )?;
+
+            // NOTE: The VFS filename uses the `ToString` impl of `LxOutPoint`
+            // rather than `lightning::chain::transaction::OutPoint` or
+            // `bitcoin::OutPoint`! `LxOutPoint`'s FromStr/Display impls are
+            // guaranteed to roundtrip, and will be stable across LDK versions.
+            let filename = funding_txo.to_string();
+            let file_id =
+                VfsFileId::new(constants::CHANNEL_MONITORS_DIR, filename);
+            self.encrypt_ldk_writeable(file_id, &*locked_monitor)
+        };
+
+        self.persist_file(&file, constants::IMPORTANT_PERSIST_RETRIES)
+            .await
+    }
+
     /// Reads all persisted events, along with their event IDs.
     async fn read_events(&self) -> anyhow::Result<Vec<(EventId, Event)>> {
         let dir = VfsDirectory::new(constants::EVENTS_DIR);