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-net-tokio/src/lib.rs

@@ -622,15 +622,17 @@ impl Hash for SocketDescriptor {
 mod tests {
 	use bitcoin::constants::ChainHash;
 	use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey};
-	use bitcoin::Network;
+	use bitcoin::{Network, Txid};
 	use lightning::ln::msgs::*;
 	use lightning::ln::peer_handler::{IgnoringMessageHandler, MessageHandler, PeerManager};
+	use lightning::ln::types::ChannelId;
 	use lightning::routing::gossip::NodeId;
 	use lightning::types::features::*;
 	use lightning::util::test_utils::TestNodeSigner;
 
 	use tokio::sync::mpsc;
 
+	use std::collections::BTreeMap;
 	use std::mem;
 	use std::sync::atomic::{AtomicBool, Ordering};
 	use std::sync::{Arc, Mutex};
@@ -723,6 +725,11 @@ mod tests {
 		) {
 		}
 		fn handle_commitment_signed(&self, _their_node_id: PublicKey, _msg: &CommitmentSigned) {}
+		fn handle_commitment_signed_batch(
+			&self, _their_node_id: PublicKey, _channel_id: ChannelId,
+			_batch: BTreeMap<Txid, CommitmentSigned>,
+		) {
+		}
 		fn handle_revoke_and_ack(&self, _their_node_id: PublicKey, _msg: &RevokeAndACK) {}
 		fn handle_update_fee(&self, _their_node_id: PublicKey, _msg: &UpdateFee) {}
 		fn handle_announcement_signatures(

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::BTreeMap;
+
 use crate::io;
 use crate::prelude::*;
 use core::{cmp,mem,fmt};
@@ -5778,6 +5780,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");
 
@@ -5805,6 +5812,53 @@ impl<SP: Deref> FundedChannel<SP> where
 	pub fn commitment_signed<L: Deref>(&mut self, msg: &msgs::CommitmentSigned, logger: &L) -> Result<Option<ChannelMonitorUpdate>, ChannelError>
 		where L::Target: Logger
 	{
+		self.commitment_signed_check_state()?;
+
+		if !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 updates = 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,
+				}]
+			)?;
+
+		self.commitment_signed_update_monitor(updates, logger)
+	}
+
+	pub fn commitment_signed_batch<L: Deref>(&mut self, batch: &BTreeMap<Txid, msgs::CommitmentSigned>, logger: &L) -> Result<Option<ChannelMonitorUpdate>, ChannelError>
+		where L::Target: Logger
+	{
+		self.commitment_signed_check_state()?;
+
+		// Any commitment_signed not associated with a FundingScope is ignored below if a
+		// pending splice transaction has confirmed since receiving the batch.
+		let updates = core::iter::once(&self.funding)
+			.chain(self.pending_funding.iter())
+			.map(|funding| {
+				let funding_txid = funding.get_funding_txo().unwrap().txid;
+				let msg = 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_update_monitor(updates, logger)
+	}
+
+	fn commitment_signed_check_state(&self) -> Result<(), ChannelError> {
 		if self.context.channel_state.is_quiescent() {
 			return Err(ChannelError::WarnAndDisconnect("Got commitment_signed message while quiescent".to_owned()));
 		}
@@ -5818,8 +5872,12 @@ 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)?;
+		Ok(())
+	}
 
+	fn commitment_signed_update_monitor<L: Deref>(&mut self, mut updates: Vec<ChannelMonitorUpdateStep>, logger: &L) -> Result<Option<ChannelMonitorUpdate>, ChannelError>
+		where L::Target: Logger
+	{
 		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
 			// waiting for our signer to unblock and provide a commitment point.
@@ -5873,18 +5931,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()),
 		};
 

lightning/src/ln/channelmanager.rs

@@ -9018,6 +9018,38 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		}
 	}
 
