Batch commitment_signed messages for splicing

A FundedChannel may have more than one pending FundingScope during
splicing, one for the splice attempt and one or more for any RBF
attempts. The counterparty will send a commitment_signed message for
each pending splice transaction and the current funding transaction.

Defer handling these commitment_signed messages until the entire batch
has arrived. Then validate them individually, also checking if all the
pending splice transactions and the current funding transaction have a
corresponding commitment_signed in the batch.

Author: jkczyz

lightning/src/ln/channel.rs

@@ -66,6 +66,8 @@ use crate::util::errors::APIError;
 use crate::util::config::{UserConfig, ChannelConfig, LegacyChannelConfig, ChannelHandshakeConfig, ChannelHandshakeLimits, MaxDustHTLCExposure};
 use crate::util::scid_utils::scid_from_parts;
 
+use alloc::collections::{btree_map, BTreeMap};
+
 use crate::io;
 use crate::prelude::*;
 use core::{cmp,mem,fmt};
@@ -1515,6 +1517,7 @@ impl<SP: Deref> Channel<SP> where
 				let mut funded_channel = FundedChannel {
 					funding: chan.funding,
 					pending_funding: vec![],
+					commitment_signed_batch: BTreeMap::new(),
 					context: chan.context,
 					interactive_tx_signing_session: chan.interactive_tx_signing_session,
 					holder_commitment_point,
@@ -4949,6 +4952,7 @@ pub(super) struct DualFundingChannelContext {
 pub(super) struct FundedChannel<SP: Deref> where SP::Target: SignerProvider {
 	pub funding: FundingScope,
 	pending_funding: Vec<FundingScope>,
+	commitment_signed_batch: BTreeMap<Txid, msgs::CommitmentSigned>,
 	pub context: ChannelContext<SP>,
 	pub interactive_tx_signing_session: Option<InteractiveTxSigningSession>,
 	holder_commitment_point: HolderCommitmentPoint,
@@ -5741,6 +5745,11 @@ impl<SP: Deref> FundedChannel<SP> where
 					ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(false) },
 				)));
 		}
+
+		if msg.batch.is_some() {
+			return Err(ChannelError::close("Peer sent initial commitment_signed with a batch".to_owned()));
+		}
+
 		let holder_commitment_point = &mut self.holder_commitment_point.clone();
 		self.context.assert_no_commitment_advancement(holder_commitment_point.transaction_number(), "initial commitment_signed");
 
@@ -5781,7 +5790,58 @@ impl<SP: Deref> FundedChannel<SP> where
 			return Err(ChannelError::close("Peer sent commitment_signed after we'd started exchanging closing_signeds".to_owned()));
 		}
 
