2/3 Use MessageSendInstructions instead of PendingOnionMessage

Now that the `MessageRouter` can `create_blinded_paths` forcing
callers of the `OnionMessenger` to provide it with a reply path up
front is unnecessary complexity, doubly so in message handlers.

Here we take the next step towards untangling that, moving from
`PendingOnionMessage` to `MessageSendInstructions` for the outbound
message queue in `OffersMessageHandler`. Better, we can also drop
the `c_bindings`-specific message queue variant, unifying the APIs.

Because `ChannelManager` needs to actually control the reply path
set in individual messages, however, we have to halfway this patch,
adding a new `MessageSendInstructions` variant that allows
specifying the `reply_path` explicitly. Still, because other message
handlers are moving this way, its nice to be consistent.

By unifying the response type with the outbound message queue type,
this also fixes #3178.

Author: TheBlueMatt

lightning/src/ln/channelmanager.rs

@@ -71,7 +71,7 @@ use crate::offers::parse::Bolt12SemanticError;
 use crate::offers::refund::{Refund, RefundBuilder};
 use crate::offers::signer;
 use crate::onion_message::async_payments::{AsyncPaymentsMessage, HeldHtlcAvailable, ReleaseHeldHtlc, AsyncPaymentsMessageHandler};
-use crate::onion_message::messenger::{new_pending_onion_message, Destination, MessageRouter, PendingOnionMessage, Responder, MessageSendInstructions};
+use crate::onion_message::messenger::{Destination, MessageRouter, PendingOnionMessage, Responder, MessageSendInstructions};
 use crate::onion_message::offers::{OffersMessage, OffersMessageHandler};
 use crate::sign::{EntropySource, NodeSigner, Recipient, SignerProvider};
 use crate::sign::ecdsa::EcdsaChannelSigner;
@@ -2277,9 +2277,9 @@ where
 	needs_persist_flag: AtomicBool,
 
 	#[cfg(not(any(test, feature = "_test_utils")))]
-	pending_offers_messages: Mutex<Vec<PendingOnionMessage<OffersMessage>>>,
+	pending_offers_messages: Mutex<Vec<(OffersMessage, MessageSendInstructions)>>,
 	#[cfg(any(test, feature = "_test_utils"))]
-	pub(crate) pending_offers_messages: Mutex<Vec<PendingOnionMessage<OffersMessage>>>,
+	pub(crate) pending_offers_messages: Mutex<Vec<(OffersMessage, MessageSendInstructions)>>,
 
 	/// Tracks the message events that are to be broadcasted when we are connected to some peer.
 	pending_broadcast_messages: Mutex<Vec<MessageSendEvent>>,
