Always remove disconnected peers with no channels

When a peer disconnects but still has channels, the peer's `peer_state`
entry in the `per_peer_state` is not removed by the `peer_disconnected`
function. If the channels of to that peer is later closed while still
being disconnected (i.e. force closed), we therefore need to remove the
peer from `peer_state` separately.

To remove the peers separately, we push such peers to a separate HashSet
that holds peers awaiting removal, and remove the peers on a timer to
limit the negative effects on paralleism as much as possible.

Author: ViktorTigerstrom

lightning/src/ln/channelmanager.rs

@@ -464,6 +464,8 @@ pub(super) struct PeerState<Signer: Sign> {
 	/// Messages to send to the peer - pushed to in the same lock that they are generated in (except
 	/// for broadcast messages, where ordering isn't as strict).
 	pub(super) pending_msg_events: Vec<MessageSendEvent>,
+	/// Represents wether we're connected to the node or not.
+	connected: bool,
 }
 
 /// Stores a PaymentSecret and any other data we may need to validate an inbound payment is
@@ -593,6 +595,8 @@ pub type SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, M, T, F, L> = C
 //  |               |
 //  |               |__`best_block`
 //  |               |
+//  |               |__`pending_peers_awaiting_removal`
+//  |               |
 //  |               |__`pending_events`
 //  |                   |
 //  |                   |__`pending_background_events`
@@ -760,6 +764,16 @@ where
 
 	/// See `ChannelManager` struct-level documentation for lock order requirements.
 	pending_events: Mutex<Vec<events::Event>>,
+	/// When a peer disconnects but still has channels, the peer's `peer_state` entry in the
+	/// `per_peer_state` is not removed by the `peer_disconnected` function. If the channels of
+	/// to that peer is later closed while still being disconnected (i.e. force closed), we
+	/// therefore need to remove the peer from `peer_state` separately.
+	/// To avoid having to take the `per_peer_state` `write` lock once the channels are closed, we
+	/// instead store such peers awaiting removal in this field, and remove them on a timer to
+	/// limit the negative effects on parallelism as much as possible.
+	///
+	/// See `ChannelManager` struct-level documentation for lock order requirements.
+	pending_peers_awaiting_removal: Mutex<HashSet<PublicKey>>,
 	/// See `ChannelManager` struct-level documentation for lock order requirements.
 	pending_background_events: Mutex<Vec<BackgroundEvent>>,
 	/// Used when we have to take a BIG lock to make sure everything is self-consistent.
@@ -1290,10 +1304,11 @@ macro_rules! try_chan_entry {
 }
 
 macro_rules! remove_channel {
-	($self: expr, $entry: expr) => {
+	($self: expr, $entry: expr, $peer_state: expr) => {
 		{
 			let channel = $entry.remove_entry().1;
 			update_maps_on_chan_removal!($self, channel);
+			$self.add_pending_peer_to_be_removed(channel.get_counterparty_node_id(), $peer_state);
 			channel
 		}
 	}
@@ -1466,6 +1481,7 @@ where
 			per_peer_state: FairRwLock::new(HashMap::new()),
 
 			pending_events: Mutex::new(Vec::new()),
+			pending_peers_awaiting_removal: Mutex::new(HashSet::new()),
 			pending_background_events: Mutex::new(Vec::new()),
 			total_consistency_lock: RwLock::new(()),
 			persistence_notifier: Notifier::new(),
@@ -1704,7 +1720,7 @@ where
 						let (result, is_permanent) =
 							handle_monitor_update_res!(self, update_res, chan_entry.get_mut(), RAACommitmentOrder::CommitmentFirst, chan_entry.key(), NO_UPDATE);
 						if is_permanent {
-							remove_channel!(self, chan_entry);
+							remove_channel!(self, chan_entry, peer_state);
 							break result;
 						}
 					}
@@ -1715,7 +1731,7 @@ where
 					});
 
 					if chan_entry.get().is_shutdown() {
-						let channel = remove_channel!(self, chan_entry);
+						let channel = remove_channel!(self, chan_entry, peer_state);
 						if let Ok(channel_update) = self.get_channel_update_for_broadcast(&channel) {
 							peer_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
 								msg: channel_update
@@ -1818,7 +1834,7 @@ where
 				} else {
 					self.issue_channel_close_events(chan.get(),ClosureReason::HolderForceClosed);
 				}
-				remove_channel!(self, chan)
+				remove_channel!(self, chan, peer_state)
 			} else {
 				return Err(APIError::ChannelUnavailable{ err: format!("Channel with id {} not found for the passed counterparty node_id {}", log_bytes!(*channel_id), peer_node_id) });
 			}
