From 0a826b81461e3beea88433b79ea77ee964ee11bc Mon Sep 17 00:00:00 2001 From: Simon Wicky Date: Tue, 30 Jan 2024 10:16:44 +0100 Subject: [PATCH] add topology access to nodes --- Cargo.lock | 3 + .../src/client/topology_control/accessor.rs | 10 +++ .../src/client/topology_control/mod.rs | 4 +- .../topology_control/nym_api_provider.rs | 4 +- gateway/Cargo.toml | 2 + gateway/src/config/mod.rs | 37 +++++++++++ gateway/src/config/old_config_v1_1_31.rs | 3 + gateway/src/node/mod.rs | 61 ++++++++++++++++++- mixnode/Cargo.toml | 1 + mixnode/src/config/mod.rs | 37 +++++++++++ mixnode/src/config/old_config_v1_1_32.rs | 3 + mixnode/src/node/mod.rs | 59 +++++++++++++++++- 12 files changed, 218 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c221eb6583..75a4c2ed9a 100644 --- a/Cargo.lock +++ b/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", diff --git a/common/client-core/src/client/topology_control/accessor.rs b/common/client-core/src/client/topology_control/accessor.rs index fcb272c934..f4b50be059 100644 --- a/common/client-core/src/client/topology_control/accessor.rs +++ b/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>> for TopologyReadPermit<'a> { diff --git a/common/client-core/src/client/topology_control/mod.rs b/common/client-core/src/client/topology_control/mod.rs index ae50128942..fa54a0ca7f 100644 --- a/common/client-core/src/client/topology_control/mod.rs +++ b/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; diff --git a/common/client-core/src/client/topology_control/nym_api_provider.rs b/common/client-core/src/client/topology_control/nym_api_provider.rs index 90fc23cc8c..829a239210 100644 --- a/common/client-core/src/client/topology_control/nym_api_provider.rs +++ b/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, @@ -18,7 +18,7 @@ pub(crate) struct NymApiTopologyProvider { } impl NymApiTopologyProvider { - pub(crate) fn new(mut nym_api_urls: Vec, client_version: String) -> Self { + pub fn new(mut nym_api_urls: Vec, client_version: String) -> Self { nym_api_urls.shuffle(&mut thread_rng()); NymApiTopologyProvider { diff --git a/gateway/Cargo.toml b/gateway/Cargo.toml index a845ba908b..4743d52add 100644 --- a/gateway/Cargo.toml +++ b/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 } diff --git a/gateway/src/config/mod.rs b/gateway/src/config/mod.rs index b872a1dffd..8d5a6ee5d9 100644 --- a/gateway/src/config/mod.rs +++ b/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, + } + } +} diff --git a/gateway/src/config/old_config_v1_1_31.rs b/gateway/src/config/old_config_v1_1_31.rs index 18841a4993..0b86b21618 100644 --- a/gateway/src/config/old_config_v1_1_31.rs +++ b/gateway/src/config/old_config_v1_1_31.rs @@ -165,6 +165,9 @@ impl From 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 } } } diff --git a/gateway/src/node/mod.rs b/gateway/src/node/mod.rs index 7c147670ac..21a98a92ef 100644 --- a/gateway/src/node/mod.rs +++ b/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 Gateway { .map_err(Into::into) } + fn setup_topology_provider(nym_api_urls: Vec) -> Box { + // 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, + 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 { // TODO: if anything, this should be getting data directly from the contract // as opposed to the validator API @@ -459,6 +506,18 @@ impl Gateway { 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")); diff --git a/mixnode/Cargo.toml b/mixnode/Cargo.toml index 9ad7c5081f..7597e19177 100644 --- a/mixnode/Cargo.toml +++ b/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 } diff --git a/mixnode/src/config/mod.rs b/mixnode/src/config/mod.rs index a1ab25a752..49b4e3ddfc 100644 --- a/mixnode/src/config/mod.rs +++ b/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//config` pub fn default_config_directory>(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, + } + } +} diff --git a/mixnode/src/config/old_config_v1_1_32.rs b/mixnode/src/config/old_config_v1_1_32.rs index a17cdeadcd..0170aa766d 100644 --- a/mixnode/src/config/old_config_v1_1_32.rs +++ b/mixnode/src/config/old_config_v1_1_32.rs @@ -109,6 +109,9 @@ impl From for Config { verloc: value.verloc.into(), logging: value.logging, debug: value.debug.into(), + // \/ ADDED + topology: Default::default(), + // /\ ADDED } } } diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index 1667b108fd..ee9567f7e9 100644 --- a/mixnode/src/node/mod.rs +++ b/mixnode/src/node/mod.rs @@ -1,7 +1,7 @@ // Copyright 2020-2023 - Nym Technologies SA // 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) -> Box { + // 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, + 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"),