Use Message Queue pallet for UMP dispatch

node/service/src/chain_spec.rs

@@ -17,7 +17,6 @@
 //! Polkadot chain configurations.
 
 use beefy_primitives::crypto::AuthorityId as BeefyId;
-use frame_support::weights::Weight;
 use grandpa::AuthorityId as GrandpaId;
 #[cfg(feature = "kusama-native")]
 use kusama_runtime as kusama;
@@ -189,7 +188,6 @@ fn default_parachains_host_configuration(
 		max_upward_queue_count: 8,
 		max_upward_queue_size: 1024 * 1024,
 		max_downward_message_size: 1024 * 1024,
-		ump_service_total_weight: Weight::from_parts(100_000_000_000, MAX_POV_SIZE as u64),
 		max_upward_message_size: 50 * 1024,
 		max_upward_message_num_per_candidate: 5,
 		hrmp_sender_deposit: 0,

primitives/src/v2/mod.rs

@@ -133,7 +133,7 @@ pub type ValidatorSignature = validator_app::Signature;
 
 /// A declarations of storage keys where an external observer can find some interesting data.
 pub mod well_known_keys {
-	use super::{HrmpChannelId, Id};
+	use super::{HrmpChannelId, Id, WellKnownKey};
 	use hex_literal::hex;
 	use parity_scale_codec::Encode as _;
 	use sp_io::hashing::twox_64;
@@ -193,6 +193,7 @@ pub mod well_known_keys {
 	///
 	/// - `count: u32`, the number of messages currently in the queue for given para,
 	/// - `total_size: u32`, the total size of all messages in the queue.
+	#[deprecated = "Use `relay_dispatch_queue_remaining_capacity` instead"]
 	pub fn relay_dispatch_queue_size(para_id: Id) -> Vec<u8> {
 		let prefix = hex!["f5207f03cfdce586301014700e2c2593fad157e461d71fd4c1f936839a5f1f3e"];
 
@@ -207,6 +208,24 @@ pub mod well_known_keys {
 		})
 	}
 
+	/// Type safe version of `relay_dispatch_queue_size`.
+	#[deprecated = "Use `relay_dispatch_queue_remaining_capacity` instead"]
+	pub fn relay_dispatch_queue_size_typed(para: Id) -> WellKnownKey<(u32, u32)> {
+		#[allow(deprecated)]
+		relay_dispatch_queue_size(para).into()
+	}
+
+	/// The upward message dispatch queue remaining capacity for the given para id.
+	///
+	/// The storage entry stores a tuple of two values:
+	///
+	/// - `count: u32`, the number of additional messages which may be enqueued for the given para,
+	/// - `total_size: u32`, the total size of additional messages which may be enqueued for the
+	/// given para.
+	pub fn relay_dispatch_queue_remaining_capacity(para_id: Id) -> WellKnownKey<(u32, u32)> {
+		(b":relay_dispatch_queue_remaining_capacity", para_id).encode().into()
+	}
+
 	/// The HRMP channel for the given identifier.
 	///
 	/// The storage entry should be accessed as an `AbridgedHrmpChannel` encoded value.
@@ -1689,6 +1708,42 @@ impl PvfCheckStatement {
 	}
 }
 
+/// A well-known and typed storage key.
+///
+/// Allows for type-safe access to raw well-known storage keys.
+pub struct WellKnownKey<T> {
+	/// The raw storage key.
+	pub key: Vec<u8>,
+	_p: sp_std::marker::PhantomData<T>,
+}
+
+impl<T> From<Vec<u8>> for WellKnownKey<T> {
+	fn from(key: Vec<u8>) -> Self {
+		Self { key, _p: Default::default() }
+	}
+}
+
+impl<T> AsRef<[u8]> for WellKnownKey<T> {
+	fn as_ref(&self) -> &[u8] {
+		self.key.as_ref()
+	}
+}
+
+impl<T: Decode> WellKnownKey<T> {
+	/// Gets the value or `None` if it does not exist or decoding failed.
+	pub fn get(&self) -> Option<T> {
+		sp_io::storage::get(&self.key)
+			.and_then(|raw| parity_scale_codec::DecodeAll::decode_all(&mut raw.as_ref()).ok())
+	}
+}
+
+impl<T: Encode> WellKnownKey<T> {
+	/// Sets the value.
+	pub fn set(&self, value: T) {
+		sp_io::storage::set(&self.key, &value.encode());
+	}
+}
+
 /// Type discriminator for PVF preparation timeouts
 #[derive(Encode, Decode, TypeInfo, Clone, Copy, Debug, PartialEq, Eq)]
 pub enum PvfPrepTimeoutKind {

roadmap/implementers-guide/src/runtime/ump.md

@@ -52,36 +52,18 @@ Candidate Acceptance Function:
 * `check_upward_messages(P: ParaId, Vec<UpwardMessage>`):
     1. Checks that there are at most `config.max_upward_message_num_per_candidate` messages.
     1. Checks that no message exceeds `config.max_upward_message_size`.
-    1. Verify that `RelayDispatchQueueSize` for `P` has enough capacity for the messages
+    1. Verify that queuing up the messages could not result in exceeding the queue's footprint
+    according to the config items. The queue's current footprint is provided in well_known_keys
+    in order to facilitate oraclisation on to the para.
 
 Candidate Enactment:
 
 * `receive_upward_messages(P: ParaId, Vec<UpwardMessage>)`:
     1. Process each upward message `M` in order:
-        1. Append the message to `RelayDispatchQueues` for `P`
-        1. Increment the size and the count in `RelayDispatchQueueSize` for `P`.
-        1. Ensure that `P` is present in `NeedsDispatch`.
-
-The following routine is meant to execute pending entries in upward message queues. This function doesn't fail, even if
-dispatching any of individual upward messages returns an error.
-
-`process_pending_upward_messages()`:
-    1. Initialize a cumulative weight counter `T` to 0
-    1. Iterate over items in `NeedsDispatch` cyclically, starting with `NextDispatchRoundStartWith`. If the item specified is `None` start from the beginning. For each `P` encountered:
-        1. Dequeue the first upward message `D` from `RelayDispatchQueues` for `P`
-        1. Decrement the size of the message from `RelayDispatchQueueSize` for `P`
-        1. Delegate processing of the message to the runtime. The weight consumed is added to `T`.
-        1. If `T >= config.ump_service_total_weight`, set `NextDispatchRoundStartWith` to `P` and finish processing.
-        1. If `RelayDispatchQueues` for `P` became empty, remove `P` from `NeedsDispatch`.
-        1. If `NeedsDispatch` became empty then finish processing and set `NextDispatchRoundStartWith` to `None`.
-        > NOTE that in practice we would need to approach the weight calculation more thoroughly, i.e. incorporate all operations
-        > that could take place on the course of handling these upward messages.
+        1. Place in the dispatch queue according to its para ID (or handle it immediately).
 
 ## Session Change
 
-1. For each `P` in `outgoing_paras` (generated by `Paras::on_new_session`):
-    1. Remove `RelayDispatchQueueSize` of `P`.
-    1. Remove `RelayDispatchQueues` of `P`.
-    1. Remove `P` if it exists in `NeedsDispatch`.
-    1. If `P` is in `NextDispatchRoundStartWith`, then reset it to `None`
-    - Note that if we don't remove the open/close requests since they are going to die out naturally at the end of the session.
+1. Nothing specific needs to be done, however the channel's dispatch queue may possibly be "swept"
+which would prevent the dispatch queue from automatically being serviced. This is a consideration
+for the chain and specific behaviour is not defined.

roadmap/implementers-guide/src/types/runtime.md

@@ -65,11 +65,6 @@ struct HostConfiguration {
 	/// no further messages may be added to it. If it exceeds this then the queue may contain only
 	/// a single message.
 	pub max_upward_queue_size: u32,
-	/// The amount of weight we wish to devote to the processing the dispatchable upward messages
-	/// stage.
-	///
-	/// NOTE that this is a soft limit and could be exceeded.
-	pub ump_service_total_weight: Weight,
 	/// The maximum size of an upward message that can be sent by a candidate.
 	///
 	/// This parameter affects the upper bound of size of `CandidateCommitments`.

runtime/common/src/assigned_slots.rs

@@ -646,6 +646,7 @@ mod tests {
 		type RuntimeEvent = RuntimeEvent;
 		type WeightInfo = parachains_paras::TestWeightInfo;
 		type UnsignedPriority = ParasUnsignedPriority;
+		type UmpQueueTracker = ();
 		type NextSessionRotation = crate::mock::TestNextSessionRotation;
 	}
 

runtime/common/src/integration_tests.rs

@@ -196,6 +196,7 @@ impl paras::Config for Test {
 	type RuntimeEvent = RuntimeEvent;
 	type WeightInfo = paras::TestWeightInfo;
 	type UnsignedPriority = ParasUnsignedPriority;
+	type UmpQueueTracker = ();
 	type NextSessionRotation = crate::mock::TestNextSessionRotation;
 }
 

runtime/common/src/paras_registrar.rs

@@ -765,6 +765,7 @@ mod tests {
 		type RuntimeEvent = RuntimeEvent;
 		type WeightInfo = paras::TestWeightInfo;
 		type UnsignedPriority = ParasUnsignedPriority;
+		type UmpQueueTracker = ();
 		type NextSessionRotation = crate::mock::TestNextSessionRotation;
 	}
 

runtime/common/src/paras_sudo_wrapper.rs

@@ -24,7 +24,7 @@ use primitives::Id as ParaId;
 use runtime_parachains::{
 	configuration, dmp, hrmp,
 	paras::{self, ParaGenesisArgs},
-	ump, ParaLifecycle,
+	ParaLifecycle,
 };
 use sp_std::boxed::Box;
 
@@ -38,10 +38,7 @@ pub mod pallet {
 
 	#[pallet::config]
 	#[pallet::disable_frame_system_supertrait_check]
-	pub trait Config:
-		configuration::Config + paras::Config + dmp::Config + ump::Config + hrmp::Config
-	{
-	}
+	pub trait Config: configuration::Config + paras::Config + dmp::Config + hrmp::Config {}
 
 	#[pallet::error]
 	pub enum Error<T> {

runtime/kusama/Cargo.toml

@@ -59,6 +59,7 @@ pallet-identity = { git = "https://github.com/paritytech/substrate", branch = "m
 pallet-im-online = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
 pallet-indices = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
 pallet-membership = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
+pallet-message-queue = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
 pallet-multisig = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
 pallet-nomination-pools = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
 pallet-offences = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
@@ -157,6 +158,7 @@ std = [
 	"pallet-im-online/std",
 	"pallet-indices/std",
 	"pallet-membership/std",
+	"pallet-message-queue/std",
 	"pallet-multisig/std",
 	"pallet-nomination-pools/std",
 	"pallet-nomination-pools-runtime-api/std",
@@ -224,6 +226,7 @@ runtime-benchmarks = [
 	"pallet-im-online/runtime-benchmarks",
 	"pallet-indices/runtime-benchmarks",
 	"pallet-membership/runtime-benchmarks",
+	"pallet-message-queue/runtime-benchmarks",
 	"pallet-multisig/runtime-benchmarks",
 	"pallet-nomination-pools/runtime-benchmarks",
 	"pallet-nomination-pools-benchmarking/runtime-benchmarks",
@@ -275,6 +278,7 @@ try-runtime = [
 	"pallet-im-online/try-runtime",
 	"pallet-indices/try-runtime",
 	"pallet-membership/try-runtime",
+	"pallet-message-queue/try-runtime",
 	"pallet-multisig/try-runtime",
 	"pallet-nomination-pools/try-runtime",
 	"pallet-offences/try-runtime",