Merge pull request #4564 from nymtech/feature/nyxd-scraper-pruning

Feature/nyxd scraper pruning

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",

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]

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 = "<strategy>"` 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.

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<Notify>,
     last_processed_height: u32,
+    last_pruned_height: u32,
     last_processed_at: Instant,
     pending_sync: PendingSync,
     queued_blocks: BTreeMap<u32, BlockToProcess>,
@@ -62,6 +66,7 @@ pub struct BlockProcessor {
 
 impl BlockProcessor {
     pub async fn new(
+        pruning_options: PruningOptions,
         cancel: CancellationToken,
         synced: Arc<Notify>,
         incoming: UnboundedReceiver<BlockToProcess>,
@@ -70,11 +75,17 @@ impl BlockProcessor {
         rpc_client: RpcClient,
     ) -> Result<Self, ScraperError> {
         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;

common/nyxd-scraper/src/block_processor/pruning.rs

@@ -0,0 +1,122 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// 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(),
+        }
+    }
+}

common/nyxd-scraper/src/error.rs

@@ -1,6 +1,9 @@
 // Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
 // 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<T> From<SendError<T>> for ScraperError {

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;

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<Self, ScraperError> {
+        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,

common/nyxd-scraper/src/storage/manager.rs

@@ -1,9 +1,11 @@
 // Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
 // 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<Option<i64>, 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<Option<i64>, 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<Option<i64>, 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<i32, sqlx::Error> {
+        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<Option<CommitSignature>, 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<Vec<Validator>, 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<Vec<Validator>, 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<i64, sqlx::Error> {
+        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<i64, sqlx::Error> {
+        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(())
 }

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<P: AsRef<Path> + Debug>(database_path: P) -> Result<Self, ScraperError> {
@@ -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<StorageTransaction, ScraperError> {
         debug!("starting storage tx");
@@ -75,6 +116,10 @@ impl ScraperStorage {
             .map_err(|source| ScraperError::StorageTxBeginFailure { source })
     }
 
+    pub async fn lowest_block_height(&self) -> Result<Option<i64>, 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<i64, ScraperError> {
         Ok(self.manager.get_last_processed_height().await?)
     }
+
+    pub async fn get_pruned_height(&self) -> Result<i64, ScraperError> {
+        Ok(self.manager.get_pruned_height().await?)
+    }
 }
 
 pub async fn persist_block(

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]

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)

nym-validator-rewarder/src/cli/mod.rs

@@ -110,7 +110,7 @@ fn try_load_current_config(custom_path: &Option<PathBuf>) -> Result<Config, NymR
     );
 
     if let Ok(cfg) = Config::read_from_toml_file(&config_path) {
-        cfg.ensure_is_valid()?;
+        cfg.validate()?;
         return Ok(cfg);
     }
 
@@ -122,7 +122,7 @@ fn try_load_current_config(custom_path: &Option<PathBuf>) -> Result<Config, NymR
         );
         err
     })?;
-    config.ensure_is_valid()?;
+    config.validate()?;
     Ok(config)
 }
 

nym-validator-rewarder/src/config/mod.rs

@@ -10,6 +10,7 @@ use nym_config::{
     DEFAULT_CONFIG_DIR, DEFAULT_CONFIG_FILENAME, DEFAULT_DATA_DIR, NYM_DIR,
 };
 use nym_validator_client::nyxd::{AccountId, Coin};
+use nyxd_scraper::PruningOptions;
 use serde::{Deserialize, Serialize};
 use serde_with::{serde_as, DisplayFromStr};
 use std::io;
@@ -34,6 +35,9 @@ const DEFAULT_MONITOR_RUN_INTERVAL: Duration = Duration::from_secs(10 * 60);
 const DEFAULT_MONITOR_MIN_VALIDATE: usize = 10;
 const DEFAULT_MONITOR_SAMPLING_RATE: f64 = 0.10;
 
+// 'worst' case scenario
+pub const TYPICAL_BLOCK_TIME: f32 = 5.;
+
 /// Get default path to rewarder's config directory.
 /// It should get resolved to `$HOME/.nym/validators-rewarder/config`
 pub fn default_config_directory() -> 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.

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 }}
 "#;

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)]

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();

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<i64>,
     ) -> Result<Option<i64>, 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;
             };