Parachain execution yields messages to send (#96)

* read head-data directly out of WASM memory

* implement ext_post_message for parachain WASM

* further refactoring of the parachain module

* add externalities error type

* accumulate posted messages when validating parachain candidate

* define Extrinsic type in primitives

* availability-store: store extrinsic data

* compute extrinsic and check against candidate

* add some egress queue tests

* grumbles & substrate update

* ensure everything builds

polkadot/consensus/src/collation.rs

@@ -22,9 +22,10 @@
 use std::sync::Arc;
 
 use polkadot_primitives::{Block, Hash, AccountId, BlockId};
-use polkadot_primitives::parachain::{Id as ParaId, Collation, Extrinsic};
-use polkadot_primitives::parachain::ParachainHost;
+use polkadot_primitives::parachain::{Id as ParaId, Collation, Extrinsic, OutgoingMessage};
+use polkadot_primitives::parachain::{CandidateReceipt, ParachainHost};
 use runtime_primitives::traits::ProvideRuntimeApi;
+use parachain::{wasm_executor::{self, ExternalitiesError}, MessageRef};
 
 use futures::prelude::*;
 
@@ -100,7 +101,9 @@ impl<C: Collators, P: ProvideRuntimeApi> Future for CollationFetch<C, P>
 			};
 
 			match validate_collation(&*self.client, &self.relay_parent, &x) {
-				Ok(e) => return Ok(Async::Ready((x, e))),
+				Ok(e) => {
+					return Ok(Async::Ready((x, e)))
+				}
 				Err(e) => {
 					debug!("Failed to validate parachain due to API error: {}", e);
 
@@ -117,36 +120,137 @@ impl<C: Collators, P: ProvideRuntimeApi> Future for CollationFetch<C, P>
 error_chain! {
 	types { Error, ErrorKind, ResultExt; }
 
+	links {
+		Client(::client::error::Error, ::client::error::ErrorKind);
+		WasmValidation(wasm_executor::Error, wasm_executor::ErrorKind);
+	}
+
 	errors {
 		InactiveParachain(id: ParaId) {
 			description("Collated for inactive parachain"),
 			display("Collated for inactive parachain: {:?}", id),
 		}
-		ValidationFailure {
-			description("Parachain candidate failed validation."),
-			display("Parachain candidate failed validation."),
+		EgressRootMismatch(id: ParaId, expected: Hash, got: Hash) {
+			description("Got unexpected egress route."),
+			display(
+				"Got unexpected egress route to {:?}. (expected: {:?}, got {:?})",
+				id, expected, got
+			),
+		}
+		MissingEgressRoute(expected: Option<ParaId>, got: Option<ParaId>) {
+			description("Missing or extra egress route."),
+			display("Missing or extra egress route. (expected: {:?}, got {:?})", expected, got),
 		}
 		WrongHeadData(expected: Vec<u8>, got: Vec<u8>) {
 			description("Parachain validation produced wrong head data."),
 			display("Parachain validation produced wrong head data (expected: {:?}, got {:?}", expected, got),
 		}
 	}
+}
 
-	links {
-		Client(::client::error::Error, ::client::error::ErrorKind);
+/// Compute the egress trie root for a set of messages.
+pub fn egress_trie_root<A, I: IntoIterator<Item=A>>(messages: I) -> Hash
+	where A: AsRef<[u8]>
+{
+	::trie::ordered_trie_root::<primitives::Blake2Hasher, _, _>(messages)
+}
+
+fn check_and_compute_extrinsic(
+	mut outgoing: Vec<OutgoingMessage>,
+	expected_egress_roots: &[(ParaId, Hash)],
+) -> Result<Extrinsic, Error> {
+	// stable sort messages by parachain ID.
+	outgoing.sort_by_key(|msg| ParaId::from(msg.target));
+
+	{
+		let mut messages_iter = outgoing.iter().peekable();
+		let mut expected_egress_roots = expected_egress_roots.iter();
+		while let Some(batch_target) = messages_iter.peek().map(|o| o.target) {
+			let expected_root = match expected_egress_roots.next() {
+				None => return Err(ErrorKind::MissingEgressRoute(Some(batch_target), None).into()),
+				Some(&(id, ref root)) => if id == batch_target {
+					root
+				} else {
+					return Err(ErrorKind::MissingEgressRoute(Some(batch_target), Some(id)).into());
+				}
+			};
+
+ 			// we borrow the iterator mutably to ensure it advances so the
+			// next iteration of the loop starts with `messages_iter` pointing to
+			// the next batch.
+			let messages_to = messages_iter
+				.clone()
+				.take_while(|o| o.target == batch_target)
+				.map(|o| { let _ = messages_iter.next(); &o.data[..] });
+
+			let computed_root = egress_trie_root(messages_to);
+			if &computed_root != expected_root {
+				return Err(ErrorKind::EgressRootMismatch(
+					batch_target,
+					expected_root.clone(),
+					computed_root,
+				).into());
+			}
+		}
+
+		// also check that there are no more additional expected roots.
+		if let Some((next_target, _)) = expected_egress_roots.next() {
+			return Err(ErrorKind::MissingEgressRoute(None, Some(*next_target)).into());
+		}
+	}
+
+	Ok(Extrinsic { outgoing_messages: outgoing })
+}
+
+struct Externalities {
+	parachain_index: ParaId,
+	outgoing: Vec<OutgoingMessage>,
+}
+
+impl wasm_executor::Externalities for Externalities {
+	fn post_message(&mut self, message: MessageRef) -> Result<(), ExternalitiesError> {
+		// TODO: https://github.com/paritytech/polkadot/issues/92
+		// check per-message and per-byte fees for the parachain.
+		let target: ParaId = message.target.into();
+		if target == self.parachain_index {
+			return Err(ExternalitiesError::CannotPostMessage("posted message to self"));
+		}
+
+		self.outgoing.push(OutgoingMessage {
+			target,
+			data: message.data.to_vec(),
+		});
+
+		Ok(())
+	}
+}
+
+impl Externalities {
+	// Performs final checks of validity, producing the extrinsic data.
+	fn final_checks(
+		self,
+		candidate: &CandidateReceipt,
+	) -> Result<Extrinsic, Error> {
+		check_and_compute_extrinsic(
+			self.outgoing,
+			&candidate.egress_queue_roots[..],
+		)
 	}
 }
 
 /// Check whether a given collation is valid. Returns `Ok` on success, error otherwise.
+///
+/// This assumes that basic validity checks have been done:
+///   - Block data hash is the same as linked in candidate receipt.
 pub fn validate_collation<P>(
 	client: &P,
 	relay_parent: &BlockId,
 	collation: &Collation
 ) -> Result<Extrinsic, Error> where
 	P: ProvideRuntimeApi,
-	P::Api: ParachainHost<Block>
+	P::Api: ParachainHost<Block>,
 {
-	use parachain::{self, ValidationParams};
+	use parachain::ValidationParams;
 
 	let api = client.runtime_api();
 	let para_id = collation.receipt.parachain_index;
@@ -161,10 +265,15 @@ pub fn validate_collation<P>(
 		block_data: collation.block_data.0.clone(),
 	};
 
-	match parachain::wasm::validate_candidate(&validation_code, params) {
+	let mut ext = Externalities {
+		parachain_index: collation.receipt.parachain_index.clone(),
+		outgoing: Vec::new(),
+	};
+
+	match wasm_executor::validate_candidate(&validation_code, params, &mut ext) {
 		Ok(result) => {
 			if result.head_data == collation.receipt.head_data.0 {
-				Ok(Extrinsic)
+				ext.final_checks(&collation.receipt)
 			} else {
 				Err(ErrorKind::WrongHeadData(
 					collation.receipt.head_data.0.clone(),
@@ -172,6 +281,60 @@ pub fn validate_collation<P>(
 				).into())
 			}
 		}
-		Err(_) => Err(ErrorKind::ValidationFailure.into())
+		Err(e) => Err(e.into())
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use parachain::wasm_executor::Externalities as ExternalitiesTrait;
+
+	#[test]
+	fn egress_roots() {
+		let messages = vec![
+			OutgoingMessage { target: 3.into(), data: vec![1, 1, 1] },
+			OutgoingMessage { target: 1.into(), data: vec![1, 2, 3] },
+			OutgoingMessage { target: 2.into(), data: vec![4, 5, 6] },
+			OutgoingMessage { target: 1.into(), data: vec![7, 8, 9] },
+		];
+
+		let root_1 = egress_trie_root(&[vec![1, 2, 3], vec![7, 8, 9]]);
+		let root_2 = egress_trie_root(&[vec![4, 5, 6]]);
+		let root_3 = egress_trie_root(&[vec![1, 1, 1]]);
+
+		assert!(check_and_compute_extrinsic(
+			messages.clone(),
+			&[(1.into(), root_1), (2.into(), root_2), (3.into(), root_3)],
+		).is_ok());
+
+		// missing root.
+		assert!(check_and_compute_extrinsic(
+			messages.clone(),
+			&[(1.into(), root_1), (3.into(), root_3)],
+		).is_err());
+
+		// extra root.
+		assert!(check_and_compute_extrinsic(
+			messages.clone(),
+			&[(1.into(), root_1), (2.into(), root_2), (3.into(), root_3), (4.into(), Default::default())],
+		).is_err());
+
+		// root mismatch.
+		assert!(check_and_compute_extrinsic(
+			messages.clone(),
+			&[(1.into(), root_2), (2.into(), root_1), (3.into(), root_3)],
+		).is_err());
+	}
+
+	#[test]
+	fn ext_rejects_local_message() {
+		let mut ext = Externalities {
+			parachain_index: 5.into(),
+			outgoing: Vec::new(),
+		};
+
+		assert!(ext.post_message(MessageRef { target: 1, data: &[] }).is_ok());
+		assert!(ext.post_message(MessageRef { target: 5, data: &[] }).is_err());
 	}
 }

polkadot/consensus/src/lib.rs

@@ -41,6 +41,7 @@ extern crate substrate_primitives as primitives;
 extern crate srml_support as runtime_support;
 extern crate sr_primitives as runtime_primitives;
 extern crate substrate_client as client;
+extern crate substrate_trie as trie;
 
 extern crate exit_future;
 extern crate tokio;
@@ -90,7 +91,7 @@ use futures::future::{self, Either};
 use collation::CollationFetch;
 use dynamic_inclusion::DynamicInclusion;
 
-pub use self::collation::{validate_collation, Collators};
+pub use self::collation::{validate_collation, egress_trie_root, Collators};
 pub use self::error::{ErrorKind, Error};
 pub use self::shared_table::{SharedTable, ParachainWork, PrimedParachainWork, Validated, Statement, SignedStatement, GenericStatement};
 
@@ -113,8 +114,6 @@ pub trait TableRouter: Clone {
 	type Error;
 	/// Future that resolves when candidate data is fetched.
 	type FetchCandidate: IntoFuture<Item=BlockData,Error=Self::Error>;
-	/// Future that resolves when extrinsic candidate data is fetched.
-	type FetchExtrinsic: IntoFuture<Item=ParachainExtrinsic,Error=Self::Error>;
 
 	/// Call with local candidate data. This will make the data available on the network,
 	/// and sign, import, and broadcast a statement about the candidate.
@@ -122,9 +121,6 @@ pub trait TableRouter: Clone {
 
 	/// Fetch block data for a specific candidate.
 	fn fetch_block_data(&self, candidate: &CandidateReceipt) -> Self::FetchCandidate;
-
-	/// Fetch extrinsic data for a specific candidate.
-	fn fetch_extrinsic_data(&self, candidate: &CandidateReceipt) -> Self::FetchExtrinsic;
 }
 
 /// A long-lived network which can create parachain statement and BFT message routing processes on demand.
@@ -229,7 +225,6 @@ struct ParachainConsensus<C, N, P> {
 	extrinsic_store: ExtrinsicStore,
 	/// Live agreements.
 	live_instances: Mutex<HashMap<Hash, Arc<AttestationTracker>>>,
-
 }
 
 impl<C, N, P> ParachainConsensus<C, N, P> where

polkadot/consensus/src/shared_table/mod.rs

@@ -156,7 +156,7 @@ pub struct Validated {
 	/// Block data to ensure availability of.
 	pub block_data: BlockData,
 	/// Extrinsic data to ensure availability of.
-	pub extrinsic: Extrinsic,
+	pub extrinsic: Option<Extrinsic>,
 }
 
 /// Future that performs parachain validation work.
@@ -172,7 +172,7 @@ impl<D: Future> ParachainWork<D> {
 	pub fn prime<P: ProvideRuntimeApi>(self, api: Arc<P>)
 		-> PrimedParachainWork<
 			D,
-			impl Send + FnMut(&BlockId, &Collation) -> bool,
+			impl Send + FnMut(&BlockId, &Collation) -> Result<Extrinsic, ()>,
 		>
 		where
 			P: Send + Sync + 'static,
@@ -186,10 +186,10 @@ impl<D: Future> ParachainWork<D> {
 			);
 
 			match res {
-				Ok(_) => true,
+				Ok(e) => Ok(e),
 				Err(e) => {
 					debug!(target: "consensus", "Encountered bad collation: {}", e);
-					false
+					Err(())
 				}
 			}
 		};
@@ -199,7 +199,7 @@ impl<D: Future> ParachainWork<D> {
 
 	/// Prime the parachain work with a custom validation function.
 	pub fn prime_with<F>(self, validate: F) -> PrimedParachainWork<D, F>
-		where F: FnMut(&BlockId, &Collation) -> bool
+		where F: FnMut(&BlockId, &Collation) -> Result<Extrinsic, ()>
 	{
 		PrimedParachainWork { inner: self, validate }
 	}
@@ -219,7 +219,7 @@ pub struct PrimedParachainWork<D: Future, F> {
 impl<D, F, Err> Future for PrimedParachainWork<D, F>
 	where
 		D: Future<Item=BlockData,Error=Err>,
-		F: FnMut(&BlockId, &Collation) -> bool,
+		F: FnMut(&BlockId, &Collation) -> Result<Extrinsic, ()>,
 		Err: From<::std::io::Error>,
 {
 	type Item = Validated;
@@ -230,27 +230,30 @@ impl<D, F, Err> Future for PrimedParachainWork<D, F>
 		let candidate = &work.candidate_receipt;
 
 		let block = try_ready!(work.fetch_block_data.poll());
-		let is_good = (self.validate)(
+		let validation_res = (self.validate)(
 			&BlockId::hash(self.inner.relay_parent),
 			&Collation { block_data: block.clone(), receipt: candidate.clone() },
 		);
 
 		let candidate_hash = candidate.hash();
 
-		debug!(target: "consensus", "Making validity statement about candidate {}: is_good? {:?}", candidate_hash, is_good);
-		let validity_statement = match is_good {
-			true => GenericStatement::Valid(candidate_hash),
-			false => GenericStatement::Invalid(candidate_hash),
-		};
+		debug!(target: "consensus", "Making validity statement about candidate {}: is_good? {:?}",
+			candidate_hash, validation_res.is_ok());
 
-		let extrinsic = Extrinsic;
-		self.inner.extrinsic_store.make_available(Data {
-			relay_parent: self.inner.relay_parent,
-			parachain_id: work.candidate_receipt.parachain_index,
-			candidate_hash,
-			block_data: block.clone(),
-			extrinsic: Some(extrinsic.clone()),
-		})?;
+		let (extrinsic, validity_statement) = match validation_res {
+			Err(()) => (None, GenericStatement::Invalid(candidate_hash)),
+			Ok(extrinsic) => {
+				self.inner.extrinsic_store.make_available(Data {
+					relay_parent: self.inner.relay_parent,
+					parachain_id: work.candidate_receipt.parachain_index,
+					candidate_hash,
+					block_data: block.clone(),
+					extrinsic: Some(extrinsic.clone()),
+				})?;
+
+				(Some(extrinsic), GenericStatement::Valid(candidate_hash))
+			}
+		};
 
 		Ok(Async::Ready(Validated {
 			validity: validity_statement,
@@ -444,7 +447,6 @@ mod tests {
 	impl TableRouter for DummyRouter {
 		type Error = ::std::io::Error;
 		type FetchCandidate = ::futures::future::FutureResult<BlockData,Self::Error>;
-		type FetchExtrinsic = ::futures::future::FutureResult<Extrinsic,Self::Error>;
 
 		fn local_candidate(&self, _candidate: CandidateReceipt, _block_data: BlockData, _extrinsic: Extrinsic) {
 
@@ -452,9 +454,6 @@ mod tests {
 		fn fetch_block_data(&self, _candidate: &CandidateReceipt) -> Self::FetchCandidate {
 			future::ok(BlockData(vec![1, 2, 3, 4, 5]))
 		}
-		fn fetch_extrinsic_data(&self, _candidate: &CandidateReceipt) -> Self::FetchExtrinsic {
-			future::ok(Extrinsic)
-		}
 	}
 
 	#[test]
@@ -586,7 +585,9 @@ mod tests {
 			extrinsic_store: store.clone(),
 		};
 
-		let produced = producer.prime_with(|_, _| true).wait().unwrap();
+		let produced = producer.prime_with(|_, _| Ok(Extrinsic { outgoing_messages: Vec::new() }))
+			.wait()
+			.unwrap();
 
 		assert_eq!(produced.block_data, block_data);
 		assert_eq!(produced.validity, GenericStatement::Valid(hash));
@@ -624,7 +625,9 @@ mod tests {
 			extrinsic_store: store.clone(),
 		};
 
-		let produced = producer.prime_with(|_, _| true).wait().unwrap();
+		let produced = producer.prime_with(|_, _| Ok(Extrinsic { outgoing_messages: Vec::new() }))
+			.wait()
+			.unwrap();
 
 		assert_eq!(produced.block_data, block_data);