Use Message Queue pallet for UMP dispatch

node/network/approval-distribution/src/lib.rs

@@ -1270,11 +1270,11 @@ impl State {
 			let candidate_entry = match block_entry.candidates.get(index as usize) {
 				None => {
 					gum::debug!(
-						target: LOG_TARGET,
-						?hash,
-						?index,
-						"`get_approval_signatures`: could not find candidate entry for given hash and index!"
-						);
+					target: LOG_TARGET,
+					?hash,
+					?index,
+					"`get_approval_signatures`: could not find candidate entry for given hash and index!"
+					);
 					continue
 				},
 				Some(e) => e,

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/v4/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;
@@ -210,6 +210,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"];
 
@@ -224,6 +225,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.
@@ -1706,6 +1725,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)]
 #[cfg_attr(feature = "std", derive(serde::Serialize, serde::Deserialize))]

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

@@ -1,6 +1,6 @@
 # Inclusion Module
 
-The inclusion module is responsible for inclusion and availability of scheduled parachains and parathreads.
+The inclusion module is responsible for inclusion and availability of scheduled parachains and parathreads. It also manages the UMP dispatch queue of each parachain/thread.
 
 ## Storage
 
@@ -35,11 +35,27 @@ PendingAvailability: map ParaId => CandidatePendingAvailability;
 PendingAvailabilityCommitments: map ParaId => CandidateCommitments;
 ```
 
+## Config Dependencies
+
+* `MessageQueue`:
+  The message queue provides general queueing and processing functionality. Currently it
+	replaces the old `UMP` dispatch queue. Other use-cases can be implemented as well by
+	adding new variants to `AggregateMessageOrigin`. Normally it should be set to an instance
+  of the `MessageQueue` pallet.
+
 ## Session Change
 
 1. Clear out all candidates pending availability.
 1. Clear out all validator bitfields.
 
+Optional:
+1. The UMP queue of all outgoing paras can be "swept". This would prevent the dispatch queue from automatically being serviced. It is a consideration for the chain and specific behaviour is not defined.
+
+## Initialization
+
+No initialization routine runs for this module. However, the initialization of the `MessageQueue` pallet will attempt to process any pending UMP messages.
+
+
 ## Routines
 
 All failed checks should lead to an unrecoverable error making the block invalid.
@@ -88,7 +104,7 @@ All failed checks should lead to an unrecoverable error making the block invalid
   1. Ensure that any code upgrade scheduled by the candidate does not happen within `config.validation_upgrade_cooldown` of `Paras::last_code_upgrade(para_id, true)`, if any, comparing against the value of `Paras::FutureCodeUpgrades` for the given para ID.
   1. Check the collator's signature on the candidate data.
   1. check the backing of the candidate using the signatures and the bitfields, comparing against the validators assigned to the groups, fetched with the `group_validators` lookup.
-  1. call `Ump::check_upward_messages(para, commitments.upward_messages)` to check that the upward messages are valid.
+  1. call `check_upward_messages(config, para, commitments.upward_messages)` to check that the upward messages are valid.
   1. call `Dmp::check_processed_downward_messages(para, commitments.processed_downward_messages)` to check that the DMQ is properly drained.
   1. call `Hrmp::check_hrmp_watermark(para, commitments.hrmp_watermark)` for each candidate to check rules of processing the HRMP watermark.
   1. using `Hrmp::check_outbound_hrmp(sender, commitments.horizontal_messages)` ensure that the each candidate sent a valid set of horizontal messages
@@ -99,7 +115,7 @@ All failed checks should lead to an unrecoverable error making the block invalid
   1. If the receipt contains a code upgrade, Call `Paras::schedule_code_upgrade(para_id, code, relay_parent_number, config)`.
     > TODO: Note that this is safe as long as we never enact candidates where the relay parent is across a session boundary. In that case, which we should be careful to avoid with contextual execution, the configuration might have changed and the para may de-sync from the host's understanding of it.
   1. Reward all backing validators of each candidate, contained within the `backers` field.
-  1. call `Ump::receive_upward_messages` for each backed candidate, using the [`UpwardMessage`s](../types/messages.md#upward-message) from the [`CandidateCommitments`](../types/candidate.md#candidate-commitments).
+  1. call `receive_upward_messages` for each backed candidate, using the [`UpwardMessage`s](../types/messages.md#upward-message) from the [`CandidateCommitments`](../types/candidate.md#candidate-commitments).
   1. call `Dmp::prune_dmq` with the para id of the candidate and the candidate's `processed_downward_messages`.
   1. call `Hrmp::prune_hrmp` with the para id of the candiate and the candidate's `hrmp_watermark`.
   1. call `Hrmp::queue_outbound_hrmp` with the para id of the candidate and the list of horizontal messages taken from the commitment,
@@ -118,3 +134,20 @@ All failed checks should lead to an unrecoverable error making the block invalid
 * `candidate_pending_availability(ParaId) -> Option<CommittedCandidateReceipt>`: returns the `CommittedCandidateReceipt` pending availability for the para provided, if any.
 * `pending_availability(ParaId) -> Option<CandidatePendingAvailability>`: returns the metadata around the candidate pending availability for the para, if any.
 * `collect_disputed(disputed: Vec<CandidateHash>) -> Vec<CoreIndex>`: Sweeps through all paras pending availability. If the candidate hash is one of the disputed candidates, then clean up the corresponding storage for that candidate and the commitments. Return a vector of cleaned-up core IDs.
+
+These functions were formerly part of the UMP pallet:
+
+* `check_upward_messages(P: ParaId, Vec<UpwardMessage>)`:
+    1. Checks that the parachain is not currently offboarding and error otherwise. 
+    1. Checks that there are at most `config.max_upward_message_num_per_candidate` messages to be enqueued.
+    1. Checks that no message exceeds `config.max_upward_message_size`.
+    1. Checks that the total resulting queue size would not exceed `co`.
+    1. Verify that queuing up the messages could not result in exceeding the queue's footprint
+    according to the config items `config.max_upward_queue_count` and `config.max_upward_queue_size`. 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. Place in the dispatch queue according to its para ID (or handle it immediately).

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

@@ -63,11 +63,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`.