Rework priority groups

client/authority-discovery/src/worker.rs

@@ -226,7 +226,7 @@ where
 				},
 				// Set peerset priority group to a new random set of addresses.
 				_ = self.priority_group_set_interval.next().fuse() => {
-					if let Err(e) = self.set_priority_group().await {
+					if let Err(e) = self.set_peers_set().await {
 						error!(
 							target: LOG_TARGET,
 							"Failed to set priority group: {:?}", e,
@@ -583,7 +583,8 @@ where
 
 	/// Set the peer set 'authority' priority group to a new random set of
 	/// [`Multiaddr`]s.
-	async fn set_priority_group(&self) -> Result<()> {
+	async fn set_peers_set(&self) -> Result<()> {
+		// TODO: no longer pass only a random subset, but the entire thing
 		let addresses = self.addr_cache.get_random_subset();
 
 		if addresses.is_empty() {
@@ -604,7 +605,7 @@ where
 		);
 
 		self.network
-			.set_priority_group(
+			.set_peers_set(
 				AUTHORITIES_PRIORITY_GROUP_NAME.to_string(),
 				addresses.into_iter().collect(),
 			).await
@@ -620,7 +621,7 @@ where
 #[async_trait]
 pub trait NetworkProvider: NetworkStateInfo {
 	/// Modify a peerset priority group.
-	async fn set_priority_group(
+	async fn set_peers_set(
 		&self,
 		group_id: String,
 		peers: HashSet<libp2p::Multiaddr>,
@@ -639,12 +640,12 @@ where
 	B: BlockT + 'static,
 	H: ExHashT,
 {
-	async fn set_priority_group(
+	async fn set_peers_set(
 		&self,
 		group_id: String,
 		peers: HashSet<libp2p::Multiaddr>,
 	) -> std::result::Result<(), String> {
-		self.set_priority_group(group_id, peers).await
+		self.set_peers_set(group_id, peers).await
 	}
 	fn put_value(&self, key: libp2p::kad::record::Key, value: Vec<u8>) {
 		self.put_value(key, value)

client/authority-discovery/src/worker/tests.rs

@@ -156,7 +156,7 @@ impl Default for TestNetwork {
 
 #[async_trait]
 impl NetworkProvider for TestNetwork {
-	async fn set_priority_group(
+	async fn set_peers_set(
 		&self,
 		group_id: String,
 		peers: HashSet<Multiaddr>,
@@ -366,7 +366,7 @@ fn publish_discover_cycle() {
 		// Make authority discovery handle the event.
 		worker.handle_dht_event(dht_event).await;
 
-		worker.set_priority_group().await.unwrap();
+		worker.set_peers_set().await.unwrap();
 
 		// Expect authority discovery to set the priority set.
 		assert_eq!(network.set_priority_group_call.lock().unwrap().len(), 1);

client/network/src/behaviour.rs

@@ -485,7 +485,7 @@ impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<DiscoveryOut>
 				// implementation for `PeerInfoEvent`.
 			}
 			DiscoveryOut::Discovered(peer_id) => {
-				self.substrate.add_discovered_nodes(iter::once(peer_id));
+				self.substrate.add_discovered_nodes(sc_peerset::SetId::from(0), iter::once(peer_id));  // TODO: correct set?
 			}
 			DiscoveryOut::ValueFound(results, duration) => {
 				self.events.push_back(BehaviourOut::Dht(DhtEvent::ValueFound(results), duration));

client/network/src/config.rs

@@ -370,16 +370,12 @@ pub struct NetworkConfiguration {
 	pub boot_nodes: Vec<MultiaddrWithPeerId>,
 	/// The node key configuration, which determines the node's network identity keypair.
 	pub node_key: NodeKeyConfig,
-	/// List of names of notifications protocols that the node supports.
-	pub notifications_protocols: Vec<Cow<'static, str>>,
 	/// List of request-response protocols that the node supports.
 	pub request_response_protocols: Vec<RequestResponseConfig>,
-	/// Maximum allowed number of incoming connections.
-	pub in_peers: u32,
-	/// Number of outgoing connections we're trying to maintain.
-	pub out_peers: u32,
-	/// List of reserved node addresses.
-	pub reserved_nodes: Vec<MultiaddrWithPeerId>,
+	/// Configuration for the default set of nodes used for block syncing and transactions.
+	pub default_peers_set: SetConfig,
+	/// Configuration for extra sets of nodes.
+	pub extra_sets: Vec<NonDefaultSetConfig>,
 	/// The non-reserved peer mode.
 	pub non_reserved_mode: NonReservedPeerMode,
 	/// Client identifier. Sent over the wire for debugging purposes.
@@ -411,11 +407,14 @@ impl NetworkConfiguration {
 			public_addresses: Vec::new(),
 			boot_nodes: Vec::new(),
 			node_key,
-			notifications_protocols: Vec::new(),
 			request_response_protocols: Vec::new(),
-			in_peers: 25,
-			out_peers: 75,
-			reserved_nodes: Vec::new(),
+			default_peers_set: SetConfig {
+				optional_notifications_protocol: Vec::new(),
+				in_peers: 25,
+				out_peers: 75,
+				reserved_nodes: Vec::new(),
+			},
+			extra_sets: Vec::new(),
 			non_reserved_mode: NonReservedPeerMode::Accept,
 			client_version: client_version.into(),
 			node_name: node_name.into(),
@@ -469,6 +468,30 @@ impl NetworkConfiguration {
 	}
 }
 
+/// Configuration for a set of nodes.
+#[derive(Clone, Debug)]
+pub struct SetConfig {
+	/// Additional notification protocols that will be attempted, but whose success isn't
+	/// mandatory.
+	pub optional_notifications_protocol: Vec<Cow<'static, str>>,
+	/// Maximum allowed number of incoming connections.
+	pub in_peers: u32,
+	/// Number of outgoing connections we're trying to maintain.
+	pub out_peers: u32,
+	/// List of reserved node addresses.
+	pub reserved_nodes: Vec<MultiaddrWithPeerId>,
+}
+
+/// Extension to [`SetConfig`] for sets that aren't the default set.
+#[derive(Clone, Debug)]
+pub struct NonDefaultSetConfig {
+	/// Name of the main notifications protocols of this set. A substream on this set will be
+	/// considered established once this protocol is open.
+	pub main_notifications_protocol: Cow<'static, str>,
+	/// Base configuration.
+	pub set_config: SetConfig,
+}
+
 /// Configuration for the transport layer.
 #[derive(Clone, Debug)]
 pub enum TransportConfig {

client/network/src/protocol.rs

@@ -371,7 +371,7 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 
 		let important_peers = {
 			let mut imp_p = HashSet::new();
-			for reserved in peerset_config.priority_groups.iter().flat_map(|(_, l)| l.iter()) {
+			for reserved in peerset_config.sets.iter().flat_map(|s| s.reserved_nodes.iter()) {
 				imp_p.insert(reserved.clone());
 			}
 			imp_p.shrink_to_fit();

client/network/src/protocol/generic_proto/behaviour.rs

@@ -479,7 +479,7 @@ impl GenericProto {
 				timer: _
 			} => {
 				debug!(target: "sub-libp2p", "PSM <= Dropped({:?})", peer_id);
-				self.peerset.dropped(peer_id.clone());
+				self.peerset.dropped("main", peer_id.clone());
 				let backoff_until = Some(if let Some(ban) = ban {
 					cmp::max(timer_deadline, Instant::now() + ban)
 				} else {
@@ -496,7 +496,7 @@ impl GenericProto {
 			// If relevant, the external API is instantly notified.
 			PeerState::Enabled { mut connections } => {
 				debug!(target: "sub-libp2p", "PSM <= Dropped({:?})", peer_id);
-				self.peerset.dropped(peer_id.clone());
+				self.peerset.dropped("main", peer_id.clone());
 
 				if connections.iter().any(|(_, s)| matches!(s, ConnectionState::Open(_))) {
 					debug!(target: "sub-libp2p", "External API <= Closed({})", peer_id);
@@ -605,12 +605,16 @@ impl GenericProto {
 		}
 	}
 
-	/// Notify the behaviour that we have learned about the existence of nodes.
+	/// Notify the behaviour that we have learned that the given nodes belong to the given set.
 	///
 	/// Can be called multiple times with the same `PeerId`s.
-	pub fn add_discovered_nodes(&mut self, peer_ids: impl Iterator<Item = PeerId>) {
+	pub fn add_discovered_nodes(
+		&mut self,
+		set_id: sc_peerset::SetId,
+		peer_ids: impl Iterator<Item = PeerId>
+	) {
 		let local_peer_id = &self.local_peer_id;
-		self.peerset.discovered(peer_ids.filter_map(|peer_id| {
+		self.peerset.discovered(set_id, peer_ids.filter_map(|peer_id| {
 			if peer_id == *local_peer_id {
 				error!(
 					target: "sub-libp2p",
@@ -968,7 +972,7 @@ impl GenericProto {
 				Some(PeerState::Enabled { .. }) => {}
 				_ => {
 					debug!(target: "sub-libp2p", "PSM <= Dropped({:?})", incoming.peer_id);
-					self.peerset.dropped(incoming.peer_id);
+					self.peerset.dropped("main", incoming.peer_id);
 				},
 			}
 			return
@@ -1204,7 +1208,7 @@ impl NetworkBehaviour for GenericProto {
 
 				if connections.is_empty() {
 					debug!(target: "sub-libp2p", "PSM <= Dropped({})", peer_id);
-					self.peerset.dropped(peer_id.clone());
+					self.peerset.dropped("main", peer_id.clone());
 					*entry.get_mut() = PeerState::Backoff { timer, timer_deadline };
 
 				} else {
@@ -1335,7 +1339,7 @@ impl NetworkBehaviour for GenericProto {
 
 				if connections.is_empty() {
 					debug!(target: "sub-libp2p", "PSM <= Dropped({})", peer_id);
-					self.peerset.dropped(peer_id.clone());
+					self.peerset.dropped("main", peer_id.clone());
 					let ban_dur = Uniform::new(5, 10).sample(&mut rand::thread_rng());
 
 					let delay_id = self.next_delay_id;
@@ -1356,7 +1360,7 @@ impl NetworkBehaviour for GenericProto {
 					matches!(s, ConnectionState::Opening | ConnectionState::Open(_)))
 				{
 					debug!(target: "sub-libp2p", "PSM <= Dropped({:?})", peer_id);
-					self.peerset.dropped(peer_id.clone());
+					self.peerset.dropped("main", peer_id.clone());
 
 					*entry.get_mut() = PeerState::Disabled {
 						connections,
@@ -1406,7 +1410,7 @@ impl NetworkBehaviour for GenericProto {
 					debug!(target: "sub-libp2p", "Libp2p => Dial failure for {:?}", peer_id);
 
 					debug!(target: "sub-libp2p", "PSM <= Dropped({:?})", peer_id);
-					self.peerset.dropped(peer_id.clone());
+					self.peerset.dropped("main", peer_id.clone());
 
 					let now = Instant::now();
 					let ban_duration = match st {
@@ -1544,7 +1548,7 @@ impl NetworkBehaviour for GenericProto {
 
 								debug!(target: "sub-libp2p", "PSM <= Incoming({}, {:?}).",
 									source, incoming_id);
-								self.peerset.incoming(source.clone(), incoming_id);
+								self.peerset.incoming("main", source.clone(), incoming_id);
 								self.incoming.push(IncomingPeer {
 									peer_id: source.clone(),
 									alive: true,
@@ -1588,7 +1592,7 @@ impl NetworkBehaviour for GenericProto {
 
 								debug!(target: "sub-libp2p", "PSM <= Incoming({}, {:?}).",
 									source, incoming_id);
-								self.peerset.incoming(source.clone(), incoming_id);
+								self.peerset.incoming("main", source.clone(), incoming_id);
 								self.incoming.push(IncomingPeer {
 									peer_id: source.clone(),
 									alive: true,
@@ -1700,7 +1704,7 @@ impl NetworkBehaviour for GenericProto {
 							// List of open connections wasn't empty before but now it is.
 							if !connections.iter().any(|(_, s)| matches!(s, ConnectionState::Opening)) {
 								debug!(target: "sub-libp2p", "PSM <= Dropped({:?})", source);
-								self.peerset.dropped(source.clone());
+								self.peerset.dropped("main", source.clone());
 								*entry.into_mut() = PeerState::Disabled {
 									connections, backoff_until: None
 								};
@@ -1858,7 +1862,7 @@ impl NetworkBehaviour for GenericProto {
 							matches!(s, ConnectionState::Opening | ConnectionState::Open(_)))
 						{
 							debug!(target: "sub-libp2p", "PSM <= Dropped({:?})", source);
-							self.peerset.dropped(source.clone());
+							self.peerset.dropped("main", source.clone());
 
 							*entry.into_mut() = PeerState::Disabled {
 								connections,
@@ -1926,18 +1930,18 @@ impl NetworkBehaviour for GenericProto {
 				}
 			}
 
-			NotifsHandlerOut::Notification { protocol_name, message } => {
+			NotifsHandlerOut::Notification { protocol, message } => {
 				if self.is_open(&source) {
 					trace!(
 						target: "sub-libp2p",
 						"Handler({:?}) => Notification({:?})",
 						source,
-						protocol_name,
+						protocol,
 					);
-					trace!(target: "sub-libp2p", "External API <= Message({:?}, {:?})", protocol_name, source);
+					trace!(target: "sub-libp2p", "External API <= Message({:?}, {:?})", protocol, source);
 					let event = GenericProtoOut::Notification {
 						peer_id: source,
-						protocol_name,
+						protocol_name: protocol,
 						message,
 					};
 
@@ -1947,7 +1951,7 @@ impl NetworkBehaviour for GenericProto {
 						target: "sub-libp2p",
 						"Handler({:?}) => Post-close notification({:?})",
 						source,
-						protocol_name,
+						protocol,
 					);
 				}
 			}
@@ -1978,11 +1982,11 @@ impl NetworkBehaviour for GenericProto {
 				Poll::Ready(Some(sc_peerset::Message::Reject(index))) => {
 					self.peerset_report_reject(index);
 				}
-				Poll::Ready(Some(sc_peerset::Message::Connect(id))) => {
-					self.peerset_report_connect(id);
+				Poll::Ready(Some(sc_peerset::Message::Connect { peer_id, .. })) => {
+					self.peerset_report_connect(peer_id);
 				}
-				Poll::Ready(Some(sc_peerset::Message::Drop(id))) => {
-					self.peerset_report_disconnect(id);
+				Poll::Ready(Some(sc_peerset::Message::Drop { peer_id, .. })) => {
+					self.peerset_report_disconnect(peer_id);
 				}
 				Poll::Ready(None) => {
 					error!(target: "sub-libp2p", "Peerset receiver stream has returned None");