Add async persistence logic in MonitorUpdatingPersister

In the next commit we'll add the ability to use an async `KVStore`
as the backing for a `ChainMonitor`. Here we tee this up by adding
an async API to `MonitorUpdatingPersisterAsync`. Its not intended
for public use and is thus only `pub(crate)` but allows spawning
all operations via a generic `FutureSpawner` trait, initiating the
write via the `KVStore` before any `await`s (or async functions).

Because we aren't going to make the `ChannelManager` (or
`ChainMonitor`) fully async, we need a way to alert the
`ChainMonitor` when a persistence completes, but we leave that for
the next commit.

Author: TheBlueMatt

lightning/src/util/persist.rs

@@ -17,6 +17,7 @@ use bitcoin::hashes::hex::FromHex;
 use bitcoin::{BlockHash, Txid};
 
 use core::future::Future;
+use core::mem;
 use core::ops::Deref;
 use core::pin::Pin;
 use core::str::FromStr;
@@ -32,7 +33,8 @@ use crate::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate};
 use crate::chain::transaction::OutPoint;
 use crate::ln::types::ChannelId;
 use crate::sign::{ecdsa::EcdsaChannelSigner, EntropySource, SignerProvider};
-use crate::util::async_poll::dummy_waker;
+use crate::sync::Mutex;
+use crate::util::async_poll::{dummy_waker, MaybeSend, MaybeSync};
 use crate::util::logger::Logger;
 use crate::util::ser::{Readable, ReadableArgs, Writeable};
 
@@ -409,6 +411,21 @@ where
 	Ok(res)
 }
 
