LSPS2: DRY up pending request insertion/removal

.. which is a prefactor to also start checking the total number of
pending requests in the next commit.

Author: tnull

lightning-liquidity/src/lsps2/service.rs

@@ -11,14 +11,14 @@
 
 use crate::events::{Event, EventQueue};
 use crate::lsps0::ser::{
-	ProtocolMessageHandler, RequestId, ResponseError, LSPS0_CLIENT_REJECTED_ERROR_CODE,
+	LSPSMessage, 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};
 use crate::message_queue::MessageQueue;
 use crate::prelude::{new_hash_map, HashMap, String, ToString, Vec};
-use crate::sync::{Arc, Mutex, RwLock};
+use crate::sync::{Arc, Mutex, MutexGuard, RwLock};
 
 use lightning::events::HTLCDestination;
 use lightning::ln::channelmanager::{AChannelManager, InterceptId};
@@ -506,9 +506,9 @@ where
 
 			match outer_state_lock.get(counterparty_node_id) {
 				Some(inner_state_lock) => {
-					let mut peer_state = inner_state_lock.lock().unwrap();
+					let mut peer_state_lock = inner_state_lock.lock().unwrap();
 
-					match peer_state.pending_requests.remove(&request_id) {
+					match self.remove_pending_request(&mut peer_state_lock, &request_id) {
 						Some(LSPS2Request::GetInfo(_)) => {
 							let response = LSPS2Response::GetInfoError(ResponseError {
 								code: LSPS2_GET_INFO_REQUEST_UNRECOGNIZED_OR_STALE_TOKEN_ERROR_CODE,
@@ -562,9 +562,9 @@ where
 
 			match outer_state_lock.get(counterparty_node_id) {
 				Some(inner_state_lock) => {
-					let mut peer_state = inner_state_lock.lock().unwrap();
+					let mut peer_state_lock = inner_state_lock.lock().unwrap();
 
-					match peer_state.pending_requests.remove(&request_id) {
+					match self.remove_pending_request(&mut peer_state_lock, &request_id) {
 						Some(LSPS2Request::GetInfo(_)) => {
 							let response = LSPS2Response::GetInfo(GetInfoResponse {
 								opening_fee_params_menu: opening_fee_params_menu
@@ -621,9 +621,9 @@ where
 
 			match outer_state_lock.get(counterparty_node_id) {
 				Some(inner_state_lock) => {
-					let mut peer_state = inner_state_lock.lock().unwrap();
+					let mut peer_state_lock = inner_state_lock.lock().unwrap();
 
-					match peer_state.pending_requests.remove(&request_id) {
+					match self.remove_pending_request(&mut peer_state_lock, &request_id) {
 						Some(LSPS2Request::Buy(buy_request)) => {
 							{
 								let mut peer_by_intercept_scid =
@@ -638,10 +638,10 @@ where
 								user_channel_id,
 							);
 
-							peer_state
+							peer_state_lock
 								.intercept_scid_by_user_channel_id
 								.insert(user_channel_id, intercept_scid);
-							peer_state
+							peer_state_lock
 								.insert_outbound_channel(intercept_scid, outbound_jit_channel);
 
 							let response = LSPS2Response::Buy(BuyResponse {
@@ -992,35 +992,24 @@ where
 				.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() < 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, MAX_PENDING_REQUESTS_PER_PEER
-				);
+			let request = LSPS2Request::GetInfo(params.clone());
+			match self.insert_pending_request(
+				&mut peer_state_lock,
+				request_id.clone(),
+				*counterparty_node_id,
+				request,
+			) {
+				(Ok(()), msg) => {
+					let event = Event::LSPS2Service(LSPS2ServiceEvent::GetInfo {
+						request_id,
+						counterparty_node_id: *counterparty_node_id,
+						token: params.token,
+					});
+					self.pending_events.enqueue(event);
 
-				let result =
-					Err(LightningError { err, action: ErrorAction::IgnoreAndLog(Level::Debug) });
-				(result, msg)
+					(Ok(()), msg)
+				},
+				(e, msg) => (e, msg),
 			}
 		};
 
@@ -1123,37 +1112,25 @@ where
 				.or_insert(Mutex::new(PeerState::new()));
 			let mut peer_state_lock = inner_state_lock.lock().unwrap();
 
-			if peer_state_lock.pending_requests.len() < 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(()), 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, MAX_PENDING_REQUESTS_PER_PEER
-				);
-				let result =
-					Err(LightningError { err, action: ErrorAction::IgnoreAndLog(Level::Debug) });
+			let request = LSPS2Request::Buy(params.clone());
+			match self.insert_pending_request(
+				&mut peer_state_lock,
+				request_id.clone(),
+				*counterparty_node_id,
+				request,
+			) {
+				(Ok(()), msg) => {
+					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);
 
-				(result, msg)
+					(Ok(()), msg)
+				},
+				(e, msg) => (e, msg),
 			}
 		};
 
@@ -1163,6 +1140,39 @@ where
 
 		result
 	}
+
+	fn insert_pending_request<'a>(
+		&self, peer_state_lock: &mut MutexGuard<'a, PeerState>, request_id: RequestId,
+		counterparty_node_id: PublicKey, request: LSPS2Request,
+	) -> (Result<(), LightningError>, Option<LSPSMessage>) {
+		if peer_state_lock.pending_requests.len() < MAX_PENDING_REQUESTS_PER_PEER {
+			peer_state_lock.pending_requests.insert(request_id, request);
+			(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, MAX_PENDING_REQUESTS_PER_PEER
+			);
+			let result =
+				Err(LightningError { err, action: ErrorAction::IgnoreAndLog(Level::Debug) });
+
+			(result, msg)
+		}
+	}
+
+	fn remove_pending_request<'a>(
+		&self, peer_state_lock: &mut MutexGuard<'a, PeerState>, request_id: &RequestId,
+	) -> Option<LSPS2Request> {
+		peer_state_lock.pending_requests.remove(request_id)
+	}
 }
 
 impl<CM: Deref + Clone> ProtocolMessageHandler for LSPS2ServiceHandler<CM>