Merge pull request #4253 from nymtech/feature/validator-rewarder

Feature/validator rewarder
This commit is contained in:
Tommy Verrall
2024-01-15 09:25:26 +01:00
committed by GitHub
73 changed files with 3445 additions and 130 deletions
Generated
+136 -10
View File
@@ -1649,6 +1649,16 @@ dependencies = [
"tendermint-proto",
]
[[package]]
name = "cosmos-sdk-proto"
version = "0.20.0"
source = "git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features#41ed4631e146268b0300033c8bbb993d79a49f58"
dependencies = [
"prost 0.12.1",
"prost-types 0.12.1",
"tendermint-proto",
]
[[package]]
name = "cosmrs"
version = "0.15.0"
@@ -1656,7 +1666,26 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "47126f5364df9387b9d8559dcef62e99010e1d4098f39eb3f7ee4b5c254e40ea"
dependencies = [
"bip32",
"cosmos-sdk-proto",
"cosmos-sdk-proto 0.20.0 (registry+https://github.com/rust-lang/crates.io-index)",
"ecdsa 0.16.8",
"eyre",
"k256",
"rand_core 0.6.4",
"serde",
"serde_json",
"signature 2.1.0",
"subtle-encoding",
"tendermint",
"thiserror",
]
[[package]]
name = "cosmrs"
version = "0.15.0"
source = "git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features#41ed4631e146268b0300033c8bbb993d79a49f58"
dependencies = [
"bip32",
"cosmos-sdk-proto 0.20.0 (git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features)",
"ecdsa 0.16.8",
"eyre",
"k256",
@@ -2212,6 +2241,16 @@ dependencies = [
"darling_macro 0.14.4",
]
[[package]]
name = "darling"
version = "0.20.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0209d94da627ab5605dcccf08bb18afa5009cfbef48d8a8b7d7bdbc79be25c5e"
dependencies = [
"darling_core 0.20.3",
"darling_macro 0.20.3",
]
[[package]]
name = "darling_core"
version = "0.13.4"
@@ -2240,6 +2279,20 @@ dependencies = [
"syn 1.0.109",
]
[[package]]
name = "darling_core"
version = "0.20.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "177e3443818124b357d8e76f53be906d60937f0d3a90773a664fa63fa253e621"
dependencies = [
"fnv",
"ident_case",
"proc-macro2",
"quote",
"strsim",
"syn 2.0.38",
]
[[package]]
name = "darling_macro"
version = "0.13.4"
@@ -2262,6 +2315,17 @@ dependencies = [
"syn 1.0.109",
]
[[package]]
name = "darling_macro"
version = "0.20.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "836a9bbc7ad63342d6d6e7b815ccab164bc77a2d95d84bc3117a8c0d5c98e2d5"
dependencies = [
"darling_core 0.20.3",
"quote",
"syn 2.0.38",
]
[[package]]
name = "dashmap"
version = "5.5.3"
@@ -5945,7 +6009,7 @@ name = "nym-api-requests"
version = "0.1.0"
dependencies = [
"bs58 0.4.0",
"cosmrs",
"cosmrs 0.15.0 (git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features)",
"cosmwasm-std",
"getset",
"nym-coconut-interface",
@@ -6003,7 +6067,7 @@ name = "nym-bity-integration"
version = "0.1.0"
dependencies = [
"anyhow",
"cosmrs",
"cosmrs 0.15.0 (git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features)",
"eyre",
"k256",
"nym-cli-commands",
@@ -6048,7 +6112,7 @@ dependencies = [
"cfg-if",
"clap 4.4.7",
"comfy-table",
"cosmrs",
"cosmrs 0.15.0 (git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features)",
"cosmwasm-std",
"cw-utils",
"handlebars",
@@ -6317,7 +6381,7 @@ name = "nym-credentials"
version = "0.1.0"
dependencies = [
"bls12_381",
"cosmrs",
"cosmrs 0.15.0 (git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features)",
"log",
"nym-api-requests",
"nym-coconut-interface",
@@ -6676,7 +6740,6 @@ dependencies = [
"nym-types",
"nym-validator-client",
"opentelemetry",
"pretty_env_logger",
"rand 0.7.3",
"serde",
"serde_json",
@@ -7405,7 +7468,7 @@ name = "nym-types"
version = "1.0.0"
dependencies = [
"base64 0.21.4",
"cosmrs",
"cosmrs 0.15.0 (git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features)",
"cosmwasm-std",
"eyre",
"hmac 0.12.1",
@@ -7439,7 +7502,7 @@ dependencies = [
"bip32",
"bip39",
"colored",
"cosmrs",
"cosmrs 0.15.0 (git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features)",
"cosmwasm-std",
"cw-controllers",
"cw-utils",
@@ -7481,6 +7544,40 @@ dependencies = [
"zeroize",
]
[[package]]
name = "nym-validator-rewarder"
version = "0.1.0"
dependencies = [
"anyhow",
"bip39",
"clap 4.4.7",
"cosmwasm-std",
"futures",
"humantime 2.1.0",
"humantime-serde",
"nym-bin-common",
"nym-coconut",
"nym-coconut-bandwidth-contract-common",
"nym-coconut-dkg-common",
"nym-config",
"nym-credentials",
"nym-crypto",
"nym-network-defaults",
"nym-task",
"nym-validator-client",
"nyxd-scraper",
"serde",
"serde_with",
"sha2 0.10.8",
"sqlx",
"thiserror",
"time",
"tokio",
"tracing",
"url",
"zeroize",
]
[[package]]
name = "nym-vesting-contract-common"
version = "0.7.0"
@@ -7499,7 +7596,7 @@ dependencies = [
name = "nym-wallet-types"
version = "1.0.0"
dependencies = [
"cosmrs",
"cosmrs 0.15.0 (registry+https://github.com/rust-lang/crates.io-index)",
"cosmwasm-std",
"hex-literal",
"nym-config",
@@ -7583,7 +7680,7 @@ version = "0.1.0"
dependencies = [
"async-trait",
"const_format",
"cosmrs",
"cosmrs 0.15.0 (git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features)",
"eyre",
"futures",
"nym-bin-common",
@@ -9807,6 +9904,35 @@ dependencies = [
"serde",
]
[[package]]
name = "serde_with"
version = "3.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "64cd236ccc1b7a29e7e2739f27c0b2dd199804abc4290e32f59f3b68d6405c23"
dependencies = [
"base64 0.21.4",
"chrono",
"hex",
"indexmap 1.9.3",
"indexmap 2.0.2",
"serde",
"serde_json",
"serde_with_macros",
"time",
]
[[package]]
name = "serde_with_macros"
version = "3.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "93634eb5f75a2323b16de4748022ac4297f9e76b6dced2be287a099f41b5e788"
dependencies = [
"darling 0.20.3",
"proc-macro2",
"quote",
"syn 2.0.38",
]
[[package]]
name = "serde_yaml"
version = "0.9.25"
+6 -1
View File
@@ -102,6 +102,7 @@ members = [
"nym-node",
"nym-node/nym-node-requests",
"nym-outfox",
"nym-validator-rewarder",
"tools/internal/ssl-inject",
"tools/internal/sdk-version-bump",
"tools/nym-cli",
@@ -124,6 +125,7 @@ default-members = [
"nym-api",
"tools/nymvisor",
"explorer-api",
"nym-validator-rewarder",
]
exclude = ["explorer", "contracts", "nym-wallet", "nym-connect/mobile/src-tauri", "nym-connect/desktop", "nym-vpn/ui/src-tauri", "cpu-cycles"]
@@ -201,7 +203,10 @@ cw-controllers = { version = "=1.1.0" }
# cosmrs-related
bip32 = "0.5.1"
cosmrs = "=0.15.0"
# temporarily using a fork again (yay.) because we need staking and slashing support
cosmrs = { git = "https://github.com/jstuczyn/cosmos-rust", branch ="nym-temp/all-validator-features" }
#cosmrs = { git = "https://github.com/jstuczyn/cosmos-rust", branch = "nym-temp/all-validator-features" } # unfortuntely we need a fork by yours truly to get the staking support
tendermint = "0.34" # same version as used by cosmrs
tendermint-rpc = "0.34" # same version as used by cosmrs
prost = "0.12"
@@ -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,
@@ -5,21 +5,24 @@ use crate::nym_api::error::NymAPIError;
use crate::nym_api::routes::{CORE_STATUS_COUNT, SINCE_ARG};
use async_trait::async_trait;
use http_api_client::{ApiClient, NO_PARAMS};
use nym_api_requests::coconut::models::{
EpochCredentialsResponse, IssuedCredentialResponse, IssuedCredentialsResponse,
pub use nym_api_requests::{
coconut::{
models::{
EpochCredentialsResponse, IssuedCredential, IssuedCredentialBody,
IssuedCredentialResponse, IssuedCredentialsResponse,
},
BlindSignRequestBody, BlindedSignatureResponse, CredentialsRequestBody,
VerifyCredentialBody, VerifyCredentialResponse,
},
models::{
ComputeRewardEstParam, DescribedGateway, GatewayBondAnnotated, GatewayCoreStatusResponse,
GatewayStatusReportResponse, GatewayUptimeHistoryResponse, InclusionProbabilityResponse,
MixNodeBondAnnotated, MixnodeCoreStatusResponse, MixnodeStatusReportResponse,
MixnodeStatusResponse, MixnodeUptimeHistoryResponse, RewardEstimationResponse,
StakeSaturationResponse, UptimeResponse,
},
};
use nym_api_requests::coconut::{
BlindSignRequestBody, BlindedSignatureResponse, CredentialsRequestBody, VerifyCredentialBody,
VerifyCredentialResponse,
};
use nym_api_requests::models::{
ComputeRewardEstParam, DescribedGateway, GatewayBondAnnotated, GatewayCoreStatusResponse,
GatewayStatusReportResponse, GatewayUptimeHistoryResponse, InclusionProbabilityResponse,
MixNodeBondAnnotated, MixnodeCoreStatusResponse, MixnodeStatusReportResponse,
MixnodeStatusResponse, MixnodeUptimeHistoryResponse, RewardEstimationResponse,
StakeSaturationResponse, UptimeResponse,
};
use nym_coconut_dkg_common::types::EpochId;
pub use nym_coconut_dkg_common::types::EpochId;
use nym_mixnet_contract_common::mixnode::MixNodeDetails;
use nym_mixnet_contract_common::{GatewayBond, IdentityKeyRef, MixId};
use nym_name_service_common::response::NamesListResponse;
@@ -8,6 +8,8 @@ use cosmwasm_std::{Fraction, Uint128};
use serde::{Deserialize, Serialize};
use std::fmt;
use std::ops::Div;
use std::str::FromStr;
use thiserror::Error;
#[derive(Serialize, Deserialize, Clone, Copy, Default, Debug, PartialEq, Eq)]
pub struct MismatchedDenoms;
@@ -126,6 +128,37 @@ impl From<CosmWasmCoin> for Coin {
}
}
// unfortunately cosmwasm didn't re-export this correct so we just redefine its
#[derive(Error, Debug, PartialEq, Eq)]
pub enum CoinFromStrError {
#[error("Missing denominator")]
MissingDenom,
#[error("Missing amount or non-digit characters in amount")]
MissingAmount,
#[error("Invalid amount: {0}")]
InvalidAmount(#[from] std::num::ParseIntError),
}
impl FromStr for Coin {
type Err = CoinFromStrError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let pos = s
.find(|c: char| !c.is_ascii_digit())
.ok_or(CoinFromStrError::MissingDenom)?;
let (amount, denom) = s.split_at(pos);
if amount.is_empty() {
return Err(CoinFromStrError::MissingAmount);
}
Ok(Coin {
amount: amount.parse::<u128>()?,
denom: denom.to_string(),
})
}
}
pub trait CoinConverter {
type Target;
@@ -7,12 +7,12 @@ use crate::nyxd::error::NyxdError;
use crate::nyxd::CosmWasmClient;
use async_trait::async_trait;
use cosmrs::AccountId;
use nym_coconut_dkg_common::dealer::{
ContractDealing, DealerDetailsResponse, PagedDealerResponse, PagedDealingsResponse,
use nym_coconut_dkg_common::{
dealer::{ContractDealing, DealerDetailsResponse, PagedDealerResponse, PagedDealingsResponse},
msg::QueryMsg as DkgQueryMsg,
types::{DealerDetails, Epoch, EpochId, InitialReplacementData},
verification_key::{ContractVKShare, PagedVKSharesResponse},
};
use nym_coconut_dkg_common::msg::QueryMsg as DkgQueryMsg;
use nym_coconut_dkg_common::types::{DealerDetails, Epoch, EpochId, InitialReplacementData};
use nym_coconut_dkg_common::verification_key::{ContractVKShare, PagedVKSharesResponse};
use serde::Deserialize;
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
@@ -52,10 +52,6 @@ use wasmtimer::tokio::sleep;
pub const DEFAULT_BROADCAST_POLLING_RATE: Duration = Duration::from_secs(4);
pub const DEFAULT_BROADCAST_TIMEOUT: Duration = Duration::from_secs(60);
#[cfg(feature = "http-client")]
#[async_trait]
impl CosmWasmClient for cosmrs::rpc::HttpClient {}
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
pub trait CosmWasmClient: TendermintRpcClient {
@@ -522,3 +518,7 @@ pub trait CosmWasmClient: TendermintRpcClient {
res.try_into()
}
}
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
impl<T> CosmWasmClient for T where T: TendermintRpcClient {}
@@ -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<_, _>>()?;
@@ -1,7 +1,7 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::nyxd::cosmwasm_client::client_traits::{CosmWasmClient, SigningCosmWasmClient};
use crate::nyxd::cosmwasm_client::client_traits::SigningCosmWasmClient;
use crate::nyxd::error::NyxdError;
use crate::nyxd::{Config, GasPrice, Hash, Height};
use crate::rpc::TendermintRpcClient;
@@ -26,6 +26,7 @@ use cosmrs::rpc::{HttpClient, HttpClientUrl};
pub mod client_traits;
mod helpers;
pub mod logs;
pub mod module_traits;
pub mod types;
#[derive(Debug)]
@@ -329,14 +330,6 @@ where
}
}
#[async_trait]
impl<C, S> CosmWasmClient for MaybeSigningClient<C, S>
where
C: TendermintRpcClient + Send + Sync,
S: Send + Sync,
{
}
#[async_trait]
impl<C, S> SigningCosmWasmClient for MaybeSigningClient<C, S>
where
@@ -0,0 +1,8 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
pub mod slashing;
pub mod staking;
pub use staking::query::StakingQueryClient;
// pub use slashing::query
@@ -0,0 +1,4 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
pub mod query;
@@ -0,0 +1,2 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
@@ -0,0 +1,8 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
pub mod query;
pub use cosmrs::staking::{
QueryHistoricalInfoResponse, QueryValidatorResponse, QueryValidatorsResponse, Validator,
};
@@ -0,0 +1,78 @@
// 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;
use cosmrs::proto::cosmos::staking::v1beta1::{
QueryHistoricalInfoRequest as ProtoQueryHistoricalInfoRequest,
QueryHistoricalInfoResponse as ProtoQueryHistoricalInfoResponse,
QueryValidatorRequest as ProtoQueryValidatorRequest,
QueryValidatorResponse as ProtoQueryValidatorResponse,
QueryValidatorsRequest as ProtoQueryValidatorsRequest,
QueryValidatorsResponse as ProtoQueryValidatorsResponse,
};
use cosmrs::staking::{QueryHistoricalInfoRequest, QueryValidatorRequest, QueryValidatorsRequest};
use cosmrs::AccountId;
// TODO: change trait restriction from `CosmWasmClient` to `TendermintRpcClient`
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
pub trait StakingQueryClient: CosmWasmClient {
async fn historical_info(&self, height: i64) -> Result<QueryHistoricalInfoResponse, NyxdError> {
let path = Some("/cosmos.staking.v1beta1.Query/HistoricalInfo".to_owned());
let req = QueryHistoricalInfoRequest { height };
let res = self
.make_abci_query::<ProtoQueryHistoricalInfoRequest, ProtoQueryHistoricalInfoResponse>(
path,
req.into(),
)
.await?;
Ok(res.try_into()?)
}
async fn validator(
&self,
validator_addr: AccountId,
) -> Result<QueryValidatorResponse, NyxdError> {
let path = Some("/cosmos.staking.v1beta1.Query/Validator".to_owned());
let req = QueryValidatorRequest { validator_addr };
let res = self
.make_abci_query::<ProtoQueryValidatorRequest, ProtoQueryValidatorResponse>(
path,
req.into(),
)
.await?;
Ok(res.try_into()?)
}
async fn validators(
&self,
status: String,
pagination: Option<PageRequest>,
) -> Result<QueryValidatorsResponse, NyxdError> {
let path = Some("/cosmos.staking.v1beta1.Query/Validators".to_owned());
let req = QueryValidatorsRequest { status, pagination };
let res = self
.make_abci_query::<ProtoQueryValidatorsRequest, ProtoQueryValidatorsResponse>(
path,
req.into(),
)
.await?;
Ok(res.try_into()?)
}
}
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
impl<T> StakingQueryClient for T where T: CosmWasmClient {}
@@ -29,24 +29,30 @@ use tendermint_rpc::endpoint::*;
use tendermint_rpc::{Error as TendermintRpcError, Order};
use url::Url;
pub use crate::nyxd::cosmwasm_client::client_traits::{CosmWasmClient, SigningCosmWasmClient};
pub use crate::nyxd::fee::Fee;
pub use crate::nyxd::{
cosmwasm_client::{
client_traits::{CosmWasmClient, SigningCosmWasmClient},
module_traits::{self, StakingQueryClient},
},
fee::Fee,
};
pub use crate::rpc::TendermintRpcClient;
pub use coin::Coin;
pub use cosmrs::bank::MsgSend;
pub use cosmrs::tendermint::abci::{
response::DeliverTx, types::ExecTxResult, Event, EventAttribute,
pub use cosmrs::{
bank::MsgSend,
bip32,
crypto::PublicKey,
query::{PageRequest, PageResponse},
tendermint::{
abci::{response::DeliverTx, types::ExecTxResult, Event, EventAttribute},
block::Height,
hash::{self, Algorithm, Hash},
validator::Info as TendermintValidatorInfo,
Time as TendermintTime,
},
tx::{self, Msg},
AccountId, Any, Coin as CosmosCoin, Denom, Gas,
};
pub use cosmrs::tendermint::block::Height;
pub use cosmrs::tendermint::hash::{self, Algorithm, Hash};
pub use cosmrs::tendermint::validator::Info as TendermintValidatorInfo;
pub use cosmrs::tendermint::Time as TendermintTime;
pub use cosmrs::tx::Msg;
pub use cosmrs::tx::{self};
pub use cosmrs::Any;
pub use cosmrs::Coin as CosmosCoin;
pub use cosmrs::Gas;
pub use cosmrs::{bip32, AccountId, Denom};
pub use cosmwasm_std::Coin as CosmWasmCoin;
pub use cw2;
pub use cw3;
@@ -56,9 +62,8 @@ pub use fee::{gas_price::GasPrice, GasAdjustable, GasAdjustment};
pub use tendermint_rpc::{
endpoint::{tx::Response as TxResponse, validators::Response as ValidatorResponse},
query::Query,
Paging,
Paging, Request, Response, SimpleRequest,
};
pub use tendermint_rpc::{Request, Response, SimpleRequest};
#[cfg(feature = "http-client")]
use crate::http_client;
@@ -98,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>,
@@ -729,7 +742,7 @@ where
where
H: Into<Height> + Send,
{
self.client.validators(height, paging).await
TendermintRpcClient::validators(&self.client, height, paging).await
}
async fn latest_consensus_params(
@@ -804,14 +817,6 @@ where
}
}
#[async_trait]
impl<C, S> CosmWasmClient for NyxdClient<C, S>
where
C: TendermintRpcClient + Send + Sync,
S: Send + Sync,
{
}
impl<C, S> OfflineSigner for NyxdClient<C, S>
where
S: OfflineSigner,
+13 -1
View File
@@ -111,6 +111,7 @@ impl NymNetworkDetails {
.with_additional_validator_endpoint(ValidatorDetails::new(
var(var_names::NYXD).expect("nyxd validator not set"),
Some(var(var_names::NYM_API).expect("nym api not set")),
get_optional_env(var_names::NYXD_WEBSOCKET),
))
.with_mixnet_contract(Some(
var(var_names::MIXNET_CONTRACT_ADDRESS).expect("mixnet contract not set"),
@@ -340,6 +341,9 @@ impl DenomDetailsOwned {
pub struct ValidatorDetails {
// it is assumed those values are always valid since they're being provided in our defaults file
pub nyxd_url: String,
//
pub websocket_url: Option<String>,
// Right now api_url is optional as we are not running the api reliably on all validators
// however, later on it should be a mandatory field
pub api_url: Option<String>,
@@ -347,9 +351,10 @@ pub struct ValidatorDetails {
}
impl ValidatorDetails {
pub fn new<S: Into<String>>(nyxd_url: S, api_url: Option<S>) -> Self {
pub fn new<S: Into<String>>(nyxd_url: S, api_url: Option<S>, websocket_url: Option<S>) -> Self {
ValidatorDetails {
nyxd_url: nyxd_url.into(),
websocket_url: websocket_url.map(Into::into),
api_url: api_url.map(Into::into),
}
}
@@ -357,6 +362,7 @@ impl ValidatorDetails {
pub fn new_nyxd_only<S: Into<String>>(nyxd_url: S) -> Self {
ValidatorDetails {
nyxd_url: nyxd_url.into(),
websocket_url: None,
api_url: None,
}
}
@@ -372,6 +378,12 @@ impl ValidatorDetails {
.as_ref()
.map(|url| url.parse().expect("the provided api url is invalid!"))
}
pub fn websocket_url(&self) -> Option<Url> {
self.websocket_url
.as_ref()
.map(|url| url.parse().expect("the provided websocket url is invalid!"))
}
}
fn fix_deprecated_environmental_variables() {
+8 -1
View File
@@ -26,6 +26,7 @@ pub const REWARDING_VALIDATOR_ADDRESS: &str = "n10yyd98e2tuwu0f7ypz9dy3hhjw7v772
pub const STATISTICS_SERVICE_DOMAIN_ADDRESS: &str = "https://mainnet-stats.nymte.ch:8090/";
pub const NYXD_URL: &str = "https://rpc.nymtech.net";
pub const NYM_API: &str = "https://validator.nymtech.net/api/";
pub const NYXD_WS: &str = "wss://rpc.nymtech.net/websocket";
pub const EXPLORER_API: &str = "https://explorer.nymtech.net/api/";
// I'm making clippy mad on purpose, because that url HAS TO be updated and deployed before merging
@@ -33,7 +34,11 @@ pub const EXIT_POLICY_URL: &str =
"https://nymtech.net/.wellknown/network-requester/exit-policy.txt";
pub(crate) fn validators() -> Vec<ValidatorDetails> {
vec![ValidatorDetails::new(NYXD_URL, Some(NYM_API))]
vec![ValidatorDetails::new(
NYXD_URL,
Some(NYM_API),
Some(NYXD_WS),
)]
}
const DEFAULT_SUFFIX: &str = "_MAINNET_DEFAULT";
@@ -111,6 +116,7 @@ pub fn export_to_env() {
);
set_var_to_default(var_names::NYXD, NYXD_URL);
set_var_to_default(var_names::NYM_API, NYM_API);
set_var_to_default(var_names::NYXD_WEBSOCKET, NYXD_WS);
set_var_to_default(var_names::EXPLORER_API, EXPLORER_API);
set_var_to_default(var_names::EXIT_POLICY_URL, EXIT_POLICY_URL);
}
@@ -159,6 +165,7 @@ pub fn export_to_env_if_not_set() {
);
set_var_conditionally_to_default(var_names::NYXD, NYXD_URL);
set_var_conditionally_to_default(var_names::NYM_API, NYM_API);
set_var_conditionally_to_default(var_names::NYXD_WEBSOCKET, NYXD_WS);
set_var_conditionally_to_default(var_names::EXPLORER_API, EXPLORER_API);
set_var_conditionally_to_default(var_names::EXIT_POLICY_URL, EXIT_POLICY_URL);
}
+1
View File
@@ -26,6 +26,7 @@ pub const SERVICE_PROVIDER_DIRECTORY_CONTRACT_ADDRESS: &str =
pub const NAME_SERVICE_CONTRACT_ADDRESS: &str = "NAME_SERVICE_CONTRACT_ADDRESS";
pub const NYXD: &str = "NYXD";
pub const NYM_API: &str = "NYM_API";
pub const NYXD_WEBSOCKET: &str = "NYXD_WS";
pub const EXPLORER_API: &str = "EXPLORER_API";
pub const EXIT_POLICY_URL: &str = "EXIT_POLICY";
+1 -1
View File
@@ -255,7 +255,7 @@ fn bench_coconut(c: &mut Criterion) {
b.iter(|| {
verify_partial_blind_signature(
&params,
&blind_sign_request,
blind_sign_request.get_private_attributes_pedersen_commitments(),
&public_attributes,
random_blind_signature,
partial_verification_key,
+2
View File
@@ -43,3 +43,5 @@ mod utils;
pub type Attribute = bls12_381::Scalar;
pub type PrivateAttribute = Attribute;
pub type PublicAttribute = Attribute;
pub use bls12_381::G1Projective;
+7 -8
View File
@@ -362,12 +362,12 @@ pub fn blind_sign(
/// The function returns `true` if the partial blind signature is valid, and `false` otherwise.
pub fn verify_partial_blind_signature(
params: &Parameters,
blind_sign_request: &BlindSignRequest,
private_attribute_commitments: &[G1Projective],
public_attributes: &[&Attribute],
blind_sig: &BlindedSignature,
partial_verification_key: &VerificationKey,
) -> bool {
let num_private_attributes = blind_sign_request.private_attributes_commitments.len();
let num_private_attributes = private_attribute_commitments.len();
if num_private_attributes + public_attributes.len() > partial_verification_key.beta_g2.len() {
return false;
}
@@ -388,8 +388,7 @@ pub fn verify_partial_blind_signature(
];
// for each private attribute, add (cm_i, beta_i) to the miller terms
for (private_attr_commit, beta_g2) in blind_sign_request
.private_attributes_commitments
for (private_attr_commit, beta_g2) in private_attribute_commitments
.iter()
.zip(&partial_verification_key.beta_g2)
{
@@ -422,7 +421,7 @@ pub fn verify_partial_blind_signature(
// is equivalent to checking e(a, b) • e(c, d)^{-1} == id
// and thus to e(a, b) • e(c^{-1}, d) == id
//
// compute e(c^1, g2) • e(s, alpha) • e(cm_0, beta_0) • e(cm_i, beta_i) • (s^pub_0, beta_{i+1}) (s^pub_j, beta_{i + j})
// compute e(c^{-1}, g2) • e(s, alpha) • e(cm_0, beta_0) • e(cm_i, beta_i) • (s^pub_0, beta_{i+1}) (s^pub_j, beta_{i + j})
multi_miller_loop(&terms_refs)
.final_exponentiation()
.is_identity()
@@ -519,7 +518,7 @@ mod tests {
assert!(verify_partial_blind_signature(
&params,
&request,
&request.private_attributes_commitments,
&public_attributes,
&blind_sig,
validator_keypair.verification_key()
@@ -539,7 +538,7 @@ mod tests {
assert!(verify_partial_blind_signature(
&params,
&request,
&request.private_attributes_commitments,
&[],
&blind_sig,
validator_keypair.verification_key()
@@ -568,7 +567,7 @@ mod tests {
// this assertion should fail, as we try to verify with a wrong validator key
assert!(!verify_partial_blind_signature(
&params,
&request,
&request.private_attributes_commitments,
&public_attributes,
&blind_sig,
validator2_keypair.verification_key()
+10 -1
View File
@@ -11,8 +11,10 @@ use crate::storage::{persist_block, ScraperStorage};
use futures::StreamExt;
use std::collections::{BTreeMap, HashSet, VecDeque};
use std::ops::{Add, Range};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc::{Sender, UnboundedReceiver};
use tokio::sync::Notify;
use tokio::time::{interval_at, Instant};
use tokio_stream::wrappers::UnboundedReceiverStream;
use tokio_util::sync::CancellationToken;
@@ -39,6 +41,7 @@ impl PendingSync {
pub struct BlockProcessor {
cancel: CancellationToken,
synced: Arc<Notify>,
last_processed_height: u32,
last_processed_at: Instant,
pending_sync: PendingSync,
@@ -60,6 +63,7 @@ pub struct BlockProcessor {
impl BlockProcessor {
pub async fn new(
cancel: CancellationToken,
synced: Arc<Notify>,
incoming: UnboundedReceiver<BlockToProcess>,
block_requester: Sender<BlockRequest>,
storage: ScraperStorage,
@@ -69,6 +73,7 @@ impl BlockProcessor {
Ok(BlockProcessor {
cancel,
synced,
last_processed_height: last_processed.try_into().unwrap_or_default(),
last_processed_at: Instant::now(),
pending_sync: Default::default(),
@@ -243,6 +248,10 @@ impl BlockProcessor {
}
self.try_request_pending().await;
if self.pending_sync.is_empty() {
self.synced.notify_one();
}
}
async fn try_request_pending(&mut self) -> bool {
@@ -294,7 +303,7 @@ impl BlockProcessor {
error!("failed to perform startup sync: {err}");
self.cancel.cancel();
return;
}
};
loop {
tokio::select! {
+1
View File
@@ -16,3 +16,4 @@ pub mod storage;
pub use modules::{BlockModule, MsgModule, TxModule};
pub use scraper::{Config, NyxdScraper};
pub use storage::models;
+14 -1
View File
@@ -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::sync::Arc;
use tendermint_rpc::WebSocketClientDriver;
use tokio::sync::mpsc::{channel, unbounded_channel};
use tokio::sync::Notify;
use tokio_util::sync::CancellationToken;
use tokio_util::task::TaskTracker;
use tracing::info;
@@ -54,6 +56,7 @@ impl NyxdScraperBuilder {
);
let mut block_processor = BlockProcessor::new(
scraper.cancel_token.clone(),
scraper.startup_sync.clone(),
processing_rx,
req_tx,
scraper.storage.clone(),
@@ -112,7 +115,8 @@ pub struct NyxdScraper {
task_tracker: TaskTracker,
cancel_token: CancellationToken,
storage: ScraperStorage,
startup_sync: Arc<Notify>,
pub storage: ScraperStorage,
}
impl NyxdScraper {
@@ -127,6 +131,7 @@ impl NyxdScraper {
config,
task_tracker: TaskTracker::new(),
cancel_token: CancellationToken::new(),
startup_sync: Arc::new(Default::default()),
storage,
})
}
@@ -164,6 +169,7 @@ impl NyxdScraper {
);
let block_processor = BlockProcessor::new(
self.cancel_token.clone(),
self.startup_sync.clone(),
processing_rx,
req_tx,
self.storage.clone(),
@@ -189,8 +195,15 @@ impl NyxdScraper {
Ok(())
}
pub async fn wait_for_startup_sync(&self) {
info!("awaiting startup chain sync");
self.startup_sync.notified().await
}
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
}
+27 -3
View File
@@ -1,7 +1,7 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::storage::models::Validator;
use crate::storage::models::{CommitSignature, Validator};
use sqlx::types::time::OffsetDateTime;
use sqlx::{Executor, Sqlite};
use tracing::{instrument, trace};
@@ -84,8 +84,8 @@ impl StorageManager {
SELECT COUNT(*) as count FROM pre_commit
WHERE
validator_address == ?
AND height > ?
AND height < ?
AND height >= ?
AND height <= ?
"#,
consensus_address,
start_height,
@@ -98,6 +98,24 @@ impl StorageManager {
Ok(count)
}
pub(crate) async fn get_precommit(
&self,
consensus_address: &str,
height: i64,
) -> Result<Option<CommitSignature>, sqlx::Error> {
sqlx::query_as(
r#"
SELECT * FROM pre_commit
WHERE validator_address = ?
AND height = ?
"#,
)
.bind(consensus_address)
.bind(height)
.fetch_optional(&self.connection_pool)
.await
}
pub(crate) async fn get_block_validators(
&self,
height: i64,
@@ -118,6 +136,12 @@ impl StorageManager {
.await
}
pub(crate) async fn get_validators(&self) -> Result<Vec<Validator>, sqlx::Error> {
sqlx::query_as("SELECT * FROM validator")
.fetch_all(&self.connection_pool)
.await
}
pub(crate) async fn get_last_processed_height(&self) -> Result<i64, sqlx::Error> {
let maybe_record = sqlx::query!(
r#"
+32 -4
View File
@@ -7,7 +7,7 @@ use crate::storage::manager::{
insert_block, insert_message, insert_precommit, insert_transaction, insert_validator,
update_last_processed, StorageManager,
};
use crate::storage::models::Validator;
use crate::storage::models::{CommitSignature, Validator};
use sqlx::types::time::OffsetDateTime;
use sqlx::{ConnectOptions, Sqlite, Transaction};
use std::fmt::Debug;
@@ -19,7 +19,7 @@ use tracing::{debug, error, info, instrument, trace, warn};
mod helpers;
mod manager;
mod models;
pub mod models;
pub type StorageTransaction = Transaction<'static, Sqlite>;
@@ -31,8 +31,6 @@ pub struct ScraperStorage {
impl ScraperStorage {
#[instrument]
pub async fn init<P: AsRef<Path> + Debug>(database_path: P) -> Result<Self, ScraperError> {
// TODO: we can inject here more stuff based on our nym-api global config
// struct. Maybe different pool size or timeout intervals?
let mut opts = sqlx::sqlite::SqliteConnectOptions::new()
.filename(database_path)
.create_if_missing(true);
@@ -91,6 +89,21 @@ impl ScraperStorage {
Ok(self.manager.get_last_block_height_before(time).await?)
}
pub async fn get_blocks_between(
&self,
start_time: OffsetDateTime,
end_time: OffsetDateTime,
) -> Result<i64, ScraperError> {
let Some(block_start) = self.get_first_block_height_after(start_time).await? else {
return Ok(0);
};
let Some(block_end) = self.get_last_block_height_before(end_time).await? else {
return Ok(0);
};
Ok(block_end - block_start)
}
pub async fn get_signed_between(
&self,
consensus_address: &str,
@@ -120,10 +133,25 @@ impl ScraperStorage {
.await
}
pub async fn get_precommit(
&self,
consensus_address: &str,
height: i64,
) -> Result<Option<CommitSignature>, ScraperError> {
Ok(self
.manager
.get_precommit(consensus_address, height)
.await?)
}
pub async fn get_block_signers(&self, height: i64) -> Result<Vec<Validator>, ScraperError> {
Ok(self.manager.get_block_validators(height).await?)
}
pub async fn get_all_known_validators(&self) -> Result<Vec<Validator>, ScraperError> {
Ok(self.manager.get_validators().await?)
}
pub async fn get_last_processed_height(&self) -> Result<i64, ScraperError> {
Ok(self.manager.get_last_processed_height().await?)
}
+1 -1
View File
@@ -4,7 +4,7 @@
use sqlx::types::time::OffsetDateTime;
use sqlx::FromRow;
#[derive(Debug, Clone, FromRow)]
#[derive(Debug, Clone, Eq, PartialEq, Hash, FromRow)]
pub struct Validator {
pub consensus_address: String,
pub consensus_pubkey: String,
+3 -3
View File
@@ -116,8 +116,8 @@ impl TaskManager {
}
#[cfg(not(target_arch = "wasm32"))]
pub async fn catch_interrupt(mut self) -> Result<(), SentError> {
let res = crate::wait_for_signal_and_error(&mut self).await;
pub async fn catch_interrupt(&mut self) -> Result<(), SentError> {
let res = crate::wait_for_signal_and_error(self).await;
log::info!("Sending shutdown");
self.signal_shutdown().ok();
@@ -629,7 +629,7 @@ impl TaskHandle {
#[cfg(not(target_arch = "wasm32"))]
pub async fn wait_for_shutdown(self) -> Result<(), SentError> {
match self {
TaskHandle::Internal(task_manager) => task_manager.catch_interrupt().await,
TaskHandle::Internal(mut task_manager) => task_manager.catch_interrupt().await,
TaskHandle::External(mut task_client) => {
task_client.recv().await;
Ok(())
+1
View File
@@ -22,6 +22,7 @@ REWARDING_VALIDATOR_ADDRESS=n10yyd98e2tuwu0f7ypz9dy3hhjw7v772q6287gy
STATISTICS_SERVICE_DOMAIN_ADDRESS="https://mainnet-stats.nymte.ch:8090"
NYXD="https://rpc.nymtech.net"
NYM_API="https://validator.nymtech.net/api/"
NYXD_WS="wss://rpc.nymtech.net/websocket"
EXPLORER_API="https://explorer.nymtech.net/api/"
DKG_TIME_CONFIGURATION="259200,300,300,60,60,1209600"
+1
View File
@@ -24,4 +24,5 @@ SERVICE_PROVIDER_DIRECTORY_CONTRACT_ADDRESS=n1ps5yutd7sufwg058qd7ac7ldnlazsvmhzq
STATISTICS_SERVICE_DOMAIN_ADDRESS="http://0.0.0.0"
EXPLORER_API=https://sandbox-explorer.nymtech.net/api
NYXD="https://rpc.sandbox.nymtech.net"
NYXD_WS="wss://rpc.sandbox.nymtech.net/websocket"
NYM_API="https://sandbox-nym-api1.nymtech.net/api"
+1 -1
View File
@@ -60,7 +60,7 @@ impl RunExternalNodeCmd {
.with_members_provider(members_provider)?
.build();
let shutdown = TaskManager::new(10);
let mut shutdown = TaskManager::new(10);
let shutdown_listener = shutdown.subscribe();
tokio::spawn(ephemera.run(shutdown_listener));
+1 -1
View File
@@ -79,7 +79,7 @@ impl ExplorerApi {
self.wait_for_interrupt(shutdown).await
}
async fn wait_for_interrupt(&self, shutdown: TaskManager) {
async fn wait_for_interrupt(&self, mut shutdown: TaskManager) {
let _res = shutdown.catch_interrupt().await;
log::info!("Stopping explorer API");
}
+3 -1
View File
@@ -391,7 +391,9 @@ impl<St> Gateway<St> {
))
}
async fn wait_for_interrupt(shutdown: TaskManager) -> Result<(), Box<dyn Error + Send + Sync>> {
async fn wait_for_interrupt(
mut shutdown: TaskManager,
) -> Result<(), Box<dyn Error + Send + Sync>> {
let res = shutdown.catch_interrupt().await;
log::info!("Stopping nym gateway");
res
+3 -4
View File
@@ -18,7 +18,7 @@ rust-version = "1.58.1"
[dependencies]
axum = { workspace = true }
anyhow = "1.0.40"
anyhow = { workspace = true }
bs58 = "0.4.0"
clap = { workspace = true, features = ["cargo", "derive"] }
colored = "2.0"
@@ -28,7 +28,6 @@ futures = { workspace = true }
humantime-serde = "1.0"
lazy_static = "1.4.0"
log = { workspace = true }
pretty_env_logger = "0.4.0"
rand = "0.7.3"
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
@@ -41,7 +40,7 @@ cfg-if = "1.0.0"
thiserror = { workspace = true }
## tracing
tracing = { version = "0.1.37", optional = true }
tracing = { workspace = true, optional = true }
opentelemetry = { version = "0.19.0", optional = true }
@@ -65,7 +64,7 @@ nym-bin-common = { path = "../common/bin-common", features = ["output_format"] }
cpu-cycles = { path = "../cpu-cycles", optional = true }
[dev-dependencies]
tokio = { version = "1.21.2", features = [
tokio = { workspace = true, features = [
"rt-multi-thread",
"net",
"signal",
+1 -1
View File
@@ -215,7 +215,7 @@ impl MixNode {
})
}
async fn wait_for_interrupt(&self, shutdown: TaskManager) {
async fn wait_for_interrupt(&self, mut shutdown: TaskManager) {
let _res = shutdown.catch_interrupt().await;
log::info!("Stopping nym mixnode");
}
@@ -187,18 +187,18 @@ pub struct EpochCredentialsResponse {
#[serde(rename_all = "camelCase")]
pub struct IssuedCredentialsResponse {
// note: BTreeMap returns ordered results so it's fine to use it with pagination
pub credentials: BTreeMap<i64, IssuedCredentialInner>,
pub credentials: BTreeMap<i64, IssuedCredentialBody>,
}
#[derive(Clone, Serialize, Deserialize, Debug)]
#[serde(rename_all = "camelCase")]
pub struct IssuedCredentialResponse {
pub credential: Option<IssuedCredentialInner>,
pub credential: Option<IssuedCredentialBody>,
}
#[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct IssuedCredentialInner {
pub struct IssuedCredentialBody {
pub credential: IssuedCredential,
pub signature: identity::Signature,
+2 -2
View File
@@ -3,7 +3,7 @@
use crate::coconut::error::Result;
use crate::coconut::storage::models::IssuedCredential;
use nym_api_requests::coconut::models::IssuedCredentialInner;
use nym_api_requests::coconut::models::IssuedCredentialBody;
use nym_api_requests::coconut::models::IssuedCredentialsResponse;
use std::collections::BTreeMap;
@@ -14,7 +14,7 @@ pub(crate) fn build_credentials_response(
for raw_credential in raw {
let id = raw_credential.id;
let api_issued = IssuedCredentialInner::try_from(raw_credential)?;
let api_issued = IssuedCredentialBody::try_from(raw_credential)?;
let old = credentials.insert(id, api_issued);
if old.is_some() {
// why do we panic here rather than return an error? because it's a critical failure because
+1 -8
View File
@@ -4,7 +4,7 @@
use crate::coconut::error::CoconutError;
use nym_api_requests::coconut::models::{
EpochCredentialsResponse, IssuedCredential as ApiIssuedCredential,
IssuedCredentialInner as ApiIssuedCredentialInner,
IssuedCredentialBody as ApiIssuedCredentialInner,
};
use nym_api_requests::coconut::BlindedSignatureResponse;
use nym_coconut::{Base58, BlindedSignature};
@@ -97,13 +97,6 @@ impl TryFrom<IssuedCredential> for BlindedSignature {
}
}
impl IssuedCredential {
// safety: this should only ever be called on sanitized data from the database,
// thus the unwraps are fine (if somebody manually entered their db file and modified it, it's on them)
// pub fn private_attribute_commitments(&self) -> Vec<Scalar>
// pub fn public_attributes(&self)
}
pub fn join_attributes<I, M>(attrs: I) -> String
where
I: IntoIterator<Item = M>,
+2 -2
View File
@@ -9,7 +9,7 @@ use async_trait::async_trait;
use cosmwasm_std::{coin, to_binary, Addr, CosmosMsg, Decimal, WasmMsg};
use cw3::ProposalResponse;
use cw4::MemberResponse;
use nym_api_requests::coconut::models::{IssuedCredentialInner, IssuedCredentialResponse};
use nym_api_requests::coconut::models::{IssuedCredentialBody, IssuedCredentialResponse};
use nym_api_requests::coconut::{
BlindSignRequestBody, BlindedSignatureResponse, VerifyCredentialBody, VerifyCredentialResponse,
};
@@ -716,7 +716,7 @@ impl TestFixture {
serde_json::from_str(&response.into_string().await.unwrap()).unwrap()
}
async fn issued_unchecked(&self, id: i64) -> IssuedCredentialInner {
async fn issued_unchecked(&self, id: i64) -> IssuedCredentialBody {
self.issued_credential(id)
.await
.unwrap()
+1 -1
View File
@@ -65,7 +65,7 @@ pub(crate) async fn execute(args: Args) -> anyhow::Result<()> {
config.validate()?;
let shutdown_handlers = start_nym_api_tasks(config).await?;
let mut shutdown_handlers = start_nym_api_tasks(config).await?;
let res = shutdown_handlers
.task_manager_handle
+2 -4
View File
@@ -1013,8 +1013,7 @@ dependencies = [
[[package]]
name = "cosmos-sdk-proto"
version = "0.20.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32560304ab4c365791fd307282f76637213d8083c1a98490c35159cd67852237"
source = "git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features#67718aa27a96efb9881e927a8f998c94b885093c"
dependencies = [
"prost",
"prost-types",
@@ -1024,8 +1023,7 @@ dependencies = [
[[package]]
name = "cosmrs"
version = "0.15.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "47126f5364df9387b9d8559dcef62e99010e1d4098f39eb3f7ee4b5c254e40ea"
source = "git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features#67718aa27a96efb9881e927a8f998c94b885093c"
dependencies = [
"bip32",
"cosmos-sdk-proto",
+48
View File
@@ -0,0 +1,48 @@
[package]
name = "nym-validator-rewarder"
version = "0.1.0"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
documentation.workspace = true
edition.workspace = true
license = "GPL-3"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
anyhow.workspace = true
bip39 = { workspace = true, features = ["zeroize"] }
cosmwasm-std.workspace = true
clap = { workspace = true, features = ["cargo"] }
futures.workspace = true
serde = { workspace = true, features = ["derive"] }
sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate", "time"] }
thiserror.workspace = true
tokio = { workspace = true, features = ["rt-multi-thread", "time", "macros"] }
tracing.workspace = true
time.workspace = true
url.workspace = true
zeroize.workspace = true
serde_with = "3.4.0"
sha2 = "0.10.8"
humantime = "2.1.0"
humantime-serde = "1.0"
# internal
nym-bin-common = { path = "../common/bin-common", features = ["output_format"] }
nym-config = { path = "../common/config" }
nym-coconut = { path = "../common/nymcoconut" }
nym-crypto = { path = "../common/crypto", features = ["asymmetric"] }
nym-credentials = { path = "../common/credentials" }
nym-network-defaults = { path = "../common/network-defaults" }
nym-task = { path = "../common/task" }
nym-validator-client = { path = "../common/client-libs/validator-client" }
nym-coconut-dkg-common = { path = "../common/cosmwasm-smart-contracts/coconut-dkg"}
nym-coconut-bandwidth-contract-common = { path = "../common/cosmwasm-smart-contracts/coconut-bandwidth-contract"}
nyxd-scraper = { path = "../common/nyxd-scraper" }
[build-dependencies]
sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] }
tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
+28
View File
@@ -0,0 +1,28 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
#[tokio::main]
async fn main() {
use sqlx::{Connection, SqliteConnection};
use std::env;
let out_dir = env::var("OUT_DIR").unwrap();
let database_path = format!("{out_dir}/scraper-example.sqlite");
let mut conn = SqliteConnection::connect(&format!("sqlite://{database_path}?mode=rwc"))
.await
.expect("Failed to create SQLx database connection");
sqlx::migrate!("./migrations")
.run(&mut conn)
.await
.expect("Failed to perform SQLx migrations");
#[cfg(target_family = "unix")]
println!("cargo:rustc-env=DATABASE_URL=sqlite://{}", &database_path);
#[cfg(target_family = "windows")]
// for some strange reason we need to add a leading `/` to the windows path even though it's
// not a valid windows path... but hey, it works...
println!("cargo:rustc-env=DATABASE_URL=sqlite:///{}", &database_path);
}
@@ -0,0 +1,71 @@
/*
* Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
* SPDX-License-Identifier: GPL-3.0-only
*/
CREATE TABLE rewarding_epoch
(
id INTEGER NOT NULL PRIMARY KEY,
start_time TIMESTAMP WITHOUT TIME ZONE NOT NULL,
end_time TIMESTAMP WITHOUT TIME ZONE NOT NULL,
budget TEXT NOT NULL,
spent TEXT NOT NULL,
rewarding_tx TEXT,
rewarding_error TEXT
);
CREATE TABLE epoch_block_signing
(
rewarding_epoch_id INTEGER NOT NULL PRIMARY KEY REFERENCES rewarding_epoch (id),
total_voting_power_at_epoch_start INTEGER NOT NULL,
num_blocks INTEGER NOT NULL,
budget TEXT NOT NULL
);
CREATE TABLE block_signing_reward
(
rewarding_epoch_id INTEGER NOT NULL REFERENCES rewarding_epoch (id),
validator_consensus_address TEXT NOT NULL,
operator_account TEXT NOT NULL,
amount TEXT NOT NULL,
voting_power BIGINT NOT NULL,
voting_power_share TEXT NOT NULL,
signed_blocks INTEGER NOT NULL,
signed_blocks_percent TEXT NOT NULL,
UNIQUE (rewarding_epoch_id, operator_account)
);
CREATE TABLE epoch_credential_issuance
(
rewarding_epoch_id INTEGER NOT NULL PRIMARY KEY REFERENCES rewarding_epoch (id),
starting_dkg_epoch INTEGER NOT NULL,
ending_dkg_epoch INTEGER NOT NULL,
total_issued_partial_credentials INTEGER NOT NULL,
budget TEXT NOT NULL
);
CREATE TABLE malformed_credential
(
rewarding_epoch_id INTEGER NOT NULL REFERENCES rewarding_epoch (id)
);
CREATE TABLE credential_issuance_reward
(
rewarding_epoch_id INTEGER NOT NULL REFERENCES rewarding_epoch (id),
operator_account TEXT NOT NULL,
amount TEXT NOT NULL,
api_endpoint TEXT NOT NULL,
issued_partial_credentials INTEGER NOT NULL,
issued_credentials_share TEXT NOT NULL,
validated_issued_credentials INTEGER NOT NULL,
UNIQUE (rewarding_epoch_id, operator_account)
);
-- CREATE TABLE credential_verification_reward
-- (
--
-- );
@@ -0,0 +1,17 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::error::NymRewarderError;
use nym_bin_common::bin_info_owned;
use nym_bin_common::output_format::OutputFormat;
#[derive(clap::Args, Debug)]
pub(crate) struct Args {
#[clap(short, long, default_value_t = OutputFormat::default())]
output: OutputFormat,
}
pub(crate) fn execute(args: Args) -> Result<(), NymRewarderError> {
println!("{}", args.output.format(&bin_info_owned!()));
Ok(())
}
+60
View File
@@ -0,0 +1,60 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::cli::ConfigOverridableArgs;
use crate::config::{default_config_directory, default_data_directory, Config};
use crate::error::NymRewarderError;
use nym_network_defaults::NymNetworkDetails;
use std::path::PathBuf;
use std::{fs, io};
#[derive(Debug, clap::Args)]
pub struct Args {
#[command(flatten)]
config_override: ConfigOverridableArgs,
/// Specifies custom location for the configuration file of nym validators rewarder.
#[clap(long)]
custom_config_path: Option<PathBuf>,
/// Mnemonic used for rewarding operations
#[clap(long)]
mnemonic: bip39::Mnemonic,
/// Overwrite existing configuration file.
#[clap(long, short)]
force: bool,
}
fn init_paths() -> io::Result<()> {
fs::create_dir_all(default_data_directory())?;
fs::create_dir_all(default_config_directory())
}
pub(crate) fn execute(args: Args) -> Result<(), NymRewarderError> {
let path = args
.custom_config_path
.clone()
.unwrap_or(Config::default_location());
if path.exists() && !args.force {
return Err(NymRewarderError::ExistingConfig { path });
}
init_paths().map_err(|source| NymRewarderError::PathInitialisationFailure { source })?;
let network = NymNetworkDetails::new_from_env();
let nyxd = network.endpoints[0].nyxd_url();
let Some(websocket) = network.endpoints[0].websocket_url() else {
return Err(NymRewarderError::UnavailableWebsocketUrl);
};
let config = Config::new(args.mnemonic, websocket, nyxd).with_override(args.config_override);
config.ensure_is_valid()?;
config
.save_to_path(&path)
.map_err(|source| NymRewarderError::ConfigSaveFailure { path, source })?;
Ok(())
}
+136
View File
@@ -0,0 +1,136 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::config::Config;
use crate::error::NymRewarderError;
use clap::{Parser, Subcommand};
use humantime_serde::re::humantime;
use nym_bin_common::bin_info;
use nym_validator_client::nyxd::Coin;
use std::path::PathBuf;
use std::sync::OnceLock;
use tracing::{debug, error};
use url::Url;
pub mod build_info;
pub mod init;
pub mod run;
pub mod upgrade_helpers;
// Helper for passing LONG_VERSION to clap
fn pretty_build_info_static() -> &'static str {
static PRETTY_BUILD_INFORMATION: OnceLock<String> = OnceLock::new();
PRETTY_BUILD_INFORMATION.get_or_init(|| bin_info!().pretty_print())
}
#[derive(Parser, Debug)]
#[clap(author = "Nymtech", version, long_version = pretty_build_info_static(), about)]
pub(crate) struct Cli {
/// Path pointing to an env file that configures the validator rewarder and overrides any preconfigured values.
#[clap(short, long)]
pub(crate) config_env_file: Option<std::path::PathBuf>,
/// Flag used for disabling the printed banner in tty.
#[clap(long)]
pub(crate) no_banner: bool,
#[clap(subcommand)]
command: Commands,
}
impl Cli {
pub(crate) async fn execute(self) -> Result<(), NymRewarderError> {
match self.command {
Commands::Init(args) => init::execute(args),
Commands::Run(args) => run::execute(args).await,
Commands::BuildInfo(args) => build_info::execute(args),
}
}
}
#[derive(Debug, clap::Args)]
pub struct ConfigOverridableArgs {
#[clap(long)]
pub disable_block_signing_rewarding: bool,
#[clap(long)]
pub disable_credential_issuance_rewarding: bool,
#[clap(long)]
pub credential_monitor_run_interval: Option<humantime::Duration>,
#[clap(long)]
pub credential_monitor_min_validation: Option<usize>,
#[clap(long)]
pub credential_monitor_sampling_rate: Option<f64>,
#[clap(long)]
pub scraper_endpoint: Option<Url>,
#[clap(long)]
pub nyxd_endpoint: Option<Url>,
#[clap(long)]
pub epoch_budget: Option<Coin>,
#[clap(long)]
pub epoch_duration: Option<humantime::Duration>,
#[clap(long)]
pub block_signing_reward_ratio: Option<f64>,
#[clap(long)]
pub credential_issuance_reward_ratio: Option<f64>,
#[clap(long)]
pub credential_verification_reward_ratio: Option<f64>,
}
#[derive(Subcommand, Debug)]
pub(crate) enum Commands {
/// Initialise a validator rewarder with persistent config.toml file.
Init(init::Args),
/// Run the validator rewarder with the preconfigured settings.
Run(run::Args),
/// Show build information of this binary
BuildInfo(build_info::Args),
}
fn try_load_current_config(custom_path: &Option<PathBuf>) -> Result<Config, NymRewarderError> {
let config_path = custom_path.clone().unwrap_or(Config::default_location());
debug!(
"attempting to load configuration file from {}",
config_path.display()
);
if let Ok(cfg) = Config::read_from_toml_file(&config_path) {
cfg.ensure_is_valid()?;
return Ok(cfg);
}
upgrade_helpers::try_upgrade_config(&config_path)?;
let config = Config::read_from_toml_file(&config_path).map_err(|err| {
error!(
"Failed to load config. Are you sure you have run `init` before? (Error was: {err})",
);
err
})?;
config.ensure_is_valid()?;
Ok(config)
}
#[cfg(test)]
mod tests {
use super::*;
use clap::CommandFactory;
#[test]
fn verify_cli() {
Cli::command().debug_assert();
}
}
+24
View File
@@ -0,0 +1,24 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::cli::{try_load_current_config, ConfigOverridableArgs};
use crate::error::NymRewarderError;
use crate::rewarder::Rewarder;
use std::path::PathBuf;
#[derive(Debug, clap::Args)]
pub struct Args {
#[command(flatten)]
config_override: ConfigOverridableArgs,
/// Specifies custom location for the configuration file of nym validators rewarder.
#[clap(long)]
custom_config_path: Option<PathBuf>,
}
pub(crate) async fn execute(args: Args) -> Result<(), NymRewarderError> {
let config =
try_load_current_config(&args.custom_config_path)?.with_override(args.config_override);
Rewarder::new(config).await?.run().await
}
@@ -0,0 +1,10 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::error::NymRewarderError;
use std::path::Path;
pub(crate) fn try_upgrade_config<P: AsRef<Path>>(config_path: P) -> Result<(), NymRewarderError> {
let _ = config_path;
Ok(())
}
+278
View File
@@ -0,0 +1,278 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::config::persistence::paths::ValidatorRewarderPaths;
use crate::config::r#override::ConfigOverride;
use crate::config::template::CONFIG_TEMPLATE;
use crate::error::NymRewarderError;
use nym_config::{
must_get_home, read_config_from_toml_file, save_formatted_config_to_file, NymConfigTemplate,
DEFAULT_CONFIG_DIR, DEFAULT_CONFIG_FILENAME, DEFAULT_DATA_DIR, NYM_DIR,
};
use nym_validator_client::nyxd::Coin;
use serde::{Deserialize, Serialize};
use serde_with::{serde_as, DisplayFromStr};
use std::io;
use std::path::{Path, PathBuf};
use std::time::Duration;
use tracing::debug;
use url::Url;
use zeroize::{Zeroize, ZeroizeOnDrop};
pub mod r#override;
pub mod persistence;
mod template;
const DEFAULT_REWARDER_DIR: &str = "validators-rewarder";
#[allow(clippy::inconsistent_digit_grouping)]
const DEFAULT_MIX_REWARDING_BUDGET: u128 = 1000_000000;
const DEFAULT_MIX_REWARDING_DENOM: &str = "unym";
const DEFAULT_EPOCH_DURATION: Duration = Duration::from_secs(60 * 60);
const DEFAULT_MONITOR_RUN_INTERVAL: Duration = Duration::from_secs(10 * 60);
const DEFAULT_MONITOR_MIN_VALIDATE: usize = 10;
const DEFAULT_MONITOR_SAMPLING_RATE: f64 = 0.10;
/// Get default path to rewarder's config directory.
/// It should get resolved to `$HOME/.nym/validators-rewarder/config`
pub fn default_config_directory() -> PathBuf {
must_get_home()
.join(NYM_DIR)
.join(DEFAULT_REWARDER_DIR)
.join(DEFAULT_CONFIG_DIR)
}
/// Get default path to rewarder's config file.
/// It should get resolved to `$HOME/.nym/validators-rewarder/config/config.toml`
pub fn default_config_filepath() -> PathBuf {
default_config_directory().join(DEFAULT_CONFIG_FILENAME)
}
/// Get default path to rewarder's data directory where files, such as keys, are stored.
/// It should get resolved to `$HOME/.nym/validators-rewarder/data`
pub fn default_data_directory() -> PathBuf {
must_get_home()
.join(NYM_DIR)
.join(DEFAULT_REWARDER_DIR)
.join(DEFAULT_DATA_DIR)
}
#[derive(Debug, Deserialize, Serialize, Zeroize, ZeroizeOnDrop)]
pub struct Config {
// additional metadata holding on-disk location of this config file
#[serde(skip)]
#[zeroize(skip)]
pub(crate) save_path: Option<PathBuf>,
#[zeroize(skip)]
#[serde(default)]
pub rewarding: Rewarding,
#[zeroize(skip)]
#[serde(default)]
pub block_signing: BlockSigning,
#[zeroize(skip)]
#[serde(default)]
pub issuance_monitor: IssuanceMonitor,
#[zeroize(skip)]
pub nyxd_scraper: NyxdScraper,
#[serde(flatten)]
pub base: Base,
#[zeroize(skip)]
pub storage_paths: ValidatorRewarderPaths,
}
impl NymConfigTemplate for Config {
fn template(&self) -> &'static str {
CONFIG_TEMPLATE
}
}
impl Config {
pub fn new(mnemonic: bip39::Mnemonic, websocket_url: Url, nyxd_url: Url) -> Self {
Config {
save_path: None,
rewarding: Rewarding::default(),
block_signing: Default::default(),
issuance_monitor: IssuanceMonitor::default(),
nyxd_scraper: NyxdScraper { websocket_url },
base: Base {
upstream_nyxd: nyxd_url,
mnemonic,
},
storage_paths: Default::default(),
}
}
pub fn scraper_config(&self) -> nyxd_scraper::Config {
nyxd_scraper::Config {
websocket_url: self.nyxd_scraper.websocket_url.clone(),
rpc_url: self.base.upstream_nyxd.clone(),
database_path: self.storage_paths.nyxd_scraper.clone(),
}
}
pub fn ensure_is_valid(&self) -> Result<(), NymRewarderError> {
self.rewarding.ratios.ensure_is_valid()?;
Ok(())
}
pub fn r#override<O: ConfigOverride>(&mut self, r#override: O) {
r#override.override_config(self)
}
pub fn with_override<O: ConfigOverride>(mut self, r#override: O) -> Self {
self.r#override(r#override);
self
}
// simple wrapper that reads config file and assigns path location
fn read_from_path<P: AsRef<Path>>(path: P) -> Result<Self, NymRewarderError> {
let path = path.as_ref();
let mut loaded: Config = read_config_from_toml_file(path).map_err(|source| {
NymRewarderError::ConfigLoadFailure {
path: path.to_path_buf(),
source,
}
})?;
loaded.ensure_is_valid()?;
loaded.save_path = Some(path.to_path_buf());
debug!("loaded config file from {}", path.display());
Ok(loaded)
}
pub fn read_from_toml_file<P: AsRef<Path>>(path: P) -> Result<Self, NymRewarderError> {
Self::read_from_path(path)
}
pub fn default_location() -> PathBuf {
default_config_filepath()
}
pub fn save_to_default_location(&self) -> io::Result<()> {
let config_save_location: PathBuf = Self::default_location();
save_formatted_config_to_file(self, config_save_location)
}
pub fn save_to_path<P: AsRef<Path>>(&self, path: P) -> io::Result<()> {
save_formatted_config_to_file(self, path)
}
}
#[derive(Debug, Deserialize, Serialize, Zeroize, ZeroizeOnDrop)]
pub struct Base {
/// Url to the upstream instance of nyxd to use for any queries and rewarding.
#[zeroize(skip)]
pub upstream_nyxd: Url,
/// Mnemonic to the nyx account distributing the rewards
pub(crate) mnemonic: bip39::Mnemonic,
}
#[serde_as]
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct Rewarding {
/// Specifies total budget for the epoch
#[serde_as(as = "DisplayFromStr")]
pub epoch_budget: Coin,
#[serde(with = "humantime_serde")]
pub epoch_duration: Duration,
pub ratios: RewardingRatios,
}
impl Default for Rewarding {
fn default() -> Self {
Rewarding {
epoch_budget: Coin::new(DEFAULT_MIX_REWARDING_BUDGET, DEFAULT_MIX_REWARDING_DENOM),
epoch_duration: DEFAULT_EPOCH_DURATION,
ratios: RewardingRatios::default(),
}
}
}
#[derive(Debug, Clone, Copy, Deserialize, Serialize)]
pub struct RewardingRatios {
/// The percent of the epoch reward being awarded for block signing.
pub block_signing: f64,
/// The percent of the epoch reward being awarded for credential issuance.
pub credential_issuance: f64,
/// The percent of the epoch reward being awarded for credential verification.
pub credential_verification: f64,
// /// The percent of the epoch reward given to Nym.
// pub nym: f64,
}
impl Default for RewardingRatios {
fn default() -> Self {
RewardingRatios {
block_signing: 0.67,
credential_issuance: 0.33,
credential_verification: 0.0,
// nym: 0.0,
}
}
}
impl RewardingRatios {
pub fn ensure_is_valid(&self) -> Result<(), NymRewarderError> {
if self.block_signing + self.credential_verification + self.credential_issuance != 1.0 {
todo!()
}
Ok(())
}
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct NyxdScraper {
/// Url to the websocket endpoint of a validator, for example `wss://rpc.nymtech.net/websocket`
pub websocket_url: Url,
// TODO: debug with everything that's currently hardcoded in the scraper
}
#[derive(Debug, Clone, Deserialize, Serialize, Copy)]
pub struct BlockSigning {
/// Specifies whether credential issuance for block signing is enabled.
pub enabled: bool,
}
impl Default for BlockSigning {
fn default() -> Self {
BlockSigning { enabled: true }
}
}
#[derive(Debug, Clone, Deserialize, Serialize, Copy)]
pub struct IssuanceMonitor {
/// Specifies whether credential issuance monitoring (and associated rewards) are enabled.
pub enabled: bool,
#[serde(with = "humantime_serde")]
pub run_interval: Duration,
/// Defines the minimum number of credentials the monitor will validate
/// regardless of the sampling rate
pub min_validate_per_issuer: usize,
/// The sampling rate of the issued credentials
pub sampling_rate: f64,
}
impl Default for IssuanceMonitor {
fn default() -> Self {
IssuanceMonitor {
enabled: true,
run_interval: DEFAULT_MONITOR_RUN_INTERVAL,
min_validate_per_issuer: DEFAULT_MONITOR_MIN_VALIDATE,
sampling_rate: DEFAULT_MONITOR_SAMPLING_RATE,
}
}
}
@@ -0,0 +1,63 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::cli::ConfigOverridableArgs;
use crate::config::Config;
pub trait ConfigOverride {
fn override_config(self, config: &mut Config);
}
impl ConfigOverride for ConfigOverridableArgs {
fn override_config(self, config: &mut Config) {
if self.disable_block_signing_rewarding {
config.block_signing.enabled = false
}
if self.disable_credential_issuance_rewarding {
config.issuance_monitor.enabled = false
}
if let Some(credential_monitor_run_interval) = self.credential_monitor_run_interval {
config.issuance_monitor.run_interval = credential_monitor_run_interval.into()
}
if let Some(credential_monitor_min_validation) = self.credential_monitor_min_validation {
config.issuance_monitor.min_validate_per_issuer = credential_monitor_min_validation
}
if let Some(credential_monitor_sampling_rate) = self.credential_monitor_sampling_rate {
config.issuance_monitor.sampling_rate = credential_monitor_sampling_rate
}
if let Some(scraper_endpoint) = self.scraper_endpoint {
config.nyxd_scraper.websocket_url = scraper_endpoint
}
if let Some(nyxd_endpoint) = self.nyxd_endpoint {
config.base.upstream_nyxd = nyxd_endpoint
}
if let Some(epoch_budget) = self.epoch_budget {
config.rewarding.epoch_budget = epoch_budget
}
if let Some(epoch_duration_secs) = self.epoch_duration {
config.rewarding.epoch_duration = epoch_duration_secs.into()
}
if let Some(block_signing_reward_ratio) = self.block_signing_reward_ratio {
config.rewarding.ratios.block_signing = block_signing_reward_ratio;
}
if let Some(credential_issuance_reward_ratio) = self.credential_issuance_reward_ratio {
config.rewarding.ratios.credential_issuance = credential_issuance_reward_ratio;
}
if let Some(credential_verification_reward_ratio) =
self.credential_verification_reward_ratio
{
config.rewarding.ratios.credential_verification = credential_verification_reward_ratio;
}
}
}
@@ -0,0 +1,4 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
pub mod paths;
@@ -0,0 +1,25 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::config::default_data_directory;
use serde::{Deserialize, Serialize};
use std::path::PathBuf;
pub const DEFAULT_SCRAPER_DB_FILENAME: &str = "nyxd_blocks.sqlite";
pub const DEFAULT_REWARD_HISTORY_DB_FILENAME: &str = "rewards.sqlite";
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct ValidatorRewarderPaths {
pub nyxd_scraper: PathBuf,
pub reward_history: PathBuf,
}
impl Default for ValidatorRewarderPaths {
fn default() -> Self {
ValidatorRewarderPaths {
nyxd_scraper: default_data_directory().join(DEFAULT_SCRAPER_DB_FILENAME),
reward_history: default_data_directory().join(DEFAULT_REWARD_HISTORY_DB_FILENAME),
}
}
}
@@ -0,0 +1,61 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
// While using normal toml marshalling would have been way simpler with less overhead,
// I think it's useful to have comments attached to the saved config file to explain behaviour of
// particular fields.
// Note: any changes to the template must be reflected in the appropriate structs.
pub(crate) const CONFIG_TEMPLATE: &str = r#"
# This is a TOML config file.
# For more information, see https://github.com/toml-lang/toml
# Url to the upstream instance of nyxd to use for any queries and rewarding.
upstream_nyxd = '{{ upstream_nyxd }}'
# Mnemonic to the nyx account distributing the rewards
mnemonic = '{{ mnemonic }}'
[storage_paths]
nyxd_scraper = '{{ storage_paths.nyxd_scraper }}'
reward_history = '{{ storage_paths.reward_history }}'
[rewarding]
# Specifies total budget for the epoch
epoch_budget = '{{ rewarding.epoch_budget }}'
epoch_duration = '{{ rewarding.epoch_duration }}'
[rewarding.ratios]
# The percent of the epoch reward being awarded for block signing.
block_signing = {{ rewarding.ratios.block_signing }}
# The percent of the epoch reward being awarded for credential issuance.
credential_issuance = {{ rewarding.ratios.credential_issuance }}
# The percent of the epoch reward being awarded for credential verification.
credential_verification = {{ rewarding.ratios.credential_verification }}
[block_signing]
# Specifies whether credential issuance for block signing is enabled.
enabled = {{ block_signing.enabled }}
[issuance_monitor]
# Specifies whether credential issuance monitoring (and associated rewards) are enabled.
enabled = {{ issuance_monitor.enabled }}
run_interval = '{{ issuance_monitor.run_interval }}'
# Defines the minimum number of credentials the monitor will validate
# regardless of the sampling rate
min_validate_per_issuer = {{ issuance_monitor.min_validate_per_issuer }}
# The sampling rate of the issued credentials
sampling_rate = {{ issuance_monitor.sampling_rate }}
[nyxd_scraper]
# Url to the websocket endpoint of a validator, for example `wss://rpc.nymtech.net/websocket`
websocket_url = '{{ nyxd_scraper.websocket_url }}'
"#;
+166
View File
@@ -0,0 +1,166 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use nym_coconut::CoconutError;
use nym_validator_client::nym_api::error::NymAPIError;
use nym_validator_client::nyxd::error::NyxdError;
use nym_validator_client::nyxd::tx::ErrorReport;
use nym_validator_client::nyxd::{AccountId, Coin, Hash};
use std::io;
use std::path::PathBuf;
use thiserror::Error;
#[derive(Debug, Error)]
pub enum NymRewarderError {
#[error("experienced internal database error: {0}")]
InternalDatabaseError(#[from] sqlx::Error),
#[error("failed to perform startup SQL migration: {0}")]
StartupMigrationFailure(#[from] sqlx::migrate::MigrateError),
#[error(
"failed to load config file using path '{}'. detailed message: {source}", path.display()
)]
ConfigLoadFailure {
path: PathBuf,
#[source]
source: io::Error,
},
#[error(
"failed to save config file using path '{}'. detailed message: {source}", path.display()
)]
ConfigSaveFailure {
path: PathBuf,
#[source]
source: io::Error,
},
#[error("failed to initialise paths")]
PathInitialisationFailure {
#[source]
source: io::Error,
},
#[error("there already exists a config file at: {}. if you want to overwrite its content, use --force flag", path.display())]
ExistingConfig { path: PathBuf },
// TODO: I think this one should get split into more, explicit, variants
#[error(transparent)]
NyxdFailure(#[from] NyxdError),
#[error("the provided rewarding ratios don't add up to 1. ratios: {ratios:?}")]
InvalidRewardingRatios { ratios: Vec<f32> },
#[error("chain scraping failure: {source}")]
ScraperFailure {
#[from]
source: nyxd_scraper::error::ScraperError,
},
// this should never happen but unwrapping everywhere was more cumbersome than just propagating the error
#[error("failed to determine epoch boundaries: {0}")]
TimeComponentFailure(#[from] time::error::ComponentRange),
#[error(
"could not convert operator address: {operator_address} to a nym account address: {source}"
)]
MalformedBech32Address {
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 },
#[error("api url ({raw}) provided by {runner_account} is invalid: {source}")]
MalformedApiUrl {
raw: String,
runner_account: AccountId,
#[source]
source: url::ParseError,
},
#[error("failed to resolve nym-api query: {0}")]
ApiQueryFailure(#[from] NymAPIError),
#[error("operator {runner_account} didn't return all requested credentials! requested {requested} but got only {received}")]
IncompleteRequest {
runner_account: AccountId,
requested: usize,
received: usize,
},
#[error("the following private attribute commitment is malformed: {raw}: {source}")]
MalformedCredentialCommitment {
raw: String,
#[source]
source: CoconutError,
},
#[error("the partial verification key for runner {runner} is malformed: {source}")]
MalformedPartialVerificationKey {
runner: String,
#[source]
source: CoconutError,
},
#[error("could not verify the blinded credential")]
BlindVerificationFailure,
#[error("the same deposit transaction ({tx_hash}) has been used for multiple issued credentials! {first} and {other}")]
DuplicateDepositHash {
tx_hash: Hash,
first: i64,
other: i64,
},
#[error("could not find the deposit value in the event of transaction {tx_hash}")]
DepositValueNotFound { tx_hash: Hash },
#[error("could not find the deposit info in the event of transaction {tx_hash}")]
DepositInfoNotFound { tx_hash: Hash },
#[error("the provided deposit value of transaction {tx_hash} is inconsistent. got '{request:?}' while the value on chain is '{on_chain}'")]
InconsistentDepositValue {
tx_hash: Hash,
request: Option<String>,
on_chain: String,
},
#[error("the provided deposit info of transaction {tx_hash} is inconsistent. got '{request:?}' while the value on chain is '{on_chain}'")]
InconsistentDepositInfo {
tx_hash: Hash,
request: Option<String>,
on_chain: String,
},
#[error("the current rewarder balance is insufficient to start the process. The epoch budget is: {} while we currently have {}. (the minimum is set to {})", .0.epoch_budget, .0.balance, .0.minimum)]
InsufficientRewarderBalance(Box<InsufficientBalance>),
#[error("the scraper websocket endpoint hasn't been provided")]
UnavailableWebsocketUrl,
}
#[derive(Debug)]
pub struct InsufficientBalance {
pub epoch_budget: Coin,
pub balance: Coin,
pub minimum: Coin,
}
+35
View File
@@ -0,0 +1,35 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
#![warn(clippy::expect_used)]
#![warn(clippy::unwrap_used)]
use crate::cli::Cli;
use clap::{crate_name, crate_version, Parser};
use nym_bin_common::logging::{maybe_print_banner, setup_tracing_logger};
use nym_network_defaults::setup_env;
pub mod cli;
pub mod config;
pub mod error;
mod rewarder;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
// std::env::set_var(
// "RUST_LOG",
// "debug,tendermint_rpc=warn,h2=warn,hyper=warn,rustls=warn,reqwest=warn,tungstenite=warn,async_tungstenite=warn",
// );
let args = Cli::parse();
setup_env(args.config_env_file.as_ref());
setup_tracing_logger();
if !args.no_banner {
maybe_print_banner(crate_name!(), crate_version!());
}
args.execute().await?;
Ok(())
}
@@ -0,0 +1,139 @@
// 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, RawValidatorResult};
use crate::rewarder::epoch::Epoch;
use crate::rewarder::nyxd_client::NyxdClient;
use nym_validator_client::nyxd::module_traits::staking;
use nym_validator_client::nyxd::PageRequest;
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) nyxd_client: NyxdClient,
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.nyxd_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.nyxd_client.validators(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_time;
let epoch_end = current_epoch.end_time;
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, RawValidatorResult::new(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)
}
}
@@ -0,0 +1,141 @@
// 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 moniker(&self) -> String {
self.staking_details
.description
.as_ref()
.map(|d| d.moniker.clone())
.unwrap_or("UNKNOWN MONIKER".to_string())
}
pub fn reward_amount(&self, signing_budget: &Coin) -> Coin {
let amount =
Uint128::new(signing_budget.amount) * self.ratio_signed * self.voting_power_ratio;
Coin::new(amount.u128(), &signing_budget.denom)
}
}
#[derive(Debug)]
pub struct EpochSigningResults {
pub blocks: i64,
pub total_voting_power_at_epoch_start: i64,
pub validators: Vec<ValidatorSigning>,
}
pub struct RawValidatorResult {
pub signed_blocks: i32,
pub voting_power: i64,
}
impl RawValidatorResult {
pub fn new(signed_blocks: i32, voting_power: i64) -> Self {
Self {
signed_blocks,
voting_power,
}
}
}
impl EpochSigningResults {
pub fn construct(
blocks: i64,
total_vp: i64,
validator_results: HashMap<models::Validator, RawValidatorResult>,
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, raw_results) in validator_results {
let vp: u64 = raw_results.voting_power.try_into().unwrap_or_default();
let signed: u64 = raw_results.signed_blocks.try_into().unwrap_or_default();
let voting_power_ratio = Decimal::from_ratio(vp, total_vp_u64);
debug_assert!(signed <= blocks_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: raw_results.voting_power,
voting_power_ratio,
signed_blocks: raw_results.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()
.inspect(|v| {
info!(
"validator {} will receive {} at address {} for block signing work",
v.moniker(),
v.reward_amount(budget),
v.operator_account
);
})
.map(|v| (v.operator_account.clone(), vec![v.reward_amount(budget)]))
.collect()
}
}
@@ -0,0 +1,58 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::config;
use crate::error::NymRewarderError;
use crate::rewarder::credential_issuance::monitor::CredentialIssuanceMonitor;
use crate::rewarder::credential_issuance::types::{CredentialIssuanceResults, MonitoringResults};
use crate::rewarder::epoch::Epoch;
use crate::rewarder::nyxd_client::NyxdClient;
use nym_task::TaskClient;
use tracing::info;
mod monitor;
pub mod types;
pub struct CredentialIssuance {
monitoring_results: MonitoringResults,
}
impl CredentialIssuance {
pub(crate) async fn new(
epoch: Epoch,
nyxd_client: &NyxdClient,
) -> Result<Self, NymRewarderError> {
Ok(CredentialIssuance {
monitoring_results: MonitoringResults::new_initial(epoch, nyxd_client).await?,
})
}
pub(crate) fn start_monitor(
&self,
monitor_config: config::IssuanceMonitor,
nyxd_client: NyxdClient,
task_client: TaskClient,
) {
let monitoring_results = self.monitoring_results.clone();
let mut monitor =
CredentialIssuanceMonitor::new(monitor_config, nyxd_client, monitoring_results);
tokio::spawn(async move { monitor.run(task_client).await });
}
pub(crate) async fn get_issued_credentials_results(
&self,
current_epoch: Epoch,
) -> Result<CredentialIssuanceResults, NymRewarderError> {
info!(
"looking up credential issuers for epoch {} ({} - {})",
current_epoch.id,
current_epoch.start_rfc3339(),
current_epoch.end_rfc3339()
);
let raw_results = self.monitoring_results.finish_epoch().await;
Ok(raw_results.into())
}
}
@@ -0,0 +1,340 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::config;
use crate::error::NymRewarderError;
use crate::rewarder::credential_issuance::types::{
CredentialIssuer, MonitoringResults, RawOperatorResult,
};
use crate::rewarder::helpers::api_client;
use crate::rewarder::nyxd_client::NyxdClient;
use bip39::rand::prelude::SliceRandom;
use bip39::rand::thread_rng;
use nym_coconut::{
hash_to_scalar, verify_partial_blind_signature, Base58, G1Projective, Parameters,
VerificationKey,
};
use nym_coconut_dkg_common::types::EpochId;
use nym_credentials::coconut::bandwidth::BandwidthVoucher;
use nym_task::TaskClient;
use nym_validator_client::nym_api::{IssuedCredential, IssuedCredentialBody, NymApiClientExt};
use nym_validator_client::nyxd::Hash;
use std::cmp::max;
use std::collections::HashMap;
use std::sync::OnceLock;
use tokio::time::interval;
use tracing::{debug, error, info, instrument, trace, warn};
pub(crate) fn bandwidth_voucher_params() -> &'static Parameters {
static BANDWIDTH_CREDENTIAL_PARAMS: OnceLock<Parameters> = OnceLock::new();
BANDWIDTH_CREDENTIAL_PARAMS.get_or_init(BandwidthVoucher::default_parameters)
}
pub struct CredentialIssuanceMonitor {
nyxd_client: NyxdClient,
monitoring_results: MonitoringResults,
config: config::IssuanceMonitor,
// map of validator address -> transaction hash -> issued credential
// (ideally we'd have hashed the AccountId directly, but it doesn't implement `Hash`)
seen_deposits: HashMap<String, HashMap<Hash, i64>>,
}
impl CredentialIssuanceMonitor {
pub fn new(
config: config::IssuanceMonitor,
nyxd_client: NyxdClient,
monitoring_results: MonitoringResults,
) -> CredentialIssuanceMonitor {
CredentialIssuanceMonitor {
config,
nyxd_client,
monitoring_results,
seen_deposits: Default::default(),
}
}
fn validate_credential_signature(
&mut self,
_issued_credential: &IssuedCredentialBody,
) -> Result<(), NymRewarderError> {
warn!("unimplemented: public key sharing mechanism");
// let plaintext = issued_credential.credential.signable_plaintext();
// if !operator_public_key.verify(&plaintext) {
// ...
// }
Ok(())
}
fn check_deposit_reuse(
&mut self,
runner: &str,
credential_id: i64,
deposit_tx: Hash,
) -> Result<bool, NymRewarderError> {
// check if we've seen this tx hash before
// TODO: we should persist them in the database in case we crash
if let Some(known_runner) = self.seen_deposits.get_mut(runner) {
if let Some(&used) = known_runner.get(&deposit_tx) {
return if used != credential_id {
Err(NymRewarderError::DuplicateDepositHash {
tx_hash: deposit_tx,
first: used,
other: credential_id,
})
} else {
debug!("we have already verified this credential before");
Ok(true)
};
} else {
known_runner.insert(deposit_tx, credential_id);
}
}
Ok(false)
}
async fn validate_deposit(
&mut self,
issued_credential: &IssuedCredentialBody,
) -> Result<(), NymRewarderError> {
// check if this deposit even exists
let deposit_tx = issued_credential.credential.tx_hash;
let (deposit_value, deposit_info) = self
.nyxd_client
.get_deposit_transaction_attributes(deposit_tx)
.await?;
trace!("deposit exists");
// check if the deposit values match
let credential_value = issued_credential.credential.public_attributes.first();
let credential_info = issued_credential.credential.public_attributes.get(1);
if credential_value != Some(&deposit_value) {
return Err(NymRewarderError::InconsistentDepositValue {
tx_hash: deposit_tx,
request: credential_value.cloned(),
on_chain: deposit_value,
});
}
trace!("credential values matches the deposit");
if credential_info != Some(&deposit_info) {
return Err(NymRewarderError::InconsistentDepositInfo {
tx_hash: deposit_tx,
request: credential_info.cloned(),
on_chain: deposit_info,
});
}
trace!("credential info matches the deposit");
Ok(())
}
fn verify_credential(
&mut self,
vk: &VerificationKey,
credential: IssuedCredential,
) -> Result<(), NymRewarderError> {
let public_attributes = credential
.public_attributes
.iter()
.map(hash_to_scalar)
.collect::<Vec<_>>();
#[allow(clippy::map_identity)]
let attributes_refs = public_attributes.iter().collect::<Vec<_>>();
let mut public_attribute_commitments =
Vec::with_capacity(credential.bs58_encoded_private_attributes_commitments.len());
for raw_cm in credential.bs58_encoded_private_attributes_commitments {
match G1Projective::try_from_bs58(&raw_cm) {
Ok(cm) => public_attribute_commitments.push(cm),
Err(source) => {
return Err(NymRewarderError::MalformedCredentialCommitment {
raw: raw_cm,
source,
})
}
}
}
// actually do verify the credential now
if !verify_partial_blind_signature(
bandwidth_voucher_params(),
&public_attribute_commitments,
&attributes_refs,
&credential.blinded_partial_credential,
vk,
) {
return Err(NymRewarderError::BlindVerificationFailure);
}
trace!("credential correctly verifies");
Ok(())
}
// TODO: currently we can't obtain public key of the runner in order to verify the signature
#[instrument(skip_all, fields(credential_id = credential_id, deposit = %issued_credential.credential.tx_hash))]
async fn validate_issued_credential(
&mut self,
runner: String,
credential_id: i64,
issued_credential: IssuedCredentialBody,
vk: &VerificationKey,
) -> Result<(), NymRewarderError> {
self.validate_credential_signature(&issued_credential)?;
let deposit_tx = issued_credential.credential.tx_hash;
// make sure the issuer is not using the same deposit for multiple credentials
let already_checked = self.check_deposit_reuse(&runner, credential_id, deposit_tx)?;
if already_checked {
return Ok(());
}
// check the correctness of the deposit itself
self.validate_deposit(&issued_credential).await?;
// check if the partial credential correctly verifies
self.verify_credential(vk, issued_credential.credential)?;
Ok(())
}
fn sample_credential_ids(&self, first_id: i64, total_issued: i64) -> Vec<i64> {
let credential_range: Vec<_> = (first_id..first_id + total_issued).collect();
let issued = credential_range.len();
let sampled = if issued <= self.config.min_validate_per_issuer {
credential_range
} else {
let mut rng = thread_rng();
let sample_size = max(
self.config.min_validate_per_issuer,
(issued as f64 * self.config.sampling_rate) as usize,
);
credential_range
.choose_multiple(&mut rng, sample_size)
.copied()
.collect::<Vec<_>>()
};
sampled
}
#[instrument(skip(self, issuer, epoch_id), fields(dkg_epoch = epoch_id, issuer = %issuer.operator_account, url = issuer.api_runner), err(Display))]
async fn check_issuer(
&mut self,
epoch_id: EpochId,
issuer: CredentialIssuer,
) -> Result<RawOperatorResult, NymRewarderError> {
info!("checking the issuer's credentials...");
debug!("checking the issuer's credentials...");
let api_client = api_client(&issuer)?;
let epoch_credentials = api_client.epoch_credentials(epoch_id).await?;
let Some(first_id) = epoch_credentials.first_epoch_credential_id else {
// no point in doing anything more - if they haven't issued anything, there's nothing to verify
debug!("no credentials issued this epoch",);
return Ok(RawOperatorResult::new_empty(
issuer.operator_account,
issuer.api_runner,
));
};
trace!("issued credentials: {epoch_credentials:?}");
let sampled = self.sample_credential_ids(first_id, epoch_credentials.total_issued as i64);
let request_size = sampled.len();
trace!("sampled credentials to validate: {sampled:?}");
let credentials = api_client.issued_credentials(sampled.clone()).await?;
if credentials.credentials.len() != request_size {
// TODO: we need some signatures here to actually show the validator is cheating
return Err(NymRewarderError::IncompleteRequest {
runner_account: issuer.operator_account,
requested: request_size,
received: credentials.credentials.len(),
});
}
for (id, credential) in credentials.credentials {
trace!("checking credential {id}...");
// TODO: insert the failure information, alongside the signature, to the evidence db
if let Err(err) = self
.validate_issued_credential(
issuer.operator_account.to_string(),
id,
credential,
&issuer.verification_key,
)
.await
{
error!(
"failed to verify credential {id} from {}!!: {err}",
issuer.operator_account
);
return Err(err);
}
}
Ok(RawOperatorResult {
operator_account: issuer.operator_account,
api_runner: issuer.api_runner,
issued_credentials: epoch_credentials.total_issued,
validated_credentials: sampled,
})
}
async fn check_issuers(&mut self) -> Result<(), NymRewarderError> {
info!("checking credential issuers");
let epoch = self.nyxd_client.dkg_epoch().await?;
let issuers = self
.nyxd_client
.get_credential_issuers(epoch.epoch_id)
.await?;
let mut results = Vec::with_capacity(issuers.len());
for issuer in issuers {
let address = issuer.operator_account.clone();
// we could parallelize it, but we're running the test so infrequently (relatively speaking)
// that doing it sequentially is fine
match self.check_issuer(epoch.epoch_id, issuer).await {
Ok(res) => results.push(res),
Err(err) => {
// TODO: insert info to the db
error!("failed to check credential issuance of {address}: {err}")
}
}
}
self.monitoring_results
.append_run_results(epoch.epoch_id as u32, results)
.await;
Ok(())
}
pub async fn run(&mut self, mut task_client: TaskClient) {
info!("starting");
let mut run_interval = interval(self.config.run_interval);
while !task_client.is_shutdown() {
tokio::select! {
biased;
_ = task_client.recv() => {
info!("received shutdown");
break
}
_ = run_interval.tick() => {
if let Err(err) = self.check_issuers().await {
error!("failed to perform credential issuance check: {err}")
}
}
}
}
}
}
@@ -0,0 +1,349 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::error::NymRewarderError;
use crate::rewarder::epoch::Epoch;
use crate::rewarder::helpers::api_client;
use crate::rewarder::nyxd_client::NyxdClient;
use cosmwasm_std::{Addr, Decimal, Uint128};
use nym_coconut::{Base58, VerificationKey};
use nym_coconut_dkg_common::verification_key::ContractVKShare;
use nym_validator_client::nym_api::NymApiClientExt;
use nym_validator_client::nyxd::{AccountId, Coin};
use std::collections::{HashMap, HashSet};
use std::mem;
use std::sync::Arc;
use tokio::sync::Mutex;
use tracing::{info, warn};
#[derive(Clone)]
pub struct MonitoringResults {
inner: Arc<Mutex<MonitoringResultsInner>>,
}
impl MonitoringResults {
pub(crate) async fn new_initial(
initial_epoch: Epoch,
nyxd_client: &NyxdClient,
) -> Result<Self, NymRewarderError> {
let epoch = nyxd_client.dkg_epoch().await?;
let issuers = nyxd_client.get_credential_issuers(epoch.epoch_id).await?;
let mut initial_results = HashMap::new();
for issuer in issuers {
let issuer_account = issuer.operator_account.to_string();
let mut raw_issuer = RawOperatorIssuing {
api_runner: issuer.api_runner.clone(),
runner_account: issuer.operator_account.clone(),
per_epoch: Default::default(),
};
let Ok(api_client) = api_client(&issuer) else {
warn!("failed to create api client for {issuer_account}");
initial_results.insert(issuer_account, raw_issuer);
continue;
};
let Ok(epoch_credentials) = api_client.epoch_credentials(epoch.epoch_id).await else {
warn!("failed to get initial epoch credentials from {issuer_account}");
initial_results.insert(issuer_account, raw_issuer);
continue;
};
raw_issuer.per_epoch.insert(
epoch.epoch_id as u32,
IssuedEpochCredentials {
issued_since_monitor_started: 0,
validated_ids: Default::default(),
last_total_issued: epoch_credentials.total_issued,
},
);
initial_results.insert(issuer_account, raw_issuer);
}
Ok(MonitoringResults {
inner: Arc::new(Mutex::new(MonitoringResultsInner::new(
initial_epoch,
epoch.epoch_id as u32,
initial_results,
))),
})
}
pub(crate) async fn append_run_results(&self, dkg_epoch: u32, results: Vec<RawOperatorResult>) {
let mut guard = self.inner.lock().await;
for result in results {
let Some(entry) = guard.operators.get_mut(result.operator_account.as_ref()) else {
// if this is the first time we're seeing this data, make sure to set the current results as the starting point
guard.operators.insert(
result.operator_account.to_string(),
RawOperatorIssuing::new_empty(dkg_epoch, result),
);
continue;
};
let Some(epoch_data) = entry.per_epoch.get_mut(&dkg_epoch) else {
// similar situation to the above, if we don't have the proper initial data, set it to what we got now
entry
.per_epoch
.insert(dkg_epoch, IssuedEpochCredentials::new_initial(&result));
continue;
};
let issued = result.issued_credentials - epoch_data.last_total_issued;
epoch_data.last_total_issued = result.issued_credentials;
for validated in result.validated_credentials {
epoch_data.validated_ids.insert(validated);
}
epoch_data.issued_since_monitor_started += issued;
}
}
pub(crate) async fn finish_epoch(&self) -> MonitoringResultsInner {
let mut guard = self.inner.lock().await;
let next_epoch = guard.epoch.next();
// safety: whenever the monitor results are constructed, we always have at least a single value there
#[allow(clippy::unwrap_used)]
let latest_dkg_epoch = guard.dkg_epochs.pop().unwrap();
// only keep results from the latest dkg epoch (but after resetting the counters)
let mut to_keep = HashMap::new();
for (runner, result) in &guard.operators {
let mut kept_epoch = HashMap::new();
if let Some(data) = result.per_epoch.get(&latest_dkg_epoch) {
kept_epoch.insert(
latest_dkg_epoch,
IssuedEpochCredentials {
issued_since_monitor_started: 0,
validated_ids: Default::default(),
last_total_issued: data.last_total_issued,
},
);
}
to_keep.insert(
runner.clone(),
RawOperatorIssuing {
api_runner: result.api_runner.clone(),
runner_account: result.runner_account.clone(),
per_epoch: kept_epoch,
},
);
}
let next_results = MonitoringResultsInner {
epoch: next_epoch,
dkg_epochs: vec![latest_dkg_epoch],
operators: to_keep,
};
mem::replace(&mut guard, next_results)
}
}
pub(crate) struct MonitoringResultsInner {
pub(crate) epoch: Epoch,
pub(crate) dkg_epochs: Vec<u32>,
// map from operator's account to their results
pub(crate) operators: HashMap<String, RawOperatorIssuing>,
}
impl From<MonitoringResultsInner> for CredentialIssuanceResults {
fn from(value: MonitoringResultsInner) -> Self {
let total_issued = value
.operators
.values()
.map(|o| {
let operator_issued: u32 = o
.per_epoch
.values()
.map(|e| e.issued_since_monitor_started)
.sum();
operator_issued
})
.sum();
CredentialIssuanceResults {
total_issued_partial_credentials: total_issued,
dkg_epochs: value.dkg_epochs,
api_runners: value
.operators
.into_values()
.map(|runner| {
let issued_ratio = if total_issued == 0 {
Decimal::zero()
} else {
Decimal::from_ratio(runner.issued_credentials(), total_issued)
};
OperatorIssuing {
issued_ratio,
issued_credentials: runner.issued_credentials(),
validated_credentials: runner.validated_credentials(),
api_runner: runner.api_runner,
runner_account: runner.runner_account,
}
})
.collect(),
}
}
}
impl MonitoringResultsInner {
fn new(
epoch: Epoch,
initial_dkg_epoch: u32,
initial_operators: HashMap<String, RawOperatorIssuing>,
) -> MonitoringResultsInner {
MonitoringResultsInner {
epoch,
dkg_epochs: vec![initial_dkg_epoch],
operators: initial_operators,
}
}
}
pub(crate) struct RawOperatorResult {
pub(crate) operator_account: AccountId,
pub(crate) api_runner: String,
// how many credentials the operator claims to have issued in **TOTAL** in this **DKG** epoch
pub(crate) issued_credentials: u32,
pub(crate) validated_credentials: Vec<i64>,
}
impl RawOperatorResult {
pub(crate) fn new_empty(operator_account: AccountId, api_runner: String) -> RawOperatorResult {
RawOperatorResult {
operator_account,
api_runner,
issued_credentials: 0,
validated_credentials: Default::default(),
}
}
}
pub struct RawOperatorIssuing {
pub api_runner: String,
pub runner_account: AccountId,
pub per_epoch: HashMap<u32, IssuedEpochCredentials>,
}
impl RawOperatorIssuing {
pub fn new_empty(epoch: u32, raw_result: RawOperatorResult) -> RawOperatorIssuing {
let mut per_epoch = HashMap::new();
per_epoch.insert(epoch, IssuedEpochCredentials::new_initial(&raw_result));
RawOperatorIssuing {
api_runner: raw_result.api_runner,
runner_account: raw_result.operator_account,
per_epoch,
}
}
pub fn issued_credentials(&self) -> u32 {
self.per_epoch
.values()
.map(|e| e.issued_since_monitor_started)
.sum()
}
pub fn validated_credentials(&self) -> u32 {
let ids: usize = self.per_epoch.values().map(|e| e.validated_ids.len()).sum();
ids as u32
}
}
pub struct IssuedEpochCredentials {
pub issued_since_monitor_started: u32,
pub validated_ids: HashSet<i64>,
pub last_total_issued: u32,
}
impl IssuedEpochCredentials {
pub fn new_initial(raw: &RawOperatorResult) -> Self {
IssuedEpochCredentials {
issued_since_monitor_started: 0,
validated_ids: raw.validated_credentials.iter().copied().collect(),
last_total_issued: raw.issued_credentials,
}
}
}
pub struct OperatorIssuing {
pub api_runner: String,
pub runner_account: AccountId,
pub issued_ratio: Decimal,
pub issued_credentials: u32,
pub validated_credentials: u32,
}
impl OperatorIssuing {
pub fn reward_amount(&self, signing_budget: &Coin) -> Coin {
let amount = Uint128::new(signing_budget.amount) * self.issued_ratio;
Coin::new(amount.u128(), &signing_budget.denom)
}
}
pub struct CredentialIssuanceResults {
pub total_issued_partial_credentials: u32,
pub dkg_epochs: Vec<u32>,
pub api_runners: Vec<OperatorIssuing>,
}
impl CredentialIssuanceResults {
pub fn rewarding_amounts(&self, budget: &Coin) -> Vec<(AccountId, Vec<Coin>)> {
self.api_runners
.iter()
.inspect(|a| {
info!(
"operator {} will receive {} at address {} for credential issuance work",
a.api_runner,
a.reward_amount(budget),
a.runner_account,
);
})
.map(|v| (v.runner_account.clone(), vec![v.reward_amount(budget)]))
.collect()
}
}
#[derive(Debug)]
pub struct CredentialIssuer {
// pub public_key: identity::PublicKey,
pub operator_account: AccountId,
pub api_runner: String,
pub verification_key: VerificationKey,
}
impl TryFrom<ContractVKShare> for CredentialIssuer {
type Error = NymRewarderError;
fn try_from(value: ContractVKShare) -> Result<Self, Self::Error> {
Ok(CredentialIssuer {
operator_account: addr_to_account_id(value.owner.clone()),
api_runner: value.announce_address,
verification_key: VerificationKey::try_from_bs58(value.share).map_err(|source| {
NymRewarderError::MalformedPartialVerificationKey {
runner: value.owner.to_string(),
source,
}
})?,
})
}
}
// safety: we're converting between different wrappers for bech32 addresses
// and we trust (reasonably so), the values coming out of registered dealers in the DKG contract
fn addr_to_account_id(addr: Addr) -> AccountId {
#[allow(clippy::unwrap_used)]
addr.as_str().parse().unwrap()
}
@@ -0,0 +1,61 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::error::NymRewarderError;
use sqlx::FromRow;
use std::ops::Add;
use std::time::Duration;
use time::format_description::well_known::Rfc3339;
use time::OffsetDateTime;
const HOUR: Duration = Duration::from_secs(60 * 60);
#[derive(Debug, Clone, Copy, FromRow)]
pub struct Epoch {
pub id: i64,
pub start_time: OffsetDateTime,
pub end_time: OffsetDateTime,
}
impl Epoch {
pub fn first(epoch_duration: Duration) -> Result<Self, NymRewarderError> {
let start = OffsetDateTime::now_utc()
.add(HOUR)
.replace_nanosecond(0)?
.replace_microsecond(0)?
.replace_second(0)?;
Ok(Epoch {
id: 0,
start_time: start,
end_time: start + epoch_duration,
})
}
pub fn until_end(&self) -> Duration {
let now = OffsetDateTime::now_utc();
(self.end_time - now).try_into().unwrap_or_default()
}
pub fn next(&self) -> Self {
let duration = self.end_time - self.start_time;
Epoch {
id: self.id + 1,
start_time: self.end_time,
end_time: self.end_time + duration,
}
}
pub fn start_rfc3339(&self) -> String {
// safety: unwrap here is fine as we're using a predefined formatter
#[allow(clippy::unwrap_used)]
self.start_time.format(&Rfc3339).unwrap()
}
pub fn end_rfc3339(&self) -> String {
// safety: unwrap here is fine as we're using a predefined formatter
#[allow(clippy::unwrap_used)]
self.end_time.format(&Rfc3339).unwrap()
}
}
@@ -0,0 +1,50 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::error::NymRewarderError;
use crate::rewarder::credential_issuance::types::CredentialIssuer;
use nym_validator_client::nym_api;
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_pubkey_to_address(
pubkey: PublicKey,
) -> Result<AccountId, NymRewarderError> {
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,
}
})
}
pub(crate) fn api_client(issuer: &CredentialIssuer) -> Result<nym_api::Client, NymRewarderError> {
let url = match issuer.api_runner.parse() {
Ok(url) => url,
Err(source) => {
return Err(NymRewarderError::MalformedApiUrl {
raw: issuer.api_runner.clone(),
runner_account: issuer.operator_account.clone(),
source,
})
}
};
Ok(nym_api::Client::new(url, None))
}
+287
View File
@@ -0,0 +1,287 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::config::Config;
use crate::error::{InsufficientBalance, NymRewarderError};
use crate::rewarder::block_signing::types::EpochSigningResults;
use crate::rewarder::block_signing::EpochSigning;
use crate::rewarder::credential_issuance::types::CredentialIssuanceResults;
use crate::rewarder::credential_issuance::CredentialIssuance;
use crate::rewarder::epoch::Epoch;
use crate::rewarder::nyxd_client::NyxdClient;
use crate::rewarder::storage::RewarderStorage;
use nym_task::TaskManager;
use nym_validator_client::nyxd::{AccountId, Coin, Hash};
use nyxd_scraper::NyxdScraper;
use std::ops::Add;
use tokio::pin;
use tokio::time::{interval_at, Instant};
use tracing::{error, info, instrument};
mod block_signing;
mod credential_issuance;
mod epoch;
mod helpers;
mod nyxd_client;
mod storage;
mod tasks;
pub struct EpochRewards {
pub epoch: Epoch,
pub signing: Option<EpochSigningResults>,
pub credentials: Option<CredentialIssuanceResults>,
pub total_budget: Coin,
pub signing_budget: Coin,
pub credentials_budget: Coin,
}
impl EpochRewards {
pub fn amounts(&self) -> Vec<(AccountId, Vec<Coin>)> {
let mut amounts = Vec::new();
if let Some(signing) = &self.signing {
for signing_amount in signing.rewarding_amounts(&self.signing_budget) {
if signing_amount.1[0].amount != 0 {
amounts.push(signing_amount)
}
}
}
if let Some(credentials) = &self.credentials {
for credential_amount in credentials.rewarding_amounts(&self.credentials_budget) {
if credential_amount.1[0].amount != 0 {
amounts.push(credential_amount)
}
}
}
amounts
}
}
pub fn total_spent(amounts: &[(AccountId, Vec<Coin>)], denom: &str) -> Coin {
let amount = amounts.iter().map(|(_, amount)| amount[0].amount).sum();
Coin::new(amount, denom)
}
pub struct Rewarder {
config: Config,
current_epoch: Epoch,
storage: RewarderStorage,
nyxd_client: NyxdClient,
epoch_signing: Option<EpochSigning>,
credential_issuance: Option<CredentialIssuance>,
}
impl Rewarder {
pub async fn new(config: Config) -> Result<Self, NymRewarderError> {
let nyxd_client = NyxdClient::new(&config)?;
let storage = RewarderStorage::init(&config.storage_paths.reward_history).await?;
let current_epoch = if let Some(last_epoch) = storage.load_last_rewarding_epoch().await? {
last_epoch.next()
} else {
Epoch::first(config.rewarding.epoch_duration)?
};
let epoch_signing = if config.block_signing.enabled {
let nyxd_scraper = NyxdScraper::new(config.scraper_config()).await?;
Some(EpochSigning {
nyxd_scraper,
nyxd_client: nyxd_client.clone(),
})
} else {
None
};
let credential_issuance = if config.issuance_monitor.enabled {
Some(CredentialIssuance::new(current_epoch, &nyxd_client).await?)
} else {
None
};
if config.issuance_monitor.enabled || config.block_signing.enabled {
let balance = nyxd_client
.balance(&config.rewarding.epoch_budget.denom)
.await?;
let minimum = Coin::new(
config.rewarding.epoch_budget.amount * 100,
&config.rewarding.epoch_budget.denom,
);
if balance.amount < minimum.amount {
return Err(NymRewarderError::InsufficientRewarderBalance(Box::new(
InsufficientBalance {
epoch_budget: config.rewarding.epoch_budget.clone(),
balance,
minimum,
},
)));
}
}
Ok(Rewarder {
current_epoch,
credential_issuance,
epoch_signing,
nyxd_client,
storage,
config,
})
}
#[instrument(skip(self))]
async fn calculate_block_signing_rewards(
&mut self,
) -> Result<Option<EpochSigningResults>, NymRewarderError> {
info!("calculating reward shares");
if let Some(epoch_signing) = &mut self.epoch_signing {
Some(
epoch_signing
.get_signed_blocks_results(self.current_epoch)
.await,
)
} else {
None
}
.transpose()
}
#[instrument(skip(self))]
async fn calculate_credential_rewards(
&mut self,
) -> Result<Option<CredentialIssuanceResults>, NymRewarderError> {
info!("calculating reward shares");
if let Some(credential_issuance) = &mut self.credential_issuance {
Some(
credential_issuance
.get_issued_credentials_results(self.current_epoch)
.await,
)
} else {
None
}
.transpose()
}
async fn determine_epoch_rewards(&mut self) -> Result<EpochRewards, NymRewarderError> {
let epoch_budget = self.config.rewarding.epoch_budget.clone();
let denom = &epoch_budget.denom;
let signing_budget = Coin::new(
(self.config.rewarding.ratios.block_signing * epoch_budget.amount as f64) as u128,
denom,
);
let credentials_budget = Coin::new(
(self.config.rewarding.ratios.credential_issuance * epoch_budget.amount as f64) as u128,
denom,
);
let signing_rewards = self.calculate_block_signing_rewards().await?;
let credential_rewards = self.calculate_credential_rewards().await?;
Ok(EpochRewards {
epoch: self.current_epoch,
signing: signing_rewards,
credentials: credential_rewards,
total_budget: epoch_budget.clone(),
signing_budget,
credentials_budget,
})
}
#[instrument(skip(self))]
async fn send_rewards(
&self,
amounts: Vec<(AccountId, Vec<Coin>)>,
) -> Result<Hash, NymRewarderError> {
info!("sending rewards");
self.nyxd_client
.send_rewards(self.current_epoch, amounts)
.await
}
async fn handle_epoch_end(&mut self) {
info!("handling the epoch end");
let rewards = match self.determine_epoch_rewards().await {
Ok(rewards) => rewards,
Err(err) => {
error!("failed to determine epoch rewards: {err}");
return;
}
};
let rewarding_amounts = rewards.amounts();
let total_spent = total_spent(
&rewarding_amounts,
&self.config.rewarding.epoch_budget.denom,
);
let rewarding_result = self.send_rewards(rewarding_amounts).await;
if let Err(err) = self
.storage
.save_rewarding_information(rewards, total_spent, rewarding_result)
.await
{
error!("failed to persist rewarding information: {err}")
}
self.current_epoch = self.current_epoch.next();
}
pub async fn run(mut self) -> Result<(), NymRewarderError> {
info!("Starting nym validators rewarder");
// setup shutdowns
let mut task_manager = TaskManager::new(5);
if let Some(ref credential_issuance) = self.credential_issuance {
credential_issuance.start_monitor(
self.config.issuance_monitor,
self.nyxd_client.clone(),
task_manager.subscribe(),
);
}
if let Some(epoch_signing) = &self.epoch_signing {
epoch_signing.nyxd_scraper.start().await?;
epoch_signing.nyxd_scraper.wait_for_startup_sync().await;
}
let until_end = self.current_epoch.until_end();
info!(
"the first epoch will finish in {} secs",
until_end.as_secs()
);
let mut epoch_ticker = interval_at(
Instant::now().add(until_end),
self.config.rewarding.epoch_duration,
);
let shutdown_future = task_manager.catch_interrupt();
pin!(shutdown_future);
loop {
tokio::select! {
biased;
interrupt_res = &mut shutdown_future => {
info!("received interrupt");
if let Err(err) = interrupt_res {
error!("runtime interrupt failure: {err}")
}
break;
}
_ = epoch_ticker.tick() => self.handle_epoch_end().await
}
}
if let Some(epoch_signing) = self.epoch_signing {
epoch_signing.nyxd_scraper.stop().await;
}
Ok(())
}
}
@@ -0,0 +1,123 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::config::Config;
use crate::error::NymRewarderError;
use crate::rewarder::credential_issuance::types::CredentialIssuer;
use nym_coconut_bandwidth_contract_common::events::{
COSMWASM_DEPOSITED_FUNDS_EVENT_TYPE, DEPOSIT_INFO, DEPOSIT_VALUE,
};
use nym_coconut_dkg_common::types::Epoch;
use nym_network_defaults::NymNetworkDetails;
use nym_validator_client::nyxd::contract_traits::{DkgQueryClient, PagedDkgQueryClient};
use nym_validator_client::nyxd::helpers::find_tx_attribute;
use nym_validator_client::nyxd::module_traits::staking::{
QueryHistoricalInfoResponse, QueryValidatorsResponse,
};
use nym_validator_client::nyxd::{
AccountId, Coin, CosmWasmClient, Hash, PageRequest, StakingQueryClient,
};
use nym_validator_client::{nyxd, DirectSigningHttpRpcNyxdClient};
use std::ops::Deref;
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) -> Result<Self, NymRewarderError> {
let client_config =
nyxd::Config::try_from_nym_network_details(&NymNetworkDetails::new_from_env())?;
let nyxd_url = config.base.upstream_nyxd.as_str();
let mnemonic = config.base.mnemonic.clone();
let inner = DirectSigningHttpRpcNyxdClient::connect_with_mnemonic(
client_config,
nyxd_url,
mnemonic,
)?;
Ok(NyxdClient {
inner: Arc::new(RwLock::new(inner)),
})
}
pub(crate) async fn balance(&self, denom: &str) -> Result<Coin, NymRewarderError> {
let guard = self.inner.read().await;
let address = guard.address();
Ok(guard
.get_balance(&address, denom.to_string())
.await?
.unwrap_or(Coin::new(0, denom)))
}
pub(crate) async fn send_rewards(
&self,
epoch: crate::rewarder::Epoch,
amounts: Vec<(AccountId, Vec<Coin>)>,
) -> Result<Hash, NymRewarderError> {
self.inner
.write()
.await
.send_multiple(amounts, format!("sending rewards for {epoch:?}"), None)
.await
.map(|res| res.hash)
.map_err(Into::into)
}
pub(crate) async fn historical_info(
&self,
height: i64,
) -> Result<QueryHistoricalInfoResponse, NymRewarderError> {
Ok(self.inner.read().await.historical_info(height).await?)
}
pub(crate) async fn validators(
&self,
pagination: Option<PageRequest>,
) -> Result<QueryValidatorsResponse, NymRewarderError> {
let guard = self.inner.read().await;
Ok(StakingQueryClient::validators(guard.deref(), "".to_string(), pagination).await?)
}
pub(crate) async fn dkg_epoch(&self) -> Result<Epoch, NymRewarderError> {
Ok(self.inner.read().await.get_current_epoch().await?)
}
pub(crate) async fn get_credential_issuers(
&self,
dkg_epoch: u64,
) -> Result<Vec<CredentialIssuer>, NymRewarderError> {
self.inner
.read()
.await
.get_all_verification_key_shares(dkg_epoch)
.await?
.into_iter()
.map(TryInto::try_into)
.collect()
}
pub(crate) async fn get_deposit_transaction_attributes(
&self,
tx_hash: Hash,
) -> Result<(String, String), NymRewarderError> {
let tx = self.inner.read().await.get_tx(tx_hash).await?;
// todo: we need to make it more concrete that the first attribute is the deposit value
// and the second one is the deposit info
let deposit_value =
find_tx_attribute(&tx, COSMWASM_DEPOSITED_FUNDS_EVENT_TYPE, DEPOSIT_VALUE)
.ok_or(NymRewarderError::DepositValueNotFound { tx_hash })?;
let deposit_info =
find_tx_attribute(&tx, COSMWASM_DEPOSITED_FUNDS_EVENT_TYPE, DEPOSIT_INFO)
.ok_or(NymRewarderError::DepositInfoNotFound { tx_hash })?;
Ok((deposit_value, deposit_info))
}
}
@@ -0,0 +1,178 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::rewarder::epoch::Epoch;
#[derive(Clone)]
pub(crate) struct StorageManager {
pub(crate) connection_pool: sqlx::SqlitePool,
}
impl StorageManager {
pub(crate) async fn load_last_rewarding_epoch(&self) -> Result<Option<Epoch>, sqlx::Error> {
sqlx::query_as(
r#"
SELECT id, start_time, end_time
FROM rewarding_epoch
ORDER BY id DESC
LIMIT 1
"#,
)
.fetch_optional(&self.connection_pool)
.await
}
pub(crate) async fn insert_rewarding_epoch(
&self,
epoch: Epoch,
rewarding_budget: String,
total_spent: String,
rewarding_tx: Option<String>,
rewarding_error: Option<String>,
) -> Result<(), sqlx::Error> {
sqlx::query!(
r#"
INSERT INTO rewarding_epoch (id, start_time, end_time, budget, spent, rewarding_tx, rewarding_error)
VALUES (?, ?, ? ,?, ?, ?, ?)
"#,
epoch.id,
epoch.start_time,
epoch.end_time,
rewarding_budget,
total_spent: String,
rewarding_tx,
rewarding_error
).execute(&self.connection_pool).await?;
Ok(())
}
pub(crate) async fn insert_rewarding_epoch_block_signing(
&self,
epoch: i64,
total_voting_power_at_epoch_start: i64,
num_blocks: i64,
budget: String,
) -> Result<(), sqlx::Error> {
sqlx::query!(
r#"
INSERT INTO epoch_block_signing (rewarding_epoch_id, total_voting_power_at_epoch_start, num_blocks, budget)
VALUES (?, ?, ?, ?)
"#,
epoch,
total_voting_power_at_epoch_start,
num_blocks,
budget,
).execute(&self.connection_pool).await?;
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn insert_rewarding_epoch_block_signing_reward(
&self,
epoch: i64,
consensus_address: String,
operator_account: String,
amount: String,
voting_power: i64,
voting_power_share: String,
signed_blocks: i32,
signed_blocks_percent: String,
) -> Result<(), sqlx::Error> {
sqlx::query!(
r#"
INSERT INTO block_signing_reward (
rewarding_epoch_id,
validator_consensus_address,
operator_account,
amount,
voting_power,
voting_power_share,
signed_blocks,
signed_blocks_percent
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
"#,
epoch,
consensus_address,
operator_account,
amount,
voting_power,
voting_power_share,
signed_blocks,
signed_blocks_percent,
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
pub(crate) async fn insert_rewarding_epoch_credential_issuance(
&self,
epoch: i64,
starting_dkg_epoch: u32,
ending_dkg_epoch: u32,
total_issued_partial_credentials: u32,
budget: String,
) -> Result<(), sqlx::Error> {
sqlx::query!(
r#"
INSERT INTO epoch_credential_issuance (
rewarding_epoch_id,
starting_dkg_epoch,
ending_dkg_epoch,
total_issued_partial_credentials,
budget
)
VALUES (?, ?, ?, ?, ?)
"#,
epoch,
starting_dkg_epoch,
ending_dkg_epoch,
total_issued_partial_credentials,
budget,
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn insert_rewarding_epoch_credential_issuance_reward(
&self,
epoch: i64,
operator_account: String,
amount: String,
api_endpoint: String,
issued_partial_credentials: u32,
issued_credentials_share: String,
validated_issued_credentials: u32,
) -> Result<(), sqlx::Error> {
sqlx::query!(
r#"
INSERT INTO credential_issuance_reward (
rewarding_epoch_id,
operator_account,
amount,
api_endpoint,
issued_partial_credentials,
issued_credentials_share,
validated_issued_credentials
) VALUES (?, ?, ?, ?, ?, ?, ?)
"#,
epoch,
operator_account,
amount,
api_endpoint,
issued_partial_credentials,
issued_credentials_share,
validated_issued_credentials,
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
}
@@ -0,0 +1,167 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::error::NymRewarderError;
use crate::rewarder::epoch::Epoch;
use crate::rewarder::storage::manager::StorageManager;
use crate::rewarder::EpochRewards;
use nym_validator_client::nyxd::{Coin, Hash};
use sqlx::ConnectOptions;
use std::fmt::Debug;
use std::path::Path;
use tracing::{error, info, instrument};
mod manager;
pub struct RewarderStorage {
pub(crate) manager: StorageManager,
}
impl RewarderStorage {
#[instrument]
pub async fn init<P: AsRef<Path> + Debug>(database_path: P) -> Result<Self, NymRewarderError> {
let mut opts = sqlx::sqlite::SqliteConnectOptions::new()
.filename(database_path)
.create_if_missing(true);
// TODO: do we want auto_vacuum ?
opts.disable_statement_logging();
let connection_pool = match sqlx::SqlitePool::connect_with(opts).await {
Ok(db) => db,
Err(err) => {
error!("Failed to connect to SQLx database: {err}");
return Err(err.into());
}
};
if let Err(err) = sqlx::migrate!("./migrations").run(&connection_pool).await {
error!("Failed to initialize SQLx database: {err}");
return Err(err.into());
}
info!("Database migration finished!");
let manager = StorageManager { connection_pool };
let storage = RewarderStorage { manager };
Ok(storage)
}
pub(crate) async fn load_last_rewarding_epoch(
&self,
) -> Result<Option<Epoch>, NymRewarderError> {
Ok(self.manager.load_last_rewarding_epoch().await?)
}
pub(crate) async fn save_rewarding_information(
&self,
reward: EpochRewards,
total_spent: Coin,
rewarding_tx: Result<Hash, NymRewarderError>,
) -> Result<(), NymRewarderError> {
info!("persisting reward details");
let (reward_tx, reward_err) = match rewarding_tx {
Ok(hash) => (Some(hash.to_string()), None),
Err(err) => (None, Some(err.to_string())),
};
let epoch_id = reward.epoch.id;
self.manager
.insert_rewarding_epoch(
reward.epoch,
reward.total_budget.to_string(),
total_spent.to_string(),
reward_tx,
reward_err,
)
.await?;
self.manager
.insert_rewarding_epoch_block_signing(
epoch_id,
reward
.signing
.as_ref()
.map(|s| s.total_voting_power_at_epoch_start)
.unwrap_or_default(),
reward
.signing
.as_ref()
.map(|s| s.blocks)
.unwrap_or_default(),
reward.signing_budget.to_string(),
)
.await?;
if let Some(signing) = reward.signing {
for validator in signing.validators {
let reward_amount = validator.reward_amount(&reward.signing_budget).to_string();
self.manager
.insert_rewarding_epoch_block_signing_reward(
epoch_id,
validator.validator.consensus_address,
validator.operator_account.to_string(),
reward_amount,
validator.voting_power_at_epoch_start,
validator.voting_power_ratio.to_string(),
validator.signed_blocks,
validator.ratio_signed.to_string(),
)
.await?;
}
}
// safety: we must have at least a single value here
#[allow(clippy::unwrap_used)]
let dkg_epoch_start = reward
.credentials
.as_ref()
.and_then(|c| c.dkg_epochs.first().copied())
.unwrap_or_default();
#[allow(clippy::unwrap_used)]
let dkg_epoch_end = reward
.credentials
.as_ref()
.and_then(|c| c.dkg_epochs.last().copied())
.unwrap_or_default();
self.manager
.insert_rewarding_epoch_credential_issuance(
epoch_id,
dkg_epoch_start,
dkg_epoch_end,
reward
.credentials
.as_ref()
.map(|c| c.total_issued_partial_credentials)
.unwrap_or_default(),
reward.credentials_budget.to_string(),
)
.await?;
if let Some(credentials) = reward.credentials {
for api_runner in credentials.api_runners {
let reward_amount = api_runner
.reward_amount(&reward.credentials_budget)
.to_string();
self.manager
.insert_rewarding_epoch_credential_issuance_reward(
epoch_id,
api_runner.runner_account.to_string(),
reward_amount,
api_runner.api_runner,
api_runner.issued_credentials,
api_runner.issued_ratio.to_string(),
api_runner.validated_credentials,
)
.await?;
}
}
Ok(())
}
}
@@ -0,0 +1,2 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
+35 -6
View File
@@ -808,6 +808,16 @@ dependencies = [
"tendermint-proto",
]
[[package]]
name = "cosmos-sdk-proto"
version = "0.20.0"
source = "git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features#67718aa27a96efb9881e927a8f998c94b885093c"
dependencies = [
"prost",
"prost-types",
"tendermint-proto",
]
[[package]]
name = "cosmrs"
version = "0.15.0"
@@ -815,7 +825,26 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "47126f5364df9387b9d8559dcef62e99010e1d4098f39eb3f7ee4b5c254e40ea"
dependencies = [
"bip32",
"cosmos-sdk-proto",
"cosmos-sdk-proto 0.20.0 (registry+https://github.com/rust-lang/crates.io-index)",
"ecdsa 0.16.8",
"eyre",
"k256 0.13.1",
"rand_core 0.6.4",
"serde",
"serde_json",
"signature 2.1.0",
"subtle-encoding",
"tendermint",
"thiserror",
]
[[package]]
name = "cosmrs"
version = "0.15.0"
source = "git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features#67718aa27a96efb9881e927a8f998c94b885093c"
dependencies = [
"bip32",
"cosmos-sdk-proto 0.20.0 (git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features)",
"ecdsa 0.16.8",
"eyre",
"k256 0.13.1",
@@ -3127,7 +3156,7 @@ name = "nym-api-requests"
version = "0.1.0"
dependencies = [
"bs58 0.4.0",
"cosmrs",
"cosmrs 0.15.0 (git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features)",
"cosmwasm-std",
"getset",
"nym-coconut-interface",
@@ -3422,7 +3451,7 @@ name = "nym-types"
version = "1.0.0"
dependencies = [
"base64 0.21.4",
"cosmrs",
"cosmrs 0.15.0 (git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features)",
"cosmwasm-std",
"eyre",
"hmac 0.12.1",
@@ -3455,7 +3484,7 @@ dependencies = [
"bip32",
"bip39",
"colored 2.0.4",
"cosmrs",
"cosmrs 0.15.0 (git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features)",
"cosmwasm-std",
"cw-controllers",
"cw-utils",
@@ -3527,7 +3556,7 @@ dependencies = [
name = "nym-wallet-types"
version = "1.0.0"
dependencies = [
"cosmrs",
"cosmrs 0.15.0 (registry+https://github.com/rust-lang/crates.io-index)",
"cosmwasm-std",
"hex-literal",
"nym-config",
@@ -3564,7 +3593,7 @@ dependencies = [
"bip39",
"cfg-if",
"colored 2.0.4",
"cosmrs",
"cosmrs 0.15.0 (git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features)",
"cosmwasm-std",
"dirs 4.0.0",
"dotenvy",
@@ -35,6 +35,7 @@ pub(crate) fn validators() -> Vec<ValidatorDetails> {
vec![ValidatorDetails::new(
"https://qa-validator.qa.nymte.ch/",
Some("https://qa-nym-api.qa.nymte.ch/api"),
Some("wss://qa-validator.qa.nymte.ch/websocket"),
)]
}
@@ -32,6 +32,7 @@ pub(crate) fn validators() -> Vec<ValidatorDetails> {
vec![ValidatorDetails::new(
"https://rpc.sandbox.nymtech.net",
Some("https://sandbox-nym-api1.nymtech.net/api"),
Some("wss://rpc.sandbox.nymtech.net/websocket"),
)]
}
+1 -1
View File
@@ -49,7 +49,7 @@ base64 = "0.13"
zeroize = { version = "1.5", features = ["zeroize_derive", "serde"] }
cosmwasm-std = "1.3.0"
cosmrs = "=0.15.0"
cosmrs = { git = "https://github.com/jstuczyn/cosmos-rust", branch ="nym-temp/all-validator-features" }
nym-validator-client = { path = "../../common/client-libs/validator-client" }
nym-crypto = { path = "../../common/crypto", features = ["asymmetric"] }
@@ -35,7 +35,7 @@ impl NetworkStatisticsAPI {
pub async fn run(self) {
let rocket_shutdown_handle = self.rocket.shutdown();
let shutdown = TaskManager::new(10);
let mut shutdown = TaskManager::new(10);
tokio::spawn(self.rocket.launch());
shutdown.catch_interrupt().await.ok();