refactor: convert EventParser to static methods and optimize caching

- Replace instance-based EventParser with static method design
  - Introduce ParserCache module for protocol config caching
  - Simplify EventProcessor to pure functional handlers
  - Optimize metrics: remove min/max tracking, keep last + avg only
  - Reduce atomic operations in hot path for better performance

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<dyn Fn(UnifiedEvent) + Send + Sync>,
+) -> Arc<dyn Fn(UnifiedEvent) + Send + Sync> {
+    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<Arc<EventParser>>,
-    pub(crate) protocols: Vec<Protocol>,
-    pub(crate) event_type_filter: Option<EventTypeFilter>,
-    pub(crate) callback: Option<Arc<dyn Fn(UnifiedEvent) + Send + Sync>>,
-}
+/// Process GRPC transaction events
+pub async fn process_grpc_transaction(
+    event_pretty: EventPretty,
+    protocols: &[Protocol],
+    event_type_filter: Option<&EventTypeFilter>,
+    callback: Arc<dyn Fn(UnifiedEvent) + Send + Sync>,
+    bot_wallet: Option<Pubkey>,
+) -> 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<Protocol>,
-        event_type_filter: Option<EventTypeFilter>,
-        callback: Option<Arc<dyn Fn(UnifiedEvent) + Send + Sync>>,
-    ) {
-        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<EventParser> {
-        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<Pubkey>,
-    ) -> 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<Pubkey>,
-    ) -> 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<dyn Fn(UnifiedEvent) + Send + Sync>,
+    bot_wallet: Option<Pubkey>,
+) -> 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);
 }

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!();
     }