NetworkBridge: validator (authorities) discovery api

node/core/runtime-api/src/lib.rs

@@ -120,6 +120,7 @@ fn make_runtime_api_request<Client>(
 		Request::CandidatePendingAvailability(para, sender) =>
 			query!(candidate_pending_availability(para), sender),
 		Request::CandidateEvents(sender) => query!(candidate_events(), sender),
+		Request::ValidatorDiscovery(ids, sender) => query!(validator_discovery(ids), sender),
 	}
 }
 
@@ -169,7 +170,7 @@ mod tests {
 	use polkadot_primitives::v1::{
 		ValidatorId, ValidatorIndex, GroupRotationInfo, CoreState, PersistedValidationData,
 		Id as ParaId, OccupiedCoreAssumption, ValidationData, SessionIndex, ValidationCode,
-		CommittedCandidateReceipt, CandidateEvent,
+		CommittedCandidateReceipt, CandidateEvent, AuthorityDiscoveryId,
 	};
 	use polkadot_node_subsystem_test_helpers as test_helpers;
 	use sp_core::testing::TaskExecutor;
@@ -258,6 +259,10 @@ mod tests {
 			fn candidate_events(&self) -> Vec<CandidateEvent> {
 				self.candidate_events.clone()
 			}
+
+			fn validator_discovery(ids: Vec<ValidatorId>) -> Vec<Option<AuthorityDiscoveryId>> {
+				vec![None; ids.len()]
+			}
 		}
 	}
 

node/network/bridge/Cargo.toml

@@ -5,12 +5,14 @@ authors = ["Parity Technologies <admin@parity.io>"]
 edition = "2018"
 
 [dependencies]
+async-trait = "0.1"
 futures = "0.3.5"
 log = "0.4.8"
 futures-timer = "3.0.2"
 streamunordered = "0.5.1"
 polkadot-primitives = { path = "../../../primitives" }
 parity-scale-codec = "1.3.4"
+sc-authority-discovery = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-network = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "master" }
 polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }

node/network/bridge/src/lib.rs

@@ -20,7 +20,7 @@ use parity_scale_codec::{Encode, Decode};
 use futures::prelude::*;
 use futures::future::BoxFuture;
 use futures::stream::BoxStream;
-use futures::channel::oneshot;
+use futures::channel::{mpsc, oneshot};
 
 use sc_network::Event as NetworkEvent;
 use sp_runtime::ConsensusEngineId;
@@ -34,16 +34,19 @@ use polkadot_subsystem::messages::{
 	BitfieldDistributionMessage, PoVDistributionMessage, StatementDistributionMessage,
 	CollatorProtocolMessage,
 };
-use polkadot_primitives::v1::{Block, Hash, ValidatorId};
+use polkadot_primitives::v1::{AuthorityDiscoveryId, Block, Hash};
 use polkadot_node_network_protocol::{
 	ObservedRole, ReputationChange, PeerId, PeerSet, View, NetworkBridgeEvent, v1 as protocol_v1
 };
 
-use std::collections::hash_map::{HashMap, Entry as HEntry};
+use std::collections::{HashMap, hash_map};
 use std::iter::ExactSizeIterator;
 use std::pin::Pin;
 use std::sync::Arc;
 
+
+mod validator_discovery;
+
 /// The maximum amount of heads a peer is allowed to have in their view at any time.
 ///
 /// We use the same limit to compute the view sent to peers locally.
@@ -188,29 +191,41 @@ impl Network for Arc<sc_network::NetworkService<Block, Hash>> {
 }
 
 /// The network bridge subsystem.
-pub struct NetworkBridge<N>(N);
+pub struct NetworkBridge<N, AD> {
+	network_service: N,
+	authority_discovery_service: AD,
+}
 
