TIER1 implementation (#8141)

Implemented TIER1 connections:
* added TIER1 support to PeerActor
* added logic connecting a TIER1 node to its proxies before broadcasting its AccountData.
* added logic making TIER1 nodes connect to other TIER1 nodes (or proxies) based on the collected AccountData.
* made TIER1 nodes send some specific message types over TIER1 connections (with a fallback to TIER2).
* made TIER1 proxies route the TIER1 messages.
* added e2e tests of the TIER1 functionality

Monitoring of the TIER1 performance will come in the next PR.

chain/network/src/config.rs

@@ -12,6 +12,7 @@ use near_crypto::{KeyType, SecretKey};
 use near_primitives::network::PeerId;
 use near_primitives::types::AccountId;
 use near_primitives::validator_signer::{InMemoryValidatorSigner, ValidatorSigner};
+use std::collections::HashSet;
 use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
 use std::sync::Arc;
 
@@ -67,9 +68,23 @@ impl ValidatorConfig {
 
 #[derive(Clone)]
 pub struct Tier1 {
+    /// Interval between attempts to connect to proxies of other TIER1 nodes.
+    pub connect_interval: time::Duration,
+    /// Maximal number of new connections established every connect_interval.
+    /// TIER1 can consists of hundreds of nodes, so it is not feasible to connect to all of them at
+    /// once.
+    pub new_connections_per_attempt: u64,
     /// Interval between broacasts of the list of validator's proxies.
     /// Before the broadcast, validator tries to establish all the missing connections to proxies.
     pub advertise_proxies_interval: time::Duration,
+    /// Support for gradual TIER1 feature rollout:
+    /// - establishing connection to node's own proxies is always enabled (it is a part of peer
+    ///   discovery mechanism). Note that unless the proxy has enable_inbound set, establishing
+    ///   those connections will fail anyway.
+    /// - a node will start accepting TIER1 inbound connections iff `enable_inbound` is true.
+    /// - a node will try to start outbound TIER1 connections iff `enable_outbound` is true.
+    pub enable_inbound: bool,
+    pub enable_outbound: bool,
 }
 
 /// Validated configuration for the peer-to-peer manager.
@@ -164,7 +179,12 @@ impl NetworkConfig {
         if cfg.public_addrs.len() > 0 && cfg.trusted_stun_servers.len() > 0 {
             anyhow::bail!("you cannot specify both public_addrs and trusted_stun_servers");
         }
+        let mut proxies = HashSet::new();
         for proxy in &cfg.public_addrs {
+            if proxies.contains(&proxy.peer_id) {
+                anyhow::bail!("public_addrs: found multiple entries with peer_id {}. Only 1 entry per peer_id is supported.",proxy.peer_id);
+            }
+            proxies.insert(proxy.peer_id.clone());
             let ip = proxy.addr.ip();
             if cfg.allow_private_ip_in_public_addrs {
                 if ip.is_unspecified() {
@@ -253,7 +273,13 @@ impl NetworkConfig {
             archive,
             accounts_data_broadcast_rate_limit: rate::Limit { qps: 0.1, burst: 1 },
             routing_table_update_rate_limit: rate::Limit { qps: 1., burst: 1 },
-            tier1: Some(Tier1 { advertise_proxies_interval: time::Duration::minutes(15) }),
+            tier1: Some(Tier1 {
+                connect_interval: cfg.experimental.tier1_connect_interval.try_into()?,
+                new_connections_per_attempt: cfg.experimental.tier1_new_connections_per_attempt,
+                advertise_proxies_interval: time::Duration::minutes(15),
+                enable_inbound: cfg.experimental.tier1_enable_inbound,
+                enable_outbound: cfg.experimental.tier1_enable_outbound,
+            }),
             inbound_disabled: cfg.experimental.inbound_disabled,
             skip_tombstones: if cfg.experimental.skip_sending_tombstones_seconds > 0 {
                 Some(time::Duration::seconds(cfg.experimental.skip_sending_tombstones_seconds))
@@ -321,7 +347,11 @@ impl NetworkConfig {
             tier1: Some(Tier1 {
                 // Interval is very large, so that it doesn't happen spontaneously in tests.
                 // It should rather be triggered manually in tests.
+                connect_interval: time::Duration::hours(1000),
+                new_connections_per_attempt: 10000,
                 advertise_proxies_interval: time::Duration::hours(1000),
+                enable_inbound: true,
+                enable_outbound: true,
             }),
             skip_tombstones: None,
             event_sink: Sink::null(),

chain/network/src/config_json.rs

@@ -176,6 +176,23 @@ pub struct Config {
     pub experimental: ExperimentalConfig,
 }
 
+fn default_tier1_enable_inbound() -> bool {
+    true
+}
+/// This default will be changed over the next releases.
+/// It allows us to gradually roll out the TIER1 feature.
+fn default_tier1_enable_outbound() -> bool {
+    false
+}
+
+fn default_tier1_connect_interval() -> Duration {
+    Duration::from_secs(60)
+}
+
+fn default_tier1_new_connections_per_attempt() -> u64 {
+    50
+}
+
 #[derive(Serialize, Deserialize, Clone, Debug)]
 pub struct ExperimentalConfig {
     // If true - don't allow any inbound connections.
@@ -192,6 +209,22 @@ pub struct ExperimentalConfig {
     // compatibility.
     #[serde(default = "default_skip_tombstones")]
     pub skip_sending_tombstones_seconds: i64,
+
+    /// See `near_network::config::Tier1::enable_inbound`.
+    #[serde(default = "default_tier1_enable_inbound")]
+    pub tier1_enable_inbound: bool,
+
+    /// See `near_network::config::Tier1::enable_outbound`.
+    #[serde(default = "default_tier1_enable_outbound")]
+    pub tier1_enable_outbound: bool,
+
+    /// See `near_network::config::Tier1::connect_interval`.
+    #[serde(default = "default_tier1_connect_interval")]
+    pub tier1_connect_interval: Duration,
+
+    /// See `near_network::config::Tier1::new_connections_per_attempt`.
+    #[serde(default = "default_tier1_new_connections_per_attempt")]
+    pub tier1_new_connections_per_attempt: u64,
 }
 
 impl Default for ExperimentalConfig {
@@ -200,6 +233,10 @@ impl Default for ExperimentalConfig {
             inbound_disabled: false,
             connect_only_to_boot_nodes: false,
             skip_sending_tombstones_seconds: default_skip_tombstones(),
+            tier1_enable_inbound: default_tier1_enable_inbound(),
+            tier1_enable_outbound: default_tier1_enable_outbound(),
+            tier1_connect_interval: default_tier1_connect_interval(),
+            tier1_new_connections_per_attempt: default_tier1_new_connections_per_attempt(),
         }
     }
 }

chain/network/src/network_protocol/testonly.rs

@@ -229,6 +229,10 @@ impl ChunkSet {
     }
 }
 
+pub fn make_hash<R: Rng>(rng: &mut R) -> CryptoHash {
+    CryptoHash::hash_bytes(&rng.gen::<[u8; 19]>())
+}
+
 pub fn make_account_keys(signers: &[InMemoryValidatorSigner]) -> AccountKeys {
     let mut account_keys = AccountKeys::new();
     for s in signers {