Move AsyncPaymentsMessageHandler to flow.rs

Author: shaavan

lightning-background-processor/src/lib.rs

@@ -1083,6 +1083,7 @@ mod tests {
 		IgnoringMessageHandler, MessageHandler, PeerManager, SocketDescriptor,
 	};
 	use lightning::ln::types::ChannelId;
+	use lightning::offers::flow;
 	use lightning::onion_message::messenger::{DefaultMessageRouter, OnionMessenger};
 	use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
 	use lightning::routing::router::{CandidateRouteHop, DefaultRouter, Path, RouteHop};
@@ -1178,6 +1179,19 @@ mod tests {
 		>,
 	>;
 
+	type OffersMessageFlow = flow::OffersMessageFlow<
+		Arc<KeysManager>,
+		Arc<ChannelManager>,
+		Arc<
+			DefaultMessageRouter<
+				Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
+				Arc<test_utils::TestLogger>,
+				Arc<KeysManager>,
+			>,
+		>,
+		Arc<test_utils::TestLogger>,
+	>;
+
 	type OM = OnionMessenger<
 		Arc<KeysManager>,
 		Arc<KeysManager>,
@@ -1191,7 +1205,7 @@ mod tests {
 			>,
 		>,
 		IgnoringMessageHandler,
-		Arc<ChannelManager>,
+		Arc<OffersMessageFlow>,
 		IgnoringMessageHandler,
 		IgnoringMessageHandler,
 	>;
@@ -1587,14 +1601,23 @@ mod tests {
 				params,
 				genesis_block.header.time,
 			));
+			let offers_message_flow = Arc::new(OffersMessageFlow::new(
+				manager.inbound_payment_key,
+				manager.get_our_node_id(),
+				keys_manager.clone(),
+				manager.clone(),
+				msg_router.clone(),
+				logger.clone(),
+			));
+
 			let messenger = Arc::new(OnionMessenger::new(
 				keys_manager.clone(),
 				keys_manager.clone(),
 				logger.clone(),
 				manager.clone(),
 				msg_router.clone(),
 				IgnoringMessageHandler {},
-				manager.clone(),
+				offers_message_flow,
 				IgnoringMessageHandler {},
 				IgnoringMessageHandler {},
 			));

lightning/src/ln/async_payments_tests.rs

@@ -265,7 +265,7 @@ fn ignore_unexpected_static_invoice() {
 	nodes[0]
 		.onion_messenger
 		.handle_onion_message(nodes[1].node.get_our_node_id(), &unexpected_static_invoice_om);
-	let async_pmts_msgs = AsyncPaymentsMessageHandler::release_pending_messages(nodes[0].node);
+	let async_pmts_msgs = nodes[0].offers_handler.release_pending_messages();
 	assert!(async_pmts_msgs.is_empty());
 	assert!(nodes[0].node.get_and_clear_pending_events().is_empty());
 
@@ -289,7 +289,7 @@ fn ignore_unexpected_static_invoice() {
 	nodes[0]
 		.onion_messenger
 		.handle_onion_message(nodes[1].node.get_our_node_id(), &static_invoice_om);
-	let async_pmts_msgs = AsyncPaymentsMessageHandler::release_pending_messages(nodes[0].node);
+	let async_pmts_msgs = nodes[0].offers_handler.release_pending_messages();
 	assert!(!async_pmts_msgs.is_empty());
 	assert!(async_pmts_msgs
 		.into_iter()
@@ -314,7 +314,7 @@ fn ignore_unexpected_static_invoice() {
 	nodes[0]
 		.onion_messenger
 		.handle_onion_message(nodes[1].node.get_our_node_id(), &dup_static_invoice_om);
-	let async_pmts_msgs = AsyncPaymentsMessageHandler::release_pending_messages(nodes[0].node);
+	let async_pmts_msgs = nodes[0].offers_handler.release_pending_messages();
 	assert!(async_pmts_msgs.is_empty());
 }
 
@@ -369,7 +369,7 @@ fn pays_static_invoice() {
 	nodes[0]
 		.onion_messenger
 		.handle_onion_message(nodes[1].node.get_our_node_id(), &static_invoice_om);
-	let mut async_pmts_msgs = AsyncPaymentsMessageHandler::release_pending_messages(nodes[0].node);
+	let mut async_pmts_msgs = nodes[0].offers_handler.release_pending_messages();
 	assert!(!async_pmts_msgs.is_empty());
 	assert!(async_pmts_msgs
 		.iter()

lightning/src/ln/channelmanager.rs

@@ -33,7 +33,7 @@ use bitcoin::secp256k1::Secp256k1;
 use bitcoin::{secp256k1, Sequence, Weight};
 
 use crate::events::FundingInfo;
-use crate::blinded_path::message::{AsyncPaymentsContext, MessageContext, OffersContext};
+use crate::blinded_path::message::{MessageContext, OffersContext};
 use crate::blinded_path::NodeIdLookUp;
 use crate::blinded_path::message::{BlindedMessagePath, MessageForwardNode};
 use crate::blinded_path::payment::{BlindedPaymentPath, Bolt12RefundContext, PaymentConstraints, PaymentContext, UnauthenticatedReceiveTlvs};
@@ -72,9 +72,8 @@ use crate::offers::offer::{Offer, OfferBuilder};
 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::dns_resolution::HumanReadableName;
-use crate::onion_message::messenger::{Destination, MessageRouter, Responder, ResponseInstruction, MessageSendInstructions};
+use crate::onion_message::messenger::{Destination, MessageRouter, MessageSendInstructions};
 use crate::onion_message::offers::OffersMessage;
 use crate::sign::{EntropySource, NodeSigner, Recipient, SignerProvider};
 use crate::sign::ecdsa::EcdsaChannelSigner;
@@ -87,6 +86,7 @@ use crate::util::ser::TransactionU16LenLimited;
 use crate::util::logger::{Level, Logger, WithContext};
 use crate::util::errors::APIError;
 #[cfg(async_payments)] use {
+	crate::blinded_path::message::AsyncPaymentsContext,
 	crate::blinded_path::payment::AsyncBolt12OfferContext,
 	crate::offers::offer::Amount,
 	crate::offers::static_invoice::{DEFAULT_RELATIVE_EXPIRY as STATIC_INVOICE_DEFAULT_RELATIVE_EXPIRY, StaticInvoice, StaticInvoiceBuilder},
@@ -2529,7 +2529,10 @@ where
 
 	our_network_pubkey: PublicKey,
 
-	pub(super) inbound_payment_key: inbound_payment::ExpandedKey,
+	/// The [`ExpandedKey`] for use in encrypting and decrypting inbound payment data.
+	///
+	/// [`ExpandedKey`]: crate::ln::inbound_payment::ExpandedKey
+	pub inbound_payment_key: inbound_payment::ExpandedKey,
 
 	/// LDK puts the [fake scids] that it generates into namespaces, to identify the type of an
 	/// incoming payment. To make it harder for a third-party to identify the type of a payment,
@@ -2622,7 +2625,6 @@ where
 	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.
 	pending_broadcast_messages: Mutex<Vec<MessageSendEvent>>,
@@ -3555,7 +3557,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()),
 
 			last_days_feerates: Mutex::new(VecDeque::new()),
@@ -4724,45 +4725,6 @@ where
 		res
 	}
 
-	#[cfg(async_payments)]
-	fn initiate_async_payment(
-		&self, invoice: &StaticInvoice, payment_id: PaymentId
-	) -> Result<(), Bolt12PaymentError> {
-
-		self.handle_static_invoice_received(invoice, payment_id)?;
-
-		let nonce = Nonce::from_entropy_source(&*self.entropy_source);
-		let hmac = payment_id.hmac_for_async_payment(nonce, &self.inbound_payment_key);
-		let reply_paths = match self.create_blinded_paths(
-			MessageContext::AsyncPayments(
-				AsyncPaymentsContext::OutboundPayment { payment_id, nonce, hmac }
-			)
-		) {
-			Ok(paths) => paths,
-			Err(()) => {
-				self.abandon_payment_with_reason(payment_id, PaymentFailureReason::BlindedPathCreationFailed);
-				return Err(Bolt12PaymentError::BlindedPathCreationFailed);
-			}
-		};
-
-		let mut pending_async_payments_messages = self.pending_async_payments_messages.lock().unwrap();
-		const HTLC_AVAILABLE_LIMIT: usize = 10;
-		reply_paths
-			.iter()
-			.flat_map(|reply_path| invoice.message_paths().iter().map(move |invoice_path| (invoice_path, reply_path)))
-			.take(HTLC_AVAILABLE_LIMIT)
-			.for_each(|(invoice_path, reply_path)| {
-				let instructions = MessageSendInstructions::WithSpecifiedReplyPath {
-					destination: Destination::BlindedPath(invoice_path.clone()),
-					reply_path: reply_path.clone(),
-				};
-				let message = AsyncPaymentsMessage::HeldHtlcAvailable(HeldHtlcAvailable {});
-				pending_async_payments_messages.push((message, instructions));
-			});
-
-		Ok(())
-	}
-
 	#[cfg(async_payments)]
 	fn send_payment_for_static_invoice(
 		&self, payment_id: PaymentId
@@ -9900,6 +9862,18 @@ where
 		self.pending_outbound_payments.received_offer(payment_id, retryable_invoice_request)
 	}
 
+	#[cfg(async_payments)]
+	fn handle_static_invoice_received(
+		&self, invoice: &StaticInvoice, payment_id: PaymentId
+	) -> Result<(), Bolt12PaymentError> {
+		self.handle_static_invoice_received(invoice, payment_id)
+	}
+
+	#[cfg(async_payments)]
+	fn send_payment_for_static_invoice(&self, payment_id: PaymentId) -> Result<(), Bolt12PaymentError> {
+		self.send_payment_for_static_invoice(payment_id)
+	}
+
 	// ----Temporary Functions----
 	// Set of functions temporarily moved to OffersMessageCommons for easier
 	// transition of code from ChannelManager to OffersMessageFlow
@@ -9919,13 +9893,6 @@ where
 	) -> Result<(), Bolt12SemanticError> {
 		self.pay_for_offer_intern(offer, quantity, amount_msats, payer_note, payment_id, human_readable_name, create_pending_payment)
 	}
-
-	#[cfg(async_payments)]
-	fn initiate_async_payment(
-		&self, invoice: &StaticInvoice, payment_id: PaymentId
-	) -> Result<(), Bolt12PaymentError> {
-		self.initiate_async_payment(invoice, payment_id)
-	}
 }
 
 /// Defines the maximum number of [`OffersMessage`] including different reply paths to be sent
@@ -11904,48 +11871,6 @@ where
 	}
 }
 
