Move pending_offers_message to flows.rs

shaavan · shaavan · commit c7ee1055a0d9 · 2024-11-29T18:07:10.000+05:30
diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
@@ -66,15 +66,12 @@ use crate::ln::outbound_payment::{OutboundPayments, PendingOutboundPayment, Retr
 use crate::ln::wire::Encode;
 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};
@@ -2106,8 +2103,6 @@ where
 //
 // Lock order tree:
 //
-// `pending_offers_messages`
-//
 // `pending_async_payments_messages`
 //
 // `total_consistency_lock`
@@ -2356,10 +2351,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.
@@ -3218,7 +3209,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()),
 
@@ -9164,9 +9154,6 @@ impl Default for Bolt11InvoiceParameters {
 ///
 /// [`OffersMessageFlow`]: crate::offers::flow::OffersMessageFlow
 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(&self) -> MutexGuard<'_, Vec<(DNSResolverMessage, MessageSendInstructions)>>;
@@ -9264,13 +9251,6 @@ pub trait OffersMessageCommons {
 	/// Errors if the `MessageRouter` errors.
 	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;
 
@@ -9310,10 +9290,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.")
@@ -9448,42 +9424,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)
 	}
@@ -9521,13 +9461,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>,
@@ -12844,7 +12777,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
@@ -1403,7 +1403,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!(),
@@ -1428,7 +1428,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!(),
@@ -1508,7 +1508,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 {
@@ -1525,7 +1525,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, .. } =>
@@ -1612,7 +1612,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!(),
@@ -1643,7 +1643,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!(),
@@ -2234,7 +2234,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
@@ -21,7 +21,7 @@ use crate::blinded_path::message::{BlindedMessagePath, MessageContext, OffersCon
 use crate::blinded_path::payment::{Bolt12OfferContext, Bolt12RefundContext, PaymentContext};
 use crate::events::{Event, PaymentFailureReason};
 use crate::ln::channelmanager::{
-	Bolt12PaymentError, OffersMessageCommons, PaymentId, Verification, OFFERS_MESSAGE_REQUEST_LIMIT,
+	Bolt12PaymentError, OffersMessageCommons, PaymentId, Verification,
 };
 use crate::ln::outbound_payment::{Retry, RetryableInvoiceRequest, StaleExpiration};
 use crate::onion_message::dns_resolution::HumanReadableName;
@@ -414,6 +414,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.
@@ -441,9 +446,13 @@ where
 
 		Self {
 			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,
@@ -471,6 +480,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,
@@ -529,6 +545,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>
@@ -585,7 +637,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)
 	}
 }
 
@@ -860,7 +912,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!(
@@ -884,7 +936,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())
 	}
 }
 
@@ -1216,7 +1268,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 {
