From 0a4be48b99a70050ae683e750ab77abc785b1cc5 Mon Sep 17 00:00:00 2001 From: ysq Date: Tue, 29 Jul 2025 14:48:43 +0800 Subject: [PATCH] feat: add event processing time tracking (v0.1.6) - Add processing duration monitoring - Update all protocol parser interfaces - Improve PumpFun dev address detection - Bump version to 0.1.6 --- Cargo.toml | 2 +- README.md | 4 +- README_CN.md | 4 +- src/streaming/event_parser/common/mod.rs | 8 +++ src/streaming/event_parser/common/types.rs | 5 +- src/streaming/event_parser/core/traits.rs | 66 +++++++++++++++---- .../event_parser/protocols/bonk/parser.rs | 4 ++ .../event_parser/protocols/pumpfun/parser.rs | 4 ++ .../event_parser/protocols/pumpswap/parser.rs | 4 ++ .../protocols/raydium_clmm/parser.rs | 4 ++ .../protocols/raydium_cpmm/parser.rs | 4 ++ src/streaming/shred_stream.rs | 2 + src/streaming/yellowstone_grpc.rs | 6 +- 13 files changed, 93 insertions(+), 24 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 59da758..07cb094 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "solana-streamer-sdk" -version = "0.1.5" +version = "0.1.6" edition = "2021" authors = ["William ", "sgxiang ", "wei <1415121722@qq.com>"] repository = "https://github.com/0xfnzero/solana-streamer" diff --git a/README.md b/README.md index 88db3de..6d9be79 100755 --- a/README.md +++ b/README.md @@ -33,14 +33,14 @@ Add the dependency to your `Cargo.toml`: ```toml # Add to your Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.5" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.6" } ``` ### Use crates.io ```toml # Add to your Cargo.toml -solana-streamer-sdk = "0.1.5" +solana-streamer-sdk = "0.1.6" ``` ## Usage Examples diff --git a/README_CN.md b/README_CN.md index c6f4f84..e3c113b 100644 --- a/README_CN.md +++ b/README_CN.md @@ -33,14 +33,14 @@ git clone https://github.com/0xfnzero/solana-streamer ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.5" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.6" } ``` ### 使用 crates.io ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = "0.1.5" +solana-streamer-sdk = "0.1.6" ``` ## 使用示例 diff --git a/src/streaming/event_parser/common/mod.rs b/src/streaming/event_parser/common/mod.rs index a3242ac..3500281 100755 --- a/src/streaming/event_parser/common/mod.rs +++ b/src/streaming/event_parser/common/mod.rs @@ -27,6 +27,14 @@ macro_rules! impl_unified_event { self.metadata.program_received_time_ms } + fn program_handle_time_consuming_ms(&self) -> i64 { + self.metadata.program_handle_time_consuming_ms + } + + fn set_program_handle_time_consuming_ms(&mut self, program_handle_time_consuming_ms: i64) { + self.metadata.program_handle_time_consuming_ms = program_handle_time_consuming_ms; + } + fn as_any(&self) -> &dyn std::any::Any { self } diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index 48eb811..2f51a9f 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -158,6 +158,7 @@ pub struct EventMetadata { pub block_time: i64, pub block_time_ms: i64, pub program_received_time_ms: i64, + pub program_handle_time_consuming_ms: i64, pub protocol: ProtocolType, pub event_type: EventType, pub program_id: Pubkey, @@ -176,6 +177,7 @@ impl EventMetadata { event_type: EventType, program_id: Pubkey, index: String, + program_received_time_ms: i64, ) -> Self { Self { id, @@ -183,7 +185,8 @@ impl EventMetadata { slot, block_time, block_time_ms, - program_received_time_ms: chrono::Utc::now().timestamp_millis(), + program_received_time_ms: program_received_time_ms, + program_handle_time_consuming_ms: 0, protocol, event_type, program_id, diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index ed95d10..26a234d 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -20,39 +20,47 @@ use crate::streaming::event_parser::{ }, }; -/// 统一事件接口 - 所有协议的事件都需要实现此trait +/// Unified Event Interface - All protocol events must implement this trait pub trait UnifiedEvent: Debug + Send + Sync { - /// 获取事件ID + /// Get event ID fn id(&self) -> &str; - /// 获取事件类型 + /// Get event type fn event_type(&self) -> EventType; - /// 获取交易签名 + /// Get transaction signature fn signature(&self) -> &str; - /// 获取槽位号 + /// Get slot number fn slot(&self) -> u64; - /// 获取程序接收的时间戳(毫秒) + /// Get program received timestamp (milliseconds) fn program_received_time_ms(&self) -> i64; - /// 将事件转换为Any以便向下转型 + /// Processing time consumption (milliseconds) + fn program_handle_time_consuming_ms(&self) -> i64; + + /// Set processing time consumption (milliseconds) + fn set_program_handle_time_consuming_ms(&mut self, program_handle_time_consuming_ms: i64); + + /// Convert event to Any for downcasting fn as_any(&self) -> &dyn std::any::Any; - /// 将事件转换为可变Any以便向下转型 + /// Convert event to mutable Any for downcasting fn as_any_mut(&mut self) -> &mut dyn std::any::Any; - /// 克隆事件 + /// Clone the event fn clone_boxed(&self) -> Box; - /// 合并事件（可选实现） + /// Merge events (optional implementation) fn merge(&mut self, _other: Box) { - // 默认实现：不进行任何合并操作 + // Default implementation: no merging operation } + /// Set transfer datas fn set_transfer_datas(&mut self, transfer_datas: Vec); + /// Get index fn index(&self) -> String; } @@ -66,6 +74,7 @@ pub trait EventParser: Send + Sync { signature: &str, slot: u64, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Vec>; @@ -77,6 +86,7 @@ pub trait EventParser: Send + Sync { signature: &str, slot: u64, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Vec>; @@ -87,6 +97,7 @@ pub trait EventParser: Send + Sync { signature: &str, slot: Option, block_time: Option, + program_received_time_ms: i64, accounts: &[Pubkey], inner_instructions: &[UiInnerInstructions], ) -> Result>> { @@ -116,6 +127,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, + program_received_time_ms, format!("{}", index), ) .await @@ -153,6 +165,7 @@ pub trait EventParser: Send + Sync { signature: &str, slot: Option, block_time: Option, + program_received_time_ms: i64, bot_wallet: Option, ) -> Result>> { let accounts: Vec = versioned_tx.message.static_account_keys().to_vec(); @@ -162,6 +175,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, + program_received_time_ms, &accounts, &vec![], ) @@ -176,6 +190,7 @@ pub trait EventParser: Send + Sync { signature: &str, slot: Option, block_time: Option, + program_received_time_ms: i64, bot_wallet: Option, ) -> Result>> { let transaction = tx.transaction; @@ -212,6 +227,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, + program_received_time_ms, &accounts, &inner_instructions, ) @@ -242,6 +258,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, + program_received_time_ms, format!("{}.{}", inner_instruction.index, index), ) .await @@ -266,6 +283,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, + program_received_time_ms, format!("{}.{}", inner_instruction.index, index), ) .await @@ -328,14 +346,20 @@ pub trait EventParser: Send + Sync { mut events: Vec>, bot_wallet: Option, ) -> Vec> { - let mut dev_address = None; + 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 = Some(token_info.user); + 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 Some(trade_info.user) == dev_address { + 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; @@ -354,6 +378,8 @@ pub trait EventParser: Send + Sync { trade_info.is_dev_create_token_trade = false; } } + let now = chrono::Utc::now().timestamp_millis(); + event.set_program_handle_time_consuming_ms(now - event.program_received_time_ms()); } events } @@ -364,6 +390,7 @@ pub trait EventParser: Send + Sync { signature: &str, slot: Option, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Result>> { let slot = slot.unwrap_or(0); @@ -372,6 +399,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, + program_received_time_ms, index, ); Ok(events) @@ -384,6 +412,7 @@ pub trait EventParser: Send + Sync { signature: &str, slot: Option, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Result>> { let slot = slot.unwrap_or(0); @@ -393,6 +422,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, + program_received_time_ms, index, ); Ok(events) @@ -475,6 +505,7 @@ impl GenericEventParser { signature: &str, slot: u64, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Option> { let timestamp = block_time.unwrap_or(Timestamp { @@ -492,6 +523,7 @@ impl GenericEventParser { config.event_type.clone(), self.program_id, index, + program_received_time_ms, ); (config.inner_instruction_parser)(data, metadata) } @@ -505,6 +537,7 @@ impl GenericEventParser { signature: &str, slot: u64, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Option> { let timestamp = block_time.unwrap_or(Timestamp { @@ -522,6 +555,7 @@ impl GenericEventParser { config.event_type.clone(), self.program_id, index, + program_received_time_ms, ); (config.instruction_parser)(data, account_pubkeys, metadata) } @@ -536,6 +570,7 @@ impl EventParser for GenericEventParser { signature: &str, slot: u64, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Vec> { let inner_instruction_data = inner_instruction.data.clone(); @@ -557,6 +592,7 @@ impl EventParser for GenericEventParser { signature, slot, block_time, + program_received_time_ms, index.clone(), ) { events.push(event); @@ -575,6 +611,7 @@ impl EventParser for GenericEventParser { signature: &str, slot: u64, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Vec> { let program_id = accounts[instruction.program_id_index as usize]; @@ -607,6 +644,7 @@ impl EventParser for GenericEventParser { signature, slot, block_time, + program_received_time_ms, index.clone(), ) { events.push(event); diff --git a/src/streaming/event_parser/protocols/bonk/parser.rs b/src/streaming/event_parser/protocols/bonk/parser.rs index cf26f4f..aebc097 100755 --- a/src/streaming/event_parser/protocols/bonk/parser.rs +++ b/src/streaming/event_parser/protocols/bonk/parser.rs @@ -421,6 +421,7 @@ impl EventParser for BonkEventParser { signature: &str, slot: u64, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Vec> { self.inner.parse_events_from_inner_instruction( @@ -428,6 +429,7 @@ impl EventParser for BonkEventParser { signature, slot, block_time, + program_received_time_ms, index, ) } @@ -439,6 +441,7 @@ impl EventParser for BonkEventParser { signature: &str, slot: u64, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Vec> { self.inner.parse_events_from_instruction( @@ -447,6 +450,7 @@ impl EventParser for BonkEventParser { signature, slot, block_time, + program_received_time_ms, index, ) } diff --git a/src/streaming/event_parser/protocols/pumpfun/parser.rs b/src/streaming/event_parser/protocols/pumpfun/parser.rs index 402937a..80c0be6 100755 --- a/src/streaming/event_parser/protocols/pumpfun/parser.rs +++ b/src/streaming/event_parser/protocols/pumpfun/parser.rs @@ -226,6 +226,7 @@ impl EventParser for PumpFunEventParser { signature: &str, slot: u64, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Vec> { self.inner.parse_events_from_inner_instruction( @@ -233,6 +234,7 @@ impl EventParser for PumpFunEventParser { signature, slot, block_time, + program_received_time_ms, index, ) } @@ -244,6 +246,7 @@ impl EventParser for PumpFunEventParser { signature: &str, slot: u64, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Vec> { self.inner.parse_events_from_instruction( @@ -252,6 +255,7 @@ impl EventParser for PumpFunEventParser { signature, slot, block_time, + program_received_time_ms, index, ) } diff --git a/src/streaming/event_parser/protocols/pumpswap/parser.rs b/src/streaming/event_parser/protocols/pumpswap/parser.rs index ce98fde..6fa4db2 100755 --- a/src/streaming/event_parser/protocols/pumpswap/parser.rs +++ b/src/streaming/event_parser/protocols/pumpswap/parser.rs @@ -374,6 +374,7 @@ impl EventParser for PumpSwapEventParser { signature: &str, slot: u64, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Vec> { self.inner.parse_events_from_inner_instruction( @@ -381,6 +382,7 @@ impl EventParser for PumpSwapEventParser { signature, slot, block_time, + program_received_time_ms, index, ) } @@ -392,6 +394,7 @@ impl EventParser for PumpSwapEventParser { signature: &str, slot: u64, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Vec> { self.inner.parse_events_from_instruction( @@ -400,6 +403,7 @@ impl EventParser for PumpSwapEventParser { signature, slot, block_time, + program_received_time_ms, index, ) } diff --git a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs index b504470..0453c55 100755 --- a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs @@ -146,6 +146,7 @@ impl EventParser for RaydiumClmmEventParser { signature: &str, slot: u64, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Vec> { self.inner.parse_events_from_inner_instruction( @@ -153,6 +154,7 @@ impl EventParser for RaydiumClmmEventParser { signature, slot, block_time, + program_received_time_ms, index, ) } @@ -164,6 +166,7 @@ impl EventParser for RaydiumClmmEventParser { signature: &str, slot: u64, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Vec> { self.inner.parse_events_from_instruction( @@ -172,6 +175,7 @@ impl EventParser for RaydiumClmmEventParser { signature, slot, block_time, + program_received_time_ms, index, ) } diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs index 3de67cf..dc0e590 100755 --- a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs @@ -135,6 +135,7 @@ impl EventParser for RaydiumCpmmEventParser { signature: &str, slot: u64, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Vec> { self.inner.parse_events_from_inner_instruction( @@ -142,6 +143,7 @@ impl EventParser for RaydiumCpmmEventParser { signature, slot, block_time, + program_received_time_ms, index, ) } @@ -153,6 +155,7 @@ impl EventParser for RaydiumCpmmEventParser { signature: &str, slot: u64, block_time: Option, + program_received_time_ms: i64, index: String, ) -> Vec> { self.inner.parse_events_from_instruction( @@ -161,6 +164,7 @@ impl EventParser for RaydiumCpmmEventParser { signature, slot, block_time, + program_received_time_ms, index, ) } diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index c36ed72..92fa5e3 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -95,6 +95,7 @@ impl ShredStreamGrpc { where F: Fn(Box) + Send + Sync, { + let program_received_time_ms = chrono::Utc::now().timestamp_millis(); let slot = transaction_with_slot.slot; let versioned_tx = transaction_with_slot.transaction; let signature = versioned_tx.signatures[0]; @@ -107,6 +108,7 @@ impl ShredStreamGrpc { &signature.to_string(), Some(slot), None, + program_received_time_ms, bot_wallet.clone(), ) .await diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index 3787017..d33bfd7 100755 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -355,7 +355,7 @@ impl YellowstoneGrpc { where F: Fn(Box) + Send + Sync, { - // let start_time = std::time::Instant::now(); + let program_received_time_ms = chrono::Utc::now().timestamp_millis(); let slot = transaction_pretty.slot; let signature = transaction_pretty.signature.to_string(); let mut futures = Vec::new(); @@ -372,6 +372,7 @@ impl YellowstoneGrpc { &signature_clone, Some(slot), transaction_pretty.block_time, + program_received_time_ms, bot_wallet_clone, ) .await @@ -387,9 +388,6 @@ impl YellowstoneGrpc { } } } - // let elapsed = start_time.elapsed(); - // println!("处理交易耗时: {:?}", elapsed); - Ok(()) } }