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

changed error types to make modules dyn compatible

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2025-07-04 11:23:25 +01:00
parent 08cff312af
commit 027bd85200
9 changed files with 65 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/nyxd-scraper-shared/src/block_processor/mod.rs
+21 -26
View File
@@ -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<S> {
impl<S> BlockProcessor<S>
where
S: NyxdScraperStorage,
S::Error: 'static,
ScraperError: From<<S as NyxdScraperStorage>::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)?;
common/nyxd-scraper-shared/src/error.rs
+10 -24
View File
@@ -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<dyn std::error::Error + Send + Sync>),
InternalDatabaseError(#[from] NyxdScraperStorageError),
// #[error("failed to perform startup SQL migration: {0}")]
// StartupMigrationFailure(#[from] Box<dyn std::error::Error + Send + Sync>),
@@ -103,13 +104,13 @@ pub enum ScraperError {
#[error("failed to begin storage tx: {source}")]
StorageTxBeginFailure {
#[source]
source: Box<dyn std::error::Error + Send + Sync>,
source: NyxdScraperStorageError,
},
#[error("failed to commit storage tx: {source}")]
StorageTxCommitFailure {
#[source]
source: Box<dyn std::error::Error + Send + Sync>,
source: NyxdScraperStorageError,
},
#[error("failed to send on a closed channel")]
@@ -148,29 +149,14 @@ pub enum ScraperError {
}
impl ScraperError {
pub fn internal_database<E>(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<E>(error: E) -> ScraperError
where
E: std::error::Error + Send + Sync + 'static,
{
ScraperError::StorageTxBeginFailure {
source: Box::new(error),
}
}
pub fn tx_commit_failure<E>(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 }
}
}
common/nyxd-scraper-shared/src/helpers.rs
+2 -2
View File
@@ -14,7 +14,7 @@ pub(crate) fn tx_hash<M: AsRef<[u8]>>(raw_tx: M) -> Hash {
Hash::Sha256(Sha256::digest(raw_tx).into())
}
pub(crate) fn validator_pubkey_to_bech32(pubkey: PublicKey) -> Result<AccountId, ScraperError> {
pub fn validator_pubkey_to_bech32(pubkey: PublicKey) -> Result<AccountId, ScraperError> {
// 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> {
common/nyxd-scraper-shared/src/lib.rs
+1 -1
View File
@@ -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;
common/nyxd-scraper-shared/src/modules/block_module.rs
+2 -12
View File
@@ -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<S>(
&mut self,
block: &FullBlockInformation,
storage_tx: &mut S::StorageTransaction,
) -> Result<(), ScraperError>
where
S: NyxdScraperStorage;
*/
}
common/nyxd-scraper-shared/src/modules/msg_module.rs
+2 -14
View File
@@ -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<S>(
&mut self,
index: usize,
msg: &Any,
tx: &ParsedTransactionResponse,
storage_tx: &mut S::StorageTransaction,
) -> Result<(), ScraperError>
where
S: NyxdScraperStorage;
*/
}
common/nyxd-scraper-shared/src/modules/tx_module.rs
+2 -12
View File
@@ -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<S>(
&mut self,
tx: &ParsedTransactionResponse,
storage_tx: &mut S::StorageTransaction,
) -> Result<(), ScraperError>
where
S: NyxdScraperStorage;
*/
}
common/nyxd-scraper-shared/src/scraper/mod.rs
-9
View File
@@ -62,9 +62,7 @@ pub struct NyxdScraperBuilder<S> {
impl<S> NyxdScraperBuilder<S>
where
S: NyxdScraperStorage + Send + Sync + 'static,
S::Error: Send + Sync + 'static,
S::StorageTransaction: Send + Sync + 'static,
ScraperError: From<<S as NyxdScraperStorage>::Error>,
{
pub async fn build_and_start(self) -> Result<NyxdScraper<S>, ScraperError> {
let scraper = NyxdScraper::<S>::new(self.config).await?;
@@ -155,9 +153,7 @@ pub struct NyxdScraper<S> {
impl<S> NyxdScraper<S>
where
S: NyxdScraperStorage + Send + Sync + 'static,
S::Error: Send + Sync + 'static,
S::StorageTransaction: Send + Sync + 'static,
ScraperError: From<<S as NyxdScraperStorage>::Error>,
{
pub fn builder(config: Config) -> NyxdScraperBuilder<S> {
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,
common/nyxd-scraper-shared/src/storage/mod.rs
+25 -18
View File
@@ -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<dyn std::error::Error + Send + Sync>);
#[async_trait]
pub trait NyxdScraperStorage: Clone + Sized {
type Error: Into<ScraperError> + std::error::Error + Send + Sync;
type StorageTransaction: NyxdScraperTransaction<Error = Self::Error>;
type StorageTransaction: NyxdScraperTransaction;
async fn initialise(storage_path: &PathBuf) -> Result<Self, Self::Error>;
async fn initialise(storage_path: &PathBuf) -> Result<Self, NyxdScraperStorageError>;
async fn begin_processing_tx(&self) -> Result<Self::StorageTransaction, Self::Error>;
async fn begin_processing_tx(
&self,
) -> Result<Self::StorageTransaction, NyxdScraperStorageError>;
async fn get_last_processed_height(&self) -> Result<u32, Self::Error>;
async fn get_last_processed_height(&self) -> Result<u32, NyxdScraperStorageError>;
async fn get_pruned_height(&self) -> Result<u32, Self::Error>;
async fn get_pruned_height(&self) -> Result<u32, NyxdScraperStorageError>;
async fn lowest_block_height(&self) -> Result<Option<i64>, Self::Error>;
async fn lowest_block_height(&self) -> Result<Option<i64>, 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<ScraperError> + 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<Tx>(
@@ -75,7 +83,6 @@ pub async fn persist_block<Tx>(
) -> Result<(), ScraperError>
where
Tx: NyxdScraperTransaction,
ScraperError: From<<Tx as NyxdScraperTransaction>::Error>,
{
let total_gas = crate::helpers::tx_gas_sum(&block.transactions);