diff --git a/Cargo.toml b/Cargo.toml index 7f6ddd4..6c3aa1b 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,7 +21,7 @@ sdk-perf-stats = ["sol-parser-sdk/perf-stats"] sdk-ultra-perf = ["sol-parser-sdk/ultra-perf"] [dependencies] -sol-parser-sdk = { version = "0.6.2", default-features = false } +sol-parser-sdk = { git = "https://github.com/0xfnzero/sol-parser-sdk.git", rev = "67b56651996901fabbc8219e65a2ba06c51fddac", default-features = false } solana-sdk = "3.0.0" solana-client = "3.1.12" solana-transaction-status = "3.1.12" diff --git a/README.md b/README.md index f9957c4..f3e5172 100755 --- a/README.md +++ b/README.md @@ -406,6 +406,25 @@ let event_type_filter = Some(EventTypeFilter::include_only(vec![ let event_type_filter = Some(EventTypeFilter::exclude_only(vec![EventType::BlockMeta])); ``` +Transaction fee, priority-fee, and SWQoS tip parsing is strictly opt-in: + +```rust +let event_type_filter = Some(EventTypeFilter::include_only(vec![ + EventType::PumpFunBuy, + EventType::TransactionCost, +])); + +let callback = |event: DexEvent| { + if let DexEvent::TransactionCostEvent(cost) = event { + println!("tips: {:?}", cost.tip_payments); + } +}; +``` + +`None`, `EventTypeFilter::default()`, and `EventTypeFilter::all()` do not enable transaction-cost +scanning. Each tip payment includes the matching SWQoS provider from the complete address registry +shared with `sol-trade-sdk`. + #### Performance Impact Event filtering can provide significant performance improvements: diff --git a/src/streaming/common/event_processor.rs b/src/streaming/common/event_processor.rs index 95d8226..fcc2a10 100644 --- a/src/streaming/common/event_processor.rs +++ b/src/streaming/common/event_processor.rs @@ -2,10 +2,11 @@ use crate::common::AnyResult; use crate::streaming::common::MetricsEventType; use crate::streaming::event_parser::common::filter::{ build_sdk_parse_event_filter, build_sdk_shred_parse_event_filter, passes_event_type_filter, - EventTypeFilter, + transaction_cost_selection, EventTypeFilter, }; use crate::streaming::event_parser::common::high_performance_clock::elapsed_micros_since; use crate::streaming::event_parser::core::common_event_parser::CommonEventParser; +use crate::streaming::event_parser::core::transaction_cost_event::TransactionCostEvent; use crate::streaming::event_parser::{core::traits::DexEvent, Protocol}; use crate::streaming::grpc::{EventPretty, MetricsManager}; use crate::streaming::parser_sdk_bridge::{ @@ -69,21 +70,46 @@ pub fn parse_grpc_transaction_events( let recv_us = transaction_pretty.recv_us; let grpc_tx = transaction_pretty.grpc_tx; let block_time_us = block_time.map(|t| t.seconds * 1_000_000 + t.nanos as i64 / 1_000); - let update = SubscribeUpdateTransaction { slot, transaction: Some(grpc_tx) }; - let sdk_parse_filter = build_sdk_parse_event_filter(event_type_filter); - let sdk_events = parse_subscribe_update_transaction_low_latency( - &update, - recv_us, - block_time_us, - sdk_parse_filter.as_ref(), - ); - let mut events = adapt_parser_events_list( - sdk_events, - block_time.as_ref(), - recv_us, - protocols, - event_type_filter, - ); + let cost_selection = transaction_cost_selection(event_type_filter); + let transaction_cost = cost_selection + .requested + .then(|| { + let transaction = grpc_tx.transaction.as_ref()?; + let meta = grpc_tx.meta.as_ref()?; + sol_parser_sdk::parse_yellowstone_transaction_cost(transaction, meta) + }) + .flatten(); + let mut events = if cost_selection.only { + Vec::with_capacity(1) + } else { + let update = SubscribeUpdateTransaction { slot, transaction: Some(grpc_tx) }; + let sdk_parse_filter = build_sdk_parse_event_filter(event_type_filter); + let sdk_events = parse_subscribe_update_transaction_low_latency( + &update, + recv_us, + block_time_us, + sdk_parse_filter.as_ref(), + ); + adapt_parser_events_list( + sdk_events, + block_time.as_ref(), + recv_us, + protocols, + event_type_filter, + ) + }; + + if let Some(cost) = transaction_cost { + events.push(DexEvent::TransactionCostEvent(TransactionCostEvent::from_parser( + cost, + transaction_pretty.signature, + slot, + transaction_pretty.tx_index, + block_time_us, + recv_us, + None, + ))); + } for event in events.iter_mut() { event.metadata_mut().handle_us = elapsed_micros_since(recv_us); @@ -216,6 +242,27 @@ pub(crate) fn parse_shred_transaction_events( bot_wallet: Option, mut on_event: impl FnMut(DexEvent), ) { + let cost_selection = transaction_cost_selection(event_type_filter); + if cost_selection.requested { + let cost = sol_parser_sdk::parse_shred_transaction_cost(tx); + let recent_blockhash = Some(tx.message.recent_blockhash().to_string()); + let mut event = DexEvent::TransactionCostEvent(TransactionCostEvent::from_parser( + cost, + signature, + slot, + tx_index, + None, + recv_us, + recent_blockhash, + )); + event.metadata_mut().handle_us = elapsed_micros_since(recv_us); + on_event(event); + } + + if cost_selection.only { + return; + } + let sdk_parse_filter = build_sdk_shred_parse_event_filter(protocols, event_type_filter); SHRED_SDK_EVENTS.with(|slot_events| { @@ -277,10 +324,17 @@ fn update_metrics_with_latency( #[cfg(test)] mod tests { - use super::process_shred_transaction; + use super::{parse_grpc_transaction_events, process_shred_transaction}; + use crate::streaming::event_parser::common::filter::EventTypeFilter; + use crate::streaming::event_parser::common::EventType; use crate::streaming::event_parser::{DexEvent, Protocol}; + use crate::streaming::grpc::TransactionPretty; use crate::streaming::shred::TransactionWithSlot; - use sol_parser_sdk::instr::program_ids::PUMPFUN_PROGRAM_ID; + use sol_parser_sdk::{ + instr::program_ids::PUMPFUN_PROGRAM_ID, + transaction_cost::{GLAIVE_TIP_ACCOUNTS, JITO_TIP_ACCOUNTS}, + SwqosProvider, + }; use solana_sdk::hash::Hash; use solana_sdk::message::{ compiled_instruction::CompiledInstruction, v0, MessageHeader, VersionedMessage, @@ -289,6 +343,9 @@ mod tests { use solana_sdk::signature::Signature; use solana_sdk::transaction::VersionedTransaction; use std::sync::{Arc, Mutex}; + use yellowstone_grpc_proto::prelude::{ + CompiledInstruction as GrpcInstruction, Message, Transaction, TransactionStatusMeta, + }; fn push_string(data: &mut Vec, value: &str) { data.extend_from_slice(&(value.len() as u32).to_le_bytes()); @@ -329,6 +386,68 @@ mod tests { } } + fn transfer(lamports: u64) -> Vec { + let mut data = 2u32.to_le_bytes().to_vec(); + data.extend_from_slice(&lamports.to_le_bytes()); + data + } + + fn cost_only_shred_tx() -> VersionedTransaction { + let source = Pubkey::new_unique(); + let system_program: Pubkey = "11111111111111111111111111111111".parse().unwrap(); + VersionedTransaction { + signatures: vec![Signature::default()], + message: VersionedMessage::V0(v0::Message { + header: MessageHeader::default(), + account_keys: vec![ + source, + system_program, + JITO_TIP_ACCOUNTS[0], + GLAIVE_TIP_ACCOUNTS[0], + ], + recent_blockhash: Hash::default(), + instructions: vec![ + CompiledInstruction::new_from_raw_parts(1, transfer(10), vec![0, 2]), + CompiledInstruction::new_from_raw_parts(1, transfer(20), vec![0, 3]), + ], + address_table_lookups: Vec::new(), + }), + } + } + + fn yellowstone_cost_transaction() -> TransactionPretty { + let source = Pubkey::new_unique(); + let system_program: Pubkey = "11111111111111111111111111111111".parse().unwrap(); + TransactionPretty { + slot: 42, + tx_index: Some(7), + signature: Signature::default(), + recv_us: 1_000_000, + grpc_tx: yellowstone_grpc_proto::geyser::SubscribeUpdateTransactionInfo { + signature: Signature::default().as_ref().to_vec(), + is_vote: false, + transaction: Some(Transaction { + signatures: vec![Signature::default().as_ref().to_vec()], + message: Some(Message { + account_keys: [source, system_program, JITO_TIP_ACCOUNTS[0]] + .iter() + .map(|key| key.to_bytes().to_vec()) + .collect(), + instructions: vec![GrpcInstruction { + program_id_index: 1, + accounts: vec![0, 2], + data: transfer(50), + }], + ..Default::default() + }), + }), + meta: Some(TransactionStatusMeta { fee: 5_000, ..Default::default() }), + index: 7, + }, + ..Default::default() + } + } + async fn parse_single_shred_create(tx_index: Option) -> DexEvent { let events = Arc::new(Mutex::new(Vec::new())); let captured = events.clone(); @@ -366,4 +485,52 @@ mod tests { assert!(matches!(event, DexEvent::PumpFunCreateTokenEvent(_))); assert_eq!(event.metadata().tx_index, Some(42)); } + + #[tokio::test] + async fn shred_cost_parsing_is_opt_in_and_provider_aware() { + let events = Arc::new(Mutex::new(Vec::new())); + let captured = events.clone(); + let callback = Arc::new(move |event| captured.lock().unwrap().push(event)); + let filter = EventTypeFilter::include_only([EventType::TransactionCost]); + + process_shred_transaction( + TransactionWithSlot::new(cost_only_shred_tx(), 42, 1_000_000, Some(7)), + &[], + Some(&filter), + callback, + None, + ) + .await + .expect("process cost transaction"); + + let events = events.lock().unwrap(); + assert_eq!(events.len(), 1); + let DexEvent::TransactionCostEvent(cost) = &events[0] else { + panic!("expected transaction cost event"); + }; + assert_eq!(cost.tip_lamports, 30); + assert_eq!(cost.tip_lamports_for(SwqosProvider::Jito), 10); + assert_eq!(cost.tip_lamports_for(SwqosProvider::Glaive), 20); + assert!(!cost.tip_payments_confirmed); + assert_eq!(cost.metadata.tx_index, Some(7)); + } + + #[test] + fn yellowstone_cost_parsing_is_opt_in() { + let without_cost = + parse_grpc_transaction_events(yellowstone_cost_transaction(), &[], None, None); + assert!(without_cost.is_empty()); + + let filter = EventTypeFilter::include_only([EventType::TransactionCost]); + let with_cost = + parse_grpc_transaction_events(yellowstone_cost_transaction(), &[], Some(&filter), None); + assert_eq!(with_cost.len(), 1); + let DexEvent::TransactionCostEvent(cost) = &with_cost[0] else { + panic!("expected transaction cost event"); + }; + assert_eq!(cost.transaction_fee_lamports, Some(5_000)); + assert_eq!(cost.tip_lamports, 50); + assert!(cost.tip_payments_confirmed); + assert_eq!(cost.metadata.tx_index, Some(7)); + } } diff --git a/src/streaming/event_parser/common/filter.rs b/src/streaming/event_parser/common/filter.rs index 9d44759..389a154 100644 --- a/src/streaming/event_parser/common/filter.rs +++ b/src/streaming/event_parser/common/filter.rs @@ -121,6 +121,9 @@ pub(crate) fn build_sdk_parse_event_filter( } let mut raw: Vec = Vec::with_capacity(f.include.len()); for et in &f.include { + if et == &EventType::TransactionCost { + continue; + } if !push_streamer_event_sdk_grpc_types(et, &mut raw, FilterMapMode::Include) { return None; } @@ -129,6 +132,38 @@ pub(crate) fn build_sdk_parse_event_filter( Some(SdkGrpcEventTypeFilter::include_only(raw)) } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(crate) struct TransactionCostSelection { + pub requested: bool, + pub only: bool, +} + +/// Empty/default filters intentionally do not enable transaction-cost scanning. +#[inline] +pub(crate) fn transaction_cost_selection( + filter: Option<&EventTypeFilter>, +) -> TransactionCostSelection { + let Some(filter) = filter else { + return TransactionCostSelection::default(); + }; + + let mut requested = false; + let mut only = true; + for event in &filter.include { + if event == &EventType::TransactionCost { + requested = true; + } else { + only = false; + } + } + + if !requested || filter.exclude.contains(&EventType::TransactionCost) { + return TransactionCostSelection::default(); + } + + TransactionCostSelection { requested, only } +} + /// Build the SDK ShredStream hot-path filter. Unlike Yellowstone, ShredStream /// subscription itself has no program filter, so protocol narrowing must be /// pushed into the SDK parser as an event-type include list. @@ -513,6 +548,49 @@ mod tests { assert!(f.include_transaction_event()); assert!(f.include_account_event()); assert!(f.include_block_event()); + assert_eq!(transaction_cost_selection(None), TransactionCostSelection::default()); + assert_eq!( + transaction_cost_selection(Some(&EventTypeFilter::all())), + TransactionCostSelection::default() + ); + } + + #[test] + fn transaction_cost_is_strictly_opt_in() { + let cost_only = EventTypeFilter::include_only([EventType::TransactionCost]); + assert_eq!( + transaction_cost_selection(Some(&cost_only)), + TransactionCostSelection { requested: true, only: true } + ); + + let combined = + EventTypeFilter::include_only([EventType::PumpFunBuy, EventType::TransactionCost]); + assert_eq!( + transaction_cost_selection(Some(&combined)), + TransactionCostSelection { requested: true, only: false } + ); + + let excluded = EventTypeFilter::include_exclude( + [EventType::TransactionCost], + [EventType::TransactionCost], + ); + assert_eq!( + transaction_cost_selection(Some(&excluded)), + TransactionCostSelection::default() + ); + } + + #[test] + fn transaction_cost_is_not_forwarded_to_the_dex_filter() { + let cost_only = EventTypeFilter::include_only([EventType::TransactionCost]); + let sdk_filter = build_sdk_parse_event_filter(Some(&cost_only)).expect("empty DEX filter"); + assert!(!sdk_filter.should_include(SdkGrpcEventType::PumpFunBuy)); + + let combined = + EventTypeFilter::include_only([EventType::PumpFunBuy, EventType::TransactionCost]); + let sdk_filter = build_sdk_parse_event_filter(Some(&combined)).expect("PumpFun filter"); + assert!(sdk_filter.should_include(SdkGrpcEventType::PumpFunBuy)); + assert!(!sdk_filter.should_include(SdkGrpcEventType::PumpFunSell)); } #[test] diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index fef0f8f..dc69ee9 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -219,6 +219,8 @@ pub enum EventType { SetComputeUnitPrice, ParserSdkError, Unknown, + // Appended to preserve existing Borsh enum discriminants. + TransactionCost, } pub const ACCOUNT_EVENT_TYPES: &[EventType] = &[ diff --git a/src/streaming/event_parser/core/event_parser/mod.rs b/src/streaming/event_parser/core/event_parser/mod.rs index 022132e..76f7b75 100644 --- a/src/streaming/event_parser/core/event_parser/mod.rs +++ b/src/streaming/event_parser/core/event_parser/mod.rs @@ -30,36 +30,68 @@ impl EventParser { } let block_us = block_time.map(|t| t.seconds * 1_000_000 + t.nanos as i64 / 1_000); + let cost_selection = + crate::streaming::event_parser::common::filter::transaction_cost_selection( + event_type_filter, + ); + let transaction_cost = cost_selection + .requested + .then(|| { + let transaction = grpc_tx.transaction.as_ref()?; + let meta = grpc_tx.meta.as_ref()?; + sol_parser_sdk::parse_yellowstone_transaction_cost(transaction, meta) + }) + .flatten(); let update = yellowstone_grpc_proto::geyser::SubscribeUpdateTransaction { slot: slot.unwrap_or(0), transaction: Some(grpc_tx), }; - 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, - 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(), + if !cost_selection.only { + 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, - 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); + 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); + } } } + if let Some(cost) = transaction_cost { + let mut event = crate::streaming::event_parser::DexEvent::TransactionCostEvent( + crate::streaming::event_parser::core::transaction_cost_event::TransactionCostEvent::from_parser( + cost, + signature, + slot.unwrap_or(0), + tx_index, + block_us, + recv_us, + None, + ), + ); + event.metadata_mut().handle_us = + crate::streaming::event_parser::common::high_performance_clock::elapsed_micros_since( + recv_us, + ); + callback(event); + } Ok(()) } diff --git a/src/streaming/event_parser/core/mod.rs b/src/streaming/event_parser/core/mod.rs index 9dba272..c7a733f 100755 --- a/src/streaming/event_parser/core/mod.rs +++ b/src/streaming/event_parser/core/mod.rs @@ -4,6 +4,7 @@ pub mod dispatcher; pub mod global_state; pub mod parser_cache; pub mod traits; +pub mod transaction_cost_event; pub use dispatcher::EventDispatcher; pub use traits::DexEvent; diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index f0749e4..d008ba7 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -5,6 +5,7 @@ use crate::streaming::event_parser::core::account_event_parser::{ use crate::streaming::event_parser::core::common_event_parser::{ SetComputeUnitLimitEvent, SetComputeUnitPriceEvent, }; +use crate::streaming::event_parser::core::transaction_cost_event::TransactionCostEvent; use crate::streaming::event_parser::protocols::block::block_meta_event::BlockMetaEvent; use crate::streaming::event_parser::protocols::bonk::events::*; use crate::streaming::event_parser::protocols::meteora_damm_v2::events::*; @@ -162,6 +163,8 @@ pub enum DexEvent { SetComputeUnitLimitEvent(SetComputeUnitLimitEvent), SetComputeUnitPriceEvent(SetComputeUnitPriceEvent), ParserSdkErrorEvent(ParserSdkErrorEvent), + // Appended to preserve existing serialized enum variant indices. + TransactionCostEvent(TransactionCostEvent), } /// Macro to generate metadata accessors for all DexEvent variants @@ -301,4 +304,5 @@ impl_dex_event_metadata!( SetComputeUnitLimitEvent, SetComputeUnitPriceEvent, ParserSdkErrorEvent, + TransactionCostEvent, ); diff --git a/src/streaming/event_parser/core/transaction_cost_event.rs b/src/streaming/event_parser/core/transaction_cost_event.rs new file mode 100644 index 0000000..6b811be --- /dev/null +++ b/src/streaming/event_parser/core/transaction_cost_event.rs @@ -0,0 +1,68 @@ +use crate::streaming::event_parser::common::{EventMetadata, EventType, ProtocolType}; +use serde::{Deserialize, Serialize}; +use sol_parser_sdk::TransactionCost; +pub use sol_parser_sdk::{SwqosProvider, TipPayment}; +use solana_sdk::{pubkey::Pubkey, signature::Signature}; + +/// Opt-in transaction fee, compute-budget, and SWQoS tip details. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct TransactionCostEvent { + pub metadata: EventMetadata, + pub transaction_fee_lamports: Option, + pub total_fee_and_tip_lamports: Option, + pub compute_units_consumed: Option, + pub compute_unit_limit: Option, + pub compute_unit_price_micro_lamports: Option, + pub priority_fee_lamports: Option, + pub tip_payments_confirmed: bool, + pub tip_lamports: u64, + pub tip_payments: Vec, +} + +impl TransactionCostEvent { + #[allow(clippy::too_many_arguments)] + pub(crate) fn from_parser( + cost: TransactionCost, + signature: Signature, + slot: u64, + tx_index: Option, + block_time_us: Option, + recv_us: i64, + recent_blockhash: Option, + ) -> Self { + let block_time_us = block_time_us.unwrap_or_default(); + Self { + metadata: EventMetadata::new( + signature, + slot, + block_time_us / 1_000_000, + block_time_us / 1_000, + ProtocolType::Common, + EventType::TransactionCost, + Pubkey::default(), + 0, + None, + recv_us, + tx_index, + recent_blockhash, + ), + transaction_fee_lamports: cost.transaction_fee_lamports, + total_fee_and_tip_lamports: cost.total_fee_and_tip_lamports, + compute_units_consumed: cost.compute_units_consumed, + compute_unit_limit: cost.compute_unit_limit, + compute_unit_price_micro_lamports: cost.compute_unit_price_micro_lamports, + priority_fee_lamports: cost.priority_fee_lamports, + tip_payments_confirmed: cost.tip_payments_confirmed, + tip_lamports: cost.tip_lamports, + tip_payments: cost.tip_payments, + } + } + + #[inline] + pub fn tip_lamports_for(&self, provider: SwqosProvider) -> u64 { + self.tip_payments + .iter() + .filter(|payment| payment.provider == provider) + .fold(0u64, |total, payment| total.saturating_add(payment.lamports)) + } +} diff --git a/src/streaming/parser_sdk_bridge/filter.rs b/src/streaming/parser_sdk_bridge/filter.rs index e69c020..44f6793 100644 --- a/src/streaming/parser_sdk_bridge/filter.rs +++ b/src/streaming/parser_sdk_bridge/filter.rs @@ -21,6 +21,7 @@ fn is_protocol_independent_event(ev: &DexEvent) -> bool { | DexEvent::BlockMetaEvent(_) | DexEvent::SetComputeUnitLimitEvent(_) | DexEvent::SetComputeUnitPriceEvent(_) + | DexEvent::TransactionCostEvent(_) | DexEvent::ParserSdkErrorEvent(_) ) } diff --git a/src/streaming/rpc_parse.rs b/src/streaming/rpc_parse.rs index 1b846c7..ce40faa 100644 --- a/src/streaming/rpc_parse.rs +++ b/src/streaming/rpc_parse.rs @@ -16,8 +16,9 @@ use solana_sdk::signature::Signature; use solana_transaction_status::{EncodedConfirmedTransactionWithStatusMeta, UiTransactionEncoding}; use crate::streaming::event_parser::common::filter::{ - build_sdk_parse_event_filter, EventTypeFilter, + build_sdk_parse_event_filter, transaction_cost_selection, EventTypeFilter, }; +use crate::streaming::event_parser::core::transaction_cost_event::TransactionCostEvent; use crate::streaming::event_parser::{DexEvent, Protocol}; use crate::streaming::parser_sdk_bridge::adapt_parser_events_list; pub use sol_parser_sdk::ParseError; @@ -31,16 +32,41 @@ pub fn parse_encoded_rpc_transaction_as_streamer_events( protocols: &[Protocol], event_type_filter: Option<&EventTypeFilter>, ) -> Result, ParseError> { - let sdk_filter = build_sdk_parse_event_filter(event_type_filter); - let pb_events = parse_rpc_transaction(rpc_tx, sdk_filter.as_ref())?; let block_ts = rpc_tx.block_time.map(|sec| Timestamp { seconds: sec, nanos: 0 }); - Ok(adapt_parser_events_list( - pb_events, - block_ts.as_ref(), - recv_wall_us, - protocols, - event_type_filter, - )) + let cost_selection = transaction_cost_selection(event_type_filter); + let mut events = if cost_selection.only { + Vec::with_capacity(1) + } else { + let sdk_filter = build_sdk_parse_event_filter(event_type_filter); + let pb_events = parse_rpc_transaction(rpc_tx, sdk_filter.as_ref())?; + adapt_parser_events_list( + pb_events, + block_ts.as_ref(), + recv_wall_us, + protocols, + event_type_filter, + ) + }; + if cost_selection.requested { + let cost = sol_parser_sdk::parse_rpc_transaction_cost(rpc_tx)?; + let signature = rpc_tx + .transaction + .transaction + .decode() + .and_then(|transaction| transaction.signatures.first().copied()) + .ok_or_else(|| ParseError::MissingField("transaction.signatures[0]".to_string()))?; + let block_time_us = rpc_tx.block_time.map(|seconds| seconds * 1_000_000); + events.push(DexEvent::TransactionCostEvent(TransactionCostEvent::from_parser( + cost, + signature, + rpc_tx.slot, + None, + block_time_us, + recv_wall_us, + None, + ))); + } + Ok(events) } /// Blocking RPC fetch by signature, then adapt SDK events to streamer events. @@ -51,6 +77,22 @@ pub fn fetch_rpc_transaction_as_streamer_events( protocols: &[Protocol], event_type_filter: Option<&EventTypeFilter>, ) -> Result, ParseError> { + if transaction_cost_selection(event_type_filter).requested { + let config = RpcTransactionConfig { + encoding: Some(UiTransactionEncoding::Base64), + commitment: None, + max_supported_transaction_version: Some(0), + }; + let rpc_tx = rpc_client + .get_transaction_with_config(signature, config) + .map_err(|error| map_async_rpc_err(error.to_string()))?; + return parse_encoded_rpc_transaction_as_streamer_events( + &rpc_tx, + recv_wall_us, + protocols, + event_type_filter, + ); + } let sdk_filter = build_sdk_parse_event_filter(event_type_filter); let pb_events = parse_transaction_from_rpc(rpc_client, signature, sdk_filter.as_ref())?; // The SDK already writes block_time_us into each event; adapter falls back to it when @@ -96,3 +138,49 @@ fn map_async_rpc_err(msg: String) -> ParseError { ParseError::RpcError(msg) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::streaming::event_parser::common::filter::EventTypeFilter; + use crate::streaming::event_parser::common::EventType; + use sol_parser_sdk::SwqosProvider; + use solana_client::rpc_client::RpcClient; + use std::str::FromStr; + + #[test] + fn current_mainnet_transaction_cost_is_reusable() { + if std::env::var_os("RUN_MAINNET_TESTS").is_none() { + return; + } + const SIGNATURE: &str = + "4yaaD6ywu8epxVTvZEDAGPhdKK2V73XqvLqQWm1KbSFQ1uTk2nnC4uW7xTrpSuQYpTivmDQQawu7x3dFbYC1KuZ6"; + let rpc_url = std::env::var("SOLANA_RPC_URL") + .unwrap_or_else(|_| "https://api.mainnet-beta.solana.com".to_string()); + let signature = Signature::from_str(SIGNATURE).expect("valid fixture signature"); + let filter = EventTypeFilter::include_only([EventType::TransactionCost]); + let events = fetch_rpc_transaction_as_streamer_events( + &RpcClient::new(rpc_url), + &signature, + 0, + &[], + Some(&filter), + ) + .expect("parse current transaction cost"); + + assert_eq!(events.len(), 1); + let DexEvent::TransactionCostEvent(cost) = &events[0] else { + panic!("expected transaction cost event"); + }; + assert_eq!(cost.metadata.slot, 438_900_232); + assert_eq!(cost.metadata.signature, signature); + assert_eq!(cost.transaction_fee_lamports, Some(29_242)); + assert_eq!(cost.compute_units_consumed, Some(135_026)); + assert_eq!(cost.compute_unit_limit, Some(300_000)); + assert_eq!(cost.compute_unit_price_micro_lamports, Some(80_805)); + assert_eq!(cost.priority_fee_lamports, Some(24_242)); + assert_eq!(cost.tip_lamports, 137_273); + assert_eq!(cost.total_fee_and_tip_lamports, Some(166_515)); + assert_eq!(cost.tip_lamports_for(SwqosProvider::Jito), 137_273); + } +}