diff --git a/Cargo.lock b/Cargo.lock index 386287164c..287f7c6706 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6999,7 +6999,8 @@ dependencies = [ "cosmrs 0.15.0 (git+https://github.com/jstuczyn/cosmos-rust?branch=nym-temp/all-validator-features)", "eyre", "futures", - "nym-bin-common", + "humantime 2.1.0", + "serde", "sha2 0.10.8", "sqlx", "tendermint", diff --git a/common/nyxd-scraper/Cargo.toml b/common/nyxd-scraper/Cargo.toml index 858f89512b..579a79a28c 100644 --- a/common/nyxd-scraper/Cargo.toml +++ b/common/nyxd-scraper/Cargo.toml @@ -16,7 +16,9 @@ const_format = "0.2.32" cosmrs.workspace = true eyre = "0.6.9" futures.workspace = true +humantime = "2.1.0" sha2 = "0.10.8" +serde = { workspace = true, features = ["derive"] } sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate", "time"] } tendermint.workspace = true tendermint-rpc = { workspace = true, features = ["websocket-client", "http-client"] } @@ -24,13 +26,13 @@ thiserror.workspace = true time = { workspace = true } tokio = { workspace = true, features = ["full"] } tokio-stream = "0.1.14" -tokio-util = { version = "0.7.10", features = ["rt"]} +tokio-util = { version = "0.7.10", features = ["rt"] } tracing.workspace = true url.workspace = true # TEMP -nym-bin-common = { path = "../bin-common", features = ["basic_tracing"]} +#nym-bin-common = { path = "../bin-common", features = ["basic_tracing"]} [build-dependencies] diff --git a/common/nyxd-scraper/README.md b/common/nyxd-scraper/README.md new file mode 100644 index 0000000000..2e22d292c1 --- /dev/null +++ b/common/nyxd-scraper/README.md @@ -0,0 +1,23 @@ +# Nyxd Scraper + +## Pruning + +Similarly to cosmos-sdk, we incorporate pruning into our (scraped) chain data. We attempt to follow their strategies as +closely as possible for convenience's sake. Therefore, the following are available: + +### Strategies + +The strategies are configured in `config.toml`, with the format `pruning = ""` where the options are: + +* `default`: only the last 362,880 states(approximately 3.5 weeks worth of state) are kept; pruning at 10 block + intervals +* `nothing`: all historic states will be saved, nothing will be deleted (i.e. archiving node) +* `everything`: 2 latest states will be kept; pruning at 10 block intervals. +* `custom`: allow pruning options to be manually specified through `pruning.keep_recent`, and `pruning.interval` + +### Custom Pruning + +These are applied if and only if the pruning strategy is `custom`: + +* `pruning.keep_recent`: N means to keep all of the last N blocks +* `pruning.interval`: N means to delete old block data from disk every Nth block. diff --git a/common/nyxd-scraper/src/block_processor/mod.rs b/common/nyxd-scraper/src/block_processor/mod.rs index d60be91e77..e6af88cdda 100644 --- a/common/nyxd-scraper/src/block_processor/mod.rs +++ b/common/nyxd-scraper/src/block_processor/mod.rs @@ -8,6 +8,7 @@ use crate::error::ScraperError; use crate::modules::{BlockModule, MsgModule, TxModule}; use crate::rpc_client::RpcClient; use crate::storage::{persist_block, ScraperStorage}; +use crate::PruningOptions; use futures::StreamExt; use std::collections::{BTreeMap, HashSet, VecDeque}; use std::ops::{Add, Range}; @@ -18,9 +19,10 @@ use tokio::sync::Notify; use tokio::time::{interval_at, Instant}; use tokio_stream::wrappers::UnboundedReceiverStream; use tokio_util::sync::CancellationToken; -use tracing::{debug, error, info, warn}; +use tracing::{debug, error, info, instrument, trace, warn}; mod helpers; +pub(crate) mod pruning; pub(crate) mod types; const MISSING_BLOCKS_CHECK_INTERVAL: Duration = Duration::from_secs(30); @@ -40,9 +42,11 @@ impl PendingSync { } pub struct BlockProcessor { + pruning_options: PruningOptions, cancel: CancellationToken, synced: Arc, last_processed_height: u32, + last_pruned_height: u32, last_processed_at: Instant, pending_sync: PendingSync, queued_blocks: BTreeMap, @@ -62,6 +66,7 @@ pub struct BlockProcessor { impl BlockProcessor { pub async fn new( + pruning_options: PruningOptions, cancel: CancellationToken, synced: Arc, incoming: UnboundedReceiver, @@ -70,11 +75,17 @@ impl BlockProcessor { rpc_client: RpcClient, ) -> Result { let last_processed = storage.get_last_processed_height().await?; + let last_processed_height = last_processed.try_into().unwrap_or_default(); + + let last_pruned = storage.get_pruned_height().await?; + let last_pruned_height = last_pruned.try_into().unwrap_or_default(); Ok(BlockProcessor { + pruning_options, cancel, synced, - last_processed_height: last_processed.try_into().unwrap_or_default(), + last_processed_height, + last_pruned_height, last_processed_at: Instant::now(), pending_sync: Default::default(), queued_blocks: Default::default(), @@ -131,12 +142,17 @@ impl BlockProcessor { } } + let commit_start = Instant::now(); tx.commit() .await .map_err(|source| ScraperError::StorageTxCommitFailure { source })?; + crate::storage::log_db_operation_time("committing processing tx", commit_start); self.last_processed_height = full_info.block.header.height.value() as u32; self.last_processed_at = Instant::now(); + if let Err(err) = self.maybe_prune_storage().await { + error!("failed to prune the storage: {err}"); + } Ok(()) } @@ -210,6 +226,61 @@ impl BlockProcessor { Ok(()) } + #[instrument(skip(self))] + async fn prune_storage(&mut self) -> Result<(), ScraperError> { + let keep_recent = self.pruning_options.strategy_keep_recent(); + let last_to_keep = self.last_processed_height - keep_recent; + + info!( + keep_recent, + oldest_to_keep = last_to_keep, + "pruning the storage" + ); + + let lowest: u32 = self + .storage + .lowest_block_height() + .await? + .unwrap_or_default() + .try_into() + .unwrap_or_default(); + + let to_prune = last_to_keep.saturating_sub(lowest); + match to_prune { + v if v > 1000 => warn!("approximately {v} blocks worth of data will be pruned"), + v if v > 100 => info!("approximately {v} blocks worth of data will be pruned"), + v if v == 0 => trace!("no blocks to prune"), + v => debug!("approximately {v} blocks worth of data will be pruned"), + } + + if to_prune == 0 { + return Ok(()); + } + + self.storage + .prune_storage(last_to_keep, self.last_processed_height) + .await?; + + self.last_pruned_height = self.last_processed_height; + Ok(()) + } + + async fn maybe_prune_storage(&mut self) -> Result<(), ScraperError> { + debug!("checking for storage pruning"); + + if self.pruning_options.strategy.is_nothing() { + trace!("the current pruning strategy is 'nothing'"); + return Ok(()); + } + + let interval = self.pruning_options.strategy_interval(); + if self.last_pruned_height + interval <= self.last_processed_height { + self.prune_storage().await?; + } + + Ok(()) + } + async fn next_incoming(&mut self, block: BlockToProcess) { let height = block.height; @@ -279,6 +350,8 @@ impl BlockProcessor { async fn startup_resync(&mut self) -> Result<(), ScraperError> { assert!(self.pending_sync.is_empty()); + self.maybe_prune_storage().await?; + let latest_block = self.rpc_client.current_block_height().await? as u32; if latest_block > self.last_processed_height && self.last_processed_height != 0 { let request_range = self.last_processed_height + 1..latest_block + 1; diff --git a/common/nyxd-scraper/src/block_processor/pruning.rs b/common/nyxd-scraper/src/block_processor/pruning.rs new file mode 100644 index 0000000000..e7a99e7dfa --- /dev/null +++ b/common/nyxd-scraper/src/block_processor/pruning.rs @@ -0,0 +1,122 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::error::ScraperError; +use serde::{Deserialize, Serialize}; + +pub const DEFAULT_PRUNING_KEEP_RECENT: u32 = 362880; +pub const DEFAULT_PRUNING_INTERVAL: u32 = 10; +pub const EVERYTHING_PRUNING_KEEP_RECENT: u32 = 2; +pub const EVERYTHING_PRUNING_INTERVAL: u32 = 10; + +/// We follow cosmos-sdk pruning strategies for convenience’s sake. +#[derive(Debug, Default, Clone, Copy, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum PruningStrategy { + /// 'Default' strategy defines a pruning strategy where the last 362880 heights are + /// kept where to-be pruned heights are pruned at every 10th height. + /// The last 362880 heights are kept(approximately 3.5 weeks worth of state) assuming the typical + /// block time is 6s. If these values do not match the applications' requirements, use the "custom" option. + #[default] + Default, + + /// 'Everything' strategy defines a pruning strategy where all committed heights are + /// deleted, storing only the current height and last 2 states. To-be pruned heights are + /// pruned at every 10th height. + Everything, + + /// 'Nothing' strategy defines a pruning strategy where all heights are kept on disk. + Nothing, + + /// 'Custom' strategy defines a pruning strategy where the user specifies the pruning. + Custom, +} + +impl PruningStrategy { + pub fn is_custom(&self) -> bool { + matches!(self, PruningStrategy::Custom) + } + + pub fn is_nothing(&self) -> bool { + matches!(self, PruningStrategy::Nothing) + } + + pub fn is_everything(&self) -> bool { + matches!(self, PruningStrategy::Everything) + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +pub struct PruningOptions { + /// keep_recent defines how many recent heights to keep on disk. + pub keep_recent: u32, + + /// interval defines the frequency of removing the pruned heights from the disk. + pub interval: u32, + + /// strategy defines the currently used kind of [PruningStrategy]. + pub strategy: PruningStrategy, +} + +impl PruningOptions { + pub fn validate(&self) -> Result<(), ScraperError> { + // if strategy is not set to custom, other options are meaningless since they won't be applied + if !self.strategy.is_custom() { + return Ok(()); + } + + if self.interval == 0 { + return Err(ScraperError::ZeroPruningInterval); + } + + if self.interval < EVERYTHING_PRUNING_INTERVAL { + return Err(ScraperError::TooSmallPruningInterval { + interval: self.interval, + }); + } + + if self.keep_recent < EVERYTHING_PRUNING_KEEP_RECENT { + return Err(ScraperError::TooSmallKeepRecent { + keep_recent: self.keep_recent, + }); + } + + Ok(()) + } + + pub fn nothing() -> Self { + PruningOptions { + keep_recent: 0, + interval: 0, + strategy: PruningStrategy::Nothing, + } + } + + pub fn strategy_interval(&self) -> u32 { + match self.strategy { + PruningStrategy::Default => DEFAULT_PRUNING_INTERVAL, + PruningStrategy::Everything => EVERYTHING_PRUNING_INTERVAL, + PruningStrategy::Nothing => 0, + PruningStrategy::Custom => self.interval, + } + } + + pub fn strategy_keep_recent(&self) -> u32 { + match self.strategy { + PruningStrategy::Default => DEFAULT_PRUNING_KEEP_RECENT, + PruningStrategy::Everything => EVERYTHING_PRUNING_KEEP_RECENT, + PruningStrategy::Nothing => 0, + PruningStrategy::Custom => self.keep_recent, + } + } +} + +impl Default for PruningOptions { + fn default() -> Self { + PruningOptions { + keep_recent: DEFAULT_PRUNING_KEEP_RECENT, + interval: DEFAULT_PRUNING_INTERVAL, + strategy: Default::default(), + } + } +} diff --git a/common/nyxd-scraper/src/error.rs b/common/nyxd-scraper/src/error.rs index daefd92ff4..079a28975d 100644 --- a/common/nyxd-scraper/src/error.rs +++ b/common/nyxd-scraper/src/error.rs @@ -1,6 +1,9 @@ // Copyright 2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use crate::block_processor::pruning::{ + EVERYTHING_PRUNING_INTERVAL, EVERYTHING_PRUNING_KEEP_RECENT, +}; use tendermint::Hash; use thiserror::Error; use tokio::sync::mpsc::error::SendError; @@ -122,6 +125,15 @@ pub enum ScraperError { "could not find validator information for {address}; the validator has signed a commit" )] MissingValidatorInfoCommitted { address: String }, + + #[error("pruning.interval must not be set to 0. If you want to disable pruning, select pruning.strategy = \"nothing\"")] + ZeroPruningInterval, + + #[error("pruning.interval must not be smaller than {}. got: {interval}. for most aggressive pruning, select pruning.strategy = \"everything\"", EVERYTHING_PRUNING_INTERVAL)] + TooSmallPruningInterval { interval: u32 }, + + #[error("pruning.keep_recent must not be smaller than {}. got: {keep_recent}. for most aggressive pruning, select pruning.strategy = \"everything\"", EVERYTHING_PRUNING_KEEP_RECENT)] + TooSmallKeepRecent { keep_recent: u32 }, } impl From> for ScraperError { diff --git a/common/nyxd-scraper/src/lib.rs b/common/nyxd-scraper/src/lib.rs index af084c4471..332e38d214 100644 --- a/common/nyxd-scraper/src/lib.rs +++ b/common/nyxd-scraper/src/lib.rs @@ -14,6 +14,7 @@ pub(crate) mod rpc_client; pub(crate) mod scraper; pub mod storage; +pub use block_processor::pruning::{PruningOptions, PruningStrategy}; pub use modules::{BlockModule, MsgModule, TxModule}; pub use scraper::{Config, NyxdScraper}; pub use storage::models; diff --git a/common/nyxd-scraper/src/scraper/mod.rs b/common/nyxd-scraper/src/scraper/mod.rs index c6c668e41d..608731b489 100644 --- a/common/nyxd-scraper/src/scraper/mod.rs +++ b/common/nyxd-scraper/src/scraper/mod.rs @@ -8,6 +8,7 @@ use crate::modules::{BlockModule, MsgModule, TxModule}; use crate::rpc_client::RpcClient; use crate::scraper::subscriber::ChainSubscriber; use crate::storage::ScraperStorage; +use crate::PruningOptions; use std::path::PathBuf; use std::sync::Arc; use tokio::sync::mpsc::{channel, unbounded_channel}; @@ -27,6 +28,8 @@ pub struct Config { pub rpc_url: Url, pub database_path: PathBuf, + + pub pruning_options: PruningOptions, } pub struct NyxdScraperBuilder { @@ -54,6 +57,7 @@ impl NyxdScraperBuilder { processing_tx.clone(), ); let mut block_processor = BlockProcessor::new( + scraper.config.pruning_options, scraper.cancel_token.clone(), scraper.startup_sync.clone(), processing_rx, @@ -119,6 +123,7 @@ impl NyxdScraper { } pub async fn new(config: Config) -> Result { + config.pruning_options.validate()?; let storage = ScraperStorage::init(&config.database_path).await?; Ok(NyxdScraper { @@ -160,6 +165,7 @@ impl NyxdScraper { processing_tx.clone(), ); let block_processor = BlockProcessor::new( + self.config.pruning_options, self.cancel_token.clone(), self.startup_sync.clone(), processing_rx, diff --git a/common/nyxd-scraper/src/storage/manager.rs b/common/nyxd-scraper/src/storage/manager.rs index 1843634714..04af3e364b 100644 --- a/common/nyxd-scraper/src/storage/manager.rs +++ b/common/nyxd-scraper/src/storage/manager.rs @@ -1,9 +1,11 @@ // Copyright 2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use crate::storage::log_db_operation_time; use crate::storage::models::{CommitSignature, Validator}; use sqlx::types::time::OffsetDateTime; use sqlx::{Executor, Sqlite}; +use tokio::time::Instant; use tracing::{instrument, trace}; #[derive(Clone)] @@ -25,10 +27,36 @@ impl StorageManager { Ok(()) } + pub(crate) async fn get_lowest_block(&self) -> Result, sqlx::Error> { + trace!("get_lowest_block"); + let start = Instant::now(); + + let maybe_record = sqlx::query!( + r#" + SELECT height + FROM block + ORDER BY height ASC + LIMIT 1 + "#, + ) + .fetch_optional(&self.connection_pool) + .await?; + log_db_operation_time("get_lowest_block", start); + + if let Some(row) = maybe_record { + Ok(row.height) + } else { + Ok(None) + } + } + pub(crate) async fn get_first_block_height_after( &self, time: OffsetDateTime, ) -> Result, sqlx::Error> { + trace!("get_first_block_height_after"); + let start = Instant::now(); + let maybe_record = sqlx::query!( r#" SELECT height @@ -41,6 +69,7 @@ impl StorageManager { ) .fetch_optional(&self.connection_pool) .await?; + log_db_operation_time("get_first_block_height_after", start); if let Some(row) = maybe_record { Ok(row.height) @@ -53,6 +82,9 @@ impl StorageManager { &self, time: OffsetDateTime, ) -> Result, sqlx::Error> { + trace!("get_last_block_height_before"); + let start = Instant::now(); + let maybe_record = sqlx::query!( r#" SELECT height @@ -65,6 +97,7 @@ impl StorageManager { ) .fetch_optional(&self.connection_pool) .await?; + log_db_operation_time("get_last_block_height_before", start); if let Some(row) = maybe_record { Ok(row.height) @@ -79,6 +112,9 @@ impl StorageManager { start_height: i64, end_height: i64, ) -> Result { + trace!("get_signed_between"); + let start = Instant::now(); + let count = sqlx::query!( r#" SELECT COUNT(*) as count FROM pre_commit @@ -94,6 +130,7 @@ impl StorageManager { .fetch_one(&self.connection_pool) .await? .count; + log_db_operation_time("get_signed_between", start); Ok(count) } @@ -103,7 +140,10 @@ impl StorageManager { consensus_address: &str, height: i64, ) -> Result, sqlx::Error> { - sqlx::query_as( + trace!("get_precommit"); + let start = Instant::now(); + + let res = sqlx::query_as( r#" SELECT * FROM pre_commit WHERE validator_address = ? @@ -113,14 +153,20 @@ impl StorageManager { .bind(consensus_address) .bind(height) .fetch_optional(&self.connection_pool) - .await + .await?; + log_db_operation_time("get_precommit", start); + + Ok(res) } pub(crate) async fn get_block_validators( &self, height: i64, ) -> Result, sqlx::Error> { - sqlx::query_as!( + trace!("get_block_validators"); + let start = Instant::now(); + + let res = sqlx::query_as!( Validator, r#" SELECT * FROM validator @@ -133,16 +179,28 @@ impl StorageManager { height ) .fetch_all(&self.connection_pool) - .await + .await?; + log_db_operation_time("get_block_validators", start); + + Ok(res) } pub(crate) async fn get_validators(&self) -> Result, sqlx::Error> { - sqlx::query_as("SELECT * FROM validator") + trace!("get_validators"); + let start = Instant::now(); + + let res = sqlx::query_as("SELECT * FROM validator") .fetch_all(&self.connection_pool) - .await + .await?; + log_db_operation_time("get_validators", start); + + Ok(res) } pub(crate) async fn get_last_processed_height(&self) -> Result { + trace!("get_last_processed_height"); + let start = Instant::now(); + let maybe_record = sqlx::query!( r#" SELECT last_processed_height FROM metadata @@ -150,6 +208,7 @@ impl StorageManager { ) .fetch_optional(&self.connection_pool) .await?; + log_db_operation_time("get_last_processed_height", start); if let Some(row) = maybe_record { Ok(row.last_processed_height) @@ -157,6 +216,27 @@ impl StorageManager { Ok(-1) } } + + pub(crate) async fn get_pruned_height(&self) -> Result { + trace!("get_pruned_height"); + let start = Instant::now(); + + let maybe_record = sqlx::query!( + r#" + SELECT last_pruned_height FROM pruning + "# + ) + .fetch_optional(&self.connection_pool) + .await?; + + log_db_operation_time("get_pruned_height", start); + + if let Some(row) = maybe_record { + Ok(row.last_pruned_height) + } else { + Ok(-1) + } + } } // make those generic over executor so that they could be performed over connection pool and a tx @@ -170,7 +250,8 @@ pub(crate) async fn insert_validator<'a, E>( where E: Executor<'a, Database = Sqlite>, { - trace!("insert validator"); + trace!("insert_validator"); + let start = Instant::now(); sqlx::query!( r#" @@ -183,6 +264,7 @@ where ) .execute(executor) .await?; + log_db_operation_time("insert_validator", start); Ok(()) } @@ -200,7 +282,8 @@ pub(crate) async fn insert_block<'a, E>( where E: Executor<'a, Database = Sqlite>, { - trace!("insert block"); + trace!("insert_block"); + let start = Instant::now(); sqlx::query!( r#" @@ -217,6 +300,7 @@ where ) .execute(executor) .await?; + log_db_operation_time("insert_block", start); Ok(()) } @@ -233,7 +317,8 @@ pub(crate) async fn insert_precommit<'a, E>( where E: Executor<'a, Database = Sqlite>, { - trace!("insert precommit"); + trace!("insert_precommit"); + let start = Instant::now(); sqlx::query!( r#" @@ -249,6 +334,7 @@ where ) .execute(executor) .await?; + log_db_operation_time("insert_precommit", start); Ok(()) } @@ -270,7 +356,8 @@ pub(crate) async fn insert_transaction<'a, E>( where E: Executor<'a, Database = Sqlite>, { - trace!("insert transaction"); + trace!("insert_transaction"); + let start = Instant::now(); sqlx::query!( r#" @@ -298,6 +385,7 @@ where ) .execute(executor) .await?; + log_db_operation_time("insert_transaction", start); Ok(()) } @@ -313,7 +401,8 @@ pub(crate) async fn insert_message<'a, E>( where E: Executor<'a, Database = Sqlite>, { - trace!("insert message"); + trace!("insert_message"); + let start = Instant::now(); sqlx::query!( r#" @@ -330,6 +419,7 @@ where ) .execute(executor) .await?; + log_db_operation_time("insert_message", start); Ok(()) } @@ -343,10 +433,100 @@ where E: Executor<'a, Database = Sqlite>, { trace!("update_last_processed"); + let start = Instant::now(); sqlx::query!("UPDATE metadata SET last_processed_height = ?", height) .execute(executor) .await?; + log_db_operation_time("update_last_processed", start); + + Ok(()) +} + +#[instrument(skip(executor))] +pub(crate) async fn update_last_pruned<'a, E>(height: i64, executor: E) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Sqlite>, +{ + trace!("update_last_pruned"); + let start = Instant::now(); + + sqlx::query!("UPDATE pruning SET last_pruned_height = ?", height) + .execute(executor) + .await?; + log_db_operation_time("update_last_pruned", start); + + Ok(()) +} + +pub(crate) async fn prune_blocks<'a, E>(oldest_to_keep: i64, executor: E) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Sqlite>, +{ + trace!("prune_blocks"); + let start = Instant::now(); + + sqlx::query!("DELETE FROM block WHERE height < ?", oldest_to_keep) + .execute(executor) + .await?; + log_db_operation_time("prune_blocks", start); + + Ok(()) +} + +pub(crate) async fn prune_pre_commits<'a, E>( + oldest_to_keep: i64, + executor: E, +) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Sqlite>, +{ + trace!("prune_pre_commits"); + let start = Instant::now(); + + sqlx::query!("DELETE FROM pre_commit WHERE height < ?", oldest_to_keep) + .execute(executor) + .await?; + log_db_operation_time("prune_pre_commits", start); + + Ok(()) +} + +pub(crate) async fn prune_transactions<'a, E>( + oldest_to_keep: i64, + executor: E, +) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Sqlite>, +{ + trace!("prune_transactions"); + let start = Instant::now(); + + sqlx::query!( + "DELETE FROM \"transaction\" WHERE height < ?", + oldest_to_keep + ) + .execute(executor) + .await?; + log_db_operation_time("prune_transactions", start); + + Ok(()) +} + +pub(crate) async fn prune_messages<'a, E>( + oldest_to_keep: i64, + executor: E, +) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Sqlite>, +{ + trace!("prune_messages"); + let start = Instant::now(); + + sqlx::query!("DELETE FROM message WHERE height < ?", oldest_to_keep) + .execute(executor) + .await?; + log_db_operation_time("prune_messages", start); Ok(()) } diff --git a/common/nyxd-scraper/src/storage/mod.rs b/common/nyxd-scraper/src/storage/mod.rs index 564f74d352..c9ecd64d40 100644 --- a/common/nyxd-scraper/src/storage/mod.rs +++ b/common/nyxd-scraper/src/storage/mod.rs @@ -5,7 +5,8 @@ use crate::block_processor::types::{FullBlockInformation, ParsedTransactionRespo use crate::error::ScraperError; use crate::storage::manager::{ insert_block, insert_message, insert_precommit, insert_transaction, insert_validator, - update_last_processed, StorageManager, + prune_blocks, prune_messages, prune_pre_commits, prune_transactions, update_last_processed, + update_last_pruned, StorageManager, }; use crate::storage::models::{CommitSignature, Validator}; use sqlx::types::time::OffsetDateTime; @@ -15,6 +16,7 @@ use std::path::Path; use tendermint::block::{Commit, CommitSig}; use tendermint::Block; use tendermint_rpc::endpoint::validators; +use tokio::time::Instant; use tracing::{debug, error, info, instrument, trace, warn}; mod helpers; @@ -28,6 +30,19 @@ pub struct ScraperStorage { pub(crate) manager: StorageManager, } +pub(crate) fn log_db_operation_time(op_name: &str, start_time: Instant) { + let elapsed = start_time.elapsed(); + let formatted = humantime::format_duration(elapsed); + + match elapsed.as_millis() { + v if v > 10000 => error!("{op_name} took {formatted} to execute"), + v if v > 1000 => warn!("{op_name} took {formatted} to execute"), + v if v > 100 => info!("{op_name} took {formatted} to execute"), + v if v > 10 => debug!("{op_name} took {formatted} to execute"), + _ => trace!("{op_name} took {formatted} to execute"), + } +} + impl ScraperStorage { #[instrument] pub async fn init + Debug>(database_path: P) -> Result { @@ -65,6 +80,32 @@ impl ScraperStorage { Ok(storage) } + #[instrument(skip(self))] + pub async fn prune_storage( + &self, + oldest_to_keep: u32, + current_height: u32, + ) -> Result<(), ScraperError> { + let start = Instant::now(); + + let mut tx = self.begin_processing_tx().await?; + + prune_messages(oldest_to_keep.into(), &mut tx).await?; + prune_transactions(oldest_to_keep.into(), &mut tx).await?; + prune_pre_commits(oldest_to_keep.into(), &mut tx).await?; + prune_blocks(oldest_to_keep.into(), &mut tx).await?; + update_last_pruned(current_height.into(), &mut tx).await?; + + let commit_start = Instant::now(); + tx.commit() + .await + .map_err(|source| ScraperError::StorageTxCommitFailure { source })?; + log_db_operation_time("committing pruning tx", commit_start); + + log_db_operation_time("pruning storage", start); + Ok(()) + } + #[instrument(skip_all)] pub async fn begin_processing_tx(&self) -> Result { debug!("starting storage tx"); @@ -75,6 +116,10 @@ impl ScraperStorage { .map_err(|source| ScraperError::StorageTxBeginFailure { source }) } + pub async fn lowest_block_height(&self) -> Result, ScraperError> { + Ok(self.manager.get_lowest_block().await?) + } + pub async fn get_first_block_height_after( &self, time: OffsetDateTime, @@ -155,6 +200,10 @@ impl ScraperStorage { pub async fn get_last_processed_height(&self) -> Result { Ok(self.manager.get_last_processed_height().await?) } + + pub async fn get_pruned_height(&self) -> Result { + Ok(self.manager.get_pruned_height().await?) + } } pub async fn persist_block( diff --git a/nym-validator-rewarder/Cargo.toml b/nym-validator-rewarder/Cargo.toml index 7100f87fb8..d08c3e8bed 100644 --- a/nym-validator-rewarder/Cargo.toml +++ b/nym-validator-rewarder/Cargo.toml @@ -31,7 +31,7 @@ humantime = "2.1.0" humantime-serde.workspace = true # internal -nym-bin-common = { path = "../common/bin-common", features = ["output_format"] } +nym-bin-common = { path = "../common/bin-common", features = ["output_format", "basic_tracing"] } nym-config = { path = "../common/config" } nym-coconut = { path = "../common/nymcoconut" } nym-crypto = { path = "../common/crypto", features = ["asymmetric"] } @@ -39,8 +39,8 @@ nym-credentials = { path = "../common/credentials" } nym-network-defaults = { path = "../common/network-defaults" } nym-task = { path = "../common/task" } nym-validator-client = { path = "../common/client-libs/validator-client" } -nym-coconut-dkg-common = { path = "../common/cosmwasm-smart-contracts/coconut-dkg"} -nym-coconut-bandwidth-contract-common = { path = "../common/cosmwasm-smart-contracts/coconut-bandwidth-contract"} +nym-coconut-dkg-common = { path = "../common/cosmwasm-smart-contracts/coconut-dkg" } +nym-coconut-bandwidth-contract-common = { path = "../common/cosmwasm-smart-contracts/coconut-bandwidth-contract" } nyxd-scraper = { path = "../common/nyxd-scraper" } [build-dependencies] diff --git a/nym-validator-rewarder/src/cli/init.rs b/nym-validator-rewarder/src/cli/init.rs index 071bc0af97..2c10d18506 100644 --- a/nym-validator-rewarder/src/cli/init.rs +++ b/nym-validator-rewarder/src/cli/init.rs @@ -50,7 +50,7 @@ pub(crate) fn execute(args: Args) -> Result<(), NymRewarderError> { }; let config = Config::new(args.mnemonic, websocket, nyxd).with_override(args.config_override); - config.ensure_is_valid()?; + config.validate()?; config .save_to_path(&path) diff --git a/nym-validator-rewarder/src/cli/mod.rs b/nym-validator-rewarder/src/cli/mod.rs index 9a68701f7a..9fd6cf4ba8 100644 --- a/nym-validator-rewarder/src/cli/mod.rs +++ b/nym-validator-rewarder/src/cli/mod.rs @@ -110,7 +110,7 @@ fn try_load_current_config(custom_path: &Option) -> Result) -> Result PathBuf { @@ -100,7 +104,10 @@ impl Config { rewarding: Rewarding::default(), block_signing: Default::default(), issuance_monitor: IssuanceMonitor::default(), - nyxd_scraper: NyxdScraper { websocket_url }, + nyxd_scraper: NyxdScraper { + websocket_url, + pruning: Default::default(), + }, base: Base { upstream_nyxd: nyxd_url, mnemonic, @@ -114,11 +121,13 @@ impl Config { websocket_url: self.nyxd_scraper.websocket_url.clone(), rpc_url: self.base.upstream_nyxd.clone(), database_path: self.storage_paths.nyxd_scraper.clone(), + pruning_options: self.nyxd_scraper.pruning, } } - pub fn ensure_is_valid(&self) -> Result<(), NymRewarderError> { - self.rewarding.ratios.ensure_is_valid()?; + pub fn validate(&self) -> Result<(), NymRewarderError> { + self.rewarding.ratios.validate()?; + self.nyxd_scraper.validate(self.rewarding.epoch_duration)?; Ok(()) } @@ -140,7 +149,7 @@ impl Config { source, } })?; - loaded.ensure_is_valid()?; + loaded.validate()?; loaded.save_path = Some(path.to_path_buf()); debug!("loaded config file from {}", path.display()); Ok(loaded) @@ -223,7 +232,7 @@ impl Default for RewardingRatios { } impl RewardingRatios { - pub fn ensure_is_valid(&self) -> Result<(), NymRewarderError> { + pub fn validate(&self) -> Result<(), NymRewarderError> { if self.block_signing + self.credential_verification + self.credential_issuance != 1.0 { return Err(NymRewarderError::InvalidRewardingRatios { ratios: *self }); } @@ -235,9 +244,44 @@ impl RewardingRatios { pub struct NyxdScraper { /// Url to the websocket endpoint of a validator, for example `wss://rpc.nymtech.net/websocket` pub websocket_url: Url, + + /// Defines the pruning options, if applicable, to be used by the underlying scraper. + // if the value is missing, use `nothing` pruning as this was the past behaviour + #[serde(default = "PruningOptions::nothing")] + pub pruning: PruningOptions, // TODO: debug with everything that's currently hardcoded in the scraper } +impl NyxdScraper { + pub fn validate(&self, epoch_duration: Duration) -> Result<(), NymRewarderError> { + // basic, sanity check, of pruning + self.pruning.validate()?; + + if self.pruning.strategy.is_nothing() { + return Ok(()); + } + + // rewarder-specific validation: + if self.pruning.strategy.is_everything() { + return Err(NymRewarderError::EverythingPruningStrategy); + } + + if self.pruning.strategy.is_custom() { + let min_to_keep = + (epoch_duration.as_secs_f32() / TYPICAL_BLOCK_TIME * 1.5).ceil() as u32; + + if self.pruning.strategy_keep_recent() < min_to_keep { + return Err(NymRewarderError::TooSmallKeepRecent { + min_to_keep, + keep_recent: self.pruning.strategy_keep_recent(), + }); + } + } + + Ok(()) + } +} + #[derive(Debug, Clone, Deserialize, Serialize)] pub struct BlockSigning { /// Specifies whether rewards for block signing is enabled. diff --git a/nym-validator-rewarder/src/config/template.rs b/nym-validator-rewarder/src/config/template.rs index 7e10b44d99..841e5e2536 100644 --- a/nym-validator-rewarder/src/config/template.rs +++ b/nym-validator-rewarder/src/config/template.rs @@ -74,4 +74,14 @@ whitelist = [ [nyxd_scraper] # Url to the websocket endpoint of a validator, for example `wss://rpc.nymtech.net/websocket` websocket_url = '{{ nyxd_scraper.websocket_url }}' + +# default: the last 362880 states are kept, pruning at 10 block intervals +# nothing: all historic states will be saved, nothing will be deleted (i.e. archiving) +# everything: 2 latest states will be kept; pruning at 10 block intervals. +# custom: allow pruning options to be manually specified through 'pruning.keep_recent' and 'pruning.interval' +pruning.strategy = '{{ nyxd_scraper.pruning.strategy }}' + +# These are applied if and only if the pruning strategy is custom. +pruning.keep_recent = {{ nyxd_scraper.pruning.keep_recent }} +pruning.interval = {{ nyxd_scraper.pruning.interval }} "#; diff --git a/nym-validator-rewarder/src/error.rs b/nym-validator-rewarder/src/error.rs index 2c78e61fbb..cfff19d0b5 100644 --- a/nym-validator-rewarder/src/error.rs +++ b/nym-validator-rewarder/src/error.rs @@ -166,6 +166,12 @@ pub enum NymRewarderError { #[error("there were no validators to reward in this epoch")] NoValidatorsToReward, + + #[error("the current pruning strategy is set to 'everything' - we won't have any block data for rewarding")] + EverythingPruningStrategy, + + #[error("pruning.keep_recent must not be smaller than {min_to_keep}. got: {keep_recent}")] + TooSmallKeepRecent { min_to_keep: u32, keep_recent: u32 }, } #[derive(Debug)] diff --git a/nym-validator-rewarder/src/main.rs b/nym-validator-rewarder/src/main.rs index eb513124d1..9b6654c3d6 100644 --- a/nym-validator-rewarder/src/main.rs +++ b/nym-validator-rewarder/src/main.rs @@ -20,7 +20,7 @@ mod rewarder; async fn main() -> anyhow::Result<()> { // std::env::set_var( // "RUST_LOG", - // "debug,tendermint_rpc=warn,h2=warn,hyper=warn,rustls=warn,reqwest=warn,tungstenite=warn,async_tungstenite=warn", + // "trace,handlebars=warn,tendermint_rpc=warn,h2=warn,hyper=warn,rustls=warn,reqwest=warn,tungstenite=warn,async_tungstenite=warn,tokio_util=warn,tokio_tungstenite=warn,tokio-util=warn", // ); let args = Cli::parse(); diff --git a/nym-validator-rewarder/src/rewarder/block_signing/mod.rs b/nym-validator-rewarder/src/rewarder/block_signing/mod.rs index b443364641..d9747a9f7a 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, trace, warn}; +use tracing::{debug, error, info, trace, warn}; pub(crate) mod types; @@ -28,7 +28,7 @@ impl EpochSigning { height_range: Range, ) -> Result, NymRewarderError> { for height in height_range { - trace!("attempting to get precommit for {address} at height {height}"); + trace!("attempting to get pre-commit for {address} at height {height}"); if let Some(precommit) = self .nyxd_scraper .storage @@ -121,7 +121,7 @@ impl EpochSigning { .get_voting_power(&validator.consensus_address, vp_range.clone()) .await? else { - warn!("failed to obtain voting power of {addr}"); + error!("failed to obtain voting power for validator {addr} for any block between heights {vp_range:?} - there were no stored pre-commits for that validator."); continue; };