@@ -9068,21 +9068,21 @@ where
 				.flat_map(|reply_path| offer.paths().iter().map(move |path| (path, reply_path)))
 				.take(OFFERS_MESSAGE_REQUEST_LIMIT)
 				.for_each(|(path, reply_path)| {
-					let message = new_pending_onion_message(
-						OffersMessage::InvoiceRequest(invoice_request.clone()),
-						Destination::BlindedPath(path.clone()),
-						Some(reply_path.clone()),
-					);
-					pending_offers_messages.push(message);
+					let instructions = MessageSendInstructions::WithSpecifiedReplyPath {
+						destination: Destination::BlindedPath(path.clone()),
+						reply_path: reply_path.clone(),
+					};
+					let message = OffersMessage::InvoiceRequest(invoice_request.clone());
+					pending_offers_messages.push((message, instructions));
 				});
 		} else if let Some(signing_pubkey) = offer.signing_pubkey() {
 			for reply_path in reply_paths {
-				let message = new_pending_onion_message(
-					OffersMessage::InvoiceRequest(invoice_request.clone()),
-					Destination::Node(signing_pubkey),
-					Some(reply_path),
-				);
-				pending_offers_messages.push(message);
+				let instructions = MessageSendInstructions::WithSpecifiedReplyPath {
+					destination: Destination::Node(signing_pubkey),
+					reply_path,
+				};
+				let message = OffersMessage::InvoiceRequest(invoice_request.clone());
+				pending_offers_messages.push((message, instructions));
 			}
 		} else {
 			debug_assert!(false);
@@ -9162,25 +9162,25 @@ where
 				let mut pending_offers_messages = self.pending_offers_messages.lock().unwrap();
 				if refund.paths().is_empty() {
 					for reply_path in reply_paths {
-						let message = new_pending_onion_message(
-							OffersMessage::Invoice(invoice.clone()),
-							Destination::Node(refund.payer_id()),
-							Some(reply_path),
-						);
-						pending_offers_messages.push(message);
+						let instructions = MessageSendInstructions::WithSpecifiedReplyPath {
+							destination: Destination::Node(refund.payer_id()),
+							reply_path,
+						};
+						let message = OffersMessage::Invoice(invoice.clone());
+						pending_offers_messages.push((message, instructions));
 					}
 				} else {
 					reply_paths
 						.iter()
 						.flat_map(|reply_path| refund.paths().iter().map(move |path| (path, reply_path)))
 						.take(OFFERS_MESSAGE_REQUEST_LIMIT)
 						.for_each(|(path, reply_path)| {
-							let message = new_pending_onion_message(
-								OffersMessage::Invoice(invoice.clone()),
-								Destination::BlindedPath(path.clone()),
-								Some(reply_path.clone()),
-							);
-							pending_offers_messages.push(message);
+							let instructions = MessageSendInstructions::WithSpecifiedReplyPath {
+								destination: Destination::BlindedPath(path.clone()),
+								reply_path: reply_path.clone(),
+							};
+							let message = OffersMessage::Invoice(invoice.clone());
+							pending_offers_messages.push((message, instructions));
 						});
 				}
 
@@ -10937,7 +10937,7 @@ where
 		}
 	}
 
-	fn release_pending_messages(&self) -> Vec<PendingOnionMessage<OffersMessage>> {
+	fn release_pending_messages(&self) -> Vec<(OffersMessage, MessageSendInstructions)> {
 		core::mem::take(&mut self.pending_offers_messages.lock().unwrap())
 	}
 }

lightning/src/ln/offers_tests.rs

@@ -59,7 +59,7 @@ use crate::offers::invoice_error::InvoiceError;
 use crate::offers::invoice_request::{InvoiceRequest, InvoiceRequestFields};
 use crate::offers::nonce::Nonce;
 use crate::offers::parse::Bolt12SemanticError;
-use crate::onion_message::messenger::{Destination, PeeledOnion, new_pending_onion_message};
+use crate::onion_message::messenger::{Destination, PeeledOnion, MessageSendInstructions};
 use crate::onion_message::offers::OffersMessage;
 use crate::onion_message::packet::ParsedOnionMessageContents;
 use crate::routing::gossip::{NodeAlias, NodeId};
@@ -1313,13 +1313,10 @@ fn fails_authentication_when_handling_invoice_request() {
 	expect_recent_payment!(david, RecentPaymentDetails::AwaitingInvoice, payment_id);
 
 	connect_peers(david, alice);
-	#[cfg(not(c_bindings))] {
-		david.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().destination =
-			Destination::Node(alice_id);
-	}
-	#[cfg(c_bindings)] {
-		david.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 =
-			Destination::Node(alice_id);
+	match &mut david.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
+		MessageSendInstructions::WithSpecifiedReplyPath { destination, .. } =>
+			*destination = Destination::Node(alice_id),
+		_ => panic!(),
 	}
 
 	let onion_message = david.onion_messenger.next_onion_message_for_peer(alice_id).unwrap();
@@ -1341,13 +1338,10 @@ fn fails_authentication_when_handling_invoice_request() {
 		.unwrap();
 	expect_recent_payment!(david, RecentPaymentDetails::AwaitingInvoice, payment_id);
 
-	#[cfg(not(c_bindings))] {
-		david.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().destination =
-			Destination::BlindedPath(invalid_path);
-	}
-	#[cfg(c_bindings)] {
-		david.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 =
-			Destination::BlindedPath(invalid_path);
+	match &mut david.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
+		MessageSendInstructions::WithSpecifiedReplyPath { destination, .. } =>
+			*destination = Destination::BlindedPath(invalid_path),
+		_ => panic!(),
 	}
 
 	connect_peers(david, bob);
