Add Message Handlers for Peer Storage

Introduce message types and handlers to enable the exchange of peer storage data between nodes.
PeerStorageMessage: Used to send encrypted backups to peers.
YourPeerStorageMessage: Used to return a peer's stored data upon reconnection.

- Define two new message types: PeerStorageMessage and YourPeerStorageMessage.
- Implement handlers for these messages in ChannelMessageHandler.
- Add SendPeerStorageMessage and SendYourPeerStorageMessage to MessageSendEvent.

Author: Aditya Sharma

lightning-net-tokio/src/lib.rs

@@ -760,6 +760,11 @@ mod tests {
 		fn handle_tx_init_rbf(&self, _their_node_id: PublicKey, _msg: &TxInitRbf) {}
 		fn handle_tx_ack_rbf(&self, _their_node_id: PublicKey, _msg: &TxAckRbf) {}
 		fn handle_tx_abort(&self, _their_node_id: PublicKey, _msg: &TxAbort) {}
+		fn handle_peer_storage(&self, _their_node_id: PublicKey, _msg: &PeerStorageMessage) {}
+		fn handle_your_peer_storage(
+			&self, _their_node_id: PublicKey, _msg: &YourPeerStorageMessage,
+		) {
+		}
 		fn peer_disconnected(&self, their_node_id: PublicKey) {
 			if their_node_id == self.expected_pubkey {
 				self.disconnected_flag.store(true, Ordering::SeqCst);

lightning/src/events/mod.rs

@@ -2622,6 +2622,23 @@ pub enum MessageSendEvent {
 		/// The gossip_timestamp_filter which should be sent.
 		msg: msgs::GossipTimestampFilter,
 	},
+	/// Sends a channel partner Peer Storage of our backup which they should store.
+	/// This should be sent on each new connection to the channel partner or whenever we want
+	/// them to update the backup that they store.
+	SendPeerStorageMessage {
+		/// The node_id of this message recipient
+		node_id: PublicKey,
+		/// The PeerStorageMessage which should be sent.
+		msg: msgs::PeerStorageMessage,
+	},
+	/// Sends a channel partner their own peer storage which we store and update when they send
+	/// a [`msgs::PeerStorageMessage`].
+	SendYourPeerStorageMessage {
+		/// The node_id of this message recipient
+		node_id: PublicKey,
+		/// The YourPeerStorageMessage which should be sent.
+		msg: msgs::YourPeerStorageMessage,
+	}
 }
 
 /// A trait indicating an object may generate message send events

lightning/src/ln/channelmanager.rs

@@ -8169,6 +8169,10 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		}
 	}
 
+	fn internal_your_peer_storage(&self, _counterparty_node_id: &PublicKey, _msg: &msgs::YourPeerStorageMessage) {}
+
+	fn internal_peer_storage(&self, _counterparty_node_id: &PublicKey, _msg: &msgs::PeerStorageMessage) {}
+
 	fn internal_funding_signed(&self, counterparty_node_id: &PublicKey, msg: &msgs::FundingSigned) -> Result<(), MsgHandleErrInternal> {
 		let best_block = *self.best_block.read().unwrap();
 		let per_peer_state = self.per_peer_state.read().unwrap();
@@ -11352,6 +11356,16 @@ where
 		let _ = handle_error!(self, self.internal_funding_signed(&counterparty_node_id, msg), counterparty_node_id);
 	}
 
