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

Add global blacklist to validator-cache (#1168)

Browse Source
This commit is contained in:
Drazen Urch
2022-03-25 17:02:01 +01:00
committed by GitHub
parent 5932974108
commit 65a45bc0a8
8 changed files with 159 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
validator-api/src/config/mod.rs
+9
View File
@@ -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
+3
View File
@@ -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
+82 -4
View File
@@ -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(&self) -> Cache<Vec<GatewayBond>> {
self.inner.read().await.gateways.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_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
+17 -2
View File
@@ -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
+32 -1
View File
@@ -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
+14 -12
View File
@@ -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
+1 -1
View File
@@ -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
+1 -1
View File
@@ -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()