diff --git a/Cargo.toml b/Cargo.toml index 54efe07..0f7ab27 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "solana-streamer-sdk" -version = "1.5.2" +version = "1.5.3" 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.2+ uses zero-copy. +# If both parser backend features are enabled, sol-parser-sdk 0.5.3+ 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.2", default-features = false } +sol-parser-sdk = { version = "0.5.3", 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 79b3bc2..b8dbf72 100755 --- a/README.md +++ b/README.md @@ -123,33 +123,37 @@ Add the dependency to your `Cargo.toml`: ```toml # Add to your Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "1.5.2" } +solana-streamer-sdk = { path = "./solana-streamer", version = "1.5.3" } ``` ### Use crates.io ```toml # Add to your Cargo.toml -solana-streamer-sdk = "1.5.2" +solana-streamer-sdk = "1.5.3" ``` Parser backend features: ```toml # Default: sol-parser-sdk parse-borsh backend -solana-streamer-sdk = "1.5.2" +solana-streamer-sdk = "1.5.3" # Zero-copy parser backend for latency-sensitive bots -solana-streamer-sdk = { version = "1.5.2", default-features = false, features = ["sdk-parse-zero-copy"] } +solana-streamer-sdk = { version = "1.5.3", 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.2+` uses the zero-copy backend. +If both `sdk-parse-borsh` and `sdk-parse-zero-copy` are enabled, `sol-parser-sdk 0.5.3+` uses the zero-copy backend. ## 🔄 Migration Guide +### Upgrading to v1.5.3 + +Version 1.5.3 uses `sol-parser-sdk 0.5.3` from crates.io. It preserves real Pump.fun v2 `ix_name` values through the streamer bridge, improves ShredStream Pump.fun v2 best-effort parsing for short account lists, and treats `PumpFunBuy` and `PumpFunBuyExactSolIn` subscriptions as compatible buy-family filters. ShredStream still uses the direct entry-reading path with automatic reconnect; callbacks should remain non-blocking and `tx_index` remains entry-local best-effort. + ### 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. +Version 1.5.2 uses `sol-parser-sdk 0.5.2` from crates.io. ShredStream delivery uses the SDK parser on a direct entry-reading path with automatic reconnect, avoiding the extra queue consumer task while keeping parser event buffers reused. ShredStream `tx_index` is entry-local best-effort, not the Yellowstone slot-level transaction index. User callbacks on the direct path run on the read task and should avoid blocking work. ### Upgrading to v1.5.1 diff --git a/README_CN.md b/README_CN.md index 1d3403a..d5c30bd 100644 --- a/README_CN.md +++ b/README_CN.md @@ -122,33 +122,37 @@ git clone https://github.com/0xfnzero/solana-streamer ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "1.5.2" } +solana-streamer-sdk = { path = "./solana-streamer", version = "1.5.3" } ``` ### 使用 crates.io ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = "1.5.2" +solana-streamer-sdk = "1.5.3" ``` 解析后端 feature： ```toml # 默认：sol-parser-sdk parse-borsh 后端 -solana-streamer-sdk = "1.5.2" +solana-streamer-sdk = "1.5.3" # 面向低延迟 Bot 的 zero-copy 解析后端 -solana-streamer-sdk = { version = "1.5.2", default-features = false, features = ["sdk-parse-zero-copy"] } +solana-streamer-sdk = { version = "1.5.3", default-features = false, features = ["sdk-parse-zero-copy"] } ``` -如果同时启用 `sdk-parse-borsh` 和 `sdk-parse-zero-copy`，`sol-parser-sdk 0.5.2+` 会优先使用 zero-copy 后端。 +如果同时启用 `sdk-parse-borsh` 和 `sdk-parse-zero-copy`，`sol-parser-sdk 0.5.3+` 会优先使用 zero-copy 后端。 ## 🔄 迁移指南 +### 升级到 v1.5.3 + +v1.5.3 使用 crates.io 上的 `sol-parser-sdk 0.5.3`。该版本会通过 streamer bridge 保留真实的 Pump.fun v2 `ix_name`，改进 ShredStream 对 Pump.fun v2 短账户列表的 best-effort 解析，并让订阅 `PumpFunBuy` 或 `PumpFunBuyExactSolIn` 都能匹配兼容的 buy-family 事件。ShredStream 仍使用直接读取 Entry 的自动重连低延迟路径；回调应避免阻塞，`tx_index` 仍是 Entry 内 best-effort 索引。 + ### 升级到 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.2 使用 crates.io 上的 `sol-parser-sdk 0.5.2`。ShredStream 投递使用直接读取 Entry 并调用 SDK parser 的低延迟路径，支持自动重连，避免额外的队列消费任务并复用解析事件缓冲。ShredStream 的 `tx_index` 是 Entry 内 best-effort 索引，不是 Yellowstone 的 slot 级交易索引。direct-callback 路径中的用户回调运行在读流任务中，应避免阻塞操作。 ### 升级到 v1.5.1 diff --git a/src/streaming/event_parser/common/filter.rs b/src/streaming/event_parser/common/filter.rs index d287e4a..e137fe5 100644 --- a/src/streaming/event_parser/common/filter.rs +++ b/src/streaming/event_parser/common/filter.rs @@ -110,7 +110,7 @@ pub(crate) fn build_sdk_parse_event_filter( if !f.exclude.is_empty() && f.include.is_empty() { let mut raw: Vec = Vec::with_capacity(f.exclude.len()); for et in &f.exclude { - push_streamer_event_sdk_grpc_types(et, &mut raw); + push_streamer_event_sdk_grpc_types(et, &mut raw, FilterMapMode::Exclude); } dedup_sdk_grpc_event_types(&mut raw); return (!raw.is_empty()).then(|| SdkGrpcEventTypeFilter::exclude_types(raw)); @@ -121,7 +121,7 @@ pub(crate) fn build_sdk_parse_event_filter( } let mut raw: Vec = Vec::with_capacity(f.include.len()); for et in &f.include { - if !push_streamer_event_sdk_grpc_types(et, &mut raw) { + if !push_streamer_event_sdk_grpc_types(et, &mut raw, FilterMapMode::Include) { return None; } } @@ -168,7 +168,17 @@ fn dedup_sdk_grpc_event_types(v: &mut Vec) { } } -fn push_streamer_event_sdk_grpc_types(t: &EventType, out: &mut Vec) -> bool { +#[derive(Clone, Copy)] +enum FilterMapMode { + Include, + Exclude, +} + +fn push_streamer_event_sdk_grpc_types( + t: &EventType, + out: &mut Vec, + mode: FilterMapMode, +) -> bool { use EventType as St; use SdkGrpcEventType as Sdk; match t { @@ -179,11 +189,24 @@ fn push_streamer_event_sdk_grpc_types(t: &EventType, out: &mut Vec out.push(Sdk::PumpFunCreateV2), St::PumpFunBuy => { + if matches!(mode, FilterMapMode::Include) { + out.push(Sdk::PumpFunTrade); + } out.push(Sdk::PumpFunBuy); out.push(Sdk::PumpFunBuyExactSolIn); } - St::PumpFunBuyExactSolIn => out.push(Sdk::PumpFunBuyExactSolIn), - St::PumpFunSell => out.push(Sdk::PumpFunSell), + St::PumpFunBuyExactSolIn => { + if matches!(mode, FilterMapMode::Include) { + out.push(Sdk::PumpFunTrade); + } + out.push(Sdk::PumpFunBuyExactSolIn); + } + St::PumpFunSell => { + if matches!(mode, FilterMapMode::Include) { + out.push(Sdk::PumpFunTrade); + } + out.push(Sdk::PumpFunSell); + } St::PumpFunMigrate => out.push(Sdk::PumpFunMigrate), St::PumpFeesCreateFeeSharingConfig => out.push(Sdk::PumpFeesCreateFeeSharingConfig), St::PumpFeesInitializeFeeConfig => out.push(Sdk::PumpFeesInitializeFeeConfig), @@ -401,6 +424,7 @@ fn event_type_matches(filter_type: &EventType, event_type: &EventType) -> bool { (filter_type, event_type), (EventType::PumpFunCreateToken, EventType::PumpFunCreateV2Token) | (EventType::PumpFunBuy, EventType::PumpFunBuyExactSolIn) + | (EventType::PumpFunBuyExactSolIn, EventType::PumpFunBuy) | (EventType::TokenAccount, EventType::TokenInfo) | (EventType::RaydiumClmmSwap, EventType::RaydiumClmmSwapV2) | (EventType::RaydiumClmmSwapV2, EventType::RaydiumClmmSwap) @@ -530,9 +554,12 @@ mod tests { fn build_sdk_filter_pumpfun_buy_only() { let f = EventTypeFilter { include: vec![EventType::PumpFunBuy], ..Default::default() }; let sdk_f = build_sdk_parse_event_filter(Some(&f)).expect("mapped"); + assert!(sdk_f.should_include(SdkGrpcEventType::PumpFunTrade)); assert!(sdk_f.should_include(SdkGrpcEventType::PumpFunBuy)); assert!(sdk_f.should_include(SdkGrpcEventType::PumpFunBuyExactSolIn)); - assert!(!sdk_f.should_include(SdkGrpcEventType::PumpFunSell)); + assert!(sdk_f.should_include(SdkGrpcEventType::PumpFunSell)); + assert!(f.passes_event_type(&EventType::PumpFunBuy)); + assert!(!f.passes_event_type(&EventType::PumpFunSell)); } #[test] @@ -569,7 +596,7 @@ mod tests { include: vec![EventType::PumpFunBuyExactSolIn], ..Default::default() }; - assert!(!f.passes_event_type(&EventType::PumpFunBuy)); + assert!(f.passes_event_type(&EventType::PumpFunBuy)); assert!(f.passes_event_type(&EventType::PumpFunBuyExactSolIn)); } @@ -602,9 +629,13 @@ mod tests { ..Default::default() }; let sdk_f = build_sdk_parse_event_filter(Some(&f)).expect("mapped"); - assert!(!sdk_f.should_include(SdkGrpcEventType::PumpFunBuy)); + assert!(sdk_f.should_include(SdkGrpcEventType::PumpFunBuy)); + assert!(sdk_f.should_include(SdkGrpcEventType::PumpFunTrade)); assert!(sdk_f.should_include(SdkGrpcEventType::PumpFunBuyExactSolIn)); - assert!(!sdk_f.should_include(SdkGrpcEventType::PumpFunSell)); + assert!(sdk_f.should_include(SdkGrpcEventType::PumpFunSell)); + assert!(f.passes_event_type(&EventType::PumpFunBuy)); + assert!(!f.passes_event_type(&EventType::PumpFunSell)); + assert!(f.passes_event_type(&EventType::PumpFunBuyExactSolIn)); } #[test] diff --git a/src/streaming/parser_sdk_bridge/mod.rs b/src/streaming/parser_sdk_bridge/mod.rs index 49c48ab..907043f 100644 --- a/src/streaming/parser_sdk_bridge/mod.rs +++ b/src/streaming/parser_sdk_bridge/mod.rs @@ -142,17 +142,17 @@ mod tests { } #[test] - fn converts_pumpfun_buy_exact_quote_in_as_buy_event() { + fn converts_pumpfun_buy_exact_quote_in_v2_as_buy_event() { let mut t = PbPumpTrade::default(); t.metadata = EventMetadata::default(); t.is_buy = true; - t.ix_name = "buy_exact_quote_in".to_string(); + t.ix_name = "buy_exact_quote_in_v2".to_string(); let ev = convert_parser_event(PbDexEvent::PumpFunBuy(t), None, 999).expect("convert"); match ev { DexEvent::PumpFunTradeEvent(st) => { assert_eq!(st.metadata.event_type, EventType::PumpFunBuy); - assert_eq!(st.ix_name, "buy_exact_quote_in"); + assert_eq!(st.ix_name, "buy_exact_quote_in_v2"); } _ => panic!("expected PumpFunTradeEvent"), } diff --git a/src/streaming/shred/connection.rs b/src/streaming/shred/connection.rs index fc78ecd..f90c8a0 100644 --- a/src/streaming/shred/connection.rs +++ b/src/streaming/shred/connection.rs @@ -25,7 +25,9 @@ impl ShredStreamGrpc { /// 创建客户端，使用自定义配置 pub async fn new_with_config(endpoint: String, config: StreamClientConfig) -> AnyResult { - let shredstream_client = ShredstreamProxyClient::connect(endpoint.clone()).await?; + let shredstream_client = ShredstreamProxyClient::connect(endpoint.clone()) + .await? + .max_decoding_message_size(config.connection.max_decoding_message_size); let sdk_config = sol_parser_sdk::shredstream::ShredStreamConfig { connection_timeout_ms: config.connection.connect_timeout.saturating_mul(1000), request_timeout_ms: config.connection.request_timeout.saturating_mul(1000), diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index e63ea17..ba84afb 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -1,25 +1,30 @@ //! ShredStream 订阅入口：底层订阅与热路径解析直接复用 `sol-parser-sdk::shredstream`， -//! 本模块只负责把 SDK `DexEvent` 适配回 streamer 原有 callback API。 -use std::sync::Arc; +//! 本模块负责把 SDK `DexEvent` 适配回 streamer 原有 callback API。 +#![allow(deprecated)] +use std::sync::Arc; +use std::time::Duration; + +use futures::StreamExt; +use solana_entry::entry::Entry as SolanaEntry; use solana_sdk::pubkey::Pubkey; +use tokio::time::timeout; use crate::common::AnyResult; +use crate::streaming::common::parse_shred_transaction_events; use crate::streaming::common::{MetricsEventType, MetricsManager, SubscriptionHandle}; -use crate::streaming::event_parser::common::filter::{ - build_sdk_shred_parse_event_filter, EventTypeFilter, -}; -use crate::streaming::event_parser::common::high_performance_clock::elapsed_micros_since; +use crate::streaming::event_parser::common::filter::EventTypeFilter; +use crate::streaming::event_parser::common::high_performance_clock::get_high_perf_clock; use crate::streaming::event_parser::{DexEvent, Protocol}; -use crate::streaming::parser_sdk_bridge::adapt_parser_event; use super::ShredStreamGrpc; impl ShredStreamGrpc { - /// 订阅ShredStream事件（支持批处理和即时处理） + /// Subscribe to ShredStream events on the direct low-latency path. /// - /// 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. + /// The callback runs on the stream read task; keep it non-blocking. `tx_index` is an + /// entry-local best-effort index because ShredStream entries do not carry the slot-level + /// Yellowstone transaction index. pub async fn shredstream_subscribe( &self, protocols: Vec, @@ -33,37 +38,128 @@ impl ShredStreamGrpc { // 如果已有活跃订阅，先停止它 self.stop().await; - let sdk_parse_filter = - build_sdk_shred_parse_event_filter(&protocols, event_type_filter.as_ref()); - - // Wrap callback once before the async block + let request_timeout = Duration::from_secs(self.config.connection.request_timeout); + let client = self.shredstream_client.as_ref().clone(); 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 { + let mut delay_ms = 100u64; - 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_for_sdk, - filter_for_sdk.as_ref(), - ) { - event.metadata_mut().handle_us = elapsed_micros_since(event.metadata().recv_us); - event = - crate::streaming::event_parser::core::event_parser::helpers::process_event( - event, bot_wallet, + loop { + let mut client = client.clone(); + let request = + tonic::Request::new(crate::protos::shredstream::SubscribeEntriesRequest {}); + let response = if request_timeout.is_zero() { + client.subscribe_entries(request).await + } else { + match timeout(request_timeout, client.subscribe_entries(request)).await { + Ok(response) => response, + Err(_) => { + log::error!( + "ShredStream subscribe request timed out after {}s - retry in {}ms", + request_timeout.as_secs(), + delay_ms + ); + tokio::time::sleep(Duration::from_millis(delay_ms)).await; + delay_ms = (delay_ms * 2).min(60_000); + continue; + } + } + }; + + let mut stream = match response { + Ok(response) => response.into_inner(), + Err(error) => { + log::error!( + "ShredStream subscribe failed: {error} - retry in {delay_ms}ms" ); + tokio::time::sleep(Duration::from_millis(delay_ms)).await; + delay_ms = (delay_ms * 2).min(60_000); + continue; + } + }; + + while let Some(message) = stream.next().await { + let entry = match message { + Ok(entry) => entry, + Err(error) => { + log::error!("ShredStream stream error: {error} - reconnecting"); + break; + } + }; + delay_ms = 100; + process_shred_entry_direct( + entry, + &protocols, + event_type_filter.as_ref(), + bot_wallet, + callback.as_ref(), + ); + } + + log::warn!("ShredStream stream ended - reconnecting in {delay_ms}ms"); + tokio::time::sleep(Duration::from_millis(delay_ms)).await; + delay_ms = (delay_ms * 2).min(60_000); + } + }); + + 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 mut handle_guard = self.subscription_handle.lock().await; + *handle_guard = Some(subscription_handle); + + Ok(()) + } +} + +#[inline] +fn process_shred_entry_direct( + entry: crate::protos::shredstream::Entry, + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, + bot_wallet: Option, + callback: &(dyn Fn(DexEvent) + Send + Sync), +) { + let slot = entry.slot; + let recv_us = get_high_perf_clock(); + let entries = match bincode::deserialize::>(&entry.entries) { + Ok(entries) => entries, + Err(error) => { + log::debug!("Failed to deserialize ShredStream entries: {error}"); + return; + } + }; + + let mut tx_index = 0u64; + for entry in entries { + for transaction in &entry.transactions { + if transaction.signatures.is_empty() { + tx_index += 1; + continue; + } + + MetricsManager::global().add_tx_process_count(); + let signature = transaction.signatures[0]; + parse_shred_transaction_events( + transaction, + signature, + slot, + Some(tx_index), + recv_us, + protocols, + event_type_filter, + bot_wallet, + |event| { let metadata = event.metadata(); let processing_time_us = metadata.handle_us as f64; let recv_us = metadata.recv_us; let block_time_ms = metadata.block_time_ms; - callback_for_sdk(event); + callback(event); MetricsManager::global().update_metrics_with_latency( MetricsEventType::Transaction, @@ -72,21 +168,9 @@ impl ShredStreamGrpc { recv_us, 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; + }, + ); + tx_index += 1; } - - // 保存订阅句柄 - let subscription_handle = SubscriptionHandle::metrics_only(metrics_handle); - let mut handle_guard = self.subscription_handle.lock().await; - *handle_guard = Some(subscription_handle); - - Ok(()) } }