Send channel_{announcement,update} msgs on connection, not timer

When we connect to a new peer, immediately send them any
channel_announcement and channel_update messages for any public
channels we have with other peers. This allows us to stop sending
those messages on a timer when they have not changed and ensures
we are sending messages when we have peers connected, rather than
broadcasting at startup when we have no peers connected.

Author: TheBlueMatt

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -1126,8 +1126,8 @@ fn test_monitor_update_fail_reestablish() {
 	nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: InitFeatures::empty(), remote_network_address: None });
 	nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: InitFeatures::empty(), remote_network_address: None });
 
-	let as_reestablish = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReestablish, nodes[1].node.get_our_node_id());
-	let bs_reestablish = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReestablish, nodes[0].node.get_our_node_id());
+	let as_reestablish = get_chan_reestablish_msgs!(nodes[0], nodes[1]).pop().unwrap();
+	let bs_reestablish = get_chan_reestablish_msgs!(nodes[1], nodes[0]).pop().unwrap();
 
 	nodes[0].node.handle_channel_reestablish(&nodes[1].node.get_our_node_id(), &bs_reestablish);
 
@@ -1145,8 +1145,8 @@ fn test_monitor_update_fail_reestablish() {
 	nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: InitFeatures::empty(), remote_network_address: None });
 	nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: InitFeatures::empty(), remote_network_address: None });
 
-	assert!(as_reestablish == get_event_msg!(nodes[0], MessageSendEvent::SendChannelReestablish, nodes[1].node.get_our_node_id()));
-	assert!(bs_reestablish == get_event_msg!(nodes[1], MessageSendEvent::SendChannelReestablish, nodes[0].node.get_our_node_id()));
+	assert_eq!(get_chan_reestablish_msgs!(nodes[0], nodes[1]).pop().unwrap(), as_reestablish);
+	assert_eq!(get_chan_reestablish_msgs!(nodes[1], nodes[0]).pop().unwrap(), bs_reestablish);
 
 	nodes[0].node.handle_channel_reestablish(&nodes[1].node.get_our_node_id(), &bs_reestablish);
 	assert_eq!(
@@ -1319,8 +1319,8 @@ fn claim_while_disconnected_monitor_update_fail() {
 	nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: InitFeatures::empty(), remote_network_address: None });
 	nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: InitFeatures::empty(), remote_network_address: None });
 
-	let as_reconnect = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReestablish, nodes[1].node.get_our_node_id());
-	let bs_reconnect = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReestablish, nodes[0].node.get_our_node_id());
+	let as_reconnect = get_chan_reestablish_msgs!(nodes[0], nodes[1]).pop().unwrap();
+	let bs_reconnect = get_chan_reestablish_msgs!(nodes[1], nodes[0]).pop().unwrap();
 
 	nodes[0].node.handle_channel_reestablish(&nodes[1].node.get_our_node_id(), &bs_reconnect);
 	let _as_channel_update = get_event_msg!(nodes[0], MessageSendEvent::SendChannelUpdate, nodes[1].node.get_our_node_id());
@@ -1451,8 +1451,8 @@ fn monitor_failed_no_reestablish_response() {
 	nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: InitFeatures::empty(), remote_network_address: None });
 	nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: InitFeatures::empty(), remote_network_address: None });
 
-	let as_reconnect = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReestablish, nodes[1].node.get_our_node_id());
-	let bs_reconnect = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReestablish, nodes[0].node.get_our_node_id());
+	let as_reconnect = get_chan_reestablish_msgs!(nodes[0], nodes[1]).pop().unwrap();
+	let bs_reconnect = get_chan_reestablish_msgs!(nodes[1], nodes[0]).pop().unwrap();
 
 	nodes[1].node.handle_channel_reestablish(&nodes[0].node.get_our_node_id(), &as_reconnect);
 	let _bs_channel_update = get_event_msg!(nodes[1], MessageSendEvent::SendChannelUpdate, nodes[0].node.get_our_node_id());
@@ -2032,9 +2032,9 @@ fn test_pending_update_fee_ack_on_reconnect() {
 	nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false);
 
 	nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: InitFeatures::known(), remote_network_address: None });
