Implement HRMP

core-primitives/src/lib.rs

@@ -102,6 +102,26 @@ pub struct InboundDownwardMessage<BlockNumber = crate::BlockNumber> {
 	pub msg: DownwardMessage,
 }
 
+/// An HRMP message seen from the perspective of a recipient.
+#[derive(codec::Encode, codec::Decode, Clone, sp_runtime::RuntimeDebug, PartialEq)]
+pub struct InboundHrmpMessage<BlockNumber = crate::BlockNumber> {
+	/// 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: sp_std::vec::Vec<u8>,
+}
+
+/// An HRMP message seen from the perspective of a sender.
+#[derive(codec::Encode, codec::Decode, Clone, sp_runtime::RuntimeDebug, PartialEq, Eq, Hash)]
+pub struct OutboundHrmpMessage<Id> {
+	/// The para that will get this message in its downward message queue.
+	pub recipient: Id,
+	/// The message payload.
+	pub data: sp_std::vec::Vec<u8>,
+}
+
 /// V1 primitives.
 pub mod v1 {
 	pub use super::*;

node/collation-generation/src/lib.rs

@@ -274,10 +274,12 @@ async fn handle_new_activations<Context: SubsystemContext>(
 
 				let commitments = CandidateCommitments {
 					upward_messages: collation.upward_messages,
+					horizontal_messages: collation.horizontal_messages,
 					new_validation_code: collation.new_validation_code,
 					head_data: collation.head_data,
 					erasure_root,
 					processed_downward_messages: collation.processed_downward_messages,
+					hrmp_watermark: collation.hrmp_watermark,
 				};
 
 				let ccr = CandidateReceipt {
@@ -382,12 +384,14 @@ mod tests {
 		fn test_collation() -> Collation {
 			Collation {
 				upward_messages: Default::default(),
+				horizontal_messages: Default::default(),
 				new_validation_code: Default::default(),
 				head_data: Default::default(),
 				proof_of_validity: PoV {
 					block_data: BlockData(Vec::new()),
 				},
 				processed_downward_messages: Default::default(),
+				hrmp_watermark: Default::default(),
 			}
 		}
 

node/core/backing/src/lib.rs

@@ -717,10 +717,12 @@ impl CandidateBackingJob {
 
 		let commitments = CandidateCommitments {
 			upward_messages: outputs.upward_messages,
+			horizontal_messages: outputs.horizontal_messages,
 			erasure_root,
 			new_validation_code: outputs.new_validation_code,
 			head_data: outputs.head_data,
 			processed_downward_messages: outputs.processed_downward_messages,
+			hrmp_watermark: outputs.hrmp_watermark,
 		};
 
 		let res = match with_commitments(commitments) {
@@ -1200,9 +1202,11 @@ mod tests {
 					tx.send(Ok(
 						ValidationResult::Valid(ValidationOutputs {
 							head_data: expected_head_data.clone(),
+							horizontal_messages: Vec::new(),
 							upward_messages: Vec::new(),
 							new_validation_code: None,
 							processed_downward_messages: 0,
+							hrmp_watermark: 0,
 						}, test_state.validation_data.persisted),
 					)).unwrap();
 				}
@@ -1328,8 +1332,10 @@ mod tests {
 						ValidationResult::Valid(ValidationOutputs {
 							head_data: expected_head_data.clone(),
 							upward_messages: Vec::new(),
+							horizontal_messages: Vec::new(),
 							new_validation_code: None,
 							processed_downward_messages: 0,
+							hrmp_watermark: 0,
 						}, test_state.validation_data.persisted),
 					)).unwrap();
 				}
@@ -1477,8 +1483,10 @@ mod tests {
 						ValidationResult::Valid(ValidationOutputs {
 							head_data: expected_head_data.clone(),
 							upward_messages: Vec::new(),
+							horizontal_messages: Vec::new(),
 							new_validation_code: None,
 							processed_downward_messages: 0,
+							hrmp_watermark: 0,
 						}, test_state.validation_data.persisted),
 					)).unwrap();
 				}
@@ -1660,8 +1668,10 @@ mod tests {
 						ValidationResult::Valid(ValidationOutputs {
 							head_data: expected_head_data.clone(),
 							upward_messages: Vec::new(),
+							horizontal_messages: Vec::new(),
 							new_validation_code: None,
 							processed_downward_messages: 0,
+							hrmp_watermark: 0,
 						}, test_state.validation_data.persisted),
 					)).unwrap();
 				}

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

