Move pending_offers_message to flows.rs

shaavan · shaavan · commit 79cfd2475d4b · 2024-12-06T21:30:07.000+05:30
diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
@@ -65,15 +65,12 @@ use crate::ln::outbound_payment;
 use crate::ln::outbound_payment::{OutboundPayments, PendingOutboundPayment, RetryableInvoiceRequest, SendAlongPathArgs, StaleExpiration};
 use crate::offers::invoice::Bolt12Invoice;
 use crate::offers::invoice::UnsignedBolt12Invoice;
-use crate::offers::invoice_request::InvoiceRequest;
 use crate::offers::nonce::Nonce;
-use crate::offers::parse::Bolt12SemanticError;
 use crate::offers::signer;
 #[cfg(async_payments)]
 use crate::offers::static_invoice::StaticInvoice;
 use crate::onion_message::async_payments::{AsyncPaymentsMessage, HeldHtlcAvailable, ReleaseHeldHtlc, AsyncPaymentsMessageHandler};
-use crate::onion_message::messenger::{DefaultMessageRouter, Destination, MessageRouter, MessageSendInstructions, Responder, ResponseInstruction};
-use crate::onion_message::offers::OffersMessage;
+use crate::onion_message::messenger::{DefaultMessageRouter, MessageRouter, MessageSendInstructions, Responder, ResponseInstruction};
 use crate::sign::{EntropySource, NodeSigner, Recipient, SignerProvider};
 use crate::sign::ecdsa::EcdsaChannelSigner;
 use crate::util::config::{UserConfig, ChannelConfig, ChannelConfigUpdate};
@@ -2113,8 +2110,6 @@ where
 //
 // Lock order tree:
 //
-// `pending_offers_messages`
-//
 // `pending_async_payments_messages`
 //
 // `total_consistency_lock`
@@ -2363,10 +2358,6 @@ where
 	event_persist_notifier: Notifier,
 	needs_persist_flag: AtomicBool,
 
-	#[cfg(not(any(test, feature = "_test_utils")))]
-	pending_offers_messages: Mutex<Vec<(OffersMessage, MessageSendInstructions)>>,
-	#[cfg(any(test, feature = "_test_utils"))]
-	pub(crate) pending_offers_messages: Mutex<Vec<(OffersMessage, MessageSendInstructions)>>,
 	pending_async_payments_messages: Mutex<Vec<(AsyncPaymentsMessage, MessageSendInstructions)>>,
 
 	/// Tracks the message events that are to be broadcasted when we are connected to some peer.
@@ -3229,7 +3220,6 @@ where
 			needs_persist_flag: AtomicBool::new(false),
 			funding_batch_states: Mutex::new(BTreeMap::new()),
 
-			pending_offers_messages: Mutex::new(Vec::new()),
 			pending_async_payments_messages: Mutex::new(Vec::new()),
 			pending_broadcast_messages: Mutex::new(Vec::new()),
 
@@ -9489,10 +9479,6 @@ where
 	MR::Target: MessageRouter,
 	L::Target: Logger,
 {
-	fn get_pending_offers_messages(&self) -> MutexGuard<'_, Vec<(OffersMessage, MessageSendInstructions)>> {
-		self.pending_offers_messages.lock().expect("Mutex is locked by other thread.")
-	}
-
 	#[cfg(feature = "dnssec")]
 	fn get_pending_dns_onion_messages(&self) -> MutexGuard<'_, Vec<(DNSResolverMessage, MessageSendInstructions)>> {
 		self.pending_dns_onion_messages.lock().expect("Mutex is locked by other thread.")
@@ -9611,42 +9597,6 @@ where
 			.and_then(|paths| (!paths.is_empty()).then(|| paths).ok_or(()))
 	}
 
