Convert BumpTransactionEventHandler to async

Changes the CoinSelectionSource and WalletSource traits to be async and
provides WalletSourceSyncWrapper to as a helper for users that want to
implement a sync wallet source.

TestWalletSource is kept sync, to prevent a cascade of async conversions
in tests.

Author: joostjager

lightning/src/events/bump_transaction.rs

@@ -30,6 +30,7 @@ use crate::sign::{
 	ChannelDerivationParameters, HTLCDescriptor, SignerProvider, P2WPKH_WITNESS_WEIGHT,
 };
 use crate::sync::Mutex;
+use crate::util::async_poll::{AsyncResult, MaybeSend, MaybeSync};
 use crate::util::logger::Logger;
 
 use bitcoin::amount::Amount;
@@ -346,21 +347,38 @@ pub trait CoinSelectionSource {
 	/// other claims, implementations must be willing to double spend their UTXOs. The choice of
 	/// which UTXOs to double spend is left to the implementation, but it must strive to keep the
 	/// set of other claims being double spent to a minimum.
-	fn select_confirmed_utxos(
-		&self, claim_id: ClaimId, must_spend: Vec<Input>, must_pay_to: &[TxOut],
+	fn select_confirmed_utxos<'a>(
+		&'a self, claim_id: ClaimId, must_spend: Vec<Input>, must_pay_to: &'a [TxOut],
 		target_feerate_sat_per_1000_weight: u32,
-	) -> Result<CoinSelection, ()>;
+	) -> AsyncResult<'a, CoinSelection>;
 	/// Signs and provides the full witness for all inputs within the transaction known to the
 	/// trait (i.e., any provided via [`CoinSelectionSource::select_confirmed_utxos`]).
 	///
 	/// If your wallet does not support signing PSBTs you can call `psbt.extract_tx()` to get the
 	/// unsigned transaction and then sign it with your wallet.
-	fn sign_psbt(&self, psbt: Psbt) -> Result<Transaction, ()>;
+	fn sign_psbt<'a>(&'a self, psbt: Psbt) -> AsyncResult<'a, Transaction>;
 }
 
 /// An alternative to [`CoinSelectionSource`] that can be implemented and used along [`Wallet`] to
 /// provide a default implementation to [`CoinSelectionSource`].
 pub trait WalletSource {
+	/// Returns all UTXOs, with at least 1 confirmation each, that are available to spend.
+	fn list_confirmed_utxos<'a>(&'a self) -> AsyncResult<'a, Vec<Utxo>>;
+	/// Returns a script to use for change above dust resulting from a successful coin selection
+	/// attempt.
+	fn get_change_script<'a>(&self) -> AsyncResult<'a, ScriptBuf>;
+	/// Signs and provides the full [`TxIn::script_sig`] and [`TxIn::witness`] for all inputs within
+	/// the transaction known to the wallet (i.e., any provided via
+	/// [`WalletSource::list_confirmed_utxos`]).
+	///
+	/// If your wallet does not support signing PSBTs you can call `psbt.extract_tx()` to get the
+	/// unsigned transaction and then sign it with your wallet.
+	fn sign_psbt<'a>(&self, psbt: Psbt) -> AsyncResult<'a, Transaction>;
+}
+
+/// A synchronous version of the [`WalletSource`] trait. Implementations of this trait should be wrapped in
+/// WalletSourceSyncWrapper for it to be used within rust-lightning.
+pub trait WalletSourceSync {
 	/// Returns all UTXOs, with at least 1 confirmation each, that are available to spend.
 	fn list_confirmed_utxos(&self) -> Result<Vec<Utxo>, ()>;
 	/// Returns a script to use for change above dust resulting from a successful coin selection
@@ -375,13 +393,54 @@ pub trait WalletSource {
 	fn sign_psbt(&self, psbt: Psbt) -> Result<Transaction, ()>;
 }
 
+/// A wrapper around [`WalletSourceSync`] to allow for async calls.
+pub struct WalletSourceSyncWrapper<T: Deref>(T)
+where
+	T::Target: WalletSourceSync;
+
+impl<T: Deref> WalletSourceSyncWrapper<T>
+where
+	T::Target: WalletSourceSync,
+{
+	/// Creates a new [`WalletSourceSyncWrapper`].
+	pub fn new(source: T) -> Self {
+		Self(source)
+	}
+}
+impl<T: Deref> WalletSource for WalletSourceSyncWrapper<T>
+where
+	T::Target: WalletSourceSync,
+{
+	/// Returns all UTXOs, with at least 1 confirmation each, that are available to spend. Wraps
+	/// [`WalletSourceSync::list_confirmed_utxos`].
+	fn list_confirmed_utxos<'a>(&'a self) -> AsyncResult<'a, Vec<Utxo>> {
+		let utxos = self.0.list_confirmed_utxos();
+		Box::pin(async move { utxos })
+	}
+
+	/// Returns a script to use for change above dust resulting from a successful coin selection attempt. Wraps
+	/// [`WalletSourceSync::get_change_script`].
+	fn get_change_script<'a>(&self) -> AsyncResult<'a, ScriptBuf> {
+		let script = self.0.get_change_script();
+		Box::pin(async move { script })
+	}
+
+	/// Signs and provides the full [`TxIn::script_sig`] and [`TxIn::witness`] for all inputs within the transaction
+	/// known to the wallet (i.e., any provided via [`WalletSource::list_confirmed_utxos`]). Wraps
+	/// [`WalletSourceSync::sign_psbt`].
+	fn sign_psbt<'a>(&self, psbt: Psbt) -> AsyncResult<'a, Transaction> {
+		let signed_psbt = self.0.sign_psbt(psbt);
+		Box::pin(async move { signed_psbt })
+	}
+}
+
 /// A wrapper over [`WalletSource`] that implements [`CoinSelection`] by preferring UTXOs that would
 /// avoid conflicting double spends. If not enough UTXOs are available to do so, conflicting double
 /// spends may happen.
