Enable ChainMonitor to distribute PeerStorage

To enable ChainMonitor sending peer storage to channel partners whenever a new block is added,
a new message handler, `SendingOnlyMessageHandler`, has been introduced.
This allows the `ChainMonitor` to handle the peer storage distribution.

Key changes:
 - Introduce the SendingOnlyMessageHandler trait to facilitate peer storage distribution.
 - Add SendingOnlyMessageHandler into the MessageHandler.
 - Implement SendingOnlyMessageHandler for ChainMonitor and IgnoringMessageHandler.
 - Process SendingOnlyMessageHandler events inside process_events().

Author: Aditya Sharma

fuzz/src/full_stack.rs

@@ -242,6 +242,7 @@ type PeerMan<'a> = PeerManager<
 	Arc<dyn Logger>,
 	IgnoringMessageHandler,
 	Arc<KeyProvider>,
+	IgnoringMessageHandler,
 >;
 
 struct MoneyLossDetector<'a> {
@@ -673,6 +674,7 @@ pub fn do_test(mut data: &[u8], logger: &Arc<dyn Logger>) {
 		route_handler: gossip_sync.clone(),
 		onion_message_handler: IgnoringMessageHandler {},
 		custom_message_handler: IgnoringMessageHandler {},
+		send_only_message_handler: IgnoringMessageHandler {},
 	};
 	let random_data = [
 		0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,

lightning-background-processor/src/lib.rs

@@ -1210,6 +1210,7 @@ mod tests {
 				Arc<test_utils::TestLogger>,
 				IgnoringMessageHandler,
 				Arc<KeysManager>,
+				IgnoringMessageHandler,
 			>,
 		>,
 		chain_monitor: Arc<ChainMonitor>,
@@ -1623,6 +1624,7 @@ mod tests {
 				route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()),
 				onion_message_handler: messenger.clone(),
 				custom_message_handler: IgnoringMessageHandler {},
+				send_only_message_handler: IgnoringMessageHandler {},
 			};
 			let peer_manager = Arc::new(PeerManager::new(
 				msg_handler,

lightning-liquidity/tests/common/mod.rs

@@ -130,6 +130,7 @@ pub(crate) struct Node {
 				>,
 			>,
 			Arc<KeysManager>,
+			Arc<test_utils::TestSendingOnlyMessageHandler>,
 		>,
 	>,
 	pub(crate) liquidity_manager:
@@ -465,6 +466,7 @@ pub(crate) fn create_liquidity_node(
 		chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new(
 			ChainHash::using_genesis_block(Network::Testnet),
 		)),
+		send_only_message_handler: Arc::new(test_utils::TestSendingOnlyMessageHandler::new()),
 		route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()),
 		onion_message_handler: IgnoringMessageHandler {},
 		custom_message_handler: Arc::clone(&liquidity_manager),

lightning-net-tokio/src/lib.rs

@@ -841,6 +841,7 @@ mod tests {
 			route_handler: Arc::clone(&a_handler),
 			onion_message_handler: Arc::new(IgnoringMessageHandler {}),
 			custom_message_handler: Arc::new(IgnoringMessageHandler {}),
+			send_only_message_handler: Arc::new(IgnoringMessageHandler {}),
 		};
 		let a_manager = Arc::new(PeerManager::new(
 			a_msg_handler,
@@ -864,6 +865,7 @@ mod tests {
 			route_handler: Arc::clone(&b_handler),
 			onion_message_handler: Arc::new(IgnoringMessageHandler {}),
 			custom_message_handler: Arc::new(IgnoringMessageHandler {}),
+			send_only_message_handler: Arc::new(IgnoringMessageHandler {}),
 		};
 		let b_manager = Arc::new(PeerManager::new(
 			b_msg_handler,
@@ -926,6 +928,7 @@ mod tests {
 			onion_message_handler: Arc::new(IgnoringMessageHandler {}),
 			route_handler: Arc::new(lightning::ln::peer_handler::IgnoringMessageHandler {}),
 			custom_message_handler: Arc::new(IgnoringMessageHandler {}),
+			send_only_message_handler: Arc::new(IgnoringMessageHandler {}),
 		};
 		let a_manager = Arc::new(PeerManager::new(
 			a_msg_handler,

lightning/src/chain/chainmonitor.rs

@@ -39,7 +39,8 @@ use crate::util::errors::APIError;
 use crate::util::persist::MonitorName;
 use crate::util::wakers::{Future, Notifier};
 use crate::ln::channel_state::ChannelDetails;
-
+use crate::ln::msgs::SendingOnlyMessageHandler;
+use crate::events::{MessageSendEvent, MessageSendEventsProvider};
 use crate::prelude::*;
 use crate::sync::{RwLock, RwLockReadGuard, Mutex, MutexGuard};
 use core::ops::Deref;
@@ -253,6 +254,7 @@ pub struct ChainMonitor<ChannelSigner: EcdsaChannelSigner, C: Deref, T: Deref, F
 	/// A [`Notifier`] used to wake up the background processor in case we have any [`Event`]s for
 	/// it to give to users (or [`MonitorEvent`]s for `ChannelManager` to process).
 	event_notifier: Notifier,
+	pending_send_only_events: Mutex<Vec<MessageSendEvent>>,
 }
 
 impl<ChannelSigner: EcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> ChainMonitor<ChannelSigner, C, T, F, L, P>
@@ -397,6 +399,7 @@ where C::Target: chain::Filter,
 			pending_monitor_events: Mutex::new(Vec::new()),
 			highest_chain_height: AtomicUsize::new(0),
 			event_notifier: Notifier::new(),
+			pending_send_only_events: Mutex::new(Vec::new()),
 		}
 	}
 
@@ -667,6 +670,32 @@ where C::Target: chain::Filter,
 	}
 }
 
+
+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,
+		T::Target: BroadcasterInterface,
+		F::Target: FeeEstimator,
+		L::Target: Logger,
+		P::Target: Persist<ChannelSigner>,
+{
+	fn get_and_clear_pending_msg_events(&self) -> Vec<MessageSendEvent> {
+		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	}
+}
+
+impl<ChannelSigner: EcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> SendingOnlyMessageHandler for ChainMonitor<ChannelSigner, C, T, F, L, P>
+where C::Target: chain::Filter,
+		T::Target: BroadcasterInterface,
+		F::Target: FeeEstimator,
+		L::Target: Logger,
+		P::Target: Persist<ChannelSigner>,
+{
+	fn send_peer_storage(&self, _their_node_id: PublicKey) {
+	}
+}
+
 impl<ChannelSigner: EcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref>
 chain::Listen for ChainMonitor<ChannelSigner, C, T, F, L, P>
 where

lightning/src/ln/msgs.rs

@@ -1517,6 +1517,19 @@ pub struct CommitmentUpdate {
 	pub commitment_signed: CommitmentSigned,
 }
 
+/// A trait for sending messages to peers without handling any incoming messages.
+/// 
+/// This trait is designed to handle outbound-only communication, allowing implementations
+/// to send specific types of messages to connected peers identified by their public keys.
+///
+/// This trait extends [`MessageSendEventsProvider`], meaning it is capable of generating 
+/// message send events, which can be processed using
+/// [`MessageSendEventsProvider::get_and_clear_pending_msg_events`].
+pub trait SendingOnlyMessageHandler: MessageSendEventsProvider {
+	/// Send `peer_storage` message to the given peer.
+	fn send_peer_storage(&self, their_node_id: PublicKey);
+}
+
 /// A trait to describe an object which can receive channel messages.
 ///
 /// Messages MAY be called in parallel when they originate from different `their_node_ids`, however