Extend EventsProvider to handle events asynchronously

Unfortunately, for both ChannelManager and ChainMonitor, we were not
able to reuse the existing process_pending_events logic (and instead had
to copy them verbatim for the async variant) due to async closures not
being supported on stable at this time.

Author: wpaulino

lightning/src/chain/chainmonitor.rs

@@ -36,12 +36,14 @@ use crate::util::atomic_counter::AtomicCounter;
 use crate::util::logger::Logger;
 use crate::util::errors::APIError;
 use crate::util::events;
-use crate::util::events::EventHandler;
+use crate::util::events::{AsyncEventHandler, EventHandler};
 use crate::ln::channelmanager::ChannelDetails;
 
 use crate::prelude::*;
 use crate::sync::{RwLock, RwLockReadGuard, Mutex, MutexGuard};
+use core::future::Future as StdFuture;
 use core::ops::Deref;
+use core::pin::Pin;
 use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use bitcoin::secp256k1::PublicKey;
 
@@ -723,6 +725,7 @@ impl<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> even
 			handler.handle_event(event);
 		}
 	}
+
 	#[cfg(anchors)]
 	/// Processes [`SpendableOutputs`] events produced from each [`ChannelMonitor`] upon maturity.
 	///
@@ -746,6 +749,24 @@ impl<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> even
 			handler.handle_event(event);
 		}
 	}
+
+	/// Processes any events asynchronously in the order they were generated since the last call
+	/// using the given event handler.
+	///
+	/// See the trait-level documentation for requirements.
+	fn process_pending_events_async<'a, H: Deref + 'a>(
+		&'a self, handler: H
+	) -> Pin<Box<dyn StdFuture<Output = ()> + '_>> where H::Target: AsyncEventHandler {
+		Box::pin(async move {
+			let mut pending_events = Vec::new();
+			for monitor_state in self.monitors.read().unwrap().values() {
+				pending_events.append(&mut monitor_state.monitor.get_and_clear_pending_events());
+			}
+			for event in pending_events.drain(..) {
+				handler.handle_event_async(&event).await;
+			}
+		})
+    }
 }
 
 #[cfg(test)]

lightning/src/ln/channelmanager.rs

@@ -53,7 +53,7 @@ use crate::ln::msgs::{ChannelMessageHandler, DecodeError, LightningError, MAX_VA
 use crate::ln::wire::Encode;
 use crate::chain::keysinterface::{Sign, KeysInterface, KeysManager, Recipient};
 use crate::util::config::{UserConfig, ChannelConfig};
-use crate::util::events::{EventHandler, EventsProvider, MessageSendEvent, MessageSendEventsProvider, ClosureReason, HTLCDestination};
+use crate::util::events::{AsyncEventHandler, EventHandler, EventsProvider, MessageSendEvent, MessageSendEventsProvider, ClosureReason, HTLCDestination};
 use crate::util::{byte_utils, events};
 use crate::util::wakers::{Future, Notifier};
 use crate::util::scid_utils::fake_scid;
@@ -67,9 +67,11 @@ use core::{cmp, mem};
 use core::cell::RefCell;
 use crate::io::Read;
 use crate::sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard};
+use core::future::Future as StdFuture;
 use core::sync::atomic::{AtomicUsize, Ordering};
 use core::time::Duration;
 use core::ops::Deref;
+use core::pin::Pin;
 
 // We hold various information about HTLC relay in the HTLC objects in Channel itself:
 //
@@ -5641,6 +5643,40 @@ where
 			result
 		});
 	}
+
+	/// Processes any events asynchronously in the order they were generated since the last call
+	/// using the given event handler.
+	///
+	/// See the trait-level documentation for requirements.
+	fn process_pending_events_async<'a, H: Deref + 'a>(
+		&'a self, handler: H
+	) -> Pin<Box<dyn StdFuture<Output = ()> + '_>> where H::Target: AsyncEventHandler {
+		// We can't use `optionally_notify` with an async closure as required by the
+		// `handle_event_async` call since it's not supported on stable.
+		Box::pin(async move {
+			let _read_guard = self.total_consistency_lock.read().unwrap();
+			let mut result = NotifyOption::SkipPersist;
+
+			// TODO: This behavior should be documented. It's unintuitive that we query
+			// ChannelMonitors when clearing other events.
+			if self.process_pending_monitor_events() {
+				result = NotifyOption::DoPersist;
+			}
+
+			let mut pending_events = mem::replace(&mut *self.pending_events.lock().unwrap(), vec![]);
+			if !pending_events.is_empty() {
+				result = NotifyOption::DoPersist;
+			}
+
+			for event in pending_events.drain(..) {
+				handler.handle_event_async(&event).await;
+			}
+
+			if result == NotifyOption::DoPersist {
+				self.persistence_notifier.notify();
+			}
+		})
+	}
 }
 
 impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> chain::Listen for ChannelManager<M, T, K, F, L>

lightning/src/util/events.rs

@@ -33,8 +33,10 @@ use bitcoin::hashes::sha256::Hash as Sha256;
 use bitcoin::secp256k1::PublicKey;
 use crate::io;
 use crate::prelude::*;
+use core::future::Future as StdFuture;
 use core::time::Duration;
 use core::ops::Deref;
+use core::pin::Pin;
 use crate::sync::Arc;
 
 /// Some information provided on receipt of payment depends on whether the payment received is a
@@ -1370,6 +1372,14 @@ pub trait EventsProvider {
 	///
 	/// See the trait-level documentation for requirements.
 	fn process_pending_events<H: Deref>(&self, handler: H) where H::Target: EventHandler;
+
+	/// Processes any events asynchronously in the order they were generated since the last call
+	/// using the given event handler.
+	///
+	/// See the trait-level documentation for requirements.
+	fn process_pending_events_async<'a, H: Deref + 'a>(
+		&'a self, handler: H
+	) -> Pin<Box<dyn StdFuture<Output = ()> + 'a>> where H::Target: AsyncEventHandler;
 }
 
 /// A trait implemented for objects handling events from [`EventsProvider`].
@@ -1391,3 +1401,27 @@ impl<T: EventHandler> EventHandler for Arc<T> {
 		self.deref().handle_event(event)
 	}
 }
+
+/// A trait implemented for objects handling events from [`EventsProvider`] asynchronously.
+pub trait AsyncEventHandler {
+	/// Handles the given [`Event`] asynchronously.
+	///
+	/// See [`EventsProvider`] for details that must be considered when implementing this method.
+	fn handle_event_async<'a>(&'a self, event: &'a Event) -> Pin<Box<dyn StdFuture<Output = ()> + 'a>>;
+}
+
+impl<F> AsyncEventHandler for F where F: Fn(&Event) {
+	fn handle_event_async<'a>(&'a self, event: &'a Event) -> Pin<Box<dyn StdFuture<Output = ()> + 'a>> {
+		Box::pin(async move {
+			self(event);
+		})
+	}
+}
+
+impl<T: AsyncEventHandler> AsyncEventHandler for Arc<T> {
+	fn handle_event_async<'a>(&'a self, event: &'a Event) -> Pin<Box<dyn StdFuture<Output = ()> + 'a>> {
+		Box::pin(async move {
+			self.deref().handle_event_async(event).await
+		})
+	}
+}