Feature/hourly set updates (#1012)
* Rename function/variables mixnodes->set * Stub utility interface * Rewarded set contract interface * Move epoch to common, epoch to contract * Move epoch to the chain * Rewarded set validator-api * [ci skip] Generate TS types * Epoch queries * Moved new code to a new module * Restored cosmwasm dependencies to their beta.3 versions for better compatibility with the rest of the codebase * Rewarded set write reorganisation * Stub for validator api module responsible for rewarded set updates * Reorganised validator api cache * Pending contract changes * Relevant updates to the validator client * Updating rewarded set based on contract state * Advancing/Setting current epoch in the contract * Using blocktime as 'now' at startup * Adjusted validator-api side rewarding code * Contract cleanup + query for epoch rewarded set heights * [ci skip] Generate TS types * Simplified rewarder processing loop and initial sync * [ci skip] Generate TS types * Fixed EXISTING query-related unit tests * Fixed existing unit tests for rewarding-related transactions * Cargo fmt * Removed some dead code * Using cosmwasm 1.0.0-beta3 for compatibility [with cw-storage-plus and rest of codebase] * Missing TryInto import * Additional storage and query related unit tests + a bug fix * Transaction-related unit tests + bug fixes * Required migration code * Update common/cosmwasm-smart-contracts/mixnet-contract/src/msg.rs Co-authored-by: Drazen Urch <drazen@urch.eu> * Update common/cosmwasm-smart-contracts/mixnet-contract/src/msg.rs Co-authored-by: Drazen Urch <drazen@urch.eu> * Constant renaming * Changed determining previous epoch return type to Option<Epoch> if they would precede the genesis * Exposed the new endpoint to the wallet * Epoch-related unit tests fixes * Recommended #[must_use] on next_epoch method * Renamed all epoch occurences to interval As they refer to the 'rewarding interval' Co-authored-by: Jędrzej Stuczyński <jedrzej.stuczynski@gmail.com> Co-authored-by: jstuczyn <jstuczyn@users.noreply.github.com>
This commit is contained in:
Drazen Urch
@@ -0,0 +1,327 @@
|
||||
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
use crate::nymd_client::Client;
|
||||
use crate::rewarding::IntervalRewardParams;
|
||||
use ::time::OffsetDateTime;
|
||||
use anyhow::Result;
|
||||
use config::defaults::VALIDATOR_API_VERSION;
|
||||
use mixnet_contract_common::{
|
||||
GatewayBond, IdentityKey, IdentityKeyRef, Interval, MixNodeBond, RewardedSetNodeStatus,
|
||||
};
|
||||
|
||||
use rocket::fairing::AdHoc;
|
||||
use serde::Serialize;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::{Notify, RwLock};
|
||||
use tokio::time;
|
||||
use validator_api_requests::models::MixnodeStatus;
|
||||
use validator_client::nymd::CosmWasmClient;
|
||||
|
||||
pub(crate) mod routes;
|
||||
|
||||
pub struct ValidatorCacheRefresher<C> {
|
||||
nymd_client: Client<C>,
|
||||
cache: ValidatorCache,
|
||||
caching_interval: Duration,
|
||||
update_rewarded_set_notify: Option<Arc<Notify>>,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct ValidatorCache {
|
||||
initialised: Arc<AtomicBool>,
|
||||
inner: Arc<RwLock<ValidatorCacheInner>>,
|
||||
}
|
||||
|
||||
struct ValidatorCacheInner {
|
||||
mixnodes: Cache<Vec<MixNodeBond>>,
|
||||
gateways: Cache<Vec<GatewayBond>>,
|
||||
|
||||
rewarded_set: Cache<Vec<MixNodeBond>>,
|
||||
active_set: Cache<Vec<MixNodeBond>>,
|
||||
|
||||
current_reward_params: Cache<IntervalRewardParams>,
|
||||
current_interval: Cache<Interval>,
|
||||
}
|
||||
|
||||
fn current_unix_timestamp() -> i64 {
|
||||
let now = OffsetDateTime::now_utc();
|
||||
now.unix_timestamp()
|
||||
}
|
||||
|
||||
#[derive(Default, Serialize, Clone)]
|
||||
pub struct Cache<T> {
|
||||
value: T,
|
||||
as_at: i64,
|
||||
}
|
||||
|
||||
impl<T: Clone> Cache<T> {
|
||||
fn new(value: T) -> Self {
|
||||
Cache {
|
||||
value,
|
||||
as_at: current_unix_timestamp(),
|
||||
}
|
||||
}
|
||||
|
||||
fn update(&mut self, value: T) {
|
||||
self.value = value;
|
||||
self.as_at = current_unix_timestamp()
|
||||
}
|
||||
|
||||
pub fn timestamp(&self) -> i64 {
|
||||
self.as_at
|
||||
}
|
||||
|
||||
pub fn into_inner(self) -> T {
|
||||
self.value
|
||||
}
|
||||
}
|
||||
|
||||
impl<C> ValidatorCacheRefresher<C> {
|
||||
pub(crate) fn new(
|
||||
nymd_client: Client<C>,
|
||||
caching_interval: Duration,
|
||||
cache: ValidatorCache,
|
||||
update_rewarded_set_notify: Option<Arc<Notify>>,
|
||||
) -> Self {
|
||||
ValidatorCacheRefresher {
|
||||
nymd_client,
|
||||
cache,
|
||||
caching_interval,
|
||||
update_rewarded_set_notify,
|
||||
}
|
||||
}
|
||||
|
||||
fn collect_rewarded_and_active_set_details(
|
||||
&self,
|
||||
all_mixnodes: &[MixNodeBond],
|
||||
rewarded_set_identities: Vec<(IdentityKey, RewardedSetNodeStatus)>,
|
||||
) -> (Vec<MixNodeBond>, Vec<MixNodeBond>) {
|
||||
let mut active_set = Vec::new();
|
||||
let mut rewarded_set = Vec::new();
|
||||
let rewarded_set_identities = rewarded_set_identities
|
||||
.into_iter()
|
||||
.collect::<HashMap<_, _>>();
|
||||
|
||||
for mix in all_mixnodes {
|
||||
if let Some(status) = rewarded_set_identities.get(mix.identity()) {
|
||||
rewarded_set.push(mix.clone());
|
||||
if status.is_active() {
|
||||
active_set.push(mix.clone())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
(rewarded_set, active_set)
|
||||
}
|
||||
|
||||
async fn refresh_cache(&self) -> Result<()>
|
||||
where
|
||||
C: CosmWasmClient + Sync,
|
||||
{
|
||||
let (mixnodes, gateways) = tokio::try_join!(
|
||||
self.nymd_client.get_mixnodes(),
|
||||
self.nymd_client.get_gateways(),
|
||||
)?;
|
||||
|
||||
let rewarded_set_identities = self.nymd_client.get_rewarded_set_identities().await?;
|
||||
let (rewarded_set, active_set) =
|
||||
self.collect_rewarded_and_active_set_details(&mixnodes, rewarded_set_identities);
|
||||
|
||||
let interval_rewarding_params = self
|
||||
.nymd_client
|
||||
.get_current_interval_reward_params()
|
||||
.await?;
|
||||
let current_interval = self.nymd_client.get_current_interval().await?;
|
||||
|
||||
info!(
|
||||
"Updating validator cache. There are {} mixnodes and {} gateways",
|
||||
mixnodes.len(),
|
||||
gateways.len(),
|
||||
);
|
||||
|
||||
self.cache
|
||||
.update_cache(
|
||||
mixnodes,
|
||||
gateways,
|
||||
rewarded_set,
|
||||
active_set,
|
||||
interval_rewarding_params,
|
||||
current_interval,
|
||||
)
|
||||
.await;
|
||||
|
||||
if let Some(notify) = &self.update_rewarded_set_notify {
|
||||
let update_details = self
|
||||
.nymd_client
|
||||
.get_current_rewarded_set_update_details()
|
||||
.await?;
|
||||
|
||||
if update_details.last_refreshed_block + (update_details.refresh_rate_blocks as u64)
|
||||
< update_details.current_height
|
||||
{
|
||||
// there's only ever a single waiter -> the set updater
|
||||
notify.notify_one()
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) async fn run(&self)
|
||||
where
|
||||
C: CosmWasmClient + Sync,
|
||||
{
|
||||
let mut interval = time::interval(self.caching_interval);
|
||||
loop {
|
||||
interval.tick().await;
|
||||
if let Err(err) = self.refresh_cache().await {
|
||||
error!("Failed to refresh validator cache - {}", err);
|
||||
} else {
|
||||
// relaxed memory ordering is fine here. worst case scenario network monitor
|
||||
// will just have to wait for an additional backoff to see the change.
|
||||
// And so this will not really incur any performance penalties by setting it every loop iteration
|
||||
self.cache.initialised.store(true, Ordering::Relaxed)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl ValidatorCache {
|
||||
fn new() -> Self {
|
||||
ValidatorCache {
|
||||
initialised: Arc::new(AtomicBool::new(false)),
|
||||
inner: Arc::new(RwLock::new(ValidatorCacheInner::new())),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn stage() -> AdHoc {
|
||||
AdHoc::on_ignite("Validator Cache Stage", |rocket| async {
|
||||
rocket.manage(Self::new()).mount(
|
||||
// this format! is so ugly...
|
||||
format!("/{}", VALIDATOR_API_VERSION),
|
||||
routes![
|
||||
routes::get_mixnodes,
|
||||
routes::get_gateways,
|
||||
routes::get_active_set,
|
||||
routes::get_rewarded_set,
|
||||
],
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
async fn update_cache(
|
||||
&self,
|
||||
mixnodes: Vec<MixNodeBond>,
|
||||
gateways: Vec<GatewayBond>,
|
||||
rewarded_set: Vec<MixNodeBond>,
|
||||
active_set: Vec<MixNodeBond>,
|
||||
interval_rewarding_params: IntervalRewardParams,
|
||||
current_interval: Interval,
|
||||
) {
|
||||
let mut inner = self.inner.write().await;
|
||||
|
||||
inner.mixnodes.update(mixnodes);
|
||||
inner.gateways.update(gateways);
|
||||
inner.rewarded_set.update(rewarded_set);
|
||||
inner.active_set.update(active_set);
|
||||
inner
|
||||
.current_reward_params
|
||||
.update(interval_rewarding_params);
|
||||
inner.current_interval.update(current_interval);
|
||||
}
|
||||
|
||||
pub async fn mixnodes(&self) -> Cache<Vec<MixNodeBond>> {
|
||||
self.inner.read().await.mixnodes.clone()
|
||||
}
|
||||
|
||||
pub async fn gateways(&self) -> Cache<Vec<GatewayBond>> {
|
||||
self.inner.read().await.gateways.clone()
|
||||
}
|
||||
|
||||
pub async fn rewarded_set(&self) -> Cache<Vec<MixNodeBond>> {
|
||||
self.inner.read().await.rewarded_set.clone()
|
||||
}
|
||||
|
||||
pub async fn active_set(&self) -> Cache<Vec<MixNodeBond>> {
|
||||
self.inner.read().await.active_set.clone()
|
||||
}
|
||||
|
||||
pub(crate) async fn interval_reward_params(&self) -> Cache<IntervalRewardParams> {
|
||||
self.inner.read().await.current_reward_params.clone()
|
||||
}
|
||||
|
||||
pub(crate) async fn current_interval(&self) -> Cache<Interval> {
|
||||
self.inner.read().await.current_interval.clone()
|
||||
}
|
||||
|
||||
pub async fn mixnode_details(
|
||||
&self,
|
||||
identity: IdentityKeyRef<'_>,
|
||||
) -> (Option<MixNodeBond>, MixnodeStatus) {
|
||||
// it might not be the most optimal to possibly iterate the entire vector to find (or not)
|
||||
// the relevant value. However, the vectors are relatively small (< 10_000 elements, < 1000 for active set)
|
||||
|
||||
let active_set = &self.inner.read().await.active_set.value;
|
||||
if let Some(bond) = active_set
|
||||
.iter()
|
||||
.find(|mix| mix.mix_node.identity_key == identity)
|
||||
{
|
||||
return (Some(bond.clone()), MixnodeStatus::Active);
|
||||
}
|
||||
|
||||
let rewarded_set = &self.inner.read().await.rewarded_set.value;
|
||||
if let Some(bond) = rewarded_set
|
||||
.iter()
|
||||
.find(|mix| mix.mix_node.identity_key == identity)
|
||||
{
|
||||
return (Some(bond.clone()), MixnodeStatus::Standby);
|
||||
}
|
||||
|
||||
let all_bonded = &self.inner.read().await.mixnodes.value;
|
||||
if let Some(bond) = all_bonded
|
||||
.iter()
|
||||
.find(|mix| mix.mix_node.identity_key == identity)
|
||||
{
|
||||
(Some(bond.clone()), MixnodeStatus::Inactive)
|
||||
} else {
|
||||
(None, MixnodeStatus::NotFound)
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn mixnode_status(&self, identity: IdentityKey) -> MixnodeStatus {
|
||||
self.mixnode_details(&identity).await.1
|
||||
}
|
||||
|
||||
pub fn initialised(&self) -> bool {
|
||||
self.initialised.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
pub(crate) async fn wait_for_initial_values(&self) {
|
||||
let initialisation_backoff = Duration::from_secs(5);
|
||||
loop {
|
||||
if self.initialised() {
|
||||
break;
|
||||
} else {
|
||||
debug!("Validator cache hasn't been initialised yet - waiting for {:?} before trying again", initialisation_backoff);
|
||||
tokio::time::sleep(initialisation_backoff).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl ValidatorCacheInner {
|
||||
fn new() -> Self {
|
||||
ValidatorCacheInner {
|
||||
mixnodes: Cache::default(),
|
||||
gateways: Cache::default(),
|
||||
rewarded_set: Cache::default(),
|
||||
active_set: Cache::default(),
|
||||
current_reward_params: Cache::new(IntervalRewardParams::new_empty()),
|
||||
current_interval: Cache::default(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
use crate::contract_cache::ValidatorCache;
|
||||
use mixnet_contract_common::{GatewayBond, MixNodeBond};
|
||||
use rocket::serde::json::Json;
|
||||
use rocket::State;
|
||||
|
||||
#[get("/mixnodes")]
|
||||
pub(crate) async fn get_mixnodes(cache: &State<ValidatorCache>) -> Json<Vec<MixNodeBond>> {
|
||||
Json(cache.mixnodes().await.value)
|
||||
}
|
||||
|
||||
#[get("/gateways")]
|
||||
pub(crate) async fn get_gateways(cache: &State<ValidatorCache>) -> Json<Vec<GatewayBond>> {
|
||||
Json(cache.gateways().await.value)
|
||||
}
|
||||
|
||||
#[get("/mixnodes/rewarded")]
|
||||
pub(crate) async fn get_rewarded_set(cache: &State<ValidatorCache>) -> Json<Vec<MixNodeBond>> {
|
||||
Json(cache.rewarded_set().await.value)
|
||||
}
|
||||
|
||||
#[get("/mixnodes/active")]
|
||||
pub(crate) async fn get_active_set(cache: &State<ValidatorCache>) -> Json<Vec<MixNodeBond>> {
|
||||
Json(cache.active_set().await.value)
|
||||
}
|
||||
Reference in New Issue
Block a user