From 1194d89f5634fe0bf31a71ae50cdba0ee0d977e3 Mon Sep 17 00:00:00 2001
From: Robert Habermeier <rphmeier@gmail.com>
Date: Tue, 14 Nov 2017 22:13:58 +0100
Subject: [PATCH 01/10] collator crate skeleton and description

---
 Cargo.lock          |  7 ++++++
 Cargo.toml          |  1 +
 collator/Cargo.toml |  8 +++++++
 collator/src/lib.rs | 55 +++++++++++++++++++++++++++++++++++++++++++++
 4 files changed, 71 insertions(+)
 create mode 100644 collator/Cargo.toml
 create mode 100644 collator/src/lib.rs

diff --git a/Cargo.lock b/Cargo.lock
index 7dfdee65905fd..35fb898fb2b4a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -404,6 +404,13 @@ dependencies = [
  "log 0.3.8 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "polkadot-collator"
+version = "0.1.0"
+dependencies = [
+ "polkadot-primitives 0.1.0",
+]
+
 [[package]]
 name = "polkadot-contracts"
 version = "0.1.0"
diff --git a/Cargo.toml b/Cargo.toml
index f99dfd8ee543a..e7666193a0c8e 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -8,6 +8,7 @@ polkadot-cli = { path = "cli", version = "0.1" }
 
 [workspace]
 members = [
+	"collator",
 	"contracts",
 	"primitives",
 	"serializer",
diff --git a/collator/Cargo.toml b/collator/Cargo.toml
new file mode 100644
index 0000000000000..58b99b001e6d7
--- /dev/null
+++ b/collator/Cargo.toml
@@ -0,0 +1,8 @@
+[package]
+name = "polkadot-collator"
+version = "0.1.0"
+authors = ["Parity Technologies <rphmeier@gmail.com>"]
+description = "Abstract collation logic"
+
+[dependencies]
+polkadot-primitives = { path = "../primitives", version = "0.1" }
diff --git a/collator/src/lib.rs b/collator/src/lib.rs
new file mode 100644
index 0000000000000..f34f5bd2bb508
--- /dev/null
+++ b/collator/src/lib.rs
@@ -0,0 +1,55 @@
+// 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 call a witness).
+//!
+//! One of the other strengths of Polkadot is its ability to route messages
+//! between chains. Each parachain has an "egress" queue of messages for each
+//! other parachain, for a total of N^2 queues all together.
+//!
+//! We will refer to the egress queue of parachain A with destination B as
+//! egress[A -> B]
+//!
+//! On every block, each parachain will be intended to route messages from some
+//! subset of all the other parachains. Routing messages from a parachain B to A
+//! involves fully draining egress[B -> A]. The proposal for parachain A may
+//! additionally post new messages to egress[A -> _].
+//!
+//! Along with the proposal and witness, the collator passes this list of posts to
+//! the validator. The posts contains a mapping of parachain identifiers to
+//! a list of two new tree roots.
+//!
+//! Since a parachain X may either fully drain egress[A -> X] or leave it untouched,
+//! the first root is the root of the empty list concatenated with the posts to X,
+//! while the second is the root of the queue at the prior relay chain block
+//! with the posts to X concatenated. These two alternatives allow the relay
+//! chain block to be constructed correctly regardless of whetherh candidates
+//! from other parachains are actually included.
+//!
+//! This crate defines traits which provide context necessary for collation logic
+//! to be performed, as the collation logic itself.
+
+extern crate polkadot_primitives as primitives;
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn it_works() {
+    }
+}

From 7977b2edb95fdd8e97b3d0b110e7e787aa8cdab4 Mon Sep 17 00:00:00 2001
From: Robert Habermeier <rphmeier@gmail.com>
Date: Wed, 15 Nov 2017 16:34:14 +0100
Subject: [PATCH 02/10] parachain primitives: proof -> witness and egress
 format

---
 primitives/src/block.rs     |  6 ++--
 primitives/src/parachain.rs | 50 ++++++++++++++------------
 primitives/src/validator.rs | 70 +++++++++++++++++--------------------
 3 files changed, 63 insertions(+), 63 deletions(-)

diff --git a/primitives/src/block.rs b/primitives/src/block.rs
index 41844b5d732af..bd8c19162b06a 100644
--- a/primitives/src/block.rs
+++ b/primitives/src/block.rs
@@ -75,17 +75,15 @@ mod tests {
 		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(),
+					witness_hash: 5.into(),
 				}
 			],
 		}), r#"{
   "paraBlocks": [
     {
-      "parachain": 5,
       "header": "0x01020304",
-      "proofHash": "0x0000000000000000000000000000000000000000000000000000000000000005"
+      "witnessHash": "0x0000000000000000000000000000000000000000000000000000000000000005"
     }
   ]
 }"#);
diff --git a/primitives/src/parachain.rs b/primitives/src/parachain.rs
index 5d80ffcf1f7d4..3533c544796d1 100644
--- a/primitives/src/parachain.rs
+++ b/primitives/src/parachain.rs
@@ -30,51 +30,57 @@ impl From<u64> for Id {
 	fn from(x: u64) -> Self { Id(x) }
 }
 
