From 78f883644ce833d70261914ca061f3eff7914a6a Mon Sep 17 00:00:00 2001 From: Camillarhi Date: Mon, 6 Jul 2026 14:28:32 +0100 Subject: [PATCH] Add LSPS5 (bLIP-55) webhook notification support Implement bLIP-55 / LSPS5 on top of the multi-LSP liquidity module. Clients register, list and remove webhooks with a given LSP via `Liquidity::lsps5`. Each call takes the LSP's node ID explicitly, as the notification delivery service verifies every notification against the node ID of the LSP that sent it. The service side delivers signed notifications over HTTPS to wake offline clients: - HTLCs to a client's private channels are intercepted and held for up to 10 seconds while the client connects. - Onion messages for a client are kept in the onion message mailbox until it reconnects. - Clients are notified when an HTLC nears the height at which the LSP would have to force-close their channel. - Operators can ask a client to come online before reclaiming liquidity. `enable_liquidity_provider` now takes optional LSPS2 and LSPS5 configs alongside `advertise_service`, which moves out of `LSPS2ServiceConfig` as the LSPS feature bit it sets is shared by all services. --- bindings/ldk_node.udl | 9 + src/builder.rs | 149 ++++++--- src/config.rs | 31 ++ src/error.rs | 59 ++++ src/event.rs | 40 ++- src/lib.rs | 52 ++- src/liquidity/client/lsps5.rs | 574 ++++++++++++++++++++++++++++++++ src/liquidity/client/mod.rs | 1 + src/liquidity/mod.rs | 108 +++++- src/liquidity/service/lsps2.rs | 2 - src/liquidity/service/lsps5.rs | 441 ++++++++++++++++++++++++ src/liquidity/service/mod.rs | 1 + tests/integration_tests_rust.rs | 313 ++++++++++++++++- 13 files changed, 1696 insertions(+), 84 deletions(-) create mode 100644 src/liquidity/client/lsps5.rs create mode 100644 src/liquidity/service/lsps5.rs diff --git a/bindings/ldk_node.udl b/bindings/ldk_node.udl index df6aa18f8d..a7a763d68f 100644 --- a/bindings/ldk_node.udl +++ b/bindings/ldk_node.udl @@ -213,6 +213,15 @@ enum NodeError { "InvalidLnurl", "ChainSourceNotSupported", "InvalidPayerProof", + "LiquiditySetWebhookFailed", + "LiquidityRemoveWebhookFailed", + "LiquidityListWebhooksFailed", + "LiquidityNotifyWebhookFailed", + "LiquidityNotifyWebhookRateLimited", + "LiquidityWebhookLimitExceeded", + "LiquidityWebhookNoPriorActivity", + "LiquidityWebhookAppNameNotFound", + "LiquidityWebhookInvalid" }; typedef dictionary NodeStatus; diff --git a/src/builder.rs b/src/builder.rs index 1158044e47..c095c0b317 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -88,7 +88,7 @@ use crate::io::{ PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, }; -use crate::liquidity::{LSPS2ServiceConfig, LiquiditySourceBuilder, LspConfig}; +use crate::liquidity::{LSPS2ServiceConfig, LSPS5ServiceConfig, LiquiditySourceBuilder, LspConfig}; use crate::lnurl_auth::LnurlAuth; use crate::logger::{log_error, LdkLogger, LogLevel, LogWriter, Logger}; use crate::message_handler::NodeCustomMessageHandler; @@ -149,10 +149,14 @@ struct PathfindingScoresSyncConfig { #[derive(Debug, Clone, Default)] struct LiquiditySourceConfig { - // Acts for both LSPS1 and LSPS2 clients connecting to the given service. + // Acts for LSPS1, LSPS2 and LSPS5 clients connecting to the given service. lsp_nodes: Vec, // Act as an LSPS2 service. lsps2_service: Option, + // Act as an LSPS5 service. + lsps5_service: Option, + // Indicates whether the LSPS service will be announced via the gossip network. + advertise_service: bool, } #[derive(Clone)] @@ -545,18 +549,30 @@ impl NodeBuilder { self } - /// Configures the [`Node`] instance to provide an [LSPS2] service, issuing just-in-time - /// channels to clients. + /// Configures the [`Node`] instance to provide [bLIP-52 / LSPS2] and/or [bLIP-55 / LSPS5] + /// services to clients. + /// + /// [bLIP-52 / LSPS2] issues just-in-time channels to clients, [bLIP-55 / LSPS5] allows clients + /// to register webhooks for push notifications. + /// + /// Passing `None` leaves the respective service disabled. + /// + /// `advertise_service` indicates whether we'll announce LSPS support via the gossip network. + /// This signals that we act as an LSP, so it applies to every service enabled here. /// /// **Caution**: LSP service support is in **alpha** and is considered an experimental feature. /// - /// [LSPS2]: https://github.com/BitcoinAndLightningLayerSpecs/lsp/blob/main/LSPS2/README.md + /// [bLIP-52 / LSPS2]: https://github.com/lightning/blips/blob/master/blip-0052.md + /// [bLIP-55 / LSPS5]: https://github.com/lightning/blips/blob/master/blip-0055.md pub fn enable_liquidity_provider( - &mut self, lsps2_service_config: LSPS2ServiceConfig, + &mut self, lsps2_service_config: Option, + lsps5_service_config: Option, advertise_service: bool, ) -> &mut Self { let liquidity_source_config = self.liquidity_source_config.get_or_insert(LiquiditySourceConfig::default()); - liquidity_source_config.lsps2_service = Some(lsps2_service_config); + liquidity_source_config.lsps2_service = lsps2_service_config; + liquidity_source_config.lsps5_service = lsps5_service_config; + liquidity_source_config.advertise_service = advertise_service; self } @@ -1194,14 +1210,30 @@ impl Builder { #[cfg(feature = "uniffi")] impl ArcedNodeBuilder { - /// Configures the [`Node`] instance to provide an [LSPS2] service, issuing just-in-time - /// channels to clients. + /// Configures the [`Node`] instance to provide [bLIP-52 / LSPS2] and/or [bLIP-55 / LSPS5] + /// services to clients. + /// + /// [bLIP-52 / LSPS2] issues just-in-time channels to clients, [bLIP-55 / LSPS5] allows clients + /// to register webhooks for push notifications. + /// + /// Passing `None` leaves the respective service disabled. + /// + /// `advertise_service` indicates whether we'll announce LSPS support via the gossip network. + /// This signals that we act as an LSP, so it applies to every service enabled here. /// /// **Caution**: LSP service support is in **alpha** and is considered an experimental feature. /// - /// [LSPS2]: https://github.com/BitcoinAndLightningLayerSpecs/lsp/blob/main/LSPS2/README.md - pub fn enable_liquidity_provider(&self, lsps2_service_config: LSPS2ServiceConfig) { - self.inner.write().expect("lock").enable_liquidity_provider(lsps2_service_config); + /// [bLIP-52 / LSPS2]: https://github.com/lightning/blips/blob/master/blip-0052.md + /// [bLIP-55 / LSPS5]: https://github.com/lightning/blips/blob/master/blip-0055.md + pub fn enable_liquidity_provider( + &self, lsps2_service_config: Option, + lsps5_service_config: Option, advertise_service: bool, + ) { + self.inner.write().expect("lock").enable_liquidity_provider( + lsps2_service_config, + lsps5_service_config, + advertise_service, + ); } } @@ -2097,12 +2129,27 @@ fn build_with_store_internal( let mut user_config = default_user_config(&config); - if liquidity_source_config.and_then(|lsc| lsc.lsps2_service.as_ref()).is_some() { + let lsps2_service = liquidity_source_config.and_then(|lsc| lsc.lsps2_service.as_ref()); + let lsps5_service = liquidity_source_config.and_then(|lsc| lsc.lsps5_service.as_ref()); + + if lsps2_service.is_some() || lsps5_service.is_some() { + let mut interception_flags = 0u8; + // If we act as an LSPS2 service, we need to be able to intercept HTLCs and forward the // information to the service handler. - user_config.htlc_interception_flags = HTLCInterceptionFlags::ToInterceptSCIDs.into(); + if lsps2_service.is_some() { + interception_flags |= HTLCInterceptionFlags::ToInterceptSCIDs as u8; + } + + // As an LSPS5 service we intercept HTLCs destined for offline clients, so we can wake them + // and forward once they connect. + if lsps5_service.is_some() { + interception_flags |= HTLCInterceptionFlags::ToOfflinePrivateChannels as u8; + } + + user_config.htlc_interception_flags = interception_flags; - // If we act as an LSPS2 service, we allow forwarding to unannounced channels. + // If we act as an LSPS2 or LSPS5 service, we allow forwarding to unannounced channels. user_config.accept_forwards_to_priv_channels = true; } @@ -2247,33 +2294,36 @@ fn build_with_store_internal( Arc::new(IgnoringMessageHandler {}); // Initialize the PeerManager - let onion_messenger: Arc = - if let Some(AsyncPaymentsRole::Server) = async_payments_role { - Arc::new(OnionMessenger::new_with_offline_peer_interception( - Arc::clone(&keys_manager), - Arc::clone(&keys_manager), - Arc::clone(&logger), - Arc::clone(&channel_manager), - message_router, - Arc::clone(&channel_manager), - Arc::clone(&channel_manager), - Arc::clone(&om_resolver), - IgnoringMessageHandler {}, - false, - )) - } else { - Arc::new(OnionMessenger::new( - Arc::clone(&keys_manager), - Arc::clone(&keys_manager), - Arc::clone(&logger), - Arc::clone(&channel_manager), - message_router, - Arc::clone(&channel_manager), - Arc::clone(&channel_manager), - Arc::clone(&om_resolver), - IgnoringMessageHandler {}, - )) - }; + // Async payments servers and LSPS5 services both hold onion messages for offline peers until + // they reconnect. LSPS5 services also wake the client up via their webhooks. + let intercept_offline_peer_messages = + matches!(async_payments_role, Some(AsyncPaymentsRole::Server)) || lsps5_service.is_some(); + let onion_messenger: Arc = if intercept_offline_peer_messages { + Arc::new(OnionMessenger::new_with_offline_peer_interception( + Arc::clone(&keys_manager), + Arc::clone(&keys_manager), + Arc::clone(&logger), + Arc::clone(&channel_manager), + message_router, + Arc::clone(&channel_manager), + Arc::clone(&channel_manager), + Arc::clone(&om_resolver), + IgnoringMessageHandler {}, + false, + )) + } else { + Arc::new(OnionMessenger::new( + Arc::clone(&keys_manager), + Arc::clone(&keys_manager), + Arc::clone(&logger), + Arc::clone(&channel_manager), + message_router, + Arc::clone(&channel_manager), + Arc::clone(&channel_manager), + Arc::clone(&om_resolver), + IgnoringMessageHandler {}, + )) + }; let ephemeral_bytes: [u8; 32] = keys_manager.get_secure_random_bytes(); // Initialize the GossipSource @@ -2311,6 +2361,7 @@ fn build_with_store_internal( Arc::clone(&tx_broadcaster), Arc::clone(&kv_store), Arc::clone(&config), + Arc::clone(&runtime), Arc::clone(&logger), ); @@ -2329,6 +2380,12 @@ fn build_with_store_internal( lsc.lsps2_service.as_ref().map(|config| { liquidity_source_builder.lsps2_service(promise_secret, config.clone()) }); + + lsc.lsps5_service + .as_ref() + .map(|config| liquidity_source_builder.lsps5_service(config.clone())); + + liquidity_source_builder.set_advertise_service(lsc.advertise_service); } let liquidity_source = runtime @@ -2389,6 +2446,8 @@ fn build_with_store_internal( liquidity_source.lsps2_service().set_peer_manager(Arc::downgrade(&peer_manager)); + liquidity_source.lsps5_service().set_peer_manager(Arc::downgrade(&peer_manager)); + let connection_manager = Arc::new(ConnectionManager::new( Arc::clone(&peer_manager), config.tor_config.clone(), @@ -2441,11 +2500,7 @@ fn build_with_store_internal( }, }; - let om_mailbox = if let Some(AsyncPaymentsRole::Server) = async_payments_role { - Some(Arc::new(OnionMessageMailbox::new())) - } else { - None - }; + let om_mailbox = intercept_offline_peer_messages.then(|| Arc::new(OnionMessageMailbox::new())); let lnurl_auth = Arc::new(LnurlAuth::new(xprv, Arc::clone(&logger))); diff --git a/src/config.rs b/src/config.rs index cb74b55c80..113808bae2 100644 --- a/src/config.rs +++ b/src/config.rs @@ -14,6 +14,7 @@ use std::time::Duration; use bitcoin::secp256k1::PublicKey; use bitcoin::Network; +use lightning::chain::channelmonitor::HTLC_FAIL_BACK_BUFFER; use lightning::ln::msgs::SocketAddress; use lightning::routing::gossip::NodeAlias; use lightning::routing::router::RouteParametersConfig; @@ -169,6 +170,36 @@ pub(crate) const LIQUIDITY_DISCOVERY_RETRY_INITIAL_DELAY: Duration = Duration::f // thereafter until every configured LSP has been discovered. pub(crate) const LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY: Duration = Duration::from_secs(60 * 60); +// The timeout after which we abort a LSPS5 webhook notification operation. +pub(crate) const LSPS5_WEBHOOK_TIMEOUT_SECS: u64 = 30; + +// The maximum size of a response body we'll accept when delivering an LSPS5 webhook notification. +pub(crate) const LSPS5_WEBHOOK_MAX_RESPONSE_SIZE: usize = 64 * 1024; + +// The time in-between checks for HTLCs approaching expiry on LSPS5 clients' channels. +pub(crate) const LSPS5_EXPIRY_CHECK_INTERVAL: Duration = Duration::from_secs(60); + +// The number of blocks we wait before notifying a client about the same expiring HTLCs again. +pub(crate) const LSPS5_EXPIRY_RENOTIFY_INTERVAL_BLOCKS: u32 = 6; + +// The number of blocks before an HTLC's deadline at which we start notifying offline LSPS5 clients. +// +// A client that doesn't come online and settle an HTLC before its deadline costs us the channel. +// We anchor the lead time on `HTLC_FAIL_BACK_BUFFER`, the margin LDK itself treats as too close to +// expiry to safely handle an HTLC, and double it to leave the client room to receive the +// notification and act on it. +pub(crate) const LSPS5_EXPIRY_NOTIFICATION_THRESHOLD_BLOCKS: u32 = HTLC_FAIL_BACK_BUFFER * 2; + +// How long we hold an HTLC intercepted for an offline LSPS5 client while waiting for them to come online. +// +// LDK requires intercepted HTLCs to be forwarded or failed within a few seconds, so this is a hard +// ceiling on the wake-up we can offer: a client that misses it has to be paid by a retry from the +// sender, which by then will find them online. +pub(crate) const LSPS5_INTERCEPT_HOLD_TIMEOUT: Duration = Duration::from_secs(10); + +// How often we re-check whether a woken LSPS5 client's channel is ready to forward over. +pub(crate) const LSPS5_INTERCEPT_POLL_INTERVAL: Duration = Duration::from_millis(200); + /// The mode used for tracking forwarded payments. /// /// In either mode, a forward is tracked only when it has exactly one incoming HTLC and one outgoing diff --git a/src/error.rs b/src/error.rs index d187c68200..357778430b 100644 --- a/src/error.rs +++ b/src/error.rs @@ -151,6 +151,35 @@ pub enum Error { ChainSourceNotSupported, /// The provided payer proof is invalid. InvalidPayerProof, + /// Failed to set a webhook with the LSP. + LiquiditySetWebhookFailed, + /// Failed to remove a webhook with the LSP. + LiquidityRemoveWebhookFailed, + /// Failed to list webhooks with the LSP. + LiquidityListWebhooksFailed, + /// Failed to send a webhook notification to a client. + LiquidityNotifyWebhookFailed, + /// The webhook notification was not sent because a notification of the same kind was sent to + /// this client too recently. + /// + /// Notifications are rate limited per client, so the call may succeed when retried after the + /// LSP's cooldown period has elapsed. + LiquidityNotifyWebhookRateLimited, + /// The LSP rejected a webhook registration because the client has reached the maximum number + /// of webhooks the LSP allows. + LiquidityWebhookLimitExceeded, + /// The LSP rejected a webhook registration because we have no prior activity with it. + /// + /// LSPs typically require an open channel, or an in-flight LSPS1 or LSPS2 flow, before + /// accepting webhook registrations. + LiquidityWebhookNoPriorActivity, + /// No webhook is registered under the given `app_name` with the LSP. + LiquidityWebhookAppNameNotFound, + /// The `app_name` or webhook URL is invalid. + /// + /// The `app_name` may exceed 64 bytes, or the URL may exceed 1024 bytes, fail to parse, or + /// not use the `https` scheme. + LiquidityWebhookInvalid, } impl fmt::Display for Error { @@ -249,6 +278,36 @@ impl fmt::Display for Error { write!(f, "The configured chain source is not supported.") }, Self::InvalidPayerProof => write!(f, "The provided payer proof is invalid."), + Self::LiquiditySetWebhookFailed => { + write!(f, "Failed to set a webhook with the LSP.") + }, + Self::LiquidityRemoveWebhookFailed => { + write!(f, "Failed to remove a webhook with the LSP.") + }, + Self::LiquidityListWebhooksFailed => { + write!(f, "Failed to list webhooks with the LSP.") + }, + Self::LiquidityNotifyWebhookFailed => { + write!(f, "Failed to send a webhook notification to a client.") + }, + Self::LiquidityNotifyWebhookRateLimited => { + write!(f, "The webhook notification was rate limited and was not sent.") + }, + Self::LiquidityWebhookLimitExceeded => { + write!( + f, + "The LSP's maximum number of webhooks for this client is already reached." + ) + }, + Self::LiquidityWebhookNoPriorActivity => { + write!(f, "The LSP rejected the webhook registration due to no prior activity.") + }, + Self::LiquidityWebhookAppNameNotFound => { + write!(f, "No webhook is registered under the given app name with this LSP.") + }, + Self::LiquidityWebhookInvalid => { + write!(f, "The given app name or webhook URL is invalid.") + }, } } } diff --git a/src/event.rs b/src/event.rs index e700873cdc..ae0f37f7fe 100644 --- a/src/event.rs +++ b/src/event.rs @@ -2024,20 +2024,39 @@ where payment_hash, .. } => { - self.liquidity_source - .lsps2_service() - .handle_htlc_intercepted( - requested_next_hop_scid, - intercept_id, - expected_outbound_amount_msat, - payment_hash, - ) - .await; + // `ToOfflinePrivateChannels` interception points at one of our own channels, while + // LSPS2's JIT flow uses intercept SCIDs that match no channel of ours. Resolving + // the SCID against our channels tells the two apart, and gives us the client we'd + // be waking. + let lsps5_service = self.liquidity_source.lsps5_service(); + match lsps5_service.resolve_client_channel(requested_next_hop_scid) { + Some((client_node_id, channel_id)) => { + lsps5_service.handle_htlc_intercepted( + client_node_id, + channel_id, + intercept_id, + expected_outbound_amount_msat, + ); + }, + None => { + self.liquidity_source + .lsps2_service() + .handle_htlc_intercepted( + requested_next_hop_scid, + intercept_id, + expected_outbound_amount_msat, + payment_hash, + ) + .await; + }, + } }, LdkEvent::InvoiceReceived { .. } => { debug_assert!(false, "We currently don't handle BOLT12 invoices manually, so this event should never be emitted."); }, LdkEvent::ConnectionNeeded { node_id, addresses } => { + self.liquidity_source.lsps5_service().notify_onion_message_incoming(node_id); + let spawn_logger = self.logger.clone(); let spawn_cm = Arc::clone(&self.connection_manager); let future = async move { @@ -2096,6 +2115,9 @@ where "Onion message intercepted, but no onion message mailbox available" ); } + self.liquidity_source + .lsps5_service() + .notify_onion_message_incoming(peer_node_id); } else { log_error!(self.logger, "Onion message intercepted for unknown SCID"); } diff --git a/src/lib.rs b/src/lib.rs index 9b0700b967..699763920b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -199,7 +199,10 @@ pub use types::{ #[cfg(feature = "storage-vss")] pub use vss_client; -use crate::config::{LIQUIDITY_DISCOVERY_RETRY_INITIAL_DELAY, LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY}; +use crate::config::{ + LIQUIDITY_DISCOVERY_RETRY_INITIAL_DELAY, LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY, + LSPS5_EXPIRY_CHECK_INTERVAL, LSPS5_INTERCEPT_POLL_INTERVAL, +}; use crate::ffi::{maybe_deref, maybe_wrap}; use crate::liquidity::Liquidity; use crate::scoring::setup_background_pathfinding_scores_sync; @@ -849,6 +852,53 @@ impl Node { } }); + // Regularly notify offline LSPS5 clients about HTLCs approaching their expiry. + if self.liquidity_source.liquidity_manager().lsps5_service_handler().is_some() { + let expiry_liquidity_source = Arc::clone(&self.liquidity_source); + let expiry_liquidity_logger = Arc::clone(&self.logger); + let mut stop_expiry = self.stop_sender.subscribe(); + self.runtime.spawn_cancellable_background_task(async move { + let mut interval = tokio::time::interval(LSPS5_EXPIRY_CHECK_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + _ = stop_expiry.changed() => { + log_debug!( + expiry_liquidity_logger, + "Stopping LSPS5 HTLC expiry checks." + ); + return; + } + _ = interval.tick() => { + expiry_liquidity_source.lsps5_service().check_expiring_htlcs(); + } + } + } + }); + } + + // Forward or fail HTLCs we're holding for offline LSPS5 clients while we wake them. + if self.liquidity_source.liquidity_manager().lsps5_service_handler().is_some() { + let intercept_liquidity_source = Arc::clone(&self.liquidity_source); + let intercept_logger = Arc::clone(&self.logger); + let mut stop_intercepts = self.stop_sender.subscribe(); + self.runtime.spawn_cancellable_background_task(async move { + let mut interval = tokio::time::interval(LSPS5_INTERCEPT_POLL_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + _ = stop_intercepts.changed() => { + log_debug!(intercept_logger, "Stopping LSPS5 HTLC intercept sweeps."); + return; + } + _ = interval.tick() => { + intercept_liquidity_source.lsps5_service().sweep_pending_intercepts(); + } + } + } + }); + } + log_info!(self.logger, "Startup complete."); *is_running_lock = true; Ok(()) diff --git a/src/liquidity/client/lsps5.rs b/src/liquidity/client/lsps5.rs new file mode 100644 index 0000000000..3feea22ac0 --- /dev/null +++ b/src/liquidity/client/lsps5.rs @@ -0,0 +1,574 @@ +// 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; +use lightning::log_debug; +use lightning_liquidity::lsps0::ser::LSPSRequestId; +use lightning_liquidity::lsps5::event::LSPS5ClientEvent; +use lightning_liquidity::lsps5::msgs::{ + LSPS5Error, LSPS5ProtocolError, ListWebhooksResponse, RemoveWebhookResponse, SetWebhookResponse, +}; +use tokio::sync::oneshot; + +use crate::connection::ConnectionManager; +use crate::liquidity::service::lsps5::LSPS5ServiceLiquiditySource; +use crate::liquidity::{ + select_lsps_for_protocol, LspConfig, LspNode, PendingRequest, PendingRequestGuard, + LIQUIDITY_REQUEST_TIMEOUT_SECS, LSPS_DISCOVERY_WAIT_TIMEOUT_SECS, +}; +use crate::logger::{log_error, log_info, LdkLogger, Logger}; +use crate::runtime::Runtime; +use crate::types::LiquidityManager; +use crate::Error; + +pub(crate) struct LSPS5Client +where + L::Target: LdkLogger, +{ + pub(crate) lsp_nodes: Arc>>, + pub(crate) pending_set_webhook_requests: + Mutex>>>, + pub(crate) pending_list_webhooks_requests: + Mutex>>>, + pub(crate) pending_remove_webhook_requests: + Mutex>>>, + pub(crate) discovery_done_rx: tokio::sync::watch::Receiver, + pub(crate) liquidity_manager: Arc, + pub(crate) logger: L, +} + +impl LSPS5Client +where + L::Target: LdkLogger, +{ + pub(crate) async fn lsps5_set_webhook( + &self, app_name: String, webhook_url: String, node_id: Option<&PublicKey>, + ) -> Result { + let lsps5_node = select_lsps_for_protocol(&self.lsp_nodes, 5, node_id) + .ok_or(Error::LiquiditySourceUnavailable)?; + let client_handler = self.liquidity_manager.lsps5_client_handler().ok_or_else(|| { + log_error!(self.logger, "LSPS5 liquidity client was not configured."); + Error::LiquiditySourceUnavailable + })?; + + let (sender, receiver) = oneshot::channel(); + let _pending_request = { + let mut pending_set_webhook_requests_lock = + self.pending_set_webhook_requests.lock().expect("lock"); + + let request_id = match client_handler.set_webhook( + lsps5_node.node_id, + app_name.clone(), + webhook_url.clone(), + ) { + Ok(request_id) => request_id, + Err(e) => { + log_error!( + self.logger, + "Failed to send set webhook request to liquidity service: {:?}", + e + ); + return Err(self.map_lsps5_error(&e, Error::LiquiditySetWebhookFailed)); + }, + }; + + PendingRequestGuard::insert( + &self.pending_set_webhook_requests, + &mut pending_set_webhook_requests_lock, + request_id, + sender, + ) + }; + + match tokio::time::timeout(Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), receiver) + .await + { + Ok(Ok(result)) => result.map(LSPS5SetWebhookResponse::from).map_err(|e| { + log_error!(self.logger, "Failed to set webhook: {:?}", e); + self.map_lsps5_error(&e, Error::LiquiditySetWebhookFailed) + }), + Ok(Err(e)) => { + log_error!( + self.logger, + "Failed to handle response from liquidity service: {:?}", + e + ); + Err(Error::LiquidityRequestFailed) + }, + Err(e) => { + log_error!(self.logger, "Liquidity request timed out: {}", e); + Err(Error::LiquidityRequestFailed) + }, + } + } + + pub(crate) async fn lsps5_list_webhooks( + &self, node_id: Option<&PublicKey>, + ) -> Result { + let lsps5_node = select_lsps_for_protocol(&self.lsp_nodes, 5, node_id) + .ok_or(Error::LiquiditySourceUnavailable)?; + let client_handler = self.liquidity_manager.lsps5_client_handler().ok_or_else(|| { + log_error!(self.logger, "LSPS5 liquidity client was not configured."); + Error::LiquiditySourceUnavailable + })?; + + let (sender, receiver) = oneshot::channel(); + let _pending_request = { + let mut pending_list_webhooks_requests_lock = + self.pending_list_webhooks_requests.lock().expect("lock"); + let request_id = client_handler.list_webhooks(lsps5_node.node_id); + PendingRequestGuard::insert( + &self.pending_list_webhooks_requests, + &mut pending_list_webhooks_requests_lock, + request_id, + sender, + ) + }; + + match tokio::time::timeout(Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), receiver) + .await + { + Ok(Ok(result)) => result.map(LSPS5ListWebhooksResponse::from).map_err(|e| { + log_error!(self.logger, "Failed to list webhooks: {:?}", e); + self.map_lsps5_error(&e, Error::LiquidityListWebhooksFailed) + }), + Ok(Err(e)) => { + log_error!( + self.logger, + "Failed to handle response from liquidity service: {:?}", + e + ); + Err(Error::LiquidityRequestFailed) + }, + Err(e) => { + log_error!(self.logger, "Liquidity request timed out: {}", e); + Err(Error::LiquidityRequestFailed) + }, + } + } + + pub(crate) async fn lsps5_remove_webhook( + &self, app_name: String, node_id: Option<&PublicKey>, + ) -> Result<(), Error> { + let lsps5_node = select_lsps_for_protocol(&self.lsp_nodes, 5, node_id) + .ok_or(Error::LiquiditySourceUnavailable)?; + let client_handler = self.liquidity_manager.lsps5_client_handler().ok_or_else(|| { + log_error!(self.logger, "LSPS5 liquidity client was not configured."); + Error::LiquiditySourceUnavailable + })?; + + let (sender, receiver) = oneshot::channel(); + let _pending_request = { + let mut pending_remove_webhook_requests_lock = + self.pending_remove_webhook_requests.lock().expect("lock"); + let request_id = + match client_handler.remove_webhook(lsps5_node.node_id, app_name.clone()) { + Ok(request_id) => request_id, + Err(e) => { + log_error!( + self.logger, + "Failed to send remove webhook request to liquidity service: {:?}", + e + ); + return Err(self.map_lsps5_error(&e, Error::LiquidityRemoveWebhookFailed)); + }, + }; + + PendingRequestGuard::insert( + &self.pending_remove_webhook_requests, + &mut pending_remove_webhook_requests_lock, + request_id, + sender, + ) + }; + + match tokio::time::timeout(Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), receiver) + .await + { + Ok(Ok(result)) => result.map(|_| ()).map_err(|e| { + log_error!(self.logger, "Failed to remove webhook: {:?}", e); + self.map_lsps5_error(&e, Error::LiquidityRemoveWebhookFailed) + }), + Ok(Err(e)) => { + log_error!( + self.logger, + "Failed to handle response from liquidity service: {:?}", + e + ); + Err(Error::LiquidityRequestFailed) + }, + Err(e) => { + log_error!(self.logger, "Liquidity request timed out: {}", e); + Err(Error::LiquidityRequestFailed) + }, + } + } + + pub(crate) async fn handle_event(&self, event: LSPS5ClientEvent) { + match event { + LSPS5ClientEvent::WebhookRegistered { + request_id, + counterparty_node_id, + num_webhooks, + max_webhooks, + no_change, + .. + } => { + if !self.is_expected_counterparty( + &counterparty_node_id, + "LSPS5Client::WebhookRegistered".into(), + ) { + return; + } + + let response = Ok(SetWebhookResponse { num_webhooks, max_webhooks, no_change }); + self.deliver_response(&self.pending_set_webhook_requests, &request_id, response); + }, + LSPS5ClientEvent::WebhookRegistrationFailed { + request_id, + counterparty_node_id, + error, + app_name, + url, + } => { + if !self.is_expected_counterparty( + &counterparty_node_id, + "LSPS5Client::WebhookRegistrationFailed".into(), + ) { + return; + } + + log_error!( + self.logger, + "Webhook registration failed for app '{}' with url '{}': {:?}", + app_name.as_str(), + url.as_str(), + error + ); + self.deliver_response(&self.pending_set_webhook_requests, &request_id, Err(error)); + }, + LSPS5ClientEvent::WebhooksListed { + request_id, + counterparty_node_id, + app_names, + max_webhooks, + } => { + if !self.is_expected_counterparty( + &counterparty_node_id, + "LSPS5Client::WebhooksListed".into(), + ) { + return; + } + + let response = Ok(ListWebhooksResponse { app_names, max_webhooks }); + self.deliver_response(&self.pending_list_webhooks_requests, &request_id, response); + }, + LSPS5ClientEvent::WebhookRemoved { request_id, counterparty_node_id, .. } => { + if !self.is_expected_counterparty( + &counterparty_node_id, + "LSPS5Client::WebhookRemoved".into(), + ) { + return; + } + + self.deliver_response( + &self.pending_remove_webhook_requests, + &request_id, + Ok(RemoveWebhookResponse {}), + ); + }, + LSPS5ClientEvent::WebhookRemovalFailed { + request_id, + counterparty_node_id, + error, + app_name, + } => { + if !self.is_expected_counterparty( + &counterparty_node_id, + "LSPS5Client::WebhookRemovalFailed".into(), + ) { + return; + } + + log_error!( + self.logger, + "Webhook removal failed for app '{}': {:?}", + app_name.as_str(), + error + ); + self.deliver_response( + &self.pending_remove_webhook_requests, + &request_id, + Err(error), + ); + }, + } + } + + fn map_lsps5_error(&self, e: &LSPS5Error, fallback: Error) -> Error { + match e { + LSPS5Error::Protocol(LSPS5ProtocolError::TooManyWebhooks) => { + Error::LiquidityWebhookLimitExceeded + }, + LSPS5Error::Protocol(LSPS5ProtocolError::NoPriorActivityError) => { + Error::LiquidityWebhookNoPriorActivity + }, + LSPS5Error::Protocol(LSPS5ProtocolError::AppNameNotFound) => { + Error::LiquidityWebhookAppNameNotFound + }, + LSPS5Error::Protocol( + LSPS5ProtocolError::AppNameTooLong + | LSPS5ProtocolError::WebhookUrlTooLong + | LSPS5ProtocolError::UrlParse + | LSPS5ProtocolError::UnsupportedProtocol, + ) => Error::LiquidityWebhookInvalid, + _ => fallback, + } + } + + fn is_expected_counterparty(&self, counterparty_node_id: &PublicKey, event: String) -> bool { + if self.lsp_nodes.read().expect("lock").iter().any(|n| n.node_id == *counterparty_node_id) { + true + } else { + log_error!(self.logger, "Received unexpected {} event!", event); + false + } + } + + fn deliver_response( + &self, pending: &Mutex>>, + request_id: &LSPSRequestId, response: T, + ) { + match pending.lock().expect("lock").remove(request_id) { + Some(request) => { + if request.sender.send(response).is_err() { + log_error!( + self.logger, + "Failed to handle response for request {:?} from liquidity service", + request_id + ); + } + }, + None => { + log_error!( + self.logger, + "Received response from liquidity service for unknown request." + ); + }, + } + } + + async fn get_lsps5_node( + &self, override_node_id: Option<&PublicKey>, + ) -> Result { + if let Some(node) = select_lsps_for_protocol(&self.lsp_nodes, 5, override_node_id) { + return Ok(node); + } + + let has_undiscovered_protocol = + self.lsp_nodes.read().expect("lock").iter().any(|n| n.supported_protocols.is_none()); + + // LSP protocol discovery may still be in flight, we wait briefly for it to finish, then re-check. + if has_undiscovered_protocol && !*self.discovery_done_rx.borrow() { + log_debug!( + self.logger, + "No LSPS5 node available yet, waiting for protocol discovery to complete." + ); + let mut rx = self.discovery_done_rx.clone(); + let _ = tokio::time::timeout( + Duration::from_secs(LSPS_DISCOVERY_WAIT_TIMEOUT_SECS), + rx.wait_for(|done| *done), + ) + .await; + } + + select_lsps_for_protocol(&self.lsp_nodes, 5, override_node_id) + .ok_or(Error::LiquiditySourceUnavailable) + } +} + +/// The response to a [`LSPS5Liquidity::set_webhook`] request. +#[derive(Debug, Clone)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct LSPS5SetWebhookResponse { + /// The current number of webhooks registered for this client. + pub num_webhooks: u32, + /// The maximum number of webhooks allowed by the LSP. + pub max_webhooks: u32, + /// Whether this was an unchanged registration (same `app_name` and URL). + pub no_change: bool, +} + +impl From for LSPS5SetWebhookResponse { + fn from(response: SetWebhookResponse) -> Self { + Self { + num_webhooks: response.num_webhooks, + max_webhooks: response.max_webhooks, + no_change: response.no_change, + } + } +} + +/// The response to a [`LSPS5Liquidity::list_webhooks`] request. +#[derive(Debug, Clone)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct LSPS5ListWebhooksResponse { + /// The app names with a currently registered webhook. + pub app_names: Vec, + /// The maximum number of webhooks allowed by the LSP. + pub max_webhooks: u32, +} + +impl From for LSPS5ListWebhooksResponse { + fn from(response: ListWebhooksResponse) -> Self { + Self { + app_names: response + .app_names + .into_iter() + .map(|name| name.as_str().to_string()) + .collect(), + max_webhooks: response.max_webhooks, + } + } +} + +/// A liquidity handler for managing LSPS5 webhook notifications. +/// +/// Should be retrieved by calling [`Liquidity::lsps5`]. +/// +/// On the client side, this handler allows registering webhook endpoints with an LSP to receive +/// push notifications for Lightning events while offline. On the service side, it allows notifying +/// clients about such events. +/// +/// [bLIP-55 / LSPS5]: https://github.com/lightning/blips/blob/master/blip-0055.md +/// [`Liquidity::lsps5`]: crate::Liquidity::lsps5 +#[derive(Clone)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Object))] +pub struct LSPS5Liquidity { + runtime: Arc, + connection_manager: Arc>>, + liquidity_source: Arc>>, + lsps5_service: Arc>>, + logger: Arc, +} + +impl LSPS5Liquidity { + pub(crate) fn new( + runtime: Arc, connection_manager: Arc>>, + liquidity_source: Arc>>, + lsps5_service: Arc>>, logger: Arc, + ) -> Self { + Self { runtime, connection_manager, liquidity_source, lsps5_service, logger } + } +} + +#[cfg_attr(feature = "uniffi", uniffi::export)] +impl LSPS5Liquidity { + /// Registers a webhook URL with a single LSP for receiving LSPS5 notifications. + /// + /// The webhook will receive signed push notifications for Lightning events such as incoming + /// payments while the client is offline. + /// + /// Webhooks are stored per LSP, so this has to be called once per LSP that should be able to + /// reach the client. Give each LSP a distinct URL: bLIP-55 expects the notification delivery + /// service to verify the `x-lsps5-signature` header against the node ID of the LSP that URL + /// belongs to. + /// + /// Note that LSPs typically reject registrations from clients they have no prior relationship + /// with. LDK Node's own LSPS5 service requires an open channel, or an in-flight LSPS1 or LSPS2 + /// flow, and otherwise fails with [`Error::LiquidityWebhookNoPriorActivity`]. + pub fn set_webhook( + &self, app_name: String, webhook_url: String, node_id: PublicKey, + ) -> Result { + let lsps5_node = self + .runtime + .block_on(async { self.liquidity_source.get_lsps5_node(Some(&node_id)).await })?; + + self.connect(&lsps5_node)?; + + let liquidity_source = Arc::clone(&self.liquidity_source); + self.runtime.block_on(async move { + liquidity_source + .lsps5_set_webhook(app_name, webhook_url, Some(&lsps5_node.node_id)) + .await + }) + } + + /// Lists all webhooks currently registered with the given LSP. + /// + /// Webhooks are stored per LSP, so this returns only the `app_name`s registered with `node_id`. + pub fn list_webhooks(&self, node_id: PublicKey) -> Result { + let lsps5_node = self + .runtime + .block_on(async { self.liquidity_source.get_lsps5_node(Some(&node_id)).await })?; + + self.connect(&lsps5_node)?; + + let liquidity_source = Arc::clone(&self.liquidity_source); + self.runtime.block_on(async move { + liquidity_source.lsps5_list_webhooks(Some(&lsps5_node.node_id)).await + }) + } + + /// Removes a previously-configured webhook from a single LSP. + /// + /// Returns an error if `node_id` is not a configured LSP supporting LSPS5, or if the LSP rejected + /// the removal. + pub fn remove_webhook(&self, app_name: String, node_id: PublicKey) -> Result<(), Error> { + let lsps5_node = self + .runtime + .block_on(async { self.liquidity_source.get_lsps5_node(Some(&node_id)).await })?; + + self.connect(&lsps5_node)?; + + let liquidity_source = Arc::clone(&self.liquidity_source); + self.runtime.block_on(async move { + liquidity_source.lsps5_remove_webhook(app_name, Some(&lsps5_node.node_id)).await + }) + } + + /// Notifies a client that we intend to manage the liquidity on their channels. + /// + /// Should be called by LSP operators when their own policy decides to reclaim or adjust + /// liquidity for `client_node_id` (e.g. before closing or splicing a channel), so the client + /// has a chance to come online and cooperate. Sends a notification to every webhook the + /// client has registered with us. + /// + /// Note that notifications are rate limited per client. Calling this repeatedly in quick + /// succession returns [`Error::LiquidityNotifyWebhookRateLimited`], and the call may succeed + /// once the LSP's cooldown period has elapsed. + /// + /// If the client is currently connected to us, no notification is sent and this returns + /// `Ok(())`, as there's no need to wake them. + pub fn notify_liquidity_management_request( + &self, client_node_id: PublicKey, + ) -> Result<(), Error> { + self.lsps5_service.notify_liquidity_management_request(client_node_id) + } +} + +impl LSPS5Liquidity { + fn connect(&self, lsps5_node: &LspConfig) -> Result<(), Error> { + let con_node_id = lsps5_node.node_id; + let con_addr = lsps5_node.address.clone(); + let con_cm = Arc::clone(&self.connection_manager); + + // We need to use our main runtime here as a local runtime might not be around to poll + // connection futures going forward. + self.runtime.block_on(async move { + con_cm.connect_peer_if_necessary(con_node_id, con_addr).await + })?; + + log_info!(self.logger, "Connected to LSP {}@{}. ", lsps5_node.node_id, lsps5_node.address); + Ok(()) + } +} diff --git a/src/liquidity/client/mod.rs b/src/liquidity/client/mod.rs index 52fad2da20..f208eb1932 100644 --- a/src/liquidity/client/mod.rs +++ b/src/liquidity/client/mod.rs @@ -7,5 +7,6 @@ pub(crate) mod lsps1; pub(crate) mod lsps2; +pub(crate) mod lsps5; pub use lsps1::LSPS1OrderStatus; diff --git a/src/liquidity/mod.rs b/src/liquidity/mod.rs index ffc1f878bc..3ba1c2fca4 100644 --- a/src/liquidity/mod.rs +++ b/src/liquidity/mod.rs @@ -18,6 +18,7 @@ use std::time::Duration; use bitcoin::secp256k1::PublicKey; pub use client::lsps1::LSPS1Liquidity; +pub use client::lsps5::{LSPS5Liquidity, LSPS5ListWebhooksResponse, LSPS5SetWebhookResponse}; pub use client::LSPS1OrderStatus; use lightning::ln::msgs::SocketAddress; use lightning_liquidity::events::LiquidityEvent; @@ -25,6 +26,8 @@ use lightning_liquidity::lsps0::event::LSPS0ClientEvent; use lightning_liquidity::lsps1::client::LSPS1ClientConfig as LdkLSPS1ClientConfig; use lightning_liquidity::lsps2::client::LSPS2ClientConfig as LdkLSPS2ClientConfig; use lightning_liquidity::lsps2::service::LSPS2ServiceConfig as LdkLSPS2ServiceConfig; +use lightning_liquidity::lsps5::client::LSPS5ClientConfig as LdkLSPS5ClientConfig; +pub use lightning_liquidity::lsps5::service::LSPS5ServiceConfig; use lightning_liquidity::{LiquidityClientConfig, LiquidityServiceConfig}; pub use service::lsps2::LSPS2ServiceConfig; use tokio::sync::oneshot; @@ -33,7 +36,9 @@ use crate::builder::BuildError; use crate::connection::ConnectionManager; use crate::liquidity::client::lsps1::LSPS1Client; use crate::liquidity::client::lsps2::LSPS2Client; +use crate::liquidity::client::lsps5::LSPS5Client; use crate::liquidity::service::lsps2::{LSPS2Service, LSPS2ServiceLiquiditySource}; +use crate::liquidity::service::lsps5::LSPS5ServiceLiquiditySource; use crate::logger::{log_debug, log_error, log_info, LdkLogger, Logger}; use crate::runtime::Runtime; use crate::types::{Broadcaster, ChannelManager, DynStore, KeysManager, LiquidityManager, Wallet}; @@ -215,6 +220,23 @@ impl Liquidity { Arc::clone(&self.logger), ) } + + /// Returns a liquidity handler for managing webhook registrations via the [bLIP-55 / LSPS5] + /// protocol. + /// + /// This allows registering, listing, and removing webhook endpoints with an LSP in order to + /// receive push notifications for events such as incoming payments while the client is offline. + /// + /// [bLIP-55 / LSPS5]: https://github.com/lightning/blips/blob/master/blip-0055.md + pub fn lsps5(&self) -> LSPS5Liquidity { + LSPS5Liquidity::new( + Arc::clone(&self.runtime), + Arc::clone(&self.connection_manager), + self.liquidity_source.lsps5_client(), + self.liquidity_source.lsps5_service(), + Arc::clone(&self.logger), + ) + } } #[derive(Debug, Clone)] @@ -240,12 +262,15 @@ where { lsp_nodes: Vec, lsps2_service: Option, + lsps5_service: Option, + advertise_service: bool, wallet: Arc, channel_manager: Arc, keys_manager: Arc, tx_broadcaster: Arc, kv_store: Arc, config: Arc, + runtime: Arc, logger: L, } @@ -255,19 +280,25 @@ where { pub(crate) fn new( wallet: Arc, channel_manager: Arc, keys_manager: Arc, - tx_broadcaster: Arc, kv_store: Arc, config: Arc, logger: L, + tx_broadcaster: Arc, kv_store: Arc, config: Arc, + runtime: Arc, logger: L, ) -> Self { let lsp_nodes = Vec::new(); let lsps2_service = None; + let lsps5_service = None; + let advertise_service = false; Self { lsp_nodes, lsps2_service, + lsps5_service, + advertise_service, wallet, channel_manager, keys_manager, tx_broadcaster, kv_store, config, + runtime, logger, } } @@ -285,18 +316,32 @@ where self } + pub(crate) fn lsps5_service(&mut self, service_config: LSPS5ServiceConfig) -> &mut Self { + self.lsps5_service = Some(service_config); + self + } + + pub(crate) fn set_advertise_service(&mut self, advertise_service: bool) -> &mut Self { + self.advertise_service = advertise_service; + self + } + pub(crate) async fn build(self) -> Result, BuildError> { - 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; - let advertise_service = s.service_config.advertise_service; - LiquidityServiceConfig { - lsps1_service_config: None, - lsps2_service_config, - lsps5_service_config, - advertise_service, - } - }); + let lsps2_service_config = + self.lsps2_service.as_ref().map(|s| s.ldk_service_config.clone()); + let lsps5_service_config = self.lsps5_service.clone(); + + let liquidity_service_config = + if lsps2_service_config.is_some() || lsps5_service_config.is_some() { + Some(LiquidityServiceConfig { + lsps1_service_config: None, + lsps2_service_config, + lsps5_service_config, + advertise_service: self.advertise_service, + }) + } else { + None + }; let (discovery_done_tx, discovery_done_rx) = tokio::sync::watch::channel(false); @@ -305,7 +350,7 @@ where let liquidity_client_config = Some(LiquidityClientConfig { lsps1_client_config: Some(LdkLSPS1ClientConfig { max_channel_fees_msat: None }), lsps2_client_config: Some(LdkLSPS2ClientConfig {}), - lsps5_client_config: None, + lsps5_client_config: Some(LdkLSPS5ClientConfig {}), }); let liquidity_manager = Arc::new( @@ -360,13 +405,31 @@ where lsps2_service: Arc::new(LSPS2ServiceLiquiditySource { lsps2_service: self.lsps2_service, wallet: self.wallet, - channel_manager: self.channel_manager, + channel_manager: self.channel_manager.clone(), peer_manager: RwLock::new(None), keys_manager: self.keys_manager, liquidity_manager: Arc::clone(&liquidity_manager), config: self.config.clone(), logger: self.logger.clone(), }), + lsps5_client: Arc::new(LSPS5Client { + lsp_nodes: Arc::clone(&lsp_nodes), + pending_set_webhook_requests: Mutex::new(HashMap::new()), + pending_list_webhooks_requests: Mutex::new(HashMap::new()), + pending_remove_webhook_requests: Mutex::new(HashMap::new()), + discovery_done_rx: discovery_done_rx.clone(), + liquidity_manager: Arc::clone(&liquidity_manager), + logger: self.logger.clone(), + }), + lsps5_service: Arc::new(LSPS5ServiceLiquiditySource { + liquidity_manager: Arc::clone(&liquidity_manager), + channel_manager: self.channel_manager, + peer_manager: RwLock::new(None), + expiry_notified_at: Mutex::new(HashMap::new()), + runtime: Arc::clone(&self.runtime), + logger: self.logger.clone(), + pending_intercepts: Mutex::new(HashMap::new()), + }), pending_lsps0_discovery: Mutex::new(HashMap::new()), discovery_done_tx, discovery_done_rx, @@ -384,6 +447,8 @@ where lsps1_client: Arc>, lsps2_client: Arc>, lsps2_service: Arc>, + lsps5_client: Arc>, + lsps5_service: Arc>, pending_lsps0_discovery: Mutex>>>, discovery_done_tx: tokio::sync::watch::Sender, discovery_done_rx: tokio::sync::watch::Receiver, @@ -411,11 +476,24 @@ where Arc::clone(&self.lsps2_service) } - pub(crate) async fn handle_next_event(&self) { + pub(crate) fn lsps5_client(&self) -> Arc> { + Arc::clone(&self.lsps5_client) + } + + pub(crate) fn lsps5_service(&self) -> Arc> { + Arc::clone(&self.lsps5_service) + } + + pub(crate) async fn handle_next_event(&self) + where + L: Clone + Send + Sync + 'static, + { match self.liquidity_manager.next_event_async().await { LiquidityEvent::LSPS1Client(event) => self.lsps1_client.handle_event(event).await, LiquidityEvent::LSPS2Client(event) => self.lsps2_client.handle_event(event).await, LiquidityEvent::LSPS2Service(event) => self.lsps2_service.handle_event(event).await, + LiquidityEvent::LSPS5Client(event) => self.lsps5_client.handle_event(event).await, + LiquidityEvent::LSPS5Service(event) => self.lsps5_service.handle_event(event).await, LiquidityEvent::LSPS0Client(LSPS0ClientEvent::ListProtocolsResponse { counterparty_node_id, diff --git a/src/liquidity/service/lsps2.rs b/src/liquidity/service/lsps2.rs index 946511c5d9..37cf8a5295 100644 --- a/src/liquidity/service/lsps2.rs +++ b/src/liquidity/service/lsps2.rs @@ -60,8 +60,6 @@ pub struct LSPS2ServiceConfig { /// /// If set, only requests matching this token will be accepted. pub require_token: Option, - /// Indicates whether the LSPS service will be announced via the gossip network. - pub advertise_service: bool, /// The fee we withhold for the channel open from the initial payment. /// /// This fee is proportional to the client-requested amount, in parts-per-million. diff --git a/src/liquidity/service/lsps5.rs b/src/liquidity/service/lsps5.rs new file mode 100644 index 0000000000..491fbb80a9 --- /dev/null +++ b/src/liquidity/service/lsps5.rs @@ -0,0 +1,441 @@ +// 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, Weak}; +use std::time::Instant; + +use bitcoin::secp256k1::PublicKey; +use lightning::chain::channelmonitor::HTLC_FAIL_BACK_BUFFER; +use lightning::ln::channel_state::InboundHTLCStateDetails; +use lightning::ln::channelmanager::InterceptId; +use lightning::ln::types::ChannelId; +use lightning_liquidity::lsps5::event::LSPS5ServiceEvent; +use lightning_liquidity::lsps5::msgs::LSPS5ProtocolError; + +use crate::config::{ + LSPS5_EXPIRY_NOTIFICATION_THRESHOLD_BLOCKS, LSPS5_EXPIRY_RENOTIFY_INTERVAL_BLOCKS, + LSPS5_INTERCEPT_HOLD_TIMEOUT, LSPS5_WEBHOOK_MAX_RESPONSE_SIZE, LSPS5_WEBHOOK_TIMEOUT_SECS, +}; +use crate::logger::{log_debug, log_error, log_info, LdkLogger}; +use crate::runtime::Runtime; +use crate::types::{ChannelManager, LiquidityManager, PeerManager}; +use crate::Error; + +#[derive(Clone, Copy)] +pub(crate) struct PendingIntercept { + client_id: PublicKey, + channel_id: ChannelId, + amt_to_forward_msat: u64, + deadline: Instant, +} + +pub(crate) struct LSPS5ServiceLiquiditySource +where + L::Target: LdkLogger, +{ + pub(crate) liquidity_manager: Arc, + pub(crate) channel_manager: Arc, + pub(crate) peer_manager: RwLock>>, + pub(crate) expiry_notified_at: Mutex>, + pub(crate) runtime: Arc, + pub(crate) logger: L, + pub(crate) pending_intercepts: Mutex>, +} + +impl LSPS5ServiceLiquiditySource +where + L::Target: LdkLogger, +{ + pub(crate) fn set_peer_manager(&self, peer_manager: Weak) { + *self.peer_manager.write().expect("lock") = Some(peer_manager); + } + + /// Notifies a client that a payment is incoming for them. + /// + /// The notification is sent while we hold the HTLC for them. + pub(crate) fn notify_payment_incoming(&self, client_id: PublicKey) { + let Some(handler) = self.liquidity_manager.lsps5_service_handler() else { + return; + }; + + if !self.is_client_offline(&client_id) { + return; + } + + handler.notify_payment_incoming(client_id).unwrap_or_else(|e| match e { + LSPS5ProtocolError::SlowDownError => log_debug!( + self.logger, + "Skipping payment incoming notification for client {}: rate limited.", + client_id + ), + _ => log_error!( + self.logger, + "Failed to notify payment incoming for client {}: {:?}", + client_id, + e + ), + }) + } + + pub(crate) fn notify_liquidity_management_request( + &self, client_id: PublicKey, + ) -> Result<(), Error> { + let Some(handler) = self.liquidity_manager.lsps5_service_handler() else { + return Err(Error::LiquiditySourceUnavailable); + }; + + if !self.is_client_offline(&client_id) { + log_debug!( + self.logger, + "Skipping liquidity management request notification for client {}: already connected.", + client_id + ); + return Ok(()); + } + + handler.notify_liquidity_management_request(client_id).map_err(|e| match e { + LSPS5ProtocolError::SlowDownError => { + log_debug!( + self.logger, + "Skipping liquidity management request notification for client {}: rate limited.", + client_id + ); + Error::LiquidityNotifyWebhookRateLimited + }, + _ => { + log_error!( + self.logger, + "Failed to notify liquidity management request for client {}: {:?}", + client_id, + e + ); + Error::LiquidityNotifyWebhookFailed + }, + }) + } + + /// Notifies a client that onion messages are waiting for them. + /// + /// The messages are held in our onion message mailbox and delivered once the client + /// reconnects. + pub(crate) fn notify_onion_message_incoming(&self, client_id: PublicKey) { + let Some(handler) = self.liquidity_manager.lsps5_service_handler() else { + return; + }; + + if !self.is_client_offline(&client_id) { + return; + } + + handler.notify_onion_message_incoming(client_id).unwrap_or_else(|e| match e { + LSPS5ProtocolError::SlowDownError => log_debug!( + self.logger, + "Skipping onion message incoming notification for client {}: rate limited.", + client_id + ), + _ => log_error!( + self.logger, + "Failed to notify onion message incoming for client {}: {:?}", + client_id, + e + ), + }) + } + + fn notify_expiry_soon(&self, client_id: PublicKey, timeout: u32) -> bool { + let Some(handler) = self.liquidity_manager.lsps5_service_handler() else { + return false; + }; + + if !self.is_client_offline(&client_id) { + return false; + } + + match handler.notify_expiry_soon(client_id, timeout) { + Ok(()) => true, + Err(LSPS5ProtocolError::SlowDownError) => { + log_debug!( + self.logger, + "Skipping expiry soon notification for client {}: rate limited.", + client_id + ); + false + }, + Err(e) => { + log_error!( + self.logger, + "Failed to notify expiry soon for client {}: {:?}", + client_id, + e + ); + false + }, + } + } + + /// Notifies offline clients about pending HTLCs that are approaching the height at which we'd + /// be forced to close their channel. + /// + /// A client that doesn't come online and settle an HTLC in time costs us the channel, as LDK + /// has to enforce it on-chain. We notify ahead of that deadline so the client has a chance to + /// come online and settle cooperatively instead. + pub(crate) fn check_expiring_htlcs(&self) { + if self.liquidity_manager.lsps5_service_handler().is_none() { + return; + } + + let current_height = self.channel_manager.current_best_block().height; + + // Notifications are rate limited per client rather than per channel, so we collapse a + // client's channels to their single most urgent deadline. + let mut deadlines: HashMap = HashMap::new(); + for channel in self.channel_manager.list_channels() { + // LDK force-closes `LATENCY_GRACE_PERIOD_BLOCKS` after an outbound HTLC's + // `cltv_expiry`, and `CLTV_CLAIM_BUFFER` before an inbound one's. Neither is exported, + // so we use `HTLC_FAIL_BACK_BUFFER` instead, which notifies a few blocks early. + let outbound = channel.pending_outbound_htlcs.iter().map(|htlc| htlc.cltv_expiry); + + // We only force-close over inbound HTLCs we know the preimage for. + let inbound = channel + .pending_inbound_htlcs + .iter() + .filter(|htlc| { + htlc.state == Some(InboundHTLCStateDetails::AwaitingRemoteRevokeToRemoveFulfill) + }) + .map(|htlc| htlc.cltv_expiry.saturating_sub(HTLC_FAIL_BACK_BUFFER)); + + for deadline in outbound.chain(inbound) { + if deadline.saturating_sub(current_height) + > LSPS5_EXPIRY_NOTIFICATION_THRESHOLD_BLOCKS + { + continue; + } + + deadlines + .entry(channel.counterparty.node_id) + .and_modify(|current| *current = (*current).min(deadline)) + .or_insert(deadline); + } + } + + let mut notified_at = self.expiry_notified_at.lock().expect("lock"); + + // Forget clients whose HTLCs resolved, so a later one notifies immediately again. + notified_at.retain(|client_id, _| deadlines.contains_key(client_id)); + + for (client_id, timeout) in deadlines { + let is_due = notified_at.get(&client_id).is_none_or(|last_height| { + current_height.saturating_sub(*last_height) >= LSPS5_EXPIRY_RENOTIFY_INTERVAL_BLOCKS + }); + + if is_due && self.notify_expiry_soon(client_id, timeout) { + notified_at.insert(client_id, current_height); + } + } + } + + /// Resolves the SCID a sender asked us to forward over to one of our channels. + /// + /// Returns `None` for SCIDs that aren't ours, in particular the intercept SCIDs LSPS2 hands + /// out. + pub(crate) fn resolve_client_channel(&self, scid: u64) -> Option<(PublicKey, ChannelId)> { + self.liquidity_manager.lsps5_service_handler()?; + + self.channel_manager.list_channels().into_iter().find_map(|channel| { + let matches = channel.short_channel_id == Some(scid) + || channel.inbound_scid_alias == Some(scid) + || channel.outbound_scid_alias == Some(scid); + matches.then(|| (channel.counterparty.node_id, channel.channel_id)) + }) + } + + /// Holds an HTLC intercepted for an offline client while we try to wake them. + /// + /// We notify the client's webhooks and then wait for them to connect and their channel to + /// become usable, forwarding as soon as it is. If they don't make it within + /// [`LSPS5_INTERCEPT_HOLD_TIMEOUT`], we fail the HTLC back so the sender can retry against a + /// client that is by then hopefully online. + pub(crate) fn handle_htlc_intercepted( + &self, client_id: PublicKey, channel_id: ChannelId, intercept_id: InterceptId, + amt_to_forward_msat: u64, + ) { + if self.liquidity_manager.lsps5_service_handler().is_none() { + self.fail_intercepted_htlc(intercept_id); + return; + } + + self.pending_intercepts.lock().expect("lock").insert( + intercept_id, + PendingIntercept { + client_id, + channel_id, + amt_to_forward_msat, + deadline: Instant::now() + LSPS5_INTERCEPT_HOLD_TIMEOUT, + }, + ); + + self.notify_payment_incoming(client_id); + } + + fn is_client_offline(&self, client_id: &PublicKey) -> bool { + match self.peer_manager.read().expect("lock").as_ref().and_then(|w| w.upgrade()) { + Some(pm) => pm.peer_by_node_id(client_id).is_none(), + None => { + log_debug!( + self.logger, + "No peer manager available, assuming client {} is offline.", + client_id + ); + true + }, + } + } + + /// Forwards HTLCs held for clients that came online, and fails those whose hold expired. + pub(crate) fn sweep_pending_intercepts(&self) { + let mut ready = Vec::new(); + let mut expired = Vec::new(); + + { + let mut pending = self.pending_intercepts.lock().expect("lock"); + if pending.is_empty() { + return; + } + + let now = Instant::now(); + pending.retain(|intercept_id, intercept| { + let is_usable = self + .channel_manager + .list_channels_with_counterparty(&intercept.client_id) + .iter() + .any(|c| c.channel_id == intercept.channel_id && c.is_usable); + + if is_usable { + ready.push((*intercept_id, *intercept)); + false + } else if now >= intercept.deadline { + expired.push((*intercept_id, intercept.client_id)); + false + } else { + true + } + }); + } + + for (intercept_id, intercept) in ready { + match self.channel_manager.forward_intercepted_htlc( + intercept_id, + &intercept.channel_id, + intercept.client_id, + intercept.amt_to_forward_msat, + ) { + Ok(()) => { + log_info!( + self.logger, + "LSPS5 client {} came online, forwarded the HTLC we held for them.", + intercept.client_id + ); + }, + Err(e) => { + // The channel may have stopped being usable between our check and this call, so + // we fail the HTLC rather than keep holding it. + log_error!( + self.logger, + "Failed to forward HTLC held for LSPS5 client {}: {:?}", + intercept.client_id, + e + ); + self.fail_intercepted_htlc(intercept_id); + }, + } + } + + for (intercept_id, client_id) in expired { + log_debug!( + self.logger, + "LSPS5 client {} did not come online in time, failing HTLC back.", + client_id + ); + self.fail_intercepted_htlc(intercept_id); + } + } + + fn fail_intercepted_htlc(&self, intercept_id: InterceptId) { + if let Err(e) = self.channel_manager.fail_intercepted_htlc(intercept_id) { + log_error!(self.logger, "Failed to fail back intercepted HTLC: {:?}", e); + } + } + + pub(crate) async fn handle_event(&self, event: LSPS5ServiceEvent) + where + L: Clone + Send + Sync + 'static, + { + match event { + LSPS5ServiceEvent::SendWebhookNotification { + counterparty_node_id: _, + app_name, + url, + notification, + headers, + } => { + if self.liquidity_manager.lsps5_service_handler().is_none() { + log_error!( + self.logger, + "Received unexpected LSPS5ServiceEvent::SendWebhookNotification event!" + ); + return; + } + + log_info!( + self.logger, + "Sending webhook notification for {}: {:?}", + app_name.as_str(), + notification + ); + + let notification_body = notification.to_request_body(); + let logger = self.logger.clone(); + + // `url` is client-supplied, so awaiting delivery here would let any client stall + // the liquidity event loop for the full timeout. Deliver out of band instead. + self.runtime.spawn_cancellable_background_task(async move { + let result = bitreq::post(url.as_str()) + .with_headers(headers) + .with_body(notification_body) + .with_timeout(LSPS5_WEBHOOK_TIMEOUT_SECS) + .with_max_redirects(0) + .with_max_body_size(LSPS5_WEBHOOK_MAX_RESPONSE_SIZE) + .send_async() + .await; + + match result { + Ok(response) => { + if response.status_code != 200 { + log_error!( + logger, + "Webhook call failed with status {} for {}", + response.status_code, + app_name.as_str(), + ); + } + }, + Err(e) => { + log_error!( + logger, + "Failed to send webhook notification for {}: {}", + app_name.as_str(), + e + ); + }, + } + }) + }, + } + } +} diff --git a/src/liquidity/service/mod.rs b/src/liquidity/service/mod.rs index cdbaf54265..9547d77fc4 100644 --- a/src/liquidity/service/mod.rs +++ b/src/liquidity/service/mod.rs @@ -6,3 +6,4 @@ // accordance with one or both of these licenses. pub(crate) mod lsps2; +pub(crate) mod lsps5; diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index 3da60800c6..9bbde8514f 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -51,10 +51,13 @@ use lightning::routing::gossip::{NodeAlias, NodeId}; use lightning::routing::router::RouteParametersConfig; use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore, PaginatedListResponse}; use lightning_invoice::{Bolt11InvoiceDescription, Description}; +use lightning_liquidity::lsps5::service::LSPS5ServiceConfig; use lightning_types::payment::{PaymentHash, PaymentPreimage}; use log::LevelFilter; use serde_json::json; +use crate::common::exponential_backoff_poll; + /// Waits until `node` has classified the funding broadcast `funding_txid` (a channel open or splice /// candidate) into a payment record carrying a `tx_type`. Classification runs off the broadcaster's /// queue, which can lag a `sync_wallets` call under load — and for a splice the counterparty also @@ -3808,7 +3811,6 @@ async fn do_lsps2_client_service_integration(client_trusts_lsp: bool) { let channel_over_provisioning_ppm = 100_000; let lsps2_service_config = LSPS2ServiceConfig { require_token: None, - advertise_service: false, channel_opening_fee_ppm, channel_over_provisioning_ppm, max_payment_size_msat: 1_000_000_000, @@ -3823,7 +3825,7 @@ async fn do_lsps2_client_service_integration(client_trusts_lsp: bool) { 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); + service_builder.enable_liquidity_provider(Some(lsps2_service_config), None, false); let service_node = service_builder.build(service_config.node_entropy.into()).unwrap(); service_node.start().unwrap(); @@ -4136,7 +4138,6 @@ async fn lsps2_client_trusts_lsp() { let channel_over_provisioning_ppm = 100_000; let lsps2_service_config = LSPS2ServiceConfig { require_token: None, - advertise_service: false, channel_opening_fee_ppm, channel_over_provisioning_ppm, max_payment_size_msat: 1_000_000_000, @@ -4151,7 +4152,7 @@ async fn lsps2_client_trusts_lsp() { 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); + service_builder.enable_liquidity_provider(Some(lsps2_service_config), None, false); let service_node = service_builder.build(service_config.node_entropy.into()).unwrap(); service_node.start().unwrap(); let service_node_id = service_node.node_id(); @@ -4313,7 +4314,6 @@ async fn lsps2_lsp_trusts_client_but_client_does_not_claim() { let channel_over_provisioning_ppm = 100_000; let lsps2_service_config = LSPS2ServiceConfig { require_token: None, - advertise_service: false, channel_opening_fee_ppm, channel_over_provisioning_ppm, max_payment_size_msat: 1_000_000_000, @@ -4328,7 +4328,7 @@ async fn lsps2_lsp_trusts_client_but_client_does_not_claim() { 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); + service_builder.enable_liquidity_provider(Some(lsps2_service_config), None, false); let service_node = service_builder.build(service_config.node_entropy.into()).unwrap(); service_node.start().unwrap(); @@ -5223,7 +5223,6 @@ async fn do_lsps2_multi_lsp_picks_cheapest(reverse_order: bool) { // Cheap LSP: 10_000 ppm. let cheap_cfg = LSPS2ServiceConfig { require_token: None, - advertise_service: false, channel_opening_fee_ppm: 10_000, channel_over_provisioning_ppm: 100_000, max_payment_size_msat: 1_000_000_000, @@ -5237,7 +5236,7 @@ async fn do_lsps2_multi_lsp_picks_cheapest(reverse_order: bool) { let cheap_node_config = random_config(); setup_builder!(cheap_builder, cheap_node_config.node_config); cheap_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); - cheap_builder.enable_liquidity_provider(cheap_cfg); + cheap_builder.enable_liquidity_provider(Some(cheap_cfg), None, false); let cheap = cheap_builder.build(cheap_node_config.node_entropy.into()).unwrap(); cheap.start().unwrap(); let cheap_id = cheap.node_id(); @@ -5246,7 +5245,6 @@ async fn do_lsps2_multi_lsp_picks_cheapest(reverse_order: bool) { // Expensive LSP: 20_000 ppm. let expensive_cfg = LSPS2ServiceConfig { require_token: None, - advertise_service: false, channel_opening_fee_ppm: 20_000, channel_over_provisioning_ppm: 200_000, max_payment_size_msat: 1_000_000_000, @@ -5260,7 +5258,7 @@ async fn do_lsps2_multi_lsp_picks_cheapest(reverse_order: bool) { let expensive_node_config = random_config(); setup_builder!(expensive_builder, expensive_node_config.node_config); expensive_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); - expensive_builder.enable_liquidity_provider(expensive_cfg); + expensive_builder.enable_liquidity_provider(Some(expensive_cfg), None, false); let expensive = expensive_builder.build(expensive_node_config.node_entropy.into()).unwrap(); expensive.start().unwrap(); let expensive_id = expensive.node_id(); @@ -5301,3 +5299,298 @@ async fn do_lsps2_multi_lsp_picks_cheapest(reverse_order: bool) { cheap.stop().unwrap(); expensive.stop().unwrap(); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn lsps5_webhook_registration() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let esplora_url = format!("http://{}", electrsd.esplora_url.as_ref().unwrap()); + let sync_config = EsploraSyncConfig::default(); + + // Setup LSPS5 service provider node + 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)); + let lsps5_service_config = LSPS5ServiceConfig { max_webhooks_per_client: 2 }; + service_builder.enable_liquidity_provider(None, Some(lsps5_service_config), false); + 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.onchain_payment().new_address().unwrap(); + let service_socket_addr = service_node.listening_addresses().unwrap().first().unwrap().clone(); + + // Setup LSPS5 client node + let client_config = random_config(); + 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_socket_addr.clone(), None, false); + let client_node = client_builder.build(client_config.node_entropy.into()).unwrap(); + client_node.start().unwrap(); + let client_node_id = client_node.node_id(); + let client_addr = client_node.onchain_payment().new_address().unwrap(); + + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![service_addr, client_addr], + Amount::from_sat(10_000_000), + ) + .await; + service_node.sync_wallets().unwrap(); + client_node.sync_wallets().unwrap(); + + open_channel(&client_node, &service_node, 5_000_000, false, &electrsd).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + service_node.sync_wallets().unwrap(); + client_node.sync_wallets().unwrap(); + expect_channel_ready_event!(client_node, service_node.node_id()); + expect_channel_ready_event!(service_node, client_node.node_id()); + + // Test webhook registration + let lsps5_client = client_node.liquidity().lsps5(); + let app_name_1 = "test-app".to_string(); + let webhook_url_1 = "https://example.com/webhook".to_string(); + + // Register first webhook + let response = lsps5_client + .set_webhook(app_name_1.clone(), webhook_url_1.clone(), service_node_id) + .expect("Failed to register webhook"); + assert_eq!(response.num_webhooks, 1, "Expected 1 webhook after first registration"); + assert_eq!(response.max_webhooks, 2, "Expected max_webhooks to be 2"); + assert!(!response.no_change, "Expected no_change to be false for new registration"); + + // Register second webhook with different app name + let app_name_2 = "test-app-2".to_string(); + let webhook_url_2 = "https://example.com/webhook-2".to_string(); + let response = lsps5_client + .set_webhook(app_name_2.clone(), webhook_url_2.clone(), service_node_id) + .expect("Failed to register second webhook"); + assert_eq!(response.num_webhooks, 2, "Expected 2 webhooks after second registration"); + assert_eq!(response.max_webhooks, 2, "Expected max_webhooks to be 2"); + assert!(!response.no_change, "Expected no_change to be false for new registration"); + + // Register the same webhook again - should return no_change=true + let response = lsps5_client + .set_webhook(app_name_2.clone(), webhook_url_2.clone(), service_node_id) + .expect("Failed to re-register webhook"); + assert_eq!(response.num_webhooks, 2, "Expected 2 webhooks after re-registering same webhook"); + assert_eq!(response.max_webhooks, 2, "Expected max_webhooks to be 2"); + assert!(response.no_change, "Expected no_change to be true for duplicate registration"); + + // Attempt to register a third webhook - should fail due to max_webhooks_per_client=2 + let app_name_3 = "test-app-3".to_string(); + let webhook_url_3 = "https://example.com/webhook-3".to_string(); + let response = + lsps5_client.set_webhook(app_name_3.clone(), webhook_url_3.clone(), service_node_id); + assert_eq!( + response.err(), + Some(NodeError::LiquidityWebhookLimitExceeded), + "Expected error when exceeding max webhooks" + ); + + // List registered webhooks + let registered_webhooks = + lsps5_client.list_webhooks(service_node_id).expect("Failed to list webhooks"); + assert_eq!(registered_webhooks.app_names.len(), 2, "Expected 2 registered webhooks"); + assert!( + registered_webhooks.app_names.iter().any(|name| name.as_str() == app_name_1), + "Expected app_name_1 in registered webhooks" + ); + assert!( + registered_webhooks.app_names.iter().any(|name| name.as_str() == app_name_2), + "Expected app_name_2 in registered webhooks" + ); + assert_eq!(registered_webhooks.max_webhooks, 2, "Expected max_webhooks to be 2"); + + // Attempt to delete non-existing webhook - should fail + let non_existing_app_name = "non-existing-app".to_string(); + let response = lsps5_client.remove_webhook(non_existing_app_name.clone(), service_node_id); + assert_eq!( + response.err(), + Some(NodeError::LiquidityWebhookAppNameNotFound), + "Expected error when removing non-existing webhook" + ); + + // Delete a registered webhook + lsps5_client + .remove_webhook(app_name_1.clone(), service_node_id) + .expect("Failed to delete first webhook"); + + // Verify webhook was deleted + let registered_webhooks = lsps5_client + .list_webhooks(service_node_id) + .expect("Failed to list webhooks after deletion"); + assert_eq!(registered_webhooks.app_names.len(), 1, "Expected 1 webhook after deletion"); + assert!( + registered_webhooks.app_names.iter().any(|name| name.as_str() == app_name_2), + "Expected app_name_2 to remain after deletion" + ); + assert!( + !registered_webhooks.app_names.iter().any(|name| name.as_str() == app_name_1), + "Expected app_name_1 to be removed" + ); + + // Requests targeting an LSP we don't know about should fail rather than silently fall back to + // another LSP. + let unknown_node_id = client_node_id; + assert_eq!( + lsps5_client.list_webhooks(unknown_node_id).err(), + Some(NodeError::LiquiditySourceUnavailable), + "Expected an error when targeting a node that is not a configured LSPS5 LSP" + ); + assert_eq!( + lsps5_client.set_webhook(app_name_1.clone(), webhook_url_1.clone(), unknown_node_id).err(), + Some(NodeError::LiquiditySourceUnavailable), + "Expected an error when targeting a node that is not a configured LSPS5 LSP" + ); + + // Test service-side notifications. + let lsps5_service = service_node.liquidity().lsps5(); + + lsps5_service + .notify_liquidity_management_request(client_node_id) + .expect("notify_liquidity_management_request failed"); + + // Notifications are rate limited per client, so an immediate second notification is rejected. + assert_eq!( + lsps5_service.notify_liquidity_management_request(client_node_id).err(), + Some(NodeError::LiquidityNotifyWebhookFailed), + "Expected the notification to be rate limited" + ); + + // A node that isn't running the LSPS5 service can't notify anyone. + assert_eq!( + client_node.liquidity().lsps5().notify_liquidity_management_request(service_node_id).err(), + Some(NodeError::LiquiditySourceUnavailable), + "Expected an error when notifying without a configured LSPS5 service" + ); + + service_node.stop().unwrap(); + client_node.stop().unwrap(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn lsps5_payment_incoming_notification() { + 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; + + // The notification leaves no trace in any store, so observe it through the service's logs. + let service_logger = Arc::new(CollectingLogWriter::new()); + 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.set_custom_logger(service_logger.clone()); + + // LSPS2 is enabled so the LSP accepts forwards to the client's unannounced channel. + let lsps2_service_config = LSPS2ServiceConfig { + require_token: None, + channel_opening_fee_ppm: 0, + 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: false, + disable_client_reserve: false, + }; + + service_builder.enable_liquidity_provider( + Some(lsps2_service_config), + Some(LSPS5ServiceConfig { max_webhooks_per_client: 2 }), + false, + ); + + 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_listening_addr = + service_node.listening_addresses().unwrap().first().unwrap().clone(); + + let client_config = random_config(); + 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_listening_addr, None, false); + let client_node = client_builder.build(client_config.node_entropy.into()).unwrap(); + client_node.start().unwrap(); + let client_node_id = client_node.node_id(); + + 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 service_addr = service_node.onchain_payment().new_address().unwrap(); + let client_addr = client_node.onchain_payment().new_address().unwrap(); + let payer_addr = payer_node.onchain_payment().new_address().unwrap(); + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![service_addr, client_addr, payer_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, false, &electrsd).await; + // Opened service -> client so the LSP has outbound liquidity towards the client, leaving the + // client being offline as the only reason the forward below can fail. + open_channel(&service_node, &client_node, 5_000_000, false, &electrsd).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + service_node.sync_wallets().unwrap(); + client_node.sync_wallets().unwrap(); + payer_node.sync_wallets().unwrap(); + expect_channel_ready_event!(payer_node, service_node_id); + expect_channel_ready_event!(client_node, service_node_id); + expect_channel_ready_events!(service_node, payer_node.node_id(), client_node_id); + + // The LSP only accepts registrations from clients it has prior activity with, which the channel + // opened above satisfies. + client_node + .liquidity() + .lsps5() + .set_webhook( + "test-app".to_string(), + "https://127.0.0.1:1/webhook".to_string(), + service_node_id, + ) + .expect("Failed to register webhook"); + + let invoice_description = + Bolt11InvoiceDescription::Direct(Description::new(String::from("lsps5")).unwrap()).into(); + let invoice = client_node + .bolt11_payment() + .receive(100_000_000, &invoice_description, 1024) + .expect("Failed to create invoice"); + + // Stopped rather than just disconnected so that the client cannot reconnect on its own. + client_node.stop().unwrap(); + + // Wait for the service to observe the disconnect, so the payment does not race against a + // channel that still looks live. + exponential_backoff_poll(|| { + let connected = + service_node.list_peers().iter().any(|p| p.node_id == client_node_id && p.is_connected); + (!connected).then_some(()) + }) + .await; + + // Not waiting on the payment to resolve: the forward failure we trigger on happens on the first + // attempt, while the payer keeps retrying in the background. + let _ = payer_node.bolt11_payment().send(&invoice, None).unwrap(); + + // Matching on the method keeps this distinct from the `lsps5.webhook_registered` notification + // the registration above already sent. + let notification_sent = service_logger.wait_for("LSPS5PaymentIncoming").await; + assert!( + notification_sent, + "Expected an HTLC held for an offline client to trigger a payment_incoming notification." + ); + + service_node.stop().unwrap(); + payer_node.stop().unwrap(); +}