Modify Peer to use ChunkedMessageQueue

Replace `pending_outbound_buffer`,
`pending_outbound_buffer_first_msg_offset`, and
`gossip_broadcast_buffer` fields in `Peer` with a single `message_queue:
ChunkedMessageQueue`.

Update `should_read` to check `message_queue.pending_bytes()` against
the buffer limit. Update `should_buffer_onion_message` and
`should_buffer_gossip_broadcast` to drop the
`pending_outbound_buffer.is_empty()` check (the contiguous chunk buffer
no longer has a meaningful "queue empty" concept). Update
`buffer_full_drop_gossip_broadcast` to use
`message_queue.total_buffered_bytes()`.

Rewrite `do_attempt_write_data` to: send full chunks first, then buffer
messages (onion -> gossip broadcast -> backfill), then pad+finalize any
partial chunk with Ping padding before sending. During handshake, send
raw bytes without chunking.

Update `enqueue_message` to use `encrypt_and_push_message`. Update
handshake act handling to use `push_raw`. Update gossip broadcast sites
to use `message_queue.gossip_broadcast_buffer`.

Co-Authored-By: HAL 9000

Author: tnull

fuzz/src/full_stack.rs

@@ -47,7 +47,7 @@ use lightning::ln::functional_test_utils::*;
 use lightning::ln::inbound_payment::ExpandedKey;
 use lightning::ln::outbound_payment::{RecipientOnionFields, Retry};
 use lightning::ln::peer_handler::{
-	IgnoringMessageHandler, MessageHandler, PeerManager, SocketDescriptor,
+	IgnoringMessageHandler, MessageHandler, PeerManager, SocketDescriptor, CHUNK_SIZE,
 };
 use lightning::ln::script::ShutdownScript;
 use lightning::ln::types::ChannelId;
@@ -201,6 +201,15 @@ struct Peer<'a> {
 }
 impl<'a> SocketDescriptor for Peer<'a> {
 	fn send_data(&mut self, data: &[u8], _continue_read: bool) -> usize {
+		// After the Noise handshake, all writes must be exactly CHUNK_SIZE bytes
+		// (or 0 for empty force-writes). The only non-chunk write is 50 bytes for
+		// the handshake act_two sent by an inbound peer before encryption is ready.
+		assert!(
+			data.len() == 0 || data.len() == 50 || data.len() == CHUNK_SIZE,
+			"Unexpected send_data size: {} (expected 0, 50, or {})",
+			data.len(),
+			CHUNK_SIZE,
+		);
 		data.len()
 	}
 	fn disconnect_socket(&mut self) {

fuzz/src/peer_crypt.rs

@@ -79,9 +79,11 @@ pub fn do_test(data: &[u8]) {
 	let mut buf = [0; 65536 + 16];
 	loop {
 		if get_slice!(1)[0] == 0 {
-			crypter.encrypt_buffer(MessageBuf::from_encoded(&get_slice!(slice_to_be16(
-				get_slice!(2)
-			))));
+			let mut dest = Vec::new();
+			crypter.encrypt_buffer_into(
+				&mut dest,
+				MessageBuf::from_encoded(&get_slice!(slice_to_be16(get_slice!(2)))),
+			);
 		} else {
 			let len = match crypter.decrypt_length_header(get_slice!(16 + 2)) {
 				Ok(len) => len,

lightning/src/ln/peer_channel_encryptor.rs

@@ -28,7 +28,7 @@ use bitcoin::secp256k1::{PublicKey, SecretKey};
 
 use crate::crypto::chacha20poly1305rfc::ChaCha20Poly1305RFC;
 use crate::crypto::utils::hkdf_extract_expand_twice;
-use crate::util::ser::{VecWriter, Writer};
+use crate::util::ser::Writer;
 
 /// Maximum Lightning message data length according to
 /// [BOLT-8](https://github.com/lightning/bolts/blob/v1.0/08-transport.md#lightning-message-specification)
@@ -540,41 +540,6 @@ impl PeerChannelEncryptor {
 		}
 	}
 
-	/// Builds sendable bytes for a message.
-	///
-	/// `msgbuf` must begin with 16 + 2 dummy/0 bytes, which will be filled with the encrypted
-	/// message length and its MAC. It should then be followed by the message bytes themselves
-	/// (including the two byte message type).
-	///
-	/// For efficiency, the [`Vec::capacity`] should be at least 16 bytes larger than the
-	/// [`Vec::len`], to avoid reallocating for the message MAC, which will be appended to the vec.
-	fn encrypt_message_with_header_0s(&mut self, msgbuf: &mut Vec<u8>) {
-		self.encrypt_with_header_0s_at(msgbuf, 0);
-	}
-
-	/// Encrypts the given pre-serialized message, returning the encrypted version.
-	/// panics if msg.len() > 65535 or Noise handshake has not finished.
-	pub fn encrypt_buffer(&mut self, mut msg: MessageBuf) -> Vec<u8> {
-		self.encrypt_message_with_header_0s(&mut msg.0);
-		msg.0
-	}
-
-	/// Encrypts the given message, returning the encrypted version.
-	/// panics if the length of `message`, once encoded, is greater than 65535 or if the Noise
-	/// handshake has not finished.
-	pub fn encrypt_message<T: wire::Type>(&mut self, message: wire::Message<T>) -> Vec<u8> {
-		// Allocate a buffer with 2KB, fitting most common messages. Reserve the first 16+2 bytes
-		// for the 2-byte message type prefix and its MAC.
-		let mut res = VecWriter(Vec::with_capacity(MSG_BUF_ALLOC_SIZE));
-		res.0.resize(16 + 2, 0);
-
-		message.type_id().write(&mut res).expect("In-memory messages must never fail to serialize");
-		message.write(&mut res).expect("In-memory messages must never fail to serialize");
-
-		self.encrypt_message_with_header_0s(&mut res.0);
-		res.0
-	}
-
 	/// Encrypts the given message directly into the destination buffer, appending encrypted bytes.
 	///
 	/// This avoids the intermediate `Vec` allocation that [`Self::encrypt_message`] creates.
@@ -1056,7 +1021,8 @@ mod tests {
 
 		for i in 0..1005 {
 			let msg = [0x68, 0x65, 0x6c, 0x6c, 0x6f];
-			let mut res = outbound_peer.encrypt_buffer(MessageBuf::from_encoded(&msg));
+			let mut res = Vec::new();
+			outbound_peer.encrypt_buffer_into(&mut res, MessageBuf::from_encoded(&msg));
 			assert_eq!(res.len(), 5 + 2 * 16 + 2);
 
 			let len_header = res[0..2 + 16].to_vec();
@@ -1101,7 +1067,8 @@ mod tests {
 	fn max_message_len_encryption() {
 		let mut outbound_peer = get_outbound_peer_for_initiator_test_vectors();
 		let msg = [4u8; LN_MAX_MSG_LEN + 1];
-		outbound_peer.encrypt_buffer(MessageBuf::from_encoded(&msg));
+		let mut dest = Vec::new();
+		outbound_peer.encrypt_buffer_into(&mut dest, MessageBuf::from_encoded(&msg));
 	}
 
 	#[test]