GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #4399 from nymtech/bugfix/signing-rewards

Browse Source 
Bugfix/signing rewards
This commit is contained in:
Jędrzej Stuczyński
2024-02-19 11:30:42 +00:00
committed by GitHub
parent 20819331f3 0b46e5b753
commit ac676760d4
13 changed files with 366 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -6694,6 +6694,7 @@ dependencies = [
"tendermint",
"tendermint-rpc",
"thiserror",
"time",
"tokio",
"tokio-stream",
"tokio-util",
common/nyxd-scraper/Cargo.toml
+1
View File
@@ -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"]}
common/nyxd-scraper/src/error.rs
+1 -1
View File
@@ -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 {
common/nyxd-scraper/src/scraper/mod.rs
+17 -22
View File
@@ -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()
}
}
common/nyxd-scraper/src/scraper/subscriber.rs
+129 -17
View File
@@ -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<BlockToProcess>,
websocket_endpoint: Url,
websocket_client: WebSocketClient,
websocket_driver: Option<WebSocketClientDriver>,
}
@@ -26,6 +33,7 @@ impl ChainSubscriber {
pub async fn new(
websocket_endpoint: &Url,
cancel: CancellationToken,
task_tracker: TaskTracker,
block_sender: UnboundedSender<BlockToProcess>,
) -> Result<Self, ScraperError> {
// 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<bool, ScraperError> {
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()
}
}
}
nym-connect/desktop/Cargo.lock
Generated
+10 -15
View File
@@ -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"
nym-validator-rewarder/src/error.rs
+3
View File
@@ -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)]
nym-validator-rewarder/src/rewarder/block_signing/mod.rs
+8 -1
View File
@@ -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<i64>,
) -> Result<Option<i64>, 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;
};
nym-validator-rewarder/src/rewarder/block_signing/types.rs
+1
View File
@@ -53,6 +53,7 @@ pub struct EpochSigningResults {
pub validators: Vec<ValidatorSigning>,
}
#[derive(Debug)]
pub struct RawValidatorResult {
pub signed_blocks: i32,
pub voting_power: i64,
nym-validator-rewarder/src/rewarder/mod.rs
+71 -33
View File
@@ -1,4 +1,4 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// Copyright 2023-2024 - Nym Technologies SA <contact@nymtech.net>
// 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<EpochSigningResults>,
pub credentials: Option<CredentialIssuanceResults>,
pub signing: Result<Option<EpochSigningResults>, NymRewarderError>,
pub credentials: Result<Option<CredentialIssuanceResults>, 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<Coin>)> {
pub fn amounts(&self) -> Result<Vec<(AccountId, Vec<Coin>)>, 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<EpochRewards, NymRewarderError> {
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<RewardingResult, NymRewarderError> {
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
}
}
nym-validator-rewarder/src/rewarder/storage/manager.rs
+3 -3
View File
@@ -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!(
nym-validator-rewarder/src/rewarder/storage/mod.rs
+114 -83
View File
@@ -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<Hash, NymRewarderError>,
rewarding_result: Result<RewardingResult, NymRewarderError>, // 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 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(())
nym-wallet/Cargo.lock
Generated
+7 -11
View File
@@ -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"