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
2026-07-22 12:35:42 +00:00 · 2019-08-29 11:49:59 +02:00
parent bd8ebbfee5
commit 55c4c830fe
22 changed files with 1981 additions and 818 deletions
@@ -0,0 +1,264 @@
+// 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());
+	}
+
+	/// Whether a message topic is considered live relative to our view. non-live
+	/// topics do not pertain to our perceived leaves, and are uninteresting to us.
+	pub(super) fn is_topic_live(&self, topic: &Hash) -> bool {
+		self.topics.contains_key(topic)
+	}
+
+	/// The relay-chain block hash corresponding to a topic.
+	pub(super) fn topic_block(&self, topic: &Hash) -> Option<&Hash> {
+		self.topics.get(topic)
+	}
+
+
+	/// Validate the signature on an attestation statement of some kind. Should be done before
+	/// any repropagation of that statement.
+	pub(super) fn validate_statement_signature<C: ChainContext + ?Sized>(
+		&mut self,
+		message: GossipStatement,
+		chain: &C,
+	)
+		-> (GossipValidationResult<Hash>, i32)
+	{
+		// message must reference one of our chain heads and
+		// if message is not a `Candidate` we should have the candidate available
+		// in `attestation_view`.
+		match self.leaf_view(&message.relay_chain_leaf) {
+			None => {
+				let cost = match chain.is_known(&message.relay_chain_leaf) {
+					Some(Known::Leaf) => {
+						warn!(
+							target: "network",
+							"Leaf block {} not considered live for attestation",
+							message.relay_chain_leaf,
+						);
+
+						0
+					}
+					Some(Known::Old) => cost::PAST_MESSAGE,
+					_ => cost::FUTURE_MESSAGE,
+				};
+
+				(GossipValidationResult::Discard, cost)
+			}
+			Some(view) => {
+				// first check that we are capable of receiving this message
+				// in a DoS-proof manner.
+				let benefit = match message.signed_statement.statement {
+					GenericStatement::Candidate(_) => benefit::NEW_CANDIDATE,
+					GenericStatement::Valid(ref h) | GenericStatement::Invalid(ref h) => {
+						if !view.knowledge.is_aware_of(h) {
+							let cost = cost::ATTESTATION_NO_CANDIDATE;
+							return (GossipValidationResult::Discard, cost);
+						}
+
+						benefit::NEW_ATTESTATION
+					}
+				};
+
+				// validate signature.
+				let res = view.validation_data.check_statement(
+					&message.relay_chain_leaf,
+					&message.signed_statement,
+				);
+
+				match res {
+					Ok(()) => {
+						let topic = attestation_topic(message.relay_chain_leaf);
+						(GossipValidationResult::ProcessAndKeep(topic), benefit)
+					}
+					Err(()) => (GossipValidationResult::Discard, cost::BAD_SIGNATURE),
+				}
+			}
+		}
+	}
+
+	/// whether it's allowed to send a statement to a peer with given knowledge
+	/// about the relay parent the statement refers to.
+	pub(super) fn statement_allowed(
+		&mut self,
+		statement: &GossipStatement,
+		relay_chain_leaf: &Hash,
+		peer_knowledge: &mut Knowledge,
+	) -> bool {
+		let signed = &statement.signed_statement;
+
+		match signed.statement {
+			GenericStatement::Valid(ref h) | GenericStatement::Invalid(ref h) => {
+				// `valid` and `invalid` statements can only be propagated after
+				// a candidate message is known by that peer.
+				peer_knowledge.is_aware_of(h)
+			}
+			GenericStatement::Candidate(ref c) => {
+				// if we are sending a `Candidate` message we should make sure that
+				// attestation_view and their_view reflects that we know about the candidate.
+				let hash = c.hash();
+				peer_knowledge.note_aware(hash);
+				if let Some(attestation_view) = self.leaf_view_mut(&relay_chain_leaf) {
+					attestation_view.knowledge.note_aware(hash);
+				}
+
+				// at this point, the peer hasn't seen the message or the candidate
+				// and has knowledge of the relevant relay-chain parent.
+				true
+			}
+		}
+	}
+}
+
+struct LeafView {
+	validation_data: MessageValidationData,
+	knowledge: Knowledge,
+}
@@ -0,0 +1,339 @@
+// 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/>.
+
+//! Data structures and synchronous logic for ICMP message gossip.
+//!
+//! The parent-module documentation describes some rationale of the general
+//! gossip protocol design.
+//!
+//! The ICMP message-routing gossip works according to those rationale.
+//!
+//! In this protocol, we perform work under 4 conditions:
+//! ### 1. Upon observation of a new leaf in the block-DAG.
+//!
+//! We first communicate the best leaves to our neighbors in the gossip graph
+//! by the means of a neighbor packet. Then, we query to discover the trie roots
+//! of all un-routed message queues from the perspective of each of those leaves.
+//!
+//! For any trie root in the unrouted set for the new leaf, if we have the corresponding
+//! queue, we send it to any peers with the new leaf in their latest advertised set.
+//!
+//! Which parachain those messages go to and from is unimportant, because this is
+//! an everybody-sees-everything style protocol. The only important property is "liveness":
+//! that the queue root is un-routed at one of the leaves we perceive to be at the head
+//! of the block-DAG.
+//!
+//! In Substrate gossip, every message is associated with a topic. Typically,
+//! many messages are grouped under a single topic. In this gossip system, each queue
+//! gets its own topic, which is based on the root hash of the queue. This is because
+//! many different chain leaves may have the same queue as un-routed, so it's better than
+//! attempting to group message packets by the leaf they appear unrouted at.
+//!
+//! ### 2. Upon a neighbor packet from a peer.
+//!
+//! The neighbor packet from a peer should contain perceived chain heads of that peer.
+//! If there is any overlap between our perceived chain heads and theirs, we send
+//! them any known, un-routed message queue from either set.
+//!
+//! ### 3. Upon receiving a message queue from a peer.
+//!
+//! If the message queue is in the un-routed set of one of the latest leaves we've updated to,
+//! we accept it and relay to any peers who need that queue as well.
+//!
+//! If not, we report the peer to the peer-set manager for sending us bad data.
+//!
+//! ### 4. Periodic Pruning
+//!
+//! We prune messages that are not un-routed from the view of any leaf and cease
+//! to attempt to send them to any peer.
+
+use sr_primitives::traits::{BlakeTwo256, Hash as HashT};
+use polkadot_primitives::Hash;
+use std::collections::{HashMap, HashSet};
+use substrate_client::error::Error as ClientError;
+use super::{MAX_CHAIN_HEADS, GossipValidationResult, LeavesVec, ChainContext};
+
+/// Construct a topic for a message queue root deterministically.
+pub fn queue_topic(queue_root: Hash) -> Hash {
+	let mut v = queue_root.as_ref().to_vec();
+	v.extend(b"message_queue");
+
+	BlakeTwo256::hash(&v[..])
+}
+
+/// A view of which queue roots are current for a given set of leaves.
+#[derive(Default)]
+pub struct View {
+	leaves: LeavesVec,
+	leaf_topics: HashMap<Hash, HashSet<Hash>>, // leaf_hash -> { topics }
+	expected_queues: HashMap<Hash, (Hash, bool)>, // topic -> (queue-root, known)
+}
+
+impl View {
+	/// Update the set of current leaves. This is called when we perceive a new bset leaf-set.
+	pub fn update_leaves<T: ChainContext + ?Sized, I>(&mut self, context: &T, new_leaves: I)
+		-> Result<(), ClientError>
+		where I: Iterator<Item=Hash>
+	{
+		let new_leaves = new_leaves.take(MAX_CHAIN_HEADS);
+		let old_leaves = std::mem::replace(&mut self.leaves, new_leaves.collect());
+
+		let expected_queues = &mut self.expected_queues;
+		let leaves = &self.leaves;
+		self.leaf_topics.retain(|l, topics| {
+			if leaves.contains(l) { return true }
+
+			// prune out all data about old leaves we don't follow anymore.
+			for topic in topics.iter() {
+				expected_queues.remove(topic);
+			}
+			false
+		});
+
+		let mut res = Ok(());
+
+		// add in new data about fresh leaves.
+		for new_leaf in &self.leaves {
+			if old_leaves.contains(new_leaf) { continue }
+
+			let mut this_leaf_topics = HashSet::new();
+
+			let r = context.leaf_unrouted_roots(new_leaf, &mut |&queue_root| {
+				let topic = queue_topic(queue_root);
+				this_leaf_topics.insert(topic);
+				expected_queues.entry(topic).or_insert((queue_root, false));
+			});
+
+			if r.is_err() {
+				if let Err(e) = res {
+					log::debug!(target: "message_routing", "Ignored duplicate error {}", e)
+				};
+				res = r;
+			}
+
+			self.leaf_topics.insert(*new_leaf, this_leaf_topics);
+		}
+
+		res
+	}
+
+	/// Validate an incoming message queue against this view. If it is accepted
+	/// by our view of un-routed message queues, we will keep and re-propagate.
+	pub fn validate_queue_and_note_known(&mut self, messages: &super::GossipParachainMessages)
+		-> (GossipValidationResult<Hash>, i32)
+	{
+		let ostensible_topic = queue_topic(messages.queue_root);
+		match self.expected_queues.get_mut(&ostensible_topic) {
+			None => (GossipValidationResult::Discard, super::cost::UNNEEDED_ICMP_MESSAGES),
+			Some(&mut (_, ref mut known)) => {
+				if !messages.queue_root_is_correct() {
+					(
+						GossipValidationResult::Discard,
+						super::cost::icmp_messages_root_mismatch(messages.messages.len()),
+					)
+				} else {
+					*known = true;
+					(
+						GossipValidationResult::ProcessAndKeep(ostensible_topic),
+						super::benefit::NEW_ICMP_MESSAGES,
+					)
+				}
+			}
+		}
+	}
+
+	/// Whether a message with given topic is live.
+	pub fn is_topic_live(&self, topic: &Hash) -> bool {
+		self.expected_queues.get(topic).is_some()
+	}
+
+	/// Whether a message is allowed under the intersection of the given leaf-set
+	/// and our own.
+	pub fn allowed_intersecting(&self, other_leaves: &LeavesVec, topic: &Hash) -> bool {
+		for i in other_leaves {
+			for j in &self.leaves {
+				if i == j {
+					let leaf_topics = self.leaf_topics.get(i)
+						.expect("leaf_topics are mutated only in update_leaves; \
+							we have an entry for each item in self.leaves; \
+							i is in self.leaves; qed");
+
+					if leaf_topics.contains(topic) {
+						return true;
+					}
+				}
+			}
+		}
+
+		false
+	}
+
+	/// Get topics of all message queues a peer is interested in - this is useful
+	/// when a peer has informed us of their new best leaves.
+	pub fn intersection_topics(&self, other_leaves: &LeavesVec) -> impl Iterator<Item=Hash> {
+		let deduplicated = other_leaves.iter()
+			.filter_map(|l| self.leaf_topics.get(l))
+			.flat_map(|topics| topics.iter().cloned())
+			.collect::<HashSet<_>>();
+
+		deduplicated.into_iter()
+	}
+
+	/// Iterate over all live message queues for which the data is marked as not locally known,
+	/// calling a closure with `(topic, root)`. The closure will return whether the queue data is
+	/// unknown.
+	///
+	/// This is called when we should send un-routed message queues that we are
+	/// newly aware of to peers - as in when we update our leaves.
+	pub fn sweep_unknown_queues(&mut self, mut check_known: impl FnMut(&Hash, &Hash) -> bool) {
+		for (topic, &mut (ref queue_root, ref mut known)) in self.expected_queues.iter_mut() {
+			if !*known {
+				*known = check_known(topic, queue_root)
+			}
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::tests::TestChainContext;
+	use crate::gossip::{Known, GossipParachainMessages};
+	use polkadot_primitives::parachain::Message as ParachainMessage;
+
+	fn hash(x: u8) -> Hash {
+		[x; 32].into()
+	}
+
+	fn message_queue(from: u8, to: u8) -> Option<[[u8; 2]; 1]> {
+		if from == to {
+			None
+		} else {
+			Some([[from, to]])
+		}
+	}
+
+	fn message_queue_root(from: u8, to: u8) -> Option<Hash> {
+		message_queue(from, to).map(
+			|q| polkadot_validation::message_queue_root(q.iter())
+		)
+	}
+
+	// check that our view has all of the roots of the message queues
+	// emitted in the heads identified in `our_heads`, and none of the others.
+	fn check_roots(view: &mut View, our_heads: &[u8], n_heads: u8) -> bool {
+		for i in 0..n_heads {
+			for j in 0..n_heads {
+				if let Some(messages) = message_queue(i, j) {
+					let queue_root = message_queue_root(i, j).unwrap();
+					let messages = GossipParachainMessages {
+						queue_root,
+						messages: messages.iter().map(|m| ParachainMessage(m.to_vec())).collect(),
+					};
+
+					let had_queue = match view.validate_queue_and_note_known(&messages).0 {
+						GossipValidationResult::ProcessAndKeep(topic) => topic == queue_topic(queue_root),
+						_ => false,
+					};
+
+					if our_heads.contains(&i) != had_queue {
+						return false
+					}
+				}
+			}
+		}
+
+		true
+	}
+
+	#[test]
+	fn update_leaves_none_in_common() {
+		let mut ctx = TestChainContext::default();
+		let n_heads = 5;
+
+		for i in 0..n_heads {
+			ctx.known_map.insert(hash(i as u8), Known::Leaf);
+
+			let messages_out: Vec<_> = (0..n_heads).filter_map(|j| message_queue_root(i, j)).collect();
+
+			if !messages_out.is_empty() {
+				ctx.ingress_roots.insert(hash(i as u8), messages_out);
+			}
+		}
+
+		// initialize the view with 2 leaves.
+
+		let mut view = View::default();
+		view.update_leaves(
+			&ctx,
+			[hash(0), hash(1)].iter().cloned(),
+		).unwrap();
+
+		// we should have all queue roots that were
+		// un-routed from the perspective of those 2
+		// leaves and no others.
+
+		assert!(check_roots(&mut view, &[0, 1], n_heads));
+
+		// after updating to a disjoint set,
+		// the property that we are aware of all un-routed
+		// from the perspective of our known leaves should
+		// remain the same.
+
+		view.update_leaves(
+			&ctx,
+			[hash(2), hash(3), hash(4)].iter().cloned(),
+		).unwrap();
+
+		assert!(check_roots(&mut view, &[2, 3, 4], n_heads));
+	}
+
+	#[test]
+	fn update_leaves_overlapping() {
+		let mut ctx = TestChainContext::default();
+		let n_heads = 5;
+
+		for i in 0..n_heads {
+			ctx.known_map.insert(hash(i as u8), Known::Leaf);
+
+			let messages_out: Vec<_> = (0..n_heads).filter_map(|j| message_queue_root(i, j)).collect();
+
+			if !messages_out.is_empty() {
+				ctx.ingress_roots.insert(hash(i as u8), messages_out);
+			}
+		}
+
+		let mut view = View::default();
+		view.update_leaves(
+			&ctx,
+			[hash(0), hash(1), hash(2)].iter().cloned(),
+		).unwrap();
+
+		assert!(check_roots(&mut view, &[0, 1, 2], n_heads));
+
+		view.update_leaves(
+			&ctx,
+			[hash(2), hash(3), hash(4)].iter().cloned(),
+		).unwrap();
+
+		// after updating to a leaf-set overlapping with the prior,
+		// the property that we are aware of all un-routed
+		// from the perspective of our known leaves should
+		// remain the same.
+
+		assert!(check_roots(&mut view, &[2, 3, 4], n_heads));
+	}
+}
@@ -16,8 +16,8 @@

 //! Polkadot-specific network implementation.
 //!
-//! This manages routing for parachain statements, parachain block and extrinsic data fetching,
-//! communication between collators and validators, and more.
+//! This manages routing for parachain statements, parachain block and outgoing message
+//! data fetching, communication between collators and validators, and more.

 mod collator_pool;
 mod local_collations;
@@ -26,23 +26,29 @@ pub mod validation;
 pub mod gossip;

 use codec::{Decode, Encode};
-use futures::sync::oneshot;
+use futures::sync::{oneshot, mpsc};
+use futures::prelude::*;
 use polkadot_primitives::{Block, Hash, Header};
 use polkadot_primitives::parachain::{
 	Id as ParaId, BlockData, CollatorId, CandidateReceipt, Collation, PoVBlock,
-	StructuredUnroutedIngress, ValidatorId
+	StructuredUnroutedIngress, ValidatorId, OutgoingMessages,
 };
 use substrate_network::{
 	PeerId, RequestId, Context, StatusMessage as GenericFullStatus,
 	specialization::{Event, NetworkSpecialization as Specialization},
 };
-use self::validation::{LiveValidationSessions, RecentValidatorIds, InsertedRecentKey};
+use substrate_network::consensus_gossip::{
+	self, TopicNotification, MessageRecipient as GossipMessageRecipient, ConsensusMessage,
+};
+use self::validation::{LiveValidationLeaves, RecentValidatorIds, InsertedRecentKey};
 use self::collator_pool::{CollatorPool, Role, Action};
 use self::local_collations::LocalCollations;
 use log::{trace, debug, warn};

 use std::collections::{HashMap, HashSet};

+use crate::gossip::{POLKADOT_ENGINE_ID, GossipMessage};
+
 #[cfg(test)]
 mod tests;

@@ -69,7 +75,112 @@ mod benefit {
 type FullStatus = GenericFullStatus<Block>;

 /// Specialization of the network service for the polkadot protocol.
-pub type NetworkService = substrate_network::NetworkService<Block, PolkadotProtocol, Hash>;
+pub type PolkadotNetworkService = substrate_network::NetworkService<Block, PolkadotProtocol, Hash>;
+
+/// Basic functionality that a network has to fulfill.
+pub trait NetworkService: Send + Sync + 'static {
+	/// Get a stream of gossip messages for a given hash.
+	fn gossip_messages_for(&self, topic: Hash) -> GossipMessageStream;
+
+	/// Gossip a message on given topic.
+	fn gossip_message(&self, topic: Hash, message: GossipMessage);
+
+	/// Execute a closure with the gossip service.
+	fn with_gossip<F: Send + 'static>(&self, with: F)
+		where F: FnOnce(&mut dyn GossipService, &mut dyn Context<Block>);
+
+	/// Execute a closure with the polkadot protocol.
+	fn with_spec<F: Send + 'static>(&self, with: F)
+		where F: FnOnce(&mut PolkadotProtocol, &mut dyn Context<Block>);
+}
+
+impl NetworkService for PolkadotNetworkService {
+	fn gossip_messages_for(&self, topic: Hash) -> GossipMessageStream {
+		let (tx, rx) = std::sync::mpsc::channel();
+
+		PolkadotNetworkService::with_gossip(self, move |gossip, _| {
+			let inner_rx = gossip.messages_for(POLKADOT_ENGINE_ID, topic);
+			let _ = tx.send(inner_rx);
+		});
+
+		let topic_stream = match rx.recv() {
+			Ok(rx) => rx,
+			Err(_) => mpsc::unbounded().1, // return empty channel.
+		};
+
+		GossipMessageStream::new(topic_stream)
+	}
+
+	fn gossip_message(&self, topic: Hash, message: GossipMessage) {
+		self.gossip_consensus_message(
+			topic,
+			POLKADOT_ENGINE_ID,
+			message.encode(),
+			GossipMessageRecipient::BroadcastToAll,
+		);
+	}
+
+	fn with_gossip<F: Send + 'static>(&self, with: F)
+		where F: FnOnce(&mut dyn GossipService, &mut dyn Context<Block>)
+	{
+		PolkadotNetworkService::with_gossip(self, move |gossip, ctx| with(gossip, ctx))
+	}
+
+	fn with_spec<F: Send + 'static>(&self, with: F)
+		where F: FnOnce(&mut PolkadotProtocol, &mut dyn Context<Block>)
+	{
+		PolkadotNetworkService::with_spec(self, with)
+	}
+}
+
+/// A gossip network subservice.
+pub trait GossipService {
+	fn send_message(&mut self, ctx: &mut dyn Context<Block>, who: &PeerId, message: ConsensusMessage);
+	fn multicast(&mut self, ctx: &mut dyn Context<Block>, topic: &Hash, message: ConsensusMessage);
+}
+
+impl GossipService for consensus_gossip::ConsensusGossip<Block> {
+	fn send_message(&mut self, ctx: &mut dyn Context<Block>, who: &PeerId, message: ConsensusMessage) {
+		consensus_gossip::ConsensusGossip::send_message(self, ctx, who, message)
+	}
+
+	fn multicast(&mut self, ctx: &mut dyn Context<Block>, topic: &Hash, message: ConsensusMessage) {
+		consensus_gossip::ConsensusGossip::multicast(self, ctx, *topic, message, false)
+	}
+}
+
+/// A stream of gossip messages and an optional sender for a topic.
+pub struct GossipMessageStream {
+	topic_stream: mpsc::UnboundedReceiver<TopicNotification>,
+}
+
+impl GossipMessageStream {
+	/// Create a new instance with the given topic stream.
+	pub fn new(topic_stream: mpsc::UnboundedReceiver<TopicNotification>) -> Self {
+		Self {
+			topic_stream
+		}
+	}
+}
+
+impl Stream for GossipMessageStream {
+	type Item = (GossipMessage, Option<PeerId>);
+	type Error = ();
+
+	fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+		loop {
+			let msg = match futures::try_ready!(self.topic_stream.poll()) {
+				Some(msg) => msg,
+				None => return Ok(Async::Ready(None)),
+			};
+
+			debug!(target: "validation", "Processing statement for live validation leaf-work");
+			if let Ok(gmsg) = GossipMessage::decode(&mut &msg.message[..]) {
+				return Ok(Async::Ready(Some((gmsg, msg.sender))))
+			}
+		}
+	}
+}

 /// Status of a Polkadot node.
 #[derive(Debug, PartialEq, Eq, Clone, Encode, Decode)]
@@ -79,7 +190,7 @@ pub struct Status {

 struct PoVBlockRequest {
 	attempted_peers: HashSet<ValidatorId>,
-	validation_session_parent: Hash,
+	validation_leaf: Hash,
 	candidate_hash: Hash,
 	block_data_hash: Hash,
 	sender: oneshot::Sender<PoVBlock>,
@@ -188,10 +299,10 @@ pub struct PolkadotProtocol {
 	collators: CollatorPool,
 	validators: HashMap<ValidatorId, PeerId>,
 	local_collations: LocalCollations<Collation>,
-	live_validation_sessions: LiveValidationSessions,
+	live_validation_leaves: LiveValidationLeaves,
 	in_flight: HashMap<(RequestId, PeerId), PoVBlockRequest>,
 	pending: Vec<PoVBlockRequest>,
-	extrinsic_store: Option<::av_store::Store>,
+	availability_store: Option<av_store::Store>,
 	next_req_id: u64,
 }

@@ -204,10 +315,10 @@ impl PolkadotProtocol {
 			collating_for,
 			validators: HashMap::new(),
 			local_collations: LocalCollations::new(),
-			live_validation_sessions: LiveValidationSessions::new(),
+			live_validation_leaves: LiveValidationLeaves::new(),
 			in_flight: HashMap::new(),
 			pending: Vec::new(),
-			extrinsic_store: None,
+			availability_store: None,
 			next_req_id: 1,
 		}
 	}
@@ -224,7 +335,7 @@ impl PolkadotProtocol {

 		self.pending.push(PoVBlockRequest {
 			attempted_peers: Default::default(),
-			validation_session_parent: relay_parent,
+			validation_leaf: relay_parent,
 			candidate_hash: candidate.hash(),
 			block_data_hash: candidate.block_data_hash,
 			sender: tx,
@@ -235,15 +346,15 @@ impl PolkadotProtocol {
 		rx
 	}

-	/// Note new validation session.
-	fn new_validation_session(
+	/// Note new leaf to do validation work at
+	fn new_validation_leaf_work(
 		&mut self,
 		ctx: &mut dyn Context<Block>,
-		params: validation::SessionParams,
-	) -> validation::ValidationSession {
+		params: validation::LeafWorkParams,
+	) -> validation::LiveValidationLeaf {

-		let (session, new_local) = self.live_validation_sessions
-			.new_validation_session(params);
+		let (work, new_local) = self.live_validation_leaves
+			.new_validation_leaf(params);

 		if let Some(new_local) = new_local {
 			for (id, peer_data) in self.peers.iter_mut()
@@ -257,12 +368,12 @@ impl PolkadotProtocol {
 			}
 		}

-		session
+		work
 	}

 	// true indicates that it was removed actually.
 	fn remove_validation_session(&mut self, parent_hash: Hash) -> bool {
-		self.live_validation_sessions.remove(parent_hash)
+		self.live_validation_leaves.remove(parent_hash)
 	}

 	fn dispatch_pending_requests(&mut self, ctx: &mut dyn Context<Block>) {
@@ -272,10 +383,10 @@ impl PolkadotProtocol {
 		let in_flight = &mut self.in_flight;

 		for mut pending in ::std::mem::replace(&mut self.pending, Vec::new()) {
-			let parent = pending.validation_session_parent;
+			let parent = pending.validation_leaf;
 			let c_hash = pending.candidate_hash;

-			let still_pending = self.live_validation_sessions.with_pov_block(&parent, &c_hash, |x| match x {
+			let still_pending = self.live_validation_leaves.with_pov_block(&parent, &c_hash, |x| match x {
 				Ok(data @ &_) => {
 					// answer locally.
 					let _ = pending.sender.send(data.clone());
@@ -305,7 +416,7 @@ impl PolkadotProtocol {
 						Some(pending)
 					}
 				}
-				Err(None) => None, // no such known validation session. prune out.
+				Err(None) => None, // no such known validation leaf-work. prune out.
 			});

 			if let Some(pending) = still_pending {
@@ -321,7 +432,7 @@ impl PolkadotProtocol {
 		match msg {
 			Message::ValidatorId(key) => self.on_session_key(ctx, who, key),
 			Message::RequestPovBlock(req_id, relay_parent, candidate_hash) => {
-				let pov_block = self.live_validation_sessions.with_pov_block(
+				let pov_block = self.live_validation_leaves.with_pov_block(
 					&relay_parent,
 					&candidate_hash,
 					|res| res.ok().map(|b| b.clone()),
@@ -330,13 +441,13 @@ impl PolkadotProtocol {
 				send_polkadot_message(ctx, who, Message::PovBlock(req_id, pov_block));
 			}
 			Message::RequestBlockData(req_id, relay_parent, candidate_hash) => {
-				let block_data = self.live_validation_sessions
+				let block_data = self.live_validation_leaves
 					.with_pov_block(
 						&relay_parent,
 						&candidate_hash,
 						|res| res.ok().map(|b| b.block_data.clone()),
 					)
-					.or_else(|| self.extrinsic_store.as_ref()
+					.or_else(|| self.availability_store.as_ref()
 						.and_then(|s| s.block_data(relay_parent, candidate_hash))
 					);

@@ -507,7 +618,7 @@ impl Specialization<Block> for PolkadotProtocol {

 		// send session keys.
 		if peer_info.should_send_key() {
-			for local_session_key in self.live_validation_sessions.recent_keys() {
+			for local_session_key in self.live_validation_leaves.recent_keys() {
 				peer_info.collator_state.send_key(local_session_key.clone(), |msg| send_polkadot_message(
 					ctx,
 					who.clone(),
@@ -549,7 +660,7 @@ impl Specialization<Block> for PolkadotProtocol {
 						let (sender, _) = oneshot::channel();
 						pending.push(::std::mem::replace(val, PoVBlockRequest {
 							attempted_peers: Default::default(),
-							validation_session_parent: Default::default(),
+							validation_leaf: Default::default(),
 							candidate_hash: Default::default(),
 							block_data_hash: Default::default(),
 							canon_roots: StructuredUnroutedIngress(Vec::new()),
@@ -676,16 +787,35 @@ impl PolkadotProtocol {

 impl PolkadotProtocol {
 	/// Add a local collation and broadcast it to the necessary peers.
+	///
+	/// This should be called by a collator intending to get the locally-collated
+	/// block into the hands of validators.
+	/// It also places the outgoing message and block data in the local availability store.
 	pub fn add_local_collation(
 		&mut self,
 		ctx: &mut dyn Context<Block>,
 		relay_parent: Hash,
 		targets: HashSet<ValidatorId>,
 		collation: Collation,
-	) {
+		outgoing_targeted: OutgoingMessages,
+	) -> std::io::Result<()> {
 		debug!(target: "p_net", "Importing local collation on relay parent {:?} and parachain {:?}",
 			relay_parent, collation.receipt.parachain_index);

+		let outgoing_queues = polkadot_validation::outgoing_queues(&outgoing_targeted)
+			.map(|(_target, root, data)| (root, data))
+			.collect();
+
+		if let Some(ref availability_store) = self.availability_store {
+			availability_store.make_available(av_store::Data {
+				relay_parent,
+				parachain_id: collation.receipt.parachain_index,
+				candidate_hash: collation.receipt.hash(),
+				block_data: collation.pov.block_data.clone(),
+				outgoing_queues: Some(outgoing_queues),
+			})?;
+		}
+
 		for (primary, cloned_collation) in self.local_collations.add_collation(relay_parent, targets, collation.clone()) {
 			match self.validators.get(&primary) {
 				Some(who) => {
@@ -700,10 +830,13 @@ impl PolkadotProtocol {
 					warn!(target: "polkadot_network", "Encountered tracked but disconnected validator {:?}", primary),
 			}
 		}
+
+		Ok(())
 	}

-	/// register availability store.
-	pub fn register_availability_store(&mut self, extrinsic_store: ::av_store::Store) {
-		self.extrinsic_store = Some(extrinsic_store);
+	/// Give the network protocol a handle to an availability store, used for
+	/// circulation of parachain data required for validation.
+	pub fn register_availability_store(&mut self, availability_store: ::av_store::Store) {
+		self.availability_store = Some(availability_store);
 	}
 }
@@ -29,7 +29,7 @@ use polkadot_validation::{
 };
 use polkadot_primitives::{Block, Hash};
 use polkadot_primitives::parachain::{
-	Extrinsic, CandidateReceipt, ParachainHost, ValidatorIndex, Collation, PoVBlock,
+	OutgoingMessages, CandidateReceipt, ParachainHost, ValidatorIndex, Collation, PoVBlock,
 };
 use crate::gossip::{RegisteredMessageValidator, GossipMessage, GossipStatement};

@@ -41,7 +41,8 @@ use std::collections::{HashMap, HashSet};
 use std::io;
 use std::sync::Arc;

-use crate::validation::{self, SessionDataFetcher, NetworkService, Executor};
+use crate::validation::{self, LeafWorkDataFetcher, Executor};
+use crate::NetworkService;

 /// Compute the gossip topic for attestations on the given parent hash.
 pub(crate) fn attestation_topic(parent_hash: Hash) -> Hash {
@@ -72,7 +73,7 @@ pub(crate) fn checked_statements<N: NetworkService>(network: &N, topic: Hash) ->
 pub struct Router<P, E, N: NetworkService, T> {
 	table: Arc<SharedTable>,
 	attestation_topic: Hash,
-	fetcher: SessionDataFetcher<P, E, N, T>,
+	fetcher: LeafWorkDataFetcher<P, E, N, T>,
 	deferred_statements: Arc<Mutex<DeferredStatements>>,
 	message_validator: RegisteredMessageValidator,
 }
@@ -80,7 +81,7 @@ pub struct Router<P, E, N: NetworkService, T> {
 impl<P, E, N: NetworkService, T> Router<P, E, N, T> {
 	pub(crate) fn new(
 		table: Arc<SharedTable>,
-		fetcher: SessionDataFetcher<P, E, N, T>,
+		fetcher: LeafWorkDataFetcher<P, E, N, T>,
 		message_validator: RegisteredMessageValidator,
 	) -> Self {
 		let parent_hash = fetcher.parent_hash();
@@ -196,7 +197,7 @@ impl<P: ProvideRuntimeApi + Send + Sync + 'static, E, N, T> Router<P, E, N, T> w
 				knowledge.lock().note_candidate(
 					candidate_hash,
 					Some(validated.pov_block().clone()),
-					validated.extrinsic().cloned(),
+					validated.outgoing_messages().cloned(),
 				);

 				// propagate the statement.
@@ -224,13 +225,13 @@ impl<P: ProvideRuntimeApi + Send, E, N, T> TableRouter for Router<P, E, N, T> wh
 	type Error = io::Error;
 	type FetchValidationProof = validation::PoVReceiver;

-	fn local_collation(&self, collation: Collation, extrinsic: Extrinsic) {
+	fn local_collation(&self, collation: Collation, outgoing: OutgoingMessages) {
 		// produce a signed statement
 		let hash = collation.receipt.hash();
 		let validated = Validated::collated_local(
 			collation.receipt,
 			collation.pov.clone(),
-			extrinsic.clone(),
+			outgoing.clone(),
 		);

 		let statement = GossipStatement::new(
@@ -242,7 +243,7 @@ impl<P: ProvideRuntimeApi + Send, E, N, T> TableRouter for Router<P, E, N, T> wh
 		);

 		// give to network to make available.
-		self.fetcher.knowledge().lock().note_candidate(hash, Some(collation.pov), Some(extrinsic));
+		self.fetcher.knowledge().lock().note_candidate(hash, Some(collation.pov), Some(outgoing));
 		self.network().gossip_message(self.attestation_topic, statement.into());
 	}

@@ -18,7 +18,7 @@

 use std::collections::HashMap;
 use super::{PolkadotProtocol, Status, Message, FullStatus};
-use crate::validation::SessionParams;
+use crate::validation::LeafWorkParams;

 use polkadot_validation::GenericStatement;
 use polkadot_primitives::{Block, Hash};
@@ -77,6 +77,28 @@ impl TestContext {
 	}
 }

+#[derive(Default)]
+pub struct TestChainContext {
+	pub known_map: HashMap<Hash, crate::gossip::Known>,
+	pub ingress_roots: HashMap<Hash, Vec<Hash>>,
+}
+
+impl crate::gossip::ChainContext for TestChainContext {
+	fn is_known(&self, block_hash: &Hash) -> Option<crate::gossip::Known> {
+		self.known_map.get(block_hash).map(|x| x.clone())
+	}
+
+	fn leaf_unrouted_roots(&self, leaf: &Hash, with_queue_root: &mut dyn FnMut(&Hash))
+		-> Result<(), substrate_client::error::Error>
+	{
+		for root in self.ingress_roots.get(leaf).into_iter().flat_map(|roots| roots) {
+			with_queue_root(root)
+		}
+
+		Ok(())
+	}
+}
+
 fn make_pov(block_data: Vec<u8>) -> PoVBlock {
 	PoVBlock {
 		block_data: BlockData(block_data),
@@ -96,8 +118,8 @@ fn make_status(status: &Status, roles: Roles) -> FullStatus {
 	}
 }

-fn make_validation_session(parent_hash: Hash, local_key: ValidatorId) -> SessionParams {
-	SessionParams {
+fn make_validation_leaf_work(parent_hash: Hash, local_key: ValidatorId) -> LeafWorkParams {
+	LeafWorkParams {
 		local_session_key: Some(local_key),
 		parent_hash,
 		authorities: Vec::new(),
@@ -129,8 +151,8 @@ fn sends_session_key() {

 	{
 		let mut ctx = TestContext::default();
-		let params = make_validation_session(parent_hash, local_key.clone());
-		protocol.new_validation_session(&mut ctx, params);
+		let params = make_validation_leaf_work(parent_hash, local_key.clone());
+		protocol.new_validation_leaf_work(&mut ctx, params);
 		assert!(ctx.has_message(peer_a, Message::ValidatorId(local_key.clone())));
 	}

@@ -169,8 +191,8 @@ fn fetches_from_those_with_knowledge() {

 	let status = Status { collating_for: None };

-	let params = make_validation_session(parent_hash, local_key.clone());
-	let session = protocol.new_validation_session(&mut TestContext::default(), params);
+	let params = make_validation_leaf_work(parent_hash, local_key.clone());
+	let session = protocol.new_validation_leaf_work(&mut TestContext::default(), params);
 	let knowledge = session.knowledge();

 	knowledge.lock().note_statement(a_key.clone(), &GenericStatement::Valid(candidate_hash));
@@ -259,7 +281,7 @@ fn fetches_available_block_data() {
 		parachain_id: para_id,
 		candidate_hash,
 		block_data: block_data.clone(),
-		extrinsic: None,
+		outgoing_queues: None,
 	}).unwrap();

 	// connect peer A
@@ -323,13 +345,13 @@ fn many_session_keys() {
 	let local_key_a: ValidatorId = [3; 32].unchecked_into();
 	let local_key_b: ValidatorId = [4; 32].unchecked_into();

-	let params_a = make_validation_session(parent_a, local_key_a.clone());
-	let params_b = make_validation_session(parent_b, local_key_b.clone());
+	let params_a = make_validation_leaf_work(parent_a, local_key_a.clone());
+	let params_b = make_validation_leaf_work(parent_b, local_key_b.clone());

-	protocol.new_validation_session(&mut TestContext::default(), params_a);
-	protocol.new_validation_session(&mut TestContext::default(), params_b);
+	protocol.new_validation_leaf_work(&mut TestContext::default(), params_a);
+	protocol.new_validation_leaf_work(&mut TestContext::default(), params_b);

-	assert_eq!(protocol.live_validation_sessions.recent_keys(), &[local_key_a.clone(), local_key_b.clone()]);
+	assert_eq!(protocol.live_validation_leaves.recent_keys(), &[local_key_a.clone(), local_key_b.clone()]);

 	let peer_a = PeerId::random();

@@ -18,18 +18,17 @@

 #![allow(unused)]

-use crate::validation::{NetworkService, GossipService, GossipMessageStream};
 use crate::gossip::GossipMessage;
 use substrate_network::Context as NetContext;
 use substrate_network::consensus_gossip::TopicNotification;
 use substrate_primitives::{NativeOrEncoded, ExecutionContext};
 use substrate_keyring::Sr25519Keyring;
-use crate::PolkadotProtocol;
+use crate::{GossipService, PolkadotProtocol, NetworkService, GossipMessageStream};

-use polkadot_validation::{SharedTable, MessagesFrom, Network};
-use polkadot_primitives::{Block, Hash, Header, BlockId};
+use polkadot_validation::{SharedTable, Network};
+use polkadot_primitives::{Block, BlockNumber, Hash, Header, BlockId};
 use polkadot_primitives::parachain::{
-	Id as ParaId, Chain, DutyRoster, ParachainHost, OutgoingMessage,
+	Id as ParaId, Chain, DutyRoster, ParachainHost, TargetedMessage,
 	ValidatorId, StructuredUnroutedIngress, BlockIngressRoots, Status,
 	FeeSchedule, HeadData,
 };
@@ -43,7 +42,7 @@ use std::sync::Arc;
 use futures::{prelude::*, sync::mpsc};
 use codec::Encode;

-use super::TestContext;
+use super::{TestContext, TestChainContext};

 type TaskExecutor = Arc<dyn futures::future::Executor<Box<dyn Future<Item = (), Error = ()> + Send>> + Send + Sync>;

@@ -315,10 +314,10 @@ impl ParachainHost<Block> for RuntimeApi {
 		&self,
 		_at: &BlockId,
 		_: ExecutionContext,
-		id: Option<ParaId>,
+		id: Option<(ParaId, Option<BlockNumber>)>,
 		_: Vec<u8>,
 	) -> ClientResult<NativeOrEncoded<Option<StructuredUnroutedIngress>>> {
-		let id = id.unwrap();
+		let (id, _) = id.unwrap();
 		Ok(NativeOrEncoded::Native(self.data.lock().ingress.get(&id).cloned()))
 	}
 }
@@ -348,7 +347,7 @@ fn build_network(n: usize, executor: TaskExecutor) -> Built {
 		});

 		let message_val = crate::gossip::RegisteredMessageValidator::new_test(
-			|_hash: &_| Some(crate::gossip::Known::Leaf),
+			TestChainContext::default(),
 			Box::new(|_, _| {}),
 		);

@@ -376,7 +375,7 @@ struct IngressBuilder {
 }

 impl IngressBuilder {
-	fn add_messages(&mut self, source: ParaId, messages: &[OutgoingMessage]) {
+	fn add_messages(&mut self, source: ParaId, messages: &[TargetedMessage]) {
 		for message in messages {
 			let target = message.target;
 			self.egress.entry((source, target)).or_insert_with(Vec::new).push(message.data.clone());
@@ -14,29 +14,24 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.

-//! The "validation session" networking code built on top of the base network service.
+//! The "validation leaf work" networking code built on top of the base network service.
 //!
 //! This fulfills the `polkadot_validation::Network` trait, providing a hook to be called
-//! each time a validation session begins on a new chain head.
+//! each time validation leaf work begins on a new chain head.

-use crate::gossip::GossipMessage;
 use sr_primitives::traits::ProvideRuntimeApi;
-use substrate_network::{PeerId, Context as NetContext};
-use substrate_network::consensus_gossip::{
-	self, TopicNotification, MessageRecipient as GossipMessageRecipient, ConsensusMessage,
-};
+use substrate_network::PeerId;
 use polkadot_validation::{
 	Network as ParachainNetwork, SharedTable, Collators, Statement, GenericStatement, SignedStatement,
 };
 use polkadot_primitives::{Block, BlockId, Hash};
 use polkadot_primitives::parachain::{
-	Id as ParaId, Collation, Extrinsic, ParachainHost, CandidateReceipt, CollatorId,
-	ValidatorId, PoVBlock, ValidatorIndex
+	Id as ParaId, Collation, OutgoingMessages, ParachainHost, CandidateReceipt, CollatorId,
+	ValidatorId, PoVBlock
 };

 use futures::prelude::*;
 use futures::future::{self, Executor as FutureExecutor};
-use futures::sync::mpsc;
 use futures::sync::oneshot::{self, Receiver};

 use std::collections::hash_map::{HashMap, Entry};
@@ -45,17 +40,15 @@ use std::sync::Arc;

 use arrayvec::ArrayVec;
 use parking_lot::Mutex;
-use log::{debug, warn};
+use log::warn;

 use crate::router::Router;
-use crate::gossip::{POLKADOT_ENGINE_ID, RegisteredMessageValidator, MessageValidationData};
+use crate::gossip::{RegisteredMessageValidator, MessageValidationData};

-use super::PolkadotProtocol;
+use super::NetworkService;

 pub use polkadot_validation::Incoming;

-use codec::{Encode, Decode};
-
 /// An executor suitable for dispatching async consensus tasks.
 pub trait Executor {
 	fn spawn<F: Future<Item=(),Error=()> + Send + 'static>(&self, f: F);
@@ -83,108 +76,8 @@ impl Executor for Arc<
 	}
 }

-/// A gossip network subservice.
-pub trait GossipService {
-	fn send_message(&mut self, ctx: &mut dyn NetContext<Block>, who: &PeerId, message: ConsensusMessage);
-}
-
-impl GossipService for consensus_gossip::ConsensusGossip<Block> {
-	fn send_message(&mut self, ctx: &mut dyn NetContext<Block>, who: &PeerId, message: ConsensusMessage) {
-		consensus_gossip::ConsensusGossip::send_message(self, ctx, who, message)
-	}
-}
-
-/// A stream of gossip messages and an optional sender for a topic.
-pub struct GossipMessageStream {
-	topic_stream: mpsc::UnboundedReceiver<TopicNotification>,
-}
-
-impl GossipMessageStream {
-	/// Create a new instance with the given topic stream.
-	pub fn new(topic_stream: mpsc::UnboundedReceiver<TopicNotification>) -> Self {
-		Self {
-			topic_stream
-		}
-	}
-}
-
-impl Stream for GossipMessageStream {
-	type Item = (GossipMessage, Option<PeerId>);
-	type Error = ();
-
-	fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
-		loop {
-			let msg = match futures::try_ready!(self.topic_stream.poll()) {
-				Some(msg) => msg,
-				None => return Ok(Async::Ready(None)),
-			};
-
-			debug!(target: "validation", "Processing statement for live validation session");
-			if let Ok(gmsg) = GossipMessage::decode(&mut &msg.message[..]) {
-				return Ok(Async::Ready(Some((gmsg, msg.sender))))
-			}
-		}
-	}
-}
-
-/// Basic functionality that a network has to fulfill.
-pub trait NetworkService: Send + Sync + 'static {
-	/// Get a stream of gossip messages for a given hash.
-	fn gossip_messages_for(&self, topic: Hash) -> GossipMessageStream;
-
-	/// Gossip a message on given topic.
-	fn gossip_message(&self, topic: Hash, message: GossipMessage);
-
-	/// Execute a closure with the gossip service.
-	fn with_gossip<F: Send + 'static>(&self, with: F)
-		where F: FnOnce(&mut dyn GossipService, &mut dyn NetContext<Block>);
-
-	/// Execute a closure with the polkadot protocol.
-	fn with_spec<F: Send + 'static>(&self, with: F)
-		where F: FnOnce(&mut PolkadotProtocol, &mut dyn NetContext<Block>);
-}
-
-impl NetworkService for super::NetworkService {
-	fn gossip_messages_for(&self, topic: Hash) -> GossipMessageStream {
-		let (tx, rx) = std::sync::mpsc::channel();
-
-		super::NetworkService::with_gossip(self, move |gossip, _| {
-			let inner_rx = gossip.messages_for(POLKADOT_ENGINE_ID, topic);
-			let _ = tx.send(inner_rx);
-		});
-
-		let topic_stream = match rx.recv() {
-			Ok(rx) => rx,
-			Err(_) => mpsc::unbounded().1, // return empty channel.
-		};
-
-		GossipMessageStream::new(topic_stream)
-	}
-
-	fn gossip_message(&self, topic: Hash, message: GossipMessage) {
-		self.gossip_consensus_message(
-			topic,
-			POLKADOT_ENGINE_ID,
-			message.encode(),
-			GossipMessageRecipient::BroadcastToAll,
-		);
-	}
-
-	fn with_gossip<F: Send + 'static>(&self, with: F)
-		where F: FnOnce(&mut dyn GossipService, &mut dyn NetContext<Block>)
-	{
-		super::NetworkService::with_gossip(self, move |gossip, ctx| with(gossip, ctx))
-	}
-
-	fn with_spec<F: Send + 'static>(&self, with: F)
-		where F: FnOnce(&mut PolkadotProtocol, &mut dyn NetContext<Block>)
-	{
-		super::NetworkService::with_spec(self, with)
-	}
-}
-
-/// Params to a current validation session.
-pub struct SessionParams {
+/// Params to instantiate validation work on a block-DAG leaf.
+pub struct LeafWorkParams {
 	/// The local session key.
 	pub local_session_key: Option<ValidatorId>,
 	/// The parent hash.
@@ -234,20 +127,22 @@ impl<P, E, N, T> ValidationNetwork<P, E, N, T> where
 	N: NetworkService,
 	T: Clone + Executor + Send + Sync + 'static,
 {
-	/// Instantiate session data fetcher at a parent hash.
+	/// Instantiate block-DAG leaf work
+	/// (i.e. the work we want to be done by validators at some chain-head)
+	/// at a parent hash.
 	///
 	/// If the used session key is new, it will be broadcast to peers.
-	/// If a validation session was already instantiated at this parent hash,
+	/// If any validation leaf-work was already instantiated at this parent hash,
 	/// the underlying instance will be shared.
 	///
-	/// If there was already a validation session instantiated and a different
+	/// If there was already validation leaf-work instantiated and a different
 	/// session key was set, then the new key will be ignored.
 	///
 	/// This implies that there can be multiple services intantiating validation
-	/// session instances safely, but they should all be coordinated on which session keys
+	/// leaf-work instances safely, but they should all be coordinated on which session keys
 	/// are being used.
-	pub fn instantiate_session(&self, params: SessionParams)
-		-> oneshot::Receiver<SessionDataFetcher<P, E, N, T>>
+	pub fn instantiate_leaf_work(&self, params: LeafWorkParams)
+		-> oneshot::Receiver<LeafWorkDataFetcher<P, E, N, T>>
 	{
 		let parent_hash = params.parent_hash;
 		let network = self.network.clone();
@@ -255,34 +150,27 @@ impl<P, E, N, T> ValidationNetwork<P, E, N, T> where
 		let task_executor = self.executor.clone();
 		let exit = self.exit.clone();
 		let message_validator = self.message_validator.clone();
-		let index_mapping = params.authorities
-			.iter()
-			.enumerate()
-			.map(|(i, k)| (i as ValidatorIndex, k.clone()))
-			.collect();
+		let authorities = params.authorities.clone();

 		let (tx, rx) = oneshot::channel();

-		{
-			let message_validator = self.message_validator.clone();
-			let authorities = params.authorities.clone();
-			self.network.with_gossip(move |gossip, ctx| {
-				message_validator.note_session(
-					parent_hash,
-					MessageValidationData { authorities, index_mapping },
-					|peer_id, message| gossip.send_message(ctx, peer_id, message),
-				);
-			});
-		}
-
 		self.network.with_spec(move |spec, ctx| {
-			let session = spec.new_validation_session(ctx, params);
-			let _ = tx.send(SessionDataFetcher {
+			let actions = message_validator.new_local_leaf(
+				parent_hash,
+				MessageValidationData { authorities },
+				|queue_root| spec.availability_store.as_ref()
+					.and_then(|store| store.queue_by_root(queue_root))
+			);
+
+			network.with_gossip(move |gossip, ctx| actions.perform(gossip, ctx));
+
+			let work = spec.new_validation_leaf_work(ctx, params);
+			let _ = tx.send(LeafWorkDataFetcher {
 				network,
 				api,
 				task_executor,
 				parent_hash,
-				knowledge: session.knowledge().clone(),
+				knowledge: work.knowledge().clone(),
 				exit,
 				message_validator,
 			});
@@ -335,7 +223,7 @@ impl<P, E, N, T> ParachainNetwork for ValidationNetwork<P, E, N, T> where
 		let parent_hash = *table.consensus_parent_hash();
 		let local_session_key = table.session_key();

-		let build_fetcher = self.instantiate_session(SessionParams {
+		let build_fetcher = self.instantiate_leaf_work(LeafWorkParams {
 			local_session_key,
 			parent_hash,
 			authorities: authorities.to_vec(),
@@ -421,9 +309,9 @@ impl<P, E: Clone, N, T: Clone> Collators for ValidationNetwork<P, E, N, T> where
 #[derive(Default)]
 struct KnowledgeEntry {
 	knows_block_data: Vec<ValidatorId>,
-	knows_extrinsic: Vec<ValidatorId>,
+	knows_outgoing: Vec<ValidatorId>,
 	pov: Option<PoVBlock>,
-	extrinsic: Option<Extrinsic>,
+	outgoing_messages: Option<OutgoingMessages>,
 }

 /// Tracks knowledge of peers.
@@ -442,18 +330,18 @@ impl Knowledge {
 	/// Note a statement seen from another validator.
 	pub(crate) fn note_statement(&mut self, from: ValidatorId, statement: &Statement) {
 		// those proposing the candidate or declaring it valid know everything.
-		// those claiming it invalid do not have the extrinsic data as it is
+		// those claiming it invalid do not have the outgoing messages data as it is
 		// generated by valid execution.
 		match *statement {
 			GenericStatement::Candidate(ref c) => {
 				let entry = self.candidates.entry(c.hash()).or_insert_with(Default::default);
 				entry.knows_block_data.push(from.clone());
-				entry.knows_extrinsic.push(from);
+				entry.knows_outgoing.push(from);
 			}
 			GenericStatement::Valid(ref hash) => {
 				let entry = self.candidates.entry(*hash).or_insert_with(Default::default);
 				entry.knows_block_data.push(from.clone());
-				entry.knows_extrinsic.push(from);
+				entry.knows_outgoing.push(from);
 			}
 			GenericStatement::Invalid(ref hash) => self.candidates.entry(*hash)
 				.or_insert_with(Default::default)
@@ -463,10 +351,15 @@ impl Knowledge {
 	}

 	/// Note a candidate collated or seen locally.
-	pub(crate) fn note_candidate(&mut self, hash: Hash, pov: Option<PoVBlock>, extrinsic: Option<Extrinsic>) {
+	pub(crate) fn note_candidate(
+		&mut self,
+		hash: Hash,
+		pov: Option<PoVBlock>,
+		outgoing_messages: Option<OutgoingMessages>,
+	) {
 		let entry = self.candidates.entry(hash).or_insert_with(Default::default);
 		entry.pov = entry.pov.take().or(pov);
-		entry.extrinsic = entry.extrinsic.take().or(extrinsic);
+		entry.outgoing_messages = entry.outgoing_messages.take().or(outgoing_messages);
 	}
 }

@@ -492,19 +385,19 @@ impl Future for IncomingReceiver {
 	}
 }

-/// A current validation session instance.
+/// A current validation leaf-work instance
 #[derive(Clone)]
-pub(crate) struct ValidationSession {
+pub(crate) struct LiveValidationLeaf {
 	parent_hash: Hash,
 	knowledge: Arc<Mutex<Knowledge>>,
 	local_session_key: Option<ValidatorId>,
 }

-impl ValidationSession {
-	/// Create a new validation session instance. Needs to be attached to the
+impl LiveValidationLeaf {
+	/// Create a new validation leaf-work instance. Needs to be attached to the
 	/// network.
-	pub(crate) fn new(params: SessionParams) -> Self {
-		ValidationSession {
+	pub(crate) fn new(params: LeafWorkParams) -> Self {
+		LiveValidationLeaf {
 			parent_hash: params.parent_hash,
 			knowledge: Arc::new(Mutex::new(Knowledge::new())),
 			local_session_key: params.local_session_key,
@@ -577,32 +470,32 @@ impl RecentValidatorIds {
 	}
 }

-/// Manages requests and keys for live validation session instances.
-pub(crate) struct LiveValidationSessions {
+/// Manages requests and keys for live validation leaf-work instances.
+pub(crate) struct LiveValidationLeaves {
 	// recent local session keys.
 	recent: RecentValidatorIds,
-	// live validation session instances, on `parent_hash`. refcount retained alongside.
-	live_instances: HashMap<Hash, (usize, ValidationSession)>,
+	// live validation leaf-work instances, on `parent_hash`. refcount retained alongside.
+	live_instances: HashMap<Hash, (usize, LiveValidationLeaf)>,
 }

-impl LiveValidationSessions {
-	/// Create a new `LiveValidationSessions`
+impl LiveValidationLeaves {
+	/// Create a new `LiveValidationLeaves`
 	pub(crate) fn new() -> Self {
-		LiveValidationSessions {
+		LiveValidationLeaves {
 			recent: Default::default(),
 			live_instances: HashMap::new(),
 		}
 	}

-	/// Note new validation session. If the used session key is new,
+	/// Note new leaf for validation work. If the used session key is new,
 	/// it returns it to be broadcasted to peers.
 	///
-	/// If there was already a validation session instantiated and a different
+	/// If there was already work instantiated at this leaf and a different
 	/// session key was set, then the new key will be ignored.
-	pub(crate) fn new_validation_session(
+	pub(crate) fn new_validation_leaf(
 		&mut self,
-		params: SessionParams,
-	) -> (ValidationSession, Option<ValidatorId>) {
+		params: LeafWorkParams,
+	) -> (LiveValidationLeaf, Option<ValidatorId>) {
 		let parent_hash = params.parent_hash;

 		let key = params.local_session_key.clone();
@@ -629,19 +522,19 @@ impl LiveValidationSessions {
 			return (prev.clone(), maybe_new)
 		}

-		let session = ValidationSession::new(params);
-		self.live_instances.insert(parent_hash, (1, session.clone()));
+		let leaf_work = LiveValidationLeaf::new(params);
+		self.live_instances.insert(parent_hash, (1, leaf_work.clone()));

-		(session, check_new_key())
+		(leaf_work, check_new_key())
 	}

-	/// Remove validation session. true indicates that it was actually removed.
+	/// Remove validation leaf-work. true indicates that it was actually removed.
 	pub(crate) fn remove(&mut self, parent_hash: Hash) -> bool {
 		let maybe_removed = if let Entry::Occupied(mut entry) = self.live_instances.entry(parent_hash) {
 			entry.get_mut().0 -= 1;
 			if entry.get().0 == 0 {
-				let (_, session) = entry.remove();
-				Some(session)
+				let (_, leaf_work) = entry.remove();
+				Some(leaf_work)
 			} else {
 				None
 			}
@@ -649,12 +542,12 @@ impl LiveValidationSessions {
 			None
 		};

-		let session = match maybe_removed {
+		let leaf_work = match maybe_removed {
 			None => return false,
 			Some(s) => s,
 		};

-		if let Some(ref key) = session.local_session_key {
+		if let Some(ref key) = leaf_work.local_session_key {
 			let key_still_used = self.live_instances.values()
 				.any(|c| c.1.local_session_key.as_ref() == Some(key));

@@ -671,12 +564,12 @@ impl LiveValidationSessions {
 		self.recent.as_slice()
 	}

-	/// Call a closure with pov-data from validation session at parent hash for a given
+	/// Call a closure with pov-data from validation leaf-work at parent hash for a given
 	/// candidate-receipt hash.
 	///
-	/// This calls the closure with `Some(data)` where the session and data are live,
-	/// `Err(Some(keys))` when the session is live but the data unknown, with a list of keys
-	/// who have the data, and `Err(None)` where the session is unknown.
+	/// This calls the closure with `Some(data)` where the leaf-work and data are live,
+	/// `Err(Some(keys))` when the leaf-work is live but the data unknown, with a list of keys
+	/// who have the data, and `Err(None)` where the leaf-work is unknown.
 	pub(crate) fn with_pov_block<F, U>(&self, parent_hash: &Hash, c_hash: &Hash, f: F) -> U
 		where F: FnOnce(Result<&PoVBlock, Option<&[ValidatorId]>>) -> U
 	{
@@ -716,8 +609,8 @@ impl Future for PoVReceiver {
 	}
 }

-/// Can fetch data for a given validation session
-pub struct SessionDataFetcher<P, E, N: NetworkService, T> {
+/// Can fetch data for a given validation leaf-work instance.
+pub struct LeafWorkDataFetcher<P, E, N: NetworkService, T> {
 	network: Arc<N>,
 	api: Arc<P>,
 	exit: E,
@@ -727,7 +620,7 @@ pub struct SessionDataFetcher<P, E, N: NetworkService, T> {
 	message_validator: RegisteredMessageValidator,
 }

-impl<P, E, N: NetworkService, T> SessionDataFetcher<P, E, N, T> {
+impl<P, E, N: NetworkService, T> LeafWorkDataFetcher<P, E, N, T> {
 	/// Get the parent hash.
 	pub(crate) fn parent_hash(&self) -> Hash {
 		self.parent_hash
@@ -759,9 +652,9 @@ impl<P, E, N: NetworkService, T> SessionDataFetcher<P, E, N, T> {
 	}
 }

-impl<P, E: Clone, N: NetworkService, T: Clone> Clone for SessionDataFetcher<P, E, N, T> {
+impl<P, E: Clone, N: NetworkService, T: Clone> Clone for LeafWorkDataFetcher<P, E, N, T> {
 	fn clone(&self) -> Self {
-		SessionDataFetcher {
+		LeafWorkDataFetcher {
 			network: self.network.clone(),
 			api: self.api.clone(),
 			task_executor: self.task_executor.clone(),
@@ -773,7 +666,7 @@ impl<P, E: Clone, N: NetworkService, T: Clone> Clone for SessionDataFetcher<P, E
 	}
 }

-impl<P: ProvideRuntimeApi + Send, E, N, T> SessionDataFetcher<P, E, N, T> where
+impl<P: ProvideRuntimeApi + Send, E, N, T> LeafWorkDataFetcher<P, E, N, T> where
 	P::Api: ParachainHost<Block>,
 	N: NetworkService,
 	T: Clone + Executor + Send + 'static,
@@ -784,7 +677,11 @@ impl<P: ProvideRuntimeApi + Send, E, N, T> SessionDataFetcher<P, E, N, T> where
 		let parachain = candidate.parachain_index;
 		let parent_hash = self.parent_hash;

-		let canon_roots = self.api.runtime_api().ingress(&BlockId::hash(parent_hash), parachain)
+		let canon_roots = self.api.runtime_api().ingress(
+			&BlockId::hash(parent_hash),
+			parachain,
+			None,
+		)
 			.map_err(|e|
 				format!(
 					"Cannot fetch ingress for parachain {:?} at {:?}: {:?}",
@@ -862,39 +759,39 @@ mod tests {
 	}

 	#[test]
-	fn add_new_sessions_works() {
-		let mut live_sessions = LiveValidationSessions::new();
+	fn add_new_leaf_work_works() {
+		let mut live_leaves = LiveValidationLeaves::new();
 		let key_a: ValidatorId = [0; 32].unchecked_into();
 		let key_b: ValidatorId = [1; 32].unchecked_into();
 		let parent_hash = [0xff; 32].into();

-		let (session, new_key) = live_sessions.new_validation_session(SessionParams {
+		let (leaf_work, new_key) = live_leaves.new_validation_leaf(LeafWorkParams {
 			parent_hash,
 			local_session_key: None,
 			authorities: Vec::new(),
 		});

-		let knowledge = session.knowledge().clone();
+		let knowledge = leaf_work.knowledge().clone();

 		assert!(new_key.is_none());

-		let (session, new_key) = live_sessions.new_validation_session(SessionParams {
+		let (leaf_work, new_key) = live_leaves.new_validation_leaf(LeafWorkParams {
 			parent_hash,
 			local_session_key: Some(key_a.clone()),
 			authorities: Vec::new(),
 		});

 		// check that knowledge points to the same place.
-		assert_eq!(&**session.knowledge() as *const _, &*knowledge as *const _);
+		assert_eq!(&**leaf_work.knowledge() as *const _, &*knowledge as *const _);
 		assert_eq!(new_key, Some(key_a.clone()));

-		let (session, new_key) = live_sessions.new_validation_session(SessionParams {
+		let (leaf_work, new_key) = live_leaves.new_validation_leaf(LeafWorkParams {
 			parent_hash,
 			local_session_key: Some(key_b.clone()),
 			authorities: Vec::new(),
 		});

-		assert_eq!(&**session.knowledge() as *const _, &*knowledge as *const _);
+		assert_eq!(&**leaf_work.knowledge() as *const _, &*knowledge as *const _);
 		assert!(new_key.is_none());
 	}
 }