diff --git a/common/nyxd-scraper-psql/Cargo.toml b/common/nyxd-scraper-psql/Cargo.toml new file mode 100644 index 0000000000..8266c23262 --- /dev/null +++ b/common/nyxd-scraper-psql/Cargo.toml @@ -0,0 +1,43 @@ +[package] +name = "nyxd-scraper" +version = "0.1.0" +authors.workspace = true +repository.workspace = true +homepage.workspace = true +documentation.workspace = true +edition.workspace = true +license.workspace = true + +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html + +[dependencies] +async-trait.workspace = true +base64.workspace = true +const_format = { workspace = true } +cosmrs.workspace = true +chrono = {workspace = true} +eyre = { workspace = true } +futures.workspace = true +humantime = { workspace = true } +sha2 = { workspace = true } +serde = { workspace = true, features = ["derive"] } +serde_json = {workspace = true} +sqlx = { workspace = true, features = ["runtime-tokio-rustls", "postgres", "macros", "migrate", "time", "json"] } +tendermint.workspace = true +tendermint-rpc = { workspace = true, features = ["websocket-client", "http-client"] } +thiserror.workspace = true +time = { workspace = true } +tokio = { workspace = true, features = ["full"] } +tokio-stream = { workspace = true } +tokio-util = { workspace = true, features = ["rt"] } +tracing.workspace = true +url.workspace = true + + +# TEMP +#nym-bin-common = { path = "../bin-common", features = ["basic_tracing"]} + + +[build-dependencies] +sqlx = { workspace = true, features = ["runtime-tokio-rustls", "postgres", "macros", "migrate"] } +tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } diff --git a/common/nyxd-scraper-psql/README.md b/common/nyxd-scraper-psql/README.md new file mode 100644 index 0000000000..2e22d292c1 --- /dev/null +++ b/common/nyxd-scraper-psql/README.md @@ -0,0 +1,23 @@ +# Nyxd Scraper + +## Pruning + +Similarly to cosmos-sdk, we incorporate pruning into our (scraped) chain data. We attempt to follow their strategies as +closely as possible for convenience's sake. Therefore, the following are available: + +### Strategies + +The strategies are configured in `config.toml`, with the format `pruning = ""` where the options are: + +* `default`: only the last 362,880 states(approximately 3.5 weeks worth of state) are kept; pruning at 10 block + intervals +* `nothing`: all historic states will be saved, nothing will be deleted (i.e. archiving node) +* `everything`: 2 latest states will be kept; pruning at 10 block intervals. +* `custom`: allow pruning options to be manually specified through `pruning.keep_recent`, and `pruning.interval` + +### Custom Pruning + +These are applied if and only if the pruning strategy is `custom`: + +* `pruning.keep_recent`: N means to keep all of the last N blocks +* `pruning.interval`: N means to delete old block data from disk every Nth block. diff --git a/common/nyxd-scraper-psql/build.rs b/common/nyxd-scraper-psql/build.rs new file mode 100644 index 0000000000..f38187f043 --- /dev/null +++ b/common/nyxd-scraper-psql/build.rs @@ -0,0 +1,41 @@ +#[tokio::main] +async fn main() { + use sqlx::{Connection, Executor, PgConnection}; + + const POSTGRES_USER: &str = "nym"; + const POSTGRES_PASSWORD: &str = "password123"; + const POSTGRES_DB: &str = "nyxd_scraper"; + + let admin_url = format!( + "postgres://{}:{}@localhost:5432/postgres", + POSTGRES_USER, POSTGRES_PASSWORD + ); + // Connect to postgres to create test database + let database_url = + format!("postgres://{POSTGRES_USER}:{POSTGRES_PASSWORD}@localhost:5432/{POSTGRES_DB}"); + + let mut conn = PgConnection::connect(&admin_url) + .await + .expect("Failed to connect to Postgres"); + + conn.execute(format!(r#"DROP DATABASE IF EXISTS {}"#, POSTGRES_DB).as_str()) + .await + .expect("Failed to drop test database"); + + conn.execute(format!(r#"CREATE DATABASE {}"#, POSTGRES_DB).as_str()) + .await + .expect("Failed to create test database"); + + let mut test_conn = PgConnection::connect(&database_url) + .await + .expect("Failed to connect to test database"); + + // Run migrations + sqlx::migrate!("./sql_migrations") + .run(&mut test_conn) + .await + .expect("Failed to perform SQLx migrations"); + + // Set the database URL as an environment variable + println!("cargo:rustc-env=DATABASE_URL={}", database_url); +} diff --git a/common/nyxd-scraper-psql/callisto b/common/nyxd-scraper-psql/callisto new file mode 160000 index 0000000000..868bbd5cbb --- /dev/null +++ b/common/nyxd-scraper-psql/callisto @@ -0,0 +1 @@ +Subproject commit 868bbd5cbb8804f25102b9042b8974c7dc71ce52 diff --git a/common/nyxd-scraper-psql/sql_migrations/01_metadata.sql b/common/nyxd-scraper-psql/sql_migrations/01_metadata.sql new file mode 100644 index 0000000000..1a525315a0 --- /dev/null +++ b/common/nyxd-scraper-psql/sql_migrations/01_metadata.sql @@ -0,0 +1,8 @@ +/* + * Copyright 2023 - Nym Technologies SA + * SPDX-License-Identifier: Apache-2.0 + */ +CREATE TABLE METADATA ( + id INTEGER PRIMARY KEY CHECK (id = 0), + last_processed_height BIGINT NOT NULL +); \ No newline at end of file diff --git a/common/nyxd-scraper-psql/sql_migrations/02_cosmos.sql b/common/nyxd-scraper-psql/sql_migrations/02_cosmos.sql new file mode 100644 index 0000000000..3762cf8804 --- /dev/null +++ b/common/nyxd-scraper-psql/sql_migrations/02_cosmos.sql @@ -0,0 +1,73 @@ +CREATE TABLE validator ( + consensus_address TEXT NOT NULL PRIMARY KEY, + /* Validator consensus address */ + consensus_pubkey TEXT NOT NULL UNIQUE + /* Validator consensus public key */ +); + +CREATE TABLE pre_commit ( + validator_address TEXT NOT NULL REFERENCES validator (consensus_address), + height BIGINT NOT NULL, + timestamp TIMESTAMPTZ NOT NULL, + voting_power BIGINT NOT NULL, + proposer_priority BIGINT NOT NULL, + UNIQUE (validator_address, timestamp) +); + +CREATE INDEX pre_commit_validator_address_index ON pre_commit (validator_address); + +CREATE INDEX pre_commit_height_index ON pre_commit (height); + +CREATE TABLE block ( + height BIGINT UNIQUE PRIMARY KEY, + hash TEXT NOT NULL UNIQUE, + num_txs INTEGER DEFAULT 0, + total_gas BIGINT DEFAULT 0, + proposer_address TEXT REFERENCES validator (consensus_address), + timestamp TIMESTAMPTZ NOT NULL +); + +CREATE INDEX block_height_index ON block (height); + +CREATE INDEX block_hash_index ON block (hash); + +CREATE INDEX block_proposer_address_index ON block (proposer_address); + +CREATE TABLE "transaction" ( + hash TEXT UNIQUE NOT NULL, + height BIGINT NOT NULL REFERENCES block (height), + "index" INTEGER NOT NULL, + success BOOLEAN NOT NULL, + /* Body */ + num_messages INTEGER NOT NULL, + messages JSONB NOT NULL DEFAULT '[]', + memo TEXT, + signatures TEXT [] NOT NULL, + /* AuthInfo */ + signer_infos JSONB NOT NULL DEFAULT '[]'::JSONB, + fee JSONB NOT NULL DEFAULT '{}'::JSONB, + /* Tx response */ + gas_wanted BIGINT DEFAULT 0, + gas_used BIGINT DEFAULT 0, + raw_log TEXT +); + +CREATE INDEX transaction_hash_index ON "transaction" (hash); + +CREATE INDEX transaction_height_index ON "transaction" (height); + +CREATE TABLE message ( + transaction_hash TEXT NOT NULL REFERENCES "transaction" (hash), + "index" BIGINT NOT NULL, + TYPE TEXT NOT NULL, + value JSONB NOT NULL, + involved_accounts_addresses TEXT [] NOT NULL, + height BIGINT NOT NULL, + CONSTRAINT unique_message_per_tx UNIQUE (transaction_hash, "index") +); + +CREATE INDEX message_transaction_hash_index ON message (transaction_hash); + +CREATE INDEX message_type_index ON message (TYPE); + +CREATE TABLE pruning (last_pruned_height BIGINT NOT NULL); \ No newline at end of file diff --git a/common/nyxd-scraper-psql/src/block_processor/helpers.rs b/common/nyxd-scraper-psql/src/block_processor/helpers.rs new file mode 100644 index 0000000000..ffc8c592c5 --- /dev/null +++ b/common/nyxd-scraper-psql/src/block_processor/helpers.rs @@ -0,0 +1,65 @@ +// 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-psql/src/block_processor/mod.rs b/common/nyxd-scraper-psql/src/block_processor/mod.rs new file mode 100644 index 0000000000..acb0cbc375 --- /dev/null +++ b/common/nyxd-scraper-psql/src/block_processor/mod.rs @@ -0,0 +1,426 @@ +// 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() + } +} + +pub struct BlockProcessor { + pruning_options: PruningOptions, + 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>, +} + +impl BlockProcessor { + pub async fn new( + pruning_options: PruningOptions, + 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(); + + Ok(BlockProcessor { + pruning_options, + 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.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).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 { + 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.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.pruning_options.strategy.is_nothing() { + trace!("the current pruning strategy is 'nothing'"); + return Ok(()); + } + + let interval = self.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()); + + self.maybe_prune_storage().await?; + + let latest_block = self.rpc_client.current_block_height().await? as u32; + + if latest_block > self.last_processed_height && self.last_processed_height != 0 { + // 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.pruning_options.strategy_keep_recent(); + let last_to_keep = latest_block - keep_recent; + self.last_processed_height = max(self.last_processed_height, last_to_keep); + + let request_range = self.last_processed_height + 1..latest_block + 1; + info!("we need to request {request_range:?} to resync"); + self.request_missing_blocks(request_range).await?; + } + + Ok(()) + } + + pub(crate) async fn run(&mut self) { + info!("starting 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-psql/src/block_processor/pruning.rs b/common/nyxd-scraper-psql/src/block_processor/pruning.rs new file mode 100644 index 0000000000..e7a99e7dfa --- /dev/null +++ b/common/nyxd-scraper-psql/src/block_processor/pruning.rs @@ -0,0 +1,122 @@ +// 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-psql/src/block_processor/types.rs b/common/nyxd-scraper-psql/src/block_processor/types.rs new file mode 100644 index 0000000000..93ffb3ba76 --- /dev/null +++ b/common/nyxd-scraper-psql/src/block_processor/types.rs @@ -0,0 +1,121 @@ +// 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 { + // we don't care about `NewBlock` until CometBFT 0.38, i.e. until we upgrade to wasmd 0.50 + EventData::NewBlock { .. } => { + return Err(ScraperError::InvalidSubscriptionEvent { + query, + kind: "NewBlock".to_string(), + }) + } + 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-psql/src/block_requester/mod.rs b/common/nyxd-scraper-psql/src/block_requester/mod.rs new file mode 100644 index 0000000000..b2a1a4d43a --- /dev/null +++ b/common/nyxd-scraper-psql/src/block_requester/mod.rs @@ -0,0 +1,91 @@ +// 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-psql/src/constants.rs b/common/nyxd-scraper-psql/src/constants.rs new file mode 100644 index 0000000000..9401cd360e --- /dev/null +++ b/common/nyxd-scraper-psql/src/constants.rs @@ -0,0 +1,47 @@ +// 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-psql/src/error.rs b/common/nyxd-scraper-psql/src/error.rs new file mode 100644 index 0000000000..9876536aa5 --- /dev/null +++ b/common/nyxd-scraper-psql/src/error.rs @@ -0,0 +1,143 @@ +// Copyright 2023 - 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 thiserror::Error; +use tokio::sync::mpsc::error::SendError; + +#[derive(Debug, Error)] +pub enum ScraperError { + #[error("experienced internal database error: {0}")] + InternalDatabaseError(#[from] sqlx::Error), + + #[error("failed to perform startup SQL migration: {0}")] + StartupMigrationFailure(#[from] sqlx::migrate::MigrateError), + + #[error("the block scraper is already running")] + ScraperAlreadyRunning, + + #[error("failed to establish websocket connection to {url}: {source}")] + WebSocketConnectionFailure { + url: String, + #[source] + source: tendermint_rpc::Error, + }, + + #[error("failed to establish rpc connection to {url}: {source}")] + HttpConnectionFailure { + url: String, + #[source] + source: tendermint_rpc::Error, + }, + + #[error("failed to create chain subscription: {source}")] + ChainSubscriptionFailure { + #[source] + source: tendermint_rpc::Error, + }, + + #[error("could not obtain basic block information at height: {height}: {source}")] + BlockQueryFailure { + height: u32, + #[source] + source: tendermint_rpc::Error, + }, + + #[error("could not obtain block results information at height: {height}: {source}")] + BlockResultsQueryFailure { + height: u32, + #[source] + source: tendermint_rpc::Error, + }, + + #[error("could not obtain validators information at height: {height}: {source}")] + ValidatorsQueryFailure { + height: u32, + #[source] + source: tendermint_rpc::Error, + }, + + #[error("could not obtain tx results for tx: {hash}: {source}")] + TxResultsQueryFailure { + hash: Hash, + #[source] + source: tendermint_rpc::Error, + }, + + #[error("could not obtain current abci info: {source}")] + AbciInfoQueryFailure { + #[source] + source: tendermint_rpc::Error, + }, + + #[error("could not parse tx {hash}: {source}")] + TxParseFailure { + hash: Hash, + #[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, + }, + + #[error("failed to commit storage tx: {source}")] + StorageTxCommitFailure { + #[source] + source: sqlx::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 }, +} + +impl From> for ScraperError { + fn from(_: SendError) -> Self { + ScraperError::ClosedChannelError + } +} diff --git a/common/nyxd-scraper-psql/src/helpers.rs b/common/nyxd-scraper-psql/src/helpers.rs new file mode 100644 index 0000000000..d11b25a2f3 --- /dev/null +++ b/common/nyxd-scraper-psql/src/helpers.rs @@ -0,0 +1,46 @@ +// 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-psql/src/lib.rs b/common/nyxd-scraper-psql/src/lib.rs new file mode 100644 index 0000000000..332e38d214 --- /dev/null +++ b/common/nyxd-scraper-psql/src/lib.rs @@ -0,0 +1,20 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +#![warn(clippy::expect_used)] +#![warn(clippy::unwrap_used)] + +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 modules::{BlockModule, MsgModule, TxModule}; +pub use scraper::{Config, NyxdScraper}; +pub use storage::models; diff --git a/common/nyxd-scraper-psql/src/modules/block_module.rs b/common/nyxd-scraper-psql/src/modules/block_module.rs new file mode 100644 index 0000000000..9ca1ba9b20 --- /dev/null +++ b/common/nyxd-scraper-psql/src/modules/block_module.rs @@ -0,0 +1,16 @@ +// 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-psql/src/modules/mod.rs b/common/nyxd-scraper-psql/src/modules/mod.rs new file mode 100644 index 0000000000..1f37d149e0 --- /dev/null +++ b/common/nyxd-scraper-psql/src/modules/mod.rs @@ -0,0 +1,10 @@ +// 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-psql/src/modules/msg_module.rs b/common/nyxd-scraper-psql/src/modules/msg_module.rs new file mode 100644 index 0000000000..df64761f7d --- /dev/null +++ b/common/nyxd-scraper-psql/src/modules/msg_module.rs @@ -0,0 +1,19 @@ +// 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 { + async fn handle_msg( + &mut self, + index: usize, + msg: &Any, + tx: &ParsedTransactionResponse, + storage_tx: &mut StorageTransaction, + ) -> Result<(), ScraperError>; +} diff --git a/common/nyxd-scraper-psql/src/modules/tx_module.rs b/common/nyxd-scraper-psql/src/modules/tx_module.rs new file mode 100644 index 0000000000..07d012ab5a --- /dev/null +++ b/common/nyxd-scraper-psql/src/modules/tx_module.rs @@ -0,0 +1,16 @@ +// 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-psql/src/rpc_client.rs b/common/nyxd-scraper-psql/src/rpc_client.rs new file mode 100644 index 0000000000..d2c141f711 --- /dev/null +++ b/common/nyxd-scraper-psql/src/rpc_client.rs @@ -0,0 +1,175 @@ +// 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, + } + })?; + + 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 }) + } + + #[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 }) + } + + 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 })?; + Ok(info.last_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, + }) + } + + #[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 }) + } +} diff --git a/common/nyxd-scraper-psql/src/scraper/mod.rs b/common/nyxd-scraper-psql/src/scraper/mod.rs new file mode 100644 index 0000000000..e95bdd5250 --- /dev/null +++ b/common/nyxd-scraper-psql/src/scraper/mod.rs @@ -0,0 +1,339 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::block_processor::types::BlockToProcess; +use crate::block_processor::BlockProcessor; +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; + +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 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 mut block_processor = BlockProcessor::new( + scraper.config.pruning_options, + 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, + pub 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.to_str().unwrap_or_default()).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, + }) + } + + fn start_tasks( + &self, + mut block_requester: BlockRequester, + mut block_processor: BlockProcessor, + mut chain_subscriber: ChainSubscriber, + ) { + self.task_tracker + .spawn(async move { block_requester.run().await }); + self.task_tracker + .spawn(async move { block_processor.run().await }); + self.task_tracker + .spawn(async move { chain_subscriber.run().await }); + + self.task_tracker.close(); + } + + pub async fn 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 + } + + pub async fn 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 current_height = self.rpc_client.current_block_height().await? as u32; + let last_processed = block_processor.last_process_height(); + + let 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 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 + } + } + }; + + 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); + } + } + } + } + + 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 { + BlockProcessor::new( + self.config.pruning_options, + 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-psql/src/scraper/subscriber.rs b/common/nyxd-scraper-psql/src/scraper/subscriber.rs new file mode 100644 index 0000000000..c2b5bdbbfd --- /dev/null +++ b/common/nyxd-scraper-psql/src/scraper/subscriber.rs @@ -0,0 +1,241 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::block_processor::types::BlockToProcess; +use crate::error::ScraperError; +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 (client, driver) = WebSocketClient::new(websocket_endpoint.as_str()) + .await + .map_err(|source| ScraperError::WebSocketConnectionFailure { + url: websocket_endpoint.to_string(), + 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, + })?; + 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 })?; + + 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-psql/src/storage/helpers.rs b/common/nyxd-scraper-psql/src/storage/helpers.rs new file mode 100644 index 0000000000..38dc36ebc3 --- /dev/null +++ b/common/nyxd-scraper-psql/src/storage/helpers.rs @@ -0,0 +1,2 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 diff --git a/common/nyxd-scraper-psql/src/storage/manager.rs b/common/nyxd-scraper-psql/src/storage/manager.rs new file mode 100644 index 0000000000..1fb9ae4d06 --- /dev/null +++ b/common/nyxd-scraper-psql/src/storage/manager.rs @@ -0,0 +1,600 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use std::vec; + +use crate::storage::log_db_operation_time; +use crate::storage::models::{CommitSignature, Validator}; +use base64; +use serde_json::Value as JsonValue; +use sqlx::types::time::OffsetDateTime; +use sqlx::{Executor, Postgres}; +use tokio::time::Instant; +use tracing::{instrument, trace}; + +#[derive(Clone)] +pub(crate) struct StorageManager { + pub(crate) connection_pool: sqlx::PgPool, +} + +impl StorageManager { + pub(crate) async fn set_initial_metadata(&self) -> Result<(), sqlx::Error> { + if sqlx::query::("SELECT * from metadata") + .fetch_optional(&self.connection_pool) + .await? + .is_none() + { + sqlx::query("INSERT INTO metadata (id, last_processed_height) VALUES (0, 0)") + .execute(&self.connection_pool) + .await?; + } + Ok(()) + } + + pub(crate) async fn get_lowest_block(&self) -> Result, sqlx::Error> { + trace!("get_lowest_block"); + let start = Instant::now(); + + let maybe_record = sqlx::query!( + r#" + SELECT height + FROM block + ORDER BY height ASC + LIMIT 1 + "#, + ) + .fetch_optional(&self.connection_pool) + .await?; + log_db_operation_time("get_lowest_block", start); + + if let Some(row) = maybe_record { + Ok(Some(row.height)) + } else { + Ok(None) + } + } + + pub(crate) async fn get_first_block_height_after( + &self, + time: OffsetDateTime, + ) -> Result, sqlx::Error> { + trace!("get_first_block_height_after"); + let start = Instant::now(); + + let maybe_record = sqlx::query!( + r#" + SELECT height + FROM block + WHERE timestamp > $1 + ORDER BY timestamp + LIMIT 1 + "#, + time + ) + .fetch_optional(&self.connection_pool) + .await?; + log_db_operation_time("get_first_block_height_after", start); + + if let Some(row) = maybe_record { + Ok(Some(row.height)) + } else { + Ok(None) + } + } + + pub(crate) async fn get_last_block_height_before( + &self, + time: OffsetDateTime, + ) -> Result, sqlx::Error> { + trace!("get_last_block_height_before"); + let start = Instant::now(); + + let maybe_record = sqlx::query!( + r#" + SELECT height + FROM block + WHERE timestamp < $1 + ORDER BY timestamp DESC + LIMIT 1 + "#, + time + ) + .fetch_optional(&self.connection_pool) + .await?; + log_db_operation_time("get_last_block_height_before", start); + + if let Some(row) = maybe_record { + Ok(Some(row.height)) + } else { + Ok(None) + } + } + + pub(crate) async fn get_signed_between( + &self, + consensus_address: &str, + start_height: i64, + end_height: i64, + ) -> Result { + trace!("get_signed_between"); + let start = Instant::now(); + + let count = sqlx::query!( + r#" + SELECT COUNT(*) as count FROM pre_commit + WHERE + validator_address = $1 + AND height >= $2 + AND height <= $3 + "#, + consensus_address, + start_height, + end_height + ) + .fetch_one(&self.connection_pool) + .await? + .count; + log_db_operation_time("get_signed_between", start); + + Ok(count.expect("Could not find the count")) + } + + pub(crate) async fn get_precommit( + &self, + consensus_address: &str, + height: i64, + ) -> Result, sqlx::Error> { + trace!("get_precommit"); + let start = Instant::now(); + + let res = sqlx::query_as( + r#" + SELECT * FROM pre_commit + WHERE validator_address = $1 + AND height = $2 + "#, + ) + .bind(consensus_address) + .bind(height) + .fetch_optional(&self.connection_pool) + .await?; + log_db_operation_time("get_precommit", start); + + Ok(res) + } + + pub(crate) async fn get_block_validators( + &self, + height: i64, + ) -> Result, sqlx::Error> { + trace!("get_block_validators"); + let start = Instant::now(); + + let res = sqlx::query_as!( + Validator, + r#" + SELECT * FROM validator + WHERE EXISTS ( + SELECT 1 FROM pre_commit + WHERE height = $1 + AND pre_commit.validator_address = validator.consensus_address + ) + "#, + height + ) + .fetch_all(&self.connection_pool) + .await?; + log_db_operation_time("get_block_validators", start); + + Ok(res) + } + + pub(crate) async fn get_validators(&self) -> Result, sqlx::Error> { + trace!("get_validators"); + let start = Instant::now(); + + let res = sqlx::query_as("SELECT * FROM validator") + .fetch_all(&self.connection_pool) + .await?; + log_db_operation_time("get_validators", start); + + Ok(res) + } + + pub(crate) async fn get_last_processed_height(&self) -> Result { + trace!("get_last_processed_height"); + let start = Instant::now(); + + let maybe_record = sqlx::query!( + r#" + SELECT last_processed_height FROM metadata + "# + ) + .fetch_optional(&self.connection_pool) + .await?; + log_db_operation_time("get_last_processed_height", start); + + if let Some(row) = maybe_record { + Ok(row.last_processed_height) + } else { + Ok(-1) + } + } + + pub(crate) async fn get_pruned_height(&self) -> Result { + trace!("get_pruned_height"); + let start = Instant::now(); + + let maybe_record = sqlx::query!( + r#" + SELECT last_pruned_height FROM pruning + "# + ) + .fetch_optional(&self.connection_pool) + .await?; + + log_db_operation_time("get_pruned_height", start); + + if let Some(row) = maybe_record { + Ok(row.last_pruned_height) + } else { + Ok(-1) + } + } +} + +// make those generic over executor so that they could be performed over connection pool and a tx + +#[instrument(skip(executor))] +pub(crate) async fn insert_validator<'a, E>( + consensus_address: String, + consensus_pubkey: String, + executor: E, +) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Postgres>, +{ + trace!("insert_validator"); + let start = Instant::now(); + + sqlx::query!( + r#" + INSERT INTO validator (consensus_address, consensus_pubkey) + VALUES ($1, $2) + ON CONFLICT (consensus_address) DO NOTHING + "#, + consensus_address, + consensus_pubkey + ) + .execute(executor) + .await?; + log_db_operation_time("insert_validator", start); + + Ok(()) +} + +#[instrument(skip(executor))] +pub(crate) async fn insert_block<'a, E>( + height: i64, + hash: String, + num_txs: u32, + total_gas: i64, + proposer_address: String, + timestamp: OffsetDateTime, + executor: E, +) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Postgres>, +{ + trace!("insert_block"); + let start = Instant::now(); + + sqlx::query!( + r#" + INSERT INTO block (height, hash, num_txs, total_gas, proposer_address, timestamp) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT DO NOTHING + "#, + height, + hash, + num_txs as i32, + total_gas, + proposer_address, + timestamp + ) + .execute(executor) + .await?; + log_db_operation_time("insert_block", start); + + Ok(()) +} + +#[instrument(skip(executor))] +pub(crate) async fn insert_precommit<'a, E>( + validator_address: String, + height: i64, + timestamp: OffsetDateTime, + voting_power: i64, + proposer_priority: i64, + executor: E, +) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Postgres>, +{ + trace!("insert_precommit"); + let start = Instant::now(); + + sqlx::query!( + r#" + INSERT INTO pre_commit (validator_address, height, timestamp, voting_power, proposer_priority) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (validator_address, timestamp) DO NOTHING + "#, + validator_address, + height, + timestamp, + voting_power, + proposer_priority + ) + .execute(executor) + .await?; + log_db_operation_time("insert_precommit", start); + + Ok(()) +} + +#[instrument(skip(executor))] +#[allow(clippy::too_many_arguments)] +pub(crate) async fn insert_transaction<'a, E>( + hash: String, + height: i64, + index: i64, + success: bool, + num_messages: i64, + messages: Vec, + memo: String, + signatures: Vec, + signer_infos: Vec, + fee: Fee, + gas_wanted: i64, + gas_used: i64, + raw_log: String, + executor: E, +) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Postgres>, +{ + trace!("insert_transaction"); + let start = Instant::now(); + + sqlx::query!( + r#" + INSERT INTO transaction (hash, height, "index", success, num_messages, messages, memo, signatures, signer_infos, fee, gas_wanted, gas_used, raw_log) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) + ON CONFLICT (hash) DO UPDATE + SET height = excluded.height, + "index" = excluded."index", + success = excluded.success, + num_messages = excluded.num_messages, + messages = excluded.messages, + memo = excluded.memo, + signatures = excluded.signatures, + signer_infos = excluded.signer_infos, + fee = excluded.fee, + gas_wanted = excluded.gas_wanted, + gas_used = excluded.gas_used, + raw_log = excluded.raw_log + "#, + hash, + height, + index as i32, + success, + num_messages as i32, + serde_json::json!(messages.iter().map(|msg| { + serde_json::json!({ + "type_url": msg.type_url, + "value": base64::encode(&msg.value) + }) + }).collect::>()), + memo, + &signatures, + serde_json::json!(signer_infos.iter().map(|info| { + serde_json::json!({ + "public_key": { + "type_url": info.public_key.type_url, + "value": base64::encode(&info.public_key.value) + }, + "mode_info": info.mode_info, + "sequence": info.sequence + }) + }).collect::>()), + serde_json::json!({ + "amount": fee.amount, + "gas_limit": fee.gas_limit, + "payer": fee.payer, + "granter": fee.granter + }), + gas_wanted, + gas_used, + raw_log + ) + .execute(executor) + .await?; + log_db_operation_time("insert_transaction", start); + + Ok(()) +} + +#[instrument(skip(executor))] +pub(crate) async fn insert_message<'a, E>( + transaction_hash: String, + index: i64, + typ: String, + value: cosmrs::Any, + involved_accounts: Vec, + height: i64, + executor: E, +) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Postgres>, +{ + trace!("insert_message"); + let start = Instant::now(); + + sqlx::query!( + r#" + INSERT INTO message (transaction_hash, "index", type, value, involved_accounts_addresses, height) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (transaction_hash, "index") DO UPDATE + SET height = excluded.height, + type = excluded.type, + value = excluded.value, + involved_accounts_addresses = excluded.involved_accounts_addresses + "#, + transaction_hash, + index, + typ, + serde_json::json!({ + "type_url": value.type_url, + "value": base64::encode(&value.value) + }), + &involved_accounts, + height + ) + .execute(executor) + .await?; + log_db_operation_time("insert_message", start); + + Ok(()) +} + +#[instrument(skip(executor))] +pub(crate) async fn update_last_processed<'a, E>( + height: i64, + executor: E, +) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Postgres>, +{ + trace!("update_last_processed"); + let start = Instant::now(); + + sqlx::query!( + "UPDATE metadata SET last_processed_height = GREATEST(last_processed_height, $1)", + height + ) + .execute(executor) + .await?; + log_db_operation_time("update_last_processed", start); + + Ok(()) +} + +#[instrument(skip(executor))] +pub(crate) async fn update_last_pruned<'a, E>(height: i64, executor: E) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Postgres>, +{ + trace!("update_last_pruned"); + let start = Instant::now(); + + sqlx::query!("UPDATE pruning SET last_pruned_height = $1", height) + .execute(executor) + .await?; + log_db_operation_time("update_last_pruned", start); + + Ok(()) +} + +pub(crate) async fn prune_blocks<'a, E>(oldest_to_keep: i64, executor: E) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Postgres>, +{ + trace!("prune_blocks"); + let start = Instant::now(); + + sqlx::query!("DELETE FROM block WHERE height < $1", oldest_to_keep) + .execute(executor) + .await?; + log_db_operation_time("prune_blocks", start); + + Ok(()) +} + +pub(crate) async fn prune_pre_commits<'a, E>( + oldest_to_keep: i64, + executor: E, +) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Postgres>, +{ + trace!("prune_pre_commits"); + let start = Instant::now(); + + sqlx::query!("DELETE FROM pre_commit WHERE height < $1", oldest_to_keep) + .execute(executor) + .await?; + log_db_operation_time("prune_pre_commits", start); + + Ok(()) +} + +pub(crate) async fn prune_transactions<'a, E>( + oldest_to_keep: i64, + executor: E, +) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Postgres>, +{ + trace!("prune_transactions"); + let start = Instant::now(); + + sqlx::query!( + "DELETE FROM transaction WHERE height < $1", + oldest_to_keep + ) + .execute(executor) + .await?; + log_db_operation_time("prune_transactions", start); + + Ok(()) +} + +pub(crate) async fn prune_messages<'a, E>( + oldest_to_keep: i64, + executor: E, +) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Postgres>, +{ + trace!("prune_messages"); + let start = Instant::now(); + + sqlx::query!("DELETE FROM message WHERE height < $1", oldest_to_keep) + .execute(executor) + .await?; + log_db_operation_time("prune_messages", start); + + Ok(()) +} + +#[derive(Debug, Clone, serde::Serialize)] +pub struct Coin { + pub denom: String, + pub amount: String, +} + +#[derive(Debug, Clone)] +pub struct SignerInfo { + pub public_key: cosmrs::Any, + pub mode_info: String, + pub sequence: i64, +} + +#[derive(Debug, Clone)] +pub struct Fee { + pub amount: Vec, + pub gas_limit: i64, + pub payer: String, + pub granter: String, +} diff --git a/common/nyxd-scraper-psql/src/storage/mod.rs b/common/nyxd-scraper-psql/src/storage/mod.rs new file mode 100644 index 0000000000..8358109a8b --- /dev/null +++ b/common/nyxd-scraper-psql/src/storage/mod.rs @@ -0,0 +1,420 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::block_processor::types::{FullBlockInformation, ParsedTransactionResponse}; +use crate::error::ScraperError; +use crate::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, Coin, Fee, SignerInfo, +}; +use crate::storage::models::{CommitSignature, Validator}; +use sqlx::types::time::OffsetDateTime; +use sqlx::{ConnectOptions, Postgres, Transaction}; +use std::str::FromStr; +use tendermint::block::{Commit, CommitSig}; +use tendermint::Block; +use tendermint_rpc::endpoint::validators; +use tokio::time::Instant; +use tracing::{debug, error, info, instrument, trace, warn}; + +mod helpers; +mod manager; +pub mod models; + +pub type StorageTransaction = Transaction<'static, Postgres>; + +#[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(database_url: &str) -> Result { + let mut opts = sqlx::postgres::PgConnectOptions::from_str(database_url) + .map_err(|err| ScraperError::InternalDatabaseError(err))?; + + // TODO: do we want auto_vacuum ? + + opts.disable_statement_logging(); + + let connection_pool = match sqlx::PgPool::connect_with(opts).await { + Ok(db) => db, + Err(err) => { + error!("Failed to connect to PostgreSQL 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, +) -> 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?; + + // persist commits + 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 + for chain_tx in &block.transactions { + persist_transaction(chain_tx, tx).await?; + } + + update_last_processed(block.block.header.height.into(), tx).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(), + &mut *tx, + ) + .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, + ) + .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(), + &mut *tx, + ) + .await?; + } + + Ok(()) +} + +async fn persist_transaction( + chain_tx: &ParsedTransactionResponse, + tx: &mut Transaction<'_, Postgres>, +) -> Result<(), ScraperError> { + let signer_infos = chain_tx + .tx + .auth_info + .signer_infos + .iter() + .map(|info| SignerInfo { + public_key: info.public_key.clone() + .map(|pk| pk.into()) + .unwrap_or_default(), + mode_info: serde_json::to_string(&serde_json::json!({ + "single": { + "mode": match &info.mode_info { + cosmrs::tx::ModeInfo::Single(s) => s.mode as u32, + _ => 0, + } + } + })).unwrap_or_default(), + sequence: info.sequence as i64, + }) + .collect(); + + let fee = Fee { + amount: chain_tx + .tx + .auth_info + .fee + .amount + .iter() + .map(|coin| Coin { + denom: coin.denom.to_string(), + amount: coin.amount.to_string(), + }) + .collect(), + gas_limit: chain_tx.tx.auth_info.fee.gas_limit as i64, + payer: chain_tx.tx.auth_info.fee.payer + .clone() + .map(|id| id.to_string()) + .unwrap_or_default(), + granter: chain_tx.tx.auth_info.fee.granter + .clone() + .map(|id| id.to_string()) + .unwrap_or_default(), + }; + + let signatures = chain_tx.tx.signatures + .iter() + .map(|sig| base64::encode(sig)) + .collect(); + + 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.messages.clone(), + chain_tx.tx.body.memo.clone(), + signatures, + signer_infos, + fee, + chain_tx.tx_result.gas_wanted, + chain_tx.tx_result.gas_used, + chain_tx.tx_result.log.clone(), + &mut *tx, + ) + .await?; + + for (index, msg) in chain_tx.tx.body.messages.iter().enumerate() { + let involved_accounts = extract_involved_accounts(msg); + insert_message( + chain_tx.hash.to_string(), + index as i64, + msg.type_url.clone(), + msg.clone(), + involved_accounts, + chain_tx.height.into(), + &mut *tx, + ) + .await?; + } + + Ok(()) +} + +fn extract_involved_accounts(_msg: &cosmrs::Any) -> Vec { + // This is a placeholder implementation + // TODO: Implement proper account extraction based on message type + vec![] +} \ No newline at end of file diff --git a/common/nyxd-scraper-psql/src/storage/models.rs b/common/nyxd-scraper-psql/src/storage/models.rs new file mode 100644 index 0000000000..74a3dd9b1c --- /dev/null +++ b/common/nyxd-scraper-psql/src/storage/models.rs @@ -0,0 +1,30 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use sqlx::types::time::OffsetDateTime; +use sqlx::FromRow; + +#[derive(Debug, Clone, Eq, PartialEq, Hash, FromRow)] +pub struct Validator { + pub consensus_address: String, + pub consensus_pubkey: String, +} + +#[derive(Debug, Clone, FromRow)] +pub struct Block { + pub height: i64, + pub hash: String, + pub num_txs: u32, + pub total_gas: i64, + pub proposer_address: String, + pub timestamp: OffsetDateTime, +} + +#[derive(Debug, Clone, FromRow)] +pub struct CommitSignature { + pub height: i64, + pub validator_address: String, + pub voting_power: i64, + pub proposer_priority: i64, + pub timestamp: OffsetDateTime, +}