diff --git a/src/streaming/common/config.rs b/src/streaming/common/config.rs index ce21ec8..c314813 100644 --- a/src/streaming/common/config.rs +++ b/src/streaming/common/config.rs @@ -26,7 +26,7 @@ pub struct BackpressureConfig { impl Default for BackpressureConfig { fn default() -> Self { - Self { permits: 1, strategy: BackpressureStrategy::default() } + Self { permits: 3000, strategy: BackpressureStrategy::default() } } } diff --git a/src/streaming/common/event_processor.rs b/src/streaming/common/event_processor.rs index 7ae82f7..300dc22 100644 --- a/src/streaming/common/event_processor.rs +++ b/src/streaming/common/event_processor.rs @@ -1,6 +1,6 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; -use std::time::Instant; + use crossbeam_queue::SegQueue; use solana_sdk::pubkey::Pubkey; @@ -13,7 +13,7 @@ use crate::streaming::common::{ 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::traits::get_high_perf_clock; + use crate::streaming::event_parser::EventParser; use crate::streaming::event_parser::{ core::traits::UnifiedEvent, protocols::mutil::parser::MutilEventParser, Protocol, @@ -225,7 +225,7 @@ impl EventProcessor { .unwrap_or_else(|| chrono::Utc::now().timestamp_millis()); let block_meta_event = CommonEventParser::generate_block_meta_event( block_meta_pretty.slot, - &block_meta_pretty.block_hash, + block_meta_pretty.block_hash, block_time_ms, block_meta_pretty.program_received_time_us, ); diff --git a/src/streaming/common/metrics.rs b/src/streaming/common/metrics.rs index 402f909..f76e866 100644 --- a/src/streaming/common/metrics.rs +++ b/src/streaming/common/metrics.rs @@ -39,7 +39,8 @@ struct AtomicEventMetrics { events_processed: AtomicU64, events_in_window: AtomicU64, window_start_nanos: AtomicU64, - events_per_second_bits: AtomicU64, // Bit representation of f64 + // Processing time statistics per event type + processing_stats: AtomicProcessingTimeStats, } impl AtomicEventMetrics { @@ -49,7 +50,7 @@ impl AtomicEventMetrics { events_processed: AtomicU64::new(0), events_in_window: AtomicU64::new(0), window_start_nanos: AtomicU64::new(now_nanos), - events_per_second_bits: AtomicU64::new(0), + processing_stats: AtomicProcessingTimeStats::new(), } } @@ -76,18 +77,6 @@ impl AtomicEventMetrics { ) } - /// Atomically update events per second - #[inline] - fn update_events_per_second(&self, eps: f64) { - self.events_per_second_bits.store(eps.to_bits(), Ordering::Relaxed); - } - - /// Get events per second - #[inline] - fn get_events_per_second(&self) -> f64 { - f64::from_bits(self.events_per_second_bits.load(Ordering::Relaxed)) - } - /// Reset window count #[inline] fn reset_window(&self, new_start_nanos: u64) { @@ -99,6 +88,18 @@ impl AtomicEventMetrics { fn get_window_start(&self) -> u64 { self.window_start_nanos.load(Ordering::Relaxed) } + + /// Get processing time statistics for this event type + #[inline] + fn get_processing_stats(&self) -> ProcessingTimeStats { + self.processing_stats.get_stats() + } + + /// Update processing time statistics for this event type + #[inline] + fn update_processing_stats(&self, time_us: f64, event_count: u64) { + self.processing_stats.update(time_us, event_count); + } } /// High-performance atomic processing time statistics @@ -139,7 +140,7 @@ impl AtomicProcessingTimeStats { // Update minimum value, check time difference and reset if over 10 seconds let mut current_min = self.min_time_bits.load(Ordering::Relaxed); let min_timestamp = self.min_time_timestamp_nanos.load(Ordering::Relaxed); - + // Check if min value timestamp exceeds 10 seconds (10_000_000_000 nanoseconds) let min_time_diff_nanos = now_nanos.saturating_sub(min_timestamp); if min_time_diff_nanos > 10_000_000_000 { @@ -148,7 +149,7 @@ impl AtomicProcessingTimeStats { self.min_time_timestamp_nanos.store(now_nanos, Ordering::Relaxed); current_min = f64::INFINITY.to_bits(); } - + // If current time is less than min value, update min value and timestamp while time_bits < current_min { match self.min_time_bits.compare_exchange_weak( @@ -236,7 +237,7 @@ pub struct ProcessingTimeStats { pub struct EventMetricsSnapshot { pub process_count: u64, pub events_processed: u64, - pub events_per_second: f64, + pub processing_stats: ProcessingTimeStats, } /// Compatibility structure - complete performance metrics @@ -253,9 +254,12 @@ pub struct PerformanceMetrics { impl PerformanceMetrics { /// Create default performance metrics (compatibility method) pub fn new() -> Self { - let default_metrics = - EventMetricsSnapshot { process_count: 0, events_processed: 0, events_per_second: 0.0 }; let default_stats = ProcessingTimeStats { min_us: 0.0, max_us: 0.0, avg_us: 0.0 }; + let default_metrics = EventMetricsSnapshot { + process_count: 0, + events_processed: 0, + processing_stats: default_stats.clone(), + }; Self { uptime: std::time::Duration::ZERO, @@ -311,9 +315,9 @@ impl HighPerformanceMetrics { pub fn get_event_metrics(&self, event_type: EventType) -> EventMetricsSnapshot { let index = event_type.as_index(); let (process_count, events_processed, _) = self.event_metrics[index].get_counts(); - let events_per_second = self.calculate_real_time_eps(event_type); + let processing_stats = self.event_metrics[index].get_processing_stats(); - EventMetricsSnapshot { process_count, events_processed, events_per_second } + EventMetricsSnapshot { process_count, events_processed, processing_stats } } /// 获取处理时间统计 @@ -328,41 +332,6 @@ impl HighPerformanceMetrics { self.dropped_events_count.load(Ordering::Relaxed) } - /// 计算实时每秒事件数（非阻塞） - fn calculate_real_time_eps(&self, event_type: EventType) -> f64 { - let now_nanos = - std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos() - as u64; - - let index = event_type.as_index(); - let event_metric = &self.event_metrics[index]; - - let window_start = event_metric.get_window_start(); - let current_window_duration_secs = - (now_nanos.saturating_sub(window_start)) as f64 / 1_000_000_000.0; - let events_in_window = event_metric.events_in_window.load(Ordering::Relaxed); - - // 优先级1: 当前窗口实时数据（≥2秒且有事件） - if current_window_duration_secs >= 2.0 && events_in_window > 0 { - return events_in_window as f64 / current_window_duration_secs; - } - - // 优先级2: 上一个窗口的结果 - let stored_eps = event_metric.get_events_per_second(); - if stored_eps > 0.0 { - return stored_eps; - } - - // 优先级3: 总体平均值（≥3秒运行时间） - let total_duration_secs = self.get_uptime_seconds(); - let total_events = event_metric.events_processed.load(Ordering::Relaxed); - if total_duration_secs >= 3.0 && total_events > 0 { - return total_events as f64 / total_duration_secs; - } - - 0.0 - } - /// 更新窗口指标（后台任务调用） fn update_window_metrics(&self, event_type: EventType, window_duration_nanos: u64) { let now_nanos = @@ -374,14 +343,6 @@ impl HighPerformanceMetrics { let window_start = event_metric.get_window_start(); if now_nanos.saturating_sub(window_start) >= window_duration_nanos { - let events_in_window = event_metric.events_in_window.load(Ordering::Relaxed); - let window_duration_secs = window_duration_nanos as f64 / 1_000_000_000.0; - - if window_duration_secs > 0.001 && events_in_window > 0 { - let eps = events_in_window as f64 / window_duration_secs; - event_metric.update_events_per_second(eps); - } - event_metric.reset_window(now_nanos); } } @@ -455,10 +416,15 @@ impl MetricsManager { return; } - // 原子更新事件计数 - self.metrics.event_metrics[event_type.as_index()].add_events_processed(count); + let index = event_type.as_index(); - // 原子更新处理时间统计 + // 原子更新事件计数 + self.metrics.event_metrics[index].add_events_processed(count); + + // 原子更新该事件类型的处理时间统计 + self.metrics.event_metrics[index].update_processing_stats(processing_time_us, count); + + // 保持全局处理时间统计的兼容性 self.metrics.processing_stats.update(processing_time_us, count); } @@ -506,35 +472,25 @@ impl MetricsManager { println!("\n⚠️ Dropped Events: {}", dropped_count); } - // 打印事件指标表格 - println!("┌─────────────┬──────────────┬──────────────────┬─────────────────┐"); - println!("│ Event Type │ Process Count│ Events Processed │ Events/Second │"); - println!("├─────────────┼──────────────┼──────────────────┼─────────────────┤"); + // 打印事件指标表格（包含处理时间统计） + println!("┌─────────────┬──────────────┬──────────────────┬─────────────┬─────────────┬─────────────┐"); + println!("│ Event Type │ Process Count│ Events Processed │ Avg Time(μs)│ Min 10s(μs) │ Max 10s(μs) │"); + println!("├─────────────┼──────────────┼──────────────────┼─────────────┼─────────────┼─────────────┤"); for event_type in [EventType::Transaction, EventType::Account, EventType::BlockMeta] { let metrics = self.get_event_metrics(event_type); println!( - "│ {:11} │ {:12} │ {:16} │ {:13.2} │", + "│ {:11} │ {:12} │ {:16} │ {:9.2} │ {:9.2} │ {:9.2} │", event_type.name(), metrics.process_count, metrics.events_processed, - metrics.events_per_second + metrics.processing_stats.avg_us, + metrics.processing_stats.min_us, + metrics.processing_stats.max_us ); } - println!("└─────────────┴──────────────┴──────────────────┴─────────────────┘"); - - // 打印处理时间统计表格 - let stats = self.get_processing_stats(); - println!("\n⏱️ Processing Time Statistics"); - println!("┌───────────────────────┬─────────────┐"); - println!("│ Metric │ Value (us) │"); - println!("├───────────────────────┼─────────────┤"); - println!("│ Average │ {:9.2} │", stats.avg_us); - println!("│ Minimum within 10s │ {:9.2} │", stats.min_us); - println!("│ Maximum within 10s │ {:9.2} │", stats.max_us); - println!("└───────────────────────┴─────────────┘"); - + println!("└─────────────┴──────────────┴──────────────────┴─────────────┴─────────────┴─────────────┘"); println!(); } diff --git a/src/streaming/event_parser/common/mod.rs b/src/streaming/event_parser/common/mod.rs index 3e9d088..6784e91 100755 --- a/src/streaming/event_parser/common/mod.rs +++ b/src/streaming/event_parser/common/mod.rs @@ -12,7 +12,7 @@ macro_rules! impl_unified_event { self.metadata.event_type.clone() } - fn signature(&self) -> &str { + fn signature(&self) -> &solana_sdk::signature::Signature { &self.metadata.signature } diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index b086d8f..2daa630 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -1,7 +1,7 @@ use borsh::{BorshDeserialize, BorshSerialize}; use crossbeam_queue::ArrayQueue; use serde::{Deserialize, Serialize}; -use solana_sdk::pubkey::Pubkey; +use solana_sdk::{pubkey::Pubkey, signature::Signature}; use std::{borrow::Cow, fmt, str::FromStr, sync::Arc}; use crate::{ @@ -282,15 +282,13 @@ pub struct SwapData { pub to_mint: Pubkey, pub from_amount: u64, pub to_amount: u64, - pub description: Option, + pub description: Option>, } /// Event metadata -#[derive( - Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize, -)] +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct EventMetadata { - pub signature: Cow<'static, str>, + pub signature: Signature, pub slot: u64, pub transaction_index: Option, // 新增：交易在slot中的索引 pub block_time: i64, @@ -308,7 +306,7 @@ pub struct EventMetadata { impl EventMetadata { #[allow(clippy::too_many_arguments)] pub fn new( - signature: Cow<'static, str>, + signature: Signature, slot: u64, block_time: i64, block_time_ms: i64, @@ -413,7 +411,7 @@ pub fn parse_swap_data_from_next_instructions( }, RaydiumClmmSwapEvent => |e: RaydiumClmmSwapEvent| { user = Some(e.payer); - swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".to_string()); + swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".into()); user_from_token = Some(e.input_token_account); user_to_token = Some(e.output_token_account); from_vault = Some(e.input_vault); @@ -430,7 +428,7 @@ pub fn parse_swap_data_from_next_instructions( }, RaydiumAmmV4SwapEvent => |e: RaydiumAmmV4SwapEvent| { user = Some(e.user_source_owner); - swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".to_string()); + swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".into()); user_from_token = Some(e.user_source_token_account); user_to_token = Some(e.user_destination_token_account); from_vault = Some(e.pool_pc_token_account); @@ -453,12 +451,12 @@ pub fn parse_swap_data_from_next_instructions( break; } let data = &compiled.data; - + // 使用 SIMD 验证数据格式 if !SimdUtils::validate_data_format(data, 8) { continue; } - + let get_pubkey = |i: usize| accounts[compiled.accounts[i] as usize]; let (source, destination, amount) = match data[0] { 12 if compiled.accounts.len() >= 4 => { diff --git a/src/streaming/event_parser/core/account_event_parser.rs b/src/streaming/event_parser/core/account_event_parser.rs index 0efa7be..001b263 100644 --- a/src/streaming/event_parser/core/account_event_parser.rs +++ b/src/streaming/event_parser/core/account_event_parser.rs @@ -1,4 +1,3 @@ -use std::borrow::Cow; use std::collections::HashMap; use std::sync::OnceLock; @@ -7,7 +6,7 @@ use solana_sdk::pubkey::Pubkey; use crate::streaming::common::SimdUtils; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::common::{EventMetadata, EventType, ProtocolType}; -use crate::streaming::event_parser::core::traits::{UnifiedEvent, get_high_perf_clock}; +use crate::streaming::event_parser::core::traits::{elapsed_micros_since, 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; @@ -177,12 +176,11 @@ impl AccountEventParser { if account.owner == config.program_id && SimdUtils::fast_discriminator_match(&account.data, config.account_discriminator) { - let signature_str = Cow::Owned(account.signature.to_string()); let event = (config.account_parser)( &account, EventMetadata { slot: account.slot, - signature: signature_str, + signature: account.signature, protocol: config.protocol_type, event_type: config.event_type, program_id: config.program_id, @@ -191,9 +189,9 @@ impl AccountEventParser { }, ); if let Some(mut event) = event { - event.set_program_handle_time_consuming_us( - get_high_perf_clock().elapsed_micros_since(account.program_received_time_us), - ); + event.set_program_handle_time_consuming_us(elapsed_micros_since( + account.program_received_time_us, + )); return Some(event); } } diff --git a/src/streaming/event_parser/core/common_event_parser.rs b/src/streaming/event_parser/core/common_event_parser.rs index ef44d96..822d1e8 100644 --- a/src/streaming/event_parser/core/common_event_parser.rs +++ b/src/streaming/event_parser/core/common_event_parser.rs @@ -1,4 +1,4 @@ -use crate::streaming::event_parser::core::traits::{UnifiedEvent, get_high_perf_clock}; +use crate::streaming::event_parser::core::traits::{elapsed_micros_since, UnifiedEvent}; use crate::streaming::event_parser::protocols::block::block_meta_event::BlockMetaEvent; pub struct CommonEventParser {} @@ -6,19 +6,14 @@ pub struct CommonEventParser {} impl CommonEventParser { pub fn generate_block_meta_event( slot: u64, - block_hash: &str, + block_hash: String, block_time_ms: i64, program_received_time_us: i64, ) -> Box { - let mut block_meta_event = BlockMetaEvent::new( - slot, - block_hash.to_string(), - block_time_ms, - program_received_time_us, - ); - block_meta_event.set_program_handle_time_consuming_us( - get_high_perf_clock().elapsed_micros_since(program_received_time_us), - ); + let mut block_meta_event = + BlockMetaEvent::new(slot, block_hash, block_time_ms, program_received_time_us); + block_meta_event + .set_program_handle_time_consuming_us(elapsed_micros_since(program_received_time_us)); Box::new(block_meta_event) } } diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index 65c93de..d45e2dd 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -29,22 +29,53 @@ use crate::streaming::event_parser::{ }, }; -/// 高性能时钟管理器，减少系统调用开销 +/// 高性能时钟管理器，减少系统调用开销并最小化延迟 #[derive(Debug)] pub struct HighPerformanceClock { /// 基准时间点（程序启动时的单调时钟时间） base_instant: Instant, /// 基准时间点对应的UTC时间戳（微秒） base_timestamp_us: i64, + /// 上次校准时间（用于检测是否需要重新校准） + last_calibration: Instant, + /// 校准间隔（秒） + calibration_interval_secs: u64, } impl HighPerformanceClock { /// 创建新的高性能时钟 pub fn new() -> Self { - let base_instant = Instant::now(); - let base_timestamp_us = chrono::Utc::now().timestamp_micros(); + Self::new_with_calibration_interval(300) // 默认5分钟校准一次 + } - Self { base_instant, base_timestamp_us } + /// 创建带自定义校准间隔的高性能时钟 + pub fn new_with_calibration_interval(calibration_interval_secs: u64) -> Self { + // 通过多次采样来减少初始化误差 + let mut best_offset = i64::MAX; + let mut best_instant = Instant::now(); + let mut best_timestamp = chrono::Utc::now().timestamp_micros(); + + // 进行3次采样，选择延迟最小的 + for _ in 0..3 { + let instant_before = Instant::now(); + let timestamp = chrono::Utc::now().timestamp_micros(); + let instant_after = Instant::now(); + + let sample_latency = instant_after.duration_since(instant_before).as_nanos() as i64; + + if sample_latency < best_offset { + best_offset = sample_latency; + best_instant = instant_before; + best_timestamp = timestamp; + } + } + + Self { + base_instant: best_instant, + base_timestamp_us: best_timestamp, + last_calibration: best_instant, + calibration_interval_secs, + } } /// 获取当前时间戳（微秒），使用单调时钟计算，避免系统调用 @@ -54,11 +85,53 @@ impl HighPerformanceClock { self.base_timestamp_us + elapsed.as_micros() as i64 } + /// 获取高精度当前时间戳（微秒），在必要时进行校准 + pub fn now_micros_with_calibration(&mut self) -> i64 { + // 检查是否需要重新校准 + if self.last_calibration.elapsed().as_secs() >= self.calibration_interval_secs { + self.recalibrate(); + } + self.now_micros() + } + + /// 重新校准时钟，减少累积漂移 + fn recalibrate(&mut self) { + let current_monotonic = Instant::now(); + let current_utc = chrono::Utc::now().timestamp_micros(); + + // 计算预期的UTC时间戳（基于单调时钟） + let expected_utc = self.base_timestamp_us + + current_monotonic.duration_since(self.base_instant).as_micros() as i64; + + // 计算漂移量 + let drift_us = current_utc - expected_utc; + + // 如果漂移超过1毫秒，进行校准 + if drift_us.abs() > 1000 { + self.base_instant = current_monotonic; + self.base_timestamp_us = current_utc; + } + + self.last_calibration = current_monotonic; + } + /// 计算从指定时间戳到现在的消耗时间（微秒） #[inline(always)] pub fn elapsed_micros_since(&self, start_timestamp_us: i64) -> i64 { self.now_micros() - start_timestamp_us } + + /// 获取高精度纳秒时间戳 + #[inline(always)] + pub fn now_nanos(&self) -> i128 { + let elapsed = self.base_instant.elapsed(); + (self.base_timestamp_us as i128 * 1000) + elapsed.as_nanos() as i128 + } + + /// 重置时钟（强制重新初始化） + pub fn reset(&mut self) { + *self = Self::new_with_calibration_interval(self.calibration_interval_secs); + } } impl Default for HighPerformanceClock { @@ -67,14 +140,21 @@ impl Default for HighPerformanceClock { } } -/// 全局高性能时钟实例（使用OnceCell避免重复初始化） +/// 全局高性能时钟实例 static HIGH_PERF_CLOCK: once_cell::sync::OnceCell = once_cell::sync::OnceCell::new(); -/// 获取全局高性能时钟实例 +/// 获取全局高性能时钟实例（最简单的实现） #[inline(always)] -pub fn get_high_perf_clock() -> &'static HighPerformanceClock { - HIGH_PERF_CLOCK.get_or_init(HighPerformanceClock::new) +pub fn get_high_perf_clock() -> i64 { + let clock = HIGH_PERF_CLOCK.get_or_init(HighPerformanceClock::new); + clock.now_micros() +} + +/// 计算从指定时间戳到现在的消耗时间（微秒） +#[inline(always)] +pub fn elapsed_micros_since(start_timestamp_us: i64) -> i64 { + get_high_perf_clock() - start_timestamp_us } /// 轻量级事件包装器，避免频繁的Box分配 @@ -147,7 +227,7 @@ pub trait UnifiedEvent: Debug + Send + Sync { fn event_type(&self) -> EventType; /// Get transaction signature - fn signature(&self) -> &str; + fn signature(&self) -> &Signature; /// Get slot number fn slot(&self) -> u64; @@ -535,7 +615,7 @@ pub trait EventParser: Send + Sync { let slot = transaction.slot; let block_time = transaction.block_time.map(|t| Timestamp { seconds: t as i64, nanos: 0 }); - let program_received_time_us = chrono::Utc::now().timestamp_micros(); + let program_received_time_us = get_high_perf_clock(); let bot_wallet = None; let transaction_index = None; // 解析指令事件 @@ -714,11 +794,10 @@ impl GenericEventParser { transaction_index: Option, ) -> Option> { if let Some(parser) = config.inner_instruction_parser { - let signature_str = Cow::Owned(signature.to_string()); let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000; let metadata = EventMetadata::new( - signature_str, + signature, slot, timestamp.seconds, block_time_ms, @@ -752,11 +831,10 @@ impl GenericEventParser { transaction_index: Option, ) -> Option> { if let Some(parser) = config.instruction_parser { - let signature_str = Cow::Owned(signature.to_string()); let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000; let metadata = EventMetadata::new( - signature_str, + signature, slot, timestamp.seconds, block_time_ms, @@ -941,9 +1019,9 @@ impl EventParser for GenericEventParser { event.merge(&*inner_instruction_event); } // 设置处理时间（使用高性能时钟） - event.set_program_handle_time_consuming_us( - get_high_perf_clock().elapsed_micros_since(program_received_time_us), - ); + event.set_program_handle_time_consuming_us(elapsed_micros_since( + program_received_time_us, + )); event = process_event(event, bot_wallet); callback(&event); } diff --git a/src/streaming/event_parser/protocols/block/block_meta_event.rs b/src/streaming/event_parser/protocols/block/block_meta_event.rs index 8fe4051..392e24b 100644 --- a/src/streaming/event_parser/protocols/block/block_meta_event.rs +++ b/src/streaming/event_parser/protocols/block/block_meta_event.rs @@ -1,9 +1,8 @@ -use std::borrow::Cow; - use crate::impl_unified_event; use crate::streaming::event_parser::common::{types::EventType, EventMetadata}; use borsh::BorshDeserialize; use serde::{Deserialize, Serialize}; +use solana_sdk::signature::Signature; /// Block元数据事件 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] @@ -22,7 +21,7 @@ impl BlockMetaEvent { program_received_time_us: i64, ) -> Self { let metadata = EventMetadata::new( - Cow::Borrowed(""), + Signature::default(), slot, block_time_ms / 1000, block_time_ms, diff --git a/src/streaming/event_parser/protocols/bonk/events.rs b/src/streaming/event_parser/protocols/bonk/events.rs index cb450ae..6571664 100755 --- a/src/streaming/event_parser/protocols/bonk/events.rs +++ b/src/streaming/event_parser/protocols/bonk/events.rs @@ -237,6 +237,7 @@ impl_unified_event!( // Migrate to CP Swap event #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct BonkMigrateToCpswapEvent { + #[borsh(skip)] pub metadata: EventMetadata, pub payer: Pubkey, pub base_mint: Pubkey, @@ -276,10 +277,10 @@ impl_unified_event!(BonkMigrateToCpswapEvent,); #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct BonkPoolStateAccountEvent { pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub pool_state: PoolState, } @@ -289,10 +290,10 @@ impl_unified_event!(BonkPoolStateAccountEvent,); #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct BonkGlobalConfigAccountEvent { pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub global_config: GlobalConfig, } @@ -302,10 +303,10 @@ impl_unified_event!(BonkGlobalConfigAccountEvent,); #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct BonkPlatformConfigAccountEvent { pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub platform_config: PlatformConfig, } diff --git a/src/streaming/event_parser/protocols/bonk/types.rs b/src/streaming/event_parser/protocols/bonk/types.rs index 476f3ae..2e82b12 100755 --- a/src/streaming/event_parser/protocols/bonk/types.rs +++ b/src/streaming/event_parser/protocols/bonk/types.rs @@ -142,10 +142,10 @@ pub fn pool_state_parser( if let Some(pool_state) = pool_state_decode(&account.data[8..POOL_STATE_SIZE + 8]) { Some(Box::new(BonkPoolStateAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, pool_state, })) @@ -193,10 +193,10 @@ pub fn global_config_parser( if let Some(global_config) = global_config_decode(&account.data[8..GLOBAL_CONFIG_SIZE + 8]) { Some(Box::new(BonkGlobalConfigAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, global_config, })) @@ -241,10 +241,10 @@ pub fn platform_config_parser( { Some(Box::new(BonkPlatformConfigAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, platform_config, })) diff --git a/src/streaming/event_parser/protocols/pumpfun/events.rs b/src/streaming/event_parser/protocols/pumpfun/events.rs index 5e06a84..32212e2 100755 --- a/src/streaming/event_parser/protocols/pumpfun/events.rs +++ b/src/streaming/event_parser/protocols/pumpfun/events.rs @@ -228,10 +228,10 @@ impl_unified_event!( pub struct PumpFunBondingCurveAccountEvent { #[borsh(skip)] pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub bonding_curve: BondingCurve, } @@ -243,10 +243,10 @@ impl_unified_event!(PumpFunBondingCurveAccountEvent,); pub struct PumpFunGlobalAccountEvent { #[borsh(skip)] pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub global: Global, } diff --git a/src/streaming/event_parser/protocols/pumpfun/types.rs b/src/streaming/event_parser/protocols/pumpfun/types.rs index c2ffdf1..d419996 100644 --- a/src/streaming/event_parser/protocols/pumpfun/types.rs +++ b/src/streaming/event_parser/protocols/pumpfun/types.rs @@ -41,10 +41,10 @@ pub fn bonding_curve_parser( if let Some(bonding_curve) = bonding_curve_decode(&account.data[8..BONDING_CURVE_SIZE + 8]) { Some(Box::new(PumpFunBondingCurveAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, bonding_curve, })) @@ -91,10 +91,10 @@ pub fn global_parser( if let Some(global) = global_decode(&account.data[8..GLOBAL_SIZE + 8]) { Some(Box::new(PumpFunGlobalAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, global, })) diff --git a/src/streaming/event_parser/protocols/pumpswap/events.rs b/src/streaming/event_parser/protocols/pumpswap/events.rs index 63a85b7..c3d2322 100755 --- a/src/streaming/event_parser/protocols/pumpswap/events.rs +++ b/src/streaming/event_parser/protocols/pumpswap/events.rs @@ -366,11 +366,12 @@ impl_unified_event!( /// 全局配置 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct PumpSwapGlobalConfigAccountEvent { + #[borsh(skip)] pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub global_config: GlobalConfig, } @@ -379,11 +380,12 @@ impl_unified_event!(PumpSwapGlobalConfigAccountEvent,); /// 池 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct PumpSwapPoolAccountEvent { + #[borsh(skip)] pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub pool: Pool, } diff --git a/src/streaming/event_parser/protocols/pumpswap/types.rs b/src/streaming/event_parser/protocols/pumpswap/types.rs index b58c0dc..a9d917b 100644 --- a/src/streaming/event_parser/protocols/pumpswap/types.rs +++ b/src/streaming/event_parser/protocols/pumpswap/types.rs @@ -43,10 +43,10 @@ pub fn global_config_parser( if let Some(config) = global_config_decode(&account.data[8..GLOBAL_CONFIG_SIZE + 8]) { Some(Box::new(PumpSwapGlobalConfigAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, global_config: config, })) @@ -88,10 +88,10 @@ pub fn pool_parser( if let Some(pool) = pool_decode(&account.data[8..POOL_SIZE + 8]) { Some(Box::new(PumpSwapPoolAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, pool: pool, })) diff --git a/src/streaming/event_parser/protocols/raydium_amm_v4/events.rs b/src/streaming/event_parser/protocols/raydium_amm_v4/events.rs index cd4eb19..9f82f6f 100755 --- a/src/streaming/event_parser/protocols/raydium_amm_v4/events.rs +++ b/src/streaming/event_parser/protocols/raydium_amm_v4/events.rs @@ -9,6 +9,7 @@ use solana_sdk::pubkey::Pubkey; /// 交易 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumAmmV4SwapEvent { + #[borsh(skip)] pub metadata: EventMetadata, // base in pub amount_in: u64, @@ -42,6 +43,7 @@ impl_unified_event!(RaydiumAmmV4SwapEvent,); /// 添加流动性 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumAmmV4DepositEvent { + #[borsh(skip)] pub metadata: EventMetadata, pub max_coin_amount: u64, pub max_pc_amount: u64, @@ -67,6 +69,7 @@ impl_unified_event!(RaydiumAmmV4DepositEvent,); /// 初始化 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumAmmV4Initialize2Event { + #[borsh(skip)] pub metadata: EventMetadata, pub nonce: u8, pub open_time: u64, @@ -100,6 +103,7 @@ impl_unified_event!(RaydiumAmmV4Initialize2Event,); /// 移除流动性 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumAmmV4WithdrawEvent { + #[borsh(skip)] pub metadata: EventMetadata, pub amount: u64, @@ -131,6 +135,7 @@ impl_unified_event!(RaydiumAmmV4WithdrawEvent,); /// 提现 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumAmmV4WithdrawPnlEvent { + #[borsh(skip)] pub metadata: EventMetadata, pub token_program: Pubkey, @@ -156,11 +161,12 @@ impl_unified_event!(RaydiumAmmV4WithdrawPnlEvent,); /// 池信息 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumAmmV4AmmInfoAccountEvent { + #[borsh(skip)] pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub amm_info: AmmInfo, } diff --git a/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs b/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs index 63fac73..e2f3078 100644 --- a/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs +++ b/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs @@ -96,10 +96,10 @@ pub fn amm_info_parser( if let Some(amm_info) = amm_info_decode(&account.data[..AMM_INFO_SIZE]) { Some(Box::new(RaydiumAmmV4AmmInfoAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, amm_info: amm_info, })) diff --git a/src/streaming/event_parser/protocols/raydium_clmm/events.rs b/src/streaming/event_parser/protocols/raydium_clmm/events.rs index f8834fa..dd405cf 100755 --- a/src/streaming/event_parser/protocols/raydium_clmm/events.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/events.rs @@ -223,10 +223,10 @@ impl_unified_event!(RaydiumClmmOpenPositionV2Event,); #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct RaydiumClmmAmmConfigAccountEvent { pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub amm_config: AmmConfig, } @@ -236,10 +236,10 @@ impl_unified_event!(RaydiumClmmAmmConfigAccountEvent,); #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct RaydiumClmmPoolStateAccountEvent { pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub pool_state: PoolState, } @@ -249,10 +249,10 @@ impl_unified_event!(RaydiumClmmPoolStateAccountEvent,); #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct RaydiumClmmTickArrayStateAccountEvent { pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub tick_array_state: TickArrayState, } diff --git a/src/streaming/event_parser/protocols/raydium_clmm/types.rs b/src/streaming/event_parser/protocols/raydium_clmm/types.rs index 06cc524..c8a8803 100644 --- a/src/streaming/event_parser/protocols/raydium_clmm/types.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/types.rs @@ -47,10 +47,10 @@ pub fn amm_config_parser( if let Some(amm_config) = amm_config_decode(&account.data[8..AMM_CONFIG_SIZE + 8]) { Some(Box::new(RaydiumClmmAmmConfigAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, amm_config: amm_config, })) @@ -135,10 +135,10 @@ pub fn pool_state_parser( if let Some(pool_state) = pool_state_decode(&account.data[8..POOL_STATE_SIZE + 8]) { Some(Box::new(RaydiumClmmPoolStateAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, pool_state: pool_state, })) @@ -218,10 +218,10 @@ pub fn tick_array_state_parser( { Some(Box::new(RaydiumClmmTickArrayStateAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, tick_array_state: tick_array_state, })) diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/events.rs b/src/streaming/event_parser/protocols/raydium_cpmm/events.rs index 6260e48..d8c1b7a 100755 --- a/src/streaming/event_parser/protocols/raydium_cpmm/events.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/events.rs @@ -10,6 +10,7 @@ use solana_sdk::pubkey::Pubkey; /// 交易 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumCpmmSwapEvent { + #[borsh(skip)] pub metadata: EventMetadata, pub amount_in: u64, pub minimum_amount_out: u64, @@ -35,6 +36,7 @@ impl_unified_event!(RaydiumCpmmSwapEvent,); /// 存款 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumCpmmDepositEvent { + #[borsh(skip)] pub metadata: EventMetadata, pub lp_token_amount: u64, pub maximum_token0_amount: u64, @@ -59,6 +61,7 @@ impl_unified_event!(RaydiumCpmmDepositEvent,); /// 初始化 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumCpmmInitializeEvent { + #[borsh(skip)] pub metadata: EventMetadata, pub init_amount0: u64, pub init_amount1: u64, @@ -90,6 +93,7 @@ impl_unified_event!(RaydiumCpmmInitializeEvent,); /// 提款 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumCpmmWithdrawEvent { + #[borsh(skip)] pub metadata: EventMetadata, pub lp_token_amount: u64, pub minimum_token0_amount: u64, @@ -115,11 +119,12 @@ impl_unified_event!(RaydiumCpmmWithdrawEvent,); /// 池配置 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumCpmmAmmConfigAccountEvent { + #[borsh(skip)] pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub amm_config: AmmConfig, } @@ -128,11 +133,12 @@ impl_unified_event!(RaydiumCpmmAmmConfigAccountEvent,); /// 池状态 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumCpmmPoolStateAccountEvent { + #[borsh(skip)] pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub pool_state: PoolState, } diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/types.rs b/src/streaming/event_parser/protocols/raydium_cpmm/types.rs index 29555c7..532690f 100644 --- a/src/streaming/event_parser/protocols/raydium_cpmm/types.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/types.rs @@ -46,10 +46,10 @@ pub fn amm_config_parser( if let Some(amm_config) = amm_config_decode(&account.data[8..AMM_CONFIG_SIZE + 8]) { Some(Box::new(RaydiumCpmmAmmConfigAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, amm_config: amm_config, })) @@ -104,10 +104,10 @@ pub fn pool_state_parser( if let Some(pool_state) = pool_state_decode(&account.data[8..POOL_STATE_SIZE + 8]) { Some(Box::new(RaydiumCpmmPoolStateAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, pool_state: pool_state, })) diff --git a/src/streaming/grpc/mod.rs b/src/streaming/grpc/mod.rs index 0d91713..6618f4e 100644 --- a/src/streaming/grpc/mod.rs +++ b/src/streaming/grpc/mod.rs @@ -1,10 +1,12 @@ // gRPC 相关模块 pub mod connection; +pub mod pool; pub mod subscription; pub mod types; // 重新导出主要类型 pub use connection::*; +pub use pool::*; pub use subscription::*; pub use types::*; diff --git a/src/streaming/grpc/pool.rs b/src/streaming/grpc/pool.rs new file mode 100644 index 0000000..5371a54 --- /dev/null +++ b/src/streaming/grpc/pool.rs @@ -0,0 +1,438 @@ +use solana_sdk::{pubkey::Pubkey, signature::Signature}; +use std::collections::VecDeque; +use std::ops::DerefMut; +use std::sync::{Arc, Mutex}; +use yellowstone_grpc_proto::{ + geyser::{SubscribeUpdateAccount, SubscribeUpdateBlockMeta, SubscribeUpdateTransaction}, + prost_types::Timestamp, +}; + +use super::types::{AccountPretty, BlockMetaPretty, TransactionPretty}; +use crate::streaming::event_parser::core::traits::get_high_perf_clock; + +/// 通用对象池特征 +pub trait ObjectPool { + fn acquire(&self) -> PooledObject; + fn return_object(&self, obj: Box); +} + +/// 带自动归还的智能指针 +pub struct PooledObject { + object: Option>, + pool: Arc>>>, + max_size: usize, +} + +impl PooledObject { + fn new(object: Box, pool: Arc>>>, max_size: usize) -> Self { + Self { object: Some(object), pool, max_size } + } +} + +impl Drop for PooledObject { + fn drop(&mut self) { + if let Some(obj) = self.object.take() { + let mut pool = self.pool.lock().unwrap(); + if pool.len() < self.max_size { + pool.push_back(obj); + } + // 超过最大容量时直接丢弃 + } + } +} + +impl std::ops::Deref for PooledObject { + type Target = T; + + fn deref(&self) -> &Self::Target { + self.object.as_ref().unwrap() + } +} + +impl std::ops::DerefMut for PooledObject { + fn deref_mut(&mut self) -> &mut Self::Target { + self.object.as_mut().unwrap() + } +} + +/// AccountPretty 对象池 +pub struct AccountPrettyPool { + pool: Arc>>>, + max_size: usize, +} + +impl AccountPrettyPool { + pub fn new(initial_size: usize, max_size: usize) -> Self { + let mut pool = VecDeque::with_capacity(initial_size); + + // 预分配对象 + for _ in 0..initial_size { + pool.push_back(Box::new(AccountPretty::default())); + } + + Self { pool: Arc::new(Mutex::new(pool)), max_size } + } + + pub fn acquire(&self) -> PooledAccountPretty { + let mut pool = self.pool.lock().unwrap(); + let account = match pool.pop_front() { + Some(reused) => reused, + None => Box::new(AccountPretty::default()), + }; + + PooledAccountPretty { account, pool: Arc::clone(&self.pool), max_size: self.max_size } + } +} + +/// 带自动归还的 AccountPretty +pub struct PooledAccountPretty { + account: Box, + pool: Arc>>>, + max_size: usize, +} + +impl PooledAccountPretty { + /// 从 gRPC 更新重置数据 + pub fn reset_from_update(&mut self, account_update: SubscribeUpdateAccount) { + let account_info = account_update.account.unwrap(); + + self.account.slot = account_update.slot; + self.account.signature = if let Some(txn_signature) = account_info.txn_signature { + Signature::try_from(txn_signature.as_slice()).expect("valid signature") + } else { + Signature::default() + }; + self.account.pubkey = + Pubkey::try_from(account_info.pubkey.as_slice()).expect("valid pubkey"); + self.account.executable = account_info.executable; + self.account.lamports = account_info.lamports; + self.account.owner = Pubkey::try_from(account_info.owner.as_slice()).expect("valid pubkey"); + self.account.rent_epoch = account_info.rent_epoch; + + // 优化数据字段的重用 + let new_data = account_info.data; + if self.account.data.capacity() >= new_data.len() { + self.account.data.clear(); + self.account.data.extend_from_slice(&new_data); + } else { + self.account.data = new_data; + } + + self.account.program_received_time_us = get_high_perf_clock(); + } +} + +impl Drop for PooledAccountPretty { + fn drop(&mut self) { + let mut pool = self.pool.lock().unwrap(); + if pool.len() < self.max_size { + // 清理敏感数据 + self.account.data.clear(); + self.account.signature = Signature::default(); + self.account.pubkey = Pubkey::default(); + self.account.owner = Pubkey::default(); + pool.push_back(std::mem::take(&mut self.account)); + } + } +} + +impl std::ops::Deref for PooledAccountPretty { + type Target = AccountPretty; + + fn deref(&self) -> &Self::Target { + &self.account + } +} + +impl std::ops::DerefMut for PooledAccountPretty { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.account + } +} + +/// BlockMetaPretty 对象池 +pub struct BlockMetaPrettyPool { + pool: Arc>>>, + max_size: usize, +} + +impl BlockMetaPrettyPool { + pub fn new(initial_size: usize, max_size: usize) -> Self { + let mut pool = VecDeque::with_capacity(initial_size); + + // 预分配对象 + for _ in 0..initial_size { + pool.push_back(Box::new(BlockMetaPretty::default())); + } + + Self { pool: Arc::new(Mutex::new(pool)), max_size } + } + + pub fn acquire(&self) -> PooledBlockMetaPretty { + let mut pool = self.pool.lock().unwrap(); + let block_meta = match pool.pop_front() { + Some(reused) => reused, + None => Box::new(BlockMetaPretty::default()), + }; + + PooledBlockMetaPretty { block_meta, pool: Arc::clone(&self.pool), max_size: self.max_size } + } +} + +/// 带自动归还的 BlockMetaPretty +pub struct PooledBlockMetaPretty { + block_meta: Box, + pool: Arc>>>, + max_size: usize, +} + +impl PooledBlockMetaPretty { + /// 从 gRPC 更新重置数据 + pub fn reset_from_update( + &mut self, + block_update: SubscribeUpdateBlockMeta, + block_time: Option, + ) { + self.block_meta.slot = block_update.slot; + self.block_meta.block_hash = block_update.blockhash; + self.block_meta.block_time = block_time; + self.block_meta.program_received_time_us = get_high_perf_clock(); + } +} + +impl Drop for PooledBlockMetaPretty { + fn drop(&mut self) { + let mut pool = self.pool.lock().unwrap(); + if pool.len() < self.max_size { + // 清理数据 + self.block_meta.block_hash.clear(); + self.block_meta.block_time = None; + pool.push_back(std::mem::take(&mut self.block_meta)); + } + } +} + +impl std::ops::Deref for PooledBlockMetaPretty { + type Target = BlockMetaPretty; + + fn deref(&self) -> &Self::Target { + &self.block_meta + } +} + +impl std::ops::DerefMut for PooledBlockMetaPretty { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.block_meta + } +} + +/// TransactionPretty 对象池 +pub struct TransactionPrettyPool { + pool: Arc>>>, + max_size: usize, +} + +impl TransactionPrettyPool { + pub fn new(initial_size: usize, max_size: usize) -> Self { + let mut pool = VecDeque::with_capacity(initial_size); + + // 预分配对象 + for _ in 0..initial_size { + pool.push_back(Box::new(TransactionPretty::default())); + } + + Self { pool: Arc::new(Mutex::new(pool)), max_size } + } + + pub fn acquire(&self) -> PooledTransactionPretty { + let mut pool = self.pool.lock().unwrap(); + let transaction = match pool.pop_front() { + Some(reused) => reused, + None => Box::new(TransactionPretty::default()), + }; + + PooledTransactionPretty { + transaction, + pool: Arc::clone(&self.pool), + max_size: self.max_size, + } + } +} + +/// 带自动归还的 TransactionPretty +pub struct PooledTransactionPretty { + transaction: Box, + pool: Arc>>>, + max_size: usize, +} + +impl PooledTransactionPretty { + /// 从 gRPC 更新重置数据 + pub fn reset_from_update( + &mut self, + tx_update: SubscribeUpdateTransaction, + block_time: Option, + ) { + let tx = tx_update.transaction.expect("should be defined"); + + self.transaction.slot = tx_update.slot; + self.transaction.transaction_index = Some(tx.index); + self.transaction.block_time = block_time; + self.transaction.block_hash.clear(); // 重置 block_hash + self.transaction.signature = + Signature::try_from(tx.signature.as_slice()).expect("valid signature"); + self.transaction.is_vote = tx.is_vote; + self.transaction.tx = yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx) + .expect("valid tx with meta"); + self.transaction.program_received_time_us = get_high_perf_clock(); + } +} + +impl Drop for PooledTransactionPretty { + fn drop(&mut self) { + let mut pool = self.pool.lock().unwrap(); + if pool.len() < self.max_size { + // 清理数据 + self.transaction.block_hash.clear(); + self.transaction.block_time = None; + self.transaction.signature = Signature::default(); + pool.push_back(std::mem::take(&mut self.transaction)); + } + } +} + +impl std::ops::Deref for PooledTransactionPretty { + type Target = TransactionPretty; + + fn deref(&self) -> &Self::Target { + &self.transaction + } +} + +impl std::ops::DerefMut for PooledTransactionPretty { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.transaction + } +} + +/// EventPretty 对象池（组合池） +pub struct EventPrettyPool { + account_pool: AccountPrettyPool, + block_pool: BlockMetaPrettyPool, + transaction_pool: TransactionPrettyPool, +} + +impl EventPrettyPool { + pub fn new() -> Self { + Self { + account_pool: AccountPrettyPool::new(10000, 20000), + block_pool: BlockMetaPrettyPool::new(500, 1000), + transaction_pool: TransactionPrettyPool::new(10000, 20000), + } + } + + /// 获取账户事件对象 + pub fn acquire_account(&self) -> PooledAccountPretty { + self.account_pool.acquire() + } + + /// 获取区块事件对象 + pub fn acquire_block(&self) -> PooledBlockMetaPretty { + self.block_pool.acquire() + } + + /// 获取交易事件对象 + pub fn acquire_transaction(&self) -> PooledTransactionPretty { + self.transaction_pool.acquire() + } +} + +/// 对象池管理器（单例） +pub struct PoolManager { + event_pool: EventPrettyPool, +} + +impl PoolManager { + pub fn new() -> Self { + Self { event_pool: EventPrettyPool::new() } + } + + pub fn get_event_pool(&self) -> &EventPrettyPool { + &self.event_pool + } +} + +impl Default for PoolManager { + fn default() -> Self { + Self::new() + } +} + +/// 工厂函数用于创建优化的 EventPretty +impl EventPrettyPool { + /// 创建账户事件 - 使用对象池优化 + pub fn create_account_event_optimized(&self, update: SubscribeUpdateAccount) -> AccountPretty { + let mut pooled_account = self.acquire_account(); + pooled_account.reset_from_update(update); + // 移动数据而不是克隆，避免多余的内存分配 + let result = std::mem::replace(pooled_account.deref_mut(), AccountPretty::default()); + result + } + + /// 创建区块事件 - 使用对象池优化 + pub fn create_block_event_optimized( + &self, + update: SubscribeUpdateBlockMeta, + block_time: Option, + ) -> BlockMetaPretty { + let mut pooled_block = self.acquire_block(); + pooled_block.reset_from_update(update, block_time); + // 移动数据而不是克隆 + let result = std::mem::replace(pooled_block.deref_mut(), BlockMetaPretty::default()); + result + } + + /// 创建交易事件 - 使用对象池优化 + pub fn create_transaction_event_optimized( + &self, + update: SubscribeUpdateTransaction, + block_time: Option, + ) -> TransactionPretty { + let mut pooled_tx = self.acquire_transaction(); + pooled_tx.reset_from_update(update, block_time); + // 移动数据而不是克隆 + let result = std::mem::replace(pooled_tx.deref_mut(), TransactionPretty::default()); + result + } +} + +// 全局池管理器实例 +lazy_static::lazy_static! { + pub static ref GLOBAL_POOL_MANAGER: PoolManager = PoolManager::new(); +} + +/// 便捷的全局工厂函数 +pub mod factory { + use super::*; + + /// 使用对象池创建账户事件（推荐用于高性能场景） + pub fn create_account_pretty_pooled(update: SubscribeUpdateAccount) -> AccountPretty { + GLOBAL_POOL_MANAGER.get_event_pool().create_account_event_optimized(update) + } + + /// 使用对象池创建区块事件（推荐用于高性能场景） + pub fn create_block_meta_pretty_pooled( + update: SubscribeUpdateBlockMeta, + block_time: Option, + ) -> BlockMetaPretty { + GLOBAL_POOL_MANAGER.get_event_pool().create_block_event_optimized(update, block_time) + } + + /// 使用对象池创建交易事件（推荐用于高性能场景） + pub fn create_transaction_pretty_pooled( + update: SubscribeUpdateTransaction, + block_time: Option, + ) -> TransactionPretty { + GLOBAL_POOL_MANAGER.get_event_pool().create_transaction_event_optimized(update, block_time) + } +} diff --git a/src/streaming/grpc/types.rs b/src/streaming/grpc/types.rs index 7cfd946..1c450de 100644 --- a/src/streaming/grpc/types.rs +++ b/src/streaming/grpc/types.rs @@ -1,11 +1,8 @@ use solana_sdk::{pubkey::Pubkey, signature::Signature}; -use solana_transaction_status::TransactionWithStatusMeta; +use solana_transaction_status::{TransactionWithStatusMeta, VersionedTransactionWithStatusMeta}; use std::{collections::HashMap, fmt}; use yellowstone_grpc_proto::{ - geyser::{ - SubscribeRequestFilterAccounts, SubscribeRequestFilterTransactions, SubscribeUpdateAccount, - SubscribeUpdateBlockMeta, SubscribeUpdateTransaction, - }, + geyser::{SubscribeRequestFilterAccounts, SubscribeRequestFilterTransactions}, prost_types::Timestamp, }; @@ -19,7 +16,7 @@ pub enum EventPretty { Account(AccountPretty), } -#[derive(Clone)] +#[derive(Clone, Default)] pub struct AccountPretty { pub slot: u64, pub signature: Signature, @@ -47,7 +44,7 @@ impl fmt::Debug for AccountPretty { } } -#[derive(Clone)] +#[derive(Clone, Default)] pub struct BlockMetaPretty { pub slot: u64, pub block_hash: String, @@ -90,63 +87,81 @@ impl fmt::Debug for TransactionPretty { } } -impl From for AccountPretty { - fn from(account: SubscribeUpdateAccount) -> Self { - let account_info = account.account.unwrap(); +impl Default for TransactionPretty { + fn default() -> Self { Self { - slot: account.slot, - signature: if let Some(txn_signature) = account_info.txn_signature { - Signature::try_from(txn_signature.as_slice()).expect("valid signature") - } else { - Signature::default() - }, - pubkey: Pubkey::try_from(account_info.pubkey.as_slice()).expect("valid pubkey"), - executable: account_info.executable, - lamports: account_info.lamports, - owner: Pubkey::try_from(account_info.owner.as_slice()).expect("valid pubkey"), - rent_epoch: account_info.rent_epoch, - data: account_info.data, - program_received_time_us: chrono::Utc::now().timestamp_micros(), + slot: 0, + transaction_index: None, + block_hash: String::new(), + block_time: None, + signature: Signature::default(), + is_vote: false, + tx: TransactionWithStatusMeta::Complete(VersionedTransactionWithStatusMeta { + transaction: solana_sdk::transaction::VersionedTransaction::default(), + meta: solana_transaction_status::TransactionStatusMeta::default(), + }), + program_received_time_us: 0, } } } -impl From<(SubscribeUpdateBlockMeta, Option)> for BlockMetaPretty { - fn from( - (SubscribeUpdateBlockMeta { slot, blockhash, .. }, block_time): ( - SubscribeUpdateBlockMeta, - Option, - ), - ) -> Self { - Self { - block_hash: blockhash.to_string(), - block_time, - slot, - program_received_time_us: chrono::Utc::now().timestamp_micros(), - } - } -} +// impl From for AccountPretty { +// fn from(account: SubscribeUpdateAccount) -> Self { +// let account_info = account.account.unwrap(); +// Self { +// slot: account.slot, +// signature: if let Some(txn_signature) = account_info.txn_signature { +// Signature::try_from(txn_signature.as_slice()).expect("valid signature") +// } else { +// Signature::default() +// }, +// pubkey: Pubkey::try_from(account_info.pubkey.as_slice()).expect("valid pubkey"), +// executable: account_info.executable, +// lamports: account_info.lamports, +// owner: Pubkey::try_from(account_info.owner.as_slice()).expect("valid pubkey"), +// rent_epoch: account_info.rent_epoch, +// data: account_info.data, +// program_received_time_us: get_high_perf_clock(), +// } +// } +// } -impl From<(SubscribeUpdateTransaction, Option)> for TransactionPretty { - fn from( - (SubscribeUpdateTransaction { transaction, slot }, block_time): ( - SubscribeUpdateTransaction, - Option, - ), - ) -> Self { - let tx = transaction.expect("should be defined"); - // 根据用户说明，交易索引在 transaction.index 中 - let transaction_index = tx.index; - Self { - slot, - transaction_index: Some(transaction_index), // 提取交易索引 - block_time, - block_hash: "".to_string(), - signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"), - is_vote: tx.is_vote, - tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx) - .expect("valid tx with meta"), - program_received_time_us: chrono::Utc::now().timestamp_micros(), - } - } -} +// impl From<(SubscribeUpdateBlockMeta, Option)> for BlockMetaPretty { +// fn from( +// (SubscribeUpdateBlockMeta { slot, blockhash, .. }, block_time): ( +// SubscribeUpdateBlockMeta, +// Option, +// ), +// ) -> Self { +// Self { +// block_hash: blockhash, +// block_time, +// slot, +// program_received_time_us: get_high_perf_clock(), +// } +// } +// } + +// impl From<(SubscribeUpdateTransaction, Option)> for TransactionPretty { +// fn from( +// (SubscribeUpdateTransaction { transaction, slot }, block_time): ( +// SubscribeUpdateTransaction, +// Option, +// ), +// ) -> Self { +// let tx = transaction.expect("should be defined"); +// // 根据用户说明，交易索引在 transaction.index 中 +// let transaction_index = tx.index; +// Self { +// slot, +// transaction_index: Some(transaction_index), // 提取交易索引 +// block_time, +// block_hash: String::new(), +// signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"), +// is_vote: tx.is_vote, +// tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx) +// .expect("valid tx with meta"), +// program_received_time_us: get_high_perf_clock(), +// } +// } +// } diff --git a/src/streaming/shred/mod.rs b/src/streaming/shred/mod.rs index 1358fe9..cd7ce22 100644 --- a/src/streaming/shred/mod.rs +++ b/src/streaming/shred/mod.rs @@ -1,9 +1,11 @@ // ShredStream 相关模块 pub mod connection; +pub mod pool; pub mod types; // 重新导出主要类型 pub use connection::*; +pub use pool::*; pub use types::*; // 从公用模块重新导出 diff --git a/src/streaming/shred/pool.rs b/src/streaming/shred/pool.rs new file mode 100644 index 0000000..07e40df --- /dev/null +++ b/src/streaming/shred/pool.rs @@ -0,0 +1,156 @@ +use std::sync::{Arc, Mutex}; +use std::collections::VecDeque; +use std::ops::DerefMut; +use solana_sdk::transaction::VersionedTransaction; + +use super::TransactionWithSlot; + + +/// TransactionWithSlot 对象池 +pub struct TransactionWithSlotPool { + pool: Arc>>>, + max_size: usize, +} + +impl TransactionWithSlotPool { + pub fn new(initial_size: usize, max_size: usize) -> Self { + let mut pool = VecDeque::with_capacity(initial_size); + + // 预分配对象 + for _ in 0..initial_size { + pool.push_back(Box::new(TransactionWithSlot::default())); + } + + Self { pool: Arc::new(Mutex::new(pool)), max_size } + } + + pub fn acquire(&self) -> PooledTransactionWithSlot { + let mut pool = self.pool.lock().unwrap(); + let transaction = match pool.pop_front() { + Some(reused) => reused, + None => Box::new(TransactionWithSlot::default()), + }; + + PooledTransactionWithSlot { + transaction, + pool: Arc::clone(&self.pool), + max_size: self.max_size + } + } +} + +/// 带自动归还的 TransactionWithSlot +pub struct PooledTransactionWithSlot { + transaction: Box, + pool: Arc>>>, + max_size: usize, +} + +impl PooledTransactionWithSlot { + /// 从原始数据重置 + pub fn reset_from_data( + &mut self, + transaction: VersionedTransaction, + slot: u64, + program_received_time_us: i64 + ) { + self.transaction.transaction = transaction; + self.transaction.slot = slot; + self.transaction.program_received_time_us = program_received_time_us; + } + + /// 使用优化的工厂方法创建 TransactionWithSlot（移动数据而不是克隆） + pub fn into_transaction_with_slot(mut self) -> TransactionWithSlot { + // 移动数据而不是克隆，避免多余的内存分配 + std::mem::replace(self.deref_mut(), TransactionWithSlot::default()) + } +} + +impl Drop for PooledTransactionWithSlot { + fn drop(&mut self) { + let mut pool = self.pool.lock().unwrap(); + if pool.len() < self.max_size { + // 清理敏感数据 + self.transaction.slot = 0; + self.transaction.program_received_time_us = 0; + // 重置交易为默认值以清理敏感数据 + self.transaction.transaction = VersionedTransaction::default(); + pool.push_back(std::mem::take(&mut self.transaction)); + } + } +} + +impl std::ops::Deref for PooledTransactionWithSlot { + type Target = TransactionWithSlot; + + fn deref(&self) -> &Self::Target { + &self.transaction + } +} + +impl std::ops::DerefMut for PooledTransactionWithSlot { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.transaction + } +} + +/// Shred 对象池管理器 +pub struct ShredPoolManager { + transaction_pool: TransactionWithSlotPool, +} + +impl ShredPoolManager { + pub fn new() -> Self { + Self { + transaction_pool: TransactionWithSlotPool::new( + 5000, // 初始大小 - Shred 事件通常较多 + 15000, // 最大大小 + ), + } + } + + pub fn get_transaction_pool(&self) -> &TransactionWithSlotPool { + &self.transaction_pool + } + + /// 创建优化的 TransactionWithSlot + pub fn create_transaction_with_slot_optimized( + &self, + transaction: VersionedTransaction, + slot: u64, + program_received_time_us: i64, + ) -> TransactionWithSlot { + let mut pooled_tx = self.transaction_pool.acquire(); + pooled_tx.reset_from_data(transaction, slot, program_received_time_us); + pooled_tx.into_transaction_with_slot() + } +} + +impl Default for ShredPoolManager { + fn default() -> Self { + Self::new() + } +} + +// 全局 Shred 池管理器实例 +lazy_static::lazy_static! { + pub static ref GLOBAL_SHRED_POOL_MANAGER: ShredPoolManager = ShredPoolManager::new(); +} + +/// 便捷的全局工厂函数 +pub mod factory { + use super::*; + + /// 使用对象池创建 TransactionWithSlot（推荐用于高性能场景） + pub fn create_transaction_with_slot_pooled( + transaction: VersionedTransaction, + slot: u64, + program_received_time_us: i64, + ) -> TransactionWithSlot { + GLOBAL_SHRED_POOL_MANAGER.create_transaction_with_slot_optimized( + transaction, + slot, + program_received_time_us + ) + } +} diff --git a/src/streaming/shred/types.rs b/src/streaming/shred/types.rs index d10094e..7e90d74 100644 --- a/src/streaming/shred/types.rs +++ b/src/streaming/shred/types.rs @@ -1,7 +1,7 @@ use solana_sdk::transaction::VersionedTransaction; /// 携带槽位信息的交易 -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Default)] pub struct TransactionWithSlot { pub transaction: VersionedTransaction, pub slot: u64, diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index 34101c0..7fe862a 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -8,7 +8,8 @@ use crate::protos::shredstream::SubscribeEntriesRequest; use crate::streaming::common::{EventProcessor, SubscriptionHandle}; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::{Protocol, UnifiedEvent}; -use crate::streaming::shred::TransactionWithSlot; +use crate::streaming::event_parser::core::traits::get_high_perf_clock; +use crate::streaming::shred::pool::factory; use log::error; use solana_entry::entry::Entry; @@ -57,10 +58,10 @@ impl ShredStreamGrpc { if let Ok(entries) = bincode::deserialize::>(&msg.entries) { for entry in entries { for transaction in entry.transactions { - let transaction_with_slot = TransactionWithSlot::new( + let transaction_with_slot = factory::create_transaction_with_slot_pooled( transaction.clone(), msg.slot, - chrono::Utc::now().timestamp_micros(), + get_high_perf_clock(), ); // 直接处理，背压控制在 EventProcessor 内部处理 if let Err(e) = event_processor_clone diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index f083b8b..bf30b31 100644 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -5,8 +5,9 @@ use crate::streaming::common::{ use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::{Protocol, UnifiedEvent}; use crate::streaming::grpc::{ - AccountPretty, BlockMetaPretty, EventPretty, SubscriptionManager, TransactionPretty, + EventPretty, SubscriptionManager, }; +use crate::streaming::grpc::pool::factory; use anyhow::anyhow; use chrono::Local; use futures::channel::mpsc; @@ -224,7 +225,7 @@ impl YellowstoneGrpc { let created_at = msg.created_at; match msg.update_oneof { Some(UpdateOneof::Account(account)) => { - let account_pretty = AccountPretty::from(account); + let account_pretty = factory::create_account_pretty_pooled(account); log::debug!("Received account: {:?}", account_pretty); if let Err(e) = event_processor .process_grpc_event_transaction_with_metrics( @@ -237,8 +238,7 @@ impl YellowstoneGrpc { } } Some(UpdateOneof::BlockMeta(sut)) => { - let block_meta_pretty = - BlockMetaPretty::from((sut, created_at)); + 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_event_transaction_with_metrics( @@ -251,8 +251,7 @@ impl YellowstoneGrpc { } } Some(UpdateOneof::Transaction(sut)) => { - let transaction_pretty = - TransactionPretty::from((sut, created_at)); + let transaction_pretty = factory::create_transaction_pretty_pooled(sut, created_at); log::debug!( "Received transaction: {} at slot {}", transaction_pretty.signature, diff --git a/src/streaming/yellowstone_sub_system.rs b/src/streaming/yellowstone_sub_system.rs index 030397d..9d0c710 100755 --- a/src/streaming/yellowstone_sub_system.rs +++ b/src/streaming/yellowstone_sub_system.rs @@ -1,9 +1,6 @@ use crate::{ common::AnyResult, - streaming::{ - grpc::{EventPretty, TransactionPretty}, - yellowstone_grpc::YellowstoneGrpc, - }, + streaming::{grpc::pool::factory, grpc::EventPretty, yellowstone_grpc::YellowstoneGrpc}, }; use futures::{SinkExt, StreamExt}; use log::error; @@ -62,13 +59,11 @@ impl YellowstoneGrpc { let created_at = msg.created_at; match msg.update_oneof { Some(UpdateOneof::Transaction(sut)) => { - let transaction_pretty = TransactionPretty::from((sut, created_at)); + let transaction_pretty = + factory::create_transaction_pretty_pooled(sut, created_at); let event_pretty = EventPretty::Transaction(transaction_pretty); - if let Err(e) = Self::process_system_transaction( - event_pretty, - &*callback, - ) - .await + if let Err(e) = + Self::process_system_transaction(event_pretty, &*callback).await { error!("Error processing transaction: {e:?}"); }