Track and accumulate ingress roots in runtime (#287)

* track unrouted ingress in runtime

* test ingress routing

* fix compilation

* add space

Co-Authored-By: Gavin Wood <github@gavwood.com>

network/src/lib.rs

@@ -30,7 +30,7 @@ use futures::sync::oneshot;
 use polkadot_primitives::{Block, SessionKey, Hash, Header};
 use polkadot_primitives::parachain::{
 	Id as ParaId, BlockData, CollatorId, CandidateReceipt, Collation, PoVBlock,
-	ConsolidatedIngressRoots,
+	StructuredUnroutedIngress,
 };
 use substrate_network::{PeerId, RequestId, Context};
 use substrate_network::{message, generic_message};
@@ -83,7 +83,7 @@ struct PoVBlockRequest {
 	candidate_hash: Hash,
 	block_data_hash: Hash,
 	sender: oneshot::Sender<PoVBlock>,
-	canon_roots: ConsolidatedIngressRoots,
+	canon_roots: StructuredUnroutedIngress,
 }
 
 impl PoVBlockRequest {
@@ -218,7 +218,7 @@ impl PolkadotProtocol {
 		ctx: &mut Context<Block>,
 		candidate: &CandidateReceipt,
 		relay_parent: Hash,
-		canon_roots: ConsolidatedIngressRoots,
+		canon_roots: StructuredUnroutedIngress,
 	) -> oneshot::Receiver<PoVBlock> {
 		let (tx, rx) = oneshot::channel();
 
@@ -547,7 +547,7 @@ impl Specialization<Block> for PolkadotProtocol {
 							validation_session_parent: Default::default(),
 							candidate_hash: Default::default(),
 							block_data_hash: Default::default(),
-							canon_roots: ConsolidatedIngressRoots(Vec::new()),
+							canon_roots: StructuredUnroutedIngress(Vec::new()),
 							sender,
 						}));
 					}

network/src/tests/mod.rs

@@ -24,7 +24,7 @@ use polkadot_validation::GenericStatement;
 use polkadot_primitives::{Block, Hash, SessionKey};
 use polkadot_primitives::parachain::{
 	CandidateReceipt, HeadData, PoVBlock, BlockData, CollatorId, ValidatorId,
-	ConsolidatedIngressRoots,
+	StructuredUnroutedIngress,
 };
 use substrate_primitives::crypto::UncheckedInto;
 use parity_codec::Encode;
