validator-api: create node status cache with selection probabilies (#1547)

* validator-api: create node status cache with selection probabilies

Create a node status cache to complement the contract cache. Initially
we store the simulated active set selection probabilities.

* validator-api: add validator cache watch channel

* changelog: add note

* validator-api: clippy fixes

* validator-api: fix clippy

* validator-api: additional fields to inclusion probabilities response

* selection chance: revert back to 3 buckets

* selection chance: revert buckets again

* rustfmt

* validator-api: remove the old get_mixnode_inclusion_probability

* node-status-cache: return error when refreshing

* inclusion-simulator: cap on wall clock time

* node status cache: tidy

validator-api/src/node_status_api/cache.rs

@@ -0,0 +1,244 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use rocket::fairing::AdHoc;
+use serde::Serialize;
+use tap::TapFallible;
+use tokio::{
+    sync::{watch, RwLock},
+    time,
+};
+
+use std::{sync::Arc, time::Duration};
+
+use mixnet_contract_common::{reward_params::EpochRewardParams, MixNodeBond};
+use task::ShutdownListener;
+use validator_api_requests::models::InclusionProbability;
+
+use crate::contract_cache::{Cache, CacheNotification, ValidatorCache};
+
+const CACHE_TIMOUT_MS: u64 = 100;
+const MAX_SIMULATION_SAMPLES: u64 = 5000;
+const MAX_SIMULATION_TIME_SEC: u64 = 15;
+
+enum NodeStatusCacheError {
+    SimulationFailed,
+}
+
+// A node status cache suitable for caching values computed in one sweep, such as active set
+// inclusion probabilities that are computed for all mixnodes at the same time.
+//
+// The cache can be triggered to update on contract cache changes, and/or periodically on a timer.
+#[derive(Clone)]
+pub struct NodeStatusCache {
+    inner: Arc<RwLock<NodeStatusCacheInner>>,
+}
+
+struct NodeStatusCacheInner {
+    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, id: &str) -> Option<&InclusionProbability> {
+        self.inclusion_probabilities.iter().find(|x| x.id == id)
+    }
+}
+
+impl NodeStatusCache {
+    fn new() -> Self {
+        NodeStatusCache {
+            inner: Arc::new(RwLock::new(NodeStatusCacheInner::new())),
+        }
+    }
+
+    pub fn stage() -> AdHoc {
+        AdHoc::on_ignite("Node Status Cache", |rocket| async {
+            rocket.manage(Self::new())
+        })
+    }
+
+    async fn update_cache(&self, inclusion_probabilities: InclusionProbabilities) {
+        match time::timeout(Duration::from_millis(CACHE_TIMOUT_MS), self.inner.write()).await {
+            Ok(mut cache) => {
+                cache
+                    .inclusion_probabilities
+                    .update(inclusion_probabilities);
+            }
+            Err(e) => error!("{e}"),
+        }
+    }
+
+    pub(crate) async fn inclusion_probabilities(&self) -> Option<Cache<InclusionProbabilities>> {
+        match time::timeout(Duration::from_millis(CACHE_TIMOUT_MS), self.inner.read()).await {
+            Ok(cache) => Some(cache.inclusion_probabilities.clone()),
+            Err(e) => {
+                error!("{e}");
+                None
+            }
+        }
+    }
+}
+
+impl NodeStatusCacheInner {
+    pub fn new() -> Self {
+        Self {
+            inclusion_probabilities: Default::default(),
+        }
+    }
+}
+
+// Long running task responsible of keeping the cache up-to-date.
+pub struct NodeStatusCacheRefresher {
+    cache: NodeStatusCache,
+    contract_cache: ValidatorCache,
+    contract_cache_listener: watch::Receiver<CacheNotification>,
+    fallback_caching_interval: Duration,
+}
+
+impl NodeStatusCacheRefresher {
+    pub(crate) fn new(
+        cache: NodeStatusCache,
+        contract_cache: ValidatorCache,
+        contract_cache_listener: watch::Receiver<CacheNotification>,
+        fallback_caching_interval: Duration,
+    ) -> Self {
+        Self {
+            cache,
+            contract_cache,
+            contract_cache_listener,
+            fallback_caching_interval,
+        }
+    }
+
+    pub async fn run(&mut self, mut shutdown: ShutdownListener) {
+        let mut fallback_interval = time::interval(self.fallback_caching_interval);
+        while !shutdown.is_shutdown() {
+            tokio::select! {
+                // Update node status cache when the contract cache / validator cache is updated
+                Ok(_) = self.contract_cache_listener.changed() => {
+                    self.update_on_notify(&mut fallback_interval).await;
+                }
+                // ... however, if we don't receive any notifications we fall back to periodic
+                // refreshes
+                _ = fallback_interval.tick() => {
+                    self.update_on_timer().await;
+                }
+                _ = shutdown.recv() => {
+                    log::trace!("NodeStatusCacheRefresher: Received shutdown");
+                }
+            }
+        }
+        log::info!("NodeStatusCacheRefresher: Exiting");
+    }
+
+    async fn update_on_notify(&self, fallback_interval: &mut time::Interval) {
+        log::debug!(
+            "Validator cache event detected: {:?}",
+            &*self.contract_cache_listener.borrow(),
+        );
+        let _ = self.refresh_cache().await;
+        fallback_interval.reset();
+    }
+
+    async fn update_on_timer(&self) {
+        log::debug!("Timed trigger for the node status cache");
+        let have_contract_cache_data =
+            *self.contract_cache_listener.borrow() != CacheNotification::Start;
+
+        if have_contract_cache_data {
+            let _ = self.refresh_cache().await;
+        } else {
+            log::trace!(
+                "Skipping updating node status cache, is the contract cache not yet available?"
+            );
+        }
+    }
+
+    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.epoch_reward_params().await.into_inner();
+        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;
+        Ok(())
+    }
+}
+
+fn compute_inclusion_probabilities(
+    mixnode_bonds: &[MixNodeBond],
+    params: EpochRewardParams,
+) -> Option<InclusionProbabilities> {
+    let active_set_size = params
+        .active_set_size()
+        .try_into()
+        .tap_err(|e| error!("Active set size unexpectantly large: {e}"))
+        .ok()?;
+    let standby_set_size = (params.rewarded_set_size() - params.active_set_size())
+        .try_into()
+        .tap_err(|e| error!("Active set size larger than rewarded set size, a contradiction: {e}"))
+        .ok()?;
+
+    // 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(mixnode_bonds);
+
+    // Compute inclusion probabilitites and keep track of how long time it took.
+    let results = inclusion_probability::simulate_selection_probability_mixnodes(
+        &mixnode_total_bonds,
+        active_set_size,
+        standby_set_size,
+        MAX_SIMULATION_SAMPLES,
+        Duration::from_secs(MAX_SIMULATION_TIME_SEC),
+    )
+    .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(
+    mixnode_bonds: &[MixNodeBond],
+) -> (Vec<&String>, Vec<u128>) {
+    mixnode_bonds
+        .iter()
+        .filter_map(|m| m.total_bond().map(|b| (m.identity(), b)))
+        .unzip()
+}
+
+fn zip_ids_together_with_results(
+    ids: &[&String],
+    results: &inclusion_probability::SelectionProbability,
+) -> Vec<InclusionProbability> {
+    ids.iter()
+        .zip(results.active_set_probability.iter())
+        .zip(results.reserve_set_probability.iter())
+        .map(|((id, a), r)| InclusionProbability {
+            id: (*id).to_string(),
+            in_active: *a,
+            in_reserve: *r,
+        })
+        .collect()
+}

validator-api/src/node_status_api/mod.rs

@@ -1,11 +1,14 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+pub(crate) use cache::{NodeStatusCache, NodeStatusCacheRefresher};
+
 use okapi::openapi3::OpenApi;
 use rocket::Route;
 use rocket_okapi::{openapi_get_routes_spec, settings::OpenApiSettings};
 use std::time::Duration;
 
+pub(crate) mod cache;
 pub(crate) mod local_guard;
 pub(crate) mod models;
 pub(crate) mod routes;
@@ -35,6 +38,7 @@ pub(crate) fn node_status_routes(
             routes::get_mixnode_inclusion_probability,
             routes::get_mixnode_avg_uptime,
             routes::get_mixnode_avg_uptimes,
+            routes::get_mixnode_inclusion_probabilities,
         ]
     } else {
         // in the minimal variant we would not have access to endpoints relying on existence
@@ -43,6 +47,7 @@ pub(crate) fn node_status_routes(
             routes::get_mixnode_status,
             routes::get_mixnode_stake_saturation,
             routes::get_mixnode_inclusion_probability,
+            routes::get_mixnode_inclusion_probabilities,
         ]
     }
 }

