Incorporate sc-peerset into sc-network (#14236)

substrate/client/network/src/behaviour.rs

@@ -20,9 +20,11 @@ use crate::{
 	discovery::{DiscoveryBehaviour, DiscoveryConfig, DiscoveryOut},
 	event::DhtEvent,
 	peer_info,
+	peerset::PeersetHandle,
 	protocol::{CustomMessageOutcome, NotificationsSink, Protocol},
 	request_responses::{self, IfDisconnected, ProtocolConfig, RequestFailure},
 	types::ProtocolName,
+	ReputationChange,
 };
 
 use bytes::Bytes;
@@ -33,7 +35,6 @@ use libp2p::{
 };
 
 use sc_network_common::role::{ObservedRole, Roles};
-use sc_peerset::{PeersetHandle, ReputationChange};
 use sp_runtime::traits::Block as BlockT;
 use std::{collections::HashSet, time::Duration};
 

substrate/client/network/src/lib.rs

@@ -243,7 +243,10 @@
 //! More precise usage details are still being worked on and will likely change in the future.
 
 mod behaviour;
+mod peer_store;
+mod peerset;
 mod protocol;
+mod protocol_controller;
 mod service;
 
 pub mod config;
@@ -267,6 +270,7 @@ pub use sc_network_common::{
 		warp::{WarpSyncPhase, WarpSyncProgress},
 		ExtendedPeerInfo, StateDownloadProgress, SyncEventStream, SyncState, SyncStatusProvider,
 	},
+	types::ReputationChange,
 };
 pub use service::{
 	signature::Signature,
@@ -281,8 +285,6 @@ pub use service::{
 };
 pub use types::ProtocolName;
 
-pub use sc_peerset::ReputationChange;
-
 /// The maximum allowed number of established connections per peer.
 ///
 /// Typically, and by design of the network behaviours in this crate,

substrate/client/network/src/peer_store.rs

@@ -0,0 +1,407 @@
+// This file is part of Substrate.
+
+// Copyright (C) Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+use libp2p::PeerId;
+use log::trace;
+use parking_lot::Mutex;
+use partial_sort::PartialSort;
+use sc_network_common::types::ReputationChange;
+use std::{
+	cmp::{Ord, Ordering, PartialOrd},
+	collections::{hash_map::Entry, HashMap, HashSet},
+	fmt::Debug,
+	sync::Arc,
+	time::{Duration, Instant},
+};
+use wasm_timer::Delay;
+
+use crate::protocol_controller::ProtocolHandle;
+
+/// Log target for this file.
+pub const LOG_TARGET: &str = "peerset";
+
+/// We don't accept nodes whose reputation is under this value.
+pub const BANNED_THRESHOLD: i32 = 82 * (i32::MIN / 100);
+/// Reputation change for a node when we get disconnected from it.
+const DISCONNECT_REPUTATION_CHANGE: i32 = -256;
+/// Relative decrement of a reputation value that is applied every second. I.e., for inverse
+/// decrement of 50 we decrease absolute value of the reputation by 1/50. This corresponds to a
+/// factor of `k = 0.98`. It takes ~ `ln(0.5) / ln(k)` seconds to reduce the reputation by half,
+/// or 34.3 seconds for the values above. In this setup the maximum allowed absolute value of
+/// `i32::MAX` becomes 0 in ~1100 seconds (actually less due to integer arithmetic).
+const INVERSE_DECREMENT: i32 = 50;
+/// Amount of time between the moment we last updated the [`PeerStore`] entry and the moment we
+/// remove it, once the reputation value reaches 0.
+const FORGET_AFTER: Duration = Duration::from_secs(3600);
+
+pub trait PeerStoreProvider: Debug + Send {
+	/// Check whether the peer is banned.
+	fn is_banned(&self, peer_id: &PeerId) -> bool;
+
+	/// Register a protocol handle to disconnect peers whose reputation drops below the threshold.
+	fn register_protocol(&self, protocol_handle: ProtocolHandle);
+
+	/// Report peer disconnection for reputation adjustment.
+	fn report_disconnect(&mut self, peer_id: PeerId);
+
+	/// Adjust peer reputation.
+	fn report_peer(&mut self, peer_id: PeerId, change: ReputationChange);
+
+	/// Get peer reputation.
+	fn peer_reputation(&self, peer_id: &PeerId) -> i32;
+
+	/// Get candidates with highest reputations for initiating outgoing connections.
+	fn outgoing_candidates(&self, count: usize, ignored: HashSet<&PeerId>) -> Vec<PeerId>;
+}
+
+#[derive(Debug, Clone)]
+pub struct PeerStoreHandle {
+	inner: Arc<Mutex<PeerStoreInner>>,
+}
+
+impl PeerStoreProvider for PeerStoreHandle {
+	fn is_banned(&self, peer_id: &PeerId) -> bool {
+		self.inner.lock().is_banned(peer_id)
+	}
+
+	fn register_protocol(&self, protocol_handle: ProtocolHandle) {
+		self.inner.lock().register_protocol(protocol_handle);
+	}
+
+	fn report_disconnect(&mut self, peer_id: PeerId) {
+		self.inner.lock().report_disconnect(peer_id)
+	}
+
+	fn report_peer(&mut self, peer_id: PeerId, change: ReputationChange) {
+		self.inner.lock().report_peer(peer_id, change)
+	}
+
+	fn peer_reputation(&self, peer_id: &PeerId) -> i32 {
+		self.inner.lock().peer_reputation(peer_id)
+	}
+
+	fn outgoing_candidates(&self, count: usize, ignored: HashSet<&PeerId>) -> Vec<PeerId> {
+		self.inner.lock().outgoing_candidates(count, ignored)
+	}
+}
+
+impl PeerStoreHandle {
+	/// Get the number of known peers.
+	///
+	/// This number might not include some connected peers in rare cases when their reputation
+	/// was not updated for one hour, because their entries in [`PeerStore`] were dropped.
+	pub fn num_known_peers(&self) -> usize {
+		self.inner.lock().peers.len()
+	}
+
+	/// Add known peer.
+	pub fn add_known_peer(&mut self, peer_id: PeerId) {
+		self.inner.lock().add_known_peer(peer_id);
+	}
+}
+
+#[derive(Debug, Clone, Copy)]
+struct PeerInfo {
+	reputation: i32,
+	last_updated: Instant,
+}
+
+impl Default for PeerInfo {
+	fn default() -> Self {
+		Self { reputation: 0, last_updated: Instant::now() }
+	}
+}
+
+impl PartialEq for PeerInfo {
+	fn eq(&self, other: &Self) -> bool {
+		self.reputation == other.reputation
+	}
+}
+
+impl Eq for PeerInfo {}
+
+impl Ord for PeerInfo {
+	// We define reverse order by reputation values.
+	fn cmp(&self, other: &Self) -> Ordering {
+		self.reputation.cmp(&other.reputation).reverse()
+	}
+}
+
+impl PartialOrd for PeerInfo {
+	fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+		Some(self.cmp(other))
+	}
+}
+
+impl PeerInfo {
+	fn is_banned(&self) -> bool {
+		self.reputation < BANNED_THRESHOLD
+	}
+
+	fn add_reputation(&mut self, increment: i32) {
+		self.reputation = self.reputation.saturating_add(increment);
+		self.bump_last_updated();
+	}
+
+	fn decay_reputation(&mut self, seconds_passed: u64) {
+		// Note that decaying the reputation value happens "on its own",
+		// so we don't do `bump_last_updated()`.
+		for _ in 0..seconds_passed {
+			let mut diff = self.reputation / INVERSE_DECREMENT;
+			if diff == 0 && self.reputation < 0 {
+				diff = -1;
+			} else if diff == 0 && self.reputation > 0 {
+				diff = 1;
+			}
+
+			self.reputation = self.reputation.saturating_sub(diff);
+
+			if self.reputation == 0 {
+				break
+			}
+		}
+	}
+
+	fn bump_last_updated(&mut self) {
+		self.last_updated = Instant::now();
+	}
+}
+
+#[derive(Debug)]
+struct PeerStoreInner {
+	peers: HashMap<PeerId, PeerInfo>,
+	protocols: Vec<ProtocolHandle>,
+}
+
+impl PeerStoreInner {
+	fn is_banned(&self, peer_id: &PeerId) -> bool {
+		self.peers.get(peer_id).map_or(false, |info| info.is_banned())
+	}
+
+	fn register_protocol(&mut self, protocol_handle: ProtocolHandle) {
+		self.protocols.push(protocol_handle);
+	}
+
+	fn report_disconnect(&mut self, peer_id: PeerId) {
+		let peer_info = self.peers.entry(peer_id).or_default();
+		peer_info.add_reputation(DISCONNECT_REPUTATION_CHANGE);
+
+		log::trace!(
+			target: LOG_TARGET,
+			"Peer {} disconnected, reputation: {:+} to {}",
+			peer_id,
+			DISCONNECT_REPUTATION_CHANGE,
+			peer_info.reputation,
+		);
+	}
+
+	fn report_peer(&mut self, peer_id: PeerId, change: ReputationChange) {
+		let peer_info = self.peers.entry(peer_id).or_default();
+		peer_info.add_reputation(change.value);
+
+		if peer_info.reputation < BANNED_THRESHOLD {
+			self.protocols.iter().for_each(|handle| handle.disconnect_peer(peer_id));
+
+			log::trace!(
+				target: LOG_TARGET,
+				"Report {}: {:+} to {}. Reason: {}. Banned, disconnecting.",
+				peer_id,
+				change.value,
+				peer_info.reputation,
+				change.reason,
+			);
+		} else {
+			log::trace!(
+				target: LOG_TARGET,
+				"Report {}: {:+} to {}. Reason: {}.",
+				peer_id,
+				change.value,
+				peer_info.reputation,
+				change.reason,
+			);
+		}
+	}
+
+	fn peer_reputation(&self, peer_id: &PeerId) -> i32 {
+		self.peers.get(peer_id).map_or(0, |info| info.reputation)
+	}
+
+	fn outgoing_candidates(&self, count: usize, ignored: HashSet<&PeerId>) -> Vec<PeerId> {
+		let mut candidates = self
+			.peers
+			.iter()
+			.filter_map(|(peer_id, info)| {
+				(!info.is_banned() && !ignored.contains(peer_id)).then_some((*peer_id, *info))
+			})
+			.collect::<Vec<_>>();
+		let count = std::cmp::min(count, candidates.len());
+		candidates.partial_sort(count, |(_, info1), (_, info2)| info1.cmp(info2));
+		candidates.iter().take(count).map(|(peer_id, _)| *peer_id).collect()
+
+		// TODO: keep the peers sorted (in a "bi-multi-map"?) to not repeat sorting every time.
+	}
+
+	fn progress_time(&mut self, seconds_passed: u64) {
+		if seconds_passed == 0 {
+			return
+		}
+
+		// Drive reputation values towards 0.
+		self.peers
+			.iter_mut()
+			.for_each(|(_, info)| info.decay_reputation(seconds_passed));
+
+		// Retain only entries with non-zero reputation values or not expired ones.
+		let now = Instant::now();
+		self.peers
+			.retain(|_, info| info.reputation != 0 || info.last_updated + FORGET_AFTER > now);
+	}
+
+	fn add_known_peer(&mut self, peer_id: PeerId) {
+		match self.peers.entry(peer_id) {
+			Entry::Occupied(mut e) => {
+				trace!(
+					target: LOG_TARGET,
+					"Trying to add an already known peer {peer_id}, bumping `last_updated`.",
+				);
+				e.get_mut().bump_last_updated();
+			},
+			Entry::Vacant(e) => {
+				trace!(target: LOG_TARGET, "Adding a new known peer {peer_id}.");
+				e.insert(PeerInfo::default());
+			},
+		}
+	}
+}
+
+#[derive(Debug)]
+pub struct PeerStore {
+	inner: Arc<Mutex<PeerStoreInner>>,
+}
+
+impl PeerStore {
+	/// Create a new peer store from the list of bootnodes.
+	pub fn new(bootnodes: Vec<PeerId>) -> Self {
+		PeerStore {
+			inner: Arc::new(Mutex::new(PeerStoreInner {
+				peers: bootnodes
+					.into_iter()
+					.map(|peer_id| (peer_id, PeerInfo::default()))
+					.collect(),
+				protocols: Vec::new(),
+			})),
+		}
+	}
+
+	/// Get `PeerStoreHandle`.
+	pub fn handle(&self) -> PeerStoreHandle {
+		PeerStoreHandle { inner: self.inner.clone() }
+	}
+
+	/// Drive the `PeerStore`, decaying reputation values over time and removing expired entries.
+	pub async fn run(self) {
+		let started = Instant::now();
+		let mut latest_time_update = started;
+
+		loop {
+			let now = Instant::now();
+			// We basically do `(now - self.latest_update).as_secs()`, except that by the way we do
+			// it we know that we're not going to miss seconds because of rounding to integers.
+			let seconds_passed = {
+				let elapsed_latest = latest_time_update - started;
+				let elapsed_now = now - started;
+				latest_time_update = now;
+				elapsed_now.as_secs() - elapsed_latest.as_secs()
+			};
+
+			self.inner.lock().progress_time(seconds_passed);
+			let _ = Delay::new(Duration::from_secs(1)).await;
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::PeerInfo;
+
+	#[test]
+	fn decaying_zero_reputation_yields_zero() {
+		let mut peer_info = PeerInfo::default();
+		assert_eq!(peer_info.reputation, 0);
+
+		peer_info.decay_reputation(1);
+		assert_eq!(peer_info.reputation, 0);
+
+		peer_info.decay_reputation(100_000);
+		assert_eq!(peer_info.reputation, 0);
+	}
+
+	#[test]
+	fn decaying_positive_reputation_decreases_it() {
+		const INITIAL_REPUTATION: i32 = 100;
+
+		let mut peer_info = PeerInfo::default();
+		peer_info.reputation = INITIAL_REPUTATION;
+
+		peer_info.decay_reputation(1);
+		assert!(peer_info.reputation >= 0);
+		assert!(peer_info.reputation < INITIAL_REPUTATION);
+	}
+
+	#[test]
+	fn decaying_negative_reputation_increases_it() {
+		const INITIAL_REPUTATION: i32 = -100;
+
+		let mut peer_info = PeerInfo::default();
+		peer_info.reputation = INITIAL_REPUTATION;
+
+		peer_info.decay_reputation(1);
+		assert!(peer_info.reputation <= 0);
+		assert!(peer_info.reputation > INITIAL_REPUTATION);
+	}
+
+	#[test]
+	fn decaying_max_reputation_finally_yields_zero() {
+		const INITIAL_REPUTATION: i32 = i32::MAX;
+		const SECONDS: u64 = 1000;
+
+		let mut peer_info = PeerInfo::default();
+		peer_info.reputation = INITIAL_REPUTATION;
+
+		peer_info.decay_reputation(SECONDS / 2);
+		assert!(peer_info.reputation > 0);
+
+		peer_info.decay_reputation(SECONDS / 2);
+		assert_eq!(peer_info.reputation, 0);
+	}
+
+	#[test]
+	fn decaying_min_reputation_finally_yields_zero() {
+		const INITIAL_REPUTATION: i32 = i32::MIN;
+		const SECONDS: u64 = 1000;
+
+		let mut peer_info = PeerInfo::default();
+		peer_info.reputation = INITIAL_REPUTATION;
+
+		peer_info.decay_reputation(SECONDS / 2);
+		assert!(peer_info.reputation < 0);
+
+		peer_info.decay_reputation(SECONDS / 2);
+		assert_eq!(peer_info.reputation, 0);
+	}
+}

substrate/client/network/src/peerset.rs

@@ -0,0 +1,394 @@
+// This file is part of Substrate.
+
+// Copyright (C) Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+//! Peer Set Manager (PSM). Contains the strategy for choosing which nodes the network should be
+//! connected to.
+//!
+//! The PSM handles *sets* of nodes. A set of nodes is defined as the nodes that are believed to
+//! support a certain capability, such as handling blocks and transactions of a specific chain,
+//! or collating a certain parachain.
+//!
+//! For each node in each set, the peerset holds a flag specifying whether the node is
+//! connected to us or not.
+//!
+//! This connected/disconnected status is specific to the node and set combination, and it is for
+//! example possible for a node to be connected through a specific set but not another.
+//!
+//! In addition, for each, set, the peerset also holds a list of reserved nodes towards which it
+//! will at all time try to maintain a connection with.
+
+use crate::{
+	peer_store::{PeerStore, PeerStoreHandle, PeerStoreProvider},
+	protocol_controller::{ProtocolController, ProtocolHandle},
+};
+
+use futures::{
+	channel::oneshot,
+	future::{join_all, BoxFuture, JoinAll},
+	prelude::*,
+	stream::Stream,
+};
+use log::debug;
+use sc_network_common::types::ReputationChange;
+use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender};
+use serde_json::json;
+use std::{
+	collections::HashSet,
+	pin::Pin,
+	task::{Context, Poll},
+};
+
+use libp2p::PeerId;
+
+pub const LOG_TARGET: &str = "peerset";
+
+#[derive(Debug)]
+enum Action {
+	AddReservedPeer(SetId, PeerId),
+	RemoveReservedPeer(SetId, PeerId),
+	SetReservedPeers(SetId, HashSet<PeerId>),
+	SetReservedOnly(SetId, bool),
+	ReportPeer(PeerId, ReputationChange),
+	AddKnownPeer(PeerId),
+	PeerReputation(PeerId, oneshot::Sender<i32>),
+}
+
+/// Identifier of a set in the peerset.
+///
+/// Can be constructed using the `From<usize>` trait implementation based on the index of the set
+/// within [`PeersetConfig::sets`]. For example, the first element of [`PeersetConfig::sets`] is
+/// later referred to with `SetId::from(0)`. It is intended that the code responsible for building
+/// the [`PeersetConfig`] is also responsible for constructing the [`SetId`]s.
+#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct SetId(usize);
+
+impl SetId {
+	pub const fn from(id: usize) -> Self {
+		Self(id)
+	}
+}
+
+impl From<usize> for SetId {
+	fn from(id: usize) -> Self {
+		Self(id)
+	}
+}
+
+impl From<SetId> for usize {
+	fn from(id: SetId) -> Self {
+		id.0
+	}
+}
+
+/// Shared handle to the peer set manager (PSM). Distributed around the code.
+#[derive(Debug, Clone)]
+pub struct PeersetHandle {
+	tx: TracingUnboundedSender<Action>,
+}
+
+impl PeersetHandle {
+	/// Adds a new reserved peer. The peerset will make an effort to always remain connected to
+	/// this peer.
+	///
+	/// Has no effect if the node was already a reserved peer.
+	///
+	/// > **Note**: Keep in mind that the networking has to know an address for this node,
+	/// > otherwise it will not be able to connect to it.
+	pub fn add_reserved_peer(&self, set_id: SetId, peer_id: PeerId) {
+		let _ = self.tx.unbounded_send(Action::AddReservedPeer(set_id, peer_id));
+	}
+
+	/// Remove a previously-added reserved peer.
+	///
+	/// Has no effect if the node was not a reserved peer.
+	pub fn remove_reserved_peer(&self, set_id: SetId, peer_id: PeerId) {
+		let _ = self.tx.unbounded_send(Action::RemoveReservedPeer(set_id, peer_id));
+	}
+
+	/// Sets whether or not the peerset only has connections with nodes marked as reserved for
+	/// the given set.
+	pub fn set_reserved_only(&self, set_id: SetId, reserved: bool) {
+		let _ = self.tx.unbounded_send(Action::SetReservedOnly(set_id, reserved));
+	}
+
+	/// Set reserved peers to the new set.
+	pub fn set_reserved_peers(&self, set_id: SetId, peer_ids: HashSet<PeerId>) {
+		let _ = self.tx.unbounded_send(Action::SetReservedPeers(set_id, peer_ids));
+	}
+
+	/// Reports an adjustment to the reputation of the given peer.
+	pub fn report_peer(&self, peer_id: PeerId, score_diff: ReputationChange) {
+		let _ = self.tx.unbounded_send(Action::ReportPeer(peer_id, score_diff));
+	}
+
+	/// Add a peer to the list of known peers.
+	pub fn add_known_peer(&self, peer_id: PeerId) {
+		let _ = self.tx.unbounded_send(Action::AddKnownPeer(peer_id));
+	}
+
+	/// Returns the reputation value of the peer.
+	pub async fn peer_reputation(self, peer_id: PeerId) -> Result<i32, ()> {
+		let (tx, rx) = oneshot::channel();
+
+		let _ = self.tx.unbounded_send(Action::PeerReputation(peer_id, tx));
+
+		// The channel can only be closed if the peerset no longer exists.
+		rx.await.map_err(|_| ())
+	}
+}
+
+/// Message that can be sent by the peer set manager (PSM).
+#[derive(Debug, PartialEq)]
+pub enum Message {
+	/// Request to open a connection to the given peer. From the point of view of the PSM, we are
+	/// immediately connected.
+	Connect {
+		/// Set id to connect on.
+		set_id: SetId,
+		/// Peer to connect to.
+		peer_id: PeerId,
+	},
+
+	/// Drop the connection to the given peer, or cancel the connection attempt after a `Connect`.
+	Drop {
+		/// Set id to disconnect on.
+		set_id: SetId,
+		/// Peer to disconnect from.
+		peer_id: PeerId,
+	},
+
+	/// Equivalent to `Connect` for the peer corresponding to this incoming index.
+	Accept(IncomingIndex),
+
+	/// Equivalent to `Drop` for the peer corresponding to this incoming index.
+	Reject(IncomingIndex),
+}
+
+/// Opaque identifier for an incoming connection. Allocated by the network.
+#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct IncomingIndex(pub u64);
+
+impl From<u64> for IncomingIndex {
+	fn from(val: u64) -> Self {
+		Self(val)
+	}
+}
+
+/// Configuration to pass when creating the peer set manager.
+#[derive(Debug)]
+pub struct PeersetConfig {
+	/// List of sets of nodes the peerset manages.
+	pub sets: Vec<SetConfig>,
+}
+
+/// Configuration for a single set of nodes.
+#[derive(Debug)]
+pub struct SetConfig {
+	/// Maximum number of ingoing links to peers.
+	pub in_peers: u32,
+
+	/// Maximum number of outgoing links to peers.
+	pub out_peers: u32,
+
+	/// List of bootstrap nodes to initialize the set with.
+	///
+	/// > **Note**: Keep in mind that the networking has to know an address for these nodes,
+	/// > otherwise it will not be able to connect to them.
+	pub bootnodes: Vec<PeerId>,
+
+	/// Lists of nodes we should always be connected to.
+	///
+	/// > **Note**: Keep in mind that the networking has to know an address for these nodes,
+	/// >			otherwise it will not be able to connect to them.
+	pub reserved_nodes: HashSet<PeerId>,
+
+	/// If true, we only accept nodes in [`SetConfig::reserved_nodes`].
+	pub reserved_only: bool,
+}
+
+/// Side of the peer set manager owned by the network. In other words, the "receiving" side.
+///
+/// Implements the `Stream` trait and can be polled for messages. The `Stream` never ends and never
+/// errors.
+pub struct Peerset {
+	/// Peer reputation store handle.
+	peer_store_handle: PeerStoreHandle,
+	/// Peer reputation store.
+	peer_store_future: BoxFuture<'static, ()>,
+	/// Protocol handles.
+	protocol_handles: Vec<ProtocolHandle>,
+	/// Protocol controllers responsible for connections, per `SetId`.
+	protocol_controller_futures: JoinAll<BoxFuture<'static, ()>>,
+	/// Commands sent from protocol controllers to `Notifications`. The size of this vector never
+	/// changes.
+	from_controllers: TracingUnboundedReceiver<Message>,
+	/// Receiver for messages from the `PeersetHandle` and from `to_self`.
+	from_handle: TracingUnboundedReceiver<Action>,
+}
+
+impl Peerset {
+	/// Builds a new peerset from the given configuration.
+	pub fn from_config(config: PeersetConfig) -> (Peerset, PeersetHandle) {
+		let default_set_config = &config.sets[0];
+		let peer_store = PeerStore::new(default_set_config.bootnodes.clone());
+
+		let (to_notifications, from_controllers) =
+			tracing_unbounded("mpsc_protocol_controllers_to_notifications", 10_000);
+
+		let controllers = config
+			.sets
+			.into_iter()
+			.enumerate()
+			.map(|(set, set_config)| {
+				ProtocolController::new(
+					SetId::from(set),
+					set_config,
+					to_notifications.clone(),
+					Box::new(peer_store.handle()),
+				)
+			})
+			.collect::<Vec<_>>();
+
+		let (protocol_handles, protocol_controllers): (Vec<ProtocolHandle>, Vec<_>) =
+			controllers.into_iter().unzip();
+
+		let (tx, from_handle) = tracing_unbounded("mpsc_peerset_messages", 10_000);
+
+		let handle = PeersetHandle { tx };
+
+		let protocol_controller_futures =
+			join_all(protocol_controllers.into_iter().map(|c| c.run().boxed()));
+
+		let peerset = Peerset {
+			peer_store_handle: peer_store.handle(),
+			peer_store_future: peer_store.run().boxed(),
+			protocol_handles,
+			protocol_controller_futures,
+			from_controllers,
+			from_handle,
+		};
+
+		(peerset, handle)
+	}
+
+	/// Returns the list of reserved peers.
+	pub fn reserved_peers(&self, set_id: SetId, pending_response: oneshot::Sender<Vec<PeerId>>) {
+		self.protocol_handles[set_id.0].reserved_peers(pending_response);
+	}
+
+	/// Indicate that we received an incoming connection. Must be answered either with
+	/// a corresponding `Accept` or `Reject`, except if we were already connected to this peer.
+	///
+	/// Note that this mechanism is orthogonal to `Connect`/`Drop`. Accepting an incoming
+	/// connection implicitly means `Connect`, but incoming connections aren't cancelled by
+	/// `dropped`.
+	// Implementation note: because of concurrency issues, it is possible that we push a `Connect`
+	// message to the output channel with a `PeerId`, and that `incoming` gets called with the same
+	// `PeerId` before that message has been read by the user. In this situation we must not answer.
+	pub fn incoming(&mut self, set_id: SetId, peer_id: PeerId, index: IncomingIndex) {
+		self.protocol_handles[set_id.0].incoming_connection(peer_id, index);
+	}
+
+	/// Indicate that we dropped an active connection with a peer, or that we failed to connect.
+	///
+	/// Must only be called after the PSM has either generated a `Connect` message with this
+	/// `PeerId`, or accepted an incoming connection with this `PeerId`.
+	pub fn dropped(&mut self, set_id: SetId, peer_id: PeerId, _reason: DropReason) {
+		self.protocol_handles[set_id.0].dropped(peer_id);
+	}
+
+	/// Produces a JSON object containing the state of the peerset manager, for debugging purposes.
+	pub fn debug_info(&mut self) -> serde_json::Value {
+		// TODO: Check what info we can include here.
+		//       Issue reference: https://github.com/paritytech/substrate/issues/14160.
+		json!("unimplemented")
+	}
+
+	/// Returns the number of peers that we have discovered.
+	pub fn num_discovered_peers(&self) -> usize {
+		self.peer_store_handle.num_known_peers()
+	}
+}
+
+impl Stream for Peerset {
+	type Item = Message;
+
+	fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+		if let Poll::Ready(msg) = self.from_controllers.poll_next_unpin(cx) {
+			if let Some(msg) = msg {
+				return Poll::Ready(Some(msg))
+			} else {
+				debug!(
+					target: LOG_TARGET,
+					"All `ProtocolController`s have terminated, terminating `Peerset`."
+				);
+				return Poll::Ready(None)
+			}
+		}
+
+		while let Poll::Ready(action) = self.from_handle.poll_next_unpin(cx) {
+			if let Some(action) = action {
+				match action {
+					Action::AddReservedPeer(set_id, peer_id) =>
+						self.protocol_handles[set_id.0].add_reserved_peer(peer_id),
+					Action::RemoveReservedPeer(set_id, peer_id) =>
+						self.protocol_handles[set_id.0].remove_reserved_peer(peer_id),
+					Action::SetReservedPeers(set_id, peer_ids) =>
+						self.protocol_handles[set_id.0].set_reserved_peers(peer_ids),
+					Action::SetReservedOnly(set_id, reserved_only) =>
+						self.protocol_handles[set_id.0].set_reserved_only(reserved_only),
+					Action::ReportPeer(peer_id, score_diff) =>
+						self.peer_store_handle.report_peer(peer_id, score_diff),
+					Action::AddKnownPeer(peer_id) => self.peer_store_handle.add_known_peer(peer_id),
+					Action::PeerReputation(peer_id, pending_response) => {
+						let _ =
+							pending_response.send(self.peer_store_handle.peer_reputation(&peer_id));
+					},
+				}
+			} else {
+				debug!(target: LOG_TARGET, "`PeersetHandle` was dropped, terminating `Peerset`.");
+				return Poll::Ready(None)
+			}
+		}
+
+		if let Poll::Ready(()) = self.peer_store_future.poll_unpin(cx) {
+			debug!(target: LOG_TARGET, "`PeerStore` has terminated, terminating `PeerSet`.");
+			return Poll::Ready(None)
+		}
+
+		if let Poll::Ready(_) = self.protocol_controller_futures.poll_unpin(cx) {
+			debug!(
+				target: LOG_TARGET,
+				"All `ProtocolHandle`s have terminated, terminating `PeerSet`."
+			);
+			return Poll::Ready(None)
+		}
+
+		Poll::Pending
+	}
+}
+
+/// Reason for calling [`Peerset::dropped`].
+#[derive(Debug)]
+pub enum DropReason {
+	/// Substream or connection has been closed for an unknown reason.
+	Unknown,
+	/// Substream or connection has been explicitly refused by the target. In other words, the
+	/// peer doesn't actually belong to this set.
+	Refused,
+}

