From 027bd85200d4c7b181fe797aa033412265f51019 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Fri, 4 Jul 2025 11:23:25 +0100 Subject: [PATCH] changed error types to make modules dyn compatible --- .../src/block_processor/mod.rs | 47 +++++++++---------- common/nyxd-scraper-shared/src/error.rs | 34 ++++---------- common/nyxd-scraper-shared/src/helpers.rs | 4 +- common/nyxd-scraper-shared/src/lib.rs | 2 +- .../src/modules/block_module.rs | 14 +----- .../src/modules/msg_module.rs | 16 +------ .../src/modules/tx_module.rs | 14 +----- common/nyxd-scraper-shared/src/scraper/mod.rs | 9 ---- common/nyxd-scraper-shared/src/storage/mod.rs | 43 ++++++++++------- 9 files changed, 65 insertions(+), 118 deletions(-) diff --git a/common/nyxd-scraper-shared/src/block_processor/mod.rs b/common/nyxd-scraper-shared/src/block_processor/mod.rs index ef53b2c008..a6c37ac984 100644 --- a/common/nyxd-scraper-shared/src/block_processor/mod.rs +++ b/common/nyxd-scraper-shared/src/block_processor/mod.rs @@ -12,7 +12,6 @@ use crate::PruningOptions; use futures::StreamExt; use std::cmp::max; use std::collections::{BTreeMap, HashSet, VecDeque}; -use std::fmt::Display; use std::ops::{Add, Range}; use std::sync::Arc; use std::time::Duration; @@ -105,8 +104,6 @@ pub struct BlockProcessor { impl BlockProcessor where S: NyxdScraperStorage, - S::Error: 'static, - ScraperError: From<::Error>, { pub async fn new( config: BlockProcessorConfig, @@ -178,30 +175,28 @@ where persist_block(&full_info, &mut tx, self.config.store_precommits).await?; - todo!(); + // let the modules do whatever they want + // the ones wanting the full block: + for block_module in &mut self.block_modules { + block_module.handle_block(&full_info, &mut tx).await?; + } - // // let the modules do whatever they want - // // the ones wanting the full block: - // for block_module in &mut self.block_modules { - // block_module.handle_block(&full_info, &mut tx).await?; - // } - // - // // the ones wanting transactions: - // for block_tx in full_info.transactions { - // for tx_module in &mut self.tx_modules { - // tx_module.handle_tx(&block_tx, &mut tx).await?; - // } - // // the ones concerned with individual messages - // for (index, msg) in block_tx.tx.body.messages.iter().enumerate() { - // for msg_module in &mut self.msg_modules { - // if msg.type_url == msg_module.type_url() { - // msg_module - // .handle_msg(index, msg, &block_tx, &mut tx) - // .await? - // } - // } - // } - // } + // the ones wanting transactions: + for block_tx in full_info.transactions { + for tx_module in &mut self.tx_modules { + tx_module.handle_tx(&block_tx, &mut tx).await?; + } + // the ones concerned with individual messages + for (index, msg) in block_tx.tx.body.messages.iter().enumerate() { + for msg_module in &mut self.msg_modules { + if msg.type_url == msg_module.type_url() { + msg_module + .handle_msg(index, msg, &block_tx, &mut tx) + .await? + } + } + } + } let commit_start = Instant::now(); tx.commit().await.map_err(ScraperError::tx_commit_failure)?; diff --git a/common/nyxd-scraper-shared/src/error.rs b/common/nyxd-scraper-shared/src/error.rs index 2394760b02..14cdeddb98 100644 --- a/common/nyxd-scraper-shared/src/error.rs +++ b/common/nyxd-scraper-shared/src/error.rs @@ -4,6 +4,7 @@ use crate::block_processor::pruning::{ EVERYTHING_PRUNING_INTERVAL, EVERYTHING_PRUNING_KEEP_RECENT, }; +use crate::storage::NyxdScraperStorageError; use tendermint::Hash; use thiserror::Error; use tokio::sync::mpsc::error::SendError; @@ -11,7 +12,7 @@ use tokio::sync::mpsc::error::SendError; #[derive(Debug, Error)] pub enum ScraperError { #[error("experienced internal database error: {0}")] - InternalDatabaseError(Box), + InternalDatabaseError(#[from] NyxdScraperStorageError), // #[error("failed to perform startup SQL migration: {0}")] // StartupMigrationFailure(#[from] Box), @@ -103,13 +104,13 @@ pub enum ScraperError { #[error("failed to begin storage tx: {source}")] StorageTxBeginFailure { #[source] - source: Box, + source: NyxdScraperStorageError, }, #[error("failed to commit storage tx: {source}")] StorageTxCommitFailure { #[source] - source: Box, + source: NyxdScraperStorageError, }, #[error("failed to send on a closed channel")] @@ -148,29 +149,14 @@ pub enum ScraperError { } impl ScraperError { - pub fn internal_database(error: E) -> ScraperError - where - E: std::error::Error + Send + Sync + 'static, - { - ScraperError::InternalDatabaseError(Box::new(error)) + pub fn tx_begin_failure(source: NyxdScraperStorageError) -> ScraperError +where { + ScraperError::StorageTxBeginFailure { source } } - pub fn tx_begin_failure(error: E) -> ScraperError - where - E: std::error::Error + Send + Sync + 'static, - { - ScraperError::StorageTxBeginFailure { - source: Box::new(error), - } - } - - pub fn tx_commit_failure(error: E) -> ScraperError - where - E: std::error::Error + Send + Sync + 'static, - { - ScraperError::StorageTxCommitFailure { - source: Box::new(error), - } + pub fn tx_commit_failure(source: NyxdScraperStorageError) -> ScraperError +where { + ScraperError::StorageTxCommitFailure { source } } } diff --git a/common/nyxd-scraper-shared/src/helpers.rs b/common/nyxd-scraper-shared/src/helpers.rs index d11b25a2f3..7c3ea91e6a 100644 --- a/common/nyxd-scraper-shared/src/helpers.rs +++ b/common/nyxd-scraper-shared/src/helpers.rs @@ -14,7 +14,7 @@ pub(crate) fn tx_hash>(raw_tx: M) -> Hash { Hash::Sha256(Sha256::digest(raw_tx).into()) } -pub(crate) fn validator_pubkey_to_bech32(pubkey: PublicKey) -> Result { +pub fn validator_pubkey_to_bech32(pubkey: PublicKey) -> Result { // TODO: this one seem to attach additional prefix to they pubkeys, is that what we want instead maybe? // Ok(pubkey.to_bech32(BECH32_CONESNSUS_PUBKEY_PREFIX)) AccountId::new(BECH32_CONESNSUS_PUBKEY_PREFIX, &pubkey.to_bytes()) @@ -30,7 +30,7 @@ pub(crate) fn tx_gas_sum(txs: &[ParsedTransactionResponse]) -> i64 { txs.iter().map(|tx| tx.tx_result.gas_used).sum() } -pub(crate) fn validator_info( +pub fn validator_info( id: account::Id, validators: &validators::Response, ) -> Result<&validator::Info, ScraperError> { diff --git a/common/nyxd-scraper-shared/src/lib.rs b/common/nyxd-scraper-shared/src/lib.rs index 89e165dc4d..d0d97c249a 100644 --- a/common/nyxd-scraper-shared/src/lib.rs +++ b/common/nyxd-scraper-shared/src/lib.rs @@ -5,7 +5,7 @@ pub(crate) mod block_processor; pub(crate) mod block_requester; pub mod constants; pub mod error; -pub(crate) mod helpers; +pub mod helpers; pub mod modules; pub(crate) mod rpc_client; pub(crate) mod scraper; diff --git a/common/nyxd-scraper-shared/src/modules/block_module.rs b/common/nyxd-scraper-shared/src/modules/block_module.rs index 6c30485083..1ea3c2899d 100644 --- a/common/nyxd-scraper-shared/src/modules/block_module.rs +++ b/common/nyxd-scraper-shared/src/modules/block_module.rs @@ -3,7 +3,7 @@ use crate::block_processor::types::FullBlockInformation; use crate::error::ScraperError; -use crate::storage::NyxdScraperStorage; +use crate::storage::NyxdScraperTransaction; use async_trait::async_trait; #[async_trait] @@ -11,16 +11,6 @@ pub trait BlockModule { async fn handle_block( &mut self, block: &FullBlockInformation, - storage_tx: &mut (), + storage_tx: &mut dyn NyxdScraperTransaction, ) -> Result<(), ScraperError>; - - /* - async fn handle_block( - &mut self, - block: &FullBlockInformation, - storage_tx: &mut S::StorageTransaction, - ) -> Result<(), ScraperError> - where - S: NyxdScraperStorage; - */ } diff --git a/common/nyxd-scraper-shared/src/modules/msg_module.rs b/common/nyxd-scraper-shared/src/modules/msg_module.rs index f30556ce07..60f53b1553 100644 --- a/common/nyxd-scraper-shared/src/modules/msg_module.rs +++ b/common/nyxd-scraper-shared/src/modules/msg_module.rs @@ -3,7 +3,7 @@ use crate::block_processor::types::ParsedTransactionResponse; use crate::error::ScraperError; -use crate::storage::NyxdScraperStorage; +use crate::storage::NyxdScraperTransaction; use async_trait::async_trait; use cosmrs::Any; @@ -16,18 +16,6 @@ pub trait MsgModule { index: usize, msg: &Any, tx: &ParsedTransactionResponse, - storage_tx: (), + storage_tx: &mut dyn NyxdScraperTransaction, ) -> Result<(), ScraperError>; - - /* - async fn handle_msg( - &mut self, - index: usize, - msg: &Any, - tx: &ParsedTransactionResponse, - storage_tx: &mut S::StorageTransaction, - ) -> Result<(), ScraperError> - where - S: NyxdScraperStorage; - */ } diff --git a/common/nyxd-scraper-shared/src/modules/tx_module.rs b/common/nyxd-scraper-shared/src/modules/tx_module.rs index 3ea741eb95..8d2f5b22b1 100644 --- a/common/nyxd-scraper-shared/src/modules/tx_module.rs +++ b/common/nyxd-scraper-shared/src/modules/tx_module.rs @@ -3,7 +3,7 @@ use crate::block_processor::types::ParsedTransactionResponse; use crate::error::ScraperError; -use crate::storage::NyxdScraperStorage; +use crate::storage::NyxdScraperTransaction; use async_trait::async_trait; #[async_trait] @@ -11,16 +11,6 @@ pub trait TxModule { async fn handle_tx( &mut self, tx: &ParsedTransactionResponse, - storage_tx: &mut (), + storage_tx: &mut dyn NyxdScraperTransaction, ) -> Result<(), ScraperError>; - - /* - async fn handle_tx( - &mut self, - tx: &ParsedTransactionResponse, - storage_tx: &mut S::StorageTransaction, - ) -> Result<(), ScraperError> - where - S: NyxdScraperStorage; - */ } diff --git a/common/nyxd-scraper-shared/src/scraper/mod.rs b/common/nyxd-scraper-shared/src/scraper/mod.rs index 9dc4ea2ca4..9d630b91dd 100644 --- a/common/nyxd-scraper-shared/src/scraper/mod.rs +++ b/common/nyxd-scraper-shared/src/scraper/mod.rs @@ -62,9 +62,7 @@ pub struct NyxdScraperBuilder { impl NyxdScraperBuilder where S: NyxdScraperStorage + Send + Sync + 'static, - S::Error: Send + Sync + 'static, S::StorageTransaction: Send + Sync + 'static, - ScraperError: From<::Error>, { pub async fn build_and_start(self) -> Result, ScraperError> { let scraper = NyxdScraper::::new(self.config).await?; @@ -155,9 +153,7 @@ pub struct NyxdScraper { impl NyxdScraper where S: NyxdScraperStorage + Send + Sync + 'static, - S::Error: Send + Sync + 'static, S::StorageTransaction: Send + Sync + 'static, - ScraperError: From<::Error>, { pub fn builder(config: Config) -> NyxdScraperBuilder { NyxdScraperBuilder::new(config) @@ -178,11 +174,6 @@ where }) } - const TODO: &'static str = "maybe remove?"; - // pub fn storage(&self) -> ScraperStorage { - // self.storage.clone() - // } - fn start_tasks( &self, mut block_requester: BlockRequester, diff --git a/common/nyxd-scraper-shared/src/storage/mod.rs b/common/nyxd-scraper-shared/src/storage/mod.rs index aed8b815d7..1146cf8837 100644 --- a/common/nyxd-scraper-shared/src/storage/mod.rs +++ b/common/nyxd-scraper-shared/src/storage/mod.rs @@ -9,63 +9,71 @@ use std::path::PathBuf; use tendermint::block::Commit; use tendermint::Block; use tendermint_rpc::endpoint::validators; +use thiserror::Error; use tracing::warn; pub(crate) mod helpers; +// a workaround for needing associated type (which is a no-no in dynamic dispatch) +#[derive(Error, Debug)] +#[error(transparent)] +pub struct NyxdScraperStorageError(Box); + #[async_trait] pub trait NyxdScraperStorage: Clone + Sized { - type Error: Into + std::error::Error + Send + Sync; - type StorageTransaction: NyxdScraperTransaction; + type StorageTransaction: NyxdScraperTransaction; - async fn initialise(storage_path: &PathBuf) -> Result; + async fn initialise(storage_path: &PathBuf) -> Result; - async fn begin_processing_tx(&self) -> Result; + async fn begin_processing_tx( + &self, + ) -> Result; - async fn get_last_processed_height(&self) -> Result; + async fn get_last_processed_height(&self) -> Result; - async fn get_pruned_height(&self) -> Result; + async fn get_pruned_height(&self) -> Result; - async fn lowest_block_height(&self) -> Result, Self::Error>; + async fn lowest_block_height(&self) -> Result, NyxdScraperStorageError>; async fn prune_storage( &self, oldest_to_keep: u32, current_height: u32, - ) -> Result<(), Self::Error>; + ) -> Result<(), NyxdScraperStorageError>; } #[async_trait] pub trait NyxdScraperTransaction { - type Error: Into + std::error::Error + Send + Sync; - - async fn commit(mut self) -> Result<(), Self::Error>; + async fn commit(mut self) -> Result<(), NyxdScraperStorageError>; async fn persist_validators( &mut self, validators: &validators::Response, - ) -> Result<(), Self::Error>; + ) -> Result<(), NyxdScraperStorageError>; async fn persist_block_data( &mut self, block: &Block, total_gas: i64, - ) -> Result<(), Self::Error>; + ) -> Result<(), NyxdScraperStorageError>; async fn persist_commits( &mut self, commits: &Commit, validators: &validators::Response, - ) -> Result<(), Self::Error>; + ) -> Result<(), NyxdScraperStorageError>; - async fn persist_txs(&mut self, txs: &[ParsedTransactionResponse]) -> Result<(), Self::Error>; + async fn persist_txs( + &mut self, + txs: &[ParsedTransactionResponse], + ) -> Result<(), NyxdScraperStorageError>; async fn persist_messages( &mut self, txs: &[ParsedTransactionResponse], - ) -> Result<(), Self::Error>; + ) -> Result<(), NyxdScraperStorageError>; - async fn update_last_processed(&mut self, height: i64) -> Result<(), Self::Error>; + async fn update_last_processed(&mut self, height: i64) -> Result<(), NyxdScraperStorageError>; } pub async fn persist_block( @@ -75,7 +83,6 @@ pub async fn persist_block( ) -> Result<(), ScraperError> where Tx: NyxdScraperTransaction, - ScraperError: From<::Error>, { let total_gas = crate::helpers::tx_gas_sum(&block.transactions);