Invoice server: treat forwarded invreqs as OM forwards

diff --git a/lightning/src/events/mod.rs b/lightning/src/events/mod.rs
index df89894..001b696 100644
--- a/lightning/src/events/mod.rs
+++ b/lightning/src/events/mod.rs
@@ -1617,6 +1617,9 @@ pub enum Event {
 	/// `OnionMessenger` was initialized with
 	/// [`OnionMessenger::new_with_offline_peer_interception`], see its docs.
 	///
+	/// The offline peer should be awoken if possible on receipt of this event, such as via the LSPS5
+	/// protocol.
+	///
 	/// # Failure Behavior and Persistence
 	/// This event will eventually be replayed after failures-to-handle (i.e., the event handler
 	/// returning `Err(ReplayEvent ())`), but won't be persisted across restarts.
@@ -1661,6 +1664,14 @@ pub enum Event {
 		/// recipient is online to provide a new invoice. This path should be persisted and
 		/// later provided to [`ChannelManager::respond_to_static_invoice_request`].
 		///
+		/// This path's [`BlindedMessagePath::introduction_node`] MUST be set to our node or one of our
+		/// peers. This is because, for DoS protection, invoice requests forwarded over this path are
+		/// treated by our node like any other onion message forward and will not generate
+		/// [`Event::ConnectionNeeded`] if the first hop in the path is not our peer.
+		///
+		/// If the next-hop peer in the path is offline, if configured to do so we will generate an
+		/// [`Event::OnionMessageIntercepted`] for the invoice request.
+		///
 		/// [`ChannelManager::respond_to_static_invoice_request`]: crate::ln::channelmanager::ChannelManager::respond_to_static_invoice_request
 		invoice_request_path: BlindedMessagePath,
 		/// Useful for the recipient to replace a specific invoice stored by us as the static invoice
diff --git a/lightning/src/ln/async_payments_tests.rs b/lightning/src/ln/async_payments_tests.rs
index ccef448..9d327d0 100644
--- a/lightning/src/ln/async_payments_tests.rs
+++ b/lightning/src/ln/async_payments_tests.rs
@@ -2887,6 +2887,21 @@ fn async_payment_e2e() {
 		.into_iter()
 		.find_map(|ev| {
 			if let Event::OnionMessageIntercepted { message, .. } = ev {
+				// At least one of the intercepted onion messages will be an invoice request that the
+				// invoice server is attempting to forward to the recipient, ignore that as we're testing
+				// the static invoice flow
+				let peeled_onion = recipient.onion_messenger.peel_onion_message(&message).unwrap();
+				if matches!(
+					peeled_onion,
+					PeeledOnion::Offers(OffersMessage::InvoiceRequest { .. }, _, _)
+				) {
+					return None;
+				}
+
+				assert!(matches!(
+					peeled_onion,
+					PeeledOnion::AsyncPayments(AsyncPaymentsMessage::HeldHtlcAvailable(_), _, _)
+				));
 				Some(message)
 			} else {
 				None
diff --git a/lightning/src/offers/flow.rs b/lightning/src/offers/flow.rs
index a6484f0..6b0132f 100644
--- a/lightning/src/offers/flow.rs
+++ b/lightning/src/offers/flow.rs
@@ -1169,9 +1169,9 @@ where
 	) {
 		let mut pending_offers_messages = self.pending_offers_messages.lock().unwrap();
 		let message = OffersMessage::InvoiceRequest(invoice_request);
-		let instructions = MessageSendInstructions::WithSpecifiedReplyPath {
+		let instructions = MessageSendInstructions::ForwardedMessage {
 			destination: Destination::BlindedPath(destination),
-			reply_path: reply_path.into_blinded_path(),
+			reply_path: Some(reply_path.into_blinded_path()),
 		};
 		pending_offers_messages.push((message, instructions));
 	}
diff --git a/lightning/src/onion_message/async_payments.rs b/lightning/src/onion_message/async_payments.rs
index 877af43..127126e 100644
--- a/lightning/src/onion_message/async_payments.rs
+++ b/lightning/src/onion_message/async_payments.rs
@@ -169,6 +169,11 @@ pub struct ServeStaticInvoice {
 	/// [`Bolt12Invoice`] if the recipient is online at the time. Use this path to forward the
 	/// [`InvoiceRequest`] to the async recipient.
 	///
+	/// This path's [`BlindedMessagePath::introduction_node`] MUST be set to the static invoice server
+	/// node or one of its peers. This is because, for DoS protection, invoice requests forwarded over
+	/// this path are treated by the server node like any other onion message forward and the server
+	/// will not directly connect to the introduction node if they are not already peers.
+	///
 	/// [`InvoiceRequest`]: crate::offers::invoice_request::InvoiceRequest
 	/// [`Bolt12Invoice`]: crate::offers::invoice::Bolt12Invoice
 	pub forward_invoice_request_path: BlindedMessagePath,
diff --git a/lightning/src/onion_message/messenger.rs b/lightning/src/onion_message/messenger.rs
index afdd97a..cb66515 100644
--- a/lightning/src/onion_message/messenger.rs
+++ b/lightning/src/onion_message/messenger.rs
@@ -490,6 +490,21 @@ pub enum MessageSendInstructions {
 		/// The instructions provided by the [`Responder`].
 		instructions: ResponseInstruction,
 	},
+	/// Indicates that this onion message did not originate from our node and is being forwarded
+	/// through us from another node on the network to the destination.
+	///
+	/// We separate out this case because forwarded onion messages are treated differently from
+	/// outbound onion messages initiated by our node. Outbounds are buffered internally, whereas, for
+	/// DoS protection, forwards should never be buffered internally and instead will either be
+	/// dropped or generate an [`Event::OnionMessageIntercepted`] if the next-hop node is
+	/// disconnected.
+	ForwardedMessage {
+		/// The destination where we need to send the forwarded onion message.
+		destination: Destination,
+		/// The reply path which should be included in the message, that terminates at the original
+		/// sender of this forwarded message.
+		reply_path: Option<BlindedMessagePath>,
+	},
 }
 
 /// A trait defining behavior for routing an [`OnionMessage`].
@@ -1467,6 +1482,7 @@ where
 	fn send_onion_message_internal<T: OnionMessageContents>(
 		&self, contents: T, instructions: MessageSendInstructions, log_suffix: fmt::Arguments,
 	) -> Result<SendSuccess, SendError> {
+		let is_forward = matches!(instructions, MessageSendInstructions::ForwardedMessage { .. });
 		let (destination, reply_path) = match instructions {
 			MessageSendInstructions::WithSpecifiedReplyPath { destination, reply_path } => {
 				(destination, Some(reply_path))
@@ -1490,12 +1506,24 @@ where
 			| MessageSendInstructions::ForReply {
 				instructions: ResponseInstruction { destination, context: None },
 			} => (destination, None),
+			MessageSendInstructions::ForwardedMessage { destination, reply_path } => {
+				(destination, reply_path)
+			},
 		};
 
-		let path = self.find_path(destination).map_err(|e| {
-			log_trace!(self.logger, "Failed to find path {}", log_suffix);
-			e
-		})?;
+		let path = if is_forward {
+			// If this onion message is being treated as a forward, we shouldn't pathfind to the next hop.
+			OnionMessagePath {
+				intermediate_nodes: Vec::new(),
+				first_node_addresses: None,
+				destination,
+			}
+		} else {
+			self.find_path(destination).map_err(|e| {
+				log_trace!(self.logger, "Failed to find path {}", log_suffix);
+				e
+			})?
+		};
 		let first_hop = path.intermediate_nodes.get(0).map(|p| *p);
 		let logger = WithContext::from(&self.logger, first_hop, None, None);
 
@@ -1514,7 +1542,17 @@ where
 			e
 		})?;
 
-		let result = self.enqueue_outbound_onion_message(onion_message, first_node_id, addresses);
+		let result = if is_forward {
+			self.enqueue_forwarded_onion_message(
+				NextMessageHop::NodeId(first_node_id),
+				onion_message,
+				log_suffix,
+			)
+			.map(|()| SendSuccess::Buffered)
+		} else {
+			self.enqueue_outbound_onion_message(onion_message, first_node_id, addresses)
+		};
+
 		match result.as_ref() {
 			Err(SendError::GetNodeIdFailed) => {
 				log_warn!(logger, "Unable to retrieve node id {}", log_suffix);