WIP

validator-api: remove storage dependency in contract cache (#1685 )
* validator-api: remove storage dependency in contract cache * validator-client: update detailed routes * contract_cache: forward to new endpoints for compat * Move reward_estimate
2022-12-05 12:04:50 +01:00 · 2022-12-02 14:40:19 +01:00 · 2022-12-02 12:36:06 +01:00 · 2022-12-02 09:23:34 +01:00 · 2022-12-01 23:09:37 +01:00
20 changed files with 713 additions and 390 deletions
@@ -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);
@@ -90,7 +90,6 @@ async fn register_with_gateway(
        gateway.owner.clone(),
        our_identity.clone(),
        timeout,
-        None,
    );
    gateway_client
        .establish_connection()
@@ -11,7 +11,8 @@ use client_core::client::base_client::{BaseClientBuilder, ClientInput, ClientOut
 use client_core::client::key_manager::KeyManager;
 use client_core::config::persistence::key_pathfinder::ClientKeyPathfinder;
 use futures::channel::mpsc;
-use futures::StreamExt;
+use futures::future::BoxFuture;
+use futures::{FutureExt, StreamExt};
 use gateway_client::bandwidth::BandwidthController;
 use log::*;
 use nymsphinx::addressing::clients::Recipient;
@@ -103,25 +104,56 @@ impl NymClient {
            shared_lane_queue_lengths,
            shutdown.clone(),
        );
-        tokio::spawn(async move {
-            // Ideally we should have a fully fledged task manager to check for errors in all
-            // tasks.
-            // However, pragmatically, we start out by at least reporting errors for some of the
-            // tasks that interact with the outside world and can fail in normal operation, such as
-            // network issues.
-            // TODO: replace this by a generic solution, such as a task manager that stores all
-            // JoinHandles of all spawned tasks.
-            if let Err(res) = sphinx_socks
-                .serve(
-                    input_sender,
-                    received_buffer_request_sender,
-                    connection_command_sender,
-                )
-                .await
-            {
-                shutdown.send_we_stopped(Box::new(res));
-            }
-        });
+        //tokio::spawn(async move {
+        //    // Ideally we should have a fully fledged task manager to check for errors in all
+        //    // tasks.
+        //    // However, pragmatically, we start out by at least reporting errors for some of the
+        //    // tasks that interact with the outside world and can fail in normal operation, such as
+        //    // network issues.
+        //    // TODO: replace this by a generic solution, such as a task manager that stores all
+        //    // JoinHandles of all spawned tasks.
+        //    if let Err(res) = sphinx_socks
+        //        .serve(
+        //            input_sender,
+        //            received_buffer_request_sender,
+        //            connection_command_sender,
+        //        )
+        //        .await
+        //    {
+        //        shutdown.send_we_stopped(Box::new(res));
+        //    }
+        //});
+
+        //let f = async move {
+        //    // Ideally we should have a fully fledged task manager to check for errors in all
+        //    // tasks.
+        //    // However, pragmatically, we start out by at least reporting errors for some of the
+        //    // tasks that interact with the outside world and can fail in normal operation, such as
+        //    // network issues.
+        //    // TODO: replace this by a generic solution, such as a task manager that stores all
+        //    // JoinHandles of all spawned tasks.
+        //    if let Err(res) = sphinx_socks
+        //        .serve(
+        //            input_sender,
+        //            received_buffer_request_sender,
+        //            connection_command_sender,
+        //        )
+        //        .await
+        //    {
+        //        shutdown.send_we_stopped(Box::new(res));
+        //    }
+        //};
+
+        let ff = sphinx_socks
+            .serve(
+                input_sender,
+                received_buffer_request_sender,
+                connection_command_sender,
+            )
+            .boxed();
+        //let b = Box::pin(ff);
+        //let box_fut = BoxFuture::new(b);
+        spawn_with_return(ff);
    }

    /// blocking version of `start` method. Will run forever (or until SIGINT is sent)
@@ -214,3 +246,31 @@ impl NymClient {
        Ok(started_client.shutdown_notifier)
    }
 }
