If needed, rebroadcast channel update of a force-closed channel

- This commit adds the ability to rebroadcast the channel update message
  for a force closed channel if we fail to broadcast it to any peer at
the time of closing of channel.
- Also it adds persistance for those message, saving them at the time of
  their addition or clearance.

Author: shaavan

lightning/src/ln/channelmanager.rs

@@ -1279,6 +1279,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,
@@ -2360,6 +2364,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,
@@ -2862,15 +2867,39 @@ 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.
+			// not try to broadcast it via whatever peer we are connected to.
+			let brodcast_message_evt = events::MessageSendEvent::BroadcastChannelUpdate {
+				msg: update
+			};
+
 			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
-				});
+
+			// Attempt to get the a_peer_state_mutex for the peer we force-closed on.
+			let a_peer_state_mutex_opt = per_peer_state
+			.get(peer_node_id)
+			.map(|v| v);
+
+			// If the particular peer is not present, select any random connected peer from the ones we are connected to.
+			let a_peer_state_mutex_opt = a_peer_state_mutex_opt.or_else(|| {
+			per_peer_state
+				.iter()
+				.find(|(_, v)| v.lock().unwrap().is_connected)
+				.map(|(_, v)| v)
+			});
+
+			match a_peer_state_mutex_opt {
+				Some(a_peer_state_mutex) => {
+					// Handle the case where a connected peer is found.
+					let mut a_peer_state = a_peer_state_mutex.lock().unwrap();
+					a_peer_state.pending_msg_events.push(brodcast_message_evt);
+				}
+				None => {
+					// Handle the case where no connected peer is found.
+					let mut pending_broadcast_messages = self.pending_broadcast_messages.lock().unwrap();
+					pending_broadcast_messages.push(brodcast_message_evt);
+					log_info!(self.logger, "Not able to broadcast channel_update of force-closed channel right now.
+						Will try rebroadcasting later.");
+				}
 			}
 		}
 
@@ -4731,6 +4760,7 @@ where
 
 			{
 				let per_peer_state = self.per_peer_state.read().unwrap();
+
 				for (counterparty_node_id, peer_state_mutex) in per_peer_state.iter() {
 					let mut peer_state_lock = peer_state_mutex.lock().unwrap();
 					let peer_state = &mut *peer_state_lock;
@@ -9006,6 +9036,23 @@ where
 						msg: chan.get_channel_reestablish(&self.logger),
 					});
 				});
+
+				{
+					// Get pending messages to be broadcasted.
+					let broadcast_evts = self.pending_broadcast_messages.lock().unwrap();
+
+					// If we have some pending message to broadcast, and we are connected to peers.
+					if broadcast_evts.len() > 0 && per_peer_state.len() > 0 {
+						let a_peer_state_mutex = per_peer_state.values().next().unwrap();
+						let mut a_peer_state = a_peer_state_mutex.lock().unwrap();
+
+						a_peer_state.pending_msg_events.extend(broadcast_evts.iter().cloned());
+
+						self.pending_broadcast_messages.lock().unwrap().clear();
+
+						return NotifyOption::DoPersist;
+					}
+				}
 			}
 
 			return NotifyOption::SkipPersistHandleEvents;
@@ -10999,6 +11046,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,