Minimal collation work-flow and necessary traits

Cargo.toml

@@ -10,6 +10,7 @@ polkadot-cli = { path = "cli", version = "0.1" }
 [workspace]
 members = [
 	"client",
+	"collator",
 	"contracts",
 	"primitives",
 	"rpc",

collator/Cargo.toml

@@ -0,0 +1,9 @@
+[package]
+name = "polkadot-collator"
+version = "0.1.0"
+authors = ["Parity Technologies <[email protected]>"]
+description = "Abstract collation logic"
+
+[dependencies]
+polkadot-primitives = { path = "../primitives", version = "0.1" }
+futures = "0.1.17"

collator/src/lib.rs

@@ -0,0 +1,215 @@
+// Copyright 2017 Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Polkadot is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Collation Logic.
+//!
+//! A collator node lives on a distinct parachain and submits a proposal for
+//! a state transition, along with a proof for its validity
+//! (what we might call a witness or block data).
+//!
+//! One of collators' other roles is to route messages between chains.
+//! Each parachain produces a list of "egress" posts of messages for each other
+//! parachain on each block, for a total of N^2 lists all together.
+//!
+//! We will refer to the egress list at relay chain block X of parachain A with
+//! destination B as egress(X)[A -> B]
+//!
+//! On every block, each parachain will be intended to route messages from some
+//! subset of all the other parachains.
+//!
+//! Since the egress information is unique to every block, when routing from a
+//! parachain a collator must gather all egress posts from that parachain
+//! up to the last point in history that messages were successfully routed
+//! from that parachain, accounting for relay chain blocks where no candidate
+//! from the collator's parachain was produced.
+//!
+//! In the case that all parachains route to each other and a candidate for the
+//! collator's parachain was included in the last relay chain block, the collator
+//! only has to gather egress posts from other parachains one block back in relay
+//! chain history.
+//!
+//! This crate defines traits which provide context necessary for collation logic
+//! to be performed, as the collation logic itself.
+
+extern crate futures;
+extern crate polkadot_primitives as primitives;
+
+use std::collections::{BTreeSet, BTreeMap};
+
+use futures::{stream, Stream, Future, IntoFuture};
+use primitives::parachain::{self, ConsolidatedIngress, Message, Id as ParaId};
+
+/// Parachain context needed for collation.
+///
+/// This can be implemented through an externally attached service or a stub.
+pub trait ParachainContext {
+	/// Produce a candidate, given the latest ingress queue information.
+	fn produce_candidate<I: IntoIterator<Item=(ParaId, Message)>>(
+		&self,
+		ingress: I,
+	) -> parachain::RawProof;
+}
+
+/// Relay chain context needed to collate.
+/// This encapsulates a network and local database which may store
+/// some of the input.
+pub trait RelayChainContext {
+	type Error;
+
+	/// Future that resolves to the un-routed egress queues of a parachain.
+	/// The first item is the oldest.
+	type FutureEgress: IntoFuture<Item=Vec<Vec<Message>>, Error=Self::Error>;
+
+	/// Provide a set of all parachains meant to be routed to at a block.
+	fn routing_parachains(&self) -> BTreeSet<ParaId>;
+
+	/// Get un-routed egress queues from a parachain to the local parachain.
+	fn unrouted_egress(&self, id: ParaId) -> Self::FutureEgress;
+}
+
+/// Collate the necessary ingress queue using the given context.
+// TODO: impl trait
+pub fn collate_ingress<'a, R>(relay_context: R)
+	-> Box<Future<Item=ConsolidatedIngress, Error=R::Error> + 'a>
+	where R: RelayChainContext,
+		  R::Error: 'a,
+		  R::FutureEgress: 'a
+{
+	let mut egress_fetch = Vec::new();
+
+	for routing_parachain in relay_context.routing_parachains() {
+		let fetch = relay_context
+			.unrouted_egress(routing_parachain)
+			.into_future()
+			.map(move |egresses| (routing_parachain, egresses));
+
+		egress_fetch.push(fetch);
+	}
+
+	// create a map ordered first by the depth of the egress queue
+	// and then by the parachain ID.
+	//
+	// then transform that into the consolidated egress queue.
+	let future = stream::futures_unordered(egress_fetch)
+		.fold(BTreeMap::new(), |mut map, (routing_id, egresses)| {
+			for (depth, egress) in egresses.into_iter().rev().enumerate() {
+				let depth = -(depth as i64);
+				map.insert((depth, routing_id), egress);
+			}
+
+			Ok(map)
+		})
+		.map(|ordered| ordered.into_iter().map(|((_, id), egress)| (id, egress)))
+		.map(|i| i.collect::<Vec<_>>())
+		.map(ConsolidatedIngress);
+
+	Box::new(future)
+}
+
+/// Produce a candidate for the parachain.
+pub fn collate<'a, R, P>(local_id: ParaId, relay_context: R, para_context: P)
+	-> Box<Future<Item=parachain::Candidate, Error=R::Error> + 'a>
+	where R: RelayChainContext,
+	      R::Error: 'a,
+		  R::FutureEgress: 'a,
+		  P: ParachainContext + 'a,
+{
+	Box::new(collate_ingress(relay_context).map(move |ingress| {
+		let block_data = para_context.produce_candidate(
+			ingress.0.iter().flat_map(|&(id, ref msgs)| msgs.iter().cloned().map(move |msg| (id, msg)))
+		);
+
+		parachain::Candidate {
+			id: local_id,
+			ingress: ingress,
+			proof: block_data,
+		}
+	}))
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	use std::collections::{HashMap, BTreeSet};
+
+	use futures::Future;
+	use primitives::parachain::{Message, Id as ParaId};
+
+	pub struct DummyRelayChainCtx {
+		egresses: HashMap<ParaId, Vec<Vec<Message>>>,
+		currently_routing: BTreeSet<ParaId>,
+	}
+
+	impl RelayChainContext for DummyRelayChainCtx {
+		type Error = ();
+		type FutureEgress = Result<Vec<Vec<Message>>, ()>;
+
+		fn routing_parachains(&self) -> BTreeSet<ParaId> {
+			self.currently_routing.clone()
+		}
+
+		fn unrouted_egress(&self, id: ParaId) -> Result<Vec<Vec<Message>>, ()> {
+			Ok(self.egresses.get(&id).cloned().unwrap_or_default())
+		}
+	}
+
+    #[test]
+	fn collates_ingress() {
+		let route_from = |x: &[ParaId]| {
+			 let mut set = BTreeSet::new();
+			 set.extend(x.iter().cloned());
+			 set
+		};
+
+		let message = |x: Vec<u8>| vec![Message(x)];
+
+		let dummy_ctx = DummyRelayChainCtx {
+			currently_routing: route_from(&[2.into(), 3.into()]),
+			egresses: vec![
+				// egresses for `2`: last routed successfully 5 blocks ago.
+				(2.into(), vec![
+					message(vec![1, 2, 3]),
+					message(vec![4, 5, 6]),
+					message(vec![7, 8]),
+					message(vec![10]),
+					message(vec![12]),
+				]),
+
+				// egresses for `3`: last routed successfully 3 blocks ago.
+				(3.into(), vec![
+					message(vec![9]),
+					message(vec![11]),
+					message(vec![13]),
+				]),
+			].into_iter().collect(),
+		};
+
+		assert_eq!(
+			collate_ingress(dummy_ctx).wait().unwrap(),
+			ConsolidatedIngress(vec![
+				(2.into(), message(vec![1, 2, 3])),
+				(2.into(), message(vec![4, 5, 6])),
+				(2.into(), message(vec![7, 8])),
+				(3.into(), message(vec![9])),
+				(2.into(), message(vec![10])),
+				(3.into(), message(vec![11])),
+				(2.into(), message(vec![12])),
+				(3.into(), message(vec![13])),
+			]
+		))
+	}
+}

primitives/src/block.rs

@@ -46,7 +46,7 @@ pub struct Header {
 #[serde(deny_unknown_fields)]
 pub struct Body {
 	/// Parachain proposal blocks.
-	pub para_blocks: Vec<parachain::Proposal>,
+	pub para_blocks: Vec<parachain::Candidate>,
 }
 
 #[cfg(test)]
@@ -73,18 +73,18 @@ mod tests {
 	fn test_body_serialization() {
 		assert_eq!(ser::to_string_pretty(&Body {
 			para_blocks: vec![
-				parachain::Proposal {
-					parachain: 5.into(),
-					header: parachain::Header(vec![1, 2, 3, 4]),
-					proof_hash: 5.into(),
+				parachain::Candidate {
+					id: 10.into(),
+					ingress: Default::default(),
+					proof: ::parachain::RawProof(vec![1, 3, 5, 8]),
 				}
 			],
 		}), r#"{
   "paraBlocks": [
     {
-      "parachain": 5,
-      "header": "0x01020304",
-      "proofHash": "0x0000000000000000000000000000000000000000000000000000000000000005"
+      "id": 10,
+      "ingress": [],
+      "proof": "0x01030508"
     }
   ]
 }"#);

primitives/src/parachain.rs

@@ -30,39 +30,89 @@ impl From<u64> for Id {
 	fn from(x: u64) -> Self { Id(x) }
 }
 
-/// A parachain block proposal.
+/// A cross-parachain message.
+#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Serialize, Deserialize)]
+pub struct Message(#[serde(with="bytes")] pub Vec<u8>);
+
+/// Posts to egress queues.
+#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct EgressPosts(pub ::std::collections::BTreeMap<::parachain::Id, Vec<::parachain::Message>>);
+
+/// A collated ingress queue.
+///
+/// This is just an ordered vector of other parachains' egress queues,
+/// obtained according to the routing rules.
+#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ConsolidatedIngress(pub Vec<(Id, Vec<Message>)>);
+
+/// A parachain block candidate.
+/// This is passed from
+#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+#[serde(deny_unknown_fields)]
+pub struct Candidate {
+	/// Parachain ID
+	pub id: Id,
+
+	/// Consolidated ingress queues.
+	///
+	/// This will always be the same for each valid candidate building on the
+	/// same relay chain block.
+	pub ingress: ConsolidatedIngress,
+
+	/// Data necessary to prove validity of the head data.
+	pub proof: RawProof,
+}
+
+/// A parachain block candidate receipt.
+///
+/// This is what is actually included on the relay-chain.
+///
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 #[serde(deny_unknown_fields)]
-pub struct Proposal {
-	/// The ID of the parachain this is a proposal for.
-	pub parachain: Id,
-	/// Parachain block header bytes.
-	pub header: Header,
-	/// Hash of data necessary to prove validity of the header.
-	pub proof_hash: ProofHash,
+pub struct CandidateReceipt {
+	/// Parachain ID
+	pub id: Id,
+
+	/// Collator ID
+	pub collator: super::Address,
+
+	/// Head data produced by the validation function.
+	pub head_data: HeadData,
+
+	// TODO: balance uploads and fees
+
+	/// Egress queue roots, sorted by chain ID.
+	pub egress_roots: Vec<(Id, ::hash::H256)>,
+
+	/// Hash of data necessary to prove validity of the head data.
+	pub proof: ProofHash,
 }
 
-/// Parachain header raw bytes wrapper type.
+/// Parachain head data raw bytes wrapper type.
+///
+/// The notion of a header is a little too specific for parachains.
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
-pub struct Header(#[serde(with="bytes")] pub Vec<u8>);
+pub struct HeadData(#[serde(with="bytes")] pub Vec<u8>);
 
-/// Hash used to refer to proof of block header.
+/// Hash used to refer to proof of block head data.
 pub type ProofHash = ::hash::H256;
 
 /// Raw proof data.
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct RawProof(#[serde(with="bytes")] pub Vec<u8>);
 
 impl RawProof {
-	/// Compute and store the hash of the proof.
+	/// Compute and store the hash of the proof
 	pub fn into_proof(self) -> Proof {
 		let hash = ::hash(&self.0);
 		Proof(self, hash)
 	}
 }
 
-/// Parachain proof data.
+/// Parachain proof data. This is passed to the validation function
+/// along with the ingress queue and produces head data.
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Proof(RawProof, ProofHash);