+/// 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 parachain block proposal.
 #[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 witness_hash: WitnessHash,
 }
 
 /// Parachain header raw bytes wrapper type.
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Header(#[serde(with="bytes")] pub Vec<u8>);
 
-/// Hash used to refer to proof of block header.
-pub type ProofHash = ::hash::H256;
+/// Hash used to refer to witness of block header.
+pub type WitnessHash = ::hash::H256;
 
-/// Raw proof data.
+/// Raw witness data.
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
-pub struct RawProof(#[serde(with="bytes")] pub Vec<u8>);
+pub struct RawWitness(#[serde(with="bytes")] pub Vec<u8>);
 
-impl RawProof {
-	/// Compute and store the hash of the proof.
-	pub fn into_proof(self) -> Proof {
+impl RawWitness {
+	/// Compute and store the hash of the witness
+	pub fn into_witness(self) -> Witness {
 		let hash = ::hash(&self.0);
-		Proof(self, hash)
+		Witness(self, hash)
 	}
 }
 
-/// Parachain proof data.
+/// Parachain witness data.
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
-pub struct Proof(RawProof, ProofHash);
+pub struct Witness(RawWitness, WitnessHash);
 
-impl Proof {
-	/// Get raw proof data.
-	pub fn raw(&self) -> &RawProof { &self.0 }
+impl Witness {
+	/// Get raw witness data.
+	pub fn raw(&self) -> &RawWitness { &self.0 }
 
-	/// Get hash of proof data.
-	pub fn hash(&self) -> &ProofHash { &self.1 }
+	/// Get hash of witness data.
+	pub fn hash(&self) -> &WitnessHash { &self.1 }
 
-	/// Decompose the proof back into raw data and hash.
-	pub fn into_inner(self) -> (RawProof, ProofHash) {
+	/// Decompose the witness back into raw data and hash.
+	pub fn into_inner(self) -> (RawWitness, WitnessHash) {
 		(self.0, self.1)
 	}
 }
@@ -89,9 +95,9 @@ mod tests {
 	use polkadot_serializer as ser;
 
 	#[test]
-	fn test_proof_serialization() {
+	fn test_witness_serialization() {
 		assert_eq!(
-			ser::to_string_pretty(&Proof(RawProof(vec![1,2,3]), 5.into())),
+			ser::to_string_pretty(&Witness(RawWitness(vec![1,2,3]), 5.into())),
 			r#"[
   "0x010203",
   "0x0000000000000000000000000000000000000000000000000000000000000005"
diff --git a/primitives/src/validator.rs b/primitives/src/validator.rs
index afdff7e20f31e..76a6b825bba97 100644
--- a/primitives/src/validator.rs
+++ b/primitives/src/validator.rs
@@ -16,48 +16,36 @@
 
 //! Validator primitives.
 
-use bytes;
+use parachain::EgressPosts;
 
-/// Parachain incoming messages.
-#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
-pub struct IngressPosts(#[serde(with="bytes")] pub Vec<u8>);
-
-/// Parachain incoming messages delta.
-#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
-pub struct IngressPostsDelta(#[serde(with="bytes")] pub Vec<u8>);
-
-/// Parachain outgoing messages.
-#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
-pub struct EgressPosts(#[serde(with="bytes")] pub Vec<u8>);
-
-/// Validity result of particular proof and ingress queue.
+/// Validity result of particular witness and ingress queue.
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(tag="type", content="data")]
 #[serde(rename_all = "camelCase")]
 #[serde(deny_unknown_fields)]
-pub enum ProofValidity {
-	/// The proof is invalid.
+pub enum WitnessValidity {
+	/// The witness is invalid.
 	Invalid,
-	/// The proof is processed and new egress queue is created.
+	/// The witness is processed and new egress queue is created.
 	/// Also includes current ingress queue delta.
-	Valid(IngressPostsDelta, EgressPosts),
+	Valid(EgressPosts),
 }
 
-impl ProofValidity {
-	/// The proof is valid.
+impl WitnessValidity {
+	/// The witness is valid.
 	pub fn is_valid(&self) -> bool {
 		match *self {
-			ProofValidity::Invalid => false,
-			ProofValidity::Valid(..) => true,
+			WitnessValidity::Invalid => false,
+			WitnessValidity::Valid(..) => true,
 		}
 	}
 }
 
-impl From<Option<(IngressPostsDelta, EgressPosts)>> for ProofValidity {
-	fn from(posts: Option<(IngressPostsDelta, EgressPosts)>) -> Self {
+impl From<Option<EgressPosts>> for WitnessValidity {
+	fn from(posts: Option<EgressPosts>) -> Self {
 		match posts {
-			Some((delta, posts)) => ProofValidity::Valid(delta, posts),
-			None => ProofValidity::Invalid,
+			Some(posts) => WitnessValidity::Valid(posts),
+			None => WitnessValidity::Invalid,
 		}
 	}
 }
@@ -68,37 +56,45 @@ pub trait Validator {
 	/// Validation error.
 	type Error: ::std::error::Error;
 
-	/// Validates if the provided proof holds given a current ingress queue.
+	/// Validates if the provided witness holds given a current ingress queue.
 	///
 	/// In case of success produces egress posts.
 	fn validate(
 		&self,
-		messages: &IngressPosts,
-		proof: &::parachain::Proof,
+		witness: &::parachain::Witness,
 		code: &[u8],
-	) -> Result<ProofValidity, Self::Error>;
+	) -> Result<WitnessValidity, Self::Error>;
 }
 
 #[cfg(test)]
 mod tests {
 	use super::*;
+	use parachain::{EgressPosts, Message};
 	use polkadot_serializer as ser;
 
 	#[test]
-	fn test_proof_validity_serialization() {
+	fn test_witness_validity_serialization() {
 		assert_eq!(
-			ser::to_string_pretty(&ProofValidity::Invalid),
+			ser::to_string_pretty(&WitnessValidity::Invalid),
 			r#"{
   "type": "invalid"
 }"#);
+
+		let mut egress = ::std::collections::BTreeMap::new();
+		egress.insert(5.into(), vec![Message(vec![1, 2, 3])]);
+		egress.insert(7.into(), vec![Message(vec![4, 5, 6])]);
 		assert_eq!(
-			ser::to_string_pretty(&ProofValidity::Valid(IngressPostsDelta(vec![1]), EgressPosts(vec![1, 2, 3]))),
+			ser::to_string_pretty(&WitnessValidity::Valid(EgressPosts(egress))),
 			r#"{
   "type": "valid",
-  "data": [
-    "0x01",
-    "0x010203"
-  ]
+  "data": {
+    "5": [
+      "0x010203"
+    ],
+    "7": [
+      "0x040506"
+    ]
+  }
 }"#);
 	}
 }

From 9c5a3b96d60419fcfd0447bd65dbe73ce5a50de9 Mon Sep 17 00:00:00 2001
From: Robert Habermeier <rphmeier@gmail.com>
Date: Wed, 15 Nov 2017 19:27:39 +0100
Subject: [PATCH 03/10] collation of ingress queues through trait

---
 Cargo.lock          |   7 +++
 collator/Cargo.toml |   1 +
 collator/src/lib.rs | 120 +++++++++++++++++++++++++++++++++++++-------
 3 files changed, 111 insertions(+), 17 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 35fb898fb2b4a..3833f21125780 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -230,6 +230,11 @@ dependencies = [
  "bitflags 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "futures"
+version = "0.1.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
 [[package]]
 name = "hashdb"
 version = "0.1.1"
@@ -408,6 +413,7 @@ dependencies = [
 name = "polkadot-collator"
 version = "0.1.0"
 dependencies = [
+ "futures 0.1.17 (registry+https://github.com/rust-lang/crates.io-index)",
  "polkadot-primitives 0.1.0",
 ]
 
@@ -762,6 +768,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 "checksum fixed-hash 0.1.0 (git+https://github.com/paritytech/primitives.git)" = "<none>"
 "checksum fuchsia-zircon 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "f6c0581a4e363262e52b87f59ee2afe3415361c6ec35e665924eb08afe8ff159"
 "checksum fuchsia-zircon-sys 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "43f3795b4bae048dc6123a6b972cadde2e676f9ded08aef6bb77f5f157684a82"
+"checksum futures 0.1.17 (registry+https://github.com/rust-lang/crates.io-index)" = "118b49cac82e04121117cbd3121ede3147e885627d82c4546b87c702debb90c1"
 "checksum hashdb 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "d97be07c358c5b461268b4ce60304024c5fa5acfd4bd8cd743639f0252003cf5"
 "checksum heapsize 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)" = "54fab2624374e5137ae4df13bf32b0b269cb804df42d13a51221bbd431d1a237"
 "checksum isatty 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)" = "00c9301a947a2eaee7ce2556b80285dcc89558d07088962e6e8b9c25730f9dc6"
diff --git a/collator/Cargo.toml b/collator/Cargo.toml
index 58b99b001e6d7..71e2606f678d3 100644
--- a/collator/Cargo.toml
+++ b/collator/Cargo.toml
@@ -6,3 +6,4 @@ description = "Abstract collation logic"
 
 [dependencies]
 polkadot-primitives = { path = "../primitives", version = "0.1" }
+futures = "0.1.17"
diff --git a/collator/src/lib.rs b/collator/src/lib.rs
index f34f5bd2bb508..92f718039ff71 100644
--- a/collator/src/lib.rs
+++ b/collator/src/lib.rs
@@ -19,34 +19,120 @@
 //! 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 call a witness).
 //!
-//! One of the other strengths of Polkadot is its ability to route messages
-//! between chains. Each parachain has an "egress" queue of messages for each
-//! other parachain, for a total of N^2 queues all together.
+//! 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 queue of parachain A with destination B as
-//! egress[A -> B]
+//! 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. Routing messages from a parachain B to A
-//! involves fully draining egress[B -> A]. The proposal for parachain A may
-//! additionally post new messages to egress[A -> _].
+//! subset of all the other parachains.
 //!
-//! Along with the proposal and witness, the collator passes this list of posts to
-//! the validator. The posts contains a mapping of parachain identifiers to
-//! a list of two new tree roots.
+//! 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.
 //!
-//! Since a parachain X may either fully drain egress[A -> X] or leave it untouched,
-//! the first root is the root of the empty list concatenated with the posts to X,
-//! while the second is the root of the queue at the prior relay chain block
-//! with the posts to X concatenated. These two alternatives allow the relay
-//! chain block to be constructed correctly regardless of whetherh candidates
-//! from other parachains are actually included.
+//! 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;
+
+use futures::{stream, Stream, Future};
+use primitives::parachain::{RawWitness, Message, Id as ParaId};
+use primitives::hash::H256;
+
+/// A collation candidate.
+pub struct Candidate {
+	/// The witness data.
+	pub witness: RawWitness,
+}
+
+/// 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=Message>>(
+		&self,
+		ingress: I,
+	) -> Candidate;
+}
+
+/// 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;
+	type FutureEgress: Future<Item=Vec<Message>, Error=Self::Error>;
+
+	/// Ancestry hashes of the relay chain, along with a flag indicating
+	/// whether the local parachain had a candidate included.
+	type Ancestry: IntoIterator<Item=(H256, bool)>;
+
+	/// Provide a set of all parachains routed from at a block.
+	///
+	/// If `None`, provide the set meant to be routed from now.
+	/// If the block hash is invalid, provide the empty set.
+	fn routing_parachains(&self, block: Option<H256>) -> BTreeSet<ParaId>;
+
+	/// Get an iterator over ancestry
+	fn ancestry_iter(&self) -> Self::Ancestry;
+
+	/// Get egress(block)[id -> local_id]
+	fn remote_egress(&self, id: ParaId, block: H256) -> Self::FutureEgress;
+}
+
+/// Collate the necessary ingress queue using the given context.
+pub fn collate_ingress<'a, R>(relay_context: R)
+	-> Box<Future<Item=Vec<Message>, Error=R::Error> + 'a>
+	where R: RelayChainContext,
+		  R::Error: 'a,
+		  R::FutureEgress: 'a
+{
+	let mut egress_fetch = Vec::new();
+	let mut currently_routing = relay_context.routing_parachains(None);
+
+	// follow the ancestry back up to the last point where
+	// 1. the parachain was meant to be routed from
+	// 2. there was a candidate included
+	for (ancestry_hash, had_candidate) in relay_context.ancestry_iter() {
+		let mut to_remove = Vec::new();
+		let routed_at_hash = relay_context.routing_parachains(Some(ancestry_hash));
+
+		// we reverse the order because we are also walking the ancestry backwards.
+		// then we can collect all the futures in reverse order and get the
+		// correctly ordered ingress queue.
+		for para_id in currently_routing.iter().rev().cloned() {
+			// fetch queue at the outset of `ancestry_hash` regardless.
+			egress_fetch.push(relay_context.remote_egress(para_id, ancestry_hash));
+
+			if had_candidate && routed_at_hash.contains(&para_id) {
+				to_remove.push(para_id);
+			}
+		}
+
+		for para_id in to_remove {
+			currently_routing.remove(&para_id);
+		}
+	}
+
+	let future = stream::futures_ordered(egress_fetch.into_iter().rev())
+		.fold(Vec::new(), |mut v, e| { v.extend(e); Ok(v) } );
+
+	Box::new(future)
+}
+
 #[cfg(test)]
 mod tests {
     #[test]

From 639a3f9e8203fd5ed85908e26af4043138ef675c Mon Sep 17 00:00:00 2001
From: Robert Habermeier <rphmeier@gmail.com>
Date: Wed, 15 Nov 2017 21:36:46 +0100
Subject: [PATCH 04/10] add ingress collation test

---
 collator/src/lib.rs | 111 ++++++++++++++++++++++++++++++++++++++++++--
 1 file changed, 107 insertions(+), 4 deletions(-)

diff --git a/collator/src/lib.rs b/collator/src/lib.rs
index 92f718039ff71..49208ff25df9d 100644
--- a/collator/src/lib.rs
+++ b/collator/src/lib.rs
@@ -48,7 +48,7 @@ extern crate polkadot_primitives as primitives;
 
 use std::collections::BTreeSet;
 
-use futures::{stream, Stream, Future};
+use futures::{stream, Stream, Future, IntoFuture};
 use primitives::parachain::{RawWitness, Message, Id as ParaId};
 use primitives::hash::H256;
 
@@ -74,7 +74,7 @@ pub trait ParachainContext {
 /// some of the input.
 pub trait RelayChainContext {
 	type Error;
-	type FutureEgress: Future<Item=Vec<Message>, Error=Self::Error>;
+	type FutureEgress: IntoFuture<Item=Vec<Message>, Error=Self::Error>;
 
 	/// Ancestry hashes of the relay chain, along with a flag indicating
 	/// whether the local parachain had a candidate included.
@@ -125,6 +125,8 @@ pub fn collate_ingress<'a, R>(relay_context: R)
 		for para_id in to_remove {
 			currently_routing.remove(&para_id);
 		}
+
+		if currently_routing.is_empty() { break }
 	}
 
 	let future = stream::futures_ordered(egress_fetch.into_iter().rev())
@@ -135,7 +137,108 @@ pub fn collate_ingress<'a, R>(relay_context: R)
 
 #[cfg(test)]
 mod tests {
+	use super::*;
+
+	use std::collections::{HashMap, BTreeSet};
+
+	use futures::Future;
+	use primitives::parachain::{Message, Id as ParaId};
+	use primitives::hash::H256;
+
+	pub struct DummyRelayChainCtx {
+		blocks: Vec<(H256, bool)>,
+		routing_parachains: HashMap<H256, BTreeSet<ParaId>>,
+		egresses: HashMap<(ParaId, H256), Vec<Message>>,
+		currently_routing: BTreeSet<ParaId>,
+	}
+
+	impl RelayChainContext for DummyRelayChainCtx {
+		type Error = ();
+		type FutureEgress = Result<Vec<Message>, ()>;
+		type Ancestry = Box<Iterator<Item=(H256, bool)>>;
+
+		fn routing_parachains(&self, block: Option<H256>) -> BTreeSet<ParaId> {
+			match block {
+				None => self.currently_routing.clone(),
+				Some(block) => self.routing_parachains.get(&block).cloned().unwrap_or_default(),
+			}
+		}
+
+		fn ancestry_iter(&self) -> Self::Ancestry {
+			Box::new(self.blocks.clone().into_iter().rev())
+		}
+
+		fn remote_egress(&self, id: ParaId, block: H256) -> Result<Vec<Message>, ()> {
+			Ok(self.egresses.get(&(id, block)).cloned().unwrap_or_default())
+		}
+	}
+
+	fn fake_hash(x: u8) -> H256 {
+		let mut hash = H256::default();
+		hash[0] = x;
+		hash
+	}
+
     #[test]
-    fn it_works() {
-    }
+	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 {
+			blocks: vec![
+				(fake_hash(1), true),
+				(fake_hash(2), false),
+				(fake_hash(3), true),
+				(fake_hash(4), false),
+				(fake_hash(5), false),
+			],
+			routing_parachains: vec![
+				(fake_hash(0), BTreeSet::new()),
+				(fake_hash(1), route_from(&[2.into()])),
+				(fake_hash(2), BTreeSet::new()),
+				(fake_hash(3), route_from(&[3.into()])),
+				(fake_hash(4), route_from(&[2.into()])),
+				(fake_hash(5), BTreeSet::new()),
+			].into_iter().collect(),
+			currently_routing: route_from(&[2.into(), 3.into()]),
+			egresses: vec![
+				// these two will not be included because they are from before
+				// the last successful times messages were routed from their
+				// chains.
+				((2.into(), fake_hash(0)), message(vec![200])),
+				((3.into(), fake_hash(2)), message(vec![101, 102, 103])),
+
+				// egresses for `2`: last routed successfully at 1.
+				((2.into(), fake_hash(1)), message(vec![1, 2, 3])),
+				((2.into(), fake_hash(2)), message(vec![4, 5, 6])),
+				((2.into(), fake_hash(3)), message(vec![7, 8])),
+				((2.into(), fake_hash(4)), message(vec![10])),
+				((2.into(), fake_hash(5)), message(vec![12])),
+
+				// egresses for `3`: last routed successfully at 3.
+				((3.into(), fake_hash(3)), message(vec![9])),
+				((3.into(), fake_hash(4)), message(vec![11])),
+				((3.into(), fake_hash(5)), message(vec![13])),
+			].into_iter().collect(),
+		};
+
+		assert_eq!(
+			collate_ingress(dummy_ctx).wait().unwrap(),
+			vec![
+				Message(vec![1, 2, 3]),
+				Message(vec![4, 5, 6]),
+				Message(vec![7, 8]),
+				Message(vec![9]),
+				Message(vec![10]),
+				Message(vec![11]),
+				Message(vec![12]),
+				Message(vec![13]),
+			]
+		)
+	}
 }

From 049196a72d3ff4107d5a673f4daec8cde017afd8 Mon Sep 17 00:00:00 2001
From: Robert Habermeier <rphmeier@gmail.com>
Date: Wed, 15 Nov 2017 21:54:54 +0100
Subject: [PATCH 05/10] structure for collated ingress

---
 collator/src/lib.rs         | 35 +++++++++++++++++++++--------------
 primitives/src/parachain.rs |  6 ++++++
 2 files changed, 27 insertions(+), 14 deletions(-)

diff --git a/collator/src/lib.rs b/collator/src/lib.rs
index 49208ff25df9d..b5028c0fafe1c 100644
--- a/collator/src/lib.rs
+++ b/collator/src/lib.rs
@@ -49,11 +49,13 @@ extern crate polkadot_primitives as primitives;
 use std::collections::BTreeSet;
 
 use futures::{stream, Stream, Future, IntoFuture};
-use primitives::parachain::{RawWitness, Message, Id as ParaId};
+use primitives::parachain::{RawWitness, CollatedIngress, Header, Message, Id as ParaId};
 use primitives::hash::H256;
 
 /// A collation candidate.
 pub struct Candidate {
+	/// The header data.
+	pub header: Header,
 	/// The witness data.
 	pub witness: RawWitness,
 }
@@ -95,7 +97,7 @@ pub trait RelayChainContext {
 
 /// Collate the necessary ingress queue using the given context.
 pub fn collate_ingress<'a, R>(relay_context: R)
-	-> Box<Future<Item=Vec<Message>, Error=R::Error> + 'a>
+	-> Box<Future<Item=CollatedIngress, Error=R::Error> + 'a>
 	where R: RelayChainContext,
 		  R::Error: 'a,
 		  R::FutureEgress: 'a
@@ -115,7 +117,11 @@ pub fn collate_ingress<'a, R>(relay_context: R)
 		// correctly ordered ingress queue.
 		for para_id in currently_routing.iter().rev().cloned() {
 			// fetch queue at the outset of `ancestry_hash` regardless.
-			egress_fetch.push(relay_context.remote_egress(para_id, ancestry_hash));
+			let fetch = relay_context.remote_egress(para_id, ancestry_hash)
+				.into_future()
+				.map(move |q| (para_id, q));
+
+			egress_fetch.push(fetch);
 
 			if had_candidate && routed_at_hash.contains(&para_id) {
 				to_remove.push(para_id);
@@ -130,7 +136,8 @@ pub fn collate_ingress<'a, R>(relay_context: R)
 	}
 
 	let future = stream::futures_ordered(egress_fetch.into_iter().rev())
-		.fold(Vec::new(), |mut v, e| { v.extend(e); Ok(v) } );
+		.fold(Vec::new(), |mut v, x| { v.push(x); Ok(v) } )
+		.map(CollatedIngress);
 
 	Box::new(future)
 }
@@ -229,16 +236,16 @@ mod tests {
 
 		assert_eq!(
 			collate_ingress(dummy_ctx).wait().unwrap(),
-			vec![
-				Message(vec![1, 2, 3]),
-				Message(vec![4, 5, 6]),
-				Message(vec![7, 8]),
-				Message(vec![9]),
-				Message(vec![10]),
-				Message(vec![11]),
-				Message(vec![12]),
-				Message(vec![13]),
+			CollatedIngress(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])),
 			]
-		)
+		))
 	}
 }
diff --git a/primitives/src/parachain.rs b/primitives/src/parachain.rs
index 3533c544796d1..49f67165a2027 100644
--- a/primitives/src/parachain.rs
+++ b/primitives/src/parachain.rs
@@ -38,6 +38,12 @@ pub struct Message(#[serde(with="bytes")] pub Vec<u8>);
 #[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(Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CollatedIngress(pub Vec<(Id, Vec<Message>)>);
+
 /// A parachain block proposal.
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]

From 852d03e2b416460ca444fc424fc82d6d411121b2 Mon Sep 17 00:00:00 2001
From: Robert Habermeier <rphmeier@gmail.com>
Date: Wed, 15 Nov 2017 21:56:57 +0100
Subject: [PATCH 06/10] add collated ingress to proposal

---
 primitives/src/parachain.rs | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git a/primitives/src/parachain.rs b/primitives/src/parachain.rs
index 49f67165a2027..239ae3ce0d573 100644
--- a/primitives/src/parachain.rs
+++ b/primitives/src/parachain.rs
@@ -39,9 +39,10 @@ pub struct Message(#[serde(with="bytes")] pub Vec<u8>);
 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(Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone PartialEq, Eq, Serialize, Deserialize)]
 pub struct CollatedIngress(pub Vec<(Id, Vec<Message>)>);
 
 /// A parachain block proposal.
@@ -51,6 +52,13 @@ pub struct CollatedIngress(pub Vec<(Id, Vec<Message>)>);
 pub struct Proposal {
 	/// Parachain block header bytes.
 	pub header: Header,
+
+	/// Collated ingress queues.
+	///
+	/// This will always be the same for each valid proposal building on the
+	/// same relay chain block.
+	pub ingress: CollatedIngress,
+
 	/// Hash of data necessary to prove validity of the header.
 	pub witness_hash: WitnessHash,
 }

From 2a2179378de25a818dbf213d6ac1746618ae22eb Mon Sep 17 00:00:00 2001
From: Robert Habermeier <rphmeier@gmail.com>
Date: Wed, 15 Nov 2017 22:39:00 +0100
Subject: [PATCH 07/10] witness -> proof

---
 collator/src/lib.rs         | 24 ++++++--------
 primitives/src/block.rs     | 14 +++++----
 primitives/src/parachain.rs | 63 +++++++++++++++++++------------------
 primitives/src/validator.rs | 34 ++++++++++----------
 4 files changed, 67 insertions(+), 68 deletions(-)

diff --git a/collator/src/lib.rs b/collator/src/lib.rs
index b5028c0fafe1c..044a98da87a40 100644
--- a/collator/src/lib.rs
+++ b/collator/src/lib.rs
@@ -17,7 +17,8 @@
 //! 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 call a witness).
+//! a state transition, along with a proof for its validity
+//! (what we might call a witness).
 //!
 //! One of collators' other roles is to route messages between chains.
 //! Each parachain produces a list of "egress" posts of messages for each other
@@ -49,26 +50,18 @@ extern crate polkadot_primitives as primitives;
 use std::collections::BTreeSet;
 
 use futures::{stream, Stream, Future, IntoFuture};
-use primitives::parachain::{RawWitness, CollatedIngress, Header, Message, Id as ParaId};
+use primitives::parachain::{self, RawProof, ConsolidatedIngress, HeadData, Message, Id as ParaId};
 use primitives::hash::H256;
 
-/// A collation candidate.
-pub struct Candidate {
-	/// The header data.
-	pub header: Header,
-	/// The witness data.
-	pub witness: RawWitness,
-}
-
 /// 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=Message>>(
+	fn produce_candidate<I: IntoIterator<Item=(ParaId, Message)>>(
 		&self,
 		ingress: I,
-	) -> Candidate;
+	) -> parachain::Proof;
 }
 
 /// Relay chain context needed to collate.
@@ -96,8 +89,9 @@ pub trait RelayChainContext {
 }
 
 /// Collate the necessary ingress queue using the given context.
+// TODO: impl trait
 pub fn collate_ingress<'a, R>(relay_context: R)
-	-> Box<Future<Item=CollatedIngress, Error=R::Error> + 'a>
+	-> Box<Future<Item=ConsolidatedIngress, Error=R::Error> + 'a>
 	where R: RelayChainContext,
 		  R::Error: 'a,
 		  R::FutureEgress: 'a
@@ -137,7 +131,7 @@ pub fn collate_ingress<'a, R>(relay_context: R)
 
 	let future = stream::futures_ordered(egress_fetch.into_iter().rev())
 		.fold(Vec::new(), |mut v, x| { v.push(x); Ok(v) } )
-		.map(CollatedIngress);
+		.map(ConsolidatedIngress);
 
 	Box::new(future)
 }
@@ -236,7 +230,7 @@ mod tests {
 
 		assert_eq!(
 			collate_ingress(dummy_ctx).wait().unwrap(),
-			CollatedIngress(vec![
+			ConsolidatedIngress(vec![
 				(2.into(), message(vec![1, 2, 3])),
 				(2.into(), message(vec![4, 5, 6])),
 				(2.into(), message(vec![7, 8])),
diff --git a/primitives/src/block.rs b/primitives/src/block.rs
index f34bf8fa8a89b..68e864ab83f98 100644
--- a/primitives/src/block.rs
+++ b/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,16 +73,18 @@ mod tests {
 	fn test_body_serialization() {
 		assert_eq!(ser::to_string_pretty(&Body {
 			para_blocks: vec![
-				parachain::Proposal {
-					header: parachain::Header(vec![1, 2, 3, 4]),
-					witness_hash: 5.into(),
+				parachain::Candidate {
+					id: 10.into(),
+					ingress: Default::default(),
+					proof_hash: 5.into(),
 				}
 			],
 		}), r#"{
   "paraBlocks": [
     {
-      "header": "0x01020304",
-      "witnessHash": "0x0000000000000000000000000000000000000000000000000000000000000005"
+      "id": 10,
+      "ingress": [],
+      "proofHash": "0x0000000000000000000000000000000000000000000000000000000000000005"
     }
   ]
 }"#);
diff --git a/primitives/src/parachain.rs b/primitives/src/parachain.rs
index 239ae3ce0d573..4345a6f2ce3fd 100644
--- a/primitives/src/parachain.rs
+++ b/primitives/src/parachain.rs
@@ -42,59 +42,62 @@ pub struct EgressPosts(pub ::std::collections::BTreeMap<::parachain::Id, Vec<::p
 ///
 /// This is just an ordered vector of other parachains' egress queues,
 /// obtained according to the routing rules.
-#[derive(Debug, Clone PartialEq, Eq, Serialize, Deserialize)]
-pub struct CollatedIngress(pub Vec<(Id, Vec<Message>)>);
+#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ConsolidatedIngress(pub Vec<(Id, Vec<Message>)>);
 
 /// A parachain block proposal.
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 #[serde(deny_unknown_fields)]
-pub struct Proposal {
-	/// Parachain block header bytes.
-	pub header: Header,
+pub struct Candidate {
+	/// Parachain ID
+	pub id: Id,
 
-	/// Collated ingress queues.
+	/// Consolidated ingress queues.
 	///
 	/// This will always be the same for each valid proposal building on the
 	/// same relay chain block.
-	pub ingress: CollatedIngress,
+	pub ingress: ConsolidatedIngress,
 
-	/// Hash of data necessary to prove validity of the header.
-	pub witness_hash: WitnessHash,
+	/// Hash of data necessary to prove validity of the head data.
+	pub proof_hash: 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 witness of block header.
-pub type WitnessHash = ::hash::H256;
+/// Hash used to refer to proof of block head data.
+pub type ProofHash = ::hash::H256;
 
-/// Raw witness data.
+/// Raw proof data.
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
-pub struct RawWitness(#[serde(with="bytes")] pub Vec<u8>);
+pub struct RawProof(#[serde(with="bytes")] pub Vec<u8>);
 
-impl RawWitness {
-	/// Compute and store the hash of the witness
-	pub fn into_witness(self) -> Witness {
+impl RawProof {
+	/// Compute and store the hash of the proof
+	pub fn into_proof(self) -> Proof {
 		let hash = ::hash(&self.0);
-		Witness(self, hash)
+		Proof(self, hash)
 	}
 }
 
-/// Parachain witness 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 Witness(RawWitness, WitnessHash);
+pub struct Proof(RawProof, ProofHash);
 
-impl Witness {
-	/// Get raw witness data.
-	pub fn raw(&self) -> &RawWitness { &self.0 }
+impl Proof {
+	/// Get raw proof data.
+	pub fn raw(&self) -> &RawProof { &self.0 }
 
-	/// Get hash of witness data.
-	pub fn hash(&self) -> &WitnessHash { &self.1 }
+	/// Get hash of proof data.
+	pub fn hash(&self) -> &ProofHash { &self.1 }
 
-	/// Decompose the witness back into raw data and hash.
-	pub fn into_inner(self) -> (RawWitness, WitnessHash) {
+	/// Decompose the proof back into raw data and hash.
+	pub fn into_inner(self) -> (RawProof, ProofHash) {
 		(self.0, self.1)
 	}
 }
@@ -109,9 +112,9 @@ mod tests {
 	use polkadot_serializer as ser;
 
 	#[test]
-	fn test_witness_serialization() {
+	fn test_proof_serialization() {
 		assert_eq!(
-			ser::to_string_pretty(&Witness(RawWitness(vec![1,2,3]), 5.into())),
+			ser::to_string_pretty(&Proof(RawProof(vec![1,2,3]), 5.into())),
 			r#"[
   "0x010203",
   "0x0000000000000000000000000000000000000000000000000000000000000005"
diff --git a/primitives/src/validator.rs b/primitives/src/validator.rs
index 76a6b825bba97..a1d3bbdd216c8 100644
--- a/primitives/src/validator.rs
+++ b/primitives/src/validator.rs
@@ -18,34 +18,34 @@
 
 use parachain::EgressPosts;
 
-/// Validity result of particular witness and ingress queue.
+/// Validity result of particular proof and ingress queue.
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(tag="type", content="data")]
 #[serde(rename_all = "camelCase")]
 #[serde(deny_unknown_fields)]
-pub enum WitnessValidity {
-	/// The witness is invalid.
+pub enum ProofValidity {
+	/// The proof is invalid.
 	Invalid,
-	/// The witness is processed and new egress queue is created.
+	/// The proof is processed and new egress queue is created.
 	/// Also includes current ingress queue delta.
 	Valid(EgressPosts),
 }
 
-impl WitnessValidity {
-	/// The witness is valid.
+impl ProofValidity {
+	/// The proof is valid.
 	pub fn is_valid(&self) -> bool {
 		match *self {
-			WitnessValidity::Invalid => false,
-			WitnessValidity::Valid(..) => true,
+			ProofValidity::Invalid => false,
+			ProofValidity::Valid(..) => true,
 		}
 	}
 }
 
-impl From<Option<EgressPosts>> for WitnessValidity {
+impl From<Option<EgressPosts>> for ProofValidity {
 	fn from(posts: Option<EgressPosts>) -> Self {
 		match posts {
-			Some(posts) => WitnessValidity::Valid(posts),
-			None => WitnessValidity::Invalid,
+			Some(posts) => ProofValidity::Valid(posts),
+			None => ProofValidity::Invalid,
 		}
 	}
 }
@@ -56,14 +56,14 @@ pub trait Validator {
 	/// Validation error.
 	type Error: ::std::error::Error;
 
-	/// Validates if the provided witness holds given a current ingress queue.
+	/// Validates if the provided proof holds given a current ingress queue.
 	///
 	/// In case of success produces egress posts.
 	fn validate(
 		&self,
-		witness: &::parachain::Witness,
+		proof: &::parachain::Proof,
 		code: &[u8],
-	) -> Result<WitnessValidity, Self::Error>;
+	) -> Result<ProofValidity, Self::Error>;
 }
 
 #[cfg(test)]
@@ -73,9 +73,9 @@ mod tests {
 	use polkadot_serializer as ser;
 
 	#[test]
-	fn test_witness_validity_serialization() {
+	fn test_proof_validity_serialization() {
 		assert_eq!(
-			ser::to_string_pretty(&WitnessValidity::Invalid),
+			ser::to_string_pretty(&ProofValidity::Invalid),
 			r#"{
   "type": "invalid"
 }"#);
@@ -84,7 +84,7 @@ mod tests {
 		egress.insert(5.into(), vec![Message(vec![1, 2, 3])]);
 		egress.insert(7.into(), vec![Message(vec![4, 5, 6])]);
 		assert_eq!(
-			ser::to_string_pretty(&WitnessValidity::Valid(EgressPosts(egress))),
+			ser::to_string_pretty(&ProofValidity::Valid(EgressPosts(egress))),
 			r#"{
   "type": "valid",
   "data": {

From 9b1a852f0e9cc464e88e0e1e5371a9ed63f7d5c7 Mon Sep 17 00:00:00 2001
From: Robert Habermeier <rphmeier@gmail.com>
Date: Thu, 16 Nov 2017 18:46:18 +0100
Subject: [PATCH 08/10] ingress collation and candidate creation + code cleanup

---
 collator/src/lib.rs         | 178 +++++++++++++++---------------------
 primitives/src/block.rs     |   4 +-
 primitives/src/parachain.rs |  33 ++++++-
 3 files changed, 106 insertions(+), 109 deletions(-)

diff --git a/collator/src/lib.rs b/collator/src/lib.rs
index 044a98da87a40..b89122aa1659b 100644
--- a/collator/src/lib.rs
+++ b/collator/src/lib.rs
@@ -18,7 +18,7 @@
 //!
 //! 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).
+//! (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
@@ -47,11 +47,10 @@
 extern crate futures;
 extern crate polkadot_primitives as primitives;
 
-use std::collections::BTreeSet;
+use std::collections::{BTreeSet, BTreeMap};
 
 use futures::{stream, Stream, Future, IntoFuture};
-use primitives::parachain::{self, RawProof, ConsolidatedIngress, HeadData, Message, Id as ParaId};
-use primitives::hash::H256;
+use primitives::parachain::{self, ConsolidatedIngress, Message, Id as ParaId};
 
 /// Parachain context needed for collation.
 ///
@@ -61,7 +60,7 @@ pub trait ParachainContext {
 	fn produce_candidate<I: IntoIterator<Item=(ParaId, Message)>>(
 		&self,
 		ingress: I,
-	) -> parachain::Proof;
+	) -> parachain::RawProof;
 }
 
 /// Relay chain context needed to collate.
@@ -69,23 +68,16 @@ pub trait ParachainContext {
 /// some of the input.
 pub trait RelayChainContext {
 	type Error;
-	type FutureEgress: IntoFuture<Item=Vec<Message>, Error=Self::Error>;
 
-	/// Ancestry hashes of the relay chain, along with a flag indicating
-	/// whether the local parachain had a candidate included.
-	type Ancestry: IntoIterator<Item=(H256, bool)>;
+	/// 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 routed from at a block.
-	///
-	/// If `None`, provide the set meant to be routed from now.
-	/// If the block hash is invalid, provide the empty set.
-	fn routing_parachains(&self, block: Option<H256>) -> BTreeSet<ParaId>;
+	/// Provide a set of all parachains meant to be routed to at a block.
+	fn routing_parachains(&self) -> BTreeSet<ParaId>;
 
-	/// Get an iterator over ancestry
-	fn ancestry_iter(&self) -> Self::Ancestry;
-
-	/// Get egress(block)[id -> local_id]
-	fn remote_egress(&self, id: ParaId, block: H256) -> Self::FutureEgress;
+	/// 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.
@@ -97,45 +89,57 @@ pub fn collate_ingress<'a, R>(relay_context: R)
 		  R::FutureEgress: 'a
 {
 	let mut egress_fetch = Vec::new();
-	let mut currently_routing = relay_context.routing_parachains(None);
-
-	// follow the ancestry back up to the last point where
-	// 1. the parachain was meant to be routed from
-	// 2. there was a candidate included
-	for (ancestry_hash, had_candidate) in relay_context.ancestry_iter() {
-		let mut to_remove = Vec::new();
-		let routed_at_hash = relay_context.routing_parachains(Some(ancestry_hash));
-
-		// we reverse the order because we are also walking the ancestry backwards.
-		// then we can collect all the futures in reverse order and get the
-		// correctly ordered ingress queue.
-		for para_id in currently_routing.iter().rev().cloned() {
-			// fetch queue at the outset of `ancestry_hash` regardless.
-			let fetch = relay_context.remote_egress(para_id, ancestry_hash)
-				.into_future()
-				.map(move |q| (para_id, q));
-
-			egress_fetch.push(fetch);
-
-			if had_candidate && routed_at_hash.contains(&para_id) {
-				to_remove.push(para_id);
-			}
-		}
 
-		for para_id in to_remove {
-			currently_routing.remove(&para_id);
-		}
+	for routing_parachain in relay_context.routing_parachains() {
+		let fetch = relay_context
+			.unrouted_egress(routing_parachain)
+			.into_future()
+			.map(move |egresses| (routing_parachain, egresses));
 
-		if currently_routing.is_empty() { break }
+		egress_fetch.push(fetch);
 	}
 
-	let future = stream::futures_ordered(egress_fetch.into_iter().rev())
-		.fold(Vec::new(), |mut v, x| { v.push(x); Ok(v) } )
+	// 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::*;
@@ -144,42 +148,25 @@ mod tests {
 
 	use futures::Future;
 	use primitives::parachain::{Message, Id as ParaId};
-	use primitives::hash::H256;
 
 	pub struct DummyRelayChainCtx {
-		blocks: Vec<(H256, bool)>,
-		routing_parachains: HashMap<H256, BTreeSet<ParaId>>,
-		egresses: HashMap<(ParaId, H256), Vec<Message>>,
+		egresses: HashMap<ParaId, Vec<Vec<Message>>>,
 		currently_routing: BTreeSet<ParaId>,
 	}
 
 	impl RelayChainContext for DummyRelayChainCtx {
 		type Error = ();
-		type FutureEgress = Result<Vec<Message>, ()>;
-		type Ancestry = Box<Iterator<Item=(H256, bool)>>;
-
-		fn routing_parachains(&self, block: Option<H256>) -> BTreeSet<ParaId> {
-			match block {
-				None => self.currently_routing.clone(),
-				Some(block) => self.routing_parachains.get(&block).cloned().unwrap_or_default(),
-			}
-		}
+		type FutureEgress = Result<Vec<Vec<Message>>, ()>;
 
-		fn ancestry_iter(&self) -> Self::Ancestry {
-			Box::new(self.blocks.clone().into_iter().rev())
+		fn routing_parachains(&self) -> BTreeSet<ParaId> {
+			self.currently_routing.clone()
 		}
 
-		fn remote_egress(&self, id: ParaId, block: H256) -> Result<Vec<Message>, ()> {
-			Ok(self.egresses.get(&(id, block)).cloned().unwrap_or_default())
+		fn unrouted_egress(&self, id: ParaId) -> Result<Vec<Vec<Message>>, ()> {
+			Ok(self.egresses.get(&id).cloned().unwrap_or_default())
 		}
 	}
 
-	fn fake_hash(x: u8) -> H256 {
-		let mut hash = H256::default();
-		hash[0] = x;
-		hash
-	}
-
     #[test]
 	fn collates_ingress() {
 		let route_from = |x: &[ParaId]| {
@@ -191,40 +178,23 @@ mod tests {
 		let message = |x: Vec<u8>| vec![Message(x)];
 
 		let dummy_ctx = DummyRelayChainCtx {
-			blocks: vec![
-				(fake_hash(1), true),
-				(fake_hash(2), false),
-				(fake_hash(3), true),
-				(fake_hash(4), false),
-				(fake_hash(5), false),
-			],
-			routing_parachains: vec![
-				(fake_hash(0), BTreeSet::new()),
-				(fake_hash(1), route_from(&[2.into()])),
-				(fake_hash(2), BTreeSet::new()),
-				(fake_hash(3), route_from(&[3.into()])),
-				(fake_hash(4), route_from(&[2.into()])),
-				(fake_hash(5), BTreeSet::new()),
-			].into_iter().collect(),
 			currently_routing: route_from(&[2.into(), 3.into()]),
 			egresses: vec![
-				// these two will not be included because they are from before
-				// the last successful times messages were routed from their
-				// chains.
-				((2.into(), fake_hash(0)), message(vec![200])),
-				((3.into(), fake_hash(2)), message(vec![101, 102, 103])),
-
-				// egresses for `2`: last routed successfully at 1.
-				((2.into(), fake_hash(1)), message(vec![1, 2, 3])),
-				((2.into(), fake_hash(2)), message(vec![4, 5, 6])),
-				((2.into(), fake_hash(3)), message(vec![7, 8])),
-				((2.into(), fake_hash(4)), message(vec![10])),
-				((2.into(), fake_hash(5)), message(vec![12])),
-
-				// egresses for `3`: last routed successfully at 3.
-				((3.into(), fake_hash(3)), message(vec![9])),
-				((3.into(), fake_hash(4)), message(vec![11])),
-				((3.into(), fake_hash(5)), message(vec![13])),
+				// 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(),
 		};
 
diff --git a/primitives/src/block.rs b/primitives/src/block.rs
index 68e864ab83f98..eb084c5b327bb 100644
--- a/primitives/src/block.rs
+++ b/primitives/src/block.rs
@@ -76,7 +76,7 @@ mod tests {
 				parachain::Candidate {
 					id: 10.into(),
 					ingress: Default::default(),
-					proof_hash: 5.into(),
+					proof: ::parachain::RawProof(vec![1, 3, 5, 8]),
 				}
 			],
 		}), r#"{
@@ -84,7 +84,7 @@ mod tests {
     {
       "id": 10,
       "ingress": [],
-      "proofHash": "0x0000000000000000000000000000000000000000000000000000000000000005"
+      "proof": "0x01030508"
     }
   ]
 }"#);
diff --git a/primitives/src/parachain.rs b/primitives/src/parachain.rs
index 4345a6f2ce3fd..3740b22a32ac0 100644
--- a/primitives/src/parachain.rs
+++ b/primitives/src/parachain.rs
@@ -45,7 +45,8 @@ pub struct EgressPosts(pub ::std::collections::BTreeMap<::parachain::Id, Vec<::p
 #[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct ConsolidatedIngress(pub Vec<(Id, Vec<Message>)>);
 
-/// A parachain block proposal.
+/// A parachain block candidate.
+/// This is passed from
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 #[serde(deny_unknown_fields)]
@@ -55,12 +56,38 @@ pub struct Candidate {
 
 	/// Consolidated ingress queues.
 	///
-	/// This will always be the same for each valid proposal building on the
+	/// 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 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_hash: ProofHash,
+	pub proof: ProofHash,
 }
 
 /// Parachain head data raw bytes wrapper type.

From 50fcc22ca7cc13525af711086bb588a48a51e623 Mon Sep 17 00:00:00 2001
From: Robert Habermeier <rphmeier@gmail.com>
Date: Wed, 6 Dec 2017 13:35:28 +0100
Subject: [PATCH 09/10] update collator lib to new definitions

---
 collator/src/lib.rs         | 11 ++++++-----
 primitives/src/parachain.rs |  9 ++++++++-
 2 files changed, 14 insertions(+), 6 deletions(-)

diff --git a/collator/src/lib.rs b/collator/src/lib.rs
index b89122aa1659b..54f0c7da364a4 100644
--- a/collator/src/lib.rs
+++ b/collator/src/lib.rs
@@ -60,7 +60,7 @@ pub trait ParachainContext {
 	fn produce_candidate<I: IntoIterator<Item=(ParaId, Message)>>(
 		&self,
 		ingress: I,
-	) -> parachain::RawProof;
+	) -> (parachain::BlockData, primitives::Signature);
 }
 
 /// Relay chain context needed to collate.
@@ -128,14 +128,15 @@ pub fn collate<'a, R, P>(local_id: ParaId, relay_context: R, para_context: P)
 		  P: ParachainContext + 'a,
 {
 	Box::new(collate_ingress(relay_context).map(move |ingress| {
-		let block_data = para_context.produce_candidate(
+		let (block_data, signature) = 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,
+			parachain_index: local_id,
+			collator_signature: signature,
+			block: block_data,
+			unprocessed_ingress: ingress,
 		}
 	}))
 }
diff --git a/primitives/src/parachain.rs b/primitives/src/parachain.rs
index 7e931bed84342..1249319aa8f4b 100644
--- a/primitives/src/parachain.rs
+++ b/primitives/src/parachain.rs
@@ -44,7 +44,7 @@ pub struct Candidate {
 	/// Unprocessed ingress queue.
 	///
 	/// Ordered by parachain ID and block number.
-	pub unprocessed_ingress: Vec<(u64, Vec<Message>)>,
+	pub unprocessed_ingress: ConsolidatedIngress,
 	/// Block data
 	pub block: BlockData,
 }
@@ -53,6 +53,13 @@ pub struct Candidate {
 #[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)]
 pub struct Message(#[serde(with="bytes")] pub Vec<u8>);
 
+/// Consolidated ingress queue data.
+///
+/// This is just an ordered vector of other parachains' egress queues,
+/// obtained according to the routing rules.
+#[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)]
+pub struct ConsolidatedIngress(pub Vec<(Id, Vec<Message>)>);
+
 /// Parachain block data.
 ///
 /// contains everything required to validate para-block, may contain block and witness data

From 7813cb60c753fb916ecb03ca0917633b91996d45 Mon Sep 17 00:00:00 2001
From: Robert Habermeier <rphmeier@gmail.com>
Date: Wed, 6 Dec 2017 13:36:43 +0100
Subject: [PATCH 10/10] address formatting grumble

---
 collator/src/lib.rs | 16 +++++++++-------
 1 file changed, 9 insertions(+), 7 deletions(-)

diff --git a/collator/src/lib.rs b/collator/src/lib.rs
index 54f0c7da364a4..1752ecdb7eab5 100644
--- a/collator/src/lib.rs
+++ b/collator/src/lib.rs
@@ -84,9 +84,10 @@ pub trait RelayChainContext {
 // 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
+	where
+		R: RelayChainContext,
+		R::Error: 'a,
+		R::FutureEgress: 'a,
 {
 	let mut egress_fetch = Vec::new();
 
@@ -122,10 +123,11 @@ pub fn collate_ingress<'a, R>(relay_context: R)
 /// 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,
+	where
+		R: RelayChainContext,
+	    R::Error: 'a,
+		R::FutureEgress: 'a,
+		P: ParachainContext + 'a,
 {
 	Box::new(collate_ingress(relay_context).map(move |ingress| {
 		let (block_data, signature) = para_context.produce_candidate(