Move code from sc-network-common back to sc-network (#13592)

* Move service tests to `client/network/tests`

These tests depend on `sc-network` and `sc-network-sync` so they should
live outside the crate.

* Move some configs from `sc-network-common` to `sc-network`

* Move `NetworkService` traits to `sc-network`

* Move request-responses to `sc-network`

* Remove more stuff

* Remove rest of configs from `sc-network-common` to `sc-network`

* Remove more stuff

* Fix warnings

* Update client/network/src/request_responses.rs

Co-authored-by: Dmitry Markin <dmitry@markin.tech>

* Fix cargo doc

---------

Co-authored-by: Dmitry Markin <dmitry@markin.tech>

substrate/client/network/src/behaviour.rs

@@ -18,9 +18,11 @@
 
 use crate::{
 	discovery::{DiscoveryBehaviour, DiscoveryConfig, DiscoveryOut},
+	event::DhtEvent,
 	peer_info,
 	protocol::{CustomMessageOutcome, NotificationsSink, Protocol},
-	request_responses,
+	request_responses::{self, IfDisconnected, ProtocolConfig, RequestFailure},
+	types::ProtocolName,
 };
 
 use bytes::Bytes;
@@ -32,14 +34,7 @@ use libp2p::{
 	swarm::NetworkBehaviour,
 };
 
-use sc_network_common::{
-	protocol::{
-		event::DhtEvent,
-		role::{ObservedRole, Roles},
-		ProtocolName,
-	},
-	request_responses::{IfDisconnected, ProtocolConfig, RequestFailure},
-};
+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/config.rs

@@ -21,21 +21,671 @@
 //! The [`Params`] struct is the struct that must be passed in order to initialize the networking.
 //! See the documentation of [`Params`].
 
-pub use sc_network_common::{
-	config::{NetworkConfiguration, ProtocolId},
-	protocol::role::Role,
+pub use crate::{
 	request_responses::{
 		IncomingRequest, OutgoingResponse, ProtocolConfig as RequestResponseConfig,
 	},
-	sync::warp::WarpSyncProvider,
-	ExHashT,
+	types::ProtocolName,
 };
 
-pub use libp2p::{build_multiaddr, core::PublicKey, identity};
-
+use codec::Encode;
+use libp2p::{identity::Keypair, multiaddr, Multiaddr, PeerId};
 use prometheus_endpoint::Registry;
-use sc_network_common::config::NonDefaultSetConfig;
-use std::{future::Future, pin::Pin, sync::Arc};
+pub use sc_network_common::{role::Role, sync::warp::WarpSyncProvider, ExHashT};
+use zeroize::Zeroize;
+
+use std::{
+	error::Error,
+	fmt, fs,
+	future::Future,
+	io::{self, Write},
+	iter,
+	net::Ipv4Addr,
+	path::{Path, PathBuf},
+	pin::Pin,
+	str::{self, FromStr},
+	sync::Arc,
+};
+
+pub use libp2p::{
+	build_multiaddr,
+	identity::{self, ed25519},
+};
+
+/// Protocol name prefix, transmitted on the wire for legacy protocol names.
+/// I.e., `dot` in `/dot/sync/2`. Should be unique for each chain. Always UTF-8.
+/// Deprecated in favour of genesis hash & fork ID based protocol names.
+#[derive(Clone, PartialEq, Eq, Hash)]
+pub struct ProtocolId(smallvec::SmallVec<[u8; 6]>);
+
+impl<'a> From<&'a str> for ProtocolId {
+	fn from(bytes: &'a str) -> ProtocolId {
+		Self(bytes.as_bytes().into())
+	}
+}
+
+impl AsRef<str> for ProtocolId {
+	fn as_ref(&self) -> &str {
+		str::from_utf8(&self.0[..])
+			.expect("the only way to build a ProtocolId is through a UTF-8 String; qed")
+	}
+}
+
+impl fmt::Debug for ProtocolId {
+	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+		fmt::Debug::fmt(self.as_ref(), f)
+	}
+}
+
+/// Parses a string address and splits it into Multiaddress and PeerId, if
+/// valid.
+///
+/// # Example
+///
+/// ```
+/// # use libp2p::{Multiaddr, PeerId};
+/// use sc_network::config::parse_str_addr;
+/// let (peer_id, addr) = parse_str_addr(
+/// 	"/ip4/198.51.100.19/tcp/30333/p2p/QmSk5HQbn6LhUwDiNMseVUjuRYhEtYj4aUZ6WfWoGURpdV"
+/// ).unwrap();
+/// assert_eq!(peer_id, "QmSk5HQbn6LhUwDiNMseVUjuRYhEtYj4aUZ6WfWoGURpdV".parse::<PeerId>().unwrap());
+/// assert_eq!(addr, "/ip4/198.51.100.19/tcp/30333".parse::<Multiaddr>().unwrap());
+/// ```
+pub fn parse_str_addr(addr_str: &str) -> Result<(PeerId, Multiaddr), ParseErr> {
+	let addr: Multiaddr = addr_str.parse()?;
+	parse_addr(addr)
+}
+
+/// Splits a Multiaddress into a Multiaddress and PeerId.
+pub fn parse_addr(mut addr: Multiaddr) -> Result<(PeerId, Multiaddr), ParseErr> {
+	let who = match addr.pop() {
+		Some(multiaddr::Protocol::P2p(key)) =>
+			PeerId::from_multihash(key).map_err(|_| ParseErr::InvalidPeerId)?,
+		_ => return Err(ParseErr::PeerIdMissing),
+	};
+
+	Ok((who, addr))
+}
+
+/// Address of a node, including its identity.
+///
+/// This struct represents a decoded version of a multiaddress that ends with `/p2p/<peerid>`.
+///
+/// # Example
+///
+/// ```
+/// # use libp2p::{Multiaddr, PeerId};
+/// use sc_network::config::MultiaddrWithPeerId;
+/// let addr: MultiaddrWithPeerId =
+/// 	"/ip4/198.51.100.19/tcp/30333/p2p/QmSk5HQbn6LhUwDiNMseVUjuRYhEtYj4aUZ6WfWoGURpdV".parse().unwrap();
+/// assert_eq!(addr.peer_id.to_base58(), "QmSk5HQbn6LhUwDiNMseVUjuRYhEtYj4aUZ6WfWoGURpdV");
+/// assert_eq!(addr.multiaddr.to_string(), "/ip4/198.51.100.19/tcp/30333");
+/// ```
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
+#[serde(try_from = "String", into = "String")]
+pub struct MultiaddrWithPeerId {
+	/// Address of the node.
+	pub multiaddr: Multiaddr,
+	/// Its identity.
+	pub peer_id: PeerId,
+}
+
+impl MultiaddrWithPeerId {
+	/// Concatenates the multiaddress and peer ID into one multiaddress containing both.
+	pub fn concat(&self) -> Multiaddr {
+		let proto = multiaddr::Protocol::P2p(From::from(self.peer_id));
+		self.multiaddr.clone().with(proto)
+	}
+}
+
+impl fmt::Display for MultiaddrWithPeerId {
+	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+		fmt::Display::fmt(&self.concat(), f)
+	}
+}
+
+impl FromStr for MultiaddrWithPeerId {
+	type Err = ParseErr;
+
+	fn from_str(s: &str) -> Result<Self, Self::Err> {
+		let (peer_id, multiaddr) = parse_str_addr(s)?;
+		Ok(Self { peer_id, multiaddr })
+	}
+}
+
+impl From<MultiaddrWithPeerId> for String {
+	fn from(ma: MultiaddrWithPeerId) -> String {
+		format!("{}", ma)
+	}
+}
+
+impl TryFrom<String> for MultiaddrWithPeerId {
+	type Error = ParseErr;
+	fn try_from(string: String) -> Result<Self, Self::Error> {
+		string.parse()
+	}
+}
+
+/// Error that can be generated by `parse_str_addr`.
+#[derive(Debug)]
+pub enum ParseErr {
+	/// Error while parsing the multiaddress.
+	MultiaddrParse(multiaddr::Error),
+	/// Multihash of the peer ID is invalid.
+	InvalidPeerId,
+	/// The peer ID is missing from the address.
+	PeerIdMissing,
+}
+
+impl fmt::Display for ParseErr {
+	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+		match self {
+			Self::MultiaddrParse(err) => write!(f, "{}", err),
+			Self::InvalidPeerId => write!(f, "Peer id at the end of the address is invalid"),
+			Self::PeerIdMissing => write!(f, "Peer id is missing from the address"),
+		}
+	}
+}
+
+impl std::error::Error for ParseErr {
+	fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+		match self {
+			Self::MultiaddrParse(err) => Some(err),
+			Self::InvalidPeerId => None,
+			Self::PeerIdMissing => None,
+		}
+	}
+}
+
+impl From<multiaddr::Error> for ParseErr {
+	fn from(err: multiaddr::Error) -> ParseErr {
+		Self::MultiaddrParse(err)
+	}
+}
+
+/// Custom handshake for the notification protocol
+#[derive(Debug, Clone)]
+pub struct NotificationHandshake(Vec<u8>);
+
+impl NotificationHandshake {
+	/// Create new `NotificationHandshake` from an object that implements `Encode`
+	pub fn new<H: Encode>(handshake: H) -> Self {
+		Self(handshake.encode())
+	}
+
+	/// Create new `NotificationHandshake` from raw bytes
+	pub fn from_bytes(bytes: Vec<u8>) -> Self {
+		Self(bytes)
+	}
+}
+
+impl std::ops::Deref for NotificationHandshake {
+	type Target = Vec<u8>;
+
+	fn deref(&self) -> &Self::Target {
+		&self.0
+	}
+}
+
+/// Configuration for the transport layer.
+#[derive(Clone, Debug)]
+pub enum TransportConfig {
+	/// Normal transport mode.
+	Normal {
+		/// If true, the network will use mDNS to discover other libp2p nodes on the local network
+		/// and connect to them if they support the same chain.
+		enable_mdns: bool,
+
+		/// If true, allow connecting to private IPv4/IPv6 addresses (as defined in
+		/// [RFC1918](https://tools.ietf.org/html/rfc1918)). Irrelevant for addresses that have
+		/// been passed in `::sc_network::config::NetworkConfiguration::boot_nodes`.
+		allow_private_ip: bool,
+	},
+
+	/// Only allow connections within the same process.
+	/// Only addresses of the form `/memory/...` will be supported.
+	MemoryOnly,
+}
+
+/// The policy for connections to non-reserved peers.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum NonReservedPeerMode {
+	/// Accept them. This is the default.
+	Accept,
+	/// Deny them.
+	Deny,
+}
+
+impl NonReservedPeerMode {
+	/// Attempt to parse the peer mode from a string.
+	pub fn parse(s: &str) -> Option<Self> {
+		match s {
+			"accept" => Some(Self::Accept),
+			"deny" => Some(Self::Deny),
+			_ => None,
+		}
+	}
+}
+
+/// Sync operation mode.
+#[derive(Copy, Clone, Debug, Eq, PartialEq)]
+pub enum SyncMode {
+	/// Full block download and verification.
+	Full,
+	/// Download blocks and the latest state.
+	Fast {
+		/// Skip state proof download and verification.
+		skip_proofs: bool,
+		/// Download indexed transactions for recent blocks.
+		storage_chain_mode: bool,
+	},
+	/// Warp sync - verify authority set transitions and the latest state.
+	Warp,
+}
+
+impl SyncMode {
+	/// Returns if `self` is [`Self::Warp`].
+	pub fn is_warp(&self) -> bool {
+		matches!(self, Self::Warp)
+	}
+
+	/// Returns if `self` is [`Self::Fast`].
+	pub fn is_fast(&self) -> bool {
+		matches!(self, Self::Fast { .. })
+	}
+}
+
+impl Default for SyncMode {
+	fn default() -> Self {
+		Self::Full
+	}
+}
+
+/// The configuration of a node's secret key, describing the type of key
+/// and how it is obtained. A node's identity keypair is the result of
+/// the evaluation of the node key configuration.
+#[derive(Clone, Debug)]
+pub enum NodeKeyConfig {
+	/// A Ed25519 secret key configuration.
+	Ed25519(Secret<ed25519::SecretKey>),
+}
+
+impl Default for NodeKeyConfig {
+	fn default() -> NodeKeyConfig {
+		Self::Ed25519(Secret::New)
+	}
+}
+
+/// The options for obtaining a Ed25519 secret key.
+pub type Ed25519Secret = Secret<ed25519::SecretKey>;
+
+/// The configuration options for obtaining a secret key `K`.
+#[derive(Clone)]
+pub enum Secret<K> {
+	/// Use the given secret key `K`.
+	Input(K),
+	/// Read the secret key from a file. If the file does not exist,
+	/// it is created with a newly generated secret key `K`. The format
+	/// of the file is determined by `K`:
+	///
+	///   * `ed25519::SecretKey`: An unencoded 32 bytes Ed25519 secret key.
+	File(PathBuf),
+	/// Always generate a new secret key `K`.
+	New,
+}
+
+impl<K> fmt::Debug for Secret<K> {
+	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+		match self {
+			Self::Input(_) => f.debug_tuple("Secret::Input").finish(),
+			Self::File(path) => f.debug_tuple("Secret::File").field(path).finish(),
+			Self::New => f.debug_tuple("Secret::New").finish(),
+		}
+	}
+}
+
+impl NodeKeyConfig {
+	/// Evaluate a `NodeKeyConfig` to obtain an identity `Keypair`:
+	///
+	///  * If the secret is configured as input, the corresponding keypair is returned.
+	///
+	///  * If the secret is configured as a file, it is read from that file, if it exists. Otherwise
+	///    a new secret is generated and stored. In either case, the keypair obtained from the
+	///    secret is returned.
+	///
+	///  * If the secret is configured to be new, it is generated and the corresponding keypair is
+	///    returned.
+	pub fn into_keypair(self) -> io::Result<Keypair> {
+		use NodeKeyConfig::*;
+		match self {
+			Ed25519(Secret::New) => Ok(Keypair::generate_ed25519()),
+
+			Ed25519(Secret::Input(k)) => Ok(Keypair::Ed25519(k.into())),
+
+			Ed25519(Secret::File(f)) => get_secret(
+				f,
+				|mut b| match String::from_utf8(b.to_vec()).ok().and_then(|s| {
+					if s.len() == 64 {
+						array_bytes::hex2bytes(&s).ok()
+					} else {
+						None
+					}
+				}) {
+					Some(s) => ed25519::SecretKey::from_bytes(s),
+					_ => ed25519::SecretKey::from_bytes(&mut b),
+				},
+				ed25519::SecretKey::generate,
+				|b| b.as_ref().to_vec(),
+			)
+			.map(ed25519::Keypair::from)
+			.map(Keypair::Ed25519),
+		}
+	}
+}
+
+/// Load a secret key from a file, if it exists, or generate a
+/// new secret key and write it to that file. In either case,
+/// the secret key is returned.
+fn get_secret<P, F, G, E, W, K>(file: P, parse: F, generate: G, serialize: W) -> io::Result<K>
+where
+	P: AsRef<Path>,
+	F: for<'r> FnOnce(&'r mut [u8]) -> Result<K, E>,
+	G: FnOnce() -> K,
+	E: Error + Send + Sync + 'static,
+	W: Fn(&K) -> Vec<u8>,
+{
+	std::fs::read(&file)
+		.and_then(|mut sk_bytes| {
+			parse(&mut sk_bytes).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))
+		})
+		.or_else(|e| {
+			if e.kind() == io::ErrorKind::NotFound {
+				file.as_ref().parent().map_or(Ok(()), fs::create_dir_all)?;
+				let sk = generate();
+				let mut sk_vec = serialize(&sk);
+				write_secret_file(file, &sk_vec)?;
+				sk_vec.zeroize();
+				Ok(sk)
+			} else {
+				Err(e)
+			}
+		})
+}
+
+/// Write secret bytes to a file.
+fn write_secret_file<P>(path: P, sk_bytes: &[u8]) -> io::Result<()>
+where
+	P: AsRef<Path>,
+{
+	let mut file = open_secret_file(&path)?;
+	file.write_all(sk_bytes)
+}
+
+/// Opens a file containing a secret key in write mode.
+#[cfg(unix)]
+fn open_secret_file<P>(path: P) -> io::Result<fs::File>
+where
+	P: AsRef<Path>,
+{
+	use std::os::unix::fs::OpenOptionsExt;
+	fs::OpenOptions::new().write(true).create_new(true).mode(0o600).open(path)
+}
+
+/// Opens a file containing a secret key in write mode.
+#[cfg(not(unix))]
+fn open_secret_file<P>(path: P) -> Result<fs::File, io::Error>
+where
+	P: AsRef<Path>,
+{
+	fs::OpenOptions::new().write(true).create_new(true).open(path)
+}
+
+/// Configuration for a set of nodes.
+#[derive(Clone, Debug)]
+pub struct SetConfig {
+	/// Maximum allowed number of incoming substreams related to this set.
+	pub in_peers: u32,
+
+	/// Number of outgoing substreams related to this set that we're trying to maintain.
+	pub out_peers: u32,
+
+	/// List of reserved node addresses.
+	pub reserved_nodes: Vec<MultiaddrWithPeerId>,
+
+	/// Whether nodes that aren't in [`SetConfig::reserved_nodes`] are accepted or automatically
+	/// refused.
+	pub non_reserved_mode: NonReservedPeerMode,
+}
+
+impl Default for SetConfig {
+	fn default() -> Self {
+		Self {
+			in_peers: 25,
+			out_peers: 75,
+			reserved_nodes: Vec::new(),
+			non_reserved_mode: NonReservedPeerMode::Accept,
+		}
+	}
+}
+
+/// Extension to [`SetConfig`] for sets that aren't the default set.
+///
+/// > **Note**: As new fields might be added in the future, please consider using the `new` method
+/// >			and modifiers instead of creating this struct manually.
+#[derive(Clone, Debug)]
+pub struct NonDefaultSetConfig {
+	/// Name of the notifications protocols of this set. A substream on this set will be
+	/// considered established once this protocol is open.
+	///
+	/// > **Note**: This field isn't present for the default set, as this is handled internally
+	/// > by the networking code.
+	pub notifications_protocol: ProtocolName,
+
+	/// If the remote reports that it doesn't support the protocol indicated in the
+	/// `notifications_protocol` field, then each of these fallback names will be tried one by
+	/// one.
+	///
+	/// If a fallback is used, it will be reported in
+	/// `sc_network::protocol::event::Event::NotificationStreamOpened::negotiated_fallback`
+	pub fallback_names: Vec<ProtocolName>,
+
+	/// Handshake of the protocol
+	///
+	/// NOTE: Currently custom handshakes are not fully supported. See issue #5685 for more
+	/// details. This field is temporarily used to allow moving the hardcoded block announcement
+	/// protocol out of `protocol.rs`.
+	pub handshake: Option<NotificationHandshake>,
+
+	/// Maximum allowed size of single notifications.
+	pub max_notification_size: u64,
+
+	/// Base configuration.
+	pub set_config: SetConfig,
+}
+
+impl NonDefaultSetConfig {
+	/// Creates a new [`NonDefaultSetConfig`]. Zero slots and accepts only reserved nodes.
+	pub fn new(notifications_protocol: ProtocolName, max_notification_size: u64) -> Self {
+		Self {
+			notifications_protocol,
+			max_notification_size,
+			fallback_names: Vec::new(),
+			handshake: None,
+			set_config: SetConfig {
+				in_peers: 0,
+				out_peers: 0,
+				reserved_nodes: Vec::new(),
+				non_reserved_mode: NonReservedPeerMode::Deny,
+			},
+		}
+	}
+
+	/// Modifies the configuration to allow non-reserved nodes.
+	pub fn allow_non_reserved(&mut self, in_peers: u32, out_peers: u32) {
+		self.set_config.in_peers = in_peers;
+		self.set_config.out_peers = out_peers;
+		self.set_config.non_reserved_mode = NonReservedPeerMode::Accept;
+	}
+
+	/// Add a node to the list of reserved nodes.
+	pub fn add_reserved(&mut self, peer: MultiaddrWithPeerId) {
+		self.set_config.reserved_nodes.push(peer);
+	}
+
+	/// Add a list of protocol names used for backward compatibility.
+	///
+	/// See the explanations in [`NonDefaultSetConfig::fallback_names`].
+	pub fn add_fallback_names(&mut self, fallback_names: Vec<ProtocolName>) {
+		self.fallback_names.extend(fallback_names);
+	}
+}
+
+/// Network service configuration.
+#[derive(Clone, Debug)]
+pub struct NetworkConfiguration {
+	/// Directory path to store network-specific configuration. None means nothing will be saved.
+	pub net_config_path: Option<PathBuf>,
+
+	/// Multiaddresses to listen for incoming connections.
+	pub listen_addresses: Vec<Multiaddr>,
+
+	/// Multiaddresses to advertise. Detected automatically if empty.
+	pub public_addresses: Vec<Multiaddr>,
+
+	/// List of initial node addresses
+	pub boot_nodes: Vec<MultiaddrWithPeerId>,
+
+	/// The node key configuration, which determines the node's network identity keypair.
+	pub node_key: NodeKeyConfig,
+
+	/// List of request-response protocols that the node supports.
+	pub request_response_protocols: Vec<RequestResponseConfig>,
+	/// Configuration for the default set of nodes used for block syncing and transactions.
+	pub default_peers_set: SetConfig,
+
+	/// Number of substreams to reserve for full nodes for block syncing and transactions.
+	/// Any other slot will be dedicated to light nodes.
+	///
+	/// This value is implicitly capped to `default_set.out_peers + default_set.in_peers`.
+	pub default_peers_set_num_full: u32,
+
+	/// Configuration for extra sets of nodes.
+	pub extra_sets: Vec<NonDefaultSetConfig>,
+
+	/// Client identifier. Sent over the wire for debugging purposes.
+	pub client_version: String,
+
+	/// Name of the node. Sent over the wire for debugging purposes.
+	pub node_name: String,
+
+	/// Configuration for the transport layer.
+	pub transport: TransportConfig,
+
+	/// Maximum number of peers to ask the same blocks in parallel.
+	pub max_parallel_downloads: u32,
+
+	/// Initial syncing mode.
+	pub sync_mode: SyncMode,
+
+	/// True if Kademlia random discovery should be enabled.
+	///
+	/// If true, the node will automatically randomly walk the DHT in order to find new peers.
+	pub enable_dht_random_walk: bool,
+
+	/// Should we insert non-global addresses into the DHT?
+	pub allow_non_globals_in_dht: bool,
+
+	/// Require iterative Kademlia DHT queries to use disjoint paths for increased resiliency in
+	/// the presence of potentially adversarial nodes.
+	pub kademlia_disjoint_query_paths: bool,
+
+	/// Enable serving block data over IPFS bitswap.
+	pub ipfs_server: bool,
+
+	/// Size of Yamux receive window of all substreams. `None` for the default (256kiB).
+	/// Any value less than 256kiB is invalid.
+	///
+	/// # Context
+	///
+	/// By design, notifications substreams on top of Yamux connections only allow up to `N` bytes
+	/// to be transferred at a time, where `N` is the Yamux receive window size configurable here.
+	/// This means, in practice, that every `N` bytes must be acknowledged by the receiver before
+	/// the sender can send more data. The maximum bandwidth of each notifications substream is
+	/// therefore `N / round_trip_time`.
+	///
+	/// It is recommended to leave this to `None`, and use a request-response protocol instead if
+	/// a large amount of data must be transferred. The reason why the value is configurable is
+	/// that some Substrate users mis-use notification protocols to send large amounts of data.
+	/// As such, this option isn't designed to stay and will likely get removed in the future.
+	///
+	/// Note that configuring a value here isn't a modification of the Yamux protocol, but rather
+	/// a modification of the way the implementation works. Different nodes with different
+	/// configured values remain compatible with each other.
+	pub yamux_window_size: Option<u32>,
+}
+
+impl NetworkConfiguration {
+	/// Create new default configuration
+	pub fn new<SN: Into<String>, SV: Into<String>>(
+		node_name: SN,
+		client_version: SV,
+		node_key: NodeKeyConfig,
+		net_config_path: Option<PathBuf>,
+	) -> Self {
+		let default_peers_set = SetConfig::default();
+		Self {
+			net_config_path,
+			listen_addresses: Vec::new(),
+			public_addresses: Vec::new(),
+			boot_nodes: Vec::new(),
+			node_key,
+			request_response_protocols: Vec::new(),
+			default_peers_set_num_full: default_peers_set.in_peers + default_peers_set.out_peers,
+			default_peers_set,
+			extra_sets: Vec::new(),
+			client_version: client_version.into(),
+			node_name: node_name.into(),
+			transport: TransportConfig::Normal { enable_mdns: false, allow_private_ip: true },
+			max_parallel_downloads: 5,
+			sync_mode: SyncMode::Full,
+			enable_dht_random_walk: true,
+			allow_non_globals_in_dht: false,
+			kademlia_disjoint_query_paths: false,
+			yamux_window_size: None,
+			ipfs_server: false,
+		}
+	}
+
+	/// Create new default configuration for localhost-only connection with random port (useful for
+	/// testing)
+	pub fn new_local() -> NetworkConfiguration {
+		let mut config =
+			NetworkConfiguration::new("test-node", "test-client", Default::default(), None);
+
+		config.listen_addresses =
+			vec![iter::once(multiaddr::Protocol::Ip4(Ipv4Addr::new(127, 0, 0, 1)))
+				.chain(iter::once(multiaddr::Protocol::Tcp(0)))
+				.collect()];
+
+		config.allow_non_globals_in_dht = true;
+		config
+	}
+
+	/// Create new default configuration for localhost-only connection with random port (useful for
+	/// testing)
+	pub fn new_memory() -> NetworkConfiguration {
+		let mut config =
+			NetworkConfiguration::new("test-node", "test-client", Default::default(), None);
+
+		config.listen_addresses =
+			vec![iter::once(multiaddr::Protocol::Ip4(Ipv4Addr::new(127, 0, 0, 1)))
+				.chain(iter::once(multiaddr::Protocol::Tcp(0)))
+				.collect()];
+
+		config.allow_non_globals_in_dht = true;
+		config
+	}
+}
 
 /// Network initialization parameters.
 pub struct Params<Client> {
@@ -67,3 +717,43 @@ pub struct Params<Client> {
 	/// Request response protocol configurations
 	pub request_response_protocol_configs: Vec<RequestResponseConfig>,
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use tempfile::TempDir;
+
+	fn tempdir_with_prefix(prefix: &str) -> TempDir {
+		tempfile::Builder::new().prefix(prefix).tempdir().unwrap()
+	}
+
+	fn secret_bytes(kp: &Keypair) -> Vec<u8> {
+		let Keypair::Ed25519(p) = kp;
+		p.secret().as_ref().iter().cloned().collect()
+	}
+
+	#[test]
+	fn test_secret_file() {
+		let tmp = tempdir_with_prefix("x");
+		std::fs::remove_dir(tmp.path()).unwrap(); // should be recreated
+		let file = tmp.path().join("x").to_path_buf();
+		let kp1 = NodeKeyConfig::Ed25519(Secret::File(file.clone())).into_keypair().unwrap();
+		let kp2 = NodeKeyConfig::Ed25519(Secret::File(file.clone())).into_keypair().unwrap();
+		assert!(file.is_file() && secret_bytes(&kp1) == secret_bytes(&kp2))
+	}
+
+	#[test]
+	fn test_secret_input() {
+		let sk = ed25519::SecretKey::generate();
+		let kp1 = NodeKeyConfig::Ed25519(Secret::Input(sk.clone())).into_keypair().unwrap();
+		let kp2 = NodeKeyConfig::Ed25519(Secret::Input(sk)).into_keypair().unwrap();
+		assert!(secret_bytes(&kp1) == secret_bytes(&kp2));
+	}
+
+	#[test]
+	fn test_secret_new() {
+		let kp1 = NodeKeyConfig::Ed25519(Secret::New).into_keypair().unwrap();
+		let kp2 = NodeKeyConfig::Ed25519(Secret::New).into_keypair().unwrap();
+		assert!(secret_bytes(&kp1) != secret_bytes(&kp2));
+	}
+}

substrate/client/network/src/discovery.rs

@@ -46,6 +46,8 @@
 //! active mechanism that asks nodes for the addresses they are listening on. Whenever we learn
 //! of a node's address, you must call `add_self_reported_address`.
 
+use crate::{config::ProtocolId, utils::LruHashSet};
+
 use array_bytes::bytes2hex;
 use futures::prelude::*;
 use futures_timer::Delay;
@@ -73,7 +75,6 @@ use libp2p::{
 	},
 };
 use log::{debug, info, trace, warn};
