Downward Message Processing implementation

core-primitives/src/lib.rs

@@ -79,20 +79,19 @@ pub use sp_runtime::OpaqueExtrinsic as UncheckedExtrinsic;
 /// the parachain.
 pub type Remark = [u8; 32];
 
-/// These are special "control" messages that can be passed from the Relaychain to a parachain.
-/// They should be handled by all parachains.
+/// A message sent from the relay-chain down to a parachain.
+///
+/// The size of the message is limited by the `config.max_downward_message_size` parameter.
+pub type DownwardMessage = sp_std::vec::Vec<u8>;
+
+/// A wrapped version of `DownwardMessage`. The difference is that it has attached the block number when
+/// the message was sent.
 #[derive(codec::Encode, codec::Decode, Clone, sp_runtime::RuntimeDebug, PartialEq)]
-pub enum DownwardMessage<AccountId = crate::AccountId> {
-	/// Some funds were transferred into the parachain's account. The hash is the identifier that
-	/// was given with the transfer.
-	TransferInto(AccountId, Balance, Remark),
-	/// An opaque blob of data. The relay chain must somehow know how to form this so that the
-	/// destination parachain does something sensible.
-	///
-	/// NOTE: Be very careful not to allow users to place arbitrary size information in here.
-	Opaque(sp_std::vec::Vec<u8>),
-	/// XCMP message for the Parachain.
-	XCMPMessage(sp_std::vec::Vec<u8>),
+pub struct InboundDownwardMessage<BlockNumber = crate::BlockNumber> {
+	/// 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,
 }
 
 /// V1 primitives.

node/collation-generation/src/lib.rs

@@ -277,6 +277,7 @@ async fn handle_new_activations<Context: SubsystemContext>(
 					new_validation_code: collation.new_validation_code,
 					head_data: collation.head_data,
 					erasure_root,
+					processed_downward_messages: collation.processed_downward_messages,
 				};
 
 				let ccr = CandidateReceipt {
@@ -387,6 +388,7 @@ mod tests {
 				proof_of_validity: PoV {
 					block_data: BlockData(Vec::new()),
 				},
+				processed_downward_messages: Default::default(),
 			}
 		}
 

node/core/av-store/src/tests.rs

@@ -70,6 +70,7 @@ impl Default for TestState {
 			parent_head: HeadData(vec![7, 8, 9]),
 			block_number: 5,
 			hrmp_mqc_heads: Vec::new(),
+			dmq_mqc_head: Default::default(),
 		};
 
 		let pruning_config = PruningConfig {

node/core/backing/src/lib.rs

@@ -682,6 +682,7 @@ impl CandidateBackingJob {
 			erasure_root,
 			new_validation_code: outputs.new_validation_code,
 			head_data: outputs.head_data,
+			processed_downward_messages: outputs.processed_downward_messages,
 		};
 
 		let res = match with_commitments(commitments) {
@@ -977,12 +978,14 @@ mod tests {
 					parent_head: HeadData(vec![7, 8, 9]),
 					block_number: Default::default(),
 					hrmp_mqc_heads: Vec::new(),
+					dmq_mqc_head: Default::default(),
 				},
 				transient: TransientValidationData {
 					max_code_size: 1000,
 					max_head_data_size: 1000,
 					balance: Default::default(),
 					code_upgrade_allowed: None,
+					dmq_length: 0,
 				},
 			};
 
@@ -1159,6 +1162,7 @@ mod tests {
 							upward_messages: Vec::new(),
 							fees: Default::default(),
 							new_validation_code: None,
+							processed_downward_messages: 0,
 						}, test_state.validation_data.persisted),
 					)).unwrap();
 				}
@@ -1278,6 +1282,7 @@ mod tests {
 							upward_messages: Vec::new(),
 							fees: Default::default(),
 							new_validation_code: None,
+							processed_downward_messages: 0,
 						}, test_state.validation_data.persisted),
 					)).unwrap();
 				}
