Track received attestations from recent blocks (#337)

* record attestations in storage ringbuffer

* remove some reliance on Aura

* fix up test configuration

* extract attestations stuff out to its own module

* add dummy inherent

* use double_map

* fix a couple more compilation errors

network/src/gossip/message_routing.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/>.
+
+//! Data structures and synchronous logic for ICMP message gossip.
+
+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>, // topic -> queue-root
+}
+
+impl View {
+	/// Update the set of current leaves.
+	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 = {
+			let mut new = LeavesVec::new();
+			for leaf in new_leaves {
+				new.push(leaf.clone());
+			}
+
+			std::mem::replace(&mut self.leaves, new)
+		};
+
+		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.insert(topic, queue_root);
+			});
+
+			if r.is_err() {
+				res = r;
+			}
+
+			self.leaf_topics.insert(*new_leaf, this_leaf_topics);
+		}
+
+		res
+	}
+
+	/// Validate an incoming message queue against this view.
+	pub fn validate_queue(&self, messages: &super::GossipParachainMessages)
+		-> (GossipValidationResult<Hash>, i32)
+	{
+		let ostensible_topic = queue_topic(messages.queue_root);
+		if !self.is_topic_live(&ostensible_topic) {
+			(GossipValidationResult::Discard, super::cost::UNNEEDED_ICMP_MESSAGES)
+		} else if !messages.queue_root_is_correct() {
+			(
+				GossipValidationResult::Discard,
+				super::cost::icmp_messages_root_mismatch(messages.messages.len()),
+			)
+		} else {
+			(
+				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
+	}
+}
+
+#[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())
+		)
+	}
+
+	fn check_roots(view: &View, i: u8, max: u8) -> bool {
+		for j in 0..max {
+			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(),
+				};
+
+				match view.validate_queue(&messages).0 {
+					GossipValidationResult::ProcessAndKeep(topic) => if topic != queue_topic(queue_root) {
+						return false
+					},
+					_ => return false,
+				}
+			}
+		}
+
+		true
+	}
+
+	#[test]
+	fn update_leaves_none_in_common() {
+		let mut ctx = TestChainContext::default();
+		let max = 5;
+
+		for i in 0..max {
+			ctx.known_map.insert(hash(i as u8), Known::Leaf);
+
+			let messages_out: Vec<_> = (0..max).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)].iter().cloned(),
+		).unwrap();
+
+		assert!(check_roots(&view, 0, max));
+		assert!(check_roots(&view, 1, max));
+
+		assert!(!check_roots(&view, 2, max));
+		assert!(!check_roots(&view, 3, max));
+		assert!(!check_roots(&view, 4, max));
+		assert!(!check_roots(&view, 5, max));
+
+		view.update_leaves(
+			&ctx,
+			[hash(2), hash(3), hash(4)].iter().cloned(),
+		).unwrap();
+
+		assert!(!check_roots(&view, 0, max));
+		assert!(!check_roots(&view, 1, max));
+
+		assert!(check_roots(&view, 2, max));
+		assert!(check_roots(&view, 3, max));
+		assert!(check_roots(&view, 4, max));
+
+		assert!(!check_roots(&view, 5, max));
+	}
+
+	#[test]
+	fn update_leaves_overlapping() {
+		let mut ctx = TestChainContext::default();
+		let max = 5;
+
+		for i in 0..max {
+			ctx.known_map.insert(hash(i as u8), Known::Leaf);
+
+			let messages_out: Vec<_> = (0..max).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();
+
+		view.update_leaves(
+			&ctx,
+			[hash(2), hash(3), hash(4)].iter().cloned(),
+		).unwrap();
+
+		assert!(!check_roots(&view, 0, max));
+		assert!(!check_roots(&view, 1, max));
+
+		assert!(check_roots(&view, 2, max));
+		assert!(check_roots(&view, 3, max));
+		assert!(check_roots(&view, 4, max));
+
+		assert!(!check_roots(&view, 5, max));
+	}
+}

