Availability/Extrinsic store (#465)

polkadot/consensus/src/collation.rs

@@ -73,6 +73,11 @@ impl<C: Collators, P: PolkadotApi> CollationFetch<C, P> {
 			live_fetch: None,
 		}
 	}
+
+	/// Access the underlying relay parent hash.
+	pub fn relay_parent(&self) -> Hash {
+		self.relay_parent_hash
+	}
 }
 
 impl<C: Collators, P: PolkadotApi> Future for CollationFetch<C, P> {

polkadot/consensus/src/lib.rs

@@ -32,6 +32,7 @@
 extern crate ed25519;
 extern crate parking_lot;
 extern crate polkadot_api;
+extern crate polkadot_availability_store as extrinsic_store;
 extern crate polkadot_statement_table as table;
 extern crate polkadot_parachain as parachain;
 extern crate polkadot_transaction_pool as transaction_pool;
@@ -66,6 +67,7 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use codec::{Decode, Encode};
+use extrinsic_store::Store as ExtrinsicStore;
 use polkadot_api::PolkadotApi;
 use polkadot_primitives::{Hash, Block, BlockId, BlockNumber, Header, Timestamp, SessionKey};
 use polkadot_primitives::parachain::{Id as ParaId, Chain, DutyRoster, BlockData, Extrinsic as ParachainExtrinsic, CandidateReceipt, CandidateSignature};
@@ -236,6 +238,8 @@ pub struct ProposerFactory<C, N, P> {
 	pub handle: TaskExecutor,
 	/// The duration after which parachain-empty blocks will be allowed.
 	pub parachain_empty_duration: Duration,
+	/// Store for extrinsic data.
+	pub extrinsic_store: ExtrinsicStore,
 }
 
 impl<C, N, P> bft::Environment<Block> for ProposerFactory<C, N, P>
@@ -279,7 +283,7 @@ impl<C, N, P> bft::Environment<Block> for ProposerFactory<C, N, P>
 		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 table = Arc::new(SharedTable::new(group_info, sign_with.clone(), parent_hash, self.extrinsic_store.clone()));
 		let (router, input, output) = self.network.communication_for(
 			authorities,
 			table.clone(),
@@ -309,6 +313,7 @@ impl<C, N, P> bft::Environment<Block> for ProposerFactory<C, N, P>
 			router.clone(),
 			&self.handle,
 			collation_work,
+			self.extrinsic_store.clone(),
 		);
 
 		let proposer = Proposer {
@@ -334,19 +339,42 @@ fn dispatch_collation_work<R, C, P>(
 	router: R,
 	handle: &TaskExecutor,
 	work: Option<CollationFetch<C, P>>,
+	extrinsic_store: ExtrinsicStore,
 ) -> exit_future::Signal where
 	C: Collators + Send + 'static,
 	P: PolkadotApi + Send + Sync + 'static,
 	<C::Collation as IntoFuture>::Future: Send + 'static,
 	R: TableRouter + Send + 'static,
 {
+	use extrinsic_store::Data;
+
 	let (signal, exit) = exit_future::signal();
+
+	let work = match work {
+		Some(w) => w,
+		None => return signal,
+	};
+
+	let relay_parent = work.relay_parent();
 	let handled_work = work.then(move |result| match result {
-		Ok(Some((collation, extrinsic))) => {
-			router.local_candidate(collation.receipt, collation.block_data, extrinsic);
+		Ok((collation, extrinsic)) => {
+			let res = extrinsic_store.make_available(Data {
+				relay_parent,
+				parachain_id: collation.receipt.parachain_index,
+				candidate_hash: collation.receipt.hash(),
+				block_data: collation.block_data.clone(),
+				extrinsic: Some(extrinsic.clone()),
+			});
+
+			match res {
+				Ok(()) =>
+					router.local_candidate(collation.receipt, collation.block_data, extrinsic),
+				Err(e) =>
+					warn!(target: "consensus", "Failed to make collation data available: {:?}", e),
+			}
+
 			Ok(())
 		}
-		Ok(None) => Ok(()),
 		Err(_e) => {
 			warn!(target: "consensus", "Failed to collate candidate");
 			Ok(())

polkadot/consensus/src/service.rs

@@ -28,12 +28,13 @@ use std::time::{Duration, Instant};
 use std::sync::Arc;
 
 use bft::{self, BftService};
-use client::{BlockchainEvents, ChainHead};
+use client::{BlockchainEvents, ChainHead, BlockBody};
 use ed25519;
 use futures::prelude::*;
 use polkadot_api::LocalPolkadotApi;
 use polkadot_primitives::{Block, Header};
 use transaction_pool::TransactionPool;
+use extrinsic_store::Store as ExtrinsicStore;
 
 use tokio::executor::current_thread::TaskExecutor as LocalThreadHandle;
 use tokio::runtime::TaskExecutor as ThreadPoolHandle;
@@ -89,6 +90,56 @@ fn start_bft<F, C>(
     }
 }
 
+// creates a task to prune redundant entries in availability store upon block finalization
+//
+// NOTE: this will need to be changed to finality notification rather than
+// block import notifications when the consensus switches to non-instant finality.
+fn prune_unneeded_availability<C>(client: Arc<C>, extrinsic_store: ExtrinsicStore)
+	-> impl Future<Item=(),Error=()> + Send
+	where C: Send + Sync + BlockchainEvents<Block> + BlockBody<Block> + 'static
+{
+	use codec::{Encode, Decode};
+	use polkadot_primitives::BlockId;
+	use polkadot_runtime::CheckedBlock;
+
+	enum NotifyError {
+		NoBody(::client::error::Error),
+		UnexpectedFormat,
+		ExtrinsicsWrong,
+	}
+
+	impl NotifyError {
+		fn log(&self, hash: &::polkadot_primitives::Hash) {
+			match *self {
+				NotifyError::NoBody(ref err) => warn!("Failed to fetch block body for imported block {:?}: {:?}", hash, err),
+				NotifyError::UnexpectedFormat => warn!("Consensus outdated: Block {:?} has unexpected body format", hash),
+				NotifyError::ExtrinsicsWrong => warn!("Consensus outdated: Failed to fetch block body for imported block {:?}", hash),
+			}
+		}
+	}
+
+	client.import_notification_stream()
+		.for_each(move |notification| {
+			let checked_block = client.block_body(&BlockId::hash(notification.hash))
+				.map_err(NotifyError::NoBody)
+				.map(|b| ::polkadot_runtime::Block::decode(&mut b.encode().as_slice()))
+				.and_then(|maybe_block| maybe_block.ok_or(NotifyError::UnexpectedFormat))
+				.and_then(|block| CheckedBlock::new(block).map_err(|_| NotifyError::ExtrinsicsWrong));
+
+			match checked_block {
+				Ok(block) => {
+					let candidate_hashes = block.parachain_heads().iter().map(|c| c.hash()).collect();
+					if let Err(e) = extrinsic_store.candidates_finalized(notification.header.parent_hash, candidate_hashes) {
+						warn!(target: "consensus", "Failed to prune unneeded available data: {:?}", e);
+					}
+				}
+				Err(e) => e.log(&notification.hash)
+			}
+
+			Ok(())
+		})
+}
+
 /// Consensus service. Starts working when created.
 pub struct Service {
 	thread: Option<thread::JoinHandle<()>>,
@@ -105,10 +156,11 @@ impl Service {
 		thread_pool: ThreadPoolHandle,
 		parachain_empty_duration: Duration,
 		key: ed25519::Pair,
+		extrinsic_store: ExtrinsicStore,
 	) -> Service
 		where
 			A: LocalPolkadotApi + Send + Sync + 'static,
-			C: BlockchainEvents<Block> + ChainHead<Block> + bft::BlockImport<Block> + bft::Authorities<Block> + Send + Sync + 'static,
+			C: BlockchainEvents<Block> + ChainHead<Block> + BlockBody<Block> + bft::BlockImport<Block> + bft::Authorities<Block> + Send + Sync + 'static,
 			N: Network + Collators + Send + 'static,
 			N::TableRouter: Send + 'static,
 			<N::Collation as IntoFuture>::Future: Send + 'static,
@@ -124,7 +176,8 @@ impl Service {
 				collators: network.clone(),
 				network,
 				parachain_empty_duration,
-				handle: thread_pool,
+				handle: thread_pool.clone(),
+				extrinsic_store: extrinsic_store.clone(),
 			};
 			let bft_service = Arc::new(BftService::new(client.clone(), key, factory));
 
@@ -172,6 +225,14 @@ impl Service {
 
 			runtime.spawn(notifications);
 			runtime.spawn(timed);
+
+			let prune_available = prune_unneeded_availability(client, extrinsic_store)
+				.select(exit.clone())
+				.then(|_| Ok(()));
+
+			// spawn this on the tokio executor since it's fine on a thread pool.
+			thread_pool.spawn(prune_available);
+
 			if let Err(e) = runtime.block_on(exit) {
 				debug!("BFT event loop error {:?}", e);
 			}

polkadot/consensus/src/shared_table/mod.rs

@@ -20,6 +20,7 @@
 use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 
+use extrinsic_store::{Data, Store as ExtrinsicStore};
 use table::{self, Table, Context as TableContextTrait};
 use polkadot_primitives::{Hash, SessionKey};
 use polkadot_primitives::parachain::{Id as ParaId, BlockData, Collation, Extrinsic, CandidateReceipt};
@@ -82,6 +83,7 @@ struct SharedTableInner {
 	checked_validity: HashSet<Hash>,
 	checked_availability: HashSet<Hash>,
 	trackers: Vec<IncludabilitySender>,
+	extrinsic_store: ExtrinsicStore,
 }
 
 impl SharedTableInner {
@@ -153,6 +155,8 @@ impl SharedTableInner {
 
 		work.map(|work| StatementProducer {
 			produced_statements: Default::default(),
+			extrinsic_store: self.extrinsic_store.clone(),
+			relay_parent: context.parent_hash.clone(),
 			work
 		})
 	}
@@ -186,6 +190,8 @@ pub struct ProducedStatements {
 pub struct StatementProducer<D: Future, E: Future> {
 	produced_statements: ProducedStatements,
 	work: Work<D, E>,
+	relay_parent: Hash,
+	extrinsic_store: ExtrinsicStore,
 }
 
 impl<D: Future, E: Future> StatementProducer<D, E> {
@@ -221,25 +227,32 @@ impl<D, E, C, Err> Future for PrimedStatementProducer<D, E, C>
 		D: Future<Item=BlockData,Error=Err>,
 		E: Future<Item=Extrinsic,Error=Err>,
 		C: FnMut(Collation) -> Option<bool>,
+		Err: From<::std::io::Error>,
 {
 	type Item = ProducedStatements;
 	type Error = Err;
 
 	fn poll(&mut self) -> Poll<ProducedStatements, Err> {
 		let work = &mut self.inner.work;
+		let candidate = &work.candidate_receipt;
+		let statements = &mut self.inner.produced_statements;
+
+		let mut candidate_hash = None;
+		let mut candidate_hash = move ||
+			candidate_hash.get_or_insert_with(|| candidate.hash()).clone();
 
 		if let Async::Ready(block_data) = work.fetch_block_data.poll()? {
-			self.inner.produced_statements.block_data = Some(block_data.clone());
+			statements.block_data = Some(block_data.clone());
 			if work.evaluate {
 				let is_good = (self.check_candidate)(Collation {
 					block_data,
 					receipt: work.candidate_receipt.clone(),
 				});
 
-				let hash = work.candidate_receipt.hash();
+				let hash = candidate_hash();
 
 				debug!(target: "consensus", "Making validity statement about candidate {}: is_good? {:?}", hash, is_good);
-				self.inner.produced_statements.validity = match is_good {
+				statements.validity = match is_good {
 					Some(true) => Some(GenericStatement::Valid(hash)),
 					Some(false) => Some(GenericStatement::Invalid(hash)),
 					None => None,
@@ -251,12 +264,11 @@ impl<D, E, C, Err> Future for PrimedStatementProducer<D, E, C>
 
 		if let Async::Ready(Some(extrinsic)) = work.fetch_extrinsic.poll()? {
 			if work.ensure_available {
-				let hash = work.candidate_receipt.hash();
+				let hash = candidate_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 =
+				statements.extrinsic = Some(extrinsic);
+				statements.availability =
 					Some(GenericStatement::Available(hash));
 
 				work.ensure_available = false;
@@ -269,7 +281,18 @@ impl<D, E, C, Err> Future for PrimedStatementProducer<D, E, C>
 		};
 
 		if done {
-			Ok(Async::Ready(::std::mem::replace(&mut self.inner.produced_statements, Default::default())))
+			// commit claimed-available data to disk before returning statements from the future.
+			if let (&Some(ref block), extrinsic) = (&statements.block_data, &statements.extrinsic) {
+				self.inner.extrinsic_store.make_available(Data {
+					relay_parent: self.inner.relay_parent,
+					parachain_id: work.candidate_receipt.parachain_index,
+					candidate_hash: candidate_hash(),
+					block_data: block.clone(),
+					extrinsic: extrinsic.clone(),
+				})?;
+			}
+
+			Ok(Async::Ready(::std::mem::replace(statements, Default::default())))
 		} else {
 			Ok(Async::NotReady)
 		}
@@ -296,7 +319,12 @@ impl SharedTable {
 	///
 	/// Provide the key to sign with, and the parent hash of the relay chain
 	/// block being built.
-	pub fn new(groups: HashMap<ParaId, GroupInfo>, key: Arc<::ed25519::Pair>, parent_hash: Hash) -> Self {
+	pub fn new(
+		groups: HashMap<ParaId, GroupInfo>,
+		key: Arc<::ed25519::Pair>,
+		parent_hash: Hash,
+		extrinsic_store: ExtrinsicStore,
+	) -> Self {
 		SharedTable {
 			context: Arc::new(TableContext { groups, key, parent_hash }),
 			inner: Arc::new(Mutex::new(SharedTableInner {
@@ -305,6 +333,7 @@ impl SharedTable {
 				checked_validity: HashSet::new(),
 				checked_availability: HashSet::new(),
 				trackers: Vec::new(),
+				extrinsic_store,
 			}))
 		}
 	}
@@ -457,9 +486,9 @@ mod tests {
 	#[derive(Clone)]
 	struct DummyRouter;
 	impl TableRouter for DummyRouter {
-		type Error = ();
-		type FetchCandidate = ::futures::future::Empty<BlockData,()>;
-		type FetchExtrinsic = ::futures::future::Empty<Extrinsic,()>;
+		type Error = ::std::io::Error;
+		type FetchCandidate = ::futures::future::Empty<BlockData,Self::Error>;
+		type FetchExtrinsic = ::futures::future::Empty<Extrinsic,Self::Error>;
 
 		fn local_candidate(&self, _candidate: CandidateReceipt, _block_data: BlockData, _extrinsic: Extrinsic) {
 
@@ -491,7 +520,12 @@ mod tests {
 			needed_availability: 0,
 		});
 
-		let shared_table = SharedTable::new(groups, local_key.clone(), parent_hash);
+		let shared_table = SharedTable::new(
+			groups,
+			local_key.clone(),
+			parent_hash,
+			ExtrinsicStore::new_in_memory(),
+		);
 
 		let candidate = CandidateReceipt {
 			parachain_index: para_id,
@@ -541,7 +575,12 @@ mod tests {
 			needed_availability: 1,
 		});
 
-		let shared_table = SharedTable::new(groups, local_key.clone(), parent_hash);
+		let shared_table = SharedTable::new(
+			groups,
+			local_key.clone(),
+			parent_hash,
+			ExtrinsicStore::new_in_memory(),
+		);
 
 		let candidate = CandidateReceipt {
 			parachain_index: para_id,
@@ -572,4 +611,93 @@ mod tests {
 		assert!(!producer.work.evaluate, "should not evaluate validity");
 		assert!(producer.work.ensure_available);
 	}
+
+	#[test]
+	fn evaluate_makes_block_data_available() {
+		let store = ExtrinsicStore::new_in_memory();
+		let relay_parent = [0; 32].into();
+		let para_id = 5.into();
+		let block_data = BlockData(vec![1, 2, 3]);
+
+		let candidate = CandidateReceipt {
+			parachain_index: para_id,
+			collator: [1; 32].into(),
+			signature: Default::default(),
+			head_data: ::polkadot_primitives::parachain::HeadData(vec![1, 2, 3, 4]),
+			balance_uploads: Vec::new(),
+			egress_queue_roots: Vec::new(),
+			fees: 1_000_000,
+			block_data_hash: [2; 32].into(),
+		};
+
+		let hash = candidate.hash();
+
+		let block_data_res: ::std::io::Result<_> = Ok(block_data.clone());
+		let producer: StatementProducer<_, future::Empty<_, _>> = StatementProducer {
+			produced_statements: Default::default(),
+			work: Work {
+				candidate_receipt: candidate,
+				fetch_block_data: block_data_res.into_future().fuse(),
+				fetch_extrinsic: None,
+				evaluate: true,
+				ensure_available: false,
+			},
+			relay_parent,
+			extrinsic_store: store.clone(),
+		};
+
+		let produced = producer.prime(|_| Some(true)).wait().unwrap();
+
+		assert_eq!(produced.block_data.as_ref(), Some(&block_data));
+		assert!(produced.validity.is_some());
+		assert!(produced.availability.is_none());
+
+		assert_eq!(store.block_data(relay_parent, hash).unwrap(), block_data);
+		assert!(store.extrinsic(relay_parent, hash).is_none());
+	}
+
+	#[test]
+	fn full_availability() {
+		let store = ExtrinsicStore::new_in_memory();
+		let relay_parent = [0; 32].into();
+		let para_id = 5.into();
+		let block_data = BlockData(vec![1, 2, 3]);
+
+		let candidate = CandidateReceipt {
+			parachain_index: para_id,
+			collator: [1; 32].into(),
+			signature: Default::default(),
+			head_data: ::polkadot_primitives::parachain::HeadData(vec![1, 2, 3, 4]),
+			balance_uploads: Vec::new(),
+			egress_queue_roots: Vec::new(),
+			fees: 1_000_000,
+			block_data_hash: [2; 32].into(),
+		};
+
+		let hash = candidate.hash();
+
+		let block_data_res: ::std::io::Result<_> = Ok(block_data.clone());
+		let extrinsic_res: ::std::io::Result<_> = Ok(Extrinsic);
+		let producer = StatementProducer {
+			produced_statements: Default::default(),
+			work: Work {
+				candidate_receipt: candidate,
+				fetch_block_data: block_data_res.into_future().fuse(),
+				fetch_extrinsic: Some(extrinsic_res.into_future().fuse()),
+				evaluate: false,
+				ensure_available: true,
+			},
+			relay_parent,
+			extrinsic_store: store.clone(),
+		};
+
+		let produced = producer.prime(|_| Some(true)).wait().unwrap();
+
+		assert_eq!(produced.block_data.as_ref(), Some(&block_data));
+		assert!(produced.validity.is_none());
+		assert!(produced.availability.is_some());
+
+		assert_eq!(store.block_data(relay_parent, hash).unwrap(), block_data);
+		assert!(store.extrinsic(relay_parent, hash).is_some());
+	}
 }