@@ -1416,6 +1421,7 @@ mod tests {
 							upward_messages: Vec::new(),
 							fees: Default::default(),
 							new_validation_code: None,
+							processed_downward_messages: 0,
 						}, test_state.validation_data.persisted),
 					)).unwrap();
 				}
@@ -1571,6 +1577,7 @@ mod tests {
 							upward_messages: Vec::new(),
 							fees: Default::default(),
 							new_validation_code: None,
+							processed_downward_messages: 0,
 						}, test_state.validation_data.persisted),
 					)).unwrap();
 				}

node/core/candidate-selection/src/lib.rs

@@ -494,11 +494,13 @@ mod tests {
 				upward_messages: Vec::new(),
 				fees: 0,
 				new_validation_code: None,
+				processed_downward_messages: 0,
 			},
 			PersistedValidationData {
 				parent_head: HeadData(parent_head_data),
 				block_number: 123,
 				hrmp_mqc_heads: Vec::new(),
+				dmq_mqc_head: Default::default(),
 			},
 		)
 	}

node/core/candidate-validation/src/lib.rs

@@ -468,6 +468,7 @@ fn validate_candidate_exhaustive<B: ValidationBackend, S: SpawnNamed + 'static>(
 		parent_head: persisted_validation_data.parent_head.clone(),
 		block_data: pov.block_data.clone(),
 		relay_chain_height: persisted_validation_data.block_number,
+		dmq_mqc_head: persisted_validation_data.dmq_mqc_head,
 		hrmp_mqc_heads: persisted_validation_data.hrmp_mqc_heads.clone(),
 	};
 
@@ -491,6 +492,7 @@ fn validate_candidate_exhaustive<B: ValidationBackend, S: SpawnNamed + 'static>(
 				upward_messages: res.upward_messages,
 				fees: 0,
 				new_validation_code: res.new_validation_code,
+				processed_downward_messages: res.processed_downward_messages,
 			};
 			Ok(ValidationResult::Valid(outputs, persisted_validation_data))
 		}

node/core/runtime-api/src/lib.rs

@@ -126,6 +126,7 @@ fn make_runtime_api_request<Client>(
 			query!(candidate_pending_availability(para), sender),
 		Request::CandidateEvents(sender) => query!(candidate_events(), sender),
 		Request::ValidatorDiscovery(ids, sender) => query!(validator_discovery(ids), sender),
+		Request::DmqContents(id, sender) => query!(dmq_contents(id), sender),
 	}
 }
 
