introduce errors with info (#1834)

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

@@ -7,26 +7,22 @@ edition = "2018"
 [dependencies]
 futures = "0.3.5"
 log = "0.4.11"
-streamunordered = "0.5.1"
 codec = { package="parity-scale-codec", version = "1.3.4", features = ["std"]  }
-derive_more = "0.99.9"
 polkadot-primitives = { path = "../../../primitives" }
 polkadot-erasure-coding = { path = "../../../erasure-coding" }
 polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }
-polkadot-network-bridge = { path = "../../network/bridge" }
 polkadot-node-network-protocol = { path = "../../network/protocol" }
 polkadot-node-subsystem-util = { path = "../../subsystem-util" }
 sp-core = { git = "https://github.com/paritytech/substrate", branch = "master", features = ["std"]  }
 sp-keystore = { git = "https://github.com/paritytech/substrate", branch = "master" }
+thiserror = "1.0.21"
 
 [dev-dependencies]
 polkadot-subsystem-testhelpers = { package = "polkadot-node-subsystem-test-helpers", path = "../../subsystem-test-helpers" }
-bitvec = { version = "0.17.4", default-features = false, features = ["alloc"] }
 sp-core = { git = "https://github.com/paritytech/substrate", branch = "master", features = ["std"] }
 sp-application-crypto = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-keyring = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-keystore = { git = "https://github.com/paritytech/substrate", branch = "master" }
-parking_lot = "0.11.0"
 futures-timer = "3.0.2"
 env_logger = "0.7.1"
 assert_matches = "1.3.0"

polkadot/node/network/availability-distribution/src/lib.rs

@@ -22,56 +22,99 @@
 //! peers. Verified in this context means, the erasure chunks contained merkle proof
 //! is checked.
 
+#![deny(unused_crate_dependencies, unused_qualifications)]
+
 use codec::{Decode, Encode};
-use futures::{channel::oneshot, FutureExt};
+use futures::{channel::oneshot, FutureExt, TryFutureExt};
 
 use sp_core::crypto::Public;
 use sp_keystore::{CryptoStore, SyncCryptoStorePtr};
 
 use log::{trace, warn};
 use polkadot_erasure_coding::branch_hash;
+use polkadot_node_network_protocol::{
+	v1 as protocol_v1, NetworkBridgeEvent, PeerId, ReputationChange as Rep, View,
+};
+use polkadot_node_subsystem_util::metrics::{self, prometheus};
 use polkadot_primitives::v1::{
-	PARACHAIN_KEY_TYPE_ID,
-	BlakeTwo256, CommittedCandidateReceipt, CoreState, ErasureChunk,
-	Hash as Hash, HashT, Id as ParaId,
-	ValidatorId, ValidatorIndex, SessionIndex,
+	BlakeTwo256, CommittedCandidateReceipt, CoreState, ErasureChunk, Hash, HashT, Id as ParaId,
+	SessionIndex, ValidatorId, ValidatorIndex, PARACHAIN_KEY_TYPE_ID,
 };
 use polkadot_subsystem::messages::{
-	AllMessages, AvailabilityDistributionMessage, NetworkBridgeMessage, RuntimeApiMessage,
-	RuntimeApiRequest, AvailabilityStoreMessage, ChainApiMessage,
+	AllMessages, AvailabilityDistributionMessage, AvailabilityStoreMessage, ChainApiMessage,
+	NetworkBridgeMessage, RuntimeApiMessage, RuntimeApiRequest,
 };
 use polkadot_subsystem::{
 	errors::{ChainApiError, RuntimeApiError},
 	ActiveLeavesUpdate, FromOverseer, OverseerSignal, SpawnedSubsystem, Subsystem,
 	SubsystemContext, SubsystemError,
 };
-use polkadot_node_subsystem_util::{
-	metrics::{self, prometheus},
-};
-use polkadot_node_network_protocol::{
-	v1 as protocol_v1, View, ReputationChange as Rep, PeerId,
-	NetworkBridgeEvent,
-};
 use std::collections::{HashMap, HashSet};
-use std::io;
 use std::iter;
+use thiserror::Error;
 
 const TARGET: &'static str = "avad";
 
