Distribute PeerStorage from ChainMonitor

Everytime a new block is added we send PeerStorage to all of our channel partner.

 - Add our_peer_storage and our_peerstorage_encryption_key to ChainMonitor
 - Write send_peer_storage() and send it to all channel partners whenever a new block is added

Author: Aditya Sharma

lightning-background-processor/src/lib.rs

@@ -1571,6 +1571,7 @@ mod tests {
 				logger.clone(),
 				fee_estimator.clone(),
 				kv_store.clone(),
+				keys_manager.get_peer_storage_key(),
 			));
 			let best_block = BestBlock::from_network(network);
 			let params = ChainParameters { network, best_block };

lightning/src/chain/chainmonitor.rs

@@ -32,6 +32,7 @@ use crate::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
 use crate::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, Balance, MonitorEvent, TransactionOutputs, WithChannelMonitor};
 use crate::chain::transaction::{OutPoint, TransactionData};
 use crate::ln::types::ChannelId;
+use crate::ln::msgs;
 use crate::sign::ecdsa::EcdsaChannelSigner;
 use crate::events::{self, Event, EventHandler, ReplayEvent};
 use crate::util::logger::{Logger, WithContext};
@@ -46,6 +47,7 @@ use crate::sync::{RwLock, RwLockReadGuard, Mutex, MutexGuard};
 use core::ops::Deref;
 use core::sync::atomic::{AtomicUsize, Ordering};
 use bitcoin::secp256k1::PublicKey;
+use crate::ln::our_peer_storage::OurPeerStorage;
 
 /// `Persist` defines behavior for persisting channel monitors: this could mean
 /// writing once to disk, and/or uploading to one or more backup services.
@@ -255,6 +257,8 @@ pub struct ChainMonitor<ChannelSigner: EcdsaChannelSigner, C: Deref, T: Deref, F
 	/// it to give to users (or [`MonitorEvent`]s for `ChannelManager` to process).
 	event_notifier: Notifier,
 	pending_send_only_events: Mutex<Vec<MessageSendEvent>>,
+	our_peer_storage: Mutex<OurPeerStorage>,
+	our_peerstorage_encryption_key: [u8;32],
 }
 
 impl<ChannelSigner: EcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> ChainMonitor<ChannelSigner, C, T, F, L, P>
@@ -388,7 +392,7 @@ where C::Target: chain::Filter,
 	/// pre-filter blocks or only fetch blocks matching a compact filter. Otherwise, clients may
 	/// always need to fetch full blocks absent another means for determining which blocks contain
 	/// transactions relevant to the watched channels.
-	pub fn new(chain_source: Option<C>, broadcaster: T, logger: L, feeest: F, persister: P) -> Self {
+	pub fn new(chain_source: Option<C>, broadcaster: T, logger: L, feeest: F, persister: P, our_peerstorage_encryption_key: [u8; 32]) -> Self {
 		Self {
 			monitors: RwLock::new(new_hash_map()),
 			chain_source,
@@ -400,6 +404,8 @@ where C::Target: chain::Filter,
 			highest_chain_height: AtomicUsize::new(0),
 			event_notifier: Notifier::new(),
 			pending_send_only_events: Mutex::new(Vec::new()),
+			our_peer_storage: Mutex::new(OurPeerStorage::new()),
+			our_peerstorage_encryption_key
 		}
 	}
 
@@ -668,8 +674,19 @@ where C::Target: chain::Filter,
 			});
 		}
 	}
-}
 
+	/// Retrieves all node IDs associated with the monitors.
+	///
+	/// This function collects the counterparty node IDs from all monitors into a `HashSet`,
+	/// ensuring unique IDs are returned.
+	fn get_peer_node_ids(&self) -> HashSet<PublicKey> {
+		let mon = self.monitors.read().unwrap();
+		mon
+		.values()
+		.map(|monitor| monitor.monitor.get_counterparty_node_id().unwrap().clone())
+		.collect()
+	}
+}
 
 impl<ChannelSigner: EcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> MessageSendEventsProvider for ChainMonitor<ChannelSigner, C, T, F, L, P>
 where C::Target: chain::Filter,
@@ -682,7 +699,8 @@ where C::Target: chain::Filter,
 		let mut pending_events = self.pending_send_only_events.lock().unwrap();
 		let mut ret = Vec::new();
 		core::mem::swap(&mut ret, &mut *pending_events);
-		ret	}
+		ret
+	}
 }
 
 impl<ChannelSigner: EcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> SendingOnlyMessageHandler for ChainMonitor<ChannelSigner, C, T, F, L, P>
@@ -692,7 +710,11 @@ where C::Target: chain::Filter,
 		L::Target: Logger,
 		P::Target: Persist<ChannelSigner>,
 {
-	fn send_peer_storage(&self, _their_node_id: PublicKey) {
+	fn send_peer_storage(&self, their_node_id: PublicKey) {
+		log_debug!(self.logger, "Sending Peer Storage from chainmonitor");
+
+		self.pending_send_only_events.lock().unwrap().push(events::MessageSendEvent::SendPeerStorage { node_id: their_node_id
+			, msg: msgs::PeerStorage { data: self.our_peer_storage.lock().unwrap().encrypt_our_peer_storage(self.our_peerstorage_encryption_key) } })
 	}
 }
 
@@ -762,6 +784,12 @@ where
 				header, height, &*self.broadcaster, &*self.fee_estimator, &self.logger
 			)
 		});
+
+		// Send peer storage everytime a new block arrives.
+		for node_id in self.get_peer_node_ids() {
+			self.send_peer_storage(node_id);
+		}
+
 		// Assume we may have some new events and wake the event processor
 		self.event_notifier.notify();
 	}

lightning/src/util/test_utils.rs

@@ -423,6 +423,7 @@ impl<'a> TestChainMonitor<'a> {
 				logger,
 				fee_estimator,
 				persister,
+				keys_manager.get_peer_storage_key(),
 			),
 			keys_manager,
 			expect_channel_force_closed: Mutex::new(None),