feature: nym topology revamp

common/client-core/Cargo.toml

@@ -3,7 +3,7 @@ name = "nym-client-core"
 version = "1.1.15"
 authors = ["Dave Hrycyszyn <[email protected]>"]
 edition = "2021"
-rust-version = "1.70"
+rust-version = "1.76"
 license.workspace = true
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -45,7 +45,7 @@ nym-nonexhaustive-delayqueue = { path = "../nonexhaustive-delayqueue" }
 nym-sphinx = { path = "../nymsphinx" }
 nym-statistics-common = { path = "../statistics" }
 nym-pemstore = { path = "../pemstore" }
-nym-topology = { path = "../topology", features = ["serializable"] }
+nym-topology = { path = "../topology", features = ["persistence"] }
 nym-mixnet-client = { path = "../client-libs/mixnet-client", default-features = false }
 nym-validator-client = { path = "../client-libs/validator-client", default-features = false }
 nym-task = { path = "../task" }

common/client-core/config-types/src/lib.rs

@@ -550,6 +550,15 @@ pub struct Topology {
     /// Specifies a minimum performance of a gateway that is used on route construction.
     /// This setting is only applicable when `NymApi` topology is used.
     pub minimum_gateway_performance: u8,
+
+    /// Specifies whether this client should attempt to retrieve all available network nodes
+    /// as opposed to just active mixnodes/gateways.
+    /// Useless without `ignore_epoch_roles = true`
+    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,
 }
 
 #[allow(clippy::large_enum_variant)]
@@ -586,6 +595,8 @@ impl Default for Topology {
             topology_structure: TopologyStructure::default(),
             minimum_mixnode_performance: DEFAULT_MIN_MIXNODE_PERFORMANCE,
             minimum_gateway_performance: DEFAULT_MIN_GATEWAY_PERFORMANCE,
+            use_extended_topology: false,
+            ignore_egress_epoch_role: false,
         }
     }
 }

common/client-core/src/cli_helpers/client_add_gateway.rs

@@ -112,7 +112,7 @@ where
                 source,
             }
         })?;
-        hardcoded_topology.get_gateways()
+        hardcoded_topology.entry_capable_nodes().cloned().collect()
     } else {
         let mut rng = rand::thread_rng();
         crate::init::helpers::current_gateways(
@@ -128,7 +128,7 @@ where
     // make sure the list of available gateways doesn't overlap the list of known gateways
     let available_gateways = available_gateways
         .into_iter()
-        .filter(|g| !registered_gateways.contains(g.identity()))
+        .filter(|g| !registered_gateways.contains(&g.identity()))
         .collect::<Vec<_>>();
 
     if available_gateways.is_empty() {

common/client-core/src/cli_helpers/client_init.rs

@@ -167,7 +167,7 @@ where
                 source,
             }
         })?;
-        hardcoded_topology.get_gateways()
+        hardcoded_topology.entry_capable_nodes().cloned().collect()
     } else {
         let mut rng = rand::thread_rng();
         crate::init::helpers::current_gateways(

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

@@ -3,7 +3,6 @@
 
 use super::received_buffer::ReceivedBufferMessage;
 use super::statistics_control::StatisticsControl;
-use super::topology_control::geo_aware_provider::GeoAwareTopologyProvider;
 use crate::client::base_client::storage::helpers::store_client_keys;
 use crate::client::base_client::storage::MixnetClientStorage;
 use crate::client::cover_traffic_stream::LoopCoverTrafficStream;
@@ -24,7 +23,7 @@ use crate::client::replies::reply_storage::{
 };
 use crate::client::topology_control::nym_api_provider::NymApiTopologyProvider;
 use crate::client::topology_control::{
-    nym_api_provider, TopologyAccessor, TopologyRefresher, TopologyRefresherConfig,
+    TopologyAccessor, TopologyRefresher, TopologyRefresherConfig,
 };
 use crate::config::{Config, DebugConfig};
 use crate::error::ClientCoreError;
@@ -464,8 +463,8 @@ where
             details_store
                 .upgrade_stored_remote_gateway_key(gateway_client.gateway_identity(), &updated_key)
                 .await.map_err(|err| {
-                    error!("failed to store upgraded gateway key! this connection might be forever broken now: {err}");
-                    ClientCoreError::GatewaysDetailsStoreError { source: Box::new(err) }
+                error!("failed to store upgraded gateway key! this connection might be forever broken now: {err}");
+                ClientCoreError::GatewaysDetailsStoreError { source: Box::new(err) }
             })?
         }
 
@@ -539,15 +538,15 @@ where
         // if no custom provider was ... provided ..., create one using nym-api
         custom_provider.unwrap_or_else(|| match config_topology.topology_structure {
             config::TopologyStructure::NymApi => Box::new(NymApiTopologyProvider::new(
-                nym_api_provider::Config {
-                    min_mixnode_performance: config_topology.minimum_mixnode_performance,
-                    min_gateway_performance: config_topology.minimum_gateway_performance,
-                },
+                config_topology,
                 nym_api_urls,
                 user_agent,
             )),
             config::TopologyStructure::GeoAware(group_by) => {
-                Box::new(GeoAwareTopologyProvider::new(nym_api_urls, group_by))
+                warn!("using deprecated 'GeoAware' topology provider - this option will be removed very soon");
+
+                #[allow(deprecated)]
+                Box::new(crate::client::topology_control::GeoAwareTopologyProvider::new(nym_api_urls, group_by))
             }
         })
     }
