lsps2: Fail over lease negotiation

Try the next eligible provider when a selected LSP rejects or times out
during the buy request. Skip remaining fee-menu entries from a failed
provider while preserving bounded whole-round retries.

Co-Authored-By: HAL 9000

Author: tnull

src/liquidity/client/lsps2/mod.rs

@@ -5,8 +5,9 @@
 // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
 // accordance with one or both of these licenses.
 
-use std::collections::{BTreeMap, HashMap};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use std::future::Future;
+use std::hash::Hash;
 use std::ops::Deref;
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::Duration;
@@ -69,6 +70,33 @@ pub(crate) struct JitInvoiceResponse {
 	pub(crate) allow_mpp: bool,
 }
 
+async fn try_lease_candidates<T, K, R, E, KF, AF, Fut>(
+	candidates: Vec<T>, candidate_key: KF, mut attempt: AF,
+) -> Result<R, E>
+where
+	K: Copy + Eq + Hash,
+	KF: Fn(&T) -> K,
+	AF: FnMut(T) -> Fut,
+	Fut: Future<Output = Result<R, E>>,
+{
+	let mut failed_candidates = HashSet::new();
+	let mut last_error = None;
+	for candidate in candidates {
+		let key = candidate_key(&candidate);
+		if failed_candidates.contains(&key) {
+			continue;
+		}
+		match attempt(candidate).await {
+			Ok(result) => return Ok(result),
+			Err(error) => {
+				failed_candidates.insert(key);
+				last_error = Some(error);
+			},
+		}
+	}
+	Err(last_error.expect("lease candidates are non-empty"))
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub(crate) enum JitInvoiceRequest {
 	Fixed { amount_msat: u64, absolute_expiry: Option<u64> },
@@ -331,13 +359,7 @@ where
 	) -> Result<(PaymentLease, u64, LspConfig), Error> {
 		let mut attempt = 1;
 		loop {
-			let result = self
-				.negotiate_fixed_lease_once(
-					amount_msat,
-					max_total_lsp_fee_limit_msat,
-					connection_manager,
-				)
-				.await;
+			let result = self.negotiate_fixed_lease_once(amount_msat, connection_manager).await;
 			match result {
 				Err(error) if should_retry_lease_negotiation(error, attempt) => {
 					log_warn!(
@@ -354,11 +376,10 @@ where
 	}
 
 	async fn negotiate_fixed_lease_once(
-		self: &Arc<Self>, amount_msat: u64, max_total_lsp_fee_limit_msat: Option<u64>,
-		connection_manager: &Arc<ConnectionManager<L>>,
+		self: &Arc<Self>, amount_msat: u64, connection_manager: &Arc<ConnectionManager<L>>,
 	) -> Result<(PaymentLease, u64, LspConfig), Error> {
 		let all_offers = self.gather_lsps2_offers(connection_manager).await?;
-		let (cheapest_lsp, min_total_fee_msat, min_opening_params) = all_offers
+		let mut candidates = all_offers
 			.into_iter()
 			.flat_map(|(lsp, resp)| {
 				resp.opening_fee_params_menu
@@ -382,11 +403,12 @@ where
 						.map(|fee| (lsp, fee, params))
 				}
 			})
-			.min_by_key(|(_, fee, _)| *fee)
-			.ok_or_else(|| {
-				log_error!(self.logger, "Failed to handle response from liquidity service",);
-				Error::LiquidityRequestFailed
-			})?;
+			.collect::<Vec<_>>();
+		candidates.sort_unstable_by_key(|(_, fee, _)| *fee);
+		let min_total_fee_msat = candidates.first().map(|(_, fee, _)| *fee).ok_or_else(|| {
+			log_error!(self.logger, "Failed to handle response from liquidity service",);
+			Error::LiquidityRequestFailed
+		})?;
 
 		if let Some(max_total_lsp_fee_limit_msat) = self.config.lsps2_max_total_lsp_fee_limit_msat {
 			if min_total_fee_msat > max_total_lsp_fee_limit_msat {
@@ -396,23 +418,44 @@ where
 				);
 				return Err(Error::LiquidityFeeTooHigh);
 			}
+			candidates.retain(|(_, fee, _)| *fee <= max_total_lsp_fee_limit_msat);
 		}
 
-		log_debug!(
-			self.logger,
-			"Choosing cheapest liquidity offer from LSP {}, will pay {}msat in total LSP fees",
-			cheapest_lsp.node_id,
-			min_total_fee_msat
-		);
-
-		let negotiated_lease = self
-			.lsps2_send_buy_request(
-				Some(amount_msat),
-				min_opening_params,
-				Some(&cheapest_lsp.node_id),
-			)
-			.await?;
-		Ok((negotiated_lease, min_total_fee_msat, cheapest_lsp))
+		try_lease_candidates(
+			candidates,
+			|(lsp, _, _)| lsp.node_id,
+			|(lsp, total_fee_msat, opening_params)| {
+				let client = Arc::clone(self);
+				async move {
+					log_debug!(
+						client.logger,
+						"Choosing liquidity offer from LSP {}, will pay {}msat in total LSP fees",
+						lsp.node_id,
+						total_fee_msat
+					);
+					match client
+						.lsps2_send_buy_request(
+							Some(amount_msat),
+							opening_params,
+							Some(&lsp.node_id),
+						)
+						.await
+					{
+						Ok(lease) => Ok((lease, total_fee_msat, lsp)),
+						Err(error) => {
+							log_warn!(
+								client.logger,
+								"Failed negotiating LSPS2 payment lease with LSP {}, trying the next candidate: {}",
+								lsp.node_id,
+								error
+							);
+							Err(error)
+						},
+					}
+				}
+			},
+		)
+		.await
 	}
 
 	async fn acquire_variable_lease(
@@ -445,12 +488,7 @@ where
 	) -> Result<(PaymentLease, u64, LspConfig), Error> {
 		let mut attempt = 1;
 		loop {
-			let result = self
-				.negotiate_variable_lease_once(
-					max_proportional_lsp_fee_limit_ppm_msat,
-					connection_manager,
-				)
-				.await;
+			let result = self.negotiate_variable_lease_once(connection_manager).await;
 			match result {
 				Err(error) if should_retry_lease_negotiation(error, attempt) => {
 					log_warn!(
@@ -467,12 +505,11 @@ where
 	}
 
 	async fn negotiate_variable_lease_once(
-		self: &Arc<Self>, max_proportional_lsp_fee_limit_ppm_msat: Option<u64>,
-		connection_manager: &Arc<ConnectionManager<L>>,
+		self: &Arc<Self>, connection_manager: &Arc<ConnectionManager<L>>,
 	) -> Result<(PaymentLease, u64, LspConfig), Error> {
 		let all_offers = self.gather_lsps2_offers(connection_manager).await?;
 		let mut rejected_for_fee = false;
-		let (cheapest_lsp, min_prop_fee_ppm_msat, min_opening_params) = all_offers
+		let mut candidates = all_offers
 			.into_iter()
 			.flat_map(|(lsp, resp)| {
 				resp.opening_fee_params_menu.into_iter().map(move |params| (lsp.clone(), params))
@@ -489,29 +526,52 @@ where
 				rejected_for_fee |= !allowed;
 				allowed
 			})
-			.min_by_key(|(_, ppm, _)| *ppm)
-			.ok_or_else(|| {
-				if rejected_for_fee {
-					log_error!(
-						self.logger,
-						"Failed to request inbound JIT channel as all LSP offers exceed our configured fee limit"
-					);
-					return Error::LiquidityFeeTooHigh;
-				}
+			.collect::<Vec<_>>();
+		candidates.sort_unstable_by_key(|(_, ppm, _)| *ppm);
+		if candidates.is_empty() {
+			return Err(if rejected_for_fee {
+				log_error!(
+					self.logger,
+					"Failed to request inbound JIT channel as all LSP offers exceed our configured fee limit"
+				);
+				Error::LiquidityFeeTooHigh
+			} else {
 				log_error!(self.logger, "Failed to handle response from liquidity service",);
 				Error::LiquidityRequestFailed
-			})?;
-		log_debug!(
-			self.logger,
-			"Choosing cheapest liquidity offer from LSP {}, will pay {}ppm msat in proportional LSP fees",
-			cheapest_lsp.node_id,
-			min_prop_fee_ppm_msat
-		);
-
-		let negotiated_lease = self
-			.lsps2_send_buy_request(None, min_opening_params, Some(&cheapest_lsp.node_id))
-			.await?;
-		Ok((negotiated_lease, min_prop_fee_ppm_msat, cheapest_lsp))
+			});
+		}
+
+		try_lease_candidates(
+			candidates,
+			|(lsp, _, _)| lsp.node_id,
+			|(lsp, proportional_fee_ppm_msat, opening_params)| {
+				let client = Arc::clone(self);
+				async move {
+					log_debug!(
+						client.logger,
+						"Choosing liquidity offer from LSP {}, will pay {}ppm msat in proportional LSP fees",
+						lsp.node_id,
+						proportional_fee_ppm_msat
+					);
+					match client
+						.lsps2_send_buy_request(None, opening_params, Some(&lsp.node_id))
+						.await
+					{
+						Ok(lease) => Ok((lease, proportional_fee_ppm_msat, lsp)),
+						Err(error) => {
+							log_warn!(
+								client.logger,
+								"Failed negotiating LSPS2 payment lease with LSP {}, trying the next candidate: {}",
+								lsp.node_id,
+								error
+							);
+							Err(error)
+						},
+					}
+				}
+			},
+		)
+		.await
 	}
 
 	fn schedule_fixed_lease_refill(
@@ -1050,6 +1110,32 @@ mod tests {
 		assert!(!should_retry_lease_negotiation(Error::LiquidityFeeTooHigh, 1));
 		assert!(!should_retry_lease_negotiation(Error::LiquiditySourceUnavailable, 1));
 	}
+
+	#[tokio::test]
+	async fn lease_negotiation_fails_over_between_lsps() {
+		let candidates = vec![(1, 10), (1, 20), (2, 30)];
+		let attempted_lsps = Arc::new(Mutex::new(Vec::new()));
+		let attempted_lsps_ref = Arc::clone(&attempted_lsps);
+		let result = try_lease_candidates(
+			candidates,
+			|candidate| candidate.0,
+			move |candidate| {
+				let attempted_lsps = Arc::clone(&attempted_lsps_ref);
+				async move {
+					attempted_lsps.lock().unwrap().push(candidate.0);
+					if candidate.0 == 1 {
+						Err(())
+					} else {
+						Ok(candidate.1)
+					}
+				}
+			},
+		)
+		.await;
+
+		assert_eq!(result, Ok(30));
+		assert_eq!(*attempted_lsps.lock().unwrap(), vec![1, 2]);
+	}
 }
 
 pub(crate) mod router;