Feature/tcp client connection timeout (#176)

* Added timeout values to configs

* Fixed possible crash when using delay larger than 2 years

* Connection timeout with hardcoded value

* Using provided timeout value

* tcp client requiring timeout value at construction

* Client using updated tcp client

* Mixnode using updated tcp client

* Healthchecker having separate timeout + new client config field

* Updated validator with connection timeout field

* Added connection_timeout to validator config template

common/clients/multi-tcp-client/src/connection_manager/mod.rs

@@ -46,17 +46,21 @@ impl<'a> ConnectionManager<'static> {
         address: SocketAddr,
         reconnection_backoff: Duration,
         maximum_reconnection_backoff: Duration,
+        connection_timeout: Duration,
     ) -> ConnectionManager<'a> {
         let (conn_tx, conn_rx) = mpsc::unbounded();
 
-        // based on initial connection we will either have a writer or a reconnector
-        let state = match tokio::net::TcpStream::connect(address).await {
-            Ok(conn) => ConnectionState::Writing(ConnectionWriter::new(conn)),
+        // the blocking call here is fine as initially we want to wait the timeout interval (at most) anyway:
+        let tcp_stream_res = std::net::TcpStream::connect_timeout(&address, connection_timeout);
+
+        let initial_state = match tcp_stream_res {
+            Ok(stream) => {
+                let tokio_stream = tokio::net::TcpStream::from_std(stream).unwrap();
+                debug!("managed to establish initial connection to {}", address);
+                ConnectionState::Writing(ConnectionWriter::new(tokio_stream))
+            }
             Err(e) => {
-                warn!(
-                    "failed to establish initial connection to {} ({}). Going into reconnection mode",
-                    address, e
-                );
+                warn!("failed to establish initial connection to {} within {:?} ({}). Going into reconnection mode", address, connection_timeout, e);
                 ConnectionState::Reconnecting(ConnectionReconnector::new(
                     address,
                     reconnection_backoff,
@@ -71,7 +75,7 @@ impl<'a> ConnectionManager<'static> {
             address,
             maximum_reconnection_backoff,
             reconnection_backoff,
-            state,
+            state: initial_state,
         }
     }
 

common/clients/multi-tcp-client/src/connection_manager/reconnector.rs

@@ -60,15 +60,25 @@ impl<'a> Future for ConnectionReconnector<'a> {
                 self.current_retry_attempt += 1;
 
                 // we failed to re-establish connection - continue exponential backoff
+
+                // according to https://github.com/tokio-rs/tokio/issues/1953 there's an undocumented
+                // limit of tokio delay of about 2 years.
+                // let's ensure our delay is always on a sane side of being maximum 1 day.
+                let maximum_sane_delay = Duration::from_secs(24 * 60 * 60);
+
                 let next_delay = std::cmp::min(
-                    self.maximum_reconnection_backoff,
+                    maximum_sane_delay,
                     self.initial_reconnection_backoff
                         .checked_mul(2_u32.pow(self.current_retry_attempt))
                         .unwrap_or_else(|| self.maximum_reconnection_backoff),
                 );
 
-                let new_delay_instant = tokio::time::Instant::now().checked_add(next_delay).unwrap_or_else(|| panic!("You seem to have an error in your config file, your maximum reconnenction backoff is {:?} !", self.maximum_reconnection_backoff));
-                self.current_backoff_delay.reset(new_delay_instant);
+                let now = self.current_backoff_delay.deadline();
+                // this can't overflow now because next_delay is limited to one day...
+                // ... well, unless you've been running the system for couple of decades without
+                // any restarts....
+                let next = now + next_delay;
+                self.current_backoff_delay.reset(next);
 
                 self.connection = tokio::net::TcpStream::connect(self.address).boxed();
 

common/clients/multi-tcp-client/src/lib.rs

@@ -13,16 +13,19 @@ mod connection_manager;
 pub struct Config {
     initial_reconnection_backoff: Duration,
     maximum_reconnection_backoff: Duration,
+    initial_connection_timeout: Duration,
 }
 
 impl Config {
     pub fn new(
         initial_reconnection_backoff: Duration,
         maximum_reconnection_backoff: Duration,
+        initial_connection_timeout: Duration,
     ) -> Self {
         Config {
             initial_reconnection_backoff,
             maximum_reconnection_backoff,
+            initial_connection_timeout,
         }
     }
 }
@@ -32,6 +35,7 @@ pub struct Client {
     connections_managers: HashMap<SocketAddr, (ConnectionManagerSender, AbortHandle)>,
     maximum_reconnection_backoff: Duration,
     initial_reconnection_backoff: Duration,
+    initial_connection_timeout: Duration,
 }
 
 impl Client {
@@ -44,6 +48,7 @@ impl Client {
             connections_managers: HashMap::new(),
             initial_reconnection_backoff: config.maximum_reconnection_backoff,
             maximum_reconnection_backoff: config.initial_reconnection_backoff,
+            initial_connection_timeout: config.initial_connection_timeout,
         }
     }
 
@@ -55,6 +60,7 @@ impl Client {
             address,
             self.initial_reconnection_backoff,
             self.maximum_reconnection_backoff,
+            self.initial_connection_timeout,
         )
         .await
         .start_abortable(&self.runtime_handle);
@@ -165,7 +171,8 @@ mod tests {
         let mut rt = tokio::runtime::Runtime::new().unwrap();
         let addr = "127.0.0.1:6000".parse().unwrap();
         let reconnection_backoff = Duration::from_secs(1);
-        let client_config = Config::new(reconnection_backoff, 10 * reconnection_backoff);
+        let timeout = Duration::from_secs(1);
+        let client_config = Config::new(reconnection_backoff, 10 * reconnection_backoff, timeout);
 
         let messages_to_send = vec![[1u8; SERVER_MSG_LEN].to_vec(), [2; SERVER_MSG_LEN].to_vec()];
 
@@ -224,7 +231,8 @@ mod tests {
         let mut rt = tokio::runtime::Runtime::new().unwrap();
         let addr = "127.0.0.1:6001".parse().unwrap();
         let reconnection_backoff = Duration::from_secs(2);
-        let client_config = Config::new(reconnection_backoff, 10 * reconnection_backoff);
+        let timeout = Duration::from_secs(1);
+        let client_config = Config::new(reconnection_backoff, 10 * reconnection_backoff, timeout);
 
         let messages_to_send = vec![[1u8; SERVER_MSG_LEN].to_vec(), [2; SERVER_MSG_LEN].to_vec()];
 

common/healthcheck/src/lib.rs

@@ -36,17 +36,20 @@ impl From<topology::NymTopologyError> for HealthCheckerError {
 pub struct HealthChecker {
     num_test_packets: usize,
     resolution_timeout: Duration,
+    connection_timeout: Duration,
     identity_keypair: MixIdentityKeyPair,
 }
 
 impl HealthChecker {
     pub fn new(
         resolution_timeout: Duration,
+        connection_timeout: Duration,
         num_test_packets: usize,
         identity_keypair: MixIdentityKeyPair,
     ) -> Self {
         HealthChecker {
             resolution_timeout,
+            connection_timeout,
             num_test_packets,
             identity_keypair,
         }
@@ -62,6 +65,7 @@ impl HealthChecker {
             current_topology,
             self.num_test_packets,
             self.resolution_timeout,
+            self.connection_timeout,
             &self.identity_keypair,
         )
         .await;

common/healthcheck/src/path_check.rs

@@ -30,6 +30,7 @@ impl PathChecker {
     pub(crate) async fn new(
         providers: Vec<provider::Node>,
         identity_keys: &MixIdentityKeyPair,
+        connection_timeout: Duration,
         check_id: [u8; 16],
     ) -> Self {
         let mut provider_clients = HashMap::new();
@@ -67,6 +68,7 @@ impl PathChecker {
         let mixnet_client_config = multi_tcp_client::Config::new(
             Duration::from_secs(1_000_000_000),
             Duration::from_secs(1_000_000_000),
+            connection_timeout,
         );
 
         PathChecker {

common/healthcheck/src/result.rs

@@ -102,6 +102,7 @@ impl HealthCheckResult {
         topology: &T,
         iterations: usize,
         resolution_timeout: Duration,
+        connection_timeout: Duration,
         identity_keys: &MixIdentityKeyPair,
     ) -> Self {
         // currently healthchecker supports only up to 255 iterations - if we somehow
@@ -127,7 +128,8 @@ impl HealthCheckResult {
 
         let providers = topology.providers();
 
-        let mut path_checker = PathChecker::new(providers, identity_keys, check_id).await;
+        let mut path_checker =
+            PathChecker::new(providers, identity_keys, connection_timeout, check_id).await;
         for i in 0..iterations {
             debug!("running healthcheck iteration {} / {}", i + 1, iterations);
             for path in &all_paths {

mixnode/src/config/mod.rs

@@ -20,6 +20,7 @@ const DEFAULT_METRICS_SENDING_DELAY: u64 = 1000;
 const DEFAULT_METRICS_RUNNING_STATS_LOGGING_DELAY: u64 = 60_000; // 1min
 const DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF: u64 = 10_000; // 10s
 const DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF: u64 = 300_000; // 5min
+const DEFAULT_INITIAL_CONNECTION_TIMEOUT: u64 = 1_500; // 1.5s
 
 #[derive(Debug, Default, Deserialize, PartialEq, Serialize)]
 #[serde(deny_unknown_fields)]
@@ -237,6 +238,10 @@ impl Config {
     pub fn get_packet_forwarding_maximum_backoff(&self) -> time::Duration {
         time::Duration::from_millis(self.debug.packet_forwarding_maximum_backoff)
     }
+
+    pub fn get_initial_connection_timeout(&self) -> time::Duration {
+        time::Duration::from_millis(self.debug.initial_connection_timeout)
+    }
 }
 
 #[derive(Debug, Deserialize, PartialEq, Serialize)]
@@ -349,6 +354,10 @@ pub struct Debug {
     /// forwarding sphinx packets.
     /// The provided value is interpreted as milliseconds.
     packet_forwarding_maximum_backoff: u64,
+
+    /// Timeout for establishing initial connection when trying to forward a sphinx packet.
+    /// The provider value is interpreted as milliseconds.
+    initial_connection_timeout: u64,
 }
 
 impl Debug {
@@ -372,6 +381,7 @@ impl Default for Debug {
             metrics_running_stats_logging_delay: DEFAULT_METRICS_RUNNING_STATS_LOGGING_DELAY,
             packet_forwarding_initial_backoff: DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF,
             packet_forwarding_maximum_backoff: DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF,
+            initial_connection_timeout: DEFAULT_INITIAL_CONNECTION_TIMEOUT,
         }
     }
 }

mixnode/src/node/mod.rs

@@ -95,6 +95,7 @@ impl MixNode {
                 packet_forwarding::PacketForwarder::new(
                     self.config.get_packet_forwarding_initial_backoff(),
                     self.config.get_packet_forwarding_maximum_backoff(),
+                    self.config.get_initial_connection_timeout(),
                 )
             })
             .start(self.runtime.handle())

mixnode/src/node/packet_forwarding.rs

@@ -14,10 +14,12 @@ impl PacketForwarder {
     pub(crate) fn new(
         initial_reconnection_backoff: Duration,
         maximum_reconnection_backoff: Duration,
+        initial_connection_timeout: Duration,
     ) -> PacketForwarder {
         let tcp_client_config = multi_tcp_client::Config::new(
             initial_reconnection_backoff,
             maximum_reconnection_backoff,
+            initial_connection_timeout,
         );
 
         let (conn_tx, conn_rx) = mpsc::unbounded();

nym-client/src/client/mix_traffic.rs

@@ -26,11 +26,13 @@ impl MixTrafficController {
     pub(crate) fn new(
         initial_reconnection_backoff: Duration,
         maximum_reconnection_backoff: Duration,
+        initial_connection_timeout: Duration,
         mix_rx: MixMessageReceiver,
     ) -> Self {
         let tcp_client_config = multi_tcp_client::Config::new(
             initial_reconnection_backoff,
             maximum_reconnection_backoff,
+            initial_connection_timeout,
         );
 
         MixTrafficController {

nym-client/src/client/mod.rs

@@ -195,6 +195,7 @@ impl NymClient {
             self.config.get_topology_refresh_rate(),
             healthcheck_keys,
             self.config.get_topology_resolution_timeout(),
+            self.config.get_healthcheck_connection_timeout(),
             self.config.get_number_of_healthcheck_test_packets() as usize,
             self.config.get_node_score_threshold(),
         );
@@ -219,6 +220,7 @@ impl NymClient {
                 MixTrafficController::new(
                     self.config.get_packet_forwarding_initial_backoff(),
                     self.config.get_packet_forwarding_maximum_backoff(),
+                    self.config.get_initial_connection_timeout(),
                     mix_rx,
                 )
             })

nym-client/src/client/topology_control.rs

@@ -110,6 +110,7 @@ pub(crate) struct TopologyRefresherConfig {
     refresh_rate: time::Duration,
     identity_keypair: MixIdentityKeyPair,
     resolution_timeout: time::Duration,
+    connection_timeout: time::Duration,
     number_test_packets: usize,
     node_score_threshold: f64,
 }
@@ -120,6 +121,7 @@ impl TopologyRefresherConfig {
         refresh_rate: time::Duration,
         identity_keypair: MixIdentityKeyPair,
         resolution_timeout: time::Duration,
+        connection_timeout: time::Duration,
         number_test_packets: usize,
         node_score_threshold: f64,
     ) -> Self {
@@ -128,6 +130,7 @@ impl TopologyRefresherConfig {
             refresh_rate,
             identity_keypair,
             resolution_timeout,
+            connection_timeout,
             number_test_packets,
             node_score_threshold,
         }
@@ -150,6 +153,7 @@ impl<T: 'static + NymTopology> TopologyRefresher<T> {
         // this is a temporary solution as the healthcheck will eventually be moved to validators
         let health_checker = healthcheck::HealthChecker::new(
             cfg.resolution_timeout,
+            cfg.connection_timeout,
             cfg.number_test_packets,
             cfg.identity_keypair,
         );

nym-client/src/config/mod.rs

@@ -20,6 +20,8 @@ const DEFAULT_TOPOLOGY_REFRESH_RATE: u64 = 10_000;
 const DEFAULT_TOPOLOGY_RESOLUTION_TIMEOUT: u64 = 5_000;
 const DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF: u64 = 10_000; // 10s
 const DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF: u64 = 300_000; // 5min
+const DEFAULT_INITIAL_CONNECTION_TIMEOUT: u64 = 1_500; // 1.5s
+const DEFAULT_HEALTHCHECK_CONNECTION_TIMEOUT: u64 = DEFAULT_INITIAL_CONNECTION_TIMEOUT;
 
 const DEFAULT_NUMBER_OF_HEALTHCHECK_TEST_PACKETS: u64 = 2;
 const DEFAULT_NODE_SCORE_THRESHOLD: f64 = 0.0;
@@ -213,6 +215,14 @@ impl Config {
     pub fn get_packet_forwarding_maximum_backoff(&self) -> time::Duration {
         time::Duration::from_millis(self.debug.packet_forwarding_maximum_backoff)
     }
+
+    pub fn get_initial_connection_timeout(&self) -> time::Duration {
+        time::Duration::from_millis(self.debug.initial_connection_timeout)
+    }
+
+    pub fn get_healthcheck_connection_timeout(&self) -> time::Duration {
+        time::Duration::from_millis(self.debug.healthcheck_connection_timeout)
+    }
 }
 
 fn de_option_string<'de, D>(deserializer: D) -> Result<Option<String>, D::Error>
@@ -376,6 +386,15 @@ pub struct Debug {
     /// forwarding sphinx packets.
     /// The provided value is interpreted as milliseconds.
     packet_forwarding_maximum_backoff: u64,
+
+    /// Timeout for establishing initial connection when trying to forward a sphinx packet.
+    /// The provider value is interpreted as milliseconds.
+    initial_connection_timeout: u64,
+
+    /// Timeout for establishing initial connection when trying to forward a sphinx packet
+    /// during healthcheck.
+    /// The provider value is interpreted as milliseconds.
+    healthcheck_connection_timeout: u64,
 }
 
 impl Default for Debug {
@@ -392,6 +411,8 @@ impl Default for Debug {
             node_score_threshold: DEFAULT_NODE_SCORE_THRESHOLD,
             packet_forwarding_initial_backoff: DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF,
             packet_forwarding_maximum_backoff: DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF,
+            initial_connection_timeout: DEFAULT_INITIAL_CONNECTION_TIMEOUT,
+            healthcheck_connection_timeout: DEFAULT_HEALTHCHECK_CONNECTION_TIMEOUT,
         }
     }
 }

validator/src/config/mod.rs

@@ -11,6 +11,7 @@ mod template;
 // 'MIXMINING'
 const DEFAULT_MIX_MINING_DELAY: u64 = 10_000;
 const DEFAULT_MIX_MINING_RESOLUTION_TIMEOUT: u64 = 5_000;
+const DEFAULT_MIX_MINING_CONNECTION_TIMEOUT: u64 = 1_500;
 
 const DEFAULT_NUMBER_OF_MIX_MINING_TEST_PACKETS: u64 = 2;
 
@@ -139,6 +140,10 @@ impl Config {
     pub fn get_mix_mining_number_of_test_packets(&self) -> u64 {
         self.mix_mining.number_of_test_packets
     }
+
+    pub fn get_healthcheck_connection_timeout(&self) -> time::Duration {
+        time::Duration::from_millis(self.mix_mining.connection_timeout)
+    }
 }
 
 #[derive(Debug, Deserialize, PartialEq, Serialize)]
@@ -190,6 +195,10 @@ pub struct MixMining {
     /// The provided value is interpreted as milliseconds.
     resolution_timeout: u64,
 
+    /// Timeout for trying to establish connection to node endpoints.
+    /// The provided value is interpreted as milliseconds.
+    connection_timeout: u64,
+
     /// How many packets should be sent through each path during the mix-mining procedure.
     number_of_test_packets: u64,
 }
@@ -201,6 +210,7 @@ impl Default for MixMining {
             run_delay: DEFAULT_MIX_MINING_DELAY,
             resolution_timeout: DEFAULT_MIX_MINING_RESOLUTION_TIMEOUT,
             number_of_test_packets: DEFAULT_NUMBER_OF_MIX_MINING_TEST_PACKETS,
+            connection_timeout: DEFAULT_MIX_MINING_CONNECTION_TIMEOUT,
         }
     }
 }

validator/src/config/template.rs

@@ -43,6 +43,10 @@ run_delay = {{ mix_mining.run_delay }}
 # The provided value is interpreted as milliseconds.
 resolution_timeout = {{ mix_mining.resolution_timeout }}
 
+# Timeout for trying to establish connection to node endpoints.
+# The provided value is interpreted as milliseconds.
+connection_timeout =  {{ mix_mining.connection_timeout }}
+    
 # How many packets should be sent through each path during the mix-mining procedure.
 number_of_test_packets = {{ mix_mining.number_of_test_packets }}
 

validator/src/validator.rs

@@ -20,6 +20,7 @@ impl Validator {
         let dummy_healthcheck_keypair = MixIdentityKeyPair::new();
         let hc = HealthChecker::new(
             config.get_mix_mining_resolution_timeout(),
+            config.get_healthcheck_connection_timeout(),
             config.get_mix_mining_number_of_test_packets() as usize,
             dummy_healthcheck_keypair,
         );