Enable ChainMonitor to distribute PeerStorage

To enable ChainMonitor sending peer storage to channel partners whenever a new block is added,
We implement BaseMessageHandler for ChainMonitor.
This allows the `ChainMonitor` to handle the peer storage distribution.

Key changes:
 - Add BaseMessageHandler into the MessageHandler.
 - Implement BaseMessageHandler for ChainMonitor.
 - Process BaseMessageHandler events inside process_events().

Author: Aditya Sharma

fuzz/src/chanmon_consistency.rs

@@ -191,6 +191,7 @@ struct TestChainMonitor {
 			Arc<FuzzEstimator>,
 			Arc<dyn Logger>,
 			Arc<TestPersister>,
+			Arc<KeyProvider>,
 		>,
 	>,
 	pub latest_monitors: Mutex<HashMap<ChannelId, LatestMonitorState>>,
@@ -207,6 +208,8 @@ impl TestChainMonitor {
 				logger.clone(),
 				feeest,
 				Arc::clone(&persister),
+				Arc::clone(&keys),
+				keys.get_peer_storage_key(),
 			)),
 			logger,
 			keys,

fuzz/src/full_stack.rs

@@ -225,6 +225,7 @@ type ChannelMan<'a> = ChannelManager<
 			Arc<FuzzEstimator>,
 			Arc<dyn Logger>,
 			Arc<TestPersister>,
+			Arc<KeyProvider>,
 		>,
 	>,
 	Arc<TestBroadcaster>,
@@ -244,6 +245,7 @@ type PeerMan<'a> = PeerManager<
 	Arc<dyn Logger>,
 	IgnoringMessageHandler,
 	Arc<KeyProvider>,
+	IgnoringMessageHandler,
 >;
 
 struct MoneyLossDetector<'a> {
@@ -256,6 +258,7 @@ struct MoneyLossDetector<'a> {
 			Arc<FuzzEstimator>,
 			Arc<dyn Logger>,
 			Arc<TestPersister>,
+			Arc<KeyProvider>,
 		>,
 	>,
 	handler: PeerMan<'a>,
@@ -280,6 +283,7 @@ impl<'a> MoneyLossDetector<'a> {
 				Arc<FuzzEstimator>,
 				Arc<dyn Logger>,
 				Arc<TestPersister>,
+				Arc<KeyProvider>,
 			>,
 		>,
 		handler: PeerMan<'a>,
@@ -580,6 +584,8 @@ pub fn do_test(mut data: &[u8], logger: &Arc<dyn Logger>) {
 		Arc::clone(&logger),
 		fee_est.clone(),
 		Arc::new(TestPersister { update_ret: Mutex::new(ChannelMonitorUpdateStatus::Completed) }),
+		Arc::clone(&keys_manager),
+		keys_manager.get_peer_storage_key(),
 	));
 
 	let network = Network::Bitcoin;
@@ -613,6 +619,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

@@ -37,6 +37,7 @@ use lightning::routing::utxo::UtxoLookup;
 use lightning::sign::ChangeDestinationSource;
 #[cfg(feature = "std")]
 use lightning::sign::ChangeDestinationSourceSync;
+use lightning::sign::EntropySource;
 use lightning::sign::OutputSpender;
 use lightning::util::logger::Logger;
 use lightning::util::persist::{KVStore, Persister};
@@ -668,14 +669,14 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
 /// #     fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize { 0 }
 /// #     fn disconnect_socket(&mut self) {}
 /// # }