@@ -489,8 +489,10 @@ fn validate_candidate_exhaustive<B: ValidationBackend, S: SpawnNamed + 'static>(
 			let outputs = ValidationOutputs {
 				head_data: res.head_data,
 				upward_messages: res.upward_messages,
+				horizontal_messages: res.horizontal_messages,
 				new_validation_code: res.new_validation_code,
 				processed_downward_messages: res.processed_downward_messages,
+				hrmp_watermark: res.hrmp_watermark,
 			};
 			Ok(ValidationResult::Valid(outputs, persisted_validation_data))
 		}
@@ -833,7 +835,9 @@ mod tests {
 			head_data: HeadData(vec![1, 1, 1]),
 			new_validation_code: Some(vec![2, 2, 2].into()),
 			upward_messages: Vec::new(),
+			horizontal_messages: Vec::new(),
 			processed_downward_messages: 0,
+			hrmp_watermark: 0,
 		};
 
 		let v = validate_candidate_exhaustive::<MockValidationBackend, _>(
@@ -848,7 +852,9 @@ mod tests {
 		assert_matches!(v, ValidationResult::Valid(outputs, used_validation_data) => {
 			assert_eq!(outputs.head_data, HeadData(vec![1, 1, 1]));
 			assert_eq!(outputs.upward_messages, Vec::<UpwardMessage>::new());
+			assert_eq!(outputs.horizontal_messages, Vec::new());
 			assert_eq!(outputs.new_validation_code, Some(vec![2, 2, 2].into()));
+			assert_eq!(outputs.hrmp_watermark, 0);
 			assert_eq!(used_validation_data, validation_data);
 		});
 	}

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

@@ -130,6 +130,7 @@ fn make_runtime_api_request<Client>(
 		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),
+		Request::InboundHrmpChannelsContents(id, sender) => query!(inbound_hrmp_channels_contents(id), sender),
 	}
 }
 
@@ -180,12 +181,11 @@ mod tests {
 		ValidatorId, ValidatorIndex, GroupRotationInfo, CoreState, PersistedValidationData,
 		Id as ParaId, OccupiedCoreAssumption, ValidationData, SessionIndex, ValidationCode,
 		CommittedCandidateReceipt, CandidateEvent, AuthorityDiscoveryId, InboundDownwardMessage,
-		BlockNumber,
+		BlockNumber, InboundHrmpMessage,
 	};
 	use polkadot_node_subsystem_test_helpers as test_helpers;
 	use sp_core::testing::TaskExecutor;
-
-	use std::collections::HashMap;
+	use std::collections::{HashMap, BTreeMap};
 	use futures::channel::oneshot;
 
 	#[derive(Default, Clone)]
@@ -201,6 +201,7 @@ mod tests {
 		candidate_pending_availability: HashMap<ParaId, CommittedCandidateReceipt>,
 		candidate_events: Vec<CandidateEvent>,
 		dmq_contents: HashMap<ParaId, Vec<InboundDownwardMessage>>,
+		hrmp_channels: HashMap<ParaId, BTreeMap<ParaId, Vec<InboundHrmpMessage>>>,
 	}
 
 	impl ProvideRuntimeApi<Block> for MockRuntimeApi {
@@ -306,9 +307,16 @@ mod tests {
 			fn dmq_contents(
 				&self,
 				recipient: ParaId,
-			) -> Vec<polkadot_primitives::v1::InboundDownwardMessage> {
+			) -> Vec<InboundDownwardMessage> {
 				self.dmq_contents.get(&recipient).map(|q| q.clone()).unwrap_or_default()
 			}
+
+			fn inbound_hrmp_channels_contents(
+				&self,
+				recipient: ParaId
+			) -> BTreeMap<ParaId, Vec<InboundHrmpMessage>> {
+				self.hrmp_channels.get(&recipient).map(|q| q.clone()).unwrap_or_default()
+			}
 		}
 	}
 
@@ -701,6 +709,72 @@ mod tests {
 		futures::executor::block_on(future::join(subsystem_task, test_task));
 	}
 
