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

state for processing bank msg

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2025-03-12 10:45:41 +00:00
parent 23755e462e
commit 0af371e844
10 changed files with 337 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/client-libs/validator-client/src/nyxd/mod.rs
+1
View File
@@ -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},
common/nyxd-scraper/src/block_processor/mod.rs
+5 -3
View File
@@ -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?
}
}
}
}
common/nyxd-scraper/src/error.rs
+9
View File
@@ -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 },
common/nyxd-scraper/src/modules/msg_module.rs
+2
View File
@@ -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,
nyx-chain-watcher/src/chain_scraper/mod.rs
+93 -62
View File
@@ -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<NyxdScraper> {
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<Coin> {
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, ScraperError> {
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 {
<MsgSend as Msg>::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;
}
}
nyx-chain-watcher/src/cli/commands/run/mod.rs
+5 -2
View File
@@ -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<anyhow::Result<CancellationToken>> = 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?;
{
nyx-chain-watcher/src/http/api/status.rs
+69 -4
View File
@@ -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<AppState> {
.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<BankScraperModuleState>,
) -> Json<BankModuleStatusResponse> {
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(),
})
}
nyx-chain-watcher/src/http/models.rs
+34
View File
@@ -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<BankMsgDetails>,
pub(crate) last_seen_watched_bank_msgs: Vec<BankMsgDetails>,
pub(crate) last_rejected_watched_bank_msgs: Vec<BankMsgRejection>,
}
#[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<String>,
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,
}
}
nyx-chain-watcher/src/http/server.rs
+3 -1
View File
@@ -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<HttpServer> {
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);
nyx-chain-watcher/src/http/state.rs
+116 -1
View File
@@ -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<PaymentWatcher>,
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<RwLock<BankScraperModuleStateInner>>,
}
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<S: Into<String>>(
&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<BankMsgDetails>,
pub(crate) last_seen_watched_bank_msgs: RingBuffer<BankMsgDetails>,
pub(crate) last_rejected_watched_bank_msgs: RingBuffer<BankMsgRejection>,
}
#[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<String>,
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<AppState> for PaymentListenerState {
fn from_ref(input: &AppState) -> Self {
input.payment_listener_state.clone()
@@ -275,3 +384,9 @@ impl FromRef<AppState> for PriceScraperState {
input.price_scraper_state.clone()
}
}
impl FromRef<AppState> for BankScraperModuleState {
fn from_ref(input: &AppState) -> Self {
input.bank_scraper_module_state.clone()
}
}