block signing related code

Cargo.lock

@@ -7506,9 +7506,7 @@ dependencies = [
  "nym-task",
  "nym-validator-client",
  "nyxd-scraper",
- "reqwest",
  "serde",
- "serde_json",
  "sha2 0.10.8",
  "thiserror",
  "time",

common/client-libs/validator-client/src/client.rs

@@ -42,6 +42,14 @@ pub struct Config {
     nyxd_config: nyxd::Config,
 }
 
+impl TryFrom<NymNetworkDetails> for Config {
+    type Error = ValidatorClientError;
+
+    fn try_from(value: NymNetworkDetails) -> Result<Self, Self::Error> {
+        Config::try_from_nym_network_details(&value)
+    }
+}
+
 impl Config {
     pub fn try_from_nym_network_details(
         details: &NymNetworkDetails,

common/client-libs/validator-client/src/nyxd/cosmwasm_client/client_traits/signing_client.rs

@@ -425,7 +425,7 @@ where
                     amount: amount.into_iter().map(Into::into).collect(),
                 }
                 .to_any()
-                .map_err(|_| NyxdError::SerializationError("MsgExecuteContract".to_owned()))
+                .map_err(|_| NyxdError::SerializationError("MsgSend".to_owned()))
             })
             .collect::<Result<_, _>>()?;
 

common/client-libs/validator-client/src/nyxd/cosmwasm_client/module_traits/staking/mod.rs

@@ -1,4 +1,8 @@
 // Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-pub mod query;
+pub mod query;
+
+pub use cosmrs::staking::{
+    QueryHistoricalInfoResponse, QueryValidatorResponse, QueryValidatorsResponse, Validator,
+};

common/client-libs/validator-client/src/nyxd/cosmwasm_client/module_traits/staking/query.rs

@@ -1,6 +1,7 @@
 // Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use super::{QueryHistoricalInfoResponse, QueryValidatorResponse, QueryValidatorsResponse};
 use crate::nyxd::error::NyxdError;
 use crate::nyxd::{CosmWasmClient, PageRequest};
 use async_trait::async_trait;
@@ -12,10 +13,7 @@ use cosmrs::proto::cosmos::staking::v1beta1::{
     QueryValidatorsRequest as ProtoQueryValidatorsRequest,
     QueryValidatorsResponse as ProtoQueryValidatorsResponse,
 };
-use cosmrs::staking::{
-    QueryHistoricalInfoRequest, QueryHistoricalInfoResponse, QueryValidatorRequest,
-    QueryValidatorResponse, QueryValidatorsRequest, QueryValidatorsResponse,
-};
+use cosmrs::staking::{QueryHistoricalInfoRequest, QueryValidatorRequest, QueryValidatorsRequest};
 use cosmrs::AccountId;
 
 // TODO: change trait restriction from `CosmWasmClient` to `TendermintRpcClient`

common/client-libs/validator-client/src/nyxd/mod.rs

@@ -41,6 +41,7 @@ pub use coin::Coin;
 pub use cosmrs::{
     bank::MsgSend,
     bip32,
+    crypto::PublicKey,
     query::{PageRequest, PageResponse},
     tendermint::{
         abci::{response::DeliverTx, types::ExecTxResult, Event, EventAttribute},
@@ -50,7 +51,7 @@ pub use cosmrs::{
         Time as TendermintTime,
     },
     tx::{self, Msg},
-    AccountId, Coin as CosmosCoin, Denom, Gas,
+    AccountId, Any, Coin as CosmosCoin, Denom, Gas,
 };
 pub use cosmwasm_std::Coin as CosmWasmCoin;
 pub use cw2;
@@ -102,6 +103,14 @@ impl Config {
     }
 }
 