validator-api/src/node_status_api/routes.rs

@@ -1,6 +1,7 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use crate::contract_cache::Cache;
 use crate::node_status_api::models::{
     ErrorResponse, GatewayStatusReport, GatewayUptimeHistory, MixnodeStatusReport,
     MixnodeUptimeHistory,
@@ -16,11 +17,12 @@ use rocket_okapi::openapi;
 use schemars::JsonSchema;
 use serde::Deserialize;
 use validator_api_requests::models::{
-    CoreNodeStatusResponse, InclusionProbabilityResponse, MixnodeStatusResponse,
-    RewardEstimationResponse, StakeSaturationResponse, UptimeResponse,
+    AllInclusionProbabilitiesResponse, CoreNodeStatusResponse, InclusionProbabilityResponse,
+    MixnodeStatusResponse, RewardEstimationResponse, StakeSaturationResponse, UptimeResponse,
 };
 
 use super::models::Uptime;
+use super::NodeStatusCache;
 
 async fn average_mixnode_uptime(
     identity: &str,
@@ -307,43 +309,21 @@ pub(crate) async fn get_mixnode_stake_saturation(
 #[openapi(tag = "status")]
 #[get("/mixnode/<identity>/inclusion-probability")]
 pub(crate) async fn get_mixnode_inclusion_probability(
-    cache: &State<ValidatorCache>,
+    node_status_cache: &State<NodeStatusCache>,
     identity: String,
 ) -> Json<Option<InclusionProbabilityResponse>> {
-    let mixnodes = cache.mixnodes().await;
-    let rewarding_params = cache.epoch_reward_params().await.into_inner();
-
-    if let Some(target_mixnode) = mixnodes.iter().find(|x| x.identity() == &identity) {
-        let total_bonded_tokens = mixnodes
-            .iter()
-            .fold(0u128, |acc, x| acc + x.total_bond().unwrap_or_default())
-            as f64;
-
-        let rewarded_set_size = rewarding_params.rewarded_set_size() as f64;
-        let active_set_size = rewarding_params.active_set_size() as f64;
-
-        let prob_one_draw =
-            target_mixnode.total_bond().unwrap_or_default() as f64 / total_bonded_tokens;
-        // Chance to be selected in any draw for active set
-        let prob_active_set = if mixnodes.len() <= active_set_size as usize {
-            1.0
-        } else {
-            active_set_size * prob_one_draw
-        };
-        // This is likely slightly too high, as we're not correcting form them not being selected in active, should be chance to be selected, minus the chance for being not selected in reserve
-        let prob_reserve_set = if mixnodes.len() <= rewarded_set_size as usize {
-            1.0
-        } else {
-            (rewarded_set_size - active_set_size) * prob_one_draw
-        };
-
-        Json(Some(InclusionProbabilityResponse {
-            in_active: prob_active_set.into(),
-            in_reserve: prob_reserve_set.into(),
-        }))
-    } else {
-        Json(None)
-    }
+    node_status_cache
+        .inclusion_probabilities()
+        .await
+        .map(Cache::into_inner)
+        .and_then(|p| p.node(&identity).cloned())
+        .map(|p| {
+            Json(Some(InclusionProbabilityResponse {
+                in_active: p.in_active.into(),
+                in_reserve: p.in_reserve.into(),
+            }))
+        })
+        .unwrap_or(Json(None))
 }
 
 #[openapi(tag = "status")]
@@ -384,3 +364,27 @@ pub(crate) async fn get_mixnode_avg_uptimes(
 
     Ok(Json(response))
 }
+
+#[openapi(tag = "status")]
+#[get("/mixnodes/inclusion_probability")]
+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,
+        ))
+    }
+}