From c92a7e3e3521821b2ef2db144a6bd0d672215df9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 7 Dec 2023 10:46:17 +0000 Subject: [PATCH 1/5] Squashed scraper globally updated sqlx to 0.6.3 wip basic processing loop wip starting on modules all of the requesting logic, catching up, etc remaining work includes persisting the data wip persisting block data initial and extremely basic nyxd block scraper --- Cargo.toml | 3 + common/client-core/Cargo.toml | 4 +- common/credential-storage/Cargo.toml | 6 +- common/nyxd-scraper/Cargo.toml | 37 ++ common/nyxd-scraper/build.rs | 28 ++ .../sql_migrations/01_metadata.sql | 10 + .../nyxd-scraper/sql_migrations/02_cosmos.sql | 76 +++++ .../nyxd-scraper/src/block_processor/mod.rs | 217 ++++++++++++ .../nyxd-scraper/src/block_processor/types.rs | 121 +++++++ .../nyxd-scraper/src/block_requester/mod.rs | 91 +++++ common/nyxd-scraper/src/constants.rs | 47 +++ common/nyxd-scraper/src/error.rs | 131 +++++++ common/nyxd-scraper/src/helpers.rs | 46 +++ common/nyxd-scraper/src/lib.rs | 18 + .../nyxd-scraper/src/modules/block_module.rs | 16 + common/nyxd-scraper/src/modules/mod.rs | 10 + common/nyxd-scraper/src/modules/msg_module.rs | 19 ++ common/nyxd-scraper/src/modules/tx_module.rs | 16 + common/nyxd-scraper/src/rpc_client.rs | 175 ++++++++++ common/nyxd-scraper/src/scraper/mod.rs | 195 +++++++++++ common/nyxd-scraper/src/scraper/subscriber.rs | 129 +++++++ common/nyxd-scraper/src/storage/helpers.rs | 2 + common/nyxd-scraper/src/storage/manager.rs | 320 ++++++++++++++++++ common/nyxd-scraper/src/storage/mod.rs | 298 ++++++++++++++++ common/nyxd-scraper/src/storage/models.rs | 30 ++ common/statistics/Cargo.toml | 2 +- gateway/Cargo.toml | 4 +- gateway/src/node/storage/inboxes.rs | 12 +- nym-api/Cargo.toml | 4 +- nym-wallet/Cargo.lock | 4 +- .../network-requester/Cargo.toml | 2 +- .../network-statistics/Cargo.toml | 4 +- 32 files changed, 2060 insertions(+), 17 deletions(-) create mode 100644 common/nyxd-scraper/Cargo.toml create mode 100644 common/nyxd-scraper/build.rs create mode 100644 common/nyxd-scraper/sql_migrations/01_metadata.sql create mode 100644 common/nyxd-scraper/sql_migrations/02_cosmos.sql create mode 100644 common/nyxd-scraper/src/block_processor/mod.rs create mode 100644 common/nyxd-scraper/src/block_processor/types.rs create mode 100644 common/nyxd-scraper/src/block_requester/mod.rs create mode 100644 common/nyxd-scraper/src/constants.rs create mode 100644 common/nyxd-scraper/src/error.rs create mode 100644 common/nyxd-scraper/src/helpers.rs create mode 100644 common/nyxd-scraper/src/lib.rs create mode 100644 common/nyxd-scraper/src/modules/block_module.rs create mode 100644 common/nyxd-scraper/src/modules/mod.rs create mode 100644 common/nyxd-scraper/src/modules/msg_module.rs create mode 100644 common/nyxd-scraper/src/modules/tx_module.rs create mode 100644 common/nyxd-scraper/src/rpc_client.rs create mode 100644 common/nyxd-scraper/src/scraper/mod.rs create mode 100644 common/nyxd-scraper/src/scraper/subscriber.rs create mode 100644 common/nyxd-scraper/src/storage/helpers.rs create mode 100644 common/nyxd-scraper/src/storage/manager.rs create mode 100644 common/nyxd-scraper/src/storage/mod.rs create mode 100644 common/nyxd-scraper/src/storage/models.rs diff --git a/Cargo.toml b/Cargo.toml index 52b8a5f382..d3f1599514 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -67,6 +67,7 @@ members = [ "common/nymsphinx/params", "common/nymsphinx/routing", "common/nymsphinx/types", + "common/nyxd-scraper", "common/pemstore", "common/socks5-client-core", "common/socks5/proxy-helpers", @@ -159,6 +160,7 @@ reqwest = "0.11.22" schemars = "0.8.1" serde = "1.0.152" serde_json = "1.0.91" +sqlx = "0.6.3" tap = "1.0.1" time = "0.3.30" thiserror = "1.0.48" @@ -200,6 +202,7 @@ cw-controllers = { version = "=1.1.0" } # cosmrs-related bip32 = "0.5.1" cosmrs = "=0.15.0" +tendermint = "0.34" # same version as used by cosmrs tendermint-rpc = "0.34" # same version as used by cosmrs tendermint = "0.34" # same version as used by cosmrs prost = "0.12" diff --git a/common/client-core/Cargo.toml b/common/client-core/Cargo.toml index e68dc29165..556978f1f8 100644 --- a/common/client-core/Cargo.toml +++ b/common/client-core/Cargo.toml @@ -59,7 +59,7 @@ features = ["time"] version = "0.20.1" [target."cfg(not(target_arch = \"wasm32\"))".dependencies.sqlx] -version = "0.6.2" +workspace = true features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] optional = true @@ -90,7 +90,7 @@ tempfile = "3.1.0" [build-dependencies] tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } -sqlx = { version = "0.6.2", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] } +sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] } [features] default = [] diff --git a/common/credential-storage/Cargo.toml b/common/credential-storage/Cargo.toml index 717cca0669..9beb030e22 100644 --- a/common/credential-storage/Cargo.toml +++ b/common/credential-storage/Cargo.toml @@ -14,14 +14,14 @@ thiserror = { workspace = true } tokio = { version = "1.24.1", features = ["sync"]} [target."cfg(not(target_arch = \"wasm32\"))".dependencies.sqlx] -version = "0.5" +workspace = true features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] [target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio] -version = "1.24.1" +workspace = true features = [ "rt-multi-thread", "net", "signal", "fs" ] [build-dependencies] -sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] } +sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] } tokio = { version = "1.24.1", features = ["rt-multi-thread", "macros"] } diff --git a/common/nyxd-scraper/Cargo.toml b/common/nyxd-scraper/Cargo.toml new file mode 100644 index 0000000000..772ab8e7b3 --- /dev/null +++ b/common/nyxd-scraper/Cargo.toml @@ -0,0 +1,37 @@ +[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 +const_format = "0.2.32" +cosmrs.workspace = true +eyre = "0.6.9" +futures.workspace = true +sha2 = "0.10.8" +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 +tokio = { workspace = true, features = ["full"] } +tokio-stream = "0.1.14" +tokio-util = { version = "0.7.10", 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", "sqlite", "macros", "migrate"] } +tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } diff --git a/common/nyxd-scraper/build.rs b/common/nyxd-scraper/build.rs new file mode 100644 index 0000000000..cfe3b9c079 --- /dev/null +++ b/common/nyxd-scraper/build.rs @@ -0,0 +1,28 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +#[tokio::main] +async fn main() { + use sqlx::{Connection, SqliteConnection}; + use std::env; + + let out_dir = env::var("OUT_DIR").unwrap(); + let database_path = format!("{out_dir}/scraper-example.sqlite"); + + let mut conn = SqliteConnection::connect(&format!("sqlite://{database_path}?mode=rwc")) + .await + .expect("Failed to create SQLx database connection"); + + sqlx::migrate!("./sql_migrations") + .run(&mut conn) + .await + .expect("Failed to perform SQLx migrations"); + + #[cfg(target_family = "unix")] + println!("cargo:rustc-env=DATABASE_URL=sqlite://{}", &database_path); + + #[cfg(target_family = "windows")] + // for some strange reason we need to add a leading `/` to the windows path even though it's + // not a valid windows path... but hey, it works... + println!("cargo:rustc-env=DATABASE_URL=sqlite:///{}", &database_path); +} diff --git a/common/nyxd-scraper/sql_migrations/01_metadata.sql b/common/nyxd-scraper/sql_migrations/01_metadata.sql new file mode 100644 index 0000000000..bce13940e8 --- /dev/null +++ b/common/nyxd-scraper/sql_migrations/01_metadata.sql @@ -0,0 +1,10 @@ +/* + * Copyright 2023 - Nym Technologies SA + * SPDX-License-Identifier: Apache-2.0 + */ + +CREATE TABLE METADATA +( + id INTEGER PRIMARY KEY CHECK (id = 0), + last_processed_height INTEGER NOT NULL +); \ No newline at end of file diff --git a/common/nyxd-scraper/sql_migrations/02_cosmos.sql b/common/nyxd-scraper/sql_migrations/02_cosmos.sql new file mode 100644 index 0000000000..373949a923 --- /dev/null +++ b/common/nyxd-scraper/sql_migrations/02_cosmos.sql @@ -0,0 +1,76 @@ +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 TIMESTAMP WITHOUT TIME ZONE 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 TIMESTAMP WITHOUT TIME ZONE 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); + +-- no JSONB in sqlite (yet) : ) +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 '[]'::JSONB, + 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 +-- logs JSONB +); +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/src/block_processor/mod.rs b/common/nyxd-scraper/src/block_processor/mod.rs new file mode 100644 index 0000000000..e1ac8cc749 --- /dev/null +++ b/common/nyxd-scraper/src/block_processor/mod.rs @@ -0,0 +1,217 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +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 futures::StreamExt; +use std::collections::BTreeMap; +use std::ops::Add; +use std::time::Duration; +use tokio::sync::mpsc::{Sender, UnboundedReceiver}; +use tokio::time::{interval_at, Instant}; +use tokio_stream::wrappers::UnboundedReceiverStream; +use tokio_util::sync::CancellationToken; +use tracing::{debug, error, info, warn}; + +pub(crate) mod types; + +const MISSING_BLOCKS_CHECK_INTERVAL: Duration = Duration::from_secs(30); +const MAX_MISSING_BLOCKS_DELAY: Duration = Duration::from_secs(15); + +pub struct BlockProcessor { + cancel: CancellationToken, + last_processed_height: u32, + last_processed_at: Instant, + 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 fn new( + cancel: CancellationToken, + incoming: UnboundedReceiver, + block_requester: Sender, + storage: ScraperStorage, + rpc_client: RpcClient, + ) -> Self { + BlockProcessor { + cancel, + last_processed_height: 0, + last_processed_at: Instant::now(), + queued_blocks: Default::default(), + rpc_client, + incoming: incoming.into(), + block_requester, + storage, + block_modules: vec![], + tx_modules: vec![], + msg_modules: vec![], + } + } + + 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? + } + } + } + + tx.commit() + .await + .map_err(|source| ScraperError::StorageTxCommitFailure { source })?; + + self.last_processed_height = full_info.block.header.height.value() as u32; + self.last_processed_at = Instant::now(); + + 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; + } + + 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(()); + } + + // 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.block_requester + .send(BlockRequest::Range(request_range)) + .await?; + + Ok(()) + } + + async fn next_incoming(&mut self, block: BlockToProcess) { + let height = block.height; + + if self.last_processed_height == 0 { + // TODO: load it from storage instead and make sure storage always has some data in there + self.last_processed_height = height - 1 + } + + if self.last_processed_height + 1 != height { + if self.queued_blocks.insert(height, block).is_some() { + error!("duplicate queued up block for height {height}") + } + return; + } else 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; + } + } + + 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, + ); + + 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/src/block_processor/types.rs b/common/nyxd-scraper/src/block_processor/types.rs new file mode 100644 index 0000000000..93ffb3ba76 --- /dev/null +++ b/common/nyxd-scraper/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/src/block_requester/mod.rs b/common/nyxd-scraper/src/block_requester/mod.rs new file mode 100644 index 0000000000..b2a1a4d43a --- /dev/null +++ b/common/nyxd-scraper/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/src/constants.rs b/common/nyxd-scraper/src/constants.rs new file mode 100644 index 0000000000..9401cd360e --- /dev/null +++ b/common/nyxd-scraper/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/src/error.rs b/common/nyxd-scraper/src/error.rs new file mode 100644 index 0000000000..dec4fb26e9 --- /dev/null +++ b/common/nyxd-scraper/src/error.rs @@ -0,0 +1,131 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +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("can't add any modules to the scraper as it's 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")] + MaximumSubscriptionFailures, + + #[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 }, +} + +impl From> for ScraperError { + fn from(_: SendError) -> Self { + ScraperError::ClosedChannelError + } +} diff --git a/common/nyxd-scraper/src/helpers.rs b/common/nyxd-scraper/src/helpers.rs new file mode 100644 index 0000000000..d11b25a2f3 --- /dev/null +++ b/common/nyxd-scraper/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/src/lib.rs b/common/nyxd-scraper/src/lib.rs new file mode 100644 index 0000000000..79372e0e30 --- /dev/null +++ b/common/nyxd-scraper/src/lib.rs @@ -0,0 +1,18 @@ +// 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 modules::{BlockModule, MsgModule, TxModule}; +pub use scraper::{Config, NyxdScraper}; diff --git a/common/nyxd-scraper/src/modules/block_module.rs b/common/nyxd-scraper/src/modules/block_module.rs new file mode 100644 index 0000000000..9ca1ba9b20 --- /dev/null +++ b/common/nyxd-scraper/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/src/modules/mod.rs b/common/nyxd-scraper/src/modules/mod.rs new file mode 100644 index 0000000000..1f37d149e0 --- /dev/null +++ b/common/nyxd-scraper/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/src/modules/msg_module.rs b/common/nyxd-scraper/src/modules/msg_module.rs new file mode 100644 index 0000000000..df64761f7d --- /dev/null +++ b/common/nyxd-scraper/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/src/modules/tx_module.rs b/common/nyxd-scraper/src/modules/tx_module.rs new file mode 100644 index 0000000000..07d012ab5a --- /dev/null +++ b/common/nyxd-scraper/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/src/rpc_client.rs b/common/nyxd-scraper/src/rpc_client.rs new file mode 100644 index 0000000000..d526a07770 --- /dev/null +++ b/common/nyxd-scraper/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 for {tx_hash}"); + + 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/src/scraper/mod.rs b/common/nyxd-scraper/src/scraper/mod.rs new file mode 100644 index 0000000000..8ba4e40398 --- /dev/null +++ b/common/nyxd-scraper/src/scraper/mod.rs @@ -0,0 +1,195 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::block_processor::BlockProcessor; +use crate::block_requester::BlockRequester; +use crate::error::ScraperError; +use crate::modules::{BlockModule, MsgModule, TxModule}; +use crate::rpc_client::RpcClient; +use crate::scraper::subscriber::{run_websocket_driver, ChainSubscriber}; +use crate::storage::ScraperStorage; +use std::path::PathBuf; +use tendermint_rpc::WebSocketClientDriver; +use tokio::sync::mpsc::{channel, unbounded_channel}; +use tokio_util::sync::CancellationToken; +use tokio_util::task::TaskTracker; +use tracing::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 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.cancel_token.clone(), + processing_rx, + req_tx, + scraper.storage.clone(), + rpc_client, + ); + 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 mut chain_subscriber = ChainSubscriber::new( + &scraper.config.websocket_url, + scraper.cancel_token.clone(), + processing_tx, + ) + .await?; + let ws_driver = chain_subscriber.ws_driver(); + + scraper.start_tasks( + block_requester, + block_processor, + chain_subscriber, + ws_driver, + ); + + 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, + storage: ScraperStorage, +} + +impl NyxdScraper { + pub fn builder(config: Config) -> NyxdScraperBuilder { + NyxdScraperBuilder::new(config) + } + + pub async fn new(config: Config) -> Result { + let storage = ScraperStorage::init(&config.database_path).await?; + + Ok(NyxdScraper { + config, + task_tracker: TaskTracker::new(), + cancel_token: CancellationToken::new(), + storage, + }) + } + + fn start_tasks( + &self, + mut block_requester: BlockRequester, + mut block_processor: BlockProcessor, + mut chain_subscriber: ChainSubscriber, + ws_driver: WebSocketClientDriver, + ) { + 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 + .spawn(run_websocket_driver(ws_driver, self.cancel_token.clone())); + self.task_tracker.close(); + } + + pub async fn start(&self) -> Result<(), ScraperError> { + let (processing_tx, processing_rx) = unbounded_channel(); + let (req_tx, req_rx) = channel(5); + + let rpc_client = RpcClient::new(&self.config.rpc_url)?; + + // create the tasks + let block_requester = BlockRequester::new( + self.cancel_token.clone(), + rpc_client.clone(), + req_rx, + processing_tx.clone(), + ); + let block_processor = BlockProcessor::new( + self.cancel_token.clone(), + processing_rx, + req_tx, + self.storage.clone(), + rpc_client, + ); + let mut chain_subscriber = ChainSubscriber::new( + &self.config.websocket_url, + self.cancel_token.clone(), + processing_tx, + ) + .await?; + let ws_driver = chain_subscriber.ws_driver(); + + // spawn them + self.start_tasks( + block_requester, + block_processor, + chain_subscriber, + ws_driver, + ); + + Ok(()) + } + + pub async fn stop(self) { + info!("stopping the chain scrapper"); + self.cancel_token.cancel(); + self.task_tracker.wait().await + } +} diff --git a/common/nyxd-scraper/src/scraper/subscriber.rs b/common/nyxd-scraper/src/scraper/subscriber.rs new file mode 100644 index 0000000000..6e9fd3e85d --- /dev/null +++ b/common/nyxd-scraper/src/scraper/subscriber.rs @@ -0,0 +1,129 @@ +// 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 tokio::sync::mpsc::UnboundedSender; +use tokio_stream::StreamExt; +use tokio_util::sync::CancellationToken; +use tracing::{error, info, warn}; +use url::Url; + +const MAX_FAILURES: usize = 10; + +pub struct ChainSubscriber { + cancel: CancellationToken, + block_sender: UnboundedSender, + + websocket_client: WebSocketClient, + websocket_driver: Option, +} + +impl ChainSubscriber { + pub async fn new( + websocket_endpoint: &Url, + cancel: CancellationToken, + 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, + block_sender, + 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(()) + } + + pub(crate) async fn run(&mut self) -> Result<(), ScraperError> { + let _drop_guard = self.cancel.clone().drop_guard(); + + 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"); + break + } + maybe_event = subs.next() => { + let Some(maybe_event) = maybe_event else { + warn!("stopped receiving new events"); + break; + }; + 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 { + // note: the drop_guard will get dropped and thus cause a shutdown + return Err(ScraperError::MaximumSubscriptionFailures); + } + } + } + } + Ok(()) + } + + pub(crate) fn ws_driver(&mut self) -> WebSocketClientDriver { + #[allow(clippy::expect_used)] + self.websocket_driver + .take() + .expect("websocket driver has already been started!") + } +} + +pub async fn run_websocket_driver(driver: WebSocketClientDriver, cancel: CancellationToken) { + info!("starting websocket driver"); + tokio::select! { + _ = cancel.cancelled() => { + info!("received cancellation token") + } + res = driver.run() => { + match res { + Ok(_) => info!("our websocket driver has finished execution"), + Err(err) => { + // TODO: in the future just attempt to reconnect + error!("our websocket driver has errored out: {err}") + } + } + cancel.cancel() + } + } +} diff --git a/common/nyxd-scraper/src/storage/helpers.rs b/common/nyxd-scraper/src/storage/helpers.rs new file mode 100644 index 0000000000..38dc36ebc3 --- /dev/null +++ b/common/nyxd-scraper/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/src/storage/manager.rs b/common/nyxd-scraper/src/storage/manager.rs new file mode 100644 index 0000000000..86e1f6898e --- /dev/null +++ b/common/nyxd-scraper/src/storage/manager.rs @@ -0,0 +1,320 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::storage::models::Validator; +use sqlx::types::time::OffsetDateTime; +use sqlx::{Executor, Sqlite}; +use tracing::{instrument, trace}; + +#[derive(Clone)] +pub(crate) struct StorageManager { + pub(crate) connection_pool: sqlx::SqlitePool, +} + +impl StorageManager { + pub(crate) async fn get_first_block_height_after( + &self, + time: OffsetDateTime, + ) -> Result, sqlx::Error> { + let maybe_record = sqlx::query!( + r#" + SELECT height + FROM block + WHERE timestamp > ? + ORDER BY timestamp + LIMIT 1 + "#, + time + ) + .fetch_optional(&self.connection_pool) + .await?; + + if let Some(row) = maybe_record { + Ok(row.height) + } else { + Ok(None) + } + } + + pub(crate) async fn get_last_block_height_before( + &self, + time: OffsetDateTime, + ) -> Result, sqlx::Error> { + let maybe_record = sqlx::query!( + r#" + SELECT height + FROM block + WHERE timestamp < ? + ORDER BY timestamp DESC + LIMIT 1 + "#, + time + ) + .fetch_optional(&self.connection_pool) + .await?; + + if let Some(row) = maybe_record { + Ok(row.height) + } else { + Ok(None) + } + } + + pub(crate) async fn get_signed_between( + &self, + consensus_address: &str, + start_height: i64, + end_height: i64, + ) -> Result { + let count = sqlx::query!( + r#" + SELECT COUNT(*) as count FROM pre_commit + WHERE + validator_address == ? + AND height > ? + AND height < ? + "#, + consensus_address, + start_height, + end_height + ) + .fetch_one(&self.connection_pool) + .await? + .count; + + Ok(count) + } + + pub(crate) async fn get_block_validators( + &self, + height: i64, + ) -> Result, sqlx::Error> { + sqlx::query_as!( + Validator, + r#" + SELECT * FROM validator + WHERE EXISTS ( + SELECT 1 FROM pre_commit + WHERE height == ? + AND pre_commit.validator_address = validator.consensus_address + ) + "#, + height + ) + .fetch_all(&self.connection_pool) + .await + } + + pub(crate) async fn get_last_processed_height(&self) -> Result { + let maybe_record = sqlx::query!( + r#" + SELECT last_processed_height FROM metadata + "# + ) + .fetch_optional(&self.connection_pool) + .await?; + + if let Some(row) = maybe_record { + Ok(row.last_processed_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 = Sqlite>, +{ + trace!("insert validator"); + + sqlx::query!( + r#" + INSERT INTO validator (consensus_address, consensus_pubkey) + VALUES (?, ?) + ON CONFLICT DO NOTHING + "#, + consensus_address, + consensus_pubkey + ) + .execute(executor) + .await?; + + 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 = Sqlite>, +{ + trace!("insert block"); + + sqlx::query!( + r#" + INSERT INTO block (height, hash, num_txs, total_gas, proposer_address, timestamp) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT DO NOTHING + "#, + height, + hash, + num_txs, + total_gas, + proposer_address, + timestamp + ) + .execute(executor) + .await?; + + 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 = Sqlite>, +{ + trace!("insert precommit"); + + sqlx::query!( + r#" + INSERT INTO pre_commit (validator_address, height, timestamp, voting_power, proposer_priority) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT (validator_address, timestamp) DO NOTHING + "#, + validator_address, + height, + timestamp, + voting_power, + proposer_priority + ) + .execute(executor) + .await?; + + 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, + messages: i64, + memo: String, + gas_wanted: i64, + gas_used: i64, + raw_log: String, + executor: E, +) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Sqlite>, +{ + trace!("insert transaction"); + + sqlx::query!( + r#" + INSERT INTO "transaction" (hash, height, "index", success, num_messages, memo, gas_wanted, gas_used, raw_log) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (hash) DO UPDATE + SET height = excluded.height, + "index" = excluded."index", + success = excluded.success, + num_messages = excluded.num_messages, + memo = excluded.memo, + gas_wanted = excluded.gas_wanted, + gas_used = excluded.gas_used, + raw_log = excluded.raw_log + "#, + hash, + height, + index, + success, + messages, + memo, + gas_wanted, + gas_used, + raw_log, + ) + .execute(executor) + .await?; + + Ok(()) +} + +#[instrument(skip(executor))] +pub(crate) async fn insert_message<'a, E>( + transaction_hash: String, + index: i64, + typ: String, + height: i64, + executor: E, +) -> Result<(), sqlx::Error> +where + E: Executor<'a, Database = Sqlite>, +{ + trace!("insert message"); + + sqlx::query!( + r#" + INSERT INTO message (transaction_hash, "index", type, height) + VALUES (?, ?, ?, ?) + ON CONFLICT (transaction_hash, "index") DO UPDATE + SET height = excluded.height, + type = excluded.type + "#, + transaction_hash, + index, + typ, + height + ) + .execute(executor) + .await?; + + 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 = Sqlite>, +{ + trace!("update_last_processed"); + + sqlx::query!( + r#" + INSERT INTO metadata (last_processed_height) VALUES (?) + "#, + height + ) + .execute(executor) + .await?; + + Ok(()) +} diff --git a/common/nyxd-scraper/src/storage/mod.rs b/common/nyxd-scraper/src/storage/mod.rs new file mode 100644 index 0000000000..03ade02a93 --- /dev/null +++ b/common/nyxd-scraper/src/storage/mod.rs @@ -0,0 +1,298 @@ +// 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, + update_last_processed, StorageManager, +}; +use crate::storage::models::Validator; +use sqlx::types::time::OffsetDateTime; +use sqlx::{ConnectOptions, Sqlite, Transaction}; +use std::fmt::Debug; +use std::path::Path; +use tendermint::block::{Commit, CommitSig}; +use tendermint::Block; +use tendermint_rpc::endpoint::validators; +use tracing::{debug, error, info, instrument, trace, warn}; + +mod helpers; +mod manager; +mod models; + +pub type StorageTransaction = Transaction<'static, Sqlite>; + +#[derive(Clone)] +pub struct ScraperStorage { + pub(crate) manager: StorageManager, +} + +impl ScraperStorage { + #[instrument] + pub async fn init + Debug>(database_path: P) -> Result { + // TODO: we can inject here more stuff based on our nym-api global config + // struct. Maybe different pool size or timeout intervals? + let mut opts = sqlx::sqlite::SqliteConnectOptions::new() + .filename(database_path) + .create_if_missing(true); + + // TODO: do we want auto_vacuum ? + + opts.disable_statement_logging(); + + 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 storage = ScraperStorage { + manager: StorageManager { connection_pool }, + }; + + Ok(storage) + } + + #[instrument(skip_all)] + pub async fn begin_processing_tx(&self) -> Result { + info!("starting storage tx"); + self.manager + .connection_pool + .begin() + .await + .map_err(|source| ScraperError::StorageTxBeginFailure { source }) + } + + 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_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_block_signers(&self, height: i64) -> Result, ScraperError> { + Ok(self.manager.get_block_validators(height).await?) + } + + pub async fn get_last_processed_height(&self) -> Result { + Ok(self.manager.get_last_processed_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 + 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).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_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(), + &mut *tx, + ) + .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(), + &mut *tx, + ) + .await? + } + } + + Ok(()) +} diff --git a/common/nyxd-scraper/src/storage/models.rs b/common/nyxd-scraper/src/storage/models.rs new file mode 100644 index 0000000000..38006bfdaf --- /dev/null +++ b/common/nyxd-scraper/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, 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, +} diff --git a/common/statistics/Cargo.toml b/common/statistics/Cargo.toml index cfdbe10a53..f431de5572 100644 --- a/common/statistics/Cargo.toml +++ b/common/statistics/Cargo.toml @@ -15,6 +15,6 @@ log = { workspace = true } reqwest = { workspace = true, features = ["json"] } serde = { version = "1.0", features = ["derive"] } serde_json = "1" -sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "chrono"]} +sqlx = { workspace = true, features = ["runtime-tokio-rustls", "chrono"]} thiserror = { workspace = true } tokio = { version = "1.24.1", features = [ "time" ] } diff --git a/gateway/Cargo.toml b/gateway/Cargo.toml index cd098b7030..a12ce97bb7 100644 --- a/gateway/Cargo.toml +++ b/gateway/Cargo.toml @@ -36,7 +36,7 @@ pretty_env_logger = "0.4" rand = "0.7" serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } -sqlx = { version = "0.5", features = [ +sqlx = { workspace = true, features = [ "runtime-tokio-rustls", "sqlite", "macros", @@ -89,7 +89,7 @@ hyper = "0.14.27" [build-dependencies] tokio = { version = "1.24.1", features = ["rt-multi-thread", "macros"] } -sqlx = { version = "0.5", features = [ +sqlx = { workspace = true, features = [ "runtime-tokio-rustls", "sqlite", "macros", diff --git a/gateway/src/node/storage/inboxes.rs b/gateway/src/node/storage/inboxes.rs index a24f962864..c94459fb88 100644 --- a/gateway/src/node/storage/inboxes.rs +++ b/gateway/src/node/storage/inboxes.rs @@ -75,7 +75,11 @@ impl InboxManager { sqlx::query_as!( StoredMessage, r#" - SELECT * FROM message_store + SELECT + id as "id!", + client_address_bs58 as "client_address_bs58!", + content as "content!" + FROM message_store WHERE client_address_bs58 = ? AND id > ? ORDER BY id ASC LIMIT ?; @@ -90,7 +94,11 @@ impl InboxManager { sqlx::query_as!( StoredMessage, r#" - SELECT * FROM message_store + SELECT + id as "id!", + client_address_bs58 as "client_address_bs58!", + content as "content!" + FROM message_store WHERE client_address_bs58 = ? ORDER BY id ASC LIMIT ?; diff --git a/nym-api/Cargo.toml b/nym-api/Cargo.toml index 785efe361e..3b19c6215b 100644 --- a/nym-api/Cargo.toml +++ b/nym-api/Cargo.toml @@ -53,7 +53,7 @@ ts-rs = { workspace = true, optional = true} anyhow = { workspace = true } getset = "0.1.1" -sqlx = { version = "0.6.2", features = [ +sqlx = { workspace = true, features = [ "runtime-tokio-rustls", "sqlite", "macros", @@ -115,7 +115,7 @@ generate-ts = ["ts-rs"] [build-dependencies] tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } -sqlx = { version = "0.6.2", features = [ +sqlx = { workspace = true, features = [ "runtime-tokio-rustls", "sqlite", "macros", diff --git a/nym-wallet/Cargo.lock b/nym-wallet/Cargo.lock index 8108a547af..89cfcaf991 100644 --- a/nym-wallet/Cargo.lock +++ b/nym-wallet/Cargo.lock @@ -4856,9 +4856,9 @@ dependencies = [ [[package]] name = "serde_json" -version = "1.0.104" +version = "1.0.108" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "076066c5f1078eac5b722a31827a8832fe108bed65dfa75e233c89f8206e976c" +checksum = "3d1c7e3eac408d115102c4c24ad393e0821bb3a5df4d506a80f85f7a742a526b" dependencies = [ "itoa 1.0.9", "ryu", diff --git a/service-providers/network-requester/Cargo.toml b/service-providers/network-requester/Cargo.toml index 39086ee399..2c6884ff90 100644 --- a/service-providers/network-requester/Cargo.toml +++ b/service-providers/network-requester/Cargo.toml @@ -32,7 +32,7 @@ regex = "1.8.4" reqwest = { workspace = true, features = ["json"] } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } -sqlx = { version = "0.6.1", features = ["runtime-tokio-rustls", "chrono"]} +sqlx = { workspace = true, features = ["runtime-tokio-rustls", "chrono"]} tap = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true, features = [ "net", "rt-multi-thread", "macros" ] } diff --git a/service-providers/network-statistics/Cargo.toml b/service-providers/network-statistics/Cargo.toml index b3d7ea663c..1895d2b88c 100644 --- a/service-providers/network-statistics/Cargo.toml +++ b/service-providers/network-statistics/Cargo.toml @@ -12,7 +12,7 @@ log = { workspace = true } pretty_env_logger = "0.4" rocket = { version = "0.5.0-rc.2", features = ["json"] } serde = { workspace = true, features = ["derive"] } -sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate", "chrono"]} +sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate", "chrono"]} thiserror = { workspace = true } tokio = { version = "1.4", features = [ "net", "rt-multi-thread", "macros", "time" ] } nym-bin-common = { path = "../../common/bin-common"} @@ -20,5 +20,5 @@ nym-statistics-common = { path = "../../common/statistics" } nym-task = { path = "../../common/task" } [build-dependencies] -sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] } +sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] } tokio = { version = "1.4", features = ["rt-multi-thread", "macros"] } From c3a9ceae525cfed1e70dd424240ff15a3ea3c975 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Tue, 12 Dec 2023 15:40:54 +0000 Subject: [PATCH 2/5] fixed update_last_processed --- common/nyxd-scraper/src/storage/manager.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/common/nyxd-scraper/src/storage/manager.rs b/common/nyxd-scraper/src/storage/manager.rs index 86e1f6898e..014aeb350f 100644 --- a/common/nyxd-scraper/src/storage/manager.rs +++ b/common/nyxd-scraper/src/storage/manager.rs @@ -309,7 +309,8 @@ where sqlx::query!( r#" - INSERT INTO metadata (last_processed_height) VALUES (?) + UPDATE metadata + SET last_processed_height = ? "#, height ) From fb1b58b5fb21402f7b87ac0dc46955f7a1744d67 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Tue, 12 Dec 2023 16:34:16 +0000 Subject: [PATCH 3/5] fixed startup sync --- .../nyxd-scraper/src/block_processor/mod.rs | 65 ++++++++++++++++--- common/nyxd-scraper/src/rpc_client.rs | 2 +- common/nyxd-scraper/src/scraper/mod.rs | 6 +- common/nyxd-scraper/src/storage/manager.rs | 25 ++++--- common/nyxd-scraper/src/storage/mod.rs | 7 +- 5 files changed, 80 insertions(+), 25 deletions(-) diff --git a/common/nyxd-scraper/src/block_processor/mod.rs b/common/nyxd-scraper/src/block_processor/mod.rs index e1ac8cc749..89e51d06b1 100644 --- a/common/nyxd-scraper/src/block_processor/mod.rs +++ b/common/nyxd-scraper/src/block_processor/mod.rs @@ -9,7 +9,7 @@ use crate::rpc_client::RpcClient; use crate::storage::{persist_block, ScraperStorage}; use futures::StreamExt; use std::collections::BTreeMap; -use std::ops::Add; +use std::ops::{Add, Range}; use std::time::Duration; use tokio::sync::mpsc::{Sender, UnboundedReceiver}; use tokio::time::{interval_at, Instant}; @@ -21,6 +21,7 @@ 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; pub struct BlockProcessor { cancel: CancellationToken, @@ -42,16 +43,18 @@ pub struct BlockProcessor { } impl BlockProcessor { - pub fn new( + pub async fn new( cancel: CancellationToken, incoming: UnboundedReceiver, block_requester: Sender, storage: ScraperStorage, rpc_client: RpcClient, - ) -> Self { - BlockProcessor { + ) -> Result { + let last_processed = storage.get_last_processed_height().await?; + + Ok(BlockProcessor { cancel, - last_processed_height: 0, + last_processed_height: last_processed.try_into().unwrap_or_default(), last_processed_at: Instant::now(), queued_blocks: Default::default(), rpc_client, @@ -61,7 +64,7 @@ impl BlockProcessor { block_modules: vec![], tx_modules: vec![], msg_modules: vec![], - } + }) } async fn process_block(&mut self, block: BlockToProcess) -> Result<(), ScraperError> { @@ -145,10 +148,25 @@ impl BlockProcessor { self.last_processed_height + 1..current_height + 1 }; + self.request_block_range(request_range).await?; + + Ok(()) + } + + // technically we're not mutating self here, + // but we need it to help the compiler figure out the future is `Send` + async fn request_block_range(&mut self, request_range: Range) -> Result<(), ScraperError> { + debug!("requesting missing blocks: {request_range:?}"); + let request_range = if request_range.len() > MAX_RANGE_SIZE { + warn!("the request range has more than {MAX_RANGE_SIZE} values. reducing it."); + request_range.start..request_range.start + MAX_RANGE_SIZE as u32 + } else { + request_range + }; + self.block_requester .send(BlockRequest::Range(request_range)) .await?; - Ok(()) } @@ -156,16 +174,24 @@ impl BlockProcessor { let height = block.height; if self.last_processed_height == 0 { - // TODO: load it from storage instead and make sure storage always has some data in there + // 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() { - error!("duplicate queued up block for height {height}") + warn!("we have already queued up block for height {height}"); } return; - } else if let Err(err) = self.process_block(block).await { + } + + if let Err(err) = self.process_block(block).await { error!("failed to process block at height {height}: {err}"); return; } @@ -180,6 +206,19 @@ impl BlockProcessor { } } + // 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> { + let latest_block = self.rpc_client.current_block_height().await? as u32; + if latest_block > self.last_processed_height && self.last_processed_height != 0 { + let request_range = self.last_processed_height + 1..latest_block + 1; + info!("we need to request {request_range:?} to resync"); + self.request_block_range(request_range).await?; + } + + Ok(()) + } + pub(crate) async fn run(&mut self) { info!("starting processing loop"); @@ -190,6 +229,12 @@ impl BlockProcessor { 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() => { diff --git a/common/nyxd-scraper/src/rpc_client.rs b/common/nyxd-scraper/src/rpc_client.rs index d526a07770..d2c141f711 100644 --- a/common/nyxd-scraper/src/rpc_client.rs +++ b/common/nyxd-scraper/src/rpc_client.rs @@ -149,7 +149,7 @@ impl RpcClient { #[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 for {tx_hash}"); + debug!("getting tx results"); self.inner .tx(tx_hash, false) diff --git a/common/nyxd-scraper/src/scraper/mod.rs b/common/nyxd-scraper/src/scraper/mod.rs index 8ba4e40398..656cd6c921 100644 --- a/common/nyxd-scraper/src/scraper/mod.rs +++ b/common/nyxd-scraper/src/scraper/mod.rs @@ -58,7 +58,8 @@ impl NyxdScraperBuilder { 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); @@ -167,7 +168,8 @@ impl NyxdScraper { req_tx, self.storage.clone(), rpc_client, - ); + ) + .await?; let mut chain_subscriber = ChainSubscriber::new( &self.config.websocket_url, self.cancel_token.clone(), diff --git a/common/nyxd-scraper/src/storage/manager.rs b/common/nyxd-scraper/src/storage/manager.rs index 014aeb350f..67ea50d610 100644 --- a/common/nyxd-scraper/src/storage/manager.rs +++ b/common/nyxd-scraper/src/storage/manager.rs @@ -12,6 +12,19 @@ pub(crate) struct StorageManager { } 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_first_block_height_after( &self, time: OffsetDateTime, @@ -307,15 +320,9 @@ where { trace!("update_last_processed"); - sqlx::query!( - r#" - UPDATE metadata - SET last_processed_height = ? - "#, - height - ) - .execute(executor) - .await?; + sqlx::query!("UPDATE metadata SET last_processed_height = ?", height) + .execute(executor) + .await?; Ok(()) } diff --git a/common/nyxd-scraper/src/storage/mod.rs b/common/nyxd-scraper/src/storage/mod.rs index 03ade02a93..ce491e5663 100644 --- a/common/nyxd-scraper/src/storage/mod.rs +++ b/common/nyxd-scraper/src/storage/mod.rs @@ -59,9 +59,10 @@ impl ScraperStorage { info!("Database migration finished!"); - let storage = ScraperStorage { - manager: StorageManager { connection_pool }, - }; + let manager = StorageManager { connection_pool }; + manager.set_initial_metadata().await?; + + let storage = ScraperStorage { manager }; Ok(storage) } From 50cc8bd0bf2c5139a9006da608bb261dd1e6ed2c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Wed, 13 Dec 2023 11:03:41 +0000 Subject: [PATCH 4/5] improved startup sync --- .../src/block_processor/helpers.rs | 65 +++++++++++++++ .../nyxd-scraper/src/block_processor/mod.rs | 79 ++++++++++++++++--- common/nyxd-scraper/src/storage/mod.rs | 2 +- 3 files changed, 136 insertions(+), 10 deletions(-) create mode 100644 common/nyxd-scraper/src/block_processor/helpers.rs diff --git a/common/nyxd-scraper/src/block_processor/helpers.rs b/common/nyxd-scraper/src/block_processor/helpers.rs new file mode 100644 index 0000000000..ffc8c592c5 --- /dev/null +++ b/common/nyxd-scraper/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/src/block_processor/mod.rs b/common/nyxd-scraper/src/block_processor/mod.rs index 89e51d06b1..bb3e886b0e 100644 --- a/common/nyxd-scraper/src/block_processor/mod.rs +++ b/common/nyxd-scraper/src/block_processor/mod.rs @@ -1,6 +1,7 @@ // 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; @@ -8,7 +9,7 @@ use crate::modules::{BlockModule, MsgModule, TxModule}; use crate::rpc_client::RpcClient; use crate::storage::{persist_block, ScraperStorage}; use futures::StreamExt; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashSet, VecDeque}; use std::ops::{Add, Range}; use std::time::Duration; use tokio::sync::mpsc::{Sender, UnboundedReceiver}; @@ -17,16 +18,30 @@ use tokio_stream::wrappers::UnboundedReceiverStream; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; +mod helpers; 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 { cancel: CancellationToken, last_processed_height: u32, last_processed_at: Instant, + pending_sync: PendingSync, queued_blocks: BTreeMap, rpc_client: RpcClient, @@ -56,6 +71,7 @@ impl BlockProcessor { cancel, last_processed_height: last_processed.try_into().unwrap_or_default(), last_processed_at: Instant::now(), + pending_sync: Default::default(), queued_blocks: Default::default(), rpc_client, incoming: incoming.into(), @@ -139,6 +155,10 @@ impl BlockProcessor { 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() { @@ -148,22 +168,37 @@ impl BlockProcessor { self.last_processed_height + 1..current_height + 1 }; - self.request_block_range(request_range).await?; + self.request_missing_blocks(request_range).await?; Ok(()) } - // technically we're not mutating self here, - // but we need it to help the compiler figure out the future is `Send` - async fn request_block_range(&mut self, request_range: Range) -> Result<(), ScraperError> { - debug!("requesting missing blocks: {request_range:?}"); + async fn request_missing_blocks( + &mut self, + request_range: Range, + ) -> Result<(), ScraperError> { let request_range = if request_range.len() > MAX_RANGE_SIZE { - warn!("the request range has more than {MAX_RANGE_SIZE} values. reducing it."); - request_range.start..request_range.start + MAX_RANGE_SIZE as u32 + 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?; @@ -173,6 +208,8 @@ impl BlockProcessor { 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"); @@ -204,16 +241,40 @@ impl BlockProcessor { } next += 1; } + + self.try_request_pending().await; + } + + 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()); + let latest_block = self.rpc_client.current_block_height().await? as u32; if latest_block > self.last_processed_height && self.last_processed_height != 0 { let request_range = self.last_processed_height + 1..latest_block + 1; info!("we need to request {request_range:?} to resync"); - self.request_block_range(request_range).await?; + self.request_missing_blocks(request_range).await?; } Ok(()) diff --git a/common/nyxd-scraper/src/storage/mod.rs b/common/nyxd-scraper/src/storage/mod.rs index ce491e5663..1889752a97 100644 --- a/common/nyxd-scraper/src/storage/mod.rs +++ b/common/nyxd-scraper/src/storage/mod.rs @@ -69,7 +69,7 @@ impl ScraperStorage { #[instrument(skip_all)] pub async fn begin_processing_tx(&self) -> Result { - info!("starting storage tx"); + debug!("starting storage tx"); self.manager .connection_pool .begin() From b5926def858effd5a3f1a98a67c034eb64ec5e60 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Mon, 18 Dec 2023 18:25:15 +0000 Subject: [PATCH 5/5] fixed rebasing artifact --- Cargo.toml | 1 - 1 file changed, 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index d3f1599514..bbd1f04c64 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -204,7 +204,6 @@ bip32 = "0.5.1" cosmrs = "=0.15.0" tendermint = "0.34" # same version as used by cosmrs tendermint-rpc = "0.34" # same version as used by cosmrs -tendermint = "0.34" # same version as used by cosmrs prost = "0.12" # wasm-related dependencies