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.
PeerStorageRetrievalMessage: Used to return a peer's stored data upon reconnection.

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

Author: Aditya Sharma

lightning-net-tokio/src/lib.rs

@@ -761,6 +761,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: PeerStorage) {}
+		fn handle_peer_storage_retrieval(
+			&self, _their_node_id: PublicKey, _msg: PeerStorageRetrieval,
+		) {
+		}
 		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

@@ -2624,6 +2624,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.
+	SendPeerStorage {
+		/// The node_id of this message recipient
+		node_id: PublicKey,
+		/// The peer_storage which should be sent.
+		msg: msgs::PeerStorage,
+	},
+	/// Sends a channel partner their own peer storage which we store and update when they send
+	/// a [`msgs::PeerStorage`].
+	SendPeerStorageRetrieval {
+		/// The node_id of this message recipient
+		node_id: PublicKey,
+		/// The peer_storage_retrieval which should be sent.
+		msg: msgs::PeerStorageRetrieval,
+	}
 }
 
 /// A trait indicating an object may generate message send events

lightning/src/ln/channelmanager.rs

@@ -8221,6 +8221,10 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		}
 	}
 
+	fn internal_peer_storage_retrieval(&self, _counterparty_node_id: PublicKey, _msg: msgs::PeerStorageRetrieval) {}
+
+	fn internal_peer_storage(&self, _counterparty_node_id: PublicKey, _msg: msgs::PeerStorage) {}
+
 	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();
@@ -11417,6 +11421,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::PeerStorage) {
+		let _persistence_guard = PersistenceNotifierGuard::optionally_notify(self, || NotifyOption::SkipPersistNoEvents);
+		self.internal_peer_storage(counterparty_node_id, msg);
+	}
+
+	fn handle_peer_storage_retrieval(&self, counterparty_node_id: PublicKey, msg: msgs::PeerStorageRetrieval) {
+		let _persistence_guard = PersistenceNotifierGuard::optionally_notify(self, || NotifyOption::SkipPersistNoEvents);
+		self.internal_peer_storage_retrieval(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
@@ -11658,6 +11672,10 @@ where
 						&events::MessageSendEvent::SendShortIdsQuery { .. } => false,
 						&events::MessageSendEvent::SendReplyChannelRange { .. } => false,
 						&events::MessageSendEvent::SendGossipTimestampFilter { .. } => false,
+
+						// Peer Storage
+						&events::MessageSendEvent::SendPeerStorage { .. } => false,
+						&events::MessageSendEvent::SendPeerStorageRetrieval { .. } => false,
 					}
 				});
 				debug_assert!(peer_state.is_connected, "A disconnected peer cannot disconnect");

lightning/src/ln/functional_test_utils.rs

@@ -880,6 +880,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::SendPeerStorage { node_id, .. } => {
+			node_id == msg_node_id
+		},
+		MessageSendEvent::SendPeerStorageRetrieval { node_id, .. } => {
+			node_id == msg_node_id
+		},
 		MessageSendEvent::SendAcceptChannel { node_id, .. } => {
 			node_id == msg_node_id
 		},

lightning/src/ln/msgs.rs

@@ -726,6 +726,36 @@ pub struct UpdateFulfillHTLC {
 	pub payment_preimage: PaymentPreimage,
 }
 
+/// A [`peer_storage`] message that can be sent to or received from a peer.
+///
+/// This message is used to distribute backup data to peers.
+/// If data is lost or corrupted, users can retrieve it through [`PeerStorageRetrieval`]  
+/// to recover critical information, such as channel states, for fund recovery.
+///
+/// [`peer_storage`] is used to send our own encrypted backup data to a peer.
+///
+/// [`peer_storage`]: https://github.com/lightning/bolts/pull/1110
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+pub struct PeerStorage {
+	/// Our encrypted backup data included in the msg.
+	pub data: Vec<u8>,
+}
+
+/// A [`peer_storage_retrieval`] message that can be sent to or received from a peer.
+///
+/// This message is sent to peers for whom we store backup data.
+/// If we receive this message, it indicates that the peer had stored our backup data.  
+/// This data can be used for fund recovery in case of data loss.
+///
+/// [`peer_storage_retrieval`] is used to send the most recent backup of the peer.
+///
+/// [`peer_storage_retrieval`]: https://github.com/lightning/bolts/pull/1110
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+pub struct PeerStorageRetrieval {
+	/// Most recent peer's 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
@@ -1508,6 +1538,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: PeerStorage);
+	/// Handle an incoming `peer_storage_retrieval` message from the given peer.
+	fn handle_peer_storage_retrieval(&self, their_node_id: PublicKey, msg: PeerStorageRetrieval);
+
 	// Channel close:
 	/// Handle an incoming `shutdown` message from the given peer.
 	fn handle_shutdown(&self, their_node_id: PublicKey, msg: &Shutdown);
