Convert BumpTransactionEventHandler to async

Author: joostjager

lightning/src/events/bump_transaction.rs

@@ -30,6 +30,7 @@ use crate::sign::{
 };
 use crate::sign::ecdsa::EcdsaChannelSigner;
 use crate::sync::Mutex;
+use crate::util::async_poll::{AsyncResult, MaybeSync, MaybeSend};
 use crate::util::logger::Logger;
 
 use bitcoin::{OutPoint, Psbt, PubkeyHash, Sequence, ScriptBuf, Transaction, TxIn, TxOut, Witness, WPubkeyHash};
@@ -344,21 +345,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
@@ -373,13 +391,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,
@@ -389,10 +448,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`].
@@ -408,7 +467,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,
@@ -461,7 +520,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) *
@@ -482,23 +541,29 @@ 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()?;
+	) -> 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 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);
@@ -524,16 +589,17 @@ where
 				preexisting_tx_weight,
 				input_amount_sat,
 				target_amount_sat,
-			);
+				).await;
 			if attempt.is_ok() {
 				return attempt;
 			}
 			last_err = Some(attempt);
 		}
 		last_err.unwrap()
+		})
 	}
 
-	fn sign_psbt(&self, psbt: Psbt) -> Result<Transaction, ()> {
+	fn sign_psbt<'a>(&'a self, psbt: Psbt) -> AsyncResult<'a, Transaction> {
 		self.source.sign_psbt(psbt)
 	}
 }
@@ -645,8 +711,11 @@ 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,
-			)?;
+				claim_id,
+				must_spend,
+				&[],
+				package_target_feerate_sat_per_1000_weight,
+			).await?;
 
 			let mut anchor_tx = Transaction {
 				version: Version::TWO,
@@ -699,7 +768,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.derive_channel_signer(anchor_descriptor.channel_derivation_parameters.keys_id);
 			let channel_parameters = &anchor_descriptor.channel_derivation_parameters.transaction_parameters;
@@ -767,8 +836,11 @@ 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,
-		)?;
+			claim_id,
+			must_spend,
+			&htlc_tx.output,
+			target_feerate_sat_per_1000_weight,
+		).await?;
 
 		#[cfg(debug_assertions)]
 		let total_satisfaction_weight = must_spend_satisfaction_weight +
@@ -800,7 +872,7 @@ where
 		let unsigned_tx_weight = htlc_psbt.unsigned_tx.weight().to_wu() - (htlc_psbt.unsigned_tx.input.len() as u64 * EMPTY_SCRIPT_SIG_WEIGHT);
 
 		log_debug!(self.logger, "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() {
@@ -843,11 +915,17 @@ where
 				log_info!(self.logger, "Handling channel close bump (claim_id = {}, commitment_txid = {})",
 					log_bytes!(claim_id.0), commitment_tx.compute_txid());
 				if let Err(_) = self.handle_channel_close(
-					*claim_id, *package_target_feerate_sat_per_1000_weight, commitment_tx,
-					*commitment_tx_fee_satoshis, anchor_descriptor,
+					*claim_id,
+					*package_target_feerate_sat_per_1000_weight,
+					commitment_tx,
+					*commitment_tx_fee_satoshis,
+					anchor_descriptor,
 				).await {
-					log_error!(self.logger, "Failed bumping commitment transaction fee for {}",
-						commitment_tx.compute_txid());
+					log_error!(
+						self.logger,
+						"Failed bumping commitment transaction fee for {}",
+						commitment_tx.compute_txid()
+					);
 				}
 			}
 			BumpTransactionEvent::HTLCResolution {
@@ -856,10 +934,16 @@ where
 				log_info!(self.logger, "Handling HTLC bump (claim_id = {}, htlcs_to_claim = {})",
 					log_bytes!(claim_id.0), log_iter!(htlc_descriptors.iter().map(|d| d.outpoint())));
 				if let Err(_) = self.handle_htlc_resolution(
-					*claim_id, *target_feerate_sat_per_1000_weight, htlc_descriptors, *tx_lock_time,
+					*claim_id,
+					*target_feerate_sat_per_1000_weight,
+					htlc_descriptors,
+					*tx_lock_time,
 				).await {
-					log_error!(self.logger, "Failed bumping HTLC transaction fee for commitment {}",
-						htlc_descriptors[0].commitment_txid);
+					log_error!(
+						self.logger,
+						"Failed bumping HTLC transaction fee for commitment {}",
+						htlc_descriptors[0].commitment_txid
+					);
 				}
 			}
 		}
@@ -886,30 +970,29 @@ 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],
-			target_feerate_sat_per_1000_weight: u32
-		) -> Result<CoinSelection, ()> {
+		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,
+		) -> 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);
+			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::payment::{PaymentPreimage, PaymentHash, PaymentSecret};
 use crate::ln::channelmanager::{AChannelManager, ChainParameters, ChannelManager, ChannelManagerReadArgs, RAACommitmentOrder, RecipientOnionFields, PaymentId, MIN_CLTV_EXPIRY_DELTA};
@@ -470,7 +470,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,
 	>,
@@ -3416,6 +3416,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,
@@ -3429,7 +3430,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

@@ -19,7 +19,7 @@ use crate::chain::channelmonitor::{Balance, ChannelMonitorUpdateStep, CLTV_CLAIM
 use crate::chain::transaction::OutPoint;
 use crate::ln::onion_utils::LocalHTLCFailureReason;
 use crate::sign::{ecdsa::EcdsaChannelSigner, EntropySource, OutputSpender, SignerProvider};
-use crate::events::bump_transaction::WalletSource;
+use crate::events::bump_transaction::WalletSourceSync;
 use crate::events::{Event, FundingInfo, PathFailure, PaymentPurpose, ClosureReason, HTLCHandlingFailureType, PaymentFailureReason};
 use crate::ln::types::ChannelId;
 use crate::types::payment::{PaymentPreimage, PaymentSecret, PaymentHash};
@@ -3138,6 +3138,7 @@ async fn test_multiple_package_conflicts_internal() {
 	let mut tx = nodes[2].tx_broadcaster.txn_broadcast();
 	assert_eq!(tx.len(), 1);
 	let conflict_tx = tx.pop().unwrap();
+
 	assert_eq!(conflict_tx.input.len(), 3);
 	assert_eq!(conflict_tx.input[0].previous_output.txid, node2_commit_tx.compute_txid());
 	assert_eq!(conflict_tx.input[1].previous_output.txid, node2_commit_tx.compute_txid());