Integrate DMP into collation, take 2 (#233)

* Integrate DMP into collation

* Integrate processed_downward_messages as well

cumulus/Cargo.lock

@@ -1074,6 +1074,7 @@ dependencies = [
  "polkadot-parachain",
  "polkadot-primitives",
  "polkadot-service",
+ "polkadot-test-client",
  "polkadot-validation",
  "sc-block-builder",
  "sc-cli",

cumulus/collator/Cargo.toml

@@ -51,6 +51,7 @@ sp-state-machine = { git = "https://github.com/paritytech/substrate", branch = "
 substrate-test-client = { git = "https://github.com/paritytech/substrate", branch = "master" }
 
 # Polkadot dependencies
+polkadot-test-client = { git = "https://github.com/paritytech/polkadot", branch = "master" }
 polkadot-node-subsystem-test-helpers = { git = "https://github.com/paritytech/polkadot", branch = "master" }
 
 # Other dependencies

cumulus/collator/src/lib.rs

@@ -68,6 +68,7 @@ pub struct Collator<Block: BlockT, PF, BI, BS, Backend> {
 	block_status: Arc<BS>,
 	wait_to_announce: Arc<Mutex<WaitToAnnounce<Block>>>,
 	backend: Arc<Backend>,
+	retrieve_dmq_contents: Arc<dyn Fn(PHash) -> Option<DownwardMessagesType> + Send + Sync>,
 }
 
 impl<Block: BlockT, PF, BI, BS, Backend> Clone for Collator<Block, PF, BI, BS, Backend> {
@@ -80,6 +81,7 @@ impl<Block: BlockT, PF, BI, BS, Backend> Clone for Collator<Block, PF, BI, BS, B
 			block_status: self.block_status.clone(),
 			wait_to_announce: self.wait_to_announce.clone(),
 			backend: self.backend.clone(),
+			retrieve_dmq_contents: self.retrieve_dmq_contents.clone(),
 		}
 	}
 }
@@ -109,6 +111,7 @@ where
 		spawner: Arc<dyn SpawnNamed + Send + Sync>,
 		announce_block: Arc<dyn Fn(Block::Hash, Vec<u8>) + Send + Sync>,
 		backend: Arc<Backend>,
+		retrieve_dmq_contents: Arc<dyn Fn(PHash) -> Option<DownwardMessagesType> + Send + Sync>,
 	) -> Self {
 		let wait_to_announce = Arc::new(Mutex::new(WaitToAnnounce::new(
 			spawner,
@@ -124,6 +127,7 @@ where
 			block_status,
 			wait_to_announce,
 			backend,
+			retrieve_dmq_contents,
 		}
 	}
 
@@ -131,7 +135,7 @@ where
 	fn inherent_data(
 		&mut self,
 		validation_data: &ValidationData,
-		downward_messages: DownwardMessagesType,
+		relay_parent: PHash,
 	) -> Option<InherentData> {
 		let mut inherent_data = self
 			.inherent_data_providers
@@ -156,6 +160,7 @@ where
 			})
 			.ok()?;
 
+		let downward_messages = (self.retrieve_dmq_contents)(relay_parent)?;
 		inherent_data
 			.put_data(DOWNWARD_MESSAGES_IDENTIFIER, &downward_messages)
 			.map_err(|e| {
@@ -242,13 +247,28 @@ where
 
 			let new_validation_code = sp_io::storage::get(well_known_keys::NEW_VALIDATION_CODE);
 
+			let processed_downward_messages = sp_io::storage::get(well_known_keys::PROCESSED_DOWNWARD_MESSAGES);
+			let processed_downward_messages = match processed_downward_messages
+				.map(|v| u32::decode(&mut &v[..]))
+			{
+				Some(Ok(processed_cnt)) => processed_cnt,
+				Some(Err(e)) => {
+					error!(
+						target: "cumulus-collator",
+						"Failed to decode the count of processed downward messages: {:?}",
+						e
+					);
+					return None
+				}
+				None => 0,
+			};
+
 			Some(Collation {
 				upward_messages,
 				new_validation_code: new_validation_code.map(Into::into),
 				head_data,
 				proof_of_validity: PoV { block_data },
-				// TODO!
+				processed_downward_messages,
-				processed_downward_messages: 0,
 				// TODO!
 				horizontal_messages: Vec::new(),
 				hrmp_watermark: relay_block_number,
@@ -297,11 +317,7 @@ where
 			})
 			.ok()?;
 
-		let inherent_data = self.inherent_data(
+		let inherent_data = self.inherent_data(&validation_data, relay_parent)?;
-			&validation_data,
-			// TODO get the downward messages
-			Vec::new(),
-		)?;
 
 		let Proposal {
 			block,
@@ -442,6 +458,26 @@ where
 	PApi: RuntimeApiCollection<StateBackend = PBackend::State>,
 	PClient: polkadot_service::AbstractClient<PBlock, PBackend, Api = PApi> + 'static,
 {
+	let retrieve_dmq_contents = {
+		let polkadot_client = polkadot_client.clone();
+		move |relay_parent: PHash| {
+			polkadot_client.runtime_api()
+				.dmq_contents_with_context(
+					&BlockId::hash(relay_parent),
+					sp_core::ExecutionContext::Importing,
+					para_id,
+				)
+				.map_err(|e| {
+					error!(
+						target: "cumulus-collator",
+						"An error occured during requesting the downward messages for {}: {:?}",
+						relay_parent, e,
+					);
+				})
+				.ok()
+		}
+	};
+
 	let follow = match cumulus_consensus::follow_polkadot(
 		para_id,
 		client,
@@ -463,6 +499,7 @@ where
 		Arc::new(spawner),
 		announce_block,
 		backend,
+		Arc::new(retrieve_dmq_contents),
 	);
 
 	let config = CollationGenerationConfig {
@@ -498,8 +535,8 @@ mod tests {
 	use sp_runtime::traits::DigestFor;
 
 	use cumulus_test_client::{
-		generate_block_inherents, Client, DefaultTestClientBuilderExt, NativeExecutor,
+		generate_block_inherents, Client, DefaultTestClientBuilderExt,
-		TestClientBuilder, TestClientBuilderExt, WasmExecutionMethod::Interpreted,
+		TestClientBuilder, TestClientBuilderExt,
 	};
 	use cumulus_test_runtime::{Block, Header};
 
@@ -594,6 +631,20 @@ mod tests {
 
 		spawner.spawn("overseer", overseer.run().then(|_| async { () }).boxed());
 
+		let (polkadot_client, relay_parent) = {
+			// Create a polkadot client with a block imported.
+			use polkadot_test_client::{
+				TestClientBuilderExt as _, DefaultTestClientBuilderExt as _,
+				InitPolkadotBlockBuilder as _, ClientBlockImportExt as _
+			};
+			let mut client = polkadot_test_client::TestClientBuilder::new().build();
+			let block_builder = client.init_polkadot_block_builder();
+			let block = block_builder.build().expect("Finalizes the block").block;
+			let hash = block.header().hash();
+			client.import_as_best(BlockOrigin::Own, block).expect("Imports the block");
+			(client, hash)
+		};
+
 		let collator_start =
 			start_collator::<_, _, _, _, _, _, _, _, polkadot_service::FullBackend, _>(
 				StartCollatorParams {
@@ -608,17 +659,7 @@ mod tests {
 					spawner,
 					para_id,
 					key: CollatorPair::generate().0,
-					polkadot_client: Arc::new(
+					polkadot_client: Arc::new(polkadot_client,),
-						substrate_test_client::TestClientBuilder::<_, _, _, ()>::default()
-							.build_with_native_executor::<polkadot_service::polkadot_runtime::RuntimeApi, _>(
-								Some(NativeExecutor::<polkadot_service::PolkadotExecutor>::new(
-									Interpreted,
-									None,
-									1,
-								)),
-							)
-							.0,
-					),
 				},
 			);
 		block_on(collator_start).expect("Should start collator");
@@ -634,7 +675,7 @@ mod tests {
 		let mut validation_data = ValidationData::default();
 		validation_data.persisted.parent_head = header.encode().into();
 
-		let collation = block_on((config.collator)(Default::default(), &validation_data))
+		let collation = block_on((config.collator)(relay_parent, &validation_data))
 			.expect("Collation is build");
 
 		let block_data = collation.proof_of_validity.block_data;

cumulus/primitives/src/lib.rs

@@ -19,7 +19,7 @@
 #![cfg_attr(not(feature = "std"), no_std)]
 
 pub use polkadot_core_primitives as relay_chain;
-pub use polkadot_core_primitives::DownwardMessage;
+pub use polkadot_core_primitives::InboundDownwardMessage;
 /// A generic upward message from a Parachain to the Relay Chain.
 ///
 /// It is "generic" in such a way, that the actual message is encoded in the `data` field.
@@ -42,7 +42,7 @@ pub mod inherents {
 	pub const DOWNWARD_MESSAGES_IDENTIFIER: InherentIdentifier = *b"cumdownm";
 
 	/// The type of the inherent downward messages.
-	pub type DownwardMessagesType = sp_std::vec::Vec<crate::DownwardMessage>;
+	pub type DownwardMessagesType = sp_std::vec::Vec<crate::InboundDownwardMessage>;
 
 	/// The identifier for the `set_validation_data` inherent.
 	pub const VALIDATION_DATA_IDENTIFIER: InherentIdentifier = *b"valfunp0";
@@ -73,7 +73,7 @@ pub mod well_known_keys {
 #[impl_trait_for_tuples::impl_for_tuples(30)]
 pub trait DownwardMessageHandler {
 	/// Handle the given downward message.
-	fn handle_downward_message(msg: &DownwardMessage);
+	fn handle_downward_message(msg: &InboundDownwardMessage);
 }
 
 /// A trait which is called when the validation data is set.

cumulus/runtime/src/validate_block/implementation.rs

@@ -155,7 +155,10 @@ pub fn validate_block<B: BlockT, E: ExecuteBlock<B>>(params: ValidationParams) -
 
 	let processed_downward_messages = overlay.storage(PROCESSED_DOWNWARD_MESSAGES)
 		.flatten()
-		.and_then(|v| Decode::decode(&mut &v[..]).ok())
+		.map(|v|
+			Decode::decode(&mut &v[..])
+				.expect("Processed downward message count is not correctly encoded in the storage")
+		)
 		.unwrap_or_default();
 
 	let validation_data: ValidationData = overlay.storage(VALIDATION_DATA).flatten()