XCM: Properly set the pricing for the DMP router (#6843)

* Properly set the pricing for the DMP router

* Publicize price types

* Use FixedU128 instead of Percent

* Add sp-arithmetic as a dependency for rococo runtime

* Add sp-arithmetic as a dependency to all runtimes

* Remove duplicate import

* Add missing import

* Fix tests

* Create an appropriate QueueDownwardMessageError variant

* Recalculate delivery fee factor based on past queue sizes

* Remove unused error variant

* Fixes

* Fixes

* Remove unused imports

* Rewrite fee factor update mechanism

* Remove unused imports

* Fixes

* Update runtime/parachains/src/dmp.rs

Co-authored-by: Squirrel <gilescope@gmail.com>

* Make DeliveryFeeFactor be a StorageMap keyed on ParaIds

* Fixes

* introduce limit for fee increase on dmp queue

* add message_size based fee factor to increment_fee_factor

* change message_size fee rate to correct value

* fix div by 0 error

* bind limit to variable

* fix message_size_factor and add DeliveryFeeFactor test

* add test for ExponentialPrice implementation

* make test formula based

* make delivery fee factor test formula based

* add max value test for DeliveryFeeFactor and move limit to config

* change threshold back to dynamic value and fix tests

* fmt

* suggested changes and fmt

* small stylistic change

* fmt

* change to tokenlocation

* small fixes

* fmt

* remove sp_arithmetic dependency

* Update runtime/parachains/src/dmp.rs

Co-authored-by: Kian Paimani <5588131+kianenigma@users.noreply.github.com>

---------

Co-authored-by: Squirrel <gilescope@gmail.com>
Co-authored-by: Just van Stam <just.van.stam@gmail.com>
Co-authored-by: Just van Stam <vstam1@users.noreply.github.com>
Co-authored-by: Kian Paimani <5588131+kianenigma@users.noreply.github.com>

polkadot/runtime/parachains/src/dmp.rs

@@ -14,13 +14,45 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
+//! To prevent Out of Memory errors on the `DownwardMessageQueue`, an
+//! exponential fee factor (`DeliveryFeeFactor`) is set. The fee factor
+//! increments exponentially after the number of messages in the
+//! `DownwardMessageQueue` pass a threshold. This threshold is set as:
+//!
+//! ```ignore
+//! // Maximum max sized messages that can be send to
+//! // the DownwardMessageQueue before it runs out of memory
+//! max_messsages = MAX_POSSIBLE_ALLOCATION / max_downward_message_size
+//! threshold = max_messages / THRESHOLD_FACTOR
+//! ```
+//! Based on the THRESHOLD_FACTOR, the threshold is set as a fraction of the
+//! total messages. The `DeliveryFeeFactor` increases for a message over the
+//! threshold by:
+//!
+//! `DeliveryFeeFactor = DeliveryFeeFactor *
+//! (EXPONENTIAL_FEE_BASE + MESSAGE_SIZE_FEE_BASE * encoded_message_size_in_KB)`
+//!
+//! And decreases when the number of messages in the `DownwardMessageQueue` fall
+//! below the threshold by:
+//!
+//! `DeliveryFeeFactor = DeliveryFeeFactor / EXPONENTIAL_FEE_BASE`
+//!
+//! As an extra defensive measure, a `max_messages` hard
+//! limit is set to the number of messages in the DownwardMessageQueue. Messages
+//! that would increase the number of messages in the queue above this hard
+//! limit are dropped.
+
 use crate::{
 	configuration::{self, HostConfiguration},
-	initializer,
+	initializer, FeeTracker,
 };
 use frame_support::pallet_prelude::*;
 use primitives::{DownwardMessage, Hash, Id as ParaId, InboundDownwardMessage};
-use sp_runtime::traits::{BlakeTwo256, Hash as HashT, SaturatedConversion};
+use sp_core::MAX_POSSIBLE_ALLOCATION;
+use sp_runtime::{
+	traits::{BlakeTwo256, Hash as HashT, SaturatedConversion},
+	FixedU128, Saturating,
+};
 use sp_std::{fmt, prelude::*};
 use xcm::latest::SendError;
 
@@ -29,7 +61,9 @@ pub use pallet::*;
 #[cfg(test)]
 mod tests;
 
-pub const MAX_MESSAGE_QUEUE_SIZE: usize = 1024;
+const THRESHOLD_FACTOR: u32 = 2;
+const EXPONENTIAL_FEE_BASE: FixedU128 = FixedU128::from_rational(105, 100); // 1.05
+const MESSAGE_SIZE_FEE_BASE: FixedU128 = FixedU128::from_rational(1, 1000); // 0.001
 
 /// An error sending a downward message.
 #[cfg_attr(test, derive(Debug))]
@@ -102,10 +136,17 @@ pub mod pallet {
 	pub(crate) type DownwardMessageQueueHeads<T: Config> =
 		StorageMap<_, Twox64Concat, ParaId, Hash, ValueQuery>;
 
-	#[pallet::call]
-	impl<T: Config> Pallet<T> {}
-}
+	/// Initialization value for the DeliveryFee factor.
+	#[pallet::type_value]
+	pub fn InitialFactor() -> FixedU128 {
+		FixedU128::from_u32(1)
+	}
 
+	/// The number to multiply the base delivery fee by.
+	#[pallet::storage]
+	pub(crate) type DeliveryFeeFactor<T: Config> =
+		StorageMap<_, Twox64Concat, ParaId, FixedU128, ValueQuery, InitialFactor>;
+}
 /// Routines and getters related to downward message passing.
 impl<T: Config> Pallet<T> {
 	/// Block initialization logic, called by initializer.
@@ -151,7 +192,8 @@ impl<T: Config> Pallet<T> {
 			return Err(QueueDownwardMessageError::ExceedsMaxMessageSize)
 		}
 
-		if DownwardMessageQueues::<T>::decode_len(para).unwrap_or(0) > MAX_MESSAGE_QUEUE_SIZE {
+		// Hard limit on Queue size
+		if Self::dmq_length(*para) > Self::dmq_max_length(config.max_downward_message_size) {
 			return Err(QueueDownwardMessageError::ExceedsMaxMessageSize)
 		}
 
@@ -176,7 +218,8 @@ impl<T: Config> Pallet<T> {
 			return Err(QueueDownwardMessageError::ExceedsMaxMessageSize)
 		}
 
-		if DownwardMessageQueues::<T>::decode_len(para).unwrap_or(0) > MAX_MESSAGE_QUEUE_SIZE {
+		// Hard limit on Queue size
+		if Self::dmq_length(para) > Self::dmq_max_length(config.max_downward_message_size) {
 			return Err(QueueDownwardMessageError::ExceedsMaxMessageSize)
 		}
 
@@ -190,10 +233,20 @@ impl<T: Config> Pallet<T> {
 			*head = new_head;
 		});
 
-		DownwardMessageQueues::<T>::mutate(para, |v| {
+		let q_len = DownwardMessageQueues::<T>::mutate(para, |v| {
 			v.push(inbound);
+			v.len()
 		});
 
+		let threshold =
+			Self::dmq_max_length(config.max_downward_message_size).saturating_div(THRESHOLD_FACTOR);
+		if q_len > (threshold as usize) {
+			let message_size_factor =
+				FixedU128::from_u32(serialized_len.saturating_div(1024) as u32)
+					.saturating_mul(MESSAGE_SIZE_FEE_BASE);
+			Self::increment_fee_factor(para, message_size_factor);
+		}
+
 		Ok(())
 	}
 
@@ -219,7 +272,7 @@ impl<T: Config> Pallet<T> {
 
 	/// Prunes the specified number of messages from the downward message queue of the given para.
 	pub(crate) fn prune_dmq(para: ParaId, processed_downward_messages: u32) -> Weight {
-		DownwardMessageQueues::<T>::mutate(para, |q| {
+		let q_len = DownwardMessageQueues::<T>::mutate(para, |q| {
 			let processed_downward_messages = processed_downward_messages as usize;
 			if processed_downward_messages > q.len() {
 				// reaching this branch is unexpected due to the constraint established by
@@ -228,7 +281,15 @@ impl<T: Config> Pallet<T> {
 			} else {
 				*q = q.split_off(processed_downward_messages);
 			}
+			q.len()
 		});
+
+		let config = configuration::ActiveConfig::<T>::get();
+		let threshold =
+			Self::dmq_max_length(config.max_downward_message_size).saturating_div(THRESHOLD_FACTOR);
+		if q_len <= (threshold as usize) {
+			Self::decrement_fee_factor(para);
+		}
 		T::DbWeight::get().reads_writes(1, 1)
 	}
 
@@ -248,10 +309,42 @@ impl<T: Config> Pallet<T> {
 			.saturated_into::<u32>()
 	}
 
+	fn dmq_max_length(max_downward_message_size: u32) -> u32 {
+		MAX_POSSIBLE_ALLOCATION.checked_div(max_downward_message_size).unwrap_or(0)
+	}
+
 	/// Returns the downward message queue contents for the given para.
 	///
 	/// The most recent messages are the latest in the vector.
 	pub(crate) fn dmq_contents(recipient: ParaId) -> Vec<InboundDownwardMessage<T::BlockNumber>> {
 		DownwardMessageQueues::<T>::get(&recipient)
 	}
+
+	/// Raise the delivery fee factor by a multiplicative factor and stores the resulting value.
+	///
+	/// Returns the new delivery fee factor after the increment.
+	pub(crate) fn increment_fee_factor(para: ParaId, message_size_factor: FixedU128) -> FixedU128 {
+		<DeliveryFeeFactor<T>>::mutate(para, |f| {
+			*f = f.saturating_mul(EXPONENTIAL_FEE_BASE + message_size_factor);
+			*f
+		})
+	}
+
+	/// Reduce the delivery fee factor by a multiplicative factor and stores the resulting value.
+	///
+	/// Does not reduce the fee factor below the initial value, which is currently set as 1.
+	///
+	/// Returns the new delivery fee factor after the decrement.
+	pub(crate) fn decrement_fee_factor(para: ParaId) -> FixedU128 {
+		<DeliveryFeeFactor<T>>::mutate(para, |f| {
+			*f = InitialFactor::get().max(*f / EXPONENTIAL_FEE_BASE);
+			*f
+		})
+	}
+}
+
+impl<T: Config> FeeTracker for Pallet<T> {
+	fn get_fee_factor(para: ParaId) -> FixedU128 {
+		DeliveryFeeFactor::<T>::get(para)
+	}
 }

polkadot/runtime/parachains/src/dmp/tests.rs

@@ -15,7 +15,11 @@
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
 use super::*;
-use crate::mock::{new_test_ext, Configuration, Dmp, MockGenesisConfig, Paras, System, Test};
+use crate::{
+	configuration::ActiveConfig,
+	mock::{new_test_ext, Configuration, Dmp, MockGenesisConfig, Paras, System, Test},
+};
+use frame_support::assert_ok;
 use hex_literal::hex;
 use parity_scale_codec::Encode;
 use primitives::BlockNumber;
@@ -205,3 +209,69 @@ fn verify_dmq_mqc_head_is_externally_accessible() {
 		);
 	});
 }
+
+#[test]
+fn verify_fee_increment_and_decrement() {
+	let a = ParaId::from(123);
+	let mut genesis = default_genesis_config();
+	genesis.configuration.config.max_downward_message_size = 16777216;
+	new_test_ext(genesis).execute_with(|| {
+		let initial = InitialFactor::get();
+		assert_eq!(DeliveryFeeFactor::<Test>::get(a), initial);
+
+		// Under fee limit
+		queue_downward_message(a, vec![1]).unwrap();
+		assert_eq!(DeliveryFeeFactor::<Test>::get(a), initial);
+
+		// Limit reached so fee is increased
+		queue_downward_message(a, vec![1]).unwrap();
+		let result = InitialFactor::get().saturating_mul(EXPONENTIAL_FEE_BASE);
+		assert_eq!(DeliveryFeeFactor::<Test>::get(a), result);
+
+		Dmp::prune_dmq(a, 1);
+		assert_eq!(DeliveryFeeFactor::<Test>::get(a), initial);
+
+		// 10 Kb message adds additional 0.001 per KB fee factor
+		let big_message = [0; 10240].to_vec();
+		let msg_len_in_kb = big_message.len().saturating_div(1024) as u32;
+		let result = initial.saturating_mul(
+			EXPONENTIAL_FEE_BASE +
+				MESSAGE_SIZE_FEE_BASE.saturating_mul(FixedU128::from_u32(msg_len_in_kb)),
+		);
+		queue_downward_message(a, big_message).unwrap();
+		assert_eq!(DeliveryFeeFactor::<Test>::get(a), result);
+
+		queue_downward_message(a, vec![1]).unwrap();
+		let result = result.saturating_mul(EXPONENTIAL_FEE_BASE);
+		assert_eq!(DeliveryFeeFactor::<Test>::get(a), result);
+
+		Dmp::prune_dmq(a, 3);
+		let result = result / EXPONENTIAL_FEE_BASE;
+		assert_eq!(DeliveryFeeFactor::<Test>::get(a), result);
+		assert_eq!(Dmp::dmq_length(a), 0);
+
+		// Messages under limit will keep decreasing fee factor until base fee factor is reached
+		queue_downward_message(a, vec![1]).unwrap();
+		Dmp::prune_dmq(a, 1);
+		queue_downward_message(a, vec![1]).unwrap();
+		Dmp::prune_dmq(a, 1);
+		assert_eq!(DeliveryFeeFactor::<Test>::get(a), initial);
+	});
+}
+
+#[test]
+fn verify_fee_factor_reaches_high_value() {
+	let a = ParaId::from(123);
+	let mut genesis = default_genesis_config();
+	genesis.configuration.config.max_downward_message_size = 51200;
+	new_test_ext(genesis).execute_with(|| {
+		let max_messages =
+			Dmp::dmq_max_length(ActiveConfig::<Test>::get().max_downward_message_size);
+		let mut total_fee_factor = FixedU128::from_float(1.0);
+		for _ in 1..max_messages {
+			assert_ok!(queue_downward_message(a, vec![]));
+			total_fee_factor = total_fee_factor + (DeliveryFeeFactor::<Test>::get(a));
+		}
+		assert!(total_fee_factor > FixedU128::from_u32(100_000_000));
+	});
+}

polkadot/runtime/parachains/src/lib.rs

@@ -51,7 +51,12 @@ mod mock;
 pub use origin::{ensure_parachain, Origin};
 pub use paras::ParaLifecycle;
 use primitives::{HeadData, Id as ParaId, ValidationCode};
-use sp_runtime::DispatchResult;
+use sp_runtime::{DispatchResult, FixedU128};
+
+/// Trait for tracking message delivery fees on a transport protocol.
+pub trait FeeTracker {
+	fn get_fee_factor(para: ParaId) -> FixedU128;
+}
 
 /// Schedule a para to be initialized at the start of the next session with the given genesis data.
 ///