From 5a9238a741197345750bd8f993ca2c3db66cf2cb Mon Sep 17 00:00:00 2001 From: ysq Date: Thu, 21 Aug 2025 21:32:53 +0800 Subject: [PATCH] feat: add transaction confirmation wait option and parallel execution optimization - Add wait_transaction_confirmed parameter to all trading methods - Support async transaction sending without waiting for confirmation for better performance - Optimize parallel execution logic to return on first successful transaction - Fix direct priority_fee modification issue - Improve error handling and resource management - Update all example code to support new parameter --- src/lib.rs | 35 +++++++++++++++++++++--------- src/main.rs | 17 +++++++++++++++ src/trading/core/executor.rs | 17 +++++++++++++-- src/trading/core/parallel.rs | 42 +++++++++++++++++++++++++----------- src/trading/core/params.rs | 6 ++++++ 5 files changed, 92 insertions(+), 25 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 85c31e8..0ef4ab1 100755 --- a/src/lib.rs +++ b/src/lib.rs @@ -136,6 +136,7 @@ impl SolanaTrade { /// * `custom_buy_tip_fee` - Optional custom tip fee for priority processing (in SOL) /// * `extension_params` - Optional protocol-specific parameters (uses defaults if None) /// * `lookup_table_key` - Optional address lookup table key for transaction optimization + /// * `wait_transaction_confirmed` - Whether to wait for the transaction to be confirmed /// /// # Returns /// @@ -184,13 +185,14 @@ impl SolanaTrade { custom_buy_tip_fee: Option, extension_params: Box, lookup_table_key: Option, + wait_transaction_confirmed: bool, ) -> Result<(), anyhow::Error> { let executor = TradeFactory::create_executor(dex_type.clone()); let protocol_params = extension_params; let final_lookup_table_key = lookup_table_key.or(self.trade_config.lookup_table_key); - let buy_params = BuyParams { + let mut buy_params = BuyParams { rpc: Some(self.rpc.clone()), payer: self.payer.clone(), mint: mint, @@ -201,13 +203,17 @@ impl SolanaTrade { lookup_table_key: final_lookup_table_key, recent_blockhash, data_size_limit: 0, + wait_transaction_confirmed: wait_transaction_confirmed, protocol_params: protocol_params.clone(), }; - let mut priority_fee = buy_params.priority_fee.clone(); if custom_buy_tip_fee.is_some() { - priority_fee.buy_tip_fee = custom_buy_tip_fee.unwrap(); - priority_fee.buy_tip_fees = - priority_fee.buy_tip_fees.iter().map(|_| custom_buy_tip_fee.unwrap()).collect(); + buy_params.priority_fee.buy_tip_fee = custom_buy_tip_fee.unwrap(); + buy_params.priority_fee.buy_tip_fees = buy_params + .priority_fee + .buy_tip_fees + .iter() + .map(|_| custom_buy_tip_fee.unwrap()) + .collect(); } let buy_with_tip_params = buy_params.clone().with_tip(self.swqos_clients.clone()); @@ -247,6 +253,7 @@ impl SolanaTrade { /// * `with_tip` - Optional boolean to indicate if the transaction should be sent with tip /// * `extension_params` - Optional protocol-specific parameters (uses defaults if None) /// * `lookup_table_key` - Optional address lookup table key for transaction optimization + /// * `wait_transaction_confirmed` - Whether to wait for the transaction to be confirmed /// /// # Returns /// @@ -298,13 +305,14 @@ impl SolanaTrade { with_tip: bool, extension_params: Box, lookup_table_key: Option, + wait_transaction_confirmed: bool, ) -> Result<(), anyhow::Error> { let executor = TradeFactory::create_executor(dex_type.clone()); let protocol_params = extension_params; let final_lookup_table_key = lookup_table_key.or(self.trade_config.lookup_table_key); - let sell_params = SellParams { + let mut sell_params = SellParams { rpc: Some(self.rpc.clone()), payer: self.payer.clone(), mint: mint, @@ -314,13 +322,17 @@ impl SolanaTrade { priority_fee: self.trade_config.priority_fee.clone(), lookup_table_key: final_lookup_table_key, recent_blockhash, + wait_transaction_confirmed: wait_transaction_confirmed, protocol_params: protocol_params.clone(), }; - let mut priority_fee = sell_params.priority_fee.clone(); if custom_buy_tip_fee.is_some() { - priority_fee.buy_tip_fee = custom_buy_tip_fee.unwrap(); - priority_fee.buy_tip_fees = - priority_fee.buy_tip_fees.iter().map(|_| custom_buy_tip_fee.unwrap()).collect(); + sell_params.priority_fee.buy_tip_fee = custom_buy_tip_fee.unwrap(); + sell_params.priority_fee.buy_tip_fees = sell_params + .priority_fee + .buy_tip_fees + .iter() + .map(|_| custom_buy_tip_fee.unwrap()) + .collect(); } let sell_with_tip_params = sell_params.clone().with_tip(self.swqos_clients.clone()); @@ -369,6 +381,7 @@ impl SolanaTrade { /// * `with_tip` - Whether to use tip for priority processing /// * `extension_params` - Optional protocol-specific parameters (uses defaults if None) /// * `lookup_table_key` - Optional lookup table key for address lookup optimization + /// * `wait_transaction_confirmed` - Whether to wait for the transaction to be confirmed /// /// # Returns /// @@ -425,6 +438,7 @@ impl SolanaTrade { with_tip: bool, extension_params: Box, lookup_table_key: Option, + wait_transaction_confirmed: bool, ) -> Result<(), anyhow::Error> { if percent == 0 || percent > 100 { return Err(anyhow::anyhow!("Percentage must be between 1 and 100")); @@ -441,6 +455,7 @@ impl SolanaTrade { with_tip, extension_params, lookup_table_key, + wait_transaction_confirmed, ) .await } diff --git a/src/main.rs b/src/main.rs index 4416c3b..09cd7e3 100755 --- a/src/main.rs +++ b/src/main.rs @@ -120,6 +120,7 @@ async fn test_middleware() -> AnyResult<()> { // Through RPC call, adds latency. Can optimize by using from_buy_trade or manually initializing PumpSwapParams Box::new(PumpSwapParams::from_pool_address_by_rpc(&client.rpc, &pool_address).await?), None, + true, ) .await?; Ok(()) @@ -148,6 +149,7 @@ async fn test_pumpfun_copy_trade_with_grpc(trade_info: PumpFunTradeEvent) -> Any None, Box::new(PumpFunParams::from_trade(&trade_info, None)), None, + true, ) .await?; @@ -166,6 +168,7 @@ async fn test_pumpfun_copy_trade_with_grpc(trade_info: PumpFunTradeEvent) -> Any false, Box::new(PumpFunParams::from_trade(&trade_info, None)), None, + true, ) .await?; @@ -205,6 +208,7 @@ async fn test_pumpfun_sniper_trade_with_shreds(trade_info: PumpFunTradeEvent) -> None, )), None, + true, ) .await?; @@ -229,6 +233,7 @@ async fn test_pumpfun_sniper_trade_with_shreds(trade_info: PumpFunTradeEvent) -> None, )), None, + true, ) .await?; @@ -260,6 +265,7 @@ async fn test_pumpswap() -> AnyResult<()> { // Through RPC call, adds latency. Can optimize by using from_buy_trade or manually initializing PumpSwapParams Box::new(PumpSwapParams::from_pool_address_by_rpc(&client.rpc, &pool_address).await?), None, + true, ) .await?; @@ -279,6 +285,7 @@ async fn test_pumpswap() -> AnyResult<()> { // Through RPC call, adds latency. Can optimize by using from_sell_trade or manually initializing PumpSwapParams Box::new(PumpSwapParams::from_pool_address_by_rpc(&client.rpc, &pool_address).await?), None, + true, ) .await?; @@ -307,6 +314,7 @@ async fn test_bonk_copy_trade_with_grpc(trade_info: BonkTradeEvent) -> AnyResult None, Box::new(BonkParams::from_trade(trade_info.clone())), None, + true, ) .await?; @@ -325,6 +333,7 @@ async fn test_bonk_copy_trade_with_grpc(trade_info: BonkTradeEvent) -> AnyResult false, Box::new(BonkParams::from_trade(trade_info)), None, + true, ) .await?; @@ -357,6 +366,7 @@ async fn test_bonk_sniper_trade_with_shreds(trade_info: BonkTradeEvent) -> AnyRe None, Box::new(BonkParams::from_dev_trade(trade_info.clone())), None, + true, ) .await?; @@ -375,6 +385,7 @@ async fn test_bonk_sniper_trade_with_shreds(trade_info: BonkTradeEvent) -> AnyRe false, Box::new(BonkParams::from_dev_trade(trade_info)), None, + true, ) .await?; @@ -404,6 +415,7 @@ async fn test_bonk() -> Result<(), Box> { // Through RPC call, adds latency. Can optimize by using from_trade or manually initializing BonkParams Box::new(BonkParams::from_mint_by_rpc(&client.rpc, &mint_pubkey).await?), None, + true, ) .await?; @@ -423,6 +435,7 @@ async fn test_bonk() -> Result<(), Box> { // Through RPC call, adds latency. Can optimize by using from_trade or manually initializing BonkParams Box::new(BonkParams::from_mint_by_rpc(&client.rpc, &mint_pubkey).await?), None, + true, ) .await?; @@ -454,6 +467,7 @@ async fn test_raydium_cpmm() -> Result<(), Box> { RaydiumCpmmParams::from_pool_address_by_rpc(&client.rpc, &pool_address).await?, ), None, + true, ) .await?; @@ -475,6 +489,7 @@ async fn test_raydium_cpmm() -> Result<(), Box> { RaydiumCpmmParams::from_pool_address_by_rpc(&client.rpc, &pool_address).await?, ), None, + true, ) .await?; @@ -504,6 +519,7 @@ async fn test_raydium_amm_v4() -> Result<(), Box> { // Through RPC call, adds latency, or from_amm_info_and_reserves or manually initialize RaydiumAmmV4Params Box::new(RaydiumAmmV4Params::from_amm_address_by_rpc(&client.rpc, amm_address).await?), None, + true, ) .await?; @@ -523,6 +539,7 @@ async fn test_raydium_amm_v4() -> Result<(), Box> { // Through RPC call, adds latency, or from_amm_info_and_reserves or manually initialize RaydiumAmmV4Params Box::new(RaydiumAmmV4Params::from_amm_address_by_rpc(&client.rpc, amm_address).await?), None, + true, ) .await?; diff --git a/src/trading/core/executor.rs b/src/trading/core/executor.rs index fbcceeb..0fffcca 100755 --- a/src/trading/core/executor.rs +++ b/src/trading/core/executor.rs @@ -76,7 +76,12 @@ impl TradeExecutor for GenericTradeExecutor { timer.stage("rpc提交确认"); // 发送交易 - rpc.send_and_confirm_transaction(&transaction).await?; + if params.wait_transaction_confirmed { + rpc.send_and_confirm_transaction(&transaction).await?; + } else { + // 异步发送交易 + rpc.send_transaction(&transaction).await?; + } timer.finish(); Ok(()) @@ -104,6 +109,7 @@ impl TradeExecutor for GenericTradeExecutor { lookup_table_key: params.lookup_table_key, recent_blockhash: params.recent_blockhash, data_size_limit: params.data_size_limit, + wait_transaction_confirmed: params.wait_transaction_confirmed, protocol_params: params.protocol_params.clone(), }; @@ -134,6 +140,7 @@ impl TradeExecutor for GenericTradeExecutor { middleware_manager, self.protocol_name.to_string(), true, + params.wait_transaction_confirmed, ) .await?; @@ -179,7 +186,11 @@ impl TradeExecutor for GenericTradeExecutor { timer.stage("卖出交易签名"); // 发送交易 - rpc.send_and_confirm_transaction(&transaction).await?; + if params.wait_transaction_confirmed { + rpc.send_and_confirm_transaction(&transaction).await?; + } else { + rpc.send_transaction(&transaction).await?; + } timer.finish(); Ok(()) @@ -203,6 +214,7 @@ impl TradeExecutor for GenericTradeExecutor { priority_fee: params.priority_fee.clone(), lookup_table_key: params.lookup_table_key, recent_blockhash: params.recent_blockhash, + wait_transaction_confirmed: params.wait_transaction_confirmed, protocol_params: params.protocol_params.clone(), }; @@ -233,6 +245,7 @@ impl TradeExecutor for GenericTradeExecutor { middleware_manager, self.protocol_name.to_string(), false, + params.wait_transaction_confirmed, ) .await?; diff --git a/src/trading/core/parallel.rs b/src/trading/core/parallel.rs index 68e60ba..93ed0a0 100755 --- a/src/trading/core/parallel.rs +++ b/src/trading/core/parallel.rs @@ -2,6 +2,7 @@ use anyhow::{anyhow, Result}; use solana_hash::Hash; use solana_sdk::{instruction::Instruction, pubkey::Pubkey, signature::Keypair}; use std::{str::FromStr, sync::Arc}; +use tokio::sync::mpsc; use tokio::task::JoinHandle; use crate::{ @@ -30,6 +31,7 @@ pub async fn parallel_execute_with_tips( middleware_manager: Option>, protocol_name: String, is_buy: bool, + wait_transaction_confirmed: bool, ) -> Result<()> { let cores = core_affinity::get_core_ids().unwrap(); let mut handles: Vec>> = vec![]; @@ -97,7 +99,7 @@ pub async fn parallel_execute_with_tips( } else { let tip_account = swqos_client.get_tip_account()?; let tip_account = Arc::new(Pubkey::from_str(&tip_account).map_err(|e| anyhow!(e))?); - priority_fee.buy_tip_fee = priority_fee.buy_tip_fees[i]; + priority_fee.buy_tip_fee = priority_fee.buy_tip_fees[i % priority_fee.buy_tip_fees.len()]; build_tip_transaction_with_priority_fee( payer, @@ -125,22 +127,36 @@ pub async fn parallel_execute_with_tips( handles.push(handle); } - // 等待所有任务完成 - let mut errors = Vec::new(); + // 任意一个成功即返回 + let (tx, mut rx) = mpsc::channel(swqos_clients.len()); + + // 启动监听任务 for handle in handles { - match handle.await { - Ok(Ok(_)) => (), + let tx = tx.clone(); + tokio::spawn(async move { + let result = handle.await; + let _ = tx.send(result).await; + }); + } + drop(tx); // 关闭发送端 + + // 等待第一个成功的结果 + let mut errors = Vec::new(); + + if !wait_transaction_confirmed { + return Ok(()); + } + + while let Some(result) = rx.recv().await { + match result { + Ok(Ok(_)) => { + return Ok(()); + } Ok(Err(e)) => errors.push(format!("Task error: {}", e)), Err(e) => errors.push(format!("Join error: {}", e)), } } - if !errors.is_empty() { - for error in &errors { - println!("{}", error); - } - return Err(anyhow!("Some tasks failed: {:?}", errors)); - } - - Ok(()) + // 如果没有成功的，返回错误 + return Err(anyhow!("所有交易都失败了: {:?}", errors)); } diff --git a/src/trading/core/params.rs b/src/trading/core/params.rs index 0f37980..ac21e63 100755 --- a/src/trading/core/params.rs +++ b/src/trading/core/params.rs @@ -36,6 +36,7 @@ pub struct BuyParams { pub lookup_table_key: Option, pub recent_blockhash: Hash, pub data_size_limit: u32, + pub wait_transaction_confirmed: bool, pub protocol_params: Box, } @@ -54,6 +55,7 @@ pub struct BuyWithTipParams { pub lookup_table_key: Option, pub recent_blockhash: Hash, pub data_size_limit: u32, + pub wait_transaction_confirmed: bool, pub protocol_params: Box, } @@ -70,6 +72,7 @@ pub struct SellParams { pub priority_fee: PriorityFee, pub lookup_table_key: Option, pub recent_blockhash: Hash, + pub wait_transaction_confirmed: bool, pub protocol_params: Box, } @@ -87,6 +90,7 @@ pub struct SellWithTipParams { pub priority_fee: PriorityFee, pub lookup_table_key: Option, pub recent_blockhash: Hash, + pub wait_transaction_confirmed: bool, pub protocol_params: Box, } @@ -456,6 +460,7 @@ impl BuyParams { lookup_table_key: self.lookup_table_key, recent_blockhash: self.recent_blockhash, data_size_limit: self.data_size_limit, + wait_transaction_confirmed: self.wait_transaction_confirmed, protocol_params: self.protocol_params, } } @@ -476,6 +481,7 @@ impl SellParams { priority_fee: self.priority_fee, lookup_table_key: self.lookup_table_key, recent_blockhash: self.recent_blockhash, + wait_transaction_confirmed: self.wait_transaction_confirmed, protocol_params: self.protocol_params, } }