@@ -175,7 +175,7 @@ fn fetches_from_those_with_knowledge() {
 	let knowledge = session.knowledge();
 
 	knowledge.lock().note_statement(a_key.clone(), &GenericStatement::Valid(candidate_hash));
-	let canon_roots = ConsolidatedIngressRoots(Vec::new());
+	let canon_roots = StructuredUnroutedIngress(Vec::new());
 	let recv = protocol.fetch_pov_block(
 		&mut TestContext::default(),
 		&candidate_receipt,

network/src/tests/validation.rs

@@ -29,7 +29,7 @@ use polkadot_validation::{SharedTable, MessagesFrom, Network};
 use polkadot_primitives::{SessionKey, Block, Hash, Header, BlockId};
 use polkadot_primitives::parachain::{
 	Id as ParaId, Chain, DutyRoster, ParachainHost, OutgoingMessage,
-	ValidatorId, ConsolidatedIngressRoots,
+	ValidatorId, StructuredUnroutedIngress, BlockIngressRoots,
 };
 use parking_lot::Mutex;
 use substrate_client::error::Result as ClientResult;
@@ -175,7 +175,7 @@ struct ApiData {
 	validators: Vec<ValidatorId>,
 	duties: Vec<Chain>,
 	active_parachains: Vec<ParaId>,
-	ingress: HashMap<ParaId, ConsolidatedIngressRoots>,
+	ingress: HashMap<ParaId, StructuredUnroutedIngress>,
 }
 
 #[derive(Default, Clone)]
@@ -306,7 +306,7 @@ impl ParachainHost<Block> for RuntimeApi {
 		_: ExecutionContext,
 		id: Option<ParaId>,
 		_: Vec<u8>,
-	) -> ClientResult<NativeOrEncoded<Option<ConsolidatedIngressRoots>>> {
+	) -> ClientResult<NativeOrEncoded<Option<StructuredUnroutedIngress>>> {
 		let id = id.unwrap();
 		Ok(NativeOrEncoded::Native(self.data.lock().ingress.get(&id).cloned()))
 	}
@@ -372,7 +372,7 @@ impl IngressBuilder {
 		}
 	}
 
-	fn build(self) -> HashMap<ParaId, ConsolidatedIngressRoots> {
+	fn build(self) -> HashMap<ParaId, BlockIngressRoots> {
 		let mut map = HashMap::new();
 		for ((source, target), messages) in self.egress {
 			map.entry(target).or_insert_with(Vec::new)
@@ -383,7 +383,7 @@ impl IngressBuilder {
 			roots.sort_by_key(|&(para_id, _)| para_id);
 		}
 
-		map.into_iter().map(|(k, v)| (k, ConsolidatedIngressRoots(v))).collect()
+		map.into_iter().map(|(k, v)| (k, BlockIngressRoots(v))).collect()
 	}
 }
 

primitives/src/parachain.rs

@@ -19,7 +19,7 @@
 use rstd::prelude::*;
 use rstd::cmp::Ordering;
 use parity_codec::{Encode, Decode};
-use super::{Hash, Balance};
+use super::{Hash, Balance, BlockNumber};
 
 #[cfg(feature = "std")]
 use serde::{Serialize, Deserialize};
@@ -200,26 +200,43 @@ pub struct PoVBlock {
 #[cfg_attr(feature = "std", derive(Serialize, Deserialize, Encode, Debug))]
 pub struct Message(#[cfg_attr(feature = "std", serde(with="bytes"))] pub Vec<u8>);
 
-/// Consolidated ingress roots.
+/// All ingress roots at one block.
 ///
-/// This is an ordered vector of other parachains' egress queue roots,
-/// obtained according to the routing rules. The same parachain may appear
-/// twice.
+/// This is an ordered vector of other parachain's egress queue roots from a specific block.
+/// empty roots are omitted. Each parachain may appear once at most.
+#[derive(Default, PartialEq, Eq, Clone, Encode)]
+#[cfg_attr(feature = "std", derive(Serialize, Deserialize, Debug, Decode))]
+pub struct BlockIngressRoots(pub Vec<(Id, Hash)>);
+
+/// All ingress roots, grouped by block number (ascending). To properly
+/// interpret this struct, the user must have knowledge of which fork of the relay
+/// chain all block numbers correspond to.
 #[derive(Default, PartialEq, Eq, Clone, Encode)]
 #[cfg_attr(feature = "std", derive(Serialize, Deserialize, Debug, Decode))]
-pub struct ConsolidatedIngressRoots(pub Vec<(Id, Hash)>);
+pub struct StructuredUnroutedIngress(pub Vec<(BlockNumber, BlockIngressRoots)>);
+
+#[cfg(feature = "std")]
+impl StructuredUnroutedIngress {
+	/// Get the length of all the ingress roots across all blocks.
+	pub fn len(&self) -> usize {
+		self.0.iter().fold(0, |a, (_, roots)| a + roots.0.len())
+	}
 
-impl From<Vec<(Id, Hash)>> for ConsolidatedIngressRoots {
-	fn from(v: Vec<(Id, Hash)>) -> Self {
-		ConsolidatedIngressRoots(v)
+	/// Returns an iterator over all ingress roots. The block number indicates
+	/// the height at which that root was posted to the relay chain. The parachain ID is the
+	/// message sender.
+	pub fn iter(&self) -> impl Iterator<Item=(BlockNumber, &Id, &Hash)> {
+		self.0.iter().flat_map(|&(n, ref roots)|
+			roots.0.iter().map(move |&(ref from, ref root)| (n, from, root))
+		)
 	}
 }
 
 /// Consolidated ingress queue data.
 ///
 /// This is just an ordered vector of other parachains' egress queues,
 /// obtained according to the routing rules. The same parachain may appear
-/// twice.
+/// more than once.
 #[derive(Default, PartialEq, Eq, Clone, Decode)]
 #[cfg_attr(feature = "std", derive(Serialize, Deserialize, Encode, Debug))]
 pub struct ConsolidatedIngress(pub Vec<(Id, Vec<Message>)>);
@@ -324,9 +341,9 @@ substrate_client::decl_runtime_apis! {
 		fn parachain_head(id: Id) -> Option<Vec<u8>>;
 		/// Get the given parachain's head code blob.
 		fn parachain_code(id: Id) -> Option<Vec<u8>>;
-		/// Get the ingress roots to a specific parachain at a
-		/// block.
-		fn ingress(to: Id) -> Option<ConsolidatedIngressRoots>;
+		/// Get all the unrouted ingress roots at the given block that
+		/// are targeting the given parachain.
+		fn ingress(to: Id) -> Option<StructuredUnroutedIngress>;
 	}
 }
 

runtime/src/lib.rs

@@ -356,8 +356,8 @@ impl_runtime_apis! {
 		fn parachain_code(id: parachain::Id) -> Option<Vec<u8>> {
 			Parachains::parachain_code(&id)
 		}
-		fn ingress(to: parachain::Id) -> Option<parachain::ConsolidatedIngressRoots> {
-			Parachains::ingress(to).map(Into::into)
+		fn ingress(to: parachain::Id) -> Option<parachain::StructuredUnroutedIngress> {
+			Parachains::ingress(to).map(parachain::StructuredUnroutedIngress)
 		}
 	}