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

Refactor to a lazy rewarding system (#1127)

Browse Source 
* Remove eager operator and delegator rewarding

* Add mixnodes snapshoting

* Add Intervals map, getter and setter

* Refactor reward params

* Refactor MixnodeToReward

* Persist node reward params and results on chain

* Update cw-storage-plus to 0.12.1

* Refactor delegation storage

* Compound delegator reward command

* Compound delegator reward command

* Add defered delegate and undelegate

* Compound on behalf command

* Scale calculations to epoch

* Rename interval -> epoch where practical

* Store epochs on chain

* Cleanup first pass

* Adapt reporting to lazy rewarding

* make clippy --all
This commit is contained in:
Drazen Urch
2022-03-09 14:28:16 +01:00
committed by GitHub
parent 379dd1f02b
commit b30f680549
51 changed files with 2510 additions and 2664 deletions
Download Patch File Download Diff File Expand all files Collapse all files
validator-api/src/config/mod.rs
+4
View File
@@ -364,6 +364,8 @@ impl Config {
self.network_monitor.eth_endpoint.clone()
}
// TODO: Remove if still unused
#[allow(dead_code)]
pub fn get_rewarding_enabled(&self) -> bool {
self.rewarding.enabled
}
@@ -438,6 +440,8 @@ impl Config {
self.network_monitor.all_validator_apis.clone()
}
// TODO: Remove if still unused
#[allow(dead_code)]
pub fn get_minimum_interval_monitor_threshold(&self) -> u8 {
self.rewarding.minimum_interval_monitor_threshold
}
validator-api/src/contract_cache/mod.rs
+15 -18
View File
@@ -2,10 +2,10 @@
// 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::reward_params::EpochRewardParams;
use mixnet_contract_common::{
GatewayBond, IdentityKey, IdentityKeyRef, Interval, MixNodeBond, RewardedSetNodeStatus,
};
@@ -23,6 +23,8 @@ use validator_client::nymd::CosmWasmClient;
pub(crate) mod routes;
type Epoch = Interval;
pub struct ValidatorCacheRefresher<C> {
nymd_client: Client<C>,
cache: ValidatorCache,
@@ -43,8 +45,8 @@ struct ValidatorCacheInner {
rewarded_set: Cache<Vec<MixNodeBond>>,
active_set: Cache<Vec<MixNodeBond>>,
current_reward_params: Cache<IntervalRewardParams>,
current_interval: Cache<Interval>,
current_reward_params: Cache<EpochRewardParams>,
current_epoch: Cache<Interval>,
}
fn current_unix_timestamp() -> i64 {
@@ -131,10 +133,7 @@ impl<C> ValidatorCacheRefresher<C> {
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 epoch_rewarding_params = self.nymd_client.get_current_epoch_reward_params().await?;
let current_interval = self.nymd_client.get_current_interval().await?;
info!(
@@ -149,7 +148,7 @@ impl<C> ValidatorCacheRefresher<C> {
gateways,
rewarded_set,
active_set,
interval_rewarding_params,
epoch_rewarding_params,
current_interval,
)
.await;
@@ -219,8 +218,8 @@ impl ValidatorCache {
gateways: Vec<GatewayBond>,
rewarded_set: Vec<MixNodeBond>,
active_set: Vec<MixNodeBond>,
interval_rewarding_params: IntervalRewardParams,
current_interval: Interval,
epoch_rewarding_params: EpochRewardParams,
current_epoch: Epoch,
) {
let mut inner = self.inner.write().await;
@@ -228,10 +227,8 @@ impl ValidatorCache {
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);
inner.current_reward_params.update(epoch_rewarding_params);
inner.current_epoch.update(current_epoch);
}
pub async fn mixnodes(&self) -> Cache<Vec<MixNodeBond>> {
@@ -250,12 +247,12 @@ impl ValidatorCache {
self.inner.read().await.active_set.clone()
}
pub(crate) async fn interval_reward_params(&self) -> Cache<IntervalRewardParams> {
pub(crate) async fn epoch_reward_params(&self) -> Cache<EpochRewardParams> {
self.inner.read().await.current_reward_params.clone()
}
pub(crate) async fn current_interval(&self) -> Cache<Interval> {
self.inner.read().await.current_interval.clone()
self.inner.read().await.current_epoch.clone()
}
pub async fn mixnode_details(
@@ -320,10 +317,10 @@ impl ValidatorCacheInner {
gateways: Cache::default(),
rewarded_set: Cache::default(),
active_set: Cache::default(),
current_reward_params: Cache::new(IntervalRewardParams::new_empty()),
current_reward_params: Cache::new(EpochRewardParams::new_empty()),
// 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_interval: Cache::new(Interval::new(
current_epoch: Cache::new(Interval::new(
u32::MAX,
OffsetDateTime::UNIX_EPOCH,
Duration::default(),
validator-api/src/main.rs
+13 -48
View File
@@ -9,7 +9,6 @@ use crate::contract_cache::ValidatorCacheRefresher;
use crate::network_monitor::NetworkMonitorBuilder;
use crate::node_status_api::uptime_updater::HistoricalUptimeUpdater;
use crate::nymd_client::Client;
use crate::rewarding::Rewarder;
use crate::storage::ValidatorApiStorage;
use ::config::NymConfig;
use anyhow::Result;
@@ -25,8 +24,8 @@ use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Notify;
use url::Url;
use validator_client::nymd::SigningNymdClient;
use validator_client::ValidatorClientError;
// use validator_client::nymd::SigningNymdClient;
// use validator_client::ValidatorClientError;
use crate::rewarded_set_updater::RewardedSetUpdater;
#[cfg(feature = "coconut")]
@@ -38,7 +37,6 @@ mod network_monitor;
mod node_status_api;
pub(crate) mod nymd_client;
mod rewarded_set_updater;
mod rewarding;
pub(crate) mod storage;
#[cfg(feature = "coconut")]
@@ -358,6 +356,8 @@ fn setup_network_monitor<'a>(
))
}
// TODO: Remove if still unused
#[allow(dead_code)]
fn expected_monitor_test_runs(config: &Config, interval_length: Duration) -> usize {
let test_delay = config.get_network_monitor_run_interval();
@@ -366,32 +366,6 @@ fn expected_monitor_test_runs(config: &Config, interval_length: Duration) -> usi
(interval_length.as_secs() / test_delay.as_secs()) as usize
}
async fn setup_rewarder(
config: &Config,
rocket: &Rocket<Ignite>,
nymd_client: &Client<SigningNymdClient>,
) -> Result<Option<Rewarder>, ValidatorClientError> {
if config.get_rewarding_enabled() && config.get_network_monitor_enabled() {
// get instances of managed states
let node_status_storage = rocket.state::<ValidatorApiStorage>().unwrap().clone();
let validator_cache = rocket.state::<ValidatorCache>().unwrap().clone();
let rewarding_interval_length = nymd_client.get_current_interval().await?.length();
Ok(Some(Rewarder::new(
nymd_client.clone(),
validator_cache,
node_status_storage,
expected_monitor_test_runs(config, rewarding_interval_length),
config.get_minimum_interval_monitor_threshold(),
)))
} else if config.get_rewarding_enabled() {
warn!("Cannot enable rewarding with the network monitor being disabled");
Ok(None)
} else {
Ok(None)
}
}
async fn setup_rocket(config: &Config, liftoff_notify: Arc<Notify>) -> Result<Rocket<Ignite>> {
// let's build our rocket!
let rocket = rocket::build()
@@ -482,27 +456,18 @@ async fn run_validator_api(matches: ArgMatches<'static>) -> Result<()> {
// setup our daily uptime updater. Note that if network monitor is disabled, then we have
// no data for the updates and hence we don't need to start it up
let storage = rocket.state::<ValidatorApiStorage>().unwrap().clone();
let uptime_updater = HistoricalUptimeUpdater::new(storage);
let uptime_updater = HistoricalUptimeUpdater::new(storage.clone());
tokio::spawn(async move { uptime_updater.run().await });
if let Some(rewarder) = setup_rewarder(&config, &rocket, &nymd_client).await? {
info!("Periodic rewarding is starting...");
let mut rewarded_set_updater = RewardedSetUpdater::new(
nymd_client,
rewarded_set_update_notify,
validator_cache.clone(),
storage,
);
let rewarded_set_updater = RewardedSetUpdater::new(
nymd_client.clone(),
rewarded_set_update_notify,
validator_cache.clone(),
);
// spawn rewarded set updater
tokio::spawn(async move { rewarded_set_updater.run().await });
// only update rewarded set if we're also distributing rewards
tokio::spawn(async move { rewarder.run().await });
} else {
info!("Periodic rewarding is disabled.");
}
// spawn rewarded set updater
tokio::spawn(async move { rewarded_set_updater.run().await.unwrap() });
} else {
let nymd_client = Client::new_query(&config);
let validator_cache_refresher = ValidatorCacheRefresher::new(
validator-api/src/node_status_api/routes.rs
+11 -9
View File
@@ -7,6 +7,7 @@ use crate::node_status_api::models::{
};
use crate::storage::ValidatorApiStorage;
use crate::ValidatorCache;
use mixnet_contract_common::reward_params::{NodeRewardParams, RewardParams};
use rocket::http::Status;
use rocket::serde::json::Json;
use rocket::State;
@@ -115,7 +116,7 @@ pub(crate) async fn get_mixnode_reward_estimation(
) -> Result<Json<RewardEstimationResponse>, ErrorResponse> {
let (bond, status) = cache.mixnode_details(&identity).await;
if let Some(bond) = bond {
let interval_reward_params = cache.interval_reward_params().await;
let interval_reward_params = cache.epoch_reward_params().await;
let as_at = interval_reward_params.timestamp();
let interval_reward_params = interval_reward_params.into_inner();
@@ -128,8 +129,9 @@ pub(crate) async fn get_mixnode_reward_estimation(
)
.await
.map_err(|err| ErrorResponse::new(err.to_string(), Status::NotFound))?;
match interval_reward_params.estimate_reward(&bond, uptime.u8(), status.is_active()) {
let node_reward_params = NodeRewardParams::new(0, uptime.u8() as u128, status.is_active());
let reward_params = RewardParams::new(interval_reward_params, node_reward_params);
match bond.estimate_reward(&reward_params) {
Ok((
estimated_total_node_reward,
estimated_operator_reward,
@@ -166,13 +168,13 @@ pub(crate) async fn get_mixnode_stake_saturation(
) -> Result<Json<StakeSaturationResponse>, ErrorResponse> {
let (bond, _) = cache.mixnode_details(&identity).await;
if let Some(bond) = bond {
let interval_reward_params = cache.interval_reward_params().await;
let interval_reward_params = cache.epoch_reward_params().await;
let as_at = interval_reward_params.timestamp();
let interval_reward_params = interval_reward_params.into_inner();
let saturation = bond.stake_saturation(
interval_reward_params.circulating_supply,
interval_reward_params.rewarded_set_size,
interval_reward_params.circulating_supply(),
interval_reward_params.rewarded_set_size() as u32,
);
Ok(Json(StakeSaturationResponse {
@@ -200,9 +202,9 @@ pub(crate) async fn get_mixnode_inclusion_probability(
.fold(0u128, |acc, x| acc + x.total_bond().unwrap_or_default())
as f64;
let rewarding_params = cache.interval_reward_params().await.into_inner();
let rewarded_set_size = rewarding_params.rewarded_set_size as f64;
let active_set_size = rewarding_params.active_set_size as f64;
let rewarding_params = cache.epoch_reward_params().await.into_inner();
let rewarded_set_size = rewarding_params.rewarded_set_size() as f64;
let active_set_size = rewarding_params.active_set_size() as f64;
let prob_one_draw =
target_mixnode.total_bond().unwrap_or_default() as f64 / total_bonded_tokens;
validator-api/src/nymd_client.rs
+45 -78
View File
@@ -2,12 +2,12 @@
// SPDX-License-Identifier: Apache-2.0
use crate::config::Config;
use crate::rewarding::{error::RewardingError, IntervalRewardParams, MixnodeToReward};
use crate::rewarded_set_updater::{error::RewardingError, MixnodeToReward};
use config::defaults::{DEFAULT_NETWORK, DEFAULT_VALIDATOR_API_PORT};
use mixnet_contract_common::{
ContractStateParams, Delegation, ExecuteMsg, GatewayBond, IdentityKey, Interval, MixNodeBond,
MixnodeRewardingStatusResponse, RewardedSetNodeStatus, RewardedSetUpdateDetails,
MIXNODE_DELEGATORS_PAGE_LIMIT,
reward_params::EpochRewardParams, ContractStateParams, Delegation, ExecuteMsg, GatewayBond,
IdentityKey, Interval, MixNodeBond, MixnodeRewardingStatusResponse, RewardedSetNodeStatus,
RewardedSetUpdateDetails,
};
use serde::Serialize;
use std::sync::Arc;
@@ -95,6 +95,7 @@ impl Client<SigningNymdClient> {
impl<C> Client<C> {
// a helper function for the future to obtain the current block timestamp
#[allow(dead_code)]
pub(crate) async fn current_block_timestamp(
&self,
) -> Result<TendermintTime, ValidatorClientError>
@@ -145,9 +146,9 @@ impl<C> Client<C> {
self.0.read().await.get_contract_settings().await
}
pub(crate) async fn get_current_interval_reward_params(
pub(crate) async fn get_current_epoch_reward_params(
&self,
) -> Result<IntervalRewardParams, ValidatorClientError>
) -> Result<EpochRewardParams, ValidatorClientError>
where
C: CosmWasmClient + Sync,
{
@@ -157,19 +158,20 @@ impl<C> Client<C> {
let reward_pool = this.get_reward_pool().await?;
let interval_reward_percent = this.get_interval_reward_percent().await?;
let interval_reward_params = IntervalRewardParams {
reward_pool,
circulating_supply: this.get_circulating_supply().await?,
sybil_resistance_percent: this.get_sybil_resistance_percent().await?,
rewarded_set_size: state.mixnode_rewarded_set_size,
active_set_size: state.mixnode_active_set_size,
period_reward_pool: (reward_pool / 100) * interval_reward_percent as u128,
active_set_work_factor: this.get_active_set_work_factor().await?,
};
let epoch_reward_params = EpochRewardParams::new(
(reward_pool / 100 / this.get_epochs_in_interval().await? as u128)
* interval_reward_percent as u128,
state.mixnode_rewarded_set_size as u128,
state.mixnode_active_set_size as u128,
this.get_circulating_supply().await?,
this.get_sybil_resistance_percent().await?,
this.get_active_set_work_factor().await?,
);
Ok(interval_reward_params)
Ok(epoch_reward_params)
}
#[allow(dead_code)]
pub(crate) async fn get_rewarding_status(
&self,
mix_identity: mixnet_contract_common::IdentityKey,
@@ -207,6 +209,7 @@ impl<C> Client<C> {
Ok(hash)
}
#[allow(dead_code)]
pub(crate) async fn get_mixnode_delegations(
&self,
identity: IdentityKey,
@@ -247,6 +250,7 @@ impl<C> Client<C> {
.await
}
#[allow(dead_code)]
pub(crate) async fn advance_current_interval(&self) -> Result<(), ValidatorClientError>
where
C: SigningCosmWasmClient + Sync,
@@ -255,6 +259,30 @@ impl<C> Client<C> {
Ok(())
}
pub(crate) async fn advance_current_epoch(&self) -> Result<(), ValidatorClientError>
where
C: SigningCosmWasmClient + Sync,
{
self.0.write().await.nymd.advance_current_epoch().await?;
Ok(())
}
pub(crate) async fn checkpoint_mixnodes(&self) -> Result<(), ValidatorClientError>
where
C: SigningCosmWasmClient + Sync,
{
self.0.write().await.nymd.checkpoint_mixnodes().await?;
Ok(())
}
pub(crate) async fn reconcile_delegations(&self) -> Result<(), ValidatorClientError>
where
C: SigningCosmWasmClient + Sync,
{
self.0.write().await.nymd.reconcile_delegations().await?;
Ok(())
}
pub(crate) async fn write_rewarded_set(
&self,
rewarded_set: Vec<IdentityKey>,
@@ -272,68 +300,7 @@ impl<C> Client<C> {
Ok(())
}
pub(crate) async fn reward_mixnode_and_all_delegators(
&self,
node: &MixnodeToReward,
interval_id: u32,
) -> Result<(), RewardingError>
where
C: SigningCosmWasmClient + Sync,
{
// start with the base call to reward operator and first page of delegators
let msgs = vec![(node.to_reward_execute_msg(interval_id), vec![])];
let memo = format!(
"operator + {} delegators rewarding",
MIXNODE_DELEGATORS_PAGE_LIMIT
);
self.execute_multiple_with_retry(msgs, Default::default(), memo)
.await?;
// reward rest of delegators (if applicable)
if node.total_delegations > MIXNODE_DELEGATORS_PAGE_LIMIT {
// the first 'batch' of delegators has been rewarded while rewarding the operator
let mut remaining_delegators = node.total_delegations - MIXNODE_DELEGATORS_PAGE_LIMIT;
let delegator_rewarding_msg = (
node.to_next_delegator_reward_execute_msg(interval_id),
vec![],
);
while remaining_delegators > 0 {
let delegators_in_call = remaining_delegators.min(MIXNODE_DELEGATORS_PAGE_LIMIT);
let msgs = vec![delegator_rewarding_msg.clone()];
let memo = format!("rewarding another {} delegators", delegators_in_call);
self.execute_multiple_with_retry(msgs, Default::default(), memo)
.await?;
remaining_delegators =
remaining_delegators.saturating_sub(MIXNODE_DELEGATORS_PAGE_LIMIT)
}
}
Ok(())
}
pub(crate) async fn reward_mix_delegators(
&self,
node: &MixnodeToReward,
interval_id: u32,
) -> Result<(), RewardingError>
where
C: SigningCosmWasmClient + Sync,
{
// the fee is a tricky subject here because we don't know exactly how many delegators we missed,
// let's aim for the worst case scenario and assume it was the entire page
let delegator_rewarding_msg = (
node.to_next_delegator_reward_execute_msg(interval_id),
vec![],
);
let memo = "rewarding delegators".to_string();
self.execute_multiple_with_retry(vec![delegator_rewarding_msg], Default::default(), memo)
.await
}
pub(crate) async fn reward_mixnodes_with_single_page_of_delegators(
pub(crate) async fn reward_mixnodes(
&self,
nodes: &[MixnodeToReward],
interval_id: u32,
validator-api/src/rewarding/error.rs → validator-api/src/rewarded_set_updater/error.rs
+3 -4
View File
@@ -7,13 +7,12 @@ use validator_client::nymd::error::NymdError;
use validator_client::ValidatorClientError;
#[derive(Debug, Error)]
pub(crate) enum RewardingError {
pub enum RewardingError {
#[error("Could not distribute rewards as the contract address was unspecified")]
UnspecifiedContractAddress,
#[error("There were no mixnodes to reward (network is dead)")]
NoMixnodesToReward,
// #[error("There were no mixnodes to reward (network is dead)")]
// NoMixnodesToReward,
#[error("Failed to execute the smart contract - {0}")]
ContractExecutionFailure(NymdError),
validator-api/src/rewarded_set_updater/mod.rs
+273 -9
View File
@@ -14,17 +14,66 @@
use crate::contract_cache::ValidatorCache;
use crate::nymd_client::Client;
use mixnet_contract_common::{IdentityKey, MixNodeBond};
use crate::storage::models::{
FailedMixnodeRewardChunk, PossiblyUnrewardedMixnode, RewardingReport,
};
use crate::storage::ValidatorApiStorage;
use mixnet_contract_common::reward_params::NodeRewardParams;
use mixnet_contract_common::ExecuteMsg;
use mixnet_contract_common::{IdentityKey, Interval, MixNodeBond};
use rand::prelude::SliceRandom;
use rand::rngs::OsRng;
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
use time::OffsetDateTime;
use tokio::sync::Notify;
use tokio::time::sleep;
use validator_client::nymd::SigningNymdClient;
pub(crate) mod error;
use error::RewardingError;
#[derive(Debug, Clone)]
pub(crate) struct MixnodeToReward {
pub(crate) identity: IdentityKey,
/// Total number of individual addresses that have delegated to this particular node
// pub(crate) total_delegations: usize,
/// Node absolute uptime over total active set uptime
pub(crate) params: NodeRewardParams,
}
impl MixnodeToReward {
fn params(&self) -> NodeRewardParams {
self.params
}
pub(crate) fn to_reward_execute_msg(&self, interval_id: u32) -> ExecuteMsg {
ExecuteMsg::RewardMixnode {
identity: self.identity.clone(),
params: self.params(),
interval_id,
}
}
}
pub(crate) struct FailedMixnodeRewardChunkDetails {
possibly_unrewarded: Vec<MixnodeToReward>,
error_message: String,
}
// Epoch has all the same semantics as interval, but has a lower set duration
type Epoch = Interval;
pub struct RewardedSetUpdater {
nymd_client: Client<SigningNymdClient>,
update_rewarded_set_notify: Arc<Notify>,
validator_cache: ValidatorCache,
storage: ValidatorApiStorage,
epoch: Epoch,
}
impl RewardedSetUpdater {
@@ -32,11 +81,15 @@ impl RewardedSetUpdater {
nymd_client: Client<SigningNymdClient>,
update_rewarded_set_notify: Arc<Notify>,
validator_cache: ValidatorCache,
storage: ValidatorApiStorage,
) -> Self {
let epoch = Epoch::new(0, OffsetDateTime::now_utc(), Duration::from_secs(3600));
RewardedSetUpdater {
nymd_client,
update_rewarded_set_notify,
validator_cache,
storage,
epoch,
}
}
@@ -75,17 +128,220 @@ impl RewardedSetUpdater {
.collect()
}
async fn update_rewarded_set(&self) {
async fn rewarding_happened_at_epoch(&self) -> Result<bool, RewardingError> {
if let Some(entry) = self
.storage
.get_interval_rewarding_entry(self.epoch.start_unix_timestamp())
.await?
{
// log error if the attempt wasn't finished. This error implies the process has crashed
// during the rewards distribution
if !entry.finished {
error!(
"It seems that we haven't successfully finished distributing rewards at {}",
self.epoch
)
}
Ok(true)
} else {
Ok(false)
}
}
async fn reward_current_rewarded_set(&self) -> Result<(), RewardingError> {
let to_reward = self.nodes_to_reward().await?;
self.storage
.insert_started_epoch_rewarding(self.epoch)
.await?;
let failure_data = self.distribute_rewards(&to_reward, false).await;
let mut rewarding_report = RewardingReport {
interval_rewarding_id: self.epoch.id() as i64,
eligible_mixnodes: to_reward.len() as i64,
possibly_unrewarded_mixnodes: 0,
};
if let Some(failure_data) = failure_data {
rewarding_report.possibly_unrewarded_mixnodes =
failure_data.possibly_unrewarded.len() as i64;
if let Err(err) = self
.save_failure_information(failure_data, self.epoch.id() as i64)
.await
{
error!("failed to save information about rewarding failures!");
// TODO: should we just terminate the process here?
return Err(err);
}
}
self.storage
.finish_rewarding_interval_and_insert_report(rewarding_report)
.await?;
Ok(())
}
async fn save_failure_information(
&self,
failed_chunk: FailedMixnodeRewardChunkDetails,
interval_rewarding_id: i64,
) -> Result<(), RewardingError> {
// save the chunk
let chunk_id = self
.storage
.insert_failed_mixnode_reward_chunk(FailedMixnodeRewardChunk {
interval_rewarding_id,
error_message: failed_chunk.error_message,
})
.await?;
// and then all associated nodes
for node in failed_chunk.possibly_unrewarded.into_iter() {
self.storage
.insert_possibly_unrewarded_mixnode(PossiblyUnrewardedMixnode {
chunk_id,
identity: node.identity,
uptime: node.params.uptime() as u8,
})
.await?;
}
Ok(())
}
async fn distribute_rewards(
&self,
eligible_mixnodes: &[MixnodeToReward],
retry: bool,
) -> Option<FailedMixnodeRewardChunkDetails> {
if retry {
info!(
"Attempting to retry rewarding {} mixnodes...",
eligible_mixnodes.len()
)
} else {
info!(
"Attempting to reward {} mixnodes...",
eligible_mixnodes.len()
)
}
let mut failed_chunks = None;
let num_retries = 5;
let mut retry = 0;
let mut success = false;
loop {
match self
.nymd_client
.reward_mixnodes(eligible_mixnodes, self.epoch.id())
.await
{
Ok(_) => {
let total_rewarded = eligible_mixnodes.len();
info!("Rewarded {} mixnodes", total_rewarded);
success = false;
break;
}
Err(err) => {
if num_retries <= retry {
break;
}
retry += 1;
// this is a super weird edge case that we didn't catch change to sequence and
// resent rewards unnecessarily, but the mempool saved us from executing it again
// however, still we want to wait until we're sure we're into the next block
if !err.is_tendermint_duplicate() {
error!("failed to reward mixnodes... - {}", err);
failed_chunks = Some(FailedMixnodeRewardChunkDetails {
possibly_unrewarded: eligible_mixnodes.to_vec(),
error_message: err.to_string(),
});
}
sleep(Duration::from_secs(11)).await;
}
}
}
// Its all or nothing since we do not chunk
if success {
failed_chunks = None
}
failed_chunks
}
async fn nodes_to_reward(&self) -> Result<Vec<MixnodeToReward>, RewardingError> {
let active_set = self
.validator_cache
.active_set()
.await
.into_inner()
.into_iter()
.map(|bond| bond.mix_node.identity_key)
.collect::<HashSet<_>>();
let rewarded_set = self.validator_cache.rewarded_set().await.into_inner();
let mut eligible_nodes = Vec::with_capacity(rewarded_set.len());
for rewarded_node in rewarded_set.into_iter() {
let uptime = self
.storage
.get_average_mixnode_uptime_in_interval(
rewarded_node.identity(),
self.epoch.start_unix_timestamp(),
self.epoch.end_unix_timestamp(),
)
.await?;
let node_reward_params = NodeRewardParams::new(
0,
uptime.u8().into(),
active_set.contains(rewarded_node.identity()),
);
eligible_nodes.push(MixnodeToReward {
identity: rewarded_node.identity().clone(),
params: node_reward_params,
})
}
Ok(eligible_nodes)
}
// This is where the epoch gets advanced, and all epoch related transactions originate
async fn update_rewarded_set(&self) -> Result<(), RewardingError> {
// 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 rewarding_params = self
let epoch_reward_params = self
.validator_cache
.interval_reward_params()
.epoch_reward_params()
.await
.into_inner();
let rewarded_set_size = rewarding_params.rewarded_set_size;
let active_set_size = rewarding_params.active_set_size;
// Reward all the nodes in the still current, soon to be previous rewarded set
if !self.rewarding_happened_at_epoch().await? {
self.reward_current_rewarded_set().await?;
}
// Reconcile delegations from the previous epoch
if let Err(err) = self.nymd_client.reconcile_delegations().await {
log::error!("failed to reconcile delegations - {}", err);
}
// Snapshot mixnodes for the next epoch
if let Err(err) = self.nymd_client.checkpoint_mixnodes().await {
log::error!("failed to checkpoint mixnodes - {}", err);
}
// Snapshot mixnodes for the next epoch
if let Err(err) = self.nymd_client.advance_current_epoch().await {
log::error!("failed to advance_epoch - {}", err);
}
let rewarded_set_size = epoch_reward_params.rewarded_set_size() as u32;
let active_set_size = epoch_reward_params.active_set_size() as u32;
// note that top k nodes are in the active set
let new_rewarded_set = self.determine_rewarded_set(all_nodes, rewarded_set_size);
@@ -94,19 +350,27 @@ impl RewardedSetUpdater {
.write_rewarded_set(new_rewarded_set, active_set_size)
.await
{
log::error!("failed to update the rewarded set - {}", err)
log::error!("failed to update the rewarded set - {}", err);
// note that if the transaction failed to get executed because, I don't know, there was a networking hiccup
// the cache will notify the updater on its next round
}
let cutoff = (self.epoch.end() - Duration::from_secs(86400)).unix_timestamp();
self.storage.purge_old_statuses(cutoff).await?;
Ok(())
}
pub(crate) async fn run(&self) {
pub(crate) async fn run(&mut self) -> Result<(), RewardingError> {
self.validator_cache.wait_for_initial_values().await;
loop {
// wait until the cache refresher determined its time to update the rewarded/active sets
self.update_rewarded_set_notify.notified().await;
self.update_rewarded_set().await;
self.epoch = self.epoch.next();
self.update_rewarded_set().await?;
}
#[allow(unreachable_code)]
Ok(())
}
}
validator-api/src/rewarding/mod.rs
-835
View File
@@ -1,835 +0,0 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::contract_cache::ValidatorCache;
use crate::node_status_api::ONE_DAY;
use crate::nymd_client::Client;
use crate::rewarding::error::RewardingError;
use crate::storage::models::{
FailedMixnodeRewardChunk, PossiblyUnrewardedMixnode, RewardingReport,
};
use crate::storage::ValidatorApiStorage;
use config::defaults::DEFAULT_NETWORK;
use log::{error, info};
use mixnet_contract_common::mixnode::NodeRewardParams;
use mixnet_contract_common::{
ExecuteMsg, IdentityKey, Interval, MixNodeBond, RewardingStatus, MIXNODE_DELEGATORS_PAGE_LIMIT,
};
use std::collections::HashSet;
use std::convert::TryInto;
use std::process;
use std::time::Duration;
use time::OffsetDateTime;
use tokio::time::sleep;
use validator_client::nymd::SigningNymdClient;
pub(crate) mod error;
#[derive(Copy, Clone)]
pub(crate) struct IntervalRewardParams {
pub(crate) reward_pool: u128,
pub(crate) circulating_supply: u128,
pub(crate) sybil_resistance_percent: u8,
pub(crate) rewarded_set_size: u32,
pub(crate) active_set_size: u32,
pub(crate) period_reward_pool: u128,
pub(crate) active_set_work_factor: u8,
}
impl IntervalRewardParams {
// technically it's identical to what would have been derived with a Default implementation,
// however, I prefer to be explicit about it, as a `Default::default` value makes no sense
// apart from the `ValidatorCacheInner` context, where this value is not going to be touched anyway
// (it's guarded behind an `initialised` flag)
pub(crate) fn new_empty() -> Self {
IntervalRewardParams {
reward_pool: 0,
circulating_supply: 0,
sybil_resistance_percent: 0,
rewarded_set_size: 0,
active_set_size: 0,
period_reward_pool: 0,
active_set_work_factor: 0,
}
}
pub(crate) fn estimate_reward(
&self,
node: &MixNodeBond,
uptime: u8,
in_active_set: bool,
) -> Result<(u64, u64, u64), RewardingError> {
let node_reward_params = NodeRewardParams::new(
self.period_reward_pool,
self.rewarded_set_size.into(),
self.active_set_size.into(),
0,
self.circulating_supply,
uptime.into(),
self.sybil_resistance_percent,
in_active_set,
self.active_set_work_factor,
);
let total_node_reward = node.reward(&node_reward_params);
let operator_reward = node.operator_reward(&node_reward_params);
let delegators_reward =
node.reward_delegation(node.total_delegation().amount, &node_reward_params);
Ok((
total_node_reward
.reward()
.checked_to_num::<u128>()
.unwrap_or_default()
.try_into()?,
operator_reward.try_into()?,
delegators_reward.try_into()?,
))
}
}
#[derive(Debug, Clone)]
pub(crate) struct MixnodeToReward {
pub(crate) identity: IdentityKey,
/// Total number of individual addresses that have delegated to this particular node
pub(crate) total_delegations: usize,
/// Node absolute uptime over total active set uptime
pub(crate) params: NodeRewardParams,
}
impl MixnodeToReward {
fn params(&self) -> NodeRewardParams {
self.params
}
}
impl MixnodeToReward {
pub(crate) fn to_reward_execute_msg(&self, interval_id: u32) -> ExecuteMsg {
ExecuteMsg::RewardMixnode {
identity: self.identity.clone(),
params: self.params(),
interval_id,
}
}
pub(crate) fn to_next_delegator_reward_execute_msg(&self, interval_id: u32) -> ExecuteMsg {
ExecuteMsg::RewardNextMixDelegators {
mix_identity: self.identity.clone(),
interval_id,
}
}
}
pub(crate) struct FailedMixnodeRewardChunkDetails {
possibly_unrewarded: Vec<MixnodeToReward>,
error_message: String,
}
#[derive(Default)]
pub(crate) struct FailureData {
mixnodes: Option<Vec<FailedMixnodeRewardChunkDetails>>,
}
pub(crate) struct Rewarder {
nymd_client: Client<SigningNymdClient>,
validator_cache: ValidatorCache,
storage: ValidatorApiStorage,
/// Ideal world, expected number of network monitor test runs per interval.
/// In reality it will be slightly lower due to network delays, but it's good enough
/// for estimations regarding percentage of available data for reward distribution.
expected_interval_monitor_runs: usize,
/// Minimum percentage of network monitor test runs reports required in order to distribute
/// rewards.
minimum_interval_monitor_threshold: u8,
}
impl Rewarder {
pub(crate) fn new(
nymd_client: Client<SigningNymdClient>,
validator_cache: ValidatorCache,
storage: ValidatorApiStorage,
expected_interval_monitor_runs: usize,
minimum_interval_monitor_threshold: u8,
) -> Self {
Rewarder {
nymd_client,
validator_cache,
storage,
expected_interval_monitor_runs,
minimum_interval_monitor_threshold,
}
}
/// Obtains the current number of delegators that have delegated their stake towards this particular mixnode.
///
/// # Arguments
///
/// * `identity`: identity key of the mixnode
async fn get_mixnode_delegators_count(
&self,
identity: IdentityKey,
) -> Result<usize, RewardingError> {
Ok(self
.nymd_client
.get_mixnode_delegations(identity)
.await?
.len())
}
/// Obtain the list of current 'rewarded' set, determine their uptime in the provided interval
/// and attach information required for rewarding.
///
/// The method also obtains the number of delegators towards the node in order to more accurately
/// approximate the required gas fees when distributing the rewards.
///
/// # Arguments
///
/// * `interval`: current rewarding interval
async fn determine_eligible_mixnodes(
&self,
interval: Interval,
) -> Result<Vec<MixnodeToReward>, RewardingError> {
let interval_reward_params = self
.nymd_client
.get_current_interval_reward_params()
.await?;
info!("Rewarding pool stats");
info!(
"-- Reward pool: {} {}",
interval_reward_params.reward_pool,
DEFAULT_NETWORK.denom()
);
info!(
"---- Interval reward pool: {} {}",
interval_reward_params.period_reward_pool,
DEFAULT_NETWORK.denom()
);
info!(
"-- Circulating supply: {} {}",
interval_reward_params.circulating_supply,
DEFAULT_NETWORK.denom()
);
// 1. get list of all currently bonded nodes
// 2. for each of them determine their delegator count
// 3. for each of them determine their uptime for the interval
// 4. for each of them determine if they're currently in the active set
// TODO: step 4 will definitely need to change.
let all_nodes = self.validator_cache.mixnodes().await.into_inner();
let active_set = self
.validator_cache
.active_set()
.await
.into_inner()
.into_iter()
.map(|bond| bond.mix_node.identity_key)
.collect::<HashSet<_>>();
let mut nodes_with_delegations = Vec::with_capacity(all_nodes.len());
for node in all_nodes {
let delegator_count = self
.get_mixnode_delegators_count(node.mix_node.identity_key.clone())
.await?;
nodes_with_delegations.push((node, delegator_count));
}
let mut eligible_nodes = Vec::with_capacity(nodes_with_delegations.len());
for (rewarded_node, total_delegations) in nodes_with_delegations.into_iter() {
let uptime = self
.storage
.get_average_mixnode_uptime_in_interval(
rewarded_node.identity(),
interval.start_unix_timestamp(),
interval.end_unix_timestamp(),
)
.await?;
eligible_nodes.push(MixnodeToReward {
identity: rewarded_node.identity().clone(),
total_delegations,
params: NodeRewardParams::new(
interval_reward_params.period_reward_pool,
interval_reward_params.rewarded_set_size.into(),
interval_reward_params.active_set_size.into(),
// Reward blockstamp gets set in the contract call
0,
interval_reward_params.circulating_supply,
uptime.u8().into(),
interval_reward_params.sybil_resistance_percent,
active_set.contains(rewarded_node.identity()),
interval_reward_params.active_set_work_factor,
),
})
}
Ok(eligible_nodes)
}
/// Check whether every node, and their delegators, on the provided list were fully rewarded
/// in the specified interval.
///
/// It is used to deal with edge cases such that mixnode had exactly full page of delegations and
/// somebody created a new delegation thus causing the "last" delegator to possibly be pushed
/// onto the next page that the validator API was not aware of.
///
/// * `eligible_mixnodes`: list of the nodes that were eligible to receive rewards.
/// * `interval_id`: nonce associated with the current rewarding interval
async fn verify_rewarding_completion(
&self,
eligible_mixnodes: &[MixnodeToReward],
current_rewarding_nonce: u32,
) -> (Vec<MixnodeToReward>, Vec<MixnodeToReward>) {
let mut unrewarded = Vec::new();
let mut further_delegators_present = Vec::new();
for mix in eligible_mixnodes {
match self
.nymd_client
.get_rewarding_status(mix.identity.clone(), current_rewarding_nonce)
.await
{
Ok(rewarding_status) => match rewarding_status.status {
// that case is super weird, it implies the node hasn't been rewarded at all!
// maybe the transaction timed out twice or something? In any case, we should attempt
// the reward for the final time!
None => unrewarded.push(mix.clone()),
Some(RewardingStatus::PendingNextDelegatorPage(_)) => {
further_delegators_present.push(mix.clone())
}
Some(RewardingStatus::Complete(_)) => {}
},
Err(err) => {
error!(
"failed to query rewarding status of {} - {}",
mix.identity, err
)
}
}
}
(unrewarded, further_delegators_present)
}
// Utility function to print to the stdout rewarding progress
fn print_rewarding_progress(&self, total_rewarded: usize, out_of: usize, is_retry: bool) {
let percentage = total_rewarded as f32 * 100.0 / out_of as f32;
if is_retry {
info!(
"Resent rewarding transaction for {} / {} mixnodes\t{:.2}%",
total_rewarded, out_of, percentage
);
} else {
info!(
"Sent rewarding transaction for {} / {} mixnodes\t{:.2}%",
total_rewarded, out_of, percentage
);
}
}
/// Using the list of mixnodes eligible for rewards, chunks it into pre-defined sized-chunks
/// and gives out the rewards by calling the smart contract.
///
/// Returns an optional vector containing list of chunks that experienced a smart contract
/// execution error during reward distribution. However, it does not necessarily imply they
/// were not rewarded. There are some edge cases where we time out waiting for block to be included
/// yet the transactions went through.
///
/// Only returns errors for problems originating from before smart contract was called, i.e.
/// we know for sure not a single node has been rewarded.
///
/// # Arguments
///
/// * `eligible_mixnodes`: list of the nodes that are eligible to receive rewards.
/// * `interval_id`: nonce associated with the current rewarding interval.
/// * `retry`: flag to indicate whether this is a retry attempt for rewarding particular nodes.
async fn distribute_rewards_to_mixnodes(
&self,
eligible_mixnodes: &[MixnodeToReward],
interval_id: u32,
retry: bool,
) -> Option<Vec<FailedMixnodeRewardChunkDetails>> {
if retry {
info!(
"Attempting to retry rewarding {} mixnodes...",
eligible_mixnodes.len()
)
} else {
info!(
"Attempting to reward {} mixnodes...",
eligible_mixnodes.len()
)
}
let mut failed_chunks = Vec::new();
// construct chunks such that we reward at most MIXNODE_DELEGATORS_PAGE_LIMIT delegators per block
// nodes with > MIXNODE_DELEGATORS_PAGE_LIMIT delegators that have to be treated in a special way,
// because we cannot batch them together
let mut individually_rewarded = Vec::new();
// sets of nodes that together they have < MIXNODE_DELEGATORS_PAGE_LIMIT delegators
let mut batch_rewarded = vec![vec![]];
let mut current_batch_i = 0;
let mut current_batch_total = 0;
// right now put mixes into batches super naively, if it doesn't fit into the current one,
// create a new one.
for mix in eligible_mixnodes {
// if mixnode has uptime of 0, no rewarding will actually happen regardless of number of delegators,
// so we can just batch it with the current batch
if mix.params.uptime() == 0 {
batch_rewarded[current_batch_i].push(mix.clone());
continue;
}
if mix.total_delegations > MIXNODE_DELEGATORS_PAGE_LIMIT {
individually_rewarded.push(mix)
} else if current_batch_total + mix.total_delegations < MIXNODE_DELEGATORS_PAGE_LIMIT {
batch_rewarded[current_batch_i].push(mix.clone());
current_batch_total += mix.total_delegations;
} else {
batch_rewarded.push(vec![mix.clone()]);
current_batch_i += 1;
current_batch_total = 0;
}
}
let mut total_rewarded = 0;
// start rewarding, first the nodes that are dealt with individually, i.e. nodes that
// need to have their own special blocks due to number of delegators
for mix in individually_rewarded {
if let Err(err) = self
.nymd_client
.reward_mixnode_and_all_delegators(mix, interval_id)
.await
{
// this is a super weird edge case that we didn't catch change to sequence and
// resent rewards unnecessarily, but the mempool saved us from executing it again
// however, still we want to wait until we're sure we're into the next block
if !err.is_tendermint_duplicate() {
error!("failed to reward mixnode with all delegators... - {}", err);
failed_chunks.push(FailedMixnodeRewardChunkDetails {
possibly_unrewarded: vec![mix.clone()],
error_message: err.to_string(),
});
}
sleep(Duration::from_secs(11)).await;
}
total_rewarded += 1;
self.print_rewarding_progress(total_rewarded, eligible_mixnodes.len(), retry);
}
// then we move onto the chunks
for mix_chunk in batch_rewarded {
if mix_chunk.is_empty() {
continue;
}
if let Err(err) = self
.nymd_client
.reward_mixnodes_with_single_page_of_delegators(&mix_chunk, interval_id)
.await
{
// this is a super weird edge case that we didn't catch change to sequence and
// resent rewards unnecessarily, but the mempool saved us from executing it again
// however, still we want to wait until we're sure we're into the next block
if !err.is_tendermint_duplicate() {
error!("failed to reward mixnodes... - {}", err);
failed_chunks.push(FailedMixnodeRewardChunkDetails {
possibly_unrewarded: mix_chunk.to_vec(),
error_message: err.to_string(),
});
}
sleep(Duration::from_secs(11)).await;
}
total_rewarded += mix_chunk.len();
self.print_rewarding_progress(total_rewarded, eligible_mixnodes.len(), retry);
}
if failed_chunks.is_empty() {
None
} else {
Some(failed_chunks)
}
}
/// For each mixnode on the list, try to "continue" rewarding its delegators.
/// Note: due to the checks inside the smart contract, it's impossible to accidentally
/// reward the same mixnode (or delegator) twice during particular rewarding interval.
///
/// Realistically if this method is ever called, it will be only done once per node, so there's
/// no need to determine the exact number of missed delegators.
///
/// * `nodes`: mixnodes which delegators did not receive all rewards in this interval.
/// * `interval_id`: nonce associated with the current rewarding interval.
async fn reward_missed_delegators(&self, nodes: &[MixnodeToReward], interval_id: u32) {
let mut total_resent = 0;
for missed_node in nodes {
total_resent += 1;
info!(
"Sending rewarding transaction for missed delegators ({} / {} mixnodes re-checked)",
total_resent,
nodes.len()
);
if let Err(err) = self
.nymd_client
.reward_mix_delegators(missed_node, interval_id)
.await
{
warn!(
"failed to attempt to reward missed delegators of node {} - {}",
missed_node.identity, err
)
}
}
}
/// Using the list of active mixnode and gateways, determine which of them are eligible for
/// rewarding and distribute the rewards.
///
/// # Arguments
///
/// * `interval_rewarding_id`: id of the current interval rewarding as stored in the database.
/// * `interval`: current rewarding interval
async fn distribute_rewards(
&self,
interval_rewarding_database_id: i64,
interval: Interval,
) -> Result<(RewardingReport, Option<FailureData>), RewardingError> {
let mut failure_data = FailureData::default();
let eligible_mixnodes = self.determine_eligible_mixnodes(interval).await?;
if eligible_mixnodes.is_empty() {
return Err(RewardingError::NoMixnodesToReward);
}
let total_eligible = eligible_mixnodes.len();
failure_data.mixnodes = self
.distribute_rewards_to_mixnodes(&eligible_mixnodes, interval.id(), false)
.await;
let mut nodes_to_verify = eligible_mixnodes;
// if there's some underlying networking error or something, don't keep retrying forever
let mut retries_allowed = 5;
loop {
if retries_allowed <= 0 {
break;
}
let (unrewarded, mut pending_delegators) = self
.verify_rewarding_completion(&nodes_to_verify, interval.id())
.await;
if unrewarded.is_empty() && pending_delegators.is_empty() {
// we're all good - everyone got their rewards
break;
}
if !unrewarded.is_empty() {
// no need to save failure data as we already know about those from the very first run
self.distribute_rewards_to_mixnodes(&unrewarded, interval.id(), true)
.await;
}
if !pending_delegators.is_empty() {
self.reward_missed_delegators(&pending_delegators, interval.id())
.await;
}
// no point in verifying EVERYTHING again, just check the nodes that went through retries
nodes_to_verify = unrewarded;
nodes_to_verify.append(&mut pending_delegators);
retries_allowed -= 1;
}
let report = RewardingReport {
interval_rewarding_id: interval_rewarding_database_id,
eligible_mixnodes: total_eligible as i64,
possibly_unrewarded_mixnodes: failure_data
.mixnodes
.as_ref()
.map(|chunks| {
chunks
.iter()
.map(|chunk| chunk.possibly_unrewarded.len())
.sum::<usize>() as i64
})
.unwrap_or_default(),
};
self.nymd_client.advance_current_interval().await?;
if failure_data.mixnodes.is_none() {
Ok((report, None))
} else {
Ok((report, Some(failure_data)))
}
}
/// Saves information about possibly failed rewarding for future manual inspection.
///
/// Currently there is no automated recovery mechanism.
///
/// # Arguments
///
/// * `failure_data`: information regarding nodes that might have not received reward this interval.
///
/// * `interval_rewarding_id`: id of the current interval rewarding.
async fn save_failure_information(
&self,
failure_data: FailureData,
interval_rewarding_id: i64,
) -> Result<(), RewardingError> {
if let Some(failed_mixnode_chunks) = failure_data.mixnodes {
for failed_chunk in failed_mixnode_chunks.into_iter() {
// save the chunk
let chunk_id = self
.storage
.insert_failed_mixnode_reward_chunk(FailedMixnodeRewardChunk {
interval_rewarding_id,
error_message: failed_chunk.error_message,
})
.await?;
// and then all associated nodes
for node in failed_chunk.possibly_unrewarded.into_iter() {
self.storage
.insert_possibly_unrewarded_mixnode(PossiblyUnrewardedMixnode {
chunk_id,
identity: node.identity,
uptime: node.params.uptime() as u8,
})
.await?;
}
}
}
Ok(())
}
/// Determines whether this validator has already distributed rewards for the specified interval
/// so that it wouldn't accidentally attempt to do it again.
///
/// # Arguments
///
/// * `interval`: interval to check
async fn check_if_rewarding_happened_at_interval(
&self,
interval: Interval,
) -> Result<bool, RewardingError> {
if let Some(entry) = self
.storage
.get_interval_rewarding_entry(interval.start_unix_timestamp())
.await?
{
// log error if the attempt wasn't finished. This error implies the process has crashed
// during the rewards distribution
if !entry.finished {
error!(
"It seems that we haven't successfully finished distributing rewards at {}",
interval
)
}
Ok(true)
} else {
Ok(false)
}
}
/// Determines whether the specified interval is eligible for rewards, i.e. it was not rewarded
/// before and we have enough network monitor test data to distribute the rewards based on them.
///
/// # Arguments
///
/// * `interval`: interval to check
async fn check_interval_eligibility(&self, interval: Interval) -> Result<bool, RewardingError> {
if self
.check_if_rewarding_happened_at_interval(interval)
.await?
|| !self.check_for_monitor_data(interval).await?
{
Ok(false)
} else {
// we haven't sent rewards during the interval and we have enough monitor test data
Ok(true)
}
}
/// Distribute rewards to all eligible mixnodes and gateways on the network.
///
/// # Arguments
///
/// * `interval`: current rewarding interval
async fn perform_rewarding(&self, interval: Interval) -> Result<(), RewardingError> {
info!(
"Starting mixnode and gateway rewarding for interval {} ...",
interval
);
// insert information about beginning the procedure (so that if we crash during it,
// we wouldn't attempt to possibly double reward operators)
let interval_rewarding_id = self
.storage
.insert_started_interval_rewarding(
interval.start_unix_timestamp(),
interval.end_unix_timestamp(),
)
.await?;
let (report, failure_data) = self
.distribute_rewards(interval_rewarding_id, interval)
.await?;
self.storage
.finish_rewarding_interval_and_insert_report(report)
.await?;
if let Some(failure_data) = failure_data {
if let Err(err) = self
.save_failure_information(failure_data, interval_rewarding_id)
.await
{
error!("failed to save information about rewarding failures!");
// TODO: should we just terminate the process here?
return Err(err);
}
}
// since we have already performed rewards, purge everything older than the end of this interval
// (+one day of buffer) as we're never going to need it again (famous last words...)
// note that usually end of interval is equal to the current time
let cutoff = (interval.end() - ONE_DAY).unix_timestamp();
self.storage.purge_old_statuses(cutoff).await?;
Ok(())
}
/// Checks whether there is enough network monitor test run data to distribute rewards
/// for the specified interval.
///
/// # Arguments
///
/// * `interval`: interval to check
async fn check_for_monitor_data(&self, interval: Interval) -> Result<bool, RewardingError> {
let since = interval.start_unix_timestamp();
let until = interval.end_unix_timestamp();
let monitor_runs = self.storage.get_monitor_runs_count(since, until).await?;
// check if we have more than threshold percentage of monitor runs for the interval
let available = monitor_runs as f32 * 100.0 / self.expected_interval_monitor_runs as f32;
Ok(available >= self.minimum_interval_monitor_threshold as f32)
}
async fn sync_up_rewarding_intervals(&self) -> Result<(), RewardingError> {
let mut last_stored_interval = self.nymd_client.get_current_interval().await?;
let block_now: OffsetDateTime = self.nymd_client.current_block_timestamp().await?.into();
let actual_current_interval = match last_stored_interval.current(block_now) {
None => return Ok(()),
Some(interval) => interval,
};
// we're waiting for the first interval to start... (same is true if the value was 'None')
if actual_current_interval.start() < last_stored_interval.start() {
return Ok(());
}
// we're already synced up
if actual_current_interval == last_stored_interval {
return Ok(());
}
// actual_current_interval > last_stored_interval
loop {
// if we can perform rewarding, do it, otherwise just go straight into the next interval
if self
.check_interval_eligibility(last_stored_interval)
.await?
{
self.perform_rewarding(last_stored_interval).await?;
} else {
self.nymd_client.advance_current_interval().await?;
}
last_stored_interval = self.nymd_client.get_current_interval().await?;
// compare by start times in case the id didn't match (TODO: is it even possible?)
if last_stored_interval.start() == actual_current_interval.start() {
break;
}
}
Ok(())
}
// pub(crate) async fn processing_loop_iteration(&self) -> Result<std::ops::ControlFlow<()>, RewardingError> {
pub(crate) async fn processing_loop_iteration(&self) -> Result<(), RewardingError> {
let last_stored_interval = self.nymd_client.get_current_interval().await?;
let block_now: OffsetDateTime = self.nymd_client.current_block_timestamp().await?.into();
let actual_current_interval = match last_stored_interval.current(block_now) {
None => return Ok(()),
Some(interval) => interval,
};
// the [stored] interval has finished - we should distribute rewards now
if last_stored_interval.start() < actual_current_interval.start() {
// it's time to distribute rewards, however, first let's see if we have enough data to go through with it
// (consider the case of rewards being distributed every 24h at 12:00pm and validator-api
// starting for the very first time at 11:00am. It's not going to have enough data for
// rewards for the *current* interval, but we couldn't have known that at startup)
if self
.check_interval_eligibility(last_stored_interval)
.await?
{
self.perform_rewarding(last_stored_interval).await?;
} else {
warn!("We do not have sufficient monitor data to perform rewarding in this interval ({}). We're advancing it forward...", last_stored_interval);
self.nymd_client.advance_current_interval().await?;
}
} else {
info!(
"rewards will be distributed around {}. Approximately {:?} remaining",
last_stored_interval.end(),
last_stored_interval.until_end(block_now)
);
}
sleep(Duration::from_secs(15 * 60)).await;
Ok(())
}
pub(crate) async fn run(&self) {
// whatever happens, we shouldn't do anything until the cache is initialised
self.validator_cache.wait_for_initial_values().await;
if let Err(err) = self.sync_up_rewarding_intervals().await {
error!(
"Failed to sync up intervals with the contract state - {}",
err
);
process::exit(1);
}
// at this point the current block time < end of current[or first] interval, so to do anything,
// we have to wait for the interval to finish
loop {
if let Err(err) = self.processing_loop_iteration().await {
error!("failed to finish rewarding loop iteration - {}", err);
// should we go into backoff here or just exit the process?
sleep(Duration::from_secs(15 * 60)).await;
}
}
}
}
validator-api/src/storage/manager.rs
+13 -11
View File
@@ -1,6 +1,8 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use mixnet_contract_common::Interval;
use crate::network_monitor::monitor::summary_producer::NodeResult;
use crate::node_status_api::models::{HistoricalUptime, Uptime};
use crate::node_status_api::utils::ActiveNodeStatuses;
@@ -673,18 +675,21 @@ impl StorageManager {
///
/// * `interval_start_timestamp`: Unix timestamp of start of this rewarding interval.
/// * `interval_end_timestamp`: Unix timestamp of end of this rewarding interval.
pub(super) async fn insert_new_interval_rewarding(
pub(super) async fn insert_new_epoch_rewarding(
&self,
interval_start_timestamp: i64,
interval_end_timestamp: i64,
epoch: Interval,
) -> Result<i64, sqlx::Error> {
let id = epoch.id();
let start = epoch.start_unix_timestamp();
let end = epoch.end_unix_timestamp();
let res = sqlx::query!(
r#"
INSERT INTO interval_rewarding (interval_start_timestamp, interval_end_timestamp, finished)
VALUES (?, ?, 0)
INSERT INTO interval_rewarding (id, interval_start_timestamp, interval_end_timestamp, finished)
VALUES (?, ?, ?, 0)
"#,
interval_start_timestamp,
interval_end_timestamp,
id,
start,
end,
)
.execute(&self.connection_pool)
.await?;
@@ -697,10 +702,7 @@ impl StorageManager {
/// # Arguments
///
/// * `id`: id of the entry we want to update.
pub(super) async fn update_finished_interval_rewarding(
&self,
id: i64,
) -> Result<(), sqlx::Error> {
pub async fn update_finished_interval_rewarding(&self, id: i64) -> Result<(), sqlx::Error> {
sqlx::query!(
r#"
UPDATE interval_rewarding
validator-api/src/storage/mod.rs
+5 -5
View File
@@ -13,6 +13,7 @@ use crate::storage::models::{
FailedMixnodeRewardChunk, IntervalRewarding, NodeStatus, PossiblyUnrewardedMixnode,
RewardingReport, TestingRoute,
};
use mixnet_contract_common::Interval;
use rocket::fairing::{self, AdHoc};
use rocket::{Build, Rocket};
use sqlx::ConnectOptions;
@@ -25,7 +26,7 @@ pub(crate) mod models;
// note that clone here is fine as upon cloning the same underlying pool will be used
#[derive(Clone)]
pub(crate) struct ValidatorApiStorage {
manager: StorageManager,
pub manager: StorageManager,
}
impl ValidatorApiStorage {
@@ -688,13 +689,12 @@ impl ValidatorApiStorage {
///
/// * `interval_start_timestamp`: Unix timestamp of start of this rewarding interval.
/// * `interval_end_timestamp`: Unix timestamp of end of this rewarding interval.
pub(crate) async fn insert_started_interval_rewarding(
pub(crate) async fn insert_started_epoch_rewarding(
&self,
interval_start_timestamp: i64,
interval_end_timestamp: i64,
epoch: Interval,
) -> Result<i64, ValidatorApiStorageError> {
self.manager
.insert_new_interval_rewarding(interval_start_timestamp, interval_end_timestamp)
.insert_new_epoch_rewarding(epoch)
.await
.map_err(|_| ValidatorApiStorageError::InternalDatabaseError)
}