From a003fd4d2f5658123c4a426ef5a3991c0de3b72d Mon Sep 17 00:00:00 2001 From: Wood Date: Sat, 7 Mar 2026 10:47:31 +0800 Subject: [PATCH] feat: Astralane QUIC, code review fixes, README & example updates - Add Astralane QUIC client (astralane_quic.rs) and SwqosTransport::Quic - README: Astralane QUIC usage, remove third-party doc links, add QUIC to examples - Code review: API key not in logs, PumpSwap no clone, PDA Result, SELL_DISCRIMINATOR, ensure_wsol_ata refactor, tracing in astralane, only supports fix - instruction/utils: pumpfun/pumpswap PDA & discriminator unit tests - trading_client example: SwqosTransport, Astralane QUIC in config - cli_trading: fix all unused variable warnings (_prefix) - Add docs/CODE_REVIEW_REPORT.md Made-with: Cursor --- Cargo.toml | 3 +- README.md | 33 +++- docs/CODE_REVIEW_REPORT.md | 137 +++++++++++++++ examples/cli_trading/src/main.rs | 64 +++---- examples/trading_client/src/main.rs | 4 +- src/constants/swqos.rs | 3 + src/instruction/pumpfun.rs | 24 +-- src/instruction/pumpswap.rs | 44 ++--- src/instruction/utils/pumpfun.rs | 36 +++- src/instruction/utils/pumpswap.rs | 32 +++- src/lib.rs | 195 ++++++++++----------- src/swqos/astralane.rs | 198 ++++++++++++---------- src/swqos/astralane_quic.rs | 252 ++++++++++++++++++++++++++++ src/swqos/mod.rs | 39 +++-- 14 files changed, 778 insertions(+), 286 deletions(-) create mode 100644 docs/CODE_REVIEW_REPORT.md create mode 100644 src/swqos/astralane_quic.rs diff --git a/Cargo.toml b/Cargo.toml index 067cdce..8fb7da0 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -107,7 +107,8 @@ parking_lot = "0.12" arc-swap = "1.7" sha2 = "0.10" tonic-prost = "0.14.2" -quinn = {version = "0.11", default-features = false, features = ["rustls"]} +quinn = { version = "0.11", default-features = false, features = ["rustls"] } +rcgen = "0.13" # Performance optimization dependencies crossbeam-queue = "0.3" diff --git a/README.md b/README.md index 6fa1c8b..2625017 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,7 @@ - [⚡ Trading Parameters](#-trading-parameters) - [📊 Usage Examples Summary Table](#-usage-examples-summary-table) - [⚙️ SWQoS Service Configuration](#️-swqos-service-configuration) + - [Astralane QUIC (Low-Latency)](#astralane-quic-low-latency) - [🔧 Middleware System](#-middleware-system) - [🔍 Address Lookup Tables](#-address-lookup-tables) - [🔍 Nonce Cache](#-nonce-cache) @@ -119,6 +120,7 @@ let swqos_configs: Vec = vec![ SwqosConfig::Default(rpc_url.clone()), SwqosConfig::Jito("your uuid".to_string(), SwqosRegion::Frankfurt, None), SwqosConfig::Bloxroute("your api_token".to_string(), SwqosRegion::Frankfurt, None), + SwqosConfig::Astralane("your_astralane_api_key".to_string(), SwqosRegion::Frankfurt, None, Some(SwqosTransport::Quic)), ]; // Create TradeConfig instance let trade_config = TradeConfig::new(rpc_url, swqos_configs, commitment); @@ -257,6 +259,29 @@ let bloxroute_config = SwqosConfig::Bloxroute( When using multiple MEV services, you need to use `Durable Nonce`. You need to use the `fetch_nonce_info` function to get the latest `nonce` value, and use it as the `durable_nonce` when trading. +#### Astralane QUIC (Low-Latency) + +Astralane supports both HTTP and **QUIC** transport. QUIC reduces connection overhead and can lower submission latency. To use the QUIC channel, pass `Some(SwqosTransport::Quic)` as the fourth parameter of `SwqosConfig::Astralane`. Astralane’s QUIC service uses a **single endpoint** (no per-region endpoints); the SDK ignores the `region` (and optional custom URL) when QUIC is selected. You can pass the same region as your other SWQoS configs for consistency. + +```rust +use sol_trade_sdk::{SwqosConfig, SwqosRegion, SwqosTransport}; + +// Astralane over QUIC (low-latency); region is ignored (single QUIC endpoint) +let swqos_configs: Vec = vec![ + SwqosConfig::Default(rpc_url.clone()), + SwqosConfig::Astralane( + "your_astralane_api_key".to_string(), + SwqosRegion::Frankfurt, // same as other services; ignored for QUIC + None, + Some(SwqosTransport::Quic), + ), +]; +// Then create TradeConfig / TradingClient as usual with swqos_configs +``` + +- **HTTP** (default): use `None` or `Some(SwqosTransport::Http)`; region and optional custom URL apply. +- **QUIC**: use `Some(SwqosTransport::Quic)`; the SDK uses a single QUIC endpoint and ignores region. Same API key as HTTP. + --- ### 🔧 Middleware System @@ -297,10 +322,10 @@ You can apply for a key through the official website: [Community Website](https: - **ZeroSlot**: Zero-latency transactions - **Temporal**: Time-sensitive transactions - **Bloxroute**: Blockchain network acceleration -- **FlashBlock**: High-speed transaction execution with API key authentication - [Official Documentation](https://doc.flashblock.trade/) -- **BlockRazor**: High-speed transaction execution with API key authentication - [Official Documentation](https://blockrazor.gitbook.io/blockrazor/) -- **Node1**: High-speed transaction execution with API key authentication - [Official Documentation](https://node1.me/docs.html) -- **Astralane**: Blockchain network acceleration +- **FlashBlock**: High-speed transaction execution with API key authentication +- **BlockRazor**: High-speed transaction execution with API key authentication +- **Node1**: High-speed transaction execution with API key authentication +- **Astralane**: Blockchain network acceleration (supports HTTP and QUIC; see [Astralane QUIC](#astralane-quic-low-latency) above) ## 📁 Project Structure diff --git a/docs/CODE_REVIEW_REPORT.md b/docs/CODE_REVIEW_REPORT.md new file mode 100644 index 0000000..b1e795e --- /dev/null +++ b/docs/CODE_REVIEW_REPORT.md @@ -0,0 +1,137 @@ +# sol-trade-sdk 代码审查报告 + +审查维度：**逻辑准确性**、**可读性**、**模块化**、**超低延迟**、**代码质量**、**安全性**。 + +--- + +## 1. 代码逻辑准确性 + +### 1.1 Instruction 与 IDL / 官方行为 + +| 模块 | 结论 | 说明 | +|------|------|------| +| PumpFun buy/sell | ✅ 一致 | 账户顺序、discriminator、track_volume 与 `idl/pump.json` 一致；cashback 时 remainingAccounts 顺序正确 | +| PumpSwap buy/sell | ✅ 一致 | 与 `idl/pump_amm.json` 一致；sell cashback 使用 quote_mint ATA | +| PDA 推导 | ✅ 一致 | bonding_curve_v2、pool_v2、user_volume_accumulator、creator_vault 等 seeds 与官方一致 | + +### 1.2 需修正的逻辑/风格 + +- **`src/instruction/utils/pumpswap.rs` 约 258、291 行**：`let program_id: &Pubkey = &&accounts::AMM_PROGRAM` 为双重引用，易误导且多余。建议改为 `&accounts::AMM_PROGRAM`。 + +--- + +## 2. 代码可读性 + +### 2.1 命名与注释 + +- 多数模块有中英文注释，instruction 与 IDL 的对应关系有标注。 +- **建议**：`src/instruction/pumpfun.rs` 约 259 行 sell 的 discriminator 使用魔法数组 `[51, 230, 133, ...]`，建议改为 `SELL_DISCRIMINATOR` 常量（与 buy 路径一致）。 + +### 2.2 错误信息与文案 + +- **建议**：`src/lib.rs` 中 “Current version only support” 应为 “only supports”；类似拼写/语法可统一检查。 + +### 2.3 过长函数 + +- **建议**：`src/lib.rs` 的 `ensure_wsol_ata` 可拆为「入口 + 重试循环」与「单次尝试 + 结果判断」，便于单测和阅读。 + +--- + +## 3. 模块化 + +### 3.1 职责与分层 + +- **instruction**：按协议分（pumpfun / pumpswap / bonk / raydium_* 等），实现 `InstructionBuilder`，边界清晰。 +- **instruction/utils**：PDA、常量、类型、池子解析与上层「组指令」分工明确。 +- **swqos**：按提供商分模块，common 放序列化、确认轮询、HTTP 客户端，无循环依赖。 +- **结论**：分层合理，模块化良好。 + +--- + +## 4. 超低延迟 + +### 4.1 必须改：避免多余 clone + +| 位置 | 问题 | 建议 | +|------|------|------| +| `src/instruction/pumpswap.rs` 约 232、429 行 | `Instruction { accounts: accounts.clone(), data }` 对已拥有的 `Vec` 做完整 clone | 改为直接移动：`Instruction { program_id, accounts, data }`，不再 clone | + +### 4.2 建议 + +- 若多路 SWQOS 并发发**同一笔**交易，可在调用方序列化一次，再传 `&[u8]` 给各 client，减少重复 bincode 序列化。 +- 热点路径未见不必要的 `Mutex`/`RwLock` 竞争，当前设计可接受。 + +--- + +## 5. 代码质量 + +### 5.1 必须改：避免 panic 的 unwrap + +以下 PDA 或关键 `Option` 使用 `.unwrap()`，在异常输入下会直接 panic，建议改为 `Result` 并向上传播错误： + +| 文件 | 行号（约） | 说明 | +|------|------------|------| +| `src/instruction/pumpfun.rs` | 67, 101, 149, 221, 291, 295 | `get_bonding_curve_pda`、`get_user_volume_accumulator_pda`、`get_bonding_curve_v2_pda` | +| `src/instruction/pumpswap.rs` | 184, 198, 385, 407 | `get_user_volume_accumulator_pda`、`get_pool_v2_pda` | +| `src/instruction/utils/pumpfun.rs` | 229 | `DEFAULT_CREATOR_VAULT.unwrap()`（LazyLock 未初始化时可能 panic） | +| `src/instruction/bonk.rs` | 多处 | `get_pool_pda`、`get_vault_pda`、`params.rpc.as_ref().unwrap()` | +| `src/instruction/utils/bonk.rs` | 110–116, 148–152 | `checked_*` 链后 `.unwrap()`，数学假设不成立会 panic | +| `src/instruction/raydium_cpmm.rs` | 46, 106, 189, 250 | PDA / 状态相关 unwrap | +| `src/instruction/utils/raydium_cpmm.rs` | 83, 85, 141 | `get_vault_pda(...).unwrap()` | + +**建议**：统一改为 `.ok_or_else(|| anyhow!("..."))?` 或返回 `Result`，在调用链顶层处理错误，避免进程退出。 + +### 5.2 建议 + +- **测试**：为 instruction 构建（或至少 PDA + discriminator/data 布局）增加单元测试，固定输入与预期 bytes/accounts 比对，便于 IDL 升级时回归。 +- **错误类型**：`claim_cashback_*` 等返回 `Option`；可考虑统一为 `Result` 并带“无法构建”原因，或在文档中明确 None 的语义。 + +--- + +## 6. 安全性 + +### 6.1 必须改：API key 不得写入日志 + +| 位置 | 问题 | 建议 | +|------|------|------| +| `src/swqos/astralane_quic.rs` 约 61 行 | `info!(..., "api_key as CN: {}", api_key)` | 移除 api_key 或改为占位（如 `***` / 仅长度） | +| `src/swqos/astralane_quic.rs` 约 74 行 | `info!(..., "Connected at {} (api_key: {})", addr, api_key)` | 同上 | + +### 6.2 必须改：SkipServerVerification 风险 + +| 位置 | 问题 | 建议 | +|------|------|------| +| `src/swqos/astralane_quic.rs` 约 179–181 行 | `with_custom_certificate_verifier(SkipServerVerification)` 完全跳过服务端证书校验 | 1）若服务端提供证书：用 `RootCertStore` 或固定证书做校验；2）若仅 dev/内网：用 feature 或配置限制，并在文档/日志中明确“仅受控环境使用”；3）默认/生产构建建议不跳过校验 | + +### 6.3 建议 + +- **敏感配置**：确保生产环境从环境变量或安全配置读取 API key，并在文档中说明。 +- **依赖**：定期执行 `cargo audit` 与依赖升级。 +- **unsafe**：`perf/hardware_optimizations.rs`、`realtime_tuning.rs` 中的 `unsafe` 使用范围可控，需保持注释中的安全约定。 + +--- + +## 7. 汇总：必须改 vs 建议改 + +### 必须改（优先处理） + +| 序号 | 项 | 位置 | +|------|----|------| +| 1 | 移除 astralane_quic 中 API key 的日志输出 | `src/swqos/astralane_quic.rs` 61、74 行 | +| 2 | SkipServerVerification：改为证书校验或仅限 dev 并文档化 | `src/swqos/astralane_quic.rs` 179–181 行 | +| 3 | instruction 中 PDA 等 `.unwrap()` 改为 `Result` 并传播错误 | pumpfun.rs、pumpswap.rs、pumpfun/utils、bonk、raydium_cpmm 等 | +| 4 | PumpSwap 构建 instruction 时避免 `accounts.clone()`，改为移动 | `src/instruction/pumpswap.rs` 232、429 行 | + +### 建议改（可分批） + +| 序号 | 项 | 位置 | +|------|----|------| +| 5 | PDA 的 `program_id` 从 `&&AMM_PROGRAM` 改为 `&AMM_PROGRAM` | `src/instruction/utils/pumpswap.rs` 258、291 行 | +| 6 | PumpFun sell discriminator 改为命名常量 | `src/instruction/pumpfun.rs` 约 259 行 | +| 7 | `ensure_wsol_ata` 拆分；修正 “only support” 等文案 | `src/lib.rs` | +| 8 | 为 instruction 构建与 PDA 增加单元测试 | 新建 tests 或模块下 | +| 9 | 生产日志用 tracing 替代 println!/eprintln! | `src/swqos/astralane.rs` 等 | + +--- + +*报告基于当前仓库与 IDL 的静态阅读；若官方 SDK 或链上程序有未公开变更，建议再与官方实现或链上行为做一次对照验证。* diff --git a/examples/cli_trading/src/main.rs b/examples/cli_trading/src/main.rs index d608999..2690523 100644 --- a/examples/cli_trading/src/main.rs +++ b/examples/cli_trading/src/main.rs @@ -531,7 +531,7 @@ async fn handle_buy( let client = initialize_real_client().await?; - let (create_mint_ata, use_seed, owner_pubkey, amount_f64, decimals) = + let (create_mint_ata, use_seed, owner_pubkey, _amount_f64, _decimals) = check_mint_ata(&client, mint).await?; match dex { @@ -565,7 +565,7 @@ async fn handle_buy_rv4( slippage: Option, ) -> Result<(), Box> { let client = initialize_real_client().await?; - let (create_mint_ata, use_seed, owner_pubkey, amount_f64, decimals) = + let (create_mint_ata, use_seed, owner_pubkey, _amount_f64, _decimals) = check_mint_ata(&client, mint).await?; handle_buy_raydium_v4(mint, amm, sol_amount, slippage, create_mint_ata, use_seed, owner_pubkey) .await?; @@ -579,7 +579,7 @@ async fn handle_buy_rcpmm( slippage: Option, ) -> Result<(), Box> { let client = initialize_real_client().await?; - let (create_mint_ata, use_seed, owner_pubkey, amount_f64, decimals) = + let (create_mint_ata, use_seed, owner_pubkey, _amount_f64, _decimals) = check_mint_ata(&client, mint).await?; handle_buy_raydium_cpmm( mint, @@ -599,8 +599,8 @@ async fn handle_buy_pumpfun( sol_amount: f64, slippage: Option, create_mint_ata: bool, - use_seed: bool, - owner_pubkey: Pubkey, + _use_seed: bool, + _owner_pubkey: Pubkey, ) -> Result<(), Box> { println!("🔥 BUY PUMPFUN COMMAND"); println!(" Token Mint: {}", mint); @@ -655,7 +655,7 @@ async fn handle_buy_pumpswap( sol_amount: f64, slippage: Option, create_mint_ata: bool, - use_seed: bool, + _use_seed: bool, _owner_pubkey: Pubkey, ) -> Result<(), Box> { let client = initialize_real_client().await?; @@ -710,8 +710,8 @@ async fn handle_buy_bonk( sol_amount: f64, slippage: Option, create_mint_ata: bool, - use_seed: bool, - owner_pubkey: Pubkey, + _use_seed: bool, + _owner_pubkey: Pubkey, ) -> Result<(), Box> { let client = initialize_real_client().await?; println!("🔥 BUY BONK COMMAND"); @@ -766,8 +766,8 @@ async fn handle_buy_raydium_v4( sol_amount: f64, slippage: Option, create_mint_ata: bool, - use_seed: bool, - owner_pubkey: Pubkey, + _use_seed: bool, + _owner_pubkey: Pubkey, ) -> Result<(), Box> { let client = initialize_real_client().await?; println!("🔥 BUY RAYDIUM V4 COMMAND"); @@ -825,8 +825,8 @@ async fn handle_buy_raydium_cpmm( sol_amount: f64, slippage: Option, create_mint_ata: bool, - use_seed: bool, - owner_pubkey: Pubkey, + _use_seed: bool, + _owner_pubkey: Pubkey, ) -> Result<(), Box> { let client = initialize_real_client().await?; println!("🔥 BUY RAYDIUM CPMM COMMAND"); @@ -992,11 +992,11 @@ async fn handle_sell_pumpfun( mint: &str, token_amount: Option, slippage: Option, - create_mint_ata: bool, - use_seed: bool, - owner_pubkey: Pubkey, + _create_mint_ata: bool, + _use_seed: bool, + _owner_pubkey: Pubkey, amount_f64: f64, - decimals: u8, + _decimals: u8, ) -> Result<(), Box> { let amount = if token_amount.is_some() { token_amount.unwrap() } else { amount_f64 }; @@ -1053,11 +1053,11 @@ async fn handle_sell_pumpswap( mint: &str, token_amount: Option, slippage: Option, - create_mint_ata: bool, - use_seed: bool, - owner_pubkey: Pubkey, + _create_mint_ata: bool, + _use_seed: bool, + _owner_pubkey: Pubkey, amount_f64: f64, - decimals: u8, + _decimals: u8, ) -> Result<(), Box> { let amount = if token_amount.is_some() { token_amount.unwrap() } else { amount_f64 }; println!("🔥 SELL PUMPSWAP COMMAND"); @@ -1111,11 +1111,11 @@ async fn handle_sell_bonk( mint: &str, token_amount: Option, slippage: Option, - create_mint_ata: bool, - use_seed: bool, - owner_pubkey: Pubkey, + _create_mint_ata: bool, + _use_seed: bool, + _owner_pubkey: Pubkey, amount_f64: f64, - decimals: u8, + _decimals: u8, ) -> Result<(), Box> { let amount = if token_amount.is_some() { token_amount.unwrap() } else { amount_f64 }; println!("🔥 SELL PUMPSWAP COMMAND"); @@ -1170,11 +1170,11 @@ async fn handle_sell_raydium_v4( mint: &str, token_amount: Option, slippage: Option, - create_mint_ata: bool, - use_seed: bool, - owner_pubkey: Pubkey, + _create_mint_ata: bool, + _use_seed: bool, + _owner_pubkey: Pubkey, amount_f64: f64, - decimals: u8, + _decimals: u8, ) -> Result<(), Box> { let amount = if token_amount.is_some() { token_amount.unwrap() } else { amount_f64 }; println!("🔥 SELL RAYDIUM V4 COMMAND"); @@ -1231,11 +1231,11 @@ async fn handle_sell_raydium_cpmm( pool_address: &str, token_amount: Option, slippage: Option, - create_mint_ata: bool, - use_seed: bool, - owner_pubkey: Pubkey, + _create_mint_ata: bool, + _use_seed: bool, + _owner_pubkey: Pubkey, amount_f64: f64, - decimals: u8, + _decimals: u8, ) -> Result<(), Box> { let amount = if token_amount.is_some() { token_amount.unwrap() } else { amount_f64 }; println!("🔥 SELL RAYDIUM CPMM COMMAND"); diff --git a/examples/trading_client/src/main.rs b/examples/trading_client/src/main.rs index 3308d26..77e3e03 100644 --- a/examples/trading_client/src/main.rs +++ b/examples/trading_client/src/main.rs @@ -10,7 +10,7 @@ use sol_trade_sdk::{ common::{AnyResult, InfrastructureConfig, TradeConfig}, swqos::{SwqosConfig, SwqosRegion}, - TradingClient, TradingInfrastructure, + SwqosTransport, TradingClient, TradingInfrastructure, }; use solana_commitment_config::CommitmentConfig; use solana_sdk::signature::Keypair; @@ -48,7 +48,7 @@ async fn create_trading_client_simple() -> AnyResult { SwqosConfig::FlashBlock("your_api_token".to_string(), SwqosRegion::Frankfurt, None), SwqosConfig::Node1("your_api_token".to_string(), SwqosRegion::Frankfurt, None), SwqosConfig::BlockRazor("your_api_token".to_string(), SwqosRegion::Frankfurt, None), - SwqosConfig::Astralane("your_api_token".to_string(), SwqosRegion::Frankfurt, None), + SwqosConfig::Astralane("your_api_token".to_string(), SwqosRegion::Frankfurt, None, Some(SwqosTransport::Quic)), // QUIC; use None for HTTP // Helius Sender: 4th param swqos_only Some(true) => min tip 0.000005 SOL; None => 0.0002 SOL SwqosConfig::Helius("".to_string(), SwqosRegion::Default, None, Some(true)), ]; diff --git a/src/constants/swqos.rs b/src/constants/swqos.rs index c60b3bc..32ce4d8 100755 --- a/src/constants/swqos.rs +++ b/src/constants/swqos.rs @@ -267,6 +267,9 @@ pub const SWQOS_ENDPOINTS_ASTRALANE: [&str; 8] = [ "http://lim.gateway.astralane.io/irisb", ]; +/// Astralane QUIC 默认端点。 +pub const ASTRALANE_QUIC_ENDPOINT: &str = "lim.gateway.astralane.io:7000"; + pub const SWQOS_ENDPOINTS_STELLIUM: [&str; 8] = [ "http://ewr1.flashrpc.com", "http://fra1.flashrpc.com", diff --git a/src/instruction/pumpfun.rs b/src/instruction/pumpfun.rs index c654d2e..e14405f 100755 --- a/src/instruction/pumpfun.rs +++ b/src/instruction/pumpfun.rs @@ -10,7 +10,7 @@ use crate::{ instruction::utils::pumpfun::{ accounts, get_bonding_curve_pda, get_bonding_curve_v2_pda, get_creator, get_mayhem_fee_recipient_meta_random, get_user_volume_accumulator_pda, - global_constants::{self}, BUY_DISCRIMINATOR, BUY_EXACT_SOL_IN_DISCRIMINATOR, + global_constants::{self}, BUY_DISCRIMINATOR, BUY_EXACT_SOL_IN_DISCRIMINATOR, SELL_DISCRIMINATOR, }, utils::calc::{ common::{calculate_with_slippage_buy, calculate_with_slippage_sell}, @@ -64,7 +64,8 @@ impl InstructionBuilder for PumpFunInstructionBuilder { ); let bonding_curve_addr = if bonding_curve.account == Pubkey::default() { - get_bonding_curve_pda(¶ms.output_mint).unwrap() + get_bonding_curve_pda(¶ms.output_mint) + .ok_or_else(|| anyhow!("bonding_curve PDA derivation failed for mint {}", params.output_mint))? } else { bonding_curve.account }; @@ -97,8 +98,8 @@ impl InstructionBuilder for PumpFunInstructionBuilder { params.open_seed_optimize, ); - let user_volume_accumulator = - get_user_volume_accumulator_pda(¶ms.payer.pubkey()).unwrap(); + let user_volume_accumulator = get_user_volume_accumulator_pda(¶ms.payer.pubkey()) + .ok_or_else(|| anyhow!("user_volume_accumulator PDA derivation failed"))?; // ======================================== // Build instructions @@ -146,7 +147,8 @@ impl InstructionBuilder for PumpFunInstructionBuilder { global_constants::FEE_RECIPIENT_META }; - let bonding_curve_v2 = get_bonding_curve_v2_pda(¶ms.output_mint).unwrap(); + let bonding_curve_v2 = get_bonding_curve_v2_pda(¶ms.output_mint) + .ok_or_else(|| anyhow!("bonding_curve_v2 PDA derivation failed for mint {}", params.output_mint))?; let mut accounts: Vec = vec![ global_constants::GLOBAL_ACCOUNT_META, fee_recipient_meta, @@ -218,7 +220,8 @@ impl InstructionBuilder for PumpFunInstructionBuilder { }; let bonding_curve_addr = if bonding_curve.account == Pubkey::default() { - get_bonding_curve_pda(¶ms.input_mint).unwrap() + get_bonding_curve_pda(¶ms.input_mint) + .ok_or_else(|| anyhow!("bonding_curve PDA derivation failed for mint {}", params.input_mint))? } else { bonding_curve.account }; @@ -257,7 +260,7 @@ impl InstructionBuilder for PumpFunInstructionBuilder { let mut instructions = Vec::with_capacity(2); let mut sell_data = [0u8; 24]; - sell_data[..8].copy_from_slice(&[51, 230, 133, 164, 1, 127, 131, 173]); // Method ID + sell_data[..8].copy_from_slice(&SELL_DISCRIMINATOR); sell_data[8..16].copy_from_slice(&token_amount.to_le_bytes()); sell_data[16..24].copy_from_slice(&min_sol_output.to_le_bytes()); @@ -287,12 +290,13 @@ impl InstructionBuilder for PumpFunInstructionBuilder { // Cashback: Bonding Curve Sell expects UserVolumeAccumulator PDA at 0th remaining account (writable) if bonding_curve.is_cashback_coin { - let user_volume_accumulator = - get_user_volume_accumulator_pda(¶ms.payer.pubkey()).unwrap(); + let user_volume_accumulator = get_user_volume_accumulator_pda(¶ms.payer.pubkey()) + .ok_or_else(|| anyhow!("user_volume_accumulator PDA derivation failed"))?; accounts.push(AccountMeta::new(user_volume_accumulator, false)); } // remainingAccounts: @pump-fun/pump-sdk sell 要求末尾传 bondingCurveV2Pda(mint)（cashback 时在 user_volume_accumulator 之后），勿删 - let bonding_curve_v2 = get_bonding_curve_v2_pda(¶ms.input_mint).unwrap(); + let bonding_curve_v2 = get_bonding_curve_v2_pda(¶ms.input_mint) + .ok_or_else(|| anyhow!("bonding_curve_v2 PDA derivation failed for mint {}", params.input_mint))?; accounts.push(AccountMeta::new_readonly(bonding_curve_v2, false)); instructions.push(Instruction::new_with_bytes( diff --git a/src/instruction/pumpswap.rs b/src/instruction/pumpswap.rs index 40de9f4..45f5a87 100755 --- a/src/instruction/pumpswap.rs +++ b/src/instruction/pumpswap.rs @@ -180,10 +180,9 @@ impl InstructionBuilder for PumpSwapInstructionBuilder { ]); if quote_is_wsol_or_usdc { accounts.push(accounts::GLOBAL_VOLUME_ACCUMULATOR_META); - accounts.push(AccountMeta::new( - get_user_volume_accumulator_pda(¶ms.payer.pubkey()).unwrap(), - false, - )); + let uva = get_user_volume_accumulator_pda(¶ms.payer.pubkey()) + .ok_or_else(|| anyhow!("user_volume_accumulator PDA derivation failed"))?; + accounts.push(AccountMeta::new(uva, false)); } accounts.push(accounts::FEE_CONFIG_META); accounts.push(accounts::FEE_PROGRAM_META); @@ -194,10 +193,9 @@ impl InstructionBuilder for PumpSwapInstructionBuilder { } } // remainingAccounts: @pump-fun/pump-swap-sdk 要求末尾传 poolV2Pda(baseMint)，勿删 - accounts.push(AccountMeta::new_readonly( - get_pool_v2_pda(&base_mint).unwrap(), - false, - )); + let pool_v2 = get_pool_v2_pda(&base_mint) + .ok_or_else(|| anyhow!("pool_v2 PDA derivation failed for base_mint {}", base_mint))?; + accounts.push(AccountMeta::new_readonly(pool_v2, false)); // Create instruction data（buy/buy_exact_quote_in 第三参数 track_volume: OptionBool，仅代币支持返现时传 Some(true)；sell 仅两参数） let track_volume = if protocol_params.is_cashback_coin { [1u8, 1u8] } else { [1u8, 0u8] }; // Some(true) / Some(false) @@ -227,13 +225,11 @@ impl InstructionBuilder for PumpSwapInstructionBuilder { buf.to_vec() }; - let buy_instruction = Instruction { + instructions.push(Instruction { program_id: accounts::AMM_PROGRAM, - accounts: accounts.clone(), + accounts, data, - }; - - instructions.push(buy_instruction); + }); if close_wsol_ata { // Close wSOL ATA account, reclaim rent instructions.extend(crate::trading::common::close_wsol(¶ms.payer.pubkey())); @@ -381,10 +377,9 @@ impl InstructionBuilder for PumpSwapInstructionBuilder { ]); if !quote_is_wsol_or_usdc { accounts.push(accounts::GLOBAL_VOLUME_ACCUMULATOR_META); - accounts.push(AccountMeta::new( - get_user_volume_accumulator_pda(¶ms.payer.pubkey()).unwrap(), - false, - )); + let uva = get_user_volume_accumulator_pda(¶ms.payer.pubkey()) + .ok_or_else(|| anyhow!("user_volume_accumulator PDA derivation failed"))?; + accounts.push(AccountMeta::new(uva, false)); } accounts.push(accounts::FEE_CONFIG_META); accounts.push(accounts::FEE_PROGRAM_META); @@ -403,10 +398,9 @@ impl InstructionBuilder for PumpSwapInstructionBuilder { } } // remainingAccounts: @pump-fun/pump-swap-sdk sell 要求末尾传 poolV2Pda(baseMint)，勿删 - accounts.push(AccountMeta::new_readonly( - get_pool_v2_pda(&base_mint).unwrap(), - false, - )); + let pool_v2 = get_pool_v2_pda(&base_mint) + .ok_or_else(|| anyhow!("pool_v2 PDA derivation failed for base_mint {}", base_mint))?; + accounts.push(AccountMeta::new_readonly(pool_v2, false)); // Create instruction data let mut data = [0u8; 24]; @@ -424,13 +418,11 @@ impl InstructionBuilder for PumpSwapInstructionBuilder { data[16..24].copy_from_slice(&token_amount.to_le_bytes()); } - let sell_instruction = Instruction { + instructions.push(Instruction { program_id: accounts::AMM_PROGRAM, - accounts: accounts.clone(), + accounts, data: data.to_vec(), - }; - - instructions.push(sell_instruction); + }); if close_wsol_ata { instructions.extend(crate::trading::common::close_wsol(¶ms.payer.pubkey())); diff --git a/src/instruction/utils/pumpfun.rs b/src/instruction/utils/pumpfun.rs index ff4b753..b23e8b6 100644 --- a/src/instruction/utils/pumpfun.rs +++ b/src/instruction/utils/pumpfun.rs @@ -226,10 +226,9 @@ pub fn get_creator(creator_vault_pda: &Pubkey) -> Pubkey { // Fast check against cached default creator vault static DEFAULT_CREATOR_VAULT: std::sync::LazyLock> = std::sync::LazyLock::new(|| get_creator_vault_pda(&Pubkey::default())); - if creator_vault_pda.eq(&DEFAULT_CREATOR_VAULT.unwrap()) { - Pubkey::default() - } else { - *creator_vault_pda + match DEFAULT_CREATOR_VAULT.as_ref() { + Some(default) if creator_vault_pda.eq(default) => Pubkey::default(), + _ => *creator_vault_pda, } } } @@ -299,3 +298,32 @@ pub fn get_buy_price( s_u64.min(real_token_reserves) } + +#[cfg(test)] +mod tests { + use super::*; + use solana_sdk::pubkey::Pubkey; + + #[test] + fn pumpfun_discriminators_are_8_bytes() { + assert_eq!(BUY_DISCRIMINATOR.len(), 8); + assert_eq!(BUY_EXACT_SOL_IN_DISCRIMINATOR.len(), 8); + assert_eq!(SELL_DISCRIMINATOR.len(), 8); + } + + #[test] + fn pumpfun_bonding_curve_and_v2_pda_differ_for_same_mint() { + let mint = Pubkey::new_unique(); + let pda = get_bonding_curve_pda(&mint).unwrap(); + let pda_v2 = get_bonding_curve_v2_pda(&mint).unwrap(); + assert_ne!(pda, pda_v2, "bonding_curve and bonding_curve_v2 PDAs must differ"); + } + + #[test] + fn pumpfun_creator_vault_pda_deterministic() { + let creator = Pubkey::new_unique(); + let a = get_creator_vault_pda(&creator).unwrap(); + let b = get_creator_vault_pda(&creator).unwrap(); + assert_eq!(a, b); + } +} diff --git a/src/instruction/utils/pumpswap.rs b/src/instruction/utils/pumpswap.rs index 943e493..31072c8 100644 --- a/src/instruction/utils/pumpswap.rs +++ b/src/instruction/utils/pumpswap.rs @@ -255,7 +255,7 @@ pub fn get_user_volume_accumulator_pda(user: &Pubkey) -> Option { crate::common::fast_fn::PdaCacheKey::PumpSwapUserVolume(*user), || { let seeds: &[&[u8]; 2] = &[&seeds::USER_VOLUME_ACCUMULATOR_SEED, user.as_ref()]; - let program_id: &Pubkey = &&accounts::AMM_PROGRAM; + let program_id: &Pubkey = &accounts::AMM_PROGRAM; let pda: Option<(Pubkey, u8)> = Pubkey::try_find_program_address(seeds, program_id); pda.map(|pubkey| pubkey.0) }, @@ -288,7 +288,7 @@ pub fn get_user_volume_accumulator_quote_ata( pub fn get_global_volume_accumulator_pda() -> Option { let seeds: &[&[u8]; 1] = &[&seeds::GLOBAL_VOLUME_ACCUMULATOR_SEED]; - let program_id: &Pubkey = &&accounts::AMM_PROGRAM; + let program_id: &Pubkey = &accounts::AMM_PROGRAM; let pda: Option<(Pubkey, u8)> = Pubkey::try_find_program_address(seeds, program_id); pda.map(|pubkey| pubkey.0) } @@ -477,3 +477,31 @@ pub fn get_fee_config_pda() -> Option { let pda: Option<(Pubkey, u8)> = Pubkey::try_find_program_address(seeds, program_id); pda.map(|pubkey| pubkey.0) } + +#[cfg(test)] +mod tests { + use super::*; + use solana_sdk::pubkey::Pubkey; + + #[test] + fn pumpswap_user_volume_accumulator_pda_deterministic() { + let user = Pubkey::new_unique(); + let a = get_user_volume_accumulator_pda(&user).unwrap(); + let b = get_user_volume_accumulator_pda(&user).unwrap(); + assert_eq!(a, b); + } + + #[test] + fn pumpswap_global_volume_accumulator_matches_constant() { + let pda = get_global_volume_accumulator_pda().unwrap(); + assert_eq!(pda, accounts::GLOBAL_VOLUME_ACCUMULATOR); + } + + #[test] + fn pumpswap_pool_v2_pda_deterministic() { + let base_mint = Pubkey::new_unique(); + let a = get_pool_v2_pda(&base_mint).unwrap(); + let b = get_pool_v2_pda(&base_mint).unwrap(); + assert_eq!(a, b); + } +} diff --git a/src/lib.rs b/src/lib.rs index 547c280..69b05eb 100755 --- a/src/lib.rs +++ b/src/lib.rs @@ -19,6 +19,8 @@ use crate::swqos::common::TradeError; use crate::swqos::SwqosClient; use crate::swqos::SwqosConfig; use crate::swqos::TradeType; +// Re-export for SWQOS HTTP/QUIC choice in SwqosConfig (e.g. Astralane) +pub use crate::swqos::SwqosTransport; use crate::trading::core::params::BonkParams; use crate::trading::core::params::MeteoraDammV2Params; use crate::trading::core::params::PumpFunParams; @@ -397,8 +399,44 @@ impl TradingClient { } } + /// 单次尝试创建 WSOL ATA：获取 blockhash、组交易、发送并确认。成功或账户已存在返回 Ok(())，否则返回 Err(错误信息)。 + async fn try_create_wsol_ata_once( + rpc: &SolanaRpcClient, + payer: &Arc, + wsol_ata: &solana_sdk::pubkey::Pubkey, + create_ata_ixs: &[solana_sdk::instruction::Instruction], + timeout_secs: u64, + ) -> Result<(), String> { + use solana_sdk::transaction::Transaction; + let recent_blockhash = rpc.get_latest_blockhash().await + .map_err(|e| format!("Failed to get blockhash: {}", e))?; + let tx = Transaction::new_signed_with_payer( + create_ata_ixs, + Some(&payer.pubkey()), + &[payer.as_ref()], + recent_blockhash, + ); + let send_result = tokio::time::timeout( + tokio::time::Duration::from_secs(timeout_secs), + rpc.send_and_confirm_transaction(&tx), + ).await; + match send_result { + Ok(Ok(_signature)) => Ok(()), + Ok(Err(e)) => { + if rpc.get_account(wsol_ata).await.is_ok() { + return Ok(()); + } + Err(format!("{}", e)) + } + Err(_) => Err(format!("Transaction confirmation timeout ({}s)", timeout_secs)), + } + } + /// 确保钱包存在 WSOL ATA；不存在则发交易创建（会花费租金 + 手续费，初始化阶段唯一会扣钱的逻辑） async fn ensure_wsol_ata(payer: &Arc, rpc: &Arc) { + const MAX_RETRIES: usize = 3; + const TIMEOUT_SECS: u64 = 10; + let wsol_ata = crate::common::fast_fn::get_associated_token_address_with_program_id_fast( &payer.pubkey(), @@ -406,119 +444,62 @@ impl TradingClient { &crate::constants::TOKEN_PROGRAM, ); - match rpc.get_account(&wsol_ata).await { - Ok(_) => { - if sdk_log::sdk_log_enabled() { - info!(target: "sol_trade_sdk", "✅ WSOL ATA already exists: {}", wsol_ata); - } - return; + if rpc.get_account(&wsol_ata).await.is_ok() { + if sdk_log::sdk_log_enabled() { + info!(target: "sol_trade_sdk", "✅ WSOL ATA already exists: {}", wsol_ata); } - Err(_) => { + return; + } + + let create_ata_ixs = + crate::trading::common::wsol_manager::create_wsol_ata(&payer.pubkey()); + if create_ata_ixs.is_empty() { + if sdk_log::sdk_log_enabled() { + info!(target: "sol_trade_sdk", "ℹ️ WSOL ATA already exists (no need to create)"); + } + return; + } + + if sdk_log::sdk_log_enabled() { + info!(target: "sol_trade_sdk", "🔨 Creating WSOL ATA: {}", wsol_ata); + } + let mut last_error = None; + for attempt in 1..=MAX_RETRIES { + if attempt > 1 { if sdk_log::sdk_log_enabled() { - info!(target: "sol_trade_sdk", "🔨 Creating WSOL ATA: {}", wsol_ata); + info!(target: "sol_trade_sdk", "🔄 Retrying WSOL ATA creation (attempt {}/{})...", attempt, MAX_RETRIES); } - let create_ata_ixs = - crate::trading::common::wsol_manager::create_wsol_ata(&payer.pubkey()); - - if !create_ata_ixs.is_empty() { - use solana_sdk::transaction::Transaction; - - // 重试逻辑：最多尝试3次，每次超时10秒 - const MAX_RETRIES: usize = 3; - const TIMEOUT_SECS: u64 = 10; - let mut last_error = None; - - for attempt in 1..=MAX_RETRIES { - if attempt > 1 { - if sdk_log::sdk_log_enabled() { - info!(target: "sol_trade_sdk", "🔄 Retrying WSOL ATA creation (attempt {}/{})...", attempt, MAX_RETRIES); - } - tokio::time::sleep(tokio::time::Duration::from_secs(2)).await; - } - - let recent_blockhash = match rpc.get_latest_blockhash().await { - Ok(hash) => hash, - Err(e) => { - if sdk_log::sdk_log_enabled() { - warn!(target: "sol_trade_sdk", "⚠️ Failed to get latest blockhash: {}", e); - } - last_error = Some(format!("Failed to get blockhash: {}", e)); - continue; - } - }; - - let tx = Transaction::new_signed_with_payer( - &create_ata_ixs, - Some(&payer.pubkey()), - &[payer.as_ref()], - recent_blockhash, - ); - - // 使用超时包装 send_and_confirm_transaction - let send_result = tokio::time::timeout( - tokio::time::Duration::from_secs(TIMEOUT_SECS), - rpc.send_and_confirm_transaction(&tx) - ).await; - - match send_result { - Ok(Ok(signature)) => { - if sdk_log::sdk_log_enabled() { - info!(target: "sol_trade_sdk", "✅ WSOL ATA created successfully: {}", signature); - } - return; - } - Ok(Err(e)) => { - last_error = Some(format!("{}", e)); - - // 检查账户是否实际已存在 - if let Ok(_) = rpc.get_account(&wsol_ata).await { - if sdk_log::sdk_log_enabled() { - info!(target: "sol_trade_sdk", "✅ WSOL ATA already exists (tx failed but account exists): {}", wsol_ata); - } - return; - } - - if attempt < MAX_RETRIES && sdk_log::sdk_log_enabled() { - warn!(target: "sol_trade_sdk", "⚠️ Attempt {} failed: {}", attempt, e); - } - } - Err(_) => { - last_error = Some(format!("Transaction confirmation timeout ({}s)", TIMEOUT_SECS)); - if sdk_log::sdk_log_enabled() { - warn!(target: "sol_trade_sdk", "⚠️ Attempt {} timed out", attempt); - } - } - } - } - - // 所有重试都失败了 - if let Some(err) = last_error { - if sdk_log::sdk_log_enabled() { - error!(target: "sol_trade_sdk", "❌ WSOL ATA creation failed after {} retries: {}", MAX_RETRIES, wsol_ata); - error!(target: "sol_trade_sdk", " Error: {}", err); - error!(target: "sol_trade_sdk", " 💡 Possible causes:"); - error!(target: "sol_trade_sdk", " 1. Insufficient SOL balance (need ~0.002 SOL for rent exemption)"); - error!(target: "sol_trade_sdk", " 2. RPC timeout or network congestion"); - error!(target: "sol_trade_sdk", " 3. Insufficient transaction fee"); - error!(target: "sol_trade_sdk", " 🔧 Solutions:"); - error!(target: "sol_trade_sdk", " 1. Fund wallet with at least 0.1 SOL"); - error!(target: "sol_trade_sdk", " 2. Retry after a few seconds"); - error!(target: "sol_trade_sdk", " 3. Check RPC connection"); - error!(target: "sol_trade_sdk", " ⚠️ Process will exit in 5 seconds, restart after fixing the above"); - } - std::thread::sleep(std::time::Duration::from_secs(5)); - panic!( - "❌ WSOL ATA creation failed and account does not exist: {}. Error: {}", - wsol_ata, err - ); - } - } else { + tokio::time::sleep(tokio::time::Duration::from_secs(2)).await; + } + match Self::try_create_wsol_ata_once(rpc.as_ref(), payer, &wsol_ata, &create_ata_ixs, TIMEOUT_SECS).await { + Ok(()) => { if sdk_log::sdk_log_enabled() { - info!(target: "sol_trade_sdk", "ℹ️ WSOL ATA already exists (no need to create)"); + info!(target: "sol_trade_sdk", "✅ WSOL ATA created or already exists"); + } + return; + } + Err(e) => { + last_error = Some(e.clone()); + if attempt < MAX_RETRIES && sdk_log::sdk_log_enabled() { + warn!(target: "sol_trade_sdk", "⚠️ Attempt {} failed: {}", attempt, e); } } } } + + if let Some(err) = last_error { + if sdk_log::sdk_log_enabled() { + error!(target: "sol_trade_sdk", "❌ WSOL ATA creation failed after {} retries: {}", MAX_RETRIES, wsol_ata); + error!(target: "sol_trade_sdk", " Error: {}", err); + error!(target: "sol_trade_sdk", " 💡 Possible causes: insufficient SOL, RPC timeout, or fee"); + error!(target: "sol_trade_sdk", " 🔧 Solutions: fund wallet (e.g. 0.1 SOL), retry, check RPC"); + } + std::thread::sleep(std::time::Duration::from_secs(5)); + panic!( + "❌ WSOL ATA creation failed and account does not exist: {}. Error: {}", + wsol_ata, err + ); + } } /// Creates a new SolTradingSDK instance with the specified configuration @@ -672,7 +653,7 @@ impl TradingClient { } if params.input_token_type == TradeTokenType::USD1 && params.dex_type != DexType::Bonk { return Err(anyhow::anyhow!( - " Current version only support USD1 trading on Bonk protocols" + " Current version only supports USD1 trading on Bonk protocols" )); } let input_token_mint = if params.input_token_type == TradeTokenType::SOL { @@ -769,7 +750,7 @@ impl TradingClient { } if params.output_token_type == TradeTokenType::USD1 && params.dex_type != DexType::Bonk { return Err(anyhow::anyhow!( - " Current version only support USD1 trading on Bonk protocols" + " Current version only supports USD1 trading on Bonk protocols" )); } let executor = TradeFactory::create_executor(params.dex_type.clone()); diff --git a/src/swqos/astralane.rs b/src/swqos/astralane.rs index ce0f5ee..d25df37 100644 --- a/src/swqos/astralane.rs +++ b/src/swqos/astralane.rs @@ -2,6 +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 std::time::Duration; use anyhow::Result; @@ -19,24 +20,37 @@ use std::sync::atomic::{AtomicBool, Ordering}; /// Empty body for getHealth POST; avoid per-request allocation. static PING_BODY: &[u8] = &[]; +use crate::swqos::astralane_quic::AstralaneQuicClient; + +#[derive(Clone)] +pub enum AstralaneBackend { + Http { + endpoint: String, + auth_token: String, + http_client: Client, + ping_handle: Arc>>>, + stop_ping: Arc, + }, + Quic(Arc), +} + #[derive(Clone)] pub struct AstralaneClient { - pub endpoint: String, - pub auth_token: String, pub rpc_client: Arc, - pub http_client: Client, - pub ping_handle: Arc>>>, - pub stop_ping: Arc, + backend: AstralaneBackend, } #[async_trait::async_trait] impl SwqosClientTrait for AstralaneClient { async fn send_transaction(&self, trade_type: TradeType, 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(&self, trade_type: TradeType, transactions: &Vec, 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 { @@ -50,59 +64,71 @@ impl SwqosClientTrait for AstralaneClient { } impl AstralaneClient { + /// 使用 HTTP（irisb）提交。 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(); - - 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)), + 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: AstralaneBackend::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; }); - client } - /// Start periodic ping task to keep connections active + /// 使用 QUIC 提交。 + pub async fn new_quic(rpc_url: String, quic_endpoint: &str, api_key: String) -> Result { + let rpc_client = SolanaRpcClient::new(rpc_url); + let quic_client = AstralaneQuicClient::connect(quic_endpoint, &api_key).await?; + Ok(Self { + rpc_client: Arc::new(rpc_client), + backend: AstralaneBackend::Quic(Arc::new(quic_client)), + }) + } + 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 { - let mut interval = tokio::time::interval(Duration::from_secs(30)); - loop { - interval.tick().await; // first tick completes immediately → one ping at start - if stop_ping.load(Ordering::Relaxed) { - break; - } - if let Err(e) = Self::send_ping_request(&http_client, &endpoint, &auth_token).await { - eprintln!("Astralane ping request failed: {}", e); + match &self.backend { + AstralaneBackend::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 { + 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_ping_request(&http_client, &endpoint, &auth_token).await { + warn!(target: "sol_trade_sdk", "Astralane ping request failed: {}", e); + } } + }); + let mut guard = ping_handle.lock().await; + if let Some(old) = guard.as_ref() { + old.abort(); } - }); - - // 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(); + *guard = Some(handle); } - *ping_guard = Some(handle); + AstralaneBackend::Quic(_) => {} } } - /// Send ping request: POST endpoint?api-key=...&method=getHealth (endpoint is irisb from constants). + /// Send ping request: POST endpoint?api-key=...&method=getHealth async fn send_ping_request(http_client: &Client, endpoint: &str, auth_token: &str) -> Result<()> { let response = http_client .post(endpoint) @@ -112,57 +138,54 @@ impl AstralaneClient { .send() .await?; let status = response.status(); - let _ = response.bytes().await; // consume body so connection returns to pool + let _ = response.bytes().await; if !status.is_success() { - eprintln!("Astralane ping request returned non-success status: {}", status); + warn!(target: "sol_trade_sdk", "Astralane ping request returned non-success status: {}", status); } Ok(()) } - /// Send transaction via /irisb binary API (no Base64; lower latency). - pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction, wait_confirmation: bool) -> Result<()> { + async fn send_transaction_impl(&self, trade_type: TradeType, transaction: &VersionedTransaction, wait_confirmation: bool) -> Result<()> { let start_time = Instant::now(); let signature = transaction.get_signature(); - let body_bytes = bincode_serialize(transaction).map_err(|e| anyhow::anyhow!("Astralane binary serialize failed: {}", e))?; - let response = self.http_client - .post(&self.endpoint) - .query(&[("api-key", self.auth_token.as_str()), ("method", "sendTransaction")]) - .header("Content-Type", "application/octet-stream") - .body(body_bytes) - .send() - .await?; - - let status = response.status(); - let _ = response.bytes().await; - if status.is_success() { - println!(" [astralane] {} submitted: {:?}", trade_type, start_time.elapsed()); - } else { - eprintln!(" [astralane] {} submission failed: status {}", trade_type, status); - return Err(anyhow::anyhow!("Astralane sendTransaction failed: {}", status)); + match &self.backend { + AstralaneBackend::Http { endpoint, auth_token, http_client, .. } => { + let response = http_client + .post(endpoint) + .query(&[("api-key", auth_token.as_str()), ("method", "sendTransaction")]) + .header("Content-Type", "application/octet-stream") + .body(body_bytes) + .send() + .await?; + let status = response.status(); + let _ = response.bytes().await; + if status.is_success() { + info!(target: "sol_trade_sdk", "[astralane] {} submitted: {:?}", trade_type, start_time.elapsed()); + } else { + error!(target: "sol_trade_sdk", "[astralane] {} submission failed: status {}", trade_type, 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()); + } } let start_time = Instant::now(); match poll_transaction_confirmation(&self.rpc_client, *signature, wait_confirmation).await { Ok(_) => (), Err(e) => { - println!(" signature: {:?}", signature); - println!(" [astralane] {} confirmation failed: {:?}", trade_type, start_time.elapsed()); + info!(target: "sol_trade_sdk", "signature: {:?}", signature); + error!(target: "sol_trade_sdk", "[astralane] {} confirmation failed: {:?}", trade_type, start_time.elapsed()); return Err(e); - }, + } } if wait_confirmation { - println!(" signature: {:?}", signature); - println!(" [astralane] {} confirmed: {:?}", trade_type, start_time.elapsed()); - } - - Ok(()) - } - - pub async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec, wait_confirmation: bool) -> Result<()> { - for transaction in transactions { - self.send_transaction(trade_type, transaction, wait_confirmation).await?; + info!(target: "sol_trade_sdk", "signature: {:?}", signature); + info!(target: "sol_trade_sdk", "[astralane] {} confirmed: {:?}", trade_type, start_time.elapsed()); } Ok(()) } @@ -170,18 +193,19 @@ impl AstralaneClient { impl Drop for AstralaneClient { 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(); + match &self.backend { + AstralaneBackend::Http { stop_ping, ping_handle, .. } => { + stop_ping.store(true, Ordering::Relaxed); + let ping_handle = ping_handle.clone(); + tokio::spawn(async move { + let mut guard = ping_handle.lock().await; + if let Some(handle) = guard.as_ref() { + handle.abort(); + } + *guard = None; + }); } - *ping_guard = None; - }); + AstralaneBackend::Quic(_) => {} + } } } diff --git a/src/swqos/astralane_quic.rs b/src/swqos/astralane_quic.rs new file mode 100644 index 0000000..e331fda --- /dev/null +++ b/src/swqos/astralane_quic.rs @@ -0,0 +1,252 @@ +//! 内联自 [Astralane/astralane-quic-client](https://github.com/Astralane/astralane-quic-client)， +//! 用于向 Astralane QUIC TPU 提交交易，不依赖外部 crate，便于审计与安全可控。 + +use anyhow::{Context, Result}; +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::str::FromStr; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::Mutex; +use tracing::{info, warn}; + +/// ALPN protocol identifier for Astralane TPU. +const ALPN_ASTRALANE_TPU: &[u8] = b"astralane-tpu"; + +/// Maximum Solana transaction size. +pub const MAX_TRANSACTION_SIZE: usize = 1232; + +/// QUIC application error codes returned by the server. +pub mod error_code { + pub const OK: u32 = 0; + pub const UNKNOWN_API_KEY: u32 = 1; + pub const CONNECTION_LIMIT: u32 = 2; + + pub fn describe(code: u32) -> &'static str { + match code { + OK => "OK", + UNKNOWN_API_KEY => "Unknown API key", + CONNECTION_LIMIT => "Connection limit exceeded", + _ => "Unknown error", + } + } +} + +/// QUIC client for sending transactions to Astralane's TPU endpoint. +pub struct AstralaneQuicClient { + endpoint: Endpoint, + connection: Mutex, + server_addr: SocketAddr, + #[allow(dead_code)] + api_key: String, +} + +impl AstralaneQuicClient { + /// 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 { + 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)) + }).context("Invalid server address")?; + + 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")?; + 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")?; + + info!("[astralane-quic] Connected at {}", addr); + + Ok(Self { + endpoint, + connection: Mutex::new(connection), + server_addr: addr, + api_key: api_key.to_string(), + }) + } + + /// 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<()> { + if transaction_bytes.len() > MAX_TRANSACTION_SIZE { + anyhow::bail!( + "Transaction too large: {} bytes (max {})", + transaction_bytes.len(), + MAX_TRANSACTION_SIZE + ); + } + + let conn = { + let mut guard = self.connection.lock().await; + if let Some(reason) = guard.close_reason() { + if let quinn::ConnectionError::ApplicationClosed(ref info) = reason { + let code = info.error_code.into_inner(); + if code != error_code::OK as u64 { + anyhow::bail!( + "Server closed connection: {} (code {})", + error_code::describe(code as u32), + code + ); + } + } + 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); + } + guard.clone() + }; + + let mut send_stream = conn + .open_uni() + .await + .context("Failed to open unidirectional stream")?; + + send_stream + .write_all(transaction_bytes) + .await + .context("Failed to write transaction data")?; + + send_stream.finish().context("Failed to finish stream")?; + info!("[astralane-quic] Transaction sent ({} bytes)", transaction_bytes.len()); + + Ok(()) + } + + /// Reconnect to the server if the connection was closed. + pub async fn reconnect(&self) -> Result<()> { + 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); + } + Ok(()) + } + + /// Check if the connection is still alive. + pub async fn is_connected(&self) -> bool { + self.connection.lock().await.close_reason().is_none() + } + + /// Close the connection gracefully. + pub async fn close(&self) { + self.connection + .lock() + .await + .close(error_code::OK.into(), b"client closing"); + } + + fn build_client_config(api_key: &str) -> Result { + let key_pair = KeyPair::generate_for(&rcgen::PKCS_ECDSA_P256_SHA256)?; + let mut cert_params = CertificateParams::new(vec![])?; + cert_params.distinguished_name.push( + rcgen::DnType::CommonName, + rcgen::DnValue::Utf8String(api_key.to_string()), + ); + let cert = cert_params.self_signed(&key_pair)?; + + let cert_der = CertificateDer::from(cert.der().to_vec()); + let key_der = PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(key_pair.serialize_der())); + + let mut crypto = rustls::ClientConfig::builder() + .dangerous() + .with_custom_certificate_verifier(Arc::new(SkipServerVerification)) + .with_client_auth_cert(vec![cert_der], key_der) + .context("Failed to set client certificate")?; + + crypto.alpn_protocols = vec![ALPN_ASTRALANE_TPU.to_vec()]; + + let mut transport = TransportConfig::default(); + transport.max_idle_timeout(Some( + IdleTimeout::try_from(Duration::from_secs(30)).unwrap(), + )); + transport.keep_alive_interval(Some(Duration::from_secs(25))); + + let mut client_config = + ClientConfig::new(Arc::new(QuicClientConfig::try_from(crypto).unwrap())); + client_config.transport_config(Arc::new(transport)); + + Ok(client_config) + } +} + +impl Drop for AstralaneQuicClient { + fn drop(&mut self) { + self.connection + .get_mut() + .close(error_code::OK.into(), b"client closing"); + } +} + +/// Skip server certificate verification (Astralane server may use self-signed cert). +#[derive(Debug)] +struct SkipServerVerification; + +impl rustls::client::danger::ServerCertVerifier for SkipServerVerification { + fn verify_server_cert( + &self, + _end_entity: &CertificateDer<'_>, + _intermediates: &[CertificateDer<'_>], + _server_name: &rustls::pki_types::ServerName<'_>, + _ocsp_response: &[u8], + _now: rustls::pki_types::UnixTime, + ) -> Result { + Ok(rustls::client::danger::ServerCertVerified::assertion()) + } + + fn verify_tls12_signature( + &self, + _message: &[u8], + _cert: &CertificateDer<'_>, + _dss: &rustls::DigitallySignedStruct, + ) -> Result { + Ok(rustls::client::danger::HandshakeSignatureValid::assertion()) + } + + fn verify_tls13_signature( + &self, + _message: &[u8], + _cert: &CertificateDer<'_>, + _dss: &rustls::DigitallySignedStruct, + ) -> Result { + Ok(rustls::client::danger::HandshakeSignatureValid::assertion()) + } + + fn supported_verify_schemes(&self) -> Vec { + vec![ + rustls::SignatureScheme::ECDSA_NISTP256_SHA256, + rustls::SignatureScheme::ECDSA_NISTP384_SHA384, + rustls::SignatureScheme::RSA_PSS_SHA256, + rustls::SignatureScheme::RSA_PSS_SHA384, + rustls::SignatureScheme::RSA_PSS_SHA512, + rustls::SignatureScheme::RSA_PKCS1_SHA256, + rustls::SignatureScheme::RSA_PKCS1_SHA384, + rustls::SignatureScheme::RSA_PKCS1_SHA512, + rustls::SignatureScheme::ED25519, + ] + } +} diff --git a/src/swqos/mod.rs b/src/swqos/mod.rs index b47a44f..7853a9d 100755 --- a/src/swqos/mod.rs +++ b/src/swqos/mod.rs @@ -1,3 +1,4 @@ +pub mod astralane_quic; pub mod common; pub mod serialization; pub mod solana_rpc; @@ -86,6 +87,14 @@ pub const SWQOS_BLACKLIST: &[SwqosType] = &[ SwqosType::NextBlock, // NextBlock is disabled by default ]; +/// SWQOS 提交通道：HTTP 或 QUIC（低延迟）。部分提供商（如 Astralane）支持 QUIC。 +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)] +pub enum SwqosTransport { + #[default] + Http, + Quic, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum TradeType { Create, @@ -209,8 +218,8 @@ pub enum SwqosConfig { FlashBlock(String, SwqosRegion, Option), /// BlockRazor(api_token, region, custom_url) BlockRazor(String, SwqosRegion, Option), - /// Astralane(api_token, region, custom_url) - Astralane(String, SwqosRegion, Option), + /// Astralane(api_token, region, custom_url, transport). transport=None 表示 Http。 + Astralane(String, SwqosRegion, Option, Option), /// Stellium(api_token, region, custom_url) Stellium(String, SwqosRegion, Option), /// Lightspeed(api_key, region, custom_url) - Solana Vibe Station @@ -239,7 +248,7 @@ impl SwqosConfig { SwqosConfig::Node1(_, _, _) => SwqosType::Node1, SwqosConfig::FlashBlock(_, _, _) => SwqosType::FlashBlock, SwqosConfig::BlockRazor(_, _, _) => SwqosType::BlockRazor, - SwqosConfig::Astralane(_, _, _) => SwqosType::Astralane, + SwqosConfig::Astralane(_, _, _, _) => SwqosType::Astralane, SwqosConfig::Stellium(_, _, _) => SwqosType::Stellium, SwqosConfig::Lightspeed(_, _, _) => SwqosType::Lightspeed, SwqosConfig::Soyas(_, _, _) => SwqosType::Soyas, @@ -351,14 +360,22 @@ impl SwqosConfig { ); Ok(Arc::new(blockrazor_client)) }, - SwqosConfig::Astralane(auth_token, region, url) => { - let endpoint = SwqosConfig::get_endpoint(SwqosType::Astralane, region, url); - let astralane_client = AstralaneClient::new( - rpc_url.clone(), - endpoint.to_string(), - auth_token - ); - Ok(Arc::new(astralane_client)) + SwqosConfig::Astralane(auth_token, region, url, transport) => { + let use_quic = transport.map_or(false, |t| t == SwqosTransport::Quic); + if use_quic { + let quic_endpoint = crate::constants::swqos::ASTRALANE_QUIC_ENDPOINT; + let astralane_client = + AstralaneClient::new_quic(rpc_url.clone(), quic_endpoint, auth_token).await?; + Ok(Arc::new(astralane_client)) + } else { + let endpoint = SwqosConfig::get_endpoint(SwqosType::Astralane, region, url); + let astralane_client = AstralaneClient::new( + rpc_url.clone(), + endpoint.to_string(), + auth_token, + ); + Ok(Arc::new(astralane_client)) + } }, SwqosConfig::Stellium(auth_token, region, url) => { let endpoint = SwqosConfig::get_endpoint(SwqosType::Stellium, region, url);