+impl TryFrom<NymNetworkDetails> for Config {
+    type Error = NyxdError;
+
+    fn try_from(value: NymNetworkDetails) -> Result<Self, Self::Error> {
+        Config::try_from_nym_network_details(&value)
+    }
+}
+
 #[derive(Debug)]
 pub struct NyxdClient<C, S = NoSigner> {
     client: MaybeSigningClient<C, S>,

common/nyxd-scraper/src/scraper/mod.rs

@@ -9,8 +9,10 @@ use crate::rpc_client::RpcClient;
 use crate::scraper::subscriber::{run_websocket_driver, ChainSubscriber};
 use crate::storage::ScraperStorage;
 use std::path::PathBuf;
+use std::time::Duration;
 use tendermint_rpc::WebSocketClientDriver;
 use tokio::sync::mpsc::{channel, unbounded_channel};
+use tokio::time::interval;
 use tokio_util::sync::CancellationToken;
 use tokio_util::task::TaskTracker;
 use tracing::info;
@@ -191,6 +193,8 @@ impl NyxdScraper {
 
     pub async fn stop(self) {
         info!("stopping the chain scrapper");
+        assert!(self.task_tracker.is_closed());
+
         self.cancel_token.cancel();
         self.task_tracker.wait().await
     }

nym-validator-rewarder/Cargo.toml

@@ -27,10 +27,6 @@ zeroize.workspace = true
 sha2 = "0.10.8"
 humantime-serde = "1.0"
 
-
-reqwest = { workspace = true, features = ["json"] }
-serde_json = "1.0.108"
-
 # internal
 nym-bin-common = { path = "../common/bin-common", features = ["output_format"] }
 nym-config = { path = "../common/config" }

nym-validator-rewarder/migrations/01_initial.sql

@@ -0,0 +1,4 @@
+/*
+ * Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+ * SPDX-License-Identifier: GPL-3.0-only
+ */

nym-validator-rewarder/src/error.rs

@@ -53,10 +53,30 @@ pub enum NymRewarderError {
     #[error("failed to determine epoch boundaries: {0}")]
     TimeComponentFailure(#[from] time::error::ComponentRange),
 
-    #[error("could not convert validators conesnsus address: {consensus_address} to a nym account address: {source}")]
+    #[error(
+        "could not convert operator address: {operator_address} to a nym account address: {source}"
+    )]
     MalformedBech32Address {
-        consensus_address: String,
+        operator_address: String,
         #[source]
         source: ErrorReport,
     },
+
+    #[error(
+        "could not convert validator public key: {public_key} into a consensus address: {source}"
+    )]
+    MalformedConsensusPublicKey {
+        public_key: String,
+        #[source]
+        source: ErrorReport,
+    },
+
+    #[error("somehow the total voting power was negative: {val}")]
+    NegativeTotalVotingPower { val: i64 },
+
+    #[error("somehow the signed blocks was negative: {val}")]
+    NegativeSignedBlocks { val: i64 },
+
+    #[error("could not find details for validator {consensus_address}")]
+    MissingValidatorDetails { consensus_address: String },
 }

nym-validator-rewarder/src/rewarder/block_signing.rs

@@ -1,78 +0,0 @@
-// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use cosmwasm_std::{Decimal, Uint128};
-use nym_validator_client::nyxd::Coin;
-use nyxd_scraper::models;
-use std::collections::HashMap;
-
-#[derive(Debug)]
-pub struct ValidatorSigning {
-    pub validator: models::Validator,
-
-    pub voting_power_at_epoch_start: i64,
-    pub voting_power_ratio: Decimal,
-
-    pub signed_blocks: i32,
-    pub ratio_signed: Decimal,
-}
-
-#[derive(Debug)]
-pub struct EpochSigning {
-    pub blocks: i64,
-    pub total_voting_power_at_epoch_start: i64,
-
-    pub validators: Vec<ValidatorSigning>,
-}
-
-impl EpochSigning {
-    pub fn construct(
-        blocks: i64,
-        total_vp: i64,
-        validator_results: HashMap<models::Validator, (i32, i64)>,
-    ) -> Self {
-        assert!(total_vp >= 0, "negative voting power!");
-        assert!(blocks >= 0, "negative blocks!");
-        let total_vp_u64: u64 = total_vp.try_into().unwrap_or_default();
-        let blocks_u64: u64 = blocks.try_into().unwrap_or_default();
-
-        let validators = validator_results
-            .into_iter()
-            .map(
-                |(validator, (signed_blocks, voting_power_at_epoch_start))| {
-                    let vp: u64 = voting_power_at_epoch_start.try_into().unwrap_or_default();
-                    let signed: u64 = signed_blocks.try_into().unwrap_or_default();
-
-                    let voting_power_ratio = Decimal::from_ratio(vp, total_vp_u64);
-                    let ratio_signed = Decimal::from_ratio(signed, blocks_u64);
-
-                    ValidatorSigning {
-                        validator,
-                        voting_power_at_epoch_start,
-                        voting_power_ratio,
-                        signed_blocks,
-                        ratio_signed,
-                    }
-                },
-            )
-            .collect();
-
-        EpochSigning {
-            blocks,
-            total_voting_power_at_epoch_start: total_vp,
-            validators,
-        }
-    }
-
-    pub fn rewarding_amounts(&self, budget: &Coin) -> HashMap<models::Validator, Coin> {
-        let denom = &budget.denom;
-        self.validators
-            .iter()
-            .map(|v| {
-                let amount = Uint128::new(budget.amount) * v.ratio_signed * v.voting_power_ratio;
-
-                (v.validator.clone(), Coin::new(amount.u128(), denom))
-            })
-            .collect()
-    }
-}

