Implementer's Guide: Integrate DMP into PersistentValidationData + DMQ watermarks

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

@@ -69,7 +69,7 @@ All failed checks should lead to an unrecoverable error making the block invalid
   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 `Router::check_upward_messages(para, commitments.upward_messages)` to check that the upward messages are valid.
-  1. call `Router::check_processed_downward_messages(para, commitments.processed_downward_messages)` to check that the DMQ is properly drained.
+  1. call `Router::check_dmq_watermark(para, commitments.dmq_watermark)` for each candidate to check rules of processing the DMQ watermark.
   1. call `Router::check_hrmp_watermark(para, commitments.hrmp_watermark)` for each candidate to check rules of processing the HRMP watermark.
   1. check that in the commitments of each candidate the horizontal messages are sorted by ascending recipient ParaId and there is no two horizontal messages have the same recipient.
   1. using `Router::verify_outbound_hrmp(sender, commitments.horizontal_messages)` ensure that the each candidate send a valid set of horizontal messages
@@ -82,7 +82,7 @@ All failed checks should lead to an unrecoverable error making the block invalid
   1. call `Router::enact_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 `Router::queue_outbound_hrmp` with the para id of the candidate and the list of horizontal messages taken from the commitment,
   1. call `Router::prune_hrmp` with the para id of the candiate and the candidate's `hrmp_watermark`.
-  1. call `Router::prune_dmq` with the para id of the candidate and the candidate's `processed_downward_messages`.
+  1. call `Router::prune_dmq` with the para id of the candidate and the candidate's `dmq_watermark`.
   1. Call `Paras::note_new_head` using the `HeadData` from the receipt and `relay_parent_number`.
 * `collect_pending`:
 

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

@@ -23,8 +23,28 @@ NeedsDispatch: Vec<ParaId>;
 /// This is the para that gets will get dispatched first during the next upward dispatchable queue
 /// execution round.
 NextDispatchRoundStartWith: Option<ParaId>;
+```
+
+### Downward Message Passing (DMP)
+
+Storage layout required for implementation of DMP.
+
+```
 /// The downward messages addressed for a certain para.
-DownwardMessageQueues: map ParaId => Vec<DownwardMessage>;
+DownwardMessageQueues: map ParaId => Vec<InboundDownwardMessage>;
+/// A mapping that stores the downward message queue MQC head for each para.
+///
+/// Each link in this chain has a form:
+/// `(prev_head, B, H(M))`, where
+/// - `prev_head`: is the previous head hash.
+/// - `B`: is the relay-chain block number in which a message was appended.
+/// - `H(M)`: is the hash of the message being appended.
+/// This value is initialized to a special value that consists of all zeroes which indicates
+/// that no messages were previously added.
+DownwardMessageQueueHeads: map ParaId => Option<Hash>;
+DownwardMessageQueueWatermarks: map ParaId => Option<BlockNumber>;
+/// A set of numbers of blocks in which at least one downward message were enqueued for a given para.
+DownwardMessagesDigest: map ParaId => Vec<BlockNumber>;
 ```
 
 ### HRMP
@@ -115,7 +135,7 @@ HrmpEgressChannelsIndex: map ParaId => Vec<ParaId>;
 HrmpChannelContents: map HrmpChannelId => Vec<InboundHrmpMessage>;
 /// Maintains a mapping that can be used to answer the question:
 /// What paras sent a message at the given block number for a given reciever.