@@ -1427,11 +1421,9 @@ fn fails_authentication_when_handling_invoice_for_offer() {
 		let mut pending_offers_messages = david.node.pending_offers_messages.lock().unwrap();
 		let pending_invoice_request = pending_offers_messages.pop().unwrap();
 		pending_offers_messages.clear();
-		#[cfg(not(c_bindings))] {
-			pending_invoice_request.reply_path
-		}
-		#[cfg(c_bindings)] {
-			pending_invoice_request.2
+		match pending_invoice_request.1 {
+			MessageSendInstructions::WithSpecifiedReplyPath { reply_path, .. } => reply_path,
+			_ => panic!(),
 		}
 	};
 
@@ -1445,11 +1437,10 @@ fn fails_authentication_when_handling_invoice_for_offer() {
 	{
 		let mut pending_offers_messages = david.node.pending_offers_messages.lock().unwrap();
 		let mut pending_invoice_request = pending_offers_messages.first_mut().unwrap();
-		#[cfg(not(c_bindings))] {
-			pending_invoice_request.reply_path = invalid_reply_path;
-		}
-		#[cfg(c_bindings)] {
-			pending_invoice_request.2 = invalid_reply_path;
+		match &mut pending_invoice_request.1 {
+			MessageSendInstructions::WithSpecifiedReplyPath { reply_path, .. } =>
+				*reply_path = invalid_reply_path,
+			_ => panic!(),
 		}
 	}
 
@@ -1531,13 +1522,10 @@ fn fails_authentication_when_handling_invoice_for_refund() {
 	let expected_invoice = alice.node.request_refund_payment(&refund).unwrap();
 
 	connect_peers(david, alice);
-	#[cfg(not(c_bindings))] {
-		alice.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().destination =
-			Destination::Node(david_id);
-	}
-	#[cfg(c_bindings)] {
-		alice.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 =
-			Destination::Node(david_id);
+	match &mut alice.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
+		MessageSendInstructions::WithSpecifiedReplyPath { destination, .. } =>
+			*destination = Destination::Node(david_id),
+		_ => panic!(),
 	}
 
 	let onion_message = alice.onion_messenger.next_onion_message_for_peer(david_id).unwrap();
@@ -1565,13 +1553,10 @@ fn fails_authentication_when_handling_invoice_for_refund() {
 
 	let expected_invoice = alice.node.request_refund_payment(&refund).unwrap();
 
-	#[cfg(not(c_bindings))] {
-		alice.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().destination =
-			Destination::BlindedPath(invalid_path);
-	}
-	#[cfg(c_bindings)] {
-		alice.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 =
-			Destination::BlindedPath(invalid_path);
+	match &mut alice.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
+		MessageSendInstructions::WithSpecifiedReplyPath { destination, .. } =>
+			*destination = Destination::BlindedPath(invalid_path),
+		_ => panic!(),
 	}
 
 	connect_peers(alice, charlie);
