diff --git a/Cargo.toml b/Cargo.toml index 040d079..d3ed2ce 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -65,3 +65,4 @@ maplit = "1.0.2" env_logger = "0.11.8" crossbeam = "0.8.4" crossbeam-queue = "0.3.12" +parking_lot = "0.12.1" diff --git a/README.md b/README.md index cc73cf0..d856b6f 100755 --- a/README.md +++ b/README.md @@ -24,7 +24,7 @@ A lightweight Rust library for real-time event streaming from Solana DEX trading 11. **Performance Monitoring**: Built-in performance metrics monitoring, including event processing speed, etc. 12. **Memory Optimization**: Object pooling and caching mechanisms to reduce memory allocations 13. **Flexible Configuration System**: Support for custom batch sizes, backpressure strategies, channel sizes, and other parameters -14. **Preset Configurations**: Provides high-performance, low-latency, ordered processing, and other preset configurations +14. **Preset Configurations**: Provides high-throughput, low-latency, and async processing preset configurations optimized for different use cases 15. **Backpressure Handling**: Supports blocking, dropping, retrying, ordered, and other backpressure strategies 16. **Runtime Configuration Updates**: Supports dynamic configuration parameter updates at runtime 17. **Full Function Performance Monitoring**: All subscribe_events functions support performance monitoring, automatically collecting and reporting performance metrics @@ -55,6 +55,91 @@ solana-streamer-sdk = { path = "./solana-streamer", version = "0.3.10" } solana-streamer-sdk = "0.3.10" ``` +## Configuration System + +### Preset Configurations + +The library provides three preset configurations optimized for different use cases: + +#### 1. High Throughput Configuration (`high_throughput()`) + +Optimized for high-concurrency scenarios, prioritizing throughput over latency: + +```rust +let config = StreamClientConfig::high_throughput(); +// Or use convenience methods +let grpc = YellowstoneGrpc::new_high_throughput(endpoint, token)?; +let shred = ShredStreamGrpc::new_high_throughput(endpoint).await?; +``` + +**Features:** +- **Backpressure Strategy**: Drop - drops messages during high load to avoid blocking +- **Buffer Size**: 5,000 permits to handle burst traffic +- **Use Case**: Scenarios where you need to process large volumes of data and can tolerate occasional message drops during peak loads + +#### 2. Low Latency Configuration (`low_latency()`) + +Optimized for real-time scenarios, prioritizing latency over throughput: + +```rust +let config = StreamClientConfig::low_latency(); +// Or use convenience methods +let grpc = YellowstoneGrpc::new_low_latency(endpoint, token)?; +let shred = ShredStreamGrpc::new_low_latency(endpoint).await?; +``` + +**Features:** +- **Backpressure Strategy**: Block - ensures no data loss +- **Buffer Size**: 1 permit to minimize memory usage +- **Immediate Processing**: No buffering, processes events immediately +- **Use Case**: Scenarios where every millisecond counts and you cannot afford to lose any events, such as trading applications or real-time monitoring + +#### 3. Async Processing Configuration (`async_processing()`) + +Balances throughput and reliability: + +```rust +let config = StreamClientConfig::async_processing(); +// Or use convenience methods +let grpc = YellowstoneGrpc::new_async_processing(endpoint, token)?; +let shred = ShredStreamGrpc::new_async_processing(endpoint).await?; +``` + +**Features:** +- **Backpressure Strategy**: Async - non-blocking operation +- **Buffer Size**: 5,000 permits for steady flow +- **Fire-and-forget**: Async processing semantics +- **Use Case**: Scenarios where you need sustained high throughput with eventual consistency, such as data ingestion pipelines or event streaming applications + +### Custom Configuration + +You can also create custom configurations: + +```rust +let config = StreamClientConfig { + connection: ConnectionConfig { + connect_timeout: 30, + request_timeout: 120, + max_decoding_message_size: 20 * 1024 * 1024, // 20MB + }, + backpressure: BackpressureConfig { + permits: 2000, + strategy: BackpressureStrategy::Block, + }, + enable_metrics: true, +}; +``` + +### Configuration Selection Guide + +| Scenario | Recommended Config | Reason | +|----------|-------------------|---------| +| Trading Bots | `low_latency()` | Need fastest response time, cannot lose trading signals | +| Data Analytics | `high_throughput()` | Need to process large amounts of historical data, can tolerate some data loss | +| Event Stream Processing | `async_processing()` | Balance performance and reliability, suitable for continuous processing | +| Real-time Monitoring | `low_latency()` | Need immediate response to anomalies | +| Bulk Data Ingestion | `high_throughput()` | Prioritize overall throughput | + ## Usage Examples ### Quick Start - Parse Transaction Events diff --git a/README_CN.md b/README_CN.md index 0a63871..cf4330a 100644 --- a/README_CN.md +++ b/README_CN.md @@ -24,7 +24,7 @@ 11. **性能监控**: 内置性能指标监控，包括事件处理速度等 12. **内存优化**: 对象池和缓存机制减少内存分配 13. **灵活配置系统**: 支持自定义批处理大小、背压策略、通道大小等参数 -14. **预设配置**: 提供高性能、低延迟、有序处理等预设配置 +14. **预设配置**: 提供高吞吐量、低延迟、异步处理等预设配置，针对不同使用场景优化 15. **背压处理**: 支持阻塞、丢弃、重试、有序等多种背压策略 16. **运行时配置更新**: 支持在运行时动态更新配置参数 17. **全函数性能监控**: 所有subscribe_events函数都支持性能监控，自动收集和报告性能指标 @@ -55,6 +55,91 @@ solana-streamer-sdk = { path = "./solana-streamer", version = "0.3.10" } solana-streamer-sdk = "0.3.10" ``` +## 配置系统 + +### 预设配置 + +库提供了三种预设配置，针对不同的使用场景进行了优化： + +#### 1. 高吞吐量配置 (`high_throughput()`) + +专为高并发场景优化，优先考虑吞吐量而非延迟： + +```rust +let config = StreamClientConfig::high_throughput(); +// 或者使用便捷方法 +let grpc = YellowstoneGrpc::new_high_throughput(endpoint, token)?; +let shred = ShredStreamGrpc::new_high_throughput(endpoint).await?; +``` + +**特性：** +- **背压策略**: Drop（丢弃策略）- 在高负载时丢弃消息以避免阻塞 +- **缓冲区大小**: 5,000 个许可证，处理突发流量 +- **适用场景**: 需要处理大量数据且可以容忍在峰值负载时偶尔丢失消息的场景 + +#### 2. 低延迟配置 (`low_latency()`) + +专为实时场景优化，优先考虑延迟而非吞吐量： + +```rust +let config = StreamClientConfig::low_latency(); +// 或者使用便捷方法 +let grpc = YellowstoneGrpc::new_low_latency(endpoint, token)?; +let shred = ShredStreamGrpc::new_low_latency(endpoint).await?; +``` + +**特性：** +- **背压策略**: Block（阻塞策略）- 确保不丢失任何数据 +- **缓冲区大小**: 1 个许可证，最小化内存使用 +- **立即处理**: 不进行缓冲，立即处理事件 +- **适用场景**: 每毫秒都很重要且不能丢失任何事件的场景，如交易应用或实时监控 + +#### 3. 异步处理配置 (`async_processing()`) + +在吞吐量和可靠性之间取得平衡： + +```rust +let config = StreamClientConfig::async_processing(); +// 或者使用便捷方法 +let grpc = YellowstoneGrpc::new_async_processing(endpoint, token)?; +let shred = ShredStreamGrpc::new_async_processing(endpoint).await?; +``` + +**特性：** +- **背压策略**: Async（异步策略）- 非阻塞操作 +- **缓冲区大小**: 5,000 个许可证，保持稳定流量 +- **Fire-and-forget**: 异步处理语义 +- **适用场景**: 需要持续高吞吐量且可接受最终一致性的场景，如数据摄取管道或事件流应用 + +### 自定义配置 + +您也可以创建自定义配置： + +```rust +let config = StreamClientConfig { + connection: ConnectionConfig { + connect_timeout: 30, + request_timeout: 120, + max_decoding_message_size: 20 * 1024 * 1024, // 20MB + }, + backpressure: BackpressureConfig { + permits: 2000, + strategy: BackpressureStrategy::Block, + }, + enable_metrics: true, +}; +``` + +### 配置选择指南 + +| 场景 | 推荐配置 | 原因 | +|------|----------|------| +| 交易机器人 | `low_latency()` | 需要最快响应时间，不能丢失交易信号 | +| 数据分析 | `high_throughput()` | 需要处理大量历史数据，可容忍部分数据丢失 | +| 事件流处理 | `async_processing()` | 平衡性能和可靠性，适合持续处理 | +| 实时监控 | `low_latency()` | 需要立即响应异常情况 | +| 批量数据摄取 | `high_throughput()` | 优先考虑整体吞吐量 | + ## 使用示例 ### 快速开始 - 解析交易事件 diff --git a/examples/parse_tx_events.rs b/examples/parse_tx_events.rs new file mode 100644 index 0000000..40b9a02 --- /dev/null +++ b/examples/parse_tx_events.rs @@ -0,0 +1,123 @@ +use anyhow::Result; +use solana_sdk::commitment_config::CommitmentConfig; + +use solana_streamer_sdk::streaming::event_parser::UnifiedEvent; +use solana_streamer_sdk::streaming::event_parser::{ + protocols::MutilEventParser, EventParser, Protocol, +}; +use solana_transaction_status::{InnerInstruction, InnerInstructions, UiInstruction}; + +use solana_sdk::bs58; +use solana_sdk::instruction::CompiledInstruction; +use std::str::FromStr; +use std::sync::Arc; + +/// Get transaction data based on transaction signature +#[tokio::main] +async fn main() -> Result<()> { + let signatures = vec![ + "4PsHYajH87x2zJPEGZczZtd2ksibuMCFPonC24jk5mTGZ46hzvjpzM5UZuLz9sRv79MkCBbtDqwJapGPTSkCFKoL", + ]; + // Validate signature format + let mut valid_signatures = Vec::new(); + for sig_str in &signatures { + match solana_sdk::signature::Signature::from_str(sig_str) { + Ok(_) => valid_signatures.push(*sig_str), + Err(e) => println!("Invalid signature format: {}", e), + } + } + if valid_signatures.is_empty() { + println!("No valid transaction signatures"); + return Ok(()); + } + for signature in valid_signatures { + println!("Starting transaction parsing: {}", signature); + get_single_transaction_details(signature).await?; + println!("Transaction parsing completed: {}\n", signature); + println!("Visit link to compare data: \nhttps://solscan.io/tx/{}\n", signature); + println!("--------------------------------"); + } + + Ok(()) +} + +/// Get details of a single transaction +async fn get_single_transaction_details(signature_str: &str) -> Result<()> { + use solana_sdk::signature::Signature; + use solana_transaction_status::UiTransactionEncoding; + + let signature = Signature::from_str(signature_str)?; + + // Create Solana RPC client + let rpc_url = "https://api.mainnet-beta.solana.com"; + println!("Connecting to Solana RPC: {}", rpc_url); + + let client = solana_client::nonblocking::rpc_client::RpcClient::new(rpc_url.to_string()); + + match client + .get_transaction_with_config( + &signature, + solana_client::rpc_config::RpcTransactionConfig { + encoding: Some(UiTransactionEncoding::Base64), + commitment: Some(CommitmentConfig::confirmed()), + max_supported_transaction_version: Some(0), + }, + ) + .await + { + Ok(transaction) => { + println!("Transaction signature: {}", signature_str); + println!("Block slot: {}", transaction.slot); + + if let Some(block_time) = transaction.block_time { + println!("Block time: {}", block_time); + } + + if let Some(meta) = &transaction.transaction.meta { + println!("Transaction fee: {} lamports", meta.fee); + println!("Status: {}", if meta.err.is_none() { "Success" } else { "Failed" }); + if let Some(err) = &meta.err { + println!("Error details: {:?}", err); + } + // Compute units consumed + if let solana_transaction_status::option_serializer::OptionSerializer::Some(units) = + &meta.compute_units_consumed + { + println!("Compute units consumed: {}", units); + } + // Display logs (all) + if let solana_transaction_status::option_serializer::OptionSerializer::Some(logs) = + &meta.log_messages + { + println!("Transaction logs (all {} entries):", logs.len()); + for (i, log) in logs.iter().enumerate() { + println!(" [{}] {}", i + 1, log); + } + } + } + let protocols = vec![ + Protocol::Bonk, + Protocol::RaydiumClmm, + Protocol::PumpSwap, + Protocol::PumpFun, + Protocol::RaydiumCpmm, + Protocol::RaydiumAmmV4, + ]; + let parser: Arc = Arc::new(MutilEventParser::new(protocols, None)); + parser + .parse_encoded_confirmed_transaction_with_status_meta( + signature, + transaction, + Arc::new(move |event: &Box| { + println!("{:?}\n", event); + }), + ) + .await?; + } + Err(e) => { + println!("Failed to get transaction: {}", e); + } + } + + Ok(()) +} diff --git a/examples/parse_tx_events.rs.bak b/examples/parse_tx_events.rs.bak deleted file mode 100644 index 4ce1b6b..0000000 --- a/examples/parse_tx_events.rs.bak +++ /dev/null @@ -1,196 +0,0 @@ -use anyhow::Result; -use solana_sdk::commitment_config::CommitmentConfig; -use solana_sdk::message::v0::LoadedAddresses; -use solana_streamer_sdk::streaming::event_parser::{ - protocols::MutilEventParser, EventParser, Protocol, -}; -use solana_transaction_status::{ - option_serializer::OptionSerializer, TransactionStatusMeta, TransactionWithStatusMeta, - VersionedTransactionWithStatusMeta, -}; -use std::str::FromStr; -use std::sync::Arc; - -/// Get transaction data based on transaction signature -#[tokio::main] -async fn main() -> Result<()> { - let signatures = vec![ - "5cnxDiHzUTUutMwnTCsvnMhyL9jEQsRWimmWU1gKxpQBngDaGTkou1YbGhUJhAhmTgvu49PYMmFQbbR38wdZDxJF", - ]; - // Validate signature format - let mut valid_signatures = Vec::new(); - for sig_str in &signatures { - match solana_sdk::signature::Signature::from_str(sig_str) { - Ok(_) => valid_signatures.push(*sig_str), - Err(e) => println!("Invalid signature format: {}", e), - } - } - if valid_signatures.is_empty() { - println!("No valid transaction signatures"); - return Ok(()); - } - for signature in valid_signatures { - println!("Starting transaction parsing: {}", signature); - get_single_transaction_details(signature).await?; - println!("Transaction parsing completed: {}\n", signature); - println!("Visit link to compare data: \nhttps://solscan.io/tx/{}\n", signature); - println!("--------------------------------"); - } - - Ok(()) -} - -/// Get details of a single transaction -async fn get_single_transaction_details(signature_str: &str) -> Result<()> { - use solana_sdk::signature::Signature; - use solana_transaction_status::UiTransactionEncoding; - - let signature = Signature::from_str(signature_str)?; - - // Create Solana RPC client - let rpc_url = "https://api.mainnet-beta.solana.com"; - println!("Connecting to Solana RPC: {}", rpc_url); - - let client = solana_client::nonblocking::rpc_client::RpcClient::new(rpc_url.to_string()); - - match client - .get_transaction_with_config( - &signature, - solana_client::rpc_config::RpcTransactionConfig { - encoding: Some(UiTransactionEncoding::Base64), - commitment: Some(CommitmentConfig::confirmed()), - max_supported_transaction_version: Some(0), - }, - ) - .await - { - Ok(transaction) => { - println!("Transaction signature: {}", signature_str); - println!("Block slot: {}", transaction.slot); - - if let Some(block_time) = transaction.block_time { - println!("Block time: {}", block_time); - } - - if let Some(meta) = &transaction.transaction.meta { - println!("Transaction fee: {} lamports", meta.fee); - println!("Status: {}", if meta.err.is_none() { "Success" } else { "Failed" }); - if let Some(err) = &meta.err { - println!("Error details: {:?}", err); - } - // Compute units consumed - if let solana_transaction_status::option_serializer::OptionSerializer::Some(units) = - &meta.compute_units_consumed - { - println!("Compute units consumed: {}", units); - } - // Display logs (all) - if let solana_transaction_status::option_serializer::OptionSerializer::Some(logs) = - &meta.log_messages - { - println!("Transaction logs (all {} entries):", logs.len()); - for (i, log) in logs.iter().enumerate() { - println!(" [{}] {}", i + 1, log); - } - } - } - let protocols = vec![ - Protocol::Bonk, - Protocol::RaydiumClmm, - Protocol::PumpSwap, - Protocol::PumpFun, - Protocol::RaydiumCpmm, - Protocol::RaydiumAmmV4, - ]; - let parser: Arc = Arc::new(MutilEventParser::new(protocols, None)); - let start_time = std::time::Instant::now(); - - // 从 EncodedTransaction 获取 VersionedTransaction - let versioned_tx = match transaction.transaction.transaction.decode() { - Some(tx) => tx, - None => { - println!("Failed to decode transaction"); - return Ok(()); - } - }; - - // 创建 TransactionWithStatusMeta - let tx = TransactionWithStatusMeta::Complete(VersionedTransactionWithStatusMeta { - transaction: versioned_tx, - meta: TransactionStatusMeta { - status: Ok(()), - fee: transaction.transaction.meta.as_ref().map_or(0, |m| m.fee), - pre_balances: transaction - .transaction - .meta - .as_ref() - .map_or(vec![], |m| m.pre_balances.clone()), - post_balances: transaction - .transaction - .meta - .as_ref() - .map_or(vec![], |m| m.post_balances.clone()), - inner_instructions: transaction.transaction.meta.as_ref().and_then(|m| { - if let OptionSerializer::Some(inner_instructions) = &m.inner_instructions { - // 手动将每个UiInnerInstructions转换为InnerInstructions - Some(inner_instructions.iter().map(|ui_inner| { - solana_transaction_status::InnerInstructions { - index: ui_inner.index, - instructions: ui_inner.instructions.iter().map(|ui_inst| { - solana_transaction_status::InnerInstruction { - instruction: solana_sdk::instruction::Instruction { - program_id: solana_sdk::pubkey::Pubkey::new_from_array([0; 32]), - accounts: vec![], - data: vec![], - }, - stack_height: None, - } - }).collect(), - } - }).collect()) - } else { - None - } - }), - log_messages: transaction.transaction.meta.as_ref().and_then(|m| { - if let OptionSerializer::Some(logs) = &m.log_messages { - Some(logs.clone()) - } else { - None - } - }), - pre_token_balances: None, - post_token_balances: None, - rewards: None, - loaded_addresses: LoadedAddresses::default(), - return_data: None, - compute_units_consumed: None, - cost_units: None, - }, - }); - - // TransactionWithStatusMeta - let events = parser - .parse_transaction( - tx, - &signature.to_string(), - Some(transaction.slot), - None, - chrono::Utc::now().timestamp_micros(), - None, - ) - .await - .unwrap_or_else(|_e| vec![]); - - println!("Parsing time: {:?}", start_time.elapsed()); - for event in events { - println!("{:?}\n", event); - } - } - Err(e) => { - println!("Failed to get transaction: {}", e); - } - } - - Ok(()) -} diff --git a/src/main.rs b/src/main.rs index b95723c..d6ad5c3 100755 --- a/src/main.rs +++ b/src/main.rs @@ -61,7 +61,7 @@ async fn test_grpc() -> Result<(), Box> { println!("Subscribing to Yellowstone gRPC events..."); // Create low-latency configuration - let mut config = ClientConfig::low_latency(); + let mut config: ClientConfig = ClientConfig::low_latency(); // Enable performance monitoring, has performance overhead, disabled by default config.enable_metrics = true; let grpc = YellowstoneGrpc::new_with_config( @@ -86,17 +86,13 @@ async fn test_grpc() -> Result<(), Box> { println!("Protocols to monitor: {:?}", protocols); - const SYSTEM_PROGRAM_ID: solana_sdk::pubkey::Pubkey = - solana_sdk::pubkey!("11111111111111111111111111111111"); - // Filter accounts let account_include = vec![ - SYSTEM_PROGRAM_ID.to_string(), - PUMPFUN_PROGRAM_ID.to_string(), // Listen to pumpfun program ID - PUMPSWAP_PROGRAM_ID.to_string(), // Listen to pumpswap program ID - BONK_PROGRAM_ID.to_string(), // Listen to bonk program ID - RAYDIUM_CPMM_PROGRAM_ID.to_string(), // Listen to raydium_cpmm program ID - RAYDIUM_CLMM_PROGRAM_ID.to_string(), // Listen to raydium_clmm program ID + PUMPFUN_PROGRAM_ID.to_string(), // Listen to pumpfun program ID + PUMPSWAP_PROGRAM_ID.to_string(), // Listen to pumpswap program ID + BONK_PROGRAM_ID.to_string(), // Listen to bonk program ID + RAYDIUM_CPMM_PROGRAM_ID.to_string(), // Listen to raydium_cpmm program ID + RAYDIUM_CLMM_PROGRAM_ID.to_string(), // Listen to raydium_clmm program ID RAYDIUM_AMM_V4_PROGRAM_ID.to_string(), // Listen to raydium_amm_v4 program ID ]; let account_exclude = vec![]; @@ -116,7 +112,7 @@ async fn test_grpc() -> Result<(), Box> { // No event filtering, includes all events let event_type_filter = None; // Only include PumpSwapBuy events and PumpSwapSell events - // let event_type_filter = EventTypeFilter { include: vec![EventType::PumpSwapBuy, EventType::PumpSwapSell] }; + // let event_type_filter = Some(EventTypeFilter { include: vec![EventType::PumpFunBuy] }); println!("Starting to listen for events, press Ctrl+C to stop..."); println!("Monitoring programs: {:?}", account_include); @@ -155,7 +151,7 @@ async fn test_shreds() -> Result<(), Box> { // Enable performance monitoring, has performance overhead, disabled by default config.enable_metrics = true; let shred_stream = - ShredStreamGrpc::new_with_config("http://64.130.37.195:10800".to_string(), config).await?; + ShredStreamGrpc::new_with_config("http://127.0.0.1:10800".to_string(), config).await?; let callback = create_event_callback(); let protocols = vec![ @@ -192,7 +188,11 @@ async fn test_shreds() -> Result<(), Box> { fn create_event_callback() -> impl Fn(Box) { |event: Box| { - println!("🎉 Event received! Type: {:?}, ID: {}", event.event_type(), event.id()); + println!( + "🎉 Event received! Type: {:?}, transaction_index: {:?}", + event.event_type(), + event.transaction_index() + ); match_event!(event, { // -------------------------- block meta ----------------------- BlockMetaEvent => |e: BlockMetaEvent| { diff --git a/src/streaming/common/config.rs b/src/streaming/common/config.rs index c050016..0c76687 100644 --- a/src/streaming/common/config.rs +++ b/src/streaming/common/config.rs @@ -1,14 +1,14 @@ use super::constants::*; -/// 背压处理策略 +/// Backpressure handling strategy #[derive(Debug, Clone, Copy)] pub enum BackpressureStrategy { - /// 阻塞等待（默认） + /// Block and wait (default) Block, - /// 丢弃消息 + /// Drop messages Drop, - /// 重试有限次数后丢弃 - Retry { max_attempts: usize, wait_ms: u64 }, + /// Execute asynchronously (don't wait for completion) + Async, } impl Default for BackpressureStrategy { @@ -17,50 +17,29 @@ impl Default for BackpressureStrategy { } } -/// 批处理配置 -#[derive(Debug, Clone)] -pub struct BatchConfig { - /// 批处理大小（默认：100） - pub batch_size: usize, - /// 批处理超时时间（毫秒，默认：5ms） - pub batch_timeout_ms: u64, - /// 是否启用批处理（默认：true） - pub enabled: bool, -} - -impl Default for BatchConfig { - fn default() -> Self { - Self { - batch_size: DEFAULT_BATCH_SIZE, - batch_timeout_ms: DEFAULT_BATCH_TIMEOUT_MS, - enabled: true, - } - } -} - -/// 背压配置 +/// Backpressure configuration #[derive(Debug, Clone)] pub struct BackpressureConfig { - /// 通道大小（默认：1000） - pub channel_size: usize, - /// 背压处理策略（默认：Block） + /// Channel size (default: 1000) + pub permits: usize, + /// Backpressure handling strategy (default: Block) pub strategy: BackpressureStrategy, } impl Default for BackpressureConfig { fn default() -> Self { - Self { channel_size: DEFAULT_CHANNEL_SIZE, strategy: BackpressureStrategy::default() } + Self { permits: 1, strategy: BackpressureStrategy::default() } } } -/// 连接配置 +/// Connection configuration #[derive(Debug, Clone)] pub struct ConnectionConfig { - /// 连接超时时间（秒，默认：10） + /// Connection timeout in seconds (default: 10) pub connect_timeout: u64, - /// 请求超时时间（秒，默认：60） + /// Request timeout in seconds (default: 60) pub request_timeout: u64, - /// 最大解码消息大小（字节，默认：10MB） + /// Maximum decoding message size in bytes (default: 10MB) pub max_decoding_message_size: usize, } @@ -74,16 +53,14 @@ impl Default for ConnectionConfig { } } -/// 通用客户端配置 +/// Common client configuration #[derive(Debug, Clone)] pub struct StreamClientConfig { - /// 连接配置 + /// Connection configuration pub connection: ConnectionConfig, - /// 批处理配置 - pub batch: BatchConfig, - /// 背压配置 + /// Backpressure configuration pub backpressure: BackpressureConfig, - /// 是否启用性能监控（默认：false） + /// Whether performance monitoring is enabled (default: false) pub enable_metrics: bool, } @@ -91,7 +68,6 @@ impl Default for StreamClientConfig { fn default() -> Self { Self { connection: ConnectionConfig::default(), - batch: BatchConfig::default(), backpressure: BackpressureConfig::default(), enable_metrics: false, } @@ -99,31 +75,57 @@ impl Default for StreamClientConfig { } impl StreamClientConfig { - /// 创建高性能配置（适合高并发场景） - pub fn high_performance() -> Self { + /// Creates a high-throughput configuration optimized for high-concurrency scenarios. + /// + /// This configuration prioritizes throughput over latency by: + /// - Implementing a drop strategy for backpressure to avoid blocking + /// - Setting a large permit buffer (5,000) to handle burst traffic + /// + /// Ideal for scenarios where you need to process large volumes of data + /// and can tolerate occasional message drops during peak loads. + pub fn high_throughput() -> Self { Self { connection: ConnectionConfig::default(), - batch: BatchConfig { batch_size: 200, batch_timeout_ms: 5, enabled: true }, backpressure: BackpressureConfig { - channel_size: 20000, + permits: 5000, strategy: BackpressureStrategy::Drop, }, enable_metrics: false, } } - /// 创建低延迟配置（适合实时场景） + /// Creates a low-latency configuration optimized for real-time scenarios. + /// + /// This configuration prioritizes latency over throughput by: + /// - Processing events immediately without buffering + /// - Implementing a blocking backpressure strategy to ensure no data loss + /// - Setting minimal permits (1) to minimize memory usage + /// + /// Ideal for scenarios where every millisecond counts and you cannot + /// afford to lose any events, such as trading applications or real-time monitoring. pub fn low_latency() -> Self { Self { connection: ConnectionConfig::default(), - batch: BatchConfig { - batch_size: 10, - batch_timeout_ms: 1, - enabled: false, // 禁用批处理，即时处理 - }, + backpressure: BackpressureConfig { permits: 1, strategy: BackpressureStrategy::Block }, + enable_metrics: false, + } + } + + /// Creates an asynchronous processing configuration optimized for high-volume scenarios. + /// + /// This configuration balances throughput and reliability by: + /// - Implementing an async backpressure strategy for non-blocking operation + /// - Setting a balanced permit buffer (5,000) for steady flow + /// + /// Ideal for scenarios where you need sustained high throughput with + /// fire-and-forget semantics, such as data ingestion pipelines or + /// event streaming applications where some eventual consistency is acceptable. + pub fn async_processing() -> Self { + Self { + connection: ConnectionConfig::default(), backpressure: BackpressureConfig { - channel_size: 1000, - strategy: BackpressureStrategy::Block, + permits: 5000, + strategy: BackpressureStrategy::Async, }, enable_metrics: false, } diff --git a/src/streaming/common/constants.rs b/src/streaming/common/constants.rs index 0b253f6..e54309c 100644 --- a/src/streaming/common/constants.rs +++ b/src/streaming/common/constants.rs @@ -5,8 +5,6 @@ pub const DEFAULT_CONNECT_TIMEOUT: u64 = 10; pub const DEFAULT_REQUEST_TIMEOUT: u64 = 60; pub const DEFAULT_CHANNEL_SIZE: usize = 1000; pub const DEFAULT_MAX_DECODING_MESSAGE_SIZE: usize = 1024 * 1024 * 10; -pub const DEFAULT_BATCH_SIZE: usize = 100; -pub const DEFAULT_BATCH_TIMEOUT_MS: u64 = 5; // 性能监控相关常量 pub const DEFAULT_METRICS_WINDOW_SECONDS: u64 = 5; diff --git a/src/streaming/common/event_processor.rs b/src/streaming/common/event_processor.rs index e7ffe13..571a6b5 100644 --- a/src/streaming/common/event_processor.rs +++ b/src/streaming/common/event_processor.rs @@ -1,7 +1,9 @@ use std::sync::Arc; +use std::time::Instant; use solana_sdk::pubkey::Pubkey; use solana_sdk::signature::Signature; +use tokio::sync::Semaphore; use crate::common::AnyResult; use crate::streaming::common::{ @@ -14,11 +16,11 @@ use crate::streaming::event_parser::EventParser; use crate::streaming::event_parser::{ core::traits::UnifiedEvent, protocols::mutil::parser::MutilEventParser, Protocol, }; -use crate::streaming::grpc::{BackpressureConfig, BatchConfig, EventPretty}; +use crate::streaming::grpc::{BackpressureConfig, EventPretty}; use crate::streaming::shred::TransactionWithSlot; use once_cell::sync::OnceCell; -/// 事件处理器 +/// Event processor pub struct EventProcessor { pub(crate) metrics_manager: MetricsManager, pub(crate) config: ClientConfig, @@ -27,13 +29,15 @@ pub struct EventProcessor { pub(crate) event_type_filter: Option, pub(crate) callback: Option) + Send + Sync>>, pub(crate) backpressure_config: BackpressureConfig, - pub(crate) batch_config: BatchConfig, + /// Backpressure semaphore for controlling concurrent processing count + pub(crate) backpressure_semaphore: Arc, } impl EventProcessor { - /// 创建新的事件处理器 + /// Create a new event processor pub fn new(metrics_manager: MetricsManager, config: ClientConfig) -> Self { let backpressure_config = config.backpressure.clone(); + let backpressure_semaphore = Arc::new(Semaphore::new(backpressure_config.permits)); Self { metrics_manager, config, @@ -41,8 +45,8 @@ impl EventProcessor { protocols: vec![], event_type_filter: None, backpressure_config, - batch_config: BatchConfig::default(), callback: None, + backpressure_semaphore, } } @@ -51,13 +55,15 @@ impl EventProcessor { protocols: Vec, event_type_filter: Option, backpressure_config: BackpressureConfig, - batch_config: BatchConfig, callback: Option) + Send + Sync>>, ) { self.protocols = protocols.clone(); self.event_type_filter = event_type_filter.clone(); + // Recreate semaphore if backpressure configuration changes + if self.backpressure_config.permits != backpressure_config.permits { + self.backpressure_semaphore = Arc::new(Semaphore::new(backpressure_config.permits)); + } self.backpressure_config = backpressure_config; - self.batch_config = batch_config; self.callback = callback; self.parser_cache .get_or_init(|| Arc::new(MutilEventParser::new(protocols, event_type_filter))); @@ -72,8 +78,73 @@ impl EventProcessor { event_pretty: EventPretty, bot_wallet: Option, ) -> AnyResult<()> { - self.process_grpc_event_transaction(event_pretty, bot_wallet).await?; - Ok(()) + // Backpressure control logic + let backpressure_start = Instant::now(); + let result = self.apply_backpressure_control(event_pretty, bot_wallet).await; + let backpressure_duration = backpressure_start.elapsed(); + + // Record backpressure-related metrics + self.metrics_manager.record_backpressure_metrics( + backpressure_duration, + result.is_ok(), + self.backpressure_semaphore.available_permits(), + ); + + result + } + + /// Apply backpressure control strategy + async fn apply_backpressure_control( + &self, + event_pretty: EventPretty, + bot_wallet: Option, + ) -> AnyResult<()> { + use crate::streaming::common::BackpressureStrategy; + + match self.backpressure_config.strategy { + BackpressureStrategy::Block => { + // Blocking strategy: acquire semaphore permit + let _permit = + self.backpressure_semaphore.acquire().await.map_err(|e| { + anyhow::anyhow!("Failed to acquire backpressure permit: {}", e) + })?; + self.process_grpc_event_transaction(event_pretty, bot_wallet).await + } + BackpressureStrategy::Drop => { + // Drop strategy: try to acquire permit, drop if failed + match self.backpressure_semaphore.try_acquire() { + Ok(_permit) => { + let result = + self.process_grpc_event_transaction(event_pretty, bot_wallet).await; + result + } + Err(_) => { + // Record dropped event + self.metrics_manager.increment_dropped_events(); + Ok(()) + } + } + } + BackpressureStrategy::Async => { + // Async strategy: process asynchronously regardless of permits + self.spawn_async_processing(event_pretty, bot_wallet).await; + Ok(()) + } + } + } + + /// Process event asynchronously (without waiting for semaphore permit) + async fn spawn_async_processing(&self, event_pretty: EventPretty, bot_wallet: Option) { + let processor = self.clone(); + + tokio::spawn(async move { + // Async strategy: no semaphore control, allow unlimited concurrency + // Execute actual event processing directly + if let Err(e) = processor.process_grpc_event_transaction(event_pretty, bot_wallet).await + { + log::error!("Error in async event processing: {}", e); + } + }); } async fn process_grpc_event_transaction( @@ -81,6 +152,9 @@ impl EventProcessor { event_pretty: EventPretty, bot_wallet: Option, ) -> AnyResult<()> { + if self.callback.is_none() { + return Ok(()); + } match event_pretty { EventPretty::Account(account_pretty) => { self.metrics_manager.add_account_process_count(); @@ -105,40 +179,36 @@ impl EventProcessor { self.metrics_manager.add_tx_process_count(); let slot = transaction_pretty.slot; let signature = transaction_pretty.signature; - // 使用缓存获取解析器 + let tx = transaction_pretty.tx; + let block_time = transaction_pretty.block_time; + let program_received_time_us = transaction_pretty.program_received_time_us; + let transaction_index = transaction_pretty.transaction_index; + // Use cache to get parser let parser = self.get_parser(); - let all_events = parser - .parse_transaction( - transaction_pretty.tx.clone(), + let callback = self.callback.clone().unwrap(); + let metrics_manager = self.metrics_manager.clone(); + let adapter_callback = Arc::new(move |event: Box| { + let processing_time_us = event.program_handle_time_consuming_us() as f64; + callback(event); + metrics_manager.update_metrics( + MetricsEventType::Transaction, + 1, + processing_time_us, + Some(signature), + ); + }); + parser + .parse_transaction_owned( + tx, signature, Some(slot), - transaction_pretty.block_time, - transaction_pretty.program_received_time_us, + block_time, + program_received_time_us, bot_wallet, - transaction_pretty.transaction_index, + transaction_index, + adapter_callback, ) - .await - .unwrap_or_else(|_e| vec![]); - - let mut all_time_consuming_us = 0; - let event_count = all_events.len(); - - // 为所有事件设置交易索引 - for mut event in all_events { - event.set_program_handle_time_consuming_us( - chrono::Utc::now().timestamp_micros() - event.program_received_time_us(), - ); - all_time_consuming_us += event.program_handle_time_consuming_us(); - self.invoke_callback(event); - } - - // 更新性能指标 - self.update_metrics( - MetricsEventType::Transaction, - event_count as u64, - all_time_consuming_us as f64, - Some(signature), - ); + .await?; } EventPretty::BlockMeta(block_meta_pretty) => { self.metrics_manager.add_block_meta_process_count(); @@ -167,7 +237,7 @@ impl EventProcessor { } } - /// 即时处理单个交易 + /// Process a single transaction immediately pub async fn process_shred_transaction_immediate( &self, transaction_with_slot: TransactionWithSlot, @@ -176,55 +246,129 @@ impl EventProcessor { self.process_shred_transaction(transaction_with_slot, bot_wallet).await } + /// Process shred transaction with backpressure control and performance monitoring + pub async fn process_shred_transaction_with_metrics( + &self, + transaction_with_slot: TransactionWithSlot, + bot_wallet: Option, + ) -> AnyResult<()> { + // Backpressure control logic + let backpressure_start = Instant::now(); + let result = self.apply_shred_backpressure_control(transaction_with_slot, bot_wallet).await; + let backpressure_duration = backpressure_start.elapsed(); + + // Record backpressure-related metrics + self.metrics_manager.record_backpressure_metrics( + backpressure_duration, + result.is_ok(), + self.backpressure_semaphore.available_permits(), + ); + + result + } + + /// Apply shred backpressure control strategy + async fn apply_shred_backpressure_control( + &self, + transaction_with_slot: TransactionWithSlot, + bot_wallet: Option, + ) -> AnyResult<()> { + use crate::streaming::common::BackpressureStrategy; + + match self.backpressure_config.strategy { + BackpressureStrategy::Block => { + // Blocking strategy: acquire semaphore permit + let _permit = + self.backpressure_semaphore.acquire().await.map_err(|e| { + anyhow::anyhow!("Failed to acquire backpressure permit: {}", e) + })?; + self.process_shred_transaction(transaction_with_slot, bot_wallet).await + } + BackpressureStrategy::Drop => { + // Drop strategy: try to acquire permit, drop if failed + match self.backpressure_semaphore.try_acquire() { + Ok(_permit) => { + let result = + self.process_shred_transaction(transaction_with_slot, bot_wallet).await; + result + } + Err(_) => { + // Record dropped event + self.metrics_manager.increment_dropped_events(); + Ok(()) + } + } + } + BackpressureStrategy::Async => { + // Async strategy: process asynchronously regardless of permits + self.spawn_async_shred_processing(transaction_with_slot, bot_wallet).await; + Ok(()) + } + } + } + + /// Process shred event asynchronously (without waiting for semaphore permit) + async fn spawn_async_shred_processing( + &self, + transaction_with_slot: TransactionWithSlot, + bot_wallet: Option, + ) { + let processor = self.clone(); + + tokio::spawn(async move { + // Async strategy: no semaphore control, allow unlimited concurrency + // Execute actual event processing directly + if let Err(e) = + processor.process_shred_transaction(transaction_with_slot, bot_wallet).await + { + log::error!("Error in async shred event processing: {}", e); + } + }); + } + pub async fn process_shred_transaction( &self, transaction_with_slot: TransactionWithSlot, bot_wallet: Option, ) -> AnyResult<()> { + if self.callback.is_none() { + return Ok(()); + } self.metrics_manager.add_tx_process_count(); - let program_received_time_us = chrono::Utc::now().timestamp_micros(); + let tx = transaction_with_slot.transaction; + let slot = transaction_with_slot.slot; - let versioned_tx = transaction_with_slot.transaction; - let signature = versioned_tx.signatures[0]; - - // 获取缓存的解析器 + let signature = tx.signatures[0]; + let program_received_time_us = transaction_with_slot.program_received_time_us; + // Use cache to get parser let parser = self.get_parser(); + let callback = self.callback.clone().unwrap(); + let metrics_manager = self.metrics_manager.clone(); - let all_events = parser - .parse_versioned_transaction( - &versioned_tx, + let adapter_callback = Arc::new(move |event: Box| { + let processing_time_us = event.program_handle_time_consuming_us() as f64; + callback(event); + metrics_manager.update_metrics( + MetricsEventType::Transaction, + 1, + processing_time_us, + Some(signature), + ); + }); + + parser + .parse_versioned_transaction_owned( + tx, signature, Some(slot), None, program_received_time_us, bot_wallet, None, + &[], + adapter_callback, ) - .await - .unwrap_or_else(|_e| vec![]); - - let mut max_time_consuming_us = 0; - - // 保存事件数量用于日志记录 - let event_count = all_events.len(); - - // 即时处理事件 - for mut event in all_events { - event.set_program_handle_time_consuming_us( - chrono::Utc::now().timestamp_micros() - event.program_received_time_us(), - ); - max_time_consuming_us = - max_time_consuming_us.max(event.program_handle_time_consuming_us()); - self.invoke_callback(event); - } - - // 实际调用性能指标更新 - self.update_metrics( - MetricsEventType::Transaction, - event_count as u64, - max_time_consuming_us as f64, - Some(signature), - ); + .await?; Ok(()) } @@ -240,7 +384,7 @@ impl EventProcessor { } } -// 实现 Clone trait 以支持模块间共享 +// Implement Clone trait to support sharing between modules impl Clone for EventProcessor { fn clone(&self) -> Self { Self { @@ -250,8 +394,8 @@ impl Clone for EventProcessor { protocols: self.protocols.clone(), event_type_filter: self.event_type_filter.clone(), backpressure_config: self.backpressure_config.clone(), - batch_config: self.batch_config.clone(), callback: self.callback.clone(), + backpressure_semaphore: self.backpressure_semaphore.clone(), } } } diff --git a/src/streaming/common/metrics.rs b/src/streaming/common/metrics.rs index a5df787..dd02697 100644 --- a/src/streaming/common/metrics.rs +++ b/src/streaming/common/metrics.rs @@ -108,15 +108,21 @@ impl AtomicEventMetrics { struct AtomicProcessingTimeStats { min_time_bits: AtomicU64, max_time_bits: AtomicU64, - total_time_us: AtomicU64, // 存储微秒的整数部分 + max_time_timestamp_nanos: AtomicU64, // 最大值更新时间戳（纳秒） + total_time_us: AtomicU64, // 存储微秒的整数部分 total_events: AtomicU64, } impl AtomicProcessingTimeStats { fn new() -> Self { + let now_nanos = + std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos() + as u64; + Self { min_time_bits: AtomicU64::new(f64::INFINITY.to_bits()), max_time_bits: AtomicU64::new(0), + max_time_timestamp_nanos: AtomicU64::new(now_nanos), total_time_us: AtomicU64::new(0), total_events: AtomicU64::new(0), } @@ -126,6 +132,9 @@ impl AtomicProcessingTimeStats { #[inline] fn update(&self, time_us: f64, event_count: u64) { let time_bits = time_us.to_bits(); + let now_nanos = + std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos() + as u64; // 更新最小值（使用 compare_exchange_weak 循环） let mut current_min = self.min_time_bits.load(Ordering::Relaxed); @@ -141,8 +150,20 @@ impl AtomicProcessingTimeStats { } } - // 更新最大值 + // 更新最大值，检查时间差并在超过10秒时清零 let mut current_max = self.max_time_bits.load(Ordering::Relaxed); + let max_timestamp = self.max_time_timestamp_nanos.load(Ordering::Relaxed); + + // 检查最大值的时间戳是否超过10秒（10_000_000_000纳秒） + let time_diff_nanos = now_nanos.saturating_sub(max_timestamp); + if time_diff_nanos > 10_000_000_000 { + // 超过10秒，清零最大值 + self.max_time_bits.store(0, Ordering::Relaxed); + self.max_time_timestamp_nanos.store(now_nanos, Ordering::Relaxed); + current_max = 0; + } + + // 如果当前时间大于最大值，更新最大值和时间戳 while time_bits > current_max { match self.max_time_bits.compare_exchange_weak( current_max, @@ -150,7 +171,11 @@ impl AtomicProcessingTimeStats { Ordering::Relaxed, Ordering::Relaxed, ) { - Ok(_) => break, + Ok(_) => { + // 成功更新最大值，同时更新时间戳 + self.max_time_timestamp_nanos.store(now_nanos, Ordering::Relaxed); + break; + } Err(x) => current_max = x, } } @@ -198,6 +223,18 @@ pub struct EventMetricsSnapshot { pub events_per_second: f64, } +/// 背压指标快照 +#[derive(Debug, Clone)] +pub struct BackpressureMetricsSnapshot { + pub total_duration_us: u64, + pub success_count: u64, + pub failure_count: u64, + pub min_permits: u64, + pub max_permits: u64, + pub avg_duration_us: f64, + pub success_rate: f64, +} + /// 兼容性结构 - 完整的性能指标 #[derive(Debug, Clone)] pub struct PerformanceMetrics { @@ -206,6 +243,8 @@ pub struct PerformanceMetrics { pub account_metrics: EventMetricsSnapshot, pub block_meta_metrics: EventMetricsSnapshot, pub processing_stats: ProcessingTimeStats, + pub backpressure_metrics: BackpressureMetricsSnapshot, + pub dropped_events_count: u64, } impl PerformanceMetrics { @@ -214,6 +253,15 @@ impl PerformanceMetrics { 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_backpressure = BackpressureMetricsSnapshot { + total_duration_us: 0, + success_count: 0, + failure_count: 0, + min_permits: 0, + max_permits: 0, + avg_duration_us: 0.0, + success_rate: 0.0, + }; Self { uptime: std::time::Duration::ZERO, @@ -221,6 +269,8 @@ impl PerformanceMetrics { account_metrics: default_metrics.clone(), block_meta_metrics: default_metrics, processing_stats: default_stats, + backpressure_metrics: default_backpressure, + dropped_events_count: 0, } } } @@ -231,6 +281,14 @@ pub struct HighPerformanceMetrics { start_nanos: u64, event_metrics: [AtomicEventMetrics; 3], processing_stats: AtomicProcessingTimeStats, + // 背压相关指标 + backpressure_total_duration_us: AtomicU64, + backpressure_success_count: AtomicU64, + backpressure_failure_count: AtomicU64, + backpressure_min_permits: AtomicU64, + backpressure_max_permits: AtomicU64, + // 丢弃事件指标 + dropped_events_count: AtomicU64, } impl HighPerformanceMetrics { @@ -247,6 +305,14 @@ impl HighPerformanceMetrics { AtomicEventMetrics::new(now_nanos), ], processing_stats: AtomicProcessingTimeStats::new(), + // 初始化背压相关指标 + backpressure_total_duration_us: AtomicU64::new(0), + backpressure_success_count: AtomicU64::new(0), + backpressure_failure_count: AtomicU64::new(0), + backpressure_min_permits: AtomicU64::new(u64::MAX), // 初始化为最大值，便于后续比较 + backpressure_max_permits: AtomicU64::new(0), + // 初始化丢弃事件指标 + dropped_events_count: AtomicU64::new(0), } } @@ -275,6 +341,38 @@ impl HighPerformanceMetrics { self.processing_stats.get_stats() } + /// 获取背压指标快照 + #[inline] + pub fn get_backpressure_metrics(&self) -> BackpressureMetricsSnapshot { + let total_duration_us = self.backpressure_total_duration_us.load(Ordering::Relaxed); + let success_count = self.backpressure_success_count.load(Ordering::Relaxed); + let failure_count = self.backpressure_failure_count.load(Ordering::Relaxed); + let min_permits = self.backpressure_min_permits.load(Ordering::Relaxed); + let max_permits = self.backpressure_max_permits.load(Ordering::Relaxed); + + let total_count = success_count + failure_count; + let avg_duration_us = + if total_count > 0 { total_duration_us as f64 / total_count as f64 } else { 0.0 }; + let success_rate = + if total_count > 0 { success_count as f64 / total_count as f64 } else { 0.0 }; + + BackpressureMetricsSnapshot { + total_duration_us, + success_count, + failure_count, + min_permits: if min_permits == u64::MAX { 0 } else { min_permits }, + max_permits, + avg_duration_us, + success_rate, + } + } + + /// 获取丢弃事件计数 + #[inline] + pub fn get_dropped_events_count(&self) -> u64 { + self.dropped_events_count.load(Ordering::Relaxed) + } + /// 计算实时每秒事件数（非阻塞） fn calculate_real_time_eps(&self, event_type: EventType) -> f64 { let now_nanos = @@ -443,11 +541,43 @@ impl MetricsManager { self.metrics.get_processing_stats() } + /// 获取背压指标 + pub fn get_backpressure_metrics(&self) -> BackpressureMetricsSnapshot { + self.metrics.get_backpressure_metrics() + } + + /// 获取丢弃事件计数 + pub fn get_dropped_events_count(&self) -> u64 { + self.metrics.get_dropped_events_count() + } + /// 打印性能指标（非阻塞） pub fn print_metrics(&self) { println!("\n📊 {} Performance Metrics", self.stream_name); println!(" Run Time: {:?}", self.get_uptime()); + // 打印背压指标表格 + let backpressure = self.get_backpressure_metrics(); + if backpressure.success_count > 0 || backpressure.failure_count > 0 { + println!("\n🚦 Backpressure Metrics"); + println!("┌──────────────────────┬─────────────┐"); + println!("│ Metric │ Value │"); + println!("├──────────────────────┼─────────────┤"); + println!("│ Success Count │ {:11} │", backpressure.success_count); + println!("│ Failure Count │ {:11} │", backpressure.failure_count); + println!("│ Success Rate │ {:11.2} │", backpressure.success_rate * 100.0); + println!("│ Avg Duration (ms) │ {:11.2} │", backpressure.avg_duration_us / 1000.0); + println!("│ Min Permits │ {:11} │", backpressure.min_permits); + println!("│ Max Permits │ {:11} │", backpressure.max_permits); + println!("└──────────────────────┴─────────────┘"); + } + + // 打印丢弃事件指标 + let dropped_count = self.get_dropped_events_count(); + if dropped_count > 0 { + println!("\n⚠️ Dropped Events: {}", dropped_count); + } + // 打印事件指标表格 println!("┌─────────────┬──────────────┬──────────────────┬─────────────────┐"); println!("│ Event Type │ Process Count│ Events Processed │ Events/Second │"); @@ -469,13 +599,14 @@ impl MetricsManager { // 打印处理时间统计表格 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 │ {:9.2} │", stats.min_us); - println!("│ Maximum │ {:9.2} │", stats.max_us); - println!("└─────────────────────┴─────────────┘"); + println!("┌───────────────────────┬─────────────┐"); + println!("│ Metric │ Value (us) │"); + println!("├───────────────────────┼─────────────┤"); + println!("│ Average │ {:9.2} │", stats.avg_us); + println!("│ Minimum │ {:9.2} │", stats.min_us); + println!("│ Maximum within 10s │ {:9.2} │", stats.max_us); + println!("└───────────────────────┴─────────────┘"); + println!(); } @@ -517,6 +648,8 @@ impl MetricsManager { account_metrics: self.get_event_metrics(EventType::Account), block_meta_metrics: self.get_event_metrics(EventType::BlockMeta), processing_stats: self.get_processing_stats(), + backpressure_metrics: self.metrics.get_backpressure_metrics(), + dropped_events_count: self.metrics.get_dropped_events_count(), } } @@ -550,6 +683,110 @@ impl MetricsManager { self.record_events(event_type, events_processed, processing_time_us); self.log_slow_processing(processing_time_us, events_processed as usize, signature); } + + /// 记录背压相关的metrics + #[inline] + pub fn record_backpressure_metrics( + &self, + backpressure_duration: std::time::Duration, + success: bool, + available_permits: usize, + ) { + if !self.enable_metrics { + return; + } + + let duration_us = backpressure_duration.as_micros() as u64; + let permits = available_permits as u64; + + // 记录总持续时间 + self.metrics.backpressure_total_duration_us.fetch_add(duration_us, Ordering::Relaxed); + + // 记录成功/失败计数 + if success { + self.metrics.backpressure_success_count.fetch_add(1, Ordering::Relaxed); + } else { + self.metrics.backpressure_failure_count.fetch_add(1, Ordering::Relaxed); + } + + // 更新最小许可数（使用 compare_exchange_weak 循环） + let mut current_min = self.metrics.backpressure_min_permits.load(Ordering::Relaxed); + while permits < current_min { + match self.metrics.backpressure_min_permits.compare_exchange_weak( + current_min, + permits, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => break, + Err(x) => current_min = x, + } + } + + // 更新最大许可数 + let mut current_max = self.metrics.backpressure_max_permits.load(Ordering::Relaxed); + while permits > current_max { + match self.metrics.backpressure_max_permits.compare_exchange_weak( + current_max, + permits, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => break, + Err(x) => current_max = x, + } + } + + // 记录慢背压操作的日志 + if duration_us > 10_000 { + // 超过10ms的背压认为是慢操作 + log::warn!( + "{} slow backpressure: {:.2}ms, success: {}, available_permits: {}", + self.stream_name, + duration_us as f64 / 1000.0, + success, + available_permits + ); + } + } + + /// 增加丢弃事件计数 + #[inline] + pub fn increment_dropped_events(&self) { + if !self.enable_metrics { + return; + } + + // 原子地增加丢弃事件计数 + let new_count = self.metrics.dropped_events_count.fetch_add(1, Ordering::Relaxed) + 1; + + // 每丢弃1000个事件记录一次警告日志 + if new_count % 1000 == 0 { + log::warn!("{} dropped events count reached: {}", self.stream_name, new_count); + } + } + + /// 批量增加丢弃事件计数 + #[inline] + pub fn increment_dropped_events_by(&self, count: u64) { + if !self.enable_metrics || count == 0 { + return; + } + + // 原子地增加丢弃事件计数 + let new_count = self.metrics.dropped_events_count.fetch_add(count, Ordering::Relaxed) + count; + + // 记录批量丢弃事件的日志 + if count > 1 { + log::warn!("{} dropped batch of {} events, total dropped: {}", + self.stream_name, count, new_count); + } + + // 每丢弃1000个事件记录一次警告日志 + if new_count % 1000 == 0 || (new_count / 1000) != ((new_count - count) / 1000) { + log::warn!("{} dropped events count reached: {}", self.stream_name, new_count); + } + } } impl Clone for MetricsManager { diff --git a/src/streaming/event_parser/core/global_state.rs b/src/streaming/event_parser/core/global_state.rs new file mode 100644 index 0000000..5e75c0f --- /dev/null +++ b/src/streaming/event_parser/core/global_state.rs @@ -0,0 +1,141 @@ +use solana_sdk::pubkey::Pubkey; +use std::sync::atomic::{AtomicU64, Ordering}; + +/// Global state management, thread-safe implementation without locks +pub struct GlobalState { + /// Last processed slot + last_slot: AtomicU64, + /// Developer address array + dev_addresses: parking_lot::RwLock>, + /// Bonk developer address array + bonk_dev_addresses: parking_lot::RwLock>, +} + +impl GlobalState { + /// Create a new global state instance + pub fn new() -> Self { + Self { + last_slot: AtomicU64::new(0), + dev_addresses: parking_lot::RwLock::new(Vec::new()), + bonk_dev_addresses: parking_lot::RwLock::new(Vec::new()), + } + } + + /// Get current slot + pub fn get_last_slot(&self) -> u64 { + self.last_slot.load(Ordering::Relaxed) + } + + /// Update slot, clear arrays if slot changes + pub fn update_slot(&self, new_slot: u64) { + let old_slot = self.last_slot.swap(new_slot, Ordering::Relaxed); + + if old_slot != new_slot { + // Clear arrays when slot changes + let mut dev_addresses = self.dev_addresses.write(); + let mut bonk_dev_addresses = self.bonk_dev_addresses.write(); + + dev_addresses.clear(); + bonk_dev_addresses.clear(); + } + } + + /// Add developer address + pub fn add_dev_address(&self, address: Pubkey) { + let mut dev_addresses = self.dev_addresses.write(); + if !dev_addresses.contains(&address) { + dev_addresses.push(address); + } + } + + /// Check if address is a developer address + pub fn is_dev_address(&self, address: &Pubkey) -> bool { + let dev_addresses = self.dev_addresses.read(); + dev_addresses.contains(address) + } + + /// Add Bonk developer address + pub fn add_bonk_dev_address(&self, address: Pubkey) { + let mut bonk_dev_addresses = self.bonk_dev_addresses.write(); + if !bonk_dev_addresses.contains(&address) { + bonk_dev_addresses.push(address); + } + } + + /// Check if address is a Bonk developer address + pub fn is_bonk_dev_address(&self, address: &Pubkey) -> bool { + let bonk_dev_addresses = self.bonk_dev_addresses.read(); + bonk_dev_addresses.contains(address) + } + + /// Get all developer addresses + pub fn get_dev_addresses(&self) -> Vec { + let dev_addresses = self.dev_addresses.read(); + dev_addresses.clone() + } + + /// Get all Bonk developer addresses + pub fn get_bonk_dev_addresses(&self) -> Vec { + let bonk_dev_addresses = self.bonk_dev_addresses.read(); + bonk_dev_addresses.clone() + } + + /// Clear all data + pub fn clear_all_data(&self) { + let mut dev_addresses = self.dev_addresses.write(); + let mut bonk_dev_addresses = self.bonk_dev_addresses.write(); + + dev_addresses.clear(); + bonk_dev_addresses.clear(); + } +} + +impl Default for GlobalState { + fn default() -> Self { + Self::new() + } +} + +/// Global state instance +static GLOBAL_STATE: once_cell::sync::Lazy = + once_cell::sync::Lazy::new(GlobalState::new); + +/// Get global state instance +pub fn get_global_state() -> &'static GlobalState { + &GLOBAL_STATE +} + +/// Convenience function: Update slot +pub fn update_slot(slot: u64) { + get_global_state().update_slot(slot); +} + +/// Convenience function: Add developer address +pub fn add_dev_address(address: Pubkey) { + get_global_state().add_dev_address(address); +} + +/// Convenience function: Check if address is a developer address +pub fn is_dev_address(address: &Pubkey) -> bool { + get_global_state().is_dev_address(address) +} + +/// Convenience function: Add Bonk developer address +pub fn add_bonk_dev_address(address: Pubkey) { + get_global_state().add_bonk_dev_address(address); +} + +/// Convenience function: Check if address is a Bonk developer address +pub fn is_bonk_dev_address(address: &Pubkey) -> bool { + get_global_state().is_bonk_dev_address(address) +} + +/// Convenience function: Get all developer addresses +pub fn get_dev_addresses() -> Vec { + get_global_state().get_dev_addresses() +} + +/// Convenience function: Get all Bonk developer addresses +pub fn get_bonk_dev_addresses() -> Vec { + get_global_state().get_bonk_dev_addresses() +} diff --git a/src/streaming/event_parser/core/macros.rs b/src/streaming/event_parser/core/macros.rs index 9ef239e..724ccc0 100644 --- a/src/streaming/event_parser/core/macros.rs +++ b/src/streaming/event_parser/core/macros.rs @@ -15,15 +15,6 @@ macro_rules! impl_event_parser_delegate { ($parser_type:ty) => { #[async_trait::async_trait] impl $crate::streaming::event_parser::core::traits::EventParser for $parser_type { - fn inner_instruction_configs( - &self, - ) -> std::collections::HashMap< - Vec, - Vec<$crate::streaming::event_parser::core::traits::GenericEventParseConfig>, - > { - self.inner.inner_instruction_configs() - } - fn instruction_configs( &self, ) -> std::collections::HashMap< @@ -42,8 +33,8 @@ macro_rules! impl_event_parser_delegate { program_received_time_us: i64, outer_index: i64, inner_index: Option, - bot_wallet: Option, transaction_index: Option, + config: &GenericEventParseConfig, ) -> Vec> { self.inner.parse_events_from_inner_instruction( inner_instruction, @@ -53,8 +44,8 @@ macro_rules! impl_event_parser_delegate { program_received_time_us, outer_index, inner_index, - bot_wallet, transaction_index, + config, ) } @@ -70,7 +61,13 @@ macro_rules! impl_event_parser_delegate { inner_index: Option, bot_wallet: Option, transaction_index: Option, - ) -> Vec> { + inner_instructions: Option<&solana_transaction_status::InnerInstructions>, + callback: std::sync::Arc< + dyn for<'a> Fn(&'a Box) + + Send + + Sync, + >, + ) -> anyhow::Result<()> { self.inner.parse_events_from_instruction( instruction, accounts, @@ -82,6 +79,8 @@ macro_rules! impl_event_parser_delegate { inner_index, bot_wallet, transaction_index, + inner_instructions, + callback, ) } diff --git a/src/streaming/event_parser/core/mod.rs b/src/streaming/event_parser/core/mod.rs index cbe0602..2e9f50e 100755 --- a/src/streaming/event_parser/core/mod.rs +++ b/src/streaming/event_parser/core/mod.rs @@ -2,4 +2,5 @@ pub mod common_event_parser; pub mod traits; pub mod account_event_parser; pub mod macros; +pub mod global_state; pub use traits::{EventParser, UnifiedEvent}; diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index b37da1a..db74152 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -1,15 +1,23 @@ -use anyhow::Result; use prost_types::Timestamp; +use solana_sdk::bs58; use solana_sdk::signature::Signature; use solana_sdk::{ instruction::CompiledInstruction, pubkey::Pubkey, transaction::VersionedTransaction, }; -use solana_transaction_status::{InnerInstructions, TransactionWithStatusMeta}; +use solana_transaction_status::{ + EncodedConfirmedTransactionWithStatusMeta, InnerInstruction, InnerInstructions, + TransactionWithStatusMeta, UiInstruction, +}; use std::collections::HashMap; use std::fmt::Debug; use std::sync::Arc; +use super::global_state::{add_dev_address, is_dev_address, update_slot}; + use crate::streaming::event_parser::common::{parse_swap_data_from_next_instructions, SwapData}; +use crate::streaming::event_parser::core::global_state::{ + add_bonk_dev_address, is_bonk_dev_address, +}; use crate::streaming::event_parser::protocols::pumpswap::{PumpSwapBuyEvent, PumpSwapSellEvent}; use crate::streaming::event_parser::{ common::{utils::*, EventMetadata, EventType, ProtocolType}, @@ -73,8 +81,6 @@ pub trait UnifiedEvent: Debug + Send + Sync { /// 事件解析器trait - 定义了事件解析的核心方法 #[async_trait::async_trait] pub trait EventParser: Send + Sync { - /// 获取内联指令解析配置 - fn inner_instruction_configs(&self) -> HashMap, Vec>; /// 获取指令解析配置 fn instruction_configs(&self) -> HashMap, Vec>; /// 从内联指令中解析事件数据 @@ -88,8 +94,8 @@ pub trait EventParser: Send + Sync { program_received_time_us: i64, outer_index: i64, inner_index: Option, - bot_wallet: Option, transaction_index: Option, + config: &GenericEventParseConfig, ) -> Vec>; /// 从指令中解析事件数据 @@ -106,7 +112,9 @@ pub trait EventParser: Send + Sync { inner_index: Option, bot_wallet: Option, transaction_index: Option, - ) -> Vec>; + inner_instructions: Option<&InnerInstructions>, + callback: Arc Fn(&'a Box) + Send + Sync>, + ) -> anyhow::Result<()>; /// 从VersionedTransaction中解析指令事件的通用方法 #[allow(clippy::too_many_arguments)] @@ -121,13 +129,11 @@ pub trait EventParser: Send + Sync { inner_instructions: &[InnerInstructions], bot_wallet: Option, transaction_index: Option, - ) -> Result>> { - // 预分配容量，避免动态扩容 - let mut instruction_events = Vec::with_capacity(16); + callback: Arc Fn(&'a Box) + Send + Sync>, + ) -> anyhow::Result<()> { // 获取交易的指令和账户 let compiled_instructions = transaction.message.instructions(); let mut accounts: Vec = accounts.to_vec(); - // 检查交易中是否包含程序 let has_program = accounts.iter().any(|account| self.should_handle(account)); if has_program { @@ -142,47 +148,60 @@ pub trait EventParser: Send + Sync { accounts.push(Pubkey::default()); } } - if let Ok(mut events) = self - .parse_instruction( - instruction, - &accounts, - signature, - slot, - block_time, - program_received_time_us, - index as i64, - None, - bot_wallet, - Some(index as u64), - ) - .await - { - if !events.is_empty() { - if let Some(inn) = - inner_instructions.iter().find(|inner_instruction| { - inner_instruction.index == index as u8 - }) - { - events.iter_mut().for_each(|event| { - let swap_data = parse_swap_data_from_next_instructions( - event.as_ref(), - inn, - -1_i8, - &accounts, - ); - if let Some(swap_data) = swap_data { - event.set_swap_data(swap_data); - } - }); - } - instruction_events.extend(events); - } - } + let inner_instructions = inner_instructions + .iter() + .find(|inner_instruction| inner_instruction.index == index as u8); + self.parse_instruction( + instruction, + &accounts, + signature, + slot, + block_time, + program_received_time_us, + index as i64, + None, + bot_wallet, + transaction_index, + inner_instructions, + Arc::clone(&callback), + ) + .await?; } } } } - Ok(instruction_events) + Ok(()) + } + + async fn parse_versioned_transaction_owned( + &self, + versioned_tx: VersionedTransaction, + signature: Signature, + slot: Option, + block_time: Option, + program_received_time_us: i64, + bot_wallet: Option, + transaction_index: Option, + inner_instructions: &[InnerInstructions], + callback: Arc) + Send + Sync>, + ) -> anyhow::Result<()> { + // 创建适配器回调，将所有权回调转换为引用回调 + let adapter_callback = Arc::new(move |event: &Box| { + callback(event.clone_boxed()); + }); + self.parse_versioned_transaction( + &versioned_tx, + signature, + slot, + block_time, + program_received_time_us, + bot_wallet, + transaction_index, + inner_instructions, + adapter_callback, + ) + .await?; + Ok(()) } async fn parse_versioned_transaction( @@ -194,23 +213,54 @@ pub trait EventParser: Send + Sync { program_received_time_us: i64, bot_wallet: Option, transaction_index: Option, - ) -> Result>> { + inner_instructions: &[InnerInstructions], + callback: Arc Fn(&'a Box) + Send + Sync>, + ) -> anyhow::Result<()> { let accounts: Vec = versioned_tx.message.static_account_keys().to_vec(); - let events = self - .parse_instruction_events_from_versioned_transaction( - versioned_tx, - signature, - slot, - block_time, - program_received_time_us, - &accounts, - &[], - bot_wallet, - transaction_index, - ) - .await - .unwrap_or_else(|_e| vec![]); - Ok(self.process_events(events, bot_wallet)) + self.parse_instruction_events_from_versioned_transaction( + versioned_tx, + signature, + slot, + block_time, + program_received_time_us, + &accounts, + inner_instructions, + bot_wallet, + transaction_index, + callback, + ) + .await?; + Ok(()) + } + + /// 解析交易，使用所有权语义的回调以避免不必要的克隆 + async fn parse_transaction_owned( + &self, + tx: TransactionWithStatusMeta, + signature: Signature, + slot: Option, + block_time: Option, + program_received_time_us: i64, + bot_wallet: Option, + transaction_index: Option, + callback: Arc) + Send + Sync>, + ) -> anyhow::Result<()> { + // 创建适配器回调，将所有权回调转换为引用回调 + let adapter_callback = Arc::new(move |event: &Box| { + callback(event.clone_boxed()); + }); + // 调用原始方法 + self.parse_transaction( + tx, + signature, + slot, + block_time, + program_received_time_us, + bot_wallet, + transaction_index, + adapter_callback, + ) + .await } async fn parse_transaction( @@ -222,10 +272,10 @@ pub trait EventParser: Send + Sync { program_received_time_us: i64, bot_wallet: Option, transaction_index: Option, - ) -> Result>> { + callback: Arc Fn(&'a Box) + Send + Sync>, + ) -> anyhow::Result<()> { let versioned_tx = tx.get_transaction(); let meta = tx.get_status_meta(); - let mut address_table_lookups: Vec = vec![]; let mut inner_instructions: Vec = vec![]; if let Some(meta) = meta { @@ -237,269 +287,183 @@ pub trait EventParser: Send + Sync { meta.loaded_addresses.writable.into_iter().chain(meta.loaded_addresses.readonly), ); } - let mut accounts = Vec::with_capacity( versioned_tx.message.static_account_keys().len() + address_table_lookups.len(), ); accounts.extend_from_slice(versioned_tx.message.static_account_keys()); accounts.extend(address_table_lookups); - // 使用 Arc 包装共享数据，避免不必要的克隆 let accounts_arc = Arc::new(accounts); let inner_instructions_arc = Arc::new(inner_instructions); + // 解析指令事件 + self.parse_instruction_events_from_versioned_transaction( + &versioned_tx, + signature, + slot, + block_time, + program_received_time_us, + &accounts_arc, + &inner_instructions_arc, + bot_wallet, + transaction_index, + callback.clone(), + ) + .await?; - // 预分配容量，避免动态扩容 - let mut instruction_events: Vec> = Vec::with_capacity(16); - let mut inner_instruction_events: Vec> = Vec::with_capacity(8); - - let accounts_for_task1 = Arc::clone(&accounts_arc); - let inner_instructions_for_task1 = Arc::clone(&inner_instructions_arc); - let task1 = async move { - self.parse_instruction_events_from_versioned_transaction( - &versioned_tx, - signature, - slot, - block_time, - program_received_time_us, - &accounts_for_task1, - &inner_instructions_for_task1, - bot_wallet, - transaction_index, - ) - .await - .unwrap_or_else(|_e| vec![]) - }; - - // 解析内联指令事件 - let inner_instructions_for_task2 = Arc::clone(&inner_instructions_arc); - let accounts_for_task2_1 = Arc::clone(&accounts_arc); - let accounts_for_task2_2 = Arc::clone(&accounts_arc); - - let mut task2_params = Vec::with_capacity(inner_instructions_for_task2.len() * 5); - for inner_instruction in inner_instructions_for_task2.iter() { + // 解析嵌套指令事件 + for inner_instruction in inner_instructions_arc.iter() { for (index, instruction) in inner_instruction.instructions.iter().enumerate() { - task2_params.push(( + self.parse_instruction( &instruction.instruction, + &accounts_arc, signature, slot, block_time, program_received_time_us, inner_instruction.index as i64, Some(index as i64), - inner_instruction, - )); + bot_wallet, + transaction_index, + Some(&inner_instruction), + callback.clone(), + ) + .await?; } } - // 转换为 Arc<[T]> 更轻量 - let task2_params: Arc<[_]> = task2_params.into(); - let task2_params_clone = Arc::clone(&task2_params); - let task2_1 = async move { - let mut instruction_events: Vec> = Vec::with_capacity(16); - for ( - instruction, - signature, - slot, - block_time, - program_received_time_us, - outer_index, - inner_index, - inner_instruction, - ) in task2_params_clone.iter() - { - if let Ok(mut events) = self - .parse_instruction( - instruction, - &accounts_for_task2_1, - *signature, - *slot, - *block_time, - *program_received_time_us, - *outer_index, - *inner_index, - bot_wallet, - transaction_index, - ) - .await - { - if !events.is_empty() { - events.iter_mut().for_each(|event| { - let swap_data = parse_swap_data_from_next_instructions( - event.as_ref(), - &inner_instruction, - inner_index.unwrap_or_default() as i8, - &accounts_for_task2_1, - ); - if let Some(swap_data) = swap_data { - event.set_swap_data(swap_data); - } - }); - instruction_events.extend(events); - } - } - } - instruction_events - }; - let task2_2 = async move { - let mut inner_instruction_events: Vec> = Vec::with_capacity(8); - for ( - instruction, - signature, - slot, - block_time, - program_received_time_us, - outer_index, - inner_index, - inner_instruction, - ) in task2_params.iter() - { - if let Ok(mut events) = self - .parse_inner_instruction( - instruction, - *signature, - *slot, - *block_time, - *program_received_time_us, - *outer_index, - *inner_index, - bot_wallet, - transaction_index, - ) - .await - { - if !events.is_empty() { - events.iter_mut().for_each(|event| { - let swap_data = parse_swap_data_from_next_instructions( - event.as_ref(), - &inner_instruction, - inner_index.unwrap_or_default() as i8, - &accounts_for_task2_2, - ); - if let Some(swap_data) = swap_data { - event.set_swap_data(swap_data); - } - }); - inner_instruction_events.extend(events); - } - } - } - inner_instruction_events - }; - let (r1, r2_1, r2_2) = tokio::join!(task1, task2_1, task2_2); - instruction_events.extend(r1); - instruction_events.extend(r2_1); - inner_instruction_events.extend(r2_2); + Ok(()) + } - if !instruction_events.is_empty() && !inner_instruction_events.is_empty() { - for instruction_event in &mut instruction_events { - for inner_instruction_event in &inner_instruction_events { - if instruction_event.id() == inner_instruction_event.id() { - if instruction_event.instruction_inner_index().is_none() - && inner_instruction_event.instruction_inner_index().is_some() - { - if inner_instruction_event.instruction_outer_index() - == instruction_event.instruction_outer_index() - { - instruction_event.merge(inner_instruction_event.as_ref()); - break; - } - } else if instruction_event.instruction_inner_index().is_some() - && inner_instruction_event.instruction_inner_index().is_some() - { - if instruction_event.instruction_outer_index() - == inner_instruction_event.instruction_outer_index() - { - if inner_instruction_event - .instruction_inner_index() - .unwrap_or_default() - > instruction_event - .instruction_inner_index() - .unwrap_or_default() - { - instruction_event.merge(inner_instruction_event.as_ref()); - break; - } + async fn parse_encoded_confirmed_transaction_with_status_meta( + &self, + signature: Signature, + transaction: EncodedConfirmedTransactionWithStatusMeta, + callback: Arc Fn(&'a Box) + Send + Sync>, + ) -> anyhow::Result<()> { + let versioned_tx = match transaction.transaction.transaction.decode() { + Some(tx) => tx, + None => { + println!("Failed to decode transaction"); + return Ok(()); + } + }; + let mut inner_instructions_vec: Vec = Vec::new(); + if let Some(meta) = &transaction.transaction.meta { + // 从meta中获取inner_instructions，处理OptionSerializer类型 + if let solana_transaction_status::option_serializer::OptionSerializer::Some( + ui_inner_insts, + ) = &meta.inner_instructions + { + // 将UiInnerInstructions转换为InnerInstructions + for ui_inner in ui_inner_insts { + let mut converted_instructions = Vec::new(); + + // 转换每个UiInstruction为InnerInstruction + for ui_instruction in &ui_inner.instructions { + if let UiInstruction::Compiled(ui_compiled) = ui_instruction { + // 解码base58编码的data + if let Ok(data) = bs58::decode(&ui_compiled.data).into_vec() { + let compiled_instruction = CompiledInstruction { + program_id_index: ui_compiled.program_id_index, + accounts: ui_compiled.accounts.clone(), + data, + }; + + let inner_instruction = InnerInstruction { + instruction: compiled_instruction, + stack_height: ui_compiled.stack_height, + }; + + converted_instructions.push(inner_instruction); } } } - } - } - } - let result = self.process_events(instruction_events, bot_wallet); - - Ok(result) - } - - fn process_events( - &self, - mut events: Vec>, - bot_wallet: Option, - ) -> Vec> { - let mut dev_address = vec![]; - let mut bonk_dev_address = None; - for event in &mut events { - if let Some(token_info) = event.as_any().downcast_ref::() { - dev_address.push(token_info.user); - if token_info.creator != Pubkey::default() && token_info.creator != token_info.user - { - dev_address.push(token_info.creator); - } - } else if let Some(trade_info) = event.as_any_mut().downcast_mut::() - { - if dev_address.contains(&trade_info.user) - || dev_address.contains(&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 + let inner_instructions = InnerInstructions { + index: ui_inner.index, + instructions: converted_instructions, }; - } - } else if let Some(trade_info) = event.as_any_mut().downcast_mut::() { - 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; - } - } else if let Some(trade_info) = event.as_any_mut().downcast_mut::() - { - 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; - } - } else if let Some(pool_info) = event.as_any().downcast_ref::() { - bonk_dev_address = Some(pool_info.creator); - } else if let Some(trade_info) = event.as_any_mut().downcast_mut::() { - if Some(trade_info.payer) == bonk_dev_address { - 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; + + inner_instructions_vec.push(inner_instructions); } } - event.clear_id(); + } + let inner_instructions: &[InnerInstructions] = &inner_instructions_vec; + + let meta = transaction.transaction.meta; + let mut address_table_lookups: Vec = vec![]; + if let Some(meta) = meta { + if let solana_transaction_status::option_serializer::OptionSerializer::Some( + loaded_addresses, + ) = &meta.loaded_addresses + { + address_table_lookups + .reserve(loaded_addresses.writable.len() + loaded_addresses.readonly.len()); + address_table_lookups.extend( + loaded_addresses + .writable + .iter() + .filter_map(|s| s.parse::().ok()) + .chain( + loaded_addresses + .readonly + .iter() + .filter_map(|s| s.parse::().ok()), + ), + ); + } + } + let mut accounts = Vec::with_capacity( + versioned_tx.message.static_account_keys().len() + address_table_lookups.len(), + ); + accounts.extend_from_slice(versioned_tx.message.static_account_keys()); + accounts.extend(address_table_lookups); + // 使用 Arc 包装共享数据，避免不必要的克隆 + let accounts_arc = Arc::new(accounts); + let inner_instructions_arc = Arc::new(inner_instructions); + + 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 bot_wallet = None; + let transaction_index = None; + // 解析指令事件 + self.parse_instruction_events_from_versioned_transaction( + &versioned_tx, + signature, + Some(slot), + block_time, + program_received_time_us, + &accounts_arc, + &inner_instructions_arc, + bot_wallet, + transaction_index, + callback.clone(), + ) + .await?; + + // 解析嵌套指令事件 + for inner_instruction in inner_instructions_arc.iter() { + for (index, instruction) in inner_instruction.instructions.iter().enumerate() { + self.parse_instruction( + &instruction.instruction, + &accounts_arc, + signature, + Some(slot), + block_time, + program_received_time_us, + inner_instruction.index as i64, + Some(index as i64), + bot_wallet, + transaction_index, + Some(&inner_instruction), + callback.clone(), + ) + .await?; + } } - events + Ok(()) } async fn parse_inner_instruction( @@ -511,9 +475,9 @@ pub trait EventParser: Send + Sync { program_received_time_us: i64, outer_index: i64, inner_index: Option, - bot_wallet: Option, transaction_index: Option, - ) -> Result>> { + config: &GenericEventParseConfig, + ) -> anyhow::Result>> { let slot = slot.unwrap_or(0); let events = self.parse_events_from_inner_instruction( instruction, @@ -523,8 +487,8 @@ pub trait EventParser: Send + Sync { program_received_time_us, outer_index, inner_index, - bot_wallet, transaction_index, + config, ); Ok(events) } @@ -542,9 +506,11 @@ pub trait EventParser: Send + Sync { inner_index: Option, bot_wallet: Option, transaction_index: Option, - ) -> Result>> { + inner_instructions: Option<&InnerInstructions>, + callback: Arc Fn(&'a Box) + Send + Sync>, + ) -> anyhow::Result<()> { let slot = slot.unwrap_or(0); - let events = self.parse_events_from_instruction( + self.parse_events_from_instruction( instruction, accounts, signature, @@ -555,8 +521,9 @@ pub trait EventParser: Send + Sync { inner_index, bot_wallet, transaction_index, - ); - Ok(events) + inner_instructions, + callback, + ) } /// 检查是否应该处理此程序ID @@ -596,7 +563,7 @@ pub type InstructionEventParser = /// 通用事件解析器基类 pub struct GenericEventParser { pub program_ids: Vec, - pub inner_instruction_configs: HashMap, Vec>, + // pub inner_instruction_configs: HashMap, Vec>, pub instruction_configs: HashMap, Vec>, } @@ -604,23 +571,16 @@ impl GenericEventParser { /// 创建新的通用事件解析器 pub fn new(program_ids: Vec, configs: Vec) -> Self { // 预分配容量，避免动态扩容 - let mut inner_instruction_configs = HashMap::with_capacity(configs.len()); let mut instruction_configs = HashMap::with_capacity(configs.len()); for config in configs { - if config.inner_instruction_discriminator.len() > 0 { - inner_instruction_configs - .entry(config.inner_instruction_discriminator.to_vec()) - .or_insert_with(Vec::new) - .push(config.clone()); - } instruction_configs .entry(config.instruction_discriminator.to_vec()) .or_insert_with(Vec::new) .push(config.clone()); } - Self { program_ids, inner_instruction_configs, instruction_configs } + Self { program_ids, instruction_configs } } /// 通用的内联指令解析方法 @@ -701,12 +661,7 @@ impl GenericEventParser { #[async_trait::async_trait] impl EventParser for GenericEventParser { - fn inner_instruction_configs(&self) -> HashMap, Vec> { - // 返回引用而非克隆，减少内存分配 - self.inner_instruction_configs.clone() - } fn instruction_configs(&self) -> HashMap, Vec> { - // 返回引用而非克隆，减少内存分配 self.instruction_configs.clone() } /// 从内联指令中解析事件数据 @@ -720,32 +675,26 @@ impl EventParser for GenericEventParser { program_received_time_us: i64, outer_index: i64, inner_index: Option, - bot_wallet: Option, transaction_index: Option, + config: &GenericEventParseConfig, ) -> Vec> { if inner_instruction.data.len() < 16 { return Vec::new(); } let data = &inner_instruction.data[16..]; let mut events = Vec::new(); - for (disc, configs) in &self.inner_instruction_configs { - if data == disc { - for config in configs { - if let Some(event) = self.parse_inner_instruction_event( - config, - data, - signature, - slot, - block_time, - program_received_time_us, - outer_index, - inner_index, - transaction_index, - ) { - events.push(event); - } - } - } + if let Some(event) = self.parse_inner_instruction_event( + config, + data, + signature, + slot, + block_time, + program_received_time_us, + outer_index, + inner_index, + transaction_index, + ) { + events.push(event); } events } @@ -764,12 +713,13 @@ impl EventParser for GenericEventParser { inner_index: Option, bot_wallet: Option, transaction_index: Option, - ) -> Vec> { + inner_instructions: Option<&InnerInstructions>, + callback: Arc Fn(&'a Box) + Send + Sync>, + ) -> anyhow::Result<()> { let program_id = accounts[instruction.program_id_index as usize]; if !self.should_handle(&program_id) { - return Vec::new(); + return Ok(()); } - let mut events = Vec::new(); for (disc, configs) in &self.instruction_configs { if instruction.data.len() < disc.len() { continue; @@ -781,14 +731,13 @@ impl EventParser for GenericEventParser { if !validate_account_indices(&instruction.accounts, accounts.len()) { continue; } - let account_pubkeys: Vec = instruction.accounts.iter().map(|&idx| accounts[idx as usize]).collect(); for config in configs { if config.program_id != program_id { continue; } - if let Some(event) = self.parse_instruction_event( + if let Some(mut event) = self.parse_instruction_event( config, data, &account_pubkeys, @@ -800,13 +749,53 @@ impl EventParser for GenericEventParser { inner_index, transaction_index, ) { - events.push(event); + let mut inner_instruction_event: Option> = None; + if inner_instructions.is_some() { + // 解析对应的内部 log 执行 + for inner_instruction in inner_instructions.unwrap().instructions.iter() + { + let result = self.parse_events_from_inner_instruction( + &inner_instruction.instruction, + signature, + slot, + block_time, + program_received_time_us, + outer_index, + inner_index, + transaction_index, + config, + ); + if result.len() > 0 { + inner_instruction_event = Some(result[0].clone()); + } + // 解析swap数据 + let swap_data = parse_swap_data_from_next_instructions( + &*event, + inner_instructions.unwrap(), + inner_index.unwrap_or(-1_i64) as i8, + &accounts, + ); + if let Some(swap_data) = swap_data { + event.set_swap_data(swap_data); + } + } + } + // 合并事件 + if let Some(inner_instruction_event) = inner_instruction_event { + event.merge(&*inner_instruction_event); + } + // 设置处理时间 + event.set_program_handle_time_consuming_us( + chrono::Utc::now().timestamp_micros() - program_received_time_us, + ); + event = process_event(event, bot_wallet); + callback(&event); + break; } } } } - - events + Ok(()) } fn should_handle(&self, program_id: &Pubkey) -> bool { @@ -817,3 +806,54 @@ impl EventParser for GenericEventParser { self.program_ids.clone() } } + +fn process_event( + mut event: Box, + bot_wallet: Option, +) -> Box { + update_slot(event.slot()); + if let Some(token_info) = event.as_any().downcast_ref::() { + add_dev_address(token_info.user); + if token_info.creator != Pubkey::default() && token_info.creator != token_info.user { + add_dev_address(token_info.creator); + } + } else if let Some(trade_info) = event.as_any_mut().downcast_mut::() { + if is_dev_address(&trade_info.user) || is_dev_address(&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 }; + } + } else if let Some(trade_info) = event.as_any_mut().downcast_mut::() { + 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; + } + } else if let Some(trade_info) = event.as_any_mut().downcast_mut::() { + 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; + } + } else if let Some(pool_info) = event.as_any().downcast_ref::() { + add_bonk_dev_address(pool_info.creator); + } else if let Some(trade_info) = event.as_any_mut().downcast_mut::() { + if is_bonk_dev_address(&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; + } + } + event.clear_id(); + event +} diff --git a/src/streaming/event_parser/protocols/mutil/parser.rs b/src/streaming/event_parser/protocols/mutil/parser.rs index c8b58a0..e64edfd 100755 --- a/src/streaming/event_parser/protocols/mutil/parser.rs +++ b/src/streaming/event_parser/protocols/mutil/parser.rs @@ -18,24 +18,6 @@ impl MutilEventParser { for protocol in protocols { let parse = EventParserFactory::create_parser(protocol); - // Merge inner_instruction_configs, append configurations to existing Vec - for (key, configs) in parse.inner_instruction_configs() { - let filtered_configs: Vec = configs - .into_iter() - .filter(|config| { - event_type_filter - .as_ref() - .map(|filter| filter.include.contains(&config.event_type)) - .unwrap_or(true) - }) - .collect(); - inner - .inner_instruction_configs - .entry(key) - .or_insert_with(Vec::new) - .extend(filtered_configs); - } - // Merge instruction_configs, append configurations to existing Vec for (key, configs) in parse.instruction_configs() { let filtered_configs: Vec = configs diff --git a/src/streaming/grpc/mod.rs b/src/streaming/grpc/mod.rs index 9c012e6..0d91713 100644 --- a/src/streaming/grpc/mod.rs +++ b/src/streaming/grpc/mod.rs @@ -1,17 +1,15 @@ // gRPC 相关模块 pub mod connection; -pub mod stream_handler; pub mod subscription; pub mod types; // 重新导出主要类型 pub use connection::*; -pub use stream_handler::*; pub use subscription::*; pub use types::*; // 从公用模块重新导出 pub use crate::streaming::common::{ - BackpressureConfig, BackpressureStrategy, BatchConfig, ConnectionConfig, MetricsManager, - PerformanceMetrics, StreamClientConfig as ClientConfig, + BackpressureConfig, BackpressureStrategy, ConnectionConfig, MetricsManager, PerformanceMetrics, + StreamClientConfig as ClientConfig, }; diff --git a/src/streaming/grpc/stream_handler.rs b/src/streaming/grpc/stream_handler.rs deleted file mode 100644 index 43dbb77..0000000 --- a/src/streaming/grpc/stream_handler.rs +++ /dev/null @@ -1,100 +0,0 @@ -use chrono::Local; -use futures::{channel::mpsc, sink::Sink, SinkExt}; -use solana_sdk::pubkey::Pubkey; -use yellowstone_grpc_proto::geyser::{ - subscribe_update::UpdateOneof, SubscribeRequest, SubscribeRequestPing, SubscribeUpdate, -}; - -use super::types::{BlockMetaPretty, EventPretty, TransactionPretty}; -use crate::common::AnyResult; -use crate::streaming::common::EventProcessor; -use crate::streaming::grpc::AccountPretty; - -/// 流消息处理器 -pub struct StreamHandler; - -impl StreamHandler { - /// 处理单个流消息 - pub async fn handle_stream_message( - msg: SubscribeUpdate, - subscribe_tx: &mut (impl Sink + Unpin), - event_processor: EventProcessor, - bot_wallet: Option, - ) -> AnyResult<()> { - let created_at = msg.created_at; - match msg.update_oneof { - Some(UpdateOneof::Account(account)) => { - let account_pretty = AccountPretty::from(account); - log::debug!("Received account: {:?}", account_pretty); - event_processor - .process_grpc_event_transaction_with_metrics( - EventPretty::Account(account_pretty), - bot_wallet, - ) - .await?; - } - Some(UpdateOneof::BlockMeta(sut)) => { - let block_meta_pretty = BlockMetaPretty::from((sut, created_at)); - log::debug!("Received block meta: {:?}", block_meta_pretty); - event_processor - .process_grpc_event_transaction_with_metrics( - EventPretty::BlockMeta(block_meta_pretty), - bot_wallet, - ) - .await?; - } - Some(UpdateOneof::Transaction(sut)) => { - let transaction_pretty = TransactionPretty::from((sut, created_at)); - log::debug!( - "Received transaction: {} at slot {}", - transaction_pretty.signature, - transaction_pretty.slot - ); - event_processor - .process_grpc_event_transaction_with_metrics( - EventPretty::Transaction(transaction_pretty), - bot_wallet, - ) - .await?; - } - Some(UpdateOneof::Ping(_)) => { - subscribe_tx - .send(SubscribeRequest { - ping: Some(SubscribeRequestPing { id: 1 }), - ..Default::default() - }) - .await?; - log::debug!("service is ping: {}", Local::now()); - } - Some(UpdateOneof::Pong(_)) => { - log::debug!("service is pong: {}", Local::now()); - } - _ => { - log::debug!("Received other message type"); - } - } - Ok(()) - } - - pub async fn handle_stream_system_message( - msg: SubscribeUpdate, - subscribe_tx: &mut (impl Sink + Unpin), - ) -> AnyResult> { - let created_at = msg.created_at; - let event_pretty = match msg.update_oneof { - Some(UpdateOneof::Transaction(sut)) => Some(TransactionPretty::from((sut, created_at))), - Some(UpdateOneof::Ping(_)) => { - subscribe_tx - .send(SubscribeRequest { - ping: Some(SubscribeRequestPing { id: 1 }), - ..Default::default() - }) - .await?; - None - } - Some(UpdateOneof::Pong(_)) => None, - _ => None, - }; - Ok(event_pretty.map(|e| EventPretty::Transaction(e))) - } -} diff --git a/src/streaming/shred/connection.rs b/src/streaming/shred/connection.rs index 44e7e4a..9c6bbe3 100644 --- a/src/streaming/shred/connection.rs +++ b/src/streaming/shred/connection.rs @@ -41,16 +41,33 @@ impl ShredStreamGrpc { }) } - /// 创建高性能客户端（适合高并发场景） - pub async fn new_high_performance(endpoint: String) -> AnyResult { - Self::new_with_config(endpoint, StreamClientConfig::high_performance()).await + /// Creates a new ShredStreamClient with high-throughput configuration. + /// + /// This is a convenience method that creates a client optimized for high-concurrency scenarios + /// where throughput is prioritized over latency. See `StreamClientConfig::high_throughput()` + /// for detailed configuration information. + pub async fn new_high_throughput(endpoint: String) -> AnyResult { + Self::new_with_config(endpoint, StreamClientConfig::high_throughput()).await } - /// 创建低延迟客户端（适合实时场景） + /// Creates a new ShredStreamClient with low-latency configuration. + /// + /// This is a convenience method that creates a client optimized for real-time scenarios + /// where latency is prioritized over throughput. See `StreamClientConfig::low_latency()` + /// for detailed configuration information. pub async fn new_low_latency(endpoint: String) -> AnyResult { Self::new_with_config(endpoint, StreamClientConfig::low_latency()).await } + /// Creates a new ShredStreamClient with asynchronous processing configuration. + /// + /// This is a convenience method that creates a client optimized for high-volume scenarios + /// with balanced throughput and reliability. See `StreamClientConfig::async_processing()` + /// for detailed configuration information. + pub async fn new_async_processing(endpoint: String) -> AnyResult { + Self::new_with_config(endpoint, StreamClientConfig::async_processing()).await + } + /// 获取当前配置 pub fn get_config(&self) -> &StreamClientConfig { &self.config diff --git a/src/streaming/shred/mod.rs b/src/streaming/shred/mod.rs index c2aa2d9..1358fe9 100644 --- a/src/streaming/shred/mod.rs +++ b/src/streaming/shred/mod.rs @@ -8,6 +8,6 @@ pub use types::*; // 从公用模块重新导出 pub use crate::streaming::common::{ - BackpressureConfig, BackpressureStrategy, BatchConfig, ConnectionConfig, MetricsEventType, - MetricsManager, PerformanceMetrics, StreamClientConfig, + BackpressureConfig, BackpressureStrategy, ConnectionConfig, MetricsEventType, MetricsManager, + PerformanceMetrics, StreamClientConfig, }; diff --git a/src/streaming/shred/types.rs b/src/streaming/shred/types.rs index 888622d..d10094e 100644 --- a/src/streaming/shred/types.rs +++ b/src/streaming/shred/types.rs @@ -5,16 +5,16 @@ use solana_sdk::transaction::VersionedTransaction; pub struct TransactionWithSlot { pub transaction: VersionedTransaction, pub slot: u64, + pub program_received_time_us: i64, } impl TransactionWithSlot { /// 创建新的带槽位的交易 - pub fn new(transaction: VersionedTransaction, slot: u64) -> Self { - Self { transaction, slot } - } - - /// 获取交易签名 - pub fn signature(&self) -> String { - self.transaction.signatures[0].to_string() + pub fn new( + transaction: VersionedTransaction, + slot: u64, + program_received_time_us: i64, + ) -> Self { + Self { transaction, slot, program_received_time_us } } } diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index 79cda1b..1ec59af 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -42,7 +42,6 @@ impl ShredStreamGrpc { protocols, event_type_filter, self.config.backpressure.clone(), - self.config.batch.clone(), Some(Arc::new(callback)), ); @@ -58,18 +57,24 @@ 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(transaction.clone(), msg.slot); - if let Err(e) = event_processor_clone - .process_shred_transaction_immediate( - transaction_with_slot, - bot_wallet, - ) - .await - { - error!("Error handling message: {e:?}"); - break; - } + let transaction_with_slot = TransactionWithSlot::new( + transaction.clone(), + msg.slot, + chrono::Utc::now().timestamp_micros(), + ); + // 异步执行，不阻塞主流，使用带背压控制的方法 + let processor_clone = event_processor_clone.clone(); + tokio::spawn(async move { + if let Err(e) = processor_clone + .process_shred_transaction_with_metrics( + transaction_with_slot, + bot_wallet, + ) + .await + { + error!("Error handling message: {e:?}"); + } + }); } } } diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index db928d9..6c12a41 100644 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -1,9 +1,11 @@ -use futures::StreamExt; +use chrono::Local; +use futures::{SinkExt, StreamExt}; use log::error; use solana_sdk::pubkey::Pubkey; use std::sync::{Arc, RwLock}; use tokio::sync::Mutex; -use yellowstone_grpc_proto::geyser::CommitmentLevel; +use yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof; +use yellowstone_grpc_proto::geyser::{CommitmentLevel, SubscribeRequest, SubscribeRequestPing}; use crate::common::AnyResult; use crate::streaming::common::{ @@ -11,7 +13,9 @@ use crate::streaming::common::{ }; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::{Protocol, UnifiedEvent}; -use crate::streaming::grpc::{StreamHandler, SubscriptionManager}; +use crate::streaming::grpc::{ + AccountPretty, BlockMetaPretty, EventPretty, SubscriptionManager, TransactionPretty, +}; /// 交易过滤器 pub struct TransactionFilter { @@ -73,20 +77,31 @@ impl YellowstoneGrpc { }) } - /// 创建高性能客户端 - pub fn new_high_performance(endpoint: String, x_token: Option) -> AnyResult { - Self::new_with_config(endpoint, x_token, StreamClientConfig::high_performance()) + /// Creates a new YellowstoneGrpcClient with high-throughput configuration. + /// + /// This is a convenience method that creates a client optimized for high-concurrency scenarios + /// where throughput is prioritized over latency. See `StreamClientConfig::high_throughput()` + /// for detailed configuration information. + pub fn new_high_throughput(endpoint: String, x_token: Option) -> AnyResult { + Self::new_with_config(endpoint, x_token, StreamClientConfig::high_throughput()) } - /// 创建低延迟客户端 + /// Creates a new YellowstoneGrpcClient with low-latency configuration. + /// + /// This is a convenience method that creates a client optimized for real-time scenarios + /// where latency is prioritized over throughput. See `StreamClientConfig::low_latency()` + /// for detailed configuration information. pub fn new_low_latency(endpoint: String, x_token: Option) -> AnyResult { Self::new_with_config(endpoint, x_token, StreamClientConfig::low_latency()) } - /// 创建即时处理客户端 - pub fn new_immediate(endpoint: String, x_token: Option) -> AnyResult { - let mut config = StreamClientConfig::low_latency(); - config.enable_metrics = false; + /// Creates a new YellowstoneGrpcClient with asynchronous processing configuration. + /// + /// This is a convenience method that creates a client optimized for high-volume scenarios + /// with balanced throughput and reliability. See `StreamClientConfig::async_processing()` + /// for detailed configuration information. + pub fn new_async_processing(endpoint: String, x_token: Option) -> AnyResult { + let config = StreamClientConfig::async_processing(); Self::new_with_config(endpoint, x_token, config) } @@ -171,35 +186,98 @@ impl YellowstoneGrpc { ); // 订阅事件 - let (mut subscribe_tx, mut stream) = self + let (subscribe_tx, mut stream) = self .subscription_manager .subscribe_with_request(transactions, accounts, commitment, event_type_filter.clone()) .await?; + // 用 Arc> 包装 subscribe_tx 以支持多线程共享 + let subscribe_tx = Arc::new(Mutex::new(subscribe_tx)); + // 启动流处理任务 let mut event_processor = self.event_processor.clone(); event_processor.set_protocols_and_event_type_filter( protocols, event_type_filter, self.config.backpressure.clone(), - self.config.batch.clone(), Some(Arc::new(callback)), ); + let event_processor = Arc::new(event_processor); let stream_handle = tokio::spawn(async move { while let Some(message) = stream.next().await { match message { Ok(msg) => { - if let Err(e) = StreamHandler::handle_stream_message( - msg, - &mut subscribe_tx, - event_processor.clone(), - bot_wallet, - ) - .await - { - error!("Error handling message: {e:?}"); - break; - } + // 不阻塞地处理消息，使用 tokio::spawn 实现并发 + let event_processor_ref = Arc::clone(&event_processor); + let subscribe_tx_ref = Arc::clone(&subscribe_tx); + tokio::spawn(async move { + let created_at = msg.created_at; + match msg.update_oneof { + Some(UpdateOneof::Account(account)) => { + let account_pretty = AccountPretty::from(account); + log::debug!("Received account: {:?}", account_pretty); + if let Err(e) = event_processor_ref + .process_grpc_event_transaction_with_metrics( + EventPretty::Account(account_pretty), + bot_wallet, + ) + .await + { + error!("Error processing account event: {e:?}"); + } + } + Some(UpdateOneof::BlockMeta(sut)) => { + let block_meta_pretty = + BlockMetaPretty::from((sut, created_at)); + log::debug!("Received block meta: {:?}", block_meta_pretty); + if let Err(e) = event_processor_ref + .process_grpc_event_transaction_with_metrics( + EventPretty::BlockMeta(block_meta_pretty), + bot_wallet, + ) + .await + { + error!("Error processing block meta event: {e:?}"); + } + } + Some(UpdateOneof::Transaction(sut)) => { + let transaction_pretty = + TransactionPretty::from((sut, created_at)); + log::debug!( + "Received transaction: {} at slot {}", + transaction_pretty.signature, + transaction_pretty.slot + ); + if let Err(e) = event_processor_ref + .process_grpc_event_transaction_with_metrics( + EventPretty::Transaction(transaction_pretty), + bot_wallet, + ) + .await + { + error!("Error processing transaction event: {e:?}"); + } + } + Some(UpdateOneof::Ping(_)) => { + // 只在需要时获取锁，并立即释放 + if let Ok(mut tx_guard) = subscribe_tx_ref.try_lock() { + let _ = tx_guard + .send(SubscribeRequest { + ping: Some(SubscribeRequestPing { id: 1 }), + ..Default::default() + }) + .await; + } + log::debug!("service is ping: {}", Local::now()); + } + Some(UpdateOneof::Pong(_)) => { + log::debug!("service is pong: {}", Local::now()); + } + _ => { + log::debug!("Received other message type"); + } + } + }); } Err(error) => { error!("Stream error: {error:?}"); diff --git a/src/streaming/yellowstone_sub_system.rs b/src/streaming/yellowstone_sub_system.rs index bccd417..91c991f 100755 --- a/src/streaming/yellowstone_sub_system.rs +++ b/src/streaming/yellowstone_sub_system.rs @@ -1,15 +1,18 @@ use crate::{ common::AnyResult, streaming::{ - grpc::{EventPretty, StreamHandler}, + grpc::{EventPretty, TransactionPretty}, yellowstone_grpc::YellowstoneGrpc, }, }; -use futures::StreamExt; +use futures::{SinkExt, StreamExt}; use log::error; use solana_program::pubkey; use solana_sdk::{pubkey::Pubkey, transaction::VersionedTransaction}; use solana_transaction_status::TransactionWithStatusMeta; +use yellowstone_grpc_proto::geyser::{ + subscribe_update::UpdateOneof, SubscribeRequest, SubscribeRequestPing, +}; const SYSTEM_PROGRAM_ID: Pubkey = pubkey!("11111111111111111111111111111111"); @@ -34,7 +37,7 @@ impl YellowstoneGrpc { account_exclude: Option>, ) -> AnyResult<()> where - F: Fn(SystemEvent) + Send + Sync + 'static, + F: Fn(SystemEvent) + Send + Sync + Clone + 'static, { let addrs = vec![SYSTEM_PROGRAM_ID.to_string()]; let account_include = account_include.unwrap_or_default(); @@ -56,16 +59,36 @@ impl YellowstoneGrpc { while let Some(message) = stream.next().await { match message { Ok(msg) => { - if let Ok(event_pretty) = - StreamHandler::handle_stream_system_message(msg, &mut subscribe_tx) - .await - { - if let Some(event_pretty) = event_pretty { - if let Err(e) = - Self::process_system_transaction(event_pretty, &*callback).await - { - error!("Error processing transaction: {e:?}"); - } + let created_at = msg.created_at; + match msg.update_oneof { + Some(UpdateOneof::Transaction(sut)) => { + let transaction_pretty = TransactionPretty::from((sut, created_at)); + let event_pretty = EventPretty::Transaction(transaction_pretty); + let callback_clone = callback.clone(); + tokio::spawn(async move { + if let Err(e) = Self::process_system_transaction( + event_pretty, + &*callback_clone, + ) + .await + { + error!("Error processing transaction: {e:?}"); + } + }); + } + Some(UpdateOneof::Ping(_)) => { + let _ = subscribe_tx + .send(SubscribeRequest { + ping: Some(SubscribeRequestPing { id: 1 }), + ..Default::default() + }) + .await; + } + Some(UpdateOneof::Pong(_)) => { + // Pong response, no action needed + } + _ => { + // Other message types, ignore for system subscription } } }