-/// Invariant: The vector is never empty.
+/// Invariant: The para ids vector is never empty.
 HrmpChannelDigests: map ParaId => Vec<(BlockNumber, Vec<ParaId>)>;
 ```
 
@@ -151,9 +171,10 @@ Candidate Acceptance Function:
       1. Check that `P` is either `ch.sender` or `ch.recipient`
       1. Check that `HrmpChannels` for `ch` exists.
       1. Check that `ch` is not in the `HrmpCloseChannelRequests` set.
-* `check_processed_downward_messages(P: ParaId, processed_downward_messages)`:
-  1. Checks that `DownwardMessageQueues` for `P` is at least `processed_downward_messages` long.
-  1. Checks that `processed_downward_messages` is at least 1 if `DownwardMessageQueues` for `P` is not empty.
+* `check_dmq_watermark(P: ParaId, new_dmq_watermark)`:
+  1. `new_dmq_watermark` should be strictly greater than the value of `DownwardMessageQueueWatermarks` for `P` (if any).
+  1. `new_dmq_watermark` must be not greater than the context's block number.
+  1. in `DownwardMessagesDigest` for `P` an entry with the block number equal to `new_dmq_watermark` should exist.
 * `check_hrmp_watermark(P: ParaId, new_hrmp_watermark)`:
   1. `new_hrmp_watermark` should be strictly greater than the value of `HrmpWatermarks` for `P` (if any).
   1. `new_hrmp_watermark` must not be greater than the context's block number.
@@ -168,7 +189,7 @@ Candidate Enactment:
 
 * `queue_outbound_hrmp(sender: ParaId, Vec<OutboundHrmpMessage>)`:
   1. For each horizontal message `HM` with the channel `C` identified by `(sender, HM.recipient)`:
-    1. Append `HM` into `HrmpChannelContents` that corresponds to `C`.
+    1. Append `HM` into `HrmpChannelContents` that corresponds to `C` with `sent_at` equals to the current block number.
     1. Locate or create an entry in ``HrmpChannelDigests`` for `HM.recipient` and append `sender` into the entry's list.
     1. Increment `C.used_places`
     1. Increment `C.used_bytes` by `HM`'s payload size
@@ -181,8 +202,9 @@ Candidate Enactment:
       1. Decrement `C.used_places`
       1. Decrement `C.used_bytes` by `M`'s payload size.
   1. Set `HrmpWatermarks` for `P` to be equal to `new_hrmp_watermark`
-* `prune_dmq(P: ParaId, processed_downward_messages)`:
-  1. Remove the first `processed_downward_messages` from the `DownwardMessageQueues` of `P`.
+* `prune_dmq(P: ParaId, new_dmq_watermark)`:
+  1. Remove all messages in `DownwardMessageQueues` for `P` up to (including) a message with `sent_at` equal to `new_dmq_watermark`.
+  1. Remove all block numbers up to (including) `new_dmq_watermark` in `DownwardMessagesDigest` for `P`
 * `enact_upward_messages(P: ParaId, Vec<UpwardMessage>)`:
   1. Process all upward messages in order depending on their kinds:
   1. If the message kind is `Dispatchable`:
@@ -228,14 +250,25 @@ any of dispatchables return an error.
       1. If `NeedsDispatch` became empty then finish processing and set `NextDispatchRoundStartWith` to `None`.
   1. Then, for each `P` and the vector of `DispatchResult` in `R`:
       1. Obtain a message by wrapping the vector into `DownwardMessage::DispatchResult`
-      1. Append the resulting message to `DownwardMessageQueues` for `P`.
+      1. Call `enqueue_downward_message` with `P` and the resulting message.
+
+Utility routines.
+
+`queue_downward_message(P: ParaId, M: DownwardMessage)`:
+    1. Wrap `M` into `InboundDownwardMessage` using the current block number for `sent_at`.
+    1. Obtain a new MQC link for the resulting `InboundDownwardMessage` and replace `DownwardMessageQueueHeads` for `P` with the resulting hash.
+    1. Adds the current block number to the set `DownwardMessagesDigest` for `P`.
+    1. Add the resulting `InboundDownwardMessage` into `DownwardMessageQueues` for `P`.
 
 ## Session Change
 
 1. Drain `OutgoingParas`. For each `P` happened to be in the list:
   1. Remove all inbound channels of `P`, i.e. `(_, P)`,
   1. Remove all outbound channels of `P`, i.e. `(P, _)`,
   1. Remove all `DownwardMessageQueues` of `P`.
+  1. Remove `DownwardMessageQueueWatermarks` for `P`.
+  1. Remove `DownwardMessageQueueHeads` for `P`.
+  1. Remove `DownwardMessagesDigest` for `P`.
   1. Remove `RelayDispatchQueueSize` of `P`.
   1. Remove `RelayDispatchQueues` of `P`.
   1. Remove `P` if it exists in `NeedsDispatch`.
@@ -259,12 +292,12 @@ any of dispatchables return an error.
         1. decrement `HrmpOpenChannelRequestCount` for `D.sender` by 1.
         1. remove `R`
         1. remove `D`
-1. For each channel designator `D` in `HrmpCloseChannelRequestsList`
+1. For each HRMP channel designator `D` in `HrmpCloseChannelRequestsList`
     1. remove the channel identified by `D`, if exists.
     1. remove `D` from `HrmpCloseChannelRequests`.
     1. remove `D` from `HrmpCloseChannelRequestsList`
 
-To remove a channel `C` identified with a tuple `(sender, recipient)`:
+To remove a HRMP channel `C` identified with a tuple `(sender, recipient)`:
 
 1. Return `C.sender_deposit` to the `sender`.
 1. Return `C.recipient_deposit` to the `recipient`.

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

@@ -129,9 +129,14 @@ struct PersistedValidationData {
 	/// vector is sorted ascending by the para id and doesn't contain multiple entries with the same
 	/// sender.
 	///
-	/// The MQC heads will be used by the validation function to authorize the input messages passed
+	/// The HRMP MQC heads will be used by the validation function to authorize the input messages passed
 	/// by the collator.
 	hrmp_mqc_heads: Vec<(ParaId, Hash)>,
+	/// The MQC head for the DMQ.
+	///
+	/// The DMQ MQC head will be used by the validation fnction to authorize the downward messages
+	/// passed by the collator.
+	dmq_mqc_head: Hash,
 }
 ```
 
