From 6493f9d9229ead587f12c0b6247a842761c3e26a Mon Sep 17 00:00:00 2001 From: Estereg <109712545+Estereg@users.noreply.github.com> Date: Sat, 7 Mar 2026 07:10:34 +0000 Subject: [PATCH 1/2] refactor: streaming core optimization and code deduplication - Remove 'wide' and 'simd_utils.rs', rely on compiler auto-vectorization. - Deduplicate swap parsing in 'types.rs' and 'merger_event.rs' using generic traits and local macros. - Improve signature cleanup in 'global_state.rs' to avoid large heap allocations. - Replace manual metadata accessors in 'traits.rs' with a macro for all variants. - Remove unused types (ParseResult, ProtocolInfo) and legacy commented code. - Fix Clippy lints (is_empty, copied) and restore missing trait imports. --- Cargo.toml | 1 - src/streaming/common/mod.rs | 4 +- src/streaming/common/simd_utils.rs | 295 -------- src/streaming/event_parser/common/types.rs | 657 ++++++------------ src/streaming/event_parser/common/utils.rs | 4 +- .../event_parser/core/global_state.rs | 11 +- src/streaming/event_parser/core/traits.rs | 196 +++--- src/streaming/grpc/subscription.rs | 2 +- src/streaming/grpc/types.rs | 60 -- 9 files changed, 301 insertions(+), 929 deletions(-) delete mode 100644 src/streaming/common/simd_utils.rs diff --git a/Cargo.toml b/Cargo.toml index 560ff26..5c3e5db 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -34,7 +34,6 @@ dashmap = "6.1.0" prost = "0.14.3" prost-types = "0.14.3" crossbeam-queue = "0.3.12" -wide = "1.1.1" spl-token = { version = "9.0.0", default-features = false, features = ["no-entrypoint"] } spl-token-2022 = { version = "10.0.0", default-features = false, features = ["no-entrypoint"] } solana-commitment-config = { version = "3.1.1", features = ["serde"] } diff --git a/src/streaming/common/mod.rs b/src/streaming/common/mod.rs index 55a76aa..bdf43c3 100644 --- a/src/streaming/common/mod.rs +++ b/src/streaming/common/mod.rs @@ -4,12 +4,10 @@ pub mod metrics; pub mod constants; pub mod subscription; pub mod event_processor; -pub mod simd_utils; // 重新导出主要类型 pub use config::*; pub use metrics::*; pub use constants::*; pub use subscription::*; -pub use event_processor::*; -pub use simd_utils::*; \ No newline at end of file +pub use event_processor::*; \ No newline at end of file diff --git a/src/streaming/common/simd_utils.rs b/src/streaming/common/simd_utils.rs deleted file mode 100644 index bb17fcf..0000000 --- a/src/streaming/common/simd_utils.rs +++ /dev/null @@ -1,295 +0,0 @@ -use wide::*; - -/// SIMD-accelerated data parsing utilities -pub struct SimdUtils; - -impl SimdUtils { - /// SIMD-accelerated byte array comparison - /// For arrays with length >= 16, uses SIMD instructions for fast comparison - #[inline(always)] - pub fn fast_bytes_equal(a: &[u8], b: &[u8]) -> bool { - if a.len() != b.len() { - return false; - } - - let len = a.len(); - - // For small arrays, use standard comparison directly - if len < 16 { - return a == b; - } - - // Use SIMD to process 16-byte chunks - let chunks = len / 16; - let remainder = len % 16; - - // Process complete 16-byte chunks - for i in 0..chunks { - let offset = i * 16; - let chunk_a = u8x16::from(&a[offset..offset + 16]); - let chunk_b = u8x16::from(&b[offset..offset + 16]); - - if !chunk_a.simd_eq(chunk_b).all() { - return false; - } - } - - // Process remaining bytes - if remainder > 0 { - let start = chunks * 16; - return &a[start..] == &b[start..]; - } - - true - } - - /// Fast discriminator matching, specifically for instruction discriminator comparison - #[inline(always)] - pub fn fast_discriminator_match(data: &[u8], discriminator: &[u8]) -> bool { - if data.len() < discriminator.len() { - return false; - } - - let disc_len = discriminator.len(); - - // Optimize for common discriminator lengths - match disc_len { - 1 => data[0] == discriminator[0], - 2 => { - let data_u16 = u16::from_le_bytes([data[0], data[1]]); - let disc_u16 = u16::from_le_bytes([discriminator[0], discriminator[1]]); - data_u16 == disc_u16 - } - 4 => { - let data_u32 = u32::from_le_bytes([data[0], data[1], data[2], data[3]]); - let disc_u32 = u32::from_le_bytes([ - discriminator[0], - discriminator[1], - discriminator[2], - discriminator[3], - ]); - data_u32 == disc_u32 - } - 8 => { - let data_u64 = u64::from_le_bytes([ - data[0], data[1], data[2], data[3], data[4], data[5], data[6], data[7], - ]); - let disc_u64 = u64::from_le_bytes([ - discriminator[0], - discriminator[1], - discriminator[2], - discriminator[3], - discriminator[4], - discriminator[5], - discriminator[6], - discriminator[7], - ]); - data_u64 == disc_u64 - } - 16 => { - // Use SIMD to process 16-byte discriminators - let data_chunk = u8x16::from(&data[..16]); - let disc_chunk = u8x16::from(discriminator); - data_chunk.simd_eq(disc_chunk).all() - } - _ => { - // For other lengths, use generic SIMD comparison - Self::fast_bytes_equal(&data[..disc_len], discriminator) - } - } - } - - /// SIMD-accelerated memory search to find specific patterns in data - #[inline(always)] - pub fn find_pattern_simd(haystack: &[u8], needle: &[u8]) -> Option { - if needle.is_empty() || haystack.len() < needle.len() { - return None; - } - - let needle_len = needle.len(); - let haystack_len = haystack.len(); - - // For single-byte search, use optimized method - if needle_len == 1 { - let target = needle[0]; - return haystack.iter().position(|&b| b == target); - } - - // For multi-byte search, use SIMD acceleration - if needle_len <= 16 && haystack_len >= 16 { - let first_byte = needle[0]; - let chunks = (haystack_len - needle_len + 1) / 16; - - for chunk_idx in 0..chunks { - let start = chunk_idx * 16; - let end = std::cmp::min(start + 16, haystack_len - needle_len + 1); - - // Use SIMD to find first byte matches - let chunk = &haystack[start..start + 16]; - let target_vec = u8x16::splat(first_byte); - let chunk_vec = u8x16::from(chunk); - let matches = chunk_vec.simd_eq(target_vec); - - // Check each match position - let matches_array: [u8; 16] = matches.into(); - for i in 0..16 { - if start + i >= end { - break; - } - - if matches_array[i] != 0 && start + i + needle_len <= haystack_len { - if Self::fast_bytes_equal( - &haystack[start + i..start + i + needle_len], - needle, - ) { - return Some(start + i); - } - } - } - } - - // Process remaining part - let remaining_start = chunks * 16; - for i in remaining_start..=(haystack_len - needle_len) { - if Self::fast_bytes_equal(&haystack[i..i + needle_len], needle) { - return Some(i); - } - } - } else { - // Fallback to standard search - for i in 0..=(haystack_len - needle_len) { - if Self::fast_bytes_equal(&haystack[i..i + needle_len], needle) { - return Some(i); - } - } - } - - None - } - - /// SIMD-accelerated data validation to check if data conforms to specific format - #[inline(always)] - pub fn validate_data_format(data: &[u8], min_length: usize) -> bool { - if data.len() < min_length { - return false; - } - - true - } - - /// Fast checksum calculation (maintains API consistency) - #[inline(always)] - pub fn fast_checksum(data: &[u8]) -> u32 { - // Simplified implementation, directly sum all bytes - data.iter().map(|&b| b as u32).sum() - } - - /// SIMD-accelerated data copy (for large data blocks) - #[inline(always)] - pub fn fast_copy(src: &[u8], dst: &mut [u8]) { - if src.len() != dst.len() { - panic!("Source and destination must have the same length"); - } - - let len = src.len(); - - if len >= 32 { - // Use 32-byte SIMD copy - let chunks = len / 32; - - for i in 0..chunks { - let start = i * 32; - let src_chunk1 = u8x16::from(&src[start..start + 16]); - let src_chunk2 = u8x16::from(&src[start + 16..start + 32]); - - let chunk1_array: [u8; 16] = src_chunk1.into(); - let chunk2_array: [u8; 16] = src_chunk2.into(); - - dst[start..start + 16].copy_from_slice(&chunk1_array); - dst[start + 16..start + 32].copy_from_slice(&chunk2_array); - } - - // Process remaining bytes - let remaining_start = chunks * 32; - dst[remaining_start..].copy_from_slice(&src[remaining_start..]); - } else { - // For small data, use standard copy - dst.copy_from_slice(src); - } - } - - /// SIMD-accelerated account indices validation - /// Validates that all indices in the account index array are less than the total account count - #[inline(always)] - pub fn validate_account_indices_simd(indices: &[u8], account_count: usize) -> bool { - if indices.is_empty() { - return true; - } - - let max_valid_index = account_count as u8; - - // For small arrays, use standard comparison directly - if indices.len() < 16 { - return indices.iter().all(|&idx| idx < max_valid_index); - } - - // Use SIMD for batch loading and comparison - let chunks = indices.len() / 16; - let remainder = indices.len() % 16; - - // Process complete 16-byte chunks - for i in 0..chunks { - let start = i * 16; - let indices_chunk = u8x16::from(&indices[start..start + 16]); - - // Convert SIMD vector to array for fast batch checking - let indices_array: [u8; 16] = indices_chunk.into(); - - // Use unrolled loop for fast comparison, compiler will optimize this - if indices_array[0] >= max_valid_index - || indices_array[1] >= max_valid_index - || indices_array[2] >= max_valid_index - || indices_array[3] >= max_valid_index - || indices_array[4] >= max_valid_index - || indices_array[5] >= max_valid_index - || indices_array[6] >= max_valid_index - || indices_array[7] >= max_valid_index - || indices_array[8] >= max_valid_index - || indices_array[9] >= max_valid_index - || indices_array[10] >= max_valid_index - || indices_array[11] >= max_valid_index - || indices_array[12] >= max_valid_index - || indices_array[13] >= max_valid_index - || indices_array[14] >= max_valid_index - || indices_array[15] >= max_valid_index - { - return false; - } - } - - // Process remaining bytes - if remainder > 0 { - let remaining_start = chunks * 16; - return indices[remaining_start..].iter().all(|&idx| idx < max_valid_index); - } - - true - } - - /// SIMD-accelerated instruction data validation - /// Validates basic format and length requirements of instruction data - #[inline(always)] - pub fn validate_instruction_data_simd( - data: &[u8], - min_length: usize, - discriminator_length: usize, - ) -> bool { - // Basic length check - if data.len() < min_length || data.len() < discriminator_length { - return false; - } - - // Use existing data format validation - Self::validate_data_format(data, min_length) - } -} diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index 31ba9f7..8396a3f 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -4,7 +4,7 @@ use serde::{Deserialize, Serialize}; use solana_sdk::{pubkey::Pubkey, signature::Signature}; use std::{borrow::Cow, fmt, str::FromStr, sync::Arc}; -use crate::streaming::{common::SimdUtils, event_parser::DexEvent}; +use crate::streaming::event_parser::DexEvent; // Object pool size configuration const EVENT_METADATA_POOL_SIZE: usize = 1000; @@ -165,124 +165,11 @@ pub const BLOCK_EVENT_TYPES: &[EventType] = &[EventType::BlockMeta]; impl fmt::Display for EventType { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - EventType::PumpSwapBuy => write!(f, "PumpSwapBuy"), - EventType::PumpSwapSell => write!(f, "PumpSwapSell"), - EventType::PumpSwapCreatePool => write!(f, "PumpSwapCreatePool"), - EventType::PumpSwapDeposit => write!(f, "PumpSwapDeposit"), - EventType::PumpSwapWithdraw => write!(f, "PumpSwapWithdraw"), - EventType::PumpFunCreateToken => write!(f, "PumpFunCreateToken"), - EventType::PumpFunCreateV2Token => write!(f, "PumpFunCreateV2Token"), - EventType::PumpFunBuy => write!(f, "PumpFunBuy"), - EventType::PumpFunSell => write!(f, "PumpFunSell"), - EventType::PumpFunMigrate => write!(f, "PumpFunMigrate"), - EventType::BonkBuyExactIn => write!(f, "BonkBuyExactIn"), - EventType::BonkBuyExactOut => write!(f, "BonkBuyExactOut"), - EventType::BonkSellExactIn => write!(f, "BonkSellExactIn"), - EventType::BonkSellExactOut => write!(f, "BonkSellExactOut"), - EventType::BonkInitialize => write!(f, "BonkInitialize"), - EventType::BonkInitializeV2 => write!(f, "BonkInitializeV2"), - EventType::BonkInitializeWithToken2022 => write!(f, "BonkInitializeWithToken2022"), - EventType::BonkMigrateToAmm => write!(f, "BonkMigrateToAmm"), - EventType::BonkMigrateToCpswap => write!(f, "BonkMigrateToCpswap"), - EventType::RaydiumCpmmSwapBaseInput => write!(f, "RaydiumCpmmSwapBaseInput"), - EventType::RaydiumCpmmSwapBaseOutput => write!(f, "RaydiumCpmmSwapBaseOutput"), - EventType::RaydiumCpmmDeposit => write!(f, "RaydiumCpmmDeposit"), - EventType::RaydiumCpmmInitialize => write!(f, "RaydiumCpmmInitialize"), - EventType::RaydiumCpmmWithdraw => write!(f, "RaydiumCpmmWithdraw"), - EventType::RaydiumClmmSwap => write!(f, "RaydiumClmmSwap"), - EventType::RaydiumClmmSwapV2 => write!(f, "RaydiumClmmSwapV2"), - EventType::RaydiumClmmClosePosition => write!(f, "RaydiumClmmClosePosition"), - EventType::RaydiumClmmDecreaseLiquidityV2 => { - write!(f, "RaydiumClmmDecreaseLiquidityV2") - } - EventType::RaydiumClmmCreatePool => write!(f, "RaydiumClmmCreatePool"), - EventType::RaydiumClmmIncreaseLiquidityV2 => { - write!(f, "RaydiumClmmIncreaseLiquidityV2") - } - EventType::RaydiumClmmOpenPositionWithToken22Nft => { - write!(f, "RaydiumClmmOpenPositionWithToken22Nft") - } - EventType::RaydiumClmmOpenPositionV2 => write!(f, "RaydiumClmmOpenPositionV2"), - EventType::RaydiumAmmV4SwapBaseIn => write!(f, "RaydiumAmmV4SwapBaseIn"), - EventType::RaydiumAmmV4SwapBaseOut => write!(f, "RaydiumAmmV4SwapBaseOut"), - EventType::RaydiumAmmV4Deposit => write!(f, "RaydiumAmmV4Deposit"), - EventType::RaydiumAmmV4Initialize2 => write!(f, "RaydiumAmmV4Initialize2"), - EventType::RaydiumAmmV4Withdraw => write!(f, "RaydiumAmmV4Withdraw"), - EventType::RaydiumAmmV4WithdrawPnl => write!(f, "RaydiumAmmV4WithdrawPnl"), - EventType::MeteoraDammV2Swap => write!(f, "MeteoraDammV2Swap"), - EventType::MeteoraDammV2Swap2 => write!(f, "MeteoraDammV2Swap2"), - EventType::MeteoraDammV2InitializePool => write!(f, "MeteoraDammV2InitializePool"), - EventType::MeteoraDammV2InitializeCustomizablePool => write!(f, "MeteoraDammV2InitializeCustomizablePool"), - EventType::MeteoraDammV2InitializePoolWithDynamicConfig => write!(f, "MeteoraDammV2InitializePoolWithDynamicConfig"), - EventType::AccountRaydiumAmmV4AmmInfo => write!(f, "AccountRaydiumAmmV4AmmInfo"), - EventType::AccountPumpSwapGlobalConfig => write!(f, "AccountPumpSwapGlobalConfig"), - EventType::AccountPumpSwapPool => write!(f, "AccountPumpSwapPool"), - EventType::AccountBonkPoolState => write!(f, "AccountBonkPoolState"), - EventType::AccountBonkGlobalConfig => write!(f, "AccountBonkGlobalConfig"), - EventType::AccountBonkPlatformConfig => write!(f, "AccountBonkPlatformConfig"), - EventType::AccountBonkVestingRecord => write!(f, "AccountBonkVestingRecord"), - EventType::AccountPumpFunBondingCurve => write!(f, "AccountPumpFunBondingCurve"), - EventType::AccountPumpFunGlobal => write!(f, "AccountPumpFunGlobal"), - EventType::AccountRaydiumClmmAmmConfig => write!(f, "AccountRaydiumClmmAmmConfig"), - EventType::AccountRaydiumClmmPoolState => write!(f, "AccountRaydiumClmmPoolState"), - EventType::AccountRaydiumClmmTickArrayState => { - write!(f, "AccountRaydiumClmmTickArrayState") - } - EventType::AccountRaydiumCpmmAmmConfig => write!(f, "AccountRaydiumCpmmAmmConfig"), - EventType::AccountRaydiumCpmmPoolState => write!(f, "AccountRaydiumCpmmPoolState"), - EventType::TokenAccount => write!(f, "TokenAccount"), - EventType::NonceAccount => write!(f, "NonceAccount"), - EventType::BlockMeta => write!(f, "BlockMeta"), - EventType::SetComputeUnitLimit => write!(f, "SetComputeUnitLimit"), - EventType::SetComputeUnitPrice => write!(f, "SetComputeUnitPrice"), - EventType::Unknown => write!(f, "Unknown"), - } + write!(f, "{:?}", self) } } -/// Parse result -#[derive(Debug, Clone)] -pub struct ParseResult { - pub success: bool, - pub data: Option, - pub error: Option, -} -impl ParseResult { - pub fn success(data: T) -> Self { - Self { success: true, data: Some(data), error: None } - } - - pub fn failure(error: String) -> Self { - Self { success: false, data: None, error: Some(error) } - } - - pub fn is_success(&self) -> bool { - self.success - } - - pub fn is_failure(&self) -> bool { - !self.success - } -} - -/// Protocol information -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub struct ProtocolInfo { - pub name: String, - pub program_ids: Vec, -} - -impl ProtocolInfo { - pub fn new(name: String, program_ids: Vec) -> Self { - Self { name, program_ids } - } - - pub fn supports_program(&self, program_id: &Pubkey) -> bool { - self.program_ids.contains(program_id) - } -} #[derive( Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize, @@ -348,11 +235,6 @@ impl EventMetadata { pub fn set_swap_data(&mut self, swap_data: SwapData) { self.swap_data = Some(swap_data); } - - /// Recycle EventMetadata to object pool - pub fn recycle(self) { - EVENT_METADATA_POOL.release(self); - } } static SOL_MINT: std::sync::LazyLock = @@ -363,341 +245,230 @@ static SYSTEM_PROGRAMS: std::sync::LazyLock<[Pubkey; 3]> = std::sync::LazyLock:: Pubkey::from_str("11111111111111111111111111111111").unwrap(), ]); -/// Parse token transfer data from next instructions +/// Trait abstracting over different inner-instruction types for swap data extraction +pub trait InnerInstructionLike { + fn program_id_index(&self) -> usize; + fn accounts(&self) -> &[u8]; + fn data(&self) -> &[u8]; +} + +/// Adapter for standard Solana compiled instructions +impl InnerInstructionLike for solana_sdk::message::compiled_instruction::CompiledInstruction { + fn program_id_index(&self) -> usize { + self.program_id_index as usize + } + fn accounts(&self) -> &[u8] { + &self.accounts + } + fn data(&self) -> &[u8] { + &self.data + } +} + +/// Adapter for gRPC inner instructions (yellowstone) +impl InnerInstructionLike for yellowstone_grpc_proto::prelude::InnerInstruction { + fn program_id_index(&self) -> usize { + self.program_id_index as usize + } + fn accounts(&self) -> &[u8] { + &self.accounts + } + fn data(&self) -> &[u8] { + &self.data + } +} + +/// Extract event context (mint/token account/vault info) from a DexEvent +fn extract_swap_context(event: &DexEvent) -> ( + SwapData, + Option, Option, + Option, Option, + Option, Option, +) { + let mut swap_data = SwapData::default(); + let mut from_mint: Option = None; + let mut to_mint: Option = None; + let mut user_from_token: Option = None; + let mut user_to_token: Option = None; + let mut from_vault: Option = None; + let mut to_vault: Option = None; + + match event { + DexEvent::BonkTradeEvent(e) => { + from_mint = Some(e.base_token_mint); + to_mint = Some(e.quote_token_mint); + user_from_token = Some(e.user_base_token); + user_to_token = Some(e.user_quote_token); + from_vault = Some(e.base_vault); + to_vault = Some(e.quote_vault); + } + DexEvent::PumpFunTradeEvent(e) => { + swap_data.from_mint = if e.is_buy { *SOL_MINT } else { e.mint }; + swap_data.to_mint = if e.is_buy { e.mint } else { *SOL_MINT }; + } + DexEvent::PumpSwapBuyEvent(e) => { + swap_data.from_mint = e.quote_mint; + swap_data.to_mint = e.base_mint; + } + DexEvent::PumpSwapSellEvent(e) => { + swap_data.from_mint = e.base_mint; + swap_data.to_mint = e.quote_mint; + } + DexEvent::RaydiumCpmmSwapEvent(e) => { + from_mint = Some(e.input_token_mint); + to_mint = Some(e.output_token_mint); + user_from_token = Some(e.input_token_account); + user_to_token = Some(e.output_token_account); + from_vault = Some(e.input_vault); + to_vault = Some(e.output_vault); + } + DexEvent::RaydiumClmmSwapEvent(e) => { + swap_data.description = + Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".into()); + user_from_token = Some(e.input_token_account); + user_to_token = Some(e.output_token_account); + from_vault = Some(e.input_vault); + to_vault = Some(e.output_vault); + } + DexEvent::RaydiumClmmSwapV2Event(e) => { + from_mint = Some(e.input_vault_mint); + to_mint = Some(e.output_vault_mint); + user_from_token = Some(e.input_token_account); + user_to_token = Some(e.output_token_account); + from_vault = Some(e.input_vault); + to_vault = Some(e.output_vault); + } + DexEvent::RaydiumAmmV4SwapEvent(e) => { + swap_data.description = + Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".into()); + user_from_token = Some(e.user_source_token_account); + user_to_token = Some(e.user_destination_token_account); + from_vault = Some(e.pool_pc_token_account); + to_vault = Some(e.pool_coin_token_account); + } + _ => {} + } + + (swap_data, from_mint, to_mint, user_from_token, user_to_token, from_vault, to_vault) +} + +/// Generic swap data extraction that works with any instruction type implementing InnerInstructionLike +fn extract_swap_data_from_instructions( + event: &DexEvent, + instructions: impl Iterator, + current_index: i8, + accounts: &[Pubkey], +) -> Option { + let (mut swap_data, fm, tm, uft, utt, fv, tv) = extract_swap_context(event); + + let user_to_token = utt.unwrap_or_default(); + let user_from_token = uft.unwrap_or_default(); + let to_vault = tv.unwrap_or_default(); + let from_vault = fv.unwrap_or_default(); + let to_mint = tm.unwrap_or_default(); + let from_mint = fm.unwrap_or_default(); + + for instruction in instructions.skip((current_index + 1) as usize) { + let program_id = accounts[instruction.program_id_index()]; + if !SYSTEM_PROGRAMS.contains(&program_id) { + break; + } + let data = instruction.data(); + let accs = instruction.accounts(); + + if data.len() < 8 { + continue; + } + + let get_pubkey = |i: usize| accounts[accs[i] as usize]; + let (source, destination, amount) = match data[0] { + 12 if accs.len() >= 4 => { + let amt = u64::from_le_bytes(data[1..9].try_into().unwrap()); + (get_pubkey(0), get_pubkey(2), amt) + } + 3 if accs.len() >= 3 => { + let amt = u64::from_le_bytes(data[1..9].try_into().unwrap()); + (get_pubkey(0), get_pubkey(1), amt) + } + 2 if accs.len() >= 2 => { + let amt = u64::from_le_bytes(data[4..12].try_into().unwrap()); + (get_pubkey(0), get_pubkey(1), amt) + } + _ => continue, + }; + + match (source, destination) { + (s, d) if s == user_to_token && d == to_vault => { + swap_data.from_mint = to_mint; + swap_data.from_amount = amount; + } + (s, d) if s == from_vault && d == user_from_token => { + swap_data.to_mint = from_mint; + swap_data.to_amount = amount; + } + (s, d) if s == user_from_token && d == from_vault => { + swap_data.from_mint = from_mint; + swap_data.from_amount = amount; + } + (s, d) if s == to_vault && d == user_to_token => { + swap_data.to_mint = to_mint; + swap_data.to_amount = amount; + } + (s, d) if s == user_from_token && d == to_vault => { + swap_data.from_mint = from_mint; + swap_data.from_amount = amount; + } + (s, d) if s == from_vault && d == user_to_token => { + swap_data.to_mint = to_mint; + swap_data.to_amount = amount; + } + _ => {} + } + if swap_data.from_mint != Pubkey::default() && swap_data.to_mint != Pubkey::default() { + break; + } + if swap_data.from_amount != 0 && swap_data.to_amount != 0 { + break; + } + } + + if swap_data.from_mint != Pubkey::default() + || swap_data.to_mint != Pubkey::default() + || swap_data.from_amount != 0 + || swap_data.to_amount != 0 + { + Some(swap_data) + } else { + None + } +} + +/// Parse token transfer data from standard Solana inner instructions pub fn parse_swap_data_from_next_instructions( event: &DexEvent, inner_instruction: &solana_transaction_status::InnerInstructions, current_index: i8, accounts: &[Pubkey], ) -> Option { - let mut swap_data = SwapData { - from_mint: Pubkey::default(), - to_mint: Pubkey::default(), - from_amount: 0, - to_amount: 0, - description: None, - }; - - // 先根据 event 取出关键信息 - // let mut user: Option = None; - let mut from_mint: Option = None; - let mut to_mint: Option = None; - let mut user_from_token: Option = None; - let mut user_to_token: Option = None; - let mut from_vault: Option = None; - let mut to_vault: Option = None; - - match event { - DexEvent::BonkTradeEvent(e) => { - // user = Some(e.payer); - from_mint = Some(e.base_token_mint); - to_mint = Some(e.quote_token_mint); - user_from_token = Some(e.user_base_token); - user_to_token = Some(e.user_quote_token); - from_vault = Some(e.base_vault); - to_vault = Some(e.quote_vault); - } - DexEvent::PumpFunTradeEvent(e) => { - swap_data.from_mint = if e.is_buy { *SOL_MINT } else { e.mint }; - swap_data.to_mint = if e.is_buy { e.mint } else { *SOL_MINT }; - } - DexEvent::PumpSwapBuyEvent(e) => { - swap_data.from_mint = e.quote_mint; - swap_data.to_mint = e.base_mint; - } - DexEvent::PumpSwapSellEvent(e) => { - swap_data.from_mint = e.base_mint; - swap_data.to_mint = e.quote_mint; - } - DexEvent::RaydiumCpmmSwapEvent(e) => { - // user = Some(e.payer); - from_mint = Some(e.input_token_mint); - to_mint = Some(e.output_token_mint); - user_from_token = Some(e.input_token_account); - user_to_token = Some(e.output_token_account); - from_vault = Some(e.input_vault); - to_vault = Some(e.output_vault); - } - DexEvent::RaydiumClmmSwapEvent(e) => { - // user = Some(e.payer); - swap_data.description = - Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".into()); - user_from_token = Some(e.input_token_account); - user_to_token = Some(e.output_token_account); - from_vault = Some(e.input_vault); - to_vault = Some(e.output_vault); - } - DexEvent::RaydiumClmmSwapV2Event(e) => { - // user = Some(e.payer); - from_mint = Some(e.input_vault_mint); - to_mint = Some(e.output_vault_mint); - user_from_token = Some(e.input_token_account); - user_to_token = Some(e.output_token_account); - from_vault = Some(e.input_vault); - to_vault = Some(e.output_vault); - } - DexEvent::RaydiumAmmV4SwapEvent(e) => { - // user = Some(e.user_source_owner); - swap_data.description = - Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".into()); - user_from_token = Some(e.user_source_token_account); - user_to_token = Some(e.user_destination_token_account); - from_vault = Some(e.pool_pc_token_account); - to_vault = Some(e.pool_coin_token_account); - } - _ => {} - } - - let user_to_token = user_to_token.unwrap_or_default(); - let user_from_token = user_from_token.unwrap_or_default(); - let to_vault = to_vault.unwrap_or_default(); - let from_vault = from_vault.unwrap_or_default(); - let to_mint = to_mint.unwrap_or_default(); - let from_mint = from_mint.unwrap_or_default(); - - // 单次循环完成提取和判断 - for instruction in inner_instruction.instructions.iter().skip((current_index + 1) as usize) { - let compiled = &instruction.instruction; - let program_id = accounts[compiled.program_id_index as usize]; - if !SYSTEM_PROGRAMS.contains(&program_id) { - break; - } - let data = &compiled.data; - - // 使用 SIMD 验证数据格式 - if !SimdUtils::validate_data_format(data, 8) { - continue; - } - - let get_pubkey = |i: usize| accounts[compiled.accounts[i] as usize]; - let (source, destination, amount) = match data[0] { - 12 if compiled.accounts.len() >= 4 => { - let amt = u64::from_le_bytes(data[1..9].try_into().unwrap()); - (get_pubkey(0), get_pubkey(2), amt) - } - 3 if compiled.accounts.len() >= 3 => { - let amt = u64::from_le_bytes(data[1..9].try_into().unwrap()); - (get_pubkey(0), get_pubkey(1), amt) - } - 2 if compiled.accounts.len() >= 2 => { - let amt = u64::from_le_bytes(data[4..12].try_into().unwrap()); - (get_pubkey(0), get_pubkey(1), amt) - } - _ => continue, - }; - - match (source, destination) { - (s, d) if s == user_to_token && d == to_vault => { - swap_data.from_mint = to_mint; - swap_data.from_amount = amount; - } - (s, d) if s == from_vault && d == user_from_token => { - swap_data.to_mint = from_mint; - swap_data.to_amount = amount; - } - (s, d) if s == user_from_token && d == from_vault => { - swap_data.from_mint = from_mint; - swap_data.from_amount = amount; - } - (s, d) if s == to_vault && d == user_to_token => { - swap_data.to_mint = to_mint; - swap_data.to_amount = amount; - } - (s, d) if s == user_from_token && d == to_vault => { - swap_data.from_mint = from_mint; - swap_data.from_amount = amount; - } - (s, d) if s == from_vault && d == user_to_token => { - swap_data.to_mint = to_mint; - swap_data.to_amount = amount; - } - _ => {} - } - if swap_data.from_mint != Pubkey::default() && swap_data.to_mint != Pubkey::default() { - break; - } - if swap_data.from_amount != 0 && swap_data.to_amount != 0 { - break; - } - } - - if swap_data.from_mint != Pubkey::default() - || swap_data.to_mint != Pubkey::default() - || swap_data.from_amount != 0 - || swap_data.to_amount != 0 - { - Some(swap_data) - } else { - None - } + extract_swap_data_from_instructions( + event, + inner_instruction.instructions.iter().map(|ix| ix.instruction.clone()), + current_index, + accounts, + ) } -/// Parse token transfer data from next instructions -/// TODO: - wait refactor +/// Parse token transfer data from gRPC inner instructions pub fn parse_swap_data_from_next_grpc_instructions( event: &DexEvent, inner_instruction: &yellowstone_grpc_proto::prelude::InnerInstructions, current_index: i8, accounts: &[Pubkey], ) -> Option { - let mut swap_data = SwapData { - from_mint: Pubkey::default(), - to_mint: Pubkey::default(), - from_amount: 0, - to_amount: 0, - description: None, - }; - - // 先根据 event 取出关键信息 - // let mut user: Option = None; - let mut from_mint: Option = None; - let mut to_mint: Option = None; - let mut user_from_token: Option = None; - let mut user_to_token: Option = None; - let mut from_vault: Option = None; - let mut to_vault: Option = None; - - match event { - DexEvent::BonkTradeEvent(e) => { - // user = Some(e.payer); - from_mint = Some(e.base_token_mint); - to_mint = Some(e.quote_token_mint); - user_from_token = Some(e.user_base_token); - user_to_token = Some(e.user_quote_token); - from_vault = Some(e.base_vault); - to_vault = Some(e.quote_vault); - } - DexEvent::PumpFunTradeEvent(e) => { - swap_data.from_mint = if e.is_buy { *SOL_MINT } else { e.mint }; - swap_data.to_mint = if e.is_buy { e.mint } else { *SOL_MINT }; - } - DexEvent::PumpSwapBuyEvent(e) => { - swap_data.from_mint = e.quote_mint; - swap_data.to_mint = e.base_mint; - } - DexEvent::PumpSwapSellEvent(e) => { - swap_data.from_mint = e.base_mint; - swap_data.to_mint = e.quote_mint; - } - DexEvent::RaydiumCpmmSwapEvent(e) => { - // user = Some(e.payer); - from_mint = Some(e.input_token_mint); - to_mint = Some(e.output_token_mint); - user_from_token = Some(e.input_token_account); - user_to_token = Some(e.output_token_account); - from_vault = Some(e.input_vault); - to_vault = Some(e.output_vault); - } - DexEvent::RaydiumClmmSwapEvent(e) => { - // user = Some(e.payer); - swap_data.description = - Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".into()); - user_from_token = Some(e.input_token_account); - user_to_token = Some(e.output_token_account); - from_vault = Some(e.input_vault); - to_vault = Some(e.output_vault); - } - DexEvent::RaydiumClmmSwapV2Event(e) => { - // user = Some(e.payer); - from_mint = Some(e.input_vault_mint); - to_mint = Some(e.output_vault_mint); - user_from_token = Some(e.input_token_account); - user_to_token = Some(e.output_token_account); - from_vault = Some(e.input_vault); - to_vault = Some(e.output_vault); - } - DexEvent::RaydiumAmmV4SwapEvent(e) => { - // user = Some(e.user_source_owner); - swap_data.description = - Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".into()); - user_from_token = Some(e.user_source_token_account); - user_to_token = Some(e.user_destination_token_account); - from_vault = Some(e.pool_pc_token_account); - to_vault = Some(e.pool_coin_token_account); - } - _ => {} - } - - let user_to_token = user_to_token.unwrap_or_default(); - let user_from_token = user_from_token.unwrap_or_default(); - let to_vault = to_vault.unwrap_or_default(); - let from_vault = from_vault.unwrap_or_default(); - let to_mint = to_mint.unwrap_or_default(); - let from_mint = from_mint.unwrap_or_default(); - - // 单次循环完成提取和判断 - for instruction in inner_instruction.instructions.iter().skip((current_index + 1) as usize) { - let compiled = &instruction; - let program_id = accounts[compiled.program_id_index as usize]; - if !SYSTEM_PROGRAMS.contains(&program_id) { - break; - } - let data = &compiled.data; - - // 使用 SIMD 验证数据格式 - if !SimdUtils::validate_data_format(data, 8) { - continue; - } - - let get_pubkey = |i: usize| accounts[compiled.accounts[i] as usize]; - let (source, destination, amount) = match data[0] { - 12 if compiled.accounts.len() >= 4 => { - let amt = u64::from_le_bytes(data[1..9].try_into().unwrap()); - (get_pubkey(0), get_pubkey(2), amt) - } - 3 if compiled.accounts.len() >= 3 => { - let amt = u64::from_le_bytes(data[1..9].try_into().unwrap()); - (get_pubkey(0), get_pubkey(1), amt) - } - 2 if compiled.accounts.len() >= 2 => { - let amt = u64::from_le_bytes(data[4..12].try_into().unwrap()); - (get_pubkey(0), get_pubkey(1), amt) - } - _ => continue, - }; - - match (source, destination) { - (s, d) if s == user_to_token && d == to_vault => { - swap_data.from_mint = to_mint; - swap_data.from_amount = amount; - } - (s, d) if s == from_vault && d == user_from_token => { - swap_data.to_mint = from_mint; - swap_data.to_amount = amount; - } - (s, d) if s == user_from_token && d == from_vault => { - swap_data.from_mint = from_mint; - swap_data.from_amount = amount; - } - (s, d) if s == to_vault && d == user_to_token => { - swap_data.to_mint = to_mint; - swap_data.to_amount = amount; - } - (s, d) if s == user_from_token && d == to_vault => { - swap_data.from_mint = from_mint; - swap_data.from_amount = amount; - } - (s, d) if s == from_vault && d == user_to_token => { - swap_data.to_mint = to_mint; - swap_data.to_amount = amount; - } - _ => {} - } - if swap_data.from_mint != Pubkey::default() && swap_data.to_mint != Pubkey::default() { - break; - } - if swap_data.from_amount != 0 && swap_data.to_amount != 0 { - break; - } - } - - if swap_data.from_mint != Pubkey::default() - || swap_data.to_mint != Pubkey::default() - || swap_data.from_amount != 0 - || swap_data.to_amount != 0 - { - Some(swap_data) - } else { - None - } + extract_swap_data_from_instructions( + event, + inner_instruction.instructions.iter().cloned(), + current_index, + accounts, + ) } diff --git a/src/streaming/event_parser/common/utils.rs b/src/streaming/event_parser/common/utils.rs index 56ec1a5..eb625c0 100755 --- a/src/streaming/event_parser/common/utils.rs +++ b/src/streaming/event_parser/common/utils.rs @@ -58,14 +58,14 @@ pub fn read_u8_le(data: &[u8], offset: usize) -> Option { } pub fn read_option_bool(data: &[u8], offset: &mut usize) -> Option> { - let has_value = data.get(*offset)?.clone(); + let has_value = data.get(*offset).copied()?; *offset += 1; if has_value == 0 { return Some(None); } - let value = data.get(*offset)?.clone(); + let value = data.get(*offset).copied()?; *offset += 1; Some(Some(value != 0)) diff --git a/src/streaming/event_parser/core/global_state.rs b/src/streaming/event_parser/core/global_state.rs index 201417c..415d7d0 100644 --- a/src/streaming/event_parser/core/global_state.rs +++ b/src/streaming/event_parser/core/global_state.rs @@ -49,16 +49,11 @@ impl GlobalState { return; // Another thread is cleaning up } - // Collect signatures to remove (random selection for simplicity) - let mut signatures_to_remove: Vec = self.signature_data.iter() + // Collect only the batch we need to remove (avoid allocating full list) + let signatures_to_remove: Vec = self.signature_data.iter() + .take(CLEANUP_BATCH_SIZE) .map(|entry| *entry.key()) .collect(); - - if signatures_to_remove.len() <= MAX_SIGNATURES { - return; // Race condition, already cleaned up - } - - signatures_to_remove.truncate(CLEANUP_BATCH_SIZE); // Remove old signatures atomically for signature in signatures_to_remove { diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index 0dc95f8..0a27eb7 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -90,122 +90,86 @@ pub enum DexEvent { SetComputeUnitPriceEvent(SetComputeUnitPriceEvent), } -impl DexEvent { - pub fn metadata(&self) -> &EventMetadata { - match self { - DexEvent::BonkTradeEvent(e) => &e.metadata, - DexEvent::BonkPoolCreateEvent(e) => &e.metadata, - DexEvent::BonkMigrateToAmmEvent(e) => &e.metadata, - DexEvent::BonkMigrateToCpswapEvent(e) => &e.metadata, - DexEvent::BonkPoolStateAccountEvent(e) => &e.metadata, - DexEvent::BonkGlobalConfigAccountEvent(e) => &e.metadata, - DexEvent::BonkPlatformConfigAccountEvent(e) => &e.metadata, - DexEvent::PumpFunCreateTokenEvent(e) => &e.metadata, - DexEvent::PumpFunCreateV2TokenEvent(e) => &e.metadata, - DexEvent::PumpFunTradeEvent(e) => &e.metadata, - DexEvent::PumpFunMigrateEvent(e) => &e.metadata, - DexEvent::PumpFunBondingCurveAccountEvent(e) => &e.metadata, - DexEvent::PumpFunGlobalAccountEvent(e) => &e.metadata, - DexEvent::PumpSwapBuyEvent(e) => &e.metadata, - DexEvent::PumpSwapSellEvent(e) => &e.metadata, - DexEvent::PumpSwapCreatePoolEvent(e) => &e.metadata, - DexEvent::PumpSwapDepositEvent(e) => &e.metadata, - DexEvent::PumpSwapWithdrawEvent(e) => &e.metadata, - DexEvent::PumpSwapGlobalConfigAccountEvent(e) => &e.metadata, - DexEvent::PumpSwapPoolAccountEvent(e) => &e.metadata, - DexEvent::RaydiumAmmV4SwapEvent(e) => &e.metadata, - DexEvent::RaydiumAmmV4DepositEvent(e) => &e.metadata, - DexEvent::RaydiumAmmV4WithdrawEvent(e) => &e.metadata, - DexEvent::RaydiumAmmV4WithdrawPnlEvent(e) => &e.metadata, - DexEvent::RaydiumAmmV4Initialize2Event(e) => &e.metadata, - DexEvent::RaydiumAmmV4AmmInfoAccountEvent(e) => &e.metadata, - DexEvent::RaydiumClmmSwapEvent(e) => &e.metadata, - DexEvent::RaydiumClmmSwapV2Event(e) => &e.metadata, - DexEvent::RaydiumClmmClosePositionEvent(e) => &e.metadata, - DexEvent::RaydiumClmmIncreaseLiquidityV2Event(e) => &e.metadata, - DexEvent::RaydiumClmmDecreaseLiquidityV2Event(e) => &e.metadata, - DexEvent::RaydiumClmmCreatePoolEvent(e) => &e.metadata, - DexEvent::RaydiumClmmOpenPositionWithToken22NftEvent(e) => &e.metadata, - DexEvent::RaydiumClmmOpenPositionV2Event(e) => &e.metadata, - DexEvent::RaydiumClmmAmmConfigAccountEvent(e) => &e.metadata, - DexEvent::RaydiumClmmPoolStateAccountEvent(e) => &e.metadata, - DexEvent::RaydiumClmmTickArrayStateAccountEvent(e) => &e.metadata, - DexEvent::RaydiumCpmmSwapEvent(e) => &e.metadata, - DexEvent::RaydiumCpmmDepositEvent(e) => &e.metadata, - DexEvent::RaydiumCpmmWithdrawEvent(e) => &e.metadata, - DexEvent::RaydiumCpmmInitializeEvent(e) => &e.metadata, - DexEvent::RaydiumCpmmAmmConfigAccountEvent(e) => &e.metadata, - DexEvent::RaydiumCpmmPoolStateAccountEvent(e) => &e.metadata, - DexEvent::MeteoraDammV2SwapEvent(e) => &e.metadata, - DexEvent::MeteoraDammV2Swap2Event(e) => &e.metadata, - DexEvent::MeteoraDammV2InitializePoolEvent(e) => &e.metadata, - DexEvent::MeteoraDammV2InitializeCustomizablePoolEvent(e) => &e.metadata, - DexEvent::MeteoraDammV2InitializePoolWithDynamicConfigEvent(e) => &e.metadata, - DexEvent::TokenAccountEvent(e) => &e.metadata, - DexEvent::NonceAccountEvent(e) => &e.metadata, - DexEvent::TokenInfoEvent(e) => &e.metadata, - DexEvent::BlockMetaEvent(e) => &e.metadata, - DexEvent::SetComputeUnitLimitEvent(e) => &e.metadata, - DexEvent::SetComputeUnitPriceEvent(e) => &e.metadata, - } - } +/// Macro to generate metadata accessors for all DexEvent variants +macro_rules! impl_dex_event_metadata { + ($($variant:ident),* $(,)?) => { + impl DexEvent { + pub fn metadata(&self) -> &EventMetadata { + match self { + $(DexEvent::$variant(e) => &e.metadata,)* + } + } - pub fn metadata_mut(&mut self) -> &mut EventMetadata { - match self { - DexEvent::BonkTradeEvent(e) => &mut e.metadata, - DexEvent::BonkPoolCreateEvent(e) => &mut e.metadata, - DexEvent::BonkMigrateToAmmEvent(e) => &mut e.metadata, - DexEvent::BonkMigrateToCpswapEvent(e) => &mut e.metadata, - DexEvent::BonkPoolStateAccountEvent(e) => &mut e.metadata, - DexEvent::BonkGlobalConfigAccountEvent(e) => &mut e.metadata, - DexEvent::BonkPlatformConfigAccountEvent(e) => &mut e.metadata, - DexEvent::PumpFunCreateTokenEvent(e) => &mut e.metadata, - DexEvent::PumpFunCreateV2TokenEvent(e) => &mut e.metadata, - DexEvent::PumpFunTradeEvent(e) => &mut e.metadata, - DexEvent::PumpFunMigrateEvent(e) => &mut e.metadata, - DexEvent::PumpFunBondingCurveAccountEvent(e) => &mut e.metadata, - DexEvent::PumpFunGlobalAccountEvent(e) => &mut e.metadata, - DexEvent::PumpSwapBuyEvent(e) => &mut e.metadata, - DexEvent::PumpSwapSellEvent(e) => &mut e.metadata, - DexEvent::PumpSwapCreatePoolEvent(e) => &mut e.metadata, - DexEvent::PumpSwapDepositEvent(e) => &mut e.metadata, - DexEvent::PumpSwapWithdrawEvent(e) => &mut e.metadata, - DexEvent::PumpSwapGlobalConfigAccountEvent(e) => &mut e.metadata, - DexEvent::PumpSwapPoolAccountEvent(e) => &mut e.metadata, - DexEvent::RaydiumAmmV4SwapEvent(e) => &mut e.metadata, - DexEvent::RaydiumAmmV4DepositEvent(e) => &mut e.metadata, - DexEvent::RaydiumAmmV4WithdrawEvent(e) => &mut e.metadata, - DexEvent::RaydiumAmmV4WithdrawPnlEvent(e) => &mut e.metadata, - DexEvent::RaydiumAmmV4Initialize2Event(e) => &mut e.metadata, - DexEvent::RaydiumAmmV4AmmInfoAccountEvent(e) => &mut e.metadata, - DexEvent::RaydiumClmmSwapEvent(e) => &mut e.metadata, - DexEvent::RaydiumClmmSwapV2Event(e) => &mut e.metadata, - DexEvent::RaydiumClmmClosePositionEvent(e) => &mut e.metadata, - DexEvent::RaydiumClmmIncreaseLiquidityV2Event(e) => &mut e.metadata, - DexEvent::RaydiumClmmDecreaseLiquidityV2Event(e) => &mut e.metadata, - DexEvent::RaydiumClmmCreatePoolEvent(e) => &mut e.metadata, - DexEvent::RaydiumClmmOpenPositionWithToken22NftEvent(e) => &mut e.metadata, - DexEvent::RaydiumClmmOpenPositionV2Event(e) => &mut e.metadata, - DexEvent::RaydiumClmmAmmConfigAccountEvent(e) => &mut e.metadata, - DexEvent::RaydiumClmmPoolStateAccountEvent(e) => &mut e.metadata, - DexEvent::RaydiumClmmTickArrayStateAccountEvent(e) => &mut e.metadata, - DexEvent::RaydiumCpmmSwapEvent(e) => &mut e.metadata, - DexEvent::RaydiumCpmmDepositEvent(e) => &mut e.metadata, - DexEvent::RaydiumCpmmWithdrawEvent(e) => &mut e.metadata, - DexEvent::RaydiumCpmmInitializeEvent(e) => &mut e.metadata, - DexEvent::RaydiumCpmmAmmConfigAccountEvent(e) => &mut e.metadata, - DexEvent::RaydiumCpmmPoolStateAccountEvent(e) => &mut e.metadata, - DexEvent::MeteoraDammV2SwapEvent(e) => &mut e.metadata, - DexEvent::MeteoraDammV2Swap2Event(e) => &mut e.metadata, - DexEvent::MeteoraDammV2InitializePoolEvent(e) => &mut e.metadata, - DexEvent::MeteoraDammV2InitializeCustomizablePoolEvent(e) => &mut e.metadata, - DexEvent::MeteoraDammV2InitializePoolWithDynamicConfigEvent(e) => &mut e.metadata, - DexEvent::TokenAccountEvent(e) => &mut e.metadata, - DexEvent::NonceAccountEvent(e) => &mut e.metadata, - DexEvent::TokenInfoEvent(e) => &mut e.metadata, - DexEvent::BlockMetaEvent(e) => &mut e.metadata, - DexEvent::SetComputeUnitLimitEvent(e) => &mut e.metadata, - DexEvent::SetComputeUnitPriceEvent(e) => &mut e.metadata, + pub fn metadata_mut(&mut self) -> &mut EventMetadata { + match self { + $(DexEvent::$variant(e) => &mut e.metadata,)* + } + } } - } + }; } + +impl_dex_event_metadata!( + // Bonk events + BonkTradeEvent, + BonkPoolCreateEvent, + BonkMigrateToAmmEvent, + BonkMigrateToCpswapEvent, + BonkPoolStateAccountEvent, + BonkGlobalConfigAccountEvent, + BonkPlatformConfigAccountEvent, + // PumpFun events + PumpFunCreateTokenEvent, + PumpFunCreateV2TokenEvent, + PumpFunTradeEvent, + PumpFunMigrateEvent, + PumpFunBondingCurveAccountEvent, + PumpFunGlobalAccountEvent, + // PumpSwap events + PumpSwapBuyEvent, + PumpSwapSellEvent, + PumpSwapCreatePoolEvent, + PumpSwapDepositEvent, + PumpSwapWithdrawEvent, + PumpSwapGlobalConfigAccountEvent, + PumpSwapPoolAccountEvent, + // Raydium AMM V4 events + RaydiumAmmV4SwapEvent, + RaydiumAmmV4DepositEvent, + RaydiumAmmV4WithdrawEvent, + RaydiumAmmV4WithdrawPnlEvent, + RaydiumAmmV4Initialize2Event, + RaydiumAmmV4AmmInfoAccountEvent, + // Raydium CLMM events + RaydiumClmmSwapEvent, + RaydiumClmmSwapV2Event, + RaydiumClmmClosePositionEvent, + RaydiumClmmIncreaseLiquidityV2Event, + RaydiumClmmDecreaseLiquidityV2Event, + RaydiumClmmCreatePoolEvent, + RaydiumClmmOpenPositionWithToken22NftEvent, + RaydiumClmmOpenPositionV2Event, + RaydiumClmmAmmConfigAccountEvent, + RaydiumClmmPoolStateAccountEvent, + RaydiumClmmTickArrayStateAccountEvent, + // Raydium CPMM events + RaydiumCpmmSwapEvent, + RaydiumCpmmDepositEvent, + RaydiumCpmmWithdrawEvent, + RaydiumCpmmInitializeEvent, + RaydiumCpmmAmmConfigAccountEvent, + RaydiumCpmmPoolStateAccountEvent, + // Meteora DAMM v2 events + MeteoraDammV2SwapEvent, + MeteoraDammV2Swap2Event, + MeteoraDammV2InitializePoolEvent, + MeteoraDammV2InitializeCustomizablePoolEvent, + MeteoraDammV2InitializePoolWithDynamicConfigEvent, + // Common events + TokenAccountEvent, + NonceAccountEvent, + TokenInfoEvent, + BlockMetaEvent, + SetComputeUnitLimitEvent, + SetComputeUnitPriceEvent, +); diff --git a/src/streaming/grpc/subscription.rs b/src/streaming/grpc/subscription.rs index 7144885..eb10515 100644 --- a/src/streaming/grpc/subscription.rs +++ b/src/streaming/grpc/subscription.rs @@ -85,7 +85,7 @@ impl SubscriptionManager { if event_type_filter.is_some() && !event_type_filter.unwrap().include_account_event() { return None; } - if account_filter.len() == 0 { + if account_filter.is_empty() { return None; } let mut accounts = HashMap::new(); diff --git a/src/streaming/grpc/types.rs b/src/streaming/grpc/types.rs index 9a526c2..b335290 100644 --- a/src/streaming/grpc/types.rs +++ b/src/streaming/grpc/types.rs @@ -104,63 +104,3 @@ impl Default for TransactionPretty { } } -// impl From for AccountPretty { -// fn from(account: SubscribeUpdateAccount) -> Self { -// let account_info = account.account.unwrap(); -// Self { -// slot: account.slot, -// signature: if let Some(txn_signature) = account_info.txn_signature { -// Signature::try_from(txn_signature.as_slice()).expect("valid signature") -// } else { -// Signature::default() -// }, -// pubkey: Pubkey::try_from(account_info.pubkey.as_slice()).expect("valid pubkey"), -// executable: account_info.executable, -// lamports: account_info.lamports, -// owner: Pubkey::try_from(account_info.owner.as_slice()).expect("valid pubkey"), -// rent_epoch: account_info.rent_epoch, -// data: account_info.data, -// recv_us: get_high_perf_clock(), -// } -// } -// } - -// impl From<(SubscribeUpdateBlockMeta, Option)> for BlockMetaPretty { -// fn from( -// (SubscribeUpdateBlockMeta { slot, blockhash, .. }, block_time): ( -// SubscribeUpdateBlockMeta, -// Option, -// ), -// ) -> Self { -// Self { -// block_hash: blockhash, -// block_time, -// slot, -// recv_us: get_high_perf_clock(), -// } -// } -// } - -// impl From<(SubscribeUpdateTransaction, Option)> for TransactionPretty { -// fn from( -// (SubscribeUpdateTransaction { transaction, slot }, block_time): ( -// SubscribeUpdateTransaction, -// Option, -// ), -// ) -> Self { -// let tx = transaction.expect("should be defined"); -// // 根据用户说明，交易索引在 transaction.index 中 -// let tx_index = tx.index; -// Self { -// slot, -// tx_index: Some(tx_index), // 提取交易索引 -// block_time, -// block_hash: String::new(), -// signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"), -// is_vote: tx.is_vote, -// tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx) -// .expect("valid tx with meta"), -// recv_us: get_high_perf_clock(), -// } -// } -// } From dad231a26b6a0d9a1a9b75f68d5727982672b4b2 Mon Sep 17 00:00:00 2001 From: Estereg <109712545+Estereg@users.noreply.github.com> Date: Sun, 8 Mar 2026 16:40:37 +0000 Subject: [PATCH 2/2] refactor: remove bs58 dependency and use solana_sdk program ID constants - Remove standalone bs58 dependency and rely on the solana_sdk re-export - Switch system and token program IDs to the idiomatic spl_token / solana_sdk constants - Use pubkey! and LazyLock for cleaner program ID initialization - Alias solana_entry::Entry as SolanaEntry to avoid a naming conflict with gRPC types - Remove unused FromStr imports and some redundant logic --- Cargo.toml | 1 - src/streaming/event_parser/common/types.rs | 12 +++++------- src/streaming/event_parser/core/event_parser.rs | 2 +- src/streaming/shred_stream.rs | 4 ++-- 4 files changed, 8 insertions(+), 11 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 33113ba..5c3e5db 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,6 @@ serde-big-array = "0.5.1" futures = "0.3.32" bincode = "1.3" anyhow = "1.0.102" -bs58 = "0.5.1" yellowstone-grpc-client = { version = "10.2.0" } yellowstone-grpc-proto = { version = "10.1.1" } tokio = { version = "1.50.0", features = ["full", "rt-multi-thread"]} diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index 700929e..f82f51e 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -2,7 +2,7 @@ use borsh::{BorshDeserialize, BorshSerialize}; use crossbeam_queue::ArrayQueue; use serde::{Deserialize, Serialize}; use solana_sdk::{pubkey::Pubkey, signature::Signature}; -use std::{borrow::Cow, fmt, str::FromStr, sync::Arc}; +use std::{borrow::Cow, fmt, sync::Arc}; use crate::streaming::event_parser::DexEvent; @@ -169,8 +169,6 @@ impl fmt::Display for EventType { } } - - #[derive( Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize, )] @@ -243,11 +241,11 @@ impl EventMetadata { } static SOL_MINT: std::sync::LazyLock = - std::sync::LazyLock::new(|| Pubkey::from_str("So11111111111111111111111111111111111111111").unwrap()); + std::sync::LazyLock::new(spl_token::native_mint::id); static SYSTEM_PROGRAMS: std::sync::LazyLock<[Pubkey; 3]> = std::sync::LazyLock::new(|| [ - Pubkey::from_str("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA").unwrap(), - Pubkey::from_str("TokenzQdBNbLqP5VEhdkAS6EPFLC1PHnBqCXEpPxuEb").unwrap(), - Pubkey::from_str("11111111111111111111111111111111").unwrap(), + spl_token::id(), + spl_token_2022::id(), + solana_sdk::pubkey!("11111111111111111111111111111111"), ]); /// Trait abstracting over different inner-instruction types for swap data extraction diff --git a/src/streaming/event_parser/core/event_parser.rs b/src/streaming/event_parser/core/event_parser.rs index d7703a0..db3b436 100644 --- a/src/streaming/event_parser/core/event_parser.rs +++ b/src/streaming/event_parser/core/event_parser.rs @@ -85,7 +85,7 @@ impl EventParser { let recent_blockhash = if message.recent_blockhash.is_empty() { None } else { - Some(bs58::encode(&message.recent_blockhash).into_string()) + Some(solana_sdk::bs58::encode(&message.recent_blockhash).into_string()) }; Self::parse_instruction_events_from_grpc_transaction( protocols, diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index 6ff6b86..bff8f36 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -12,7 +12,7 @@ use crate::streaming::event_parser::{Protocol, DexEvent}; use crate::streaming::grpc::MetricsManager; use crate::streaming::shred::pool::factory; use log::error; -use solana_entry::entry::Entry; +use solana_entry::entry::Entry as SolanaEntry; use super::ShredStreamGrpc; @@ -49,7 +49,7 @@ impl ShredStreamGrpc { while let Some(message) = stream.next().await { match message { Ok(msg) => { - if let Ok(entries) = bincode::deserialize::>(&msg.entries) { + if let Ok(entries) = bincode::deserialize::>(&msg.entries) { for entry in entries { for (tx_index, transaction) in entry.transactions.iter().enumerate() { let transaction_with_slot =