+	#[test]
+	fn requests_inbound_hrmp_channels_contents() {
+		let (ctx, mut ctx_handle) = test_helpers::make_subsystem_context(TaskExecutor::new());
+
+		let relay_parent = [1; 32].into();
+		let para_a = 99.into();
+		let para_b = 66.into();
+		let para_c = 33.into();
+
+		let para_b_inbound_channels = [
+			(para_a, vec![]),
+			(
+				para_c,
+				vec![InboundHrmpMessage {
+					sent_at: 1,
+					data: "𝙀=𝙈𝘾²".as_bytes().to_owned(),
+				}],
+			),
+		]
+		.iter()
+		.cloned()
+		.collect::<BTreeMap<_, _>>();
+
+		let runtime_api = Arc::new({
+			let mut runtime_api = MockRuntimeApi::default();
+
+			runtime_api.hrmp_channels.insert(para_a, BTreeMap::new());
+			runtime_api
+				.hrmp_channels
+				.insert(para_b, para_b_inbound_channels.clone());
+
+			runtime_api
+		});
+
+		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::InboundHrmpChannelsContents(para_a, tx),
+					),
+				})
+				.await;
+			assert_eq!(rx.await.unwrap().unwrap(), BTreeMap::new());
+
+			let (tx, rx) = oneshot::channel();
+			ctx_handle
+				.send(FromOverseer::Communication {
+					msg: RuntimeApiMessage::Request(
+						relay_parent,
+						Request::InboundHrmpChannelsContents(para_b, tx),
+					),
+				})
+				.await;
+			assert_eq!(rx.await.unwrap().unwrap(), para_b_inbound_channels,);
+
+			ctx_handle
+				.send(FromOverseer::Signal(OverseerSignal::Conclude))
+				.await;
+		};
+		futures::executor::block_on(future::join(subsystem_task, test_task));
+	}
+
 	#[test]
 	fn requests_historical_code() {
 		let (ctx, mut ctx_handle) = test_helpers::make_subsystem_context(TaskExecutor::new());

node/primitives/src/lib.rs

@@ -28,7 +28,7 @@ use polkadot_primitives::v1::{
 	Hash, CommittedCandidateReceipt, CandidateReceipt, CompactStatement,
 	EncodeAs, Signed, SigningContext, ValidatorIndex, ValidatorId,
 	UpwardMessage, ValidationCode, PersistedValidationData, ValidationData,
-	HeadData, PoV, CollatorPair, Id as ParaId, ValidationOutputs, CandidateHash,
+	HeadData, PoV, CollatorPair, Id as ParaId, OutboundHrmpMessage, ValidationOutputs, CandidateHash,
 };
 use polkadot_statement_table::{
 	generic::{
@@ -252,9 +252,11 @@ impl std::convert::TryFrom<FromTableMisbehavior> for MisbehaviorReport {
 /// - does not contain the erasure root; that's computed at the Polkadot level, not at Cumulus
 /// - contains a proof of validity.
 #[derive(Clone, Encode, Decode)]
-pub struct Collation {
+pub struct Collation<BlockNumber = polkadot_primitives::v1::BlockNumber> {
 	/// Messages destined to be interpreted by the Relay chain itself.
 	pub upward_messages: Vec<UpwardMessage>,
+	/// The horizontal messages sent by the parachain.
+	pub horizontal_messages: Vec<OutboundHrmpMessage<ParaId>>,
 	/// New validation code.
 	pub new_validation_code: Option<ValidationCode>,
 	/// The head-data produced as a result of execution.
@@ -263,6 +265,8 @@ pub struct Collation {
 	pub proof_of_validity: PoV,
 	/// The number of messages processed from the DMQ.
 	pub processed_downward_messages: u32,
+	/// The mark which specifies the block number up to which all inbound HRMP messages are processed.
+	pub hrmp_watermark: BlockNumber,
 }
 
 /// Configuration for the collation generator

node/subsystem/src/messages.rs

@@ -37,9 +37,10 @@ use polkadot_primitives::v1::{
 	GroupRotationInfo, Hash, Id as ParaId, OccupiedCoreAssumption,
 	PersistedValidationData, PoV, SessionIndex, SignedAvailabilityBitfield,
 	ValidationCode, ValidatorId, ValidationData, CandidateHash,
-	ValidatorIndex, ValidatorSignature, InboundDownwardMessage,
+	ValidatorIndex, ValidatorSignature, InboundDownwardMessage, InboundHrmpMessage,
 };
 use std::sync::Arc;
+use std::collections::btree_map::BTreeMap;
 
 /// A notification of a new backed candidate.
 #[derive(Debug)]
@@ -452,6 +453,12 @@ pub enum RuntimeApiRequest {
 		ParaId,
 		RuntimeApiSender<Vec<InboundDownwardMessage<BlockNumber>>>,
 	),
+	/// Get the contents of all channels addressed to the given recipient. Channels that have no
+	/// messages in them are also included.
+	InboundHrmpChannelsContents(
+		ParaId,
+		RuntimeApiSender<BTreeMap<ParaId, Vec<InboundHrmpMessage<BlockNumber>>>>,
+	),
 }
 
 /// A message to the Runtime API subsystem.

parachain/src/primitives.rs

@@ -28,7 +28,7 @@ use serde::{Serialize, Deserialize};
 #[cfg(feature = "std")]
 use sp_core::bytes;
 
-use polkadot_core_primitives::Hash;
+use polkadot_core_primitives::{Hash, OutboundHrmpMessage};
 
 /// Block number type used by the relay chain.
 pub use polkadot_core_primitives::BlockNumber as RelayChainBlockNumber;
@@ -186,6 +186,21 @@ impl<T: Encode + Decode + Default> AccountIdConversion<T> for Id {
 	}
 }
 
+/// A type that uniquely identifies an HRMP channel. An HRMP channel is established between two paras.
+/// In text, we use the notation `(A, B)` to specify a channel between A and B. The channels are
+/// unidirectional, meaning that `(A, B)` and `(B, A)` refer to different channels. The convention is
+/// that we use the first item tuple for the sender and the second for the recipient. Only one channel
+/// is allowed between two participants in one direction, i.e. there cannot be 2 different channels
+/// identified by `(A, B)`.
+#[derive(Clone, PartialEq, Eq, Encode, Decode, RuntimeDebug)]
+#[cfg_attr(feature = "std", derive(Hash))]
+pub struct HrmpChannelId {
+	/// The para that acts as the sender in this channel.
+	pub sender: Id,
+	/// The para that acts as the recipient in this channel.
+	pub recipient: Id,
+}
+
 /// A message from a parachain to its Relay Chain.
 pub type UpwardMessage = Vec<u8>;
 
@@ -212,7 +227,7 @@ pub struct ValidationParams {
 }
 
 /// The result of parachain validation.
-// TODO: egress and balance uploads (https://github.com/paritytech/polkadot/issues/220)
+// TODO: balance uploads (https://github.com/paritytech/polkadot/issues/220)
 #[derive(PartialEq, Eq, Encode)]
 #[cfg_attr(feature = "std", derive(Debug, Decode))]
 pub struct ValidationResult {
@@ -222,8 +237,12 @@ pub struct ValidationResult {
 	pub new_validation_code: Option<ValidationCode>,
 	/// Upward messages send by the Parachain.
 	pub upward_messages: Vec<UpwardMessage>,
+	/// Outbound horizontal messages sent by the parachain.
+	pub horizontal_messages: Vec<OutboundHrmpMessage<Id>>,
 	/// Number of downward messages that were processed by the Parachain.
 	///
 	/// It is expected that the Parachain processes them from first to last.
 	pub processed_downward_messages: u32,
+	/// The mark which specifies the block number up to which all inbound HRMP messages are processed.
+	pub hrmp_watermark: RelayChainBlockNumber,
 }

parachain/test-parachains/adder/collator/src/lib.rs

@@ -114,10 +114,12 @@ impl Collator {
 
 			let collation = Collation {
 				upward_messages: Vec::new(),
+				horizontal_messages: Vec::new(),
 				new_validation_code: None,
 				head_data: head_data.encode().into(),
 				proof_of_validity: PoV { block_data: block_data.encode().into() },
 				processed_downward_messages: 0,
+				hrmp_watermark: validation_data.persisted.block_number,
 			};
 
 			async move { Some(collation) }.boxed()