Simplify message handling logic

Now that we have the `MessageQueueNotifierGuard`, we can be sure that we
always dropped locks before notifying. Hence, we can save *a lot* of
error-prone boilerplate that we used to ensure we'd only enqueue if we
dropped locks.

Author: tnull

lightning-liquidity/src/lsps1/client.rs

@@ -202,27 +202,19 @@ where
 	) -> LSPSRequestId {
 		let mut message_queue_notifier = self.pending_messages.notifier();
 
-		let (request_id, request_msg) = {
-			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::default()));
-			let mut peer_state_lock = inner_state_lock.lock().unwrap();
-
-			let request_id = crate::utils::generate_request_id(&self.entropy_source);
-			let request = LSPS1Request::CreateOrder(LSPS1CreateOrderRequest {
-				order,
-				refund_onchain_address,
-			});
-			let msg = LSPS1Message::Request(request_id.clone(), request).into();
-			peer_state_lock.pending_create_order_requests.insert(request_id.clone());
+		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::default()));
+		let mut peer_state_lock = inner_state_lock.lock().unwrap();
 
-			(request_id, Some(msg))
-		};
+		let request_id = crate::utils::generate_request_id(&self.entropy_source);
+		let request =
+			LSPS1Request::CreateOrder(LSPS1CreateOrderRequest { order, refund_onchain_address });
+		let msg = LSPS1Message::Request(request_id.clone(), request).into();
+		peer_state_lock.pending_create_order_requests.insert(request_id.clone());
 
-		if let Some(msg) = request_msg {
-			message_queue_notifier.enqueue(&counterparty_node_id, msg);
-		}
+		message_queue_notifier.enqueue(&counterparty_node_id, msg);
 
 		request_id
 	}
@@ -328,26 +320,19 @@ where
 	) -> LSPSRequestId {
 		let mut message_queue_notifier = self.pending_messages.notifier();
 
-		let (request_id, request_msg) = {
-			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::default()));
-			let mut peer_state_lock = inner_state_lock.lock().unwrap();
-
-			let request_id = crate::utils::generate_request_id(&self.entropy_source);
-			peer_state_lock.pending_get_order_requests.insert(request_id.clone());
+		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::default()));
+		let mut peer_state_lock = inner_state_lock.lock().unwrap();
 
-			let request =
-				LSPS1Request::GetOrder(LSPS1GetOrderRequest { order_id: order_id.clone() });
-			let msg = LSPS1Message::Request(request_id.clone(), request).into();
+		let request_id = crate::utils::generate_request_id(&self.entropy_source);
+		peer_state_lock.pending_get_order_requests.insert(request_id.clone());
 
-			(request_id, Some(msg))
-		};
+		let request = LSPS1Request::GetOrder(LSPS1GetOrderRequest { order_id: order_id.clone() });
+		let msg = LSPS1Message::Request(request_id.clone(), request).into();
 
-		if let Some(msg) = request_msg {
-			message_queue_notifier.enqueue(&counterparty_node_id, msg);
-		}
+		message_queue_notifier.enqueue(&counterparty_node_id, msg);
 
 		request_id
 	}

lightning-liquidity/src/lsps1/service.rs