+
+//fn spawn_with_return<F, T, S>(future: F)
+//where
+//    F: std::future::Future<Output = Result<T, S>> + Send + 'static,
+//    F::Output: Send + 'static,
+//    //S: 'static,
+//    //T: 'static,
+//{
+//    let f = async move {
+//        if let Err(_err) = future.await {
+//            println!("error");
+//        }
+//    };
+//    tokio::spawn(f);
+//}
+
+fn spawn_with_return<T, E>(future: BoxFuture<'static, Result<T, E>>)
+where
+    T: 'static,
+    E: 'static,
+{
+    let f = async move {
+        if let Err(_err) = future.await {
+            println!("error");
+        }
+    };
+    tokio::spawn(f);
+}
@@ -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<ShutdownListener>,
+    shutdown: ShutdownListener,
 }

 impl GatewayClient {
@@ -85,7 +83,7 @@ impl GatewayClient {
        ack_sender: AcknowledgementSender,
        response_timeout_duration: Duration,
        bandwidth_controller: Option<BandwidthController<PersistentStorage>>,
-        shutdown: Option<ShutdownListener>,
+        shutdown: ShutdownListener,
    ) -> Self {
        GatewayClient {
            authenticated: false,
@@ -130,7 +128,6 @@ impl GatewayClient {
        gateway_owner: String,
        local_identity: Arc<identity::KeyPair>,
        response_timeout_duration: Duration,
-        shutdown: Option<ShutdownListener>,
    ) -> 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(),
                    )
                }
@@ -21,14 +21,14 @@ pub type AcknowledgementReceiver = mpsc::UnboundedReceiver<Vec<Vec<u8>>>;
 pub struct PacketRouter {
    ack_sender: AcknowledgementSender,
    mixnet_message_sender: MixnetMessageSender,
-    shutdown: Option<ShutdownListener>,
+    shutdown: ShutdownListener,
 }

 impl PacketRouter {
    pub fn new(
        ack_sender: AcknowledgementSender,
        mixnet_message_sender: MixnetMessageSender,
-        shutdown: Option<ShutdownListener>,
+        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
@@ -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<SharedKeys>,
-        #[cfg(not(target_arch = "wasm32"))] shutdown: Option<ShutdownListener>,
+        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()
            {
@@ -136,7 +136,12 @@ impl Client {
        &self,
    ) -> Result<Vec<MixNodeBondAnnotated>, 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,
@@ -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::*;
@@ -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/"
@@ -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<C> {
    cache: ValidatorCache,
    caching_interval: Duration,

-    // Readonly: some of the quantities cached depends on values from the storage.
-    storage: Option<ValidatorApiStorage>,
-
    // Notify listeners that the cache has been updated
    update_notifier: watch::Sender<CacheNotification>,
 }
@@ -56,14 +50,14 @@ pub struct ValidatorCache {
 }

 struct ValidatorCacheInner {
-    mixnodes: Cache<Vec<MixNodeBondAnnotated>>,
+    mixnodes: Cache<Vec<MixNodeDetails>>,
    gateways: Cache<Vec<GatewayBond>>,

    mixnodes_blacklist: Cache<HashSet<MixId>>,
    gateways_blacklist: Cache<HashSet<IdentityKey>>,

-    rewarded_set: Cache<Vec<MixNodeBondAnnotated>>,
-    active_set: Cache<Vec<MixNodeBondAnnotated>>,
+    rewarded_set: Cache<Vec<MixNodeDetails>>,
+    active_set: Cache<Vec<MixNodeDetails>>,

    current_reward_params: Cache<Option<RewardingParams>>,
    current_interval: Cache<Option<Interval>>,
@@ -102,90 +96,33 @@ impl<T: Clone> Cache<T> {
    }
 }

+impl<T> Deref for Cache<T> {
+    type Target = T;
+
+    fn deref(&self) -> &Self::Target {
+        &self.value
+    }
+}
+
 impl<C> ValidatorCacheRefresher<C> {
    pub(crate) fn new(
        nymd_client: Client<C>,
        caching_interval: Duration,
        cache: ValidatorCache,
-        storage: Option<ValidatorApiStorage>,
    ) -> 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<Performance> {
-        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<CacheNotification> {
        self.update_notifier.subscribe()
    }

-    async fn annotate_node_with_details(
-        &self,
-        mixnodes: Vec<MixNodeDetails>,
-        interval_reward_params: RewardingParams,
-        current_interval: Interval,
-        rewarded_set: &HashMap<MixId, RewardedSetNodeStatus>,
-    ) -> Vec<MixNodeBondAnnotated> {
-        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<MixId, RewardedSetNodeStatus>
    where
        C: CosmWasmClient + Sync + Send,
@@ -198,9 +135,9 @@ impl<C> ValidatorCacheRefresher<C> {
    }

    fn collect_rewarded_and_active_set_details(
-        all_mixnodes: &[MixNodeBondAnnotated],
+        all_mixnodes: &[MixNodeDetails],
        rewarded_set_nodes: &HashMap<MixId, RewardedSetNodeStatus>,
-    ) -> (Vec<MixNodeBondAnnotated>, Vec<MixNodeBondAnnotated>) {
+    ) -> (Vec<MixNodeDetails>, Vec<MixNodeDetails>) {
        let mut active_set = Vec::new();
        let mut rewarded_set = Vec::new();

@@ -226,14 +163,10 @@ impl<C> ValidatorCacheRefresher<C> {
        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<MixNodeBondAnnotated>,
+        mixnodes: Vec<MixNodeDetails>,
        gateways: Vec<GatewayBond>,
-        rewarded_set: Vec<MixNodeBondAnnotated>,
-        active_set: Vec<MixNodeBondAnnotated>,
+        rewarded_set: Vec<MixNodeDetails>,
+        active_set: Vec<MixNodeDetails>,
        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<MixNodeBondAnnotated> {
+    pub async fn mixnodes(&self) -> Vec<MixNodeDetails> {
        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<MixNodeDetails> {
-        self.mixnodes_detailed()
-            .await
-            .into_iter()
-            .map(|bond| bond.mixnode_details)
-            .collect()
-    }
-
    pub async fn mixnodes_basic(&self) -> Vec<MixNodeBond> {
        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<Vec<MixNodeBondAnnotated>> {
+    pub async fn rewarded_set(&self) -> Cache<Vec<MixNodeDetails>> {
        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<MixNodeDetails> {
-        self.rewarded_set_detailed()
-            .await
-            .value
-            .into_iter()
-            .map(|bond| bond.mixnode_details)
-            .collect()
-    }
-
-    pub async fn active_set_detailed(&self) -> Cache<Vec<MixNodeBondAnnotated>> {
+    pub async fn active_set(&self) -> Cache<Vec<MixNodeDetails>> {
        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<MixNodeDetails> {
-        self.active_set_detailed()
-            .await
-            .value
-            .into_iter()
-            .map(|bond| bond.mixnode_details)
-            .collect()
-    }
-
    pub(crate) async fn interval_reward_params(&self) -> Cache<Option<RewardingParams>> {
        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<MixNodeBondAnnotated>, MixnodeStatus) {
+    pub async fn mixnode_details(&self, mix_id: MixId) -> (Option<MixNodeDetails>, 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 {
@@ -1,15 +1,21 @@
 // Copyright 2021-2022 - Nym Technologies SA <contact@nymtech.net>
 // 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<ValidatorCache>) -> Json<Vec<MixNodeDeta
    Json(cache.mixnodes().await)
 }

+// DEPRECATED: this endpoint now lives in `node_status_api`. Once all consumers are updated,
+// replace this with
+// ```
+//  pub fn get_mixnodes_detailed() -> Redirect {
+//      Redirect::to(uri!("/v1/status/mixnodes/detailed"))
+//  }
+// ```
 #[openapi(tag = "contract-cache")]
 #[get("/mixnodes/detailed")]
 pub async fn get_mixnodes_detailed(
-    cache: &State<ValidatorCache>,
+    cache: &State<NodeStatusCache>,
 ) -> Json<Vec<MixNodeBondAnnotated>> {
-    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<ValidatorCache>) -> Json<Vec<GatewayBond
 #[openapi(tag = "contract-cache")]
 #[get("/mixnodes/rewarded")]
 pub async fn get_rewarded_set(cache: &State<ValidatorCache>) -> Json<Vec<MixNodeDetails>> {
-    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<ValidatorCache>,
+    cache: &State<NodeStatusCache>,
 ) -> Json<Vec<MixNodeBondAnnotated>> {
-    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<ValidatorCache>) -> Json<Vec<MixNodeDetails>> {
-    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<ValidatorCache>,
+    cache: &State<NodeStatusCache>,
 ) -> Json<Vec<MixNodeBondAnnotated>> {
-    Json(cache.active_set_detailed().await.value)
+    Json(_get_active_set_detailed(cache).await)
 }

 #[openapi(tag = "contract-cache")]
@@ -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()
@@ -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::<ValidatorApiStorage>().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 });
@@ -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
@@ -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<RwLock<NodeStatusCacheInner>>,
 }

+#[derive(Default)]
 struct NodeStatusCacheInner {
+    mixnodes_annotated: Cache<Vec<MixNodeBondAnnotated>>,
+    rewarded_set_annotated: Cache<Vec<MixNodeBondAnnotated>>,
+    active_set_annotated: Cache<Vec<MixNodeBondAnnotated>>,
+
+    // Estimated active set inclusion probabilities from Monte Carlo simulation
    inclusion_probabilities: Cache<InclusionProbabilities>,
 }

-#[derive(Clone, Default, Serialize, schemars::JsonSchema)]
-pub(crate) struct InclusionProbabilities {
-    pub inclusion_probabilities: Vec<InclusionProbability>,
-    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<MixNodeBondAnnotated>,
+        rewarded_set: Vec<MixNodeBondAnnotated>,
+        active_set: Vec<MixNodeBondAnnotated>,
+        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<Cache<InclusionProbabilities>> {
+    async fn get_cache<T>(
+        &self,
+        fn_arg: impl FnOnce(RwLockReadGuard<'_, NodeStatusCacheInner>) -> Cache<T>,
+    ) -> Option<Cache<T>> {
        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<Cache<Vec<MixNodeBondAnnotated>>> {
+        self.get_cache(|c| c.mixnodes_annotated.clone()).await
+    }
+
+    pub(crate) async fn rewarded_set_annotated(&self) -> Option<Cache<Vec<MixNodeBondAnnotated>>> {
+        self.get_cache(|c| c.rewarded_set_annotated.clone()).await
+    }
+
+    pub(crate) async fn active_set_annotated(&self) -> Option<Cache<Vec<MixNodeBondAnnotated>>> {
+        self.get_cache(|c| c.active_set_annotated.clone()).await
+    }
+
+    pub(crate) async fn inclusion_probabilities(&self) -> Option<Cache<InclusionProbabilities>> {
+        self.get_cache(|c| c.inclusion_probabilities.clone()).await
+    }
+
+    pub async fn mixnode_details(
+        &self,
+        mix_id: MixId,
+    ) -> (Option<MixNodeBondAnnotated>, 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<CacheNotification>,
-    fallback_caching_interval: Duration,
+    storage: Option<ValidatorApiStorage>,
 }

 impl NodeStatusCacheRefresher {
    pub(crate) fn new(
        cache: NodeStatusCache,
+        fallback_caching_interval: Duration,
        contract_cache: ValidatorCache,
        contract_cache_listener: watch::Receiver<CacheNotification>,
-        fallback_caching_interval: Duration,
+        storage: Option<ValidatorApiStorage>,
    ) -> 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<Performance> {
+        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<MixNodeDetails>,
+        interval_reward_params: RewardingParams,
+        current_interval: Interval,
+        rewarded_set: &HashMap<MixId, RewardedSetNodeStatus>,
+    ) -> Vec<MixNodeBondAnnotated> {
+        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<InclusionProbabilities> {
-    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<MixId>, Vec<u128>) {
-    mixnodes
+fn to_rewarded_set_node_status(
+    rewarded_set: &[MixNodeDetails],
+    active_set: &[MixNodeDetails],
+) -> HashMap<MixId, RewardedSetNodeStatus> {
+    let mut rewarded_set_node_status: HashMap<MixId, RewardedSetNodeStatus> = 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<InclusionProbability> {
-    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<u32, RewardedSetNodeStatus>,
+) -> (Vec<MixNodeBondAnnotated>, Vec<MixNodeBondAnnotated>) {
+    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)
 }
@@ -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<InclusionProbability>,
+    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<InclusionProbabilities> {
+        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<InclusionProbabilities> {
+    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<MixId>, Vec<u128>) {
+    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<InclusionProbability> {
+    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()
+}
@@ -1,7 +1,6 @@
 // Copyright 2021-2022 - Nym Technologies SA <contact@nymtech.net>
 // 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<ValidatorCache>,
+    cache: &State<NodeStatusCache>,
+    validator_cache: &State<ValidatorCache>,
    mix_id: MixId,
 ) -> Result<RewardEstimationResponse, ErrorResponse> {
    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<RewardEstimationResponse, ErrorResponse> {
    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<StakeSaturationResponse, ErrorResponse> {
    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<AllInclusionProbabilitiesResponse, ErrorResponse> {
+    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<MixNodeBondAnnotated> {
+    cache
+        .mixnodes_annotated()
+        .await
+        .unwrap_or_default()
+        .into_inner()
+}
+
+pub(crate) async fn _get_rewarded_set_detailed(
+    cache: &NodeStatusCache,
+) -> Vec<MixNodeBondAnnotated> {
+    cache
+        .rewarded_set_annotated()
+        .await
+        .unwrap_or_default()
+        .into_inner()
+}
+
+pub(crate) async fn _get_active_set_detailed(cache: &NodeStatusCache) -> Vec<MixNodeBondAnnotated> {
+    cache
+        .active_set_annotated()
+        .await
+        .unwrap_or_default()
+        .into_inner()
+}
@@ -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,
        ]
    }
 }
@@ -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();
    }
@@ -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/<mix_id>/reward-estimation")]
 pub(crate) async fn get_mixnode_reward_estimation(
-    cache: &State<ValidatorCache>,
+    cache: &State<NodeStatusCache>,
+    validator_cache: &State<ValidatorCache>,
    mix_id: MixId,
 ) -> Result<Json<RewardEstimationResponse>, 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<ComputeRewardEstParam>,
-    cache: &State<ValidatorCache>,
+    cache: &State<NodeStatusCache>,
+    validator_cache: &State<ValidatorCache>,
    mix_id: MixId,
 ) -> Result<Json<RewardEstimationResponse>, 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/<mix_id>/stake-saturation")]
 pub(crate) async fn get_mixnode_stake_saturation(
-    cache: &State<ValidatorCache>,
+    cache: &State<NodeStatusCache>,
+    validator_cache: &State<ValidatorCache>,
    mix_id: MixId,
 ) -> Result<Json<StakeSaturationResponse>, 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<NodeStatusCache>,
 ) -> Result<Json<AllInclusionProbabilitiesResponse>, 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<NodeStatusCache>,
+) -> Json<Vec<MixNodeBondAnnotated>> {
+    Json(_get_mixnodes_detailed(cache).await)
+}
+
+#[openapi(tag = "status")]
+#[get("/mixnodes/rewarded/detailed")]
+pub async fn get_rewarded_set_detailed(
+    cache: &State<NodeStatusCache>,
+) -> Json<Vec<MixNodeBondAnnotated>> {
+    Json(_get_rewarded_set_detailed(cache).await)
+}
+
+#[openapi(tag = "status")]
+#[get("/mixnodes/active/detailed")]
+pub async fn get_active_set_detailed(
+    cache: &State<NodeStatusCache>,
+) -> Json<Vec<MixNodeBondAnnotated>> {
+    Json(_get_active_set_detailed(cache).await)
 }
Author	SHA1	Message	Date
Jon Häggblad	16cf6581ad	WIP	2022-12-05 12:04:50 +01:00
Jon Häggblad	6943271942	validator-api: remove storage dependency in contract cache (#1685 ) * validator-api: remove storage dependency in contract cache * validator-client: update detailed routes * contract_cache: forward to new endpoints for compat * Move reward_estimate	2022-12-02 14:40:19 +01:00
Jon Häggblad	a872b478d2	gateway-client: handle shutdown listener (#1829 ) * WIP * WIP: try another approach * WIP * Reworked * Tidy * fix	2022-12-02 12:36:06 +01:00
Jon Häggblad	11051ba929	validator-api: add missing shortform for --config-env-file (#1830 )	2022-12-02 09:23:34 +01:00
Jon Häggblad	10e28c6e5c	envs/mainnet: update to latest mixnet contract and nymd validator url	2022-12-01 23:09:37 +01:00