Introduce MessageQueueNotifier RAII type

Previously, when enqueuing new messages to the `MessageQueue`, we'd
directly attempt to call back into `PeerHandler::process_events` (via
the `process_msgs_callback`), which always held the risk of deadlocks on
re-entrancy, i.e., when we would still hold the peer state mutexes when
doing so. To mitigate this we had to always implement ~ugly patterns
ensuring we don't hold the `Mutex` when we call `MessageQueue::enqueue`.

Here, we instead introduce a `MessageQueueNotifier` RAII type that will
call `process_msgs_callback` when dropped, which allows us to clean up
the code and reduces the risk of reintroducing such deadlock bugs: we
now simply have to ensure that any method calling `enqueue` holds the
notifier before retrieving any locks.

Author: tnull

lightning-liquidity/src/lsps0/client.rs

@@ -49,6 +49,7 @@ where
 	/// specifcation](https://github.com/BitcoinAndLightningLayerSpecs/lsp/tree/main/LSPS0#lsps-specification-support-query)
 	/// for more information.
 	pub fn list_protocols(&self, counterparty_node_id: &PublicKey) {
+		let _msg_queue_notifier = self.pending_messages.notifier();
 		let msg = LSPS0Message::Request(
 			utils::generate_request_id(&self.entropy_source),
 			LSPS0Request::ListProtocols(ListProtocolsRequest {}),

lightning-liquidity/src/lsps0/service.rs

@@ -39,6 +39,7 @@ impl LSPS0ServiceHandler {
 	fn handle_request(
 		&self, request_id: RequestId, request: LSPS0Request, counterparty_node_id: &PublicKey,
 	) -> Result<(), lightning::ln::msgs::LightningError> {
+		let _msg_queue_notifier = self.pending_messages.notifier();
 		match request {
 			LSPS0Request::ListProtocols(_) => {
 				let msg = LSPS0Message::Response(

lightning-liquidity/src/lsps1/client.rs

@@ -84,6 +84,7 @@ where
 	///
 	/// [`SupportedOptionsReady`]: crate::lsps1::event::LSPS1ClientEvent::SupportedOptionsReady
 	pub fn request_supported_options(&self, counterparty_node_id: PublicKey) -> RequestId {
+		let _msg_queue_notifier = self.pending_messages.notifier();
 		let request_id = crate::utils::generate_request_id(&self.entropy_source);
 		{
 			let mut outer_state_lock = self.per_peer_state.write().unwrap();
@@ -191,25 +192,19 @@ where
 		&self, counterparty_node_id: &PublicKey, order: OrderParameters,
 		refund_onchain_address: Option<Address>,
 	) -> RequestId {
-		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(CreateOrderRequest { 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 _msg_queue_notifier = self.pending_messages.notifier();
+		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))
-		};
-
-		if let Some(msg) = request_msg {
-			self.pending_messages.enqueue(&counterparty_node_id, msg);
-		}
+		let request_id = crate::utils::generate_request_id(&self.entropy_source);
+		let request =
+			LSPS1Request::CreateOrder(CreateOrderRequest { order, refund_onchain_address });
+		let msg = LSPS1Message::Request(request_id.clone(), request).into();
+		self.pending_messages.enqueue(&counterparty_node_id, msg);
+		peer_state_lock.pending_create_order_requests.insert(request_id.clone());
 
 		request_id
 	}
@@ -310,25 +305,19 @@ where
 	pub fn check_order_status(
 		&self, counterparty_node_id: &PublicKey, order_id: OrderId,
 	) -> RequestId {
-		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 _msg_queue_notifier = self.pending_messages.notifier();
+		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(GetOrderRequest { order_id: order_id.clone() });
-			let msg = LSPS1Message::Request(request_id.clone(), request).into();
-
-			(request_id, Some(msg))
-		};
+		let request_id = crate::utils::generate_request_id(&self.entropy_source);
+		peer_state_lock.pending_get_order_requests.insert(request_id.clone());
 
-		if let Some(msg) = request_msg {
-			self.pending_messages.enqueue(&counterparty_node_id, msg);
-		}
+		let request = LSPS1Request::GetOrder(GetOrderRequest { order_id: order_id.clone() });
+		let msg = LSPS1Message::Request(request_id.clone(), request).into();
+		self.pending_messages.enqueue(&counterparty_node_id, msg);
 
 		request_id
 	}

lightning-liquidity/src/lsps2/client.rs

@@ -111,6 +111,7 @@ where
 	pub fn request_opening_params(
 		&self, counterparty_node_id: PublicKey, token: Option<String>,
 	) -> RequestId {
+		let _msg_queue_notifier = self.pending_messages.notifier();
 		let request_id = crate::utils::generate_request_id(&self.entropy_source);
 
 		{
@@ -151,6 +152,7 @@ where
 		&self, counterparty_node_id: PublicKey, payment_size_msat: Option<u64>,
 		opening_fee_params: OpeningFeeParams,
 	) -> Result<RequestId, APIError> {
+		let _msg_queue_notifier = self.pending_messages.notifier();
 		let request_id = crate::utils::generate_request_id(&self.entropy_source);
 
 		{