@@ -2634,6 +2670,14 @@ impl_writeable_msg!(UpdateFulfillHTLC, {
 	payment_preimage
 }, {});
 
+impl_writeable_msg!(PeerStorage, {
+	data
+}, {});
+
+impl_writeable_msg!(PeerStorageRetrieval, {
+	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.
@@ -4536,6 +4580,26 @@ mod tests {
 		assert_eq!(encoded_value, target_value);
 	}
 
+	#[test]
+	fn encoding_peer_storage() {
+		let peer_storage = msgs::PeerStorage {
+			data: <Vec<u8>>::from_hex("01020304050607080910").unwrap()
+		};
+		let encoded_value = peer_storage.encode();
+		let target_value = <Vec<u8>>::from_hex("000a01020304050607080910").unwrap();
+		assert_eq!(encoded_value, target_value);
+	}
+
+	#[test]
+	fn encoding_peer_storage_retrieval() {
+		let peer_storage_retrieval = msgs::PeerStorageRetrieval {
+			data: <Vec<u8>>::from_hex("01020304050607080910").unwrap()
+		};
+		let encoded_value = peer_storage_retrieval.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

@@ -327,6 +327,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::PeerStorage) {}
+	fn handle_peer_storage_retrieval(&self, _their_node_id: PublicKey, _msg: msgs::PeerStorageRetrieval) {}
 	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) {}
@@ -1850,6 +1852,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::PeerStorage(msg) => {
+				self.message_handler.chan_handler.handle_peer_storage(their_node_id, msg);
+			},
+			wire::Message::PeerStorageRetrieval(msg) => {
+				self.message_handler.chan_handler.handle_peer_storage_retrieval(their_node_id, msg);
+			},
 
 			// Quiescence messages:
 			wire::Message::Stfu(msg) => {
@@ -2163,6 +2171,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::SendPeerStorage { ref node_id, ref msg } => {
+							log_debug!(self.logger, "Handling SendPeerStorage event in peer_handler for {}", log_pubkey!(node_id));
+							self.enqueue_message(&mut *get_peer_for_forwarding!(node_id)?, msg);
+						},
+						MessageSendEvent::SendPeerStorageRetrieval { ref node_id, ref msg } => {
+							log_debug!(self.logger, "Handling SendPeerStorageRetrieval 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),
+	PeerStorage(msgs::PeerStorage),
+	PeerStorageRetrieval(msgs::PeerStorageRetrieval),
 	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::PeerStorage(ref msg) => msg.write(writer),
+			&Message::PeerStorageRetrieval(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::PeerStorage(ref msg) => msg.type_id(),
+			&Message::PeerStorageRetrieval(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::PeerStorage::TYPE => {
+			Ok(Message::PeerStorage(Readable::read(buffer)?))
+		},
+		msgs::PeerStorageRetrieval::TYPE => {
+			Ok(Message::PeerStorageRetrieval(Readable::read(buffer)?))
+		},
 		msgs::OpenChannel::TYPE => {
 			Ok(Message::OpenChannel(Readable::read(buffer)?))
 		},
@@ -448,6 +460,14 @@ impl Encode for msgs::Stfu {
 	const TYPE: u16 = 2;
 }
 
+impl Encode for msgs::PeerStorage {
+	const TYPE: u16 = 7;
+}
+
+impl Encode for msgs::PeerStorageRetrieval {
+	const TYPE: u16 = 9;
+}
+
 impl Encode for msgs::Init {
 	const TYPE: u16 = 16;
 }

lightning/src/util/test_utils.rs

@@ -1118,6 +1118,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::PeerStorage) {
+		self.received_msg(wire::Message::PeerStorage(msg));
+	}
+
+	fn handle_peer_storage_retrieval(
+		&self, _their_node_id: PublicKey, msg: msgs::PeerStorageRetrieval,
+	) {
+		self.received_msg(wire::Message::PeerStorageRetrieval(msg));
+	}
+
 	fn message_received(&self) {}
 }