Pass all BroadcastChannelUpdates through pending_broadcast_message pipeline

And fix the consequent test failures.

1. Resolved issues stemming from lack of peers to broadcast `msg_events`.
2. Updated handling of `BroadcastChannelUpdate` and `HandleErrorMessage`.

Introduced `connect_dummy_node` and `disconnect_dummy_node` functions in functional test utils to rectify the first type of failure.
Additionally, manually adjusted other tests to align with the updated `msg_events` ordering.

Author: shaavan

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -3282,6 +3282,7 @@ fn do_test_durable_preimages_on_closed_channel(close_chans_before_reload: bool,
 	check_spends!(bs_preimage_tx, as_closing_tx[0]);
 
 	if !close_chans_before_reload {
+		// Connect a dummy node to allow broadcasting the close channel event.
 		check_closed_broadcast(&nodes[1], 1, true);
 		check_closed_event(&nodes[1], 1, ClosureReason::CommitmentTxConfirmed, false, &[nodes[0].node.get_our_node_id()], 100000);
 	} else {

lightning/src/ln/channelmanager.rs

@@ -1980,7 +1980,8 @@ macro_rules! handle_error {
 		match $internal {
 			Ok(msg) => Ok(msg),
 			Err(MsgHandleErrInternal { err, shutdown_finish, .. }) => {
-				let mut msg_events = Vec::with_capacity(2);
+				let mut msg_events = Vec::with_capacity(1);
+				let mut broadcast_events = Vec::with_capacity(1);
 
 				if let Some((shutdown_res, update_option)) = shutdown_finish {
 					let counterparty_node_id = shutdown_res.counterparty_node_id;
@@ -1992,7 +1993,7 @@ macro_rules! handle_error {
 
 					$self.finish_close_channel(shutdown_res);
 					if let Some(update) = update_option {
-						msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+						broadcast_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
 							msg: update
 						});
 					}
@@ -2008,6 +2009,11 @@ macro_rules! handle_error {
 					});
 				}
 
+				if !broadcast_events.is_empty() {
+					let mut pending_broadcast_messages = $self.pending_broadcast_messages.lock().unwrap();
+					pending_broadcast_messages.append(&mut broadcast_events);
+				}
+
 				if !msg_events.is_empty() {
 					let per_peer_state = $self.per_peer_state.read().unwrap();
 					if let Some(peer_state_mutex) = per_peer_state.get(&$counterparty_node_id) {
@@ -4042,6 +4048,8 @@ where
 			.ok_or_else(|| APIError::ChannelUnavailable { err: format!("Can't find a peer matching the passed counterparty node_id {}", counterparty_node_id) })?;
 		let mut peer_state_lock = peer_state_mutex.lock().unwrap();
 		let peer_state = &mut *peer_state_lock;
+		let mut pending_broadcast_messages = self.pending_broadcast_messages.lock().unwrap();
+
 		for channel_id in channel_ids {
 			if !peer_state.has_channel(channel_id) {
 				return Err(APIError::ChannelUnavailable {
@@ -4058,7 +4066,7 @@ where
 				}
 				if let ChannelPhase::Funded(channel) = channel_phase {
 					if let Ok(msg) = self.get_channel_update_for_broadcast(channel) {
-						peer_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate { msg });
+						pending_broadcast_messages.push(events::MessageSendEvent::BroadcastChannelUpdate { msg });
 					} else if let Ok(msg) = self.get_channel_update_for_unicast(channel) {
 						peer_state.pending_msg_events.push(events::MessageSendEvent::SendChannelUpdate {
 							node_id: channel.context.get_counterparty_node_id(),
@@ -4938,6 +4946,7 @@ where
 					let mut peer_state_lock = peer_state_mutex.lock().unwrap();
 					let peer_state = &mut *peer_state_lock;
 					let pending_msg_events = &mut peer_state.pending_msg_events;
+					let mut pending_broadcast_messages = self.pending_broadcast_messages.lock().unwrap();
 					let counterparty_node_id = *counterparty_node_id;
 					peer_state.channel_by_id.retain(|chan_id, phase| {
 						match phase {
@@ -4968,7 +4977,7 @@ where
 										if n >= DISABLE_GOSSIP_TICKS {
 											chan.set_channel_update_status(ChannelUpdateStatus::Disabled);
 											if let Ok(update) = self.get_channel_update_for_broadcast(&chan) {
-												pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+												pending_broadcast_messages.push(events::MessageSendEvent::BroadcastChannelUpdate {
 													msg: update
 												});
 											}
@@ -4982,7 +4991,7 @@ where
 										if n >= ENABLE_GOSSIP_TICKS {
 											chan.set_channel_update_status(ChannelUpdateStatus::Enabled);
 											if let Ok(update) = self.get_channel_update_for_broadcast(&chan) {
-												pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+												pending_broadcast_messages.push(events::MessageSendEvent::BroadcastChannelUpdate {
 													msg: update
 												});
 											}
@@ -6641,9 +6650,8 @@ where
 		}
 		if let Some(ChannelPhase::Funded(chan)) = chan_option {
 			if let Ok(update) = self.get_channel_update_for_broadcast(&chan) {
-				let mut peer_state_lock = peer_state_mutex.lock().unwrap();
-				let peer_state = &mut *peer_state_lock;
-				peer_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+				let mut pending_broadcast_messages = self.pending_broadcast_messages.lock().unwrap();
+				pending_broadcast_messages.push(events::MessageSendEvent::BroadcastChannelUpdate {
 					msg: update
 				});
 			}
@@ -7299,11 +7307,12 @@ where
 								let mut peer_state_lock = peer_state_mutex.lock().unwrap();
 								let peer_state = &mut *peer_state_lock;
 								let pending_msg_events = &mut peer_state.pending_msg_events;
+								let mut pending_broadcast_messages = self.pending_broadcast_messages.lock().unwrap();
 								if let hash_map::Entry::Occupied(chan_phase_entry) = peer_state.channel_by_id.entry(channel_id) {
 									if let ChannelPhase::Funded(mut chan) = remove_channel_phase!(self, chan_phase_entry) {
 										failed_channels.push(chan.context.force_shutdown(false, ClosureReason::HolderForceClosed));
 										if let Ok(update) = self.get_channel_update_for_broadcast(&chan) {
-											pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+											pending_broadcast_messages.push(events::MessageSendEvent::BroadcastChannelUpdate {
 												msg: update
 											});
 										}
@@ -7468,6 +7477,7 @@ where
 				let mut peer_state_lock = peer_state_mutex.lock().unwrap();
 				let peer_state = &mut *peer_state_lock;
 				let pending_msg_events = &mut peer_state.pending_msg_events;
+				let mut pending_broadcast_messages = self.pending_broadcast_messages.lock().unwrap();
 				peer_state.channel_by_id.retain(|channel_id, phase| {
 					match phase {
 						ChannelPhase::Funded(chan) => {
@@ -7488,7 +7498,7 @@ where
 										// We're done with this channel. We got a closing_signed and sent back
 										// a closing_signed with a closing transaction to broadcast.
 										if let Ok(update) = self.get_channel_update_for_broadcast(&chan) {
-											pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+											pending_broadcast_messages.push(events::MessageSendEvent::BroadcastChannelUpdate {
 												msg: update
 											});
 										}
@@ -8098,6 +8108,16 @@ where
 		self.pending_outbound_payments.clear_pending_payments()
 	}
 
+	/// Checks if at least one peer is connected.
+	pub fn is_some_peer_connected(&self) -> bool {
+		let peer_state = self.per_peer_state.read().unwrap();
+		for (_, peer_mutex) in peer_state.iter() {
+			let peer = peer_mutex.lock().unwrap();
+			if peer.is_connected { return true; }
+		}
+		false
+	}
+
 	/// When something which was blocking a channel from updating its [`ChannelMonitor`] (e.g. an
 	/// [`Event`] being handled) completes, this should be called to restore the channel to normal
 	/// operation. It will double-check that nothing *else* is also blocking the same channel from
@@ -8450,6 +8470,8 @@ where
 				let mut peer_state_lock = peer_state_mutex.lock().unwrap();
 				let peer_state = &mut *peer_state_lock;
 				let pending_msg_events = &mut peer_state.pending_msg_events;
+				let mut pending_broadcast_messages = self.pending_broadcast_messages.lock().unwrap();
+
 				peer_state.channel_by_id.retain(|_, phase| {
 					match phase {
 						// Retain unfunded channels.
@@ -8522,7 +8544,7 @@ where
 								let reason_message = format!("{}", reason);
 								failed_channels.push(channel.context.force_shutdown(true, reason));
 								if let Ok(update) = self.get_channel_update_for_broadcast(&channel) {
-									pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+									pending_broadcast_messages.push(events::MessageSendEvent::BroadcastChannelUpdate {
 										msg: update
 									});
 								}
@@ -8969,7 +8991,9 @@ where
 						// Gossip
 						&events::MessageSendEvent::SendChannelAnnouncement { .. } => false,
 						&events::MessageSendEvent::BroadcastChannelAnnouncement { .. } => true,
-						&events::MessageSendEvent::BroadcastChannelUpdate { .. } => true,
+						// [`ChannelManager::pending_broadcast_events`] holds the [`BroadcastChannelUpdate`]
+						// This check here is to ensure exhaustivity.
+						&events::MessageSendEvent::BroadcastChannelUpdate { .. } => false,
 						&events::MessageSendEvent::BroadcastNodeAnnouncement { .. } => true,
 						&events::MessageSendEvent::SendChannelUpdate { .. } => false,
 						&events::MessageSendEvent::SendChannelRangeQuery { .. } => false,
@@ -11880,7 +11904,6 @@ mod tests {
 			assert_eq!(nodes_0_lock.len(), 1);
 			assert!(nodes_0_lock.contains_key(&funding_output));
 		}
-
 		{
 			// At this stage, `nodes[1]` has proposed a fee for the closing transaction in the
 			// `handle_closing_signed` call above. As `nodes[1]` has not yet received the signature

lightning/src/ln/functional_test_utils.rs

@@ -1534,8 +1534,16 @@ macro_rules! check_warn_msg {
 /// Check that a channel's closing channel update has been broadcasted, and optionally
 /// check whether an error message event has occurred.
 pub fn check_closed_broadcast(node: &Node, num_channels: usize, with_error_msg: bool) -> Vec<msgs::ErrorMessage> {
+	let mut dummy_connected = false;
+	if !node.node.is_some_peer_connected() {
+		connect_dummy_node(&node);
+		dummy_connected = true;
+	}
 	let msg_events = node.node.get_and_clear_pending_msg_events();
 	assert_eq!(msg_events.len(), if with_error_msg { num_channels * 2 } else { num_channels });
+	if dummy_connected {
+		disconnect_dummy_node(&node);
+	}
 	msg_events.into_iter().filter_map(|msg_event| {
 		match msg_event {
 			MessageSendEvent::BroadcastChannelUpdate { ref msg } => {
@@ -3039,6 +3047,26 @@ pub fn create_network<'a, 'b: 'a, 'c: 'b>(node_count: usize, cfgs: &'b Vec<NodeC
 	nodes
 }
 
+pub fn connect_dummy_node<'a, 'b: 'a, 'c: 'b>(node: &Node<'a, 'b, 'c>) {
+	let node_id_dummy = PublicKey::from_slice(&[2; 33]).unwrap();
+
+	let mut dummy_init_features = InitFeatures::empty();
+	dummy_init_features.set_static_remote_key_required();
+
+	let init_dummy = msgs::Init {
+		features: dummy_init_features,
+		networks: None,
+		remote_network_address: None
+	};
+
+	node.node.peer_connected(&node_id_dummy, &init_dummy, true).unwrap();
+	node.onion_messenger.peer_connected(&node_id_dummy, &init_dummy, true).unwrap();
+}
+
+pub fn disconnect_dummy_node<'a, 'b: 'a, 'c: 'b>(node: &Node<'a, 'b, 'c>) {
+	node.node.peer_disconnected(&PublicKey::from_slice(&[2; 33]).unwrap());
+}
+
 // Note that the following only works for CLTV values up to 128
 pub const ACCEPTED_HTLC_SCRIPT_WEIGHT: usize = 137; // Here we have a diff due to HTLC CLTV expiry being < 2^15 in test
 pub const ACCEPTED_HTLC_SCRIPT_WEIGHT_ANCHORS: usize = 140; // Here we have a diff due to HTLC CLTV expiry being < 2^15 in test
@@ -3150,15 +3178,20 @@ pub fn check_preimage_claim<'a, 'b, 'c>(node: &Node<'a, 'b, 'c>, prev_txn: &Vec<
 }
 
 pub fn handle_announce_close_broadcast_events<'a, 'b, 'c>(nodes: &Vec<Node<'a, 'b, 'c>>, a: usize, b: usize, needs_err_handle: bool, expected_error: &str)  {
+	let mut dummy_connected = false;
+	if !nodes[a].node.is_some_peer_connected() {
+		connect_dummy_node(&nodes[a]);
+		dummy_connected = true
+	}
 	let events_1 = nodes[a].node.get_and_clear_pending_msg_events();
 	assert_eq!(events_1.len(), 2);
-	let as_update = match events_1[0] {
+	let as_update = match events_1[1] {
 		MessageSendEvent::BroadcastChannelUpdate { ref msg } => {
 			msg.clone()
 		},
 		_ => panic!("Unexpected event"),
 	};
-	match events_1[1] {
+	match events_1[0] {
 		MessageSendEvent::HandleError { node_id, action: msgs::ErrorAction::SendErrorMessage { ref msg } } => {
 			assert_eq!(node_id, nodes[b].node.get_our_node_id());
 			assert_eq!(msg.data, expected_error);
@@ -3175,17 +3208,24 @@ pub fn handle_announce_close_broadcast_events<'a, 'b, 'c>(nodes: &Vec<Node<'a, '
 		},
 		_ => panic!("Unexpected event"),
 	}
-
+	if dummy_connected {
+		disconnect_dummy_node(&nodes[a]);
+		dummy_connected = false;
+	}
+	if !nodes[b].node.is_some_peer_connected() {
+		connect_dummy_node(&nodes[b]);
+		dummy_connected = true;
+	}
 	let events_2 = nodes[b].node.get_and_clear_pending_msg_events();
 	assert_eq!(events_2.len(), if needs_err_handle { 1 } else { 2 });
-	let bs_update = match events_2[0] {
+	let bs_update = match events_2.last().unwrap() {
 		MessageSendEvent::BroadcastChannelUpdate { ref msg } => {
 			msg.clone()
 		},
 		_ => panic!("Unexpected event"),
 	};
 	if !needs_err_handle {
-		match events_2[1] {
+		match events_2[0] {
 			MessageSendEvent::HandleError { node_id, action: msgs::ErrorAction::SendErrorMessage { ref msg } } => {
 				assert_eq!(node_id, nodes[a].node.get_our_node_id());
 				assert_eq!(msg.data, expected_error);
@@ -3197,7 +3237,11 @@ pub fn handle_announce_close_broadcast_events<'a, 'b, 'c>(nodes: &Vec<Node<'a, '
 			_ => panic!("Unexpected event"),
 		}
 	}
-
+	if dummy_connected {
+		disconnect_dummy_node(&nodes[b]);
+		// Commenting the assignment to remove `unused_assignments` warning.
+		// dummy_connected = false;
+	}
 	for node in nodes {
 		node.gossip_sync.handle_channel_update(&as_update).unwrap();
 		node.gossip_sync.handle_channel_update(&bs_update).unwrap();

lightning/src/ln/functional_tests.rs

@@ -2371,13 +2371,13 @@ fn channel_monitor_network_test() {
 		connect_blocks(&nodes[3], TEST_FINAL_CLTV + LATENCY_GRACE_PERIOD_BLOCKS + 1);
 		let events = nodes[3].node.get_and_clear_pending_msg_events();
 		assert_eq!(events.len(), 2);
-		let close_chan_update_1 = match events[0] {
+		let close_chan_update_1 = match events[1] {
 			MessageSendEvent::BroadcastChannelUpdate { ref msg } => {
 				msg.clone()
 			},
 			_ => panic!("Unexpected event"),
 		};
-		match events[1] {
+		match events[0] {
 			MessageSendEvent::HandleError { action: ErrorAction::DisconnectPeer { .. }, node_id } => {
 				assert_eq!(node_id, nodes[4].node.get_our_node_id());
 			},
@@ -2403,13 +2403,13 @@ fn channel_monitor_network_test() {
 		connect_blocks(&nodes[4], TEST_FINAL_CLTV - CLTV_CLAIM_BUFFER + 2);
 		let events = nodes[4].node.get_and_clear_pending_msg_events();
 		assert_eq!(events.len(), 2);
-		let close_chan_update_2 = match events[0] {
+		let close_chan_update_2 = match events[1] {
 			MessageSendEvent::BroadcastChannelUpdate { ref msg } => {
 				msg.clone()
 			},
 			_ => panic!("Unexpected event"),
 		};
-		match events[1] {
+		match events[0] {
 			MessageSendEvent::HandleError { action: ErrorAction::DisconnectPeer { .. }, node_id } => {
 				assert_eq!(node_id, nodes[3].node.get_our_node_id());
 			},
@@ -4605,7 +4605,7 @@ fn test_static_spendable_outputs_preimage_tx() {
 		MessageSendEvent::UpdateHTLCs { .. } => {},
 		_ => panic!("Unexpected event"),
 	}
-	match events[1] {
+	match events[2] {
 		MessageSendEvent::BroadcastChannelUpdate { .. } => {},
 		_ => panic!("Unexepected event"),
 	}
@@ -4648,7 +4648,7 @@ fn test_static_spendable_outputs_timeout_tx() {
 	mine_transaction(&nodes[1], &commitment_tx[0]);
 	check_added_monitors!(nodes[1], 1);
 	let events = nodes[1].node.get_and_clear_pending_msg_events();
-	match events[0] {
+	match events[1] {
 		MessageSendEvent::BroadcastChannelUpdate { .. } => {},
 		_ => panic!("Unexpected event"),
 	}
@@ -5062,7 +5062,7 @@ fn test_duplicate_payment_hash_one_failure_one_success() {
 		MessageSendEvent::UpdateHTLCs { .. } => {},
 		_ => panic!("Unexpected event"),
 	}
-	match events[1] {
+	match events[2] {
 		MessageSendEvent::BroadcastChannelUpdate { .. } => {},
 		_ => panic!("Unexepected event"),
 	}
@@ -5140,7 +5140,7 @@ fn test_dynamic_spendable_outputs_local_htlc_success_tx() {
 		MessageSendEvent::UpdateHTLCs { .. } => {},
 		_ => panic!("Unexpected event"),
 	}
-	match events[1] {
+	match events[2] {
 		MessageSendEvent::BroadcastChannelUpdate { .. } => {},
 		_ => panic!("Unexepected event"),
 	}
@@ -7321,6 +7321,9 @@ fn test_announce_disable_channels() {
 	let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
 	let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
 
+	// Connect a dummy node for proper future events broadcasting
+	connect_dummy_node(&nodes[0]);
+
 	create_announced_chan_between_nodes(&nodes, 0, 1);
 	create_announced_chan_between_nodes(&nodes, 1, 0);
 	create_announced_chan_between_nodes(&nodes, 0, 1);