@@ -2155,10 +2140,11 @@ fn fails_paying_invoice_with_unknown_required_features() {
 		.build_and_sign(&secp_ctx).unwrap();
 
 	// Enqueue an onion message containing the new invoice.
-	let pending_message = new_pending_onion_message(
-		OffersMessage::Invoice(invoice), Destination::BlindedPath(reply_path), None
-	);
-	alice.node.pending_offers_messages.lock().unwrap().push(pending_message);
+	let instructions = MessageSendInstructions::WithoutReplyPath {
+		destination: Destination::BlindedPath(reply_path),
+	};
+	let message = OffersMessage::Invoice(invoice);
+	alice.node.pending_offers_messages.lock().unwrap().push((message, instructions));
 
 	let onion_message = alice.onion_messenger.next_onion_message_for_peer(charlie_id).unwrap();
 	charlie.onion_messenger.handle_onion_message(&alice_id, &onion_message);

lightning/src/onion_message/messenger.rs

@@ -385,6 +385,14 @@ impl Responder {
 /// Instructions for how and where to send a message.
 #[derive(Clone)]
 pub enum MessageSendInstructions {
+	/// Indicates that a message should be sent including the provided reply path for the recipient
+	/// to respond.
+	WithSpecifiedReplyPath {
+		/// The desination where we need to send our message.
+		destination: Destination,
+		/// The reply path which should be included in the message.
+		reply_path: BlindedMessagePath,
+	},
 	/// Indicates that a message should be sent including a reply path for the recipient to
 	/// respond.
 	WithReplyPath {
@@ -1165,24 +1173,25 @@ where
 	fn handle_onion_message_send<T: OnionMessageContents>(
 		&self, response_message: T, response: MessageSendInstructions, log_suffix: fmt::Arguments,
 	) -> Result<Option<SendSuccess>, SendError> {
-		let (destination, context) = match response {
-			MessageSendInstructions::WithReplyPath { destination, context } => (destination, Some(context)),
-			MessageSendInstructions::WithoutReplyPath { destination } => (destination, None),
-		};
-
-		let reply_path = if let Some(context) = context {
-			match self.create_blinded_path(context) {
-				Ok(reply_path) => Some(reply_path),
-				Err(err) => {
-					log_trace!(
-						self.logger,
-						"Failed to create reply path {}: {:?}",
-						log_suffix, err
-					);
-					return Err(err);
+		let (destination, reply_path) = match response {
+			MessageSendInstructions::WithSpecifiedReplyPath { destination, reply_path } =>
+				(destination, Some(reply_path)),
+			MessageSendInstructions::WithReplyPath { destination, context } => {
+				match self.create_blinded_path(context) {
+					Ok(reply_path) => (destination, Some(reply_path)),
+					Err(err) => {
+						log_trace!(
+							self.logger,
+							"Failed to create reply path {}: {:?}",
+							log_suffix, err
+						);
+						return Err(err);
+					}
 				}
-			}
-		} else { None };
+			},
+			MessageSendInstructions::WithoutReplyPath { destination } =>
+				(destination, None),
+		};
 
 		self.find_path_and_enqueue_onion_message(
 			response_message, destination, reply_path, log_suffix,
@@ -1719,13 +1728,9 @@ where
 	// node, and then enqueue the message for sending to the first peer in the full path.
 	fn next_onion_message_for_peer(&self, peer_node_id: PublicKey) -> Option<OnionMessage> {
 		// Enqueue any initiating `OffersMessage`s to send.
-		for message in self.offers_handler.release_pending_messages() {
-			#[cfg(not(c_bindings))]
-			let PendingOnionMessage { contents, destination, reply_path } = message;
-			#[cfg(c_bindings)]
-			let (contents, destination, reply_path) = message;
-			let _ = self.find_path_and_enqueue_onion_message(
-				contents, destination, reply_path, format_args!("when sending OffersMessage")
+		for (message, instructions) in self.offers_handler.release_pending_messages() {
+			let _ = self.handle_onion_message_send(
+				message, instructions, format_args!("when sending OffersMessage")
 			);
 		}
 

lightning/src/onion_message/offers.rs

@@ -23,8 +23,6 @@ use crate::onion_message::packet::OnionMessageContents;
 use crate::util::logger::Logger;
 use crate::util::ser::{Readable, ReadableArgs, Writeable, Writer};
 use crate::onion_message::messenger::{MessageSendInstructions, Responder};
-#[cfg(not(c_bindings))]
-use crate::onion_message::messenger::PendingOnionMessage;
 
 use crate::prelude::*;
 
@@ -53,15 +51,7 @@ pub trait OffersMessageHandler {
 	///
 	/// Typically, this is used for messages initiating a payment flow rather than in response to
 	/// another message. The latter should use the return value of [`Self::handle_message`].
-	#[cfg(not(c_bindings))]
-	fn release_pending_messages(&self) -> Vec<PendingOnionMessage<OffersMessage>> { vec![] }
-
-	/// Releases any [`OffersMessage`]s that need to be sent.
-	///
-	/// Typically, this is used for messages initiating a payment flow rather than in response to
-	/// another message. The latter should use the return value of [`Self::handle_message`].
-	#[cfg(c_bindings)]
-	fn release_pending_messages(&self) -> Vec<(OffersMessage, crate::onion_message::messenger::Destination, Option<crate::blinded_path::message::BlindedMessagePath>)> { vec![] }
+	fn release_pending_messages(&self) -> Vec<(OffersMessage, MessageSendInstructions)> { vec![] }
 }
 
 /// Possible BOLT 12 Offers messages sent and received via an [`OnionMessage`].