primitives/src/lib.rs

@@ -76,6 +76,13 @@ pub type AuraPair = ed25519::Pair;
 /// The Ed25519 pub key of an session that belongs to an Aura authority of the chain.
 pub type AuraId = ed25519::Public;
 
+/// The Parachain crypto scheme defined via the keypair type.
+#[cfg(feature = "std")]
+pub type ParachainPair = ed25519::Pair;
+
+/// The Ed25519 public key used to authenticate signatures on parachain data.
+pub type ParachainPublic = ed25519::Public;
+
 /// Header type.
 pub type Header = generic::Header<BlockNumber, BlakeTwo256>;
 /// Block type.

runtime/src/attestations.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/>.
+
+//! A module for tracking all attestations that fell on a given candidate receipt.
+//!
+//! In the future, it is planned that this module will handle dispute resolution
+//! as well.
+
+use rstd::prelude::*;
+use parity_codec::{Encode, Decode};
+use srml_support::{decl_storage, decl_module, ensure};
+
+use primitives::{Hash, parachain::{AttestedCandidate, CandidateReceipt, Id as ParaId}};
+use {system, session::{self, SessionIndex}};
+use srml_support::{
+	StorageValue, StorageMap, StorageDoubleMap, dispatch::Result, traits::Get,
+};
+
+use inherents::{ProvideInherent, InherentData, RuntimeString, MakeFatalError, InherentIdentifier};
+use system::ensure_none;
+
+/// Parachain blocks included in a recent relay-chain block.
+#[derive(Encode, Decode)]
+pub struct IncludedBlocks<T: Trait> {
+	/// The actual relay chain block number where blocks were included.
+	pub actual_number: T::BlockNumber,
+	/// The session index at this block.
+	pub session: SessionIndex,
+	/// The randomness seed at this block.
+	pub random_seed: [u8; 32],
+	/// All parachain IDs active at this block.
+	pub active_parachains: Vec<ParaId>,
+	/// Hashes of the parachain candidates included at this block.
+	pub para_blocks: Vec<Hash>,
+}
+
+/// Attestations kept over time on a parachain block.
+#[derive(Encode, Decode)]
+pub struct BlockAttestations<T: Trait> {
+	receipt: CandidateReceipt,
+	valid: Vec<T::AccountId>, // stash account ID of voter.
+	invalid: Vec<T::AccountId>, // stash account ID of voter.
+}
+
+/// Additional attestations on a parachain block, after it was included.
+#[derive(Encode, Decode, Clone, PartialEq)]
+#[cfg_attr(feature = "std", derive(Debug))]
+pub struct MoreAttestations;
+
+pub trait Trait: session::Trait {
+	/// How many blocks ago we're willing to accept attestations for.
+	type AttestationPeriod: Get<Self::BlockNumber>;
+
+	/// Get a list of the validators' underlying identities.
+	type ValidatorIdentities: Get<Vec<Self::AccountId>>;
+}
+
+decl_storage! {
+	trait Store for Module<T: Trait> as Attestations {
+		/// A mapping from modular block number (n % AttestationPeriod)
+		/// to session index and the list of candidate hashes.
+		pub RecentParaBlocks: map T::BlockNumber => Option<IncludedBlocks<T>>;
+
+		/// Attestations on a recent parachain block.
+		pub ParaBlockAttestations: double_map T::BlockNumber, blake2_128(Hash) => Option<BlockAttestations<T>>;
+
+		// Did we already have more attestations included in this block?
+		DidUpdate: bool;
+	}
+}
+
+decl_module! {
+	/// Parachain-attestations module.
+	pub struct Module<T: Trait> for enum Call where origin: <T as system::Trait>::Origin {
+		/// Provide candidate receipts for parachains, in ascending order by id.
+		fn more_attestations(origin, _more: MoreAttestations) -> Result {
+			ensure_none(origin)?;
+			ensure!(!<DidUpdate>::exists(), "More attestations can be added only once in a block.");
+			<DidUpdate>::put(true);
+
+			Ok(())
+		}
+
+		fn on_finalize(_n: T::BlockNumber) {
+			<DidUpdate>::kill();
+		}
+	}
+}
+
+impl<T: Trait> Module<T> {
+	/// Update recent candidates to contain the already-checked parachain candidates.
+	pub(crate) fn note_included(heads: &[AttestedCandidate], para_blocks: IncludedBlocks<T>) {
+		let attestation_period = T::AttestationPeriod::get();
+		let mod_num = para_blocks.actual_number % attestation_period;
+
+		// clear old entry that was in this place.
+		if let Some(old_entry) = <RecentParaBlocks<T>>::take(&mod_num) {
+			<ParaBlockAttestations<T>>::remove_prefix(&old_entry.actual_number);
+		}
+
+		let validators = T::ValidatorIdentities::get();
+
+		// make new entry.
+		for (head, hash) in heads.iter().zip(&para_blocks.para_blocks) {
+			let mut valid = Vec::new();
+			let invalid = Vec::new();
+
+			for (auth_index, _) in head.validator_indices
+				.iter()
+				.enumerate()
+				.filter(|(_, bit)| *bit)
+			{
+				let stash_id = validators.get(auth_index)
+					.expect("auth_index checked to be within bounds in `check_candidates`; qed")
+					.clone();
+
+				valid.push(stash_id);
+			}
+
+			let summary = BlockAttestations {
+				receipt: head.candidate().clone(),
+				valid,
+				invalid,
+			};
+
+			<ParaBlockAttestations<T>>::insert(&para_blocks.actual_number, hash, &summary);
+		}
+
+		<RecentParaBlocks<T>>::insert(&mod_num, &para_blocks);
+	}
+}
+
+/// An identifier for inherent data that provides after-the-fact attestations
+/// on already included parachain blocks.
+pub const MORE_ATTESTATIONS_IDENTIFIER: InherentIdentifier = *b"par-atts";
+
+pub type InherentType = MoreAttestations;
+
+impl<T: Trait> ProvideInherent for Module<T> {
+	type Call = Call<T>;
+	type Error = MakeFatalError<RuntimeString>;
+	const INHERENT_IDENTIFIER: InherentIdentifier = MORE_ATTESTATIONS_IDENTIFIER;
+
+	fn create_inherent(data: &InherentData) -> Option<Self::Call> {
+		data.get_data::<InherentType>(&MORE_ATTESTATIONS_IDENTIFIER)
+			.ok()
+			.and_then(|x| x.map(Call::more_attestations))
+	}
+}

