Prepare sc-network for ProtocolController/NotificationService (#14080)

* Prepare `sc-network` for `ProtocolController`/`NotificationService`

The upcoming notification protocol refactoring requires that protocols
are able to communicate with `sc-network` over unique and direct links.
This means that `sc-network` side of the link has to be created before
`sc-network` is initialized and that it is allowed to consume the object
as the receiver half of the link may not implement `Clone`.

Remove request-response and notification protocols from `NetworkConfiguration`
and create a new object that contains the configurations of these protocols
and which is consumable by `sc-network`. This is needed needed because, e.g.,
the receiver half of `NotificationService` is not clonable so `sc-network`
must consume it when it's initializing the protocols in `Notifications`.

Similar principe applies to `PeerStore`/`ProtocolController`: as per current
design, protocols are created before the network so `Protocol` cannot be
the one creating the `PeerStore` object. `FullNetworkConfiguration` will be
used to store the objects that `sc-network` will use to communicate with
protocols and it will also allow protocols to allocate handles so they
can directly communicate with `sc-network`.

* Fixes

* Update client/service/src/builder.rs

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

* Updates

* Doc updates + cargo-fmt

---------

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

substrate/client/network/src/config.rs

@@ -32,15 +32,16 @@ pub use crate::{
 use codec::Encode;
 use libp2p::{identity::Keypair, multiaddr, Multiaddr, PeerId};
 use prometheus_endpoint::Registry;
+use zeroize::Zeroize;
+
 pub use sc_network_common::{
 	role::{Role, Roles},
 	sync::warp::WarpSyncProvider,
 	ExHashT,
 };
 use sc_utils::mpsc::TracingUnboundedSender;
-use zeroize::Zeroize;
-
 use sp_runtime::traits::Block as BlockT;
+
 use std::{
 	error::Error,
 	fmt, fs,
@@ -564,9 +565,6 @@ pub struct NetworkConfiguration {
 	/// 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,
 
@@ -576,9 +574,6 @@ pub struct NetworkConfiguration {
 	/// 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,
 
@@ -649,10 +644,8 @@ impl NetworkConfiguration {
 			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 },
@@ -707,7 +700,7 @@ pub struct Params<Block: BlockT> {
 	pub executor: Box<dyn Fn(Pin<Box<dyn Future<Output = ()> + Send>>) + Send>,
 
 	/// Network layer configuration.
-	pub network_config: NetworkConfiguration,
+	pub network_config: FullNetworkConfiguration,
 
 	/// Legacy name of the protocol to use on the wire. Should be different for each chain.
 	pub protocol_id: ProtocolId,
@@ -727,9 +720,44 @@ pub struct Params<Block: BlockT> {
 
 	/// TX channel for direct communication with `SyncingEngine` and `Protocol`.
 	pub tx: TracingUnboundedSender<crate::event::SyncEvent<Block>>,
+}
 
-	/// Request response protocol configurations
-	pub request_response_protocol_configs: Vec<RequestResponseConfig>,
+/// Full network configuration.
+pub struct FullNetworkConfiguration {
+	/// Installed notification protocols.
+	pub(crate) notification_protocols: Vec<NonDefaultSetConfig>,
+
+	/// List of request-response protocols that the node supports.
+	pub(crate) request_response_protocols: Vec<RequestResponseConfig>,
+
+	/// Network configuration.
+	pub network_config: NetworkConfiguration,
+}
+
+impl FullNetworkConfiguration {
+	/// Create new [`FullNetworkConfiguration`].
+	pub fn new(network_config: &NetworkConfiguration) -> Self {
+		Self {
+			notification_protocols: Vec::new(),
+			request_response_protocols: Vec::new(),
+			network_config: network_config.clone(),
+		}
+	}
+
+	/// Add a notification protocol.
+	pub fn add_notification_protocol(&mut self, config: NonDefaultSetConfig) {
+		self.notification_protocols.push(config);
+	}
+
+	/// Get reference to installed notification protocols.
+	pub fn notification_protocols(&self) -> &Vec<NonDefaultSetConfig> {
+		&self.notification_protocols
+	}
+
+	/// Add a request-response protocol.
+	pub fn add_request_response_protocol(&mut self, config: RequestResponseConfig) {
+		self.request_response_protocols.push(config);
+	}
 }
 
 #[cfg(test)]

substrate/client/network/src/protocol.rs

@@ -102,6 +102,7 @@ impl<B: BlockT> Protocol<B> {
 	pub fn new(
 		roles: Roles,
 		network_config: &config::NetworkConfiguration,
+		notification_protocols: Vec<config::NonDefaultSetConfig>,
 		block_announces_protocol: config::NonDefaultSetConfig,
 		tx: TracingUnboundedSender<crate::event::SyncEvent<B>>,
 	) -> error::Result<(Self, sc_peerset::PeersetHandle, Vec<(PeerId, Multiaddr)>)> {
@@ -109,7 +110,7 @@ impl<B: BlockT> Protocol<B> {
 
 		let (peerset, peerset_handle) = {
 			let mut sets =
-				Vec::with_capacity(NUM_HARDCODED_PEERSETS + network_config.extra_sets.len());
+				Vec::with_capacity(NUM_HARDCODED_PEERSETS + notification_protocols.len());
 
 			let mut default_sets_reserved = HashSet::new();
 			for reserved in network_config.default_peers_set.reserved_nodes.iter() {
@@ -135,7 +136,7 @@ impl<B: BlockT> Protocol<B> {
 					NonReservedPeerMode::Deny,
 			});
 
-			for set_cfg in &network_config.extra_sets {
+			for set_cfg in &notification_protocols {
 				let mut reserved_nodes = HashSet::new();
 				for reserved in set_cfg.set_config.reserved_nodes.iter() {
 					reserved_nodes.insert(reserved.peer_id);
@@ -169,7 +170,7 @@ impl<B: BlockT> Protocol<B> {
 					handshake: block_announces_protocol.handshake.as_ref().unwrap().to_vec(),
 					max_notification_size: block_announces_protocol.max_notification_size,
 				})
-				.chain(network_config.extra_sets.iter().map(|s| notifications::ProtocolConfig {
+				.chain(notification_protocols.iter().map(|s| notifications::ProtocolConfig {
 					name: s.notifications_protocol.clone(),
 					fallback_names: s.fallback_names.clone(),
 					handshake: s.handshake.as_ref().map_or(roles.encode(), |h| (*h).to_vec()),
@@ -182,7 +183,7 @@ impl<B: BlockT> Protocol<B> {
 			peerset_handle: peerset_handle.clone(),
 			behaviour,
 			notification_protocols: iter::once(block_announces_protocol.notifications_protocol)
-				.chain(network_config.extra_sets.iter().map(|s| s.notifications_protocol.clone()))
+				.chain(notification_protocols.iter().map(|s| s.notifications_protocol.clone()))
 				.collect(),
 			bad_handshake_substreams: Default::default(),
 			peers: HashMap::new(),

substrate/client/network/src/service.rs

@@ -29,7 +29,7 @@
 
 use crate::{
 	behaviour::{self, Behaviour, BehaviourOut},
-	config::{MultiaddrWithPeerId, Params, TransportConfig},
+	config::{FullNetworkConfiguration, MultiaddrWithPeerId, Params, TransportConfig},
 	discovery::DiscoveryConfig,
 	error::Error,
 	event::{DhtEvent, Event},
@@ -147,25 +147,24 @@ where
 	/// Returns a `NetworkWorker` that implements `Future` and must be regularly polled in order
 	/// for the network processing to advance. From it, you can extract a `NetworkService` using
 	/// `worker.service()`. The `NetworkService` can be shared through the codebase.
-	pub fn new(mut params: Params<B>) -> Result<Self, Error> {
+	pub fn new(params: Params<B>) -> Result<Self, Error> {
+		let FullNetworkConfiguration {
+			notification_protocols,
+			request_response_protocols,
+			mut network_config,
+		} = params.network_config;
+
 		// Private and public keys configuration.
-		let local_identity = params.network_config.node_key.clone().into_keypair()?;
+		let local_identity = network_config.node_key.clone().into_keypair()?;
 		let local_public = local_identity.public();
 		let local_peer_id = local_public.to_peer_id();
 
-		params
-			.network_config
-			.request_response_protocols
-			.extend(params.request_response_protocol_configs);
-
-		params.network_config.boot_nodes = params
-			.network_config
+		network_config.boot_nodes = network_config
 			.boot_nodes
 			.into_iter()
 			.filter(|boot_node| boot_node.peer_id != local_peer_id)
 			.collect();
-		params.network_config.default_peers_set.reserved_nodes = params
-			.network_config
+		network_config.default_peers_set.reserved_nodes = network_config
 			.default_peers_set
 			.reserved_nodes
 			.into_iter()
@@ -185,36 +184,31 @@ where
 
 		// Ensure the listen addresses are consistent with the transport.
 		ensure_addresses_consistent_with_transport(
-			params.network_config.listen_addresses.iter(),
-			&params.network_config.transport,
+			network_config.listen_addresses.iter(),
+			&network_config.transport,
 		)?;
 		ensure_addresses_consistent_with_transport(
-			params.network_config.boot_nodes.iter().map(|x| &x.multiaddr),
-			&params.network_config.transport,
+			network_config.boot_nodes.iter().map(|x| &x.multiaddr),
+			&network_config.transport,
 		)?;
 		ensure_addresses_consistent_with_transport(
-			params
-				.network_config
-				.default_peers_set
-				.reserved_nodes
-				.iter()
-				.map(|x| &x.multiaddr),
-			&params.network_config.transport,
+			network_config.default_peers_set.reserved_nodes.iter().map(|x| &x.multiaddr),
+			&network_config.transport,
 		)?;
-		for extra_set in &params.network_config.extra_sets {
+		for notification_protocol in &notification_protocols {
 			ensure_addresses_consistent_with_transport(
-				extra_set.set_config.reserved_nodes.iter().map(|x| &x.multiaddr),
-				&params.network_config.transport,
+				notification_protocol.set_config.reserved_nodes.iter().map(|x| &x.multiaddr),
+				&network_config.transport,
 			)?;
 		}
 		ensure_addresses_consistent_with_transport(
-			params.network_config.public_addresses.iter(),
-			&params.network_config.transport,
+			network_config.public_addresses.iter(),
+			&network_config.transport,
 		)?;
 
 		let (to_worker, from_service) = tracing_unbounded("mpsc_network_worker", 100_000);
 
-		if let Some(path) = &params.network_config.net_config_path {
+		if let Some(path) = &network_config.net_config_path {
 			fs::create_dir_all(path)?;
 		}
 
@@ -224,9 +218,58 @@ where
 			local_peer_id.to_base58(),
 		);
 
+		let (transport, bandwidth) = {
+			let config_mem = match network_config.transport {
+				TransportConfig::MemoryOnly => true,
+				TransportConfig::Normal { .. } => false,
+			};
+
+			// The yamux buffer size limit is configured to be equal to the maximum frame size
+			// of all protocols. 10 bytes are added to each limit for the length prefix that
+			// is not included in the upper layer protocols limit but is still present in the
+			// yamux buffer. These 10 bytes correspond to the maximum size required to encode
+			// a variable-length-encoding 64bits number. In other words, we make the
+			// assumption that no notification larger than 2^64 will ever be sent.
+			let yamux_maximum_buffer_size = {
+				let requests_max = request_response_protocols
+					.iter()
+					.map(|cfg| usize::try_from(cfg.max_request_size).unwrap_or(usize::MAX));
+				let responses_max = request_response_protocols
+					.iter()
+					.map(|cfg| usize::try_from(cfg.max_response_size).unwrap_or(usize::MAX));
+				let notifs_max = notification_protocols
+					.iter()
+					.map(|cfg| usize::try_from(cfg.max_notification_size).unwrap_or(usize::MAX));
+
+				// A "default" max is added to cover all the other protocols: ping, identify,
+				// kademlia, block announces, and transactions.
+				let default_max = cmp::max(
+					1024 * 1024,
+					usize::try_from(protocol::BLOCK_ANNOUNCES_TRANSACTIONS_SUBSTREAM_SIZE)
+						.unwrap_or(usize::MAX),
+				);
+
+				iter::once(default_max)
+					.chain(requests_max)
+					.chain(responses_max)
+					.chain(notifs_max)
+					.max()
+					.expect("iterator known to always yield at least one element; qed")
+					.saturating_add(10)
+			};
+
+			transport::build_transport(
+				local_identity.clone(),
+				config_mem,
+				network_config.yamux_window_size,
+				yamux_maximum_buffer_size,
+			)
+		};
+
 		let (protocol, peerset_handle, mut known_addresses) = Protocol::new(
 			From::from(&params.role),
-			&params.network_config,
+			&network_config,
+			notification_protocols,
 			params.block_announce_config,
 			params.tx,
 		)?;
@@ -235,7 +278,7 @@ where
 		let mut boot_node_ids = HashSet::new();
 
 		// Process the bootnodes.
-		for bootnode in params.network_config.boot_nodes.iter() {
+		for bootnode in network_config.boot_nodes.iter() {
 			boot_node_ids.insert(bootnode.peer_id);
 			known_addresses.push((bootnode.peer_id, bootnode.multiaddr.clone()));
 		}
@@ -243,9 +286,8 @@ where
 		let boot_node_ids = Arc::new(boot_node_ids);
 
 		// Check for duplicate bootnodes.
-		params.network_config.boot_nodes.iter().try_for_each(|bootnode| {
-			if let Some(other) = params
-				.network_config
+		network_config.boot_nodes.iter().try_for_each(|bootnode| {
+			if let Some(other) = network_config
 				.boot_nodes
 				.iter()
 				.filter(|o| o.multiaddr == bootnode.multiaddr)
@@ -265,29 +307,25 @@ where
 
 		// Build the swarm.
 		let (mut swarm, bandwidth): (Swarm<Behaviour<B>>, _) = {
-			let user_agent = format!(
-				"{} ({})",
-				params.network_config.client_version, params.network_config.node_name
-			);
+			let user_agent =
+				format!("{} ({})", network_config.client_version, network_config.node_name);
 
 			let discovery_config = {
 				let mut config = DiscoveryConfig::new(local_public.clone());
 				config.with_permanent_addresses(known_addresses);
-				config.discovery_limit(
-					u64::from(params.network_config.default_peers_set.out_peers) + 15,
-				);
+				config.discovery_limit(u64::from(network_config.default_peers_set.out_peers) + 15);
 				config.with_kademlia(
 					params.genesis_hash,
 					params.fork_id.as_deref(),
 					&params.protocol_id,
 				);
-				config.with_dht_random_walk(params.network_config.enable_dht_random_walk);
-				config.allow_non_globals_in_dht(params.network_config.allow_non_globals_in_dht);
+				config.with_dht_random_walk(network_config.enable_dht_random_walk);
+				config.allow_non_globals_in_dht(network_config.allow_non_globals_in_dht);
 				config.use_kademlia_disjoint_query_paths(
-					params.network_config.kademlia_disjoint_query_paths,
+					network_config.kademlia_disjoint_query_paths,
 				);
 
-				match params.network_config.transport {
+				match network_config.transport {
 					TransportConfig::MemoryOnly => {
 						config.with_mdns(false);
 						config.allow_private_ip(false);
@@ -305,64 +343,13 @@ where
 				config
 			};
 
-			let (transport, bandwidth) = {
-				let config_mem = match params.network_config.transport {
-					TransportConfig::MemoryOnly => true,
-					TransportConfig::Normal { .. } => false,
-				};
-
-				// The yamux buffer size limit is configured to be equal to the maximum frame size
-				// of all protocols. 10 bytes are added to each limit for the length prefix that
-				// is not included in the upper layer protocols limit but is still present in the
-				// yamux buffer. These 10 bytes correspond to the maximum size required to encode
-				// a variable-length-encoding 64bits number. In other words, we make the
-				// assumption that no notification larger than 2^64 will ever be sent.
-				let yamux_maximum_buffer_size = {
-					let requests_max = params
-						.network_config
-						.request_response_protocols
-						.iter()
-						.map(|cfg| usize::try_from(cfg.max_request_size).unwrap_or(usize::MAX));
-					let responses_max =
-						params.network_config.request_response_protocols.iter().map(|cfg| {
-							usize::try_from(cfg.max_response_size).unwrap_or(usize::MAX)
-						});
-					let notifs_max = params.network_config.extra_sets.iter().map(|cfg| {
-						usize::try_from(cfg.max_notification_size).unwrap_or(usize::MAX)
-					});
-
-					// A "default" max is added to cover all the other protocols: ping, identify,
-					// kademlia, block announces, and transactions.
-					let default_max = cmp::max(
-						1024 * 1024,
-						usize::try_from(protocol::BLOCK_ANNOUNCES_TRANSACTIONS_SUBSTREAM_SIZE)
-							.unwrap_or(usize::MAX),
-					);
-
-					iter::once(default_max)
-						.chain(requests_max)
-						.chain(responses_max)
-						.chain(notifs_max)
-						.max()
-						.expect("iterator known to always yield at least one element; qed")
-						.saturating_add(10)
-				};
-
-				transport::build_transport(
-					local_identity.clone(),
-					config_mem,
-					params.network_config.yamux_window_size,
-					yamux_maximum_buffer_size,
-				)
-			};
-
 			let behaviour = {
 				let result = Behaviour::new(
 					protocol,
 					user_agent,
 					local_public,
 					discovery_config,
-					params.network_config.request_response_protocols,
+					request_response_protocols,
 					peerset_handle.clone(),
 				);
 
@@ -416,14 +403,14 @@ where
 		};
 
 		// Listen on multiaddresses.
-		for addr in &params.network_config.listen_addresses {
+		for addr in &network_config.listen_addresses {
 			if let Err(err) = Swarm::<Behaviour<B>>::listen_on(&mut swarm, addr.clone()) {
 				warn!(target: "sub-libp2p", "Can't listen on {} because: {:?}", addr, err)
 			}
 		}
 
 		// Add external addresses.
-		for addr in &params.network_config.public_addresses {
+		for addr in &network_config.public_addresses {
 			Swarm::<Behaviour<B>>::add_external_address(
 				&mut swarm,
 				addr.clone(),