substrate/client/network/src/protocol.rs

@@ -20,6 +20,7 @@ use crate::{
 	config::{self, NonReservedPeerMode},
 	error,
 	types::ProtocolName,
+	ReputationChange,
 };
 
 use bytes::Bytes;
@@ -61,13 +62,13 @@ pub mod message;
 pub(crate) const BLOCK_ANNOUNCES_TRANSACTIONS_SUBSTREAM_SIZE: u64 = 16 * 1024 * 1024;
 
 /// Identifier of the peerset for the block announces protocol.
-const HARDCODED_PEERSETS_SYNC: sc_peerset::SetId = sc_peerset::SetId::from(0);
+const HARDCODED_PEERSETS_SYNC: crate::peerset::SetId = crate::peerset::SetId::from(0);
 /// Number of hardcoded peersets (the constants right above). Any set whose identifier is equal or
 /// superior to this value corresponds to a user-defined protocol.
 const NUM_HARDCODED_PEERSETS: usize = 1;
 
 mod rep {
-	use sc_peerset::ReputationChange as Rep;
+	use crate::ReputationChange as Rep;
 	/// We received a message that failed to decode.
 	pub const BAD_MESSAGE: Rep = Rep::new(-(1 << 12), "Bad message");
 }
@@ -78,7 +79,7 @@ type PendingSyncSubstreamValidation =
 // Lock must always be taken in order declared here.
 pub struct Protocol<B: BlockT> {
 	/// Used to report reputation changes.
-	peerset_handle: sc_peerset::PeersetHandle,
+	peerset_handle: crate::peerset::PeersetHandle,
 	/// Handles opening the unique substream and sending and receiving raw messages.
 	behaviour: Notifications,
 	/// List of notifications protocols that have been registered.
@@ -89,7 +90,7 @@ pub struct Protocol<B: BlockT> {
 	/// event to the outer layers, we also shouldn't propagate this "substream closed" event. To
 	/// solve this, an entry is added to this map whenever an invalid handshake is received.
 	/// Entries are removed when the corresponding "substream closed" is later received.
-	bad_handshake_substreams: HashSet<(PeerId, sc_peerset::SetId)>,
+	bad_handshake_substreams: HashSet<(PeerId, crate::peerset::SetId)>,
 	/// Connected peers.
 	peers: HashMap<PeerId, Roles>,
 	sync_substream_validations: FuturesUnordered<PendingSyncSubstreamValidation>,
@@ -105,7 +106,7 @@ impl<B: BlockT> Protocol<B> {
 		notification_protocols: Vec<config::NonDefaultSetConfig>,
 		block_announces_protocol: config::NonDefaultSetConfig,
 		tx: TracingUnboundedSender<crate::event::SyncEvent<B>>,
-	) -> error::Result<(Self, sc_peerset::PeersetHandle, Vec<(PeerId, Multiaddr)>)> {
+	) -> error::Result<(Self, crate::peerset::PeersetHandle, Vec<(PeerId, Multiaddr)>)> {
 		let mut known_addresses = Vec::new();
 
 		let (peerset, peerset_handle) = {
@@ -127,7 +128,7 @@ impl<B: BlockT> Protocol<B> {
 			}
 
 			// Set number 0 is used for block announces.
-			sets.push(sc_peerset::SetConfig {
+			sets.push(crate::peerset::SetConfig {
 				in_peers: network_config.default_peers_set.in_peers,
 				out_peers: network_config.default_peers_set.out_peers,
 				bootnodes,
@@ -146,7 +147,7 @@ impl<B: BlockT> Protocol<B> {
 				let reserved_only =
 					set_cfg.set_config.non_reserved_mode == NonReservedPeerMode::Deny;
 
-				sets.push(sc_peerset::SetConfig {
+				sets.push(crate::peerset::SetConfig {
 					in_peers: set_cfg.set_config.in_peers,
 					out_peers: set_cfg.set_config.out_peers,
 					bootnodes: Vec::new(),
@@ -155,7 +156,7 @@ impl<B: BlockT> Protocol<B> {
 				});
 			}
 
-			sc_peerset::Peerset::from_config(sc_peerset::PeersetConfig { sets })
+			crate::peerset::Peerset::from_config(crate::peerset::PeersetConfig { sets })
 		};
 
 		let behaviour = {
@@ -210,7 +211,7 @@ impl<B: BlockT> Protocol<B> {
 	pub fn disconnect_peer(&mut self, peer_id: &PeerId, protocol_name: ProtocolName) {
 		if let Some(position) = self.notification_protocols.iter().position(|p| *p == protocol_name)
 		{
-			self.behaviour.disconnect_peer(peer_id, sc_peerset::SetId::from(position));
+			self.behaviour.disconnect_peer(peer_id, crate::peerset::SetId::from(position));
 			self.peers.remove(peer_id);
 		} else {
 			warn!(target: "sub-libp2p", "disconnect_peer() with invalid protocol name")
@@ -228,7 +229,7 @@ impl<B: BlockT> Protocol<B> {
 	}
 
 	/// Adjusts the reputation of a node.
-	pub fn report_peer(&self, who: PeerId, reputation: sc_peerset::ReputationChange) {
+	pub fn report_peer(&self, who: PeerId, reputation: ReputationChange) {
 		self.peerset_handle.report_peer(who, reputation)
 	}
 
@@ -236,7 +237,7 @@ impl<B: BlockT> Protocol<B> {
 	pub fn set_notification_handshake(&mut self, protocol: ProtocolName, handshake: Vec<u8>) {
 		if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
 			self.behaviour
-				.set_notif_protocol_handshake(sc_peerset::SetId::from(index), handshake);
+				.set_notif_protocol_handshake(crate::peerset::SetId::from(index), handshake);
 		} else {
 			error!(
 				target: "sub-libp2p",
@@ -274,7 +275,8 @@ impl<B: BlockT> Protocol<B> {
 	/// Sets the list of reserved peers for the given protocol/peerset.
 	pub fn set_reserved_peerset_peers(&self, protocol: ProtocolName, peers: HashSet<PeerId>) {
 		if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
-			self.peerset_handle.set_reserved_peers(sc_peerset::SetId::from(index), peers);
+			self.peerset_handle
+				.set_reserved_peers(crate::peerset::SetId::from(index), peers);
 		} else {
 			error!(
 				target: "sub-libp2p",
@@ -287,7 +289,8 @@ impl<B: BlockT> Protocol<B> {
 	/// Removes a `PeerId` from the list of reserved peers.
 	pub fn remove_set_reserved_peer(&self, protocol: ProtocolName, peer: PeerId) {
 		if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
-			self.peerset_handle.remove_reserved_peer(sc_peerset::SetId::from(index), peer);
+			self.peerset_handle
+				.remove_reserved_peer(crate::peerset::SetId::from(index), peer);
 		} else {
 			error!(
 				target: "sub-libp2p",
@@ -300,7 +303,7 @@ impl<B: BlockT> Protocol<B> {
 	/// Adds a `PeerId` to the list of reserved peers.
 	pub fn add_set_reserved_peer(&self, protocol: ProtocolName, peer: PeerId) {
 		if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
-			self.peerset_handle.add_reserved_peer(sc_peerset::SetId::from(index), peer);
+			self.peerset_handle.add_reserved_peer(crate::peerset::SetId::from(index), peer);
 		} else {
 			error!(
 				target: "sub-libp2p",

substrate/client/network/src/protocol/notifications/behaviour.rs

@@ -17,6 +17,7 @@
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
 use crate::{
+	peerset::DropReason,
 	protocol::notifications::handler::{
 		self, NotificationsSink, NotifsHandler, NotifsHandlerIn, NotifsHandlerOut,
 	},
@@ -38,7 +39,6 @@ use libp2p::{
 use log::{debug, error, info, trace, warn};
 use parking_lot::RwLock;
 use rand::distributions::{Distribution as _, Uniform};
-use sc_peerset::DropReason;
 use smallvec::SmallVec;
 use std::{
 	cmp,
@@ -108,10 +108,10 @@ pub struct Notifications {
 	notif_protocols: Vec<handler::ProtocolConfig>,
 
 	/// Receiver for instructions about who to connect to or disconnect from.
-	peerset: sc_peerset::Peerset,
+	peerset: crate::peerset::Peerset,
 
 	/// List of peers in our state.
-	peers: FnvHashMap<(PeerId, sc_peerset::SetId), PeerState>,
+	peers: FnvHashMap<(PeerId, crate::peerset::SetId), PeerState>,
 
 	/// The elements in `peers` occasionally contain `Delay` objects that we would normally have
 	/// to be polled one by one. In order to avoid doing so, as an optimization, every `Delay` is
@@ -121,7 +121,7 @@ pub struct Notifications {
 	/// By design, we never remove elements from this list. Elements are removed only when the
 	/// `Delay` triggers. As such, this stream may produce obsolete elements.
 	delays: stream::FuturesUnordered<
-		Pin<Box<dyn Future<Output = (DelayId, PeerId, sc_peerset::SetId)> + Send>>,
+		Pin<Box<dyn Future<Output = (DelayId, PeerId, crate::peerset::SetId)> + Send>>,
 	>,
 
 	/// [`DelayId`] to assign to the next delay.
@@ -133,7 +133,7 @@ pub struct Notifications {
 
 	/// We generate indices to identify incoming connections. This is the next value for the index
 	/// to use when a connection is incoming.
-	next_incoming_index: sc_peerset::IncomingIndex,
+	next_incoming_index: crate::peerset::IncomingIndex,
 
 	/// Events to produce from `poll()`.
 	events: VecDeque<ToSwarm<NotificationsOut, NotifsHandlerIn>>,
@@ -232,7 +232,7 @@ enum PeerState {
 		backoff_until: Option<Instant>,
 
 		/// Incoming index tracking this connection.
-		incoming_index: sc_peerset::IncomingIndex,
+		incoming_index: crate::peerset::IncomingIndex,
 
 		/// List of connections with this peer, and their state.
 		connections: SmallVec<[(ConnectionId, ConnectionState); crate::MAX_CONNECTIONS_PER_PEER]>,
@@ -296,12 +296,12 @@ struct IncomingPeer {
 	/// Id of the remote peer of the incoming substream.
 	peer_id: PeerId,
 	/// Id of the set the incoming substream would belong to.
-	set_id: sc_peerset::SetId,
+	set_id: crate::peerset::SetId,
 	/// If true, this "incoming" still corresponds to an actual connection. If false, then the
 	/// connection corresponding to it has been closed or replaced already.
 	alive: bool,
 	/// Id that the we sent to the peerset.
-	incoming_id: sc_peerset::IncomingIndex,
+	incoming_id: crate::peerset::IncomingIndex,
 }
 
 /// Event that can be emitted by the `Notifications`.
@@ -312,7 +312,7 @@ pub enum NotificationsOut {
 		/// Id of the peer we are connected to.
 		peer_id: PeerId,
 		/// Peerset set ID the substream is tied to.
-		set_id: sc_peerset::SetId,
+		set_id: crate::peerset::SetId,
 		/// If `Some`, a fallback protocol name has been used rather the main protocol name.
 		/// Always matches one of the fallback names passed at initialization.
 		negotiated_fallback: Option<ProtocolName>,
@@ -332,7 +332,7 @@ pub enum NotificationsOut {
 		/// Id of the peer we are connected to.
 		peer_id: PeerId,
 		/// Peerset set ID the substream is tied to.
-		set_id: sc_peerset::SetId,
+		set_id: crate::peerset::SetId,
 		/// Replacement for the previous [`NotificationsSink`].
 		notifications_sink: NotificationsSink,
 	},
@@ -343,7 +343,7 @@ pub enum NotificationsOut {
 		/// Id of the peer we were connected to.
 		peer_id: PeerId,
 		/// Peerset set ID the substream was tied to.
-		set_id: sc_peerset::SetId,
+		set_id: crate::peerset::SetId,
 	},
 
 	/// Receives a message on a custom protocol substream.
@@ -353,7 +353,7 @@ pub enum NotificationsOut {
 		/// Id of the peer the message came from.
 		peer_id: PeerId,
 		/// Peerset set ID the substream is tied to.
-		set_id: sc_peerset::SetId,
+		set_id: crate::peerset::SetId,
 		/// Message that has been received.
 		message: BytesMut,
 	},
@@ -362,7 +362,7 @@ pub enum NotificationsOut {
 impl Notifications {
 	/// Creates a `CustomProtos`.
 	pub fn new(
-		peerset: sc_peerset::Peerset,
+		peerset: crate::peerset::Peerset,
 		notif_protocols: impl Iterator<Item = ProtocolConfig>,
 	) -> Self {
 		let notif_protocols = notif_protocols
@@ -383,7 +383,7 @@ impl Notifications {
 			delays: Default::default(),
 			next_delay_id: DelayId(0),
 			incoming: SmallVec::new(),
-			next_incoming_index: sc_peerset::IncomingIndex(0),
+			next_incoming_index: crate::peerset::IncomingIndex(0),
 			events: VecDeque::new(),
 		}
 	}
@@ -391,7 +391,7 @@ impl Notifications {
 	/// Modifies the handshake of the given notifications protocol.
 	pub fn set_notif_protocol_handshake(
 		&mut self,
-		set_id: sc_peerset::SetId,
+		set_id: crate::peerset::SetId,
 		handshake_message: impl Into<Vec<u8>>,
 	) {
 		if let Some(p) = self.notif_protocols.get_mut(usize::from(set_id)) {
@@ -413,18 +413,18 @@ impl Notifications {
 	}
 
 	/// Returns true if we have an open substream to the given peer.
-	pub fn is_open(&self, peer_id: &PeerId, set_id: sc_peerset::SetId) -> bool {
+	pub fn is_open(&self, peer_id: &PeerId, set_id: crate::peerset::SetId) -> bool {
 		self.peers.get(&(*peer_id, set_id)).map(|p| p.is_open()).unwrap_or(false)
 	}
 
 	/// Disconnects the given peer if we are connected to it.
-	pub fn disconnect_peer(&mut self, peer_id: &PeerId, set_id: sc_peerset::SetId) {
+	pub fn disconnect_peer(&mut self, peer_id: &PeerId, set_id: crate::peerset::SetId) {
 		trace!(target: "sub-libp2p", "External API => Disconnect({}, {:?})", peer_id, set_id);
 		self.disconnect_peer_inner(peer_id, set_id);
 	}
 
 	/// Inner implementation of `disconnect_peer`.
-	fn disconnect_peer_inner(&mut self, peer_id: &PeerId, set_id: sc_peerset::SetId) {
+	fn disconnect_peer_inner(&mut self, peer_id: &PeerId, set_id: crate::peerset::SetId) {
 		let mut entry = if let Entry::Occupied(entry) = self.peers.entry((*peer_id, set_id)) {
 			entry
 		} else {
@@ -541,7 +541,7 @@ impl Notifications {
 	/// Returns the list of reserved peers.
 	pub fn reserved_peers(
 		&self,
-		set_id: sc_peerset::SetId,
+		set_id: crate::peerset::SetId,
 		pending_response: oneshot::Sender<Vec<PeerId>>,
 	) {
 		self.peerset.reserved_peers(set_id, pending_response);
@@ -553,7 +553,7 @@ impl Notifications {
 	}
 
 	/// Function that is called when the peerset wants us to connect to a peer.
-	fn peerset_report_connect(&mut self, peer_id: PeerId, set_id: sc_peerset::SetId) {
+	fn peerset_report_connect(&mut self, peer_id: PeerId, set_id: crate::peerset::SetId) {
 		// If `PeerId` is unknown to us, insert an entry, start dialing, and return early.
 		let mut occ_entry = match self.peers.entry((peer_id, set_id)) {
 			Entry::Occupied(entry) => entry,
@@ -731,7 +731,7 @@ impl Notifications {
 	}
 
 	/// Function that is called when the peerset wants us to disconnect from a peer.
-	fn peerset_report_disconnect(&mut self, peer_id: PeerId, set_id: sc_peerset::SetId) {
+	fn peerset_report_disconnect(&mut self, peer_id: PeerId, set_id: crate::peerset::SetId) {
 		let mut entry = match self.peers.entry((peer_id, set_id)) {
 			Entry::Occupied(entry) => entry,
 			Entry::Vacant(entry) => {
@@ -839,7 +839,7 @@ impl Notifications {
 
 	/// Function that is called when the peerset wants us to accept a connection
 	/// request from a peer.
-	fn peerset_report_accept(&mut self, index: sc_peerset::IncomingIndex) {
+	fn peerset_report_accept(&mut self, index: crate::peerset::IncomingIndex) {
 		let incoming = if let Some(pos) = self.incoming.iter().position(|i| i.incoming_id == index)
 		{
 			self.incoming.remove(pos)
@@ -925,7 +925,7 @@ impl Notifications {
 	}
 
 	/// Function that is called when the peerset wants us to reject an incoming peer.
-	fn peerset_report_reject(&mut self, index: sc_peerset::IncomingIndex) {
+	fn peerset_report_reject(&mut self, index: crate::peerset::IncomingIndex) {
 		let incoming = if let Some(pos) = self.incoming.iter().position(|i| i.incoming_id == index)
 		{
 			self.incoming.remove(pos)
@@ -1059,7 +1059,7 @@ impl NetworkBehaviour for Notifications {
 				connection_id,
 				..
 			}) => {
-				for set_id in (0..self.notif_protocols.len()).map(sc_peerset::SetId::from) {
+				for set_id in (0..self.notif_protocols.len()).map(crate::peerset::SetId::from) {
 					match self.peers.entry((peer_id, set_id)).or_insert(PeerState::Poisoned) {
 						// Requested | PendingRequest => Enabled
 						st @ &mut PeerState::Requested |
@@ -1113,7 +1113,7 @@ impl NetworkBehaviour for Notifications {
 				}
 			},
 			FromSwarm::ConnectionClosed(ConnectionClosed { peer_id, connection_id, .. }) => {
-				for set_id in (0..self.notif_protocols.len()).map(sc_peerset::SetId::from) {
+				for set_id in (0..self.notif_protocols.len()).map(crate::peerset::SetId::from) {
 					let mut entry = if let Entry::Occupied(entry) =
 						self.peers.entry((peer_id, set_id))
 					{
@@ -1406,7 +1406,7 @@ impl NetworkBehaviour for Notifications {
 				if let Some(peer_id) = peer_id {
 					trace!(target: "sub-libp2p", "Libp2p => Dial failure for {:?}", peer_id);
 
-					for set_id in (0..self.notif_protocols.len()).map(sc_peerset::SetId::from) {
+					for set_id in (0..self.notif_protocols.len()).map(crate::peerset::SetId::from) {
 						if let Entry::Occupied(mut entry) = self.peers.entry((peer_id, set_id)) {
 							match mem::replace(entry.get_mut(), PeerState::Poisoned) {
 								// The peer is not in our list.
@@ -1485,7 +1485,7 @@ impl NetworkBehaviour for Notifications {
 	) {
 		match event {
 			NotifsHandlerOut::OpenDesiredByRemote { protocol_index } => {
-				let set_id = sc_peerset::SetId::from(protocol_index);
+				let set_id = crate::peerset::SetId::from(protocol_index);
 
 				trace!(target: "sub-libp2p",
 					"Handler({:?}, {:?}]) => OpenDesiredByRemote({:?})",
@@ -1675,7 +1675,7 @@ impl NetworkBehaviour for Notifications {
 			},
 
 			NotifsHandlerOut::CloseDesired { protocol_index } => {
-				let set_id = sc_peerset::SetId::from(protocol_index);
+				let set_id = crate::peerset::SetId::from(protocol_index);
 
 				trace!(target: "sub-libp2p",
 					"Handler({}, {:?}) => CloseDesired({:?})",
@@ -1775,7 +1775,7 @@ impl NetworkBehaviour for Notifications {
 			},
 
 			NotifsHandlerOut::CloseResult { protocol_index } => {
-				let set_id = sc_peerset::SetId::from(protocol_index);
+				let set_id = crate::peerset::SetId::from(protocol_index);
 
 				trace!(target: "sub-libp2p",
 					"Handler({}, {:?}) => CloseResult({:?})",
@@ -1814,7 +1814,7 @@ impl NetworkBehaviour for Notifications {
 				notifications_sink,
 				..
 			} => {
-				let set_id = sc_peerset::SetId::from(protocol_index);
+				let set_id = crate::peerset::SetId::from(protocol_index);
 				trace!(target: "sub-libp2p",
 					"Handler({}, {:?}) => OpenResultOk({:?})",
 					peer_id, connection_id, set_id);
@@ -1880,7 +1880,7 @@ impl NetworkBehaviour for Notifications {
 			},
 
 			NotifsHandlerOut::OpenResultErr { protocol_index } => {
-				let set_id = sc_peerset::SetId::from(protocol_index);
+				let set_id = crate::peerset::SetId::from(protocol_index);
 				trace!(target: "sub-libp2p",
 					"Handler({:?}, {:?}) => OpenResultErr({:?})",
 					peer_id, connection_id, set_id);
@@ -1969,7 +1969,7 @@ impl NetworkBehaviour for Notifications {
 			},
 
 			NotifsHandlerOut::Notification { protocol_index, message } => {
-				let set_id = sc_peerset::SetId::from(protocol_index);
+				let set_id = crate::peerset::SetId::from(protocol_index);
 				if self.is_open(&peer_id, set_id) {
 					trace!(
 						target: "sub-libp2p",
@@ -2015,16 +2015,16 @@ impl NetworkBehaviour for Notifications {
 		// Note that the peerset is a *best effort* crate, and we have to use defensive programming.
 		loop {
 			match futures::Stream::poll_next(Pin::new(&mut self.peerset), cx) {
-				Poll::Ready(Some(sc_peerset::Message::Accept(index))) => {
+				Poll::Ready(Some(crate::peerset::Message::Accept(index))) => {
 					self.peerset_report_accept(index);
 				},
-				Poll::Ready(Some(sc_peerset::Message::Reject(index))) => {
+				Poll::Ready(Some(crate::peerset::Message::Reject(index))) => {
 					self.peerset_report_reject(index);
 				},
-				Poll::Ready(Some(sc_peerset::Message::Connect { peer_id, set_id, .. })) => {
+				Poll::Ready(Some(crate::peerset::Message::Connect { peer_id, set_id, .. })) => {
 					self.peerset_report_connect(peer_id, set_id);
 				},
-				Poll::Ready(Some(sc_peerset::Message::Drop { peer_id, set_id, .. })) => {
+				Poll::Ready(Some(crate::peerset::Message::Drop { peer_id, set_id, .. })) => {
 					self.peerset_report_disconnect(peer_id, set_id);
 				},
 				Poll::Ready(None) => {
@@ -2105,9 +2105,8 @@ impl NetworkBehaviour for Notifications {
 #[allow(deprecated)]
 mod tests {
 	use super::*;
-	use crate::protocol::notifications::handler::tests::*;
+	use crate::{peerset::IncomingIndex, protocol::notifications::handler::tests::*};
 	use libp2p::swarm::AddressRecord;
-	use sc_peerset::IncomingIndex;
 	use std::{collections::HashSet, iter};
 
 	impl PartialEq for ConnectionState {
@@ -2153,11 +2152,11 @@ mod tests {
 		}
 	}
 
-	fn development_notifs() -> (Notifications, sc_peerset::PeersetHandle) {
+	fn development_notifs() -> (Notifications, crate::peerset::PeersetHandle) {
 		let (peerset, peerset_handle) = {
 			let mut sets = Vec::with_capacity(1);
 
-			sets.push(sc_peerset::SetConfig {
+			sets.push(crate::peerset::SetConfig {
 				in_peers: 25,
 				out_peers: 25,
 				bootnodes: Vec::new(),
@@ -2165,7 +2164,7 @@ mod tests {
 				reserved_only: false,
 			});
 
-			sc_peerset::Peerset::from_config(sc_peerset::PeersetConfig { sets })
+			crate::peerset::Peerset::from_config(crate::peerset::PeersetConfig { sets })
 		};
 
 		(
@@ -2311,7 +2310,7 @@ mod tests {
 
 		assert!(std::matches!(
 			notif.incoming.pop(),
-			Some(IncomingPeer { alive: true, incoming_id: sc_peerset::IncomingIndex(0), .. }),
+			Some(IncomingPeer { alive: true, incoming_id: crate::peerset::IncomingIndex(0), .. }),
 		));
 	}
 
@@ -2354,7 +2353,7 @@ mod tests {
 	#[test]
 	fn peerset_report_connect_backoff() {
 		let (mut notif, _peerset) = development_notifs();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
@@ -2421,7 +2420,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -2455,7 +2454,7 @@ mod tests {
 	#[test]
 	fn peerset_disconnect_disable_pending_enable() {
 		let (mut notif, _peerset) = development_notifs();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
@@ -2504,7 +2503,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -2539,7 +2538,7 @@ mod tests {
 	fn peerset_disconnect_requested() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 
 		// Set peer into `Requested` state.
 		notif.peerset_report_connect(peer, set_id);
@@ -2553,7 +2552,7 @@ mod tests {
 	#[test]
 	fn peerset_disconnect_pending_request() {
 		let (mut notif, _peerset) = development_notifs();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
@@ -2608,7 +2607,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -2635,17 +2634,17 @@ mod tests {
 
 		assert!(std::matches!(
 			notif.incoming[0],
-			IncomingPeer { alive: true, incoming_id: sc_peerset::IncomingIndex(0), .. },
+			IncomingPeer { alive: true, incoming_id: crate::peerset::IncomingIndex(0), .. },
 		));
 
 		notif.disconnect_peer(&peer, set_id);
 		assert!(std::matches!(notif.peers.get(&(peer, set_id)), Some(&PeerState::Disabled { .. })));
 		assert!(std::matches!(
 			notif.incoming[0],
-			IncomingPeer { alive: false, incoming_id: sc_peerset::IncomingIndex(0), .. },
+			IncomingPeer { alive: false, incoming_id: crate::peerset::IncomingIndex(0), .. },
 		));
 
-		notif.peerset_report_accept(sc_peerset::IncomingIndex(0));
+		notif.peerset_report_accept(crate::peerset::IncomingIndex(0));
 		assert_eq!(notif.incoming.len(), 0);
 		assert!(std::matches!(notif.peers.get(&(peer, set_id)), Some(PeerState::Disabled { .. })));
 	}
@@ -2656,7 +2655,7 @@ mod tests {
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
 		let conn2 = ConnectionId::new_unchecked(1);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -2710,7 +2709,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -2744,7 +2743,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -2780,7 +2779,7 @@ mod tests {
 		assert!(notif.peers.get(&(peer, set_id)).is_none());
 		assert!(std::matches!(
 			notif.incoming[0],
-			IncomingPeer { alive: false, incoming_id: sc_peerset::IncomingIndex(0), .. },
+			IncomingPeer { alive: false, incoming_id: crate::peerset::IncomingIndex(0), .. },
 		));
 	}
 
@@ -2790,7 +2789,7 @@ mod tests {
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
 		let conn1 = ConnectionId::new_unchecked(1);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -2858,7 +2857,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -2886,7 +2885,7 @@ mod tests {
 
 		// We rely on the implementation detail that incoming indices are counted
 		// from 0 to not mock the `Peerset`.
-		notif.peerset_report_accept(sc_peerset::IncomingIndex(0));
+		notif.peerset_report_accept(crate::peerset::IncomingIndex(0));
 		assert!(std::matches!(notif.peers.get(&(peer, set_id)), Some(&PeerState::Enabled { .. })));
 
 		// open new substream
@@ -2913,7 +2912,7 @@ mod tests {
 		let peer = PeerId::random();
 		let conn1 = ConnectionId::new_unchecked(0);
 		let conn2 = ConnectionId::new_unchecked(1);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3006,7 +3005,7 @@ mod tests {
 	fn dial_failure_for_requested_peer() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 
 		// Set peer into `Requested` state.
 		notif.peerset_report_connect(peer, set_id);
@@ -3030,7 +3029,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3077,7 +3076,7 @@ mod tests {
 	#[test]
 	fn peerset_report_connect_backoff_expired() {
 		let (mut notif, _peerset) = development_notifs();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
@@ -3126,7 +3125,7 @@ mod tests {
 	fn peerset_report_disconnect_disabled() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -3151,7 +3150,7 @@ mod tests {
 	#[test]
 	fn peerset_report_disconnect_backoff() {
 		let (mut notif, _peerset) = development_notifs();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
@@ -3197,7 +3196,7 @@ mod tests {
 	#[test]
 	fn peer_is_backed_off_if_both_connections_get_closed_while_peer_is_disabled_with_back_off() {
 		let (mut notif, _peerset) = development_notifs();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let peer = PeerId::random();
 		let conn1 = ConnectionId::new_unchecked(0);
 		let conn2 = ConnectionId::new_unchecked(1);
@@ -3271,7 +3270,7 @@ mod tests {
 	fn inject_connection_closed_incoming_with_backoff() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -3324,7 +3323,7 @@ mod tests {
 		let peer = PeerId::random();
 		let conn1 = ConnectionId::new_unchecked(0);
 		let conn2 = ConnectionId::new_unchecked(1);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3379,7 +3378,7 @@ mod tests {
 		let peer = PeerId::random();
 		let conn1 = ConnectionId::new_unchecked(0);
 		let conn2 = ConnectionId::new_unchecked(1);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3437,7 +3436,7 @@ mod tests {
 		let peer = PeerId::random();
 		let conn1 = ConnectionId::new_unchecked(0);
 		let conn2 = ConnectionId::new_unchecked(1);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3502,7 +3501,7 @@ mod tests {
 	#[test]
 	fn inject_dial_failure_for_pending_request() {
 		let (mut notif, _peerset) = development_notifs();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
@@ -3566,7 +3565,7 @@ mod tests {
 	fn peerstate_incoming_open_desired_by_remote() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let conn1 = ConnectionId::new_unchecked(0);
 		let conn2 = ConnectionId::new_unchecked(1);
 		let connected = ConnectedPoint::Listener {
@@ -3620,7 +3619,7 @@ mod tests {
 	async fn remove_backoff_peer_after_timeout() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -3699,7 +3698,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3817,7 +3816,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3865,7 +3864,7 @@ mod tests {
 	#[cfg(debug_assertions)]
 	fn peerset_report_connect_with_disabled_pending_enable_peer() {
 		let (mut notif, _peerset) = development_notifs();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
@@ -3912,7 +3911,7 @@ mod tests {
 	fn peerset_report_connect_with_requested_peer() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 
 		// Set peer into `Requested` state.
 		notif.peerset_report_connect(peer, set_id);
@@ -3927,7 +3926,7 @@ mod tests {
 	#[cfg(debug_assertions)]
 	fn peerset_report_connect_with_pending_requested() {
 		let (mut notif, _peerset) = development_notifs();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
@@ -3985,7 +3984,7 @@ mod tests {
 	fn peerset_report_connect_with_incoming_peer() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -4020,7 +4019,7 @@ mod tests {
 	fn peerset_report_disconnect_with_incoming_peer() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -4057,7 +4056,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -4084,11 +4083,11 @@ mod tests {
 
 		assert!(std::matches!(
 			notif.incoming[0],
-			IncomingPeer { alive: true, incoming_id: sc_peerset::IncomingIndex(0), .. },
+			IncomingPeer { alive: true, incoming_id: crate::peerset::IncomingIndex(0), .. },
 		));
 
 		notif.peers.remove(&(peer, set_id));
-		notif.peerset_report_accept(sc_peerset::IncomingIndex(0));
+		notif.peerset_report_accept(crate::peerset::IncomingIndex(0));
 	}
 
 	#[test]
@@ -4098,7 +4097,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -4126,7 +4125,7 @@ mod tests {
 
 		assert!(std::matches!(
 			notif.incoming[0],
-			IncomingPeer { alive: true, incoming_id: sc_peerset::IncomingIndex(0), .. },
+			IncomingPeer { alive: true, incoming_id: crate::peerset::IncomingIndex(0), .. },
 		));
 
 		notif.peerset_report_connect(peer, set_id);
@@ -4137,7 +4136,7 @@ mod tests {
 
 		assert!(std::matches!(notif.peers.get(&(peer, set_id)), Some(&PeerState::Enabled { .. })));
 		notif.incoming[0].alive = true;
-		notif.peerset_report_accept(sc_peerset::IncomingIndex(0));
+		notif.peerset_report_accept(crate::peerset::IncomingIndex(0));
 	}
 
 	#[test]
@@ -4166,7 +4165,7 @@ mod tests {
 	fn disconnect_non_existent_peer() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 
 		notif.peerset_report_disconnect(peer, set_id);
 
@@ -4199,7 +4198,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -4237,7 +4236,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -4263,7 +4262,7 @@ mod tests {
 		assert!(std::matches!(notif.peers.get(&(peer, set_id)), Some(&PeerState::Incoming { .. })));
 		assert!(std::matches!(
 			notif.incoming[0],
-			IncomingPeer { alive: true, incoming_id: sc_peerset::IncomingIndex(0), .. },
+			IncomingPeer { alive: true, incoming_id: crate::peerset::IncomingIndex(0), .. },
 		));
 
 		notif.peers.remove(&(peer, set_id));
@@ -4277,7 +4276,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -4318,7 +4317,7 @@ mod tests {
 	#[cfg(debug_assertions)]
 	fn inject_non_existent_connection_closed_for_disabled_peer() {
 		let (mut notif, _peerset) = development_notifs();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
@@ -4353,7 +4352,7 @@ mod tests {
 	#[cfg(debug_assertions)]
 	fn inject_non_existent_connection_closed_for_disabled_pending_enable() {
 		let (mut notif, _peerset) = development_notifs();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
@@ -4406,7 +4405,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -4450,7 +4449,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -4495,7 +4494,7 @@ mod tests {
 	#[cfg(debug_assertions)]
 	fn inject_connection_closed_for_backoff_peer() {
 		let (mut notif, _peerset) = development_notifs();
-		let set_id = sc_peerset::SetId::from(0);
+		let set_id = crate::peerset::SetId::from(0);
 		let peer = PeerId::random();
 		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {

substrate/client/network/src/protocol/notifications/tests.rs

@@ -65,19 +65,24 @@ fn build_nodes() -> (Swarm<CustomProtoWithAddr>, Swarm<CustomProtoWithAddr>) {
 			.timeout(Duration::from_secs(20))
 			.boxed();
 
-		let (peerset, _) = sc_peerset::Peerset::from_config(sc_peerset::PeersetConfig {
-			sets: vec![sc_peerset::SetConfig {
-				in_peers: 25,
-				out_peers: 25,
-				bootnodes: if index == 0 {
-					keypairs.iter().skip(1).map(|keypair| keypair.public().to_peer_id()).collect()
-				} else {
-					vec![]
-				},
-				reserved_nodes: Default::default(),
-				reserved_only: false,
-			}],
-		});
+		let (peerset, handle) =
+			crate::peerset::Peerset::from_config(crate::peerset::PeersetConfig {
+				sets: vec![crate::peerset::SetConfig {
+					in_peers: 25,
+					out_peers: 25,
+					bootnodes: if index == 0 {
+						keypairs
+							.iter()
+							.skip(1)
+							.map(|keypair| keypair.public().to_peer_id())
+							.collect()
+					} else {
+						vec![]
+					},
+					reserved_nodes: Default::default(),
+					reserved_only: false,
+				}],
+			});
 
 		let behaviour = CustomProtoWithAddr {
 			inner: Notifications::new(
@@ -89,6 +94,7 @@ fn build_nodes() -> (Swarm<CustomProtoWithAddr>, Swarm<CustomProtoWithAddr>) {
 					max_notification_size: 1024 * 1024,
 				}),
 			),
+			_peerset_handle: handle,
 			addrs: addrs
 				.iter()
 				.enumerate()
@@ -124,6 +130,8 @@ fn build_nodes() -> (Swarm<CustomProtoWithAddr>, Swarm<CustomProtoWithAddr>) {
 /// Wraps around the `CustomBehaviour` network behaviour, and adds hardcoded node addresses to it.
 struct CustomProtoWithAddr {
 	inner: Notifications,
+	// We need to keep `PeersetHandle` for `Peerset` not to shut down.
+	_peerset_handle: crate::peerset::PeersetHandle,
 	addrs: Vec<(PeerId, Multiaddr)>,
 }
 
@@ -266,7 +274,7 @@ fn reconnect_after_disconnect() {
 						if service2_state == ServiceState::FirstConnec {
 							service1.behaviour_mut().disconnect_peer(
 								Swarm::local_peer_id(&service2),
-								sc_peerset::SetId::from(0),
+								crate::peerset::SetId::from(0),
 							);
 						}
 					},
@@ -289,7 +297,7 @@ fn reconnect_after_disconnect() {
 						if service1_state == ServiceState::FirstConnec {
 							service1.behaviour_mut().disconnect_peer(
 								Swarm::local_peer_id(&service2),
-								sc_peerset::SetId::from(0),
+								crate::peerset::SetId::from(0),
 							);
 						}
 					},

substrate/client/network/src/request_responses.rs

@@ -34,7 +34,9 @@
 //! - If provided, a ["requests processing"](ProtocolConfig::inbound_queue) channel
 //! is used to handle incoming requests.
 
-use crate::{types::ProtocolName, ReputationChange};
+use crate::{
+	peer_store::BANNED_THRESHOLD, peerset::PeersetHandle, types::ProtocolName, ReputationChange,
+};
 
 use futures::{channel::oneshot, prelude::*};
 use libp2p::{
@@ -49,8 +51,6 @@ use libp2p::{
 	PeerId,
 };
 
-use sc_peerset::{PeersetHandle, BANNED_THRESHOLD};
-
 use std::{
 	collections::{hash_map::Entry, HashMap},
 	io, iter,
@@ -1040,6 +1040,7 @@ impl Codec for GenericCodec {
 mod tests {
 	use super::*;
 
+	use crate::peerset::{Peerset, PeersetConfig, SetConfig};
 	use futures::{channel::oneshot, executor::LocalPool, task::Spawn};
 	use libp2p::{
 		core::{
@@ -1051,7 +1052,6 @@ mod tests {
 		swarm::{Executor, Swarm, SwarmBuilder, SwarmEvent},
 		Multiaddr,
 	};
-	use sc_peerset::{Peerset, PeersetConfig, SetConfig};
 	use std::{iter, time::Duration};
 
 	struct TokioExecutor(tokio::runtime::Runtime);

substrate/client/network/src/service.rs

@@ -36,6 +36,7 @@ use crate::{
 	network_state::{
 		NetworkState, NotConnectedPeer as NetworkStateNotConnectedPeer, Peer as NetworkStatePeer,
 	},
+	peerset::PeersetHandle,
 	protocol::{self, NotifsHandlerError, Protocol, Ready},
 	request_responses::{IfDisconnected, RequestFailure},
 	service::{
@@ -73,7 +74,6 @@ use metrics::{Histogram, HistogramVec, MetricSources, Metrics};
 use parking_lot::Mutex;
 
 use sc_network_common::ExHashT;
-use sc_peerset::PeersetHandle;
 use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender};
 use sp_runtime::traits::Block as BlockT;
 

substrate/client/network/src/service/traits.rs

@@ -24,13 +24,12 @@ use crate::{
 	request_responses::{IfDisconnected, RequestFailure},
 	service::signature::Signature,
 	types::ProtocolName,
+	ReputationChange,
 };
 
 use futures::{channel::oneshot, Stream};
 use libp2p::{Multiaddr, PeerId};
 
-use sc_peerset::ReputationChange;
-
 use std::{collections::HashSet, future::Future, pin::Pin, sync::Arc};
 
 pub use libp2p::{identity::SigningError, kad::record::Key as KademliaKey};