Feature/active sets (#764)

* Added separate gateway active set size

* Grabbing contract state

* Defined PartialOrd on MixnodeBond and GatewayBond

* Some initial stub for active set

* Unit tests for mixnode and gateway bond partialord implementation

* Obtaining active sets

* Active nodes routes

* Additional methods on validator client

* Added state migration

* Feature locking unused import

* Fixed State test fixture

* Included block height for partial_ord

* Missing post-merge imports

* api on the client for active nodes

* Native/socks5/wasm clients using active nodes

* Rewarding only active nodes

* Updated validator client StateParams definition

* Gateway active set size

* Contract migration update

* Cargo fmt

* Updated TauriStateParams

* [ci skip] Generate TS types

Co-authored-by: jstuczyn <jstuczyn@users.noreply.github.com>

validator-api/src/cache/mod.rs

@@ -4,10 +4,10 @@
 use crate::nymd_client::Client;
 use anyhow::Result;
 use config::defaults::VALIDATOR_API_VERSION;
-use mixnet_contract::{GatewayBond, MixNodeBond};
+use mixnet_contract::{GatewayBond, MixNodeBond, StateParams};
 use rocket::fairing::AdHoc;
 use serde::Serialize;
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tokio::sync::RwLock;
@@ -31,6 +31,12 @@ struct ValidatorCacheInner {
     initialised: AtomicBool,
     mixnodes: RwLock<Cache<Vec<MixNodeBond>>>,
     gateways: RwLock<Cache<Vec<GatewayBond>>>,
+
+    active_mixnodes_available: AtomicBool,
+    active_gateways_available: AtomicBool,
+
+    current_mixnode_active_set_size: AtomicUsize,
+    current_gateway_active_set_size: AtomicUsize,
 }
 
 #[derive(Default, Serialize, Clone)]
@@ -75,13 +81,17 @@ impl<C> ValidatorCacheRefresher<C> {
             self.nymd_client.get_gateways()
         )?;
 
+        let state_params = self.nymd_client.get_state_params().await?;
+
         info!(
             "Updating validator cache. There are {} mixnodes and {} gateways",
             mixnodes.len(),
             gateways.len()
         );
 
-        self.cache.update_cache(mixnodes, gateways).await;
+        self.cache
+            .update_cache(mixnodes, gateways, state_params)
+            .await;
 
         Ok(())
     }
@@ -117,12 +127,95 @@ impl ValidatorCache {
             rocket.manage(Self::new()).mount(
                 // this format! is so ugly...
                 format!("/{}", VALIDATOR_API_VERSION),
-                routes![routes::get_mixnodes, routes::get_gateways],
+                routes![
+                    routes::get_mixnodes,
+                    routes::get_gateways,
+                    routes::get_active_mixnodes,
+                    routes::get_active_gateways
+                ],
             )
         })
     }
 