-impl<M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, MR: Deref, L: Deref>
-AsyncPaymentsMessageHandler for ChannelManager<M, T, ES, NS, SP, F, R, MR, L>
-where
-	M::Target: chain::Watch<<SP::Target as SignerProvider>::EcdsaSigner>,
-	T::Target: BroadcasterInterface,
-	ES::Target: EntropySource,
-	NS::Target: NodeSigner,
-	SP::Target: SignerProvider,
-	F::Target: FeeEstimator,
-	R::Target: Router,
-	MR::Target: MessageRouter,
-	L::Target: Logger,
-{
-	fn handle_held_htlc_available(
-		&self, _message: HeldHtlcAvailable, _context: AsyncPaymentsContext,
-		_responder: Option<Responder>
-	) -> Option<(ReleaseHeldHtlc, ResponseInstruction)> {
-		None
-	}
-
-	fn handle_release_held_htlc(&self, _message: ReleaseHeldHtlc, _context: AsyncPaymentsContext) {
-		#[cfg(async_payments)] {
-			let (payment_id, nonce, hmac) = match _context {
-				AsyncPaymentsContext::OutboundPayment { payment_id, hmac, nonce } => {
-					(payment_id, nonce, hmac)
-				},
-				_ => return
-			};
-			if payment_id.verify_for_async_payment(hmac, nonce, &self.inbound_payment_key).is_err() { return }
-			if let Err(e) = self.send_payment_for_static_invoice(payment_id) {
-				log_trace!(
-					self.logger, "Failed to release held HTLC with payment id {}: {:?}", payment_id, e
-				);
-			}
-		}
-	}
-
-	fn release_pending_messages(&self) -> Vec<(AsyncPaymentsMessage, MessageSendInstructions)> {
-		core::mem::take(&mut self.pending_async_payments_messages.lock().unwrap())
-	}
-}
-
 impl<M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, MR: Deref, L: Deref>
 NodeIdLookUp for ChannelManager<M, T, ES, NS, SP, F, R, MR, L>
 where
