Batch commitment_signed messages in PeerManager

During splicing, commitment_signed messages need to be collected into a
single batch before they are handled. Rather than including this as part
of the channel state machine logic, batch when reading messages from the
wire since they can be considered one logical message.

Author: jkczyz

lightning/src/ln/peer_handler.rs

@@ -15,6 +15,7 @@
 //! call into the provided message handlers (probably a ChannelManager and P2PGossipSync) with
 //! messages they should handle, and encoding/sending response messages.
 
+use bitcoin::Txid;
 use bitcoin::constants::ChainHash;
 use bitcoin::secp256k1::{self, Secp256k1, SecretKey, PublicKey};
 
@@ -41,6 +42,8 @@ use crate::util::string::PrintableString;
 #[allow(unused_imports)]
 use crate::prelude::*;
 
+use alloc::collections::{btree_map, BTreeMap};
+
 use crate::io;
 use crate::sync::{Mutex, MutexGuard, FairRwLock};
 use core::sync::atomic::{AtomicBool, AtomicU32, AtomicI32, Ordering};
@@ -608,6 +611,8 @@ struct Peer {
 	received_channel_announce_since_backlogged: bool,
 
 	inbound_connection: bool,
+
+	commitment_signed_batch: Option<(ChannelId, BTreeMap<Txid, msgs::CommitmentSigned>)>,
 }
 
 impl Peer {
@@ -860,6 +865,7 @@ pub struct PeerManager<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: D
 
 enum LogicalMessage<T: core::fmt::Debug + wire::Type + wire::TestEq> {
 	FromWire(wire::Message<T>),
+	CommitmentSignedBatch(ChannelId, BTreeMap<Txid, msgs::CommitmentSigned>),
 }
 
 enum MessageHandlingError {
@@ -1144,6 +1150,8 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 
 					received_channel_announce_since_backlogged: false,
 					inbound_connection: false,
+
+					commitment_signed_batch: None,
 				}));
 				Ok(res)
 			}
@@ -1200,6 +1208,8 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 
 					received_channel_announce_since_backlogged: false,
 					inbound_connection: true,
+
+					commitment_signed_batch: None,
 				}));
 				Ok(())
 			}
@@ -1653,6 +1663,11 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 			Some(LogicalMessage::FromWire(message)) => {
 				self.do_handle_message_without_peer_lock(peer_mutex, message, their_node_id, &logger)
 			},
+			Some(LogicalMessage::CommitmentSignedBatch(channel_id, batch)) => {
+				log_trace!(logger, "Received commitment_signed batch {:?} from {}", batch, log_pubkey!(their_node_id));
+				self.message_handler.chan_handler.handle_commitment_signed_batch(their_node_id, channel_id, batch);
+				return Ok(None);
+			},
 			None => Ok(None),
 		}
 	}
@@ -1747,6 +1762,51 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 			return Err(PeerHandleError { }.into());
 		}
 
+		// During splicing, commitment_signed messages need to be collected into a single batch
+		// before they are handled.
+		if let wire::Message::CommitmentSigned(msg) = message {
+			if let Some(ref batch) = msg.batch {
+				let (channel_id, buffer) = peer_lock
+					.commitment_signed_batch
+					.get_or_insert_with(|| (msg.channel_id, BTreeMap::new()));
+
+				if msg.channel_id != *channel_id {
+					log_debug!(logger, "Peer {} sent batched commitment_signed for the wrong channel (expected: {}, actual: {})", log_pubkey!(their_node_id), channel_id, &msg.channel_id);
+					return Err(PeerHandleError { }.into());
+				}
+
+				const COMMITMENT_SIGNED_BATCH_LIMIT: usize = 100;
+				if buffer.len() == COMMITMENT_SIGNED_BATCH_LIMIT {
+					log_debug!(logger, "Peer {} sent batched commitment_signed for channel {} exceeding the limit", log_pubkey!(their_node_id), channel_id);
+					return Err(PeerHandleError { }.into());
+				}
+
+				let batch_size = batch.batch_size as usize;
+				match buffer.entry(batch.funding_txid) {
+					btree_map::Entry::Vacant(entry) => { entry.insert(msg); },
+					btree_map::Entry::Occupied(_) => {
+						log_debug!(logger, "Peer {} sent batched commitment_signed with duplicate funding_txid {} for channel {}", log_pubkey!(their_node_id), channel_id, &batch.funding_txid);
+						return Err(PeerHandleError { }.into());
+					}
+				}
+
+				if buffer.len() >= batch_size {
+					let (channel_id, batch) = peer_lock.commitment_signed_batch.take().expect("batch should have been inserted");
+					return Ok(Some(LogicalMessage::CommitmentSignedBatch(channel_id, batch)));
+				} else {
+					return Ok(None);
+				}
+			} else if peer_lock.commitment_signed_batch.is_some() {
+				log_debug!(logger, "Peer {} sent non-batched commitment_signed for channel {} when expecting batched commitment_signed", log_pubkey!(their_node_id), &msg.channel_id);
+				return Err(PeerHandleError { }.into());
+			} else {
+				return Ok(Some(LogicalMessage::FromWire(wire::Message::CommitmentSigned(msg))));
+			}
+		} else if peer_lock.commitment_signed_batch.is_some() {
+			log_debug!(logger, "Peer {} sent non-commitment_signed message when expecting batched commitment_signed", log_pubkey!(their_node_id));
+			return Err(PeerHandleError { }.into());
+		}
+
 		if let wire::Message::GossipTimestampFilter(_msg) = message {
 			// When supporting gossip messages, start initial gossip sync only after we receive
 			// a GossipTimestampFilter