Merge pull request #76: add opt-in transaction cost events

Add opt-in transaction cost events

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 = { version = "0.6.3", default-features = false }
 solana-sdk = "3.0.0"
 solana-client = "3.1.12"
 solana-transaction-status = "3.1.12"

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:

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<Pubkey>,
     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<u8>, value: &str) {
         data.extend_from_slice(&(value.len() as u32).to_le_bytes());
@@ -329,6 +386,68 @@ mod tests {
         }
     }
 
+    fn transfer(lamports: u64) -> Vec<u8> {
+        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<u64>) -> 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));
+    }
 }

src/streaming/event_parser/common/filter.rs

@@ -121,6 +121,9 @@ pub(crate) fn build_sdk_parse_event_filter(
     }
     let mut raw: Vec<SdkGrpcEventType> = 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]

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] = &[

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(())
     }
 

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;

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,
 );

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<u64>,
+    pub total_fee_and_tip_lamports: Option<u64>,
+    pub compute_units_consumed: Option<u64>,
+    pub compute_unit_limit: Option<u32>,
+    pub compute_unit_price_micro_lamports: Option<u64>,
+    pub priority_fee_lamports: Option<u64>,
+    pub tip_payments_confirmed: bool,
+    pub tip_lamports: u64,
+    pub tip_payments: Vec<TipPayment>,
+}
+
+impl TransactionCostEvent {
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) fn from_parser(
+        cost: TransactionCost,
+        signature: Signature,
+        slot: u64,
+        tx_index: Option<u64>,
+        block_time_us: Option<i64>,
+        recv_us: i64,
+        recent_blockhash: Option<String>,
+    ) -> 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))
+    }
+}

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(_)
     )
 }

src/streaming/parser_sdk_bridge/mod.rs

@@ -508,6 +508,16 @@ mod tests {
             is_buy: true,
             trade_direction: PbBonkDir::Buy,
             exact_in: true,
+            global_config: Pubkey::default(),
+            platform_config: Pubkey::default(),
+            user_base_token: Pubkey::default(),
+            user_quote_token: Pubkey::default(),
+            base_vault: Pubkey::default(),
+            quote_vault: Pubkey::default(),
+            base_mint: Pubkey::default(),
+            quote_mint: Pubkey::default(),
+            base_token_program: Pubkey::default(),
+            quote_token_program: Pubkey::default(),
         };
         let dex =
             convert_parser_event(PbDexEvent::RaydiumLaunchlabTrade(b), None, 0).expect("convert");
@@ -530,6 +540,16 @@ mod tests {
             is_buy: false,
             trade_direction: PbBonkDir::Sell,
             exact_in: false,
+            global_config: Pubkey::default(),
+            platform_config: Pubkey::default(),
+            user_base_token: Pubkey::default(),
+            user_quote_token: Pubkey::default(),
+            base_vault: Pubkey::default(),
+            quote_vault: Pubkey::default(),
+            base_mint: Pubkey::default(),
+            quote_mint: Pubkey::default(),
+            base_token_program: Pubkey::default(),
+            quote_token_program: Pubkey::default(),
         };
         let dex =
             convert_parser_event(PbDexEvent::RaydiumLaunchlabTrade(b), None, 0).expect("convert");

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<Vec<DexEvent>, 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<Vec<DexEvent>, 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);
+    }
+}