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

implemented traits for sqlite instance

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2025-07-04 12:09:45 +01:00
parent 027bd85200
commit 81eaf7b1cc
27 changed files with 561 additions and 2526 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
-12
View File
@@ -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]]
common/nyxd-scraper-shared/src/error.rs
+5 -20
View File
@@ -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<dyn std::error::Error + Send + Sync>),
#[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,
common/nyxd-scraper-shared/src/helpers.rs
+29 -9
View File
@@ -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<M: AsRef<[u8]>>(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<M: AsRef<[u8]>>(raw_tx: M) -> Hash {
Hash::Sha256(Sha256::digest(raw_tx).into())
}
pub fn validator_pubkey_to_bech32(pubkey: PublicKey) -> Result<AccountId, ScraperError> {
pub fn validator_pubkey_to_bech32(pubkey: PublicKey) -> Result<AccountId, MalformedDataError> {
// 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<AccountId, ScraperError> {
pub fn validator_consensus_address(id: account::Id) -> Result<AccountId, MalformedDataError> {
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(),
})
}
common/nyxd-scraper-shared/src/lib.rs
+1 -1
View File
@@ -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;
common/nyxd-scraper-shared/src/storage/helpers.rs
+1 -1
View File
@@ -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);
common/nyxd-scraper-shared/src/storage/mod.rs
+20 -10
View File
@@ -1,37 +1,47 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// 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<dyn std::error::Error + Send + Sync>);
impl NyxdScraperStorageError {
pub fn new<E>(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<Self, NyxdScraperStorageError>;
async fn initialise(storage_path: &Path) -> Result<Self, NyxdScraperStorageError>;
async fn begin_processing_tx(
&self,
) -> Result<Self::StorageTransaction, NyxdScraperStorageError>;
async fn get_last_processed_height(&self) -> Result<u32, NyxdScraperStorageError>;
async fn get_last_processed_height(&self) -> Result<i64, NyxdScraperStorageError>;
async fn get_pruned_height(&self) -> Result<u32, NyxdScraperStorageError>;
async fn get_pruned_height(&self) -> Result<i64, NyxdScraperStorageError>;
async fn lowest_block_height(&self) -> Result<Option<i64>, NyxdScraperStorageError>;
common/nyxd-scraper-sqlite/Cargo.toml
+2 -18
View File
@@ -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"] }
common/nyxd-scraper-sqlite/src/block_processor/helpers.rs
-65
View File
@@ -1,65 +0,0 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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<u32>) -> VecDeque<Range<u32>> {
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));
}
}
common/nyxd-scraper-sqlite/src/block_processor/mod.rs
-515
View File
@@ -1,515 +0,0 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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<u32>,
queued_requests: VecDeque<Range<u32>>,
}
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<u32>,
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<u32>,
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<Notify>,
last_processed_height: u32,
last_pruned_height: u32,
last_processed_at: Instant,
pending_sync: PendingSync,
queued_blocks: BTreeMap<u32, BlockToProcess>,
rpc_client: RpcClient,
incoming: UnboundedReceiverStream<BlockToProcess>,
block_requester: Sender<BlockRequest>,
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<Box<dyn BlockModule + Send>>,
tx_modules: Vec<Box<dyn TxModule + Send>>,
msg_modules: Vec<Box<dyn MsgModule + Send>>,
}
#[allow(clippy::too_many_arguments)]
impl BlockProcessor {
pub async fn new(
config: BlockProcessorConfig,
cancel: CancellationToken,
synced: Arc<Notify>,
incoming: UnboundedReceiver<BlockToProcess>,
block_requester: Sender<BlockRequest>,
storage: ScraperStorage,
rpc_client: RpcClient,
) -> Result<Self, ScraperError> {
let last_processed = storage.get_last_processed_height().await?;
let last_processed_height = last_processed.try_into().unwrap_or_default();
let last_pruned = storage.get_pruned_height().await?;
let last_pruned_height = last_pruned.try_into().unwrap_or_default();
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<Box<dyn BlockModule + Send>>) {
self.block_modules = modules;
}
pub fn set_tx_modules(&mut self, modules: Vec<Box<dyn TxModule + Send>>) {
self.tx_modules = modules;
}
pub fn set_msg_modules(&mut self, modules: Vec<Box<dyn MsgModule + Send>>) {
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<u32>,
) -> 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<u32>) -> 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
}
}
}
}
}
}
}
common/nyxd-scraper-sqlite/src/block_processor/pruning.rs
-122
View File
@@ -1,122 +0,0 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::error::ScraperError;
use serde::{Deserialize, Serialize};
pub const DEFAULT_PRUNING_KEEP_RECENT: u32 = 362880;
pub const DEFAULT_PRUNING_INTERVAL: u32 = 10;
pub const EVERYTHING_PRUNING_KEEP_RECENT: u32 = 2;
pub const EVERYTHING_PRUNING_INTERVAL: u32 = 10;
/// We follow cosmos-sdk pruning strategies for convenience’s sake.
#[derive(Debug, Default, Clone, Copy, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PruningStrategy {
/// 'Default' strategy defines a pruning strategy where the last 362880 heights are
/// kept where to-be pruned heights are pruned at every 10th height.
/// The last 362880 heights are kept(approximately 3.5 weeks worth of state) assuming the typical
/// block time is 6s. If these values do not match the applications' requirements, use the "custom" option.
#[default]
Default,
/// 'Everything' strategy defines a pruning strategy where all committed heights are
/// deleted, storing only the current height and last 2 states. To-be pruned heights are
/// pruned at every 10th height.
Everything,
/// 'Nothing' strategy defines a pruning strategy where all heights are kept on disk.
Nothing,
/// 'Custom' strategy defines a pruning strategy where the user specifies the pruning.
Custom,
}
impl PruningStrategy {
pub fn is_custom(&self) -> bool {
matches!(self, PruningStrategy::Custom)
}
pub fn is_nothing(&self) -> bool {
matches!(self, PruningStrategy::Nothing)
}
pub fn is_everything(&self) -> bool {
matches!(self, PruningStrategy::Everything)
}
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
pub struct PruningOptions {
/// keep_recent defines how many recent heights to keep on disk.
pub keep_recent: u32,
/// interval defines the frequency of removing the pruned heights from the disk.
pub interval: u32,
/// strategy defines the currently used kind of [PruningStrategy].
pub strategy: PruningStrategy,
}
impl PruningOptions {
pub fn validate(&self) -> Result<(), ScraperError> {
// if strategy is not set to custom, other options are meaningless since they won't be applied
if !self.strategy.is_custom() {
return Ok(());
}
if self.interval == 0 {
return Err(ScraperError::ZeroPruningInterval);
}
if self.interval < EVERYTHING_PRUNING_INTERVAL {
return Err(ScraperError::TooSmallPruningInterval {
interval: self.interval,
});
}
if self.keep_recent < EVERYTHING_PRUNING_KEEP_RECENT {
return Err(ScraperError::TooSmallKeepRecent {
keep_recent: self.keep_recent,
});
}
Ok(())
}
pub fn nothing() -> Self {
PruningOptions {
keep_recent: 0,
interval: 0,
strategy: PruningStrategy::Nothing,
}
}
pub fn strategy_interval(&self) -> u32 {
match self.strategy {
PruningStrategy::Default => DEFAULT_PRUNING_INTERVAL,
PruningStrategy::Everything => EVERYTHING_PRUNING_INTERVAL,
PruningStrategy::Nothing => 0,
PruningStrategy::Custom => self.interval,
}
}
pub fn strategy_keep_recent(&self) -> u32 {
match self.strategy {
PruningStrategy::Default => DEFAULT_PRUNING_KEEP_RECENT,
PruningStrategy::Everything => EVERYTHING_PRUNING_KEEP_RECENT,
PruningStrategy::Nothing => 0,
PruningStrategy::Custom => self.keep_recent,
}
}
}
impl Default for PruningOptions {
fn default() -> Self {
PruningOptions {
keep_recent: DEFAULT_PRUNING_KEEP_RECENT,
interval: DEFAULT_PRUNING_INTERVAL,
strategy: Default::default(),
}
}
}
common/nyxd-scraper-sqlite/src/block_processor/types.rs
-115
View File
@@ -1,115 +0,0 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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<tx::Proof>,
}
#[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<ParsedTransactionResponse>,
}
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<Block> for BlockToProcess {
fn from(block: Block) -> Self {
BlockToProcess {
height: block.header.height.value() as u32,
block,
}
}
}
impl TryFrom<Event> for BlockToProcess {
type Error = ScraperError;
fn try_from(event: Event) -> Result<Self, Self::Error> {
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<block_endpoint::Response> for BlockToProcess {
fn from(value: block_endpoint::Response) -> Self {
value.block.into()
}
}
common/nyxd-scraper-sqlite/src/block_requester/mod.rs
-91
View File
@@ -1,91 +0,0 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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<u32>),
// UNIMPLEMENTED:
#[allow(dead_code)]
Specific(Vec<u32>),
}
pub(crate) struct BlockRequester {
cancel: CancellationToken,
rpc_client: RpcClient,
requests: ReceiverStream<BlockRequest>,
blocks: UnboundedSender<BlockToProcess>,
}
impl BlockRequester {
pub(crate) fn new(
cancel: CancellationToken,
rpc_client: RpcClient,
requests: Receiver<BlockRequest>,
blocks: UnboundedSender<BlockToProcess>,
) -> 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<I: IntoIterator<Item = u32>>(&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
}
}
}
}
}
}
}
common/nyxd-scraper-sqlite/src/constants.rs
-47
View File
@@ -1,47 +0,0 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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
);
common/nyxd-scraper-sqlite/src/error.rs
+13 -132
View File
@@ -1,155 +1,36 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::block_processor::pruning::{
EVERYTHING_PRUNING_INTERVAL, EVERYTHING_PRUNING_KEEP_RECENT,
};
use tendermint::Hash;
use 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<tendermint_rpc::Error>,
},
#[error("failed to establish rpc connection to {url}: {source}")]
HttpConnectionFailure {
url: String,
#[source]
source: Box<tendermint_rpc::Error>,
},
#[error("failed to create chain subscription: {source}")]
ChainSubscriptionFailure {
#[source]
source: Box<tendermint_rpc::Error>,
},
#[error("could not obtain basic block information at height: {height}: {source}")]
BlockQueryFailure {
height: u32,
#[source]
source: Box<tendermint_rpc::Error>,
},
#[error("could not obtain block results information at height: {height}: {source}")]
BlockResultsQueryFailure {
height: u32,
#[source]
source: Box<tendermint_rpc::Error>,
},
#[error("could not obtain validators information at height: {height}: {source}")]
ValidatorsQueryFailure {
height: u32,
#[source]
source: Box<tendermint_rpc::Error>,
},
#[error("could not obtain tx results for tx: {hash}: {source}")]
TxResultsQueryFailure {
hash: Hash,
#[source]
source: Box<tendermint_rpc::Error>,
},
#[error("could not obtain current abci info: {source}")]
AbciInfoQueryFailure {
#[source]
source: Box<tendermint_rpc::Error>,
},
#[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<T> From<SendError<T>> for ScraperError {
fn from(_: SendError<T>) -> Self {
ScraperError::ClosedChannelError
impl From<SqliteScraperError> for NyxdScraperStorageError {
fn from(err: SqliteScraperError) -> Self {
NyxdScraperStorageError::new(err)
}
}
common/nyxd-scraper-sqlite/src/helpers.rs
-46
View File
@@ -1,46 +0,0 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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<M: AsRef<[u8]>>(raw_tx: M) -> Hash {
Hash::Sha256(Sha256::digest(raw_tx).into())
}
pub(crate) fn validator_pubkey_to_bech32(pubkey: PublicKey) -> Result<AccountId, ScraperError> {
// 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, ScraperError> {
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(),
})
}
}
}
common/nyxd-scraper-sqlite/src/lib.rs
-11
View File
@@ -1,18 +1,7 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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;
common/nyxd-scraper-sqlite/src/modules/block_module.rs
-16
View File
@@ -1,16 +0,0 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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>;
}
common/nyxd-scraper-sqlite/src/modules/mod.rs
-10
View File
@@ -1,10 +0,0 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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;
common/nyxd-scraper-sqlite/src/modules/msg_module.rs
-21
View File
@@ -1,21 +0,0 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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>;
}
common/nyxd-scraper-sqlite/src/modules/tx_module.rs
-16
View File
@@ -1,16 +0,0 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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>;
}
common/nyxd-scraper-sqlite/src/rpc_client.rs
-198
View File
@@ -1,198 +0,0 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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<HttpClient>,
}
impl RpcClient {
pub fn new(url: &Url) -> Result<Self, ScraperError> {
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<FullBlockInformation, ScraperError> {
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<block::Response, ScraperError> {
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<block_results::Response, ScraperError> {
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<u64, ScraperError> {
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<u64, ScraperError> {
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<u8>],
) -> Result<Vec<tx::Response>, 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<tx::Response, ScraperError> {
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<validators::Response, ScraperError> {
debug!("getting validators set");
self.inner
.validators(height, Paging::All)
.await
.map_err(|source| ScraperError::ValidatorsQueryFailure {
height,
source: Box::new(source),
})
}
}
common/nyxd-scraper-sqlite/src/scraper/mod.rs
-407
View File
@@ -1,407 +0,0 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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<u32>,
/// 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<Box<dyn BlockModule + Send>>,
tx_modules: Vec<Box<dyn TxModule + Send>>,
msg_modules: Vec<Box<dyn MsgModule + Send>>,
}
impl NyxdScraperBuilder {
pub async fn build_and_start(self) -> Result<NyxdScraper, ScraperError> {
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<M: BlockModule + Send + 'static>(mut self, module: M) -> Self {
self.block_modules.push(Box::new(module));
self
}
pub fn with_tx_module<M: TxModule + Send + 'static>(mut self, module: M) -> Self {
self.tx_modules.push(Box::new(module));
self
}
pub fn with_msg_module<M: MsgModule + Send + 'static>(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<Notify>,
storage: ScraperStorage,
rpc_client: RpcClient,
}
impl NyxdScraper {
pub fn builder(config: Config) -> NyxdScraperBuilder {
NyxdScraperBuilder::new(config)
}
pub async fn new(config: Config) -> Result<Self, ScraperError> {
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<u32>,
end_height: Option<u32>,
) -> 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::<Vec<_>>();
// 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<BlockRequest>,
processing_tx: UnboundedSender<BlockToProcess>,
) -> 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<BlockRequest>,
processing_rx: UnboundedReceiver<BlockToProcess>,
) -> Result<BlockProcessor, ScraperError> {
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<BlockToProcess>,
) -> Result<ChainSubscriber, ScraperError> {
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()
}
}
common/nyxd-scraper-sqlite/src/scraper/subscriber.rs
-253
View File
@@ -1,253 +0,0 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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<BlockToProcess>,
websocket_endpoint: Url,
websocket_client: WebSocketClient,
websocket_driver: Option<WebSocketClientDriver>,
}
impl ChainSubscriber {
pub async fn new(
websocket_endpoint: &Url,
cancel: CancellationToken,
task_tracker: TaskTracker,
block_sender: UnboundedSender<BlockToProcess>,
) -> Result<Self, ScraperError> {
// 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<bool, ScraperError> {
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}");
}
}
}
}
}
common/nyxd-scraper-sqlite/src/storage/block_storage.rs
+251
View File
@@ -0,0 +1,251 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// 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<P: AsRef<Path> + Debug>(
database_path: P,
) -> Result<Self, SqliteScraperError> {
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<SqliteStorageTransaction, SqliteScraperError> {
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<Option<i64>, SqliteScraperError> {
Ok(self.manager.get_lowest_block().await?)
}
pub async fn get_first_block_height_after(
&self,
time: OffsetDateTime,
) -> Result<Option<i64>, SqliteScraperError> {
Ok(self.manager.get_first_block_height_after(time).await?)
}
pub async fn get_last_block_height_before(
&self,
time: OffsetDateTime,
) -> Result<Option<i64>, 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<i64, SqliteScraperError> {
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<i64, SqliteScraperError> {
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<i64, SqliteScraperError> {
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<Option<CommitSignature>, SqliteScraperError> {
Ok(self
.manager
.get_precommit(consensus_address, height)
.await?)
}
pub async fn get_block_signers(
&self,
height: i64,
) -> Result<Vec<Validator>, SqliteScraperError> {
Ok(self.manager.get_block_validators(height).await?)
}
pub async fn get_all_known_validators(&self) -> Result<Vec<Validator>, SqliteScraperError> {
Ok(self.manager.get_validators().await?)
}
pub async fn get_last_processed_height(&self) -> Result<i64, SqliteScraperError> {
Ok(self.manager.get_last_processed_height().await?)
}
pub async fn get_pruned_height(&self) -> Result<i64, SqliteScraperError> {
Ok(self.manager.get_pruned_height().await?)
}
}
#[async_trait]
impl NyxdScraperStorage for SqliteScraperStorage {
type StorageTransaction = SqliteStorageTransaction;
async fn initialise(storage_path: &Path) -> Result<Self, NyxdScraperStorageError> {
SqliteScraperStorage::init(&storage_path)
.await
.map_err(NyxdScraperStorageError::from)
}
async fn begin_processing_tx(
&self,
) -> Result<Self::StorageTransaction, NyxdScraperStorageError> {
self.begin_processing_tx()
.await
.map_err(NyxdScraperStorageError::from)
}
async fn get_last_processed_height(&self) -> Result<i64, NyxdScraperStorageError> {
self.get_last_processed_height()
.await
.map_err(NyxdScraperStorageError::from)
}
async fn get_pruned_height(&self) -> Result<i64, NyxdScraperStorageError> {
self.get_pruned_height()
.await
.map_err(NyxdScraperStorageError::from)
}
async fn lowest_block_height(&self) -> Result<Option<i64>, 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)
}
}
common/nyxd-scraper-sqlite/src/storage/helpers.rs
-2
View File
@@ -1,2 +0,0 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
common/nyxd-scraper-sqlite/src/storage/mod.rs
+3 -388
View File
@@ -1,394 +1,9 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// 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<P: AsRef<Path> + Debug>(database_path: P) -> Result<Self, ScraperError> {
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<StorageTransaction, ScraperError> {
debug!("starting storage tx");
self.manager
.connection_pool
.begin()
.await
.map_err(|source| ScraperError::StorageTxBeginFailure { source })
}
pub async fn lowest_block_height(&self) -> Result<Option<i64>, ScraperError> {
Ok(self.manager.get_lowest_block().await?)
}
pub async fn get_first_block_height_after(
&self,
time: OffsetDateTime,
) -> Result<Option<i64>, ScraperError> {
Ok(self.manager.get_first_block_height_after(time).await?)
}
pub async fn get_last_block_height_before(
&self,
time: OffsetDateTime,
) -> Result<Option<i64>, 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<i64, ScraperError> {
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<i64, ScraperError> {
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<i64, ScraperError> {
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<Option<CommitSignature>, ScraperError> {
Ok(self
.manager
.get_precommit(consensus_address, height)
.await?)
}
pub async fn get_block_signers(&self, height: i64) -> Result<Vec<Validator>, ScraperError> {
Ok(self.manager.get_block_validators(height).await?)
}
pub async fn get_all_known_validators(&self) -> Result<Vec<Validator>, ScraperError> {
Ok(self.manager.get_validators().await?)
}
pub async fn get_last_processed_height(&self) -> Result<i64, ScraperError> {
Ok(self.manager.get_last_processed_height().await?)
}
pub async fn get_pruned_height(&self) -> Result<i64, ScraperError> {
Ok(self.manager.get_pruned_height().await?)
}
}
pub async fn persist_block(
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;
common/nyxd-scraper-sqlite/src/storage/transaction.rs
+236
View File
@@ -0,0 +1,236 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// 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)
}
}