@@ -1257,6 +1257,10 @@ where
1257
1257
1258
1258
pending_offers_messages: Mutex<Vec<PendingOnionMessage<OffersMessage>>>,
1259
1259
1260
+ /// Tracks the channel_update message that were not broadcasted because
1261
+ /// we were not connected to any peers.
1262
+ pending_broadcast_messages: Mutex<Vec<MessageSendEvent>>,
1263
+
1260
1264
entropy_source: ES,
1261
1265
node_signer: NS,
1262
1266
signer_provider: SP,
@@ -2338,6 +2342,7 @@ where
2338
2342
funding_batch_states: Mutex::new(BTreeMap::new()),
2339
2343
2340
2344
pending_offers_messages: Mutex::new(Vec::new()),
2345
+ pending_broadcast_messages: Mutex::new(Vec::new()),
2341
2346
2342
2347
entropy_source,
2343
2348
node_signer,
@@ -2841,14 +2846,30 @@ where
2841
2846
if let Some(update) = update_opt {
2842
2847
// Try to send the `BroadcastChannelUpdate` to the peer we just force-closed on, but if
2843
2848
// not try to broadcast it via whatever peer we have.
2849
+ let brodcast_message_evt = events::MessageSendEvent::BroadcastChannelUpdate {
2850
+ msg: update
2851
+ };
2852
+
2844
2853
let per_peer_state = self.per_peer_state.read().unwrap();
2845
- let a_peer_state_opt = per_peer_state.get(peer_node_id)
2846
- .ok_or(per_peer_state.values().next());
2847
- if let Ok(a_peer_state_mutex) = a_peer_state_opt {
2848
- let mut a_peer_state = a_peer_state_mutex.lock().unwrap();
2849
- a_peer_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
2850
- msg: update
2851
- });
2854
+
2855
+ // Attempt to get the a_peer_state_mutex for the peer we disconnected on.
2856
+ let a_peer_state_mutex_opt = per_peer_state.get(peer_node_id).map(|v| v);
2857
+
2858
+ // If the particular peer is not present. Select any random peer from the ones we are connected on.
2859
+ let a_peer_state_mutex_opt = a_peer_state_mutex_opt.or_else(|| per_peer_state.iter().next().map(|(_, v)| v));
2860
+
2861
+ match a_peer_state_mutex_opt {
2862
+ Some(a_peer_state_mutex) => {
2863
+ let mut a_peer_state = a_peer_state_mutex.lock().unwrap();
2864
+ a_peer_state.pending_msg_events.push(brodcast_message_evt);
2865
+ }
2866
+ // If we are connected to no peer.
2867
+ None => {
2868
+ let mut pending_broadcast_messages = self.pending_broadcast_messages.lock().unwrap();
2869
+ pending_broadcast_messages.push(brodcast_message_evt);
2870
+ log_info!(self.logger, "Not able to broadcast channel_update of force-closed channel right now.
2871
+ Will try rebroadcasting later.");
2872
+ }
2852
2873
}
2853
2874
}
2854
2875
@@ -4915,6 +4936,22 @@ where
4915
4936
4916
4937
{
4917
4938
let per_peer_state = self.per_peer_state.read().unwrap();
4939
+
4940
+ {
4941
+ // Get pending messages to be broadcasted.
4942
+ let broadcast_evts = self.pending_broadcast_messages.lock().unwrap();
4943
+
4944
+ // If we have some pending message to broadcast, and we are connected to peers.
4945
+ if broadcast_evts.len() > 0 && per_peer_state.len() > 0 {
4946
+ let a_peer_state_mutex = per_peer_state.values().next().unwrap();
4947
+ let mut a_peer_state = a_peer_state_mutex.lock().unwrap();
4948
+
4949
+ a_peer_state.pending_msg_events.extend(broadcast_evts.iter().cloned());
4950
+
4951
+ self.pending_broadcast_messages.lock().unwrap().clear();
4952
+ }
4953
+ }
4954
+
4918
4955
for (counterparty_node_id, peer_state_mutex) in per_peer_state.iter() {
4919
4956
let mut peer_state_lock = peer_state_mutex.lock().unwrap();
4920
4957
let peer_state = &mut *peer_state_lock;
@@ -10810,6 +10847,8 @@ where
10810
10847
10811
10848
pending_offers_messages: Mutex::new(Vec::new()),
10812
10849
10850
+ pending_broadcast_messages: Mutex::new(Vec::new()),
10851
+
10813
10852
entropy_source: args.entropy_source,
10814
10853
node_signer: args.node_signer,
10815
10854
signer_provider: args.signer_provider,
0 commit comments