Add support for native async `KVStore` persist to `ChainMonitor`

diff --git a/lightning/src/chain/chainmonitor.rs b/lightning/src/chain/chainmonitor.rs
index 36d26ae..4abd0cd 100644
--- a/lightning/src/chain/chainmonitor.rs
+++ b/lightning/src/chain/chainmonitor.rs
@@ -46,12 +46,14 @@ use crate::ln::our_peer_storage::{DecryptedOurPeerStorage, PeerStorageMonitorHol
 use crate::ln::types::ChannelId;
 use crate::prelude::*;
 use crate::sign::ecdsa::EcdsaChannelSigner;
-use crate::sign::{EntropySource, PeerStorageKey};
+use crate::sign::{EntropySource, PeerStorageKey, SignerProvider};
 use crate::sync::{Mutex, MutexGuard, RwLock, RwLockReadGuard};
 use crate::types::features::{InitFeatures, NodeFeatures};
+use crate::util::async_poll::{MaybeSend, MaybeSync};
 use crate::util::errors::APIError;
 use crate::util::logger::{Logger, WithContext};
-use crate::util::persist::MonitorName;
+use crate::util::native_async::FutureSpawner;
+use crate::util::persist::{KVStore, MonitorName, MonitorUpdatingPersisterAsync};
 #[cfg(peer_storage)]
 use crate::util::ser::{VecWriter, Writeable};
 use crate::util::wakers::{Future, Notifier};
@@ -192,6 +194,17 @@ pub trait Persist<ChannelSigner: EcdsaChannelSigner> {
 	/// restart, this method must in that case be idempotent, ensuring it can handle scenarios where
 	/// the monitor already exists in the archive.
 	fn archive_persisted_channel(&self, monitor_name: MonitorName);
+
+	/// Fetches the set of [`ChannelMonitorUpdate`]s, previously persisted with
+	/// [`Self::update_persisted_channel`], which have completed.
+	///
+	/// Returning an update here is equivalent to calling
+	/// [`ChainMonitor::channel_monitor_updated`]. Because of this, this method is defaulted and
+	/// hidden in the docs.
+	#[doc(hidden)]
+	fn get_and_clear_completed_updates(&self) -> Vec<(ChannelId, u64)> {
+		Vec::new()
+	}
 }
 
 struct MonitorHolder<ChannelSigner: EcdsaChannelSigner> {
@@ -235,6 +248,93 @@ impl<ChannelSigner: EcdsaChannelSigner> Deref for LockedChannelMonitor<'_, Chann
 	}
 }
 
+/// An unconstructable [`Persist`]er which is used under the hood when you call
+/// [`ChainMonitor::new_async_beta`].
+pub struct AsyncPersister<
+	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,
+> where
+	K::Target: KVStore + MaybeSync,
+	L::Target: Logger,
+	ES::Target: EntropySource + Sized,
+	SP::Target: SignerProvider + Sized,
+	BI::Target: BroadcasterInterface,
+	FE::Target: FeeEstimator,
+{
+	persister: MonitorUpdatingPersisterAsync<K, S, L, ES, SP, BI, FE>,
+}
+
+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,
+	> Deref for AsyncPersister<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,
+{
+	type Target = Self;
+	fn deref(&self) -> &Self {
+		self
+	}
+}
+
+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,
+	> Persist<<SP::Target as SignerProvider>::EcdsaSigner> for AsyncPersister<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,
+{
+	fn persist_new_channel(
+		&self, monitor_name: MonitorName,
+		monitor: &ChannelMonitor<<SP::Target as SignerProvider>::EcdsaSigner>,
+	) -> ChannelMonitorUpdateStatus {
+		self.persister.spawn_async_persist_new_channel(monitor_name, monitor);
+		ChannelMonitorUpdateStatus::InProgress
+	}
+
+	fn update_persisted_channel(
+		&self, monitor_name: MonitorName, monitor_update: Option<&ChannelMonitorUpdate>,
+		monitor: &ChannelMonitor<<SP::Target as SignerProvider>::EcdsaSigner>,
+	) -> ChannelMonitorUpdateStatus {
+		self.persister.spawn_async_update_persisted_channel(monitor_name, monitor_update, monitor);
+		ChannelMonitorUpdateStatus::InProgress
+	}
+
+	fn archive_persisted_channel(&self, monitor_name: MonitorName) {
+		self.persister.spawn_async_archive_persisted_channel(monitor_name);
+	}
+
+	fn get_and_clear_completed_updates(&self) -> Vec<(ChannelId, u64)> {
+		self.persister.get_and_clear_completed_updates()
+	}
+}
+
 /// An implementation of [`chain::Watch`] for monitoring channels.
 ///
 /// Connected and disconnected blocks must be provided to `ChainMonitor` as documented by