nym-validator-rewarder/src/rewarder/block_signing/mod.rs

@@ -0,0 +1,142 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::error::NymRewarderError;
+use crate::rewarder::block_signing::types::EpochSigningResults;
+use crate::rewarder::epoch::Epoch;
+use nym_validator_client::nyxd::module_traits::staking;
+use nym_validator_client::nyxd::{PageRequest, StakingQueryClient};
+use nym_validator_client::QueryHttpRpcNyxdClient;
+use nyxd_scraper::NyxdScraper;
+use std::cmp::min;
+use std::collections::HashMap;
+use std::ops::Range;
+use tracing::info;
+
+pub(crate) mod types;
+
+pub struct EpochSigning {
+    pub(crate) rpc_client: QueryHttpRpcNyxdClient,
+    pub(crate) nyxd_scraper: NyxdScraper,
+}
+
+impl EpochSigning {
+    async fn get_voting_power(
+        &self,
+        address: &str,
+        height_range: Range<i64>,
+    ) -> Result<Option<i64>, NymRewarderError> {
+        for height in height_range {
+            if let Some(precommit) = self
+                .nyxd_scraper
+                .storage
+                .get_precommit(address, height)
+                .await?
+            {
+                return Ok(Some(precommit.voting_power));
+            }
+        }
+
+        Ok(None)
+    }
+
+    // TODO: eventually this will be replaced by scraping the data from the staking module in the scraper itself
+    async fn get_validator_details(
+        &self,
+        height: i64,
+    ) -> Result<Vec<staking::Validator>, NymRewarderError> {
+        // first attempt to get it via the historical info.
+        // if that fails, attempt to use current block information to at least get **something**
+        if let Some(validators) = self.rpc_client.historical_info(height).await?.hist {
+            Ok(validators.valset)
+        } else {
+            let mut page_request = None;
+            let mut response = Vec::new();
+
+            loop {
+                let mut res = self
+                    .rpc_client
+                    .validators("".to_string(), page_request)
+                    .await?;
+                response.append(&mut res.validators);
+
+                let Some(pagination) = res.pagination else {
+                    break;
+                };
+
+                page_request = Some(PageRequest {
+                    key: pagination.next_key,
+                    offset: 0,
+                    limit: 0,
+                    count_total: false,
+                    reverse: false,
+                });
+            }
+
+            Ok(response)
+        }
+    }
+
+    pub(crate) async fn get_signed_blocks_results(
+        &self,
+        current_epoch: Epoch,
+    ) -> Result<EpochSigningResults, NymRewarderError> {
+        info!(
+            "looking up block signers for epoch {} ({} - {})",
+            current_epoch.id,
+            current_epoch.start_rfc3339(),
+            current_epoch.end_rfc3339()
+        );
+
+        let validators = self.nyxd_scraper.storage.get_all_known_validators().await?;
+        let epoch_start = current_epoch.start;
+        let epoch_end = current_epoch.end;
+        let first_block = self
+            .nyxd_scraper
+            .storage
+            .get_first_block_height_after(epoch_start)
+            .await?
+            .unwrap_or_default();
+        let last_block = self
+            .nyxd_scraper
+            .storage
+            .get_last_block_height_before(epoch_end)
+            .await?
+            .unwrap_or_default();
+
+        // each validator MUST be online at some point during the first 20 blocks, otherwise they're not getting anything.
+        let vp_range_end = min(first_block + 20, last_block);
+        let vp_range = first_block..vp_range_end;
+
+        let mut total_vp = 0;
+        let mut signed_in_epoch = HashMap::new();
+
+        // for each validator, with a valid voting power, get number of signed blocks in the rewarding epoch
+        for validator in validators {
+            let Some(vp) = self
+                .get_voting_power(&validator.consensus_address, vp_range.clone())
+                .await?
+            else {
+                continue;
+            };
+            total_vp += vp;
+
+            let signed = self
+                .nyxd_scraper
+                .storage
+                .get_signed_between_times(&validator.consensus_address, epoch_start, epoch_end)
+                .await?;
+            signed_in_epoch.insert(validator, (signed, vp));
+        }
+
+        let total = self
+            .nyxd_scraper
+            .storage
+            .get_blocks_between(epoch_start, epoch_end)
+            .await?;
+
+        let details = self.get_validator_details(last_block).await?;
+
+        EpochSigningResults::construct(total, total_vp, signed_in_epoch, details)
+    }
+}

