UMP: Support Overweight messages (#3575)

* Introduce new config: ump_max_individual_weight

* Implement overweight msg stashing

* Test

* Add migration module.

Also introduces a test for migration

* Integrate ExecuteOverweightOrigin to runtimes

* Fix more stuff

* Add `yeet` into dictionary

* Use suggested `Error` variant names

* typo

* Use 20ms as the maximum individual message weight

* Update the test value

* rustfmt

* Clean up

* Remove deprecated field from host config

* Remove missed _hrmp_open_request_ttl

* Apply typo fix suggestion

Co-authored-by: Alexander Popiak <alexander.popiak@parity.io>

* Rename `migration::migrate_to_latest`

* Restore `_hrmp_open_request_ttl` in `v0::HostConfiguration`

* Apply suggestion for a rustdoc

* Apply the suggestion

* Test v0 config with the raw production data fetched from Kusama

* Update runtime/parachains/src/ump.rs

Co-authored-by: Alexander Popiak <alexander.popiak@parity.io>

* Expose migration functions

* Fix spellcheck

Co-authored-by: Alexander Popiak <alexander.popiak@parity.io>
Co-authored-by: Shawn Tabrizi <shawntabrizi@gmail.com>
Co-authored-by: Keith Yeung <kungfukeith11@gmail.com>

polkadot/node/service/src/chain_spec.rs

@@ -185,7 +185,6 @@ fn default_parachains_host_configuration(
 		ump_service_total_weight: 4 * 1_000_000_000,
 		max_upward_message_size: 1024 * 1024,
 		max_upward_message_num_per_candidate: 5,
-		_hrmp_open_request_ttl: 5,
 		hrmp_sender_deposit: 0,
 		hrmp_recipient_deposit: 0,
 		hrmp_channel_max_capacity: 8,

polkadot/runtime/kusama/src/lib.rs

@@ -1118,6 +1118,7 @@ impl parachains_ump::Config for Runtime {
 	type Event = Event;
 	type UmpSink = crate::parachains_ump::XcmSink<XcmExecutor<XcmConfig>, Runtime>;
 	type FirstMessageFactorPercent = FirstMessageFactorPercent;
+	type ExecuteOverweightOrigin = EnsureRoot<AccountId>;
 }
 
 impl parachains_dmp::Config for Runtime {}

polkadot/runtime/parachains/src/configuration.rs

@@ -19,7 +19,7 @@
 //! Configuration can change only at session boundaries and is buffered until then.
 
 use crate::shared;
-use frame_support::pallet_prelude::*;
+use frame_support::{pallet_prelude::*, weights::constants::WEIGHT_PER_MILLIS};
 use frame_system::pallet_prelude::*;
 use parity_scale_codec::{Decode, Encode};
 use primitives::v1::{Balance, SessionIndex, MAX_CODE_SIZE, MAX_POV_SIZE};
@@ -28,6 +28,10 @@ use sp_std::prelude::*;
 
 pub use pallet::*;
 
+pub mod migration;
+
+const LOG_TARGET: &str = "runtime::configuration";
+
 /// All configuration of the runtime with respect to parachains and parathreads.
 #[derive(Clone, Encode, Decode, PartialEq, sp_core::RuntimeDebug)]
 #[cfg_attr(feature = "std", derive(serde::Serialize, serde::Deserialize))]
@@ -91,10 +95,6 @@ pub struct HostConfiguration<BlockNumber> {
 	pub hrmp_max_parachain_outbound_channels: u32,
 	/// The maximum number of outbound HRMP channels a parathread is allowed to open.
 	pub hrmp_max_parathread_outbound_channels: u32,
-	/// NOTE: this field is deprecated. Channel open requests became non-expiring. Changing this value
-	/// doesn't have any effect. This field doesn't have a `deprecated` attribute because that would
-	/// trigger warnings coming from macros.
-	pub _hrmp_open_request_ttl: u32,
 	/// The deposit that the sender should provide for opening an HRMP channel.
 	pub hrmp_sender_deposit: Balance,
 	/// The deposit that the recipient should provide for accepting opening an HRMP channel.
@@ -170,6 +170,9 @@ pub struct HostConfiguration<BlockNumber> {
 	pub needed_approvals: u32,
 	/// The number of samples to do of the `RelayVRFModulo` approval assignment criterion.
 	pub relay_vrf_modulo_samples: u32,
+	/// The maximum amount of weight any individual upward message may consume. Messages above this
+	/// weight go into the overweight queue and may only be serviced explicitly.
+	pub ump_max_individual_weight: Weight,
 }
 
 impl<BlockNumber: Default + From<u32>> Default for HostConfiguration<BlockNumber> {
@@ -204,7 +207,6 @@ impl<BlockNumber: Default + From<u32>> Default for HostConfiguration<BlockNumber
 			ump_service_total_weight: Default::default(),
 			max_upward_message_size: Default::default(),
 			max_upward_message_num_per_candidate: Default::default(),
-			_hrmp_open_request_ttl: Default::default(),
 			hrmp_sender_deposit: Default::default(),
 			hrmp_recipient_deposit: Default::default(),
 			hrmp_channel_max_capacity: Default::default(),
@@ -215,6 +217,7 @@ impl<BlockNumber: Default + From<u32>> Default for HostConfiguration<BlockNumber
 			hrmp_max_parachain_outbound_channels: Default::default(),
 			hrmp_max_parathread_outbound_channels: Default::default(),
 			hrmp_max_message_num_per_candidate: Default::default(),
+			ump_max_individual_weight: 20 * WEIGHT_PER_MILLIS,
 		}
 	}
 }
@@ -261,6 +264,7 @@ pub mod pallet {
 
 	#[pallet::pallet]
 	#[pallet::generate_store(pub(super) trait Store)]
+	#[pallet::storage_version(migration::STORAGE_VERSION)]
 	pub struct Pallet<T>(_);
 
 	#[pallet::config]
@@ -764,10 +768,24 @@ pub mod pallet {
 			});
 			Ok(())
 		}
