Cache and Broadcast channel update later

- Cache the force-close channel update message, and broadcast them
  later.
- Update get_and_clear_pending_msg_events to broadcast the
  pending_broadcast_events along with other messages.
- Also appropriately update the tests which are affected by this change

Author: shaavan

lightning/src/ln/channelmanager.rs

@@ -1381,6 +1381,10 @@ where
 
 	pending_offers_messages: Mutex<Vec<PendingOnionMessage<OffersMessage>>>,
 
+	/// Tracks the channel_update message that were not broadcasted because
+	/// we were not connected to any peers.
+	pending_broadcast_messages: Mutex<Vec<MessageSendEvent>>,
+
 	entropy_source: ES,
 	node_signer: NS,
 	signer_provider: SP,
@@ -2455,6 +2459,7 @@ where
 			funding_batch_states: Mutex::new(BTreeMap::new()),
 
 			pending_offers_messages: Mutex::new(Vec::new()),
+			pending_broadcast_messages: Mutex::new(Vec::new()),
 
 			entropy_source,
 			node_signer,
@@ -2946,17 +2951,11 @@ where
 			}
 		};
 		if let Some(update) = update_opt {
-			// Try to send the `BroadcastChannelUpdate` to the peer we just force-closed on, but if
-			// not try to broadcast it via whatever peer we have.
-			let per_peer_state = self.per_peer_state.read().unwrap();
-			let a_peer_state_opt = per_peer_state.get(peer_node_id)
-				.ok_or(per_peer_state.values().next());
-			if let Ok(a_peer_state_mutex) = a_peer_state_opt {
-				let mut a_peer_state = a_peer_state_mutex.lock().unwrap();
-				a_peer_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
-					msg: update
-				});
-			}
+			// If we have some Channel Update to broadcast, we cache it and broadcast it later.
+			let mut pending_broadcast_messages = self.pending_broadcast_messages.lock().unwrap();
+			pending_broadcast_messages.push(events::MessageSendEvent::BroadcastChannelUpdate {
+				msg: update
+			});
 		}
 
 		Ok(counterparty_node_id)
@@ -8194,6 +8193,7 @@ where
 				result = NotifyOption::DoPersist;
 			}
 
+			let mut is_some_peer_connected = false;
 			let mut pending_events = Vec::new();
 			let per_peer_state = self.per_peer_state.read().unwrap();
 			for (_cp_id, peer_state_mutex) in per_peer_state.iter() {
@@ -8202,6 +8202,15 @@ where
 				if peer_state.pending_msg_events.len() > 0 {
 					pending_events.append(&mut peer_state.pending_msg_events);
 				}
+				if peer_state.is_connected {
+					is_some_peer_connected = true
+				}
+			}
+
+			// Ensure that we are connected to some peers before getting broadcast messages.
+			if is_some_peer_connected {
+				let mut broadcast_msgs = self.pending_broadcast_messages.lock().unwrap();
+				pending_events.append(&mut broadcast_msgs);
 			}
 
 			if !pending_events.is_empty() {
@@ -11099,6 +11108,8 @@ where
 
 			pending_offers_messages: Mutex::new(Vec::new()),
 
+			pending_broadcast_messages: Mutex::new(Vec::new()),
+
 			entropy_source: args.entropy_source,
 			node_signer: args.node_signer,
 			signer_provider: args.signer_provider,
@@ -11630,10 +11641,10 @@ mod tests {
 
 	#[test]
 	fn test_drop_disconnected_peers_when_removing_channels() {
-		let chanmon_cfgs = create_chanmon_cfgs(2);
-		let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
-		let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
-		let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+		let chanmon_cfgs = create_chanmon_cfgs(3);
+		let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
+		let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]);
+		let nodes = create_network(3, &node_cfgs, &node_chanmgrs);
 
 		let chan = create_announced_chan_between_nodes(&nodes, 0, 1);
 
@@ -11650,15 +11661,15 @@ mod tests {
 			// disconnected and the channel between has been force closed.
 			let nodes_0_per_peer_state = nodes[0].node.per_peer_state.read().unwrap();
 			// Assert that nodes[1] isn't removed before `timer_tick_occurred` has been executed.
-			assert_eq!(nodes_0_per_peer_state.len(), 1);
+			assert_eq!(nodes_0_per_peer_state.len(), 2);
 			assert!(nodes_0_per_peer_state.get(&nodes[1].node.get_our_node_id()).is_some());
 		}
 
 		nodes[0].node.timer_tick_occurred();
 
 		{
 			// Assert that nodes[1] has now been removed.
-			assert_eq!(nodes[0].node.per_peer_state.read().unwrap().len(), 0);
+			assert_eq!(nodes[0].node.per_peer_state.read().unwrap().len(), 1);
 		}
 	}
 