nym-validator-rewarder/src/rewarder/block_signing/types.rs

@@ -0,0 +1,121 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::error::NymRewarderError;
+use crate::rewarder::helpers::{consensus_pubkey_to_address, operator_account_to_owner_account};
+use cosmwasm_std::{Decimal, Uint128};
+use nym_validator_client::nyxd::module_traits::staking;
+use nym_validator_client::nyxd::{AccountId, Coin};
+use nyxd_scraper::models;
+use std::collections::HashMap;
+use tracing::info;
+
+#[derive(Debug)]
+pub struct ValidatorSigning {
+    pub validator: models::Validator,
+    pub staking_details: staking::Validator,
+    pub operator_account: AccountId,
+
+    pub voting_power_at_epoch_start: i64,
+    pub voting_power_ratio: Decimal,
+
+    pub signed_blocks: i32,
+    pub ratio_signed: Decimal,
+}
+
+impl ValidatorSigning {
+    pub fn reward_amount(&self, signing_budget: &Coin) -> Coin {
+        let amount =
+            Uint128::new(signing_budget.amount) * self.ratio_signed * self.voting_power_ratio;
+
+        let amount = Coin::new(amount.u128(), &signing_budget.denom);
+        let moniker = self
+            .staking_details
+            .description
+            .as_ref()
+            .map(|d| d.moniker.clone())
+            .unwrap_or("UNKNOWN MONIKER".to_string());
+
+        info!(
+            "validator {moniker} will receive {amount} at address {} for block signing work",
+            self.operator_account
+        );
+        amount
+    }
+}
+
+#[derive(Debug)]
+pub struct EpochSigningResults {
+    pub blocks: i64,
+    pub total_voting_power_at_epoch_start: i64,
+
+    pub validators: Vec<ValidatorSigning>,
+}
+
+impl EpochSigningResults {
+    pub fn construct(
+        blocks: i64,
+        total_vp: i64,
+        validator_results: HashMap<models::Validator, (i32, i64)>,
+        validator_details: Vec<staking::Validator>,
+    ) -> Result<Self, NymRewarderError> {
+        let Ok(total_vp_u64): Result<u64, _> = total_vp.try_into() else {
+            return Err(NymRewarderError::NegativeTotalVotingPower { val: total_vp });
+        };
+        let Ok(blocks_u64): Result<u64, _> = blocks.try_into() else {
+            return Err(NymRewarderError::NegativeSignedBlocks { val: blocks });
+        };
+
+        let mut validator_details: HashMap<_, _> = validator_details
+            .into_iter()
+            .filter(|v| v.consensus_pubkey.is_some())
+            .map(|v| {
+                // safety: we know the key is definitely set as we just filtered the iterator based on that condition
+                #[allow(clippy::unwrap_used)]
+                consensus_pubkey_to_address(v.consensus_pubkey.unwrap())
+                    .map(|addr| (addr.to_string(), v))
+            })
+            .collect::<Result<_, _>>()?;
+
+        let mut validators = Vec::new();
+
+        for (validator, (signed_blocks, voting_power_at_epoch_start)) in validator_results {
+            let vp: u64 = voting_power_at_epoch_start.try_into().unwrap_or_default();
+            let signed: u64 = signed_blocks.try_into().unwrap_or_default();
+
+            let voting_power_ratio = Decimal::from_ratio(vp, total_vp_u64);
+            let ratio_signed = Decimal::from_ratio(signed, blocks_u64);
+            let staking_details = validator_details
+                .remove(&validator.consensus_address)
+                .ok_or_else(|| NymRewarderError::MissingValidatorDetails {
+                    consensus_address: validator.consensus_address.clone(),
+                })?;
+
+            let operator_account =
+                operator_account_to_owner_account(&staking_details.operator_address)?;
+
+            validators.push(ValidatorSigning {
+                validator,
+                staking_details,
+                operator_account,
+                voting_power_at_epoch_start,
+                voting_power_ratio,
+                signed_blocks,
+                ratio_signed,
+            })
+        }
+
+        Ok(EpochSigningResults {
+            blocks,
+            total_voting_power_at_epoch_start: total_vp,
+            validators,
+        })
+    }
+
+    pub fn rewarding_amounts(&self, budget: &Coin) -> Vec<(AccountId, Vec<Coin>)> {
+        self.validators
+            .iter()
+            .map(|v| (v.operator_account.clone(), vec![v.reward_amount(budget)]))
+            .collect()
+    }
+}

