From 727037260493921ffe6dbdb8d731f30b913e2e52 Mon Sep 17 00:00:00 2001 From: Wood Date: Sat, 7 Mar 2026 12:35:57 +0800 Subject: [PATCH] feat(shred): tx_index, Migrate without CPI, merge fix, doc limits - 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 --- docs/SHREDSTREAM_LIMITATIONS.md | 101 ++++++++++++++++++ examples/grpc_example.rs | 4 +- examples/parse_tx_events.rs | 4 +- examples/shred_example.rs | 4 +- src/streaming/common/event_processor.rs | 11 +- src/streaming/event_parser/common/types.rs | 6 +- .../event_parser/core/event_parser.rs | 42 +++----- .../event_parser/core/merger_event.rs | 80 ++++++++++---- .../event_parser/protocols/pumpfun/parser.rs | 2 + src/streaming/grpc/pool.rs | 2 +- src/streaming/grpc/types.rs | 10 +- src/streaming/shred/pool.rs | 22 ++-- src/streaming/shred/types.rs | 5 +- src/streaming/shred_stream.rs | 3 +- 14 files changed, 219 insertions(+), 77 deletions(-) create mode 100644 docs/SHREDSTREAM_LIMITATIONS.md diff --git a/docs/SHREDSTREAM_LIMITATIONS.md b/docs/SHREDSTREAM_LIMITATIONS.md new file mode 100644 index 0000000..9c11ee8 --- /dev/null +++ b/docs/SHREDSTREAM_LIMITATIONS.md @@ -0,0 +1,101 @@ +# Shredstream 解析限制与差异说明 + +与 gRPC 订阅相比，shredstream 路径存在以下限制和解析差异，使用时请注意。 + +## 1. 数据源差异 + +| 数据 | gRPC | Shredstream | +|------|------|-------------| +| 账户列表 | 完整 resolved 列表（static + loaded_addresses） | 仅 `static_account_keys()` | +| Inner instructions (CPI) | 有（来自区块执行结果） | **无**（Entry 仅含原始交易） | +| block_time | 有 | **无**（恒为 0） | +| tx_index | slot 内交易索引 | entry 内交易索引（best-effort） | + +## 2. 解析问题与遗漏 + +**漏掉事件小结**：Shred 会**整笔漏掉**「仅通过 CPI 触发的」所有协议事件（例如经 Jupiter 等聚合器路由的 PumpFun/PumpSwap 等），因为 shred 不解析 inner instructions。详见 2.3。 + +### 2.1 使用 Address Lookup Tables (ALT) 的交易 + +- **现象**：指令中的账户索引指向「static + loaded」的完整列表，shred 只传入 static，越界索引会被填成 `Pubkey::default()`（11111...）。 +- **影响协议/指令**：所有依赖「按索引取账户」的指令，在交易使用 ALT 时都可能出现错误或默认账户。 +- **典型表现**： + - **PumpFun**：Create / CreateV2 的 token_program、global、event_authority 等为 11111...；Buy/Sell 的 creator_vault、token_program 等可能错误。 + - **PumpSwap / Bonk / Raydium / Meteora**：依赖高索引账户的指令同样可能得到错误或 default 账户。 +- **建议**：若需完整且正确的账户字段，请使用 gRPC 订阅。 + +### 2.2 无 Inner Instructions → 无 CPI 合并 + +- **原因**：Shred/Entry 只包含原始 `VersionedTransaction`，inner instructions 是执行阶段产物，不在 shred 载荷中。 +- **影响**： + - **PumpFun** + - Create / CreateV2：无 CPI 合并 → `timestamp`、`virtual_*_reserves`、`real_*_reserves`、`token_total_supply`、`token_program`（来自 log）等多为 0 或默认。 + - Trade：无 CPI 合并 → 无 log 中的成交额、reserves、fee 等明细，仅保留指令层数据。 + - Migrate：此前因「必须带 CPI」被直接跳过，**现已改为** shred 下仍发出仅含指令数据的 Migrate 事件（user/mint 等来自指令账户；mint_amount、sol_amount、timestamp、pool 等来自 CPI 的字段为 0/默认）。 + - **PumpSwap**：buy/sell/deposit/withdraw/create_pool 无 CPI 合并，无 log 中的金额、reserves 等；**swap_data** 依赖后续指令解析，inner 为空时恒为空。 + - **Bonk**：trade、pool_create 无 CPI 合并，缺少 log 明细。 + - **Meteora Damm V2**：swap、initialize_pool 无 CPI 合并。 + - **Raydium**：依赖 inner 的解析/合并与 gRPC 一致缺失。 + +### 2.3 漏掉的事件：仅通过 CPI 触发的调用 + +- **原因**：Shred 路径只遍历并解析**外层指令**（`transaction.message.instructions()`）。内层指令（inner instructions）只有在传入非空的 `inner_instructions` 时才会被解析；shred 传入的为 `&[]`，因此**从不**解析任何 inner。 +- **结果**：当协议**仅作为 CPI 被调用**时（例如用户通过 Jupiter/Raydium 聚合器等路由，外层指令是聚合器，PumpFun/PumpSwap 等只在 inner 中出现），gRPC 会解析该 inner 并发出对应事件，**shred 则整笔交易都不会产生该协议的任何事件**。 +- **影响**：所有协议（PumpFun、PumpSwap、Bonk、Raydium、Meteora 等）在「仅 CPI 调用」场景下，shred 都会**漏掉整笔事件**，不是字段缺失，而是事件本身不会出现。 +- **建议**：若需要统计或处理通过聚合器/路由产生的交易，必须使用 gRPC 订阅；shred 只适合「用户直接与协议交互」的链路。 + +### 2.4 其他明确「漏掉」或弱化的解析 + +- **PumpFun Migrate**：shred 下**会**发出事件，但仅包含指令解析出的账户与部分字段（如 user、mint）；mint_amount、sol_amount、pool_migration_fee、timestamp、pool 等来自 CPI 的字段为 0/默认。 +- **所有协议的 CPI 维度的数据**：shred 路径一律缺失（无 inner instructions 即无 CPI 解析与 merge）。 + +## 3. 使用建议 + +- 需要**完整、正确**的账户与 log 字段（reserves、timestamp、amounts、swap_data 等）时，使用 **gRPC 订阅**。 +- Shredstream 更适合：对延迟更敏感、可接受「仅指令层 + 部分字段缺失/默认」的场景，或确认交易**未使用 ALT** 时的账户解析。 + +## 4. 各事件 Shred 路径字段完整性 + +以下为「直接外层调用」场景下，shred 能拿到的字段 vs 仅 CPI 合并才有的字段（shred 下为 0/默认）。若交易使用 ALT，标注为「指令」的账户类字段也可能错误或为 default。 + +**元数据（所有事件）** +- Shred 有：signature, slot, recv_us, program_id, outer_index, tx_index（entry 内索引）, event_type, protocol +- Shred 缺失：**block_time / block_time_ms**（恒为 0），**swap_data**（恒为 None，依赖 inner 后续指令解析） + +### 4.1 PumpFun + +| 事件 | 指令解析有（Shred 有） | 仅 CPI 合并有（Shred 缺失） | +|------|------------------------|-----------------------------| +| **CreateToken** | name, symbol, uri, creator, mint, 各账户(0..13) | timestamp, virtual_*_reserves, real_*_reserves, token_total_supply, token_program(来自 log), is_mayhem_mode, is_cashback_enabled | +| **CreateV2Token** | name, symbol, uri, creator, mint, 各账户(0..15) | timestamp, virtual_*_reserves, real_*_reserves, token_total_supply, token_program(来自 log), is_mayhem_mode, is_cashback_enabled | +| **Trade** (Buy/Sell) | is_buy, amount/max_sol_cost/min_sol_output, 各账户(含 user, mint, creator_vault 等) | sol_amount, token_amount, timestamp, virtual_*_reserves, real_*_reserves, fee_recipient, fee_basis_points, fee, creator, creator_fee_*, track_volume, total_unclaimed/claimed_tokens, current_sol_volume, last_update_timestamp, ix_name, mayhem_mode, cashback_* | +| **Migrate** | user, mint, bonding_curve, 全部 24 个账户 | mint_amount, sol_amount, pool_migration_fee, timestamp, pool（CPI 的 pool） | + +### 4.2 PumpSwap + +| 事件 | 指令解析有（Shred 有） | 仅 CPI 合并有（Shred 缺失） | +|------|------------------------|-----------------------------| +| **Buy** | base_amount_out, max_quote_amount_in, pool, user, base_mint, quote_mint, 各 token account / fee recipient / program，coin_creator_vault_ata/authority(若 accounts≥19) | timestamp, 实际 quote_amount_in, user/pool *_reserves, lp_fee, protocol_fee, coin_creator_fee_*, track_volume, total_unclaimed/claimed_tokens, current_sol_volume, last_update_timestamp | +| **Sell** | base_amount_in, min_quote_amount_out, pool, user, base_mint, quote_mint, 各账户，coin_creator_vault_* | timestamp, 实际 quote_amount_out, *_reserves, 各项 fee, coin_creator_fee_* | +| **CreatePool** | index, base_amount_in, quote_amount_in, coin_creator(若 data≥50), pool, creator, base/quote_mint, lp_mint, 各 token account | timestamp, base_mint_decimals, quote_mint_decimals, pool_base/quote_amount, minimum/initial_liquidity, lp_token_amount_out, pool_bump | +| **Deposit** | lp_token_amount_out, max_base/quote_amount_in, pool, user, 各 mint / token account | timestamp, user/pool *_reserves, base_amount_in, quote_amount_in, lp_mint_supply 等 | +| **Withdraw** | lp_token_amount_in, min_base/quote_amount_out, pool, user, 各账户 | timestamp, *_reserves, base/quote_amount_out, lp_mint_supply 等 | + +### 4.3 Bonk + +| 事件 | 指令解析有（Shred 有） | 仅 CPI 合并有（Shred 缺失） | +|------|------------------------|-----------------------------| +| **Trade** | amount_in/out, minimum/maximum_*, share_fee_rate, payer, pool_state, 各 vault/mint/program 账户, trade_direction | pool_state(来自 log), total_base_sell, virtual_base/quote, real_*_before/after, amount_in/out(实际成交), protocol_fee, platform_fee, creator_fee, share_fee, pool_status, exact_in | +| **PoolCreate** | payer, creator, global_config, platform_config, pool_state, base/quote_mint, base/quote_vault, base_mint_param, curve_param, vesting_param(, amm_fee_on for V2) | config, base_mint_param/curve_param/vesting_param(来自 log 的完整值), amm_fee_on(来自 log) | +| **MigrateToAmm / MigrateToCpswap** | 指令侧账户与参数 | base_lot_size, quote_lot_size, market_vault_signer_nonce（CPI 才有） | + +### 4.4 Raydium / Meteora Damm V2 + +- **Raydium CLMM/CPMM/AMM**：指令解析会填账户与指令内参数（如 amount、min_out 等）；实际成交额、reserves、fee 等来自 log 的字段在 shred 下均为 0/默认。 +- **Meteora Damm V2**：Swap / InitializePool 等同上，指令层有账户与部分参数，CPI 的 timestamp、reserves、实际 amount 等 shred 缺失。 + +## 5. 代码位置参考 + +- Shred 入口：`streaming/common/event_processor.rs` → `process_shred_transaction` +- 账户与 inner 传入：`accounts = tx.message.static_account_keys()`，`inner_instructions: &[]` +- 合并逻辑（CPI 覆盖/补充字段）：`streaming/event_parser/core/merger_event.rs` → `merge()` diff --git a/examples/grpc_example.rs b/examples/grpc_example.rs index 98deb5e..3dab927 100644 --- a/examples/grpc_example.rs +++ b/examples/grpc_example.rs @@ -113,9 +113,9 @@ async fn test_grpc() -> Result<(), Box> { fn create_event_callback() -> impl Fn(DexEvent) { |event: DexEvent| { println!( - "🎉 Event received! Type: {:?}, transaction_index: {:?}", + "🎉 Event received! Type: {:?}, tx_index: {:?}", event.metadata().event_type, - event.metadata().transaction_index + event.metadata().tx_index ); match event { DexEvent::BlockMetaEvent(e) => { diff --git a/examples/parse_tx_events.rs b/examples/parse_tx_events.rs index 0afe5f0..10506a7 100644 --- a/examples/parse_tx_events.rs +++ b/examples/parse_tx_events.rs @@ -177,7 +177,7 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> { .unwrap() .as_micros() as i64; let bot_wallet = None; - let transaction_index = None; + let tx_index = None; let protocols = vec![ Protocol::Bonk, @@ -206,7 +206,7 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> { &accounts, &inner_instructions_vec, bot_wallet, - transaction_index, + tx_index, callback, ) .await?; diff --git a/examples/shred_example.rs b/examples/shred_example.rs index 6659aab..74666f2 100644 --- a/examples/shred_example.rs +++ b/examples/shred_example.rs @@ -57,9 +57,9 @@ async fn test_shreds() -> Result<(), Box> { fn create_event_callback() -> impl Fn(DexEvent) { |event: DexEvent| { println!( - "🎉 Event received! Type: {:?}, transaction_index: {:?}", + "🎉 Event received! Type: {:?}, tx_index: {:?}", event.metadata().event_type, - event.metadata().transaction_index + event.metadata().tx_index ); match event { DexEvent::BlockMetaEvent(e) => { diff --git a/src/streaming/common/event_processor.rs b/src/streaming/common/event_processor.rs index 87e986d..894a209 100644 --- a/src/streaming/common/event_processor.rs +++ b/src/streaming/common/event_processor.rs @@ -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?; diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index 7bcaa96..f33d78f 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -301,7 +301,7 @@ pub struct SwapData { pub struct EventMetadata { pub signature: Signature, pub slot: u64, - pub transaction_index: Option, // 新增：交易在slot中的索引 + pub tx_index: Option, // 新增：交易在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, recv_us: i64, - transaction_index: Option, + tx_index: Option, ) -> Self { Self { signature, @@ -342,7 +342,7 @@ impl EventMetadata { swap_data: None, outer_index, inner_index, - transaction_index, + tx_index, } } diff --git a/src/streaming/event_parser/core/event_parser.rs b/src/streaming/event_parser/core/event_parser.rs index 93a8d4e..782427a 100644 --- a/src/streaming/event_parser/core/event_parser.rs +++ b/src/streaming/event_parser/core/event_parser.rs @@ -39,7 +39,7 @@ impl EventParser { block_time: Option, recv_us: i64, bot_wallet: Option, - transaction_index: Option, + tx_index: Option, callback: Arc, ) -> 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, - transaction_index: Option, + tx_index: Option, callback: Arc, ) -> 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, - transaction_index: Option, + tx_index: Option, callback: Arc 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, bot_wallet: Option, - transaction_index: Option, + tx_index: Option, inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>, callback: Arc 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, bot_wallet: Option, - transaction_index: Option, + tx_index: Option, inner_instructions: Option<&InnerInstructions>, callback: Arc 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 { diff --git a/src/streaming/event_parser/core/merger_event.rs b/src/streaming/event_parser/core/merger_event.rs index 8d7d588..ce4258b 100644 --- a/src/streaming/event_parser/core/merger_event.rs +++ b/src/streaming/event_parser/core/merger_event.rs @@ -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; } diff --git a/src/streaming/event_parser/protocols/pumpfun/parser.rs b/src/streaming/event_parser/protocols/pumpfun/parser.rs index 7cf78ed..7968fe8 100755 --- a/src/streaming/event_parser/protocols/pumpfun/parser.rs +++ b/src/streaming/event_parser/protocols/pumpfun/parser.rs @@ -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], diff --git a/src/streaming/grpc/pool.rs b/src/streaming/grpc/pool.rs index c8e6b88..de671de 100644 --- a/src/streaming/grpc/pool.rs +++ b/src/streaming/grpc/pool.rs @@ -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 = diff --git a/src/streaming/grpc/types.rs b/src/streaming/grpc/types.rs index 6abfca3..9a526c2 100644 --- a/src/streaming/grpc/types.rs +++ b/src/streaming/grpc/types.rs @@ -68,7 +68,7 @@ impl fmt::Debug for BlockMetaPretty { #[derive(Clone)] pub struct TransactionPretty { pub slot: u64, - pub transaction_index: Option, // 新增：交易在slot中的索引 + pub tx_index: Option, // 新增：交易在slot中的索引 pub block_hash: String, pub block_time: Option, 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"), diff --git a/src/streaming/shred/pool.rs b/src/streaming/shred/pool.rs index 5dea151..47805fb 100644 --- a/src/streaming/shred/pool.rs +++ b/src/streaming/shred/pool.rs @@ -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, ) { 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, ) -> 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, ) -> TransactionWithSlot { GLOBAL_SHRED_POOL_MANAGER.create_transaction_with_slot_optimized( - transaction, - slot, - recv_us + transaction, + slot, + recv_us, + tx_index, ) } } diff --git a/src/streaming/shred/types.rs b/src/streaming/shred/types.rs index ca87cc4..08691d6 100644 --- a/src/streaming/shred/types.rs +++ b/src/streaming/shred/types.rs @@ -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, } impl TransactionWithSlot { @@ -14,7 +16,8 @@ impl TransactionWithSlot { transaction: VersionedTransaction, slot: u64, recv_us: i64, + tx_index: Option, ) -> Self { - Self { transaction, slot, recv_us } + Self { transaction, slot, recv_us, tx_index } } } diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index f98a412..6ff6b86 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -51,12 +51,13 @@ impl ShredStreamGrpc { Ok(msg) => { if let Ok(entries) = bincode::deserialize::>(&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(