-/// # type ChainMonitor<B, F, FE> = lightning::chain::chainmonitor::ChainMonitor<lightning::sign::InMemorySigner, Arc<F>, Arc<B>, Arc<FE>, Arc<Logger>, Arc<Store>>;
+/// # type ChainMonitor<B, F, FE> = lightning::chain::chainmonitor::ChainMonitor<lightning::sign::InMemorySigner, Arc<F>, Arc<B>, Arc<FE>, Arc<Logger>, Arc<Store>, Arc<lightning::sign::KeysManager>>;
 /// # type NetworkGraph = lightning::routing::gossip::NetworkGraph<Arc<Logger>>;
 /// # type P2PGossipSync<UL> = lightning::routing::gossip::P2PGossipSync<Arc<NetworkGraph>, Arc<UL>, Arc<Logger>>;
 /// # type ChannelManager<B, F, FE> = lightning::ln::channelmanager::SimpleArcChannelManager<ChainMonitor<B, F, FE>, B, FE, Logger>;
 /// # type OnionMessenger<B, F, FE> = lightning::onion_message::messenger::OnionMessenger<Arc<lightning::sign::KeysManager>, Arc<lightning::sign::KeysManager>, Arc<Logger>, Arc<ChannelManager<B, F, FE>>, Arc<lightning::onion_message::messenger::DefaultMessageRouter<Arc<NetworkGraph>, Arc<Logger>, Arc<lightning::sign::KeysManager>>>, Arc<ChannelManager<B, F, FE>>, lightning::ln::peer_handler::IgnoringMessageHandler, lightning::ln::peer_handler::IgnoringMessageHandler, lightning::ln::peer_handler::IgnoringMessageHandler>;
 /// # type LiquidityManager<B, F, FE> = lightning_liquidity::LiquidityManager<Arc<lightning::sign::KeysManager>, Arc<ChannelManager<B, F, FE>>, Arc<F>>;
 /// # type Scorer = RwLock<lightning::routing::scoring::ProbabilisticScorer<Arc<NetworkGraph>, Arc<Logger>>>;
-/// # type PeerManager<B, F, FE, UL> = lightning::ln::peer_handler::SimpleArcPeerManager<SocketDescriptor, ChainMonitor<B, F, FE>, B, FE, Arc<UL>, Logger>;
+/// # type PeerManager<B, F, FE, UL> = lightning::ln::peer_handler::SimpleArcPeerManager<SocketDescriptor, ChainMonitor<B, F, FE>, B, FE, Arc<UL>, Logger, F, Store>;
 /// #
 /// # struct Node<
 /// #     B: lightning::chain::chaininterface::BroadcasterInterface + Send + Sync + 'static,
@@ -780,8 +781,9 @@ pub async fn process_events_async<
 	EventHandlerFuture: core::future::Future<Output = Result<(), ReplayEvent>>,
 	EventHandler: Fn(Event) -> EventHandlerFuture,
 	PS: 'static + Deref + Send,
+	ES: 'static + Deref + Send,
 	M: 'static
-		+ Deref<Target = ChainMonitor<<CM::Target as AChannelManager>::Signer, CF, T, F, L, P>>
+		+ Deref<Target = ChainMonitor<<CM::Target as AChannelManager>::Signer, CF, T, F, L, P, ES>>
 		+ Send
 		+ Sync,
 	CM: 'static + Deref,
@@ -813,6 +815,7 @@ where
 	L::Target: 'static + Logger,
 	P::Target: 'static + Persist<<CM::Target as AChannelManager>::Signer>,
 	PS::Target: 'static + Persister<'a, CM, L, S>,
+	ES::Target: 'static + EntropySource,
 	CM::Target: AChannelManager,
 	OM::Target: AOnionMessenger,
 	PM::Target: APeerManager,
