Buffer onion messages requiring a connection

MessageRouter::find_path returns a path to use when sending an onion
message. If the first node on the path is not connected or does not
support onion messages, sending will fail with InvalidFirstHop. Instead
of failing outright, buffer the message for later sending once the first
node is a connected peer.

Author: jkczyz

lightning/src/onion_message/messenger.rs

@@ -313,7 +313,7 @@ pub enum SendError {
 	/// hops.
 	TooFewBlindedHops,
 	/// Our next-hop peer was offline or does not support onion message forwarding.
-	InvalidFirstHop,
+	InvalidFirstHop(PublicKey, OnionMessage),
 	/// Onion message contents must have a TLV type >= 64.
 	InvalidMessage,
 	/// Our next-hop peer's buffer was full or our total outbound buffer was full.
@@ -580,7 +580,9 @@ where
 		}
 
 		match message_buffers.entry(first_node_id) {
-			hash_map::Entry::Vacant(_) => Err(SendError::InvalidFirstHop),
+			hash_map::Entry::Vacant(_) => {
+				Err(SendError::InvalidFirstHop(first_node_id, onion_message))
+			},
 			hash_map::Entry::Occupied(mut e) => {
 				e.get_mut().enqueue_message(onion_message);
 				Ok(())
@@ -617,7 +619,11 @@ where
 			}
 		};
 
-		let peers = self.message_buffers.lock().unwrap().keys().copied().collect();
+		let peers = self.message_buffers.lock().unwrap()
+			.iter()
+			.filter(|(_, buffer)| matches!(buffer, OnionMessageBuffer::ConnectedPeer(_)))
+			.map(|(node_id, _)| *node_id)
+			.collect();
 		let path = match self.message_router.find_path(sender, peers, destination) {
 			Ok(path) => path,
 			Err(()) => {
@@ -629,7 +635,19 @@ where
 		log_trace!(self.logger, "Sending onion message {}", log_suffix);
 
 		if let Err(e) = self.send_onion_message(path, contents, reply_path) {
-			log_trace!(self.logger, "Failed sending onion message {}: {:?}", log_suffix, e);
+			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);
+				log_trace!(
+					self.logger, "Buffered onion message waiting on peer connection {}: {:?}",
+					log_suffix, first_node_id
+				);
+			} else {
+				log_trace!(self.logger, "Failed sending onion message {}: {:?}", log_suffix, e);
+			}
 		}
 	}