nym-validator-rewarder/src/rewarder/epoch.rs

@@ -1,5 +1,5 @@
 // Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
+// SPDX-License-Identifier: GPL-3.0-only
 
 use crate::error::NymRewarderError;
 use std::ops::Add;
@@ -9,6 +9,7 @@ use time::OffsetDateTime;
 
 const HOUR: Duration = Duration::from_secs(60 * 60);
 
+#[derive(Debug, Clone, Copy)]
 pub struct Epoch {
     pub id: i64,
 

nym-validator-rewarder/src/rewarder/helpers.rs

@@ -1,23 +1,32 @@
 // Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
+// SPDX-License-Identifier: GPL-3.0-only
 
 use crate::error::NymRewarderError;
-use nym_validator_client::nyxd::AccountId;
-use nyxd_scraper::constants::BECH32_PREFIX;
+use nym_validator_client::nyxd::{AccountId, PublicKey};
+use nyxd_scraper::constants::{BECH32_CONSENSUS_ADDRESS_PREFIX, BECH32_PREFIX};
+use sha2::{Digest, Sha256};
 
-pub(crate) fn consensus_address_to_account(
-    consensus_address: &str,
+pub(crate) fn consensus_pubkey_to_address(
+    pubkey: PublicKey,
 ) -> Result<AccountId, NymRewarderError> {
-    let consensus_addr: AccountId =
-        consensus_address
-            .parse()
-            .map_err(|source| NymRewarderError::MalformedBech32Address {
-                consensus_address: consensus_address.to_string(),
-                source,
-            })?;
-    AccountId::new(BECH32_PREFIX, &consensus_addr.to_bytes()).map_err(|source| {
-        NymRewarderError::MalformedBech32Address {
-            consensus_address: consensus_address.to_string(),
+    let digest = Sha256::digest(pubkey.to_bytes()).to_vec();
+
+    // TODO: make those configurable, etc
+    AccountId::new(BECH32_CONSENSUS_ADDRESS_PREFIX, &digest[..20]).map_err(|source| {
+        NymRewarderError::MalformedConsensusPublicKey {
+            public_key: pubkey.to_string(),
+            source,
+        }
+    })
+}
+
+// it's just a matter of swapping bech32 prefixes and recalculating the checksum
+pub(crate) fn operator_account_to_owner_account(
+    operator_address: &AccountId,
+) -> Result<AccountId, NymRewarderError> {
+    AccountId::new(BECH32_PREFIX, &operator_address.to_bytes()).map_err(|source| {
+        NymRewarderError::MalformedBech32Address {
+            operator_address: operator_address.to_string(),
             source,
         }
     })

nym-validator-rewarder/src/rewarder/mod.rs

@@ -5,34 +5,28 @@ use crate::config::Config;
 use crate::error::NymRewarderError;
 use crate::rewarder::block_signing::EpochSigning;
 use crate::rewarder::epoch::Epoch;
-use crate::rewarder::helpers::consensus_address_to_account;
-use nym_network_defaults::NymNetworkDetails;
 use nym_task::TaskManager;
-use nym_validator_client::nyxd::{AccountId, Coin, StakingQueryClient};
+use nym_validator_client::nyxd::{AccountId, Coin};
 use nym_validator_client::QueryHttpRpcNyxdClient;
-use nyxd_scraper::{models, NyxdScraper};
-use sha2::{Digest, Sha256};
-use std::cmp::min;
-use std::collections::HashMap;
-use std::ops::{Add, Range};
-use std::str::FromStr;
+use nyxd_scraper::NyxdScraper;
+use std::ops::Add;
 use std::time::Duration;
-use time::format_description::well_known::Rfc3339;
 use time::OffsetDateTime;
+use tokio::pin;
 use tokio::time::{interval_at, Instant};
 use tracing::{debug, error, info, instrument};
 
 mod block_signing;
 mod epoch;
 mod helpers;
+mod nyxd_client;
 mod tasks;
 
 pub struct Rewarder {
-    current_epoch: Epoch,
-    rpc_client: QueryHttpRpcNyxdClient,
-
     config: Config,
-    nyxd_scraper: NyxdScraper,
+    current_epoch: Epoch,
+
+    epoch_signing: EpochSigning,
 }
 
 impl Rewarder {
@@ -45,95 +39,24 @@ impl Rewarder {
 
         Ok(Rewarder {
             current_epoch: Epoch::first()?,
-            rpc_client,
             config,
-            nyxd_scraper,
+            epoch_signing: EpochSigning {
+                nyxd_scraper,
+                rpc_client,
+            },
         })
     }
 
-    async fn get_voting_power(
-        &self,
-        address: &str,
-        height_range: Range<i64>,
-    ) -> Result<Option<i64>, NymRewarderError> {
-        for height in height_range {
-            if let Some(precommit) = self
-                .nyxd_scraper
-                .storage
-                .get_precommit(address, height)
-                .await?
-            {
-                return Ok(Some(precommit.voting_power));
-            }
-        }
-
-        Ok(None)
-    }
-
-    async fn get_signed_blocks(&self) -> Result<EpochSigning, NymRewarderError> {
-        info!(
-            "looking up block signers for epoch {} ({} - {})",
-            self.current_epoch.id,
-            self.current_epoch.start_rfc3339(),
-            self.current_epoch.end_rfc3339()
-        );
-
-        let validators = self.nyxd_scraper.storage.get_all_known_validators().await?;
-        let epoch_start = self.current_epoch.start;
-        let epoch_end = self.current_epoch.end;
-        let first_block = self
-            .nyxd_scraper
-            .storage
-            .get_first_block_height_after(epoch_start)
-            .await?
-            .unwrap_or_default();
-        let last_block = self
-            .nyxd_scraper
-            .storage
-            .get_last_block_height_before(epoch_end)
-            .await?
-            .unwrap_or_default();
-
-        // each validator MUST be online at some point during the first 20 blocks, otherwise they're not getting anything.
-        let vp_range_end = min(first_block + 20, last_block);
-        let vp_range = first_block..vp_range_end;
-
-        let mut total_vp = 0;
-        let mut signed_in_epoch = HashMap::new();
-        for validator in validators {
-            let Some(vp) = self
-                .get_voting_power(&validator.consensus_address, vp_range.clone())
-                .await?
-            else {
-                continue;
-            };
-            total_vp += vp;
-
-            let signed = self
-                .nyxd_scraper
-                .storage
-                .get_signed_between_times(&validator.consensus_address, epoch_start, epoch_end)
-                .await?;
-            signed_in_epoch.insert(validator, (signed, vp));
-        }
-
-        let total = self
-            .nyxd_scraper
-            .storage
-            .get_blocks_between(epoch_start, epoch_end)
-            .await?;
-
-        Ok(EpochSigning::construct(total, total_vp, signed_in_epoch))
-    }
-
     #[instrument(skip(self,budget), fields(budget = %budget))]
-
     async fn calculate_block_signing_rewards(
         &mut self,
         budget: Coin,
-    ) -> Result<HashMap<models::Validator, Coin>, NymRewarderError> {
+    ) -> Result<Vec<(AccountId, Vec<Coin>)>, NymRewarderError> {
         info!("calculating reward shares");
-        let signed = self.get_signed_blocks().await?;
+        let signed = self
+            .epoch_signing
+            .get_signed_blocks_results(self.current_epoch)
+            .await?;
 
         debug!("details: {signed:?}");
 
@@ -144,14 +67,14 @@ impl Rewarder {
     async fn calculate_credential_rewards(
         &mut self,
         budget: Coin,
-    ) -> Result<HashMap<models::Validator, Coin>, NymRewarderError> {
+    ) -> Result<Vec<(AccountId, Vec<Coin>)>, NymRewarderError> {
         info!("calculating reward shares");
-        Ok(HashMap::new())
+        Ok(Vec::new())
     }
 
     async fn determine_epoch_rewards(
         &mut self,
-    ) -> Result<HashMap<models::Validator, (AccountId, Coin)>, NymRewarderError> {
+    ) -> Result<Vec<(AccountId, Vec<Coin>)>, NymRewarderError> {
         let epoch_budget = &self.config.rewarding.epoch_budget;
         let denom = &epoch_budget.denom;
         let signing_budget = Coin::new(
@@ -164,65 +87,16 @@ impl Rewarder {
         );
 
         let signing_rewards = self.calculate_block_signing_rewards(signing_budget).await?;
-        let credential_rewards = self.calculate_credential_rewards(credential_budget).await?;
-
-        let mut rewards: HashMap<models::Validator, (AccountId, Coin)> = HashMap::new();
-        let validators2 = self
-            .rpc_client
-            .validators("".to_string(), None)
-            .await
-            .unwrap();
-
-        let mut monikers = HashMap::new();
-        for v in validators2.validators {
-            let val = v.consensus_pubkey.unwrap();
-            // println!("{:?}", val);
-
-            let digest = Sha256::digest(&val.to_bytes()).to_vec();
-
-            // println!("{}", String::from_utf8_lossy(&val));
-            // assert_eq!(val.len(), 32);
-
-            let consensus_key = AccountId::new("nvalcons", &digest[..20])
-                .unwrap()
-                .to_string();
-            let moniker = v.description.unwrap().moniker;
-            let acc = v.operator_address;
-            let acc = AccountId::new("n", &acc.to_bytes()).unwrap();
-            monikers.insert(consensus_key, (moniker, acc));
-        }
-
-        for (val, amount) in &signing_rewards {
-            // let oper = AccountId::new("nvaloper", &acc.to_bytes()).unwrap();
-            // let moniker = get_moniker(acc.clone()).await;
-
-            let (moniker, acc) = monikers.get(&val.consensus_address).unwrap();
-            println!("{moniker}: {acc} signing: {amount} credentials: XXX")
-            //
-        }
-
-        // for (validator, amount) in signing_rewards {
-        //     let account = consensus_address_to_account(&validator)?;
-        //     rewards.insert(validator, (account, amount));
-        // }
-        //
-        // for (validator, amount) in credential_rewards {
-        //     let account = consensus_address_to_account(&validator)?;
-        //
-        //     if let Some(existing) = rewards.get_mut(&validator) {
-        //         assert_eq!(existing.0, account);
-        //         existing.1.amount += amount.amount;
-        //     } else {
-        //         rewards.insert(validator, (account, amount));
-        //     }
-        // }
+        let mut credential_rewards = self.calculate_credential_rewards(credential_budget).await?;
 
+        let mut rewards = signing_rewards;
+        rewards.append(&mut credential_rewards);
         Ok(rewards)
     }
 
     async fn send_rewards(
         &self,
-        amounts: HashMap<AccountId, Coin>,
+        amounts: Vec<(AccountId, Vec<Coin>)>,
     ) -> Result<(), NymRewarderError> {
         Ok(())
     }
@@ -238,30 +112,11 @@ impl Rewarder {
             }
         };
 
-        // for (_, (acc, amount)) in rewards {
-        //     // let oper = AccountId::new("nvaloper", &acc.to_bytes()).unwrap();
-        //     // let moniker = get_moniker(acc.clone()).await;
-        //
-        //     let moniker = validators2.validators.iter().find(|v| {
-        //         v.consensus_pubkey
-        //
-        //     })
-        //
-        //     println!("{moniker}: {acc} signing: {amount} credentials: XXX")
-        //     //
-        // }
+        if let Err(err) = self.send_rewards(rewards).await {
+            error!("failed to send epoch rewards: {err}");
+            return;
+        };
 
-        /*
-
-        let budget = Coin::new(667_000_000, "unym");
-        let rewards = foo.rewarding_amounts(&budget);
-
-        println!("{rewards:#?}");
-        666_378_383
-        let bar: u128 = rewards.into_values().map(|v| v.amount).sum();
-        println!("summed: {bar}");
-
-         */
         self.current_epoch = self.current_epoch.next();
     }
 
@@ -271,17 +126,12 @@ impl Rewarder {
         // setup shutdowns
         let mut task_manager = TaskManager::new(5);
 
-        self.nyxd_scraper.start().await?;
-        //
-        // tokio::time::sleep(Duration::from_secs(3000)).await;
+        self.epoch_signing.nyxd_scraper.start().await?;
 
         // rewarding epochs last from :00 to :00
         self.current_epoch.end = OffsetDateTime::now_utc();
         self.current_epoch.start = self.current_epoch.end - Duration::from_secs(60 * 60);
 
-        println!("sleepiing for 10");
-        tokio::time::sleep(Duration::from_secs(10)).await;
-
         let until_end = self.current_epoch.until_end();
 
         info!(
@@ -289,11 +139,13 @@ impl Rewarder {
             until_end.as_secs()
         );
         let mut epoch_ticker = interval_at(Instant::now().add(until_end), Epoch::LENGTH);
+        let shutdown_future = task_manager.catch_interrupt();
+        pin!(shutdown_future);
 
         loop {
             tokio::select! {
                 biased;
-                interrupt_res = task_manager.catch_interrupt() => {
+                interrupt_res = &mut shutdown_future => {
                     info!("received interrupt");
                     if let Err(err) = interrupt_res {
                         error!("runtime interrupt failure: {err}")
@@ -304,7 +156,7 @@ impl Rewarder {
             }
         }
 
-        self.nyxd_scraper.stop().await;
+        self.epoch_signing.nyxd_scraper.stop().await;
 
         /*
            task 1:
@@ -323,75 +175,3 @@ impl Rewarder {
         todo!()
     }
 }
-
-fn val_to_nym(addr: &str) -> String {
-    let foo: AccountId = addr.parse().unwrap();
-    let bar = AccountId::new("n", &foo.to_bytes()).unwrap();
-    bar.to_string()
-}
-
-fn make_url(oper: &str) -> String {
-    format!("https://rpc.nymtech.net/api/cosmos/staking/v1beta1/validators/{oper}")
-}
-
-async fn get_moniker(addr: AccountId) -> String {
-    let oper = AccountId::new("nvaloper", &addr.to_bytes())
-        .unwrap()
-        .to_string();
-
-    let foo: serde_json::Value = reqwest::get(make_url(&oper))
-        .await
-        .unwrap()
-        .json()
-        .await
-        .unwrap();
-
-    println!("raw: {foo:?}");
-
-    let Some(a) = foo.as_object() else {
-        return "UNKNOWN".to_string();
-    };
-    let Some(b) = a.get("validator").and_then(|o| o.as_object()) else {
-        return "UNKNOWN".to_string();
-    };
-    let Some(c) = b.get("description").and_then(|o| o.as_object()) else {
-        return "UNKNOWN".to_string();
-    };
-    let moniker = c.get("moniker").unwrap().as_str().unwrap();
-    moniker.to_string()
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use std::str::FromStr;
-
-    #[test]
-    fn aaa() {
-        let addr = AccountId::from_str("nvalcons1yn8kzqna703x5a6wh449ylw70u5drjejx5t6dz").unwrap();
-        println!("{}", AccountId::new("n", &addr.to_bytes()).unwrap());
-        println!("{}", AccountId::new("nvaloper", &addr.to_bytes()).unwrap());
-        println!("{}", AccountId::new("nvalcons", &addr.to_bytes()).unwrap());
-        // println!("{}", AccountId::new("n", &addr.to_bytes()).unwrap());
-
-        // let b = val_to_nym("nvaloper1q8cnx8s06q7ralnskqvj0acvqgacau6djqkm3z");
-        // println!("{b}");
-    }
-
-    #[tokio::test]
-    async fn bar() {
-        let oper = "nvaloper18xr68spwm96vvehuvwf6ay9er0gd7q7ae8w8ns";
-        let foo: serde_json::Value = reqwest::get(make_url(oper))
-            .await
-            .unwrap()
-            .json()
-            .await
-            .unwrap();
-
-        let a = foo.as_object().unwrap();
-        let b = a.get("validator").unwrap().as_object().unwrap();
-        let c = b.get("description").unwrap().as_object().unwrap();
-        let moniker = c.get("moniker").unwrap().as_str().unwrap();
-        println!("moniker: {moniker}")
-    }
-}

nym-validator-rewarder/src/rewarder/nyxd_client.rs

@@ -0,0 +1,35 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::config::Config;
+use nym_validator_client::{nyxd, DirectSigningHttpRpcNyxdClient};
+use std::sync::Arc;
+use tokio::sync::RwLock;
+
+#[derive(Clone)]
+pub struct NyxdClient {
+    inner: Arc<RwLock<DirectSigningHttpRpcNyxdClient>>,
+}
+
+impl NyxdClient {
+    pub(crate) fn new(config: &Config) -> Self {
+        let details = config.get_network_details();
+        let nyxd_url = config.get_nyxd_url();
+
+        let client_config = nyxd::Config::try_from_nym_network_details(&details)
+            .expect("failed to construct valid validator client config with the provided network");
+
+        let mnemonic = config.base.mnemonic.clone();
+
+        let inner = DirectSigningHttpRpcNyxdClient::connect_with_mnemonic(
+            client_config,
+            nyxd_url.as_str(),
+            mnemonic,
+        )
+        .expect("Failed to connect to nyxd!");
+
+        NyxdClient {
+            inner: Arc::new(RwLock::new(inner)),
+        }
+    }
+}