diff --git a/Cargo.toml b/Cargo.toml index 7fc945833e..e1709cb3ae 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -185,16 +185,16 @@ harness = false #vss-client-ng = { path = "../vss-client" } #vss-client-ng = { git = "https://github.com/lightningdevkit/vss-client", branch = "main" } # -#[patch."https://github.com/lightningdevkit/rust-lightning"] -#lightning = { path = "../rust-lightning/lightning" } -#lightning-types = { path = "../rust-lightning/lightning-types" } -#lightning-invoice = { path = "../rust-lightning/lightning-invoice" } +[patch."https://github.com/lightningdevkit/rust-lightning"] +lightning = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "47dcb88874b11ddcf836bb294f861eb7bbf0d09b" } +lightning-types = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "47dcb88874b11ddcf836bb294f861eb7bbf0d09b" } +lightning-invoice = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "47dcb88874b11ddcf836bb294f861eb7bbf0d09b" } #lightning-net-tokio = { path = "../rust-lightning/lightning-net-tokio" } #lightning-persister = { path = "../rust-lightning/lightning-persister" } #lightning-background-processor = { path = "../rust-lightning/lightning-background-processor" } #lightning-rapid-gossip-sync = { path = "../rust-lightning/lightning-rapid-gossip-sync" } #lightning-block-sync = { path = "../rust-lightning/lightning-block-sync" } #lightning-transaction-sync = { path = "../rust-lightning/lightning-transaction-sync" } -#lightning-liquidity = { path = "../rust-lightning/lightning-liquidity" } -#lightning-macros = { path = "../rust-lightning/lightning-macros" } +lightning-liquidity = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "47dcb88874b11ddcf836bb294f861eb7bbf0d09b" } +lightning-macros = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "47dcb88874b11ddcf836bb294f861eb7bbf0d09b" } #lightning-dns-resolver = { path = "../rust-lightning/lightning-dns-resolver" } diff --git a/src/builder.rs b/src/builder.rs index a70b04b2ab..5861025349 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -43,6 +43,7 @@ use lightning::util::persist::{ use lightning::util::ser::ReadableArgs; use lightning::util::sweep::OutputSweeper; use lightning_dns_resolver::OMDomainResolver; +use lightning_liquidity::lsps2::router::LSPS2Router; use vss_client::headers::VssHeaderProvider; use crate::chain::ChainSource; @@ -74,6 +75,7 @@ use crate::lnurl_auth::LnurlAuth; use crate::logger::{log_error, LdkLogger, LogLevel, LogWriter, Logger}; use crate::message_handler::NodeCustomMessageHandler; use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox; +use crate::payment::LSPS2PaymentMetadataDecoder; use crate::peer_store::PeerStore; use crate::probing::{ HighDegreeStrategy, Prober, ProbingConfig, ProbingStrategy, ProbingStrategyKind, @@ -1900,15 +1902,20 @@ fn build_with_store_internal( } let scoring_fee_params = ProbabilisticScoringFeeParameters::default(); - let router = Arc::new(DefaultRouter::new( - Arc::clone(&network_graph), - Arc::clone(&logger), + let router = Arc::new(LSPS2Router::new_with_payment_metadata_decoder( + DefaultRouter::new( + Arc::clone(&network_graph), + Arc::clone(&logger), + Arc::clone(&keys_manager), + Arc::clone(&scorer), + scoring_fee_params, + ), Arc::clone(&keys_manager), - Arc::clone(&scorer), - scoring_fee_params, + LSPS2PaymentMetadataDecoder, )); let mut user_config = default_user_config(&config); + user_config.manually_handle_bolt12_invoice_requests = true; if liquidity_source_config.and_then(|lsc| lsc.lsps2_service.as_ref()).is_some() { // If we act as an LSPS2 service, we need to be able to intercept HTLCs and forward the @@ -2270,12 +2277,16 @@ fn build_with_store_internal( if let Some(penalty) = probing_cfg.diversity_penalty_msat { probing_fee_params.probing_diversity_penalty_msat = penalty; } - let probing_router = Arc::new(DefaultRouter::new( - Arc::clone(&network_graph), - Arc::clone(&logger), + let probing_router = Arc::new(LSPS2Router::new_with_payment_metadata_decoder( + DefaultRouter::new( + Arc::clone(&network_graph), + Arc::clone(&logger), + Arc::clone(&keys_manager), + Arc::clone(&scorer), + probing_fee_params, + ), Arc::clone(&keys_manager), - Arc::clone(&scorer), - probing_fee_params, + LSPS2PaymentMetadataDecoder, )); Arc::new(HighDegreeStrategy::new( Arc::clone(&network_graph), diff --git a/src/config.rs b/src/config.rs index 958eb14fcd..0ddf9d1c1b 100644 --- a/src/config.rs +++ b/src/config.rs @@ -28,6 +28,8 @@ const DEFAULT_BDK_WALLET_SYNC_INTERVAL_SECS: u64 = 80; const DEFAULT_LDK_WALLET_SYNC_INTERVAL_SECS: u64 = 30; const DEFAULT_FEE_RATE_CACHE_UPDATE_INTERVAL_SECS: u64 = 60 * 10; const DEFAULT_PROBING_LIQUIDITY_LIMIT_MULTIPLIER: u64 = 3; +/// The default maximum number of pending BOLT12 LSPS2 offers. +pub const DEFAULT_LSPS2_PENDING_OFFER_CACHE_SIZE: usize = 100; pub(crate) const DEFAULT_PROBING_INTERVAL_SECS: u64 = 10; pub(crate) const MIN_PROBING_INTERVAL: Duration = Duration::from_millis(100); pub(crate) const DEFAULT_PROBED_NODE_COOLDOWN_SECS: u64 = 60 * 60; // 1 hour @@ -146,6 +148,7 @@ pub(crate) const LNURL_AUTH_TIMEOUT_SECS: u64 = 15; /// | `probing_liquidity_limit_multiplier` | 3 | /// | `anchor_channels_config` | AnchorChannelsConfig::default() | /// | `route_parameters` | None | +/// | `lsps2_pending_offer_cache_size` | 100 | /// | `tor_config` | None | /// | `hrn_config` | HumanReadableNamesConfig::default() | /// @@ -201,6 +204,11 @@ pub struct Config { /// **Note:** If unset, default parameters will be used, and you will be able to override the /// parameters on a per-payment basis in the corresponding method calls. pub route_parameters: Option, + /// The maximum number of BOLT12 offers using LSPS2 JIT channels that are retained for future + /// invoice requests. Once full, the least recently used offer is evicted. + /// + /// Values below one are treated as one. + pub lsps2_pending_offer_cache_size: usize, /// Configuration options for enabling peer connections via the Tor network. /// /// Setting [`TorConfig`] enables connecting to peers with OnionV3 addresses. No other connections @@ -226,6 +234,7 @@ impl Default for Config { anchor_channels_config: AnchorChannelsConfig::default(), tor_config: None, route_parameters: None, + lsps2_pending_offer_cache_size: DEFAULT_LSPS2_PENDING_OFFER_CACHE_SIZE, node_alias: None, hrn_config: HumanReadableNamesConfig::default(), } diff --git a/src/event.rs b/src/event.rs index 528a7085ed..b21baa70de 100644 --- a/src/event.rs +++ b/src/event.rs @@ -7,7 +7,7 @@ use core::future::Future; use core::task::{Poll, Waker}; -use std::collections::VecDeque; +use std::collections::{BTreeMap, VecDeque}; use std::ops::Deref; use std::sync::{Arc, Mutex}; @@ -45,7 +45,7 @@ use crate::io::{ EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE, }; use crate::liquidity::LiquiditySource; -use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; +use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger}; use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox; use crate::payment::asynchronous::static_invoice_store::StaticInvoiceStore; use crate::payment::store::{ @@ -533,7 +533,7 @@ where connection_manager: Arc>, output_sweeper: Arc, network_graph: Arc, - liquidity_source: Arc>>, + liquidity_source: Arc>, payment_store: Arc, peer_store: Arc>, keys_manager: Arc, @@ -555,7 +555,7 @@ where bump_tx_event_handler: Arc, channel_manager: Arc, connection_manager: Arc>, output_sweeper: Arc, network_graph: Arc, - liquidity_source: Arc>>, payment_store: Arc, + liquidity_source: Arc>, payment_store: Arc, peer_store: Arc>, keys_manager: Arc, static_invoice_store: Option, onion_messenger: Arc, om_mailbox: Option>, prober: Option>, @@ -664,13 +664,40 @@ where } } - fn lsps2_max_total_opening_fee_msat(payment_metadata: &[u8], amount_msat: u64) -> Option { + fn lsps2_max_total_opening_fee_msat( + payment_metadata: &[u8], amount_msat: u64, counterparty_skimmed_fee_msat: u64, + ) -> Option { let metadata = PaymentMetadata::read(&mut &payment_metadata[..]).ok()?; + Self::lsps2_max_total_opening_fee_msat_from_metadata( + metadata, + amount_msat, + counterparty_skimmed_fee_msat, + ) + } + + fn lsps2_bolt12_max_total_opening_fee_msat( + payment_metadata: &BTreeMap>, amount_msat: u64, + counterparty_skimmed_fee_msat: u64, + ) -> Option { + let metadata = PaymentMetadata::decode_from_bolt12_payment_metadata(payment_metadata)?; + Self::lsps2_max_total_opening_fee_msat_from_metadata( + metadata, + amount_msat, + counterparty_skimmed_fee_msat, + ) + } + + fn lsps2_max_total_opening_fee_msat_from_metadata( + metadata: PaymentMetadata, amount_msat: u64, counterparty_skimmed_fee_msat: u64, + ) -> Option { let lsps2_parameters = metadata.lsps2_parameters?; lsps2_parameters.max_total_opening_fee_msat.or_else(|| { lsps2_parameters.max_proportional_opening_fee_ppm_msat.and_then(|max_prop_fee| { - // If it's a variable amount payment, compute the actual fee. - compute_opening_fee(amount_msat, 0, max_prop_fee) + // `PaymentClaimable::amount_msat` excludes the fee withheld by the LSP. LSPS2 + // proportional fees are calculated from the amount sent by the payer, so add the + // withheld fee back before computing the permitted fee. + let payment_size_msat = amount_msat.checked_add(counterparty_skimmed_fee_msat)?; + compute_opening_fee(payment_size_msat, 0, max_prop_fee) }) }) } @@ -864,15 +891,28 @@ where .as_ref() .and_then(|fields| fields.payment_metadata.as_ref()) .and_then(|metadata| { - Self::lsps2_max_total_opening_fee_msat(metadata, amount_msat) + Self::lsps2_max_total_opening_fee_msat( + metadata, + amount_msat, + counterparty_skimmed_fee_msat, + ) }), + PaymentPurpose::Bolt12OfferPayment { payment_context, .. } => { + payment_context.payment_metadata.as_ref().and_then(|metadata| { + Self::lsps2_bolt12_max_total_opening_fee_msat( + metadata, + amount_msat, + counterparty_skimmed_fee_msat, + ) + }) + }, _ => None, }; let Some(max_total_opening_fee_msat) = max_total_opening_fee_msat else { log_info!( self.logger, - "Refusing inbound payment with hash {} as the counterparty withheld {}msat without valid BOLT11 LSPS2 payment metadata", + "Refusing inbound payment with hash {} as the counterparty withheld {}msat without valid LSPS2 payment metadata", hex_utils::to_string(&payment_hash.0), counterparty_skimmed_fee_msat, ); @@ -894,7 +934,7 @@ where if let Some(info) = payment_info.as_ref() { match &info.kind { - PaymentKind::Bolt11 { .. } => { + PaymentKind::Bolt11 { .. } | PaymentKind::Bolt12Offer { .. } => { let update = PaymentDetailsUpdate { counterparty_skimmed_fee_msat: Some(Some(counterparty_skimmed_fee_msat)), ..PaymentDetailsUpdate::new(payment_id) @@ -907,7 +947,7 @@ where }, }; }, - _ => debug_assert!(false, "We only expect the counterparty to get away with withholding fees for BOLT11 payments."), + _ => debug_assert!(false, "We only expect the counterparty to withhold fees for LSPS2 invoice payments."), } } } @@ -976,6 +1016,8 @@ where hash: Some(payment_hash), preimage: payment_preimage, secret: Some(payment_secret), + counterparty_skimmed_fee_msat: (counterparty_skimmed_fee_msat > 0) + .then_some(counterparty_skimmed_fee_msat), offer_id, payer_note, quantity, @@ -1499,7 +1541,7 @@ where from_prev_str, next_htlcs.len(), to_next_str, - outbound_amount_forwarded_msat.unwrap_or(0), + outbound_amount_forwarded_msat, fee_earned, ); } else { @@ -1510,7 +1552,7 @@ where from_prev_str, next_htlcs.len(), to_next_str, - outbound_amount_forwarded_msat.unwrap_or(0), + outbound_amount_forwarded_msat, fee_earned, ); } @@ -1544,7 +1586,7 @@ where total_fee_earned_msat, skimmed_fee_msat, claim_from_onchain_tx, - outbound_amount_forwarded_msat, + outbound_amount_forwarded_msat: Some(outbound_amount_forwarded_msat), }; self.event_queue.add_event(event).await.map_err(|e| { log_error!(self.logger, "Failed to push to event queue: {}", e); @@ -1773,6 +1815,27 @@ where LdkEvent::InvoiceReceived { .. } => { debug_assert!(false, "We currently don't handle BOLT12 invoices manually, so this event should never be emitted."); }, + LdkEvent::InvoiceRequestReceived { offer_id, invoice_request, context, responder } => { + let lsps2_client = self.liquidity_source.lsps2_client(); + let connection_manager = Arc::clone(&self.connection_manager); + let logger = self.logger.clone(); + // LSPS2 negotiation completes through liquidity events handled by the background + // processor, so awaiting it from this event handler would deadlock the response. + self.runtime.spawn_cancellable_background_task(async move { + if let Err(error) = lsps2_client + .respond_to_invoice_request( + offer_id, + invoice_request, + context, + responder, + connection_manager, + ) + .await + { + log_error!(logger, "Failed handling BOLT12 invoice request: {}", error); + } + }); + }, LdkEvent::ConnectionNeeded { node_id, addresses } => { let spawn_logger = self.logger.clone(); let spawn_cm = Arc::clone(&self.connection_manager); @@ -2031,48 +2094,112 @@ mod tests { max_total_opening_fee_msat: Some(42_000), max_proportional_opening_fee_ppm_msat: None, }), + lsps2_lease_parameters: None, }; assert_eq!( EventHandler::>::lsps2_max_total_opening_fee_msat( &metadata.encode(), - 100_000 + 100_000, + 0, ), Some(42_000) ); } + #[test] + fn lsps2_proportional_fee_uses_gross_payment_amount() { + let metadata = PaymentMetadata { + lsps2_parameters: Some(LSPS2Parameters { + max_total_opening_fee_msat: None, + max_proportional_opening_fee_ppm_msat: Some(10_000), + }), + lsps2_lease_parameters: None, + }; + + assert_eq!( + EventHandler::>::lsps2_max_total_opening_fee_msat( + &metadata.encode(), + 198_000, + 2_000, + ), + Some(2_000) + ); + } + #[test] fn lsps2_payment_metadata_missing_or_malformed_limit_is_rejected() { - let empty_metadata = PaymentMetadata { lsps2_parameters: None }.encode(); + let empty_metadata = + PaymentMetadata { lsps2_parameters: None, lsps2_lease_parameters: None }.encode(); let metadata_without_fee_limit = PaymentMetadata { lsps2_parameters: Some(LSPS2Parameters { max_total_opening_fee_msat: None, max_proportional_opening_fee_ppm_msat: None, }), + lsps2_lease_parameters: None, } .encode(); assert_eq!( EventHandler::>::lsps2_max_total_opening_fee_msat( &empty_metadata, - 100_000 + 100_000, + 0, ), None ); assert_eq!( - EventHandler::>::lsps2_max_total_opening_fee_msat(&[0xff], 100_000), + EventHandler::>::lsps2_max_total_opening_fee_msat(&[0xff], 100_000, 0,), None ); assert_eq!( EventHandler::>::lsps2_max_total_opening_fee_msat( &metadata_without_fee_limit, - 100_000 + 100_000, + 0, ), None ); } + #[test] + fn bolt12_lsps2_payment_metadata_decodes_fee_limit() { + let metadata = PaymentMetadata { + lsps2_parameters: Some(LSPS2Parameters { + max_total_opening_fee_msat: None, + max_proportional_opening_fee_ppm_msat: Some(10_000), + }), + lsps2_lease_parameters: None, + } + .encode_as_bolt12_payment_metadata(); + + assert_eq!( + EventHandler::>::lsps2_bolt12_max_total_opening_fee_msat( + &metadata, 100_000, 0, + ), + Some(1_000) + ); + } + + #[test] + fn bolt12_lsps2_payment_metadata_requires_fee_limit() { + let empty_metadata = BTreeMap::new(); + let metadata_without_fee_limit = + PaymentMetadata { lsps2_parameters: None, lsps2_lease_parameters: None } + .encode_as_bolt12_payment_metadata(); + let mut malformed_metadata = metadata_without_fee_limit.clone(); + *malformed_metadata.values_mut().next().unwrap() = vec![0xff]; + + for metadata in [empty_metadata, malformed_metadata, metadata_without_fee_limit] { + assert_eq!( + EventHandler::>::lsps2_bolt12_max_total_opening_fee_msat( + &metadata, 100_000, 0, + ), + None + ); + } + } + #[tokio::test] async fn event_queue_persistence() { let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); diff --git a/src/io/mod.rs b/src/io/mod.rs index a01aa59a83..8386bc6ad7 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -33,6 +33,14 @@ pub(crate) const PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; pub(crate) const PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE: &str = "pending_payments"; pub(crate) const PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; +/// Cached LSPS2 payment leases are persisted under this prefix. +pub(crate) const LSPS2_LEASE_PERSISTENCE_PRIMARY_NAMESPACE: &str = "lsps2_leases"; +pub(crate) const LSPS2_LEASE_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; + +/// Pending BOLT12 LSPS2 offers are persisted under this prefix. +pub(crate) const LSPS2_PENDING_OFFER_PERSISTENCE_PRIMARY_NAMESPACE: &str = "lsps2_pending_offers"; +pub(crate) const LSPS2_PENDING_OFFER_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; + /// The node metrics will be persisted under this key. pub(crate) const NODE_METRICS_PRIMARY_NAMESPACE: &str = ""; pub(crate) const NODE_METRICS_SECONDARY_NAMESPACE: &str = ""; diff --git a/src/lib.rs b/src/lib.rs index cb570243cc..5469a6ce6e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -544,6 +544,28 @@ impl Node { } }); + // Periodically prune LSPS2 leases before they expire so stale parameters don't accumulate. + let lsps2_client = self.liquidity_source.lsps2_client(); + let prune_logger = Arc::clone(&self.logger); + let mut stop_pruning = self.stop_sender.subscribe(); + self.runtime.spawn_cancellable_background_task(async move { + let mut interval = tokio::time::interval(Duration::from_secs(60 * 60)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + _ = stop_pruning.changed() => return, + _ = interval.tick() => { + if let Err(error) = lsps2_client.prune_stale_leases().await { + log_error!(prune_logger, "Failed pruning stale LSPS2 leases: {}", error); + } + if let Err(error) = lsps2_client.prune_stale_pending_offers().await { + log_error!(prune_logger, "Failed pruning stale LSPS2 offers: {}", error); + } + }, + } + } + }); + // Regularly broadcast node announcements. let bcast_cm = Arc::clone(&self.channel_manager); let bcast_pm = Arc::clone(&self.peer_manager); @@ -805,6 +827,17 @@ impl Node { liquidity_logger, "LSP protocols discovery complete.", ); + if let Err(error) = liquidity_handler + .lsps2_client() + .refill_pending_offer_leases(&discovery_cm) + .await + { + log_error!( + liquidity_logger, + "Failed scheduling LSPS2 offer lease refills: {}", + error + ); + } } } } @@ -1044,6 +1077,7 @@ impl Node { Bolt12Payment::new( Arc::clone(&self.runtime), Arc::clone(&self.channel_manager), + Arc::clone(&self.liquidity_source), Arc::clone(&self.keys_manager), Arc::clone(&self.payment_store), Arc::clone(&self.config), @@ -1061,6 +1095,7 @@ impl Node { Arc::new(Bolt12Payment::new( Arc::clone(&self.runtime), Arc::clone(&self.channel_manager), + Arc::clone(&self.liquidity_source), Arc::clone(&self.keys_manager), Arc::clone(&self.payment_store), Arc::clone(&self.config), diff --git a/src/liquidity/client/lsps2.rs b/src/liquidity/client/lsps2.rs deleted file mode 100644 index 3033f8d827..0000000000 --- a/src/liquidity/client/lsps2.rs +++ /dev/null @@ -1,553 +0,0 @@ -// This file is Copyright its original authors, visible in version control history. -// -// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in -// accordance with one or both of these licenses. - -use std::collections::HashMap; -use std::ops::Deref; -use std::sync::{Arc, Mutex, RwLock}; -use std::time::Duration; - -use bitcoin::secp256k1::{PublicKey, Secp256k1}; -use lightning::ln::channelmanager::MIN_FINAL_CLTV_EXPIRY_DELTA; -use lightning::log_warn; -use lightning::routing::router::{RouteHint, RouteHintHop}; -use lightning::util::ser::Writeable; -use lightning_invoice::{Bolt11Invoice, Bolt11InvoiceDescription, InvoiceBuilder, RoutingFees}; -use lightning_liquidity::lsps0::ser::LSPSRequestId; -use lightning_liquidity::lsps2::event::LSPS2ClientEvent; -use lightning_liquidity::lsps2::msgs::LSPS2OpeningFeeParams; -use lightning_liquidity::lsps2::utils::compute_opening_fee; -use lightning_types::payment::PaymentHash; -use tokio::sync::oneshot; -use tokio::task::JoinSet; - -use crate::connection::ConnectionManager; -use crate::liquidity::{ - select_all_lsps_for_protocol, select_lsps_for_protocol, LspConfig, LspNode, - LIQUIDITY_REQUEST_TIMEOUT_SECS, LSPS_DISCOVERY_WAIT_TIMEOUT_SECS, -}; -use crate::logger::{log_debug, log_error, log_info, LdkLogger}; -use crate::payment::store::LSPS2Parameters; -use crate::payment::PaymentMetadata; -use crate::types::{ChannelManager, KeysManager, LiquidityManager}; -use crate::{Config, Error}; - -pub(crate) struct LSPS2Client -where - L::Target: LdkLogger, -{ - pub(crate) lsp_nodes: Arc>>, - pub(crate) pending_lsps2_fee_requests: - Mutex>>, - pub(crate) pending_buy_requests: - Mutex>>, - pub(crate) channel_manager: Arc, - pub(crate) keys_manager: Arc, - pub(crate) discovery_done_rx: tokio::sync::watch::Receiver, - pub(crate) liquidity_manager: Arc, - pub(crate) config: Arc, - pub(crate) logger: L, -} - -impl LSPS2Client -where - L::Target: LdkLogger, -{ - pub(crate) async fn lsps2_receive_to_jit_channel( - self: Arc, amount_msat: u64, description: &Bolt11InvoiceDescription, - expiry_secs: u32, max_total_lsp_fee_limit_msat: Option, - payment_hash: Option, connection_manager: Arc>, - ) -> Result<(Bolt11Invoice, LspConfig), Error> { - // Connect to all candidate LSPs before querying fees. - let all_offers = self.gather_lsps2_offers(&connection_manager).await?; - let (cheapest_lsp, min_total_fee_msat, min_opening_params) = all_offers - .into_iter() - .flat_map(|(lsp, resp)| { - resp.opening_fee_params_menu - .into_iter() - .map(move |params| (lsp.clone(), params)) - }) - .filter_map(|(lsp, params)| { - if amount_msat < params.min_payment_size_msat - || amount_msat > params.max_payment_size_msat - { - log_debug!(self.logger, - "Skipping LSP {}'s JIT offer as the payment of {}msat doesn't meet LSP limits (min: {}msat, max: {}msat)", - lsp.node_id, - amount_msat, - params.min_payment_size_msat, - params.max_payment_size_msat - ); - None - } else { - compute_opening_fee(amount_msat, params.min_fee_msat, params.proportional as u64) - .map(|fee| (lsp, fee, params)) - } - }) - .min_by_key(|(_, fee, _)| *fee) - .ok_or_else(|| { - log_error!(self.logger, "Failed to handle response from liquidity service",); - Error::LiquidityRequestFailed - })?; - - if let Some(max_total_lsp_fee_limit_msat) = max_total_lsp_fee_limit_msat { - if min_total_fee_msat > max_total_lsp_fee_limit_msat { - log_error!(self.logger, - "Failed to request inbound JIT channel as LSP's requested total opening fee of {}msat exceeds our fee limit of {}msat", - min_total_fee_msat, max_total_lsp_fee_limit_msat - ); - return Err(Error::LiquidityFeeTooHigh); - } - } - - log_debug!( - self.logger, - "Choosing cheapest liquidity offer from LSP {}, will pay {}msat in total LSP fees", - cheapest_lsp.node_id, - min_total_fee_msat - ); - - let buy_response = self - .lsps2_send_buy_request( - Some(amount_msat), - min_opening_params, - Some(&cheapest_lsp.node_id), - ) - .await?; - let lsps2_parameters = LSPS2Parameters { - max_total_opening_fee_msat: Some(min_total_fee_msat), - max_proportional_opening_fee_ppm_msat: None, - }; - - let invoice = self.lsps2_create_jit_invoice( - buy_response, - Some(amount_msat), - description, - expiry_secs, - payment_hash, - lsps2_parameters, - Some(&cheapest_lsp.node_id), - )?; - - log_info!(self.logger, "JIT-channel invoice created: {}", invoice); - Ok((invoice, cheapest_lsp)) - } - - pub(crate) async fn lsps2_receive_variable_amount_to_jit_channel( - self: Arc, description: &Bolt11InvoiceDescription, expiry_secs: u32, - max_proportional_lsp_fee_limit_ppm_msat: Option, payment_hash: Option, - connection_manager: Arc>, - ) -> Result<(Bolt11Invoice, LspConfig), Error> { - // Connect to all candidate LSPs before querying fees. - let all_offers = self.gather_lsps2_offers(&connection_manager).await?; - let (cheapest_lsp, min_prop_fee_ppm_msat, min_opening_params) = all_offers - .into_iter() - .flat_map(|(lsp, resp)| { - resp.opening_fee_params_menu.into_iter().map(move |params| (lsp.clone(), params)) - }) - .map(|(lsp, params)| { - let ppm = params.proportional as u64; - (lsp, ppm, params) - }) - .min_by_key(|(_, ppm, _)| *ppm) - .ok_or_else(|| { - log_error!(self.logger, "Failed to handle response from liquidity service",); - Error::LiquidityRequestFailed - })?; - - if let Some(max_proportional_lsp_fee_limit_ppm_msat) = - max_proportional_lsp_fee_limit_ppm_msat - { - if min_prop_fee_ppm_msat > max_proportional_lsp_fee_limit_ppm_msat { - log_error!(self.logger, - "Failed to request inbound JIT channel as LSP's requested proportional opening fee of {} ppm msat exceeds our fee limit of {} ppm msat", - min_prop_fee_ppm_msat, - max_proportional_lsp_fee_limit_ppm_msat - ); - return Err(Error::LiquidityFeeTooHigh); - } - } - - log_debug!( - self.logger, - "Choosing cheapest liquidity offer from LSP {}, will pay {}ppm msat in proportional LSP fees", - cheapest_lsp.node_id, - min_prop_fee_ppm_msat - ); - - let buy_response = self - .lsps2_send_buy_request(None, min_opening_params, Some(&cheapest_lsp.node_id)) - .await?; - let lsps2_parameters = LSPS2Parameters { - max_total_opening_fee_msat: None, - max_proportional_opening_fee_ppm_msat: Some(min_prop_fee_ppm_msat), - }; - let invoice = self.lsps2_create_jit_invoice( - buy_response, - None, - description, - expiry_secs, - payment_hash, - lsps2_parameters, - Some(&cheapest_lsp.node_id), - )?; - - log_info!(self.logger, "JIT-channel invoice created: {}", invoice); - Ok((invoice, cheapest_lsp)) - } - - async fn gather_lsps2_offers( - self: &Arc, connection_manager: &Arc>, - ) -> Result, Error> { - let lsps2_nodes = self.get_lsps2_nodes().await?; - - // Connect to all candidate LSPs in parallel. - let mut connect_set = JoinSet::new(); - for lsp_node in &lsps2_nodes { - let cm = Arc::clone(connection_manager); - let node_id = lsp_node.node_id; - let addr = lsp_node.address.clone(); - let logger = self.logger.clone(); - connect_set.spawn(async move { - if let Err(e) = cm.connect_peer_if_necessary(node_id, addr).await { - log_warn!(logger, "Failed to connect to LSP {} for fee query: {}", node_id, e); - } - }); - } - while connect_set.join_next().await.is_some() {} - - let mut all_offers: Vec<(LspConfig, LSPS2FeeResponse)> = - Vec::with_capacity(lsps2_nodes.len()); - let mut fee_set: JoinSet<(LspConfig, Result)> = JoinSet::new(); - for lsp_node in &lsps2_nodes { - let lsp = lsp_node.clone(); - let client = Arc::clone(self); - fee_set.spawn(async move { - let res = client.lsps2_request_opening_fee_params(Some(&lsp.node_id)).await; - (lsp, res) - }); - } - while let Some(join_result) = fee_set.join_next().await { - match join_result { - Ok((lsp, Ok(fees))) => all_offers.push((lsp, fees)), - Ok((lsp, Err(e))) => { - log_warn!(self.logger, "Failed to get fees from LSP {}: {}", lsp.node_id, e) - }, - Err(e) => { - log_warn!(self.logger, "Failed to get fees from LSP: {}", e) - }, - } - } - - Ok(all_offers) - } -} - -impl LSPS2Client -where - L::Target: LdkLogger, -{ - async fn lsps2_request_opening_fee_params( - &self, node_id: Option<&PublicKey>, - ) -> Result { - let lsps2_node = select_lsps_for_protocol(&self.lsp_nodes, 2, node_id) - .ok_or(Error::LiquiditySourceUnavailable)?; - - let client_handler = self.liquidity_manager.lsps2_client_handler().ok_or_else(|| { - log_error!(self.logger, "Liquidity client was not configured.",); - Error::LiquiditySourceUnavailable - })?; - - let (fee_request_sender, fee_request_receiver) = oneshot::channel(); - { - let mut pending_fee_requests_lock = - self.pending_lsps2_fee_requests.lock().expect("lock"); - let request_id = - client_handler.request_opening_params(lsps2_node.node_id, lsps2_node.token.clone()); - pending_fee_requests_lock.insert(request_id, fee_request_sender); - } - - tokio::time::timeout( - Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), - fee_request_receiver, - ) - .await - .map_err(|e| { - log_error!(self.logger, "Liquidity request timed out: {}", e); - Error::LiquidityRequestFailed - })? - .map_err(|e| { - log_error!(self.logger, "Failed to handle response from liquidity service: {}", e); - Error::LiquidityRequestFailed - }) - } - - async fn lsps2_send_buy_request( - &self, amount_msat: Option, opening_fee_params: LSPS2OpeningFeeParams, - node_id: Option<&PublicKey>, - ) -> Result { - let lsps2_node = select_lsps_for_protocol(&self.lsp_nodes, 2, node_id) - .ok_or(Error::LiquiditySourceUnavailable)?; - - let client_handler = self.liquidity_manager.lsps2_client_handler().ok_or_else(|| { - log_error!(self.logger, "Liquidity client was not configured.",); - Error::LiquiditySourceUnavailable - })?; - - let (buy_request_sender, buy_request_receiver) = oneshot::channel(); - { - let mut pending_buy_requests_lock = self.pending_buy_requests.lock().expect("lock"); - let request_id = client_handler - .select_opening_params(lsps2_node.node_id, amount_msat, opening_fee_params) - .map_err(|e| { - log_error!( - self.logger, - "Failed to send buy request to liquidity service: {:?}", - e - ); - Error::LiquidityRequestFailed - })?; - pending_buy_requests_lock.insert(request_id, buy_request_sender); - } - - let buy_response = tokio::time::timeout( - Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), - buy_request_receiver, - ) - .await - .map_err(|e| { - log_error!(self.logger, "Liquidity request timed out: {}", e); - Error::LiquidityRequestFailed - })? - .map_err(|e| { - log_error!(self.logger, "Failed to handle response from liquidity service: {:?}", e); - Error::LiquidityRequestFailed - })?; - - Ok(buy_response) - } - - fn lsps2_create_jit_invoice( - &self, buy_response: LSPS2BuyResponse, amount_msat: Option, - description: &Bolt11InvoiceDescription, expiry_secs: u32, - payment_hash: Option, lsps2_parameters: LSPS2Parameters, - node_id: Option<&PublicKey>, - ) -> Result { - let lsps2_node = select_lsps_for_protocol(&self.lsp_nodes, 2, node_id) - .ok_or(Error::LiquiditySourceUnavailable)?; - - // LSPS2 requires min_final_cltv_expiry_delta to be at least 2 more than usual. - let min_final_cltv_expiry_delta = MIN_FINAL_CLTV_EXPIRY_DELTA + 2; - let encoded_payment_metadata = - PaymentMetadata { lsps2_parameters: Some(lsps2_parameters) }.encode(); - let (payment_hash, payment_secret, payment_metadata) = match payment_hash { - Some(payment_hash) => { - let (payment_secret, payment_metadata) = self - .channel_manager - .create_inbound_payment_for_hash( - payment_hash, - None, - expiry_secs, - Some(min_final_cltv_expiry_delta), - Some(encoded_payment_metadata), - ) - .map_err(|e| { - log_error!(self.logger, "Failed to register inbound payment: {:?}", e); - Error::InvoiceCreationFailed - })?; - (payment_hash, payment_secret, payment_metadata) - }, - None => self - .channel_manager - .create_inbound_payment( - None, - expiry_secs, - Some(min_final_cltv_expiry_delta), - Some(encoded_payment_metadata), - ) - .map_err(|e| { - log_error!(self.logger, "Failed to register inbound payment: {:?}", e); - Error::InvoiceCreationFailed - })?, - }; - - let route_hint = RouteHint(vec![RouteHintHop { - src_node_id: lsps2_node.node_id, - short_channel_id: buy_response.intercept_scid, - fees: RoutingFees { base_msat: 0, proportional_millionths: 0 }, - cltv_expiry_delta: buy_response.cltv_expiry_delta as u16, - htlc_minimum_msat: None, - htlc_maximum_msat: None, - }]); - - let currency = self.config.network.into(); - let mut invoice_builder = InvoiceBuilder::new(currency) - .invoice_description(description.clone()) - .payment_hash(payment_hash) - .payment_secret(payment_secret) - .current_timestamp() - .min_final_cltv_expiry_delta(min_final_cltv_expiry_delta.into()) - .expiry_time(Duration::from_secs(expiry_secs.into())) - .private_route(route_hint); - - if let Some(amount_msat) = amount_msat { - invoice_builder = invoice_builder.amount_milli_satoshis(amount_msat).basic_mpp(); - } - - let invoice = if let Some(payment_metadata) = payment_metadata { - invoice_builder.payment_metadata(payment_metadata).build_signed(|hash| { - Secp256k1::new() - .sign_ecdsa_recoverable(hash, &self.keys_manager.get_node_secret_key()) - }) - } else { - invoice_builder.build_signed(|hash| { - Secp256k1::new() - .sign_ecdsa_recoverable(hash, &self.keys_manager.get_node_secret_key()) - }) - }; - invoice.map_err(|e| { - log_error!(self.logger, "Failed to build and sign invoice: {}", e); - Error::InvoiceCreationFailed - }) - } - - pub(crate) async fn handle_event(&self, event: LSPS2ClientEvent) { - match event { - LSPS2ClientEvent::OpeningParametersReady { - request_id, - counterparty_node_id, - opening_fee_params_menu, - } => { - if self - .lsp_nodes - .read() - .expect("lock") - .iter() - .any(|n| n.node_id == counterparty_node_id) - { - if let Some(sender) = - self.pending_lsps2_fee_requests.lock().expect("lock").remove(&request_id) - { - let response = LSPS2FeeResponse { opening_fee_params_menu }; - - match sender.send(response) { - Ok(()) => (), - Err(_) => { - log_error!( - self.logger, - "Failed to handle response for request {:?} from liquidity service", - request_id - ); - }, - } - } else { - debug_assert!( - false, - "Received response from liquidity service for unknown request." - ); - log_error!( - self.logger, - "Received response from liquidity service for unknown request." - ); - } - } else { - log_error!( - self.logger, - "Received unexpected LSPS2Client::OpeningParametersReady event!" - ); - } - }, - LSPS2ClientEvent::InvoiceParametersReady { - request_id, - counterparty_node_id, - intercept_scid, - cltv_expiry_delta, - .. - } => { - if self - .lsp_nodes - .read() - .expect("lock") - .iter() - .any(|n| n.node_id == counterparty_node_id) - { - if let Some(sender) = - self.pending_buy_requests.lock().expect("lock").remove(&request_id) - { - let response = LSPS2BuyResponse { intercept_scid, cltv_expiry_delta }; - - match sender.send(response) { - Ok(()) => (), - Err(_) => { - log_error!( - self.logger, - "Failed to handle response for request {:?} from liquidity service", - request_id - ); - }, - } - } else { - debug_assert!( - false, - "Received response from liquidity service for unknown request." - ); - log_error!( - self.logger, - "Received response from liquidity service for unknown request." - ); - } - } else { - log_error!( - self.logger, - "Received unexpected LSPS2Client::InvoiceParametersReady event!" - ); - } - }, - _ => { - log_error!(self.logger, "Received unexpected LSPS2Client liquidity event!"); - }, - } - } - - async fn get_lsps2_nodes(&self) -> Result, Error> { - let has_undiscovered_protocol = - self.lsp_nodes.read().expect("lock").iter().any(|n| n.supported_protocols.is_none()); - - if has_undiscovered_protocol { - // LSP protocol discovery is still in flight, we wait briefly for it to finish, then re-check. - let mut rx = self.discovery_done_rx.clone(); - if !*rx.borrow() { - log_debug!( - self.logger, - "Waiting for LSP protocol discovery to complete before selecting LSPS2 nodes." - ); - let _ = tokio::time::timeout( - Duration::from_secs(LSPS_DISCOVERY_WAIT_TIMEOUT_SECS), - rx.wait_for(|done| *done), - ) - .await; - } - } - - let lsps2_nodes = select_all_lsps_for_protocol(&self.lsp_nodes, 2); - if lsps2_nodes.is_empty() { - log_error!(self.logger, "No LSPs available for LSPS2 protocol."); - return Err(Error::LiquiditySourceUnavailable); - }; - Ok(lsps2_nodes) - } -} - -#[derive(Debug, Clone)] -pub(crate) struct LSPS2FeeResponse { - opening_fee_params_menu: Vec, -} - -#[derive(Debug, Clone)] -pub(crate) struct LSPS2BuyResponse { - intercept_scid: u64, - cltv_expiry_delta: u32, -} diff --git a/src/liquidity/client/lsps2/mod.rs b/src/liquidity/client/lsps2/mod.rs new file mode 100644 index 0000000000..674fb88c16 --- /dev/null +++ b/src/liquidity/client/lsps2/mod.rs @@ -0,0 +1,1352 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. + +use std::collections::{HashMap, HashSet}; +use std::future::Future; +use std::hash::Hash; +use std::ops::Deref; +use std::sync::{Arc, Mutex, RwLock}; +use std::time::Duration; + +use bitcoin::secp256k1::{PublicKey, Secp256k1}; +use lightning::blinded_path::message::OffersContext; +use lightning::ln::channelmanager::MIN_FINAL_CLTV_EXPIRY_DELTA; +use lightning::log_warn; +use lightning::offers::invoice_request::InvoiceRequest; +use lightning::offers::offer::OfferId; +use lightning::onion_message::messenger::Responder; +use lightning::routing::router::{RouteHint, RouteHintHop}; +use lightning::util::ser::Writeable; +use lightning_invoice::{Bolt11Invoice, Bolt11InvoiceDescription, InvoiceBuilder, RoutingFees}; +use lightning_liquidity::lsps0::ser::LSPSRequestId; +use lightning_liquidity::lsps2::event::LSPS2ClientEvent; +use lightning_liquidity::lsps2::msgs::LSPS2OpeningFeeParams; +use lightning_liquidity::lsps2::router::PaymentParameters; +use lightning_liquidity::lsps2::utils::compute_opening_fee; +use lightning_types::payment::PaymentHash; +use tokio::sync::oneshot; +use tokio::task::JoinSet; + +use crate::connection::ConnectionManager; +use crate::liquidity::{ + select_all_lsps_for_protocol, select_lsps_for_protocol, LspConfig, LspNode, + LIQUIDITY_REQUEST_TIMEOUT_SECS, LSPS_DISCOVERY_WAIT_TIMEOUT_SECS, +}; +use crate::logger::{log_debug, log_error, log_info, LdkLogger}; +use crate::payment::store::LSPS2Parameters; +use crate::payment::PaymentMetadata; +use crate::types::{ChannelManager, KeysManager, LiquidityManager}; +use crate::{Config, Error}; + +use self::state::{ + now_secs, LSPS2LeaseState, LeaseRequestKey, PaymentLease, PaymentLeaseId, PaymentLeaseStore, + PendingInvoiceRequestState, PendingLeaseRequestState, PendingOffer, PendingOfferAmount, + PendingOfferId, PendingOfferState, PendingOfferStore, +}; + +async fn consume_after_persisted_removal( + value: T, persist_removal: RF, consume: CF, +) -> Result +where + T: Clone, + RF: FnOnce(T) -> Fut, + CF: FnOnce(&T), + Fut: Future>, +{ + persist_removal(value.clone()).await?; + consume(&value); + Ok(value) +} + +const DEFAULT_BOLT12_INVOICE_EXPIRY_SECS: u32 = 2 * 60 * 60; +const LEASE_NEGOTIATION_MAX_ATTEMPTS: usize = 3; + +fn should_retry_lease_negotiation(error: Error, attempt: usize) -> bool { + error == Error::LiquidityRequestFailed && attempt < LEASE_NEGOTIATION_MAX_ATTEMPTS +} + +async fn try_lease_candidates( + candidates: Vec, candidate_key: KF, mut attempt: AF, +) -> Result +where + K: Copy + Eq + Hash, + KF: Fn(&T) -> K, + AF: FnMut(T) -> Fut, + Fut: Future>, +{ + let mut failed_candidates = HashSet::new(); + let mut last_error = None; + for candidate in candidates { + let key = candidate_key(&candidate); + if failed_candidates.contains(&key) { + continue; + } + match attempt(candidate).await { + Ok(result) => return Ok(result), + Err(error) => { + failed_candidates.insert(key); + last_error = Some(error); + }, + } + } + Err(last_error.expect("lease candidates are non-empty")) +} + +#[derive(Debug, PartialEq, Eq)] +enum JitInvoiceRequest { + Fixed { amount_msat: u64, max_total_fee_msat: Option }, + Variable { amount_msat: u64, max_proportional_fee_ppm_msat: Option }, +} + +impl JitInvoiceRequest { + fn allow_mpp(&self) -> bool { + matches!(self, Self::Fixed { .. }) + } +} + +fn jit_invoice_request( + pending_offer: &PendingOffer, invoice_request: &InvoiceRequest, +) -> Result { + let amount_msat = invoice_request.amount_msats().ok_or(Error::InvalidAmount)?; + match pending_offer.amount { + PendingOfferAmount::Fixed { amount_msat: offer_amount_msat, max_total_fee_msat } => { + if amount_msat < offer_amount_msat { + return Err(Error::InvalidAmount); + } + Ok(JitInvoiceRequest::Fixed { amount_msat, max_total_fee_msat }) + }, + PendingOfferAmount::Variable { max_proportional_fee_ppm_msat } => { + Ok(JitInvoiceRequest::Variable { amount_msat, max_proportional_fee_ppm_msat }) + }, + } +} + +pub(crate) struct LSPS2Client +where + L::Target: LdkLogger, +{ + pub(crate) lsp_nodes: Arc>>, + pub(crate) pending_lsps2_fee_requests: + Mutex>>, + pub(crate) pending_buy_requests: + Mutex>>, + pub(crate) lease_store: Arc>, + pub(crate) lease_state: Mutex, + pub(crate) pending_offer_store: Arc>, + pub(crate) pending_offer_state: Mutex, + pub(crate) pending_invoice_request_state: Mutex, + pub(crate) pending_lease_request_state: Mutex, + pub(crate) channel_manager: Arc, + pub(crate) keys_manager: Arc, + pub(crate) discovery_done_rx: tokio::sync::watch::Receiver, + pub(crate) liquidity_manager: Arc, + pub(crate) config: Arc, + pub(crate) logger: L, +} + +impl LSPS2Client +where + L::Target: LdkLogger, +{ + pub(crate) async fn prune_stale_leases(&self) -> Result<(), Error> { + let stale_ids = self + .lease_store + .list_filter(|lease| !state::is_lease_usable(lease)) + .into_iter() + .map(|lease| lease.id) + .collect::>(); + for id in stale_ids { + self.lease_store.remove(&id).await?; + } + self.lease_state.lock().expect("lock").prune(); + Ok(()) + } + + pub(crate) async fn prune_stale_pending_offers(&self) -> Result<(), Error> { + let stale_ids = self.pending_offer_state.lock().expect("lock").prune(now_secs()); + for id in stale_ids { + self.pending_offer_store.remove(&id).await?; + } + Ok(()) + } + + pub(crate) async fn register_pending_offer(&self, offer: PendingOffer) -> Result<(), Error> { + self.pending_offer_store.insert_or_update(offer.clone()).await?; + let evicted = self.pending_offer_state.lock().expect("lock").register(offer); + if let Some(id) = evicted { + self.pending_offer_store.remove(&id).await?; + } + Ok(()) + } + + pub(crate) async fn refill_pending_offer_leases( + self: &Arc, connection_manager: &Arc>, + ) -> Result<(), Error> { + self.prune_stale_pending_offers().await?; + let pending_offers = self.pending_offer_state.lock().expect("lock").offers(); + for pending_offer in pending_offers { + match pending_offer.amount { + PendingOfferAmount::Fixed { amount_msat, max_total_fee_msat } => { + self.schedule_fixed_lease_refill( + amount_msat, + max_total_fee_msat, + connection_manager, + ); + }, + PendingOfferAmount::Variable { max_proportional_fee_ppm_msat } => { + self.schedule_variable_lease_refill( + max_proportional_fee_ppm_msat, + connection_manager, + ); + }, + } + } + Ok(()) + } + + pub(crate) async fn pending_offer( + &self, offer_id: PendingOfferId, + ) -> Result, Error> { + let offer = self.pending_offer_state.lock().expect("lock").get(&offer_id, now_secs()); + if let Some(offer) = offer { + self.pending_offer_store.insert_or_update(offer.clone()).await?; + Ok(Some(offer)) + } else { + self.pending_offer_store.remove(&offer_id).await?; + Ok(None) + } + } + + pub(crate) async fn respond_to_invoice_request( + self: Arc, offer_id: OfferId, invoice_request: InvoiceRequest, + context: Option, responder: Responder, + connection_manager: Arc>, + ) -> Result<(), Error> { + let pending_offer_id = offer_id.into(); + let is_jit_offer = + self.pending_offer_state.lock().expect("lock").contains(&pending_offer_id); + if !is_jit_offer { + let payment_metadata = match context.as_ref() { + Some(OffersContext::InvoiceRequest { payment_metadata, .. }) => { + payment_metadata.clone() + }, + _ => None, + }; + return self + .channel_manager + .respond_to_invoice_request( + invoice_request, + context, + responder, + payment_metadata, + DEFAULT_BOLT12_INVOICE_EXPIRY_SECS, + true, + ) + .map_err(|error| { + log_error!( + self.logger, + "Failed responding to BOLT12 invoice request: {}", + error + ); + Error::InvoiceCreationFailed + }); + } + + let request_lock = + self.pending_invoice_request_state.lock().expect("lock").request_lock(pending_offer_id); + let result = { + let _guard = request_lock.lock().await; + self.respond_to_jit_invoice_request( + pending_offer_id, + invoice_request, + context, + responder, + &connection_manager, + ) + .await + }; + drop(request_lock); + self.pending_invoice_request_state.lock().expect("lock").prune(); + result + } + + async fn respond_to_jit_invoice_request( + self: &Arc, pending_offer_id: PendingOfferId, invoice_request: InvoiceRequest, + context: Option, responder: Responder, + connection_manager: &Arc>, + ) -> Result<(), Error> { + let pending_offer = + self.pending_offer(pending_offer_id).await?.ok_or(Error::InvalidOffer)?; + let request = jit_invoice_request(&pending_offer, &invoice_request)?; + let allow_mpp = request.allow_mpp(); + let (lease, fee_parameters) = match request { + JitInvoiceRequest::Fixed { amount_msat, max_total_fee_msat } => { + let (lease, total_fee_msat, _, _) = self + .acquire_fixed_lease(amount_msat, max_total_fee_msat, connection_manager) + .await?; + ( + lease, + LSPS2Parameters { + max_total_opening_fee_msat: Some(total_fee_msat), + max_proportional_opening_fee_ppm_msat: None, + }, + ) + }, + JitInvoiceRequest::Variable { amount_msat: _, max_proportional_fee_ppm_msat } => { + let (lease, proportional_fee, _, _) = self + .acquire_variable_lease(max_proportional_fee_ppm_msat, connection_manager) + .await?; + ( + lease, + LSPS2Parameters { + max_total_opening_fee_msat: None, + max_proportional_opening_fee_ppm_msat: Some(proportional_fee), + }, + ) + }, + }; + + let remaining_validity_secs = lease + .valid_until + .checked_sub(now_secs()) + .filter(|remaining| *remaining > 0) + .ok_or(Error::LiquidityRequestFailed)?; + let relative_expiry_secs = DEFAULT_BOLT12_INVOICE_EXPIRY_SECS + .min(remaining_validity_secs.try_into().unwrap_or(u32::MAX)); + let lease_parameters = PaymentParameters { + lsp_node_id: lease.id.lsp_node_id, + intercept_scid: lease.id.intercept_scid, + cltv_expiry_delta: lease + .cltv_expiry_delta + .try_into() + .map_err(|_| Error::LiquidityRequestFailed)?, + payment_size_msat: lease.payment_size_msat, + valid_until: lease.valid_until, + }; + let payment_metadata = PaymentMetadata { + lsps2_parameters: Some(fee_parameters), + lsps2_lease_parameters: Some(lease_parameters), + } + .encode_as_bolt12_payment_metadata(); + + self.channel_manager + .respond_to_invoice_request( + invoice_request, + context, + responder, + Some(payment_metadata), + relative_expiry_secs, + allow_mpp, + ) + .map_err(|error| { + log_error!(self.logger, "Failed responding to LSPS2 invoice request: {}", error); + Error::InvoiceCreationFailed + }) + } + + pub(crate) async fn lsps2_receive_to_jit_channel( + self: Arc, amount_msat: u64, description: &Bolt11InvoiceDescription, + expiry_secs: u32, max_total_lsp_fee_limit_msat: Option, + payment_hash: Option, connection_manager: Arc>, + ) -> Result<(Bolt11Invoice, LspConfig), Error> { + let (lease, total_fee_msat, lsp, was_negotiated) = self + .acquire_fixed_lease(amount_msat, max_total_lsp_fee_limit_msat, &connection_manager) + .await?; + let lsps2_parameters = LSPS2Parameters { + max_total_opening_fee_msat: Some(total_fee_msat), + max_proportional_opening_fee_ppm_msat: None, + }; + + let invoice = self.lsps2_create_jit_invoice( + LSPS2BuyResponse::from(&lease), + Some(amount_msat), + description, + expiry_secs, + payment_hash, + lsps2_parameters, + Some(&lsp.node_id), + )?; + + if was_negotiated { + log_info!(self.logger, "JIT-channel invoice created: {}", invoice); + } + Ok((invoice, lsp)) + } + + pub(crate) async fn lsps2_receive_variable_amount_to_jit_channel( + self: Arc, description: &Bolt11InvoiceDescription, expiry_secs: u32, + max_proportional_lsp_fee_limit_ppm_msat: Option, payment_hash: Option, + connection_manager: Arc>, + ) -> Result<(Bolt11Invoice, LspConfig), Error> { + let (lease, proportional_fee, lsp, was_negotiated) = self + .acquire_variable_lease(max_proportional_lsp_fee_limit_ppm_msat, &connection_manager) + .await?; + let lsps2_parameters = LSPS2Parameters { + max_total_opening_fee_msat: None, + max_proportional_opening_fee_ppm_msat: Some(proportional_fee), + }; + let invoice = self.lsps2_create_jit_invoice( + LSPS2BuyResponse::from(&lease), + None, + description, + expiry_secs, + payment_hash, + lsps2_parameters, + Some(&lsp.node_id), + )?; + + if was_negotiated { + log_info!(self.logger, "JIT-channel invoice created: {}", invoice); + } + Ok((invoice, lsp)) + } + + async fn acquire_fixed_lease( + self: &Arc, amount_msat: u64, max_total_lsp_fee_limit_msat: Option, + connection_manager: &Arc>, + ) -> Result<(PaymentLease, u64, LspConfig, bool), Error> { + if let Some((lease, total_fee_msat, lsp)) = + self.take_cached_fixed_lease(amount_msat, max_total_lsp_fee_limit_msat).await? + { + self.schedule_fixed_lease_refill( + amount_msat, + max_total_lsp_fee_limit_msat, + connection_manager, + ); + return Ok((lease, total_fee_msat, lsp, false)); + } + let request_lock = self + .pending_lease_request_state + .lock() + .expect("lock") + .request_lock(LeaseRequestKey::Fixed(amount_msat)); + let _request_guard = request_lock.lock().await; + if let Some((lease, total_fee_msat, lsp)) = + self.take_cached_fixed_lease(amount_msat, max_total_lsp_fee_limit_msat).await? + { + self.schedule_fixed_lease_refill( + amount_msat, + max_total_lsp_fee_limit_msat, + connection_manager, + ); + return Ok((lease, total_fee_msat, lsp, false)); + } + + let (negotiated_lease, min_total_fee_msat, cheapest_lsp) = self + .negotiate_fixed_lease(amount_msat, max_total_lsp_fee_limit_msat, connection_manager) + .await?; + let lease = self.consume_lease(&negotiated_lease.id).await?; + self.schedule_fixed_lease_refill( + amount_msat, + max_total_lsp_fee_limit_msat, + connection_manager, + ); + Ok((lease, min_total_fee_msat, cheapest_lsp, true)) + } + + async fn negotiate_fixed_lease( + self: &Arc, amount_msat: u64, max_total_lsp_fee_limit_msat: Option, + connection_manager: &Arc>, + ) -> Result<(PaymentLease, u64, LspConfig), Error> { + let mut attempt = 1; + loop { + let result = self + .negotiate_fixed_lease_once( + amount_msat, + max_total_lsp_fee_limit_msat, + connection_manager, + ) + .await; + match result { + Err(error) if should_retry_lease_negotiation(error, attempt) => { + log_warn!( + self.logger, + "LSPS2 lease negotiation attempt {} failed, retrying: {}", + attempt, + error + ); + attempt += 1; + }, + result => return result, + } + } + } + + async fn negotiate_fixed_lease_once( + self: &Arc, amount_msat: u64, max_total_lsp_fee_limit_msat: Option, + connection_manager: &Arc>, + ) -> Result<(PaymentLease, u64, LspConfig), Error> { + let all_offers = self.gather_lsps2_offers(connection_manager).await?; + let mut candidates = all_offers + .into_iter() + .flat_map(|(lsp, resp)| { + resp.opening_fee_params_menu + .into_iter() + .map(move |params| (lsp.clone(), params)) + }) + .filter_map(|(lsp, params)| { + if amount_msat < params.min_payment_size_msat + || amount_msat > params.max_payment_size_msat + { + log_debug!(self.logger, + "Skipping LSP {}'s JIT offer as the payment of {}msat doesn't meet LSP limits (min: {}msat, max: {}msat)", + lsp.node_id, + amount_msat, + params.min_payment_size_msat, + params.max_payment_size_msat + ); + None + } else { + compute_opening_fee(amount_msat, params.min_fee_msat, params.proportional as u64) + .map(|fee| (lsp, fee, params)) + } + }) + .collect::>(); + candidates.sort_unstable_by_key(|(_, fee, _)| *fee); + let min_total_fee_msat = candidates.first().map(|(_, fee, _)| *fee).ok_or_else(|| { + log_error!(self.logger, "Failed to handle response from liquidity service",); + Error::LiquidityRequestFailed + })?; + + if let Some(max_total_lsp_fee_limit_msat) = max_total_lsp_fee_limit_msat { + if min_total_fee_msat > max_total_lsp_fee_limit_msat { + log_error!(self.logger, + "Failed to request inbound JIT channel as LSP's requested total opening fee of {}msat exceeds our fee limit of {}msat", + min_total_fee_msat, max_total_lsp_fee_limit_msat + ); + return Err(Error::LiquidityFeeTooHigh); + } + candidates.retain(|(_, fee, _)| *fee <= max_total_lsp_fee_limit_msat); + } + + try_lease_candidates( + candidates, + |(lsp, _, _)| lsp.node_id, + |(lsp, total_fee_msat, opening_params)| { + let client = Arc::clone(self); + async move { + log_debug!( + client.logger, + "Choosing liquidity offer from LSP {}, will pay {}msat in total LSP fees", + lsp.node_id, + total_fee_msat + ); + match client + .lsps2_send_buy_request( + Some(amount_msat), + opening_params, + Some(&lsp.node_id), + ) + .await + { + Ok(lease) => Ok((lease, total_fee_msat, lsp)), + Err(error) => { + log_warn!( + client.logger, + "Failed negotiating LSPS2 payment lease with LSP {}, trying the next candidate: {}", + lsp.node_id, + error + ); + Err(error) + }, + } + } + }, + ) + .await + } + + async fn acquire_variable_lease( + self: &Arc, max_proportional_lsp_fee_limit_ppm_msat: Option, + connection_manager: &Arc>, + ) -> Result<(PaymentLease, u64, LspConfig, bool), Error> { + if let Some((lease, proportional_fee, lsp)) = + self.take_cached_variable_lease(max_proportional_lsp_fee_limit_ppm_msat).await? + { + self.schedule_variable_lease_refill( + max_proportional_lsp_fee_limit_ppm_msat, + connection_manager, + ); + return Ok((lease, proportional_fee, lsp, false)); + } + let request_lock = self + .pending_lease_request_state + .lock() + .expect("lock") + .request_lock(LeaseRequestKey::Variable); + let _request_guard = request_lock.lock().await; + if let Some((lease, proportional_fee, lsp)) = + self.take_cached_variable_lease(max_proportional_lsp_fee_limit_ppm_msat).await? + { + self.schedule_variable_lease_refill( + max_proportional_lsp_fee_limit_ppm_msat, + connection_manager, + ); + return Ok((lease, proportional_fee, lsp, false)); + } + + let (negotiated_lease, min_prop_fee_ppm_msat, cheapest_lsp) = self + .negotiate_variable_lease(max_proportional_lsp_fee_limit_ppm_msat, connection_manager) + .await?; + let lease = self.consume_lease(&negotiated_lease.id).await?; + self.schedule_variable_lease_refill( + max_proportional_lsp_fee_limit_ppm_msat, + connection_manager, + ); + Ok((lease, min_prop_fee_ppm_msat, cheapest_lsp, true)) + } + + async fn negotiate_variable_lease( + self: &Arc, max_proportional_lsp_fee_limit_ppm_msat: Option, + connection_manager: &Arc>, + ) -> Result<(PaymentLease, u64, LspConfig), Error> { + let mut attempt = 1; + loop { + let result = self + .negotiate_variable_lease_once( + max_proportional_lsp_fee_limit_ppm_msat, + connection_manager, + ) + .await; + match result { + Err(error) if should_retry_lease_negotiation(error, attempt) => { + log_warn!( + self.logger, + "LSPS2 lease negotiation attempt {} failed, retrying: {}", + attempt, + error + ); + attempt += 1; + }, + result => return result, + } + } + } + + async fn negotiate_variable_lease_once( + self: &Arc, max_proportional_lsp_fee_limit_ppm_msat: Option, + connection_manager: &Arc>, + ) -> Result<(PaymentLease, u64, LspConfig), Error> { + let all_offers = self.gather_lsps2_offers(connection_manager).await?; + let mut candidates = all_offers + .into_iter() + .flat_map(|(lsp, resp)| { + resp.opening_fee_params_menu.into_iter().map(move |params| (lsp.clone(), params)) + }) + .map(|(lsp, params)| { + let ppm = params.proportional as u64; + (lsp, ppm, params) + }) + .collect::>(); + candidates.sort_unstable_by_key(|(_, ppm, _)| *ppm); + let min_prop_fee_ppm_msat = + candidates.first().map(|(_, ppm, _)| *ppm).ok_or_else(|| { + log_error!(self.logger, "Failed to handle response from liquidity service",); + Error::LiquidityRequestFailed + })?; + + if let Some(max_proportional_lsp_fee_limit_ppm_msat) = + max_proportional_lsp_fee_limit_ppm_msat + { + if min_prop_fee_ppm_msat > max_proportional_lsp_fee_limit_ppm_msat { + log_error!(self.logger, + "Failed to request inbound JIT channel as LSP's requested proportional opening fee of {} ppm msat exceeds our fee limit of {} ppm msat", + min_prop_fee_ppm_msat, + max_proportional_lsp_fee_limit_ppm_msat + ); + return Err(Error::LiquidityFeeTooHigh); + } + candidates.retain(|(_, ppm, _)| *ppm <= max_proportional_lsp_fee_limit_ppm_msat); + } + + try_lease_candidates( + candidates, + |(lsp, _, _)| lsp.node_id, + |(lsp, proportional_fee_ppm_msat, opening_params)| { + let client = Arc::clone(self); + async move { + log_debug!( + client.logger, + "Choosing liquidity offer from LSP {}, will pay {}ppm msat in proportional LSP fees", + lsp.node_id, + proportional_fee_ppm_msat + ); + match client + .lsps2_send_buy_request(None, opening_params, Some(&lsp.node_id)) + .await + { + Ok(lease) => Ok((lease, proportional_fee_ppm_msat, lsp)), + Err(error) => { + log_warn!( + client.logger, + "Failed negotiating LSPS2 payment lease with LSP {}, trying the next candidate: {}", + lsp.node_id, + error + ); + Err(error) + }, + } + } + }, + ) + .await + } + + fn schedule_fixed_lease_refill( + self: &Arc, amount_msat: u64, max_total_lsp_fee_limit_msat: Option, + connection_manager: &Arc>, + ) { + let client = Arc::clone(self); + let connection_manager = Arc::clone(connection_manager); + tokio::spawn(async move { + if let Err(error) = client + .cache_fixed_lease(amount_msat, max_total_lsp_fee_limit_msat, &connection_manager) + .await + { + log_warn!(client.logger, "Failed refilling LSPS2 payment lease: {}", error); + } + }); + } + + fn schedule_variable_lease_refill( + self: &Arc, max_proportional_lsp_fee_limit_ppm_msat: Option, + connection_manager: &Arc>, + ) { + let client = Arc::clone(self); + let connection_manager = Arc::clone(connection_manager); + tokio::spawn(async move { + if let Err(error) = client + .cache_variable_lease(max_proportional_lsp_fee_limit_ppm_msat, &connection_manager) + .await + { + log_warn!(client.logger, "Failed refilling LSPS2 payment lease: {}", error); + } + }); + } + + async fn cache_fixed_lease( + self: &Arc, amount_msat: u64, max_total_lsp_fee_limit_msat: Option, + connection_manager: &Arc>, + ) -> Result<(), Error> { + let request_lock = self + .pending_lease_request_state + .lock() + .expect("lock") + .request_lock(LeaseRequestKey::Fixed(amount_msat)); + let _request_guard = request_lock.lock().await; + if self + .lease_state + .lock() + .expect("lock") + .has_fixed_amount(amount_msat, max_total_lsp_fee_limit_msat) + { + return Ok(()); + } + self.negotiate_fixed_lease(amount_msat, max_total_lsp_fee_limit_msat, connection_manager) + .await?; + Ok(()) + } + + async fn cache_variable_lease( + self: &Arc, max_proportional_lsp_fee_limit_ppm_msat: Option, + connection_manager: &Arc>, + ) -> Result<(), Error> { + let request_lock = self + .pending_lease_request_state + .lock() + .expect("lock") + .request_lock(LeaseRequestKey::Variable); + let _request_guard = request_lock.lock().await; + if self + .lease_state + .lock() + .expect("lock") + .has_variable_amount(max_proportional_lsp_fee_limit_ppm_msat) + { + return Ok(()); + } + self.negotiate_variable_lease(max_proportional_lsp_fee_limit_ppm_msat, connection_manager) + .await?; + Ok(()) + } + + async fn gather_lsps2_offers( + self: &Arc, connection_manager: &Arc>, + ) -> Result, Error> { + let lsps2_nodes = self.get_lsps2_nodes().await?; + + // Connect to all candidate LSPs in parallel. + let mut connect_set = JoinSet::new(); + for lsp_node in &lsps2_nodes { + let cm = Arc::clone(connection_manager); + let node_id = lsp_node.node_id; + let addr = lsp_node.address.clone(); + let logger = self.logger.clone(); + connect_set.spawn(async move { + if let Err(e) = cm.connect_peer_if_necessary(node_id, addr).await { + log_warn!(logger, "Failed to connect to LSP {} for fee query: {}", node_id, e); + } + }); + } + while connect_set.join_next().await.is_some() {} + + let mut all_offers: Vec<(LspConfig, LSPS2FeeResponse)> = + Vec::with_capacity(lsps2_nodes.len()); + let mut fee_set: JoinSet<(LspConfig, Result)> = JoinSet::new(); + for lsp_node in &lsps2_nodes { + let lsp = lsp_node.clone(); + let client = Arc::clone(self); + fee_set.spawn(async move { + let res = client.lsps2_request_opening_fee_params(Some(&lsp.node_id)).await; + (lsp, res) + }); + } + while let Some(join_result) = fee_set.join_next().await { + match join_result { + Ok((lsp, Ok(fees))) => all_offers.push((lsp, fees)), + Ok((lsp, Err(e))) => { + log_warn!(self.logger, "Failed to get fees from LSP {}: {}", lsp.node_id, e) + }, + Err(e) => { + log_warn!(self.logger, "Failed to get fees from LSP: {}", e) + }, + } + } + + Ok(all_offers) + } +} + +impl LSPS2Client +where + L::Target: LdkLogger, +{ + async fn lsps2_request_opening_fee_params( + &self, node_id: Option<&PublicKey>, + ) -> Result { + let lsps2_node = select_lsps_for_protocol(&self.lsp_nodes, 2, node_id) + .ok_or(Error::LiquiditySourceUnavailable)?; + + let client_handler = self.liquidity_manager.lsps2_client_handler().ok_or_else(|| { + log_error!(self.logger, "Liquidity client was not configured.",); + Error::LiquiditySourceUnavailable + })?; + + let (fee_request_sender, fee_request_receiver) = oneshot::channel(); + { + let mut pending_fee_requests_lock = + self.pending_lsps2_fee_requests.lock().expect("lock"); + let request_id = + client_handler.request_opening_params(lsps2_node.node_id, lsps2_node.token.clone()); + pending_fee_requests_lock.insert(request_id, fee_request_sender); + } + + tokio::time::timeout( + Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), + fee_request_receiver, + ) + .await + .map_err(|e| { + log_error!(self.logger, "Liquidity request timed out: {}", e); + Error::LiquidityRequestFailed + })? + .map_err(|e| { + log_error!(self.logger, "Failed to handle response from liquidity service: {}", e); + Error::LiquidityRequestFailed + }) + } + + async fn lsps2_send_buy_request( + &self, amount_msat: Option, opening_fee_params: LSPS2OpeningFeeParams, + node_id: Option<&PublicKey>, + ) -> Result { + let lsps2_node = select_lsps_for_protocol(&self.lsp_nodes, 2, node_id) + .ok_or(Error::LiquiditySourceUnavailable)?; + + let client_handler = self.liquidity_manager.lsps2_client_handler().ok_or_else(|| { + log_error!(self.logger, "Liquidity client was not configured.",); + Error::LiquiditySourceUnavailable + })?; + + let (buy_request_sender, buy_request_receiver) = oneshot::channel(); + { + let mut pending_buy_requests_lock = self.pending_buy_requests.lock().expect("lock"); + let request_id = client_handler + .select_opening_params(lsps2_node.node_id, amount_msat, opening_fee_params.clone()) + .map_err(|e| { + log_error!( + self.logger, + "Failed to send buy request to liquidity service: {:?}", + e + ); + Error::LiquidityRequestFailed + })?; + pending_buy_requests_lock.insert(request_id, buy_request_sender); + } + + let buy_response = tokio::time::timeout( + Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), + buy_request_receiver, + ) + .await + .map_err(|e| { + log_error!(self.logger, "Liquidity request timed out: {}", e); + Error::LiquidityRequestFailed + })? + .map_err(|e| { + log_error!(self.logger, "Failed to handle response from liquidity service: {:?}", e); + Error::LiquidityRequestFailed + })?; + + let valid_until = opening_fee_params + .valid_until + .0 + .timestamp() + .try_into() + .map_err(|_| Error::LiquidityRequestFailed)?; + let lease = PaymentLease { + id: PaymentLeaseId { + lsp_node_id: lsps2_node.node_id, + intercept_scid: buy_response.intercept_scid, + }, + params: opening_fee_params, + cltv_expiry_delta: buy_response.cltv_expiry_delta, + payment_size_msat: amount_msat, + valid_until, + }; + self.lease_store.insert(lease.clone()).await?; + self.lease_state.lock().expect("lock").insert(lease.clone()); + Ok(lease) + } + + async fn consume_lease(&self, id: &PaymentLeaseId) -> Result { + let lease = self + .lease_state + .lock() + .expect("lock") + .valid(id) + .ok_or(Error::LiquidityRequestFailed)?; + self.consume_selected_lease(lease).await + } + + async fn consume_selected_lease(&self, lease: PaymentLease) -> Result { + let lease_store = Arc::clone(&self.lease_store); + consume_after_persisted_removal( + lease, + move |lease| async move { lease_store.remove(&lease.id).await }, + |lease| { + self.lease_state.lock().expect("lock").remove(&lease.id); + }, + ) + .await + } + + async fn take_cached_fixed_lease( + &self, amount_msat: u64, max_fee_msat: Option, + ) -> Result, Error> { + loop { + let Some((lease, fee_msat)) = + self.lease_state.lock().expect("lock").fixed_amount(amount_msat, max_fee_msat) + else { + return Ok(None); + }; + let lease = self.consume_selected_lease(lease).await?; + if let Some(lsp) = + select_lsps_for_protocol(&self.lsp_nodes, 2, Some(&lease.id.lsp_node_id)) + { + return Ok(Some((lease, fee_msat, lsp))); + } + } + } + + async fn take_cached_variable_lease( + &self, max_proportional_fee_ppm_msat: Option, + ) -> Result, Error> { + loop { + let Some((lease, proportional_fee)) = self + .lease_state + .lock() + .expect("lock") + .variable_amount(max_proportional_fee_ppm_msat) + else { + return Ok(None); + }; + let lease = self.consume_selected_lease(lease).await?; + if let Some(lsp) = + select_lsps_for_protocol(&self.lsp_nodes, 2, Some(&lease.id.lsp_node_id)) + { + return Ok(Some((lease, proportional_fee, lsp))); + } + } + } + + fn lsps2_create_jit_invoice( + &self, buy_response: LSPS2BuyResponse, amount_msat: Option, + description: &Bolt11InvoiceDescription, expiry_secs: u32, + payment_hash: Option, lsps2_parameters: LSPS2Parameters, + node_id: Option<&PublicKey>, + ) -> Result { + let lsps2_node = select_lsps_for_protocol(&self.lsp_nodes, 2, node_id) + .ok_or(Error::LiquiditySourceUnavailable)?; + + // LSPS2 requires min_final_cltv_expiry_delta to be at least 2 more than usual. + let min_final_cltv_expiry_delta = MIN_FINAL_CLTV_EXPIRY_DELTA + 2; + let encoded_payment_metadata = PaymentMetadata { + lsps2_parameters: Some(lsps2_parameters), + lsps2_lease_parameters: None, + } + .encode(); + let (payment_hash, payment_secret, payment_metadata) = match payment_hash { + Some(payment_hash) => { + let (payment_secret, payment_metadata) = self + .channel_manager + .create_inbound_payment_for_hash( + payment_hash, + None, + expiry_secs, + Some(min_final_cltv_expiry_delta), + Some(encoded_payment_metadata), + ) + .map_err(|e| { + log_error!(self.logger, "Failed to register inbound payment: {:?}", e); + Error::InvoiceCreationFailed + })?; + (payment_hash, payment_secret, payment_metadata) + }, + None => self + .channel_manager + .create_inbound_payment( + None, + expiry_secs, + Some(min_final_cltv_expiry_delta), + Some(encoded_payment_metadata), + ) + .map_err(|e| { + log_error!(self.logger, "Failed to register inbound payment: {:?}", e); + Error::InvoiceCreationFailed + })?, + }; + + let route_hint = RouteHint(vec![RouteHintHop { + src_node_id: lsps2_node.node_id, + short_channel_id: buy_response.intercept_scid, + fees: RoutingFees { base_msat: 0, proportional_millionths: 0 }, + cltv_expiry_delta: buy_response.cltv_expiry_delta as u16, + htlc_minimum_msat: None, + htlc_maximum_msat: None, + }]); + + let currency = self.config.network.into(); + let mut invoice_builder = InvoiceBuilder::new(currency) + .invoice_description(description.clone()) + .payment_hash(payment_hash) + .payment_secret(payment_secret) + .current_timestamp() + .min_final_cltv_expiry_delta(min_final_cltv_expiry_delta.into()) + .expiry_time(Duration::from_secs(expiry_secs.into())) + .private_route(route_hint); + + if let Some(amount_msat) = amount_msat { + invoice_builder = invoice_builder.amount_milli_satoshis(amount_msat).basic_mpp(); + } + + let invoice = if let Some(payment_metadata) = payment_metadata { + invoice_builder.payment_metadata(payment_metadata).build_signed(|hash| { + Secp256k1::new() + .sign_ecdsa_recoverable(hash, &self.keys_manager.get_node_secret_key()) + }) + } else { + invoice_builder.build_signed(|hash| { + Secp256k1::new() + .sign_ecdsa_recoverable(hash, &self.keys_manager.get_node_secret_key()) + }) + }; + invoice.map_err(|e| { + log_error!(self.logger, "Failed to build and sign invoice: {}", e); + Error::InvoiceCreationFailed + }) + } + + pub(crate) async fn handle_event(&self, event: LSPS2ClientEvent) { + match event { + LSPS2ClientEvent::OpeningParametersReady { + request_id, + counterparty_node_id, + opening_fee_params_menu, + } => { + if self + .lsp_nodes + .read() + .expect("lock") + .iter() + .any(|n| n.node_id == counterparty_node_id) + { + if let Some(sender) = + self.pending_lsps2_fee_requests.lock().expect("lock").remove(&request_id) + { + let response = LSPS2FeeResponse { opening_fee_params_menu }; + + match sender.send(response) { + Ok(()) => (), + Err(_) => { + log_error!( + self.logger, + "Failed to handle response for request {:?} from liquidity service", + request_id + ); + }, + } + } else { + debug_assert!( + false, + "Received response from liquidity service for unknown request." + ); + log_error!( + self.logger, + "Received response from liquidity service for unknown request." + ); + } + } else { + log_error!( + self.logger, + "Received unexpected LSPS2Client::OpeningParametersReady event!" + ); + } + }, + LSPS2ClientEvent::InvoiceParametersReady { + request_id, + counterparty_node_id, + intercept_scid, + cltv_expiry_delta, + .. + } => { + if self + .lsp_nodes + .read() + .expect("lock") + .iter() + .any(|n| n.node_id == counterparty_node_id) + { + if let Some(sender) = + self.pending_buy_requests.lock().expect("lock").remove(&request_id) + { + let response = LSPS2BuyResponse { intercept_scid, cltv_expiry_delta }; + + match sender.send(response) { + Ok(()) => (), + Err(_) => { + log_error!( + self.logger, + "Failed to handle response for request {:?} from liquidity service", + request_id + ); + }, + } + } else { + debug_assert!( + false, + "Received response from liquidity service for unknown request." + ); + log_error!( + self.logger, + "Received response from liquidity service for unknown request." + ); + } + } else { + log_error!( + self.logger, + "Received unexpected LSPS2Client::InvoiceParametersReady event!" + ); + } + }, + _ => { + log_error!(self.logger, "Received unexpected LSPS2Client liquidity event!"); + }, + } + } + + async fn get_lsps2_nodes(&self) -> Result, Error> { + let has_undiscovered_protocol = + self.lsp_nodes.read().expect("lock").iter().any(|n| n.supported_protocols.is_none()); + + if has_undiscovered_protocol { + // LSP protocol discovery is still in flight, we wait briefly for it to finish, then re-check. + let mut rx = self.discovery_done_rx.clone(); + if !*rx.borrow() { + log_debug!( + self.logger, + "Waiting for LSP protocol discovery to complete before selecting LSPS2 nodes." + ); + let _ = tokio::time::timeout( + Duration::from_secs(LSPS_DISCOVERY_WAIT_TIMEOUT_SECS), + rx.wait_for(|done| *done), + ) + .await; + } + } + + let lsps2_nodes = select_all_lsps_for_protocol(&self.lsp_nodes, 2); + if lsps2_nodes.is_empty() { + log_error!(self.logger, "No LSPs available for LSPS2 protocol."); + return Err(Error::LiquiditySourceUnavailable); + }; + Ok(lsps2_nodes) + } +} + +#[derive(Debug, Clone)] +pub(crate) struct LSPS2FeeResponse { + opening_fee_params_menu: Vec, +} + +#[derive(Debug, Clone)] +pub(crate) struct LSPS2BuyResponse { + intercept_scid: u64, + cltv_expiry_delta: u32, +} + +impl From<&PaymentLease> for LSPS2BuyResponse { + fn from(lease: &PaymentLease) -> Self { + Self { intercept_scid: lease.id.intercept_scid, cltv_expiry_delta: lease.cltv_expiry_delta } + } +} +#[cfg(test)] +mod tests { + use std::num::NonZeroU64; + + use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey}; + use lightning::ln::channelmanager::PaymentId; + use lightning::ln::inbound_payment::ExpandedKey; + use lightning::offers::nonce::Nonce; + use lightning::offers::offer::{OfferBuilder, Quantity}; + use lightning::sign::EntropySource; + + use super::*; + use crate::liquidity::client::lsps2::state::PendingOfferAmount; + + struct FixedEntropy; + + impl EntropySource for FixedEntropy { + fn get_secure_random_bytes(&self) -> [u8; 32] { + [42; 32] + } + } + + fn recipient_pubkey() -> PublicKey { + PublicKey::from_secret_key(&Secp256k1::new(), &SecretKey::from_slice(&[43; 32]).unwrap()) + } + + #[test] + fn fixed_offer_request_uses_quantity_resolved_amount() { + let expanded_key = ExpandedKey::new([44; 32]); + let nonce = Nonce::from_entropy_source(&FixedEntropy); + let secp_ctx = Secp256k1::new(); + let offer = OfferBuilder::new(recipient_pubkey()) + .amount_msats(1_000) + .supported_quantity(Quantity::Bounded(NonZeroU64::new(10).unwrap())) + .build() + .unwrap(); + let invoice_request = offer + .request_invoice(&expanded_key, nonce, &secp_ctx, PaymentId([45; 32])) + .unwrap() + .quantity(3) + .unwrap() + .build_and_sign() + .unwrap(); + let pending_offer = PendingOffer::new( + offer.id(), + PendingOfferAmount::Fixed { amount_msat: 1_000, max_total_fee_msat: Some(250) }, + None, + ); + + assert_eq!( + jit_invoice_request(&pending_offer, &invoice_request).unwrap(), + JitInvoiceRequest::Fixed { amount_msat: 3_000, max_total_fee_msat: Some(250) } + ); + } + + #[test] + fn variable_offer_request_disables_mpp() { + let expanded_key = ExpandedKey::new([46; 32]); + let nonce = Nonce::from_entropy_source(&FixedEntropy); + let secp_ctx = Secp256k1::new(); + let offer = OfferBuilder::new(recipient_pubkey()).build().unwrap(); + let invoice_request = offer + .request_invoice(&expanded_key, nonce, &secp_ctx, PaymentId([47; 32])) + .unwrap() + .amount_msats(2_500) + .unwrap() + .build_and_sign() + .unwrap(); + let pending_offer = PendingOffer::new( + offer.id(), + PendingOfferAmount::Variable { max_proportional_fee_ppm_msat: Some(10_000) }, + None, + ); + + assert_eq!( + jit_invoice_request(&pending_offer, &invoice_request).unwrap(), + JitInvoiceRequest::Variable { + amount_msat: 2_500, + max_proportional_fee_ppm_msat: Some(10_000), + } + ); + assert!(!jit_invoice_request(&pending_offer, &invoice_request).unwrap().allow_mpp()); + } + + #[tokio::test] + async fn failed_persisted_removal_does_not_consume() { + let consumed = Arc::new(Mutex::new(false)); + let consumed_ref = Arc::clone(&consumed); + let result = consume_after_persisted_removal( + 42, + |_| async { Err(()) }, + move |_| *consumed_ref.lock().unwrap() = true, + ) + .await; + + assert_eq!(result, Err(())); + assert!(!*consumed.lock().unwrap()); + } + + #[test] + fn lease_negotiation_retries_only_transient_failures() { + assert!(should_retry_lease_negotiation(Error::LiquidityRequestFailed, 1)); + assert!(should_retry_lease_negotiation(Error::LiquidityRequestFailed, 2)); + assert!(!should_retry_lease_negotiation(Error::LiquidityRequestFailed, 3)); + assert!(!should_retry_lease_negotiation(Error::LiquidityFeeTooHigh, 1)); + assert!(!should_retry_lease_negotiation(Error::LiquiditySourceUnavailable, 1)); + } + + #[tokio::test] + async fn lease_negotiation_fails_over_between_lsps() { + let candidates = vec![(1, 10), (1, 20), (2, 30)]; + let attempted_lsps = Arc::new(Mutex::new(Vec::new())); + let attempted_lsps_ref = Arc::clone(&attempted_lsps); + let result = try_lease_candidates( + candidates, + |candidate| candidate.0, + move |candidate| { + let attempted_lsps = Arc::clone(&attempted_lsps_ref); + async move { + attempted_lsps.lock().unwrap().push(candidate.0); + if candidate.0 == 1 { + Err(()) + } else { + Ok(candidate.1) + } + } + }, + ) + .await; + + assert_eq!(result, Ok(30)); + assert_eq!(*attempted_lsps.lock().unwrap(), vec![1, 2]); + } +} + +pub(crate) mod state; diff --git a/src/liquidity/client/lsps2/state.rs b/src/liquidity/client/lsps2/state.rs new file mode 100644 index 0000000000..28bbd6dd57 --- /dev/null +++ b/src/liquidity/client/lsps2/state.rs @@ -0,0 +1,683 @@ +use std::collections::HashMap; +use std::sync::{Arc, Weak}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use bitcoin::secp256k1::PublicKey; +use lightning::offers::offer::OfferId; +use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum}; +use lightning_liquidity::lsps2::msgs::LSPS2OpeningFeeParams; +use lightning_liquidity::lsps2::utils::compute_opening_fee; +use tokio::sync::Mutex as AsyncMutex; + +use crate::data_store::{DataStore, StorableObject, StorableObjectId, StorableObjectUpdate}; +use crate::hex_utils; + +pub(crate) const MIN_LEASE_REMAINING_SECS: u64 = 24 * 60 * 60; + +pub(crate) type PaymentLeaseStore = DataStore; +pub(crate) type PendingOfferStore = DataStore; + +#[derive(Clone, Copy, Debug, Hash, PartialEq, Eq, PartialOrd, Ord)] +pub(crate) struct PendingOfferId { + offer_id: [u8; 32], +} + +impl From for PendingOfferId { + fn from(offer_id: OfferId) -> Self { + Self { offer_id: offer_id.0 } + } +} + +impl From for OfferId { + fn from(offer_id: PendingOfferId) -> Self { + Self(offer_id.offer_id) + } +} + +impl_writeable_tlv_based!(PendingOfferId, { + (0, offer_id, required), +}); + +impl StorableObjectId for PendingOfferId { + fn encode_to_hex_str(&self) -> String { + hex_utils::to_string(&self.offer_id) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum PendingOfferAmount { + Fixed { amount_msat: u64, max_total_fee_msat: Option }, + Variable { max_proportional_fee_ppm_msat: Option }, +} + +impl_writeable_tlv_based_enum!(PendingOfferAmount, + (0, Fixed) => { + (0, amount_msat, required), + (2, max_total_fee_msat, option), + }, + (2, Variable) => { + (0, max_proportional_fee_ppm_msat, option), + }, +); + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct PendingOffer { + pub(crate) id: PendingOfferId, + pub(crate) amount: PendingOfferAmount, + pub(crate) absolute_expiry: Option, + last_accessed: u64, +} + +impl PendingOffer { + pub(crate) fn new( + offer_id: OfferId, amount: PendingOfferAmount, absolute_expiry: Option, + ) -> Self { + Self { id: offer_id.into(), amount, absolute_expiry, last_accessed: now_secs() } + } +} + +impl_writeable_tlv_based!(PendingOffer, { + (0, id, required), + (2, amount, required), + (4, absolute_expiry, option), + (6, last_accessed, required), +}); + +#[derive(Clone, Debug)] +pub(crate) struct PendingOfferUpdate(PendingOffer); + +impl StorableObjectUpdate for PendingOfferUpdate { + fn id(&self) -> PendingOfferId { + self.0.id + } +} + +impl StorableObject for PendingOffer { + type Id = PendingOfferId; + type Update = PendingOfferUpdate; + + fn id(&self) -> Self::Id { + self.id + } + + fn update(&mut self, update: Self::Update) -> bool { + if *self == update.0 { + false + } else { + *self = update.0; + true + } + } + + fn to_update(&self) -> Self::Update { + PendingOfferUpdate(self.clone()) + } +} + +pub(crate) struct PendingOfferState { + offers: HashMap, + capacity: usize, +} + +impl PendingOfferState { + pub(crate) fn from_offers( + offers: Vec, capacity: usize, now: u64, + ) -> (Self, Vec) { + let mut removed = Vec::new(); + let mut offers = offers + .into_iter() + .filter_map(|offer| { + if is_pending_offer_expired(&offer, now) { + removed.push(offer.id); + None + } else { + Some((offer.id, offer)) + } + }) + .collect::>(); + let capacity = capacity.max(1); + while offers.len() > capacity { + let id = least_recently_used(&offers).expect("offers are non-empty"); + offers.remove(&id); + removed.push(id); + } + (Self { offers, capacity }, removed) + } + + pub(crate) fn register(&mut self, offer: PendingOffer) -> Option { + let is_new = !self.offers.contains_key(&offer.id); + let evicted = if is_new && self.offers.len() == self.capacity { + let id = least_recently_used(&self.offers).expect("offers are non-empty"); + self.offers.remove(&id); + Some(id) + } else { + None + }; + self.offers.insert(offer.id, offer); + evicted + } + + pub(crate) fn get(&mut self, id: &PendingOfferId, now: u64) -> Option { + if self.offers.get(id).is_some_and(|offer| is_pending_offer_expired(offer, now)) { + self.offers.remove(id); + return None; + } + let last_accessed = self.next_access_timestamp(now); + let offer = self.offers.get_mut(id)?; + offer.last_accessed = last_accessed; + Some(offer.clone()) + } + + pub(crate) fn contains(&self, id: &PendingOfferId) -> bool { + self.offers.contains_key(id) + } + + pub(crate) fn offers(&self) -> Vec { + self.offers.values().cloned().collect() + } + + pub(crate) fn prune(&mut self, now: u64) -> Vec { + let removed = self + .offers + .iter() + .filter(|(_, offer)| is_pending_offer_expired(offer, now)) + .map(|(id, _)| *id) + .collect::>(); + for id in &removed { + self.offers.remove(id); + } + removed + } + + fn next_access_timestamp(&self, now: u64) -> u64 { + self.offers + .values() + .map(|offer| offer.last_accessed) + .max() + .map_or(now, |last_accessed| now.max(last_accessed.saturating_add(1))) + } +} + +#[derive(Default)] +pub(crate) struct PendingInvoiceRequestState { + locks: HashMap>>, +} + +impl PendingInvoiceRequestState { + pub(crate) fn request_lock(&mut self, id: PendingOfferId) -> Arc> { + self.prune(); + if let Some(lock) = self.locks.get(&id).and_then(Weak::upgrade) { + return lock; + } + let lock = Arc::new(AsyncMutex::new(())); + self.locks.insert(id, Arc::downgrade(&lock)); + lock + } + + pub(crate) fn prune(&mut self) { + self.locks.retain(|_, lock| lock.strong_count() > 0); + } + + #[cfg(test)] + fn is_empty(&self) -> bool { + self.locks.is_empty() + } +} + +#[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)] +pub(crate) enum LeaseRequestKey { + Fixed(u64), + Variable, +} + +#[derive(Default)] +pub(crate) struct PendingLeaseRequestState { + locks: HashMap>>, +} + +impl PendingLeaseRequestState { + pub(crate) fn request_lock(&mut self, key: LeaseRequestKey) -> Arc> { + self.prune(); + if let Some(lock) = self.locks.get(&key).and_then(Weak::upgrade) { + return lock; + } + let lock = Arc::new(AsyncMutex::new(())); + self.locks.insert(key, Arc::downgrade(&lock)); + lock + } + + pub(crate) fn prune(&mut self) { + self.locks.retain(|_, lock| lock.strong_count() > 0); + } +} + +fn least_recently_used(offers: &HashMap) -> Option { + offers.values().min_by_key(|offer| (offer.last_accessed, offer.id)).map(|offer| offer.id) +} + +fn is_pending_offer_expired(offer: &PendingOffer, now: u64) -> bool { + offer.absolute_expiry.is_some_and(|expiry| expiry <= now) +} + +#[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)] +pub(crate) struct PaymentLeaseId { + pub(crate) lsp_node_id: PublicKey, + pub(crate) intercept_scid: u64, +} + +impl_writeable_tlv_based!(PaymentLeaseId, { + (0, lsp_node_id, required), + (2, intercept_scid, required), +}); + +impl StorableObjectId for PaymentLeaseId { + fn encode_to_hex_str(&self) -> String { + format!( + "{}{:016x}", + hex_utils::to_string(&self.lsp_node_id.serialize()), + self.intercept_scid + ) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct PaymentLease { + pub(crate) id: PaymentLeaseId, + pub(crate) params: LSPS2OpeningFeeParams, + pub(crate) cltv_expiry_delta: u32, + pub(crate) payment_size_msat: Option, + pub(crate) valid_until: u64, +} + +impl_writeable_tlv_based!(PaymentLease, { + (0, id, required), + (2, params, required), + (4, valid_until, required), + (6, cltv_expiry_delta, required), + (8, payment_size_msat, option), +}); + +#[derive(Clone, Debug)] +pub(crate) struct PaymentLeaseUpdate(PaymentLease); + +impl StorableObjectUpdate for PaymentLeaseUpdate { + fn id(&self) -> PaymentLeaseId { + self.0.id + } +} + +impl StorableObject for PaymentLease { + type Id = PaymentLeaseId; + type Update = PaymentLeaseUpdate; + + fn id(&self) -> Self::Id { + self.id + } + + fn update(&mut self, update: Self::Update) -> bool { + if *self == update.0 { + false + } else { + *self = update.0; + true + } + } + + fn to_update(&self) -> Self::Update { + PaymentLeaseUpdate(self.clone()) + } +} + +#[derive(Default)] +pub(crate) struct LSPS2LeaseState { + leases: HashMap, +} + +impl LSPS2LeaseState { + pub(crate) fn from_leases(leases: Vec) -> Self { + Self { leases: leases.into_iter().map(|lease| (lease.id, lease)).collect() } + } + + pub(crate) fn insert(&mut self, lease: PaymentLease) { + self.leases.insert(lease.id, lease); + } + + pub(crate) fn valid(&self, id: &PaymentLeaseId) -> Option { + self.leases.get(id).filter(|lease| is_lease_usable(lease)).cloned() + } + + pub(crate) fn remove(&mut self, id: &PaymentLeaseId) -> Option { + self.leases.remove(id) + } + + pub(crate) fn fixed_amount( + &self, amount_msat: u64, max_fee_msat: Option, + ) -> Option<(PaymentLease, u64)> { + let (id, fee_msat) = self + .leases + .iter() + .filter(|(_, lease)| lease.payment_size_msat == Some(amount_msat)) + .filter(|(_, lease)| is_lease_usable(lease)) + .filter_map(|(id, lease)| { + compute_opening_fee( + amount_msat, + lease.params.min_fee_msat, + lease.params.proportional as u64, + ) + .map(|fee_msat| (*id, fee_msat)) + }) + .filter(|(_, fee_msat)| max_fee_msat.map_or(true, |max| *fee_msat <= max)) + .min_by_key(|(_, fee_msat)| *fee_msat)?; + self.leases.get(&id).cloned().map(|lease| (lease, fee_msat)) + } + + pub(crate) fn has_fixed_amount(&self, amount_msat: u64, max_fee_msat: Option) -> bool { + self.leases + .values() + .filter(|lease| lease.payment_size_msat == Some(amount_msat)) + .filter(|lease| is_lease_usable(lease)) + .filter_map(|lease| { + compute_opening_fee( + amount_msat, + lease.params.min_fee_msat, + lease.params.proportional as u64, + ) + }) + .any(|fee_msat| max_fee_msat.map_or(true, |max| fee_msat <= max)) + } + + pub(crate) fn variable_amount( + &self, max_proportional_fee_ppm_msat: Option, + ) -> Option<(PaymentLease, u64)> { + let (id, proportional_fee) = self + .leases + .iter() + .filter(|(_, lease)| lease.payment_size_msat.is_none()) + .filter(|(_, lease)| is_lease_usable(lease)) + .map(|(id, lease)| (*id, lease.params.proportional as u64)) + .filter(|(_, fee)| max_proportional_fee_ppm_msat.map_or(true, |max| *fee <= max)) + .min_by_key(|(_, fee)| *fee)?; + self.leases.get(&id).cloned().map(|lease| (lease, proportional_fee)) + } + + pub(crate) fn has_variable_amount(&self, max_proportional_fee_ppm_msat: Option) -> bool { + self.leases + .values() + .filter(|lease| lease.payment_size_msat.is_none()) + .filter(|lease| is_lease_usable(lease)) + .map(|lease| lease.params.proportional as u64) + .any(|fee| max_proportional_fee_ppm_msat.map_or(true, |max| fee <= max)) + } + + pub(crate) fn prune(&mut self) { + self.leases.retain(|_, lease| is_lease_usable(lease)); + } +} + +pub(crate) fn is_lease_usable(lease: &PaymentLease) -> bool { + lease.valid_until.saturating_sub(now_secs()) >= MIN_LEASE_REMAINING_SECS +} + +pub(crate) fn now_secs() -> u64 { + SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or(Duration::ZERO).as_secs() +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use super::*; + use bitcoin::secp256k1::{Secp256k1, SecretKey}; + use lightning::offers::offer::OfferId; + use lightning::util::ser::{Readable, Writeable}; + + fn lease( + node_key_byte: u8, intercept_scid: u64, min_fee_msat: u64, payment_size_msat: Option, + valid_until: u64, + ) -> PaymentLease { + PaymentLease { + id: PaymentLeaseId { + lsp_node_id: PublicKey::from_secret_key( + &Secp256k1::new(), + &SecretKey::from_slice(&[node_key_byte; 32]).unwrap(), + ), + intercept_scid, + }, + params: LSPS2OpeningFeeParams { + min_fee_msat, + proportional: 1, + valid_until: "2030-01-01T00:00:00Z".parse().unwrap(), + min_lifetime: 1, + max_client_to_self_delay: 1, + min_payment_size_msat: 1, + max_payment_size_msat: 2_000, + promise: String::new(), + }, + cltv_expiry_delta: 18, + payment_size_msat, + valid_until, + } + } + + #[test] + fn leases_are_consumed_once() { + let lease = lease(2, 42, 1, Some(1_000), now_secs() + MIN_LEASE_REMAINING_SECS + 60); + let id = lease.id; + let mut state = LSPS2LeaseState::default(); + state.insert(lease); + assert!(state.valid(&id).is_some()); + assert!(state.remove(&id).is_some()); + assert!(state.valid(&id).is_none()); + } + + #[test] + fn payment_leases_roundtrip() { + let lease = lease(2, 48, 100, Some(1_000), now_secs() + MIN_LEASE_REMAINING_SECS + 60); + + let encoded = lease.encode(); + let decoded = PaymentLease::read(&mut &encoded[..]).unwrap(); + + assert_eq!(decoded, lease); + } + + #[test] + fn fixed_and_variable_lease_pools_are_isolated() { + let valid_until = now_secs() + MIN_LEASE_REMAINING_SECS + 60; + let fixed = lease(2, 49, 100, Some(1_000), valid_until); + let variable = lease(3, 50, 50, None, valid_until); + let mut state = LSPS2LeaseState::from_leases(vec![fixed.clone(), variable.clone()]); + + let (selected_fixed, _) = state.fixed_amount(1_000, None).unwrap(); + assert_eq!(selected_fixed.id, fixed.id); + state.remove(&selected_fixed.id); + assert!(state.fixed_amount(1_000, None).is_none()); + + let (selected_variable, _) = state.variable_amount(None).unwrap(); + assert_eq!(selected_variable.id, variable.id); + } + + #[test] + fn prunes_leases_close_to_expiry() { + let lease = lease(2, 43, 1, Some(1_000), now_secs() + MIN_LEASE_REMAINING_SECS - 1); + let id = lease.id; + let mut state = LSPS2LeaseState::default(); + state.insert(lease); + state.prune(); + assert!(state.valid(&id).is_none()); + } + + #[test] + fn selects_cheapest_matching_lease_across_lsps() { + let valid_until = now_secs() + MIN_LEASE_REMAINING_SECS + 60; + let expensive = lease(2, 44, 100, Some(1_000), valid_until); + let cheap = lease(3, 45, 50, Some(1_000), valid_until); + let mut state = LSPS2LeaseState::from_leases(vec![expensive.clone(), cheap.clone()]); + + let (selected, _) = state.fixed_amount(1_000, None).unwrap(); + assert_eq!(selected.id, cheap.id); + state.remove(&selected.id); + let (remaining, _) = state.fixed_amount(1_000, None).unwrap(); + assert_eq!(remaining.id, expensive.id); + } + + #[test] + fn detects_cached_leases_for_refill() { + let valid_until = now_secs() + MIN_LEASE_REMAINING_SECS + 60; + let fixed = lease(2, 46, 100, Some(1_000), valid_until); + let variable = lease(3, 47, 50, None, valid_until); + let state = LSPS2LeaseState::from_leases(vec![fixed, variable]); + + assert!(state.has_fixed_amount(1_000, Some(100))); + assert!(!state.has_fixed_amount(1_000, Some(99))); + assert!(!state.has_fixed_amount(2_000, None)); + assert!(state.has_variable_amount(Some(1))); + assert!(!state.has_variable_amount(Some(0))); + } + + fn pending_offer( + id_byte: u8, amount: PendingOfferAmount, absolute_expiry: Option, last_accessed: u64, + ) -> PendingOffer { + PendingOffer { + id: PendingOfferId::from(OfferId([id_byte; 32])), + amount, + absolute_expiry, + last_accessed, + } + } + + #[test] + fn pending_offers_roundtrip() { + let offer = pending_offer( + 1, + PendingOfferAmount::Fixed { amount_msat: 1_000, max_total_fee_msat: Some(100) }, + Some(42), + 7, + ); + + let encoded = offer.encode(); + let decoded = PendingOffer::read(&mut &encoded[..]).unwrap(); + assert_eq!(decoded, offer); + } + + #[test] + fn pending_offer_registry_evicts_least_recently_used() { + let first = pending_offer( + 1, + PendingOfferAmount::Fixed { amount_msat: 1_000, max_total_fee_msat: None }, + None, + 1, + ); + let second = pending_offer( + 2, + PendingOfferAmount::Variable { max_proportional_fee_ppm_msat: None }, + None, + 2, + ); + let third = pending_offer( + 3, + PendingOfferAmount::Fixed { amount_msat: 3_000, max_total_fee_msat: None }, + None, + 3, + ); + let first_id = first.id; + let second_id = second.id; + let third_id = third.id; + let (mut state, removed) = PendingOfferState::from_offers(vec![first, second], 2, 10); + assert!(removed.is_empty()); + + assert!(state.get(&first_id, 11).is_some()); + assert_eq!(state.register(third), Some(second_id)); + assert!(state.get(&first_id, 13).is_some()); + assert!(state.get(&second_id, 14).is_none()); + assert!(state.get(&third_id, 15).is_some()); + } + + #[test] + fn pending_offer_registry_prunes_only_expired_offers() { + let expired = pending_offer( + 1, + PendingOfferAmount::Fixed { amount_msat: 1_000, max_total_fee_msat: None }, + Some(9), + 1, + ); + let unexpired = pending_offer( + 2, + PendingOfferAmount::Fixed { amount_msat: 2_000, max_total_fee_msat: None }, + Some(11), + 2, + ); + let long_lived = pending_offer( + 3, + PendingOfferAmount::Variable { max_proportional_fee_ppm_msat: Some(100) }, + None, + 3, + ); + let expired_id = expired.id; + let unexpired_id = unexpired.id; + let long_lived_id = long_lived.id; + + let (state, removed) = + PendingOfferState::from_offers(vec![expired, unexpired, long_lived], 3, 10); + + assert_eq!(removed, vec![expired_id]); + assert!(state.contains(&unexpired_id)); + assert!(state.contains(&long_lived_id)); + } + + #[test] + fn pending_offer_registry_lists_cached_offers() { + let fixed = pending_offer( + 1, + PendingOfferAmount::Fixed { amount_msat: 1_000, max_total_fee_msat: None }, + None, + 1, + ); + let variable = pending_offer( + 2, + PendingOfferAmount::Variable { max_proportional_fee_ppm_msat: None }, + None, + 2, + ); + let (state, removed) = + PendingOfferState::from_offers(vec![fixed.clone(), variable.clone()], 2, 10); + assert!(removed.is_empty()); + + let mut ids = state.offers().into_iter().map(|offer| offer.id).collect::>(); + ids.sort(); + assert_eq!(ids, vec![fixed.id, variable.id]); + } + + #[test] + fn invoice_requests_are_serialized_per_offer() { + let first_id = PendingOfferId::from(OfferId([1; 32])); + let second_id = PendingOfferId::from(OfferId([2; 32])); + let mut state = PendingInvoiceRequestState::default(); + + let first = state.request_lock(first_id); + let same_offer = state.request_lock(first_id); + let other_offer = state.request_lock(second_id); + + assert!(Arc::ptr_eq(&first, &same_offer)); + assert!(!Arc::ptr_eq(&first, &other_offer)); + + drop(first); + drop(same_offer); + drop(other_offer); + state.prune(); + assert!(state.is_empty()); + } + + #[test] + fn lease_requests_are_serialized_per_key() { + let mut state = PendingLeaseRequestState::default(); + + let fixed = state.request_lock(LeaseRequestKey::Fixed(1_000)); + let same_fixed = state.request_lock(LeaseRequestKey::Fixed(1_000)); + let other_fixed = state.request_lock(LeaseRequestKey::Fixed(2_000)); + let variable = state.request_lock(LeaseRequestKey::Variable); + + assert!(Arc::ptr_eq(&fixed, &same_fixed)); + assert!(!Arc::ptr_eq(&fixed, &other_fixed)); + assert!(!Arc::ptr_eq(&fixed, &variable)); + } +} diff --git a/src/liquidity/mod.rs b/src/liquidity/mod.rs index 87a0650c83..811d56d2fe 100644 --- a/src/liquidity/mod.rs +++ b/src/liquidity/mod.rs @@ -31,7 +31,17 @@ use tokio::sync::oneshot; use crate::builder::BuildError; use crate::connection::ConnectionManager; +use crate::io::utils::read_all_objects; +use crate::io::{ + LSPS2_LEASE_PERSISTENCE_PRIMARY_NAMESPACE, LSPS2_LEASE_PERSISTENCE_SECONDARY_NAMESPACE, + LSPS2_PENDING_OFFER_PERSISTENCE_PRIMARY_NAMESPACE, + LSPS2_PENDING_OFFER_PERSISTENCE_SECONDARY_NAMESPACE, +}; use crate::liquidity::client::lsps1::LSPS1Client; +use crate::liquidity::client::lsps2::state::{ + is_lease_usable, now_secs, LSPS2LeaseState, PaymentLeaseStore, PendingInvoiceRequestState, + PendingLeaseRequestState, PendingOfferState, PendingOfferStore, +}; use crate::liquidity::client::lsps2::LSPS2Client; use crate::liquidity::service::lsps2::{LSPS2Service, LSPS2ServiceLiquiditySource}; use crate::logger::{log_debug, log_error, log_info, LdkLogger, Logger}; @@ -247,6 +257,49 @@ where } pub(crate) async fn build(self) -> Result, BuildError> { + let (leases, pending_offers) = tokio::join!( + read_all_objects( + &*self.kv_store, + LSPS2_LEASE_PERSISTENCE_PRIMARY_NAMESPACE, + LSPS2_LEASE_PERSISTENCE_SECONDARY_NAMESPACE, + self.logger.clone(), + ), + read_all_objects( + &*self.kv_store, + LSPS2_PENDING_OFFER_PERSISTENCE_PRIMARY_NAMESPACE, + LSPS2_PENDING_OFFER_PERSISTENCE_SECONDARY_NAMESPACE, + self.logger.clone(), + ), + ); + let leases = leases.map_err(|_| BuildError::ReadFailed)?; + let pending_offers = pending_offers.map_err(|_| BuildError::ReadFailed)?; + let lease_store = Arc::new(PaymentLeaseStore::new( + leases.clone(), + LSPS2_LEASE_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + LSPS2_LEASE_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&self.kv_store), + self.logger.clone(), + )); + for lease in leases.iter().filter(|lease| !is_lease_usable(lease)) { + lease_store.remove(&lease.id).await.map_err(|_| BuildError::WriteFailed)?; + } + let lease_state = + LSPS2LeaseState::from_leases(leases.into_iter().filter(is_lease_usable).collect()); + let pending_offer_store = Arc::new(PendingOfferStore::new( + pending_offers.clone(), + LSPS2_PENDING_OFFER_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + LSPS2_PENDING_OFFER_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&self.kv_store), + self.logger.clone(), + )); + let (pending_offer_state, removed_offer_ids) = PendingOfferState::from_offers( + pending_offers, + self.config.lsps2_pending_offer_cache_size, + now_secs(), + ); + for id in removed_offer_ids { + pending_offer_store.remove(&id).await.map_err(|_| BuildError::WriteFailed)?; + } let liquidity_service_config = self.lsps2_service.as_ref().map(|s| { let lsps2_service_config = Some(s.ldk_service_config.clone()); let lsps5_service_config = None; @@ -311,6 +364,12 @@ where lsp_nodes: Arc::clone(&lsp_nodes), pending_lsps2_fee_requests: Mutex::new(HashMap::new()), pending_buy_requests: Mutex::new(HashMap::new()), + lease_store, + lease_state: Mutex::new(lease_state), + pending_offer_store, + pending_offer_state: Mutex::new(pending_offer_state), + pending_invoice_request_state: Mutex::new(PendingInvoiceRequestState::default()), + pending_lease_request_state: Mutex::new(PendingLeaseRequestState::default()), channel_manager: self.channel_manager.clone(), keys_manager: self.keys_manager.clone(), discovery_done_rx: discovery_done_rx.clone(), diff --git a/src/liquidity/service/lsps2.rs b/src/liquidity/service/lsps2.rs index 946511c5d9..4732e6f92c 100644 --- a/src/liquidity/service/lsps2.rs +++ b/src/liquidity/service/lsps2.rs @@ -26,7 +26,7 @@ use crate::logger::{log_error, LdkLogger}; use crate::types::{ChannelManager, KeysManager, LiquidityManager, PeerManager, Wallet}; use crate::{total_anchor_channels_reserve_sats, Config}; -const LSPS2_GETINFO_REQUEST_EXPIRY: Duration = Duration::from_secs(60 * 60 * 24); +const LSPS2_GETINFO_REQUEST_EXPIRY: Duration = Duration::from_secs(60 * 60 * 48); const LSPS2_CHANNEL_CLTV_EXPIRY_DELTA: u32 = 72; pub(crate) struct LSPS2Service { diff --git a/src/payment/bolt11.rs b/src/payment/bolt11.rs index 4503dfa061..5eff643581 100644 --- a/src/payment/bolt11.rs +++ b/src/payment/bolt11.rs @@ -13,7 +13,6 @@ use std::sync::{Arc, RwLock}; use bitcoin::hashes::sha256::Hash as Sha256; use bitcoin::hashes::Hash; -use lightning::impl_writeable_tlv_based; use lightning::ln::channelmanager::{ Bolt11InvoiceParameters, OptionalBolt11PaymentParams, PaymentId, }; @@ -32,8 +31,7 @@ use crate::ffi::{maybe_deref, maybe_try_convert_enum, maybe_wrap}; use crate::liquidity::LiquiditySource; use crate::logger::{log_error, log_info, LdkLogger, Logger}; use crate::payment::store::{ - LSPS2Parameters, PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentKind, - PaymentStatus, + PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentKind, PaymentStatus, }; use crate::peer_store::{PeerInfo, PeerStore}; use crate::runtime::Runtime; @@ -49,16 +47,6 @@ type Bolt11InvoiceDescription = LdkBolt11InvoiceDescription; #[cfg(feature = "uniffi")] type Bolt11InvoiceDescription = crate::ffi::Bolt11InvoiceDescription; -/// Metadata carried in BOLT11 invoice `payment_metadata`. -#[derive(Clone, Debug, PartialEq, Eq)] -pub(crate) struct PaymentMetadata { - pub(crate) lsps2_parameters: Option, -} - -impl_writeable_tlv_based!(PaymentMetadata, { - (0, lsps2_parameters, option), -}); - /// A payment handler allowing to create and pay [BOLT 11] invoices. /// /// Should be retrieved by calling [`Node::bolt11_payment`]. @@ -233,37 +221,6 @@ impl Bolt11Payment { } } -#[cfg(test)] -mod tests { - use lightning::util::ser::{Readable, Writeable}; - - use super::*; - - #[test] - fn empty_metadata_roundtrips() { - let metadata = PaymentMetadata { lsps2_parameters: None }; - - let encoded = metadata.encode(); - let decoded = PaymentMetadata::read(&mut &*encoded).unwrap(); - - assert_eq!(metadata, decoded); - } - - #[test] - fn lsps2_parameters_roundtrip() { - let lsps2_parameters = LSPS2Parameters { - max_total_opening_fee_msat: Some(42_000), - max_proportional_opening_fee_ppm_msat: Some(17_000), - }; - let metadata = PaymentMetadata { lsps2_parameters: Some(lsps2_parameters) }; - - let encoded = metadata.encode(); - let decoded = PaymentMetadata::read(&mut &*encoded).unwrap(); - - assert_eq!(metadata, decoded); - } -} - impl Bolt11Payment { fn send_internal( &self, invoice: &LdkBolt11Invoice, amount_msat: Option, diff --git a/src/payment/bolt12.rs b/src/payment/bolt12.rs index d79aca6c24..c8aae11b90 100644 --- a/src/payment/bolt12.rs +++ b/src/payment/bolt12.rs @@ -18,6 +18,7 @@ use lightning::ln::channelmanager::{OptionalOfferPaymentParams, PaymentId}; use lightning::ln::outbound_payment::Retry; use lightning::offers::offer::{Amount, Offer as LdkOffer, OfferFromHrn, Quantity}; use lightning::offers::parse::Bolt12SemanticError; +use lightning::onion_message::messenger::NullMessageRouter; use lightning::routing::router::RouteParametersConfig; use lightning::sign::EntropySource; #[cfg(feature = "uniffi")] @@ -27,6 +28,8 @@ use lightning_types::string::UntrustedString; use crate::config::{AsyncPaymentsRole, Config, LDK_PAYMENT_RETRY_TIMEOUT}; use crate::error::Error; use crate::ffi::{maybe_deref, maybe_wrap}; +use crate::liquidity::client::lsps2::state::{PendingOffer, PendingOfferAmount}; +use crate::liquidity::LiquiditySource; use crate::logger::{log_error, log_info, LdkLogger, Logger}; use crate::payment::store::{PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus}; use crate::runtime::Runtime; @@ -62,6 +65,7 @@ type HumanReadableName = Arc; pub struct Bolt12Payment { runtime: Arc, channel_manager: Arc, + liquidity_source: Arc>>, keys_manager: Arc, payment_store: Arc, config: Arc, @@ -73,13 +77,14 @@ pub struct Bolt12Payment { impl Bolt12Payment { pub(crate) fn new( runtime: Arc, channel_manager: Arc, - keys_manager: Arc, payment_store: Arc, config: Arc, - is_running: Arc>, logger: Arc, - async_payments_role: Option, + liquidity_source: Arc>>, keys_manager: Arc, + payment_store: Arc, config: Arc, is_running: Arc>, + logger: Arc, async_payments_role: Option, ) -> Self { Self { runtime, channel_manager, + liquidity_source, keys_manager, payment_store, config, @@ -155,6 +160,7 @@ impl Bolt12Payment { hash: None, preimage: None, secret: None, + counterparty_skimmed_fee_msat: None, offer_id: offer.id(), payer_note: payer_note.map(UntrustedString), quantity, @@ -180,6 +186,7 @@ impl Bolt12Payment { hash: None, preimage: None, secret: None, + counterparty_skimmed_fee_msat: None, offer_id: offer.id(), payer_note: payer_note.map(UntrustedString), quantity, @@ -201,9 +208,15 @@ impl Bolt12Payment { } pub(crate) fn receive_inner( - &self, amount_msat: u64, description: &str, expiry_secs: Option, quantity: Option, + &self, amount_msat: u64, description: &str, expiry_secs: Option, + quantity: Option, address_by_node_id: bool, ) -> Result { - let mut offer_builder = self.channel_manager.create_offer_builder().map_err(|e| { + let offer_builder = if address_by_node_id { + self.channel_manager.create_offer_builder_using_router(NullMessageRouter {}) + } else { + self.channel_manager.create_offer_builder() + }; + let mut offer_builder = offer_builder.map_err(|e| { log_error!(self.logger, "Failed to create offer builder: {:?}", e); Error::OfferCreationFailed })?; @@ -237,6 +250,40 @@ impl Bolt12Payment { Ok(finalized_offer) } + fn receive_variable_amount_inner( + &self, description: &str, expiry_secs: Option, address_by_node_id: bool, + ) -> Result { + let offer_builder = if address_by_node_id { + self.channel_manager.create_offer_builder_using_router(NullMessageRouter {}) + } else { + self.channel_manager.create_offer_builder() + }; + let mut offer_builder = offer_builder.map_err(|e| { + log_error!(self.logger, "Failed to create offer builder: {:?}", e); + Error::OfferCreationFailed + })?; + + if let Some(expiry_secs) = expiry_secs { + let absolute_expiry = (SystemTime::now() + Duration::from_secs(expiry_secs as u64)) + .duration_since(UNIX_EPOCH) + .expect("system time must be after Unix epoch"); + offer_builder = offer_builder.absolute_expiry(absolute_expiry); + } + + offer_builder.description(description.to_string()).build().map_err(|e| { + log_error!(self.logger, "Failed to create offer: {:?}", e); + Error::OfferCreationFailed + }) + } + + fn register_jit_offer( + &self, offer: &LdkOffer, amount: PendingOfferAmount, + ) -> Result<(), Error> { + let pending_offer = pending_jit_offer(offer, amount); + self.runtime + .block_on(self.liquidity_source.lsps2_client().register_pending_offer(pending_offer)) + } + fn blinded_paths_for_async_recipient_internal( &self, recipient_id: Vec, ) -> Result, Error> { @@ -253,6 +300,55 @@ impl Bolt12Payment { } } +fn pending_jit_offer(offer: &LdkOffer, amount: PendingOfferAmount) -> PendingOffer { + PendingOffer::new(offer.id(), amount, offer.absolute_expiry().map(|expiry| expiry.as_secs())) +} + +#[cfg(test)] +mod tests { + use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey}; + use lightning::offers::offer::{OfferBuilder, OfferId}; + + use super::*; + use crate::liquidity::client::lsps2::state::{PendingOfferAmount, PendingOfferId}; + + #[test] + fn jit_offer_descriptor_preserves_offer_lifetime() { + let node_id = PublicKey::from_secret_key( + &Secp256k1::new(), + &SecretKey::from_slice(&[42; 32]).unwrap(), + ); + let offer = OfferBuilder::new(node_id).description("test".to_string()).build().unwrap(); + let descriptor = pending_jit_offer( + &offer, + PendingOfferAmount::Fixed { amount_msat: 1_000, max_total_fee_msat: Some(100) }, + ); + + assert_eq!(OfferId::from(descriptor.id), offer.id()); + assert_eq!(descriptor.absolute_expiry, None); + } + + #[test] + fn jit_offer_descriptor_preserves_offer_expiry() { + let node_id = PublicKey::from_secret_key( + &Secp256k1::new(), + &SecretKey::from_slice(&[43; 32]).unwrap(), + ); + let offer = OfferBuilder::new(node_id) + .absolute_expiry(Duration::from_secs(42)) + .description("test".to_string()) + .build() + .unwrap(); + let descriptor = pending_jit_offer( + &offer, + PendingOfferAmount::Variable { max_proportional_fee_ppm_msat: Some(100) }, + ); + + assert_eq!(PendingOfferId::from(offer.id()), descriptor.id); + assert_eq!(descriptor.absolute_expiry, Some(42)); + } +} + #[cfg_attr(feature = "uniffi", uniffi::export)] impl Bolt12Payment { /// Send a payment given an offer. @@ -317,6 +413,7 @@ impl Bolt12Payment { hash: None, preimage: None, secret: None, + counterparty_skimmed_fee_msat: None, offer_id: offer.id(), payer_note: payer_note.map(UntrustedString), quantity, @@ -342,6 +439,7 @@ impl Bolt12Payment { hash: None, preimage: None, secret: None, + counterparty_skimmed_fee_msat: None, offer_id: offer.id(), payer_note: payer_note.map(UntrustedString), quantity, @@ -394,7 +492,7 @@ impl Bolt12Payment { pub fn receive( &self, amount_msat: u64, description: &str, expiry_secs: Option, quantity: Option, ) -> Result { - let offer = self.receive_inner(amount_msat, description, expiry_secs, quantity)?; + let offer = self.receive_inner(amount_msat, description, expiry_secs, quantity, false)?; Ok(maybe_wrap(offer)) } @@ -403,23 +501,52 @@ impl Bolt12Payment { pub fn receive_variable_amount( &self, description: &str, expiry_secs: Option, ) -> Result { - let mut offer_builder = self.channel_manager.create_offer_builder().map_err(|e| { - log_error!(self.logger, "Failed to create offer builder: {:?}", e); - Error::OfferCreationFailed - })?; - - if let Some(expiry_secs) = expiry_secs { - let absolute_expiry = (SystemTime::now() + Duration::from_secs(expiry_secs as u64)) - .duration_since(UNIX_EPOCH) - .expect("system time must be after Unix epoch"); - offer_builder = offer_builder.absolute_expiry(absolute_expiry); - } + let offer = self.receive_variable_amount_inner(description, expiry_secs, false)?; + Ok(maybe_wrap(offer)) + } - let offer = offer_builder.description(description.to_string()).build().map_err(|e| { - log_error!(self.logger, "Failed to create offer: {:?}", e); - Error::OfferCreationFailed - })?; + /// Returns a payable offer that can be used to request a payment of the amount given and + /// receive it via a just-in-time (JIT) channel. + /// + /// Creating the offer does not contact an LSP. LSPS2 payment parameters are selected only when + /// an invoice request for the offer is received, allowing the offer to outlive those parameters. + /// The offer contains no blinded message paths and is addressed directly to this node's ID. + /// If sufficient inbound liquidity is available over pre-existing channels, the payment may be + /// received over those channels without opening a new JIT channel. + pub fn receive_via_jit_channel( + &self, amount_msat: u64, description: &str, expiry_secs: Option, + quantity: Option, max_total_lsp_fee_limit_msat: Option, + ) -> Result { + let offer = self.receive_inner(amount_msat, description, expiry_secs, quantity, true)?; + self.register_jit_offer( + &offer, + PendingOfferAmount::Fixed { + amount_msat, + max_total_fee_msat: max_total_lsp_fee_limit_msat, + }, + )?; + Ok(maybe_wrap(offer)) + } + /// Returns a payable offer that can be used to request a variable amount payment and receive it + /// via a just-in-time (JIT) channel. + /// + /// Creating the offer does not contact an LSP. LSPS2 payment parameters are selected only when + /// an invoice request for the offer is received, allowing the offer to outlive those parameters. + /// The offer contains no blinded message paths and is addressed directly to this node's ID. + /// If sufficient inbound liquidity is available over pre-existing channels, the payment may be + /// received over those channels without opening a new JIT channel. + pub fn receive_variable_amount_via_jit_channel( + &self, description: &str, expiry_secs: Option, + max_proportional_lsp_fee_limit_ppm_msat: Option, + ) -> Result { + let offer = self.receive_variable_amount_inner(description, expiry_secs, true)?; + self.register_jit_offer( + &offer, + PendingOfferAmount::Variable { + max_proportional_fee_ppm_msat: max_proportional_lsp_fee_limit_ppm_msat, + }, + )?; Ok(maybe_wrap(offer)) } diff --git a/src/payment/metadata.rs b/src/payment/metadata.rs new file mode 100644 index 0000000000..3d513ae1c4 --- /dev/null +++ b/src/payment/metadata.rs @@ -0,0 +1,136 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. + +use std::collections::BTreeMap; + +use lightning::impl_writeable_tlv_based; +use lightning::util::ser::{Readable, Writeable}; +use lightning_liquidity::lsps2::router::{PaymentMetadataDecoder, PaymentParameters}; + +use crate::payment::store::LSPS2Parameters; + +pub(crate) const LDK_NODE_BOLT12_PAYMENT_METADATA_KEY: u64 = 0; + +/// Metadata carried in BOLT11 invoice metadata or BOLT12 payment-context metadata. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct PaymentMetadata { + /// Fee limits retained until payment claim to validate the LSP's withholding. + pub(crate) lsps2_parameters: Option, + /// Single-use routing parameters consumed while constructing blinded payment paths. + pub(crate) lsps2_lease_parameters: Option, +} + +impl PaymentMetadata { + pub(crate) fn encode_as_bolt12_payment_metadata(&self) -> BTreeMap> { + let mut metadata = BTreeMap::new(); + metadata.insert(LDK_NODE_BOLT12_PAYMENT_METADATA_KEY, self.encode()); + metadata + } + + pub(crate) fn decode_from_bolt12_payment_metadata( + payment_metadata: &BTreeMap>, + ) -> Option { + payment_metadata + .get(&LDK_NODE_BOLT12_PAYMENT_METADATA_KEY) + .and_then(|encoded| Self::read(&mut &encoded[..]).ok()) + } +} + +impl_writeable_tlv_based!(PaymentMetadata, { + (0, lsps2_parameters, option), + (2, lsps2_lease_parameters, option), +}); + +#[derive(Clone, Copy)] +pub(crate) struct LSPS2PaymentMetadataDecoder; + +impl PaymentMetadataDecoder for LSPS2PaymentMetadataDecoder { + fn decode_payment_parameters( + &self, payment_metadata: &BTreeMap>, + ) -> Vec { + PaymentMetadata::decode_from_bolt12_payment_metadata(payment_metadata) + .and_then(|metadata| metadata.lsps2_lease_parameters) + .into_iter() + .collect() + } +} + +#[cfg(test)] +mod tests { + use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey}; + use lightning::util::ser::{Readable, Writeable}; + use lightning_liquidity::lsps2::router::{PaymentMetadataDecoder, PaymentParameters}; + + use super::*; + + fn pubkey(byte: u8) -> PublicKey { + PublicKey::from_secret_key(&Secp256k1::new(), &SecretKey::from_slice(&[byte; 32]).unwrap()) + } + + #[test] + fn empty_metadata_roundtrips() { + let metadata = PaymentMetadata { lsps2_parameters: None, lsps2_lease_parameters: None }; + + let encoded = metadata.encode(); + let decoded = PaymentMetadata::read(&mut &*encoded).unwrap(); + + assert_eq!(metadata, decoded); + } + + #[test] + fn lsps2_parameters_roundtrip() { + let lsps2_parameters = LSPS2Parameters { + max_total_opening_fee_msat: Some(42_000), + max_proportional_opening_fee_ppm_msat: Some(17_000), + }; + let metadata = PaymentMetadata { + lsps2_parameters: Some(lsps2_parameters), + lsps2_lease_parameters: None, + }; + + let encoded = metadata.encode(); + let decoded = PaymentMetadata::read(&mut &*encoded).unwrap(); + + assert_eq!(metadata, decoded); + } + + #[test] + fn bolt12_decoder_extracts_lease_parameters() { + let lease_parameters = PaymentParameters { + lsp_node_id: pubkey(3), + intercept_scid: 43, + cltv_expiry_delta: 72, + payment_size_msat: None, + valid_until: 1_000_000, + }; + let metadata = PaymentMetadata { + lsps2_parameters: None, + lsps2_lease_parameters: Some(lease_parameters), + } + .encode_as_bolt12_payment_metadata(); + + assert_eq!( + LSPS2PaymentMetadataDecoder.decode_payment_parameters(&metadata), + vec![lease_parameters] + ); + } + + #[test] + fn bolt12_metadata_roundtrips() { + let metadata = PaymentMetadata { + lsps2_parameters: Some(LSPS2Parameters { + max_total_opening_fee_msat: Some(42_000), + max_proportional_opening_fee_ppm_msat: None, + }), + lsps2_lease_parameters: None, + }; + + let encoded = metadata.encode_as_bolt12_payment_metadata(); + + assert_eq!(PaymentMetadata::decode_from_bolt12_payment_metadata(&encoded), Some(metadata)); + } +} diff --git a/src/payment/mod.rs b/src/payment/mod.rs index fd75322ceb..9540d21f1e 100644 --- a/src/payment/mod.rs +++ b/src/payment/mod.rs @@ -10,6 +10,7 @@ pub(crate) mod asynchronous; mod bolt11; mod bolt12; +mod metadata; mod onchain; pub(crate) mod pending_payment_store; mod spontaneous; @@ -17,8 +18,8 @@ pub(crate) mod store; mod unified; pub use bolt11::Bolt11Payment; -pub(crate) use bolt11::PaymentMetadata; pub use bolt12::Bolt12Payment; +pub(crate) use metadata::{LSPS2PaymentMetadataDecoder, PaymentMetadata}; pub use onchain::OnchainPayment; pub(crate) use pending_payment_store::{FundingTxCandidate, PendingPaymentDetails}; pub use spontaneous::SpontaneousPayment; diff --git a/src/payment/store.rs b/src/payment/store.rs index d2b92747a2..ae7ab206b3 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -253,12 +253,13 @@ impl StorableObject for PaymentDetails { if let Some(skimmed_fee_msat) = update.counterparty_skimmed_fee_msat { match self.kind { - PaymentKind::Bolt11 { ref mut counterparty_skimmed_fee_msat, .. } => { + PaymentKind::Bolt11 { ref mut counterparty_skimmed_fee_msat, .. } + | PaymentKind::Bolt12Offer { ref mut counterparty_skimmed_fee_msat, .. } => { update_if_necessary!(*counterparty_skimmed_fee_msat, skimmed_fee_msat); }, _ => debug_assert!( false, - "We should only ever override counterparty_skimmed_fee_msat for BOLT11 payments" + "We should only override counterparty_skimmed_fee_msat for invoice payments" ), } } @@ -545,6 +546,17 @@ pub enum PaymentKind { preimage: Option, /// The secret used by the payment. secret: Option, + /// The value, in thousands of a satoshi, that was deducted from this payment as an extra + /// fee taken by our channel counterparty. + /// + /// Will only ever be `Some` for inbound payments received via an [bLIP-52 / LSPS 2] + /// just-in-time channel, and only after the payment is observed; `None` otherwise. + /// + /// This will always be `None` for payments serialized by versions that did not record + /// BOLT12 LSPS2 fees. + /// + /// [bLIP-52 / LSPS 2]: https://github.com/lightning/blips/blob/master/blip-0052.md + counterparty_skimmed_fee_msat: Option, /// The ID of the offer this payment is for. offer_id: OfferId, /// The payer note for the payment. @@ -617,6 +629,7 @@ impl_writeable_tlv_based_enum!(PaymentKind, (2, preimage, option), (3, quantity, option), (4, secret, option), + (5, counterparty_skimmed_fee_msat, option), (6, offer_id, required), }, (8, Spontaneous) => { @@ -732,7 +745,8 @@ impl From<&PaymentDetails> for PaymentDetailsUpdate { }; let counterparty_skimmed_fee_msat = match value.kind { - PaymentKind::Bolt11 { counterparty_skimmed_fee_msat, .. } => { + PaymentKind::Bolt11 { counterparty_skimmed_fee_msat, .. } + | PaymentKind::Bolt12Offer { counterparty_skimmed_fee_msat, .. } => { Some(counterparty_skimmed_fee_msat) }, _ => None, @@ -1022,6 +1036,38 @@ mod tests { } } + #[test] + fn bolt12_offer_records_counterparty_skimmed_fee() { + let payment_id = PaymentId([41; 32]); + let mut payment = PaymentDetails::new( + payment_id, + PaymentKind::Bolt12Offer { + hash: Some(PaymentHash([42; 32])), + preimage: None, + secret: None, + counterparty_skimmed_fee_msat: None, + offer_id: OfferId([43; 32]), + payer_note: None, + quantity: None, + }, + Some(100_000), + None, + PaymentDirection::Inbound, + PaymentStatus::Pending, + ); + + assert!(payment.update(PaymentDetailsUpdate { + counterparty_skimmed_fee_msat: Some(Some(1_000)), + ..PaymentDetailsUpdate::new(payment_id) + })); + assert!(matches!( + payment.kind, + PaymentKind::Bolt12Offer { counterparty_skimmed_fee_msat: Some(1_000), .. } + )); + let encoded = payment.encode(); + assert_eq!(payment, PaymentDetails::read(&mut &*encoded).unwrap()); + } + #[derive(Clone, Debug, PartialEq, Eq)] struct LegacyBolt11JitKind { hash: PaymentHash, diff --git a/src/payment/unified.rs b/src/payment/unified.rs index cb51174140..508c4f9b7f 100644 --- a/src/payment/unified.rs +++ b/src/payment/unified.rs @@ -134,7 +134,7 @@ impl UnifiedPayment { let onchain_address = self.onchain_payment.new_address()?; let bolt12_offer = - match self.bolt12_payment.receive_inner(amount_msats, description, None, None) { + match self.bolt12_payment.receive_inner(amount_msats, description, None, None, false) { Ok(offer) => Some(maybe_wrap(offer)), Err(e) => { log_error!(self.logger, "Failed to create offer: {}", e); diff --git a/src/types.rs b/src/types.rs index 5552877ef8..303365ca06 100644 --- a/src/types.rs +++ b/src/types.rs @@ -37,6 +37,7 @@ use lightning::util::persist::{ use lightning::util::ser::{Readable, Writeable, Writer}; use lightning::util::sweep::OutputSweeper; use lightning_block_sync::gossip::GossipVerifier; +use lightning_liquidity::lsps2::router::LSPS2Router; use lightning_liquidity::utils::time::DefaultTimeProvider; use lightning_net_tokio::SocketDescriptor; @@ -48,7 +49,7 @@ use crate::fee_estimator::OnchainFeeEstimator; use crate::ffi::maybe_wrap; use crate::logger::Logger; use crate::message_handler::NodeCustomMessageHandler; -use crate::payment::{PaymentDetails, PendingPaymentDetails}; +use crate::payment::{LSPS2PaymentMetadataDecoder, PaymentDetails, PendingPaymentDetails}; use crate::runtime::RuntimeSpawner; #[cfg(not(feature = "uniffi"))] @@ -273,7 +274,8 @@ pub(crate) type Broadcaster = crate::tx_broadcaster::TransactionBroadcaster, LSPS2PaymentMetadataDecoder>; +pub(crate) type BaseRouter = DefaultRouter< Arc, Arc, Arc, diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index ba15f6fa30..9f9e82c1f0 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -2312,12 +2312,14 @@ async fn simple_bolt12_send_receive() { hash, preimage, secret: _, + counterparty_skimmed_fee_msat, offer_id, quantity: ref qty, payer_note: ref note, } => { assert!(hash.is_some()); assert!(preimage.is_some()); + assert_eq!(counterparty_skimmed_fee_msat, None); assert_eq!(offer_id, offer.id()); assert_eq!(&expected_quantity, qty); assert_eq!(expected_payer_note.unwrap(), note.clone().unwrap().0); @@ -2379,12 +2381,14 @@ async fn simple_bolt12_send_receive() { hash, preimage, secret: _, + counterparty_skimmed_fee_msat, offer_id, quantity: ref qty, payer_note: ref note, } => { assert!(hash.is_some()); assert!(preimage.is_some()); + assert_eq!(counterparty_skimmed_fee_msat, None); assert_eq!(offer_id, offer.id()); assert_eq!(&expected_quantity, qty); assert_eq!(expected_payer_note.unwrap(), note.clone().unwrap().0); @@ -3125,6 +3129,230 @@ async fn do_lsps2_client_service_integration(client_trusts_lsp: bool) { assert_eq!(client_node.payment(&payment_id).unwrap().status, PaymentStatus::Failed); } +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn bolt12_lsps2_client_service_integration() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let esplora_url = format!("http://{}", electrsd.esplora_url.as_ref().unwrap()); + + let mut sync_config = EsploraSyncConfig::default(); + sync_config.background_sync_config = None; + + let channel_opening_fee_ppm = 10_000; + let lsps2_service_config = LSPS2ServiceConfig { + require_token: None, + advertise_service: false, + channel_opening_fee_ppm, + channel_over_provisioning_ppm: 100_000, + max_payment_size_msat: 1_000_000_000, + min_payment_size_msat: 0, + min_channel_lifetime: 100, + min_channel_opening_fee_msat: 0, + max_client_to_self_delay: 1024, + client_trusts_lsp: true, + disable_client_reserve: false, + }; + + let service_config = random_config(); + setup_builder!(service_builder, service_config.node_config); + service_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); + service_builder.enable_liquidity_provider(lsps2_service_config); + let service_node = service_builder.build(service_config.node_entropy.into()).unwrap(); + service_node.start().unwrap(); + + let service_node_id = service_node.node_id(); + let service_addr = service_node.listening_addresses().unwrap().first().unwrap().clone(); + + let client_config = random_config(); + let client_store = + TestSyncStore::new(client_config.node_config.storage_dir_path.clone().into()); + setup_builder!(client_builder, client_config.node_config); + client_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); + client_builder.add_liquidity_source(service_node_id, service_addr.clone(), None, true); + let client_node = client_builder + .build_with_store(client_config.node_entropy.into(), client_store.clone()) + .unwrap(); + client_node.start().unwrap(); + + let payer_config = random_config(); + setup_builder!(payer_builder, payer_config.node_config); + payer_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); + let payer_node = payer_builder.build(payer_config.node_entropy.into()).unwrap(); + payer_node.start().unwrap(); + + let client_addr = client_node.listening_addresses().unwrap().first().unwrap().clone(); + payer_node.connect(client_node.node_id(), client_addr, false).unwrap(); + + let service_onchain_addr = service_node.onchain_payment().new_address().unwrap(); + let client_onchain_addr = client_node.onchain_payment().new_address().unwrap(); + let payer_onchain_addr = payer_node.onchain_payment().new_address().unwrap(); + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![service_onchain_addr, client_onchain_addr, payer_onchain_addr], + Amount::from_sat(10_000_000), + ) + .await; + service_node.sync_wallets().unwrap(); + client_node.sync_wallets().unwrap(); + payer_node.sync_wallets().unwrap(); + + open_channel(&payer_node, &service_node, 5_000_000, true, &electrsd).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + service_node.sync_wallets().unwrap(); + payer_node.sync_wallets().unwrap(); + expect_channel_ready_event!(payer_node, service_node.node_id()); + expect_channel_ready_event!(service_node, payer_node.node_id()); + while payer_node.status().latest_node_announcement_broadcast_timestamp.is_none() { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + + let fixed_amount_msat = 100_000_000; + let fixed_offer = client_node + .bolt12_payment() + .receive_via_jit_channel(fixed_amount_msat, "fixed", None, None, None) + .unwrap(); + assert!(fixed_offer.paths().is_empty()); + assert_eq!(fixed_offer.issuer_signing_pubkey(), Some(client_node.node_id())); + let fixed_payment_id = + payer_node.bolt12_payment().send(&fixed_offer, None, None, None).unwrap(); + + expect_channel_pending_event!(service_node, client_node.node_id()); + expect_channel_ready_event!(service_node, client_node.node_id()); + expect_event!(service_node, PaymentForwarded); + expect_channel_pending_event!(client_node, service_node.node_id()); + expect_channel_ready_event!(client_node, service_node.node_id()); + expect_payment_successful_event!(payer_node, Some(fixed_payment_id), None); + let fixed_fee_msat = fixed_amount_msat * channel_opening_fee_ppm as u64 / 1_000_000; + let fixed_received_msat = fixed_amount_msat - fixed_fee_msat; + let fixed_receiver_payment_id = + expect_payment_received_event!(client_node, fixed_received_msat).unwrap(); + match client_node.payment(&fixed_receiver_payment_id).unwrap().kind { + PaymentKind::Bolt12Offer { counterparty_skimmed_fee_msat, .. } => { + assert_eq!(counterparty_skimmed_fee_msat, Some(fixed_fee_msat)); + }, + _ => panic!("Unexpected payment kind"), + } + + let persisted_lease_key = + tokio::time::timeout(std::time::Duration::from_secs(common::INTEROP_TIMEOUT_SECS), async { + loop { + let keys = KVStore::list(&client_store, "lsps2_leases", "").await.unwrap(); + if let Some(key) = keys.first() { + return key.clone(); + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("replacement lease should be persisted"); + assert!(KVStore::read(&client_store, "lsps2_leases", "", &persisted_lease_key).await.is_ok()); + + let client_node_id = client_node.node_id(); + client_node.stop().unwrap(); + drop(client_node); + tokio::time::timeout(std::time::Duration::from_secs(common::INTEROP_TIMEOUT_SECS), async { + while payer_node.list_peers().iter().any(|peer| peer.node_id == client_node_id) + || service_node.list_peers().iter().any(|peer| peer.node_id == client_node_id) + { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("peers should observe the stopped client"); + + setup_builder!(restarted_client_builder, client_config.node_config); + restarted_client_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); + restarted_client_builder.add_liquidity_source( + service_node_id, + service_addr.clone(), + None, + true, + ); + let client_node = restarted_client_builder + .build_with_store(client_config.node_entropy.into(), client_store.clone()) + .unwrap(); + assert_eq!(client_node.node_id(), client_node_id); + assert!(KVStore::read(&client_store, "lsps2_leases", "", &persisted_lease_key).await.is_ok()); + client_node.start().unwrap(); + let client_addr = client_node.listening_addresses().unwrap().first().unwrap().clone(); + payer_node.connect(client_node.node_id(), client_addr, false).unwrap(); + + let restored_payment_id = + payer_node.bolt12_payment().send(&fixed_offer, None, None, None).unwrap(); + expect_channel_pending_event!(service_node, client_node.node_id()); + expect_channel_ready_event!(service_node, client_node.node_id()); + expect_event!(service_node, PaymentForwarded); + loop { + let event = tokio::time::timeout( + std::time::Duration::from_secs(common::INTEROP_TIMEOUT_SECS), + client_node.next_event_async(), + ) + .await + .expect("restored client should receive a payment event"); + match event { + Event::ChannelPending { counterparty_node_id, .. } => { + assert_eq!(counterparty_node_id, service_node.node_id()); + client_node.event_handled().unwrap(); + break; + }, + Event::PaymentReceived { payment_id, amount_msat, .. } => { + assert_eq!(payment_id, Some(fixed_receiver_payment_id)); + assert_eq!(amount_msat, fixed_received_msat); + client_node.event_handled().unwrap(); + }, + unexpected => panic!("Unexpected event after restoring client: {unexpected:?}"), + } + } + expect_channel_ready_event!(client_node, service_node.node_id()); + expect_payment_successful_event!(payer_node, Some(restored_payment_id), None); + let restored_receiver_payment_id = + expect_payment_received_event!(client_node, fixed_received_msat).unwrap(); + match client_node.payment(&restored_receiver_payment_id).unwrap().kind { + PaymentKind::Bolt12Offer { counterparty_skimmed_fee_msat, .. } => { + assert_eq!(counterparty_skimmed_fee_msat, Some(fixed_fee_msat)); + }, + _ => panic!("Unexpected payment kind"), + } + assert!( + KVStore::read(&client_store, "lsps2_leases", "", &persisted_lease_key).await.is_err(), + "the restored lease must be consumed instead of renegotiating" + ); + + // The fixed JIT channels cannot carry this larger variable payment. The variable offer disables + // MPP, so the payer must use the fresh JIT path rather than splitting across the existing paths. + let variable_amount_msat = 200_000_000; + let variable_offer = client_node + .bolt12_payment() + .receive_variable_amount_via_jit_channel("variable", None, None) + .unwrap(); + service_node.stop().unwrap(); + service_node.start().unwrap(); + let _ = payer_node.connect(service_node_id, service_addr.clone(), false); + let _ = client_node.connect(service_node_id, service_addr, false); + let variable_payment_id = payer_node + .bolt12_payment() + .send_using_amount(&variable_offer, variable_amount_msat, None, None, None) + .unwrap(); + + expect_channel_pending_event!(service_node, client_node.node_id()); + expect_channel_ready_event!(service_node, client_node.node_id()); + expect_event!(service_node, PaymentForwarded); + expect_channel_pending_event!(client_node, service_node.node_id()); + expect_channel_ready_event!(client_node, service_node.node_id()); + expect_payment_successful_event!(payer_node, Some(variable_payment_id), None); + let variable_fee_msat = variable_amount_msat * channel_opening_fee_ppm as u64 / 1_000_000; + let variable_received_msat = variable_amount_msat - variable_fee_msat; + let variable_receiver_payment_id = + expect_payment_received_event!(client_node, variable_received_msat).unwrap(); + match client_node.payment(&variable_receiver_payment_id).unwrap().kind { + PaymentKind::Bolt12Offer { counterparty_skimmed_fee_msat, .. } => { + assert_eq!(counterparty_skimmed_fee_msat, Some(variable_fee_msat)); + }, + _ => panic!("Unexpected payment kind"), + } +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn facade_logging() { let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); @@ -4302,17 +4530,18 @@ async fn lsps2_multi_lsp_picks_cheapest() { } async fn do_lsps2_multi_lsp_picks_cheapest(reverse_order: bool) { - let (_bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); let esplora_url = format!("http://{}", electrsd.esplora_url.as_ref().unwrap()); let mut sync_config = EsploraSyncConfig::default(); sync_config.background_sync_config = None; // Cheap LSP: 10_000 ppm. + let cheap_opening_fee_ppm = 10_000; let cheap_cfg = LSPS2ServiceConfig { require_token: None, advertise_service: false, - channel_opening_fee_ppm: 10_000, + channel_opening_fee_ppm: cheap_opening_fee_ppm, channel_over_provisioning_ppm: 100_000, max_payment_size_msat: 1_000_000_000, min_payment_size_msat: 0, @@ -4368,6 +4597,40 @@ async fn do_lsps2_multi_lsp_picks_cheapest(reverse_order: bool) { let client = client_builder.build(client_config.node_entropy.into()).unwrap(); client.start().unwrap(); + let payer_config = random_config(); + setup_builder!(payer_builder, payer_config.node_config); + payer_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); + let payer = payer_builder.build(payer_config.node_entropy.into()).unwrap(); + payer.start().unwrap(); + + let client_addr = client.listening_addresses().unwrap().first().unwrap().clone(); + payer.connect(client.node_id(), client_addr, false).unwrap(); + + let cheap_onchain_addr = cheap.onchain_payment().new_address().unwrap(); + let client_onchain_addr = client.onchain_payment().new_address().unwrap(); + let payer_onchain_addr = payer.onchain_payment().new_address().unwrap(); + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![cheap_onchain_addr, client_onchain_addr, payer_onchain_addr], + Amount::from_sat(10_000_000), + ) + .await; + cheap.sync_wallets().unwrap(); + client.sync_wallets().unwrap(); + payer.sync_wallets().unwrap(); + + open_channel(&payer, &cheap, 5_000_000, true, &electrsd).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + cheap.sync_wallets().unwrap(); + payer.sync_wallets().unwrap(); + expect_channel_ready_event!(payer, cheap.node_id()); + expect_channel_ready_event!(cheap, payer.node_id()); + while payer.status().latest_node_announcement_broadcast_timestamp.is_none() { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + let invoice_description = Bolt11InvoiceDescription::Direct(Description::new(String::from("asdf")).unwrap()).into(); let jit_invoice = client @@ -4385,6 +4648,32 @@ async fn do_lsps2_multi_lsp_picks_cheapest(reverse_order: bool) { let route_hint_src = first_hop.expect("route hint should have at least one hop").src_node_id; assert_eq!(route_hint_src, cheap_id, "expected cheaper LSP to be selected."); + // Consuming the BOLT11 lease schedules a replacement. The BOLT12 flow shares that cache and + // must retain the same cheapest-LSP selection regardless of registration order. + let payment_amount_msat = 100_000_000; + let offer = client + .bolt12_payment() + .receive_via_jit_channel(payment_amount_msat, "multi LSP", None, None, None) + .unwrap(); + let payment_id = payer.bolt12_payment().send(&offer, None, None, None).unwrap(); + + expect_channel_pending_event!(cheap, client.node_id()); + expect_channel_ready_event!(cheap, client.node_id()); + expect_event!(cheap, PaymentForwarded); + expect_channel_pending_event!(client, cheap.node_id()); + expect_channel_ready_event!(client, cheap.node_id()); + expect_payment_successful_event!(payer, Some(payment_id), None); + let fee_msat = payment_amount_msat * cheap_opening_fee_ppm as u64 / 1_000_000; + let receiver_payment_id = + expect_payment_received_event!(client, payment_amount_msat - fee_msat).unwrap(); + match client.payment(&receiver_payment_id).unwrap().kind { + PaymentKind::Bolt12Offer { counterparty_skimmed_fee_msat, .. } => { + assert_eq!(counterparty_skimmed_fee_msat, Some(fee_msat)); + }, + _ => panic!("Unexpected payment kind"), + } + + payer.stop().unwrap(); client.stop().unwrap(); cheap.stop().unwrap(); expensive.stop().unwrap();