diff --git a/clients/client-core/src/client/base_client/mod.rs b/clients/client-core/src/client/base_client/mod.rs index 9788cb26f8..5295b08e55 100644 --- a/clients/client-core/src/client/base_client/mod.rs +++ b/clients/client-core/src/client/base_client/mod.rs @@ -270,7 +270,7 @@ impl<'a> BaseClientBuilder<'a> { ack_sender, self.debug_config.gateway_response_timeout, self.bandwidth_controller.take(), - Some(shutdown), + shutdown, ); gateway_client.set_disabled_credentials_mode(self.disabled_credentials); diff --git a/clients/client-core/src/config/mod.rs b/clients/client-core/src/config/mod.rs index 82317a6431..98af55133f 100644 --- a/clients/client-core/src/config/mod.rs +++ b/clients/client-core/src/config/mod.rs @@ -141,8 +141,15 @@ impl Config { pub fn set_high_default_traffic_volume(&mut self) { self.debug.average_packet_delay = Duration::from_millis(10); - self.debug.loop_cover_traffic_average_delay = Duration::from_millis(2_000_000); // basically don't really send cover messages - self.debug.message_sending_average_delay = Duration::from_millis(4); // 250 "real" messages / s + // basically don't really send cover messages + self.debug.loop_cover_traffic_average_delay = Duration::from_millis(2_000_000); + // 250 "real" messages / s + self.debug.message_sending_average_delay = Duration::from_millis(4); + } + + pub fn set_no_cover_traffic(&mut self) { + self.debug.disable_loop_cover_traffic_stream = true; + self.debug.disable_main_poisson_packet_distribution = true; } pub fn set_custom_version(&mut self, version: &str) { diff --git a/clients/client-core/src/init.rs b/clients/client-core/src/init.rs index 5e7c1e6c2a..44c20a7597 100644 --- a/clients/client-core/src/init.rs +++ b/clients/client-core/src/init.rs @@ -90,7 +90,6 @@ async fn register_with_gateway( gateway.owner.clone(), our_identity.clone(), timeout, - None, ); gateway_client .establish_connection() diff --git a/clients/native/src/commands/init.rs b/clients/native/src/commands/init.rs index d97bcc8e60..dfcd98bab2 100644 --- a/clients/native/src/commands/init.rs +++ b/clients/native/src/commands/init.rs @@ -46,6 +46,10 @@ pub(crate) struct Init { #[clap(long, hidden = true)] fastmode: bool, + /// Disable loop cover traffic and the Poisson rate limiter (for debugging only) + #[clap(long, hidden = true)] + no_cover: bool, + /// Set this client to work in a enabled credentials mode that would attempt to use gateway /// with bandwidth credential requirement. #[cfg(feature = "coconut")] @@ -61,6 +65,7 @@ impl From for OverrideConfig { disable_socket: init_config.disable_socket, port: init_config.port, fastmode: init_config.fastmode, + no_cover: init_config.no_cover, #[cfg(feature = "coconut")] enabled_credentials_mode: init_config.enabled_credentials_mode, diff --git a/clients/native/src/commands/mod.rs b/clients/native/src/commands/mod.rs index 648a1d6002..c3232165d8 100644 --- a/clients/native/src/commands/mod.rs +++ b/clients/native/src/commands/mod.rs @@ -80,6 +80,7 @@ pub(crate) struct OverrideConfig { disable_socket: bool, port: Option, fastmode: bool, + no_cover: bool, #[cfg(feature = "coconut")] enabled_credentials_mode: bool, @@ -141,6 +142,10 @@ pub(crate) fn override_config(mut config: Config, args: OverrideConfig) -> Confi config.get_base_mut().set_high_default_traffic_volume(); } + if args.no_cover { + config.get_base_mut().set_no_cover_traffic(); + } + config } diff --git a/clients/native/src/commands/run.rs b/clients/native/src/commands/run.rs index df49471ea9..4267fcbac5 100644 --- a/clients/native/src/commands/run.rs +++ b/clients/native/src/commands/run.rs @@ -39,6 +39,15 @@ pub(crate) struct Run { #[clap(short, long)] port: Option, + /// Mostly debug-related option to increase default traffic rate so that you would not need to + /// modify config post init + #[clap(long, hidden = true)] + fastmode: bool, + + /// Disable loop cover traffic and the Poisson rate limiter (for debugging only) + #[clap(long, hidden = true)] + no_cover: bool, + /// Set this client to work in a enabled credentials mode that would attempt to use gateway /// with bandwidth credential requirement. #[cfg(feature = "coconut")] @@ -53,7 +62,8 @@ impl From for OverrideConfig { api_validators: run_config.api_validators, disable_socket: run_config.disable_socket, port: run_config.port, - fastmode: false, + fastmode: run_config.fastmode, + no_cover: run_config.no_cover, #[cfg(feature = "coconut")] enabled_credentials_mode: run_config.enabled_credentials_mode, } diff --git a/clients/socks5/src/commands/init.rs b/clients/socks5/src/commands/init.rs index a1d123fc75..a727d0d8e9 100644 --- a/clients/socks5/src/commands/init.rs +++ b/clients/socks5/src/commands/init.rs @@ -46,6 +46,10 @@ pub(crate) struct Init { #[clap(long, hidden = true)] fastmode: bool, + /// Disable loop cover traffic and the Poisson rate limiter (for debugging only) + #[clap(long, hidden = true)] + no_cover: bool, + /// Set this client to work in a enabled credentials mode that would attempt to use gateway /// with bandwidth credential requirement. #[cfg(feature = "coconut")] @@ -60,6 +64,7 @@ impl From for OverrideConfig { api_validators: init_config.api_validators, port: init_config.port, fastmode: init_config.fastmode, + no_cover: init_config.no_cover, #[cfg(feature = "coconut")] enabled_credentials_mode: init_config.enabled_credentials_mode, } diff --git a/clients/socks5/src/commands/mod.rs b/clients/socks5/src/commands/mod.rs index e1d42e0fb1..6a141808bb 100644 --- a/clients/socks5/src/commands/mod.rs +++ b/clients/socks5/src/commands/mod.rs @@ -83,6 +83,7 @@ pub(crate) struct OverrideConfig { api_validators: Option, port: Option, fastmode: bool, + no_cover: bool, #[cfg(feature = "coconut")] enabled_credentials_mode: bool, @@ -136,6 +137,10 @@ pub(crate) fn override_config(mut config: Config, args: OverrideConfig) -> Confi config.get_base_mut().set_high_default_traffic_volume(); } + if args.no_cover { + config.get_base_mut().set_no_cover_traffic(); + } + config } diff --git a/clients/socks5/src/commands/run.rs b/clients/socks5/src/commands/run.rs index d74e6678fe..16f4b7bc53 100644 --- a/clients/socks5/src/commands/run.rs +++ b/clients/socks5/src/commands/run.rs @@ -43,6 +43,15 @@ pub(crate) struct Run { #[clap(short, long)] port: Option, + /// Mostly debug-related option to increase default traffic rate so that you would not need to + /// modify config post init + #[clap(long, hidden = true)] + fastmode: bool, + + /// Disable loop cover traffic and the Poisson rate limiter (for debugging only) + #[clap(long, hidden = true)] + no_cover: bool, + /// Set this client to work in a enabled credentials mode that would attempt to use gateway /// with bandwidth credential requirement. #[cfg(feature = "coconut")] @@ -56,8 +65,8 @@ impl From for OverrideConfig { nymd_validators: run_config.nymd_validators, api_validators: run_config.api_validators, port: run_config.port, - fastmode: false, - + fastmode: run_config.fastmode, + no_cover: run_config.no_cover, #[cfg(feature = "coconut")] enabled_credentials_mode: run_config.enabled_credentials_mode, } diff --git a/common/client-libs/gateway-client/src/client.rs b/common/client-libs/gateway-client/src/client.rs index 56ff03737a..158af85bdd 100644 --- a/common/client-libs/gateway-client/src/client.rs +++ b/common/client-libs/gateway-client/src/client.rs @@ -10,7 +10,7 @@ pub use crate::packet_router::{ use crate::socket_state::{PartiallyDelegated, SocketState}; use crate::{cleanup_socket_message, try_decrypt_binary_message}; use crypto::asymmetric::identity; -use futures::{FutureExt, SinkExt, StreamExt}; +use futures::{SinkExt, StreamExt}; use gateway_requests::authentication::encrypted_address::EncryptedAddressBytes; use gateway_requests::iv::IV; use gateway_requests::registration::handshake::{client_handshake, SharedKeys}; @@ -67,9 +67,7 @@ pub struct GatewayClient { reconnection_backoff: Duration, /// Listen to shutdown messages. - // TODO: fix this - #[cfg_attr(target_arch = "wasm32", allow(dead_code))] - shutdown: Option, + shutdown: ShutdownListener, } impl GatewayClient { @@ -85,7 +83,7 @@ impl GatewayClient { ack_sender: AcknowledgementSender, response_timeout_duration: Duration, bandwidth_controller: Option>, - shutdown: Option, + shutdown: ShutdownListener, ) -> Self { GatewayClient { authenticated: false, @@ -130,7 +128,6 @@ impl GatewayClient { gateway_owner: String, local_identity: Arc, response_timeout_duration: Duration, - shutdown: Option, ) -> Self { use futures::channel::mpsc; @@ -138,6 +135,7 @@ impl GatewayClient { // perfectly fine here, because it's not meant to be used let (ack_tx, _) = mpsc::unbounded(); let (mix_tx, _) = mpsc::unbounded(); + let shutdown = ShutdownListener::dummy(); let packet_router = PacketRouter::new(ack_tx, mix_tx, shutdown.clone()); GatewayClient { @@ -283,44 +281,19 @@ impl GatewayClient { // technically the `wasm_timer` also works outside wasm, but unless required, // I really prefer to just stick to tokio #[cfg(target_arch = "wasm32")] - let timeout = wasm_timer::Delay::new(self.response_timeout_duration); - - let mut fused_timeout = timeout.fuse(); - let mut fused_stream = conn.fuse(); - - // Bit of an ugly workaround for selecting on an `Option` without having access to - // `tokio::select` - #[cfg(not(target_arch = "wasm32"))] - let shutdown = { - let m_shutdown = self.shutdown.clone(); - async { - if let Some(mut s) = m_shutdown { - // TODO: fix this by marking as success _after_ the select - s.mark_as_success(); - s.recv().await - } else { - std::future::pending::<()>().await - } - } - .fuse() - }; - #[cfg(not(target_arch = "wasm32"))] - tokio::pin!(shutdown); - - #[cfg(target_arch = "wasm32")] - let mut shutdown = std::future::pending::<()>().fuse(); + let mut timeout = wasm_timer::Delay::new(self.response_timeout_duration); loop { - futures::select! { - _ = shutdown => { + tokio::select! { + _ = self.shutdown.recv() => { log::trace!("GatewayClient control response: Received shutdown"); log::debug!("GatewayClient control response: Exiting"); break Err(GatewayClientError::ConnectionClosedGatewayShutdown); } - _ = &mut fused_timeout => { + _ = &mut timeout => { break Err(GatewayClientError::Timeout); } - msg = fused_stream.next() => { + msg = conn.next() => { let ws_msg = match cleanup_socket_message(msg) { Err(err) => break Err(err), Ok(msg) => msg @@ -770,7 +743,6 @@ impl GatewayClient { .as_ref() .expect("no shared key present even though we're authenticated!"), ), - #[cfg(not(target_arch = "wasm32"))] self.shutdown.clone(), ) } diff --git a/common/client-libs/gateway-client/src/packet_router.rs b/common/client-libs/gateway-client/src/packet_router.rs index 59d1d8add1..fd63077ee6 100644 --- a/common/client-libs/gateway-client/src/packet_router.rs +++ b/common/client-libs/gateway-client/src/packet_router.rs @@ -21,14 +21,14 @@ pub type AcknowledgementReceiver = mpsc::UnboundedReceiver>>; pub struct PacketRouter { ack_sender: AcknowledgementSender, mixnet_message_sender: MixnetMessageSender, - shutdown: Option, + shutdown: ShutdownListener, } impl PacketRouter { pub fn new( ack_sender: AcknowledgementSender, mixnet_message_sender: MixnetMessageSender, - shutdown: Option, + shutdown: ShutdownListener, ) -> Self { PacketRouter { ack_sender, @@ -82,12 +82,10 @@ impl PacketRouter { if !received_messages.is_empty() { trace!("routing 'real'"); if let Err(err) = self.mixnet_message_sender.unbounded_send(received_messages) { - if let Some(shutdown) = &mut self.shutdown { - if shutdown.is_shutdown_poll() { - // This should ideally not happen, but it's ok - log::warn!("Failed to send mixnet message due to receiver task shutdown"); - return Err(GatewayClientError::MixnetMsgSenderFailedToSend); - } + if self.shutdown.is_shutdown_poll() || self.shutdown.is_dummy() { + // This should ideally not happen, but it's ok + log::warn!("Failed to send mixnet message due to receiver task shutdown"); + return Err(GatewayClientError::MixnetMsgSenderFailedToSend); } // This should never happen during ordinary operation the way it's currently used. // Abort to be on the safe side diff --git a/common/client-libs/gateway-client/src/socket_state.rs b/common/client-libs/gateway-client/src/socket_state.rs index 2d66e782e8..aa96b9a4fa 100644 --- a/common/client-libs/gateway-client/src/socket_state.rs +++ b/common/client-libs/gateway-client/src/socket_state.rs @@ -10,7 +10,6 @@ use futures::{SinkExt, StreamExt}; use gateway_requests::registration::handshake::SharedKeys; use log::*; use std::sync::Arc; -#[cfg(not(target_arch = "wasm32"))] use task::ShutdownListener; use tungstenite::Message; @@ -85,7 +84,7 @@ impl PartiallyDelegated { conn: WsConn, packet_router: PacketRouter, shared_key: Arc, - #[cfg(not(target_arch = "wasm32"))] shutdown: Option, + mut shutdown: ShutdownListener, ) -> Self { // when called for, it NEEDS TO yield back the stream so that we could merge it and // read control request responses. @@ -99,27 +98,9 @@ impl PartiallyDelegated { let mut chunk_stream = (&mut stream).ready_chunks(8); let mut packet_router = packet_router; - // Bit of an ugly workaround for selecting on an `Option` without having access to - // `tokio::select` - #[cfg(not(target_arch = "wasm32"))] - let shutdown = { - async { - if let Some(mut s) = shutdown { - s.recv().await - } else { - std::future::pending::<()>().await - } - } - }; - #[cfg(not(target_arch = "wasm32"))] - tokio::pin!(shutdown); - - #[cfg(target_arch = "wasm32")] - let mut shutdown = std::future::pending::<()>(); - let ret_err = loop { tokio::select! { - _ = &mut shutdown => { + _ = shutdown.recv() => { log::trace!("GatewayClient listener: Received shutdown"); log::debug!("GatewayClient listener: Exiting"); return; @@ -142,7 +123,10 @@ impl PartiallyDelegated { if match ret_err { Err(err) => stream_sender.send(Err(err)), - Ok(_) => stream_sender.send(Ok(stream)), + Ok(_) => { + shutdown.mark_as_success(); + stream_sender.send(Ok(stream)) + } } .is_err() { diff --git a/common/client-libs/validator-client/src/validator_api/mod.rs b/common/client-libs/validator-client/src/validator_api/mod.rs index 4a7b7af3cd..3cb7ae2ce9 100644 --- a/common/client-libs/validator-client/src/validator_api/mod.rs +++ b/common/client-libs/validator-client/src/validator_api/mod.rs @@ -136,7 +136,12 @@ impl Client { &self, ) -> Result, ValidatorAPIError> { self.query_validator_api( - &[routes::API_VERSION, routes::MIXNODES, routes::DETAILED], + &[ + routes::API_VERSION, + routes::STATUS, + routes::MIXNODES, + routes::DETAILED, + ], NO_PARAMS, ) .await @@ -161,6 +166,7 @@ impl Client { self.query_validator_api( &[ routes::API_VERSION, + routes::STATUS, routes::MIXNODES, routes::ACTIVE, routes::DETAILED, @@ -252,6 +258,7 @@ impl Client { self.query_validator_api( &[ routes::API_VERSION, + routes::STATUS, routes::MIXNODES, routes::REWARDED, routes::DETAILED, diff --git a/common/task/src/shutdown.rs b/common/task/src/shutdown.rs index 0f8efa51d7..0f88135f9c 100644 --- a/common/task/src/shutdown.rs +++ b/common/task/src/shutdown.rs @@ -3,7 +3,7 @@ use std::{error::Error, time::Duration}; -use futures::FutureExt; +use futures::{future::pending, FutureExt}; use tokio::{ sync::{ mpsc, @@ -149,9 +149,8 @@ pub struct ShutdownListener { // Also notify if we dropped without shutdown being registered drop_error: ErrorSender, - // Sometimes it's necessary to clone and drop the shutdown listener during normal operation, - // for those situations we need to explicitly not drop (and trigger shutdown). - set_not_drop: bool, + // The current operating mode + mode: ShutdownListenerMode, } impl ShutdownListener { @@ -168,15 +167,40 @@ impl ShutdownListener { notify, return_error, drop_error, - set_not_drop: false, + mode: ShutdownListenerMode::Listening, } } + // Create a dummy that will never report that we should shutdown. + pub fn dummy() -> ShutdownListener { + let (_notify_tx, notify_rx) = watch::channel(()); + let (task_halt_tx, _task_halt_rx) = mpsc::unbounded_channel(); + let (task_drop_tx, _task_drop_rx) = mpsc::unbounded_channel(); + ShutdownListener { + shutdown: false, + notify: notify_rx, + return_error: task_halt_tx, + drop_error: task_drop_tx, + mode: ShutdownListenerMode::Dummy, + } + } + + pub fn is_dummy(&self) -> bool { + self.mode.is_dummy() + } + pub fn is_shutdown(&self) -> bool { - self.shutdown + if self.mode.is_dummy() { + false + } else { + self.shutdown + } } pub async fn recv(&mut self) { + if self.mode.is_dummy() { + return pending().await; + } if self.shutdown { return; } @@ -185,6 +209,9 @@ impl ShutdownListener { } pub async fn recv_timeout(&mut self) { + if self.mode.is_dummy() { + return pending().await; + } #[cfg(not(target_arch = "wasm32"))] tokio::time::timeout(Self::SHUTDOWN_TIMEOUT, self.recv()) .await @@ -192,6 +219,9 @@ impl ShutdownListener { } pub fn is_shutdown_poll(&mut self) -> bool { + if self.mode.is_dummy() { + return false; + } if self.shutdown { return true; } @@ -210,21 +240,30 @@ impl ShutdownListener { } } + // This listener should to *not* notify the ShutdownNotifier to shutdown when dropped. For + // example when we clone the listener for a task handling connections, we often want to drop + // without signal failure. + pub fn mark_as_success(&mut self) { + self.mode.set_should_not_signal_on_drop(); + } + pub fn send_we_stopped(&mut self, err: SentError) { + if self.mode.is_dummy() { + return; + } log::trace!("Notifying we stopped: {:?}", err); if self.return_error.send(err).is_err() { log::error!("Failed to send back error message"); } } - - pub fn mark_as_success(&mut self) { - self.set_not_drop = true; - } } impl Drop for ShutdownListener { fn drop(&mut self) { - if !self.set_not_drop && !self.is_shutdown_poll() { + if !self.mode.should_signal_on_drop() { + return; + } + if !self.is_shutdown_poll() { log::trace!("Notifying stop on unexpected drop"); // If we can't send, well then there is not much to do self.drop_error @@ -234,6 +273,37 @@ impl Drop for ShutdownListener { } } +#[derive(Clone, Debug, PartialEq, Eq)] +enum ShutdownListenerMode { + // Normal operations + Listening, + // Normal operations, but we don't report back if the we stop by getting dropped. + ListeningButDontReportHalt, + // Dummy mode, for when we don't do anything at all. + Dummy, +} + +impl ShutdownListenerMode { + fn is_dummy(&self) -> bool { + self == &ShutdownListenerMode::Dummy + } + + fn should_signal_on_drop(&self) -> bool { + match self { + ShutdownListenerMode::Listening => true, + ShutdownListenerMode::ListeningButDontReportHalt | ShutdownListenerMode::Dummy => false, + } + } + + fn set_should_not_signal_on_drop(&mut self) { + use ShutdownListenerMode::{Dummy, Listening, ListeningButDontReportHalt}; + *self = match &self { + ListeningButDontReportHalt | Listening => ListeningButDontReportHalt, + Dummy => Dummy, + }; + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/envs/mainnet.env b/envs/mainnet.env index d53b7f4b8b..751081adb8 100644 --- a/envs/mainnet.env +++ b/envs/mainnet.env @@ -9,7 +9,7 @@ MIX_DENOM_DISPLAY=nym STAKE_DENOM=unyx STAKE_DENOM_DISPLAY=nyx DENOMS_EXPONENT=6 -MIXNET_CONTRACT_ADDRESS=n14hj2tavq8fpesdwxxcu44rty3hh90vhujrvcmstl4zr3txmfvw9sjyvg3g +MIXNET_CONTRACT_ADDRESS=n17srjznxl9dvzdkpwpw24gg668wc73val88a6m5ajg6ankwvz9wtst0cznr VESTING_CONTRACT_ADDRESS=n1nc5tatafv6eyq7llkr2gv50ff9e22mnf70qgjlv737ktmt4eswrq73f2nw BANDWIDTH_CLAIM_CONTRACT_ADDRESS=n19lc9u84cz0yz3fww5283nucc9yvr8gsjmgeul0 COCONUT_BANDWIDTH_CONTRACT_ADDRESS=n19lc9u84cz0yz3fww5283nucc9yvr8gsjmgeul0 @@ -17,5 +17,5 @@ MULTISIG_CONTRACT_ADDRESS=n19lc9u84cz0yz3fww5283nucc9yvr8gsjmgeul0 COCONUT_DKG_CONTRACT_ADDRESS=n19lc9u84cz0yz3fww5283nucc9yvr8gsjmgeul0 REWARDING_VALIDATOR_ADDRESS=n10yyd98e2tuwu0f7ypz9dy3hhjw7v772q6287gy STATISTICS_SERVICE_DOMAIN_ADDRESS="https://mainnet-stats.nymte.ch:8090" -NYMD_VALIDATOR="https://rpc.nyx.nodes.guru/" +NYMD_VALIDATOR="https://rpc.nymtech.net"; API_VALIDATOR="https://validator.nymtech.net/api/" diff --git a/nym-connect/Cargo.lock b/nym-connect/Cargo.lock index 0dba93e249..8d2d5d1ceb 100644 --- a/nym-connect/Cargo.lock +++ b/nym-connect/Cargo.lock @@ -630,7 +630,7 @@ dependencies = [ [[package]] name = "client-core" -version = "1.1.0" +version = "1.1.1" dependencies = [ "client-connections", "config", @@ -3352,7 +3352,7 @@ dependencies = [ [[package]] name = "nym-socks5-client" -version = "1.1.0" +version = "1.1.1" dependencies = [ "clap", "client-connections", diff --git a/validator-api/src/contract_cache/mod.rs b/validator-api/src/contract_cache/mod.rs index 95d984533c..b4538ea36b 100644 --- a/validator-api/src/contract_cache/mod.rs +++ b/validator-api/src/contract_cache/mod.rs @@ -2,13 +2,11 @@ // SPDX-License-Identifier: Apache-2.0 use crate::nymd_client::Client; -use crate::storage::ValidatorApiStorage; use ::time::OffsetDateTime; use anyhow::Result; -use mixnet_contract_common::mixnode::MixNodeDetails; -use mixnet_contract_common::reward_params::{Performance, RewardingParams}; use mixnet_contract_common::{ - GatewayBond, IdentityKey, Interval, MixId, MixNodeBond, RewardedSetNodeStatus, + mixnode::MixNodeDetails, reward_params::RewardingParams, GatewayBond, IdentityKey, Interval, + MixId, MixNodeBond, RewardedSetNodeStatus, }; use okapi::openapi3::OpenApi; use rocket::fairing::AdHoc; @@ -16,18 +14,17 @@ use rocket::Route; use rocket_okapi::openapi_get_routes_spec; use rocket_okapi::settings::OpenApiSettings; use serde::Serialize; -use std::collections::HashMap; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; +use std::ops::Deref; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::Duration; use task::ShutdownListener; use tokio::sync::{watch, RwLock}; use tokio::time; -use validator_api_requests::models::{MixNodeBondAnnotated, MixnodeStatus}; +use validator_api_requests::models::MixnodeStatus; use validator_client::nymd::CosmWasmClient; -pub(crate) mod reward_estimate; pub(crate) mod routes; // The cache can emit notifications to listeners about the current state @@ -42,9 +39,6 @@ pub struct ValidatorCacheRefresher { cache: ValidatorCache, caching_interval: Duration, - // Readonly: some of the quantities cached depends on values from the storage. - storage: Option, - // Notify listeners that the cache has been updated update_notifier: watch::Sender, } @@ -56,14 +50,14 @@ pub struct ValidatorCache { } struct ValidatorCacheInner { - mixnodes: Cache>, + mixnodes: Cache>, gateways: Cache>, mixnodes_blacklist: Cache>, gateways_blacklist: Cache>, - rewarded_set: Cache>, - active_set: Cache>, + rewarded_set: Cache>, + active_set: Cache>, current_reward_params: Cache>, current_interval: Cache>, @@ -102,90 +96,33 @@ impl Cache { } } +impl Deref for Cache { + type Target = T; + + fn deref(&self) -> &Self::Target { + &self.value + } +} + impl ValidatorCacheRefresher { pub(crate) fn new( nymd_client: Client, caching_interval: Duration, cache: ValidatorCache, - storage: Option, ) -> Self { let (tx, _) = watch::channel(CacheNotification::Start); ValidatorCacheRefresher { nymd_client, cache, caching_interval, - storage, update_notifier: tx, } } - async fn get_performance(&self, mix_id: MixId, epoch: Interval) -> Option { - self.storage - .as_ref()? - .get_average_mixnode_uptime_in_the_last_24hrs( - mix_id, - epoch.current_epoch_end_unix_timestamp(), - ) - .await - .ok() - .map(Into::into) - } - pub fn subscribe(&self) -> watch::Receiver { self.update_notifier.subscribe() } - async fn annotate_node_with_details( - &self, - mixnodes: Vec, - interval_reward_params: RewardingParams, - current_interval: Interval, - rewarded_set: &HashMap, - ) -> Vec { - let mut annotated = Vec::new(); - for mixnode in mixnodes { - let stake_saturation = mixnode - .rewarding_details - .bond_saturation(&interval_reward_params); - - let uncapped_stake_saturation = mixnode - .rewarding_details - .uncapped_bond_saturation(&interval_reward_params); - - let performance = self - .get_performance(mixnode.mix_id(), current_interval) - .await - .unwrap_or_default(); - - let rewarded_set_status = rewarded_set.get(&mixnode.mix_id()).cloned(); - - let reward_estimate = reward_estimate::compute_reward_estimate( - &mixnode, - performance, - rewarded_set_status, - interval_reward_params, - current_interval, - ); - - let (estimated_operator_apy, estimated_delegators_apy) = - reward_estimate::compute_apy_from_reward( - &mixnode, - reward_estimate, - current_interval, - ); - - annotated.push(MixNodeBondAnnotated { - mixnode_details: mixnode, - stake_saturation, - uncapped_stake_saturation, - performance, - estimated_operator_apy, - estimated_delegators_apy, - }); - } - annotated - } - async fn get_rewarded_set_map(&self) -> HashMap where C: CosmWasmClient + Sync + Send, @@ -198,9 +135,9 @@ impl ValidatorCacheRefresher { } fn collect_rewarded_and_active_set_details( - all_mixnodes: &[MixNodeBondAnnotated], + all_mixnodes: &[MixNodeDetails], rewarded_set_nodes: &HashMap, - ) -> (Vec, Vec) { + ) -> (Vec, Vec) { let mut active_set = Vec::new(); let mut rewarded_set = Vec::new(); @@ -226,14 +163,10 @@ impl ValidatorCacheRefresher { let mixnodes = self.nymd_client.get_mixnodes().await?; let gateways = self.nymd_client.get_gateways().await?; - let rewarded_set = self.get_rewarded_set_map().await; - - let mixnodes = self - .annotate_node_with_details(mixnodes, rewarding_params, current_interval, &rewarded_set) - .await; + let rewarded_set_map = self.get_rewarded_set_map().await; let (rewarded_set, active_set) = - Self::collect_rewarded_and_active_set_details(&mixnodes, &rewarded_set); + Self::collect_rewarded_and_active_set_details(&mixnodes, &rewarded_set_map); info!( "Updating validator cache. There are {} mixnodes and {} gateways", @@ -324,10 +257,10 @@ impl ValidatorCache { async fn update_cache( &self, - mixnodes: Vec, + mixnodes: Vec, gateways: Vec, - rewarded_set: Vec, - active_set: Vec, + rewarded_set: Vec, + active_set: Vec, rewarding_params: RewardingParams, current_interval: Interval, ) { @@ -422,7 +355,7 @@ impl ValidatorCache { error!("Failed to update gateways blacklist"); } - pub async fn mixnodes_detailed(&self) -> Vec { + pub async fn mixnodes(&self) -> Vec { let blacklist = self.mixnodes_blacklist().await; let mixnodes = match time::timeout(Duration::from_millis(100), self.inner.read()).await { Ok(cache) => cache.mixnodes.clone(), @@ -444,14 +377,6 @@ impl ValidatorCache { } } - pub async fn mixnodes(&self) -> Vec { - self.mixnodes_detailed() - .await - .into_iter() - .map(|bond| bond.mixnode_details) - .collect() - } - pub async fn mixnodes_basic(&self) -> Vec { match time::timeout(Duration::from_millis(100), self.inner.read()).await { Ok(cache) => cache @@ -459,7 +384,7 @@ impl ValidatorCache { .clone() .into_inner() .into_iter() - .map(|bond| bond.mixnode_details.bond_information) + .map(|bond| bond.bond_information) .collect(), Err(e) => { error!("{}", e); @@ -500,7 +425,7 @@ impl ValidatorCache { } } - pub async fn rewarded_set_detailed(&self) -> Cache> { + pub async fn rewarded_set(&self) -> Cache> { match time::timeout(Duration::from_millis(100), self.inner.read()).await { Ok(cache) => cache.rewarded_set.clone(), Err(e) => { @@ -510,16 +435,7 @@ impl ValidatorCache { } } - pub async fn rewarded_set(&self) -> Vec { - self.rewarded_set_detailed() - .await - .value - .into_iter() - .map(|bond| bond.mixnode_details) - .collect() - } - - pub async fn active_set_detailed(&self) -> Cache> { + pub async fn active_set(&self) -> Cache> { match time::timeout(Duration::from_millis(100), self.inner.read()).await { Ok(cache) => cache.active_set.clone(), Err(e) => { @@ -529,15 +445,6 @@ impl ValidatorCache { } } - pub async fn active_set(&self) -> Vec { - self.active_set_detailed() - .await - .value - .into_iter() - .map(|bond| bond.mixnode_details) - .collect() - } - pub(crate) async fn interval_reward_params(&self) -> Cache> { match time::timeout(Duration::from_millis(100), self.inner.read()).await { Ok(cache) => cache.current_reward_params.clone(), @@ -558,24 +465,21 @@ impl ValidatorCache { } } - pub async fn mixnode_details( - &self, - mix_id: MixId, - ) -> (Option, MixnodeStatus) { + pub async fn mixnode_details(&self, mix_id: MixId) -> (Option, 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.active_set_detailed().await.value; + let active_set = &self.active_set().await.value; if let Some(bond) = active_set.iter().find(|mix| mix.mix_id() == mix_id) { return (Some(bond.clone()), MixnodeStatus::Active); } - let rewarded_set = &self.rewarded_set_detailed().await.value; + let rewarded_set = &self.rewarded_set().await.value; if let Some(bond) = rewarded_set.iter().find(|mix| mix.mix_id() == mix_id) { return (Some(bond.clone()), MixnodeStatus::Standby); } - let all_bonded = &self.mixnodes_detailed().await; + let all_bonded = &self.mixnodes().await; if let Some(bond) = all_bonded.iter().find(|mix| mix.mix_id() == mix_id) { (Some(bond.clone()), MixnodeStatus::Inactive) } else { diff --git a/validator-api/src/contract_cache/routes.rs b/validator-api/src/contract_cache/routes.rs index 45d0b290ac..6aba1ba4f9 100644 --- a/validator-api/src/contract_cache/routes.rs +++ b/validator-api/src/contract_cache/routes.rs @@ -1,15 +1,21 @@ // Copyright 2021-2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::contract_cache::ValidatorCache; -use mixnet_contract_common::mixnode::MixNodeDetails; -use mixnet_contract_common::reward_params::RewardingParams; -use mixnet_contract_common::{GatewayBond, Interval, MixId}; -use rocket::serde::json::Json; -use rocket::State; +use crate::{ + contract_cache::ValidatorCache, + node_status_api::{ + helpers::{_get_active_set_detailed, _get_mixnodes_detailed, _get_rewarded_set_detailed}, + NodeStatusCache, + }, +}; +use mixnet_contract_common::{ + mixnode::MixNodeDetails, reward_params::RewardingParams, GatewayBond, Interval, MixId, +}; +use validator_api_requests::models::MixNodeBondAnnotated; + +use rocket::{serde::json::Json, State}; use rocket_okapi::openapi; use std::collections::HashSet; -use validator_api_requests::models::MixNodeBondAnnotated; #[openapi(tag = "contract-cache")] #[get("/mixnodes")] @@ -17,12 +23,19 @@ pub async fn get_mixnodes(cache: &State) -> Json Redirect { +// Redirect::to(uri!("/v1/status/mixnodes/detailed")) +// } +// ``` #[openapi(tag = "contract-cache")] #[get("/mixnodes/detailed")] pub async fn get_mixnodes_detailed( - cache: &State, + cache: &State, ) -> Json> { - Json(cache.mixnodes_detailed().await) + Json(_get_mixnodes_detailed(cache).await) } #[openapi(tag = "contract-cache")] @@ -34,29 +47,43 @@ pub async fn get_gateways(cache: &State) -> Json) -> Json> { - Json(cache.rewarded_set().await) + Json(cache.rewarded_set().await.value) } +// DEPRECATED: this endpoint now lives in `node_status_api`. Once all consumers are updated, +// replace this with +// ``` +// pub fn get_mixnodes_set_detailed() -> Redirect { +// Redirect::to(uri!("/v1/status/mixnodes/rewarded/detailed")) +// } +// ``` #[openapi(tag = "contract-cache")] #[get("/mixnodes/rewarded/detailed")] pub async fn get_rewarded_set_detailed( - cache: &State, + cache: &State, ) -> Json> { - Json(cache.rewarded_set_detailed().await.value) + Json(_get_rewarded_set_detailed(cache).await) } #[openapi(tag = "contract-cache")] #[get("/mixnodes/active")] pub async fn get_active_set(cache: &State) -> Json> { - Json(cache.active_set().await) + Json(cache.active_set().await.value) } +// DEPRECATED: this endpoint now lives in `node_status_api`. Once all consumers are updated, +// replace this with +// ``` +// pub fn get_active_set_detailed() -> Redirect { +// Redirect::to(uri!("/status/mixnodes/active/detailed")) +// } +// ``` #[openapi(tag = "contract-cache")] #[get("/mixnodes/active/detailed")] pub async fn get_active_set_detailed( - cache: &State, + cache: &State, ) -> Json> { - Json(cache.active_set_detailed().await.value) + Json(_get_active_set_detailed(cache).await) } #[openapi(tag = "contract-cache")] diff --git a/validator-api/src/epoch_operations/mod.rs b/validator-api/src/epoch_operations/mod.rs index cb33962890..06141e7863 100644 --- a/validator-api/src/epoch_operations/mod.rs +++ b/validator-api/src/epoch_operations/mod.rs @@ -156,7 +156,7 @@ impl RewardedSetUpdater { Err(err) => { warn!("failed to obtain the current rewarded set - {}. falling back to the cached version", err); self.validator_cache - .rewarded_set_detailed() + .rewarded_set() .await .into_inner() .into_iter() diff --git a/validator-api/src/main.rs b/validator-api/src/main.rs index 5116896858..3055d57f88 100644 --- a/validator-api/src/main.rs +++ b/validator-api/src/main.rs @@ -121,6 +121,7 @@ fn parse_args() -> ArgMatches { Arg::with_name(CONFIG_ENV_FILE) .help("Path pointing to an env file that configures the validator API") .long(CONFIG_ENV_FILE) + .short('c') .takes_value(true) ) .arg( @@ -577,7 +578,6 @@ async fn run_validator_api(matches: ArgMatches) -> Result<()> { signing_nymd_client.clone(), config.get_caching_interval(), validator_cache.clone(), - Some(storage.clone()), ); let validator_cache_listener = validator_cache_refresher.subscribe(); let shutdown_listener = shutdown.subscribe(); @@ -599,7 +599,6 @@ async fn run_validator_api(matches: ArgMatches) -> Result<()> { nymd_client, config.get_caching_interval(), validator_cache.clone(), - None, ); let validator_cache_listener = validator_cache_refresher.subscribe(); let shutdown_listener = shutdown.subscribe(); @@ -611,11 +610,13 @@ async fn run_validator_api(matches: ArgMatches) -> Result<()> { // Spawn the node status cache refresher. // It is primarily refreshed in-sync with the validator cache, however provide a fallback // caching interval that is twice the validator cache + let storage = rocket.state::().cloned(); let mut validator_api_cache_refresher = node_status_api::NodeStatusCacheRefresher::new( node_status_cache, + config.get_caching_interval().saturating_mul(2), validator_cache, validator_cache_listener, - config.get_caching_interval().saturating_mul(2), + storage, ); let shutdown_listener = shutdown.subscribe(); tokio::spawn(async move { validator_api_cache_refresher.run(shutdown_listener).await }); diff --git a/validator-api/src/network_monitor/monitor/sender.rs b/validator-api/src/network_monitor/monitor/sender.rs index 4ec0e9ce45..d2fa815a11 100644 --- a/validator-api/src/network_monitor/monitor/sender.rs +++ b/validator-api/src/network_monitor/monitor/sender.rs @@ -209,7 +209,7 @@ impl PacketSender { ack_sender, fresh_gateway_client_data.gateway_response_timeout, Some(fresh_gateway_client_data.bandwidth_controller.clone()), - None, + task::ShutdownListener::dummy(), ); gateway_client diff --git a/validator-api/src/node_status_api/cache.rs b/validator-api/src/node_status_api/cache.rs index ef5df963bf..dc8dc7447f 100644 --- a/validator-api/src/node_status_api/cache.rs +++ b/validator-api/src/node_status_api/cache.rs @@ -2,25 +2,33 @@ // SPDX-License-Identifier: Apache-2.0 use crate::contract_cache::{Cache, CacheNotification, ValidatorCache}; -use contracts_common::truncate_decimal; -use mixnet_contract_common::{MixId, MixNodeDetails, RewardingParams}; +use crate::storage::ValidatorApiStorage; +use mixnet_contract_common::reward_params::Performance; +use mixnet_contract_common::{ + Interval, MixId, MixNodeDetails, RewardedSetNodeStatus, RewardingParams, +}; use rocket::fairing::AdHoc; -use serde::Serialize; +use std::collections::HashMap; use std::{sync::Arc, time::Duration}; -use tap::TapFallible; use task::ShutdownListener; +use tokio::sync::RwLockReadGuard; use tokio::{ sync::{watch, RwLock}, time, }; -use validator_api_requests::models::InclusionProbability; +use validator_api_requests::models::{MixNodeBondAnnotated, MixnodeStatus}; + +use self::inclusion_probabilities::InclusionProbabilities; + +use super::reward_estimate::{compute_apy_from_reward, compute_reward_estimate}; + +mod inclusion_probabilities; const CACHE_TIMOUT_MS: u64 = 100; -const MAX_SIMULATION_SAMPLES: u64 = 5000; -const MAX_SIMULATION_TIME_SEC: u64 = 15; enum NodeStatusCacheError { SimulationFailed, + SourceDataMissing, } // A node status cache suitable for caching values computed in one sweep, such as active set @@ -32,27 +40,16 @@ pub struct NodeStatusCache { inner: Arc>, } +#[derive(Default)] struct NodeStatusCacheInner { + mixnodes_annotated: Cache>, + rewarded_set_annotated: Cache>, + active_set_annotated: Cache>, + + // Estimated active set inclusion probabilities from Monte Carlo simulation inclusion_probabilities: Cache, } -#[derive(Clone, Default, Serialize, schemars::JsonSchema)] -pub(crate) struct InclusionProbabilities { - pub inclusion_probabilities: Vec, - pub samples: u64, - pub elapsed: Duration, - pub delta_max: f64, - pub delta_l2: f64, -} - -impl InclusionProbabilities { - pub fn node(&self, mix_id: MixId) -> Option<&InclusionProbability> { - self.inclusion_probabilities - .iter() - .find(|x| x.mix_id == mix_id) - } -} - impl NodeStatusCache { fn new() -> Self { NodeStatusCache { @@ -66,9 +63,18 @@ impl NodeStatusCache { }) } - async fn update_cache(&self, inclusion_probabilities: InclusionProbabilities) { + async fn update_cache( + &self, + mixnodes: Vec, + rewarded_set: Vec, + active_set: Vec, + inclusion_probabilities: InclusionProbabilities, + ) { match time::timeout(Duration::from_millis(CACHE_TIMOUT_MS), self.inner.write()).await { Ok(mut cache) => { + cache.mixnodes_annotated.update(mixnodes); + cache.rewarded_set_annotated.update(rewarded_set); + cache.active_set_annotated.update(active_set); cache .inclusion_probabilities .update(inclusion_probabilities); @@ -77,45 +83,93 @@ impl NodeStatusCache { } } - pub(crate) async fn inclusion_probabilities(&self) -> Option> { + async fn get_cache( + &self, + fn_arg: impl FnOnce(RwLockReadGuard<'_, NodeStatusCacheInner>) -> Cache, + ) -> Option> { match time::timeout(Duration::from_millis(CACHE_TIMOUT_MS), self.inner.read()).await { - Ok(cache) => Some(cache.inclusion_probabilities.clone()), + Ok(cache) => Some(fn_arg(cache)), Err(e) => { error!("{e}"); None } } } + + pub(crate) async fn mixnodes_annotated(&self) -> Option>> { + self.get_cache(|c| c.mixnodes_annotated.clone()).await + } + + pub(crate) async fn rewarded_set_annotated(&self) -> Option>> { + self.get_cache(|c| c.rewarded_set_annotated.clone()).await + } + + pub(crate) async fn active_set_annotated(&self) -> Option>> { + self.get_cache(|c| c.active_set_annotated.clone()).await + } + + pub(crate) async fn inclusion_probabilities(&self) -> Option> { + self.get_cache(|c| c.inclusion_probabilities.clone()).await + } + + pub async fn mixnode_details( + &self, + mix_id: MixId, + ) -> (Option, 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.active_set_annotated().await.unwrap().into_inner(); + if let Some(bond) = active_set.iter().find(|mix| mix.mix_id() == mix_id) { + return (Some(bond.clone()), MixnodeStatus::Active); + } + + let rewarded_set = &self.rewarded_set_annotated().await.unwrap().into_inner(); + if let Some(bond) = rewarded_set.iter().find(|mix| mix.mix_id() == mix_id) { + return (Some(bond.clone()), MixnodeStatus::Standby); + } + + let all_bonded = &self.mixnodes_annotated().await.unwrap().into_inner(); + if let Some(bond) = all_bonded.iter().find(|mix| mix.mix_id() == mix_id) { + (Some(bond.clone()), MixnodeStatus::Inactive) + } else { + (None, MixnodeStatus::NotFound) + } + } } impl NodeStatusCacheInner { pub fn new() -> Self { - Self { - inclusion_probabilities: Default::default(), - } + Self::default() } } // Long running task responsible of keeping the cache up-to-date. pub struct NodeStatusCacheRefresher { + // Main stored data cache: NodeStatusCache, + fallback_caching_interval: Duration, + + // Sources for when refreshing data contract_cache: ValidatorCache, contract_cache_listener: watch::Receiver, - fallback_caching_interval: Duration, + storage: Option, } impl NodeStatusCacheRefresher { pub(crate) fn new( cache: NodeStatusCache, + fallback_caching_interval: Duration, contract_cache: ValidatorCache, contract_cache_listener: watch::Receiver, - fallback_caching_interval: Duration, + storage: Option, ) -> Self { Self { cache, + fallback_caching_interval, contract_cache, contract_cache_listener, - fallback_caching_interval, + storage, } } @@ -176,78 +230,154 @@ impl NodeStatusCacheRefresher { async fn refresh_cache(&self) -> Result<(), NodeStatusCacheError> { log::info!("Updating node status cache"); - let mixnode_bonds = self.contract_cache.mixnodes().await; - let params = self - .contract_cache - .interval_reward_params() - .await - .into_inner() - .ok_or(NodeStatusCacheError::SimulationFailed)?; - let inclusion_probabilities = compute_inclusion_probabilities(&mixnode_bonds, params) - .ok_or_else(|| { - error!( - "Failed to simulate selection probabilties for mixnodes, not updating cache" - ); - NodeStatusCacheError::SimulationFailed - })?; - self.cache.update_cache(inclusion_probabilities).await; + // Fetch contract cache data to work with + let mixnode_details = self.contract_cache.mixnodes().await; + let interval_reward_params = self.contract_cache.interval_reward_params().await; + let current_interval = self.contract_cache.current_interval().await; + + let rewarded_set = self.contract_cache.rewarded_set().await; + let active_set = self.contract_cache.active_set().await; + + let interval_reward_params = + interval_reward_params.ok_or(NodeStatusCacheError::SourceDataMissing)?; + let current_interval = current_interval.ok_or(NodeStatusCacheError::SourceDataMissing)?; + + // Compute inclusion probabilities + let inclusion_probabilities = InclusionProbabilities::compute( + &mixnode_details, + interval_reward_params, + ) + .ok_or_else(|| { + error!("Failed to simulate selection probabilties for mixnodes, not updating cache"); + NodeStatusCacheError::SimulationFailed + })?; + + // Create annotated data + let rewarded_set_node_status = to_rewarded_set_node_status(&rewarded_set, &active_set); + let mixnodes_annotated = self + .annotate_node_with_details( + mixnode_details, + interval_reward_params, + current_interval, + &rewarded_set_node_status, + ) + .await; + + // Create the annotated rewarded and active sets + let (rewarded_set, active_set) = + split_into_active_and_rewarded_set(&mixnodes_annotated, &rewarded_set_node_status); + + self.cache + .update_cache( + mixnodes_annotated, + rewarded_set, + active_set, + inclusion_probabilities, + ) + .await; Ok(()) } + + async fn get_performance_from_storage( + &self, + mix_id: MixId, + epoch: Interval, + ) -> Option { + self.storage + .as_ref()? + .get_average_mixnode_uptime_in_the_last_24hrs( + mix_id, + epoch.current_epoch_end_unix_timestamp(), + ) + .await + .ok() + .map(Into::into) + } + + async fn annotate_node_with_details( + &self, + mixnodes: Vec, + interval_reward_params: RewardingParams, + current_interval: Interval, + rewarded_set: &HashMap, + ) -> Vec { + let mut annotated = Vec::new(); + for mixnode in mixnodes { + let stake_saturation = mixnode + .rewarding_details + .bond_saturation(&interval_reward_params); + + let uncapped_stake_saturation = mixnode + .rewarding_details + .uncapped_bond_saturation(&interval_reward_params); + + // If the performance can't be obtained, because the validator-api was not started with + // the monitoring (and hence, storage), then reward estimates will be all zero + let performance = self + .get_performance_from_storage(mixnode.mix_id(), current_interval) + .await + .unwrap_or_default(); + + let rewarded_set_status = rewarded_set.get(&mixnode.mix_id()).copied(); + + let reward_estimate = compute_reward_estimate( + &mixnode, + performance, + rewarded_set_status, + interval_reward_params, + current_interval, + ); + + let (estimated_operator_apy, estimated_delegators_apy) = + compute_apy_from_reward(&mixnode, reward_estimate, current_interval); + + annotated.push(MixNodeBondAnnotated { + mixnode_details: mixnode, + stake_saturation, + uncapped_stake_saturation, + performance, + estimated_operator_apy, + estimated_delegators_apy, + }); + } + annotated + } } -fn compute_inclusion_probabilities( - mixnodes: &[MixNodeDetails], - params: RewardingParams, -) -> Option { - let active_set_size = params.active_set_size; - let standby_set_size = params.rewarded_set_size - active_set_size; - - // Unzip list of total bonds into ids and bonds. - // We need to go through this zip/unzip procedure to make sure we have matching identities - // for the input to the simulator, which assumes the identity is the position in the vec - let (ids, mixnode_total_bonds) = unzip_into_mixnode_ids_and_total_bonds(mixnodes); - - // Compute inclusion probabilitites and keep track of how long time it took. - let mut rng = rand::thread_rng(); - let results = inclusion_probability::simulate_selection_probability_mixnodes( - &mixnode_total_bonds, - active_set_size as usize, - standby_set_size as usize, - MAX_SIMULATION_SAMPLES, - Duration::from_secs(MAX_SIMULATION_TIME_SEC), - &mut rng, - ) - .tap_err(|err| error!("{err}")) - .ok()?; - - Some(InclusionProbabilities { - inclusion_probabilities: zip_ids_together_with_results(&ids, &results), - samples: results.samples, - elapsed: results.time, - delta_max: results.delta_max, - delta_l2: results.delta_l2, - }) -} - -fn unzip_into_mixnode_ids_and_total_bonds(mixnodes: &[MixNodeDetails]) -> (Vec, Vec) { - mixnodes +fn to_rewarded_set_node_status( + rewarded_set: &[MixNodeDetails], + active_set: &[MixNodeDetails], +) -> HashMap { + let mut rewarded_set_node_status: HashMap = rewarded_set .iter() - .map(|m| (m.mix_id(), truncate_decimal(m.total_stake()).u128())) - .unzip() + .map(|m| (m.mix_id(), RewardedSetNodeStatus::Standby)) + .collect(); + for mixnode in active_set { + *rewarded_set_node_status + .get_mut(&mixnode.mix_id()) + .expect("All active nodes are rewarded nodes") = RewardedSetNodeStatus::Active; + } + rewarded_set_node_status } -fn zip_ids_together_with_results( - ids: &[MixId], - results: &inclusion_probability::SelectionProbability, -) -> Vec { - ids.iter() - .zip(results.active_set_probability.iter()) - .zip(results.reserve_set_probability.iter()) - .map(|((&mix_id, a), r)| InclusionProbability { - mix_id, - in_active: *a, - in_reserve: *r, +fn split_into_active_and_rewarded_set( + mixnodes_annotated: &[MixNodeBondAnnotated], + rewarded_set_node_status: &HashMap, +) -> (Vec, Vec) { + let rewarded_set: Vec<_> = mixnodes_annotated + .iter() + .filter(|mixnode| rewarded_set_node_status.get(&mixnode.mix_id()).is_some()) + .cloned() + .collect(); + let active_set: Vec<_> = rewarded_set + .iter() + .filter(|mixnode| { + rewarded_set_node_status + .get(&mixnode.mix_id()) + .map_or(false, RewardedSetNodeStatus::is_active) }) - .collect() + .cloned() + .collect(); + (rewarded_set, active_set) } diff --git a/validator-api/src/node_status_api/cache/inclusion_probabilities.rs b/validator-api/src/node_status_api/cache/inclusion_probabilities.rs new file mode 100644 index 0000000000..21c1bdcab5 --- /dev/null +++ b/validator-api/src/node_status_api/cache/inclusion_probabilities.rs @@ -0,0 +1,89 @@ +use contracts_common::truncate_decimal; +use mixnet_contract_common::{MixId, MixNodeDetails, RewardingParams}; +use serde::Serialize; +use std::time::Duration; +use tap::TapFallible; +use validator_api_requests::models::InclusionProbability; + +const MAX_SIMULATION_SAMPLES: u64 = 5000; +const MAX_SIMULATION_TIME_SEC: u64 = 15; + +#[derive(Clone, Default, Serialize, schemars::JsonSchema)] +pub(crate) struct InclusionProbabilities { + pub inclusion_probabilities: Vec, + pub samples: u64, + pub elapsed: Duration, + pub delta_max: f64, + pub delta_l2: f64, +} + +impl InclusionProbabilities { + pub(crate) fn compute( + mixnodes: &[MixNodeDetails], + params: RewardingParams, + ) -> Option { + compute_inclusion_probabilities(mixnodes, params) + } + + pub(crate) fn node(&self, mix_id: MixId) -> Option<&InclusionProbability> { + self.inclusion_probabilities + .iter() + .find(|x| x.mix_id == mix_id) + } +} + +fn compute_inclusion_probabilities( + mixnodes: &[MixNodeDetails], + params: RewardingParams, +) -> Option { + let active_set_size = params.active_set_size; + let standby_set_size = params.rewarded_set_size - active_set_size; + + // Unzip list of total bonds into ids and bonds. + // We need to go through this zip/unzip procedure to make sure we have matching identities + // for the input to the simulator, which assumes the identity is the position in the vec + let (ids, mixnode_total_bonds) = unzip_into_mixnode_ids_and_total_bonds(mixnodes); + + // Compute inclusion probabilitites and keep track of how long time it took. + let mut rng = rand::thread_rng(); + let results = inclusion_probability::simulate_selection_probability_mixnodes( + &mixnode_total_bonds, + active_set_size as usize, + standby_set_size as usize, + MAX_SIMULATION_SAMPLES, + Duration::from_secs(MAX_SIMULATION_TIME_SEC), + &mut rng, + ) + .tap_err(|err| error!("{err}")) + .ok()?; + + Some(InclusionProbabilities { + inclusion_probabilities: zip_ids_together_with_results(&ids, &results), + samples: results.samples, + elapsed: results.time, + delta_max: results.delta_max, + delta_l2: results.delta_l2, + }) +} + +fn unzip_into_mixnode_ids_and_total_bonds(mixnodes: &[MixNodeDetails]) -> (Vec, Vec) { + mixnodes + .iter() + .map(|m| (m.mix_id(), truncate_decimal(m.total_stake()).u128())) + .unzip() +} + +fn zip_ids_together_with_results( + ids: &[MixId], + results: &inclusion_probability::SelectionProbability, +) -> Vec { + ids.iter() + .zip(results.active_set_probability.iter()) + .zip(results.reserve_set_probability.iter()) + .map(|((&mix_id, a), r)| InclusionProbability { + mix_id, + in_active: *a, + in_reserve: *r, + }) + .collect() +} diff --git a/validator-api/src/node_status_api/helpers.rs b/validator-api/src/node_status_api/helpers.rs index c0c2a544bd..4b084533bc 100644 --- a/validator-api/src/node_status_api/helpers.rs +++ b/validator-api/src/node_status_api/helpers.rs @@ -1,7 +1,6 @@ // Copyright 2021-2022 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::contract_cache::reward_estimate::compute_reward_estimate; use crate::contract_cache::Cache; use crate::node_status_api::models::ErrorResponse; use crate::storage::ValidatorApiStorage; @@ -12,11 +11,14 @@ use mixnet_contract_common::{Interval, MixId, RewardedSetNodeStatus}; use rocket::http::Status; use rocket::State; use validator_api_requests::models::{ - ComputeRewardEstParam, InclusionProbabilityResponse, MixnodeCoreStatusResponse, - MixnodeStatusReportResponse, MixnodeStatusResponse, MixnodeUptimeHistoryResponse, - RewardEstimationResponse, StakeSaturationResponse, UptimeResponse, + AllInclusionProbabilitiesResponse, ComputeRewardEstParam, InclusionProbabilityResponse, + MixNodeBondAnnotated, MixnodeCoreStatusResponse, MixnodeStatusReportResponse, + MixnodeStatusResponse, MixnodeUptimeHistoryResponse, RewardEstimationResponse, + StakeSaturationResponse, UptimeResponse, }; +use super::reward_estimate::compute_reward_estimate; + pub(crate) async fn _mixnode_report( storage: &ValidatorApiStorage, mix_id: MixId, @@ -62,17 +64,18 @@ pub(crate) async fn _get_mixnode_status( } pub(crate) async fn _get_mixnode_reward_estimation( - cache: &State, + cache: &State, + validator_cache: &State, mix_id: MixId, ) -> Result { let (mixnode, status) = cache.mixnode_details(mix_id).await; if let Some(mixnode) = mixnode { - let reward_params = cache.interval_reward_params().await; + let reward_params = validator_cache.interval_reward_params().await; let as_at = reward_params.timestamp(); let reward_params = reward_params .into_inner() .ok_or_else(|| ErrorResponse::new("server error", Status::InternalServerError))?; - let current_interval = cache + let current_interval = validator_cache .current_interval() .await .into_inner() @@ -117,17 +120,18 @@ async fn average_mixnode_performance( pub(crate) async fn _compute_mixnode_reward_estimation( user_reward_param: ComputeRewardEstParam, - cache: &ValidatorCache, + cache: &NodeStatusCache, + validator_cache: &ValidatorCache, mix_id: MixId, ) -> Result { let (mixnode, actual_status) = cache.mixnode_details(mix_id).await; if let Some(mut mixnode) = mixnode { - let reward_params = cache.interval_reward_params().await; + let reward_params = validator_cache.interval_reward_params().await; let as_at = reward_params.timestamp(); let reward_params = reward_params .into_inner() .ok_or_else(|| ErrorResponse::new("server error", Status::InternalServerError))?; - let current_interval = cache + let current_interval = validator_cache .current_interval() .await .into_inner() @@ -200,14 +204,15 @@ pub(crate) async fn _compute_mixnode_reward_estimation( } pub(crate) async fn _get_mixnode_stake_saturation( - cache: &ValidatorCache, + cache: &NodeStatusCache, + validator_cache: &ValidatorCache, mix_id: MixId, ) -> Result { let (mixnode, _) = cache.mixnode_details(mix_id).await; if let Some(mixnode) = mixnode { // Recompute the stake saturation just so that we can confidently state that the `as_at` // field is consistent and correct. Luckily this is very cheap. - let reward_params = cache.interval_reward_params().await; + let reward_params = validator_cache.interval_reward_params().await; let as_at = reward_params.timestamp(); let rewarding_params = reward_params .into_inner() @@ -266,3 +271,51 @@ pub(crate) async fn _get_mixnode_avg_uptime( avg_uptime: performance.round_to_integer(), }) } + +pub(crate) async fn _get_mixnode_inclusion_probabilities( + cache: &NodeStatusCache, +) -> Result { + if let Some(prob) = cache.inclusion_probabilities().await { + let as_at = prob.timestamp(); + let prob = prob.into_inner(); + Ok(AllInclusionProbabilitiesResponse { + inclusion_probabilities: prob.inclusion_probabilities, + samples: prob.samples, + elapsed: prob.elapsed, + delta_max: prob.delta_max, + delta_l2: prob.delta_l2, + as_at, + }) + } else { + Err(ErrorResponse::new( + "No data available".to_string(), + Status::ServiceUnavailable, + )) + } +} + +pub(crate) async fn _get_mixnodes_detailed(cache: &NodeStatusCache) -> Vec { + cache + .mixnodes_annotated() + .await + .unwrap_or_default() + .into_inner() +} + +pub(crate) async fn _get_rewarded_set_detailed( + cache: &NodeStatusCache, +) -> Vec { + cache + .rewarded_set_annotated() + .await + .unwrap_or_default() + .into_inner() +} + +pub(crate) async fn _get_active_set_detailed(cache: &NodeStatusCache) -> Vec { + cache + .active_set_annotated() + .await + .unwrap_or_default() + .into_inner() +} diff --git a/validator-api/src/node_status_api/mod.rs b/validator-api/src/node_status_api/mod.rs index 08cf612dba..cbcf1784f2 100644 --- a/validator-api/src/node_status_api/mod.rs +++ b/validator-api/src/node_status_api/mod.rs @@ -10,6 +10,7 @@ pub(crate) mod cache; pub(crate) mod helpers; pub(crate) mod local_guard; pub(crate) mod models; +pub(crate) mod reward_estimate; pub(crate) mod routes; pub(crate) mod uptime_updater; pub(crate) mod utils; @@ -37,6 +38,9 @@ pub(crate) fn node_status_routes( routes::get_mixnode_inclusion_probability, routes::get_mixnode_avg_uptime, routes::get_mixnode_inclusion_probabilities, + routes::get_mixnodes_detailed, + routes::get_rewarded_set_detailed, + routes::get_active_set_detailed, ] } else { // in the minimal variant we would not have access to endpoints relying on existence @@ -46,6 +50,9 @@ pub(crate) fn node_status_routes( routes::get_mixnode_stake_saturation, routes::get_mixnode_inclusion_probability, routes::get_mixnode_inclusion_probabilities, + routes::get_mixnodes_detailed, + routes::get_rewarded_set_detailed, + routes::get_active_set_detailed, ] } } diff --git a/validator-api/src/contract_cache/reward_estimate.rs b/validator-api/src/node_status_api/reward_estimate.rs similarity index 96% rename from validator-api/src/contract_cache/reward_estimate.rs rename to validator-api/src/node_status_api/reward_estimate.rs index 41e285a0a0..33bbd8320c 100644 --- a/validator-api/src/contract_cache/reward_estimate.rs +++ b/validator-api/src/node_status_api/reward_estimate.rs @@ -7,7 +7,7 @@ use mixnet_contract_common::reward_params::{NodeRewardParams, Performance, Rewar use mixnet_contract_common::rewarding::RewardEstimate; use mixnet_contract_common::{Interval, RewardedSetNodeStatus}; -pub fn compute_apy(epochs_in_year: Decimal, reward: Decimal, pledge_amount: Decimal) -> Decimal { +fn compute_apy(epochs_in_year: Decimal, reward: Decimal, pledge_amount: Decimal) -> Decimal { if pledge_amount.is_zero() { return Decimal::zero(); } diff --git a/validator-api/src/node_status_api/routes.rs b/validator-api/src/node_status_api/routes.rs index 7665f1bac1..037ca7b117 100644 --- a/validator-api/src/node_status_api/routes.rs +++ b/validator-api/src/node_status_api/routes.rs @@ -3,9 +3,10 @@ use super::NodeStatusCache; use crate::node_status_api::helpers::{ - _compute_mixnode_reward_estimation, _get_mixnode_avg_uptime, - _get_mixnode_inclusion_probability, _get_mixnode_reward_estimation, - _get_mixnode_stake_saturation, _get_mixnode_status, _mixnode_core_status_count, + _compute_mixnode_reward_estimation, _get_active_set_detailed, _get_mixnode_avg_uptime, + _get_mixnode_inclusion_probabilities, _get_mixnode_inclusion_probability, + _get_mixnode_reward_estimation, _get_mixnode_stake_saturation, _get_mixnode_status, + _get_mixnodes_detailed, _get_rewarded_set_detailed, _mixnode_core_status_count, _mixnode_report, _mixnode_uptime_history, }; use crate::node_status_api::models::ErrorResponse; @@ -19,9 +20,9 @@ use rocket_okapi::openapi; use validator_api_requests::models::{ AllInclusionProbabilitiesResponse, ComputeRewardEstParam, GatewayCoreStatusResponse, GatewayStatusReportResponse, GatewayUptimeHistoryResponse, InclusionProbabilityResponse, - MixnodeCoreStatusResponse, MixnodeStatusReportResponse, MixnodeStatusResponse, - MixnodeUptimeHistoryResponse, RewardEstimationResponse, StakeSaturationResponse, - UptimeResponse, + MixNodeBondAnnotated, MixnodeCoreStatusResponse, MixnodeStatusReportResponse, + MixnodeStatusResponse, MixnodeUptimeHistoryResponse, RewardEstimationResponse, + StakeSaturationResponse, UptimeResponse, }; #[openapi(tag = "status")] @@ -112,10 +113,13 @@ pub(crate) async fn get_mixnode_status( #[openapi(tag = "status")] #[get("/mixnode//reward-estimation")] pub(crate) async fn get_mixnode_reward_estimation( - cache: &State, + cache: &State, + validator_cache: &State, mix_id: MixId, ) -> Result, ErrorResponse> { - Ok(Json(_get_mixnode_reward_estimation(cache, mix_id).await?)) + Ok(Json( + _get_mixnode_reward_estimation(cache, validator_cache, mix_id).await?, + )) } #[openapi(tag = "status")] @@ -125,21 +129,31 @@ pub(crate) async fn get_mixnode_reward_estimation( )] pub(crate) async fn compute_mixnode_reward_estimation( user_reward_param: Json, - cache: &State, + cache: &State, + validator_cache: &State, mix_id: MixId, ) -> Result, ErrorResponse> { Ok(Json( - _compute_mixnode_reward_estimation(user_reward_param.into_inner(), cache, mix_id).await?, + _compute_mixnode_reward_estimation( + user_reward_param.into_inner(), + cache, + validator_cache, + mix_id, + ) + .await?, )) } #[openapi(tag = "status")] #[get("/mixnode//stake-saturation")] pub(crate) async fn get_mixnode_stake_saturation( - cache: &State, + cache: &State, + validator_cache: &State, mix_id: MixId, ) -> Result, ErrorResponse> { - Ok(Json(_get_mixnode_stake_saturation(cache, mix_id).await?)) + Ok(Json( + _get_mixnode_stake_saturation(cache, validator_cache, mix_id).await?, + )) } #[openapi(tag = "status")] @@ -168,21 +182,29 @@ pub(crate) async fn get_mixnode_avg_uptime( pub(crate) async fn get_mixnode_inclusion_probabilities( cache: &State, ) -> Result, ErrorResponse> { - if let Some(prob) = cache.inclusion_probabilities().await { - let as_at = prob.timestamp(); - let prob = prob.into_inner(); - Ok(Json(AllInclusionProbabilitiesResponse { - inclusion_probabilities: prob.inclusion_probabilities, - samples: prob.samples, - elapsed: prob.elapsed, - delta_max: prob.delta_max, - delta_l2: prob.delta_l2, - as_at, - })) - } else { - Err(ErrorResponse::new( - "No data available".to_string(), - Status::ServiceUnavailable, - )) - } + Ok(Json(_get_mixnode_inclusion_probabilities(cache).await?)) +} + +#[openapi(tag = "status")] +#[get("/mixnodes/detailed")] +pub async fn get_mixnodes_detailed( + cache: &State, +) -> Json> { + Json(_get_mixnodes_detailed(cache).await) +} + +#[openapi(tag = "status")] +#[get("/mixnodes/rewarded/detailed")] +pub async fn get_rewarded_set_detailed( + cache: &State, +) -> Json> { + Json(_get_rewarded_set_detailed(cache).await) +} + +#[openapi(tag = "status")] +#[get("/mixnodes/active/detailed")] +pub async fn get_active_set_detailed( + cache: &State, +) -> Json> { + Json(_get_active_set_detailed(cache).await) }