@@ -13862,7 +13787,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()),
 

lightning/src/ln/functional_test_utils.rs

@@ -427,7 +427,7 @@ type TestOnionMessenger<'chan_man, 'node_cfg, 'chan_mon_cfg> = OnionMessenger<
 	&'chan_man TestChannelManager<'node_cfg, 'chan_mon_cfg>,
 	&'node_cfg test_utils::TestMessageRouter<'chan_mon_cfg>,
 	Arc<TestOffersMessageFlow<'chan_man, 'node_cfg, 'chan_mon_cfg>>,
-	&'chan_man TestChannelManager<'node_cfg, 'chan_mon_cfg>,
+	Arc<TestOffersMessageFlow<'chan_man, 'node_cfg, 'chan_mon_cfg>>,
 	IgnoringMessageHandler,
 	IgnoringMessageHandler,
 >;
@@ -440,7 +440,7 @@ type TestOnionMessenger<'chan_man, 'node_cfg, 'chan_mon_cfg> = OnionMessenger<
 	&'chan_man TestChannelManager<'node_cfg, 'chan_mon_cfg>,
 	&'node_cfg test_utils::TestMessageRouter<'chan_mon_cfg>,
 	Arc<TestOffersMessageFlow<'chan_man, 'node_cfg, 'chan_mon_cfg>>,
