Merge pull request lightningdevkit#3594 from TheBlueMatt/2025-02-peer…

…-connected-trait

Unify common message handler traits into one trait

fuzz/src/chanmon_consistency.rs

@@ -43,7 +43,6 @@ use lightning::chain::{
 	chainmonitor, channelmonitor, BestBlock, ChannelMonitorUpdateStatus, Confirm, Watch,
 };
 use lightning::events;
-use lightning::events::MessageSendEventsProvider;
 use lightning::ln::channel::FEE_SPIKE_BUFFER_FEE_INCREASE_MULTIPLE;
 use lightning::ln::channel_state::ChannelDetails;
 use lightning::ln::channelmanager::{
@@ -52,7 +51,10 @@ use lightning::ln::channelmanager::{
 };
 use lightning::ln::functional_test_utils::*;
 use lightning::ln::inbound_payment::ExpandedKey;
-use lightning::ln::msgs::{ChannelMessageHandler, CommitmentUpdate, Init, UpdateAddHTLC};
+use lightning::ln::msgs::{
+	BaseMessageHandler, ChannelMessageHandler, CommitmentUpdate, Init, MessageSendEvent,
+	UpdateAddHTLC,
+};
 use lightning::ln::script::ShutdownScript;
 use lightning::ln::types::ChannelId;
 use lightning::offers::invoice::UnsignedBolt12Invoice;
@@ -768,7 +770,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 			let open_channel = {
 				let events = $source.get_and_clear_pending_msg_events();
 				assert_eq!(events.len(), 1);
-				if let events::MessageSendEvent::SendOpenChannel { ref msg, .. } = events[0] {
+				if let MessageSendEvent::SendOpenChannel { ref msg, .. } = events[0] {
 					msg.clone()
 				} else {
 					panic!("Wrong event type");
@@ -804,7 +806,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 				}
 				let events = $dest.get_and_clear_pending_msg_events();
 				assert_eq!(events.len(), 1);
-				if let events::MessageSendEvent::SendAcceptChannel { ref msg, .. } = events[0] {
+				if let MessageSendEvent::SendAcceptChannel { ref msg, .. } = events[0] {
 					msg.clone()
 				} else {
 					panic!("Wrong event type");
@@ -847,7 +849,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 			let funding_created = {
 				let events = $source.get_and_clear_pending_msg_events();
 				assert_eq!(events.len(), 1);
-				if let events::MessageSendEvent::SendFundingCreated { ref msg, .. } = events[0] {
+				if let MessageSendEvent::SendFundingCreated { ref msg, .. } = events[0] {
 					msg.clone()
 				} else {
 					panic!("Wrong event type");
@@ -858,7 +860,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 			let funding_signed = {
 				let events = $dest.get_and_clear_pending_msg_events();
 				assert_eq!(events.len(), 1);
-				if let events::MessageSendEvent::SendFundingSigned { ref msg, .. } = events[0] {
+				if let MessageSendEvent::SendFundingSigned { ref msg, .. } = events[0] {
 					msg.clone()
 				} else {
 					panic!("Wrong event type");
@@ -913,9 +915,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 			}
 			for (idx, node_event) in node_events.iter().enumerate() {
 				for event in node_event {
-					if let events::MessageSendEvent::SendChannelReady { ref node_id, ref msg } =
-						event
-					{
+					if let MessageSendEvent::SendChannelReady { ref node_id, ref msg } = event {
 						for node in $nodes.iter() {
 							if node.get_our_node_id() == *node_id {
 								node.handle_channel_ready($nodes[idx].get_our_node_id(), msg);
@@ -930,7 +930,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 			for node in $nodes.iter() {
 				let events = node.get_and_clear_pending_msg_events();
 				for event in events {
-					if let events::MessageSendEvent::SendAnnouncementSignatures { .. } = event {
+					if let MessageSendEvent::SendAnnouncementSignatures { .. } = event {
 					} else {
 						panic!("Wrong event type");
 					}
@@ -1015,25 +1015,25 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 				let expect_drop_id = if let Some(id) = expect_drop_node { Some(nodes[id].get_our_node_id()) } else { None };
 				for event in $excess_events {
 					let push_a = match event {
-						events::MessageSendEvent::UpdateHTLCs { ref node_id, .. } => {
+						MessageSendEvent::UpdateHTLCs { ref node_id, .. } => {
 							if Some(*node_id) == expect_drop_id { panic!("peer_disconnected should drop msgs bound for the disconnected peer"); }
 							*node_id == a_id
 						},
-						events::MessageSendEvent::SendRevokeAndACK { ref node_id, .. } => {
+						MessageSendEvent::SendRevokeAndACK { ref node_id, .. } => {
 							if Some(*node_id) == expect_drop_id { panic!("peer_disconnected should drop msgs bound for the disconnected peer"); }
 							*node_id == a_id
 						},
-						events::MessageSendEvent::SendChannelReestablish { ref node_id, .. } => {
+						MessageSendEvent::SendChannelReestablish { ref node_id, .. } => {
 							if Some(*node_id) == expect_drop_id { panic!("peer_disconnected should drop msgs bound for the disconnected peer"); }
 							*node_id == a_id
 						},
-						events::MessageSendEvent::SendStfu { ref node_id, .. } => {
+						MessageSendEvent::SendStfu { ref node_id, .. } => {
 							if Some(*node_id) == expect_drop_id { panic!("peer_disconnected should drop msgs bound for the disconnected peer"); }
 							*node_id == a_id
 						},
-						events::MessageSendEvent::SendChannelReady { .. } => continue,
-						events::MessageSendEvent::SendAnnouncementSignatures { .. } => continue,
-						events::MessageSendEvent::SendChannelUpdate { ref node_id, ref msg } => {
+						MessageSendEvent::SendChannelReady { .. } => continue,
+						MessageSendEvent::SendAnnouncementSignatures { .. } => continue,
+						MessageSendEvent::SendChannelUpdate { ref node_id, ref msg } => {
 							assert_eq!(msg.contents.channel_flags & 2, 0); // The disable bit must never be set!
 							if Some(*node_id) == expect_drop_id { panic!("peer_disconnected should drop msgs bound for the disconnected peer"); }
 							*node_id == a_id
@@ -1089,7 +1089,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 				for event in &mut events_iter {
 					had_events = true;
 					match event {
-						events::MessageSendEvent::UpdateHTLCs { node_id, updates: CommitmentUpdate { update_add_htlcs, update_fail_htlcs, update_fulfill_htlcs, update_fail_malformed_htlcs, update_fee, commitment_signed } } => {
+						MessageSendEvent::UpdateHTLCs { node_id, updates: CommitmentUpdate { update_add_htlcs, update_fail_htlcs, update_fulfill_htlcs, update_fail_malformed_htlcs, update_fee, commitment_signed } } => {
 							for (idx, dest) in nodes.iter().enumerate() {
 								if dest.get_our_node_id() == node_id {
 									for update_add in update_add_htlcs.iter() {
@@ -1127,7 +1127,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 										!update_fail_htlcs.is_empty() || !update_fail_malformed_htlcs.is_empty();
 									if $limit_events != ProcessMessages::AllMessages && processed_change {
 										// If we only want to process some messages, don't deliver the CS until later.
-										extra_ev = Some(events::MessageSendEvent::UpdateHTLCs { node_id, updates: CommitmentUpdate {
+										extra_ev = Some(MessageSendEvent::UpdateHTLCs { node_id, updates: CommitmentUpdate {
 											update_add_htlcs: Vec::new(),
 											update_fail_htlcs: Vec::new(),
 											update_fulfill_htlcs: Vec::new(),
@@ -1143,37 +1143,37 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 								}
 							}
 						},
-						events::MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => {
+						MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => {
 							for (idx, dest) in nodes.iter().enumerate() {
 								if dest.get_our_node_id() == *node_id {
 									out.locked_write(format!("Delivering revoke_and_ack from node {} to node {}.\n", $node, idx).as_bytes());
 									dest.handle_revoke_and_ack(nodes[$node].get_our_node_id(), msg);
 								}
 							}
 						},
-						events::MessageSendEvent::SendChannelReestablish { ref node_id, ref msg } => {
+						MessageSendEvent::SendChannelReestablish { ref node_id, ref msg } => {
 							for (idx, dest) in nodes.iter().enumerate() {
 								if dest.get_our_node_id() == *node_id {
 									out.locked_write(format!("Delivering channel_reestablish from node {} to node {}.\n", $node, idx).as_bytes());
 									dest.handle_channel_reestablish(nodes[$node].get_our_node_id(), msg);
 								}
 							}
 						},
-						events::MessageSendEvent::SendStfu { ref node_id, ref msg } => {
+						MessageSendEvent::SendStfu { ref node_id, ref msg } => {
 							for (idx, dest) in nodes.iter().enumerate() {
 								if dest.get_our_node_id() == *node_id {
 									out.locked_write(format!("Delivering stfu from node {} to node {}.\n", $node, idx).as_bytes());
 									dest.handle_stfu(nodes[$node].get_our_node_id(), msg);
 								}
 							}
 						}
-						events::MessageSendEvent::SendChannelReady { .. } => {
+						MessageSendEvent::SendChannelReady { .. } => {
 							// Can be generated as a reestablish response
 						},
-						events::MessageSendEvent::SendAnnouncementSignatures { .. } => {
+						MessageSendEvent::SendAnnouncementSignatures { .. } => {
 							// Can be generated as a reestablish response
 						},
-						events::MessageSendEvent::SendChannelUpdate { ref msg, .. } => {
+						MessageSendEvent::SendChannelUpdate { ref msg, .. } => {
 							// When we reconnect we will resend a channel_update to make sure our
 							// counterparty has the latest parameters for receiving payments
 							// through us. We do, however, check that the message does not include
@@ -1216,13 +1216,13 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 				if $counterparty_id == 0 {
 					for event in nodes[0].get_and_clear_pending_msg_events() {
 						match event {
-							events::MessageSendEvent::UpdateHTLCs { .. } => {},
-							events::MessageSendEvent::SendRevokeAndACK { .. } => {},
-							events::MessageSendEvent::SendChannelReestablish { .. } => {},
-							events::MessageSendEvent::SendStfu { .. } => {},
-							events::MessageSendEvent::SendChannelReady { .. } => {},
-							events::MessageSendEvent::SendAnnouncementSignatures { .. } => {},
-							events::MessageSendEvent::SendChannelUpdate { ref msg, .. } => {
+							MessageSendEvent::UpdateHTLCs { .. } => {},
+							MessageSendEvent::SendRevokeAndACK { .. } => {},
+							MessageSendEvent::SendChannelReestablish { .. } => {},
+							MessageSendEvent::SendStfu { .. } => {},
+							MessageSendEvent::SendChannelReady { .. } => {},
+							MessageSendEvent::SendAnnouncementSignatures { .. } => {},
+							MessageSendEvent::SendChannelUpdate { ref msg, .. } => {
 								assert_eq!(msg.contents.channel_flags & 2, 0); // The disable bit must never be set!
 							},
 							_ => {
@@ -1243,13 +1243,13 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 				} else {
 					for event in nodes[2].get_and_clear_pending_msg_events() {
 						match event {
-							events::MessageSendEvent::UpdateHTLCs { .. } => {},
-							events::MessageSendEvent::SendRevokeAndACK { .. } => {},
-							events::MessageSendEvent::SendChannelReestablish { .. } => {},
-							events::MessageSendEvent::SendStfu { .. } => {},
-							events::MessageSendEvent::SendChannelReady { .. } => {},
-							events::MessageSendEvent::SendAnnouncementSignatures { .. } => {},
-							events::MessageSendEvent::SendChannelUpdate { ref msg, .. } => {
+							MessageSendEvent::UpdateHTLCs { .. } => {},
+							MessageSendEvent::SendRevokeAndACK { .. } => {},
+							MessageSendEvent::SendChannelReestablish { .. } => {},
+							MessageSendEvent::SendStfu { .. } => {},
+							MessageSendEvent::SendChannelReady { .. } => {},
+							MessageSendEvent::SendAnnouncementSignatures { .. } => {},
+							MessageSendEvent::SendChannelUpdate { ref msg, .. } => {
 								assert_eq!(msg.contents.channel_flags & 2, 0); // The disable bit must never be set!
 							},
 							_ => {

fuzz/src/onion_message.rs

@@ -10,7 +10,7 @@ use lightning::blinded_path::message::{
 };
 use lightning::blinded_path::EmptyNodeIdLookUp;
 use lightning::ln::inbound_payment::ExpandedKey;
-use lightning::ln::msgs::{self, OnionMessageHandler};
+use lightning::ln::msgs::{self, BaseMessageHandler, DecodeError, OnionMessageHandler};
 use lightning::ln::peer_handler::IgnoringMessageHandler;
 use lightning::ln::script::ShutdownScript;
 use lightning::offers::invoice::UnsignedBolt12Invoice;
@@ -170,7 +170,7 @@ impl CustomOnionMessageHandler for TestCustomMessageHandler {
 	}
 	fn read_custom_message<R: io::Read>(
 		&self, _message_type: u64, buffer: &mut R,
-	) -> Result<Option<Self::CustomMessage>, msgs::DecodeError> {
+	) -> Result<Option<Self::CustomMessage>, DecodeError> {
 		let mut buf = Vec::new();
 		buffer.read_to_limit(&mut buf, u64::MAX)?;
 		return Ok(Some(TestCustomMessage {}));

lightning-background-processor/src/lib.rs

@@ -1070,15 +1070,13 @@ mod tests {
 	use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
 	use lightning::chain::transaction::OutPoint;
 	use lightning::chain::{chainmonitor, BestBlock, Confirm, Filter};
-	use lightning::events::{
-		Event, MessageSendEvent, MessageSendEventsProvider, PathFailure, ReplayEvent,
-	};
+	use lightning::events::{Event, PathFailure, ReplayEvent};
 	use lightning::ln::channelmanager;
 	use lightning::ln::channelmanager::{
 		ChainParameters, PaymentId, BREAKDOWN_TIMEOUT, MIN_CLTV_EXPIRY_DELTA,
 	};
 	use lightning::ln::functional_test_utils::*;
-	use lightning::ln::msgs::{ChannelMessageHandler, Init};
+	use lightning::ln::msgs::{BaseMessageHandler, ChannelMessageHandler, Init, MessageSendEvent};
 	use lightning::ln::peer_handler::{
 		IgnoringMessageHandler, MessageHandler, PeerManager, SocketDescriptor,
 	};

lightning-dns-resolver/src/lib.rs

@@ -164,7 +164,9 @@ mod test {
 	use lightning::events::{Event, PaymentPurpose};
 	use lightning::ln::channelmanager::{PaymentId, Retry};
 	use lightning::ln::functional_test_utils::*;
-	use lightning::ln::msgs::{ChannelMessageHandler, Init, OnionMessageHandler};
+	use lightning::ln::msgs::{
+		BaseMessageHandler, ChannelMessageHandler, Init, OnionMessageHandler,
+	};
 	use lightning::ln::peer_handler::IgnoringMessageHandler;
 	use lightning::onion_message::dns_resolution::{HumanReadableName, OMNameResolver};
 	use lightning::onion_message::messenger::{

lightning-liquidity/tests/common/mod.rs

@@ -15,7 +15,7 @@ use lightning::chain::{chainmonitor, BestBlock, Confirm};
 use lightning::ln::channelmanager;
 use lightning::ln::channelmanager::ChainParameters;
 use lightning::ln::functional_test_utils::*;
-use lightning::ln::msgs::{ChannelMessageHandler, Init};
+use lightning::ln::msgs::{BaseMessageHandler, ChannelMessageHandler, Init};
 use lightning::ln::peer_handler::{
 	IgnoringMessageHandler, MessageHandler, PeerManager, SocketDescriptor,
 };

lightning-net-tokio/src/lib.rs

@@ -623,7 +623,6 @@ mod tests {
 	use bitcoin::constants::ChainHash;
 	use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey};
 	use bitcoin::Network;
-	use lightning::events::*;
 	use lightning::ln::msgs::*;
 	use lightning::ln::peer_handler::{IgnoringMessageHandler, MessageHandler, PeerManager};
 	use lightning::routing::gossip::NodeId;
@@ -684,12 +683,6 @@ mod tests {
 		) -> Option<NodeAnnouncement> {
 			None
 		}
-		fn peer_connected(
-			&self, _their_node_id: PublicKey, _init_msg: &Init, _inbound: bool,
-		) -> Result<(), ()> {
-			Ok(())
-		}
-		fn peer_disconnected(&self, _their_node_id: PublicKey) {}
 		fn handle_reply_channel_range(
 			&self, _their_node_id: PublicKey, _msg: ReplyChannelRange,
 		) -> Result<(), LightningError> {
@@ -710,12 +703,6 @@ mod tests {
 		) -> Result<(), LightningError> {
 			Ok(())
 		}
-		fn provided_node_features(&self) -> NodeFeatures {
-			NodeFeatures::empty()
-		}
-		fn provided_init_features(&self, _their_node_id: PublicKey) -> InitFeatures {
-			InitFeatures::empty()
-		}
 		fn processing_queue_high(&self) -> bool {
 			false
 		}
@@ -766,35 +753,39 @@ mod tests {
 			&self, _their_node_id: PublicKey, _msg: PeerStorageRetrieval,
 		) {
 		}
+		fn handle_channel_reestablish(&self, _their_node_id: PublicKey, _msg: &ChannelReestablish) {
+		}
+		fn handle_error(&self, _their_node_id: PublicKey, _msg: &ErrorMessage) {}
+		fn get_chain_hashes(&self) -> Option<Vec<ChainHash>> {
+			Some(vec![ChainHash::using_genesis_block(Network::Testnet)])
+		}
+		fn message_received(&self) {}
+	}
+	impl BaseMessageHandler for MsgHandler {
 		fn peer_disconnected(&self, their_node_id: PublicKey) {
 			if their_node_id == self.expected_pubkey {
 				self.disconnected_flag.store(true, Ordering::SeqCst);
-				self.pubkey_disconnected.clone().try_send(()).unwrap();
+				// This method is called twice as we're two message handlers. `try_send` will fail
+				// the second time.
+				let _ = self.pubkey_disconnected.clone().try_send(());
 			}
 		}
 		fn peer_connected(
 			&self, their_node_id: PublicKey, _init_msg: &Init, _inbound: bool,
 		) -> Result<(), ()> {
 			if their_node_id == self.expected_pubkey {
-				self.pubkey_connected.clone().try_send(()).unwrap();
+				// This method is called twice as we're two message handlers. `try_send` will fail
+				// the second time.
+				let _ = self.pubkey_connected.clone().try_send(());
 			}
 			Ok(())
 		}
-		fn handle_channel_reestablish(&self, _their_node_id: PublicKey, _msg: &ChannelReestablish) {
-		}
-		fn handle_error(&self, _their_node_id: PublicKey, _msg: &ErrorMessage) {}
 		fn provided_node_features(&self) -> NodeFeatures {
 			NodeFeatures::empty()
 		}
 		fn provided_init_features(&self, _their_node_id: PublicKey) -> InitFeatures {
 			InitFeatures::empty()
 		}
-		fn get_chain_hashes(&self) -> Option<Vec<ChainHash>> {
-			Some(vec![ChainHash::using_genesis_block(Network::Testnet)])
-		}
-		fn message_received(&self) {}
-	}
-	impl MessageSendEventsProvider for MsgHandler {
 		fn get_and_clear_pending_msg_events(&self) -> Vec<MessageSendEvent> {
 			let mut ret = Vec::new();
 			mem::swap(&mut *self.msg_events.lock().unwrap(), &mut ret);

lightning-persister/src/fs_store.rs

@@ -501,8 +501,9 @@ mod tests {
 	use lightning::chain::chainmonitor::Persist;
 	use lightning::chain::ChannelMonitorUpdateStatus;
 	use lightning::check_closed_event;
-	use lightning::events::{ClosureReason, MessageSendEventsProvider};
+	use lightning::events::ClosureReason;
 	use lightning::ln::functional_test_utils::*;
+	use lightning::ln::msgs::BaseMessageHandler;
 	use lightning::util::persist::read_channel_monitors;
 	use lightning::util::test_utils;
 

lightning/src/chain/chainmonitor.rs

@@ -928,9 +928,9 @@ mod tests {
 	use crate::{get_htlc_update_msgs, get_revoke_commit_msgs};
 	use crate::chain::{ChannelMonitorUpdateStatus, Watch};
 	use crate::chain::channelmonitor::ANTI_REORG_DELAY;
-	use crate::events::{ClosureReason, Event, MessageSendEvent, MessageSendEventsProvider};
+	use crate::events::{ClosureReason, Event};
 	use crate::ln::functional_test_utils::*;
-	use crate::ln::msgs::ChannelMessageHandler;
+	use crate::ln::msgs::{BaseMessageHandler, ChannelMessageHandler, MessageSendEvent};
 
 	const CHAINSYNC_MONITOR_PARTITION_FACTOR: u32 = 5;