-	fn enqueue_invoice_request(
-		&self,
-		invoice_request: InvoiceRequest,
-		reply_paths: Vec<BlindedMessagePath>,
-	) -> Result<(), Bolt12SemanticError> {
-		let mut pending_offers_messages = self.pending_offers_messages.lock().unwrap();
-		if !invoice_request.paths().is_empty() {
-			reply_paths
-				.iter()
-				.flat_map(|reply_path| invoice_request.paths().iter().map(move |path| (path, reply_path)))
-				.take(OFFERS_MESSAGE_REQUEST_LIMIT)
-				.for_each(|(path, reply_path)| {
-					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(node_id) = invoice_request.issuer_signing_pubkey() {
-			for reply_path in reply_paths {
-				let instructions = MessageSendInstructions::WithSpecifiedReplyPath {
-					destination: Destination::Node(node_id),
-					reply_path,
-				};
-				let message = OffersMessage::InvoiceRequest(invoice_request.clone());
-				pending_offers_messages.push((message, instructions));
-			}
-		} else {
-			debug_assert!(false);
-			return Err(Bolt12SemanticError::MissingIssuerSigningPubkey);
-		}
-
-		Ok(())
-	}
-
 	fn get_current_blocktime(&self) -> Duration {
 		Duration::from_secs(self.highest_seen_timestamp.load(Ordering::Acquire) as u64)
 	}
@@ -9687,13 +9637,6 @@ where
 	}
 }
 
-/// Defines the maximum number of [`OffersMessage`] including different reply paths to be sent
-/// along different paths.
-/// Sending multiple requests increases the chances of successful delivery in case some
-/// paths are unavailable. However, only one invoice for a given [`PaymentId`] will be paid,
-/// even if multiple invoices are received.
-pub const OFFERS_MESSAGE_REQUEST_LIMIT: usize = 10;
-
 impl<M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, MR: Deref, L: Deref> ChannelManager<M, T, ES, NS, SP, F, R, MR, L>
 where
 	M::Target: chain::Watch<<SP::Target as SignerProvider>::EcdsaSigner>,
@@ -13029,7 +12972,6 @@ where
 
 			funding_batch_states: Mutex::new(BTreeMap::new()),
 
-			pending_offers_messages: Mutex::new(Vec::new()),
 			pending_async_payments_messages: Mutex::new(Vec::new()),
 
 			pending_broadcast_messages: Mutex::new(Vec::new()),
diff --git a/lightning/src/ln/offers_tests.rs b/lightning/src/ln/offers_tests.rs
@@ -1326,7 +1326,7 @@ fn fails_authentication_when_handling_invoice_request() {
 	expect_recent_payment!(david, RecentPaymentDetails::AwaitingInvoice, payment_id);
 
 	connect_peers(david, alice);
-	match &mut david.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
+	match &mut david.offers_handler.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
 		MessageSendInstructions::WithSpecifiedReplyPath { destination, .. } =>
 			*destination = Destination::Node(alice_id),
 		_ => panic!(),
@@ -1351,7 +1351,7 @@ fn fails_authentication_when_handling_invoice_request() {
 		.unwrap();
 	expect_recent_payment!(david, RecentPaymentDetails::AwaitingInvoice, payment_id);
 
-	match &mut david.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
+	match &mut david.offers_handler.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
 		MessageSendInstructions::WithSpecifiedReplyPath { destination, .. } =>
 			*destination = Destination::BlindedPath(invalid_path),
 		_ => panic!(),
@@ -1431,7 +1431,7 @@ fn fails_authentication_when_handling_invoice_for_offer() {
 
 	// Don't send the invoice request, but grab its reply path to use with a different request.
 	let invalid_reply_path = {
-		let mut pending_offers_messages = david.node.pending_offers_messages.lock().unwrap();
+		let mut pending_offers_messages = david.offers_handler.pending_offers_messages.lock().unwrap();
 		let pending_invoice_request = pending_offers_messages.pop().unwrap();
 		pending_offers_messages.clear();
 		match pending_invoice_request.1 {
@@ -1448,7 +1448,7 @@ fn fails_authentication_when_handling_invoice_for_offer() {
 	// Swap out the reply path to force authentication to fail when handling the invoice since it
 	// will be sent over the wrong blinded path.
 	{
-		let mut pending_offers_messages = david.node.pending_offers_messages.lock().unwrap();
+		let mut pending_offers_messages = david.offers_handler.pending_offers_messages.lock().unwrap();
 		let mut pending_invoice_request = pending_offers_messages.first_mut().unwrap();
 		match &mut pending_invoice_request.1 {
 			MessageSendInstructions::WithSpecifiedReplyPath { reply_path, .. } =>
@@ -1535,7 +1535,7 @@ fn fails_authentication_when_handling_invoice_for_refund() {
 	let expected_invoice = alice.offers_handler.request_refund_payment(&refund).unwrap();
 
 	connect_peers(david, alice);
-	match &mut alice.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
+	match &mut alice.offers_handler.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
 		MessageSendInstructions::WithSpecifiedReplyPath { destination, .. } =>
 			*destination = Destination::Node(david_id),
 		_ => panic!(),
@@ -1566,7 +1566,7 @@ fn fails_authentication_when_handling_invoice_for_refund() {
 
 	let expected_invoice = alice.offers_handler.request_refund_payment(&refund).unwrap();
 
-	match &mut alice.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
+	match &mut alice.offers_handler.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
 		MessageSendInstructions::WithSpecifiedReplyPath { destination, .. } =>
 			*destination = Destination::BlindedPath(invalid_path),
 		_ => panic!(),
@@ -2157,7 +2157,7 @@ fn fails_paying_invoice_with_unknown_required_features() {
 		destination: Destination::BlindedPath(reply_path),
 	};
 	let message = OffersMessage::Invoice(invoice);
-	alice.node.pending_offers_messages.lock().unwrap().push((message, instructions));
+	alice.offers_handler.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);
diff --git a/lightning/src/offers/flow.rs b/lightning/src/offers/flow.rs
@@ -28,9 +28,7 @@ use crate::blinded_path::payment::{
 	BlindedPaymentPath, Bolt12OfferContext, Bolt12RefundContext, PaymentContext,
 };
 use crate::events::PaymentFailureReason;
-use crate::ln::channelmanager::{
-	Bolt12PaymentError, PaymentId, Verification, OFFERS_MESSAGE_REQUEST_LIMIT,
-};
+use crate::ln::channelmanager::{Bolt12PaymentError, PaymentId, Verification};
 use crate::ln::inbound_payment;
 use crate::ln::outbound_payment::{Retry, RetryableInvoiceRequest, StaleExpiration};
 use crate::offers::invoice::{
@@ -73,11 +71,6 @@ use {
 ///
 /// [`ChannelManager`]: crate::ln::channelmanager::ChannelManager
 pub trait OffersMessageCommons {
-	/// Get pending offers messages
-	fn get_pending_offers_messages(
-		&self,
-	) -> MutexGuard<'_, Vec<(OffersMessage, MessageSendInstructions)>>;
-
 	#[cfg(feature = "dnssec")]
 	/// Get pending DNS onion messages
 	fn get_pending_dns_onion_messages(
@@ -176,11 +169,6 @@ pub trait OffersMessageCommons {
 	/// [`MessageRouter::create_blinded_paths`]: crate::onion_message::messenger::MessageRouter::create_blinded_paths
 	fn create_blinded_paths(&self, context: MessageContext) -> Result<Vec<BlindedMessagePath>, ()>;
 
-	/// Enqueue invoice request
-	fn enqueue_invoice_request(
-		&self, invoice_request: InvoiceRequest, reply_paths: Vec<BlindedMessagePath>,
-	) -> Result<(), Bolt12SemanticError>;
-
 	/// Get the current time determined by highest seen timestamp
 	fn get_current_blocktime(&self) -> Duration;
 
@@ -575,6 +563,11 @@ where
 
 	message_router: MR,
 
+	#[cfg(not(any(test, feature = "_test_utils")))]
+	pending_offers_messages: Mutex<Vec<(OffersMessage, MessageSendInstructions)>>,
+	#[cfg(any(test, feature = "_test_utils"))]
+	pub(crate) pending_offers_messages: Mutex<Vec<(OffersMessage, MessageSendInstructions)>>,
+
 	#[cfg(feature = "_test_utils")]
 	/// In testing, it is useful be able to forge a name -> offer mapping so that we can pay an
 	/// offer generated in the test.
@@ -607,9 +600,13 @@ where
 			inbound_payment_key: expanded_inbound_key,
 			our_network_pubkey,
 			secp_ctx,
+			entropy_source,
+
 			commons,
+
 			message_router,
-			entropy_source,
+
+			pending_offers_messages: Mutex::new(Vec::new()),
 			#[cfg(feature = "_test_utils")]
 			testing_dnssec_proof_offer_resolution_override: Mutex::new(new_hash_map()),
 			logger,
@@ -642,6 +639,13 @@ where
 /// [`Refund`]: crate::offers::refund
 pub const MAX_SHORT_LIVED_RELATIVE_EXPIRY: Duration = Duration::from_secs(60 * 60 * 24);
 
+/// Defines the maximum number of [`OffersMessage`] including different reply paths to be sent
+/// along different paths.
+/// Sending multiple requests increases the chances of successful delivery in case some
+/// paths are unavailable. However, only one invoice for a given [`PaymentId`] will be paid,
+/// even if multiple invoices are received.
+pub const OFFERS_MESSAGE_REQUEST_LIMIT: usize = 10;
+
 impl<ES: Deref, OMC: Deref, MR: Deref, L: Deref> OffersMessageFlow<ES, OMC, MR, L>
 where
 	ES::Target: EntropySource,
@@ -700,6 +704,42 @@ where
 			)
 			.and_then(|paths| (!paths.is_empty()).then(|| paths).ok_or(()))
 	}
+
+	fn enqueue_invoice_request(
+		&self, invoice_request: InvoiceRequest, reply_paths: Vec<BlindedMessagePath>,
+	) -> Result<(), Bolt12SemanticError> {
+		let mut pending_offers_messages = self.pending_offers_messages.lock().unwrap();
+		if !invoice_request.paths().is_empty() {
+			reply_paths
+				.iter()
+				.flat_map(|reply_path| {
+					invoice_request.paths().iter().map(move |path| (path, reply_path))
+				})
+				.take(OFFERS_MESSAGE_REQUEST_LIMIT)
+				.for_each(|(path, reply_path)| {
+					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(node_id) = invoice_request.issuer_signing_pubkey() {
+			for reply_path in reply_paths {
+				let instructions = MessageSendInstructions::WithSpecifiedReplyPath {
+					destination: Destination::Node(node_id),
+					reply_path,
+				};
+				let message = OffersMessage::InvoiceRequest(invoice_request.clone());
+				pending_offers_messages.push((message, instructions));
+			}
+		} else {
+			debug_assert!(false);
+			return Err(Bolt12SemanticError::MissingIssuerSigningPubkey);
+		}
+
+		Ok(())
+	}
 }
 
 impl<ES: Deref, OMC: Deref, MR: Deref, L: Deref> OffersMessageFlow<ES, OMC, MR, L>
@@ -756,7 +796,7 @@ where
 
 		create_pending_payment(&invoice_request, nonce)?;
 
-		self.commons.enqueue_invoice_request(invoice_request, reply_paths)
+		self.enqueue_invoice_request(invoice_request, reply_paths)
 	}
 }
 
@@ -1024,7 +1064,7 @@ where
 			});
 			match self.commons.create_blinded_paths(context) {
 				Ok(reply_paths) => {
-					match self.commons.enqueue_invoice_request(invoice_request, reply_paths) {
+					match self.enqueue_invoice_request(invoice_request, reply_paths) {
 						Ok(_) => {},
 						Err(_) => {
 							log_warn!(
@@ -1048,7 +1088,7 @@ where
 	}
 
 	fn release_pending_messages(&self) -> Vec<(OffersMessage, MessageSendInstructions)> {
-		core::mem::take(&mut self.commons.get_pending_offers_messages())
+		core::mem::take(&mut self.pending_offers_messages.lock().unwrap())
 	}
 }
 
@@ -1382,7 +1422,7 @@ where
 					.create_blinded_paths(context)
 					.map_err(|_| Bolt12SemanticError::MissingPaths)?;
 
-				let mut pending_offers_messages = self.commons.get_pending_offers_messages();
+				let mut pending_offers_messages = self.pending_offers_messages.lock().unwrap();
 				if refund.paths().is_empty() {
 					for reply_path in reply_paths {
 						let instructions = MessageSendInstructions::WithSpecifiedReplyPath {