-use sc_network_common::{config::ProtocolId, utils::LruHashSet};
 use sp_core::hexdisplay::HexDisplay;
 use std::{
 	cmp,
@@ -904,6 +905,7 @@ mod tests {
 	use super::{
 		kademlia_protocol_name, legacy_kademlia_protocol_name, DiscoveryConfig, DiscoveryOut,
 	};
+	use crate::config::ProtocolId;
 	use futures::prelude::*;
 	use libp2p::{
 		core::{
@@ -915,7 +917,6 @@ mod tests {
 		swarm::{Executor, Swarm, SwarmEvent},
 		yamux, Multiaddr,
 	};
-	use sc_network_common::config::ProtocolId;
 	use sp_core::hash::H256;
 	use std::{collections::HashSet, pin::Pin, task::Poll};
 

substrate/client/network/src/error.rs

@@ -0,0 +1,78 @@
+// 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/>.
+
+//! Substrate network possible errors.
+
+use crate::{config::TransportConfig, types::ProtocolName};
+
+use libp2p::{Multiaddr, PeerId};
+
+use std::fmt;
+
+/// Result type alias for the network.
+pub type Result<T> = std::result::Result<T, Error>;
+
+/// Error type for the network.
+#[derive(thiserror::Error)]
+pub enum Error {
+	/// Io error
+	#[error(transparent)]
+	Io(#[from] std::io::Error),
+
+	/// Client error
+	#[error(transparent)]
+	Client(#[from] Box<sp_blockchain::Error>),
+	/// The same bootnode (based on address) is registered with two different peer ids.
+	#[error(
+		"The same bootnode (`{address}`) is registered with two different peer ids: `{first_id}` and `{second_id}`"
+	)]
+	DuplicateBootnode {
+		/// The address of the bootnode.
+		address: Multiaddr,
+		/// The first peer id that was found for the bootnode.
+		first_id: PeerId,
+		/// The second peer id that was found for the bootnode.
+		second_id: PeerId,
+	},
+	/// Prometheus metrics error.
+	#[error(transparent)]
+	Prometheus(#[from] prometheus_endpoint::PrometheusError),
+	/// The network addresses are invalid because they don't match the transport.
+	#[error(
+		"The following addresses are invalid because they don't match the transport: {addresses:?}"
+	)]
+	AddressesForAnotherTransport {
+		/// Transport used.
+		transport: TransportConfig,
+		/// The invalid addresses.
+		addresses: Vec<Multiaddr>,
+	},
+	/// The same request-response protocol has been registered multiple times.
+	#[error("Request-response protocol registered multiple times: {protocol}")]
+	DuplicateRequestResponseProtocol {
+		/// Name of the protocol registered multiple times.
+		protocol: ProtocolName,
+	},
+}
+
+// Make `Debug` use the `Display` implementation.
+impl fmt::Debug for Error {
+	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+		fmt::Display::fmt(self, f)
+	}
+}

substrate/client/network/src/event.rs

@@ -0,0 +1,92 @@
+// 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/>.
+
+//! Network event types. These are are not the part of the protocol, but rather
+//! events that happen on the network like DHT get/put results received.
+
+use crate::types::ProtocolName;
+
+use bytes::Bytes;
+use libp2p::{core::PeerId, kad::record::Key};
+
+use sc_network_common::role::ObservedRole;
+
+/// Events generated by DHT as a response to get_value and put_value requests.
+#[derive(Debug, Clone)]
+#[must_use]
+pub enum DhtEvent {
+	/// The value was found.
+	ValueFound(Vec<(Key, Vec<u8>)>),
+
+	/// The requested record has not been found in the DHT.
+	ValueNotFound(Key),
+
+	/// The record has been successfully inserted into the DHT.
+	ValuePut(Key),
+
+	/// An error has occurred while putting a record into the DHT.
+	ValuePutFailed(Key),
+}
+
+/// Type for events generated by networking layer.
+#[derive(Debug, Clone)]
+#[must_use]
+pub enum Event {
+	/// Event generated by a DHT.
+	Dht(DhtEvent),
+
+	/// Opened a substream with the given node with the given notifications protocol.
+	///
+	/// The protocol is always one of the notification protocols that have been registered.
+	NotificationStreamOpened {
+		/// Node we opened the substream with.
+		remote: PeerId,
+		/// The concerned protocol. Each protocol uses a different substream.
+		/// This is always equal to the value of
+		/// `sc_network::config::NonDefaultSetConfig::notifications_protocol` of one of the
+		/// configured sets.
+		protocol: ProtocolName,
+		/// If the negotiation didn't use the main name of the protocol (the one in
+		/// `notifications_protocol`), then this field contains which name has actually been
+		/// used.
+		/// Always contains a value equal to the value in
+		/// `sc_network::config::NonDefaultSetConfig::fallback_names`.
+		negotiated_fallback: Option<ProtocolName>,
+		/// Role of the remote.
+		role: ObservedRole,
+		/// Received handshake.
+		received_handshake: Vec<u8>,
+	},
+
+	/// Closed a substream with the given node. Always matches a corresponding previous
+	/// `NotificationStreamOpened` message.
+	NotificationStreamClosed {
+		/// Node we closed the substream with.
+		remote: PeerId,
+		/// The concerned protocol. Each protocol uses a different substream.
+		protocol: ProtocolName,
+	},
+
+	/// Received one or more messages from the given node using the given protocol.
+	NotificationsReceived {
+		/// Node we received the message from.
+		remote: PeerId,
+		/// Concerned protocol and associated message.
+		messages: Vec<(ProtocolName, Bytes)>,
+	},
+}

substrate/client/network/src/lib.rs

@@ -248,36 +248,40 @@ mod behaviour;
 mod discovery;
 mod peer_info;
 mod protocol;
-mod request_responses;
 mod service;
 mod transport;
 
 pub mod config;
+pub mod error;
+pub mod event;
 pub mod network_state;
+pub mod request_responses;
+pub mod types;
+pub mod utils;
 
+pub use event::{DhtEvent, Event};
 #[doc(inline)]
 pub use libp2p::{multiaddr, Multiaddr, PeerId};
+pub use request_responses::{IfDisconnected, RequestFailure, RequestResponseConfig};
 pub use sc_network_common::{
-	protocol::{
-		event::{DhtEvent, Event},
-		role::ObservedRole,
-		ProtocolName,
-	},
-	request_responses::{IfDisconnected, RequestFailure},
-	service::{
-		KademliaKey, NetworkBlock, NetworkDHTProvider, NetworkRequest, NetworkSigner,
-		NetworkStateInfo, NetworkStatus, NetworkStatusProvider, NetworkSyncForkRequest, Signature,
-		SigningError,
-	},
+	role::ObservedRole,
 	sync::{
 		warp::{WarpSyncPhase, WarpSyncProgress},
 		ExtendedPeerInfo, StateDownloadProgress, SyncEventStream, SyncState, SyncStatusProvider,
 	},
 };
 pub use service::{
-	DecodingError, Keypair, NetworkService, NetworkWorker, NotificationSender,
-	NotificationSenderReady, OutboundFailure, PublicKey,
+	signature::Signature,
+	traits::{
+		KademliaKey, NetworkBlock, NetworkDHTProvider, NetworkEventStream, NetworkNotification,
+		NetworkPeers, NetworkRequest, NetworkSigner, NetworkStateInfo, NetworkStatus,
+		NetworkStatusProvider, NetworkSyncForkRequest, NotificationSender as NotificationSenderT,
+		NotificationSenderError, NotificationSenderReady,
+	},
+	DecodingError, Keypair, NetworkService, NetworkWorker, NotificationSender, OutboundFailure,
+	PublicKey,
 };
+pub use types::ProtocolName;
 
 pub use sc_peerset::ReputationChange;
 

substrate/client/network/src/peer_info.rs

@@ -16,6 +16,8 @@
 // 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 crate::utils::interval;
+
 use fnv::FnvHashMap;
 use futures::prelude::*;
 use libp2p::{
@@ -36,8 +38,8 @@ use libp2p::{
 	Multiaddr,
 };
 use log::{debug, error, trace};
-use sc_network_common::utils::interval;
 use smallvec::SmallVec;
+
 use std::{
 	collections::hash_map::Entry,
 	pin::Pin,

substrate/client/network/src/protocol.rs

@@ -16,7 +16,11 @@
 // 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 crate::config;
+use crate::{
+	config::{self, NonReservedPeerMode},
+	error,
+	types::ProtocolName,
+};
 
 use bytes::Bytes;
 use codec::{DecodeAll, Encode};
@@ -29,27 +33,25 @@ use libp2p::{
 	Multiaddr, PeerId,
 };
 use log::{debug, error, warn};
-use message::{generic::Message as GenericMessage, Message};
-use notifications::{Notifications, NotificationsOut};
-use sc_network_common::{
-	config::NonReservedPeerMode,
-	error,
-	protocol::{role::Roles, ProtocolName},
-	sync::message::BlockAnnouncesHandshake,
-};
+
+use sc_network_common::{role::Roles, sync::message::BlockAnnouncesHandshake};
 use sp_runtime::traits::Block as BlockT;
+
 use std::{
 	collections::{HashMap, HashSet, VecDeque},
 	iter,
 	task::Poll,
 };
 
+use message::{generic::Message as GenericMessage, Message};
+use notifications::{Notifications, NotificationsOut};
+
+pub use notifications::{NotificationsSink, NotifsHandlerError, Ready};
+
 mod notifications;
 
 pub mod message;
 
-pub use notifications::{NotificationsSink, NotifsHandlerError, Ready};
-
 /// Maximum size used for notifications in the block announce and transaction protocols.
 // Must be equal to `max(MAX_BLOCK_ANNOUNCE_SIZE, MAX_TRANSACTIONS_SIZE)`.
 pub(crate) const BLOCK_ANNOUNCES_TRANSACTIONS_SUBSTREAM_SIZE: u64 = 16 * 1024 * 1024;
@@ -93,7 +95,7 @@ impl<B: BlockT> Protocol<B> {
 	pub fn new(
 		roles: Roles,
 		network_config: &config::NetworkConfiguration,
-		block_announces_protocol: sc_network_common::config::NonDefaultSetConfig,
+		block_announces_protocol: config::NonDefaultSetConfig,
 	) -> error::Result<(Self, sc_peerset::PeersetHandle, Vec<(PeerId, Multiaddr)>)> {
 		let mut known_addresses = Vec::new();
 

substrate/client/network/src/protocol/message.rs

@@ -61,7 +61,7 @@ pub mod generic {
 	use sc_client_api::StorageProof;
 	use sc_network_common::{
 		message::RequestId,
-		protocol::role::Roles,
+		role::Roles,
 		sync::message::{
 			generic::{BlockRequest, BlockResponse},
 			BlockAnnounce,

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

@@ -16,8 +16,11 @@
 // 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 crate::protocol::notifications::handler::{
-	self, NotificationsSink, NotifsHandlerIn, NotifsHandlerOut, NotifsHandlerProto,
+use crate::{
+	protocol::notifications::handler::{
+		self, NotificationsSink, NotifsHandlerIn, NotifsHandlerOut, NotifsHandlerProto,
+	},
+	types::ProtocolName,
 };
 
 use bytes::BytesMut;
@@ -35,7 +38,6 @@ use libp2p::{
 use log::{error, trace, warn};
 use parking_lot::RwLock;
 use rand::distributions::{Distribution as _, Uniform};
-use sc_network_common::protocol::ProtocolName;
 use sc_peerset::DropReason;
 use smallvec::SmallVec;
 use std::{

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

@@ -57,9 +57,12 @@
 //! It is illegal to send a [`NotifsHandlerIn::Open`] before a previously-emitted
 //! [`NotifsHandlerIn::Open`] has gotten an answer.
 
-use crate::protocol::notifications::upgrade::{
-	NotificationsIn, NotificationsInSubstream, NotificationsOut, NotificationsOutSubstream,
-	UpgradeCollec,
+use crate::{
+	protocol::notifications::upgrade::{
+		NotificationsIn, NotificationsInSubstream, NotificationsOut, NotificationsOutSubstream,
+		UpgradeCollec,
+	},
+	types::ProtocolName,
 };
 
 use bytes::BytesMut;
@@ -77,7 +80,6 @@ use libp2p::{
 };
 use log::error;
 use parking_lot::{Mutex, RwLock};
-use sc_network_common::protocol::ProtocolName;
 use std::{
 	collections::VecDeque,
 	mem,
@@ -945,8 +947,9 @@ pub mod tests {
 				Poll::Ready(Some(NotificationsSinkMessage::Notification { message })) =>
 					Poll::Ready(Some(message)),
 				Poll::Pending => Poll::Ready(None),
-				Poll::Ready(Some(NotificationsSinkMessage::ForceClose)) | Poll::Ready(None) =>
-					panic!("sink closed"),
+				Poll::Ready(Some(NotificationsSinkMessage::ForceClose)) | Poll::Ready(None) => {
+					panic!("sink closed")
+				},
 			})
 			.await
 		}

substrate/client/network/src/protocol/notifications/upgrade/collec.rs

@@ -103,8 +103,8 @@ impl<T: Future<Output = Result<O, E>>, O, E> Future for FutWithUsize<T> {
 #[cfg(test)]
 mod tests {
 	use super::*;
+	use crate::types::ProtocolName as ProtoName;
 	use libp2p::core::upgrade::{ProtocolName, UpgradeInfo};
-	use sc_network_common::protocol::ProtocolName as ProtoName;
 
 	// TODO: move to mocks
 	mockall::mock! {

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

@@ -16,7 +16,6 @@
 // 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 asynchronous_codec::Framed;
 /// Notifications protocol.
 ///
 /// The Substrate notifications protocol consists in the following:
@@ -35,11 +34,15 @@ use asynchronous_codec::Framed;
 ///
 /// Notification substreams are unidirectional. If A opens a substream with B, then B is
 /// encouraged but not required to open a substream to A as well.
+use crate::types::ProtocolName;
+
+use asynchronous_codec::Framed;
 use bytes::BytesMut;
 use futures::prelude::*;
 use libp2p::core::{upgrade, InboundUpgrade, OutboundUpgrade, UpgradeInfo};
 use log::{error, warn};
-use sc_network_common::protocol::ProtocolName;
+use unsigned_varint::codec::UviBytes;
+
 use std::{
 	convert::Infallible,
 	io, mem,
@@ -47,7 +50,6 @@ use std::{
 	task::{Context, Poll},
 	vec,
 };
-use unsigned_varint::codec::UviBytes;
 
 /// Maximum allowed size of the two handshake messages, in bytes.
 const MAX_HANDSHAKE_SIZE: usize = 1024;

substrate/client/network/src/request_responses.rs

@@ -34,7 +34,8 @@
 //! - If provided, a ["requests processing"](ProtocolConfig::inbound_queue) channel
 //! is used to handle incoming requests.
 
-use crate::ReputationChange;
+use crate::{types::ProtocolName, ReputationChange};
+
 use futures::{
 	channel::{mpsc, oneshot},
 	prelude::*,
@@ -43,7 +44,7 @@ use libp2p::{
 	core::{connection::ConnectionId, Multiaddr, PeerId},
 	request_response::{
 		handler::RequestResponseHandler, ProtocolSupport, RequestResponse, RequestResponseCodec,
-		RequestResponseConfig, RequestResponseEvent, RequestResponseMessage, ResponseChannel,
+		RequestResponseEvent, RequestResponseMessage, ResponseChannel,
 	},
 	swarm::{
 		behaviour::{ConnectionClosed, DialFailure, FromSwarm, ListenFailure},
@@ -52,12 +53,9 @@ use libp2p::{
 		PollParameters,
 	},
 };
-use sc_network_common::{
-	protocol::ProtocolName,
-	request_responses::{
-		IfDisconnected, IncomingRequest, OutgoingResponse, ProtocolConfig, RequestFailure,
-	},
-};
+
+use sc_peerset::{PeersetHandle, BANNED_THRESHOLD};
+
 use std::{
 	collections::{hash_map::Entry, HashMap},
 	io, iter,
@@ -66,8 +64,138 @@ use std::{
 	time::{Duration, Instant},
 };
 
-pub use libp2p::request_response::{InboundFailure, OutboundFailure, RequestId};
-use sc_peerset::{PeersetHandle, BANNED_THRESHOLD};
+pub use libp2p::request_response::{
+	InboundFailure, OutboundFailure, RequestId, RequestResponseConfig,
+};
+
+/// Error in a request.
+#[derive(Debug, thiserror::Error)]
+#[allow(missing_docs)]
+pub enum RequestFailure {
+	#[error("We are not currently connected to the requested peer.")]
+	NotConnected,
+	#[error("Given protocol hasn't been registered.")]
+	UnknownProtocol,
+	#[error("Remote has closed the substream before answering, thereby signaling that it considers the request as valid, but refused to answer it.")]
+	Refused,
+	#[error("The remote replied, but the local node is no longer interested in the response.")]
+	Obsolete,
+	#[error("Problem on the network: {0}")]
+	Network(OutboundFailure),
+}
+
+/// Configuration for a single request-response protocol.
+#[derive(Debug, Clone)]
+pub struct ProtocolConfig {
+	/// Name of the protocol on the wire. Should be something like `/foo/bar`.
+	pub name: ProtocolName,
+
+	/// Fallback on the wire protocol names to support.
+	pub fallback_names: Vec<ProtocolName>,
+
+	/// Maximum allowed size, in bytes, of a request.
+	///
+	/// Any request larger than this value will be declined as a way to avoid allocating too
+	/// much memory for it.
+	pub max_request_size: u64,
+
+	/// Maximum allowed size, in bytes, of a response.
+	///
+	/// Any response larger than this value will be declined as a way to avoid allocating too
+	/// much memory for it.
+	pub max_response_size: u64,
+
+	/// Duration after which emitted requests are considered timed out.
+	///
+	/// If you expect the response to come back quickly, you should set this to a smaller duration.
+	pub request_timeout: Duration,
+
+	/// Channel on which the networking service will send incoming requests.
+	///
+	/// Every time a peer sends a request to the local node using this protocol, the networking
+	/// service will push an element on this channel. The receiving side of this channel then has
+	/// to pull this element, process the request, and send back the response to send back to the
+	/// peer.
+	///
+	/// The size of the channel has to be carefully chosen. If the channel is full, the networking
+	/// service will discard the incoming request send back an error to the peer. Consequently,
+	/// the channel being full is an indicator that the node is overloaded.
+	///
+	/// You can typically set the size of the channel to `T / d`, where `T` is the
+	/// `request_timeout` and `d` is the expected average duration of CPU and I/O it takes to
+	/// build a response.
+	///
+	/// Can be `None` if the local node does not support answering incoming requests.
+	/// If this is `None`, then the local node will not advertise support for this protocol towards
+	/// other peers. If this is `Some` but the channel is closed, then the local node will
+	/// advertise support for this protocol, but any incoming request will lead to an error being
+	/// sent back.
+	pub inbound_queue: Option<mpsc::Sender<IncomingRequest>>,
+}
+
+/// A single request received by a peer on a request-response protocol.
+#[derive(Debug)]
+pub struct IncomingRequest {
+	/// Who sent the request.
+	pub peer: PeerId,
+
+	/// Request sent by the remote. Will always be smaller than
+	/// [`ProtocolConfig::max_request_size`].
+	pub payload: Vec<u8>,
+
+	/// Channel to send back the response.
+	///
+	/// There are two ways to indicate that handling the request failed:
+	///
+	/// 1. Drop `pending_response` and thus not changing the reputation of the peer.
+	///
+	/// 2. Sending an `Err(())` via `pending_response`, optionally including reputation changes for
+	/// the given peer.
+	pub pending_response: oneshot::Sender<OutgoingResponse>,
+}
+
+/// Response for an incoming request to be send by a request protocol handler.
+#[derive(Debug)]
+pub struct OutgoingResponse {
+	/// The payload of the response.
+	///
+	/// `Err(())` if none is available e.g. due an error while handling the request.
+	pub result: Result<Vec<u8>, ()>,
+
+	/// Reputation changes accrued while handling the request. To be applied to the reputation of
+	/// the peer sending the request.
+	pub reputation_changes: Vec<ReputationChange>,
+
+	/// If provided, the `oneshot::Sender` will be notified when the request has been sent to the
+	/// peer.
+	///
+	/// > **Note**: Operating systems typically maintain a buffer of a few dozen kilobytes of
+	/// >			outgoing data for each TCP socket, and it is not possible for a user
+	/// >			application to inspect this buffer. This channel here is not actually notified
+	/// >			when the response has been fully sent out, but rather when it has fully been
+	/// >			written to the buffer managed by the operating system.
+	pub sent_feedback: Option<oneshot::Sender<()>>,
+}
+
+/// When sending a request, what to do on a disconnected recipient.
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
+pub enum IfDisconnected {
+	/// Try to connect to the peer.
+	TryConnect,
+	/// Just fail if the destination is not yet connected.
+	ImmediateError,
+}
+
+/// Convenience functions for `IfDisconnected`.
+impl IfDisconnected {
+	/// Shall we connect to a disconnected peer?
+	pub fn should_connect(self) -> bool {
+		match self {
+			Self::TryConnect => true,
+			Self::ImmediateError => false,
+		}
+	}
+}
 
 /// Event generated by the [`RequestResponsesBehaviour`].
 #[derive(Debug)]
@@ -103,7 +231,12 @@ pub enum Event {
 	},
 
 	/// A request protocol handler issued reputation changes for the given peer.
-	ReputationChanges { peer: PeerId, changes: Vec<ReputationChange> },
+	ReputationChanges {
+		/// Peer whose reputation needs to be adjust.
+		peer: PeerId,
+		/// Reputation changes.
+		changes: Vec<ReputationChange>,
+	},
 }
 
 /// Combination of a protocol name and a request id.
@@ -344,7 +477,7 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
 						)
 					}
 				},
-			FromSwarm::DialFailure(DialFailure { peer_id, error, handler }) =>
+			FromSwarm::DialFailure(DialFailure { peer_id, error, handler }) => {
 				for (p_name, p_handler) in handler.into_iter() {
 					if let Some((proto, _)) = self.protocols.get_mut(p_name.as_str()) {
 						proto.on_swarm_event(FromSwarm::DialFailure(DialFailure {
@@ -359,7 +492,8 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
 						  p_name,
 						)
 					}
-				},
+				}
+			},
 			FromSwarm::ListenerClosed(e) =>
 				for (p, _) in self.protocols.values_mut() {
 					NetworkBehaviour::on_swarm_event(p, FromSwarm::ListenerClosed(e));

substrate/client/network/src/service.rs

@@ -29,13 +29,27 @@
 
 use crate::{
 	behaviour::{self, Behaviour, BehaviourOut},
-	config::Params,
+	config::{MultiaddrWithPeerId, Params, TransportConfig},
 	discovery::DiscoveryConfig,
+	error::Error,
+	event::{DhtEvent, Event},
 	network_state::{
 		NetworkState, NotConnectedPeer as NetworkStateNotConnectedPeer, Peer as NetworkStatePeer,
 	},
 	protocol::{self, NotificationsSink, NotifsHandlerError, Protocol, Ready},
-	transport, ReputationChange,
+	request_responses::{IfDisconnected, RequestFailure},
+	service::{
+		signature::{Signature, SigningError},
+		traits::{
+			NetworkDHTProvider, NetworkEventStream, NetworkNotification, NetworkPeers,
+			NetworkRequest, NetworkSigner, NetworkStateInfo, NetworkStatus, NetworkStatusProvider,
+			NotificationSender as NotificationSenderT, NotificationSenderError,
+			NotificationSenderReady as NotificationSenderReadyT,
+		},
+	},
+	transport,
+	types::ProtocolName,
+	ReputationChange,
 };
 
 use futures::{channel::oneshot, prelude::*};
@@ -55,26 +69,13 @@ use libp2p::{
 use log::{debug, error, info, trace, warn};
 use metrics::{Histogram, HistogramVec, MetricSources, Metrics};
 use parking_lot::Mutex;
-use sc_network_common::{
-	config::{MultiaddrWithPeerId, TransportConfig},
-	error::Error,
-	protocol::{
-		event::{DhtEvent, Event},
-		ProtocolName,
-	},
-	request_responses::{IfDisconnected, RequestFailure},
-	service::{
-		NetworkDHTProvider, NetworkEventStream, NetworkNotification, NetworkPeers, NetworkSigner,
-		NetworkStateInfo, NetworkStatus, NetworkStatusProvider,
-		NotificationSender as NotificationSenderT, NotificationSenderError,
-		NotificationSenderReady as NotificationSenderReadyT, Signature, SigningError,
-	},
-	ExHashT,
-};
+
+use sc_network_common::ExHashT;
 use sc_peerset::PeersetHandle;
 use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender};
 use sp_blockchain::HeaderBackend;
 use sp_runtime::traits::{Block as BlockT, Zero};
+
 use std::{
 	cmp,
 	collections::{HashMap, HashSet},
@@ -90,14 +91,13 @@ use std::{
 };
 
 pub use behaviour::{InboundFailure, OutboundFailure, ResponseFailure};
+pub use libp2p::identity::{error::DecodingError, Keypair, PublicKey};
 
 mod metrics;
 mod out_events;
-#[cfg(test)]
-mod tests;
 
-pub use libp2p::identity::{error::DecodingError, Keypair, PublicKey};
-use sc_network_common::service::NetworkRequest;
+pub mod signature;
+pub mod traits;
 
 /// Custom error that can be produced by the [`ConnectionHandler`] of the [`NetworkBehaviour`].
 /// Used as a template parameter of [`SwarmEvent`] below.
@@ -1432,10 +1432,11 @@ where
 						},
 					}
 				},
-			SwarmEvent::Behaviour(BehaviourOut::ReputationChanges { peer, changes }) =>
+			SwarmEvent::Behaviour(BehaviourOut::ReputationChanges { peer, changes }) => {
 				for change in changes {
 					self.network_service.behaviour().user_protocol().report_peer(peer, change);
-				},
+				}
+			},
 			SwarmEvent::Behaviour(BehaviourOut::PeerIdentify {
 				peer_id,
 				info:
@@ -1467,10 +1468,11 @@ where
 					.user_protocol_mut()
 					.add_default_set_discovered_nodes(iter::once(peer_id));
 			},
-			SwarmEvent::Behaviour(BehaviourOut::RandomKademliaStarted) =>
+			SwarmEvent::Behaviour(BehaviourOut::RandomKademliaStarted) => {
 				if let Some(metrics) = self.metrics.as_ref() {
 					metrics.kademlia_random_queries_total.inc();
-				},
+				}
+			},
 			SwarmEvent::Behaviour(BehaviourOut::NotificationStreamOpened {
 				remote,
 				protocol,

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

@@ -31,11 +31,12 @@
 //! - Send events by calling [`OutChannels::send`]. Events are cloned for each sender in the
 //! collection.
 
+use crate::event::Event;
+
 use futures::{prelude::*, ready, stream::FusedStream};
 use log::error;
 use parking_lot::Mutex;
 use prometheus_endpoint::{register, CounterVec, GaugeVec, Opts, PrometheusError, Registry, U64};
-use sc_network_common::protocol::event::Event;
 use std::{
 	backtrace::Backtrace,
 	cell::RefCell,

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

@@ -0,0 +1,55 @@
+// 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/>.
+//
+// If you read this, you are very thorough, congratulations.
+
+use libp2p::{
+	identity::{Keypair, PublicKey},
+	PeerId,
+};
+
+pub use libp2p::identity::error::SigningError;
+
+/// A result of signing a message with a network identity. Since `PeerId` is potentially a hash of a
+/// `PublicKey`, you need to reveal the `PublicKey` next to the signature, so the verifier can check
+/// if the signature was made by the entity that controls a given `PeerId`.
+pub struct Signature {
+	/// The public key derived from the network identity that signed the message.
+	pub public_key: PublicKey,
+	/// The actual signature made for the message signed.
+	pub bytes: Vec<u8>,
+}
+
+impl Signature {
+	/// Create a signature for a message with a given network identity.
+	pub fn sign_message(
+		message: impl AsRef<[u8]>,
+		keypair: &Keypair,
+	) -> Result<Self, SigningError> {
+		let public_key = keypair.public();
+		let bytes = keypair.sign(message.as_ref())?;
+		Ok(Self { public_key, bytes })
+	}
+
+	/// Verify whether the signature was made for the given message by the entity that controls the
+	/// given `PeerId`.
+	pub fn verify(&self, message: impl AsRef<[u8]>, peer_id: &PeerId) -> bool {
+		*peer_id == self.public_key.to_peer_id() &&
+			self.public_key.verify(message.as_ref(), &self.bytes)
+	}
+}

substrate/client/network/src/service/tests/mod.rs

@@ -1,240 +0,0 @@
-// 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 crate::{config, NetworkService, NetworkWorker};
-
-use futures::prelude::*;
-use libp2p::Multiaddr;
-use sc_consensus::{ImportQueue, Link};
-use sc_network_common::{
-	config::{NonDefaultSetConfig, ProtocolId, SetConfig, TransportConfig},
-	protocol::{event::Event, role::Roles},
-	service::NetworkEventStream,
-};
-use sc_network_light::light_client_requests::handler::LightClientRequestHandler;
-use sc_network_sync::{
-	block_request_handler::BlockRequestHandler,
-	engine::SyncingEngine,
-	service::network::{NetworkServiceHandle, NetworkServiceProvider},
-	state_request_handler::StateRequestHandler,
-};
-use sp_runtime::traits::Block as BlockT;
-use std::sync::Arc;
-use substrate_test_runtime_client::{
-	runtime::{Block as TestBlock, Hash as TestHash},
-	TestClientBuilder, TestClientBuilderExt as _,
-};
-
-#[cfg(test)]
-mod service;
-
-type TestNetworkWorker = NetworkWorker<TestBlock, TestHash>;
-type TestNetworkService = NetworkService<TestBlock, TestHash>;
-
-const PROTOCOL_NAME: &str = "/foo";
-
-struct TestNetwork {
-	network: TestNetworkWorker,
-}
-
-impl TestNetwork {
-	pub fn new(network: TestNetworkWorker) -> Self {
-		Self { network }
-	}
-
-	pub fn start_network(
-		self,
-	) -> (Arc<TestNetworkService>, (impl Stream<Item = Event> + std::marker::Unpin)) {
-		let worker = self.network;
-		let service = worker.service().clone();
-		let event_stream = service.event_stream("test");
-
-		tokio::spawn(worker.run());
-
-		(service, event_stream)
-	}
-}
-
-struct TestNetworkBuilder {
-	import_queue: Option<Box<dyn ImportQueue<TestBlock>>>,
-	link: Option<Box<dyn Link<TestBlock>>>,
-	client: Option<Arc<substrate_test_runtime_client::TestClient>>,
-	listen_addresses: Vec<Multiaddr>,
-	set_config: Option<SetConfig>,
-	chain_sync_network: Option<(NetworkServiceProvider, NetworkServiceHandle)>,
-	config: Option<config::NetworkConfiguration>,
-}
-
-impl TestNetworkBuilder {
-	pub fn new() -> Self {
-		Self {
-			import_queue: None,
-			link: None,
-			client: None,
-			listen_addresses: Vec::new(),
-			set_config: None,
-			chain_sync_network: None,
-			config: None,
-		}
-	}
-
-	pub fn with_config(mut self, config: config::NetworkConfiguration) -> Self {
-		self.config = Some(config);
-		self
-	}
-
-	pub fn with_listen_addresses(mut self, addresses: Vec<Multiaddr>) -> Self {
-		self.listen_addresses = addresses;
-		self
-	}
-
-	pub fn with_set_config(mut self, set_config: SetConfig) -> Self {
-		self.set_config = Some(set_config);
-		self
-	}
-
-	pub fn build(mut self) -> TestNetwork {
-		let client = self.client.as_mut().map_or(
-			Arc::new(TestClientBuilder::with_default_backend().build_with_longest_chain().0),
-			|v| v.clone(),
-		);
-
-		let network_config = self.config.unwrap_or(config::NetworkConfiguration {
-			extra_sets: vec![NonDefaultSetConfig {
-				notifications_protocol: PROTOCOL_NAME.into(),
-				fallback_names: Vec::new(),
-				max_notification_size: 1024 * 1024,
-				handshake: None,
-				set_config: self.set_config.unwrap_or_default(),
-			}],
-			listen_addresses: self.listen_addresses,
-			transport: TransportConfig::MemoryOnly,
-			..config::NetworkConfiguration::new_local()
-		});
-
-		#[derive(Clone)]
-		struct PassThroughVerifier(bool);
-
-		#[async_trait::async_trait]
-		impl<B: BlockT> sc_consensus::Verifier<B> for PassThroughVerifier {
-			async fn verify(
-				&mut self,
-				mut block: sc_consensus::BlockImportParams<B, ()>,
-			) -> Result<sc_consensus::BlockImportParams<B, ()>, String> {
-				block.finalized = self.0;
-				block.fork_choice = Some(sc_consensus::ForkChoiceStrategy::LongestChain);
-				Ok(block)
-			}
-		}
-
-		let mut import_queue =
-			self.import_queue.unwrap_or(Box::new(sc_consensus::BasicQueue::new(
-				PassThroughVerifier(false),
-				Box::new(client.clone()),
-				None,
-				&sp_core::testing::TaskExecutor::new(),
-				None,
-			)));
-
-		let protocol_id = ProtocolId::from("test-protocol-name");
-		let fork_id = Some(String::from("test-fork-id"));
-
-		let block_request_protocol_config = {
-			let (handler, protocol_config) =
-				BlockRequestHandler::new(&protocol_id, None, client.clone(), 50);
-			tokio::spawn(handler.run().boxed());
-			protocol_config
-		};
-
-		let state_request_protocol_config = {
-			let (handler, protocol_config) =
-				StateRequestHandler::new(&protocol_id, None, client.clone(), 50);
-			tokio::spawn(handler.run().boxed());
-			protocol_config
-		};
-
-		let light_client_request_protocol_config = {
-			let (handler, protocol_config) =
-				LightClientRequestHandler::new(&protocol_id, None, client.clone());
-			tokio::spawn(handler.run().boxed());
-			protocol_config
-		};
-
-		let (chain_sync_network_provider, chain_sync_network_handle) =
-			self.chain_sync_network.unwrap_or(NetworkServiceProvider::new());
-
-		let (engine, chain_sync_service, block_announce_config) = SyncingEngine::new(
-			Roles::from(&config::Role::Full),
-			client.clone(),
-			None,
-			&network_config,
-			protocol_id.clone(),
-			&None,
-			Box::new(sp_consensus::block_validation::DefaultBlockAnnounceValidator),
-			None,
-			chain_sync_network_handle,
-			import_queue.service(),
-			block_request_protocol_config.name.clone(),
-			state_request_protocol_config.name.clone(),
-			None,
-		)
-		.unwrap();
-		let mut link = self.link.unwrap_or(Box::new(chain_sync_service.clone()));
-		let worker = NetworkWorker::<
-			substrate_test_runtime_client::runtime::Block,
-			substrate_test_runtime_client::runtime::Hash,
-		>::new(config::Params {
-			block_announce_config,
-			role: config::Role::Full,
-			executor: Box::new(|f| {
-				tokio::spawn(f);
-			}),
-			network_config,
-			chain: client.clone(),
-			protocol_id,
-			fork_id,
-			metrics_registry: None,
-			request_response_protocol_configs: [
-				block_request_protocol_config,
-				state_request_protocol_config,
-				light_client_request_protocol_config,
-			]
-			.to_vec(),
-		})
-		.unwrap();
-
-		let service = worker.service().clone();
-		tokio::spawn(async move {
-			let _ = chain_sync_network_provider.run(service).await;
-		});
-		tokio::spawn(async move {
-			loop {
-				futures::future::poll_fn(|cx| {
-					import_queue.poll_actions(cx, &mut *link);
-					std::task::Poll::Ready(())
-				})
-				.await;
-				tokio::time::sleep(std::time::Duration::from_millis(250)).await;
-			}
-		});
-		let stream = worker.service().event_stream("syncing");
-		tokio::spawn(engine.run(stream));
-
-		TestNetwork::new(worker)
-	}
-}

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

@@ -1,586 +0,0 @@
-// 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 crate::{config, service::tests::TestNetworkBuilder, NetworkService};
-
-use futures::prelude::*;
-use libp2p::PeerId;
-use sc_network_common::{
-	config::{MultiaddrWithPeerId, NonDefaultSetConfig, SetConfig, TransportConfig},
-	protocol::event::Event,
-	service::{NetworkNotification, NetworkPeers, NetworkStateInfo},
-};
-use std::{sync::Arc, time::Duration};
-
-type TestNetworkService = NetworkService<
-	substrate_test_runtime_client::runtime::Block,
-	substrate_test_runtime_client::runtime::Hash,
->;
-
-const PROTOCOL_NAME: &str = "/foo";
-
-/// Builds two nodes and their associated events stream.
-/// The nodes are connected together and have the `PROTOCOL_NAME` protocol registered.
-fn build_nodes_one_proto() -> (
-	Arc<TestNetworkService>,
-	impl Stream<Item = Event>,
-	Arc<TestNetworkService>,
-	impl Stream<Item = Event>,
-) {
-	let listen_addr = config::build_multiaddr![Memory(rand::random::<u64>())];
-
-	let (node1, events_stream1) = TestNetworkBuilder::new()
-		.with_listen_addresses(vec![listen_addr.clone()])
-		.build()
-		.start_network();
-
-	let (node2, events_stream2) = TestNetworkBuilder::new()
-		.with_set_config(SetConfig {
-			reserved_nodes: vec![MultiaddrWithPeerId {
-				multiaddr: listen_addr,
-				peer_id: node1.local_peer_id(),
-			}],
-			..Default::default()
-		})
-		.build()
-		.start_network();
-
-	(node1, events_stream1, node2, events_stream2)
-}
-
-#[tokio::test]
-async fn notifications_state_consistent() {
-	// Runs two nodes and ensures that events are propagated out of the API in a consistent
-	// correct order, which means no notification received on a closed substream.
-
-	let (node1, mut events_stream1, node2, mut events_stream2) = build_nodes_one_proto();
-
-	// Write some initial notifications that shouldn't get through.
-	for _ in 0..(rand::random::<u8>() % 5) {
-		node1.write_notification(
-			node2.local_peer_id(),
-			PROTOCOL_NAME.into(),
-			b"hello world".to_vec(),
-		);
-	}
-	for _ in 0..(rand::random::<u8>() % 5) {
-		node2.write_notification(
-			node1.local_peer_id(),
-			PROTOCOL_NAME.into(),
-			b"hello world".to_vec(),
-		);
-	}
-
-	// True if we have an active substream from node1 to node2.
-	let mut node1_to_node2_open = false;
-	// True if we have an active substream from node2 to node1.
-	let mut node2_to_node1_open = false;
-	// We stop the test after a certain number of iterations.
-	let mut iterations = 0;
-	// Safe guard because we don't want the test to pass if no substream has been open.
-	let mut something_happened = false;
-
-	loop {
-		iterations += 1;
-		if iterations >= 1_000 {
-			assert!(something_happened);
-			break
-		}
-
-		// Start by sending a notification from node1 to node2 and vice-versa. Part of the
-		// test consists in ensuring that notifications get ignored if the stream isn't open.
-		if rand::random::<u8>() % 5 >= 3 {
-			node1.write_notification(
-				node2.local_peer_id(),
-				PROTOCOL_NAME.into(),
-				b"hello world".to_vec(),
-			);
-		}
-		if rand::random::<u8>() % 5 >= 3 {
-			node2.write_notification(
-				node1.local_peer_id(),
-				PROTOCOL_NAME.into(),
-				b"hello world".to_vec(),
-			);
-		}
-
-		// Also randomly disconnect the two nodes from time to time.
-		if rand::random::<u8>() % 20 == 0 {
-			node1.disconnect_peer(node2.local_peer_id(), PROTOCOL_NAME.into());
-		}
-		if rand::random::<u8>() % 20 == 0 {
-			node2.disconnect_peer(node1.local_peer_id(), PROTOCOL_NAME.into());
-		}
-
-		// Grab next event from either `events_stream1` or `events_stream2`.
-		let next_event = {
-			let next1 = events_stream1.next();
-			let next2 = events_stream2.next();
-			// We also await on a small timer, otherwise it is possible for the test to wait
-			// forever while nothing at all happens on the network.
-			let continue_test = futures_timer::Delay::new(Duration::from_millis(20));
-			match future::select(future::select(next1, next2), continue_test).await {
-				future::Either::Left((future::Either::Left((Some(ev), _)), _)) =>
-					future::Either::Left(ev),
-				future::Either::Left((future::Either::Right((Some(ev), _)), _)) =>
-					future::Either::Right(ev),
-				future::Either::Right(_) => continue,
-				_ => break,
-			}
-		};
-
-		match next_event {
-			future::Either::Left(Event::NotificationStreamOpened { remote, protocol, .. }) =>
-				if protocol == PROTOCOL_NAME.into() {
-					something_happened = true;
-					assert!(!node1_to_node2_open);
-					node1_to_node2_open = true;
-					assert_eq!(remote, node2.local_peer_id());
-				},
-			future::Either::Right(Event::NotificationStreamOpened { remote, protocol, .. }) =>
-				if protocol == PROTOCOL_NAME.into() {
-					something_happened = true;
-					assert!(!node2_to_node1_open);
-					node2_to_node1_open = true;
-					assert_eq!(remote, node1.local_peer_id());
-				},
-			future::Either::Left(Event::NotificationStreamClosed { remote, protocol, .. }) =>
-				if protocol == PROTOCOL_NAME.into() {
-					assert!(node1_to_node2_open);
-					node1_to_node2_open = false;
-					assert_eq!(remote, node2.local_peer_id());
-				},
-			future::Either::Right(Event::NotificationStreamClosed { remote, protocol, .. }) =>
-				if protocol == PROTOCOL_NAME.into() {
-					assert!(node2_to_node1_open);
-					node2_to_node1_open = false;
-					assert_eq!(remote, node1.local_peer_id());
-				},
-			future::Either::Left(Event::NotificationsReceived { remote, .. }) => {
-				assert!(node1_to_node2_open);
-				assert_eq!(remote, node2.local_peer_id());
-				if rand::random::<u8>() % 5 >= 4 {
-					node1.write_notification(
-						node2.local_peer_id(),
-						PROTOCOL_NAME.into(),
-						b"hello world".to_vec(),
-					);
-				}
-			},
-			future::Either::Right(Event::NotificationsReceived { remote, .. }) => {
-				assert!(node2_to_node1_open);
-				assert_eq!(remote, node1.local_peer_id());
-				if rand::random::<u8>() % 5 >= 4 {
-					node2.write_notification(
-						node1.local_peer_id(),
-						PROTOCOL_NAME.into(),
-						b"hello world".to_vec(),
-					);
-				}
-			},
-
-			// Add new events here.
-			future::Either::Left(Event::Dht(_)) => {},
-			future::Either::Right(Event::Dht(_)) => {},
-		};
-	}
-}
-
-#[tokio::test]
-async fn lots_of_incoming_peers_works() {
-	sp_tracing::try_init_simple();
-	let listen_addr = config::build_multiaddr![Memory(rand::random::<u64>())];
-
-	let (main_node, _) = TestNetworkBuilder::new()
-		.with_listen_addresses(vec![listen_addr.clone()])
-		.with_set_config(SetConfig { in_peers: u32::MAX, ..Default::default() })
-		.build()
-		.start_network();
-
-	let main_node_peer_id = main_node.local_peer_id();
-
-	// We spawn background tasks and push them in this `Vec`. They will all be waited upon before
-	// this test ends.
-	let mut background_tasks_to_wait = Vec::new();
-
-	for _ in 0..32 {
-		let (_dialing_node, event_stream) = TestNetworkBuilder::new()
-			.with_set_config(SetConfig {
-				reserved_nodes: vec![MultiaddrWithPeerId {
-					multiaddr: listen_addr.clone(),
-					peer_id: main_node_peer_id,
-				}],
-				..Default::default()
-			})
-			.build()
-			.start_network();
-
-		background_tasks_to_wait.push(tokio::spawn(async move {
-			// Create a dummy timer that will "never" fire, and that will be overwritten when we
-			// actually need the timer. Using an Option would be technically cleaner, but it would
-			// make the code below way more complicated.
-			let mut timer = futures_timer::Delay::new(Duration::from_secs(3600 * 24 * 7)).fuse();
-
-			let mut event_stream = event_stream.fuse();
-			let mut sync_protocol_name = None;
-			loop {
-				futures::select! {
-					_ = timer => {
-						// Test succeeds when timer fires.
-						return;
-					}
-					ev = event_stream.next() => {
-						match ev.unwrap() {
-							Event::NotificationStreamOpened { protocol, remote, .. } => {
-								if let None = sync_protocol_name {
-									sync_protocol_name = Some(protocol.clone());
-								}
-
-								assert_eq!(remote, main_node_peer_id);
-								// Test succeeds after 5 seconds. This timer is here in order to
-								// detect a potential problem after opening.
-								timer = futures_timer::Delay::new(Duration::from_secs(5)).fuse();
-							}
-							Event::NotificationStreamClosed { protocol, .. } => {
-								if Some(protocol) != sync_protocol_name {
-									// Test failed.
-									panic!();
-								}
-							}
-							_ => {}
-						}
-					}
-				}
-			}
-		}));
-	}
-
-	future::join_all(background_tasks_to_wait).await;
-}
-
-#[tokio::test]
-async fn notifications_back_pressure() {
-	// Node 1 floods node 2 with notifications. Random sleeps are done on node 2 to simulate the
-	// node being busy. We make sure that all notifications are received.
-
-	const TOTAL_NOTIFS: usize = 10_000;
-
-	let (node1, mut events_stream1, node2, mut events_stream2) = build_nodes_one_proto();
-	let node2_id = node2.local_peer_id();
-
-	let receiver = tokio::spawn(async move {
-		let mut received_notifications = 0;
-		let mut sync_protocol_name = None;
-
-		while received_notifications < TOTAL_NOTIFS {
-			match events_stream2.next().await.unwrap() {
-				Event::NotificationStreamOpened { protocol, .. } =>
-					if let None = sync_protocol_name {
-						sync_protocol_name = Some(protocol);
-					},
-				Event::NotificationStreamClosed { protocol, .. } => {
-					if Some(&protocol) != sync_protocol_name.as_ref() {
-						panic!()
-					}
-				},
-				Event::NotificationsReceived { messages, .. } =>
-					for message in messages {
-						assert_eq!(message.0, PROTOCOL_NAME.into());
-						assert_eq!(message.1, format!("hello #{}", received_notifications));
-						received_notifications += 1;
-					},
-				_ => {},
-			};
-
-			if rand::random::<u8>() < 2 {
-				tokio::time::sleep(Duration::from_millis(rand::random::<u64>() % 750)).await;
-			}
-		}
-	});
-
-	// Wait for the `NotificationStreamOpened`.
-	loop {
-		match events_stream1.next().await.unwrap() {
-			Event::NotificationStreamOpened { .. } => break,
-			_ => {},
-		};
-	}
-
-	// Sending!
-	for num in 0..TOTAL_NOTIFS {
-		let notif = node1.notification_sender(node2_id, PROTOCOL_NAME.into()).unwrap();
-		notif
-			.ready()
-			.await
-			.unwrap()
-			.send(format!("hello #{}", num).into_bytes())
-			.unwrap();
-	}
-
-	receiver.await.unwrap();
-}
-
-#[tokio::test]
-async fn fallback_name_working() {
-	// Node 1 supports the protocols "new" and "old". Node 2 only supports "old". Checks whether
-	// they can connect.
-	const NEW_PROTOCOL_NAME: &str = "/new-shiny-protocol-that-isnt-PROTOCOL_NAME";
-
-	let listen_addr = config::build_multiaddr![Memory(rand::random::<u64>())];
-	let (node1, mut events_stream1) = TestNetworkBuilder::new()
-		.with_config(config::NetworkConfiguration {
-			extra_sets: vec![NonDefaultSetConfig {
-				notifications_protocol: NEW_PROTOCOL_NAME.into(),
-				fallback_names: vec![PROTOCOL_NAME.into()],
-				max_notification_size: 1024 * 1024,
-				handshake: None,
-				set_config: Default::default(),
-			}],
-			listen_addresses: vec![listen_addr.clone()],
-			transport: TransportConfig::MemoryOnly,
-			..config::NetworkConfiguration::new_local()
-		})
-		.build()
-		.start_network();
-
-	let (_, mut events_stream2) = TestNetworkBuilder::new()
-		.with_set_config(SetConfig {
-			reserved_nodes: vec![MultiaddrWithPeerId {
-				multiaddr: listen_addr,
-				peer_id: node1.local_peer_id(),
-			}],
-			..Default::default()
-		})
-		.build()
-		.start_network();
-
-	let receiver = tokio::spawn(async move {
-		// Wait for the `NotificationStreamOpened`.
-		loop {
-			match events_stream2.next().await.unwrap() {
-				Event::NotificationStreamOpened { protocol, negotiated_fallback, .. } => {
-					assert_eq!(protocol, PROTOCOL_NAME.into());
-					assert_eq!(negotiated_fallback, None);
-					break
-				},
-				_ => {},
-			};
-		}
-	});
-
-	// Wait for the `NotificationStreamOpened`.
-	loop {
-		match events_stream1.next().await.unwrap() {
-			Event::NotificationStreamOpened { protocol, negotiated_fallback, .. }
-				if protocol == NEW_PROTOCOL_NAME.into() =>
-			{
-				assert_eq!(negotiated_fallback, Some(PROTOCOL_NAME.into()));
-				break
-			},
-			_ => {},
-		};
-	}
-
-	receiver.await.unwrap();
-}
-
-#[tokio::test]
-#[should_panic(expected = "don't match the transport")]
-async fn ensure_listen_addresses_consistent_with_transport_memory() {
-	let listen_addr = config::build_multiaddr![Ip4([127, 0, 0, 1]), Tcp(0_u16)];
-
-	let _ = TestNetworkBuilder::new()
-		.with_config(config::NetworkConfiguration {
-			listen_addresses: vec![listen_addr.clone()],
-			transport: TransportConfig::MemoryOnly,
-			..config::NetworkConfiguration::new(
-				"test-node",
-				"test-client",
-				Default::default(),
-				None,
-			)
-		})
-		.build()
-		.start_network();
-}
-
-#[tokio::test]
-#[should_panic(expected = "don't match the transport")]
-async fn ensure_listen_addresses_consistent_with_transport_not_memory() {
-	let listen_addr = config::build_multiaddr![Memory(rand::random::<u64>())];
-
-	let _ = TestNetworkBuilder::new()
-		.with_config(config::NetworkConfiguration {
-			listen_addresses: vec![listen_addr.clone()],
-			..config::NetworkConfiguration::new(
-				"test-node",
-				"test-client",
-				Default::default(),
-				None,
-			)
-		})
-		.build()
-		.start_network();
-}
-
-#[tokio::test]
-#[should_panic(expected = "don't match the transport")]
-async fn ensure_boot_node_addresses_consistent_with_transport_memory() {
-	let listen_addr = config::build_multiaddr![Memory(rand::random::<u64>())];
-	let boot_node = MultiaddrWithPeerId {
-		multiaddr: config::build_multiaddr![Ip4([127, 0, 0, 1]), Tcp(0_u16)],
-		peer_id: PeerId::random(),
-	};
-
-	let _ = TestNetworkBuilder::new()
-		.with_config(config::NetworkConfiguration {
-			listen_addresses: vec![listen_addr.clone()],
-			transport: TransportConfig::MemoryOnly,
-			boot_nodes: vec![boot_node],
-			..config::NetworkConfiguration::new(
-				"test-node",
-				"test-client",
-				Default::default(),
-				None,
-			)
-		})
-		.build()
-		.start_network();
-}
-
-#[tokio::test]
-#[should_panic(expected = "don't match the transport")]
-async fn ensure_boot_node_addresses_consistent_with_transport_not_memory() {
-	let listen_addr = config::build_multiaddr![Ip4([127, 0, 0, 1]), Tcp(0_u16)];
-	let boot_node = MultiaddrWithPeerId {
-		multiaddr: config::build_multiaddr![Memory(rand::random::<u64>())],
-		peer_id: PeerId::random(),
-	};
-
-	let _ = TestNetworkBuilder::new()
-		.with_config(config::NetworkConfiguration {
-			listen_addresses: vec![listen_addr.clone()],
-			boot_nodes: vec![boot_node],
-			..config::NetworkConfiguration::new(
-				"test-node",
-				"test-client",
-				Default::default(),
-				None,
-			)
-		})
-		.build()
-		.start_network();
-}
-
-#[tokio::test]
-#[should_panic(expected = "don't match the transport")]
-async fn ensure_reserved_node_addresses_consistent_with_transport_memory() {
-	let listen_addr = config::build_multiaddr![Memory(rand::random::<u64>())];
-	let reserved_node = MultiaddrWithPeerId {
-		multiaddr: config::build_multiaddr![Ip4([127, 0, 0, 1]), Tcp(0_u16)],
-		peer_id: PeerId::random(),
-	};
-
-	let _ = TestNetworkBuilder::new()
-		.with_config(config::NetworkConfiguration {
-			listen_addresses: vec![listen_addr.clone()],
-			transport: TransportConfig::MemoryOnly,
-			default_peers_set: SetConfig {
-				reserved_nodes: vec![reserved_node],
-				..Default::default()
-			},
-			..config::NetworkConfiguration::new(
-				"test-node",
-				"test-client",
-				Default::default(),
-				None,
-			)
-		})
-		.build()
-		.start_network();
-}
-
-#[tokio::test]
-#[should_panic(expected = "don't match the transport")]
-async fn ensure_reserved_node_addresses_consistent_with_transport_not_memory() {
-	let listen_addr = config::build_multiaddr![Ip4([127, 0, 0, 1]), Tcp(0_u16)];
-	let reserved_node = MultiaddrWithPeerId {
-		multiaddr: config::build_multiaddr![Memory(rand::random::<u64>())],
-		peer_id: PeerId::random(),
-	};
-
-	let _ = TestNetworkBuilder::new()
-		.with_config(config::NetworkConfiguration {
-			listen_addresses: vec![listen_addr.clone()],
-			default_peers_set: SetConfig {
-				reserved_nodes: vec![reserved_node],
-				..Default::default()
-			},
-			..config::NetworkConfiguration::new(
-				"test-node",
-				"test-client",
-				Default::default(),
-				None,
-			)
-		})
-		.build()
-		.start_network();
-}
-
-#[tokio::test]
-#[should_panic(expected = "don't match the transport")]
-async fn ensure_public_addresses_consistent_with_transport_memory() {
-	let listen_addr = config::build_multiaddr![Memory(rand::random::<u64>())];
-	let public_address = config::build_multiaddr![Ip4([127, 0, 0, 1]), Tcp(0_u16)];
-
-	let _ = TestNetworkBuilder::new()
-		.with_config(config::NetworkConfiguration {
-			listen_addresses: vec![listen_addr.clone()],
-			transport: TransportConfig::MemoryOnly,
-			public_addresses: vec![public_address],
-			..config::NetworkConfiguration::new(
-				"test-node",
-				"test-client",
-				Default::default(),
-				None,
-			)
-		})
-		.build()
-		.start_network();
-}
-
-#[tokio::test]
-#[should_panic(expected = "don't match the transport")]
-async fn ensure_public_addresses_consistent_with_transport_not_memory() {
-	let listen_addr = config::build_multiaddr![Ip4([127, 0, 0, 1]), Tcp(0_u16)];
-	let public_address = config::build_multiaddr![Memory(rand::random::<u64>())];
-
-	let _ = TestNetworkBuilder::new()
-		.with_config(config::NetworkConfiguration {
-			listen_addresses: vec![listen_addr.clone()],
-			public_addresses: vec![public_address],
-			..config::NetworkConfiguration::new(
-				"test-node",
-				"test-client",
-				Default::default(),
-				None,
-			)
-		})
-		.build()
-		.start_network();
-}

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

@@ -0,0 +1,634 @@
+// 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/>.
+//
+// If you read this, you are very thorough, congratulations.
+
+use crate::{
+	config::MultiaddrWithPeerId,
+	event::Event,
+	request_responses::{IfDisconnected, RequestFailure},
+	service::signature::Signature,
+	types::ProtocolName,
+};
+
+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::error::SigningError, kad::record::Key as KademliaKey};
+
+/// Signer with network identity
+pub trait NetworkSigner {
+	/// Signs the message with the `KeyPair` that defines the local [`PeerId`].
+	fn sign_with_local_identity(&self, msg: impl AsRef<[u8]>) -> Result<Signature, SigningError>;
+}
+
+impl<T> NetworkSigner for Arc<T>
+where
+	T: ?Sized,
+	T: NetworkSigner,
+{
+	fn sign_with_local_identity(&self, msg: impl AsRef<[u8]>) -> Result<Signature, SigningError> {
+		T::sign_with_local_identity(self, msg)
+	}
+}
+
+/// Provides access to the networking DHT.
+pub trait NetworkDHTProvider {
+	/// Start getting a value from the DHT.
+	fn get_value(&self, key: &KademliaKey);
+
+	/// Start putting a value in the DHT.
+	fn put_value(&self, key: KademliaKey, value: Vec<u8>);
+}
+
+impl<T> NetworkDHTProvider for Arc<T>
+where
+	T: ?Sized,
+	T: NetworkDHTProvider,
+{
+	fn get_value(&self, key: &KademliaKey) {
+		T::get_value(self, key)
+	}
+
+	fn put_value(&self, key: KademliaKey, value: Vec<u8>) {
+		T::put_value(self, key, value)
+	}
+}
+
+/// Provides an ability to set a fork sync request for a particular block.
+pub trait NetworkSyncForkRequest<BlockHash, BlockNumber> {
+	/// Notifies the sync service to try and sync the given block from the given
+	/// peers.
+	///
+	/// If the given vector of peers is empty then the underlying implementation
+	/// should make a best effort to fetch the block from any peers it is
+	/// connected to (NOTE: this assumption will change in the future #3629).
+	fn set_sync_fork_request(&self, peers: Vec<PeerId>, hash: BlockHash, number: BlockNumber);
+}
+
+impl<T, BlockHash, BlockNumber> NetworkSyncForkRequest<BlockHash, BlockNumber> for Arc<T>
+where
+	T: ?Sized,
+	T: NetworkSyncForkRequest<BlockHash, BlockNumber>,
+{
+	fn set_sync_fork_request(&self, peers: Vec<PeerId>, hash: BlockHash, number: BlockNumber) {
+		T::set_sync_fork_request(self, peers, hash, number)
+	}
+}
+
+/// Overview status of the network.
+#[derive(Clone)]
+pub struct NetworkStatus {
+	/// Total number of connected peers.
+	pub num_connected_peers: usize,
+	/// The total number of bytes received.
+	pub total_bytes_inbound: u64,
+	/// The total number of bytes sent.
+	pub total_bytes_outbound: u64,
+}
+
+/// Provides high-level status information about network.
+#[async_trait::async_trait]
+pub trait NetworkStatusProvider {
+	/// High-level network status information.
+	///
+	/// Returns an error if the `NetworkWorker` is no longer running.
+	async fn status(&self) -> Result<NetworkStatus, ()>;
+}
+
+// Manual implementation to avoid extra boxing here
+impl<T> NetworkStatusProvider for Arc<T>
+where
+	T: ?Sized,
+	T: NetworkStatusProvider,
+{
+	fn status<'life0, 'async_trait>(
+		&'life0 self,
+	) -> Pin<Box<dyn Future<Output = Result<NetworkStatus, ()>> + Send + 'async_trait>>
+	where
+		'life0: 'async_trait,
+		Self: 'async_trait,
+	{
+		T::status(self)
+	}
+}
+
+/// Provides low-level API for manipulating network peers.
+pub trait NetworkPeers {
+	/// Set authorized peers.
+	///
+	/// Need a better solution to manage authorized peers, but now just use reserved peers for
+	/// prototyping.
+	fn set_authorized_peers(&self, peers: HashSet<PeerId>);
+
+	/// Set authorized_only flag.
+	///
+	/// Need a better solution to decide authorized_only, but now just use reserved_only flag for
+	/// prototyping.
+	fn set_authorized_only(&self, reserved_only: bool);
+
+	/// Adds an address known to a node.
+	fn add_known_address(&self, peer_id: PeerId, addr: Multiaddr);
+
+	/// Report a given peer as either beneficial (+) or costly (-) according to the
+	/// given scalar.
+	fn report_peer(&self, who: PeerId, cost_benefit: ReputationChange);
+
+	/// Disconnect from a node as soon as possible.
+	///
+	/// This triggers the same effects as if the connection had closed itself spontaneously.
+	///
+	/// See also [`NetworkPeers::remove_from_peers_set`], which has the same effect but also
+	/// prevents the local node from re-establishing an outgoing substream to this peer until it
+	/// is added again.
+	fn disconnect_peer(&self, who: PeerId, protocol: ProtocolName);
+
+	/// Connect to unreserved peers and allow unreserved peers to connect for syncing purposes.
+	fn accept_unreserved_peers(&self);
+
+	/// Disconnect from unreserved peers and deny new unreserved peers to connect for syncing
+	/// purposes.
+	fn deny_unreserved_peers(&self);
+
+	/// Adds a `PeerId` and its `Multiaddr` as reserved for a sync protocol (default peer set).
+	///
+	/// Returns an `Err` if the given string is not a valid multiaddress
+	/// or contains an invalid peer ID (which includes the local peer ID).
+	fn add_reserved_peer(&self, peer: MultiaddrWithPeerId) -> Result<(), String>;
+
+	/// Removes a `PeerId` from the list of reserved peers for a sync protocol (default peer set).
+	fn remove_reserved_peer(&self, peer_id: PeerId);
+
+	/// Sets the reserved set of a protocol to the given set of peers.
+	///
+	/// Each `Multiaddr` must end with a `/p2p/` component containing the `PeerId`. It can also
+	/// consist of only `/p2p/<peerid>`.
+	///
+	/// The node will start establishing/accepting connections and substreams to/from peers in this
+	/// set, if it doesn't have any substream open with them yet.
+	///
+	/// Note however, if a call to this function results in less peers on the reserved set, they
+	/// will not necessarily get disconnected (depending on available free slots in the peer set).
+	/// If you want to also disconnect those removed peers, you will have to call
+	/// `remove_from_peers_set` on those in addition to updating the reserved set. You can omit
+	/// this step if the peer set is in reserved only mode.
+	///
+	/// Returns an `Err` if one of the given addresses is invalid or contains an
+	/// invalid peer ID (which includes the local peer ID).
+	fn set_reserved_peers(
+		&self,
+		protocol: ProtocolName,
+		peers: HashSet<Multiaddr>,
+	) -> Result<(), String>;
+
+	/// Add peers to a peer set.
+	///
+	/// Each `Multiaddr` must end with a `/p2p/` component containing the `PeerId`. It can also
+	/// consist of only `/p2p/<peerid>`.
+	///
+	/// Returns an `Err` if one of the given addresses is invalid or contains an
+	/// invalid peer ID (which includes the local peer ID).
+	fn add_peers_to_reserved_set(
+		&self,
+		protocol: ProtocolName,
+		peers: HashSet<Multiaddr>,
+	) -> Result<(), String>;
+
+	/// Remove peers from a peer set.
+	fn remove_peers_from_reserved_set(&self, protocol: ProtocolName, peers: Vec<PeerId>);
+
+	/// Add a peer to a set of peers.
+	///
+	/// If the set has slots available, it will try to open a substream with this peer.
+	///
+	/// Each `Multiaddr` must end with a `/p2p/` component containing the `PeerId`. It can also
+	/// consist of only `/p2p/<peerid>`.
+	///
+	/// Returns an `Err` if one of the given addresses is invalid or contains an
+	/// invalid peer ID (which includes the local peer ID).
+	fn add_to_peers_set(
+		&self,
+		protocol: ProtocolName,
+		peers: HashSet<Multiaddr>,
+	) -> Result<(), String>;
+
+	/// Remove peers from a peer set.
+	///
+	/// If we currently have an open substream with this peer, it will soon be closed.
+	fn remove_from_peers_set(&self, protocol: ProtocolName, peers: Vec<PeerId>);
+
+	/// Returns the number of peers in the sync peer set we're connected to.
+	fn sync_num_connected(&self) -> usize;
+}
+
+// Manual implementation to avoid extra boxing here
+impl<T> NetworkPeers for Arc<T>
+where
+	T: ?Sized,
+	T: NetworkPeers,
+{
+	fn set_authorized_peers(&self, peers: HashSet<PeerId>) {
+		T::set_authorized_peers(self, peers)
+	}
+
+	fn set_authorized_only(&self, reserved_only: bool) {
+		T::set_authorized_only(self, reserved_only)
+	}
+
+	fn add_known_address(&self, peer_id: PeerId, addr: Multiaddr) {
+		T::add_known_address(self, peer_id, addr)
+	}
+
+	fn report_peer(&self, who: PeerId, cost_benefit: ReputationChange) {
+		T::report_peer(self, who, cost_benefit)
+	}
+
+	fn disconnect_peer(&self, who: PeerId, protocol: ProtocolName) {
+		T::disconnect_peer(self, who, protocol)
+	}
+
+	fn accept_unreserved_peers(&self) {
+		T::accept_unreserved_peers(self)
+	}
+
+	fn deny_unreserved_peers(&self) {
+		T::deny_unreserved_peers(self)
+	}
+
+	fn add_reserved_peer(&self, peer: MultiaddrWithPeerId) -> Result<(), String> {
+		T::add_reserved_peer(self, peer)
+	}
+
+	fn remove_reserved_peer(&self, peer_id: PeerId) {
+		T::remove_reserved_peer(self, peer_id)
+	}
+
+	fn set_reserved_peers(
+		&self,
+		protocol: ProtocolName,
+		peers: HashSet<Multiaddr>,
+	) -> Result<(), String> {
+		T::set_reserved_peers(self, protocol, peers)
+	}
+
+	fn add_peers_to_reserved_set(
+		&self,
+		protocol: ProtocolName,
+		peers: HashSet<Multiaddr>,
+	) -> Result<(), String> {
+		T::add_peers_to_reserved_set(self, protocol, peers)
+	}
+
+	fn remove_peers_from_reserved_set(&self, protocol: ProtocolName, peers: Vec<PeerId>) {
+		T::remove_peers_from_reserved_set(self, protocol, peers)
+	}
+
+	fn add_to_peers_set(
+		&self,
+		protocol: ProtocolName,
+		peers: HashSet<Multiaddr>,
+	) -> Result<(), String> {
+		T::add_to_peers_set(self, protocol, peers)
+	}
+
+	fn remove_from_peers_set(&self, protocol: ProtocolName, peers: Vec<PeerId>) {
+		T::remove_from_peers_set(self, protocol, peers)
+	}
+
+	fn sync_num_connected(&self) -> usize {
+		T::sync_num_connected(self)
+	}
+}
+
+/// Provides access to network-level event stream.
+pub trait NetworkEventStream {
+	/// Returns a stream containing the events that happen on the network.
+	///
+	/// If this method is called multiple times, the events are duplicated.
+	///
+	/// The stream never ends (unless the `NetworkWorker` gets shut down).
+	///
+	/// The name passed is used to identify the channel in the Prometheus metrics. Note that the
+	/// parameter is a `&'static str`, and not a `String`, in order to avoid accidentally having
+	/// an unbounded set of Prometheus metrics, which would be quite bad in terms of memory
+	fn event_stream(&self, name: &'static str) -> Pin<Box<dyn Stream<Item = Event> + Send>>;
+}
+
+impl<T> NetworkEventStream for Arc<T>
+where
+	T: ?Sized,
+	T: NetworkEventStream,
+{
+	fn event_stream(&self, name: &'static str) -> Pin<Box<dyn Stream<Item = Event> + Send>> {
+		T::event_stream(self, name)
+	}
+}
+
+/// Trait for providing information about the local network state
+pub trait NetworkStateInfo {
+	/// Returns the local external addresses.
+	fn external_addresses(&self) -> Vec<Multiaddr>;
+
+	/// Returns the listening addresses (without trailing `/p2p/` with our `PeerId`).
+	fn listen_addresses(&self) -> Vec<Multiaddr>;
+
+	/// Returns the local Peer ID.
+	fn local_peer_id(&self) -> PeerId;
+}
+
+impl<T> NetworkStateInfo for Arc<T>
+where
+	T: ?Sized,
+	T: NetworkStateInfo,
+{
+	fn external_addresses(&self) -> Vec<Multiaddr> {
+		T::external_addresses(self)
+	}
+
+	fn listen_addresses(&self) -> Vec<Multiaddr> {
+		T::listen_addresses(self)
+	}
+
+	fn local_peer_id(&self) -> PeerId {
+		T::local_peer_id(self)
+	}
+}
+
+/// Reserved slot in the notifications buffer, ready to accept data.
+pub trait NotificationSenderReady {
+	/// Consumes this slots reservation and actually queues the notification.
+	///
+	/// NOTE: Traits can't consume itself, but calling this method second time will return an error.
+	fn send(&mut self, notification: Vec<u8>) -> Result<(), NotificationSenderError>;
+}
+
+/// A `NotificationSender` allows for sending notifications to a peer with a chosen protocol.
+#[async_trait::async_trait]
+pub trait NotificationSender: Send + Sync + 'static {
+	/// Returns a future that resolves when the `NotificationSender` is ready to send a
+	/// notification.
+	async fn ready(&self)
+		-> Result<Box<dyn NotificationSenderReady + '_>, NotificationSenderError>;
+}
+
+/// Error returned by [`NetworkNotification::notification_sender`].
+#[derive(Debug, thiserror::Error)]
+pub enum NotificationSenderError {
+	/// The notification receiver has been closed, usually because the underlying connection
+	/// closed.
+	///
+	/// Some of the notifications most recently sent may not have been received. However,
+	/// the peer may still be connected and a new `NotificationSender` for the same
+	/// protocol obtained from [`NetworkNotification::notification_sender`].
+	#[error("The notification receiver has been closed")]
+	Closed,
+	/// Protocol name hasn't been registered.
+	#[error("Protocol name hasn't been registered")]
+	BadProtocol,
+}
+
+/// Provides ability to send network notifications.
+pub trait NetworkNotification {
+	/// Appends a notification to the buffer of pending outgoing notifications with the given peer.
+	/// Has no effect if the notifications channel with this protocol name is not open.
+	///
+	/// If the buffer of pending outgoing notifications with that peer is full, the notification
+	/// is silently dropped and the connection to the remote will start being shut down. This
+	/// happens if you call this method at a higher rate than the rate at which the peer processes
+	/// these notifications, or if the available network bandwidth is too low.
+	///
+	/// For this reason, this method is considered soft-deprecated. You are encouraged to use
+	/// [`NetworkNotification::notification_sender`] instead.
+	///
+	/// > **Note**: The reason why this is a no-op in the situation where we have no channel is
+	/// >			that we don't guarantee message delivery anyway. Networking issues can cause
+	/// >			connections to drop at any time, and higher-level logic shouldn't differentiate
+	/// >			between the remote voluntarily closing a substream or a network error
+	/// >			preventing the message from being delivered.
+	///
+	/// The protocol must have been registered with
+	/// `crate::config::NetworkConfiguration::notifications_protocols`.
+	fn write_notification(&self, target: PeerId, protocol: ProtocolName, message: Vec<u8>);
+
+	/// Obtains a [`NotificationSender`] for a connected peer, if it exists.
+	///
+	/// A `NotificationSender` is scoped to a particular connection to the peer that holds
+	/// a receiver. With a `NotificationSender` at hand, sending a notification is done in two
+	/// steps:
+	///
+	/// 1.  [`NotificationSender::ready`] is used to wait for the sender to become ready
+	/// for another notification, yielding a [`NotificationSenderReady`] token.
+	/// 2.  [`NotificationSenderReady::send`] enqueues the notification for sending. This operation
+	/// can only fail if the underlying notification substream or connection has suddenly closed.
+	///
+	/// An error is returned by [`NotificationSenderReady::send`] if there exists no open
+	/// notifications substream with that combination of peer and protocol, or if the remote
+	/// has asked to close the notifications substream. If that happens, it is guaranteed that an
+	/// [`Event::NotificationStreamClosed`] has been generated on the stream returned by
+	/// [`NetworkEventStream::event_stream`].
+	///
+	/// If the remote requests to close the notifications substream, all notifications successfully
+	/// enqueued using [`NotificationSenderReady::send`] will finish being sent out before the
+	/// substream actually gets closed, but attempting to enqueue more notifications will now
+	/// return an error. It is however possible for the entire connection to be abruptly closed,
+	/// in which case enqueued notifications will be lost.
+	///
+	/// The protocol must have been registered with
+	/// `crate::config::NetworkConfiguration::notifications_protocols`.
+	///
+	/// # Usage
+	///
+	/// This method returns a struct that allows waiting until there is space available in the
+	/// buffer of messages towards the given peer. If the peer processes notifications at a slower
+	/// rate than we send them, this buffer will quickly fill up.
+	///
+	/// As such, you should never do something like this:
+	///
+	/// ```ignore
+	/// // Do NOT do this
+	/// for peer in peers {
+	/// 	if let Ok(n) = network.notification_sender(peer, ...) {
+	/// 			if let Ok(s) = n.ready().await {
+	/// 				let _ = s.send(...);
+	/// 			}
+	/// 	}
+	/// }
+	/// ```
+	///
+	/// Doing so would slow down all peers to the rate of the slowest one. A malicious or
+	/// malfunctioning peer could intentionally process notifications at a very slow rate.
+	///
+	/// Instead, you are encouraged to maintain your own buffer of notifications on top of the one
+	/// maintained by `sc-network`, and use `notification_sender` to progressively send out
+	/// elements from your buffer. If this additional buffer is full (which will happen at some
+	/// point if the peer is too slow to process notifications), appropriate measures can be taken,
+	/// such as removing non-critical notifications from the buffer or disconnecting the peer
+	/// using [`NetworkPeers::disconnect_peer`].
+	///
+	///
+	/// Notifications              Per-peer buffer
+	///   broadcast    +------->   of notifications   +-->  `notification_sender`  +-->  Internet
+	///                    ^       (not covered by
+	///                    |         sc-network)
+	///                    +
+	///      Notifications should be dropped
+	///             if buffer is full
+	///
+	///
+	/// See also the `sc-network-gossip` crate for a higher-level way to send notifications.
+	fn notification_sender(
+		&self,
+		target: PeerId,
+		protocol: ProtocolName,
+	) -> Result<Box<dyn NotificationSender>, NotificationSenderError>;
+
+	/// Set handshake for the notification protocol.
+	fn set_notification_handshake(&self, protocol: ProtocolName, handshake: Vec<u8>);
+}
+
+impl<T> NetworkNotification for Arc<T>
+where
+	T: ?Sized,
+	T: NetworkNotification,
+{
+	fn write_notification(&self, target: PeerId, protocol: ProtocolName, message: Vec<u8>) {
+		T::write_notification(self, target, protocol, message)
+	}
+
+	fn notification_sender(
+		&self,
+		target: PeerId,
+		protocol: ProtocolName,
+	) -> Result<Box<dyn NotificationSender>, NotificationSenderError> {
+		T::notification_sender(self, target, protocol)
+	}
+
+	fn set_notification_handshake(&self, protocol: ProtocolName, handshake: Vec<u8>) {
+		T::set_notification_handshake(self, protocol, handshake)
+	}
+}
+
+/// Provides ability to send network requests.
+#[async_trait::async_trait]
+pub trait NetworkRequest {
+	/// Sends a single targeted request to a specific peer. On success, returns the response of
+	/// the peer.
+	///
+	/// Request-response protocols are a way to complement notifications protocols, but
+	/// notifications should remain the default ways of communicating information. For example, a
+	/// peer can announce something through a notification, after which the recipient can obtain
+	/// more information by performing a request.
+	/// As such, call this function with `IfDisconnected::ImmediateError` for `connect`. This way
+	/// you will get an error immediately for disconnected peers, instead of waiting for a
+	/// potentially very long connection attempt, which would suggest that something is wrong
+	/// anyway, as you are supposed to be connected because of the notification protocol.
+	///
+	/// No limit or throttling of concurrent outbound requests per peer and protocol are enforced.
+	/// Such restrictions, if desired, need to be enforced at the call site(s).
+	///
+	/// The protocol must have been registered through
+	/// `NetworkConfiguration::request_response_protocols`.
+	async fn request(
+		&self,
+		target: PeerId,
+		protocol: ProtocolName,
+		request: Vec<u8>,
+		connect: IfDisconnected,
+	) -> Result<Vec<u8>, RequestFailure>;
+
+	/// Variation of `request` which starts a request whose response is delivered on a provided
+	/// channel.
+	///
+	/// Instead of blocking and waiting for a reply, this function returns immediately, sending
+	/// responses via the passed in sender. This alternative API exists to make it easier to
+	/// integrate with message passing APIs.
+	///
+	/// Keep in mind that the connected receiver might receive a `Canceled` event in case of a
+	/// closing connection. This is expected behaviour. With `request` you would get a
+	/// `RequestFailure::Network(OutboundFailure::ConnectionClosed)` in that case.
+	fn start_request(
+		&self,
+		target: PeerId,
+		protocol: ProtocolName,
+		request: Vec<u8>,
+		tx: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
+		connect: IfDisconnected,
+	);
+}
+
+// Manual implementation to avoid extra boxing here
+impl<T> NetworkRequest for Arc<T>
+where
+	T: ?Sized,
+	T: NetworkRequest,
+{
+	fn request<'life0, 'async_trait>(
+		&'life0 self,
+		target: PeerId,
+		protocol: ProtocolName,
+		request: Vec<u8>,
+		connect: IfDisconnected,
+	) -> Pin<Box<dyn Future<Output = Result<Vec<u8>, RequestFailure>> + Send + 'async_trait>>
+	where
+		'life0: 'async_trait,
+		Self: 'async_trait,
+	{
+		T::request(self, target, protocol, request, connect)
+	}
+
+	fn start_request(
+		&self,
+		target: PeerId,
+		protocol: ProtocolName,
+		request: Vec<u8>,
+		tx: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
+		connect: IfDisconnected,
+	) {
+		T::start_request(self, target, protocol, request, tx, connect)
+	}
+}
+
+/// Provides ability to announce blocks to the network.
+pub trait NetworkBlock<BlockHash, BlockNumber> {
+	/// Make sure an important block is propagated to peers.
+	///
+	/// In chain-based consensus, we often need to make sure non-best forks are
+	/// at least temporarily synced. This function forces such an announcement.
+	fn announce_block(&self, hash: BlockHash, data: Option<Vec<u8>>);
+
+	/// Inform the network service about new best imported block.
+	fn new_best_block_imported(&self, hash: BlockHash, number: BlockNumber);
+}
+
+impl<T, BlockHash, BlockNumber> NetworkBlock<BlockHash, BlockNumber> for Arc<T>
+where
+	T: ?Sized,
+	T: NetworkBlock<BlockHash, BlockNumber>,
+{
+	fn announce_block(&self, hash: BlockHash, data: Option<Vec<u8>>) {
+		T::announce_block(self, hash, data)
+	}
+
+	fn new_best_block_imported(&self, hash: BlockHash, number: BlockNumber) {
+		T::new_best_block_imported(self, hash, number)
+	}
+}

substrate/client/network/src/types.rs

@@ -0,0 +1,146 @@
+// 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/>.
+
+//! `sc-network` type definitions
+
+use libp2p::core::upgrade;
+
+use std::{
+	borrow::Borrow,
+	fmt,
+	hash::{Hash, Hasher},
+	ops::Deref,
+	sync::Arc,
+};
+
+/// The protocol name transmitted on the wire.
+#[derive(Debug, Clone)]
+pub enum ProtocolName {
+	/// The protocol name as a static string.
+	Static(&'static str),
+	/// The protocol name as a dynamically allocated string.
+	OnHeap(Arc<str>),
+}
+
+impl From<&'static str> for ProtocolName {
+	fn from(name: &'static str) -> Self {
+		Self::Static(name)
+	}
+}
+
+impl From<Arc<str>> for ProtocolName {
+	fn from(name: Arc<str>) -> Self {
+		Self::OnHeap(name)
+	}
+}
+
+impl From<String> for ProtocolName {
+	fn from(name: String) -> Self {
+		Self::OnHeap(Arc::from(name))
+	}
+}
+
+impl Deref for ProtocolName {
+	type Target = str;
+
+	fn deref(&self) -> &str {
+		match self {
+			Self::Static(name) => name,
+			Self::OnHeap(name) => &name,
+		}
+	}
+}
+
+impl Borrow<str> for ProtocolName {
+	fn borrow(&self) -> &str {
+		self
+	}
+}
+
+impl PartialEq for ProtocolName {
+	fn eq(&self, other: &Self) -> bool {
+		(self as &str) == (other as &str)
+	}
+}
+
+impl Eq for ProtocolName {}
+
+impl Hash for ProtocolName {
+	fn hash<H: Hasher>(&self, state: &mut H) {
+		(self as &str).hash(state)
+	}
+}
+
+impl fmt::Display for ProtocolName {
+	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+		f.write_str(self)
+	}
+}
+
+impl upgrade::ProtocolName for ProtocolName {
+	fn protocol_name(&self) -> &[u8] {
+		(self as &str).as_bytes()
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::ProtocolName;
+	use std::{
+		borrow::Borrow,
+		collections::hash_map::DefaultHasher,
+		hash::{Hash, Hasher},
+	};
+
+	#[test]
+	fn protocol_name_keys_are_equivalent_to_str_keys() {
+		const PROTOCOL: &'static str = "/some/protocol/1";
+		let static_protocol_name = ProtocolName::from(PROTOCOL);
+		let on_heap_protocol_name = ProtocolName::from(String::from(PROTOCOL));
+
+		assert_eq!(<ProtocolName as Borrow<str>>::borrow(&static_protocol_name), PROTOCOL);
+		assert_eq!(<ProtocolName as Borrow<str>>::borrow(&on_heap_protocol_name), PROTOCOL);
+		assert_eq!(static_protocol_name, on_heap_protocol_name);
+
+		assert_eq!(hash(static_protocol_name), hash(PROTOCOL));
+		assert_eq!(hash(on_heap_protocol_name), hash(PROTOCOL));
+	}
+
+	#[test]
+	fn different_protocol_names_do_not_compare_equal() {
+		const PROTOCOL1: &'static str = "/some/protocol/1";
+		let static_protocol_name1 = ProtocolName::from(PROTOCOL1);
+		let on_heap_protocol_name1 = ProtocolName::from(String::from(PROTOCOL1));
+
+		const PROTOCOL2: &'static str = "/some/protocol/2";
+		let static_protocol_name2 = ProtocolName::from(PROTOCOL2);
+		let on_heap_protocol_name2 = ProtocolName::from(String::from(PROTOCOL2));
+
+		assert_ne!(<ProtocolName as Borrow<str>>::borrow(&static_protocol_name1), PROTOCOL2);
+		assert_ne!(<ProtocolName as Borrow<str>>::borrow(&on_heap_protocol_name1), PROTOCOL2);
+		assert_ne!(static_protocol_name1, static_protocol_name2);
+		assert_ne!(static_protocol_name1, on_heap_protocol_name2);
+		assert_ne!(on_heap_protocol_name1, on_heap_protocol_name2);
+	}
+
+	fn hash<T: Hash>(x: T) -> u64 {
+		let mut hasher = DefaultHasher::new();
+		x.hash(&mut hasher);
+		hasher.finish()
+	}
+}

substrate/client/network/src/utils.rs

@@ -0,0 +1,88 @@
+// 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/>.
+
+//! `sc-network` utilities
+
+use futures::{stream::unfold, FutureExt, Stream, StreamExt};
+use futures_timer::Delay;
+use linked_hash_set::LinkedHashSet;
+
+use std::{hash::Hash, num::NonZeroUsize, time::Duration};
+
+/// Creates a stream that returns a new value every `duration`.
+pub fn interval(duration: Duration) -> impl Stream<Item = ()> + Unpin {
+	unfold((), move |_| Delay::new(duration).map(|_| Some(((), ())))).map(drop)
+}
+
+/// Wrapper around `LinkedHashSet` with bounded growth.
+///
+/// In the limit, for each element inserted the oldest existing element will be removed.
+#[derive(Debug, Clone)]
+pub struct LruHashSet<T: Hash + Eq> {
+	set: LinkedHashSet<T>,
+	limit: NonZeroUsize,
+}
+
+impl<T: Hash + Eq> LruHashSet<T> {
+	/// Create a new `LruHashSet` with the given (exclusive) limit.
+	pub fn new(limit: NonZeroUsize) -> Self {
+		Self { set: LinkedHashSet::new(), limit }
+	}
+
+	/// Insert element into the set.
+	///
+	/// Returns `true` if this is a new element to the set, `false` otherwise.
+	/// Maintains the limit of the set by removing the oldest entry if necessary.
+	/// Inserting the same element will update its LRU position.
+	pub fn insert(&mut self, e: T) -> bool {
+		if self.set.insert(e) {
+			if self.set.len() == usize::from(self.limit) {
+				self.set.pop_front(); // remove oldest entry
+			}
+			return true
+		}
+		false
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn maintains_limit() {
+		let three = NonZeroUsize::new(3).unwrap();
+		let mut set = LruHashSet::<u8>::new(three);
+
+		// First element.
+		assert!(set.insert(1));
+		assert_eq!(vec![&1], set.set.iter().collect::<Vec<_>>());
+
+		// Second element.
+		assert!(set.insert(2));
+		assert_eq!(vec![&1, &2], set.set.iter().collect::<Vec<_>>());
+
+		// Inserting the same element updates its LRU position.
+		assert!(!set.insert(1));
+		assert_eq!(vec![&2, &1], set.set.iter().collect::<Vec<_>>());
+
+		// We reached the limit. The next element forces the oldest one out.
+		assert!(set.insert(3));
+		assert_eq!(vec![&1, &3], set.set.iter().collect::<Vec<_>>());
+	}
+}