-pub struct Wallet<W: Deref, L: Deref>
+pub struct Wallet<W: Deref + MaybeSync + MaybeSend, L: Deref + MaybeSync + MaybeSend>
 where
-	W::Target: WalletSource,
-	L::Target: Logger,
+	W::Target: WalletSource + MaybeSend,
+	L::Target: Logger + MaybeSend,
 {
 	source: W,
 	logger: L,
@@ -391,10 +450,10 @@ where
 	locked_utxos: Mutex<HashMap<OutPoint, ClaimId>>,
 }
 
-impl<W: Deref, L: Deref> Wallet<W, L>
+impl<W: Deref + MaybeSync + MaybeSend, L: Deref + MaybeSync + MaybeSend> Wallet<W, L>
 where
-	W::Target: WalletSource,
-	L::Target: Logger,
+	W::Target: WalletSource + MaybeSend,
+	L::Target: Logger + MaybeSend,
 {
 	/// Returns a new instance backed by the given [`WalletSource`] that serves as an implementation
 	/// of [`CoinSelectionSource`].
@@ -410,7 +469,7 @@ where
 	/// `tolerate_high_network_feerates` is set, we'll attempt to spend UTXOs that contribute at
 	/// least 1 satoshi at the current feerate, otherwise, we'll only attempt to spend those which
 	/// contribute at least twice their fee.
-	fn select_confirmed_utxos_internal(
+	async fn select_confirmed_utxos_internal(
 		&self, utxos: &[Utxo], claim_id: ClaimId, force_conflicting_utxo_spend: bool,
 		tolerate_high_network_feerates: bool, target_feerate_sat_per_1000_weight: u32,
 		preexisting_tx_weight: u64, input_amount_sat: Amount, target_amount_sat: Amount,
@@ -484,7 +543,7 @@ where
 		}
 
 		let remaining_amount = selected_amount - target_amount_sat - total_fees;
-		let change_script = self.source.get_change_script()?;
+		let change_script = self.source.get_change_script().await?;
 		let change_output_fee = fee_for_weight(
 			target_feerate_sat_per_1000_weight,
 			(8 /* value */ + change_script.consensus_encode(&mut sink()).unwrap() as u64)
@@ -503,60 +562,67 @@ where
 	}
 }
 
-impl<W: Deref, L: Deref> CoinSelectionSource for Wallet<W, L>
+impl<W: Deref + MaybeSync + MaybeSend, L: Deref + MaybeSync + MaybeSend> CoinSelectionSource
+	for Wallet<W, L>
 where
-	W::Target: WalletSource,
-	L::Target: Logger,
+	W::Target: WalletSource + MaybeSend + MaybeSync,
+	L::Target: Logger + MaybeSend + MaybeSync,
 {
-	fn select_confirmed_utxos(
-		&self, claim_id: ClaimId, must_spend: Vec<Input>, must_pay_to: &[TxOut],
+	fn select_confirmed_utxos<'a>(
+		&'a self, claim_id: ClaimId, must_spend: Vec<Input>, must_pay_to: &'a [TxOut],
 		target_feerate_sat_per_1000_weight: u32,
-	) -> Result<CoinSelection, ()> {
-		let utxos = self.source.list_confirmed_utxos()?;
-		// TODO: Use fee estimation utils when we upgrade to bitcoin v0.30.0.
-		const BASE_TX_SIZE: u64 = 4 /* version */ + 1 /* input count */ + 1 /* output count */ + 4 /* locktime */;
-		let total_output_size: u64 = must_pay_to
-			.iter()
-			.map(|output| 8 /* value */ + 1 /* script len */ + output.script_pubkey.len() as u64)
-			.sum();
-		let total_satisfaction_weight: u64 =
-			must_spend.iter().map(|input| input.satisfaction_weight).sum();
-		let total_input_weight =
-			(BASE_INPUT_WEIGHT * must_spend.len() as u64) + total_satisfaction_weight;
-
-		let preexisting_tx_weight = 2 /* segwit marker & flag */ + total_input_weight +
+	) -> AsyncResult<'a, CoinSelection> {
+		Box::pin(async move {
+			let utxos = self.source.list_confirmed_utxos().await?;
+			// TODO: Use fee estimation utils when we upgrade to bitcoin v0.30.0.
+			const BASE_TX_SIZE: u64 = 4 /* version */ + 1 /* input count */ + 1 /* output count */ + 4 /* locktime */;
+			let total_output_size: u64 = must_pay_to
+				.iter()
+				.map(
+					|output| 8 /* value */ + 1 /* script len */ + output.script_pubkey.len() as u64,
+				)
+				.sum();
+			let total_satisfaction_weight: u64 =
+				must_spend.iter().map(|input| input.satisfaction_weight).sum();
+			let total_input_weight =
+				(BASE_INPUT_WEIGHT * must_spend.len() as u64) + total_satisfaction_weight;
+
+			let preexisting_tx_weight = 2 /* segwit marker & flag */ + total_input_weight +
 			((BASE_TX_SIZE + total_output_size) * WITNESS_SCALE_FACTOR as u64);
-		let input_amount_sat = must_spend.iter().map(|input| input.previous_utxo.value).sum();
-		let target_amount_sat = must_pay_to.iter().map(|output| output.value).sum();
+			let input_amount_sat = must_spend.iter().map(|input| input.previous_utxo.value).sum();
+			let target_amount_sat = must_pay_to.iter().map(|output| output.value).sum();
 
-		let configs = [(false, false), (false, true), (true, false), (true, true)];
-		for (force_conflicting_utxo_spend, tolerate_high_network_feerates) in configs {
-			log_debug!(
+			let configs = [(false, false), (false, true), (true, false), (true, true)];
+			for (force_conflicting_utxo_spend, tolerate_high_network_feerates) in configs {
+				log_debug!(
 				self.logger,
 				"Attempting coin selection targeting {} sat/kW (force_conflicting_utxo_spend = {}, tolerate_high_network_feerates = {})",
 				target_feerate_sat_per_1000_weight,
 				force_conflicting_utxo_spend,
 				tolerate_high_network_feerates
 			);
-			let attempt = self.select_confirmed_utxos_internal(
-				&utxos,
-				claim_id,
-				force_conflicting_utxo_spend,
-				tolerate_high_network_feerates,
-				target_feerate_sat_per_1000_weight,
-				preexisting_tx_weight,
-				input_amount_sat,
-				target_amount_sat,
-			);
-			if attempt.is_ok() {
-				return attempt;
+				let attempt = self
+					.select_confirmed_utxos_internal(
+						&utxos,
+						claim_id,
+						force_conflicting_utxo_spend,
+						tolerate_high_network_feerates,
+						target_feerate_sat_per_1000_weight,
+						preexisting_tx_weight,
+						input_amount_sat,
+						target_amount_sat,
+					)
+					.await;
+				if attempt.is_ok() {
+					return attempt;
+				}
 			}
-		}
-		Err(())
+			Err(())
+		})
 	}
 
-	fn sign_psbt(&self, psbt: Psbt) -> Result<Transaction, ()> {
-		self.source.sign_psbt(psbt)
+	fn sign_psbt<'a>(&'a self, psbt: Psbt) -> AsyncResult<'a, Transaction> {
+		Box::pin(async move { self.source.sign_psbt(psbt).await })
 	}
 }
 
@@ -662,12 +728,15 @@ where
 
 			log_debug!(self.logger, "Performing coin selection for commitment package (commitment and anchor transaction) targeting {} sat/kW",
 				package_target_feerate_sat_per_1000_weight);
-			let coin_selection: CoinSelection = self.utxo_source.select_confirmed_utxos(
-				claim_id,
-				must_spend,
-				&[],
-				package_target_feerate_sat_per_1000_weight,
-			)?;
+			let coin_selection: CoinSelection = self
+				.utxo_source
+				.select_confirmed_utxos(
+					claim_id,
+					must_spend,
+					&[],
+					package_target_feerate_sat_per_1000_weight,
+				)
+				.await?;
 
 			let mut anchor_tx = Transaction {
 				version: Version::TWO,
@@ -733,7 +802,7 @@ where
 			}
 
 			log_debug!(self.logger, "Signing anchor transaction {}", anchor_txid);
-			anchor_tx = self.utxo_source.sign_psbt(anchor_psbt)?;
+			anchor_tx = self.utxo_source.sign_psbt(anchor_psbt).await?;
 
 			let signer = self
 				.signer_provider
@@ -821,12 +890,15 @@ where
 		let must_spend_amount =
 			must_spend.iter().map(|input| input.previous_utxo.value.to_sat()).sum::<u64>();
 
-		let coin_selection: CoinSelection = self.utxo_source.select_confirmed_utxos(
-			claim_id,
-			must_spend,
-			&htlc_tx.output,
-			target_feerate_sat_per_1000_weight,
-		)?;
+		let coin_selection: CoinSelection = self
+			.utxo_source
+			.select_confirmed_utxos(
+				claim_id,
+				must_spend,
+				&htlc_tx.output,
+				target_feerate_sat_per_1000_weight,
+			)
+			.await?;
 
 		#[cfg(debug_assertions)]
 		let input_satisfaction_weight: u64 =
@@ -870,7 +942,7 @@ where
 			"Signing HTLC transaction {}",
 			htlc_psbt.unsigned_tx.compute_txid()
 		);