-	let as_connect_msg = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReestablish, nodes[1].node.get_our_node_id());
+	let as_connect_msg = get_chan_reestablish_msgs!(nodes[0], nodes[1]).pop().unwrap();
 	nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: InitFeatures::known(), remote_network_address: None });
-	let bs_connect_msg = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReestablish, nodes[0].node.get_our_node_id());
+	let bs_connect_msg = get_chan_reestablish_msgs!(nodes[1], nodes[0]).pop().unwrap();
 
 	nodes[1].node.handle_channel_reestablish(&nodes[0].node.get_our_node_id(), &as_connect_msg);
 	let bs_resend_msgs = nodes[1].node.get_and_clear_pending_msg_events();
@@ -2160,9 +2160,9 @@ fn do_update_fee_resend_test(deliver_update: bool, parallel_updates: bool) {
 	nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false);
 
 	nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: InitFeatures::known(), remote_network_address: None });
-	let as_connect_msg = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReestablish, nodes[1].node.get_our_node_id());
+	let as_connect_msg = get_chan_reestablish_msgs!(nodes[0], nodes[1]).pop().unwrap();
 	nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: InitFeatures::known(), remote_network_address: None });
-	let bs_connect_msg = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReestablish, nodes[0].node.get_our_node_id());
+	let bs_connect_msg = get_chan_reestablish_msgs!(nodes[1], nodes[0]).pop().unwrap();
 
 	nodes[1].node.handle_channel_reestablish(&nodes[0].node.get_our_node_id(), &as_connect_msg);
 	get_event_msg!(nodes[1], MessageSendEvent::SendChannelUpdate, nodes[0].node.get_our_node_id());

lightning/src/ln/channelmanager.rs

@@ -2985,19 +2985,10 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 
 		let mut announced_chans = false;
 		for (_, chan) in channel_state.by_id.iter() {
-			if let Some(msg) = chan.get_signed_channel_announcement(self.get_our_node_id(), self.genesis_hash.clone(), self.best_block.read().unwrap().height()) {
-				channel_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelAnnouncement {
-					msg,
-					update_msg: match self.get_channel_update_for_broadcast(chan) {
-						Ok(msg) => msg,
-						Err(_) => continue,
-					},
-				});
+			if chan.get_signed_channel_announcement(self.get_our_node_id(), self.genesis_hash.clone(), self.best_block.read().unwrap().height()).is_some()
+				&& self.get_channel_update_for_broadcast(chan).is_ok()
+			{
 				announced_chans = true;
-			} else {
-				// If the channel is not public or has not yet reached channel_ready, check the
-				// next channel. If we don't yet have any public channels, we'll skip the broadcast
-				// below as peers may not accept it without channels on chain first.
 			}
 		}
 
@@ -6143,6 +6134,7 @@ impl<Signer: Sign, M: Deref , T: Deref , K: Deref , F: Deref , L: Deref >
 					&events::MessageSendEvent::SendClosingSigned { ref node_id, .. } => node_id != counterparty_node_id,
 					&events::MessageSendEvent::SendShutdown { ref node_id, .. } => node_id != counterparty_node_id,
 					&events::MessageSendEvent::SendChannelReestablish { ref node_id, .. } => node_id != counterparty_node_id,
+					&events::MessageSendEvent::SendChannelAnnouncement { ref node_id, .. } => node_id != counterparty_node_id,
 					&events::MessageSendEvent::BroadcastChannelAnnouncement { .. } => true,
 					&events::MessageSendEvent::BroadcastNodeAnnouncement { .. } => true,
 					&events::MessageSendEvent::BroadcastChannelUpdate { .. } => true,
