node: Proper async channel monitor persist

MaxFangX · MaxFangX · commit 9412c4215095 · 2022-08-16T18:36:16.000-07:00
diff --git a/node/src/init.rs b/node/src/init.rs
@@ -19,20 +19,24 @@ use common::rng::Crng;
 use lightning::chain;
 use lightning::chain::chainmonitor::ChainMonitor;
 use lightning::chain::keysinterface::KeysInterface;
+use lightning::chain::transaction::OutPoint;
 use lightning::routing::gossip::P2PGossipSync;
 use lightning_invoice::payment;
 use lightning_invoice::utils::DefaultRouter;
 use tokio::net::TcpListener;
 use tokio::runtime::Handle;
 use tokio::sync::{broadcast, mpsc};
 use tracing::{debug, error, info, instrument};
+use tokio::task::JoinHandle;
 
 use crate::api::ApiClient;
 use crate::event_handler::LdkEventHandler;
 use crate::inactivity_timer::InactivityTimer;
 use crate::lexe::background_processor::LexeBackgroundProcessor;
 use crate::lexe::bitcoind::LexeBitcoind;
-use crate::lexe::channel_manager::LexeChannelManager;
+use crate::lexe::channel_manager::{
+    LexeChannelManager, LxChannelMonitorUpdate,
+};
 use crate::lexe::keys_manager::LexeKeysManager;
 use crate::lexe::logger::LexeTracingLogger;
 use crate::lexe::peer_manager::LexePeerManager;
