Limit pending requests and peers in LSPS1 service

diff --git a/lightning-liquidity/src/lsps1/peer_state.rs b/lightning-liquidity/src/lsps1/peer_state.rs
index d2b806c..6e18897 100644
--- a/lightning-liquidity/src/lsps1/peer_state.rs
+++ b/lightning-liquidity/src/lsps1/peer_state.rs
@@ -22,6 +22,8 @@ use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum};
 
 use core::fmt;
 
+const MAX_PENDING_REQUESTS_PER_PEER: usize = 10;
+
 /// Indicates which payment method was used for the order.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum PaymentMethod {
@@ -340,6 +342,9 @@ impl PeerState {
 	pub(super) fn register_request(
 		&mut self, request_id: LSPSRequestId, request: LSPS1Request,
 	) -> Result<(), PeerStateError> {
+		if self.pending_requests_and_unpaid_orders() >= MAX_PENDING_REQUESTS_PER_PEER {
+			return Err(PeerStateError::TooManyPendingRequests);
+		}
 		if self.pending_requests.contains_key(&request_id) {
 			return Err(PeerStateError::DuplicateRequestId);
 		}
@@ -376,8 +381,10 @@ impl PeerState {
 		self.pending_requests.is_empty() && self.outbound_channels_by_order_id.is_empty()
 	}
 
-	pub(super) fn prune_pending_requests(&mut self) {
-		self.pending_requests.clear()
+	pub(super) fn prune_pending_requests(&mut self) -> usize {
+		let num_pruned = self.pending_requests.len();
+		self.pending_requests.clear();
+		num_pruned
 	}
 
 	pub(super) fn prune_expired_request_state(&mut self) {
@@ -389,6 +396,23 @@ impl PeerState {
 			true
 		});
 	}
+
+	fn pending_requests_and_unpaid_orders(&self) -> usize {
+		let pending_requests = self.pending_requests.len();
+		// We exclude paid and completed orders.
+		let unpaid_orders = self
+			.outbound_channels_by_order_id
+			.iter()
+			.filter(|(_, v)| {
+				!matches!(
+					v.state,
+					ChannelOrderState::OrderPaid { .. }
+						| ChannelOrderState::CompletedAndChannelOpened { .. }
+				)
+			})
+			.count();
+		pending_requests + unpaid_orders
+	}
 }
 
 impl_writeable_tlv_based!(PeerState, {
@@ -403,6 +427,7 @@ pub(super) enum PeerStateError {
 	DuplicateRequestId,
 	UnknownOrderId,
 	InvalidStateTransition(ChannelOrderStateError),
+	TooManyPendingRequests,
 }
 
 impl fmt::Display for PeerStateError {
@@ -412,6 +437,7 @@ impl fmt::Display for PeerStateError {
 			Self::DuplicateRequestId => write!(f, "duplicate request id"),
 			Self::UnknownOrderId => write!(f, "unknown order id"),
 			Self::InvalidStateTransition(e) => write!(f, "{}", e),
+			Self::TooManyPendingRequests => write!(f, "too many pending requests"),
 		}
 	}
 }
diff --git a/lightning-liquidity/src/lsps1/service.rs b/lightning-liquidity/src/lsps1/service.rs
index 9d58ea0..7cf0412 100644
--- a/lightning-liquidity/src/lsps1/service.rs
+++ b/lightning-liquidity/src/lsps1/service.rs
@@ -34,6 +34,7 @@ use crate::message_queue::MessageQueue;
 use crate::events::EventQueue;
 use crate::lsps0::ser::{
 	LSPSDateTime, LSPSProtocolMessageHandler, LSPSRequestId, LSPSResponseError,
+	LSPS0_CLIENT_REJECTED_ERROR_CODE,
 };
 use crate::persist::{
 	LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, LSPS1_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
@@ -62,6 +63,8 @@ pub struct LSPS1ServiceConfig {
 	pub supported_options: LSPS1Options,
 }
 
+const MAX_TOTAL_PEERS: usize = 100000;
+
 /// The main object allowing to send and receive bLIP-51 / LSPS1 messages.
 pub struct LSPS1ServiceHandler<
 	ES: EntropySource,
@@ -308,11 +311,30 @@ where
 
 		{
 			let mut outer_state_lock = self.per_peer_state.write().unwrap();
+			let num_peers = outer_state_lock.len();
 
-			let inner_state_lock = outer_state_lock
-				.entry(*counterparty_node_id)
-				.or_insert(Mutex::new(PeerState::default()));
-			let mut peer_state_lock = inner_state_lock.lock().unwrap();
+			let inner_state_entry = outer_state_lock.entry(*counterparty_node_id);
+
+			if matches!(inner_state_entry, Entry::Vacant(_)) && num_peers >= MAX_TOTAL_PEERS {
+				let response = LSPS1Response::CreateOrderError(LSPSResponseError {
+					code: LSPS0_CLIENT_REJECTED_ERROR_CODE,
+					message: "Reached maximum number of pending requests. Please try again later."
+						.to_string(),
+					data: None,
+				});
+				let msg = LSPS1Message::Response(request_id, response).into();
+				message_queue_notifier.enqueue(counterparty_node_id, msg);
+				return Err(LightningError {
+					err: format!(
+						"Dropping request from peer {} due to reaching maximally allowed number of total peers: {}",
+						counterparty_node_id, MAX_TOTAL_PEERS
+					),
+					action: ErrorAction::IgnoreAndLog(Level::Debug),
+				});
+			}
+
+			let mut peer_state_lock =
+				inner_state_entry.or_insert(Mutex::new(PeerState::default())).lock().unwrap();
 
 			let request = LSPS1Request::CreateOrder(params.clone());
 			peer_state_lock.register_request(request_id.clone(), request).map_err(|e| {
@@ -734,16 +756,19 @@ where
 		&self, message: Self::ProtocolMessage, counterparty_node_id: &PublicKey,
 	) -> Result<(), LightningError> {
 		match message {
-			LSPS1Message::Request(request_id, request) => match request {
-				LSPS1Request::GetInfo(_) => {
-					self.handle_get_info_request(request_id, counterparty_node_id)
-				},
-				LSPS1Request::CreateOrder(params) => {
-					self.handle_create_order_request(request_id, counterparty_node_id, params)
-				},
-				LSPS1Request::GetOrder(params) => {
-					self.handle_get_order_request(request_id, counterparty_node_id, params)
-				},
+			LSPS1Message::Request(request_id, request) => {
+				let res = match request {
+					LSPS1Request::GetInfo(_) => {
+						self.handle_get_info_request(request_id, counterparty_node_id)
+					},
+					LSPS1Request::CreateOrder(params) => {
+						self.handle_create_order_request(request_id, counterparty_node_id, params)
+					},
+					LSPS1Request::GetOrder(params) => {
+						self.handle_get_order_request(request_id, counterparty_node_id, params)
+					},
+				};
+				res
 			},
 			_ => {
 				debug_assert!(
diff --git a/lightning-liquidity/tests/lsps1_integration_tests.rs b/lightning-liquidity/tests/lsps1_integration_tests.rs
index 6318566..a177b33 100644
--- a/lightning-liquidity/tests/lsps1_integration_tests.rs
+++ b/lightning-liquidity/tests/lsps1_integration_tests.rs
@@ -34,6 +34,8 @@ use lightning::ln::functional_test_utils::{create_network, Node};
 use lightning_liquidity::lsps1::msgs::LSPS1OrderId;
 use lightning_liquidity::utils::time::TimeProvider;
 
+const MAX_PENDING_REQUESTS_PER_PEER: usize = 10;
+
 fn build_lsps1_configs(
 	supported_options: LSPS1Options,
 ) -> (LiquidityServiceConfig, LiquidityClientConfig) {
@@ -1139,3 +1141,92 @@ fn lsps1_expired_orders_are_pruned_and_not_persisted() {
 		}
 	}
 }
+
+#[test]
+fn max_pending_requests_per_peer_rejected() {
+	let chanmon_cfgs = create_chanmon_cfgs(2);
+	let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
+	let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
+	let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+
+	let supported_options = LSPS1Options {
+		min_required_channel_confirmations: 0,
+		min_funding_confirms_within_blocks: 6,
+		supports_zero_channel_reserve: true,
+		max_channel_expiry_blocks: 144,
+		min_initial_client_balance_sat: 10_000_000,
+		max_initial_client_balance_sat: 100_000_000,
+		min_initial_lsp_balance_sat: 100_000,
+		max_initial_lsp_balance_sat: 100_000_000,
+		min_channel_balance_sat: 100_000,
+		max_channel_balance_sat: 100_000_000,
+	};
+
+	let LSPSNodes { service_node, client_node } =
+		setup_test_lsps1_nodes(nodes, supported_options.clone());
+	let service_node_id = service_node.inner.node.get_our_node_id();
+	let client_node_id = client_node.inner.node.get_our_node_id();
+	let client_handler = client_node.liquidity_manager.lsps1_client_handler().unwrap();
+
+	let order_params = LSPS1OrderParams {
+		lsp_balance_sat: 100_000,
+		client_balance_sat: 10_000_000,
+		required_channel_confirmations: 0,
+		funding_confirms_within_blocks: 6,
+		channel_expiry_blocks: 144,
+		token: None,
+		announce_channel: true,
+	};
+
+	let refund_onchain_address =
+		Address::from_str("bc1p5uvtaxzkjwvey2tfy49k5vtqfpjmrgm09cvs88ezyy8h2zv7jhas9tu4yr")
+			.unwrap()
+			.assume_checked();
+
+	// Send MAX_PENDING_REQUESTS_PER_PEER create_order requests, all should succeed.
+	for _ in 0..MAX_PENDING_REQUESTS_PER_PEER {
+		let _ = client_handler.create_order(
+			&service_node_id,
+			order_params.clone(),
+			Some(refund_onchain_address.clone()),
+		);
+		let req_msg = get_lsps_message!(client_node, service_node_id);
+		let result = service_node.liquidity_manager.handle_custom_message(req_msg, client_node_id);
+		assert!(result.is_ok());
+		let event = service_node.liquidity_manager.next_event().unwrap();
+		assert!(matches!(
+			event,
+			LiquidityEvent::LSPS1Service(LSPS1ServiceEvent::RequestForPaymentDetails { .. })
+		));
+	}
+
+	// The next request should be rejected due to per-peer limit.
+	let rejected_req_id = client_handler.create_order(
+		&service_node_id,
+		order_params.clone(),
+		Some(refund_onchain_address),
+	);
+	let rejected_req_msg = get_lsps_message!(client_node, service_node_id);
+	let result =
+		service_node.liquidity_manager.handle_custom_message(rejected_req_msg, client_node_id);
+	assert!(result.is_err(), "We should have hit the per-peer limit");
+
+	let error_response = get_lsps_message!(service_node, client_node_id);
+	let result =
+		client_node.liquidity_manager.handle_custom_message(error_response, service_node_id);
+	assert!(result.is_err());
+
+	let event = client_node.liquidity_manager.next_event().unwrap();
+	if let LiquidityEvent::LSPS1Client(LSPS1ClientEvent::OrderRequestFailed {
+		request_id,
+		counterparty_node_id,
+		error,
+	}) = event
+	{
+		assert_eq!(request_id, rejected_req_id);
+		assert_eq!(counterparty_node_id, service_node_id);
+		assert_eq!(error.code, 1); // LSPS0_CLIENT_REJECTED_ERROR_CODE
+	} else {
+		panic!("Expected LSPS1ClientEvent::OrderRequestFailed event");
+	}
+}