XCM: Properly set the pricing for the DMP router

runtime/common/src/xcm_sender.rs

@@ -21,8 +21,9 @@ use parity_scale_codec::Encode;
 use primitives::Id as ParaId;
 use runtime_parachains::{
 	configuration::{self, HostConfiguration},
-	dmp,
+	dmp, FeeTracker,
 };
+use sp_runtime::FixedPointNumber;
 use sp_std::{marker::PhantomData, prelude::*};
 use xcm::prelude::*;
 use SendError::*;
@@ -47,6 +48,24 @@ impl<T: Get<MultiAssets>> PriceForParachainDelivery for ConstantPrice<T> {
 	}
 }
 
+/// Implementation of `PriceForParachainDelivery` which returns an exponentially increasing price.
+/// The `A` type parameter is used to denote the asset ID that will be used for paying the delivery
+/// fee.
+///
+/// The formula for the fee is based on the sum of a base fee plus a message length fee, multiplied
+/// by a specified factor. In mathematical form, it is `F * (B + encoded_msg_len * M)`.
+pub struct ExponentialPrice<A, B, M, F>(sp_std::marker::PhantomData<(A, B, M, F)>);
+impl<A: Get<AssetId>, B: Get<u128>, M: Get<u128>, F: FeeTracker> PriceForParachainDelivery
+	for ExponentialPrice<A, B, M, F>
+{
+	fn price_for_parachain_delivery(para: ParaId, msg: &Xcm<()>) -> MultiAssets {
+		let msg_fee = (msg.encoded_size() as u128).saturating_mul(M::get());
+		let fee_sum = B::get().saturating_add(msg_fee);
+		let amount = F::get_fee_factor(para).saturating_mul_int(fee_sum);
+		(A::get(), amount).into()
+	}
+}
+
 /// XCM sender for relay chain. It only sends downward message.
 pub struct ChildParachainRouter<T, W, P>(PhantomData<(T, W, P)>);
 
@@ -88,3 +107,61 @@ impl<T: configuration::Config + dmp::Config, W: xcm::WrapVersion, P: PriceForPar
 			.map_err(|_| SendError::Transport(&"Error placing into DMP queue"))
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use frame_support::parameter_types;
+	use runtime_parachains::FeeTracker;
+	use sp_runtime::FixedU128;
+
+	parameter_types! {
+		pub const BaseDeliveryFee: u128 = 300_000_000;
+		pub const TransactionByteFee: u128 = 1_000_000;
+		pub FeeAssetId: AssetId = Concrete(Here.into());
+	}
+
+	struct TestFeeTracker;
+	impl FeeTracker for TestFeeTracker {
+		fn get_fee_factor(_: ParaId) -> FixedU128 {
+			FixedU128::from_rational(101, 100)
+		}
+	}
+
+	type TestExponentialPrice =
+		ExponentialPrice<FeeAssetId, BaseDeliveryFee, TransactionByteFee, TestFeeTracker>;
+
+	#[test]
+	fn exponential_price_correct_price_calculation() {
+		let id: ParaId = 123.into();
+		let b: u128 = BaseDeliveryFee::get();
+		let m: u128 = TransactionByteFee::get();
+
+		// F * (B + msg_length * M)
+		// message_length = 1
+		let result: u128 = TestFeeTracker::get_fee_factor(id.clone()).saturating_mul_int(b + m);
+		assert_eq!(
+			TestExponentialPrice::price_for_parachain_delivery(id.clone(), &Xcm(vec![])),
+			(FeeAssetId::get(), result).into()
+		);
+
+		// message size = 2
+		let result: u128 =
+			TestFeeTracker::get_fee_factor(id.clone()).saturating_mul_int(b + (2 * m));
+		assert_eq!(
+			TestExponentialPrice::price_for_parachain_delivery(id.clone(), &Xcm(vec![ClearOrigin])),
+			(FeeAssetId::get(), result).into()
+		);
+
+		// message size = 4
+		let result: u128 =
+			TestFeeTracker::get_fee_factor(id.clone()).saturating_mul_int(b + (4 * m));
+		assert_eq!(
+			TestExponentialPrice::price_for_parachain_delivery(
+				id.clone(),
+				&Xcm(vec![SetAppendix(Xcm(vec![ClearOrigin]))])
+			),
+			(FeeAssetId::get(), result).into()
+		);
+	}
+}