@@ -255,66 +255,46 @@ where
 		payment: LSPS1PaymentInfo, created_at: LSPSDateTime,
 	) -> Result<(), APIError> {
 		let mut message_queue_notifier = self.pending_messages.notifier();
-		let (result, response) = {
-			let outer_state_lock = self.per_peer_state.read().unwrap();
-
-			match outer_state_lock.get(counterparty_node_id) {
-				Some(inner_state_lock) => {
-					let mut peer_state_lock = inner_state_lock.lock().unwrap();
-
-					match peer_state_lock.pending_requests.remove(&request_id) {
-						Some(LSPS1Request::CreateOrder(params)) => {
-							let order_id = self.generate_order_id();
-							let channel = OutboundCRChannel::new(
-								params.order.clone(),
-								created_at.clone(),
-								order_id.clone(),
-								payment.clone(),
-							);
-
-							peer_state_lock.insert_outbound_channel(order_id.clone(), channel);
-
-							let response = LSPS1Response::CreateOrder(LSPS1CreateOrderResponse {
-								order: params.order,
-								order_id,
-								order_state: LSPS1OrderState::Created,
-								created_at,
-								payment,
-								channel: None,
-							});
-
-							(Ok(()), Some(response))
-						},
-
-						_ => (
-							Err(APIError::APIMisuseError {
-								err: format!(
-									"No pending buy request for request_id: {:?}",
-									request_id
-								),
-							}),
-							None,
-						),
-					}
-				},
-				None => (
-					Err(APIError::APIMisuseError {
-						err: format!(
-							"No state for the counterparty exists: {:?}",
-							counterparty_node_id
-						),
-					}),
-					None,
-				),
-			}
-		};
 
-		if let Some(response) = response {
-			let msg = LSPS1Message::Response(request_id, response).into();
-			message_queue_notifier.enqueue(counterparty_node_id, msg);
-		}
+		let outer_state_lock = self.per_peer_state.read().unwrap();
+		match outer_state_lock.get(counterparty_node_id) {
+			Some(inner_state_lock) => {
+				let mut peer_state_lock = inner_state_lock.lock().unwrap();
+
+				match peer_state_lock.pending_requests.remove(&request_id) {
+					Some(LSPS1Request::CreateOrder(params)) => {
+						let order_id = self.generate_order_id();
+						let channel = OutboundCRChannel::new(
+							params.order.clone(),
+							created_at.clone(),
+							order_id.clone(),
+							payment.clone(),
+						);
 
-		result
+						peer_state_lock.insert_outbound_channel(order_id.clone(), channel);
+
+						let response = LSPS1Response::CreateOrder(LSPS1CreateOrderResponse {
+							order: params.order,
+							order_id,
+							order_state: LSPS1OrderState::Created,
+							created_at,
+							payment,
+							channel: None,
+						});
+						let msg = LSPS1Message::Response(request_id, response).into();
+						message_queue_notifier.enqueue(counterparty_node_id, msg);
+						Ok(())
+					},
+
+					_ => Err(APIError::APIMisuseError {
+						err: format!("No pending buy request for request_id: {:?}", request_id),
+					}),
+				}
+			},
+			None => Err(APIError::APIMisuseError {
+				err: format!("No state for the counterparty exists: {:?}", counterparty_node_id),
+			}),
+		}
 	}
 
 	fn handle_get_order_request(
@@ -383,54 +363,38 @@ where
 	) -> Result<(), APIError> {
 		let mut message_queue_notifier = self.pending_messages.notifier();
 
-		let (result, response) = {
-			let outer_state_lock = self.per_peer_state.read().unwrap();
-
-			match outer_state_lock.get(&counterparty_node_id) {
-				Some(inner_state_lock) => {
-					let mut peer_state_lock = inner_state_lock.lock().unwrap();
+		let outer_state_lock = self.per_peer_state.read().unwrap();
 
-					if let Some(outbound_channel) =
-						peer_state_lock.outbound_channels_by_order_id.get_mut(&order_id)
-					{
-						let config = &outbound_channel.config;
+		match outer_state_lock.get(&counterparty_node_id) {
+			Some(inner_state_lock) => {
+				let mut peer_state_lock = inner_state_lock.lock().unwrap();
 
-						let response = LSPS1Response::GetOrder(LSPS1CreateOrderResponse {
-							order_id,
-							order: config.order.clone(),
-							order_state,
-							created_at: config.created_at.clone(),
-							payment: config.payment.clone(),
-							channel,
-						});
-						(Ok(()), Some(response))
-					} else {
-						(
-							Err(APIError::APIMisuseError {
-								err: format!("Channel with order_id {} not found", order_id.0),
-							}),
-							None,
-						)
-					}
-				},
-				None => (
+				if let Some(outbound_channel) =
+					peer_state_lock.outbound_channels_by_order_id.get_mut(&order_id)
+				{
+					let config = &outbound_channel.config;
+
+					let response = LSPS1Response::GetOrder(LSPS1CreateOrderResponse {
+						order_id,
+						order: config.order.clone(),
+						order_state,
+						created_at: config.created_at.clone(),
+						payment: config.payment.clone(),
+						channel,
+					});
+					let msg = LSPS1Message::Response(request_id, response).into();
+					message_queue_notifier.enqueue(&counterparty_node_id, msg);
+					Ok(())
+				} else {
 					Err(APIError::APIMisuseError {
-						err: format!(
-							"No existing state with counterparty {}",
-							counterparty_node_id
-						),
-					}),
-					None,
-				),
-			}
-		};
-
-		if let Some(response) = response {
-			let msg = LSPS1Message::Response(request_id, response).into();
-			message_queue_notifier.enqueue(&counterparty_node_id, msg);
+						err: format!("Channel with order_id {} not found", order_id.0),
+					})
+				}
+			},
+			None => Err(APIError::APIMisuseError {
+				err: format!("No existing state with counterparty {}", counterparty_node_id),
+			}),
 		}
-
-		result
 	}
 
 	fn generate_order_id(&self) -> LSPS1OrderId {