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,7 +3282,10 @@ 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.
+		connect_dummy_node(&nodes[1]);
 		check_closed_broadcast(&nodes[1], 1, true);
+		disconnect_dummy_node(&nodes[1]);
 		check_closed_event(&nodes[1], 1, ClosureReason::CommitmentTxConfirmed, false, &[nodes[0].node.get_our_node_id()], 100000);
 	} else {
 		// While we forwarded the payment a while ago, we don't want to process events too early or
@@ -3395,7 +3398,9 @@ fn do_test_reload_mon_update_completion_actions(close_during_reload: bool) {
 		// (as learned about during the on-reload block connection).
 		nodes[0].node.force_close_broadcasting_latest_txn(&chan_id_ab, &nodes[1].node.get_our_node_id()).unwrap();
 		check_added_monitors!(nodes[0], 1);
+		connect_dummy_node(&nodes[0]);
 		check_closed_broadcast!(nodes[0], true);
+		disconnect_dummy_node(&nodes[0]);
 		check_closed_event(&nodes[0], 1, ClosureReason::HolderForceClosed, false, &[nodes[1].node.get_our_node_id()], 100_000);
 		let as_closing_tx = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
 		mine_transaction_without_consistency_checks(&nodes[1], &as_closing_tx[0]);
@@ -3410,7 +3415,9 @@ fn do_test_reload_mon_update_completion_actions(close_during_reload: bool) {
 			Event::ChannelClosed { .. } => {},
 			_ => panic!(),
 		}
+		connect_dummy_node(&nodes[1]);
 		check_closed_broadcast!(nodes[1], true);
+		disconnect_dummy_node(&nodes[1]);
 	}
 
 	// Once we run event processing the monitor should free, check that it was indeed the B<->C

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
 											});
 										}
@@ -8450,6 +8460,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 +8534,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 +8981,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 +11894,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

@@ -3039,6 +3039,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
@@ -3152,13 +3172,13 @@ 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 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);
@@ -3178,14 +3198,14 @@ pub fn handle_announce_close_broadcast_events<'a, 'b, 'c>(nodes: &Vec<Node<'a, '
 
 	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);

lightning/src/ln/functional_tests.rs

@@ -2272,7 +2272,9 @@ fn channel_monitor_network_test() {
 	// Simple case with no pending HTLCs:
 	nodes[1].node.force_close_broadcasting_latest_txn(&chan_1.2, &nodes[0].node.get_our_node_id()).unwrap();
 	check_added_monitors!(nodes[1], 1);
+	connect_dummy_node(&nodes[1]);
 	check_closed_broadcast!(nodes[1], true);
+	disconnect_dummy_node(&nodes[1]);
 	check_closed_event!(nodes[1], 1, ClosureReason::HolderForceClosed, [nodes[0].node.get_our_node_id()], 100000);
 	{
 		let mut node_txn = test_txn_broadcast(&nodes[1], &chan_1, None, HTLCType::NONE);
@@ -2371,13 +2373,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 +2405,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 +4607,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 +4650,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 +5064,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 +5142,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 +7323,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);

lightning/src/ln/monitor_tests.rs

@@ -2683,7 +2683,9 @@ fn do_test_anchors_monitor_fixes_counterparty_payment_script_on_reload(confirm_c
 		reload_node!(nodes[1], user_config, &nodes[1].node.encode(), &[&serialized_monitor], persister, chain_monitor, node_deserialized);
 		let commitment_tx_conf_height = block_from_scid(&mine_transaction(&nodes[1], &commitment_tx));
 		check_added_monitors(&nodes[1], 1);
+		connect_dummy_node(&nodes[1]);
 		check_closed_broadcast(&nodes[1], 1, true);
+		disconnect_dummy_node(&nodes[1]);
 		commitment_tx_conf_height
 	};
 	check_closed_event!(&nodes[1], 1, ClosureReason::CommitmentTxConfirmed, false,

lightning/src/ln/payment_tests.rs

@@ -1046,6 +1046,10 @@ fn do_test_dup_htlc_onchain_fails_on_reload(persist_manager_post_event: bool, co
 	let nodes_0_deserialized;
 	let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs);
 
+	// Connect a dummy node for proper future events broadcasting
+	connect_dummy_node(&nodes[0]);
+	connect_dummy_node(&nodes[1]);
+
 	let (_, _, chan_id, funding_tx) = create_announced_chan_between_nodes(&nodes, 0, 1);
 
 	// Route a payment, but force-close the channel before the HTLC fulfill message arrives at

lightning/src/ln/reorg_tests.rs

@@ -352,7 +352,9 @@ fn do_test_unconf_chan(reload_node: bool, reorg_after_reload: bool, use_funding_
 	check_added_monitors!(nodes[0], 1);
 	let expected_err = "Funding transaction was un-confirmed. Locked at 6 confs, now have 0 confs.";
 	if reorg_after_reload || !reload_node {
+		connect_dummy_node(&nodes[0]);
 		handle_announce_close_broadcast_events(&nodes, 0, 1, true, "Channel closed because of an exception: Funding transaction was un-confirmed. Locked at 6 confs, now have 0 confs.");
+		disconnect_dummy_node(&nodes[0]);
 		check_added_monitors!(nodes[1], 1);
 		check_closed_event!(nodes[1], 1, ClosureReason::CounterpartyForceClosed { peer_msg: UntrustedString(format!("Channel closed because of an exception: {}", expected_err)) }
 			, [nodes[0].node.get_our_node_id()], 100000);