Allow RoutingMessageHandler to signal backpressure

Now that we allow `handle_channel_announcement` to (indirectly)
spawn async tasks which will complete later, we have to ensure it
can apply backpressure all the way up to the TCP socket to ensure
we don't end up with too many buffers allocated for UTXO
validation.

We do this by adding a new method to `RoutingMessageHandler` which
allows it to signal if there are "many" checks pending and
`channel_announcement` messages should be delayed. The actual
`PeerManager` implementation thereof is done in the next commit.

Author: TheBlueMatt

lightning-net-tokio/src/lib.rs

@@ -622,6 +622,7 @@ mod tests {
 		fn handle_query_short_channel_ids(&self, _their_node_id: &PublicKey, _msg: QueryShortChannelIds) -> Result<(), LightningError> { Ok(()) }
 		fn provided_node_features(&self) -> NodeFeatures { NodeFeatures::empty() }
 		fn provided_init_features(&self, _their_node_id: &PublicKey) -> InitFeatures { InitFeatures::empty() }
+		fn processing_queue_high(&self) -> bool { false }
 	}
 	impl ChannelMessageHandler for MsgHandler {
 		fn handle_open_channel(&self, _their_node_id: &PublicKey, _msg: &OpenChannel) {}

lightning/src/ln/msgs.rs

@@ -1080,6 +1080,14 @@ pub trait RoutingMessageHandler : MessageSendEventsProvider {
 	/// list of `short_channel_id`s.
 	fn handle_query_short_channel_ids(&self, their_node_id: &PublicKey, msg: QueryShortChannelIds) -> Result<(), LightningError>;
 
+	// Handler queueing status
+
+	/// Indicates that there are a large number of [`ChannelAnnouncement`] (or other) messages
+	/// pending some async action. While there is no guarantee of the rate of future messages, the
+	/// caller should seek to reduce the rate of new gossip messages handled, and especially
+	/// [`ChannelAnnouncement`]s.
+	fn processing_queue_high(&self) -> bool;
+
 	// Handler information:
 	/// Gets the node feature flags which this handler itself supports. All available handlers are
 	/// queried similarly and their feature flags are OR'd together to form the [`NodeFeatures`]

lightning/src/ln/peer_handler.rs

@@ -81,6 +81,7 @@ impl RoutingMessageHandler for IgnoringMessageHandler {
 	fn provided_init_features(&self, _their_node_id: &PublicKey) -> InitFeatures {
 		InitFeatures::empty()
 	}
+	fn processing_queue_high(&self) -> bool { false }
 }
 impl OnionMessageProvider for IgnoringMessageHandler {
 	fn next_onion_message_for_peer(&self, _peer_node_id: PublicKey) -> Option<msgs::OnionMessage> { None }

lightning/src/routing/gossip.rs

@@ -637,6 +637,10 @@ where C::Target: ChainAccess, L::Target: Logger
 		features.set_gossip_queries_optional();
 		features
 	}
+
+	fn processing_queue_high(&self) -> bool {
+		self.network_graph.pending_checks.too_many_checks_pending()
+	}
 }
 
 impl<G: Deref<Target=NetworkGraph<L>>, C: Deref, L: Deref> MessageSendEventsProvider for P2PGossipSync<G, C, L>

lightning/src/routing/gossip_checking.rs

@@ -477,4 +477,28 @@ impl PendingChecks {
 			}
 		}
 	}
+
+	const MAX_PENDING_LOOKUPS: usize = 32;
+
+	/// Signls if there are a large number of async checks pending and future channel_announcement
+	/// messages should be delayed. Note that this is only a hint and messages already in-flight
+	/// may still have to be handled for various reasons.
+	pub(super) fn too_many_checks_pending(&self) -> bool {
+		let mut pending_checks = self.internal.lock().unwrap();
+		if pending_checks.channels.len() > Self::MAX_PENDING_LOOKUPS {
+			// If we have many channel checks pending, ensure we don't have any dangling checks
+			// (i.e. checks where the user told us they'd call back but drop'd the `AccessFuture`
+			// instead) before we commit to applying backpressure.
+			pending_checks.channels.retain(|_, chan| {
+				Weak::upgrade(&chan).is_some()
+			});
+			pending_checks.nodes.retain(|_, channels| {
+				channels.retain(|chan| Weak::upgrade(&chan).is_some());
+				!channels.is_empty()
+			});
+			pending_checks.channels.len() > Self::MAX_PENDING_LOOKUPS
+		} else {
+			false
+		}
+	}
 }

lightning/src/util/test_utils.rs

@@ -550,6 +550,8 @@ impl msgs::RoutingMessageHandler for TestRoutingMessageHandler {
 		features.set_gossip_queries_optional();
 		features
 	}
+
+	fn processing_queue_high(&self) -> bool { false }
 }
 
 impl events::MessageSendEventsProvider for TestRoutingMessageHandler {