GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: ability to inject custom topology into clients (#3055)

Browse Source 
* ability to specify custom TopologyProvider in TopologyRefresher

* topology provider builder method for base client

* ability to take manual control over topology

* wasm fixes

* added topology injection to nym-sdk API

* added examples to nym-sdk and exposed additional helper methods
This commit is contained in:
Jędrzej Stuczyński
2023-03-06 13:50:06 +00:00
committed by GitHub
parent d4f0b4772b
commit 2ff6bfbdd8
29 changed files with 764 additions and 372 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+2
View File
@@ -3776,6 +3776,7 @@ dependencies = [
"nym-network-defaults",
"nym-sphinx",
"nym-task",
"nym-topology",
"pretty_env_logger",
"rand 0.7.3",
"tap",
@@ -3983,6 +3984,7 @@ dependencies = [
name = "nym-topology"
version = "0.1.0"
dependencies = [
"async-trait",
"bs58",
"log",
"nym-bin-common",
Cargo.toml
+1 -1
View File
@@ -105,7 +105,7 @@ edition = "2021"
license = "Apache-2.0"
[workspace.dependencies]
async-trait = "0.1.63"
async-trait = "0.1.64"
cfg-if = "1.0.0"
dotenv = "0.15.0"
lazy_static = "1.4.0"
clients/client-core/Cargo.toml
+1 -1
View File
@@ -8,7 +8,7 @@ rust-version = "1.66"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
async-trait = { version = "0.1.58" }
async-trait = { workspace = true }
dirs = "4.0"
dashmap = "5.4.0"
futures = "0.3"
clients/client-core/src/client/base_client/mod.rs
+40 -10
View File
@@ -15,6 +15,7 @@ use crate::client::replies::reply_controller::{ReplyControllerReceiver, ReplyCon
use crate::client::replies::reply_storage::{
CombinedReplyStorage, PersistentReplyStorage, ReplyStorageBackend, SentReplyKeys,
};
use crate::client::topology_control::nym_api_provider::NymApiTopologyProvider;
use crate::client::topology_control::{
TopologyAccessor, TopologyRefresher, TopologyRefresherConfig,
};
@@ -37,10 +38,12 @@ use nym_sphinx::addressing::nodes::NodeIdentity;
use nym_sphinx::receiver::ReconstructedMessage;
use nym_task::connections::{ConnectionCommandReceiver, ConnectionCommandSender, LaneQueueLengths};
use nym_task::{TaskClient, TaskManager};
use nym_topology::provider_trait::TopologyProvider;
use std::sync::Arc;
use std::time::Duration;
use tap::TapFallible;
use url::Url;
#[cfg(not(target_arch = "wasm32"))]
use validator_client::nyxd::CosmWasmClient;
@@ -90,6 +93,7 @@ impl ClientOutput {
pub struct ClientState {
pub shared_lane_queue_lengths: LaneQueueLengths,
pub reply_controller_sender: ReplyControllerSender,
pub topology_accessor: TopologyAccessor,
}
pub enum ClientInputStatus {
@@ -154,6 +158,7 @@ pub struct BaseClientBuilder<'a, B, C: Clone> {
nym_api_endpoints: Vec<Url>,
reply_storage_backend: B,
custom_topology_provider: Option<Box<dyn TopologyProvider>>,
bandwidth_controller: Option<BandwidthController<C>>,
key_manager: KeyManager,
}
@@ -177,6 +182,7 @@ where
bandwidth_controller,
reply_storage_backend,
key_manager,
custom_topology_provider: None,
}
}
@@ -195,11 +201,17 @@ where
disabled_credentials: credentials_toggle.is_disabled(),
nym_api_endpoints,
reply_storage_backend,
custom_topology_provider: None,
bandwidth_controller,
key_manager,
}
}
pub fn with_topology_provider(mut self, provider: Box<dyn TopologyProvider>) -> Self {
self.custom_topology_provider = Some(provider);
self
}
pub fn as_mix_recipient(&self) -> Recipient {
Recipient::new(
*self.key_manager.identity_keypair().public_key(),
@@ -341,25 +353,38 @@ where
Ok(gateway_client)
}
fn setup_topology_provider(
custom_provider: Option<Box<dyn TopologyProvider>>,
nym_api_urls: Vec<Url>,
) -> Box<dyn TopologyProvider> {
// if no custom provider was ... provided ..., create one using nym-api
custom_provider.unwrap_or_else(|| {
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(
nym_api_urls: Vec<Url>,
topology_provider: Box<dyn TopologyProvider>,
refresh_rate: Duration,
topology_accessor: TopologyAccessor,
shutdown: TaskClient,
) -> Result<(), ClientCoreError> {
let topology_refresher_config = TopologyRefresherConfig::new(
nym_api_urls,
refresh_rate,
env!("CARGO_PKG_VERSION").to_string(),
let topology_refresher_config = TopologyRefresherConfig::new(refresh_rate);
let mut topology_refresher = TopologyRefresher::new(
topology_refresher_config,
topology_accessor,
topology_provider,
);
let mut topology_refresher =
TopologyRefresher::new(topology_refresher_config, topology_accessor);
// 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.refresh().await;
topology_refresher.try_refresh().await;
if let Err(err) = topology_refresher.ensure_topology_is_routable().await {
log::error!(
@@ -468,8 +493,12 @@ where
)
.await?;
let topology_provider = Self::setup_topology_provider(
self.custom_topology_provider.take(),
self.nym_api_endpoints,
);
Self::start_topology_refresher(
self.nym_api_endpoints.clone(),
topology_provider,
self.debug_config.topology_refresh_rate,
shared_topology_accessor.clone(),
task_manager.subscribe(),
@@ -530,7 +559,7 @@ where
self.debug_config,
self.key_manager.ack_key(),
self_address,
shared_topology_accessor,
shared_topology_accessor.clone(),
sphinx_message_sender,
task_manager.subscribe(),
);
@@ -554,6 +583,7 @@ where
client_state: ClientState {
shared_lane_queue_lengths,
reply_controller_sender,
topology_accessor: shared_topology_accessor,
},
task_manager,
})
clients/client-core/src/client/topology_control.rs
-336
View File
@@ -1,336 +0,0 @@
// Copyright 2021-2022 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::spawn_future;
use futures::StreamExt;
use log::*;
use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::params::DEFAULT_NUM_MIX_HOPS;
use nym_topology::{nym_topology_from_detailed, NymTopology, NymTopologyError};
use rand::seq::SliceRandom;
use rand::thread_rng;
use std::ops::Deref;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{RwLock, RwLockReadGuard};
use url::Url;
// I'm extremely curious why compiler NEVER complained about lack of Debug here before
#[derive(Debug)]
pub struct TopologyAccessorInner(Option<NymTopology>);
impl AsRef<Option<NymTopology>> for TopologyAccessorInner {
fn as_ref(&self) -> &Option<NymTopology> {
&self.0
}
}
impl TopologyAccessorInner {
fn new() -> Self {
TopologyAccessorInner(None)
}
fn update(&mut self, new: Option<NymTopology>) {
self.0 = new;
}
}
pub struct TopologyReadPermit<'a> {
permit: RwLockReadGuard<'a, TopologyAccessorInner>,
}
impl<'a> Deref for TopologyReadPermit<'a> {
type Target = TopologyAccessorInner;
fn deref(&self) -> &Self::Target {
&self.permit
}
}
impl<'a> TopologyReadPermit<'a> {
/// Using provided topology read permit, tries to get an immutable reference to the underlying
/// topology. For obvious reasons the lifetime of the topology reference is bound to the permit.
pub(super) fn try_get_valid_topology_ref(
&'a self,
ack_recipient: &Recipient,
packet_recipient: Option<&Recipient>,
) -> 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()
.as_ref()
.ok_or(NymTopologyError::EmptyNetworkTopology)?;
// 2. does it have any mixnode at all?
// 3. does it have any gateways at all?
// 4. does it have a mixnode on each layer?
topology.ensure_can_construct_path_through(DEFAULT_NUM_MIX_HOPS)?;
// 5. does it contain OUR gateway (so that we could create an ack packet)?
if !topology.gateway_exists(ack_recipient.gateway()) {
return Err(NymTopologyError::NonExistentGatewayError {
identity_key: ack_recipient.gateway().to_base58_string(),
});
}
// 6. for our target recipient, does it contain THEIR gateway (so that we could create
if let Some(recipient) = packet_recipient {
if !topology.gateway_exists(recipient.gateway()) {
return Err(NymTopologyError::NonExistentGatewayError {
identity_key: recipient.gateway().to_base58_string(),
});
}
}
Ok(topology)
}
}
impl<'a> From<RwLockReadGuard<'a, TopologyAccessorInner>> for TopologyReadPermit<'a> {
fn from(read_permit: RwLockReadGuard<'a, TopologyAccessorInner>) -> Self {
TopologyReadPermit {
permit: read_permit,
}
}
}
#[derive(Clone, Debug)]
pub struct TopologyAccessor {
// `RwLock` *seems to* be the better approach for this as write access is only requested every
// few seconds, while reads are needed every single packet generated.
// However, proper benchmarks will be needed to determine if `RwLock` is indeed a better
// approach than a `Mutex`
inner: Arc<RwLock<TopologyAccessorInner>>,
}
impl TopologyAccessor {
pub fn new() -> Self {
TopologyAccessor {
inner: Arc::new(RwLock::new(TopologyAccessorInner::new())),
}
}
pub async fn get_read_permit(&self) -> TopologyReadPermit<'_> {
self.inner.read().await.into()
}
async fn update_global_topology(&self, new_topology: Option<NymTopology>) {
self.inner.write().await.update(new_topology);
}
// only used by the client at startup to get a slightly more reasonable error message
// (currently displays as unused because health checker is disabled due to required changes)
pub async fn ensure_is_routable(&self) -> Result<(), NymTopologyError> {
match &self.inner.read().await.0 {
None => Err(NymTopologyError::EmptyNetworkTopology),
Some(ref topology) => topology.ensure_can_construct_path_through(DEFAULT_NUM_MIX_HOPS),
}
}
}
impl Default for TopologyAccessor {
fn default() -> Self {
TopologyAccessor::new()
}
}
pub struct TopologyRefresherConfig {
nym_api_urls: Vec<Url>,
refresh_rate: Duration,
client_version: String,
}
impl TopologyRefresherConfig {
pub fn new(nym_api_urls: Vec<Url>, refresh_rate: Duration, client_version: String) -> Self {
TopologyRefresherConfig {
nym_api_urls,
refresh_rate,
client_version,
}
}
}
pub struct TopologyRefresher {
validator_client: validator_client::client::NymApiClient,
client_version: String,
nym_api_urls: Vec<Url>,
topology_accessor: TopologyAccessor,
refresh_rate: Duration,
currently_used_api: usize,
was_latest_valid: bool,
}
impl TopologyRefresher {
pub fn new(mut cfg: TopologyRefresherConfig, topology_accessor: TopologyAccessor) -> Self {
cfg.nym_api_urls.shuffle(&mut thread_rng());
TopologyRefresher {
validator_client: validator_client::client::NymApiClient::new(
cfg.nym_api_urls[0].clone(),
),
client_version: cfg.client_version,
nym_api_urls: cfg.nym_api_urls,
topology_accessor,
refresh_rate: cfg.refresh_rate,
currently_used_api: 0,
was_latest_valid: true,
}
}
fn use_next_nym_api(&mut self) {
if self.nym_api_urls.len() == 1 {
warn!("There's only a single nym API available - it won't be possible to use a different one");
return;
}
self.currently_used_api = (self.currently_used_api + 1) % self.nym_api_urls.len();
self.validator_client
.change_nym_api(self.nym_api_urls[self.currently_used_api].clone())
}
/// Verifies whether nodes a reasonably distributed among all mix layers.
///
/// In ideal world we would have 33% nodes on layer 1, 33% on layer 2 and 33% on layer 3.
/// However, this is a rather unrealistic expectation, instead we check whether there exists
/// a layer with more than 66% of nodes or with fewer than 15% and if so, we trigger a failure.
///
/// # Arguments
///
/// * `topology`: active topology constructed from validator api data
fn check_layer_distribution(&self, active_topology: &NymTopology) -> bool {
let mixes = active_topology.mixes();
let mixnodes_count = active_topology.num_mixnodes();
if active_topology.gateways().is_empty() {
return false;
}
// trivial check to see if have at least a single node on each layer (regardless of active set size)
if mixes.get(&1).is_none() || mixes.get(&2).is_none() || mixes.get(&3).is_none() {
return false;
}
let upper_bound = (mixnodes_count as f32 * 0.66) as usize;
let lower_bound = (mixnodes_count as f32 * 0.15) as usize;
let layer1 = mixes.get(&1).unwrap().len();
let layer2 = mixes.get(&2).unwrap().len();
let layer3 = mixes.get(&3).unwrap().len();
if layer1 < lower_bound || layer1 > upper_bound {
warn!(
"nodes: {}, layer1: {}, layer2: {}, layer3: {}",
mixnodes_count, layer1, layer2, layer3
);
return false;
}
if layer2 < lower_bound || layer2 > upper_bound {
warn!(
"nodes: {}, layer1: {}, layer2: {}, layer3: {}",
mixnodes_count, layer1, layer2, layer3
);
return false;
}
if layer3 < lower_bound || layer3 > upper_bound {
warn!(
"nodes: {}, layer1: {}, layer2: {}, layer3: {}",
mixnodes_count, layer1, layer2, layer3
);
return false;
}
true
}
async fn get_current_compatible_topology(&self) -> Option<NymTopology> {
// TODO: optimization for the future:
// only refresh mixnodes on timer and refresh gateways only when
// we have to send to a new, unknown, gateway
let mixnodes = match self.validator_client.get_cached_active_mixnodes().await {
Err(err) => {
error!("failed to get network mixnodes - {err}");
return None;
}
Ok(mixes) => mixes,
};
let gateways = match self.validator_client.get_cached_gateways().await {
Err(err) => {
error!("failed to get network gateways - {err}");
return None;
}
Ok(gateways) => gateways,
};
let topology = nym_topology_from_detailed(mixnodes, gateways)
.filter_system_version(&self.client_version);
if !self.check_layer_distribution(&topology) {
warn!("The current filtered active topology has extremely skewed layer distribution. It cannot be used.");
None
} else {
Some(topology)
}
}
pub async fn refresh(&mut self) {
trace!("Refreshing the topology");
let new_topology = self.get_current_compatible_topology().await;
if new_topology.is_none() {
self.use_next_nym_api();
}
if new_topology.is_none() && self.was_latest_valid {
// if we failed to grab this topology, but the one before it was alright, let's assume
// validator had a tiny hiccup and use the old data
warn!("we're going to keep on using the old topology for this iteration");
self.was_latest_valid = false;
return;
} else if new_topology.is_some() {
self.was_latest_valid = true;
}
self.topology_accessor
.update_global_topology(new_topology)
.await;
}
pub async fn ensure_topology_is_routable(&self) -> Result<(), NymTopologyError> {
self.topology_accessor.ensure_is_routable().await
}
pub fn start_with_shutdown(mut self, mut shutdown: nym_task::TaskClient) {
spawn_future(async move {
debug!("Started TopologyRefresher with graceful shutdown support");
#[cfg(not(target_arch = "wasm32"))]
let mut interval = tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(
self.refresh_rate,
));
#[cfg(target_arch = "wasm32")]
let mut interval =
gloo_timers::future::IntervalStream::new(self.refresh_rate.as_millis() as u32);
while !shutdown.is_shutdown() {
tokio::select! {
_ = interval.next() => {
self.refresh().await;
},
_ = shutdown.recv() => {
log::trace!("TopologyRefresher: Received shutdown");
},
}
}
shutdown.recv_timeout().await;
log::debug!("TopologyRefresher: Exiting");
})
}
}
clients/client-core/src/client/topology_control/accessor.rs
+154
View File
@@ -0,0 +1,154 @@
// Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::params::DEFAULT_NUM_MIX_HOPS;
use nym_topology::{NymTopology, NymTopologyError};
use std::ops::Deref;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use tokio::sync::{Notify, RwLock, RwLockReadGuard};
#[derive(Debug)]
pub struct TopologyAccessorInner {
controlled_manually: AtomicBool,
released_manual_control: Notify,
// `RwLock` *seems to* be the better approach for this as write access is only requested every
// few seconds, while reads are needed every single packet generated.
// However, proper benchmarks will be needed to determine if `RwLock` is indeed a better
// approach than a `Mutex`
topology: RwLock<Option<NymTopology>>,
}
impl TopologyAccessorInner {
fn new() -> Self {
TopologyAccessorInner {
controlled_manually: AtomicBool::new(false),
released_manual_control: Notify::new(),
topology: RwLock::new(None),
}
}
async fn update(&self, new: Option<NymTopology>) {
*self.topology.write().await = new;
}
}
pub struct TopologyReadPermit<'a> {
permit: RwLockReadGuard<'a, Option<NymTopology>>,
}
impl<'a> Deref for TopologyReadPermit<'a> {
type Target = Option<NymTopology>;
fn deref(&self) -> &Self::Target {
&self.permit
}
}
impl<'a> TopologyReadPermit<'a> {
/// Using provided topology read permit, tries to get an immutable reference to the underlying
/// topology. For obvious reasons the lifetime of the topology reference is bound to the permit.
pub(crate) fn try_get_valid_topology_ref(
&'a self,
ack_recipient: &Recipient,
packet_recipient: Option<&Recipient>,
) -> 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)?;
// 2. does it have any mixnode at all?
// 3. does it have any gateways at all?
// 4. does it have a mixnode on each layer?
topology.ensure_can_construct_path_through(DEFAULT_NUM_MIX_HOPS)?;
// 5. does it contain OUR gateway (so that we could create an ack packet)?
if !topology.gateway_exists(ack_recipient.gateway()) {
return Err(NymTopologyError::NonExistentGatewayError {
identity_key: ack_recipient.gateway().to_base58_string(),
});
}
// 6. for our target recipient, does it contain THEIR gateway (so that we could create
if let Some(recipient) = packet_recipient {
if !topology.gateway_exists(recipient.gateway()) {
return Err(NymTopologyError::NonExistentGatewayError {
identity_key: recipient.gateway().to_base58_string(),
});
}
}
Ok(topology)
}
}
impl<'a> From<RwLockReadGuard<'a, Option<NymTopology>>> for TopologyReadPermit<'a> {
fn from(read_permit: RwLockReadGuard<'a, Option<NymTopology>>) -> Self {
TopologyReadPermit {
permit: read_permit,
}
}
}
#[derive(Clone, Debug)]
pub struct TopologyAccessor {
inner: Arc<TopologyAccessorInner>,
}
impl TopologyAccessor {
pub fn new() -> Self {
TopologyAccessor {
inner: Arc::new(TopologyAccessorInner::new()),
}
}
pub fn controlled_manually(&self) -> bool {
self.inner.controlled_manually.load(Ordering::SeqCst)
}
pub async fn get_read_permit(&self) -> TopologyReadPermit<'_> {
self.inner.topology.read().await.into()
}
pub(crate) async fn update_global_topology(&self, new_topology: Option<NymTopology>) {
self.inner.update(new_topology).await;
}
pub(crate) async fn wait_for_released_manual_control(&self) {
self.inner.released_manual_control.notified().await
}
pub async fn current_topology(&self) -> Option<NymTopology> {
self.inner.topology.read().await.clone()
}
pub async fn manually_change_topology(&self, new_topology: NymTopology) {
self.inner.controlled_manually.store(true, Ordering::SeqCst);
self.inner.update(Some(new_topology)).await;
}
pub fn release_manual_control(&self) {
self.inner
.controlled_manually
.store(false, Ordering::SeqCst);
self.inner.released_manual_control.notify_waiters();
}
// only used by the client at startup to get a slightly more reasonable error message
// (currently displays as unused because health checker is disabled due to required changes)
pub async fn ensure_is_routable(&self) -> Result<(), NymTopologyError> {
match self.inner.topology.read().await.deref() {
None => Err(NymTopologyError::EmptyNetworkTopology),
Some(ref topology) => topology.ensure_can_construct_path_through(DEFAULT_NUM_MIX_HOPS),
}
}
}
impl Default for TopologyAccessor {
fn default() -> Self {
TopologyAccessor::new()
}
}
clients/client-core/src/client/topology_control/mod.rs
+115
View File
@@ -0,0 +1,115 @@
// Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::spawn_future;
pub(crate) use accessor::{TopologyAccessor, TopologyReadPermit};
use futures::StreamExt;
use log::*;
use nym_topology::provider_trait::TopologyProvider;
use nym_topology::NymTopologyError;
use std::time::Duration;
mod accessor;
pub(crate) mod nym_api_provider;
// TODO: move it to config later
const MAX_FAILURE_COUNT: usize = 10;
pub struct TopologyRefresherConfig {
refresh_rate: Duration,
}
impl TopologyRefresherConfig {
pub fn new(refresh_rate: Duration) -> Self {
TopologyRefresherConfig { refresh_rate }
}
}
pub struct TopologyRefresher {
topology_provider: Box<dyn TopologyProvider>,
topology_accessor: TopologyAccessor,
refresh_rate: Duration,
consecutive_failure_count: usize,
}
impl TopologyRefresher {
pub fn new(
cfg: TopologyRefresherConfig,
topology_accessor: TopologyAccessor,
topology_provider: Box<dyn TopologyProvider>,
) -> Self {
TopologyRefresher {
topology_provider,
topology_accessor,
refresh_rate: cfg.refresh_rate,
consecutive_failure_count: 0,
}
}
pub fn change_topology_provider(&mut self, provider: Box<dyn TopologyProvider>) {
self.topology_provider = provider;
}
pub async fn try_refresh(&mut self) {
trace!("Refreshing the topology");
if self.topology_accessor.controlled_manually() {
info!("topology is being controlled manually - we're going to wait until the control is released...");
self.topology_accessor
.wait_for_released_manual_control()
.await;
}
let new_topology = self.topology_provider.get_new_topology().await;
if new_topology.is_none() {
warn!("failed to obtain new network topology");
}
if new_topology.is_none() && self.consecutive_failure_count < MAX_FAILURE_COUNT {
// if we failed to grab this topology, but the one before it was alright, let's assume
// validator had a tiny hiccup and use the old data
warn!("we're going to keep on using the old topology for this iteration");
self.consecutive_failure_count += 1;
return;
} else if new_topology.is_some() {
self.consecutive_failure_count = 0;
}
self.topology_accessor
.update_global_topology(new_topology)
.await;
}
pub async fn ensure_topology_is_routable(&self) -> Result<(), NymTopologyError> {
self.topology_accessor.ensure_is_routable().await
}
pub fn start_with_shutdown(mut self, mut shutdown: nym_task::TaskClient) {
spawn_future(async move {
debug!("Started TopologyRefresher with graceful shutdown support");
#[cfg(not(target_arch = "wasm32"))]
let mut interval = tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(
self.refresh_rate,
));
#[cfg(target_arch = "wasm32")]
let mut interval =
gloo_timers::future::IntervalStream::new(self.refresh_rate.as_millis() as u32);
while !shutdown.is_shutdown() {
tokio::select! {
_ = interval.next() => {
self.try_refresh().await;
},
_ = shutdown.recv() => {
log::trace!("TopologyRefresher: Received shutdown");
},
}
}
shutdown.recv_timeout().await;
log::debug!("TopologyRefresher: Exiting");
})
}
}
clients/client-core/src/client/topology_control/nym_api_provider.rs
+106
View File
@@ -0,0 +1,106 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use async_trait::async_trait;
use log::{error, warn};
use nym_topology::provider_trait::TopologyProvider;
use nym_topology::{nym_topology_from_detailed, NymTopology, NymTopologyError};
use rand::prelude::SliceRandom;
use rand::thread_rng;
use url::Url;
pub(crate) struct NymApiTopologyProvider {
validator_client: validator_client::client::NymApiClient,
nym_api_urls: Vec<Url>,
client_version: String,
currently_used_api: usize,
}
impl NymApiTopologyProvider {
pub(crate) fn new(mut nym_api_urls: Vec<Url>, client_version: String) -> Self {
nym_api_urls.shuffle(&mut thread_rng());
NymApiTopologyProvider {
validator_client: validator_client::client::NymApiClient::new(nym_api_urls[0].clone()),
nym_api_urls,
client_version,
currently_used_api: 0,
}
}
fn use_next_nym_api(&mut self) {
if self.nym_api_urls.len() == 1 {
warn!("There's only a single nym API available - it won't be possible to use a different one");
return;
}
self.currently_used_api = (self.currently_used_api + 1) % self.nym_api_urls.len();
self.validator_client
.change_nym_api(self.nym_api_urls[self.currently_used_api].clone())
}
/// Verifies whether nodes a reasonably distributed among all mix layers.
///
/// In ideal world we would have 33% nodes on layer 1, 33% on layer 2 and 33% on layer 3.
/// However, this is a rather unrealistic expectation, instead we check whether there exists
/// a layer with more than 66% of nodes or with fewer than 15% and if so, we trigger a failure.
///
/// # Arguments
///
/// * `topology`: active topology constructed from validator api data
fn check_layer_distribution(
&self,
active_topology: &NymTopology,
) -> Result<(), NymTopologyError> {
let lower_threshold = 0.15;
let upper_threshold = 0.66;
active_topology.ensure_even_layer_distribution(lower_threshold, upper_threshold)
}
async fn get_current_compatible_topology(&mut self) -> Option<NymTopology> {
let mixnodes = match self.validator_client.get_cached_active_mixnodes().await {
Err(err) => {
error!("failed to get network mixnodes - {err}");
return None;
}
Ok(mixes) => mixes,
};
let gateways = match self.validator_client.get_cached_gateways().await {
Err(err) => {
error!("failed to get network gateways - {err}");
return None;
}
Ok(gateways) => gateways,
};
let topology = nym_topology_from_detailed(mixnodes, gateways)
.filter_system_version(&self.client_version);
if let Err(err) = self.check_layer_distribution(&topology) {
warn!("The current filtered active topology has extremely skewed layer distribution. It cannot be used: {err}");
self.use_next_nym_api();
None
} else {
Some(topology)
}
}
}
// hehe, wasm
#[cfg(not(target_arch = "wasm32"))]
#[async_trait]
impl TopologyProvider for NymApiTopologyProvider {
async fn get_new_topology(&mut self) -> Option<NymTopology> {
self.get_current_compatible_topology().await
}
}
#[cfg(target_arch = "wasm32")]
#[async_trait(?Send)]
impl TopologyProvider for NymApiTopologyProvider {
async fn get_new_topology(&mut self) -> Option<NymTopology> {
self.get_current_compatible_topology().await
}
}
clients/native/src/client/mod.rs
+1
View File
@@ -101,6 +101,7 @@ impl SocketClient {
let ClientState {
shared_lane_queue_lengths,
reply_controller_sender,
..
} = client_state;
let websocket_handler = websocket::HandlerBuilder::new(
clients/socks5/src/client/mod.rs
+1 -1
View File
@@ -123,7 +123,7 @@ impl NymClient {
let ClientState {
shared_lane_queue_lengths,
reply_controller_sender: _,
..
} = client_status;
let authenticator = Authenticator::new(auth_methods, allowed_users);
common/client-libs/gateway-client/Cargo.toml
+1 -1
View File
@@ -14,7 +14,7 @@ log = { workspace = true }
thiserror = "1.0"
url = "2.2"
rand = { version = "0.7.3", features = ["wasm-bindgen"] }
async-trait = { version = "0.1.51" }
async-trait = { workspace = true }
tokio = { version = "1.24.1", features = ["macros"] }
# internal
common/client-libs/validator-client/Cargo.toml
+1 -1
View File
@@ -35,7 +35,7 @@ nym-api-requests = { path = "../../../nym-api/nym-api-requests" }
# required for nyxd-client
# at some point it might be possible to make it wasm-compatible
# perhaps after https://github.com/cosmos/cosmos-rust/pull/97 is resolved (and tendermint-rs is updated)
async-trait = { version = "0.1.51", optional = true }
async-trait = { workspace = true, optional = true }
bip39 = { version = "1", features = ["rand"], optional = true }
nym-config = { path = "../../config", optional = true }
cosmrs = { git = "https://github.com/neacsu/cosmos-rust", branch = "neacsu/feegrant_support", features = ["rpc", "bip32", "cosmwasm"], optional = true}
common/credential-storage/Cargo.toml
+1 -1
View File
@@ -6,7 +6,7 @@ edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
async-trait = { version = "0.1.51" }
async-trait = { workspace = true }
log = { workspace = true }
sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"]}
common/crypto/src/asymmetric/encryption/mod.rs
+9
View File
@@ -3,6 +3,7 @@
use nym_pemstore::traits::{PemStorableKey, PemStorableKeyPair};
use std::fmt::{self, Display, Formatter};
use std::str::FromStr;
use thiserror::Error;
#[cfg(feature = "rand")]
@@ -131,6 +132,14 @@ impl PublicKey {
}
}
impl FromStr for PublicKey {
type Err = KeyRecoveryError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
PublicKey::from_base58_string(s)
}
}
#[cfg(feature = "serde")]
impl Serialize for PublicKey {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
common/mobile-storage/Cargo.toml
+1 -1
View File
@@ -6,6 +6,6 @@ edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
async-trait = { version = "0.1.51" }
async-trait = { workspace = true }
thiserror = "1.0"
common/statistics/Cargo.toml
+1 -1
View File
@@ -9,7 +9,7 @@ edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
async-trait = { version = "0.1.51" }
async-trait = { workspace = true }
log = { workspace = true }
reqwest = { version = "0.11", features = ["json"] }
serde = { version = "1.0", features = ["derive"] }
common/topology/Cargo.toml
+5
View File
@@ -16,6 +16,7 @@ bs58 = "0.4"
log = { workspace = true }
rand = { version = "0.7.3", features = ["wasm-bindgen"] }
thiserror = "1.0.37"
async-trait = { workspace = true, optional = true }
## internal
nym-crypto = { path = "../crypto" }
@@ -23,3 +24,7 @@ nym-mixnet-contract-common = { path = "../cosmwasm-smart-contracts/mixnet-contra
nym-sphinx-addressing = { path = "../nymsphinx/addressing" }
nym-sphinx-types = { path = "../nymsphinx/types" }
nym-bin-common = { path = "../bin-common" }
[features]
default = ["provider-trait"]
provider-trait = ["async-trait"]
common/topology/src/lib.rs
+54 -1
View File
@@ -20,6 +20,9 @@ pub mod filter;
pub mod gateway;
pub mod mix;
#[cfg(feature = "provider-trait")]
pub mod provider_trait;
#[derive(Debug, Clone, Error)]
pub enum NymTopologyError {
#[error("The provided network topology is empty - there are no mixnodes and no gateways on it - the network request(s) probably failed")]
@@ -39,6 +42,16 @@ pub enum NymTopologyError {
#[error("No mixnodes available on layer {layer}")]
EmptyMixLayer { layer: MixLayer },
#[error("Uneven layer distribution. Layer {layer} has {nodes} on it, while we expected a value between {lower_bound} and {upper_bound} as we have {total_nodes} nodes in total. Full breakdown: {layer_distribution:?}")]
UnevenLayerDistribution {
layer: MixLayer,
nodes: usize,
lower_bound: usize,
upper_bound: usize,
total_nodes: usize,
layer_distribution: Vec<(MixLayer, usize)>,
},
}
#[derive(Debug, Clone)]
@@ -97,7 +110,7 @@ impl NymTopology {
}
pub fn num_mixnodes(&self) -> usize {
self.mixes.values().flat_map(|m| m.iter()).count()
self.mixes.values().map(|m| m.len()).sum()
}
pub fn mixes_as_vec(&self) -> Vec<mix::Node> {
@@ -238,6 +251,46 @@ impl NymTopology {
Ok(())
}
pub fn ensure_even_layer_distribution(
&self,
lower_threshold: f32,
upper_threshold: f32,
) -> Result<(), NymTopologyError> {
let mixnodes_count = self.num_mixnodes();
let layers = self
.mixes
.iter()
.map(|(k, v)| (*k, v.len()))
.collect::<Vec<_>>();
if self.gateways.is_empty() {
return Err(NymTopologyError::NoGatewaysAvailable);
}
if layers.is_empty() {
return Err(NymTopologyError::NoMixnodesAvailable);
}
let upper_bound = (mixnodes_count as f32 * upper_threshold) as usize;
let lower_bound = (mixnodes_count as f32 * lower_threshold) as usize;
for (layer, nodes) in &layers {
if nodes < &lower_bound || nodes > &upper_bound {
return Err(NymTopologyError::UnevenLayerDistribution {
layer: *layer,
nodes: *nodes,
lower_bound,
upper_bound,
total_nodes: mixnodes_count,
layer_distribution: layers,
});
}
}
Ok(())
}
#[must_use]
pub fn filter_system_version(&self, expected_version: &str) -> Self {
self.filter_node_versions(expected_version)
common/topology/src/mix.rs
+2 -1
View File
@@ -3,7 +3,8 @@
use crate::{filter, NetworkAddress};
use nym_crypto::asymmetric::{encryption, identity};
use nym_mixnet_contract_common::{Layer, MixId, MixNodeBond};
pub use nym_mixnet_contract_common::Layer;
use nym_mixnet_contract_common::{MixId, MixNodeBond};
use nym_sphinx_addressing::nodes::NymNodeRoutingAddress;
use nym_sphinx_types::Node as SphinxNode;
use std::convert::{TryFrom, TryInto};
common/topology/src/provider_trait.rs
+44
View File
@@ -0,0 +1,44 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::NymTopology;
pub use async_trait::async_trait;
// hehe, wasm
#[cfg(not(target_arch = "wasm32"))]
#[async_trait]
pub trait TopologyProvider: Send {
async fn get_new_topology(&mut self) -> Option<NymTopology>;
}
#[cfg(target_arch = "wasm32")]
#[async_trait(?Send)]
pub trait TopologyProvider {
async fn get_new_topology(&mut self) -> Option<NymTopology>;
}
pub struct HardcodedTopologyProvider {
topology: NymTopology,
}
impl HardcodedTopologyProvider {
pub fn new(topology: NymTopology) -> Self {
HardcodedTopologyProvider { topology }
}
}
#[cfg(not(target_arch = "wasm32"))]
#[async_trait]
impl TopologyProvider for HardcodedTopologyProvider {
async fn get_new_topology(&mut self) -> Option<NymTopology> {
Some(self.topology.clone())
}
}
#[cfg(target_arch = "wasm32")]
#[async_trait(?Send)]
impl TopologyProvider for HardcodedTopologyProvider {
async fn get_new_topology(&mut self) -> Option<NymTopology> {
Some(self.topology.clone())
}
}
gateway/Cargo.toml
+1 -1
View File
@@ -16,7 +16,7 @@ rust-version = "1.56"
[dependencies]
anyhow = "1.0.53"
async-trait = { version = "0.1.51" }
async-trait = { workspace = true }
bip39 = "1.0.1"
bs58 = "0.4.0"
clap = { version = "4.0", features = ["cargo", "derive"] }
nym-api/Cargo.toml
+1 -1
View File
@@ -15,7 +15,7 @@ rust-version = "1.56"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
async-trait = "0.1.52"
async-trait = { workspace = true }
bs58 = {version = "0.4.0" }
bip39 = "1"
cfg-if = "1.0"
nym-api/src/network_monitor/monitor/mod.rs
+3 -10
View File
@@ -88,10 +88,7 @@ impl Monitor {
)
.await
{
error!(
"Failed to submit monitor run information to the database - {}",
err
);
error!("Failed to submit monitor run information to the database - {err}",);
// TODO: slightly more graceful shutdown here
process::exit(1);
@@ -124,12 +121,8 @@ impl Monitor {
for route in routes {
let mut packet_preparer = self.packet_preparer.clone();
let route = route.clone();
let route_test_packets = self.route_test_packets;
let gateway_packets = tokio::spawn(async move {
packet_preparer.prepare_test_route_viability_packets(&route, route_test_packets)
})
.await
.unwrap();
let gateway_packets = packet_preparer
.prepare_test_route_viability_packets(&route, self.route_test_packets);
packets.push(gateway_packets);
}
sdk/rust/nym-sdk/Cargo.toml
+1
View File
@@ -13,6 +13,7 @@ gateway-requests = { path = "../../../gateway/gateway-requests" }
nym-network-defaults = { path = "../../../common/network-defaults" }
nym-sphinx = { path = "../../../common/nymsphinx" }
nym-task = { path = "../../../common/task" }
nym-topology = { path = "../../../common/topology" }
validator-client = { path = "../../../common/client-libs/validator-client", features = ["nyxd-client"] }
futures = "0.3"
sdk/rust/nym-sdk/examples/custom_topology_provider.rs
+78
View File
@@ -0,0 +1,78 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use nym_sdk::mixnet;
use nym_topology::provider_trait::{async_trait, TopologyProvider};
use nym_topology::{nym_topology_from_detailed, NymTopology};
use url::Url;
struct MyTopologyProvider {
validator_client: validator_client::client::NymApiClient,
}
impl MyTopologyProvider {
fn new(nym_api_url: Url) -> MyTopologyProvider {
MyTopologyProvider {
validator_client: validator_client::client::NymApiClient::new(nym_api_url),
}
}
async fn get_topology(&self) -> NymTopology {
let mixnodes = self
.validator_client
.get_cached_active_mixnodes()
.await
.unwrap();
// in our topology provider only use mixnodes that have mix_id divisible by 3
// and have more than 100k nym (i.e. 100'000'000'000 unym) in stake
// why? because this is just an example to showcase arbitrary uses and capabilities of this trait
let filtered_mixnodes = mixnodes
.into_iter()
.filter(|mix| {
mix.mix_id() % 3 == 0 && mix.total_stake() > "100000000000".parse().unwrap()
})
.collect::<Vec<_>>();
let gateways = self.validator_client.get_cached_gateways().await.unwrap();
nym_topology_from_detailed(filtered_mixnodes, gateways)
}
}
#[async_trait]
impl TopologyProvider for MyTopologyProvider {
// this will be manually refreshed on a timer specified inside mixnet client config
async fn get_new_topology(&mut self) -> Option<NymTopology> {
Some(self.get_topology().await)
}
}
#[tokio::main]
async fn main() {
nym_bin_common::logging::setup_logging();
let nym_api = "https://validator.nymtech.net/api/".parse().unwrap();
let my_topology_provider = MyTopologyProvider::new(nym_api);
// Passing no config makes the client fire up an ephemeral session and figure shit out on its own
let mut client = mixnet::MixnetClientBuilder::new()
.custom_topology_provider(Box::new(my_topology_provider))
.build::<mixnet::EmptyReplyStorage>()
.await
.unwrap()
.connect_to_mixnet()
.await
.unwrap();
let our_address = client.nym_address();
println!("Our client nym address is: {our_address}");
// Send a message through the mixnet to ourselves
client.send_str(*our_address, "hello there").await;
println!("Waiting for message (ctrl-c to exit)");
client
.on_messages(|msg| println!("Received: {}", String::from_utf8_lossy(&msg.message)))
.await;
}
sdk/rust/nym-sdk/examples/manually_overwrite_topology.rs
+90
View File
@@ -0,0 +1,90 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use nym_sdk::mixnet;
use nym_topology::mix::Layer;
use nym_topology::{mix, NymTopology};
use std::collections::HashMap;
#[tokio::main]
async fn main() {
nym_bin_common::logging::setup_logging();
// Passing no config makes the client fire up an ephemeral session and figure shit out on its own
let mut client = mixnet::MixnetClient::connect_new().await.unwrap();
let starting_topology = client.read_current_topology().await.unwrap();
// but we don't like our default topology, we want to use only those very specific, hardcoded, nodes:
let mut mixnodes = HashMap::new();
mixnodes.insert(
1,
vec![mix::Node {
mix_id: 63,
owner: "n1k52k5n45cqt5qpjh8tcwmgqm0wkt355yy0g5vu".to_string(),
host: "172.105.92.48".parse().unwrap(),
mix_host: "172.105.92.48:1789".parse().unwrap(),
identity_key: "GLdR2NRVZBiCoCbv4fNqt9wUJZAnNjGXHkx3TjVAUzrK"
.parse()
.unwrap(),
sphinx_key: "CBmYewWf43iarBq349KhbfYMc9ys2ebXWd4Vp4CLQ5Rq"
.parse()
.unwrap(),
layer: Layer::One,
version: "1.1.0".to_string(),
}],
);
mixnodes.insert(
2,
vec![mix::Node {
mix_id: 23,
owner: "n1fzv4jc7fanl9s0qj02ge2ezk3kts545kjtek47".to_string(),
host: "178.79.143.65".parse().unwrap(),
mix_host: "178.79.143.65:1789".parse().unwrap(),
identity_key: "4Yr4qmEHd9sgsuQ83191FR2hD88RfsbMmB4tzhhZWriz"
.parse()
.unwrap(),
sphinx_key: "8ndjk5oZ6HxUZNScLJJ7hk39XtUqGexdKgW7hSX6kpWG"
.parse()
.unwrap(),
layer: Layer::Two,
version: "1.1.0".to_string(),
}],
);
mixnodes.insert(
3,
vec![mix::Node {
mix_id: 66,
owner: "n1ae2pjd7q9p0dea65pqkvcm4x9s264v4fktpyru".to_string(),
host: "139.162.247.97".parse().unwrap(),
mix_host: "139.162.247.97:1789".parse().unwrap(),
identity_key: "66UngapebhJRni3Nj52EW1qcNsWYiuonjkWJzHFsmyYY"
.parse()
.unwrap(),
sphinx_key: "7KyZh8Z8KxuVunqytAJ2eXFuZkCS7BLTZSzujHJZsGa2"
.parse()
.unwrap(),
layer: Layer::Three,
version: "1.1.0".to_string(),
}],
);
// but we like the available gateways, so keep using them!
// (we like them because the author of this example is too lazy to use the same hardcoded gateway
// during client initialisation to make sure we are able to send to ourselves : ) )
let custom_topology = NymTopology::new(mixnodes, starting_topology.gateways().to_vec());
client.manually_overwrite_topology(custom_topology).await;
// and everything we send now should only ever go via those nodes
let our_address = client.nym_address();
println!("Our client nym address is: {our_address}");
// Send a message through the mixnet to ourselves
client.send_str(*our_address, "hello there").await;
println!("Waiting for message (ctrl-c to exit)");
client
.on_messages(|msg| println!("Received: {}", String::from_utf8_lossy(&msg.message)))
.await;
}
sdk/rust/nym-sdk/src/mixnet.rs
+1
View File
@@ -52,4 +52,5 @@ pub use nym_sphinx::{
addressing::clients::{ClientIdentity, Recipient},
receiver::ReconstructedMessage,
};
pub use nym_topology::{provider_trait::TopologyProvider, NymTopology};
pub use paths::{GatewayKeyMode, KeyMode, StoragePaths};
sdk/rust/nym-sdk/src/mixnet/client.rs
+48 -3
View File
@@ -23,6 +23,8 @@ use nym_task::{
};
use futures::StreamExt;
use nym_topology::provider_trait::TopologyProvider;
use nym_topology::NymTopology;
use validator_client::nyxd::SigningNyxdClient;
use crate::{Error, Result};
@@ -38,6 +40,7 @@ pub struct MixnetClientBuilder {
storage_paths: Option<StoragePaths>,
keys: Option<Keys>,
gateway_config: Option<GatewayEndpointConfig>,
custom_topology_provider: Option<Box<dyn TopologyProvider>>,
}
impl MixnetClientBuilder {
@@ -70,6 +73,15 @@ impl MixnetClientBuilder {
self
}
#[must_use]
pub fn custom_topology_provider(
mut self,
topology_provider: Box<dyn TopologyProvider>,
) -> Self {
self.custom_topology_provider = Some(topology_provider);
self
}
/// Construct a [`DisconnectedMixnetClient`] from the setup specified.
pub async fn build<B>(self) -> Result<DisconnectedMixnetClient<B>>
where
@@ -79,7 +91,12 @@ impl MixnetClientBuilder {
let config = self.config.unwrap_or_default();
let storage_paths = self.storage_paths;
let mut client = DisconnectedMixnetClient::new(Some(config), storage_paths).await?;
let mut client = DisconnectedMixnetClient::new(
Some(config),
storage_paths,
self.custom_topology_provider,
)
.await?;
if let Some(keys) = self.keys {
client.set_keys(keys);
@@ -118,6 +135,9 @@ where
/// The storage backend for reply-SURBs
reply_storage_backend: B,
/// Alternative provider of network topology used for constructing sphinx packets.
custom_topology_provider: Option<Box<dyn TopologyProvider>>,
}
impl<B> DisconnectedMixnetClient<B>
@@ -133,6 +153,7 @@ where
async fn new(
config: Option<Config>,
paths: Option<StoragePaths>,
custom_topology_provider: Option<Box<dyn TopologyProvider>>,
) -> Result<DisconnectedMixnetClient<B>>
where
<B as ReplyStorageBackend>::StorageError: Send + Sync,
@@ -189,6 +210,7 @@ where
storage_paths: paths,
state: BuilderState::New,
reply_storage_backend,
custom_topology_provider,
})
}
@@ -377,7 +399,7 @@ where
// TODO: we currently don't support having a bandwidth controller
let bandwidth_controller = None;
let base_builder: BaseClientBuilder<'_, _, SigningNyxdClient> = BaseClientBuilder::new(
let mut base_builder: BaseClientBuilder<'_, _, SigningNyxdClient> = BaseClientBuilder::new(
&gateway_endpoint_config,
&self.config.debug_config,
self.key_manager.clone(),
@@ -387,6 +409,10 @@ where
self.config.nym_api_endpoints.clone(),
);
if let Some(topology_provider) = self.custom_topology_provider {
base_builder = base_builder.with_topology_provider(topology_provider);
}
let mut started_client = base_builder.start_base().await?;
let client_input = started_client.client_input.register_producer();
let mut client_output = started_client.client_output.register_consumer();
@@ -502,7 +528,7 @@ impl MixnetClient {
}
/// Get a shallow clone of [`ConnectionCommandSender`]. This is useful if you want to e.g
/// explictly close a transmission lane that is still sending data even though it should
/// explicitly close a transmission lane that is still sending data even though it should
/// cancel.
pub fn connection_command_sender(&self) -> ConnectionCommandSender {
self.client_input.connection_command_sender.clone()
@@ -514,6 +540,25 @@ impl MixnetClient {
self.client_state.shared_lane_queue_lengths.clone()
}
/// Change the network topology used by this client for constructing sphinx packets into the
/// provided one.
pub async fn manually_overwrite_topology(&self, new_topology: NymTopology) {
self.client_state
.topology_accessor
.manually_change_topology(new_topology)
.await
}
/// Gets the value of the currently used network topology.
pub async fn read_current_topology(&self) -> Option<NymTopology> {
self.client_state.topology_accessor.current_topology().await
}
/// Restore default topology refreshing behaviour of this client.
pub fn restore_automatic_topology_refreshing(&self) {
self.client_state.topology_accessor.release_manual_control()
}
/// Sends stringy data to the supplied Nym address
///
/// # Example
service-providers/network-requester/Cargo.toml
+1 -1
View File
@@ -11,7 +11,7 @@ rust-version = "1.65"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
async-trait = { version = "0.1.51" }
async-trait = { workspace = true }
clap = {version = "4.0", features = ["cargo", "derive"]}
dirs = "4.0"
futures = "0.3.24"