@@ -1857,6 +1873,13 @@ where
 		}
 	}
 
+	fn add_pending_peer_to_be_removed(&self, counterparty_node_id: PublicKey, peer_state: &mut PeerState<<SP::Target as SignerProvider>::Signer>) {
+		let peer_should_be_removed = !peer_state.connected && peer_state.channel_by_id.len() == 0;
+		if peer_should_be_removed {
+			self.pending_peers_awaiting_removal.lock().unwrap().insert(counterparty_node_id);
+		}
+	}
+
 	/// Force closes a channel, immediately broadcasting the latest local transaction(s) and
 	/// rejecting new HTLCs on the given channel. Fails if `channel_id` is unknown to
 	/// the manager, or if the `counterparty_node_id` isn't the counterparty of the corresponding
@@ -3281,6 +3304,34 @@ where
 		true
 	}
 
+	/// Removes peers which have been been added to `pending_peers_awaiting_removal` which are
+	/// still disconnected and we have no channels to.
+	///
+	/// Must be called without the `per_peer_state` lock acquired.
+	fn remove_peers_awaiting_removal(&self) {
+		let mut pending_peers_awaiting_removal = HashSet::new();
+		mem::swap(&mut *self.pending_peers_awaiting_removal.lock().unwrap(), &mut pending_peers_awaiting_removal);
+		if pending_peers_awaiting_removal.len() > 0 {
+			let mut per_peer_state = self.per_peer_state.write().unwrap();
+			for counterparty_node_id in pending_peers_awaiting_removal.drain() {
+				match per_peer_state.entry(counterparty_node_id) {
+					hash_map::Entry::Occupied(entry) => {
+						// Remove the entry if the peer is still disconnected and we still
+						// have no channels to the peer.
+						let remove_entry = {
+							let peer_state = entry.get().lock().unwrap();
+							!peer_state.connected && peer_state.channel_by_id.len() == 0
+						};
+						if remove_entry {
+							entry.remove_entry();
+						}
+					},
+					hash_map::Entry::Vacant(_) => { /* The PeerState has already been removed */ }
+				}
+			}
+		}
+	}
+
 	#[cfg(any(test, feature = "_test_utils"))]
 	/// Process background events, for functional testing
 	pub fn test_process_background_events(&self) {
@@ -3359,13 +3410,14 @@ 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 counterparty_node_id = *counterparty_node_id;
 					peer_state.channel_by_id.retain(|chan_id, chan| {
 						let chan_needs_persist = self.update_channel_fee(chan_id, chan, new_feerate);
 						if chan_needs_persist == NotifyOption::DoPersist { should_persist = NotifyOption::DoPersist; }
 
 						if let Err(e) = chan.timer_check_closing_negotiation_progress() {
 							let (needs_close, err) = convert_chan_err!(self, e, chan, chan_id);
-							handle_errors.push((Err(err), *counterparty_node_id));
+							handle_errors.push((Err(err), counterparty_node_id));
 							if needs_close { return false; }
 						}
 
@@ -3399,8 +3451,10 @@ where
 
 						true
 					});
+					self.add_pending_peer_to_be_removed(counterparty_node_id, peer_state);
 				}
 			}
