diff --git a/Cargo.lock b/Cargo.lock index 04b9c19eb7..eb1e717ee6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7508,6 +7508,7 @@ dependencies = [ "nyxd-scraper", "serde", "sha2 0.10.8", + "sqlx", "thiserror", "time", "tokio", diff --git a/common/nyxd-scraper/src/storage/mod.rs b/common/nyxd-scraper/src/storage/mod.rs index 2722a8558d..6dba07ae0e 100644 --- a/common/nyxd-scraper/src/storage/mod.rs +++ b/common/nyxd-scraper/src/storage/mod.rs @@ -31,8 +31,6 @@ pub struct ScraperStorage { impl ScraperStorage { #[instrument] pub async fn init + Debug>(database_path: P) -> Result { - // 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); diff --git a/nym-validator-rewarder/Cargo.toml b/nym-validator-rewarder/Cargo.toml index d4e908955b..081d32c925 100644 --- a/nym-validator-rewarder/Cargo.toml +++ b/nym-validator-rewarder/Cargo.toml @@ -17,6 +17,7 @@ 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 @@ -33,4 +34,8 @@ nym-config = { path = "../common/config" } nym-network-defaults = { path = "../common/network-defaults" } nym-task = { path = "../common/task" } nym-validator-client = { path = "../common/client-libs/validator-client" } -nyxd-scraper = { path = "../common/nyxd-scraper" } \ No newline at end of file +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"] } diff --git a/nym-validator-rewarder/build.rs b/nym-validator-rewarder/build.rs new file mode 100644 index 0000000000..6bd3636b5e --- /dev/null +++ b/nym-validator-rewarder/build.rs @@ -0,0 +1,28 @@ +// Copyright 2023 - Nym Technologies SA +// 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); +} diff --git a/nym-validator-rewarder/migrations/01_initial.sql b/nym-validator-rewarder/migrations/01_initial.sql index 0719d1f4fb..ab618ffc91 100644 --- a/nym-validator-rewarder/migrations/01_initial.sql +++ b/nym-validator-rewarder/migrations/01_initial.sql @@ -1,4 +1,67 @@ /* * Copyright 2023 - Nym Technologies SA * SPDX-License-Identifier: GPL-3.0-only - */ \ No newline at end of file + */ + +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, + 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 DECIMAL NOT NULL, + signed_blocks INTEGER NOT NULL, + signed_blocks_percent DECIMAL 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), + dkg_epoch_id INTEGER NOT NULL,-- currently not incrementing, needs to change + total_issued_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), + amount TEXT NOT NULL, + operator_account TEXT NOT NULL, + api_endpoint TEXT NOT NULL, + + + UNIQUE (rewarding_epoch_id, operator_account) +); + +-- CREATE TABLE credential_verification_reward +-- ( +-- +-- ); + diff --git a/nym-validator-rewarder/src/config/mod.rs b/nym-validator-rewarder/src/config/mod.rs index d02cd97672..6c9c86cbac 100644 --- a/nym-validator-rewarder/src/config/mod.rs +++ b/nym-validator-rewarder/src/config/mod.rs @@ -112,7 +112,8 @@ impl Config { pub fn rpc_client_config(&self) -> nyxd::Config { // TEMP - nyxd::Config::try_from_nym_network_details(&NymNetworkDetails::new_from_env()).unwrap() + nyxd::Config::try_from_nym_network_details(&NymNetworkDetails::new_from_env()) + .expect("failed to create nyxd client config") } pub fn ensure_is_valid(&self) -> Result<(), NymRewarderError> { diff --git a/nym-validator-rewarder/src/config/persistence/paths.rs b/nym-validator-rewarder/src/config/persistence/paths.rs index 4317339677..daee529076 100644 --- a/nym-validator-rewarder/src/config/persistence/paths.rs +++ b/nym-validator-rewarder/src/config/persistence/paths.rs @@ -11,12 +11,15 @@ 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), } } } diff --git a/nym-validator-rewarder/src/error.rs b/nym-validator-rewarder/src/error.rs index aca9dd61a3..c6d722e415 100644 --- a/nym-validator-rewarder/src/error.rs +++ b/nym-validator-rewarder/src/error.rs @@ -9,6 +9,12 @@ 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() )] diff --git a/nym-validator-rewarder/src/rewarder/block_signing/mod.rs b/nym-validator-rewarder/src/rewarder/block_signing/mod.rs index caaf2be59d..18f4e85f3a 100644 --- a/nym-validator-rewarder/src/rewarder/block_signing/mod.rs +++ b/nym-validator-rewarder/src/rewarder/block_signing/mod.rs @@ -4,9 +4,9 @@ use crate::error::NymRewarderError; use crate::rewarder::block_signing::types::EpochSigningResults; 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, StakingQueryClient}; -use nym_validator_client::QueryHttpRpcNyxdClient; +use nym_validator_client::nyxd::PageRequest; use nyxd_scraper::NyxdScraper; use std::cmp::min; use std::collections::HashMap; @@ -16,7 +16,7 @@ use tracing::info; pub(crate) mod types; pub struct EpochSigning { - pub(crate) rpc_client: QueryHttpRpcNyxdClient, + pub(crate) nyxd_client: NyxdClient, pub(crate) nyxd_scraper: NyxdScraper, } @@ -47,17 +47,14 @@ impl EpochSigning { ) -> Result, NymRewarderError> { // first attempt to get it via the historical info. // if that fails, attempt to use current block information to at least get **something** - if let Some(validators) = self.rpc_client.historical_info(height).await?.hist { + 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 - .rpc_client - .validators("".to_string(), page_request) - .await?; + let mut res = self.nyxd_client.validators(page_request).await?; response.append(&mut res.validators); let Some(pagination) = res.pagination else { diff --git a/nym-validator-rewarder/src/rewarder/credential_issuance/mod.rs b/nym-validator-rewarder/src/rewarder/credential_issuance/mod.rs new file mode 100644 index 0000000000..697d33d779 --- /dev/null +++ b/nym-validator-rewarder/src/rewarder/credential_issuance/mod.rs @@ -0,0 +1,27 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +use crate::error::NymRewarderError; +use crate::rewarder::credential_issuance::types::CredentialIssuanceResults; +use crate::rewarder::epoch::Epoch; +use tracing::info; + +pub mod types; + +pub struct CredentialIssuance {} + +impl CredentialIssuance { + pub(crate) async fn get_signed_blocks_results( + &self, + current_epoch: Epoch, + ) -> Result { + info!( + "looking up credential issuers for epoch {} ({} - {})", + current_epoch.id, + current_epoch.start_rfc3339(), + current_epoch.end_rfc3339() + ); + + Ok(CredentialIssuanceResults {}) + } +} diff --git a/nym-validator-rewarder/src/rewarder/credential_issuance/types.rs b/nym-validator-rewarder/src/rewarder/credential_issuance/types.rs new file mode 100644 index 0000000000..a0766af2a9 --- /dev/null +++ b/nym-validator-rewarder/src/rewarder/credential_issuance/types.rs @@ -0,0 +1,13 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use nym_validator_client::nyxd::{AccountId, Coin}; + +pub struct CredentialIssuanceResults {} + +impl CredentialIssuanceResults { + pub fn rewarding_amounts(&self, budget: &Coin) -> Vec<(AccountId, Vec)> { + let _ = budget; + Vec::new() + } +} \ No newline at end of file diff --git a/nym-validator-rewarder/src/rewarder/mod.rs b/nym-validator-rewarder/src/rewarder/mod.rs index a464b3f8dc..3372fad544 100644 --- a/nym-validator-rewarder/src/rewarder/mod.rs +++ b/nym-validator-rewarder/src/rewarder/mod.rs @@ -3,102 +3,138 @@ use crate::config::Config; use crate::error::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}; -use nym_validator_client::QueryHttpRpcNyxdClient; +use nym_validator_client::nyxd::{AccountId, Coin, Hash}; use nyxd_scraper::NyxdScraper; use std::ops::Add; use std::time::Duration; use time::OffsetDateTime; use tokio::pin; use tokio::time::{interval_at, Instant}; -use tracing::{debug, error, info, instrument}; +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: EpochSigningResults, + pub credentials: CredentialIssuanceResults, + + pub total_budget: Coin, + pub signing_budget: Coin, + pub credentials_budget: Coin, +} + +impl EpochRewards { + pub fn amounts(&self) -> Vec<(AccountId, Vec)> { + let signing = self.signing.rewarding_amounts(&self.signing_budget); + let mut credentials = Vec::new(); + + let mut amounts = signing; + amounts.append(&mut credentials); + + amounts + } +} + pub struct Rewarder { config: Config, current_epoch: Epoch, + storage: RewarderStorage, + nyxd_client: NyxdClient, epoch_signing: EpochSigning, + credential_issuance: CredentialIssuance, } impl Rewarder { pub async fn new(config: Config) -> Result { let nyxd_scraper = NyxdScraper::new(config.scraper_config()).await?; - let rpc_client = QueryHttpRpcNyxdClient::connect( - config.rpc_client_config(), - config.base.upstream_nyxd.as_str(), - )?; + let nyxd_client = NyxdClient::new(&config); + let storage = RewarderStorage::init(&config.storage_paths.reward_history).await?; Ok(Rewarder { current_epoch: Epoch::first()?, config, epoch_signing: EpochSigning { nyxd_scraper, - rpc_client, + nyxd_client: nyxd_client.clone(), }, + nyxd_client, + storage, + credential_issuance: CredentialIssuance {}, }) } - #[instrument(skip(self,budget), fields(budget = %budget))] + #[instrument(skip(self))] async fn calculate_block_signing_rewards( &mut self, - budget: Coin, - ) -> Result)>, NymRewarderError> { + ) -> Result { info!("calculating reward shares"); - let signed = self - .epoch_signing + self.epoch_signing .get_signed_blocks_results(self.current_epoch) - .await?; - - debug!("details: {signed:?}"); - - Ok(signed.rewarding_amounts(&budget)) + .await } - #[instrument(skip(self,budget), fields(budget = %budget))] + #[instrument(skip(self))] async fn calculate_credential_rewards( &mut self, - budget: Coin, - ) -> Result)>, NymRewarderError> { + ) -> Result { info!("calculating reward shares"); - Ok(Vec::new()) + self.credential_issuance + .get_signed_blocks_results(self.current_epoch) + .await } - async fn determine_epoch_rewards( - &mut self, - ) -> Result)>, NymRewarderError> { - let epoch_budget = &self.config.rewarding.epoch_budget; + async fn determine_epoch_rewards(&mut self) -> Result { + 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 credential_budget = Coin::new( + 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(signing_budget).await?; - let mut credential_rewards = self.calculate_credential_rewards(credential_budget).await?; + let signing_rewards = self.calculate_block_signing_rewards().await?; + let credential_rewards = self.calculate_credential_rewards().await?; - let mut rewards = signing_rewards; - rewards.append(&mut credential_rewards); - Ok(rewards) + Ok(EpochRewards { + epoch: self.current_epoch, + signing: signing_rewards, + credentials: credential_rewards, + total_budget: epoch_budget.clone(), + signing_budget, + credentials_budget, + }) } async fn send_rewards( &self, amounts: Vec<(AccountId, Vec)>, - ) -> Result<(), NymRewarderError> { - Ok(()) + ) -> Result { + let address = self.nyxd_client.address().await; + info!("here we ({address}) will be sending the following rewards:"); + for (target, amount) in amounts { + info!("{amount:?} to {target}") + } + + Ok(Hash::Sha256([0u8; 32])) } async fn handle_epoch_end(&mut self) { @@ -112,10 +148,14 @@ impl Rewarder { } }; - if let Err(err) = self.send_rewards(rewards).await { - error!("failed to send epoch rewards: {err}"); - return; - }; + let rewarding_result = self.send_rewards(rewards.amounts()).await; + if let Err(err) = self + .storage + .save_rewarding_information(rewards, rewarding_result) + .await + { + error!("failed to persist rewarding information: {err}") + } self.current_epoch = self.current_epoch.next(); } @@ -139,6 +179,7 @@ impl Rewarder { until_end.as_secs() ); let mut epoch_ticker = interval_at(Instant::now().add(until_end), Epoch::LENGTH); + let shutdown_future = task_manager.catch_interrupt(); pin!(shutdown_future); @@ -158,20 +199,6 @@ impl Rewarder { self.epoch_signing.nyxd_scraper.stop().await; - /* - task 1: - on timer: - - go to DKG contract - - get all coconut signers - - for each of them get the info, verify, etc - - task 2: - on timer (or maybe per block?): - - query abci endpoint for VP - - also maybe missed blocks, etc - - */ - - todo!() + Ok(()) } } diff --git a/nym-validator-rewarder/src/rewarder/nyxd_client.rs b/nym-validator-rewarder/src/rewarder/nyxd_client.rs index 7084e7c7f0..25f8bc46c6 100644 --- a/nym-validator-rewarder/src/rewarder/nyxd_client.rs +++ b/nym-validator-rewarder/src/rewarder/nyxd_client.rs @@ -2,6 +2,12 @@ // SPDX-License-Identifier: GPL-3.0-only use crate::config::Config; +use crate::error::NymRewarderError; +use nym_validator_client::nyxd::error::NyxdError; +use nym_validator_client::nyxd::module_traits::staking::{ + QueryHistoricalInfoResponse, QueryValidatorResponse, QueryValidatorsResponse, +}; +use nym_validator_client::nyxd::{AccountId, CosmWasmClient, PageRequest, StakingQueryClient}; use nym_validator_client::{nyxd, DirectSigningHttpRpcNyxdClient}; use std::sync::Arc; use tokio::sync::RwLock; @@ -13,17 +19,14 @@ pub struct NyxdClient { impl NyxdClient { pub(crate) fn new(config: &Config) -> Self { - let details = config.get_network_details(); - let nyxd_url = config.get_nyxd_url(); - - let client_config = nyxd::Config::try_from_nym_network_details(&details) - .expect("failed to construct valid validator client config with the provided network"); + let client_config = config.rpc_client_config(); + 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.as_str(), + nyxd_url, mnemonic, ) .expect("Failed to connect to nyxd!"); @@ -32,4 +35,27 @@ impl NyxdClient { inner: Arc::new(RwLock::new(inner)), } } + + pub(crate) async fn address(&self) -> AccountId { + self.inner.read().await.address() + } + + pub(crate) async fn historical_info( + &self, + height: i64, + ) -> Result { + Ok(self.inner.read().await.historical_info(height).await?) + } + + pub(crate) async fn validators( + &self, + pagination: Option, + ) -> Result { + Ok(self + .inner + .read() + .await + .validators("".to_string(), pagination) + .await?) + } } diff --git a/nym-validator-rewarder/src/rewarder/storage/manager.rs b/nym-validator-rewarder/src/rewarder/storage/manager.rs new file mode 100644 index 0000000000..4c4c2d59fb --- /dev/null +++ b/nym-validator-rewarder/src/rewarder/storage/manager.rs @@ -0,0 +1,52 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +use crate::rewarder::epoch::Epoch; +use sqlx::types::time::OffsetDateTime; +use sqlx::{Executor, Sqlite}; +use tracing::{instrument, trace}; + +#[derive(Clone)] +pub(crate) struct StorageManager { + pub(crate) connection_pool: sqlx::SqlitePool, +} + +impl StorageManager { + pub(crate) async fn insert_rewarding_epoch( + &self, + epoch: Epoch, + rewarding_budget: String, + rewarding_tx: Option, + rewarding_error: Option, + ) -> Result<(), sqlx::Error> { + todo!() + } + + 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> { + todo!() + } + + pub(crate) async fn insert_rewarding_epoch_block_signing_reward( + &self, + ) -> Result<(), sqlx::Error> { + todo!() + } + + pub(crate) async fn insert_rewarding_epoch_credential_issuance( + &self, + ) -> Result<(), sqlx::Error> { + todo!() + } + + pub(crate) async fn insert_rewarding_epoch_credential_issuance_reward( + &self, + ) -> Result<(), sqlx::Error> { + todo!() + } +} diff --git a/nym-validator-rewarder/src/rewarder/storage/mod.rs b/nym-validator-rewarder/src/rewarder/storage/mod.rs new file mode 100644 index 0000000000..8227231f24 --- /dev/null +++ b/nym-validator-rewarder/src/rewarder/storage/mod.rs @@ -0,0 +1,64 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +use crate::error::NymRewarderError; +use crate::rewarder::storage::manager::StorageManager; +use crate::rewarder::EpochRewards; +use nym_validator_client::nyxd::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 + Debug>(database_path: P) -> Result { + 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 save_rewarding_information( + &self, + reward: EpochRewards, + rewarding_tx: Result, + ) -> 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())), + }; + + Ok(()) + } +}