diff --git a/common/client-libs/validator-client/src/nyxd/mod.rs b/common/client-libs/validator-client/src/nyxd/mod.rs index 458ba45c27..03746e305e 100644 --- a/common/client-libs/validator-client/src/nyxd/mod.rs +++ b/common/client-libs/validator-client/src/nyxd/mod.rs @@ -62,6 +62,7 @@ pub use cw3; pub use cw4; pub use cw_controllers; pub use fee::{gas_price::GasPrice, GasAdjustable, GasAdjustment}; +pub use prost::Name; pub use tendermint_rpc::endpoint::block::Response as BlockResponse; pub use tendermint_rpc::{ endpoint::{tx::Response as TxResponse, validators::Response as ValidatorResponse}, diff --git a/common/nyxd-scraper/src/block_processor/mod.rs b/common/nyxd-scraper/src/block_processor/mod.rs index e75d62b7df..d4007de96b 100644 --- a/common/nyxd-scraper/src/block_processor/mod.rs +++ b/common/nyxd-scraper/src/block_processor/mod.rs @@ -182,9 +182,11 @@ impl BlockProcessor { // 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? + if msg.type_url == msg_module.type_url() { + msg_module + .handle_msg(index, msg, &block_tx, &mut tx) + .await? + } } } } diff --git a/common/nyxd-scraper/src/error.rs b/common/nyxd-scraper/src/error.rs index 6e413983bd..3337ee969c 100644 --- a/common/nyxd-scraper/src/error.rs +++ b/common/nyxd-scraper/src/error.rs @@ -83,6 +83,15 @@ pub enum ScraperError { source: cosmrs::ErrorReport, }, + #[error("could not parse msg in tx {hash} at index {index} into {type_url}: {source}")] + MsgParseFailure { + hash: Hash, + index: usize, + type_url: String, + #[source] + source: cosmrs::ErrorReport, + }, + #[error("received an invalid chain subscription event of kind {kind} while we were waiting for new block data (query: '{query}')")] InvalidSubscriptionEvent { query: String, kind: String }, diff --git a/common/nyxd-scraper/src/modules/msg_module.rs b/common/nyxd-scraper/src/modules/msg_module.rs index df64761f7d..1d195bee14 100644 --- a/common/nyxd-scraper/src/modules/msg_module.rs +++ b/common/nyxd-scraper/src/modules/msg_module.rs @@ -9,6 +9,8 @@ use cosmrs::Any; #[async_trait] pub trait MsgModule { + fn type_url(&self) -> String; + async fn handle_msg( &mut self, index: usize, diff --git a/nyx-chain-watcher/src/chain_scraper/mod.rs b/nyx-chain-watcher/src/chain_scraper/mod.rs index eba1865c6e..b3be54d046 100644 --- a/nyx-chain-watcher/src/chain_scraper/mod.rs +++ b/nyx-chain-watcher/src/chain_scraper/mod.rs @@ -3,18 +3,21 @@ use crate::env::vars::{ NYXD_SCRAPER_START_HEIGHT, NYXD_SCRAPER_UNSAFE_NUKE_DB, NYXD_SCRAPER_USE_BEST_EFFORT_START_HEIGHT, }; +use crate::http::state::BankScraperModuleState; use async_trait::async_trait; +use nym_validator_client::nyxd::{Any, Coin, CosmosCoin, Hash, Msg, MsgSend, Name}; use nyxd_scraper::{ - error::ScraperError, storage::StorageTransaction, NyxdScraper, ParsedTransactionResponse, - PruningOptions, TxModule, + error::ScraperError, storage::StorageTransaction, MsgModule, NyxdScraper, + ParsedTransactionResponse, PruningOptions, }; use sqlx::SqlitePool; use std::fs; -use tracing::{error, info, warn}; +use tracing::{info, warn}; pub(crate) async fn run_chain_scraper( config: &crate::config::Config, db_pool: SqlitePool, + shared_state: BankScraperModuleState, ) -> anyhow::Result { let websocket_url = std::env::var("NYXD_WS").expect("NYXD_WS not defined"); @@ -58,9 +61,10 @@ pub(crate) async fn run_chain_scraper( use_best_effort_start_height, }, }) - .with_tx_module(EventScraperModule::new( + .with_msg_module(BankScraperModule::new( db_pool, config.payment_watcher_config.clone(), + shared_state, )); let instance = scraper.build_and_start().await?; @@ -71,16 +75,22 @@ pub(crate) async fn run_chain_scraper( Ok(instance) } -pub struct EventScraperModule { +pub struct BankScraperModule { db_pool: SqlitePool, payment_config: PaymentWatchersConfig, + shared_state: BankScraperModuleState, } -impl EventScraperModule { - pub fn new(db_pool: SqlitePool, payment_config: PaymentWatchersConfig) -> Self { +impl BankScraperModule { + pub fn new( + db_pool: SqlitePool, + payment_config: PaymentWatchersConfig, + shared_state: BankScraperModuleState, + ) -> Self { Self { db_pool, payment_config, + shared_state, } } @@ -108,23 +118,47 @@ impl EventScraperModule { amount, memo ) - .execute(&self.db_pool) - .await?; + .execute(&self.db_pool) + .await?; Ok(()) } -} + fn get_unym_coin(&self, coins: &[CosmosCoin]) -> Option { + coins + .into_iter() + .find(|coin| coin.denom.as_ref() == "unym") + .map(|c| c.clone().into()) + } + + // TODO: ideally this should be done by the scraper itself + fn recover_bank_msg( + &self, + tx_hash: Hash, + index: usize, + msg: &Any, + ) -> Result { + MsgSend::from_any(msg).map_err(|source| ScraperError::MsgParseFailure { + hash: tx_hash, + index, + type_url: self.type_url(), + source, + }) + } +} #[async_trait] -impl TxModule for EventScraperModule { - async fn handle_tx( +impl MsgModule for BankScraperModule { + fn type_url(&self) -> String { + ::Proto::type_url() + } + + async fn handle_msg( &mut self, + index: usize, + msg: &Any, tx: &ParsedTransactionResponse, - _: &mut StorageTransaction, + _storage_tx: &mut StorageTransaction, ) -> Result<(), ScraperError> { - let events = &tx.tx_result.events; - let height = tx.height.value() as i64; - let tx_hash = tx.hash.to_string(); let memo = tx.tx.body.memo.clone(); // Don't process failed transactions @@ -132,56 +166,53 @@ impl TxModule for EventScraperModule { return Ok(()); } - if tx.tx.body.messages.len() > 1 { - error!( - "this transaction has more than 1 message in it - payment information will be lost" - ); - } + let msg = self.recover_bank_msg(tx.hash, index, &msg)?; - // Process each event - for event in events { - // Only process transfer events - if event.kind == "transfer" { - let mut recipient = None; - let mut sender = None; - let mut amount = None; - // TODO: get message index from event - let message_index = 0; + // Check if any watcher is watching this recipient + let is_watched = self + .payment_config + .is_being_watched(msg.to_address.as_ref()); - // Extract transfer event attributes - for attr in &event.attributes { - if let (Ok(key), Ok(value)) = (attr.key_str(), attr.value_str()) { - match key { - "recipient" => recipient = Some(value.to_string()), - "sender" => sender = Some(value.to_string()), - "amount" => amount = Some(value.to_string()), - _ => continue, - } - } - } + self.shared_state + .new_bank_msg(tx, index, &msg, is_watched) + .await; - // If we have all required fields, check if recipient is watched and store - if let (Some(recipient), Some(sender), Some(amount)) = (recipient, sender, amount) { - // Check if any watcher is watching this recipient - let is_watched = self.payment_config.is_being_watched(&recipient); + if is_watched { + let Some(unym_coin) = self.get_unym_coin(&msg.amount) else { + let warn = format!( + "{} sent {:?} instead of unym!", + msg.from_address, msg.amount + ); + warn!("{warn}"); + self.shared_state + .new_rejection(tx.hash.to_string(), tx.height.value(), index as u32, warn) + .await; - if is_watched { - if let Err(e) = self - .store_transfer_event( - &tx_hash, - height, - message_index, - sender, - recipient, - amount, - Some(memo.clone()), - ) - .await - { - warn!("Failed to store transfer event: {}", e); - } - } - } + // we don't want to fail the whole processing - this is not a failure in that sense! + return Ok(()); + }; + + if let Err(err) = self + .store_transfer_event( + &tx.hash.to_string(), + tx.height.value() as i64, + index as i64, + msg.from_address.to_string(), + msg.to_address.to_string(), + unym_coin.to_string(), + Some(memo.clone()), + ) + .await + { + warn!("Failed to store transfer event: {err}"); + self.shared_state + .new_rejection( + tx.hash.to_string(), + tx.height.value(), + index as u32, + format!("storage failure: {err}"), + ) + .await; } } diff --git a/nyx-chain-watcher/src/cli/commands/run/mod.rs b/nyx-chain-watcher/src/cli/commands/run/mod.rs index a8bacd6268..84d9f10f3b 100644 --- a/nyx-chain-watcher/src/cli/commands/run/mod.rs +++ b/nyx-chain-watcher/src/cli/commands/run/mod.rs @@ -14,7 +14,7 @@ mod config; use crate::chain_scraper::run_chain_scraper; use crate::db::DbPool; -use crate::http::state::{PaymentListenerState, PriceScraperState}; +use crate::http::state::{BankScraperModuleState, PaymentListenerState, PriceScraperState}; use crate::payment_listener::PaymentListener; use crate::price_scraper::PriceScraper; use crate::{db, http}; @@ -147,15 +147,17 @@ pub(crate) async fn execute(args: Args, http_port: u16) -> Result<(), NyxChainWa // construct shared state let payment_listener_shared_state = PaymentListenerState::new(); let price_scraper_shared_state = PriceScraperState::new(); + let bank_scraper_module_shared_state = BankScraperModuleState::new(); // spawn all the tasks // 1. chain scraper (note: this doesn't really spawn the full scraper on this task, but we don't want to be blocking waiting for its startup) let scraper_token_handle: JoinHandle> = tokio::spawn({ let config = config.clone(); + let shared_state = bank_scraper_module_shared_state.clone(); async move { // this only blocks until startup sync is done; it then runs on its own set of tasks - let scraper = run_chain_scraper(&config, scraper_pool).await?; + let scraper = run_chain_scraper(&config, scraper_pool, shared_state).await?; Ok(scraper.cancel_token()) } }); @@ -200,6 +202,7 @@ pub(crate) async fn execute(args: Args, http_port: u16) -> Result<(), NyxChainWa http_port, payment_listener_shared_state, price_scraper_shared_state, + bank_scraper_module_shared_state, ) .await?; { diff --git a/nyx-chain-watcher/src/http/api/status.rs b/nyx-chain-watcher/src/http/api/status.rs index 9b48e183fb..df6ec70a91 100644 --- a/nyx-chain-watcher/src/http/api/status.rs +++ b/nyx-chain-watcher/src/http/api/status.rs @@ -2,11 +2,14 @@ // SPDX-License-Identifier: GPL-3.0-only use crate::http::models::status::{ - ActivePaymentWatchersResponse, ApiStatus, HealthResponse, PaymentListenerFailureDetails, - PaymentListenerStatusResponse, PriceScraperLastError, PriceScraperLastSuccess, - PriceScraperStatusResponse, ProcessedPayment, WatcherFailureDetails, WatcherState, + ActivePaymentWatchersResponse, ApiStatus, BankModuleStatusResponse, BankMsgDetails, + BankMsgRejection, HealthResponse, PaymentListenerFailureDetails, PaymentListenerStatusResponse, + PriceScraperLastError, PriceScraperLastSuccess, PriceScraperStatusResponse, ProcessedPayment, + WatcherFailureDetails, WatcherState, +}; +use crate::http::state::{ + AppState, BankScraperModuleState, PaymentListenerState, PriceScraperState, StatusState, }; -use crate::http::state::{AppState, PaymentListenerState, PriceScraperState, StatusState}; use axum::extract::State; use axum::routing::get; use axum::{Json, Router}; @@ -20,6 +23,7 @@ pub(crate) fn routes() -> Router { .route("/active-payment-watchers", get(active_payment_watchers)) .route("/payment-listener", get(payment_listener_status)) .route("/price-scraper", get(price_scraper_status)) + .route("/bank-module-scraper", get(bank_module_status)) } #[utoipa::path( @@ -161,3 +165,64 @@ pub(crate) async fn price_scraper_status( }), }) } + +#[utoipa::path( + tag = "Status", + get, + path = "/bank-module-scraper", + context_path = "/v1/status", + responses( + (status = 200, body = BankModuleStatusResponse) + ) +)] +pub(crate) async fn bank_module_status( + State(state): State, +) -> Json { + let guard = state.inner.read().await; + Json(BankModuleStatusResponse { + processed_bank_msgs_since_startup: guard.processed_bank_msgs_since_startup, + processed_bank_msgs_to_watched_addresses_since_startup: guard + .processed_bank_msgs_to_watched_addresses_since_startup, + rejected_bank_msgs_to_watched_addresses_since_startup: guard + .rejected_bank_msgs_to_watched_addresses_since_startup, + last_seen_bank_msgs: guard + .last_seen_bank_msgs + .iter() + .map(|msg| BankMsgDetails { + processed_at: msg.processed_at, + tx_hash: msg.tx_hash.clone(), + height: msg.height, + index: msg.index, + from: msg.from.clone(), + to: msg.to.clone(), + amount: msg.amount.clone(), + memo: msg.memo.clone(), + }) + .collect(), + last_seen_watched_bank_msgs: guard + .last_seen_watched_bank_msgs + .iter() + .map(|msg| BankMsgDetails { + processed_at: msg.processed_at, + tx_hash: msg.tx_hash.clone(), + height: msg.height, + index: msg.index, + from: msg.from.clone(), + to: msg.to.clone(), + amount: msg.amount.clone(), + memo: msg.memo.clone(), + }) + .collect(), + last_rejected_watched_bank_msgs: guard + .last_rejected_watched_bank_msgs + .iter() + .map(|r| BankMsgRejection { + rejected_at: r.rejected_at, + tx_hash: r.tx_hash.clone(), + height: r.height, + index: r.index, + error: r.error.clone(), + }) + .collect(), + }) +} diff --git a/nyx-chain-watcher/src/http/models.rs b/nyx-chain-watcher/src/http/models.rs index 4c7173dbdf..c433ea9702 100644 --- a/nyx-chain-watcher/src/http/models.rs +++ b/nyx-chain-watcher/src/http/models.rs @@ -125,4 +125,38 @@ pub mod status { pub(crate) timestamp: OffsetDateTime, pub(crate) message: String, } + + #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] + pub(crate) struct BankModuleStatusResponse { + pub(crate) processed_bank_msgs_since_startup: usize, + pub(crate) processed_bank_msgs_to_watched_addresses_since_startup: usize, + pub(crate) rejected_bank_msgs_to_watched_addresses_since_startup: usize, + + pub(crate) last_seen_bank_msgs: Vec, + pub(crate) last_seen_watched_bank_msgs: Vec, + pub(crate) last_rejected_watched_bank_msgs: Vec, + } + + #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] + pub(crate) struct BankMsgDetails { + #[serde(with = "time::serde::rfc3339")] + pub(crate) processed_at: OffsetDateTime, + pub(crate) tx_hash: String, + pub(crate) height: u64, + pub(crate) index: u32, + pub(crate) from: String, + pub(crate) to: String, + pub(crate) amount: Vec, + pub(crate) memo: String, + } + + #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] + pub(crate) struct BankMsgRejection { + #[serde(with = "time::serde::rfc3339")] + pub(crate) rejected_at: OffsetDateTime, + pub(crate) tx_hash: String, + pub(crate) height: u64, + pub(crate) index: u32, + pub(crate) error: String, + } } diff --git a/nyx-chain-watcher/src/http/server.rs b/nyx-chain-watcher/src/http/server.rs index 35e6551788..96e7d47f95 100644 --- a/nyx-chain-watcher/src/http/server.rs +++ b/nyx-chain-watcher/src/http/server.rs @@ -4,7 +4,7 @@ use tokio::net::TcpListener; use tokio_util::sync::WaitForCancellationFutureOwned; use crate::config::Config; -use crate::http::state::{PaymentListenerState, PriceScraperState}; +use crate::http::state::{BankScraperModuleState, PaymentListenerState, PriceScraperState}; use crate::{ db::DbPool, http::{api::RouterBuilder, state::AppState}, @@ -16,6 +16,7 @@ pub(crate) async fn build_http_api( http_port: u16, payment_listener_state: PaymentListenerState, price_scraper_state: PriceScraperState, + bank_scraper_module_shared_state: BankScraperModuleState, ) -> anyhow::Result { let router_builder = RouterBuilder::with_default_routes(); @@ -29,6 +30,7 @@ pub(crate) async fn build_http_api( .collect(), payment_listener_state, price_scraper_state, + bank_scraper_module_shared_state, ); let router = router_builder.with_state(state); diff --git a/nyx-chain-watcher/src/http/state.rs b/nyx-chain-watcher/src/http/state.rs index 75d0568830..3f0d813f44 100644 --- a/nyx-chain-watcher/src/http/state.rs +++ b/nyx-chain-watcher/src/http/state.rs @@ -6,7 +6,8 @@ use crate::models::WebhookPayload; use axum::extract::FromRef; use nym_bin_common::bin_info; use nym_bin_common::build_information::BinaryBuildInformation; -use nym_validator_client::nyxd::Coin; +use nym_validator_client::nyxd::{Coin, MsgSend}; +use nyxd_scraper::ParsedTransactionResponse; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::ops::Deref; @@ -22,6 +23,7 @@ pub(crate) struct AppState { pub(crate) payment_listener_state: PaymentListenerState, pub(crate) status_state: StatusState, pub(crate) price_scraper_state: PriceScraperState, + pub(crate) bank_scraper_module_state: BankScraperModuleState, } impl AppState { @@ -30,6 +32,7 @@ impl AppState { registered_payment_watchers: Vec, payment_listener_state: PaymentListenerState, price_scraper_state: PriceScraperState, + bank_scraper_module_state: BankScraperModuleState, ) -> Self { Self { db_pool, @@ -37,6 +40,7 @@ impl AppState { payment_listener_state, status_state: Default::default(), price_scraper_state, + bank_scraper_module_state, } } @@ -259,6 +263,111 @@ impl WatcherFailureDetails { } } +#[derive(Debug, Clone)] +pub(crate) struct BankScraperModuleState { + pub(crate) inner: Arc>, +} + +impl BankScraperModuleState { + // TODO: make those configurable + const MAX_LAST_BANK_MSGS: usize = 20; + const MAX_LAST_WATCHED_BANK_MSGS: usize = 10; + const MAX_LAST_REJECTED_BANK_MSGS: usize = 25; + + pub(crate) fn new() -> Self { + BankScraperModuleState { + inner: Arc::new(RwLock::new(BankScraperModuleStateInner { + processed_bank_msgs_since_startup: 0, + processed_bank_msgs_to_watched_addresses_since_startup: 0, + rejected_bank_msgs_to_watched_addresses_since_startup: 0, + last_seen_bank_msgs: RingBuffer::new(Self::MAX_LAST_BANK_MSGS), + last_seen_watched_bank_msgs: RingBuffer::new(Self::MAX_LAST_WATCHED_BANK_MSGS), + last_rejected_watched_bank_msgs: RingBuffer::new(Self::MAX_LAST_REJECTED_BANK_MSGS), + })), + } + } + + pub(crate) async fn new_bank_msg( + &self, + tx: &ParsedTransactionResponse, + index: usize, + msg: &MsgSend, + is_watched: bool, + ) { + let mut guard = self.inner.write().await; + guard.processed_bank_msgs_since_startup += 1; + + let details = BankMsgDetails { + processed_at: OffsetDateTime::now_utc(), + tx_hash: tx.hash.to_string(), + height: tx.height.value(), + index: index as u32, + from: msg.from_address.to_string(), + to: msg.to_address.to_string(), + amount: msg.amount.iter().map(|c| c.to_string()).collect(), + memo: tx.tx.body.memo.clone(), + }; + guard.last_seen_bank_msgs.push(details.clone()); + + if is_watched { + guard.processed_bank_msgs_to_watched_addresses_since_startup += 1; + guard.last_seen_watched_bank_msgs.push(details.clone()); + } + } + + pub(crate) async fn new_rejection>( + &self, + tx_hash: String, + height: u64, + index: u32, + error: S, + ) { + self.inner + .write() + .await + .last_rejected_watched_bank_msgs + .push(BankMsgRejection { + rejected_at: OffsetDateTime::now_utc(), + tx_hash, + height, + index, + error: error.into(), + }) + } +} + +#[derive(Debug)] +pub(crate) struct BankScraperModuleStateInner { + pub(crate) processed_bank_msgs_since_startup: usize, + pub(crate) processed_bank_msgs_to_watched_addresses_since_startup: usize, + pub(crate) rejected_bank_msgs_to_watched_addresses_since_startup: usize, + + pub(crate) last_seen_bank_msgs: RingBuffer, + pub(crate) last_seen_watched_bank_msgs: RingBuffer, + pub(crate) last_rejected_watched_bank_msgs: RingBuffer, +} + +#[derive(Debug, Clone)] +pub(crate) struct BankMsgDetails { + pub(crate) processed_at: OffsetDateTime, + pub(crate) tx_hash: String, + pub(crate) height: u64, + pub(crate) index: u32, + pub(crate) from: String, + pub(crate) to: String, + pub(crate) amount: Vec, + pub(crate) memo: String, +} + +#[derive(Debug)] +pub(crate) struct BankMsgRejection { + pub(crate) rejected_at: OffsetDateTime, + pub(crate) tx_hash: String, + pub(crate) height: u64, + pub(crate) index: u32, + pub(crate) error: String, +} + impl FromRef for PaymentListenerState { fn from_ref(input: &AppState) -> Self { input.payment_listener_state.clone() @@ -275,3 +384,9 @@ impl FromRef for PriceScraperState { input.price_scraper_state.clone() } } + +impl FromRef for BankScraperModuleState { + fn from_ref(input: &AppState) -> Self { + input.bank_scraper_module_state.clone() + } +}