+	fn handle_peer_storage(&self, counterparty_node_id: PublicKey, msg: &msgs::PeerStorageMessage) {
+		let _persistence_guard = PersistenceNotifierGuard::optionally_notify(self, || NotifyOption::SkipPersistNoEvents);
+		self.internal_peer_storage(&counterparty_node_id, msg);
+	}
+
+	fn handle_your_peer_storage(&self, counterparty_node_id: PublicKey, msg: &msgs::YourPeerStorageMessage) {
+		let _persistence_guard = PersistenceNotifierGuard::optionally_notify(self, || NotifyOption::SkipPersistNoEvents);
+		self.internal_your_peer_storage(&counterparty_node_id, msg);
+	}
+
 	fn handle_channel_ready(&self, counterparty_node_id: PublicKey, msg: &msgs::ChannelReady) {
 		// Note that we never need to persist the updated ChannelManager for an inbound
 		// channel_ready message - while the channel's state will change, any channel_ready message
@@ -11593,6 +11607,10 @@ where
 						&events::MessageSendEvent::SendShortIdsQuery { .. } => false,
 						&events::MessageSendEvent::SendReplyChannelRange { .. } => false,
 						&events::MessageSendEvent::SendGossipTimestampFilter { .. } => false,
+
+						// Peer Storage
+						&events::MessageSendEvent::SendPeerStorageMessage { .. } => false,
+						&events::MessageSendEvent::SendYourPeerStorageMessage { .. } => false,
 					}
 				});
 				debug_assert!(peer_state.is_connected, "A disconnected peer cannot disconnect");

lightning/src/ln/functional_test_utils.rs

