From 02c8a6d168835914c4050c0ab8c04fa74a54b99f Mon Sep 17 00:00:00 2001 From: 0xfnzero <0xfnzero@users.noreply.github.com> Date: Tue, 26 May 2026 21:23:27 +0800 Subject: [PATCH] Release solana-streamer 1.5.2 --- Cargo.toml | 6 +- README.md | 14 +- README_CN.md | 14 +- src/streaming/common/event_processor.rs | 172 ++++++++++++++++-- src/streaming/common/subscription.rs | 16 +- src/streaming/event_parser/common/filter.rs | 32 +++- .../event_parser/core/event_parser/mod.rs | 34 +--- src/streaming/shred_stream.rs | 44 ++--- 8 files changed, 245 insertions(+), 87 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index c062ab6..54efe07 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "solana-streamer-sdk" -version = "1.5.1" +version = "1.5.2" edition = "2021" authors = ["William ", "sgxiang ", "wei <1415121722@qq.com>"] repository = "https://github.com/0xfnzero/solana-streamer" @@ -15,13 +15,13 @@ crate-type = ["cdylib", "rlib"] [features] default = ["sdk-parse-borsh"] sdk-parse-borsh = ["sol-parser-sdk/parse-borsh"] -# If both parser backend features are enabled, sol-parser-sdk 0.5.1+ uses zero-copy. +# If both parser backend features are enabled, sol-parser-sdk 0.5.2+ uses zero-copy. sdk-parse-zero-copy = ["sol-parser-sdk/parse-zero-copy"] sdk-perf-stats = ["sol-parser-sdk/perf-stats"] sdk-ultra-perf = ["sol-parser-sdk/ultra-perf"] [dependencies] -sol-parser-sdk = { version = "0.5.1", default-features = false } +sol-parser-sdk = { version = "0.5.2", 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 5c26760..79b3bc2 100755 --- a/README.md +++ b/README.md @@ -123,30 +123,34 @@ Add the dependency to your `Cargo.toml`: ```toml # Add to your Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "1.5.1" } +solana-streamer-sdk = { path = "./solana-streamer", version = "1.5.2" } ``` ### Use crates.io ```toml # Add to your Cargo.toml -solana-streamer-sdk = "1.5.1" +solana-streamer-sdk = "1.5.2" ``` Parser backend features: ```toml # Default: sol-parser-sdk parse-borsh backend -solana-streamer-sdk = "1.5.1" +solana-streamer-sdk = "1.5.2" # Zero-copy parser backend for latency-sensitive bots -solana-streamer-sdk = { version = "1.5.1", default-features = false, features = ["sdk-parse-zero-copy"] } +solana-streamer-sdk = { version = "1.5.2", default-features = false, features = ["sdk-parse-zero-copy"] } ``` -If both `sdk-parse-borsh` and `sdk-parse-zero-copy` are enabled, `sol-parser-sdk 0.5.1+` uses the zero-copy backend. +If both `sdk-parse-borsh` and `sdk-parse-zero-copy` are enabled, `sol-parser-sdk 0.5.2+` uses the zero-copy backend. ## 🔄 Migration Guide +### Upgrading to v1.5.2 + +Version 1.5.2 uses `sol-parser-sdk 0.5.2` from crates.io. ShredStream delivery now uses the SDK direct-callback path to avoid the extra queue consumer task, reuses parser event buffers across entries, preserves unknown ShredStream `tx_index` as `None`, and keeps `PumpFunCreateToken` filters backward-compatible with `CreateV2` events. User callbacks on the direct path should avoid blocking work. + ### Upgrading to v1.5.1 Version 1.5.1 uses `sol-parser-sdk 0.5.1` from crates.io. It improves ShredStream ALT handling by parsing outer instructions best-effort with default placeholders for ALT-loaded accounts, adds dropped-event queue observability, and keeps CPI/inner-only events documented as a ShredStream limitation. diff --git a/README_CN.md b/README_CN.md index 4b6cad0..1d3403a 100644 --- a/README_CN.md +++ b/README_CN.md @@ -122,30 +122,34 @@ git clone https://github.com/0xfnzero/solana-streamer ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "1.5.1" } +solana-streamer-sdk = { path = "./solana-streamer", version = "1.5.2" } ``` ### 使用 crates.io ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = "1.5.1" +solana-streamer-sdk = "1.5.2" ``` 解析后端 feature： ```toml # 默认：sol-parser-sdk parse-borsh 后端 -solana-streamer-sdk = "1.5.1" +solana-streamer-sdk = "1.5.2" # 面向低延迟 Bot 的 zero-copy 解析后端 -solana-streamer-sdk = { version = "1.5.1", default-features = false, features = ["sdk-parse-zero-copy"] } +solana-streamer-sdk = { version = "1.5.2", default-features = false, features = ["sdk-parse-zero-copy"] } ``` -如果同时启用 `sdk-parse-borsh` 和 `sdk-parse-zero-copy`，`sol-parser-sdk 0.5.1+` 会优先使用 zero-copy 后端。 +如果同时启用 `sdk-parse-borsh` 和 `sdk-parse-zero-copy`，`sol-parser-sdk 0.5.2+` 会优先使用 zero-copy 后端。 ## 🔄 迁移指南 +### 升级到 v1.5.2 + +v1.5.2 使用 crates.io 上的 `sol-parser-sdk 0.5.2`。ShredStream 投递改为 SDK direct-callback 路径，避免额外的队列消费任务；解析事件缓冲会跨 Entry 复用；ShredStream 拿不到的 `tx_index` 会保留为 `None`；`PumpFunCreateToken` 过滤继续向后兼容 `CreateV2` 事件。direct-callback 路径中的用户回调应避免阻塞操作。 + ### 升级到 v1.5.1 v1.5.1 使用 crates.io 上的 `sol-parser-sdk 0.5.1`。该版本改进 ShredStream 的 ALT 处理：ALT-loaded 账户会以默认账户占位，外层指令按 data/discriminator 尽量解析；同时增加队列满时的 dropped event 可观测性，并明确 CPI/inner-only 事件仍是 ShredStream 的固有限制。 diff --git a/src/streaming/common/event_processor.rs b/src/streaming/common/event_processor.rs index 48d72dd..132a4d2 100644 --- a/src/streaming/common/event_processor.rs +++ b/src/streaming/common/event_processor.rs @@ -15,9 +15,17 @@ use crate::streaming::parser_sdk_bridge::{ use crate::streaming::shred::TransactionWithSlot; use sol_parser_sdk::grpc::parse_subscribe_update_transaction_low_latency; use solana_sdk::pubkey::Pubkey; +use solana_sdk::signature::Signature; +use solana_sdk::transaction::VersionedTransaction; +use std::cell::RefCell; use std::sync::Arc; use yellowstone_grpc_proto::geyser::SubscribeUpdateTransaction; +thread_local! { + static SHRED_SDK_EVENTS: RefCell> = + RefCell::new(Vec::with_capacity(4)); +} + /// Wrap the user callback and update transaction metrics after delivery. #[inline] fn create_metrics_callback( @@ -182,33 +190,68 @@ pub async fn process_shred_transaction( let recv_us = transaction_with_slot.recv_us; let adapter_callback = create_metrics_callback(callback); - let sdk_parse_filter = build_sdk_shred_parse_event_filter(protocols, event_type_filter); - let mut sdk_events = Vec::with_capacity(4); - sol_parser_sdk::shredstream::parse_transaction_dex_events_with_filter( + parse_shred_transaction_events( &tx, signature, slot, - tx_index.unwrap_or(0), + tx_index, recv_us, - sdk_parse_filter.as_ref(), - &mut sdk_events, + protocols, + event_type_filter, + bot_wallet, + |event| adapter_callback(event), ); - for sdk_event in sdk_events { - if let Some(mut event) = - adapt_parser_event(sdk_event, None, recv_us, protocols, event_type_filter) - { - event.metadata_mut().handle_us = elapsed_micros_since(recv_us); - event = crate::streaming::event_parser::core::event_parser::helpers::process_event( - event, bot_wallet, - ); - adapter_callback(event); - } - } - Ok(()) } +#[allow(clippy::too_many_arguments)] +pub(crate) fn parse_shred_transaction_events( + tx: &VersionedTransaction, + signature: Signature, + slot: u64, + tx_index: Option, + recv_us: i64, + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, + bot_wallet: Option, + mut on_event: impl FnMut(DexEvent), +) { + let sdk_parse_filter = build_sdk_shred_parse_event_filter(protocols, event_type_filter); + + SHRED_SDK_EVENTS.with(|slot_events| { + let mut events = { + let mut slot_events = slot_events.borrow_mut(); + std::mem::take(&mut *slot_events) + }; + events.clear(); + sol_parser_sdk::shredstream::parse_transaction_dex_events_with_filter( + tx, + signature, + slot, + tx_index.unwrap_or(0), + recv_us, + sdk_parse_filter.as_ref(), + &mut events, + ); + + for sdk_event in events.drain(..) { + if let Some(mut event) = + adapt_parser_event(sdk_event, None, recv_us, protocols, event_type_filter) + { + event.metadata_mut().tx_index = tx_index; + event.metadata_mut().handle_us = elapsed_micros_since(recv_us); + event = crate::streaming::event_parser::core::event_parser::helpers::process_event( + event, bot_wallet, + ); + on_event(event); + } + } + + *slot_events.borrow_mut() = events; + }); +} + /// Update metrics for event processing (with optional latency check) #[inline] fn update_metrics(ty: MetricsEventType, count: u64, time_us: f64) { @@ -232,3 +275,96 @@ fn update_metrics_with_latency( block_time_ms, ); } + +#[cfg(test)] +mod tests { + use super::process_shred_transaction; + use crate::streaming::event_parser::{DexEvent, Protocol}; + use crate::streaming::shred::TransactionWithSlot; + use sol_parser_sdk::instr::program_ids::PUMPFUN_PROGRAM_ID; + use solana_sdk::hash::Hash; + use solana_sdk::message::{ + compiled_instruction::CompiledInstruction, v0, MessageHeader, VersionedMessage, + }; + use solana_sdk::pubkey::Pubkey; + use solana_sdk::signature::Signature; + use solana_sdk::transaction::VersionedTransaction; + use std::sync::{Arc, Mutex}; + + fn push_string(data: &mut Vec, value: &str) { + data.extend_from_slice(&(value.len() as u32).to_le_bytes()); + data.extend_from_slice(value.as_bytes()); + } + + fn pumpfun_create_data() -> Vec { + let mut data = Vec::new(); + data.extend_from_slice(&[24, 30, 200, 40, 5, 28, 7, 119]); + push_string(&mut data, "Index Test"); + push_string(&mut data, "IDX"); + push_string(&mut data, "https://example.invalid/index.json"); + data.extend_from_slice(Pubkey::new_unique().as_ref()); + data + } + + fn pumpfun_create_tx() -> VersionedTransaction { + let mut account_keys = (0..10).map(|_| Pubkey::new_unique()).collect::>(); + account_keys.push(PUMPFUN_PROGRAM_ID); + + VersionedTransaction { + signatures: vec![Signature::default()], + message: VersionedMessage::V0(v0::Message { + header: MessageHeader { + num_required_signatures: 1, + num_readonly_signed_accounts: 0, + num_readonly_unsigned_accounts: 0, + }, + account_keys, + recent_blockhash: Hash::default(), + instructions: vec![CompiledInstruction::new_from_raw_parts( + 10, + pumpfun_create_data(), + (0..10).collect(), + )], + address_table_lookups: Vec::new(), + }), + } + } + + async fn parse_single_shred_create(tx_index: Option) -> DexEvent { + let events = Arc::new(Mutex::new(Vec::new())); + let captured = events.clone(); + let callback = Arc::new(move |event| { + captured.lock().unwrap().push(event); + }); + + process_shred_transaction( + TransactionWithSlot::new(pumpfun_create_tx(), 42, 1_000_000, tx_index), + &[Protocol::PumpFun], + None, + callback, + None, + ) + .await + .expect("process shred transaction"); + + let mut events = events.lock().unwrap(); + assert_eq!(events.len(), 1); + events.pop().unwrap() + } + + #[tokio::test] + async fn shred_transaction_preserves_unknown_tx_index() { + let event = parse_single_shred_create(None).await; + + assert!(matches!(event, DexEvent::PumpFunCreateTokenEvent(_))); + assert_eq!(event.metadata().tx_index, None); + } + + #[tokio::test] + async fn shred_transaction_preserves_present_tx_index() { + let event = parse_single_shred_create(Some(42)).await; + + assert!(matches!(event, DexEvent::PumpFunCreateTokenEvent(_))); + assert_eq!(event.metadata().tx_index, Some(42)); + } +} diff --git a/src/streaming/common/subscription.rs b/src/streaming/common/subscription.rs index c0abf16..79284c3 100644 --- a/src/streaming/common/subscription.rs +++ b/src/streaming/common/subscription.rs @@ -2,7 +2,7 @@ use tokio::task::JoinHandle; /// Subscription handle for managing and stopping subscriptions pub struct SubscriptionHandle { - stream_handle: JoinHandle<()>, + stream_handle: Option>, event_handle: Option>, metrics_handle: Option>, } @@ -14,12 +14,18 @@ impl SubscriptionHandle { event_handle: Option>, metrics_handle: Option>, ) -> Self { - Self { stream_handle, event_handle, metrics_handle } + Self { stream_handle: Some(stream_handle), event_handle, metrics_handle } + } + + pub fn metrics_only(metrics_handle: Option>) -> Self { + Self { stream_handle: None, event_handle: None, metrics_handle } } /// Stop subscription and abort all related tasks pub fn stop(self) { - self.stream_handle.abort(); + if let Some(handle) = self.stream_handle { + handle.abort(); + } if let Some(handle) = self.event_handle { handle.abort(); } @@ -30,7 +36,9 @@ impl SubscriptionHandle { /// Asynchronously wait for all tasks to complete pub async fn join(self) -> Result<(), tokio::task::JoinError> { - let _ = self.stream_handle.await; + if let Some(handle) = self.stream_handle { + let _ = handle.await; + } if let Some(handle) = self.event_handle { let _ = handle.await; } diff --git a/src/streaming/event_parser/common/filter.rs b/src/streaming/event_parser/common/filter.rs index 42c193c..d287e4a 100644 --- a/src/streaming/event_parser/common/filter.rs +++ b/src/streaming/event_parser/common/filter.rs @@ -173,7 +173,10 @@ fn push_streamer_event_sdk_grpc_types(t: &EventType, out: &mut Vec out.push(Sdk::BlockMeta), - St::PumpFunCreateToken => out.push(Sdk::PumpFunCreate), + St::PumpFunCreateToken => { + out.push(Sdk::PumpFunCreate); + out.push(Sdk::PumpFunCreateV2); + } St::PumpFunCreateV2Token => out.push(Sdk::PumpFunCreateV2), St::PumpFunBuy => { out.push(Sdk::PumpFunBuy); @@ -396,7 +399,8 @@ fn event_type_matches(filter_type: &EventType, event_type: &EventType) -> bool { filter_type == event_type || matches!( (filter_type, event_type), - (EventType::PumpFunBuy, EventType::PumpFunBuyExactSolIn) + (EventType::PumpFunCreateToken, EventType::PumpFunCreateV2Token) + | (EventType::PumpFunBuy, EventType::PumpFunBuyExactSolIn) | (EventType::TokenAccount, EventType::TokenInfo) | (EventType::RaydiumClmmSwap, EventType::RaydiumClmmSwapV2) | (EventType::RaydiumClmmSwapV2, EventType::RaydiumClmmSwap) @@ -531,6 +535,30 @@ mod tests { assert!(!sdk_f.should_include(SdkGrpcEventType::PumpFunSell)); } + #[test] + fn pumpfun_create_filter_matches_create_v2_for_backward_compat() { + let f = + EventTypeFilter { include: vec![EventType::PumpFunCreateToken], ..Default::default() }; + assert!(f.passes_event_type(&EventType::PumpFunCreateToken)); + assert!(f.passes_event_type(&EventType::PumpFunCreateV2Token)); + + let sdk_f = build_sdk_parse_event_filter(Some(&f)).expect("mapped"); + assert!(sdk_f.should_include(SdkGrpcEventType::PumpFunCreate)); + assert!(sdk_f.should_include(SdkGrpcEventType::PumpFunCreateV2)); + assert!(!sdk_f.should_include(SdkGrpcEventType::PumpFunBuy)); + + let f = EventTypeFilter { + include: vec![EventType::PumpFunCreateV2Token], + ..Default::default() + }; + assert!(!f.passes_event_type(&EventType::PumpFunCreateToken)); + assert!(f.passes_event_type(&EventType::PumpFunCreateV2Token)); + + let sdk_f = build_sdk_parse_event_filter(Some(&f)).expect("mapped"); + assert!(!sdk_f.should_include(SdkGrpcEventType::PumpFunCreate)); + assert!(sdk_f.should_include(SdkGrpcEventType::PumpFunCreateV2)); + } + #[test] fn pumpfun_buy_filter_matches_exact_sol_in_for_backward_compat() { let f = EventTypeFilter { include: vec![EventType::PumpFunBuy], ..Default::default() }; diff --git a/src/streaming/event_parser/core/event_parser/mod.rs b/src/streaming/event_parser/core/event_parser/mod.rs index 60500ff..c00f8e9 100644 --- a/src/streaming/event_parser/core/event_parser/mod.rs +++ b/src/streaming/event_parser/core/event_parser/mod.rs @@ -78,37 +78,19 @@ impl EventParser { tx_index: Option, callback: std::sync::Arc, ) -> anyhow::Result<()> { - let sdk_parse_filter = - crate::streaming::event_parser::common::filter::build_sdk_shred_parse_event_filter( - protocols, - event_type_filter, - ); - let mut sdk_events = Vec::with_capacity(4); - sol_parser_sdk::shredstream::parse_transaction_dex_events_with_filter( + crate::streaming::common::parse_shred_transaction_events( transaction, signature, slot.unwrap_or(0), - tx_index.unwrap_or(0), + tx_index, recv_us, - sdk_parse_filter.as_ref(), - &mut sdk_events, - ); - for sdk_event in sdk_events { - if let Some(mut event) = crate::streaming::parser_sdk_bridge::adapt_parser_event( - sdk_event, - None, - 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); + protocols, + event_type_filter, + bot_wallet, + |event| { callback(event); - } - } + }, + ); Ok(()) } } diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index b6a9ad5..e63ea17 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -17,6 +17,9 @@ use super::ShredStreamGrpc; impl ShredStreamGrpc { /// 订阅ShredStream事件（支持批处理和即时处理） + /// + /// Uses the SDK direct-callback path for minimum latency. User callbacks should avoid + /// blocking work; use the SDK queue API directly when heavy asynchronous processing is needed. pub async fn shredstream_subscribe( &self, protocols: Vec, @@ -30,38 +33,25 @@ impl ShredStreamGrpc { // 如果已有活跃订阅，先停止它 self.stop().await; - let mut metrics_handle = None; - // 启动自动性能监控（如果启用） - if self.config.enable_metrics { - metrics_handle = MetricsManager::global().start_auto_monitoring().await; - } - let sdk_parse_filter = build_sdk_shred_parse_event_filter(&protocols, event_type_filter.as_ref()); - let queue = self - .sdk_shredstream_client - .subscribe_with_filter(sdk_parse_filter) - .await - .map_err(|e| anyhow::anyhow!(e.to_string()))?; // Wrap callback once before the async block let callback = Arc::new(callback); + let callback_for_sdk = callback.clone(); + let protocols_for_sdk = protocols; + let filter_for_sdk = event_type_filter; - let stream_task = tokio::spawn(async move { - loop { - let Some(sdk_event) = queue.pop() else { - tokio::task::yield_now().await; - continue; - }; - + self.sdk_shredstream_client + .subscribe_with_filter_callback(sdk_parse_filter, move |sdk_event| { MetricsManager::global().add_tx_process_count(); let recv_wall_us = sdk_event.metadata().grpc_recv_us; if let Some(mut event) = adapt_parser_event( sdk_event, None, recv_wall_us, - &protocols, - event_type_filter.as_ref(), + &protocols_for_sdk, + filter_for_sdk.as_ref(), ) { event.metadata_mut().handle_us = elapsed_micros_since(event.metadata().recv_us); event = @@ -73,7 +63,7 @@ impl ShredStreamGrpc { let recv_us = metadata.recv_us; let block_time_ms = metadata.block_time_ms; - callback(event); + callback_for_sdk(event); MetricsManager::global().update_metrics_with_latency( MetricsEventType::Transaction, @@ -83,11 +73,17 @@ impl ShredStreamGrpc { block_time_ms, ); } - } - }); + }) + .await + .map_err(|e| anyhow::anyhow!(e.to_string()))?; + + let mut metrics_handle = None; + if self.config.enable_metrics { + metrics_handle = MetricsManager::global().start_auto_monitoring().await; + } // 保存订阅句柄 - let subscription_handle = SubscriptionHandle::new(stream_task, None, metrics_handle); + let subscription_handle = SubscriptionHandle::metrics_only(metrics_handle); let mut handle_guard = self.subscription_handle.lock().await; *handle_guard = Some(subscription_handle);