-		htlc_tx = self.utxo_source.sign_psbt(htlc_psbt)?;
+		htlc_tx = self.utxo_source.sign_psbt(htlc_psbt).await?;
 
 		let mut signers = BTreeMap::new();
 		for (idx, htlc_descriptor) in htlc_descriptors.iter().enumerate() {
@@ -993,27 +1065,27 @@ mod tests {
 		expected_selects: Mutex<Vec<(u64, u64, u32, CoinSelection)>>,
 	}
 	impl CoinSelectionSource for TestCoinSelectionSource {
-		fn select_confirmed_utxos(
-			&self, _claim_id: ClaimId, must_spend: Vec<Input>, _must_pay_to: &[TxOut],
+		fn select_confirmed_utxos<'a>(
+			&'a self, _claim_id: ClaimId, must_spend: Vec<Input>, _must_pay_to: &'a [TxOut],
 			target_feerate_sat_per_1000_weight: u32,
-		) -> Result<CoinSelection, ()> {
+		) -> AsyncResult<'a, CoinSelection> {
 			let mut expected_selects = self.expected_selects.lock().unwrap();
 			let (weight, value, feerate, res) = expected_selects.remove(0);
 			assert_eq!(must_spend.len(), 1);
 			assert_eq!(must_spend[0].satisfaction_weight, weight);
 			assert_eq!(must_spend[0].previous_utxo.value.to_sat(), value);
 			assert_eq!(target_feerate_sat_per_1000_weight, feerate);
-			Ok(res)
+			Box::pin(async move { Ok(res) })
 		}
-		fn sign_psbt(&self, psbt: Psbt) -> Result<Transaction, ()> {
+		fn sign_psbt<'a>(&'a self, psbt: Psbt) -> AsyncResult<'a, Transaction> {
 			let mut tx = psbt.unsigned_tx;
 			for input in tx.input.iter_mut() {
 				if input.previous_output.txid != Txid::from_byte_array([44; 32]) {
 					// Channel output, add a realistic size witness to make the assertions happy
 					input.witness = Witness::from_slice(&[vec![42; 162]]);
 				}
 			}
-			Ok(tx)
+			Box::pin(async move { Ok(tx) })
 		}
 	}
 

