Collator for the "adder" (formerly basic-add) parachain and various small fixes

Cargo.toml

@@ -32,6 +32,9 @@ members = [
 	"polkadot/transaction-pool",
 	"polkadot/service",
 
+	"polkadot/test-parachains/adder",
+	"polkadot/test-parachains/adder/collator",
+
 	"substrate/bft",
 	"substrate/cli",
 	"substrate/client",
@@ -77,6 +80,7 @@ members = [
 ]
 exclude = [
 	"polkadot/runtime/wasm",
+	"polkadot/test-parachains/adder/wasm",
 	"demo/runtime/wasm",
 	"substrate/executor/wasm",
 	"substrate/pwasm-alloc",
@@ -92,4 +96,4 @@ is-it-maintained-open-issues = { repository = "paritytech/polkadot" }
 
 [profile.release]
 # Substrate runtime requires unwinding.
-panic = "unwind" 
+panic = "unwind"

common.sh

@@ -8,7 +8,7 @@ SRCS=(
 	"substrate/executor/wasm"
 	"demo/runtime/wasm"
 	"substrate/test-runtime/wasm"
-	"polkadot/parachain/test-chains/basic_add"
+	"polkadot/test-parachains/"
 )
 
 # Make pushd/popd silent.

polkadot/collator/src/lib.rs

@@ -60,6 +60,7 @@ extern crate polkadot_primitives;
 extern crate log;
 
 use std::collections::{BTreeSet, BTreeMap, HashSet};
+use std::fmt;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
@@ -68,12 +69,36 @@ use client::BlockchainEvents;
 use polkadot_api::PolkadotApi;
 use polkadot_primitives::{AccountId, BlockId, SessionKey};
 use polkadot_primitives::parachain::{self, BlockData, DutyRoster, HeadData, ConsolidatedIngress, Message, Id as ParaId};
-use polkadot_cli::{ServiceComponents, Service, CustomConfiguration, VersionInfo};
+use polkadot_cli::{ServiceComponents, Service, CustomConfiguration};
 use polkadot_cli::{Worker, IntoExit};
 use tokio::timer::Deadline;
 
+pub use polkadot_cli::VersionInfo;
+
 const COLLATION_TIMEOUT: Duration = Duration::from_secs(30);
 
+/// Error to return when the head data was invalid.
+#[derive(Clone, Copy, Debug)]
+pub struct InvalidHead;
+
+/// Collation errors.
+#[derive(Debug)]
+pub enum Error<R> {
+	/// Error on the relay-chain side of things.
+	Polkadot(R),
+	/// Error on the collator side of things.
+	Collator(InvalidHead),
+}
+
+impl<R: fmt::Display> fmt::Display for Error<R> {
+	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+		match *self {
+			Error::Polkadot(ref err) => write!(f, "Polkadot node error: {}", err),
+			Error::Collator(_) => write!(f, "Collator node error: Invalid head data"),
+		}
+	}
+}
+
 /// Parachain context needed for collation.
 ///
 /// This can be implemented through an externally attached service or a stub.
@@ -84,7 +109,7 @@ pub trait ParachainContext: Clone {
 		&self,
 		last_head: HeadData,
 		ingress: I,
-	) -> (BlockData, HeadData);
+	) -> Result<(BlockData, HeadData), InvalidHead>;
 }
 
 /// Relay chain context needed to collate.
@@ -154,18 +179,18 @@ pub fn collate<'a, R, P>(
 	para_context: P,
 	key: Arc<ed25519::Pair>,
 )
