diff --git a/Cargo.lock b/Cargo.lock index 0b602f6a58..ac8415dbcf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6694,6 +6694,7 @@ dependencies = [ "tendermint", "tendermint-rpc", "thiserror", + "time", "tokio", "tokio-stream", "tokio-util", diff --git a/common/nyxd-scraper/Cargo.toml b/common/nyxd-scraper/Cargo.toml index 772ab8e7b3..858f89512b 100644 --- a/common/nyxd-scraper/Cargo.toml +++ b/common/nyxd-scraper/Cargo.toml @@ -21,6 +21,7 @@ sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros tendermint.workspace = true tendermint-rpc = { workspace = true, features = ["websocket-client", "http-client"] } thiserror.workspace = true +time = { workspace = true } tokio = { workspace = true, features = ["full"] } tokio-stream = "0.1.14" tokio-util = { version = "0.7.10", features = ["rt"]} diff --git a/common/nyxd-scraper/src/error.rs b/common/nyxd-scraper/src/error.rs index dec4fb26e9..daefd92ff4 100644 --- a/common/nyxd-scraper/src/error.rs +++ b/common/nyxd-scraper/src/error.rs @@ -84,7 +84,7 @@ pub enum ScraperError { EmptyBlockData { query: String }, #[error("reached maximum number of allowed errors for subscription events")] - MaximumSubscriptionFailures, + MaximumWebSocketFailures, #[error("failed to begin storage tx: {source}")] StorageTxBeginFailure { diff --git a/common/nyxd-scraper/src/scraper/mod.rs b/common/nyxd-scraper/src/scraper/mod.rs index f417318227..c6c668e41d 100644 --- a/common/nyxd-scraper/src/scraper/mod.rs +++ b/common/nyxd-scraper/src/scraper/mod.rs @@ -6,11 +6,10 @@ use crate::block_requester::BlockRequester; use crate::error::ScraperError; use crate::modules::{BlockModule, MsgModule, TxModule}; use crate::rpc_client::RpcClient; -use crate::scraper::subscriber::{run_websocket_driver, ChainSubscriber}; +use crate::scraper::subscriber::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; @@ -67,20 +66,15 @@ impl NyxdScraperBuilder { block_processor.set_tx_modules(self.tx_modules); block_processor.set_msg_modules(self.msg_modules); - let mut chain_subscriber = ChainSubscriber::new( + let chain_subscriber = ChainSubscriber::new( &scraper.config.websocket_url, scraper.cancel_token.clone(), + scraper.task_tracker.clone(), processing_tx, ) .await?; - let ws_driver = chain_subscriber.ws_driver(); - scraper.start_tasks( - block_requester, - block_processor, - chain_subscriber, - ws_driver, - ); + scraper.start_tasks(block_requester, block_processor, chain_subscriber); Ok(scraper) } @@ -141,7 +135,6 @@ impl NyxdScraper { mut block_requester: BlockRequester, mut block_processor: BlockProcessor, mut chain_subscriber: ChainSubscriber, - ws_driver: WebSocketClientDriver, ) { self.task_tracker .spawn(async move { block_requester.run().await }); @@ -149,8 +142,7 @@ impl NyxdScraper { .spawn(async move { block_processor.run().await }); self.task_tracker .spawn(async move { chain_subscriber.run().await }); - self.task_tracker - .spawn(run_websocket_driver(ws_driver, self.cancel_token.clone())); + self.task_tracker.close(); } @@ -176,21 +168,16 @@ impl NyxdScraper { rpc_client, ) .await?; - let mut chain_subscriber = ChainSubscriber::new( + let chain_subscriber = ChainSubscriber::new( &self.config.websocket_url, self.cancel_token.clone(), + self.task_tracker.clone(), processing_tx, ) .await?; - let ws_driver = chain_subscriber.ws_driver(); // spawn them - self.start_tasks( - block_requester, - block_processor, - chain_subscriber, - ws_driver, - ); + self.start_tasks(block_requester, block_processor, chain_subscriber); Ok(()) } @@ -201,10 +188,18 @@ impl NyxdScraper { } pub async fn stop(self) { - info!("stopping the chain scrapper"); + info!("stopping the chain scraper"); assert!(self.task_tracker.is_closed()); self.cancel_token.cancel(); self.task_tracker.wait().await } + + pub fn cancel_token(&self) -> CancellationToken { + self.cancel_token.clone() + } + + pub fn is_cancelled(&self) -> bool { + self.cancel_token.is_cancelled() + } } diff --git a/common/nyxd-scraper/src/scraper/subscriber.rs b/common/nyxd-scraper/src/scraper/subscriber.rs index 6e9fd3e85d..461d362eb5 100644 --- a/common/nyxd-scraper/src/scraper/subscriber.rs +++ b/common/nyxd-scraper/src/scraper/subscriber.rs @@ -6,18 +6,25 @@ use crate::error::ScraperError; use tendermint_rpc::event::Event; use tendermint_rpc::query::EventType; use tendermint_rpc::{SubscriptionClient, WebSocketClient, WebSocketClientDriver}; +use time::{Duration, OffsetDateTime}; use tokio::sync::mpsc::UnboundedSender; use tokio_stream::StreamExt; use tokio_util::sync::CancellationToken; +use tokio_util::task::TaskTracker; use tracing::{error, info, warn}; use url::Url; const MAX_FAILURES: usize = 10; +const MAX_RECONNECTION_ATTEMPTS: usize = 8; +const SOCKET_FAILURE_RESET: Duration = Duration::hours(2); pub struct ChainSubscriber { cancel: CancellationToken, + task_tracker: TaskTracker, + block_sender: UnboundedSender, + websocket_endpoint: Url, websocket_client: WebSocketClient, websocket_driver: Option, } @@ -26,6 +33,7 @@ impl ChainSubscriber { pub async fn new( websocket_endpoint: &Url, cancel: CancellationToken, + task_tracker: TaskTracker, block_sender: UnboundedSender, ) -> Result { // sure, we could have just used websocket client entirely, but let's keep the logic for @@ -39,7 +47,9 @@ impl ChainSubscriber { Ok(ChainSubscriber { cancel, + task_tracker, block_sender, + websocket_endpoint: websocket_endpoint.clone(), websocket_client: client, websocket_driver: Some(driver), }) @@ -53,8 +63,48 @@ impl ChainSubscriber { Ok(()) } - pub(crate) async fn run(&mut self) -> Result<(), ScraperError> { - let _drop_guard = self.cancel.clone().drop_guard(); + async fn remake_connection(&mut self) -> Result<(), ScraperError> { + info!( + "attempting to reestablish connection to {}", + self.websocket_endpoint + ); + + let (client, driver) = WebSocketClient::new(self.websocket_endpoint.as_str()) + .await + .map_err(|source| ScraperError::WebSocketConnectionFailure { + url: self.websocket_endpoint.to_string(), + source, + })?; + self.websocket_client = client; + self.websocket_driver = Some(driver); + + info!( + "managed to reestablish the websocket connection to {}", + self.websocket_endpoint + ); + Ok(()) + } + + /// Returns whether the method exited due to the cancellation + async fn run_chain_subscription(&mut self) -> Result { + let Some(ws_driver) = self.websocket_driver.take() else { + error!("the websocket driver hasn't been created - we probably failed to establish the connection"); + return Ok(false); + }; + + let driver_cancel = CancellationToken::new(); + let _driver_guard = driver_cancel.clone().drop_guard(); + + // spawn the websocket driver task + let driver_handle = { + self.task_tracker.reopen(); + let handle = self + .task_tracker + .spawn(run_websocket_driver(ws_driver, driver_cancel)); + self.task_tracker.close(); + handle + }; + tokio::pin!(driver_handle); info!("creating chain subscription"); let mut subs = self @@ -70,12 +120,17 @@ impl ChainSubscriber { tokio::select! { _ = self.cancel.cancelled() => { info!("received cancellation token"); - break + // note: `_driver_guard` will get dropped here thus causing cancellation of the driver task + return Ok(true) + } + _ = &mut driver_handle => { + error!("our websocket driver has finished execution"); + return Ok(self.cancel.is_cancelled()) } maybe_event = subs.next() => { let Some(maybe_event) = maybe_event else { warn!("stopped receiving new events"); - break; + return Ok(false) }; match maybe_event { Ok(event) => { @@ -92,38 +147,95 @@ impl ChainSubscriber { } } if failures >= MAX_FAILURES { - // note: the drop_guard will get dropped and thus cause a shutdown - return Err(ScraperError::MaximumSubscriptionFailures); + return Ok(false) } } } } - Ok(()) } - pub(crate) fn ws_driver(&mut self) -> WebSocketClientDriver { - #[allow(clippy::expect_used)] - self.websocket_driver - .take() - .expect("websocket driver has already been started!") + async fn websocket_backoff(&mut self, failure_count: usize) -> bool { + const MINIMUM_WAIT_MS: u64 = 10_000; + const INCREMENTAL_WAIT_MS: u64 = 30_000; + + let backoff_duration_ms = MINIMUM_WAIT_MS + INCREMENTAL_WAIT_MS * failure_count as u64; + info!("going to wait {backoff_duration_ms} ms before re-attempting the reconnection"); + + tokio::select! { + _ = self.cancel.cancelled() => { + info!("received cancellation token"); + true + } + _ = tokio::time::sleep(std::time::Duration::from_millis(backoff_duration_ms)) => false, + } + } + + pub(crate) async fn run(&mut self) -> Result<(), ScraperError> { + let _drop_guard = self.cancel.clone().drop_guard(); + let mut socket_failures = 0; + let mut last_failure = OffsetDateTime::now_utc(); + + loop { + if self.cancel.is_cancelled() { + return Ok(()); + } + + match self.run_chain_subscription().await { + Ok(cancelled) => { + if cancelled { + // we're in the middle of a shutdown + return Ok(()); + } + socket_failures += 1; + } + Err(err) => { + error!("failed to create chain subscription: {err}"); + socket_failures += 1; + } + } + + warn!("current socket failure count: {socket_failures}. the last failure was at {last_failure}"); + + let now = OffsetDateTime::now_utc(); + + // if it's been a while since the last failure, reset the count + if now - last_failure > SOCKET_FAILURE_RESET { + warn!("resetting the failure count to 1"); + socket_failures = 1; + } + last_failure = now; + + if socket_failures >= MAX_RECONNECTION_ATTEMPTS { + error!("reached the maximum allowed failure count"); + return Err(ScraperError::MaximumWebSocketFailures); + } + + // BACKOFF + let cancelled = self.websocket_backoff(socket_failures).await; + if cancelled { + return Ok(()); + } + + if let Err(err) = self.remake_connection().await { + error!("failed to re-establish the websocket connection: {err}"); + } + } } } -pub async fn run_websocket_driver(driver: WebSocketClientDriver, cancel: CancellationToken) { +pub async fn run_websocket_driver(driver: WebSocketClientDriver, driver_cancel: CancellationToken) { info!("starting websocket driver"); tokio::select! { - _ = cancel.cancelled() => { + _ = driver_cancel.cancelled() => { info!("received cancellation token") } res = driver.run() => { match res { Ok(_) => info!("our websocket driver has finished execution"), Err(err) => { - // TODO: in the future just attempt to reconnect - error!("our websocket driver has errored out: {err}") + error!("our websocket driver has errored out: {err}"); } } - cancel.cancel() } } } diff --git a/nym-connect/desktop/Cargo.lock b/nym-connect/desktop/Cargo.lock index e3e8cf14c3..c284610844 100644 --- a/nym-connect/desktop/Cargo.lock +++ b/nym-connect/desktop/Cargo.lock @@ -4513,7 +4513,6 @@ dependencies = [ "nym-network-defaults", "nym-service-provider-directory-common", "nym-vesting-contract-common", - "openssl", "prost", "reqwest", "serde", @@ -4660,15 +4659,6 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf" -[[package]] -name = "openssl-src" -version = "111.27.0+1.1.1v" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "06e8f197c82d7511c5b014030c9b1efeda40d7d5f99d23b4ceed3524a5e63f02" -dependencies = [ - "cc", -] - [[package]] name = "openssl-sys" version = "0.9.91" @@ -4677,7 +4667,6 @@ checksum = "866b5f16f90776b9bb8dc1e1802ac6f0513de3a7a7465867bfbc563dc737faac" dependencies = [ "cc", "libc", - "openssl-src", "pkg-config", "vcpkg", ] @@ -5518,6 +5507,7 @@ dependencies = [ "wasm-bindgen-futures", "wasm-streams", "web-sys", + "webpki-roots 0.25.4", "winreg 0.50.0", ] @@ -6473,7 +6463,7 @@ dependencies = [ "thiserror", "tokio-stream", "url", - "webpki-roots", + "webpki-roots 0.22.6", ] [[package]] @@ -7262,9 +7252,8 @@ checksum = "212d5dcb2a1ce06d81107c3d0ffa3121fe974b73f068c8282cb1c32328113b6c" dependencies = [ "futures-util", "log", - "native-tls", + "rustls 0.21.7", "tokio", - "tokio-native-tls", "tungstenite", ] @@ -7444,8 +7433,8 @@ dependencies = [ "http", "httparse", "log", - "native-tls", "rand 0.8.5", + "rustls 0.21.7", "sha1", "thiserror", "url", @@ -7889,6 +7878,12 @@ dependencies = [ "webpki", ] +[[package]] +name = "webpki-roots" +version = "0.25.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5f20c57d8d7db6d3b86154206ae5d8fba62dd39573114de97c2cb0578251f8e1" + [[package]] name = "webview2-com" version = "0.19.1" diff --git a/nym-validator-rewarder/src/error.rs b/nym-validator-rewarder/src/error.rs index 3d00f0324d..2c78e61fbb 100644 --- a/nym-validator-rewarder/src/error.rs +++ b/nym-validator-rewarder/src/error.rs @@ -163,6 +163,9 @@ pub enum NymRewarderError { #[error("credential issuance rewarding is enabled, but the validator whitelist is empty")] EmptyCredentialIssuanceWhitelist, + + #[error("there were no validators to reward in this epoch")] + NoValidatorsToReward, } #[derive(Debug)] diff --git a/nym-validator-rewarder/src/rewarder/block_signing/mod.rs b/nym-validator-rewarder/src/rewarder/block_signing/mod.rs index b87c6efaef..b443364641 100644 --- a/nym-validator-rewarder/src/rewarder/block_signing/mod.rs +++ b/nym-validator-rewarder/src/rewarder/block_signing/mod.rs @@ -11,7 +11,7 @@ use nyxd_scraper::NyxdScraper; use std::cmp::min; use std::collections::HashMap; use std::ops::Range; -use tracing::{debug, info, warn}; +use tracing::{debug, info, trace, warn}; pub(crate) mod types; @@ -28,6 +28,7 @@ impl EpochSigning { height_range: Range, ) -> Result, NymRewarderError> { for height in height_range { + trace!("attempting to get precommit for {address} at height {height}"); if let Some(precommit) = self .nyxd_scraper .storage @@ -87,6 +88,8 @@ impl EpochSigning { ); let validators = self.nyxd_scraper.storage.get_all_known_validators().await?; + debug!("retrieved {} known validators", validators.len()); + let epoch_start = current_epoch.start_time; let epoch_end = current_epoch.end_time; let first_block = self @@ -111,10 +114,14 @@ impl EpochSigning { // for each validator, with a valid voting power, get number of signed blocks in the rewarding epoch for validator in validators { + let addr = &validator.consensus_address; + debug!("getting voting power and signed blocks of {addr}"); + let Some(vp) = self .get_voting_power(&validator.consensus_address, vp_range.clone()) .await? else { + warn!("failed to obtain voting power of {addr}"); continue; }; diff --git a/nym-validator-rewarder/src/rewarder/block_signing/types.rs b/nym-validator-rewarder/src/rewarder/block_signing/types.rs index 8cf31a5ba5..ebdb24432f 100644 --- a/nym-validator-rewarder/src/rewarder/block_signing/types.rs +++ b/nym-validator-rewarder/src/rewarder/block_signing/types.rs @@ -53,6 +53,7 @@ pub struct EpochSigningResults { pub validators: Vec, } +#[derive(Debug)] pub struct RawValidatorResult { pub signed_blocks: i32, pub voting_power: i64, diff --git a/nym-validator-rewarder/src/rewarder/mod.rs b/nym-validator-rewarder/src/rewarder/mod.rs index 216ee8eee8..5d813a11f0 100644 --- a/nym-validator-rewarder/src/rewarder/mod.rs +++ b/nym-validator-rewarder/src/rewarder/mod.rs @@ -1,4 +1,4 @@ -// Copyright 2023 - Nym Technologies SA +// Copyright 2023-2024 - Nym Technologies SA // SPDX-License-Identifier: GPL-3.0-only use crate::config::Config; @@ -10,13 +10,15 @@ use crate::rewarder::credential_issuance::CredentialIssuance; use crate::rewarder::epoch::Epoch; use crate::rewarder::nyxd_client::NyxdClient; use crate::rewarder::storage::RewarderStorage; +use futures::future::{FusedFuture, OptionFuture}; +use futures::FutureExt; 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}; +use tracing::{error, info, instrument, warn}; mod block_signing; mod credential_issuance; @@ -26,10 +28,15 @@ mod nyxd_client; mod storage; mod tasks; +pub struct RewardingResult { + pub total_spent: Coin, + pub rewarding_tx: Hash, +} + pub struct EpochRewards { pub epoch: Epoch, - pub signing: Option, - pub credentials: Option, + pub signing: Result, NymRewarderError>, + pub credentials: Result, NymRewarderError>, pub total_budget: Coin, pub signing_budget: Coin, @@ -37,10 +44,10 @@ pub struct EpochRewards { } impl EpochRewards { - pub fn amounts(&self) -> Vec<(AccountId, Vec)> { + pub fn amounts(&self) -> Result)>, NymRewarderError> { let mut amounts = Vec::new(); - if let Some(signing) = &self.signing { + if let Ok(Some(signing)) = &self.signing { for (account, signing_amount) in signing.rewarding_amounts(&self.signing_budget) { if signing_amount[0].amount != 0 { amounts.push((account, signing_amount)) @@ -48,7 +55,7 @@ impl EpochRewards { } } - if let Some(credentials) = &self.credentials { + if let Ok(Some(credentials)) = &self.credentials { for (account, credential_amount) in credentials.rewarding_amounts(&self.credentials_budget) { @@ -58,7 +65,7 @@ impl EpochRewards { } } - amounts + Ok(amounts) } } @@ -93,6 +100,10 @@ impl Rewarder { return Err(NymRewarderError::EmptyBlockSigningWhitelist); } + if config.block_signing.monitor_only { + info!("the block signing rewarding is running in monitor only mode"); + } + let nyxd_scraper = NyxdScraper::new(config.scraper_config()).await?; Some(EpochSigning { @@ -115,7 +126,9 @@ impl Rewarder { None }; - if config.issuance_monitor.enabled || config.block_signing.enabled { + if config.issuance_monitor.enabled + || (config.block_signing.enabled && !config.block_signing.monitor_only) + { let balance = nyxd_client .balance(&config.rewarding.epoch_budget.denom) .await?; @@ -179,7 +192,7 @@ impl Rewarder { .transpose() } - async fn determine_epoch_rewards(&mut self) -> Result { + async fn determine_epoch_rewards(&mut self) -> EpochRewards { let epoch_budget = self.config.rewarding.epoch_budget.clone(); let denom = &epoch_budget.denom; let signing_budget = Coin::new( @@ -191,17 +204,17 @@ impl Rewarder { denom, ); - let signing_rewards = self.calculate_block_signing_rewards().await?; - let credential_rewards = self.calculate_credential_rewards().await?; + let signing_rewards = self.calculate_block_signing_rewards().await; + let credential_rewards = self.calculate_credential_rewards().await; - Ok(EpochRewards { + EpochRewards { epoch: self.current_epoch, signing: signing_rewards, credentials: credential_rewards, total_budget: epoch_budget.clone(), signing_budget, credentials_budget, - }) + } } #[instrument(skip(self))] @@ -214,33 +227,47 @@ impl Rewarder { return Ok(Hash::Sha256([0u8; 32])); } + if amounts.is_empty() { + warn!("no rewards to send"); + return Err(NymRewarderError::NoValidatorsToReward); + } + 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(); + async fn calculate_and_send_epoch_rewards( + &mut self, + rewards: &EpochRewards, + ) -> Result { + 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; + let rewarding_tx = self.send_rewards(rewarding_amounts).await?; + + Ok(RewardingResult { + total_spent, + rewarding_tx, + }) + } + + async fn handle_epoch_end(&mut self) { + info!("handling the epoch end"); + let base_rewards = self.determine_epoch_rewards().await; + + let rewarding_result = self + .calculate_and_send_epoch_rewards(&base_rewards) + .await + .inspect_err(|err| error!("failed to determine and send epoch_rewards: {err}")); + if let Err(err) = self .storage - .save_rewarding_information(rewards, total_spent, rewarding_result) + .save_rewarding_information(base_rewards, rewarding_result) .await { error!("failed to persist rewarding information: {err}") @@ -263,15 +290,22 @@ impl Rewarder { ); } - if let Some(epoch_signing) = &self.epoch_signing { - epoch_signing.nyxd_scraper.start().await?; - epoch_signing.nyxd_scraper.wait_for_startup_sync().await; - } + let mut scraper_cancellation: OptionFuture<_> = + if let Some(epoch_signing) = &self.epoch_signing { + let cancellation_token = epoch_signing.nyxd_scraper.cancel_token(); + epoch_signing.nyxd_scraper.start().await?; + epoch_signing.nyxd_scraper.wait_for_startup_sync().await; + Some(Box::pin(async move { cancellation_token.cancelled().await }).fuse()) + } else { + None + } + .into(); let until_end = self.current_epoch.until_end(); info!( - "the first epoch will finish in {} secs", + "the initial epoch (id: {}) will finish in {} secs", + self.current_epoch.id, until_end.as_secs() ); let mut epoch_ticker = interval_at( @@ -292,6 +326,10 @@ impl Rewarder { } break; } + _ = &mut scraper_cancellation, if !scraper_cancellation.is_terminated() => { + warn!("the nyxd scraper has been cancelled"); + break + } _ = epoch_ticker.tick() => self.handle_epoch_end().await } } diff --git a/nym-validator-rewarder/src/rewarder/storage/manager.rs b/nym-validator-rewarder/src/rewarder/storage/manager.rs index b8f84bbd18..1eb9e985cf 100644 --- a/nym-validator-rewarder/src/rewarder/storage/manager.rs +++ b/nym-validator-rewarder/src/rewarder/storage/manager.rs @@ -114,9 +114,9 @@ impl StorageManager { 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, + starting_dkg_epoch: i64, + ending_dkg_epoch: i64, + total_issued_partial_credentials: i64, budget: String, ) -> Result<(), sqlx::Error> { sqlx::query!( diff --git a/nym-validator-rewarder/src/rewarder/storage/mod.rs b/nym-validator-rewarder/src/rewarder/storage/mod.rs index 0c35fb0400..dd5e8f68a7 100644 --- a/nym-validator-rewarder/src/rewarder/storage/mod.rs +++ b/nym-validator-rewarder/src/rewarder/storage/mod.rs @@ -4,8 +4,8 @@ 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 crate::rewarder::{EpochRewards, RewardingResult}; +use nym_validator_client::nyxd::Coin; use sqlx::ConnectOptions; use std::fmt::Debug; use std::path::Path; @@ -55,20 +55,45 @@ impl RewarderStorage { Ok(self.manager.load_last_rewarding_epoch().await?) } + async fn insert_failed_rewarding_epoch_block_signing( + &self, + epoch: i64, + budget: &Coin, + ) -> Result<(), NymRewarderError> { + Ok(self + .manager + .insert_rewarding_epoch_block_signing(epoch, -1, -1, budget.to_string()) + .await?) + } + + async fn insert_failed_rewarding_epoch_credential_issuance( + &self, + epoch: i64, + budget: &Coin, + ) -> Result<(), NymRewarderError> { + Ok(self + .manager + .insert_rewarding_epoch_credential_issuance(epoch, -1, -1, -1, budget.to_string()) + .await?) + } + pub(crate) async fn save_rewarding_information( &self, reward: EpochRewards, - total_spent: Coin, - rewarding_tx: Result, + rewarding_result: Result, // total_spent: Coin, + // 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())), + let denom = &reward.total_budget.denom; + + let (reward_tx, total_spent, reward_err) = match rewarding_result { + Ok(res) => (Some(res.rewarding_tx.to_string()), res.total_spent, None), + Err(err) => (None, Coin::new(0, denom), Some(err.to_string())), }; let epoch_id = reward.epoch.id; + // general epoch info self.manager .insert_rewarding_epoch( reward.epoch, @@ -79,89 +104,95 @@ impl RewarderStorage { ) .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(), - validator.whitelisted, - reward_amount, - validator.voting_power_at_epoch_start, - validator.voting_power_ratio.to_string(), - validator.signed_blocks, - validator.ratio_signed.to_string(), - ) - .await?; + // block signing info + if let Ok(block_signing) = reward.signing { + self.manager + .insert_rewarding_epoch_block_signing( + epoch_id, + block_signing + .as_ref() + .map(|s| s.total_voting_power_at_epoch_start) + .unwrap_or_default(), + block_signing.as_ref().map(|s| s.blocks).unwrap_or_default(), + reward.signing_budget.to_string(), + ) + .await?; + if let Some(signing) = block_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(), + validator.whitelisted, + reward_amount, + validator.voting_power_at_epoch_start, + validator.voting_power_ratio.to_string(), + validator.signed_blocks, + validator.ratio_signed.to_string(), + ) + .await?; + } } + } else { + self.insert_failed_rewarding_epoch_block_signing(epoch_id, &reward.signing_budget) + .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(); + // credential info + if let Ok(credential_issuance) = reward.credentials { + // safety: we must have at least a single value here + #[allow(clippy::unwrap_used)] + let dkg_epoch_start = credential_issuance + .as_ref() + .and_then(|c| c.dkg_epochs.first().copied()) + .unwrap_or_default() as i64; + #[allow(clippy::unwrap_used)] + let dkg_epoch_end = credential_issuance + .as_ref() + .and_then(|c| c.dkg_epochs.last().copied()) + .unwrap_or_default() as i64; - self.manager - .insert_rewarding_epoch_credential_issuance( + self.manager + .insert_rewarding_epoch_credential_issuance( + epoch_id, + dkg_epoch_start, + dkg_epoch_end, + credential_issuance + .as_ref() + .map(|c| c.total_issued_partial_credentials) + .unwrap_or_default() as i64, + reward.credentials_budget.to_string(), + ) + .await?; + + if let Some(credentials) = credential_issuance { + 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(), + api_runner.whitelisted, + reward_amount, + api_runner.api_runner, + api_runner.issued_credentials, + api_runner.issued_ratio.to_string(), + api_runner.validated_credentials, + ) + .await?; + } + } + } else { + self.insert_failed_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(), + &reward.credentials_budget, ) .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(), - api_runner.whitelisted, - reward_amount, - api_runner.api_runner, - api_runner.issued_credentials, - api_runner.issued_ratio.to_string(), - api_runner.validated_credentials, - ) - .await?; - } } Ok(()) diff --git a/nym-wallet/Cargo.lock b/nym-wallet/Cargo.lock index 93c8462986..a197456fb9 100644 --- a/nym-wallet/Cargo.lock +++ b/nym-wallet/Cargo.lock @@ -3454,7 +3454,6 @@ dependencies = [ "nym-network-defaults", "nym-service-provider-directory-common", "nym-vesting-contract-common", - "openssl", "prost", "reqwest", "serde", @@ -3686,15 +3685,6 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf" -[[package]] -name = "openssl-src" -version = "111.27.0+1.1.1v" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "06e8f197c82d7511c5b014030c9b1efeda40d7d5f99d23b4ceed3524a5e63f02" -dependencies = [ - "cc", -] - [[package]] name = "openssl-sys" version = "0.9.91" @@ -3703,7 +3693,6 @@ checksum = "866b5f16f90776b9bb8dc1e1802ac6f0513de3a7a7465867bfbc563dc737faac" dependencies = [ "cc", "libc", - "openssl-src", "pkg-config", "vcpkg", ] @@ -4454,6 +4443,7 @@ dependencies = [ "wasm-bindgen", "wasm-bindgen-futures", "web-sys", + "webpki-roots", "winreg", ] @@ -6270,6 +6260,12 @@ dependencies = [ "system-deps 6.1.1", ] +[[package]] +name = "webpki-roots" +version = "0.25.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5f20c57d8d7db6d3b86154206ae5d8fba62dd39573114de97c2cb0578251f8e1" + [[package]] name = "webview2-com" version = "0.19.1"