diff --git a/Cargo.toml b/Cargo.toml index 3147f66..040d079 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,7 +33,6 @@ serde-big-array = "0.5.1" futures = "0.3.31" futures-util = "0.3.31" base64 = "0.22.1" -bs58 = "0.5.1" rand = "0.9.0" bincode = "1.3.3" anyhow = "1.0.90" @@ -64,3 +63,5 @@ borsh-derive = "1.5.5" indicatif = "0.18.0" maplit = "1.0.2" env_logger = "0.11.8" +crossbeam = "0.8.4" +crossbeam-queue = "0.3.12" diff --git a/examples/parse_tx_events.rs b/examples/parse_tx_events.rs.bak similarity index 53% rename from examples/parse_tx_events.rs rename to examples/parse_tx_events.rs.bak index cbb5bb6..4ce1b6b 100644 --- a/examples/parse_tx_events.rs +++ b/examples/parse_tx_events.rs.bak @@ -1,8 +1,13 @@ use anyhow::Result; use solana_sdk::commitment_config::CommitmentConfig; +use solana_sdk::message::v0::LoadedAddresses; use solana_streamer_sdk::streaming::event_parser::{ protocols::MutilEventParser, EventParser, Protocol, }; +use solana_transaction_status::{ + option_serializer::OptionSerializer, TransactionStatusMeta, TransactionWithStatusMeta, + VersionedTransactionWithStatusMeta, +}; use std::str::FromStr; use std::sync::Arc; @@ -10,7 +15,7 @@ use std::sync::Arc; #[tokio::main] async fn main() -> Result<()> { let signatures = vec![ - "42agNk1heHabNAVRzEKqQEt5adGkQzRYf9M1Q81uBJPCCHyP4cyCA1RNkgxXrtEAWeeGcytyh2TsnkBDgqnHeq4z", + "5cnxDiHzUTUutMwnTCsvnMhyL9jEQsRWimmWU1gKxpQBngDaGTkou1YbGhUJhAhmTgvu49PYMmFQbbR38wdZDxJF", ]; // Validate signature format let mut valid_signatures = Vec::new(); @@ -52,7 +57,7 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> { .get_transaction_with_config( &signature, solana_client::rpc_config::RpcTransactionConfig { - encoding: Some(UiTransactionEncoding::Binary), + encoding: Some(UiTransactionEncoding::Base64), commitment: Some(CommitmentConfig::confirmed()), max_supported_transaction_version: Some(0), }, @@ -99,21 +104,85 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> { ]; let parser: Arc = Arc::new(MutilEventParser::new(protocols, None)); let start_time = std::time::Instant::now(); + + // 从 EncodedTransaction 获取 VersionedTransaction + let versioned_tx = match transaction.transaction.transaction.decode() { + Some(tx) => tx, + None => { + println!("Failed to decode transaction"); + return Ok(()); + } + }; + + // 创建 TransactionWithStatusMeta + let tx = TransactionWithStatusMeta::Complete(VersionedTransactionWithStatusMeta { + transaction: versioned_tx, + meta: TransactionStatusMeta { + status: Ok(()), + fee: transaction.transaction.meta.as_ref().map_or(0, |m| m.fee), + pre_balances: transaction + .transaction + .meta + .as_ref() + .map_or(vec![], |m| m.pre_balances.clone()), + post_balances: transaction + .transaction + .meta + .as_ref() + .map_or(vec![], |m| m.post_balances.clone()), + inner_instructions: transaction.transaction.meta.as_ref().and_then(|m| { + if let OptionSerializer::Some(inner_instructions) = &m.inner_instructions { + // 手动将每个UiInnerInstructions转换为InnerInstructions + Some(inner_instructions.iter().map(|ui_inner| { + solana_transaction_status::InnerInstructions { + index: ui_inner.index, + instructions: ui_inner.instructions.iter().map(|ui_inst| { + solana_transaction_status::InnerInstruction { + instruction: solana_sdk::instruction::Instruction { + program_id: solana_sdk::pubkey::Pubkey::new_from_array([0; 32]), + accounts: vec![], + data: vec![], + }, + stack_height: None, + } + }).collect(), + } + }).collect()) + } else { + None + } + }), + log_messages: transaction.transaction.meta.as_ref().and_then(|m| { + if let OptionSerializer::Some(logs) = &m.log_messages { + Some(logs.clone()) + } else { + None + } + }), + pre_token_balances: None, + post_token_balances: None, + rewards: None, + loaded_addresses: LoadedAddresses::default(), + return_data: None, + compute_units_consumed: None, + cost_units: None, + }, + }); + + // TransactionWithStatusMeta let events = parser .parse_transaction( - transaction.transaction.clone(), + tx, &signature.to_string(), Some(transaction.slot), None, - 0, + chrono::Utc::now().timestamp_micros(), None, ) .await .unwrap_or_else(|_e| vec![]); - let end_time = std::time::Instant::now(); - let duration = end_time.duration_since(start_time); - println!("Parsing time: {:?}", duration); + println!("Parsing time: {:?}", start_time.elapsed()); for event in events { println!("{:?}\n", event); } diff --git a/src/streaming/common/constants.rs b/src/streaming/common/constants.rs index d4b3d18..8cc2be8 100644 --- a/src/streaming/common/constants.rs +++ b/src/streaming/common/constants.rs @@ -11,4 +11,4 @@ pub const DEFAULT_BATCH_TIMEOUT_MS: u64 = 5; // 性能监控相关常量 pub const DEFAULT_METRICS_WINDOW_SECONDS: u64 = 5; pub const DEFAULT_METRICS_PRINT_INTERVAL_SECONDS: u64 = 10; -pub const SLOW_PROCESSING_THRESHOLD_MS: f64 = 10.0; +pub const SLOW_PROCESSING_THRESHOLD_US: f64 = 500.0; diff --git a/src/streaming/common/event_processor.rs b/src/streaming/common/event_processor.rs new file mode 100644 index 0000000..e69d699 --- /dev/null +++ b/src/streaming/common/event_processor.rs @@ -0,0 +1,254 @@ +use std::sync::Arc; +use tokio::task::JoinHandle; + +use solana_sdk::pubkey::Pubkey; + +use crate::common::AnyResult; +use crate::streaming::common::{ + MetricsEventType, MetricsManager, StreamClientConfig as ClientConfig, +}; +use crate::streaming::event_parser::common::filter::EventTypeFilter; +use crate::streaming::event_parser::core::account_event_parser::AccountEventParser; +use crate::streaming::event_parser::core::common_event_parser::CommonEventParser; +use crate::streaming::event_parser::EventParser; +use crate::streaming::event_parser::{ + core::traits::UnifiedEvent, protocols::mutil::parser::MutilEventParser, Protocol, +}; +use crate::streaming::grpc::{BackpressureStrategy, BatchConfig, EventPretty}; +use crate::streaming::shred::TransactionWithSlot; +use once_cell::sync::OnceCell; + +/// 事件处理器 +pub struct EventProcessor { + pub(crate) metrics_manager: MetricsManager, + pub(crate) config: ClientConfig, + pub(crate) parser_cache: OnceCell>, + pub(crate) protocols: Vec, + pub(crate) event_type_filter: Option, + pub(crate) backpressure_strategy: BackpressureStrategy, + pub(crate) batch_config: BatchConfig, +} + +impl EventProcessor { + /// 创建新的事件处理器 + pub fn new(metrics_manager: MetricsManager, config: ClientConfig) -> Self { + Self { + metrics_manager, + config, + parser_cache: OnceCell::new(), + protocols: vec![], + event_type_filter: None, + backpressure_strategy: BackpressureStrategy::Block, + batch_config: BatchConfig::default(), + } + } + + pub fn set_protocols_and_event_type_filter( + &mut self, + protocols: Vec, + event_type_filter: Option, + backpressure_strategy: BackpressureStrategy, + batch_config: BatchConfig, + ) { + self.protocols = protocols.clone(); + self.event_type_filter = event_type_filter.clone(); + self.backpressure_strategy = backpressure_strategy; + self.batch_config = batch_config; + self.parser_cache + .get_or_init(|| Arc::new(MutilEventParser::new(protocols, event_type_filter))); + } + + pub fn get_parser(&self) -> Arc { + self.parser_cache.get().unwrap().clone() + } + + pub fn get_event_handle(&self) -> Option> { + return None; + } + + pub async fn process_grpc_event_transaction_with_metrics( + &self, + event_pretty: EventPretty, + callback: &F, + bot_wallet: Option, + ) -> AnyResult<()> + where + F: Fn(Box) + Send + Sync, + { + self.process_grpc_event_transaction(event_pretty, callback, bot_wallet).await?; + Ok(()) + } + + async fn process_grpc_event_transaction( + &self, + event_pretty: EventPretty, + callback: &F, + bot_wallet: Option, + ) -> AnyResult<()> + where + F: Fn(Box) + Send + Sync, + { + match event_pretty { + EventPretty::Account(account_pretty) => { + self.metrics_manager.add_account_process_count(); + let account_event = AccountEventParser::parse_account_event( + self.protocols.clone(), + account_pretty, + self.event_type_filter.clone(), + ); + if let Some(event) = account_event { + let processing_time_us = event.program_handle_time_consuming_us() as f64; + callback(event); + // 更新性能指标（如果启用） + self.metrics_manager.update_metrics( + MetricsEventType::Account, + 1, + processing_time_us, + ); + } + } + EventPretty::Transaction(transaction_pretty) => { + self.metrics_manager.add_tx_process_count(); + let slot = transaction_pretty.slot; + let signature = transaction_pretty.signature; + // 使用缓存获取解析器 + let parser = self.get_parser(); + let all_events = parser + .parse_transaction( + transaction_pretty.tx.clone(), + signature, + Some(slot), + transaction_pretty.block_time, + transaction_pretty.program_received_time_us, + bot_wallet, + ) + .await + .unwrap_or_else(|_e| vec![]); + + let max_time_consuming_us = all_events + .iter() + .map(|event| event.program_handle_time_consuming_us()) + .max() + .unwrap_or(0); + + // 保存事件数量用于日志记录 + let event_count = all_events.len(); + + // 批量处理事件 + if !all_events.is_empty() { + for mut event in all_events { + event.set_program_handle_time_consuming_us( + chrono::Utc::now().timestamp_micros() + - event.program_received_time_us(), + ); + callback(event); + } + } + + // 更新性能指标 + // 更新性能指标（如果启用） + self.metrics_manager.update_metrics( + MetricsEventType::Tx, + event_count as u64, + max_time_consuming_us as f64, + ); + } + EventPretty::BlockMeta(block_meta_pretty) => { + self.metrics_manager.add_block_meta_process_count(); + let block_time_ms = block_meta_pretty + .block_time + .map(|ts| ts.seconds * 1000 + ts.nanos as i64 / 1_000_000) + .unwrap_or_else(|| chrono::Utc::now().timestamp_millis()); + let block_meta_event = CommonEventParser::generate_block_meta_event( + block_meta_pretty.slot, + &block_meta_pretty.block_hash, + block_time_ms, + block_meta_pretty.program_received_time_us, + ); + let processing_time_us = block_meta_event.program_handle_time_consuming_us() as f64; + callback(block_meta_event); + // 更新性能指标（如果启用） + self.metrics_manager.update_metrics( + MetricsEventType::BlockMeta, + 1, + processing_time_us, + ); + } + } + + Ok(()) + } + + /// 即时处理单个交易 + pub async fn process_shred_transaction_immediate( + &self, + transaction_with_slot: TransactionWithSlot, + bot_wallet: Option, + callback: &F, + ) -> AnyResult<()> + where + F: Fn(Box) + Send + Sync, + { + self.metrics_manager.add_tx_process_count(); + let program_received_time_us = chrono::Utc::now().timestamp_micros(); + let slot = transaction_with_slot.slot; + let versioned_tx = transaction_with_slot.transaction; + let signature = versioned_tx.signatures[0]; + + // 获取缓存的解析器 + let parser = self.get_parser(); + + let all_events = parser + .parse_versioned_transaction( + &versioned_tx, + signature, + Some(slot), + None, + program_received_time_us, + bot_wallet, + ) + .await + .unwrap_or_else(|_e| vec![]); + + let max_time_consuming_us = all_events + .iter() + .map(|event| event.program_handle_time_consuming_us()) + .max() + .unwrap_or(0); + + // 保存事件数量用于日志记录 + let event_count = all_events.len(); + + // 即时处理事件 + for mut event in all_events { + event.set_program_handle_time_consuming_us( + chrono::Utc::now().timestamp_micros() - event.program_received_time_us(), + ); + callback(event); + } + + // 实际调用性能指标更新 + self.metrics_manager.update_metrics( + MetricsEventType::Tx, + event_count as u64, + max_time_consuming_us as f64, + ); + + Ok(()) + } +} + +// 实现 Clone trait 以支持模块间共享 +impl Clone for EventProcessor { + fn clone(&self) -> Self { + Self { + metrics_manager: self.metrics_manager.clone(), + config: self.config.clone(), + parser_cache: self.parser_cache.clone(), + protocols: self.protocols.clone(), + event_type_filter: self.event_type_filter.clone(), + backpressure_strategy: self.backpressure_strategy.clone(), + batch_config: self.batch_config.clone(), + } + } +} diff --git a/src/streaming/common/metrics.rs b/src/streaming/common/metrics.rs index 2ccdebe..6dd80e5 100644 --- a/src/streaming/common/metrics.rs +++ b/src/streaming/common/metrics.rs @@ -1,5 +1,7 @@ use std::sync::Arc; -use tokio::sync::Mutex; +use crossbeam::utils::Backoff; +use crossbeam::atomic::AtomicCell; +use std::sync::RwLock; use super::config::StreamClientConfig; use super::constants::*; @@ -31,9 +33,9 @@ impl EventMetrics { pub struct PerformanceMetrics { pub start_time: std::time::Instant, pub event_metrics: [EventMetrics; 3], // [Tx, Account, BlockMeta] - pub average_processing_time_ms: f64, - pub min_processing_time_ms: f64, - pub max_processing_time_ms: f64, + pub average_processing_time_us: f64, + pub min_processing_time_us: f64, + pub max_processing_time_us: f64, pub last_update_time: std::time::Instant, } @@ -65,9 +67,9 @@ impl PerformanceMetrics { Self { start_time: now, event_metrics: [EventMetrics::new(now), EventMetrics::new(now), EventMetrics::new(now)], - average_processing_time_ms: 0.0, - min_processing_time_ms: 0.0, - max_processing_time_ms: 0.0, + average_processing_time_us: 0.0, + min_processing_time_us: 0.0, + max_processing_time_us: 0.0, last_update_time: now, } } @@ -132,7 +134,7 @@ impl PerformanceMetrics { /// 通用性能监控管理器 pub struct MetricsManager { - metrics: Arc>, + metrics: Arc>, config: Arc, stream_name: String, } @@ -140,7 +142,7 @@ pub struct MetricsManager { impl MetricsManager { /// 创建新的性能监控管理器 pub fn new( - metrics: Arc>, + metrics: Arc>, config: Arc, stream_name: String, ) -> Self { @@ -148,14 +150,24 @@ impl MetricsManager { } /// 获取性能指标 - pub async fn get_metrics(&self) -> PerformanceMetrics { - let metrics = self.metrics.lock().await; - metrics.clone() + pub fn get_metrics(&self) -> PerformanceMetrics { + // 使用 Backoff 策略进行读取尝试 + let backoff = Backoff::new(); + loop { + match self.metrics.read() { + Ok(metrics) => return metrics.clone(), + Err(_) => { + // 如果获取读锁失败，使用指数退避策略 + backoff.snooze(); + continue; + } + } + } } /// 打印性能指标 - pub async fn print_metrics(&self) { - let metrics = self.get_metrics().await; + pub fn print_metrics(&self) { + let metrics = self.get_metrics(); let event_names = ["TX", "Account", "Block Meta"]; let event_types = [MetricsEventType::Tx, MetricsEventType::Account, MetricsEventType::BlockMeta]; @@ -189,11 +201,11 @@ impl MetricsManager { // 打印处理时间统计表格 println!("\n⏱️ Processing Time Statistics"); println!("┌─────────────────────┬─────────────┐"); - println!("│ Metric │ Value (ms) │"); + println!("│ Metric │ Value (us) │"); println!("├─────────────────────┼─────────────┤"); - println!("│ Average │ {:9.2} │", metrics.average_processing_time_ms); - println!("│ Minimum │ {:9.2} │", metrics.min_processing_time_ms); - println!("│ Maximum │ {:9.2} │", metrics.max_processing_time_ms); + println!("│ Average │ {:9.2} │", metrics.average_processing_time_us); + println!("│ Minimum │ {:9.2} │", metrics.min_processing_time_us); + println!("│ Maximum │ {:9.2} │", metrics.max_processing_time_us); println!("└─────────────────────┴─────────────┘"); println!(); } @@ -212,91 +224,118 @@ impl MetricsManager { )); loop { interval.tick().await; - metrics_manager.print_metrics().await; + metrics_manager.print_metrics(); } }); Some(handle) } /// 更新处理次数 - pub async fn add_process_count(&self, event_type: MetricsEventType) { + pub fn add_process_count(&self, event_type: MetricsEventType) { if !self.config.enable_metrics { return; } - let mut metrics = self.metrics.lock().await; - metrics.event_metrics[event_type.as_index()].process_count += 1; + + // 使用 Backoff 策略进行写入尝试 + let backoff = Backoff::new(); + loop { + match self.metrics.write() { + Ok(mut metrics) => { + metrics.event_metrics[event_type.as_index()].process_count += 1; + break; + }, + Err(_) => { + // 如果获取写锁失败，使用指数退避策略 + backoff.snooze(); + continue; + } + } + } } // 保持向后兼容的方法 - pub async fn add_tx_process_count(&self) { - self.add_process_count(MetricsEventType::Tx).await; + pub fn add_tx_process_count(&self) { + self.add_process_count(MetricsEventType::Tx); } - pub async fn add_account_process_count(&self) { - self.add_process_count(MetricsEventType::Account).await; + pub fn add_account_process_count(&self) { + self.add_process_count(MetricsEventType::Account); } - pub async fn add_block_meta_process_count(&self) { - self.add_process_count(MetricsEventType::BlockMeta).await; + pub fn add_block_meta_process_count(&self) { + self.add_process_count(MetricsEventType::BlockMeta); } /// 更新性能指标 - pub async fn update_metrics( + pub fn update_metrics( &self, event_type: MetricsEventType, events_processed: u64, - processing_time_ms: f64, + processing_time_us: f64, ) { // 检查是否启用性能监控 if !self.config.enable_metrics { return; } + + // 使用 Backoff 策略进行写入尝试 + let backoff = Backoff::new(); + loop { + match self.metrics.write() { + Ok(mut metrics) => { + let now = std::time::Instant::now(); + let index = event_type.as_index(); - let mut metrics = self.metrics.lock().await; - let now = std::time::Instant::now(); - let index = event_type.as_index(); + // 更新事件计数 + metrics.event_metrics[index].events_processed += events_processed; + metrics.event_metrics[index].events_in_window += events_processed; - // 更新事件计数 - metrics.event_metrics[index].events_processed += events_processed; - metrics.event_metrics[index].events_in_window += events_processed; + metrics.last_update_time = now; - metrics.last_update_time = now; + // 更新处理时间统计 + if processing_time_us < metrics.min_processing_time_us + || metrics.min_processing_time_us == 0.0 + { + metrics.min_processing_time_us = processing_time_us; + } + if processing_time_us > metrics.max_processing_time_us { + metrics.max_processing_time_us = processing_time_us; + } - // 更新处理时间统计 - if processing_time_ms < metrics.min_processing_time_ms - || metrics.min_processing_time_ms == 0.0 - { - metrics.min_processing_time_ms = processing_time_ms; + // 计算平均处理时间 - 使用增量更新避免重复计算 + let total_events = metrics.event_metrics[index].events_processed; + if total_events > 0 { + let total_events_f64 = total_events as f64; + let old_total = (total_events_f64 - events_processed as f64).max(0.0); + + metrics.average_processing_time_us = if old_total > 0.0 { + (metrics.average_processing_time_us * old_total + + processing_time_us * events_processed as f64) + / total_events_f64 + } else { + processing_time_us + }; + } + + // 更新时间窗口指标 + let window_duration = std::time::Duration::from_secs(DEFAULT_METRICS_WINDOW_SECONDS); + metrics.update_window_metrics(&event_type, now, window_duration); + break; + }, + Err(_) => { + // 如果获取写锁失败，使用指数退避策略 + backoff.snooze(); + continue; + } + } } - if processing_time_ms > metrics.max_processing_time_ms { - metrics.max_processing_time_ms = processing_time_ms; - } - - // 计算平均处理时间 - 使用增量更新避免重复计算 - let total_events = metrics.event_metrics[index].events_processed; - if total_events > 0 { - let total_events_f64 = total_events as f64; - let old_total = (total_events_f64 - events_processed as f64).max(0.0); - - metrics.average_processing_time_ms = if old_total > 0.0 { - (metrics.average_processing_time_ms * old_total - + processing_time_ms * events_processed as f64) - / total_events_f64 - } else { - processing_time_ms - }; - } - - // 更新时间窗口指标 - let window_duration = std::time::Duration::from_secs(DEFAULT_METRICS_WINDOW_SECONDS); - metrics.update_window_metrics(&event_type, now, window_duration); } /// 记录慢处理操作 - pub fn log_slow_processing(&self, processing_time_ms: f64, event_count: usize) { - if processing_time_ms > SLOW_PROCESSING_THRESHOLD_MS { + pub fn log_slow_processing(&self, processing_time_us: f64, event_count: usize) { + if processing_time_us > SLOW_PROCESSING_THRESHOLD_US { log::warn!( - "{} slow processing: {processing_time_ms}ms for {event_count} events", + "{} slow processing: {processing_time_us}us for {event_count} events", self.stream_name ); } diff --git a/src/streaming/common/mod.rs b/src/streaming/common/mod.rs index bbbc5e1..011bab8 100644 --- a/src/streaming/common/mod.rs +++ b/src/streaming/common/mod.rs @@ -4,6 +4,7 @@ pub mod metrics; pub mod batch; pub mod constants; pub mod subscription; +pub mod event_processor; // 重新导出主要类型 pub use config::*; @@ -11,3 +12,4 @@ pub use metrics::*; pub use batch::*; pub use constants::*; pub use subscription::*; +pub use event_processor::*; \ No newline at end of file diff --git a/src/streaming/common/subscription.rs b/src/streaming/common/subscription.rs index ca29cbb..c0abf16 100644 --- a/src/streaming/common/subscription.rs +++ b/src/streaming/common/subscription.rs @@ -3,7 +3,7 @@ use tokio::task::JoinHandle; /// Subscription handle for managing and stopping subscriptions pub struct SubscriptionHandle { stream_handle: JoinHandle<()>, - event_handle: JoinHandle<()>, + event_handle: Option>, metrics_handle: Option>, } @@ -11,7 +11,7 @@ impl SubscriptionHandle { /// Create a new subscription handle pub fn new( stream_handle: JoinHandle<()>, - event_handle: JoinHandle<()>, + event_handle: Option>, metrics_handle: Option>, ) -> Self { Self { stream_handle, event_handle, metrics_handle } @@ -20,7 +20,9 @@ impl SubscriptionHandle { /// Stop subscription and abort all related tasks pub fn stop(self) { self.stream_handle.abort(); - self.event_handle.abort(); + if let Some(handle) = self.event_handle { + handle.abort(); + } if let Some(handle) = self.metrics_handle { handle.abort(); } @@ -29,7 +31,9 @@ impl SubscriptionHandle { /// Asynchronously wait for all tasks to complete pub async fn join(self) -> Result<(), tokio::task::JoinError> { let _ = self.stream_handle.await; - let _ = self.event_handle.await; + if let Some(handle) = self.event_handle { + let _ = handle.await; + } if let Some(handle) = self.metrics_handle { let _ = handle.await; } diff --git a/src/streaming/event_parser/common/mod.rs b/src/streaming/event_parser/common/mod.rs index 3ecc161..d421559 100755 --- a/src/streaming/event_parser/common/mod.rs +++ b/src/streaming/event_parser/common/mod.rs @@ -2,6 +2,8 @@ pub mod types; pub mod utils; pub mod filter; +pub const EMPTY_ID: &str = ""; + /// 自动生成UnifiedEvent trait实现的宏 #[macro_export] macro_rules! impl_unified_event { @@ -12,6 +14,10 @@ macro_rules! impl_unified_event { &self.metadata.id } + fn clear_id(&mut self) { + self.metadata.id = $crate::streaming::event_parser::common::EMPTY_ID.to_string(); + } + fn event_type(&self) -> $crate::streaming::event_parser::common::types::EventType { self.metadata.event_type.clone() } @@ -24,16 +30,16 @@ macro_rules! impl_unified_event { self.metadata.slot } - fn program_received_time_ms(&self) -> i64 { - self.metadata.program_received_time_ms + fn program_received_time_us(&self) -> i64 { + self.metadata.program_received_time_us } - fn program_handle_time_consuming_ms(&self) -> i64 { - self.metadata.program_handle_time_consuming_ms + fn program_handle_time_consuming_us(&self) -> i64 { + self.metadata.program_handle_time_consuming_us } - fn set_program_handle_time_consuming_ms(&mut self, program_handle_time_consuming_ms: i64) { - self.metadata.program_handle_time_consuming_ms = program_handle_time_consuming_ms; + fn set_program_handle_time_consuming_us(&mut self, program_handle_time_consuming_us: i64) { + self.metadata.program_handle_time_consuming_us = program_handle_time_consuming_us; } fn as_any(&self) -> &dyn std::any::Any { @@ -56,8 +62,8 @@ macro_rules! impl_unified_event { } } - fn set_transfer_datas(&mut self, transfer_datas: Vec<$crate::streaming::event_parser::common::types::TransferData>, swap_data: Option<$crate::streaming::event_parser::common::types::SwapData>) { - self.metadata.set_transfer_datas(transfer_datas, swap_data); + fn set_swap_data(&mut self, swap_data: $crate::streaming::event_parser::common::types::SwapData) { + self.metadata.set_swap_data(swap_data); } fn index(&self) -> String { diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index 9794cc8..37e75e4 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -1,13 +1,14 @@ use borsh::{BorshDeserialize, BorshSerialize}; +use crossbeam_queue::ArrayQueue; use serde::{Deserialize, Serialize}; use solana_sdk::pubkey::Pubkey; -use solana_transaction_status::UiInstruction; +use solana_transaction_status::{InnerInstruction, UiInstruction}; use std::{ + fmt, hash::{DefaultHasher, Hash, Hasher}, str::FromStr, sync::Arc, }; -use tokio::sync::Mutex; use crate::{ match_event, @@ -30,7 +31,7 @@ const TRANSFER_DATA_POOL_SIZE: usize = 2000; /// Event metadata object pool pub struct EventMetadataPool { - pool: Arc>>, + pool: Arc>, } impl Default for EventMetadataPool { @@ -41,25 +42,22 @@ impl Default for EventMetadataPool { impl EventMetadataPool { pub fn new() -> Self { - Self { pool: Arc::new(Mutex::new(Vec::with_capacity(EVENT_METADATA_POOL_SIZE))) } + Self { pool: Arc::new(ArrayQueue::new(EVENT_METADATA_POOL_SIZE)) } } - pub async fn acquire(&self) -> Option { - let mut pool = self.pool.lock().await; - pool.pop() + pub fn acquire(&self) -> Option { + self.pool.pop() } - pub async fn release(&self, metadata: EventMetadata) { - let mut pool = self.pool.lock().await; - if pool.len() < EVENT_METADATA_POOL_SIZE { - pool.push(metadata); - } + pub fn release(&self, metadata: EventMetadata) { + // 如果队列已满，push 会失败，但不会阻塞 + let _ = self.pool.push(metadata); } } /// Transfer data object pool pub struct TransferDataPool { - pool: Arc>>, + pool: Arc>, } impl Default for TransferDataPool { @@ -70,19 +68,16 @@ impl Default for TransferDataPool { impl TransferDataPool { pub fn new() -> Self { - Self { pool: Arc::new(Mutex::new(Vec::with_capacity(TRANSFER_DATA_POOL_SIZE))) } + Self { pool: Arc::new(ArrayQueue::new(TRANSFER_DATA_POOL_SIZE)) } } - pub async fn acquire(&self) -> Option { - let mut pool = self.pool.lock().await; - pool.pop() + pub fn acquire(&self) -> Option { + self.pool.pop() } - pub async fn release(&self, transfer_data: TransferData) { - let mut pool = self.pool.lock().await; - if pool.len() < TRANSFER_DATA_POOL_SIZE { - pool.push(transfer_data); - } + pub fn release(&self, transfer_data: TransferData) { + // 如果队列已满，push 会失败，但不会阻塞 + let _ = self.pool.push(transfer_data); } } @@ -199,70 +194,69 @@ pub const ACCOUNT_EVENT_TYPES: &[EventType] = &[ ]; pub const BLOCK_EVENT_TYPES: &[EventType] = &[EventType::BlockMeta]; -impl EventType { - #[allow(clippy::inherent_to_string)] - pub fn to_string(&self) -> String { +impl fmt::Display for EventType { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { - EventType::PumpSwapBuy => "PumpSwapBuy".to_string(), - EventType::PumpSwapSell => "PumpSwapSell".to_string(), - EventType::PumpSwapCreatePool => "PumpSwapCreatePool".to_string(), - EventType::PumpSwapDeposit => "PumpSwapDeposit".to_string(), - EventType::PumpSwapWithdraw => "PumpSwapWithdraw".to_string(), - EventType::PumpFunCreateToken => "PumpFunCreateToken".to_string(), - EventType::PumpFunBuy => "PumpFunBuy".to_string(), - EventType::PumpFunSell => "PumpFunSell".to_string(), - EventType::PumpFunMigrate => "PumpFunMigrate".to_string(), - EventType::BonkBuyExactIn => "BonkBuyExactIn".to_string(), - EventType::BonkBuyExactOut => "BonkBuyExactOut".to_string(), - EventType::BonkSellExactIn => "BonkSellExactIn".to_string(), - EventType::BonkSellExactOut => "BonkSellExactOut".to_string(), - EventType::BonkInitialize => "BonkInitialize".to_string(), - EventType::BonkInitializeV2 => "BonkInitializeV2".to_string(), - EventType::BonkMigrateToAmm => "BonkMigrateToAmm".to_string(), - EventType::BonkMigrateToCpswap => "BonkMigrateToCpswap".to_string(), - EventType::AccountPumpFunBondingCurve => "AccountPumpFunBondingCurve".to_string(), - EventType::AccountPumpFunGlobal => "AccountPumpFunGlobal".to_string(), - EventType::AccountPumpSwapGlobalConfig => "AccountPumpSwapGlobalConfig".to_string(), - EventType::AccountPumpSwapPool => "AccountPumpSwapPool".to_string(), - EventType::AccountBonkPoolState => "AccountBonkPoolState".to_string(), - EventType::AccountBonkGlobalConfig => "AccountBonkGlobalConfig".to_string(), - EventType::AccountBonkPlatformConfig => "AccountBonkPlatformConfig".to_string(), - EventType::AccountBonkVestingRecord => "AccountBonkVestingRecord".to_string(), - EventType::RaydiumCpmmSwapBaseInput => "RaydiumCpmmSwapBaseInput".to_string(), - EventType::RaydiumCpmmSwapBaseOutput => "RaydiumCpmmSwapBaseOutput".to_string(), - EventType::RaydiumCpmmDeposit => "RaydiumCpmmDeposit".to_string(), - EventType::RaydiumCpmmInitialize => "RaydiumCpmmInitialize".to_string(), - EventType::RaydiumCpmmWithdraw => "RaydiumCpmmWithdraw".to_string(), - EventType::RaydiumClmmSwap => "RaydiumClmmSwap".to_string(), - EventType::RaydiumClmmSwapV2 => "RaydiumClmmSwapV2".to_string(), - EventType::RaydiumClmmClosePosition => "RaydiumClmmClosePosition".to_string(), + EventType::PumpSwapBuy => write!(f, "PumpSwapBuy"), + EventType::PumpSwapSell => write!(f, "PumpSwapSell"), + EventType::PumpSwapCreatePool => write!(f, "PumpSwapCreatePool"), + EventType::PumpSwapDeposit => write!(f, "PumpSwapDeposit"), + EventType::PumpSwapWithdraw => write!(f, "PumpSwapWithdraw"), + EventType::PumpFunCreateToken => write!(f, "PumpFunCreateToken"), + EventType::PumpFunBuy => write!(f, "PumpFunBuy"), + EventType::PumpFunSell => write!(f, "PumpFunSell"), + EventType::PumpFunMigrate => write!(f, "PumpFunMigrate"), + EventType::BonkBuyExactIn => write!(f, "BonkBuyExactIn"), + EventType::BonkBuyExactOut => write!(f, "BonkBuyExactOut"), + EventType::BonkSellExactIn => write!(f, "BonkSellExactIn"), + EventType::BonkSellExactOut => write!(f, "BonkSellExactOut"), + EventType::BonkInitialize => write!(f, "BonkInitialize"), + EventType::BonkInitializeV2 => write!(f, "BonkInitializeV2"), + EventType::BonkMigrateToAmm => write!(f, "BonkMigrateToAmm"), + EventType::BonkMigrateToCpswap => write!(f, "BonkMigrateToCpswap"), + EventType::RaydiumCpmmSwapBaseInput => write!(f, "RaydiumCpmmSwapBaseInput"), + EventType::RaydiumCpmmSwapBaseOutput => write!(f, "RaydiumCpmmSwapBaseOutput"), + EventType::RaydiumCpmmDeposit => write!(f, "RaydiumCpmmDeposit"), + EventType::RaydiumCpmmInitialize => write!(f, "RaydiumCpmmInitialize"), + EventType::RaydiumCpmmWithdraw => write!(f, "RaydiumCpmmWithdraw"), + EventType::RaydiumClmmSwap => write!(f, "RaydiumClmmSwap"), + EventType::RaydiumClmmSwapV2 => write!(f, "RaydiumClmmSwapV2"), + EventType::RaydiumClmmClosePosition => write!(f, "RaydiumClmmClosePosition"), EventType::RaydiumClmmDecreaseLiquidityV2 => { - "RaydiumClmmDecreaseLiquidityV2".to_string() + write!(f, "RaydiumClmmDecreaseLiquidityV2") } - EventType::RaydiumClmmCreatePool => "RaydiumClmmCreatePool".to_string(), + EventType::RaydiumClmmCreatePool => write!(f, "RaydiumClmmCreatePool"), EventType::RaydiumClmmIncreaseLiquidityV2 => { - "RaydiumClmmIncreaseLiquidityV2".to_string() + write!(f, "RaydiumClmmIncreaseLiquidityV2") } EventType::RaydiumClmmOpenPositionWithToken22Nft => { - "RaydiumClmmOpenPositionWithToken22Nft".to_string() + write!(f, "RaydiumClmmOpenPositionWithToken22Nft") } - EventType::RaydiumClmmOpenPositionV2 => "RaydiumClmmOpenPositionV2".to_string(), - EventType::RaydiumAmmV4SwapBaseIn => "RaydiumAmmV4SwapBaseIn".to_string(), - EventType::RaydiumAmmV4SwapBaseOut => "RaydiumAmmV4SwapBaseOut".to_string(), - EventType::RaydiumAmmV4Deposit => "RaydiumAmmV4Deposit".to_string(), - EventType::RaydiumAmmV4Initialize2 => "RaydiumAmmV4Initialize2".to_string(), - EventType::RaydiumAmmV4Withdraw => "RaydiumAmmV4Withdraw".to_string(), - EventType::RaydiumAmmV4WithdrawPnl => "RaydiumAmmV4WithdrawPnl".to_string(), - EventType::AccountRaydiumAmmV4AmmInfo => "AccountRaydiumAmmV4AmmInfo".to_string(), - EventType::AccountRaydiumClmmAmmConfig => "AccountRaydiumClmmAmmConfig".to_string(), - EventType::AccountRaydiumClmmPoolState => "AccountRaydiumClmmPoolState".to_string(), + EventType::RaydiumClmmOpenPositionV2 => write!(f, "RaydiumClmmOpenPositionV2"), + EventType::RaydiumAmmV4SwapBaseIn => write!(f, "RaydiumAmmV4SwapBaseIn"), + EventType::RaydiumAmmV4SwapBaseOut => write!(f, "RaydiumAmmV4SwapBaseOut"), + EventType::RaydiumAmmV4Deposit => write!(f, "RaydiumAmmV4Deposit"), + EventType::RaydiumAmmV4Initialize2 => write!(f, "RaydiumAmmV4Initialize2"), + EventType::RaydiumAmmV4Withdraw => write!(f, "RaydiumAmmV4Withdraw"), + EventType::RaydiumAmmV4WithdrawPnl => write!(f, "RaydiumAmmV4WithdrawPnl"), + EventType::AccountRaydiumAmmV4AmmInfo => write!(f, "AccountRaydiumAmmV4AmmInfo"), + EventType::AccountPumpSwapGlobalConfig => write!(f, "AccountPumpSwapGlobalConfig"), + EventType::AccountPumpSwapPool => write!(f, "AccountPumpSwapPool"), + EventType::AccountBonkPoolState => write!(f, "AccountBonkPoolState"), + EventType::AccountBonkGlobalConfig => write!(f, "AccountBonkGlobalConfig"), + EventType::AccountBonkPlatformConfig => write!(f, "AccountBonkPlatformConfig"), + EventType::AccountBonkVestingRecord => write!(f, "AccountBonkVestingRecord"), + EventType::AccountPumpFunBondingCurve => write!(f, "AccountPumpFunBondingCurve"), + EventType::AccountPumpFunGlobal => write!(f, "AccountPumpFunGlobal"), + EventType::AccountRaydiumClmmAmmConfig => write!(f, "AccountRaydiumClmmAmmConfig"), + EventType::AccountRaydiumClmmPoolState => write!(f, "AccountRaydiumClmmPoolState"), EventType::AccountRaydiumClmmTickArrayState => { - "AccountRaydiumClmmTickArrayState".to_string() + write!(f, "AccountRaydiumClmmTickArrayState") } - EventType::AccountRaydiumCpmmAmmConfig => "AccountRaydiumCpmmAmmConfig".to_string(), - EventType::AccountRaydiumCpmmPoolState => "AccountRaydiumCpmmPoolState".to_string(), - EventType::BlockMeta => "BlockMeta".to_string(), - EventType::Unknown => "Unknown".to_string(), + EventType::AccountRaydiumCpmmAmmConfig => write!(f, "AccountRaydiumCpmmAmmConfig"), + EventType::AccountRaydiumCpmmPoolState => write!(f, "AccountRaydiumCpmmPoolState"), + EventType::BlockMeta => write!(f, "BlockMeta"), + EventType::Unknown => write!(f, "Unknown"), } } } @@ -345,11 +339,12 @@ pub struct EventMetadata { pub slot: u64, pub block_time: i64, pub block_time_ms: i64, - pub program_received_time_ms: i64, - pub program_handle_time_consuming_ms: i64, + pub program_received_time_us: i64, + pub program_handle_time_consuming_us: i64, pub protocol: ProtocolType, pub event_type: EventType, pub program_id: Pubkey, + #[deprecated(note = "Please use swap_data instead")] pub transfer_datas: Vec, pub swap_data: Option, pub index: String, @@ -367,7 +362,7 @@ impl EventMetadata { event_type: EventType, program_id: Pubkey, index: String, - program_received_time_ms: i64, + program_received_time_us: i64, ) -> Self { Self { id, @@ -375,250 +370,196 @@ impl EventMetadata { slot, block_time, block_time_ms, - program_received_time_ms, - program_handle_time_consuming_ms: 0, + program_received_time_us, + program_handle_time_consuming_us: 0, protocol, event_type, program_id, - transfer_datas: Vec::with_capacity(4), // Pre-allocate capacity + transfer_datas: vec![], swap_data: None, index, } } pub fn set_id(&mut self, id: String) { - let _id = format!("{}-{}-{}", self.signature, self.event_type.to_string(), id); - let mut hasher = DefaultHasher::new(); - _id.hash(&mut hasher); - let hash_value = hasher.finish(); - self.id = format!("{:x}", hash_value); + self.id = format!("{}-{}-{}", self.signature, self.event_type, id); } - pub fn set_transfer_datas( - &mut self, - transfer_datas: Vec, - swap_data: Option, - ) { - self.transfer_datas = transfer_datas; - self.swap_data = swap_data; + pub fn set_swap_data(&mut self, swap_data: SwapData) { + self.swap_data = Some(swap_data); } /// Recycle EventMetadata to object pool - pub async fn recycle(self) { - EVENT_METADATA_POOL.release(self).await; + pub fn recycle(self) { + EVENT_METADATA_POOL.release(self); } } -/// Parse token transfer data from next instructions -pub fn parse_transfer_datas_from_next_instructions( - event: Box, - inner_instruction: &solana_transaction_status::UiInnerInstructions, - current_index: i8, - accounts: &[Pubkey], -) -> (Vec, Option) { - let mut transfer_datas = vec![]; - // Get the next two instructions after the current instruction - let next_instructions: Vec<&UiInstruction> = - inner_instruction.instructions.iter().skip((current_index + 1) as usize).collect(); - - let system_programs = vec![ - // Token Program +lazy_static::lazy_static! { + static ref SOL_MINT: Pubkey = Pubkey::from_str("So11111111111111111111111111111111111111111").unwrap(); + static ref SYSTEM_PROGRAMS: [Pubkey; 3] = [ Pubkey::from_str("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA").unwrap(), - // Token 2022 Program Pubkey::from_str("TokenzQdBNbLqP5VEhdkAS6EPFLC1PHnBqCXEpPxuEb").unwrap(), - // System Program Pubkey::from_str("11111111111111111111111111111111").unwrap(), ]; - for instruction in next_instructions { - if let UiInstruction::Compiled(compiled) = instruction { - if !system_programs.contains(&accounts[compiled.program_id_index as usize]) { - break; - } - if let Ok(data) = bs58::decode(compiled.data.clone()).into_vec() { - // Token Program: transferChecked - // Token 2022 Program: transferChecked - if data[0] == 12 { - let account_pubkeys: Vec = - compiled.accounts.iter().map(|a| accounts[*a as usize]).collect(); - if account_pubkeys.len() < 4 { - continue; - } - let (source, mint, destination, authority) = ( - account_pubkeys[0], - account_pubkeys[1], - account_pubkeys[2], - account_pubkeys[3], - ); - let amount = u64::from_le_bytes(data[1..9].try_into().unwrap()); - let decimals = data[9]; - let token_program = accounts[compiled.program_id_index as usize]; - transfer_datas.push(TransferData { - amount, - decimals: Some(decimals), - mint: Some(mint), - source, - destination, - authority: Some(authority), - token_program, - }); - } - // Token Program: transfer - else if data[0] == 3 { - let account_pubkeys: Vec = - compiled.accounts.iter().map(|a| accounts[*a as usize]).collect(); - if account_pubkeys.len() < 3 { - continue; - } - let (source, destination, authority) = - (account_pubkeys[0], account_pubkeys[1], account_pubkeys[2]); - let amount = u64::from_le_bytes(data[1..9].try_into().unwrap()); - let token_program = accounts[compiled.program_id_index as usize]; - transfer_datas.push(TransferData { - amount, - decimals: None, - mint: None, - source, - destination, - authority: Some(authority), - token_program, - }); - } - //System Program: transfer - else if data[0] == 2 { - let account_pubkeys: Vec = - compiled.accounts.iter().map(|a| accounts[*a as usize]).collect(); - if account_pubkeys.len() < 2 { - continue; - } - let (source, destination) = (account_pubkeys[0], account_pubkeys[1]); - let amount = u64::from_le_bytes(data[4..12].try_into().unwrap()); - let token_program = accounts[compiled.program_id_index as usize]; - transfer_datas.push(TransferData { - amount, - decimals: None, - mint: None, - source, - destination, - authority: None, - token_program, - }); - } - } - } - } - let mut swap_data: SwapData = SwapData { +} + +/// Parse token transfer data from next instructions +pub fn parse_swap_data_from_next_instructions( + event: Box, + inner_instruction: &solana_transaction_status::InnerInstructions, + current_index: i8, + accounts: &[Pubkey], +) -> Option { + let mut swap_data = SwapData { from_mint: Pubkey::default(), to_mint: Pubkey::default(), from_amount: 0, to_amount: 0, description: None, }; - let sol_mint = Pubkey::from_str("So11111111111111111111111111111111111111111").unwrap(); - if transfer_datas.len() > 0 { - let mut user: Option = None; - let mut from_mint: Option = None; - let mut to_mint: Option = None; - let mut user_from_token: Option = None; - let mut user_to_token: Option = None; - let mut from_vault: Option = None; - let mut to_vault: Option = None; - match_event!(event, { - BonkTradeEvent => |e: BonkTradeEvent| { - user = Some(e.payer); - from_mint = Some(e.base_token_mint); - to_mint = Some(e.quote_token_mint); - user_from_token = Some(e.user_base_token); - user_to_token = Some(e.user_quote_token); - from_vault = Some(e.base_vault); - to_vault = Some(e.quote_vault); - }, - PumpFunTradeEvent => |e: PumpFunTradeEvent| { - swap_data.from_mint = if e.is_buy { - sol_mint - } else { - e.mint - }; - swap_data.to_mint = if e.is_buy { - e.mint - } else { - sol_mint - }; - }, - PumpSwapBuyEvent => |e: PumpSwapBuyEvent| { - swap_data.from_mint = e.quote_mint; - swap_data.to_mint = e.base_mint; - }, - PumpSwapSellEvent => |e: PumpSwapSellEvent| { - swap_data.from_mint = e.base_mint; - swap_data.to_mint = e.quote_mint; - }, - RaydiumCpmmSwapEvent => |e: RaydiumCpmmSwapEvent| { - user = Some(e.payer); - from_mint = Some(e.input_token_mint); - to_mint = Some(e.output_token_mint); - user_from_token = Some(e.input_token_account); - user_to_token = Some(e.output_token_account); - from_vault = Some(e.input_vault); - to_vault = Some(e.output_vault); - }, - RaydiumClmmSwapEvent => |e: RaydiumClmmSwapEvent| { - user = Some(e.payer); - swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".to_string()); - user_from_token = Some(e.input_token_account); - user_to_token = Some(e.output_token_account); - from_vault = Some(e.input_vault); - to_vault = Some(e.output_vault); - }, - RaydiumClmmSwapV2Event => |e: RaydiumClmmSwapV2Event| { - user = Some(e.payer); - from_mint = Some(e.input_vault_mint); - to_mint = Some(e.output_vault_mint); - user_from_token = Some(e.input_token_account); - user_to_token = Some(e.output_token_account); - from_vault = Some(e.input_vault); - to_vault = Some(e.output_vault); - }, - RaydiumAmmV4SwapEvent => |e: RaydiumAmmV4SwapEvent| { - user = Some(e.user_source_owner); - swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".to_string()); - user_from_token = Some(e.user_source_token_account); - user_to_token = Some(e.user_destination_token_account); - from_vault = Some(e.pool_pc_token_account); - to_vault = Some(e.pool_coin_token_account); - }, - }); - for transfer_data in transfer_datas.clone() { - if transfer_data.source == user_to_token.unwrap_or_default() - && transfer_data.destination == to_vault.unwrap_or_default() - { - swap_data.from_mint = to_mint.unwrap_or_default(); - swap_data.from_amount = transfer_data.amount; - } else if transfer_data.source == from_vault.unwrap_or_default() - && transfer_data.destination == user_from_token.unwrap_or_default() - { - swap_data.to_mint = from_mint.unwrap_or_default(); - swap_data.to_amount = transfer_data.amount; - } else if transfer_data.source == user_from_token.unwrap_or_default() - && transfer_data.destination == from_vault.unwrap_or_default() - { - swap_data.from_mint = from_mint.unwrap_or_default(); - swap_data.from_amount = transfer_data.amount; - } else if transfer_data.source == to_vault.unwrap_or_default() - && transfer_data.destination == user_to_token.unwrap_or_default() - { - swap_data.to_mint = to_mint.unwrap_or_default(); - swap_data.to_amount = transfer_data.amount; + // 先根据 event 取出关键信息 + let mut user: Option = None; + let mut from_mint: Option = None; + let mut to_mint: Option = None; + let mut user_from_token: Option = None; + let mut user_to_token: Option = None; + let mut from_vault: Option = None; + let mut to_vault: Option = None; + + match_event!(event, { + BonkTradeEvent => |e: BonkTradeEvent| { + user = Some(e.payer); + from_mint = Some(e.base_token_mint); + to_mint = Some(e.quote_token_mint); + user_from_token = Some(e.user_base_token); + user_to_token = Some(e.user_quote_token); + from_vault = Some(e.base_vault); + to_vault = Some(e.quote_vault); + }, + PumpFunTradeEvent => |e: PumpFunTradeEvent| { + swap_data.from_mint = if e.is_buy { *SOL_MINT } else { e.mint }; + swap_data.to_mint = if e.is_buy { e.mint } else { *SOL_MINT }; + }, + PumpSwapBuyEvent => |e: PumpSwapBuyEvent| { + swap_data.from_mint = e.quote_mint; + swap_data.to_mint = e.base_mint; + }, + PumpSwapSellEvent => |e: PumpSwapSellEvent| { + swap_data.from_mint = e.base_mint; + swap_data.to_mint = e.quote_mint; + }, + RaydiumCpmmSwapEvent => |e: RaydiumCpmmSwapEvent| { + user = Some(e.payer); + from_mint = Some(e.input_token_mint); + to_mint = Some(e.output_token_mint); + user_from_token = Some(e.input_token_account); + user_to_token = Some(e.output_token_account); + from_vault = Some(e.input_vault); + to_vault = Some(e.output_vault); + }, + RaydiumClmmSwapEvent => |e: RaydiumClmmSwapEvent| { + user = Some(e.payer); + swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".to_string()); + user_from_token = Some(e.input_token_account); + user_to_token = Some(e.output_token_account); + from_vault = Some(e.input_vault); + to_vault = Some(e.output_vault); + }, + RaydiumClmmSwapV2Event => |e: RaydiumClmmSwapV2Event| { + user = Some(e.payer); + from_mint = Some(e.input_vault_mint); + to_mint = Some(e.output_vault_mint); + user_from_token = Some(e.input_token_account); + user_to_token = Some(e.output_token_account); + from_vault = Some(e.input_vault); + to_vault = Some(e.output_vault); + }, + RaydiumAmmV4SwapEvent => |e: RaydiumAmmV4SwapEvent| { + user = Some(e.user_source_owner); + swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".to_string()); + user_from_token = Some(e.user_source_token_account); + user_to_token = Some(e.user_destination_token_account); + from_vault = Some(e.pool_pc_token_account); + to_vault = Some(e.pool_coin_token_account); + }, + }); + + let user_to_token = user_to_token.unwrap_or_default(); + let user_from_token = user_from_token.unwrap_or_default(); + let to_vault = to_vault.unwrap_or_default(); + let from_vault = from_vault.unwrap_or_default(); + let to_mint = to_mint.unwrap_or_default(); + let from_mint = from_mint.unwrap_or_default(); + + // 单次循环完成提取和判断 + for instruction in inner_instruction.instructions.iter().skip((current_index + 1) as usize) { + let compiled = &instruction.instruction; + let program_id = accounts[compiled.program_id_index as usize]; + if !SYSTEM_PROGRAMS.contains(&program_id) { + break; + } + let data = &compiled.data; + let get_pubkey = |i: usize| accounts[compiled.accounts[i] as usize]; + let (source, destination, amount) = match data[0] { + 12 if compiled.accounts.len() >= 4 => { + let amt = u64::from_le_bytes(data[1..9].try_into().unwrap()); + (get_pubkey(0), get_pubkey(2), amt) } + 3 if compiled.accounts.len() >= 3 => { + let amt = u64::from_le_bytes(data[1..9].try_into().unwrap()); + (get_pubkey(0), get_pubkey(1), amt) + } + 2 if compiled.accounts.len() >= 2 => { + let amt = u64::from_le_bytes(data[4..12].try_into().unwrap()); + (get_pubkey(0), get_pubkey(1), amt) + } + _ => continue, + }; + + match (source, destination) { + (s, d) if s == user_to_token && d == to_vault => { + swap_data.from_mint = to_mint; + swap_data.from_amount = amount; + } + (s, d) if s == from_vault && d == user_from_token => { + swap_data.to_mint = from_mint; + swap_data.to_amount = amount; + } + (s, d) if s == user_from_token && d == from_vault => { + swap_data.from_mint = from_mint; + swap_data.from_amount = amount; + } + (s, d) if s == to_vault && d == user_to_token => { + swap_data.to_mint = to_mint; + swap_data.to_amount = amount; + } + (s, d) if s == user_from_token && d == to_vault => { + swap_data.from_mint = from_mint; + swap_data.from_amount = amount; + } + (s, d) if s == from_vault && d == user_to_token => { + swap_data.to_mint = to_mint; + swap_data.to_amount = amount; + } + _ => {} + } + if swap_data.from_mint != Pubkey::default() && swap_data.to_mint != Pubkey::default() { + break; + } + if swap_data.from_amount != 0 && swap_data.to_amount != 0 { + break; } } + if swap_data.from_mint != Pubkey::default() || swap_data.to_mint != Pubkey::default() || swap_data.from_amount != 0 || swap_data.to_amount != 0 { - (transfer_datas, Some(swap_data)) + Some(swap_data) } else { - (transfer_datas, None) + None } } diff --git a/src/streaming/event_parser/common/utils.rs b/src/streaming/event_parser/common/utils.rs index 2d372b2..80458d4 100755 --- a/src/streaming/event_parser/common/utils.rs +++ b/src/streaming/event_parser/common/utils.rs @@ -1,5 +1,3 @@ -use base64::engine::general_purpose; -use base64::Engine; use std::time::{SystemTime, UNIX_EPOCH}; /// 获取当前时间戳 @@ -7,16 +5,6 @@ pub fn current_timestamp() -> i64 { SystemTime::now().duration_since(UNIX_EPOCH).expect("Time went backwards").as_secs() as i64 } -/// 从base64字符串解码数据 -pub fn decode_base64(data: &str) -> Result, base64::DecodeError> { - general_purpose::STANDARD.decode(data) -} - -/// 将数据编码为base64字符串 -pub fn encode_base64(data: &[u8]) -> String { - general_purpose::STANDARD.encode(data) -} - /// 从字节数组中提取鉴别器和剩余数据 pub fn extract_discriminator(length: usize, data: &[u8]) -> Option<(&[u8], &[u8])> { if data.len() < length { diff --git a/src/streaming/event_parser/core/account_event_parser.rs b/src/streaming/event_parser/core/account_event_parser.rs index 7e95698..030266a 100644 --- a/src/streaming/event_parser/core/account_event_parser.rs +++ b/src/streaming/event_parser/core/account_event_parser.rs @@ -158,12 +158,11 @@ impl AccountEventParser { pub fn parse_account_event( protocols: Vec, account: AccountPretty, - program_received_time_ms: i64, event_type_filter: Option, ) -> Option> { let configs = Self::configs(protocols, event_type_filter); for config in configs { - if account.owner == config.program_id.to_string() + if account.owner == config.program_id && account.data[..config.account_discriminator.len()] == *config.account_discriminator { @@ -171,17 +170,17 @@ impl AccountEventParser { &account, EventMetadata { slot: account.slot, - signature: account.signature.clone(), + signature: account.signature.to_string(), protocol: config.protocol_type, event_type: config.event_type, program_id: config.program_id, - program_received_time_ms, + program_received_time_us: account.program_received_time_us, ..Default::default() }, ); if let Some(mut event) = event { - event.set_program_handle_time_consuming_ms( - chrono::Utc::now().timestamp_millis() - program_received_time_ms, + event.set_program_handle_time_consuming_us( + chrono::Utc::now().timestamp_micros() - account.program_received_time_us, ); return Some(event); } diff --git a/src/streaming/event_parser/core/common_event_parser.rs b/src/streaming/event_parser/core/common_event_parser.rs index 8442446..6ec37ae 100644 --- a/src/streaming/event_parser/core/common_event_parser.rs +++ b/src/streaming/event_parser/core/common_event_parser.rs @@ -8,8 +8,17 @@ impl CommonEventParser { slot: u64, block_hash: &str, block_time_ms: i64, + program_received_time_us: i64, ) -> Box { - let block_meta_event = BlockMetaEvent::new(slot, block_hash.to_string(), block_time_ms); + let mut block_meta_event = BlockMetaEvent::new( + slot, + block_hash.to_string(), + block_time_ms, + program_received_time_us, + ); + block_meta_event.set_program_handle_time_consuming_us( + chrono::Utc::now().timestamp_micros() - program_received_time_us, + ); Box::new(block_meta_event) } } diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index a6d8574..d29e52e 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -1,17 +1,14 @@ use anyhow::Result; use prost_types::Timestamp; +use solana_sdk::signature::Signature; use solana_sdk::{ instruction::CompiledInstruction, pubkey::Pubkey, transaction::VersionedTransaction, }; -use solana_transaction_status::{ - EncodedTransactionWithStatusMeta, UiCompiledInstruction, UiInnerInstructions, UiInstruction, -}; +use solana_transaction_status::{InnerInstructions, TransactionWithStatusMeta}; +use std::collections::HashMap; use std::fmt::Debug; -use std::{collections::HashMap, str::FromStr}; -use crate::streaming::event_parser::common::{ - parse_transfer_datas_from_next_instructions, SwapData, TransferData, -}; +use crate::streaming::event_parser::common::{parse_swap_data_from_next_instructions, SwapData}; use crate::streaming::event_parser::protocols::pumpswap::{PumpSwapBuyEvent, PumpSwapSellEvent}; use crate::streaming::event_parser::{ common::{utils::*, EventMetadata, EventType, ProtocolType}, @@ -20,12 +17,16 @@ use crate::streaming::event_parser::{ pumpfun::{PumpFunCreateTokenEvent, PumpFunTradeEvent}, }, }; +use crate::streaming::shred::MetricsEventType; /// Unified Event Interface - All protocol events must implement this trait pub trait UnifiedEvent: Debug + Send + Sync { /// Get event ID fn id(&self) -> &str; + /// Set event ID + fn clear_id(&mut self); + /// Get event type fn event_type(&self) -> EventType; @@ -36,13 +37,13 @@ pub trait UnifiedEvent: Debug + Send + Sync { fn slot(&self) -> u64; /// Get program received timestamp (milliseconds) - fn program_received_time_ms(&self) -> i64; + fn program_received_time_us(&self) -> i64; /// Processing time consumption (milliseconds) - fn program_handle_time_consuming_ms(&self) -> i64; + fn program_handle_time_consuming_us(&self) -> i64; /// Set processing time consumption (milliseconds) - fn set_program_handle_time_consuming_ms(&mut self, program_handle_time_consuming_ms: i64); + fn set_program_handle_time_consuming_us(&mut self, program_handle_time_consuming_us: i64); /// Convert event to Any for downcasting fn as_any(&self) -> &dyn std::any::Any; @@ -58,12 +59,8 @@ pub trait UnifiedEvent: Debug + Send + Sync { // Default implementation: no merging operation } - /// Set transfer datas - fn set_transfer_datas( - &mut self, - transfer_datas: Vec, - swap_data: Option, - ); + /// Set swap data + fn set_swap_data(&mut self, swap_data: SwapData); /// Get index fn index(&self) -> String; @@ -80,11 +77,11 @@ pub trait EventParser: Send + Sync { #[allow(clippy::too_many_arguments)] fn parse_events_from_inner_instruction( &self, - inner_instruction: &UiCompiledInstruction, - signature: &str, + inner_instruction: &CompiledInstruction, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec>; @@ -94,10 +91,10 @@ pub trait EventParser: Send + Sync { &self, instruction: &CompiledInstruction, accounts: &[Pubkey], - signature: &str, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec>; @@ -106,12 +103,12 @@ pub trait EventParser: Send + Sync { async fn parse_instruction_events_from_versioned_transaction( &self, transaction: &VersionedTransaction, - signature: &str, + signature: Signature, slot: Option, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, accounts: &[Pubkey], - inner_instructions: &[UiInnerInstructions], + inner_instructions: &[InnerInstructions], ) -> Result>> { // 预分配容量，避免动态扩容 let mut instruction_events = Vec::with_capacity(16); @@ -140,7 +137,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, format!("{index}"), ) .await @@ -152,14 +149,15 @@ pub trait EventParser: Send + Sync { }) { events.iter_mut().for_each(|event| { - let (transfer_datas, swap_data) = - parse_transfer_datas_from_next_instructions( - event.clone_boxed(), - inn, - -1_i8, - &accounts, - ); - event.set_transfer_datas(transfer_datas, swap_data); + let swap_data = parse_swap_data_from_next_instructions( + event.clone_boxed(), + inn, + -1_i8, + &accounts, + ); + if let Some(swap_data) = swap_data { + event.set_swap_data(swap_data); + } }); } instruction_events.extend(events); @@ -175,10 +173,10 @@ pub trait EventParser: Send + Sync { async fn parse_versioned_transaction( &self, versioned_tx: &VersionedTransaction, - signature: &str, + signature: Signature, slot: Option, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, bot_wallet: Option, ) -> Result>> { let accounts: Vec = versioned_tx.message.static_account_keys().to_vec(); @@ -188,7 +186,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, &accounts, &[], ) @@ -199,146 +197,108 @@ pub trait EventParser: Send + Sync { async fn parse_transaction( &self, - tx: EncodedTransactionWithStatusMeta, - signature: &str, + tx: TransactionWithStatusMeta, + signature: Signature, slot: Option, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, bot_wallet: Option, ) -> Result>> { - // TODO: bug - 待优化 - // // 生成缓存键 - // let cache_key = format!("{}_{}_{}", signature, slot.unwrap_or(0), program_received_time_ms); - - // // 尝试从缓存获取 - // if let Some(cached_events) = PARSE_CACHE.get(&cache_key).await { - // return Ok(cached_events); - // } - - let transaction = tx.transaction; - // 检查交易元数据 - let meta = - tx.meta.as_ref().ok_or_else(|| anyhow::anyhow!("Missing transaction metadata"))?; + let versioned_tx = tx.get_transaction(); + let meta = tx.get_status_meta(); let mut address_table_lookups: Vec = vec![]; - let mut inner_instructions: Vec = vec![]; - if meta.err.is_none() { - // 正确处理OptionSerializer类型 - if let solana_transaction_status::option_serializer::OptionSerializer::Some( - meta_inner_instructions, - ) = &meta.inner_instructions - { - inner_instructions = meta_inner_instructions.clone(); + let mut inner_instructions: Vec = vec![]; + if let Some(meta) = meta { + inner_instructions = meta.inner_instructions.unwrap_or_default(); + for loopup in meta.loaded_addresses.writable { + address_table_lookups.push(loopup); } - if let solana_transaction_status::option_serializer::OptionSerializer::Some( - loaded_addresses, - ) = &meta.loaded_addresses - { - for lookup in &loaded_addresses.writable { - if let Ok(pubkey) = Pubkey::from_str(lookup) { - address_table_lookups.push(pubkey); - } - } - for lookup in &loaded_addresses.readonly { - if let Ok(pubkey) = Pubkey::from_str(lookup) { - address_table_lookups.push(pubkey); - } - } + for loopup in meta.loaded_addresses.readonly { + address_table_lookups.push(loopup); } } let mut accounts: Vec = vec![]; // 预分配容量，避免动态扩容 - let mut instruction_events = Vec::with_capacity(16); + let mut instruction_events: Vec> = Vec::with_capacity(16); // 解析指令事件 - if let Some(versioned_tx) = transaction.decode() { - accounts = versioned_tx.message.static_account_keys().to_vec(); - accounts.extend(address_table_lookups.clone()); + accounts = versioned_tx.message.static_account_keys().to_vec(); + accounts.extend(address_table_lookups.clone()); - instruction_events = self - .parse_instruction_events_from_versioned_transaction( - &versioned_tx, - signature, - slot, - block_time, - program_received_time_ms, - &accounts, - &inner_instructions, - ) - .await - .unwrap_or_else(|_e| vec![]); - } else { - accounts.extend(address_table_lookups.clone()); - } + instruction_events = self + .parse_instruction_events_from_versioned_transaction( + &versioned_tx, + signature, + slot, + block_time, + program_received_time_us, + &accounts, + &inner_instructions, + ) + .await + .unwrap_or_else(|_e| vec![]); // 解析内联指令事件 // 预分配容量，避免动态扩容 - let mut inner_instruction_events = Vec::with_capacity(8); + let mut inner_instruction_events: Vec> = Vec::with_capacity(8); // 检查交易是否成功 - if meta.err.is_none() { - for inner_instruction in inner_instructions { - for (index, instruction) in inner_instruction.instructions.iter().enumerate() { - if let UiInstruction::Compiled(compiled) = instruction { - // 解析嵌套指令 - let compiled_instruction = CompiledInstruction { - program_id_index: compiled.program_id_index, - accounts: compiled.accounts.clone(), - data: bs58::decode(compiled.data.clone()) - .into_vec() - .unwrap_or_else(|_| vec![]), - }; - if let Ok(mut events) = self - .parse_instruction( - &compiled_instruction, + for inner_instruction in inner_instructions { + for (index, instruction) in inner_instruction.instructions.iter().enumerate() { + // 解析嵌套指令 + let compiled_instruction = instruction.instruction.clone(); + if let Ok(mut events) = self + .parse_instruction( + &compiled_instruction, + &accounts, + signature, + slot, + block_time, + program_received_time_us, + format!("{}.{}", inner_instruction.index, index), + ) + .await + { + if !events.is_empty() { + events.iter_mut().for_each(|event| { + let swap_data = parse_swap_data_from_next_instructions( + event.clone_boxed(), + &inner_instruction, + index as i8, &accounts, - signature, - slot, - block_time, - program_received_time_ms, - format!("{}.{}", inner_instruction.index, index), - ) - .await - { - if !events.is_empty() { - events.iter_mut().for_each(|event| { - let (transfer_datas, swap_data) = - parse_transfer_datas_from_next_instructions( - event.clone_boxed(), - &inner_instruction, - index as i8, - &accounts, - ); - event.set_transfer_datas(transfer_datas, swap_data); - }); - instruction_events.extend(events); + ); + if let Some(swap_data) = swap_data { + event.set_swap_data(swap_data); } - } - if let Ok(mut events) = self - .parse_inner_instruction( - compiled, - signature, - slot, - block_time, - program_received_time_ms, - format!("{}.{}", inner_instruction.index, index), - ) - .await - { - if !events.is_empty() { - events.iter_mut().for_each(|event| { - let (transfer_datas, swap_data) = - parse_transfer_datas_from_next_instructions( - event.clone_boxed(), - &inner_instruction, - index as i8, - &accounts, - ); - event.set_transfer_datas(transfer_datas, swap_data); - }); - inner_instruction_events.extend(events); + }); + instruction_events.extend(events); + } + } + if let Ok(mut events) = self + .parse_inner_instruction( + &compiled_instruction, + signature, + slot, + block_time, + program_received_time_us, + format!("{}.{}", inner_instruction.index, index), + ) + .await + { + if !events.is_empty() { + events.iter_mut().for_each(|event| { + let swap_data = parse_swap_data_from_next_instructions( + event.clone_boxed(), + &inner_instruction, + index as i8, + &accounts, + ); + if let Some(swap_data) = swap_data { + event.set_swap_data(swap_data); } - } + }); + inner_instruction_events.extend(events); } } } @@ -386,9 +346,6 @@ pub trait EventParser: Send + Sync { let result = self.process_events(instruction_events, bot_wallet); - // 缓存结果 - // PARSE_CACHE.set(cache_key, result.clone()).await; - Ok(result) } @@ -397,7 +354,6 @@ pub trait EventParser: Send + Sync { mut events: Vec>, bot_wallet: Option, ) -> Vec> { - let start_time = std::time::Instant::now(); let mut dev_address = vec![]; let mut bonk_dev_address = None; for event in &mut events { @@ -458,18 +414,7 @@ pub trait EventParser: Send + Sync { trade_info.is_dev_create_token_trade = false; } } - let now = chrono::Utc::now().timestamp_millis(); - event.set_program_handle_time_consuming_ms(now - event.program_received_time_ms()); - } - - // 记录处理时间 - let processing_time = start_time.elapsed(); - if processing_time.as_millis() > 10 { - log::warn!( - "Event processing took {}ms for {} events", - processing_time.as_millis(), - events.len() - ); + event.clear_id(); } events @@ -477,11 +422,11 @@ pub trait EventParser: Send + Sync { async fn parse_inner_instruction( &self, - instruction: &UiCompiledInstruction, - signature: &str, + instruction: &CompiledInstruction, + signature: Signature, slot: Option, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Result>> { let slot = slot.unwrap_or(0); @@ -490,7 +435,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index, ); Ok(events) @@ -501,10 +446,10 @@ pub trait EventParser: Send + Sync { &self, instruction: &CompiledInstruction, accounts: &[Pubkey], - signature: &str, + signature: Signature, slot: Option, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Result>> { let slot = slot.unwrap_or(0); @@ -514,7 +459,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index, ); Ok(events) @@ -588,10 +533,10 @@ impl GenericEventParser { &self, config: &GenericEventParseConfig, data: &[u8], - signature: &str, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Option> { if let Some(parser) = config.inner_instruction_parser { @@ -607,7 +552,7 @@ impl GenericEventParser { config.event_type.clone(), config.program_id, index, - program_received_time_ms, + program_received_time_us, ); parser(data, metadata) } else { @@ -622,10 +567,10 @@ impl GenericEventParser { config: &GenericEventParseConfig, data: &[u8], account_pubkeys: &[Pubkey], - signature: &str, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Option> { if let Some(parser) = config.instruction_parser { @@ -641,7 +586,7 @@ impl GenericEventParser { config.event_type.clone(), config.program_id, index, - program_received_time_ms, + program_received_time_us, ); parser(data, account_pubkeys, metadata) } else { @@ -662,16 +607,14 @@ impl EventParser for GenericEventParser { #[allow(clippy::too_many_arguments)] fn parse_events_from_inner_instruction( &self, - inner_instruction: &UiCompiledInstruction, - signature: &str, + inner_instruction: &CompiledInstruction, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec> { - let inner_instruction_data = inner_instruction.data.clone(); - let inner_instruction_data_decoded = - bs58::decode(inner_instruction_data).into_vec().unwrap_or_else(|_| vec![]); + let inner_instruction_data_decoded = inner_instruction.data.clone(); if inner_instruction_data_decoded.len() < 16 { return Vec::new(); } @@ -688,7 +631,7 @@ impl EventParser for GenericEventParser { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index.clone(), ) { events.push(event); @@ -705,10 +648,10 @@ impl EventParser for GenericEventParser { &self, instruction: &CompiledInstruction, accounts: &[Pubkey], - signature: &str, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec> { let program_id = accounts[instruction.program_id_index as usize]; @@ -741,7 +684,7 @@ impl EventParser for GenericEventParser { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index.clone(), ) { events.push(event); diff --git a/src/streaming/event_parser/protocols/block/block_meta_event.rs b/src/streaming/event_parser/protocols/block/block_meta_event.rs index f4bee6d..2987246 100644 --- a/src/streaming/event_parser/protocols/block/block_meta_event.rs +++ b/src/streaming/event_parser/protocols/block/block_meta_event.rs @@ -13,7 +13,12 @@ pub struct BlockMetaEvent { } impl BlockMetaEvent { - pub fn new(slot: u64, block_hash: String, block_time_ms: i64) -> Self { + pub fn new( + slot: u64, + block_hash: String, + block_time_ms: i64, + program_received_time_us: i64, + ) -> Self { let metadata = EventMetadata::new( format!("block_{}_{}", slot, block_hash), "".to_string(), @@ -24,7 +29,7 @@ impl BlockMetaEvent { EventType::BlockMeta, solana_sdk::pubkey::Pubkey::default(), "".to_string(), - chrono::Utc::now().timestamp_millis(), + program_received_time_us, ); Self { metadata, slot, block_hash } } diff --git a/src/streaming/event_parser/protocols/bonk/parser.rs b/src/streaming/event_parser/protocols/bonk/parser.rs index 056b39d..a03a38e 100755 --- a/src/streaming/event_parser/protocols/bonk/parser.rs +++ b/src/streaming/event_parser/protocols/bonk/parser.rs @@ -1,14 +1,16 @@ use std::collections::HashMap; use prost_types::Timestamp; -use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; -use solana_transaction_status::UiCompiledInstruction; +use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey, signature::Signature}; use crate::streaming::event_parser::{ common::{utils::*, EventMetadata, EventType, ProtocolType}, core::traits::{EventParser, GenericEventParseConfig, GenericEventParser, UnifiedEvent}, protocols::bonk::{ - bonk_pool_create_event_log_decode, bonk_trade_event_log_decode, discriminators, AmmFeeOn, BonkMigrateToAmmEvent, BonkMigrateToCpswapEvent, BonkPoolCreateEvent, BonkTradeEvent, ConstantCurve, CurveParams, FixedCurve, LinearCurve, MintParams, TradeDirection, VestingParams + bonk_pool_create_event_log_decode, bonk_trade_event_log_decode, discriminators, AmmFeeOn, + BonkMigrateToAmmEvent, BonkMigrateToCpswapEvent, BonkPoolCreateEvent, BonkTradeEvent, + ConstantCurve, CurveParams, FixedCurve, LinearCurve, MintParams, TradeDirection, + VestingParams, }, }; @@ -611,11 +613,11 @@ impl EventParser for BonkEventParser { } fn parse_events_from_inner_instruction( &self, - inner_instruction: &UiCompiledInstruction, - signature: &str, + inner_instruction: &CompiledInstruction, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec> { self.inner.parse_events_from_inner_instruction( @@ -623,7 +625,7 @@ impl EventParser for BonkEventParser { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index, ) } @@ -632,10 +634,10 @@ impl EventParser for BonkEventParser { &self, instruction: &CompiledInstruction, accounts: &[Pubkey], - signature: &str, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec> { self.inner.parse_events_from_instruction( @@ -644,7 +646,7 @@ impl EventParser for BonkEventParser { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index, ) } diff --git a/src/streaming/event_parser/protocols/mutil/parser.rs b/src/streaming/event_parser/protocols/mutil/parser.rs index af3902d..9748124 100755 --- a/src/streaming/event_parser/protocols/mutil/parser.rs +++ b/src/streaming/event_parser/protocols/mutil/parser.rs @@ -1,8 +1,8 @@ use std::collections::HashMap; use prost_types::Timestamp; +use solana_sdk::signature::Signature; use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; -use solana_transaction_status::UiCompiledInstruction; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::{ @@ -23,18 +23,38 @@ impl MutilEventParser { // Merge inner_instruction_configs, append configurations to existing Vec for (key, configs) in parse.inner_instruction_configs() { - let filtered_configs: Vec = configs.into_iter().filter(|config| { - event_type_filter.as_ref().map(|filter| filter.include.contains(&config.event_type)).unwrap_or(true) - }).collect(); - inner.inner_instruction_configs.entry(key).or_insert_with(Vec::new).extend(filtered_configs); + let filtered_configs: Vec = configs + .into_iter() + .filter(|config| { + event_type_filter + .as_ref() + .map(|filter| filter.include.contains(&config.event_type)) + .unwrap_or(true) + }) + .collect(); + inner + .inner_instruction_configs + .entry(key) + .or_insert_with(Vec::new) + .extend(filtered_configs); } // Merge instruction_configs, append configurations to existing Vec for (key, configs) in parse.instruction_configs() { - let filtered_configs: Vec = configs.into_iter().filter(|config| { - event_type_filter.as_ref().map(|filter| filter.include.contains(&config.event_type)).unwrap_or(true) - }).collect(); - inner.instruction_configs.entry(key).or_insert_with(Vec::new).extend(filtered_configs); + let filtered_configs: Vec = configs + .into_iter() + .filter(|config| { + event_type_filter + .as_ref() + .map(|filter| filter.include.contains(&config.event_type)) + .unwrap_or(true) + }) + .collect(); + inner + .instruction_configs + .entry(key) + .or_insert_with(Vec::new) + .extend(filtered_configs); } // Append program_ids (this is already appending) @@ -54,11 +74,11 @@ impl EventParser for MutilEventParser { } fn parse_events_from_inner_instruction( &self, - inner_instruction: &UiCompiledInstruction, - signature: &str, + inner_instruction: &CompiledInstruction, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec> { self.inner.parse_events_from_inner_instruction( @@ -66,7 +86,7 @@ impl EventParser for MutilEventParser { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index, ) } @@ -75,10 +95,10 @@ impl EventParser for MutilEventParser { &self, instruction: &CompiledInstruction, accounts: &[Pubkey], - signature: &str, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec> { self.inner.parse_events_from_instruction( @@ -87,7 +107,7 @@ impl EventParser for MutilEventParser { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index, ) } diff --git a/src/streaming/event_parser/protocols/pumpfun/parser.rs b/src/streaming/event_parser/protocols/pumpfun/parser.rs index 4faaf01..e907c11 100755 --- a/src/streaming/event_parser/protocols/pumpfun/parser.rs +++ b/src/streaming/event_parser/protocols/pumpfun/parser.rs @@ -1,8 +1,7 @@ use std::collections::HashMap; use prost_types::Timestamp; -use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; -use solana_transaction_status::UiCompiledInstruction; +use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey, signature::Signature}; use crate::streaming::event_parser::{ common::{EventMetadata, EventType, ProtocolType}, @@ -295,11 +294,11 @@ impl EventParser for PumpFunEventParser { } fn parse_events_from_inner_instruction( &self, - inner_instruction: &UiCompiledInstruction, - signature: &str, + inner_instruction: &CompiledInstruction, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec> { self.inner.parse_events_from_inner_instruction( @@ -307,7 +306,7 @@ impl EventParser for PumpFunEventParser { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index, ) } @@ -316,10 +315,10 @@ impl EventParser for PumpFunEventParser { &self, instruction: &CompiledInstruction, accounts: &[Pubkey], - signature: &str, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec> { self.inner.parse_events_from_instruction( @@ -328,7 +327,7 @@ impl EventParser for PumpFunEventParser { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index, ) } diff --git a/src/streaming/event_parser/protocols/pumpswap/parser.rs b/src/streaming/event_parser/protocols/pumpswap/parser.rs index ec84ad1..6bd9b57 100755 --- a/src/streaming/event_parser/protocols/pumpswap/parser.rs +++ b/src/streaming/event_parser/protocols/pumpswap/parser.rs @@ -1,8 +1,7 @@ use std::collections::HashMap; use prost_types::Timestamp; -use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; -use solana_transaction_status::UiCompiledInstruction; +use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey, signature::Signature}; use crate::streaming::event_parser::{ common::{read_u64_le, EventMetadata, EventType, ProtocolType}, @@ -385,11 +384,11 @@ impl EventParser for PumpSwapEventParser { } fn parse_events_from_inner_instruction( &self, - inner_instruction: &UiCompiledInstruction, - signature: &str, + inner_instruction: &CompiledInstruction, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec> { self.inner.parse_events_from_inner_instruction( @@ -397,7 +396,7 @@ impl EventParser for PumpSwapEventParser { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index, ) } @@ -406,10 +405,10 @@ impl EventParser for PumpSwapEventParser { &self, instruction: &CompiledInstruction, accounts: &[Pubkey], - signature: &str, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec> { self.inner.parse_events_from_instruction( @@ -418,7 +417,7 @@ impl EventParser for PumpSwapEventParser { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index, ) } diff --git a/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs b/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs index ba4a5cf..593e1af 100755 --- a/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs @@ -1,8 +1,7 @@ use std::collections::HashMap; use prost_types::Timestamp; -use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; -use solana_transaction_status::UiCompiledInstruction; +use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey, signature::Signature}; use crate::streaming::event_parser::{ common::{read_u64_le, EventMetadata, EventType, ProtocolType}, @@ -387,11 +386,11 @@ impl EventParser for RaydiumAmmV4EventParser { } fn parse_events_from_inner_instruction( &self, - inner_instruction: &UiCompiledInstruction, - signature: &str, + inner_instruction: &CompiledInstruction, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec> { self.inner.parse_events_from_inner_instruction( @@ -399,7 +398,7 @@ impl EventParser for RaydiumAmmV4EventParser { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index, ) } @@ -408,10 +407,10 @@ impl EventParser for RaydiumAmmV4EventParser { &self, instruction: &CompiledInstruction, accounts: &[Pubkey], - signature: &str, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec> { self.inner.parse_events_from_instruction( @@ -420,7 +419,7 @@ impl EventParser for RaydiumAmmV4EventParser { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index, ) } diff --git a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs index f06f238..aef8ba9 100755 --- a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs @@ -1,8 +1,7 @@ use std::collections::HashMap; use prost_types::Timestamp; -use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; -use solana_transaction_status::UiCompiledInstruction; +use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey, signature::Signature}; use crate::streaming::event_parser::{ common::{ @@ -428,11 +427,11 @@ impl EventParser for RaydiumClmmEventParser { } fn parse_events_from_inner_instruction( &self, - inner_instruction: &UiCompiledInstruction, - signature: &str, + inner_instruction: &CompiledInstruction, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec> { self.inner.parse_events_from_inner_instruction( @@ -440,7 +439,7 @@ impl EventParser for RaydiumClmmEventParser { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index, ) } @@ -449,10 +448,10 @@ impl EventParser for RaydiumClmmEventParser { &self, instruction: &CompiledInstruction, accounts: &[Pubkey], - signature: &str, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec> { self.inner.parse_events_from_instruction( @@ -461,7 +460,7 @@ impl EventParser for RaydiumClmmEventParser { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index, ) } diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs index fdc453a..6dfdd6b 100755 --- a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs @@ -1,8 +1,7 @@ use std::collections::HashMap; use prost_types::Timestamp; -use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; -use solana_transaction_status::UiCompiledInstruction; +use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey, signature::Signature}; use crate::streaming::event_parser::{ common::{read_u64_le, EventMetadata, EventType, ProtocolType}, @@ -278,11 +277,11 @@ impl EventParser for RaydiumCpmmEventParser { } fn parse_events_from_inner_instruction( &self, - inner_instruction: &UiCompiledInstruction, - signature: &str, + inner_instruction: &CompiledInstruction, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec> { self.inner.parse_events_from_inner_instruction( @@ -290,7 +289,7 @@ impl EventParser for RaydiumCpmmEventParser { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index, ) } @@ -299,10 +298,10 @@ impl EventParser for RaydiumCpmmEventParser { &self, instruction: &CompiledInstruction, accounts: &[Pubkey], - signature: &str, + signature: Signature, slot: u64, block_time: Option, - program_received_time_ms: i64, + program_received_time_us: i64, index: String, ) -> Vec> { self.inner.parse_events_from_instruction( @@ -311,7 +310,7 @@ impl EventParser for RaydiumCpmmEventParser { signature, slot, block_time, - program_received_time_ms, + program_received_time_us, index, ) } diff --git a/src/streaming/grpc/event_processor.rs b/src/streaming/grpc/event_processor.rs deleted file mode 100644 index 10454fe..0000000 --- a/src/streaming/grpc/event_processor.rs +++ /dev/null @@ -1,290 +0,0 @@ -use std::sync::{Arc, Mutex}; - -use solana_sdk::pubkey::Pubkey; - -use super::types::EventPretty; -use crate::common::AnyResult; -use crate::streaming::common::{ - EventBatchProcessor as EventBatchCollector, MetricsEventType, MetricsManager, - StreamClientConfig as ClientConfig, -}; -use crate::streaming::event_parser::common::filter::EventTypeFilter; -use crate::streaming::event_parser::core::account_event_parser::AccountEventParser; -use crate::streaming::event_parser::core::common_event_parser::CommonEventParser; -use crate::streaming::event_parser::EventParser; -use crate::streaming::event_parser::{ - core::traits::UnifiedEvent, protocols::mutil::parser::MutilEventParser, Protocol, -}; - -/// 事件处理器 -pub struct EventProcessor { - pub(crate) metrics_manager: MetricsManager, - pub(crate) config: ClientConfig, - pub(crate) parser_cache: Arc>>>, -} - -impl EventProcessor { - /// 创建新的事件处理器 - pub fn new(metrics_manager: MetricsManager, config: ClientConfig) -> Self { - Self { metrics_manager, config, parser_cache: Arc::new(Mutex::new(None)) } - } - - /// 获取或创建解析器，使用缓存机制避免重复创建 - fn get_or_create_parser( - &self, - protocols: Vec, - event_type_filter: Option, - ) -> Arc { - let mut cache = self.parser_cache.lock().unwrap(); - if let Some(cached_parser) = cache.clone() { - return cached_parser.clone(); - } - let parser: Arc = - Arc::new(MutilEventParser::new(protocols.clone(), event_type_filter.clone())); - *cache = Some(parser.clone()); - parser - } - - /// 使用性能监控处理事件交易 - pub async fn process_event_transaction_with_metrics( - &self, - event_pretty: EventPretty, - callback: &F, - bot_wallet: Option, - protocols: Vec, - event_type_filter: Option, - ) -> AnyResult<()> - where - F: Fn(Box) + Send + Sync, - { - match event_pretty { - EventPretty::Account(account_pretty) => { - self.metrics_manager.add_account_process_count().await; - let start_time = std::time::Instant::now(); - let program_received_time_ms = chrono::Utc::now().timestamp_millis(); - let account_event = AccountEventParser::parse_account_event( - protocols.clone(), - account_pretty, - program_received_time_ms, - event_type_filter, - ); - if let Some(event) = account_event { - callback(event); - // 更新性能指标 - let processing_time = start_time.elapsed(); - let processing_time_ms = processing_time.as_millis() as f64; - // 更新性能指标（如果启用） - self.metrics_manager - .update_metrics(MetricsEventType::Account, 1, processing_time_ms) - .await; - // 记录慢处理操作 - self.metrics_manager.log_slow_processing(processing_time_ms, 1); - } - } - EventPretty::Transaction(transaction_pretty) => { - self.metrics_manager.add_tx_process_count().await; - let start_time = std::time::Instant::now(); - let program_received_time_ms = chrono::Utc::now().timestamp_millis(); - let slot = transaction_pretty.slot; - let signature = transaction_pretty.signature.to_string(); - - // 使用缓存获取解析器 - let parser = self.get_or_create_parser(protocols.clone(), event_type_filter); - let all_events = parser - .parse_transaction( - transaction_pretty.tx.clone(), - &signature, - Some(slot), - transaction_pretty.block_time.map(|ts| prost_types::Timestamp { - seconds: ts.seconds, - nanos: ts.nanos, - }), - program_received_time_ms, - bot_wallet, - ) - .await - .unwrap_or_else(|_e| vec![]); - - // 保存事件数量用于日志记录 - let event_count = all_events.len(); - - // 批量处理事件 - if !all_events.is_empty() { - for event in all_events { - callback(event); - } - } - - // 更新性能指标 - let processing_time = start_time.elapsed(); - let processing_time_ms = processing_time.as_millis() as f64; - - // 更新性能指标（如果启用） - self.metrics_manager - .update_metrics(MetricsEventType::Tx, event_count as u64, processing_time_ms) - .await; - // 记录慢处理操作 - self.metrics_manager.log_slow_processing(processing_time_ms, event_count); - } - EventPretty::BlockMeta(block_meta_pretty) => { - let start_time = std::time::Instant::now(); - self.metrics_manager.add_block_meta_process_count().await; - let block_time_ms = block_meta_pretty - .block_time - .map(|ts| ts.seconds * 1000 + ts.nanos as i64 / 1_000_000) - .unwrap_or_else(|| chrono::Utc::now().timestamp_millis()); - let block_meta_event = CommonEventParser::generate_block_meta_event( - block_meta_pretty.slot, - &block_meta_pretty.block_hash, - block_time_ms, - ); - callback(block_meta_event); - // 更新性能指标 - let processing_time = start_time.elapsed(); - let processing_time_ms = processing_time.as_millis() as f64; - // 更新性能指标（如果启用） - self.metrics_manager - .update_metrics(MetricsEventType::BlockMeta, 1, processing_time_ms) - .await; - // 记录慢处理操作 - self.metrics_manager.log_slow_processing(processing_time_ms, 1); - } - } - - Ok(()) - } - - /// 使用批处理处理事件交易 - pub async fn process_event_transaction_with_batch( - &self, - event_pretty: EventPretty, - batch_processor: &mut EventBatchCollector, - bot_wallet: Option, - protocols: Vec, - event_type_filter: Option, - ) -> AnyResult<()> - where - F: Fn(Vec>) + Send + Sync + 'static, - { - match event_pretty { - EventPretty::Account(account_pretty) => { - self.metrics_manager.add_account_process_count().await; - let start_time = std::time::Instant::now(); - let program_received_time_ms = chrono::Utc::now().timestamp_millis(); - let account_event = AccountEventParser::parse_account_event( - protocols.clone(), - account_pretty, - program_received_time_ms, - event_type_filter, - ); - if let Some(event) = account_event { - (batch_processor.callback)(vec![event]); - // 更新性能指标 - let processing_time = start_time.elapsed(); - let processing_time_ms = processing_time.as_millis() as f64; - // 实际调用性能指标更新 - self.metrics_manager - .update_metrics(MetricsEventType::Account, 1, processing_time_ms) - .await; - // 记录慢处理操作 - self.metrics_manager.log_slow_processing(processing_time_ms, 1); - } - } - EventPretty::Transaction(transaction_pretty) => { - self.metrics_manager.add_tx_process_count().await; - let start_time = std::time::Instant::now(); - let program_received_time_ms = chrono::Utc::now().timestamp_millis(); - let slot = transaction_pretty.slot; - let signature = transaction_pretty.signature.to_string(); - - // 使用缓存获取解析器 - let parser = self.get_or_create_parser(protocols.clone(), event_type_filter); - let result = parser - .parse_transaction( - transaction_pretty.tx.clone(), - &signature, - Some(slot), - transaction_pretty.block_time.map(|ts| prost_types::Timestamp { - seconds: ts.seconds, - nanos: ts.nanos, - }), - program_received_time_ms, - bot_wallet, - ) - .await; - - // 处理解析结果并使用批处理器 - let total_events = match result { - Ok(events) => { - let event_count = events.len(); - if !events.is_empty() { - log::debug!("Parsed {} events", event_count); - log::debug!("Adding {} events to batch processor", event_count); - for event in events { - if self.config.batch.enabled { - batch_processor.add_event(event); - } else { - // 如果批处理被禁用，直接调用回调 - // 这里需要将单个事件包装成Vec来调用批处理回调 - let single_event_batch = vec![event]; - (batch_processor.callback)(single_event_batch); - } - } - } - event_count - } - Err(e) => { - log::warn!("Failed to parse transaction: {:?}", e); - 0 - } - }; - - // 添加调试信息 - if total_events > 0 { - log::debug!( - "Total events parsed: {} for transaction {}", - total_events, - signature - ); - } - - // 更新性能指标 - let processing_time = start_time.elapsed(); - let processing_time_ms = processing_time.as_millis() as f64; - - // 实际调用性能指标更新 - self.metrics_manager - .update_metrics(MetricsEventType::Tx, total_events as u64, processing_time_ms) - .await; - - // 记录慢处理操作 - self.metrics_manager.log_slow_processing(processing_time_ms, total_events); - } - EventPretty::BlockMeta(block_meta_pretty) => { - let start_time = std::time::Instant::now(); - self.metrics_manager.add_block_meta_process_count().await; - let block_time_ms = block_meta_pretty - .block_time - .map(|ts| ts.seconds * 1000 + ts.nanos as i64 / 1_000_000) - .unwrap_or_else(|| chrono::Utc::now().timestamp_millis()); - let block_meta_event = CommonEventParser::generate_block_meta_event( - block_meta_pretty.slot, - &block_meta_pretty.block_hash, - block_time_ms, - ); - (batch_processor.callback)(vec![block_meta_event]); - // 更新性能指标 - let processing_time = start_time.elapsed(); - let processing_time_ms = processing_time.as_millis() as f64; - // 更新性能指标（如果启用） - self.metrics_manager - .update_metrics(MetricsEventType::BlockMeta, 1, processing_time_ms) - .await; - // 记录慢处理操作 - self.metrics_manager.log_slow_processing(processing_time_ms, 1); - } - } - - Ok(()) - } -} diff --git a/src/streaming/grpc/mod.rs b/src/streaming/grpc/mod.rs index 44713a9..9c012e6 100644 --- a/src/streaming/grpc/mod.rs +++ b/src/streaming/grpc/mod.rs @@ -1,25 +1,17 @@ // gRPC 相关模块 pub mod connection; -pub mod types; -pub mod subscription; pub mod stream_handler; -pub mod event_processor; +pub mod subscription; +pub mod types; // 重新导出主要类型 pub use connection::*; -pub use types::*; -pub use subscription::*; pub use stream_handler::*; -pub use event_processor::*; +pub use subscription::*; +pub use types::*; // 从公用模块重新导出 pub use crate::streaming::common::{ - StreamClientConfig as ClientConfig, - PerformanceMetrics, - MetricsManager, - EventBatchProcessor as EventBatchCollector, - BackpressureStrategy, - BatchConfig, - BackpressureConfig, - ConnectionConfig, -}; \ No newline at end of file + BackpressureConfig, BackpressureStrategy, BatchConfig, ConnectionConfig, MetricsManager, + PerformanceMetrics, StreamClientConfig as ClientConfig, +}; diff --git a/src/streaming/grpc/stream_handler.rs b/src/streaming/grpc/stream_handler.rs index 362d963..ff6fb00 100644 --- a/src/streaming/grpc/stream_handler.rs +++ b/src/streaming/grpc/stream_handler.rs @@ -1,12 +1,14 @@ use chrono::Local; use futures::{channel::mpsc, sink::Sink, SinkExt}; +use solana_sdk::pubkey::Pubkey; use yellowstone_grpc_proto::geyser::{ subscribe_update::UpdateOneof, SubscribeRequest, SubscribeRequestPing, SubscribeUpdate, }; use super::types::{BlockMetaPretty, EventPretty, TransactionPretty}; use crate::common::AnyResult; -use crate::streaming::common::BackpressureStrategy; +use crate::streaming::common::EventProcessor; +use crate::streaming::event_parser::UnifiedEvent; use crate::streaming::grpc::AccountPretty; /// 流消息处理器 @@ -14,33 +16,39 @@ pub struct StreamHandler; impl StreamHandler { /// 处理单个流消息 - pub async fn handle_stream_message( + pub async fn handle_stream_message( msg: SubscribeUpdate, - tx: &mut mpsc::Sender, subscribe_tx: &mut (impl Sink + Unpin), - backpressure_strategy: BackpressureStrategy, - ) -> AnyResult<()> { + event_processor: EventProcessor, + callback: &F, + bot_wallet: Option, + ) -> AnyResult<()> + where + F: Fn(Box) + Send + Sync, + { let created_at = msg.created_at; match msg.update_oneof { Some(UpdateOneof::Account(account)) => { let account_pretty = AccountPretty::from(account); log::debug!("Received account: {:?}", account_pretty); - Self::handle_backpressure( - tx, - EventPretty::Account(account_pretty), - backpressure_strategy, - ) - .await?; + event_processor + .process_grpc_event_transaction_with_metrics( + EventPretty::Account(account_pretty), + callback, + bot_wallet, + ) + .await?; } Some(UpdateOneof::BlockMeta(sut)) => { let block_meta_pretty = BlockMetaPretty::from((sut, created_at)); log::debug!("Received block meta: {:?}", block_meta_pretty); - Self::handle_backpressure( - tx, - EventPretty::BlockMeta(block_meta_pretty), - backpressure_strategy, - ) - .await?; + event_processor + .process_grpc_event_transaction_with_metrics( + EventPretty::BlockMeta(block_meta_pretty), + callback, + bot_wallet, + ) + .await?; } Some(UpdateOneof::Transaction(sut)) => { let transaction_pretty = TransactionPretty::from((sut, created_at)); @@ -49,14 +57,13 @@ impl StreamHandler { transaction_pretty.signature, transaction_pretty.slot ); - - // 根据背压策略处理发送 - Self::handle_backpressure( - tx, - EventPretty::Transaction(transaction_pretty), - backpressure_strategy, - ) - .await?; + event_processor + .process_grpc_event_transaction_with_metrics( + EventPretty::Transaction(transaction_pretty), + callback, + bot_wallet, + ) + .await?; } Some(UpdateOneof::Ping(_)) => { subscribe_tx @@ -77,58 +84,58 @@ impl StreamHandler { Ok(()) } - /// 处理背压策略 - async fn handle_backpressure( - tx: &mut mpsc::Sender, - event_pretty: EventPretty, - backpressure_strategy: BackpressureStrategy, - ) -> AnyResult<()> { - match backpressure_strategy { - BackpressureStrategy::Block => { - // 阻塞等待，直到有空间 - if let Err(e) = tx.send(event_pretty).await { - log::error!("Failed to send transaction to channel: {:?}", e); - return Err(anyhow::anyhow!("Channel send failed: {:?}", e)); - } - } - BackpressureStrategy::Drop => { - // 尝试发送，如果失败则丢弃 - if let Err(e) = tx.try_send(event_pretty) { - if e.is_full() { - log::warn!("Channel is full, dropping transaction"); - } else { - log::error!("Channel is closed: {:?}", e); - return Err(anyhow::anyhow!("Channel is closed: {:?}", e)); - } - } - } - BackpressureStrategy::Retry { max_attempts, wait_ms } => { - // 重试有限次数 - let mut retry_count = 0; - loop { - match tx.try_send(event_pretty.clone()) { - Ok(_) => break, - Err(e) => { - if e.is_full() { - retry_count += 1; - if retry_count >= max_attempts { - log::warn!( - "Channel is full after {} attempts, dropping transaction", - retry_count - ); - break; - } - tokio::time::sleep(tokio::time::Duration::from_millis(wait_ms)) - .await; - } else { - log::error!("Channel is closed: {:?}", e); - return Err(anyhow::anyhow!("Channel is closed: {:?}", e)); - } - } - } - } - } - } - Ok(()) - } + // /// 处理背压策略 + // async fn handle_backpressure( + // tx: &mut mpsc::Sender, + // event_pretty: EventPretty, + // backpressure_strategy: BackpressureStrategy, + // ) -> AnyResult<()> { + // match backpressure_strategy { + // BackpressureStrategy::Block => { + // // 阻塞等待，直到有空间 + // if let Err(e) = tx.send(event_pretty).await { + // log::error!("Failed to send transaction to channel: {:?}", e); + // return Err(anyhow::anyhow!("Channel send failed: {:?}", e)); + // } + // } + // BackpressureStrategy::Drop => { + // // 尝试发送，如果失败则丢弃 + // if let Err(e) = tx.try_send(event_pretty) { + // if e.is_full() { + // log::warn!("Channel is full, dropping transaction"); + // } else { + // log::error!("Channel is closed: {:?}", e); + // return Err(anyhow::anyhow!("Channel is closed: {:?}", e)); + // } + // } + // } + // BackpressureStrategy::Retry { max_attempts, wait_ms } => { + // // 重试有限次数 + // let mut retry_count = 0; + // loop { + // match tx.try_send(event_pretty.clone()) { + // Ok(_) => break, + // Err(e) => { + // if e.is_full() { + // retry_count += 1; + // if retry_count >= max_attempts { + // log::warn!( + // "Channel is full after {} attempts, dropping transaction", + // retry_count + // ); + // break; + // } + // tokio::time::sleep(tokio::time::Duration::from_millis(wait_ms)) + // .await; + // } else { + // log::error!("Channel is closed: {:?}", e); + // return Err(anyhow::anyhow!("Channel is closed: {:?}", e)); + // } + // } + // } + // } + // } + // } + // Ok(()) + // } } diff --git a/src/streaming/grpc/types.rs b/src/streaming/grpc/types.rs index 8d80a7c..b0ef05f 100644 --- a/src/streaming/grpc/types.rs +++ b/src/streaming/grpc/types.rs @@ -1,5 +1,5 @@ -use solana_sdk::signature::Signature; -use solana_transaction_status::{EncodedTransactionWithStatusMeta, UiTransactionEncoding}; +use solana_sdk::{pubkey::Pubkey, signature::Signature}; +use solana_transaction_status::TransactionWithStatusMeta; use std::{collections::HashMap, fmt}; use yellowstone_grpc_proto::{ geyser::{ @@ -22,13 +22,14 @@ pub enum EventPretty { #[derive(Clone)] pub struct AccountPretty { pub slot: u64, - pub signature: String, - pub pubkey: String, + pub signature: Signature, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub data: Vec, + pub program_received_time_us: i64, } impl fmt::Debug for AccountPretty { @@ -51,6 +52,7 @@ pub struct BlockMetaPretty { pub slot: u64, pub block_hash: String, pub block_time: Option, + pub program_received_time_us: i64, } impl fmt::Debug for BlockMetaPretty { @@ -59,6 +61,7 @@ impl fmt::Debug for BlockMetaPretty { .field("slot", &self.slot) .field("block_hash", &self.block_hash) .field("block_time", &self.block_time) + .field("program_received_time_us", &self.program_received_time_us) .finish() } } @@ -70,24 +73,17 @@ pub struct TransactionPretty { pub block_time: Option, pub signature: Signature, pub is_vote: bool, - pub tx: EncodedTransactionWithStatusMeta, + pub tx: TransactionWithStatusMeta, + pub program_received_time_us: i64, } impl fmt::Debug for TransactionPretty { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - struct TxWrap<'a>(&'a EncodedTransactionWithStatusMeta); - impl<'a> fmt::Debug for TxWrap<'a> { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - let serialized = serde_json::to_string(self.0).expect("failed to serialize"); - fmt::Display::fmt(&serialized, f) - } - } - f.debug_struct("TransactionPretty") .field("slot", &self.slot) .field("signature", &self.signature) .field("is_vote", &self.is_vote) - .field("tx", &TxWrap(&self.tx)) + .field("program_received_time_us", &self.program_received_time_us) .finish() } } @@ -97,13 +93,17 @@ impl From for AccountPretty { let account_info = account.account.unwrap(); Self { slot: account.slot, - signature: bs58::encode(&account_info.txn_signature.unwrap_or_default()).into_string(), - pubkey: bs58::encode(&account_info.pubkey).into_string(), + signature: Signature::try_from( + account_info.txn_signature.unwrap_or_default().as_slice(), + ) + .expect("valid signature"), + pubkey: Pubkey::try_from(account_info.pubkey.as_slice()).expect("valid pubkey"), executable: account_info.executable, lamports: account_info.lamports, - owner: bs58::encode(&account_info.owner).into_string(), + owner: Pubkey::try_from(account_info.owner.as_slice()).expect("valid pubkey"), rent_epoch: account_info.rent_epoch, data: account_info.data, + program_received_time_us: chrono::Utc::now().timestamp_micros(), } } } @@ -115,7 +115,12 @@ impl From<(SubscribeUpdateBlockMeta, Option)> for BlockMetaPretty { Option, ), ) -> Self { - Self { block_hash: blockhash.to_string(), block_time, slot } + Self { + block_hash: blockhash.to_string(), + block_time, + slot, + program_received_time_us: chrono::Utc::now().timestamp_micros(), + } } } @@ -134,9 +139,8 @@ impl From<(SubscribeUpdateTransaction, Option)> for TransactionPretty signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"), is_vote: tx.is_vote, tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx) - .expect("valid tx with meta") - .encode(UiTransactionEncoding::Base64, Some(u8::MAX), true) - .expect("failed to encode"), + .expect("valid tx with meta"), + program_received_time_us: chrono::Utc::now().timestamp_micros(), } } } diff --git a/src/streaming/mod.rs b/src/streaming/mod.rs index 8747e2a..9aed6b5 100755 --- a/src/streaming/mod.rs +++ b/src/streaming/mod.rs @@ -4,8 +4,8 @@ pub mod grpc; pub mod shred; pub mod shred_stream; pub mod yellowstone_grpc; -pub mod yellowstone_sub_system; +// pub mod yellowstone_sub_system; pub use shred::ShredStreamGrpc; pub use yellowstone_grpc::YellowstoneGrpc; -pub use yellowstone_sub_system::{SystemEvent, TransferInfo}; +// pub use yellowstone_sub_system::{SystemEvent, TransferInfo}; diff --git a/src/streaming/shred/connection.rs b/src/streaming/shred/connection.rs index 8a27a8a..90bc64a 100644 --- a/src/streaming/shred/connection.rs +++ b/src/streaming/shred/connection.rs @@ -1,4 +1,5 @@ use std::sync::Arc; +use std::sync::RwLock; use tokio::sync::Mutex; use tonic::transport::Channel; @@ -13,7 +14,7 @@ use crate::streaming::common::{ pub struct ShredStreamGrpc { pub shredstream_client: Arc>, pub config: StreamClientConfig, - pub metrics: Arc>, + pub metrics: Arc>, pub metrics_manager: MetricsManager, pub subscription_handle: Arc>>, } @@ -27,7 +28,7 @@ impl ShredStreamGrpc { /// 创建客户端，使用自定义配置 pub async fn new_with_config(endpoint: String, config: StreamClientConfig) -> AnyResult { let shredstream_client = ShredstreamProxyClient::connect(endpoint.clone()).await?; - let metrics = Arc::new(Mutex::new(PerformanceMetrics::new())); + let metrics = Arc::new(RwLock::new(PerformanceMetrics::new())); let config_arc = Arc::new(config.clone()); let metrics_manager = @@ -36,7 +37,7 @@ impl ShredStreamGrpc { Ok(Self { shredstream_client: Arc::new(shredstream_client), config, - metrics, + metrics: metrics.clone(), metrics_manager, subscription_handle: Arc::new(Mutex::new(None)), }) @@ -63,8 +64,8 @@ impl ShredStreamGrpc { } /// 获取性能指标 - pub async fn get_metrics(&self) -> PerformanceMetrics { - self.metrics_manager.get_metrics().await + pub fn get_metrics(&self) -> PerformanceMetrics { + self.metrics_manager.get_metrics() } /// 启用或禁用性能监控 @@ -73,8 +74,8 @@ impl ShredStreamGrpc { } /// 打印性能指标 - pub async fn print_metrics(&self) { - self.metrics_manager.print_metrics().await; + pub fn print_metrics(&self) { + self.metrics_manager.print_metrics(); } /// 启动自动性能监控任务 diff --git a/src/streaming/shred/event_processor.rs b/src/streaming/shred/event_processor.rs deleted file mode 100644 index c72895a..0000000 --- a/src/streaming/shred/event_processor.rs +++ /dev/null @@ -1,170 +0,0 @@ -use solana_sdk::pubkey::Pubkey; -use std::sync::{Arc, Mutex}; - -use crate::common::AnyResult; -use crate::streaming::common::{ - EventBatchProcessor, MetricsEventType, MetricsManager, StreamClientConfig, -}; -use crate::streaming::event_parser::common::filter::EventTypeFilter; -use crate::streaming::event_parser::protocols::MutilEventParser; -use crate::streaming::event_parser::{EventParser, Protocol, UnifiedEvent}; -use crate::streaming::shred::TransactionWithSlot; - -/// ShredStream 事件处理器 -pub struct ShredEventProcessor { - pub(crate) metrics_manager: MetricsManager, - pub(crate) config: StreamClientConfig, - pub(crate) parser_cache: Arc>>>, -} - -impl ShredEventProcessor { - /// 创建新的事件处理器 - pub fn new(metrics_manager: MetricsManager, config: StreamClientConfig) -> Self { - Self { metrics_manager, config, parser_cache: Arc::new(Mutex::new(None)) } - } - - /// 获取或创建解析器，使用缓存机制避免重复创建 - fn get_or_create_parser( - &self, - protocols: Vec, - event_type_filter: Option, - ) -> Arc { - let mut cache = self.parser_cache.lock().unwrap(); - if let Some(cached_parser) = cache.clone() { - return cached_parser.clone(); - } - let parser: Arc = - Arc::new(MutilEventParser::new(protocols.clone(), event_type_filter.clone())); - *cache = Some(parser.clone()); - parser - } - - /// 即时处理单个交易 - pub async fn process_transaction_immediate( - &self, - transaction_with_slot: TransactionWithSlot, - protocols: Vec, - bot_wallet: Option, - event_type_filter: Option, - callback: &F, - ) -> AnyResult<()> - where - F: Fn(Box) + Send + Sync, - { - let start_time = std::time::Instant::now(); - self.metrics_manager.add_tx_process_count().await; - let program_received_time_ms = chrono::Utc::now().timestamp_millis(); - let slot = transaction_with_slot.slot; - let versioned_tx = transaction_with_slot.transaction; - let signature = versioned_tx.signatures[0]; - - // 获取缓存的解析器 - let parser = self.get_or_create_parser(protocols, event_type_filter); - - let all_events = parser - .parse_versioned_transaction( - &versioned_tx, - &signature.to_string(), - Some(slot), - None, - program_received_time_ms, - bot_wallet, - ) - .await - .unwrap_or_else(|_e| vec![]); - - // 保存事件数量用于日志记录 - let event_count = all_events.len(); - - // 即时处理事件 - for event in all_events { - callback(event); - } - - // 更新性能指标 - let processing_time = start_time.elapsed(); - let processing_time_ms = processing_time.as_millis() as f64; - - // 实际调用性能指标更新 - self.update_metrics(event_count as u64, processing_time_ms).await; - - // 记录慢处理操作 - self.metrics_manager.log_slow_processing(processing_time_ms, event_count); - - Ok(()) - } - - /// 批处理模式处理单个交易 - pub async fn process_transaction_with_batch( - &self, - transaction_with_slot: TransactionWithSlot, - protocols: Vec, - bot_wallet: Option, - batch_processor: &mut EventBatchProcessor, - event_type_filter: Option, - ) -> AnyResult<()> - where - F: FnMut(Vec>) + Send + Sync + 'static, - { - let start_time = std::time::Instant::now(); - self.metrics_manager.add_tx_process_count().await; - let program_received_time_ms = chrono::Utc::now().timestamp_millis(); - let slot = transaction_with_slot.slot; - let versioned_tx = transaction_with_slot.transaction; - let signature = versioned_tx.signatures[0]; - - // 获取缓存的解析器 - let parser = self.get_or_create_parser(protocols, event_type_filter); - - let all_events = parser - .parse_versioned_transaction( - &versioned_tx, - &signature.to_string(), - Some(slot), - None, - program_received_time_ms, - bot_wallet, - ) - .await - .unwrap_or_else(|_e| vec![]); - - // 保存事件数量用于日志记录 - let event_count = all_events.len(); - - // 使用批处理器处理事件 - for event in all_events { - batch_processor.add_event(event); - } - - // 更新性能指标 - let processing_time = start_time.elapsed(); - let processing_time_ms = processing_time.as_millis() as f64; - - // 实际调用性能指标更新 - self.update_metrics(event_count as u64, processing_time_ms).await; - - // 记录慢处理操作 - self.metrics_manager.log_slow_processing(processing_time_ms, event_count); - - Ok(()) - } - - /// 更新性能指标 - async fn update_metrics(&self, events_processed: u64, processing_time_ms: f64) { - // 使用统一的指标管理器，这里假设 ShredStream 主要处理交易事件 - self.metrics_manager - .update_metrics(MetricsEventType::Tx, events_processed, processing_time_ms) - .await; - } -} - -// 实现 Clone trait 以支持模块间共享 -impl Clone for ShredEventProcessor { - fn clone(&self) -> Self { - Self { - metrics_manager: self.metrics_manager.clone(), - config: self.config.clone(), - parser_cache: self.parser_cache.clone(), - } - } -} diff --git a/src/streaming/shred/mod.rs b/src/streaming/shred/mod.rs index 5ba7daf..c2aa2d9 100644 --- a/src/streaming/shred/mod.rs +++ b/src/streaming/shred/mod.rs @@ -1,17 +1,13 @@ // ShredStream 相关模块 pub mod connection; pub mod types; -pub mod stream_handler; -pub mod event_processor; // 重新导出主要类型 pub use connection::*; pub use types::*; -pub use stream_handler::*; -pub use event_processor::*; // 从公用模块重新导出 pub use crate::streaming::common::{ - BackpressureConfig, BackpressureStrategy, BatchConfig, ConnectionConfig, EventBatchProcessor, - MetricsEventType, MetricsManager, PerformanceMetrics, StreamClientConfig, + BackpressureConfig, BackpressureStrategy, BatchConfig, ConnectionConfig, MetricsEventType, + MetricsManager, PerformanceMetrics, StreamClientConfig, }; diff --git a/src/streaming/shred/stream_handler.rs b/src/streaming/shred/stream_handler.rs deleted file mode 100644 index b5de4c2..0000000 --- a/src/streaming/shred/stream_handler.rs +++ /dev/null @@ -1,116 +0,0 @@ -use futures::{channel::mpsc, StreamExt}; -use log::error; -use solana_entry::entry::Entry; -use tokio::task::JoinHandle; - -use crate::common::AnyResult; -use crate::protos::shredstream::{ - shredstream_proxy_client::ShredstreamProxyClient, SubscribeEntriesRequest, -}; -use crate::streaming::shred::TransactionWithSlot; - -/// ShredStream 流处理器 -pub struct ShredStreamHandler; - -impl ShredStreamHandler { - /// 启动 ShredStream 流处理任务 - /// - /// # 参数 - /// * `client` - ShredStream 客户端 - /// * `tx` - 事务发送通道 - /// * `channel_size` - 通道缓冲区大小 - /// - /// # 返回值 - /// 返回 ShredStream 流处理任务句柄和事务接收通道 - pub async fn start_stream_processing( - mut client: ShredstreamProxyClient, - channel_size: usize, - ) -> AnyResult<(JoinHandle<()>, mpsc::Receiver)> { - let request = tonic::Request::new(SubscribeEntriesRequest {}); - let stream = client.subscribe_entries(request).await?.into_inner(); - let (tx, rx) = mpsc::channel::(channel_size); - - let stream_task = tokio::spawn(Self::process_stream_messages(stream, tx)); - - Ok((stream_task, rx)) - } - - /// 处理流消息 - /// - /// # 参数 - /// * `stream` - ShredStream 数据流 - /// * `tx` - 事务发送通道 - async fn process_stream_messages( - mut stream: tonic::codec::Streaming, - mut tx: mpsc::Sender, - ) { - while let Some(message) = stream.next().await { - match message { - Ok(msg) => { - if let Err(e) = Self::handle_stream_message(msg, &mut tx).await { - error!("Error handling stream message: {e:?}"); - continue; - } - } - Err(error) => { - error!("Stream error: {error:?}"); - break; - } - } - } - } - - /// 处理单个流消息 - /// - /// # 参数 - /// * `msg` - ShredStream 消息 - /// * `tx` - 事务发送通道 - async fn handle_stream_message( - msg: crate::protos::shredstream::Entry, - tx: &mut mpsc::Sender, - ) -> AnyResult<()> { - if let Ok(entries) = bincode::deserialize::>(&msg.entries) { - for entry in entries { - for transaction in entry.transactions { - let transaction_with_slot = - TransactionWithSlot::new(transaction.clone(), msg.slot); - - if let Err(e) = tx.try_send(transaction_with_slot) { - // 如果通道满了，记录警告但不中断处理 - if e.is_full() { - log::warn!("Transaction channel is full, dropping transaction"); - } else { - // 通道已关闭，返回错误 - return Err(e.into()); - } - } - } - } - } - Ok(()) - } - - /// 启动事务处理任务 - /// - /// # 参数 - /// * `rx` - 事务接收通道 - /// * `processor` - 事务处理器 - pub fn start_transaction_processing( - mut rx: mpsc::Receiver, - processor: F, - ) -> JoinHandle<()> - where - F: Fn(TransactionWithSlot) -> Result<(), Box> - + Send - + Sync - + 'static, - { - tokio::spawn(async move { - while let Some(transaction_with_slot) = rx.next().await { - if let Err(e) = processor(transaction_with_slot) { - error!("Error processing transaction: {e:?}"); - } - } - }) - } -} diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index 356e774..860429a 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -1,10 +1,14 @@ +use futures::StreamExt; use solana_sdk::pubkey::Pubkey; use crate::common::AnyResult; -use crate::streaming::common::{EventBatchProcessor, SubscriptionHandle}; +use crate::protos::shredstream::SubscribeEntriesRequest; +use crate::streaming::common::{EventProcessor, SubscriptionHandle}; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::{Protocol, UnifiedEvent}; -use crate::streaming::shred::{ShredEventProcessor, ShredStreamHandler, TransactionWithSlot}; +use crate::streaming::shred::TransactionWithSlot; +use log::error; +use solana_entry::entry::Entry; use super::ShredStreamGrpc; @@ -29,120 +33,63 @@ impl ShredStreamGrpc { metrics_handle = self.metrics_manager.start_auto_monitoring().await; } - // 启动流处理 - let client = (*self.shredstream_client).clone(); - let (stream_task, rx) = ShredStreamHandler::start_stream_processing( - client, - self.config.backpressure.channel_size, - ) - .await?; + // 创建事件处理器 + let mut event_processor = + EventProcessor::new(self.metrics_manager.clone(), self.config.clone()); + event_processor.set_protocols_and_event_type_filter( + protocols, + event_type_filter, + self.config.backpressure.strategy, + self.config.batch.clone(), + ); - // 根据配置选择处理模式并获取事件处理任务句柄 - let event_handle = if self.config.batch.enabled { - // 批处理模式 - self.process_with_batch(rx, protocols, bot_wallet, event_type_filter, callback).await? - } else { - // 即时处理模式 - self.process_immediate(rx, protocols, bot_wallet, event_type_filter, callback).await? - }; + // 启动流处理 + let mut client = (*self.shredstream_client).clone(); + let request = tonic::Request::new(SubscribeEntriesRequest {}); + let mut stream = client.subscribe_entries(request).await?.into_inner(); + let event_processor_clone = event_processor.clone(); + let stream_task = tokio::spawn(async move { + while let Some(message) = stream.next().await { + match message { + Ok(msg) => { + if let Ok(entries) = bincode::deserialize::>(&msg.entries) { + for entry in entries { + for transaction in entry.transactions { + let transaction_with_slot = + TransactionWithSlot::new(transaction.clone(), msg.slot); + if let Err(e) = event_processor_clone + .process_shred_transaction_immediate( + transaction_with_slot, + bot_wallet, + &callback, + ) + .await + { + error!("Error handling message: {e:?}"); + break; + } + } + } + } + continue; + } + Err(error) => { + error!("Stream error: {error:?}"); + break; + } + } + } + }); // 保存订阅句柄 - let subscription_handle = SubscriptionHandle::new(stream_task, event_handle, metrics_handle); + let subscription_handle = SubscriptionHandle::new( + stream_task, + event_processor.get_event_handle(), + metrics_handle, + ); let mut handle_guard = self.subscription_handle.lock().await; *handle_guard = Some(subscription_handle); Ok(()) } - - /// 批处理模式 - async fn process_with_batch( - &self, - mut rx: futures::channel::mpsc::Receiver, - protocols: Vec, - bot_wallet: Option, - event_type_filter: Option, - callback: F, - ) -> AnyResult> - where - F: Fn(Box) + Send + Sync + 'static, - { - use futures::StreamExt; - - // 创建批处理器，将单个事件回调转换为批量回调 - let batch_callback = move |events: Vec>| { - for event in events { - callback(event); - } - }; - - let mut batch_processor = EventBatchProcessor::new( - batch_callback, - self.config.batch.batch_size, - self.config.batch.batch_timeout_ms, - ); - - // 创建事件处理器 - let event_processor = - ShredEventProcessor::new(self.metrics_manager.clone(), self.config.clone()); - - let event_handle = tokio::spawn(async move { - while let Some(transaction_with_slot) = rx.next().await { - if let Err(e) = event_processor - .process_transaction_with_batch( - transaction_with_slot, - protocols.clone(), - bot_wallet, - &mut batch_processor, - event_type_filter.clone(), - ) - .await - { - log::error!("Error processing transaction: {e:?}"); - } - } - - // 处理剩余的事件 - batch_processor.flush(); - }); - - Ok(event_handle) - } - - /// 即时处理模式 - async fn process_immediate( - &self, - mut rx: futures::channel::mpsc::Receiver, - protocols: Vec, - bot_wallet: Option, - event_type_filter: Option, - callback: F, - ) -> AnyResult> - where - F: Fn(Box) + Send + Sync + 'static, - { - use futures::StreamExt; - - // 创建事件处理器 - let event_processor = - ShredEventProcessor::new(self.metrics_manager.clone(), self.config.clone()); - - let event_handle = tokio::spawn(async move { - while let Some(transaction_with_slot) = rx.next().await { - if let Err(e) = event_processor - .process_transaction_immediate( - transaction_with_slot, - protocols.clone(), - bot_wallet, - event_type_filter.clone(), - &callback, - ) - .await - { - log::error!("Error processing transaction: {e:?}"); - } - } - }); - - Ok(event_handle) - } } diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index f88e88b..58075cd 100644 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -1,17 +1,17 @@ -use futures::{channel::mpsc, StreamExt}; +use futures::StreamExt; use log::error; use solana_sdk::pubkey::Pubkey; -use std::sync::Arc; +use std::sync::{Arc, RwLock}; use tokio::sync::Mutex; use yellowstone_grpc_proto::geyser::CommitmentLevel; use crate::common::AnyResult; use crate::streaming::common::{ - EventBatchProcessor, MetricsManager, PerformanceMetrics, StreamClientConfig, SubscriptionHandle, + EventProcessor, MetricsManager, PerformanceMetrics, StreamClientConfig, SubscriptionHandle, }; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::{Protocol, UnifiedEvent}; -use crate::streaming::grpc::{EventPretty, EventProcessor, StreamHandler, SubscriptionManager}; +use crate::streaming::grpc::{StreamHandler, SubscriptionManager}; /// 交易过滤器 pub struct TransactionFilter { @@ -30,7 +30,7 @@ pub struct YellowstoneGrpc { pub endpoint: String, pub x_token: Option, pub config: StreamClientConfig, - pub metrics: Arc>, + pub metrics: Arc>, pub subscription_manager: SubscriptionManager, pub metrics_manager: MetricsManager, pub event_processor: EventProcessor, @@ -50,7 +50,7 @@ impl YellowstoneGrpc { config: StreamClientConfig, ) -> AnyResult { let _ = rustls::crypto::ring::default_provider().install_default().ok(); - let metrics = Arc::new(Mutex::new(PerformanceMetrics::new())); + let metrics = Arc::new(RwLock::new(PerformanceMetrics::new())); let config_arc = Arc::new(config.clone()); let subscription_manager = @@ -63,7 +63,7 @@ impl YellowstoneGrpc { endpoint, x_token, config, - metrics, + metrics: metrics.clone(), subscription_manager, metrics_manager, event_processor, @@ -99,13 +99,13 @@ impl YellowstoneGrpc { } /// 获取性能指标 - pub async fn get_metrics(&self) -> PerformanceMetrics { - self.metrics_manager.get_metrics().await + pub fn get_metrics(&self) -> PerformanceMetrics { + self.metrics_manager.get_metrics() } /// 打印性能指标 - pub async fn print_metrics(&self) { - self.metrics_manager.print_metrics().await; + pub fn print_metrics(&self) { + self.metrics_manager.print_metrics(); } /// 启用或禁用性能监控 @@ -174,20 +174,24 @@ impl YellowstoneGrpc { .subscribe_with_request(transactions, accounts, commitment, event_type_filter.clone()) .await?; - // 创建通道，使用配置中的通道大小 - let (mut tx, mut rx) = mpsc::channel::(self.config.backpressure.channel_size); - // 启动流处理任务 - let backpressure_strategy = self.config.backpressure.strategy; + let mut event_processor = self.event_processor.clone(); + event_processor.set_protocols_and_event_type_filter( + protocols, + event_type_filter, + self.config.backpressure.strategy, + self.config.batch.clone(), + ); let stream_handle = tokio::spawn(async move { while let Some(message) = stream.next().await { match message { Ok(msg) => { if let Err(e) = StreamHandler::handle_stream_message( msg, - &mut tx, &mut subscribe_tx, - backpressure_strategy, + event_processor.clone(), + &callback, + bot_wallet, ) .await { @@ -203,160 +207,12 @@ impl YellowstoneGrpc { } }); - // 即时处理交易，无批处理 - let event_processor = self.event_processor.clone(); - let event_handle = tokio::spawn(async move { - while let Some(event_pretty) = rx.next().await { - if let Err(e) = event_processor - .process_event_transaction_with_metrics( - event_pretty, - &callback, - bot_wallet, - protocols.clone(), - event_type_filter.clone(), - ) - .await - { - error!("Error processing transaction: {e:?}"); - } - } - }); - // 保存订阅句柄 - let subscription_handle = - SubscriptionHandle::new(stream_handle, event_handle, metrics_handle); - let mut handle_guard = self.subscription_handle.lock().await; - *handle_guard = Some(subscription_handle); - - Ok(()) - } - - /// Advanced event subscription with batch processing and backpressure handling - /// - /// # Parameters - /// * `protocols` - List of protocols to monitor - /// * `bot_wallet` - Optional bot wallet address for filtering related transactions - /// * `transaction_filter` - Transaction filter specifying accounts to include/exclude - /// * `account_filter` - Account filter specifying accounts and owners to monitor - /// * `event_filter` - Optional event filter for further event filtering, no filtering if None - /// * `commitment` - Optional commitment level, defaults to Confirmed - /// * `callback` - Event callback function that receives parsed unified events - /// - /// # Features - /// * Batch processing for improved throughput - /// * Backpressure handling to prevent memory overflow - /// * Automatic performance monitoring (if enabled) - /// * Configurable batch size and timeout - /// - /// # Returns - /// Returns `AnyResult<()>`, `Ok(())` on success, error information on failure - pub async fn subscribe_events_advanced( - &self, - protocols: Vec, - bot_wallet: Option, - transaction_filter: TransactionFilter, - account_filter: AccountFilter, - event_type_filter: Option, - commitment: Option, - callback: F, - ) -> AnyResult<()> - where - F: Fn(Box) + Send + Sync + 'static, - { - // 如果已有活跃订阅，先停止它 - self.stop().await; - - let mut metrics_handle = None; - // 启动自动性能监控（如果启用） - if self.config.enable_metrics { - metrics_handle = self.metrics_manager.start_auto_monitoring().await; - } - - let transactions = self.subscription_manager.get_subscribe_request_filter( - transaction_filter.account_include, - transaction_filter.account_exclude, - transaction_filter.account_required, - event_type_filter.clone(), + let subscription_handle = SubscriptionHandle::new( + stream_handle, + self.event_processor.get_event_handle(), + metrics_handle, ); - let accounts = self.subscription_manager.subscribe_with_account_request( - account_filter.account, - account_filter.owner, - event_type_filter.clone(), - ); - - // Subscribe to events - let (mut subscribe_tx, mut stream) = self - .subscription_manager - .subscribe_with_request(transactions, accounts, commitment, event_type_filter.clone()) - .await?; - - // Create channel - let (mut tx, mut rx) = mpsc::channel::(self.config.backpressure.channel_size); - - // 创建批处理器，将单个事件回调转换为批量回调 - let batch_callback = move |events: Vec>| { - for event in events { - callback(event); - } - }; - - let mut batch_processor = EventBatchProcessor::new( - batch_callback, - self.config.batch.batch_size, - self.config.batch.batch_timeout_ms, - ); - - // Start task to process the stream - let backpressure_strategy = self.config.backpressure.strategy; - let stream_handle = tokio::spawn(async move { - while let Some(message) = stream.next().await { - match message { - Ok(msg) => { - if let Err(e) = StreamHandler::handle_stream_message( - msg, - &mut tx, - &mut subscribe_tx, - backpressure_strategy, - ) - .await - { - error!("Error handling message: {e:?}"); - break; - } - } - Err(error) => { - error!("Stream error: {error:?}"); - break; - } - } - } - }); - - // Process transactions with batch processing - let event_processor = self.event_processor.clone(); - let event_handle = tokio::spawn(async move { - while let Some(event_pretty) = rx.next().await { - if let Err(e) = event_processor - .process_event_transaction_with_batch( - event_pretty, - &mut batch_processor, - bot_wallet, - protocols.clone(), - event_type_filter.clone(), - ) - .await - { - error!("Error processing transaction: {e:?}"); - } - } - - // 处理剩余的事件 - batch_processor.flush(); - }); - - // 保存订阅句柄 - let subscription_handle = - SubscriptionHandle::new(stream_handle, event_handle, metrics_handle); let mut handle_guard = self.subscription_handle.lock().await; *handle_guard = Some(subscription_handle); @@ -379,14 +235,3 @@ impl Clone for YellowstoneGrpc { } } } - -// 实现 Clone trait 以支持模块间共享 -impl Clone for EventProcessor { - fn clone(&self) -> Self { - Self { - metrics_manager: self.metrics_manager.clone(), - config: self.config.clone(), - parser_cache: self.parser_cache.clone(), - } - } -} diff --git a/src/streaming/yellowstone_sub_system.rs b/src/streaming/yellowstone_sub_system.rs index ebb4aa9..4d19f95 100755 --- a/src/streaming/yellowstone_sub_system.rs +++ b/src/streaming/yellowstone_sub_system.rs @@ -9,7 +9,7 @@ use futures::{channel::mpsc, StreamExt}; use log::error; use solana_program::pubkey; use solana_sdk::{pubkey::Pubkey, transaction::VersionedTransaction}; -use solana_transaction_status::EncodedTransactionWithStatusMeta; +use solana_transaction_status::TransactionWithStatusMeta; const SYSTEM_PROGRAM_ID: Pubkey = pubkey!("11111111111111111111111111111111"); // 根据实际并发量调整通道大小，避免背压 @@ -93,20 +93,19 @@ impl YellowstoneGrpc { { match event_pretty { EventPretty::Transaction(transaction_pretty) => { - let trade_raw: EncodedTransactionWithStatusMeta = transaction_pretty.tx; - let meta = trade_raw - .meta - .as_ref() - .ok_or_else(|| anyhow::anyhow!("Missing transaction metadata"))?; + let trade_raw: TransactionWithStatusMeta = transaction_pretty.tx; + let meta = trade_raw.get_status_meta(); - if meta.err.is_some() { + if meta.is_none() { return Ok(()); } + let transaction = trade_raw.get_transaction(); + callback(SystemEvent::NewTransfer(TransferInfo { slot: transaction_pretty.slot, signature: transaction_pretty.signature.to_string(), - tx: trade_raw.transaction.decode(), + tx: Some(transaction), })); } _ => {}