Skip to content

Commit

Permalink
Move pending_offers_message to flows.rs
Browse files Browse the repository at this point in the history
  • Loading branch information
@shaavan
shaavan committed Nov 30, 2024
1 parent cb5d6d1 commit 003a409
Show file tree
Hide file tree
Showing 3 changed files with 66 additions and 82 deletions.
70 changes: 1 addition & 69 deletions lightning/src/ln/channelmanager.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -64,15 +64,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};
Expand Down Expand Up @@ -2113,8 +2110,6 @@ where
//
// Lock order tree:
//
// `pending_offers_messages`
//
// `pending_async_payments_messages`
//
// `total_consistency_lock`
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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()),

Expand Down Expand Up @@ -9475,9 +9465,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)>>;
Expand Down Expand Up @@ -9575,13 +9562,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;

Expand Down Expand Up @@ -9621,10 +9601,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.")
Expand Down Expand Up @@ -9759,42 +9735,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)
}
Expand Down Expand Up @@ -9832,13 +9772,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>,
Expand Down Expand Up @@ -13174,7 +13107,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()),
Expand Down
14 changes: 7 additions & 7 deletions lightning/src/ln/offers_tests.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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!(),
Expand All @@ -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!(),
Expand Down Expand Up @@ -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 {
Expand All @@ -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, .. } =>
Expand Down Expand Up @@ -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!(),
Expand Down Expand Up @@ -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!(),
Expand Down Expand Up @@ -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);
Expand Down
64 changes: 58 additions & 6 deletions lightning/src/offers/flow.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -416,6 +416,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.
Expand Down Expand Up @@ -443,9 +448,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,
Expand Down Expand Up @@ -473,6 +482,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,
Expand Down Expand Up @@ -531,6 +547,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>
Expand Down Expand Up @@ -587,7 +639,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)
}
}

Expand Down Expand Up @@ -862,7 +914,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!(
Expand All @@ -886,7 +938,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())
}
}

Expand Down Expand Up @@ -1218,7 +1270,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 {
Expand Down

0 comments on commit 003a409

Please sign in to comment.