diff --git a/common/cosmwasm-smart-contracts/mixnet-contract/src/types.rs b/common/cosmwasm-smart-contracts/mixnet-contract/src/types.rs index b3c4e5b2be..14bcf94fa0 100644 --- a/common/cosmwasm-smart-contracts/mixnet-contract/src/types.rs +++ b/common/cosmwasm-smart-contracts/mixnet-contract/src/types.rs @@ -110,6 +110,10 @@ impl RewardedSet { } None } + + pub fn all_ids(&self) -> Vec { + Vec::new() + } } #[cw_serde] diff --git a/common/topology/src/lib.rs b/common/topology/src/lib.rs index 0a5e76f658..22a237074f 100644 --- a/common/topology/src/lib.rs +++ b/common/topology/src/lib.rs @@ -182,7 +182,7 @@ impl NymRouteProvider { } -#[derive(Clone, Debug, Default, Serialize, Deserialize)] +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)] pub struct NymTopology { // for the purposes of future VRF, everyone will need the same view of the network, regardless of performance filtering // so we use the same 'master' rewarded set information for that @@ -558,6 +558,10 @@ impl NymTopology { .values() } + pub fn all_node_ids(&self) -> impl Iterator { + self.node_details.keys() + } + pub fn gateways(&self) -> impl Iterator { self.node_details .values() diff --git a/common/topology/src/node.rs b/common/topology/src/node.rs index 2019a3ee85..8c710c6573 100644 --- a/common/topology/src/node.rs +++ b/common/topology/src/node.rs @@ -20,7 +20,7 @@ pub enum RoutingNodeError { NoIpAddressesProvided { node_id: NodeId, identity: String }, } -#[derive(Clone, Debug, Serialize, Deserialize)] +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] pub struct EntryDetails { // to allow client to choose ipv6 preference, if available pub ip_addresses: Vec, @@ -29,7 +29,7 @@ pub struct EntryDetails { pub clients_wss_port: Option, } -#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)] pub struct SupportedRoles { pub mixnode: bool, pub mixnet_entry: bool, @@ -46,7 +46,7 @@ impl From for SupportedRoles { } } -#[derive(Clone, Debug, Serialize, Deserialize)] +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] pub struct RoutingNode { pub node_id: NodeId, diff --git a/common/topology/src/providers/piecewise.rs b/common/topology/src/providers/piecewise.rs index e40d154608..8eccba782c 100644 --- a/common/topology/src/providers/piecewise.rs +++ b/common/topology/src/providers/piecewise.rs @@ -5,13 +5,19 @@ //! #![warn(missing_docs)] -use crate::{EpochRewardedSet, NymTopology, Role, RoutingNode, TopologyProvider}; +use crate::{EpochRewardedSet, NymTopology, RoutingNode, TopologyProvider}; use async_trait::async_trait; use time::OffsetDateTime; use tokio::sync::Mutex; +use tracing::{debug, warn}; -use std::{cmp::min, collections::HashMap, sync::Arc, time::Duration}; +use std::{ + cmp::min, + collections::{HashMap, HashSet}, + sync::Arc, + time::Duration, +}; #[derive(Debug)] pub struct Config { @@ -59,6 +65,8 @@ impl Config { } } +/// Topology Provider build around a cached piecewise provider that uses the Nym API to +/// fetch changes and node details. #[derive(Clone)] pub struct NymTopologyProvider { inner: Arc>>, @@ -81,18 +89,18 @@ impl NymTopologyProvider { /// Bypass the caching for the topology and force a check for the latest updates next time the /// topology is requested. pub async fn force_refresh(&self) { - let mut guard = self.inner.lock().await; - guard.cached_at = OffsetDateTime::UNIX_EPOCH; - } + 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. /// /// WARNING: This may be slow / require non-trivial bandwidth. pub async fn force_clear(&self) { - let mut guard = self.inner.lock().await; - guard.cached = None; - } + let mut guard = self.inner.lock().await; + guard.cached = None; + } } #[async_trait] @@ -104,9 +112,9 @@ impl TopologyProvider for NymTopologyProvider { return Some(cached); } - // not cached, or cache expired. try update. + // not cached, or cache expired. try update. guard.update_cache().await; - guard.get_current_compatible_topology().await + guard.get_current_compatible_topology().await } } @@ -144,37 +152,73 @@ impl NymTopologyProviderInner { } async fn update_cache(&mut self) { - todo!("pull layer assignments and then batch"); - // let updated_cache = self.get_new_topology().await?; + if let Some(ref mut cached_topology) = self.cached { + // get layer assignment map + let response = self.topology_manager.get_layer_assignments().await; + if response.is_none() { + warn!("pulled layer assignments and got no response"); + self.cached_at = OffsetDateTime::now_utc(); + return; + } - // self.cached_at = OffsetDateTime::now_utc(); - // self.cached = Some(updated_cache.clone()); + let layer_assignments = response.unwrap(); - // Some(updated_cache) + // Check if we already know about the epoch + if cached_topology.rewarded_set.epoch_id == layer_assignments.epoch_id { + debug!("pulled layer assignments, epoch already known"); + self.cached_at = OffsetDateTime::now_utc(); + return; + } + + // get the set of node IDs + let newly_assigned_node_ids = layer_assignments.assignment.all_ids(); + let new_id_set = HashSet::::from_iter(newly_assigned_node_ids); + 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).copied().collect(); + + // Pull node descriptors for unknown IDs + let response = self + .topology_manager + .get_descriptor_batch(&unknown_node_ids[..]) + .await; + + // Add the new nodes to our cached topology + if let Some(new_descriptors) = response { + cached_topology.add_routing_nodes(new_descriptors.values()); + } + } else { + self.cached = self.topology_manager.get_full_topology().await; + } + + self.cached_at = OffsetDateTime::now_utc(); } - /// Gets the current topology state using `Self::cached_topology` and then applies any filters - /// defined in the provided Config. + /// Gets the current topology state using `Self::cached_topology` and then applies any filters + /// defined in the provided Config. async fn get_current_compatible_topology(&mut self) -> Option { - let full_topology = self.cached_topology()?; + let full_topology = self.cached_topology()?; let mut topology = NymTopology::new_empty(full_topology.rewarded_set().clone()); if self.config.use_extended_topology { - topology.add_additional_nodes(full_topology.all_nodes().filter(|n| { + topology.add_additional_nodes(full_topology.all_nodes().filter(|n| { n.performance.round_to_integer() >= self.config.min_node_performance() })); - return Some(full_topology); - } + return Some(full_topology); + } + + topology.add_additional_nodes( + full_topology.mixnodes().filter(|m| { + m.performance.round_to_integer() >= self.config.min_mixnode_performance + }), + ); + topology.add_additional_nodes( + full_topology.gateways().filter(|m| { + m.performance.round_to_integer() >= self.config.min_gateway_performance + }), + ); - topology.add_additional_nodes(full_topology.mixnodes().filter(|m| { - m.performance.round_to_integer() >= self.config.min_mixnode_performance - })); - topology.add_additional_nodes(full_topology.gateways().filter(|m| { - m.performance.round_to_integer() >= self.config.min_gateway_performance - })); - Some(topology) } } @@ -186,9 +230,7 @@ impl TopologyProvider for NymTopologyProviderInner } } -/// -/// -/// This is intentionally private, such that we can modify it at any time in the future. +/// Trait allowing construction and upkeep of a #[async_trait] pub trait PiecewiseTopologyProvider: Send { async fn get_full_topology(&mut self) -> Option; @@ -198,11 +240,17 @@ pub trait PiecewiseTopologyProvider: Send { async fn get_layer_assignments(&mut self) -> Option; } - #[cfg(test)] mod test { - use super::*; + use std::net::SocketAddr; + use nym_mixnet_contract_common::Percent; + use nym_crypto::asymmetric::identity::PublicKey as IdentityPubkey; + use nym_crypto::asymmetric::encryption::PublicKey as SphinxPubkey; + use super::*; + use crate::SupportedRoles; + + #[derive(Clone)] struct PassthroughPiecewiseTopologyProvider { topo: NymTopology, } @@ -215,28 +263,78 @@ mod test { async fn get_descriptor_batch(&mut self, ids: &[u32]) -> Option> { let mut nodes = HashMap::new(); - ids.iter().for_each(|id| { - if let Some(node) = self.topo.node_details.get(id) { - nodes.insert(*id, node.clone()); - } - }); + ids.iter().for_each(|id| { + if let Some(node) = self.topo.node_details.get(id) { + nodes.insert(*id, node.clone()); + } + }); - Some(nodes) + Some(nodes) } async fn get_layer_assignments(&mut self) -> Option { - return Some(self.topo.rewarded_set.clone().into()) + return Some(self.topo.rewarded_set.clone().into()); } } #[tokio::test] async fn test_topology_provider() -> Result<(), Box> { - let topo_mgr = PassthroughPiecewiseTopologyProvider { + let mut topo_mgr = PassthroughPiecewiseTopologyProvider { topo: NymTopology::default(), }; - let topo_provider = NymTopologyProviderInner::new(Config::default(), topo_mgr, None); + let mut topo_provider = + NymTopologyProviderInner::new(Config::default(), topo_mgr.clone(), None); + + // No initial topology was provided, No update has run yet, None should be returned + assert_eq!(topo_provider.cached_topology(), None); + + // force an update of the cached topology + topo_provider.update_cache(); + + let topo = topo_provider.cached_topology(); + assert!(topo.is_some()); + + // create a change in the manager to make sure it is propogated to the provider cache on update + 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); + topo_provider.topology_manager = topo_mgr.clone(); + + // force an update of the cached topology + topo_provider.update_cache(); + + let topo = topo_provider.cached_topology(); + assert!(topo.is_some()); Ok(()) } + + #[tokio::test] + async fn test_topology_provider_by_trait() -> Result<(), Box> { + let mut topo_mgr = PassthroughPiecewiseTopologyProvider { + topo: NymTopology::default(), + }; + + let mut topo_provider = + NymTopologyProviderInner::new(Config::default(), topo_mgr.clone(), None); + + Ok(()) + } + + fn fake_node(node_id: u32) -> RoutingNode { + RoutingNode { + node_id, + mix_host: "127.0.0.1:2345".parse().unwrap(), + entry: None, + identity_key: IdentityPubkey::from_bytes(&[0u8; 32][..]).unwrap(), + sphinx_key: SphinxPubkey::from_bytes(&[0u8; 32][..]).unwrap(), + supported_roles: SupportedRoles { + mixnode: true, + mixnet_entry: true, + mixnet_exit: true, + }, + performance: Percent::hundred(), + } + } } diff --git a/common/topology/src/rewarded_set.rs b/common/topology/src/rewarded_set.rs index 0d06239be6..22f50e9b65 100644 --- a/common/topology/src/rewarded_set.rs +++ b/common/topology/src/rewarded_set.rs @@ -6,7 +6,7 @@ use nym_mixnet_contract_common::{EpochId, EpochRewardedSet, NodeId, RewardedSet} use serde::{Deserialize, Serialize}; use std::collections::HashSet; -#[derive(Clone, Debug, Default, Serialize, Deserialize)] +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)] pub struct CachedEpochRewardedSet { pub epoch_id: EpochId,