runtime/kusama/Cargo.toml

@@ -24,7 +24,7 @@ sp-api = { git = "https://github.com/paritytech/substrate", branch = "master", d
 inherents = { package = "sp-inherents", git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
 offchain-primitives = { package = "sp-offchain", git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
 sp-std = { package = "sp-std", git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
-sp-arithmetic = { package = "sp-arithmetic", git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
+sp-arithmetic = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
 sp-io = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
 sp-mmr-primitives = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
 sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
@@ -128,6 +128,7 @@ std = [
 	"parity-scale-codec/std",
 	"scale-info/std",
 	"inherents/std",
+	"sp-arithmetic/std",
 	"sp-core/std",
 	"sp-api/std",
 	"tx-pool-api/std",

runtime/kusama/src/lib.rs

@@ -1430,7 +1430,7 @@ construct_runtime! {
 		ParaScheduler: parachains_scheduler::{Pallet, Storage} = 55,
 		Paras: parachains_paras::{Pallet, Call, Storage, Event, Config, ValidateUnsigned} = 56,
 		Initializer: parachains_initializer::{Pallet, Call, Storage} = 57,
-		Dmp: parachains_dmp::{Pallet, Call, Storage} = 58,
+		Dmp: parachains_dmp::{Pallet, Storage} = 58,
 		Ump: parachains_ump::{Pallet, Call, Storage, Event} = 59,
 		Hrmp: parachains_hrmp::{Pallet, Call, Storage, Event<T>, Config} = 60,
 		ParaSessionInfo: parachains_session_info::{Pallet, Storage} = 61,

runtime/kusama/src/xcm_config.rs

@@ -17,16 +17,22 @@
 //! XCM configurations for the Kusama runtime.
 
 use super::{
-	parachains_origin, AccountId, AllPalletsWithSystem, Balances, Fellows, ParaId, Runtime,
-	RuntimeCall, RuntimeEvent, RuntimeOrigin, StakingAdmin, WeightToFee, XcmPallet,
+	parachains_origin, AccountId, AllPalletsWithSystem, Balances, Dmp, Fellows, ParaId, Runtime,
+	RuntimeCall, RuntimeEvent, RuntimeOrigin, StakingAdmin, TransactionByteFee, WeightToFee,
+	XcmPallet,
 };
 use frame_support::{
 	match_types, parameter_types,
 	traits::{Contains, Everything, Nothing},
 	weights::Weight,
 };
 use frame_system::EnsureRoot;
-use runtime_common::{crowdloan, paras_registrar, xcm_sender, ToAuthor};
+use kusama_runtime_constants::currency::CENTS;
+use runtime_common::{
+	crowdloan, paras_registrar,
+	xcm_sender::{ChildParachainRouter, ExponentialPrice},
+	ToAuthor,
+};
 use sp_core::ConstU32;
 use xcm::latest::prelude::*;
 use xcm_builder::{
@@ -101,13 +107,21 @@ parameter_types! {
 	/// Maximum number of instructions in a single XCM fragment. A sanity check against weight
 	/// calculations getting too crazy.
 	pub const MaxInstructions: u32 = 100;
+	/// The asset ID for the asset that we use to pay for message delivery fees.
+	pub FeeAssetId: AssetId = Concrete(Here.into());
+	/// The base fee for the message delivery fees.
+	pub const BaseDeliveryFee: u128 = CENTS.saturating_mul(3);
 }
 
 /// The XCM router. When we want to send an XCM message, we use this type. It amalgamates all of our
 /// individual routers.
 pub type XcmRouter = (
 	// Only one router so far - use DMP to communicate with child parachains.
-	xcm_sender::ChildParachainRouter<Runtime, XcmPallet, ()>,
+	ChildParachainRouter<
+		Runtime,
+		XcmPallet,
+		ExponentialPrice<FeeAssetId, BaseDeliveryFee, TransactionByteFee, Dmp>,
+	>,
 );
 
 parameter_types! {

runtime/parachains/src/dmp.rs

@@ -14,13 +14,34 @@
 // 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 +50,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 +125,18 @@ 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]
+	#[pallet::getter(fn delivery_fee_factor)]
+	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 +182,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 +208,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 +223,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 +262,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 +271,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 +299,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 {
+		Pallet::<T>::delivery_fee_factor(para)
+	}
 }