Support async signing of interactive-tx initial commitment signatures

This commit allows for an async signer to immediately return upon a call
to `EcdsaChannelSigner::sign_counterparty_commitment` for the initial
commitment signatures of an interactively funded transaction, such that
they can call back in via `ChannelManager::signer_unblocked` once the
signatures are ready. This is done for both splices and dual-funded
channels, though note that the latter still require more work to be
integrated. Since `tx_signatures` must be sent only after exchanging
`commitment_signed`, we make sure to hold them back if they're ready to
be sent until our `commitment_signed` is also ready.

Author: wpaulino

lightning/src/ln/async_signer_tests.rs

@@ -10,9 +10,13 @@
 //! Tests for asynchronous signing. These tests verify that the channel state machine behaves
 //! properly with a signer implementation that asynchronously derives signatures.
 
+use crate::events::bump_transaction::sync::WalletSourceSync;
+use crate::ln::funding::SpliceContribution;
+use crate::ln::splicing_tests::negotiate_splice_tx;
 use crate::prelude::*;
 use crate::util::ser::Writeable;
 use bitcoin::secp256k1::Secp256k1;
+use bitcoin::{Amount, TxOut};
 
 use crate::chain::channelmonitor::LATENCY_GRACE_PERIOD_BLOCKS;
 use crate::chain::ChannelMonitorUpdateStatus;
@@ -1549,3 +1553,105 @@ fn test_async_force_close_on_invalid_secret_for_stale_state() {
 	check_closed_broadcast(&nodes[1], 1, true);
 	check_closed_event(&nodes[1], 1, closure_reason, &[node_id_0], 100_000);
 }
+
+#[test]
+fn test_async_splice_initial_commit_sig() {
+	let chanmon_cfgs = create_chanmon_cfgs(2);
+	let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
+	let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
+	let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+
+	let channel_id = create_announced_chan_between_nodes(&nodes, 0, 1).2;
+	send_payment(&nodes[0], &[&nodes[1]], 1_000);
+
+	let (initiator, acceptor) = (&nodes[0], &nodes[1]);
+	let initiator_node_id = initiator.node.get_our_node_id();
+	let acceptor_node_id = acceptor.node.get_our_node_id();
+
+	initiator.disable_channel_signer_op(
+		&acceptor_node_id,
+		&channel_id,
+		SignerOp::SignCounterpartyCommitment,
+	);
+	acceptor.disable_channel_signer_op(
+		&initiator_node_id,
+		&channel_id,
+		SignerOp::SignCounterpartyCommitment,
+	);
+
+	// Negotiate a splice up until the signature exchange.
+	let contribution = SpliceContribution::splice_out(vec![TxOut {
+		value: Amount::from_sat(1_000),
+		script_pubkey: nodes[0].wallet_source.get_change_script().unwrap(),
+	}]);
+	negotiate_splice_tx(initiator, acceptor, channel_id, contribution);
+
+	assert!(initiator.node.get_and_clear_pending_msg_events().is_empty());
+	assert!(acceptor.node.get_and_clear_pending_msg_events().is_empty());
+
+	// Have the initiator sign the funding transaction. We won't see their initial commitment signed
+	// go out until their signer returns.
+	let event = get_event!(initiator, Event::FundingTransactionReadyForSigning);
+	if let Event::FundingTransactionReadyForSigning { unsigned_transaction, .. } = event {
+		let partially_signed_tx = initiator.wallet_source.sign_tx(unsigned_transaction).unwrap();
+		initiator
+			.node
+			.funding_transaction_signed(&channel_id, &acceptor_node_id, partially_signed_tx)
+			.unwrap();
+	}
+
+	assert!(initiator.node.get_and_clear_pending_msg_events().is_empty());
+	assert!(acceptor.node.get_and_clear_pending_msg_events().is_empty());
+
+	initiator.enable_channel_signer_op(
+		&acceptor_node_id,
+		&channel_id,
+		SignerOp::SignCounterpartyCommitment,
+	);
+	initiator.node.signer_unblocked(None);
+
+	// Have the acceptor process the message. They should be able to send their `tx_signatures` as
+	// they go first, but it is held back as their initial `commitment_signed` is not ready yet.
+	let initiator_commit_sig = get_htlc_update_msgs(initiator, &acceptor_node_id);
+	acceptor
+		.node
+		.handle_commitment_signed(initiator_node_id, &initiator_commit_sig.commitment_signed[0]);
+	check_added_monitors(acceptor, 1);
+	assert!(acceptor.node.get_and_clear_pending_msg_events().is_empty());
+
+	// Reestablish the channel to make sure the acceptor doesn't attempt to retransmit any messages
+	// that are not ready yet.
+	initiator.node.peer_disconnected(acceptor_node_id);
+	acceptor.node.peer_disconnected(initiator_node_id);
+	reconnect_nodes(ReconnectArgs::new(initiator, acceptor));
+
+	// Re-enable the acceptor's signer. We should see both their initial `commitment_signed` and
+	// `tx_signatures` go out.
+	acceptor.enable_channel_signer_op(
+		&initiator_node_id,
+		&channel_id,
+		SignerOp::SignCounterpartyCommitment,
+	);
+	acceptor.node.signer_unblocked(None);
+
+	let msg_events = acceptor.node.get_and_clear_pending_msg_events();
+	assert_eq!(msg_events.len(), 2, "{msg_events:?}");
+	if let MessageSendEvent::UpdateHTLCs { updates, .. } = &msg_events[0] {
+		initiator.node.handle_commitment_signed(acceptor_node_id, &updates.commitment_signed[0]);
+		check_added_monitors(initiator, 1);
+	} else {
+		panic!("Unexpected event");
+	}
+	if let MessageSendEvent::SendTxSignatures { msg, .. } = &msg_events[1] {
+		initiator.node.handle_tx_signatures(acceptor_node_id, &msg);
+	} else {
+		panic!("Unexpected event");
+	}
+
+	let tx_signatures =
+		get_event_msg!(initiator, MessageSendEvent::SendTxSignatures, acceptor_node_id);
+	acceptor.node.handle_tx_signatures(initiator_node_id, &tx_signatures);
+
+	let _ = get_event!(initiator, Event::SplicePending);
+	let _ = get_event!(acceptor, Event::SplicePending);
+}

