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

starting on persistence

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2023-12-14 17:17:41 +00:00
committed by Jon Häggblad
parent 397ef8723d
commit 4f6fe88b4c
15 changed files with 383 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -7508,6 +7508,7 @@ dependencies = [
"nyxd-scraper",
"serde",
"sha2 0.10.8",
"sqlx",
"thiserror",
"time",
"tokio",
common/nyxd-scraper/src/storage/mod.rs
-2
View File
@@ -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);
nym-validator-rewarder/Cargo.toml
+6 -1
View File
@@ -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" }
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"] }
nym-validator-rewarder/build.rs
+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);
}
nym-validator-rewarder/migrations/01_initial.sql
+64 -1
View File
@@ -1,4 +1,67 @@
/*
* 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,
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
-- (
--
-- );
nym-validator-rewarder/src/config/mod.rs
+2 -1
View File
@@ -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> {
nym-validator-rewarder/src/config/persistence/paths.rs
+3
View File
@@ -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),
}
}
}
nym-validator-rewarder/src/error.rs
+6
View File
@@ -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()
)]
nym-validator-rewarder/src/rewarder/block_signing/mod.rs
+5 -8
View File
@@ -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<Vec<staking::Validator>, NymRewarderError> {
// first attempt to get it via the historical info.
// if that fails, attempt to use current block information to at least get **something**
if let Some(validators) = self.rpc_client.historical_info(height).await?.hist {
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 {
nym-validator-rewarder/src/rewarder/credential_issuance/mod.rs
+27
View File
@@ -0,0 +1,27 @@
// 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::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<CredentialIssuanceResults, NymRewarderError> {
info!(
"looking up credential issuers for epoch {} ({} - {})",
current_epoch.id,
current_epoch.start_rfc3339(),
current_epoch.end_rfc3339()
);
Ok(CredentialIssuanceResults {})
}
}
nym-validator-rewarder/src/rewarder/credential_issuance/types.rs
+13
View File
@@ -0,0 +1,13 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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<Coin>)> {
let _ = budget;
Vec::new()
}
}
nym-validator-rewarder/src/rewarder/mod.rs
+80 -53
View File
@@ -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<Coin>)> {
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<Self, NymRewarderError> {
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<Vec<(AccountId, Vec<Coin>)>, NymRewarderError> {
) -> Result<EpochSigningResults, NymRewarderError> {
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<Vec<(AccountId, Vec<Coin>)>, NymRewarderError> {
) -> Result<CredentialIssuanceResults, NymRewarderError> {
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<Vec<(AccountId, Vec<Coin>)>, NymRewarderError> {
let epoch_budget = &self.config.rewarding.epoch_budget;
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 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<Coin>)>,
) -> Result<(), NymRewarderError> {
Ok(())
) -> Result<Hash, NymRewarderError> {
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(())
}
}
nym-validator-rewarder/src/rewarder/nyxd_client.rs
+32 -6
View File
@@ -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<QueryHistoricalInfoResponse, NymRewarderError> {
Ok(self.inner.read().await.historical_info(height).await?)
}
pub(crate) async fn validators(
&self,
pagination: Option<PageRequest>,
) -> Result<QueryValidatorsResponse, NymRewarderError> {
Ok(self
.inner
.read()
.await
.validators("".to_string(), pagination)
.await?)
}
}
nym-validator-rewarder/src/rewarder/storage/manager.rs
+52
View File
@@ -0,0 +1,52 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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<String>,
rewarding_error: Option<String>,
) -> 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!()
}
}
nym-validator-rewarder/src/rewarder/storage/mod.rs
+64
View File
@@ -0,0 +1,64 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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<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 save_rewarding_information(
&self,
reward: EpochRewards,
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())),
};
Ok(())
}
}