+	fn internal_commitment_signed_batch(&self, counterparty_node_id: &PublicKey, channel_id: ChannelId, batch: &BTreeMap<Txid, msgs::CommitmentSigned>) -> Result<(), MsgHandleErrInternal> {
+		let per_peer_state = self.per_peer_state.read().unwrap();
+		let peer_state_mutex = per_peer_state.get(counterparty_node_id)
+			.ok_or_else(|| {
+				debug_assert!(false);
+				MsgHandleErrInternal::send_err_msg_no_close(format!("Can't find a peer matching the passed counterparty node_id {}", counterparty_node_id), channel_id)
+			})?;
+		let mut peer_state_lock = peer_state_mutex.lock().unwrap();
+		let peer_state = &mut *peer_state_lock;
+		match peer_state.channel_by_id.entry(channel_id) {
+			hash_map::Entry::Occupied(mut chan_entry) => {
+				let chan = chan_entry.get_mut();
+				let logger = WithChannelContext::from(&self.logger, &chan.context(), None);
+				let funding_txo = chan.funding().get_funding_txo();
+				if let Some(chan) = chan.as_funded_mut() {
+					let monitor_update_opt = try_channel_entry!(
+						self, peer_state, chan.commitment_signed_batch(batch, &&logger), chan_entry
+					);
+
+					if let Some(monitor_update) = monitor_update_opt {
+						handle_new_monitor_update!(
+							self, funding_txo.unwrap(), monitor_update, peer_state_lock, peer_state,
+							per_peer_state, chan
+						);
+					}
+				}
+				Ok(())
+			},
+			hash_map::Entry::Vacant(_) => Err(MsgHandleErrInternal::send_err_msg_no_close(format!("Got a message for a channel from the wrong node! No such channel for the passed counterparty_node_id {}", counterparty_node_id), channel_id))
+		}
+	}
+
 	fn push_decode_update_add_htlcs(&self, mut update_add_htlcs: (u64, Vec<msgs::UpdateAddHTLC>)) {
 		let mut push_forward_event = self.forward_htlcs.lock().unwrap().is_empty();
 		let mut decode_update_add_htlcs = self.decode_update_add_htlcs.lock().unwrap();
@@ -12128,6 +12160,11 @@ where
 		let _ = handle_error!(self, self.internal_commitment_signed(&counterparty_node_id, msg), counterparty_node_id);
 	}
 
+	fn handle_commitment_signed_batch(&self, counterparty_node_id: PublicKey, channel_id: ChannelId, batch: BTreeMap<Txid, msgs::CommitmentSigned>) {
+		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(self);
+		let _ = handle_error!(self, self.internal_commitment_signed_batch(&counterparty_node_id, channel_id, &batch), counterparty_node_id);
+	}
+
 	fn handle_revoke_and_ack(&self, counterparty_node_id: PublicKey, msg: &msgs::RevokeAndACK) {
 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(self);
 		let _ = handle_error!(self, self.internal_revoke_and_ack(&counterparty_node_id, msg), counterparty_node_id);

lightning/src/ln/msgs.rs

@@ -1938,7 +1938,7 @@ pub trait ChannelMessageHandler : BaseMessageHandler {
 	fn handle_commitment_signed_batch(
 		&self, their_node_id: PublicKey, channel_id: ChannelId,
 		batch: BTreeMap<Txid, CommitmentSigned>,
-	) {}
+	);
 	/// Handle an incoming `revoke_and_ack` message from the given peer.
 	fn handle_revoke_and_ack(&self, their_node_id: PublicKey, msg: &RevokeAndACK);
 

lightning/src/ln/peer_handler.rs

@@ -334,6 +334,12 @@ impl ChannelMessageHandler for ErroringMessageHandler {
 	fn handle_commitment_signed(&self, their_node_id: PublicKey, msg: &msgs::CommitmentSigned) {
 		ErroringMessageHandler::push_error(self, their_node_id, msg.channel_id);
 	}
+	fn handle_commitment_signed_batch(
+		&self, their_node_id: PublicKey, channel_id: ChannelId,
+		_batch: BTreeMap<Txid, msgs::CommitmentSigned>,
+	) {
+		ErroringMessageHandler::push_error(self, their_node_id, channel_id);
+	}
 	fn handle_revoke_and_ack(&self, their_node_id: PublicKey, msg: &msgs::RevokeAndACK) {
 		ErroringMessageHandler::push_error(self, their_node_id, msg.channel_id);
 	}

lightning/src/util/test_utils.rs

@@ -79,6 +79,8 @@ use bitcoin::secp256k1::{self, PublicKey, Scalar, Secp256k1, SecretKey};
 
 use lightning_invoice::RawBolt11Invoice;
 
+use alloc::collections::BTreeMap;
+
 use crate::io;
 use crate::prelude::*;
 use crate::sign::{EntropySource, NodeSigner, RandomBytes, Recipient, SignerProvider};
@@ -1053,6 +1055,12 @@ impl msgs::ChannelMessageHandler for TestChannelMessageHandler {
 	fn handle_commitment_signed(&self, _their_node_id: PublicKey, msg: &msgs::CommitmentSigned) {
 		self.received_msg(wire::Message::CommitmentSigned(msg.clone()));
 	}
+	fn handle_commitment_signed_batch(
+		&self, _their_node_id: PublicKey, _channel_id: ChannelId,
+		_batch: BTreeMap<Txid, msgs::CommitmentSigned>,
+	) {
+		unreachable!()
+	}
 	fn handle_revoke_and_ack(&self, _their_node_id: PublicKey, msg: &msgs::RevokeAndACK) {
 		self.received_msg(wire::Message::RevokeAndACK(msg.clone()));
 	}