@@ -6187,7 +6179,7 @@ impl<Signer: Sign, M: Deref , T: Deref , K: Deref , F: Deref , L: Deref >
 		let channel_state = &mut *channel_state_lock;
 		let pending_msg_events = &mut channel_state.pending_msg_events;
 		channel_state.by_id.retain(|_, chan| {
-			if chan.get_counterparty_node_id() == *counterparty_node_id {
+			let retain = if chan.get_counterparty_node_id() == *counterparty_node_id {
 				if !chan.have_received_message() {
 					// If we created this (outbound) channel while we were disconnected from the
 					// peer we probably failed to send the open_channel message, which is now
@@ -6201,7 +6193,18 @@ impl<Signer: Sign, M: Deref , T: Deref , K: Deref , F: Deref , L: Deref >
 					});
 					true
 				}
-			} else { true }
+			} else { true };
+			if retain && chan.get_counterparty_node_id() != *counterparty_node_id {
+				if let Some(msg) = chan.get_signed_channel_announcement(self.get_our_node_id(), self.genesis_hash.clone(), self.best_block.read().unwrap().height()) {
+					if let Ok(update_msg) = self.get_channel_update_for_broadcast(chan) {
+						pending_msg_events.push(events::MessageSendEvent::SendChannelAnnouncement {
+							node_id: *counterparty_node_id,
+							msg, update_msg,
+						});
+					}
+				}
+			}
+			retain
 		});
 		//TODO: Also re-broadcast announcement_signatures
 	}

lightning/src/ln/functional_test_utils.rs

