From ebaa7edb839450ad117a01a14d0913e7c8e972f3 Mon Sep 17 00:00:00 2001 From: Fernando Ledesma Date: Tue, 31 Mar 2026 12:04:24 -0500 Subject: [PATCH 1/4] lsps1: Add `prune_orders` API to remove completed order state Add `prune_orders(counterparty_node_id, max_age: Duration)` to both `LSPS1ServiceHandler` and `LSPS1ServiceHandlerSync`. It removes all terminal orders (`CompletedAndChannelOpened` / `FailedAndRefunded`) for a given peer that are at least `max_age` old, persists the updated state, and returns the number of entries removed. Passing `Duration::ZERO` prunes all terminal orders immediately regardless of age, which is the recommended approach to unblock a client that has hit the per-peer request limit due to accumulated failed orders. On the `PeerState` layer, `prune_terminal_orders(now, max_age)` uses `retain` for a single-pass removal and sets `needs_persist` only when at least one entry is removed. --- lightning-liquidity/src/lsps1/peer_state.rs | 177 ++++++++++++++++++++ lightning-liquidity/src/lsps1/service.rs | 66 ++++++++ 2 files changed, 243 insertions(+) diff --git a/lightning-liquidity/src/lsps1/peer_state.rs b/lightning-liquidity/src/lsps1/peer_state.rs index 6e1889749ae..d814d6641ce 100644 --- a/lightning-liquidity/src/lsps1/peer_state.rs +++ b/lightning-liquidity/src/lsps1/peer_state.rs @@ -17,6 +17,8 @@ use super::msgs::{ use crate::lsps0::ser::{LSPSDateTime, LSPSRequestId}; use crate::prelude::HashMap; +use core::time::Duration; + use lightning::util::hash_tables::new_hash_map; use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum}; @@ -397,6 +399,36 @@ impl PeerState { }); } + /// Removes all terminal orders from state that are at least `max_age` old. + /// + /// Terminal orders are those in the [`ChannelOrderState::CompletedAndChannelOpened`] or + /// [`ChannelOrderState::FailedAndRefunded`] state. `max_age` is measured from the order's + /// `created_at` timestamp. Pass [`Duration::ZERO`] to prune all terminal orders regardless + /// of age, which is useful to immediately free per-peer quota when a client is blocked by + /// the request limit due to accumulated `FailedAndRefunded` entries. + /// + /// Returns the number of orders removed. + pub(super) fn prune_terminal_orders(&mut self, now: &LSPSDateTime, max_age: Duration) -> usize { + let mut pruned = 0usize; + self.outbound_channels_by_order_id.retain(|_order_id, order| { + let is_terminal = matches!( + order.state, + ChannelOrderState::CompletedAndChannelOpened { .. } + | ChannelOrderState::FailedAndRefunded { .. } + ); + if is_terminal && now.duration_since(&order.created_at) >= max_age { + pruned += 1; + false + } else { + true + } + }); + if pruned > 0 { + self.needs_persist |= true; + } + pruned + } + fn pending_requests_and_unpaid_orders(&self) -> usize { let pending_requests = self.pending_requests.len(); // We exclude paid and completed orders. @@ -778,4 +810,149 @@ mod tests { // Available in CompletedAndChannelOpened assert_eq!(state.channel_info(), Some(&channel_info)); } + + fn create_test_order_params() -> LSPS1OrderParams { + LSPS1OrderParams { + lsp_balance_sat: 100_000, + client_balance_sat: 0, + required_channel_confirmations: 0, + funding_confirms_within_blocks: 6, + channel_expiry_blocks: 144, + token: None, + announce_channel: false, + } + } + + #[test] + fn test_prune_terminal_orders_completed() { + let mut peer_state = PeerState::default(); + let order_id = LSPS1OrderId("order1".to_string()); + peer_state.new_order( + order_id.clone(), + create_test_order_params(), + LSPSDateTime::from_str("2024-01-01T00:00:00Z").unwrap(), + create_test_payment_info_bolt11_only(), + ); + peer_state.order_payment_received(&order_id, PaymentMethod::Bolt11).unwrap(); + peer_state.order_channel_opened(&order_id, create_test_channel_info()).unwrap(); + + // max_age=0 prunes all terminal orders regardless of age. + let now = LSPSDateTime::from_str("2024-01-01T01:00:00Z").unwrap(); + assert_eq!(peer_state.prune_terminal_orders(&now, Duration::ZERO), 1); + assert!(peer_state.get_order(&order_id).is_err()); + } + + #[test] + fn test_prune_terminal_orders_failed_and_refunded() { + let mut peer_state = PeerState::default(); + let order_id = LSPS1OrderId("order2".to_string()); + // Non-expired invoice: verify we do not require invoice expiry before pruning. + peer_state.new_order( + order_id.clone(), + create_test_order_params(), + LSPSDateTime::from_str("2024-01-01T00:00:00Z").unwrap(), + create_test_payment_info_bolt11_only(), + ); + peer_state.order_failed_and_refunded(&order_id).unwrap(); + + let now = LSPSDateTime::from_str("2024-01-01T01:00:00Z").unwrap(); + assert_eq!(peer_state.prune_terminal_orders(&now, Duration::ZERO), 1); + assert!(peer_state.get_order(&order_id).is_err()); + } + + #[test] + fn test_prune_terminal_orders_age_filter() { + let mut peer_state = PeerState::default(); + + // Old order (2 hours before now) — must be pruned when max_age = 1 hour. + let old_id = LSPS1OrderId("old".to_string()); + peer_state.new_order( + old_id.clone(), + create_test_order_params(), + LSPSDateTime::from_str("2024-01-01T00:00:00Z").unwrap(), + create_test_payment_info_bolt11_only(), + ); + peer_state.order_failed_and_refunded(&old_id).unwrap(); + + // Recent order (10 minutes before now) — must NOT be pruned when max_age = 1 hour. + let recent_id = LSPS1OrderId("recent".to_string()); + peer_state.new_order( + recent_id.clone(), + create_test_order_params(), + LSPSDateTime::from_str("2024-01-01T01:50:00Z").unwrap(), + create_test_payment_info_bolt11_only(), + ); + peer_state.order_failed_and_refunded(&recent_id).unwrap(); + + let now = LSPSDateTime::from_str("2024-01-01T02:00:00Z").unwrap(); + let pruned = peer_state.prune_terminal_orders(&now, Duration::from_secs(3600)); + assert_eq!(pruned, 1); + assert!(peer_state.get_order(&old_id).is_err()); + assert!(peer_state.get_order(&recent_id).is_ok()); + } + + #[test] + fn test_prune_terminal_orders_non_terminal_skipped() { + let mut peer_state = PeerState::default(); + + // ExpectingPayment is not a terminal state. + let expecting_id = LSPS1OrderId("expecting".to_string()); + peer_state.new_order( + expecting_id.clone(), + create_test_order_params(), + LSPSDateTime::from_str("2024-01-01T00:00:00Z").unwrap(), + create_test_payment_info_bolt11_only(), + ); + + // OrderPaid is not a terminal state. + let paid_id = LSPS1OrderId("paid".to_string()); + peer_state.new_order( + paid_id.clone(), + create_test_order_params(), + LSPSDateTime::from_str("2024-01-01T00:00:00Z").unwrap(), + create_test_payment_info_bolt11_only(), + ); + peer_state.order_payment_received(&paid_id, PaymentMethod::Bolt11).unwrap(); + + let now = LSPSDateTime::from_str("2024-01-01T02:00:00Z").unwrap(); + assert_eq!(peer_state.prune_terminal_orders(&now, Duration::ZERO), 0); + assert!(peer_state.get_order(&expecting_id).is_ok()); + assert!(peer_state.get_order(&paid_id).is_ok()); + } + + #[test] + fn test_prune_terminal_orders_frees_quota() { + let mut peer_state = PeerState::default(); + + // Fill up to the limit with FailedAndRefunded orders. + for i in 0..MAX_PENDING_REQUESTS_PER_PEER { + let order_id = LSPS1OrderId(format!("order{}", i)); + peer_state.new_order( + order_id.clone(), + create_test_order_params(), + LSPSDateTime::from_str("2024-01-01T00:00:00Z").unwrap(), + create_test_payment_info_bolt11_only(), + ); + peer_state.order_failed_and_refunded(&order_id).unwrap(); + } + + // Registering another request must fail: quota is exhausted. + let dummy_request = LSPS1Request::GetInfo(Default::default()); + assert!(matches!( + peer_state.register_request(LSPSRequestId("r0".to_string()), dummy_request.clone()), + Err(PeerStateError::TooManyPendingRequests) + )); + + // Prune all failed orders with max_age=0. + let now = LSPSDateTime::from_str("2024-01-01T01:00:00Z").unwrap(); + assert_eq!( + peer_state.prune_terminal_orders(&now, Duration::ZERO), + MAX_PENDING_REQUESTS_PER_PEER + ); + + // Now registering a new request must succeed. + assert!(peer_state + .register_request(LSPSRequestId("r1".to_string()), dummy_request) + .is_ok()); + } } diff --git a/lightning-liquidity/src/lsps1/service.rs b/lightning-liquidity/src/lsps1/service.rs index 0e139907589..c77329b508b 100644 --- a/lightning-liquidity/src/lsps1/service.rs +++ b/lightning-liquidity/src/lsps1/service.rs @@ -17,6 +17,7 @@ use core::ops::Deref; use core::pin::pin; use core::sync::atomic::{AtomicUsize, Ordering}; use core::task; +use core::time::Duration; use super::event::LSPS1ServiceEvent; use super::msgs::{ @@ -752,6 +753,52 @@ where Ok(()) } + /// Prunes terminal orders for a peer that are at least `max_age` old, freeing memory and + /// per-peer quota. + /// + /// Terminal orders are those in the [`LSPS1OrderState::Completed`] or + /// [`LSPS1OrderState::Failed`] state. `max_age` is measured from each order's `created_at` + /// timestamp. Pass [`Duration::ZERO`] to prune all terminal orders regardless of age, + /// which is useful to immediately free per-peer quota when a client is blocked by the + /// per-peer request limit due to accumulated failed orders. + /// + /// Returns the number of orders removed, or an [`APIError::APIMisuseError`] if no state + /// exists for the given counterparty. + pub async fn prune_orders( + &self, counterparty_node_id: PublicKey, max_age: Duration, + ) -> Result { + let now = + LSPSDateTime::new_from_duration_since_epoch(self.time_provider.duration_since_epoch()); + let pruned; + { + let outer_state_lock = self.per_peer_state.read().unwrap(); + let inner_state_lock = + outer_state_lock.get(&counterparty_node_id).ok_or_else(|| { + APIError::APIMisuseError { + err: format!( + "No existing state with counterparty {}", + counterparty_node_id + ), + } + })?; + let mut peer_state = inner_state_lock.lock().unwrap(); + pruned = peer_state.prune_terminal_orders(&now, max_age); + } + + if pruned > 0 { + self.persist_peer_state(counterparty_node_id).await.map_err(|e| { + APIError::APIMisuseError { + err: format!( + "Failed to persist peer state for {}: {}", + counterparty_node_id, e + ), + } + })?; + } + + Ok(pruned) + } + fn generate_order_id(&self) -> LSPS1OrderId { let bytes = self.entropy_source.get_secure_random_bytes(); LSPS1OrderId(utils::hex_str(&bytes[0..16])) @@ -930,6 +977,25 @@ where }, } } + + /// Prunes terminal orders for a peer that are at least `max_age` old. + /// + /// Wraps [`LSPS1ServiceHandler::prune_orders`]. + pub fn prune_orders( + &self, counterparty_node_id: PublicKey, max_age: Duration, + ) -> Result { + let mut fut = pin!(self.inner.prune_orders(counterparty_node_id, max_age)); + + let mut waker = dummy_waker(); + let mut ctx = task::Context::from_waker(&mut waker); + match fut.as_mut().poll(&mut ctx) { + task::Poll::Ready(result) => result, + task::Poll::Pending => { + // In a sync context, we can't wait for the future to complete. + unreachable!("Should not be pending in a sync context"); + }, + } + } } fn check_range(min: u64, max: u64, value: u64) -> bool { From 057fdc3d64b6b60754907bf48582641513b897f0 Mon Sep 17 00:00:00 2001 From: Fernando Ledesma Date: Mon, 13 Apr 2026 16:55:07 -0500 Subject: [PATCH 2/4] lsps2: Track channel creation time via `created_at` field Add a `created_at: LSPSDateTime` field to `OutboundJITChannel` to record when each JIT channel was created (i.e., when the buy request was accepted by the LSP). This timestamp is needed to implement time-based bulk pruning of completed channel state. The field is persisted as TLV type 10 with a `default_value` of Unix epoch, ensuring old serialized data (without TLV 10) is read back successfully with the epoch sentinel rather than failing deserialization. --- lightning-liquidity/src/lsps2/service.rs | 85 +++++++++++++++++++++--- lightning-liquidity/src/manager.rs | 9 +-- 2 files changed, 79 insertions(+), 15 deletions(-) diff --git a/lightning-liquidity/src/lsps2/service.rs b/lightning-liquidity/src/lsps2/service.rs index b7f6f2fc64d..df17534c752 100644 --- a/lightning-liquidity/src/lsps2/service.rs +++ b/lightning-liquidity/src/lsps2/service.rs @@ -19,13 +19,15 @@ use core::ops::Deref; use core::pin::pin; use core::sync::atomic::{AtomicUsize, Ordering}; use core::task; +use core::time::Duration; use crate::events::EventQueue; use crate::lsps0::ser::{ - LSPSMessage, LSPSProtocolMessageHandler, LSPSRequestId, LSPSResponseError, + LSPSDateTime, LSPSMessage, LSPSProtocolMessageHandler, LSPSRequestId, LSPSResponseError, JSONRPC_INTERNAL_ERROR_ERROR_CODE, JSONRPC_INTERNAL_ERROR_ERROR_MESSAGE, LSPS0_CLIENT_REJECTED_ERROR_CODE, }; +use crate::utils::time::TimeProvider; use crate::lsps2::event::LSPS2ServiceEvent; use crate::lsps2::payment_queue::{InterceptedHTLC, PaymentQueue}; use crate::lsps2::utils::{ @@ -497,6 +499,8 @@ struct OutboundJITChannel { opening_fee_params: LSPS2OpeningFeeParams, payment_size_msat: Option, trust_model: TrustModel, + /// The time at which the JIT channel was created (i.e., the buy request was accepted). + created_at: LSPSDateTime, } impl_writeable_tlv_based!(OutboundJITChannel, { @@ -505,12 +509,13 @@ impl_writeable_tlv_based!(OutboundJITChannel, { (4, opening_fee_params, required), (6, payment_size_msat, option), (8, trust_model, required), + (10, created_at, (default_value, LSPSDateTime::new_from_duration_since_epoch(Duration::ZERO))), }); impl OutboundJITChannel { fn new( payment_size_msat: Option, opening_fee_params: LSPS2OpeningFeeParams, - user_channel_id: u128, client_trusts_lsp: bool, + user_channel_id: u128, client_trusts_lsp: bool, created_at: LSPSDateTime, ) -> Self { Self { user_channel_id, @@ -518,6 +523,7 @@ impl OutboundJITChannel { opening_fee_params, payment_size_msat, trust_model: TrustModel::new(client_trusts_lsp), + created_at, } } @@ -702,9 +708,10 @@ macro_rules! get_or_insert_peer_state_entry { } /// The main object allowing to send and receive bLIP-52 / LSPS2 messages. -pub struct LSPS2ServiceHandler +pub struct LSPS2ServiceHandler where CM::Target: AChannelManager, + TP::Target: TimeProvider, { channel_manager: CM, kv_store: K, @@ -717,17 +724,20 @@ where total_pending_requests: AtomicUsize, config: LSPS2ServiceConfig, persistence_in_flight: AtomicUsize, + time_provider: TP, } -impl LSPS2ServiceHandler +impl + LSPS2ServiceHandler where CM::Target: AChannelManager, + TP::Target: TimeProvider, { /// Constructs a `LSPS2ServiceHandler`. pub(crate) fn new( per_peer_state: HashMap>, pending_messages: Arc, pending_events: Arc>, channel_manager: CM, kv_store: K, tx_broadcaster: T, - config: LSPS2ServiceConfig, + config: LSPS2ServiceConfig, time_provider: TP, ) -> Result { let mut peer_by_intercept_scid = new_hash_map(); let mut peer_by_channel_id = new_hash_map(); @@ -768,6 +778,7 @@ where kv_store, tx_broadcaster, config, + time_provider, }) } @@ -921,11 +932,15 @@ where peer_by_intercept_scid.insert(intercept_scid, *counterparty_node_id); } + let created_at = LSPSDateTime::new_from_duration_since_epoch( + self.time_provider.duration_since_epoch(), + ); let outbound_jit_channel = OutboundJITChannel::new( buy_request.payment_size_msat, buy_request.opening_fee_params, user_channel_id, client_trusts_lsp, + created_at, ); peer_state_lock @@ -2050,10 +2065,11 @@ where } } -impl LSPSProtocolMessageHandler - for LSPS2ServiceHandler +impl + LSPSProtocolMessageHandler for LSPS2ServiceHandler where CM::Target: AChannelManager, + TP::Target: TimeProvider, { type ProtocolMessage = LSPS2Message; const PROTOCOL_NUMBER: Option = Some(2); @@ -2128,18 +2144,21 @@ pub struct LSPS2ServiceHandlerSync< CM: Deref, K: KVStore + Clone, T: BroadcasterInterface + Clone, + TP: Deref + Clone, > where CM::Target: AChannelManager, + TP::Target: TimeProvider, { - inner: &'a LSPS2ServiceHandler, + inner: &'a LSPS2ServiceHandler, } -impl<'a, CM: Deref, K: KVStore + Clone, T: BroadcasterInterface + Clone> - LSPS2ServiceHandlerSync<'a, CM, K, T> +impl<'a, CM: Deref, K: KVStore + Clone, T: BroadcasterInterface + Clone, TP: Deref + Clone> + LSPS2ServiceHandlerSync<'a, CM, K, T, TP> where CM::Target: AChannelManager, + TP::Target: TimeProvider, { - pub(crate) fn from_inner(inner: &'a LSPS2ServiceHandler) -> Self { + pub(crate) fn from_inner(inner: &'a LSPS2ServiceHandler) -> Self { Self { inner } } @@ -2785,6 +2804,7 @@ mod tests { opening_fee_params.clone(), user_channel_id, true, + LSPSDateTime::from_str("2024-01-01T00:00:00Z").unwrap(), ); let opening_payment_hash = PaymentHash([42; 32]); @@ -2864,4 +2884,47 @@ mod tests { "Broadcast was not allowed even though all the skimmed fees were collected" ); } + + #[test] + fn test_outbound_jit_channel_created_at_stored() { + let opening_fee_params = LSPS2OpeningFeeParams { + min_fee_msat: 1_000, + proportional: 0, + valid_until: LSPSDateTime::from_str("2035-05-20T08:30:45Z").unwrap(), + min_lifetime: 144, + max_client_to_self_delay: 128, + min_payment_size_msat: 1, + max_payment_size_msat: 10_000_000_000, + promise: "ignore".to_string(), + }; + let created_at = LSPSDateTime::from_str("2024-06-15T12:00:00Z").unwrap(); + let channel = + OutboundJITChannel::new(Some(1_000_000), opening_fee_params, 1u128, true, created_at); + assert_eq!(channel.created_at, created_at); + } + + #[test] + fn test_outbound_jit_channel_created_at_round_trips() { + use lightning::util::ser::{Readable, Writeable}; + + let opening_fee_params = LSPS2OpeningFeeParams { + min_fee_msat: 1_000, + proportional: 0, + valid_until: LSPSDateTime::from_str("2035-05-20T08:30:45Z").unwrap(), + min_lifetime: 144, + max_client_to_self_delay: 128, + min_payment_size_msat: 1, + max_payment_size_msat: 10_000_000_000, + promise: "ignore".to_string(), + }; + let created_at = LSPSDateTime::from_str("2024-06-15T12:00:00Z").unwrap(); + let channel = + OutboundJITChannel::new(Some(1_000_000), opening_fee_params, 1u128, true, created_at); + + let mut buf = Vec::new(); + channel.write(&mut buf).unwrap(); + + let decoded = ::read(&mut &buf[..]).unwrap(); + assert_eq!(decoded.created_at, created_at); + } } diff --git a/lightning-liquidity/src/manager.rs b/lightning-liquidity/src/manager.rs index f1b098dbfaa..b34dc8d98b9 100644 --- a/lightning-liquidity/src/manager.rs +++ b/lightning-liquidity/src/manager.rs @@ -283,7 +283,7 @@ pub struct LiquidityManager< lsps0_service_handler: Option, lsps1_service_handler: Option>, lsps1_client_handler: Option>, - lsps2_service_handler: Option>, + lsps2_service_handler: Option>, lsps2_client_handler: Option>, lsps5_service_handler: Option>, lsps5_client_handler: Option>, @@ -377,7 +377,7 @@ where let lsps2_service_handler = if let Some(service_config) = service_config.as_ref() { if let Some(lsps2_service_config) = service_config.lsps2_service_config.as_ref() { if let Some(number) = - as LSPSProtocolMessageHandler>::PROTOCOL_NUMBER + as LSPSProtocolMessageHandler>::PROTOCOL_NUMBER { supported_protocols.push(number); } @@ -391,6 +391,7 @@ where kv_store.clone(), transaction_broadcaster.clone(), lsps2_service_config.clone(), + time_provider.clone(), )?) } else { None @@ -540,7 +541,7 @@ where /// Returns a reference to the LSPS2 server-side handler. /// /// The returned hendler allows to initiate the LSPS2 service-side flow. - pub fn lsps2_service_handler(&self) -> Option<&LSPS2ServiceHandler> { + pub fn lsps2_service_handler(&self) -> Option<&LSPS2ServiceHandler> { self.lsps2_service_handler.as_ref() } @@ -1053,7 +1054,7 @@ where /// Wraps [`LiquidityManager::lsps2_service_handler`]. pub fn lsps2_service_handler<'a>( &'a self, - ) -> Option, T>> { + ) -> Option, T, TP>> { self.inner.lsps2_service_handler.as_ref().map(|r| LSPS2ServiceHandlerSync::from_inner(r)) } From cc4cc39b7b4da380bd81f248f2a2534c504d568d Mon Sep 17 00:00:00 2001 From: Fernando Ledesma Date: Thu, 16 Apr 2026 10:56:32 -0500 Subject: [PATCH 3/4] lsps2: Add `prune_channels` API to remove completed JIT channel state Add `LSPS2ServiceHandler::prune_channels` that lets the LSP operator remove all channels in the `PaymentForwarded` terminal state whose `created_at` timestamp is at least `max_age` old. Passing `Duration::ZERO` prunes all terminal channels regardless of age. All associated state is cleaned up atomically: - per-peer `intercept_scid_by_channel_id` and `intercept_scid_by_user_channel_id` - handler-level `peer_by_intercept_scid` and `peer_by_channel_id` A new `PeerState::prune_terminal_channels` helper handles the intra-peer map cleanup and returns the removed `(scid, channel_id)` pairs for the handler to clean up the outer maps. Integration tests cover: non-terminal channels not pruned, unknown counterparty errors, age filtering, and successful bulk prune. --- lightning-liquidity/src/lsps2/service.rs | 233 ++++++++++++-- .../tests/lsps2_integration_tests.rs | 301 ++++++++++++++++++ 2 files changed, 513 insertions(+), 21 deletions(-) diff --git a/lightning-liquidity/src/lsps2/service.rs b/lightning-liquidity/src/lsps2/service.rs index df17534c752..07d7101a27f 100644 --- a/lightning-liquidity/src/lsps2/service.rs +++ b/lightning-liquidity/src/lsps2/service.rs @@ -664,6 +664,37 @@ impl PeerState { // Return whether the entire state is empty. self.pending_requests.is_empty() && self.outbound_channels_by_intercept_scid.is_empty() } + + /// Removes all channels in the [`PaymentForwarded`] terminal state whose `created_at` + /// timestamp is at least `max_age` old. Passing [`Duration::ZERO`] removes all terminal + /// channels regardless of age. + /// + /// Cleans up the intra-peer auxiliary maps for each removed channel and returns the + /// `(intercept_scid, channel_id)` pairs so the caller can remove them from the + /// handler-level peer lookup maps. + /// + /// [`PaymentForwarded`]: OutboundJITChannelState::PaymentForwarded + /// [`Duration::ZERO`]: core::time::Duration::ZERO + fn prune_terminal_channels( + &mut self, now: &LSPSDateTime, max_age: Duration, + ) -> Vec<(u64, ChannelId)> { + let mut removed = Vec::new(); + self.outbound_channels_by_intercept_scid.retain(|scid, channel| { + if let OutboundJITChannelState::PaymentForwarded { channel_id } = &channel.state { + let should_prune = max_age == Duration::ZERO + || now.duration_since(&channel.created_at) >= max_age; + if should_prune { + removed.push((*scid, *channel_id)); + self.intercept_scid_by_channel_id.retain(|_, iscid| iscid != scid); + self.intercept_scid_by_user_channel_id.retain(|_, iscid| iscid != scid); + self.needs_persist = true; + return false; + } + } + true + }); + removed + } } impl_writeable_tlv_based!(PeerState, { @@ -1267,6 +1298,66 @@ where Ok(()) } + /// Prunes completed JIT channels from state for a given peer, freeing memory. + /// + /// Removes all channels in the [`OutboundJITChannelState::PaymentForwarded`] terminal state + /// whose `created_at` timestamp is at least `max_age` old. Pass [`Duration::ZERO`] to prune + /// all terminal channels regardless of age. + /// + /// All associated state is cleaned up for each removed channel, including the per-peer + /// `intercept_scid_by_channel_id` and `intercept_scid_by_user_channel_id` maps as well as + /// the handler-level `peer_by_intercept_scid` and `peer_by_channel_id` lookups. + /// + /// Returns the number of channels pruned, or an [`APIError::APIMisuseError`] if the + /// counterparty has no state. + /// + /// [`Duration::ZERO`]: core::time::Duration::ZERO + pub async fn prune_channels( + &self, counterparty_node_id: PublicKey, max_age: Duration, + ) -> Result { + let now = LSPSDateTime::new_from_duration_since_epoch( + self.time_provider.duration_since_epoch(), + ); + + let removed = { + let outer_state_lock = self.per_peer_state.read().unwrap(); + let inner_state_lock = + outer_state_lock.get(&counterparty_node_id).ok_or_else(|| { + APIError::APIMisuseError { + err: format!( + "No existing state with counterparty {}", + counterparty_node_id + ), + } + })?; + let mut peer_state = inner_state_lock.lock().unwrap(); + peer_state.prune_terminal_channels(&now, max_age) + }; + + let pruned = removed.len(); + if pruned > 0 { + let mut peer_by_intercept_scid = self.peer_by_intercept_scid.write().unwrap(); + let mut peer_by_channel_id = self.peer_by_channel_id.write().unwrap(); + for (scid, channel_id) in &removed { + peer_by_intercept_scid.remove(scid); + peer_by_channel_id.remove(channel_id); + } + drop(peer_by_intercept_scid); + drop(peer_by_channel_id); + + self.persist_peer_state(counterparty_node_id).await.map_err(|e| { + APIError::APIMisuseError { + err: format!( + "Failed to persist peer state for {}: {}", + counterparty_node_id, e + ), + } + })?; + } + + Ok(pruned) + } + /// Abandons a pending JIT‐open flow for `user_channel_id`, removing all local state. /// /// This removes the intercept SCID, any outbound channel state, and associated @@ -2349,6 +2440,24 @@ where } } + /// Prunes completed JIT channels from state for a given peer, freeing memory. + /// + /// Wraps [`LSPS2ServiceHandler::prune_channels`]. + pub fn prune_channels( + &self, counterparty_node_id: PublicKey, max_age: Duration, + ) -> Result { + let mut fut = pin!(self.inner.prune_channels(counterparty_node_id, max_age)); + + let mut waker = dummy_waker(); + let mut ctx = task::Context::from_waker(&mut waker); + match fut.as_mut().poll(&mut ctx) { + task::Poll::Ready(result) => result, + task::Poll::Pending => { + unreachable!("Should not be pending in a sync context"); + }, + } + } + /// Forward [`Event::ChannelReady`] event parameters into this function. /// /// Wraps [`LSPS2ServiceHandler::channel_ready`]. @@ -2885,21 +2994,29 @@ mod tests { ); } - #[test] - fn test_outbound_jit_channel_created_at_stored() { - let opening_fee_params = LSPS2OpeningFeeParams { - min_fee_msat: 1_000, - proportional: 0, - valid_until: LSPSDateTime::from_str("2035-05-20T08:30:45Z").unwrap(), + fn make_test_opening_fee_params() -> LSPS2OpeningFeeParams { + LSPS2OpeningFeeParams { + min_fee_msat: 1000, + proportional: 100, + valid_until: LSPSDateTime::from_str("2035-01-01T00:00:00Z").unwrap(), min_lifetime: 144, max_client_to_self_delay: 128, min_payment_size_msat: 1, max_payment_size_msat: 10_000_000_000, promise: "ignore".to_string(), - }; + } + } + + #[test] + fn test_outbound_jit_channel_created_at_stored() { let created_at = LSPSDateTime::from_str("2024-06-15T12:00:00Z").unwrap(); - let channel = - OutboundJITChannel::new(Some(1_000_000), opening_fee_params, 1u128, true, created_at); + let channel = OutboundJITChannel::new( + Some(1_000_000), + make_test_opening_fee_params(), + 1u128, + true, + created_at, + ); assert_eq!(channel.created_at, created_at); } @@ -2907,19 +3024,14 @@ mod tests { fn test_outbound_jit_channel_created_at_round_trips() { use lightning::util::ser::{Readable, Writeable}; - let opening_fee_params = LSPS2OpeningFeeParams { - min_fee_msat: 1_000, - proportional: 0, - valid_until: LSPSDateTime::from_str("2035-05-20T08:30:45Z").unwrap(), - min_lifetime: 144, - max_client_to_self_delay: 128, - min_payment_size_msat: 1, - max_payment_size_msat: 10_000_000_000, - promise: "ignore".to_string(), - }; let created_at = LSPSDateTime::from_str("2024-06-15T12:00:00Z").unwrap(); - let channel = - OutboundJITChannel::new(Some(1_000_000), opening_fee_params, 1u128, true, created_at); + let channel = OutboundJITChannel::new( + Some(1_000_000), + make_test_opening_fee_params(), + 1u128, + true, + created_at, + ); let mut buf = Vec::new(); channel.write(&mut buf).unwrap(); @@ -2927,4 +3039,83 @@ mod tests { let decoded = ::read(&mut &buf[..]).unwrap(); assert_eq!(decoded.created_at, created_at); } + + // Verify that a PeerState entry in PaymentForwarded state is correctly removed along with + // all auxiliary lookup maps when prune_channel logic is exercised manually. + #[test] + fn test_peer_state_prune_payment_forwarded_channel() { + let intercept_scid = 42u64; + let user_channel_id = 1u128; + let channel_id = ChannelId([1; 32]); + let created_at = LSPSDateTime::from_str("2024-01-01T00:00:00Z").unwrap(); + + let mut jit_channel = OutboundJITChannel::new( + Some(1_000_000), + make_test_opening_fee_params(), + user_channel_id, + false, + created_at, + ); + + // Drive the channel through to PaymentForwarded state. + let htlc = InterceptedHTLC { + intercept_id: InterceptId([0; 32]), + expected_outbound_amount_msat: 1_000_000, + payment_hash: PaymentHash([1; 32]), + }; + let action = jit_channel.htlc_intercepted(htlc).unwrap(); + assert!(matches!(action, Some(HTLCInterceptedAction::OpenChannel(_)))); + jit_channel.channel_ready(channel_id).unwrap(); + // Provide enough fee to transition to PaymentForwarded. + jit_channel.payment_forwarded(1_000).unwrap(); + + // Build a minimal PeerState with the channel and auxiliary maps. + let mut peer_state = PeerState::new(); + peer_state.outbound_channels_by_intercept_scid.insert(intercept_scid, jit_channel); + peer_state.intercept_scid_by_user_channel_id.insert(user_channel_id, intercept_scid); + peer_state.intercept_scid_by_channel_id.insert(channel_id, intercept_scid); + + // Confirm the channel is in PaymentForwarded state. + assert!(matches!( + peer_state.outbound_channels_by_intercept_scid.get(&intercept_scid).unwrap().state, + OutboundJITChannelState::PaymentForwarded { .. } + )); + + // Simulate what prune_channel does internally. + peer_state.outbound_channels_by_intercept_scid.remove(&intercept_scid); + peer_state.intercept_scid_by_channel_id.retain(|_, iscid| *iscid != intercept_scid); + peer_state.intercept_scid_by_user_channel_id.retain(|_, iscid| *iscid != intercept_scid); + peer_state.needs_persist = true; + + // All maps must be empty after pruning. + assert!(peer_state.outbound_channels_by_intercept_scid.is_empty()); + assert!(peer_state.intercept_scid_by_channel_id.is_empty()); + assert!(peer_state.intercept_scid_by_user_channel_id.is_empty()); + assert!(peer_state.needs_persist); + } + + // Verify that prune_channel rejects non-terminal states. + #[test] + fn test_peer_state_prune_channel_non_terminal_rejected() { + let intercept_scid = 99u64; + let user_channel_id = 2u128; + let created_at = LSPSDateTime::from_str("2024-01-01T00:00:00Z").unwrap(); + let jit_channel = OutboundJITChannel::new( + Some(500_000), + make_test_opening_fee_params(), + user_channel_id, + false, + created_at, + ); + + // Channel is in PendingInitialPayment — a non-terminal state. + assert!(matches!(jit_channel.state, OutboundJITChannelState::PendingInitialPayment { .. })); + + // Verify the guard logic that prune_channel uses. + let is_prunable = + matches!(jit_channel.state, OutboundJITChannelState::PaymentForwarded { .. }); + assert!(!is_prunable, "PendingInitialPayment must not be considered prunable"); + + let _ = intercept_scid; // silence unused warning + } } diff --git a/lightning-liquidity/tests/lsps2_integration_tests.rs b/lightning-liquidity/tests/lsps2_integration_tests.rs index fbff2eae4cd..41e9d1094be 100644 --- a/lightning-liquidity/tests/lsps2_integration_tests.rs +++ b/lightning-liquidity/tests/lsps2_integration_tests.rs @@ -2352,3 +2352,304 @@ fn client_trusts_lsp_partial_fee_does_not_trigger_broadcast() { client_node.inner.chain_monitor.added_monitors.lock().unwrap().clear(); payer_node.chain_monitor.added_monitors.lock().unwrap().clear(); } + +/// Drive the buy-request protocol on two nodes (no payer, no payment), leaving the channel +/// in `PendingInitialPayment` state on the service side. +fn run_buy_request_protocol<'a, 'b, 'c>( + lsps_nodes: &LSPSNodes<'a, 'b, 'c>, intercept_scid: u64, user_channel_id: u128, + cltv_expiry_delta: u32, +) { + let service_node = &lsps_nodes.service_node; + let client_node = &lsps_nodes.client_node; + + let service_node_id = service_node.inner.node.get_our_node_id(); + let client_node_id = client_node.inner.node.get_our_node_id(); + + let client_handler = client_node.liquidity_manager.lsps2_client_handler().unwrap(); + let service_handler = service_node.liquidity_manager.lsps2_service_handler().unwrap(); + + let _get_info_request_id = client_handler.request_opening_params(service_node_id, None); + let get_info_request = get_lsps_message!(client_node, service_node_id); + service_node.liquidity_manager.handle_custom_message(get_info_request, client_node_id).unwrap(); + + let get_info_event = service_node.liquidity_manager.next_event().unwrap(); + let get_info_request_id = match get_info_event { + LiquidityEvent::LSPS2Service(LSPS2ServiceEvent::GetInfo { request_id, .. }) => request_id, + _ => panic!("Unexpected event"), + }; + + let raw_opening_params = LSPS2RawOpeningFeeParams { + min_fee_msat: 100, + proportional: 0, + valid_until: LSPSDateTime::from_str("2035-05-20T08:30:45Z").unwrap(), + min_lifetime: 144, + max_client_to_self_delay: 128, + min_payment_size_msat: 1, + max_payment_size_msat: 100_000_000, + }; + service_handler + .opening_fee_params_generated( + &client_node_id, + get_info_request_id.clone(), + vec![raw_opening_params], + ) + .unwrap(); + + let get_info_response = get_lsps_message!(service_node, client_node_id); + client_node.liquidity_manager.handle_custom_message(get_info_response, service_node_id).unwrap(); + + let opening_fee_params = match client_node.liquidity_manager.next_event().unwrap() { + LiquidityEvent::LSPS2Client(LSPS2ClientEvent::OpeningParametersReady { + opening_fee_params_menu, + .. + }) => opening_fee_params_menu.first().unwrap().clone(), + _ => panic!("Unexpected event"), + }; + + let buy_request_id = client_handler + .select_opening_params(service_node_id, Some(1_000_000), opening_fee_params) + .unwrap(); + + let buy_request = get_lsps_message!(client_node, service_node_id); + service_node.liquidity_manager.handle_custom_message(buy_request, client_node_id).unwrap(); + + let buy_event = service_node.liquidity_manager.next_event().unwrap(); + if let LiquidityEvent::LSPS2Service(LSPS2ServiceEvent::BuyRequest { request_id, .. }) = buy_event + { + assert_eq!(request_id, buy_request_id); + } else { + panic!("Unexpected event"); + } + + service_handler + .invoice_parameters_generated( + &client_node_id, + buy_request_id.clone(), + intercept_scid, + cltv_expiry_delta, + true, + user_channel_id, + ) + .unwrap(); + + let buy_response = get_lsps_message!(service_node, client_node_id); + client_node.liquidity_manager.handle_custom_message(buy_response, service_node_id).unwrap(); + let _invoice_params_event = client_node.liquidity_manager.next_event().unwrap(); +} + +#[test] +fn prune_channels_non_terminal_not_pruned() { + let chanmon_cfgs = create_chanmon_cfgs(2); + let node_cfgs = create_node_cfgs(2, &chanmon_cfgs); + let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]); + let nodes = create_network(2, &node_cfgs, &node_chanmgrs); + let (lsps_nodes, _promise_secret) = setup_test_lsps2_nodes(nodes); + let LSPSNodes { ref service_node, ref client_node } = lsps_nodes; + + let client_node_id = client_node.inner.node.get_our_node_id(); + let service_handler = service_node.liquidity_manager.lsps2_service_handler().unwrap(); + + let intercept_scid = service_node.node.get_intercept_scid(); + run_buy_request_protocol(&lsps_nodes, intercept_scid, 42u128, 144); + + // Channel is in PendingInitialPayment — prune_channels must not remove it. + let pruned = service_handler.prune_channels(client_node_id, Duration::ZERO).unwrap(); + assert_eq!(pruned, 0, "PendingInitialPayment channel must not be pruned"); +} + +#[test] +fn prune_channels_unknown_counterparty_fails() { + let chanmon_cfgs = create_chanmon_cfgs(2); + let node_cfgs = create_node_cfgs(2, &chanmon_cfgs); + let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]); + let nodes = create_network(2, &node_cfgs, &node_chanmgrs); + let (lsps_nodes, _) = setup_test_lsps2_nodes(nodes); + let LSPSNodes { ref service_node, .. } = lsps_nodes; + + let service_handler = service_node.liquidity_manager.lsps2_service_handler().unwrap(); + + let unknown_pubkey = PublicKey::from_slice(&[2u8; 33]).unwrap(); + assert!( + service_handler.prune_channels(unknown_pubkey, Duration::ZERO).is_err(), + "Expected error for unknown counterparty" + ); +} + +#[test] +fn prune_channels_payment_forwarded_and_age_filter() { + let chanmon_cfgs = create_chanmon_cfgs(3); + let node_cfgs = create_node_cfgs(3, &chanmon_cfgs); + let mut service_node_config = test_default_channel_config(); + service_node_config.htlc_interception_flags = HTLCInterceptionFlags::ToInterceptSCIDs as u8; + + let mut client_node_config = test_default_channel_config(); + client_node_config.channel_config.accept_underpaying_htlcs = true; + let node_chanmgrs = create_node_chanmgrs( + 3, + &node_cfgs, + &[Some(service_node_config), Some(client_node_config), None], + ); + let nodes = create_network(3, &node_cfgs, &node_chanmgrs); + let (lsps_nodes, promise_secret) = setup_test_lsps2_nodes_with_payer(nodes); + let LSPSNodesWithPayer { ref service_node, ref client_node, ref payer_node } = lsps_nodes; + + let payer_node_id = payer_node.node.get_our_node_id(); + let service_node_id = service_node.inner.node.get_our_node_id(); + let client_node_id = client_node.inner.node.get_our_node_id(); + + let service_handler = service_node.liquidity_manager.lsps2_service_handler().unwrap(); + + create_chan_between_nodes_with_value(&payer_node, &service_node.inner, 2_000_000, 100_000); + + let intercept_scid = service_node.node.get_intercept_scid(); + let user_channel_id = 42u128; + let cltv_expiry_delta: u32 = 144; + let payment_size_msat = Some(1_000_000); + let fee_base_msat: u64 = 1_000; + + execute_lsps2_dance( + &lsps_nodes, + intercept_scid, + user_channel_id, + cltv_expiry_delta, + promise_secret, + payment_size_msat, + fee_base_msat, + ); + + let invoice = create_jit_invoice( + &client_node, + service_node_id, + intercept_scid, + cltv_expiry_delta, + payment_size_msat, + "prune-test", + 3600, + ) + .unwrap(); + + payer_node + .node + .pay_for_bolt11_invoice( + &invoice, + PaymentId(invoice.payment_hash().0), + None, + OptionalBolt11PaymentParams::default(), + ) + .unwrap(); + + check_added_monitors(&payer_node, 1); + let events = payer_node.node.get_and_clear_pending_msg_events(); + let ev = SendEvent::from_event(events[0].clone()); + service_node.inner.node.handle_update_add_htlc(payer_node_id, &ev.msgs[0]); + do_commitment_signed_dance(&service_node.inner, &payer_node, &ev.commitment_msg, false, true); + service_node.inner.node.process_pending_htlc_forwards(); + + let events = service_node.inner.node.get_and_clear_pending_events(); + assert_eq!(events.len(), 1); + let expected_outbound_amount_msat = match &events[0] { + Event::HTLCIntercepted { + intercept_id, + requested_next_hop_scid, + payment_hash, + expected_outbound_amount_msat, + .. + } => { + assert_eq!(*requested_next_hop_scid, intercept_scid); + service_handler + .htlc_intercepted( + *requested_next_hop_scid, + *intercept_id, + *expected_outbound_amount_msat, + *payment_hash, + ) + .unwrap(); + *expected_outbound_amount_msat + }, + other => panic!("Expected HTLCIntercepted, got {:?}", other), + }; + + let open_channel_event = service_node.liquidity_manager.next_event().unwrap(); + match open_channel_event { + LiquidityEvent::LSPS2Service(LSPS2ServiceEvent::OpenChannel { .. }) => {}, + other => panic!("Expected OpenChannel, got: {:?}", other), + }; + + let (channel_id, funding_tx) = create_channel_with_manual_broadcast( + &service_node_id, + &client_node_id, + &service_node, + &client_node, + user_channel_id, + &expected_outbound_amount_msat, + true, + ); + + service_handler.channel_ready(user_channel_id, &channel_id, &client_node_id).unwrap(); + service_node.inner.node.process_pending_htlc_forwards(); + + { + let mut added_monitors = + service_node.inner.chain_monitor.added_monitors.lock().unwrap(); + assert_eq!(added_monitors.len(), 1); + added_monitors.clear(); + } + let mut events = service_node.inner.node.get_and_clear_pending_msg_events(); + assert_eq!(events.len(), 1); + let pay_event = SendEvent::from_event(events.remove(0)); + + client_node.inner.node.handle_update_add_htlc(service_node_id, &pay_event.msgs[0]); + do_commitment_signed_dance( + &client_node.inner, + &service_node.inner, + &pay_event.commitment_msg, + false, + true, + ); + client_node.inner.node.process_pending_htlc_forwards(); + + let client_events = client_node.inner.node.get_and_clear_pending_events(); + assert_eq!(client_events.len(), 1); + let preimage = match &client_events[0] { + Event::PaymentClaimable { purpose, .. } => purpose.preimage(), + other => panic!("Expected PaymentClaimable, got {:?}", other), + }; + + client_node.inner.node.claim_funds(preimage.unwrap()); + + claim_and_assert_forwarded_only( + &payer_node, + &service_node.inner, + &client_node.inner, + preimage.unwrap(), + ); + + let service_events = service_node.node.get_and_clear_pending_events(); + assert_eq!(service_events.len(), 1); + let total_fee_msat = match service_events[0].clone() { + Event::PaymentForwarded { skimmed_fee_msat, total_fee_earned_msat, .. } => { + service_handler.payment_forwarded(channel_id, skimmed_fee_msat.unwrap_or(0)).unwrap(); + total_fee_earned_msat.map(|total| total - skimmed_fee_msat.unwrap_or(0)) + }, + _ => panic!("Expected PaymentForwarded, got: {:?}", service_events[0]), + }; + + // Channel is now in PaymentForwarded. Verify age filtering: a max_age of 1 year should + // not prune a channel created moments ago. + let pruned = + service_handler.prune_channels(client_node_id, Duration::from_secs(365 * 24 * 3600)).unwrap(); + assert_eq!(pruned, 0, "Channel created moments ago must not exceed 1-year max_age"); + + // Duration::ZERO prunes all terminal channels regardless of age. + let pruned = service_handler.prune_channels(client_node_id, Duration::ZERO).unwrap(); + assert_eq!(pruned, 1, "Expected one PaymentForwarded channel to be pruned"); + + // A second call must find nothing to prune. + let pruned = service_handler.prune_channels(client_node_id, Duration::ZERO).unwrap(); + assert_eq!(pruned, 0, "Second call must find nothing to prune"); + + let broadcasted = service_node.inner.tx_broadcaster.txn_broadcasted.lock().unwrap(); + assert!(broadcasted.iter().any(|b| b.compute_txid() == funding_tx.compute_txid())); + + expect_payment_sent(&payer_node, preimage.unwrap(), Some(total_fee_msat), true, true); +} From e56052ddc0f01a0a5bcdba447383d9c00d0da1d1 Mon Sep 17 00:00:00 2001 From: Fernando Ledesma Date: Thu, 16 Apr 2026 10:59:14 -0500 Subject: [PATCH 4/4] lsps5: Add `prune_webhook` API to remove a client's webhook entry Add `LSPS5ServiceHandler::prune_webhook` that removes a single webhook entry identified by `counterparty_node_id` and `LSPS5AppName`. The method is synchronous, consistent with all other public `notify_*` methods on this handler: it marks the peer state as dirty and relies on the normal `LiquidityManager::persist` loop to flush the change to the KVStore. The method reuses the existing private `PeerState::remove_webhook` helper and returns an `APIError::APIMisuseError` if the counterparty has no registered state or the given `app_name` is not found. --- lightning-liquidity/src/lsps5/service.rs | 33 +++++++ .../tests/lsps5_integration_tests.rs | 86 +++++++++++++++++++ 2 files changed, 119 insertions(+) diff --git a/lightning-liquidity/src/lsps5/service.rs b/lightning-liquidity/src/lsps5/service.rs index 4678d38dc9a..e4170b1d5f0 100644 --- a/lightning-liquidity/src/lsps5/service.rs +++ b/lightning-liquidity/src/lsps5/service.rs @@ -31,6 +31,7 @@ use lightning::impl_writeable_tlv_based; use lightning::ln::channelmanager::AChannelManager; use lightning::ln::msgs::{ErrorAction, LightningError}; use lightning::sign::NodeSigner; +use lightning::util::errors::APIError; use lightning::util::logger::Level; use lightning::util::persist::KVStore; use lightning::util::ser::Writeable; @@ -580,6 +581,38 @@ where self.send_notifications_to_client_webhooks(client_id, notification) } + /// Removes a specific webhook registration for a client. + /// + /// This can be used to prune webhook state for a client that is no longer active or whose + /// webhooks are no longer relevant, complementing the automatic 30-day stale webhook pruning. + /// The state change will be persisted on the next call to [`LiquidityManager::persist`]. + /// + /// Returns an [`APIError::APIMisuseError`] if the client has no registered state or no + /// webhook with the given `app_name` is registered for that client. + /// + /// [`LiquidityManager::persist`]: crate::LiquidityManager::persist + pub fn prune_webhook( + &self, counterparty_node_id: PublicKey, app_name: &LSPS5AppName, + ) -> Result<(), APIError> { + let mut outer_state_lock = self.per_peer_state.write().unwrap(); + match outer_state_lock.get_mut(&counterparty_node_id) { + Some(peer_state) => { + if !peer_state.remove_webhook(app_name) { + return Err(APIError::APIMisuseError { + err: format!( + "No webhook with app_name '{}' registered for counterparty {}", + app_name, counterparty_node_id + ), + }); + } + Ok(()) + }, + None => Err(APIError::APIMisuseError { + err: format!("No existing state with counterparty {}", counterparty_node_id), + }), + } + } + fn send_notifications_to_client_webhooks( &self, client_id: PublicKey, notification: WebhookNotification, ) -> Result<(), LSPS5ProtocolError> { diff --git a/lightning-liquidity/tests/lsps5_integration_tests.rs b/lightning-liquidity/tests/lsps5_integration_tests.rs index 2b32b4dcbc6..8115b4db7b0 100644 --- a/lightning-liquidity/tests/lsps5_integration_tests.rs +++ b/lightning-liquidity/tests/lsps5_integration_tests.rs @@ -1633,3 +1633,89 @@ fn lsps5_service_handler_persistence_across_restarts() { } } } + +#[test] +fn prune_webhook_removes_registered_webhook() { + let chanmon_cfgs = create_chanmon_cfgs(2); + let node_cfgs = create_node_cfgs(2, &chanmon_cfgs); + let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]); + let nodes = create_network(2, &node_cfgs, &node_chanmgrs); + let (lsps_nodes, _) = lsps5_test_setup(nodes, Arc::new(DefaultTimeProvider)); + let LSPSNodes { service_node, client_node, .. } = lsps_nodes; + create_chan_between_nodes(&service_node.inner, &client_node.inner); + + let service_node_id = service_node.inner.node.get_our_node_id(); + let client_node_id = client_node.inner.node.get_our_node_id(); + let service_handler = service_node.liquidity_manager.lsps5_service_handler().unwrap(); + + // assert_lsps5_accept registers a webhook with app_name "App". + let app_name = LSPS5AppName::from_string("App".to_string()).unwrap(); + + // Register a webhook through the normal request flow. + assert_lsps5_accept(&service_node, &client_node); + + // The notification must be emitted before pruning to confirm the webhook is registered. + let result = service_handler.notify_payment_incoming(client_node_id); + assert!(result.is_ok()); + assert!(service_node.liquidity_manager.next_event().is_some(), "Webhook notification expected"); + + // Prune the webhook. + let prune_result = service_handler.prune_webhook(client_node_id, &app_name); + assert!(prune_result.is_ok(), "prune_webhook should succeed for a registered webhook"); + + // After pruning, notify should succeed but produce no event (no webhooks left). + let result_after = service_handler.notify_payment_incoming(client_node_id); + assert!(result_after.is_ok()); + assert!( + service_node.liquidity_manager.next_event().is_none(), + "No notification event expected after pruning" + ); + + // A second prune of the same app_name must fail. + let second_prune = service_handler.prune_webhook(client_node_id, &app_name); + assert!(second_prune.is_err(), "prune_webhook should fail when the webhook is already removed"); + + let _ = service_node_id; // suppress unused warning +} + +#[test] +fn prune_webhook_fails_for_unknown_counterparty() { + let chanmon_cfgs = create_chanmon_cfgs(2); + let node_cfgs = create_node_cfgs(2, &chanmon_cfgs); + let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]); + let nodes = create_network(2, &node_cfgs, &node_chanmgrs); + let (lsps_nodes, _) = lsps5_test_setup(nodes, Arc::new(DefaultTimeProvider)); + let LSPSNodes { service_node, client_node, .. } = lsps_nodes; + create_chan_between_nodes(&service_node.inner, &client_node.inner); + + let client_node_id = client_node.inner.node.get_our_node_id(); + let service_handler = service_node.liquidity_manager.lsps5_service_handler().unwrap(); + + let app_name = LSPS5AppName::from_string("SomeApp".to_string()).unwrap(); + + // Pruning for a peer with no state must return an error. + let result = service_handler.prune_webhook(client_node_id, &app_name); + assert!(result.is_err(), "prune_webhook should fail when counterparty has no state"); +} + +#[test] +fn prune_webhook_fails_for_unknown_app_name() { + let chanmon_cfgs = create_chanmon_cfgs(2); + let node_cfgs = create_node_cfgs(2, &chanmon_cfgs); + let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]); + let nodes = create_network(2, &node_cfgs, &node_chanmgrs); + let (lsps_nodes, _) = lsps5_test_setup(nodes, Arc::new(DefaultTimeProvider)); + let LSPSNodes { service_node, client_node, .. } = lsps_nodes; + create_chan_between_nodes(&service_node.inner, &client_node.inner); + + let client_node_id = client_node.inner.node.get_our_node_id(); + let service_handler = service_node.liquidity_manager.lsps5_service_handler().unwrap(); + + // Register one webhook. + assert_lsps5_accept(&service_node, &client_node); + + // Pruning with a different app_name must fail. + let unknown_app_name = LSPS5AppName::from_string("UnknownApp".to_string()).unwrap(); + let result = service_handler.prune_webhook(client_node_id, &unknown_app_name); + assert!(result.is_err(), "prune_webhook should fail for an unregistered app_name"); +}