@@ -558,7 +557,7 @@ where
         topology_provider: Box<dyn TopologyProvider + Send + Sync>,
         topology_config: config::Topology,
         topology_accessor: TopologyAccessor,
-        local_gateway: &NodeIdentity,
+        local_gateway: NodeIdentity,
         wait_for_gateway: bool,
         mut shutdown: TaskClient,
     ) -> Result<(), ClientCoreError> {
@@ -590,7 +589,7 @@ where
         };
 
         if let Err(err) = topology_refresher
-            .ensure_contains_gateway(local_gateway)
+            .ensure_contains_routable_egress(local_gateway)
             .await
         {
             if let Some(waiting_timeout) = gateway_wait_timeout {
@@ -740,7 +739,8 @@ where
 
         // channels responsible for controlling ack messages
         let (ack_sender, ack_receiver) = mpsc::unbounded();
-        let shared_topology_accessor = TopologyAccessor::new();
+        let shared_topology_accessor =
+            TopologyAccessor::new(self.config.debug.topology.ignore_egress_epoch_role);
 
         // Shutdown notifier for signalling tasks to stop
         let shutdown = self

common/client-core/src/client/cover_traffic_stream.rs

@@ -163,6 +163,7 @@ impl LoopCoverTrafficStream<OsRng> {
         // poisson delay, but is it really a problem?
         let topology_permit = self.topology_access.get_read_permit().await;
         // the ack is sent back to ourselves (and then ignored)
+
         let topology_ref = match topology_permit.try_get_valid_topology_ref(
             &self.our_full_destination,
             Some(&self.our_full_destination),

common/client-core/src/client/inbound_messages.rs

@@ -28,7 +28,6 @@ pub enum InputMessage {
         recipient: Recipient,
         data: Vec<u8>,
         lane: TransmissionLane,
-        mix_hops: Option<u8>,
     },
 
     /// Creates a message used for a duplex anonymous communication where the recipient
@@ -44,7 +43,6 @@ pub enum InputMessage {
         data: Vec<u8>,
         reply_surbs: u32,
         lane: TransmissionLane,
-        mix_hops: Option<u8>,
     },
 
     /// Attempt to use our internally received and stored `ReplySurb` to send the message back
@@ -94,29 +92,6 @@ impl InputMessage {
             recipient,
             data,
             lane,
-            mix_hops: None,
-        };
-        if let Some(packet_type) = packet_type {
-            InputMessage::new_wrapper(message, packet_type)
-        } else {
-            message
-        }
-    }
-
-    // IMHO `new_regular` should take `mix_hops: Option<u8>` as an argument instead of creating
-    // this function, but that would potentially break backwards compatibility with the current API
-    pub fn new_regular_with_custom_hops(
-        recipient: Recipient,
-        data: Vec<u8>,
-        lane: TransmissionLane,
-        packet_type: Option<PacketType>,
-        mix_hops: Option<u8>,
-    ) -> Self {
-        let message = InputMessage::Regular {
-            recipient,
-            data,
-            lane,
-            mix_hops,
         };
         if let Some(packet_type) = packet_type {
             InputMessage::new_wrapper(message, packet_type)
@@ -137,7 +112,6 @@ impl InputMessage {
             data,
             reply_surbs,
             lane,
-            mix_hops: None,
         };
         if let Some(packet_type) = packet_type {
             InputMessage::new_wrapper(message, packet_type)
@@ -154,14 +128,12 @@ impl InputMessage {
         reply_surbs: u32,
         lane: TransmissionLane,
         packet_type: Option<PacketType>,
-        mix_hops: Option<u8>,
     ) -> Self {
         let message = InputMessage::Anonymous {
             recipient,
             data,
             reply_surbs,
             lane,
-            mix_hops,
         };
         if let Some(packet_type) = packet_type {
             InputMessage::new_wrapper(message, packet_type)

common/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs

@@ -73,11 +73,10 @@ where
         content: Vec<u8>,
         lane: TransmissionLane,
         packet_type: PacketType,
-        mix_hops: Option<u8>,
     ) {
         if let Err(err) = self
             .message_handler
-            .try_send_plain_message(recipient, content, lane, packet_type, mix_hops)
+            .try_send_plain_message(recipient, content, lane, packet_type)
             .await
         {
             warn!("failed to send a plain message - {err}")
@@ -91,18 +90,10 @@ where
         reply_surbs: u32,
         lane: TransmissionLane,
         packet_type: PacketType,
-        mix_hops: Option<u8>,
     ) {
         if let Err(err) = self
             .message_handler
-            .try_send_message_with_reply_surbs(
-                recipient,
-                content,
-                reply_surbs,
-                lane,
-                packet_type,
-                mix_hops,
-            )
+            .try_send_message_with_reply_surbs(recipient, content, reply_surbs, lane, packet_type)
             .await
         {
             warn!("failed to send a repliable message - {err}")
@@ -115,27 +106,18 @@ where
                 recipient,
                 data,
                 lane,
-                mix_hops,
             } => {
-                self.handle_plain_message(recipient, data, lane, PacketType::Mix, mix_hops)
+                self.handle_plain_message(recipient, data, lane, PacketType::Mix)
                     .await
             }
             InputMessage::Anonymous {
                 recipient,
                 data,
                 reply_surbs,
                 lane,
-                mix_hops,
             } => {
-                self.handle_repliable_message(
-                    recipient,
-                    data,
-                    reply_surbs,
-                    lane,
-                    PacketType::Mix,
-                    mix_hops,
-                )
-                .await
+                self.handle_repliable_message(recipient, data, reply_surbs, lane, PacketType::Mix)
+                    .await
             }
             InputMessage::Reply {
                 recipient_tag,
@@ -153,27 +135,18 @@ where
                     recipient,
                     data,
                     lane,
-                    mix_hops,
                 } => {
-                    self.handle_plain_message(recipient, data, lane, packet_type, mix_hops)
+                    self.handle_plain_message(recipient, data, lane, packet_type)
                         .await
                 }
                 InputMessage::Anonymous {
                     recipient,
                     data,
                     reply_surbs,
                     lane,
-                    mix_hops,
                 } => {
-                    self.handle_repliable_message(
-                        recipient,
-                        data,
-                        reply_surbs,
-                        lane,
-                        packet_type,
-                        mix_hops,
-                    )
-                    .await
+                    self.handle_repliable_message(recipient, data, reply_surbs, lane, packet_type)
+                        .await
                 }
                 InputMessage::Reply {
                     recipient_tag,

common/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs

@@ -70,7 +70,6 @@ pub(crate) struct PendingAcknowledgement {
     message_chunk: Fragment,
     delay: SphinxDelay,
     destination: PacketDestination,
-    mix_hops: Option<u8>,
     retransmissions: u32,
 }
 
@@ -80,13 +79,11 @@ impl PendingAcknowledgement {
         message_chunk: Fragment,
         delay: SphinxDelay,
         recipient: Recipient,
-        mix_hops: Option<u8>,
     ) -> Self {
         PendingAcknowledgement {
             message_chunk,
             delay,
             destination: PacketDestination::KnownRecipient(recipient.into()),
-            mix_hops,
             retransmissions: 0,
         }
     }
@@ -104,9 +101,6 @@ impl PendingAcknowledgement {
                 recipient_tag,
                 extra_surb_request,
             },
-            // Messages sent using SURBs are using the number of mix hops set by the recipient when
-            // they provided the SURBs, so it doesn't make sense to include it here.
-            mix_hops: None,
             retransmissions: 0,
         }
     }