@@ -828,20 +828,7 @@ pub fn create_unannounced_chan_between_nodes_with_value<'a, 'b, 'c, 'd>(nodes: &
 pub fn update_nodes_with_chan_announce<'a, 'b, 'c, 'd>(nodes: &'a Vec<Node<'b, 'c, 'd>>, a: usize, b: usize, ann: &msgs::ChannelAnnouncement, upd_1: &msgs::ChannelUpdate, upd_2: &msgs::ChannelUpdate) {
 	nodes[a].node.broadcast_node_announcement([0, 0, 0], [0; 32], Vec::new());
 	let a_events = nodes[a].node.get_and_clear_pending_msg_events();
-	assert!(a_events.len() >= 2);
-
-	// ann should be re-generated by broadcast_node_announcement - check that we have it.
-	let mut found_ann_1 = false;
-	for event in a_events.iter() {
-		match event {
-			MessageSendEvent::BroadcastChannelAnnouncement { ref msg, .. } => {
-				if msg == ann { found_ann_1 = true; }
-			},
-			MessageSendEvent::BroadcastNodeAnnouncement { .. } => {},
-			_ => panic!("Unexpected event {:?}", event),
-		}
-	}
-	assert!(found_ann_1);
+	assert_eq!(a_events.len(), 1);
 
 	let a_node_announcement = match a_events.last().unwrap() {
 		MessageSendEvent::BroadcastNodeAnnouncement { ref msg } => {
@@ -852,20 +839,7 @@ pub fn update_nodes_with_chan_announce<'a, 'b, 'c, 'd>(nodes: &'a Vec<Node<'b, '
 
 	nodes[b].node.broadcast_node_announcement([1, 1, 1], [1; 32], Vec::new());
 	let b_events = nodes[b].node.get_and_clear_pending_msg_events();
-	assert!(b_events.len() >= 2);
-
-	// ann should be re-generated by broadcast_node_announcement - check that we have it.
-	let mut found_ann_2 = false;
-	for event in b_events.iter() {
-		match event {
-			MessageSendEvent::BroadcastChannelAnnouncement { ref msg, .. } => {
-				if msg == ann { found_ann_2 = true; }
-			},
-			MessageSendEvent::BroadcastNodeAnnouncement { .. } => {},
-			_ => panic!("Unexpected event"),
-		}
-	}
-	assert!(found_ann_2);
+	assert_eq!(b_events.len(), 1);
 
 	let b_node_announcement = match b_events.last().unwrap() {
 		MessageSendEvent::BroadcastNodeAnnouncement { ref msg } => {
@@ -2258,15 +2232,27 @@ macro_rules! get_channel_value_stat {
 macro_rules! get_chan_reestablish_msgs {
 	($src_node: expr, $dst_node: expr) => {
 		{
+			let mut announcements = std::collections::HashSet::new();
 			let mut res = Vec::with_capacity(1);
 			for msg in $src_node.node.get_and_clear_pending_msg_events() {
 				if let MessageSendEvent::SendChannelReestablish { ref node_id, ref msg } = msg {
 					assert_eq!(*node_id, $dst_node.node.get_our_node_id());
 					res.push(msg.clone());
+				} else if let MessageSendEvent::SendChannelAnnouncement { ref node_id, ref msg, .. } = msg {
+					assert_eq!(*node_id, $dst_node.node.get_our_node_id());
+					announcements.insert(msg.contents.short_channel_id);
 				} else {
 					panic!("Unexpected event")
 				}
 			}
+			for chan in $src_node.node.list_channels() {
+				if chan.is_public && chan.counterparty.node_id != $dst_node.node.get_our_node_id() {
+					if let Some(scid) = chan.short_channel_id {
+						assert!(announcements.remove(&scid));
+					}
+				}
+			}
+			assert!(announcements.is_empty());
 			res
 		}
 	}

lightning/src/ln/functional_tests.rs

@@ -3895,9 +3895,9 @@ fn test_funding_peer_disconnect() {
 	assert!(events_2.is_empty());
 
 	nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: InitFeatures::empty(), remote_network_address: None });
-	let as_reestablish = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReestablish, nodes[1].node.get_our_node_id());
+	let as_reestablish = get_chan_reestablish_msgs!(nodes[0], nodes[1]).pop().unwrap();
 	nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: InitFeatures::empty(), remote_network_address: None });
-	let bs_reestablish = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReestablish, nodes[0].node.get_our_node_id());
+	let bs_reestablish = get_chan_reestablish_msgs!(nodes[1], nodes[0]).pop().unwrap();
 
 	// nodes[0] hasn't yet received a channel_ready, so it only sends that on reconnect.
 	nodes[0].node.handle_channel_reestablish(&nodes[1].node.get_our_node_id(), &bs_reestablish);
@@ -4038,21 +4038,6 @@ fn test_funding_peer_disconnect() {
 	check_added_monitors!(nodes[0], 1);
 
 	reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
-
-	// The channel announcement should be re-generated exactly by broadcast_node_announcement.
-	nodes[0].node.broadcast_node_announcement([0, 0, 0], [0; 32], Vec::new());
-	let msgs = nodes[0].node.get_and_clear_pending_msg_events();
-	let mut found_announcement = false;
-	for event in msgs.iter() {
-		match event {
-			MessageSendEvent::BroadcastChannelAnnouncement { ref msg, .. } => {
-				if *msg == chan_announcement { found_announcement = true; }
-			},
-			MessageSendEvent::BroadcastNodeAnnouncement { .. } => {},
-			_ => panic!("Unexpected event"),
-		}
-	}
-	assert!(found_announcement);
 }
 
 #[test]
@@ -4737,19 +4722,23 @@ fn test_manager_serialize_deserialize_inconsistent_monitor() {
 	claim_payment(&nodes[2], &[&nodes[0], &nodes[1]], our_payment_preimage);
 
 	nodes[3].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: InitFeatures::empty(), remote_network_address: None });
-	let reestablish = get_event_msg!(nodes[3], MessageSendEvent::SendChannelReestablish, nodes[0].node.get_our_node_id());
+	let reestablish = get_chan_reestablish_msgs!(nodes[3], nodes[0]).pop().unwrap();
 	nodes[0].node.peer_connected(&nodes[3].node.get_our_node_id(), &msgs::Init { features: InitFeatures::empty(), remote_network_address: None });
 	nodes[0].node.handle_channel_reestablish(&nodes[3].node.get_our_node_id(), &reestablish);
-	let msg_events = nodes[0].node.get_and_clear_pending_msg_events();
-	assert_eq!(msg_events.len(), 1);
-	if let MessageSendEvent::HandleError { ref action, .. } = msg_events[0] {
-		match action {
-			&ErrorAction::SendErrorMessage { ref msg } => {
-				assert_eq!(msg.channel_id, channel_id);
-			},
-			_ => panic!("Unexpected event!"),
+	let mut found_err = false;
+	for msg_event in nodes[0].node.get_and_clear_pending_msg_events() {
+		if let MessageSendEvent::HandleError { ref action, .. } = msg_event {
+			match action {
+				&ErrorAction::SendErrorMessage { ref msg } => {
+					assert_eq!(msg.channel_id, channel_id);
+					assert!(!found_err);
+					found_err = true;
+				},
+				_ => panic!("Unexpected event!"),
+			}
 		}
 	}
+	assert!(found_err);
 }
 
 macro_rules! check_spendable_outputs {

lightning/src/ln/payment_tests.rs

@@ -457,7 +457,7 @@ fn do_retry_with_no_persist(confirm_before_reload: bool) {
 	// Now nodes[1] should send a channel reestablish, which nodes[0] will respond to with an
 	// error, as the channel has hit the chain.
 	nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: InitFeatures::known(), remote_network_address: None });
-	let bs_reestablish = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReestablish, nodes[0].node.get_our_node_id());
+	let bs_reestablish = get_chan_reestablish_msgs!(nodes[1], nodes[0]).pop().unwrap();
 	nodes[0].node.handle_channel_reestablish(&nodes[1].node.get_our_node_id(), &bs_reestablish);
 	let as_err = nodes[0].node.get_and_clear_pending_msg_events();
 	assert_eq!(as_err.len(), 1);

lightning/src/ln/peer_handler.rs

@@ -1613,6 +1613,13 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 								log_bytes!(msg.channel_id));
 						self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
 					},
+					MessageSendEvent::SendChannelAnnouncement { ref node_id, ref msg, ref update_msg } => {
+						log_debug!(self.logger, "Handling SendChannelAnnouncement event in peer_handler for node {} for short channel id {}",
+								log_pubkey!(node_id),
+								msg.contents.short_channel_id);
+						self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+						self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), update_msg);
+					},
 					MessageSendEvent::BroadcastChannelAnnouncement { msg, update_msg } => {
 						log_debug!(self.logger, "Handling BroadcastChannelAnnouncement event in peer_handler for short channel id {}", msg.contents.short_channel_id);
 						match self.message_handler.route_handler.handle_channel_announcement(&msg) {

lightning/src/ln/priv_short_conf_tests.rs

@@ -144,17 +144,17 @@ fn test_priv_forwarding_rejection() {
 
 	nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: InitFeatures::known(), remote_network_address: None });
 	nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: InitFeatures::empty(), remote_network_address: None });
