Convert BumpTransactionEventHandler to async

Author: joostjager

lightning/Cargo.toml

@@ -54,6 +54,7 @@ regex = "1.5.6"
 lightning-types = { version = "0.3.0", path = "../lightning-types", features = ["_test_utils"] }
 lightning-macros = { path = "../lightning-macros" }
 parking_lot = { version = "0.12", default-features = false }
+tokio = { version = "1.35", features = [ "macros", "rt" ] }
 
 [dev-dependencies.bitcoin]
 version = "0.32.2"

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;
 use crate::util::logger::Logger;
 
 use bitcoin::{OutPoint, Psbt, PubkeyHash, Sequence, ScriptBuf, Transaction, TxIn, TxOut, Witness, WPubkeyHash};
@@ -603,7 +604,7 @@ where
 	/// Handles a [`BumpTransactionEvent::ChannelClose`] event variant by producing a fully-signed
 	/// transaction spending an anchor output of the commitment transaction to bump its fee and
 	/// broadcasts them to the network as a package.
-	fn handle_channel_close(
+	async fn handle_channel_close(
 		&self, claim_id: ClaimId, package_target_feerate_sat_per_1000_weight: u32,
 		commitment_tx: &Transaction, commitment_tx_fee_sat: u64, anchor_descriptor: &AnchorDescriptor,
 	) -> Result<(), ()> {
@@ -714,7 +715,7 @@ where
 
 	/// Handles a [`BumpTransactionEvent::HTLCResolution`] event variant by producing a
 	/// fully-signed, fee-bumped HTLC transaction that is broadcast to the network.
-	fn handle_htlc_resolution(
+	async fn handle_htlc_resolution(
 		&self, claim_id: ClaimId, target_feerate_sat_per_1000_weight: u32,
 		htlc_descriptors: &[HTLCDescriptor], tx_lock_time: LockTime,
 	) -> Result<(), ()> {
@@ -818,7 +819,7 @@ where
 	}
 
 	/// Handles all variants of [`BumpTransactionEvent`].
-	pub fn handle_event(&self, event: &BumpTransactionEvent) {
+	pub async fn handle_event(&self, event: &BumpTransactionEvent) {
 		match event {
 			BumpTransactionEvent::ChannelClose {
 				claim_id, package_target_feerate_sat_per_1000_weight, commitment_tx,
@@ -829,7 +830,7 @@ where
 				if let Err(_) = self.handle_channel_close(
 					*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());
 				}
@@ -841,7 +842,7 @@ where
 					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,
-				) {
+				).await {
 					log_error!(self.logger, "Failed bumping HTLC transaction fee for commitment {}",
 						htlc_descriptors[0].commitment_txid);
 				}
@@ -903,8 +904,8 @@ mod tests {
 		}
 	}
 
-	#[test]
-	fn test_op_return_under_funds() {
+	#[tokio::test]
+	async fn test_op_return_under_funds() {
 		// Test what happens if we have to select coins but the anchor output value itself suffices
 		// to pay the required fee.
 		//
@@ -964,6 +965,6 @@ mod tests {
 				outpoint: OutPoint { txid: Txid::from_byte_array([42; 32]), vout: 0 },
 			},
 			pending_htlcs: Vec::new(),
-		});
+		}).await;
 	}
 }

lightning/src/ln/async_signer_tests.rs

@@ -805,7 +805,7 @@ fn do_test_async_commitment_signature_ordering(monitor_update_failure: bool) {
 	claim_payment(&nodes[0], &[&nodes[1]], payment_preimage_2);
 }
 