lightning/src/ln/channel.rs

@@ -1171,6 +1171,8 @@ pub(super) struct SignerResumeUpdates {
 	pub accept_channel: Option<msgs::AcceptChannel>,
 	pub funding_created: Option<msgs::FundingCreated>,
 	pub funding_signed: Option<msgs::FundingSigned>,
+	pub funding_commit_sig: Option<msgs::CommitmentSigned>,
+	pub tx_signatures: Option<msgs::TxSignatures>,
 	pub channel_ready: Option<msgs::ChannelReady>,
 	pub order: RAACommitmentOrder,
 	pub closing_signed: Option<msgs::ClosingSigned>,
@@ -1634,6 +1636,8 @@ where
 					accept_channel: None,
 					funding_created,
 					funding_signed: None,
+					funding_commit_sig: None,
+					tx_signatures: None,
 					channel_ready: None,
 					order: chan.context.resend_order.clone(),
 					closing_signed: None,
@@ -1650,6 +1654,8 @@ where
 					accept_channel,
 					funding_created: None,
 					funding_signed: None,
+					funding_commit_sig: None,
+					tx_signatures: None,
 					channel_ready: None,
 					order: chan.context.resend_order.clone(),
 					closing_signed: None,
@@ -6501,7 +6507,7 @@ where
 	}
 
 	fn get_initial_commitment_signed_v2<L: Deref>(
-		&self, funding: &FundingScope, logger: &L,
+		&mut self, funding: &FundingScope, logger: &L,
 	) -> Option<msgs::CommitmentSigned>
 	where
 		SP::Target: SignerProvider,
@@ -6514,6 +6520,7 @@ where
 				// We shouldn't expect any HTLCs before `ChannelReady`.
 				debug_assert!(htlc_signatures.is_empty());
 			}
+			self.signer_pending_funding = false;
 			Some(msgs::CommitmentSigned {
 				channel_id: self.channel_id,
 				htlc_signatures,
@@ -6523,7 +6530,11 @@ where
 				partial_signature_with_nonce: None,
 			})
 		} else {
-			// TODO(splicing): Support async signing
+			log_debug!(
+				logger,
+				"Initial counterparty commitment signature not available, waiting on async signer"
+			);
+			self.signer_pending_funding = true;
 			None
 		}
 	}
