diff --git a/src/event.rs b/src/event.rs index 88501ae6ef..38c78c9388 100644 --- a/src/event.rs +++ b/src/event.rs @@ -2461,6 +2461,11 @@ where None }; + self.liquidity_source + .lsps2_service() + .handle_channel_closed(user_channel_id, channel_id, &counterparty_node_id) + .await; + let event = Event::ChannelClosed { channel_id, user_channel_id: UserChannelId(user_channel_id), diff --git a/src/liquidity/service/lsps2.rs b/src/liquidity/service/lsps2.rs index 14fff4b624..de14300292 100644 --- a/src/liquidity/service/lsps2.rs +++ b/src/liquidity/service/lsps2.rs @@ -22,7 +22,7 @@ use lightning_liquidity::lsps2::msgs::LSPS2RawOpeningFeeParams; use lightning_liquidity::lsps2::service::LSPS2ServiceConfig as LdkLSPS2ServiceConfig; use lightning_types::payment::PaymentHash; -use crate::logger::{log_error, LdkLogger}; +use crate::logger::{log_error, log_trace, LdkLogger}; use crate::types::{ChannelManager, KeysManager, LiquidityManager, PeerManager, Wallet}; use crate::{total_anchor_channels_reserve_sats, Config}; @@ -260,6 +260,50 @@ where } } + pub(crate) async fn handle_channel_closed( + &self, user_channel_id: u128, channel_id: ChannelId, counterparty_node_id: &PublicKey, + ) { + if let Some(lsps2_service_handler) = self.liquidity_manager.lsps2_service_handler() { + // A JIT channel closing before it became ready is a failed channel open. This errors + // for any channel that isn't such a JIT channel, which is the common case. + if let Err(e) = lsps2_service_handler + .channel_open_failed(counterparty_node_id, user_channel_id) + .await + { + log_trace!( + self.logger, + "LSPS2 service did not treat the closure of channel {} as a failed channel open: {:?}", + channel_id, + e + ); + } + + if let Err(e) = lsps2_service_handler.channel_closed(channel_id).await { + log_error!( + self.logger, + "LSPS2 service failed to handle ChannelClosed event: {:?}", + e + ); + } + } + } + + /// Fails the HTLCs held for the given JIT channel back, allowing the payer to retry. + async fn fail_channel_open(&self, counterparty_node_id: &PublicKey, user_channel_id: u128) { + if let Some(lsps2_service_handler) = self.liquidity_manager.lsps2_service_handler() { + if let Err(e) = lsps2_service_handler + .channel_open_failed(counterparty_node_id, user_channel_id) + .await + { + log_error!( + self.logger, + "LSPS2 service failed to handle failed channel open: {:?}", + e + ); + } + } + } + pub(crate) async fn handle_event(&self, event: LSPS2ServiceEvent) { match event { LSPS2ServiceEvent::GetInfo { request_id, counterparty_node_id, token } => { @@ -421,29 +465,33 @@ where return; }; - let init_features = if let Some(Some(peer_manager)) = - self.peer_manager.read().expect("lock").as_ref().map(|weak| weak.upgrade()) - { - // Fail if we're not connected to the prospective channel partner. - if let Some(peer) = peer_manager.peer_by_node_id(&their_network_key) { - peer.init_features - } else { - // TODO: We just silently fail here. Eventually we will need to remember - // the pending requests and regularly retry opening the channel until we - // succeed. - log_error!( - self.logger, - "Failed to open LSPS2 channel to {} due to peer not being not connected.", - their_network_key, - ); - return; - } - } else { + let peer_manager = self + .peer_manager + .read() + .expect("lock") + .as_ref() + .and_then(|weak| weak.upgrade()); + let Some(peer_manager) = peer_manager else { debug_assert!(false, "Failed to handle LSPS2ServiceEvent as peer manager isn't available. This should never happen.",); log_error!(self.logger, "Failed to handle LSPS2ServiceEvent as peer manager isn't available. This should never happen.",); return; }; + // Fail if we're not connected to the prospective channel partner. + let Some(peer) = peer_manager.peer_by_node_id(&their_network_key) else { + log_error!( + self.logger, + "Failed to open LSPS2 channel to {} due to peer not being not connected.", + their_network_key, + ); + // TODO: We fail the payment back here. Eventually we will need to remember + // the pending requests and regularly retry opening the channel until we + // succeed. + self.fail_channel_open(&their_network_key, user_channel_id).await; + return; + }; + let init_features = peer.init_features; + // Fail if we have insufficient onchain funds available. let over_provisioning_msat = amt_to_forward_msat .saturating_mul(service_config.channel_over_provisioning_ppm as u64) @@ -468,9 +516,10 @@ where "Unable to create channel due to insufficient funds. Available: {}sats, Required: {}sats", spendable_amount_sats, required_funds_sats, ); - // TODO: We just silently fail here. Eventually we will need to remember + // TODO: We fail the payment back here. Eventually we will need to remember // the pending requests and regularly retry opening the channel until we // succeed. + self.fail_channel_open(&their_network_key, user_channel_id).await; return; } @@ -517,9 +566,6 @@ where match result { Ok(_) => {}, Err(e) => { - // TODO: We just silently fail here. Eventually we will need to remember - // the pending requests and regularly retry opening the channel until we - // succeed. let zero_reserve_string = if service_config.disable_client_reserve { "0reserve " } else { "" }; log_error!( @@ -529,6 +575,10 @@ where their_network_key, e ); + // TODO: We fail the payment back here. Eventually we will need to remember + // the pending requests and regularly retry opening the channel until we + // succeed. + self.fail_channel_open(&their_network_key, user_channel_id).await; return; }, } diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index 691668c56a..4b94a8c6d8 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -7673,6 +7673,201 @@ async fn lsps2_lsp_trusts_client_but_client_does_not_claim() { ); } +/// Sets up an LSPS2 service around `service_store`, a client of it, and a payer with a channel to +/// the service. The service's on-chain wallet is funded with the given amount, and the client +/// accepts the service's channels without confirmations if `client_trusts_0conf` is set. +async fn setup_lsps2_nodes( + bitcoind: &BitcoinD, electrsd: &ElectrsD, service_store: S, service_funds_sat: u64, + client_trusts_0conf: bool, +) -> (TestNode, TestNode, TestNode) { + let esplora_url = format!("http://{}", electrsd.esplora_url.as_ref().unwrap()); + + let mut sync_config = EsploraSyncConfig::default(); + sync_config.background_sync_config = None; + + let lsps2_service_config = 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, + 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, + }; + + let service_config = random_config(); + setup_builder!(service_builder, service_config.node_config); + service_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); + service_builder.enable_liquidity_provider(lsps2_service_config); + let service_node = service_builder + .build_with_store(service_config.node_entropy.into(), service_store) + .unwrap(); + service_node.start().unwrap(); + let service_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.node_id(), + service_addr, + None, + client_trusts_0conf, + ); + let client_node = client_builder.build(client_config.node_entropy.into()).unwrap(); + client_node.start().unwrap(); + + let payer_config = random_config(); + setup_builder!(payer_builder, payer_config.node_config); + payer_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); + let payer_node = payer_builder.build(payer_config.node_entropy.into()).unwrap(); + payer_node.start().unwrap(); + + premine_blocks(&bitcoind.client, &electrsd.client).await; + for (node, funds_sat) in + [(&service_node, service_funds_sat), (&client_node, 10_000_000), (&payer_node, 10_000_000)] + { + let addr = node.onchain_payment().new_address().unwrap(); + distribute_funds_unconfirmed( + &bitcoind.client, + &electrsd.client, + vec![addr], + Amount::from_sat(funds_sat), + ) + .await; + } + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).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; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + service_node.sync_wallets().unwrap(); + payer_node.sync_wallets().unwrap(); + expect_channel_ready_event!(payer_node, service_node.node_id()); + expect_channel_ready_event!(service_node, payer_node.node_id()); + + (service_node, client_node, payer_node) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn lsps2_service_prunes_closed_channel_state() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let service_store = TestSyncStore::new(random_config().node_config.storage_dir_path.into()); + let (service_node, client_node, payer_node) = + setup_lsps2_nodes(&bitcoind, &electrsd, service_store.clone(), 10_000_000, true).await; + + let list_lsps2_peer_states = || async { + KVStore::list(&service_store, "lightning_liquidity_state", "lsps2_service").await.unwrap() + }; + + let invoice_description = + Bolt11InvoiceDescription::Direct(Description::new(String::from("asdf")).unwrap()); + let jit_amount_msat = 100_000_000; + let jit_invoice = client_node + .bolt11_payment() + .receive_via_jit_channel(jit_amount_msat, &invoice_description.into(), 1024, None) + .unwrap(); + + let payer_payment_id = payer_node.bolt11_payment().send(&jit_invoice, None).unwrap(); + expect_channel_pending_event!(service_node, client_node.node_id()); + expect_channel_ready_event!(service_node, client_node.node_id()); + expect_event!(service_node, PaymentForwarded); + expect_channel_pending_event!(client_node, service_node.node_id()); + let client_user_channel_id = expect_channel_ready_event!(client_node, service_node.node_id()); + expect_payment_successful_event!(payer_node, payer_payment_id, None); + expect_payment_received_event!(client_node, jit_amount_msat * 99 / 100); + + assert_eq!(list_lsps2_peer_states().await, vec![client_node.node_id().to_string()]); + + client_node.close_channel(&client_user_channel_id, service_node.node_id()).unwrap(); + expect_event!(client_node, ChannelClosed); + expect_event!(service_node, ChannelClosed); + + // The service drops the client's state once it holds no channel anymore, which happens in the + // background. + tokio::time::timeout(Duration::from_secs(common::INTEROP_TIMEOUT_SECS), async { + while !list_lsps2_peer_states().await.is_empty() { + tokio::time::sleep(Duration::from_millis(100)).await; + } + }) + .await + .expect("the service should prune the state of the closed JIT channel"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn lsps2_service_fails_payment_if_it_cannot_open_channel() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + // The service can cover the Anchor reserve for the payer's channel, but not the JIT channel. + let service_store = TestSyncStore::new(random_config().node_config.storage_dir_path.into()); + let (service_node, client_node, payer_node) = + setup_lsps2_nodes(&bitcoind, &electrsd, service_store, 60_000, true).await; + + let invoice_description = + Bolt11InvoiceDescription::Direct(Description::new(String::from("asdf")).unwrap()); + let jit_amount_msat = 100_000_000; + let jit_invoice = client_node + .bolt11_payment() + .receive_via_jit_channel(jit_amount_msat, &invoice_description.into(), 1024, None) + .unwrap(); + + payer_node.bolt11_payment().send(&jit_invoice, None).unwrap(); + expect_event!(payer_node, PaymentFailed); + + // Once the service can open the channel, a retry of the payment succeeds. + let service_addr = service_node.onchain_payment().new_address().unwrap(); + distribute_funds_unconfirmed( + &bitcoind.client, + &electrsd.client, + vec![service_addr], + Amount::from_sat(10_000_000), + ) + .await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; + service_node.sync_wallets().unwrap(); + + let payer_payment_id = payer_node.bolt11_payment().send(&jit_invoice, None).unwrap(); + expect_channel_pending_event!(service_node, client_node.node_id()); + expect_channel_ready_event!(service_node, client_node.node_id()); + expect_event!(service_node, PaymentForwarded); + expect_channel_pending_event!(client_node, service_node.node_id()); + expect_channel_ready_event!(client_node, service_node.node_id()); + expect_payment_successful_event!(payer_node, payer_payment_id, None); + expect_payment_received_event!(client_node, jit_amount_msat * 99 / 100); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn lsps2_service_fails_payment_if_channel_closes_before_ready() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + // The client doesn't trust the service's unconfirmed channels, leaving the JIT channel pending + // until its funding confirms. + let service_store = TestSyncStore::new(random_config().node_config.storage_dir_path.into()); + let (service_node, client_node, payer_node) = + setup_lsps2_nodes(&bitcoind, &electrsd, service_store, 10_000_000, false).await; + + let invoice_description = + Bolt11InvoiceDescription::Direct(Description::new(String::from("asdf")).unwrap()); + let jit_invoice = client_node + .bolt11_payment() + .receive_via_jit_channel(100_000_000, &invoice_description.into(), 1024, None) + .unwrap(); + + payer_node.bolt11_payment().send(&jit_invoice, None).unwrap(); + expect_channel_pending_event!(service_node, client_node.node_id()); + expect_channel_pending_event!(client_node, service_node.node_id()); + + let client_user_channel_id = client_node.list_channels().first().unwrap().user_channel_id; + client_node.force_close_channel(&client_user_channel_id, service_node.node_id(), None).unwrap(); + expect_event!(client_node, ChannelClosed); + expect_event!(service_node, ChannelClosed); + expect_event!(payer_node, PaymentFailed); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn payment_persistence_after_restart() { let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();