@@ -175,7 +176,7 @@ mod tests {
 	use polkadot_primitives::v1::{
 		ValidatorId, ValidatorIndex, GroupRotationInfo, CoreState, PersistedValidationData,
 		Id as ParaId, OccupiedCoreAssumption, ValidationData, SessionIndex, ValidationCode,
-		CommittedCandidateReceipt, CandidateEvent, AuthorityDiscoveryId,
+		CommittedCandidateReceipt, CandidateEvent, AuthorityDiscoveryId, InboundDownwardMessage,
 	};
 	use polkadot_node_subsystem_test_helpers as test_helpers;
 	use sp_core::testing::TaskExecutor;
@@ -194,6 +195,7 @@ mod tests {
 		validation_outputs_results: HashMap<ParaId, bool>,
 		candidate_pending_availability: HashMap<ParaId, CommittedCandidateReceipt>,
 		candidate_events: Vec<CandidateEvent>,
+		dmq_contents: HashMap<ParaId, Vec<InboundDownwardMessage>>,
 	}
 
 	impl ProvideRuntimeApi<Block> for MockRuntimeApi {
@@ -282,6 +284,13 @@ mod tests {
 			fn validator_discovery(ids: Vec<ValidatorId>) -> Vec<Option<AuthorityDiscoveryId>> {
 				vec![None; ids.len()]
 			}
+
+			fn dmq_contents(
+				&self,
+				recipient: ParaId,
+			) -> Vec<polkadot_primitives::v1::InboundDownwardMessage> {
+				self.dmq_contents.get(&recipient).map(|q| q.clone()).unwrap_or_default()
+			}
 		}
 	}
 
@@ -614,4 +623,54 @@ mod tests {
 
 		futures::executor::block_on(future::join(subsystem_task, test_task));
 	}
+
+	#[test]
+	fn requests_dmq_contents() {
+		let (ctx, mut ctx_handle) = test_helpers::make_subsystem_context(TaskExecutor::new());
+		let mut runtime_api = MockRuntimeApi::default();
+		let relay_parent = [1; 32].into();
+		let para_a = 5.into();
+		let para_b = 6.into();
+
+		runtime_api.dmq_contents.insert(para_a, vec![]);
+		runtime_api.dmq_contents.insert(
+			para_b,
+			vec![InboundDownwardMessage {
+				sent_at: 228,
+				msg: b"Novus Ordo Seclorum".to_vec(),
+			}],
+		);
+
+		let subsystem = RuntimeApiSubsystem::new(runtime_api.clone(), Metrics(None));
+		let subsystem_task = run(ctx, subsystem).map(|x| x.unwrap());
+		let test_task = async move {
+			let (tx, rx) = oneshot::channel();
+			ctx_handle
+				.send(FromOverseer::Communication {
+					msg: RuntimeApiMessage::Request(relay_parent, Request::DmqContents(para_a, tx)),
+				})
+				.await;
+			assert_eq!(rx.await.unwrap().unwrap(), vec![]);
+
+			let (tx, rx) = oneshot::channel();
+			ctx_handle
+				.send(FromOverseer::Communication {
+					msg: RuntimeApiMessage::Request(relay_parent, Request::DmqContents(para_b, tx)),
+				})
+				.await;
+			assert_eq!(
+				rx.await.unwrap().unwrap(),
+				vec![InboundDownwardMessage {
+					sent_at: 228,
+					msg: b"Novus Ordo Seclorum".to_vec(),
+				}]
+			);
+
+			ctx_handle
+				.send(FromOverseer::Signal(OverseerSignal::Conclude))
+				.await;
+		};
+		futures::executor::block_on(future::join(subsystem_task, test_task));
+	}
+
 }

node/network/availability-distribution/src/tests.rs

@@ -217,6 +217,7 @@ impl Default for TestState {
 			parent_head: HeadData(vec![7, 8, 9]),
 			block_number: Default::default(),
 			hrmp_mqc_heads: Vec::new(),
+			dmq_mqc_head: Default::default(),
 		};
 
 		let validator_index = Some((validators.len() - 1) as ValidatorIndex);

node/primitives/src/lib.rs

@@ -263,6 +263,8 @@ pub struct Collation {
 	pub head_data: HeadData,
 	/// Proof to verify the state transition of the parachain.
 	pub proof_of_validity: PoV,
+	/// The number of messages processed from the DMQ.
+	pub processed_downward_messages: u32,
 }
 
 /// Configuration for the collation generator

node/subsystem/src/messages.rs

@@ -37,7 +37,7 @@ use polkadot_primitives::v1::{
 	GroupRotationInfo, Hash, Id as ParaId, OccupiedCoreAssumption,
 	PersistedValidationData, PoV, SessionIndex, SignedAvailabilityBitfield,
 	ValidationCode, ValidatorId, ValidationData,
-	ValidatorIndex, ValidatorSignature,
+	ValidatorIndex, ValidatorSignature, InboundDownwardMessage,
 };
 use std::sync::Arc;
 
