Fix/gateway message processing

libs/traits/src/liquidity_pools.rs

@@ -152,7 +152,7 @@ pub trait MessageProcessor {
 	/// Process a message.
 	fn process(msg: Self::Message) -> (DispatchResult, Weight);
 
-	/// Max weight that processing a message can take
+	/// Max weight that processing a message can take.
 	fn max_processing_weight(msg: &Self::Message) -> Weight;
 }
 

pallets/liquidity-pools-gateway-queue/src/lib.rs

@@ -202,9 +202,15 @@ pub mod pallet {
 		fn service_message_queue(max_weight: Weight) -> Weight {
 			let mut weight_used = Weight::zero();
 
-			let mut processed_entries = Vec::new();
+			let mut nonces = MessageQueue::<T>::iter_keys().collect::<Vec<_>>();
+			nonces.sort();
+
+			for nonce in nonces {
+				let message =
+					MessageQueue::<T>::get(nonce).expect("valid nonce ensured by `iter_keys`");
+
+				weight_used.saturating_accrue(T::DbWeight::get().reads(1));
 
-			for (nonce, message) in MessageQueue::<T>::iter() {
 				let remaining_weight = max_weight.saturating_sub(weight_used);
 				let next_weight = T::MessageProcessor::max_processing_weight(&message);
 
@@ -214,36 +220,21 @@ pub mod pallet {
 				}
 
 				let weight = match Self::process_message_and_deposit_event(nonce, message.clone()) {
-					(Ok(()), weight) => {
-						// Extra weight breakdown:
-						//
-						// 1 read for the message
-						// 1 write for the message removal
-						weight.saturating_add(T::DbWeight::get().reads_writes(1, 1))
-					}
+					(Ok(()), weight) => weight,
 					(Err(e), weight) => {
 						FailedMessageQueue::<T>::insert(nonce, (message, e));
 
-						// Extra weight breakdown:
-						//
-						// 1 read for the message
 						// 1 write for the failed message
-						// 1 write for the message removal
-						weight.saturating_add(T::DbWeight::get().reads_writes(1, 2))
+						weight.saturating_add(T::DbWeight::get().writes(1))
 					}
 				};
 
-				processed_entries.push(nonce);
-
-				weight_used = weight_used.saturating_add(weight);
+				weight_used.saturating_accrue(weight);
 
-				if weight_used.all_gte(max_weight) {
-					break;
-				}
-			}
+				MessageQueue::<T>::remove(nonce);
 
-			for entry in processed_entries {
-				MessageQueue::<T>::remove(entry);
+				// 1 write for removing the message
+				weight_used.saturating_accrue(T::DbWeight::get().writes(1));
 			}
 
 			weight_used

pallets/liquidity-pools-gateway/src/lib.rs

@@ -35,14 +35,17 @@ use cfg_traits::liquidity_pools::{
 	OutboundMessageHandler, RouterProvider,
 };
 use cfg_types::domain_address::{Domain, DomainAddress};
-use frame_support::{dispatch::DispatchResult, pallet_prelude::*};
+use frame_support::{
+	dispatch::DispatchResult,
+	pallet_prelude::*,
+	storage::{with_transaction, TransactionOutcome},
+};
 use frame_system::pallet_prelude::{ensure_signed, OriginFor};
 use message::GatewayMessage;
 use orml_traits::GetByKey;
 pub use pallet::*;
 use parity_scale_codec::FullCodec;
 use sp_arithmetic::traits::{BaseArithmetic, EnsureAddAssign, One};
-use sp_runtime::SaturatedConversion;
 use sp_std::convert::TryInto;
 
 use crate::{
@@ -469,8 +472,8 @@ pub mod pallet {
 				router_ids.iter().any(|x| x == &router_id),
 				Error::<T>::UnknownRouter
 			);
-			// Message recovery shouldn't be supported for setups that have less than 1
-			// router since no proofs are required in that case.
+			// Message recovery shouldn't be supported for setups that have less than 2
+			// routers since no proofs are required in that case.
 			ensure!(router_ids.len() > 1, Error::<T>::NotEnoughRoutersForDomain);
 
 			let session_id = SessionIdStore::<T>::get();
@@ -621,45 +624,33 @@ pub mod pallet {
 		type Message = GatewayMessage<T::Message, T::RouterId>;
 
 		fn process(msg: Self::Message) -> (DispatchResult, Weight) {
-			match msg {
-				GatewayMessage::Inbound {
-					domain_address,
-					message,
-					router_id,
-				} => {
-					let mut counter = 0;
-
-					let res = Self::process_inbound_message(
+			// The #[transactional] macro only works for functions that return a
+			// `DispatchResult` therefore, we need to manually add this here.
+			let res = with_transaction(|| {
+				let res = match msg {
+					GatewayMessage::Inbound {
 						domain_address,
 						message,
 						router_id,
-						&mut counter,
-					);
-
-					let weight = match counter {
-						0 => LP_DEFENSIVE_WEIGHT / 10,
-						n => LP_DEFENSIVE_WEIGHT.saturating_mul(n),
-					};
+					} => Self::process_inbound_message(domain_address, message, router_id),
+					GatewayMessage::Outbound { message, router_id } => {
+						T::MessageSender::send(router_id, T::Sender::get(), message)
+					}
+				};
 
-					(res, weight)
+				if res.is_ok() {
+					TransactionOutcome::Commit(res)
+				} else {
+					TransactionOutcome::Rollback(res)
 				}
-				GatewayMessage::Outbound { message, router_id } => {
-					let res = T::MessageSender::send(router_id, T::Sender::get(), message);
+			});
 
-					(res, LP_DEFENSIVE_WEIGHT)
-				}
-			}
+			(res, LP_DEFENSIVE_WEIGHT)
 		}
 
-		/// Returns the max processing weight for a message, based on its
-		/// direction.
-		fn max_processing_weight(msg: &Self::Message) -> Weight {
-			match msg {
-				GatewayMessage::Inbound { message, .. } => {
-					LP_DEFENSIVE_WEIGHT.saturating_mul(message.submessages().len().saturated_into())
-				}
-				GatewayMessage::Outbound { .. } => LP_DEFENSIVE_WEIGHT,
-			}
+		/// Returns the maximum weight for processing one message.
+		fn max_processing_weight(_: &Self::Message) -> Weight {
+			LP_DEFENSIVE_WEIGHT
 		}
 	}
 

pallets/liquidity-pools-gateway/src/message_processing.rs

@@ -1,6 +1,5 @@
 use cfg_traits::liquidity_pools::{
-	InboundMessageHandler, LpMessageBatch, LpMessageHash, LpMessageProof, MessageHash,
-	MessageQueue, RouterProvider,
+	InboundMessageHandler, LpMessageHash, LpMessageProof, MessageHash, MessageQueue, RouterProvider,
 };
 use cfg_types::domain_address::{Domain, DomainAddress};
 use frame_support::{
@@ -333,7 +332,11 @@ impl<T: Config> Pallet<T> {
 				// we can return.
 				None => return Ok(()),
 				Some(stored_inbound_entry) => match stored_inbound_entry {
-					InboundEntry::Message(message_entry) => message = Some(message_entry.message),
+					InboundEntry::Message(message_entry)
+						if message_entry.session_id == session_id =>
+					{
+						message = Some(message_entry.message)
+					}
 					InboundEntry::Proof(proof_entry)
 						if proof_entry.has_valid_vote_for_session(session_id) =>
 					{
@@ -349,10 +352,10 @@ impl<T: Config> Pallet<T> {
 		}
 
 		if let Some(msg) = message {
-			Self::execute_post_voting_dispatch(message_hash, router_ids, expected_proof_count)?;
-
 			T::InboundMessageHandler::handle(domain_address.clone(), msg)?;
 
+			Self::execute_post_voting_dispatch(message_hash, router_ids, expected_proof_count)?;
+
 			Self::deposit_event(Event::<T>::InboundMessageExecuted {
 				domain_address,
 				message_hash,
@@ -401,42 +404,36 @@ impl<T: Config> Pallet<T> {
 		domain_address: DomainAddress,
 		message: T::Message,
 		router_id: T::RouterId,
-		counter: &mut u64,
 	) -> DispatchResult {
 		let router_ids = Self::get_router_ids_for_domain(domain_address.domain())?;
 		let session_id = SessionIdStore::<T>::get();
 		let expected_proof_count = Self::get_expected_proof_count(&router_ids)?;
+		let message_hash = message.get_message_hash();
+		let inbound_entry: InboundEntry<T> = InboundEntry::create(
+			message.clone(),
+			session_id,
+			domain_address.clone(),
+			expected_proof_count,
+		);
 
-		for submessage in message.submessages() {
-			counter.ensure_add_assign(1)?;
+		inbound_entry.validate(&router_ids, &router_id.clone())?;
 
-			let message_hash = submessage.get_message_hash();
+		Self::upsert_pending_entry(message_hash, &router_id, inbound_entry)?;
 
-			let inbound_entry: InboundEntry<T> = InboundEntry::create(
-				submessage.clone(),
-				session_id,
-				domain_address.clone(),
-				expected_proof_count,
-			);
-
-			inbound_entry.validate(&router_ids, &router_id.clone())?;
-			Self::upsert_pending_entry(message_hash, &router_id, inbound_entry)?;
-
-			Self::deposit_processing_event(
-				domain_address.clone(),
-				submessage,
-				message_hash,
-				router_id.clone(),
-			);
+		Self::deposit_processing_event(
+			domain_address.clone(),
+			message,
+			message_hash,
+			router_id.clone(),
+		);
 
-			Self::execute_if_requirements_are_met(
-				message_hash,
-				&router_ids,
-				session_id,
-				expected_proof_count,
-				domain_address.clone(),
-			)?;
-		}
+		Self::execute_if_requirements_are_met(
+			message_hash,
+			&router_ids,
+			session_id,
+			expected_proof_count,
+			domain_address.clone(),
+		)?;
 
 		Ok(())
 	}