diff --git a/common/topology/src/lib.rs b/common/topology/src/lib.rs index b9249cd2ed..c99a5a1d67 100644 --- a/common/topology/src/lib.rs +++ b/common/topology/src/lib.rs @@ -8,10 +8,8 @@ use nym_sphinx_types::Node as SphinxNode; use rand::prelude::IteratorRandom; use rand::{CryptoRng, Rng}; use std::borrow::Borrow; -use std::collections::hash_map::Values; use std::collections::{HashMap, HashSet}; use std::fmt::Display; -use std::iter::Filter; use std::net::IpAddr; use tracing::{debug, trace, warn}; diff --git a/common/topology/src/providers/piecewise.rs b/common/topology/src/providers/piecewise.rs index 2c5ec70df6..b14516d51f 100644 --- a/common/topology/src/providers/piecewise.rs +++ b/common/topology/src/providers/piecewise.rs @@ -19,11 +19,23 @@ use std::{ time::Duration, }; +/// Topology filtering and caching configuration #[derive(Debug)] pub struct Config { + /// Specifies a minimum performance of a mixnode that is used on route construction. + /// This setting is only applicable when `NymApi` topology is used. pub min_mixnode_performance: u8, + + /// Specifies a minimum performance of a gateway that is used on route construction. + /// This setting is only applicable when `NymApi` topology is used. pub min_gateway_performance: u8, + + /// Specifies whether this client should attempt to retrieve all available network nodes + /// as opposed to just active mixnodes/gateways. pub use_extended_topology: bool, + + /// Specifies whether this client should ignore the current epoch role of the target egress node + /// when constructing the final hop packets. pub ignore_egress_epoch_role: bool, /// Minimum duration during which querying the topology will NOT attempt to re-fetch data, and @@ -87,14 +99,16 @@ impl NymTopologyProvider { } /// Bypass the caching for the topology and force a check for the latest updates next time the - /// topology is requested. + /// topology is requested. This fn requires async to get lock in case other threads have access + /// to the cached topology state. pub async fn force_refresh(&self) { let mut guard = self.inner.lock().await; guard.cached_at = OffsetDateTime::UNIX_EPOCH; } /// Remove all stored topology state. The next time the topology is requested this will force a - /// pull of all topology information. + /// pull of all topology information. This fn requires async to get lock in case other threads + /// have access to the cached topology state. /// /// WARNING: This may be slow / require non-trivial bandwidth. pub async fn force_clear(&self) { @@ -186,6 +200,16 @@ impl NymTopologyProviderInner { if let Some(new_descriptors) = response { cached_topology.add_routing_nodes(new_descriptors.values()); } + + // double check that we have the expected nodes + let known_id_set = HashSet::::from_iter(cached_topology.all_node_ids().copied()); + let unknown_node_ids: Vec<_> = new_id_set.difference(&known_id_set).collect(); + if !unknown_node_ids.is_empty() { + warn!( + "still missing descriptors for nodes in the assigned set: {:?}", + unknown_node_ids + ); + } } else { self.cached = self.topology_manager.get_full_topology().await; } @@ -233,10 +257,16 @@ impl TopologyProvider for NymTopologyProviderInner /// Trait allowing construction and upkeep of a #[async_trait] pub trait PiecewiseTopologyProvider: Send { + /// Pull a copy of the full topology. + /// + /// This is intended to be used sparingly as repeated usage could result in fetching duplicate + /// information more often than necessary. async fn get_full_topology(&mut self) -> Option; + /// Fetch a node descriptors for the set of provided IDs if available. async fn get_descriptor_batch(&mut self, ids: &[u32]) -> Option>; + /// Fetch the latest mapping of node IDs to Nym Network layer. async fn get_layer_assignments(&mut self) -> Option; } @@ -315,17 +345,81 @@ mod test { let topo2 = topo_provider.cached_topology().unwrap(); assert_eq!(topo1, topo2); + // Add a node without a descriptor to make sure warning is printed. + topo_mgr.topo.rewarded_set.epoch_id += 1; + topo_mgr.topo.rewarded_set.entry_gateways = HashSet::from([123, 456]); + topo_provider.topology_manager = topo_mgr.clone(); + + // try forcing an update even though the epoch has not changed. Should result in no change + topo_provider.update_cache().await; + let _ = topo_provider.cached_topology().unwrap(); + Ok(()) } #[tokio::test] async fn test_topology_provider_by_trait() -> Result<(), Box> { - let topo_mgr = PassthroughPiecewiseTopologyProvider { + let mut topo_mgr = PassthroughPiecewiseTopologyProvider { topo: NymTopology::default(), }; - let _topo_provider = - NymTopologyProviderInner::new(Config::default(), topo_mgr.clone(), None); + let mut topo_provider = NymTopologyProvider::new(topo_mgr.clone(), Config::default(), None); + + // No initial topology was provided, the NymTopologyProvider should do an update from the + // manager to build its cache. This should be our empty topology initialized in the manage + // above + let maybe_topo = topo_provider.get_new_topology().await; + assert!(maybe_topo.is_some()); + let topo1 = maybe_topo.unwrap(); + assert!(topo1.is_empty()); + + // Try pulling again, should give response from cache because we are under ttl + let maybe_topo = topo_provider.get_new_topology().await; + assert!(maybe_topo.is_some()); + let topo2 = maybe_topo.unwrap(); + assert_eq!(topo1, topo2); + + // create a change in the manager + topo_mgr.topo.rewarded_set.epoch_id += 1; + topo_mgr.topo.rewarded_set.entry_gateways = HashSet::from([123]); + assert_eq!(topo_mgr.topo.node_details.insert(123, fake_node(123)), None); + { + let mut guard = topo_provider.inner.lock().await; + guard.topology_manager = topo_mgr.clone(); + drop(guard) + } + + // The NymTopologyProvider should still serve from cache because we haven't crossed ttl + // despite updates being available in the manager + let maybe_topo = topo_provider.get_new_topology().await; + assert!(maybe_topo.is_some()); + let topo3 = maybe_topo.unwrap(); + assert_eq!(topo2, topo3); + + // force ttl timeout should allow refresh that includes latest changes from manager + topo_provider.force_refresh().await; + let maybe_topo = topo_provider.get_new_topology().await; + assert!(maybe_topo.is_some()); + let topo4 = maybe_topo.unwrap(); + assert_ne!(topo3, topo4); + assert!(topo4.node_details.contains_key(&123)); + + // create another change in the manager + topo_mgr.topo.rewarded_set.epoch_id += 1; + topo_mgr.topo.rewarded_set.entry_gateways = HashSet::from([123, 456]); + assert_eq!(topo_mgr.topo.node_details.insert(456, fake_node(456)), None); + { + let mut guard = topo_provider.inner.lock().await; + guard.topology_manager = topo_mgr.clone(); + drop(guard) + } + + // force clear cache should also pull latest full topology + topo_provider.force_clear().await; + let maybe_topo = topo_provider.get_new_topology().await; + assert!(maybe_topo.is_some()); + let topo5 = maybe_topo.unwrap(); + assert!(topo5.node_details.contains_key(&456)); Ok(()) }