diff --git a/Cargo.lock b/Cargo.lock index 6ea79a120361e..a9a0eef551179 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -468,6 +468,7 @@ dependencies = [ "sc-finality-grandpa", "sc-keystore", "sc-network", + "sc-network-common", "sc-network-gossip", "sc-network-test", "sc-utils", @@ -7934,6 +7935,7 @@ dependencies = [ "sc-client-db", "sc-keystore", "sc-network", + "sc-network-common", "sc-service", "sc-telemetry", "sc-tracing", @@ -8547,7 +8549,9 @@ dependencies = [ "bitflags", "bytes", "futures", + "futures-timer", "libp2p", + "linked_hash_set", "parity-scale-codec", "prost-build 0.10.4", "sc-consensus", @@ -8558,6 +8562,7 @@ dependencies = [ "sp-consensus", "sp-finality-grandpa", "sp-runtime", + "substrate-prometheus-endpoint", "thiserror", ] @@ -8663,6 +8668,24 @@ dependencies = [ "substrate-test-runtime-client", ] +[[package]] +name = "sc-network-transactions" +version = "0.10.0-dev" +dependencies = [ + "array-bytes", + "futures", + "hex", + "libp2p", + "log", + "parity-scale-codec", + "pin-project", + "sc-network-common", + "sc-peerset", + "sp-consensus", + "sp-runtime", + "substrate-prometheus-endpoint", +] + [[package]] name = "sc-offchain" version = "4.0.0-dev" @@ -8851,6 +8874,7 @@ dependencies = [ "sc-network-common", "sc-network-light", "sc-network-sync", + "sc-network-transactions", "sc-offchain", "sc-rpc", "sc-rpc-server", diff --git a/bin/node-template/node/src/service.rs b/bin/node-template/node/src/service.rs index 6ec9a33749a69..96de6e17f3bfd 100644 --- a/bin/node-template/node/src/service.rs +++ b/bin/node-template/node/src/service.rs @@ -191,7 +191,7 @@ pub fn new_full(mut config: Configuration) -> Result Vec::default(), )); - let (network, system_rpc_tx, network_starter) = + let (network, system_rpc_tx, tx_handler_controller, network_starter) = sc_service::build_network(sc_service::BuildNetworkParams { config: &config, client: client.clone(), @@ -238,6 +238,7 @@ pub fn new_full(mut config: Configuration) -> Result rpc_builder: rpc_extensions_builder, backend, system_rpc_tx, + tx_handler_controller, config, telemetry: telemetry.as_mut(), })?; diff --git a/bin/node/cli/src/service.rs b/bin/node/cli/src/service.rs index a3098eac6402f..6c29f0c08ee13 100644 --- a/bin/node/cli/src/service.rs +++ b/bin/node/cli/src/service.rs @@ -354,7 +354,7 @@ pub fn new_full_base( Vec::default(), )); - let (network, system_rpc_tx, network_starter) = + let (network, system_rpc_tx, tx_handler_controller, network_starter) = sc_service::build_network(sc_service::BuildNetworkParams { config: &config, client: client.clone(), @@ -392,6 +392,7 @@ pub fn new_full_base( transaction_pool: transaction_pool.clone(), task_manager: &mut task_manager, system_rpc_tx, + tx_handler_controller, telemetry: telemetry.as_mut(), })?; diff --git a/client/beefy/Cargo.toml b/client/beefy/Cargo.toml index 47a3be859cbbb..a125d4c8d4f07 100644 --- a/client/beefy/Cargo.toml +++ b/client/beefy/Cargo.toml @@ -27,6 +27,7 @@ sc-consensus = { version = "0.10.0-dev", path = "../consensus/common" } sc-finality-grandpa = { version = "0.10.0-dev", path = "../../client/finality-grandpa" } sc-keystore = { version = "4.0.0-dev", path = "../keystore" } sc-network = { version = "0.10.0-dev", path = "../network" } +sc-network-common = { version = "0.10.0-dev", path = "../network/common" } sc-network-gossip = { version = "0.10.0-dev", path = "../network-gossip" } sc-utils = { version = "4.0.0-dev", path = "../utils" } sp-api = { version = "4.0.0-dev", path = "../../primitives/api" } diff --git a/client/beefy/src/lib.rs b/client/beefy/src/lib.rs index 41eeec43d64bd..ad527b2929585 100644 --- a/client/beefy/src/lib.rs +++ b/client/beefy/src/lib.rs @@ -83,8 +83,8 @@ pub(crate) mod beefy_protocol_name { /// For standard protocol name see [`beefy_protocol_name::standard_name`]. pub fn beefy_peers_set_config( protocol_name: ProtocolName, -) -> sc_network::config::NonDefaultSetConfig { - let mut cfg = sc_network::config::NonDefaultSetConfig::new(protocol_name, 1024 * 1024); +) -> sc_network_common::config::NonDefaultSetConfig { + let mut cfg = sc_network_common::config::NonDefaultSetConfig::new(protocol_name, 1024 * 1024); cfg.allow_non_reserved(25, 25); cfg.add_fallback_names(beefy_protocol_name::LEGACY_NAMES.iter().map(|&n| n.into()).collect()); diff --git a/client/cli/Cargo.toml b/client/cli/Cargo.toml index e5cd6167596c0..37a8fd2e0b64d 100644 --- a/client/cli/Cargo.toml +++ b/client/cli/Cargo.toml @@ -34,6 +34,7 @@ sc-client-api = { version = "4.0.0-dev", path = "../api" } sc-client-db = { version = "0.10.0-dev", default-features = false, path = "../db" } sc-keystore = { version = "4.0.0-dev", path = "../keystore" } sc-network = { version = "0.10.0-dev", path = "../network" } +sc-network-common = { version = "0.10.0-dev", path = "../network/common" } sc-service = { version = "0.10.0-dev", default-features = false, path = "../service" } sc-telemetry = { version = "4.0.0-dev", path = "../telemetry" } sc-tracing = { version = "4.0.0-dev", path = "../tracing" } diff --git a/client/cli/src/params/network_params.rs b/client/cli/src/params/network_params.rs index 74c2db92c3215..0450b5f0e2566 100644 --- a/client/cli/src/params/network_params.rs +++ b/client/cli/src/params/network_params.rs @@ -19,11 +19,10 @@ use crate::{arg_enums::SyncMode, params::node_key_params::NodeKeyParams}; use clap::Args; use sc_network::{ - config::{ - NetworkConfiguration, NodeKeyConfig, NonReservedPeerMode, SetConfig, TransportConfig, - }, + config::{NetworkConfiguration, NodeKeyConfig}, multiaddr::Protocol, }; +use sc_network_common::config::{NonReservedPeerMode, SetConfig, TransportConfig}; use sc_service::{ config::{Multiaddr, MultiaddrWithPeerId}, ChainSpec, ChainType, diff --git a/client/finality-grandpa/src/lib.rs b/client/finality-grandpa/src/lib.rs index 7e47b70bd6b98..d5c05fea78aa2 100644 --- a/client/finality-grandpa/src/lib.rs +++ b/client/finality-grandpa/src/lib.rs @@ -688,18 +688,18 @@ pub struct GrandpaParams { /// For standard protocol name see [`crate::protocol_standard_name`]. pub fn grandpa_peers_set_config( protocol_name: ProtocolName, -) -> sc_network::config::NonDefaultSetConfig { +) -> sc_network_common::config::NonDefaultSetConfig { use communication::grandpa_protocol_name; - sc_network::config::NonDefaultSetConfig { + sc_network_common::config::NonDefaultSetConfig { notifications_protocol: protocol_name, fallback_names: grandpa_protocol_name::LEGACY_NAMES.iter().map(|&n| n.into()).collect(), // Notifications reach ~256kiB in size at the time of writing on Kusama and Polkadot. max_notification_size: 1024 * 1024, - set_config: sc_network::config::SetConfig { + set_config: sc_network_common::config::SetConfig { in_peers: 0, out_peers: 0, reserved_nodes: Vec::new(), - non_reserved_mode: sc_network::config::NonReservedPeerMode::Deny, + non_reserved_mode: sc_network_common::config::NonReservedPeerMode::Deny, }, } } diff --git a/client/network/common/Cargo.toml b/client/network/common/Cargo.toml index 47d43e8b4b03f..1ee7b15538366 100644 --- a/client/network/common/Cargo.toml +++ b/client/network/common/Cargo.toml @@ -24,7 +24,10 @@ codec = { package = "parity-scale-codec", version = "3.0.0", features = [ "derive", ] } futures = "0.3.21" +futures-timer = "3.0.2" libp2p = "0.46.1" +linked_hash_set = "0.1.3" +prometheus-endpoint = { package = "substrate-prometheus-endpoint", version = "0.10.0-dev", path = "../../../utils/prometheus" } smallvec = "1.8.0" sc-consensus = { version = "0.10.0-dev", path = "../../consensus/common" } sc-peerset = { version = "4.0.0-dev", path = "../../peerset" } diff --git a/client/network/common/src/config.rs b/client/network/common/src/config.rs index 8b7e045780d7d..fb23cd0174922 100644 --- a/client/network/common/src/config.rs +++ b/client/network/common/src/config.rs @@ -18,6 +18,8 @@ //! Configuration of the networking layer. +use crate::protocol; + use libp2p::{multiaddr, Multiaddr, PeerId}; use std::{fmt, str, str::FromStr}; @@ -171,3 +173,129 @@ impl From for ParseErr { Self::MultiaddrParse(err) } } + +/// 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, + /// 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: 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, + /// 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: protocol::ProtocolName, max_notification_size: u64) -> Self { + Self { + notifications_protocol, + max_notification_size, + fallback_names: Vec::new(), + 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) { + self.fallback_names.extend(fallback_names); + } +} + +/// 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 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_ipv4: 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 { + match s { + "accept" => Some(Self::Accept), + "deny" => Some(Self::Deny), + _ => None, + } + } +} diff --git a/client/network/src/error.rs b/client/network/common/src/error.rs similarity index 96% rename from client/network/src/error.rs rename to client/network/common/src/error.rs index b4287ffbd55db..4326b1af52836 100644 --- a/client/network/src/error.rs +++ b/client/network/common/src/error.rs @@ -18,9 +18,8 @@ //! Substrate network possible errors. -use crate::config::TransportConfig; +use crate::{config::TransportConfig, protocol::ProtocolName}; use libp2p::{Multiaddr, PeerId}; -use sc_network_common::protocol::ProtocolName; use std::fmt; diff --git a/client/network/common/src/lib.rs b/client/network/common/src/lib.rs index 3a30d24900199..36e67f11e5cff 100644 --- a/client/network/common/src/lib.rs +++ b/client/network/common/src/lib.rs @@ -19,8 +19,16 @@ //! Common data structures of the networking layer. pub mod config; +pub mod error; pub mod message; pub mod protocol; pub mod request_responses; pub mod service; pub mod sync; +pub mod utils; + +/// Minimum Requirements for a Hash within Networking +pub trait ExHashT: std::hash::Hash + Eq + std::fmt::Debug + Clone + Send + Sync + 'static {} + +impl ExHashT for T where T: std::hash::Hash + Eq + std::fmt::Debug + Clone + Send + Sync + 'static +{} diff --git a/client/network/common/src/service.rs b/client/network/common/src/service.rs index 88583832e4c38..aa4967ba51700 100644 --- a/client/network/common/src/service.rs +++ b/client/network/common/src/service.rs @@ -604,35 +604,6 @@ where } } -/// Provides ability to propagate transactions over the network. -pub trait NetworkTransaction { - /// You may call this when new transactions are imported by the transaction pool. - /// - /// All transactions will be fetched from the `TransactionPool` that was passed at - /// initialization as part of the configuration and propagated to peers. - fn trigger_repropagate(&self); - - /// You must call when new transaction is imported by the transaction pool. - /// - /// This transaction will be fetched from the `TransactionPool` that was passed at - /// initialization as part of the configuration and propagated to peers. - fn propagate_transaction(&self, hash: H); -} - -impl NetworkTransaction for Arc -where - T: ?Sized, - T: NetworkTransaction, -{ - fn trigger_repropagate(&self) { - T::trigger_repropagate(self) - } - - fn propagate_transaction(&self, hash: H) { - T::propagate_transaction(self, hash) - } -} - /// Provides ability to announce blocks to the network. pub trait NetworkBlock { /// Make sure an important block is propagated to peers. diff --git a/client/network/src/utils.rs b/client/network/common/src/utils.rs similarity index 100% rename from client/network/src/utils.rs rename to client/network/common/src/utils.rs diff --git a/client/network/src/config.rs b/client/network/src/config.rs index 202a628884d79..b2adfa81d065b 100644 --- a/client/network/src/config.rs +++ b/client/network/src/config.rs @@ -27,24 +27,24 @@ pub use sc_network_common::{ IncomingRequest, OutgoingResponse, ProtocolConfig as RequestResponseConfig, }, sync::warp::WarpSyncProvider, + ExHashT, }; pub use libp2p::{build_multiaddr, core::PublicKey, identity}; -use crate::ExHashT; - use core::{fmt, iter}; -use futures::future; use libp2p::{ identity::{ed25519, Keypair}, multiaddr, Multiaddr, }; use prometheus_endpoint::Registry; use sc_consensus::ImportQueue; -use sc_network_common::{config::MultiaddrWithPeerId, protocol::ProtocolName, sync::ChainSync}; +use sc_network_common::{ + config::{MultiaddrWithPeerId, NonDefaultSetConfig, SetConfig, TransportConfig}, + sync::ChainSync, +}; use sp_runtime::traits::Block as BlockT; use std::{ - collections::HashMap, error::Error, fs, future::Future, @@ -52,16 +52,14 @@ use std::{ net::Ipv4Addr, path::{Path, PathBuf}, pin::Pin, - str, sync::Arc, }; use zeroize::Zeroize; /// Network initialization parameters. -pub struct Params +pub struct Params where B: BlockT + 'static, - H: ExHashT, { /// Assigned role for our node (full, light, ...). pub role: Role, @@ -70,21 +68,12 @@ where /// default. pub executor: Option + Send>>) + Send>>, - /// How to spawn the background task dedicated to the transactions handler. - pub transactions_handler_executor: Box + Send>>) + Send>, - /// Network layer configuration. pub network_config: NetworkConfiguration, /// Client that contains the blockchain. pub chain: Arc, - /// Pool of transactions. - /// - /// The network worker will fetch transactions from this object in order to propagate them on - /// the network. - pub transaction_pool: Arc>, - /// Legacy name of the protocol to use on the wire. Should be different for each chain. pub protocol_id: ProtocolId, @@ -166,66 +155,6 @@ impl fmt::Display for Role { } } -/// Result of the transaction import. -#[derive(Clone, Copy, Debug)] -pub enum TransactionImport { - /// Transaction is good but already known by the transaction pool. - KnownGood, - /// Transaction is good and not yet known. - NewGood, - /// Transaction is invalid. - Bad, - /// Transaction import was not performed. - None, -} - -/// Future resolving to transaction import result. -pub type TransactionImportFuture = Pin + Send>>; - -/// Transaction pool interface -pub trait TransactionPool: Send + Sync { - /// Get transactions from the pool that are ready to be propagated. - fn transactions(&self) -> Vec<(H, B::Extrinsic)>; - /// Get hash of transaction. - fn hash_of(&self, transaction: &B::Extrinsic) -> H; - /// Import a transaction into the pool. - /// - /// This will return future. - fn import(&self, transaction: B::Extrinsic) -> TransactionImportFuture; - /// Notify the pool about transactions broadcast. - fn on_broadcasted(&self, propagations: HashMap>); - /// Get transaction by hash. - fn transaction(&self, hash: &H) -> Option; -} - -/// Dummy implementation of the [`TransactionPool`] trait for a transaction pool that is always -/// empty and discards all incoming transactions. -/// -/// Requires the "hash" type to implement the `Default` trait. -/// -/// Useful for testing purposes. -pub struct EmptyTransactionPool; - -impl TransactionPool for EmptyTransactionPool { - fn transactions(&self) -> Vec<(H, B::Extrinsic)> { - Vec::new() - } - - fn hash_of(&self, _transaction: &B::Extrinsic) -> H { - Default::default() - } - - fn import(&self, _transaction: B::Extrinsic) -> TransactionImportFuture { - Box::pin(future::ready(TransactionImport::KnownGood)) - } - - fn on_broadcasted(&self, _: HashMap>) {} - - fn transaction(&self, _h: &H) -> Option { - None - } -} - /// Sync operation mode. #[derive(Copy, Clone, Debug, Eq, PartialEq)] pub enum SyncMode { @@ -394,132 +323,6 @@ impl NetworkConfiguration { } } -/// 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, - /// 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 - /// [`crate::Event::NotificationStreamOpened::negotiated_fallback`]. - pub fallback_names: Vec, - /// 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(), - 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) { - self.fallback_names.extend(fallback_names); - } -} - -/// 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 addresses (as defined in - /// [RFC1918](https://tools.ietf.org/html/rfc1918)). Irrelevant for addresses that have - /// been passed in [`NetworkConfiguration::boot_nodes`]. - allow_private_ipv4: 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 { - match s { - "accept" => Some(Self::Accept), - "deny" => Some(Self::Deny), - _ => None, - } - } -} - /// 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. diff --git a/client/network/src/discovery.rs b/client/network/src/discovery.rs index ab93662968dc2..8422e34485125 100644 --- a/client/network/src/discovery.rs +++ b/client/network/src/discovery.rs @@ -46,7 +46,6 @@ //! 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::utils::LruHashSet; use futures::prelude::*; use futures_timer::Delay; use ip_network::IpNetwork; @@ -72,7 +71,7 @@ use libp2p::{ }, }; use log::{debug, error, info, trace, warn}; -use sc_network_common::config::ProtocolId; +use sc_network_common::{config::ProtocolId, utils::LruHashSet}; use sp_core::hexdisplay::HexDisplay; use std::{ cmp, diff --git a/client/network/src/lib.rs b/client/network/src/lib.rs index 320104d0f9554..d17f47328b804 100644 --- a/client/network/src/lib.rs +++ b/client/network/src/lib.rs @@ -251,12 +251,9 @@ mod protocol; mod request_responses; mod service; mod transport; -mod utils; pub mod config; -pub mod error; pub mod network_state; -pub mod transactions; #[doc(inline)] pub use libp2p::{multiaddr, Multiaddr, PeerId}; @@ -269,8 +266,8 @@ pub use sc_network_common::{ request_responses::{IfDisconnected, RequestFailure}, service::{ KademliaKey, NetworkBlock, NetworkDHTProvider, NetworkRequest, NetworkSigner, - NetworkStateInfo, NetworkStatus, NetworkStatusProvider, NetworkSyncForkRequest, - NetworkTransaction, Signature, SigningError, + NetworkStateInfo, NetworkStatus, NetworkStatusProvider, NetworkSyncForkRequest, Signature, + SigningError, }, sync::{ warp::{WarpSyncPhase, WarpSyncProgress}, @@ -295,9 +292,3 @@ const MAX_CONNECTIONS_PER_PEER: usize = 2; /// The maximum number of concurrent established connections that were incoming. const MAX_CONNECTIONS_ESTABLISHED_INCOMING: u32 = 10_000; - -/// Minimum Requirements for a Hash within Networking -pub trait ExHashT: std::hash::Hash + Eq + std::fmt::Debug + Clone + Send + Sync + 'static {} - -impl ExHashT for T where T: std::hash::Hash + Eq + std::fmt::Debug + Clone + Send + Sync + 'static -{} diff --git a/client/network/src/peer_info.rs b/client/network/src/peer_info.rs index d668cb25ea455..c62c2ea1c5d98 100644 --- a/client/network/src/peer_info.rs +++ b/client/network/src/peer_info.rs @@ -16,7 +16,6 @@ // You should have received a copy of the GNU General Public License // along with this program. If not, see . -use crate::utils::interval; use fnv::FnvHashMap; use futures::prelude::*; use libp2p::{ @@ -33,6 +32,7 @@ use libp2p::{ Multiaddr, }; use log::{debug, error, trace}; +use sc_network_common::utils::interval; use smallvec::SmallVec; use std::{ collections::hash_map::Entry, diff --git a/client/network/src/protocol.rs b/client/network/src/protocol.rs index c7a3cf4b2160f..fbf651de9d49a 100644 --- a/client/network/src/protocol.rs +++ b/client/network/src/protocol.rs @@ -16,10 +16,7 @@ // You should have received a copy of the GNU General Public License // along with this program. If not, see . -use crate::{ - config, error, - utils::{interval, LruHashSet}, -}; +use crate::config; use bytes::Bytes; use codec::{Decode, DecodeAll, Encode}; @@ -45,7 +42,8 @@ use sc_consensus::import_queue::{ BlockImportError, BlockImportStatus, IncomingBlock, RuntimeOrigin, }; use sc_network_common::{ - config::ProtocolId, + config::{NonReservedPeerMode, ProtocolId}, + error, protocol::ProtocolName, request_responses::RequestFailure, sync::{ @@ -57,6 +55,7 @@ use sc_network_common::{ OpaqueBlockResponse, OpaqueStateRequest, OpaqueStateResponse, PollBlockAnnounceValidation, SyncStatus, }, + utils::{interval, LruHashSet}, }; use sp_arithmetic::traits::SaturatedConversion; use sp_consensus::BlockOrigin; @@ -341,7 +340,7 @@ where bootnodes, reserved_nodes: default_sets_reserved.clone(), reserved_only: network_config.default_peers_set.non_reserved_mode == - config::NonReservedPeerMode::Deny, + NonReservedPeerMode::Deny, }); for set_cfg in &network_config.extra_sets { @@ -352,7 +351,7 @@ where } let reserved_only = - set_cfg.set_config.non_reserved_mode == config::NonReservedPeerMode::Deny; + set_cfg.set_config.non_reserved_mode == NonReservedPeerMode::Deny; sets.push(sc_peerset::SetConfig { in_peers: set_cfg.set_config.in_peers, diff --git a/client/network/src/protocol/message.rs b/client/network/src/protocol/message.rs index 50c4a264a5f95..3e1281753b82c 100644 --- a/client/network/src/protocol/message.rs +++ b/client/network/src/protocol/message.rs @@ -36,9 +36,6 @@ pub type Message = generic::Message< ::Extrinsic, >; -/// A set of transactions. -pub type Transactions = Vec; - /// Remote call response. #[derive(Debug, PartialEq, Eq, Clone, Encode, Decode)] pub struct RemoteCallResponse { @@ -59,7 +56,7 @@ pub struct RemoteReadResponse { /// Generic types. pub mod generic { - use super::{RemoteCallResponse, RemoteReadResponse, Transactions}; + use super::{RemoteCallResponse, RemoteReadResponse}; use bitflags::bitflags; use codec::{Decode, Encode, Input, Output}; use sc_client_api::StorageProof; @@ -146,9 +143,10 @@ pub mod generic { BlockResponse(BlockResponse), /// Block announce. BlockAnnounce(BlockAnnounce
), - /// Transactions. - Transactions(Transactions), /// Consensus protocol message. + // NOTE: index is incremented by 1 due to transaction-related + // message that was removed + #[codec(index = 6)] Consensus(ConsensusMessage), /// Remote method call request. RemoteCallRequest(RemoteCallRequest), diff --git a/client/network/src/service.rs b/client/network/src/service.rs index dceb57d9e695c..180482e75ece2 100644 --- a/client/network/src/service.rs +++ b/client/network/src/service.rs @@ -29,9 +29,8 @@ use crate::{ behaviour::{self, Behaviour, BehaviourOut}, - config::{Params, TransportConfig}, + config::Params, discovery::DiscoveryConfig, - error::Error, network_state::{ NetworkState, NotConnectedPeer as NetworkStateNotConnectedPeer, Peer as NetworkStatePeer, }, @@ -39,7 +38,7 @@ use crate::{ self, message::generic::Roles, NotificationsSink, NotifsHandlerError, PeerInfo, Protocol, Ready, }, - transactions, transport, ExHashT, ReputationChange, + transport, ReputationChange, }; use codec::Encode as _; @@ -60,7 +59,8 @@ use metrics::{Histogram, HistogramVec, MetricSources, Metrics}; use parking_lot::Mutex; use sc_consensus::{BlockImportError, BlockImportStatus, ImportQueue, Link}; use sc_network_common::{ - config::MultiaddrWithPeerId, + config::{MultiaddrWithPeerId, TransportConfig}, + error::Error, protocol::{ event::{DhtEvent, Event}, ProtocolName, @@ -73,6 +73,7 @@ use sc_network_common::{ NotificationSenderReady as NotificationSenderReadyT, Signature, SigningError, }, sync::{SyncState, SyncStatus}, + ExHashT, }; use sc_peerset::PeersetHandle; use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender}; @@ -101,7 +102,7 @@ mod out_events; mod tests; pub use libp2p::identity::{error::DecodingError, Keypair, PublicKey}; -use sc_network_common::service::{NetworkBlock, NetworkRequest, NetworkTransaction}; +use sc_network_common::service::{NetworkBlock, NetworkRequest}; /// Substrate network service. Handles network IO and manages connectivity. pub struct NetworkService { @@ -121,7 +122,7 @@ pub struct NetworkService { /// nodes it should be connected to or not. peerset: PeersetHandle, /// Channel that sends messages to the actual worker. - to_worker: TracingUnboundedSender>, + to_worker: TracingUnboundedSender>, /// For each peer and protocol combination, an object that allows sending notifications to /// that peer. Updated by the [`NetworkWorker`]. peers_notifications_sinks: Arc>>, @@ -144,7 +145,7 @@ 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) -> Result { + pub fn new(mut params: Params) -> Result { // Private and public keys configuration. let local_identity = params.network_config.node_key.clone().into_keypair()?; let local_public = local_identity.public(); @@ -215,21 +216,6 @@ where fs::create_dir_all(path)?; } - let transactions_handler_proto = transactions::TransactionsHandlerPrototype::new( - params.protocol_id.clone(), - params - .chain - .hash(0u32.into()) - .ok() - .flatten() - .expect("Genesis block exists; qed"), - params.fork_id.clone(), - ); - params - .network_config - .extra_sets - .insert(0, transactions_handler_proto.set_config()); - info!( target: "sub-libp2p", "🏷 Local node identity is: {}", @@ -244,11 +230,8 @@ where params.protocol_id.clone(), ¶ms.fork_id, ¶ms.network_config, - iter::once(Vec::new()) - .chain( - (0..params.network_config.extra_sets.len() - 1) - .map(|_| default_notif_handshake_message.clone()), - ) + (0..params.network_config.extra_sets.len()) + .map(|_| default_notif_handshake_message.clone()) .collect(), params.metrics_registry.as_ref(), params.chain_sync, @@ -465,13 +448,6 @@ where _marker: PhantomData, }); - let (tx_handler, tx_handler_controller) = transactions_handler_proto.build( - service.clone(), - params.transaction_pool, - params.metrics_registry.as_ref(), - )?; - (params.transactions_handler_executor)(tx_handler.run().boxed()); - Ok(NetworkWorker { external_addresses, num_connected, @@ -482,9 +458,9 @@ where from_service, event_streams: out_events::OutChannels::new(params.metrics_registry.as_ref())?, peers_notifications_sinks, - tx_handler_controller, metrics, boot_node_ids, + _marker: Default::default(), }) } @@ -1149,20 +1125,6 @@ where } } -impl NetworkTransaction for NetworkService -where - B: BlockT + 'static, - H: ExHashT, -{ - fn trigger_repropagate(&self) { - let _ = self.to_worker.unbounded_send(ServiceToWorkerMsg::PropagateTransactions); - } - - fn propagate_transaction(&self, hash: H) { - let _ = self.to_worker.unbounded_send(ServiceToWorkerMsg::PropagateTransaction(hash)); - } -} - impl NetworkBlock> for NetworkService where B: BlockT + 'static, @@ -1249,9 +1211,7 @@ impl<'a> NotificationSenderReadyT for NotificationSenderReady<'a> { /// Messages sent from the `NetworkService` to the `NetworkWorker`. /// /// Each entry corresponds to a method of `NetworkService`. -enum ServiceToWorkerMsg { - PropagateTransaction(H), - PropagateTransactions, +enum ServiceToWorkerMsg { RequestJustification(B::Hash, NumberFor), ClearJustificationRequests, AnnounceBlock(B::Hash, Option>), @@ -1309,7 +1269,7 @@ where /// The import queue that was passed at initialization. import_queue: Box>, /// Messages from the [`NetworkService`] that must be processed. - from_service: TracingUnboundedReceiver>, + from_service: TracingUnboundedReceiver>, /// Senders for events that happen on the network. event_streams: out_events::OutChannels, /// Prometheus network metrics. @@ -1319,8 +1279,9 @@ where /// For each peer and protocol combination, an object that allows sending notifications to /// that peer. Shared with the [`NetworkService`]. peers_notifications_sinks: Arc>>, - /// Controller for the handler of incoming and outgoing transactions. - tx_handler_controller: transactions::TransactionsHandlerController, + /// Marker to pin the `H` generic. Serves no purpose except to not break backwards + /// compatibility. + _marker: PhantomData, } impl Future for NetworkWorker @@ -1376,10 +1337,6 @@ where .behaviour_mut() .user_protocol_mut() .clear_justification_requests(), - ServiceToWorkerMsg::PropagateTransaction(hash) => - this.tx_handler_controller.propagate_transaction(hash), - ServiceToWorkerMsg::PropagateTransactions => - this.tx_handler_controller.propagate_transactions(), ServiceToWorkerMsg::GetValue(key) => this.network_service.behaviour_mut().get_value(key), ServiceToWorkerMsg::PutValue(key, value) => @@ -1922,8 +1879,6 @@ where SyncState::Downloading => true, }; - this.tx_handler_controller.set_gossip_enabled(!is_major_syncing); - this.is_major_syncing.store(is_major_syncing, Ordering::Relaxed); if let Some(metrics) = this.metrics.as_ref() { diff --git a/client/network/src/service/tests.rs b/client/network/src/service/tests.rs index a9505c5341c3d..c8f137f79c6dc 100644 --- a/client/network/src/service/tests.rs +++ b/client/network/src/service/tests.rs @@ -21,7 +21,7 @@ use crate::{config, NetworkService, NetworkWorker}; use futures::prelude::*; use libp2p::PeerId; use sc_network_common::{ - config::{MultiaddrWithPeerId, ProtocolId}, + config::{MultiaddrWithPeerId, NonDefaultSetConfig, ProtocolId, SetConfig, TransportConfig}, protocol::event::Event, service::{NetworkEventStream, NetworkNotification, NetworkPeers, NetworkStateInfo}, }; @@ -135,12 +135,8 @@ fn build_test_full_node( let worker = NetworkWorker::new(config::Params { role: config::Role::Full, executor: None, - transactions_handler_executor: Box::new(|task| { - async_std::task::spawn(task); - }), network_config, chain: client.clone(), - transaction_pool: Arc::new(config::EmptyTransactionPool), protocol_id, fork_id, import_queue, @@ -178,23 +174,23 @@ fn build_nodes_one_proto() -> ( let listen_addr = config::build_multiaddr![Memory(rand::random::())]; let (node1, events_stream1) = build_test_full_node(config::NetworkConfiguration { - extra_sets: vec![config::NonDefaultSetConfig { + extra_sets: vec![NonDefaultSetConfig { notifications_protocol: PROTOCOL_NAME.into(), fallback_names: Vec::new(), max_notification_size: 1024 * 1024, set_config: Default::default(), }], listen_addresses: vec![listen_addr.clone()], - transport: config::TransportConfig::MemoryOnly, + transport: TransportConfig::MemoryOnly, ..config::NetworkConfiguration::new_local() }); let (node2, events_stream2) = build_test_full_node(config::NetworkConfiguration { - extra_sets: vec![config::NonDefaultSetConfig { + extra_sets: vec![NonDefaultSetConfig { notifications_protocol: PROTOCOL_NAME.into(), fallback_names: Vec::new(), max_notification_size: 1024 * 1024, - set_config: config::SetConfig { + set_config: SetConfig { reserved_nodes: vec![MultiaddrWithPeerId { multiaddr: listen_addr, peer_id: node1.local_peer_id(), @@ -203,7 +199,7 @@ fn build_nodes_one_proto() -> ( }, }], listen_addresses: vec![], - transport: config::TransportConfig::MemoryOnly, + transport: TransportConfig::MemoryOnly, ..config::NetworkConfiguration::new_local() }); @@ -368,13 +364,13 @@ fn lots_of_incoming_peers_works() { let (main_node, _) = build_test_full_node(config::NetworkConfiguration { listen_addresses: vec![listen_addr.clone()], - extra_sets: vec![config::NonDefaultSetConfig { + extra_sets: vec![NonDefaultSetConfig { notifications_protocol: PROTOCOL_NAME.into(), fallback_names: Vec::new(), max_notification_size: 1024 * 1024, - set_config: config::SetConfig { in_peers: u32::MAX, ..Default::default() }, + set_config: SetConfig { in_peers: u32::MAX, ..Default::default() }, }], - transport: config::TransportConfig::MemoryOnly, + transport: TransportConfig::MemoryOnly, ..config::NetworkConfiguration::new_local() }); @@ -387,11 +383,11 @@ fn lots_of_incoming_peers_works() { for _ in 0..32 { let (_dialing_node, event_stream) = build_test_full_node(config::NetworkConfiguration { listen_addresses: vec![], - extra_sets: vec![config::NonDefaultSetConfig { + extra_sets: vec![NonDefaultSetConfig { notifications_protocol: PROTOCOL_NAME.into(), fallback_names: Vec::new(), max_notification_size: 1024 * 1024, - set_config: config::SetConfig { + set_config: SetConfig { reserved_nodes: vec![MultiaddrWithPeerId { multiaddr: listen_addr.clone(), peer_id: main_node_peer_id, @@ -399,7 +395,7 @@ fn lots_of_incoming_peers_works() { ..Default::default() }, }], - transport: config::TransportConfig::MemoryOnly, + transport: TransportConfig::MemoryOnly, ..config::NetworkConfiguration::new_local() }); @@ -504,23 +500,23 @@ fn fallback_name_working() { let listen_addr = config::build_multiaddr![Memory(rand::random::())]; let (node1, mut events_stream1) = build_test_full_node(config::NetworkConfiguration { - extra_sets: vec![config::NonDefaultSetConfig { + extra_sets: vec![NonDefaultSetConfig { notifications_protocol: NEW_PROTOCOL_NAME.into(), fallback_names: vec![PROTOCOL_NAME.into()], max_notification_size: 1024 * 1024, set_config: Default::default(), }], listen_addresses: vec![listen_addr.clone()], - transport: config::TransportConfig::MemoryOnly, + transport: TransportConfig::MemoryOnly, ..config::NetworkConfiguration::new_local() }); let (_, mut events_stream2) = build_test_full_node(config::NetworkConfiguration { - extra_sets: vec![config::NonDefaultSetConfig { + extra_sets: vec![NonDefaultSetConfig { notifications_protocol: PROTOCOL_NAME.into(), fallback_names: Vec::new(), max_notification_size: 1024 * 1024, - set_config: config::SetConfig { + set_config: SetConfig { reserved_nodes: vec![MultiaddrWithPeerId { multiaddr: listen_addr, peer_id: node1.local_peer_id(), @@ -529,7 +525,7 @@ fn fallback_name_working() { }, }], listen_addresses: vec![], - transport: config::TransportConfig::MemoryOnly, + transport: TransportConfig::MemoryOnly, ..config::NetworkConfiguration::new_local() }); @@ -572,7 +568,7 @@ fn ensure_listen_addresses_consistent_with_transport_memory() { let _ = build_test_full_node(config::NetworkConfiguration { listen_addresses: vec![listen_addr.clone()], - transport: config::TransportConfig::MemoryOnly, + transport: TransportConfig::MemoryOnly, ..config::NetworkConfiguration::new("test-node", "test-client", Default::default(), None) }); } @@ -599,7 +595,7 @@ fn ensure_boot_node_addresses_consistent_with_transport_memory() { let _ = build_test_full_node(config::NetworkConfiguration { listen_addresses: vec![listen_addr.clone()], - transport: config::TransportConfig::MemoryOnly, + transport: TransportConfig::MemoryOnly, boot_nodes: vec![boot_node], ..config::NetworkConfiguration::new("test-node", "test-client", Default::default(), None) }); @@ -632,11 +628,8 @@ fn ensure_reserved_node_addresses_consistent_with_transport_memory() { let _ = build_test_full_node(config::NetworkConfiguration { listen_addresses: vec![listen_addr.clone()], - transport: config::TransportConfig::MemoryOnly, - default_peers_set: config::SetConfig { - reserved_nodes: vec![reserved_node], - ..Default::default() - }, + transport: TransportConfig::MemoryOnly, + default_peers_set: SetConfig { reserved_nodes: vec![reserved_node], ..Default::default() }, ..config::NetworkConfiguration::new("test-node", "test-client", Default::default(), None) }); } @@ -652,10 +645,7 @@ fn ensure_reserved_node_addresses_consistent_with_transport_not_memory() { let _ = build_test_full_node(config::NetworkConfiguration { listen_addresses: vec![listen_addr.clone()], - default_peers_set: config::SetConfig { - reserved_nodes: vec![reserved_node], - ..Default::default() - }, + default_peers_set: SetConfig { reserved_nodes: vec![reserved_node], ..Default::default() }, ..config::NetworkConfiguration::new("test-node", "test-client", Default::default(), None) }); } @@ -668,7 +658,7 @@ fn ensure_public_addresses_consistent_with_transport_memory() { let _ = build_test_full_node(config::NetworkConfiguration { listen_addresses: vec![listen_addr.clone()], - transport: config::TransportConfig::MemoryOnly, + transport: TransportConfig::MemoryOnly, public_addresses: vec![public_address], ..config::NetworkConfiguration::new("test-node", "test-client", Default::default(), None) }); diff --git a/client/network/test/src/lib.rs b/client/network/test/src/lib.rs index e78b91a4e04ee..2f6b788e368b3 100644 --- a/client/network/test/src/lib.rs +++ b/client/network/test/src/lib.rs @@ -47,16 +47,14 @@ use sc_consensus::{ ForkChoiceStrategy, ImportResult, JustificationImport, JustificationSyncLink, LongestChain, Verifier, }; -pub use sc_network::config::EmptyTransactionPool; use sc_network::{ - config::{ - NetworkConfiguration, NonDefaultSetConfig, NonReservedPeerMode, Role, SyncMode, - TransportConfig, - }, + config::{NetworkConfiguration, Role, SyncMode}, Multiaddr, NetworkService, NetworkWorker, }; use sc_network_common::{ - config::{MultiaddrWithPeerId, ProtocolId}, + config::{ + MultiaddrWithPeerId, NonDefaultSetConfig, NonReservedPeerMode, ProtocolId, TransportConfig, + }, protocol::ProtocolName, service::{NetworkBlock, NetworkStateInfo, NetworkSyncForkRequest}, sync::warp::{AuthorityList, EncodedProof, SetId, VerificationResult, WarpSyncProvider}, @@ -879,12 +877,8 @@ where let network = NetworkWorker::new(sc_network::config::Params { role: if config.is_authority { Role::Authority } else { Role::Full }, executor: None, - transactions_handler_executor: Box::new(|task| { - async_std::task::spawn(task); - }), network_config, chain: client.clone(), - transaction_pool: Arc::new(EmptyTransactionPool), protocol_id, fork_id, import_queue, diff --git a/client/network/transactions/Cargo.toml b/client/network/transactions/Cargo.toml new file mode 100644 index 0000000000000..5578bb2c7191e --- /dev/null +++ b/client/network/transactions/Cargo.toml @@ -0,0 +1,28 @@ +[package] +description = "Substrate transaction protocol" +name = "sc-network-transactions" +version = "0.10.0-dev" +license = "GPL-3.0-or-later WITH Classpath-exception-2.0" +authors = ["Parity Technologies "] +edition = "2021" +homepage = "https://substrate.io" +repository = "https://github.com/paritytech/substrate/" +documentation = "https://docs.rs/sc-network-transactions" +readme = "README.md" + +[package.metadata.docs.rs] +targets = ["x86_64-unknown-linux-gnu"] + +[dependencies] +array-bytes = "4.1" +codec = { package = "parity-scale-codec", version = "3.0.0", features = ["derive"] } +futures = "0.3.21" +hex = "0.4.0" +libp2p = "0.46.1" +log = "0.4.17" +pin-project = "1.0.10" +prometheus-endpoint = { package = "substrate-prometheus-endpoint", version = "0.10.0-dev", path = "../../../utils/prometheus" } +sc-network-common = { version = "0.10.0-dev", path = "../common" } +sc-peerset = { version = "4.0.0-dev", path = "../../peerset" } +sp-runtime = { version = "6.0.0", path = "../../../primitives/runtime" } +sp-consensus = { version = "0.10.0-dev", path = "../../../primitives/consensus/common" } diff --git a/client/network/transactions/src/config.rs b/client/network/transactions/src/config.rs new file mode 100644 index 0000000000000..abb8cccd301ac --- /dev/null +++ b/client/network/transactions/src/config.rs @@ -0,0 +1,98 @@ +// This file is part of Substrate. + +// Copyright (C) 2017-2022 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 . + +//! Configuration of the transaction protocol + +use futures::prelude::*; +use sc_network_common::ExHashT; +use sp_runtime::traits::Block as BlockT; +use std::{collections::HashMap, future::Future, pin::Pin, time}; + +/// Interval at which we propagate transactions; +pub(crate) const PROPAGATE_TIMEOUT: time::Duration = time::Duration::from_millis(2900); + +/// Maximum number of known transaction hashes to keep for a peer. +/// +/// This should be approx. 2 blocks full of transactions for the network to function properly. +pub(crate) const MAX_KNOWN_TRANSACTIONS: usize = 10240; // ~300kb per peer + overhead. + +/// Maximum allowed size for a transactions notification. +pub(crate) const MAX_TRANSACTIONS_SIZE: u64 = 16 * 1024 * 1024; + +/// Maximum number of transaction validation request we keep at any moment. +pub(crate) const MAX_PENDING_TRANSACTIONS: usize = 8192; + +/// Result of the transaction import. +#[derive(Clone, Copy, Debug)] +pub enum TransactionImport { + /// Transaction is good but already known by the transaction pool. + KnownGood, + /// Transaction is good and not yet known. + NewGood, + /// Transaction is invalid. + Bad, + /// Transaction import was not performed. + None, +} + +/// Future resolving to transaction import result. +pub type TransactionImportFuture = Pin + Send>>; + +/// Transaction pool interface +pub trait TransactionPool: Send + Sync { + /// Get transactions from the pool that are ready to be propagated. + fn transactions(&self) -> Vec<(H, B::Extrinsic)>; + /// Get hash of transaction. + fn hash_of(&self, transaction: &B::Extrinsic) -> H; + /// Import a transaction into the pool. + /// + /// This will return future. + fn import(&self, transaction: B::Extrinsic) -> TransactionImportFuture; + /// Notify the pool about transactions broadcast. + fn on_broadcasted(&self, propagations: HashMap>); + /// Get transaction by hash. + fn transaction(&self, hash: &H) -> Option; +} + +/// Dummy implementation of the [`TransactionPool`] trait for a transaction pool that is always +/// empty and discards all incoming transactions. +/// +/// Requires the "hash" type to implement the `Default` trait. +/// +/// Useful for testing purposes. +pub struct EmptyTransactionPool; + +impl TransactionPool for EmptyTransactionPool { + fn transactions(&self) -> Vec<(H, B::Extrinsic)> { + Vec::new() + } + + fn hash_of(&self, _transaction: &B::Extrinsic) -> H { + Default::default() + } + + fn import(&self, _transaction: B::Extrinsic) -> TransactionImportFuture { + Box::pin(future::ready(TransactionImport::KnownGood)) + } + + fn on_broadcasted(&self, _: HashMap>) {} + + fn transaction(&self, _h: &H) -> Option { + None + } +} diff --git a/client/network/src/transactions.rs b/client/network/transactions/src/lib.rs similarity index 84% rename from client/network/src/transactions.rs rename to client/network/transactions/src/lib.rs index da4547aefeab3..b75bd411b39c4 100644 --- a/client/network/src/transactions.rs +++ b/client/network/transactions/src/lib.rs @@ -26,27 +26,22 @@ //! - Use [`TransactionsHandlerPrototype::build`] then [`TransactionsHandler::run`] to obtain a //! `Future` that processes transactions. -use crate::{ - config::{self, TransactionImport, TransactionImportFuture, TransactionPool}, - error, - protocol::message, - service::NetworkService, - utils::{interval, LruHashSet}, - ExHashT, -}; - +use crate::config::*; use codec::{Decode, Encode}; use futures::{channel::mpsc, prelude::*, stream::FuturesUnordered}; use libp2p::{multiaddr, PeerId}; use log::{debug, trace, warn}; use prometheus_endpoint::{register, Counter, PrometheusError, Registry, U64}; use sc_network_common::{ - config::ProtocolId, + config::{NonDefaultSetConfig, NonReservedPeerMode, ProtocolId, SetConfig}, + error, protocol::{ event::{Event, ObservedRole}, ProtocolName, }, service::{NetworkEventStream, NetworkNotification, NetworkPeers}, + utils::{interval, LruHashSet}, + ExHashT, }; use sp_runtime::traits::Block as BlockT; use std::{ @@ -54,27 +49,14 @@ use std::{ iter, num::NonZeroUsize, pin::Pin, - sync::{ - atomic::{AtomicBool, Ordering}, - Arc, - }, + sync::Arc, task::Poll, - time, }; -/// Interval at which we propagate transactions; -const PROPAGATE_TIMEOUT: time::Duration = time::Duration::from_millis(2900); - -/// Maximum number of known transaction hashes to keep for a peer. -/// -/// This should be approx. 2 blocks full of transactions for the network to function properly. -const MAX_KNOWN_TRANSACTIONS: usize = 10240; // ~300kb per peer + overhead. +pub mod config; -/// Maximum allowed size for a transactions notification. -const MAX_TRANSACTIONS_SIZE: u64 = 16 * 1024 * 1024; - -/// Maximum number of transaction validation request we keep at any moment. -const MAX_PENDING_TRANSACTIONS: usize = 8192; +/// A set of transactions. +pub type Transactions = Vec; mod rep { use sc_peerset::ReputationChange as Rep; @@ -141,7 +123,7 @@ impl TransactionsHandlerPrototype { pub fn new>( protocol_id: ProtocolId, genesis_hash: Hash, - fork_id: Option, + fork_id: Option<&str>, ) -> Self { let genesis_hash = genesis_hash.as_ref(); let protocol_name = if let Some(fork_id) = fork_id { @@ -158,16 +140,16 @@ impl TransactionsHandlerPrototype { } /// Returns the configuration of the set to put in the network configuration. - pub fn set_config(&self) -> config::NonDefaultSetConfig { - config::NonDefaultSetConfig { + pub fn set_config(&self) -> NonDefaultSetConfig { + NonDefaultSetConfig { notifications_protocol: self.protocol_name.clone(), fallback_names: self.fallback_protocol_names.clone(), max_notification_size: MAX_TRANSACTIONS_SIZE, - set_config: config::SetConfig { + set_config: SetConfig { in_peers: 0, out_peers: 0, reserved_nodes: Vec::new(), - non_reserved_mode: config::NonReservedPeerMode::Deny, + non_reserved_mode: NonReservedPeerMode::Deny, }, } } @@ -176,23 +158,25 @@ impl TransactionsHandlerPrototype { /// the behaviour of the handler while it's running. /// /// Important: the transactions handler is initially disabled and doesn't gossip transactions. - /// You must call [`TransactionsHandlerController::set_gossip_enabled`] to enable it. - pub fn build( + /// Gossiping is enabled when major syncing is done. + pub fn build< + B: BlockT + 'static, + H: ExHashT, + S: NetworkPeers + NetworkEventStream + NetworkNotification + sp_consensus::SyncOracle, + >( self, - service: Arc>, + service: S, transaction_pool: Arc>, metrics_registry: Option<&Registry>, - ) -> error::Result<(TransactionsHandler, TransactionsHandlerController)> { + ) -> error::Result<(TransactionsHandler, TransactionsHandlerController)> { let event_stream = service.event_stream("transactions-handler"); let (to_handler, from_controller) = mpsc::unbounded(); - let gossip_enabled = Arc::new(AtomicBool::new(false)); let handler = TransactionsHandler { protocol_name: self.protocol_name, propagate_timeout: Box::pin(interval(PROPAGATE_TIMEOUT)), pending_transactions: FuturesUnordered::new(), pending_transactions_peers: HashMap::new(), - gossip_enabled: gossip_enabled.clone(), service, event_stream, peers: HashMap::new(), @@ -205,7 +189,7 @@ impl TransactionsHandlerPrototype { }, }; - let controller = TransactionsHandlerController { to_handler, gossip_enabled }; + let controller = TransactionsHandlerController { to_handler }; Ok((handler, controller)) } @@ -214,15 +198,9 @@ impl TransactionsHandlerPrototype { /// Controls the behaviour of a [`TransactionsHandler`] it is connected to. pub struct TransactionsHandlerController { to_handler: mpsc::UnboundedSender>, - gossip_enabled: Arc, } impl TransactionsHandlerController { - /// Controls whether transactions are being gossiped on the network. - pub fn set_gossip_enabled(&mut self, enabled: bool) { - self.gossip_enabled.store(enabled, Ordering::Relaxed); - } - /// You may call this when new transactions are imported by the transaction pool. /// /// All transactions will be fetched from the `TransactionPool` that was passed at @@ -246,7 +224,11 @@ enum ToHandler { } /// Handler for transactions. Call [`TransactionsHandler::run`] to start the processing. -pub struct TransactionsHandler { +pub struct TransactionsHandler< + B: BlockT + 'static, + H: ExHashT, + S: NetworkPeers + NetworkEventStream + NetworkNotification + sp_consensus::SyncOracle, +> { protocol_name: ProtocolName, /// Interval at which we call `propagate_transactions`. propagate_timeout: Pin + Send>>, @@ -258,13 +240,12 @@ pub struct TransactionsHandler { /// multiple times concurrently. pending_transactions_peers: HashMap>, /// Network service to use to send messages and manage peers. - service: Arc>, + service: S, /// Stream of networking events. event_stream: Pin + Send>>, // All connected peers peers: HashMap>, transaction_pool: Arc>, - gossip_enabled: Arc, from_controller: mpsc::UnboundedReceiver>, /// Prometheus metrics. metrics: Option, @@ -278,7 +259,12 @@ struct Peer { role: ObservedRole, } -impl TransactionsHandler { +impl TransactionsHandler +where + B: BlockT + 'static, + H: ExHashT, + S: NetworkPeers + NetworkEventStream + NetworkNotification + sp_consensus::SyncOracle, +{ /// Turns the [`TransactionsHandler`] into a future that should run forever and not be /// interrupted. pub async fn run(mut self) { @@ -360,9 +346,9 @@ impl TransactionsHandler { continue } - if let Ok(m) = as Decode>::decode( - &mut message.as_ref(), - ) { + if let Ok(m) = + as Decode>::decode(&mut message.as_ref()) + { self.on_transactions(remote, m); } else { warn!(target: "sub-libp2p", "Failed to decode transactions list"); @@ -376,10 +362,10 @@ impl TransactionsHandler { } /// Called when peer sends us new transactions - fn on_transactions(&mut self, who: PeerId, transactions: message::Transactions) { - // Accept transactions only when enabled - if !self.gossip_enabled.load(Ordering::Relaxed) { - trace!(target: "sync", "{} Ignoring transactions while disabled", who); + fn on_transactions(&mut self, who: PeerId, transactions: Transactions) { + // Accept transactions only when node is not major syncing + if self.service.is_major_syncing() { + trace!(target: "sync", "{} Ignoring transactions while major syncing", who); return } @@ -428,10 +414,11 @@ impl TransactionsHandler { /// Propagate one transaction. pub fn propagate_transaction(&mut self, hash: &H) { - // Accept transactions only when enabled - if !self.gossip_enabled.load(Ordering::Relaxed) { + // Accept transactions only when node is not major syncing + if self.service.is_major_syncing() { return } + debug!(target: "sync", "Propagating transaction [{:?}]", hash); if let Some(transaction) = self.transaction_pool.transaction(hash) { let propagated_to = self.do_propagate_transactions(&[(hash.clone(), transaction)]); @@ -479,10 +466,11 @@ impl TransactionsHandler { /// Call when we must propagate ready transactions to peers. fn propagate_transactions(&mut self) { - // Accept transactions only when enabled - if !self.gossip_enabled.load(Ordering::Relaxed) { + // Accept transactions only when node is not major syncing + if self.service.is_major_syncing() { return } + debug!(target: "sync", "Propagating transactions"); let transactions = self.transaction_pool.transactions(); let propagated_to = self.do_propagate_transactions(&transactions); diff --git a/client/service/Cargo.toml b/client/service/Cargo.toml index 3c574ef13c8e6..e46c65cf018f5 100644 --- a/client/service/Cargo.toml +++ b/client/service/Cargo.toml @@ -56,6 +56,7 @@ sc-network-bitswap = { version = "0.10.0-dev", path = "../network/bitswap" } sc-network-common = { version = "0.10.0-dev", path = "../network/common" } sc-network-light = { version = "0.10.0-dev", path = "../network/light" } sc-network-sync = { version = "0.10.0-dev", path = "../network/sync" } +sc-network-transactions = { version = "0.10.0-dev", path = "../network/transactions" } sc-chain-spec = { version = "4.0.0-dev", path = "../chain-spec" } sc-client-api = { version = "4.0.0-dev", path = "../api" } sp-api = { version = "4.0.0-dev", path = "../../primitives/api" } diff --git a/client/service/src/builder.rs b/client/service/src/builder.rs index 5a2f4cf978b41..dfd532a14c172 100644 --- a/client/service/src/builder.rs +++ b/client/service/src/builder.rs @@ -40,7 +40,7 @@ use sc_keystore::LocalKeystore; use sc_network::{config::SyncMode, NetworkService}; use sc_network_bitswap::BitswapRequestHandler; use sc_network_common::{ - service::{NetworkStateInfo, NetworkStatusProvider, NetworkTransaction}, + service::{NetworkStateInfo, NetworkStatusProvider}, sync::warp::WarpSyncProvider, }; use sc_network_light::light_client_requests::handler::LightClientRequestHandler; @@ -326,7 +326,6 @@ where pub trait SpawnTaskNetwork: sc_offchain::NetworkProvider + NetworkStateInfo - + NetworkTransaction + NetworkStatusProvider + Send + Sync @@ -339,7 +338,6 @@ where Block: BlockT, T: sc_offchain::NetworkProvider + NetworkStateInfo - + NetworkTransaction + NetworkStatusProvider + Send + Sync @@ -368,6 +366,9 @@ pub struct SpawnTasksParams<'a, TBl: BlockT, TCl, TExPool, TRpc, Backend> { pub network: Arc>, /// A Sender for RPC requests. pub system_rpc_tx: TracingUnboundedSender>, + /// Controller for transactions handlers + pub tx_handler_controller: + sc_network_transactions::TransactionsHandlerController<::Hash>, /// Telemetry instance for this node. pub telemetry: Option<&'a mut Telemetry>, } @@ -446,6 +447,7 @@ where rpc_builder, network, system_rpc_tx, + tx_handler_controller, telemetry, } = params; @@ -481,7 +483,11 @@ where spawn_handle.spawn( "on-transaction-imported", Some("transaction-pool"), - transaction_notifications(transaction_pool.clone(), network.clone(), telemetry.clone()), + transaction_notifications( + transaction_pool.clone(), + tx_handler_controller, + telemetry.clone(), + ), ); // Prometheus metrics. @@ -544,20 +550,21 @@ where Ok(rpc_handlers) } -async fn transaction_notifications( +async fn transaction_notifications( transaction_pool: Arc, - network: Network, + tx_handler_controller: sc_network_transactions::TransactionsHandlerController< + ::Hash, + >, telemetry: Option, ) where Block: BlockT, ExPool: MaintainedTransactionPool::Hash>, - Network: NetworkTransaction<::Hash> + Send + Sync, { // transaction notifications transaction_pool .import_notification_stream() .for_each(move |hash| { - network.propagate_transaction(hash); + tx_handler_controller.propagate_transaction(hash); let status = transaction_pool.status(); telemetry!( telemetry; @@ -719,6 +726,7 @@ pub fn build_network( ( Arc::Hash>>, TracingUnboundedSender>, + sc_network_transactions::TransactionsHandlerController<::Hash>, NetworkStarter, ), Error, @@ -761,9 +769,6 @@ where } } - let transaction_pool_adapter = - Arc::new(TransactionPoolAdapter { pool: transaction_pool, client: client.clone() }); - let protocol_id = config.protocol_id(); let block_announce_validator = if let Some(f) = block_announce_validator_builder { @@ -845,7 +850,7 @@ where protocol_config })); - let network_params = sc_network::config::Params { + let mut network_params = sc_network::config::Params { role: config.role.clone(), executor: { let spawn_handle = Clone::clone(&spawn_handle); @@ -853,16 +858,9 @@ where spawn_handle.spawn("libp2p-node", Some("networking"), fut); })) }, - transactions_handler_executor: { - let spawn_handle = Clone::clone(&spawn_handle); - Box::new(move |fut| { - spawn_handle.spawn("network-transactions-handler", Some("networking"), fut); - }) - }, network_config: config.network.clone(), chain: client.clone(), - transaction_pool: transaction_pool_adapter as _, - protocol_id, + protocol_id: protocol_id.clone(), fork_id: config.chain_spec.fork_id().map(ToOwned::to_owned), import_queue: Box::new(import_queue), chain_sync: Box::new(chain_sync), @@ -877,10 +875,32 @@ where .collect::>(), }; + // crate transactions protocol and add it to the list of supported protocols of `network_params` + let transactions_handler_proto = sc_network_transactions::TransactionsHandlerPrototype::new( + protocol_id.clone(), + client + .block_hash(0u32.into()) + .ok() + .flatten() + .expect("Genesis block exists; qed"), + config.chain_spec.fork_id(), + ); + network_params + .network_config + .extra_sets + .insert(0, transactions_handler_proto.set_config()); + let has_bootnodes = !network_params.network_config.boot_nodes.is_empty(); let network_mut = sc_network::NetworkWorker::new(network_params)?; let network = network_mut.service().clone(); + let (tx_handler, tx_handler_controller) = transactions_handler_proto.build( + network.clone(), + Arc::new(TransactionPoolAdapter { pool: transaction_pool, client: client.clone() }), + config.prometheus_config.as_ref().map(|config| &config.registry), + )?; + spawn_handle.spawn("network-transactions-handler", Some("networking"), tx_handler.run()); + let (system_rpc_tx, system_rpc_rx) = tracing_unbounded("mpsc_system_rpc"); let future = build_network_future( @@ -928,7 +948,7 @@ where future.await }); - Ok((network, system_rpc_tx, NetworkStarter(network_start_tx))) + Ok((network, system_rpc_tx, tx_handler_controller, NetworkStarter(network_start_tx))) } /// Object used to start the network. diff --git a/client/service/src/config.rs b/client/service/src/config.rs index 44153e3b914f3..bca0697bcbd08 100644 --- a/client/service/src/config.rs +++ b/client/service/src/config.rs @@ -24,13 +24,11 @@ pub use sc_executor::WasmExecutionMethod; #[cfg(feature = "wasmtime")] pub use sc_executor::WasmtimeInstantiationStrategy; pub use sc_network::{ - config::{ - NetworkConfiguration, NodeKeyConfig, NonDefaultSetConfig, Role, SetConfig, TransportConfig, - }, + config::{NetworkConfiguration, NodeKeyConfig, Role}, Multiaddr, }; pub use sc_network_common::{ - config::{MultiaddrWithPeerId, ProtocolId}, + config::{MultiaddrWithPeerId, NonDefaultSetConfig, ProtocolId, SetConfig, TransportConfig}, request_responses::{ IncomingRequest, OutgoingResponse, ProtocolConfig as RequestResponseConfig, }, diff --git a/client/service/src/error.rs b/client/service/src/error.rs index 0d702c7f37b98..001a83922d776 100644 --- a/client/service/src/error.rs +++ b/client/service/src/error.rs @@ -19,7 +19,6 @@ //! Errors that can occur during the service operation. use sc_keystore; -use sc_network; use sp_blockchain; use sp_consensus; @@ -41,7 +40,7 @@ pub enum Error { Consensus(#[from] sp_consensus::Error), #[error(transparent)] - Network(#[from] sc_network::error::Error), + Network(#[from] sc_network_common::error::Error), #[error(transparent)] Keystore(#[from] sc_keystore::Error), diff --git a/client/service/src/lib.rs b/client/service/src/lib.rs index 19358c1e5bc4c..091b4bbe9fe5f 100644 --- a/client/service/src/lib.rs +++ b/client/service/src/lib.rs @@ -72,7 +72,7 @@ pub use sc_chain_spec::{ pub use sc_consensus::ImportQueue; pub use sc_executor::NativeExecutionDispatch; #[doc(hidden)] -pub use sc_network::config::{TransactionImport, TransactionImportFuture}; +pub use sc_network_transactions::config::{TransactionImport, TransactionImportFuture}; pub use sc_rpc::{ RandomIntegerSubscriptionId, RandomStringSubscriptionId, RpcSubscriptionIdProvider, }; @@ -148,7 +148,7 @@ async fn build_network_future< + Send + Sync + 'static, - H: sc_network::ExHashT, + H: sc_network_common::ExHashT, >( role: Role, mut network: sc_network::NetworkWorker, @@ -415,7 +415,8 @@ where .collect() } -impl sc_network::config::TransactionPool for TransactionPoolAdapter +impl sc_network_transactions::config::TransactionPool + for TransactionPoolAdapter where C: HeaderBackend + BlockBackend diff --git a/client/service/test/src/lib.rs b/client/service/test/src/lib.rs index 23245d46cba10..5d29d34a3cbf2 100644 --- a/client/service/test/src/lib.rs +++ b/client/service/test/src/lib.rs @@ -22,12 +22,9 @@ use futures::{task::Poll, Future, TryFutureExt as _}; use log::{debug, info}; use parking_lot::Mutex; use sc_client_api::{Backend, CallExecutor}; -use sc_network::{ - config::{NetworkConfiguration, TransportConfig}, - multiaddr, -}; +use sc_network::{config::NetworkConfiguration, multiaddr}; use sc_network_common::{ - config::MultiaddrWithPeerId, + config::{MultiaddrWithPeerId, TransportConfig}, service::{NetworkBlock, NetworkPeers, NetworkStateInfo}, }; use sc_service::{