@@ -9581,6 +9592,11 @@ where
 			// We want to clear that the monitor update for our `tx_signatures` has completed, but
 			// we may still need to hold back the message until it's ready to be sent.
 			self.context.monitor_pending_tx_signatures = false;
+
+			if self.context.signer_pending_funding {
+				tx_signatures.take();
+			}
+
 			let signing_session = self.context.interactive_tx_signing_session.as_ref()
 				.expect("We have a tx_signatures message so we must have a valid signing session");
 			if !signing_session.holder_sends_tx_signatures_first()
@@ -9758,7 +9774,12 @@ where
 			log_trace!(logger, "Attempting to update holder per-commitment point...");
 			self.holder_commitment_point.try_resolve_pending(&self.context.holder_signer, &self.context.secp_ctx, logger);
 		}
-		let funding_signed = if self.context.signer_pending_funding && !self.funding.is_outbound() {
+
+		let funding_signed = if self.context.signer_pending_funding
+			&& !self.is_v2_established()
+			&& !self.funding.is_outbound()
+			&& self.pending_splice.is_none()
+		{
 			let commitment_data = self.context.build_commitment_transaction(&self.funding,
 				// The previous transaction number (i.e., when adding 1) is used because this field
 				// is advanced when handling funding_created, but the point is not advanced until
@@ -9768,6 +9789,43 @@ where
 			let counterparty_initial_commitment_tx = commitment_data.tx;
 			self.context.get_funding_signed_msg(&self.funding.channel_transaction_parameters, logger, counterparty_initial_commitment_tx)
 		} else { None };
+
+		let funding_commit_sig = if self.context.signer_pending_funding
+			&& (self.is_v2_established() || self.pending_splice.is_some())
+		{
+			log_debug!(logger, "Attempting to generate pending initial commitment_signed...");
+			let funding = self
+				.pending_splice
+				.as_ref()
+				.and_then(|pending_splice| pending_splice.funding_negotiation.as_ref())
+				.and_then(|funding_negotiation| {
+					debug_assert!(matches!(
+							funding_negotiation,
+							FundingNegotiation::AwaitingSignatures { .. }
+					));
+					funding_negotiation.as_funding()
+				})
+			.unwrap_or(&self.funding);
+			self.context.get_initial_commitment_signed_v2(funding, logger)
+		} else {
+			None
+		};
+
+		let tx_signatures = if funding_commit_sig.is_some() {
+			if let Some(signing_session) = self.context.interactive_tx_signing_session.as_ref() {
+				let should_send_tx_signatures = signing_session.holder_sends_tx_signatures_first()
+					|| signing_session.has_received_tx_signatures();
+				should_send_tx_signatures
+					.then(|| ())
+					.and_then(|_| signing_session.holder_tx_signatures().clone())
+			} else {
+				debug_assert!(false);
+				None
+			}
+		} else {
+			None
+		};
+
 		// Provide a `channel_ready` message if we need to, but only if we're _not_ still pending
 		// funding.
 		let channel_ready = if self.context.signer_pending_channel_ready && !self.context.signer_pending_funding {
@@ -9826,12 +9884,14 @@ where
 			} else { (None, None, None) }
 		} else { (None, None, None) };
 
-		log_trace!(logger, "Signer unblocked with {} commitment_update, {} revoke_and_ack, with resend order {:?}, {} funding_signed, {} channel_ready,
-				{} closing_signed, {} signed_closing_tx, and {} shutdown result",
+		log_trace!(logger, "Signer unblocked with {} commitment_update, {} revoke_and_ack, with resend order {:?}, {} funding_signed, \
+			{} funding commit_sig, {} tx_signatures, {} channel_ready, {} closing_signed, {} signed_closing_tx, and {} shutdown result",
 			if commitment_update.is_some() { "a" } else { "no" },
 			if revoke_and_ack.is_some() { "a" } else { "no" },
 			self.context.resend_order,
 			if funding_signed.is_some() { "a" } else { "no" },
+			if funding_commit_sig.is_some() { "a" } else { "no" },
+			if tx_signatures.is_some() { "a" } else { "no" },
 			if channel_ready.is_some() { "a" } else { "no" },
 			if closing_signed.is_some() { "a" } else { "no" },
 			if signed_closing_tx.is_some() { "a" } else { "no" },
@@ -9844,6 +9904,8 @@ where
 			accept_channel: None,
 			funding_created: None,
 			funding_signed,
+			funding_commit_sig,
+			tx_signatures,
 			channel_ready,
 			order: self.context.resend_order.clone(),
 			closing_signed,
@@ -10150,6 +10212,7 @@ where
 
 		// A receiving node:
 		//   - if the `next_funding` TLV is set:
+		let mut retransmit_funding_commit_sig = None;
 		if let Some(next_funding) = &msg.next_funding {
 			// - if `next_funding_txid` matches the latest interactive funding transaction
 			//   or the current channel funding transaction:
@@ -10172,49 +10235,7 @@ where
 					&& next_funding.should_retransmit(msgs::NextFundingFlag::CommitmentSigned)
 				{
 					// - MUST retransmit its `commitment_signed` for that funding transaction.
-					let funding = self
-						.pending_splice
-						.as_ref()
-						.and_then(|pending_splice| pending_splice.funding_negotiation.as_ref())
-						.and_then(|funding_negotiation| {
-							if let FundingNegotiation::AwaitingSignatures { funding, .. } = &funding_negotiation {
-								Some(funding)
-							} else {
-								None
-							}
-						})
-						.or_else(|| Some(&self.funding))
-						.filter(|funding| funding.get_funding_txid() == Some(next_funding.txid))
-						.ok_or_else(|| {
-							let message = "Failed to find funding for new commitment_signed".to_owned();
-							ChannelError::Close(
-								(
-									message.clone(),
-									ClosureReason::HolderForceClosed { message, broadcasted_latest_txn: Some(false) },
-								)
-							)
-						})?;
-
-					let commitment_signed = self.context.get_initial_commitment_signed_v2(&funding, logger)
-						// TODO(splicing): Support async signing
-						.ok_or_else(|| {
-							let message = "Failed to get signatures for new commitment_signed".to_owned();
-							ChannelError::Close(
-								(
-									message.clone(),
-									ClosureReason::HolderForceClosed { message, broadcasted_latest_txn: Some(false) },
-								)
-							)
-						})?;
-
-					commitment_update = Some(msgs::CommitmentUpdate {
-						commitment_signed: vec![commitment_signed],
-						update_add_htlcs: vec![],
-						update_fulfill_htlcs: vec![],
-						update_fail_htlcs: vec![],
-						update_fail_malformed_htlcs: vec![],
-						update_fee: None,
-					});
+					retransmit_funding_commit_sig = Some(next_funding.txid);
 				}
 
 				// - if it has already received `commitment_signed` and it should sign first
@@ -10246,6 +10267,47 @@ where
 						"No active signing session. The associated funding transaction may have already been broadcast.".as_bytes().to_vec() });
 			}
 		}
