Implement the Statement Distribution Subsystem (#1326)

* set up data types and control flow for statement distribution

* add some set-like methods to View

* implement sending to peers

* start fixing equivocation handling

* Add a section to the statement distribution subsystem on equivocations and flood protection

* fix typo and amend wording

* implement flood protection

* have peer knowledge tracker follow when peer first learns about a candidate

* send dependents after circulating

* add another TODO

* trigger send in one more place

* refactors from review

* send new statements to candidate backing

* instantiate active head data with runtime API values

* track our view changes and peer view changes

* apply a benefit to peers who send us statements we want

* remove unneeded TODO

* add some comments and improve Hash implementation

* start tests and fix `note_statement`

* test active_head seconding logic

* test that the per-peer tracking logic works

* test per-peer knowledge tracker

* test that peer view updates lead to messages being sent

* test statement circulation

* address review comments

* have view set methods return references

polkadot/node/network/statement-distribution/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "polkadot-statement-distribution"
+version = "0.1.0"
+authors = ["Parity Technologies <admin@parity.io>"]
+description = "Statement Distribution Subsystem"
+edition = "2018"
+
+[dependencies]
+futures = "0.3.5"
+log = "0.4.8"
+futures-timer = "3.0.2"
+streamunordered = "0.5.1"
+polkadot-primitives = { path = "../../../primitives" }
+node-primitives = { package = "polkadot-node-primitives", path = "../../primitives" }
+parity-scale-codec = "1.3.0"
+sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-staking = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
+polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }
+arrayvec = "0.5.1"
+indexmap = "1.4.0"
+
+[dev-dependencies]
+parking_lot = "0.10.0"
+subsystem-test = { package = "polkadot-subsystem-test-helpers", path = "../../test-helpers/subsystem" }
+assert_matches = "1.3.0"
+sp-keyring = { git = "https://github.com/paritytech/substrate", branch = "master" }

polkadot/node/primitives/src/lib.rs

@@ -29,7 +29,7 @@ use polkadot_primitives::{Hash,
 };
 
 /// A statement, where the candidate receipt is included in the `Seconded` variant.
-#[derive(Debug, Clone, PartialEq, Encode, Decode)]
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
 pub enum Statement {
 	/// A statement that a validator seconds a candidate.
 	#[codec(index = "1")]
@@ -42,16 +42,19 @@ pub enum Statement {
 	Invalid(Hash),
 }
 
+impl Statement {
+	pub fn to_compact(&self) -> CompactStatement {
+		match *self {
+			Statement::Seconded(ref c) => CompactStatement::Candidate(c.hash()),
+			Statement::Valid(hash) => CompactStatement::Valid(hash),
+			Statement::Invalid(hash) => CompactStatement::Invalid(hash),
+		}
+	}
+}
+
 impl EncodeAs<CompactStatement> for Statement {
 	fn encode_as(&self) -> Vec<u8> {
-		let statement = match *self {
-			Statement::Seconded(ref c) => {
-				polkadot_primitives::parachain::CompactStatement::Candidate(c.hash())
-			}
-			Statement::Valid(hash) => polkadot_primitives::parachain::CompactStatement::Valid(hash),
-			Statement::Invalid(hash) => polkadot_primitives::parachain::CompactStatement::Invalid(hash),
-		};
-		statement.encode()
+		self.to_compact().encode()
 	}
 }
 
@@ -87,5 +90,22 @@ pub type ProtocolId = [u8; 4];
 /// A succinct representation of a peer's view. This consists of a bounded amount of chain heads.
 ///
 /// Up to `N` (5?) chain heads.
-#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+#[derive(Default, Debug, Clone, PartialEq, Eq, Encode, Decode)]
 pub struct View(pub Vec<Hash>);
+
+impl View {
+	/// Returns an iterator of the hashes present in `Self` but not in `other`.
+	pub fn difference<'a>(&'a self, other: &'a View) -> impl Iterator<Item = &'a Hash> + 'a {
+		self.0.iter().filter(move |h| !other.contains(h))
+	}
+
+	/// An iterator containing hashes present in both `Self` and in `other`.
+	pub fn intersection<'a>(&'a self, other: &'a View) -> impl Iterator<Item = &'a Hash> + 'a {
+		self.0.iter().filter(move |h| other.contains(h))
+	}
+
+	/// Whether the view contains a given hash.
+	pub fn contains(&self, hash: &Hash) -> bool {
+		self.0.contains(hash)
+	}
+}

polkadot/node/subsystem/src/messages.rs

@@ -24,7 +24,6 @@
 
 use futures::channel::{mpsc, oneshot};
 
-use sc_network::{ObservedRole, ReputationChange, PeerId};
 use polkadot_primitives::{BlockNumber, Hash, Signature};
 use polkadot_primitives::parachain::{
 	AbridgedCandidateReceipt, PoVBlock, ErasureChunk, BackedCandidate, Id as ParaId,
@@ -34,6 +33,8 @@ use polkadot_node_primitives::{
 	MisbehaviorReport, SignedFullStatement, View, ProtocolId,
 };
 
+pub use sc_network::{ObservedRole, ReputationChange, PeerId};
+
 /// A notification of a new backed candidate.
 #[derive(Debug)]
 pub struct NewBackedCandidate(pub BackedCandidate);
@@ -234,4 +235,6 @@ pub enum AllMessages {
 	RuntimeApi(RuntimeApiMessage),
 	/// Message for the availability store subsystem.
 	AvailabilityStore(AvailabilityStoreMessage),
+	/// Message for the network bridge subsystem.
+	NetworkBridge(NetworkBridgeMessage),
 }