-#[derive(Debug, derive_more::From)]
+#[derive(Debug, Error)]
 enum Error {
-	#[from]
-	Erasure(polkadot_erasure_coding::Error),
-	#[from]
-	Io(io::Error),
-	#[from]
-	Oneshot(oneshot::Canceled),
-	#[from]
-	Subsystem(SubsystemError),
-	#[from]
-	RuntimeApi(RuntimeApiError),
-	#[from]
-	ChainApi(ChainApiError),
+	#[error("Sending PendingAvailability query failed")]
+	QueryPendingAvailabilitySendQuery(#[source] SubsystemError),
+	#[error("Response channel to obtain PendingAvailability failed")]
+	QueryPendingAvailabilityResponseChannel(#[source] oneshot::Canceled),
+	#[error("RuntimeAPI to obtain PendingAvailability failed")]
+	QueryPendingAvailability(#[source] RuntimeApiError),
+
+	#[error("Sending StoreChunk query failed")]
+	StoreChunkSendQuery(#[source] SubsystemError),
+	#[error("Response channel to obtain StoreChunk failed")]
+	StoreChunkResponseChannel(#[source] oneshot::Canceled),
+
+	#[error("Sending QueryChunk query failed")]
+	QueryChunkSendQuery(#[source] SubsystemError),
+	#[error("Response channel to obtain QueryChunk failed")]
+	QueryChunkResponseChannel(#[source] oneshot::Canceled),
+
+	#[error("Sending QueryAncestors query failed")]
+	QueryAncestorsSendQuery(#[source] SubsystemError),
+	#[error("Response channel to obtain QueryAncestors failed")]
+	QueryAncestorsResponseChannel(#[source] oneshot::Canceled),
+	#[error("RuntimeAPI to obtain QueryAncestors failed")]
+	QueryAncestors(#[source] ChainApiError),
+
+	#[error("Sending QuerySession query failed")]
+	QuerySessionSendQuery(#[source] SubsystemError),
+	#[error("Response channel to obtain QuerySession failed")]
+	QuerySessionResponseChannel(#[source] oneshot::Canceled),
+	#[error("RuntimeAPI to obtain QuerySession failed")]
+	QuerySession(#[source] RuntimeApiError),
+
+	#[error("Sending QueryValidators query failed")]
+	QueryValidatorsSendQuery(#[source] SubsystemError),
+	#[error("Response channel to obtain QueryValidators failed")]
+	QueryValidatorsResponseChannel(#[source] oneshot::Canceled),
+	#[error("RuntimeAPI to obtain QueryValidators failed")]
+	QueryValidators(#[source] RuntimeApiError),
+
+	#[error("Sending AvailabilityCores query failed")]
+	AvailabilityCoresSendQuery(#[source] SubsystemError),
+	#[error("Response channel to obtain AvailabilityCores failed")]
+	AvailabilityCoresResponseChannel(#[source] oneshot::Canceled),
+	#[error("RuntimeAPI to obtain AvailabilityCores failed")]
+	AvailabilityCores(#[source] RuntimeApiError),
+
+	#[error("Sending AvailabilityCores query failed")]
+	QueryAvailabilitySendQuery(#[source] SubsystemError),
+	#[error("Response channel to obtain AvailabilityCores failed")]
+	QueryAvailabilityResponseChannel(#[source] oneshot::Canceled),
+
+	#[error("Sending out a peer report message")]
+	ReportPeerMessageSend(#[source] SubsystemError),
+
+	#[error("Sending a gossip message")]
+	TrackedGossipMessage(#[source] SubsystemError),
+
+	#[error("Receive channel closed")]
+	IncomingMessageChannel(#[source] SubsystemError),
 }
 
 type Result<T> = std::result::Result<T, Error>;
@@ -199,22 +242,18 @@ impl ProtocolState {
 	where
 		Context: SubsystemContext<Message = AvailabilityDistributionMessage>,
 	{
-		let candidates =
-			query_live_candidates(ctx, self, std::iter::once(relay_parent)).await?;
+		let candidates = query_live_candidates(ctx, self, std::iter::once(relay_parent)).await?;
 
 		// register the relation of relay_parent to candidate..
 		// ..and the reverse association.
 		for (relay_parent_or_ancestor, (receipt_hash, receipt)) in candidates.clone() {
-			self
-				.reverse
+			self.reverse
 				.insert(receipt_hash.clone(), relay_parent_or_ancestor.clone());
-			let per_candidate = self.per_candidate.entry(receipt_hash.clone())
-				.or_default();
+			let per_candidate = self.per_candidate.entry(receipt_hash.clone()).or_default();
 			per_candidate.validator_index = validator_index.clone();
 			per_candidate.validators = validators.clone();
 
-			self
-				.receipts
+			self.receipts
 				.entry(relay_parent_or_ancestor)
 				.or_default()
 				.insert((receipt_hash, receipt));
@@ -240,8 +279,7 @@ impl ProtocolState {
 				.insert(relay_parent);
 		}
 
-		self
-			.per_relay_parent
+		self.per_relay_parent
 			.entry(relay_parent)
 			.or_default()
 			.ancestors = ancestors;
@@ -314,17 +352,21 @@ where
 		}
 		NetworkBridgeEvent::PeerMessage(remote, msg) => {
 			let gossiped_availability = match msg {
-				protocol_v1::AvailabilityDistributionMessage::Chunk(candidate_hash, chunk) =>
-					AvailabilityGossipMessage { candidate_hash, erasure_chunk: chunk }
+				protocol_v1::AvailabilityDistributionMessage::Chunk(candidate_hash, chunk) => {
+					AvailabilityGossipMessage {
+						candidate_hash,
+						erasure_chunk: chunk,
+					}
+				}
 			};
 
-			process_incoming_peer_message(ctx, state, remote, gossiped_availability, metrics).await?;
+			process_incoming_peer_message(ctx, state, remote, gossiped_availability, metrics)
+				.await?;
 		}
 	}
 	Ok(())
 }
 
-
 /// Handle the changes necessary when our view changes.
 async fn handle_our_view_change<Context>(
 	ctx: &mut Context,
@@ -346,19 +388,15 @@ where
 	for added in added.iter() {
 		let added = **added;
 		let validators = query_validators(ctx, added).await?;
-		let validator_index = obtain_our_validator_index(
-			&validators,
-			keystore.clone(),
-		).await;
-		state.add_relay_parent(ctx, added, validators, validator_index).await?;
+		let validator_index = obtain_our_validator_index(&validators, keystore.clone()).await;
+		state
+			.add_relay_parent(ctx, added, validators, validator_index)
+			.await?;
 	}
 
 	// handle all candidates
 	for (candidate_hash, _receipt) in state.cached_live_candidates_unioned(added) {
-		let per_candidate = state
-			.per_candidate
-			.entry(candidate_hash)
-			.or_default();
+		let per_candidate = state.per_candidate.entry(candidate_hash).or_default();
 
 		// assure the node has the validator role
 		if per_candidate.validator_index.is_none() {
@@ -388,19 +426,18 @@ where
 
 		// distribute all erasure messages to interested peers
 		for chunk_index in 0u32..(validator_count as u32) {
-
 			// only the peers which did not receive this particular erasure chunk
-			let per_candidate = state
-				.per_candidate
-				.entry(candidate_hash)
-				.or_default();
+			let per_candidate = state.per_candidate.entry(candidate_hash).or_default();
 
 			// obtain the chunks from the cache, if not fallback
 			// and query the availability store
 			let message_id = (candidate_hash, chunk_index);
-			let erasure_chunk = if let Some(message) = per_candidate.message_vault.get(&chunk_index) {
+			let erasure_chunk = if let Some(message) = per_candidate.message_vault.get(&chunk_index)
+			{
 				message.erasure_chunk.clone()
-			} else if let Some(erasure_chunk) = query_chunk(ctx, candidate_hash, chunk_index as ValidatorIndex).await? {
+			} else if let Some(erasure_chunk) =
+				query_chunk(ctx, candidate_hash, chunk_index as ValidatorIndex).await?
+			{
 				erasure_chunk
 			} else {
 				continue;
@@ -415,9 +452,7 @@ where
 					!per_candidate
 						.sent_messages
 						.get(*peer)
-						.filter(|set| {
-							set.contains(&message_id)
-						})
+						.filter(|set| set.contains(&message_id))
 						.is_some()
 				})
 				.map(|peer| peer.clone())
@@ -427,7 +462,8 @@ where
 				erasure_chunk,
 			};
 
-			send_tracked_gossip_message_to_peers(ctx, per_candidate, metrics, peers, message).await?;
+			send_tracked_gossip_message_to_peers(ctx, per_candidate, metrics, peers, message)
+				.await?;
 		}
 	}
 
@@ -450,7 +486,8 @@ async fn send_tracked_gossip_message_to_peers<Context>(
 where
 	Context: SubsystemContext<Message = AvailabilityDistributionMessage>,
 {
-	send_tracked_gossip_messages_to_peers(ctx, per_candidate, metrics, peers, iter::once(message)).await
+	send_tracked_gossip_messages_to_peers(ctx, per_candidate, metrics, peers, iter::once(message))
+		.await
 }
 
 #[inline(always)]
@@ -464,7 +501,8 @@ async fn send_tracked_gossip_messages_to_peer<Context>(
 where
 	Context: SubsystemContext<Message = AvailabilityDistributionMessage>,
 {
-	send_tracked_gossip_messages_to_peers(ctx, per_candidate, metrics, vec![peer], message_iter).await
+	send_tracked_gossip_messages_to_peers(ctx, per_candidate, metrics, vec![peer], message_iter)
+		.await
 }
 
 async fn send_tracked_gossip_messages_to_peers<Context>(
@@ -478,7 +516,7 @@ where
 	Context: SubsystemContext<Message = AvailabilityDistributionMessage>,
 {
 	if peers.is_empty() {
-		return Ok(())
+		return Ok(());
 	}
 	for message in message_iter {
 		for peer in peers.iter() {
@@ -506,7 +544,7 @@ where
 			),
 		))
 		.await
-		.map_err::<Error, _>(Into::into)?;
+		.map_err(|e| Error::TrackedGossipMessage(e))?;
 
 		metrics.on_chunk_distributed();
 	}
@@ -543,8 +581,7 @@ where
 		let per_candidate = state.per_candidate.entry(candidate_hash).or_default();
 
 		// obtain the relevant chunk indices not sent yet
-		let messages = ((0 as ValidatorIndex)
-			..(per_candidate.validators.len() as ValidatorIndex))
+		let messages = ((0 as ValidatorIndex)..(per_candidate.validators.len() as ValidatorIndex))
 			.into_iter()
 			.filter_map(|erasure_chunk_index: ValidatorIndex| {
 				let message_id = (candidate_hash, erasure_chunk_index);
@@ -567,7 +604,8 @@ where
 			.cloned()
 			.collect::<HashSet<_>>();
 
-		send_tracked_gossip_messages_to_peer(ctx, per_candidate, metrics, origin.clone(), messages).await?;
+		send_tracked_gossip_messages_to_peer(ctx, per_candidate, metrics, origin.clone(), messages)
+			.await?;
 	}
 	Ok(())
 }
@@ -580,8 +618,13 @@ async fn obtain_our_validator_index(
 	keystore: SyncCryptoStorePtr,
 ) -> Option<ValidatorIndex> {
 	for (idx, validator) in validators.iter().enumerate() {
-		if CryptoStore::has_keys(&*keystore, &[(validator.to_raw_vec(), PARACHAIN_KEY_TYPE_ID)]).await {
-			return Some(idx as ValidatorIndex)
+		if CryptoStore::has_keys(
+			&*keystore,
+			&[(validator.to_raw_vec(), PARACHAIN_KEY_TYPE_ID)],
+		)
+		.await
+		{
+			return Some(idx as ValidatorIndex);
 		}
 	}
 	None
@@ -664,8 +707,13 @@ where
 						live_candidate.descriptor.relay_parent.clone(),
 						message.erasure_chunk.index,
 						message.erasure_chunk.clone(),
-					).await? {
-						warn!(target: TARGET, "Failed to store erasure chunk to availability store");
+					)
+					.await?
+					{
+						warn!(
+							target: TARGET,
+							"Failed to store erasure chunk to availability store"
+						);
 					}
 				}
 			}
@@ -729,7 +777,10 @@ impl AvailabilityDistributionSubsystem {
 		// work: process incoming messages from the overseer.
 		let mut state = ProtocolState::default();
 		loop {
-			let message = ctx.recv().await.map_err::<Error, _>(Into::into)?;
+			let message = ctx
+				.recv()
+				.await
+				.map_err(|e| Error::IncomingMessageChannel(e))?;
 			match message {
 				FromOverseer::Communication {
 					msg: AvailabilityDistributionMessage::NetworkBridgeUpdateV1(event),
@@ -740,7 +791,9 @@ impl AvailabilityDistributionSubsystem {
 						&mut state,
 						&self.metrics,
 						event,
-					).await {
+					)
+					.await
+					{
 						warn!(
 							target: TARGET,
 							"Failed to handle incomming network messages: {:?}", e
@@ -767,9 +820,15 @@ where
 	Context: SubsystemContext<Message = AvailabilityDistributionMessage> + Sync + Send,
 {
 	fn start(self, ctx: Context) -> SpawnedSubsystem {
+		let future = self
+			.run(ctx)
+			.map_err(|e| SubsystemError::with_origin("availability-distribution", e))
+			.map(|_| ())
+			.boxed();
+
 		SpawnedSubsystem {
 			name: "availability-distribution-subsystem",
-			future: Box::pin(async move { self.run(ctx) }.map(|_| ())),
+			future,
 		}
 	}
 }
@@ -816,7 +875,6 @@ where
 		HashMap::<Hash, (Hash, CommittedCandidateReceipt)>::with_capacity(capacity);
 
 	for relay_parent in iter {
-
 		// register one of relay parents (not the ancestors)
 		let mut ancestors = query_up_to_k_ancestors_in_same_session(
 			ctx,
@@ -827,7 +885,6 @@ where
 
 		ancestors.push(relay_parent);
 
-
 		// ancestors might overlap, so check the cache too
 		let unknown = ancestors
 			.into_iter()
@@ -841,10 +898,7 @@ where
 						// directly extend the live_candidates with the cached value
 						live_candidates.extend(receipts.into_iter().map(
 							|(receipt_hash, receipt)| {
-								(
-									relay_parent,
-									(receipt_hash.clone(), receipt.clone()),
-								)
+								(relay_parent, (receipt_hash.clone(), receipt.clone()))
 							},
 						));
 						Some(())
@@ -877,10 +931,12 @@ where
 		RuntimeApiRequest::AvailabilityCores(tx),
 	)))
 	.await
-	.map_err::<Error, _>(Into::into)?;
+	.map_err(|e| Error::AvailabilityCoresSendQuery(e))?;
 
 	let all_para_ids: Vec<_> = rx
-		.await??;
+		.await
+		.map_err(|e| Error::AvailabilityCoresResponseChannel(e))?
+		.map_err(|e| Error::AvailabilityCores(e))?;
 
 	let occupied_para_ids = all_para_ids
 		.into_iter()
@@ -910,14 +966,11 @@ where
 		NetworkBridgeMessage::ReportPeer(peer, rep),
 	))
 	.await
-	.map_err::<Error, _>(Into::into)
+	.map_err(|e| Error::ReportPeerMessageSend(e))
 }
 
 /// Query the proof of validity for a particular candidate hash.
-async fn query_data_availability<Context>(
-	ctx: &mut Context,
-	candidate_hash: Hash,
-) -> Result<bool>
+async fn query_data_availability<Context>(ctx: &mut Context, candidate_hash: Hash) -> Result<bool>
 where
 	Context: SubsystemContext<Message = AvailabilityDistributionMessage>,
 {
@@ -925,11 +978,12 @@ where
 	ctx.send_message(AllMessages::AvailabilityStore(
 		AvailabilityStoreMessage::QueryDataAvailability(candidate_hash, tx),
 	))
-	.await?;
-	rx.await.map_err::<Error, _>(Into::into)
+	.await
+	.map_err(|e| Error::QueryAvailabilitySendQuery(e))?;
+	rx.await
+		.map_err(|e| Error::QueryAvailabilityResponseChannel(e))
 }
 
-
 async fn query_chunk<Context>(
 	ctx: &mut Context,
 	candidate_hash: Hash,
@@ -940,13 +994,13 @@ where
 {
 	let (tx, rx) = oneshot::channel();
 	ctx.send_message(AllMessages::AvailabilityStore(
-			AvailabilityStoreMessage::QueryChunk(candidate_hash, validator_index, tx),
-		))
-		.await?;
-	rx.await.map_err::<Error, _>(Into::into)
+		AvailabilityStoreMessage::QueryChunk(candidate_hash, validator_index, tx),
+	))
+	.await
+	.map_err(|e| Error::QueryChunkSendQuery(e))?;
+	rx.await.map_err(|e| Error::QueryChunkResponseChannel(e))
 }
 
-
 async fn store_chunk<Context>(
 	ctx: &mut Context,
 	candidate_hash: Hash,
@@ -958,16 +1012,19 @@ where
 	Context: SubsystemContext<Message = AvailabilityDistributionMessage>,
 {
 	let (tx, rx) = oneshot::channel();
-	ctx.send_message(AllMessages::AvailabilityStore(
-		AvailabilityStoreMessage::StoreChunk {
-			candidate_hash,
-			relay_parent,
-			validator_index,
-			chunk: erasure_chunk,
-			tx,
-		}
-	)).await?;
-	rx.await.map_err::<Error, _>(Into::into)
+	ctx.send_message(
+        AllMessages::AvailabilityStore(
+                AvailabilityStoreMessage::StoreChunk {
+                candidate_hash,
+                relay_parent,
+                validator_index,
+                chunk: erasure_chunk,
+                tx,
+            }
+        )).await
+        .map_err(|e| Error::StoreChunkSendQuery(e))?;
+    
+    rx.await.map_err(|e| Error::StoreChunkResponseChannel(e))
 }
 
 /// Request the head data for a particular para.
@@ -981,12 +1038,15 @@ where
 {
 	let (tx, rx) = oneshot::channel();
 	ctx.send_message(AllMessages::RuntimeApi(RuntimeApiMessage::Request(
-			relay_parent,
-			RuntimeApiRequest::CandidatePendingAvailability(para, tx),
-		)))
-		.await?;
-	rx.await?
-		.map_err::<Error, _>(Into::into)
+		relay_parent,
+		RuntimeApiRequest::CandidatePendingAvailability(para, tx),
+	)))
+	.await
+	.map_err(|e| Error::QueryPendingAvailabilitySendQuery(e))?;
+
+	rx.await
+		.map_err(|e| Error::QueryPendingAvailabilityResponseChannel(e))?
+		.map_err(|e| Error::QueryPendingAvailability(e))
 }
 
 /// Query the validator set.
@@ -1004,9 +1064,11 @@ where
 	));
 
 	ctx.send_message(query_validators)
-		.await?;
-	rx.await?
-		.map_err::<Error, _>(Into::into)
+		.await
+		.map_err(|e| Error::QueryValidatorsSendQuery(e))?;
+	rx.await
+		.map_err(|e| Error::QueryValidatorsResponseChannel(e))?
+		.map_err(|e| Error::QueryValidators(e))
 }
 
 /// Query the hash of the `K` ancestors
@@ -1026,9 +1088,11 @@ where
 	});
 
 	ctx.send_message(query_ancestors)
-		.await?;
-	rx.await?
-		.map_err::<Error, _>(Into::into)
+		.await
+		.map_err(|e| Error::QueryAncestorsSendQuery(e))?;
+	rx.await
+		.map_err(|e| Error::QueryAncestorsResponseChannel(e))?
+		.map_err(|e| Error::QueryAncestors(e))
 }
 
 /// Query the session index of a relay parent
@@ -1046,9 +1110,11 @@ where
 	));
 
 	ctx.send_message(query_session_idx_for_child)
-		.await?;
-	rx.await?
-		.map_err::<Error, _>(Into::into)
+		.await
+		.map_err(|e| Error::QuerySessionSendQuery(e))?;
+	rx.await
+		.map_err(|e| Error::QuerySessionResponseChannel(e))?
+		.map_err(|e| Error::QuerySession(e))
 }
 
 /// Queries up to k ancestors with the constraints of equiv session
@@ -1089,7 +1155,6 @@ where
 	Ok(acc)
 }
 
-
 #[derive(Clone)]
 struct MetricsInner {
 	gossipped_availability_chunks: prometheus::Counter<prometheus::U64>,
@@ -1108,12 +1173,14 @@ impl Metrics {
 }
 
 impl metrics::Metrics for Metrics {
-	fn try_register(registry: &prometheus::Registry) -> std::result::Result<Self, prometheus::PrometheusError> {
+	fn try_register(
+		registry: &prometheus::Registry,
+	) -> std::result::Result<Self, prometheus::PrometheusError> {
 		let metrics = MetricsInner {
 			gossipped_availability_chunks: prometheus::register(
 				prometheus::Counter::new(
 					"parachain_gossipped_availability_chunks_total",
-					"Number of availability chunks gossipped to other peers."
+					"Number of availability chunks gossipped to other peers.",
 				)?,
 				registry,
 			)?,

polkadot/node/network/availability-distribution/src/tests.rs

@@ -17,22 +17,21 @@
 use super::*;
 use assert_matches::assert_matches;
 use polkadot_erasure_coding::{branches, obtain_chunks_v1 as obtain_chunks};
+use polkadot_node_network_protocol::ObservedRole;
+use polkadot_node_subsystem_util::TimeoutExt;
 use polkadot_primitives::v1::{
 	AvailableData, BlockData, CandidateCommitments, CandidateDescriptor, GroupIndex,
-	GroupRotationInfo, HeadData, PersistedValidationData, OccupiedCore,
-	PoV, ScheduledCore,
+	GroupRotationInfo, HeadData, OccupiedCore, PersistedValidationData, PoV, ScheduledCore,
 };
 use polkadot_subsystem_testhelpers::{self as test_helpers};
-use polkadot_node_subsystem_util::TimeoutExt;
-use polkadot_node_network_protocol::ObservedRole;
 
 use futures::{executor, future, Future};
 use futures_timer::Delay;
-use smallvec::smallvec;
-use std::{sync::Arc, time::Duration};
 use sc_keystore::LocalKeystore;
-use sp_keystore::{SyncCryptoStorePtr, SyncCryptoStore};
+use smallvec::smallvec;
 use sp_application_crypto::AppKey;
+use sp_keystore::{SyncCryptoStore, SyncCryptoStorePtr};
+use std::{sync::Arc, time::Duration};
 
 macro_rules! view {
 		( $( $hash:expr ),* $(,)? ) => [
@@ -46,9 +45,9 @@ macro_rules! delay {
 	};
 }
 
-fn chunk_protocol_message(message: AvailabilityGossipMessage)
-	-> protocol_v1::AvailabilityDistributionMessage
-{
+fn chunk_protocol_message(
+	message: AvailabilityGossipMessage,
+) -> protocol_v1::AvailabilityDistributionMessage {
 	protocol_v1::AvailabilityDistributionMessage::Chunk(
 		message.candidate_hash,
 		message.erasure_chunk,
@@ -175,8 +174,12 @@ impl Default for TestState {
 
 		let keystore: SyncCryptoStorePtr = Arc::new(LocalKeystore::in_memory());
 
-		SyncCryptoStore::sr25519_generate_new(&*keystore, ValidatorId::ID, Some(&validators[0].to_seed()))
-			.expect("Insert key into keystore");
+		SyncCryptoStore::sr25519_generate_new(
+			&*keystore,
+			ValidatorId::ID,
+			Some(&validators[0].to_seed()),
+		)
+		.expect("Insert key into keystore");
 
 		let validator_public = validator_pubkeys(&validators);
 
@@ -867,10 +870,7 @@ fn reputation_verification() {
 			overseer_send(
 				&mut virtual_overseer,
 				AvailabilityDistributionMessage::NetworkBridgeUpdateV1(
-					NetworkBridgeEvent::PeerMessage(
-						peer_a.clone(),
-						chunk_protocol_message(valid2),
-					),
+					NetworkBridgeEvent::PeerMessage(peer_a.clone(), chunk_protocol_message(valid2)),
 				),
 			)
 			.await;

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

@@ -6,17 +6,12 @@ edition = "2018"
 
 [dependencies]
 futures = "0.3.5"
-futures-timer = "3.0.2"
-log = "0.4.8"
-streamunordered = "0.5.1"
+log = "0.4.11"
 codec = { package="parity-scale-codec", version = "1.3.4" }
-node-primitives = { package = "polkadot-node-primitives", path = "../../primitives" }
 polkadot-primitives = { path = "../../../primitives" }
 polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }
 polkadot-node-subsystem-util = { path = "../../subsystem-util" }
-polkadot-network-bridge = { path = "../../network/bridge" }
 polkadot-node-network-protocol = { path = "../../network/protocol" }
-sc-network = { git = "https://github.com/paritytech/substrate", branch = "master" }
 
 [dev-dependencies]
 polkadot-node-subsystem-test-helpers = { path = "../../subsystem-test-helpers" }
@@ -25,9 +20,7 @@ sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-application-crypto = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-keystore = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-keystore = { git = "https://github.com/paritytech/substrate", branch = "master" }
-parking_lot = "0.11.0"
 maplit = "1.0.2"
-smol = "0.3.3"
 env_logger = "0.7.1"
 assert_matches = "1.3.0"
 tempfile = "3.1.0"

polkadot/node/network/bitfield-distribution/src/lib.rs

@@ -20,8 +20,10 @@
 //! for a particular relay parent.
 //! Independently of that, gossips on received messages from peers to other interested peers.
 
+#![deny(unused_crate_dependencies)]
+
 use codec::{Decode, Encode};
-use futures::{channel::oneshot, FutureExt};
+use futures::{channel::oneshot, FutureExt, TryFutureExt};
 
 use log::{trace, warn};
 use polkadot_subsystem::messages::*;
@@ -33,6 +35,7 @@ use polkadot_node_subsystem_util::{
 };
 use polkadot_primitives::v1::{Hash, SignedAvailabilityBitfield, SigningContext, ValidatorId};
 use polkadot_node_network_protocol::{v1 as protocol_v1, PeerId, NetworkBridgeEvent, View, ReputationChange};
+use polkadot_subsystem::SubsystemError;
 use std::collections::{HashMap, HashSet};
 
 const COST_SIGNATURE_INVALID: ReputationChange =
@@ -578,9 +581,15 @@ where
 	C: SubsystemContext<Message = BitfieldDistributionMessage> + Sync + Send,
 {
 	fn start(self, ctx: C) -> SpawnedSubsystem {
+		let future = self.run(ctx)
+			.map_err(|e| {
+				SubsystemError::with_origin("bitfield-distribution", e)
+			})
+			.map(|_| ()).boxed();
+
 		SpawnedSubsystem {
 			name: "bitfield-distribution-subsystem",
-			future: Box::pin(async move { Self::run(self, ctx) }.map(|_| ())),
+			future,
 		}
 	}
 }

polkadot/node/network/bridge/Cargo.toml

@@ -7,9 +7,7 @@ edition = "2018"
 [dependencies]
 async-trait = "0.1"
 futures = "0.3.5"
-log = "0.4.8"
-futures-timer = "3.0.2"
-streamunordered = "0.5.1"
+log = "0.4.11"
 polkadot-primitives = { path = "../../../primitives" }
 parity-scale-codec = "1.3.4"
 sc-authority-discovery = { git = "https://github.com/paritytech/substrate", branch = "master" }

polkadot/node/network/bridge/src/lib.rs

@@ -16,6 +16,10 @@
 
 //! The Network Bridge Subsystem - protocol multiplexer for Polkadot.
 
+#![deny(unused_crate_dependencies, unused_results)]
+#![warn(missing_docs)]
+
+
 use parity_scale_codec::{Encode, Decode};
 use futures::prelude::*;
 use futures::future::BoxFuture;
@@ -219,13 +223,19 @@ impl<Net, AD, Context> Subsystem<Context> for NetworkBridge<Net, AD>
 		// Swallow error because failure is fatal to the node and we log with more precision
 		// within `run_network`.
 		let Self { network_service, authority_discovery_service } = self;
-		SpawnedSubsystem {
-			name: "network-bridge-subsystem",
-			future: run_network(
+		let future = run_network(
 				network_service,
 				authority_discovery_service,
 				ctx,
-			).map(|_| ()).boxed(),
+			)
+			.map_err(|e| {
+				SubsystemError::with_origin("network-bridge", e)
+			})
+			.map(|_| ())
+			.boxed();
+		SpawnedSubsystem {
+			name: "network-bridge-subsystem",
+			future,
 		}
 	}
 }
@@ -654,7 +664,7 @@ where
 				match peer_map.entry(peer.clone()) {
 					hash_map::Entry::Occupied(_) => continue,
 					hash_map::Entry::Vacant(vacant) => {
-						vacant.insert(PeerData {
+						let _ = vacant.insert(PeerData {
 							view: View(Vec::new()),
 						});
 
@@ -937,7 +947,7 @@ mod tests {
 		futures::pin_mut!(test_fut);
 		futures::pin_mut!(network_bridge);
 
-		executor::block_on(future::select(test_fut, network_bridge));
+		let _ = executor::block_on(future::select(test_fut, network_bridge));
 	}
 
 	async fn assert_sends_validation_event_to_all(

polkadot/node/network/bridge/src/validator_discovery.rs

@@ -192,7 +192,7 @@ impl<N: Network, AD: AuthorityDiscovery> Service<N, AD> {
 				Err(e) if e.is_disconnected() => {
 					// the request is already revoked
 					for peer_id in validator_ids {
-						on_revoke(&mut self.requested_validators, peer_id);
+						let _ = on_revoke(&mut self.requested_validators, peer_id);
 					}
 					return (network_service, authority_discovery_service);
 				}
@@ -217,7 +217,7 @@ impl<N: Network, AD: AuthorityDiscovery> Service<N, AD> {
 				// They are going to be removed soon though:
 				// https://github.com/paritytech/substrate/issues/6845
 				for addr in addresses.into_iter().take(MAX_ADDR_PER_PEER) {
-					multiaddr_to_add.insert(addr);
+					let _ = multiaddr_to_add.insert(addr);
 				}
 			}
 		}
@@ -247,7 +247,7 @@ impl<N: Network, AD: AuthorityDiscovery> Service<N, AD> {
 			let result = authority_discovery_service.get_addresses_by_authority_id(id).await;
 			if let Some(addresses) = result {
 				for addr in addresses.into_iter().take(MAX_ADDR_PER_PEER) {
-					multiaddr_to_remove.insert(addr);
+					let _ = multiaddr_to_remove.insert(addr);
 				}
 			}
 		}
@@ -283,16 +283,16 @@ impl<N: Network, AD: AuthorityDiscovery> Service<N, AD> {
 		let maybe_authority = authority_discovery_service.get_authority_id_by_peer_id(peer_id.clone()).await;
 		if let Some(authority) = maybe_authority {
 			for request in self.non_revoked_discovery_requests.iter_mut() {
-				request.on_authority_connected(&authority, peer_id);
+				let _ = request.on_authority_connected(&authority, peer_id);
 			}
-			self.connected_validators.insert(authority, peer_id.clone());
+			let _ = self.connected_validators.insert(authority, peer_id.clone());
 		}
 	}
 
 	pub async fn on_peer_disconnected(&mut self, peer_id: &PeerId, authority_discovery_service: &mut AD) {
 		let maybe_authority = authority_discovery_service.get_authority_id_by_peer_id(peer_id.clone()).await;
 		if let Some(authority) = maybe_authority {
-			self.connected_validators.remove(&authority);
+			let _ = self.connected_validators.remove(&authority);
 		}
 	}
 }

polkadot/node/network/collator-protocol/Cargo.toml

@@ -7,12 +7,10 @@ edition = "2018"
 [dependencies]
 futures = "0.3.5"
 log = "0.4.11"
-derive_more = "0.99.9"
+thiserror = "1.0.21"
 
-codec = { package="parity-scale-codec", version = "1.3.4", features = ["std"]  }
 
 polkadot-primitives = { path = "../../../primitives" }
-polkadot-network-bridge = { path = "../../network/bridge" }
 polkadot-node-network-protocol = { path = "../../network/protocol" }
 polkadot-node-subsystem-util = { path = "../../subsystem-util" }
 polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }
@@ -20,7 +18,6 @@ polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsys
 [dev-dependencies]
 env_logger = "0.7.1"
 assert_matches = "1.3.0"
-smol-timeout = "0.1.0"
 smallvec = "1.4.2"
 futures-timer = "3.0.2"
 

polkadot/node/network/collator-protocol/src/lib.rs

@@ -17,11 +17,12 @@
 //! The Collator Protocol allows collators and validators talk to each other.
 //! This subsystem implements both sides of the collator protocol.
 
-#![deny(missing_docs)]
+#![deny(missing_docs, unused_crate_dependencies)]
 
 use std::time::Duration;
 use futures::{channel::oneshot, FutureExt};
 use log::trace;
+use thiserror::Error;
 
 use polkadot_subsystem::{
 	Subsystem, SubsystemContext, SubsystemError, SpawnedSubsystem,
@@ -45,18 +46,18 @@ mod validator_side;
 const TARGET: &'static str = "colp";
 const REQUEST_TIMEOUT: Duration = Duration::from_secs(1);
 
-#[derive(Debug, derive_more::From)]
+#[derive(Debug, Error)]
 enum Error {
-	#[from]
-	Subsystem(SubsystemError),
-	#[from]
-	Oneshot(oneshot::Canceled),
-	#[from]
-	RuntimeApi(RuntimeApiError),
-	#[from]
-	UtilError(util::Error),
-	#[from]
-	Prometheus(prometheus::PrometheusError),
+	#[error(transparent)]
+	Subsystem(#[from] SubsystemError),
+	#[error(transparent)]
+	Oneshot(#[from] oneshot::Canceled),
+	#[error(transparent)]
+	RuntimeApi(#[from] RuntimeApiError),
+	#[error(transparent)]
+	UtilError(#[from] util::Error),
+	#[error(transparent)]
+	Prometheus(#[from] prometheus::PrometheusError),
 }
 
 impl From<util::validator_discovery::Error> for Error {
@@ -113,7 +114,9 @@ impl CollatorProtocolSubsystem {
 				id,
 				metrics,
 			).await,
-		}
+		}.map_err(|e| {
+			SubsystemError::with_origin("collator-protocol", e).into()
+		})
 	}
 }
 

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

@@ -6,19 +6,13 @@ edition = "2018"
 
 [dependencies]
 futures = "0.3.5"
-log = "0.4.8"
-futures-timer = "3.0.2"
-streamunordered = "0.5.1"
+log = "0.4.11"
 polkadot-primitives = { path = "../../../primitives" }
-node-primitives = { package = "polkadot-node-primitives", path = "../../primitives" }
-parity-scale-codec = "1.3.4"
-sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "master" }
 polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }
 polkadot-node-subsystem-util = { path = "../../subsystem-util" }
 polkadot-node-network-protocol = { path = "../../network/protocol" }
 
 [dev-dependencies]
-parking_lot = "0.10.0"
 assert_matches = "1.3.0"
 sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }
 polkadot-node-subsystem-test-helpers = { path = "../../subsystem-test-helpers" }

polkadot/node/network/pov-distribution/src/lib.rs

@@ -19,9 +19,13 @@
 //! This is a gossip implementation of code that is responsible for distributing PoVs
 //! among validators.
 
+#![deny(unused_crate_dependencies)]
+#![warn(missing_docs)]
+
 use polkadot_primitives::v1::{Hash, PoV, CandidateDescriptor};
 use polkadot_subsystem::{
-	ActiveLeavesUpdate, OverseerSignal, SubsystemContext, Subsystem, SubsystemResult, FromOverseer, SpawnedSubsystem,
+	ActiveLeavesUpdate, OverseerSignal, SubsystemContext, Subsystem, SubsystemResult, SubsystemError,
+	FromOverseer, SpawnedSubsystem,
 	messages::{
 		PoVDistributionMessage, RuntimeApiMessage, RuntimeApiRequest, AllMessages, NetworkBridgeMessage,
 	},
@@ -60,9 +64,13 @@ impl<C> Subsystem<C> for PoVDistribution
 	fn start(self, ctx: C) -> SpawnedSubsystem {
 		// Swallow error because failure is fatal to the node and we log with more precision
 		// within `run`.
+		let future = self.run(ctx)
+			.map_err(|e| SubsystemError::with_origin("pov-distribution", e))
+			.map(|_| ())
+			.boxed();
 		SpawnedSubsystem {
 			name: "pov-distribution-subsystem",
-			future: self.run(ctx).map(|_| ()).boxed(),
+			future,
 		}
 	}
 }
@@ -608,918 +616,4 @@ impl metrics::Metrics for Metrics {
 }
 
 #[cfg(test)]
-mod tests {
-	use super::*;
-	use futures::executor;
-	use polkadot_primitives::v1::BlockData;
-	use assert_matches::assert_matches;
-
-	fn make_pov(data: Vec<u8>) -> PoV {
-		PoV { block_data: BlockData(data) }
-	}
-
-	fn make_peer_state(awaited: Vec<(Hash, Vec<Hash>)>)
-		-> PeerState
-	{
-		PeerState {
-			awaited: awaited.into_iter().map(|(rp, h)| (rp, h.into_iter().collect())).collect()
-		}
-	}
-
-	#[test]
-	fn distributes_to_those_awaiting_and_completes_local() {
-		let hash_a: Hash = [0; 32].into();
-		let hash_b: Hash = [1; 32].into();
-
-		let peer_a = PeerId::random();
-		let peer_b = PeerId::random();
-		let peer_c = PeerId::random();
-
-		let (pov_send, pov_recv) = oneshot::channel();
-		let pov = make_pov(vec![1, 2, 3]);
-		let pov_hash = pov.hash();
-
-		let mut state = State {
-			relay_parent_state: {
-				let mut s = HashMap::new();
-				let mut b = BlockBasedState {
-					known: HashMap::new(),
-					fetching: HashMap::new(),
-					n_validators: 10,
-				};
-
-				b.fetching.insert(pov_hash, vec![pov_send]);
-				s.insert(hash_a, b);
-				s
-			},
-			peer_state: {
-				let mut s = HashMap::new();
-
-				// peer A has hash_a in its view and is awaiting the PoV.
-				s.insert(
-					peer_a.clone(),
-					make_peer_state(vec![(hash_a, vec![pov_hash])]),
-				);
-
-				// peer B has hash_a in its view but is not awaiting.
-				s.insert(
-					peer_b.clone(),
-					make_peer_state(vec![(hash_a, vec![])]),
-				);
-
-				// peer C doesn't have hash_a in its view but is awaiting the PoV under hash_b.
-				s.insert(
-					peer_c.clone(),
-					make_peer_state(vec![(hash_b, vec![pov_hash])]),
-				);
-
-				s
-			},
-			our_view: View(vec![hash_a, hash_b]),
-			metrics: Default::default(),
-		};
-
-		let pool = sp_core::testing::TaskExecutor::new();
-		let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
-		let mut descriptor = CandidateDescriptor::default();
-		descriptor.pov_hash = pov_hash;
-
-		executor::block_on(async move {
-			handle_distribute(
-				&mut state,
-				&mut ctx,
-				hash_a,
-				descriptor,
-				Arc::new(pov.clone()),
-			).await.unwrap();
-
-			assert!(!state.peer_state[&peer_a].awaited[&hash_a].contains(&pov_hash));
-			assert!(state.peer_state[&peer_c].awaited[&hash_b].contains(&pov_hash));
-
-			// our local sender also completed
-			assert_eq!(&*pov_recv.await.unwrap(), &pov);
-
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::NetworkBridge(
-					NetworkBridgeMessage::SendValidationMessage(peers, message)
-				) => {
-					assert_eq!(peers, vec![peer_a.clone()]);
-					assert_eq!(
-						message,
-						send_pov_message(hash_a, pov_hash, pov.clone()),
-					);
-				}
-			)
-		});
-	}
-
-	#[test]
-	fn we_inform_peers_with_same_view_we_are_awaiting() {
-		let hash_a: Hash = [0; 32].into();
-		let hash_b: Hash = [1; 32].into();
-
-		let peer_a = PeerId::random();
-		let peer_b = PeerId::random();
-
-		let (pov_send, _) = oneshot::channel();
-		let pov = make_pov(vec![1, 2, 3]);
-		let pov_hash = pov.hash();
-
-		let mut state = State {
-			relay_parent_state: {
-				let mut s = HashMap::new();
-				let b = BlockBasedState {
-					known: HashMap::new(),
-					fetching: HashMap::new(),
-					n_validators: 10,
-				};
-
-				s.insert(hash_a, b);
-				s
-			},
-			peer_state: {
-				let mut s = HashMap::new();
-
-				// peer A has hash_a in its view.
-				s.insert(
-					peer_a.clone(),
-					make_peer_state(vec![(hash_a, vec![])]),
-				);
-
-				// peer B doesn't have hash_a in its view.
-				s.insert(
-					peer_b.clone(),
-					make_peer_state(vec![(hash_b, vec![])]),
-				);
-
-				s
-			},
-			our_view: View(vec![hash_a]),
-			metrics: Default::default(),
-		};
-
-		let pool = sp_core::testing::TaskExecutor::new();
-		let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
-		let mut descriptor = CandidateDescriptor::default();
-		descriptor.pov_hash = pov_hash;
-
-		executor::block_on(async move {
-			handle_fetch(
-				&mut state,
-				&mut ctx,
-				hash_a,
-				descriptor,
-				pov_send,
-			).await.unwrap();
-
-			assert_eq!(state.relay_parent_state[&hash_a].fetching[&pov_hash].len(), 1);
-
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::NetworkBridge(
-					NetworkBridgeMessage::SendValidationMessage(peers, message)
-				) => {
-					assert_eq!(peers, vec![peer_a.clone()]);
-					assert_eq!(
-						message,
-						awaiting_message(hash_a, vec![pov_hash]),
-					);
-				}
-			)
-		});
-	}
-
-	#[test]
-	fn peer_view_change_leads_to_us_informing() {
-		let hash_a: Hash = [0; 32].into();
-		let hash_b: Hash = [1; 32].into();
-
-		let peer_a = PeerId::random();
-
-		let (pov_a_send, _) = oneshot::channel();
-
-		let pov_a = make_pov(vec![1, 2, 3]);
-		let pov_a_hash = pov_a.hash();
-
-		let pov_b = make_pov(vec![4, 5, 6]);
-		let pov_b_hash = pov_b.hash();
-
-		let mut state = State {
-			relay_parent_state: {
-				let mut s = HashMap::new();
-				let mut b = BlockBasedState {
-					known: HashMap::new(),
-					fetching: HashMap::new(),
-					n_validators: 10,
-				};
-
-				// pov_a is still being fetched, whereas the fetch of pov_b has already
-				// completed, as implied by the empty vector.
-				b.fetching.insert(pov_a_hash, vec![pov_a_send]);
-				b.fetching.insert(pov_b_hash, vec![]);
-
-				s.insert(hash_a, b);
-				s
-			},
-			peer_state: {
-				let mut s = HashMap::new();
-
-				// peer A doesn't yet have hash_a in its view.
-				s.insert(
-					peer_a.clone(),
-					make_peer_state(vec![(hash_b, vec![])]),
-				);
-
-				s
-			},
-			our_view: View(vec![hash_a]),
-			metrics: Default::default(),
-		};
-
-		let pool = sp_core::testing::TaskExecutor::new();
-		let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
-
-		executor::block_on(async move {
-			handle_network_update(
-				&mut state,
-				&mut ctx,
-				NetworkBridgeEvent::PeerViewChange(peer_a.clone(), View(vec![hash_a, hash_b])),
-			).await.unwrap();
-
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::NetworkBridge(
-					NetworkBridgeMessage::SendValidationMessage(peers, message)
-				) => {
-					assert_eq!(peers, vec![peer_a.clone()]);
-					assert_eq!(
-						message,
-						awaiting_message(hash_a, vec![pov_a_hash]),
-					);
-				}
-			)
-		});
-	}
-
-	#[test]
-	fn peer_complete_fetch_and_is_rewarded() {
-		let hash_a: Hash = [0; 32].into();
-
-		let peer_a = PeerId::random();
-		let peer_b = PeerId::random();
-
-		let (pov_send, pov_recv) = oneshot::channel();
-
-		let pov = make_pov(vec![1, 2, 3]);
-		let pov_hash = pov.hash();
-
-		let mut state = State {
-			relay_parent_state: {
-				let mut s = HashMap::new();
-				let mut b = BlockBasedState {
-					known: HashMap::new(),
-					fetching: HashMap::new(),
-					n_validators: 10,
-				};
-
-				// pov is being fetched.
-				b.fetching.insert(pov_hash, vec![pov_send]);
-
-				s.insert(hash_a, b);
-				s
-			},
-			peer_state: {
-				let mut s = HashMap::new();
-
-				// peers A and B are functionally the same.
-				s.insert(
-					peer_a.clone(),
-					make_peer_state(vec![(hash_a, vec![])]),
-				);
-
-				s.insert(
-					peer_b.clone(),
-					make_peer_state(vec![(hash_a, vec![])]),
-				);
-
-				s
-			},
-			our_view: View(vec![hash_a]),
-			metrics: Default::default(),
-		};
-
-		let pool = sp_core::testing::TaskExecutor::new();
-		let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
-
-		executor::block_on(async move {
-			// Peer A answers our request before peer B.
-			handle_network_update(
-				&mut state,
-				&mut ctx,
-				NetworkBridgeEvent::PeerMessage(
-					peer_a.clone(),
-					send_pov_message(hash_a, pov_hash, pov.clone()),
-				).focus().unwrap(),
-			).await.unwrap();
-
-			handle_network_update(
-				&mut state,
-				&mut ctx,
-				NetworkBridgeEvent::PeerMessage(
-					peer_b.clone(),
-					send_pov_message(hash_a, pov_hash, pov.clone()),
-				).focus().unwrap(),
-			).await.unwrap();
-
-			assert_eq!(&*pov_recv.await.unwrap(), &pov);
-
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::NetworkBridge(
-					NetworkBridgeMessage::ReportPeer(peer, rep)
-				) => {
-					assert_eq!(peer, peer_a);
-					assert_eq!(rep, BENEFIT_FRESH_POV);
-				}
-			);
-
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::NetworkBridge(
-					NetworkBridgeMessage::ReportPeer(peer, rep)
-				) => {
-					assert_eq!(peer, peer_b);
-					assert_eq!(rep, BENEFIT_LATE_POV);
-				}
-			);
-		});
-	}
-
-	#[test]
-	fn peer_punished_for_sending_bad_pov() {
-		let hash_a: Hash = [0; 32].into();
-
-		let peer_a = PeerId::random();
-
-		let (pov_send, _) = oneshot::channel();
-
-		let pov = make_pov(vec![1, 2, 3]);
-		let pov_hash = pov.hash();
-
-		let bad_pov = make_pov(vec![6, 6, 6]);
-
-		let mut state = State {
-			relay_parent_state: {
-				let mut s = HashMap::new();
-				let mut b = BlockBasedState {
-					known: HashMap::new(),
-					fetching: HashMap::new(),
-					n_validators: 10,
-				};
-
-				// pov is being fetched.
-				b.fetching.insert(pov_hash, vec![pov_send]);
-
-				s.insert(hash_a, b);
-				s
-			},
-			peer_state: {
-				let mut s = HashMap::new();
-
-				s.insert(
-					peer_a.clone(),
-					make_peer_state(vec![(hash_a, vec![])]),
-				);
-
-				s
-			},
-			our_view: View(vec![hash_a]),
-			metrics: Default::default(),
-		};
-
-		let pool = sp_core::testing::TaskExecutor::new();
-		let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
-
-		executor::block_on(async move {
-			// Peer A answers our request: right relay parent, awaited hash, wrong PoV.
-			handle_network_update(
-				&mut state,
-				&mut ctx,
-				NetworkBridgeEvent::PeerMessage(
-					peer_a.clone(),
-					send_pov_message(hash_a, pov_hash, bad_pov.clone()),
-				).focus().unwrap(),
-			).await.unwrap();
-
-			// didn't complete our sender.
-			assert_eq!(state.relay_parent_state[&hash_a].fetching[&pov_hash].len(), 1);
-
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::NetworkBridge(
-					NetworkBridgeMessage::ReportPeer(peer, rep)
-				) => {
-					assert_eq!(peer, peer_a);
-					assert_eq!(rep, COST_UNEXPECTED_POV);
-				}
-			);
-		});
-	}
-
-	#[test]
-	fn peer_punished_for_sending_unexpected_pov() {
-		let hash_a: Hash = [0; 32].into();
-
-		let peer_a = PeerId::random();
-
-		let pov = make_pov(vec![1, 2, 3]);
-		let pov_hash = pov.hash();
-
-		let mut state = State {
-			relay_parent_state: {
-				let mut s = HashMap::new();
-				let b = BlockBasedState {
-					known: HashMap::new(),
-					fetching: HashMap::new(),
-					n_validators: 10,
-				};
-
-				s.insert(hash_a, b);
-				s
-			},
-			peer_state: {
-				let mut s = HashMap::new();
-
-				s.insert(
-					peer_a.clone(),
-					make_peer_state(vec![(hash_a, vec![])]),
-				);
-
-				s
-			},
-			our_view: View(vec![hash_a]),
-			metrics: Default::default(),
-		};
-
-		let pool = sp_core::testing::TaskExecutor::new();
-		let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
-
-		executor::block_on(async move {
-			// Peer A answers our request: right relay parent, awaited hash, wrong PoV.
-			handle_network_update(
-				&mut state,
-				&mut ctx,
-				NetworkBridgeEvent::PeerMessage(
-					peer_a.clone(),
-					send_pov_message(hash_a, pov_hash, pov.clone()),
-				).focus().unwrap(),
-			).await.unwrap();
-
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::NetworkBridge(
-					NetworkBridgeMessage::ReportPeer(peer, rep)
-				) => {
-					assert_eq!(peer, peer_a);
-					assert_eq!(rep, COST_UNEXPECTED_POV);
-				}
-			);
-		});
-	}
-
-	#[test]
-	fn peer_punished_for_sending_pov_out_of_our_view() {
-		let hash_a: Hash = [0; 32].into();
-		let hash_b: Hash = [1; 32].into();
-
-		let peer_a = PeerId::random();
-
-		let pov = make_pov(vec![1, 2, 3]);
-		let pov_hash = pov.hash();
-
-		let mut state = State {
-			relay_parent_state: {
-				let mut s = HashMap::new();
-				let b = BlockBasedState {
-					known: HashMap::new(),
-					fetching: HashMap::new(),
-					n_validators: 10,
-				};
-
-				s.insert(hash_a, b);
-				s
-			},
-			peer_state: {
-				let mut s = HashMap::new();
-
-				s.insert(
-					peer_a.clone(),
-					make_peer_state(vec![(hash_a, vec![])]),
-				);
-
-				s
-			},
-			our_view: View(vec![hash_a]),
-			metrics: Default::default(),
-		};
-
-		let pool = sp_core::testing::TaskExecutor::new();
-		let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
-
-		executor::block_on(async move {
-			// Peer A answers our request: right relay parent, awaited hash, wrong PoV.
-			handle_network_update(
-				&mut state,
-				&mut ctx,
-				NetworkBridgeEvent::PeerMessage(
-					peer_a.clone(),
-					send_pov_message(hash_b, pov_hash, pov.clone()),
-				).focus().unwrap(),
-			).await.unwrap();
-
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::NetworkBridge(
-					NetworkBridgeMessage::ReportPeer(peer, rep)
-				) => {
-					assert_eq!(peer, peer_a);
-					assert_eq!(rep, COST_UNEXPECTED_POV);
-				}
-			);
-		});
-	}
-
-	#[test]
-	fn peer_reported_for_awaiting_too_much() {
-		let hash_a: Hash = [0; 32].into();
-
-		let peer_a = PeerId::random();
-		let n_validators = 10;
-
-		let mut state = State {
-			relay_parent_state: {
-				let mut s = HashMap::new();
-				let b = BlockBasedState {
-					known: HashMap::new(),
-					fetching: HashMap::new(),
-					n_validators,
-				};
-
-				s.insert(hash_a, b);
-				s
-			},
-			peer_state: {
-				let mut s = HashMap::new();
-
-				s.insert(
-					peer_a.clone(),
-					make_peer_state(vec![(hash_a, vec![])]),
-				);
-
-				s
-			},
-			our_view: View(vec![hash_a]),
-			metrics: Default::default(),
-		};
-
-		let pool = sp_core::testing::TaskExecutor::new();
-		let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
-
-		executor::block_on(async move {
-			let max_plausibly_awaited = n_validators * 2;
-
-			// The peer awaits a plausible (albeit unlikely) amount of PoVs.
-			for i in 0..max_plausibly_awaited {
-				let pov_hash = make_pov(vec![i as u8; 32]).hash();
-				handle_network_update(
-					&mut state,
-					&mut ctx,
-					NetworkBridgeEvent::PeerMessage(
-						peer_a.clone(),
-						awaiting_message(hash_a, vec![pov_hash]),
-					).focus().unwrap(),
-				).await.unwrap();
-			}
-
-			assert_eq!(state.peer_state[&peer_a].awaited[&hash_a].len(), max_plausibly_awaited);
-
-			// The last straw:
-			let last_pov_hash = make_pov(vec![max_plausibly_awaited as u8; 32]).hash();
-			handle_network_update(
-				&mut state,
-				&mut ctx,
-				NetworkBridgeEvent::PeerMessage(
-					peer_a.clone(),
-					awaiting_message(hash_a, vec![last_pov_hash]),
-				).focus().unwrap(),
-			).await.unwrap();
-
-			// No more bookkeeping for you!
-			assert_eq!(state.peer_state[&peer_a].awaited[&hash_a].len(), max_plausibly_awaited);
-
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::NetworkBridge(
-					NetworkBridgeMessage::ReportPeer(peer, rep)
-				) => {
-					assert_eq!(peer, peer_a);
-					assert_eq!(rep, COST_APPARENT_FLOOD);
-				}
-			);
-		});
-	}
-
-	#[test]
-	fn peer_reported_for_awaiting_outside_their_view() {
-		let hash_a: Hash = [0; 32].into();
-		let hash_b: Hash = [1; 32].into();
-
-		let peer_a = PeerId::random();
-
-		let mut state = State {
-			relay_parent_state: {
-				let mut s = HashMap::new();
-				s.insert(hash_a, BlockBasedState {
-					known: HashMap::new(),
-					fetching: HashMap::new(),
-					n_validators: 10,
-				});
-
-				s.insert(hash_b, BlockBasedState {
-					known: HashMap::new(),
-					fetching: HashMap::new(),
-					n_validators: 10,
-				});
-
-				s
-			},
-			peer_state: {
-				let mut s = HashMap::new();
-
-				// Peer has only hash A in its view.
-				s.insert(
-					peer_a.clone(),
-					make_peer_state(vec![(hash_a, vec![])]),
-				);
-
-				s
-			},
-			our_view: View(vec![hash_a, hash_b]),
-			metrics: Default::default(),
-		};
-
-		let pool = sp_core::testing::TaskExecutor::new();
-		let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
-
-		executor::block_on(async move {
-			let pov_hash = make_pov(vec![1, 2, 3]).hash();
-
-			// Hash B is in our view but not the peer's
-			handle_network_update(
-				&mut state,
-				&mut ctx,
-				NetworkBridgeEvent::PeerMessage(
-					peer_a.clone(),
-					awaiting_message(hash_b, vec![pov_hash]),
-				).focus().unwrap(),
-			).await.unwrap();
-
-			assert!(state.peer_state[&peer_a].awaited.get(&hash_b).is_none());
-
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::NetworkBridge(
-					NetworkBridgeMessage::ReportPeer(peer, rep)
-				) => {
-					assert_eq!(peer, peer_a);
-					assert_eq!(rep, COST_AWAITED_NOT_IN_VIEW);
-				}
-			);
-		});
-	}
-
-	#[test]
-	fn peer_reported_for_awaiting_outside_our_view() {
-		let hash_a: Hash = [0; 32].into();
-		let hash_b: Hash = [1; 32].into();
-
-		let peer_a = PeerId::random();
-
-		let mut state = State {
-			relay_parent_state: {
-				let mut s = HashMap::new();
-				s.insert(hash_a, BlockBasedState {
-					known: HashMap::new(),
-					fetching: HashMap::new(),
-					n_validators: 10,
-				});
-
-				s
-			},
-			peer_state: {
-				let mut s = HashMap::new();
-
-				// Peer has hashes A and B in their view.
-				s.insert(
-					peer_a.clone(),
-					make_peer_state(vec![(hash_a, vec![]), (hash_b, vec![])]),
-				);
-
-				s
-			},
-			our_view: View(vec![hash_a]),
-			metrics: Default::default(),
-		};
-
-		let pool = sp_core::testing::TaskExecutor::new();
-		let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
-
-		executor::block_on(async move {
-			let pov_hash = make_pov(vec![1, 2, 3]).hash();
-
-			// Hash B is in peer's view but not ours.
-			handle_network_update(
-				&mut state,
-				&mut ctx,
-				NetworkBridgeEvent::PeerMessage(
-					peer_a.clone(),
-					awaiting_message(hash_b, vec![pov_hash]),
-				).focus().unwrap(),
-			).await.unwrap();
-
-			// Illegal `awaited` is ignored.
-			assert!(state.peer_state[&peer_a].awaited[&hash_b].is_empty());
-
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::NetworkBridge(
-					NetworkBridgeMessage::ReportPeer(peer, rep)
-				) => {
-					assert_eq!(peer, peer_a);
-					assert_eq!(rep, COST_AWAITED_NOT_IN_VIEW);
-				}
-			);
-		});
-	}
-
-	#[test]
-	fn peer_complete_fetch_leads_to_us_completing_others() {
-		let hash_a: Hash = [0; 32].into();
-
-		let peer_a = PeerId::random();
-		let peer_b = PeerId::random();
-
-		let (pov_send, pov_recv) = oneshot::channel();
-
-		let pov = make_pov(vec![1, 2, 3]);
-		let pov_hash = pov.hash();
-
-		let mut state = State {
-			relay_parent_state: {
-				let mut s = HashMap::new();
-				let mut b = BlockBasedState {
-					known: HashMap::new(),
-					fetching: HashMap::new(),
-					n_validators: 10,
-				};
-
-				// pov is being fetched.
-				b.fetching.insert(pov_hash, vec![pov_send]);
-
-				s.insert(hash_a, b);
-				s
-			},
-			peer_state: {
-				let mut s = HashMap::new();
-
-				s.insert(
-					peer_a.clone(),
-					make_peer_state(vec![(hash_a, vec![])]),
-				);
-
-				// peer B is awaiting peer A's request.
-				s.insert(
-					peer_b.clone(),
-					make_peer_state(vec![(hash_a, vec![pov_hash])]),
-				);
-
-				s
-			},
-			our_view: View(vec![hash_a]),
-			metrics: Default::default(),
-		};
-
-		let pool = sp_core::testing::TaskExecutor::new();
-		let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
-
-		executor::block_on(async move {
-			handle_network_update(
-				&mut state,
-				&mut ctx,
-				NetworkBridgeEvent::PeerMessage(
-					peer_a.clone(),
-					send_pov_message(hash_a, pov_hash, pov.clone()),
-				).focus().unwrap(),
-			).await.unwrap();
-
-			assert_eq!(&*pov_recv.await.unwrap(), &pov);
-
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::NetworkBridge(
-					NetworkBridgeMessage::ReportPeer(peer, rep)
-				) => {
-					assert_eq!(peer, peer_a);
-					assert_eq!(rep, BENEFIT_FRESH_POV);
-				}
-			);
-
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::NetworkBridge(
-					NetworkBridgeMessage::SendValidationMessage(peers, message)
-				) => {
-					assert_eq!(peers, vec![peer_b.clone()]);
-					assert_eq!(
-						message,
-						send_pov_message(hash_a, pov_hash, pov.clone()),
-					);
-				}
-			);
-
-			assert!(!state.peer_state[&peer_b].awaited[&hash_a].contains(&pov_hash));
-		});
-	}
-
-	#[test]
-	fn peer_completing_request_no_longer_awaiting() {
-		let hash_a: Hash = [0; 32].into();
-
-		let peer_a = PeerId::random();
-
-		let (pov_send, pov_recv) = oneshot::channel();
-
-		let pov = make_pov(vec![1, 2, 3]);
-		let pov_hash = pov.hash();
-
-		let mut state = State {
-			relay_parent_state: {
-				let mut s = HashMap::new();
-				let mut b = BlockBasedState {
-					known: HashMap::new(),
-					fetching: HashMap::new(),
-					n_validators: 10,
-				};
-
-				// pov is being fetched.
-				b.fetching.insert(pov_hash, vec![pov_send]);
-
-				s.insert(hash_a, b);
-				s
-			},
-			peer_state: {
-				let mut s = HashMap::new();
-
-				// peer A is registered as awaiting.
-				s.insert(
-					peer_a.clone(),
-					make_peer_state(vec![(hash_a, vec![pov_hash])]),
-				);
-
-				s
-			},
-			our_view: View(vec![hash_a]),
-			metrics: Default::default(),
-		};
-
-		let pool = sp_core::testing::TaskExecutor::new();
-		let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
-
-		executor::block_on(async move {
-			handle_network_update(
-				&mut state,
-				&mut ctx,
-				NetworkBridgeEvent::PeerMessage(
-					peer_a.clone(),
-					send_pov_message(hash_a, pov_hash, pov.clone()),
-				).focus().unwrap(),
-			).await.unwrap();
-
-			assert_eq!(&*pov_recv.await.unwrap(), &pov);
-
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::NetworkBridge(
-					NetworkBridgeMessage::ReportPeer(peer, rep)
-				) => {
-					assert_eq!(peer, peer_a);
-					assert_eq!(rep, BENEFIT_FRESH_POV);
-				}
-			);
-
-			// We received the PoV from peer A, so we do not consider it awaited by peer A anymore.
-			assert!(!state.peer_state[&peer_a].awaited[&hash_a].contains(&pov_hash));
-		});
-	}
-}
+mod tests;

polkadot/node/network/pov-distribution/src/tests.rs

@@ -0,0 +1,913 @@
+use super::*;
+use futures::executor;
+use polkadot_primitives::v1::BlockData;
+use assert_matches::assert_matches;
+
+fn make_pov(data: Vec<u8>) -> PoV {
+	PoV { block_data: BlockData(data) }
+}
+
+fn make_peer_state(awaited: Vec<(Hash, Vec<Hash>)>)
+	-> PeerState
+{
+	PeerState {
+		awaited: awaited.into_iter().map(|(rp, h)| (rp, h.into_iter().collect())).collect()
+	}
+}
+
+#[test]
+fn distributes_to_those_awaiting_and_completes_local() {
+	let hash_a: Hash = [0; 32].into();
+	let hash_b: Hash = [1; 32].into();
+
+	let peer_a = PeerId::random();
+	let peer_b = PeerId::random();
+	let peer_c = PeerId::random();
+
+	let (pov_send, pov_recv) = oneshot::channel();
+	let pov = make_pov(vec![1, 2, 3]);
+	let pov_hash = pov.hash();
+
+	let mut state = State {
+		relay_parent_state: {
+			let mut s = HashMap::new();
+			let mut b = BlockBasedState {
+				known: HashMap::new(),
+				fetching: HashMap::new(),
+				n_validators: 10,
+			};
+
+			b.fetching.insert(pov_hash, vec![pov_send]);
+			s.insert(hash_a, b);
+			s
+		},
+		peer_state: {
+			let mut s = HashMap::new();
+
+			// peer A has hash_a in its view and is awaiting the PoV.
+			s.insert(
+				peer_a.clone(),
+				make_peer_state(vec![(hash_a, vec![pov_hash])]),
+			);
+
+			// peer B has hash_a in its view but is not awaiting.
+			s.insert(
+				peer_b.clone(),
+				make_peer_state(vec![(hash_a, vec![])]),
+			);
+
+			// peer C doesn't have hash_a in its view but is awaiting the PoV under hash_b.
+			s.insert(
+				peer_c.clone(),
+				make_peer_state(vec![(hash_b, vec![pov_hash])]),
+			);
+
+			s
+		},
+		our_view: View(vec![hash_a, hash_b]),
+		metrics: Default::default(),
+	};
+
+	let pool = sp_core::testing::TaskExecutor::new();
+	let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
+	let mut descriptor = CandidateDescriptor::default();
+	descriptor.pov_hash = pov_hash;
+
+	executor::block_on(async move {
+		handle_distribute(
+			&mut state,
+			&mut ctx,
+			hash_a,
+			descriptor,
+			Arc::new(pov.clone()),
+		).await.unwrap();
+
+		assert!(!state.peer_state[&peer_a].awaited[&hash_a].contains(&pov_hash));
+		assert!(state.peer_state[&peer_c].awaited[&hash_b].contains(&pov_hash));
+
+		// our local sender also completed
+		assert_eq!(&*pov_recv.await.unwrap(), &pov);
+
+		assert_matches!(
+			handle.recv().await,
+			AllMessages::NetworkBridge(
+				NetworkBridgeMessage::SendValidationMessage(peers, message)
+			) => {
+				assert_eq!(peers, vec![peer_a.clone()]);
+				assert_eq!(
+					message,
+					send_pov_message(hash_a, pov_hash, pov.clone()),
+				);
+			}
+		)
+	});
+}
+
+#[test]
+fn we_inform_peers_with_same_view_we_are_awaiting() {
+	let hash_a: Hash = [0; 32].into();
+	let hash_b: Hash = [1; 32].into();
+
+	let peer_a = PeerId::random();
+	let peer_b = PeerId::random();
+
+	let (pov_send, _) = oneshot::channel();
+	let pov = make_pov(vec![1, 2, 3]);
+	let pov_hash = pov.hash();
+
+	let mut state = State {
+		relay_parent_state: {
+			let mut s = HashMap::new();
+			let b = BlockBasedState {
+				known: HashMap::new(),
+				fetching: HashMap::new(),
+				n_validators: 10,
+			};
+
+			s.insert(hash_a, b);
+			s
+		},
+		peer_state: {
+			let mut s = HashMap::new();
+
+			// peer A has hash_a in its view.
+			s.insert(
+				peer_a.clone(),
+				make_peer_state(vec![(hash_a, vec![])]),
+			);
+
+			// peer B doesn't have hash_a in its view.
+			s.insert(
+				peer_b.clone(),
+				make_peer_state(vec![(hash_b, vec![])]),
+			);
+
+			s
+		},
+		our_view: View(vec![hash_a]),
+		metrics: Default::default(),
+	};
+
+	let pool = sp_core::testing::TaskExecutor::new();
+	let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
+	let mut descriptor = CandidateDescriptor::default();
+	descriptor.pov_hash = pov_hash;
+
+	executor::block_on(async move {
+		handle_fetch(
+			&mut state,
+			&mut ctx,
+			hash_a,
+			descriptor,
+			pov_send,
+		).await.unwrap();
+
+		assert_eq!(state.relay_parent_state[&hash_a].fetching[&pov_hash].len(), 1);
+
+		assert_matches!(
+			handle.recv().await,
+			AllMessages::NetworkBridge(
+				NetworkBridgeMessage::SendValidationMessage(peers, message)
+			) => {
+				assert_eq!(peers, vec![peer_a.clone()]);
+				assert_eq!(
+					message,
+					awaiting_message(hash_a, vec![pov_hash]),
+				);
+			}
+		)
+	});
+}
+
+#[test]
+fn peer_view_change_leads_to_us_informing() {
+	let hash_a: Hash = [0; 32].into();
+	let hash_b: Hash = [1; 32].into();
+
+	let peer_a = PeerId::random();
+
+	let (pov_a_send, _) = oneshot::channel();
+
+	let pov_a = make_pov(vec![1, 2, 3]);
+	let pov_a_hash = pov_a.hash();
+
+	let pov_b = make_pov(vec![4, 5, 6]);
+	let pov_b_hash = pov_b.hash();
+
+	let mut state = State {
+		relay_parent_state: {
+			let mut s = HashMap::new();
+			let mut b = BlockBasedState {
+				known: HashMap::new(),
+				fetching: HashMap::new(),
+				n_validators: 10,
+			};
+
+			// pov_a is still being fetched, whereas the fetch of pov_b has already
+			// completed, as implied by the empty vector.
+			b.fetching.insert(pov_a_hash, vec![pov_a_send]);
+			b.fetching.insert(pov_b_hash, vec![]);
+
+			s.insert(hash_a, b);
+			s
+		},
+		peer_state: {
+			let mut s = HashMap::new();
+
+			// peer A doesn't yet have hash_a in its view.
+			s.insert(
+				peer_a.clone(),
+				make_peer_state(vec![(hash_b, vec![])]),
+			);
+
+			s
+		},
+		our_view: View(vec![hash_a]),
+		metrics: Default::default(),
+	};
+
+	let pool = sp_core::testing::TaskExecutor::new();
+	let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
+
+	executor::block_on(async move {
+		handle_network_update(
+			&mut state,
+			&mut ctx,
+			NetworkBridgeEvent::PeerViewChange(peer_a.clone(), View(vec![hash_a, hash_b])),
+		).await.unwrap();
+
+		assert_matches!(
+			handle.recv().await,
+			AllMessages::NetworkBridge(
+				NetworkBridgeMessage::SendValidationMessage(peers, message)
+			) => {
+				assert_eq!(peers, vec![peer_a.clone()]);
+				assert_eq!(
+					message,
+					awaiting_message(hash_a, vec![pov_a_hash]),
+				);
+			}
+		)
+	});
+}
+
+#[test]
+fn peer_complete_fetch_and_is_rewarded() {
+	let hash_a: Hash = [0; 32].into();
+
+	let peer_a = PeerId::random();
+	let peer_b = PeerId::random();
+
+	let (pov_send, pov_recv) = oneshot::channel();
+
+	let pov = make_pov(vec![1, 2, 3]);
+	let pov_hash = pov.hash();
+
+	let mut state = State {
+		relay_parent_state: {
+			let mut s = HashMap::new();
+			let mut b = BlockBasedState {
+				known: HashMap::new(),
+				fetching: HashMap::new(),
+				n_validators: 10,
+			};
+
+			// pov is being fetched.
+			b.fetching.insert(pov_hash, vec![pov_send]);
+
+			s.insert(hash_a, b);
+			s
+		},
+		peer_state: {
+			let mut s = HashMap::new();
+
+			// peers A and B are functionally the same.
+			s.insert(
+				peer_a.clone(),
+				make_peer_state(vec![(hash_a, vec![])]),
+			);
+
+			s.insert(
+				peer_b.clone(),
+				make_peer_state(vec![(hash_a, vec![])]),
+			);
+
+			s
+		},
+		our_view: View(vec![hash_a]),
+		metrics: Default::default(),
+	};
+
+	let pool = sp_core::testing::TaskExecutor::new();
+	let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
+
+	executor::block_on(async move {
+		// Peer A answers our request before peer B.
+		handle_network_update(
+			&mut state,
+			&mut ctx,
+			NetworkBridgeEvent::PeerMessage(
+				peer_a.clone(),
+				send_pov_message(hash_a, pov_hash, pov.clone()),
+			).focus().unwrap(),
+		).await.unwrap();
+
+		handle_network_update(
+			&mut state,
+			&mut ctx,
+			NetworkBridgeEvent::PeerMessage(
+				peer_b.clone(),
+				send_pov_message(hash_a, pov_hash, pov.clone()),
+			).focus().unwrap(),
+		).await.unwrap();
+
+		assert_eq!(&*pov_recv.await.unwrap(), &pov);
+
+		assert_matches!(
+			handle.recv().await,
+			AllMessages::NetworkBridge(
+				NetworkBridgeMessage::ReportPeer(peer, rep)
+			) => {
+				assert_eq!(peer, peer_a);
+				assert_eq!(rep, BENEFIT_FRESH_POV);
+			}
+		);
+
+		assert_matches!(
+			handle.recv().await,
+			AllMessages::NetworkBridge(
+				NetworkBridgeMessage::ReportPeer(peer, rep)
+			) => {
+				assert_eq!(peer, peer_b);
+				assert_eq!(rep, BENEFIT_LATE_POV);
+			}
+		);
+	});
+}
+
+#[test]
+fn peer_punished_for_sending_bad_pov() {
+	let hash_a: Hash = [0; 32].into();
+
+	let peer_a = PeerId::random();
+
+	let (pov_send, _) = oneshot::channel();
+
+	let pov = make_pov(vec![1, 2, 3]);
+	let pov_hash = pov.hash();
+
+	let bad_pov = make_pov(vec![6, 6, 6]);
+
+	let mut state = State {
+		relay_parent_state: {
+			let mut s = HashMap::new();
+			let mut b = BlockBasedState {
+				known: HashMap::new(),
+				fetching: HashMap::new(),
+				n_validators: 10,
+			};
+
+			// pov is being fetched.
+			b.fetching.insert(pov_hash, vec![pov_send]);
+
+			s.insert(hash_a, b);
+			s
+		},
+		peer_state: {
+			let mut s = HashMap::new();
+
+			s.insert(
+				peer_a.clone(),
+				make_peer_state(vec![(hash_a, vec![])]),
+			);
+
+			s
+		},
+		our_view: View(vec![hash_a]),
+		metrics: Default::default(),
+	};
+
+	let pool = sp_core::testing::TaskExecutor::new();
+	let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
+
+	executor::block_on(async move {
+		// Peer A answers our request: right relay parent, awaited hash, wrong PoV.
+		handle_network_update(
+			&mut state,
+			&mut ctx,
+			NetworkBridgeEvent::PeerMessage(
+				peer_a.clone(),
+				send_pov_message(hash_a, pov_hash, bad_pov.clone()),
+			).focus().unwrap(),
+		).await.unwrap();
+
+		// didn't complete our sender.
+		assert_eq!(state.relay_parent_state[&hash_a].fetching[&pov_hash].len(), 1);
+
+		assert_matches!(
+			handle.recv().await,
+			AllMessages::NetworkBridge(
+				NetworkBridgeMessage::ReportPeer(peer, rep)
+			) => {
+				assert_eq!(peer, peer_a);
+				assert_eq!(rep, COST_UNEXPECTED_POV);
+			}
+		);
+	});
+}
+
+#[test]
+fn peer_punished_for_sending_unexpected_pov() {
+	let hash_a: Hash = [0; 32].into();
+
+	let peer_a = PeerId::random();
+
+	let pov = make_pov(vec![1, 2, 3]);
+	let pov_hash = pov.hash();
+
+	let mut state = State {
+		relay_parent_state: {
+			let mut s = HashMap::new();
+			let b = BlockBasedState {
+				known: HashMap::new(),
+				fetching: HashMap::new(),
+				n_validators: 10,
+			};
+
+			s.insert(hash_a, b);
+			s
+		},
+		peer_state: {
+			let mut s = HashMap::new();
+
+			s.insert(
+				peer_a.clone(),
+				make_peer_state(vec![(hash_a, vec![])]),
+			);
+
+			s
+		},
+		our_view: View(vec![hash_a]),
+		metrics: Default::default(),
+	};
+
+	let pool = sp_core::testing::TaskExecutor::new();
+	let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
+
+	executor::block_on(async move {
+		// Peer A answers our request: right relay parent, awaited hash, wrong PoV.
+		handle_network_update(
+			&mut state,
+			&mut ctx,
+			NetworkBridgeEvent::PeerMessage(
+				peer_a.clone(),
+				send_pov_message(hash_a, pov_hash, pov.clone()),
+			).focus().unwrap(),
+		).await.unwrap();
+
+		assert_matches!(
+			handle.recv().await,
+			AllMessages::NetworkBridge(
+				NetworkBridgeMessage::ReportPeer(peer, rep)
+			) => {
+				assert_eq!(peer, peer_a);
+				assert_eq!(rep, COST_UNEXPECTED_POV);
+			}
+		);
+	});
+}
+
+#[test]
+fn peer_punished_for_sending_pov_out_of_our_view() {
+	let hash_a: Hash = [0; 32].into();
+	let hash_b: Hash = [1; 32].into();
+
+	let peer_a = PeerId::random();
+
+	let pov = make_pov(vec![1, 2, 3]);
+	let pov_hash = pov.hash();
+
+	let mut state = State {
+		relay_parent_state: {
+			let mut s = HashMap::new();
+			let b = BlockBasedState {
+				known: HashMap::new(),
+				fetching: HashMap::new(),
+				n_validators: 10,
+			};
+
+			s.insert(hash_a, b);
+			s
+		},
+		peer_state: {
+			let mut s = HashMap::new();
+
+			s.insert(
+				peer_a.clone(),
+				make_peer_state(vec![(hash_a, vec![])]),
+			);
+
+			s
+		},
+		our_view: View(vec![hash_a]),
+		metrics: Default::default(),
+	};
+
+	let pool = sp_core::testing::TaskExecutor::new();
+	let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
+
+	executor::block_on(async move {
+		// Peer A answers our request: right relay parent, awaited hash, wrong PoV.
+		handle_network_update(
+			&mut state,
+			&mut ctx,
+			NetworkBridgeEvent::PeerMessage(
+				peer_a.clone(),
+				send_pov_message(hash_b, pov_hash, pov.clone()),
+			).focus().unwrap(),
+		).await.unwrap();
+
+		assert_matches!(
+			handle.recv().await,
+			AllMessages::NetworkBridge(
+				NetworkBridgeMessage::ReportPeer(peer, rep)
+			) => {
+				assert_eq!(peer, peer_a);
+				assert_eq!(rep, COST_UNEXPECTED_POV);
+			}
+		);
+	});
+}
+
+#[test]
+fn peer_reported_for_awaiting_too_much() {
+	let hash_a: Hash = [0; 32].into();
+
+	let peer_a = PeerId::random();
+	let n_validators = 10;
+
+	let mut state = State {
+		relay_parent_state: {
+			let mut s = HashMap::new();
+			let b = BlockBasedState {
+				known: HashMap::new(),
+				fetching: HashMap::new(),
+				n_validators,
+			};
+
+			s.insert(hash_a, b);
+			s
+		},
+		peer_state: {
+			let mut s = HashMap::new();
+
+			s.insert(
+				peer_a.clone(),
+				make_peer_state(vec![(hash_a, vec![])]),
+			);
+
+			s
+		},
+		our_view: View(vec![hash_a]),
+		metrics: Default::default(),
+	};
+
+	let pool = sp_core::testing::TaskExecutor::new();
+	let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
+
+	executor::block_on(async move {
+		let max_plausibly_awaited = n_validators * 2;
+
+		// The peer awaits a plausible (albeit unlikely) amount of PoVs.
+		for i in 0..max_plausibly_awaited {
+			let pov_hash = make_pov(vec![i as u8; 32]).hash();
+			handle_network_update(
+				&mut state,
+				&mut ctx,
+				NetworkBridgeEvent::PeerMessage(
+					peer_a.clone(),
+					awaiting_message(hash_a, vec![pov_hash]),
+				).focus().unwrap(),
+			).await.unwrap();
+		}
+
+		assert_eq!(state.peer_state[&peer_a].awaited[&hash_a].len(), max_plausibly_awaited);
+
+		// The last straw:
+		let last_pov_hash = make_pov(vec![max_plausibly_awaited as u8; 32]).hash();
+		handle_network_update(
+			&mut state,
+			&mut ctx,
+			NetworkBridgeEvent::PeerMessage(
+				peer_a.clone(),
+				awaiting_message(hash_a, vec![last_pov_hash]),
+			).focus().unwrap(),
+		).await.unwrap();
+
+		// No more bookkeeping for you!
+		assert_eq!(state.peer_state[&peer_a].awaited[&hash_a].len(), max_plausibly_awaited);
+
+		assert_matches!(
+			handle.recv().await,
+			AllMessages::NetworkBridge(
+				NetworkBridgeMessage::ReportPeer(peer, rep)
+			) => {
+				assert_eq!(peer, peer_a);
+				assert_eq!(rep, COST_APPARENT_FLOOD);
+			}
+		);
+	});
+}
+
+#[test]
+fn peer_reported_for_awaiting_outside_their_view() {
+	let hash_a: Hash = [0; 32].into();
+	let hash_b: Hash = [1; 32].into();
+
+	let peer_a = PeerId::random();
+
+	let mut state = State {
+		relay_parent_state: {
+			let mut s = HashMap::new();
+			s.insert(hash_a, BlockBasedState {
+				known: HashMap::new(),
+				fetching: HashMap::new(),
+				n_validators: 10,
+			});
+
+			s.insert(hash_b, BlockBasedState {
+				known: HashMap::new(),
+				fetching: HashMap::new(),
+				n_validators: 10,
+			});
+
+			s
+		},
+		peer_state: {
+			let mut s = HashMap::new();
+
+			// Peer has only hash A in its view.
+			s.insert(
+				peer_a.clone(),
+				make_peer_state(vec![(hash_a, vec![])]),
+			);
+
+			s
+		},
+		our_view: View(vec![hash_a, hash_b]),
+		metrics: Default::default(),
+	};
+
+	let pool = sp_core::testing::TaskExecutor::new();
+	let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
+
+	executor::block_on(async move {
+		let pov_hash = make_pov(vec![1, 2, 3]).hash();
+
+		// Hash B is in our view but not the peer's
+		handle_network_update(
+			&mut state,
+			&mut ctx,
+			NetworkBridgeEvent::PeerMessage(
+				peer_a.clone(),
+				awaiting_message(hash_b, vec![pov_hash]),
+			).focus().unwrap(),
+		).await.unwrap();
+
+		assert!(state.peer_state[&peer_a].awaited.get(&hash_b).is_none());
+
+		assert_matches!(
+			handle.recv().await,
+			AllMessages::NetworkBridge(
+				NetworkBridgeMessage::ReportPeer(peer, rep)
+			) => {
+				assert_eq!(peer, peer_a);
+				assert_eq!(rep, COST_AWAITED_NOT_IN_VIEW);
+			}
+		);
+	});
+}
+
+#[test]
+fn peer_reported_for_awaiting_outside_our_view() {
+	let hash_a: Hash = [0; 32].into();
+	let hash_b: Hash = [1; 32].into();
+
+	let peer_a = PeerId::random();
+
+	let mut state = State {
+		relay_parent_state: {
+			let mut s = HashMap::new();
+			s.insert(hash_a, BlockBasedState {
+				known: HashMap::new(),
+				fetching: HashMap::new(),
+				n_validators: 10,
+			});
+
+			s
+		},
+		peer_state: {
+			let mut s = HashMap::new();
+
+			// Peer has hashes A and B in their view.
+			s.insert(
+				peer_a.clone(),
+				make_peer_state(vec![(hash_a, vec![]), (hash_b, vec![])]),
+			);
+
+			s
+		},
+		our_view: View(vec![hash_a]),
+		metrics: Default::default(),
+	};
+
+	let pool = sp_core::testing::TaskExecutor::new();
+	let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
+
+	executor::block_on(async move {
+		let pov_hash = make_pov(vec![1, 2, 3]).hash();
+
+		// Hash B is in peer's view but not ours.
+		handle_network_update(
+			&mut state,
+			&mut ctx,
+			NetworkBridgeEvent::PeerMessage(
+				peer_a.clone(),
+				awaiting_message(hash_b, vec![pov_hash]),
+			).focus().unwrap(),
+		).await.unwrap();
+
+		// Illegal `awaited` is ignored.
+		assert!(state.peer_state[&peer_a].awaited[&hash_b].is_empty());
+
+		assert_matches!(
+			handle.recv().await,
+			AllMessages::NetworkBridge(
+				NetworkBridgeMessage::ReportPeer(peer, rep)
+			) => {
+				assert_eq!(peer, peer_a);
+				assert_eq!(rep, COST_AWAITED_NOT_IN_VIEW);
+			}
+		);
+	});
+}
+
+#[test]
+fn peer_complete_fetch_leads_to_us_completing_others() {
+	let hash_a: Hash = [0; 32].into();
+
+	let peer_a = PeerId::random();
+	let peer_b = PeerId::random();
+
+	let (pov_send, pov_recv) = oneshot::channel();
+
+	let pov = make_pov(vec![1, 2, 3]);
+	let pov_hash = pov.hash();
+
+	let mut state = State {
+		relay_parent_state: {
+			let mut s = HashMap::new();
+			let mut b = BlockBasedState {
+				known: HashMap::new(),
+				fetching: HashMap::new(),
+				n_validators: 10,
+			};
+
+			// pov is being fetched.
+			b.fetching.insert(pov_hash, vec![pov_send]);
+
+			s.insert(hash_a, b);
+			s
+		},
+		peer_state: {
+			let mut s = HashMap::new();
+
+			s.insert(
+				peer_a.clone(),
+				make_peer_state(vec![(hash_a, vec![])]),
+			);
+
+			// peer B is awaiting peer A's request.
+			s.insert(
+				peer_b.clone(),
+				make_peer_state(vec![(hash_a, vec![pov_hash])]),
+			);
+
+			s
+		},
+		our_view: View(vec![hash_a]),
+		metrics: Default::default(),
+	};
+
+	let pool = sp_core::testing::TaskExecutor::new();
+	let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
+
+	executor::block_on(async move {
+		handle_network_update(
+			&mut state,
+			&mut ctx,
+			NetworkBridgeEvent::PeerMessage(
+				peer_a.clone(),
+				send_pov_message(hash_a, pov_hash, pov.clone()),
+			).focus().unwrap(),
+		).await.unwrap();
+
+		assert_eq!(&*pov_recv.await.unwrap(), &pov);
+
+		assert_matches!(
+			handle.recv().await,
+			AllMessages::NetworkBridge(
+				NetworkBridgeMessage::ReportPeer(peer, rep)
+			) => {
+				assert_eq!(peer, peer_a);
+				assert_eq!(rep, BENEFIT_FRESH_POV);
+			}
+		);
+
+		assert_matches!(
+			handle.recv().await,
+			AllMessages::NetworkBridge(
+				NetworkBridgeMessage::SendValidationMessage(peers, message)
+			) => {
+				assert_eq!(peers, vec![peer_b.clone()]);
+				assert_eq!(
+					message,
+					send_pov_message(hash_a, pov_hash, pov.clone()),
+				);
+			}
+		);
+
+		assert!(!state.peer_state[&peer_b].awaited[&hash_a].contains(&pov_hash));
+	});
+}
+
+#[test]
+fn peer_completing_request_no_longer_awaiting() {
+	let hash_a: Hash = [0; 32].into();
+
+	let peer_a = PeerId::random();
+
+	let (pov_send, pov_recv) = oneshot::channel();
+
+	let pov = make_pov(vec![1, 2, 3]);
+	let pov_hash = pov.hash();
+
+	let mut state = State {
+		relay_parent_state: {
+			let mut s = HashMap::new();
+			let mut b = BlockBasedState {
+				known: HashMap::new(),
+				fetching: HashMap::new(),
+				n_validators: 10,
+			};
+
+			// pov is being fetched.
+			b.fetching.insert(pov_hash, vec![pov_send]);
+
+			s.insert(hash_a, b);
+			s
+		},
+		peer_state: {
+			let mut s = HashMap::new();
+
+			// peer A is registered as awaiting.
+			s.insert(
+				peer_a.clone(),
+				make_peer_state(vec![(hash_a, vec![pov_hash])]),
+			);
+
+			s
+		},
+		our_view: View(vec![hash_a]),
+		metrics: Default::default(),
+	};
+
+	let pool = sp_core::testing::TaskExecutor::new();
+	let (mut ctx, mut handle) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
+
+	executor::block_on(async move {
+		handle_network_update(
+			&mut state,
+			&mut ctx,
+			NetworkBridgeEvent::PeerMessage(
+				peer_a.clone(),
+				send_pov_message(hash_a, pov_hash, pov.clone()),
+			).focus().unwrap(),
+		).await.unwrap();
+
+		assert_eq!(&*pov_recv.await.unwrap(), &pov);
+
+		assert_matches!(
+			handle.recv().await,
+			AllMessages::NetworkBridge(
+				NetworkBridgeMessage::ReportPeer(peer, rep)
+			) => {
+				assert_eq!(peer, peer_a);
+				assert_eq!(rep, BENEFIT_FRESH_POV);
+			}
+		);
+
+		// We received the PoV from peer A, so we do not consider it awaited by peer A anymore.
+		assert!(!state.peer_state[&peer_a].awaited[&hash_a].contains(&pov_hash));
+	});
+}

polkadot/node/network/protocol/Cargo.toml

@@ -9,6 +9,4 @@ description = "Primitives types for the Node-side"
 polkadot-primitives = { path = "../../../primitives" }
 polkadot-node-primitives = { path = "../../primitives" }
 parity-scale-codec = { version = "1.3.4", default-features = false, features = ["derive"] }
-runtime_primitives = { package = "sp-runtime", git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
-sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-network = { git = "https://github.com/paritytech/substrate", branch = "master" }

polkadot/node/network/protocol/src/lib.rs

@@ -16,9 +16,13 @@
 
 //! Network protocol types for parachains.
 
+#![deny(unused_crate_dependencies, unused_results)]
+#![warn(missing_docs)]
+
 use polkadot_primitives::v1::Hash;
 use parity_scale_codec::{Encode, Decode};
 use std::convert::TryFrom;
+use std::fmt;
 
 pub use sc_network::{ReputationChange, PeerId};
 
@@ -32,6 +36,15 @@ pub type ProtocolVersion = u32;
 #[derive(Debug, Clone, Copy, PartialEq)]
 pub struct WrongVariant;
 
+impl fmt::Display for WrongVariant {
+	fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+		write!(formatter, "Wrong message variant")
+	}
+}
+
+impl std::error::Error for WrongVariant {}
+
+
 /// The peer-sets that the network manages. Different subsystems will use different peer-sets.
 #[derive(Debug, Clone, Copy, PartialEq)]
 pub enum PeerSet {

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

@@ -7,13 +7,9 @@ edition = "2018"
 
 [dependencies]
 futures = "0.3.5"
-log = "0.4.8"
-futures-timer = "3.0.2"
-streamunordered = "0.5.1"
+log = "0.4.11"
 polkadot-primitives = { path = "../../../primitives" }
 node-primitives = { package = "polkadot-node-primitives", path = "../../primitives" }
-parity-scale-codec = "1.3.4"
-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" }
 polkadot-node-subsystem-util = { path = "../../subsystem-util" }
@@ -22,7 +18,6 @@ arrayvec = "0.5.1"
 indexmap = "1.4.0"
 
 [dev-dependencies]
-parking_lot = "0.10.0"
 polkadot-node-subsystem-test-helpers = { path = "../../subsystem-test-helpers" }
 assert_matches = "1.3.0"
 sp-keyring = { git = "https://github.com/paritytech/substrate", branch = "master" }

polkadot/node/network/statement-distribution/src/lib.rs

@@ -19,6 +19,9 @@
 //! This is responsible for distributing signed statements about candidate
 //! validity amongst validators.
 
+#![deny(unused_crate_dependencies)]
+#![warn(missing_docs)]
+
 use polkadot_subsystem::{
 	Subsystem, SubsystemResult, SubsystemContext, SpawnedSubsystem,
 	ActiveLeavesUpdate, FromOverseer, OverseerSignal,