Handle batched commitment_signed messages

Update commitment_signed message to contain the funding_txid instead of
both that and a batch_size. The spec was updated to batch messages using
start_batch, which contains the batch_size. This commit also updates
PeerManager to batch commitment_signed messages in this manner instead
of the previous custom approach.

Author: jkczyz

lightning/src/ln/channel.rs

@@ -4978,7 +4978,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,
 		})
@@ -5948,10 +5948,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");
 
@@ -9349,20 +9345,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/dual_funding_tests.rs

@@ -185,7 +185,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

@@ -807,15 +807,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
@@ -827,8 +818,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>,
@@ -1986,15 +1977,14 @@ 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;
+				.map(|cs| {
+					let funding_txid = cs.funding_txid.unwrap();
 					(funding_txid, cs)
 				})
 				.collect();
@@ -2768,18 +2758,14 @@ 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),
+	// TOOD(splicing): Change this to 1 once the spec is finalized
+	(1001, funding_txid, option),
 });
 
 #[cfg(taproot)]
@@ -2788,8 +2774,9 @@ impl_writeable_msg!(CommitmentSigned, {
 	signature,
 	htlc_signatures
 }, {
-	(0, batch, option),
 	(2, partial_signature_with_nonce, option),
+	// TOOD(splicing): Change this to 1 and reorder once the spec is finalized
+	(1001, funding_txid, option),
 });
 
 impl_writeable!(DecodedOnionErrorPacket, {
@@ -5649,13 +5636,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(
-					"c2d4449afa8d26140898dd54d3390b057ba2a5afcf03ba29d7dc0d8b9ffe966e",
-				)
-				.unwrap(),
-			}),
+			funding_txid: Some(
+				Txid::from_str("c2d4449afa8d26140898dd54d3390b057ba2a5afcf03ba29d7dc0d8b9ffe966e")
+					.unwrap(),
+			),
 			#[cfg(taproot)]
 			partial_signature_with_nonce: None,
 		};
@@ -5666,7 +5650,9 @@ mod tests {
 		} else {
 			target_value += "0000";
 		}
-		target_value += "002200036e96fe9f8b0ddcd729ba03cfafa5a27b050b39d354dd980814268dfa9a44d4c2"; // batch
+		target_value += "fd03e9"; // Type (funding_txid)
+		target_value += "20"; // Length (funding_txid)
+		target_value += "6e96fe9f8b0ddcd729ba03cfafa5a27b050b39d354dd980814268dfa9a44d4c2"; // Value
 		assert_eq!(encoded_value.as_hex().to_string(), target_value);
 	}
 

lightning/src/ln/peer_handler.rs

@@ -620,7 +620,7 @@ struct Peer {
 
 	inbound_connection: bool,
 
-	commitment_signed_batch: Option<(ChannelId, BTreeMap<Txid, msgs::CommitmentSigned>)>,
+	commitment_signed_batch: Option<(ChannelId, usize, BTreeMap<Txid, msgs::CommitmentSigned>)>,
 }
 
 impl Peer {
@@ -1772,41 +1772,95 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 
 		// 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 let wire::Message::StartBatch(msg) = message {
+			if peer_lock.commitment_signed_batch.is_some() {
+				let error = format!("Peer {} sent start_batch for channel {} before previous batch completed", log_pubkey!(their_node_id), &msg.channel_id);
+				log_debug!(logger, "{}", error);
+				return Err(LightningError {
+					err: error.clone(),
+					action: msgs::ErrorAction::DisconnectPeerWithWarning {
+						msg: msgs::WarningMessage {
+							channel_id: msg.channel_id,
+							data: error,
+						},
+					},
+				}.into());
+			}
+
+			let batch_size = msg.batch_size as usize;
+			if batch_size <= 1 {
+				let error = format!("Peer {} sent start_batch for channel {} not strictly greater than 1", log_pubkey!(their_node_id), &msg.channel_id);
+				log_debug!(logger, "{}", error);
+				return Err(LightningError {
+					err: error.clone(),
+					action: msgs::ErrorAction::SendWarningMessage {
+						msg: msgs::WarningMessage {
+							channel_id: msg.channel_id,
+							data: error,
+						},
+						log_level: Level::Debug,
+					},
+				}.into());
+			}
 
+			const COMMITMENT_SIGNED_BATCH_LIMIT: usize = 20;
+			if batch_size > COMMITMENT_SIGNED_BATCH_LIMIT {
+				let error = format!("Peer {} sent start_batch for channel {} exceeding the limit", log_pubkey!(their_node_id), &msg.channel_id);
+				log_debug!(logger, "{}", error);
+				return Err(LightningError {
+					err: error.clone(),
+					action: msgs::ErrorAction::DisconnectPeerWithWarning {
+						msg: msgs::WarningMessage {
+							channel_id: msg.channel_id,
+							data: error,
+						},
+					},
+				}.into());
+			}
+
+			peer_lock.commitment_signed_batch = Some((msg.channel_id, batch_size, BTreeMap::new()));
+
+			return Ok(None);
+		}
+
+		if let wire::Message::CommitmentSigned(msg) = message {
+			if let Some((channel_id, batch_size, buffer)) = &mut peer_lock.commitment_signed_batch {
 				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());
+					let error = format!("Peer {} sent batched commitment_signed for the wrong channel (expected: {}, actual: {})", log_pubkey!(their_node_id), channel_id, &msg.channel_id);
+					log_debug!(logger, "{}", error);
+					return Err(LightningError {
+						err: error.clone(),
+						action: msgs::ErrorAction::DisconnectPeerWithWarning {
+							msg: msgs::WarningMessage {
+								channel_id: msg.channel_id,
+								data: error,
+							},
+						},
+					}.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 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), channel_id);
+						return Err(PeerHandleError { }.into());
+					},
+				};
 
-				let batch_size = batch.batch_size as usize;
-				match buffer.entry(batch.funding_txid) {
+				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), channel_id, &batch.funding_txid);
+						log_debug!(logger, "Peer {} sent batched commitment_signed with duplicate funding_txid {} for channel {}", log_pubkey!(their_node_id), funding_txid, channel_id);
 						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");
+				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))));
 			}
@@ -2407,6 +2461,13 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 							if let &Some(ref msg) = update_fee {
 								self.enqueue_message(&mut *peer, msg);
 							}
+							if commitment_signed.len() > 1 {
+								let msg = msgs::StartBatch {
+									channel_id: *channel_id,
+									batch_size: commitment_signed.len() as u16,
+								};
+								self.enqueue_message(&mut *peer, &msg);
+							}
 							for msg in commitment_signed {
 								self.enqueue_message(&mut *peer, msg);
 							}

lightning/src/ln/update_fee_tests.rs

@@ -520,7 +520,7 @@ pub fn do_test_update_fee_that_funder_cannot_afford(channel_type_features: Chann
 		channel_id: chan.2,
 		signature: res.0,
 		htlc_signatures: res.1,
-		batch: None,
+		funding_txid: None,
 		#[cfg(taproot)]
 		partial_signature_with_nonce: None,
 	};
@@ -621,7 +621,7 @@ pub fn test_update_fee_that_saturates_subs() {
 		channel_id: chan_id,
 		signature: res.0,
 		htlc_signatures: res.1,
-		batch: None,
+		funding_txid: None,
 		#[cfg(taproot)]
 		partial_signature_with_nonce: None,
 	};