Introduce XcmFeesToAccount fee manager (#1234)

Combination of paritytech/polkadot#7005, its addon PR
paritytech/polkadot#7585 and its companion paritytech/cumulus#2433.

This PR introduces a new XcmFeesToAccount struct which implements the
`FeeManager` trait, and assigns this struct as the `FeeManager` in the
XCM config for all runtimes.

The struct simply deposits all fees handled by the XCM executor to a
specified account. In all runtimes, the specified account is configured
as the treasury account.

XCM __delivery__ fees are now being introduced (unless the root origin
is sending a message to a system parachain on behalf of the originating
chain).

# Note for reviewers

Most file changes are tests that had to be modified to account for the
new fees.
Main changes are in:
- cumulus/pallets/xcmp-queue/src/lib.rs <- To make it track the delivery
fees exponential factor
- polkadot/xcm/xcm-builder/src/fee_handling.rs <- Added. Has the
FeeManager implementation
- All runtime xcm_config files <- To add the FeeManager to the XCM
configuration

# Important note

After this change, instructions that create and send a new XCM (Query*,
Report*, ExportMessage, InitiateReserveWithdraw, InitiateTeleport,
DepositReserveAsset, TransferReserveAsset, LockAsset and RequestUnlock)
will require the corresponding origin account in the origin register to
pay for transport delivery fees, and the onward message will fail to be
sent if the origin account does not have the required amount. This
delivery fee is on top of what we already collect as tx fees in
pallet-xcm and XCM BuyExecution fees!

Wallet UIs that want to expose the new delivery fee can do so using the
formula:

```
delivery_fee_factor * (base_fee + encoded_msg_len * per_byte_fee)
```

where the delivery fee factor can be obtained from the corresponding
pallet based on which transport you are using (UMP, HRMP or bridges),
the base fee is a constant, the encoded message length from the message
itself and the per byte fee is the same as the configured per byte fee
for txs (i.e. `TransactionByteFee`).

---------

Co-authored-by: Branislav Kontur <bkontur@gmail.com>
Co-authored-by: joe petrowski <25483142+joepetrowski@users.noreply.github.com>
Co-authored-by: Giles Cope <gilescope@gmail.com>
Co-authored-by: command-bot <>
Co-authored-by: Francisco Aguirre <franciscoaguirreperez@gmail.com>
Co-authored-by: Liam Aharon <liam.aharon@hotmail.com>
Co-authored-by: Kian Paimani <5588131+kianenigma@users.noreply.github.com>

cumulus/pallets/parachain-system/Cargo.toml

@@ -29,6 +29,7 @@ sp-version = { path = "../../../substrate/primitives/version", default-features
 
 # Polkadot
 polkadot-parachain-primitives = { path = "../../../polkadot/parachain", default-features = false, features = [ "wasm-api" ]}
+polkadot-runtime-parachains = { path = "../../../polkadot/runtime/parachains", default-features = false }
 xcm = { package = "staging-xcm", path = "../../../polkadot/xcm", default-features = false}
 
 # Cumulus
@@ -63,6 +64,7 @@ std = [
 	"frame-system/std",
 	"log/std",
 	"polkadot-parachain-primitives/std",
+	"polkadot-runtime-parachains/std",
 	"scale-info/std",
 	"sp-core/std",
 	"sp-externalities/std",
@@ -80,12 +82,14 @@ runtime-benchmarks = [
 	"frame-support/runtime-benchmarks",
 	"frame-system/runtime-benchmarks",
 	"polkadot-parachain-primitives/runtime-benchmarks",
+	"polkadot-runtime-parachains/runtime-benchmarks",
 	"sp-runtime/runtime-benchmarks",
 ]
 
 try-runtime = [
 	"frame-support/try-runtime",
 	"frame-system/try-runtime",
+	"polkadot-runtime-parachains/try-runtime",
 	"sp-runtime/try-runtime",
 ]
 

cumulus/pallets/parachain-system/src/lib.rs

@@ -29,10 +29,10 @@
 
 use codec::{Decode, Encode, MaxEncodedLen};
 use cumulus_primitives_core::{
-	relay_chain, AbridgedHostConfiguration, ChannelStatus, CollationInfo, DmpMessageHandler,
-	GetChannelInfo, InboundDownwardMessage, InboundHrmpMessage, MessageSendError,
-	OutboundHrmpMessage, ParaId, PersistedValidationData, UpwardMessage, UpwardMessageSender,
-	XcmpMessageHandler, XcmpMessageSource,
+	relay_chain, AbridgedHostConfiguration, ChannelInfo, ChannelStatus, CollationInfo,
+	DmpMessageHandler, GetChannelInfo, InboundDownwardMessage, InboundHrmpMessage,
+	MessageSendError, OutboundHrmpMessage, ParaId, PersistedValidationData, UpwardMessage,
+	UpwardMessageSender, XcmpMessageHandler, XcmpMessageSource,
 };
 use cumulus_primitives_parachain_inherent::{MessageQueueChain, ParachainInherentData};
 use frame_support::{
@@ -45,6 +45,7 @@ use frame_support::{
 };
 use frame_system::{ensure_none, ensure_root, pallet_prelude::HeaderFor};
 use polkadot_parachain_primitives::primitives::RelayChainBlockNumber;
+use polkadot_runtime_parachains::FeeTracker;
 use scale_info::TypeInfo;
 use sp_runtime::{
 	traits::{Block as BlockT, BlockNumberProvider, Hash},
@@ -52,7 +53,7 @@ use sp_runtime::{
 		InvalidTransaction, TransactionLongevity, TransactionSource, TransactionValidity,
 		ValidTransaction,
 	},
-	DispatchError, RuntimeDebug,
+	DispatchError, FixedU128, RuntimeDebug, Saturating,
 };
 use sp_std::{cmp, collections::btree_map::BTreeMap, prelude::*};
 use xcm::latest::XcmHash;
@@ -177,6 +178,20 @@ where
 	check_version: bool,
 }
 
+pub mod ump_constants {
+	use super::FixedU128;
+
+	/// `host_config.max_upward_queue_size / THRESHOLD_FACTOR` is the threshold after which delivery
+	/// starts getting exponentially more expensive.
+	/// `2` means the price starts to increase when queue is half full.
+	pub const THRESHOLD_FACTOR: u32 = 2;
+	/// The base number the delivery fee factor gets multiplied by every time it is increased.
+	/// Also the number it gets divided by when decreased.
+	pub const EXPONENTIAL_FEE_BASE: FixedU128 = FixedU128::from_rational(105, 100); // 1.05
+	/// The base number message size in KB is multiplied by before increasing the fee factor.
+	pub const MESSAGE_SIZE_FEE_BASE: FixedU128 = FixedU128::from_rational(1, 1000); // 0.001
+}
+
 #[frame_support::pallet]
 pub mod pallet {
 	use super::*;
@@ -240,6 +255,10 @@ pub mod pallet {
 
 	#[pallet::hooks]
 	impl<T: Config> Hooks<BlockNumberFor<T>> for Pallet<T> {
+		/// Handles actually sending upward messages by moving them from `PendingUpwardMessages` to
+		/// `UpwardMessages`. Decreases the delivery fee factor if after sending messages, the queue
+		/// total size is less than the threshold (see [`ump_constants::THRESHOLD_FACTOR`]).
+		/// Also does the sending for HRMP messages it takes from `OutboundXcmpMessageSource`.
 		fn on_finalize(_: BlockNumberFor<T>) {
 			<DidSetValidationCode<T>>::kill();
 			<UpgradeRestrictionSignal<T>>::kill();
@@ -326,6 +345,17 @@ pub mod pallet {
 				UpwardMessages::<T>::put(&up[..num as usize]);
 				*up = up.split_off(num as usize);
 
+				// If the total size of the pending messages is less than the threshold,
+				// we decrease the fee factor, since the queue is less congested.
+				// This makes delivery of new messages cheaper.
+				let threshold = host_config
+					.max_upward_queue_size
+					.saturating_div(ump_constants::THRESHOLD_FACTOR);
+				let remaining_total_size: usize = up.iter().map(UpwardMessage::len).sum();
+				if remaining_total_size <= threshold as usize {
+					Self::decrease_fee_factor(());
+				}
+
 				(num, total_size)
 			});
 
@@ -721,7 +751,7 @@ pub mod pallet {
 		StorageValue<_, Vec<Ancestor<T::Hash>>, ValueQuery>;
 
 	/// Storage field that keeps track of bandwidth used by the unincluded segment along with the
-	/// latest the latest HRMP watermark. Used for limiting the acceptance of new blocks with
+	/// latest HRMP watermark. Used for limiting the acceptance of new blocks with
 	/// respect to relay chain constraints.
 	#[pallet::storage]
 	pub(super) type AggregatedUnincludedSegment<T: Config> =
@@ -860,6 +890,17 @@ pub mod pallet {
 	pub(super) type PendingUpwardMessages<T: Config> =
 		StorageValue<_, Vec<UpwardMessage>, ValueQuery>;
 
+	/// Initialization value for the delivery fee factor for UMP.
+	#[pallet::type_value]
+	pub fn UpwardInitialDeliveryFeeFactor() -> FixedU128 {
+		FixedU128::from_u32(1)
+	}
+
+	/// The factor to multiply the base delivery fee by for UMP.
+	#[pallet::storage]
+	pub(super) type UpwardDeliveryFeeFactor<T: Config> =
+		StorageValue<_, FixedU128, ValueQuery, UpwardInitialDeliveryFeeFactor>;
+
 	/// The number of HRMP messages we observed in `on_initialize` and thus used that number for
 	/// announcing the weight of `on_initialize` and `on_finalize`.
 	#[pallet::storage]
@@ -976,6 +1017,31 @@ impl<T: Config> Pallet<T> {
 	}
 }
 
+impl<T: Config> FeeTracker for Pallet<T> {
+	type Id = ();
+
+	fn get_fee_factor(_: Self::Id) -> FixedU128 {
+		UpwardDeliveryFeeFactor::<T>::get()
+	}
+
+	fn increase_fee_factor(_: Self::Id, message_size_factor: FixedU128) -> FixedU128 {
+		<UpwardDeliveryFeeFactor<T>>::mutate(|f| {
+			*f = f.saturating_mul(
+				ump_constants::EXPONENTIAL_FEE_BASE.saturating_add(message_size_factor),
+			);
+			*f
+		})
+	}
+
+	fn decrease_fee_factor(_: Self::Id) -> FixedU128 {
+		<UpwardDeliveryFeeFactor<T>>::mutate(|f| {
+			*f =
+				UpwardInitialDeliveryFeeFactor::get().max(*f / ump_constants::EXPONENTIAL_FEE_BASE);
+			*f
+		})
+	}
+}
+
 impl<T: Config> GetChannelInfo for Pallet<T> {
 	fn get_channel_status(id: ParaId) -> ChannelStatus {
 		// Note, that we are using `relevant_messaging_state` which may be from the previous
@@ -1019,10 +1085,17 @@ impl<T: Config> GetChannelInfo for Pallet<T> {
 		ChannelStatus::Ready(max_size_now as usize, max_size_ever as usize)
 	}
 
-	fn get_channel_max(id: ParaId) -> Option<usize> {
+	fn get_channel_info(id: ParaId) -> Option<ChannelInfo> {
 		let channels = Self::relevant_messaging_state()?.egress_channels;
 		let index = channels.binary_search_by_key(&id, |item| item.0).ok()?;
-		Some(channels[index].1.max_message_size as usize)
+		let info = ChannelInfo {
+			max_capacity: channels[index].1.max_capacity,
+			max_total_size: channels[index].1.max_total_size,
+			max_message_size: channels[index].1.max_message_size,
+			msg_count: channels[index].1.msg_count,
+			total_size: channels[index].1.total_size,
+		};
+		Some(info)
 	}
 }
 
@@ -1427,6 +1500,23 @@ impl<T: Config> Pallet<T> {
 		})
 	}
 
+	/// Open HRMP channel for using it in benchmarks or tests.
+	///
+	/// The caller assumes that the pallet will accept regular outbound message to the sibling
+	/// `target_parachain` after this call. No other assumptions are made.
+	#[cfg(any(feature = "runtime-benchmarks", feature = "std"))]
+	pub fn open_custom_outbound_hrmp_channel_for_benchmarks_or_tests(
+		target_parachain: ParaId,
+		channel: cumulus_primitives_core::AbridgedHrmpChannel,
+	) {
+		RelevantMessagingState::<T>::put(MessagingStateSnapshot {
+			dmq_mqc_head: Default::default(),
+			relay_dispatch_queue_remaining_capacity: Default::default(),
+			ingress_channels: Default::default(),
+			egress_channels: vec![(target_parachain, channel)],
+		})
+	}
+
 	/// Prepare/insert relevant data for `schedule_code_upgrade` for benchmarks.
 	#[cfg(feature = "runtime-benchmarks")]
 	pub fn initialize_for_set_code_benchmark(max_code_size: u32) {
@@ -1468,7 +1558,13 @@ impl<T: Config> frame_system::SetCode<T> for ParachainSetCode<T> {
 }
 
 impl<T: Config> Pallet<T> {
+	/// Puts a message in the `PendingUpwardMessages` storage item.
+	/// The message will be later sent in `on_finalize`.
+	/// Checks host configuration to see if message is too big.
+	/// Increases the delivery fee factor if the queue is sufficiently (see
+	/// [`ump_constants::THRESHOLD_FACTOR`]) congested.
 	pub fn send_upward_message(message: UpwardMessage) -> Result<(u32, XcmHash), MessageSendError> {
+		let message_len = message.len();
 		// Check if the message fits into the relay-chain constraints.
 		//
 		// Note, that we are using `host_configuration` here which may be from the previous
@@ -1482,9 +1578,22 @@ impl<T: Config> Pallet<T> {
 		//
 		// However, changing this setting is expected to be rare.
 		if let Some(cfg) = Self::host_configuration() {
-			if message.len() > cfg.max_upward_message_size as usize {
+			if message_len > cfg.max_upward_message_size as usize {
 				return Err(MessageSendError::TooBig)
 			}
+			let threshold =
+				cfg.max_upward_queue_size.saturating_div(ump_constants::THRESHOLD_FACTOR);
+			// We check the threshold against total size and not number of messages since messages
+			// could be big or small.
+			<PendingUpwardMessages<T>>::append(message.clone());
+			let pending_messages = PendingUpwardMessages::<T>::get();
+			let total_size: usize = pending_messages.iter().map(UpwardMessage::len).sum();
+			if total_size > threshold as usize {
+				// We increase the fee factor by a factor based on the new message's size in KB
+				let message_size_factor = FixedU128::from((message_len / 1024) as u128)
+					.saturating_mul(ump_constants::MESSAGE_SIZE_FEE_BASE);
+				Self::increase_fee_factor((), message_size_factor);
+			}
 		} else {
 			// This storage field should carry over from the previous block. So if it's None
 			// then it must be that this is an edge-case where a message is attempted to be
@@ -1495,8 +1604,8 @@ impl<T: Config> Pallet<T> {
 			// returned back to the sender.
 			//
 			// Thus fall through here.
+			<PendingUpwardMessages<T>>::append(message.clone());
 		};
-		<PendingUpwardMessages<T>>::append(message.clone());
 
 		// The relay ump does not use using_encoded
 		// We apply the same this to use the same hash

cumulus/pallets/parachain-system/src/tests.rs

@@ -1496,3 +1496,53 @@ fn deposits_relay_parent_storage_root() {
 		},
 	);
 }
+
+#[test]
+fn ump_fee_factor_increases_and_decreases() {
+	BlockTests::new()
+		.with_relay_sproof_builder(|_, _, sproof| {
+			sproof.host_config.max_upward_queue_size = 100;
+			sproof.host_config.max_upward_message_num_per_candidate = 1;
+		})
+		.add_with_post_test(
+			1,
+			|| {
+				// Fee factor increases in `send_upward_message`
+				ParachainSystem::send_upward_message(b"Test".to_vec()).unwrap();
+				assert_eq!(UpwardDeliveryFeeFactor::<Test>::get(), FixedU128::from_u32(1));
+
+				ParachainSystem::send_upward_message(
+					b"This message will be enough to increase the fee factor".to_vec(),
+				)
+				.unwrap();
+				assert_eq!(
+					UpwardDeliveryFeeFactor::<Test>::get(),
+					FixedU128::from_rational(105, 100)
+				);
+			},
+			|| {
+				// Factor decreases in `on_finalize`, but only if we are below the threshold
+				let messages = UpwardMessages::<Test>::get();
+				assert_eq!(messages, vec![b"Test".to_vec()]);
+				assert_eq!(
+					UpwardDeliveryFeeFactor::<Test>::get(),
+					FixedU128::from_rational(105, 100)
+				);
+			},
+		)
+		.add_with_post_test(
+			2,
+			|| {
+				// We do nothing here
+			},
+			|| {
+				let messages = UpwardMessages::<Test>::get();
+				assert_eq!(
+					messages,
+					vec![b"This message will be enough to increase the fee factor".to_vec(),]
+				);
+				// Now the delivery fee factor is decreased, since we are below the threshold
+				assert_eq!(UpwardDeliveryFeeFactor::<Test>::get(), FixedU128::from_u32(1));
+			},
+		);
+}

cumulus/pallets/xcmp-queue/Cargo.toml

@@ -14,13 +14,15 @@ scale-info = { version = "2.9.0", default-features = false, features = ["derive"
 frame-support = { path = "../../../substrate/frame/support", default-features = false}
 frame-system = { path = "../../../substrate/frame/system", default-features = false}
 sp-io = { path = "../../../substrate/primitives/io", default-features = false}
+sp-core = { path = "../../../substrate/primitives/core", default-features = false }
 sp-runtime = { path = "../../../substrate/primitives/runtime", default-features = false}
 sp-std = { path = "../../../substrate/primitives/std", default-features = false}
 
 # Polkadot
-polkadot-runtime-common = { path = "../../../polkadot/runtime/common", default-features = false}
-xcm = { package = "staging-xcm", path = "../../../polkadot/xcm", default-features = false}
-xcm-executor = { package = "staging-xcm-executor", path = "../../../polkadot/xcm/xcm-executor", default-features = false}
+polkadot-runtime-common = { path = "../../../polkadot/runtime/common", default-features = false }
+polkadot-runtime-parachains = { path = "../../../polkadot/runtime/parachains", default-features = false }
+xcm = { package = "staging-xcm", path = "../../../polkadot/xcm", default-features = false }
+xcm-executor = { package = "staging-xcm-executor", path = "../../../polkadot/xcm/xcm-executor", default-features = false }
 
 # Cumulus
 cumulus-primitives-core = { path = "../../primitives/core", default-features = false }
@@ -54,7 +56,9 @@ std = [
 	"frame-system/std",
 	"log/std",
 	"polkadot-runtime-common/std",
+	"polkadot-runtime-parachains/std",
 	"scale-info/std",
+	"sp-core/std",
 	"sp-io/std",
 	"sp-runtime/std",
 	"sp-std/std",
@@ -69,6 +73,7 @@ runtime-benchmarks = [
 	"frame-system/runtime-benchmarks",
 	"pallet-balances/runtime-benchmarks",
 	"polkadot-runtime-common/runtime-benchmarks",
+	"polkadot-runtime-parachains/runtime-benchmarks",
 	"sp-runtime/runtime-benchmarks",
 	"xcm-builder/runtime-benchmarks",
 	"xcm-executor/runtime-benchmarks",
@@ -79,6 +84,7 @@ try-runtime = [
 	"frame-system/try-runtime",
 	"pallet-balances/try-runtime",
 	"polkadot-runtime-common/try-runtime",
+	"polkadot-runtime-parachains/try-runtime",
 	"sp-runtime/try-runtime",
 ]
 bridging = [ "bp-xcm-bridge-hub-router" ]

cumulus/pallets/xcmp-queue/src/lib.rs

@@ -22,6 +22,16 @@
 //! Also provides an implementation of `SendXcm` which can be placed in a router tuple for relaying
 //! XCM over XCMP if the destination is `Parent/Parachain`. It requires an implementation of
 //! `XcmExecutor` for dispatching incoming XCM messages.
+//!
+//! To prevent out of memory errors on the `OutboundXcmpMessages` queue, an exponential fee factor
+//! (`DeliveryFeeFactor`) is set, much like the one used in DMP.
+//! The fee factor increases whenever the total size of messages in a particular channel passes a
+//! threshold. This threshold is defined as a percentage of the maximum total size the channel can
+//! have. More concretely, the threshold is `max_total_size` / `THRESHOLD_FACTOR`, where:
+//! - `max_total_size` is the maximum size, in bytes, of the channel, not number of messages.
+//! It is defined in the channel configuration.
+//! - `THRESHOLD_FACTOR` just declares which percentage of the max size is the actual threshold.
+//! If it's 2, then the threshold is half of the max size, if it's 4, it's a quarter, and so on.
 
 #![cfg_attr(not(feature = "std"), no_std)]
 
@@ -49,13 +59,15 @@ use frame_support::{
 	traits::{EnsureOrigin, Get},
 	weights::{constants::WEIGHT_REF_TIME_PER_MILLIS, Weight},
 };
-use polkadot_runtime_common::xcm_sender::PriceForParachainDelivery;
+use polkadot_runtime_common::xcm_sender::PriceForMessageDelivery;
+use polkadot_runtime_parachains::FeeTracker;
 use rand_chacha::{
 	rand_core::{RngCore, SeedableRng},
 	ChaChaRng,
 };
 use scale_info::TypeInfo;
-use sp_runtime::RuntimeDebug;
+use sp_core::MAX_POSSIBLE_ALLOCATION;
+use sp_runtime::{FixedU128, RuntimeDebug, Saturating};
 use sp_std::{convert::TryFrom, prelude::*};
 use xcm::{latest::prelude::*, VersionedXcm, WrapVersion, MAX_XCM_DECODE_DEPTH};
 use xcm_executor::traits::ConvertOrigin;
@@ -68,6 +80,19 @@ pub type OverweightIndex = u64;
 const LOG_TARGET: &str = "xcmp_queue";
 const DEFAULT_POV_SIZE: u64 = 64 * 1024; // 64 KB
 
+/// Constants related to delivery fee calculation
+pub mod delivery_fee_constants {
+	use super::FixedU128;
+
+	/// Fees will start increasing when queue is half full
+	pub const THRESHOLD_FACTOR: u32 = 2;
+	/// The base number the delivery fee factor gets multiplied by every time it is increased.
+	/// Also, the number it gets divided by when decreased.
+	pub const EXPONENTIAL_FEE_BASE: FixedU128 = FixedU128::from_rational(105, 100); // 1.05
+	/// The contribution of each KB to a fee factor increase
+	pub const MESSAGE_SIZE_FEE_BASE: FixedU128 = FixedU128::from_rational(1, 1000); // 0.001
+}
+
 // Maximum amount of messages to process per block. This is a temporary measure until we properly
 // account for proof size weights.
 const MAX_MESSAGES_PER_BLOCK: u8 = 10;
@@ -77,7 +102,7 @@ const MAX_OVERWEIGHT_MESSAGES: u32 = 1000;
 #[frame_support::pallet]
 pub mod pallet {
 	use super::*;
-	use frame_support::pallet_prelude::*;
+	use frame_support::{pallet_prelude::*, Twox64Concat};
 	use frame_system::pallet_prelude::*;
 
 	#[pallet::pallet]
@@ -109,7 +134,7 @@ pub mod pallet {
 		type ControllerOriginConverter: ConvertOrigin<Self::RuntimeOrigin>;
 
 		/// The price for delivering an XCM to a sibling parachain destination.
-		type PriceForSiblingDelivery: PriceForParachainDelivery;
+		type PriceForSiblingDelivery: PriceForMessageDelivery<Id = ParaId>;
 
 		/// The weight information of this pallet.
 		type WeightInfo: WeightInfo;
@@ -374,6 +399,17 @@ pub mod pallet {
 	/// Whether or not the XCMP queue is suspended from executing incoming XCMs or not.
 	#[pallet::storage]
 	pub(super) type QueueSuspended<T: Config> = StorageValue<_, bool, ValueQuery>;
+
+	/// Initialization value for the DeliveryFee factor.
+	#[pallet::type_value]
+	pub fn InitialFactor() -> FixedU128 {
+		FixedU128::from_u32(1)
+	}
+
+	/// The factor to multiply the base delivery fee by.
+	#[pallet::storage]
+	pub(super) type DeliveryFeeFactor<T: Config> =
+		StorageMap<_, Twox64Concat, ParaId, FixedU128, ValueQuery, InitialFactor>;
 }
 
 #[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Encode, Decode, RuntimeDebug, TypeInfo)]
@@ -403,7 +439,7 @@ pub struct InboundChannelDetails {
 }
 
 /// Struct containing detailed information about the outbound channel.
-#[derive(Clone, Eq, PartialEq, Encode, Decode, TypeInfo)]
+#[derive(Debug, Clone, Eq, PartialEq, Encode, Decode, TypeInfo)]
 pub struct OutboundChannelDetails {
 	/// The `ParaId` of the parachain that this channel is connected with.
 	recipient: ParaId,
@@ -503,56 +539,90 @@ impl<T: Config> Pallet<T> {
 	/// length prefixed and can thus decode each fragment from the aggregate stream. With this,
 	/// we can concatenate them into a single aggregate blob without needing to be concerned
 	/// about encoding fragment boundaries.
+	///
+	/// If successful, returns the number of pages in the outbound queue after enqueuing the new
+	/// fragment.
 	fn send_fragment<Fragment: Encode>(
 		recipient: ParaId,
 		format: XcmpMessageFormat,
 		fragment: Fragment,
 	) -> Result<u32, MessageSendError> {
-		let data = fragment.encode();
+		let encoded_fragment = fragment.encode();
 
-		// Optimization note: `max_message_size` could potentially be stored in
-		// `OutboundXcmpMessages` once known; that way it's only accessed when a new page is needed.
-
-		let max_message_size =
-			T::ChannelInfo::get_channel_max(recipient).ok_or(MessageSendError::NoChannel)?;
-		if data.len() > max_message_size {
+		let channel_info =
+			T::ChannelInfo::get_channel_info(recipient).ok_or(MessageSendError::NoChannel)?;
+		let max_message_size = channel_info.max_message_size as usize;
+		// Max message size refers to aggregates, or pages. Not to individual fragments.
+		if encoded_fragment.len() > max_message_size {
 			return Err(MessageSendError::TooBig)
 		}
 
-		let mut s = <OutboundXcmpStatus<T>>::get();
-		let details = if let Some(details) = s.iter_mut().find(|item| item.recipient == recipient) {
+		let mut all_channels = <OutboundXcmpStatus<T>>::get();
+		let channel_details = if let Some(details) =
+			all_channels.iter_mut().find(|channel| channel.recipient == recipient)
+		{
 			details
 		} else {
-			s.push(OutboundChannelDetails::new(recipient));
-			s.last_mut().expect("can't be empty; a new element was just pushed; qed")
+			all_channels.push(OutboundChannelDetails::new(recipient));
+			all_channels
+				.last_mut()
+				.expect("can't be empty; a new element was just pushed; qed")
 		};
-		let have_active = details.last_index > details.first_index;
-		let appended = have_active &&
-			<OutboundXcmpMessages<T>>::mutate(recipient, details.last_index - 1, |s| {
-				if XcmpMessageFormat::decode_with_depth_limit(MAX_XCM_DECODE_DEPTH, &mut &s[..]) !=
-					Ok(format)
-				{
-					return false
-				}
-				if s.len() + data.len() > max_message_size {
-					return false
-				}
-				s.extend_from_slice(&data[..]);
-				true
-			});
-		if appended {
-			Ok((details.last_index - details.first_index - 1) as u32)
+		let have_active = channel_details.last_index > channel_details.first_index;
+		// Try to append fragment to the last page, if there is enough space.
+		// We return the size of the last page inside of the option, to not calculate it again.
+		let appended_to_last_page = have_active
+			.then(|| {
+				<OutboundXcmpMessages<T>>::mutate(
+					recipient,
+					channel_details.last_index - 1,
+					|page| {
+						if XcmpMessageFormat::decode_with_depth_limit(
+							MAX_XCM_DECODE_DEPTH,
+							&mut &page[..],
+						) != Ok(format)
+						{
+							return None
+						}
+						if page.len() + encoded_fragment.len() > max_message_size {
+							return None
+						}
+						page.extend_from_slice(&encoded_fragment[..]);
+						Some(page.len())
+					},
+				)
+			})
+			.flatten();
+
+		let (number_of_pages, last_page_size) = if let Some(size) = appended_to_last_page {
+			let number_of_pages = (channel_details.last_index - channel_details.first_index) as u32;
+			(number_of_pages, size)
 		} else {
 			// Need to add a new page.
-			let page_index = details.last_index;
-			details.last_index += 1;
+			let page_index = channel_details.last_index;
+			channel_details.last_index += 1;
 			let mut new_page = format.encode();
-			new_page.extend_from_slice(&data[..]);
+			new_page.extend_from_slice(&encoded_fragment[..]);
+			let last_page_size = new_page.len();
+			let number_of_pages = (channel_details.last_index - channel_details.first_index) as u32;
 			<OutboundXcmpMessages<T>>::insert(recipient, page_index, new_page);
-			let r = (details.last_index - details.first_index - 1) as u32;
-			<OutboundXcmpStatus<T>>::put(s);
-			Ok(r)
+			<OutboundXcmpStatus<T>>::put(all_channels);
+			(number_of_pages, last_page_size)
+		};
+
+		// We have to count the total size here since `channel_info.total_size` is not updated at
+		// this point in time. We assume all previous pages are filled, which, in practice, is not
+		// always the case.
+		let total_size =
+			number_of_pages.saturating_sub(1) * max_message_size as u32 + last_page_size as u32;
+		let threshold = channel_info.max_total_size / delivery_fee_constants::THRESHOLD_FACTOR;
+		if total_size > threshold {
+			let message_size_factor = FixedU128::from((encoded_fragment.len() / 1024) as u128)
+				.saturating_mul(delivery_fee_constants::MESSAGE_SIZE_FEE_BASE);
+			Self::increase_fee_factor(recipient, message_size_factor);
 		}
+
+		Ok(number_of_pages)
 	}
 
 	/// Sends a signal to the `dest` chain over XCMP. This is guaranteed to be dispatched on this
@@ -1004,9 +1074,8 @@ impl<T: Config> XcmpMessageHandler for Pallet<T> {
 				// Record the fact we received it.
 				match status.binary_search_by_key(&sender, |item| item.sender) {
 					Ok(i) => {
-						let count = status[i].message_metadata.len();
-						if count as u32 >= suspend_threshold && status[i].state == InboundState::Ok
-						{
+						let count = status[i].message_metadata.len() as u32;
+						if count >= suspend_threshold && status[i].state == InboundState::Ok {
 							status[i].state = InboundState::Suspended;
 							let r = Self::send_signal(sender, ChannelSignal::Suspend);
 							if r.is_err() {
@@ -1015,7 +1084,7 @@ impl<T: Config> XcmpMessageHandler for Pallet<T> {
 								);
 							}
 						}
-						if (count as u32) < drop_threshold {
+						if count < drop_threshold {
 							status[i].message_metadata.push((sent_at, format));
 						} else {
 							debug_assert!(
@@ -1023,6 +1092,13 @@ impl<T: Config> XcmpMessageHandler for Pallet<T> {
 								"XCMP channel queue full. Silently dropping message"
 							);
 						}
+						// Update the delivery fee factor, if applicable.
+						if count > suspend_threshold {
+							let message_size_factor =
+								FixedU128::from((data_ref.len() / 1024) as u128)
+									.saturating_mul(delivery_fee_constants::MESSAGE_SIZE_FEE_BASE);
+							Self::increase_fee_factor(sender, message_size_factor);
+						}
 					},
 					Err(_) => status.push(InboundChannelDetails {
 						sender,
@@ -1120,6 +1196,21 @@ impl<T: Config> XcmpMessageSource for Pallet<T> {
 				result.push((para_id, page));
 			}
 
+			let max_total_size = match T::ChannelInfo::get_channel_info(para_id) {
+				Some(channel_info) => channel_info.max_total_size,
+				None => {
+					log::warn!("calling `get_channel_info` with no RelevantMessagingState?!");
+					MAX_POSSIBLE_ALLOCATION // We use this as a fallback in case the messaging state is not present
+				},
+			};
+			let threshold = max_total_size.saturating_div(delivery_fee_constants::THRESHOLD_FACTOR);
+			let remaining_total_size: usize = (first_index..last_index)
+				.map(|index| OutboundXcmpMessages::<T>::decode_len(para_id, index).unwrap())
+				.sum();
+			if remaining_total_size <= threshold as usize {
+				Self::decrease_fee_factor(para_id);
+			}
+
 			*status = OutboundChannelDetails {
 				recipient: para_id,
 				state: outbound_state,
@@ -1172,7 +1263,7 @@ impl<T: Config> SendXcm for Pallet<T> {
 			MultiLocation { parents: 1, interior: X1(Parachain(id)) } => {
 				let xcm = msg.take().ok_or(SendError::MissingArgument)?;
 				let id = ParaId::from(*id);
-				let price = T::PriceForSiblingDelivery::price_for_parachain_delivery(id, &xcm);
+				let price = T::PriceForSiblingDelivery::price_for_delivery(id, &xcm);
 				let versioned_xcm = T::VersionWrapper::wrap_version(&d, xcm)
 					.map_err(|()| SendError::DestinationUnsupported)?;
 				Ok(((id, versioned_xcm), price))
@@ -1198,3 +1289,27 @@ impl<T: Config> SendXcm for Pallet<T> {
 		}
 	}
 }
+
+impl<T: Config> FeeTracker for Pallet<T> {
+	type Id = ParaId;
+
+	fn get_fee_factor(id: Self::Id) -> FixedU128 {
+		<DeliveryFeeFactor<T>>::get(id)
+	}
+
+	fn increase_fee_factor(id: Self::Id, message_size_factor: FixedU128) -> FixedU128 {
+		<DeliveryFeeFactor<T>>::mutate(id, |f| {
+			*f = f.saturating_mul(
+				delivery_fee_constants::EXPONENTIAL_FEE_BASE.saturating_add(message_size_factor),
+			);
+			*f
+		})
+	}
+
+	fn decrease_fee_factor(id: Self::Id) -> FixedU128 {
+		<DeliveryFeeFactor<T>>::mutate(id, |f| {
+			*f = InitialFactor::get().max(*f / delivery_fee_constants::EXPONENTIAL_FEE_BASE);
+			*f
+		})
+	}
+}

cumulus/pallets/xcmp-queue/src/mock.rs

@@ -85,8 +85,10 @@ parameter_types! {
 	pub const MaxReserves: u32 = 50;
 }
 
+pub type Balance = u64;
+
 impl pallet_balances::Config for Test {
-	type Balance = u64;
+	type Balance = Balance;
 	type RuntimeEvent = RuntimeEvent;
 	type DustRemoval = ();
 	type ExistentialDeposit = ExistentialDeposit;
@@ -196,6 +198,22 @@ impl<RuntimeOrigin: OriginTrait> ConvertOrigin<RuntimeOrigin>
 	}
 }
 
+parameter_types! {
+	/// The asset ID for the asset that we use to pay for message delivery fees.
+	pub FeeAssetId: AssetId = Concrete(RelayChain::get());
+	/// The base fee for the message delivery fees.
+	pub const BaseDeliveryFee: Balance = 300_000_000;
+	/// The fee per byte
+	pub const ByteFee: Balance = 1_000_000;
+}
+
+pub type PriceForSiblingParachainDelivery = polkadot_runtime_common::xcm_sender::ExponentialPrice<
+	FeeAssetId,
+	BaseDeliveryFee,
+	ByteFee,
+	XcmpQueue,
+>;
+
 impl Config for Test {
 	type RuntimeEvent = RuntimeEvent;
 	type XcmExecutor = xcm_executor::XcmExecutor<XcmConfig>;
@@ -205,7 +223,7 @@ impl Config for Test {
 	type ControllerOrigin = EnsureRoot<AccountId>;
 	type ControllerOriginConverter = SystemParachainAsSuperuser<RuntimeOrigin>;
 	type WeightInfo = ();
-	type PriceForSiblingDelivery = ();
+	type PriceForSiblingDelivery = PriceForSiblingParachainDelivery;
 }
 
 pub fn new_test_ext() -> sp_io::TestExternalities {

cumulus/pallets/xcmp-queue/src/tests.rs

@@ -324,13 +324,14 @@ fn xcmp_queue_consumes_dest_and_msg_on_ok_validate() {
 	let dest = (Parent, X1(Parachain(5555)));
 	let mut dest_wrapper = Some(dest.into());
 	let mut msg_wrapper = Some(message.clone());
-	assert!(<XcmpQueue as SendXcm>::validate(&mut dest_wrapper, &mut msg_wrapper).is_ok());
-
-	// check wrapper were consumed
-	assert_eq!(None, dest_wrapper.take());
-	assert_eq!(None, msg_wrapper.take());
 
 	new_test_ext().execute_with(|| {
+		assert!(<XcmpQueue as SendXcm>::validate(&mut dest_wrapper, &mut msg_wrapper).is_ok());
+
+		// check wrapper were consumed
+		assert_eq!(None, dest_wrapper.take());
+		assert_eq!(None, msg_wrapper.take());
+
 		// another try with router chain with asserting sender
 		assert_eq!(
 			Err(SendError::Transport("NoChannel")),
@@ -370,3 +371,74 @@ fn xcmp_queue_send_xcm_works() {
 			.any(|(para_id, _)| para_id == &sibling_para_id));
 	})
 }
+
+#[test]
+fn verify_fee_factor_increase_and_decrease() {
+	use cumulus_primitives_core::AbridgedHrmpChannel;
+	use sp_runtime::FixedU128;
+
+	let sibling_para_id = ParaId::from(12345);
+	let destination = (Parent, Parachain(sibling_para_id.into())).into();
+	let xcm = Xcm(vec![ClearOrigin; 100]);
+	let versioned_xcm = VersionedXcm::from(xcm.clone());
+	let mut xcmp_message = XcmpMessageFormat::ConcatenatedVersionedXcm.encode();
+	xcmp_message.extend(versioned_xcm.encode());
+
+	new_test_ext().execute_with(|| {
+		let initial = InitialFactor::get();
+		assert_eq!(DeliveryFeeFactor::<Test>::get(sibling_para_id), initial);
+
+		// Open channel so messages can actually be sent
+		ParachainSystem::open_custom_outbound_hrmp_channel_for_benchmarks_or_tests(
+			sibling_para_id,
+			AbridgedHrmpChannel {
+				max_capacity: 10,
+				max_total_size: 1000,
+				max_message_size: 104,
+				msg_count: 0,
+				total_size: 0,
+				mqc_head: None,
+			},
+		);
+
+		// Fee factor is only increased in `send_fragment`, which is called by `send_xcm`.
+		// When queue is not congested, fee factor doesn't change.
+		assert_ok!(send_xcm::<XcmpQueue>(destination, xcm.clone())); // Size 104
+		assert_ok!(send_xcm::<XcmpQueue>(destination, xcm.clone())); // Size 208
+		assert_ok!(send_xcm::<XcmpQueue>(destination, xcm.clone())); // Size 312
+		assert_ok!(send_xcm::<XcmpQueue>(destination, xcm.clone())); // Size 416
+		assert_eq!(DeliveryFeeFactor::<Test>::get(sibling_para_id), initial);
+
+		// Sending the message right now is cheap
+		let (_, delivery_fees) = validate_send::<XcmpQueue>(destination, xcm.clone())
+			.expect("message can be sent; qed");
+		let Fungible(delivery_fee_amount) = delivery_fees.inner()[0].fun else { unreachable!("asset is fungible; qed"); };
+		assert_eq!(delivery_fee_amount, 402_000_000);
+
+		let smaller_xcm = Xcm(vec![ClearOrigin; 30]);
+
+		// When we get to half of `max_total_size`, because `THRESHOLD_FACTOR` is 2,
+		// then the fee factor starts to increase.
+		assert_ok!(send_xcm::<XcmpQueue>(destination, xcm.clone())); // Size 520
+		assert_eq!(DeliveryFeeFactor::<Test>::get(sibling_para_id), FixedU128::from_float(1.05));
+
+		for _ in 0..12 { // We finish at size 929
+			assert_ok!(send_xcm::<XcmpQueue>(destination, smaller_xcm.clone()));
+		}
+		assert!(DeliveryFeeFactor::<Test>::get(sibling_para_id) > FixedU128::from_float(1.88));
+
+		// Sending the message right now is expensive
+		let (_, delivery_fees) = validate_send::<XcmpQueue>(destination, xcm.clone())
+			.expect("message can be sent; qed");
+		let Fungible(delivery_fee_amount) = delivery_fees.inner()[0].fun else { unreachable!("asset is fungible; qed"); };
+		assert_eq!(delivery_fee_amount, 758_030_955);
+
+		// Fee factor only decreases in `take_outbound_messages`
+		for _ in 0..5 { // We take 5 100 byte pages
+			XcmpQueue::take_outbound_messages(1);
+		}
+		assert!(DeliveryFeeFactor::<Test>::get(sibling_para_id) < FixedU128::from_float(1.72));
+		XcmpQueue::take_outbound_messages(1);
+		assert!(DeliveryFeeFactor::<Test>::get(sibling_para_id) < FixedU128::from_float(1.63));
+	});
+}