diff --git a/examples/parse_tx_events.rs b/examples/parse_tx_events.rs index 8b0a488..e191b80 100644 --- a/examples/parse_tx_events.rs +++ b/examples/parse_tx_events.rs @@ -96,16 +96,16 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> { Protocol::RaydiumCpmm, Protocol::RaydiumAmmV4, ]; - let parser: Arc = Arc::new(EventParser::new(protocols, None)); - parser - .parse_encoded_confirmed_transaction_with_status_meta( - signature, - transaction, - Arc::new(move |event: &UnifiedEvent| { - println!("{:?}\n", event); - }), - ) - .await?; + EventParser::parse_encoded_confirmed_transaction_with_status_meta( + &protocols, + None, + signature, + transaction, + Arc::new(move |event: &UnifiedEvent| { + println!("{:?}\n", event); + }), + ) + .await?; } Err(e) => { println!("Failed to get transaction: {}", e); diff --git a/src/streaming/common/event_processor.rs b/src/streaming/common/event_processor.rs index 33bdd36..ba5e856 100644 --- a/src/streaming/common/event_processor.rs +++ b/src/streaming/common/event_processor.rs @@ -1,208 +1,149 @@ -use std::sync::Arc; - -use solana_sdk::pubkey::Pubkey; - use crate::common::AnyResult; -use crate::streaming::common::{MetricsEventType, StreamClientConfig as ClientConfig}; +use crate::streaming::common::MetricsEventType; 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::core::event_parser::EventParser; use crate::streaming::event_parser::{core::traits::UnifiedEvent, Protocol}; use crate::streaming::grpc::{EventPretty, MetricsManager}; use crate::streaming::shred::TransactionWithSlot; -use once_cell::sync::OnceCell; +use solana_sdk::pubkey::Pubkey; +use std::sync::Arc; -pub enum EventSource { - Grpc, - Shred, +/// 创建带 metrics 统计的 callback 包装器 +/// +/// 用于 Transaction 事件处理，在调用原始 callback 的同时更新 metrics +#[inline] +fn create_metrics_callback( + callback: Arc, +) -> Arc { + Arc::new(move |event: UnifiedEvent| { + let processing_time_us = event.metadata().handle_us as f64; + callback(event); + MetricsManager::global().update_metrics( + MetricsEventType::Transaction, + 1, + processing_time_us, + ); + }) } -/// High-performance Event processor -pub struct EventProcessor { - pub(crate) config: ClientConfig, - pub(crate) parser_cache: OnceCell>, - pub(crate) protocols: Vec, - pub(crate) event_type_filter: Option, - pub(crate) callback: Option>, -} +/// Process GRPC transaction events +pub async fn process_grpc_transaction( + event_pretty: EventPretty, + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, + callback: Arc, + bot_wallet: Option, +) -> AnyResult<()> { + match event_pretty { + EventPretty::Account(account_pretty) => { + MetricsManager::global().add_account_process_count(); -impl EventProcessor { - pub fn new(config: ClientConfig) -> Self { - Self { - config, - parser_cache: OnceCell::new(), - protocols: vec![], - event_type_filter: None, - callback: None, - } - } - - pub fn set_protocols_and_event_type_filter( - &mut self, - _source: EventSource, - protocols: Vec, - event_type_filter: Option, - callback: Option>, - ) { - self.protocols = protocols; - self.event_type_filter = event_type_filter; - self.callback = callback; - let protocols_ref = &self.protocols; - let event_type_filter_ref = self.event_type_filter.as_ref(); - self.parser_cache.get_or_init(|| { - Arc::new(EventParser::new(protocols_ref.clone(), event_type_filter_ref.cloned())) - }); - } - - pub fn get_parser(&self) -> Arc { - self.parser_cache.get().unwrap().clone() - } - - fn invoke_callback(&self, event: UnifiedEvent) { - if let Some(callback) = self.callback.as_ref() { - callback(event); - } - } - - pub async fn process_grpc_transaction( - &self, - event_pretty: EventPretty, - bot_wallet: Option, - ) -> AnyResult<()> { - if self.callback.is_none() { - return Ok(()); - } - match event_pretty { - EventPretty::Account(account_pretty) => { - MetricsManager::global().add_account_process_count(); - let account_event = AccountEventParser::parse_account_event( - &self.protocols, - account_pretty, - self.event_type_filter.as_ref(), - ); - if let Some(mut event) = account_event { - let processing_time_us = event.metadata().handle_us as f64; - self.invoke_callback(event); - self.update_metrics(MetricsEventType::Account, 1, processing_time_us); - } - } - EventPretty::Transaction(transaction_pretty) => { - MetricsManager::global().add_tx_process_count(); - let slot = transaction_pretty.slot; - let signature = transaction_pretty.signature; - let block_time = transaction_pretty.block_time; - let recv_us = transaction_pretty.recv_us; - let transaction_index = transaction_pretty.transaction_index; - let grpc_tx = transaction_pretty.grpc_tx; - - let parser = self.get_parser(); - let callback = self.callback.clone().unwrap(); - let adapter_callback = Arc::new(move |mut event: UnifiedEvent| { - let processing_time_us = event.metadata().handle_us as f64; - callback(event); - MetricsManager::global().update_metrics( - MetricsEventType::Transaction, - 1, - processing_time_us, - ); - }); - - parser - .parse_grpc_transaction_owned( - grpc_tx, - signature, - Some(slot), - block_time, - recv_us, - bot_wallet, - transaction_index, - adapter_callback, - ) - .await?; - } - EventPretty::BlockMeta(block_meta_pretty) => { - MetricsManager::global().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 mut block_meta_event = CommonEventParser::generate_block_meta_event( - block_meta_pretty.slot, - block_meta_pretty.block_hash, - block_time_ms, - block_meta_pretty.recv_us, - ); - let processing_time_us = block_meta_event.metadata().handle_us as f64; - self.invoke_callback(block_meta_event); - self.update_metrics(MetricsEventType::BlockMeta, 1, processing_time_us); - } - } - - Ok(()) - } - - pub async fn process_shred_transaction( - &self, - transaction_with_slot: TransactionWithSlot, - bot_wallet: Option, - ) -> AnyResult<()> { - if self.callback.is_none() { - return Ok(()); - } - MetricsManager::global().add_tx_process_count(); - let tx = transaction_with_slot.transaction; - - let slot = transaction_with_slot.slot; - if tx.signatures.is_empty() { - return Ok(()); - } - let signature = tx.signatures[0]; - let recv_us = transaction_with_slot.recv_us; - - let parser = self.get_parser(); - let callback = self.callback.clone().unwrap(); - let adapter_callback = Arc::new(move |mut event: UnifiedEvent| { - let processing_time_us = event.metadata().handle_us as f64; - callback(event); - MetricsManager::global().update_metrics( - MetricsEventType::Transaction, - 1, - processing_time_us, + let account_event = AccountEventParser::parse_account_event( + protocols, + account_pretty, + event_type_filter, ); - }); - parser - .parse_versioned_transaction_owned( - tx, + if let Some(event) = account_event { + let processing_time_us = event.metadata().handle_us as f64; + callback(event); + update_metrics(MetricsEventType::Account, 1, processing_time_us); + } + } + EventPretty::Transaction(transaction_pretty) => { + MetricsManager::global().add_tx_process_count(); + + let slot = transaction_pretty.slot; + let signature = transaction_pretty.signature; + let block_time = transaction_pretty.block_time; + let recv_us = transaction_pretty.recv_us; + let transaction_index = transaction_pretty.transaction_index; + let grpc_tx = transaction_pretty.grpc_tx; + + let adapter_callback = create_metrics_callback(callback.clone()); + + EventParser::parse_grpc_transaction_owned( + protocols, + event_type_filter, + grpc_tx, signature, Some(slot), - None, + block_time, recv_us, bot_wallet, - None, - &[], + transaction_index, adapter_callback, ) .await?; + } + EventPretty::BlockMeta(block_meta_pretty) => { + MetricsManager::global().add_block_meta_process_count(); - Ok(()) - } + 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()); - fn update_metrics(&self, ty: MetricsEventType, count: u64, time_us: f64) { - MetricsManager::global().update_metrics(ty, count, time_us); - } -} + let block_meta_event = CommonEventParser::generate_block_meta_event( + block_meta_pretty.slot, + block_meta_pretty.block_hash, + block_time_ms, + block_meta_pretty.recv_us, + ); -impl Clone for EventProcessor { - fn clone(&self) -> Self { - Self { - config: self.config.clone(), - parser_cache: self.parser_cache.clone(), - protocols: self.protocols.clone(), - event_type_filter: self.event_type_filter.clone(), - callback: self.callback.clone(), + let processing_time_us = block_meta_event.metadata().handle_us as f64; + callback(block_meta_event); + update_metrics(MetricsEventType::BlockMeta, 1, processing_time_us); } } + + Ok(()) +} + +/// Process Shred transaction events +pub async fn process_shred_transaction( + transaction_with_slot: TransactionWithSlot, + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, + callback: Arc, + bot_wallet: Option, +) -> AnyResult<()> { + MetricsManager::global().add_tx_process_count(); + + let tx = transaction_with_slot.transaction; + let slot = transaction_with_slot.slot; + + if tx.signatures.is_empty() { + return Ok(()); + } + + let signature = tx.signatures[0]; + let recv_us = transaction_with_slot.recv_us; + + let adapter_callback = create_metrics_callback(callback); + + EventParser::parse_versioned_transaction_owned( + protocols, + event_type_filter, + tx, + signature, + Some(slot), + None, + recv_us, + bot_wallet, + None, + &[], + adapter_callback, + ) + .await?; + + Ok(()) +} + +/// Update metrics for event processing +#[inline] +fn update_metrics(ty: MetricsEventType, count: u64, time_us: f64) { + MetricsManager::global().update_metrics(ty, count, time_us); } diff --git a/src/streaming/common/metrics.rs b/src/streaming/common/metrics.rs index e3a668a..dea8bbf 100644 --- a/src/streaming/common/metrics.rs +++ b/src/streaming/common/metrics.rs @@ -104,21 +104,15 @@ impl AtomicEventMetrics { /// High-performance atomic processing time statistics #[derive(Debug)] struct AtomicProcessingTimeStats { - min_time_bits: AtomicU64, - max_time_bits: AtomicU64, - min_time_timestamp_nanos: AtomicU64, // Timestamp of min value update (nanoseconds) - max_time_timestamp_nanos: AtomicU64, // Timestamp of max value update (nanoseconds) - total_time_us: AtomicU64, // Store integer part of microseconds + last_time_bits: AtomicU64, // Last processing time (f64 as u64 bits) + total_time_us: AtomicU64, // Store integer part of microseconds total_events: AtomicU64, } impl AtomicProcessingTimeStats { const fn new_const() -> Self { Self { - min_time_bits: AtomicU64::new(f64::INFINITY.to_bits()), - max_time_bits: AtomicU64::new(0), - min_time_timestamp_nanos: AtomicU64::new(0), - max_time_timestamp_nanos: AtomicU64::new(0), + last_time_bits: AtomicU64::new(0), total_time_us: AtomicU64::new(0), total_events: AtomicU64::new(0), } @@ -129,33 +123,8 @@ impl AtomicProcessingTimeStats { fn update(&self, time_us: f64, event_count: u64) { let time_bits = time_us.to_bits(); - // Fast path: Update min value without timestamp check (checked in background task) - let mut current_min = self.min_time_bits.load(Ordering::Relaxed); - while time_bits < current_min { - match self.min_time_bits.compare_exchange_weak( - current_min, - time_bits, - Ordering::Relaxed, - Ordering::Relaxed, - ) { - Ok(_) => break, - Err(x) => current_min = x, - } - } - - // Fast path: Update max value without timestamp check (checked in background task) - let mut current_max = self.max_time_bits.load(Ordering::Relaxed); - while time_bits > current_max { - match self.max_time_bits.compare_exchange_weak( - current_max, - time_bits, - Ordering::Relaxed, - Ordering::Relaxed, - ) { - Ok(_) => break, - Err(x) => current_max = x, - } - } + // Update last processing time (simple store, no compare-exchange needed) + self.last_time_bits.store(time_bits, Ordering::Relaxed); // Update cumulative values (convert microseconds to integers to avoid floating point accumulation issues) let total_time_us_int = (time_us * event_count as f64) as u64; @@ -163,55 +132,26 @@ impl AtomicProcessingTimeStats { self.total_events.fetch_add(event_count, Ordering::Relaxed); } - /// Reset min/max if they are older than 10 seconds (called by background task) - #[inline] - fn reset_stale_min_max(&self) { - let now_nanos = - std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos() - as u64; - - // Check and reset min if stale - let min_timestamp = self.min_time_timestamp_nanos.load(Ordering::Relaxed); - if now_nanos.saturating_sub(min_timestamp) > 10_000_000_000 { - self.min_time_bits.store(f64::INFINITY.to_bits(), Ordering::Relaxed); - self.min_time_timestamp_nanos.store(now_nanos, Ordering::Relaxed); - } - - // Check and reset max if stale - let max_timestamp = self.max_time_timestamp_nanos.load(Ordering::Relaxed); - if now_nanos.saturating_sub(max_timestamp) > 10_000_000_000 { - self.max_time_bits.store(0, Ordering::Relaxed); - self.max_time_timestamp_nanos.store(now_nanos, Ordering::Relaxed); - } - } - /// Get statistics (non-blocking) #[inline] fn get_stats(&self) -> ProcessingTimeStats { - let min_bits = self.min_time_bits.load(Ordering::Relaxed); - let max_bits = self.max_time_bits.load(Ordering::Relaxed); + let last_bits = self.last_time_bits.load(Ordering::Relaxed); let total_time_us_int = self.total_time_us.load(Ordering::Relaxed); let total_events = self.total_events.load(Ordering::Relaxed); - let min_time = f64::from_bits(min_bits); - let max_time = f64::from_bits(max_bits); + let last_time = f64::from_bits(last_bits); let avg_time = if total_events > 0 { total_time_us_int as f64 / total_events as f64 } else { 0.0 }; - ProcessingTimeStats { - min_us: if min_time == f64::INFINITY { 0.0 } else { min_time }, - max_us: max_time, - avg_us: avg_time, - } + ProcessingTimeStats { last_us: last_time, avg_us: avg_time } } } /// Processing time statistics result #[derive(Debug, Clone)] pub struct ProcessingTimeStats { - pub min_us: f64, - pub max_us: f64, - pub avg_us: f64, + pub last_us: f64, // Last processing time in microseconds + pub avg_us: f64, // Average processing time in microseconds } /// Event metrics snapshot @@ -236,7 +176,7 @@ pub struct PerformanceMetrics { impl PerformanceMetrics { /// Create default performance metrics (compatibility method) pub fn new() -> Self { - let default_stats = ProcessingTimeStats { min_us: 0.0, max_us: 0.0, avg_us: 0.0 }; + let default_stats = ProcessingTimeStats { last_us: 0.0, avg_us: 0.0 }; let default_metrics = EventMetricsSnapshot { process_count: 0, events_processed: 0, @@ -384,12 +324,6 @@ impl MetricsManager { GLOBAL_METRICS.update_window_metrics(EventType::Account, window_duration_nanos); GLOBAL_METRICS .update_window_metrics(EventType::BlockMeta, window_duration_nanos); - - // Reset stale min/max values (10 second expiry) - moved from hot path - GLOBAL_METRICS.processing_stats.reset_stale_min_max(); - for event_metric in &GLOBAL_METRICS.event_metrics { - event_metric.processing_stats.reset_stale_min_max(); - } } }); } @@ -467,24 +401,23 @@ impl MetricsManager { } // 打印事件指标表格（包含处理时间统计） - println!("┌─────────────┬──────────────┬──────────────────┬─────────────┬─────────────┬─────────────┐"); - println!("│ Event Type │ Process Count│ Events Processed │ Avg Time(μs)│ Min 10s(μs) │ Max 10s(μs) │"); - println!("├─────────────┼──────────────┼──────────────────┼─────────────┼─────────────┼─────────────┤"); + println!("┌─────────────┬──────────────┬──────────────────┬─────────────┬─────────────┐"); + println!("│ Event Type │ Process Count│ Events Processed │ Last(μs) │ Avg(μs) │"); + println!("├─────────────┼──────────────┼──────────────────┼─────────────┼─────────────┤"); for event_type in [EventType::Transaction, EventType::Account, EventType::BlockMeta] { let metrics = self.get_event_metrics(event_type); println!( - "│ {:11} │ {:12} │ {:16} │ {:9.2} │ {:9.2} │ {:9.2} │", + "│ {:11} │ {:12} │ {:16} │ {:9.2} │ {:9.2} │", event_type.name(), metrics.process_count, metrics.events_processed, - metrics.processing_stats.avg_us, - metrics.processing_stats.min_us, - metrics.processing_stats.max_us + metrics.processing_stats.last_us, + metrics.processing_stats.avg_us ); } - println!("└─────────────┴──────────────┴──────────────────┴─────────────┴─────────────┴─────────────┘"); + println!("└─────────────┴──────────────┴──────────────────┴─────────────┴─────────────┘"); println!(); } diff --git a/src/streaming/event_parser/common/filter.rs b/src/streaming/event_parser/common/filter.rs index 2ac0743..3c22fe5 100644 --- a/src/streaming/event_parser/common/filter.rs +++ b/src/streaming/event_parser/common/filter.rs @@ -2,7 +2,7 @@ use crate::streaming::event_parser::common::{ types::EventType, ACCOUNT_EVENT_TYPES, BLOCK_EVENT_TYPES, }; -#[derive(Debug, Clone, Default)] +#[derive(Debug, Clone, Default, PartialEq, Eq, Hash)] pub struct EventTypeFilter { pub include: Vec, } diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index c28c048..f9bd2a9 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -56,7 +56,7 @@ pub enum ProtocolType { /// Event type enumeration #[derive( - Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize, + Debug, Clone, Default, PartialEq, Eq, Hash, Serialize, Deserialize, BorshSerialize, BorshDeserialize, )] pub enum EventType { // PumpSwap events diff --git a/src/streaming/event_parser/core/account_event_parser.rs b/src/streaming/event_parser/core/account_event_parser.rs index f605b94..4f29f19 100644 --- a/src/streaming/event_parser/core/account_event_parser.rs +++ b/src/streaming/event_parser/core/account_event_parser.rs @@ -1,14 +1,9 @@ use crate::streaming::common::SimdUtils; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::common::high_performance_clock::elapsed_micros_since; -use crate::streaming::event_parser::common::{EventMetadata, EventType, ProtocolType}; +use crate::streaming::event_parser::common::EventMetadata; +use crate::streaming::event_parser::core::parser_cache::get_account_configs; use crate::streaming::event_parser::core::traits::UnifiedEvent; -use crate::streaming::event_parser::protocols::bonk::parser::BONK_PROGRAM_ID; -use crate::streaming::event_parser::protocols::pumpfun::parser::PUMPFUN_PROGRAM_ID; -use crate::streaming::event_parser::protocols::pumpswap::parser::PUMPSWAP_PROGRAM_ID; -use crate::streaming::event_parser::protocols::raydium_amm_v4::parser::RAYDIUM_AMM_V4_PROGRAM_ID; -use crate::streaming::event_parser::protocols::raydium_clmm::parser::RAYDIUM_CLMM_PROGRAM_ID; -use crate::streaming::event_parser::protocols::raydium_cpmm::parser::RAYDIUM_CPMM_PROGRAM_ID; use crate::streaming::event_parser::Protocol; use crate::streaming::grpc::AccountPretty; use serde::{Deserialize, Serialize}; @@ -20,18 +15,6 @@ use spl_token_2022::{ extension::StateWithExtensions, state::{Account as Account2022, Mint as Mint2022}, }; -use std::collections::HashMap; -use std::sync::OnceLock; - -/// 通用事件解析器配置 -#[derive(Debug, Clone)] -pub struct AccountEventParseConfig { - pub program_id: Pubkey, - pub protocol_type: ProtocolType, - pub event_type: EventType, - pub account_discriminator: &'static [u8], - pub account_parser: AccountEventParserFn, -} /// 通用账户事件 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] @@ -72,188 +55,21 @@ pub struct TokenInfoEvent { pub decimals: u8, } -/// 账户事件解析器 -pub type AccountEventParserFn = - fn(account: &AccountPretty, metadata: EventMetadata) -> Option; - -static PROTOCOL_CONFIGS_CACHE: OnceLock>> = - OnceLock::new(); - -// 通用账户解析配置的静态缓存 -static COMMON_CONFIG: OnceLock = OnceLock::new(); -// Nonce account config -static NONCE_CONFIG: OnceLock = OnceLock::new(); - pub struct AccountEventParser {} impl AccountEventParser { - pub fn configs( - protocols: &[Protocol], - event_type_filter: Option<&EventTypeFilter>, - ) -> Vec { - let protocols_map = PROTOCOL_CONFIGS_CACHE.get_or_init(|| { - let mut map: HashMap> = HashMap::new(); - map.insert(Protocol::PumpSwap, vec![ - AccountEventParseConfig { - program_id: PUMPSWAP_PROGRAM_ID, - protocol_type: ProtocolType::PumpSwap, - event_type: EventType::AccountPumpSwapGlobalConfig, - account_discriminator: crate::streaming::event_parser::protocols::pumpswap::discriminators::GLOBAL_CONFIG_ACCOUNT, - account_parser: crate::streaming::event_parser::protocols::pumpswap::types::global_config_parser, - }, - AccountEventParseConfig { - program_id: PUMPSWAP_PROGRAM_ID, - protocol_type: ProtocolType::PumpSwap, - event_type: EventType::AccountPumpSwapPool, - account_discriminator: crate::streaming::event_parser::protocols::pumpswap::discriminators::POOL_ACCOUNT, - account_parser: crate::streaming::event_parser::protocols::pumpswap::types::pool_parser, - }, - ]); - map.insert(Protocol::PumpFun, vec![ - AccountEventParseConfig { - program_id: PUMPFUN_PROGRAM_ID, - protocol_type: ProtocolType::PumpFun, - event_type: EventType::AccountPumpFunBondingCurve, - account_discriminator: crate::streaming::event_parser::protocols::pumpfun::discriminators::BONDING_CURVE_ACCOUNT, - account_parser: crate::streaming::event_parser::protocols::pumpfun::types::bonding_curve_parser, - }, - AccountEventParseConfig { - program_id: PUMPFUN_PROGRAM_ID, - protocol_type: ProtocolType::PumpFun, - event_type: EventType::AccountPumpFunGlobal, - account_discriminator: crate::streaming::event_parser::protocols::pumpfun::discriminators::GLOBAL_ACCOUNT, - account_parser: crate::streaming::event_parser::protocols::pumpfun::types::global_parser, - }, - ]); - map.insert(Protocol::Bonk, vec![ - AccountEventParseConfig { - program_id: BONK_PROGRAM_ID, - protocol_type: ProtocolType::Bonk, - event_type: EventType::AccountBonkPoolState, - account_discriminator: crate::streaming::event_parser::protocols::bonk::discriminators::POOL_STATE_ACCOUNT, - account_parser: crate::streaming::event_parser::protocols::bonk::types::pool_state_parser, - }, - AccountEventParseConfig { - program_id: BONK_PROGRAM_ID, - protocol_type: ProtocolType::Bonk, - event_type: EventType::AccountBonkGlobalConfig, - account_discriminator: crate::streaming::event_parser::protocols::bonk::discriminators::GLOBAL_CONFIG_ACCOUNT, - account_parser: crate::streaming::event_parser::protocols::bonk::types::global_config_parser, - }, - AccountEventParseConfig { - program_id: BONK_PROGRAM_ID, - protocol_type: ProtocolType::Bonk, - event_type: EventType::AccountBonkPlatformConfig, - account_discriminator: crate::streaming::event_parser::protocols::bonk::discriminators::PLATFORM_CONFIG_ACCOUNT, - account_parser: crate::streaming::event_parser::protocols::bonk::types::platform_config_parser, - }, - ]); - map.insert(Protocol::RaydiumCpmm, vec![ - AccountEventParseConfig { - program_id: RAYDIUM_CPMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumCpmm, - event_type: EventType::AccountRaydiumCpmmAmmConfig, - account_discriminator: crate::streaming::event_parser::protocols::raydium_cpmm::discriminators::AMM_CONFIG, - account_parser: crate::streaming::event_parser::protocols::raydium_cpmm::types::amm_config_parser, - }, - AccountEventParseConfig { - program_id: RAYDIUM_CPMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumCpmm, - event_type: EventType::AccountRaydiumCpmmPoolState, - account_discriminator: crate::streaming::event_parser::protocols::raydium_cpmm::discriminators::POOL_STATE, - account_parser: crate::streaming::event_parser::protocols::raydium_cpmm::types::pool_state_parser, - }, - ]); - map.insert(Protocol::RaydiumClmm, vec![ - AccountEventParseConfig { - program_id: RAYDIUM_CLMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumClmm, - event_type: EventType::AccountRaydiumClmmAmmConfig, - account_discriminator: crate::streaming::event_parser::protocols::raydium_clmm::discriminators::AMM_CONFIG, - account_parser: crate::streaming::event_parser::protocols::raydium_clmm::types::amm_config_parser, - }, - AccountEventParseConfig { - program_id: RAYDIUM_CLMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumClmm, - event_type: EventType::AccountRaydiumClmmPoolState, - account_discriminator: crate::streaming::event_parser::protocols::raydium_clmm::discriminators::POOL_STATE, - account_parser: crate::streaming::event_parser::protocols::raydium_clmm::types::pool_state_parser, - }, - AccountEventParseConfig { - program_id: RAYDIUM_CLMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumClmm, - event_type: EventType::AccountRaydiumClmmTickArrayState, - account_discriminator: crate::streaming::event_parser::protocols::raydium_clmm::discriminators::TICK_ARRAY_STATE, - account_parser: crate::streaming::event_parser::protocols::raydium_clmm::types::tick_array_state_parser, - }, - ]); - map.insert(Protocol::RaydiumAmmV4, vec![ - AccountEventParseConfig { - program_id: RAYDIUM_AMM_V4_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumAmmV4, - event_type: EventType::AccountRaydiumAmmV4AmmInfo, - account_discriminator: crate::streaming::event_parser::protocols::raydium_amm_v4::discriminators::AMM_INFO, - account_parser: crate::streaming::event_parser::protocols::raydium_amm_v4::types::amm_info_parser, - }, - ]); - map - }); - - let mut configs = Vec::new(); - let empty_vec = Vec::new(); - - // 预估容量以减少重新分配 - let estimated_capacity = protocols.len() * 3; // 大多数协议有2-3个配置 - configs.reserve(estimated_capacity); - - for protocol in protocols { - let protocol_configs = protocols_map.get(protocol).unwrap_or(&empty_vec); - // 如果没有过滤器，直接扩展所有配置 - if event_type_filter.is_none() { - configs.extend(protocol_configs.iter().cloned()); - } else { - // 有过滤器时才进行过滤 - let filter = event_type_filter.unwrap(); - configs.extend( - protocol_configs - .iter() - .filter(|config| filter.include.contains(&config.event_type)) - .cloned(), - ); - } - } - - if event_type_filter.is_none() - || event_type_filter.unwrap().include.contains(&EventType::NonceAccount) - { - let nonce_config = NONCE_CONFIG.get_or_init(|| AccountEventParseConfig { - program_id: Pubkey::default(), - protocol_type: ProtocolType::Common, - event_type: EventType::NonceAccount, - account_discriminator: &[1, 0, 0, 0, 1, 0, 0, 0], - account_parser: Self::parse_nonce_account_event, - }); - configs.push(nonce_config.clone()); - } - - let common_config = COMMON_CONFIG.get_or_init(|| AccountEventParseConfig { - program_id: Pubkey::default(), - protocol_type: ProtocolType::Common, - event_type: EventType::TokenAccount, - account_discriminator: &[], - account_parser: Self::parse_token_account_event, - }); - configs.push(common_config.clone()); - - configs - } - pub fn parse_account_event( protocols: &[Protocol], account: AccountPretty, event_type_filter: Option<&EventTypeFilter>, ) -> Option { - let configs = Self::configs(protocols, event_type_filter); + // 直接从 parser_cache 获取配置 + let configs = get_account_configs( + protocols, + event_type_filter, + Self::parse_nonce_account_event, + Self::parse_token_account_event, + ); for config in configs { if config.program_id == Pubkey::default() || (account.owner == config.program_id @@ -275,7 +91,9 @@ impl AccountEventParser { ..Default::default() }, ); - return event; + if event.is_some() { + return event; + } } } None diff --git a/src/streaming/event_parser/core/event_parser.rs b/src/streaming/event_parser/core/event_parser.rs index 11251b2..90436cf 100644 --- a/src/streaming/event_parser/core/event_parser.rs +++ b/src/streaming/event_parser/core/event_parser.rs @@ -5,18 +5,17 @@ use crate::streaming::{ filter::EventTypeFilter, high_performance_clock::{elapsed_micros_since, get_high_perf_clock}, parse_swap_data_from_next_grpc_instructions, parse_swap_data_from_next_instructions, - EventMetadata, EventType, ProtocolType, + EventMetadata, }, - core::global_state::{ - add_bonk_dev_address, add_dev_address, is_bonk_dev_address_in_signature, - is_dev_address_in_signature, - }, - protocols::{ - bonk::parser::BONK_PROGRAM_ID, pumpfun::parser::PUMPFUN_PROGRAM_ID, - pumpswap::parser::PUMPSWAP_PROGRAM_ID, - raydium_amm_v4::parser::RAYDIUM_AMM_V4_PROGRAM_ID, - raydium_clmm::parser::RAYDIUM_CLMM_PROGRAM_ID, - raydium_cpmm::parser::RAYDIUM_CPMM_PROGRAM_ID, + core::{ + global_state::{ + add_bonk_dev_address, add_dev_address, is_bonk_dev_address_in_signature, + is_dev_address_in_signature, + }, + parser_cache::{ + build_account_pubkeys_with_cache, get_global_instruction_configs, + get_global_program_ids, GenericEventParseConfig, + }, }, Protocol, UnifiedEvent, }, @@ -29,170 +28,16 @@ use solana_sdk::{ use solana_transaction_status::{ EncodedConfirmedTransactionWithStatusMeta, InnerInstruction, InnerInstructions, UiInstruction, }; -use std::{ - collections::HashMap, - sync::{Arc, LazyLock}, -}; +use std::sync::Arc; use yellowstone_grpc_proto::geyser::SubscribeUpdateTransactionInfo; -/// 高性能账户公钥缓存，避免重复Vec分配 -#[derive(Debug)] -pub struct AccountPubkeyCache { - /// 预分配的账户公钥向量，避免每次重新分配 - cache: Vec, -} - -impl AccountPubkeyCache { - /// 创建新的账户公钥缓存 - pub fn new() -> Self { - Self { - cache: Vec::with_capacity(32), // 预分配32个位置，覆盖大多数交易 - } - } - - /// 从指令账户索引构建账户公钥向量，重用缓存内存 - #[inline] - pub fn build_account_pubkeys( - &mut self, - instruction_accounts: &[u8], - all_accounts: &[Pubkey], - ) -> &[Pubkey] { - self.cache.clear(); - - // 确保容量足够，避免动态扩容 - if self.cache.capacity() < instruction_accounts.len() { - self.cache.reserve(instruction_accounts.len() - self.cache.capacity()); - } - - // 快速填充账户公钥 - for &idx in instruction_accounts.iter() { - if (idx as usize) < all_accounts.len() { - self.cache.push(all_accounts[idx as usize]); - } - } - - &self.cache - } -} - -impl Default for AccountPubkeyCache { - fn default() -> Self { - Self::new() - } -} - -/// 内联指令事件解析器 -pub type InnerInstructionEventParser = - fn(data: &[u8], metadata: EventMetadata) -> Option; - -/// 指令事件解析器 -pub type InstructionEventParser = - fn(data: &[u8], accounts: &[Pubkey], metadata: EventMetadata) -> Option; - -/// 通用事件解析器配置 -#[derive(Debug, Clone)] -pub struct GenericEventParseConfig { - pub program_id: Pubkey, - pub protocol_type: ProtocolType, - pub inner_instruction_discriminator: &'static [u8], - pub instruction_discriminator: &'static [u8], - pub event_type: EventType, - pub inner_instruction_parser: Option, - pub instruction_parser: Option, - pub requires_inner_instruction: bool, -} - -pub static EVENT_PARSERS: LazyLock> = - LazyLock::new(|| { - // 预分配容量，避免动态扩容 - let mut parsers: HashMap = - HashMap::with_capacity(6); - parsers.insert( - Protocol::PumpSwap, - ( - PUMPSWAP_PROGRAM_ID, - crate::streaming::event_parser::protocols::pumpswap::parser::CONFIGS, - ), - ); - parsers.insert( - Protocol::PumpFun, - ( - PUMPFUN_PROGRAM_ID, - crate::streaming::event_parser::protocols::pumpfun::parser::CONFIGS, - ), - ); - parsers.insert( - Protocol::Bonk, - (BONK_PROGRAM_ID, crate::streaming::event_parser::protocols::bonk::parser::CONFIGS), - ); - parsers.insert( - Protocol::RaydiumCpmm, - ( - RAYDIUM_CPMM_PROGRAM_ID, - crate::streaming::event_parser::protocols::raydium_cpmm::parser::CONFIGS, - ), - ); - parsers.insert( - Protocol::RaydiumClmm, - ( - RAYDIUM_CLMM_PROGRAM_ID, - crate::streaming::event_parser::protocols::raydium_clmm::parser::CONFIGS, - ), - ); - parsers.insert( - Protocol::RaydiumAmmV4, - ( - RAYDIUM_AMM_V4_PROGRAM_ID, - crate::streaming::event_parser::protocols::raydium_amm_v4::parser::CONFIGS, - ), - ); - parsers - }); - -/// 通用事件解析器基类 -pub struct EventParser { - pub program_ids: Vec, - // pub inner_instruction_configs: HashMap, Vec>, - pub instruction_configs: HashMap, Vec>, - /// 账户公钥缓存，避免重复分配 - pub account_cache: parking_lot::Mutex, -} +pub struct EventParser {} impl EventParser { - pub fn new(protocols: Vec, event_type_filter: Option) -> Self { - let mut instruction_configs = HashMap::with_capacity(protocols.len()); - let mut program_ids = Vec::with_capacity(protocols.len()); - // Configure all event types - for protocol in protocols { - let parse = EVENT_PARSERS.get(&protocol).unwrap(); - // Merge instruction_configs, append configurations to existing Vec - parse - .1 - .iter() - .filter(|config| { - event_type_filter - .as_ref() - .map(|filter| filter.include.contains(&config.event_type)) - .unwrap_or(true) - }) - .for_each(|config| { - instruction_configs - .entry(config.instruction_discriminator.to_vec()) - .or_insert_with(Vec::new) - .push(config.clone()); - }); - - // Append program_ids (this is already appending) - program_ids.push(parse.0); - } - let account_cache = parking_lot::Mutex::new(AccountPubkeyCache::new()); - - Self { program_ids, instruction_configs, account_cache } - } - #[allow(clippy::too_many_arguments)] async fn parse_instruction_events_from_grpc_transaction( - &self, + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, compiled_instructions: &[yellowstone_grpc_proto::prelude::CompiledInstruction], signature: Signature, slot: Option, @@ -207,7 +52,9 @@ impl EventParser { // 获取交易的指令和账户 let mut accounts = accounts.to_vec(); // 检查交易中是否包含程序 - let has_program = accounts.iter().any(|account| self.should_handle(account)); + let has_program = accounts + .iter() + .any(|account| Self::should_handle(protocols, event_type_filter, account)); if has_program { // 解析每个指令 for (index, instruction) in compiled_instructions.iter().enumerate() { @@ -221,8 +68,10 @@ impl EventParser { if *max_idx as usize >= accounts.len() { accounts.resize(*max_idx as usize + 1, Pubkey::default()); } - if self.should_handle(&program_id) { - self.parse_events_from_grpc_instruction( + if Self::should_handle(protocols, event_type_filter, &program_id) { + Self::parse_events_from_grpc_instruction( + protocols, + event_type_filter, instruction, &accounts, signature, @@ -250,7 +99,9 @@ impl EventParser { accounts: inner_accounts.to_vec(), data: data.to_vec(), }; - self.parse_events_from_grpc_instruction( + Self::parse_events_from_grpc_instruction( + protocols, + event_type_filter, &instruction, &accounts, signature, @@ -275,7 +126,8 @@ impl EventParser { /// 从VersionedTransaction中解析指令事件的通用方法 #[allow(clippy::too_many_arguments)] async fn parse_instruction_events_from_versioned_transaction( - &self, + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, transaction: &VersionedTransaction, signature: Signature, slot: Option, @@ -291,7 +143,9 @@ impl EventParser { let compiled_instructions = transaction.message.instructions(); let mut accounts: Vec = accounts.to_vec(); // 检查交易中是否包含程序 - let has_program = accounts.iter().any(|account| self.should_handle(account)); + let has_program = accounts + .iter() + .any(|account| Self::should_handle(protocols, event_type_filter, account)); if has_program { // 解析每个指令 for (index, instruction) in compiled_instructions.iter().enumerate() { @@ -300,13 +154,15 @@ impl EventParser { let inner_instructions = inner_instructions .iter() .find(|inner_instruction| inner_instruction.index == index as u8); - if self.should_handle(&program_id) { + if Self::should_handle(protocols, event_type_filter, &program_id) { let max_idx = instruction.accounts.iter().max().unwrap_or(&0); // 补齐accounts(使用Pubkey::default()) if *max_idx as usize >= accounts.len() { accounts.resize(*max_idx as usize + 1, Pubkey::default()); } - self.parse_events_from_instruction( + Self::parse_events_from_instruction( + protocols, + event_type_filter, instruction, &accounts, signature, @@ -326,7 +182,9 @@ impl EventParser { for (inner_index, inner_instruction) in inner_instructions.instructions.iter().enumerate() { - self.parse_events_from_instruction( + Self::parse_events_from_instruction( + protocols, + event_type_filter, &inner_instruction.instruction, &accounts, signature, @@ -349,7 +207,8 @@ impl EventParser { } pub async fn parse_versioned_transaction_owned( - &self, + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, versioned_tx: VersionedTransaction, signature: Signature, slot: Option, @@ -364,7 +223,9 @@ impl EventParser { let adapter_callback = Arc::new(move |event: &UnifiedEvent| { callback(event.clone()); }); - self.parse_versioned_transaction( + Self::parse_versioned_transaction( + protocols, + event_type_filter, &versioned_tx, signature, slot, @@ -380,7 +241,8 @@ impl EventParser { } async fn parse_versioned_transaction( - &self, + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, versioned_tx: &VersionedTransaction, signature: Signature, slot: Option, @@ -392,7 +254,9 @@ impl EventParser { callback: Arc Fn(&'a UnifiedEvent) + Send + Sync>, ) -> anyhow::Result<()> { let accounts: Vec = versioned_tx.message.static_account_keys().to_vec(); - self.parse_instruction_events_from_versioned_transaction( + Self::parse_instruction_events_from_versioned_transaction( + protocols, + event_type_filter, versioned_tx, signature, slot, @@ -409,7 +273,8 @@ impl EventParser { } pub async fn parse_grpc_transaction_owned( - &self, + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, grpc_tx: SubscribeUpdateTransactionInfo, signature: Signature, slot: Option, @@ -424,7 +289,9 @@ impl EventParser { callback(event.clone()); }); // 调用原始方法 - self.parse_grpc_transaction( + Self::parse_grpc_transaction( + protocols, + event_type_filter, grpc_tx, signature, slot, @@ -438,7 +305,8 @@ impl EventParser { } async fn parse_grpc_transaction( - &self, + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, grpc_tx: SubscribeUpdateTransactionInfo, signature: Signature, slot: Option, @@ -458,7 +326,7 @@ impl EventParser { if let Some(meta) = grpc_tx.meta { inner_instructions = meta.inner_instructions; address_table_lookups.reserve( - meta.loaded_writable_addresses.len() + meta.loaded_writable_addresses.len(), + meta.loaded_writable_addresses.len() + meta.loaded_readonly_addresses.len(), ); let loaded_writable_addresses = meta.loaded_writable_addresses; let loaded_readonly_addresses = meta.loaded_readonly_addresses; @@ -487,7 +355,9 @@ impl EventParser { let inner_instructions_arc = Arc::new(inner_instructions); // 解析指令事件 let instructions = &message.instructions; - self.parse_instruction_events_from_grpc_transaction( + Self::parse_instruction_events_from_grpc_transaction( + protocols, + event_type_filter, &instructions, signature, slot, @@ -507,7 +377,8 @@ impl EventParser { } pub async fn parse_encoded_confirmed_transaction_with_status_meta( - &self, + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, signature: Signature, transaction: EncodedConfirmedTransactionWithStatusMeta, callback: Arc Fn(&'a UnifiedEvent) + Send + Sync>, @@ -600,7 +471,9 @@ impl EventParser { let bot_wallet = None; let transaction_index = None; // 解析指令事件 - self.parse_instruction_events_from_versioned_transaction( + Self::parse_instruction_events_from_versioned_transaction( + protocols, + event_type_filter, &versioned_tx, signature, Some(slot), @@ -620,7 +493,6 @@ impl EventParser { /// 通用的内联指令解析方法 #[allow(clippy::too_many_arguments)] fn parse_inner_instruction_event( - &self, config: &GenericEventParseConfig, data: &[u8], signature: Signature, @@ -656,7 +528,6 @@ impl EventParser { /// 通用的指令解析方法 #[allow(clippy::too_many_arguments)] fn parse_instruction_event( - &self, config: &GenericEventParseConfig, data: &[u8], account_pubkeys: &[Pubkey], @@ -693,7 +564,6 @@ impl EventParser { /// 从内联指令中解析事件数据 #[allow(clippy::too_many_arguments)] fn parse_events_from_inner_instruction( - &self, inner_instruction: &CompiledInstruction, signature: Signature, slot: u64, @@ -724,7 +594,7 @@ impl EventParser { let data = &inner_instruction.data[16..]; let mut events = Vec::new(); - if let Some(event) = self.parse_inner_instruction_event( + if let Some(event) = Self::parse_inner_instruction_event( config, data, signature, @@ -743,7 +613,6 @@ impl EventParser { /// 从内联指令中解析事件数据 #[allow(clippy::too_many_arguments)] fn parse_events_from_grpc_inner_instruction( - &self, inner_instruction: &yellowstone_grpc_proto::prelude::InnerInstruction, signature: Signature, slot: u64, @@ -774,7 +643,7 @@ impl EventParser { let data = &inner_instruction.data[16..]; let mut events = Vec::new(); - if let Some(event) = self.parse_inner_instruction_event( + if let Some(event) = Self::parse_inner_instruction_event( config, data, signature, @@ -793,7 +662,8 @@ impl EventParser { /// 从指令中解析事件 #[allow(clippy::too_many_arguments)] fn parse_events_from_instruction( - &self, + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, instruction: &CompiledInstruction, accounts: &[Pubkey], signature: Signature, @@ -807,13 +677,18 @@ impl EventParser { inner_instructions: Option<&InnerInstructions>, callback: Arc Fn(&'a UnifiedEvent) + Send + Sync>, ) -> anyhow::Result<()> { - let program_id = accounts[instruction.program_id_index as usize]; - if !self.should_handle(&program_id) { + // 添加边界检查以防止越界访问 + let program_id_index = instruction.program_id_index as usize; + if program_id_index >= accounts.len() { + return Ok(()); + } + let program_id = accounts[program_id_index]; + if !Self::should_handle(protocols, event_type_filter, &program_id) { return Ok(()); } // 一维化并行处理：将所有 (discriminator, config) 组合展开并行处理 - let all_processing_params: Vec<_> = self - .instruction_configs + let instruction_configs = get_global_instruction_configs(protocols, event_type_filter); + let all_processing_params: Vec<_> = instruction_configs .iter() .filter(|(disc, _)| { // Use SIMD-optimized data validation and discriminator matching @@ -833,18 +708,15 @@ impl EventParser { return Ok(()); } - // 使用缓存构建账户公钥列表，避免重复分配 (只需构建一次) - let account_pubkeys = { - let mut cache_guard = self.account_cache.lock(); - cache_guard.build_account_pubkeys(&instruction.accounts, accounts).to_vec() - }; + // 使用线程局部缓存构建账户公钥列表，避免重复分配 (只需构建一次) + let account_pubkeys = build_account_pubkeys_with_cache(&instruction.accounts, accounts); // 并行处理所有 (discriminator, config) 组合 let all_results: Vec<_> = all_processing_params .iter() .filter_map(|(disc, config)| { let data = &instruction.data[disc.len()..]; - self.parse_instruction_event( + Self::parse_instruction_event( config, data, &account_pubkeys, @@ -870,7 +742,7 @@ impl EventParser { let (inner_event_result, swap_data_result) = std::thread::scope(|s| { let inner_event_handle = s.spawn(|| { for inner_instruction in inner_instructions_ref.instructions.iter() { - let result = self.parse_events_from_inner_instruction( + let result = Self::parse_events_from_inner_instruction( &inner_instruction.instruction, signature, slot, @@ -881,7 +753,7 @@ impl EventParser { transaction_index, &config, ); - if result.len() > 0 { + if !result.is_empty() { return Some(result[0].clone()); } } @@ -922,7 +794,7 @@ impl EventParser { } // 设置处理时间（使用高性能时钟） event.metadata_mut().handle_us = elapsed_micros_since(recv_us); - event = process_event(event, bot_wallet); + event = Self::process_event(event, bot_wallet); callback(&event); } Ok(()) @@ -932,7 +804,8 @@ impl EventParser { /// TODO: - wait refactor #[allow(clippy::too_many_arguments)] fn parse_events_from_grpc_instruction( - &self, + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, instruction: &yellowstone_grpc_proto::prelude::CompiledInstruction, accounts: &[Pubkey], signature: Signature, @@ -946,13 +819,18 @@ impl EventParser { inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>, callback: Arc Fn(&'a UnifiedEvent) + Send + Sync>, ) -> anyhow::Result<()> { - let program_id = accounts[instruction.program_id_index as usize]; - if !self.should_handle(&program_id) { + // 添加边界检查以防止越界访问 + let program_id_index = instruction.program_id_index as usize; + if program_id_index >= accounts.len() { + return Ok(()); + } + let program_id = accounts[program_id_index]; + if !Self::should_handle(protocols, event_type_filter, &program_id) { return Ok(()); } // 一维化并行处理：将所有 (discriminator, config) 组合展开并行处理 - let all_processing_params: Vec<_> = self - .instruction_configs + let instruction_configs = get_global_instruction_configs(protocols, event_type_filter); + let all_processing_params: Vec<_> = instruction_configs .iter() .filter(|(disc, _)| { // Use SIMD-optimized data validation and discriminator matching @@ -972,18 +850,15 @@ impl EventParser { return Ok(()); } - // 使用缓存构建账户公钥列表，避免重复分配 (只需构建一次) - let account_pubkeys = { - let mut cache_guard = self.account_cache.lock(); - cache_guard.build_account_pubkeys(&instruction.accounts, accounts).to_vec() - }; + // 使用线程局部缓存构建账户公钥列表，避免重复分配 (只需构建一次) + let account_pubkeys = build_account_pubkeys_with_cache(&instruction.accounts, accounts); // 并行处理所有 (discriminator, config) 组合 let all_results: Vec<_> = all_processing_params .iter() .filter_map(|(disc, config)| { let data = &instruction.data[disc.len()..]; - self.parse_instruction_event( + Self::parse_instruction_event( config, data, &account_pubkeys, @@ -1009,7 +884,7 @@ impl EventParser { let (inner_event_result, swap_data_result) = std::thread::scope(|s| { let inner_event_handle = s.spawn(|| { for inner_instruction in inner_instructions_ref.instructions.iter() { - let result = self.parse_events_from_grpc_inner_instruction( + let result = Self::parse_events_from_grpc_inner_instruction( &inner_instruction, signature, slot, @@ -1020,7 +895,7 @@ impl EventParser { transaction_index, &config, ); - if result.len() > 0 { + if !result.is_empty() { return Some(result[0].clone()); } } @@ -1061,81 +936,90 @@ impl EventParser { } // 设置处理时间（使用高性能时钟） event.metadata_mut().handle_us = elapsed_micros_since(recv_us); - event = process_event(event, bot_wallet); + event = Self::process_event(event, bot_wallet); callback(&event); } Ok(()) } - fn should_handle(&self, program_id: &Pubkey) -> bool { - self.program_ids.contains(program_id) + fn should_handle( + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, + program_id: &Pubkey, + ) -> bool { + get_global_program_ids(protocols, event_type_filter).contains(program_id) } - // fn supported_program_ids(&self) -> Vec { - // self.program_ids.clone() - // } -} - -fn process_event(mut event: UnifiedEvent, bot_wallet: Option) -> UnifiedEvent { - let signature = event.metadata().signature; // Copy the signature to avoid borrowing issues - match event { - UnifiedEvent::PumpFunCreateTokenEvent(token_info) => { - add_dev_address(&signature, token_info.user); - if token_info.creator != Pubkey::default() && token_info.creator != token_info.user { - add_dev_address(&signature, token_info.creator); + fn process_event(mut event: UnifiedEvent, bot_wallet: Option) -> UnifiedEvent { + let signature = event.metadata().signature; // Copy the signature to avoid borrowing issues + match event { + UnifiedEvent::PumpFunCreateTokenEvent(token_info) => { + add_dev_address(&signature, token_info.user); + if token_info.creator != Pubkey::default() && token_info.creator != token_info.user + { + add_dev_address(&signature, token_info.creator); + } + UnifiedEvent::PumpFunCreateTokenEvent(token_info) } - UnifiedEvent::PumpFunCreateTokenEvent(token_info) - } - UnifiedEvent::PumpFunTradeEvent(mut trade_info) => { - if is_dev_address_in_signature(&signature, &trade_info.user) - || is_dev_address_in_signature(&signature, &trade_info.creator) - { - trade_info.is_dev_create_token_trade = true; - } else if Some(trade_info.user) == bot_wallet { - trade_info.is_bot = true; - } else { - trade_info.is_dev_create_token_trade = false; + UnifiedEvent::PumpFunTradeEvent(mut trade_info) => { + if is_dev_address_in_signature(&signature, &trade_info.user) + || is_dev_address_in_signature(&signature, &trade_info.creator) + { + trade_info.is_dev_create_token_trade = true; + } else if Some(trade_info.user) == bot_wallet { + trade_info.is_bot = true; + } else { + trade_info.is_dev_create_token_trade = false; + } + if trade_info.metadata.swap_data.is_some() { + trade_info.metadata.swap_data.as_mut().unwrap().from_amount = + if trade_info.is_buy { + trade_info.sol_amount + } else { + trade_info.token_amount + }; + trade_info.metadata.swap_data.as_mut().unwrap().to_amount = if trade_info.is_buy + { + trade_info.token_amount + } else { + trade_info.sol_amount + }; + } + UnifiedEvent::PumpFunTradeEvent(trade_info) } - if trade_info.metadata.swap_data.is_some() { - trade_info.metadata.swap_data.as_mut().unwrap().from_amount = - if trade_info.is_buy { trade_info.sol_amount } else { trade_info.token_amount }; - trade_info.metadata.swap_data.as_mut().unwrap().to_amount = - if trade_info.is_buy { trade_info.token_amount } else { trade_info.sol_amount }; + UnifiedEvent::PumpSwapBuyEvent(mut trade_info) => { + if trade_info.metadata.swap_data.is_some() { + trade_info.metadata.swap_data.as_mut().unwrap().from_amount = + trade_info.user_quote_amount_in; + trade_info.metadata.swap_data.as_mut().unwrap().to_amount = + trade_info.base_amount_out; + } + UnifiedEvent::PumpSwapBuyEvent(trade_info) } - UnifiedEvent::PumpFunTradeEvent(trade_info) - } - UnifiedEvent::PumpSwapBuyEvent(mut trade_info) => { - if trade_info.metadata.swap_data.is_some() { - trade_info.metadata.swap_data.as_mut().unwrap().from_amount = - trade_info.user_quote_amount_in; - trade_info.metadata.swap_data.as_mut().unwrap().to_amount = - trade_info.base_amount_out; + UnifiedEvent::PumpSwapSellEvent(mut trade_info) => { + if trade_info.metadata.swap_data.is_some() { + trade_info.metadata.swap_data.as_mut().unwrap().from_amount = + trade_info.base_amount_in; + trade_info.metadata.swap_data.as_mut().unwrap().to_amount = + trade_info.user_quote_amount_out; + } + UnifiedEvent::PumpSwapSellEvent(trade_info) } - UnifiedEvent::PumpSwapBuyEvent(trade_info) - } - UnifiedEvent::PumpSwapSellEvent(mut trade_info) => { - if trade_info.metadata.swap_data.is_some() { - trade_info.metadata.swap_data.as_mut().unwrap().from_amount = - trade_info.base_amount_in; - trade_info.metadata.swap_data.as_mut().unwrap().to_amount = - trade_info.user_quote_amount_out; + UnifiedEvent::BonkPoolCreateEvent(pool_info) => { + add_bonk_dev_address(&signature, pool_info.creator); + UnifiedEvent::BonkPoolCreateEvent(pool_info) } - UnifiedEvent::PumpSwapSellEvent(trade_info) - } - UnifiedEvent::BonkPoolCreateEvent(pool_info) => { - add_bonk_dev_address(&signature, pool_info.creator); - UnifiedEvent::BonkPoolCreateEvent(pool_info) - } - UnifiedEvent::BonkTradeEvent(mut trade_info) => { - if is_bonk_dev_address_in_signature(&signature, &trade_info.payer) { - trade_info.is_dev_create_token_trade = true; - } else if Some(trade_info.payer) == bot_wallet { - trade_info.is_bot = true; - } else { - trade_info.is_dev_create_token_trade = false; + UnifiedEvent::BonkTradeEvent(mut trade_info) => { + if is_bonk_dev_address_in_signature(&signature, &trade_info.payer) { + trade_info.is_dev_create_token_trade = true; + } else if Some(trade_info.payer) == bot_wallet { + trade_info.is_bot = true; + } else { + trade_info.is_dev_create_token_trade = false; + } + UnifiedEvent::BonkTradeEvent(trade_info) } - UnifiedEvent::BonkTradeEvent(trade_info) + _ => event, } - _ => event, } } diff --git a/src/streaming/event_parser/core/mod.rs b/src/streaming/event_parser/core/mod.rs index 39fda22..dcce995 100755 --- a/src/streaming/event_parser/core/mod.rs +++ b/src/streaming/event_parser/core/mod.rs @@ -1,7 +1,12 @@ pub mod account_event_parser; pub mod common_event_parser; pub mod global_state; +pub mod parser_cache; pub mod traits; pub use traits::UnifiedEvent; +pub use parser_cache::{ + AccountEventParseConfig, AccountEventParserFn, GenericEventParseConfig, + InnerInstructionEventParser, InstructionEventParser, get_account_configs, +}; pub mod event_parser; diff --git a/src/streaming/event_parser/core/parser_cache.rs b/src/streaming/event_parser/core/parser_cache.rs new file mode 100644 index 0000000..affcedd --- /dev/null +++ b/src/streaming/event_parser/core/parser_cache.rs @@ -0,0 +1,604 @@ +//! # 事件解析器配置缓存模块 +//! +//! 本模块统一管理所有事件解析器的配置和缓存，包括： +//! - 指令事件解析器（Instruction Event Parser） +//! - 账户事件解析器（Account Event Parser） +//! - 高性能缓存工具 +//! +//! ## 设计目标 +//! - **统一配置管理**：所有协议的解析器配置集中管理 +//! - **高性能缓存**：避免重复初始化和内存分配 +//! - **易于扩展**：添加新协议只需修改配置映射 + +use crate::streaming::{ + event_parser::{ + common::{filter::EventTypeFilter, EventMetadata, EventType, ProtocolType}, + protocols::{ + bonk::parser::BONK_PROGRAM_ID, + pumpfun::parser::PUMPFUN_PROGRAM_ID, + pumpswap::parser::PUMPSWAP_PROGRAM_ID, + raydium_amm_v4::parser::RAYDIUM_AMM_V4_PROGRAM_ID, + raydium_clmm::parser::RAYDIUM_CLMM_PROGRAM_ID, + raydium_cpmm::parser::RAYDIUM_CPMM_PROGRAM_ID, + }, + Protocol, UnifiedEvent, + }, + grpc::AccountPretty, +}; +use solana_sdk::pubkey::Pubkey; +use std::{ + collections::HashMap, + sync::{Arc, LazyLock, OnceLock}, +}; + +// ============================================================================ +// 第一部分：指令事件解析器配置（Instruction Event Parser） +// ============================================================================ + +/// 内联指令事件解析器函数类型 +/// +/// 用于解析内联指令（Inner Instruction）生成的事件 +pub type InnerInstructionEventParser = + fn(data: &[u8], metadata: EventMetadata) -> Option; + +/// 指令事件解析器函数类型 +/// +/// 用于解析普通指令（Instruction）生成的事件，需要账户信息 +pub type InstructionEventParser = + fn(data: &[u8], accounts: &[Pubkey], metadata: EventMetadata) -> Option; + +/// 指令事件解析器配置 +/// +/// 定义如何解析特定协议的指令事件 +#[derive(Debug, Clone)] +pub struct GenericEventParseConfig { + /// 程序ID（Program ID） + pub program_id: Pubkey, + /// 协议类型 + pub protocol_type: ProtocolType, + /// 内联指令判别器（8字节） + pub inner_instruction_discriminator: &'static [u8], + /// 普通指令判别器（8字节） + pub instruction_discriminator: &'static [u8], + /// 事件类型 + pub event_type: EventType, + /// 内联指令解析器 + pub inner_instruction_parser: Option, + /// 普通指令解析器 + pub instruction_parser: Option, + /// 是否需要内联指令数据 + pub requires_inner_instruction: bool, +} + +/// 全局指令事件解析器注册表 +/// +/// 存储所有协议的指令解析器配置，启动时初始化一次 +pub static EVENT_PARSERS: LazyLock> = + LazyLock::new(|| { + let mut parsers = HashMap::with_capacity(6); + + // 注册各协议的解析器配置 + parsers.insert( + Protocol::PumpSwap, + ( + PUMPSWAP_PROGRAM_ID, + crate::streaming::event_parser::protocols::pumpswap::parser::CONFIGS, + ), + ); + parsers.insert( + Protocol::PumpFun, + ( + PUMPFUN_PROGRAM_ID, + crate::streaming::event_parser::protocols::pumpfun::parser::CONFIGS, + ), + ); + parsers.insert( + Protocol::Bonk, + (BONK_PROGRAM_ID, crate::streaming::event_parser::protocols::bonk::parser::CONFIGS), + ); + parsers.insert( + Protocol::RaydiumCpmm, + ( + RAYDIUM_CPMM_PROGRAM_ID, + crate::streaming::event_parser::protocols::raydium_cpmm::parser::CONFIGS, + ), + ); + parsers.insert( + Protocol::RaydiumClmm, + ( + RAYDIUM_CLMM_PROGRAM_ID, + crate::streaming::event_parser::protocols::raydium_clmm::parser::CONFIGS, + ), + ); + parsers.insert( + Protocol::RaydiumAmmV4, + ( + RAYDIUM_AMM_V4_PROGRAM_ID, + crate::streaming::event_parser::protocols::raydium_amm_v4::parser::CONFIGS, + ), + ); + + parsers + }); + +// ---------------------------------------------------------------------------- +// 指令解析器缓存工具 +// ---------------------------------------------------------------------------- + +/// 缓存键：用于标识不同的协议和过滤器组合 +/// +/// 通过对协议和事件类型排序，确保相同组合生成相同的哈希键 +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct CacheKey { + /// 协议列表（已排序） + pub protocols: Vec, + /// 事件类型过滤器（可选，已排序） + pub event_types: Option>, +} + +impl CacheKey { + /// 创建新的缓存键 + /// + /// # 参数 + /// - `protocols`: 协议列表 + /// - `filter`: 事件类型过滤器 + /// + /// # 优化 + /// 使用 `sort_by_cached_key` 避免重复调用 `format!` + pub fn new(mut protocols: Vec, filter: Option<&EventTypeFilter>) -> Self { + // 排序协议列表，确保相同协议组合生成相同的key + protocols.sort_by_cached_key(|p| format!("{:?}", p)); + + let event_types = filter.map(|f| { + let mut types = f.include.clone(); + types.sort_by_cached_key(|t| format!("{:?}", t)); + types + }); + + Self { protocols, event_types } + } +} + +/// 全局程序ID缓存（使用读写锁保护） +static GLOBAL_PROGRAM_IDS_CACHE: LazyLock< + parking_lot::RwLock>>>, +> = LazyLock::new(|| parking_lot::RwLock::new(HashMap::new())); + +/// 全局指令配置缓存（使用读写锁保护） +static GLOBAL_INSTRUCTION_CONFIGS_CACHE: LazyLock< + parking_lot::RwLock, Vec>>>>, +> = LazyLock::new(|| parking_lot::RwLock::new(HashMap::new())); + +/// 获取指定协议的程序ID列表 +/// +/// # 参数 +/// - `protocols`: 协议列表 +/// - `filter`: 事件类型过滤器（可选） +/// +/// # 返回 +/// Arc包装的程序ID向量，支持零成本共享 +/// +/// # 缓存策略 +/// - 快速路径：从缓存读取（读锁） +/// - 慢速路径：构建并缓存（写锁） +pub fn get_global_program_ids( + protocols: &[Protocol], + filter: Option<&EventTypeFilter>, +) -> Arc> { + let cache_key = CacheKey::new(protocols.to_vec(), filter); + + // 快速路径：尝试读取缓存 + { + let cache = GLOBAL_PROGRAM_IDS_CACHE.read(); + if let Some(program_ids) = cache.get(&cache_key) { + return program_ids.clone(); + } + } + + // 慢速路径：构建并缓存 + let mut program_ids = Vec::with_capacity(protocols.len()); + for protocol in protocols { + if let Some(parse) = EVENT_PARSERS.get(protocol) { + program_ids.push(parse.0); + } + } + let program_ids = Arc::new(program_ids); + + // 缓存结果（写锁） + GLOBAL_PROGRAM_IDS_CACHE.write().insert(cache_key, program_ids.clone()); + + program_ids +} + +/// 获取指定协议的指令配置 +/// +/// # 参数 +/// - `protocols`: 协议列表 +/// - `filter`: 事件类型过滤器（可选） +/// +/// # 返回 +/// Arc包装的指令配置映射表，按判别器（Discriminator）索引 +/// +/// # 缓存策略 +/// - 快速路径：从缓存读取（读锁） +/// - 慢速路径：构建并缓存（写锁） +pub fn get_global_instruction_configs( + protocols: &[Protocol], + filter: Option<&EventTypeFilter>, +) -> Arc, Vec>> { + let cache_key = CacheKey::new(protocols.to_vec(), filter); + + // 快速路径：尝试读取缓存 + { + let cache = GLOBAL_INSTRUCTION_CONFIGS_CACHE.read(); + if let Some(configs) = cache.get(&cache_key) { + return configs.clone(); + } + } + + // 慢速路径：构建并缓存 + let mut instruction_configs = HashMap::with_capacity(protocols.len() * 4); + for protocol in protocols { + if let Some(parse) = EVENT_PARSERS.get(protocol) { + parse + .1 + .iter() + .filter(|config| { + filter.as_ref().map(|f| f.include.contains(&config.event_type)).unwrap_or(true) + }) + .for_each(|config| { + instruction_configs + .entry(config.instruction_discriminator.to_vec()) + .or_insert_with(Vec::new) + .push(config.clone()); + }); + } + } + let instruction_configs = Arc::new(instruction_configs); + + // 缓存结果（写锁） + GLOBAL_INSTRUCTION_CONFIGS_CACHE.write().insert(cache_key, instruction_configs.clone()); + + instruction_configs +} + +// ============================================================================ +// 第二部分：账户公钥缓存工具（Account Pubkey Cache） +// ============================================================================ + +/// 高性能账户公钥缓存 +/// +/// 通过重用内存避免重复Vec分配，提升性能 +#[derive(Debug)] +pub struct AccountPubkeyCache { + /// 预分配的账户公钥向量 + cache: Vec, +} + +impl AccountPubkeyCache { + /// 创建新的账户公钥缓存 + /// + /// 预分配32个位置，覆盖大多数交易场景 + pub fn new() -> Self { + Self { + cache: Vec::with_capacity(32), + } + } + + /// 从指令账户索引构建账户公钥向量 + /// + /// # 参数 + /// - `instruction_accounts`: 指令账户索引列表 + /// - `all_accounts`: 所有账户公钥列表 + /// + /// # 返回 + /// 账户公钥切片引用 + /// + /// # 性能优化 + /// - 重用内部缓存，避免重新分配 + /// - 仅在必要时扩容 + #[inline] + pub fn build_account_pubkeys( + &mut self, + instruction_accounts: &[u8], + all_accounts: &[Pubkey], + ) -> &[Pubkey] { + self.cache.clear(); + + // 确保容量足够，避免动态扩容 + if self.cache.capacity() < instruction_accounts.len() { + self.cache.reserve(instruction_accounts.len() - self.cache.capacity()); + } + + // 快速填充账户公钥（带边界检查） + for &idx in instruction_accounts.iter() { + if (idx as usize) < all_accounts.len() { + self.cache.push(all_accounts[idx as usize]); + } + } + + &self.cache + } +} + +impl Default for AccountPubkeyCache { + fn default() -> Self { + Self::new() + } +} + +/// 线程局部账户公钥缓存 +/// +/// 每个线程独立维护一个缓存实例，避免锁竞争 +thread_local! { + static THREAD_LOCAL_ACCOUNT_CACHE: std::cell::RefCell = + std::cell::RefCell::new(AccountPubkeyCache::new()); +} + +/// 从线程局部缓存构建账户公钥列表 +/// +/// # 参数 +/// - `instruction_accounts`: 指令账户索引列表 +/// - `all_accounts`: 所有账户公钥列表 +/// +/// # 返回 +/// 账户公钥向量 +/// +/// # 线程安全 +/// 使用线程局部存储，每个线程独立缓存 +#[inline] +pub fn build_account_pubkeys_with_cache( + instruction_accounts: &[u8], + all_accounts: &[Pubkey], +) -> Vec { + THREAD_LOCAL_ACCOUNT_CACHE.with(|cache| { + let mut cache = cache.borrow_mut(); + cache.build_account_pubkeys(instruction_accounts, all_accounts).to_vec() + }) +} + +// ============================================================================ +// 第三部分：账户事件解析器配置（Account Event Parser） +// ============================================================================ + +/// 账户事件解析器函数类型 +/// +/// 用于解析账户状态变更生成的事件 +pub type AccountEventParserFn = + fn(account: &AccountPretty, metadata: EventMetadata) -> Option; + +/// 账户事件解析器配置 +/// +/// 定义如何解析特定协议的账户事件 +#[derive(Debug, Clone)] +pub struct AccountEventParseConfig { + /// 程序ID（Program ID） + pub program_id: Pubkey, + /// 协议类型 + pub protocol_type: ProtocolType, + /// 事件类型 + pub event_type: EventType, + /// 账户判别器（Account Discriminator） + pub account_discriminator: &'static [u8], + /// 账户解析器函数 + pub account_parser: AccountEventParserFn, +} + +/// 协议账户配置缓存 +/// +/// 存储各协议的账户解析器配置 +static PROTOCOL_CONFIGS_CACHE: OnceLock>> = + OnceLock::new(); + +/// Nonce 账户配置缓存 +static NONCE_CONFIG: OnceLock = OnceLock::new(); + +/// 通用 Token 账户配置缓存 +static COMMON_CONFIG: OnceLock = OnceLock::new(); + +/// 获取账户解析器配置列表 +/// +/// # 参数 +/// - `protocols`: 协议列表 +/// - `event_type_filter`: 事件类型过滤器（可选） +/// - `nonce_parser`: Nonce账户解析器 +/// - `token_parser`: Token账户解析器 +/// +/// # 返回 +/// 账户解析器配置向量 +/// +/// # 配置组成 +/// 1. 协议特定配置（根据protocols参数） +/// 2. Nonce账户配置（通用） +/// 3. Token账户配置（通用，作为兜底） +/// +/// # 示例 +/// ```ignore +/// let configs = get_account_configs( +/// &[Protocol::PumpFun, Protocol::Bonk], +/// None, +/// parse_nonce_account, +/// parse_token_account, +/// ); +/// ``` +pub fn get_account_configs( + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, + nonce_parser: AccountEventParserFn, + token_parser: AccountEventParserFn, +) -> Vec { + // 初始化协议配置缓存（仅在首次调用时执行） + let protocols_map = PROTOCOL_CONFIGS_CACHE.get_or_init(|| { + let mut map = HashMap::new(); + + // PumpSwap 协议 + map.insert(Protocol::PumpSwap, vec![ + AccountEventParseConfig { + program_id: PUMPSWAP_PROGRAM_ID, + protocol_type: ProtocolType::PumpSwap, + event_type: EventType::AccountPumpSwapGlobalConfig, + account_discriminator: crate::streaming::event_parser::protocols::pumpswap::discriminators::GLOBAL_CONFIG_ACCOUNT, + account_parser: crate::streaming::event_parser::protocols::pumpswap::types::global_config_parser, + }, + AccountEventParseConfig { + program_id: PUMPSWAP_PROGRAM_ID, + protocol_type: ProtocolType::PumpSwap, + event_type: EventType::AccountPumpSwapPool, + account_discriminator: crate::streaming::event_parser::protocols::pumpswap::discriminators::POOL_ACCOUNT, + account_parser: crate::streaming::event_parser::protocols::pumpswap::types::pool_parser, + }, + ]); + + // PumpFun 协议 + map.insert(Protocol::PumpFun, vec![ + AccountEventParseConfig { + program_id: PUMPFUN_PROGRAM_ID, + protocol_type: ProtocolType::PumpFun, + event_type: EventType::AccountPumpFunBondingCurve, + account_discriminator: crate::streaming::event_parser::protocols::pumpfun::discriminators::BONDING_CURVE_ACCOUNT, + account_parser: crate::streaming::event_parser::protocols::pumpfun::types::bonding_curve_parser, + }, + AccountEventParseConfig { + program_id: PUMPFUN_PROGRAM_ID, + protocol_type: ProtocolType::PumpFun, + event_type: EventType::AccountPumpFunGlobal, + account_discriminator: crate::streaming::event_parser::protocols::pumpfun::discriminators::GLOBAL_ACCOUNT, + account_parser: crate::streaming::event_parser::protocols::pumpfun::types::global_parser, + }, + ]); + + // Bonk 协议 + map.insert(Protocol::Bonk, vec![ + AccountEventParseConfig { + program_id: BONK_PROGRAM_ID, + protocol_type: ProtocolType::Bonk, + event_type: EventType::AccountBonkPoolState, + account_discriminator: crate::streaming::event_parser::protocols::bonk::discriminators::POOL_STATE_ACCOUNT, + account_parser: crate::streaming::event_parser::protocols::bonk::types::pool_state_parser, + }, + AccountEventParseConfig { + program_id: BONK_PROGRAM_ID, + protocol_type: ProtocolType::Bonk, + event_type: EventType::AccountBonkGlobalConfig, + account_discriminator: crate::streaming::event_parser::protocols::bonk::discriminators::GLOBAL_CONFIG_ACCOUNT, + account_parser: crate::streaming::event_parser::protocols::bonk::types::global_config_parser, + }, + AccountEventParseConfig { + program_id: BONK_PROGRAM_ID, + protocol_type: ProtocolType::Bonk, + event_type: EventType::AccountBonkPlatformConfig, + account_discriminator: crate::streaming::event_parser::protocols::bonk::discriminators::PLATFORM_CONFIG_ACCOUNT, + account_parser: crate::streaming::event_parser::protocols::bonk::types::platform_config_parser, + }, + ]); + + // Raydium CPMM 协议 + map.insert(Protocol::RaydiumCpmm, vec![ + AccountEventParseConfig { + program_id: RAYDIUM_CPMM_PROGRAM_ID, + protocol_type: ProtocolType::RaydiumCpmm, + event_type: EventType::AccountRaydiumCpmmAmmConfig, + account_discriminator: crate::streaming::event_parser::protocols::raydium_cpmm::discriminators::AMM_CONFIG, + account_parser: crate::streaming::event_parser::protocols::raydium_cpmm::types::amm_config_parser, + }, + AccountEventParseConfig { + program_id: RAYDIUM_CPMM_PROGRAM_ID, + protocol_type: ProtocolType::RaydiumCpmm, + event_type: EventType::AccountRaydiumCpmmPoolState, + account_discriminator: crate::streaming::event_parser::protocols::raydium_cpmm::discriminators::POOL_STATE, + account_parser: crate::streaming::event_parser::protocols::raydium_cpmm::types::pool_state_parser, + }, + ]); + + // Raydium CLMM 协议 + map.insert(Protocol::RaydiumClmm, vec![ + AccountEventParseConfig { + program_id: RAYDIUM_CLMM_PROGRAM_ID, + protocol_type: ProtocolType::RaydiumClmm, + event_type: EventType::AccountRaydiumClmmAmmConfig, + account_discriminator: crate::streaming::event_parser::protocols::raydium_clmm::discriminators::AMM_CONFIG, + account_parser: crate::streaming::event_parser::protocols::raydium_clmm::types::amm_config_parser, + }, + AccountEventParseConfig { + program_id: RAYDIUM_CLMM_PROGRAM_ID, + protocol_type: ProtocolType::RaydiumClmm, + event_type: EventType::AccountRaydiumClmmPoolState, + account_discriminator: crate::streaming::event_parser::protocols::raydium_clmm::discriminators::POOL_STATE, + account_parser: crate::streaming::event_parser::protocols::raydium_clmm::types::pool_state_parser, + }, + AccountEventParseConfig { + program_id: RAYDIUM_CLMM_PROGRAM_ID, + protocol_type: ProtocolType::RaydiumClmm, + event_type: EventType::AccountRaydiumClmmTickArrayState, + account_discriminator: crate::streaming::event_parser::protocols::raydium_clmm::discriminators::TICK_ARRAY_STATE, + account_parser: crate::streaming::event_parser::protocols::raydium_clmm::types::tick_array_state_parser, + }, + ]); + + // Raydium AMM V4 协议 + map.insert(Protocol::RaydiumAmmV4, vec![ + AccountEventParseConfig { + program_id: RAYDIUM_AMM_V4_PROGRAM_ID, + protocol_type: ProtocolType::RaydiumAmmV4, + event_type: EventType::AccountRaydiumAmmV4AmmInfo, + account_discriminator: crate::streaming::event_parser::protocols::raydium_amm_v4::discriminators::AMM_INFO, + account_parser: crate::streaming::event_parser::protocols::raydium_amm_v4::types::amm_info_parser, + }, + ]); + + map + }); + + // 构建配置列表 + let mut configs = Vec::new(); + let empty_vec = Vec::new(); + + // 预估容量（大多数协议有2-3个配置） + let estimated_capacity = protocols.len() * 3; + configs.reserve(estimated_capacity); + + // 1. 添加协议特定配置 + for protocol in protocols { + let protocol_configs = protocols_map.get(protocol).unwrap_or(&empty_vec); + + if event_type_filter.is_none() { + // 无过滤器：添加所有配置 + configs.extend(protocol_configs.iter().cloned()); + } else { + // 有过滤器：仅添加匹配的配置 + let filter = event_type_filter.unwrap(); + configs.extend( + protocol_configs + .iter() + .filter(|config| filter.include.contains(&config.event_type)) + .cloned(), + ); + } + } + + // 2. 添加 Nonce 账户配置（通用配置） + if event_type_filter.is_none() + || event_type_filter.unwrap().include.contains(&EventType::NonceAccount) + { + let nonce_config = NONCE_CONFIG.get_or_init(|| AccountEventParseConfig { + program_id: Pubkey::default(), + protocol_type: ProtocolType::Common, + event_type: EventType::NonceAccount, + account_discriminator: &[1, 0, 0, 0, 1, 0, 0, 0], + account_parser: nonce_parser, + }); + configs.push(nonce_config.clone()); + } + + // 3. 添加通用 Token 账户配置（兜底配置，始终添加） + let common_config = COMMON_CONFIG.get_or_init(|| AccountEventParseConfig { + program_id: Pubkey::default(), + protocol_type: ProtocolType::Common, + event_type: EventType::TokenAccount, + account_discriminator: &[], + account_parser: token_parser, + }); + configs.push(common_config.clone()); + + configs +} diff --git a/src/streaming/event_parser/protocols/bonk/parser.rs b/src/streaming/event_parser/protocols/bonk/parser.rs index 4ef974d..fc22718 100755 --- a/src/streaming/event_parser/protocols/bonk/parser.rs +++ b/src/streaming/event_parser/protocols/bonk/parser.rs @@ -2,7 +2,7 @@ use solana_sdk::pubkey::Pubkey; use crate::streaming::event_parser::{ common::{utils::*, EventMetadata, EventType, ProtocolType}, - core::event_parser::GenericEventParseConfig, + core::GenericEventParseConfig, protocols::bonk::{ bonk_pool_create_event_log_decode, bonk_trade_event_log_decode, discriminators, AmmFeeOn, BonkMigrateToAmmEvent, BonkMigrateToCpswapEvent, BonkPoolCreateEvent, BonkTradeEvent, diff --git a/src/streaming/event_parser/protocols/pumpfun/parser.rs b/src/streaming/event_parser/protocols/pumpfun/parser.rs index a80d44b..b47992f 100755 --- a/src/streaming/event_parser/protocols/pumpfun/parser.rs +++ b/src/streaming/event_parser/protocols/pumpfun/parser.rs @@ -1,6 +1,6 @@ use crate::streaming::event_parser::{ common::{EventMetadata, EventType, ProtocolType}, - core::event_parser::GenericEventParseConfig, + core::GenericEventParseConfig, protocols::pumpfun::{ discriminators, pumpfun_create_token_event_log_decode, pumpfun_migrate_event_log_decode, pumpfun_trade_event_log_decode, PumpFunCreateTokenEvent, PumpFunMigrateEvent, diff --git a/src/streaming/event_parser/protocols/pumpswap/parser.rs b/src/streaming/event_parser/protocols/pumpswap/parser.rs index ee29942..5df431d 100755 --- a/src/streaming/event_parser/protocols/pumpswap/parser.rs +++ b/src/streaming/event_parser/protocols/pumpswap/parser.rs @@ -1,6 +1,6 @@ use crate::streaming::event_parser::{ common::{read_u64_le, EventMetadata, EventType, ProtocolType}, - core::event_parser::GenericEventParseConfig, + core::GenericEventParseConfig, protocols::pumpswap::{ discriminators, pump_swap_buy_event_log_decode, pump_swap_create_pool_event_log_decode, pump_swap_deposit_event_log_decode, pump_swap_sell_event_log_decode, 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 944b60f..94f1957 100755 --- a/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs @@ -1,6 +1,6 @@ use crate::streaming::event_parser::{ common::{read_u64_le, EventMetadata, EventType, ProtocolType}, - core::event_parser::GenericEventParseConfig, + core::GenericEventParseConfig, protocols::raydium_amm_v4::{ discriminators, RaydiumAmmV4DepositEvent, RaydiumAmmV4Initialize2Event, RaydiumAmmV4SwapEvent, RaydiumAmmV4WithdrawEvent, RaydiumAmmV4WithdrawPnlEvent, diff --git a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs index b786b05..77e37cd 100755 --- a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs @@ -3,7 +3,7 @@ use crate::streaming::event_parser::{ read_i32_le, read_option_bool, read_u128_le, read_u64_le, read_u8_le, EventMetadata, EventType, ProtocolType, }, - core::event_parser::GenericEventParseConfig, + core::GenericEventParseConfig, protocols::raydium_clmm::{ discriminators, RaydiumClmmClosePositionEvent, RaydiumClmmCreatePoolEvent, RaydiumClmmDecreaseLiquidityV2Event, RaydiumClmmIncreaseLiquidityV2Event, diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs index f7d5afa..5571a33 100755 --- a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs @@ -2,7 +2,7 @@ use solana_sdk::pubkey::Pubkey; use crate::streaming::event_parser::{ common::{read_u64_le, EventMetadata, EventType, ProtocolType}, - core::event_parser::GenericEventParseConfig, + core::GenericEventParseConfig, protocols::raydium_cpmm::{ discriminators, RaydiumCpmmDepositEvent, RaydiumCpmmInitializeEvent, RaydiumCpmmSwapEvent, RaydiumCpmmWithdrawEvent, diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index c92dace..272c5f6 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -5,7 +5,7 @@ use solana_sdk::pubkey::Pubkey; use crate::common::AnyResult; use crate::protos::shredstream::SubscribeEntriesRequest; -use crate::streaming::common::{EventProcessor, SubscriptionHandle}; +use crate::streaming::common::{process_shred_transaction, SubscriptionHandle}; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::common::high_performance_clock::get_high_perf_clock; use crate::streaming::event_parser::{Protocol, UnifiedEvent}; @@ -37,20 +37,14 @@ impl ShredStreamGrpc { metrics_handle = MetricsManager::global().start_auto_monitoring().await; } - // 创建事件处理器 - let mut event_processor = EventProcessor::new(self.config.clone()); - event_processor.set_protocols_and_event_type_filter( - super::common::EventSource::Shred, - protocols, - event_type_filter, - Some(Arc::new(callback)), - ); - // 启动流处理 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(); + + // Wrap callback once before the async block + let callback = Arc::new(callback); + let stream_task = tokio::spawn(async move { while let Some(message) = stream.next().await { match message { @@ -64,13 +58,15 @@ impl ShredStreamGrpc { msg.slot, get_high_perf_clock(), ); - // 直接处理，背压控制在 EventProcessor 内部处理 - if let Err(e) = event_processor_clone - .process_shred_transaction( - transaction_with_slot, - bot_wallet, - ) - .await + // Process transaction - clone Arc and Vec for each call + if let Err(e) = process_shred_transaction( + transaction_with_slot, + &protocols, + event_type_filter.as_ref(), + callback.clone(), + bot_wallet, + ) + .await { error!("Error handling message: {e:?}"); } diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index 356baf9..8147028 100644 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -1,6 +1,7 @@ use crate::common::AnyResult; use crate::streaming::common::{ - EventProcessor, MetricsManager, PerformanceMetrics, StreamClientConfig, SubscriptionHandle, + process_grpc_transaction, MetricsManager, PerformanceMetrics, StreamClientConfig, + SubscriptionHandle, }; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::{Protocol, UnifiedEvent}; @@ -41,7 +42,6 @@ pub struct YellowstoneGrpc { pub x_token: Option, pub config: StreamClientConfig, pub subscription_manager: SubscriptionManager, - pub event_processor: EventProcessor, pub subscription_handle: Arc>>, // Dynamic subscription management fields pub active_subscription: Arc, @@ -67,14 +67,12 @@ impl YellowstoneGrpc { let subscription_manager = SubscriptionManager::new(endpoint.clone(), x_token.clone(), config.clone()); MetricsManager::init(config.enable_metrics); - let event_processor = EventProcessor::new(config.clone()); Ok(Self { endpoint, x_token, config, subscription_manager, - event_processor, subscription_handle: Arc::new(Mutex::new(None)), active_subscription: Arc::new(AtomicBool::new(false)), control_tx: Arc::new(tokio::sync::Mutex::new(None)), @@ -179,14 +177,9 @@ impl YellowstoneGrpc { let (control_tx, mut control_rx) = mpsc::channel(100); *self.control_tx.lock().await = Some(control_tx); - // 启动流处理任务 - let mut event_processor = self.event_processor.clone(); - event_processor.set_protocols_and_event_type_filter( - super::common::EventSource::Grpc, - protocols, - event_type_filter, - Some(Arc::new(callback)), - ); + // Wrap callback once before the async block + let callback = Arc::new(callback); + let stream_handle = tokio::spawn(async move { loop { tokio::select! { @@ -198,12 +191,14 @@ impl YellowstoneGrpc { Some(UpdateOneof::Account(account)) => { let account_pretty = factory::create_account_pretty_pooled(account); log::debug!("Received account: {:?}", account_pretty); - if let Err(e) = event_processor - .process_grpc_transaction( - EventPretty::Account(account_pretty), - bot_wallet, - ) - .await + if let Err(e) = process_grpc_transaction( + EventPretty::Account(account_pretty), + &protocols, + event_type_filter.as_ref(), + callback.clone(), + bot_wallet, + ) + .await { error!("Error processing account event: {e:?}"); } @@ -211,12 +206,14 @@ impl YellowstoneGrpc { Some(UpdateOneof::BlockMeta(sut)) => { let block_meta_pretty = factory::create_block_meta_pretty_pooled(sut, created_at); log::debug!("Received block meta: {:?}", block_meta_pretty); - if let Err(e) = event_processor - .process_grpc_transaction( - EventPretty::BlockMeta(block_meta_pretty), - bot_wallet, - ) - .await + if let Err(e) = process_grpc_transaction( + EventPretty::BlockMeta(block_meta_pretty), + &protocols, + event_type_filter.as_ref(), + callback.clone(), + bot_wallet, + ) + .await { error!("Error processing block meta event: {e:?}"); } @@ -228,12 +225,14 @@ impl YellowstoneGrpc { transaction_pretty.signature, transaction_pretty.slot ); - if let Err(e) = event_processor - .process_grpc_transaction( - EventPretty::Transaction(transaction_pretty), - bot_wallet, - ) - .await + if let Err(e) = process_grpc_transaction( + EventPretty::Transaction(transaction_pretty), + &protocols, + event_type_filter.as_ref(), + callback.clone(), + bot_wallet, + ) + .await { error!("Error processing transaction event: {e:?}"); } @@ -352,7 +351,6 @@ impl Clone for YellowstoneGrpc { x_token: self.x_token.clone(), config: self.config.clone(), subscription_manager: self.subscription_manager.clone(), - event_processor: self.event_processor.clone(), subscription_handle: self.subscription_handle.clone(), // 共享同一个 Arc> active_subscription: self.active_subscription.clone(), control_tx: self.control_tx.clone(),