Compare commits

..

11 Commits

Author SHA1 Message Date
Wood 19f11aa620 Release v4.0.2 with Astralane QUIC reliability updates.
Bump crate/docs version to 4.0.2 and add release notes covering IPv4/IPv6 address-family handling, direct-IP candidate preference, and endpoint failover rotation for more stable low-latency QUIC submission.

Made-with: Cursor
2026-03-24 18:48:44 +08:00
Wood 61cea7546c Fix Astralane QUIC address-family mismatch and add endpoint failover.
Prefer IPv4 candidates (including official Astralane direct IPs), bind local QUIC socket by remote address family, and rotate reconnect attempts across candidates for higher reliability under DNS/Cloudflare variance.

Made-with: Cursor
2026-03-24 18:28:41 +08:00
Wood d52503b8de Release v4.0.1: Fix BlockRazor gRPC endpoints and set gRPC as default transport
## Changes since v4.0.0

### Bug Fixes
- Fixed BlockRazor gRPC endpoints to use correct gRPC service URLs
- Set gRPC as the default transport layer for BlockRazor with HTTP fallback

This release improves the reliability and performance of BlockRazor integration by prioritizing gRPC communication while maintaining HTTP fallback compatibility.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-23 01:34:48 +08:00
Wood 9e53b7694c Fix BlockRazor gRPC endpoints and set gRPC as default transport
- Add SWQOS_ENDPOINTS_BLOCKRAZOR_GRPC with correct gRPC endpoints
- Add Grpc variant to SwqosTransport enum
- Add get_endpoint_with_transport() method for transport-aware endpoint selection
- Change BlockRazor default: transport=None now uses gRPC (was HTTP)
- Require explicit Some(SwqosTransport::Http) for HTTP fallback

This fixes the issue where BlockRazor was incorrectly using HTTP endpoints
when gRPC was intended. Users can now use gRPC by default or explicitly
specify HTTP transport when needed.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-20 01:17:06 +08:00
Wood 07e45d136f Release v4.0.0: SWQoS transport improvements and Binary-Tx response handling
Major changes:
- Switch BlockRazor default transport from gRPC to HTTP to avoid FRAME_SIZE_ERROR
  - gRPC mode still available via explicit SwqosTransport configuration
- Fix ZeroSlot Binary-Tx JSON-RPC 2.0 response parsing
  - Properly handle success responses with "result" field
  - Properly handle error responses with "error" field containing code and message
- Update version to 4.0.0
- Update README files to reflect new version

Technical details:
- BlockRazor: HTTP mode is now the default (SwqosTransport::Http or None)
- ZeroSlot: Parse JSON-RPC 2.0 format responses instead of plain text
- Proto code: Pre-generated gRPC code for BlockRazor (no protoc required)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-18 21:32:23 +08:00
Wood 20d053bab3 Add gRPC support for BlockRazor with HTTP fallback (gRPC by default)
- Add BlockRazorBackend enum to support both gRPC and HTTP transport
- Default to gRPC for better performance, HTTP available via explicit selection
- Update SwqosConfig::BlockRazor to accept optional SwqosTransport parameter
- Implement keep-alive ping task for both gRPC and HTTP backends
- Follow same backend pattern as Astralane (Quic/Http) and Node1 (Quic/Http)
- Manual gRPC client implementation to avoid proto compilation issues

Usage:
- Default: BlockRazorClient::new() uses gRPC
- HTTP: BlockRazorClient::new_http() for HTTP transport
- Config: SwqosConfig::BlockRazor(token, region, url, None) for gRPC
- Config: SwqosConfig::BlockRazor(token, region, url, Some(Http)) for HTTP

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-18 18:01:50 +08:00
Wood 5d921f23ff Improve ZeroSlot client: add HTTP keep-alive ping and switch to Binary-Tx
- Add HTTP keep-alive ping task (30s interval) using free getHealth method
- Switch from JSON-RPC to faster Binary-Tx endpoint (/txb) for transaction submission
- Send raw binary transaction bytes directly to avoid encoding/decoding overhead
- Update response handling for Binary-Tx plain text status codes (200/403/419/500)
- Follow same keep-alive pattern as BlockRazor, Stellium, and Astralane clients

These changes reduce first-submit cold start latency and overall transaction submission time.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-18 17:29:14 +08:00
Wood a187126554 Add 9 new Astralane tip wallets for improved routing and reduced write-lock contention
Made-with: Cursor
2026-03-18 14:40:06 +08:00
Wood 6ea8c27824 docs: sync README and README_CN version to 3.6.5
Made-with: Cursor
2026-03-17 17:51:09 +08:00
Wood 8c4e3ee0c5 Release v3.6.5: SWQOS core affinity and recommended sender thread indices
- Add TradeConfig::with_swqos_cores_from_end(bool) to use last N CPU cores for SWQOS, reducing contention with main thread and default tokio workers.
- Add recommended_sender_thread_core_indices(swqos_count) to get the same last-N core indices for with_dedicated_sender_threads (recommended combo for lower latency).
- Document core affinity and latency in async_executor and with_dedicated_sender_threads.

Made-with: Cursor
2026-03-17 17:19:04 +08:00
Wood 624b1843a2 Improve SDK and SWQOS logging: alignment, errors, duration format
- Rename wait_transaction_confirmed to wait_tx_confirmed (params, lib, docs, examples)
- SDK timing: use [SDK][provider] style with fixed-width labels; add newline before block
- Move print_sdk_timing_block to common::sdk_log; unify SWQOS_LABEL_WIDTH
- SWQOS submitted/failed: aligned [Provider] labels via log_swqos_submitted/submission_failed
- Extract short error message from JSON (message/data) or quoted string; prefix with 'error: '
- Format elapsed as 'X.XXXX ms' or 'X.XXXX µs' (4 decimals, space before unit); use comma before error
- Add SwqosType::as_str() for zero-allocation log labels; only parse JSON when input starts with '{'

