lsps2: Serialize lease negotiation

Make same-amount and variable callers wait for one in-flight LSPS2
request, then recheck the shared cache before negotiating.

Co-Authored-By: HAL 9000

Author: tnull

src/liquidity/client/lsps2/mod.rs

@@ -39,7 +39,7 @@ use crate::{Config, Error};
 use self::router::LSPS2LeaseParameters;
 use self::state::{
 	now_secs, LSPS2LeaseState, LeaseCacheTarget, LeaseCacheTargetId, LeaseCacheTargetStore,
-	PaymentLease, PaymentLeaseId, PaymentLeaseStore,
+	LeaseRequestKey, PaymentLease, PaymentLeaseId, PaymentLeaseStore, PendingLeaseRequestState,
 };
 
 async fn consume_after_persisted_removal<T, E, RF, CF, Fut>(
@@ -102,6 +102,7 @@ where
 	pub(crate) lease_store: Arc<PaymentLeaseStore<L>>,
 	pub(crate) lease_state: Mutex<LSPS2LeaseState>,
 	pub(crate) cache_target_store: Arc<LeaseCacheTargetStore<L>>,
+	pub(crate) pending_lease_request_state: Mutex<PendingLeaseRequestState>,
 	pub(crate) channel_manager: Arc<ChannelManager>,
 	pub(crate) keys_manager: Arc<KeysManager>,
 	pub(crate) discovery_done_rx: tokio::sync::watch::Receiver<bool>,
@@ -275,6 +276,17 @@ where
 		{
 			return Ok((lease, total_fee_msat, lsp, false));
 		}
+		let request_lock = self
+			.pending_lease_request_state
+			.lock()
+			.expect("lock")
+			.request_lock(LeaseRequestKey::Fixed(amount_msat));
+		let _request_guard = request_lock.lock().await;
+		if let Some((lease, total_fee_msat, lsp)) =
+			self.take_cached_fixed_lease(amount_msat).await?
+		{
+			return Ok((lease, total_fee_msat, lsp, false));
+		}
 
 		let all_offers = self.gather_lsps2_offers(connection_manager).await?;
 		let (cheapest_lsp, min_total_fee_msat, min_opening_params) = all_offers
@@ -341,6 +353,15 @@ where
 		if let Some((lease, proportional_fee, lsp)) = self.take_cached_variable_lease().await? {
 			return Ok((lease, proportional_fee, lsp, false));
 		}
+		let request_lock = self
+			.pending_lease_request_state
+			.lock()
+			.expect("lock")
+			.request_lock(LeaseRequestKey::Variable);
+		let _request_guard = request_lock.lock().await;
+		if let Some((lease, proportional_fee, lsp)) = self.take_cached_variable_lease().await? {
+			return Ok((lease, proportional_fee, lsp, false));
+		}
 
 		let all_offers = self.gather_lsps2_offers(connection_manager).await?;
 		let mut rejected_for_fee = false;

src/liquidity/client/lsps2/state.rs

@@ -1,6 +1,6 @@
 use std::collections::HashMap;
 use std::ops::Deref;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, Weak};
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use bitcoin::secp256k1::PublicKey;
@@ -9,6 +9,7 @@ use lightning::util::ser::{Readable, Writeable};
 use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum};
 use lightning_liquidity::lsps2::msgs::LSPS2OpeningFeeParams;
 use lightning_liquidity::lsps2::utils::compute_opening_fee;
+use tokio::sync::Mutex as AsyncMutex;
 
 use crate::data_store::{DataStore, StorableObject, StorableObjectId, StorableObjectUpdate};
 use crate::hex_utils;
@@ -261,6 +262,33 @@ fn merge_absolute_expiry(current: Option<u64>, new: Option<u64>) -> Option<u64>
 	}
 }
 
+#[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)]
+pub(crate) enum LeaseRequestKey {
+	Fixed(u64),
+	Variable,
+}
+
+#[derive(Default)]
+pub(crate) struct PendingLeaseRequestState {
+	locks: HashMap<LeaseRequestKey, Weak<AsyncMutex<()>>>,
+}
+
+impl PendingLeaseRequestState {
+	pub(crate) fn request_lock(&mut self, key: LeaseRequestKey) -> Arc<AsyncMutex<()>> {
+		self.prune();
+		if let Some(lock) = self.locks.get(&key).and_then(Weak::upgrade) {
+			return lock;
+		}
+		let lock = Arc::new(AsyncMutex::new(()));
+		self.locks.insert(key, Arc::downgrade(&lock));
+		lock
+	}
+
+	pub(crate) fn prune(&mut self) {
+		self.locks.retain(|_, lock| lock.strong_count() > 0);
+	}
+}
+
 fn is_cache_target_expired(target: &LeaseCacheTarget, now: u64) -> bool {
 	target.absolute_expiry.is_some_and(|expiry| expiry <= now)
 }
@@ -649,4 +677,18 @@ mod tests {
 		state.register(cache_target(id, None, 3));
 		assert_eq!(state.targets().first().unwrap().absolute_expiry, None);
 	}
+
+	#[test]
+	fn lease_requests_are_serialized_per_key() {
+		let mut state = PendingLeaseRequestState::default();
+
+		let fixed = state.request_lock(LeaseRequestKey::Fixed(1_000));
+		let same_fixed = state.request_lock(LeaseRequestKey::Fixed(1_000));
+		let other_fixed = state.request_lock(LeaseRequestKey::Fixed(2_000));
+		let variable = state.request_lock(LeaseRequestKey::Variable);
+
+		assert!(Arc::ptr_eq(&fixed, &same_fixed));
+		assert!(!Arc::ptr_eq(&fixed, &other_fixed));
+		assert!(!Arc::ptr_eq(&fixed, &variable));
+	}
 }

src/liquidity/mod.rs

@@ -38,7 +38,7 @@ use crate::io::{
 use crate::liquidity::client::lsps1::LSPS1Client;
 use crate::liquidity::client::lsps2::state::{
 	is_lease_usable, now_secs, read_lease_cache_targets, LSPS2LeaseState, LeaseCacheTargetState,
-	LeaseCacheTargetStore, PaymentLeaseStore,
+	LeaseCacheTargetStore, PaymentLeaseStore, PendingLeaseRequestState,
 };
 use crate::liquidity::client::lsps2::LSPS2Client;
 use crate::liquidity::service::lsps2::{LSPS2Service, LSPS2ServiceLiquiditySource};
@@ -361,6 +361,7 @@ where
 				lease_store,
 				lease_state: Mutex::new(lease_state),
 				cache_target_store,
+				pending_lease_request_state: Mutex::new(PendingLeaseRequestState::default()),
 				channel_manager: self.channel_manager.clone(),
 				keys_manager: self.keys_manager.clone(),
 				discovery_done_rx: discovery_done_rx.clone(),