Cache peers in OffersMessageFlow

In upcoming commits, we'll be creating blinded paths during the process of
creating a revoke_and_ack message within the Channel struct. These paths will
be included in said RAA to be used as reply paths for often-offline senders
held_htlc_available messages.

Because we hold the per-peer lock corresponding to the Channel while creating
this RAA, we can't use our typical approach of calling
ChannelManager::get_peers_for_blinded_path to create these blinded paths.
The ::get_peers method takes each peer's lock in turn in order to check for
usable channels/onion message feature support, and it's not permitted to hold
multiple peer state locks at the same time due to the potential for deadlocks
(see the debug_sync module).

To avoid taking other peer state locks while holding a particular Channel's
peer state lock, here we cache the set of peers in the OffersMessageFlow, which
is the struct that ultimately creates the blinded paths for the RAA.

Author: valentinewallace

lightning/src/ln/channelmanager.rs

@@ -13156,6 +13156,8 @@ where
 		for (err, counterparty_node_id) in failed_channels.drain(..) {
 			let _ = handle_error!(self, err, counterparty_node_id);
 		}
+
+		let _ = self.flow.peer_disconnected(counterparty_node_id);
 	}
 
 	#[rustfmt::skip]
@@ -13167,14 +13169,15 @@ where
 		}
 
 		let mut res = Ok(());
-
+		let mut peer_has_live_channel = false;
 		PersistenceNotifierGuard::optionally_notify(self, || {
 			// If we have too many peers connected which don't have funded channels, disconnect the
 			// peer immediately (as long as it doesn't have funded channels). If we have a bunch of
 			// unfunded channels taking up space in memory for disconnected peers, we still let new
 			// peers connect, but we'll reject new channels from them.
 			let connected_peers_without_funded_channels = self.peers_without_funded_channels(|node| node.is_connected);
 			let inbound_peer_limited = inbound && connected_peers_without_funded_channels >= MAX_NO_CHANNEL_PEERS;
+			let best_block_height = self.best_block.read().unwrap().height;
 
 			{
 				let mut peer_state_lock = self.per_peer_state.write().unwrap();
@@ -13201,7 +13204,6 @@ where
 						let mut peer_state = e.get().lock().unwrap();
 						peer_state.latest_features = init_msg.features.clone();
 
-						let best_block_height = self.best_block.read().unwrap().height;
 						if inbound_peer_limited &&
 							Self::unfunded_channel_count(&*peer_state, best_block_height) ==
 							peer_state.channel_by_id.len()
@@ -13253,6 +13255,11 @@ where
 							}),
 						ReconnectionMsg::None => {},
 					}
+					if chan.context().is_usable()
+						&& chan.context().channel_creation_height <= best_block_height.saturating_sub(6)
+					{
+						peer_has_live_channel = true;
+					}
 				}
 			}
 
@@ -13265,6 +13272,7 @@ where
 		// until we have some peer connection(s) to receive onion messages over, so as a minor optimization
 		// refresh the cache when a peer connects.
 		self.check_refresh_async_receive_offer_cache(false);
+		let _ = self.flow.peer_connected(counterparty_node_id, &init_msg.features, peer_has_live_channel);
 		res
 	}
 

lightning/src/offers/flow.rs

@@ -61,6 +61,7 @@ use crate::sign::{EntropySource, NodeSigner, ReceiveAuthKey};
 
 use crate::offers::static_invoice::{StaticInvoice, StaticInvoiceBuilder};
 use crate::sync::{Mutex, RwLock};
+use crate::types::features::InitFeatures;
 use crate::types::payment::{PaymentHash, PaymentSecret};
 use crate::util::logger::Logger;
 use crate::util::ser::Writeable;
@@ -100,6 +101,7 @@ where
 
 	pending_async_payments_messages: Mutex<Vec<(AsyncPaymentsMessage, MessageSendInstructions)>>,
 	async_receive_offer_cache: Mutex<AsyncReceiveOfferCache>,
+	peers_cache: Mutex<Vec<MessageForwardNode>>,
 
 	#[cfg(feature = "dnssec")]
 	pub(crate) hrn_resolver: OMNameResolver,
@@ -136,6 +138,7 @@ where
 
 			pending_offers_messages: Mutex::new(Vec::new()),
 			pending_async_payments_messages: Mutex::new(Vec::new()),
+			peers_cache: Mutex::new(Vec::new()),
 
 			#[cfg(feature = "dnssec")]
 			hrn_resolver: OMNameResolver::new(current_timestamp, best_block.height),
@@ -1683,4 +1686,43 @@ where
 	pub fn writeable_async_receive_offer_cache(&self) -> Vec<u8> {
 		self.async_receive_offer_cache.encode()
 	}
+
+	/// Indicates that a peer was connected to our node. Useful for the [`OffersMessageFlow`] to keep
+	/// track of which peers are connected, which allows for methods that can create blinded paths
+	/// without requiring a fresh set of [`MessageForwardNode`]s to be passed in.
+	///
+	/// `live_channel` should be set if we have an open, usable channel with this peer that has at
+	/// least six onchain confirmations.
+	///
+	/// MUST be called by always-online nodes that support holding HTLCs on behalf of often-offline
+	/// senders.
+	///
+	/// Errors if the peer does not support onion messages or we don't have a channel with them.
+	pub fn peer_connected(
+		&self, peer_node_id: PublicKey, features: &InitFeatures, live_channel: bool,
+	) -> Result<(), ()> {
+		if !features.supports_onion_messages() || !live_channel {
+			return Err(());
+		}
+
+		let mut peers_cache = self.peers_cache.lock().unwrap();
+		let peer = MessageForwardNode { node_id: peer_node_id, short_channel_id: None };
+		peers_cache.push(peer);
+
+		Ok(())
+	}
+
+	/// Indicates that a peer was disconnected from our node. See [`Self::peer_connected`].
+	///
+	/// Errors if the peer is unknown.
+	pub fn peer_disconnected(&self, peer_node_id: PublicKey) -> Result<(), ()> {
+		let mut peers_cache = self.peers_cache.lock().unwrap();
+		let peer_idx = match peers_cache.iter().position(|peer| peer.node_id == peer_node_id) {
+			Some(idx) => idx,
+			None => return Err(()),
+		};
+		peers_cache.swap_remove(peer_idx);
+
+		Ok(())
+	}
 }