gavindiaz/solana-streamer
1
0
Fork 0
mirror of https://github.com/0xfnzero/solana-streamer.git synced 2026-08-16 18:38:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

Release solana-streamer-sdk v1.4.3

Browse Source 
Route gRPC, ShredStream, and account parsing through sol-parser-sdk.

Remove local protocol parsing implementations.

Depend on sol-parser-sdk v0.4.8 from crates.io.
This commit is contained in:
0xfnzero
2026-05-16 06:45:35 +08:00
parent 55e6bc6459
commit 92bac16fd4
43 changed files with 837 additions and 5460 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/streaming/event_parser/core/account_event_parser.rs
+23 -179
View File
@@ -1,18 +1,10 @@
use crate::streaming::event_parser::common::filter::EventTypeFilter;
use crate::streaming::event_parser::common::high_performance_clock::elapsed_micros_since;
use crate::streaming::event_parser::common::{EventMetadata, EventType, ProtocolType};
use crate::streaming::event_parser::common::{EventMetadata, EventType};
use crate::streaming::event_parser::core::traits::DexEvent;
use crate::streaming::event_parser::Protocol;
use crate::streaming::grpc::AccountPretty;
use serde::{Deserialize, Serialize};
use solana_account_decoder::parse_nonce::parse_nonce;
use solana_sdk::pubkey::Pubkey;
use spl_token::solana_program::program_pack::Pack;
use spl_token::state::{Account, Mint};
use spl_token_2022::{
extension::StateWithExtensions,
state::{Account as Account2022, Mint as Mint2022},
};
/// 通用账户事件
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
@@ -61,186 +53,38 @@ impl AccountEventParser {
account: AccountPretty,
event_type_filter: Option<&EventTypeFilter>,
) -> Option<DexEvent> {
use crate::streaming::event_parser::core::dispatcher::EventDispatcher;
// 1. 尝试从账户 discriminator 解析（协议特定账户）
if account.data.len() >= 8 {
let discriminator = &account.data[0..8];
// 尝试识别协议类型
if let Some(protocol) = EventDispatcher::match_protocol_by_program_id(&account.owner) {
// 检查是否在请求的协议列表中
if protocols.contains(&protocol) {
// 构建临时元数据（protocol会被dispatcher设置，event_type会在parser中设置）
let metadata = EventMetadata {
slot: account.slot,
signature: account.signature,
protocol: ProtocolType::Common, // 会被 EventDispatcher::dispatch_account 设置
event_type: EventType::default(), // 会被具体 parser 设置
program_id: account.owner,
recv_us: account.recv_us,
handle_us: elapsed_micros_since(account.recv_us),
..Default::default()
};
// 使用 dispatcher 解析
if let Some(event) = EventDispatcher::dispatch_account(
protocol,
discriminator,
&account,
metadata,
) {
// 应用事件类型过滤
if let Some(filter) = event_type_filter {
if filter.passes_event_type(&event.metadata().event_type) {
return Some(event);
}
// 不匹配过滤器，继续尝试其他解析方式
} else {
return Some(event);
}
}
}
}
}
// 2. 尝试解析特殊账户类型（Token、Nonce等）
// 这些是通用的，不属于特定协议
let metadata = EventMetadata {
slot: account.slot,
signature: account.signature,
protocol: ProtocolType::Common,
event_type: EventType::default(),
program_id: account.owner,
recv_us: account.recv_us,
handle_us: elapsed_micros_since(account.recv_us),
..Default::default()
};
// 尝试解析 Nonce 账户
if let Some(event) = Self::parse_nonce_account_event(&account, metadata.clone()) {
if let Some(filter) = event_type_filter {
if filter.passes_event_type(&event.metadata().event_type) {
return Some(event);
}
} else {
return Some(event);
}
}
// 尝试解析 Token 账户
if let Some(event) = Self::parse_token_account_event(&account, metadata) {
if let Some(filter) = event_type_filter {
if filter.passes_event_type(&event.metadata().event_type) {
return Some(event);
}
} else {
return Some(event);
}
}
None
crate::streaming::parser_sdk_bridge::parse_sdk_account_event(
&account,
protocols,
event_type_filter,
)
}
pub fn parse_token_account_event(
account: &AccountPretty,
mut metadata: EventMetadata,
metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::TokenAccount;
let pubkey = account.pubkey;
let executable = account.executable;
let lamports = account.lamports;
let owner = account.owner;
let rent_epoch = account.rent_epoch;
// Spl Token Mint
if account.data.len() >= Mint::LEN {
if let Ok(mint) = Mint::unpack_from_slice(&account.data) {
let mut metadata = metadata.clone();
metadata.event_type = EventType::TokenInfo;
let mut event = TokenInfoEvent {
metadata,
pubkey,
executable,
lamports,
owner,
rent_epoch,
supply: mint.supply,
decimals: mint.decimals,
};
let recv_delta = elapsed_micros_since(account.recv_us);
event.metadata.handle_us = recv_delta;
return Some(DexEvent::TokenInfoEvent(event));
}
}
// Spl Token2022 Mint
if account.data.len() >= Account2022::LEN {
if let Ok(mint) = StateWithExtensions::<Mint2022>::unpack(&account.data) {
let mut metadata = metadata.clone();
metadata.event_type = EventType::TokenInfo;
let mut event = TokenInfoEvent {
metadata,
pubkey,
executable,
lamports,
owner,
rent_epoch,
supply: mint.base.supply,
decimals: mint.base.decimals,
};
let recv_delta = elapsed_micros_since(account.recv_us);
event.metadata.handle_us = recv_delta;
return Some(DexEvent::TokenInfoEvent(event));
}
}
let amount = if account.owner.to_bytes() == spl_token_2022::ID.to_bytes() {
StateWithExtensions::<Account2022>::unpack(&account.data)
.ok()
.map(|info| info.base.amount)
} else {
Account::unpack(&account.data).ok().map(|info| info.amount)
};
let mut event = TokenAccountEvent {
metadata,
pubkey,
executable,
lamports,
owner,
rent_epoch,
amount,
token_owner: account.owner,
};
let recv_delta = elapsed_micros_since(account.recv_us);
event.metadata.handle_us = recv_delta;
Some(DexEvent::TokenAccountEvent(event))
let mut account = account.clone();
overlay_metadata(&mut account, &metadata);
let filter = EventTypeFilter::include_only([EventType::TokenAccount]);
crate::streaming::parser_sdk_bridge::parse_sdk_account_event(&account, &[], Some(&filter))
}
pub fn parse_nonce_account_event(
account: &AccountPretty,
mut metadata: EventMetadata,
metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::NonceAccount;
let mut account = account.clone();
overlay_metadata(&mut account, &metadata);
let filter = EventTypeFilter::include_only([EventType::NonceAccount]);
crate::streaming::parser_sdk_bridge::parse_sdk_account_event(&account, &[], Some(&filter))
}
}
if let Ok(info) = parse_nonce(&account.data) {
match info {
solana_account_decoder::parse_nonce::UiNonceState::Initialized(details) => {
let mut event = NonceAccountEvent {
metadata,
pubkey: account.pubkey,
executable: account.executable,
lamports: account.lamports,
owner: account.owner,
rent_epoch: account.rent_epoch,
nonce: details.blockhash,
authority: details.authority,
};
event.metadata.handle_us = elapsed_micros_since(account.recv_us);
return Some(DexEvent::NonceAccountEvent(event));
}
solana_account_decoder::parse_nonce::UiNonceState::Uninitialized => {}
}
}
None
fn overlay_metadata(account: &mut AccountPretty, metadata: &EventMetadata) {
account.slot = metadata.slot;
account.signature = metadata.signature;
if metadata.recv_us != 0 {
account.recv_us = metadata.recv_us;
}
}
src/streaming/event_parser/core/common_event_parser.rs
-44
View File
@@ -1,15 +1,9 @@
use crate::streaming::event_parser::common::high_performance_clock::elapsed_micros_since;
use crate::streaming::event_parser::common::types::{EventType, ProtocolType};
use crate::streaming::event_parser::common::EventMetadata;
use crate::streaming::event_parser::core::traits::DexEvent;
use crate::streaming::event_parser::protocols::block::block_meta_event::BlockMetaEvent;
use borsh::BorshDeserialize;
use serde::{Deserialize, Serialize};
use solana_sdk::pubkey::Pubkey;
// Compute Budget Program ID
pub const COMPUTE_BUDGET_PROGRAM_ID: Pubkey =
solana_sdk::pubkey!("ComputeBudget111111111111111111111111111111");
/// SetComputeUnitLimit 事件
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
@@ -42,42 +36,4 @@ impl CommonEventParser {
block_meta_event.metadata.handle_us = elapsed_micros_since(recv_us);
DexEvent::BlockMetaEvent(block_meta_event)
}
/// 解析 Compute Budget 指令
pub fn parse_compute_budget_instruction(
instruction_data: &[u8],
mut metadata: EventMetadata,
) -> Option<DexEvent> {
if instruction_data.is_empty() {
return None;
}
// 设置 protocol 为 Common
metadata.protocol = ProtocolType::Common;
// Compute Budget 指令使用单字节判别器
match instruction_data[0] {
// SetComputeUnitLimit: discriminator = 2
2 => {
if instruction_data.len() < 5 {
return None;
}
let units = u32::from_le_bytes(instruction_data[1..5].try_into().ok()?);
metadata.event_type = EventType::SetComputeUnitLimit;
let event = SetComputeUnitLimitEvent { metadata, units };
Some(DexEvent::SetComputeUnitLimitEvent(event))
}
// SetComputeUnitPrice: discriminator = 3
3 => {
if instruction_data.len() < 9 {
return None;
}
let micro_lamports = u64::from_le_bytes(instruction_data[1..9].try_into().ok()?);
metadata.event_type = EventType::SetComputeUnitPrice;
let event = SetComputeUnitPriceEvent { metadata, micro_lamports };
Some(DexEvent::SetComputeUnitPriceEvent(event))
}
_ => None,
}
}
}
src/streaming/event_parser/core/dispatcher.rs
+119 -257
View File
@@ -1,331 +1,193 @@
//! 事件路由入口（类比 sol-parser-sdk 的 `instr`，区分「原生字节解析」与「sdk 事件对齐」）。
//! SDK-backed event dispatcher kept for compatibility with older streamer APIs.
//!
//! ## 代码去哪找
//! - **`protocols/<协议>/parser.rs`** — Yellowstone / shred 路径下的顶层与 inner 指令解析（手写）。
//! - **`streaming/parser_sdk_bridge/`** — `sol-parser-sdk::DexEvent` → streamer `DexEvent` 字段映射。
//! - **`protocols/sol_parser_forward/native.rs`** — Orca / Meteora Pools & DLMM：调用 sdk `instr` 后再走 bridge。
//!
//! ## 设计原则
//! - **单一职责**: 每个函数只负责一件事（路由、解析、合并分离）
//! - **灵活性**: 调用方可以选择是否合并，或自定义合并逻辑
//! - **可测试性**: 每个函数都可以独立测试
//! Streamer no longer owns protocol parsers here. The dispatcher only adapts
//! streamer metadata and routes to `sol-parser-sdk` parsers, then converts SDK
//! events back to streamer `DexEvent`.
use crate::streaming::event_parser::{
common::EventMetadata,
core::common_event_parser::{CommonEventParser, COMPUTE_BUDGET_PROGRAM_ID},
protocols::{
bonk::parser as bonk, meteora_damm_v2::parser as meteora_damm_v2,
pumpfun::parser as pumpfun, pumpswap::parser as pumpswap,
raydium_amm_v4::parser as raydium_amm_v4, raydium_clmm::parser as raydium_clmm,
raydium_cpmm::parser as raydium_cpmm, sol_parser_forward,
},
DexEvent, Protocol,
use crate::streaming::event_parser::{common::EventMetadata, DexEvent, Protocol};
use crate::streaming::parser_sdk_bridge::{
block_timestamp_from_stream_meta, convert_parser_event, fuse_streamer_ix_ctx,
parse_sdk_account_event,
};
use sol_parser_sdk::core::events::EventMetadata as PbEventMetadata;
use sol_parser_sdk::instr::{
all_inner, program_ids, pump_amm_inner, pump_inner, raydium_clmm_inner,
};
use solana_sdk::pubkey::Pubkey;
/// 中心事件解析调度器
///
/// 负责将解析请求路由到对应协议的解析函数
pub struct EventDispatcher;
impl EventDispatcher {
/// 解析 instruction 事件（只解析，不合并）
///
/// # 参数
/// - `protocol`: 协议类型
/// - `instruction_discriminator`: 指令判别器 (8 bytes)
/// - `instruction_data`: 指令数据
/// - `accounts`: 账户公钥列表
/// - `metadata`: 事件元数据
///
/// # 返回
/// 解析成功返回 `Some(DexEvent)`，否则返回 `None`
#[inline]
pub fn dispatch_instruction(
protocol: Protocol,
instruction_discriminator: &[u8],
instruction_data: &[u8],
accounts: &[Pubkey],
mut metadata: EventMetadata,
metadata: EventMetadata,
) -> Option<DexEvent> {
// 根据协议类型设置 metadata.protocol
use crate::streaming::event_parser::common::ProtocolType;
metadata.protocol = match protocol {
Protocol::PumpFun => ProtocolType::PumpFun,
Protocol::PumpSwap => ProtocolType::PumpSwap,
Protocol::Bonk => ProtocolType::Bonk,
Protocol::RaydiumCpmm => ProtocolType::RaydiumCpmm,
Protocol::RaydiumClmm => ProtocolType::RaydiumClmm,
Protocol::RaydiumAmmV4 => ProtocolType::RaydiumAmmV4,
Protocol::MeteoraDammV2 => ProtocolType::MeteoraDammV2,
Protocol::OrcaWhirlpool => ProtocolType::OrcaWhirlpool,
Protocol::MeteoraPools => ProtocolType::MeteoraPools,
Protocol::MeteoraDlmm => ProtocolType::MeteoraDlmm,
};
let mut full = Vec::with_capacity(instruction_discriminator.len() + instruction_data.len());
full.extend_from_slice(instruction_discriminator);
full.extend_from_slice(instruction_data);
match protocol {
Protocol::OrcaWhirlpool | Protocol::MeteoraPools | Protocol::MeteoraDlmm => {
sol_parser_forward::native::dispatch_instruction(
protocol.clone(),
instruction_discriminator,
instruction_data,
accounts,
&metadata,
)
}
Protocol::PumpFun => pumpfun::parse_pumpfun_instruction_data(
instruction_discriminator,
instruction_data,
accounts,
metadata,
),
Protocol::PumpSwap => pumpswap::parse_pumpswap_instruction_data(
instruction_discriminator,
instruction_data,
accounts,
metadata,
),
Protocol::Bonk => bonk::parse_bonk_instruction_data(
instruction_discriminator,
instruction_data,
accounts,
metadata,
),
Protocol::RaydiumCpmm => raydium_cpmm::parse_raydium_cpmm_instruction_data(
instruction_discriminator,
instruction_data,
accounts,
metadata,
),
Protocol::RaydiumClmm => raydium_clmm::parse_raydium_clmm_instruction_data(
instruction_discriminator,
instruction_data,
accounts,
metadata,
),
Protocol::RaydiumAmmV4 => raydium_amm_v4::parse_raydium_amm_v4_instruction_data(
instruction_discriminator,
instruction_data,
accounts,
metadata,
),
Protocol::MeteoraDammV2 => meteora_damm_v2::parse_meteora_damm_v2_instruction_data(
instruction_discriminator,
instruction_data,
accounts,
metadata,
),
}
let program_id = Self::get_program_id(protocol);
let pb = sol_parser_sdk::instr::parse_instruction_unified(
&full,
accounts,
metadata.signature,
metadata.slot,
metadata.tx_index.unwrap_or(0),
Some(metadata.block_time_ms.saturating_mul(1000)),
metadata.recv_us,
None,
&program_id,
)?;
let ts = block_timestamp_from_stream_meta(&metadata);
let ev = convert_parser_event(pb, Some(&ts), metadata.recv_us)?;
Some(fuse_streamer_ix_ctx(ev, &metadata))
}
/// 解析 inner instruction 事件（只解析，不合并）
///
/// # 参数
/// - `protocol`: 协议类型
/// - `inner_instruction_discriminator`: 内联指令判别器 (16 bytes)
/// - `inner_instruction_data`: 内联指令数据
/// - `metadata`: 事件元数据
///
/// # 返回
/// 解析成功返回 `Some(DexEvent)`，否则返回 `None`
#[inline]
pub fn dispatch_inner_instruction(
protocol: Protocol,
inner_instruction_discriminator: &[u8],
inner_instruction_data: &[u8],
mut metadata: EventMetadata,
metadata: EventMetadata,
) -> Option<DexEvent> {
// 根据协议类型设置 metadata.protocol
use crate::streaming::event_parser::common::ProtocolType;
metadata.protocol = match protocol {
Protocol::PumpFun => ProtocolType::PumpFun,
Protocol::PumpSwap => ProtocolType::PumpSwap,
Protocol::Bonk => ProtocolType::Bonk,
Protocol::RaydiumCpmm => ProtocolType::RaydiumCpmm,
Protocol::RaydiumClmm => ProtocolType::RaydiumClmm,
Protocol::RaydiumAmmV4 => ProtocolType::RaydiumAmmV4,
Protocol::MeteoraDammV2 => ProtocolType::MeteoraDammV2,
Protocol::OrcaWhirlpool => ProtocolType::OrcaWhirlpool,
Protocol::MeteoraPools => ProtocolType::MeteoraPools,
Protocol::MeteoraDlmm => ProtocolType::MeteoraDlmm,
};
let disc: [u8; 16] = inner_instruction_discriminator.try_into().ok()?;
let pm = pb_meta_from_streamer(&metadata);
match protocol {
Protocol::OrcaWhirlpool | Protocol::MeteoraPools | Protocol::MeteoraDlmm => {
sol_parser_forward::native::dispatch_inner_instruction(
protocol.clone(),
inner_instruction_discriminator,
inner_instruction_data,
&metadata,
)
let pb = match protocol {
Protocol::PumpFun => pump_inner::parse_pumpfun_inner_instruction(
&disc,
inner_instruction_data,
pm,
false,
),
Protocol::PumpSwap => {
pump_amm_inner::parse_pumpswap_inner_instruction(&disc, inner_instruction_data, pm)
}
Protocol::PumpFun => pumpfun::parse_pumpfun_inner_instruction_data(
inner_instruction_discriminator,
Protocol::PumpFees => all_inner::pump_fees::parse(&disc, inner_instruction_data, pm),
Protocol::Bonk | Protocol::RaydiumLaunchpad => {
all_inner::bonk::parse(&disc, inner_instruction_data, pm)
}
Protocol::RaydiumCpmm => {
all_inner::raydium_cpmm::parse(&disc, inner_instruction_data, pm)
}
Protocol::RaydiumClmm => raydium_clmm_inner::parse_raydium_clmm_inner_instruction(
&disc,
inner_instruction_data,
metadata,
),
Protocol::PumpSwap => pumpswap::parse_pumpswap_inner_instruction_data(
inner_instruction_discriminator,
inner_instruction_data,
metadata,
),
Protocol::Bonk => bonk::parse_bonk_inner_instruction_data(
inner_instruction_discriminator,
inner_instruction_data,
metadata,
),
Protocol::RaydiumCpmm => raydium_cpmm::parse_raydium_cpmm_inner_instruction_data(
inner_instruction_discriminator,
inner_instruction_data,
metadata,
),
Protocol::RaydiumClmm => raydium_clmm::parse_raydium_clmm_inner_instruction_data(
inner_instruction_discriminator,
inner_instruction_data,
metadata,
),
Protocol::RaydiumAmmV4 => raydium_amm_v4::parse_raydium_amm_v4_inner_instruction_data(
inner_instruction_discriminator,
inner_instruction_data,
metadata,
pm,
),
Protocol::RaydiumAmmV4 => {
all_inner::raydium_amm::parse(&disc, inner_instruction_data, pm)
}
Protocol::MeteoraDammV2 => {
meteora_damm_v2::parse_meteora_damm_v2_inner_instruction_data(
inner_instruction_discriminator,
inner_instruction_data,
metadata,
)
all_inner::meteora_damm::parse(&disc, inner_instruction_data, pm)
}
}
Protocol::OrcaWhirlpool => all_inner::orca::parse(&disc, inner_instruction_data, pm),
Protocol::MeteoraPools => {
all_inner::meteora_amm::parse(&disc, inner_instruction_data, pm)
}
Protocol::MeteoraDlmm => {
all_inner::meteora_dlmm::parse(&disc, inner_instruction_data, pm)
}
}?;
let ts = block_timestamp_from_stream_meta(&metadata);
let ev = convert_parser_event(pb, Some(&ts), metadata.recv_us)?;
Some(fuse_streamer_ix_ctx(ev, &metadata))
}
/// 通过 program_id 匹配协议类型
#[inline]
pub fn match_protocol_by_program_id(program_id: &Pubkey) -> Option<Protocol> {
if program_id == &pumpfun::PUMPFUN_PROGRAM_ID {
if program_id == &program_ids::PUMPFUN_PROGRAM_ID {
Some(Protocol::PumpFun)
} else if program_id == &pumpswap::PUMPSWAP_PROGRAM_ID {
} else if program_id == &program_ids::PUMP_FEES_PROGRAM_ID {
Some(Protocol::PumpFees)
} else if program_id == &program_ids::PUMPSWAP_PROGRAM_ID {
Some(Protocol::PumpSwap)
} else if program_id == &bonk::BONK_PROGRAM_ID {
} else if program_id == &program_ids::BONK_PROGRAM_ID {
Some(Protocol::Bonk)
} else if program_id == &raydium_cpmm::RAYDIUM_CPMM_PROGRAM_ID {
} else if program_id == &program_ids::RAYDIUM_CPMM_PROGRAM_ID {
Some(Protocol::RaydiumCpmm)
} else if program_id == &raydium_clmm::RAYDIUM_CLMM_PROGRAM_ID {
} else if program_id == &program_ids::RAYDIUM_CLMM_PROGRAM_ID {
Some(Protocol::RaydiumClmm)
} else if program_id == &raydium_amm_v4::RAYDIUM_AMM_V4_PROGRAM_ID {
} else if program_id == &program_ids::RAYDIUM_AMM_V4_PROGRAM_ID {
Some(Protocol::RaydiumAmmV4)
} else if program_id == &meteora_damm_v2::METEORA_DAMM_V2_PROGRAM_ID {
} else if program_id == &program_ids::METEORA_DAMM_V2_PROGRAM_ID {
Some(Protocol::MeteoraDammV2)
} else if program_id == &sol_parser_forward::ORCA_WHIRLPOOL_PROGRAM_ID {
} else if program_id == &program_ids::ORCA_WHIRLPOOL_PROGRAM_ID {
Some(Protocol::OrcaWhirlpool)
} else if program_id == &sol_parser_forward::METEORA_POOLS_PROGRAM_ID {
} else if program_id == &program_ids::METEORA_POOLS_PROGRAM_ID {
Some(Protocol::MeteoraPools)
} else if program_id == &sol_parser_forward::METEORA_DLMM_PROGRAM_ID {
} else if program_id == &program_ids::METEORA_DLMM_PROGRAM_ID {
Some(Protocol::MeteoraDlmm)
} else {
None
}
}
/// 检查是否为 Compute Budget Program
#[inline]
pub fn is_compute_budget_program(program_id: &Pubkey) -> bool {
program_id == &COMPUTE_BUDGET_PROGRAM_ID
program_id == &solana_sdk::pubkey!("ComputeBudget111111111111111111111111111111")
}
/// 解析 Compute Budget 指令
///
/// # 参数
/// - `instruction_data`: 指令数据
/// - `metadata`: 事件元数据
///
/// # 返回
/// 解析成功返回 `Some(DexEvent)`，否则返回 `None`
#[inline]
pub fn dispatch_compute_budget_instruction(
instruction_data: &[u8],
metadata: EventMetadata,
_instruction_data: &[u8],
_metadata: EventMetadata,
) -> Option<DexEvent> {
CommonEventParser::parse_compute_budget_instruction(instruction_data, metadata)
None
}
/// 获取指定协议的 program_id
#[inline]
pub fn get_program_id(protocol: Protocol) -> Pubkey {
match protocol {
Protocol::PumpFun => pumpfun::PUMPFUN_PROGRAM_ID,
Protocol::PumpSwap => pumpswap::PUMPSWAP_PROGRAM_ID,
Protocol::Bonk => bonk::BONK_PROGRAM_ID,
Protocol::RaydiumCpmm => raydium_cpmm::RAYDIUM_CPMM_PROGRAM_ID,
Protocol::RaydiumClmm => raydium_clmm::RAYDIUM_CLMM_PROGRAM_ID,
Protocol::RaydiumAmmV4 => raydium_amm_v4::RAYDIUM_AMM_V4_PROGRAM_ID,
Protocol::MeteoraDammV2 => meteora_damm_v2::METEORA_DAMM_V2_PROGRAM_ID,
Protocol::OrcaWhirlpool => sol_parser_forward::ORCA_WHIRLPOOL_PROGRAM_ID,
Protocol::MeteoraPools => sol_parser_forward::METEORA_POOLS_PROGRAM_ID,
Protocol::MeteoraDlmm => sol_parser_forward::METEORA_DLMM_PROGRAM_ID,
Protocol::PumpFun => program_ids::PUMPFUN_PROGRAM_ID,
Protocol::PumpFees => program_ids::PUMP_FEES_PROGRAM_ID,
Protocol::PumpSwap => program_ids::PUMPSWAP_PROGRAM_ID,
Protocol::Bonk | Protocol::RaydiumLaunchpad => program_ids::BONK_PROGRAM_ID,
Protocol::RaydiumCpmm => program_ids::RAYDIUM_CPMM_PROGRAM_ID,
Protocol::RaydiumClmm => program_ids::RAYDIUM_CLMM_PROGRAM_ID,
Protocol::RaydiumAmmV4 => program_ids::RAYDIUM_AMM_V4_PROGRAM_ID,
Protocol::MeteoraDammV2 => program_ids::METEORA_DAMM_V2_PROGRAM_ID,
Protocol::OrcaWhirlpool => program_ids::ORCA_WHIRLPOOL_PROGRAM_ID,
Protocol::MeteoraPools => program_ids::METEORA_POOLS_PROGRAM_ID,
Protocol::MeteoraDlmm => program_ids::METEORA_DLMM_PROGRAM_ID,
}
}
/// 批量获取 program_ids
pub fn get_program_ids(protocols: &[Protocol]) -> Vec<Pubkey> {
protocols.iter().map(|p| Self::get_program_id(p.clone())).collect()
let mut ids = Vec::with_capacity(protocols.len());
for protocol in protocols {
let id = Self::get_program_id(protocol.clone());
if !ids.contains(&id) {
ids.push(id);
}
}
ids
}
/// 解析账户数据
///
/// 根据账户的 discriminator 路由到对应协议的账户解析函数
///
/// # 参数
/// - `protocol`: 协议类型
/// - `discriminator`: 账户判别器
/// - `account`: 账户信息
/// - `metadata`: 事件元数据
///
/// # 返回
/// 解析成功返回 `Some(DexEvent)`，否则返回 `None`
pub fn dispatch_account(
protocol: Protocol,
discriminator: &[u8],
_discriminator: &[u8],
account: &crate::streaming::grpc::AccountPretty,
mut metadata: crate::streaming::event_parser::common::EventMetadata,
_metadata: EventMetadata,
) -> Option<DexEvent> {
// 根据协议类型设置 metadata.protocol
use crate::streaming::event_parser::common::ProtocolType;
metadata.protocol = match protocol {
Protocol::PumpFun => ProtocolType::PumpFun,
Protocol::PumpSwap => ProtocolType::PumpSwap,
Protocol::Bonk => ProtocolType::Bonk,
Protocol::RaydiumCpmm => ProtocolType::RaydiumCpmm,
Protocol::RaydiumClmm => ProtocolType::RaydiumClmm,
Protocol::RaydiumAmmV4 => ProtocolType::RaydiumAmmV4,
Protocol::MeteoraDammV2 => ProtocolType::MeteoraDammV2,
Protocol::OrcaWhirlpool => ProtocolType::OrcaWhirlpool,
Protocol::MeteoraPools => ProtocolType::MeteoraPools,
Protocol::MeteoraDlmm => ProtocolType::MeteoraDlmm,
};
parse_sdk_account_event(account, &[protocol], None)
}
}
match protocol {
Protocol::OrcaWhirlpool | Protocol::MeteoraPools | Protocol::MeteoraDlmm => None,
Protocol::PumpFun => {
pumpfun::parse_pumpfun_account_data(discriminator, account, metadata)
}
Protocol::PumpSwap => {
pumpswap::parse_pumpswap_account_data(discriminator, account, metadata)
}
Protocol::Bonk => bonk::parse_bonk_account_data(discriminator, account, metadata),
Protocol::RaydiumCpmm => {
raydium_cpmm::parse_raydium_cpmm_account_data(discriminator, account, metadata)
}
Protocol::RaydiumClmm => {
raydium_clmm::parse_raydium_clmm_account_data(discriminator, account, metadata)
}
Protocol::RaydiumAmmV4 => {
raydium_amm_v4::parse_raydium_amm_v4_account_data(discriminator, account, metadata)
}
Protocol::MeteoraDammV2 => {
// Meteora DAMM 目前不需要解析账户数据，返回 None
None
}
}
#[inline]
fn pb_meta_from_streamer(sm: &EventMetadata) -> PbEventMetadata {
PbEventMetadata {
signature: sm.signature,
slot: sm.slot,
tx_index: sm.tx_index.unwrap_or(0),
block_time_us: sm.block_time_ms.saturating_mul(1000),
grpc_recv_us: sm.recv_us,
recent_blockhash: sm.recent_blockhash.clone(),
}
}
src/streaming/event_parser/core/event_parser/compiled_path/compiled_instruction.rs
-195
View File
@@ -1,195 +0,0 @@
//! Single Solana [`CompiledInstruction`] parsing with local inner merge and swap enrichment.
use crate::streaming::event_parser::{
common::{
filter::{passes_event_type_filter, EventTypeFilter},
high_performance_clock::elapsed_micros_since,
parse_swap_data_from_next_instructions, EventMetadata,
},
core::{dispatcher::EventDispatcher, merger_event::merge},
protocols::{
raydium_amm_v4::parser::RAYDIUM_AMM_V4_PROGRAM_ID,
sol_parser_forward::METEORA_DLMM_PROGRAM_ID,
},
DexEvent, Protocol,
};
use prost_types::Timestamp;
use solana_sdk::{
message::compiled_instruction::CompiledInstruction, pubkey::Pubkey, signature::Signature,
};
use solana_transaction_status::InnerInstructions;
use std::sync::Arc;
pub(super) fn parse_events_from_instruction(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
instruction: &CompiledInstruction,
accounts: &[Pubkey],
signature: Signature,
slot: u64,
block_time: Option<Timestamp>,
recv_us: i64,
outer_index: i64,
inner_index: Option<i64>,
bot_wallet: Option<Pubkey>,
tx_index: Option<u64>,
recent_blockhash: Option<&str>,
inner_instructions: Option<&InnerInstructions>,
callback: Arc<dyn Fn(DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
// Bounds check before reading the program id index.
let program_id_index = instruction.program_id_index as usize;
if program_id_index >= accounts.len() {
return Ok(());
}
let program_id = accounts[program_id_index];
if !super::super::helpers::should_handle(protocols, event_type_filter, &program_id) {
return Ok(());
}
let is_cu_program = EventDispatcher::is_compute_budget_program(&program_id);
let disc_len = match program_id {
RAYDIUM_AMM_V4_PROGRAM_ID | METEORA_DLMM_PROGRAM_ID => 1,
_ => 8,
};
// Non-ComputeBudget instructions need at least a discriminator.
if !is_cu_program && instruction.data.len() < disc_len {
return Ok(());
}
// Build streamer metadata.
let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 });
let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000;
let metadata = EventMetadata::new(
signature,
slot,
timestamp.seconds,
block_time_ms,
Default::default(), // protocol will be set by dispatcher
Default::default(), // event_type will be set by dispatcher
program_id,
outer_index,
inner_index,
recv_us,
tx_index,
recent_blockhash.map(|s| s.to_string()),
);
if is_cu_program {
if let Some(event) = EventDispatcher::dispatch_compute_budget_instruction(
&instruction.data,
metadata.clone(),
) {
if passes_event_type_filter(event_type_filter, &event) {
callback(event);
}
}
return Ok(());
}
// Match the parser protocol.
let protocol = match EventDispatcher::match_protocol_by_program_id(&program_id) {
Some(p) => p,
None => return Ok(()),
};
// Split discriminator and instruction payload.
let instruction_discriminator = &instruction.data[..disc_len];
let instruction_data = &instruction.data[disc_len..];
// Build the account pubkey list for this instruction.
let account_pubkeys: Vec<Pubkey> = instruction
.accounts
.iter()
.filter_map(|&idx| accounts.get(idx as usize).copied())
.collect();
// Parse the instruction event.
let mut event = match EventDispatcher::dispatch_instruction(
protocol.clone(),
instruction_discriminator,
instruction_data,
&account_pubkeys,
metadata.clone(),
) {
Some(e) => e,
None => return Ok(()),
};
// Find the next CPI log for merge.
let mut inner_instruction_event: Option<DexEvent> = None;
if let Some(inner_instructions_ref) = inner_instructions {
let raw = inner_index.unwrap_or(-1);
let current_inner_idx = raw.clamp(i32::MIN as i64, i32::MAX as i64) as i32;
// Parse the inner event and swap data in parallel on the compiled local path.
let (inner_event_result, swap_data_result) = std::thread::scope(|s| {
let inner_event_handle = s.spawn(|| {
for (idx, inner_instruction) in
inner_instructions_ref.instructions.iter().enumerate()
{
// Only inspect CPI logs after the current inner instruction.
if (idx as i32) <= current_inner_idx {
continue;
}
let inner_data = &inner_instruction.instruction.data;
// Inner CPI logs use a 16-byte discriminator.
if inner_data.len() < 16 {
continue;
}
let inner_discriminator = &inner_data[..16];
let inner_instruction_data = &inner_data[16..];
if let Some(inner_event) = EventDispatcher::dispatch_inner_instruction(
protocol.clone(),
inner_discriminator,
inner_instruction_data,
metadata.clone(),
) {
return Some(inner_event);
}
}
None
});
let swap_data_handle = s.spawn(|| {
if event.metadata().swap_data.is_none() {
parse_swap_data_from_next_instructions(
&event,
inner_instructions_ref,
current_inner_idx,
accounts,
)
} else {
None
}
});
// Wait for both local tasks.
(inner_event_handle.join().unwrap(), swap_data_handle.join().unwrap())
});
inner_instruction_event = inner_event_result;
if let Some(swap_data) = swap_data_result {
event.metadata_mut().set_swap_data(swap_data);
}
}
// PumpFun MIGRATE emits instruction-only data when no CPI log exists.
// Merge CPI details into the outer event.
if let Some(inner_instruction_event) = inner_instruction_event {
merge(&mut event, inner_instruction_event);
}
// Stamp handling latency using the high-performance clock.
event.metadata_mut().handle_us = elapsed_micros_since(recv_us);
event = super::super::helpers::process_event(event, bot_wallet);
if passes_event_type_filter(event_type_filter, &event) {
callback(event);
}
Ok(())
}
src/streaming/event_parser/core/event_parser/compiled_path/compiled_transaction.rs
-87
View File
@@ -1,87 +0,0 @@
//! Sequential top-level and inner ix traversal for [`VersionedTransaction`].
use crate::streaming::event_parser::{common::filter::EventTypeFilter, DexEvent, Protocol};
use prost_types::Timestamp;
use solana_sdk::{pubkey::Pubkey, signature::Signature, transaction::VersionedTransaction};
use solana_transaction_status::InnerInstructions;
use std::sync::Arc;
pub(crate) async fn parse_instruction_events_from_versioned_transaction(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
transaction: &VersionedTransaction,
signature: Signature,
slot: Option<u64>,
block_time: Option<Timestamp>,
recv_us: i64,
accounts: &[Pubkey],
inner_instructions: &[InnerInstructions],
bot_wallet: Option<Pubkey>,
tx_index: Option<u64>,
callback: Arc<dyn Fn(DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
let compiled_instructions = transaction.message.instructions();
let recent_blockhash = Some(transaction.message.recent_blockhash().to_string());
let mut accounts: Vec<Pubkey> = accounts.to_vec();
let has_program = accounts
.iter()
.any(|account| super::super::helpers::should_handle(protocols, event_type_filter, account));
if has_program {
// Parse each instruction in order.
for (index, instruction) in compiled_instructions.iter().enumerate() {
if let Some(program_id) = accounts.get(instruction.program_id_index as usize) {
let program_id = *program_id;
let inner_instructions = inner_instructions
.iter()
.find(|inner_instruction| inner_instruction.index == index as u8);
if super::super::helpers::should_handle(protocols, event_type_filter, &program_id) {
let max_idx = instruction.accounts.iter().max().unwrap_or(&0);
if *max_idx as usize >= accounts.len() {
accounts.resize(*max_idx as usize + 1, Pubkey::default());
}
super::compiled_instruction::parse_events_from_instruction(
protocols,
event_type_filter,
instruction,
&accounts,
signature,
slot.unwrap_or(0),
block_time,
recv_us,
index as i64,
None,
bot_wallet,
tx_index,
recent_blockhash.as_deref(),
inner_instructions,
callback.clone(),
)?;
}
// Immediately process inner instructions for correct ordering
if let Some(inner_instructions) = inner_instructions {
for (inner_index, inner_instruction) in
inner_instructions.instructions.iter().enumerate()
{
super::compiled_instruction::parse_events_from_instruction(
protocols,
event_type_filter,
&inner_instruction.instruction,
&accounts,
signature,
slot.unwrap_or(0),
block_time,
recv_us,
index as i64,
Some(inner_index as i64),
bot_wallet,
tx_index,
recent_blockhash.as_deref(),
Some(&inner_instructions),
callback.clone(),
)?;
}
}
}
}
}
Ok(())
}
src/streaming/event_parser/core/event_parser/compiled_path/mod.rs
-11
View File
@@ -1,11 +0,0 @@
//! Standard [`VersionedTransaction`] / [`CompiledInstruction`] path for RPC and replay.
//!
//! | Module | Responsibility |
//! |--------|------|
//! | [`compiled_transaction`] | top-level ix loop |
//! | [`compiled_instruction`] | single Solana `CompiledInstruction` |
mod compiled_instruction;
mod compiled_transaction;
pub(super) use compiled_transaction::parse_instruction_events_from_versioned_transaction;
src/streaming/event_parser/core/event_parser/grpc_path/grpc_instruction.rs
-171
View File
@@ -1,171 +0,0 @@
//! Single Yellowstone [`CompiledInstruction`] parsing: dispatch, inner merge, swap enrichment.
use crate::streaming::event_parser::{
common::{
filter::{passes_event_type_filter, EventTypeFilter},
high_performance_clock::elapsed_micros_since,
parse_swap_data_from_next_grpc_instructions, EventMetadata,
},
core::{dispatcher::EventDispatcher, merger_event::merge},
protocols::{
raydium_amm_v4::parser::RAYDIUM_AMM_V4_PROGRAM_ID,
sol_parser_forward::METEORA_DLMM_PROGRAM_ID,
},
DexEvent, Protocol,
};
use prost_types::Timestamp;
use solana_sdk::{pubkey::Pubkey, signature::Signature};
use std::sync::Arc;
pub(super) fn parse_events_from_grpc_instruction(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
instruction: &yellowstone_grpc_proto::prelude::CompiledInstruction,
accounts: &[Pubkey],
signature: Signature,
slot: u64,
block_time: Option<Timestamp>,
recv_us: i64,
outer_index: i64,
inner_index: Option<i64>,
bot_wallet: Option<Pubkey>,
tx_index: Option<u64>,
recent_blockhash: Option<&str>,
inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>,
callback: Arc<dyn Fn(DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
// Bounds check before reading the program id index.
let program_id_index = instruction.program_id_index as usize;
if program_id_index >= accounts.len() {
return Ok(());
}
let program_id = accounts[program_id_index];
if !super::super::helpers::should_handle(protocols, event_type_filter, &program_id) {
return Ok(());
}
let is_cu_program = EventDispatcher::is_compute_budget_program(&program_id);
let disc_len = match program_id {
RAYDIUM_AMM_V4_PROGRAM_ID | METEORA_DLMM_PROGRAM_ID => 1,
_ => 8,
};
// Non-ComputeBudget instructions need at least a discriminator.
if !is_cu_program && instruction.data.len() < disc_len {
return Ok(());
}
// Build streamer metadata.
let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 });
let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000;
let metadata = EventMetadata::new(
signature,
slot,
timestamp.seconds,
block_time_ms,
Default::default(), // protocol will be set by dispatcher
Default::default(), // event_type will be set by dispatcher
program_id,
outer_index,
inner_index,
recv_us,
tx_index,
recent_blockhash.map(|s| s.to_string()),
);
if is_cu_program {
if let Some(event) = EventDispatcher::dispatch_compute_budget_instruction(
&instruction.data,
metadata.clone(),
) {
if passes_event_type_filter(event_type_filter, &event) {
callback(event);
}
}
return Ok(());
}
// Match the parser protocol.
let protocol = match EventDispatcher::match_protocol_by_program_id(&program_id) {
Some(p) => p,
None => return Ok(()),
};
// Split discriminator and instruction payload.
let instruction_discriminator = &instruction.data[..disc_len];
let instruction_data = &instruction.data[disc_len..];
// Build the account pubkey list for this instruction.
let account_pubkeys: Vec<Pubkey> = instruction
.accounts
.iter()
.filter_map(|&idx| accounts.get(idx as usize).copied())
.collect();
// Parse the instruction event.
let mut event = match EventDispatcher::dispatch_instruction(
protocol.clone(),
instruction_discriminator,
instruction_data,
&account_pubkeys,
metadata.clone(),
) {
Some(e) => e,
None => return Ok(()),
};
// Find the next CPI log for merge. The gRPC hot path stays sequential to avoid
// thread::scope spawn/join overhead.
let mut inner_instruction_event: Option<DexEvent> = None;
if let Some(inner_instructions_ref) = inner_instructions {
let raw = inner_index.unwrap_or(-1);
let current_inner_idx = raw.clamp(i32::MIN as i64, i32::MAX as i64) as i32;
for (idx, inner_instruction) in inner_instructions_ref.instructions.iter().enumerate() {
if (idx as i32) <= current_inner_idx {
continue;
}
let inner_data = &inner_instruction.data;
if inner_data.len() < 16 {
continue;
}
let inner_discriminator = &inner_data[..16];
let inner_instruction_data = &inner_data[16..];
if let Some(inner_event) = EventDispatcher::dispatch_inner_instruction(
protocol.clone(),
inner_discriminator,
inner_instruction_data,
metadata.clone(),
) {
inner_instruction_event = Some(inner_event);
break;
}
}
if event.metadata().swap_data.is_none() {
if let Some(swap_data) = parse_swap_data_from_next_grpc_instructions(
&event,
inner_instructions_ref,
current_inner_idx,
accounts,
) {
event.metadata_mut().set_swap_data(swap_data);
}
}
}
// PumpFun MIGRATE emits instruction-only data when no CPI log exists.
// Merge CPI details into the outer event.
if let Some(inner_instruction_event) = inner_instruction_event {
merge(&mut event, inner_instruction_event);
}
// Stamp handling latency using the high-performance clock.
event.metadata_mut().handle_us = elapsed_micros_since(recv_us);
event = super::super::helpers::process_event(event, bot_wallet);
if passes_event_type_filter(event_type_filter, &event) {
callback(event);
}
Ok(())
}
src/streaming/event_parser/core/event_parser/grpc_path/grpc_ix_mode.rs
-9
View File
@@ -1,9 +0,0 @@
//! Top-level ix parsing strategy for the gRPC path.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum GrpcIxParseMode {
/// Parse all subscribed instructions, used when transaction meta is missing.
Full,
/// Parse only ComputeBudget locally; DEX events come from sol-parser-sdk.
ComputeBudgetOnly,
}
src/streaming/event_parser/core/event_parser/grpc_path/grpc_transaction.rs
-265
View File
@@ -1,265 +0,0 @@
//! Yellowstone transaction parsing: SDK-first DEX parsing plus optional local ix fallback.
//!
//! When `transaction.meta` is missing, the SDK low-latency parser cannot see logs / complete inner
//! instruction context and may return no events. In that case streamer uses the local full ix path.
//!
//! When meta exists, DEX events come from `sol-parser-sdk`; the local second pass is limited to
//! ComputeBudget events when the user asked for them.
use crate::streaming::event_parser::{
common::{
filter::{
build_sdk_parse_event_filter, filter_includes_compute_budget_types, EventTypeFilter,
},
high_performance_clock::elapsed_micros_since,
},
core::dispatcher::EventDispatcher,
DexEvent, Protocol,
};
use prost_types::Timestamp;
use sol_parser_sdk::grpc::parse_subscribe_update_transaction_low_latency;
use solana_sdk::{pubkey::Pubkey, signature::Signature};
use std::sync::Arc;
use yellowstone_grpc_proto::geyser::{SubscribeUpdateTransaction, SubscribeUpdateTransactionInfo};
use super::grpc_ix_mode::GrpcIxParseMode;
pub(crate) async fn parse_grpc_transaction(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
mut grpc_tx: SubscribeUpdateTransactionInfo,
signature: Signature,
slot: Option<u64>,
block_time: Option<Timestamp>,
recv_us: i64,
bot_wallet: Option<Pubkey>,
tx_index: Option<u64>,
callback: Arc<dyn Fn(DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
let slot_u = slot.unwrap_or(0);
let block_us_micro = block_time.map(|t| t.seconds * 1_000_000 + t.nanos as i64 / 1_000);
if grpc_tx.transaction.as_ref().and_then(|tx| tx.message.as_ref()).is_none() {
return Ok(());
}
let use_sol_parser_sdk = grpc_tx.meta.is_some();
let skip_ix_pass =
use_sol_parser_sdk && !filter_includes_compute_budget_types(event_type_filter);
if use_sol_parser_sdk {
let mut update = SubscribeUpdateTransaction {
slot: slot_u,
transaction: Some(grpc_tx),
..Default::default()
};
let sdk_parse_filter = build_sdk_parse_event_filter(event_type_filter);
let pb_events = parse_subscribe_update_transaction_low_latency(
&update,
recv_us,
block_us_micro,
sdk_parse_filter.as_ref(),
);
let adapted = crate::streaming::parser_sdk_bridge::adapt_parser_events_list(
pb_events,
block_time.as_ref(),
recv_us,
protocols,
event_type_filter,
);
for mut ev in adapted {
ev.metadata_mut().handle_us = elapsed_micros_since(recv_us);
ev = super::super::helpers::process_event(ev, bot_wallet);
callback(ev);
}
if skip_ix_pass {
return Ok(());
}
let Some(tx) = update.transaction.take() else {
return Ok(());
};
grpc_tx = tx;
}
let Some(transition) = grpc_tx.transaction.as_ref() else {
return Ok(());
};
let Some(message) = transition.message.as_ref() else {
return Ok(());
};
let ix_mode =
if use_sol_parser_sdk { GrpcIxParseMode::ComputeBudgetOnly } else { GrpcIxParseMode::Full };
let accounts = build_account_keys(message, grpc_tx.meta.as_ref());
let inner_instructions =
grpc_tx.meta.as_ref().map(|meta| meta.inner_instructions.as_slice()).unwrap_or_default();
let recent_blockhash = if message.recent_blockhash.len() == 32 {
Some(solana_sdk::bs58::encode(&message.recent_blockhash).into_string())
} else {
None
};
parse_instruction_events_from_grpc_transaction(
protocols,
event_type_filter,
ix_mode,
&message.instructions,
signature,
slot,
block_time,
recv_us,
&accounts,
inner_instructions,
bot_wallet,
tx_index,
recent_blockhash,
callback,
)
.await?;
Ok(())
}
fn build_account_keys(
message: &yellowstone_grpc_proto::prelude::Message,
meta: Option<&yellowstone_grpc_proto::prelude::TransactionStatusMeta>,
) -> Vec<Pubkey> {
let loaded_len = meta
.map(|m| m.loaded_writable_addresses.len() + m.loaded_readonly_addresses.len())
.unwrap_or(0);
let mut accounts = Vec::with_capacity(message.account_keys.len() + loaded_len);
for account in &message.account_keys {
push_account_key(&mut accounts, account);
}
if let Some(meta) = meta {
for account in
meta.loaded_writable_addresses.iter().chain(meta.loaded_readonly_addresses.iter())
{
push_account_key(&mut accounts, account);
}
}
accounts
}
#[inline]
fn push_account_key(accounts: &mut Vec<Pubkey>, account: &[u8]) {
let pubkey = if account.len() == 32 {
Pubkey::try_from(account).unwrap_or_default()
} else {
Pubkey::default()
};
accounts.push(pubkey);
}
pub(super) async fn parse_instruction_events_from_grpc_transaction(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
ix_mode: GrpcIxParseMode,
compiled_instructions: &[yellowstone_grpc_proto::prelude::CompiledInstruction],
signature: Signature,
slot: Option<u64>,
block_time: Option<Timestamp>,
recv_us: i64,
accounts: &[Pubkey],
inner_instructions: &[yellowstone_grpc_proto::solana::storage::confirmed_block::InnerInstructions],
bot_wallet: Option<Pubkey>,
tx_index: Option<u64>,
recent_blockhash: Option<String>,
callback: Arc<dyn Fn(DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
let mut accounts = accounts.to_vec();
let has_program = match ix_mode {
GrpcIxParseMode::Full => accounts.iter().any(|account| {
super::super::helpers::should_handle(protocols, event_type_filter, account)
}),
GrpcIxParseMode::ComputeBudgetOnly => compiled_instructions.iter().any(|ix| {
accounts
.get(ix.program_id_index as usize)
.map(EventDispatcher::is_compute_budget_program)
.unwrap_or(false)
}),
};
if has_program {
// Parse each instruction in order.
for (index, instruction) in compiled_instructions.iter().enumerate() {
if let Some(program_id) = accounts.get(instruction.program_id_index as usize) {
let program_id = *program_id;
let inner_instructions_ref = inner_instructions
.iter()
.find(|inner_instruction| inner_instruction.index == index as u32);
let max_idx = instruction.accounts.iter().max().unwrap_or(&0);
if *max_idx as usize >= accounts.len() {
accounts.resize(*max_idx as usize + 1, Pubkey::default());
}
let handle_outer = match ix_mode {
GrpcIxParseMode::Full => super::super::helpers::should_handle(
protocols,
event_type_filter,
&program_id,
),
GrpcIxParseMode::ComputeBudgetOnly => {
EventDispatcher::is_compute_budget_program(&program_id)
}
};
if handle_outer {
super::grpc_instruction::parse_events_from_grpc_instruction(
protocols,
event_type_filter,
instruction,
&accounts,
signature,
slot.unwrap_or(0),
block_time,
recv_us,
index as i64,
None,
bot_wallet,
tx_index,
recent_blockhash.as_deref(),
inner_instructions_ref,
callback.clone(),
)?;
}
if ix_mode == GrpcIxParseMode::Full {
if let Some(inner_instructions) = inner_instructions_ref {
for (inner_index, inner_instruction) in
inner_instructions.instructions.iter().enumerate()
{
let inner_accounts = &inner_instruction.accounts;
let data = &inner_instruction.data;
let instruction =
yellowstone_grpc_proto::prelude::CompiledInstruction {
program_id_index: inner_instruction.program_id_index,
accounts: inner_accounts.to_vec(),
data: data.to_vec(),
};
super::grpc_instruction::parse_events_from_grpc_instruction(
protocols,
event_type_filter,
&instruction,
&accounts,
signature,
slot.unwrap_or(0),
block_time,
recv_us,
inner_instructions.index as i64,
Some(inner_index as i64),
bot_wallet,
tx_index,
recent_blockhash.as_deref(),
Some(inner_instructions),
callback.clone(),
)?;
}
}
}
}
}
}
Ok(())
}
src/streaming/event_parser/core/event_parser/grpc_path/mod.rs
-13
View File
@@ -1,13 +0,0 @@
//! Yellowstone gRPC path for `SubscribeUpdateTransactionInfo` and SDK aggregate parsing.
//!
//! | Module | Responsibility |
//! |--------|------|
//! | [`grpc_ix_mode`] | `GrpcIxParseMode` |
//! | [`grpc_transaction`] | whole subscription message and top-level ix loop |
//! | [`grpc_instruction`] | single Yellowstone `CompiledInstruction` |
mod grpc_instruction;
mod grpc_ix_mode;
mod grpc_transaction;
pub(super) use grpc_transaction::parse_grpc_transaction;
src/streaming/event_parser/core/event_parser/helpers.rs
+2 -19
View File
@@ -1,30 +1,13 @@
//! Protocol filtering and event enrichment for PumpFun / PumpSwap / Bonk / bot flags.
//! Event enrichment for PumpFun / PumpSwap / Bonk / bot flags.
use crate::streaming::event_parser::{
common::filter::{filter_includes_compute_budget_types, EventTypeFilter},
core::dispatcher::EventDispatcher,
core::global_state::{
add_bonk_dev_address, add_dev_address, is_bonk_dev_address_in_signature,
is_dev_address_in_signature,
},
DexEvent, Protocol,
DexEvent,
};
use solana_sdk::pubkey::Pubkey;
pub(super) fn should_handle(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
program_id: &Pubkey,
) -> bool {
if EventDispatcher::is_compute_budget_program(program_id) {
return filter_includes_compute_budget_types(event_type_filter);
}
if let Some(protocol) = EventDispatcher::match_protocol_by_program_id(program_id) {
protocols.contains(&protocol)
} else {
false
}
}
// ================================================================================================
// Event Post-Processing
// ================================================================================================
src/streaming/event_parser/core/event_parser/mod.rs
+76 -36
View File
@@ -1,13 +1,9 @@
//! Transaction parser entry point with separate gRPC and standard ix paths.
//! Transaction parser entry point.
//!
//! | Module | Path |
//! |--------|------|
//! | [`grpc_path`] | Yellowstone gRPC |
//! | [`compiled_path`] | standard transaction / RPC replay |
//! | [`helpers`] | `should_handle`、`process_event` |
//! gRPC and ShredStream parsing are delegated to `sol-parser-sdk`; this layer
//! only adapts SDK events into streamer event structs and applies streamer
//! metadata enrichment.
mod compiled_path;
mod grpc_path;
pub(crate) mod helpers;
pub struct EventParser;
@@ -16,7 +12,7 @@ impl EventParser {
pub async fn parse_grpc_transaction(
protocols: &[crate::streaming::event_parser::Protocol],
event_type_filter: Option<&crate::streaming::event_parser::common::filter::EventTypeFilter>,
grpc_tx: yellowstone_grpc_proto::geyser::SubscribeUpdateTransactionInfo,
mut grpc_tx: yellowstone_grpc_proto::geyser::SubscribeUpdateTransactionInfo,
signature: solana_sdk::signature::Signature,
slot: Option<u64>,
block_time: Option<prost_types::Timestamp>,
@@ -25,19 +21,46 @@ impl EventParser {
tx_index: Option<u64>,
callback: std::sync::Arc<dyn Fn(crate::streaming::event_parser::DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
grpc_path::parse_grpc_transaction(
protocols,
event_type_filter,
grpc_tx,
signature,
slot,
block_time,
if grpc_tx.signature.is_empty() {
grpc_tx.signature = signature.as_ref().to_vec();
}
if grpc_tx.index == 0 {
grpc_tx.index = tx_index.unwrap_or(0);
}
let block_us = block_time.map(|t| t.seconds * 1_000_000 + t.nanos as i64 / 1_000);
let update = yellowstone_grpc_proto::geyser::SubscribeUpdateTransaction {
slot: slot.unwrap_or(0),
transaction: Some(grpc_tx),
..Default::default()
};
let sdk_parse_filter =
crate::streaming::event_parser::common::filter::build_sdk_parse_event_filter(
event_type_filter,
);
let sdk_events = sol_parser_sdk::grpc::parse_subscribe_update_transaction_low_latency(
&update,
recv_us,
bot_wallet,
tx_index,
callback,
)
.await
block_us,
sdk_parse_filter.as_ref(),
);
for sdk_event in sdk_events {
if let Some(mut event) = crate::streaming::parser_sdk_bridge::adapt_parser_event(
sdk_event,
block_time.as_ref(),
recv_us,
protocols,
event_type_filter,
) {
event.metadata_mut().handle_us =
crate::streaming::event_parser::common::high_performance_clock::elapsed_micros_since(
recv_us,
);
event = helpers::process_event(event, bot_wallet);
callback(event);
}
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
@@ -47,28 +70,45 @@ impl EventParser {
transaction: &solana_sdk::transaction::VersionedTransaction,
signature: solana_sdk::signature::Signature,
slot: Option<u64>,
block_time: Option<prost_types::Timestamp>,
_block_time: Option<prost_types::Timestamp>,
recv_us: i64,
accounts: &[solana_sdk::pubkey::Pubkey],
inner_instructions: &[solana_transaction_status::InnerInstructions],
_accounts: &[solana_sdk::pubkey::Pubkey],
_inner_instructions: &[solana_transaction_status::InnerInstructions],
bot_wallet: Option<solana_sdk::pubkey::Pubkey>,
tx_index: Option<u64>,
callback: std::sync::Arc<dyn Fn(crate::streaming::event_parser::DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
compiled_path::parse_instruction_events_from_versioned_transaction(
protocols,
event_type_filter,
let sdk_parse_filter =
crate::streaming::event_parser::common::filter::build_sdk_shred_parse_event_filter(
protocols,
event_type_filter,
);
let mut sdk_events = Vec::with_capacity(4);
sol_parser_sdk::shredstream::parse_transaction_dex_events_with_filter(
transaction,
signature,
slot,
block_time,
slot.unwrap_or(0),
tx_index.unwrap_or(0),
recv_us,
accounts,
inner_instructions,
bot_wallet,
tx_index,
callback,
)
.await
sdk_parse_filter.as_ref(),
&mut sdk_events,
);
for sdk_event in sdk_events {
if let Some(mut event) = crate::streaming::parser_sdk_bridge::adapt_parser_event(
sdk_event,
None,
recv_us,
protocols,
event_type_filter,
) {
event.metadata_mut().handle_us =
crate::streaming::event_parser::common::high_performance_clock::elapsed_micros_since(
recv_us,
);
event = helpers::process_event(event, bot_wallet);
callback(event);
}
}
Ok(())
}
}