OnionMessageHandler::release_pending_connections

An OnionMessageHandler may buffer messages that can't be sent because
the recipient is not a peer. Expand the interface with a method for
returning any nodes that fall into this category so that an
Event::ConnectionNeeded can be generated. Also, implement the new
interface for OnionMessenger.

Author: jkczyz

lightning/src/ln/msgs.rs

@@ -1580,6 +1580,17 @@ pub trait OnionMessageHandler {
 	/// drop and refuse to forward onion messages to this peer.
 	fn peer_disconnected(&self, their_node_id: &PublicKey);
 
+	/// Returns nodes that have messages buffered for calls to [`Self::next_onion_message_for_peer`]
+	/// but aren't connected as peers. Used to determine if an [`Event::ConnectionNeeded`] should be
+	/// generated.
+	///
+	/// Subsequent calls should not include the same nodes unless it's appropriate to generate
+	/// another event (e.g., if the implementation drops the buffered messages after a timeout but
+	/// more messages are then buffered).
+	///
+	/// [`Event::ConnectionNeeded`]: crate::events::Event::ConnectionNeeded
+	fn release_pending_connections(&self) -> Vec<NodeAnnouncement> { vec![] }
+
 	// 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/onion_message/messenger.rs

@@ -22,7 +22,7 @@ use crate::sign::{EntropySource, KeysManager, NodeSigner, Recipient};
 #[cfg(not(c_bindings))]
 use crate::ln::channelmanager::{SimpleArcChannelManager, SimpleRefChannelManager};
 use crate::ln::features::{InitFeatures, NodeFeatures};
-use crate::ln::msgs::{self, OnionMessage, OnionMessageHandler};
+use crate::ln::msgs::{self, NodeAnnouncement, OnionMessage, OnionMessageHandler};
 use crate::ln::onion_utils;
 use crate::ln::peer_handler::IgnoringMessageHandler;
 use crate::routing::gossip::{NetworkGraph, NodeId};
@@ -167,21 +167,21 @@ enum OnionMessageBuffer {
 	ConnectedPeer(VecDeque<OnionMessage>),
 
 	/// Messages for a node that is not yet connected.
-	PendingConnection(VecDeque<OnionMessage>),
+	PendingConnection(VecDeque<OnionMessage>, Option<NodeAnnouncement>),
 }
 
 impl OnionMessageBuffer {
 	fn pending_messages(&self) -> &VecDeque<OnionMessage> {
 		match self {
 			OnionMessageBuffer::ConnectedPeer(pending_messages) => pending_messages,
-			OnionMessageBuffer::PendingConnection(pending_messages) => pending_messages,
+			OnionMessageBuffer::PendingConnection(pending_messages, _) => pending_messages,
 		}
 	}
 
 	fn enqueue_message(&mut self, message: OnionMessage) {
 		let pending_messages = match self {
 			OnionMessageBuffer::ConnectedPeer(pending_messages) => pending_messages,
-			OnionMessageBuffer::PendingConnection(pending_messages) => pending_messages,
+			OnionMessageBuffer::PendingConnection(pending_messages, _) => pending_messages,
 		};
 
 		pending_messages.push_back(message);
@@ -190,7 +190,7 @@ impl OnionMessageBuffer {
 	fn dequeue_message(&mut self) -> Option<OnionMessage> {
 		let pending_messages = match self {
 			OnionMessageBuffer::ConnectedPeer(pending_messages) => pending_messages,
-			OnionMessageBuffer::PendingConnection(pending_messages) => pending_messages,
+			OnionMessageBuffer::PendingConnection(pending_messages, _) => pending_messages,
 		};
 
 		pending_messages.pop_front()
@@ -200,14 +200,14 @@ impl OnionMessageBuffer {
 	fn release_pending_messages(&mut self) -> VecDeque<OnionMessage> {
 		let pending_messages = match self {
 			OnionMessageBuffer::ConnectedPeer(pending_messages) => pending_messages,
-			OnionMessageBuffer::PendingConnection(pending_messages) => pending_messages,
+			OnionMessageBuffer::PendingConnection(pending_messages, _) => pending_messages,
 		};
 
 		core::mem::take(pending_messages)
 	}
 
 	fn mark_connected(&mut self) {
-		if let OnionMessageBuffer::PendingConnection(pending_messages) = self {
+		if let OnionMessageBuffer::PendingConnection(pending_messages, _) = self {
 			let mut new_pending_messages = VecDeque::new();
 			core::mem::swap(pending_messages, &mut new_pending_messages);
 			*self = OnionMessageBuffer::ConnectedPeer(new_pending_messages);
@@ -640,22 +640,36 @@ where
 			},
 		};
 
-		log_trace!(self.logger, "Sending onion message {}", log_suffix);
-
-		if let Err(e) = self.send_onion_message(path, contents, reply_path) {
-			if let SendError::InvalidFirstHop(first_node_id, onion_message) = e {
-				// TODO: Check NetworkGraph for node id and onion message support before enqueuing
-				self.message_buffers.lock().unwrap()
-					.entry(first_node_id)
-					.or_insert_with(|| OnionMessageBuffer::PendingConnection(VecDeque::new()))
-					.enqueue_message(onion_message);
+		match self.send_onion_message(path, contents, reply_path) {
+			Err(SendError::InvalidFirstHop(first_node_id, onion_message)) => {
+				let network_graph = self.network_graph.deref().read_only();
+				let node_announcement = network_graph
+					.node(&NodeId::from_pubkey(&first_node_id))
+					.and_then(|node_info| node_info.announcement_info.as_ref())
+					.and_then(|announcement_info| announcement_info.announcement_message.as_ref());
+
+				if let Some(node_announcement) = node_announcement {
+					if node_announcement.contents.features.supports_onion_messages() {
+						self.message_buffers.lock().unwrap()
+							.entry(first_node_id)
+							.or_insert_with(|| OnionMessageBuffer::PendingConnection(VecDeque::new(), Some(node_announcement.clone())))
+							.enqueue_message(onion_message);
+						log_trace!(
+							self.logger, "Buffered onion message waiting on peer connection {}: {:?}",
+							log_suffix, first_node_id
+						);
+						return;
+					}
+				}
 				log_trace!(
-					self.logger, "Buffered onion message waiting on peer connection {}: {:?}",
-					log_suffix, first_node_id
+					self.logger, "Failed sending onion message {}: {:?}", log_suffix,
+					SendError::InvalidFirstHop(first_node_id, onion_message)
 				);
-			} else {
+			},
+			Err(e) => {
 				log_trace!(self.logger, "Failed sending onion message {}: {:?}", log_suffix, e);
-			}
+			},
+			Ok(_) => log_trace!(self.logger, "Sending onion message {}", log_suffix),
 		}
 	}
 
@@ -782,6 +796,18 @@ where
 		self.message_buffers.lock().unwrap().remove(their_node_id);
 	}
 
+	fn release_pending_connections(&self) -> Vec<NodeAnnouncement> {
+		self.message_buffers.lock().unwrap()
+			.values_mut()
+			.filter_map(|recipient| match recipient {
+				OnionMessageBuffer::PendingConnection(_, node_announcement) => {
+					node_announcement.take()
+				},
+				_ => None,
+			})
+			.collect()
+	}
+
 	fn provided_node_features(&self) -> NodeFeatures {
 		let mut features = NodeFeatures::empty();
 		features.set_onion_messages_optional();