@@ -132,8 +136,20 @@ impl LexeNode {
         let fee_estimator = bitcoind.clone();
         let broadcaster = bitcoind.clone();
 
+        // Init Tokio channels
+        let (activity_tx, activity_rx) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
+        let (shutdown_tx, shutdown_rx) =
+            broadcast::channel(DEFAULT_CHANNEL_SIZE);
+        let (channel_monitor_updated_tx, channel_monitor_updated_rx) =
+            mpsc::channel(DEFAULT_CHANNEL_SIZE);
+
         // Initialize Persister
-        let persister = LexePersister::new(api.clone(), node_pk, measurement);
+        let persister = LexePersister::new(
+            api.clone(),
+            node_pk,
+            measurement,
+            channel_monitor_updated_tx,
+        );
 
         // Initialize the ChainMonitor
         let chain_monitor = Arc::new(ChainMonitor::new(
@@ -144,6 +160,16 @@ impl LexeNode {
             persister.clone(),
         ));
 
+        // Set up the persister -> chain monitor channel
+        // TODO(max): Handle the handle
+        let channel_monitor_updated_shutdown_rx = shutdown_tx.subscribe();
+        let _channel_monitor_updated_handle =
+            spawn_channel_monitor_updated_task(
+                chain_monitor.clone(),
+                channel_monitor_updated_rx,
+                channel_monitor_updated_shutdown_rx,
+            );
+
         // Read the `ChannelMonitor`s and initialize the `P2PGossipSync`
         let (channel_monitors_res, gossip_sync_res) = tokio::join!(
             channel_monitors(&persister, keys_manager.clone()),
@@ -198,12 +224,7 @@ impl LexeNode {
         )
         .await;
 
-        // Init Tokio channels
-        let (activity_tx, activity_rx) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
-        let (shutdown_tx, shutdown_rx) =
-            broadcast::channel(DEFAULT_CHANNEL_SIZE);
-
-        // build owner service TLS config for authenticating owner
+        // Build owner service TLS config for authenticating owner
         let node_dns = args.node_dns_name.clone();
         let owner_tls = node_run_tls_config(rng, root_seed, vec![node_dns])
             .context("Failed to build owner service TLS config")?;
@@ -297,7 +318,8 @@ impl LexeNode {
         );
 
         // Spawn a task to regularly reconnect to channel peers
-        spawn_p2p_reconnect_task(
+        // TODO(max): Handle the handle
+        let _reconnect_handle = spawn_p2p_reconnect_task(
             channel_manager.clone(),
             peer_manager.clone(),
             stop_listen_connect.clone(),
@@ -577,7 +599,7 @@ fn spawn_p2p_reconnect_task(
     peer_manager: LexePeerManager,
     stop_listen_connect: Arc<AtomicBool>,
     persister: LexePersister,
-) {
+) -> JoinHandle<()> {
     tokio::spawn(async move {
         let mut interval = tokio::time::interval(Duration::from_secs(60));
         loop {
@@ -611,5 +633,39 @@ fn spawn_p2p_reconnect_task(
                 }
             }
         }
-    });
+    })
+}
+
+/// Spawns a task that that lets the persister make calls to the chain monitor.
+/// For now, it simply listens on `channel_monitor_updated_rx` and calls
+/// `ChainMonitor::channel_monitor_updated()` with any received values. This is
+/// required because (a) the chain monitor cannot be initialized without the
+/// persister, therefore (b) the persister cannot hold the chain monitor,
+/// therefore there needs to be another means of letting the persister notify
+/// the channel manager of events.
+pub fn spawn_channel_monitor_updated_task(
+    chain_monitor: Arc<ChainMonitorType>,
+    mut channel_monitor_updated_rx: mpsc::Receiver<LxChannelMonitorUpdate>,
+    mut shutdown_rx: broadcast::Receiver<()>,
+) -> JoinHandle<()> {
+    info!("Starting channel_monitor_updated task");
+    tokio::spawn(async move {
+        loop {
+            tokio::select! {
+                Some(update) = channel_monitor_updated_rx.recv() => {
+                    if let Err(e) = chain_monitor.channel_monitor_updated(
+                        OutPoint::from(update.funding_txo),
+                        update.update_id,
+                    ) {
+                        // ApiError impls Debug but not std::error::Error
+                        error!("channel_monitor_updated returned Err: {:?}", e);
+                    }
+                }
+                _ = shutdown_rx.recv() => {
+                    info!("channel_monitor_updated task shutting down");
+                    break;
+                }
+            }
+        }
+    })
 }
diff --git a/node/src/lexe/channel_manager/types.rs b/node/src/lexe/channel_manager/types.rs
@@ -1,9 +1,15 @@
 use bitcoin::secp256k1::PublicKey;
+use lightning::chain::chainmonitor::MonitorUpdateId;
 use lightning::ln::channelmanager::{ChannelCounterparty, ChannelDetails};
 use serde::Serialize;
 
 use crate::lexe::types::LxOutPoint;
 
+pub struct LxChannelMonitorUpdate {
+    pub funding_txo: LxOutPoint,
+    pub update_id: MonitorUpdateId,
+}
+
 #[derive(Serialize)]
 pub struct LxChannelDetails {
     pub channel_id: [u8; 32],
diff --git a/node/src/lexe/persister.rs b/node/src/lexe/persister.rs
@@ -18,10 +18,10 @@ use lightning::routing::scoring::{
     ProbabilisticScorer, ProbabilisticScoringParameters,
 };
 use lightning::util::ser::{ReadableArgs, Writeable};
-use tokio::runtime::Handle;
+use tokio::sync::mpsc;
 use tracing::{debug, error};
 
-use crate::lexe::channel_manager::USER_CONFIG;
+use crate::lexe::channel_manager::{LxChannelMonitorUpdate, USER_CONFIG};
 use crate::lexe::keys_manager::LexeKeysManager;
 use crate::lexe::logger::LexeTracingLogger;
 use crate::lexe::peer_manager::ChannelPeer;
@@ -53,8 +53,14 @@ impl LexePersister {
         api: ApiClientType,
         node_pk: PublicKey,
         measurement: Measurement,
+        channel_monitor_updated_tx: mpsc::Sender<LxChannelMonitorUpdate>,
     ) -> Self {
-        let inner = InnerPersister::new(api, node_pk, measurement);
+        let inner = InnerPersister::new(
+            api,
+            node_pk,
+            measurement,
+            channel_monitor_updated_tx,
+        );
         Self { inner }
     }
 }
@@ -73,18 +79,21 @@ pub struct InnerPersister {
     api: ApiClientType,
     node_pk: PublicKey,
     measurement: Measurement,
+    channel_monitor_updated_tx: mpsc::Sender<LxChannelMonitorUpdate>,
 }
 
 impl InnerPersister {
     fn new(
         api: ApiClientType,
         node_pk: PublicKey,
         measurement: Measurement,
+        channel_monitor_updated_tx: mpsc::Sender<LxChannelMonitorUpdate>,
     ) -> Self {
         Self {
             api,
             node_pk,
             measurement,
+            channel_monitor_updated_tx,
         }
     }
 
@@ -153,6 +162,7 @@ impl InnerPersister {
         keys_manager: LexeKeysManager,
     ) -> anyhow::Result<Vec<(BlockHash, ChannelMonitorType)>> {
         debug!("Reading channel monitors");
+        // TODO Also attempt to read from the cloud
 
         let cm_dir = Directory {
             node_pk: self.node_pk,
@@ -396,11 +406,10 @@ impl Persist<SignerType> for InnerPersister {
         &self,
         funding_txo: OutPoint,
         monitor: &ChannelMonitorType,
-        _update_id: MonitorUpdateId,
+        update_id: MonitorUpdateId,
     ) -> Result<(), ChannelMonitorUpdateErr> {
-        let outpoint = LxOutPoint::from(funding_txo);
-        let outpoint_str = outpoint.to_string();
-        debug!("Persisting new channel {}", outpoint_str);
+        let funding_txo = LxOutPoint::from(funding_txo);
+        debug!("Persisting new channel {}", funding_txo);
 
         // FIXME(encrypt): Encrypt under key derived from seed
         let data = monitor.encode();
@@ -409,31 +418,53 @@ impl Persist<SignerType> for InnerPersister {
             self.node_pk,
             self.measurement,
             CHANNEL_MONITORS_DIRECTORY.to_owned(),
-            outpoint_str,
+            funding_txo.to_string(),
             data,
         );
+        let update = LxChannelMonitorUpdate {
+            funding_txo,
+            update_id,
+        };
 
-        // Run an async fn inside a sync fn inside a Tokio runtime
-        Handle::current()
-            .block_on(async move { self.api.create_file(&cm_file).await })
-            .map(|_| ())
-            .map_err(|e| {
-                // TODO(max): Implement durability then make this err permanent
-                error!("Could not persist new channel monitor: {:#}", e);
-                ChannelMonitorUpdateErr::TemporaryFailure
-            })
+        // Spawn a task for persisting the channel monitor
+        let api_clone = self.api.clone();
+        let channel_monitor_updated_tx =
+            self.channel_monitor_updated_tx.clone();
+        tokio::spawn(async move {
+            // Retry indefinitely until it succeeds
+            loop {
+                // TODO Also attempt to persist to cloud backup
+                match api_clone.create_file(&cm_file).await {
+                    Ok(_file) => {
+                        if let Err(e) =
+                            channel_monitor_updated_tx.try_send(update)
+                        {
+                            error!("Couldn't notify chain monitor: {:#}", e);
+                        }
+                        return;
+                    }
+                    Err(e) => {
+                        error!("Couldn't persist new channel monitor: {:#}", e)
+                    }
+                }
+            }
+        });
+
+        // As documented in the `Persist` trait docs, return `TemporaryFailure`,
+        // which freezes the channel until persistence succeeds.
+        Err(ChannelMonitorUpdateErr::TemporaryFailure)
     }
 
     fn update_persisted_channel(
         &self,
         funding_txo: OutPoint,
+        // TODO: We probably want to use this for rollback protection
         _update: &Option<ChannelMonitorUpdate>,
         monitor: &ChannelMonitorType,
-        _update_id: MonitorUpdateId,
+        update_id: MonitorUpdateId,
     ) -> Result<(), ChannelMonitorUpdateErr> {
-        let outpoint = LxOutPoint::from(funding_txo);
-        let outpoint_str = outpoint.to_string();
-        debug!("Updating persisted channel {}", outpoint_str);
+        let funding_txo = LxOutPoint::from(funding_txo);
+        debug!("Updating persisted channel {}", funding_txo);
 
         // FIXME(encrypt): Encrypt under key derived from seed
         let data = monitor.encode();
@@ -442,18 +473,40 @@ impl Persist<SignerType> for InnerPersister {
             self.node_pk,
             self.measurement,
             CHANNEL_MONITORS_DIRECTORY.to_owned(),
-            outpoint_str,
+            funding_txo.to_string(),
             data,
         );
+        let update = LxChannelMonitorUpdate {
+            funding_txo,
+            update_id,
+        };
 
-        // Run an async fn inside a sync fn inside a Tokio runtime
-        Handle::current()
-            .block_on(async move { self.api.upsert_file(&cm_file).await })
-            .map(|_| ())
-            .map_err(|e| {
-                // TODO(max): Implement durability then make this err permanent
-                error!("Could not update persisted channel monitor: {:#}", e);
-                ChannelMonitorUpdateErr::TemporaryFailure
-            })
+        // Spawn a task for persisting the channel monitor
+        let api_clone = self.api.clone();
+        let channel_monitor_updated_tx =
+            self.channel_monitor_updated_tx.clone();
+        tokio::spawn(async move {
+            // Retry indefinitely until it succeeds
+            loop {
+                // TODO Also attempt to persist to cloud backup
+                match api_clone.upsert_file(&cm_file).await {
+                    Ok(_) => {
+                        if let Err(e) =
+                            channel_monitor_updated_tx.try_send(update)
+                        {
+                            error!("Couldn't notify chain monitor: {:#}", e);
+                        }
+                        return;
+                    }
+                    Err(e) => {
+                        error!("Could not update channel monitor: {:#}", e)
+                    }
+                }
+            }
+        });
+
+        // As documented in the `Persist` trait docs, return `TemporaryFailure`,
+        // which freezes the channel until persistence succeeds.
+        Err(ChannelMonitorUpdateErr::TemporaryFailure)
     }
 }
diff --git a/node/src/lexe/types.rs b/node/src/lexe/types.rs
@@ -34,6 +34,15 @@ impl From<OutPoint> for LxOutPoint {
     }
 }
 
+impl From<LxOutPoint> for OutPoint {
+    fn from(op: LxOutPoint) -> Self {
+        Self {
+            txid: op.txid,
+            index: op.index,
+        }
+    }
+}
+
 /// Deserializes from <txid>_<index>
 impl FromStr for LxOutPoint {
     type Err = anyhow::Error;