runtime/src/lib.rs

@@ -14,15 +14,16 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
-//! The Polkadot runtime. This can be compiled with ``#[no_std]`, ready for Wasm.
+//! The Polkadot runtime. This can be compiled with `#[no_std]`, ready for Wasm.
 
 #![cfg_attr(not(feature = "std"), no_std)]
 // `construct_runtime!` does a lot of recursion and requires us to increase the limit to 256.
 #![recursion_limit="256"]
 
+mod attestations;
+mod claims;
 mod curated_grandpa;
 mod parachains;
-mod claims;
 mod slot_range;
 mod slots;
 
@@ -57,7 +58,8 @@ pub use staking::StakerStatus;
 pub use sr_primitives::BuildStorage;
 pub use timestamp::Call as TimestampCall;
 pub use balances::Call as BalancesCall;
-pub use parachains::{Call as ParachainsCall, INHERENT_IDENTIFIER as PARACHAIN_INHERENT_IDENTIFIER};
+pub use attestations::{Call as AttestationsCall, MORE_ATTESTATIONS_IDENTIFIER};
+pub use parachains::{Call as ParachainsCall, NEW_HEADS_IDENTIFIER};
 pub use sr_primitives::{Permill, Perbill};
 pub use srml_support::StorageValue;
 
@@ -244,6 +246,8 @@ parameter_types! {
 	pub const MinimumDeposit: Balance = 100 * BUCKS;
 	pub const EnactmentPeriod: BlockNumber = 30 * 24 * 60 * MINUTES;
 	pub const CooloffPeriod: BlockNumber = 30 * 24 * 60 * MINUTES;
+
+	pub const AttestationPeriod: BlockNumber = 60 * MINUTES * 3;
 }
 
 impl democracy::Trait for Runtime {
@@ -343,6 +347,11 @@ impl finality_tracker::Trait for Runtime {
 	type ReportLatency = ReportLatency;
 }
 
+impl attestations::Trait for Runtime {
+	type AttestationPeriod = AttestationPeriod;
+	type ValidatorIdentities = parachains::ValidatorIdentities<Runtime>;
+}
+
 impl parachains::Trait for Runtime {
 	type Origin = Origin;
 	type Call = Call;
@@ -392,6 +401,7 @@ construct_runtime!(
 		CuratedGrandpa: curated_grandpa::{Module, Call, Config<T>, Storage},
 		Treasury: treasury::{Module, Call, Storage, Event<T>},
 		Parachains: parachains::{Module, Call, Storage, Config<T>, Inherent, Origin},
+		Attestations: attestations::{Module, Call, Storage},
 		Slots: slots::{Module, Call, Storage, Event<T>},
 		Sudo: sudo,
 	}
@@ -481,7 +491,7 @@ impl_runtime_apis! {
 			Aura::authorities()  // only possible as long as parachain validator crypto === aura crypto
 		}
 		fn duty_roster() -> parachain::DutyRoster {
-			Parachains::calculate_duty_roster()
+			Parachains::calculate_duty_roster().0
 		}
 		fn active_parachains() -> Vec<parachain::Id> {
 			Parachains::active_parachains()

runtime/src/parachains.rs

-// Copyright 2017 Parity Technologies (UK) Ltd.
+// Copyright 2017-2019 Parity Technologies (UK) Ltd.
 // This file is part of Polkadot.
 
 // Polkadot is free software: you can redistribute it and/or modify
@@ -18,19 +18,19 @@
 
 use rstd::prelude::*;
 use rstd::collections::btree_map::BTreeMap;
-use parity_codec::{Decode, HasCompact};
+use parity_codec::{Encode, Decode, HasCompact};
 use srml_support::{decl_storage, decl_module, fail, ensure};
 
 use sr_primitives::traits::{Hash as HashT, BlakeTwo256, Member, CheckedConversion, Saturating, One};
 use sr_primitives::weights::SimpleDispatchInfo;
-use primitives::{Hash, Balance, parachain::{
+use primitives::{Hash, Balance, ParachainPublic, parachain::{
 	self, Id as ParaId, Chain, DutyRoster, AttestedCandidate, Statement, AccountIdConversion,
 	ParachainDispatchOrigin, UpwardMessage, BlockIngressRoots,
 }};
 use {system, session};
 use srml_support::{
 	StorageValue, StorageMap, storage::AppendableStorageMap, Parameter, Dispatchable, dispatch::Result,
-	traits::{Currency, WithdrawReason, ExistenceRequirement}
+	traits::{Currency, Get, WithdrawReason, ExistenceRequirement}
 };
 
 #[cfg(feature = "std")]
@@ -45,6 +45,7 @@ use sr_primitives::{StorageOverlay, ChildrenStorageOverlay};
 use rstd::marker::PhantomData;
 
 use system::{ensure_none, ensure_root};
+use crate::attestations::{self, IncludedBlocks};
 
 // ranges for iteration of general block number don't work, so this
 // is a utility to get around that.
@@ -172,7 +173,16 @@ impl<AccountId, T: Currency<AccountId>> ParachainCurrency<AccountId> for T where
 	}
 }
 
-pub trait Trait: session::Trait {
+/// Interface to the persistent (stash) identities of the current validators.
+pub struct ValidatorIdentities<T>(rstd::marker::PhantomData<T>);
+