-	&'chan_man TestChannelManager<'node_cfg, 'chan_mon_cfg>,
+	Arc<TestOffersMessageFlow<'chan_man, 'node_cfg, 'chan_mon_cfg>>,
 	Arc<TestOffersMessageFlow<'chan_man, 'node_cfg, 'chan_mon_cfg>>,
 	IgnoringMessageHandler,
 >;
@@ -3352,13 +3352,13 @@ pub fn create_network<'a, 'b: 'a, 'c: 'b>(node_count: usize, cfgs: &'b Vec<NodeC
 		#[cfg(feature = "dnssec")]
 		let onion_messenger = OnionMessenger::new(
 			dedicated_entropy, cfgs[i].keys_manager, cfgs[i].logger, &chan_mgrs[i],
-			&cfgs[i].message_router, offers_handler.clone(), &chan_mgrs[i], offers_handler.clone(),
+			&cfgs[i].message_router, offers_handler.clone(), offers_handler.clone(), offers_handler.clone(),
 			IgnoringMessageHandler {},
 		);
 		#[cfg(not(feature = "dnssec"))]
 		let onion_messenger = OnionMessenger::new(
 			dedicated_entropy, cfgs[i].keys_manager, cfgs[i].logger, &chan_mgrs[i],
-			&cfgs[i].message_router, offers_handler.clone(), &chan_mgrs[i], IgnoringMessageHandler {},
+			&cfgs[i].message_router, offers_handler.clone(), offers_handler.clone(), IgnoringMessageHandler {},
 			IgnoringMessageHandler {},
 		);
 		let gossip_sync = P2PGossipSync::new(cfgs[i].network_graph.as_ref(), None, cfgs[i].logger);