diff --git a/common/clients/multi-tcp-client/src/connection_manager/mod.rs b/common/clients/multi-tcp-client/src/connection_manager/mod.rs index 57a56bb810..b4f26b6c59 100644 --- a/common/clients/multi-tcp-client/src/connection_manager/mod.rs +++ b/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, } } diff --git a/common/clients/multi-tcp-client/src/connection_manager/reconnector.rs b/common/clients/multi-tcp-client/src/connection_manager/reconnector.rs index b8e1613093..293ae32869 100644 --- a/common/clients/multi-tcp-client/src/connection_manager/reconnector.rs +++ b/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(); diff --git a/common/clients/multi-tcp-client/src/lib.rs b/common/clients/multi-tcp-client/src/lib.rs index 38c0b88511..86013adb0f 100644 --- a/common/clients/multi-tcp-client/src/lib.rs +++ b/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, 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()]; diff --git a/common/healthcheck/src/lib.rs b/common/healthcheck/src/lib.rs index 04196db3ed..0aee269d13 100644 --- a/common/healthcheck/src/lib.rs +++ b/common/healthcheck/src/lib.rs @@ -36,17 +36,20 @@ impl From 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; diff --git a/common/healthcheck/src/path_check.rs b/common/healthcheck/src/path_check.rs index b55f95afd5..fb68953ee2 100644 --- a/common/healthcheck/src/path_check.rs +++ b/common/healthcheck/src/path_check.rs @@ -30,6 +30,7 @@ impl PathChecker { pub(crate) async fn new( providers: Vec, 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 { diff --git a/common/healthcheck/src/result.rs b/common/healthcheck/src/result.rs index 2db51ddc3a..12137fb55d 100644 --- a/common/healthcheck/src/result.rs +++ b/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 { diff --git a/mixnode/src/config/mod.rs b/mixnode/src/config/mod.rs index ed8cd4728a..75f206aa52 100644 --- a/mixnode/src/config/mod.rs +++ b/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, } } } diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index 547d6c2e63..338734eda1 100644 --- a/mixnode/src/node/mod.rs +++ b/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()) diff --git a/mixnode/src/node/packet_forwarding.rs b/mixnode/src/node/packet_forwarding.rs index bdba773d84..58d2fce1b7 100644 --- a/mixnode/src/node/packet_forwarding.rs +++ b/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(); diff --git a/nym-client/src/client/mix_traffic.rs b/nym-client/src/client/mix_traffic.rs index 515eeaa1f1..5e322db539 100644 --- a/nym-client/src/client/mix_traffic.rs +++ b/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 { diff --git a/nym-client/src/client/mod.rs b/nym-client/src/client/mod.rs index b009b7c7b8..4a6af32df8 100644 --- a/nym-client/src/client/mod.rs +++ b/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, ) }) diff --git a/nym-client/src/client/topology_control.rs b/nym-client/src/client/topology_control.rs index 06cfa4618c..4f4de04ef0 100644 --- a/nym-client/src/client/topology_control.rs +++ b/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 TopologyRefresher { // 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, ); diff --git a/nym-client/src/config/mod.rs b/nym-client/src/config/mod.rs index 53d45ce298..d747a22eb4 100644 --- a/nym-client/src/config/mod.rs +++ b/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, 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, } } } diff --git a/validator/src/config/mod.rs b/validator/src/config/mod.rs index 47fba50311..1396287272 100644 --- a/validator/src/config/mod.rs +++ b/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, } } } diff --git a/validator/src/config/template.rs b/validator/src/config/template.rs index 71164f6737..17ba8d1ca8 100644 --- a/validator/src/config/template.rs +++ b/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 }} diff --git a/validator/src/validator.rs b/validator/src/validator.rs index 9c12cef95a..51c78d9402 100644 --- a/validator/src/validator.rs +++ b/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, );