+/// A generic trait which is able to spawn futures in the background.
+pub trait FutureSpawner: Send + Sync + 'static {
+	/// Spawns the given future as a background task.
+	///
+	/// This method MUST NOT block on the given future immediately.
+	fn spawn<T: Future<Output = ()> + MaybeSend + 'static>(&self, future: T);
+}
+
+struct PanicingSpawner;
+impl FutureSpawner for PanicingSpawner {
+	fn spawn<T: Future<Output = ()> + MaybeSend + 'static>(&self, _: T) {
+		unreachable!();
+	}
+}
+
 fn poll_sync_future<F: Future>(future: F) -> F::Output {
 	let mut waker = dummy_waker();
 	let mut ctx = task::Context::from_waker(&mut waker);
@@ -507,7 +524,7 @@ fn poll_sync_future<F: Future>(future: F) -> F::Output {
 /// would like to get rid of them, consider using the
 /// [`MonitorUpdatingPersister::cleanup_stale_updates`] function.
 pub struct MonitorUpdatingPersister<K: Deref, L: Deref, ES: Deref, SP: Deref, BI: Deref, FE: Deref>
-(MonitorUpdatingPersisterAsync<KVStoreSyncWrapper<K>, L, ES, SP, BI, FE>)
+(MonitorUpdatingPersisterAsync<KVStoreSyncWrapper<K>, PanicingSpawner, L, ES, SP, BI, FE>)
 where
 	K::Target: KVStoreSync,
 	L::Target: Logger,
@@ -552,6 +569,7 @@ where
 	) -> Self {
 		MonitorUpdatingPersister(MonitorUpdatingPersisterAsync::new(
 			KVStoreSyncWrapper(kv_store),
+			PanicingSpawner,
 			logger,
 			maximum_pending_updates,
 			entropy_source,
@@ -664,7 +682,8 @@ where
 		&self, monitor_name: MonitorName, update: Option<&ChannelMonitorUpdate>,
 		monitor: &ChannelMonitor<ChannelSigner>,
 	) -> chain::ChannelMonitorUpdateStatus {
-		let res = poll_sync_future(self.0.0.update_persisted_channel(monitor_name, update, monitor));
+		let inner = Arc::clone(&self.0.0);
+		let res = poll_sync_future(inner.update_persisted_channel(monitor_name, update, monitor));
 		match res {
 			Ok(()) => chain::ChannelMonitorUpdateStatus::Completed,
 			Err(e) => {
@@ -689,8 +708,11 @@ where
 /// async versions of the public accessors.
 ///
 /// Note that async monitor updating is considered beta, and bugs may be triggered by its use.
-pub struct MonitorUpdatingPersisterAsync<K: Deref, L: Deref, ES: Deref, SP: Deref, BI: Deref, FE: Deref>
-(Arc<MonitorUpdatingPersisterAsyncInner<K, L, ES, SP, BI, FE>>)
+///
+/// Unlike [`MonitorUpdatingPersister`], this does not implement [`Persist`], but is instead used
+/// directly by the [`ChainMonitor`].
+pub struct MonitorUpdatingPersisterAsync<K: Deref, S: FutureSpawner, L: Deref, ES: Deref, SP: Deref, BI: Deref, FE: Deref>
+(Arc<MonitorUpdatingPersisterAsyncInner<K, S, L, ES, SP, BI, FE>>)
 where
 	K::Target: KVStore,
 	L::Target: Logger,
@@ -699,7 +721,7 @@ where
 	BI::Target: BroadcasterInterface,
 	FE::Target: FeeEstimator;
 
-struct MonitorUpdatingPersisterAsyncInner<K: Deref, L: Deref, ES: Deref, SP: Deref, BI: Deref, FE: Deref>
+struct MonitorUpdatingPersisterAsyncInner<K: Deref, S: FutureSpawner, L: Deref, ES: Deref, SP: Deref, BI: Deref, FE: Deref>
 where
 	K::Target: KVStore,
 	L::Target: Logger,
@@ -709,6 +731,7 @@ where
 	FE::Target: FeeEstimator,
 {
 	kv_store: K,
+	future_spawner: S,
 	logger: L,
 	maximum_pending_updates: u64,
 	entropy_source: ES,
@@ -717,8 +740,8 @@ where
 	fee_estimator: FE,
 }
 
-impl<K: Deref, L: Deref, ES: Deref, SP: Deref, BI: Deref, FE: Deref>
-	MonitorUpdatingPersisterAsync<K, L, ES, SP, BI, FE>
+impl<K: Deref, S: FutureSpawner, L: Deref, ES: Deref, SP: Deref, BI: Deref, FE: Deref>
+	MonitorUpdatingPersisterAsync<K, S, L, ES, SP, BI, FE>
 where
 	K::Target: KVStore,
 	L::Target: Logger,
@@ -731,11 +754,12 @@ where
 	///
 	/// See [`MonitorUpdatingPersister::new`] for more info.
 	pub fn new(
-		kv_store: K, logger: L, maximum_pending_updates: u64, entropy_source: ES,
-		signer_provider: SP, broadcaster: BI, fee_estimator: FE,
+		kv_store: K, future_spawner: S, logger: L, maximum_pending_updates: u64,
+		entropy_source: ES, signer_provider: SP, broadcaster: BI, fee_estimator: FE,
 	) -> Self {
 		MonitorUpdatingPersisterAsync(Arc::new(MonitorUpdatingPersisterAsyncInner {
 			kv_store,
+			future_spawner,
 			logger,
 			maximum_pending_updates,
 			entropy_source,
@@ -805,9 +829,70 @@ where
 	}
 }
 
+impl<K: Deref + MaybeSend + MaybeSync + 'static, S: FutureSpawner, L: Deref + MaybeSend + MaybeSync + 'static, ES: Deref + MaybeSend + MaybeSync + 'static, SP: Deref + MaybeSend + MaybeSync + 'static, BI: Deref + MaybeSend + MaybeSync + 'static, FE: Deref + MaybeSend + MaybeSync + 'static>
+	MonitorUpdatingPersisterAsync<K, S, L, ES, SP, BI, FE>
+where
+	K::Target: KVStore + MaybeSync,
+	L::Target: Logger,
+	ES::Target: EntropySource + Sized,
+	SP::Target: SignerProvider + Sized,
+	BI::Target: BroadcasterInterface,
+	FE::Target: FeeEstimator,
+	<SP::Target as SignerProvider>::EcdsaSigner: MaybeSend + 'static,
+{
+	pub(crate) fn spawn_async_persist_new_channel(
+		&self, monitor_name: MonitorName, monitor: &ChannelMonitor<<SP::Target as SignerProvider>::EcdsaSigner>,
+	) {
+		let inner = Arc::clone(&self.0);
+		let future = inner.persist_new_channel(monitor_name, monitor);
+		let channel_id = monitor.channel_id();
+		self.0.future_spawner.spawn(async move {
+			match future.await {
+				Ok(()) => {}, // TODO: expose completions
+				Err(e) => {
+					log_error!(
+						inner.logger,
+						"Failed to persist new ChannelMonitor {channel_id}: {e}. The node will now likely stall as this channel will not be able to make progress. You should restart as soon as possible.",
+					);
+				},
+			}
+		});
+	}
 
-impl<K: Deref, L: Deref, ES: Deref, SP: Deref, BI: Deref, FE: Deref>
-	MonitorUpdatingPersisterAsyncInner<K, L, ES, SP, BI, FE>
+	pub(crate) fn spawn_async_update_persisted_channel(
+		&self, monitor_name: MonitorName, update: Option<&ChannelMonitorUpdate>,
+		monitor: &ChannelMonitor<<SP::Target as SignerProvider>::EcdsaSigner>,
+	) {
+		let inner = Arc::clone(&self.0);
+		let future = inner.update_persisted_channel(monitor_name, update, monitor);
+		let channel_id = monitor.channel_id();
+		let inner = Arc::clone(&self.0);
+		self.0.future_spawner.spawn(async move {
+			match future.await {
+				Ok(()) => {}, // TODO: expose completions
+				Err(e) => {
+					log_error!(
+						inner.logger,
+						"Failed to persist new ChannelMonitor {channel_id}: {e}. The node will now likely stall as this channel will not be able to make progress. You should restart as soon as possible.",
+					);
+				},
+			}
+		});
+	}
+
+	pub(crate) fn spawn_async_archive_persisted_channel(
+		&self, monitor_name: MonitorName,
+	) {
+		let inner = Arc::clone(&self.0);
+		self.0.future_spawner.spawn(async move {
+			inner.archive_persisted_channel(monitor_name).await;
+		});
+	}
+}
+
+
+impl<K: Deref, S: FutureSpawner, L: Deref, ES: Deref, SP: Deref, BI: Deref, FE: Deref>
+	MonitorUpdatingPersisterAsyncInner<K, S, L, ES, SP, BI, FE>
 where
 	K::Target: KVStore,
 	L::Target: Logger,
@@ -931,7 +1016,7 @@ where
 			let monitor_name = MonitorName::from_str(&monitor_key)?;
 			let (_, current_monitor) = self.read_monitor(&monitor_name, &monitor_key).await?;
 			let latest_update_id = current_monitor.get_latest_update_id();
-			self.cleanup_stale_updates_for_monitor_to(&monitor_key, latest_update_id, lazy).await;
+			self.cleanup_stale_updates_for_monitor_to(&monitor_key, latest_update_id, lazy).await?;
 		}
 		Ok(())
 	}
@@ -958,9 +1043,9 @@ where
 		Ok(())
 	}
 
-	async fn persist_new_channel<ChannelSigner: EcdsaChannelSigner>(
+	fn persist_new_channel<ChannelSigner: EcdsaChannelSigner>(
 		&self, monitor_name: MonitorName, monitor: &ChannelMonitor<ChannelSigner>,
-	) -> Result<(), io::Error> {
+	) -> impl Future<Output = Result<(), io::Error>> {
 		// Determine the proper key for this monitor
 		let monitor_key = monitor_name.to_string();
 		// Serialize and write the new monitor
@@ -974,55 +1059,77 @@ where
 			monitor_bytes.extend_from_slice(MONITOR_UPDATING_PERSISTER_PREPEND_SENTINEL);
 		}
 		monitor.write(&mut monitor_bytes).unwrap();
+		// Note that this is NOT an async function, but rather calls the *sync* KVStore write
+		// method, allowing it to do its queueing immediately, and then return a future for the
+		// completion of the write. This ensures monitor persistence ordering is preserved.
 		self.kv_store.write(
 			CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
 			CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
 			monitor_key.as_str(),
 			monitor_bytes,
-		).await
+		)
 	}
 
-	async fn update_persisted_channel<ChannelSigner: EcdsaChannelSigner>(
-		&self, monitor_name: MonitorName, update: Option<&ChannelMonitorUpdate>,
+	fn update_persisted_channel<'a, ChannelSigner: EcdsaChannelSigner + 'a>(
+		self: Arc<Self>, monitor_name: MonitorName, update: Option<&ChannelMonitorUpdate>,
 		monitor: &ChannelMonitor<ChannelSigner>,
-	) -> Result<(), io::Error> {
+	) -> impl Future<Output = Result<(), io::Error>> + 'a where Self: 'a {
 		const LEGACY_CLOSED_CHANNEL_UPDATE_ID: u64 = u64::MAX;
+		let mut res_a = None;
+		let mut res_b = None;
+		let mut res_c = None;
 		if let Some(update) = update {
 			let persist_update = update.update_id != LEGACY_CLOSED_CHANNEL_UPDATE_ID
 				&& self.maximum_pending_updates != 0
 				&& update.update_id % self.maximum_pending_updates != 0;
 			if persist_update {
 				let monitor_key = monitor_name.to_string();
 				let update_name = UpdateName::from(update.update_id);
-				self.kv_store.write(
+				res_a = Some(self.kv_store.write(
 					CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
 					monitor_key.as_str(),
 					update_name.as_str(),
 					update.encode(),
-				).await
+				));
 			} else {
 				// We could write this update, but it meets criteria of our design that calls for a full monitor write.
-				let write_status = self.persist_new_channel(monitor_name, monitor).await;
-
-				if let Ok(()) = write_status {
-					let channel_closed_legacy =
-						monitor.get_latest_update_id() == LEGACY_CLOSED_CHANNEL_UPDATE_ID;
-					let latest_update_id = monitor.get_latest_update_id();
-					if channel_closed_legacy {
-						let monitor_key = monitor_name.to_string();
-						self.cleanup_stale_updates_for_monitor_to(&monitor_key, latest_update_id, true).await;
-					} else {
-						let end = latest_update_id;
-						let start = end.saturating_sub(self.maximum_pending_updates);
-						self.cleanup_in_range(monitor_name, start, end).await;
+				let write_fut = self.persist_new_channel(monitor_name, monitor);
+				let latest_update_id = monitor.get_latest_update_id();
+
+				res_b = Some(async move {
+					let write_status = write_fut.await;
+					if let Ok(()) = write_status {
+						if latest_update_id == LEGACY_CLOSED_CHANNEL_UPDATE_ID {
+							let monitor_key = monitor_name.to_string();
+							self.cleanup_stale_updates_for_monitor_to(&monitor_key, latest_update_id, true).await?;
+						} else {
+							let end = latest_update_id;
+							let start = end.saturating_sub(self.maximum_pending_updates);
+							self.cleanup_in_range(monitor_name, start, end).await;
+						}
 					}
-				}
 
-				write_status
+					write_status
+				});
 			}
 		} else {
 			// There is no update given, so we must persist a new monitor.
-			self.persist_new_channel(monitor_name, monitor).await
+			// Note that this is NOT an async function, but rather calls the *sync* KVStore write
+			// method, allowing it to do its queueing immediately, and then return a future for the
+			// completion of the write. This ensures monitor persistence ordering is preserved.
+			res_c = Some(self.persist_new_channel(monitor_name, monitor));
+		}
+		async move {
+			if let Some(a) = res_a {
+				a.await?;
+			}
+			if let Some(b) = res_b {
+				b.await?;
+			}
+			if let Some(c) = res_c {
+				c.await?;
+			}
+			Ok(())
 		}
 	}