@@ -168,10 +173,14 @@ struct TransientValidationData {
 	/// This informs a relay-chain backing check and the parachain logic.
 	code_upgrade_allowed: Option<BlockNumber>,
 	/// A copy of `config.max_upward_message_num_per_candidate` for checking that a candidate doesn't
-	/// send more messages that permitted.
+	/// send more messages than permitted.
 	config_max_upward_message_num_per_candidate: u32,
-	/// The number of messages pending of the downward message queue.
-	dmq_length: u32,
+	/// A vector that enumerates the list of blocks in which there was at least one DMQ messages
+	/// enqueued.
+	///
+	/// The first number in the vector, if any, is always greater than the DMQ watermark. The
+	/// elements are ordered by ascending the block number. The vector doesn't contain duplicates.
+	dmq_digest: Vec<BlockNumber>,
 	/// A part of transient validation data related to HRMP.
 	hrmp: HrmpTransientValidationData,
 }
@@ -246,8 +255,8 @@ struct CandidateCommitments {
 	new_validation_code: Option<ValidationCode>,
 	/// The head-data produced as a result of execution.
 	head_data: HeadData,
-	/// The number of messages processed from the DMQ.
-	processed_downward_messages: u32,
+	/// The mark which specifies the block number up to which all inbound DMP messages are processed.
+	dmq_watermark: BlockNumber,
 	/// The mark which specifies the block number up to which all inbound HRMP messages are processed.
 	hrmp_watermark: BlockNumber,
 }
@@ -284,8 +293,8 @@ struct ValidationOutputs {
 	fees: Balance,
 	/// The new validation code submitted by the execution, if any.
 	new_validation_code: Option<ValidationCode>,
-	/// The number of messages processed from the DMQ.
-	processed_downward_messages: u32,
+	/// The mark which specifies the block number up to which all inbound DMP messages are processed.
+	dmq_watermark: BlockNumber,
 	/// The mark which specifies the block number up to which all inbound HRMP messages are processed.
 	hrmp_watermark: BlockNumber,
 }

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

@@ -90,6 +90,9 @@ struct OutboundHrmpMessage {
 }
 
 struct InboundHrmpMessage {
+	/// The block number at which this message was sent.
+	/// Specifically, it is the block number at which the candidate that sends this message was
+	/// enacted.
 	pub sent_at: BlockNumber,
 	/// The message payload.
 	pub data: Vec<u8>,
@@ -125,4 +128,13 @@ enum DownwardMessage {
 	/// paras.
 	ParachainSpecific(Vec<u8>),
 }
+
+/// A wrapped version of `DownwardMessage`. The difference is that it has attached the block number when
+/// the message was sent.
+struct InboundDownwardMessage {
+	/// The block number at which this messages was put into the downward message queue.
+	pub sent_at: BlockNumber,
+	/// The actual downward message to processes.
+	pub msg: DownwardMessage,
+}
 ```