Implement start_batch message batching

lightning-net-tokio/src/lib.rs

@@ -622,7 +622,7 @@ impl Hash for SocketDescriptor {
 mod tests {
 	use bitcoin::constants::ChainHash;
 	use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey};
-	use bitcoin::{Network, Txid};
+	use bitcoin::Network;
 	use lightning::ln::msgs::*;
 	use lightning::ln::peer_handler::{IgnoringMessageHandler, MessageHandler, PeerManager};
 	use lightning::ln::types::ChannelId;
@@ -632,7 +632,6 @@ mod tests {
 
 	use tokio::sync::mpsc;
 
-	use std::collections::BTreeMap;
 	use std::mem;
 	use std::sync::atomic::{AtomicBool, Ordering};
 	use std::sync::{Arc, Mutex};
@@ -726,8 +725,7 @@ 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>,
+			&self, _their_node_id: PublicKey, _channel_id: ChannelId, _batch: Vec<CommitmentSigned>,
 		) {
 		}
 		fn handle_revoke_and_ack(&self, _their_node_id: PublicKey, _msg: &RevokeAndACK) {}

lightning/src/ln/channel.rs

@@ -66,7 +66,7 @@ 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 alloc::collections::{btree_map, BTreeMap};
 
 use crate::io;
 use crate::prelude::*;
@@ -4943,7 +4943,7 @@ impl<SP: Deref> ChannelContext<SP> where SP::Target: SignerProvider {
 			channel_id: self.channel_id,
 			htlc_signatures: vec![],
 			signature,
-			batch: None,
+			funding_txid: funding.get_funding_txo().map(|funding_txo| funding_txo.txid),
 			#[cfg(taproot)]
 			partial_signature_with_nonce: None,
 		})
@@ -5909,10 +5909,6 @@ impl<SP: Deref> FundedChannel<SP> where
 				)));
 		}
 
-		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");
 
@@ -5954,18 +5950,35 @@ impl<SP: Deref> FundedChannel<SP> where
 		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>
+	pub fn commitment_signed_batch<L: Deref>(&mut self, batch: Vec<msgs::CommitmentSigned>, logger: &L) -> Result<Option<ChannelMonitorUpdate>, ChannelError>
 		where L::Target: Logger
 	{
 		self.commitment_signed_check_state()?;
 
+		let mut messages = BTreeMap::new();
+		for msg in batch {
+			let funding_txid = match msg.funding_txid {
+				Some(funding_txid) => funding_txid,
+				None => {
+					return Err(ChannelError::close("Peer sent batched commitment_signed without a funding_txid".to_string()));
+				},
+			};
+
+			match messages.entry(funding_txid) {
+				btree_map::Entry::Vacant(entry) => { entry.insert(msg); },
+				btree_map::Entry::Occupied(_) => {
+					return Err(ChannelError::close(format!("Peer sent batched commitment_signed with duplicate funding_txid {}", funding_txid)));
+				}
+			}
+		}
+
 		// 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
+				let msg = messages
 					.get(&funding_txid)
 					.ok_or_else(|| ChannelError::close(format!("Peer did not send a commitment_signed for pending splice transaction: {}", funding_txid)))?;
 				self.context
@@ -9176,20 +9189,11 @@ impl<SP: Deref> FundedChannel<SP> where
 					}
 				}
 
-				let batch = if self.pending_funding.is_empty() { None } else {
-					Some(msgs::CommitmentSignedBatch {
-						batch_size: self.pending_funding.len() as u16 + 1,
-						funding_txid: funding
-							.get_funding_txo()
-							.expect("splices should have their funding transactions negotiated before exiting quiescence while un-negotiated splices are discarded on reload")
-							.txid,
-					})
-				};
 				Ok(msgs::CommitmentSigned {
 					channel_id: self.context.channel_id,
 					signature,
 					htlc_signatures,
-					batch,
+					funding_txid: funding.get_funding_txo().map(|funding_txo| funding_txo.txid),
 					#[cfg(taproot)]
 					partial_signature_with_nonce: None,
 				})