lightning/src/ln/async_signer_tests.rs

@@ -18,7 +18,7 @@ use bitcoin::transaction::Version;
 
 use crate::chain::channelmonitor::LATENCY_GRACE_PERIOD_BLOCKS;
 use crate::chain::ChannelMonitorUpdateStatus;
-use crate::events::bump_transaction::WalletSource;
+use crate::events::bump_transaction::WalletSourceSync;
 use crate::events::{ClosureReason, Event};
 use crate::ln::chan_utils::ClosingTransaction;
 use crate::ln::channel::DISCONNECT_PEER_AWAITING_RESPONSE_TICKS;

lightning/src/ln/functional_test_utils.rs

@@ -14,7 +14,7 @@ use crate::chain::{BestBlock, ChannelMonitorUpdateStatus, Confirm, Listen, Watch
 use crate::chain::channelmonitor::ChannelMonitor;
 use crate::chain::transaction::OutPoint;
 use crate::events::{ClaimedHTLC, ClosureReason, Event, HTLCHandlingFailureType, PaidBolt12Invoice, PathFailure, PaymentFailureReason, PaymentPurpose};
-use crate::events::bump_transaction::{BumpTransactionEvent, BumpTransactionEventHandler, Wallet, WalletSource};
+use crate::events::bump_transaction::{BumpTransactionEvent, BumpTransactionEventHandler, Wallet, WalletSourceSync, WalletSourceSyncWrapper};
 use crate::ln::types::ChannelId;
 use crate::types::features::ChannelTypeFeatures;
 use crate::types::payment::{PaymentPreimage, PaymentHash, PaymentSecret};
@@ -472,7 +472,7 @@ pub struct Node<'chan_man, 'node_cfg: 'chan_man, 'chan_mon_cfg: 'node_cfg> {
 	pub wallet_source: Arc<test_utils::TestWalletSource>,
 	pub bump_tx_handler: BumpTransactionEventHandler<
 		&'chan_mon_cfg test_utils::TestBroadcaster,
-		Arc<Wallet<Arc<test_utils::TestWalletSource>, &'chan_mon_cfg test_utils::TestLogger>>,
+		Arc<Wallet<Arc<WalletSourceSyncWrapper<Arc<test_utils::TestWalletSource>>>, &'chan_mon_cfg test_utils::TestLogger>>,
 		&'chan_mon_cfg test_utils::TestKeysInterface,
 		&'chan_mon_cfg test_utils::TestLogger,
 	>,
@@ -3422,6 +3422,7 @@ pub fn create_network<'a, 'b: 'a, 'c: 'b>(node_count: usize, cfgs: &'b Vec<NodeC
 		);
 		let gossip_sync = P2PGossipSync::new(cfgs[i].network_graph.as_ref(), None, cfgs[i].logger);
 		let wallet_source = Arc::new(test_utils::TestWalletSource::new(SecretKey::from_slice(&[i as u8 + 1; 32]).unwrap()));
