Extract util to forward onion messages

In an upcoming commit, we will need to repurpose this logic when, as a static
invoice server node, we update our support for forwarding invoice requests from
payers to often-offline recipients. We currently treat these forwarded invreqs
as outbound onion messages that are initiated by our node, when they should
really be treated as forwarded onion messages.

Author: valentinewallace

fuzz/src/onion_message.rs

@@ -430,7 +430,7 @@ mod tests {
 		super::do_test(&<Vec<u8>>::from_hex(two_unblinded_hops_om).unwrap(), &logger);
 		{
 			let log_entries = logger.lines.lock().unwrap();
-			assert_eq!(log_entries.get(&("lightning::onion_message::messenger".to_string(), "Forwarding an onion message to peer 020202020202020202020202020202020202020202020202020202020202020202".to_string())), Some(&1));
+			assert_eq!(log_entries.get(&("lightning::onion_message::messenger".to_string(), "Forwarding an onion message to peer 020202020202020202020202020202020202020202020202020202020202020202 when forwarding peeled onion message from 020000000000000000000000000000000000000000000000000000000000000002".to_string())), Some(&1));
 		}
 
 		let two_unblinded_two_blinded_om = "\
@@ -471,7 +471,7 @@ mod tests {
 		super::do_test(&<Vec<u8>>::from_hex(two_unblinded_two_blinded_om).unwrap(), &logger);
 		{
 			let log_entries = logger.lines.lock().unwrap();
-			assert_eq!(log_entries.get(&("lightning::onion_message::messenger".to_string(), "Forwarding an onion message to peer 020202020202020202020202020202020202020202020202020202020202020202".to_string())), Some(&1));
+			assert_eq!(log_entries.get(&("lightning::onion_message::messenger".to_string(), "Forwarding an onion message to peer 020202020202020202020202020202020202020202020202020202020202020202 when forwarding peeled onion message from 020000000000000000000000000000000000000000000000000000000000000002".to_string())), Some(&1));
 		}
 
 		let three_blinded_om = "\
@@ -512,7 +512,7 @@ mod tests {
 		super::do_test(&<Vec<u8>>::from_hex(three_blinded_om).unwrap(), &logger);
 		{
 			let log_entries = logger.lines.lock().unwrap();
-			assert_eq!(log_entries.get(&("lightning::onion_message::messenger".to_string(), "Forwarding an onion message to peer 020202020202020202020202020202020202020202020202020202020202020202".to_string())), Some(&1));
+			assert_eq!(log_entries.get(&("lightning::onion_message::messenger".to_string(), "Forwarding an onion message to peer 020202020202020202020202020202020202020202020202020202020202020202 when forwarding peeled onion message from 020000000000000000000000000000000000000000000000000000000000000002".to_string())), Some(&1));
 		}
 	}
 }

lightning/src/onion_message/messenger.rs

@@ -1620,6 +1620,74 @@ where
 		}
 	}
 
