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

@@ -930,6 +930,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
@@ -2117,7 +2119,7 @@ impl<SP: Deref> PendingV2Channel<SP> where SP::Target: SignerProvider {
 
 		let mut output_index = None;
 		let expected_spk = self.context.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.context.get_value_satoshis() {
 				if output_index.is_some() {
 					return Err(ChannelError::Close(
@@ -2130,7 +2132,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(
 				(
@@ -2145,7 +2147,7 @@ impl<SP: Deref> PendingV2Channel<SP> where SP::Target: SignerProvider {
 		let commitment_signed = self.context.get_initial_commitment_signed(logger);
 		let commitment_signed = match commitment_signed {
 			Ok(commitment_signed) => {
-				self.context.funding_transaction = Some(signing_session.unsigned_tx.build_unsigned_tx());
+				self.context.funding_transaction = Some(signing_session.unsigned_tx().build_unsigned_tx());
 				commitment_signed
 			},
 			Err(err) => {
@@ -6003,7 +6005,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(),
@@ -6633,7 +6635,7 @@ 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() {
 			return Err(ChannelError::close("Peer sent an invalid channel_reestablish to force close in a non-standard way".to_owned()));
 		}
 
@@ -6697,6 +6699,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,
 				});
 			}
 
@@ -6706,6 +6710,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,
 			});
 		}
 
@@ -6751,11 +6757,67 @@ 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:
+					if session.unsigned_tx().compute_txid() == next_funding_txid {
+						// if it has not received tx_signatures for that funding transaction:
+						if !session.counterparty_sent_tx_signatures() {
+							// if next_commitment_number is zero:
+							let commitment_update = if msg.next_local_commitment_number == 0 {
+								// MUST retransmit its commitment_signed for that funding transaction.
+								let commitment_signed = self.context.get_initial_commitment_signed(logger)?;
+								Some(msgs::CommitmentUpdate {
+									commitment_signed,
+									update_add_htlcs: vec![],
+									update_fulfill_htlcs: vec![],
+									update_fail_htlcs: vec![],
+									update_fail_malformed_htlcs: vec![],
+									update_fee: None,
+								})
+							} else { None };
+							// if it has already received commitment_signed and it should sign first, as specified in the tx_signatures requirements:
+							if session.has_received_commitment_signed() && session.holder_sends_tx_signatures_first() {
+								// MUST send its tx_signatures for that funding transaction.
+								(commitment_update, session.holder_tx_signatures().clone(), None)
+							} else {
+								(commitment_update, None, None)
+							}
+						} else {
+							// if it has already received tx_signatures for that funding transaction:
+							// MUST send its tx_signatures for that funding transaction.
+							(None, session.holder_tx_signatures().clone(), None)
+						}
+					} else {
+						// MUST send tx_abort to let the sending node know that they can forget this funding transaction.
+						(None, None, Some(msgs::TxAbort { channel_id: self.context.channel_id(), data: vec![] }))
+					}
+				} else {
+					// Counterparty set `next_funding_txid` at incorrect state.
+					// TODO(dual_funding): Should probably error here (or send tx_abort) but not in spec.
+					(None, None, None)
+				}
+			} else {
+				// if `next_funding_txid` is not set, and `next_commitment_number` is zero:
+				if msg.next_local_commitment_number == 0 {
+					// MUST immediately fail the channel and broadcast any relevant latest commitment transaction.
+					return Err(ChannelError::close(format!(
+						"Peer attempted to reestablish channel expecting a future local commitment transaction: {} (received) vs {} (expected)",
+						msg.next_remote_commitment_number,
+						our_commitment_transaction
+					)));
+				}
+				(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 {
@@ -6770,6 +6832,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
@@ -6792,6 +6856,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 {
@@ -8069,7 +8135,7 @@ 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.counterparty_sent_tx_signatures() {
 				// ...but we didn't receive a `tx_signatures` from the counterparty yet.
 				Some(self.funding_outpoint().txid)
 			} else {

lightning/src/ln/channelmanager.rs

@@ -3208,7 +3208,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);
 		}
@@ -7522,10 +7522,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" },
@@ -7534,6 +7534,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();
@@ -7567,6 +7568,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(events::MessageSendEvent::SendTxAbort {
+				node_id: counterparty_node_id,
+				msg,
+			});
+		}
 
 		macro_rules! handle_cs { () => {
 			if let Some(update) = commitment_update {
@@ -9238,7 +9245,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

@@ -289,16 +289,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,
+	counterparty_sent_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 counterparty_sent_tx_signatures(&self) -> bool {
+		self.counterparty_sent_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 {
@@ -307,7 +327,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
@@ -988,7 +1008,7 @@ 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,
 				};