diff --git a/common/nyxd-scraper-shared/examples/watcher.rs b/common/nyxd-scraper-shared/examples/watcher.rs new file mode 100644 index 0000000000..3d72c8bb5a --- /dev/null +++ b/common/nyxd-scraper-shared/examples/watcher.rs @@ -0,0 +1,64 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use nyxd_scraper_shared::error::ScraperError; +use nyxd_scraper_shared::storage::FullBlockInformation; +use nyxd_scraper_shared::watcher::{NyxdWatcher, WatcherConfig}; +use nyxd_scraper_shared::{BlockModule, ParsedTransactionResponse, TxModule}; + +struct FancyBlockModule; + +struct FancyTxModule; + +#[async_trait::async_trait] +impl BlockModule for FancyBlockModule { + async fn handle_block(&mut self, block: &FullBlockInformation) -> Result<(), ScraperError> { + println!("🚀 got new block for height {}", block.block.header.height); + + // should be false + println!("results scraped: {}", block.results.is_some()); + // should be false + println!("validators scraped: {}", block.validators.is_some()); + // should be true + println!("transactions scraped: {}", block.transactions.is_some()); + + println!(); + + Ok(()) + } +} + +#[async_trait::async_trait] +impl TxModule for FancyTxModule { + async fn handle_tx(&mut self, tx: &ParsedTransactionResponse) -> Result<(), ScraperError> { + println!( + "✨ got new tx for height {}: {} ({} msgs)", + tx.block.header.height, + tx.hash, + tx.parsed_messages.len() + ); + + Ok(()) + } +} + +#[tokio::main] +async fn main() -> eyre::Result<()> { + let cfg = WatcherConfig { + websocket_url: "wss://rpc.nymtech.net/websocket".parse()?, + rpc_url: "https://rpc.nymtech.net".parse()?, + }; + + let watcher = NyxdWatcher::builder(cfg) + .with_block_module(FancyBlockModule) + .with_tx_module(FancyTxModule) + .build_and_start() + .await?; + + // run for 30s before shutting down + tokio::time::sleep(std::time::Duration::from_secs(30)).await; + + watcher.stop().await; + + Ok(()) +} diff --git a/common/nyxd-scraper-shared/src/block_processor/ephemeral_storage.rs b/common/nyxd-scraper-shared/src/block_processor/ephemeral_storage.rs new file mode 100644 index 0000000000..435771d589 --- /dev/null +++ b/common/nyxd-scraper-shared/src/block_processor/ephemeral_storage.rs @@ -0,0 +1,94 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::storage::NyxdScraperStorageError; +use crate::{NyxdScraperStorage, NyxdScraperTransaction, ParsedTransactionResponse}; +use tendermint::Block; +use tendermint::block::Commit; +use tendermint_rpc::endpoint::validators::Response; +use thiserror::Error; + +#[derive(Clone)] +pub struct Ephemeral; + +#[derive(Debug, Error)] +#[error("no storage backend enabled")] +pub struct EphemeralStorageError; + +pub struct EphemeralTransaction; + +#[async_trait::async_trait] +impl NyxdScraperTransaction for EphemeralTransaction { + async fn commit(self) -> Result<(), NyxdScraperStorageError> { + Err(NyxdScraperStorageError::new(EphemeralStorageError)) + } + + async fn persist_validators(&mut self, _: &Response) -> Result<(), NyxdScraperStorageError> { + Err(NyxdScraperStorageError::new(EphemeralStorageError)) + } + + async fn persist_block_data( + &mut self, + _: &Block, + _: i64, + ) -> Result<(), NyxdScraperStorageError> { + Err(NyxdScraperStorageError::new(EphemeralStorageError)) + } + + async fn persist_commits( + &mut self, + _: &Commit, + _: &Response, + ) -> Result<(), NyxdScraperStorageError> { + Err(NyxdScraperStorageError::new(EphemeralStorageError)) + } + + async fn persist_txs( + &mut self, + _: &[ParsedTransactionResponse], + ) -> Result<(), NyxdScraperStorageError> { + Err(NyxdScraperStorageError::new(EphemeralStorageError)) + } + + async fn persist_messages( + &mut self, + _: &[ParsedTransactionResponse], + ) -> Result<(), NyxdScraperStorageError> { + Err(NyxdScraperStorageError::new(EphemeralStorageError)) + } + + async fn update_last_processed(&mut self, _: i64) -> Result<(), NyxdScraperStorageError> { + Err(NyxdScraperStorageError::new(EphemeralStorageError)) + } +} + +#[async_trait::async_trait] +impl NyxdScraperStorage for Ephemeral { + type StorageTransaction = EphemeralTransaction; + + async fn initialise(_: &str, _: &bool) -> Result { + Err(NyxdScraperStorageError::new(EphemeralStorageError)) + } + + async fn begin_processing_tx( + &self, + ) -> Result { + Err(NyxdScraperStorageError::new(EphemeralStorageError)) + } + + async fn get_last_processed_height(&self) -> Result { + Err(NyxdScraperStorageError::new(EphemeralStorageError)) + } + + async fn get_pruned_height(&self) -> Result { + Err(NyxdScraperStorageError::new(EphemeralStorageError)) + } + + async fn lowest_block_height(&self) -> Result, NyxdScraperStorageError> { + Err(NyxdScraperStorageError::new(EphemeralStorageError)) + } + + async fn prune_storage(&self, _: u32, _: u32) -> Result<(), NyxdScraperStorageError> { + Err(NyxdScraperStorageError::new(EphemeralStorageError)) + } +} diff --git a/common/nyxd-scraper-shared/src/block_processor/mod.rs b/common/nyxd-scraper-shared/src/block_processor/mod.rs index 0d30df40b7..d29129b790 100644 --- a/common/nyxd-scraper-shared/src/block_processor/mod.rs +++ b/common/nyxd-scraper-shared/src/block_processor/mod.rs @@ -2,13 +2,16 @@ // SPDX-License-Identifier: Apache-2.0 use crate::PruningOptions; +use crate::block_processor::ephemeral_storage::Ephemeral; use crate::block_processor::helpers::split_request_range; use crate::block_processor::types::BlockToProcess; use crate::block_requester::BlockRequest; use crate::error::ScraperError; use crate::modules::{BlockModule, MsgModule, TxModule}; -use crate::rpc_client::RpcClient; -use crate::storage::{NyxdScraperStorage, NyxdScraperTransaction, persist_block}; +use crate::rpc_client::{RetrievalConfig, RpcClient}; +use crate::storage::{ + FullBlockInformation, NyxdScraperStorage, NyxdScraperTransaction, persist_block, +}; use futures::StreamExt; use std::cmp::max; use std::collections::{BTreeMap, HashSet, VecDeque}; @@ -22,6 +25,7 @@ use tokio_stream::wrappers::UnboundedReceiverStream; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, instrument, trace, warn}; +mod ephemeral_storage; mod helpers; pub(crate) mod pruning; pub(crate) mod types; @@ -77,20 +81,148 @@ impl BlockProcessorConfig { } } -pub struct BlockProcessor { +pub struct BlockProcessorPersistence { config: BlockProcessorConfig, - cancel: CancellationToken, synced: Arc, - last_processed_height: u32, last_pruned_height: u32, + + storage: S, +} + +impl BlockProcessorPersistence +where + S: NyxdScraperStorage, +{ + pub(crate) async fn new( + config: BlockProcessorConfig, + synced: Arc, + storage: S, + ) -> Result { + let last_pruned = storage.get_pruned_height().await?; + let last_pruned_height = last_pruned.try_into().unwrap_or_default(); + + debug!(pruned_height = %last_pruned_height, "setting up block processor..."); + + Ok(Self { + config, + synced, + last_pruned_height, + storage, + }) + } + + #[must_use] + pub fn with_pruning(mut self, pruning_options: PruningOptions) -> Self { + self.config.pruning_options = pruning_options; + self + } + + async fn stored_last_processed_height(&self) -> Result { + let last_processed = self.storage.get_last_processed_height().await?; + let last_processed_height = last_processed.try_into().unwrap_or_default(); + Ok(last_processed_height) + } + + async fn persist_block( + &mut self, + full_info: &FullBlockInformation, + ) -> Result<(), ScraperError> { + // process the entire block as a transaction so that if anything fails, + // we wouldn't end up with a corrupted storage. + let mut tx = self + .storage + .begin_processing_tx() + .await + .map_err(ScraperError::tx_begin_failure)?; + + persist_block(full_info, &mut tx, self.config.store_precommits).await?; + + let commit_start = Instant::now(); + tx.commit().await.map_err(ScraperError::tx_commit_failure)?; + crate::storage::helpers::log_db_operation_time("committing processing tx", commit_start); + + let last_processed_height = full_info.block.header.height.value() as u32; + if let Err(err) = self.maybe_prune_storage(last_processed_height).await { + error!("failed to prune the storage: {err}"); + } + + Ok(()) + } + + #[instrument(skip(self))] + async fn prune_storage(&mut self, last_processed_height: u32) -> Result<(), ScraperError> { + let keep_recent = self.config.pruning_options.strategy_keep_recent(); + let last_to_keep = 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"), + 0 => trace!("no blocks to prune"), + v => debug!("approximately {v} blocks worth of data will be pruned"), + } + + if to_prune == 0 { + self.last_pruned_height = last_processed_height; + return Ok(()); + } + + self.storage + .prune_storage(last_to_keep, last_processed_height) + .await?; + + self.last_pruned_height = last_processed_height; + Ok(()) + } + + async fn maybe_prune_storage( + &mut self, + last_processed_height: u32, + ) -> Result<(), ScraperError> { + debug!("checking for storage pruning"); + + if self.config.pruning_options.strategy.is_nothing() { + trace!("the current pruning strategy is 'nothing'"); + return Ok(()); + } + + let interval = self.config.pruning_options.strategy_interval(); + if self.last_pruned_height + interval <= last_processed_height { + self.prune_storage(last_processed_height).await?; + } + + Ok(()) + } +} + +pub struct BlockProcessor { + cancel: CancellationToken, + last_processed_height: u32, last_processed_at: Instant, pending_sync: PendingSync, queued_blocks: BTreeMap, + /// Specifies how much data to actually retrieve per block + retrieval_config: RetrievalConfig, + rpc_client: RpcClient, incoming: UnboundedReceiverStream, block_requester: Sender, - storage: S, + persistence: Option>, // future work: rather than sending each msg to every msg module, // let them subscribe based on `type_url` inside the message itself @@ -105,108 +237,97 @@ impl BlockProcessor where S: NyxdScraperStorage, { - pub async fn new( - config: BlockProcessorConfig, + pub fn new( cancel: CancellationToken, - synced: Arc, incoming: UnboundedReceiver, block_requester: Sender, - storage: S, 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(); - - debug!(last_processed_height = %last_processed_height, pruned_height = %last_pruned_height, "setting up block processor..."); - - Ok(BlockProcessor { - config, + ) -> Self { + BlockProcessor { cancel, - synced, - last_processed_height, - last_pruned_height, + last_processed_height: Default::default(), last_processed_at: Instant::now(), pending_sync: Default::default(), queued_blocks: Default::default(), + retrieval_config: RetrievalConfig::default(), rpc_client, incoming: incoming.into(), block_requester, - storage, + persistence: None, block_modules: vec![], tx_modules: vec![], msg_modules: vec![], - }) + } } - pub fn with_pruning(mut self, pruning_options: PruningOptions) -> Self { - self.config.pruning_options = pruning_options; + #[must_use] + pub fn with_retrieval_config(mut self, retrieval_config: RetrievalConfig) -> Self { + self.retrieval_config = retrieval_config; self } + pub async fn with_persistence( + mut self, + persistence: BlockProcessorPersistence, + ) -> Result { + let last_processed_height = persistence.stored_last_processed_height().await?; + debug!(last_processed_height = %last_processed_height, "setting up block processor..."); + + self.persistence = Some(persistence); + Ok(self) + } + pub(super) async fn process_block( &mut self, block: BlockToProcess, ) -> Result<(), ScraperError> { info!("processing block at height {}", block.height); - let full_info = self.rpc_client.try_get_full_details(block).await?; + let full_info = self + .rpc_client + .try_get_full_details(block, self.retrieval_config) + .await?; - debug!( - "this block has {} transaction(s)", - full_info.transactions.len() - ); - for tx in &full_info.transactions { - debug!("{} has {} message(s)", tx.hash, tx.tx.body.messages.len()); - for (index, msg) in tx.tx.body.messages.iter().enumerate() { - debug!("{index}: {:?}", msg.type_url) + if let Some(tx_info) = &full_info.transactions { + debug!("this block has {} transaction(s)", tx_info.len()); + for tx in tx_info { + debug!("{} has {} message(s)", tx.hash, tx.tx.body.messages.len()); + for (index, msg) in tx.tx.body.messages.iter().enumerate() { + debug!("{index}: {:?}", msg.type_url) + } } } - // process the entire block as a transaction so that if anything fails, - // we won't end up with a corrupted storage. - let mut tx = self - .storage - .begin_processing_tx() - .await - .map_err(ScraperError::tx_begin_failure)?; - - persist_block(&full_info, &mut tx, self.config.store_precommits).await?; + // if we have enabled persistence, do try to store the block information + if let Some(persistence) = &mut self.persistence { + persistence.persist_block(&full_info).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?; + block_module.handle_block(&full_info).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 (assuming tx retrieval is enabled): + if let Some(tx_info) = &full_info.transactions { + for block_tx in tx_info { + for tx_module in &mut self.tx_modules { + tx_module.handle_tx(block_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).await? + } } } } } - let commit_start = Instant::now(); - tx.commit().await.map_err(ScraperError::tx_commit_failure)?; - crate::storage::helpers::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(()) } @@ -284,62 +405,6 @@ where Ok(()) } - #[instrument(skip(self))] - async fn prune_storage(&mut self) -> Result<(), ScraperError> { - let keep_recent = self.config.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"), - 0 => trace!("no blocks to prune"), - v => debug!("approximately {v} blocks worth of data will be pruned"), - } - - if to_prune == 0 { - self.last_pruned_height = self.last_processed_height; - 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.config.pruning_options.strategy.is_nothing() { - trace!("the current pruning strategy is 'nothing'"); - return Ok(()); - } - - let interval = self.config.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; @@ -379,8 +444,10 @@ where self.try_request_pending().await; - if self.pending_sync.is_empty() { - self.synced.notify_one(); + if let Some(persistence) = &self.persistence + && self.pending_sync.is_empty() + { + persistence.synced.notify_one(); } } @@ -410,7 +477,14 @@ where assert!(self.pending_sync.is_empty()); info!("attempting to run startup resync..."); - self.maybe_prune_storage().await?; + let Some(persistence) = self.persistence.as_mut() else { + // without data persistence, we're always starting from scratch + return Ok(()); + }; + + persistence + .maybe_prune_storage(self.last_processed_height) + .await?; let latest_block = self.rpc_client.current_block_height().await? as u32; info!("obtained latest block height: {latest_block}"); @@ -419,10 +493,10 @@ where info!("we have already processed some blocks in the past - attempting to resume..."); // in case we were offline for a while, // make sure we don't request blocks we'd have to prune anyway - let keep_recent = self.config.pruning_options.strategy_keep_recent(); + let keep_recent = persistence.config.pruning_options.strategy_keep_recent(); let last_to_keep = latest_block - keep_recent; - if !self.config.pruning_options.strategy.is_nothing() { + if !persistence.config.pruning_options.strategy.is_nothing() { self.last_processed_height = max(self.last_processed_height, last_to_keep); } @@ -440,7 +514,7 @@ where // this is the first time starting up if self.last_processed_height == 0 { info!("this is the first time starting up"); - let Some(starting_height) = self.config.explicit_starting_block_height else { + let Some(starting_height) = persistence.config.explicit_starting_block_height else { info!("no starting block height set - will use the default behaviour"); // nothing to do return Ok(()); @@ -451,7 +525,9 @@ where self.rpc_client.earliest_available_block_height().await? as u32; info!("earliest available block height: {earliest_available}"); - if earliest_available > starting_height && self.config.use_best_effort_start_height { + if earliest_available > starting_height + && persistence.config.use_best_effort_start_height + { error!("the earliest available block is higher than the desired starting height"); return Err(ScraperError::BlocksUnavailable { height: starting_height, diff --git a/common/nyxd-scraper-shared/src/block_processor/types.rs b/common/nyxd-scraper-shared/src/block_processor/types.rs index 8bf184c0d9..044d9bcffc 100644 --- a/common/nyxd-scraper-shared/src/block_processor/types.rs +++ b/common/nyxd-scraper-shared/src/block_processor/types.rs @@ -2,8 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use crate::error::ScraperError; -use crate::helpers; -use std::collections::HashMap; +use std::collections::BTreeMap; use tendermint::{Block, Hash, abci, block, tx}; use tendermint_rpc::endpoint::{block as block_endpoint, block_results, validators}; use tendermint_rpc::event::{Event, EventData}; @@ -28,9 +27,9 @@ pub struct ParsedTransactionResponse { pub proof: Option, - pub parsed_messages: HashMap, + pub parsed_messages: BTreeMap, - pub parsed_message_urls: HashMap, + pub parsed_message_urls: BTreeMap, pub block: Block, } @@ -41,32 +40,13 @@ pub struct FullBlockInformation { pub block: Block, /// All of the emitted events alongside any tx results. - pub results: block_results::Response, + pub results: Option, /// Validator set for this particular block - pub validators: validators::Response, + pub validators: Option, /// Transaction results from this particular block - pub transactions: Vec, -} - -impl FullBlockInformation { - pub fn ensure_proposer(&self) -> Result<(), ScraperError> { - let block_proposer = self.block.header.proposer_address; - if !self - .validators - .validators - .iter() - .any(|v| v.address == block_proposer) - { - let proposer = helpers::validator_consensus_address(block_proposer)?; - return Err(ScraperError::BlockProposerNotInValidatorSet { - height: self.block.header.height.value() as u32, - proposer: proposer.to_string(), - }); - } - Ok(()) - } + pub transactions: Option>, } pub(crate) struct BlockToProcess { diff --git a/common/nyxd-scraper-shared/src/error.rs b/common/nyxd-scraper-shared/src/error.rs index 76723950be..41264896ee 100644 --- a/common/nyxd-scraper-shared/src/error.rs +++ b/common/nyxd-scraper-shared/src/error.rs @@ -172,16 +172,20 @@ pub enum ScraperError { #[source] error: base64::DecodeError, }, + + #[error("no modules configured for the chain watcher")] + NoModulesConfigured, + + #[error("no storage configured for the chain watcher")] + NoStorageConfigured, } impl ScraperError { - pub fn tx_begin_failure(source: NyxdScraperStorageError) -> ScraperError -where { + pub fn tx_begin_failure(source: NyxdScraperStorageError) -> ScraperError { ScraperError::StorageTxBeginFailure { source } } - pub fn tx_commit_failure(source: NyxdScraperStorageError) -> ScraperError -where { + pub fn tx_commit_failure(source: NyxdScraperStorageError) -> ScraperError { ScraperError::StorageTxCommitFailure { source } } } diff --git a/common/nyxd-scraper-shared/src/lib.rs b/common/nyxd-scraper-shared/src/lib.rs index 7d6b18509c..43afe4f83d 100644 --- a/common/nyxd-scraper-shared/src/lib.rs +++ b/common/nyxd-scraper-shared/src/lib.rs @@ -11,6 +11,8 @@ pub mod modules; pub(crate) mod rpc_client; pub(crate) mod scraper; pub mod storage; +pub(crate) mod subscriber; +pub mod watcher; pub use block_processor::pruning::{PruningOptions, PruningStrategy}; pub use block_processor::types::ParsedTransactionResponse; diff --git a/common/nyxd-scraper-shared/src/modules/block_module.rs b/common/nyxd-scraper-shared/src/modules/block_module.rs index 1ea3c2899d..af69a3fd26 100644 --- a/common/nyxd-scraper-shared/src/modules/block_module.rs +++ b/common/nyxd-scraper-shared/src/modules/block_module.rs @@ -3,14 +3,9 @@ use crate::block_processor::types::FullBlockInformation; use crate::error::ScraperError; -use crate::storage::NyxdScraperTransaction; use async_trait::async_trait; #[async_trait] pub trait BlockModule { - async fn handle_block( - &mut self, - block: &FullBlockInformation, - storage_tx: &mut dyn NyxdScraperTransaction, - ) -> Result<(), ScraperError>; + async fn handle_block(&mut self, block: &FullBlockInformation) -> Result<(), ScraperError>; } diff --git a/common/nyxd-scraper-shared/src/modules/msg_module.rs b/common/nyxd-scraper-shared/src/modules/msg_module.rs index 60f53b1553..93b09bf427 100644 --- a/common/nyxd-scraper-shared/src/modules/msg_module.rs +++ b/common/nyxd-scraper-shared/src/modules/msg_module.rs @@ -3,7 +3,6 @@ use crate::block_processor::types::ParsedTransactionResponse; use crate::error::ScraperError; -use crate::storage::NyxdScraperTransaction; use async_trait::async_trait; use cosmrs::Any; @@ -16,6 +15,5 @@ pub trait MsgModule { index: usize, msg: &Any, tx: &ParsedTransactionResponse, - storage_tx: &mut dyn NyxdScraperTransaction, ) -> Result<(), ScraperError>; } diff --git a/common/nyxd-scraper-shared/src/modules/tx_module.rs b/common/nyxd-scraper-shared/src/modules/tx_module.rs index 8d2f5b22b1..71774c44bc 100644 --- a/common/nyxd-scraper-shared/src/modules/tx_module.rs +++ b/common/nyxd-scraper-shared/src/modules/tx_module.rs @@ -3,14 +3,9 @@ use crate::block_processor::types::ParsedTransactionResponse; use crate::error::ScraperError; -use crate::storage::NyxdScraperTransaction; use async_trait::async_trait; #[async_trait] pub trait TxModule { - async fn handle_tx( - &mut self, - tx: &ParsedTransactionResponse, - storage_tx: &mut dyn NyxdScraperTransaction, - ) -> Result<(), ScraperError>; + async fn handle_tx(&mut self, tx: &ParsedTransactionResponse) -> Result<(), ScraperError>; } diff --git a/common/nyxd-scraper-shared/src/rpc_client.rs b/common/nyxd-scraper-shared/src/rpc_client.rs index 3f4ee84d26..8d1df1890e 100644 --- a/common/nyxd-scraper-shared/src/rpc_client.rs +++ b/common/nyxd-scraper-shared/src/rpc_client.rs @@ -9,15 +9,32 @@ use crate::helpers::tx_hash; use crate::{Any, MessageRegistry, default_message_registry}; use futures::StreamExt; use futures::future::join3; -use std::collections::{BTreeMap, HashMap}; +use std::collections::BTreeMap; use std::sync::Arc; -use tendermint::Hash; +use tendermint::{Block, Hash}; use tendermint_rpc::endpoint::{block, block_results, tx, validators}; use tendermint_rpc::{Client, HttpClient, Paging}; use tokio::sync::Mutex; use tracing::{debug, instrument, warn}; use url::Url; +#[derive(Debug, Clone, Copy)] +pub struct RetrievalConfig { + pub get_validators: bool, + pub get_transactions: bool, + pub get_block_results: bool, +} + +impl Default for RetrievalConfig { + fn default() -> Self { + Self { + get_validators: true, + get_transactions: true, + get_block_results: true, + } + } +} + #[derive(Clone)] pub struct RpcClient { // right now I don't care about anything nym specific, so a simple http client is sufficient, @@ -53,27 +70,15 @@ impl RpcClient { } } - #[instrument(skip(self, block), fields(height = block.height))] - pub async fn try_get_full_details( + fn parse_transactions( &self, - block: BlockToProcess, - ) -> Result { - debug!("getting complete block details"); - let height = block.height; - - // make all the http requests concurrently - let (results, validators, raw_transactions) = join3( - self.get_block_results(height), - self.get_validators_details(height), - self.get_transaction_results(&block.block.data), - ) - .await; - - let raw_transactions = raw_transactions?; + raw_transactions: Vec, + block: &Block, + ) -> Result, ScraperError> { let mut transactions = Vec::with_capacity(raw_transactions.len()); for raw_tx in raw_transactions { - let mut parsed_messages = HashMap::new(); - let mut parsed_message_urls = HashMap::new(); + let mut parsed_messages = BTreeMap::new(); + let mut parsed_message_urls = BTreeMap::new(); let tx = cosmrs::Tx::from_bytes(&raw_tx.tx).map_err(|source| { ScraperError::TxParseFailure { hash: raw_tx.hash, @@ -97,9 +102,33 @@ impl RpcClient { proof: raw_tx.proof, parsed_messages, parsed_message_urls, - block: block.block.clone(), + block: block.clone(), }) } + Ok(transactions) + } + + #[instrument(skip(self, block), fields(height = block.height))] + pub async fn try_get_full_details( + &self, + block: BlockToProcess, + config: RetrievalConfig, + ) -> Result { + debug!("getting complete block details"); + let height = block.height; + + // make all the http requests run concurrently + let (results, validators, raw_transactions) = join3( + self.maybe_get_block_results(height, config.get_block_results), + self.maybe_get_validators_details(height, config.get_validators), + self.maybe_get_transaction_results(&block.block.data, config.get_transactions), + ) + .await; + + let transactions = match raw_transactions? { + Some(raw) => Some(self.parse_transactions(raw, &block.block)?), + None => None, + }; Ok(FullBlockInformation { block: block.block, @@ -140,6 +169,18 @@ impl RpcClient { }) } + async fn maybe_get_block_results( + &self, + height: u32, + retrieve: bool, + ) -> Result, ScraperError> { + if retrieve { + self.get_block_results(height).await.map(Some) + } else { + Ok(None) + } + } + pub(crate) async fn current_block_height(&self) -> Result { debug!("getting current block height"); @@ -196,6 +237,18 @@ impl RpcClient { inner.into_values().collect() } + async fn maybe_get_transaction_results( + &self, + raw: &[Vec], + retrieve: bool, + ) -> Result>, ScraperError> { + if retrieve { + self.get_transaction_results(raw).await.map(Some) + } else { + Ok(None) + } + } + #[instrument(skip(self, tx_hash), fields(tx_hash = %tx_hash), err(Display))] async fn get_transaction_result(&self, tx_hash: Hash) -> Result { debug!("getting tx results"); @@ -224,4 +277,16 @@ impl RpcClient { source: Box::new(source), }) } + + async fn maybe_get_validators_details( + &self, + height: u32, + retrieve: bool, + ) -> Result, ScraperError> { + if retrieve { + self.get_validators_details(height).await.map(Some) + } else { + Ok(None) + } + } } diff --git a/common/nyxd-scraper-shared/src/scraper/mod.rs b/common/nyxd-scraper-shared/src/scraper.rs similarity index 87% rename from common/nyxd-scraper-shared/src/scraper/mod.rs rename to common/nyxd-scraper-shared/src/scraper.rs index 76b17a2a54..878cb7639a 100644 --- a/common/nyxd-scraper-shared/src/scraper/mod.rs +++ b/common/nyxd-scraper-shared/src/scraper.rs @@ -1,15 +1,15 @@ -// Copyright 2023 - Nym Technologies SA +// Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 use crate::PruningOptions; use crate::block_processor::types::BlockToProcess; -use crate::block_processor::{BlockProcessor, BlockProcessorConfig}; +use crate::block_processor::{BlockProcessor, BlockProcessorConfig, BlockProcessorPersistence}; use crate::block_requester::{BlockRequest, BlockRequester}; use crate::error::ScraperError; use crate::modules::{BlockModule, MsgModule, TxModule}; use crate::rpc_client::RpcClient; -use crate::scraper::subscriber::ChainSubscriber; use crate::storage::NyxdScraperStorage; +use crate::subscriber::ChainSubscriber; use futures::future::join_all; use std::marker::PhantomData; use std::sync::Arc; @@ -22,8 +22,6 @@ use tokio_util::task::TaskTracker; use tracing::{error, info}; use url::Url; -mod subscriber; - #[derive(Default, Clone, Copy)] pub struct StartingBlockOpts { pub start_block_height: Option, @@ -72,7 +70,7 @@ where let (processing_tx, processing_rx) = unbounded_channel(); let (req_tx, req_rx) = channel(5); - let rpc_client = RpcClient::new(&scraper.config.rpc_url)?; + let rpc_client = scraper.rpc_client.clone(); // create the tasks let block_requester = BlockRequester::new( @@ -90,14 +88,19 @@ where ); let mut block_processor = BlockProcessor::new( - block_processor_config, scraper.cancel_token.clone(), - scraper.startup_sync.clone(), processing_rx, req_tx, - scraper.storage.clone(), rpc_client, ) + .with_persistence( + BlockProcessorPersistence::new( + block_processor_config, + scraper.startup_sync.clone(), + scraper.storage.clone(), + ) + .await?, + ) .await?; block_processor.set_block_modules(self.block_modules); block_processor.set_tx_modules(self.tx_modules); @@ -126,16 +129,19 @@ where } } + #[must_use] pub fn with_block_module(mut self, module: M) -> Self { self.block_modules.push(Box::new(module)); self } + #[must_use] pub fn with_tx_module(mut self, module: M) -> Self { self.tx_modules.push(Box::new(module)); self } + #[must_use] pub fn with_msg_module(mut self, module: M) -> Self { self.msg_modules.push(Box::new(module)); self @@ -209,9 +215,12 @@ where let (req_tx, _) = channel(5); let mut block_processor = self - .new_block_processor(req_tx.clone(), processing_rx) - .await? - .with_pruning(PruningOptions::nothing()); + .new_block_processor_with_persistence( + req_tx.clone(), + processing_rx, + PruningOptions::nothing(), + ) + .await?; let block = self.rpc_client.get_basic_block_details(height).await?; @@ -234,9 +243,12 @@ where let (req_tx, _) = channel(5); let mut block_processor = self - .new_block_processor(req_tx.clone(), processing_rx) - .await? - .with_pruning(PruningOptions::nothing()); + .new_block_processor_with_persistence( + req_tx.clone(), + processing_rx, + PruningOptions::nothing(), + ) + .await?; let mut current_height = self.rpc_client.current_block_height().await? as u32; let last_processed = block_processor.last_process_height(); @@ -345,10 +357,24 @@ where ) } - async fn new_block_processor( + fn new_block_processor( &self, req_tx: Sender, processing_rx: UnboundedReceiver, + ) -> BlockProcessor { + BlockProcessor::::new( + self.cancel_token.clone(), + processing_rx, + req_tx, + self.rpc_client.clone(), + ) + } + + async fn new_block_processor_with_persistence( + &self, + req_tx: Sender, + processing_rx: UnboundedReceiver, + pruning_options: impl Into>, ) -> Result, ScraperError> { let block_processor_config = BlockProcessorConfig::new( self.config.pruning_options, @@ -357,16 +383,27 @@ where self.config.start_block.use_best_effort_start_height, ); - BlockProcessor::::new( - block_processor_config, - self.cancel_token.clone(), - self.startup_sync.clone(), - processing_rx, - req_tx, - self.storage.clone(), - self.rpc_client.clone(), - ) - .await + let persistence = match pruning_options.into() { + Some(options) => BlockProcessorPersistence::new( + block_processor_config, + self.startup_sync.clone(), + self.storage.clone(), + ) + .await? + .with_pruning(options), + None => { + BlockProcessorPersistence::new( + block_processor_config, + self.startup_sync.clone(), + self.storage.clone(), + ) + .await? + } + }; + + self.new_block_processor(req_tx, processing_rx) + .with_persistence(persistence) + .await } async fn new_chain_subscriber( @@ -388,7 +425,9 @@ where // create the tasks let block_requester = self.new_block_requester(req_rx, processing_tx.clone()); - let block_processor = self.new_block_processor(req_tx, processing_rx).await?; + let block_processor = self + .new_block_processor_with_persistence(req_tx, processing_rx, None) + .await?; let chain_subscriber = self.new_chain_subscriber(processing_tx).await?; // spawn them diff --git a/common/nyxd-scraper-shared/src/storage/mod.rs b/common/nyxd-scraper-shared/src/storage/mod.rs index c0564e343f..14e17bea9b 100644 --- a/common/nyxd-scraper-shared/src/storage/mod.rs +++ b/common/nyxd-scraper-shared/src/storage/mod.rs @@ -3,6 +3,7 @@ use crate::error::ScraperError; use async_trait::async_trait; +use tendermint::block; use thiserror::Error; use tracing::warn; @@ -89,6 +90,25 @@ pub trait NyxdScraperTransaction { async fn update_last_processed(&mut self, height: i64) -> Result<(), NyxdScraperStorageError>; } +fn ensure_proposer_present( + block_header: &block::Header, + validators: &validators::Response, +) -> Result<(), ScraperError> { + let block_proposer = block_header.proposer_address; + if !validators + .validators + .iter() + .any(|v| v.address == block_proposer) + { + let proposer = crate::helpers::validator_consensus_address(block_proposer)?; + return Err(ScraperError::BlockProposerNotInValidatorSet { + height: block_header.height.value() as u32, + proposer: proposer.to_string(), + }); + } + Ok(()) +} + pub async fn persist_block( block: &FullBlockInformation, tx: &mut Tx, @@ -97,28 +117,34 @@ pub async fn persist_block( where Tx: NyxdScraperTransaction, { - let total_gas = crate::helpers::tx_gas_sum(&block.transactions); - - // SANITY CHECK: make sure the block proposer is present in the validator set - block.ensure_proposer()?; - - tx.persist_validators(&block.validators).await?; + let total_gas = match block.transactions.as_ref() { + Some(txs) => crate::helpers::tx_gas_sum(txs), + None => 0, + }; tx.persist_block_data(&block.block, total_gas).await?; - if store_precommits { - if let Some(commit) = &block.block.last_commit { - tx.persist_commits(commit, &block.validators).await?; - } else { - warn!("no commits for block {}", block.block.header.height) + if let Some(validators) = &block.validators { + // SANITY CHECK: make sure the block proposer is present in the validator set + ensure_proposer_present(&block.block.header, validators)?; + tx.persist_validators(validators).await?; + + if store_precommits { + if let Some(commit) = &block.block.last_commit { + tx.persist_commits(commit, validators).await?; + } else { + warn!("no commits for block {}", block.block.header.height) + } } } - // persist txs - tx.persist_txs(&block.transactions).await?; + if let Some(transactions) = &block.transactions { + // persist txs + tx.persist_txs(transactions).await?; - // persist messages (inside the transactions) - tx.persist_messages(&block.transactions).await?; + // persist messages (inside the transactions) + tx.persist_messages(transactions).await?; + } tx.update_last_processed(block.block.header.height.into()) .await?; diff --git a/common/nyxd-scraper-shared/src/scraper/subscriber.rs b/common/nyxd-scraper-shared/src/subscriber.rs similarity index 100% rename from common/nyxd-scraper-shared/src/scraper/subscriber.rs rename to common/nyxd-scraper-shared/src/subscriber.rs diff --git a/common/nyxd-scraper-shared/src/watcher.rs b/common/nyxd-scraper-shared/src/watcher.rs new file mode 100644 index 0000000000..3ba3ea5b4e --- /dev/null +++ b/common/nyxd-scraper-shared/src/watcher.rs @@ -0,0 +1,160 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::block_processor::BlockProcessor; +use crate::block_requester::BlockRequester; +use crate::error::ScraperError; +use crate::rpc_client::{RetrievalConfig, RpcClient}; +use crate::subscriber::ChainSubscriber; +use crate::{BlockModule, MsgModule, TxModule}; +use tokio::sync::mpsc::{channel, unbounded_channel}; +use tokio_util::sync::CancellationToken; +use tokio_util::task::TaskTracker; +use tracing::info; +use url::Url; + +pub struct WatcherConfig { + /// Url to the websocket endpoint of a validator, for example `wss://rpc.nymtech.net/websocket` + pub websocket_url: Url, + + /// Url to the rpc endpoint of a validator, for example `https://rpc.nymtech.net/` + pub rpc_url: Url, +} + +pub struct NyxdWatcherBuilder { + config: WatcherConfig, + + block_modules: Vec>, + tx_modules: Vec>, + msg_modules: Vec>, +} + +impl NyxdWatcherBuilder { + pub fn new(config: WatcherConfig) -> Self { + NyxdWatcherBuilder { + config, + block_modules: vec![], + tx_modules: vec![], + msg_modules: vec![], + } + } + + #[must_use] + pub fn with_block_module(mut self, module: M) -> Self { + self.block_modules.push(Box::new(module)); + self + } + + #[must_use] + pub fn with_tx_module(mut self, module: M) -> Self { + self.tx_modules.push(Box::new(module)); + self + } + + #[must_use] + pub fn with_msg_module(mut self, module: M) -> Self { + self.msg_modules.push(Box::new(module)); + self + } + + pub async fn build_and_start(self) -> Result { + // we must have at least something configured to run the watcher + if self.block_modules.is_empty() + && self.tx_modules.is_empty() + && self.msg_modules.is_empty() + { + return Err(ScraperError::NoModulesConfigured); + } + + let watcher = NyxdWatcher::new(); + let rpc_client = RpcClient::new(&self.config.rpc_url)?; + + let (processing_tx, processing_rx) = unbounded_channel(); + let (req_tx, req_rx) = channel(5); + + // create the tasks + let block_requester = BlockRequester::new( + watcher.cancel_token(), + rpc_client.clone(), + req_rx, + processing_tx.clone(), + ); + + let mut block_processor = + BlockProcessor::new(watcher.cancel_token(), processing_rx, req_tx, rpc_client) + .with_retrieval_config(RetrievalConfig { + get_validators: false, + get_transactions: true, + get_block_results: false, + }); + block_processor.set_block_modules(self.block_modules); + block_processor.set_tx_modules(self.tx_modules); + block_processor.set_msg_modules(self.msg_modules); + + let chain_subscriber = ChainSubscriber::new( + &self.config.websocket_url, + watcher.cancel_token(), + watcher.task_tracker.clone(), + processing_tx, + ) + .await?; + + watcher.start_tasks(block_requester, block_processor, chain_subscriber); + + Ok(watcher) + } +} + +/// A simpler alternative to the `NyxdScraper` that does not persist any received block information. +/// Instead, it only calls the registered modules on the processed data. +/// +/// Furthermore, it also does not retrieve any validator information or detailed block information +pub struct NyxdWatcher { + task_tracker: TaskTracker, + cancel_token: CancellationToken, +} + +impl NyxdWatcher { + pub fn builder(config: WatcherConfig) -> NyxdWatcherBuilder { + NyxdWatcherBuilder::new(config) + } + + fn new() -> Self { + Self { + task_tracker: TaskTracker::new(), + cancel_token: CancellationToken::new(), + } + } + + fn start_tasks( + &self, + mut block_requester: BlockRequester, + mut block_processor: BlockProcessor, + mut chain_subscriber: ChainSubscriber, + ) { + self.task_tracker + .spawn(async move { block_requester.run().await }); + self.task_tracker + .spawn(async move { block_processor.run().await }); + self.task_tracker + .spawn(async move { chain_subscriber.run().await }); + + self.task_tracker.close(); + } + + pub async fn stop(self) { + info!("stopping the chain watcher"); + assert!(self.task_tracker.is_closed()); + + self.cancel_token.cancel(); + self.task_tracker.wait().await + } + + pub fn cancel_token(&self) -> CancellationToken { + self.cancel_token.clone() + } + + pub fn is_cancelled(&self) -> bool { + self.cancel_token.is_cancelled() + } +} diff --git a/nym-data-observatory/src/chain_scraper/webhook.rs b/nym-data-observatory/src/chain_scraper/webhook.rs index 641f5741d7..240bcbdccf 100644 --- a/nym-data-observatory/src/chain_scraper/webhook.rs +++ b/nym-data-observatory/src/chain_scraper/webhook.rs @@ -5,9 +5,7 @@ use crate::config::data_observatory::{HttpAuthenticationOptions, WebhookConfig}; use crate::models::WebhookPayload; use anyhow::Context; use async_trait::async_trait; -use nyxd_scraper_psql::{ - NyxdScraperTransaction, ParsedTransactionResponse, ScraperError, TxModule, -}; +use nyxd_scraper_psql::{ParsedTransactionResponse, ScraperError, TxModule}; use reqwest::{Client, Url}; use tracing::{error, info}; @@ -29,11 +27,7 @@ impl WebhookModule { #[async_trait] impl TxModule for WebhookModule { - async fn handle_tx( - &mut self, - tx: &ParsedTransactionResponse, - _: &mut dyn NyxdScraperTransaction, - ) -> Result<(), ScraperError> { + async fn handle_tx(&mut self, tx: &ParsedTransactionResponse) -> Result<(), ScraperError> { for (index, msg) in &tx.parsed_messages { if let Some(parsed_message_type_url) = tx.parsed_message_urls.get(index) { let payload = WebhookPayload { diff --git a/nym-data-observatory/src/modules/wasm.rs b/nym-data-observatory/src/modules/wasm.rs index 98fe1dc809..bbe90e2c37 100644 --- a/nym-data-observatory/src/modules/wasm.rs +++ b/nym-data-observatory/src/modules/wasm.rs @@ -8,9 +8,7 @@ use cosmrs::proto::cosmwasm::wasm::v1::MsgExecuteContract; use cosmrs::proto::prost::Message; use nym_validator_client::nyxd::{Any, Name}; use nyxd_scraper_psql::models::DbCoin; -use nyxd_scraper_psql::{ - MsgModule, NyxdScraperTransaction, ParsedTransactionResponse, ScraperError, -}; +use nyxd_scraper_psql::{MsgModule, ParsedTransactionResponse, ScraperError}; use serde_json::Value; use time::{OffsetDateTime, PrimitiveDateTime}; use tracing::{error, trace}; @@ -37,7 +35,6 @@ impl MsgModule for WasmModule { index: usize, msg: &Any, tx: &ParsedTransactionResponse, - _storage_tx: &mut dyn NyxdScraperTransaction, ) -> Result<(), ScraperError> { let message = serde_json::to_value(tx.parsed_messages.get(&index)).unwrap_or_default(); let value = serde_json::to_value(message.clone()).unwrap_or_default(); diff --git a/nyx-chain-watcher/src/chain_scraper/mod.rs b/nyx-chain-watcher/src/chain_scraper/mod.rs index fe0d2187ff..5bf6eff337 100644 --- a/nyx-chain-watcher/src/chain_scraper/mod.rs +++ b/nyx-chain-watcher/src/chain_scraper/mod.rs @@ -7,8 +7,7 @@ use crate::http::state::BankScraperModuleState; use async_trait::async_trait; use nym_validator_client::nyxd::{Any, Coin, CosmosCoin, Hash, Msg, MsgSend, Name}; use nyxd_scraper_sqlite::{ - MsgModule, NyxdScraperTransaction, ParsedTransactionResponse, PruningOptions, ScraperError, - SqliteNyxdScraper, + MsgModule, ParsedTransactionResponse, PruningOptions, ScraperError, SqliteNyxdScraper, }; use sqlx::SqlitePool; use std::fs; @@ -158,7 +157,6 @@ impl MsgModule for BankScraperModule { index: usize, msg: &Any, tx: &ParsedTransactionResponse, - _storage_tx: &mut dyn NyxdScraperTransaction, ) -> Result<(), ScraperError> { let memo = tx.tx.body.memo.clone();