@@ -291,6 +391,63 @@ pub struct ChainMonitor<
 	our_peerstorage_encryption_key: PeerStorageKey,
 }
 
+impl<
+		K: Deref + MaybeSend + MaybeSync + 'static,
+		S: FutureSpawner,
+		SP: Deref + MaybeSend + MaybeSync + 'static,
+		C: Deref,
+		T: Deref + MaybeSend + MaybeSync + 'static,
+		F: Deref + MaybeSend + MaybeSync + 'static,
+		L: Deref + MaybeSend + MaybeSync + 'static,
+		ES: Deref + MaybeSend + MaybeSync + 'static,
+	>
+	ChainMonitor<
+		<SP::Target as SignerProvider>::EcdsaSigner,
+		C,
+		T,
+		F,
+		L,
+		AsyncPersister<K, S, L, ES, SP, T, F>,
+		ES,
+	> where
+	K::Target: KVStore + MaybeSync,
+	SP::Target: SignerProvider + Sized,
+	C::Target: chain::Filter,
+	T::Target: BroadcasterInterface,
+	F::Target: FeeEstimator,
+	L::Target: Logger,
+	ES::Target: EntropySource + Sized,
+	<SP::Target as SignerProvider>::EcdsaSigner: MaybeSend + 'static,
+{
+	/// Creates a new `ChainMonitor` used to watch on-chain activity pertaining to channels.
+	///
+	/// This behaves the same as [`ChainMonitor::new`] except that it relies on
+	/// [`MonitorUpdatingPersisterAsync`] and thus allows persistence to be completed async.
+	///
+	/// Note that async monitor updating is considered beta, and bugs may be triggered by its use.
+	pub fn new_async_beta(
+		chain_source: Option<C>, broadcaster: T, logger: L, feeest: F,
+		persister: MonitorUpdatingPersisterAsync<K, S, L, ES, SP, T, F>, _entropy_source: ES,
+		_our_peerstorage_encryption_key: PeerStorageKey,
+	) -> Self {
+		Self {
+			monitors: RwLock::new(new_hash_map()),
+			chain_source,
+			broadcaster,
+			logger,
+			fee_estimator: feeest,
+			persister: AsyncPersister { persister },
+			_entropy_source,
+			pending_monitor_events: Mutex::new(Vec::new()),
+			highest_chain_height: AtomicUsize::new(0),
+			event_notifier: Notifier::new(),
+			pending_send_only_events: Mutex::new(Vec::new()),
+			#[cfg(peer_storage)]
+			our_peerstorage_encryption_key: _our_peerstorage_encryption_key,
+		}
+	}
+}
+
 impl<
 		ChannelSigner: EcdsaChannelSigner,
 		C: Deref,
