f Account for async KVStore and async background persistence

Author: joostjager

src/builder.rs

@@ -1609,7 +1609,7 @@ fn build_with_store_internal(
 		Ok(output_sweeper) => Arc::new(output_sweeper),
 		Err(e) => {
 			if e.kind() == std::io::ErrorKind::NotFound {
-				Arc::new(OutputSweeper::new(
+				Arc::new(OutputSweeper::new_with_kv_store_sync(
 					channel_manager.current_best_block(),
 					Arc::clone(&tx_broadcaster),
 					Arc::clone(&fee_estimator),

src/event.rs

@@ -1062,7 +1062,11 @@ where
 				self.runtime.spawn_cancellable_background_task(future);
 			},
 			LdkEvent::SpendableOutputs { outputs, channel_id } => {
-				match self.output_sweeper.track_spendable_outputs(outputs, channel_id, true, None) {
+				match self
+					.output_sweeper
+					.track_spendable_outputs(outputs, channel_id, true, None)
+					.await
+				{
 					Ok(_) => return Ok(()),
 					Err(_) => {
 						log_error!(self.logger, "Failed to track spendable outputs");

src/io/utils.rs

@@ -252,7 +252,7 @@ pub(crate) fn read_output_sweeper(
 		kv_store,
 		logger.clone(),
 	);
-	OutputSweeper::read(&mut reader, args).map_err(|e| {
+	OutputSweeper::read_with_kv_store_sync(&mut reader, args).map_err(|e| {
 		log_error!(logger, "Failed to deserialize OutputSweeper: {}", e);
 		std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize OutputSweeper")
 	})

src/lib.rs

@@ -158,7 +158,7 @@ use lightning::ln::channelmanager::PaymentId;
 use lightning::ln::msgs::SocketAddress;
 use lightning::routing::gossip::NodeAlias;
 
-use lightning_background_processor::process_events_async;
+use lightning_background_processor::process_events_async_with_kv_store_sync;
 
 use bitcoin::secp256k1::PublicKey;
 
@@ -549,7 +549,7 @@ impl Node {
 		};
 
 		self.runtime.spawn_background_processor_task(async move {
-			process_events_async(
+			process_events_async_with_kv_store_sync(
 				background_persister,
 				|e| background_event_handler.handle_event(e),
 				background_chain_mon,

src/liquidity.rs

@@ -484,7 +484,7 @@ where
 						if token != Some(required) {
 							log_error!(
 								self.logger,
-								"Rejecting LSPS2 request {:?} from counterparty {} as the client provided an invalid token.", 
+								"Rejecting LSPS2 request {:?} from counterparty {} as the client provided an invalid token.",
 								request_id,
 								counterparty_node_id
 							);

src/types.rs

@@ -26,9 +26,10 @@ use lightning::routing::router::DefaultRouter;
 use lightning::routing::scoring::{ProbabilisticScorer, ProbabilisticScoringFeeParameters};
 use lightning::sign::InMemorySigner;
 use lightning::util::persist::KVStoreSync;
+use lightning::util::persist::KVStoreSyncWrapper;
 use lightning::util::ser::{Readable, Writeable, Writer};
-use lightning::util::sweep::OutputSweeper;
 
+use lightning::util::sweep::OutputSweeper;
 use lightning_block_sync::gossip::{GossipVerifier, UtxoSource};
 
 use lightning_net_tokio::SocketDescriptor;
@@ -131,7 +132,7 @@ pub(crate) type Sweeper = OutputSweeper<
 	Arc<KeysManager>,
 	Arc<OnchainFeeEstimator>,
 	Arc<ChainSource>,
-	Arc<DynStore>,
+	KVStoreSyncWrapper<Arc<DynStore>>,
 	Arc<Logger>,
 	Arc<KeysManager>,
 >;