+	fn enqueue_forwarded_onion_message(
+		&self, next_hop: NextMessageHop, onion_message: OnionMessage, log_suffix: fmt::Arguments,
+	) -> Result<(), SendError> {
+		let next_node_id = match next_hop {
+			NextMessageHop::NodeId(pubkey) => pubkey,
+			NextMessageHop::ShortChannelId(scid) => match self.node_id_lookup.next_node_id(scid) {
+				Some(pubkey) => pubkey,
+				None => {
+					log_trace!(self.logger, "Dropping forwarded onion messager: unable to resolve next hop using SCID {} {}", scid, log_suffix);
+					return Err(SendError::GetNodeIdFailed);
+				},
+			},
+		};
+
+		let mut message_recipients = self.message_recipients.lock().unwrap();
+		if outbound_buffer_full(&next_node_id, &message_recipients) {
+			log_trace!(
+				self.logger,
+				"Dropping forwarded onion message to peer {}: outbound buffer full {}",
+				next_node_id,
+				log_suffix
+			);
+			return Err(SendError::BufferFull);
+		}
+
+		#[cfg(fuzzing)]
+		message_recipients
+			.entry(next_node_id)
+			.or_insert_with(|| OnionMessageRecipient::ConnectedPeer(VecDeque::new()));
+
+		match message_recipients.entry(next_node_id) {
+			hash_map::Entry::Occupied(mut e)
+				if matches!(e.get(), OnionMessageRecipient::ConnectedPeer(..)) =>
+			{
+				e.get_mut().enqueue_message(onion_message);
+				log_trace!(
+					self.logger,
+					"Forwarding an onion message to peer {} {}",
+					next_node_id,
+					log_suffix
+				);
+				Ok(())
+			},
+			_ if self.intercept_messages_for_offline_peers => {
+				log_trace!(
+					self.logger,
+					"Generating OnionMessageIntercepted event for peer {} {}",
+					next_node_id,
+					log_suffix
+				);
+				self.enqueue_intercepted_event(Event::OnionMessageIntercepted {
+					peer_node_id: next_node_id,
+					message: onion_message,
+				});
+				Ok(())
+			},
+			_ => {
+				log_trace!(
+					self.logger,
+					"Dropping forwarded onion message to disconnected peer {} {}",
+					next_node_id,
+					log_suffix
+				);
+				Err(SendError::InvalidFirstHop(next_node_id))
+			},
+		}
+	}
+
 	/// Forwards an [`OnionMessage`] to `peer_node_id`. Useful if we initialized
 	/// the [`OnionMessenger`] with [`Self::new_with_offline_peer_interception`]
 	/// and want to forward a previously intercepted onion message to a peer that
@@ -2204,56 +2272,11 @@ where
 				}
 			},
 			Ok(PeeledOnion::Forward(next_hop, onion_message)) => {
-				let next_node_id = match next_hop {
-					NextMessageHop::NodeId(pubkey) => pubkey,
-					NextMessageHop::ShortChannelId(scid) => {
-						match self.node_id_lookup.next_node_id(scid) {
-							Some(pubkey) => pubkey,
-							None => {
-								log_trace!(self.logger, "Dropping forwarded onion messager: unable to resolve next hop using SCID {}", scid);
-								return;
-							},
-						}
-					},
-				};
-
-				let mut message_recipients = self.message_recipients.lock().unwrap();
-				if outbound_buffer_full(&next_node_id, &message_recipients) {
-					log_trace!(
-						logger,
-						"Dropping forwarded onion message to peer {}: outbound buffer full",
-						next_node_id
-					);
-					return;
-				}
-
-				#[cfg(fuzzing)]
-				message_recipients
-					.entry(next_node_id)
-					.or_insert_with(|| OnionMessageRecipient::ConnectedPeer(VecDeque::new()));
-
-				match message_recipients.entry(next_node_id) {
-					hash_map::Entry::Occupied(mut e)
-						if matches!(e.get(), OnionMessageRecipient::ConnectedPeer(..)) =>
-					{
-						e.get_mut().enqueue_message(onion_message);
-						log_trace!(logger, "Forwarding an onion message to peer {}", next_node_id);
-					},
-					_ if self.intercept_messages_for_offline_peers => {
-						self.enqueue_intercepted_event(Event::OnionMessageIntercepted {
-							peer_node_id: next_node_id,
-							message: onion_message,
-						});
-					},
-					_ => {
-						log_trace!(
-							logger,
-							"Dropping forwarded onion message to disconnected peer {}",
-							next_node_id
-						);
-						return;
-					},
-				}
+				let _ = self.enqueue_forwarded_onion_message(
+					next_hop,
+					onion_message,
+					format_args!("when forwarding peeled onion message from {}", peer_node_id),
+				);
 			},
 			Err(e) => {
 				log_error!(logger, "Failed to process onion message {:?}", e);