Fail channel for batched commitment_signed appropriately

If a message in a commitment_signed batch does not contain a
funding_txid or has duplicates, the channel should be failed. Move this
check from PeerManager to FundedChannel such that this can be done.

Author: jkczyz

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::*;
@@ -5950,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

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/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;
@@ -1952,8 +1950,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);
@@ -1967,15 +1964,7 @@ pub trait ChannelMessageHandler: BaseMessageHandler {
 			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(|cs| {
-					let funding_txid = cs.funding_txid.unwrap();
-					(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());
 		}
 	}
 

lightning/src/ln/peer_handler.rs

@@ -15,7 +15,6 @@
 //! 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};
 
@@ -42,8 +41,6 @@ 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};
@@ -335,8 +332,7 @@ impl ChannelMessageHandler for ErroringMessageHandler {
 		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>,
+		&self, their_node_id: PublicKey, channel_id: ChannelId, _batch: Vec<msgs::CommitmentSigned>,
 	) {
 		ErroringMessageHandler::push_error(self, their_node_id, channel_id);
 	}
@@ -551,8 +547,8 @@ struct MessageBatch {
 
 /// The representation of the message batch, which may different for each message type.
 enum MessageBatchImpl {
-	/// A batch of `commitment_signed` messages, where each has a unique `funding_txid`.
-	CommitmentSigned(BTreeMap<Txid, msgs::CommitmentSigned>),
+	/// A batch of `commitment_signed` messages used when there are pending splices.
+	CommitmentSigned(Vec<msgs::CommitmentSigned>),
 }
 
 /// The ratio between buffer sizes at which we stop sending initial sync messages vs when we stop
@@ -889,7 +885,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>),
+	CommitmentSignedBatch(ChannelId, Vec<msgs::CommitmentSigned>),
 }
 
 enum MessageHandlingError {
@@ -1836,7 +1832,8 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 
 			let messages = match msg.message_type {
 				Some(message_type) if message_type == msgs::CommitmentSigned::TYPE => {
-					MessageBatchImpl::CommitmentSigned(BTreeMap::new())
+					let messages = Vec::with_capacity(batch_size);
+					MessageBatchImpl::CommitmentSigned(messages)
 				},
 				_ => {
 					let error = format!("Peer {} sent start_batch for channel {} without a known message type", log_pubkey!(their_node_id), &msg.channel_id);
@@ -1865,7 +1862,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 
 		if let wire::Message::CommitmentSigned(msg) = message {
 			if let Some(message_batch) = &mut peer_lock.message_batch {
-				let buffer = match &mut message_batch.messages {
+				let messages = match &mut message_batch.messages {
 					MessageBatchImpl::CommitmentSigned(ref mut messages) => messages,
 				};
 
@@ -1883,23 +1880,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 					}.into());
 				}
 
-				let funding_txid = match msg.funding_txid {
-					Some(funding_txid) => funding_txid,
-					None => {
-						log_debug!(logger, "Peer {} sent batched commitment_signed without a funding_txid for channel {}", log_pubkey!(their_node_id), message_batch.channel_id);
-						return Err(PeerHandleError { }.into());
-					},
-				};
-
-				match buffer.entry(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), funding_txid, message_batch.channel_id);
-						return Err(PeerHandleError { }.into());
-					}
-				}
+				messages.push(msg);
 
-				if buffer.len() == message_batch.batch_size {
+				if messages.len() == message_batch.batch_size {
 					let MessageBatch { channel_id, batch_size: _, messages } = peer_lock.message_batch.take().expect("batch should have been inserted");
 					let batch = match messages {
 						MessageBatchImpl::CommitmentSigned(messages) => messages,

lightning/src/util/test_utils.rs

@@ -79,8 +79,6 @@ 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};
@@ -1057,7 +1055,7 @@ impl msgs::ChannelMessageHandler for TestChannelMessageHandler {
 	}
 	fn handle_commitment_signed_batch(
 		&self, _their_node_id: PublicKey, _channel_id: ChannelId,
-		_batch: BTreeMap<Txid, msgs::CommitmentSigned>,
+		_batch: Vec<msgs::CommitmentSigned>,
 	) {
 		unreachable!()
 	}