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

@@ -1,6 +1,6 @@
 [package]
 name = "solana-streamer-sdk"
-version = "0.1.5"
+version = "0.1.6"
 edition = "2021"
 authors = ["William <byteblock6@gmail.com>", "sgxiang <sgxiang@gmail.com>", "wei <1415121722@qq.com>"]
 repository = "https://github.com/0xfnzero/solana-streamer"

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

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"
 ```
 
 ## 使用示例

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
             }

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,

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<dyn UnifiedEvent>;
 
-    /// 合并事件（可选实现）
+    /// Merge events (optional implementation)
     fn merge(&mut self, _other: Box<dyn UnifiedEvent>) {
-        // 默认实现：不进行任何合并操作
+        // Default implementation: no merging operation
     }
 
+    /// Set transfer datas
     fn set_transfer_datas(&mut self, transfer_datas: Vec<TransferData>);
 
+    /// Get index
     fn index(&self) -> String;
 }
 
@@ -66,6 +74,7 @@ pub trait EventParser: Send + Sync {
         signature: &str,
         slot: u64,
         block_time: Option<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Vec<Box<dyn UnifiedEvent>>;
 
@@ -77,6 +86,7 @@ pub trait EventParser: Send + Sync {
         signature: &str,
         slot: u64,
         block_time: Option<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Vec<Box<dyn UnifiedEvent>>;
 
@@ -87,6 +97,7 @@ pub trait EventParser: Send + Sync {
         signature: &str,
         slot: Option<u64>,
         block_time: Option<Timestamp>,
+        program_received_time_ms: i64,
         accounts: &[Pubkey],
         inner_instructions: &[UiInnerInstructions],
     ) -> Result<Vec<Box<dyn UnifiedEvent>>> {
@@ -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<u64>,
         block_time: Option<Timestamp>,
+        program_received_time_ms: i64,
         bot_wallet: Option<Pubkey>,
     ) -> Result<Vec<Box<dyn UnifiedEvent>>> {
         let accounts: Vec<Pubkey> = 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<u64>,
         block_time: Option<Timestamp>,
+        program_received_time_ms: i64,
         bot_wallet: Option<Pubkey>,
     ) -> Result<Vec<Box<dyn UnifiedEvent>>> {
         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<Box<dyn UnifiedEvent>>,
         bot_wallet: Option<Pubkey>,
     ) -> Vec<Box<dyn UnifiedEvent>> {
-        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::<PumpFunCreateTokenEvent>() {
-                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::<PumpFunTradeEvent>()
             {
-                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<u64>,
         block_time: Option<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Result<Vec<Box<dyn UnifiedEvent>>> {
         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<u64>,
         block_time: Option<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Result<Vec<Box<dyn UnifiedEvent>>> {
         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<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Option<Box<dyn UnifiedEvent>> {
         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<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Option<Box<dyn UnifiedEvent>> {
         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<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         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<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         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);

src/streaming/event_parser/protocols/bonk/parser.rs

@@ -421,6 +421,7 @@ impl EventParser for BonkEventParser {
         signature: &str,
         slot: u64,
         block_time: Option<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         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<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         self.inner.parse_events_from_instruction(
@@ -447,6 +450,7 @@ impl EventParser for BonkEventParser {
             signature,
             slot,
             block_time,
+            program_received_time_ms,
             index,
         )
     }

src/streaming/event_parser/protocols/pumpfun/parser.rs

@@ -226,6 +226,7 @@ impl EventParser for PumpFunEventParser {
         signature: &str,
         slot: u64,
         block_time: Option<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         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<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         self.inner.parse_events_from_instruction(
@@ -252,6 +255,7 @@ impl EventParser for PumpFunEventParser {
             signature,
             slot,
             block_time,
+            program_received_time_ms,
             index,
         )
     }

src/streaming/event_parser/protocols/pumpswap/parser.rs

@@ -374,6 +374,7 @@ impl EventParser for PumpSwapEventParser {
         signature: &str,
         slot: u64,
         block_time: Option<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         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<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         self.inner.parse_events_from_instruction(
@@ -400,6 +403,7 @@ impl EventParser for PumpSwapEventParser {
             signature,
             slot,
             block_time,
+            program_received_time_ms,
             index,
         )
     }

src/streaming/event_parser/protocols/raydium_clmm/parser.rs

@@ -146,6 +146,7 @@ impl EventParser for RaydiumClmmEventParser {
         signature: &str,
         slot: u64,
         block_time: Option<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         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<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         self.inner.parse_events_from_instruction(
@@ -172,6 +175,7 @@ impl EventParser for RaydiumClmmEventParser {
             signature,
             slot,
             block_time,
+            program_received_time_ms,
             index,
         )
     }

src/streaming/event_parser/protocols/raydium_cpmm/parser.rs

@@ -135,6 +135,7 @@ impl EventParser for RaydiumCpmmEventParser {
         signature: &str,
         slot: u64,
         block_time: Option<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         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<Timestamp>,
+        program_received_time_ms: i64,
         index: String,
     ) -> Vec<Box<dyn UnifiedEvent>> {
         self.inner.parse_events_from_instruction(
@@ -161,6 +164,7 @@ impl EventParser for RaydiumCpmmEventParser {
             signature,
             slot,
             block_time,
+            program_received_time_ms,
             index,
         )
     }

src/streaming/shred_stream.rs

@@ -95,6 +95,7 @@ impl ShredStreamGrpc {
     where
         F: Fn(Box<dyn UnifiedEvent>) + 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

src/streaming/yellowstone_grpc.rs

@@ -355,7 +355,7 @@ impl YellowstoneGrpc {
     where
         F: Fn(Box<dyn UnifiedEvent>) + 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(())
     }
 }