mirror of
https://github.com/0xfnzero/solana-streamer.git
synced 2026-11-02 07:16:31 +00:00
feat: release solana-streamer-sdk 3.0.0
This commit is contained in:
0xfnzero
+13
-21
@@ -1,7 +1,8 @@
|
||||
[package]
|
||||
name = "solana-streamer-sdk"
|
||||
version = "2.0.3"
|
||||
version = "3.0.0"
|
||||
edition = "2021"
|
||||
rust-version = "1.91"
|
||||
authors = ["William <byteblock6@gmail.com>", "sgxiang <sgxiang@gmail.com>", "wei <1415121722@qq.com>"]
|
||||
repository = "https://github.com/0xfnzero/solana-streamer"
|
||||
description = "A low-latency Solana DEX event streaming facade over sol-parser-sdk with Yellowstone gRPC, ShredStream, and RPC parsing helpers."
|
||||
@@ -21,38 +22,29 @@ sdk-perf-stats = ["sol-parser-sdk/perf-stats"]
|
||||
sdk-ultra-perf = ["sol-parser-sdk/ultra-perf"]
|
||||
|
||||
[dependencies]
|
||||
sol-parser-sdk = { version = "0.6.6", default-features = false }
|
||||
solana-sdk = "3.0.0"
|
||||
solana-client = "3.1.12"
|
||||
solana-transaction-status = "3.1.12"
|
||||
solana-account-decoder = "3.1.12"
|
||||
solana-entry = "3.0.0"
|
||||
sol-parser-sdk = { path = "../sol-parser-sdk", version = "0.7.0", default-features = false }
|
||||
solana-sdk = "=4.0.1"
|
||||
solana-client = "=4.2.2"
|
||||
solana-transaction-status = { version = "=4.2.2", features = ["agave-unstable-api"] }
|
||||
solana-entry = { version = "=4.2.2", features = ["agave-unstable-api"] }
|
||||
borsh = { version = "1.6.0", features = ["derive"] }
|
||||
serde = { version = "1.0.228", features = ["derive"] }
|
||||
serde-big-array = "0.5.1"
|
||||
futures = "0.3.32"
|
||||
bincode = "1.3"
|
||||
wincode = "=0.5.5"
|
||||
anyhow = "1.0.102"
|
||||
yellowstone-grpc-client = "12.1.0"
|
||||
yellowstone-grpc-proto = "=12.4.0"
|
||||
tokio = { version = "1.50.0", features = ["full", "rt-multi-thread"]}
|
||||
yellowstone-grpc-client = "=13.5.0"
|
||||
yellowstone-grpc-proto = "=12.7.0"
|
||||
tokio = { version = "1.50.0", features = ["full"] }
|
||||
tonic = { version = "0.14.5", features = ["transport"] }
|
||||
rustls = { version = "0.23.37", features = ["ring"], default-features = false }
|
||||
log = "0.4.29"
|
||||
dashmap = "6.1.0"
|
||||
dashmap = "=5.5.3"
|
||||
prost = "0.14.3"
|
||||
prost-types = "0.14.3"
|
||||
crossbeam-queue = "0.3.12"
|
||||
spl-token = { version = "9.0.0", default-features = false, features = ["no-entrypoint"] }
|
||||
spl-token-2022 = { version = "10.0.0", default-features = false, features = ["no-entrypoint"] }
|
||||
# spl-token-2022 10.0.0 is incompatible with spl-token-group-interface 0.7.2+ (E0308).
|
||||
# Keep token-group-interface pinned to 0.7.1 until the upstream token-2022 bump is available.
|
||||
spl-token-group-interface = "=0.7.1"
|
||||
# spl-list-view 0.1.0 does not compile with spl-pod 0.7.3 due to its widened
|
||||
# PodLength::TryFrom error type. Pin the compatible patch for lock-free builds.
|
||||
spl-pod = "=0.7.2"
|
||||
solana-commitment-config = { version = "3.1.1", features = ["serde"] }
|
||||
tonic-prost = "0.14.5"
|
||||
|
||||
[dev-dependencies]
|
||||
serde_json = "1.0.149"
|
||||
solana-commitment-config = { version = "3.1.1", features = ["serde"] }
|
||||
|
||||
@@ -129,30 +129,34 @@ Add the dependency to your `Cargo.toml`:
|
||||
|
||||
```toml
|
||||
# Add to your Cargo.toml
|
||||
solana-streamer-sdk = { path = "./solana-streamer", version = "2.0.3" }
|
||||
solana-streamer-sdk = { path = "./solana-streamer", version = "3.0.0" }
|
||||
```
|
||||
|
||||
### Use crates.io
|
||||
|
||||
```toml
|
||||
# Add to your Cargo.toml
|
||||
solana-streamer-sdk = "2.0.3"
|
||||
solana-streamer-sdk = "3.0.0"
|
||||
```
|
||||
|
||||
Parser backend features:
|
||||
|
||||
```toml
|
||||
# Default: sol-parser-sdk parse-borsh backend
|
||||
solana-streamer-sdk = "2.0.3"
|
||||
solana-streamer-sdk = "3.0.0"
|
||||
|
||||
# Zero-copy parser backend for latency-sensitive bots
|
||||
solana-streamer-sdk = { version = "2.0.3", default-features = false, features = ["sdk-parse-zero-copy"] }
|
||||
solana-streamer-sdk = { version = "3.0.0", default-features = false, features = ["sdk-parse-zero-copy"] }
|
||||
```
|
||||
|
||||
If both `sdk-parse-borsh` and `sdk-parse-zero-copy` are enabled, `sol-parser-sdk 0.6.1+` uses the zero-copy backend.
|
||||
|
||||
## 🔄 Migration Guide
|
||||
|
||||
### Upgrading to v3.0.0
|
||||
|
||||
Version 3.0.0 uses `sol-parser-sdk 0.7.0` and Solana 4 public types. ShredStream Entry decoding now uses `wincode 0.5.5` in production and supports Solana V1 transactions. The normal dependency graph dropped from 677 to 544 package/version entries, while duplicated crate names dropped from 119 to 30 by removing redundant direct Solana/SPL dependencies and aligning versions with the parser. This release requires Rust 1.91 or newer and also forwards Meteora DLMM user token input/output accounts.
|
||||
|
||||
### Upgrading to v2.0.3
|
||||
|
||||
Version 2.0.3 uses `sol-parser-sdk 0.6.6` and pins `spl-pod 0.7.2` to keep fresh, lock-free builds compatible with `spl-list-view 0.1.0`. PumpFun trade events preserve the parser SDK's optional pre/post token balances in raw mint units and pre/post SOL balances in lamports, populated from transaction metadata without additional RPC calls. Meteora DLMM swaps also preserve `min_amount_out` separately from the executed `amount_out`.
|
||||
|
||||
+8
-4
@@ -128,30 +128,34 @@ git clone https://github.com/0xfnzero/solana-streamer
|
||||
|
||||
```toml
|
||||
# 添加到您的 Cargo.toml
|
||||
solana-streamer-sdk = { path = "./solana-streamer", version = "2.0.3" }
|
||||
solana-streamer-sdk = { path = "./solana-streamer", version = "3.0.0" }
|
||||
```
|
||||
|
||||
### 使用 crates.io
|
||||
|
||||
```toml
|
||||
# 添加到您的 Cargo.toml
|
||||
solana-streamer-sdk = "2.0.3"
|
||||
solana-streamer-sdk = "3.0.0"
|
||||
```
|
||||
|
||||
解析后端 feature：
|
||||
|
||||
```toml
|
||||
# 默认：sol-parser-sdk parse-borsh 后端
|
||||
solana-streamer-sdk = "2.0.3"
|
||||
solana-streamer-sdk = "3.0.0"
|
||||
|
||||
# 面向低延迟 Bot 的 zero-copy 解析后端
|
||||
solana-streamer-sdk = { version = "2.0.3", default-features = false, features = ["sdk-parse-zero-copy"] }
|
||||
solana-streamer-sdk = { version = "3.0.0", default-features = false, features = ["sdk-parse-zero-copy"] }
|
||||
```
|
||||
|
||||
如果同时启用 `sdk-parse-borsh` 和 `sdk-parse-zero-copy`，`sol-parser-sdk 0.6.1+` 会优先使用 zero-copy 后端。
|
||||
|
||||
## 🔄 迁移指南
|
||||
|
||||
### 升级到 v3.0.0
|
||||
|
||||
v3.0.0 使用 `sol-parser-sdk 0.7.0` 和 Solana 4 公开类型。生产 ShredStream Entry 解码已切换到 `wincode 0.5.5`，并支持 Solana V1 交易。通过删除重复的 Solana/SPL 直接依赖并与 parser 对齐版本，normal 依赖图从 677 个 package/version 项降到 544 个，重复 crate 名从 119 个降到 30 个。此版本要求 Rust 1.91 或更新版本，并会继续转发 Meteora DLMM 用户 token 输入/输出账户。
|
||||
|
||||
### 升级到 v2.0.3
|
||||
|
||||
v2.0.3 使用 `sol-parser-sdk 0.6.6`，并固定 `spl-pod 0.7.2`，确保无锁文件的全新构建与 `spl-list-view 0.1.0` 兼容。PumpFun 交易事件会保留 parser SDK 提供的可选交易前后 token 余额（mint 原始精度）和 SOL 余额（lamports），这些字段直接来自交易 metadata，无需额外 RPC 请求。Meteora DLMM swap 也会单独保留 `min_amount_out`，不再与实际成交的 `amount_out` 混淆。
|
||||
|
||||
@@ -108,3 +108,73 @@ ShredStream 路径仍为下面的原始交易解析限制，不使用上述日
|
||||
- Shred 入口：`streaming/common/event_processor.rs` → `process_shred_transaction`
|
||||
- 账户与 inner 传入：`accounts = tx.message.static_account_keys()`，`inner_instructions: &[]`
|
||||
- 合并逻辑（CPI 覆盖/补充字段）：`streaming/event_parser/core/merger_event.rs` → `merge()`
|
||||
|
||||
## 6. 待解决事项
|
||||
|
||||
以下事项已确认存在，但当前版本暂不处理。实现前必须先保存带 ALT 的真实
|
||||
ShredStream Entry fixture，并按现有基准流程记录优化前后的正确性与延迟数据。
|
||||
|
||||
### 6.1 恢复 V0 ALT-loaded addresses
|
||||
|
||||
**状态**：待解决，暂缓。
|
||||
|
||||
**根因**：当前 ShredStream proxy 的 `Entry` 消息只携带 `slot` 和序列化后的
|
||||
`Vec<solana_entry::entry::Entry>`。V0 transaction 只记录 lookup table 账户地址及
|
||||
writable/readonly 索引，不包含索引对应的 Pubkey。仅使用 transaction bytes 无法无损
|
||||
恢复 loaded addresses。
|
||||
|
||||
**推荐方案**：
|
||||
|
||||
1. 优先扩展 ShredStream proxy 协议，由靠近 validator/bank 的服务端附带 slot 对应的
|
||||
resolved writable/readonly addresses。这是可以保证完整性且客户端延迟最低的方案。
|
||||
2. 若无法修改 proxy，增加异步 ALT cache/resolver。热路径只读取内存缓存，cache miss
|
||||
不得同步调用 RPC；后台通过 RPC 或 Yellowstone account stream 加载和刷新 table。
|
||||
3. 为公开 API 增加明确模式：
|
||||
- `Strict`：无法解析 program id 或必要账户时跳过指令，不产生猜测事件；
|
||||
- `Cached`：仅在 ALT cache 命中时完整解析，未命中时异步加载并跳过；
|
||||
- `BestEffort`：保留当前默认 Pubkey 占位和 discriminator fallback，仅用于兼容。
|
||||
|
||||
**验收条件**：
|
||||
|
||||
- 用真实 V0/ALT Shred fixture 与同一交易的 Yellowstone resolved accounts 对比，所有
|
||||
writable/readonly keys、program id 和事件账户字段完全一致；
|
||||
- cache miss 不阻塞 ShredStream 读流任务，不在热路径发起网络请求；
|
||||
- cache 命中路径必须有前后 Criterion 对比，新增解析开销需单独报告；
|
||||
- table 扩展、失效、重连和并发读取必须有测试；
|
||||
- legacy、V1 和不使用 ALT 的 V0 交易不得发生行为或性能回退。
|
||||
|
||||
### 6.2 消除未知 program discriminator 碰撞
|
||||
|
||||
**状态**：待解决，暂缓。
|
||||
|
||||
**根因**：当外层 `program_id_index` 指向 ALT-loaded key 且真实 program id 无法恢复时，
|
||||
当前 parser 会按启用的协议依次尝试 discriminator-only 解析，并采用第一个成功候选。
|
||||
不同程序可能共享 Anchor discriminator，恶意程序也可以构造相同前缀，因此该路径不能
|
||||
提供严格的程序身份保证。
|
||||
|
||||
**推荐方案**：
|
||||
|
||||
1. 完成 6.1 的 program id 解析后，只向真实 program parser 派发，这是根本解决方案。
|
||||
2. 在 ALT 仍未解析时，`Strict`/`Cached` 模式不得执行 discriminator fallback。
|
||||
3. 保留 `BestEffort` 时必须进一步收紧：要求单协议 filter、完整 payload 布局、合法账户
|
||||
数量、可见固定 program/token/system/event-authority 账户一致，并且只在候选唯一时输出；
|
||||
多个候选同时通过时应丢弃，不得按遍历顺序选择第一个。
|
||||
4. 可在事件元数据中增加解析可信度或来源标记，但这属于公开 API 变更，需要单独评审。
|
||||
|
||||
**验收条件**：
|
||||
|
||||
- 加入共享 discriminator、恶意伪造 payload、账户数量相同及多候选同时命中的负向 fixture；
|
||||
- `Strict`/`Cached` cache-miss 路径不得产生错误协议事件；
|
||||
- `BestEffort` 只允许唯一候选通过，结果不得依赖协议遍历顺序；
|
||||
- 对静态 program id 的常规热路径保持零额外分配，基准不得出现可测量回退；
|
||||
- 分别统计误报、漏报、ALT cache hit/miss，并提供可观测日志或计数器。
|
||||
|
||||
### 6.3 当前决策
|
||||
|
||||
在上述事项完成前：
|
||||
|
||||
- 要求账户字段和程序身份严格正确的业务继续使用 Yellowstone gRPC；
|
||||
- ShredStream 保持当前低延迟 best-effort 语义；
|
||||
- 使用方应尽量配置窄协议/事件 filter，并将默认 Pubkey 账户视为字段不可恢复，而不是
|
||||
有效链上账户；
|
||||
- 本节仅记录后续工作，不代表当前版本已经提供 ALT resolver、严格模式或零误报保证。
|
||||
|
||||
@@ -31,7 +31,7 @@ async fn main() -> Result<()> {
|
||||
solana_client::rpc_config::RpcTransactionConfig {
|
||||
encoding: Some(solana_transaction_status::UiTransactionEncoding::Base64),
|
||||
commitment: Some(CommitmentConfig::confirmed()),
|
||||
max_supported_transaction_version: Some(0),
|
||||
max_supported_transaction_version: Some(1),
|
||||
},
|
||||
)
|
||||
.await
|
||||
|
||||
@@ -30,7 +30,7 @@ async fn main() -> Result<()> {
|
||||
solana_client::rpc_config::RpcTransactionConfig {
|
||||
encoding: Some(solana_transaction_status::UiTransactionEncoding::Base64),
|
||||
commitment: Some(CommitmentConfig::confirmed()),
|
||||
max_supported_transaction_version: Some(0),
|
||||
max_supported_transaction_version: Some(1),
|
||||
},
|
||||
)
|
||||
.await
|
||||
|
||||
@@ -184,7 +184,7 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> {
|
||||
solana_client::rpc_config::RpcTransactionConfig {
|
||||
encoding: Some(UiTransactionEncoding::Base64),
|
||||
commitment: Some(CommitmentConfig::confirmed()),
|
||||
max_supported_transaction_version: Some(0),
|
||||
max_supported_transaction_version: Some(1),
|
||||
},
|
||||
)
|
||||
.await
|
||||
|
||||
@@ -242,6 +242,11 @@ pub(crate) fn parse_shred_transaction_events(
|
||||
bot_wallet: Option<Pubkey>,
|
||||
mut on_event: impl FnMut(DexEvent),
|
||||
) {
|
||||
if tx.sanitize().is_err() {
|
||||
log::debug!("Ignoring unsanitized ShredStream transaction");
|
||||
return;
|
||||
}
|
||||
|
||||
let cost_selection = transaction_cost_selection(event_type_filter);
|
||||
if cost_selection.requested {
|
||||
let cost = sol_parser_sdk::parse_shred_transaction_cost(tx);
|
||||
@@ -372,7 +377,7 @@ mod tests {
|
||||
header: MessageHeader {
|
||||
num_required_signatures: 1,
|
||||
num_readonly_signed_accounts: 0,
|
||||
num_readonly_unsigned_accounts: 0,
|
||||
num_readonly_unsigned_accounts: 1,
|
||||
},
|
||||
account_keys,
|
||||
recent_blockhash: Hash::default(),
|
||||
@@ -398,17 +403,21 @@ mod tests {
|
||||
VersionedTransaction {
|
||||
signatures: vec![Signature::default()],
|
||||
message: VersionedMessage::V0(v0::Message {
|
||||
header: MessageHeader::default(),
|
||||
header: MessageHeader {
|
||||
num_required_signatures: 1,
|
||||
num_readonly_signed_accounts: 0,
|
||||
num_readonly_unsigned_accounts: 1,
|
||||
},
|
||||
account_keys: vec![
|
||||
source,
|
||||
system_program,
|
||||
JITO_TIP_ACCOUNTS[0],
|
||||
GLAIVE_TIP_ACCOUNTS[0],
|
||||
system_program,
|
||||
],
|
||||
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]),
|
||||
CompiledInstruction::new_from_raw_parts(3, transfer(10), vec![0, 1]),
|
||||
CompiledInstruction::new_from_raw_parts(3, transfer(20), vec![0, 2]),
|
||||
],
|
||||
address_table_lookups: Vec::new(),
|
||||
}),
|
||||
@@ -515,6 +524,29 @@ mod tests {
|
||||
assert_eq!(cost.metadata.tx_index, Some(7));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn shred_transaction_rejects_unsanitized_message() {
|
||||
let mut transaction = pumpfun_create_tx();
|
||||
let VersionedMessage::V0(message) = &mut transaction.message else {
|
||||
panic!("expected V0 fixture");
|
||||
};
|
||||
message.instructions[0].program_id_index = u8::MAX;
|
||||
let events = Arc::new(Mutex::new(Vec::new()));
|
||||
let captured = events.clone();
|
||||
|
||||
process_shred_transaction(
|
||||
TransactionWithSlot::new(transaction, 42, 1_000_000, Some(0)),
|
||||
&[Protocol::PumpFun],
|
||||
None,
|
||||
Arc::new(move |event| captured.lock().unwrap().push(event)),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("reject malformed Shred transaction");
|
||||
|
||||
assert!(events.lock().unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn yellowstone_cost_parsing_is_opt_in() {
|
||||
let without_cost =
|
||||
|
||||
@@ -314,6 +314,10 @@ pub struct MeteoraDlmmSwapEvent {
|
||||
pub token_x_mint: Pubkey,
|
||||
#[serde(default)]
|
||||
pub token_y_mint: Pubkey,
|
||||
#[serde(default)]
|
||||
pub user_token_in: Pubkey,
|
||||
#[serde(default)]
|
||||
pub user_token_out: Pubkey,
|
||||
pub pool: Pubkey,
|
||||
pub from: Pubkey,
|
||||
pub start_bin_id: i32,
|
||||
|
||||
@@ -4,7 +4,7 @@ use crate::streaming::common::constants::{
|
||||
};
|
||||
use std::time::Duration;
|
||||
use tonic::transport::channel::ClientTlsConfig;
|
||||
use yellowstone_grpc_client::{GeyserGrpcClient, Interceptor};
|
||||
use yellowstone_grpc_client::GeyserGrpcClient;
|
||||
|
||||
/// gRPC连接池 - 简化版本
|
||||
pub struct GrpcConnectionPool {
|
||||
@@ -17,7 +17,7 @@ impl GrpcConnectionPool {
|
||||
Self { endpoint, x_token }
|
||||
}
|
||||
|
||||
pub async fn create_connection(&self) -> AnyResult<GeyserGrpcClient<impl Interceptor>> {
|
||||
pub async fn create_connection(&self) -> AnyResult<GeyserGrpcClient> {
|
||||
let builder = GeyserGrpcClient::build_from_shared(self.endpoint.clone())?
|
||||
.x_token(self.x_token.clone())?
|
||||
.tls_config(ClientTlsConfig::new().with_native_roots())?
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
use futures::{channel::mpsc, sink::Sink, Stream};
|
||||
use futures::{sink::Sink, Stream};
|
||||
use std::{collections::HashMap, time::Duration};
|
||||
use tonic::{transport::channel::ClientTlsConfig, Status};
|
||||
use yellowstone_grpc_client::{GeyserGrpcClient, Interceptor};
|
||||
use yellowstone_grpc_client::{GeyserGrpcClient, SubscribeRequestSinkError};
|
||||
use yellowstone_grpc_proto::geyser::{
|
||||
CommitmentLevel, SubscribeRequest, SubscribeRequestFilterAccounts,
|
||||
SubscribeRequestFilterBlocksMeta, SubscribeRequestFilterTransactions, SubscribeUpdate,
|
||||
@@ -30,7 +30,7 @@ impl SubscriptionManager {
|
||||
}
|
||||
|
||||
/// Create gRPC connection
|
||||
pub async fn connect(&self) -> AnyResult<GeyserGrpcClient<impl Interceptor>> {
|
||||
pub async fn connect(&self) -> AnyResult<GeyserGrpcClient> {
|
||||
let builder = GeyserGrpcClient::build_from_shared(self.endpoint.clone())?
|
||||
.x_token(self.x_token.clone())?
|
||||
.tls_config(ClientTlsConfig::new().with_native_roots())?
|
||||
@@ -48,7 +48,7 @@ impl SubscriptionManager {
|
||||
commitment: Option<CommitmentLevel>,
|
||||
event_type_filter: Option<&EventTypeFilter>,
|
||||
) -> AnyResult<(
|
||||
impl Sink<SubscribeRequest, Error = mpsc::SendError>,
|
||||
impl Sink<SubscribeRequest, Error = SubscribeRequestSinkError>,
|
||||
impl Stream<Item = Result<SubscribeUpdate, Status>>,
|
||||
SubscribeRequest,
|
||||
)> {
|
||||
@@ -123,6 +123,7 @@ impl SubscriptionManager {
|
||||
account_include: tf.account_include.clone(),
|
||||
account_exclude: tf.account_exclude.clone(),
|
||||
account_required: tf.account_required.clone(),
|
||||
..Default::default()
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
@@ -362,6 +362,8 @@ pub(crate) fn meteora_dlmm_swap_from_pb(
|
||||
metadata: meta,
|
||||
token_x_mint: e.token_x_mint,
|
||||
token_y_mint: e.token_y_mint,
|
||||
user_token_in: e.user_token_in,
|
||||
user_token_out: e.user_token_out,
|
||||
pool: e.pool,
|
||||
from: e.from,
|
||||
start_bin_id: e.start_bin_id,
|
||||
|
||||
@@ -448,10 +448,14 @@ mod tests {
|
||||
let from = Pubkey::new_unique();
|
||||
let token_x_mint = Pubkey::new_unique();
|
||||
let token_y_mint = Pubkey::new_unique();
|
||||
let user_token_in = Pubkey::new_unique();
|
||||
let user_token_out = Pubkey::new_unique();
|
||||
let pb = PbDlmmSwap {
|
||||
metadata: EventMetadata::default(),
|
||||
token_x_mint,
|
||||
token_y_mint,
|
||||
user_token_in,
|
||||
user_token_out,
|
||||
pool,
|
||||
from,
|
||||
start_bin_id: -5,
|
||||
@@ -470,6 +474,8 @@ mod tests {
|
||||
DexEvent::MeteoraDlmmSwapEvent(e) => {
|
||||
assert_eq!(e.token_x_mint, token_x_mint);
|
||||
assert_eq!(e.token_y_mint, token_y_mint);
|
||||
assert_eq!(e.user_token_in, user_token_in);
|
||||
assert_eq!(e.user_token_out, user_token_out);
|
||||
assert_eq!(e.pool, pool);
|
||||
assert_eq!(e.from, from);
|
||||
assert_eq!(e.start_bin_id, -5);
|
||||
|
||||
+94
-20
@@ -9,7 +9,10 @@
|
||||
//! [`parse_encoded_rpc_transaction_as_streamer_events`].
|
||||
|
||||
use prost_types::Timestamp;
|
||||
use sol_parser_sdk::{parse_rpc_transaction, parse_transaction_from_rpc};
|
||||
use sol_parser_sdk::{
|
||||
parse_rpc_transaction, parse_rpc_transaction_cost_with_signature,
|
||||
parse_rpc_transaction_with_cost, parse_transaction_from_rpc,
|
||||
};
|
||||
use solana_client::rpc_client::RpcClient;
|
||||
use solana_client::rpc_config::RpcTransactionConfig;
|
||||
use solana_sdk::signature::Signature;
|
||||
@@ -34,27 +37,28 @@ pub fn parse_encoded_rpc_transaction_as_streamer_events(
|
||||
) -> Result<Vec<DexEvent>, ParseError> {
|
||||
let block_ts = rpc_tx.block_time.map(|sec| Timestamp { seconds: sec, nanos: 0 });
|
||||
let cost_selection = transaction_cost_selection(event_type_filter);
|
||||
let mut events = if cost_selection.only {
|
||||
Vec::with_capacity(1)
|
||||
let (mut events, transaction_cost) = if cost_selection.only {
|
||||
(Vec::with_capacity(1), Some(parse_rpc_transaction_cost_with_signature(rpc_tx)?))
|
||||
} 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,
|
||||
let (pb_events, cost) = if cost_selection.requested {
|
||||
let parsed = parse_rpc_transaction_with_cost(rpc_tx, sdk_filter.as_ref())?;
|
||||
(parsed.events, Some((parsed.cost, parsed.signature)))
|
||||
} else {
|
||||
(parse_rpc_transaction(rpc_tx, sdk_filter.as_ref())?, None)
|
||||
};
|
||||
(
|
||||
adapt_parser_events_list(
|
||||
pb_events,
|
||||
block_ts.as_ref(),
|
||||
recv_wall_us,
|
||||
protocols,
|
||||
event_type_filter,
|
||||
),
|
||||
cost,
|
||||
)
|
||||
};
|
||||
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()))?;
|
||||
if let Some((cost, signature)) = transaction_cost {
|
||||
let block_time_us = rpc_tx.block_time.map(|seconds| seconds * 1_000_000);
|
||||
events.push(DexEvent::TransactionCostEvent(TransactionCostEvent::from_parser(
|
||||
cost,
|
||||
@@ -81,7 +85,7 @@ pub fn fetch_rpc_transaction_as_streamer_events(
|
||||
let config = RpcTransactionConfig {
|
||||
encoding: Some(UiTransactionEncoding::Base64),
|
||||
commitment: None,
|
||||
max_supported_transaction_version: Some(0),
|
||||
max_supported_transaction_version: Some(1),
|
||||
};
|
||||
let rpc_tx = rpc_client
|
||||
.get_transaction_with_config(signature, config)
|
||||
@@ -111,7 +115,7 @@ pub async fn fetch_rpc_transaction_as_streamer_events_async(
|
||||
let config = RpcTransactionConfig {
|
||||
encoding: Some(UiTransactionEncoding::Base64),
|
||||
commitment: None,
|
||||
max_supported_transaction_version: Some(0),
|
||||
max_supported_transaction_version: Some(1),
|
||||
};
|
||||
let rpc_tx = rpc_client
|
||||
.get_transaction_with_config(signature, config)
|
||||
@@ -146,8 +150,78 @@ mod tests {
|
||||
use crate::streaming::event_parser::common::EventType;
|
||||
use sol_parser_sdk::SwqosProvider;
|
||||
use solana_client::rpc_client::RpcClient;
|
||||
use solana_client::rpc_response::UiTransactionStatusMeta;
|
||||
use solana_sdk::{
|
||||
hash::Hash,
|
||||
message::{v1, MessageHeader, VersionedMessage},
|
||||
pubkey::Pubkey,
|
||||
transaction::VersionedTransaction,
|
||||
};
|
||||
use solana_transaction_status::{
|
||||
option_serializer::OptionSerializer, Encodable, EncodedTransactionWithStatusMeta,
|
||||
};
|
||||
use std::str::FromStr;
|
||||
|
||||
fn local_rpc_fixture() -> EncodedConfirmedTransactionWithStatusMeta {
|
||||
let transaction = VersionedTransaction {
|
||||
signatures: vec![Signature::from([7; 64])],
|
||||
message: VersionedMessage::V1(v1::Message {
|
||||
header: MessageHeader { num_required_signatures: 1, ..Default::default() },
|
||||
config: v1::TransactionConfig::empty()
|
||||
.with_compute_unit_limit(200_000)
|
||||
.with_priority_fee(1_000),
|
||||
lifetime_specifier: Hash::new_unique(),
|
||||
account_keys: vec![Pubkey::new_unique()],
|
||||
instructions: Vec::new(),
|
||||
}),
|
||||
};
|
||||
EncodedConfirmedTransactionWithStatusMeta {
|
||||
slot: 42,
|
||||
transaction: EncodedTransactionWithStatusMeta {
|
||||
transaction: transaction.encode(UiTransactionEncoding::Base64),
|
||||
meta: Some(UiTransactionStatusMeta {
|
||||
err: None,
|
||||
status: Ok(()),
|
||||
fee: 6_000,
|
||||
pre_balances: vec![1_000_000],
|
||||
post_balances: vec![994_000],
|
||||
inner_instructions: OptionSerializer::None,
|
||||
log_messages: OptionSerializer::None,
|
||||
pre_token_balances: OptionSerializer::None,
|
||||
post_token_balances: OptionSerializer::None,
|
||||
rewards: OptionSerializer::None,
|
||||
loaded_addresses: OptionSerializer::None,
|
||||
return_data: OptionSerializer::None,
|
||||
compute_units_consumed: OptionSerializer::Some(10_000),
|
||||
cost_units: OptionSerializer::None,
|
||||
}),
|
||||
version: None,
|
||||
},
|
||||
block_time: Some(1_700_000_000),
|
||||
transaction_index: Some(0),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn local_rpc_cost_only_and_combined_filters_preserve_cost_output() {
|
||||
let rpc_tx = local_rpc_fixture();
|
||||
for filter in [
|
||||
EventTypeFilter::include_only([EventType::TransactionCost]),
|
||||
EventTypeFilter::include_only([EventType::PumpFunBuy, EventType::TransactionCost]),
|
||||
] {
|
||||
let events =
|
||||
parse_encoded_rpc_transaction_as_streamer_events(&rpc_tx, 123, &[], Some(&filter))
|
||||
.expect("parse local RPC fixture");
|
||||
assert_eq!(events.len(), 1);
|
||||
let DexEvent::TransactionCostEvent(cost) = &events[0] else {
|
||||
panic!("expected transaction cost event");
|
||||
};
|
||||
assert_eq!(cost.metadata.signature, Signature::from([7; 64]));
|
||||
assert_eq!(cost.compute_unit_limit, Some(200_000));
|
||||
assert_eq!(cost.priority_fee_lamports, Some(1_000));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn current_mainnet_transaction_cost_is_reusable() {
|
||||
if std::env::var_os("RUN_MAINNET_TESTS").is_none() {
|
||||
|
||||
@@ -126,7 +126,7 @@ fn process_shred_entry_direct(
|
||||
) {
|
||||
let slot = entry.slot;
|
||||
let recv_us = get_high_perf_clock();
|
||||
let entries = match bincode::deserialize::<Vec<SolanaEntry>>(&entry.entries) {
|
||||
let entries = match decode_shred_entries(&entry.entries) {
|
||||
Ok(entries) => entries,
|
||||
Err(error) => {
|
||||
log::debug!("Failed to deserialize ShredStream entries: {error}");
|
||||
@@ -174,3 +174,44 @@ fn process_shred_entry_direct(
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn decode_shred_entries(bytes: &[u8]) -> wincode::ReadResult<Vec<SolanaEntry>> {
|
||||
wincode::deserialize(bytes)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::decode_shred_entries;
|
||||
use solana_entry::entry::Entry;
|
||||
use solana_sdk::{
|
||||
hash::Hash,
|
||||
message::{v1, MessageHeader, VersionedMessage},
|
||||
pubkey::Pubkey,
|
||||
signature::Signature,
|
||||
transaction::VersionedTransaction,
|
||||
};
|
||||
|
||||
#[test]
|
||||
fn wincode_decodes_solana4_v1_entries() {
|
||||
let entries = vec![Entry {
|
||||
num_hashes: 1,
|
||||
hash: Hash::new_unique(),
|
||||
transactions: vec![VersionedTransaction {
|
||||
signatures: vec![Signature::from([9; 64])],
|
||||
message: VersionedMessage::V1(v1::Message {
|
||||
header: MessageHeader { num_required_signatures: 1, ..Default::default() },
|
||||
config: v1::TransactionConfig::empty()
|
||||
.with_compute_unit_limit(250_000)
|
||||
.with_priority_fee(1_500),
|
||||
lifetime_specifier: Hash::new_unique(),
|
||||
account_keys: vec![Pubkey::new_unique()],
|
||||
instructions: Vec::new(),
|
||||
}),
|
||||
}],
|
||||
}];
|
||||
let bytes = wincode::serialize(&entries).expect("serialize V1 Entry");
|
||||
|
||||
assert_eq!(decode_shred_entries(&bytes).expect("decode V1 Entry"), entries);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user