Dispute distribution implementation

Cargo.toml

@@ -63,6 +63,7 @@ members = [
 	"node/network/availability-recovery",
 	"node/network/collator-protocol",
 	"node/network/gossip-support",
+	"node/network/dispute-distribution",
 	"node/overseer",
 	"node/malus",
 	"node/primitives",

node/network/availability-distribution/src/pov_requester/mod.rs

@@ -19,10 +19,8 @@
 use futures::{FutureExt, channel::oneshot, future::BoxFuture};
 
 use polkadot_subsystem::jaeger;
-use polkadot_node_network_protocol::{
-	request_response::{OutgoingRequest, Recipient, request::{RequestError, Requests},
-	v1::{PoVFetchingRequest, PoVFetchingResponse}}
-};
+use polkadot_node_network_protocol::request_response::{OutgoingRequest, Recipient, request::{RequestError, Requests},
+	v1::{PoVFetchingRequest, PoVFetchingResponse}};
 use polkadot_primitives::v1::{
 	CandidateHash, Hash, ValidatorIndex,
 };
@@ -49,7 +47,7 @@ pub async fn fetch_pov<Context>(
 where
 	Context: SubsystemContext,
 {
-	let info = &runtime.get_session_info(ctx, parent).await?.session_info;
+	let info = &runtime.get_session_info(ctx.sender(), parent).await?.session_info;
 	let authority_id = info.discovery_keys.get(from_validator.0 as usize)
 		.ok_or(NonFatal::InvalidValidatorIndex)?
 		.clone();

node/network/availability-distribution/src/requester/session_cache.rs

@@ -106,7 +106,7 @@ impl SessionCache {
 		Context: SubsystemContext,
 		F: FnOnce(&SessionInfo) -> R,
 	{
-		let session_index = runtime.get_session_index(ctx, parent).await?;
+		let session_index = runtime.get_session_index(ctx.sender(), parent).await?;
 
 		if let Some(o_info) = self.session_info_cache.get(&session_index) {
 			tracing::trace!(target: LOG_TARGET, session_index, "Got session from lru");
@@ -183,7 +183,7 @@ impl SessionCache {
 	where
 		Context: SubsystemContext,
 	{
-		let info = runtime.get_session_info_by_index(ctx, parent, session_index).await?;
+		let info = runtime.get_session_info_by_index(ctx.sender(), parent, session_index).await?;
 
 		let discovery_keys = info.session_info.discovery_keys.clone();
 		let mut validator_groups = info.session_info.validator_groups.clone();

node/network/bridge/Cargo.toml

@@ -10,7 +10,6 @@ futures = "0.3.15"
 tracing = "0.1.26"
 polkadot-primitives = { path = "../../../primitives" }
 parity-scale-codec = { version = "2.0.0", default-features = false, features = ["derive"] }
-sc-authority-discovery = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-network = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-consensus = { git = "https://github.com/paritytech/substrate", branch = "master" }
 polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }

node/network/bridge/src/lib.rs

@@ -24,11 +24,15 @@ use parity_scale_codec::{Encode, Decode};
 use parking_lot::Mutex;
 use futures::prelude::*;
 use futures::stream::BoxStream;
+use polkadot_subsystem::ActiveLeavesUpdate;
+use polkadot_subsystem::FromOverseer;
+use polkadot_subsystem::OverseerSignal;
+use polkadot_subsystem::messages::DisputeDistributionMessage;
 use sc_network::Event as NetworkEvent;
 use sp_consensus::SyncOracle;
 
 use polkadot_subsystem::{
-	ActivatedLeaf, ActiveLeavesUpdate, FromOverseer, OverseerSignal, SpawnedSubsystem,
+	ActivatedLeaf, SpawnedSubsystem,
 	Subsystem, SubsystemContext, SubsystemError, SubsystemResult, SubsystemSender,
 	messages::StatementDistributionMessage
 };
@@ -48,7 +52,8 @@ use polkadot_node_subsystem_util::metrics::{self, prometheus};
 /// To be added to [`NetworkConfiguration::extra_sets`].
 pub use polkadot_node_network_protocol::peer_set::{peer_sets_info, IsAuthority};
 
-use std::collections::{HashMap, hash_map, HashSet};
+use std::collections::HashSet;
+use std::collections::{HashMap, hash_map};
 use std::sync::Arc;
 
 mod validator_discovery;
@@ -57,12 +62,14 @@ mod validator_discovery;
 ///
 /// Defines the `Network` trait with an implementation for an `Arc<NetworkService>`.
 mod network;
-use network::{Network, send_message, get_peer_id_by_authority_id};
+use network::{Network, send_message};
 
 /// Request multiplexer for combining the multiple request sources into a single `Stream` of `AllMessages`.
 mod multiplexer;
 pub use multiplexer::RequestMultiplexer;
 
+use crate::network::get_peer_id_by_authority_id;
+
 #[cfg(test)]
 mod tests;
 
@@ -295,7 +302,7 @@ impl<N, AD> NetworkBridge<N, AD> {
 impl<Net, AD, Context> Subsystem<Context> for NetworkBridge<Net, AD>
 	where
 		Net: Network + Sync,
-		AD: validator_discovery::AuthorityDiscovery,
+		AD: validator_discovery::AuthorityDiscovery + Clone,
 		Context: SubsystemContext<Message=NetworkBridgeMessage>,
 {
 	fn start(mut self, ctx: Context) -> SpawnedSubsystem {
@@ -364,7 +371,7 @@ async fn handle_subsystem_messages<Context, N, AD>(
 where
 	Context: SubsystemContext<Message = NetworkBridgeMessage>,
 	N: Network,
-	AD: validator_discovery::AuthorityDiscovery,
+	AD: validator_discovery::AuthorityDiscovery + Clone,
 {
 	// This is kept sorted, descending, by block number.
 	let mut live_heads: Vec<ActivatedLeaf> = Vec::with_capacity(MAX_VIEW_HEADS);
@@ -556,6 +563,15 @@ where
 						network_service = ns;
 						authority_discovery_service = ads;
 					}
+					NetworkBridgeMessage::GetAuthorityDiscoveryService(tx) => {
+						if let Err(err) = tx.send(Box::new(authority_discovery_service.clone())) {
+							tracing::debug!(
+								target: LOG_TARGET,
+								?err,
+								"Answering `GetAuthorityDiscoveryService` request failed."
+							);
+						}
+					}
 					NetworkBridgeMessage::NewGossipTopology {
 						our_neighbors,
 					} => {
@@ -861,7 +877,7 @@ async fn run_network<N, AD>(
 ) -> SubsystemResult<()>
 where
 	N: Network,
-	AD: validator_discovery::AuthorityDiscovery,
+	AD: validator_discovery::AuthorityDiscovery + Clone,
 {
 	let shared = Shared::default();
 
@@ -877,6 +893,10 @@ where
 		.get_statement_fetching()
 		.expect("Gets initialized, must be `Some` on startup. qed.");
 
+	let dispute_receiver = request_multiplexer
+		.get_dispute_sending()
+		.expect("Gets initialized, must be `Some` on startup. qed.");
+
 	let (remote, network_event_handler) = handle_network_messages(
 		ctx.sender().clone(),
 		network_service.clone(),
@@ -889,6 +909,9 @@ where
 
 	ctx.spawn("network-bridge-network-worker", Box::pin(remote)).await?;
 
+	ctx.send_message(AllMessages::DisputeDistribution(
+		DisputeDistributionMessage::DisputeSendingReceiver(dispute_receiver)
+	)).await;
 	ctx.send_message(AllMessages::StatementDistribution(
 		StatementDistributionMessage::StatementFetchingReceiver(statement_receiver)
 	)).await;

node/network/bridge/src/multiplexer.rs

@@ -42,6 +42,7 @@ use polkadot_subsystem::messages::AllMessages;
 pub struct RequestMultiplexer {
 	receivers: Vec<(Protocol, mpsc::Receiver<network::IncomingRequest>)>,
 	statement_fetching: Option<mpsc::Receiver<network::IncomingRequest>>,
+	dispute_sending: Option<mpsc::Receiver<network::IncomingRequest>>,
 	next_poll: usize,
 }
 
@@ -68,19 +69,31 @@ impl RequestMultiplexer {
 			})
 			.unzip();
 
+		// Ok this code is ugly as hell, it is also a hack, see #2842. But it works and is executed
+		// on startup so, if anything is wrong here it will be noticed immediately.
 		let index = receivers.iter().enumerate().find_map(|(i, (p, _))|
 			if let Protocol::StatementFetching = p {
 				Some(i)
 			} else {
 				None
 			}
-		).expect("Statement fetching must be registered. qed.");
+		).expect("Statement fetching must be registered.");
 		let statement_fetching = Some(receivers.remove(index).1);
 
+		let index = receivers.iter().enumerate().find_map(|(i, (p, _))|
+			if let Protocol::DisputeSending = p {
+				Some(i)
+			} else {
+				None
+			}
+		).expect("Dispute sending must be registered.");
+		let dispute_sending = Some(receivers.remove(index).1);
+
 		(
 			Self {
 				receivers,
 				statement_fetching,
+                dispute_sending,
 				next_poll: 0,
 			},
 			cfgs,
@@ -93,6 +106,13 @@ impl RequestMultiplexer {
 	pub fn get_statement_fetching(&mut self) -> Option<mpsc::Receiver<network::IncomingRequest>> {
 		std::mem::take(&mut self.statement_fetching)
 	}
+
+	/// Get the receiver for handling dispute sending requests.
+	///
+	/// This function will only return `Some` once.
+	pub fn get_dispute_sending(&mut self) -> Option<mpsc::Receiver<network::IncomingRequest>> {
+		std::mem::take(&mut self.dispute_sending)
+	}
 }
 
 impl Stream for RequestMultiplexer {
@@ -174,6 +194,9 @@ fn multiplex_single(
 		Protocol::StatementFetching => {
 			panic!("Statement fetching requests are handled directly. qed.");
 		}
+		Protocol::DisputeSending => {
+			panic!("Dispute sending request are handled directly. qed.");
+		}
 	};
 	Ok(r)
 }

node/network/bridge/src/tests.rs

@@ -67,7 +67,7 @@ struct TestNetwork {
 	_req_configs: Vec<RequestResponseConfig>,
 }
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 struct TestAuthorityDiscovery;
 
 // The test's view of the network. This receives updates from the subsystem in the form
@@ -688,6 +688,12 @@ fn peer_view_updates_sent_via_overseer() {
 
 		let view = view![Hash::repeat_byte(1)];
 
+		assert_matches!(
+			virtual_overseer.recv().await,
+			AllMessages::DisputeDistribution(
+				DisputeDistributionMessage::DisputeSendingReceiver(_)
+			)
+		);
 		assert_matches!(
 			virtual_overseer.recv().await,
 			AllMessages::StatementDistribution(
@@ -740,6 +746,12 @@ fn peer_messages_sent_via_overseer() {
 			ObservedRole::Full,
 		).await;
 
+		assert_matches!(
+			virtual_overseer.recv().await,
+			AllMessages::DisputeDistribution(
+				DisputeDistributionMessage::DisputeSendingReceiver(_)
+			)
+		);
 		assert_matches!(
 			virtual_overseer.recv().await,
 			AllMessages::StatementDistribution(
@@ -812,6 +824,12 @@ fn peer_disconnect_from_just_one_peerset() {
 		network_handle.connect_peer(peer.clone(), PeerSet::Validation, ObservedRole::Full).await;
 		network_handle.connect_peer(peer.clone(), PeerSet::Collation, ObservedRole::Full).await;
 
+		assert_matches!(
+			virtual_overseer.recv().await,
+			AllMessages::DisputeDistribution(
+				DisputeDistributionMessage::DisputeSendingReceiver(_)
+			)
+		);
 		assert_matches!(
 			virtual_overseer.recv().await,
 			AllMessages::StatementDistribution(
@@ -894,6 +912,12 @@ fn relays_collation_protocol_messages() {
 		let peer_a = PeerId::random();
 		let peer_b = PeerId::random();
 
+		assert_matches!(
+			virtual_overseer.recv().await,
+			AllMessages::DisputeDistribution(
+				DisputeDistributionMessage::DisputeSendingReceiver(_)
+			)
+		);
 		assert_matches!(
 			virtual_overseer.recv().await,
 			AllMessages::StatementDistribution(
@@ -992,6 +1016,12 @@ fn different_views_on_different_peer_sets() {
 		network_handle.connect_peer(peer.clone(), PeerSet::Validation, ObservedRole::Full).await;
 		network_handle.connect_peer(peer.clone(), PeerSet::Collation, ObservedRole::Full).await;
 
+		assert_matches!(
+			virtual_overseer.recv().await,
+			AllMessages::DisputeDistribution(
+				DisputeDistributionMessage::DisputeSendingReceiver(_)
+			)
+		);
 		assert_matches!(
 			virtual_overseer.recv().await,
 			AllMessages::StatementDistribution(
@@ -1153,6 +1183,12 @@ fn send_messages_to_peers() {
 		network_handle.connect_peer(peer.clone(), PeerSet::Validation, ObservedRole::Full).await;
 		network_handle.connect_peer(peer.clone(), PeerSet::Collation, ObservedRole::Full).await;
 
+		assert_matches!(
+			virtual_overseer.recv().await,
+			AllMessages::DisputeDistribution(
+				DisputeDistributionMessage::DisputeSendingReceiver(_)
+			)
+		);
 		assert_matches!(
 			virtual_overseer.recv().await,
 			AllMessages::StatementDistribution(
@@ -1278,7 +1314,8 @@ fn spread_event_to_subsystems_is_up_to_date() {
 			AllMessages::ApprovalDistribution(_) => { cnt += 1; }
 			AllMessages::GossipSupport(_) => unreachable!("Not interested in network events"),
 			AllMessages::DisputeCoordinator(_) => unreachable!("Not interested in network events"),
-			AllMessages::DisputeParticipation(_) => unreachable!("Not interetsed in network events"),
+			AllMessages::DisputeParticipation(_) => unreachable!("Not interested in network events"),
+			AllMessages::DisputeDistribution(_) => unreachable!("Not interested in network events"),
 			// Add variants here as needed, `{ cnt += 1; }` for those that need to be
 			// notified, `unreachable!()` for those that should not.
 		}
@@ -1323,6 +1360,12 @@ fn our_view_updates_decreasing_order_and_limited_to_max() {
 			0,
 		);
 
+		assert_matches!(
+			virtual_overseer.recv().await,
+			AllMessages::DisputeDistribution(
+				DisputeDistributionMessage::DisputeSendingReceiver(_)
+			)
+		);
 		assert_matches!(
 			virtual_overseer.recv().await,
 			AllMessages::StatementDistribution(

node/network/bridge/src/validator_discovery.rs

@@ -21,37 +21,16 @@ use crate::Network;
 use core::marker::PhantomData;
 use std::collections::HashSet;
 
-use async_trait::async_trait;
 use futures::channel::oneshot;
 
 use sc_network::multiaddr::Multiaddr;
-use sc_authority_discovery::Service as AuthorityDiscoveryService;
-use polkadot_node_network_protocol::PeerId;
+
 use polkadot_primitives::v1::AuthorityDiscoveryId;
 use polkadot_node_network_protocol::peer_set::{PeerSet, PerPeerSet};
+pub use polkadot_node_network_protocol::authority_discovery::AuthorityDiscovery;
 
 const LOG_TARGET: &str = "parachain::validator-discovery";
 
-/// An abstraction over the authority discovery service.
-#[async_trait]
-pub trait AuthorityDiscovery: Send + Clone + 'static {
-	/// Get the addresses for the given [`AuthorityId`] from the local address cache.
-	async fn get_addresses_by_authority_id(&mut self, authority: AuthorityDiscoveryId) -> Option<Vec<Multiaddr>>;
-	/// Get the [`AuthorityId`] for the given [`PeerId`] from the local address cache.
-	async fn get_authority_id_by_peer_id(&mut self, peer_id: PeerId) -> Option<AuthorityDiscoveryId>;
-}
-
-#[async_trait]
-impl AuthorityDiscovery for AuthorityDiscoveryService {
-	async fn get_addresses_by_authority_id(&mut self, authority: AuthorityDiscoveryId) -> Option<Vec<Multiaddr>> {
-		AuthorityDiscoveryService::get_addresses_by_authority_id(self, authority).await
-	}
-
-	async fn get_authority_id_by_peer_id(&mut self, peer_id: PeerId) -> Option<AuthorityDiscoveryId> {
-		AuthorityDiscoveryService::get_authority_id_by_peer_id(self, peer_id).await
-	}
-}
-
 pub(super) struct Service<N, AD> {
 	state: PerPeerSet<StatePerPeerSet>,
 	// PhantomData used to make the struct generic instead of having generic methods
@@ -147,9 +126,10 @@ mod tests {
 
 	use std::{borrow::Cow, collections::HashMap};
 	use futures::stream::BoxStream;
+	use async_trait::async_trait;
 	use sc_network::{Event as NetworkEvent, IfDisconnected};
 	use sp_keyring::Sr25519Keyring;
-	use polkadot_node_network_protocol::request_response::request::Requests;
+	use polkadot_node_network_protocol::{PeerId, request_response::request::Requests};
 
 	fn new_service() -> Service<TestNetwork, TestAuthorityDiscovery> {
 		Service::new()
@@ -164,7 +144,7 @@ mod tests {
 		peers_set: HashSet<Multiaddr>,
 	}
 
-	#[derive(Default, Clone)]
+	#[derive(Default, Clone, Debug)]
 	struct TestAuthorityDiscovery {
 		by_authority_id: HashMap<AuthorityDiscoveryId, Multiaddr>,
 		by_peer_id: HashMap<PeerId, AuthorityDiscoveryId>,