[Runtime] Bound XCMP queue (#2302)

Remove `without_storage_info` from the XCMP queue pallet. Part of
https://github.com/paritytech/polkadot-sdk/issues/323

Changes:
- Limit the number of channels that can be suspended at the same time.
- Limit the number of channels that can have messages or signals pending
at the same time.

A No-OP migration is put in place to ensure that all `BoundedVec`s still
decode and not truncate after upgrade. The storage version is thereby
bumped to 4 to have our tooling remind us to deploy that migration.

---------

Signed-off-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io>
Co-authored-by: Francisco Aguirre <franciscoaguirreperez@gmail.com>

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

@@ -30,7 +30,7 @@
 use codec::{Decode, Encode};
 use cumulus_primitives_core::{
 	relay_chain, AbridgedHostConfiguration, ChannelInfo, ChannelStatus, CollationInfo,
-	GetChannelInfo, InboundDownwardMessage, InboundHrmpMessage, MessageSendError,
+	GetChannelInfo, InboundDownwardMessage, InboundHrmpMessage, ListChannelInfos, MessageSendError,
 	OutboundHrmpMessage, ParaId, PersistedValidationData, UpwardMessage, UpwardMessageSender,
 	XcmpMessageHandler, XcmpMessageSource,
 };
@@ -1023,6 +1023,13 @@ impl<T: Config> FeeTracker for Pallet<T> {
 	}
 }
 
+impl<T: Config> ListChannelInfos for Pallet<T> {
+	fn outgoing_channels() -> Vec<ParaId> {
+		let Some(state) = Self::relevant_messaging_state() else { return Vec::new() };
+		state.egress_channels.into_iter().map(|(id, _)| id).collect()
+	}
+}
+
 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

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

@@ -51,7 +51,7 @@ pub mod weights;
 pub use weights::WeightInfo;
 
 use bounded_collections::BoundedBTreeSet;