-		let commitment_tx_info = self.context.validate_commitment_signed(&self.funding, &self.holder_commitment_point, msg, logger)?;
+		if msg.batch.is_none() && !self.pending_funding.is_empty() {
+			return Err(ChannelError::close("Peer sent commitment_signed without a batch when there's a pending splice".to_owned()));
+		}
+
+		let mut updates = match &msg.batch {
+			// No pending splice
+			None => {
+				debug_assert!(self.pending_funding.is_empty());
+				debug_assert!(self.commitment_signed_batch.is_empty());
+				self.context
+					.validate_commitment_signed(&self.funding, &self.holder_commitment_point, msg, logger)
+					.map(|LatestHolderCommitmentTXInfo { commitment_tx, htlc_outputs, nondust_htlc_sources }|
+						vec![ChannelMonitorUpdateStep::LatestHolderCommitmentTXInfo {
+							commitment_tx, htlc_outputs, claimed_htlcs: vec![], nondust_htlc_sources,
+						}]
+					)?
+			},
+			// May or may not have a pending splice
+			Some(batch) => {
+				match self.commitment_signed_batch.entry(batch.funding_txid) {
+					btree_map::Entry::Vacant(entry) => { entry.insert(msg.clone()); },
+					btree_map::Entry::Occupied(entry) => {
+						return Err(ChannelError::close(format!("Peer sent commitment_signed with duplicate funding_txid {} in a batch", entry.key())));
+					},
+				}
+
+				if self.commitment_signed_batch.len() < batch.batch_size as usize {
+					return Ok(None);
+				}
+
+				// Any commitment_signed not associated with a FundingScope is ignored below if a
+				// pending splice transaction has confirmed since receiving the batch.
+				core::iter::once(&self.funding)
+					.chain(self.pending_funding.iter())
+					.map(|funding| {
+						let funding_txid = funding.get_funding_txo().unwrap().txid;
+						let msg = self.commitment_signed_batch
+							.get(&funding_txid)
+							.ok_or_else(|| ChannelError::close(format!("Peer did not send a commitment_signed for pending splice transaction: {}", funding_txid)))?;
+						self.context
+							.validate_commitment_signed(funding, &self.holder_commitment_point, msg, logger)
+							.map(|LatestHolderCommitmentTXInfo { commitment_tx, htlc_outputs, nondust_htlc_sources }|
+								ChannelMonitorUpdateStep::LatestHolderCommitmentTXInfo {
+									commitment_tx, htlc_outputs, claimed_htlcs: vec![], nondust_htlc_sources,
+								}
+							)
+						}
+					)
+					.collect::<Result<Vec<_>, ChannelError>>()?
+			},
+		};
+		self.commitment_signed_batch.clear();
 
 		if self.holder_commitment_point.advance(&self.context.holder_signer, &self.context.secp_ctx, logger).is_err() {
 			// We only fail to advance our commitment point/number if we're currently
@@ -5836,18 +5896,21 @@ impl<SP: Deref> FundedChannel<SP> where
 			}
 		}
 
-		let LatestHolderCommitmentTXInfo {
-			commitment_tx, htlc_outputs, nondust_htlc_sources,
-		} = commitment_tx_info;
+		for mut update in updates.iter_mut() {
+			if let ChannelMonitorUpdateStep::LatestHolderCommitmentTXInfo {
+				claimed_htlcs: ref mut update_claimed_htlcs, ..
+			} = &mut update {
+				debug_assert!(update_claimed_htlcs.is_empty());
+				*update_claimed_htlcs = claimed_htlcs.clone();
+			} else {
+				debug_assert!(false);
+			}
+		}
+
 		self.context.latest_monitor_update_id += 1;
 		let mut monitor_update = ChannelMonitorUpdate {
 			update_id: self.context.latest_monitor_update_id,
-			updates: vec![ChannelMonitorUpdateStep::LatestHolderCommitmentTXInfo {
-				commitment_tx,
-				htlc_outputs,
-				claimed_htlcs,
-				nondust_htlc_sources,
-			}],
+			updates,
 			channel_id: Some(self.context.channel_id()),
 		};
 
@@ -9541,6 +9604,7 @@ impl<SP: Deref> OutboundV1Channel<SP> where SP::Target: SignerProvider {
 		let mut channel = FundedChannel {
 			funding: self.funding,
 			pending_funding: vec![],
+			commitment_signed_batch: BTreeMap::new(),
 			context: self.context,
 			interactive_tx_signing_session: None,
 			is_v2_established: false,
@@ -9818,6 +9882,7 @@ impl<SP: Deref> InboundV1Channel<SP> where SP::Target: SignerProvider {
 		let mut channel = FundedChannel {
 			funding: self.funding,
 			pending_funding: vec![],
+			commitment_signed_batch: BTreeMap::new(),
 			context: self.context,
 			interactive_tx_signing_session: None,
 			is_v2_established: false,
@@ -11084,6 +11149,7 @@ impl<'a, 'b, 'c, ES: Deref, SP: Deref> ReadableArgs<(&'a ES, &'b SP, &'c Channel
 				funding_transaction,
 			},
 			pending_funding: pending_funding.unwrap(),
+			commitment_signed_batch: BTreeMap::new(),
 			context: ChannelContext {
 				user_id,