Made-with: Cursor
2026-03-17 13:11:33 +08:00
33 changed files with 1591 additions and 390 deletions
+9 -27
View File
@@ -1,30 +1,12 @@
# Generated by Cargo
# will have compiled files and executables
debug/
target/
# Proto 生成工具和生成的代码(用户不需要)
/proto/gen/
# Remove Cargo.lock from gitignore if creating an executable, leave it for libraries
# More information here https://doc.rust-lang.org/cargo/guide/cargo-toml-vs-cargo-lock.html
# 预生成的 proto Rust 代码(提交到仓库,但用户不应修改)
# /src/swqos/pb/serverpb.rs <- 这个文件已经提交,用户不应修改
# Build artifacts
/target/
Cargo.lock
# These are backup files generated by rustfmt
**/*.rs.bk
# MSVC Windows builds of rustc generate these, which store debugging information
*.pdb
# RustRover
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
.cargo/
tmp_*.rs
tmp_*.log
.claude/
.serena/
# Proto sources
/proto/
+5 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "sol-trade-sdk"
version = "3.6.4"
version = "4.0.2"
edition = "2021"
authors = [
"William <byteblock6@gmail.com>",
@@ -135,6 +135,10 @@ incremental = true # 增量编译 - 大幅加速重新编译
opt-level = 1 # 开发时适度优化
overflow-checks = true # 开发时启用溢出检查
# 🚀 构建依赖
# 注意:proto 代码已预生成在 src/swqos/pb/serverpb.rs
# 开发者如需重新生成代码,请运行 gen_proto 目录下的工具
# 🚀 性能关键依赖的特殊优化
[profile.release.package.solana-sdk]
opt-level = 3
+2 -2
View File
@@ -90,14 +90,14 @@ Add the dependency to your `Cargo.toml`:
```toml
# Add to your Cargo.toml
sol-trade-sdk = { path = "./sol-trade-sdk", version = "3.6.4" }
sol-trade-sdk = { path = "./sol-trade-sdk", version = "4.0.2" }
```
### Use crates.io
```toml
# Add to your Cargo.toml
sol-trade-sdk = "3.6.4"
sol-trade-sdk = "4.0.2"
```
## 🛠️ Usage Examples
+2 -2
View File
@@ -90,14 +90,14 @@ git clone https://github.com/0xfnzero/sol-trade-sdk
```toml
# 添加到您的 Cargo.toml
sol-trade-sdk = { path = "./sol-trade-sdk", version = "3.6.4" }
sol-trade-sdk = { path = "./sol-trade-sdk", version = "4.0.2" }
```
### 使用 crates.io
```toml
# 添加到您的 Cargo.toml
sol-trade-sdk = "3.6.4"
sol-trade-sdk = "4.0.2"
```
## 🛠️ 使用示例
-8
View File
@@ -1,8 +0,0 @@
# Release v3.6.4
## Changes from v3.6.3
- **Examples**: All workspace examples are verified to build successfully (`cargo build --workspace`).
- **Docs**: Version references in README.md and README_CN.md updated to 3.6.4.
No API or behavior changes in this release.
+19
View File
@@ -0,0 +1,19 @@
## sol-trade-sdk v4.0.2
This release focuses on QUIC reliability and low-latency submission stability for Astralane.
### Highlights
- Fixed Astralane QUIC address-family mismatch that could produce `invalid remote address` when DNS returned IPv6 first and local endpoint was IPv4-only.
- Added remote-family-aware local QUIC bind selection:
- IPv4 remote -> bind `0.0.0.0:0`
- IPv6 remote -> bind `[::]:0`
- Added Astralane direct-IP candidate support (official region IPs), with IPv4-first selection for better QUIC stability.
- Added automatic endpoint failover and reconnect rotation across candidate addresses, reducing single-endpoint/DNS variance impact.
- Kept existing SDK interfaces compatible while improving submit-path resiliency.
### Also included from recent updates
- BlockRazor gRPC endpoint fixes and gRPC default transport behavior improvements (v4.0.1).
- SWQOS transport path hardening and Binary-Tx response handling improvements (v4.0.0).
+3 -3
View File
@@ -30,7 +30,7 @@ The `TradeBuyParams` struct contains all parameters required for executing buy o
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| `address_lookup_table_account` | `Option<AddressLookupTableAccount>` | ❌ | Address lookup table for transaction optimization |
| `wait_transaction_confirmed` | `bool` | ✅ | Whether to wait for transaction confirmation |
| `wait_tx_confirmed` | `bool` | ✅ | Whether to wait for transaction confirmation |
| `create_input_token_ata` | `bool` | ✅ | Whether to create input token Associated Token Account |
| `close_input_token_ata` | `bool` | ✅ | Whether to close input token ATA after transaction |
| `create_mint_ata` | `bool` | ✅ | Whether to create token mint ATA |
@@ -62,7 +62,7 @@ The `TradeSellParams` struct contains all parameters required for executing sell
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| `address_lookup_table_account` | `Option<Pubkey>` | ❌ | Address lookup table for transaction optimization |
| `wait_transaction_confirmed` | `bool` | ✅ | Whether to wait for transaction confirmation |
| `wait_tx_confirmed` | `bool` | ✅ | Whether to wait for transaction confirmation |
| `create_output_token_ata` | `bool` | ✅ | Whether to create output token Associated Token Account |
| `close_output_token_ata` | `bool` | ✅ | Whether to close output token ATA after transaction |
| `durable_nonce` | `Option<DurableNonceInfo>` | ❌ | Durable nonce information containing nonce account and current nonce value |
@@ -88,7 +88,7 @@ These parameters are essential for defining the basic trading operation:
These parameters control how the transaction is processed:
- **slippage_basis_points**: Controls acceptable price slippage
- **wait_transaction_confirmed**: Controls whether to wait for confirmation
- **wait_tx_confirmed**: Controls whether to wait for confirmation
### 🔧 Account Management Parameters
+3 -3
View File
@@ -30,7 +30,7 @@
| 参数 | 类型 | 必需 | 描述 |
|------|------|------|------|
| `address_lookup_table_account` | `Option<Pubkey>` | ❌ | 用于交易优化的地址查找表 |
| `wait_transaction_confirmed` | `bool` | ✅ | 是否等待交易确认 |
| `wait_tx_confirmed` | `bool` | ✅ | 是否等待交易确认 |
| `create_input_token_ata` | `bool` | ✅ | 是否创建输入代币关联代币账户 |
| `close_input_token_ata` | `bool` | ✅ | 交易后是否关闭输入代币 ATA |
| `create_mint_ata` | `bool` | ✅ | 是否创建代币 mint ATA |
@@ -62,7 +62,7 @@
| 参数 | 类型 | 必需 | 描述 |
|------|------|------|------|
| `address_lookup_table_account` | `Option<AddressLookupTableAccount>` | ❌ | 用于交易优化的地址查找表 |
| `wait_transaction_confirmed` | `bool` | ✅ | 是否等待交易确认 |
| `wait_tx_confirmed` | `bool` | ✅ | 是否等待交易确认 |
| `create_output_token_ata` | `bool` | ✅ | 是否创建输出代币关联代币账户 |
| `close_output_token_ata` | `bool` | ✅ | 交易后是否关闭输出代币 ATA |
| `durable_nonce` | `Option<DurableNonceInfo>` | ❌ | 持久 nonce 信息,包含 nonce 账户和当前 nonce 值 |
@@ -88,7 +88,7 @@
这些参数控制交易的处理方式:
- **slippage_basis_points**: 控制可接受的价格滑点
- **wait_transaction_confirmed**: 控制是否等待确认
- **wait_tx_confirmed**: 控制是否等待确认
### 🔧 账户管理参数
+149
View File
@@ -3,10 +3,59 @@
//! Controlled by `TradeConfig::log_enabled`, set in `TradingClient::new`.
//! All SDK logs (timing, SWQOS submit/confirm, WSOL, blacklist, etc.) should check this before output.
use std::fmt;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
/// Format duration for log: "97.9396 ms" or "13.936 µs", 4 decimal places, space before unit.
fn format_elapsed(d: Duration) -> String {
let secs = d.as_secs_f64();
if secs < 0.001 {
format!("{:.4} µs", secs * 1_000_000.0)
} else {
format!("{:.4} ms", secs * 1000.0)
}
}
/// Extract a short error message for SWQOS submission failed log.
/// Tries JSON "message"/"data" and quoted string; on any parse failure returns original (no panic).
fn extract_swqos_error_message(s: &str) -> String {
let s = s.trim();
if s.is_empty() {
return String::new();
}
// Plain double-quoted string (no inner JSON): unquote
if s.starts_with('"') && s.ends_with('"') && s.len() >= 2 {
let inner = &s[1..s.len() - 1];
if !inner.contains('{') {
return inner.replace("\\\"", "\"");
}
}
// Try parse as JSON only when input looks like JSON (avoid parsing long non-JSON strings)
if s.starts_with('{') {
if let Ok(v) = serde_json::from_str::<serde_json::Value>(s) {
let obj = v
.get("error")
.and_then(|e| e.as_object())
.or_else(|| v.as_object());
if let Some(o) = obj {
if let Some(m) = o.get("message").and_then(|x| x.as_str()) {
return m.to_string();
}
if let Some(d) = o.get("data").and_then(|x| x.as_str()) {
return d.to_string();
}
}
}
}
s.to_string()
}
static SDK_LOG_ENABLED: AtomicBool = AtomicBool::new(true);
/// Width of [provider] label so SWQOS submit/confirm lines align (longest: Speedlanding).
pub const SWQOS_LABEL_WIDTH: usize = 12;
/// Whether SDK logging is enabled (set from TradeConfig.log_enabled in TradingClient::new).
#[inline(always)]
pub fn sdk_log_enabled() -> bool {
@@ -17,3 +66,103 @@ pub fn sdk_log_enabled() -> bool {
pub fn set_sdk_log_enabled(enabled: bool) {
SDK_LOG_ENABLED.store(enabled, Ordering::Relaxed);
}
/// Aligned log: ` [Soyas ] Buy submitted: 13.936 µs`. Call only when sdk_log_enabled().
#[inline]
pub fn log_swqos_submitted(
provider: &str,
trade_type: impl fmt::Display,
elapsed: Duration,
) {
println!(
" [{:width$}] {} submitted: {}",
provider,
trade_type,
format_elapsed(elapsed),
width = SWQOS_LABEL_WIDTH
);
}
/// Prints one SDK timing block (build_instructions, before_submit, per-channel submit_done).
/// When confirm_us is Some, prints confirmed + total; when None, prints "confirmed: -, total: submit_ms".
/// Call only when sdk_log_enabled().
pub fn print_sdk_timing_block(
dir: &str,
start_us: Option<i64>,
build_end_us: Option<i64>,
before_submit_us: Option<i64>,
submit_timings: &[(crate::swqos::SwqosType, i64)],
confirm_us: Option<i64>,
) {
println!();
let start_us = match start_us {
Some(u) => u,
None => return,
};
if let Some(end_us) = build_end_us {
println!(
" [SDK][{:width$}] {} build_instructions: {:.4} ms",
"-",
dir,
(end_us - start_us) as f64 / 1000.0,
width = SWQOS_LABEL_WIDTH
);
}
if let Some(end_us) = before_submit_us {
println!(
" [SDK][{:width$}] {} before_submit: {:.4} ms",
"-",
dir,
(end_us - start_us) as f64 / 1000.0,
width = SWQOS_LABEL_WIDTH
);
}
if let Some(confirm_done_us) = confirm_us {
let total_ms = (confirm_done_us - start_us) as f64 / 1000.0;
for (swqos_type, submit_done_us) in submit_timings {
let submit_ms = (*submit_done_us - start_us).max(0) as f64 / 1000.0;
let confirmed_ms = (confirm_done_us - *submit_done_us).max(0) as f64 / 1000.0;
println!(
" [SDK][{:width$}] {} submit_done: {:.4} ms, confirmed: {:.4} ms, total: {:.4} ms",
swqos_type.as_str(),
dir,
submit_ms,
confirmed_ms,
total_ms,
width = SWQOS_LABEL_WIDTH
);
}
} else {
for (swqos_type, submit_done_us) in submit_timings {
let submit_ms = (*submit_done_us - start_us).max(0) as f64 / 1000.0;
println!(
" [SDK][{:width$}] {} submit_done: {:.4} ms, confirmed: -, total: {:.4} ms",
swqos_type.as_str(),
dir,
submit_ms,
submit_ms,
width = SWQOS_LABEL_WIDTH
);
}
}
}
/// Aligned log: ` [Stellium ] Buy submission failed after 97.9396 ms: ...`. Call only when sdk_log_enabled().
/// Error is normalized: JSON "message"/"data" or quoted string is shown; raw JSON is not.
#[inline]
pub fn log_swqos_submission_failed(
provider: &str,
trade_type: impl fmt::Display,
elapsed: Duration,
err: impl fmt::Display,
) {
let msg = extract_swqos_error_message(&format!("{}", err));
eprintln!(
" [{:width$}] {} submission failed after {}, error: {}",
provider,
trade_type,
format_elapsed(elapsed),
msg,
width = SWQOS_LABEL_WIDTH
);
}
+21 -3
View File
@@ -9,6 +9,8 @@ pub struct InfrastructureConfig {
pub rpc_url: String,
pub swqos_configs: Vec<SwqosConfig>,
pub commitment: CommitmentConfig,
/// When true, SWQOS sender threads use the *last* N cores instead of the first N. Reduces contention with main thread / default tokio workers that often use low-numbered cores. Default false.
pub swqos_cores_from_end: bool,
}
impl InfrastructureConfig {
@@ -17,7 +19,12 @@ impl InfrastructureConfig {
swqos_configs: Vec<SwqosConfig>,
commitment: CommitmentConfig,
) -> Self {
Self { rpc_url, swqos_configs, commitment }
Self {
rpc_url,
swqos_configs,
commitment,
swqos_cores_from_end: false,
}
}
/// Create from TradeConfig (extract infrastructure-only settings)
@@ -26,6 +33,7 @@ impl InfrastructureConfig {
rpc_url: config.rpc_url.clone(),
swqos_configs: config.swqos_configs.clone(),
commitment: config.commitment.clone(),
swqos_cores_from_end: config.swqos_cores_from_end,
}
}
@@ -43,8 +51,8 @@ impl Hash for InfrastructureConfig {
fn hash<H: Hasher>(&self, state: &mut H) {
self.rpc_url.hash(state);
self.swqos_configs.hash(state);
// Hash commitment level as string since CommitmentConfig doesn't impl Hash
format!("{:?}", self.commitment).hash(state);
self.swqos_cores_from_end.hash(state);
}
}
@@ -53,6 +61,7 @@ impl PartialEq for InfrastructureConfig {
self.rpc_url == other.rpc_url
&& self.swqos_configs == other.swqos_configs
&& self.commitment == other.commitment
&& self.swqos_cores_from_end == other.swqos_cores_from_end
}
}
@@ -72,6 +81,8 @@ pub struct TradeConfig {
pub log_enabled: bool,
/// Whether to check minimum tip per SWQOS provider (filter out configs below min). Default false to save latency.
pub check_min_tip: bool,
/// When true, SWQOS uses the *last* N cores (instead of the first N). Use when main thread / tokio use low-numbered cores to reduce CPU contention. Default false.
pub swqos_cores_from_end: bool,
}
impl TradeConfig {
@@ -91,7 +102,8 @@ impl TradeConfig {
create_wsol_ata_on_startup: true, // default: check and create on startup
use_seed_optimize: true, // default: use seed optimization
log_enabled: true, // default: enable all SDK logs
check_min_tip: false, // default: skip min tip check to reduce latency
check_min_tip: false, // default: skip min tip check to reduce latency
swqos_cores_from_end: false,
}
}
@@ -111,6 +123,12 @@ impl TradeConfig {
self.check_min_tip = check_min_tip;
self
}
/// Use the *last* N cores for SWQOS (instead of the first N). Call this when the main thread or tokio workers use low-numbered cores to avoid binding SWQOS to busy cores. Default false.
pub fn with_swqos_cores_from_end(mut self, from_end: bool) -> Self {
self.swqos_cores_from_end = from_end;
self
}
}
pub type SolanaRpcClient = solana_client::nonblocking::rpc_client::RpcClient;
+24
View File
@@ -111,6 +111,7 @@ pub const BLOCKRAZOR_TIP_ACCOUNTS: &[Pubkey] = &[
pubkey!("AP6qExwrbRgBAVaehg4b5xHENX815sMabtBzUzVB4v8S"),
];
/// Astralane tip wallets. Extended with new addresses for improved routing and reduced write-lock contention (see portal.astralane.io/blockline).
pub const ASTRALANE_TIP_ACCOUNTS: &[Pubkey] = &[
pubkey!("astrazznxsGUhWShqgNtAdfrzP2G83DzcWVJDxwV9bF"),
pubkey!("astra4uejePWneqNaJKuFFA8oonqCE1sqF6b45kDMZm"),
@@ -120,6 +121,16 @@ pub const ASTRALANE_TIP_ACCOUNTS: &[Pubkey] = &[
pubkey!("astraubkDw81n4LuutzSQ8uzHCv4BhPVhfvTcYv8SKC"),
pubkey!("astraZW5GLFefxNPAatceHhYjfA1ciq9gvfEg2S47xk"),
pubkey!("astrawVNP4xDBKT7rAdxrLYiTSTdqtUr63fSMduivXK"),
// New tip wallets (2025) for improved performance and reduced write-lock delays
pubkey!("AstrA1ejL4UeXC2SBP4cpeEmtcFPZVLxx3XGKXyCW6to"),
pubkey!("AsTra79FET4aCKWspPqeSFvjJNyp96SvAnrmyAxqg5b7"),
pubkey!("AstrABAu8CBTyuPXpV4eSCJ5fePEPnxN8NqBaPKQ9fHR"),
pubkey!("AsTRADtvb6tTmrsqULQ9Wji9PigDMjhfEMza6zkynEvV"),
pubkey!("AsTRAEoyMofR3vUPpf9k68Gsfb6ymTZttEtsAbv8Bk4d"),
pubkey!("AStrAJv2RN2hKCHxwUMtqmSxgdcNZbihCwc1mCSnG83W"),
pubkey!("Astran35aiQUF57XZsmkWMtNCtXGLzs8upfiqXxth2bz"),
pubkey!("AStRAnpi6kFrKypragExgeRoJ1QnKH7pbSjLAKQVWUum"),
pubkey!("ASTRaoF93eYt73TYvwtsv6fMWHWbGmMUZfVZPo3CRU9C"),
];
pub const STELLIUM_TIP_ACCOUNTS: &[Pubkey] = &[
@@ -266,6 +277,19 @@ pub const SWQOS_ENDPOINTS_BLOCKRAZOR: [&str; 8] = [
"http://frankfurt.solana.blockrazor.xyz:443/v2/sendTransaction",
];
/// BlockRazor gRPC endpoints. Region order: NewYork, Frankfurt, Amsterdam, SLC, Tokyo, London, LosAngeles, Default.
/// Port 80 for gRPC protocol. Auth: apikey metadata in gRPC headers.
pub const SWQOS_ENDPOINTS_BLOCKRAZOR_GRPC: [&str; 8] = [
"http://newyork.solana-grpc.blockrazor.xyz:80",
"http://frankfurt.solana-grpc.blockrazor.xyz:80",
"http://amsterdam.solana-grpc.blockrazor.xyz:80",
"http://newyork.solana-grpc.blockrazor.xyz:80",
"http://tokyo.solana-grpc.blockrazor.xyz:80",
"http://london.solana-grpc.blockrazor.xyz:80",
"http://newyork.solana-grpc.blockrazor.xyz:80",
"http://frankfurt.solana-grpc.blockrazor.xyz:80",
];
/// Astralane binary API path (no Base64; use with ?api-key=...&method=sendTransaction|getHealth).
pub const ASTRALANE_PATH_IRISB: &str = "irisb";
+34 -6
View File
@@ -185,7 +185,15 @@ impl TradingInfrastructure {
let max_by_cores = (num_cores * 2 / 3).max(1);
let cap = swqos_count.min(max_by_cores).max(1);
let ids = core_affinity::get_core_ids()
.map(|all| all.into_iter().take(cap).collect::<Vec<_>>())
.map(|all| {
let v: Vec<_> = all.into_iter().collect();
let len = v.len();
if config.swqos_cores_from_end && len >= cap {
v.into_iter().skip(len - cap).collect()
} else {
v.into_iter().take(cap).collect()
}
})
.unwrap_or_default();
(cap, Arc::new(ids))
};
@@ -200,6 +208,23 @@ impl TradingInfrastructure {
}
}
/// When using `TradeConfig::with_swqos_cores_from_end(true)`, returns the same "last N" core indices
/// that the infrastructure uses. Pass the result to `TradingClient::with_dedicated_sender_threads`
/// for 方式 C (组合使用): SWQOS on last N cores and dedicated sender threads pinned to those cores.
///
/// Returns `None` if core count cannot be determined. `swqos_count` is typically `swqos_configs.len()`.
pub fn recommended_sender_thread_core_indices(swqos_count: usize) -> Option<Vec<usize>> {
let all = core_affinity::get_core_ids()?;
let num_cores = all.len();
if num_cores == 0 {
return None;
}
let max_by_cores = (num_cores * 2 / 3).max(1);
let cap = swqos_count.min(max_by_cores).max(1).min(num_cores);
let start = num_cores.saturating_sub(cap);
Some((start..num_cores).collect())
}
/// Main trading client for Solana DeFi protocols
///
/// `SolTradingSDK` provides a unified interface for trading across multiple Solana DEXs
@@ -277,7 +302,7 @@ pub struct TradeBuyParams {
/// Optional address lookup table for transaction size optimization
pub address_lookup_table_account: Option<AddressLookupTableAccount>,
/// Whether to wait for transaction confirmation before returning
pub wait_transaction_confirmed: bool,
pub wait_tx_confirmed: bool,
/// Whether to create input token associated token account
pub create_input_token_ata: bool,
/// Whether to close input token associated token account after trade
@@ -328,7 +353,7 @@ pub struct TradeSellParams {
/// Optional address lookup table for transaction size optimization
pub address_lookup_table_account: Option<AddressLookupTableAccount>,
/// Whether to wait for transaction confirmation before returning
pub wait_transaction_confirmed: bool,
pub wait_tx_confirmed: bool,
/// Whether to create output token associated token account
pub create_output_token_ata: bool,
/// Whether to close output token associated token account after trade
@@ -635,7 +660,10 @@ impl TradingClient {
/// Concurrency and core count are capped internally (≤ swqos count, ≤ 2/3 of CPU cores).
/// - `None`: keep default (shared tokio pool).
/// - `Some(vec![])`: dedicated threads with default count, no core pinning.
/// - `Some(indices)`: dedicated threads pinned to those core indices (trimmed to cap).
/// - `Some(indices)`: dedicated threads pinned to those core indices (trimmed to cap).
///
/// **Latency note:** If a core is busy with other work (node, bot), SWQOS submit on that core can be delayed.
/// For lowest latency, pass core indices that are *reserved* for SWQOS (do not run other CPU-heavy work on those cores).
pub fn with_dedicated_sender_threads(mut self, core_indices: Option<Vec<usize>>) -> Self {
match core_indices {
None => {
@@ -760,7 +788,7 @@ impl TradingClient {
slippage_basis_points: params.slippage_basis_points,
address_lookup_table_account: params.address_lookup_table_account,
recent_blockhash: params.recent_blockhash,
wait_transaction_confirmed: params.wait_transaction_confirmed,
wait_tx_confirmed: params.wait_tx_confirmed,
protocol_params,
open_seed_optimize: self.use_seed_optimize, // 使用全局seed优化配置
swqos_clients: self.infrastructure.swqos_clients.clone(),
@@ -867,7 +895,7 @@ impl TradingClient {
slippage_basis_points: params.slippage_basis_points,
address_lookup_table_account: params.address_lookup_table_account,
recent_blockhash: params.recent_blockhash,
wait_transaction_confirmed: params.wait_transaction_confirmed,
wait_tx_confirmed: params.wait_tx_confirmed,
protocol_params,
with_tip: params.with_tip,
open_seed_optimize: self.use_seed_optimize, // 使用全局seed优化配置
+23 -10
View File
@@ -2,7 +2,7 @@ use crate::swqos::common::{default_http_client_builder, poll_transaction_confirm
use rand::seq::IndexedRandom;
use reqwest::Client;
use std::{sync::Arc, time::Instant};
use tracing::{error, info, warn};
use tracing::warn;
use crate::swqos::SwqosClientTrait;
use crate::swqos::{SwqosType, TradeType};
@@ -193,15 +193,26 @@ impl AstralaneClient {
let status = response.status();
let _ = response.bytes().await;
if status.is_success() {
info!(target: "sol_trade_sdk", "[astralane] {} submitted: {:?}", trade_type, start_time.elapsed());
if crate::common::sdk_log::sdk_log_enabled() {
crate::common::sdk_log::log_swqos_submitted("Astralane", trade_type, start_time.elapsed());
}
} else {
error!(target: "sol_trade_sdk", "[astralane] {} submission failed: status {}", trade_type, status);
if crate::common::sdk_log::sdk_log_enabled() {
crate::common::sdk_log::log_swqos_submission_failed("Astralane", trade_type, start_time.elapsed(), format!("status {}", status));
}
return Err(anyhow::anyhow!("Astralane sendTransaction failed: {}", status));
}
}
AstralaneBackend::Quic(quic) => {
quic.send_transaction(&body_bytes).await?;
info!(target: "sol_trade_sdk", "[astralane-quic] {} submitted: {:?}", trade_type, start_time.elapsed());
if let Err(e) = quic.send_transaction(&body_bytes).await {
if crate::common::sdk_log::sdk_log_enabled() {
crate::common::sdk_log::log_swqos_submission_failed("Astralane", trade_type, start_time.elapsed(), &e);
}
return Err(e);
}
if crate::common::sdk_log::sdk_log_enabled() {
crate::common::sdk_log::log_swqos_submitted("Astralane", trade_type, start_time.elapsed());
}
}
}
@@ -209,14 +220,16 @@ impl AstralaneClient {
match poll_transaction_confirmation(&self.rpc_client, *signature, wait_confirmation).await {
Ok(_) => (),
Err(e) => {
info!(target: "sol_trade_sdk", "signature: {:?}", signature);
error!(target: "sol_trade_sdk", "[astralane] {} confirmation failed: {:?}", trade_type, start_time.elapsed());
if crate::common::sdk_log::sdk_log_enabled() {
println!(" signature: {:?}", signature);
println!(" [{:width$}] {} confirmation failed: {:?}", "Astralane", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
}
return Err(e);
}
}
if wait_confirmation {
info!(target: "sol_trade_sdk", "signature: {:?}", signature);
info!(target: "sol_trade_sdk", "[astralane] {} confirmed: {:?}", trade_type, start_time.elapsed());
if wait_confirmation && crate::common::sdk_log::sdk_log_enabled() {
println!(" signature: {:?}", signature);
println!(" [{:width$}] {} confirmed: {:?}", "Astralane", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
}
Ok(())
}
+146 -30
View File
@@ -6,8 +6,9 @@ use quinn::crypto::rustls::QuicClientConfig;
use quinn::{ClientConfig, Connection, Endpoint, IdleTimeout, TransportConfig};
use rcgen::{CertificateParams, KeyPair};
use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
use std::net::SocketAddr;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::str::FromStr;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
@@ -40,49 +41,171 @@ pub struct AstralaneQuicClient {
endpoint: Endpoint,
connection: Mutex<Connection>,
server_addr: SocketAddr,
server_candidates: Vec<SocketAddr>,
next_server_idx: AtomicUsize,
#[allow(dead_code)]
api_key: String,
}
impl AstralaneQuicClient {
#[inline]
fn astralane_quic_ip_candidates(host: &str, port: u16) -> Vec<SocketAddr> {
// Official recommended direct-IP list (faster/more stable than DNS-only for QUIC).
// We intentionally avoid fr2/ams2 per prior guidance.
match host {
"fr.gateway.astralane.io" => vec![
SocketAddr::new(IpAddr::V4(Ipv4Addr::new(185, 191, 117, 97)), 7000),
SocketAddr::new(IpAddr::V4(Ipv4Addr::new(45, 139, 132, 160)), 7000),
],
"ny.gateway.astralane.io" => {
vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::new(64, 130, 45, 19)), 7000)]
}
"ams.gateway.astralane.io" => vec![
SocketAddr::new(IpAddr::V4(Ipv4Addr::new(64, 130, 43, 43)), 7000),
SocketAddr::new(IpAddr::V4(Ipv4Addr::new(84, 32, 186, 73)), 7000),
],
"la.gateway.astralane.io" => {
vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::new(74, 118, 142, 151)), 7000)]
}
"lim.gateway.astralane.io" => {
vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::new(162, 19, 222, 232)), 7000)]
}
"sg.gateway.astralane.io" => {
vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::new(67, 209, 54, 176)), 7000)]
}
"lit.gateway.astralane.io" => {
vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::new(84, 32, 97, 47)), 7000)]
}
_ => {
let _ = port; // keep signature consistent; fall back to DNS below.
Vec::new()
}
}
}
#[inline]
fn parse_host_port(server_addr: &str) -> Option<(&str, u16)> {
let (host, port_str) = server_addr.rsplit_once(':')?;
let port = port_str.parse::<u16>().ok()?;
Some((host, port))
}
/// Resolve `host:port` and prefer IPv4 result to avoid v6-remote/v4-local mismatch.
#[inline]
fn resolve_server_candidates(server_addr: &str) -> Result<Vec<SocketAddr>> {
if let Ok(addr) = SocketAddr::from_str(server_addr) {
return Ok(vec![addr]);
}
let mut candidates: Vec<SocketAddr> = Vec::with_capacity(8);
if let Some((host, port)) = Self::parse_host_port(server_addr) {
candidates.extend(Self::astralane_quic_ip_candidates(host, port));
}
use std::net::ToSocketAddrs;
let mut addrs: Vec<SocketAddr> = server_addr
.to_socket_addrs()
.with_context(|| format!("Cannot resolve address: {}", server_addr))?
.collect();
if addrs.is_empty() && candidates.is_empty() {
anyhow::bail!("Cannot resolve address: {}", server_addr);
}
// QUIC in many bot/VPS environments is primarily IPv4; prefer A over AAAA.
addrs.sort_by_key(|a| if a.is_ipv4() { 0 } else { 1 });
for addr in addrs {
if !candidates.contains(&addr) {
candidates.push(addr);
}
}
Ok(candidates)
}
#[inline]
fn local_bind_for_remote(remote: SocketAddr) -> SocketAddr {
match remote.ip() {
IpAddr::V4(_) => SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0),
IpAddr::V6(_) => SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0),
}
}
/// Connect to an Astralane QUIC server.
/// Generates a self-signed TLS certificate with the API key as the Common Name (CN).
pub async fn connect(server_addr: &str, api_key: &str) -> Result<Self> {
let _ = rustls::crypto::ring::default_provider().install_default();
let addr = SocketAddr::from_str(server_addr)
.or_else(|_| {
use std::net::ToSocketAddrs;
server_addr
.to_socket_addrs()
.ok()
.and_then(|mut addrs| addrs.next())
.ok_or_else(|| anyhow::anyhow!("Cannot resolve address: {}", server_addr))
})
let candidates = Self::resolve_server_candidates(server_addr)
.context("Invalid server address")?;
let addr = candidates[0];
info!("[astralane-quic] Building TLS config (CN = api_key)");
let client_config = Self::build_client_config(api_key)?;
let mut endpoint =
Endpoint::client("0.0.0.0:0".parse()?).context("Failed to create QUIC endpoint")?;
let mut endpoint = Endpoint::client(Self::local_bind_for_remote(addr))
.context("Failed to create QUIC endpoint")?;
endpoint.set_default_client_config(client_config);
info!("[astralane-quic] Connecting to {} ...", addr);
let connection = endpoint
.connect(addr, "astralane")?
.await
.context("Failed to connect to Astralane QUIC server")?;
let mut last_err: Option<anyhow::Error> = None;
let mut selected_addr = addr;
let mut connection_opt: Option<Connection> = None;
for candidate in &candidates {
selected_addr = *candidate;
match endpoint.connect(*candidate, "astralane") {
Ok(connecting) => match connecting.await {
Ok(conn) => {
connection_opt = Some(conn);
break;
}
Err(e) => {
warn!("[astralane-quic] connect failed for {}: {}", candidate, e);
last_err = Some(e.into());
}
},
Err(e) => {
warn!("[astralane-quic] connect setup failed for {}: {}", candidate, e);
last_err = Some(e.into());
}
}
}
let connection = connection_opt.ok_or_else(|| {
last_err.unwrap_or_else(|| anyhow::anyhow!("Failed to connect to Astralane QUIC server"))
})?;
info!("[astralane-quic] Connected at {}", addr);
info!("[astralane-quic] Connected at {}", selected_addr);
Ok(Self {
endpoint,
connection: Mutex::new(connection),
server_addr: addr,
server_addr: selected_addr,
server_candidates: candidates,
next_server_idx: AtomicUsize::new(0),
api_key: api_key.to_string(),
})
}
async fn reconnect_next_candidate(&self) -> Result<Connection> {
let total = self.server_candidates.len().max(1);
let start = self.next_server_idx.fetch_add(1, Ordering::Relaxed) % total;
let mut last_err: Option<anyhow::Error> = None;
for offset in 0..total {
let idx = (start + offset) % total;
let addr = self.server_candidates[idx];
match self.endpoint.connect(addr, "astralane") {
Ok(connecting) => match connecting.await {
Ok(conn) => return Ok(conn),
Err(e) => {
warn!("[astralane-quic] reconnect failed for {}: {}", addr, e);
last_err = Some(e.into());
}
},
Err(e) => {
warn!("[astralane-quic] reconnect setup failed for {}: {}", addr, e);
last_err = Some(e.into());
}
}
}
Err(last_err.unwrap_or_else(|| anyhow::anyhow!("Failed to reconnect to Astralane QUIC server")))
}
/// Send a single bincode-serialized `VersionedTransaction`.
/// Fire-and-forget; automatically reconnects if the connection is dead.
pub async fn send_transaction(&self, transaction_bytes: &[u8]) -> Result<()> {
@@ -108,12 +231,9 @@ impl AstralaneQuicClient {
}
}
warn!("[astralane-quic] Connection dead, reconnecting to {} ...", self.server_addr);
*guard = self
.endpoint
.connect(self.server_addr, "astralane")?
.await
.context("Failed to reconnect to Astralane QUIC server")?;
info!("[astralane-quic] Reconnected to {}", self.server_addr);
let new_conn = self.reconnect_next_candidate().await?;
*guard = new_conn.clone();
info!("[astralane-quic] Reconnected");
}
guard.clone()
};
@@ -137,12 +257,8 @@ impl AstralaneQuicClient {
let mut guard = self.connection.lock().await;
if guard.close_reason().is_some() {
info!("[astralane-quic] Reconnecting at {}", self.server_addr);
*guard = self
.endpoint
.connect(self.server_addr, "astralane")?
.await
.context("Failed to reconnect to Astralane QUIC server")?;
info!("[astralane-quic] Reconnected to {}", self.server_addr);
*guard = self.reconnect_next_candidate().await?;
info!("[astralane-quic] Reconnected");
}
Ok(())
}
+297 -106
View File
@@ -17,15 +17,91 @@ use crate::{common::SolanaRpcClient, constants::swqos::BLOCKRAZOR_TIP_ACCOUNTS};
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::task::JoinHandle;
use tonic::transport::Channel;
use tonic::metadata::AsciiMetadataValue;
// Include pre-generated gRPC code
pub mod serverpb {
include!("pb/serverpb.rs");
}
// gRPC client wrapper
#[derive(Clone)]
pub struct BlockRazorGrpcClient {
channel: Channel,
auth_token: String,
}
impl BlockRazorGrpcClient {
pub fn new(channel: Channel, auth_token: String) -> Self {
Self { channel, auth_token }
}
pub async fn get_health(&self) -> Result<String> {
let mut client = serverpb::server_client::ServerClient::new(self.channel.clone());
let apikey = AsciiMetadataValue::try_from(self.auth_token.as_str())
.map_err(|e| anyhow::anyhow!("Invalid API key format: {}", e))?;
let mut request = tonic::Request::new(serverpb::HealthRequest {});
request.metadata_mut().insert("apikey", apikey);
let response = client.get_health(request).await
.map_err(|e| anyhow::anyhow!("gRPC health check failed: {}", e))?;
Ok(response.into_inner().status)
}
pub async fn send_transaction(
&self,
transaction: String,
mode: String,
safe_window: Option<i32>,
revert_protection: bool,
) -> Result<String> {
// 检查交易数据大小
if crate::common::sdk_log::sdk_log_enabled() {
eprintln!("BlockRazor transaction size: {} bytes", transaction.len());
}
let mut client = serverpb::server_client::ServerClient::new(self.channel.clone());
let apikey = AsciiMetadataValue::try_from(self.auth_token.as_str())
.map_err(|e| anyhow::anyhow!("Invalid API key format: {}", e))?;
let mut request = tonic::Request::new(serverpb::SendRequest {
transaction,
mode: String::from(mode),
safe_window,
revert_protection,
});
request.metadata_mut().insert("apikey", apikey);
let response = client.send_transaction(request).await
.map_err(|e| anyhow::anyhow!("gRPC send transaction failed: {}", e))?;
Ok(response.into_inner().signature)
}
}
#[derive(Clone)]
pub enum BlockRazorBackend {
Grpc {
endpoint: String,
auth_token: String,
grpc_client: Arc<BlockRazorGrpcClient>,
ping_handle: Arc<tokio::sync::Mutex<Option<JoinHandle<()>>>>,
stop_ping: Arc<AtomicBool>,
},
Http {
endpoint: String,
auth_token: String,
http_client: Client,
ping_handle: Arc<tokio::sync::Mutex<Option<JoinHandle<()>>>>,
stop_ping: Arc<AtomicBool>,
},
}
#[derive(Clone)]
pub struct BlockRazorClient {
pub endpoint: String,
pub auth_token: String,
pub rpc_client: Arc<SolanaRpcClient>,
pub http_client: Client,
pub ping_handle: Arc<tokio::sync::Mutex<Option<JoinHandle<()>>>>,
pub stop_ping: Arc<AtomicBool>,
backend: BlockRazorBackend,
}
#[async_trait::async_trait]
@@ -36,7 +112,7 @@ impl SwqosClientTrait for BlockRazorClient {
transaction: &VersionedTransaction,
wait_confirmation: bool,
) -> Result<()> {
self.send_transaction(trade_type, transaction, wait_confirmation).await
self.send_transaction_impl(trade_type, transaction, wait_confirmation).await
}
async fn send_transactions(
@@ -45,7 +121,10 @@ impl SwqosClientTrait for BlockRazorClient {
transactions: &Vec<VersionedTransaction>,
wait_confirmation: bool,
) -> Result<()> {
self.send_transactions(trade_type, transactions, wait_confirmation).await
for transaction in transactions {
self.send_transaction_impl(trade_type, transaction, wait_confirmation).await?;
}
Ok(())
}
fn get_tip_account(&self) -> Result<String> {
@@ -62,21 +141,62 @@ impl SwqosClientTrait for BlockRazorClient {
}
impl BlockRazorClient {
pub fn new(rpc_url: String, endpoint: String, auth_token: String) -> Self {
pub async fn new(rpc_url: String, endpoint: String, auth_token: String) -> Result<Self> {
// 默认使用 HTTP 模式,避免 gRPC FRAME_SIZE_ERROR
Ok(Self::new_http(rpc_url, endpoint, auth_token))
}
pub async fn new_grpc(rpc_url: String, endpoint: String, auth_token: String) -> Result<Self> {
let rpc_client = SolanaRpcClient::new(rpc_url);
// 官方文档:請求中唯一允許的 header 是 Content-Type: text/plain;避免默认 User-Agent 等导致 500
let http_client = default_http_client_builder().user_agent("").build().unwrap();
// 配置 Channel,增加连接超时
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.map_err(|e| anyhow::anyhow!("Invalid gRPC endpoint: {}", e))?
.timeout(Duration::from_secs(30))
.connect()
.await
.map_err(|e| anyhow::anyhow!("Failed to connect to gRPC endpoint: {}", e))?;
let grpc_client = Arc::new(BlockRazorGrpcClient::new(channel, auth_token.clone()));
let ping_handle = Arc::new(tokio::sync::Mutex::new(None));
let stop_ping = Arc::new(AtomicBool::new(false));
let client = Self {
rpc_client: Arc::new(rpc_client),
endpoint,
auth_token,
http_client,
ping_handle: Arc::new(tokio::sync::Mutex::new(None)),
stop_ping: Arc::new(AtomicBool::new(false)),
backend: BlockRazorBackend::Grpc {
endpoint,
auth_token,
grpc_client,
ping_handle,
stop_ping,
},
};
let client_clone = client.clone();
tokio::spawn(async move {
client_clone.start_ping_task().await;
});
Ok(client)
}
pub fn new_http(rpc_url: String, endpoint: String, auth_token: String) -> Self {
let rpc_client = SolanaRpcClient::new(rpc_url);
let http_client = default_http_client_builder().user_agent("").build().unwrap();
let ping_handle = Arc::new(tokio::sync::Mutex::new(None));
let stop_ping = Arc::new(AtomicBool::new(false));
let client = Self {
rpc_client: Arc::new(rpc_client),
backend: BlockRazorBackend::Http {
endpoint,
auth_token,
http_client,
ping_handle,
stop_ping,
},
};
// Start ping task
let client_clone = client.clone();
tokio::spawn(async move {
client_clone.start_ping_task().await;
@@ -85,47 +205,87 @@ impl BlockRazorClient {
client
}
/// Start periodic ping task to keep connections active
async fn start_ping_task(&self) {
let endpoint = self.endpoint.clone();
let auth_token = self.auth_token.clone();
let http_client = self.http_client.clone();
let stop_ping = self.stop_ping.clone();
match &self.backend {
BlockRazorBackend::Grpc {
grpc_client,
ping_handle,
stop_ping,
..
} => {
let grpc_client = grpc_client.clone();
let ping_handle = ping_handle.clone();
let stop_ping = stop_ping.clone();
let handle = tokio::spawn(async move {
// Immediate first ping to warm connection and reduce first-submit cold start latency
if let Err(e) = Self::send_ping_request(&http_client, &endpoint, &auth_token).await {
if crate::common::sdk_log::sdk_log_enabled() {
eprintln!("BlockRazor ping request failed: {}", e);
}
}
let mut interval = tokio::time::interval(Duration::from_secs(30)); // 30s keepalive to avoid server ~5min idle close
loop {
interval.tick().await;
if stop_ping.load(Ordering::Relaxed) {
break;
}
if let Err(e) = Self::send_ping_request(&http_client, &endpoint, &auth_token).await
{
if crate::common::sdk_log::sdk_log_enabled() {
eprintln!("BlockRazor ping request failed: {}", e);
let handle = tokio::spawn(async move {
if let Err(e) = grpc_client.get_health().await {
if crate::common::sdk_log::sdk_log_enabled() {
eprintln!("BlockRazor gRPC ping request failed: {}", e);
}
}
}
}
});
let mut interval = tokio::time::interval(Duration::from_secs(30));
loop {
interval.tick().await;
if stop_ping.load(Ordering::Relaxed) {
break;
}
if let Err(e) = grpc_client.get_health().await {
if crate::common::sdk_log::sdk_log_enabled() {
eprintln!("BlockRazor gRPC ping request failed: {}", e);
}
}
}
});
// Update ping_handle - use Mutex to safely update
{
let mut ping_guard = self.ping_handle.lock().await;
if let Some(old_handle) = ping_guard.as_ref() {
old_handle.abort();
let mut ping_guard = ping_handle.lock().await;
if let Some(old_handle) = ping_guard.as_ref() {
old_handle.abort();
}
*ping_guard = Some(handle);
}
BlockRazorBackend::Http {
endpoint,
auth_token,
http_client,
ping_handle,
stop_ping,
} => {
let endpoint = endpoint.clone();
let auth_token = auth_token.clone();
let http_client = http_client.clone();
let ping_handle = ping_handle.clone();
let stop_ping = stop_ping.clone();
let handle = tokio::spawn(async move {
if let Err(e) = Self::send_http_ping(&http_client, &endpoint, &auth_token).await {
if crate::common::sdk_log::sdk_log_enabled() {
eprintln!("BlockRazor HTTP ping request failed: {}", e);
}
}
let mut interval = tokio::time::interval(Duration::from_secs(30));
loop {
interval.tick().await;
if stop_ping.load(Ordering::Relaxed) {
break;
}
if let Err(e) = Self::send_http_ping(&http_client, &endpoint, &auth_token).await {
if crate::common::sdk_log::sdk_log_enabled() {
eprintln!("BlockRazor HTTP ping request failed: {}", e);
}
}
}
});
let mut ping_guard = ping_handle.lock().await;
if let Some(old_handle) = ping_guard.as_ref() {
old_handle.abort();
}
*ping_guard = Some(handle);
}
*ping_guard = Some(handle);
}
}
/// Send ping request: POST /v2/health?auth=... (Keep Alive). Only required param: auth.
async fn send_ping_request(
async fn send_http_ping(
http_client: &Client,
endpoint: &str,
auth_token: &str,
@@ -142,63 +302,105 @@ impl BlockRazorClient {
let status = response.status();
let _ = response.bytes().await;
if !status.is_success() {
eprintln!("BlockRazor ping request failed with status: {}", status);
eprintln!("BlockRazor HTTP ping request failed with status: {}", status);
}
Ok(())
}
/// Send transaction via v2 API: plain Base64 body, Content-Type: text/plain. Only required URI param: auth.
/// 文档要求:auth 以 URI 参数传入;body 为纯 Base64 编码交易;唯一允许的 header 为 Content-Type: text/plain。
pub async fn send_transaction(
async fn send_transaction_impl(
&self,
trade_type: TradeType,
transaction: &VersionedTransaction,
wait_confirmation: bool,
) -> Result<()> {
let start_time = Instant::now();
let (content, signature) =
serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64)?;
let response = self
.http_client
.post(&self.endpoint)
.query(&[("auth", self.auth_token.as_str())])
.header("Content-Type", "text/plain")
.body(content)
.send()
.await?;
match &self.backend {
BlockRazorBackend::Grpc {
grpc_client,
..
} => {
let (content, _signature) =
serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64)?;
let status = response.status();
if status.is_success() {
let _ = response.bytes().await;
if crate::common::sdk_log::sdk_log_enabled() {
println!(" [blockrazor] {} submitted: {:?}", trade_type, start_time.elapsed());
let signature = grpc_client.send_transaction(
content,
"fast".to_string(),
None,
false,
).await;
match signature {
Ok(sig) => {
if !sig.is_empty() {
if crate::common::sdk_log::sdk_log_enabled() {
crate::common::sdk_log::log_swqos_submitted("BlockRazor", trade_type, start_time.elapsed());
}
} else {
if crate::common::sdk_log::sdk_log_enabled() {
crate::common::sdk_log::log_swqos_submission_failed("BlockRazor", trade_type, start_time.elapsed(), "empty signature".to_string());
}
return Err(anyhow::anyhow!("BlockRazor gRPC returned empty signature"));
}
}
Err(e) => {
if crate::common::sdk_log::sdk_log_enabled() {
crate::common::sdk_log::log_swqos_submission_failed("BlockRazor", trade_type, start_time.elapsed(), format!("gRPC error: {}", e));
}
return Err(anyhow::anyhow!("BlockRazor gRPC sendTransaction failed: {}", e));
}
}
}
} else {
let body = response.text().await.unwrap_or_default();
if crate::common::sdk_log::sdk_log_enabled() {
eprintln!(
" [blockrazor] {} submission failed: status {} body: {}",
trade_type, status, body
);
BlockRazorBackend::Http {
endpoint,
auth_token,
http_client,
..
} => {
let (content, _signature) =
serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64)?;
let response = http_client
.post(endpoint)
.query(&[("auth", auth_token.as_str())])
.header("Content-Type", "text/plain")
.body(content)
.send()
.await?;
let status = response.status();
if status.is_success() {
let _ = response.bytes().await;
if crate::common::sdk_log::sdk_log_enabled() {
crate::common::sdk_log::log_swqos_submitted("blockrazor", trade_type, start_time.elapsed());
}
} else {
let body = response.text().await.unwrap_or_default();
if crate::common::sdk_log::sdk_log_enabled() {
crate::common::sdk_log::log_swqos_submission_failed("blockrazor", trade_type, start_time.elapsed(), format!("status {} body: {}", status, body));
}
return Err(anyhow::anyhow!(
"BlockRazor HTTP sendTransaction failed: status {} body: {}",
status,
body
));
}
}
return Err(anyhow::anyhow!(
"BlockRazor sendTransaction failed: status {} body: {}",
status,
body
));
}
let start_time = Instant::now();
let signature = transaction.signatures[0];
match poll_transaction_confirmation(&self.rpc_client, signature, wait_confirmation).await {
Ok(_) => (),
Err(e) => {
if crate::common::sdk_log::sdk_log_enabled() {
println!(" signature: {:?}", signature);
println!(
" [blockrazor] {} confirmation failed: {:?}",
" [{:width$}] {} confirmation failed: {:?}",
"blockrazor",
trade_type,
start_time.elapsed()
start_time.elapsed(),
width = crate::common::sdk_log::SWQOS_LABEL_WIDTH
);
}
return Err(e);
@@ -206,39 +408,28 @@ impl BlockRazorClient {
}
if wait_confirmation && crate::common::sdk_log::sdk_log_enabled() {
println!(" signature: {:?}", signature);
println!(" [blockrazor] {} confirmed: {:?}", trade_type, start_time.elapsed());
println!(" [{:width$}] {} confirmed: {:?}", "blockrazor", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
}
Ok(())
}
pub async fn send_transactions(
&self,
trade_type: TradeType,
transactions: &Vec<VersionedTransaction>,
wait_confirmation: bool,
) -> Result<()> {
for transaction in transactions {
self.send_transaction(trade_type, transaction, wait_confirmation).await?;
}
Ok(())
}
}
impl Drop for BlockRazorClient {
fn drop(&mut self) {
// Ensure ping task stops when client is destroyed
self.stop_ping.store(true, Ordering::Relaxed);
match &self.backend {
BlockRazorBackend::Grpc { stop_ping, ping_handle, .. } | BlockRazorBackend::Http { stop_ping, ping_handle, .. } => {
stop_ping.store(true, Ordering::Relaxed);
// Try to stop ping task immediately
// Use tokio::spawn to avoid blocking Drop
let ping_handle = self.ping_handle.clone();
tokio::spawn(async move {
let mut ping_guard = ping_handle.lock().await;
if let Some(handle) = ping_guard.as_ref() {
handle.abort();
let ping_handle = ping_handle.clone();
tokio::spawn(async move {
let mut ping_guard = ping_handle.lock().await;
if let Some(handle) = ping_guard.as_ref() {
handle.abort();
}
*ping_guard = None;
});
}
*ping_guard = None;
});
}
}
}
+9 -7
View File
@@ -100,13 +100,13 @@ impl BloxrouteClient {
if let Ok(response_json) = serde_json::from_str::<serde_json::Value>(&response_text) {
if crate::common::sdk_log::sdk_log_enabled() {
if response_json.get("result").is_some() {
println!(" [bloxroute] {} submitted: {:?}", trade_type, start_time.elapsed());
crate::common::sdk_log::log_swqos_submitted("bloxroute", trade_type, start_time.elapsed());
} else if let Some(_error) = response_json.get("error") {
eprintln!(" [bloxroute] {} submission failed: {:?}", trade_type, _error);
eprintln!(" [bloxroute] {} submission failed after {:?}: {:?}", trade_type, start_time.elapsed(), _error);
}
}
} else if crate::common::sdk_log::sdk_log_enabled() {
eprintln!(" [bloxroute] {} submission failed: {:?}", trade_type, response_text);
crate::common::sdk_log::log_swqos_submission_failed("bloxroute", trade_type, start_time.elapsed(), response_text);
}
let start_time: Instant = Instant::now();
@@ -116,9 +116,11 @@ impl BloxrouteClient {
if crate::common::sdk_log::sdk_log_enabled() {
println!(" signature: {:?}", signature);
println!(
" [bloxroute] {} confirmation failed: {:?}",
" [{:width$}] {} confirmation failed: {:?}",
"bloxroute",
trade_type,
start_time.elapsed()
start_time.elapsed(),
width = crate::common::sdk_log::SWQOS_LABEL_WIDTH
);
}
return Err(e);
@@ -126,7 +128,7 @@ impl BloxrouteClient {
}
if wait_confirmation && crate::common::sdk_log::sdk_log_enabled() {
println!(" signature: {:?}", signature);
println!(" [bloxroute] {} confirmed: {:?}", trade_type, start_time.elapsed());
println!(" [{:width$}] {} confirmed: {:?}", "bloxroute", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
}
Ok(())
@@ -168,7 +170,7 @@ impl BloxrouteClient {
if response_json.get("result").is_some() {
println!(" bloxroute {} submitted: {:?}", trade_type, start_time.elapsed());
} else if let Some(_error) = response_json.get("error") {
eprintln!(" bloxroute {} submission failed: {:?}", trade_type, _error);
eprintln!(" bloxroute {} submission failed after {:?}: {:?}", trade_type, start_time.elapsed(), _error);
}
}
}
+8 -6
View File
@@ -97,12 +97,12 @@ impl FlashBlockClient {
// Parse response
if let Ok(response_json) = serde_json::from_str::<serde_json::Value>(&response_text) {
if response_json.get("success").is_some() || response_json.get("result").is_some() {
println!(" [FlashBlock] {} submitted: {:?}", trade_type, start_time.elapsed());
crate::common::sdk_log::log_swqos_submitted("FlashBlock", trade_type, start_time.elapsed());
} else if let Some(_error) = response_json.get("error") {
eprintln!(" [FlashBlock] {} submission failed: {:?}", trade_type, _error);
eprintln!(" [FlashBlock] {} submission failed after {:?}: {:?}", trade_type, start_time.elapsed(), _error);
}
} else {
eprintln!(" [FlashBlock] {} submission failed: {:?}", trade_type, response_text);
crate::common::sdk_log::log_swqos_submission_failed("FlashBlock", trade_type, start_time.elapsed(), response_text);
}
let start_time: Instant = Instant::now();
@@ -111,16 +111,18 @@ impl FlashBlockClient {
Err(e) => {
println!(" signature: {:?}", signature);
println!(
" [FlashBlock] {} confirmation failed: {:?}",
" [{:width$}] {} confirmation failed: {:?}",
"FlashBlock",
trade_type,
start_time.elapsed()
start_time.elapsed(),
width = crate::common::sdk_log::SWQOS_LABEL_WIDTH
);
return Err(e);
}
}
if wait_confirmation {
println!(" signature: {:?}", signature);
println!(" [FlashBlock] {} confirmed: {:?}", trade_type, start_time.elapsed());
println!(" [{:width$}] {} confirmed: {:?}", "FlashBlock", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
}
Ok(())
+10 -8
View File
@@ -106,8 +106,8 @@ impl HeliusClient {
if !status.is_success() {
if crate::common::sdk_log::sdk_log_enabled() {
eprintln!(
" [helius] {} submission failed status={} body={}",
trade_type, status, response_text
" [helius] {} submission failed after {:?} status={} body={}",
trade_type, start_time.elapsed(), status, response_text
);
}
return Err(anyhow::anyhow!(
@@ -124,15 +124,15 @@ impl HeliusClient {
.and_then(|v| v.as_str())
.unwrap_or("unknown");
if crate::common::sdk_log::sdk_log_enabled() {
eprintln!(" [helius] {} submission error: {}", trade_type, err_msg);
crate::common::sdk_log::log_swqos_submission_failed("helius", trade_type, start_time.elapsed(), err_msg);
}
return Err(anyhow::anyhow!("Helius Sender error: {}", err_msg));
}
if response_json.get("result").is_some() && crate::common::sdk_log::sdk_log_enabled() {
println!(" [helius] {} submitted: {:?}", trade_type, start_time.elapsed());
crate::common::sdk_log::log_swqos_submitted("helius", trade_type, start_time.elapsed());
}
} else if crate::common::sdk_log::sdk_log_enabled() {
eprintln!(" [helius] {} submission failed: {:?}", trade_type, response_text);
crate::common::sdk_log::log_swqos_submission_failed("helius", trade_type, start_time.elapsed(), response_text);
}
match poll_transaction_confirmation(&self.rpc_client, signature, wait_confirmation).await {
@@ -140,9 +140,11 @@ impl HeliusClient {
Err(e) => {
if crate::common::sdk_log::sdk_log_enabled() {
eprintln!(
" [helius] {} confirmation failed: {:?}",
" [{:width$}] {} confirmation failed: {:?}",
"helius",
trade_type,
start_time.elapsed()
start_time.elapsed(),
width = crate::common::sdk_log::SWQOS_LABEL_WIDTH
);
}
return Err(e);
@@ -150,7 +152,7 @@ impl HeliusClient {
}
if wait_confirmation && crate::common::sdk_log::sdk_log_enabled() {
println!(" signature: {:?}", signature);
println!(" [helius] {} confirmed: {:?}", trade_type, start_time.elapsed());
println!(" [{:width$}] {} confirmed: {:?}", "helius", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
}
Ok(())
}
+6 -6
View File
@@ -105,12 +105,12 @@ impl JitoClient {
if let Ok(response_json) = serde_json::from_str::<serde_json::Value>(&response_text) {
if response_json.get("result").is_some() {
println!(" [jito] {} submitted: {:?}", trade_type, start_time.elapsed());
crate::common::sdk_log::log_swqos_submitted("jito", trade_type, start_time.elapsed());
} else if let Some(_error) = response_json.get("error") {
eprintln!(" [jito] {} submission failed: {:?}", trade_type, _error);
eprintln!(" [jito] {} submission failed after {:?}: {:?}", trade_type, start_time.elapsed(), _error);
}
} else {
eprintln!(" [jito] {} submission failed: {:?}", trade_type, response_text);
crate::common::sdk_log::log_swqos_submission_failed("jito", trade_type, start_time.elapsed(), response_text);
}
let start_time: Instant = Instant::now();
@@ -118,13 +118,13 @@ impl JitoClient {
Ok(_) => (),
Err(e) => {
println!(" signature: {:?}", signature);
println!(" [jito] {} confirmation failed: {:?}", trade_type, start_time.elapsed());
println!(" [{:width$}] {} confirmation failed: {:?}", "jito", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
return Err(e);
}
}
if wait_confirmation {
println!(" signature: {:?}", signature);
println!(" [jito] {} confirmed: {:?}", trade_type, start_time.elapsed());
println!(" [{:width$}] {} confirmed: {:?}", "jito", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
}
Ok(())
@@ -171,7 +171,7 @@ impl JitoClient {
if response_json.get("result").is_some() {
println!(" jito {} submitted: {:?}", trade_type, start_time.elapsed());
} else if let Some(_error) = response_json.get("error") {
eprintln!(" jito {} submission failed: {:?}", trade_type, _error);
eprintln!(" jito {} submission failed after {:?}: {:?}", trade_type, start_time.elapsed(), _error);
}
}
+8 -6
View File
@@ -103,12 +103,12 @@ impl LightspeedClient {
if let Ok(response_json) = serde_json::from_str::<serde_json::Value>(&response_text) {
if response_json.get("result").is_some() {
println!(" [lightspeed] {} submitted: {:?}", trade_type, start_time.elapsed());
crate::common::sdk_log::log_swqos_submitted("lightspeed", trade_type, start_time.elapsed());
} else if let Some(_error) = response_json.get("error") {
eprintln!(" [lightspeed] {} submission failed: {:?}", trade_type, _error);
crate::common::sdk_log::log_swqos_submission_failed("lightspeed", trade_type, start_time.elapsed(), _error);
}
} else {
eprintln!(" [lightspeed] {} submission failed: {:?}", trade_type, response_text);
crate::common::sdk_log::log_swqos_submission_failed("lightspeed", trade_type, start_time.elapsed(), response_text);
}
let start_time: Instant = Instant::now();
@@ -117,16 +117,18 @@ impl LightspeedClient {
Err(e) => {
println!(" signature: {:?}", signature);
println!(
" [lightspeed] {} confirmation failed: {:?}",
" [{:width$}] {} confirmation failed: {:?}",
"lightspeed",
trade_type,
start_time.elapsed()
start_time.elapsed(),
width = crate::common::sdk_log::SWQOS_LABEL_WIDTH
);
return Err(e);
}
}
if wait_confirmation {
println!(" signature: {:?}", signature);
println!(" [lightspeed] {} confirmed: {:?}", trade_type, start_time.elapsed());
println!(" [{:width$}] {} confirmed: {:?}", "lightspeed", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
}
Ok(())
+93 -18
View File
@@ -30,15 +30,16 @@ use crate::{
common::SolanaRpcClient,
constants::swqos::{
SWQOS_ENDPOINTS_ASTRALANE, SWQOS_ENDPOINTS_ASTRALANE_QUIC, SWQOS_ENDPOINTS_BLOCKRAZOR,
SWQOS_ENDPOINTS_BLOX, SWQOS_ENDPOINTS_FLASHBLOCK, SWQOS_ENDPOINTS_HELIUS,
SWQOS_ENDPOINTS_JITO, SWQOS_ENDPOINTS_NEXTBLOCK, SWQOS_ENDPOINTS_NODE1,
SWQOS_ENDPOINTS_NODE1_QUIC, SWQOS_ENDPOINTS_SOYAS, SWQOS_ENDPOINTS_SPEEDLANDING,
SWQOS_ENDPOINTS_STELLIUM, SWQOS_ENDPOINTS_TEMPORAL, SWQOS_ENDPOINTS_ZERO_SLOT,
SWQOS_MIN_TIP_ASTRALANE, SWQOS_MIN_TIP_BLOCKRAZOR, SWQOS_MIN_TIP_BLOXROUTE,
SWQOS_MIN_TIP_DEFAULT, SWQOS_MIN_TIP_FLASHBLOCK, SWQOS_MIN_TIP_HELIUS, SWQOS_MIN_TIP_JITO,
SWQOS_MIN_TIP_LIGHTSPEED, SWQOS_MIN_TIP_NEXTBLOCK, SWQOS_MIN_TIP_NODE1,
SWQOS_MIN_TIP_SOYAS, SWQOS_MIN_TIP_SPEEDLANDING, SWQOS_MIN_TIP_STELLIUM,
SWQOS_MIN_TIP_TEMPORAL, SWQOS_MIN_TIP_ZERO_SLOT,
SWQOS_ENDPOINTS_BLOCKRAZOR_GRPC, SWQOS_ENDPOINTS_BLOX, SWQOS_ENDPOINTS_FLASHBLOCK,
SWQOS_ENDPOINTS_HELIUS, SWQOS_ENDPOINTS_JITO, SWQOS_ENDPOINTS_NEXTBLOCK,
SWQOS_ENDPOINTS_NODE1, SWQOS_ENDPOINTS_NODE1_QUIC, SWQOS_ENDPOINTS_SOYAS,
SWQOS_ENDPOINTS_SPEEDLANDING, SWQOS_ENDPOINTS_STELLIUM, SWQOS_ENDPOINTS_TEMPORAL,
SWQOS_ENDPOINTS_ZERO_SLOT, SWQOS_MIN_TIP_ASTRALANE, SWQOS_MIN_TIP_BLOCKRAZOR,
SWQOS_MIN_TIP_BLOXROUTE, SWQOS_MIN_TIP_DEFAULT, SWQOS_MIN_TIP_FLASHBLOCK,
SWQOS_MIN_TIP_HELIUS, SWQOS_MIN_TIP_JITO, SWQOS_MIN_TIP_LIGHTSPEED,
SWQOS_MIN_TIP_NEXTBLOCK, SWQOS_MIN_TIP_NODE1, SWQOS_MIN_TIP_SOYAS,
SWQOS_MIN_TIP_SPEEDLANDING, SWQOS_MIN_TIP_STELLIUM, SWQOS_MIN_TIP_TEMPORAL,
SWQOS_MIN_TIP_ZERO_SLOT,
},
swqos::{
astralane::AstralaneClient, blockrazor::BlockRazorClient, bloxroute::BloxrouteClient,
@@ -63,11 +64,14 @@ pub const SWQOS_BLACKLIST: &[SwqosType] = &[
SwqosType::NextBlock, // NextBlock is disabled by default
];
/// SWQOS 提交通道:HTTP 或 QUIC(低延迟)。部分提供商(如 Astralane)支持 QUIC。
/// SWQOS 提交通道:HTTP、gRPC 或 QUIC(低延迟)。
/// BlockRazor 支持 gRPC 和 HTTP。
/// Astralane 和 Node1 支持 QUIC。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
pub enum SwqosTransport {
#[default]
Http,
Grpc,
Quic,
}
@@ -111,6 +115,28 @@ pub enum SwqosType {
}
impl SwqosType {
/// Label for log alignment; same as Debug output (e.g. "Soyas", "Speedlanding").
#[inline]
pub fn as_str(self) -> &'static str {
match self {
Self::Jito => "Jito",
Self::NextBlock => "NextBlock",
Self::ZeroSlot => "ZeroSlot",
Self::Temporal => "Temporal",
Self::Bloxroute => "Bloxroute",
Self::Node1 => "Node1",
Self::FlashBlock => "FlashBlock",
Self::BlockRazor => "BlockRazor",
Self::Astralane => "Astralane",
Self::Stellium => "Stellium",
Self::Lightspeed => "Lightspeed",
Self::Soyas => "Soyas",
Self::Speedlanding => "Speedlanding",
Self::Helius => "Helius",
Self::Default => "Default",
}
}
pub fn values() -> Vec<Self> {
vec![
Self::Jito,
@@ -202,8 +228,8 @@ pub enum SwqosConfig {
Node1(String, SwqosRegion, Option<String>, Option<SwqosTransport>),
/// FlashBlock(api_token, region, custom_url)
FlashBlock(String, SwqosRegion, Option<String>),
/// BlockRazor(api_token, region, custom_url)
BlockRazor(String, SwqosRegion, Option<String>),
/// BlockRazor(api_token, region, custom_url, transport). transport=None 或 Grpc => gRPC; Some(Http) => HTTP.
BlockRazor(String, SwqosRegion, Option<String>, Option<SwqosTransport>),
/// Astralane(api_token, region, custom_url, transport). transport=None 表示 Http。
Astralane(String, SwqosRegion, Option<String>, Option<SwqosTransport>),
/// Stellium(api_token, region, custom_url)
@@ -233,7 +259,7 @@ impl SwqosConfig {
SwqosConfig::ZeroSlot(_, _, _) => SwqosType::ZeroSlot,
SwqosConfig::Node1(_, _, _, _) => SwqosType::Node1,
SwqosConfig::FlashBlock(_, _, _) => SwqosType::FlashBlock,
SwqosConfig::BlockRazor(_, _, _) => SwqosType::BlockRazor,
SwqosConfig::BlockRazor(_, _, _, _) => SwqosType::BlockRazor,
SwqosConfig::Astralane(_, _, _, _) => SwqosType::Astralane,
SwqosConfig::Stellium(_, _, _) => SwqosType::Stellium,
SwqosConfig::Lightspeed(_, _, _) => SwqosType::Lightspeed,
@@ -272,6 +298,46 @@ impl SwqosConfig {
}
}
pub fn get_endpoint_with_transport(
swqos_type: SwqosType,
region: SwqosRegion,
url: Option<String>,
transport: Option<SwqosTransport>,
) -> String {
if let Some(custom_url) = url {
return custom_url;
}
match swqos_type {
SwqosType::BlockRazor => {
// transport=None 或 transport=Grpc => gRPC; transport=Http => HTTP
let use_http = transport.map_or(false, |t| t == SwqosTransport::Http);
if use_http {
SWQOS_ENDPOINTS_BLOCKRAZOR[region as usize].to_string()
} else {
SWQOS_ENDPOINTS_BLOCKRAZOR_GRPC[region as usize].to_string()
}
}
SwqosType::Node1 => {
let use_quic = transport.map_or(false, |t| t == SwqosTransport::Quic);
if use_quic {
SWQOS_ENDPOINTS_NODE1_QUIC[region as usize].to_string()
} else {
SWQOS_ENDPOINTS_NODE1[region as usize].to_string()
}
}
SwqosType::Astralane => {
let use_quic = transport.map_or(false, |t| t == SwqosTransport::Quic);
if use_quic {
SWQOS_ENDPOINTS_ASTRALANE_QUIC[region as usize].to_string()
} else {
SWQOS_ENDPOINTS_ASTRALANE[region as usize].to_string()
}
}
_ => Self::get_endpoint(swqos_type, region, url),
}
}
pub async fn get_swqos_client(
rpc_url: String,
commitment: CommitmentConfig,
@@ -329,11 +395,20 @@ impl SwqosConfig {
FlashBlockClient::new(rpc_url.clone(), endpoint.to_string(), auth_token);
Ok(Arc::new(flashblock_client))
}
SwqosConfig::BlockRazor(auth_token, region, url) => {
let endpoint = SwqosConfig::get_endpoint(SwqosType::BlockRazor, region, url);
let blockrazor_client =
BlockRazorClient::new(rpc_url.clone(), endpoint.to_string(), auth_token);
Ok(Arc::new(blockrazor_client))
SwqosConfig::BlockRazor(auth_token, region, url, transport) => {
// BlockRazor: transport=None 或 transport=Grpc 时使用 gRPCtransport=Http 时使用 HTTP
let use_http = transport.map_or(false, |t| t == SwqosTransport::Http);
let endpoint = SwqosConfig::get_endpoint_with_transport(SwqosType::BlockRazor, region, url, transport);
if use_http {
let blockrazor_client =
BlockRazorClient::new_http(rpc_url.clone(), endpoint.to_string(), auth_token);
Ok(Arc::new(blockrazor_client))
} else {
// 使用 gRPC 模式(默认或用户明确指定了 gRPC)
let blockrazor_client =
BlockRazorClient::new_grpc(rpc_url.clone(), endpoint.to_string(), auth_token).await?;
Ok(Arc::new(blockrazor_client))
}
}
SwqosConfig::Astralane(auth_token, region, url, transport) => {
let use_quic = transport.map_or(false, |t| t == SwqosTransport::Quic);
+8 -6
View File
@@ -99,12 +99,12 @@ impl NextBlockClient {
if let Ok(response_json) = serde_json::from_str::<serde_json::Value>(&response_text) {
if response_json.get("result").is_some() {
println!(" [nextblock] {} submitted: {:?}", trade_type, start_time.elapsed());
crate::common::sdk_log::log_swqos_submitted("nextblock", trade_type, start_time.elapsed());
} else if let Some(_error) = response_json.get("error") {
eprintln!(" [nextblock] {} submission failed: {:?}", trade_type, _error);
crate::common::sdk_log::log_swqos_submission_failed("nextblock", trade_type, start_time.elapsed(), _error);
}
} else {
eprintln!(" [nextblock] {} submission failed: {:?}", trade_type, response_text);
crate::common::sdk_log::log_swqos_submission_failed("nextblock", trade_type, start_time.elapsed(), response_text);
}
let start_time: Instant = Instant::now();
@@ -113,16 +113,18 @@ impl NextBlockClient {
Err(e) => {
println!(" signature: {:?}", signature);
println!(
" [nextblock] {} confirmation failed: {:?}",
" [{:width$}] {} confirmation failed: {:?}",
"nextblock",
trade_type,
start_time.elapsed()
start_time.elapsed(),
width = crate::common::sdk_log::SWQOS_LABEL_WIDTH
);
return Err(e);
}
}
if wait_confirmation {
println!(" signature: {:?}", signature);
println!(" [nextblock] {} confirmed: {:?}", trade_type, start_time.elapsed());
println!(" [{:width$}] {} confirmed: {:?}", "nextblock", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
}
Ok(())
+8 -6
View File
@@ -184,13 +184,13 @@ impl Node1Client {
if let Ok(response_json) = serde_json::from_str::<serde_json::Value>(&response_text) {
if crate::common::sdk_log::sdk_log_enabled() {
if response_json.get("result").is_some() {
println!(" [node1] {} submitted: {:?}", trade_type, start_time.elapsed());
crate::common::sdk_log::log_swqos_submitted("node1", trade_type, start_time.elapsed());
} else if let Some(_error) = response_json.get("error") {
eprintln!(" [node1] {} submission failed: {:?}", trade_type, _error);
eprintln!(" [node1] {} submission failed after {:?}: {:?}", trade_type, start_time.elapsed(), _error);
}
}
} else if crate::common::sdk_log::sdk_log_enabled() {
eprintln!(" [node1] {} submission failed: {:?}", trade_type, response_text);
crate::common::sdk_log::log_swqos_submission_failed("node1", trade_type, start_time.elapsed(), response_text);
}
let start_time: Instant = Instant::now();
@@ -200,9 +200,11 @@ impl Node1Client {
if crate::common::sdk_log::sdk_log_enabled() {
println!(" signature: {:?}", signature);
println!(
" [node1] {} confirmation failed: {:?}",
" [{:width$}] {} confirmation failed: {:?}",
"node1",
trade_type,
start_time.elapsed()
start_time.elapsed(),
width = crate::common::sdk_log::SWQOS_LABEL_WIDTH
);
}
return Err(e);
@@ -210,7 +212,7 @@ impl Node1Client {
}
if wait_confirmation && crate::common::sdk_log::sdk_log_enabled() {
println!(" signature: {:?}", signature);
println!(" [node1] {} confirmed: {:?}", trade_type, start_time.elapsed());
println!(" [{:width$}] {} confirmed: {:?}", "node1", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
}
Ok(())
+42
View File
@@ -0,0 +1,42 @@
# Proto 生成的代码说明
## 概述
这个目录包含了从 `.proto` 文件预生成的 Rust 代码。
## 文件说明
- `serverpb.rs` - 从 `blockrazor.proto` 生成的 gRPC 代码
- 消息类型: `SendRequest`, `SendResponse`, `HealthRequest`, `HealthResponse`
- gRPC 客户端: `server_client::ServerClient`
- gRPC 服务端: `server_server::Server`
## 用户使用
用户**不需要**安装 `protoc` 或编译 proto 文件。这些代码已经预生成好了,可以直接使用。
`blockrazor.rs` 中使用:
```rust
pub mod serverpb {
include!("pb/serverpb.rs");
}
```
## 开发者如何重新生成代码
如果你修改了 `.proto` 文件并需要重新生成代码:
```bash
cd sol-trade-sdk/proto/gen
cargo run
```
这会在 `src/swqos/pb/serverpb.rs` 生成新的代码。
## 技术细节
生成工具使用 `tonic-prost-build` crate
- 输出目录: `src/swqos/pb`
- Proto 文件: `proto/blockrazor.proto`
- 生成工具: `proto/gen/`
- 包含完整的客户端和服务端代码
+383
View File
@@ -0,0 +1,383 @@
// This file is @generated by prost-build.
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
pub struct SendRequest {
#[prost(string, tag = "1")]
pub transaction: ::prost::alloc::string::String,
#[prost(string, tag = "2")]
pub mode: ::prost::alloc::string::String,
/// only take effect in sandwichMitigation mode
#[prost(int32, optional, tag = "3")]
pub safe_window: ::core::option::Option<i32>,
#[prost(bool, tag = "4")]
pub revert_protection: bool,
}
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
pub struct SendResponse {
#[prost(string, tag = "1")]
pub signature: ::prost::alloc::string::String,
}
#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
pub struct HealthRequest {}
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
pub struct HealthResponse {
#[prost(string, tag = "1")]
pub status: ::prost::alloc::string::String,
}
/// Generated client implementations.
pub mod server_client {
#![allow(
unused_variables,
dead_code,
missing_docs,
clippy::wildcard_imports,
clippy::let_unit_value,
)]
use tonic::codegen::*;
use tonic::codegen::http::Uri;
#[derive(Debug, Clone)]
pub struct ServerClient<T> {
inner: tonic::client::Grpc<T>,
}
impl ServerClient<tonic::transport::Channel> {
/// Attempt to create a new client by connecting to a given endpoint.
pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
where
D: TryInto<tonic::transport::Endpoint>,
D::Error: Into<StdError>,
{
let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
Ok(Self::new(conn))
}
}
impl<T> ServerClient<T>
where
T: tonic::client::GrpcService<tonic::body::Body>,
T::Error: Into<StdError>,
T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
<T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
{
pub fn new(inner: T) -> Self {
let inner = tonic::client::Grpc::new(inner);
Self { inner }
}
pub fn with_origin(inner: T, origin: Uri) -> Self {
let inner = tonic::client::Grpc::with_origin(inner, origin);
Self { inner }
}
pub fn with_interceptor<F>(
inner: T,
interceptor: F,
) -> ServerClient<InterceptedService<T, F>>
where
F: tonic::service::Interceptor,
T::ResponseBody: Default,
T: tonic::codegen::Service<
http::Request<tonic::body::Body>,
Response = http::Response<
<T as tonic::client::GrpcService<tonic::body::Body>>::ResponseBody,
>,
>,
<T as tonic::codegen::Service<
http::Request<tonic::body::Body>,
>>::Error: Into<StdError> + std::marker::Send + std::marker::Sync,
{
ServerClient::new(InterceptedService::new(inner, interceptor))
}
/// Compress requests with the given encoding.
///
/// This requires the server to support it otherwise it might respond with an
/// error.
#[must_use]
pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
self.inner = self.inner.send_compressed(encoding);
self
}
/// Enable decompressing responses.
#[must_use]
pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
self.inner = self.inner.accept_compressed(encoding);
self
}
/// Limits the maximum size of a decoded message.
///
/// Default: `4MB`
#[must_use]
pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
self.inner = self.inner.max_decoding_message_size(limit);
self
}
/// Limits the maximum size of an encoded message.
///
/// Default: `usize::MAX`
#[must_use]
pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
self.inner = self.inner.max_encoding_message_size(limit);
self
}
pub async fn send_transaction(
&mut self,
request: impl tonic::IntoRequest<super::SendRequest>,
) -> std::result::Result<tonic::Response<super::SendResponse>, tonic::Status> {
self.inner
.ready()
.await
.map_err(|e| {
tonic::Status::unknown(
format!("Service was not ready: {}", e.into()),
)
})?;
let codec = tonic_prost::ProstCodec::default();
let path = http::uri::PathAndQuery::from_static(
"/serverpb.Server/SendTransaction",
);
let mut req = request.into_request();
req.extensions_mut()
.insert(GrpcMethod::new("serverpb.Server", "SendTransaction"));
self.inner.unary(req, path, codec).await
}
pub async fn get_health(
&mut self,
request: impl tonic::IntoRequest<super::HealthRequest>,
) -> std::result::Result<tonic::Response<super::HealthResponse>, tonic::Status> {
self.inner
.ready()
.await
.map_err(|e| {
tonic::Status::unknown(
format!("Service was not ready: {}", e.into()),
)
})?;
let codec = tonic_prost::ProstCodec::default();
let path = http::uri::PathAndQuery::from_static(
"/serverpb.Server/GetHealth",
);
let mut req = request.into_request();
req.extensions_mut().insert(GrpcMethod::new("serverpb.Server", "GetHealth"));
self.inner.unary(req, path, codec).await
}
}
}
/// Generated server implementations.
pub mod server_server {
#![allow(
unused_variables,
dead_code,
missing_docs,
clippy::wildcard_imports,
clippy::let_unit_value,
)]
use tonic::codegen::*;
/// Generated trait containing gRPC methods that should be implemented for use with ServerServer.
#[async_trait]
pub trait Server: std::marker::Send + std::marker::Sync + 'static {
async fn send_transaction(
&self,
request: tonic::Request<super::SendRequest>,
) -> std::result::Result<tonic::Response<super::SendResponse>, tonic::Status>;
async fn get_health(
&self,
request: tonic::Request<super::HealthRequest>,
) -> std::result::Result<tonic::Response<super::HealthResponse>, tonic::Status>;
}
#[derive(Debug)]
pub struct ServerServer<T> {
inner: Arc<T>,
accept_compression_encodings: EnabledCompressionEncodings,
send_compression_encodings: EnabledCompressionEncodings,
max_decoding_message_size: Option<usize>,
max_encoding_message_size: Option<usize>,
}
impl<T> ServerServer<T> {
pub fn new(inner: T) -> Self {
Self::from_arc(Arc::new(inner))
}
pub fn from_arc(inner: Arc<T>) -> Self {
Self {
inner,
accept_compression_encodings: Default::default(),
send_compression_encodings: Default::default(),
max_decoding_message_size: None,
max_encoding_message_size: None,
}
}
pub fn with_interceptor<F>(
inner: T,
interceptor: F,
) -> InterceptedService<Self, F>
where
F: tonic::service::Interceptor,
{
InterceptedService::new(Self::new(inner), interceptor)
}
/// Enable decompressing requests with the given encoding.
#[must_use]
pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
self.accept_compression_encodings.enable(encoding);
self
}
/// Compress responses with the given encoding, if the client supports it.
#[must_use]
pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
self.send_compression_encodings.enable(encoding);
self
}
/// Limits the maximum size of a decoded message.
///
/// Default: `4MB`
#[must_use]
pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
self.max_decoding_message_size = Some(limit);
self
}
/// Limits the maximum size of an encoded message.
///
/// Default: `usize::MAX`
#[must_use]
pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
self.max_encoding_message_size = Some(limit);
self
}
}
impl<T, B> tonic::codegen::Service<http::Request<B>> for ServerServer<T>
where
T: Server,
B: Body + std::marker::Send + 'static,
B::Error: Into<StdError> + std::marker::Send + 'static,
{
type Response = http::Response<tonic::body::Body>;
type Error = std::convert::Infallible;
type Future = BoxFuture<Self::Response, Self::Error>;
fn poll_ready(
&mut self,
_cx: &mut Context<'_>,
) -> Poll<std::result::Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, req: http::Request<B>) -> Self::Future {
match req.uri().path() {
"/serverpb.Server/SendTransaction" => {
#[allow(non_camel_case_types)]
struct SendTransactionSvc<T: Server>(pub Arc<T>);
impl<T: Server> tonic::server::UnaryService<super::SendRequest>
for SendTransactionSvc<T> {
type Response = super::SendResponse;
type Future = BoxFuture<
tonic::Response<Self::Response>,
tonic::Status,
>;
fn call(
&mut self,
request: tonic::Request<super::SendRequest>,
) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move {
<T as Server>::send_transaction(&inner, request).await
};
Box::pin(fut)
}
}
let accept_compression_encodings = self.accept_compression_encodings;
let send_compression_encodings = self.send_compression_encodings;
let max_decoding_message_size = self.max_decoding_message_size;
let max_encoding_message_size = self.max_encoding_message_size;
let inner = self.inner.clone();
let fut = async move {
let method = SendTransactionSvc(inner);
let codec = tonic_prost::ProstCodec::default();
let mut grpc = tonic::server::Grpc::new(codec)
.apply_compression_config(
accept_compression_encodings,
send_compression_encodings,
)
.apply_max_message_size_config(
max_decoding_message_size,
max_encoding_message_size,
);
let res = grpc.unary(method, req).await;
Ok(res)
};
Box::pin(fut)
}
"/serverpb.Server/GetHealth" => {
#[allow(non_camel_case_types)]
struct GetHealthSvc<T: Server>(pub Arc<T>);
impl<T: Server> tonic::server::UnaryService<super::HealthRequest>
for GetHealthSvc<T> {
type Response = super::HealthResponse;
type Future = BoxFuture<
tonic::Response<Self::Response>,
tonic::Status,
>;
fn call(
&mut self,
request: tonic::Request<super::HealthRequest>,
) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move {
<T as Server>::get_health(&inner, request).await
};
Box::pin(fut)
}
}
let accept_compression_encodings = self.accept_compression_encodings;
let send_compression_encodings = self.send_compression_encodings;
let max_decoding_message_size = self.max_decoding_message_size;
let max_encoding_message_size = self.max_encoding_message_size;
let inner = self.inner.clone();
let fut = async move {
let method = GetHealthSvc(inner);
let codec = tonic_prost::ProstCodec::default();
let mut grpc = tonic::server::Grpc::new(codec)
.apply_compression_config(
accept_compression_encodings,
send_compression_encodings,
)
.apply_max_message_size_config(
max_decoding_message_size,
max_encoding_message_size,
);
let res = grpc.unary(method, req).await;
Ok(res)
};
Box::pin(fut)
}
_ => {
Box::pin(async move {
let mut response = http::Response::new(
tonic::body::Body::default(),
);
let headers = response.headers_mut();
headers
.insert(
tonic::Status::GRPC_STATUS,
(tonic::Code::Unimplemented as i32).into(),
);
headers
.insert(
http::header::CONTENT_TYPE,
tonic::metadata::GRPC_CONTENT_TYPE,
);
Ok(response)
})
}
}
}
}
impl<T> Clone for ServerServer<T> {
fn clone(&self) -> Self {
let inner = self.inner.clone();
Self {
inner,
accept_compression_encodings: self.accept_compression_encodings,
send_compression_encodings: self.send_compression_encodings,
max_decoding_message_size: self.max_decoding_message_size,
max_encoding_message_size: self.max_encoding_message_size,
}
}
}
/// Generated gRPC service name
pub const SERVICE_NAME: &str = "serverpb.Server";
impl<T> tonic::server::NamedService for ServerServer<T> {
const NAME: &'static str = SERVICE_NAME;
}
}
+16 -6
View File
@@ -109,25 +109,35 @@ impl SwqosClientTrait for SoyasClient {
let serialized_tx = bincode::serialize(transaction)?;
let connection = self.connection.load_full();
if Self::try_send_bytes(&connection, &serialized_tx).await.is_err() {
eprintln!(" [soyas] {} submission failed, reconnecting", trade_type);
if crate::common::sdk_log::sdk_log_enabled() {
crate::common::sdk_log::log_swqos_submission_failed("Soyas", trade_type, start_time.elapsed(), "reconnecting");
}
self.reconnect().await?;
let connection = self.connection.load_full();
if let Err(e) = Self::try_send_bytes(&connection, &serialized_tx).await {
eprintln!(" [soyas] {} submission failed: {:?}", trade_type, e);
if crate::common::sdk_log::sdk_log_enabled() {
crate::common::sdk_log::log_swqos_submission_failed("Soyas", trade_type, start_time.elapsed(), &e);
}
return Err(e.into());
}
}
if crate::common::sdk_log::sdk_log_enabled() {
crate::common::sdk_log::log_swqos_submitted("Soyas", trade_type, start_time.elapsed());
}
let start_time = Instant::now();
match poll_transaction_confirmation(&self.rpc_client, *signature, wait_confirmation).await {
Ok(_) => (),
Err(e) => {
println!(" signature: {:?}", signature);
println!(" [soyas] {} confirmation failed: {:?}", trade_type, start_time.elapsed());
if crate::common::sdk_log::sdk_log_enabled() {
println!(" signature: {:?}", signature);
println!(" [{:width$}] {} confirmation failed: {:?}", "Soyas", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
}
return Err(e);
}
}
if wait_confirmation {
if wait_confirmation && crate::common::sdk_log::sdk_log_enabled() {
println!(" signature: {:?}", signature);
println!(" [soyas] {} confirmed: {:?}", trade_type, start_time.elapsed());
println!(" [{:width$}] {} confirmed: {:?}", "Soyas", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
}
Ok(())
}
+25 -5
View File
@@ -161,22 +161,42 @@ impl SwqosClientTrait for SpeedlandingClient {
};
if need_retry {
if crate::common::sdk_log::sdk_log_enabled() {
eprintln!(" [speedlanding] {} send failed or timeout, reconnecting", trade_type);
eprintln!(" [Speedlanding] {} submission failed after {:?}, reconnecting", trade_type, start_time.elapsed());
}
let connection = self.ensure_connected().await?;
send_result =
timeout(SEND_TIMEOUT, Self::try_send_bytes(&connection, &*buf_guard)).await;
}
send_result.context("Speedlanding QUIC send timeout")??;
match send_result.context("Speedlanding QUIC send timeout") {
Ok(Ok(())) => {
if crate::common::sdk_log::sdk_log_enabled() {
crate::common::sdk_log::log_swqos_submitted("Speedlanding", trade_type, start_time.elapsed());
}
}
Ok(Err(e)) => {
if crate::common::sdk_log::sdk_log_enabled() {
crate::common::sdk_log::log_swqos_submission_failed("Speedlanding", trade_type, start_time.elapsed(), &e);
}
return Err(e.into());
}
Err(e) => {
if crate::common::sdk_log::sdk_log_enabled() {
crate::common::sdk_log::log_swqos_submission_failed("Speedlanding", trade_type, start_time.elapsed(), "timeout");
}
return Err(e.into());
}
}
match poll_transaction_confirmation(&self.rpc_client, signature, wait_confirmation).await {
Ok(_) => (),
Err(e) => {
if crate::common::sdk_log::sdk_log_enabled() {
println!(" signature: {:?}", signature);
println!(
" [speedlanding] {} confirmation failed: {:?}",
" [{:width$}] {} confirmation failed: {:?}",
"Speedlanding",
trade_type,
start_time.elapsed()
start_time.elapsed(),
width = crate::common::sdk_log::SWQOS_LABEL_WIDTH
);
}
return Err(e);
@@ -184,7 +204,7 @@ impl SwqosClientTrait for SpeedlandingClient {
}
if wait_confirmation && crate::common::sdk_log::sdk_log_enabled() {
println!(" signature: {:?}", signature);
println!(" [speedlanding] {} confirmed: {:?}", trade_type, start_time.elapsed());
println!(" [{:width$}] {} confirmed: {:?}", "Speedlanding", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
}
Ok(())
}
+8 -6
View File
@@ -168,13 +168,13 @@ impl StelliumClient {
if let Ok(response_json) = serde_json::from_str::<serde_json::Value>(&response_text) {
if crate::common::sdk_log::sdk_log_enabled() {
if response_json.get("result").is_some() {
println!(" [Stellium] {} submitted: {:?}", trade_type, start_time.elapsed());
crate::common::sdk_log::log_swqos_submitted("Stellium", trade_type, start_time.elapsed());
} else if let Some(_error) = response_json.get("error") {
eprintln!(" [Stellium] {} submission failed: {:?}", trade_type, _error);
crate::common::sdk_log::log_swqos_submission_failed("Stellium", trade_type, start_time.elapsed(), _error);
}
}
} else if crate::common::sdk_log::sdk_log_enabled() {
eprintln!(" [Stellium] {} submission failed: {:?}", trade_type, response_text);
crate::common::sdk_log::log_swqos_submission_failed("Stellium", trade_type, start_time.elapsed(), response_text);
}
let start_time: Instant = Instant::now();
@@ -184,9 +184,11 @@ impl StelliumClient {
if crate::common::sdk_log::sdk_log_enabled() {
println!(" signature: {:?}", signature);
println!(
" [Stellium] {} confirmation failed: {:?}",
" [{:width$}] {} confirmation failed: {:?}",
"Stellium",
trade_type,
start_time.elapsed()
start_time.elapsed(),
width = crate::common::sdk_log::SWQOS_LABEL_WIDTH
);
}
return Err(e);
@@ -194,7 +196,7 @@ impl StelliumClient {
}
if wait_confirmation && crate::common::sdk_log::sdk_log_enabled() {
println!(" signature: {:?}", signature);
println!(" [Stellium] {} confirmed: {:?}", trade_type, start_time.elapsed());
println!(" [{:width$}] {} confirmed: {:?}", "Stellium", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
}
Ok(())
+4 -4
View File
@@ -209,12 +209,12 @@ impl TemporalClient {
if let Ok(response_json) = serde_json::from_str::<serde_json::Value>(&response_text) {
if response_json.get("result").is_some() {
println!(" [nozomi] {} submitted: {:?}", trade_type, start_time.elapsed());
crate::common::sdk_log::log_swqos_submitted("nozomi", trade_type, start_time.elapsed());
} else if let Some(_error) = response_json.get("error") {
// eprintln!("nozomi transaction submission failed: {:?}", _error);
crate::common::sdk_log::log_swqos_submission_failed("nozomi", trade_type, start_time.elapsed(), _error);
}
} else {
eprintln!(" [nozomi] {} submission failed: {:?}", trade_type, response_text);
crate::common::sdk_log::log_swqos_submission_failed("nozomi", trade_type, start_time.elapsed(), response_text);
}
let start_time: Instant = Instant::now();
@@ -232,7 +232,7 @@ impl TemporalClient {
}
if wait_confirmation {
println!(" signature: {:?}", signature);
println!(" [nozomi] {} confirmed: {:?}", trade_type, start_time.elapsed());
println!(" [{:width$}] {} confirmed: {:?}", "nozomi", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
}
Ok(())
+160 -35
View File
@@ -1,12 +1,12 @@
use crate::swqos::common::{
default_http_client_builder, poll_transaction_confirmation, serialize_transaction_and_encode,
default_http_client_builder, poll_transaction_confirmation,
};
use rand::seq::IndexedRandom;
use reqwest::Client;
use serde_json::json;
use std::{sync::Arc, time::Instant};
use solana_transaction_status::UiTransactionEncoding;
use std::{sync::Arc, time::Instant, time::Duration};
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::task::JoinHandle;
use bincode;
use crate::swqos::SwqosClientTrait;
use crate::swqos::{SwqosType, TradeType};
@@ -21,6 +21,8 @@ pub struct ZeroSlotClient {
pub auth_token: String,
pub rpc_client: Arc<SolanaRpcClient>,
pub http_client: Client,
pub ping_handle: Arc<tokio::sync::Mutex<Option<JoinHandle<()>>>>,
pub stop_ping: Arc<AtomicBool>,
}
#[async_trait::async_trait]
@@ -60,7 +62,79 @@ impl ZeroSlotClient {
pub fn new(rpc_url: String, endpoint: String, auth_token: String) -> Self {
let rpc_client = SolanaRpcClient::new(rpc_url);
let http_client = default_http_client_builder().build().unwrap();
Self { rpc_client: Arc::new(rpc_client), endpoint, auth_token, http_client }
let client = Self {
rpc_client: Arc::new(rpc_client),
endpoint,
auth_token,
http_client,
ping_handle: Arc::new(tokio::sync::Mutex::new(None)),
stop_ping: Arc::new(AtomicBool::new(false)),
};
// Start ping task
let client_clone = client.clone();
tokio::spawn(async move {
client_clone.start_ping_task().await;
});
client
}
/// Start periodic ping task to keep connections active
async fn start_ping_task(&self) {
let endpoint = self.endpoint.clone();
let auth_token = self.auth_token.clone();
let http_client = self.http_client.clone();
let stop_ping = self.stop_ping.clone();
let handle = tokio::spawn(async move {
// Immediate first ping to warm connection and reduce first-submit cold start latency
if let Err(e) = Self::send_ping_request(&http_client, &endpoint, &auth_token).await {
if crate::common::sdk_log::sdk_log_enabled() {
eprintln!("0slot ping request failed: {}", e);
}
}
let mut interval = tokio::time::interval(Duration::from_secs(30)); // 30s keepalive under 65s server timeout
loop {
interval.tick().await;
if stop_ping.load(Ordering::Relaxed) {
break;
}
if let Err(e) = Self::send_ping_request(&http_client, &endpoint, &auth_token).await {
if crate::common::sdk_log::sdk_log_enabled() {
eprintln!("0slot ping request failed: {}", e);
}
}
}
});
// Update ping_handle - use Mutex to safely update
{
let mut ping_guard = self.ping_handle.lock().await;
if let Some(old_handle) = ping_guard.as_ref() {
old_handle.abort();
}
*ping_guard = Some(handle);
}
}
/// Send ping request: POST with getHealth method (Keep Alive). Free operation, not counted toward TPS.
async fn send_ping_request(
http_client: &Client,
endpoint: &str,
auth_token: &str,
) -> Result<()> {
let url = format!("{}/?api-key={}", endpoint, auth_token);
let response = http_client
.post(&url)
.header("Content-Type", "application/json")
.timeout(Duration::from_millis(1500))
.body(r#"{"jsonrpc":"2.0","id":1,"method":"getHealth"}"#)
.send()
.await?;
let _ = response.bytes().await;
Ok(())
}
pub async fn send_transaction(
@@ -70,58 +144,91 @@ impl ZeroSlotClient {
wait_confirmation: bool,
) -> Result<()> {
let start_time = Instant::now();
let (content, signature) =
serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64)?;
let request_body = serde_json::to_string(&json!({
"jsonrpc": "2.0",
"id": 1,
"method": "sendTransaction",
"params": [
content,
{ "encoding": "base64", "skipPreflight": true }
]
}))?;
// Binary-Tx: Send raw binary transaction bytes directly
// This is faster than JSON-RPC as it avoids unnecessary encoding/decoding
let tx_bytes = bincode::serialize(transaction)?;
// Build URL for Binary-Tx endpoint: {endpoint}/txb?api-key={auth_token}
let mut url = String::with_capacity(self.endpoint.len() + self.auth_token.len() + 20);
url.push_str(&self.endpoint);
url.push_str("/?api-key=");
url.push_str("/txb?api-key=");
url.push_str(&self.auth_token);
// 4. Use `text().await?` directly, avoiding async JSON parsing from `json().await?`
let response_text = self
// Send binary transaction directly
let response = self
.http_client
.post(&url)
.body(request_body) // Pass string directly, avoiding `json()` overhead
.header("Content-Type", "application/json") // Explicitly specify JSON header
.header("User-Agent", "") // Optional: 0slot recommends empty User-Agent
.body(tx_bytes)
.send()
.await?
.text()
.await?;
// 5. Use `serde_json::from_str()` to parse JSON, reducing extra wait from `.json().await?`
if let Ok(response_json) = serde_json::from_str::<serde_json::Value>(&response_text) {
if response_json.get("result").is_some() {
println!(" [0slot] {} submitted: {:?}", trade_type, start_time.elapsed());
} else if let Some(_error) = response_json.get("error") {
eprintln!(" [0slot] {} submission failed: {:?}", trade_type, _error);
let status = response.status();
let response_text = response.text().await?;
// Binary-Tx returns JSON-RPC 2.0 format responses
// 200: success with result field containing signature, or error field with code/message
// 403: api-key error (null, doesn't exist, or expired)
// 419: rate limit exceeded
// 500: submission failed
match status.as_u16() {
200 => {
if let Ok(json_value) = serde_json::from_str::<serde_json::Value>(&response_text) {
if json_value.get("result").is_some() {
crate::common::sdk_log::log_swqos_submitted("0slot", trade_type, start_time.elapsed());
} else if let Some(error) = json_value.get("error") {
let code = error.get("code")
.and_then(|c| c.as_i64())
.map(|c| c.to_string())
.unwrap_or_else(|| "unknown".to_string());
let message = error.get("message")
.and_then(|m| m.as_str())
.unwrap_or("unknown error");
crate::common::sdk_log::log_swqos_submission_failed("0slot", trade_type, start_time.elapsed(), format!("code {}: {}", code, message));
return Err(anyhow::anyhow!("0slot Binary-Tx error: {}", message));
} else {
crate::common::sdk_log::log_swqos_submission_failed("0slot", trade_type, start_time.elapsed(), format!("unexpected JSON: {}", response_text));
return Err(anyhow::anyhow!("0slot Binary-Tx unexpected JSON: {}", response_text));
}
} else {
crate::common::sdk_log::log_swqos_submission_failed("0slot", trade_type, start_time.elapsed(), format!("invalid JSON: {}", response_text));
return Err(anyhow::anyhow!("0slot Binary-Tx invalid JSON: {}", response_text));
}
}
403 => {
crate::common::sdk_log::log_swqos_submission_failed("0slot", trade_type, start_time.elapsed(), response_text.clone());
return Err(anyhow::anyhow!("0slot API key error: {}", response_text));
}
419 => {
crate::common::sdk_log::log_swqos_submission_failed("0slot", trade_type, start_time.elapsed(), response_text.clone());
return Err(anyhow::anyhow!("0slot rate limit exceeded"));
}
500 => {
crate::common::sdk_log::log_swqos_submission_failed("0slot", trade_type, start_time.elapsed(), "submission failed".to_string());
return Err(anyhow::anyhow!("0slot transaction submission failed"));
}
_ => {
crate::common::sdk_log::log_swqos_submission_failed("0slot", trade_type, start_time.elapsed(), format!("status {} body: {}", status, response_text));
return Err(anyhow::anyhow!("0slot Binary-Tx failed with status {}: {}", status, response_text));
}
} else {
eprintln!(" [0slot] {} submission failed: {:?}", trade_type, response_text);
}
let start_time: Instant = Instant::now();
// Get transaction signature from the transaction for confirmation polling
let signature = transaction.signatures[0];
let start_time = Instant::now();
match poll_transaction_confirmation(&self.rpc_client, signature, wait_confirmation).await {
Ok(_) => (),
Err(e) => {
println!(" signature: {:?}", signature);
println!(" [0slot] {} confirmation failed: {:?}", trade_type, start_time.elapsed());
println!(" [{:width$}] {} confirmation failed: {:?}", "0slot", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
return Err(e);
}
}
if wait_confirmation {
println!(" signature: {:?}", signature);
println!(" [0slot] {} confirmed: {:?}", trade_type, start_time.elapsed());
println!(" [{:width$}] {} confirmed: {:?}", "0slot", trade_type, start_time.elapsed(), width = crate::common::sdk_log::SWQOS_LABEL_WIDTH);
}
Ok(())
@@ -139,3 +246,21 @@ impl ZeroSlotClient {
Ok(())
}
}
impl Drop for ZeroSlotClient {
fn drop(&mut self) {
// Ensure ping task stops when client is destroyed
self.stop_ping.store(true, Ordering::Relaxed);
// Try to stop ping task immediately
// Use tokio::spawn to avoid blocking Drop
let ping_handle = self.ping_handle.clone();
tokio::spawn(async move {
let mut ping_guard = ping_handle.lock().await;
if let Some(handle) = ping_guard.as_ref() {
handle.abort();
}
*ping_guard = None;
});
}
}
+5
View File
@@ -5,6 +5,11 @@
//! - **Dedicated threads** (opt-in via `with_dedicated_sender_threads`): N OS threads run sender work only, optionally pinned to cores.
//! - **Arc**: Shared data behind `Arc` → clone = refcount increment (no data copy).
//! - **Refs**: `build_transaction` takes refs only; worker path avoids extra clones.
//!
//! **Core affinity & latency:** Each job is assigned a core (round-robin from `effective_core_ids`). When a worker runs a job,
//! it sets thread affinity to that core. If that core is busy with other work (e.g. node sync, bot logic), SWQOS submit on that
//! core will compete for CPU and latency can increase. For lowest latency, reserve a subset of cores for SWQOS only via
//! `with_dedicated_sender_threads(Some(indices))` and avoid running other CPU-heavy work on those core indices.
use anyhow::{anyhow, Result};
use crossbeam_queue::ArrayQueue;
+60 -69
View File
@@ -118,32 +118,45 @@ impl TradeExecutor for GenericTradeExecutor {
if crate::common::sdk_log::sdk_log_enabled() {
let dir = if is_buy { "Buy" } else { "Sell" };
println!();
if let (Some(start_us), Some(end_us)) = (timing_start_us, build_end_us) {
println!(
" [SDK] {} build_instructions: {:.4} ms",
" [SDK][{:width$}] {} build_instructions: {:.4} ms",
"-",
dir,
(end_us - start_us) as f64 / 1000.0
(end_us - start_us) as f64 / 1000.0,
width = crate::common::sdk_log::SWQOS_LABEL_WIDTH
);
}
if let (Some(start_us), Some(end_us)) = (timing_start_us, before_submit_us) {
println!(
" [SDK] {} before_submit: {:.4} ms",
" [SDK][{:width$}] {} before_submit: {:.4} ms",
"-",
dir,
(end_us - start_us) as f64 / 1000.0
(end_us - start_us) as f64 / 1000.0,
width = crate::common::sdk_log::SWQOS_LABEL_WIDTH
);
}
println!(
" [SDK] {} simulate (dry-run): {:.4} ms",
" [SDK][{:width$}] {} simulate (dry-run): {:.4} ms",
"-",
dir,
send_elapsed.as_secs_f64() * 1000.0
send_elapsed.as_secs_f64() * 1000.0,
width = crate::common::sdk_log::SWQOS_LABEL_WIDTH
);
println!(
" [SDK][{:width$}] {} total: {:.4} ms",
"-",
dir,
total_elapsed.as_secs_f64() * 1000.0,
width = crate::common::sdk_log::SWQOS_LABEL_WIDTH
);
println!(" [SDK] {} total: {:.4} ms", dir, total_elapsed.as_secs_f64() * 1000.0);
}
return result;
}
let need_confirm = params.wait_transaction_confirmed;
let need_confirm = params.wait_tx_confirmed;
let sender_config = params.sender_concurrency_config();
let result = execute_parallel(
params.swqos_clients.as_slice(),
@@ -185,32 +198,14 @@ impl TradeExecutor for GenericTradeExecutor {
let confirm_done_us = log_enabled.then(crate::common::clock::now_micros);
if log_enabled {
let dir = if is_buy { "Buy" } else { "Sell" };
if let Some(start_us) = timing_start_us {
if let Some(end_us) = build_end_us {
println!(
" [SDK] {} build_instructions: {:.4} ms",
dir,
(end_us - start_us) as f64 / 1000.0
);
}
if let Some(end_us) = before_submit_us {
println!(
" [SDK] {} before_submit: {:.4} ms",
dir,
(end_us - start_us) as f64 / 1000.0
);
}
if let Some(confirm_us) = confirm_done_us {
let total_ms = (confirm_us - start_us) as f64 / 1000.0;
for (swqos_type, submit_done_us) in submit_timings_ref {
let submit_ms =
(*submit_done_us - start_us).max(0) as f64 / 1000.0;
let confirmed_ms =
(confirm_us - *submit_done_us).max(0) as f64 / 1000.0;
println!(" [SDK] {} {:?} submit: {:.4} ms, confirmed: {:.4} ms, total: {:.4} ms", dir, swqos_type, submit_ms, confirmed_ms, total_ms);
}
}
}
crate::common::sdk_log::print_sdk_timing_block(
dir,
timing_start_us,
build_end_us,
before_submit_us,
submit_timings_ref,
confirm_done_us,
);
}
match poll_res {
Ok(_) => (true, signatures, None),
@@ -222,31 +217,17 @@ impl TradeExecutor for GenericTradeExecutor {
};
Ok(confirm_result)
} else {
// Not waiting for confirmation: confirmed is not measured (-); total is per-channel submit time only.
if log_enabled {
let dir = if is_buy { "Buy" } else { "Sell" };
if let Some(start_us) = timing_start_us {
if let Some(end_us) = build_end_us {
println!(
" [SDK] {} build_instructions: {:.4} ms",
dir,
(end_us - start_us) as f64 / 1000.0
);
}
if let Some(end_us) = before_submit_us {
println!(
" [SDK] {} before_submit: {:.4} ms",
dir,
(end_us - start_us) as f64 / 1000.0
);
}
for (swqos_type, submit_done_us) in submit_timings_ref {
let submit_ms = (*submit_done_us - start_us).max(0) as f64 / 1000.0;
println!(
" [SDK] {} {:?} submit: {:.4} ms, confirmed: -, total: {:.4} ms",
dir, swqos_type, submit_ms, submit_ms
);
}
}
crate::common::sdk_log::print_sdk_timing_block(
dir,
timing_start_us,
build_end_us,
before_submit_us,
submit_timings_ref,
None,
);
}
Ok((ok, signatures, err))
};
@@ -377,37 +358,47 @@ mod tests {
let dir = "Buy";
let build_ms = 12.34;
let before_submit_ms = 15.67;
let w = 12usize; // same as crate::common::sdk_log::SWQOS_LABEL_WIDTH
println!("\n--- 1. 构建指令耗时 / 提交前耗时(各打印一次,统一 ms,保留 4 位小数)---\n");
println!(" [SDK] {} build_instructions: {:.4} ms", dir, build_ms);
println!(" [SDK] {} before_submit: {:.4} ms", dir, before_submit_ms);
println!(" [SDK][{:width$}] {} build_instructions: {:.4} ms", "-", dir, build_ms, width = w);
println!(" [SDK][{:width$}] {} before_submit: {:.4} ms", "-", dir, before_submit_ms, width = w);
println!("\n--- 2. 每个 SWQOS 独立耗时:submit=起点→该通道返回, confirmed=该通道提交→链上确认, total=起点→链上确认 ---\n");
println!("\n--- 2. 每个 SWQOS 独立耗时:submit_done=起点→该通道提交完成, confirmed=该通道提交→链上确认, total=起点→链上确认 ---\n");
for (swqos_type, submit_ms, confirmed_ms, total_ms) in [
(SwqosType::Jito, 45.12, 83.38, 128.50),
(SwqosType::Helius, 52.30, 76.20, 128.50),
(SwqosType::ZeroSlot, 48.90, 79.60, 128.50),
] {
println!(
" [SDK] {} {:?} submit: {:.4} ms, confirmed: {:.4} ms, total: {:.4} ms",
dir, swqos_type, submit_ms, confirmed_ms, total_ms
" [SDK][{:width$}] {} submit_done: {:.4} ms, confirmed: {:.4} ms, total: {:.4} ms",
swqos_type.as_str(),
dir,
submit_ms,
confirmed_ms,
total_ms,
width = w
);
}
println!("\n--- 3. 不等待链上确认时:每行 total = 该通道 submit 耗时(独立---\n");
println!("\n--- 3. 不等待链上确认时:每行 total = 该通道 submit_done(提交完成总耗时---\n");
for (swqos_type, submit_ms, total_ms) in
[(SwqosType::Jito, 44.20, 44.20), (SwqosType::Helius, 51.80, 51.80)]
{
println!(
" [SDK] {} {:?} submit: {:.4} ms, confirmed: -, total: {:.4} ms",
dir, swqos_type, submit_ms, total_ms
" [SDK][{:width$}] {} submit_done: {:.4} ms, confirmed: -, total: {:.4} ms",
swqos_type.as_str(),
dir,
submit_ms,
total_ms,
width = w
);
}
println!("\n--- 4. Simulate 模式(build/before_submit 仍从 grpc_recv_us 起算)---\n");
println!(" [SDK] {} build_instructions: {:.4} ms", dir, build_ms);
println!(" [SDK] {} before_submit: {:.4} ms", dir, before_submit_ms);
println!(" [SDK] {} simulate (dry-run): {:.4} ms", dir, 8.50);
println!(" [SDK] {} total: {:.4} ms", dir, 36.51);
println!(" [SDK][{:width$}] {} build_instructions: {:.4} ms", "-", dir, build_ms, width = w);
println!(" [SDK][{:width$}] {} before_submit: {:.4} ms", "-", dir, before_submit_ms, width = w);
println!(" [SDK][{:width$}] {} simulate (dry-run): {:.4} ms", "-", dir, 8.50, width = w);
println!(" [SDK][{:width$}] {} total: {:.4} ms", "-", dir, 36.51, width = w);
println!();
}
}
+1 -1
View File
@@ -62,7 +62,7 @@ pub struct SwapParams {
pub slippage_basis_points: Option<u64>,
pub address_lookup_table_account: Option<AddressLookupTableAccount>,
pub recent_blockhash: Option<Hash>,
pub wait_transaction_confirmed: bool,
pub wait_tx_confirmed: bool,
pub protocol_params: DexParamEnum,
pub open_seed_optimize: bool,
/// Arc<Vec<..>> so cloning from infrastructure is a single Arc clone.