-use codec::{Decode, DecodeLimit, Encode};
+use codec::{Decode, DecodeLimit, Encode, MaxEncodedLen};
 use cumulus_primitives_core::{
 	relay_chain::BlockNumber as RelayBlockNumber, ChannelStatus, GetChannelInfo, MessageSendError,
 	ParaId, XcmpMessageFormat, XcmpMessageHandler, XcmpMessageSource,
@@ -105,7 +105,6 @@ pub mod pallet {
 
 	#[pallet::pallet]
 	#[pallet::storage_version(migration::STORAGE_VERSION)]
-	#[pallet::without_storage_info]
 	pub struct Pallet<T>(_);
 
 	#[pallet::config]
@@ -132,6 +131,25 @@ pub mod pallet {
 		#[pallet::constant]
 		type MaxInboundSuspended: Get<u32>;
 
+		/// Maximal number of outbound XCMP channels that can have messages queued at the same time.
+		///
+		/// If this is reached, then no further messages can be sent to channels that do not yet
+		/// have a message queued. This should be set to the expected maximum of outbound channels
+		/// which is determined by [`Self::ChannelInfo`]. It is important to set this correctly,
+		/// since otherwise the congestion control protocol will not work as intended and messages
+		/// may be dropped. This value increases the PoV and should therefore not be picked too
+		/// high.
+		#[pallet::constant]
+		type MaxActiveOutboundChannels: Get<u32>;
+
+		/// The maximal page size for HRMP message pages.
+		///
+		/// A lower limit can be set dynamically, but this is the hard-limit for the PoV worst case
+		/// benchmarking. The limit for the size of a message is slightly below this, since some
+		/// overhead is incurred for encoding the format.
+		#[pallet::constant]
+		type MaxPageSize: Get<u32>;
+
 		/// The origin that is allowed to resume or suspend the XCMP queue.
 		type ControllerOrigin: EnsureOrigin<Self::RuntimeOrigin>;
 
@@ -276,6 +294,10 @@ pub mod pallet {
 		AlreadySuspended,
 		/// The execution is already resumed.
 		AlreadyResumed,
+		/// There are too many active outbound channels.
+		TooManyOutboundChannels,
+		/// The message is too big.
+		TooBig,
 	}
 
 	/// The suspended inbound XCMP channels. All others are not suspended.
@@ -297,19 +319,28 @@ pub mod pallet {
 	/// case of the need to send a high-priority signal message this block.
 	/// The bool is true if there is a signal message waiting to be sent.
 	#[pallet::storage]
-	pub(super) type OutboundXcmpStatus<T: Config> =
+	pub(super) type OutboundXcmpStatus<T: Config> = StorageValue<
-		StorageValue<_, Vec<OutboundChannelDetails>, ValueQuery>;
+		_,
+		BoundedVec<OutboundChannelDetails, T::MaxActiveOutboundChannels>,
+		ValueQuery,
+	>;
 
-	// The new way of doing it:
 	/// The messages outbound in a given XCMP channel.
 	#[pallet::storage]
-	pub(super) type OutboundXcmpMessages<T: Config> =
+	pub(super) type OutboundXcmpMessages<T: Config> = StorageDoubleMap<
-		StorageDoubleMap<_, Blake2_128Concat, ParaId, Twox64Concat, u16, Vec<u8>, ValueQuery>;
+		_,
+		Blake2_128Concat,
+		ParaId,
+		Twox64Concat,
+		u16,
+		BoundedVec<u8, T::MaxPageSize>,
+		ValueQuery,
+	>;
 
 	/// Any signal messages waiting to be sent.
 	#[pallet::storage]
 	pub(super) type SignalMessages<T: Config> =
-		StorageMap<_, Blake2_128Concat, ParaId, Vec<u8>, ValueQuery>;
+		StorageMap<_, Blake2_128Concat, ParaId, BoundedVec<u8, T::MaxPageSize>, ValueQuery>;
 
 	/// The configuration which controls the dynamics of the outbound queue.
 	#[pallet::storage]
@@ -331,15 +362,14 @@ pub mod pallet {
 		StorageMap<_, Twox64Concat, ParaId, FixedU128, ValueQuery, InitialFactor>;
 }
 
-#[derive(Copy, Clone, Eq, PartialEq, Encode, Decode, RuntimeDebug, TypeInfo)]
+#[derive(Copy, Clone, Eq, PartialEq, Encode, Decode, RuntimeDebug, TypeInfo, MaxEncodedLen)]
 pub enum OutboundState {
 	Ok,
 	Suspended,
 }
 
 /// Struct containing detailed information about the outbound channel.
-#[derive(Clone, Eq, PartialEq, Encode, Decode, TypeInfo)]
+#[derive(Clone, Eq, PartialEq, Encode, Decode, TypeInfo, RuntimeDebug, MaxEncodedLen)]
-#[cfg_attr(feature = "std", derive(Debug))]
 pub struct OutboundChannelDetails {
 	/// The `ParaId` of the parachain that this channel is connected with.
 	recipient: ParaId,
@@ -375,7 +405,7 @@ impl OutboundChannelDetails {
 	}
 }
 
-#[derive(Copy, Clone, Eq, PartialEq, Encode, Decode, RuntimeDebug, TypeInfo)]
+#[derive(Copy, Clone, Eq, PartialEq, Encode, Decode, RuntimeDebug, TypeInfo, MaxEncodedLen)]
 pub struct QueueConfigData {
 	/// The number of pages which must be in the queue for the other side to be told to suspend
 	/// their sending.
@@ -478,7 +508,9 @@ impl<T: Config> Pallet<T> {
 		{
 			details
 		} else {
-			all_channels.push(OutboundChannelDetails::new(recipient));
+			all_channels
+				.try_push(OutboundChannelDetails::new(recipient))
+				.map_err(|_| MessageSendError::TooManyChannels)?;
 			all_channels
 				.last_mut()
 				.expect("can't be empty; a new element was just pushed; qed")
@@ -503,7 +535,7 @@ impl<T: Config> Pallet<T> {
 						if page.len() + encoded_fragment.len() > max_message_size {
 							return None
 						}
-						page.extend_from_slice(&encoded_fragment[..]);
+						page.try_extend(encoded_fragment.iter().cloned()).ok()?;
 						Some(page.len())
 					},
 				)
@@ -521,7 +553,9 @@ impl<T: Config> Pallet<T> {
 			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 bounded_page =
+				BoundedVec::try_from(new_page).map_err(|_| MessageSendError::TooBig)?;
+			<OutboundXcmpMessages<T>>::insert(recipient, page_index, bounded_page);
 			<OutboundXcmpStatus<T>>::put(all_channels);
 			(number_of_pages, last_page_size)
 		};
@@ -543,17 +577,21 @@ impl<T: Config> Pallet<T> {
 
 	/// Sends a signal to the `dest` chain over XCMP. This is guaranteed to be dispatched on this
 	/// block.
-	fn send_signal(dest: ParaId, signal: ChannelSignal) {
+	fn send_signal(dest: ParaId, signal: ChannelSignal) -> Result<(), Error<T>> {
 		let mut s = <OutboundXcmpStatus<T>>::get();
 		if let Some(details) = s.iter_mut().find(|item| item.recipient == dest) {
 			details.signals_exist = true;
 		} else {
-			s.push(OutboundChannelDetails::new(dest).with_signals());
+			s.try_push(OutboundChannelDetails::new(dest).with_signals())
+				.map_err(|_| Error::<T>::TooManyOutboundChannels)?;
 		}
-		<SignalMessages<T>>::mutate(dest, |page| {
+
-			*page = (XcmpMessageFormat::Signals, signal).encode();
+		let page = BoundedVec::try_from((XcmpMessageFormat::Signals, signal).encode())
-		});
+			.map_err(|_| Error::<T>::TooBig)?;
+
+		<SignalMessages<T>>::insert(dest, page);
 		<OutboundXcmpStatus<T>>::put(s);
+		Ok(())
 	}
 
 	fn suspend_channel(target: ParaId) {
@@ -563,7 +601,13 @@ impl<T: Config> Pallet<T> {
 				defensive_assert!(ok, "WARNING: Attempt to suspend channel that was not Ok.");
 				details.state = OutboundState::Suspended;
 			} else {
-				s.push(OutboundChannelDetails::new(target).with_suspended_state());
+				if s.try_push(OutboundChannelDetails::new(target).with_suspended_state()).is_err() {
+					// Nothing that we can do here. The outbound channel does not exist either, so
+					// there should be no message going out as well. The next time that the channel
+					// can be created it will again get the suspension from the remote side. It can
+					// therefore result in a few lost messages, but should eventually self-repair.
+					defensive!("Cannot pause channel; too many outbound channels");
+				}
 			}
 		});
 	}
@@ -664,13 +708,17 @@ impl<T: Config> OnQueueChanged<ParaId> for Pallet<T> {
 		let suspended = suspended_channels.contains(&para);
 
 		if suspended && fp.pages <= resume_threshold {
-			Self::send_signal(para, ChannelSignal::Resume);
+			// If the resuming fails then it is not critical. We will retry in the future.
+			let _ = Self::send_signal(para, ChannelSignal::Resume);
 
 			suspended_channels.remove(&para);
 			<InboundXcmpSuspended<T>>::put(suspended_channels);
 		} else if !suspended && fp.pages >= suspend_threshold {
 			log::warn!("XCMP queue for sibling {:?} is full; suspending channel.", para);
-			Self::send_signal(para, ChannelSignal::Suspend);
+			if let Err(_) = Self::send_signal(para, ChannelSignal::Suspend) {
+				// It will retry if `drop_threshold` is not reached, but it could be too late.
+				defensive!("Could not send suspension signal; future messages may be dropped.");
+			}
 
 			if let Err(err) = suspended_channels.try_insert(para) {
 				log::error!("Too many channels suspended; cannot suspend sibling {:?}: {:?}; further messages may be dropped.", para, err);
@@ -842,7 +890,7 @@ impl<T: Config> XcmpMessageSource for Pallet<T> {
 				//   since it's so unlikely then for now we just drop it.
 				defensive!("WARNING: oversize message in queue - dropping");
 			} else {
-				result.push((para_id, page));
+				result.push((para_id, page.into_inner()));
 			}
 
 			let max_total_size = match T::ChannelInfo::get_channel_info(para_id) {
@@ -890,7 +938,14 @@ impl<T: Config> XcmpMessageSource for Pallet<T> {
 		let pruned = old_statuses_len - statuses.len();
 		// removing an item from status implies a message being sent, so the result messages must
 		// be no less than the pruned channels.
-		statuses.rotate_left(result.len().saturating_sub(pruned));
+
+		// TODO <https://github.com/paritytech/parity-common/pull/800>
+		{
+			let mut statuses_inner = statuses.into_inner();
+			statuses_inner.rotate_left(result.len().saturating_sub(pruned));
+			statuses = BoundedVec::try_from(statuses_inner)
+				.expect("Rotating does not change the length; it still fits; qed");
+		}
 
 		<OutboundXcmpStatus<T>>::put(statuses);
 

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

@@ -16,6 +16,8 @@
 
 //! A module that is responsible for migration of storage.
 
+pub mod v5;
+
 use crate::{Config, OverweightIndex, Pallet, QueueConfig, QueueConfigData, DEFAULT_POV_SIZE};
 use cumulus_primitives_core::XcmpMessageFormat;
 use frame_support::{
@@ -25,7 +27,7 @@ use frame_support::{
 };
 
 /// The current storage version.
-pub const STORAGE_VERSION: StorageVersion = StorageVersion::new(4);
+pub const STORAGE_VERSION: StorageVersion = StorageVersion::new(5);
 
 pub const LOG: &str = "runtime::xcmp-queue-migration";
 
@@ -264,9 +266,9 @@ pub mod v4 {
 
 	/// Migrates `QueueConfigData` to v4, removing deprecated fields and bumping page
 	/// thresholds to at least the default values.
-	pub struct UncheckedMigrationToV4<T: Config>(PhantomData<T>);
+	pub struct UncheckedMigrateV3ToV4<T: Config>(PhantomData<T>);
 
-	impl<T: Config> OnRuntimeUpgrade for UncheckedMigrationToV4<T> {
+	impl<T: Config> OnRuntimeUpgrade for UncheckedMigrateV3ToV4<T> {
 		fn on_runtime_upgrade() -> Weight {
 			let translate = |pre: v2::QueueConfigData| -> QueueConfigData {
 				let pre_default = v2::QueueConfigData::default();
@@ -299,13 +301,13 @@ pub mod v4 {
 		}
 	}
 
-	/// [`UncheckedMigrationToV4`] wrapped in a
+	/// [`UncheckedMigrateV3ToV4`] wrapped in a
 	/// [`VersionedMigration`](frame_support::migrations::VersionedMigration), ensuring the
 	/// migration is only performed when on-chain version is 3.
-	pub type MigrationToV4<T> = frame_support::migrations::VersionedMigration<
+	pub type MigrateV3ToV4<T> = frame_support::migrations::VersionedMigration<
 		3,
 		4,
-		UncheckedMigrationToV4<T>,
+		UncheckedMigrateV3ToV4<T>,
 		Pallet<T>,
 		<T as frame_system::Config>::DbWeight,
 	>;
@@ -372,10 +374,10 @@ mod tests {
 				&v2.encode(),
 			);
 
-			let bytes = v4::MigrationToV4::<Test>::pre_upgrade();
+			let bytes = v4::MigrateV3ToV4::<Test>::pre_upgrade();
 			assert!(bytes.is_ok());
-			v4::MigrationToV4::<Test>::on_runtime_upgrade();
+			v4::MigrateV3ToV4::<Test>::on_runtime_upgrade();
-			assert!(v4::MigrationToV4::<Test>::post_upgrade(bytes.unwrap()).is_ok());
+			assert!(v4::MigrateV3ToV4::<Test>::post_upgrade(bytes.unwrap()).is_ok());
 
 			let v4 = QueueConfig::<Test>::get();
 
@@ -401,10 +403,10 @@ mod tests {
 				&v2.encode(),
 			);
 
-			let bytes = v4::MigrationToV4::<Test>::pre_upgrade();
+			let bytes = v4::MigrateV3ToV4::<Test>::pre_upgrade();
 			assert!(bytes.is_ok());
-			v4::MigrationToV4::<Test>::on_runtime_upgrade();
+			v4::MigrateV3ToV4::<Test>::on_runtime_upgrade();
-			assert!(v4::MigrationToV4::<Test>::post_upgrade(bytes.unwrap()).is_ok());
+			assert!(v4::MigrateV3ToV4::<Test>::post_upgrade(bytes.unwrap()).is_ok());
 
 			let v4 = QueueConfig::<Test>::get();
 

cumulus/pallets/xcmp-queue/src/migration/v5.rs

@@ -0,0 +1,115 @@
+// Copyright (C) Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Polkadot is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Migrates the storage to version 5.
+
+use crate::*;
+use cumulus_primitives_core::ListChannelInfos;
+use frame_support::{pallet_prelude::*, traits::OnRuntimeUpgrade};
+
+/// Configs needed to run the V5 migration.
+pub trait V5Config: Config {
+	/// List all outbound channels with their target `ParaId` and maximum message size.
+	type ChannelList: ListChannelInfos;
+}
+
+/// Ensures that the storage migrates cleanly to V5.
+///
+/// The migration itself is a no-op, but it checks that none of the `BoundedVec`s would truncate on
+/// the next decode after the upgrade was applied.
+pub type MigrateV4ToV5<T> = frame_support::migrations::VersionedMigration<
+	4,
+	5,
+	unversioned::UncheckedMigrateV4ToV5<T>,
+	Pallet<T>,
+	<T as frame_system::Config>::DbWeight,
+>;
+
+// V4 storage aliases
+mod v4 {
+	use super::*;
+
+	#[frame_support::storage_alias]
+	pub(super) type OutboundXcmpStatus<T: Config> =
+		StorageValue<Pallet<T>, Vec<OutboundChannelDetails>, ValueQuery>;
+
+	#[frame_support::storage_alias]
+	pub(super) type OutboundXcmpMessages<T: Config> = StorageDoubleMap<
+		Pallet<T>,
+		Blake2_128Concat,
+		ParaId,
+		Twox64Concat,
+		u16,
+		Vec<u8>,
+		ValueQuery,
+	>;
+
+	#[frame_support::storage_alias]
+	pub(super) type SignalMessages<T: Config> =
+		StorageMap<Pallet<T>, Blake2_128Concat, ParaId, Vec<u8>, ValueQuery>;
+}
+
+// Private module to hide the migration.
+mod unversioned {
+	/// Please use [`MigrateV4ToV5`] instead.
+	pub struct UncheckedMigrateV4ToV5<T: super::V5Config>(core::marker::PhantomData<T>);
+}
+
+impl<T: V5Config> OnRuntimeUpgrade for unversioned::UncheckedMigrateV4ToV5<T> {
+	fn on_runtime_upgrade() -> frame_support::weights::Weight {
+		Default::default()
+	}
+
+	#[cfg(feature = "try-runtime")]
+	fn post_upgrade(_: Vec<u8>) -> Result<(), sp_runtime::DispatchError> {
+		// We dont need any front-run protection for this since channels are opened by governance.
+		ensure!(
+			v4::OutboundXcmpStatus::<T>::get().len() as u32 <= T::MaxActiveOutboundChannels::get(),
+			"Too many outbound channels. Close some channels or increase `MaxActiveOutboundChannels`."
+		);
+
+		// Check if any channels have a too large message max sizes.
+		let max_msg_len = T::MaxPageSize::get() - XcmpMessageFormat::max_encoded_len() as u32;
+		for channel in T::ChannelList::outgoing_channels() {
+			let info = T::ChannelInfo::get_channel_info(channel)
+				.expect("All listed channels must provide info");
+
+			ensure!(
+				info.max_message_size <= max_msg_len,
+				"Max message size for channel is too large. This means that the V5 migration can \
+				be front-run and an attacker could place a large message just right before the \
+				migration to make other messages un-decodable. Please either increase \
+				`MaxPageSize` or decrease the `max_message_size` for this channel.",
+			);
+		}
+
+		// Now check that all pages still fit into the new `BoundedVec`s:
+		for page in v4::OutboundXcmpMessages::<T>::iter_values() {
+			ensure!(
+				page.len() < T::MaxPageSize::get() as usize,
+				"Too long message in storage. Either manually truncate the pages or increase `MaxPageSize`."
+			);
+		}
+		for page in v4::SignalMessages::<T>::iter_values() {
+			ensure!(
+				page.len() < T::MaxPageSize::get() as usize,
+				"Too long signal in storage. Either manually truncate the pages or increase `MaxPageSize`."
+			);
+		}
+
+		Ok(())
+	}
+}

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

@@ -275,7 +275,9 @@ impl Config for Test {
 	type ChannelInfo = MockedChannelInfo;
 	type VersionWrapper = ();
 	type XcmpQueue = EnqueueToLocalStorage<Pallet<Test>>;
-	type MaxInboundSuspended = sp_core::ConstU32<1_000>;
+	type MaxInboundSuspended = ConstU32<1_000>;
+	type MaxActiveOutboundChannels = ConstU32<128>;
+	type MaxPageSize = ConstU32<{ 1 << 16 }>;
 	type ControllerOrigin = EnsureRoot<AccountId>;
 	type ControllerOriginConverter = SystemParachainAsSuperuser<RuntimeOrigin>;
 	type WeightInfo = ();

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

@@ -520,7 +520,7 @@ fn hrmp_signals_are_prioritized() {
 		});
 
 		// But a signal gets prioritized instead of the messages:
-		XcmpQueue::send_signal(sibling_para_id.into(), ChannelSignal::Suspend);
+		assert_ok!(XcmpQueue::send_signal(sibling_para_id.into(), ChannelSignal::Suspend));
 
 		let taken = XcmpQueue::take_outbound_messages(130);
 		assert_eq!(

cumulus/parachain-template/runtime/src/lib.rs

@@ -422,7 +422,9 @@ impl cumulus_pallet_xcmp_queue::Config for Runtime {
 	type VersionWrapper = ();
 	// Enqueue XCMP messages from siblings for later processing.
 	type XcmpQueue = TransformOrigin<MessageQueue, AggregateMessageOrigin, ParaId, ParaIdToSibling>;
-	type MaxInboundSuspended = sp_core::ConstU32<1_000>;
+	type MaxInboundSuspended = ConstU32<1_000>;
+	type MaxActiveOutboundChannels = ConstU32<128>;
+	type MaxPageSize = ConstU32<{ 1 << 16 }>;
 	type ControllerOrigin = EnsureRoot<AccountId>;
 	type ControllerOriginConverter = XcmOriginToTransactDispatchOrigin;
 	type WeightInfo = ();

cumulus/parachains/runtimes/assets/asset-hub-rococo/src/lib.rs

@@ -716,12 +716,19 @@ impl cumulus_pallet_xcmp_queue::Config for Runtime {
 	type ChannelInfo = ParachainSystem;
 	type VersionWrapper = PolkadotXcm;
 	type XcmpQueue = TransformOrigin<MessageQueue, AggregateMessageOrigin, ParaId, ParaIdToSibling>;
-	type MaxInboundSuspended = sp_core::ConstU32<1_000>;
+	type MaxInboundSuspended = ConstU32<1_000>;
+	type MaxActiveOutboundChannels = ConstU32<128>;
+	type MaxPageSize = ConstU32<{ 1 << 16 }>;
 	type ControllerOrigin = EnsureRoot<AccountId>;
 	type ControllerOriginConverter = xcm_config::XcmOriginToTransactDispatchOrigin;
 	type PriceForSiblingDelivery = PriceForSiblingParachainDelivery;
 }
 
+impl cumulus_pallet_xcmp_queue::migration::v5::V5Config for Runtime {
+	// This must be the same as the `ChannelInfo` from the `Config`:
+	type ChannelList = ParachainSystem;
+}
+
 parameter_types! {
 	pub const RelayOrigin: AggregateMessageOrigin = AggregateMessageOrigin::Parent;
 }
@@ -993,8 +1000,8 @@ pub type UncheckedExtrinsic =
 pub type Migrations = (
 	pallet_collator_selection::migration::v1::MigrateToV1<Runtime>,
 	InitStorageVersions,
-	// unreleased
+	cumulus_pallet_xcmp_queue::migration::v4::MigrateV3ToV4<Runtime>,
-	cumulus_pallet_xcmp_queue::migration::v4::MigrationToV4<Runtime>,
+	cumulus_pallet_xcmp_queue::migration::v5::MigrateV4ToV5<Runtime>,
 );
 
 /// Migration to initialize storage versions for pallets added after genesis.

cumulus/parachains/runtimes/assets/asset-hub-westend/src/lib.rs

@@ -668,13 +668,20 @@ impl cumulus_pallet_xcmp_queue::Config for Runtime {
 	type VersionWrapper = PolkadotXcm;
 	// Enqueue XCMP messages from siblings for later processing.
 	type XcmpQueue = TransformOrigin<MessageQueue, AggregateMessageOrigin, ParaId, ParaIdToSibling>;
-	type MaxInboundSuspended = sp_core::ConstU32<1_000>;
+	type MaxInboundSuspended = ConstU32<1_000>;
+	type MaxActiveOutboundChannels = ConstU32<128>;
+	type MaxPageSize = ConstU32<{ 1 << 16 }>;
 	type ControllerOrigin = EnsureRoot<AccountId>;
 	type ControllerOriginConverter = XcmOriginToTransactDispatchOrigin;
 	type WeightInfo = weights::cumulus_pallet_xcmp_queue::WeightInfo<Runtime>;
 	type PriceForSiblingDelivery = PriceForSiblingParachainDelivery;
 }
 
+impl cumulus_pallet_xcmp_queue::migration::v5::V5Config for Runtime {
+	// This must be the same as the `ChannelInfo` from the `Config`:
+	type ChannelList = ParachainSystem;
+}
+
 parameter_types! {
 	pub const RelayOrigin: AggregateMessageOrigin = AggregateMessageOrigin::Parent;
 }
@@ -947,8 +954,8 @@ pub type Migrations = (
 	InitStorageVersions,
 	// unreleased
 	DeleteUndecodableStorage,
-	// unreleased
+	cumulus_pallet_xcmp_queue::migration::v4::MigrateV3ToV4<Runtime>,
-	cumulus_pallet_xcmp_queue::migration::v4::MigrationToV4<Runtime>,
+	cumulus_pallet_xcmp_queue::migration::v5::MigrateV4ToV5<Runtime>,
 );
 
 /// Asset Hub Westend has some undecodable storage, delete it.

cumulus/parachains/runtimes/bridge-hubs/bridge-hub-rococo/src/lib.rs

@@ -144,8 +144,9 @@ pub type Migrations = (
 	pallet_collator_selection::migration::v1::MigrateToV1<Runtime>,
 	pallet_multisig::migrations::v1::MigrateToV1<Runtime>,
 	InitStorageVersions,
-	cumulus_pallet_xcmp_queue::migration::v4::MigrationToV4<Runtime>,
 	// unreleased
+	cumulus_pallet_xcmp_queue::migration::v4::MigrateV3ToV4<Runtime>,
+	cumulus_pallet_xcmp_queue::migration::v5::MigrateV4ToV5<Runtime>,
 	snowbridge_pallet_system::migration::v0::InitializeOnUpgrade<
 		Runtime,
 		ConstU32<BRIDGE_HUB_ID>,
@@ -405,13 +406,20 @@ impl cumulus_pallet_xcmp_queue::Config for Runtime {
 	type VersionWrapper = PolkadotXcm;
 	// Enqueue XCMP messages from siblings for later processing.
 	type XcmpQueue = TransformOrigin<MessageQueue, AggregateMessageOrigin, ParaId, ParaIdToSibling>;
-	type MaxInboundSuspended = sp_core::ConstU32<1_000>;
+	type MaxInboundSuspended = ConstU32<1_000>;
+	type MaxActiveOutboundChannels = ConstU32<128>;
+	type MaxPageSize = ConstU32<{ 1 << 16 }>;
 	type ControllerOrigin = EnsureRoot<AccountId>;
 	type ControllerOriginConverter = XcmOriginToTransactDispatchOrigin;
 	type WeightInfo = weights::cumulus_pallet_xcmp_queue::WeightInfo<Runtime>;
 	type PriceForSiblingDelivery = PriceForSiblingParachainDelivery;
 }
 
+impl cumulus_pallet_xcmp_queue::migration::v5::V5Config for Runtime {
+	// This must be the same as the `ChannelInfo` from the `Config`:
+	type ChannelList = ParachainSystem;
+}
+
 parameter_types! {
 	pub const RelayOrigin: AggregateMessageOrigin = AggregateMessageOrigin::Parent;
 }

cumulus/parachains/runtimes/bridge-hubs/bridge-hub-westend/src/lib.rs

@@ -123,8 +123,8 @@ pub type Migrations = (
 	pallet_collator_selection::migration::v1::MigrateToV1<Runtime>,
 	pallet_multisig::migrations::v1::MigrateToV1<Runtime>,
 	InitStorageVersions,
-	// unreleased
+	cumulus_pallet_xcmp_queue::migration::v4::MigrateV3ToV4<Runtime>,
-	cumulus_pallet_xcmp_queue::migration::v4::MigrationToV4<Runtime>,
+	cumulus_pallet_xcmp_queue::migration::v5::MigrateV4ToV5<Runtime>,
 );
 
 /// Migration to initialize storage versions for pallets added after genesis.
@@ -371,13 +371,20 @@ impl cumulus_pallet_xcmp_queue::Config for Runtime {
 	type ChannelInfo = ParachainSystem;
 	type VersionWrapper = PolkadotXcm;
 	type XcmpQueue = TransformOrigin<MessageQueue, AggregateMessageOrigin, ParaId, ParaIdToSibling>;
-	type MaxInboundSuspended = sp_core::ConstU32<1_000>;
+	type MaxInboundSuspended = ConstU32<1_000>;
+	type MaxActiveOutboundChannels = ConstU32<128>;
+	type MaxPageSize = ConstU32<{ 1 << 16 }>;
 	type ControllerOrigin = EnsureRoot<AccountId>;
 	type ControllerOriginConverter = XcmOriginToTransactDispatchOrigin;
 	type WeightInfo = weights::cumulus_pallet_xcmp_queue::WeightInfo<Runtime>;
 	type PriceForSiblingDelivery = PriceForSiblingParachainDelivery;
 }
 
+impl cumulus_pallet_xcmp_queue::migration::v5::V5Config for Runtime {
+	// This must be the same as the `ChannelInfo` from the `Config`:
+	type ChannelList = ParachainSystem;
+}
+
 parameter_types! {
 	pub const RelayOrigin: AggregateMessageOrigin = AggregateMessageOrigin::Parent;
 }

cumulus/parachains/runtimes/collectives/collectives-westend/src/lib.rs

@@ -445,13 +445,20 @@ impl cumulus_pallet_xcmp_queue::Config for Runtime {
 	type VersionWrapper = PolkadotXcm;
 	// Enqueue XCMP messages from siblings for later processing.
 	type XcmpQueue = TransformOrigin<MessageQueue, AggregateMessageOrigin, ParaId, ParaIdToSibling>;
-	type MaxInboundSuspended = sp_core::ConstU32<1_000>;
+	type MaxInboundSuspended = ConstU32<1_000>;
+	type MaxActiveOutboundChannels = ConstU32<128>;
+	type MaxPageSize = ConstU32<{ 1 << 16 }>;
 	type ControllerOrigin = EitherOfDiverse<EnsureRoot<AccountId>, Fellows>;
 	type ControllerOriginConverter = XcmOriginToTransactDispatchOrigin;
 	type WeightInfo = weights::cumulus_pallet_xcmp_queue::WeightInfo<Runtime>;
 	type PriceForSiblingDelivery = PriceForSiblingParachainDelivery;
 }
 
+impl cumulus_pallet_xcmp_queue::migration::v5::V5Config for Runtime {
+	// This must be the same as the `ChannelInfo` from the `Config`:
+	type ChannelList = ParachainSystem;
+}
+
 parameter_types! {
 	pub const RelayOrigin: AggregateMessageOrigin = AggregateMessageOrigin::Parent;
 }
@@ -720,8 +727,8 @@ pub type UncheckedExtrinsic =
 type Migrations = (
 	// unreleased
 	pallet_collator_selection::migration::v1::MigrateToV1<Runtime>,
-	// unreleased
+	cumulus_pallet_xcmp_queue::migration::v4::MigrateV3ToV4<Runtime>,
-	cumulus_pallet_xcmp_queue::migration::v4::MigrationToV4<Runtime>,
+	cumulus_pallet_xcmp_queue::migration::v5::MigrateV4ToV5<Runtime>,
 );
 
 /// Executive: handles dispatch to the various modules.

cumulus/parachains/runtimes/contracts/contracts-rococo/src/lib.rs

@@ -104,7 +104,8 @@ pub type Migrations = (
 	cumulus_pallet_xcmp_queue::migration::v3::MigrationToV3<Runtime>,
 	pallet_contracts::Migration<Runtime>,
 	// unreleased
-	cumulus_pallet_xcmp_queue::migration::v4::MigrationToV4<Runtime>,
+	cumulus_pallet_xcmp_queue::migration::v4::MigrateV3ToV4<Runtime>,
+	cumulus_pallet_xcmp_queue::migration::v5::MigrateV4ToV5<Runtime>,
 );
 
 type EventRecord = frame_system::EventRecord<

cumulus/parachains/runtimes/contracts/contracts-rococo/src/xcm_config.rs

@@ -281,7 +281,9 @@ impl cumulus_pallet_xcmp_queue::Config for Runtime {
 		cumulus_primitives_core::ParaId,
 		parachains_common::message_queue::ParaIdToSibling,
 	>;
-	type MaxInboundSuspended = sp_core::ConstU32<1_000>;
+	type MaxInboundSuspended = ConstU32<1_000>;
+	type MaxActiveOutboundChannels = ConstU32<128>;
+	type MaxPageSize = ConstU32<{ 1 << 16 }>;
 	type ControllerOrigin = EitherOfDiverse<
 		EnsureRoot<AccountId>,
 		EnsureXcm<IsMajorityOfBody<RelayLocation, ExecutiveBody>>,
@@ -291,6 +293,11 @@ impl cumulus_pallet_xcmp_queue::Config for Runtime {
 	type PriceForSiblingDelivery = PriceForSiblingParachainDelivery;
 }
 
+impl cumulus_pallet_xcmp_queue::migration::v5::V5Config for Runtime {
+	// This must be the same as the `ChannelInfo` from the `Config`:
+	type ChannelList = ParachainSystem;
+}
+
 parameter_types! {
 	pub const RelayOrigin: AggregateMessageOrigin = AggregateMessageOrigin::Parent;
 }

cumulus/parachains/runtimes/coretime/coretime-rococo/src/lib.rs

@@ -106,7 +106,10 @@ pub type UncheckedExtrinsic =
 	generic::UncheckedExtrinsic<Address, RuntimeCall, Signature, SignedExtra>;
 
 /// Migrations to apply on runtime upgrade.
-pub type Migrations = (cumulus_pallet_xcmp_queue::migration::v4::MigrationToV4<Runtime>,);
+pub type Migrations = (
+	cumulus_pallet_xcmp_queue::migration::v4::MigrateV3ToV4<Runtime>,
+	cumulus_pallet_xcmp_queue::migration::v5::MigrateV4ToV5<Runtime>,
+);
 
 /// Executive: handles dispatch to the various modules.
 pub type Executive = frame_executive::Executive<
@@ -332,13 +335,20 @@ impl cumulus_pallet_xcmp_queue::Config for Runtime {
 	type ChannelInfo = ParachainSystem;
 	type VersionWrapper = PolkadotXcm;
 	type XcmpQueue = TransformOrigin<MessageQueue, AggregateMessageOrigin, ParaId, ParaIdToSibling>;
-	type MaxInboundSuspended = sp_core::ConstU32<1_000>;
+	type MaxInboundSuspended = ConstU32<1_000>;
+	type MaxActiveOutboundChannels = ConstU32<128>;
+	type MaxPageSize = ConstU32<{ 1 << 16 }>;
 	type ControllerOrigin = RootOrFellows;
 	type ControllerOriginConverter = XcmOriginToTransactDispatchOrigin;
 	type WeightInfo = weights::cumulus_pallet_xcmp_queue::WeightInfo<Runtime>;
 	type PriceForSiblingDelivery = PriceForSiblingParachainDelivery;
 }
 
+impl cumulus_pallet_xcmp_queue::migration::v5::V5Config for Runtime {
+	// This must be the same as the `ChannelInfo` from the `Config`:
+	type ChannelList = ParachainSystem;
+}
+
 pub const PERIOD: u32 = 6 * HOURS;
 pub const OFFSET: u32 = 0;
 

cumulus/parachains/runtimes/coretime/coretime-westend/src/lib.rs

@@ -323,13 +323,20 @@ impl cumulus_pallet_xcmp_queue::Config for Runtime {
 	type ChannelInfo = ParachainSystem;
 	type VersionWrapper = PolkadotXcm;
 	type XcmpQueue = TransformOrigin<MessageQueue, AggregateMessageOrigin, ParaId, ParaIdToSibling>;
-	type MaxInboundSuspended = sp_core::ConstU32<1_000>;
+	type MaxInboundSuspended = ConstU32<1_000>;
+	type MaxActiveOutboundChannels = ConstU32<128>;
+	type MaxPageSize = ConstU32<{ 1 << 16 }>;
 	type ControllerOrigin = RootOrFellows;
 	type ControllerOriginConverter = XcmOriginToTransactDispatchOrigin;
 	type WeightInfo = weights::cumulus_pallet_xcmp_queue::WeightInfo<Runtime>;
 	type PriceForSiblingDelivery = PriceForSiblingParachainDelivery;
 }
 
+impl cumulus_pallet_xcmp_queue::migration::v5::V5Config for Runtime {
+	// This must be the same as the `ChannelInfo` from the `Config`:
+	type ChannelList = ParachainSystem;
+}
+
 pub const PERIOD: u32 = 6 * HOURS;
 pub const OFFSET: u32 = 0;
 

cumulus/parachains/runtimes/people/people-rococo/src/lib.rs

@@ -301,13 +301,20 @@ impl cumulus_pallet_xcmp_queue::Config for Runtime {
 	type ChannelInfo = ParachainSystem;
 	type VersionWrapper = PolkadotXcm;
 	type XcmpQueue = TransformOrigin<MessageQueue, AggregateMessageOrigin, ParaId, ParaIdToSibling>;
-	type MaxInboundSuspended = sp_core::ConstU32<1_000>;
+	type MaxInboundSuspended = ConstU32<1_000>;
+	type MaxActiveOutboundChannels = ConstU32<128>;
+	type MaxPageSize = ConstU32<{ 1 << 16 }>;
 	type ControllerOrigin = RootOrFellows;
 	type ControllerOriginConverter = XcmOriginToTransactDispatchOrigin;
 	type PriceForSiblingDelivery = PriceForSiblingParachainDelivery;
 	type WeightInfo = weights::cumulus_pallet_xcmp_queue::WeightInfo<Runtime>;
 }
 
+impl cumulus_pallet_xcmp_queue::migration::v5::V5Config for Runtime {
+	// This must be the same as the `ChannelInfo` from the `Config`:
+	type ChannelList = ParachainSystem;
+}
+
 pub const PERIOD: u32 = 6 * HOURS;
 pub const OFFSET: u32 = 0;
 

cumulus/parachains/runtimes/people/people-westend/src/lib.rs

@@ -301,13 +301,20 @@ impl cumulus_pallet_xcmp_queue::Config for Runtime {
 	type ChannelInfo = ParachainSystem;
 	type VersionWrapper = PolkadotXcm;
 	type XcmpQueue = TransformOrigin<MessageQueue, AggregateMessageOrigin, ParaId, ParaIdToSibling>;
-	type MaxInboundSuspended = sp_core::ConstU32<1_000>;
+	type MaxInboundSuspended = ConstU32<1_000>;
+	type MaxActiveOutboundChannels = ConstU32<128>;
+	type MaxPageSize = ConstU32<{ 1 << 16 }>;
 	type ControllerOrigin = RootOrFellows;
 	type ControllerOriginConverter = XcmOriginToTransactDispatchOrigin;
 	type WeightInfo = weights::cumulus_pallet_xcmp_queue::WeightInfo<Runtime>;
 	type PriceForSiblingDelivery = PriceForSiblingParachainDelivery;
 }
 
+impl cumulus_pallet_xcmp_queue::migration::v5::V5Config for Runtime {
+	// This must be the same as the `ChannelInfo` from the `Config`:
+	type ChannelList = ParachainSystem;
+}
+
 pub const PERIOD: u32 = 6 * HOURS;
 pub const OFFSET: u32 = 0;
 

cumulus/parachains/runtimes/testing/penpal/src/lib.rs

@@ -549,7 +549,9 @@ impl cumulus_pallet_xcmp_queue::Config for Runtime {
 	type VersionWrapper = PolkadotXcm;
 	// Enqueue XCMP messages from siblings for later processing.
 	type XcmpQueue = TransformOrigin<MessageQueue, AggregateMessageOrigin, ParaId, ParaIdToSibling>;
-	type MaxInboundSuspended = sp_core::ConstU32<1_000>;
+	type MaxInboundSuspended = ConstU32<1_000>;
+	type MaxActiveOutboundChannels = ConstU32<128>;
+	type MaxPageSize = ConstU32<{ 1 << 16 }>;
 	type ControllerOrigin = EnsureRoot<AccountId>;
 	type ControllerOriginConverter = XcmOriginToTransactDispatchOrigin;
 	type WeightInfo = ();

cumulus/parachains/runtimes/testing/rococo-parachain/src/lib.rs

@@ -542,7 +542,9 @@ impl cumulus_pallet_xcmp_queue::Config for Runtime {
 	type VersionWrapper = ();
 	// Enqueue XCMP messages from siblings for later processing.
 	type XcmpQueue = TransformOrigin<MessageQueue, AggregateMessageOrigin, ParaId, ParaIdToSibling>;
-	type MaxInboundSuspended = sp_core::ConstU32<1_000>;
+	type MaxInboundSuspended = ConstU32<1_000>;
+	type MaxActiveOutboundChannels = ConstU32<128>;
+	type MaxPageSize = ConstU32<{ 1 << 16 }>;
 	type ControllerOrigin = EnsureRoot<AccountId>;
 	type ControllerOriginConverter = XcmOriginToTransactDispatchOrigin;
 	type WeightInfo = cumulus_pallet_xcmp_queue::weights::SubstrateWeight<Runtime>;

cumulus/primitives/core/src/lib.rs

@@ -64,6 +64,8 @@ pub enum MessageSendError {
 	TooBig,
 	/// Some other error.
 	Other,
+	/// There are too many channels open at once.
+	TooManyChannels,
 }
 
 impl From<MessageSendError> for &'static str {
@@ -74,6 +76,7 @@ impl From<MessageSendError> for &'static str {
 			NoChannel => "NoChannel",
 			TooBig => "TooBig",
 			Other => "Other",
+			TooManyChannels => "TooManyChannels",
 		}
 	}
 }
@@ -135,6 +138,11 @@ pub trait GetChannelInfo {
 	fn get_channel_info(id: ParaId) -> Option<ChannelInfo>;
 }
 
+/// List all open outgoing channels.
+pub trait ListChannelInfos {
+	fn outgoing_channels() -> Vec<ParaId>;
+}
+
 /// Something that should be called when sending an upward message.
 pub trait UpwardMessageSender {
 	/// Send the given UMP message; return the expected number of blocks before the message will

polkadot/parachain/src/primitives.rs

@@ -333,7 +333,19 @@ impl DmpMessageHandler for () {
 }
 
 /// The aggregate XCMP message format.
-#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Encode, Decode, TypeInfo, RuntimeDebug)]
+#[derive(
+	Copy,
+	Clone,
+	Eq,
+	PartialEq,
+	Ord,
+	PartialOrd,
+	Encode,
+	Decode,
+	TypeInfo,
+	RuntimeDebug,
+	MaxEncodedLen,
+)]
 pub enum XcmpMessageFormat {
 	/// Encoded `VersionedXcm` messages, all concatenated.
 	ConcatenatedVersionedXcm,

prdoc/pr_2302.prdoc

@@ -0,0 +1,20 @@
+title: Storage bound the XCMP queue pallet
+
+doc:
+  - audience: Runtime Dev
+    description: |
+      Enforce upper limits for the number of suspended channels and on
+      the number of channels that have messages pending. This is needed to use
+      the worst-case PoV benchmarking.
+
+migrations:
+  db: []
+
+  runtime:
+    - reference: cumulus_pallet_xcmp_queue::migration::v5::MigrateV4ToV5
+      description: A No-OP migration is deployed to ensure that all `BoundedVec`s` still decode as expected.
+
+crates:
+  - name: cumulus-pallet-xcmp-queue
+
+host_functions: []