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

@@ -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<f64>,
         extension_params: Box<dyn ProtocolParams>,
         lookup_table_key: Option<Pubkey>,
+        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();
+            buy_params.priority_fee.buy_tip_fee = custom_buy_tip_fee.unwrap();
-            priority_fee.buy_tip_fees =
+            buy_params.priority_fee.buy_tip_fees = buy_params
-                priority_fee.buy_tip_fees.iter().map(|_| custom_buy_tip_fee.unwrap()).collect();
+                .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<dyn ProtocolParams>,
         lookup_table_key: Option<Pubkey>,
+        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();
+            sell_params.priority_fee.buy_tip_fee = custom_buy_tip_fee.unwrap();
-            priority_fee.buy_tip_fees =
+            sell_params.priority_fee.buy_tip_fees = sell_params
-                priority_fee.buy_tip_fees.iter().map(|_| custom_buy_tip_fee.unwrap()).collect();
+                .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<dyn ProtocolParams>,
         lookup_table_key: Option<Pubkey>,
+        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
     }

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<dyn std::error::Error>> {
             // 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<dyn std::error::Error>> {
             // 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<dyn std::error::Error>> {
                 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<dyn std::error::Error>> {
                 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<dyn std::error::Error>> {
             // 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<dyn std::error::Error>> {
             // 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?;
 

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?;
 

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<Arc<MiddlewareManager>>,
     protocol_name: String,
     is_buy: bool,
+    wait_transaction_confirmed: bool,
 ) -> Result<()> {
     let cores = core_affinity::get_core_ids().unwrap();
     let mut handles: Vec<JoinHandle<Result<()>>> = 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 {
+        let tx = tx.clone();
-            Ok(Ok(_)) => (),
+        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 {
+    return Err(anyhow!("所有交易都失败了: {:?}", errors));
-            println!("{}", error);
-        }
-        return Err(anyhow!("Some tasks failed: {:?}", errors));
-    }
-
-    Ok(())
 }

src/trading/core/params.rs

@@ -36,6 +36,7 @@ pub struct BuyParams {
     pub lookup_table_key: Option<Pubkey>,
     pub recent_blockhash: Hash,
     pub data_size_limit: u32,
+    pub wait_transaction_confirmed: bool,
     pub protocol_params: Box<dyn ProtocolParams>,
 }
 
@@ -54,6 +55,7 @@ pub struct BuyWithTipParams {
     pub lookup_table_key: Option<Pubkey>,
     pub recent_blockhash: Hash,
     pub data_size_limit: u32,
+    pub wait_transaction_confirmed: bool,
     pub protocol_params: Box<dyn ProtocolParams>,
 }
 
@@ -70,6 +72,7 @@ pub struct SellParams {
     pub priority_fee: PriorityFee,
     pub lookup_table_key: Option<Pubkey>,
     pub recent_blockhash: Hash,
+    pub wait_transaction_confirmed: bool,
     pub protocol_params: Box<dyn ProtocolParams>,
 }
 
@@ -87,6 +90,7 @@ pub struct SellWithTipParams {
     pub priority_fee: PriorityFee,
     pub lookup_table_key: Option<Pubkey>,
     pub recent_blockhash: Hash,
+    pub wait_transaction_confirmed: bool,
     pub protocol_params: Box<dyn ProtocolParams>,
 }
 
@@ -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,
         }
     }