Forward channel closures and failed opens to the LSPS2 service

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),

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;
 					},
 				}

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<S: PaginatedKVStore + Send + Sync + 'static>(
+	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();