-	-> impl Future<Item=parachain::Collation, Error=R::Error> + 'a
+	-> impl Future<Item=parachain::Collation, Error=Error<R::Error>> + 'a
 	where
 		R: RelayChainContext + 'a,
 		R::Error: 'a,
 		R::FutureEgress: 'a,
 		P: ParachainContext + 'a,
 {
-	collate_ingress(relay_context).map(move |ingress| {
+	collate_ingress(relay_context).map_err(Error::Polkadot).and_then(move |ingress| {
 		let (block_data, head_data) = para_context.produce_candidate(
 			last_head,
 			ingress.0.iter().flat_map(|&(id, ref msgs)| msgs.iter().cloned().map(move |msg| (id, msg)))
-		);
+		).map_err(Error::Collator)?;
 
 		let block_data_hash = block_data.hash();
 		let signature = key.sign(&block_data_hash.0[..]).into();
@@ -181,10 +206,10 @@ pub fn collate<'a, R, P>(
 			block_data_hash,
 		};
 
-		parachain::Collation {
+		Ok(parachain::Collation {
 			receipt,
 			block_data,
-		}
+		})
 	})
 }
 
@@ -248,7 +273,7 @@ impl<P, E> Worker for CollationNode<P, E> where
 					($e:expr) => {
 						match $e {
 							Ok(x) => x,
-							Err(e) => return future::Either::A(future::err(e)),
+							Err(e) => return future::Either::A(future::err(Error::Polkadot(e))),
 						}
 					}
 				}
@@ -323,17 +348,19 @@ fn compute_targets(para_id: ParaId, session_keys: &[SessionKey], roster: DutyRos
 ///
 /// Provide a future which resolves when the node should exit.
 /// This function blocks until done.
-pub fn run_collator<P, E>(
+pub fn run_collator<P, E, I, ArgT>(
 	parachain_context: P,
 	para_id: ParaId,
 	exit: E,
 	key: Arc<ed25519::Pair>,
-	args: Vec<::std::ffi::OsString>,
+	args: I,
 	version: VersionInfo,
 ) -> polkadot_cli::error::Result<()> where
 	P: ParachainContext + Send + 'static,
 	E: IntoFuture<Item=(),Error=()>,
 	E::Future: Send + Clone + 'static,
+	I: IntoIterator<Item=ArgT>,
+	ArgT: Into<std::ffi::OsString> + Clone,
 {
 	let node_logic = CollationNode { parachain_context, exit: exit.into_future(), para_id, key };
 	polkadot_cli::run(args, node_logic, version)

polkadot/consensus/src/lib.rs

@@ -273,8 +273,13 @@ impl<C, N, P> bft::Environment<Block> for ProposerFactory<C, N, P>
 			sign_with.public().into(),
 		)?;
 
+		info!("Starting consensus session on top of parent {:?}. Local parachain duty is {:?}",
+			parent_hash, local_duty.validation);
+
 		let active_parachains = self.client.active_parachains(&id)?;
 
+		debug!(target: "consensus", "Active parachains: {:?}", active_parachains);
+
 		let n_parachains = active_parachains.len();
 		let table = Arc::new(SharedTable::new(group_info, sign_with.clone(), parent_hash));
 		let (router, input, output) = self.network.communication_for(

polkadot/consensus/src/shared_table/mod.rs

@@ -143,6 +143,7 @@ impl SharedTableInner {
 						fetch_block_data,
 						fetch_extrinsic,
 						evaluate: checking_validity,
+						ensure_available: checking_availability,
 					})
 				}
 			}
@@ -206,6 +207,7 @@ struct Work<D: Future, E: Future> {
 	fetch_block_data: future::Fuse<D>,
 	fetch_extrinsic: Option<future::Fuse<E>>,
 	evaluate: bool,
+	ensure_available: bool,
 }
 
 /// Primed statement producer.
@@ -235,31 +237,35 @@ impl<D, E, C, Err> Future for PrimedStatementProducer<D, E, C>
 				});
 
 				let hash = work.candidate_receipt.hash();
+
+				debug!(target: "consensus", "Making validity statement about candidate {}: is_good? {:?}", hash, is_good);
 				self.inner.produced_statements.validity = match is_good {
 					Some(true) => Some(GenericStatement::Valid(hash)),
 					Some(false) => Some(GenericStatement::Invalid(hash)),
 					None => None,
 				};
