Speed up remove_stale_channels_and_tracking nontrivially

During startup, the lightning protocol forces us to fetch a ton of
gossip for channels where there is a `channel_update` in only one
direction. We then have to wait around a while until we can prune
the crap cause we don't know when the gossip sync has completed.

Sadly, doing a large prune via `remove_stale_channels_and_tracking`
is somewhat slow. Removing a large portion of our graph currently
takes a bit more than 7.5 seconds on an i9-14900K, which can
ultimately ~hang a node with a few less GHz ~forever.

The bulk of this time is in our `IndexedMap` removals, where we
walk the entire `keys` `Vec` to remove the entry, then shift it
down after removing.

In the previous commit we shifted to a bulk removal model for
channels, here we do the same for nodes. This reduces the same test
to around 340 milliseconds on the same hardware.

Author: TheBlueMatt

lightning/src/routing/gossip.rs

@@ -37,7 +37,9 @@ use crate::ln::types::ChannelId;
 use crate::routing::utxo::{self, UtxoLookup, UtxoResolver};
 use crate::types::features::{ChannelFeatures, InitFeatures, NodeFeatures};
 use crate::types::string::PrintableString;
-use crate::util::indexed_map::{Entry as IndexedMapEntry, IndexedMap};
+use crate::util::indexed_map::{
+	Entry as IndexedMapEntry, IndexedMap, OccupiedEntry as IndexedMapOccupiedEntry,
+};
 use crate::util::logger::{Level, Logger};
 use crate::util::scid_utils::{block_from_scid, scid_from_parts, MAX_SCID_BLOCK};
 use crate::util::ser::{MaybeReadable, Readable, ReadableArgs, RequiredWrapper, Writeable, Writer};
@@ -2389,11 +2391,15 @@ where
 			let mut removed_channels_lck = self.removed_channels.lock().unwrap();
 
 			let channels_removed_bulk = channels.remove_fetch_bulk(&scids_to_remove);
-			removed_channels_lck.reserve(channels_removed_bulk.len());
+			self.removed_node_counters.lock().unwrap().reserve(channels_removed_bulk.len());
+			let mut nodes_to_remove = hash_set_with_capacity(channels_removed_bulk.len());
 			for (scid, info) in channels_removed_bulk {
-				self.remove_channel_in_nodes(&mut nodes, &info, scid);
+				self.remove_channel_in_nodes_callback(&mut nodes, &info, scid, |e| {
+					nodes_to_remove.insert(*e.key());
+				});
 				removed_channels_lck.insert(scid, Some(current_time_unix));
 			}
+			nodes.remove_bulk(&nodes_to_remove);
 		}
 
 		let should_keep_tracking = |time: &mut Option<u64>| {
@@ -2632,16 +2638,17 @@ where
 		Ok(())
 	}
 
-	fn remove_channel_in_nodes(
+	fn remove_channel_in_nodes_callback<RM: FnMut(IndexedMapOccupiedEntry<NodeId, NodeInfo>)>(
 		&self, nodes: &mut IndexedMap<NodeId, NodeInfo>, chan: &ChannelInfo, short_channel_id: u64,
+		mut remove_node: RM,
 	) {
 		macro_rules! remove_from_node {
 			($node_id: expr) => {
 				if let IndexedMapEntry::Occupied(mut entry) = nodes.entry($node_id) {
 					entry.get_mut().channels.retain(|chan_id| short_channel_id != *chan_id);
 					if entry.get().channels.is_empty() {
 						self.removed_node_counters.lock().unwrap().push(entry.get().node_counter);
-						entry.remove_entry();
+						remove_node(entry);
 					}
 				} else {
 					panic!(
@@ -2654,6 +2661,14 @@ where
 		remove_from_node!(chan.node_one);
 		remove_from_node!(chan.node_two);
 	}
+
+	fn remove_channel_in_nodes(
+		&self, nodes: &mut IndexedMap<NodeId, NodeInfo>, chan: &ChannelInfo, short_channel_id: u64,
+	) {
+		self.remove_channel_in_nodes_callback(nodes, chan, short_channel_id, |e| {
+			e.remove_entry();
+		});
+	}
 }
 
 impl ReadOnlyNetworkGraph<'_> {

lightning/src/util/indexed_map.rs

@@ -84,6 +84,14 @@ impl<K: Clone + Hash + Ord, V> IndexedMap<K, V> {
 		res
 	}
 
+	/// Removes elements with the given `keys` in bulk.
+	pub fn remove_bulk(&mut self, keys: &HashSet<K>) {
+		for key in keys.iter() {
+			self.map.remove(key);
+		}
+		self.keys.retain(|k| !keys.contains(k));
+	}
+
 	/// Inserts the given `key`/`value` pair into the map, returning the element that was
 	/// previously stored at the given `key`, if one exists.
 	pub fn insert(&mut self, key: K, value: V) -> Option<V> {
@@ -222,6 +230,11 @@ impl<'a, K: Hash + Ord, V> OccupiedEntry<'a, K, V> {
 		res
 	}
 
+	/// Get a reference to the key at the position described by this entry.
+	pub fn key(&self) -> &K {
+		self.underlying_entry.key()
+	}
+
 	/// Get a reference to the value at the position described by this entry.
 	pub fn get(&self) -> &V {
 		self.underlying_entry.get()