+			self.remove_peers_awaiting_removal();
 
 			self.claimable_payments.lock().unwrap().claimable_htlcs.retain(|payment_hash, (_, htlcs)| {
 				if htlcs.is_empty() {
@@ -4136,7 +4190,7 @@ where
 						}
 					};
 					peer_state.pending_msg_events.push(send_msg_err_event);
-					let _ = remove_channel!(self, channel);
+					let _ = remove_channel!(self, channel, peer_state);
 					return Err(APIError::APIMisuseError { err: "Please use accept_inbound_channel_from_trusted_peer_0conf to accept channels with zero confirmations.".to_owned() });
 				}
 
@@ -4422,7 +4476,7 @@ where
 						let (result, is_permanent) =
 							handle_monitor_update_res!(self, update_res, chan_entry.get_mut(), RAACommitmentOrder::CommitmentFirst, chan_entry.key(), NO_UPDATE);
 						if is_permanent {
-							remove_channel!(self, chan_entry);
+							remove_channel!(self, chan_entry, peer_state);
 							break result;
 						}
 					}
@@ -4471,7 +4525,7 @@ where
 						// also implies there are no pending HTLCs left on the channel, so we can
 						// fully delete it from tracking (the channel monitor is still around to
 						// watch for old state broadcasts)!
-						(tx, Some(remove_channel!(self, chan_entry)))
+						(tx, Some(remove_channel!(self, chan_entry, peer_state)))
 					} else { (tx, None) }
 				},
 				hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close(format!("Got a message for a channel from the wrong node! No such channel for the passed counterparty_node_id {}", counterparty_node_id), msg.channel_id))
@@ -4974,12 +5028,11 @@ where
 							if let Some(peer_state_mutex) = per_peer_state.get(&counterparty_node_id) {
 								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;
 								if let hash_map::Entry::Occupied(chan_entry) = peer_state.channel_by_id.entry(funding_outpoint.to_channel_id()) {
-									let mut chan = remove_channel!(self, chan_entry);
+									let mut chan = remove_channel!(self, chan_entry, peer_state);
 									failed_channels.push(chan.force_shutdown(false));
 									if let Ok(update) = self.get_channel_update_for_broadcast(&chan) {
-										pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+										peer_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
 											msg: update
 										});
 									}
@@ -4989,7 +5042,7 @@ where
 										ClosureReason::CommitmentTxConfirmed
 									};
 									self.issue_channel_close_events(&chan, reason);
-									pending_msg_events.push(events::MessageSendEvent::HandleError {
+									peer_state.pending_msg_events.push(events::MessageSendEvent::HandleError {
 										node_id: chan.get_counterparty_node_id(),
 										action: msgs::ErrorAction::SendErrorMessage {
 											msg: msgs::ErrorMessage { channel_id: chan.channel_id(), data: "Channel force-closed".to_owned() }
@@ -5031,7 +5084,7 @@ where
 		{
 			let per_peer_state = self.per_peer_state.read().unwrap();
 
-			for (_cp_id, peer_state_mutex) in per_peer_state.iter() {
+			for (cp_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;
 				let pending_msg_events = &mut peer_state.pending_msg_events;
@@ -5071,6 +5124,7 @@ where
 						}
 					}
 				});
+				self.add_pending_peer_to_be_removed(*cp_id, peer_state);
 			}
 		}
 
@@ -5095,7 +5149,7 @@ where
 		{
 			let per_peer_state = self.per_peer_state.read().unwrap();
 
-			for (_cp_id, peer_state_mutex) in per_peer_state.iter() {
+			for (cp_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;
 				let pending_msg_events = &mut peer_state.pending_msg_events;
@@ -5133,6 +5187,7 @@ where
 						}
 					}
 				});
+				self.add_pending_peer_to_be_removed(*cp_id, peer_state);
 			}
 		}
 
@@ -5696,7 +5751,7 @@ where
 		let mut timed_out_htlcs = Vec::new();
 		{
 			let per_peer_state = self.per_peer_state.read().unwrap();
-			for (_cp_id, peer_state_mutex) in per_peer_state.iter() {
+			for (cp_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;
 				let pending_msg_events = &mut peer_state.pending_msg_events;
@@ -5780,6 +5835,7 @@ where
 					}
 					true
 				});
