Erasure encoding availability (#345)

* Erasure encoding availability initial commit

 * Modifications to availability store to keep chunks as well as
   reconstructed blocks and extrinsics.
 * Gossip messages containig signed erasure chunks.
 * Requesting eraure chunks with polkadot-specific messages.
 * Validation of erasure chunk messages.

* Apply suggestions from code review

Co-Authored-By: Luke Schoen <ltfschoen@users.noreply.github.com>

* Fix build after a merge

* Gossip erasure chunk messages under their own topic

* erasure_chunks should use the appropriate topic

* Updates Cargo.lock

* Fixes after merge

* Removes a couple of leftover pieces of code

* Fixes simple stuff from review

* Updates erasure and storage for more flexible logic

* Changes validation and candidate receipt production.

* Adds add_erasure_chunks method

* Fixes most of the nits

* Better validate_collation and validate_receipt functions

* Fixes the tests

* Apply suggestions from code review

Co-Authored-By: Robert Habermeier <rphmeier@gmail.com>

* Removes unwrap() calls

* Removes ErasureChunks primitive

* Removes redundant fields from ErasureChunk struct

* AvailabilityStore should store CandidateReceipt

* Changes the way chunk messages are imported and validated.

 * Availability store now stores a validator_index and n_validators for
 each relay_parent.
 * Availability store now also stores candidate receipts.
 * Removes importing chunks in the table and moves it into network
 gossip validation.
 * Validation of erasure messages id done against receipts that are
 stored in the availability store.

* Correctly compute topics for erasure messages

* Removes an unused parameter

* Refactors availability db querying into a helper

* Adds the apis described in the writeup

* Adds a runtime api to extract erasure roots form raw extrinsics.

* Adds a barebone BlockImport impl for avalability store

* Adds the implementation of the availability worker

* Fix build after the merge with master.

* Make availability store API async

* Bring back the default wasmtime feature

* Lines width

* Bump runtime version

* Formatting and dead code elimination

* some style nits (#1)

* More nits and api cleanup

* Disable wasm CI for availability-store

* Another nit

* Formatting

.gitlab-ci.yml

@@ -144,7 +144,6 @@ check-web-wasm:                    &test
   script:
     # WASM support is in progress. As more and more crates support WASM, we
     # should add entries here. See https://github.com/paritytech/polkadot/issues/625
-    - time cargo build --locked --target=wasm32-unknown-unknown --manifest-path availability-store/Cargo.toml
     - time cargo build --locked --target=wasm32-unknown-unknown --manifest-path executor/Cargo.toml
     - time cargo build --locked --target=wasm32-unknown-unknown --manifest-path erasure-coding/Cargo.toml
     - time cargo build --locked --target=wasm32-unknown-unknown --manifest-path parachain/Cargo.toml

Cargo.lock

@@ -777,6 +777,19 @@ dependencies = [
  "syn 0.15.44 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "derive_more"
+version = "0.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "lazy_static 1.4.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "proc-macro2 0.4.30 (registry+https://github.com/rust-lang/crates.io-index)",
+ "quote 0.6.13 (registry+https://github.com/rust-lang/crates.io-index)",
+ "regex 1.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "rustc_version 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)",
+ "syn 0.15.44 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "derive_more"
 version = "0.99.2"
@@ -3455,14 +3468,29 @@ dependencies = [
 name = "polkadot-availability-store"
 version = "0.7.5"
 dependencies = [
+ "derive_more 0.99.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "exit-future 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures01 0.1.29 (registry+https://github.com/rust-lang/crates.io-index)",
  "kvdb 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "kvdb-memorydb 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "kvdb-rocksdb 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)",
  "parity-scale-codec 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "parking_lot 0.9.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "polkadot-erasure-coding 0.7.5",
  "polkadot-primitives 0.7.5",
+ "polkadot-runtime 0.7.5",
+ "sc-client 2.0.0 (git+https://github.com/paritytech/substrate?branch=polkadot-master)",
+ "sc-client-api 2.0.0 (git+https://github.com/paritytech/substrate?branch=polkadot-master)",
+ "sc-keystore 2.0.0 (git+https://github.com/paritytech/substrate?branch=polkadot-master)",
+ "sc-network 2.0.0 (git+https://github.com/paritytech/substrate?branch=polkadot-master)",
+ "sp-api 2.0.0 (git+https://github.com/paritytech/substrate?branch=polkadot-master)",
+ "sp-blockchain 2.0.0 (git+https://github.com/paritytech/substrate?branch=polkadot-master)",
+ "sp-consensus 2.0.0 (git+https://github.com/paritytech/substrate?branch=polkadot-master)",
  "sp-core 2.0.0 (git+https://github.com/paritytech/substrate?branch=polkadot-master)",
+ "sp-runtime 2.0.0 (git+https://github.com/paritytech/substrate?branch=polkadot-master)",
+ "tokio 0.1.22 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -3505,6 +3533,7 @@ dependencies = [
 name = "polkadot-erasure-coding"
 version = "0.7.5"
 dependencies = [
+ "derive_more 0.15.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "parity-scale-codec 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "polkadot-primitives 0.7.5",
  "reed-solomon-erasure 4.0.0 (git+https://github.com/paritytech/reed-solomon-erasure)",
@@ -3532,6 +3561,7 @@ dependencies = [
  "parity-scale-codec 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "parking_lot 0.9.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "polkadot-availability-store 0.7.5",
+ "polkadot-erasure-coding 0.7.5",
  "polkadot-primitives 0.7.5",
  "polkadot-validation 0.7.5",
  "sc-client 2.0.0 (git+https://github.com/paritytech/substrate?branch=polkadot-master)",
@@ -3578,6 +3608,7 @@ dependencies = [
  "sp-runtime 2.0.0 (git+https://github.com/paritytech/substrate?branch=polkadot-master)",
  "sp-serializer 2.0.0 (git+https://github.com/paritytech/substrate?branch=polkadot-master)",
  "sp-std 2.0.0 (git+https://github.com/paritytech/substrate?branch=polkadot-master)",
+ "sp-trie 2.0.0 (git+https://github.com/paritytech/substrate?branch=polkadot-master)",
  "sp-version 2.0.0 (git+https://github.com/paritytech/substrate?branch=polkadot-master)",
 ]
 
@@ -3731,6 +3762,7 @@ dependencies = [
  "parity-scale-codec 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "parking_lot 0.9.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "polkadot-availability-store 0.7.5",
+ "polkadot-erasure-coding 0.7.5",
  "polkadot-parachain 0.7.5",
  "polkadot-primitives 0.7.5",
  "polkadot-runtime 0.7.5",
@@ -6923,6 +6955,7 @@ dependencies = [
 "checksum curve25519-dalek 1.2.3 (registry+https://github.com/rust-lang/crates.io-index)" = "8b7dcd30ba50cdf88b55b033456138b7c0ac4afdc436d82e1b79f370f24cc66d"
 "checksum data-encoding 2.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "f4f47ca1860a761136924ddd2422ba77b2ea54fe8cc75b9040804a0d9d32ad97"
 "checksum derive_more 0.14.1 (registry+https://github.com/rust-lang/crates.io-index)" = "6d944ac6003ed268757ef1ee686753b57efc5fcf0ebe7b64c9fc81e7e32ff839"
+"checksum derive_more 0.15.0 (registry+https://github.com/rust-lang/crates.io-index)" = "7a141330240c921ec6d074a3e188a7c7ef95668bb95e7d44fa0e5778ec2a7afe"
 "checksum derive_more 0.99.2 (registry+https://github.com/rust-lang/crates.io-index)" = "2159be042979966de68315bce7034bb000c775f22e3e834e1c52ff78f041cae8"
 "checksum difference 2.0.0 (registry+https://github.com/rust-lang/crates.io-index)" = "524cbf6897b527295dff137cec09ecf3a05f4fddffd7dfcd1585403449e74198"
 "checksum digest 0.8.1 (registry+https://github.com/rust-lang/crates.io-index)" = "f3d0c8c8752312f9713efd397ff63acb9f85585afbf179282e720e7704954dd5"

availability-store/Cargo.toml

@@ -7,9 +7,24 @@ edition = "2018"
 
 [dependencies]
 polkadot-primitives = { path = "../primitives" }
+polkadot-erasure-coding = { path = "../erasure-coding" }
+polkadot-runtime = { path = "../runtime" }
 parking_lot = "0.9.0"
+derive_more = "0.99"
 log = "0.4.8"
+futures01 = "0.1.17"
+futures = { package = "futures", version = "0.3.1", features = ["compat"] }
+tokio = "0.1.7"
+exit-future = "0.1"
 codec = { package = "parity-scale-codec", version = "1.1.0", default-features = false, features = ["derive"] }
+sc-network = { git = "https://github.com/paritytech/substrate", branch = "polkadot-master" }
+consensus_common = { package = "sp-consensus", git = "https://github.com/paritytech/substrate", branch = "polkadot-master" }
+client = { package = "sc-client-api", git = "https://github.com/paritytech/substrate", branch = "polkadot-master" }
+sc-client = { git = "https://github.com/paritytech/substrate", branch = "polkadot-master" }
+sp-runtime = { git = "https://github.com/paritytech/substrate", default-features = false, branch = "polkadot-master" }
+sp-blockchain = { git = "https://github.com/paritytech/substrate", branch = "polkadot-master" }
+sp-api = { git = "https://github.com/paritytech/substrate", branch = "polkadot-master" }
+keystore = { package = "sc-keystore", git = "https://github.com/paritytech/substrate", branch = "polkadot-master" }
 sp-core = { git = "https://github.com/paritytech/substrate", branch = "polkadot-master" }
 kvdb = "0.1.1"
 kvdb-memorydb = "0.1.2"

availability-store/src/lib.rs

@@ -14,28 +14,56 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
-//! Persistent database for parachain data: PoV block data and outgoing messages.
+//! Persistent database for parachain data: PoV block data, erasure-coding chunks and outgoing messages.
 //!
 //! This will be written into during the block validation pipeline, and queried
 //! by networking code in order to circulate required data and maintain availability
 //! of it.
 
-use codec::{Encode, Decode};
-use kvdb::{KeyValueDB, DBTransaction};
-use polkadot_primitives::Hash;
-use polkadot_primitives::parachain::{Id as ParaId, BlockData, Message};
+#![warn(missing_docs)]
+
+use futures::prelude::*;
+use futures::channel::{mpsc, oneshot};
+use keystore::KeyStorePtr;
+use polkadot_primitives::{
+	Hash, Block,
+	parachain::{
+		Id as ParaId, BlockData, CandidateReceipt, Message, AvailableMessages, ErasureChunk,
+		ParachainHost,
+	},
+};
+use sp_runtime::traits::{BlakeTwo256, Hash as HashT, ProvideRuntimeApi};
+use sp_blockchain::{Result as ClientResult};
+use client::{
+	BlockchainEvents, BlockBody,
+};
+use sp_api::ApiExt;
+
 use log::warn;
 
+use std::sync::Arc;
 use std::collections::HashSet;
 use std::path::PathBuf;
-use std::sync::Arc;
 use std::io;
 
-mod columns {
-	pub const DATA: Option<u32> = Some(0);
-	pub const META: Option<u32> = Some(1);
-	pub const NUM_COLUMNS: u32 = 2;
-}
+mod worker;
+mod store;
+
+pub use worker::AvailabilityBlockImport;
+
+use worker::{
+	Worker, WorkerHandle, Chunks, ParachainBlocks, WorkerMsg, MakeAvailable,
+};
+
+use store::{Store as InnerStore};
+
+/// Abstraction over an executor that lets you spawn tasks in the background.
+pub(crate) type TaskExecutor =
+	Arc<dyn futures01::future::Executor<
+		Box<dyn futures01::Future<Item = (), Error = ()> + Send>
+	> + Send + Sync>;
+
+const LOG_TARGET: &str = "availability";
 
 /// Configuration for the availability store.
 pub struct Config {
@@ -45,67 +73,153 @@ pub struct Config {
 	pub path: PathBuf,
 }
 
+/// Compute gossip topic for the erasure chunk messages given the relay parent,
+/// root and the chunk index.
+///
+/// Since at this point we are not able to use [`network`] directly, but both
+/// of them need to compute these topics, this lives here and not there.
+///
+/// [`network`]: ../polkadot_network/index.html
+pub fn erasure_coding_topic(relay_parent: Hash, erasure_root: Hash, index: u32) -> Hash {
+	let mut v = relay_parent.as_ref().to_vec();
+	v.extend(erasure_root.as_ref());
+	v.extend(&index.to_le_bytes()[..]);
+	v.extend(b"erasure_chunks");
+
+	BlakeTwo256::hash(&v[..])
+}
+
+/// A trait that provides a shim for the [`NetworkService`] trait.
+///
+/// Currently it is not possible to use the networking code in the availability store
+/// core directly due to a number of loop dependencies it require:
+///
+/// `availability-store` -> `network` -> `availability-store`
+///
+/// `availability-store` -> `network` -> `validation` -> `availability-store`
+///
+/// So we provide this shim trait that gets implemented for a wrapper newtype in
+/// the [`network`] module.
+///
+/// [`NetworkService`]: ../polkadot_network/trait.NetworkService.html
+/// [`network`]: ../polkadot_network/index.html
+pub trait ProvideGossipMessages {
+	/// Get a stream of gossip erasure chunk messages for a given topic.
+	///
+	/// Each item is a tuple (relay_parent, candidate_hash, erasure_chunk)
+	fn gossip_messages_for(
+		&self,
+		topic: Hash,
+	) -> Box<dyn Stream<Item = (Hash, Hash, ErasureChunk)> + Send + Unpin>;
+
+	/// Gossip an erasure chunk message.
+	fn gossip_erasure_chunk(
+		&self,
+		relay_parent: Hash,
+		candidate_hash: Hash,
+		erasure_root: Hash,
+		chunk: ErasureChunk,
+	);
+}
+
 /// Some data to keep available about a parachain block candidate.
+#[derive(Debug)]
 pub struct Data {
 	/// The relay chain parent hash this should be localized to.
 	pub relay_parent: Hash,
 	/// The parachain index for this candidate.
 	pub parachain_id: ParaId,
-	/// Unique candidate receipt hash.
-	pub candidate_hash: Hash,
 	/// Block data.
 	pub block_data: BlockData,
 	/// Outgoing message queues from execution of the block, if any.
 	///
 	/// The tuple pairs the message queue root and the queue data.
-	pub outgoing_queues: Option<Vec<(Hash, Vec<Message>)>>,
-}
-
-fn block_data_key(relay_parent: &Hash, candidate_hash: &Hash) -> Vec<u8> {
-	(relay_parent, candidate_hash, 0i8).encode()
+	pub outgoing_queues: Option<AvailableMessages>,
 }
 
 /// Handle to the availability store.
+///
+/// This provides a proxying API that
+///   * in case of write operations provides async methods that send data to
+///     the background worker and resolve when that data is processed by the worker
+///   * in case of read opeartions queries the underlying storage synchronously.
 #[derive(Clone)]
 pub struct Store {
-	inner: Arc<dyn KeyValueDB>,
+	inner: InnerStore,
+	worker: Arc<WorkerHandle>,
+	to_worker: mpsc::UnboundedSender<WorkerMsg>,
 }
 
 impl Store {
-	/// Create a new `Store` with given config on disk.
-	#[cfg(not(target_os = "unknown"))]
-	pub fn new(config: Config) -> io::Result<Self> {
-		use kvdb_rocksdb::{Database, DatabaseConfig};
-		let mut db_config = DatabaseConfig::with_columns(Some(columns::NUM_COLUMNS));
-
-		if let Some(cache_size) = config.cache_size {
-			let mut memory_budget = std::collections::HashMap::new();
-			for i in 0..columns::NUM_COLUMNS {
-				memory_budget.insert(Some(i), cache_size / columns::NUM_COLUMNS as usize);
-			}
-
-			db_config.memory_budget = memory_budget;
-		}
-
-		let path = config.path.to_str().ok_or_else(|| io::Error::new(
-			io::ErrorKind::Other,
-			format!("Bad database path: {:?}", config.path),
-		))?;
-
-		let db = Database::open(&db_config, &path)?;
-
-		Ok(Store {
-			inner: Arc::new(db),
+	/// Create a new `Store` with given condig on disk.
+	///
+	/// Creating a store among other things starts a background worker thread which
+	/// handles most of the write operations to the storage.
+	pub fn new<PGM>(config: Config, gossip: PGM) -> io::Result<Self>
+		where PGM: ProvideGossipMessages + Send + Sync + Clone + 'static
+	{
+		let inner = InnerStore::new(config)?;
+		let worker = Arc::new(Worker::start(inner.clone(), gossip));
+		let to_worker = worker.to_worker().clone();
+
+		Ok(Self {
+			inner,
+			worker,
+			to_worker,
 		})
 	}
 
 	/// Create a new `Store` in-memory. Useful for tests.
-	pub fn new_in_memory() -> Self {
-		Store {
-			inner: Arc::new(::kvdb_memorydb::create(columns::NUM_COLUMNS)),
+	///
+	/// Creating a store among other things starts a background worker thread
+	/// which handles most of the write operations to the storage.
+	pub fn new_in_memory<PGM>(gossip: PGM) -> Self
+		where PGM: ProvideGossipMessages + Send + Sync + Clone + 'static
+	{
+		let inner = InnerStore::new_in_memory();
+		let worker = Arc::new(Worker::start(inner.clone(), gossip));
+		let to_worker = worker.to_worker().clone();
+
+		Self {
+			inner,
+			worker,
+			to_worker,
 		}
 	}
 
+	/// Obtain a [`BlockImport`] implementation to import blocks into this store.
+	///
+	/// This block import will act upon all newly imported blocks sending information
+	/// about parachain heads included in them to this `Store`'s background worker.
+	/// The user may create multiple instances of [`BlockImport`]s with this call.
+	///
+	/// [`BlockImport`]: https://substrate.dev/rustdocs/v1.0/substrate_consensus_common/trait.BlockImport.html
+	pub fn block_import<I, P>(
+		&self,
+		wrapped_block_import: I,
+		client: Arc<P>,
+		thread_pool: TaskExecutor,
+		keystore: KeyStorePtr,
+	) -> ClientResult<(AvailabilityBlockImport<I, P>)>
+	where
+		P: ProvideRuntimeApi + BlockchainEvents<Block> + BlockBody<Block> + Send + Sync + 'static,
+		P::Api: ParachainHost<Block>,
+		P::Api: ApiExt<Block, Error=sp_blockchain::Error>,
+	{
+		let to_worker = self.to_worker.clone();
+
+		let import = AvailabilityBlockImport::new(
+			self.inner.clone(),
+			client,
+			wrapped_block_import,
+			thread_pool,
+			keystore,
+			to_worker,
+		);
+
+		Ok(import)
+	}
+
 	/// Make some data available provisionally.
 	///
 	/// Validators with the responsibility of maintaining availability
@@ -117,174 +231,164 @@ impl Store {
 	/// to be present with the exception of the case where there is no message data
 	/// due to the block's invalidity. Determination of invalidity is beyond the
 	/// scope of this function.
-	pub fn make_available(&self, data: Data) -> io::Result<()> {
-		let mut tx = DBTransaction::new();
-
-		// note the meta key.
-		let mut v = match self.inner.get(columns::META, data.relay_parent.as_ref()) {
-			Ok(Some(raw)) => Vec::decode(&mut &raw[..]).expect("all stored data serialized correctly; qed"),
-			Ok(None) => Vec::new(),
-			Err(e) => {
-				warn!(target: "availability", "Error reading from availability store: {:?}", e);
-				Vec::new()
-			}
-		};
-
-		v.push(data.candidate_hash);
-		tx.put_vec(columns::META, &data.relay_parent[..], v.encode());
-
-		tx.put_vec(
-			columns::DATA,
-			block_data_key(&data.relay_parent, &data.candidate_hash).as_slice(),
-			data.block_data.encode()
-		);
-
-		if let Some(outgoing_queues) = data.outgoing_queues {
-			// This is kept forever and not pruned.
-			for (root, messages) in outgoing_queues {
-				tx.put_vec(
-					columns::DATA,
-					root.as_ref(),
-					messages.encode(),
-				);
-			}
-
+	///
+	/// This method will send the `Data` to the background worker, allowing caller to
+	/// asynchrounously wait for the result.
+	pub async fn make_available(&self, data: Data) -> io::Result<()> {
+		let (s, r) = oneshot::channel();
+		let msg = WorkerMsg::MakeAvailable(MakeAvailable {
+			data,
+			result: s,
+		});
+
+		let _ = self.to_worker.unbounded_send(msg);
+
+		if let Ok(Ok(())) = r.await {
+			Ok(())
+		} else {
+			Err(io::Error::new(io::ErrorKind::Other, format!("adding erasure chunks failed")))
 		}
 
-		self.inner.write(tx)
 	}
 
-	/// Note that a set of candidates have been included in a finalized block with given hash and parent hash.
-	pub fn candidates_finalized(&self, parent: Hash, finalized_candidates: HashSet<Hash>) -> io::Result<()> {
-		let mut tx = DBTransaction::new();
-
-		let v = match self.inner.get(columns::META, &parent[..]) {
-			Ok(Some(raw)) => Vec::decode(&mut &raw[..]).expect("all stored data serialized correctly; qed"),
-			Ok(None) => Vec::new(),
-			Err(e) => {
-				warn!(target: "availability", "Error reading from availability store: {:?}", e);
-				Vec::new()
-			}
-		};
-		tx.delete(columns::META, &parent[..]);
-
-		for candidate_hash in v {
-			if !finalized_candidates.contains(&candidate_hash) {
-				tx.delete(columns::DATA, block_data_key(&parent, &candidate_hash).as_slice());
-			}
-		}
+	/// Get a set of all chunks we are waiting for grouped by
+	/// `(relay_parent, erasure_root, candidate_hash, our_id)`.
+	pub fn awaited_chunks(&self) -> Option<HashSet<(Hash, Hash, Hash, u32)>> {
+		self.inner.awaited_chunks()
+	}
 
-		self.inner.write(tx)
+	/// Qery which candidates were included in the relay chain block by block's parent.
+	pub fn get_candidates_in_relay_block(&self, relay_block: &Hash) -> Option<Vec<Hash>> {
+		self.inner.get_candidates_in_relay_block(relay_block)
 	}
 
-	/// Query block data.
-	pub fn block_data(&self, relay_parent: Hash, candidate_hash: Hash) -> Option<BlockData> {
-		let encoded_key = block_data_key(&relay_parent, &candidate_hash);
-		match self.inner.get(columns::DATA, &encoded_key[..]) {
-			Ok(Some(raw)) => Some(
-				BlockData::decode(&mut &raw[..]).expect("all stored data serialized correctly; qed")
-			),
-			Ok(None) => None,
-			Err(e) => {
-				warn!(target: "availability", "Error reading from availability store: {:?}", e);
-				None
-			}
-		}
+	/// Make a validator's index and a number of validators at a relay parent available.
+	///
+	/// This information is needed before the `add_candidates_in_relay_block` is called
+	/// since that call forms the awaited frontier of chunks.
+	/// In the current implementation this function is called in the `get_or_instantiate` at
+	/// the start of the parachain agreement process on top of some parent hash.
+	pub fn add_validator_index_and_n_validators(
+		&self,
+		relay_parent: &Hash,
+		validator_index: u32,
+		n_validators: u32,
+	) -> io::Result<()> {
+		self.inner.add_validator_index_and_n_validators(
+			relay_parent,
+			validator_index,
+			n_validators,
+		)
 	}
 
-	/// Query message queue data by message queue root hash.
-	pub fn queue_by_root(&self, queue_root: &Hash) -> Option<Vec<Message>> {
-		match self.inner.get(columns::DATA, queue_root.as_ref()) {
-			Ok(Some(raw)) => Some(
-				<_>::decode(&mut &raw[..]).expect("all stored data serialized correctly; qed")
-			),
-			Ok(None) => None,
-			Err(e) => {
-				warn!(target: "availability", "Error reading from availability store: {:?}", e);
-				None
-			}
-		}
+	/// Query a validator's index and n_validators by relay parent.
+	pub fn get_validator_index_and_n_validators(&self, relay_parent: &Hash) -> Option<(u32, u32)> {
+		self.inner.get_validator_index_and_n_validators(relay_parent)
 	}
-}
 
-#[cfg(test)]
-mod tests {
-	use super::*;
+	/// Adds an erasure chunk to storage.
+	///
+	/// The chunk should be checked for validity against the root of encoding
+	/// and its proof prior to calling this.
+	///
+	/// This method will send the chunk to the background worker, allowing caller to
+	/// asynchrounously wait for the result.
+	pub async fn add_erasure_chunk(
+		&self,
+		relay_parent: Hash,
+		receipt: CandidateReceipt,
+		chunk: ErasureChunk,
+	) -> io::Result<()> {
+		self.add_erasure_chunks(relay_parent, receipt, vec![chunk]).await
+	}
 
-	#[test]
-	fn finalization_removes_unneeded() {
-		let relay_parent = [1; 32].into();
+	/// Adds a set of erasure chunks to storage.
+	///
+	/// The chunks should be checked for validity against the root of encoding
+	/// and it's proof prior to calling this.
+	///
+	/// This method will send the chunks to the background worker, allowing caller to
+	/// asynchrounously waiting for the result.
+	pub async fn add_erasure_chunks<I>(
+		&self,
+		relay_parent: Hash,
+		receipt: CandidateReceipt,
+		chunks: I,
+	) -> io::Result<()>
+		where I: IntoIterator<Item = ErasureChunk>
+	{
+		self.add_candidate(relay_parent, receipt.clone()).await?;
+		let (s, r) = oneshot::channel();
+		let chunks = chunks.into_iter().collect();
+		let candidate_hash = receipt.hash();
+		let msg = WorkerMsg::Chunks(Chunks {
+			relay_parent,
+			candidate_hash,
+			chunks,
+			result: s,
+		});
 
-		let para_id_1 = 5.into();
-		let para_id_2 = 6.into();
+		let _ = self.to_worker.unbounded_send(msg);
 
-		let candidate_1 = [2; 32].into();
-		let candidate_2 = [3; 32].into();
+		if let Ok(Ok(())) = r.await {
+			Ok(())
+		} else {
+			Err(io::Error::new(io::ErrorKind::Other, format!("adding erasure chunks failed")))
+		}
+	}
 
-		let block_data_1 = BlockData(vec![1, 2, 3]);
-		let block_data_2 = BlockData(vec![4, 5, 6]);
+	/// Queries an erasure chunk by its block's parent and hash and index.
+	pub fn get_erasure_chunk(
+		&self,
+		relay_parent: &Hash,
+		block_data_hash: Hash,
+		index: usize,
+	) -> Option<ErasureChunk> {
+		self.inner.get_erasure_chunk(relay_parent, block_data_hash, index)
+	}
 
-		let store = Store::new_in_memory();
-		store.make_available(Data {
-			relay_parent,
-			parachain_id: para_id_1,
-			candidate_hash: candidate_1,
-			block_data: block_data_1.clone(),
-			outgoing_queues: None,
-		}).unwrap();
+	/// Stores a candidate receipt.
+	pub async fn add_candidate(
+		&self,
+		relay_parent: Hash,
+		receipt: CandidateReceipt,
+	) -> io::Result<()> {
+		let (s, r) = oneshot::channel();
 
-		store.make_available(Data {
+		let msg = WorkerMsg::ParachainBlocks(ParachainBlocks {
 			relay_parent,