@@ -1357,6 +1514,9 @@ where
 	fn release_pending_monitor_events(
 		&self,
 	) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, PublicKey)> {
+		for (channel_id, update_id) in self.persister.get_and_clear_completed_updates() {
+			let _ = self.channel_monitor_updated(channel_id, update_id);
+		}
 		let mut pending_monitor_events = self.pending_monitor_events.lock().unwrap().split_off(0);
 		for monitor_state in self.monitors.read().unwrap().values() {
 			let monitor_events = monitor_state.monitor.get_and_clear_pending_monitor_events();
diff --git a/lightning/src/util/persist.rs b/lightning/src/util/persist.rs
index 8547b10..9036a27 100644
--- a/lightning/src/util/persist.rs
+++ b/lightning/src/util/persist.rs
@@ -561,6 +561,9 @@ where
 		kv_store: K, logger: L, maximum_pending_updates: u64, entropy_source: ES,
 		signer_provider: SP, broadcaster: BI, fee_estimator: FE,
 	) -> Self {
+		// Note that calling the spawner only happens in the `pub(crate)` `spawn_*` methods defined
+		// with additional bounds on `MonitorUpdatingPersisterAsync`. Thus its safe to provide a
+		// dummy always-panic implementation here.
 		MonitorUpdatingPersister(MonitorUpdatingPersisterAsync::new(
 			KVStoreSyncWrapper(kv_store),
 			PanicingSpawner,
@@ -704,9 +707,10 @@ where
 /// Note that async monitor updating is considered beta, and bugs may be triggered by its use.
 ///
 /// Unlike [`MonitorUpdatingPersister`], this does not implement [`Persist`], but is instead used
-/// directly by the [`ChainMonitor`].
+/// directly by the [`ChainMonitor`] via [`ChainMonitor::new_async_beta`].
 ///
 /// [`ChainMonitor`]: crate::chain::chainmonitor::ChainMonitor
+/// [`ChainMonitor::new_async_beta`]: crate::chain::chainmonitor::ChainMonitor::new_async_beta
 pub struct MonitorUpdatingPersisterAsync<
 	K: Deref,
 	S: FutureSpawner,
@@ -741,6 +745,7 @@ struct MonitorUpdatingPersisterAsyncInner<
 	FE::Target: FeeEstimator,
 {
 	kv_store: K,
+	async_completed_updates: Mutex<Vec<(ChannelId, u64)>>,
 	future_spawner: S,
 	logger: L,
 	maximum_pending_updates: u64,
@@ -769,6 +774,7 @@ where
 	) -> Self {
 		MonitorUpdatingPersisterAsync(Arc::new(MonitorUpdatingPersisterAsyncInner {
 			kv_store,
+			async_completed_updates: Mutex::new(Vec::new()),
 			future_spawner,
 			logger,
 			maximum_pending_updates,
@@ -861,11 +867,14 @@ where
 		monitor: &ChannelMonitor<<SP::Target as SignerProvider>::EcdsaSigner>,
 	) {
 		let inner = Arc::clone(&self.0);
+		// Note that `persist_new_channel` is a sync method which calls all the way through to the
+		// sync KVStore::write method (which returns a future) to ensure writes are well-ordered.
 		let future = inner.persist_new_channel(monitor_name, monitor);
 		let channel_id = monitor.channel_id();
+		let completion = (monitor.channel_id(), monitor.get_latest_update_id());
 		self.0.future_spawner.spawn(async move {
 			match future.await {
-				Ok(()) => {}, // TODO: expose completions
+				Ok(()) => inner.async_completed_updates.lock().unwrap().push(completion),
 				Err(e) => {
 					log_error!(
 						inner.logger,
@@ -881,12 +890,21 @@ where
 		monitor: &ChannelMonitor<<SP::Target as SignerProvider>::EcdsaSigner>,
 	) {
 		let inner = Arc::clone(&self.0);
+		// Note that `update_persisted_channel` is a sync method which calls all the way through to
+		// the sync KVStore::write method (which returns a future) to ensure writes are well-ordered
 		let future = inner.update_persisted_channel(monitor_name, update, monitor);
 		let channel_id = monitor.channel_id();
+		let completion = if let Some(update) = update {
+			Some((monitor.channel_id(), update.update_id))
+		} else {
+			None
+		};
 		let inner = Arc::clone(&self.0);
 		self.0.future_spawner.spawn(async move {
 			match future.await {
-				Ok(()) => {}, // TODO: expose completions
+				Ok(()) => if let Some(completion) = completion {
+					inner.async_completed_updates.lock().unwrap().push(completion);
+				},
 				Err(e) => {
 					log_error!(
 						inner.logger,
@@ -903,6 +921,10 @@ where
 			inner.archive_persisted_channel(monitor_name).await;
 		});
 	}
+
+	pub(crate) fn get_and_clear_completed_updates(&self) -> Vec<(ChannelId, u64)> {
+		mem::take(&mut *self.0.async_completed_updates.lock().unwrap())
+	}
 }
 
 impl<K: Deref, S: FutureSpawner, L: Deref, ES: Deref, SP: Deref, BI: Deref, FE: Deref>