@@ -419,7 +419,11 @@ pub enum RuntimeApiRequest {
 	/// Get the validation code for a para, taking the given `OccupiedCoreAssumption`, which
 	/// will inform on how the validation data should be computed if the para currently
 	/// occupies a core.
-	ValidationCode(ParaId, OccupiedCoreAssumption, RuntimeApiSender<Option<ValidationCode>>),
+	ValidationCode(
+		ParaId,
+		OccupiedCoreAssumption,
+		RuntimeApiSender<Option<ValidationCode>>,
+	),
 	/// Get a the candidate pending availability for a particular parachain by parachain / core index
 	CandidatePendingAvailability(ParaId, RuntimeApiSender<Option<CommittedCandidateReceipt>>),
 	/// Get all events concerning candidates (backing, inclusion, time-out) in the parent of
@@ -429,7 +433,15 @@ pub enum RuntimeApiRequest {
 	/// Currently this request is limited to validators in the current session.
 	///
 	/// Returns `None` for validators not found in the current session.
-	ValidatorDiscovery(Vec<ValidatorId>, RuntimeApiSender<Vec<Option<AuthorityDiscoveryId>>>),
+	ValidatorDiscovery(
+		Vec<ValidatorId>,
+		RuntimeApiSender<Vec<Option<AuthorityDiscoveryId>>>,
+	),
+	/// Get all the pending inbound messages in the downward message queue for a para.
+	DmqContents(
+		ParaId,
+		RuntimeApiSender<Vec<InboundDownwardMessage<BlockNumber>>>,
+	),
 }
 
 /// A message to the Runtime API subsystem.

parachain/src/primitives.rs

@@ -248,6 +248,11 @@ pub struct ValidationParams {
 	pub block_data: BlockData,
 	/// The current relay-chain block number.
 	pub relay_chain_height: RelayChainBlockNumber,
+	/// The MQC head for the DMQ.
+	///
+	/// The DMQ MQC head will be used by the validation function to authorize the downward messages
+	/// passed by the collator.
+	pub dmq_mqc_head: Hash,
 	/// The list of MQC heads for the inbound HRMP channels paired with the sender para ids. This
 	/// vector is sorted ascending by the para id and doesn't contain multiple entries with the same
 	/// sender.

parachain/test-parachains/tests/adder/mod.rs

@@ -97,6 +97,7 @@ fn execute_good_on_parent(execution_mode: ExecutionMode) {
 			block_data: GenericBlockData(block_data.encode()),
 			relay_chain_height: 1,
 			hrmp_mqc_heads: Vec::new(),
+			dmq_mqc_head: Default::default(),
 		},
 		&execution_mode,
 		sp_core::testing::TaskExecutor::new(),
@@ -135,6 +136,7 @@ fn execute_good_chain_on_parent() {
 				block_data: GenericBlockData(block_data.encode()),
 				relay_chain_height: number as RelayChainBlockNumber + 1,
 				hrmp_mqc_heads: Vec::new(),
+				dmq_mqc_head: Default::default(),
 			},
 			&execution_mode,
 			sp_core::testing::TaskExecutor::new(),
@@ -174,6 +176,7 @@ fn execute_bad_on_parent() {
 			block_data: GenericBlockData(block_data.encode()),
 			relay_chain_height: 1,
 			hrmp_mqc_heads: Vec::new(),
+			dmq_mqc_head: Default::default(),
 		},
 		&execution_mode,
 		sp_core::testing::TaskExecutor::new(),

parachain/test-parachains/tests/wasm_executor/mod.rs

@@ -43,6 +43,7 @@ fn terminates_on_timeout() {
 			parent_head: Default::default(),
 			relay_chain_height: 1,
 			hrmp_mqc_heads: Vec::new(),
+			dmq_mqc_head: Default::default(),
 		},
 		&execution_mode,
 		sp_core::testing::TaskExecutor::new(),
@@ -71,6 +72,7 @@ fn parallel_execution() {
 			parent_head: Default::default(),
 			relay_chain_height: 1,
 			hrmp_mqc_heads: Vec::new(),
+			dmq_mqc_head: Default::default(),
 		},
 		&execution_mode,
 		sp_core::testing::TaskExecutor::new(),
@@ -82,6 +84,7 @@ fn parallel_execution() {
 			parent_head: Default::default(),
 			relay_chain_height: 1,
 			hrmp_mqc_heads: Vec::new(),
+			dmq_mqc_head: Default::default(),
 		},
 		&execution_mode2,
 		sp_core::testing::TaskExecutor::new(),