diff --git a/Cargo.toml b/Cargo.toml index 52b8a5f382..bbd1f04c64 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,8 +202,8 @@ cw-controllers = { version = "=1.1.0" } # cosmrs-related bip32 = "0.5.1" cosmrs = "=0.15.0" -tendermint-rpc = "0.34" # same version as used by cosmrs tendermint = "0.34" # same version as used by cosmrs +tendermint-rpc = "0.34" # same version as used by cosmrs prost = "0.12" # wasm-related dependencies 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/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 new file mode 100644 index 0000000000..bb3e886b0e --- /dev/null +++ b/common/nyxd-scraper/src/block_processor/mod.rs @@ -0,0 +1,323 @@ +// 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 futures::StreamExt; +use std::collections::{BTreeMap, HashSet, VecDeque}; +use std::ops::{Add, Range}; +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}; + +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, + 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( + cancel: CancellationToken, + incoming: UnboundedReceiver, + block_requester: Sender, + storage: ScraperStorage, + rpc_client: RpcClient, + ) -> Result { + let last_processed = storage.get_last_processed_height().await?; + + Ok(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(), + 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(()); + } + + 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(()) + } + + 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; + } + + 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_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/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..d2c141f711 --- /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"); + + 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..656cd6c921 --- /dev/null +++ b/common/nyxd-scraper/src/scraper/mod.rs @@ -0,0 +1,197 @@ +// 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, + ) + .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 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, + ) + .await?; + 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..67ea50d610 --- /dev/null +++ b/common/nyxd-scraper/src/storage/manager.rs @@ -0,0 +1,328 @@ +// 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 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, + ) -> 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!("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 new file mode 100644 index 0000000000..1889752a97 --- /dev/null +++ b/common/nyxd-scraper/src/storage/mod.rs @@ -0,0 +1,299 @@ +// 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 manager = StorageManager { connection_pool }; + manager.set_initial_metadata().await?; + + let storage = ScraperStorage { manager }; + + Ok(storage) + } + + #[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 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"] }