make async

Author: joostjager

lightning-background-processor/src/lib.rs

@@ -313,7 +313,7 @@ macro_rules! define_run_body {
 		$channel_manager: ident, $process_channel_manager_events: expr,
 		$onion_messenger: ident, $process_onion_message_handler_events: expr,
 		$peer_manager: ident, $gossip_sync: ident,
-		$sweeper: ident,
+		$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,
 	) => { {
@@ -473,7 +473,7 @@ macro_rules! define_run_body {
 
 			if $timer_elapsed(&mut last_sweeper_call, SWEEPER_TIMER) {
 				log_trace!($logger, "Regenerate sweeper spends if necessary");
-				let _ = $sweeper.regenerate_and_broadcast_spend_if_necessary_locked();
+				let _ = $sweeper;
 				last_sweeper_call = $get_timer(SWEEPER_TIMER);
 			}
 		}
@@ -757,6 +757,10 @@ pub async fn process_events_async<
 		+ Sync,
 	CM: 'static + Deref + Send + Sync,
 	OM: 'static + Deref + Send + Sync,
+	D: 'static + Deref + Send + Sync,
+	O: 'static + Deref + Send + Sync,
+	K: 'static + Deref + Send + Sync,
+	OS: 'static + Deref<Target = OutputSweeper<T, D, F, CF, K, L, O>> + Send + Sync,
 	PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
 	RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
 	PM: 'static + Deref + Send + Sync,
@@ -768,12 +772,13 @@ pub async fn process_events_async<
 >(
 	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,
+	sweeper: 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,
+	CF::Target: 'static + chain::Filter + Sync + Send,
 	T::Target: 'static + BroadcasterInterface,
 	F::Target: 'static + FeeEstimator,
 	L::Target: 'static + Logger,
@@ -782,6 +787,9 @@ where
 	CM::Target: AChannelManager + Send + Sync,
 	OM::Target: AOnionMessenger + Send + Sync,
 	PM::Target: APeerManager + Send + Sync,
+	O::Target: 'static + OutputSpender + Send + Sync,
+	D::Target: 'static + ChangeDestinationSource + Send + Sync,
+	K::Target: 'static + KVStore + Send + Sync,
 {
 	let mut should_break = false;
 	let async_event_handler = |event| {
@@ -825,6 +833,7 @@ where
 		},
 		peer_manager,
 		gossip_sync,
+		sweeper.regenerate_and_broadcast_spend_if_necessary_locked().await,
 		logger,
 		scorer,
 		should_break,
@@ -1109,7 +1118,7 @@ mod tests {
 	use lightning::routing::router::{CandidateRouteHop, DefaultRouter, Path, RouteHop};
 	use lightning::routing::scoring::{ChannelUsage, LockableScore, ScoreLookUp, ScoreUpdate};
 	use lightning::routing::utxo::UtxoLookup;
-	use lightning::sign::{ChangeDestinationSource, InMemorySigner, KeysManager};
+	use lightning::sign::{AsyncGetChangeDestinationScriptResult, ChangeDestinationSource, InMemorySigner, KeysManager};
 	use lightning::types::features::{ChannelFeatures, NodeFeatures};
 	use lightning::types::payment::PaymentHash;
 	use lightning::util::config::UserConfig;
@@ -1544,8 +1553,10 @@ mod tests {
 	struct TestWallet {}
 
 	impl ChangeDestinationSource for TestWallet {
-		fn get_change_destination_script(&self) -> Result<ScriptBuf, ()> {
-			Ok(ScriptBuf::new())
+		fn get_change_destination_script<'a>(&self) -> AsyncGetChangeDestinationScriptResult<'a, ScriptBuf> {
+			Box::pin(async move {
+				Ok(ScriptBuf::new())
+			})
 		}
 	}
 
@@ -2027,6 +2038,7 @@ mod tests {
 			Some(nodes[0].messenger.clone()),
 			nodes[0].rapid_gossip_sync(),
 			nodes[0].peer_manager.clone(),
+			nodes[0].sweeper.clone(),
 			nodes[0].logger.clone(),
 			Some(nodes[0].scorer.clone()),
 			move |dur: Duration| {
@@ -2514,6 +2526,7 @@ mod tests {
 			Some(nodes[0].messenger.clone()),
 			nodes[0].rapid_gossip_sync(),
 			nodes[0].peer_manager.clone(),
+			nodes[0].sweeper.clone(),
 			nodes[0].logger.clone(),
 			Some(nodes[0].scorer.clone()),
 			move |dur: Duration| {
@@ -2727,6 +2740,7 @@ mod tests {
 			Some(nodes[0].messenger.clone()),
 			nodes[0].no_gossip_sync(),
 			nodes[0].peer_manager.clone(),
+			nodes[0].sweeper.clone(),
 			nodes[0].logger.clone(),
 			Some(nodes[0].scorer.clone()),
 			move |dur: Duration| {

lightning/src/sign/mod.rs

@@ -67,6 +67,8 @@ use crate::sign::ecdsa::EcdsaChannelSigner;
 use crate::sign::taproot::TaprootChannelSigner;
 use crate::util::atomic_counter::AtomicCounter;
 use core::convert::TryInto;
+use core::future::Future;
+use core::pin::Pin;
 use core::sync::atomic::{AtomicUsize, Ordering};
 #[cfg(taproot)]
 use musig2::types::{PartialSignature, PublicNonce};
@@ -975,6 +977,14 @@ pub trait SignerProvider {
 	fn get_shutdown_scriptpubkey(&self) -> Result<ShutdownScript, ()>;
 }
 
+
+/// Result type for `BlockSource` requests.
+pub type GetChangeDestinationScriptResult<T> = Result<T, ()>;
+
+/// A type alias for a future that returns a [`GetChangeDestinationScriptResult`].
+pub type AsyncGetChangeDestinationScriptResult<'a, T> =
+	Pin<Box<dyn Future<Output = GetChangeDestinationScriptResult<T>> + 'a + Send>>;
+
 /// A helper trait that describes an on-chain wallet capable of returning a (change) destination
 /// script.
 pub trait ChangeDestinationSource {
@@ -983,7 +993,7 @@ pub trait ChangeDestinationSource {
 	///
 	/// This method should return a different value each time it is called, to avoid linking
 	/// on-chain funds controlled to the same user.
-	fn get_change_destination_script(&self) -> Result<ScriptBuf, ()>;
+	fn get_change_destination_script<'a>(&self) -> AsyncGetChangeDestinationScriptResult<'a, ScriptBuf>;
 }
 
 mod sealed {

lightning/src/util/sweep.rs

@@ -450,15 +450,15 @@ where
 	}
 
 	/// Regenerates and broadcasts the spending transaction for any outputs that are pending
-	pub fn regenerate_and_broadcast_spend_if_necessary_locked(&self) -> Result<(), ()> {
+	pub async fn regenerate_and_broadcast_spend_if_necessary_locked(&self) -> Result<(), ()> {
 		let mut sweeper_state = self.sweeper_state.lock().unwrap();
-		self.regenerate_and_broadcast_spend_if_necessary(&mut *sweeper_state)
+		self.regenerate_and_broadcast_spend_if_necessary(&mut *sweeper_state).await
 	}
 
-	fn regenerate_and_broadcast_spend_if_necessary(
+	async fn regenerate_and_broadcast_spend_if_necessary(
 		&self, sweeper_state: &mut SweeperState,
 	) -> Result<(), ()> {
-		let spending_tx_opt = self.regenerate_spend_if_necessary(sweeper_state);
+		let spending_tx_opt = self.regenerate_spend_if_necessary(sweeper_state).await;
 		if let Some(spending_tx) = spending_tx_opt {
 			self.persist_state(&*sweeper_state).map_err(|e| {
 				log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
@@ -470,7 +470,7 @@ where
 		Ok(())
 	}
 
-	fn regenerate_spend_if_necessary(
+	async fn regenerate_spend_if_necessary(
 		&self, sweeper_state: &mut SweeperState,
 	) -> Option<Transaction> {
 		let cur_height = sweeper_state.best_block.height;
@@ -502,7 +502,7 @@ where
 			return None;
 		}
 
-		let spending_tx = match self.spend_outputs(&*sweeper_state, respend_descriptors) {
+		let spending_tx = match self.spend_outputs(&*sweeper_state, respend_descriptors).await {
 			Ok(spending_tx) => {
 				log_debug!(
 					self.logger,
@@ -573,13 +573,13 @@ where
 			})
 	}
 
-	fn spend_outputs(
+	async fn spend_outputs(
 		&self, sweeper_state: &SweeperState, descriptors: Vec<&SpendableOutputDescriptor>,
 	) -> Result<Transaction, ()> {
 		let tx_feerate =
 			self.fee_estimator.get_est_sat_per_1000_weight(ConfirmationTarget::OutputSpendingFee);
 		let change_destination_script =
-			self.change_destination_source.get_change_destination_script()?;
+			self.change_destination_source.get_change_destination_script().await?;
 		let cur_height = sweeper_state.best_block.height;
 		let locktime = Some(LockTime::from_height(cur_height).unwrap_or(LockTime::ZERO));
 		self.output_spender.spend_spendable_outputs(