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>

Author: 2 people

Cargo.lock

@@ -23,6 +23,7 @@ futures = { version = "0.3.21", features = ["thread-pool"]}
 sc-cli = { version = "0.10.0-dev", path = "../../../client/cli" }
 sp-core = { version = "7.0.0", path = "../../../primitives/core" }
 sc-executor = { version = "0.10.0-dev", path = "../../../client/executor" }
+sc-network = { version = "0.10.0-dev", path = "../../../client/network" }
 sc-service = { version = "0.10.0-dev", path = "../../../client/service" }
 sc-telemetry = { version = "4.0.0-dev", path = "../../../client/telemetry" }
 sc-keystore = { version = "4.0.0-dev", path = "../../../client/keystore" }

bin/node-template/node/Cargo.toml

@@ -138,7 +138,7 @@ pub fn new_partial(
 }
 
 /// Builds a new service for a full client.
-pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError> {
+pub fn new_full(config: Configuration) -> Result<TaskManager, ServiceError> {
 	let sc_service::PartialComponents {
 		client,
 		backend,
@@ -150,15 +150,16 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
 		other: (block_import, grandpa_link, mut telemetry),
 	} = new_partial(&config)?;
 
+	let mut net_config = sc_network::config::FullNetworkConfiguration::new(&config.network);
+
 	let grandpa_protocol_name = sc_consensus_grandpa::protocol_standard_name(
 		&client.block_hash(0).ok().flatten().expect("Genesis block exists; qed"),
 		&config.chain_spec,
 	);
+	net_config.add_notification_protocol(sc_consensus_grandpa::grandpa_peers_set_config(
+		grandpa_protocol_name.clone(),
+	));
 
-	config
-		.network
-		.extra_sets
-		.push(sc_consensus_grandpa::grandpa_peers_set_config(grandpa_protocol_name.clone()));
 	let warp_sync = Arc::new(sc_consensus_grandpa::warp_proof::NetworkProvider::new(
 		backend.clone(),
 		grandpa_link.shared_authority_set().clone(),
@@ -168,6 +169,7 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
 	let (network, system_rpc_tx, tx_handler_controller, network_starter, sync_service) =
 		sc_service::build_network(sc_service::BuildNetworkParams {
 			config: &config,
+			net_config,
 			client: client.clone(),
 			transaction_pool: transaction_pool.clone(),
 			spawn_handle: task_manager.spawn_handle(),

bin/node-template/node/src/service.rs

@@ -316,7 +316,7 @@ pub struct NewFullBase {
 
 /// Creates a full service from the configuration.
 pub fn new_full_base(
-	mut config: Configuration,
+	config: Configuration,
 	disable_hardware_benchmarks: bool,
 	with_startup_data: impl FnOnce(
 		&sc_consensus_babe::BabeBlockImport<Block, FullClient, FullGrandpaBlockImport>,
@@ -343,10 +343,15 @@ pub fn new_full_base(
 
 	let shared_voter_state = rpc_setup;
 	let auth_disc_publish_non_global_ips = config.network.allow_non_globals_in_dht;
+	let mut net_config = sc_network::config::FullNetworkConfiguration::new(&config.network);
+
 	let grandpa_protocol_name = grandpa::protocol_standard_name(
 		&client.block_hash(0).ok().flatten().expect("Genesis block exists; qed"),
 		&config.chain_spec,
 	);
+	net_config.add_notification_protocol(grandpa::grandpa_peers_set_config(
+		grandpa_protocol_name.clone(),
+	));
 
 	let statement_handler_proto = sc_network_statement::StatementHandlerPrototype::new(
 		client
@@ -356,12 +361,8 @@ pub fn new_full_base(
 			.expect("Genesis block exists; qed"),
 		config.chain_spec.fork_id(),
 	);
-	config.network.extra_sets.push(statement_handler_proto.set_config());
+	net_config.add_notification_protocol(statement_handler_proto.set_config());
 
-	config
-		.network
-		.extra_sets
-		.push(grandpa::grandpa_peers_set_config(grandpa_protocol_name.clone()));
 	let warp_sync = Arc::new(grandpa::warp_proof::NetworkProvider::new(
 		backend.clone(),
 		import_setup.1.shared_authority_set().clone(),
@@ -371,6 +372,7 @@ pub fn new_full_base(
 	let (network, system_rpc_tx, tx_handler_controller, network_starter, sync_service) =
 		sc_service::build_network(sc_service::BuildNetworkParams {
 			config: &config,
+			net_config,
 			client: client.clone(),
 			transaction_pool: transaction_pool.clone(),
 			spawn_handle: task_manager.spawn_handle(),

bin/node/cli/src/service.rs

@@ -221,8 +221,6 @@ impl NetworkParams {
 			default_peers_set_num_full: self.in_peers + self.out_peers,
 			listen_addresses,
 			public_addresses,
-			extra_sets: Vec::new(),
-			request_response_protocols: Vec::new(),
 			node_key,
 			node_name: node_name.to_string(),
 			client_version: client_id.to_string(),

client/cli/src/params/network_params.rs

@@ -63,7 +63,7 @@ pub(crate) mod beefy_protocol_name {
 }
 
 /// Returns the configuration value to put in
-/// [`sc_network::config::NetworkConfiguration::extra_sets`].
+/// [`sc_network::config::FullNetworkConfiguration`].
 /// For standard protocol name see [`beefy_protocol_name::gossip_protocol_name`].
 pub fn beefy_peers_set_config(
 	gossip_protocol_name: sc_network::ProtocolName,

client/consensus/beefy/src/communication/mod.rs

@@ -691,7 +691,7 @@ pub struct GrandpaParams<Block: BlockT, C, N, S, SC, VR> {
 }
 
 /// Returns the configuration value to put in
-/// [`sc_network::config::NetworkConfiguration::extra_sets`].
+/// [`sc_network::config::FullNetworkConfiguration`].
 /// For standard protocol name see [`crate::protocol_standard_name`].
 pub fn grandpa_peers_set_config(
 	protocol_name: ProtocolName,

client/consensus/grandpa/src/lib.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>>,
+}
+
+/// 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>,
 
-	/// Request response protocol configurations
-	pub request_response_protocol_configs: 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)]

client/network/src/config.rs

@@ -102,14 +102,15 @@ 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)>)> {
 		let mut known_addresses = Vec::new();
 
 		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(),