diff --git a/Cargo.lock b/Cargo.lock index 1212868dd7..4a7a46578e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8110,24 +8110,12 @@ name = "nyxd-scraper-sqlite" version = "0.1.0" dependencies = [ "async-trait", - "const_format", - "cosmrs", - "eyre", - "futures", "humantime", "nyxd-scraper-shared", - "serde", - "sha2 0.10.9", "sqlx", - "tendermint", - "tendermint-rpc", "thiserror 2.0.12", - "time", "tokio", - "tokio-stream", - "tokio-util", "tracing", - "url", ] [[package]] diff --git a/common/nyxd-scraper-shared/src/error.rs b/common/nyxd-scraper-shared/src/error.rs index 14cdeddb98..d52cd145cd 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::helpers::MalformedDataError; use crate::storage::NyxdScraperStorageError; use tendermint::Hash; use thiserror::Error; @@ -11,11 +12,9 @@ use tokio::sync::mpsc::error::SendError; #[derive(Debug, Error)] pub enum ScraperError { - #[error("experienced internal database error: {0}")] - InternalDatabaseError(#[from] NyxdScraperStorageError), + #[error("storage error: {0}")] + StorageError(#[from] NyxdScraperStorageError), - // #[error("failed to perform startup SQL migration: {0}")] - // StartupMigrationFailure(#[from] Box), #[error("the block scraper is already running")] ScraperAlreadyRunning, @@ -116,28 +115,14 @@ pub enum ScraperError { #[error("failed to send on a closed channel")] ClosedChannelError, - #[error("failed to parse validator's address: {source}")] - MalformedValidatorAddress { - #[source] - source: eyre::Report, - }, - - #[error("failed to parse validator's address: {source}")] - MalformedValidatorPubkey { - #[source] - source: eyre::Report, - }, + #[error(transparent)] + MalformedData(#[from] MalformedDataError), #[error( "could not find the block proposer ('{proposer}') for height {height} in the validator set" )] BlockProposerNotInValidatorSet { height: u32, proposer: String }, - #[error( - "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, diff --git a/common/nyxd-scraper-shared/src/helpers.rs b/common/nyxd-scraper-shared/src/helpers.rs index 7c3ea91e6a..2a20a2751b 100644 --- a/common/nyxd-scraper-shared/src/helpers.rs +++ b/common/nyxd-scraper-shared/src/helpers.rs @@ -3,42 +3,62 @@ use crate::block_processor::types::ParsedTransactionResponse; use crate::constants::{BECH32_CONESNSUS_PUBKEY_PREFIX, BECH32_CONSENSUS_ADDRESS_PREFIX}; -use crate::error::ScraperError; use cosmrs::AccountId; use sha2::{Digest, Sha256}; use tendermint::{account, PublicKey}; use tendermint::{validator, Hash}; use tendermint_rpc::endpoint::validators; +use thiserror::Error; -pub(crate) fn tx_hash>(raw_tx: M) -> Hash { +#[derive(Error, Debug)] +pub enum MalformedDataError { + #[error("failed to parse validator's address: {source}")] + MalformedValidatorAddress { + #[source] + source: eyre::Report, + }, + + #[error("failed to parse validator's address: {source}")] + MalformedValidatorPubkey { + #[source] + source: eyre::Report, + }, + + #[error( + "could not find validator information for {address}; the validator has signed a commit" + )] + MissingValidatorInfoCommitted { address: String }, +} + +pub fn tx_hash>(raw_tx: M) -> Hash { Hash::Sha256(Sha256::digest(raw_tx).into()) } -pub 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()) - .map_err(|source| ScraperError::MalformedValidatorPubkey { source }) + .map_err(|source| MalformedDataError::MalformedValidatorPubkey { source }) } -pub(crate) fn validator_consensus_address(id: account::Id) -> Result { +pub fn validator_consensus_address(id: account::Id) -> Result { AccountId::new(BECH32_CONSENSUS_ADDRESS_PREFIX, id.as_ref()) - .map_err(|source| ScraperError::MalformedValidatorAddress { source }) + .map_err(|source| MalformedDataError::MalformedValidatorAddress { source }) } -pub(crate) fn tx_gas_sum(txs: &[ParsedTransactionResponse]) -> i64 { +pub fn tx_gas_sum(txs: &[ParsedTransactionResponse]) -> i64 { txs.iter().map(|tx| tx.tx_result.gas_used).sum() } pub fn validator_info( id: account::Id, validators: &validators::Response, -) -> Result<&validator::Info, ScraperError> { +) -> Result<&validator::Info, MalformedDataError> { match validators.validators.iter().find(|v| v.address == id) { Some(info) => Ok(info), None => { let addr = validator_consensus_address(id)?; - Err(ScraperError::MissingValidatorInfoCommitted { + Err(MalformedDataError::MissingValidatorInfoCommitted { address: addr.to_string(), }) } diff --git a/common/nyxd-scraper-shared/src/lib.rs b/common/nyxd-scraper-shared/src/lib.rs index d0d97c249a..43f6ffdcf2 100644 --- a/common/nyxd-scraper-shared/src/lib.rs +++ b/common/nyxd-scraper-shared/src/lib.rs @@ -9,7 +9,7 @@ pub mod helpers; pub mod modules; pub(crate) mod rpc_client; pub(crate) mod scraper; -mod storage; +pub mod storage; pub use block_processor::pruning::{PruningOptions, PruningStrategy}; pub use block_processor::types::ParsedTransactionResponse; diff --git a/common/nyxd-scraper-shared/src/storage/helpers.rs b/common/nyxd-scraper-shared/src/storage/helpers.rs index d49d4baeb1..488103fcc6 100644 --- a/common/nyxd-scraper-shared/src/storage/helpers.rs +++ b/common/nyxd-scraper-shared/src/storage/helpers.rs @@ -4,7 +4,7 @@ use tokio::time::Instant; use tracing::{debug, error, info, trace, warn}; -pub(crate) fn log_db_operation_time(op_name: &str, start_time: Instant) { +pub fn log_db_operation_time(op_name: &str, start_time: Instant) { let elapsed = start_time.elapsed(); let formatted = humantime::format_duration(elapsed); diff --git a/common/nyxd-scraper-shared/src/storage/mod.rs b/common/nyxd-scraper-shared/src/storage/mod.rs index 1146cf8837..b57c3cb510 100644 --- a/common/nyxd-scraper-shared/src/storage/mod.rs +++ b/common/nyxd-scraper-shared/src/storage/mod.rs @@ -1,37 +1,47 @@ // Copyright 2025 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::block_processor::types::FullBlockInformation; use crate::error::ScraperError; -use crate::ParsedTransactionResponse; use async_trait::async_trait; -use std::path::PathBuf; -use tendermint::block::Commit; -use tendermint::Block; -use tendermint_rpc::endpoint::validators; +use std::path::Path; use thiserror::Error; use tracing::warn; -pub(crate) mod helpers; +pub use crate::block_processor::types::FullBlockInformation; +pub use crate::ParsedTransactionResponse; +pub use tendermint::block::{Commit, CommitSig}; +pub use tendermint::Block; +pub use tendermint_rpc::endpoint::validators; + +pub 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); +impl NyxdScraperStorageError { + pub fn new(error: E) -> Self + where + E: std::error::Error + Send + Sync + 'static, + { + NyxdScraperStorageError(Box::new(error)) + } +} + #[async_trait] pub trait NyxdScraperStorage: Clone + Sized { type StorageTransaction: NyxdScraperTransaction; - async fn initialise(storage_path: &PathBuf) -> Result; + async fn initialise(storage_path: &Path) -> 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, NyxdScraperStorageError>; diff --git a/common/nyxd-scraper-sqlite/Cargo.toml b/common/nyxd-scraper-sqlite/Cargo.toml index bf32afbb8a..0035689224 100644 --- a/common/nyxd-scraper-sqlite/Cargo.toml +++ b/common/nyxd-scraper-sqlite/Cargo.toml @@ -11,32 +11,16 @@ license.workspace = true # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -async-trait.workspace = true -const_format = { workspace = true } -cosmrs.workspace = true -eyre = { workspace = true } -futures.workspace = true +async-trait = { workspace = true } humantime = { workspace = true } -sha2 = { workspace = true } -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"] } -thiserror.workspace = true -time = { workspace = true } +thiserror = { workspace = true } tokio = { workspace = true, features = ["full"] } -tokio-stream = { workspace = true } -tokio-util = { workspace = true, features = ["rt"] } tracing.workspace = true -url.workspace = true nyxd-scraper-shared = { path = "../nyxd-scraper-shared" } -# TEMP -#nym-bin-common = { path = "../bin-common", features = ["basic_tracing"]} - - [build-dependencies] sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] } tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } diff --git a/common/nyxd-scraper-sqlite/src/block_processor/helpers.rs b/common/nyxd-scraper-sqlite/src/block_processor/helpers.rs deleted file mode 100644 index ffc8c592c5..0000000000 --- a/common/nyxd-scraper-sqlite/src/block_processor/helpers.rs +++ /dev/null @@ -1,65 +0,0 @@ -// Copyright 2023 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::block_processor::MAX_RANGE_SIZE; -use std::cmp::min; -use std::collections::VecDeque; -use std::ops::Range; - -pub(crate) fn split_request_range(request_range: Range) -> VecDeque> { - let mut requests = VecDeque::new(); - - let mut start = request_range.start; - let mut end = min(request_range.end, start + MAX_RANGE_SIZE as u32); - - loop { - requests.push_back(start..end); - start = min(start + MAX_RANGE_SIZE as u32, request_range.end); - end = min(end + MAX_RANGE_SIZE as u32, request_range.end); - - if start == end { - break; - } - } - - requests -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn splitting_request_range() { - let range = 0..100; - let mut expected = VecDeque::new(); - expected.push_back(0..30); - expected.push_back(30..60); - expected.push_back(60..90); - expected.push_back(90..100); - assert_eq!(expected, split_request_range(range)); - - let range = 0..30; - let mut expected = VecDeque::new(); - expected.push_back(0..30); - assert_eq!(expected, split_request_range(range)); - - let range = 0..60; - let mut expected = VecDeque::new(); - expected.push_back(0..30); - expected.push_back(30..60); - assert_eq!(expected, split_request_range(range)); - - let range = 0..5; - let mut expected = VecDeque::new(); - expected.push_back(0..5); - assert_eq!(expected, split_request_range(range)); - - let range = 123..200; - let mut expected = VecDeque::new(); - expected.push_back(123..153); - expected.push_back(153..183); - expected.push_back(183..200); - assert_eq!(expected, split_request_range(range)); - } -} diff --git a/common/nyxd-scraper-sqlite/src/block_processor/mod.rs b/common/nyxd-scraper-sqlite/src/block_processor/mod.rs deleted file mode 100644 index d4007de96b..0000000000 --- a/common/nyxd-scraper-sqlite/src/block_processor/mod.rs +++ /dev/null @@ -1,515 +0,0 @@ -// Copyright 2023 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -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::{persist_block, ScraperStorage}; -use crate::PruningOptions; -use futures::StreamExt; -use std::cmp::max; -use std::collections::{BTreeMap, HashSet, VecDeque}; -use std::ops::{Add, Range}; -use std::sync::Arc; -use std::time::Duration; -use tokio::sync::mpsc::{Sender, UnboundedReceiver}; -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, instrument, trace, warn}; - -mod helpers; -pub(crate) mod pruning; -pub(crate) mod types; - -const MISSING_BLOCKS_CHECK_INTERVAL: Duration = Duration::from_secs(30); -const MAX_MISSING_BLOCKS_DELAY: Duration = Duration::from_secs(15); -const MAX_RANGE_SIZE: usize = 30; - -#[derive(Debug, Default)] -struct PendingSync { - request_in_flight: HashSet, - queued_requests: VecDeque>, -} - -impl PendingSync { - fn is_empty(&self) -> bool { - self.request_in_flight.is_empty() && self.queued_requests.is_empty() - } -} - -#[derive(Debug, Clone)] -pub struct BlockProcessorConfig { - pub pruning_options: PruningOptions, - pub store_precommits: bool, - pub explicit_starting_block_height: Option, - pub use_best_effort_start_height: bool, -} - -impl Default for BlockProcessorConfig { - fn default() -> Self { - Self { - pruning_options: PruningOptions::nothing(), - store_precommits: true, - explicit_starting_block_height: None, - use_best_effort_start_height: false, - } - } -} - -impl BlockProcessorConfig { - pub fn new( - pruning_options: PruningOptions, - store_precommits: bool, - explicit_starting_block_height: Option, - use_best_effort_start_height: bool, - ) -> Self { - Self { - pruning_options, - store_precommits, - explicit_starting_block_height, - use_best_effort_start_height, - } - } -} - -pub struct BlockProcessor { - config: BlockProcessorConfig, - cancel: CancellationToken, - synced: Arc, - last_processed_height: u32, - last_pruned_height: u32, - last_processed_at: Instant, - pending_sync: PendingSync, - queued_blocks: BTreeMap, - - rpc_client: RpcClient, - incoming: UnboundedReceiverStream, - block_requester: Sender, - storage: ScraperStorage, - - // future work: rather than sending each msg to every msg module, - // let them subscribe based on `type_url` inside the message itself - // (like "/cosmwasm.wasm.v1.MsgExecuteContract") - block_modules: Vec>, - tx_modules: Vec>, - msg_modules: Vec>, -} - -#[allow(clippy::too_many_arguments)] -impl BlockProcessor { - pub async fn new( - config: BlockProcessorConfig, - cancel: CancellationToken, - synced: Arc, - incoming: UnboundedReceiver, - block_requester: Sender, - storage: ScraperStorage, - 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, - cancel, - synced, - last_processed_height, - last_pruned_height, - last_processed_at: Instant::now(), - pending_sync: Default::default(), - queued_blocks: Default::default(), - rpc_client, - incoming: incoming.into(), - block_requester, - storage, - 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; - 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?; - - 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) - } - } - - // 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?; - - persist_block(&full_info, &mut tx, self.config.store_precommits).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? - } - } - } - } - - 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(()) - } - - pub fn set_block_modules(&mut self, modules: Vec>) { - self.block_modules = modules; - } - - pub fn set_tx_modules(&mut self, modules: Vec>) { - self.tx_modules = modules; - } - - pub fn set_msg_modules(&mut self, modules: Vec>) { - self.msg_modules = modules; - } - - pub(super) fn last_process_height(&self) -> u32 { - self.last_processed_height - } - - async fn maybe_request_missing_blocks(&mut self) -> Result<(), ScraperError> { - // we're still processing, so we're good - if self.last_processed_at.elapsed() < MAX_MISSING_BLOCKS_DELAY { - debug!("no need to request missing blocks"); - return Ok(()); - } - - if self.try_request_pending().await { - return Ok(()); - } - - // TODO: properly fill in the gaps later with BlockRequest::Specific, - let request_range = if let Some((next_available, _)) = self.queued_blocks.first_key_value() - { - self.last_processed_height + 1..*next_available - } else { - let current_height = self.rpc_client.current_block_height().await? as u32; - self.last_processed_height + 1..current_height + 1 - }; - - self.request_missing_blocks(request_range).await?; - - Ok(()) - } - - async fn request_missing_blocks( - &mut self, - request_range: Range, - ) -> Result<(), ScraperError> { - let request_range = if request_range.len() > MAX_RANGE_SIZE { - let mut split = split_request_range(request_range); - - // SAFETY: we know that after the split of a non-empty range we have AT LEAST one value - #[allow(clippy::unwrap_used)] - let first = split.pop_front().unwrap(); - self.pending_sync.queued_requests = split; - self.pending_sync.request_in_flight = first.clone().collect(); - - first - } else { - request_range - }; - - self.send_blocks_request(request_range).await - } - - // technically we're not mutating self here, - // but we need it to help the compiler figure out the future is `Send` - async fn send_blocks_request(&mut self, request_range: Range) -> Result<(), ScraperError> { - debug!("requesting missing blocks: {request_range:?}"); - - self.block_requester - .send(BlockRequest::Range(request_range)) - .await?; - 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; - - self.pending_sync.request_in_flight.remove(&height); - - if self.last_processed_height == 0 { - // this is the first time we've started up the process - debug!("setting up initial processing height"); - self.last_processed_height = height - 1 - } - - if height <= self.last_processed_height { - warn!("we have already processed block for height {height}"); - return; - } - - if self.last_processed_height + 1 != height { - if self.queued_blocks.insert(height, block).is_some() { - warn!("we have already queued up block for height {height}"); - } - return; - } - - if let Err(err) = self.process_block(block).await { - error!("failed to process block at height {height}: {err}"); - return; - } - - // process as much as we can from the queue - let mut next = height + 1; - while let Some(next_block) = self.queued_blocks.remove(&next) { - if let Err(err) = self.process_block(next_block).await { - error!("failed to process queued-up block at height {next}: {err}") - } - next += 1; - } - - self.try_request_pending().await; - - if self.pending_sync.is_empty() { - self.synced.notify_one(); - } - } - - async fn try_request_pending(&mut self) -> bool { - if self.pending_sync.request_in_flight.is_empty() { - if let Some(next_sync) = self.pending_sync.queued_requests.pop_front() { - debug!( - "current request range has been resolved. requesting another bunch of blocks" - ); - if let Err(err) = self.send_blocks_request(next_sync.clone()).await { - error!("failed to request resync blocks: {err}"); - self.pending_sync.queued_requests.push_front(next_sync); - } else { - self.pending_sync.request_in_flight = next_sync.collect() - } - - return true; - } - } - - false - } - - // technically we're not mutating self here, - // but we need it to help the compiler figure out the future is `Send` - async fn startup_resync(&mut self) -> Result<(), ScraperError> { - assert!(self.pending_sync.is_empty()); - info!("attempting to run startup resync..."); - - self.maybe_prune_storage().await?; - - let latest_block = self.rpc_client.current_block_height().await? as u32; - info!("obtained latest block height: {latest_block}"); - - if latest_block > self.last_processed_height && self.last_processed_height != 0 { - 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 last_to_keep = latest_block - keep_recent; - - if !self.config.pruning_options.strategy.is_nothing() { - self.last_processed_height = max(self.last_processed_height, last_to_keep); - } - - let request_range = self.last_processed_height + 1..latest_block + 1; - info!( - keep_recent = %keep_recent, - last_to_keep = %last_to_keep, - last_processed_height = %self.last_processed_height, - "we need to request {request_range:?} to resync" - ); - self.request_missing_blocks(request_range).await?; - return Ok(()); - } - - // 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 { - info!("no starting block height set - will use the default behaviour"); - // nothing to do - return Ok(()); - }; - - info!("attempting to start the scraper from block {starting_height}"); - let earliest_available = - 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 { - error!("the earliest available block is higher than the desired starting height"); - return Err(ScraperError::BlocksUnavailable { - height: starting_height, - }); - } - - let starting_height = if earliest_available > starting_height { - // add few additional blocks to account for all the startup waiting - // because the node might have pruned few blocks since - earliest_available + 10 - } else { - starting_height - }; - - let request_range = starting_height..latest_block + 1; - - info!("going to start the scraper from block {starting_height}"); - info!("we need to request {request_range:?} before properly starting up"); - - self.request_missing_blocks(request_range).await?; - } - - Ok(()) - } - - pub(crate) async fn run(&mut self) { - info!("starting block processor processing loop"); - - // sure, we could be more efficient and reset it on every processed block, - // but the overhead is so minimal that it doesn't matter - let mut missing_check_interval = interval_at( - Instant::now().add(MISSING_BLOCKS_CHECK_INTERVAL), - MISSING_BLOCKS_CHECK_INTERVAL, - ); - - if let Err(err) = self.startup_resync().await { - error!("failed to perform startup sync: {err}"); - self.cancel.cancel(); - return; - }; - - loop { - tokio::select! { - _ = self.cancel.cancelled() => { - info!("received cancellation token"); - break - } - _ = missing_check_interval.tick() => { - if let Err(err) = self.maybe_request_missing_blocks().await { - error!("failed to request missing blocks: {err}") - } - } - block = self.incoming.next() => { - match block { - Some(block) => self.next_incoming(block).await, - None => { - warn!("stopped receiving new blocks"); - self.cancel.cancel(); - break - } - } - } - } - } - } -} diff --git a/common/nyxd-scraper-sqlite/src/block_processor/pruning.rs b/common/nyxd-scraper-sqlite/src/block_processor/pruning.rs deleted file mode 100644 index e7a99e7dfa..0000000000 --- a/common/nyxd-scraper-sqlite/src/block_processor/pruning.rs +++ /dev/null @@ -1,122 +0,0 @@ -// Copyright 2024 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::error::ScraperError; -use serde::{Deserialize, Serialize}; - -pub const DEFAULT_PRUNING_KEEP_RECENT: u32 = 362880; -pub const DEFAULT_PRUNING_INTERVAL: u32 = 10; -pub const EVERYTHING_PRUNING_KEEP_RECENT: u32 = 2; -pub const EVERYTHING_PRUNING_INTERVAL: u32 = 10; - -/// We follow cosmos-sdk pruning strategies for convenience’s sake. -#[derive(Debug, Default, Clone, Copy, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum PruningStrategy { - /// 'Default' strategy defines a pruning strategy where the last 362880 heights are - /// kept where to-be pruned heights are pruned at every 10th height. - /// The last 362880 heights are kept(approximately 3.5 weeks worth of state) assuming the typical - /// block time is 6s. If these values do not match the applications' requirements, use the "custom" option. - #[default] - Default, - - /// 'Everything' strategy defines a pruning strategy where all committed heights are - /// deleted, storing only the current height and last 2 states. To-be pruned heights are - /// pruned at every 10th height. - Everything, - - /// 'Nothing' strategy defines a pruning strategy where all heights are kept on disk. - Nothing, - - /// 'Custom' strategy defines a pruning strategy where the user specifies the pruning. - Custom, -} - -impl PruningStrategy { - pub fn is_custom(&self) -> bool { - matches!(self, PruningStrategy::Custom) - } - - pub fn is_nothing(&self) -> bool { - matches!(self, PruningStrategy::Nothing) - } - - pub fn is_everything(&self) -> bool { - matches!(self, PruningStrategy::Everything) - } -} - -#[derive(Debug, Clone, Copy, Serialize, Deserialize)] -pub struct PruningOptions { - /// keep_recent defines how many recent heights to keep on disk. - pub keep_recent: u32, - - /// interval defines the frequency of removing the pruned heights from the disk. - pub interval: u32, - - /// strategy defines the currently used kind of [PruningStrategy]. - pub strategy: PruningStrategy, -} - -impl PruningOptions { - pub fn validate(&self) -> Result<(), ScraperError> { - // if strategy is not set to custom, other options are meaningless since they won't be applied - if !self.strategy.is_custom() { - return Ok(()); - } - - if self.interval == 0 { - return Err(ScraperError::ZeroPruningInterval); - } - - if self.interval < EVERYTHING_PRUNING_INTERVAL { - return Err(ScraperError::TooSmallPruningInterval { - interval: self.interval, - }); - } - - if self.keep_recent < EVERYTHING_PRUNING_KEEP_RECENT { - return Err(ScraperError::TooSmallKeepRecent { - keep_recent: self.keep_recent, - }); - } - - Ok(()) - } - - pub fn nothing() -> Self { - PruningOptions { - keep_recent: 0, - interval: 0, - strategy: PruningStrategy::Nothing, - } - } - - pub fn strategy_interval(&self) -> u32 { - match self.strategy { - PruningStrategy::Default => DEFAULT_PRUNING_INTERVAL, - PruningStrategy::Everything => EVERYTHING_PRUNING_INTERVAL, - PruningStrategy::Nothing => 0, - PruningStrategy::Custom => self.interval, - } - } - - pub fn strategy_keep_recent(&self) -> u32 { - match self.strategy { - PruningStrategy::Default => DEFAULT_PRUNING_KEEP_RECENT, - PruningStrategy::Everything => EVERYTHING_PRUNING_KEEP_RECENT, - PruningStrategy::Nothing => 0, - PruningStrategy::Custom => self.keep_recent, - } - } -} - -impl Default for PruningOptions { - fn default() -> Self { - PruningOptions { - keep_recent: DEFAULT_PRUNING_KEEP_RECENT, - interval: DEFAULT_PRUNING_INTERVAL, - strategy: Default::default(), - } - } -} diff --git a/common/nyxd-scraper-sqlite/src/block_processor/types.rs b/common/nyxd-scraper-sqlite/src/block_processor/types.rs deleted file mode 100644 index 6909ff32a0..0000000000 --- a/common/nyxd-scraper-sqlite/src/block_processor/types.rs +++ /dev/null @@ -1,115 +0,0 @@ -// Copyright 2023 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::error::ScraperError; -use crate::helpers; -use tendermint::{abci, block, tx, Block, Hash}; -use tendermint_rpc::endpoint::{block as block_endpoint, block_results, validators}; -use tendermint_rpc::event::{Event, EventData}; - -// just get all everything out of tx::Response, but parse raw `tx` bytes -#[derive(Clone, Debug)] -pub struct ParsedTransactionResponse { - /// The hash of the transaction. - /// - /// Deserialized from a hex-encoded string (there is a discrepancy between - /// the format used for the request and the format used for the response in - /// the Tendermint RPC). - pub hash: Hash, - - pub height: block::Height, - - pub index: u32, - - pub tx_result: abci::types::ExecTxResult, - - pub tx: cosmrs::tx::Tx, - - pub proof: Option, -} - -#[derive(Debug)] -pub struct FullBlockInformation { - /// Basic block information, including its signers. - pub block: Block, - - /// All of the emitted events alongside any tx results. - pub results: block_results::Response, - - /// Validator set for this particular block - pub validators: validators::Response, - - /// 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(crate) struct BlockToProcess { - pub(crate) height: u32, - pub(crate) block: Block, -} - -impl From for BlockToProcess { - fn from(block: Block) -> Self { - BlockToProcess { - height: block.header.height.value() as u32, - block, - } - } -} - -impl TryFrom for BlockToProcess { - type Error = ScraperError; - - fn try_from(event: Event) -> Result { - let query = event.query.clone(); - - // TODO: we're losing `result_begin_block` and `result_end_block` here but maybe that's fine? - let maybe_block = match event.data { - EventData::NewBlock { block, .. } => block, - EventData::LegacyNewBlock { block, .. } => block, - EventData::Tx { .. } => { - return Err(ScraperError::InvalidSubscriptionEvent { - query, - kind: "Tx".to_string(), - }) - } - EventData::GenericJsonEvent(_) => { - return Err(ScraperError::InvalidSubscriptionEvent { - query, - kind: "GenericJsonEvent".to_string(), - }) - } - }; - - let Some(block) = maybe_block else { - return Err(ScraperError::EmptyBlockData { query }); - }; - - Ok((*block).into()) - } -} - -impl From for BlockToProcess { - fn from(value: block_endpoint::Response) -> Self { - value.block.into() - } -} diff --git a/common/nyxd-scraper-sqlite/src/block_requester/mod.rs b/common/nyxd-scraper-sqlite/src/block_requester/mod.rs deleted file mode 100644 index b2a1a4d43a..0000000000 --- a/common/nyxd-scraper-sqlite/src/block_requester/mod.rs +++ /dev/null @@ -1,91 +0,0 @@ -// Copyright 2023 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::block_processor::types::BlockToProcess; -use crate::error::ScraperError; -use crate::rpc_client::RpcClient; -use futures::StreamExt; -use std::ops::Range; -use tokio::sync::mpsc::{Receiver, UnboundedSender}; -use tokio_stream::wrappers::ReceiverStream; -use tokio_util::sync::CancellationToken; -use tracing::{error, info, instrument, warn}; - -#[derive(Debug)] -pub enum BlockRequest { - Range(Range), - - // UNIMPLEMENTED: - #[allow(dead_code)] - Specific(Vec), -} - -pub(crate) struct BlockRequester { - cancel: CancellationToken, - rpc_client: RpcClient, - requests: ReceiverStream, - blocks: UnboundedSender, -} - -impl BlockRequester { - pub(crate) fn new( - cancel: CancellationToken, - rpc_client: RpcClient, - requests: Receiver, - blocks: UnboundedSender, - ) -> Self { - BlockRequester { - cancel, - rpc_client, - requests: requests.into(), - blocks, - } - } - - async fn request_and_send(&self, height: u32) -> Result<(), ScraperError> { - let block = self.rpc_client.get_basic_block_details(height).await?; - self.blocks.send(block.into())?; - Ok(()) - } - - async fn request_blocks>(&self, heights: I) { - futures::stream::iter(heights) - .for_each_concurrent(4, |height| async move { - if let Err(err) = self.request_and_send(height).await { - error!("failed to request block data: {err}") - } - }) - .await - } - - #[instrument(skip(self))] - async fn handle_blocks_request(&self, request: BlockRequest) { - info!("received request for missed blocks"); - - match request { - BlockRequest::Range(range) => self.request_blocks(range).await, - BlockRequest::Specific(heights) => self.request_blocks(heights).await, - } - } - - pub(crate) async fn run(&mut self) { - loop { - tokio::select! { - _ = self.cancel.cancelled() => { - info!("received cancellation token"); - break - } - maybe_request = self.requests.next() => { - match maybe_request { - Some(request) => self.handle_blocks_request(request).await, - None => { - warn!("stopped receiving new requests"); - self.cancel.cancel(); - break - } - } - } - } - } - } -} diff --git a/common/nyxd-scraper-sqlite/src/constants.rs b/common/nyxd-scraper-sqlite/src/constants.rs deleted file mode 100644 index 9401cd360e..0000000000 --- a/common/nyxd-scraper-sqlite/src/constants.rs +++ /dev/null @@ -1,47 +0,0 @@ -// Copyright 2023 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use const_format::concatcp; - -// TODO: make those configurable via 'NymNetworkDetails' - -// BECH32_PREFIX defines the main SDK Bech32 prefix of an account's address -pub const BECH32_PREFIX: &str = "n"; - -// ACCOUNT_PREFIX is the prefix for account keys -pub const ACCOUNT_PREFIX: &str = "acc"; -// VALIDATOR_PREFIX is the prefix for validator keys -pub const VALIDATOR_PREFIX: &str = "val"; -// CONSENSUS_PREFIX is the prefix for consensus keys -pub const CONSENSUS_PREFIX: &str = "cons"; -// PUBKEY_PREFIX is the prefix for public keys -pub const PUBKEY_PREFIX: &str = "pub"; -// OPERATOR_PREFIX is the prefix for operator keys -pub const OPERATOR_PREFIX: &str = "oper"; -// ADDRESS_PREFIX is the prefix for addresses -pub const ADDRESS_PREFIX: &str = "addr"; - -// BECH32_ACCOUNT_ADDRESS_PREFIX defines the Bech32 prefix of an account's address -pub const BECH32_ACCOUNT_ADDRESS_PREFIX: &str = BECH32_PREFIX; -// BECH32_ACCOUNT_PUBKEY_PREFIX defines the Bech32 prefix of an account's public key -pub const BECH32_ACCOUNT_PUBKEY_PREFIX: &str = concatcp!(BECH32_PREFIX, PUBKEY_PREFIX); -// BECH32_VALIDATOR_ADDRESS_PREFIX defines the Bech32 prefix of a validator's operator address -pub const BECH32_VALIDATOR_ADDRESS_PREFIX: &str = - concatcp!(BECH32_PREFIX, VALIDATOR_PREFIX, OPERATOR_PREFIX); -// BECH32_VALIDATOR_PUBKEY_PREFIX defines the Bech32 prefix of a validator's operator public key -pub const BECH32_VALIDATOR_PUBKEY_PREFIX: &str = concatcp!( - BECH32_PREFIX, - VALIDATOR_PREFIX, - OPERATOR_PREFIX, - PUBKEY_PREFIX -); -// BECH32_CONSENSUS_ADDRESS_PREFIX defines the Bech32 prefix of a consensus node address -pub const BECH32_CONSENSUS_ADDRESS_PREFIX: &str = - concatcp!(BECH32_PREFIX, VALIDATOR_PREFIX, CONSENSUS_PREFIX); -// BECH32_CONESNSUS_PUBKEY_PREFIX defines the Bech32 prefix of a consensus node public key -pub const BECH32_CONESNSUS_PUBKEY_PREFIX: &str = concatcp!( - BECH32_PREFIX, - VALIDATOR_PREFIX, - CONSENSUS_PREFIX, - PUBKEY_PREFIX -); diff --git a/common/nyxd-scraper-sqlite/src/error.rs b/common/nyxd-scraper-sqlite/src/error.rs index 9f9bfd12d2..c91a2c2b8d 100644 --- a/common/nyxd-scraper-sqlite/src/error.rs +++ b/common/nyxd-scraper-sqlite/src/error.rs @@ -1,155 +1,36 @@ -// Copyright 2023 - Nym Technologies SA +// Copyright 2025 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::block_processor::pruning::{ - EVERYTHING_PRUNING_INTERVAL, EVERYTHING_PRUNING_KEEP_RECENT, -}; -use tendermint::Hash; +use nyxd_scraper_shared::helpers::MalformedDataError; +use nyxd_scraper_shared::storage::NyxdScraperStorageError; use thiserror::Error; -use tokio::sync::mpsc::error::SendError; -#[derive(Debug, Error)] -pub enum ScraperError { +#[derive(Error, Debug)] +pub enum SqliteScraperError { #[error("experienced internal database error: {0}")] - InternalDatabaseError(#[from] sqlx::Error), + InternalDatabaseError(#[from] sqlx::error::Error), #[error("failed to perform startup SQL migration: {0}")] StartupMigrationFailure(#[from] sqlx::migrate::MigrateError), - #[error("the block scraper is already running")] - ScraperAlreadyRunning, - - #[error("block information for height {height} is not available on the provided rpc endpoint")] - BlocksUnavailable { height: u32 }, - - #[error("failed to establish websocket connection to {url}: {source}")] - WebSocketConnectionFailure { - url: String, - #[source] - source: Box, - }, - - #[error("failed to establish rpc connection to {url}: {source}")] - HttpConnectionFailure { - url: String, - #[source] - source: Box, - }, - - #[error("failed to create chain subscription: {source}")] - ChainSubscriptionFailure { - #[source] - source: Box, - }, - - #[error("could not obtain basic block information at height: {height}: {source}")] - BlockQueryFailure { - height: u32, - #[source] - source: Box, - }, - - #[error("could not obtain block results information at height: {height}: {source}")] - BlockResultsQueryFailure { - height: u32, - #[source] - source: Box, - }, - - #[error("could not obtain validators information at height: {height}: {source}")] - ValidatorsQueryFailure { - height: u32, - #[source] - source: Box, - }, - - #[error("could not obtain tx results for tx: {hash}: {source}")] - TxResultsQueryFailure { - hash: Hash, - #[source] - source: Box, - }, - - #[error("could not obtain current abci info: {source}")] - AbciInfoQueryFailure { - #[source] - source: Box, - }, - - #[error("could not parse tx {hash}: {source}")] - TxParseFailure { - hash: Hash, - #[source] - source: cosmrs::ErrorReport, - }, - - #[error("could not parse msg in tx {hash} at index {index} into {type_url}: {source}")] - MsgParseFailure { - hash: Hash, - index: usize, - type_url: String, - #[source] - source: cosmrs::ErrorReport, - }, - - #[error("received an invalid chain subscription event of kind {kind} while we were waiting for new block data (query: '{query}')")] - InvalidSubscriptionEvent { query: String, kind: String }, - - #[error("received block data was empty (query: '{query}')")] - EmptyBlockData { query: String }, - - #[error("reached maximum number of allowed errors for subscription events")] - MaximumWebSocketFailures, - #[error("failed to begin storage tx: {source}")] StorageTxBeginFailure { #[source] - source: sqlx::Error, + source: sqlx::error::Error, }, #[error("failed to commit storage tx: {source}")] StorageTxCommitFailure { #[source] - source: sqlx::Error, + source: sqlx::error::Error, }, - #[error("failed to send on a closed channel")] - ClosedChannelError, - - #[error("failed to parse validator's address: {source}")] - MalformedValidatorAddress { - #[source] - source: eyre::Report, - }, - - #[error("failed to parse validator's address: {source}")] - MalformedValidatorPubkey { - #[source] - source: eyre::Report, - }, - - #[error( - "could not find the block proposer ('{proposer}') for height {height} in the validator set" - )] - BlockProposerNotInValidatorSet { height: u32, proposer: String }, - - #[error( - "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 }, + #[error(transparent)] + MalformedData(#[from] MalformedDataError), } -impl From> for ScraperError { - fn from(_: SendError) -> Self { - ScraperError::ClosedChannelError +impl From for NyxdScraperStorageError { + fn from(err: SqliteScraperError) -> Self { + NyxdScraperStorageError::new(err) } } diff --git a/common/nyxd-scraper-sqlite/src/helpers.rs b/common/nyxd-scraper-sqlite/src/helpers.rs deleted file mode 100644 index d11b25a2f3..0000000000 --- a/common/nyxd-scraper-sqlite/src/helpers.rs +++ /dev/null @@ -1,46 +0,0 @@ -// Copyright 2023 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::block_processor::types::ParsedTransactionResponse; -use crate::constants::{BECH32_CONESNSUS_PUBKEY_PREFIX, BECH32_CONSENSUS_ADDRESS_PREFIX}; -use crate::error::ScraperError; -use cosmrs::AccountId; -use sha2::{Digest, Sha256}; -use tendermint::{account, PublicKey}; -use tendermint::{validator, Hash}; -use tendermint_rpc::endpoint::validators; - -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 { - // 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()) - .map_err(|source| ScraperError::MalformedValidatorPubkey { source }) -} - -pub(crate) fn validator_consensus_address(id: account::Id) -> Result { - AccountId::new(BECH32_CONSENSUS_ADDRESS_PREFIX, id.as_ref()) - .map_err(|source| ScraperError::MalformedValidatorAddress { source }) -} - -pub(crate) fn tx_gas_sum(txs: &[ParsedTransactionResponse]) -> i64 { - txs.iter().map(|tx| tx.tx_result.gas_used).sum() -} - -pub(crate) fn validator_info( - id: account::Id, - validators: &validators::Response, -) -> Result<&validator::Info, ScraperError> { - match validators.validators.iter().find(|v| v.address == id) { - Some(info) => Ok(info), - None => { - let addr = validator_consensus_address(id)?; - Err(ScraperError::MissingValidatorInfoCommitted { - address: addr.to_string(), - }) - } - } -} diff --git a/common/nyxd-scraper-sqlite/src/lib.rs b/common/nyxd-scraper-sqlite/src/lib.rs index 1d01f4719f..d3068b5fd5 100644 --- a/common/nyxd-scraper-sqlite/src/lib.rs +++ b/common/nyxd-scraper-sqlite/src/lib.rs @@ -1,18 +1,7 @@ // Copyright 2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -pub(crate) mod block_processor; -pub(crate) mod block_requester; -pub mod constants; pub mod error; -pub(crate) mod helpers; -pub mod modules; -pub(crate) mod rpc_client; -pub(crate) mod scraper; pub mod storage; -pub use block_processor::pruning::{PruningOptions, PruningStrategy}; -pub use block_processor::types::ParsedTransactionResponse; -pub use modules::{BlockModule, MsgModule, TxModule}; -pub use scraper::{Config, NyxdScraper, StartingBlockOpts}; pub use storage::models; diff --git a/common/nyxd-scraper-sqlite/src/modules/block_module.rs b/common/nyxd-scraper-sqlite/src/modules/block_module.rs deleted file mode 100644 index 9ca1ba9b20..0000000000 --- a/common/nyxd-scraper-sqlite/src/modules/block_module.rs +++ /dev/null @@ -1,16 +0,0 @@ -// Copyright 2023 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::block_processor::types::FullBlockInformation; -use crate::error::ScraperError; -use crate::storage::StorageTransaction; -use async_trait::async_trait; - -#[async_trait] -pub trait BlockModule { - async fn handle_block( - &mut self, - block: &FullBlockInformation, - storage_tx: &mut StorageTransaction, - ) -> Result<(), ScraperError>; -} diff --git a/common/nyxd-scraper-sqlite/src/modules/mod.rs b/common/nyxd-scraper-sqlite/src/modules/mod.rs deleted file mode 100644 index 1f37d149e0..0000000000 --- a/common/nyxd-scraper-sqlite/src/modules/mod.rs +++ /dev/null @@ -1,10 +0,0 @@ -// Copyright 2023 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -mod block_module; -mod msg_module; -mod tx_module; - -pub use block_module::BlockModule; -pub use msg_module::MsgModule; -pub use tx_module::TxModule; diff --git a/common/nyxd-scraper-sqlite/src/modules/msg_module.rs b/common/nyxd-scraper-sqlite/src/modules/msg_module.rs deleted file mode 100644 index 1d195bee14..0000000000 --- a/common/nyxd-scraper-sqlite/src/modules/msg_module.rs +++ /dev/null @@ -1,21 +0,0 @@ -// Copyright 2023 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::block_processor::types::ParsedTransactionResponse; -use crate::error::ScraperError; -use crate::storage::StorageTransaction; -use async_trait::async_trait; -use cosmrs::Any; - -#[async_trait] -pub trait MsgModule { - fn type_url(&self) -> String; - - async fn handle_msg( - &mut self, - index: usize, - msg: &Any, - tx: &ParsedTransactionResponse, - storage_tx: &mut StorageTransaction, - ) -> Result<(), ScraperError>; -} diff --git a/common/nyxd-scraper-sqlite/src/modules/tx_module.rs b/common/nyxd-scraper-sqlite/src/modules/tx_module.rs deleted file mode 100644 index 07d012ab5a..0000000000 --- a/common/nyxd-scraper-sqlite/src/modules/tx_module.rs +++ /dev/null @@ -1,16 +0,0 @@ -// Copyright 2023 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::block_processor::types::ParsedTransactionResponse; -use crate::error::ScraperError; -use crate::storage::StorageTransaction; -use async_trait::async_trait; - -#[async_trait] -pub trait TxModule { - async fn handle_tx( - &mut self, - tx: &ParsedTransactionResponse, - storage_tx: &mut StorageTransaction, - ) -> Result<(), ScraperError>; -} diff --git a/common/nyxd-scraper-sqlite/src/rpc_client.rs b/common/nyxd-scraper-sqlite/src/rpc_client.rs deleted file mode 100644 index b20a89a098..0000000000 --- a/common/nyxd-scraper-sqlite/src/rpc_client.rs +++ /dev/null @@ -1,198 +0,0 @@ -// Copyright 2023 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::block_processor::types::{ - BlockToProcess, FullBlockInformation, ParsedTransactionResponse, -}; -use crate::error::ScraperError; -use crate::helpers::tx_hash; -use futures::future::join3; -use futures::StreamExt; -use std::collections::BTreeMap; -use std::sync::Arc; -use tendermint::Hash; -use tendermint_rpc::endpoint::{block, block_results, tx, validators}; -use tendermint_rpc::{Client, HttpClient, Paging}; -use tokio::sync::Mutex; -use tracing::{debug, instrument}; -use url::Url; - -#[derive(Clone)] -pub struct RpcClient { - // right now I don't care about anything nym specific, so a simple http client is sufficient, - // once this is inadequate, we can switch to a NyxdClient - inner: Arc, -} - -impl RpcClient { - pub fn new(url: &Url) -> Result { - let http_client = HttpClient::new(url.as_str()).map_err(|source| { - ScraperError::HttpConnectionFailure { - url: url.to_string(), - source: Box::new(source), - } - })?; - - Ok(RpcClient { - inner: Arc::new(http_client), - }) - } - - #[instrument(skip(self, block), fields(height = block.height))] - pub async fn try_get_full_details( - &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?; - let mut transactions = Vec::with_capacity(raw_transactions.len()); - for tx in raw_transactions { - transactions.push(ParsedTransactionResponse { - hash: tx.hash, - height: tx.height, - index: tx.index, - tx_result: tx.tx_result, - tx: cosmrs::Tx::from_bytes(&tx.tx).map_err(|source| { - ScraperError::TxParseFailure { - hash: tx.hash, - source, - } - })?, - proof: tx.proof, - }) - } - - Ok(FullBlockInformation { - block: block.block, - results: results?, - validators: validators?, - transactions, - }) - } - - #[instrument(skip(self), err(Display))] - pub async fn get_basic_block_details( - &self, - height: u32, - ) -> Result { - debug!("getting basic block details"); - - self.inner - .block(height) - .await - .map_err(|source| ScraperError::BlockQueryFailure { - height, - source: Box::new(source), - }) - } - - #[instrument(skip(self), err(Display))] - pub async fn get_block_results( - &self, - height: u32, - ) -> Result { - debug!("getting block results"); - - self.inner.block_results(height).await.map_err(|source| { - ScraperError::BlockResultsQueryFailure { - height, - source: Box::new(source), - } - }) - } - - pub(crate) async fn current_block_height(&self) -> Result { - debug!("getting current block height"); - - let info = - self.inner - .abci_info() - .await - .map_err(|source| ScraperError::AbciInfoQueryFailure { - source: Box::new(source), - })?; - Ok(info.last_block_height.value()) - } - - pub(crate) async fn earliest_available_block_height(&self) -> Result { - debug!("getting earliest available block height"); - - let status = - self.inner - .status() - .await - .map_err(|source| ScraperError::AbciInfoQueryFailure { - source: Box::new(source), - })?; - Ok(status.sync_info.earliest_block_height.value()) - } - - async fn get_transaction_results( - &self, - raw: &[Vec], - ) -> Result, ScraperError> { - let ordered_results = Arc::new(Mutex::new(BTreeMap::new())); - - // "Data is just a wrapper for a list of transactions, where transactions are arbitrary byte arrays" - // source: https://github.com/tendermint/spec/blob/d46cd7f573a2c6a2399fcab2cde981330aa63f37/spec/core/data_structures.md#data - // - // I hate that zip as much as you, dear reader, but for some reason the compiler didn't let me remove the `move` - futures::stream::iter( - raw.iter() - .map(tx_hash) - .enumerate() - .zip(std::iter::repeat(ordered_results.clone())), - ) - .for_each_concurrent(4, |((id, tx_hash), ordered_results)| async move { - let res = self.get_transaction_result(tx_hash).await; - ordered_results.lock().await.insert(id, res); - }) - .await; - - // safety the futures have completed so we MUST have the only arc reference - #[allow(clippy::unwrap_used)] - let inner = Arc::into_inner(ordered_results).unwrap().into_inner(); - - // BTreeMap is ordered by its keys so we're guaranteed to get txs in correct order - inner.into_values().collect() - } - - #[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"); - - self.inner - .tx(tx_hash, false) - .await - .map_err(|source| ScraperError::TxResultsQueryFailure { - hash: tx_hash, - source: Box::new(source), - }) - } - - #[instrument(skip(self))] - pub async fn get_validators_details( - &self, - height: u32, - ) -> Result { - debug!("getting validators set"); - - self.inner - .validators(height, Paging::All) - .await - .map_err(|source| ScraperError::ValidatorsQueryFailure { - height, - source: Box::new(source), - }) - } -} diff --git a/common/nyxd-scraper-sqlite/src/scraper/mod.rs b/common/nyxd-scraper-sqlite/src/scraper/mod.rs deleted file mode 100644 index 71cb535645..0000000000 --- a/common/nyxd-scraper-sqlite/src/scraper/mod.rs +++ /dev/null @@ -1,407 +0,0 @@ -// Copyright 2023 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::block_processor::types::BlockToProcess; -use crate::block_processor::{BlockProcessor, BlockProcessorConfig}; -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::ScraperStorage; -use crate::PruningOptions; -use futures::future::join_all; -use std::path::PathBuf; -use std::sync::Arc; -use tokio::sync::mpsc::{ - channel, unbounded_channel, Receiver, Sender, UnboundedReceiver, UnboundedSender, -}; -use tokio::sync::Notify; -use tokio_util::sync::CancellationToken; -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, - - /// If the scraper fails to start from the desired height, rather than failing, - /// attempt to use the next available height - pub use_best_effort_start_height: bool, -} - -pub struct Config { - /// 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 database_path: PathBuf, - - pub pruning_options: PruningOptions, - - pub store_precommits: bool, - - pub start_block: StartingBlockOpts, -} - -pub struct NyxdScraperBuilder { - config: Config, - - block_modules: Vec>, - tx_modules: Vec>, - msg_modules: Vec>, -} - -impl NyxdScraperBuilder { - pub async fn build_and_start(self) -> Result { - let scraper = NyxdScraper::new(self.config).await?; - - let (processing_tx, processing_rx) = unbounded_channel(); - let (req_tx, req_rx) = channel(5); - - let rpc_client = RpcClient::new(&scraper.config.rpc_url)?; - - // create the tasks - let block_requester = BlockRequester::new( - scraper.cancel_token.clone(), - rpc_client.clone(), - req_rx, - processing_tx.clone(), - ); - - let block_processor_config = BlockProcessorConfig::new( - scraper.config.pruning_options, - scraper.config.store_precommits, - scraper.config.start_block.start_block_height, - scraper.config.start_block.use_best_effort_start_height, - ); - - 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, - ) - .await?; - 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( - &scraper.config.websocket_url, - scraper.cancel_token.clone(), - scraper.task_tracker.clone(), - processing_tx, - ) - .await?; - - scraper.start_tasks(block_requester, block_processor, chain_subscriber); - - Ok(scraper) - } - - pub fn new(config: Config) -> Self { - NyxdScraperBuilder { - config, - block_modules: vec![], - tx_modules: vec![], - msg_modules: vec![], - } - } - - pub fn with_block_module(mut self, module: M) -> Self { - self.block_modules.push(Box::new(module)); - self - } - - pub fn with_tx_module(mut self, module: M) -> Self { - self.tx_modules.push(Box::new(module)); - self - } - - pub fn with_msg_module(mut self, module: M) -> Self { - self.msg_modules.push(Box::new(module)); - self - } -} - -pub struct NyxdScraper { - config: Config, - - task_tracker: TaskTracker, - cancel_token: CancellationToken, - startup_sync: Arc, - storage: ScraperStorage, - rpc_client: RpcClient, -} - -impl NyxdScraper { - pub fn builder(config: Config) -> NyxdScraperBuilder { - NyxdScraperBuilder::new(config) - } - - pub async fn new(config: Config) -> Result { - config.pruning_options.validate()?; - let storage = ScraperStorage::init(&config.database_path).await?; - let rpc_client = RpcClient::new(&config.rpc_url)?; - - Ok(NyxdScraper { - config, - task_tracker: TaskTracker::new(), - cancel_token: CancellationToken::new(), - startup_sync: Arc::new(Default::default()), - storage, - rpc_client, - }) - } - - pub fn storage(&self) -> ScraperStorage { - self.storage.clone() - } - - 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(); - } - - // DO NOT USE UNLESS YOU KNOW EXACTLY WHAT YOU'RE DOING - // AS THIS WILL NOT USE ANY OF YOUR REGISTERED MODULES - // YOU WILL BE FIRED IF YOU USE IT : ) - pub async fn unsafe_process_single_block(&self, height: u32) -> Result<(), ScraperError> { - info!(height = height, "attempting to process a single block"); - if !self.task_tracker.is_empty() { - return Err(ScraperError::ScraperAlreadyRunning); - } - - let (_, processing_rx) = unbounded_channel(); - let (req_tx, _) = channel(5); - - let mut block_processor = self - .new_block_processor(req_tx.clone(), processing_rx) - .await? - .with_pruning(PruningOptions::nothing()); - - let block = self.rpc_client.get_basic_block_details(height).await?; - - block_processor.process_block(block.into()).await - } - - // DO NOT USE UNLESS YOU KNOW EXACTLY WHAT YOU'RE DOING - // AS THIS WILL NOT USE ANY OF YOUR REGISTERED MODULES - // YOU WILL BE FIRED IF YOU USE IT : ) - pub async fn unsafe_process_block_range( - &self, - starting_height: Option, - end_height: Option, - ) -> Result<(), ScraperError> { - if !self.task_tracker.is_empty() { - return Err(ScraperError::ScraperAlreadyRunning); - } - - let (_, processing_rx) = unbounded_channel(); - let (req_tx, _) = channel(5); - - let mut block_processor = self - .new_block_processor(req_tx.clone(), processing_rx) - .await? - .with_pruning(PruningOptions::nothing()); - - let mut current_height = self.rpc_client.current_block_height().await? as u32; - let last_processed = block_processor.last_process_height(); - - let mut starting_height = match starting_height { - // always attempt to use whatever the user has provided - Some(explicit) => explicit, - None => { - // otherwise, attempt to resume where we last stopped - // and if we haven't processed anything, start from the current height - if last_processed != 0 { - last_processed - } else { - current_height - } - } - }; - - let must_catch_up = end_height.is_none(); - let mut end_height = match end_height { - // always attempt to use whatever the user has provided - Some(explicit) => explicit, - None => { - // otherwise, attempt to either go from the start height to the height right - // before the final processed block held in the storage (in case there are gaps) - // or finally, just go to the current block height - if last_processed > starting_height { - last_processed - 1 - } else { - current_height - } - } - }; - - let mut last_processed = starting_height; - - while last_processed < current_height { - info!( - starting_height = starting_height, - end_height = end_height, - "attempting to process block range" - ); - - let range = (starting_height..=end_height).collect::>(); - - // the most likely bottleneck here are going to be the chain queries, - // so batch multiple requests - for batch in range.chunks(4) { - let batch_result = join_all( - batch - .iter() - .map(|height| self.rpc_client.get_basic_block_details(*height)), - ) - .await; - for result in batch_result { - match result { - Ok(block) => block_processor.process_block(block.into()).await?, - Err(err) => { - error!("failed to retrieve the block: {err}. stopping..."); - return Err(err); - } - } - } - } - - // if we don't need to catch up, return early - if !must_catch_up { - return Ok(()); - } - - // check if we have caught up to the current block height - last_processed = end_height; - current_height = self.rpc_client.current_block_height().await? as u32; - - info!( - last_processed = last_processed, - current_height = current_height, - "🏃 still need to catch up..." - ); - - starting_height = last_processed + 1; - end_height = current_height; - } - - if must_catch_up { - info!( - last_processed = last_processed, - current_height = current_height, - "✅ block processing has caught up!" - ); - } - - Ok(()) - } - - fn new_block_requester( - &self, - req_rx: Receiver, - processing_tx: UnboundedSender, - ) -> BlockRequester { - BlockRequester::new( - self.cancel_token.clone(), - self.rpc_client.clone(), - req_rx, - processing_tx.clone(), - ) - } - - async fn new_block_processor( - &self, - req_tx: Sender, - processing_rx: UnboundedReceiver, - ) -> Result { - let block_processor_config = BlockProcessorConfig::new( - self.config.pruning_options, - self.config.store_precommits, - self.config.start_block.start_block_height, - 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 - } - - async fn new_chain_subscriber( - &self, - processing_tx: UnboundedSender, - ) -> Result { - ChainSubscriber::new( - &self.config.websocket_url, - self.cancel_token.clone(), - self.task_tracker.clone(), - processing_tx, - ) - .await - } - - pub async fn start(&self) -> Result<(), ScraperError> { - let (processing_tx, processing_rx) = unbounded_channel(); - let (req_tx, req_rx) = channel(5); - - // 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 chain_subscriber = self.new_chain_subscriber(processing_tx).await?; - - // spawn them - self.start_tasks(block_requester, block_processor, chain_subscriber); - - Ok(()) - } - - pub async fn wait_for_startup_sync(&self) { - info!("awaiting startup chain sync"); - self.startup_sync.notified().await - } - - pub async fn stop(self) { - info!("stopping the chain scraper"); - 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/common/nyxd-scraper-sqlite/src/scraper/subscriber.rs b/common/nyxd-scraper-sqlite/src/scraper/subscriber.rs deleted file mode 100644 index f3a9411647..0000000000 --- a/common/nyxd-scraper-sqlite/src/scraper/subscriber.rs +++ /dev/null @@ -1,253 +0,0 @@ -// Copyright 2023 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::block_processor::types::BlockToProcess; -use crate::error::ScraperError; -use tendermint_rpc::client::CompatMode; -use tendermint_rpc::event::Event; -use tendermint_rpc::query::EventType; -use tendermint_rpc::{SubscriptionClient, WebSocketClient, WebSocketClientDriver}; -use time::{Duration, OffsetDateTime}; -use tokio::sync::mpsc::UnboundedSender; -use tokio_stream::StreamExt; -use tokio_util::sync::CancellationToken; -use tokio_util::task::TaskTracker; -use tracing::{error, info, warn}; -use url::Url; - -const MAX_FAILURES: usize = 10; -const MAX_RECONNECTION_ATTEMPTS: usize = 8; -const SOCKET_FAILURE_RESET: Duration = Duration::minutes(15); - -pub struct ChainSubscriber { - cancel: CancellationToken, - task_tracker: TaskTracker, - - block_sender: UnboundedSender, - - websocket_endpoint: Url, - websocket_client: WebSocketClient, - websocket_driver: Option, -} - -impl ChainSubscriber { - pub async fn new( - websocket_endpoint: &Url, - cancel: CancellationToken, - task_tracker: TaskTracker, - block_sender: UnboundedSender, - ) -> Result { - // sure, we could have just used websocket client entirely, but let's keep the logic for - // getting current blocks and historical blocks completely separate with the dual connection - let websocket_url = websocket_endpoint.as_str().try_into().map_err(|source| { - ScraperError::WebSocketConnectionFailure { - url: websocket_endpoint.to_string(), - source: Box::new(source), - } - })?; - - let (client, driver) = WebSocketClient::builder(websocket_url) - .compat_mode(CompatMode::V0_37) - .build() - .await - .map_err(|source| ScraperError::WebSocketConnectionFailure { - url: websocket_endpoint.to_string(), - source: Box::new(source), - })?; - - Ok(ChainSubscriber { - cancel, - task_tracker, - block_sender, - websocket_endpoint: websocket_endpoint.clone(), - websocket_client: client, - websocket_driver: Some(driver), - }) - } - - fn handle_new_event(&mut self, event: Event) -> Result<(), ScraperError> { - if let Err(err) = self.block_sender.send(event.try_into()?) { - // this error has nothing to do with the websocket or chain - error!("failed to send block for processing: {err} - are we shutting down?") - } - Ok(()) - } - - async fn remake_connection(&mut self) -> Result<(), ScraperError> { - info!( - "attempting to reestablish connection to {}", - self.websocket_endpoint - ); - - let (client, driver) = WebSocketClient::new(self.websocket_endpoint.as_str()) - .await - .map_err(|source| ScraperError::WebSocketConnectionFailure { - url: self.websocket_endpoint.to_string(), - source: Box::new(source), - })?; - self.websocket_client = client; - self.websocket_driver = Some(driver); - - info!( - "managed to reestablish the websocket connection to {}", - self.websocket_endpoint - ); - Ok(()) - } - - /// Returns whether the method exited due to the cancellation - async fn run_chain_subscription(&mut self) -> Result { - let Some(ws_driver) = self.websocket_driver.take() else { - error!("the websocket driver hasn't been created - we probably failed to establish the connection"); - return Ok(false); - }; - - let driver_cancel = CancellationToken::new(); - let _driver_guard = driver_cancel.clone().drop_guard(); - - // spawn the websocket driver task - let driver_handle = { - self.task_tracker.reopen(); - let handle = self - .task_tracker - .spawn(run_websocket_driver(ws_driver, driver_cancel)); - self.task_tracker.close(); - handle - }; - tokio::pin!(driver_handle); - - info!("creating chain subscription"); - let mut subs = self - .websocket_client - .subscribe(EventType::NewBlock.into()) - .await - .map_err(|source| ScraperError::ChainSubscriptionFailure { - source: Box::new(source), - })?; - - let mut failures = 0; - - info!("starting processing loop"); - loop { - tokio::select! { - _ = self.cancel.cancelled() => { - info!("received cancellation token"); - // note: `_driver_guard` will get dropped here thus causing cancellation of the driver task - return Ok(true) - } - _ = &mut driver_handle => { - error!("our websocket driver has finished execution"); - return Ok(self.cancel.is_cancelled()) - } - maybe_event = subs.next() => { - let Some(maybe_event) = maybe_event else { - warn!("stopped receiving new events"); - return Ok(false) - }; - match maybe_event { - Ok(event) => { - if let Err(err) = self.handle_new_event(event) { - error!("failed to process received block: {err}"); - failures += 1 - } else { - failures = 0; - } - } - Err(err) => { - error!("failed to receive a valid subscription event: {err}"); - failures += 1 - } - } - if failures >= MAX_FAILURES { - return Ok(false) - } - } - } - } - } - - async fn websocket_backoff(&mut self, failure_count: usize) -> bool { - const MINIMUM_WAIT_MS: u64 = 10_000; - const INCREMENTAL_WAIT_MS: u64 = 30_000; - - let backoff_duration_ms = MINIMUM_WAIT_MS + INCREMENTAL_WAIT_MS * failure_count as u64; - info!("going to wait {backoff_duration_ms} ms before re-attempting the reconnection"); - - tokio::select! { - _ = self.cancel.cancelled() => { - info!("received cancellation token"); - true - } - _ = tokio::time::sleep(std::time::Duration::from_millis(backoff_duration_ms)) => false, - } - } - - pub(crate) async fn run(&mut self) -> Result<(), ScraperError> { - let _drop_guard = self.cancel.clone().drop_guard(); - let mut socket_failures = 0; - let mut last_failure = OffsetDateTime::now_utc(); - - loop { - if self.cancel.is_cancelled() { - return Ok(()); - } - - match self.run_chain_subscription().await { - Ok(cancelled) => { - if cancelled { - // we're in the middle of a shutdown - return Ok(()); - } - socket_failures += 1; - } - Err(err) => { - error!("failed to create chain subscription: {err}"); - socket_failures += 1; - } - } - - warn!("current socket failure count: {socket_failures}. the last failure was at {last_failure}"); - - let now = OffsetDateTime::now_utc(); - - // if it's been a while since the last failure, reset the count - if now - last_failure > SOCKET_FAILURE_RESET { - warn!("resetting the failure count to 1"); - socket_failures = 1; - } - last_failure = now; - - if socket_failures >= MAX_RECONNECTION_ATTEMPTS { - error!("reached the maximum allowed failure count"); - return Err(ScraperError::MaximumWebSocketFailures); - } - - // BACKOFF - let cancelled = self.websocket_backoff(socket_failures).await; - if cancelled { - return Ok(()); - } - - if let Err(err) = self.remake_connection().await { - error!("failed to re-establish the websocket connection: {err}"); - } - } - } -} - -pub async fn run_websocket_driver(driver: WebSocketClientDriver, driver_cancel: CancellationToken) { - info!("starting websocket driver"); - tokio::select! { - _ = driver_cancel.cancelled() => { - info!("received cancellation token") - } - res = driver.run() => { - match res { - Ok(_) => info!("our websocket driver has finished execution"), - Err(err) => { - error!("our websocket driver has errored out: {err}"); - } - } - } - } -} diff --git a/common/nyxd-scraper-sqlite/src/storage/block_storage.rs b/common/nyxd-scraper-sqlite/src/storage/block_storage.rs new file mode 100644 index 0000000000..0f416f1963 --- /dev/null +++ b/common/nyxd-scraper-sqlite/src/storage/block_storage.rs @@ -0,0 +1,251 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::error::SqliteScraperError; +use crate::models::{CommitSignature, Validator}; +use crate::storage::manager::{ + prune_blocks, prune_messages, prune_pre_commits, prune_transactions, update_last_pruned, + StorageManager, +}; +use crate::storage::transaction::SqliteStorageTransaction; +use async_trait::async_trait; +use nyxd_scraper_shared::storage::helpers::log_db_operation_time; +use nyxd_scraper_shared::storage::{NyxdScraperStorage, NyxdScraperStorageError}; +use sqlx::sqlite::{SqliteAutoVacuum, SqliteSynchronous}; +use sqlx::types::time::OffsetDateTime; +use sqlx::ConnectOptions; +use std::fmt::Debug; +use std::path::Path; +use tokio::time::Instant; +use tracing::{debug, error, info, instrument}; + +#[derive(Clone)] +pub struct SqliteScraperStorage { + pub(crate) manager: StorageManager, +} + +impl SqliteScraperStorage { + #[instrument] + pub async fn init + Debug>( + database_path: P, + ) -> Result { + let database_path = database_path.as_ref(); + debug!( + "initialising scraper database path to '{}'", + database_path.display() + ); + + let opts = sqlx::sqlite::SqliteConnectOptions::new() + .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal) + .synchronous(SqliteSynchronous::Normal) + .auto_vacuum(SqliteAutoVacuum::Incremental) + .filename(database_path) + .create_if_missing(true) + .disable_statement_logging(); + + // TODO: do we want auto_vacuum ? + + let connection_pool = match sqlx::SqlitePool::connect_with(opts).await { + Ok(db) => db, + Err(err) => { + error!("Failed to connect to SQLx database: {err}"); + return Err(err.into()); + } + }; + + if let Err(err) = sqlx::migrate!("./sql_migrations") + .run(&connection_pool) + .await + { + error!("Failed to initialize SQLx database: {err}"); + return Err(err.into()); + } + + info!("Database migration finished!"); + + let manager = StorageManager { connection_pool }; + manager.set_initial_metadata().await?; + + let storage = SqliteScraperStorage { manager }; + + Ok(storage) + } + + #[instrument(skip(self))] + pub async fn prune_storage( + &self, + oldest_to_keep: u32, + current_height: u32, + ) -> Result<(), SqliteScraperError> { + 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.0.commit() + .await + .map_err(|source| SqliteScraperError::StorageTxCommitFailure { source })?; + log_db_operation_time("committing pruning tx", commit_start); + + log_db_operation_time("pruning storage", start); + Ok(()) + } + + #[instrument(skip_all)] + pub async fn begin_processing_tx( + &self, + ) -> Result { + debug!("starting storage tx"); + self.manager + .connection_pool + .begin() + .await + .map(SqliteStorageTransaction) + .map_err(|source| SqliteScraperError::StorageTxBeginFailure { source }) + } + + pub async fn lowest_block_height(&self) -> Result, SqliteScraperError> { + Ok(self.manager.get_lowest_block().await?) + } + + pub async fn get_first_block_height_after( + &self, + time: OffsetDateTime, + ) -> Result, SqliteScraperError> { + Ok(self.manager.get_first_block_height_after(time).await?) + } + + pub async fn get_last_block_height_before( + &self, + time: OffsetDateTime, + ) -> Result, SqliteScraperError> { + Ok(self.manager.get_last_block_height_before(time).await?) + } + + pub async fn get_blocks_between( + &self, + start_time: OffsetDateTime, + end_time: OffsetDateTime, + ) -> Result { + let Some(block_start) = self.get_first_block_height_after(start_time).await? else { + return Ok(0); + }; + let Some(block_end) = self.get_last_block_height_before(end_time).await? else { + return Ok(0); + }; + + Ok(block_end - block_start) + } + + pub async fn get_signed_between( + &self, + consensus_address: &str, + start_height: i64, + end_height: i64, + ) -> Result { + Ok(self + .manager + .get_signed_between(consensus_address, start_height, end_height) + .await?) + } + + pub async fn get_signed_between_times( + &self, + consensus_address: &str, + start_time: OffsetDateTime, + end_time: OffsetDateTime, + ) -> Result { + let Some(block_start) = self.get_first_block_height_after(start_time).await? else { + return Ok(0); + }; + let Some(block_end) = self.get_last_block_height_before(end_time).await? else { + return Ok(0); + }; + + self.get_signed_between(consensus_address, block_start, block_end) + .await + } + + pub async fn get_precommit( + &self, + consensus_address: &str, + height: i64, + ) -> Result, SqliteScraperError> { + Ok(self + .manager + .get_precommit(consensus_address, height) + .await?) + } + + pub async fn get_block_signers( + &self, + height: i64, + ) -> Result, SqliteScraperError> { + Ok(self.manager.get_block_validators(height).await?) + } + + pub async fn get_all_known_validators(&self) -> Result, SqliteScraperError> { + Ok(self.manager.get_validators().await?) + } + + pub async fn get_last_processed_height(&self) -> Result { + Ok(self.manager.get_last_processed_height().await?) + } + + pub async fn get_pruned_height(&self) -> Result { + Ok(self.manager.get_pruned_height().await?) + } +} + +#[async_trait] +impl NyxdScraperStorage for SqliteScraperStorage { + type StorageTransaction = SqliteStorageTransaction; + + async fn initialise(storage_path: &Path) -> Result { + SqliteScraperStorage::init(&storage_path) + .await + .map_err(NyxdScraperStorageError::from) + } + + async fn begin_processing_tx( + &self, + ) -> Result { + self.begin_processing_tx() + .await + .map_err(NyxdScraperStorageError::from) + } + + async fn get_last_processed_height(&self) -> Result { + self.get_last_processed_height() + .await + .map_err(NyxdScraperStorageError::from) + } + + async fn get_pruned_height(&self) -> Result { + self.get_pruned_height() + .await + .map_err(NyxdScraperStorageError::from) + } + + async fn lowest_block_height(&self) -> Result, NyxdScraperStorageError> { + self.lowest_block_height() + .await + .map_err(NyxdScraperStorageError::from) + } + + async fn prune_storage( + &self, + oldest_to_keep: u32, + current_height: u32, + ) -> Result<(), NyxdScraperStorageError> { + self.prune_storage(oldest_to_keep, current_height) + .await + .map_err(NyxdScraperStorageError::from) + } +} diff --git a/common/nyxd-scraper-sqlite/src/storage/helpers.rs b/common/nyxd-scraper-sqlite/src/storage/helpers.rs deleted file mode 100644 index 38dc36ebc3..0000000000 --- a/common/nyxd-scraper-sqlite/src/storage/helpers.rs +++ /dev/null @@ -1,2 +0,0 @@ -// Copyright 2023 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 diff --git a/common/nyxd-scraper-sqlite/src/storage/mod.rs b/common/nyxd-scraper-sqlite/src/storage/mod.rs index 754eba45ee..b255c0c0f3 100644 --- a/common/nyxd-scraper-sqlite/src/storage/mod.rs +++ b/common/nyxd-scraper-sqlite/src/storage/mod.rs @@ -1,394 +1,9 @@ // Copyright 2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::{ - block_processor::types::{FullBlockInformation, ParsedTransactionResponse}, - error::ScraperError, - storage::{ - manager::{ - insert_block, insert_message, insert_precommit, insert_transaction, insert_validator, - prune_blocks, prune_messages, prune_pre_commits, prune_transactions, - update_last_processed, update_last_pruned, StorageManager, - }, - models::{CommitSignature, Validator}, - }, -}; -use sqlx::{ - sqlite::{SqliteAutoVacuum, SqliteSynchronous}, - types::time::OffsetDateTime, - ConnectOptions, Sqlite, Transaction, -}; -use std::{fmt::Debug, path::Path}; -use tendermint::{ - block::{Commit, CommitSig}, - Block, -}; -use tendermint_rpc::endpoint::validators; -use tokio::time::Instant; -use tracing::{debug, error, info, instrument, trace, warn}; +use nyxd_scraper_shared::storage::helpers::log_db_operation_time; -mod helpers; +pub mod block_storage; mod manager; pub mod models; - -pub type StorageTransaction = Transaction<'static, Sqlite>; - -#[derive(Clone)] -pub struct ScraperStorage { - pub(crate) manager: StorageManager, -} - -pub(crate) fn log_db_operation_time(op_name: &str, start_time: Instant) { - let elapsed = start_time.elapsed(); - let formatted = humantime::format_duration(elapsed); - - match elapsed.as_millis() { - v if v > 10000 => error!("{op_name} took {formatted} to execute"), - v if v > 1000 => warn!("{op_name} took {formatted} to execute"), - v if v > 100 => info!("{op_name} took {formatted} to execute"), - v if v > 10 => debug!("{op_name} took {formatted} to execute"), - _ => trace!("{op_name} took {formatted} to execute"), - } -} - -impl ScraperStorage { - #[instrument] - pub async fn init + Debug>(database_path: P) -> Result { - let database_path = database_path.as_ref(); - debug!( - "initialising scraper database path to '{}'", - database_path.display() - ); - - let opts = sqlx::sqlite::SqliteConnectOptions::new() - .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal) - .synchronous(SqliteSynchronous::Normal) - .auto_vacuum(SqliteAutoVacuum::Incremental) - .filename(database_path) - .create_if_missing(true) - .disable_statement_logging(); - - // TODO: do we want auto_vacuum ? - - let connection_pool = match sqlx::SqlitePool::connect_with(opts).await { - Ok(db) => db, - Err(err) => { - error!("Failed to connect to SQLx database: {err}"); - return Err(err.into()); - } - }; - - if let Err(err) = sqlx::migrate!("./sql_migrations") - .run(&connection_pool) - .await - { - error!("Failed to initialize SQLx database: {err}"); - return Err(err.into()); - } - - info!("Database migration finished!"); - - let manager = StorageManager { connection_pool }; - manager.set_initial_metadata().await?; - - let storage = ScraperStorage { manager }; - - Ok(storage) - } - - #[instrument(skip(self))] - pub async fn prune_storage( - &self, - oldest_to_keep: u32, - current_height: u32, - ) -> Result<(), ScraperError> { - let start = Instant::now(); - - let mut tx = self.begin_processing_tx().await?; - - prune_messages(oldest_to_keep.into(), &mut *tx).await?; - prune_transactions(oldest_to_keep.into(), &mut *tx).await?; - prune_pre_commits(oldest_to_keep.into(), &mut *tx).await?; - prune_blocks(oldest_to_keep.into(), &mut *tx).await?; - update_last_pruned(current_height.into(), &mut *tx).await?; - - let commit_start = Instant::now(); - tx.commit() - .await - .map_err(|source| ScraperError::StorageTxCommitFailure { source })?; - log_db_operation_time("committing pruning tx", commit_start); - - log_db_operation_time("pruning storage", start); - Ok(()) - } - - #[instrument(skip_all)] - pub async fn begin_processing_tx(&self) -> Result { - debug!("starting storage tx"); - self.manager - .connection_pool - .begin() - .await - .map_err(|source| ScraperError::StorageTxBeginFailure { source }) - } - - pub async fn lowest_block_height(&self) -> Result, ScraperError> { - Ok(self.manager.get_lowest_block().await?) - } - - pub async fn get_first_block_height_after( - &self, - time: OffsetDateTime, - ) -> Result, ScraperError> { - Ok(self.manager.get_first_block_height_after(time).await?) - } - - pub async fn get_last_block_height_before( - &self, - time: OffsetDateTime, - ) -> Result, ScraperError> { - Ok(self.manager.get_last_block_height_before(time).await?) - } - - pub async fn get_blocks_between( - &self, - start_time: OffsetDateTime, - end_time: OffsetDateTime, - ) -> Result { - let Some(block_start) = self.get_first_block_height_after(start_time).await? else { - return Ok(0); - }; - let Some(block_end) = self.get_last_block_height_before(end_time).await? else { - return Ok(0); - }; - - Ok(block_end - block_start) - } - - pub async fn get_signed_between( - &self, - consensus_address: &str, - start_height: i64, - end_height: i64, - ) -> Result { - Ok(self - .manager - .get_signed_between(consensus_address, start_height, end_height) - .await?) - } - - pub async fn get_signed_between_times( - &self, - consensus_address: &str, - start_time: OffsetDateTime, - end_time: OffsetDateTime, - ) -> Result { - let Some(block_start) = self.get_first_block_height_after(start_time).await? else { - return Ok(0); - }; - let Some(block_end) = self.get_last_block_height_before(end_time).await? else { - return Ok(0); - }; - - self.get_signed_between(consensus_address, block_start, block_end) - .await - } - - pub async fn get_precommit( - &self, - consensus_address: &str, - height: i64, - ) -> Result, ScraperError> { - Ok(self - .manager - .get_precommit(consensus_address, height) - .await?) - } - - pub async fn get_block_signers(&self, height: i64) -> Result, ScraperError> { - Ok(self.manager.get_block_validators(height).await?) - } - - pub async fn get_all_known_validators(&self) -> Result, ScraperError> { - Ok(self.manager.get_validators().await?) - } - - pub async fn get_last_processed_height(&self) -> Result { - Ok(self.manager.get_last_processed_height().await?) - } - - pub async fn get_pruned_height(&self) -> Result { - Ok(self.manager.get_pruned_height().await?) - } -} - -pub async fn persist_block( - block: &FullBlockInformation, - tx: &mut StorageTransaction, - store_precommits: bool, -) -> Result<(), ScraperError> { - 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()?; - - // persist validators - persist_validators(&block.validators, tx).await?; - - // persist block data - persist_block_data(&block.block, total_gas, tx).await?; - - if store_precommits { - if let Some(commit) = &block.block.last_commit { - persist_commits(commit, &block.validators, tx).await?; - } else { - warn!("no commits for block {}", block.block.header.height) - } - } - - // persist txs - persist_txs(&block.transactions, tx).await?; - - // persist messages (inside the transactions) - persist_messages(&block.transactions, tx).await?; - - update_last_processed(block.block.header.height.into(), tx.as_mut()).await?; - - Ok(()) -} - -async fn persist_validators( - validators: &validators::Response, - tx: &mut StorageTransaction, -) -> Result<(), ScraperError> { - debug!("persisting {} validators", validators.total); - for validator in &validators.validators { - let consensus_address = crate::helpers::validator_consensus_address(validator.address)?; - let consensus_pubkey = crate::helpers::validator_pubkey_to_bech32(validator.pub_key)?; - - insert_validator( - consensus_address.to_string(), - consensus_pubkey.to_string(), - tx.as_mut(), - ) - .await?; - } - - Ok(()) -} - -async fn persist_block_data( - block: &Block, - total_gas: i64, - tx: &mut StorageTransaction, -) -> Result<(), ScraperError> { - let proposer_address = - crate::helpers::validator_consensus_address(block.header.proposer_address)?.to_string(); - - insert_block( - block.header.height.into(), - block.header.hash().to_string(), - block.data.len() as u32, - total_gas, - proposer_address, - block.header.time.into(), - tx.as_mut(), - ) - .await?; - Ok(()) -} - -async fn persist_commits( - commits: &Commit, - validators: &validators::Response, - tx: &mut StorageTransaction, -) -> Result<(), ScraperError> { - debug!("persisting up to {} commits", commits.signatures.len()); - let height: i64 = commits.height.into(); - - for commit_sig in &commits.signatures { - let (validator_id, timestamp, signature) = match commit_sig { - CommitSig::BlockIdFlagAbsent => { - trace!("absent signature"); - continue; - } - CommitSig::BlockIdFlagCommit { - validator_address, - timestamp, - signature, - } => (validator_address, timestamp, signature), - CommitSig::BlockIdFlagNil { - validator_address, - timestamp, - signature, - } => (validator_address, timestamp, signature), - }; - - let validator = crate::helpers::validator_info(*validator_id, validators)?; - let validator_address = crate::helpers::validator_consensus_address(*validator_id)?; - - if signature.is_none() { - warn!("empty signature for {validator_address} at height {height}"); - continue; - } - - insert_precommit( - validator_address.to_string(), - height, - (*timestamp).into(), - validator.power.into(), - validator.proposer_priority.value(), - tx.as_mut(), - ) - .await?; - } - - Ok(()) -} - -async fn persist_txs( - txs: &[ParsedTransactionResponse], - tx: &mut StorageTransaction, -) -> Result<(), ScraperError> { - debug!("persisting {} txs", txs.len()); - - for chain_tx in txs { - insert_transaction( - chain_tx.hash.to_string(), - chain_tx.height.into(), - chain_tx.index as i64, - chain_tx.tx_result.code.is_ok(), - chain_tx.tx.body.messages.len() as i64, - chain_tx.tx.body.memo.clone(), - chain_tx.tx_result.gas_wanted, - chain_tx.tx_result.gas_used, - chain_tx.tx_result.log.clone(), - tx.as_mut(), - ) - .await?; - } - - Ok(()) -} - -async fn persist_messages( - txs: &[ParsedTransactionResponse], - tx: &mut StorageTransaction, -) -> Result<(), ScraperError> { - debug!("persisting messages"); - - for chain_tx in txs { - for (index, msg) in chain_tx.tx.body.messages.iter().enumerate() { - insert_message( - chain_tx.hash.to_string(), - index as i64, - msg.type_url.clone(), - chain_tx.height.into(), - tx.as_mut(), - ) - .await? - } - } - - Ok(()) -} +pub mod transaction; diff --git a/common/nyxd-scraper-sqlite/src/storage/transaction.rs b/common/nyxd-scraper-sqlite/src/storage/transaction.rs new file mode 100644 index 0000000000..713cbb3261 --- /dev/null +++ b/common/nyxd-scraper-sqlite/src/storage/transaction.rs @@ -0,0 +1,236 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::error::SqliteScraperError; +use crate::storage::manager::{ + insert_block, insert_message, insert_precommit, insert_transaction, insert_validator, +}; +use async_trait::async_trait; +use nyxd_scraper_shared::helpers::{ + validator_consensus_address, validator_info, validator_pubkey_to_bech32, +}; +use nyxd_scraper_shared::storage::validators::Response; +use nyxd_scraper_shared::storage::{ + validators, Block, Commit, CommitSig, NyxdScraperStorageError, NyxdScraperTransaction, +}; +use nyxd_scraper_shared::ParsedTransactionResponse; +use sqlx::{Sqlite, Transaction}; +use std::ops::{Deref, DerefMut}; +use tracing::{debug, trace, warn}; + +pub struct SqliteStorageTransaction(pub(crate) Transaction<'static, Sqlite>); + +impl Deref for SqliteStorageTransaction { + type Target = Transaction<'static, Sqlite>; + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +impl DerefMut for SqliteStorageTransaction { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.0 + } +} + +impl SqliteStorageTransaction { + async fn persist_validators( + &mut self, + validators: &validators::Response, + ) -> Result<(), SqliteScraperError> { + debug!("persisting {} validators", validators.total); + for validator in &validators.validators { + let consensus_address = validator_consensus_address(validator.address)?; + let consensus_pubkey = validator_pubkey_to_bech32(validator.pub_key)?; + + insert_validator( + consensus_address.to_string(), + consensus_pubkey.to_string(), + self.0.as_mut(), + ) + .await?; + } + + Ok(()) + } + + async fn persist_block_data( + &mut self, + block: &Block, + total_gas: i64, + ) -> Result<(), SqliteScraperError> { + let proposer_address = + validator_consensus_address(block.header.proposer_address)?.to_string(); + + insert_block( + block.header.height.into(), + block.header.hash().to_string(), + block.data.len() as u32, + total_gas, + proposer_address, + block.header.time.into(), + self.0.as_mut(), + ) + .await?; + Ok(()) + } + + async fn persist_commits( + &mut self, + commits: &Commit, + validators: &validators::Response, + ) -> Result<(), SqliteScraperError> { + debug!("persisting up to {} commits", commits.signatures.len()); + let height: i64 = commits.height.into(); + + for commit_sig in &commits.signatures { + let (validator_id, timestamp, signature) = match commit_sig { + CommitSig::BlockIdFlagAbsent => { + trace!("absent signature"); + continue; + } + CommitSig::BlockIdFlagCommit { + validator_address, + timestamp, + signature, + } => (validator_address, timestamp, signature), + CommitSig::BlockIdFlagNil { + validator_address, + timestamp, + signature, + } => (validator_address, timestamp, signature), + }; + + let validator = validator_info(*validator_id, validators)?; + let validator_address = validator_consensus_address(*validator_id)?; + + if signature.is_none() { + warn!("empty signature for {validator_address} at height {height}"); + continue; + } + + insert_precommit( + validator_address.to_string(), + height, + (*timestamp).into(), + validator.power.into(), + validator.proposer_priority.value(), + self.0.as_mut(), + ) + .await?; + } + + Ok(()) + } + + async fn persist_txs( + &mut self, + txs: &[ParsedTransactionResponse], + ) -> Result<(), SqliteScraperError> { + debug!("persisting {} txs", txs.len()); + + for chain_tx in txs { + insert_transaction( + chain_tx.hash.to_string(), + chain_tx.height.into(), + chain_tx.index as i64, + chain_tx.tx_result.code.is_ok(), + chain_tx.tx.body.messages.len() as i64, + chain_tx.tx.body.memo.clone(), + chain_tx.tx_result.gas_wanted, + chain_tx.tx_result.gas_used, + chain_tx.tx_result.log.clone(), + self.0.as_mut(), + ) + .await?; + } + + Ok(()) + } + + async fn persist_messages( + &mut self, + txs: &[ParsedTransactionResponse], + ) -> Result<(), SqliteScraperError> { + debug!("persisting messages"); + + for chain_tx in txs { + for (index, msg) in chain_tx.tx.body.messages.iter().enumerate() { + insert_message( + chain_tx.hash.to_string(), + index as i64, + msg.type_url.clone(), + chain_tx.height.into(), + self.0.as_mut(), + ) + .await? + } + } + + Ok(()) + } +} + +#[async_trait] +impl NyxdScraperTransaction for SqliteStorageTransaction { + async fn commit(self) -> Result<(), NyxdScraperStorageError> { + self.0 + .commit() + .await + .map_err(SqliteScraperError::from) + .map_err(NyxdScraperStorageError::from) + } + + async fn persist_validators( + &mut self, + validators: &Response, + ) -> Result<(), NyxdScraperStorageError> { + self.persist_validators(validators) + .await + .map_err(NyxdScraperStorageError::from) + } + + async fn persist_block_data( + &mut self, + block: &Block, + total_gas: i64, + ) -> Result<(), NyxdScraperStorageError> { + self.persist_block_data(block, total_gas) + .await + .map_err(NyxdScraperStorageError::from) + } + + async fn persist_commits( + &mut self, + commits: &Commit, + validators: &Response, + ) -> Result<(), NyxdScraperStorageError> { + self.persist_commits(commits, validators) + .await + .map_err(NyxdScraperStorageError::from) + } + + async fn persist_txs( + &mut self, + txs: &[ParsedTransactionResponse], + ) -> Result<(), NyxdScraperStorageError> { + self.persist_txs(txs) + .await + .map_err(NyxdScraperStorageError::from) + } + + async fn persist_messages( + &mut self, + txs: &[ParsedTransactionResponse], + ) -> Result<(), NyxdScraperStorageError> { + self.persist_messages(txs) + .await + .map_err(NyxdScraperStorageError::from) + } + + async fn update_last_processed(&mut self, height: i64) -> Result<(), NyxdScraperStorageError> { + self.update_last_processed(height) + .await + .map_err(NyxdScraperStorageError::from) + } +}