Add global blacklist to validator-cache (#1168)

Author: durch

validator-api/src/config/mod.rs

@@ -34,6 +34,7 @@ const DEFAULT_PER_NODE_TEST_PACKETS: usize = 3;
 
 const DEFAULT_CACHE_INTERVAL: Duration = Duration::from_secs(30);
 const DEFAULT_MONITOR_THRESHOLD: u8 = 60;
+const DEFAULT_MIN_RELIABILITY: u8 = 50;
 
 #[derive(Debug, Default, Deserialize, PartialEq, Serialize)]
 pub struct Config {
@@ -107,6 +108,9 @@ impl Default for Base {
 #[derive(Debug, Deserialize, PartialEq, Serialize)]
 #[serde(default)]
 pub struct NetworkMonitor {
+    //  Mixnodes and gateways with relialability lower the this get blacklisted by network monitor, get no traffic and cannot be selected into a rewarded set.
+    min_reliability: u8,
+
     /// Specifies whether network monitoring service is enabled in this process.
     enabled: bool,
 
@@ -188,6 +192,7 @@ impl NetworkMonitor {
 impl Default for NetworkMonitor {
     fn default() -> Self {
         NetworkMonitor {
+            min_reliability: DEFAULT_MIN_RELIABILITY,
             enabled: false,
             testnet_mode: false,
             all_validator_apis: default_api_endpoints(),
@@ -418,6 +423,10 @@ impl Config {
         self.network_monitor.minimum_test_routes
     }
 
+    pub fn get_min_reliability(&self) -> u8 {
+        self.network_monitor.min_reliability
+    }
+
     pub fn get_route_test_packets(&self) -> usize {
         self.network_monitor.route_test_packets
     }

validator-api/src/config/template.rs

@@ -20,6 +20,9 @@ mixnet_contract_address = '{{ base.mixnet_contract_address }}'
 
 [network_monitor]
 
+# Mixnodes and gateways with relialability lower the this get blacklisted by network monitor, get no traffic and cannot be selected into a rewarded set.
+min_reliability = {{ network_monitor.min_reliability }}
+
 # Specifies whether network monitoring service is enabled in this process.
 enabled = {{ network_monitor.enabled }}
 

validator-api/src/contract_cache/mod.rs

@@ -13,6 +13,7 @@ use mixnet_contract_common::{
 use rocket::fairing::AdHoc;
 use serde::Serialize;
 use std::collections::HashMap;
+use std::collections::HashSet;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
@@ -40,6 +41,9 @@ struct ValidatorCacheInner {
     mixnodes: Cache<Vec<MixNodeBond>>,
     gateways: Cache<Vec<GatewayBond>>,
 
+    mixnodes_blacklist: Cache<HashSet<IdentityKey>>,
+    gateways_blacklist: Cache<HashSet<IdentityKey>>,
+
     rewarded_set: Cache<Vec<MixNodeBond>>,
     active_set: Cache<Vec<MixNodeBond>>,
 
@@ -203,6 +207,8 @@ impl ValidatorCache {
                     routes::get_gateways,
                     routes::get_active_set,
                     routes::get_rewarded_set,
+                    routes::get_blacklisted_mixnodes,
+                    routes::get_blacklisted_gateways,
                 ],
             )
         })
@@ -225,12 +231,82 @@ impl ValidatorCache {
         inner.current_reward_params.update(epoch_rewarding_params);
     }
 
-    pub async fn mixnodes(&self) -> Cache<Vec<MixNodeBond>> {
-        self.inner.read().await.mixnodes.clone()
+    pub async fn mixnodes_blacklist(&self) -> Cache<HashSet<IdentityKey>> {
+        self.inner.read().await.mixnodes_blacklist.clone()
+    }
+
+    pub async fn gateways_blacklist(&self) -> Cache<HashSet<IdentityKey>> {
+        self.inner.read().await.gateways_blacklist.clone()
+    }
+
+    pub async fn insert_mixnodes_blacklist(&mut self, mix_identity: IdentityKey) {
+        self.inner
+            .write()
+            .await
+            .mixnodes_blacklist
+            .value
+            .insert(mix_identity);
+    }
+
+    pub async fn remove_mixnodes_blacklist(&mut self, mix_identity: &str) {
+        self.inner
+            .write()
+            .await
+            .mixnodes_blacklist
+            .value
+            .remove(mix_identity);
+    }
+
+    pub async fn insert_gateways_blacklist(&mut self, gateway_identity: IdentityKey) {
+        self.inner
+            .write()
+            .await
+            .gateways_blacklist
+            .value
+            .insert(gateway_identity);
+    }
+
+    pub async fn remove_gateways_blacklist(&mut self, gateway_identity: &str) {
+        self.inner
+            .write()
+            .await
+            .gateways_blacklist
+            .value
+            .remove(gateway_identity);
+    }
+
+    pub async fn mixnodes(&self) -> Vec<MixNodeBond> {
+        let blacklist = self.mixnodes_blacklist().await.value;
+        self.inner
+            .read()
+            .await
+            .mixnodes
+            .value
+            .iter()
+            .filter(|mix| !blacklist.contains(mix.identity()))
+            .cloned()
+            .collect()
+    }
+
+    pub async fn mixnodes_all(&self) -> Vec<MixNodeBond> {
+        self.inner.read().await.mixnodes.value.clone()
+    }
+
+    pub async fn gateways(&self) -> Vec<GatewayBond> {
+        let blacklist = self.gateways_blacklist().await.value;
+        self.inner
+            .read()
+            .await
+            .gateways
+            .value
+            .iter()
+            .filter(|gateway| !blacklist.contains(gateway.identity()))
+            .cloned()
+            .collect()
     }
 
-    pub async fn gateways(&self) -> Cache<Vec<GatewayBond>> {
-        self.inner.read().await.gateways.clone()
+    pub async fn gateways_all(&self) -> Vec<GatewayBond> {
+        self.inner.read().await.gateways.value.clone()
     }
 
     pub async fn rewarded_set(&self) -> Cache<Vec<MixNodeBond>> {
@@ -312,6 +388,8 @@ impl ValidatorCacheInner {
             rewarded_set: Cache::default(),
             active_set: Cache::default(),
             current_reward_params: Cache::new(EpochRewardParams::new_empty()),
+            mixnodes_blacklist: Cache::default(),
+            gateways_blacklist: Cache::default(),
             // setting it to a dummy value on creation is fine, as nothing will be able to ready from it
             // since 'initialised' flag won't be set
             current_epoch: Cache::new(None),

validator-api/src/contract_cache/routes.rs

@@ -5,15 +5,16 @@ use crate::contract_cache::ValidatorCache;
 use mixnet_contract_common::{GatewayBond, MixNodeBond};
 use rocket::serde::json::Json;
 use rocket::State;
+use std::collections::HashSet;
 
 #[get("/mixnodes")]
 pub(crate) async fn get_mixnodes(cache: &State<ValidatorCache>) -> Json<Vec<MixNodeBond>> {
-    Json(cache.mixnodes().await.value)
+    Json(cache.mixnodes().await)
 }
 
 #[get("/gateways")]
 pub(crate) async fn get_gateways(cache: &State<ValidatorCache>) -> Json<Vec<GatewayBond>> {
-    Json(cache.gateways().await.value)
+    Json(cache.gateways().await)
 }
 
 #[get("/mixnodes/rewarded")]
@@ -25,3 +26,17 @@ pub(crate) async fn get_rewarded_set(cache: &State<ValidatorCache>) -> Json<Vec<
 pub(crate) async fn get_active_set(cache: &State<ValidatorCache>) -> Json<Vec<MixNodeBond>> {
     Json(cache.active_set().await.value)
 }
+
+#[get("/mixnodes/blacklisted")]
+pub(crate) async fn get_blacklisted_mixnodes(
+    cache: &State<ValidatorCache>,
+) -> Json<HashSet<String>> {
+    Json(cache.mixnodes_blacklist().await.value)
+}
+
+#[get("/gateways/blacklisted")]
+pub(crate) async fn get_blacklisted_gateways(
+    cache: &State<ValidatorCache>,
+) -> Json<HashSet<String>> {
+    Json(cache.gateways_blacklist().await.value)
+}

validator-api/src/network_monitor/monitor/mod.rs

@@ -42,6 +42,7 @@ pub(super) struct Monitor {
     /// The minimum number of test routes that need to be constructed (and working) in order for
     /// a monitor test run to be valid.
     minimum_test_routes: usize,
+    min_reliability: u8,
 }
 
 impl Monitor {
@@ -66,14 +67,44 @@ impl Monitor {
             route_test_packets: config.get_route_test_packets(),
             test_routes: config.get_test_routes(),
             minimum_test_routes: config.get_minimum_test_routes(),
+            min_reliability: config.get_min_reliability(),
         }
     }
 
     // while it might have been cleaner to put this into a separate `Notifier` structure,
     // I don't see much point considering it's only a single, small, method
-    async fn submit_new_node_statuses(&self, test_summary: TestSummary) {
+    async fn submit_new_node_statuses(&mut self, test_summary: TestSummary) {
         // indicate our run has completed successfully and should be used in any future
         // uptime calculations
+
+        for result in test_summary.mixnode_results.iter() {
+            if result.reliability < self.min_reliability {
+                self.packet_preparer
+                    .validator_cache()
+                    .insert_mixnodes_blacklist(result.identity.clone())
+                    .await;
+            } else {
+                self.packet_preparer
+                    .validator_cache()
+                    .remove_mixnodes_blacklist(&result.identity)
+                    .await;
+            }
+        }
+
+        for result in test_summary.gateway_results.iter() {
+            if result.reliability < self.min_reliability {
+                self.packet_preparer
+                    .validator_cache()
+                    .insert_gateways_blacklist(result.identity.clone())
+                    .await;
+            } else {
+                self.packet_preparer
+                    .validator_cache()
+                    .remove_gateways_blacklist(&result.identity)
+                    .await;
+            }
+        }
+
         if let Err(err) = self
             .node_status_storage
             .insert_monitor_run_results(

validator-api/src/network_monitor/monitor/preparer.rs

@@ -150,6 +150,10 @@ impl PacketPreparer {
         }
     }
 
+    pub fn validator_cache(&mut self) -> &mut ValidatorCache {
+        &mut self.validator_cache
+    }
+
     async fn wrap_test_packet(
         &mut self,
         packet: &TestPacket,
@@ -187,7 +191,7 @@ impl PacketPreparer {
             let gateways = self.validator_cache.gateways().await;
             let mixnodes = self.validator_cache.rewarded_set().await;
 
-            if gateways.into_inner().len() < minimum_full_routes {
+            if gateways.len() < minimum_full_routes {
                 info!(
                     "Minimal topology is still not online. Going to check again in {:?}",
                     initialisation_backoff
@@ -224,11 +228,11 @@ impl PacketPreparer {
         }
     }
 
-    async fn get_rewarded_nodes(&self) -> (Vec<MixNodeBond>, Vec<GatewayBond>) {
+    async fn all_mixnodes_and_gateways(&self) -> (Vec<MixNodeBond>, Vec<GatewayBond>) {
         info!(target: "Monitor", "Obtaining network topology...");
 
-        let mixnodes = self.validator_cache.rewarded_set().await.into_inner();
-        let gateways = self.validator_cache.gateways().await.into_inner();
+        let mixnodes = self.validator_cache.mixnodes_all().await;
+        let gateways = self.validator_cache.gateways_all().await;
 
         (mixnodes, gateways)
     }
@@ -262,19 +266,17 @@ impl PacketPreparer {
         n: usize,
         blacklist: &mut HashSet<String>,
     ) -> Option<Vec<TestRoute>> {
-        let rewarded_set = self.validator_cache.rewarded_set().await.into_inner();
-        let gateways = self.validator_cache.gateways().await.into_inner();
-
+        let (mixnodes, gateways) = self.all_mixnodes_and_gateways().await;
         // separate mixes into layers for easier selection
         let mut layered_mixes = HashMap::new();
-        for rewarded_mix in rewarded_set {
+        for mix in mixnodes {
             // filter out mixes on the blacklist
-            if blacklist.contains(&rewarded_mix.mix_node.identity_key) {
+            if blacklist.contains(&mix.mix_node.identity_key) {
                 continue;
             }
-            let layer = rewarded_mix.layer;
+            let layer = mix.layer;
             let mixes = layered_mixes.entry(layer).or_insert_with(Vec::new);
-            mixes.push(rewarded_mix)
+            mixes.push(mix)
         }
         // filter out gateways on the blacklist
         let gateways = gateways
@@ -458,7 +460,7 @@ impl PacketPreparer {
         // (remember that "idle" nodes are still part of that set)
         // we don't care about other nodes, i.e. nodes that are bonded but will not get
         // any reward during the current rewarding interval
-        let (rewarded_set, all_gateways) = self.get_rewarded_nodes().await;
+        let (rewarded_set, all_gateways) = self.all_mixnodes_and_gateways().await;
 
         let (mixes, invalid_mixnodes) = self.filter_outdated_and_malformed_mixnodes(rewarded_set);
         let (gateways, invalid_gateways) =

validator-api/src/node_status_api/routes.rs

@@ -212,7 +212,7 @@ pub(crate) async fn get_mixnode_inclusion_probability(
     cache: &State<ValidatorCache>,
     identity: String,
 ) -> Json<Option<InclusionProbabilityResponse>> {
-    let mixnodes = cache.mixnodes().await.into_inner();
+    let mixnodes = cache.mixnodes().await;
 
     if let Some(target_mixnode) = mixnodes.iter().find(|x| x.identity() == &identity) {
         let total_bonded_tokens = mixnodes

validator-api/src/rewarded_set_updater/mod.rs

@@ -204,7 +204,7 @@ impl RewardedSetUpdater {
         let epoch = self.epoch().await?;
         log::info!("Starting rewarded set update");
         // we know the entries are not stale, as a matter of fact they were JUST updated, since we got notified
-        let all_nodes = self.validator_cache.mixnodes().await.into_inner();
+        let all_nodes = self.validator_cache.mixnodes().await;
         let epoch_reward_params = self
             .validator_cache
             .epoch_reward_params()