add topology access to nodes

Cargo.lock

@@ -6547,6 +6547,7 @@ dependencies = [
  "log",
  "nym-api-requests",
  "nym-bin-common",
+ "nym-client-core",
  "nym-coconut-interface",
  "nym-config",
  "nym-credentials",
@@ -6562,6 +6563,7 @@ dependencies = [
  "nym-sphinx",
  "nym-statistics-common",
  "nym-task",
+ "nym-topology",
  "nym-types",
  "nym-validator-client",
  "nym-wireguard",
@@ -6752,6 +6754,7 @@ dependencies = [
  "lazy_static",
  "log",
  "nym-bin-common",
+ "nym-client-core",
  "nym-config",
  "nym-contracts-common",
  "nym-crypto",

common/client-core/src/client/topology_control/accessor.rs

@@ -83,6 +83,16 @@ impl<'a> TopologyReadPermit<'a> {
 
         Ok(topology)
     }
+
+    pub fn try_get_raw_topology_ref(&'a self) -> Result<&'a NymTopology, NymTopologyError> {
+        // 1. Have we managed to get anything from the refresher, i.e. have the nym-api queries gone through?
+        let topology = self
+            .permit
+            .as_ref()
+            .ok_or(NymTopologyError::EmptyNetworkTopology)?;
+
+        Ok(topology)
+    }
 }
 
 impl<'a> From<RwLockReadGuard<'a, Option<NymTopology>>> for TopologyReadPermit<'a> {

common/client-core/src/client/topology_control/mod.rs

@@ -16,9 +16,9 @@ use tokio::time::sleep;
 #[cfg(target_arch = "wasm32")]
 use wasmtimer::tokio::sleep;
 
-mod accessor;
+pub mod accessor;
 pub mod geo_aware_provider;
-pub(crate) mod nym_api_provider;
+pub mod nym_api_provider;
 
 // TODO: move it to config later
 const MAX_FAILURE_COUNT: usize = 10;

common/client-core/src/client/topology_control/nym_api_provider.rs

@@ -9,7 +9,7 @@ use rand::prelude::SliceRandom;
 use rand::thread_rng;
 use url::Url;
 
-pub(crate) struct NymApiTopologyProvider {
+pub struct NymApiTopologyProvider {
     validator_client: nym_validator_client::client::NymApiClient,
     nym_api_urls: Vec<Url>,
 
@@ -18,7 +18,7 @@ pub(crate) struct NymApiTopologyProvider {
 }
 
 impl NymApiTopologyProvider {
-    pub(crate) fn new(mut nym_api_urls: Vec<Url>, client_version: String) -> Self {
+    pub fn new(mut nym_api_urls: Vec<Url>, client_version: String) -> Self {
         nym_api_urls.shuffle(&mut thread_rng());
 
         NymApiTopologyProvider {

gateway/Cargo.toml

@@ -78,6 +78,8 @@ nym-task = { path = "../common/task" }
 nym-types = { path = "../common/types" }
 nym-validator-client = { path = "../common/client-libs/validator-client" }
 nym-ip-packet-router = { path = "../service-providers/ip-packet-router" }
+nym-client-core = { path = "../common/client-core"}
+nym-topology = { path = "../common/topology" }
 
 nym-wireguard = { path = "../common/wireguard", optional = true }
 defguard_wireguard_rs = { git = "https://github.com/neacsu/wireguard-rs.git", rev = "c2cd0c1119f699f4bc43f5e6ffd6fc242caa42ed", optional = true }

gateway/src/config/mod.rs

@@ -38,6 +38,9 @@ const DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF: Duration = Duration::from_milli
 const DEFAULT_INITIAL_CONNECTION_TIMEOUT: Duration = Duration::from_millis(1_500);
 const DEFAULT_MAXIMUM_CONNECTION_BUFFER_SIZE: usize = 2000;
 
+const DEFAULT_TOPOLOGY_REFRESH_RATE: Duration = Duration::from_secs(5 * 60); // every 5min
+const DEFAULT_TOPOLOGY_RESOLUTION_TIMEOUT: Duration = Duration::from_millis(5_000);
+
 const DEFAULT_STORED_MESSAGE_FILENAME_LENGTH: u16 = 16;
 const DEFAULT_MESSAGE_RETRIEVAL_LIMIT: i64 = 100;
 
@@ -109,6 +112,9 @@ pub struct Config {
 
     #[serde(default)]
     pub debug: Debug,
+
+    #[serde(default)]
+    pub topology: Topology,
 }
 
 impl NymConfigTemplate for Config {
@@ -135,6 +141,7 @@ impl Config {
             ip_packet_router: Default::default(),
             logging: Default::default(),
             debug: Default::default(),
+            topology: Default::default(),
         }
     }
 
@@ -442,3 +449,33 @@ impl Default for Debug {
         }
     }
 }
+
+#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Serialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct Topology {
+    /// The uniform delay every which clients are querying the directory server
+    /// to try to obtain a compatible network topology to send sphinx packets through.
+    #[serde(with = "humantime_serde")]
+    pub topology_refresh_rate: Duration,
+
+    /// During topology refresh, test packets are sent through every single possible network
+    /// path. This timeout determines waiting period until it is decided that the packet
+    /// did not reach its destination.
+    #[serde(with = "humantime_serde")]
+    pub topology_resolution_timeout: Duration,
+
+    /// Specifies whether the client should not refresh the network topology after obtaining
+    /// the first valid instance.
+    /// Supersedes `topology_refresh_rate_ms`.
+    pub disable_refreshing: bool,
+}
+
+impl Default for Topology {
+    fn default() -> Self {
+        Topology {
+            topology_refresh_rate: DEFAULT_TOPOLOGY_REFRESH_RATE,
+            topology_resolution_timeout: DEFAULT_TOPOLOGY_RESOLUTION_TIMEOUT,
+            disable_refreshing: false,
+        }
+    }
+}

gateway/src/config/old_config_v1_1_31.rs

@@ -165,6 +165,9 @@ impl From<ConfigV1_1_31> for Config {
                 message_retrieval_limit: value.debug.message_retrieval_limit,
                 use_legacy_framed_packet_version: value.debug.use_legacy_framed_packet_version,
             },
+            // \/ ADDED
+            topology: Default::default(),
+            // /\ ADDED
         }
     }
 }

gateway/src/node/mod.rs

@@ -7,7 +7,7 @@ use crate::commands::helpers::{
     override_ip_packet_router_config, override_network_requester_config,
     OverrideIpPacketRouterConfig, OverrideNetworkRequesterConfig,
 };
-use crate::config::Config;
+use crate::config::{Config, Topology};
 use crate::error::GatewayError;
 use crate::http::HttpApiBuilder;
 use crate::node::client_handling::active_clients::ActiveClientsStore;
@@ -24,6 +24,10 @@ use anyhow::bail;
 use dashmap::DashMap;
 use futures::channel::{mpsc, oneshot};
 use log::*;
+use nym_client_core::client::topology_control::accessor::TopologyAccessor;
+use nym_client_core::client::topology_control::nym_api_provider::NymApiTopologyProvider;
+use nym_client_core::client::topology_control::TopologyRefresher;
+use nym_client_core::client::topology_control::TopologyRefresherConfig;
 use nym_crypto::asymmetric::{encryption, identity};
 use nym_mixnet_client::forwarder::{MixForwardingSender, PacketForwarder};
 use nym_network_defaults::NymNetworkDetails;
@@ -31,6 +35,7 @@ use nym_network_requester::{LocalGateway, NRServiceProviderBuilder, RequestFilte
 use nym_node::wireguard::types::GatewayClientRegistry;
 use nym_statistics_common::collector::StatisticsSender;
 use nym_task::{TaskClient, TaskManager};
+use nym_topology::provider_trait::TopologyProvider;
 use nym_validator_client::{nyxd, DirectSigningHttpRpcNyxdClient};
 use rand::seq::SliceRandom;
 use rand::thread_rng;
@@ -38,6 +43,7 @@ use std::error::Error;
 use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::sync::Arc;
+use url::Url;
 
 pub(crate) mod client_handling;
 pub(crate) mod helpers;
@@ -425,6 +431,47 @@ impl<St> Gateway<St> {
         .map_err(Into::into)
     }
 
+    fn setup_topology_provider(nym_api_urls: Vec<Url>) -> Box<dyn TopologyProvider + Send + Sync> {
+        // if no custom provider was ... provided ..., create one using nym-api
+        Box::new(NymApiTopologyProvider::new(
+            nym_api_urls,
+            env!("CARGO_PKG_VERSION").to_string(),
+        ))
+    }
+
+    // future responsible for periodically polling directory server and updating
+    // the current global view of topology
+    async fn start_topology_refresher(
+        topology_provider: Box<dyn TopologyProvider + Send + Sync>,
+        topology_config: Topology,
+        topology_accessor: TopologyAccessor,
+        mut shutdown: TaskClient,
+    ) {
+        let topology_refresher_config =
+            TopologyRefresherConfig::new(topology_config.topology_refresh_rate);
+
+        let mut topology_refresher = TopologyRefresher::new(
+            topology_refresher_config,
+            topology_accessor,
+            topology_provider,
+        );
+        // before returning, block entire runtime to refresh the current network view so that any
+        // components depending on topology would see a non-empty view
+        info!("Obtaining initial network topology");
+        topology_refresher.try_refresh().await;
+
+        if topology_config.disable_refreshing {
+            // if we're not spawning the refresher, don't cause shutdown immediately
+            info!("The topology refesher is not going to be started");
+            shutdown.mark_as_success();
+        } else {
+            // don't spawn the refresher if we don't want to be refreshing the topology.
+            // only use the initial values obtained
+            info!("Starting topology refresher...");
+            topology_refresher.start_with_shutdown(shutdown);
+        }
+    }
+
     async fn check_if_bonded(&self) -> Result<bool, GatewayError> {
         // TODO: if anything, this should be getting data directly from the contract
         // as opposed to the validator API
@@ -459,6 +506,18 @@ impl<St> Gateway<St> {
             CoconutVerifier::new(nyxd_client).await
         };
 
+        let topology_provider = Self::setup_topology_provider(self.config.get_nym_api_endpoints());
+
+        let shared_topology_access = TopologyAccessor::new();
+
+        Self::start_topology_refresher(
+            topology_provider,
+            self.config.topology,
+            shared_topology_access.clone(),
+            shutdown.subscribe(),
+        )
+        .await;
+
         let mix_forwarding_channel =
             self.start_packet_forwarder(shutdown.subscribe().named("PacketForwarder"));
 

mixnode/Cargo.toml

@@ -59,6 +59,7 @@ nym-pemstore = { path = "../common/pemstore", version = "0.3.0" }
 nym-task = { path = "../common/task" }
 nym-types = { path = "../common/types" }
 nym-topology = { path = "../common/topology" }
+nym-client-core = { path = "../common/client-core/" }
 nym-validator-client = { path = "../common/client-libs/validator-client" }
 nym-bin-common = { path = "../common/bin-common", features = ["output_format"] }
 cpu-cycles = { path = "../cpu-cycles", optional = true }

mixnode/src/config/mod.rs

@@ -47,6 +47,9 @@ const DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF: Duration = Duration::from_milli
 const DEFAULT_INITIAL_CONNECTION_TIMEOUT: Duration = Duration::from_millis(1_500);
 const DEFAULT_MAXIMUM_CONNECTION_BUFFER_SIZE: usize = 2000;
 
+const DEFAULT_TOPOLOGY_REFRESH_RATE: Duration = Duration::from_secs(5 * 60); // every 5min
+const DEFAULT_TOPOLOGY_RESOLUTION_TIMEOUT: Duration = Duration::from_millis(5_000);
+
 /// Derive default path to mixnodes's config directory.
 /// It should get resolved to `$HOME/.nym/mixnodes/<id>/config`
 pub fn default_config_directory<P: AsRef<Path>>(id: P) -> PathBuf {
@@ -107,6 +110,9 @@ pub struct Config {
 
     #[serde(default)]
     pub debug: Debug,
+
+    #[serde(default)]
+    pub topology: Topology,
 }
 
 impl NymConfigTemplate for Config {
@@ -132,6 +138,7 @@ impl Config {
             verloc: Default::default(),
             logging: Default::default(),
             debug: Default::default(),
+            topology: Default::default(),
         }
     }
 
@@ -339,3 +346,33 @@ impl Default for Debug {
         }
     }
 }
+
+#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Serialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct Topology {
+    /// The uniform delay every which clients are querying the directory server
+    /// to try to obtain a compatible network topology to send sphinx packets through.
+    #[serde(with = "humantime_serde")]
+    pub topology_refresh_rate: Duration,
+
+    /// During topology refresh, test packets are sent through every single possible network
+    /// path. This timeout determines waiting period until it is decided that the packet
+    /// did not reach its destination.
+    #[serde(with = "humantime_serde")]
+    pub topology_resolution_timeout: Duration,
+
+    /// Specifies whether the client should not refresh the network topology after obtaining
+    /// the first valid instance.
+    /// Supersedes `topology_refresh_rate_ms`.
+    pub disable_refreshing: bool,
+}
+
+impl Default for Topology {
+    fn default() -> Self {
+        Topology {
+            topology_refresh_rate: DEFAULT_TOPOLOGY_REFRESH_RATE,
+            topology_resolution_timeout: DEFAULT_TOPOLOGY_RESOLUTION_TIMEOUT,
+            disable_refreshing: false,
+        }
+    }
+}

mixnode/src/config/old_config_v1_1_32.rs

@@ -109,6 +109,9 @@ impl From<ConfigV1_1_32> for Config {
             verloc: value.verloc.into(),
             logging: value.logging,
             debug: value.debug.into(),
+            // \/ ADDED
+            topology: Default::default(),
+            // /\ ADDED
         }
     }
 }

mixnode/src/node/mod.rs

@@ -1,7 +1,7 @@
 // Copyright 2020-2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: GPL-3.0-only
 
-use crate::config::Config;
+use crate::config::{Config, Topology};
 use crate::error::MixnodeError;
 use crate::node::helpers::{load_identity_keys, load_sphinx_keys};
 use crate::node::http::legacy::verloc::VerlocState;
@@ -15,14 +15,20 @@ use crate::node::packet_delayforwarder::{DelayForwarder, PacketDelayForwardSende
 use log::{error, info, warn};
 use nym_bin_common::output_format::OutputFormat;
 use nym_bin_common::version_checker::parse_version;
+use nym_client_core::client::topology_control::accessor::TopologyAccessor;
+use nym_client_core::client::topology_control::nym_api_provider::NymApiTopologyProvider;
+use nym_client_core::client::topology_control::TopologyRefresher;
+use nym_client_core::client::topology_control::TopologyRefresherConfig;
 use nym_crypto::asymmetric::{encryption, identity};
 use nym_mixnode_common::verloc::{self, AtomicVerlocResult, VerlocMeasurer};
 use nym_task::{TaskClient, TaskManager};
+use nym_topology::provider_trait::TopologyProvider;
 use rand::seq::SliceRandom;
 use rand::thread_rng;
 use std::net::SocketAddr;
 use std::process;
 use std::sync::Arc;
+use url::Url;
 
 pub(crate) mod helpers;
 mod http;
@@ -185,6 +191,47 @@ impl MixNode {
         atomic_verloc_results
     }
 
+    fn setup_topology_provider(nym_api_urls: Vec<Url>) -> Box<dyn TopologyProvider + Send + Sync> {
+        // if no custom provider was ... provided ..., create one using nym-api
+        Box::new(NymApiTopologyProvider::new(
+            nym_api_urls,
+            env!("CARGO_PKG_VERSION").to_string(),
+        ))
+    }
+
+    // future responsible for periodically polling directory server and updating
+    // the current global view of topology
+    async fn start_topology_refresher(
+        topology_provider: Box<dyn TopologyProvider + Send + Sync>,
+        topology_config: Topology,
+        topology_accessor: TopologyAccessor,
+        mut shutdown: TaskClient,
+    ) {
+        let topology_refresher_config =
+            TopologyRefresherConfig::new(topology_config.topology_refresh_rate);
+
+        let mut topology_refresher = TopologyRefresher::new(
+            topology_refresher_config,
+            topology_accessor,
+            topology_provider,
+        );
+        // before returning, block entire runtime to refresh the current network view so that any
+        // components depending on topology would see a non-empty view
+        info!("Obtaining initial network topology");
+        topology_refresher.try_refresh().await;
+
+        if topology_config.disable_refreshing {
+            // if we're not spawning the refresher, don't cause shutdown immediately
+            info!("The topology refesher is not going to be started");
+            shutdown.mark_as_success();
+        } else {
+            // don't spawn the refresher if we don't want to be refreshing the topology.
+            // only use the initial values obtained
+            info!("Starting topology refresher...");
+            topology_refresher.start_with_shutdown(shutdown);
+        }
+    }
+
     fn random_api_client(&self) -> nym_validator_client::NymApiClient {
         let endpoints = self.config.get_nym_api_endpoints();
         let nym_api = endpoints
@@ -231,6 +278,16 @@ impl MixNode {
 
         let (node_stats_pointer, node_stats_update_sender) = self
             .start_node_stats_controller(shutdown.subscribe().named("node_statistics::Controller"));
+
+        let topology_provider = Self::setup_topology_provider(self.config.get_nym_api_endpoints());
+        let shared_topology_access = TopologyAccessor::new();
+        Self::start_topology_refresher(
+            topology_provider,
+            self.config.topology,
+            shared_topology_access.clone(),
+            shutdown.subscribe().named("TopologyRefresher"),
+        )
+        .await;
         let delay_forwarding_channel = self.start_packet_delay_forwarder(
             node_stats_update_sender.clone(),
             shutdown.subscribe().named("DelayForwarder"),