@@ -12362,11 +12373,11 @@ mod tests {
 
 	#[test]
 	fn test_trigger_lnd_force_close() {
-		let chanmon_cfg = create_chanmon_cfgs(2);
-		let node_cfg = create_node_cfgs(2, &chanmon_cfg);
+		let chanmon_cfg = create_chanmon_cfgs(3);
+		let node_cfg = create_node_cfgs(3, &chanmon_cfg);
 		let user_config = test_default_channel_config();
-		let node_chanmgr = create_node_chanmgrs(2, &node_cfg, &[Some(user_config), Some(user_config)]);
-		let nodes = create_network(2, &node_cfg, &node_chanmgr);
+		let node_chanmgr = create_node_chanmgrs(3, &node_cfg, &[Some(user_config), Some(user_config), Some(user_config)]);
+		let nodes = create_network(3, &node_cfg, &node_chanmgr);
 
 		// Open a channel, immediately disconnect each other, and broadcast Alice's latest state.
 		let (_, _, chan_id, funding_tx) = create_announced_chan_between_nodes(&nodes, 0, 1);

lightning/src/ln/reorg_tests.rs

@@ -763,21 +763,21 @@ fn test_htlc_preimage_claim_prev_counterparty_commitment_after_current_counterpa
 fn do_test_retries_own_commitment_broadcast_after_reorg(anchors: bool, revoked_counterparty_commitment: bool) {
 	// Tests that a node will retry broadcasting its own commitment after seeing a confirmed
 	// counterparty commitment be reorged out.
-	let mut chanmon_cfgs = create_chanmon_cfgs(2);
+	let mut chanmon_cfgs = create_chanmon_cfgs(3);
 	if revoked_counterparty_commitment {
 		chanmon_cfgs[1].keys_manager.disable_revocation_policy_check = true;
 	}
-	let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
+	let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
 	let mut config = test_default_channel_config();
 	if anchors {
 		config.channel_handshake_config.negotiate_anchors_zero_fee_htlc_tx = true;
 		config.manually_accept_inbound_channels = true;
 	}
 	let persister;
 	let new_chain_monitor;
-	let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[Some(config), Some(config)]);
+	let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[Some(config), Some(config), Some(config)]);
 	let nodes_1_deserialized;
-	let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+	let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs);
 
 	let (_, _, chan_id, funding_tx) = create_announced_chan_between_nodes(&nodes, 0, 1);
 
@@ -801,6 +801,11 @@ fn do_test_retries_own_commitment_broadcast_after_reorg(anchors: bool, revoked_c
 		reload_node!(
 			nodes[1], config, &serialized_node, &[&serialized_monitor], persister, new_chain_monitor, nodes_1_deserialized
 		);
+
+		// Reconnect node[1] with node[2] to allow successful channel_update broadcast later
+		nodes[1].node.peer_connected(&nodes[2].node.get_our_node_id(), &Init {
+			features: nodes[2].node.init_features(), networks: None, remote_network_address: None
+		}, true).unwrap();
 	}
 
 	// Connect blocks until the HTLC expiry is met, prompting a commitment broadcast by A.