gavindiaz/solana-streamer
1
0
Fork 0
mirror of https://github.com/0xfnzero/solana-streamer.git synced 2026-08-17 10:58:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(shred): tx_index, Migrate without CPI, merge fix, doc limits

Browse Source 
- Add tx_index to TransactionWithSlot (entry index), pass to parser
- Emit PumpFun Migrate even when no CPI (shred gets instruction-only)
- Merge: only overwrite Create/CreateV2 with CPI when value non-default
- Rename transaction_index -> tx_index across codebase
- Add docs/SHREDSTREAM_LIMITATIONS.md (ALT/CPI limits, field completeness)
- Comment shred path: static_account_keys only, no inner_instructions

Made-with: Cursor
This commit is contained in:
Wood
2026-03-07 12:35:57 +08:00
parent 04df6db0b0
commit 7270372604
14 changed files with 219 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/streaming/common/event_processor.rs
+7 -4
View File
@@ -66,7 +66,7 @@ pub async fn process_grpc_transaction(
let signature = transaction_pretty.signature;
let block_time = transaction_pretty.block_time;
let recv_us = transaction_pretty.recv_us;
let transaction_index = transaction_pretty.transaction_index;
let tx_index = transaction_pretty.tx_index;
let grpc_tx = transaction_pretty.grpc_tx;
let adapter_callback = create_metrics_callback(callback.clone());
@@ -80,7 +80,7 @@ pub async fn process_grpc_transaction(
block_time,
recv_us,
bot_wallet,
transaction_index,
tx_index,
adapter_callback,
)
.await?;
@@ -121,6 +121,7 @@ pub async fn process_shred_transaction(
let tx = transaction_with_slot.transaction;
let slot = transaction_with_slot.slot;
let tx_index = transaction_with_slot.tx_index;
if tx.signatures.is_empty() {
return Ok(());
@@ -130,6 +131,8 @@ pub async fn process_shred_transaction(
let recv_us = transaction_with_slot.recv_us;
let adapter_callback = create_metrics_callback(callback);
// Shred 路径仅能拿到 static_account_keys，且无 inner_instructions，解析限制见 docs/SHREDSTREAM_LIMITATIONS.md
// 若交易使用 ALT，账户可能为 default/错误；无 CPI 合并，timestamp/reserves 等多为 0。
let accounts = tx.message.static_account_keys();
EventParser::parse_instruction_events_from_versioned_transaction(
@@ -138,12 +141,12 @@ pub async fn process_shred_transaction(
&tx,
signature,
Some(slot),
None,
None, // shred 无 block_time
recv_us,
accounts,
&[],
bot_wallet,
None,
tx_index,
adapter_callback,
)
.await?;
src/streaming/event_parser/common/types.rs
+3 -3
View File
@@ -301,7 +301,7 @@ pub struct SwapData {
pub struct EventMetadata {
pub signature: Signature,
pub slot: u64,
pub transaction_index: Option<u64>, // 新增：交易在slot中的索引
pub tx_index: Option<u64>, // 新增：交易在slot中的索引
pub block_time: i64,
pub block_time_ms: i64,
pub recv_us: i64,
@@ -327,7 +327,7 @@ impl EventMetadata {
outer_index: i64,
inner_index: Option<i64>,
recv_us: i64,
transaction_index: Option<u64>,
tx_index: Option<u64>,
) -> Self {
Self {
signature,
@@ -342,7 +342,7 @@ impl EventMetadata {
swap_data: None,
outer_index,
inner_index,
transaction_index,
tx_index,
}
}
src/streaming/event_parser/core/event_parser.rs
+14 -28
View File
@@ -39,7 +39,7 @@ impl EventParser {
block_time: Option<Timestamp>,
recv_us: i64,
bot_wallet: Option<Pubkey>,
transaction_index: Option<u64>,
tx_index: Option<u64>,
callback: Arc<dyn Fn(DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
// 创建适配器回调，将所有权回调转换为引用回调
@@ -93,7 +93,7 @@ impl EventParser {
&accounts,
&inner_instructions,
bot_wallet,
transaction_index,
tx_index,
adapter_callback,
)
.await?;
@@ -119,7 +119,7 @@ impl EventParser {
accounts: &[Pubkey],
inner_instructions: &[InnerInstructions],
bot_wallet: Option<Pubkey>,
transaction_index: Option<u64>,
tx_index: Option<u64>,
callback: Arc<dyn Fn(DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
// 创建适配器回调，将所有权回调转换为引用回调
@@ -159,7 +159,7 @@ impl EventParser {
index as i64,
None,
bot_wallet,
transaction_index,
tx_index,
inner_instructions,
adapter_callback.clone(),
)?;
@@ -181,7 +181,7 @@ impl EventParser {
index as i64,
Some(inner_index as i64),
bot_wallet,
transaction_index,
tx_index,
Some(&inner_instructions),
adapter_callback.clone(),
)?;
@@ -213,7 +213,7 @@ impl EventParser {
accounts: &[Pubkey],
inner_instructions: &[yellowstone_grpc_proto::prelude::InnerInstructions],
bot_wallet: Option<Pubkey>,
transaction_index: Option<u64>,
tx_index: Option<u64>,
callback: Arc<dyn for<'a> Fn(&'a DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
// 获取交易的指令和账户
@@ -248,7 +248,7 @@ impl EventParser {
index as i64,
None,
bot_wallet,
transaction_index,
tx_index,
inner_instructions,
callback.clone(),
)?;
@@ -278,7 +278,7 @@ impl EventParser {
inner_instructions.index as i64,
Some(inner_index as i64),
bot_wallet,
transaction_index,
tx_index,
Some(&inner_instructions),
callback.clone(),
)?;
@@ -307,7 +307,7 @@ impl EventParser {
outer_index: i64,
inner_index: Option<i64>,
bot_wallet: Option<Pubkey>,
transaction_index: Option<u64>,
tx_index: Option<u64>,
inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>,
callback: Arc<dyn for<'a> Fn(&'a DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
@@ -346,7 +346,7 @@ impl EventParser {
outer_index,
inner_index,
recv_us,
transaction_index,
tx_index,
);
if is_cu_program {
@@ -428,14 +428,7 @@ impl EventParser {
}
}
// 特殊处理: PumpFun MIGRATE 指令需要 inner instruction data
if matches!(protocol, Protocol::PumpFun) {
const PUMPFUN_MIGRATE_IX: &[u8] = &[155, 234, 231, 146, 236, 158, 162, 30];
if instruction_discriminator == PUMPFUN_MIGRATE_IX && inner_instruction_event.is_none()
{
return Ok(());
}
}
// PumpFun MIGRATE: 有 CPI 时合并 log；无 CPI 时仍发出仅含指令数据的事件。
// 合并事件
if let Some(inner_instruction_event) = inner_instruction_event {
@@ -471,7 +464,7 @@ impl EventParser {
outer_index: i64,
inner_index: Option<i64>,
bot_wallet: Option<Pubkey>,
transaction_index: Option<u64>,
tx_index: Option<u64>,
inner_instructions: Option<&InnerInstructions>,
callback: Arc<dyn for<'a> Fn(&'a DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
@@ -511,7 +504,7 @@ impl EventParser {
outer_index,
inner_index,
recv_us,
transaction_index,
tx_index,
);
if is_cu_program {
@@ -611,14 +604,7 @@ impl EventParser {
}
}
// 特殊处理: PumpFun MIGRATE 指令需要 inner instruction data
if matches!(protocol, Protocol::PumpFun) {
const PUMPFUN_MIGRATE_IX: &[u8] = &[155, 234, 231, 146, 236, 158, 162, 30];
if instruction_discriminator == PUMPFUN_MIGRATE_IX && inner_instruction_event.is_none()
{
return Ok(());
}
}
// PumpFun MIGRATE: 有 CPI 时合并 log；无 CPI（如 shred）仍发出仅含指令数据的事件。
// 合并事件
if let Some(inner_instruction_event) = inner_instruction_event {
src/streaming/event_parser/core/merger_event.rs
+60 -20
View File
@@ -36,16 +36,36 @@ pub fn merge(instruction_event: &mut DexEvent, cpi_log_event: DexEvent) {
},
DexEvent::PumpFunCreateTokenEvent(e) => match cpi_log_event {
DexEvent::PumpFunCreateV2TokenEvent(cpie) => {
e.mint = cpie.mint;
e.bonding_curve = cpie.bonding_curve;
e.user = cpie.user;
e.creator = cpie.creator;
e.timestamp = cpie.timestamp;
e.virtual_token_reserves = cpie.virtual_token_reserves;
e.virtual_sol_reserves = cpie.virtual_sol_reserves;
e.real_token_reserves = cpie.real_token_reserves;
e.token_total_supply = cpie.token_total_supply;
e.token_program = cpie.token_program;
if cpie.mint != solana_sdk::pubkey::Pubkey::default() {
e.mint = cpie.mint;
}
if cpie.bonding_curve != solana_sdk::pubkey::Pubkey::default() {
e.bonding_curve = cpie.bonding_curve;
}
if cpie.user != solana_sdk::pubkey::Pubkey::default() {
e.user = cpie.user;
}
if cpie.creator != solana_sdk::pubkey::Pubkey::default() {
e.creator = cpie.creator;
}
if cpie.timestamp != 0 {
e.timestamp = cpie.timestamp;
}
if cpie.virtual_token_reserves != 0 {
e.virtual_token_reserves = cpie.virtual_token_reserves;
}
if cpie.virtual_sol_reserves != 0 {
e.virtual_sol_reserves = cpie.virtual_sol_reserves;
}
if cpie.real_token_reserves != 0 {
e.real_token_reserves = cpie.real_token_reserves;
}
if cpie.token_total_supply != 0 {
e.token_total_supply = cpie.token_total_supply;
}
if cpie.token_program != solana_sdk::pubkey::Pubkey::default() {
e.token_program = cpie.token_program;
}
e.is_mayhem_mode = cpie.is_mayhem_mode;
e.is_cashback_enabled = cpie.is_cashback_enabled;
}
@@ -53,16 +73,36 @@ pub fn merge(instruction_event: &mut DexEvent, cpi_log_event: DexEvent) {
},
DexEvent::PumpFunCreateV2TokenEvent(e) => match cpi_log_event {
DexEvent::PumpFunCreateV2TokenEvent(cpie) => {
e.mint = cpie.mint;
e.bonding_curve = cpie.bonding_curve;
e.user = cpie.user;
e.creator = cpie.creator;
e.timestamp = cpie.timestamp;
e.virtual_token_reserves = cpie.virtual_token_reserves;
e.virtual_sol_reserves = cpie.virtual_sol_reserves;
e.real_token_reserves = cpie.real_token_reserves;
e.token_total_supply = cpie.token_total_supply;
e.token_program = cpie.token_program;
if cpie.mint != solana_sdk::pubkey::Pubkey::default() {
e.mint = cpie.mint;
}
if cpie.bonding_curve != solana_sdk::pubkey::Pubkey::default() {
e.bonding_curve = cpie.bonding_curve;
}
if cpie.user != solana_sdk::pubkey::Pubkey::default() {
e.user = cpie.user;
}
if cpie.creator != solana_sdk::pubkey::Pubkey::default() {
e.creator = cpie.creator;
}
if cpie.timestamp != 0 {
e.timestamp = cpie.timestamp;
}
if cpie.virtual_token_reserves != 0 {
e.virtual_token_reserves = cpie.virtual_token_reserves;
}
if cpie.virtual_sol_reserves != 0 {
e.virtual_sol_reserves = cpie.virtual_sol_reserves;
}
if cpie.real_token_reserves != 0 {
e.real_token_reserves = cpie.real_token_reserves;
}
if cpie.token_total_supply != 0 {
e.token_total_supply = cpie.token_total_supply;
}
if cpie.token_program != solana_sdk::pubkey::Pubkey::default() {
e.token_program = cpie.token_program;
}
e.is_mayhem_mode = cpie.is_mayhem_mode;
e.is_cashback_enabled = cpie.is_cashback_enabled;
}
src/streaming/event_parser/protocols/pumpfun/parser.rs
+2
View File
@@ -192,6 +192,8 @@ fn parse_create_token_instruction(
/// 5: user, 6: system_program, 7: token_program, 8: associated_token_program, 9: mayhem_program_id,
/// 10: global_params, 11: sol_vault, 12: mayhem_state, 13: mayhem_token_vault, 14: event_authority, 15: program.
/// 共 16 个固定账户，不足时返回 None 避免越界。
/// 注意：shredstream 路径仅传入 static_account_keys，若交易使用 Address Lookup Tables，
/// 无法解析 loaded_addresses，部分账户会以 default 填充，导致 token_program/global 等错误。
fn parse_create_v2_token_instruction(
data: &[u8],
accounts: &[Pubkey],
src/streaming/grpc/pool.rs
+1 -1
View File
@@ -276,7 +276,7 @@ impl PooledTransactionPretty {
let tx = tx_update.transaction.expect("should be defined");
self.transaction.slot = tx_update.slot;
self.transaction.transaction_index = Some(tx.index);
self.transaction.tx_index = Some(tx.index);
self.transaction.block_time = block_time;
self.transaction.block_hash.clear(); // 重置 block_hash
self.transaction.signature =
src/streaming/grpc/types.rs
+5 -5
View File
@@ -68,7 +68,7 @@ impl fmt::Debug for BlockMetaPretty {
#[derive(Clone)]
pub struct TransactionPretty {
pub slot: u64,
pub transaction_index: Option<u64>, // 新增：交易在slot中的索引
pub tx_index: Option<u64>, // 新增：交易在slot中的索引
pub block_hash: String,
pub block_time: Option<Timestamp>,
pub signature: Signature,
@@ -81,7 +81,7 @@ impl fmt::Debug for TransactionPretty {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("TransactionPretty")
.field("slot", &self.slot)
.field("transaction_index", &self.transaction_index)
.field("tx_index", &self.tx_index)
.field("signature", &self.signature)
.field("is_vote", &self.is_vote)
.field("recv_us", &self.recv_us)
@@ -93,7 +93,7 @@ impl Default for TransactionPretty {
fn default() -> Self {
Self {
slot: 0,
transaction_index: None,
tx_index: None,
block_hash: String::new(),
block_time: None,
signature: Signature::default(),
@@ -150,10 +150,10 @@ impl Default for TransactionPretty {
// ) -> Self {
// let tx = transaction.expect("should be defined");
// // 根据用户说明，交易索引在 transaction.index 中
// let transaction_index = tx.index;
// let tx_index = tx.index;
// Self {
// slot,
// transaction_index: Some(transaction_index), // 提取交易索引
// tx_index: Some(tx_index), // 提取交易索引
// block_time,
// block_hash: String::new(),
// signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"),
src/streaming/shred/pool.rs
+14 -8
View File
@@ -49,14 +49,16 @@ pub struct PooledTransactionWithSlot {
impl PooledTransactionWithSlot {
/// 从原始数据重置
pub fn reset_from_data(
&mut self,
transaction: VersionedTransaction,
slot: u64,
recv_us: i64
&mut self,
transaction: VersionedTransaction,
slot: u64,
recv_us: i64,
tx_index: Option<u64>,
) {
self.transaction.transaction = transaction;
self.transaction.slot = slot;
self.transaction.recv_us = recv_us;
self.transaction.tx_index = tx_index;
}
/// 使用优化的工厂方法创建 TransactionWithSlot（移动数据而不是克隆）
@@ -73,6 +75,7 @@ impl Drop for PooledTransactionWithSlot {
// 清理敏感数据
self.transaction.slot = 0;
self.transaction.recv_us = 0;
self.transaction.tx_index = None;
// 重置交易为默认值以清理敏感数据
self.transaction.transaction = VersionedTransaction::default();
pool.push_back(std::mem::take(&mut self.transaction));
@@ -119,9 +122,10 @@ impl ShredPoolManager {
transaction: VersionedTransaction,
slot: u64,
recv_us: i64,
tx_index: Option<u64>,
) -> TransactionWithSlot {
let mut pooled_tx = self.transaction_pool.acquire();
pooled_tx.reset_from_data(transaction, slot, recv_us);
pooled_tx.reset_from_data(transaction, slot, recv_us, tx_index);
pooled_tx.into_transaction_with_slot()
}
}
@@ -146,11 +150,13 @@ pub mod factory {
transaction: VersionedTransaction,
slot: u64,
recv_us: i64,
tx_index: Option<u64>,
) -> TransactionWithSlot {
GLOBAL_SHRED_POOL_MANAGER.create_transaction_with_slot_optimized(
transaction,
slot,
recv_us
transaction,
slot,
recv_us,
tx_index,
)
}
}
src/streaming/shred/types.rs
+4 -1
View File
@@ -6,6 +6,8 @@ pub struct TransactionWithSlot {
pub transaction: VersionedTransaction,
pub slot: u64,
pub recv_us: i64,
/// 交易在 entry 内的索引（shredstream 无 slot 级 index 时用作 best-effort）
pub tx_index: Option<u64>,
}
impl TransactionWithSlot {
@@ -14,7 +16,8 @@ impl TransactionWithSlot {
transaction: VersionedTransaction,
slot: u64,
recv_us: i64,
tx_index: Option<u64>,
) -> Self {
Self { transaction, slot, recv_us }
Self { transaction, slot, recv_us, tx_index }
}
}
src/streaming/shred_stream.rs
+2 -1
View File
@@ -51,12 +51,13 @@ impl ShredStreamGrpc {
Ok(msg) => {
if let Ok(entries) = bincode::deserialize::<Vec<Entry>>(&msg.entries) {
for entry in entries {
for transaction in entry.transactions {
for (tx_index, transaction) in entry.transactions.iter().enumerate() {
let transaction_with_slot =
factory::create_transaction_with_slot_pooled(
transaction.clone(),
msg.slot,
get_high_perf_clock(),
Some(tx_index as u64),
);
// Process transaction - clone Arc and Vec for each call
if let Err(e) = process_shred_transaction(