-fn do_test_async_holder_signatures(anchors: bool, remote_commitment: bool) {
+async fn do_test_async_holder_signatures(anchors: bool, remote_commitment: bool) {
 	// Ensures that we can obtain holder signatures for commitment and HTLC transactions
 	// asynchronously by allowing their retrieval to fail and retrying via
 	// `ChannelMonitor::signer_unblocked`.
@@ -909,7 +909,7 @@ fn do_test_async_holder_signatures(anchors: bool, remote_commitment: bool) {
 
 	// No HTLC transaction should be broadcast as the signer is not available yet.
 	if anchors && !remote_commitment {
-		handle_bump_htlc_event(&nodes[0], 1);
+		handle_bump_htlc_event(&nodes[0], 1).await;
 	}
 	let txn = nodes[0].tx_broadcaster.txn_broadcast();
 	assert!(txn.is_empty(), "expected no transaction to be broadcast, got {:?}", txn);
@@ -920,7 +920,7 @@ fn do_test_async_holder_signatures(anchors: bool, remote_commitment: bool) {
 	get_monitor!(nodes[0], chan_id).signer_unblocked(nodes[0].tx_broadcaster, nodes[0].fee_estimator, &nodes[0].logger);
 
 	if anchors && !remote_commitment {
-		handle_bump_htlc_event(&nodes[0], 1);
+		handle_bump_htlc_event(&nodes[0], 1).await;
 	}
 	{
 		let txn = nodes[0].tx_broadcaster.txn_broadcast();
@@ -929,24 +929,24 @@ fn do_test_async_holder_signatures(anchors: bool, remote_commitment: bool) {
 	}
 }
 
-#[test]
-fn test_async_holder_signatures_no_anchors() {
-	do_test_async_holder_signatures(false, false);
+#[tokio::test]
+async fn test_async_holder_signatures_no_anchors() {
+	do_test_async_holder_signatures(false, false).await;
 }
 
-#[test]
-fn test_async_holder_signatures_remote_commitment_no_anchors() {
-	do_test_async_holder_signatures(false, true);
+#[tokio::test]
+async fn test_async_holder_signatures_remote_commitment_no_anchors() {
+	do_test_async_holder_signatures(false, true).await;
 }
 
-#[test]
-fn test_async_holder_signatures_anchors() {
-	do_test_async_holder_signatures(true, false);
+#[tokio::test]
+async fn test_async_holder_signatures_anchors() {
+	do_test_async_holder_signatures(true, false).await;
 }
 
-#[test]
-fn test_async_holder_signatures_remote_commitment_anchors() {
-	do_test_async_holder_signatures(true, true);
+#[tokio::test]
+async fn test_async_holder_signatures_remote_commitment_anchors() {
+	do_test_async_holder_signatures(true, true).await;
 }
 
 #[test]

lightning/src/ln/functional_test_utils.rs

@@ -1840,15 +1840,15 @@ macro_rules! check_closed_event {
 	}
 }
 
-pub fn handle_bump_htlc_event(node: &Node, count: usize) {
+pub async fn handle_bump_htlc_event<'a, 'b, 'c>(node: &Node<'a, 'b, 'c>, count: usize) {
 	let events = node.chain_monitor.chain_monitor.get_and_clear_pending_events();
 	assert_eq!(events.len(), count);
 	for event in events {
 		match event {
 			Event::BumpTransaction(bump_event) => {
 				if let BumpTransactionEvent::HTLCResolution { .. } = &bump_event {}
 				else { panic!(); }
-				node.bump_tx_handler.handle_event(&bump_event);
+				node.bump_tx_handler.handle_event(&bump_event).await;
 			},
 			_ => panic!(),
 		}

lightning/src/ln/functional_tests.rs

@@ -2958,6 +2958,26 @@ pub fn claim_htlc_outputs() {
 // This is a regression test for https://github.com/lightningdevkit/rust-lightning/issues/3537.
 #[xtest(feature = "_externalize_tests")]
 pub fn test_multiple_package_conflicts() {
+	// Wanted to use #[tokio::test] for this, but that macro adds a test attribute and then the external test runner
+	// can't find the test anymore.
+	let body = async {
+		test_multiple_package_conflicts_internal().await;
+	};
+	let mut body = body;
+	#[allow(unused_mut)]
+	let mut body = unsafe { tokio::macros::support::Pin::new_unchecked(&mut body) };
+	let body: ::core::pin::Pin<&mut dyn ::core::future::Future<Output = ()>> = body;
+	#[allow(clippy::expect_used, clippy::diverging_sub_expression)]
+	{
+		return tokio::runtime::Builder::new_current_thread()
+			.enable_all()
+			.build()
+			.expect("Failed building the Runtime")
+			.block_on(body);
+	}
+}
+
+async fn test_multiple_package_conflicts_internal() {
 	let chanmon_cfgs = create_chanmon_cfgs(3);
 	let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
 	let mut user_cfg = test_default_channel_config();
@@ -3106,21 +3126,18 @@ pub fn test_multiple_package_conflicts() {
 	check_closed_broadcast!(nodes[2], true);
 	check_added_monitors(&nodes[2], 1);
 
-	let process_bump_event = |node: &Node| {
-		let events = node.chain_monitor.chain_monitor.get_and_clear_pending_events();
-		assert_eq!(events.len(), 1);
-		let bump_event = match &events[0] {
-			Event::BumpTransaction(bump_event) => bump_event,
-			_ => panic!("Unexepected event"),
-		};
-		node.bump_tx_handler.handle_event(bump_event);
-
-		let mut tx = node.tx_broadcaster.txn_broadcast();
-		assert_eq!(tx.len(), 1);
-		tx.pop().unwrap()
+	let events = nodes[2].chain_monitor.chain_monitor.get_and_clear_pending_events();
+	assert_eq!(events.len(), 1);
+	let bump_event = match &events[0] {
+		Event::BumpTransaction(bump_event) => bump_event,
+		_ => panic!("Unexpected event"),
 	};
+	nodes[2].bump_tx_handler.handle_event(bump_event).await;
+
+	let mut tx = nodes[2].tx_broadcaster.txn_broadcast();
+	assert_eq!(tx.len(), 1);
+	let conflict_tx = tx.pop().unwrap();
 
-	let conflict_tx = process_bump_event(&nodes[2]);
 	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());