@@ -882,6 +882,12 @@ macro_rules! get_htlc_update_msgs {
 /// such messages are intended to all peers.
 pub fn remove_first_msg_event_to_node(msg_node_id: &PublicKey, msg_events: &mut Vec<MessageSendEvent>) -> MessageSendEvent {
 	let ev_index = msg_events.iter().position(|e| { match e {
+		MessageSendEvent::SendPeerStorageMessage { node_id, .. } => {
+			node_id == msg_node_id
+		},
+		MessageSendEvent::SendYourPeerStorageMessage { node_id, .. } => {
+			node_id == msg_node_id
+		},
 		MessageSendEvent::SendAcceptChannel { node_id, .. } => {
 			node_id == msg_node_id
 		},

lightning/src/ln/msgs.rs

@@ -725,6 +725,24 @@ pub struct UpdateFulfillHTLC {
 	pub payment_preimage: PaymentPreimage,
 }
 
+/// A [`PeerStorage`] message to be sent to or received from a peer.
+///
+/// [`PeerStorage`]: https://github.com/lightning/bolts/pull/1110
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+pub struct PeerStorageMessage {
+	/// Data included in the msg
+	pub data: Vec<u8>,
+}
+
+/// An [`YourPeerStorage`] message to be sent to or received from a peer.
+/// 
+/// [`YourPeerStorage`]: https://github.com/lightning/bolts/pull/1110
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+pub struct YourPeerStorageMessage {
+	/// Data included in the msg
+	pub data: Vec<u8>,
+}
+
 /// An [`update_fail_htlc`] message to be sent to or received from a peer.
 ///
 /// [`update_fail_htlc`]: https://github.com/lightning/bolts/blob/master/02-peer-protocol.md#removing-an-htlc-update_fulfill_htlc-update_fail_htlc-and-update_fail_malformed_htlc
@@ -1507,6 +1525,12 @@ pub trait ChannelMessageHandler : MessageSendEventsProvider {
 	/// Handle an incoming `channel_ready` message from the given peer.
 	fn handle_channel_ready(&self, their_node_id: PublicKey, msg: &ChannelReady);
 
+	// Peer Storage
+	/// Handle an incoming `peer_storage` message from the given peer.
+	fn handle_peer_storage(&self, their_node_id: PublicKey, msg: &PeerStorageMessage);
+	/// Handle an incoming `your_peer_storage` message from the given peer.
+	fn handle_your_peer_storage(&self, their_node_id: PublicKey, msg: &YourPeerStorageMessage);
+
 	// Channel close:
 	/// Handle an incoming `shutdown` message from the given peer.
 	fn handle_shutdown(&self, their_node_id: PublicKey, msg: &Shutdown);
@@ -2628,6 +2652,14 @@ impl_writeable_msg!(UpdateFulfillHTLC, {
 	payment_preimage
 }, {});
 
+impl_writeable_msg!(PeerStorageMessage, {
+	data
+}, {});
+
+impl_writeable_msg!(YourPeerStorageMessage, {
+	data
+}, {});
+
 // Note that this is written as a part of ChannelManager objects, and thus cannot change its
 // serialization format in a way which assumes we know the total serialized length/message end
 // position.
@@ -4527,6 +4559,26 @@ mod tests {
 		assert_eq!(encoded_value, target_value);
 	}
 
+	#[test]
+	fn encoding_peer_storage() {
+		let peerstorage = msgs::PeerStorageMessage {
+			data: <Vec<u8>>::from_hex("01020304050607080910").unwrap()
+		};
+		let encoded_value = peerstorage.encode();
+		let target_value = <Vec<u8>>::from_hex("000a01020304050607080910").unwrap();
+		assert_eq!(encoded_value, target_value);
+	}
+
+	#[test]
+	fn encoding_your_peer_storage() {
+		let yourpeerstorage = msgs::YourPeerStorageMessage {
+			data: <Vec<u8>>::from_hex("01020304050607080910").unwrap()
+		};
+		let encoded_value = yourpeerstorage.encode();
+		let target_value = <Vec<u8>>::from_hex("000a01020304050607080910").unwrap();
+		assert_eq!(encoded_value, target_value);
+	}
+
 	#[test]
 	fn encoding_pong() {
 		let pong = msgs::Pong {

lightning/src/ln/peer_handler.rs

@@ -324,6 +324,8 @@ impl ChannelMessageHandler for ErroringMessageHandler {
 	}
 	// msgs::ChannelUpdate does not contain the channel_id field, so we just drop them.
 	fn handle_channel_update(&self, _their_node_id: PublicKey, _msg: &msgs::ChannelUpdate) {}
+	fn handle_peer_storage(&self, _their_node_id: PublicKey, _msg: &msgs::PeerStorageMessage) {}
+	fn handle_your_peer_storage(&self, _their_node_id: PublicKey, _msg: &msgs::YourPeerStorageMessage) {}
 	fn peer_disconnected(&self, _their_node_id: PublicKey) {}
 	fn peer_connected(&self, _their_node_id: PublicKey, _init: &msgs::Init, _inbound: bool) -> Result<(), ()> { Ok(()) }
 	fn handle_error(&self, _their_node_id: PublicKey, _msg: &msgs::ErrorMessage) {}
@@ -1841,6 +1843,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 			wire::Message::ChannelReady(msg) => {
 				self.message_handler.chan_handler.handle_channel_ready(their_node_id, &msg);
 			},
+			wire::Message::PeerStorageMessage(msg) => {
+				self.message_handler.chan_handler.handle_peer_storage(their_node_id, &msg);
+			},
+			wire::Message::YourPeerStorageMessage(msg) => {
+				self.message_handler.chan_handler.handle_your_peer_storage(their_node_id, &msg);
+			},
 
 			// Quiescence messages:
 			wire::Message::Stfu(msg) => {
@@ -2154,6 +2162,14 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 				// robustly gossip broadcast events even if a peer's message buffer is full.
 				let mut handle_event = |event, from_chan_handler| {
 					match event {
+						MessageSendEvent::SendPeerStorageMessage { ref node_id, ref msg } => {
+							log_debug!(self.logger, "Handling SendPeerStorageMessage event in peer_handler for {}", log_pubkey!(node_id));
+							self.enqueue_message(&mut *get_peer_for_forwarding!(node_id)?, msg);
+						},
+						MessageSendEvent::SendYourPeerStorageMessage { ref node_id, ref msg } => {
+							log_debug!(self.logger, "Handling SendYourPeerStorageMessage event in peer_handler for {}", log_pubkey!(node_id));
+							self.enqueue_message(&mut *get_peer_for_forwarding!(node_id)?, msg);
+						},
 						MessageSendEvent::SendAcceptChannel { ref node_id, ref msg } => {
 							log_debug!(WithContext::from(&self.logger, Some(*node_id), Some(msg.common_fields.temporary_channel_id), None), "Handling SendAcceptChannel event in peer_handler for node {} for channel {}",
 									log_pubkey!(node_id),

lightning/src/ln/wire.rs

@@ -53,6 +53,8 @@ pub(crate) enum Message<T> where T: core::fmt::Debug + Type + TestEq {
 	Warning(msgs::WarningMessage),
 	Ping(msgs::Ping),
 	Pong(msgs::Pong),
+	PeerStorageMessage(msgs::PeerStorageMessage),
+	YourPeerStorageMessage(msgs::YourPeerStorageMessage),
 	OpenChannel(msgs::OpenChannel),
 	OpenChannelV2(msgs::OpenChannelV2),
 	AcceptChannel(msgs::AcceptChannel),
@@ -111,6 +113,8 @@ impl<T> Writeable for Message<T> where T: core::fmt::Debug + Type + TestEq {
 			&Message::Warning(ref msg) => msg.write(writer),
 			&Message::Ping(ref msg) => msg.write(writer),
 			&Message::Pong(ref msg) => msg.write(writer),
+			&Message::PeerStorageMessage(ref msg) => msg.write(writer),
+			&Message::YourPeerStorageMessage(ref msg) => msg.write(writer),
 			&Message::OpenChannel(ref msg) => msg.write(writer),
 			&Message::OpenChannelV2(ref msg) => msg.write(writer),
 			&Message::AcceptChannel(ref msg) => msg.write(writer),
@@ -169,6 +173,8 @@ impl<T> Type for Message<T> where T: core::fmt::Debug + Type + TestEq {
 			&Message::Warning(ref msg) => msg.type_id(),
 			&Message::Ping(ref msg) => msg.type_id(),
 			&Message::Pong(ref msg) => msg.type_id(),
+			&Message::PeerStorageMessage(ref msg) => msg.type_id(),
+			&Message::YourPeerStorageMessage(ref msg) => msg.type_id(),
 			&Message::OpenChannel(ref msg) => msg.type_id(),
 			&Message::OpenChannelV2(ref msg) => msg.type_id(),
 			&Message::AcceptChannel(ref msg) => msg.type_id(),
@@ -261,6 +267,12 @@ fn do_read<R: io::Read, T, H: core::ops::Deref>(buffer: &mut R, message_type: u1
 		msgs::Pong::TYPE => {
 			Ok(Message::Pong(Readable::read(buffer)?))
 		},
+		msgs::PeerStorageMessage::TYPE => {
+			Ok(Message::PeerStorageMessage(Readable::read(buffer)?))
+		},
+		msgs::YourPeerStorageMessage::TYPE => {
+			Ok(Message::YourPeerStorageMessage(Readable::read(buffer)?))
+		},
 		msgs::OpenChannel::TYPE => {
 			Ok(Message::OpenChannel(Readable::read(buffer)?))
 		},
@@ -625,6 +637,14 @@ impl Encode for msgs::GossipTimestampFilter {
 	const TYPE: u16 = 265;
 }
 
+impl Encode for msgs::PeerStorageMessage {
+	const TYPE: u16 = 7;
+}
+
+impl Encode for msgs::YourPeerStorageMessage {
+	const TYPE: u16 = 9;
+}
+
 #[cfg(test)]
 mod tests {
 	use super::*;

lightning/src/util/test_utils.rs

@@ -1087,6 +1087,16 @@ impl msgs::ChannelMessageHandler for TestChannelMessageHandler {
 		self.received_msg(wire::Message::TxAbort(msg.clone()));
 	}
 
+	fn handle_peer_storage(&self, _their_node_id: PublicKey, msg: &msgs::PeerStorageMessage) {
+		self.received_msg(wire::Message::PeerStorageMessage(msg.clone()));
+	}
+
+	fn handle_your_peer_storage(
+		&self, _their_node_id: PublicKey, msg: &msgs::YourPeerStorageMessage,
+	) {
+		self.received_msg(wire::Message::YourPeerStorageMessage(msg.clone()));
+	}
+
 	fn message_received(&self) {}
 }