LSPS2: Introduce max_pending_requests_per_peer service limit

We introduce a new `LSPS2ServiceLimits` object and use it to limit the
number of pending (get_info and buy) requests per peer.

Author: tnull

lightning-liquidity/src/lsps0/ser.rs

@@ -41,7 +41,7 @@ pub(crate) const JSONRPC_ERROR_FIELD_KEY: &str = "error";
 pub(crate) const JSONRPC_INVALID_MESSAGE_ERROR_CODE: i32 = -32700;
 pub(crate) const JSONRPC_INVALID_MESSAGE_ERROR_MESSAGE: &str = "parse error";
 
-pub(crate) const _LSPS0_CLIENT_REJECTED_ERROR_CODE: i32 = 1;
+pub(crate) const LSPS0_CLIENT_REJECTED_ERROR_CODE: i32 = 1;
 
 #[derive(Copy, Clone, Debug, PartialEq, Eq)]
 pub(crate) enum LSPSMethod {

lightning-liquidity/src/lsps2/service.rs

@@ -10,7 +10,9 @@
 //! Contains the main LSPS2 server-side object, [`LSPS2ServiceHandler`].
 
 use crate::events::{Event, EventQueue};
-use crate::lsps0::ser::{ProtocolMessageHandler, RequestId, ResponseError};
+use crate::lsps0::ser::{
+	ProtocolMessageHandler, RequestId, ResponseError, LSPS0_CLIENT_REJECTED_ERROR_CODE,
+};
 use crate::lsps2::event::LSPS2ServiceEvent;
 use crate::lsps2::payment_queue::{InterceptedHTLC, PaymentQueue};
 use crate::lsps2::utils::{compute_opening_fee, is_valid_opening_fee_params};
@@ -40,13 +42,33 @@ use crate::lsps2::msgs::{
 	LSPS2_GET_INFO_REQUEST_UNRECOGNIZED_OR_STALE_TOKEN_ERROR_CODE,
 };
 
+/// The default value applied for [`LSPS2ServiceLimits::max_pending_requests_per_peer`].
+pub const DEFAULT_MAX_PENDING_REQUESTS_PER_PEER: usize = 10;
+
 /// Server-side configuration options for JIT channels.
 #[derive(Clone, Debug)]
 pub struct LSPS2ServiceConfig {
 	/// Used to calculate the promise for channel parameters supplied to clients.
 	///
 	/// Note: If this changes then old promises given out will be considered invalid.
 	pub promise_secret: [u8; 32],
+	/// Configuration limits for JIT channels.
+	pub service_limits: LSPS2ServiceLimits,
+}
+
+/// Server-side configuration limits for JIT channels.
+#[derive(Clone, Debug)]
+pub struct LSPS2ServiceLimits {
+	/// The maximum number of pending requests we allow on a per-peer basis.
+	///
+	/// Any requests beyond this limit will be ignored.
+	pub max_pending_requests_per_peer: usize,
+}
+
+impl Default for LSPS2ServiceLimits {
+	fn default() -> Self {
+		Self { max_pending_requests_per_peer: DEFAULT_MAX_PENDING_REQUESTS_PER_PEER }
+	}
 }
 
 /// Information about the initial payment size and JIT channel opening fee.
@@ -982,21 +1004,51 @@ where
 	fn handle_get_info_request(
 		&self, request_id: RequestId, counterparty_node_id: &PublicKey, params: GetInfoRequest,
 	) -> Result<(), LightningError> {
-		let mut outer_state_lock = self.per_peer_state.write().unwrap();
-		let inner_state_lock: &mut Mutex<PeerState> =
-			outer_state_lock.entry(*counterparty_node_id).or_insert(Mutex::new(PeerState::new()));
-		let mut peer_state_lock = inner_state_lock.lock().unwrap();
-		peer_state_lock
-			.pending_requests
-			.insert(request_id.clone(), LSPS2Request::GetInfo(params.clone()));
-
-		let event = Event::LSPS2Service(LSPS2ServiceEvent::GetInfo {
-			request_id,
-			counterparty_node_id: *counterparty_node_id,
-			token: params.token,
-		});
-		self.pending_events.enqueue(event);
-		Ok(())
+		let (result, response) = {
+			let mut outer_state_lock = self.per_peer_state.write().unwrap();
+			let inner_state_lock: &mut Mutex<PeerState> = outer_state_lock
+				.entry(*counterparty_node_id)
+				.or_insert(Mutex::new(PeerState::new()));
+			let mut peer_state_lock = inner_state_lock.lock().unwrap();
+			if peer_state_lock.pending_requests.len()
+				< self.config.service_limits.max_pending_requests_per_peer
+			{
+				peer_state_lock
+					.pending_requests
+					.insert(request_id.clone(), LSPS2Request::GetInfo(params.clone()));
+
+				let event = Event::LSPS2Service(LSPS2ServiceEvent::GetInfo {
+					request_id,
+					counterparty_node_id: *counterparty_node_id,
+					token: params.token,
+				});
+				self.pending_events.enqueue(event);
+				(Ok(()), None)
+			} else {
+				let response = LSPS2Response::GetInfoError(ResponseError {
+					code: LSPS0_CLIENT_REJECTED_ERROR_CODE,
+					message: "Reached maximum number of pending requests. Please try again later."
+						.to_string(),
+					data: None,
+				});
+				let msg = Some(LSPS2Message::Response(request_id, response).into());
+
+				let err = format!(
+					"Peer {} reached maximum number of pending requests: {}",
+					counterparty_node_id, self.config.service_limits.max_pending_requests_per_peer
+				);
+
+				let result =
+					Err(LightningError { err, action: ErrorAction::IgnoreAndLog(Level::Debug) });
+				(result, msg)
+			}
+		};
+
+		if let Some(msg) = response {
+			self.pending_messages.enqueue(counterparty_node_id, msg);
+		}
+
+		result
 	}
 
 	fn handle_buy_request(
@@ -1070,7 +1122,6 @@ where
 		}
 
 		// TODO: if payment_size_msat is specified, make sure our node has sufficient incoming liquidity from public network to receive it.
-
 		if !is_valid_opening_fee_params(&params.opening_fee_params, &self.config.promise_secret) {
 			let response = LSPS2Response::BuyError(ResponseError {
 				code: LSPS2_BUY_REQUEST_INVALID_OPENING_FEE_PARAMS_ERROR_CODE,
@@ -1085,26 +1136,54 @@ where
 			});
 		}
 
-		{
+		let (result, response) = {
 			let mut outer_state_lock = self.per_peer_state.write().unwrap();
 			let inner_state_lock = outer_state_lock
 				.entry(*counterparty_node_id)
 				.or_insert(Mutex::new(PeerState::new()));
 			let mut peer_state_lock = inner_state_lock.lock().unwrap();
-			peer_state_lock
-				.pending_requests
-				.insert(request_id.clone(), LSPS2Request::Buy(params.clone()));
-		}
 
-		let event = Event::LSPS2Service(LSPS2ServiceEvent::BuyRequest {
-			request_id,
-			counterparty_node_id: *counterparty_node_id,
-			opening_fee_params: params.opening_fee_params,
-			payment_size_msat: params.payment_size_msat,
-		});
-		self.pending_events.enqueue(event);
+			if peer_state_lock.pending_requests.len()
+				< self.config.service_limits.max_pending_requests_per_peer
+			{
+				peer_state_lock
+					.pending_requests
+					.insert(request_id.clone(), LSPS2Request::Buy(params.clone()));
+
+				let event = Event::LSPS2Service(LSPS2ServiceEvent::BuyRequest {
+					request_id,
+					counterparty_node_id: *counterparty_node_id,
+					opening_fee_params: params.opening_fee_params,
+					payment_size_msat: params.payment_size_msat,
+				});
+				self.pending_events.enqueue(event);
 
-		Ok(())
+				(Ok(()), None)
+			} else {
+				let response = LSPS2Response::BuyError(ResponseError {
+					code: LSPS0_CLIENT_REJECTED_ERROR_CODE,
+					message: "Reached maximum number of pending requests. Please try again later."
+						.to_string(),
+					data: None,
+				});
+				let msg = Some(LSPS2Message::Response(request_id, response).into());
+
+				let err = format!(
+					"Peer {} reached maximum number of pending requests: {}",
+					counterparty_node_id, self.config.service_limits.max_pending_requests_per_peer
+				);
+				let result =
+					Err(LightningError { err, action: ErrorAction::IgnoreAndLog(Level::Debug) });
+
+				(result, msg)
+			}
+		};
+
+		if let Some(msg) = response {
+			self.pending_messages.enqueue(counterparty_node_id, msg);
+		}
+
+		result
 	}
 }
 

lightning-liquidity/tests/lsps2_integration_tests.rs

@@ -83,7 +83,8 @@ fn create_jit_invoice(
 #[test]
 fn invoice_generation_flow() {
 	let promise_secret = [42; 32];
-	let lsps2_service_config = LSPS2ServiceConfig { promise_secret };
+	let service_limits = Default::default();
+	let lsps2_service_config = LSPS2ServiceConfig { promise_secret, service_limits };
 	let service_config = LiquidityServiceConfig {
 		#[cfg(lsps1_service)]
 		lsps1_service_config: None,