-    async fn update_cache(&self, mixnodes: Vec<MixNodeBond>, gateways: Vec<GatewayBond>) {
+    // TODO: check if all nodes can be compared together,
+    // i.e. they all have the same denom for bonds and delegations
+    fn verify_mixnodes(&self, mixnodes: &[MixNodeBond]) -> bool {
+        if mixnodes.is_empty() {
+            return true;
+        }
+        let expected_denom = &mixnodes[0].bond_amount.denom;
+        for mixnode in mixnodes {
+            if &mixnode.bond_amount.denom != expected_denom
+                || &mixnode.total_delegation.denom != expected_denom
+            {
+                return false;
+            }
+        }
+
+        true
+    }
+
+    // TODO: check if all nodes can be compared together,
+    // i.e. they all have the same denom for bonds and delegations
+    fn verify_gateways(&self, gateways: &[GatewayBond]) -> bool {
+        if gateways.is_empty() {
+            return true;
+        }
+        let expected_denom = &gateways[0].bond_amount.denom;
+        for gateway in gateways {
+            if &gateway.bond_amount.denom != expected_denom
+                || &gateway.total_delegation.denom != expected_denom
+            {
+                return false;
+            }
+        }
+
+        false
+    }
+
+    async fn update_cache(
+        &self,
+        mut mixnodes: Vec<MixNodeBond>,
+        mut gateways: Vec<GatewayBond>,
+        state: StateParams,
+    ) {
+        // if our data is valid, it means the active sets are available,
+        // otherwise we must explicitly indicate nobody can read this data
+
+        if self.verify_mixnodes(&mixnodes) {
+            // partial_cmp can only fail if the nodes have different denomination,
+            // but we just checked for that hence the unwraps are fine here
+            // Note the reverse order of comparison so that the "highest" node would be first
+            mixnodes.sort_by(|a, b| b.partial_cmp(a).unwrap());
+            self.inner
+                .active_mixnodes_available
+                .store(true, Ordering::SeqCst);
+            self.inner
+                .current_mixnode_active_set_size
+                .store(state.mixnode_active_set_size as usize, Ordering::SeqCst);
+        } else {
+            self.inner
+                .active_mixnodes_available
+                .store(false, Ordering::SeqCst);
+        }
+
+        if self.verify_gateways(&gateways) {
+            // partial_cmp can only fail if the nodes have different denomination,
+            // but we just checked for that hence the unwraps are fine here
+            // Note the reverse order of comparison so that the "highest" node would be first
+            gateways.sort_by(|a, b| b.partial_cmp(a).unwrap());
+            self.inner
+                .active_gateways_available
+                .store(true, Ordering::SeqCst);
+            self.inner
+                .current_gateway_active_set_size
+                .store(state.gateway_active_set_size as usize, Ordering::SeqCst);
+        } else {
+            self.inner
+                .active_gateways_available
+                .store(false, Ordering::SeqCst);
+        }
+
         self.inner.mixnodes.write().await.set(mixnodes);
         self.inner.gateways.write().await.set(gateways);
     }
@@ -135,6 +228,54 @@ impl ValidatorCache {
         self.inner.gateways.read().await.clone()
     }
 
+    pub async fn active_mixnodes(&self) -> Option<Cache<Vec<MixNodeBond>>> {
+        // if active set is available, it means it is already sorted
+        if self.inner.active_mixnodes_available.load(Ordering::SeqCst) {
+            let cache = self.inner.mixnodes.read().await;
+            let timestamp = cache.as_at;
+            let nodes = cache
+                .value
+                .iter()
+                .take(
+                    self.inner
+                        .current_mixnode_active_set_size
+                        .load(Ordering::SeqCst),
+                )
+                .cloned()
+                .collect();
+            Some(Cache {
+                value: nodes,
+                as_at: timestamp,
+            })
+        } else {
+            None
+        }
+    }
+
+    pub async fn active_gateways(&self) -> Option<Cache<Vec<GatewayBond>>> {
+        // if active set is available, it means it is already sorted
+        if self.inner.active_gateways_available.load(Ordering::SeqCst) {
+            let cache = self.inner.gateways.read().await;
+            let timestamp = cache.as_at;
+            let nodes = cache
+                .value
+                .iter()
+                .take(
+                    self.inner
+                        .current_gateway_active_set_size
+                        .load(Ordering::SeqCst),
+                )
+                .cloned()
+                .collect();
+            Some(Cache {
+                value: nodes,
+                as_at: timestamp,
+            })
+        } else {
+            None
+        }
+    }
+
     pub fn initialised(&self) -> bool {
         self.inner.initialised.load(Ordering::Relaxed)
     }
@@ -158,6 +299,10 @@ impl ValidatorCacheInner {
             initialised: AtomicBool::new(false),
             mixnodes: RwLock::new(Cache::default()),
             gateways: RwLock::new(Cache::default()),
+            active_mixnodes_available: AtomicBool::new(false),
+            active_gateways_available: AtomicBool::new(false),
+            current_mixnode_active_set_size: Default::default(),
+            current_gateway_active_set_size: Default::default(),
         }
     }
 }

validator-api/src/cache/routes.rs

@@ -15,3 +15,17 @@ pub(crate) async fn get_mixnodes(cache: &State<ValidatorCache>) -> Json<Vec<MixN
 pub(crate) async fn get_gateways(cache: &State<ValidatorCache>) -> Json<Vec<GatewayBond>> {
     Json(cache.gateways().await.value)
 }
+
+#[get("/mixnodes/active")]
+pub(crate) async fn get_active_mixnodes(
+    cache: &State<ValidatorCache>,
+) -> Option<Json<Vec<MixNodeBond>>> {
+    cache.active_mixnodes().await.map(|cache| Json(cache.value))
+}
+
+#[get("/gateways/active")]
+pub(crate) async fn get_active_gateways(
+    cache: &State<ValidatorCache>,
+) -> Option<Json<Vec<GatewayBond>>> {
+    cache.active_gateways().await.map(|cache| Json(cache.value))
+}