+
+		/// Sets the maximum amount of weight any individual upward message may consume.
+		#[pallet::weight((1_000, DispatchClass::Operational))]
+		pub fn set_ump_max_individual_weight(origin: OriginFor<T>, new: Weight) -> DispatchResult {
+			ensure_root(origin)?;
+			Self::update_config_member(|config| {
+				sp_std::mem::replace(&mut config.ump_max_individual_weight, new) != new
+			});
+			Ok(())
+		}
 	}
 
 	#[pallet::hooks]
 	impl<T: Config> Hooks<BlockNumberFor<T>> for Pallet<T> {
+		fn on_runtime_upgrade() -> Weight {
+			migration::migrate_to_latest::<T>()
+		}
+
 		fn integrity_test() {
 			assert_eq!(
 				&ActiveConfig::<T>::hashed_key(),
@@ -888,7 +906,6 @@ mod tests {
 				ump_service_total_weight: 20000,
 				max_upward_message_size: 448,
 				max_upward_message_num_per_candidate: 5,
-				_hrmp_open_request_ttl: 0,
 				hrmp_sender_deposit: 22,
 				hrmp_recipient_deposit: 4905,
 				hrmp_channel_max_capacity: 3921,
@@ -899,6 +916,7 @@ mod tests {
 				hrmp_max_parachain_outbound_channels: 100,
 				hrmp_max_parathread_outbound_channels: 200,
 				hrmp_max_message_num_per_candidate: 20,
+				ump_max_individual_weight: 909,
 			};
 
 			assert!(<Configuration as Store>::PendingConfig::get(shared::SESSION_DELAY).is_none());
@@ -1060,6 +1078,11 @@ mod tests {
 				new_config.hrmp_max_message_num_per_candidate,
 			)
 			.unwrap();
+			Configuration::set_ump_max_individual_weight(
+				Origin::root(),
+				new_config.ump_max_individual_weight,
+			)
+			.unwrap();
 
 			assert_eq!(
 				<Configuration as Store>::PendingConfig::get(shared::SESSION_DELAY),

polkadot/runtime/parachains/src/configuration/migration.rs

@@ -0,0 +1,314 @@
+// Copyright 2021 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/>.
+
+//! A module that is responsible for migration of storage.
+
+use crate::configuration::{self, Config, Pallet, Store};
+use frame_support::{pallet_prelude::*, traits::StorageVersion, weights::Weight};
+use frame_system::pallet_prelude::BlockNumberFor;
+
+/// The current storage version.
+pub const STORAGE_VERSION: StorageVersion = StorageVersion::new(1);
+
+/// Migrates the pallet storage to the most recent version, checking and setting the `StorageVersion`.
+pub fn migrate_to_latest<T: Config>() -> Weight {
+	let mut weight = 0;
+
+	if StorageVersion::get::<Pallet<T>>() == 0 {
+		weight += migrate_to_v1::<T>();
+		StorageVersion::new(1).put::<Pallet<T>>();
+	}
+
+	weight
+}
+
+mod v0 {
+	use super::*;
+	use primitives::v1::{Balance, SessionIndex};
+
+	#[derive(parity_scale_codec::Encode, parity_scale_codec::Decode, Debug)]
+	pub struct HostConfiguration<BlockNumber> {
+		pub max_code_size: u32,
+		pub max_head_data_size: u32,
+		pub max_upward_queue_count: u32,
+		pub max_upward_queue_size: u32,
+		pub max_upward_message_size: u32,
+		pub max_upward_message_num_per_candidate: u32,
+		pub hrmp_max_message_num_per_candidate: u32,
+		pub validation_upgrade_frequency: BlockNumber,
+		pub validation_upgrade_delay: BlockNumber,
+		pub max_pov_size: u32,
+		pub max_downward_message_size: u32,
+		pub ump_service_total_weight: Weight,
+		pub hrmp_max_parachain_outbound_channels: u32,
+		pub hrmp_max_parathread_outbound_channels: u32,
+		pub _hrmp_open_request_ttl: u32,
+		pub hrmp_sender_deposit: Balance,
+		pub hrmp_recipient_deposit: Balance,
+		pub hrmp_channel_max_capacity: u32,
+		pub hrmp_channel_max_total_size: u32,
+		pub hrmp_max_parachain_inbound_channels: u32,
+		pub hrmp_max_parathread_inbound_channels: u32,
+		pub hrmp_channel_max_message_size: u32,
+		pub code_retention_period: BlockNumber,
+		pub parathread_cores: u32,
+		pub parathread_retries: u32,
+		pub group_rotation_frequency: BlockNumber,
+		pub chain_availability_period: BlockNumber,
+		pub thread_availability_period: BlockNumber,
+		pub scheduling_lookahead: u32,
+		pub max_validators_per_core: Option<u32>,
+		pub max_validators: Option<u32>,
+		pub dispute_period: SessionIndex,
+		pub dispute_post_conclusion_acceptance_period: BlockNumber,
+		pub dispute_max_spam_slots: u32,
+		pub dispute_conclusion_by_time_out_period: BlockNumber,
+		pub no_show_slots: u32,
+		pub n_delay_tranches: u32,
+		pub zeroth_delay_tranche_width: u32,
+		pub needed_approvals: u32,
+		pub relay_vrf_modulo_samples: u32,
+	}
+
+	impl<BlockNumber: Default + From<u32>> Default for HostConfiguration<BlockNumber> {
+		fn default() -> Self {
+			HostConfiguration {
+				group_rotation_frequency: 1u32.into(),
+				chain_availability_period: 1u32.into(),
+				thread_availability_period: 1u32.into(),
+				no_show_slots: 1u32.into(),
+				validation_upgrade_frequency: Default::default(),
+				validation_upgrade_delay: Default::default(),
+				code_retention_period: Default::default(),
+				max_code_size: Default::default(),
+				max_pov_size: Default::default(),
+				max_head_data_size: Default::default(),
+				parathread_cores: Default::default(),
+				parathread_retries: Default::default(),
+				scheduling_lookahead: Default::default(),
+				max_validators_per_core: Default::default(),
+				max_validators: None,
+				dispute_period: 6,
+				dispute_post_conclusion_acceptance_period: 100.into(),
+				dispute_max_spam_slots: 2,
+				dispute_conclusion_by_time_out_period: 200.into(),
+				n_delay_tranches: Default::default(),
+				zeroth_delay_tranche_width: Default::default(),
+				needed_approvals: Default::default(),
+				relay_vrf_modulo_samples: Default::default(),
+				max_upward_queue_count: Default::default(),
+				max_upward_queue_size: Default::default(),
+				max_downward_message_size: Default::default(),
+				ump_service_total_weight: Default::default(),
+				max_upward_message_size: Default::default(),
+				max_upward_message_num_per_candidate: Default::default(),
+				_hrmp_open_request_ttl: Default::default(),
+				hrmp_sender_deposit: Default::default(),
+				hrmp_recipient_deposit: Default::default(),
+				hrmp_channel_max_capacity: Default::default(),
+				hrmp_channel_max_total_size: Default::default(),
+				hrmp_max_parachain_inbound_channels: Default::default(),
+				hrmp_max_parathread_inbound_channels: Default::default(),
+				hrmp_channel_max_message_size: Default::default(),
+				hrmp_max_parachain_outbound_channels: Default::default(),
+				hrmp_max_parathread_outbound_channels: Default::default(),
+				hrmp_max_message_num_per_candidate: Default::default(),
+			}
+		}
+	}
+}
+
+/// Migrates the `HostConfiguration` from v0 (with deprecated `hrmp_open_request_ttl` and without
+/// `ump_max_individual_weight`) to v1 (without HRMP TTL and with max individual weight).
+/// Uses the `Default` implementation of `HostConfiguration` to choose a value for `ump_max_individual_weight`.
+///
+/// NOTE: Only use this function if you know what you are doing. Default to using `migrate_to_latest`.
+pub fn migrate_to_v1<T: Config>() -> Weight {
+	// Unusual formatting is justified:
+	// - make it easier to verify that fields assign what they supposed to assign.
+	// - this code is transient and will be removed after all migrations are done.
+	// - this code is important enough to optimize for legibility sacrificing consistency.
+	#[rustfmt::skip]
+	let translate =
+		|pre: v0::HostConfiguration<BlockNumberFor<T>>| -> configuration::HostConfiguration<BlockNumberFor<T>>
+	{
+		super::HostConfiguration {
+
+max_code_size                            : pre.max_code_size,
+max_head_data_size                       : pre.max_head_data_size,
+max_upward_queue_count                   : pre.max_upward_queue_count,
+max_upward_queue_size                    : pre.max_upward_queue_size,
+max_upward_message_size                  : pre.max_upward_message_size,
+max_upward_message_num_per_candidate     : pre.max_upward_message_num_per_candidate,
+hrmp_max_message_num_per_candidate       : pre.hrmp_max_message_num_per_candidate,
+validation_upgrade_frequency             : pre.validation_upgrade_frequency,
+validation_upgrade_delay                 : pre.validation_upgrade_delay,
+max_pov_size                             : pre.max_pov_size,
+max_downward_message_size                : pre.max_downward_message_size,
+ump_service_total_weight                 : pre.ump_service_total_weight,
+hrmp_max_parachain_outbound_channels     : pre.hrmp_max_parachain_outbound_channels,
+hrmp_max_parathread_outbound_channels    : pre.hrmp_max_parathread_outbound_channels,
+hrmp_sender_deposit                      : pre.hrmp_sender_deposit,
+hrmp_recipient_deposit                   : pre.hrmp_recipient_deposit,
+hrmp_channel_max_capacity                : pre.hrmp_channel_max_capacity,
+hrmp_channel_max_total_size              : pre.hrmp_channel_max_total_size,
+hrmp_max_parachain_inbound_channels      : pre.hrmp_max_parachain_inbound_channels,
+hrmp_max_parathread_inbound_channels     : pre.hrmp_max_parathread_inbound_channels,
+hrmp_channel_max_message_size            : pre.hrmp_channel_max_message_size,
+code_retention_period                    : pre.code_retention_period,
+parathread_cores                         : pre.parathread_cores,
+parathread_retries                       : pre.parathread_retries,
+group_rotation_frequency                 : pre.group_rotation_frequency,
+chain_availability_period                : pre.chain_availability_period,
+thread_availability_period               : pre.thread_availability_period,
+scheduling_lookahead                     : pre.scheduling_lookahead,
+max_validators_per_core                  : pre.max_validators_per_core,
+max_validators                           : pre.max_validators,
+dispute_period                           : pre.dispute_period,
+dispute_post_conclusion_acceptance_period: pre.dispute_post_conclusion_acceptance_period,
+dispute_max_spam_slots                   : pre.dispute_max_spam_slots,
+dispute_conclusion_by_time_out_period    : pre.dispute_conclusion_by_time_out_period,
+no_show_slots                            : pre.no_show_slots,
+n_delay_tranches                         : pre.n_delay_tranches,
+zeroth_delay_tranche_width               : pre.zeroth_delay_tranche_width,
+needed_approvals                         : pre.needed_approvals,
+relay_vrf_modulo_samples                 : pre.relay_vrf_modulo_samples,
+
+ump_max_individual_weight: <configuration::HostConfiguration<BlockNumberFor<T>>>::default().ump_max_individual_weight,
+		}
+	};
+
+	if let Err(_) = <Pallet<T> as Store>::ActiveConfig::translate(|pre| pre.map(translate)) {
+		// `Err` is returned when the pre-migration type cannot be deserialized. This
+		// cannot happen if the migration runs correctly, i.e. against the expected version.
+		//
+		// This happening almost surely will lead to a panic somewhere else. Corruption seems
+		// to be unlikely to be caused by this. So we just log. Maybe it'll work out still?
+		log::error!(
+			target: configuration::LOG_TARGET,
+			"unexpected error when performing translation of the configuration type during storage upgrade to v1."
+		);
+	}
+
+	T::DbWeight::get().reads_writes(1, 1)
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::mock::{new_test_ext, Test};
+
+	#[test]
+	fn v0_deserialized_from_actual_data() {
+		// Fetched at Kusama 9,207,703 (0xbfe5227324c08b3ab67e0473a360acbce43efbd7b42041d0033adaf9ff2c5330)
+		//
+		// This exceeds the maximal line width length, but that's fine, since this is not code and
+		// doesn't need to be read and also leaving it as one line allows to easily copy it.
+		let raw_config = hex_literal::hex!["0000a000005000000a00000000c8000000c800000a0000000a00000040380000580200000000500000c8000000e87648170000000a0000000000000048000000c09e5d9a2f3d00000000000000000000c09e5d9a2f3d00000000000000000000e8030000009001000a00000000000000009001008070000000000000000000000a0000000a0000000a00000001000000010500000001c8000000060000005802000002000000580200000200000059000000000000001e00000028000000"];
+
+		let v0 = v0::HostConfiguration::<primitives::v1::BlockNumber>::decode(&mut &raw_config[..])
+			.unwrap();
+
+		// We check only a sample of the values here. If we missed any fields or messed up data types
+		// that would skew all the fields coming after.
+		assert_eq!(v0.max_code_size, 10_485_760);
+		assert_eq!(v0.validation_upgrade_frequency, 14_400);
+		assert_eq!(v0.max_pov_size, 5_242_880);
+		assert_eq!(v0._hrmp_open_request_ttl, 72);
+		assert_eq!(v0.hrmp_channel_max_message_size, 102_400);
+		assert_eq!(v0.dispute_max_spam_slots, 2);
+		assert_eq!(v0.n_delay_tranches, 89);
+		assert_eq!(v0.relay_vrf_modulo_samples, 40);
+	}
+
+	#[test]
+	fn test_migrate_to_v1() {
+		// Host configuration has lots of fields. However, in this migration we add one and remove one
+		// field. The most important part to check are a couple of the last fields. We also pick
+		// extra fields to check arbitrarily, e.g. depending on their position (i.e. the middle) and
+		// also their type.
+		//
+		// We specify only the picked fields and the rest should be provided by the `Default`
+		// implementation. That implementation is copied over between the two types and should work
+		// fine.
+		let v0 = v0::HostConfiguration::<primitives::v1::BlockNumber> {
+			relay_vrf_modulo_samples: 0xFEEDBEEFu32,
+			needed_approvals: 69,
+			thread_availability_period: 55,
+			hrmp_recipient_deposit: 1337,
+			max_pov_size: 1111,
+			..Default::default()
+		};
+
+		new_test_ext(Default::default()).execute_with(|| {
+			// Implant the v0 version in the state.
+			frame_support::storage::unhashed::put_raw(
+				&configuration::ActiveConfig::<Test>::hashed_key(),
+				&v0.encode(),
+			);
+
+			migrate_to_v1::<Test>();
+
+			let v1 = configuration::ActiveConfig::<Test>::get();
+
+			// The same motivation as for the migration code. See `migrate_to_v1`.
+			#[rustfmt::skip]
+			{
+				assert_eq!(v0.max_code_size                            , v1.max_code_size);
+				assert_eq!(v0.max_head_data_size                       , v1.max_head_data_size);
+				assert_eq!(v0.max_upward_queue_count                   , v1.max_upward_queue_count);
+				assert_eq!(v0.max_upward_queue_size                    , v1.max_upward_queue_size);
+				assert_eq!(v0.max_upward_message_size                  , v1.max_upward_message_size);
+				assert_eq!(v0.max_upward_message_num_per_candidate     , v1.max_upward_message_num_per_candidate);
+				assert_eq!(v0.hrmp_max_message_num_per_candidate       , v1.hrmp_max_message_num_per_candidate);
+				assert_eq!(v0.validation_upgrade_frequency             , v1.validation_upgrade_frequency);
+				assert_eq!(v0.validation_upgrade_delay                 , v1.validation_upgrade_delay);
+				assert_eq!(v0.max_pov_size                             , v1.max_pov_size);
+				assert_eq!(v0.max_downward_message_size                , v1.max_downward_message_size);
+				assert_eq!(v0.ump_service_total_weight                 , v1.ump_service_total_weight);
+				assert_eq!(v0.hrmp_max_parachain_outbound_channels     , v1.hrmp_max_parachain_outbound_channels);
+				assert_eq!(v0.hrmp_max_parathread_outbound_channels    , v1.hrmp_max_parathread_outbound_channels);
+				assert_eq!(v0.hrmp_sender_deposit                      , v1.hrmp_sender_deposit);
+				assert_eq!(v0.hrmp_recipient_deposit                   , v1.hrmp_recipient_deposit);
+				assert_eq!(v0.hrmp_channel_max_capacity                , v1.hrmp_channel_max_capacity);
+				assert_eq!(v0.hrmp_channel_max_total_size              , v1.hrmp_channel_max_total_size);
+				assert_eq!(v0.hrmp_max_parachain_inbound_channels      , v1.hrmp_max_parachain_inbound_channels);
+				assert_eq!(v0.hrmp_max_parathread_inbound_channels     , v1.hrmp_max_parathread_inbound_channels);
+				assert_eq!(v0.hrmp_channel_max_message_size            , v1.hrmp_channel_max_message_size);
+				assert_eq!(v0.code_retention_period                    , v1.code_retention_period);
+				assert_eq!(v0.parathread_cores                         , v1.parathread_cores);
+				assert_eq!(v0.parathread_retries                       , v1.parathread_retries);
+				assert_eq!(v0.group_rotation_frequency                 , v1.group_rotation_frequency);
+				assert_eq!(v0.chain_availability_period                , v1.chain_availability_period);
+				assert_eq!(v0.thread_availability_period               , v1.thread_availability_period);
+				assert_eq!(v0.scheduling_lookahead                     , v1.scheduling_lookahead);
+				assert_eq!(v0.max_validators_per_core                  , v1.max_validators_per_core);
+				assert_eq!(v0.max_validators                           , v1.max_validators);
+				assert_eq!(v0.dispute_period                           , v1.dispute_period);
+				assert_eq!(v0.dispute_post_conclusion_acceptance_period, v1.dispute_post_conclusion_acceptance_period);
+				assert_eq!(v0.dispute_max_spam_slots                   , v1.dispute_max_spam_slots);
+				assert_eq!(v0.dispute_conclusion_by_time_out_period    , v1.dispute_conclusion_by_time_out_period);
+				assert_eq!(v0.no_show_slots                            , v1.no_show_slots);
+				assert_eq!(v0.n_delay_tranches                         , v1.n_delay_tranches);
+				assert_eq!(v0.zeroth_delay_tranche_width               , v1.zeroth_delay_tranche_width);
+				assert_eq!(v0.needed_approvals                         , v1.needed_approvals);
+				assert_eq!(v0.relay_vrf_modulo_samples                 , v1.relay_vrf_modulo_samples);
+
+				assert_eq!(v1.ump_max_individual_weight, 20_000_000_000);
+			}; // ; makes this a statement. `rustfmt::skip` cannot be put on an expression.
+		});
+	}
+}

polkadot/runtime/parachains/src/mock.rs

@@ -133,6 +133,7 @@ impl crate::ump::Config for Test {
 	type Event = Event;
 	type UmpSink = TestUmpSink;
 	type FirstMessageFactorPercent = FirstMessageFactorPercent;
+	type ExecuteOverweightOrigin = frame_system::EnsureRoot<AccountId>;
 }
 
 impl crate::hrmp::Config for Test {
@@ -309,3 +310,11 @@ pub struct MockGenesisConfig {
 	pub configuration: crate::configuration::GenesisConfig<Test>,
 	pub paras: crate::paras::GenesisConfig,
 }
+
+pub fn assert_last_event(generic_event: Event) {
+	let events = frame_system::Pallet::<Test>::events();
+	let system_event: <Test as frame_system::Config>::Event = generic_event.into();
+	// compare to the last event record
+	let frame_system::EventRecord { event, .. } = &events[events.len() - 1];
+	assert_eq!(event, &system_event);
+}

polkadot/runtime/parachains/src/ump.rs

@@ -18,10 +18,11 @@ use crate::{
 	configuration::{self, HostConfiguration},
 	initializer,
 };
-use frame_support::pallet_prelude::*;
+use frame_support::{pallet_prelude::*, traits::EnsureOrigin};
+use frame_system::pallet_prelude::*;
 use primitives::v1::{Id as ParaId, UpwardMessage};
 use sp_std::{
-	collections::btree_map::BTreeMap, convert::TryFrom, fmt, marker::PhantomData, prelude::*,
+	collections::btree_map::BTreeMap, convert::TryFrom, fmt, marker::PhantomData, mem, prelude::*,
 };
 use xcm::latest::Outcome;
 
@@ -70,10 +71,18 @@ impl UmpSink for () {
 /// if the message content is unique.
 pub type MessageId = [u8; 32];
 
+/// Index used to identify overweight messages.
+pub type OverweightIndex = u64;
+
 /// A specific implementation of a `UmpSink` where messages are in the XCM format
 /// and will be forwarded to the XCM Executor.
 pub struct XcmSink<XcmExecutor, Config>(PhantomData<(XcmExecutor, Config)>);
 
+/// Returns a [`MessageId`] for the given upward message payload.
+fn upward_message_id(data: &[u8]) -> MessageId {
+	sp_io::hashing::blake2_256(data)
+}
+
 impl<XcmExecutor: xcm::latest::ExecuteXcm<C::Call>, C: Config> UmpSink for XcmSink<XcmExecutor, C> {
 	fn process_upward_message(
 		origin: ParaId,
@@ -86,7 +95,7 @@ impl<XcmExecutor: xcm::latest::ExecuteXcm<C::Call>, C: Config> UmpSink for XcmSi
 			VersionedXcm,
 		};
 
-		let id = sp_io::hashing::blake2_256(&data[..]);
+		let id = upward_message_id(&data[..]);
 		let maybe_msg = VersionedXcm::<C::Call>::decode_all_with_depth_limit(
 			xcm::MAX_XCM_DECODE_DEPTH,
 			&mut &data[..],
@@ -177,6 +186,9 @@ pub mod pallet {
 		///
 		/// Generally you'll want this to be a bit more - 150 or 200 would be good values.
 		type FirstMessageFactorPercent: Get<Weight>;
+
+		/// Origin which is allowed to execute overweight messages.
+		type ExecuteOverweightOrigin: EnsureOrigin<Self::Origin>;
 	}
 
 	#[pallet::event]
@@ -197,6 +209,26 @@ pub mod pallet {
 		/// Some downward messages have been received and will be processed.
 		/// \[ para, count, size \]
 		UpwardMessagesReceived(ParaId, u32, u32),
+		/// The weight budget was exceeded for an individual downward message.
+		///
+		/// This message can be later dispatched manually using `service_overweight` dispatchable
+		/// using the assigned `overweight_index`.
+		///
+		/// \[ para, id, overweight_index, required \]
+		OverweightEnqueued(ParaId, MessageId, OverweightIndex, Weight),
+		/// Downward message from the overweight queue was executed with the given actual weight
+		/// used.
+		///
+		/// \[ overweight_index, used \]
+		OverweightServiced(OverweightIndex, Weight),
+	}
+
+	#[pallet::error]
+	pub enum Error<T> {
+		/// The message index given is unknown.
+		UnknownMessageIndex,
+		/// The amount of weight given is possibly not enough for executing the message.
+		WeightOverLimit,
 	}
 
 	/// The messages waiting to be handled by the relay-chain originating from a certain parachain.
@@ -242,8 +274,49 @@ pub mod pallet {
 	#[pallet::storage]
 	pub type NextDispatchRoundStartWith<T: Config> = StorageValue<_, ParaId>;
 
+	/// The messages that exceeded max individual message weight budget.
+	///
+	/// These messages stay there until manually dispatched.
+	#[pallet::storage]
+	pub type Overweight<T: Config> =
+		StorageMap<_, Twox64Concat, OverweightIndex, (ParaId, Vec<u8>), OptionQuery>;
+
+	/// The number of overweight messages ever recorded in `Overweight` (and thus the lowest free
+	/// index).
+	#[pallet::storage]
+	pub type OverweightCount<T: Config> = StorageValue<_, OverweightIndex, ValueQuery>;
+
 	#[pallet::call]
-	impl<T: Config> Pallet<T> {}
+	impl<T: Config> Pallet<T> {
+		/// Service a single overweight upward message.
+		///
+		/// - `origin`: Must pass `ExecuteOverweightOrigin`.
+		/// - `index`: The index of the overweight message to service.
+		/// - `weight_limit`: The amount of weight that message execution may take.
+		///
+		/// Errors:
+		/// - `UnknownMessageIndex`: Message of `index` is unknown.
+		/// - `WeightOverLimit`: Message execution may use greater than `weight_limit`.
+		///
+		/// Events:
+		/// - `OverweightServiced`: On success.
+		#[pallet::weight(weight_limit.saturating_add(1_000_000))]
+		pub fn service_overweight(
+			origin: OriginFor<T>,
+			index: OverweightIndex,
+			weight_limit: Weight,
+		) -> DispatchResultWithPostInfo {
+			T::ExecuteOverweightOrigin::ensure_origin(origin)?;
+
+			let (sender, data) =
+				Overweight::<T>::get(index).ok_or(Error::<T>::UnknownMessageIndex)?;
+			let used = T::UmpSink::process_upward_message(sender, &data[..], weight_limit)
+				.map_err(|_| Error::<T>::WeightOverLimit)?;
+			Overweight::<T>::remove(index);
+			Self::deposit_event(Event::OverweightServiced(index, used));
+			Ok(Some(used.saturating_add(1_000_000)).into())
+		}
+	}
 }
 
 /// Routines related to the upward message passing.
@@ -406,26 +479,36 @@ impl<T: Config> Pallet<T> {
 			// attempt to process the next message from the queue of the dispatchee; if not beyond
 			// our remaining weight limit, then consume it.
 			let maybe_next = queue_cache.peek_front::<T>(dispatchee);
-			let became_empty = if let Some(upward_message) = maybe_next {
+			if let Some(upward_message) = maybe_next {
 				match T::UmpSink::process_upward_message(dispatchee, upward_message, max_weight) {
 					Ok(used) => {
 						weight_used += used;
-						queue_cache.consume_front::<T>(dispatchee)
+						let _ = queue_cache.consume_front::<T>(dispatchee);
 					},
 					Err((id, required)) => {
-						// we process messages in order and don't drop them if we run out of weight,
-						// so need to break here without calling `consume_front`.
-						Self::deposit_event(Event::WeightExhausted(id, max_weight, required));
-						break
+						if required > config.ump_max_individual_weight {
+							// overweight - add to overweight queue and continue with message
+							// execution consuming the message.
+							let upward_message = queue_cache.consume_front::<T>(dispatchee).expect(
+								"`consume_front` should return the same msg as `peek_front`;\
+								if we get into this branch then `peek_front` returned `Some`;\
+								thus `upward_message` cannot be `None`; qed",
+							);
+							let index = Self::stash_overweight(dispatchee, upward_message);
+							Self::deposit_event(Event::OverweightEnqueued(
+								dispatchee, id, index, required,
+							));
+						} else {
+							// we process messages in order and don't drop them if we run out of weight,
+							// so need to break here without calling `consume_front`.
+							Self::deposit_event(Event::WeightExhausted(id, max_weight, required));
+							break
+						}
 					},
 				}
-			} else {
-				// this should never happen, since the cursor should never point to an empty queue.
-				// it is resolved harmlessly here anyway.
-				true
-			};
+			}
 
-			if became_empty {
+			if queue_cache.is_empty::<T>(dispatchee) {
 				// the queue is empty now - this para doesn't need attention anymore.
 				cursor.remove();
 			} else {
@@ -438,6 +521,19 @@ impl<T: Config> Pallet<T> {
 
 		weight_used
 	}
+
+	/// Puts a given upward message into the list of overweight messages allowing it to be executed
+	/// later.
+	fn stash_overweight(sender: ParaId, upward_message: Vec<u8>) -> OverweightIndex {
+		let index = <Self as Store>::OverweightCount::mutate(|count| {
+			let index = *count;
+			*count += 1;
+			index
+		});
+
+		<Self as Store>::Overweight::insert(index, (sender, upward_message));
+		index
+	}
 }
 
 /// To avoid constant fetching, deserializing and serialization the queues are cached.
@@ -490,16 +586,27 @@ impl QueueCache {
 
 	/// Attempts to remove one message from the front of `para`'s queue. If the queue is empty, then
 	/// does nothing.
-	///
-	/// Returns `true` iff there are no more messages in the queue after the removal attempt.
-	fn consume_front<T: Config>(&mut self, para: ParaId) -> bool {
+	fn consume_front<T: Config>(&mut self, para: ParaId) -> Option<UpwardMessage> {
 		let cache_entry = self.ensure_cached::<T>(para);
-		let upward_message = cache_entry.queue.get(cache_entry.consumed_count);
-		if let Some(msg) = upward_message {
-			cache_entry.consumed_count += 1;
-			cache_entry.consumed_size += msg.len();
-		}
 
+		match cache_entry.queue.get_mut(cache_entry.consumed_count) {
+			Some(msg) => {
+				cache_entry.consumed_count += 1;
+				cache_entry.consumed_size += msg.len();
+
+				Some(mem::take(msg))
+			},
+			None => None,
+		}
+	}
+
+	/// Returns if the queue for the given para is empty.
+	///
+	/// That is, if this returns `true` then the next call to [`peek_front`] will return `None`.
+	///
+	/// Does not mutate the queue.
+	fn is_empty<T: Config>(&mut self, para: ParaId) -> bool {
+		let cache_entry = self.ensure_cached::<T>(para);
 		cache_entry.consumed_count >= cache_entry.queue.len()
 	}
 
@@ -600,8 +707,11 @@ impl NeedsDispatchCursor {
 #[cfg(test)]
 pub(crate) mod tests {
 	use super::*;
-	use crate::mock::{new_test_ext, take_processed, Configuration, MockGenesisConfig, Ump};
-	use frame_support::weights::Weight;
+	use crate::mock::{
+		assert_last_event, new_test_ext, take_processed, Configuration, MockGenesisConfig, Origin,
+		System, Test, Ump,
+	};
+	use frame_support::{assert_noop, assert_ok, weights::Weight};
 	use std::collections::HashSet;
 
 	struct GenesisConfigBuilder {
@@ -610,6 +720,7 @@ pub(crate) mod tests {
 		max_upward_queue_count: u32,
 		max_upward_queue_size: u32,
 		ump_service_total_weight: Weight,
+		ump_max_individual_weight: Weight,
 	}
 
 	impl Default for GenesisConfigBuilder {
@@ -620,6 +731,7 @@ pub(crate) mod tests {
 				max_upward_queue_count: 4,
 				max_upward_queue_size: 64,
 				ump_service_total_weight: 1000,
+				ump_max_individual_weight: 100,
 			}
 		}
 	}
@@ -634,6 +746,7 @@ pub(crate) mod tests {
 			config.max_upward_queue_count = self.max_upward_queue_count;
 			config.max_upward_queue_size = self.max_upward_queue_size;
 			config.ump_service_total_weight = self.ump_service_total_weight;
+			config.ump_max_individual_weight = self.ump_max_individual_weight;
 			genesis
 		}
 	}
@@ -778,7 +891,12 @@ pub(crate) mod tests {
 		let a_msg_2 = (300u32, "a_msg_2").encode();
 
 		new_test_ext(
-			GenesisConfigBuilder { ump_service_total_weight: 500, ..Default::default() }.build(),
+			GenesisConfigBuilder {
+				ump_service_total_weight: 500,
+				ump_max_individual_weight: 300,
+				..Default::default()
+			}
+			.build(),
 		)
 		.execute_with(|| {
 			queue_upward_msg(a, a_msg_1.clone());
@@ -856,4 +974,71 @@ pub(crate) mod tests {
 			assert_eq!(size, 3);
 		});
 	}
+
+	#[test]
+	fn service_overweight_unknown() {
+		// This test just makes sure that 0 is not a valid index and we can use it not worrying in
+		// the next test.
+		new_test_ext(GenesisConfigBuilder::default().build()).execute_with(|| {
+			assert_noop!(
+				Ump::service_overweight(Origin::root(), 0, 1000),
+				Error::<Test>::UnknownMessageIndex
+			);
+		});
+	}
+
+	#[test]
+	fn overweight_queue_works() {
+		let para_a = ParaId::from(2021);
+
+		let a_msg_1 = (301u32, "a_msg_1").encode();
+		let a_msg_2 = (500u32, "a_msg_2").encode();
+		let a_msg_3 = (500u32, "a_msg_3").encode();
+
+		new_test_ext(
+			GenesisConfigBuilder {
+				ump_service_total_weight: 900,
+				ump_max_individual_weight: 300,
+				..Default::default()
+			}
+			.build(),
+		)
+		.execute_with(|| {
+			// HACK: Start with the block number 1. This is needed because should an event be
+			// emitted during the genesis block they will be implicitly wiped.
+			System::set_block_number(1);
+
+			// This one is overweight. However, the weight is plenty and we can afford to execute
+			// this message, thus expect it.
+			queue_upward_msg(para_a, a_msg_1.clone());
+			Ump::process_pending_upward_messages();
+			assert_eq!(take_processed(), vec![(para_a, a_msg_1)]);
+
+			// This is overweight and this message cannot fit into the total weight budget.
+			queue_upward_msg(para_a, a_msg_2.clone());
+			queue_upward_msg(para_a, a_msg_3.clone());
+			Ump::process_pending_upward_messages();
+			assert_last_event(
+				Event::OverweightEnqueued(para_a, upward_message_id(&a_msg_3[..]), 0, 500).into(),
+			);
+
+			// Now verify that if we wanted to service this overweight message with less than enough
+			// weight it will fail.
+			assert_noop!(
+				Ump::service_overweight(Origin::root(), 0, 499),
+				Error::<Test>::WeightOverLimit
+			);
+
+			// ... and if we try to service it with just enough weight it will succeed as well.
+			assert_ok!(Ump::service_overweight(Origin::root(), 0, 500));
+			assert_last_event(Event::OverweightServiced(0, 500).into());
+
+			// ... and if we try to service a message with index that doesn't exist it will error
+			// out.
+			assert_noop!(
+				Ump::service_overweight(Origin::root(), 1, 1000),
+				Error::<Test>::UnknownMessageIndex
+			);
+		});
+	}
 }

polkadot/runtime/rococo/src/lib.rs

@@ -718,6 +718,7 @@ impl parachains_ump::Config for Runtime {
 	type Event = Event;
 	type UmpSink = crate::parachains_ump::XcmSink<XcmExecutor<XcmConfig>, Runtime>;
 	type FirstMessageFactorPercent = FirstMessageFactorPercent;
+	type ExecuteOverweightOrigin = EnsureRoot<AccountId>;
 }
 
 impl parachains_dmp::Config for Runtime {}

polkadot/runtime/test-runtime/src/lib.rs

@@ -491,6 +491,7 @@ impl parachains_ump::Config for Runtime {
 	type Event = Event;
 	type UmpSink = ();
 	type FirstMessageFactorPercent = FirstMessageFactorPercent;
+	type ExecuteOverweightOrigin = frame_system::EnsureRoot<AccountId>;
 }
 
 parameter_types! {

polkadot/runtime/westend/src/lib.rs

@@ -786,6 +786,7 @@ impl parachains_ump::Config for Runtime {
 	type Event = Event;
 	type UmpSink = crate::parachains_ump::XcmSink<XcmExecutor<XcmConfig>, Runtime>;
 	type FirstMessageFactorPercent = FirstMessageFactorPercent;
+	type ExecuteOverweightOrigin = EnsureRoot<AccountId>;
 }
 
 impl parachains_dmp::Config for Runtime {}

polkadot/scripts/gitlab/lingua.dic

@@ -249,6 +249,7 @@ timestamp/MS
 transitionary
 trie/MS
 trustless/Y
+TTL
 tuple/SM
 typesystem
 ubuntu/M
@@ -292,6 +293,7 @@ WND/S
 Wococo
 XCM/S
 XCMP/M
+yeet
 decrement
 DM
-ParaId
+ParaId

polkadot/xcm/xcm-simulator/example/src/relay_chain.rs

@@ -166,6 +166,7 @@ impl ump::Config for Runtime {
 	type Event = Event;
 	type UmpSink = ump::XcmSink<XcmExecutor<XcmConfig>, Runtime>;
 	type FirstMessageFactorPercent = FirstMessageFactorPercent;
+	type ExecuteOverweightOrigin = frame_system::EnsureRoot<AccountId>;
 }
 
 impl origin::Config for Runtime {}