@@ -976,8 +979,11 @@ impl BackgroundProcessor {
 		P: 'static + Deref,
 		EH: 'static + EventHandler + Send,
 		PS: 'static + Deref + Send,
+		ES: 'static + Deref + Send,
 		M: 'static
-			+ Deref<Target = ChainMonitor<<CM::Target as AChannelManager>::Signer, CF, T, F, L, P>>
+			+ Deref<
+				Target = ChainMonitor<<CM::Target as AChannelManager>::Signer, CF, T, F, L, P, ES>,
+			>
 			+ Send
 			+ Sync,
 		CM: 'static + Deref + Send,
@@ -1005,6 +1011,7 @@ impl BackgroundProcessor {
 		L::Target: 'static + Logger,
 		P::Target: 'static + Persist<<CM::Target as AChannelManager>::Signer>,
 		PS::Target: 'static + Persister<'a, CM, L, S>,
+		ES::Target: 'static + EntropySource,
 		CM::Target: AChannelManager,
 		OM::Target: AOnionMessenger,
 		PM::Target: APeerManager,
@@ -1174,7 +1181,7 @@ mod tests {
 	use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
 	use lightning::routing::router::{CandidateRouteHop, DefaultRouter, Path, RouteHop};
 	use lightning::routing::scoring::{ChannelUsage, LockableScore, ScoreLookUp, ScoreUpdate};
-	use lightning::sign::{ChangeDestinationSourceSync, InMemorySigner, KeysManager};
+	use lightning::sign::{ChangeDestinationSourceSync, InMemorySigner, KeysManager, NodeSigner};
 	use lightning::types::features::{ChannelFeatures, NodeFeatures};
 	use lightning::types::payment::PaymentHash;
 	use lightning::util::config::UserConfig;
@@ -1250,6 +1257,7 @@ mod tests {
 		Arc<test_utils::TestFeeEstimator>,
 		Arc<test_utils::TestLogger>,
 		Arc<FilesystemStore>,
+		Arc<KeysManager>,
 	>;
 
 	type PGS = Arc<
@@ -1301,6 +1309,7 @@ mod tests {
 				Arc<test_utils::TestLogger>,
 				IgnoringMessageHandler,
 				Arc<KeysManager>,
+				IgnoringMessageHandler,
 			>,
 		>,
 		liquidity_manager: Arc<LM>,
@@ -1662,6 +1671,8 @@ mod tests {
 				logger.clone(),
 				fee_estimator.clone(),
 				kv_store.clone(),
+				keys_manager.clone(),
+				keys_manager.get_peer_storage_key(),
 			));
 			let best_block = BestBlock::from_network(network);
 			let params = ChainParameters { network, best_block };
@@ -1715,6 +1726,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-block-sync/src/init.rs

@@ -76,7 +76,7 @@ where
 /// 	P: chainmonitor::Persist<SP::EcdsaSigner>,
 /// >(
 /// 	block_source: &B,
-/// 	chain_monitor: &ChainMonitor<SP::EcdsaSigner, &C, &T, &F, &L, &P>,
+/// 	chain_monitor: &ChainMonitor<SP::EcdsaSigner, &C, &T, &F, &L, &P, &ES>,
 /// 	config: UserConfig,
 /// 	entropy_source: &ES,
 /// 	node_signer: &NS,
@@ -109,7 +109,7 @@ where
 /// 			config,
 /// 			vec![&mut monitor],
 /// 		);
-/// 		<(BlockHash, ChannelManager<&ChainMonitor<SP::EcdsaSigner, &C, &T, &F, &L, &P>, &T, &ES, &NS, &SP, &F, &R, &MR, &L>)>::read(
+/// 		<(BlockHash, ChannelManager<&ChainMonitor<SP::EcdsaSigner, &C, &T, &F, &L, &P, &ES>, &T, &ES, &NS, &SP, &F, &R, &MR, &L>)>::read(
 /// 			&mut Cursor::new(&serialized_manager), read_args).unwrap()
 /// 	};
 ///

lightning-liquidity/tests/common/mod.rs

@@ -5,7 +5,7 @@
 #![allow(unused_macros)]
 
 use lightning::chain::Filter;
-use lightning::sign::EntropySource;
+use lightning::sign::{EntropySource, NodeSigner};
 
 use bitcoin::blockdata::constants::{genesis_block, ChainHash};
 use bitcoin::blockdata::transaction::Transaction;
@@ -101,6 +101,7 @@ type ChainMonitor = chainmonitor::ChainMonitor<
 	Arc<test_utils::TestFeeEstimator>,
 	Arc<test_utils::TestLogger>,
 	Arc<FilesystemStore>,
+	Arc<KeysManager>,
 >;
 
 type PGS = Arc<
@@ -130,6 +131,7 @@ pub(crate) struct Node {
 				>,
 			>,
 			Arc<KeysManager>,
+			Arc<ChainMonitor>,
 		>,
 	>,
 	pub(crate) liquidity_manager:
@@ -429,6 +431,8 @@ pub(crate) fn create_liquidity_node(
 		logger.clone(),
 		fee_estimator.clone(),
 		kv_store.clone(),
+		keys_manager.clone(),
+		keys_manager.get_peer_storage_key(),
 	));
 	let best_block = BestBlock::from_network(network);
 	let chain_params = ChainParameters { network, best_block };
@@ -467,6 +471,7 @@ pub(crate) fn create_liquidity_node(
 		route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()),
 		onion_message_handler: IgnoringMessageHandler {},
 		custom_message_handler: Arc::clone(&liquidity_manager),
+		send_only_message_handler: Arc::clone(&chain_monitor),
 	};
 	let peer_manager =
 		Arc::new(PeerManager::new(msg_handler, 0, &seed, logger.clone(), keys_manager.clone()));

lightning-net-tokio/src/lib.rs

@@ -839,6 +839,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,
@@ -862,6 +863,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,
@@ -924,6 +926,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,