sol_parser_sdk/grpc/
yellowstone_tx_parse.rs

1//! Yellowstone `SubscribeUpdateTransaction` 单笔解析（logs ∥ instructions + 去重）。
2//! 从 [`super::client`] 抽出，供 crate 内与下游 streamer 复用。
3
4use smallvec::SmallVec;
5use solana_sdk::pubkey::Pubkey;
6use yellowstone_grpc_proto::prelude::{
7    SubscribeUpdateTransaction, Transaction, TransactionStatusMeta,
8};
9
10use super::transaction_meta::try_yellowstone_signature;
11use super::types::EventTypeFilter;
12use crate::DexEvent;
13
14const PROGRAM_DATA_PREFIX: &[u8] = b"Program data: ";
15
16struct ActiveProgram<'a> {
17    encoded: &'a str,
18    pubkey: Pubkey,
19}
20
21/// 解析单笔 Yellowstone 交易更新（含 meta）：并行 logs + enhanced instructions，再 log/ix 去重合并。
22#[inline]
23pub fn parse_subscribe_update_transaction(
24    tx: &SubscribeUpdateTransaction,
25    grpc_recv_us: i64,
26    block_us: Option<i64>,
27    filter: Option<&EventTypeFilter>,
28) -> Vec<DexEvent> {
29    parse_transaction_core(tx, grpc_recv_us, block_us, filter)
30}
31
32#[inline]
33pub(crate) fn parse_transaction_core(
34    tx: &SubscribeUpdateTransaction,
35    grpc_us: i64,
36    block_us: Option<i64>,
37    filter: Option<&EventTypeFilter>,
38) -> Vec<DexEvent> {
39    let Some(info) = &tx.transaction else { return Vec::new() };
40    let Some(meta) = &info.meta else { return Vec::new() };
41
42    let Some(sig) = try_yellowstone_signature(&info.signature) else {
43        return Vec::new();
44    };
45    let slot = tx.slot;
46    let idx = info.index;
47    let needs_pumpfun = filter.map(EventTypeFilter::includes_pumpfun).unwrap_or(true);
48    let is_created_buy =
49        needs_pumpfun && crate::logs::optimized_matcher::detect_pumpfun_create(&meta.log_messages);
50
51    let (log_events, instr_events) = rayon::join(
52        || {
53            parse_logs(
54                meta,
55                &info.transaction,
56                &meta.log_messages,
57                sig,
58                slot,
59                idx,
60                block_us,
61                grpc_us,
62                filter,
63                is_created_buy,
64            )
65        },
66        || {
67            parse_instructions(
68                meta,
69                &info.transaction,
70                sig,
71                slot,
72                idx,
73                block_us,
74                grpc_us,
75                filter,
76                is_created_buy,
77            )
78        },
79    );
80
81    let mut events =
82        crate::grpc::log_instr_dedup::dedupe_log_instruction_events(log_events, instr_events);
83    crate::grpc::transaction_meta::fill_recent_blockhash(&mut events, &info.transaction);
84    for event in &mut events {
85        crate::core::common_filler::fill_token_balances(event, meta, &info.transaction);
86    }
87    if let Some(filter) = filter {
88        events.into_iter().map(|e| filter.normalize_dex_event(e)).collect()
89    } else {
90        events
91    }
92}
93
94/// 单笔交易解析：**顺序**执行 logs → instructions 再合并。
95///
96/// 与 [`parse_subscribe_update_transaction`]（内部 `rayon::join` 并行）算法一致，但避免工作窃取与线程池调度，
97/// 在「单笔极低延迟」场景通常更快；适合嵌入 latency-sensitive 的订阅流水线。
98#[inline]
99pub fn parse_subscribe_update_transaction_low_latency(
100    tx: &SubscribeUpdateTransaction,
101    grpc_recv_us: i64,
102    block_us: Option<i64>,
103    filter: Option<&EventTypeFilter>,
104) -> Vec<DexEvent> {
105    parse_transaction_core_sequential(tx, grpc_recv_us, block_us, filter)
106}
107
108#[inline]
109fn parse_transaction_core_sequential(
110    tx: &SubscribeUpdateTransaction,
111    grpc_us: i64,
112    block_us: Option<i64>,
113    filter: Option<&EventTypeFilter>,
114) -> Vec<DexEvent> {
115    let Some(info) = &tx.transaction else {
116        return Vec::new();
117    };
118    let Some(meta) = &info.meta else {
119        return Vec::new();
120    };
121
122    let Some(sig) = try_yellowstone_signature(&info.signature) else {
123        return Vec::new();
124    };
125    let slot = tx.slot;
126    let idx = info.index;
127    let needs_pumpfun = filter.map(EventTypeFilter::includes_pumpfun).unwrap_or(true);
128    let is_created_buy =
129        needs_pumpfun && crate::logs::optimized_matcher::detect_pumpfun_create(&meta.log_messages);
130
131    let log_events = parse_logs(
132        meta,
133        &info.transaction,
134        &meta.log_messages,
135        sig,
136        slot,
137        idx,
138        block_us,
139        grpc_us,
140        filter,
141        is_created_buy,
142    );
143    let instr_events = parse_instructions(
144        meta,
145        &info.transaction,
146        sig,
147        slot,
148        idx,
149        block_us,
150        grpc_us,
151        filter,
152        is_created_buy,
153    );
154
155    let mut events =
156        crate::grpc::log_instr_dedup::dedupe_log_instruction_events(log_events, instr_events);
157    crate::grpc::transaction_meta::fill_recent_blockhash(&mut events, &info.transaction);
158    for event in &mut events {
159        crate::core::common_filler::fill_token_balances(event, meta, &info.transaction);
160    }
161    if let Some(filter) = filter {
162        events.into_iter().map(|e| filter.normalize_dex_event(e)).collect()
163    } else {
164        events
165    }
166}
167
168#[inline]
169fn parse_logs(
170    meta: &TransactionStatusMeta,
171    transaction: &Option<Transaction>,
172    logs: &[String],
173    sig: solana_sdk::signature::Signature,
174    slot: u64,
175    tx_idx: u64,
176    block_us: Option<i64>,
177    grpc_us: i64,
178    filter: Option<&EventTypeFilter>,
179    is_created_buy: bool,
180) -> Vec<DexEvent> {
181    let mut outer_idx: i32 = -1;
182    let mut inner_idx: i32 = -1;
183    let mut invokes = crate::core::invoke_context::InvokeContext::default();
184    let mut active_program_stack: SmallVec<[ActiveProgram<'_>; 8]> = SmallVec::new();
185    let mut result = Vec::with_capacity(4);
186
187    for log in logs {
188        if log.as_bytes().starts_with(PROGRAM_DATA_PREFIX) {
189            let current_program = active_program_stack.last().map(|active| &active.pubkey);
190            if let Some(mut e) = crate::logs::parse_log_with_program_id(
191                log,
192                sig,
193                slot,
194                tx_idx,
195                block_us,
196                grpc_us,
197                filter,
198                is_created_buy,
199                None,
200                current_program,
201            ) {
202                crate::core::account_dispatcher::fill_accounts_with_invoke_context(
203                    &mut e,
204                    meta,
205                    transaction,
206                    &invokes,
207                );
208                crate::core::common_filler::fill_data_with_invoke_context(
209                    &mut e,
210                    meta,
211                    transaction,
212                    &invokes,
213                );
214                result.push(e);
215            }
216            continue;
217        }
218
219        if let Some((pid, depth)) = crate::logs::optimized_matcher::parse_invoke_info(log) {
220            if depth == 1 {
221                inner_idx = -1;
222                outer_idx += 1;
223            } else {
224                inner_idx += 1;
225            }
226            let pk = crate::grpc::program_ids::known_program_id(pid).unwrap_or_default();
227            active_program_stack.truncate(depth - 1);
228            active_program_stack.push(ActiveProgram { encoded: pid, pubkey: pk });
229            if crate::grpc::program_ids::needs_invoke_context(&pk) {
230                invokes.push(pk, (outer_idx, inner_idx));
231            }
232            continue;
233        }
234
235        if let Some(pid) = crate::logs::optimized_matcher::parse_program_complete_info(log) {
236            if let Some(pos) = active_program_stack.iter().rposition(|active| active.encoded == pid)
237            {
238                active_program_stack.truncate(pos);
239            }
240        }
241    }
242    result
243}
244
245#[inline]
246fn parse_instructions(
247    meta: &TransactionStatusMeta,
248    transaction: &Option<Transaction>,
249    sig: solana_sdk::signature::Signature,
250    slot: u64,
251    tx_idx: u64,
252    block_us: Option<i64>,
253    grpc_us: i64,
254    filter: Option<&EventTypeFilter>,
255    is_created_buy: bool,
256) -> Vec<DexEvent> {
257    crate::grpc::instruction_parser::parse_instructions_enhanced_with_created_buy(
258        meta,
259        transaction,
260        sig,
261        slot,
262        tx_idx,
263        block_us,
264        grpc_us,
265        filter,
266        is_created_buy,
267    )
268}
sol_parser_sdk/grpc/yellowstone_tx_parse.rs

sol_parser_sdk/grpc/
yellowstone_tx_parse.rs