-	let as_reestablish = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReestablish, nodes[1].node.get_our_node_id());
-	let bs_reestablish = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReestablish, nodes[0].node.get_our_node_id());
+	let as_reestablish = get_chan_reestablish_msgs!(nodes[0], nodes[1]).pop().unwrap();
+	let bs_reestablish = get_chan_reestablish_msgs!(nodes[1], nodes[0]).pop().unwrap();
 	nodes[1].node.handle_channel_reestablish(&nodes[0].node.get_our_node_id(), &as_reestablish);
 	nodes[0].node.handle_channel_reestablish(&nodes[1].node.get_our_node_id(), &bs_reestablish);
 	get_event_msg!(nodes[0], MessageSendEvent::SendChannelUpdate, nodes[1].node.get_our_node_id());
 	get_event_msg!(nodes[1], MessageSendEvent::SendChannelUpdate, nodes[0].node.get_our_node_id());
 
 	nodes[1].node.peer_connected(&nodes[2].node.get_our_node_id(), &msgs::Init { features: InitFeatures::known(), remote_network_address: None });
 	nodes[2].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: InitFeatures::empty(), remote_network_address: None });
-	let bs_reestablish = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReestablish, nodes[2].node.get_our_node_id());
-	let cs_reestablish = get_event_msg!(nodes[2], MessageSendEvent::SendChannelReestablish, nodes[1].node.get_our_node_id());
+	let bs_reestablish = get_chan_reestablish_msgs!(nodes[1], nodes[2]).pop().unwrap();
+	let cs_reestablish = get_chan_reestablish_msgs!(nodes[2], nodes[1]).pop().unwrap();
 	nodes[2].node.handle_channel_reestablish(&nodes[1].node.get_our_node_id(), &bs_reestablish);
 	nodes[1].node.handle_channel_reestablish(&nodes[2].node.get_our_node_id(), &cs_reestablish);
 	get_event_msg!(nodes[1], MessageSendEvent::SendChannelUpdate, nodes[2].node.get_our_node_id());