Merge pull request #2708 from TheBlueMatt/2023-11-less-graph-memory-frag

Reduce common allocations across the codebase

Author: TheBlueMatt

CONTRIBUTING.md

@@ -88,10 +88,11 @@ be covered by functional tests.
 When refactoring, structure your PR to make it easy to review and don't
 hesitate to split it into multiple small, focused PRs.
 
-The Minimum Supported Rust Version (MSRV) currently is 1.41.1 (enforced by
-our GitHub Actions). Also, the compatibility for LDK object serialization is
-currently ensured back to and including crate version 0.0.99 (see the
-[changelog](CHANGELOG.md)).
+The Minimum Supported Rust Version (MSRV) currently is 1.48.0 (enforced by
+our GitHub Actions). We support reading serialized LDK objects written by any
+version of LDK 0.0.99 and above. We support LDK versions 0.0.113 and above
+reading serialized LDK objects written by modern LDK. Any expected issues with
+upgrades or downgrades should be mentioned in the [changelog](CHANGELOG.md).
 
 Commits should cover both the issue fixed and the solution's rationale. These
 [guidelines](https://chris.beams.io/posts/git-commit/) should be kept in mind.

fuzz/src/peer_crypt.rs

@@ -7,7 +7,7 @@
 // You may not use this file except in accordance with one or both of these
 // licenses.
 
-use lightning::ln::peer_channel_encryptor::PeerChannelEncryptor;
+use lightning::ln::peer_channel_encryptor::{PeerChannelEncryptor, MessageBuf};
 use lightning::util::test_utils::TestNodeSigner;
 
 use bitcoin::secp256k1::{Secp256k1, PublicKey, SecretKey};
@@ -74,15 +74,17 @@ pub fn do_test(data: &[u8]) {
 		assert!(crypter.is_ready_for_encryption());
 		crypter
 	};
+	let mut buf = [0; 65536 + 16];
 	loop {
 		if get_slice!(1)[0] == 0 {
-			crypter.encrypt_buffer(get_slice!(slice_to_be16(get_slice!(2))));
+			crypter.encrypt_buffer(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,
 				Err(_) => return,
 			};
-			match crypter.decrypt_message(get_slice!(len as usize + 16)) {
+			buf.copy_from_slice(&get_slice!(len as usize + 16));
+			match crypter.decrypt_message(&mut buf[..len as usize + 16]) {
 				Ok(_) => {},
 				Err(_) => return,
 			}

lightning-net-tokio/src/lib.rs

@@ -422,7 +422,11 @@ const SOCK_WAKER_VTABLE: task::RawWakerVTable =
 	task::RawWakerVTable::new(clone_socket_waker, wake_socket_waker, wake_socket_waker_by_ref, drop_socket_waker);
 
 fn clone_socket_waker(orig_ptr: *const ()) -> task::RawWaker {
-	write_avail_to_waker(orig_ptr as *const mpsc::Sender<()>)
+	let new_waker = unsafe { Arc::from_raw(orig_ptr as *const mpsc::Sender<()>) };
+	let res = write_avail_to_waker(&new_waker);
+	// Don't decrement the refcount when dropping new_waker by turning it back `into_raw`.
+	let _ = Arc::into_raw(new_waker);
+	res
 }
 // When waking, an error should be fine. Most likely we got two send_datas in a row, both of which
 // failed to fully write, but we only need to call write_buffer_space_avail() once. Otherwise, the
@@ -435,29 +439,36 @@ fn wake_socket_waker(orig_ptr: *const ()) {
 }
 fn wake_socket_waker_by_ref(orig_ptr: *const ()) {
 	let sender_ptr = orig_ptr as *const mpsc::Sender<()>;
-	let sender = unsafe { (*sender_ptr).clone() };
+	let sender = unsafe { &*sender_ptr };
 	let _ = sender.try_send(());
 }
 fn drop_socket_waker(orig_ptr: *const ()) {
-	let _orig_box = unsafe { Box::from_raw(orig_ptr as *mut mpsc::Sender<()>) };
-	// _orig_box is now dropped
+	let _orig_arc = unsafe { Arc::from_raw(orig_ptr as *mut mpsc::Sender<()>) };
+	// _orig_arc is now dropped
 }
-fn write_avail_to_waker(sender: *const mpsc::Sender<()>) -> task::RawWaker {
-	let new_box = Box::leak(Box::new(unsafe { (*sender).clone() }));
-	let new_ptr = new_box as *const mpsc::Sender<()>;
+fn write_avail_to_waker(sender: &Arc<mpsc::Sender<()>>) -> task::RawWaker {
+	let new_ptr = Arc::into_raw(Arc::clone(&sender));
 	task::RawWaker::new(new_ptr as *const (), &SOCK_WAKER_VTABLE)
 }
 
 /// The SocketDescriptor used to refer to sockets by a PeerHandler. This is pub only as it is a
 /// type in the template of PeerHandler.
 pub struct SocketDescriptor {
 	conn: Arc<Mutex<Connection>>,
+	// We store a copy of the mpsc::Sender to wake the read task in an Arc here. While we can
+	// simply clone the sender and store a copy in each waker, that would require allocating for
+	// each waker. Instead, we can simply `Arc::clone`, creating a new reference and store the
+	// pointer in the waker.
+	write_avail_sender: Arc<mpsc::Sender<()>>,
 	id: u64,
 }
 impl SocketDescriptor {
 	fn new(conn: Arc<Mutex<Connection>>) -> Self {
-		let id = conn.lock().unwrap().id;
-		Self { conn, id }
+		let (id, write_avail_sender) = {
+			let us = conn.lock().unwrap();
+			(us.id, Arc::new(us.write_avail.clone()))
+		};
+		Self { conn, id, write_avail_sender }
 	}
 }
 impl peer_handler::SocketDescriptor for SocketDescriptor {
@@ -480,7 +491,7 @@ impl peer_handler::SocketDescriptor for SocketDescriptor {
 			let _ = us.read_waker.try_send(());
 		}
 		if data.is_empty() { return 0; }
-		let waker = unsafe { task::Waker::from_raw(write_avail_to_waker(&us.write_avail)) };
+		let waker = unsafe { task::Waker::from_raw(write_avail_to_waker(&self.write_avail_sender)) };
 		let mut ctx = task::Context::from_waker(&waker);
 		let mut written_len = 0;
 		loop {
@@ -522,6 +533,7 @@ impl Clone for SocketDescriptor {
 		Self {
 			conn: Arc::clone(&self.conn),
 			id: self.id,
+			write_avail_sender: Arc::clone(&self.write_avail_sender),
 		}
 	}
 }

lightning/src/blinded_path/utils.rs

@@ -20,7 +20,7 @@ use crate::ln::msgs::DecodeError;
 use crate::ln::onion_utils;
 use crate::onion_message::Destination;
 use crate::util::chacha20poly1305rfc::ChaChaPolyWriteAdapter;
-use crate::util::ser::{Readable, VecWriter, Writeable};
+use crate::util::ser::{Readable, Writeable};
 
 use crate::io;
 use crate::prelude::*;
@@ -129,10 +129,8 @@ where
 
 /// Encrypt TLV payload to be used as a [`crate::blinded_path::BlindedHop::encrypted_payload`].
 fn encrypt_payload<P: Writeable>(payload: P, encrypted_tlvs_rho: [u8; 32]) -> Vec<u8> {
-	let mut writer = VecWriter(Vec::new());
 	let write_adapter = ChaChaPolyWriteAdapter::new(encrypted_tlvs_rho, &payload);
-	write_adapter.write(&mut writer).expect("In-memory writes cannot fail");
-	writer.0
+	write_adapter.encode()
 }
 
 /// Blinded path encrypted payloads may be padded to ensure they are equal length.

lightning/src/ln/channel.rs

@@ -39,7 +39,7 @@ use crate::chain::transaction::{OutPoint, TransactionData};
 use crate::sign::{EcdsaChannelSigner, WriteableEcdsaChannelSigner, EntropySource, ChannelSigner, SignerProvider, NodeSigner, Recipient};
 use crate::events::ClosureReason;
 use crate::routing::gossip::NodeId;
-use crate::util::ser::{Readable, ReadableArgs, Writeable, Writer, VecWriter};
+use crate::util::ser::{Readable, ReadableArgs, Writeable, Writer};
 use crate::util::logger::Logger;
 use crate::util::errors::APIError;
 use crate::util::config::{UserConfig, ChannelConfig, LegacyChannelConfig, ChannelHandshakeConfig, ChannelHandshakeLimits, MaxDustHTLCExposure};
@@ -6893,7 +6893,7 @@ impl<SP: Deref> InboundV1Channel<SP> where SP::Target: SignerProvider {
 }
 
 const SERIALIZATION_VERSION: u8 = 3;
-const MIN_SERIALIZATION_VERSION: u8 = 2;
+const MIN_SERIALIZATION_VERSION: u8 = 3;
 
 impl_writeable_tlv_based_enum!(InboundHTLCRemovalReason,;
 	(0, FailRelay),
@@ -6973,14 +6973,6 @@ impl<SP: Deref> Writeable for Channel<SP> where SP::Target: SignerProvider {
 
 		self.context.latest_monitor_update_id.write(writer)?;
 
-		let mut key_data = VecWriter(Vec::new());
-		// TODO (taproot|arik): Introduce serialization distinction for non-ECDSA signers.
-		self.context.holder_signer.as_ecdsa().expect("Only ECDSA signers may be serialized").write(&mut key_data)?;
-		assert!(key_data.0.len() < core::usize::MAX);
-		assert!(key_data.0.len() < core::u32::MAX as usize);
-		(key_data.0.len() as u32).write(writer)?;
-		writer.write_all(&key_data.0[..])?;
-
 		// Write out the old serialization for shutdown_pubkey for backwards compatibility, if
 		// deserialized from that format.
 		match self.context.shutdown_scriptpubkey.as_ref().and_then(|script| script.as_legacy_pubkey()) {

lightning/src/ln/peer_channel_encryptor.rs

@@ -34,6 +34,10 @@ use core::ops::Deref;
 /// and [BOLT-1](https://github.com/lightning/bolts/blob/master/01-messaging.md#lightning-message-format):
 pub const LN_MAX_MSG_LEN: usize = ::core::u16::MAX as usize; // Must be equal to 65535
 
+/// The (rough) size buffer to pre-allocate when encoding a message. Messages should reliably be
+/// smaller than this size by at least 32 bytes or so.
+pub const MSG_BUF_ALLOC_SIZE: usize = 2048;
+
 // Sha256("Noise_XK_secp256k1_ChaChaPoly_SHA256")
 const NOISE_CK: [u8; 32] = [0x26, 0x40, 0xf5, 0x2e, 0xeb, 0xcd, 0x9e, 0x88, 0x29, 0x58, 0x95, 0x1c, 0x79, 0x42, 0x50, 0xee, 0xdb, 0x28, 0x00, 0x2c, 0x05, 0xd7, 0xdc, 0x2e, 0xa0, 0xf1, 0x95, 0x40, 0x60, 0x42, 0xca, 0xf1];
 // Sha256(NOISE_CK || "lightning")
@@ -165,6 +169,18 @@ impl PeerChannelEncryptor {
 		res.extend_from_slice(&tag);
 	}
 
+	fn decrypt_in_place_with_ad(inout: &mut [u8], n: u64, key: &[u8; 32], h: &[u8]) -> Result<(), LightningError> {
+		let mut nonce = [0; 12];
+		nonce[4..].copy_from_slice(&n.to_le_bytes()[..]);
+
+		let mut chacha = ChaCha20Poly1305RFC::new(key, &nonce, h);
+		let (inout, tag) = inout.split_at_mut(inout.len() - 16);
+		if chacha.check_decrypt_in_place(inout, tag).is_err() {
+			return Err(LightningError{err: "Bad MAC".to_owned(), action: msgs::ErrorAction::DisconnectPeer{ msg: None }});
+		}
+		Ok(())
+	}
+
 	#[inline]
 	fn decrypt_with_ad(res: &mut[u8], n: u64, key: &[u8; 32], h: &[u8], cyphertext: &[u8]) -> Result<(), LightningError> {
 		let mut nonce = [0; 12];
@@ -411,16 +427,20 @@ impl PeerChannelEncryptor {
 		Ok(self.their_node_id.unwrap().clone())
 	}
 
-	/// 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, msg: &[u8]) -> Vec<u8> {
-		if msg.len() > LN_MAX_MSG_LEN {
+	/// 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 effeciency, 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>) {
+		let msg_len = msgbuf.len() - 16 - 2;
+		if msg_len > LN_MAX_MSG_LEN {
 			panic!("Attempted to encrypt message longer than 65535 bytes!");
 		}
 
-		let mut res = Vec::with_capacity(msg.len() + 16*2 + 2);
-		res.resize(msg.len() + 16*2 + 2, 0);
-
 		match self.noise_state {
 			NoiseState::Finished { ref mut sk, ref mut sn, ref mut sck, rk: _, rn: _, rck: _ } => {
 				if *sn >= 1000 {
@@ -430,16 +450,21 @@ impl PeerChannelEncryptor {
 					*sn = 0;
 				}
 
-				Self::encrypt_with_ad(&mut res[0..16+2], *sn, sk, &[0; 0], &(msg.len() as u16).to_be_bytes());
+				Self::encrypt_with_ad(&mut msgbuf[0..16+2], *sn, sk, &[0; 0], &(msg_len as u16).to_be_bytes());
 				*sn += 1;
 
-				Self::encrypt_with_ad(&mut res[16+2..], *sn, sk, &[0; 0], msg);
+				Self::encrypt_in_place_with_ad(msgbuf, 16+2, *sn, sk, &[0; 0]);
 				*sn += 1;
 			},
 			_ => panic!("Tried to encrypt a message prior to noise handshake completion"),
 		}
+	}
 
-		res
+	/// 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.
@@ -448,33 +473,11 @@ impl PeerChannelEncryptor {
 	pub fn encrypt_message<M: wire::Type>(&mut self, message: &M) -> 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(2048));
+		let mut res = VecWriter(Vec::with_capacity(MSG_BUF_ALLOC_SIZE));
 		res.0.resize(16 + 2, 0);
 		wire::write(message, &mut res).expect("In-memory messages must never fail to serialize");
 
-		let msg_len = res.0.len() - 16 - 2;
-		if msg_len > LN_MAX_MSG_LEN {
-			panic!("Attempted to encrypt message longer than 65535 bytes!");
-		}
-
-		match self.noise_state {
-			NoiseState::Finished { ref mut sk, ref mut sn, ref mut sck, rk: _, rn: _, rck: _ } => {
-				if *sn >= 1000 {
-					let (new_sck, new_sk) = hkdf_extract_expand_twice(sck, sk);
-					*sck = new_sck;
-					*sk = new_sk;
-					*sn = 0;
-				}
-
-				Self::encrypt_with_ad(&mut res.0[0..16+2], *sn, sk, &[0; 0], &(msg_len as u16).to_be_bytes());
-				*sn += 1;
-
-				Self::encrypt_in_place_with_ad(&mut res.0, 16+2, *sn, sk, &[0; 0]);
-				*sn += 1;
-			},
-			_ => panic!("Tried to encrypt a message prior to noise handshake completion"),
-		}
-
+		self.encrypt_message_with_header_0s(&mut res.0);
 		res.0
 	}
 
@@ -501,21 +504,20 @@ impl PeerChannelEncryptor {
 		}
 	}
 
-	/// Decrypts the given message.
+	/// Decrypts the given message up to msg.len() - 16. Bytes after msg.len() - 16 will be left
+	/// undefined (as they contain the Poly1305 tag bytes).
+	///
 	/// panics if msg.len() > 65535 + 16
-	pub fn decrypt_message(&mut self, msg: &[u8]) -> Result<Vec<u8>, LightningError> {
+	pub fn decrypt_message(&mut self, msg: &mut [u8]) -> Result<(), LightningError> {
 		if msg.len() > LN_MAX_MSG_LEN + 16 {
 			panic!("Attempted to decrypt message longer than 65535 + 16 bytes!");
 		}
 
 		match self.noise_state {
 			NoiseState::Finished { sk: _, sn: _, sck: _, ref rk, ref mut rn, rck: _ } => {
-				let mut res = Vec::with_capacity(msg.len() - 16);
-				res.resize(msg.len() - 16, 0);
-				Self::decrypt_with_ad(&mut res[..], *rn, rk, &[0; 0], msg)?;
+				Self::decrypt_in_place_with_ad(&mut msg[..], *rn, rk, &[0; 0])?;
 				*rn += 1;
-
-				Ok(res)
+				Ok(())
 			},
 			_ => panic!("Tried to decrypt a message prior to noise handshake completion"),
 		}
@@ -542,9 +544,30 @@ impl PeerChannelEncryptor {
 	}
 }
 
+/// A buffer which stores an encoded message (including the two message-type bytes) with some
+/// padding to allow for future encryption/MACing.
+pub struct MessageBuf(Vec<u8>);
+impl MessageBuf {
+	/// Creates a new buffer from an encoded message (i.e. the two message-type bytes followed by
+	/// the message contents).
+	///
+	/// Panics if the message is longer than 2^16.
+	pub fn from_encoded(encoded_msg: &[u8]) -> Self {
+		if encoded_msg.len() > LN_MAX_MSG_LEN {
+			panic!("Attempted to encrypt message longer than 65535 bytes!");
+		}
+		// In addition to the message (continaing the two message type bytes), we also have to add
+		// the message length header (and its MAC) and the message MAC.
+		let mut res = Vec::with_capacity(encoded_msg.len() + 16*2 + 2);
+		res.resize(encoded_msg.len() + 16 + 2, 0);
+		res[16 + 2..].copy_from_slice(&encoded_msg);
+		Self(res)
+	}
+}
+
 #[cfg(test)]
 mod tests {
-	use super::LN_MAX_MSG_LEN;
+	use super::{MessageBuf, LN_MAX_MSG_LEN};
 
 	use bitcoin::secp256k1::{PublicKey, SecretKey};
 	use bitcoin::secp256k1::Secp256k1;
@@ -760,12 +783,11 @@ mod tests {
 
 		for i in 0..1005 {
 			let msg = [0x68, 0x65, 0x6c, 0x6c, 0x6f];
-			let res = outbound_peer.encrypt_buffer(&msg);
+			let mut res = outbound_peer.encrypt_buffer(MessageBuf::from_encoded(&msg));
 			assert_eq!(res.len(), 5 + 2*16 + 2);
 
 			let len_header = res[0..2+16].to_vec();
 			assert_eq!(inbound_peer.decrypt_length_header(&len_header[..]).unwrap() as usize, msg.len());
-			assert_eq!(inbound_peer.decrypt_message(&res[2+16..]).unwrap()[..], msg[..]);
 
 			if i == 0 {
 				assert_eq!(res, hex::decode("cf2b30ddf0cf3f80e7c35a6e6730b59fe802473180f396d88a8fb0db8cbcf25d2f214cf9ea1d95").unwrap());
@@ -780,6 +802,9 @@ mod tests {
 			} else if i == 1001 {
 				assert_eq!(res, hex::decode("2ecd8c8a5629d0d02ab457a0fdd0f7b90a192cd46be5ecb6ca570bfc5e268338b1a16cf4ef2d36").unwrap());
 			}
+
+			inbound_peer.decrypt_message(&mut res[2+16..]).unwrap();
+			assert_eq!(res[2 + 16..res.len() - 16], msg[..]);
 		}
 	}
 
@@ -794,7 +819,7 @@ 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(&msg);
+		outbound_peer.encrypt_buffer(MessageBuf::from_encoded(&msg));
 	}
 
 	#[test]
@@ -803,7 +828,7 @@ mod tests {
 		let mut inbound_peer = get_inbound_peer_for_test_vectors();
 
 		// MSG should not exceed LN_MAX_MSG_LEN + 16
-		let msg = [4u8; LN_MAX_MSG_LEN + 17];
-		inbound_peer.decrypt_message(&msg).unwrap();
+		let mut msg = [4u8; LN_MAX_MSG_LEN + 17];
+		inbound_peer.decrypt_message(&mut msg).unwrap();
 	}
 }