diff --git a/src/constants/swqos.rs b/src/constants/swqos.rs index 32ce4d8..070d457 100755 --- a/src/constants/swqos.rs +++ b/src/constants/swqos.rs @@ -242,6 +242,7 @@ pub const SWQOS_ENDPOINTS_FLASHBLOCK: [&str; 8] = [ ]; /// BlockRazor Send Transaction v2: plain-text Base64 body, auth in URI, Content-Type: text/plain. Keep-alive: POST /v2/health. +/// 若 HTTP 返回 500，可尝试 HTTPS：https://.solana.blockrazor.io/v2/sendTransaction（Frankfurt/NewYork/Tokyo），通过 custom_url 覆盖。 pub const SWQOS_ENDPOINTS_BLOCKRAZOR: [&str; 8] = [ "http://newyork.solana.blockrazor.xyz:443/v2/sendTransaction", "http://frankfurt.solana.blockrazor.xyz:443/v2/sendTransaction", diff --git a/src/instruction/bonk.rs b/src/instruction/bonk.rs index f7655a7..9873654 100755 --- a/src/instruction/bonk.rs +++ b/src/instruction/bonk.rs @@ -4,12 +4,9 @@ use crate::{ accounts, get_pool_pda, get_vault_pda, BUY_EXECT_IN_DISCRIMINATOR, SELL_EXECT_IN_DISCRIMINATOR, }, - trading::{ - common::utils::get_token_balance, - core::{ - params::{BonkParams, SwapParams}, - traits::InstructionBuilder, - }, + trading::core::{ + params::{BonkParams, SwapParams}, + traits::InstructionBuilder, }, utils::calc::bonk::{ get_buy_token_amount_from_sol_amount, get_sell_sol_amount_from_token_amount, @@ -177,9 +174,10 @@ impl InstructionBuilder for BonkInstructionBuilder { // ======================================== // Parameter validation and basic data preparation // ======================================== - if params.rpc.is_none() { - return Err(anyhow!("RPC is not set")); - } + let amount = params + .input_amount + .filter(|&a| a > 0) + .ok_or_else(|| anyhow!("Bonk sell requires input_amount (token amount to sell); fetch balance via RPC before calling build_sell"))?; let protocol_params = params .protocol_params @@ -189,20 +187,6 @@ impl InstructionBuilder for BonkInstructionBuilder { let usd1_pool = protocol_params.global_config == accounts::USD1_GLOBAL_CONFIG; - let rpc = params.rpc.as_ref().unwrap().clone(); - - let mut amount = params.input_amount; - if params.input_amount.is_none() || params.input_amount.unwrap_or(0) == 0 { - let balance_u64 = - get_token_balance(rpc.as_ref(), ¶ms.payer.pubkey(), ¶ms.input_mint).await?; - amount = Some(balance_u64); - } - let amount = amount.unwrap_or(0); - - if amount == 0 { - return Err(anyhow!("Amount cannot be zero")); - } - let pool_state = if protocol_params.pool_state == Pubkey::default() { if usd1_pool { get_pool_pda(¶ms.input_mint, &crate::constants::USD1_TOKEN_ACCOUNT).unwrap() diff --git a/src/instruction/meteora_damm_v2.rs b/src/instruction/meteora_damm_v2.rs index fd8b54f..35d6017 100644 --- a/src/instruction/meteora_damm_v2.rs +++ b/src/instruction/meteora_damm_v2.rs @@ -27,7 +27,7 @@ impl InstructionBuilder for MeteoraDammV2InstructionBuilder { .protocol_params .as_any() .downcast_ref::() - .ok_or_else(|| anyhow!("Invalid protocol params for RaydiumCpmm"))?; + .ok_or_else(|| anyhow!("Invalid protocol params for MeteoraDammV2"))?; let is_wsol = protocol_params.token_a_mint == crate::constants::WSOL_TOKEN_ACCOUNT || protocol_params.token_b_mint == crate::constants::WSOL_TOKEN_ACCOUNT; let is_usdc = protocol_params.token_a_mint == crate::constants::USDC_TOKEN_ACCOUNT || protocol_params.token_b_mint == crate::constants::USDC_TOKEN_ACCOUNT; @@ -135,7 +135,7 @@ impl InstructionBuilder for MeteoraDammV2InstructionBuilder { .protocol_params .as_any() .downcast_ref::() - .ok_or_else(|| anyhow!("Invalid protocol params for RaydiumCpmm"))?; + .ok_or_else(|| anyhow!("Invalid protocol params for MeteoraDammV2"))?; if params.input_amount.is_none() || params.input_amount.unwrap_or(0) == 0 { return Err(anyhow!("Token amount is not set")); diff --git a/src/instruction/raydium_amm_v4.rs b/src/instruction/raydium_amm_v4.rs index 6a87503..822e7bf 100755 --- a/src/instruction/raydium_amm_v4.rs +++ b/src/instruction/raydium_amm_v4.rs @@ -29,7 +29,7 @@ impl InstructionBuilder for RaydiumAmmV4InstructionBuilder { .protocol_params .as_any() .downcast_ref::() - .ok_or_else(|| anyhow!("Invalid protocol params for RaydiumCpmm"))?; + .ok_or_else(|| anyhow!("Invalid protocol params for RaydiumAmmV4"))?; let is_wsol = protocol_params.coin_mint == crate::constants::WSOL_TOKEN_ACCOUNT || protocol_params.pc_mint == crate::constants::WSOL_TOKEN_ACCOUNT; @@ -144,7 +144,7 @@ impl InstructionBuilder for RaydiumAmmV4InstructionBuilder { .protocol_params .as_any() .downcast_ref::() - .ok_or_else(|| anyhow!("Invalid protocol params for RaydiumCpmm"))?; + .ok_or_else(|| anyhow!("Invalid protocol params for RaydiumAmmV4"))?; if params.input_amount.is_none() || params.input_amount.unwrap_or(0) == 0 { return Err(anyhow!("Token amount is not set")); diff --git a/src/instruction/utils/pumpswap.rs b/src/instruction/utils/pumpswap.rs index 31072c8..1e880c7 100644 --- a/src/instruction/utils/pumpswap.rs +++ b/src/instruction/utils/pumpswap.rs @@ -353,8 +353,8 @@ pub async fn find_by_base_mint( } pools.sort_by(|a, b| b.1.lp_supply.cmp(&a.1.lp_supply)); - let (address, pool) = pools[0].clone(); - Ok((address, pool)) + let first = &pools[0]; + Ok((first.0, first.1.clone())) } pub async fn find_by_quote_mint( @@ -405,8 +405,8 @@ pub async fn find_by_quote_mint( } pools.sort_by(|a, b| b.1.lp_supply.cmp(&a.1.lp_supply)); - let (address, pool) = pools[0].clone(); - Ok((address, pool)) + let first = &pools[0]; + Ok((first.0, first.1.clone())) } /// 按 mint 查找 PumpSwap 池（本函数仅用于 PumpSwap，其他 DEX 勿用）。 diff --git a/src/lib.rs b/src/lib.rs index 69b05eb..8ffd1aa 100755 --- a/src/lib.rs +++ b/src/lib.rs @@ -643,6 +643,11 @@ impl TradingClient { &self, params: TradeBuyParams, ) -> Result<(bool, Vec, Option), anyhow::Error> { + if params.recent_blockhash.is_none() && params.durable_nonce.is_none() { + return Err(anyhow::anyhow!( + "Must provide either recent_blockhash or durable_nonce for buy (required for transaction validity)" + )); + } #[cfg(feature = "perf-trace")] if sdk_log::sdk_log_enabled() && params.slippage_basis_points.is_none() { debug!( @@ -656,6 +661,13 @@ impl TradingClient { " Current version only supports USD1 trading on Bonk protocols" )); } + let protocol_params = params.extension_params; + if !validate_protocol_params(params.dex_type, &protocol_params) { + return Err(anyhow::anyhow!( + "Invalid protocol params for Trade (dex={:?})", + params.dex_type + )); + } let input_token_mint = if params.input_token_type == TradeTokenType::SOL { SOL_TOKEN_ACCOUNT } else if params.input_token_type == TradeTokenType::WSOL { @@ -665,8 +677,7 @@ impl TradingClient { } else { USD1_TOKEN_ACCOUNT }; - let executor = TradeFactory::create_executor(params.dex_type.clone()); - let protocol_params = params.extension_params; + let executor = TradeFactory::create_executor(params.dex_type); let buy_params = SwapParams { rpc: Some(self.infrastructure.rpc.clone()), payer: self.payer.clone(), @@ -680,7 +691,7 @@ impl TradingClient { address_lookup_table_account: params.address_lookup_table_account, recent_blockhash: params.recent_blockhash, wait_transaction_confirmed: params.wait_transaction_confirmed, - protocol_params: protocol_params.clone(), + protocol_params, open_seed_optimize: self.use_seed_optimize, // 使用全局seed优化配置 swqos_clients: self.infrastructure.swqos_clients.clone(), middleware_manager: self.middleware_manager.clone(), @@ -700,10 +711,6 @@ impl TradingClient { use_exact_sol_amount: params.use_exact_sol_amount, }; - if !validate_protocol_params(params.dex_type, &protocol_params) { - return Err(anyhow::anyhow!("Invalid protocol params for Trade")); - } - let swap_result = executor.swap(buy_params).await; let result = swap_result.map(|(success, sigs, err)| (success, sigs, err.map(TradeError::from))); @@ -748,13 +755,24 @@ impl TradingClient { DEFAULT_SLIPPAGE ); } + if params.recent_blockhash.is_none() && params.durable_nonce.is_none() { + return Err(anyhow::anyhow!( + "Must provide either recent_blockhash or durable_nonce for sell (required for transaction validity)" + )); + } if params.output_token_type == TradeTokenType::USD1 && params.dex_type != DexType::Bonk { return Err(anyhow::anyhow!( " Current version only supports USD1 trading on Bonk protocols" )); } - let executor = TradeFactory::create_executor(params.dex_type.clone()); let protocol_params = params.extension_params; + if !validate_protocol_params(params.dex_type, &protocol_params) { + return Err(anyhow::anyhow!( + "Invalid protocol params for Trade (dex={:?})", + params.dex_type + )); + } + let executor = TradeFactory::create_executor(params.dex_type); let output_token_mint = if params.output_token_type == TradeTokenType::SOL { SOL_TOKEN_ACCOUNT } else if params.output_token_type == TradeTokenType::WSOL { @@ -777,7 +795,7 @@ impl TradingClient { address_lookup_table_account: params.address_lookup_table_account, recent_blockhash: params.recent_blockhash, wait_transaction_confirmed: params.wait_transaction_confirmed, - protocol_params: protocol_params.clone(), + protocol_params, with_tip: params.with_tip, open_seed_optimize: self.use_seed_optimize, // 使用全局seed优化配置 swqos_clients: self.infrastructure.swqos_clients.clone(), @@ -797,10 +815,6 @@ impl TradingClient { use_exact_sol_amount: None, }; - if !validate_protocol_params(params.dex_type, &protocol_params) { - return Err(anyhow::anyhow!("Invalid protocol params for Trade")); - } - let swap_result = executor.swap(sell_params).await; let result = swap_result.map(|(success, sigs, err)| (success, sigs, err.map(TradeError::from))); diff --git a/src/swqos/blockrazor.rs b/src/swqos/blockrazor.rs index ed9c52c..6286351 100644 --- a/src/swqos/blockrazor.rs +++ b/src/swqos/blockrazor.rs @@ -49,7 +49,11 @@ impl SwqosClientTrait for BlockRazorClient { impl BlockRazorClient { 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(); + // 官方文档：請求中唯一允許的 header 是 Content-Type: text/plain；避免默认 User-Agent 等导致 500 + let http_client = default_http_client_builder() + .user_agent("") + .build() + .unwrap(); let client = Self { rpc_client: Arc::new(rpc_client), @@ -127,6 +131,7 @@ impl BlockRazorClient { } /// 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(&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)?; @@ -140,16 +145,21 @@ impl BlockRazorClient { .await?; let status = response.status(); - let _ = response.bytes().await; if status.is_success() { + let _ = response.bytes().await; if crate::common::sdk_log::sdk_log_enabled() { println!(" [blockrazor] {} submitted: {:?}", trade_type, start_time.elapsed()); } } else { + let body = response.text().await.unwrap_or_default(); if crate::common::sdk_log::sdk_log_enabled() { - eprintln!(" [blockrazor] {} submission failed: status {}", trade_type, status); + eprintln!(" [blockrazor] {} submission failed: status {} body: {}", trade_type, status, body); } - return Err(anyhow::anyhow!("BlockRazor sendTransaction failed: {}", status)); + return Err(anyhow::anyhow!( + "BlockRazor sendTransaction failed: status {} body: {}", + status, + body + )); } let start_time = Instant::now(); diff --git a/src/swqos/mod.rs b/src/swqos/mod.rs index 7853a9d..c5f881b 100755 --- a/src/swqos/mod.rs +++ b/src/swqos/mod.rs @@ -77,6 +77,8 @@ use crate::{ }; lazy_static::lazy_static! { + /// Reserved for future per-SWQOS tip account caching (currently unused). + #[allow(dead_code)] static ref TIP_ACCOUNT_CACHE: RwLock> = RwLock::new(Vec::new()); } diff --git a/src/trading/common/compute_budget_manager.rs b/src/trading/common/compute_budget_manager.rs index cccae75..dff33b3 100755 --- a/src/trading/common/compute_budget_manager.rs +++ b/src/trading/common/compute_budget_manager.rs @@ -3,6 +3,7 @@ use once_cell::sync::Lazy; use smallvec::SmallVec; use solana_sdk::instruction::Instruction; use solana_compute_budget_interface::ComputeBudgetInstruction; +use std::sync::Arc; /// Cache key containing all parameters for compute budget instructions #[derive(Debug, Clone, PartialEq, Eq, Hash)] @@ -11,43 +12,52 @@ struct ComputeBudgetCacheKey { unit_limit: u32, } -/// Global cache storing compute budget instructions -/// Uses DashMap for high-performance lock-free concurrent access -static COMPUTE_BUDGET_CACHE: Lazy>> = +/// Global cache storing compute budget instructions (Arc to avoid clone on hit). +/// Uses DashMap for high-performance lock-free concurrent access. +static COMPUTE_BUDGET_CACHE: Lazy>>> = Lazy::new(|| DashMap::new()); +/// Extend `instructions` with compute budget instructions; on cache hit extends from cached Arc (no SmallVec clone). #[inline(always)] -pub fn compute_budget_instructions( +pub fn extend_compute_budget_instructions( + instructions: &mut Vec, unit_price: u64, unit_limit: u32, -) -> SmallVec<[Instruction; 2]> { - // Create cache key - let cache_key = ComputeBudgetCacheKey { - unit_price: unit_price, - unit_limit: unit_limit, - }; +) { + let cache_key = ComputeBudgetCacheKey { unit_price, unit_limit }; - // Try to get from cache first - if let Some(cached_insts) = COMPUTE_BUDGET_CACHE.get(&cache_key) { - return cached_insts.clone(); + if let Some(cached) = COMPUTE_BUDGET_CACHE.get(&cache_key) { + instructions.extend(cached.iter().cloned()); + return; } - // Cache miss, generate new instructions let mut insts = SmallVec::<[Instruction; 2]>::new(); - - // Only add compute unit price instruction if > 0 if unit_price > 0 { insts.push(ComputeBudgetInstruction::set_compute_unit_price(unit_price)); } - - // Only add compute unit limit instruction if > 0 if unit_limit > 0 { insts.push(ComputeBudgetInstruction::set_compute_unit_limit(unit_limit)); } + let arc = Arc::new(insts); + instructions.extend(arc.iter().cloned()); + COMPUTE_BUDGET_CACHE.insert(cache_key, arc); +} - // Store result in cache - let insts_clone = insts.clone(); - COMPUTE_BUDGET_CACHE.insert(cache_key, insts_clone); - +/// Returns compute budget instructions (allocates on cache hit; prefer `extend_compute_budget_instructions` on hot path). +#[inline(always)] +pub fn compute_budget_instructions(unit_price: u64, unit_limit: u32) -> SmallVec<[Instruction; 2]> { + let cache_key = ComputeBudgetCacheKey { unit_price, unit_limit }; + if let Some(cached) = COMPUTE_BUDGET_CACHE.get(&cache_key) { + return (**cached).clone(); + } + let mut insts = SmallVec::<[Instruction; 2]>::new(); + if unit_price > 0 { + insts.push(ComputeBudgetInstruction::set_compute_unit_price(unit_price)); + } + if unit_limit > 0 { + insts.push(ComputeBudgetInstruction::set_compute_unit_limit(unit_limit)); + } + let arc = Arc::new(insts.clone()); + COMPUTE_BUDGET_CACHE.insert(cache_key, arc); insts } diff --git a/src/trading/common/nonce_manager.rs b/src/trading/common/nonce_manager.rs index 299150f..6fa32f0 100755 --- a/src/trading/common/nonce_manager.rs +++ b/src/trading/common/nonce_manager.rs @@ -12,9 +12,7 @@ use crate::common::nonce_cache::DurableNonceInfo; pub fn add_nonce_instruction( instructions: &mut Vec, payer: &Keypair, - // nonce_account: Option, - // current_nonce: Option, - durable_nonce: Option, + durable_nonce: Option<&DurableNonceInfo>, ) -> Result<(), anyhow::Error> { if let Some(durable_nonce) = durable_nonce { let nonce_advance_ix = advance_nonce_account(&durable_nonce.nonce_account.unwrap(), &payer.pubkey()); @@ -24,17 +22,22 @@ pub fn add_nonce_instruction( Ok(()) } -/// Get blockhash for transaction -/// If nonce account is used, return blockhash from nonce, otherwise return the provided recent_blockhash +/// Get blockhash for transaction. +/// If nonce account is used, returns blockhash from nonce; otherwise returns the provided recent_blockhash. +/// Returns error when neither durable_nonce nor recent_blockhash is set (caller must provide one for low latency). pub fn get_transaction_blockhash( recent_blockhash: Option, - durable_nonce: Option, - // nonce_account: Option, - // current_nonce: Option, -) -> Hash { + durable_nonce: Option<&DurableNonceInfo>, +) -> Result { if let Some(durable_nonce) = durable_nonce { - durable_nonce.current_nonce.unwrap() + durable_nonce + .current_nonce + .ok_or_else(|| anyhow::anyhow!("durable_nonce.current_nonce is None")) + } else if let Some(hash) = recent_blockhash { + Ok(hash) } else { - recent_blockhash.unwrap() + Err(anyhow::anyhow!( + "Must provide either recent_blockhash or durable_nonce for transaction" + )) } } diff --git a/src/trading/common/transaction_builder.rs b/src/trading/common/transaction_builder.rs index 56cf8e3..ce11c66 100755 --- a/src/trading/common/transaction_builder.rs +++ b/src/trading/common/transaction_builder.rs @@ -1,70 +1,66 @@ use solana_hash::Hash; use solana_sdk::{ - instruction::Instruction, message::AddressLookupTableAccount, native_token::sol_str_to_lamports, pubkey::Pubkey, signature::Keypair, signer::Signer, transaction::VersionedTransaction + instruction::Instruction, message::AddressLookupTableAccount, pubkey::Pubkey, + signature::Keypair, signer::Signer, transaction::VersionedTransaction, }; use solana_system_interface::instruction::transfer; use std::sync::Arc; -use super::{ - compute_budget_manager::compute_budget_instructions, - nonce_manager::{add_nonce_instruction, get_transaction_blockhash}, -}; +use super::nonce_manager::{add_nonce_instruction, get_transaction_blockhash}; use crate::{ common::{nonce_cache::DurableNonceInfo, SolanaRpcClient}, trading::{MiddlewareManager, core::transaction_pool::{acquire_builder, release_builder}}, }; +/// Convert SOL amount (f64) to lamports without string allocation (hot path). +#[inline(always)] +fn sol_f64_to_lamports(sol: f64) -> u64 { + if sol <= 0.0 { + return 0; + } + let lamports = sol * 1_000_000_000.0; + (lamports.min(u64::MAX as f64)).round() as u64 +} + /// Build standard RPC transaction. -/// Takes `business_instructions` by reference to avoid per-task Vec clone in execute_parallel. +/// Takes Arc/context by reference to avoid clone in worker hot path (Arc::clone is cheap but ref is zero-cost). pub async fn build_transaction( - payer: Arc, - _rpc: Option>, + payer: &Arc, + _rpc: Option<&Arc>, unit_limit: u32, unit_price: u64, business_instructions: &[Instruction], - address_lookup_table_account: Option, + address_lookup_table_account: Option<&AddressLookupTableAccount>, recent_blockhash: Option, - middleware_manager: Option>, + middleware_manager: Option<&Arc>, protocol_name: &str, is_buy: bool, with_tip: bool, tip_account: &Pubkey, tip_amount: f64, - durable_nonce: Option, - // nonce_account: Option, - // current_nonce: Option, + durable_nonce: Option<&DurableNonceInfo>, ) -> Result { let mut instructions = Vec::with_capacity(business_instructions.len() + 5); - // Add nonce instruction - if let Err(e) = - add_nonce_instruction(&mut instructions, payer.as_ref(), durable_nonce.clone()) - { + if let Err(e) = add_nonce_instruction(&mut instructions, payer.as_ref(), durable_nonce) { return Err(e); } - // Add tip transfer instruction if with_tip && tip_amount > 0.0 { - instructions.push(transfer( - &payer.pubkey(), - tip_account, - sol_str_to_lamports(tip_amount.to_string().as_str()).unwrap_or(0), - )); + let tip_lamports = sol_f64_to_lamports(tip_amount); + instructions.push(transfer(&payer.pubkey(), tip_account, tip_lamports)); } - // Add compute budget instructions - instructions.extend(compute_budget_instructions( + super::compute_budget_manager::extend_compute_budget_instructions( + &mut instructions, unit_price, unit_limit, - )); + ); - // Add business instructions (clone only here; avoids per-task Vec clone in execute_parallel) instructions.extend_from_slice(business_instructions); - // Get blockhash for transaction - let blockhash = get_transaction_blockhash(recent_blockhash, durable_nonce.clone()); + let blockhash = get_transaction_blockhash(recent_blockhash, durable_nonce)?; - // Build transaction build_versioned_transaction( payer, instructions, @@ -77,23 +73,18 @@ pub async fn build_transaction( .await } -/// Low-level function for building versioned transactions async fn build_versioned_transaction( - payer: Arc, + payer: &Arc, instructions: Vec, - address_lookup_table_account: Option, + address_lookup_table_account: Option<&AddressLookupTableAccount>, blockhash: Hash, - middleware_manager: Option>, + middleware_manager: Option<&Arc>, protocol_name: &str, is_buy: bool, ) -> Result { let full_instructions = match middleware_manager { Some(middleware_manager) => middleware_manager - .apply_middlewares_process_full_instructions( - instructions, - protocol_name.to_string(), - is_buy, - )?, + .apply_middlewares_process_full_instructions(instructions, protocol_name, is_buy)?, None => instructions, }; @@ -108,7 +99,7 @@ async fn build_versioned_transaction( ); let msg_bytes = versioned_msg.serialize(); - let signature = payer.try_sign_message(&msg_bytes).expect("sign failed"); + let signature = payer.as_ref().try_sign_message(&msg_bytes).expect("sign failed"); let tx = VersionedTransaction { signatures: vec![signature], message: versioned_msg }; // 归还构建器到池 diff --git a/src/trading/core/async_executor.rs b/src/trading/core/async_executor.rs index 149eaa7..b55ba2a 100644 --- a/src/trading/core/async_executor.rs +++ b/src/trading/core/async_executor.rs @@ -1,15 +1,26 @@ //! Parallel executor for multi-SWQOS submit. +//! +//! - **Pool**: Pre-spawned workers; hot path only enqueues jobs (no per-call tokio::spawn). +//! - **Arc**: Shared data is behind `Arc` so "clone" is just a refcount increment (no data copy). +//! - **Refs**: `build_transaction` takes `&Arc<..>`, `Option<&DurableNonceInfo>`, `Option<&AddressLookupTableAccount>` so the worker passes refs only (zero clone on worker path). use anyhow::{anyhow, Result}; use crossbeam_queue::ArrayQueue; +use once_cell::sync::OnceCell; use solana_hash::Hash; use solana_sdk::message::AddressLookupTableAccount; use solana_sdk::{ instruction::Instruction, pubkey::Pubkey, signature::Keypair, signature::Signature, }; +use std::collections::HashMap; +use std::hash::BuildHasherDefault; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::{str::FromStr, sync::Arc, time::Instant}; +use fnv::FnvHasher; + +type FnvHashMap = HashMap>; + use crate::{ common::nonce_cache::DurableNonceInfo, common::{GasFeeStrategy, SolanaRpcClient}, @@ -17,6 +28,133 @@ use crate::{ trading::{common::build_transaction, MiddlewareManager}, }; +const SWQOS_POOL_WORKERS: usize = 32; +const SWQOS_QUEUE_CAP: usize = 128; + +/// Shared across all jobs in one batch; built once, cloned as single Arc per job (minimal hot-path clone). +struct SwqosSharedContext { + payer: Arc, + instructions: Arc>, + rpc: Option>, + address_lookup_table_account: Option, + recent_blockhash: Option, + durable_nonce: Option, + middleware_manager: Option>, + protocol_name: &'static str, + is_buy: bool, + wait_transaction_confirmed: bool, + with_tip: bool, + collector: Arc, +} + +/// One SWQOS submit task; only per-task data + one Arc to shared (reduces hot-path clones). +struct SwqosJob { + shared: Arc, + tip: f64, + unit_limit: u32, + unit_price: u64, + tip_account: Arc, + swqos_client: Arc, + swqos_type: SwqosType, + core_id: Option, + use_affinity: bool, +} + +async fn run_one_swqos_job(job: SwqosJob) { + let s = &job.shared; + if job.use_affinity { + if let Some(cid) = job.core_id { + core_affinity::set_for_current(cid); + } + } + + let tip_amount = if s.with_tip { job.tip } else { 0.0 }; + + let transaction = match build_transaction( + &s.payer, + s.rpc.as_ref(), + job.unit_limit, + job.unit_price, + s.instructions.as_ref(), + s.address_lookup_table_account.as_ref(), + s.recent_blockhash, + s.middleware_manager.as_ref(), + s.protocol_name, + s.is_buy, + job.swqos_type != SwqosType::Default, + &job.tip_account, + tip_amount, + s.durable_nonce.as_ref(), + ) + .await + { + Ok(tx) => tx, + Err(e) => { + s.collector.submit(TaskResult { + success: false, + signature: Signature::default(), + error: Some(e), + swqos_type: job.swqos_type, + landed_on_chain: false, + submit_done_us: crate::common::clock::now_micros(), + }); + return; + } + }; + + let (success, err, landed_on_chain) = match job + .swqos_client + .send_transaction( + if s.is_buy { + TradeType::Buy + } else { + TradeType::Sell + }, + &transaction, + s.wait_transaction_confirmed, + ) + .await + { + Ok(()) => (true, None, true), + Err(e) => { + let landed = is_landed_error(&e); + (false, Some(e), landed) + } + }; + + let sig = transaction.signatures.first().copied().unwrap_or_default(); + s.collector.submit(TaskResult { + success, + signature: sig, + error: err, + swqos_type: job.swqos_type, + landed_on_chain, + submit_done_us: crate::common::clock::now_micros(), + }); +} + +async fn swqos_worker_loop(queue: Arc>) { + loop { + if let Some(job) = queue.pop() { + run_one_swqos_job(job).await; + } else { + tokio::task::yield_now().await; + } + } +} + +static SWQOS_QUEUE: OnceCell>> = OnceCell::new(); +static SWQOS_WORKERS_STARTED: AtomicBool = AtomicBool::new(false); + +fn ensure_swqos_pool(queue: Arc>) { + if SWQOS_WORKERS_STARTED.swap(true, Ordering::AcqRel) { + return; + } + for _ in 0..SWQOS_POOL_WORKERS { + tokio::spawn(swqos_worker_loop(queue.clone())); + } +} + #[repr(align(64))] struct TaskResult { success: bool, @@ -282,116 +420,64 @@ pub async fn execute_parallel( return Err(anyhow!("Multiple swqos transactions require durable_nonce to be set.",)); } - // Task preparation completed - + // Task preparation completed: one shared context (clone once per batch), then minimal per-task data. let collector = Arc::new(ResultCollector::new(task_configs.len())); - let _spawn_start = Instant::now(); + let shared = Arc::new(SwqosSharedContext { + payer, + instructions, + rpc, + address_lookup_table_account, + recent_blockhash, + durable_nonce, + middleware_manager, + protocol_name, + is_buy, + wait_transaction_confirmed, + with_tip, + collector: collector.clone(), + }); - for (i, swqos_client, gas_fee_strategy_config) in task_configs { - let core_id = cores.get(i % cores.len().max(1)).copied(); - let use_affinity = use_core_affinity; - let payer = payer.clone(); - let instructions = instructions.clone(); - let middleware_manager = middleware_manager.clone(); - let swqos_type = swqos_client.get_swqos_type(); - let tip_account_str = swqos_client.get_tip_account()?; - let tip_account = Arc::new(Pubkey::from_str(&tip_account_str).unwrap_or_default()); - let collector = collector.clone(); + let queue = SWQOS_QUEUE.get_or_init(|| Arc::new(ArrayQueue::new(SWQOS_QUEUE_CAP))); + ensure_swqos_pool(queue.clone()); - let tip = gas_fee_strategy_config.2.tip; - let unit_limit = gas_fee_strategy_config.2.cu_limit; - let unit_price = gas_fee_strategy_config.2.cu_price; - let rpc = rpc.clone(); - let durable_nonce = durable_nonce.clone(); - let address_lookup_table_account = address_lookup_table_account.clone(); - let recent_blockhash_task = recent_blockhash.clone(); - - tokio::spawn(async move { - let _task_start = Instant::now(); - if use_affinity { - if let Some(cid) = core_id { - core_affinity::set_for_current(cid); + { + // Cache tip_account per client (one get_tip_account/from_str per unique client per batch). Dropped before await so future stays Send. + let mut tip_cache: FnvHashMap<*const (), Arc> = + FnvHashMap::with_capacity_and_hasher(task_configs.len(), BuildHasherDefault::default()); + for (i, swqos_client, gas_fee_strategy_config) in task_configs { + let core_id = cores.get(i % cores.len().max(1)).copied(); + let swqos_type = swqos_client.get_swqos_type(); + let key = Arc::as_ptr(&swqos_client) as *const (); + let tip_account = match tip_cache.get(&key) { + Some(tip) => tip.clone(), + None => { + let s = swqos_client.get_tip_account()?; + let tip = Arc::new(Pubkey::from_str(&s).unwrap_or_default()); + tip_cache.insert(key, tip.clone()); + tip } - } - - let tip_amount = if with_tip { tip } else { 0.0 }; - - let _build_start = Instant::now(); - let transaction = match build_transaction( - payer, - rpc, + }; + let (tip, unit_limit, unit_price) = ( + gas_fee_strategy_config.2.tip, + gas_fee_strategy_config.2.cu_limit, + gas_fee_strategy_config.2.cu_price, + ); + let job = SwqosJob { + shared: shared.clone(), + tip, unit_limit, unit_price, - instructions.as_ref(), - address_lookup_table_account, - recent_blockhash_task, - middleware_manager, - protocol_name, - is_buy, - swqos_type != SwqosType::Default, - &tip_account, - tip_amount, - durable_nonce, - ) - .await - { - Ok(tx) => tx, - Err(e) => { - // Build transaction failed - collector.submit(TaskResult { - success: false, - signature: Signature::default(), - error: Some(e), - swqos_type, - landed_on_chain: false, - submit_done_us: crate::common::clock::now_micros(), - }); - return; - } - }; - - // Transaction built - - let _send_start = Instant::now(); - let mut err: Option = None; - #[allow(unused_assignments)] - let mut landed_on_chain = false; - let success = match swqos_client - .send_transaction( - if is_buy { TradeType::Buy } else { TradeType::Sell }, - &transaction, - wait_transaction_confirmed, - ) - .await - { - Ok(()) => { - landed_on_chain = true; // Success means tx confirmed on-chain - true - } - Err(e) => { - // Check if this error indicates the tx landed but failed (e.g., ExceededSlippage) - landed_on_chain = is_landed_error(&e); - err = Some(e); - // Send transaction failed - false - } - }; - - // Transaction sent: always submit a result so collector never has "no result" for this task. - // If transaction has no signatures (malformed), submit with default signature and success=false. - let sig = transaction.signatures.first().copied().unwrap_or_default(); - collector.submit(TaskResult { - success, - signature: sig, - error: err, + tip_account, + swqos_client, swqos_type, - landed_on_chain, - submit_done_us: crate::common::clock::now_micros(), - }); - }); + core_id, + use_affinity: use_core_affinity, + }; + let _ = queue.push(job); + } } - // All tasks spawned + // All jobs enqueued (no spawn on hot path) if !wait_transaction_confirmed { const SUBMIT_TIMEOUT_SECS: u64 = 30; diff --git a/src/trading/core/executor.rs b/src/trading/core/executor.rs index 89b1372..c4debb7 100755 --- a/src/trading/core/executor.rs +++ b/src/trading/core/executor.rs @@ -77,7 +77,7 @@ impl TradeExecutor for GenericTradeExecutor { Some(middleware_manager) => middleware_manager .apply_middlewares_process_protocol_instructions( instructions, - self.protocol_name.to_string(), + self.protocol_name, is_buy, )?, None => instructions, @@ -144,53 +144,51 @@ impl TradeExecutor for GenericTradeExecutor { .await; let log_enabled = params.log_enabled && crate::common::sdk_log::sdk_log_enabled(); - let submit_timings = if log_enabled { - result.as_ref().ok().map(|(_, _, _, t)| t.clone()).unwrap_or_default() - } else { - Vec::new() + + let (ok, signatures, err, submit_timings) = match result { + Ok((success, sigs, last_error, timings)) => ( + success, + sigs, + last_error.map(|e| anyhow::anyhow!("{}", e)), + timings, + ), + Err(e) => (false, vec![], Some(anyhow::anyhow!("{}", e)), vec![]), }; + let submit_timings_ref: &[(crate::swqos::SwqosType, i64)] = submit_timings.as_slice(); let result = if need_confirm { - let (ok, sigs, err) = match &result { - Ok((success, signatures, last_error, _)) => ( - *success, - signatures.clone(), - last_error.as_ref().map(|e| anyhow::anyhow!("{}", e)), - ), - Err(e) => (false, vec![], Some(anyhow::anyhow!("{}", e))), - }; let confirm_result = if let Some(rpc) = params.rpc.as_ref() { - if sigs.is_empty() { - (ok, sigs, err) + if signatures.is_empty() { + (ok, signatures, err) } else { - let poll_res = poll_any_transaction_confirmation(rpc, &sigs, true).await; - 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 { - 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); + let poll_res = poll_any_transaction_confirmation(rpc, &signatures, true).await; + 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); + } } } } - } - match poll_res { - Ok(_) => (true, sigs, None), - Err(e) => (false, sigs, Some(e)), - } + match poll_res { + Ok(_) => (true, signatures, None), + Err(e) => (false, signatures, Some(e)), + } } } else { - (ok, sigs, err) + (ok, signatures, err) }; Ok(confirm_result) } else { @@ -203,13 +201,13 @@ impl TradeExecutor for GenericTradeExecutor { 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 { + 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); } } } - result.map(|(a, b, c, _)| (a, b, c)) + Ok((ok, signatures, err)) }; result @@ -256,22 +254,21 @@ async fn simulate_transaction( let unit_limit = default_config.2.cu_limit; let unit_price = default_config.2.cu_price; - // Build transaction for simulation let transaction = build_transaction( - payer.clone(), - Some(rpc.clone()), + &payer, + Some(&rpc), unit_limit, unit_price, &instructions, - address_lookup_table_account, + address_lookup_table_account.as_ref(), recent_blockhash, - middleware_manager, + middleware_manager.as_ref(), protocol_name, is_buy, - false, // simulate doesn't need tip instruction + false, &Pubkey::default(), tip, - durable_nonce, + durable_nonce.as_ref(), ) .await?; diff --git a/src/trading/core/params.rs b/src/trading/core/params.rs index 461dbb7..93b57ed 100755 --- a/src/trading/core/params.rs +++ b/src/trading/core/params.rs @@ -398,7 +398,7 @@ impl PumpSwapParams { ); Ok(Self { - pool: pool_address.clone(), + pool: *pool_address, base_mint: pool_data.base_mint, quote_mint: pool_data.quote_mint, pool_base_token_account: pool_data.pool_base_token_account, @@ -671,7 +671,7 @@ impl RaydiumCpmmParams { ) .await?; Ok(Self { - pool_state: pool_address.clone(), + pool_state: *pool_address, amm_config: pool.amm_config, base_mint: pool.token0_mint, quote_mint: pool.token1_mint, @@ -778,7 +778,7 @@ impl MeteoraDammV2Params { let pool_data = crate::instruction::utils::meteora_damm_v2::fetch_pool(rpc, pool_address).await?; Ok(Self { - pool: pool_address.clone(), + pool: *pool_address, token_a_vault: pool_data.token_a_vault, token_b_vault: pool_data.token_b_vault, token_a_mint: pool_data.token_a_mint, diff --git a/src/trading/core/transaction_pool.rs b/src/trading/core/transaction_pool.rs index d8d0473..28220c6 100644 --- a/src/trading/core/transaction_pool.rs +++ b/src/trading/core/transaction_pool.rs @@ -6,6 +6,15 @@ //! - 零拷贝 I/O //! - 内存预热 +/// 预分配指令容量（单笔交易常见指令数） +const TX_BUILDER_INSTRUCTION_CAP: usize = 32; +/// 预分配地址查找表数量 +const TX_BUILDER_LOOKUP_TABLE_CAP: usize = 8; +/// 对象池最大容量 +const TX_BUILDER_POOL_CAP: usize = 1000; +/// 启动时预填充对象池数量 +const TX_BUILDER_POOL_PREFILL: usize = 100; + use crossbeam_queue::ArrayQueue; use once_cell::sync::Lazy; use solana_sdk::{ @@ -23,8 +32,8 @@ pub struct PreallocatedTxBuilder { impl PreallocatedTxBuilder { fn new() -> Self { Self { - instructions: Vec::with_capacity(32), // 预分配32条指令空间 - lookup_tables: Vec::with_capacity(8), // 预分配8个查找表空间 + instructions: Vec::with_capacity(TX_BUILDER_INSTRUCTION_CAP), + lookup_tables: Vec::with_capacity(TX_BUILDER_LOOKUP_TABLE_CAP), } } @@ -65,23 +74,20 @@ impl PreallocatedTxBuilder { &mut self, payer: &Pubkey, instructions: &[Instruction], - address_lookup_table_account: Option, + address_lookup_table_account: Option<&AddressLookupTableAccount>, recent_blockhash: Hash, ) -> VersionedMessage { - // 重用已分配的 vector self.reset(); self.instructions.extend_from_slice(instructions); - // ✅ 如果有查找表，使用 V0 消息 - if let Some(address_lookup_table_account) = address_lookup_table_account { - let message = v0::Message::try_compile( + if let Some(alt) = address_lookup_table_account { + let message = v0::Message::try_compile( payer, &self.instructions, - &[address_lookup_table_account], + std::slice::from_ref(alt), recent_blockhash, - ).expect("v0 message compile failed"); - - + ) + .expect("v0 message compile failed"); VersionedMessage::V0(message) } else { // ✅ 没有查找表，使用 Legacy 消息（兼容所有 RPC） @@ -97,10 +103,9 @@ impl PreallocatedTxBuilder { /// 🚀 全局交易构建器对象池 static TX_BUILDER_POOL: Lazy>> = Lazy::new(|| { - let pool = ArrayQueue::new(1000); // 1000个预分配构建器 + let pool = ArrayQueue::new(TX_BUILDER_POOL_CAP); - // 预填充池 - for _ in 0..100 { + for _ in 0..TX_BUILDER_POOL_PREFILL { let _ = pool.push(PreallocatedTxBuilder::new()); } diff --git a/src/trading/middleware/builtin.rs b/src/trading/middleware/builtin.rs index 65dc101..90b36d3 100644 --- a/src/trading/middleware/builtin.rs +++ b/src/trading/middleware/builtin.rs @@ -14,7 +14,7 @@ impl InstructionMiddleware for LoggingMiddleware { fn process_protocol_instructions( &self, protocol_instructions: Vec, - protocol_name: String, + protocol_name: &str, is_buy: bool, ) -> Result> { println!("-------------------[{}]-------------------", self.name()); @@ -32,7 +32,7 @@ impl InstructionMiddleware for LoggingMiddleware { fn process_full_instructions( &self, full_instructions: Vec, - protocol_name: String, + protocol_name: &str, is_buy: bool, ) -> Result> { println!("-------------------[{}]-------------------", self.name()); diff --git a/src/trading/middleware/traits.rs b/src/trading/middleware/traits.rs index 081f6a8..b2453dd 100644 --- a/src/trading/middleware/traits.rs +++ b/src/trading/middleware/traits.rs @@ -20,7 +20,7 @@ pub trait InstructionMiddleware: Send + Sync { fn process_protocol_instructions( &self, protocol_instructions: Vec, - protocol_name: String, + protocol_name: &str, is_buy: bool, ) -> Result>; @@ -36,7 +36,7 @@ pub trait InstructionMiddleware: Send + Sync { fn process_full_instructions( &self, full_instructions: Vec, - protocol_name: String, + protocol_name: &str, is_buy: bool, ) -> Result>; @@ -72,13 +72,13 @@ impl MiddlewareManager { pub fn apply_middlewares_process_full_instructions( &self, mut full_instructions: Vec, - protocol_name: String, + protocol_name: &str, is_buy: bool, ) -> Result> { for middleware in &self.middlewares { full_instructions = middleware.process_full_instructions( full_instructions, - protocol_name.clone(), + protocol_name, is_buy, )?; if full_instructions.is_empty() { @@ -92,13 +92,13 @@ impl MiddlewareManager { pub fn apply_middlewares_process_protocol_instructions( &self, mut protocol_instructions: Vec, - protocol_name: String, + protocol_name: &str, is_buy: bool, ) -> Result> { for middleware in &self.middlewares { protocol_instructions = middleware.process_protocol_instructions( protocol_instructions, - protocol_name.clone(), + protocol_name, is_buy, )?; if protocol_instructions.is_empty() {