ICMP message-routing gossip (#304)

* core logic for ICMP gossip

* refactor gossip to make more extension friendly

* move files aroun

* extract attestation-gossip logic to its own module

* message validation and broadcast logic

* fix upstream crates' compilation

* add a test

* another test for overlapping

* Some grammar and phrasing tweaks

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

* add since parameter to ingress runtime API

* broadcast out known unrouted message queues

* fix compilation of service and collator

* remove useless index_mapping

* some tests for icmp propagation

* fix decoding bug and test icmp queue validation

* simplify engine-id definition

Co-Authored-By: Sergei Pepyakin <sergei@parity.io>

* address some grumbles

* some cleanup of old circulation code

* give network a handle to extrinsic store on startup

* an honest collator ensures data available as well

* address some grumbles

* add docs; rename the attestation session to "leaf work"

* module docs

* move gossip back to gossip.rs

* clean up and document attestation-gossip a bit

* some more docs on the availability store

* store all outgoing message queues in the availability store

* filter `Extrinsic` out of validation crate

* expunge Extrinsic from network

* expunge Extrinsic from erasure-coding

* expunge Extrinsic from collator

* expunge from adder-collator

* rename ExtrinsicStore to AvailabilityStore everywhere

* annotate and clean up message-routing tests

availability-store/src/lib.rs

@@ -14,13 +14,17 @@
 // 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.
+//! Persistent database for parachain data: PoV block data 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 kvdb_rocksdb::{Database, DatabaseConfig};
 use polkadot_primitives::Hash;
-use polkadot_primitives::parachain::{Id as ParaId, BlockData, Extrinsic};
+use polkadot_primitives::parachain::{Id as ParaId, BlockData, Message};
 use log::warn;
 
 use std::collections::HashSet;
@@ -42,7 +46,7 @@ pub struct Config {
 	pub path: PathBuf,
 }
 
-/// Some data to keep available.
+/// Some data to keep available about a parachain block candidate.
 pub struct Data {
 	/// The relay chain parent hash this should be localized to.
 	pub relay_parent: Hash,
@@ -52,18 +56,16 @@ pub struct Data {
 	pub candidate_hash: Hash,
 	/// Block data.
 	pub block_data: BlockData,
-	/// Extrinsic data.
-	pub extrinsic: Option<Extrinsic>,
+	/// 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()
 }
 
-fn extrinsic_key(relay_parent: &Hash, candidate_hash: &Hash) -> Vec<u8> {
-	(relay_parent, candidate_hash, 1i8).encode()
-}
-
 /// Handle to the availability store.
 #[derive(Clone)]
 pub struct Store {
@@ -96,6 +98,16 @@ impl Store {
 	}
 
 	/// Make some data available provisionally.
+	///
+	/// Validators with the responsibility of maintaining availability
+	/// for a block or collators collating a block will call this function
+	/// in order to persist that data to disk and so it can be queried and provided
+	/// to other nodes in the network.
+	///
+	/// The message data of `Data` is optional but is expected
+	/// 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();
 
@@ -118,12 +130,16 @@ impl Store {
 			data.block_data.encode()
 		);
 
-		if let Some(extrinsic) = data.extrinsic {
-			tx.put_vec(
-				columns::DATA,
-				extrinsic_key(&data.relay_parent, &data.candidate_hash).as_slice(),
-				extrinsic.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(),
+				);
+			}
+
 		}
 
 		self.inner.write(tx)
@@ -146,7 +162,6 @@ impl Store {
 		for candidate_hash in v {
 			if !finalized_candidates.contains(&candidate_hash) {
 				tx.delete(columns::DATA, block_data_key(&parent, &candidate_hash).as_slice());
-				tx.delete(columns::DATA, extrinsic_key(&parent, &candidate_hash).as_slice());
 			}
 		}
 
@@ -168,12 +183,11 @@ impl Store {
 		}
 	}
 
-	/// Query extrinsic data.
-	pub fn extrinsic(&self, relay_parent: Hash, candidate_hash: Hash) -> Option<Extrinsic> {
-		let encoded_key = extrinsic_key(&relay_parent, &candidate_hash);
-		match self.inner.get(columns::DATA, &encoded_key[..]) {
+	/// 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(
-				Extrinsic::decode(&mut &raw[..]).expect("all stored data serialized correctly; qed")
+				<_>::decode(&mut &raw[..]).expect("all stored data serialized correctly; qed")
 			),
 			Ok(None) => None,
 			Err(e) => {
@@ -207,7 +221,7 @@ mod tests {
 			parachain_id: para_id_1,
 			candidate_hash: candidate_1,
 			block_data: block_data_1.clone(),
-			extrinsic: Some(Extrinsic { outgoing_messages: Vec::new() }),
+			outgoing_queues: None,
 		}).unwrap();
 
 		store.make_available(Data {
@@ -215,21 +229,53 @@ mod tests {
 			parachain_id: para_id_2,
 			candidate_hash: candidate_2,
 			block_data: block_data_2.clone(),
-			extrinsic: Some(Extrinsic { outgoing_messages: Vec::new() }),
+			outgoing_queues: None,
 		}).unwrap();
 
 		assert_eq!(store.block_data(relay_parent, candidate_1).unwrap(), block_data_1);
 		assert_eq!(store.block_data(relay_parent, candidate_2).unwrap(), block_data_2);
 
-		assert!(store.extrinsic(relay_parent, candidate_1).is_some());
-		assert!(store.extrinsic(relay_parent, candidate_2).is_some());
-
 		store.candidates_finalized(relay_parent, [candidate_1].iter().cloned().collect()).unwrap();
 
 		assert_eq!(store.block_data(relay_parent, candidate_1).unwrap(), block_data_1);
 		assert!(store.block_data(relay_parent, candidate_2).is_none());
+	}
+
+	#[test]
+	fn queues_available_by_queue_root() {
+		let relay_parent = [1; 32].into();
+		let para_id = 5.into();
+		let candidate = [2; 32].into();
+		let block_data = BlockData(vec![1, 2, 3]);
+
+		let message_queue_root_1 = [0x42; 32].into();
+		let message_queue_root_2 = [0x43; 32].into();
 
-		assert!(store.extrinsic(relay_parent, candidate_1).is_some());
-		assert!(store.extrinsic(relay_parent, candidate_2).is_none());
+		let message_a = Message(vec![1, 2, 3, 4]);
+		let message_b = Message(vec![4, 5, 6, 7]);
+
+		let outgoing_queues = vec![
+			(message_queue_root_1, vec![message_a.clone()]),
+			(message_queue_root_2, vec![message_b.clone()]),
+		];
+
+		let store = Store::new_in_memory();
+		store.make_available(Data {
+			relay_parent,
+			parachain_id: para_id,
+			candidate_hash: candidate,
+			block_data: block_data.clone(),
+			outgoing_queues: Some(outgoing_queues),
+		}).unwrap();
+
+		assert_eq!(
+			store.queue_by_root(&message_queue_root_1),
+			Some(vec![message_a]),
+		);
+
+		assert_eq!(
+			store.queue_by_root(&message_queue_root_2),
+			Some(vec![message_b]),
+		);
 	}
 }

collator/src/lib.rs

@@ -57,7 +57,7 @@ use primitives::Pair;
 use polkadot_primitives::{
 	BlockId, Hash, Block,
 	parachain::{
-		self, BlockData, DutyRoster, HeadData, ConsolidatedIngress, Message, Id as ParaId, Extrinsic,
+		self, BlockData, DutyRoster, HeadData, ConsolidatedIngress, Message, Id as ParaId, OutgoingMessages,
 		PoVBlock, Status as ParachainStatus, ValidatorId, CollatorPair,
 	}
 };
@@ -65,8 +65,8 @@ use polkadot_cli::{
 	Worker, IntoExit, ProvideRuntimeApi, TaskExecutor, AbstractService,
 	CustomConfiguration, ParachainHost,
 };
-use polkadot_network::validation::{SessionParams, ValidationNetwork};
-use polkadot_network::{NetworkService, PolkadotProtocol};
+use polkadot_network::validation::{LeafWorkParams, ValidationNetwork};
+use polkadot_network::{PolkadotNetworkService, PolkadotProtocol};
 use tokio::timer::Timeout;
 
 pub use polkadot_cli::VersionInfo;
@@ -91,7 +91,7 @@ pub trait Network: Send + Sync {
 	fn checked_statements(&self, relay_parent: Hash) -> Box<dyn Stream<Item=SignedStatement, Error=()>>;
 }
 
-impl<P, E> Network for ValidationNetwork<P, E, NetworkService, TaskExecutor> where
+impl<P, E> Network for ValidationNetwork<P, E, PolkadotNetworkService, TaskExecutor> where
 	P: 'static + Send + Sync,
 	E: 'static + Send + Sync,
 {
@@ -142,7 +142,7 @@ pub trait BuildParachainContext {
 /// This can be implemented through an externally attached service or a stub.
 /// This is expected to be a lightweight, shared type like an Arc.
 pub trait ParachainContext: Clone {
-	type ProduceCandidate: IntoFuture<Item=(BlockData, HeadData, Extrinsic), Error=InvalidHead>;
+	type ProduceCandidate: IntoFuture<Item=(BlockData, HeadData, OutgoingMessages), Error=InvalidHead>;
 
 	/// Produce a candidate, given the relay parent hash, the latest ingress queue information
 	/// and the last parachain head.
@@ -177,7 +177,7 @@ pub fn collate<'a, R, P>(
 	para_context: P,
 	key: Arc<CollatorPair>,
 )
-	-> impl Future<Item=parachain::Collation, Error=Error<R::Error>> + 'a
+	-> impl Future<Item=(parachain::Collation, OutgoingMessages), Error=Error<R::Error>> + 'a
 	where
 		R: RelayChainContext,
 		R::Error: 'a,
@@ -197,11 +197,11 @@ pub fn collate<'a, R, P>(
 				.map(move |x| (ingress, x))
 				.map_err(Error::Collator)
 		})
-		.and_then(move |(ingress, (block_data, head_data, mut extrinsic))| {
+		.and_then(move |(ingress, (block_data, head_data, mut outgoing))| {
 			let block_data_hash = block_data.hash();
 			let signature = key.sign(block_data_hash.as_ref()).into();
 			let egress_queue_roots =
-				polkadot_validation::egress_roots(&mut extrinsic.outgoing_messages);
+				polkadot_validation::egress_roots(&mut outgoing.outgoing_messages);
 
 			let receipt = parachain::CandidateReceipt {
 				parachain_index: local_id,
@@ -214,19 +214,21 @@ pub fn collate<'a, R, P>(
 				upward_messages: Vec::new(),
 			};
 
-			Ok(parachain::Collation {
+			let collation = parachain::Collation {
 				receipt,
 				pov: PoVBlock {
 					block_data,
 					ingress,
 				},
-			})
+			};
+
+			Ok((collation, outgoing))
 		})
 }
 
 /// Polkadot-api context.
 struct ApiContext<P, E> {
-	network: Arc<ValidationNetwork<P, E, NetworkService, TaskExecutor>>,
+	network: Arc<ValidationNetwork<P, E, PolkadotNetworkService, TaskExecutor>>,
 	parent_hash: Hash,
 	validators: Vec<ValidatorId>,
 }
@@ -243,7 +245,7 @@ impl<P: 'static, E: 'static> RelayChainContext for ApiContext<P, E> where
 		// TODO: https://github.com/paritytech/polkadot/issues/253
 		//
 		// Fetch ingress and accumulate all unrounted egress
-		let _session = self.network.instantiate_session(SessionParams {
+		let _session = self.network.instantiate_leaf_work(LeafWorkParams {
 			local_session_key: None,
 			parent_hash: self.parent_hash,
 			authorities: self.validators.clone(),
@@ -303,26 +305,28 @@ impl<P, E> Worker for CollationNode<P, E> where
 			return Box::new(future::err(()));
 		};
 
+		let is_known = move |block_hash: &Hash| {
+			use client::BlockStatus;
+			use polkadot_network::gossip::Known;
+
+			match known_oracle.block_status(&BlockId::hash(*block_hash)) {
+				Err(_) | Ok(BlockStatus::Unknown) | Ok(BlockStatus::Queued) => None,
+				Ok(BlockStatus::KnownBad) => Some(Known::Bad),
+				Ok(BlockStatus::InChainWithState) | Ok(BlockStatus::InChainPruned) =>
+					match select_chain.leaves() {
+						Err(_) => None,
+						Ok(leaves) => if leaves.contains(block_hash) {
+							Some(Known::Leaf)
+						} else {
+							Some(Known::Old)
+						},
+					}
+			}
+		};
+
 		let message_validator = polkadot_network::gossip::register_validator(
 			network.clone(),
-			move |block_hash: &Hash| {
-				use client::BlockStatus;
-				use polkadot_network::gossip::Known;
-
-				match known_oracle.block_status(&BlockId::hash(*block_hash)) {
-					Err(_) | Ok(BlockStatus::Unknown) | Ok(BlockStatus::Queued) => None,
-					Ok(BlockStatus::KnownBad) => Some(Known::Bad),
-					Ok(BlockStatus::InChainWithState) | Ok(BlockStatus::InChainPruned) =>
-						match select_chain.leaves() {
-							Err(_) => None,
-							Ok(leaves) => if leaves.contains(block_hash) {
-								Some(Known::Leaf)
-							} else {
-								Some(Known::Old)
-							},
-						}
-				}
-			},
+			(is_known, client.clone()),
 		);
 
 		let validation_network = Arc::new(ValidationNetwork::new(
@@ -386,13 +390,20 @@ impl<P, E> Worker for CollationNode<P, E> where
 						context,
 						parachain_context,
 						key,
-					).map(move |collation| {
-						network.with_spec(move |spec, ctx| spec.add_local_collation(
-							ctx,
-							relay_parent,
-							targets,
-							collation,
-						));
+					).map(move |(collation, outgoing)| {
+						network.with_spec(move |spec, ctx| {
+							let res = spec.add_local_collation(
+								ctx,
+								relay_parent,
+								targets,
+								collation,
+								outgoing,
+							);
+
+							if let Err(e) = res {
+								warn!("Unable to broadcast local collation: {:?}", e);
+							}
+						})
 					});
 
 					future::Either::B(collation_work)
@@ -450,7 +461,7 @@ pub fn run_collator<P, E>(
 #[cfg(test)]
 mod tests {
 	use std::collections::HashMap;
-	use polkadot_primitives::parachain::{OutgoingMessage, FeeSchedule};
+	use polkadot_primitives::parachain::{TargetedMessage, FeeSchedule};
 	use keyring::Sr25519Keyring;
 	use super::*;
 
@@ -475,20 +486,20 @@ mod tests {
 	struct DummyParachainContext;
 
 	impl ParachainContext for DummyParachainContext {
-		type ProduceCandidate = Result<(BlockData, HeadData, Extrinsic), InvalidHead>;
+		type ProduceCandidate = Result<(BlockData, HeadData, OutgoingMessages), InvalidHead>;
 
 		fn produce_candidate<I: IntoIterator<Item=(ParaId, Message)>>(
 			&self,
 			_relay_parent: Hash,
 			_status: ParachainStatus,
 			ingress: I,
-		) -> Result<(BlockData, HeadData, Extrinsic), InvalidHead> {
+		) -> Result<(BlockData, HeadData, OutgoingMessages), InvalidHead> {
 			// send messages right back.
 			Ok((
 				BlockData(vec![1, 2, 3, 4, 5,]),
 				HeadData(vec![9, 9, 9]),
-				Extrinsic {
-					outgoing_messages: ingress.into_iter().map(|(id, msg)| OutgoingMessage {
+				OutgoingMessages {
+					outgoing_messages: ingress.into_iter().map(|(id, msg)| TargetedMessage {
 						target: id,
 						data: msg.0,
 					}).collect(),
@@ -542,7 +553,7 @@ mod tests {
 			context.clone(),
 			DummyParachainContext,
 			Arc::new(Sr25519Keyring::Alice.pair().into()),
-		).wait().unwrap();
+		).wait().unwrap().0;
 
 		// ascending order by root.
 		assert_eq!(collation.receipt.egress_queue_roots, vec![(a, root_a), (b, root_b)]);

erasure-coding/src/lib.rs

@@ -27,7 +27,7 @@
 use codec::{Encode, Decode};
 use reed_solomon::galois_16::{self, ReedSolomon};
 use primitives::{Hash as H256, BlakeTwo256, HashT};
-use primitives::parachain::{BlockData, Extrinsic};
+use primitives::parachain::{BlockData, OutgoingMessages};
 use substrate_primitives::Blake2Hasher;
 use trie::{EMPTY_PREFIX, MemoryDB, Trie, TrieMut, trie_types::{TrieDBMut, TrieDB}};
 
@@ -124,11 +124,11 @@ fn code_params(n_validators: usize) -> Result<CodeParams, Error> {
 /// Obtain erasure-coded chunks, one for each validator.
 ///
 /// Works only up to 65536 validators, and `n_validators` must be non-zero.
-pub fn obtain_chunks(n_validators: usize, block_data: &BlockData, extrinsic: &Extrinsic)
+pub fn obtain_chunks(n_validators: usize, block_data: &BlockData, outgoing: &OutgoingMessages)
 	-> Result<Vec<Vec<u8>>, Error>
 {
 	let params  = code_params(n_validators)?;
-	let encoded = (block_data, extrinsic).encode();
+	let encoded = (block_data, outgoing).encode();
 
 	if encoded.is_empty() {
 		return Err(Error::BadPayload);
@@ -150,7 +150,7 @@ pub fn obtain_chunks(n_validators: usize, block_data: &BlockData, extrinsic: &Ex
 ///
 /// Works only up to 65536 validators, and `n_validators` must be non-zero.
 pub fn reconstruct<'a, I: 'a>(n_validators: usize, chunks: I)
-	-> Result<(BlockData, Extrinsic), Error>
+	-> Result<(BlockData, OutgoingMessages), Error>
 	where I: IntoIterator<Item=(&'a [u8], usize)>
 {
 	let params = code_params(n_validators)?;
@@ -399,7 +399,7 @@ mod tests {
     #[test]
 	fn round_trip_block_data() {
 		let block_data = BlockData((0..255).collect());
-		let ex = Extrinsic { outgoing_messages: Vec::new() };
+		let ex = OutgoingMessages { outgoing_messages: Vec::new() };
 		let chunks = obtain_chunks(
 			10,
 			&block_data,
@@ -428,7 +428,7 @@ mod tests {
 		let chunks = obtain_chunks(
 			10,
 			&block_data,
-			&Extrinsic { outgoing_messages: Vec::new() },
+			&OutgoingMessages { outgoing_messages: Vec::new() },
 		).unwrap();
 		let chunks: Vec<_> = chunks.iter().map(|c| &c[..]).collect();
 

network/Cargo.toml

@@ -18,7 +18,7 @@ sr-primitives = { git = "https://github.com/paritytech/substrate", branch = "pol
 futures = "0.1"
 log = "0.4"
 exit-future = "0.1.4"
+substrate-client = { git = "https://github.com/paritytech/substrate", branch = "polkadot-master" }
 
 [dev-dependencies]
-substrate-client = { git = "https://github.com/paritytech/substrate", branch = "polkadot-master" }
 substrate-keyring = { git = "https://github.com/paritytech/substrate", branch = "polkadot-master" }

network/src/gossip/attestation.rs

+// Copyright 2019 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/>.
+
+//! Gossip messages and structures for dealing with attestations (statements of
+//! validity of invalidity on parachain candidates).
+//!
+//! This follows the same principles as other gossip modules (see parent
+//! documentation for more details) by being aware of our current chain
+//! heads and accepting only information relative to them. Attestations are localized to
+//! relay chain head, so this is easily doable.
+//!
+//! This module also provides a filter, so we can only broadcast messages to
+//! peers that are relevant to chain heads they have advertised.
+//!
+//! Furthermore, since attestations are bottlenecked by the `Candidate` statement,
+//! we only accept attestations which are themselves `Candidate` messages, or reference
+//! a `Candidate` we are aware of. Otherwise, it is possible we could be forced to
+//! consider an infinite amount of attestations produced by a misbehaving validator.
+
+use substrate_network::consensus_gossip::{ValidationResult as GossipValidationResult};
+use polkadot_validation::GenericStatement;
+use polkadot_primitives::Hash;
+
+use std::collections::{HashMap, HashSet};
+
+use log::warn;
+use crate::router::attestation_topic;
+
+use super::{cost, benefit, MAX_CHAIN_HEADS, LeavesVec, ChainContext, Known, MessageValidationData, GossipStatement};
+
+// knowledge about attestations on a single parent-hash.
+#[derive(Default)]
+pub(super) struct Knowledge {
+	candidates: HashSet<Hash>,
+}
+
+impl Knowledge {
+	// whether the peer is aware of a candidate with given hash.
+	fn is_aware_of(&self, candidate_hash: &Hash) -> bool {
+		self.candidates.contains(candidate_hash)
+	}
+
+	// note that the peer is aware of a candidate with given hash. this should
+	// be done after observing an incoming candidate message via gossip.
+	fn note_aware(&mut self, candidate_hash: Hash) {
+		self.candidates.insert(candidate_hash);
+	}
+}
+
+#[derive(Default)]
+pub(super) struct PeerData {
+	live: HashMap<Hash, Knowledge>,
+}
+
+impl PeerData {
+	/// Update leaves, returning a list of which leaves are new.
+	pub(super) fn update_leaves(&mut self, leaves: &LeavesVec) -> LeavesVec {
+		let mut new = LeavesVec::new();
+		self.live.retain(|k, _| leaves.contains(k));
+		for &leaf in leaves {
+			self.live.entry(leaf).or_insert_with(|| {
+				new.push(leaf);
+				Default::default()
+			});
+		}
+
+		new
+	}
+
+	#[cfg(test)]
+	pub(super) fn note_aware_under_leaf(&mut self, relay_chain_leaf: &Hash, candidate_hash: Hash) {
+		if let Some(knowledge) = self.live.get_mut(relay_chain_leaf) {
+			knowledge.note_aware(candidate_hash);
+		}
+	}
+
+	pub(super) fn knowledge_at_mut(&mut self, parent_hash: &Hash) -> Option<&mut Knowledge> {
+		self.live.get_mut(parent_hash)
+	}
+
+	/// Get an iterator over all live leaves of this peer.
+	pub(super) fn leaves(&self) -> impl Iterator<Item = &Hash> {
+		self.live.keys()
+	}
+}
+
+/// An impartial view of what topics and data are valid based on attestation session data.
+pub(super) struct View {
+	leaf_work: Vec<(Hash, LeafView)>, // hashes of the best DAG-leaves paired with validation data.
+	topics: HashMap<Hash, Hash>, // maps topic hashes to block hashes.
+}
+
+impl Default for View {
+	fn default() -> Self {
+		View {
+			leaf_work: Vec::with_capacity(MAX_CHAIN_HEADS),
+			topics: Default::default(),
+		}
+	}
+}
+
+impl View {
+	fn leaf_view(&self, relay_chain_leaf: &Hash) -> Option<&LeafView> {
+		self.leaf_work.iter()
+			.find_map(|&(ref h, ref leaf)| if h == relay_chain_leaf { Some(leaf) } else { None } )
+	}
+
+	fn leaf_view_mut(&mut self, relay_chain_leaf: &Hash) -> Option<&mut LeafView> {
+		self.leaf_work.iter_mut()
+			.find_map(|&mut (ref h, ref mut leaf)| if h == relay_chain_leaf { Some(leaf) } else { None } )
+	}
+
+	/// Get our leaves-set. Guaranteed to have length <= MAX_CHAIN_HEADS.
+	pub(super) fn neighbor_info<'a>(&'a self) -> impl Iterator<Item=Hash> + 'a + Clone {
+		self.leaf_work.iter().take(MAX_CHAIN_HEADS).map(|(p, _)| p.clone())
+	}
+
+	/// Note new leaf in our local view and validation data necessary to check signatures
+	/// of statements issued under this leaf.
+	///
+	/// This will be pruned later on a call to `prune_old_leaves`, when this leaf
+	/// is not a leaf anymore.
+	pub(super) fn new_local_leaf(&mut self, relay_chain_leaf: Hash, validation_data: MessageValidationData) {
+		self.leaf_work.push((
+			relay_chain_leaf,
+			LeafView {
+				validation_data,
+				knowledge: Default::default(),
+			},
+		));
+		self.topics.insert(attestation_topic(relay_chain_leaf), relay_chain_leaf);
+	}
+
+	/// Prune old leaf-work that fails the leaf predicate.
+	pub(super) fn prune_old_leaves<F: Fn(&Hash) -> bool>(&mut self, is_leaf: F) {
+		let leaf_work = &mut self.leaf_work;
+		leaf_work.retain(|&(ref relay_chain_leaf, _)| is_leaf(relay_chain_leaf));
+		self.topics.retain(|_, v| leaf_work.iter().find(|(p, _)| p == v).is_some());
+	}
+