Convert process_events_async to take an asynchronous Persister

Also provide a wrapper to allow a sync kvstore to be used.

Author: joostjager

lightning-background-processor/src/lib.rs

@@ -40,7 +40,9 @@ use lightning::sign::ChangeDestinationSourceSync;
 use lightning::sign::EntropySource;
 use lightning::sign::OutputSpender;
 use lightning::util::logger::Logger;
-use lightning::util::persist::{KVStoreSync, PersisterSync};
+#[cfg(feature = "std")]
+use lightning::util::persist::PersisterSync;
+use lightning::util::persist::{KVStoreSync, Persister};
 use lightning::util::sweep::OutputSweeper;
 #[cfg(feature = "std")]
 use lightning::util::sweep::OutputSweeperSync;
@@ -50,7 +52,9 @@ use lightning_rapid_gossip_sync::RapidGossipSync;
 
 use lightning_liquidity::ALiquidityManager;
 
+use core::future::Future;
 use core::ops::Deref;
+use core::pin::Pin;
 use core::time::Duration;
 
 #[cfg(feature = "std")]
@@ -311,6 +315,15 @@ fn update_scorer<'a, S: 'static + Deref<Target = SC> + Send + Sync, SC: 'a + Wri
 	true
 }
 
+macro_rules! maybe_await {
+	(true, $e:expr) => {
+		$e.await
+	};
+	(false, $e:expr) => {
+		$e
+	};
+}
+
 macro_rules! define_run_body {
 	(
 		$persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr,
@@ -319,7 +332,7 @@ macro_rules! define_run_body {
 		$peer_manager: ident, $gossip_sync: ident,
 		$process_sweeper: expr,
 		$logger: ident, $scorer: ident, $loop_exit_check: expr, $await: expr, $get_timer: expr,
-		$timer_elapsed: expr, $check_slow_await: expr, $time_fetch: expr,
+		$timer_elapsed: expr, $check_slow_await: expr, $time_fetch: expr, $async: tt,
 	) => { {
 		log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
 		$channel_manager.get_cm().timer_tick_occurred();
@@ -375,7 +388,7 @@ macro_rules! define_run_body {
 
 			if $channel_manager.get_cm().get_and_clear_needs_persistence() {
 				log_trace!($logger, "Persisting ChannelManager...");
-				$persister.persist_manager(&$channel_manager)?;
+				maybe_await!($async, $persister.persist_manager(&$channel_manager))?;
 				log_trace!($logger, "Done persisting ChannelManager.");
 			}
 			if $timer_elapsed(&mut last_freshness_call, FRESHNESS_TIMER) {
@@ -436,7 +449,7 @@ macro_rules! define_run_body {
 						log_trace!($logger, "Persisting network graph.");
 					}
 
-					if let Err(e) = $persister.persist_graph(network_graph) {
+					if let Err(e) = maybe_await!($async, $persister.persist_graph(network_graph)) {
 						log_error!($logger, "Error: Failed to persist network graph, check your disk and permissions {}", e)
 					}
 
@@ -464,7 +477,7 @@ macro_rules! define_run_body {
 					} else {
 						log_trace!($logger, "Persisting scorer");
 					}
-					if let Err(e) = $persister.persist_scorer(&scorer) {
+					if let Err(e) = maybe_await!($async, $persister.persist_scorer(&scorer)) {
 						log_error!($logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
 					}
 				}
@@ -487,16 +500,16 @@ macro_rules! define_run_body {
 		// After we exit, ensure we persist the ChannelManager one final time - this avoids
 		// some races where users quit while channel updates were in-flight, with
 		// ChannelMonitor update(s) persisted without a corresponding ChannelManager update.
-		$persister.persist_manager(&$channel_manager)?;
+		maybe_await!($async, $persister.persist_manager(&$channel_manager))?;
 
 		// Persist Scorer on exit
 		if let Some(ref scorer) = $scorer {
-			$persister.persist_scorer(&scorer)?;
+			maybe_await!($async, $persister.persist_scorer(&scorer))?;
 		}
 
 		// Persist NetworkGraph on exit
 		if let Some(network_graph) = $gossip_sync.network_graph() {
-			$persister.persist_graph(network_graph)?;
+			maybe_await!($async, $persister.persist_graph(network_graph))?;
 		}
 
 		Ok(())
@@ -643,22 +656,30 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
 /// ```
 /// # use lightning::io;
 /// # use lightning::events::ReplayEvent;
-/// # use lightning::util::sweep::OutputSweeper;
 /// # use std::sync::{Arc, RwLock};
 /// # use std::sync::atomic::{AtomicBool, Ordering};
 /// # use std::time::SystemTime;
 /// # use lightning_background_processor::{process_events_async, GossipSync};
+/// # use core::future::Future;
+/// # use core::pin::Pin;
 /// # struct Logger {}
 /// # impl lightning::util::logger::Logger for Logger {
 /// #     fn log(&self, _record: lightning::util::logger::Record) {}
 /// # }
-/// # struct Store {}
-/// # impl lightning::util::persist::KVStore for Store {
+/// # struct StoreSync {}
+/// # impl lightning::util::persist::KVStoreSync for StoreSync {
 /// #     fn read(&self, primary_namespace: &str, secondary_namespace: &str, key: &str) -> io::Result<Vec<u8>> { Ok(Vec::new()) }
 /// #     fn write(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8]) -> io::Result<()> { Ok(()) }
 /// #     fn remove(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool) -> io::Result<()> { Ok(()) }
 /// #     fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> io::Result<Vec<String>> { Ok(Vec::new()) }
 /// # }
+/// # struct Store {}
+/// # impl lightning::util::persist::KVStore for Store {
+/// #     fn read(&self, primary_namespace: &str, secondary_namespace: &str, key: &str) -> Pin<Box<dyn Future<Output = Result<Vec<u8>, io::Error>> + 'static + Send>> { todo!() }
+/// #     fn write(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8]) -> Pin<Box<dyn Future<Output = Result<(), io::Error>> + 'static + Send>> { todo!() }
+/// #     fn remove(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool) -> Pin<Box<dyn Future<Output = Result<(), io::Error>> + 'static + Send>> { todo!() }
+/// #     fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> Pin<Box<dyn Future<Output = Result<Vec<String>, io::Error>> + 'static + Send>> { todo!() }
+/// # }
 /// # struct EventHandler {}
 /// # impl EventHandler {
 /// #     async fn handle_event(&self, _: lightning::events::Event) -> Result<(), ReplayEvent> { Ok(()) }
@@ -669,22 +690,22 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
 /// #     fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize { 0 }
 /// #     fn disconnect_socket(&mut self) {}
 /// # }
-/// # type ChainMonitor<B, F, FE> = lightning::chain::chainmonitor::ChainMonitor<lightning::sign::InMemorySigner, Arc<F>, Arc<B>, Arc<FE>, Arc<Logger>, Arc<Store>, Arc<lightning::sign::KeysManager>>;
+/// # type ChainMonitor<B, F, FE> = lightning::chain::chainmonitor::ChainMonitor<lightning::sign::InMemorySigner, Arc<F>, Arc<B>, Arc<FE>, Arc<Logger>, Arc<StoreSync>, Arc<lightning::sign::KeysManager>>;
 /// # type NetworkGraph = lightning::routing::gossip::NetworkGraph<Arc<Logger>>;
 /// # type P2PGossipSync<UL> = lightning::routing::gossip::P2PGossipSync<Arc<NetworkGraph>, Arc<UL>, Arc<Logger>>;
 /// # type ChannelManager<B, F, FE> = lightning::ln::channelmanager::SimpleArcChannelManager<ChainMonitor<B, F, FE>, B, FE, Logger>;
 /// # type OnionMessenger<B, F, FE> = lightning::onion_message::messenger::OnionMessenger<Arc<lightning::sign::KeysManager>, Arc<lightning::sign::KeysManager>, Arc<Logger>, Arc<ChannelManager<B, F, FE>>, Arc<lightning::onion_message::messenger::DefaultMessageRouter<Arc<NetworkGraph>, Arc<Logger>, Arc<lightning::sign::KeysManager>>>, Arc<ChannelManager<B, F, FE>>, lightning::ln::peer_handler::IgnoringMessageHandler, lightning::ln::peer_handler::IgnoringMessageHandler, lightning::ln::peer_handler::IgnoringMessageHandler>;
 /// # type LiquidityManager<B, F, FE> = lightning_liquidity::LiquidityManager<Arc<lightning::sign::KeysManager>, Arc<ChannelManager<B, F, FE>>, Arc<F>>;
 /// # type Scorer = RwLock<lightning::routing::scoring::ProbabilisticScorer<Arc<NetworkGraph>, Arc<Logger>>>;
-/// # type PeerManager<B, F, FE, UL> = lightning::ln::peer_handler::SimpleArcPeerManager<SocketDescriptor, ChainMonitor<B, F, FE>, B, FE, Arc<UL>, Logger, F, Store>;
-/// #
+/// # type PeerManager<B, F, FE, UL> = lightning::ln::peer_handler::SimpleArcPeerManager<SocketDescriptor, ChainMonitor<B, F, FE>, B, FE, Arc<UL>, Logger, F, StoreSync>;
+/// # type OutputSweeper<B, D, FE, F, O> = lightning::util::sweep::OutputSweeper<Arc<B>, Arc<D>, Arc<FE>, Arc<F>, Arc<StoreSync>, Arc<Logger>, Arc<O>>;
+///
 /// # struct Node<
 /// #     B: lightning::chain::chaininterface::BroadcasterInterface + Send + Sync + 'static,
 /// #     F: lightning::chain::Filter + Send + Sync + 'static,
 /// #     FE: lightning::chain::chaininterface::FeeEstimator + Send + Sync + 'static,
 /// #     UL: lightning::routing::utxo::UtxoLookup + Send + Sync + 'static,
 /// #     D: lightning::sign::ChangeDestinationSource + Send + Sync + 'static,
-/// #     K: lightning::util::persist::KVStore + Send + Sync + 'static,
 /// #     O: lightning::sign::OutputSpender + Send + Sync + 'static,
 /// # > {
 /// #     peer_manager: Arc<PeerManager<B, F, FE, UL>>,
@@ -697,7 +718,7 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
 /// #     persister: Arc<Store>,
 /// #     logger: Arc<Logger>,
 /// #     scorer: Arc<Scorer>,
-/// #     sweeper: Arc<OutputSweeper<Arc<B>, Arc<D>, Arc<FE>, Arc<F>, Arc<K>, Arc<Logger>, Arc<O>>>,
+/// #     sweeper: Arc<OutputSweeper<B, D, FE, F, O>>,
 /// # }
 /// #
 /// # async fn setup_background_processing<
@@ -706,10 +727,9 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
 /// #     FE: lightning::chain::chaininterface::FeeEstimator + Send + Sync + 'static,
 /// #     UL: lightning::routing::utxo::UtxoLookup + Send + Sync + 'static,
 /// #     D: lightning::sign::ChangeDestinationSource + Send + Sync + 'static,
-/// #     K: lightning::util::persist::KVStore + Send + Sync + 'static,
 /// #     O: lightning::sign::OutputSpender + Send + Sync + 'static,
-/// # >(node: Node<B, F, FE, UL, D, K, O>) {
-///	let background_persister = Arc::clone(&node.persister);
+/// # >(node: Node<B, F, FE, UL, D, O>) {
+///	let background_persister = Arc::new(Arc::clone(&node.persister));
 ///	let background_event_handler = Arc::clone(&node.event_handler);
 ///	let background_chain_mon = Arc::clone(&node.chain_monitor);
 ///	let background_chan_man = Arc::clone(&node.channel_manager);
@@ -814,7 +834,7 @@ where
 	F::Target: 'static + FeeEstimator,
 	L::Target: 'static + Logger,
 	P::Target: 'static + Persist<<CM::Target as AChannelManager>::Signer>,
-	PS::Target: 'static + PersisterSync<'a, CM, L, S>,
+	PS::Target: 'static + Persister<'a, CM, L, S>,
 	ES::Target: 'static + EntropySource,
 	CM::Target: AChannelManager,
 	OM::Target: AOnionMessenger,
@@ -841,7 +861,7 @@ where
 				if let Some(duration_since_epoch) = fetch_time() {
 					if update_scorer(scorer, &event, duration_since_epoch) {
 						log_trace!(logger, "Persisting scorer after update");
-						if let Err(e) = persister.persist_scorer(&*scorer) {
+						if let Err(e) = persister.persist_scorer(&*scorer).await {
 							log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e);
 							// We opt not to abort early on persistence failure here as persisting
 							// the scorer is non-critical and we still hope that it will have
@@ -919,7 +939,134 @@ where
 		},
 		mobile_interruptable_platform,
 		fetch_time,
+		true,
+	)
+}
+
+/// A wrapper that turns a synchronous [`PersisterSync`] into an async [`Persister`].
+struct PersisterSyncWrapper<'a, PS, CM, L, S> {
+	inner: Arc<PS>,
+	_phantom: std::marker::PhantomData<(&'a (), CM, L, S)>,
+}
+
+impl<'a, PS, CM, L, S> Deref for PersisterSyncWrapper<'_, PS, CM, L, S> {
+	type Target = Self;
+	fn deref(&self) -> &Self {
+		self
+	}
+}
+
+impl<'a, PS, CM, L, S> PersisterSyncWrapper<'a, PS, CM, L, S> {
+	/// Constructs a new [`PersisterSyncWrapper`] from the given sync persister.
+	pub fn new(inner: PS) -> Self {
+		Self { inner: Arc::new(inner), _phantom: std::marker::PhantomData }
+	}
+}
+
+impl<'a, PS: Deref, CM: Deref + 'static, L: Deref + 'static, S: Deref + 'static>
+	Persister<'a, CM, L, S> for PersisterSyncWrapper<'a, PS, CM, L, S>
+where
+	PS::Target: PersisterSync<'a, CM, L, S>,
+	CM::Target: AChannelManager,
+	L::Target: Logger,
+	S::Target: WriteableScore<'a>,
+{
+	fn persist_manager(
+		&self, channel_manager: &CM,
+	) -> Pin<Box<dyn Future<Output = Result<(), lightning::io::Error>> + 'static + Send>> {
+		let res = self.inner.persist_manager(&channel_manager);
+		Box::pin(async move { res })
+	}
+
+	fn persist_graph(
+		&self, network_graph: &NetworkGraph<L>,
+	) -> Pin<Box<dyn Future<Output = Result<(), lightning::io::Error>> + 'static + Send>> {
+		let res = self.inner.persist_graph(&network_graph);
+		Box::pin(async move { res })
+	}
+
+	fn persist_scorer(
+		&self, scorer: &S,
+	) -> Pin<Box<dyn Future<Output = Result<(), lightning::io::Error>> + 'static + Send>> {
+		let res = self.inner.persist_scorer(&scorer);
+		Box::pin(async move { res })
+	}
+}
+
+/// Async events processor the is based on [`process_events_async`] but allows for [`PersisterSync`] to be used for
+/// synchronous background persistence.
+pub async fn process_events_partial_async<
+	UL: 'static + Deref,
+	CF: 'static + Deref,
+	T: 'static + Deref,
+	F: 'static + Deref,
+	G: 'static + Deref<Target = NetworkGraph<L>>,
+	L: 'static + Deref + Send + Sync,
+	P: 'static + Deref,
+	EventHandlerFuture: core::future::Future<Output = Result<(), ReplayEvent>>,
+	EventHandler: Fn(Event) -> EventHandlerFuture,
+	PS: 'static + Deref + Send + Sync,
+	ES: 'static + Deref + Send,
+	M: 'static
+		+ Deref<Target = ChainMonitor<<CM::Target as AChannelManager>::Signer, CF, T, F, L, P, ES>>
+		+ Send
+		+ Sync,
+	CM: 'static + Deref + Send + Sync,
+	OM: 'static + Deref,
+	PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>>,
+	RGS: 'static + Deref<Target = RapidGossipSync<G, L>>,
+	PM: 'static + Deref,
+	LM: 'static + Deref,
+	D: 'static + Deref,
+	O: 'static + Deref,
+	K: 'static + Deref,
+	OS: 'static + Deref<Target = OutputSweeper<T, D, F, CF, K, L, O>>,
+	S: 'static + Deref<Target = SC> + Send + Sync,
+	SC: for<'b> WriteableScore<'b>,
+	SleepFuture: core::future::Future<Output = bool> + core::marker::Unpin,
+	Sleeper: Fn(Duration) -> SleepFuture,
+	FetchTime: Fn() -> Option<Duration>,
+>(
+	persister: PS, event_handler: EventHandler, chain_monitor: M, channel_manager: CM,
+	onion_messenger: Option<OM>, gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM,
+	liquidity_manager: Option<LM>, sweeper: Option<OS>, logger: L, scorer: Option<S>,
+	sleeper: Sleeper, mobile_interruptable_platform: bool, fetch_time: FetchTime,
+) -> Result<(), lightning::io::Error>
+where
+	UL::Target: 'static + UtxoLookup,
+	CF::Target: 'static + chain::Filter,
+	T::Target: 'static + BroadcasterInterface,
+	F::Target: 'static + FeeEstimator,
+	L::Target: 'static + Logger,
+	P::Target: 'static + Persist<<CM::Target as AChannelManager>::Signer>,
+	PS::Target: 'static + PersisterSync<'static, CM, L, S>,
+	ES::Target: 'static + EntropySource,
+	CM::Target: AChannelManager,
+	OM::Target: AOnionMessenger,
+	PM::Target: APeerManager,
+	LM::Target: ALiquidityManager,
+	O::Target: 'static + OutputSpender,
+	D::Target: 'static + ChangeDestinationSource,
+	K::Target: 'static + KVStoreSync,
+{
+	let persister = PersisterSyncWrapper::<'static, PS, CM, L, S>::new(persister);
+	process_events_async(
+		persister,
+		event_handler,
+		chain_monitor,
+		channel_manager,
+		onion_messenger,
+		gossip_sync,
+		peer_manager,
+		liquidity_manager,
+		sweeper,
+		logger,
+		scorer,
+		sleeper,
+		mobile_interruptable_platform,
+		fetch_time,
 	)
+	.await
 }
 
 #[cfg(feature = "std")]
@@ -1098,6 +1245,7 @@ impl BackgroundProcessor {
 							.expect("Time should be sometime after 1970"),
 					)
 				},
+				false,
 			)
 		});
 		Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
@@ -2111,7 +2259,7 @@ mod tests {
 			PersisterSync::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"),
 		);
 
-		let bp_future = super::process_events_async(
+		let bp_future = super::process_events_partial_async(
 			persister,
 			|_: _| async { Ok(()) },
 			Arc::clone(&nodes[0].chain_monitor),
@@ -2622,7 +2770,7 @@ mod tests {
 			Arc::new(PersisterSync::new(data_dir).with_graph_persistence_notifier(sender));
 
 		let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
-		let bp_future = super::process_events_async(
+		let bp_future = super::process_events_partial_async(
 			persister,
 			|_: _| async { Ok(()) },
 			Arc::clone(&nodes[0].chain_monitor),
@@ -2839,7 +2987,7 @@ mod tests {
 
 		let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
 
-		let bp_future = super::process_events_async(
+		let bp_future = super::process_events_partial_async(
 			persister,
 			event_handler,
 			Arc::clone(&nodes[0].chain_monitor),