-			}
-		}
 
-		if let Some(ref mut fetch_extrinsic) = work.fetch_extrinsic {
-			if let Async::Ready(extrinsic) = fetch_extrinsic.poll()? {
-				self.inner.produced_statements.extrinsic = Some(extrinsic);
+				work.evaluate = false;
 			}
 		}
 
-		let done = self.inner.produced_statements.block_data.is_some() && {
-			if work.evaluate {
-				true
-			} else if self.inner.produced_statements.extrinsic.is_some() {
+		if let Async::Ready(Some(extrinsic)) = work.fetch_extrinsic.poll()? {
+			if work.ensure_available {
+				let hash = work.candidate_receipt.hash();
+				debug!(target: "consensus", "Claiming candidate {} available.", hash);
+
+				// TODO: actually wait for block data and then ensure availability.
+				self.inner.produced_statements.extrinsic = Some(extrinsic);
 				self.inner.produced_statements.availability =
-					Some(GenericStatement::Available(work.candidate_receipt.hash()));
+					Some(GenericStatement::Available(hash));
 
-				true
-			} else {
-				false
+				work.ensure_available = false;
 			}
+		}
+
+		let done = match (work.evaluate, work.ensure_available) {
+			(false, false) => true,
+			_ => false,
 		};
 
 		if done {
@@ -356,10 +362,25 @@ impl SharedTable {
 	}
 
 	/// Sign and import a local statement.
-	pub fn sign_and_import(&self, statement: table::Statement) -> SignedStatement {
-		let proposed_digest = match statement {
-			GenericStatement::Candidate(ref c) => Some(c.hash()),
-			_ => None,
+	///
+	/// For candidate statements, this may also produce a second signed statement
+	/// concerning the availability of the candidate data.
+	pub fn sign_and_import(&self, statement: table::Statement)
+		-> (SignedStatement, Option<SignedStatement>)
+	{
+		let (proposed_digest, availability) = match statement {
+			GenericStatement::Candidate(ref c) => {
+				let mut availability = None;
+				let hash = c.hash();
+
+				// TODO: actually store the data in an availability store of some kind.
+				if self.context.is_availability_guarantor_of(&self.context.local_id(), &c.parachain_index) {
+					availability = Some(self.context.sign_statement(GenericStatement::Available(hash)));
+				}
+
+				(Some(hash), availability)
+			}
+			_ => (None, None),
 		};
 
 		let signed_statement = self.context.sign_statement(statement);
@@ -370,7 +391,13 @@ impl SharedTable {
 		}
 
 		inner.table.import_statement(&*self.context, signed_statement.clone());
-		signed_statement
+
+		// ensure the availability statement is imported after the candidate.
+		if let Some(a) = availability.clone() {
+			inner.table.import_statement(&*self.context, a);
+		}
+
+		(signed_statement, availability)
 	}
 
 	/// Execute a closure using a specific candidate.
@@ -543,5 +570,6 @@ mod tests {
 
 		assert!(producer.work.fetch_extrinsic.is_some(), "should fetch extrinsic when guaranteeing availability");
 		assert!(!producer.work.evaluate, "should not evaluate validity");
+		assert!(producer.work.ensure_available);
 	}
 }

polkadot/network/src/consensus.rs

@@ -176,6 +176,7 @@ impl<P: LocalPolkadotApi + Send + Sync + 'static> MessageProcessTask<P> {
 				}
 			}
 			ConsensusMessage::ChainSpecific(msg, _) => {
+				debug!(target: "consensus", "Processing consensus statement for live consensus");
 				if let Some(Message::Statement(parent_hash, statement)) = Decode::decode(&mut msg.as_slice()) {
 					if ::polkadot_consensus::check_statement(&statement.statement, &statement.signature, statement.sender, &parent_hash) {
 						self.table_router.import_statement(statement);