Handle receiving channel_reestablish with next_funding_txid

This follows the the specification closely in branching without being
too verbose, so that it should be easy to follow the logic.

See: https://github.com/lightning/bolts/blob/aa5207a/02-peer-protocol.md?plain=1#L2520-L2531

Author: dunxen

lightning/src/ln/channel.rs

@@ -1074,6 +1074,8 @@ pub(super) struct ReestablishResponses {
 	pub order: RAACommitmentOrder,
 	pub announcement_sigs: Option<msgs::AnnouncementSignatures>,
 	pub shutdown_msg: Option<msgs::Shutdown>,
+	pub tx_signatures: Option<msgs::TxSignatures>,
+	pub tx_abort: Option<msgs::TxAbort>,
 }
 
 /// The first message we send to our peer after connection
@@ -2540,7 +2542,7 @@ impl<SP: Deref> PendingV2Channel<SP> where SP::Target: SignerProvider {
 
 		let mut output_index = None;
 		let expected_spk = self.funding.get_funding_redeemscript().to_p2wsh();
-		for (idx, outp) in signing_session.unsigned_tx.outputs().enumerate() {
+		for (idx, outp) in signing_session.unsigned_tx().outputs().enumerate() {
 			if outp.script_pubkey() == &expected_spk && outp.value() == self.funding.get_value_satoshis() {
 				if output_index.is_some() {
 					return Err(ChannelError::Close(
@@ -2553,7 +2555,7 @@ impl<SP: Deref> PendingV2Channel<SP> where SP::Target: SignerProvider {
 			}
 		}
 		let outpoint = if let Some(output_index) = output_index {
-			OutPoint { txid: signing_session.unsigned_tx.compute_txid(), index: output_index }
+			OutPoint { txid: signing_session.unsigned_tx().compute_txid(), index: output_index }
 		} else {
 			return Err(ChannelError::Close(
 				(
@@ -2567,7 +2569,7 @@ impl<SP: Deref> PendingV2Channel<SP> where SP::Target: SignerProvider {
 		let commitment_signed = self.context.get_initial_commitment_signed(&self.funding, logger);
 		let commitment_signed = match commitment_signed {
 			Ok(commitment_signed) => {
-				self.funding.funding_transaction = Some(signing_session.unsigned_tx.build_unsigned_tx());
+				self.funding.funding_transaction = Some(signing_session.unsigned_tx().build_unsigned_tx());
 				commitment_signed
 			},
 			Err(err) => {
@@ -6543,7 +6545,7 @@ impl<SP: Deref> FundedChannel<SP> where
 		}
 
 		if let Some(ref mut signing_session) = self.interactive_tx_signing_session {
-			if msg.tx_hash != signing_session.unsigned_tx.compute_txid() {
+			if msg.tx_hash != signing_session.unsigned_tx().compute_txid() {
 				return Err(ChannelError::Close(
 					(
 						"The txid for the transaction does not match".to_string(),
@@ -7193,7 +7195,10 @@ impl<SP: Deref> FundedChannel<SP> where
 		}
 
 		if msg.next_local_commitment_number >= INITIAL_COMMITMENT_NUMBER || msg.next_remote_commitment_number >= INITIAL_COMMITMENT_NUMBER ||
-			msg.next_local_commitment_number == 0 {
+			(msg.next_local_commitment_number == 0 && msg.next_funding_txid.is_none()) {
+			// Note: This also covers the following case in the V2 channel establishment specification:
+			//   if `next_funding_txid` is not set, and `next_commitment_number` is zero:
+			//     MUST immediately fail the channel and broadcast any relevant latest commitment transaction.
 			return Err(ChannelError::close("Peer sent an invalid channel_reestablish to force close in a non-standard way".to_owned()));
 		}
 
@@ -7257,6 +7262,8 @@ impl<SP: Deref> FundedChannel<SP> where
 					raa: None, commitment_update: None,
 					order: RAACommitmentOrder::CommitmentFirst,
 					shutdown_msg, announcement_sigs,
+					tx_signatures: None,
+					tx_abort: None,
 				});
 			}
 
@@ -7266,6 +7273,8 @@ impl<SP: Deref> FundedChannel<SP> where
 				raa: None, commitment_update: None,
 				order: RAACommitmentOrder::CommitmentFirst,
 				shutdown_msg, announcement_sigs,
+				tx_signatures: None,
+				tx_abort: None,
 			});
 		}
 
@@ -7308,11 +7317,90 @@ impl<SP: Deref> FundedChannel<SP> where
 				log_debug!(logger, "Reconnected channel {} with no loss", &self.context.channel_id());
 			}
 
+			// if next_funding_txid is set:
+			let (commitment_update, tx_signatures, tx_abort) = if let Some(next_funding_txid) = msg.next_funding_txid {
+				if let Some(session) = &self.interactive_tx_signing_session {
+					// if next_funding_txid matches the latest interactive funding transaction:
+					let our_next_funding_txid = session.unsigned_tx().compute_txid();
+					if our_next_funding_txid == next_funding_txid {
+						debug_assert_eq!(session.unsigned_tx().compute_txid(), self.maybe_get_next_funding_txid().unwrap());
+
+						let commitment_update = if !session.has_received_tx_signatures() && msg.next_local_commitment_number == 0 {
+							// if it has not received tx_signatures for that funding transaction AND
+							// if next_commitment_number is zero:
+							//   MUST retransmit its commitment_signed for that funding transaction.
+							let commitment_signed = self.context.get_initial_commitment_signed(&self.funding, logger)?;
+							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,
+							})
+						} else { None };
+						// TODO(dual_funding): For async signing support we need to hold back `tx_signatures` until the `commitment_signed` is ready.
+						let tx_signatures = if (
+							// if it has not received tx_signatures for that funding transaction AND
+							// if it has already received commitment_signed AND it should sign first, as specified in the tx_signatures requirements:
+							//   MUST send its tx_signatures for that funding transaction.
+							!session.has_received_tx_signatures() && session.has_received_commitment_signed() && session.holder_sends_tx_signatures_first()
+							// else if it has already received tx_signatures for that funding transaction:
+							//   MUST send its tx_signatures for that funding transaction.
+						) || session.has_received_tx_signatures() {
+							if self.context.channel_state.is_monitor_update_in_progress() {
+								// The `monitor_pending_tx_signatures` field should have already been set in `commitment_signed_initial_v2`
+								// if we were up first for signing and had a monitor update in progress, but check again just in case.
+								debug_assert!(self.context.monitor_pending_tx_signatures.is_some(), "monitor_pending_tx_signatures should already be set");
+								log_debug!(logger, "Not sending tx_signatures: a monitor update is in progress. Setting monitor_pending_tx_signatures.");
+								if self.context.monitor_pending_tx_signatures.is_none() {
+									self.context.monitor_pending_tx_signatures = session.holder_tx_signatures().clone();
+								}
+								None
+							} else {
+								// If `holder_tx_signatures` is `None` here, the `tx_signatures` message will be sent
+								// when the holder provides their witnesses as this will queue a `tx_signatures` if the
+								// holder must send one.
+								session.holder_tx_signatures().clone()
+							}
+						} else {
+							None
+						};
+						if !session.has_received_commitment_signed() {
+							self.context.expecting_peer_commitment_signed = true;
+						}
+						(commitment_update, tx_signatures, None)
+					} else {
+						// The `next_funding_txid` does not match the latest interactive funding transaction so we
+						// MUST send tx_abort to let the remote  know that they can forget this funding transaction.
+						(None, None, Some(msgs::TxAbort {
+							channel_id: self.context.channel_id(),
+							data: format!(
+								"next_funding_txid {} does match our latest interactive funding txid {}",
+								next_funding_txid, our_next_funding_txid,
+							).into_bytes() }))
+					}
+				} else {
+					// We'll just send a `tx_abort` here if we don't have a signing session for this channel
+					// on reestablish and tell our peer to just forget about it.
+					// Our peer is doing something strange, but it doesn't warrant closing the channel.
+					(None, None, Some(msgs::TxAbort {
+						channel_id: self.context.channel_id(),
+						data:
+							"No active signing session. The associated funding transaction may have already been broadcast.".as_bytes().to_vec() }))
+				}
+			} else {
+				// Don't send anything related to interactive signing if `next_funding_txid` is not set.
+				(None, None, None)
+			};
+
 			Ok(ReestablishResponses {
 				channel_ready, shutdown_msg, announcement_sigs,
 				raa: required_revoke,
-				commitment_update: None,
+				commitment_update,
 				order: self.context.resend_order.clone(),
+				tx_signatures,
+				tx_abort,
 			})
 		} else if msg.next_local_commitment_number == next_counterparty_commitment_number - 1 {
 			if required_revoke.is_some() || self.context.signer_pending_revoke_and_ack {
@@ -7327,6 +7415,8 @@ impl<SP: Deref> FundedChannel<SP> where
 					channel_ready, shutdown_msg, announcement_sigs,
 					commitment_update: None, raa: None,
 					order: self.context.resend_order.clone(),
+					tx_signatures: None,
+					tx_abort: None,
 				})
 			} else {
 				let commitment_update = if self.context.resend_order == RAACommitmentOrder::RevokeAndACKFirst
@@ -7349,6 +7439,8 @@ impl<SP: Deref> FundedChannel<SP> where
 					channel_ready, shutdown_msg, announcement_sigs,
 					raa, commitment_update,
 					order: self.context.resend_order.clone(),
+					tx_signatures: None,
+					tx_abort: None,
 				})
 			}
 		} else if msg.next_local_commitment_number < next_counterparty_commitment_number {
@@ -8636,9 +8728,9 @@ impl<SP: Deref> FundedChannel<SP> where
 		// to the txid of that interactive transaction, else we MUST NOT set it.
 		if let Some(signing_session) = &self.interactive_tx_signing_session {
 			// Since we have a signing_session, this implies we've sent an initial `commitment_signed`...
-			if !signing_session.counterparty_sent_tx_signatures {
+			if !signing_session.has_received_tx_signatures() {
 				// ...but we didn't receive a `tx_signatures` from the counterparty yet.
-				Some(signing_session.unsigned_tx.compute_txid())
+				Some(signing_session.unsigned_tx().compute_txid())
 			} else {
 				// ...and we received a `tx_signatures` from the counterparty.
 				None

lightning/src/ln/channelmanager.rs

@@ -3253,7 +3253,7 @@ macro_rules! handle_monitor_update_completion {
 			&mut $peer_state.pending_msg_events, $chan, updates.raa,
 			updates.commitment_update, updates.order, updates.accepted_htlcs, updates.pending_update_adds,
 			updates.funding_broadcastable, updates.channel_ready,
-			updates.announcement_sigs, updates.tx_signatures);
+			updates.announcement_sigs, updates.tx_signatures, None);
 		if let Some(upd) = channel_update {
 			$peer_state.pending_msg_events.push(upd);
 		}
@@ -7665,10 +7665,10 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		pending_forwards: Vec<(PendingHTLCInfo, u64)>, pending_update_adds: Vec<msgs::UpdateAddHTLC>,
 		funding_broadcastable: Option<Transaction>,
 		channel_ready: Option<msgs::ChannelReady>, announcement_sigs: Option<msgs::AnnouncementSignatures>,
-		tx_signatures: Option<msgs::TxSignatures>
+		tx_signatures: Option<msgs::TxSignatures>, tx_abort: Option<msgs::TxAbort>,
 	) -> (Option<(u64, Option<PublicKey>, OutPoint, ChannelId, u128, Vec<(PendingHTLCInfo, u64)>)>, Option<(u64, Vec<msgs::UpdateAddHTLC>)>) {
 		let logger = WithChannelContext::from(&self.logger, &channel.context, None);
-		log_trace!(logger, "Handling channel resumption for channel {} with {} RAA, {} commitment update, {} pending forwards, {} pending update_add_htlcs, {}broadcasting funding, {} channel ready, {} announcement, {} tx_signatures",
+		log_trace!(logger, "Handling channel resumption for channel {} with {} RAA, {} commitment update, {} pending forwards, {} pending update_add_htlcs, {}broadcasting funding, {} channel ready, {} announcement, {} tx_signatures, {} tx_abort",
 			&channel.context.channel_id(),
 			if raa.is_some() { "an" } else { "no" },
 			if commitment_update.is_some() { "a" } else { "no" },
@@ -7677,6 +7677,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 			if channel_ready.is_some() { "sending" } else { "without" },
 			if announcement_sigs.is_some() { "sending" } else { "without" },
 			if tx_signatures.is_some() { "sending" } else { "without" },
+			if tx_abort.is_some() { "sending" } else { "without" },
 		);
 
 		let counterparty_node_id = channel.context.get_counterparty_node_id();
@@ -7710,6 +7711,12 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 				msg,
 			});
 		}
+		if let Some(msg) = tx_abort {
+			pending_msg_events.push(MessageSendEvent::SendTxAbort {
+				node_id: counterparty_node_id,
+				msg,
+			});
+		}
 
 		macro_rules! handle_cs { () => {
 			if let Some(update) = commitment_update {
@@ -9491,7 +9498,8 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 						let need_lnd_workaround = chan.context.workaround_lnd_bug_4006.take();
 						let (htlc_forwards, decode_update_add_htlcs) = self.handle_channel_resumption(
 							&mut peer_state.pending_msg_events, chan, responses.raa, responses.commitment_update, responses.order,
-							Vec::new(), Vec::new(), None, responses.channel_ready, responses.announcement_sigs, None);
+							Vec::new(), Vec::new(), None, responses.channel_ready, responses.announcement_sigs,
+							responses.tx_signatures, responses.tx_abort);
 						debug_assert!(htlc_forwards.is_none());
 						debug_assert!(decode_update_add_htlcs.is_none());
 						if let Some(upd) = channel_update {

lightning/src/ln/interactivetxs.rs

@@ -309,16 +309,36 @@ impl ConstructedTransaction {
 /// https://github.com/lightning/bolts/blob/master/02-peer-protocol.md#sharing-funding-signatures-tx_signatures
 #[derive(Debug, Clone, PartialEq)]
 pub(crate) struct InteractiveTxSigningSession {
-	pub unsigned_tx: ConstructedTransaction,
-	pub counterparty_sent_tx_signatures: bool,
+	unsigned_tx: ConstructedTransaction,
+	has_received_tx_signatures: bool,
 	holder_sends_tx_signatures_first: bool,
-	received_commitment_signed: bool,
+	has_received_commitment_signed: bool,
 	holder_tx_signatures: Option<TxSignatures>,
 }
 
 impl InteractiveTxSigningSession {
+	pub fn unsigned_tx(&self) -> &ConstructedTransaction {
+		&self.unsigned_tx
+	}
+
+	pub fn has_received_tx_signatures(&self) -> bool {
+		self.has_received_tx_signatures
+	}
+
+	pub fn holder_sends_tx_signatures_first(&self) -> bool {
+		self.holder_sends_tx_signatures_first
+	}
+
+	pub fn has_received_commitment_signed(&self) -> bool {
+		self.has_received_commitment_signed
+	}
+
+	pub fn holder_tx_signatures(&self) -> &Option<TxSignatures> {
+		&self.holder_tx_signatures
+	}
+
 	pub fn received_commitment_signed(&mut self) -> Option<TxSignatures> {
-		self.received_commitment_signed = true;
+		self.has_received_commitment_signed = true;
 		if self.holder_sends_tx_signatures_first {
 			self.holder_tx_signatures.clone()
 		} else {
@@ -327,7 +347,7 @@ impl InteractiveTxSigningSession {
 	}
 
 	pub fn get_tx_signatures(&self) -> Option<TxSignatures> {
-		if self.received_commitment_signed {
+		if self.has_received_commitment_signed {
 			self.holder_tx_signatures.clone()
 		} else {
 			None
@@ -352,7 +372,7 @@ impl InteractiveTxSigningSession {
 			return Err(());
 		}
 		self.unsigned_tx.add_remote_witnesses(tx_signatures.witnesses.clone());
-		self.counterparty_sent_tx_signatures = true;
+		self.has_received_tx_signatures = true;
 
 		let holder_tx_signatures = if !self.holder_sends_tx_signatures_first {
 			self.holder_tx_signatures.clone()
@@ -1008,9 +1028,9 @@ macro_rules! define_state_transitions {
 				let signing_session = InteractiveTxSigningSession {
 					holder_sends_tx_signatures_first: tx.holder_sends_tx_signatures_first,
 					unsigned_tx: tx,
-					received_commitment_signed: false,
+					has_received_commitment_signed: false,
 					holder_tx_signatures: None,
-					counterparty_sent_tx_signatures: false,
+					has_received_tx_signatures: false,
 				};
 				Ok(NegotiationComplete(signing_session))
 			}