lightning/src/ln/channelmanager.rs

@@ -9043,7 +9043,7 @@ 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> {
+	fn internal_commitment_signed_batch(&self, counterparty_node_id: &PublicKey, channel_id: ChannelId, batch: Vec<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(|| {
@@ -12195,9 +12195,9 @@ 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>) {
+	fn handle_commitment_signed_batch(&self, counterparty_node_id: PublicKey, channel_id: ChannelId, batch: Vec<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);
+		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) {

lightning/src/ln/dual_funding_tests.rs

@@ -183,7 +183,7 @@ fn do_test_v2_channel_establishment(session: V2ChannelEstablishmentTestSession)
 			)
 			.unwrap(),
 		htlc_signatures: vec![],
-		batch: None,
+		funding_txid: None,
 		#[cfg(taproot)]
 		partial_signature_with_nonce: None,
 	};

lightning/src/ln/htlc_reserve_unit_tests.rs

@@ -899,7 +899,7 @@ pub fn test_fee_spike_violation_fails_htlc() {
 		channel_id: chan.2,
 		signature: res.0,
 		htlc_signatures: res.1,
-		batch: None,
+		funding_txid: None,
 		#[cfg(taproot)]
 		partial_signature_with_nonce: None,
 	};

lightning/src/ln/msgs.rs

@@ -47,8 +47,6 @@ use crate::types::payment::{PaymentHash, PaymentPreimage, PaymentSecret};
 #[allow(unused_imports)]
 use crate::prelude::*;
 
-use alloc::collections::BTreeMap;
-
 use crate::io::{self, Cursor, Read};
 use crate::io_extras::read_to_end;
 use core::fmt;
@@ -686,6 +684,18 @@ pub struct ClosingSigned {
 	pub fee_range: Option<ClosingSignedFeeRange>,
 }
 
+/// A [`start_batch`] message to be sent to group together multiple channel messages as a single
+/// logical message.
+///
+/// [`start_batch`]: https://github.com/lightning/bolts/blob/master/02-peer-protocol.md#batching-channel-messages
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+pub struct StartBatch {
+	/// The channel ID of all messages in the batch.
+	pub channel_id: ChannelId,
+	/// The number of messages to follow.
+	pub batch_size: u16,
+}
+
 /// An [`update_add_htlc`] message to be sent to or received from a peer.
 ///
 /// [`update_add_htlc`]: https://github.com/lightning/bolts/blob/master/02-peer-protocol.md#adding-an-htlc-update_add_htlc
@@ -795,15 +805,6 @@ pub struct UpdateFailMalformedHTLC {
 	pub failure_code: u16,
 }
 
-/// Optional batch parameters for `commitment_signed` message.
-#[derive(Clone, Debug, Hash, PartialEq, Eq)]
-pub struct CommitmentSignedBatch {
-	/// Batch size N: all N `commitment_signed` messages must be received before being processed
-	pub batch_size: u16,
-	/// The funding transaction, to discriminate among multiple pending funding transactions (e.g. in case of splicing)
-	pub funding_txid: Txid,
-}
-
 /// A [`commitment_signed`] message to be sent to or received from a peer.
 ///
 /// [`commitment_signed`]: https://github.com/lightning/bolts/blob/master/02-peer-protocol.md#committing-updates-so-far-commitment_signed
@@ -815,8 +816,8 @@ pub struct CommitmentSigned {
 	pub signature: Signature,
 	/// Signatures on the HTLC transactions
 	pub htlc_signatures: Vec<Signature>,
-	/// Optional batch size and other parameters
-	pub batch: Option<CommitmentSignedBatch>,
+	/// The funding transaction, to discriminate among multiple pending funding transactions (e.g. in case of splicing)
+	pub funding_txid: Option<Txid>,
 	#[cfg(taproot)]
 	/// The partial Taproot signature on the commitment transaction
 	pub partial_signature_with_nonce: Option<PartialSignatureWithNonce>,
@@ -1947,8 +1948,7 @@ pub trait ChannelMessageHandler: BaseMessageHandler {
 	fn handle_commitment_signed(&self, their_node_id: PublicKey, msg: &CommitmentSigned);
 	/// Handle a batch of incoming `commitment_signed` message from the given peer.
 	fn handle_commitment_signed_batch(
-		&self, their_node_id: PublicKey, channel_id: ChannelId,
-		batch: BTreeMap<Txid, CommitmentSigned>,
+		&self, their_node_id: PublicKey, channel_id: ChannelId, batch: Vec<CommitmentSigned>,
 	);
 	/// Handle an incoming `revoke_and_ack` message from the given peer.
 	fn handle_revoke_and_ack(&self, their_node_id: PublicKey, msg: &RevokeAndACK);
@@ -1959,19 +1959,10 @@ pub trait ChannelMessageHandler: BaseMessageHandler {
 	) {
 		assert!(!batch.is_empty());
 		if batch.len() == 1 {
-			assert!(batch[0].batch.is_none());
 			self.handle_commitment_signed(their_node_id, &batch[0]);
 		} else {
 			let channel_id = batch[0].channel_id;
-			let batch: BTreeMap<Txid, CommitmentSigned> = batch
-				.iter()
-				.cloned()
-				.map(|mut cs| {
-					let funding_txid = cs.batch.take().unwrap().funding_txid;
-					(funding_txid, cs)
-				})
-				.collect();
-			self.handle_commitment_signed_batch(their_node_id, channel_id, batch);
+			self.handle_commitment_signed_batch(their_node_id, channel_id, batch.clone());
 		}
 	}
 
@@ -2741,18 +2732,13 @@ impl_writeable!(ClosingSignedFeeRange, {
 	max_fee_satoshis
 });
 
-impl_writeable_msg!(CommitmentSignedBatch, {
-	batch_size,
-	funding_txid,
-}, {});
-
 #[cfg(not(taproot))]
 impl_writeable_msg!(CommitmentSigned, {
 	channel_id,
 	signature,
 	htlc_signatures
 }, {
-	(0, batch, option),
+	(1, funding_txid, option),
 });
 
 #[cfg(taproot)]
@@ -2761,7 +2747,7 @@ impl_writeable_msg!(CommitmentSigned, {
 	signature,
 	htlc_signatures
 }, {
-	(0, batch, option),
+	(1, funding_txid, option),
 	(2, partial_signature_with_nonce, option),
 });
 
@@ -3082,6 +3068,11 @@ impl_writeable_msg!(PeerStorage, { data }, {});
 
 impl_writeable_msg!(PeerStorageRetrieval, { data }, {});
 
+impl_writeable_msg!(StartBatch, {
+	channel_id,
+	batch_size
+}, {});
+
 // Note that this is written as a part of ChannelManager objects, and thus cannot change its
 // serialization format in a way which assumes we know the total serialized length/message end
 // position.
@@ -5617,13 +5608,10 @@ mod tests {
 			channel_id: ChannelId::from_bytes([2; 32]),
 			signature: sig_1,
 			htlc_signatures: if htlcs { vec![sig_2, sig_3, sig_4] } else { Vec::new() },
-			batch: Some(msgs::CommitmentSignedBatch {
-				batch_size: 3,
-				funding_txid: Txid::from_str(
+			funding_txid: Some(Txid::from_str(
 					"c2d4449afa8d26140898dd54d3390b057ba2a5afcf03ba29d7dc0d8b9ffe966e",
 				)
-				.unwrap(),
-			}),
+				.unwrap()),
 			#[cfg(taproot)]
 			partial_signature_with_nonce: None,
 		};
@@ -5634,7 +5622,7 @@ mod tests {
 		} else {
 			target_value += "0000";
 		}
-		target_value += "002200036e96fe9f8b0ddcd729ba03cfafa5a27b050b39d354dd980814268dfa9a44d4c2"; // batch
+		target_value += "01206e96fe9f8b0ddcd729ba03cfafa5a27b050b39d354dd980814268dfa9a44d4c2"; // funding_txid
 		assert_eq!(encoded_value.as_hex().to_string(), target_value);
 	}