feat(json_rpc): improve queryPowers: distinct, limit, offset, order_by.

Author: guidiaz

data_structures/src/capabilities.rs

@@ -1,14 +1,16 @@
 use serde::{Deserialize, Serialize};
-use strum_macros::{EnumString, IntoStaticStr};
+use strum_macros::{EnumIter, EnumString, IntoStaticStr};
 
 #[repr(u8)]
 #[derive(
     Clone,
     Copy,
     Debug,
     Default,
+    EnumIter,
     EnumString,
     Eq,
+    Hash,
     IntoStaticStr,
     PartialEq,
     serde::Deserialize,

data_structures/src/transaction_factory.rs

@@ -96,8 +96,8 @@ pub enum NodeBalance2 {
 impl NodeBalance2 {
     pub fn add_utxo_value(&mut self, utxo_value: u64, utxo_locked: bool) {
         if let NodeBalance2::One {
-                unlocked, locked, ..
-            } = self 
+            unlocked, locked, ..
+        } = self
         {
             if utxo_locked {
                 locked.add_assign(utxo_value);

node/src/actors/chain_manager/handlers.rs

@@ -20,7 +20,11 @@ use witnet_data_structures::{
     get_protocol_version,
     proto::versioning::ProtocolVersion,
     refresh_protocol_version,
-    staking::{errors::StakesError, prelude::StakeKey, stakes::QueryStakesKey},
+    staking::{
+        errors::StakesError,
+        prelude::{Power, StakeKey},
+        stakes::QueryStakesKey,
+    },
     transaction::{
         DRTransaction, StakeTransaction, Transaction, UnstakeTransaction, VTTransaction,
     },
@@ -46,7 +50,7 @@ use crate::{
             GetMemoryTransaction, GetMempool, GetMempoolResult, GetNodeStats, GetProtocolInfo,
             GetReputation, GetReputationResult, GetSignalingInfo, GetState, GetSuperBlockVotes,
             GetSupplyInfo, GetSupplyInfo2, GetUtxoInfo, IsConfirmedBlock, PeersBeacons,
-            QueryStakePowers, QueryStakes, QueryStakesOrderByOptions, ReputationStats, Rewind,
+            QueryStakes, QueryStakesOrderByOptions, QueryStakingPowers, ReputationStats, Rewind,
             SendLastBeacon, SessionUnitResult, SetLastBeacon, SetPeersLimits, SignalingInfo,
             SnapshotExport, SnapshotImport, TryMineBlock,
         },
@@ -1652,19 +1656,38 @@ impl Handler<QueryStakes> for ChainManager {
     }
 }
 
-impl Handler<QueryStakePowers> for ChainManager {
-    type Result = <QueryStakePowers as Message>::Result;
+impl Handler<QueryStakingPowers> for ChainManager {
+    type Result = <QueryStakingPowers as Message>::Result;
 
-    fn handle(&mut self, msg: QueryStakePowers, _ctx: &mut Self::Context) -> Self::Result {
-        let mut powers = HashMap::new();
+    fn handle(&mut self, msg: QueryStakingPowers, _ctx: &mut Self::Context) -> Self::Result {
         let current_epoch = self.current_epoch.unwrap();
-        for capability in msg.capabilities {
-            for (key, power) in self.chain_state.stakes.by_rank(capability, current_epoch) {
-                powers.entry(key).or_insert(vec![]).push(power);
-            }
-        }
+        let limit = msg.limit.unwrap_or(u16::MAX) as usize;
+        let offset = msg.offset.unwrap_or_default();
+
+        // Enumerate power entries order by specified capability,
+        // mapping into a record containing: the ranking, the stake key and current power
+        let powers = self
+            .chain_state
+            .stakes
+            .by_rank(msg.order_by, current_epoch)
+            .enumerate()
+            .map(|(index, record)| (index, record.0, record.1));
+
+        // Skip first `offset` entries and take next `limit`.
+        // If distinct is specified, retain just first appearance per validator.
+        let powers: Vec<(usize, StakeKey<PublicKeyHash>, Power)> = if msg.distinct.unwrap_or(false)
+        {
+            powers
+                .unique_by(|(_, key, _)| key.validator)
+                .skip(offset)
+                .take(limit)
+                .into_iter()
+                .collect()
+        } else {
+            powers.skip(offset).take(limit).into_iter().collect()
+        };
 
-        powers.into_iter().collect()
+        powers
     }
 }
 

node/src/actors/json_rpc/api.rs

@@ -19,7 +19,6 @@ use itertools::Itertools;
 use jsonrpc_core::{BoxFuture, Error, Params, Value};
 use jsonrpc_pubsub::{Subscriber, SubscriptionId};
 use serde::{Deserialize, Serialize};
-use strum_macros::IntoStaticStr;
 
 use witnet_crypto::key::KeyPath;
 use witnet_data_structures::{
@@ -49,7 +48,7 @@ use crate::{
             GetItemTransaction, GetKnownPeers, GetMemoryTransaction, GetMempool, GetNodeStats,
             GetProtocolInfo, GetReputation, GetSignalingInfo, GetState, GetSupplyInfo,
             GetSupplyInfo2, GetUtxoInfo, InitializePeers, IsConfirmedBlock, MagicEither,
-            QueryStakePowers, QueryStakes, Rewind, SnapshotExport, SnapshotImport,
+            QueryStakes, QueryStakingPowers, Rewind, SnapshotExport, SnapshotImport,
             StakeAuthorization,
         },
         peers_manager::PeersManager,
@@ -2267,44 +2266,33 @@ pub async fn query_stakes(params: Result<Option<QueryStakes>, Error>) -> JsonRpc
         .await
 }
 
-/// Param for query_stakes
-#[derive(Clone, Debug, Deserialize, Eq, PartialEq, IntoStaticStr, Serialize)]
-#[serde(rename_all = "lowercase")]
-pub enum QueryPowersParams {
-    /// To query all validators' power for a specific capability
-    Capability(Capability),
-    /// To query all powers
-    All(bool),
-}
-
 /// Format of the output of query_powers
 #[derive(Clone, Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
-pub struct QueryPowersRecord {
-    /// Current power
-    pub powers: Vec<u64>,
+pub struct QueryStakingPowersRecord {
+    /// Staking power
+    pub power: u64,
+    /// Ranking index (1 .. N)
+    pub ranking: usize,
     /// Validator's stringified pkh
     pub validator: String,
     /// Withdrawer's stringified pkh
     pub withdrawer: String,
 }
 
 /// Query the amount of nanowits staked by an address.
-pub async fn query_powers(params: Result<QueryPowersParams, Error>) -> JsonRpcResult {
+pub async fn query_powers(params: Result<QueryStakingPowers, Error>) -> JsonRpcResult {
     // Short-circuit if parameters are wrong
-    let params = params?;
+    let msg = params?;
 
-    let capabilities = match params {
-        QueryPowersParams::All(_) => ALL_CAPABILITIES.to_vec(),
-        QueryPowersParams::Capability(c) => vec![c],
-    };
     ChainManager::from_registry()
-        .send(QueryStakePowers { capabilities })
+        .send(msg)
         .map(|res| match res {
             Ok(candidates) => {
-                let candidates: Vec<QueryPowersRecord> = candidates
+                let candidates: Vec<QueryStakingPowersRecord> = candidates
                     .iter()
-                    .map(|(key, powers)| QueryPowersRecord {
-                        powers: powers.clone(),
+                    .map(|(ranking, key, power)| QueryStakingPowersRecord {
+                        power: *power,
+                        ranking: *ranking,
                         validator: key.validator.to_string(),
                         withdrawer: key.withdrawer.to_string(),
                     })

node/src/actors/messages.rs

@@ -346,14 +346,32 @@ impl Message for StakeAuthorization {
 }
 
 /// Message for querying stakes
-#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
-pub struct QueryStakePowers {
-    /// capabilities being searched for
-    pub capabilities: Vec<Capability>,
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct QueryStakingPowers {
+    /// Retrieve only first appearence for every distinct validator (default: false)
+    pub distinct: Option<bool>,
+    /// Limits max number of entries to return (default: 0 == u16::MAX)
+    pub limit: Option<u16>,
+    /// Skips first found entries (default: 0)
+    pub offset: Option<usize>,
+    /// Order by specified capability (default: reverse order by coins)
+    pub order_by: Capability,
+}
+
+impl Default for QueryStakingPowers {
+    fn default() -> Self {
+        QueryStakingPowers {
+            distinct: Some(false),
+            limit: Some(u16::MAX),
+            offset: Some(0),
+            order_by: Capability::Mining,
+        }
+    }
 }
 
-impl Message for QueryStakePowers {
-    type Result = Vec<(StakeKey<PublicKeyHash>, Vec<Power>)>;
+impl Message for QueryStakingPowers {
+    type Result = Vec<(usize, StakeKey<PublicKeyHash>, Power)>;
 }
 
 /// Stake key for quering stakes

src/cli/node/json_rpc_client.rs

@@ -14,7 +14,7 @@ use ansi_term::Color::{Purple, Red, White, Yellow};
 use failure::{bail, Fail};
 use itertools::Itertools;
 use num_format::{Locale, ToFormattedString};
-use prettytable::{row, Cell, Row, Table};
+use prettytable::{row, Table};
 use qrcode::render::unicode;
 use serde::{de::DeserializeOwned, Deserialize, Serialize};
 
@@ -24,7 +24,7 @@ use witnet_crypto::{
     key::{ExtendedPK, ExtendedSK},
 };
 use witnet_data_structures::{
-    capabilities::{Capability, ALL_CAPABILITIES},
+    capabilities::Capability,
     chain::{
         priority::{PrioritiesEstimate, Priority, PriorityEstimate, TimeToBlock},
         tapi::{current_active_wips, ActiveWips},
@@ -50,13 +50,12 @@ use witnet_data_structures::{
 use witnet_node::actors::{
     chain_manager::run_dr_locally,
     json_rpc::api::{
-        AddrType, GetBlockChainParams, GetTransactionOutput, PeersResult, QueryPowersParams,
-        QueryPowersRecord,
+        AddrType, GetBlockChainParams, GetTransactionOutput, PeersResult, QueryStakingPowersRecord,
     },
     messages::{
         AuthorizeStake, BuildDrt, BuildStakeParams, BuildStakeResponse, BuildUnstakeParams,
         BuildVtt, GetBalanceTarget, GetReputationResult, MagicEither, QueryStakes,
-        QueryStakesFilter, SignalingInfo, StakeAuthorization,
+        QueryStakesFilter, QueryStakingPowers, SignalingInfo, StakeAuthorization,
     },
 };
 use witnet_rad::types::RadonTypes;
@@ -2019,19 +2018,18 @@ pub fn query_stakes(
 pub fn query_powers(
     addr: SocketAddr,
     capability: Option<String>,
-    all: bool,
+    distinct: bool,
 ) -> Result<(), failure::Error> {
     let mut stream = start_client(addr)?;
-    let params = if all {
-        QueryPowersParams::All(true)
-    } else {
-        match capability {
-            Some(c) => match Capability::from_str(&c) {
-                Ok(c) => QueryPowersParams::Capability(c),
-                Err(_) => QueryPowersParams::Capability(Capability::Mining),
-            },
-            None => QueryPowersParams::Capability(Capability::Mining),
-        }
+
+    let mut params = QueryStakingPowers::default();
+    params.distinct = Some(distinct);
+    params.order_by = match capability {
+        Some(c) => match Capability::from_str(&c) {
+            Ok(c) => c,
+            Err(_) => Capability::Mining,
+        },
+        None => Capability::Mining,
     };
 
     let response = send_request(
@@ -2042,42 +2040,22 @@ pub fn query_powers(
         ),
     )?;
 
-    let mut powers: Vec<QueryPowersRecord> = parse_response(&response)?;
-    powers.sort_by_key(|power| Reverse(power.powers[0]));
+    let records: Vec<QueryStakingPowersRecord> = parse_response(&response)?;
 
     let mut powers_table = Table::new();
     powers_table.set_format(*prettytable::format::consts::FORMAT_NO_BORDER_LINE_SEPARATOR);
-    if all {
-        let mut header = vec![
-            Cell::new("Validator").style_spec("c"),
-            Cell::new("Withdrawer").style_spec("c"),
-        ];
-        for capability in ALL_CAPABILITIES {
-            let capability_str: &'static str = capability.into();
-            header.push(Cell::new(capability_str).style_spec("c"));
-        }
-        powers_table.set_titles(Row::new(header));
-        for power in powers.iter() {
-            let mut row = vec![
-                Cell::new(&power.validator.to_string()),
-                Cell::new(&power.withdrawer.to_string()),
-            ];
-            for p in &power.powers {
-                row.push(Cell::new(&p.to_formatted_string(&Locale::en)).style_spec("r"));
-            }
-            powers_table.add_row(Row::new(row));
-        }
-    } else {
-        let capability_str: &'static str = params.into();
-        powers_table.set_titles(row![c->"Validator", c->"Withdrawer", c->capability_str]);
-        for power in powers.iter() {
-            powers_table.add_row(row![
-                power.validator,
-                power.withdrawer,
-                r->(power.powers[0]).to_formatted_string(&Locale::en),
-            ]);
-        }
+
+    let capability_str: &'static str = params.order_by.into();
+    powers_table.set_titles(row![c->"Ranking", c->"Validator", c->"Withdrawer", c->capability_str]);
+    for record in records.iter() {
+        powers_table.add_row(row![
+            record.ranking,
+            record.validator,
+            record.withdrawer,
+            record.power.to_formatted_string(&Locale::en),
+        ]);
     }
+
     powers_table.printstd();
     println!();
 

src/cli/node/with_node.rs

@@ -300,8 +300,8 @@ pub fn exec_cmd(
         Command::QueryPowers {
             node,
             capability,
-            all,
-        } => rpc::query_powers(node.unwrap_or(default_jsonrpc), capability, all),
+            distinct,
+        } => rpc::query_powers(node.unwrap_or(default_jsonrpc), capability, distinct),
         Command::Unstake {
             node,
             operator,
@@ -848,8 +848,8 @@ pub enum Command {
         node: Option<SocketAddr>,
         #[structopt(short = "c", long = "capability")]
         capability: Option<String>,
-        #[structopt(short = "a", long = "all")]
-        all: bool,
+        #[structopt(short = "d", long = "distinct")]
+        distinct: bool,
     },
     #[structopt(name = "unstake", about = "Create an unstake transaction")]
     Unstake {