+		if let Some(funding_txid) = retransmit_funding_commit_sig {
+			let funding = self
+				.pending_splice
+				.as_ref()
+				.and_then(|pending_splice| pending_splice.funding_negotiation.as_ref())
+				.and_then(|funding_negotiation| {
+					if let FundingNegotiation::AwaitingSignatures { funding, .. } = &funding_negotiation {
+						Some(funding)
+					} else {
+						None
+					}
+				})
+			.or_else(|| Some(&self.funding))
+				.filter(|funding| funding.get_funding_txid() == Some(funding_txid))
+				.ok_or_else(|| {
+					let message = "Failed to find funding for new commitment_signed".to_owned();
+					ChannelError::Close(
+						(
+							message.clone(),
+							ClosureReason::HolderForceClosed { message, broadcasted_latest_txn: Some(false) },
+						)
+					)
+				})?;
+
+			commitment_update = self
+				.context
+				.get_initial_commitment_signed_v2(&funding, logger)
+				.map(|commitment_signed|
+					msgs::CommitmentUpdate {
+						commitment_signed: vec![commitment_signed],
+						update_add_htlcs: vec![],
+						update_fulfill_htlcs: vec![],
+						update_fail_htlcs: vec![],
+						update_fail_malformed_htlcs: vec![],
+						update_fee: None,
+					}
+				);
+			if commitment_update.is_none() {
+				tx_signatures.take();
+			}
+		}
 
 		if matches!(self.context.channel_state, ChannelState::AwaitingChannelReady(_)) {
 			// If we're waiting on a monitor update, we shouldn't re-send any channel_ready's.