sweeper async wip

Author: joostjager

lightning/src/util/sweep.rs

@@ -411,7 +411,7 @@ where
 	/// Returns `Err` on persistence failure, in which case the call may be safely retried.
 	///
 	/// [`Event::SpendableOutputs`]: crate::events::Event::SpendableOutputs
-	pub fn track_spendable_outputs(
+	pub async fn track_spendable_outputs(
 		&self, output_descriptors: Vec<SpendableOutputDescriptor>, channel_id: Option<ChannelId>,
 		exclude_static_outputs: bool, delay_until_height: Option<u32>,
 	) -> Result<(), ()> {
@@ -444,7 +444,7 @@ where
 
 			state_lock.outputs.push(output_info);
 		}
-		self.persist_state(&*state_lock).map_err(|e| {
+		self.persist_state(&*state_lock).await.map_err(|e| {
 			log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
 		})
 	}
@@ -560,7 +560,7 @@ where
 				output_info.status.broadcast(cur_hash, cur_height, spending_tx.clone());
 			}
 
-			self.persist_state(&sweeper_state).map_err(|e| {
+			self.persist_state(&sweeper_state).await.map_err(|e| {
 				log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
 			})?;
 
@@ -590,14 +590,15 @@ where
 		});
 	}
 
-	fn persist_state(&self, sweeper_state: &SweeperState) -> Result<(), io::Error> {
+	async fn persist_state(&self, sweeper_state: &SweeperState) -> Result<(), io::Error> {
 		self.kv_store
-			.write(
+			.write_async(
 				OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE,
 				OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE,
 				OUTPUT_SWEEPER_PERSISTENCE_KEY,
 				&sweeper_state.encode(),
 			)
+			.await
 			.map_err(|e| {
 				log_error!(
 					self.logger,
@@ -970,16 +971,18 @@ where
 	}
 
 	/// Tells the sweeper to track the given outputs descriptors. Wraps [`OutputSweeper::track_spendable_outputs`].
-	pub fn track_spendable_outputs(
+	pub async fn track_spendable_outputs(
 		&self, output_descriptors: Vec<SpendableOutputDescriptor>, channel_id: Option<ChannelId>,
 		exclude_static_outputs: bool, delay_until_height: Option<u32>,
 	) -> Result<(), ()> {
-		self.sweeper.track_spendable_outputs(
-			output_descriptors,
-			channel_id,
-			exclude_static_outputs,
-			delay_until_height,
-		)
+		self.sweeper
+			.track_spendable_outputs(
+				output_descriptors,
+				channel_id,
+				exclude_static_outputs,
+				delay_until_height,
+			)
+			.await
 	}
 
 	/// Returns a list of the currently tracked spendable outputs. Wraps [`OutputSweeper::tracked_spendable_outputs`].