+				self.add_pending_peer_to_be_removed(*cp_id, peer_state);
 			}
 		}
 
@@ -6026,6 +6082,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;
+				peer_state.connected = false;
 				peer_state.channel_by_id.retain(|_, chan| {
 					chan.remove_uncommitted_htlcs_and_mark_paused(&self.logger);
 					if chan.is_shutdown() {
@@ -6091,17 +6148,20 @@ where
 						channel_by_id: HashMap::new(),
 						latest_features: init_msg.features.clone(),
 						pending_msg_events: Vec::new(),
+						connected: true,
 					}));
 				},
 				hash_map::Entry::Occupied(e) => {
-					e.get().lock().unwrap().latest_features = init_msg.features.clone();
+					let mut peer_state = e.get().lock().unwrap();
+					peer_state.latest_features = init_msg.features.clone();
+					peer_state.connected = true;
 				},
 			}
 		}
 
 		let per_peer_state = self.per_peer_state.read().unwrap();
 
-		for (_cp_id, peer_state_mutex) in per_peer_state.iter() {
+		for (cp_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;
 			let pending_msg_events = &mut peer_state.pending_msg_events;
@@ -6133,6 +6193,7 @@ where
 				}
 				retain
 			});
+			self.add_pending_peer_to_be_removed(*cp_id, peer_state);
 		}
 		//TODO: Also re-broadcast announcement_signatures
 		Ok(())
@@ -6646,6 +6707,8 @@ where
 
 		write_ver_prefix!(writer, SERIALIZATION_VERSION, MIN_SERIALIZATION_VERSION);
 
+		self.remove_peers_awaiting_removal();
+
 		self.genesis_hash.write(writer)?;
 		{
 			let best_block = self.best_block.read().unwrap();
@@ -7113,6 +7176,7 @@ where
 				channel_by_id: peer_channels.remove(&peer_pubkey).unwrap_or(HashMap::new()),
 				latest_features: Readable::read(reader)?,
 				pending_msg_events: Vec::new(),
+				connected: false,
 			};
 			per_peer_state.insert(peer_pubkey, Mutex::new(peer_state));
 		}
@@ -7466,6 +7530,7 @@ where
 			per_peer_state: FairRwLock::new(per_peer_state),
 
 			pending_events: Mutex::new(pending_events_read),
+			pending_peers_awaiting_removal: Mutex::new(HashSet::new()),
 			pending_background_events: Mutex::new(pending_background_events_read),
 			total_consistency_lock: RwLock::new(()),
 			persistence_notifier: Notifier::new(),
@@ -7933,6 +7998,44 @@ 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 chan = create_announced_chan_between_nodes(&nodes, 0, 1);
+
+		nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false);
+		nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false);
+
+		nodes[0].node.force_close_broadcasting_latest_txn(&chan.2, &nodes[1].node.get_our_node_id()).unwrap();
+		check_closed_broadcast!(nodes[0], true);
+		check_added_monitors!(nodes[0], 1);
+		check_closed_event!(nodes[0], 1, ClosureReason::HolderForceClosed);
+
+		{
+			// Assert that nodes[1] is awaiting removal for nodes[0] once nodes[1] has been
+			// disconnected and the channel between has been force closed.
+			let nodes_0_per_peer_state = nodes[0].node.per_peer_state.read().unwrap();
+			let nodes_0_pending_peers_awaiting_removal = nodes[0].node.pending_peers_awaiting_removal.lock().unwrap();
+			assert_eq!(nodes_0_pending_peers_awaiting_removal.len(), 1);
+			assert!(nodes_0_pending_peers_awaiting_removal.get(&nodes[1].node.get_our_node_id()).is_some());
+			// Assert that nodes[1] isn't removed before `timer_tick_occurred` has been executed.
+			assert_eq!(nodes_0_per_peer_state.len(), 1);
+			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.pending_peers_awaiting_removal.lock().unwrap().len(), 0);
+		}
+	}
+
 	#[test]
 	fn bad_inbound_payment_hash() {
 		// Add coverage for checking that a user-provided payment hash matches the payment secret.