Handle PeerStorage Message and its Persistence

This commit introduces the handling and persistence of PeerStorage messages on a per-peer basis.
The peer storage is stored within the PeerState to simplify management, ensuring we do not need to remove it
when there are no active channels with the peer.

Key changes include:

 - Add PeerStorage to PeerState for persistent storage.
 - Implement internal_peer_storage to manage PeerStorage and its updates.
 - Add resend logic in peer_connected() to resend PeerStorage before sending the channel reestablish message upon reconnection.
 - Update PeerState's write() and read() methods to support PeerStorage persistence.

Author: Aditya Sharma

lightning/src/ln/channelmanager.rs

@@ -1380,6 +1380,8 @@ pub(super) struct PeerState<SP: Deref> where SP::Target: SignerProvider {
 	/// [`ChannelMessageHandler::peer_connected`] and no corresponding
 	/// [`ChannelMessageHandler::peer_disconnected`].
 	pub is_connected: bool,
+	/// Holds the peer storage data for the channel partner on a per-peer basis.
+	peer_storage: Vec<u8>,
 }
 
 impl <SP: Deref> PeerState<SP> where SP::Target: SignerProvider {
@@ -8170,9 +8172,41 @@ 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) {
+		let per_peer_state = self.per_peer_state.read().unwrap();
+		let peer_state_mutex = match per_peer_state.get(counterparty_node_id) {
+			Some(peer_state_mutex) => peer_state_mutex,
+			None => return,
+		};
+		let mut peer_state_lock = peer_state_mutex.lock().unwrap();
+		let peer_state = &mut *peer_state_lock;
+		let logger = WithContext::from(&self.logger, Some(*counterparty_node_id), None, None);
+
+		// Check if we have any channels with the peer (Currently we only provide the servie to peers we have a channel with).
+		if !peer_state.channel_by_id.values().any(|phase| phase.is_funded()) {
+			log_debug!(logger, "Ignoring peer storage request from {} as we don't have any funded channels with them.", log_pubkey!(counterparty_node_id));
+			return;
+		}
+
+		#[cfg(not(test))]
+		if msg.data.len() > 1024 {
+			log_debug!(logger, "Sending warning to peer and ignoring peer storage request from {} as its over 1KiB", log_pubkey!(counterparty_node_id));
+			peer_state.pending_msg_events.push(events::MessageSendEvent::HandleError {
+				node_id: counterparty_node_id.clone(),
+				action: msgs::ErrorAction::SendWarningMessage {
+					msg: msgs::WarningMessage {
+						channel_id: ChannelId([0; 32]),
+						data: "Supports only data up to 1 KiB in peer storage.".to_owned()
+					},
+					log_level: Level::Trace,
+				}
+			});
+			return;
+		}
 
-	fn internal_peer_storage(&self, _counterparty_node_id: &PublicKey, _msg: &msgs::PeerStorageMessage) {}
+		log_trace!(logger, "Received Peer Storage from {}", log_pubkey!(counterparty_node_id));
+		peer_state.peer_storage = msg.data.clone();
+	}
 
 	fn internal_funding_signed(&self, counterparty_node_id: &PublicKey, msg: &msgs::FundingSigned) -> Result<(), MsgHandleErrInternal> {
 		let best_block = *self.best_block.read().unwrap();
@@ -11664,6 +11698,7 @@ where
 							actions_blocking_raa_monitor_updates: BTreeMap::new(),
 							closed_channel_monitor_update_ids: BTreeMap::new(),
 							is_connected: true,
+							peer_storage: Vec::new(),
 						}));
 					},
 					hash_map::Entry::Occupied(e) => {
@@ -11693,6 +11728,15 @@ where
 				let peer_state = &mut *peer_state_lock;
 				let pending_msg_events = &mut peer_state.pending_msg_events;
 
+				if !peer_state.peer_storage.is_empty() {
+					pending_msg_events.push(events::MessageSendEvent::SendPeerStorageRetrievalMessage {
+						node_id: counterparty_node_id.clone(),
+						msg: msgs::PeerStorageRetrievalMessage {
+							data: peer_state.peer_storage.clone()
+						},
+					});
+				}
+
 				for (_, chan) in peer_state.channel_by_id.iter_mut() {
 					let logger = WithChannelContext::from(&self.logger, &chan.context(), None);
 					match chan.peer_connected_get_handshake(self.chain_hash, &&logger) {
@@ -12859,6 +12903,9 @@ where
 			if !peer_state.ok_to_remove(false) {
 				peer_pubkey.write(writer)?;
 				peer_state.latest_features.write(writer)?;
+
+				peer_state.peer_storage.write(writer)?;
+
 				if !peer_state.monitor_update_blocked_actions.is_empty() {
 					monitor_update_blocked_actions_per_peer
 						.get_or_insert_with(Vec::new)
@@ -13212,6 +13259,7 @@ where
 				monitor_update_blocked_actions: BTreeMap::new(),
 				actions_blocking_raa_monitor_updates: BTreeMap::new(),
 				closed_channel_monitor_update_ids: BTreeMap::new(),
+				peer_storage: Vec::new(),
 				is_connected: false,
 			}
 		};
@@ -13435,8 +13483,10 @@ where
 		for _ in 0..peer_count {
 			let peer_pubkey: PublicKey = Readable::read(reader)?;
 			let latest_features = Readable::read(reader)?;
+			let peer_storage = Readable::read(reader)?;
 			if let Some(peer_state) = per_peer_state.get_mut(&peer_pubkey) {
 				peer_state.get_mut().unwrap().latest_features = latest_features;
+				peer_state.get_mut().unwrap().peer_storage = peer_storage;
 			}
 		}