-impl<N> NetworkBridge<N> {
-	/// Create a new network bridge subsystem with underlying network service.
+impl<N, AD> NetworkBridge<N, AD> {
+	/// Create a new network bridge subsystem with underlying network service and authority discovery service.
 	///
 	/// This assumes that the network service has had the notifications protocol for the network
 	/// bridge already registered. See [`notifications_protocol_info`](notifications_protocol_info).
-	pub fn new(net_service: N) -> Self {
-		NetworkBridge(net_service)
+	pub fn new(network_service: N, authority_discovery_service: AD) -> Self {
+		NetworkBridge {
+			network_service,
+			authority_discovery_service,
+		}
 	}
 }
 
-impl<Net, Context> Subsystem<Context> for NetworkBridge<Net>
+impl<Net, AD, Context> Subsystem<Context> for NetworkBridge<Net, AD>
 	where
-		Net: Network,
+		Net: Network + validator_discovery::Network,
+		AD: validator_discovery::AuthorityDiscovery,
 		Context: SubsystemContext<Message=NetworkBridgeMessage>,
 {
 	fn start(self, ctx: Context) -> SpawnedSubsystem {
 		// 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(self.0, ctx).map(|_| ()).boxed(),
+			future: run_network(
+				network_service,
+				authority_discovery_service,
+				ctx,
+			).map(|_| ()).boxed(),
 		}
 	}
 }
@@ -224,7 +239,11 @@ struct PeerData {
 enum Action {
 	SendValidationMessage(Vec<PeerId>, protocol_v1::ValidationProtocol),
 	SendCollationMessage(Vec<PeerId>, protocol_v1::CollationProtocol),
-	ConnectToValidators(PeerSet, Vec<ValidatorId>, oneshot::Sender<Vec<(ValidatorId, PeerId)>>),
+	ConnectToValidators {
+		validator_ids: Vec<AuthorityDiscoveryId>,
+		connected: mpsc::Sender<(AuthorityDiscoveryId, PeerId)>,
+		revoke: oneshot::Receiver<()>,
+	},
 	ReportPeer(PeerId, ReputationChange),
 
 	ActiveLeaves(ActiveLeavesUpdate),
@@ -254,8 +273,11 @@ fn action_from_overseer_message(
 				=> Action::SendValidationMessage(peers, msg),
 			NetworkBridgeMessage::SendCollationMessage(peers, msg)
 				=> Action::SendCollationMessage(peers, msg),
-			NetworkBridgeMessage::ConnectToValidators(peer_set, validators, res)
-				=> Action::ConnectToValidators(peer_set, validators, res),
+			NetworkBridgeMessage::ConnectToValidators {
+				validator_ids,
+				connected,
+				revoke,
+			} => Action::ConnectToValidators { validator_ids, connected, revoke },
 		},
 		Ok(FromOverseer::Signal(OverseerSignal::BlockFinalized(_)))
 			=> Action::Nop,
@@ -538,11 +560,16 @@ async fn dispatch_collation_events_to_all<I>(
 	ctx.send_messages(events.into_iter().flat_map(messages_for)).await
 }
 
-async fn run_network<N: Network>(
-	mut net: N,
+async fn run_network<N, AD>(
+	mut network_service: N,
+	mut authority_discovery_service: AD,
 	mut ctx: impl SubsystemContext<Message=NetworkBridgeMessage>,
-) -> SubsystemResult<()> {
-	let mut event_stream = net.event_stream().fuse();
+) -> SubsystemResult<()>
+where
+	N: Network + validator_discovery::Network,
+	AD: validator_discovery::AuthorityDiscovery,
+{
+	let mut event_stream = network_service.event_stream().fuse();
 
 	// Most recent heads are at the back.
 	let mut live_heads: Vec<Hash> = Vec::with_capacity(MAX_VIEW_HEADS);
@@ -551,7 +578,10 @@ async fn run_network<N: Network>(
 	let mut validation_peers: HashMap<PeerId, PeerData> = HashMap::new();
 	let mut collation_peers: HashMap<PeerId, PeerData> = HashMap::new();
 
+	let mut validator_discovery = validator_discovery::Service::<N, AD>::new();
+
 	loop {
+
 		let action = {
 			let subsystem_next = ctx.recv().fuse();
 			let mut net_event_next = event_stream.next().fuse();
@@ -568,31 +598,43 @@ async fn run_network<N: Network>(
 			Action::Abort => return Ok(()),
 
 			Action::SendValidationMessage(peers, msg) => send_message(
-					&mut net,
+					&mut network_service,
 					peers,
 					PeerSet::Validation,
 					WireMessage::ProtocolMessage(msg),
 			).await?,
 
 			Action::SendCollationMessage(peers, msg) => send_message(
-					&mut net,
+					&mut network_service,
 					peers,
 					PeerSet::Collation,
 					WireMessage::ProtocolMessage(msg),
 			).await?,
 
-			Action::ConnectToValidators(_peer_set, _validators, _res) => {
-				// TODO: https://github.com/paritytech/polkadot/issues/1461
-			}
+			Action::ConnectToValidators {
+				validator_ids,
+				connected,
+				revoke,
+			} => {
+				let (ns, ads) = validator_discovery.on_request(
+					validator_ids,
+					connected,
+					revoke,
+					network_service,
+					authority_discovery_service,
+				).await;
+				network_service = ns;
+				authority_discovery_service = ads;
+			},
 
-			Action::ReportPeer(peer, rep) => net.report_peer(peer, rep).await?,
+			Action::ReportPeer(peer, rep) => network_service.report_peer(peer, rep).await?,
 
 			Action::ActiveLeaves(ActiveLeavesUpdate { activated, deactivated }) => {
 				live_heads.extend(activated);
 				live_heads.retain(|h| !deactivated.contains(h));
 
 				update_view(
-					&mut net,
+					&mut network_service,
 					&mut ctx,
 					&live_heads,
 					&mut local_view,
@@ -607,9 +649,11 @@ async fn run_network<N: Network>(
 					PeerSet::Collation => &mut collation_peers,
 				};
 
+				validator_discovery.on_peer_connected(&peer, &mut authority_discovery_service).await;
+
 				match peer_map.entry(peer.clone()) {
-					HEntry::Occupied(_) => continue,
-					HEntry::Vacant(vacant) => {
+					hash_map::Entry::Occupied(_) => continue,
+					hash_map::Entry::Vacant(vacant) => {
 						vacant.insert(PeerData {
 							view: View(Vec::new()),
 						});
@@ -650,6 +694,8 @@ async fn run_network<N: Network>(
 					PeerSet::Collation => &mut collation_peers,
 				};
 
+				validator_discovery.on_peer_disconnected(&peer, &mut authority_discovery_service).await;
+
 				if peer_map.remove(&peer).is_some() {
 					let res = match peer_set {
 						PeerSet::Validation => dispatch_validation_event_to_all(
@@ -677,7 +723,7 @@ async fn run_network<N: Network>(
 						peer.clone(),
 						&mut validation_peers,
 						v_messages,
-						&mut net,
+						&mut network_service,
 					).await?;
 
 					if let Err(e) = dispatch_validation_events_to_all(
@@ -697,7 +743,7 @@ async fn run_network<N: Network>(
 						peer.clone(),
 						&mut collation_peers,
 						c_messages,
-						&mut net,
+						&mut network_service,
 					).await?;
 
 					if let Err(e) = dispatch_collation_events_to_all(
@@ -716,20 +762,24 @@ async fn run_network<N: Network>(
 	}
 }
 
+
 #[cfg(test)]
 mod tests {
 	use super::*;
 	use futures::channel::mpsc;
 	use futures::executor;
 
 	use std::sync::Arc;
+	use std::collections::HashSet;
+	use async_trait::async_trait;
 	use parking_lot::Mutex;
 	use assert_matches::assert_matches;
 
 	use polkadot_subsystem::messages::{StatementDistributionMessage, BitfieldDistributionMessage};
 	use polkadot_node_subsystem_test_helpers::{
 		SingleItemSink, SingleItemStream, TestSubsystemContextHandle,
 	};
+	use sc_network::Multiaddr;
 	use sp_keyring::Sr25519Keyring;
 
 	// The subsystem's view of the network - only supports a single call to `event_stream`.
@@ -738,6 +788,8 @@ mod tests {
 		action_tx: mpsc::UnboundedSender<NetworkAction>,
 	}
 
+	struct TestAuthorityDiscovery;
+
 	// The test's view of the network. This receives updates from the subsystem in the form
 	// of `NetworkAction`s.
 	struct TestNetworkHandle {
@@ -748,6 +800,7 @@ mod tests {
 	fn new_test_network() -> (
 		TestNetwork,
 		TestNetworkHandle,
+		TestAuthorityDiscovery,
 	) {
 		let (net_tx, net_rx) = polkadot_node_subsystem_test_helpers::single_item_sink();
 		let (action_tx, action_rx) = mpsc::unbounded();
@@ -761,6 +814,7 @@ mod tests {
 				action_rx,
 				net_tx,
 			},
+			TestAuthorityDiscovery,
 		)
 	}
 
@@ -786,6 +840,23 @@ mod tests {
 		}
 	}
 
+	impl validator_discovery::Network for TestNetwork {
+		fn set_priority_group(&self, _group_id: String, _multiaddresses: HashSet<Multiaddr>) -> Result<(), String> {
+			Ok(())
+		}
+	}
+
+	#[async_trait]
+	impl validator_discovery::AuthorityDiscovery for TestAuthorityDiscovery {
+		async fn get_addresses_by_authority_id(&mut self, _authority: AuthorityDiscoveryId) -> Option<Vec<Multiaddr>> {
+			None
+		}
+
+		async fn get_authority_id_by_peer_id(&mut self, _peer_id: PeerId) -> Option<AuthorityDiscoveryId> {
+			None
+		}
+	}
+
 	impl TestNetworkHandle {
 		// Get the next network action.
 		async fn next_network_action(&mut self) -> NetworkAction {
@@ -842,11 +913,12 @@ mod tests {
 
 	fn test_harness<T: Future<Output=()>>(test: impl FnOnce(TestHarness) -> T) {
 		let pool = sp_core::testing::TaskExecutor::new();
-		let (network, network_handle) = new_test_network();
+		let (network, network_handle, discovery) = new_test_network();
 		let (context, virtual_overseer) = polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
 
 		let network_bridge = run_network(
 			network,
+			discovery,
 			context,
 		)
 			.map_err(|_| panic!("subsystem execution failed"))