+		let wallet_source_async = Arc::new(WalletSourceSyncWrapper::new(wallet_source.clone()));
 		nodes.push(Node{
 			chain_source: cfgs[i].chain_source, tx_broadcaster: cfgs[i].tx_broadcaster,
 			fee_estimator: cfgs[i].fee_estimator, router: &cfgs[i].router,
@@ -3435,7 +3436,7 @@ pub fn create_network<'a, 'b: 'a, 'c: 'b>(node_count: usize, cfgs: &'b Vec<NodeC
 			override_init_features: Rc::clone(&cfgs[i].override_init_features),
 			wallet_source: Arc::clone(&wallet_source),
 			bump_tx_handler: BumpTransactionEventHandler::new(
-				cfgs[i].tx_broadcaster, Arc::new(Wallet::new(Arc::clone(&wallet_source), cfgs[i].logger)),
+				cfgs[i].tx_broadcaster, Arc::new(Wallet::new(wallet_source_async, cfgs[i].logger)),
 				&cfgs[i].keys_manager, cfgs[i].logger,
 			),
 		})

lightning/src/ln/functional_tests.rs

@@ -20,7 +20,7 @@ use crate::chain::channelmonitor::{
 };
 use crate::chain::transaction::OutPoint;
 use crate::chain::{ChannelMonitorUpdateStatus, Confirm, Listen, Watch};
-use crate::events::bump_transaction::WalletSource;
+use crate::events::bump_transaction::WalletSourceSync;
 use crate::events::{
 	ClosureReason, Event, FundingInfo, HTLCHandlingFailureType, PathFailure, PaymentFailureReason,
 	PaymentPurpose,