From f1434a82c5009df02947cd73c60bfb0d32a8a14d Mon Sep 17 00:00:00 2001 From: vibes Date: Fri, 19 Dec 2025 08:48:54 +0000 Subject: [PATCH] fix: early exit when tx lands but fails on-chain MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When using multiple SWQOS channels with durable nonce, if one channel's transaction lands on-chain but fails (e.g., ExceededSlippage), the nonce is consumed and other channels cannot succeed. Previously, the SDK would wait for all channels to timeout (~15s) before returning, and would often return a generic "timeout" error instead of the actual on-chain error. This fix: - Adds `landed_on_chain` field to TaskResult to track if tx landed - Adds `is_landed_error()` to detect on-chain failures vs timeouts - Adds `landed_failed_flag` to ResultCollector for early exit signaling - Returns immediately with the actual error when a tx lands but fails This makes behavior consistent with the success case - both return immediately once the nonce is consumed. Benefits: 1. Get the actual error (e.g., ExceededSlippage) instead of timeout 2. Return in ~1s instead of ~15s 3. Enable faster retry logic for callers 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- src/trading/core/async_executor.rs | 66 +++++++++++++++++++++++++++--- 1 file changed, 61 insertions(+), 5 deletions(-) diff --git a/src/trading/core/async_executor.rs b/src/trading/core/async_executor.rs index e1948d4..e46ef5f 100644 --- a/src/trading/core/async_executor.rs +++ b/src/trading/core/async_executor.rs @@ -37,11 +37,37 @@ struct TaskResult { signature: Signature, error: Option, swqos_type: SwqosType, // 🔧 增加：记录SWQOS类型 + landed_on_chain: bool, // 🔧 Whether tx landed on-chain (even if failed) +} + +/// Check if an error indicates the transaction landed on-chain (vs network/timeout error) +fn is_landed_error(error: &anyhow::Error) -> bool { + use crate::swqos::common::TradeError; + + // If it's a TradeError with a non-zero code, the tx landed but failed on-chain + if let Some(trade_error) = error.downcast_ref::() { + // Code 500 with "timed out" message means tx never landed + if trade_error.code == 500 && trade_error.message.contains("timed out") { + return false; + } + // Any other TradeError means the tx landed (e.g., ExceededSlippage = 6004) + return trade_error.code > 0; + } + + // Check error message for timeout indication + let msg = error.to_string(); + if msg.contains("timed out") || msg.contains("timeout") { + return false; + } + + // Assume other errors might indicate landed tx (be conservative) + false } struct ResultCollector { results: Arc>, success_flag: Arc, + landed_failed_flag: Arc, // 🔧 Tx landed on-chain but failed (nonce consumed) completed_count: Arc, total_tasks: usize, } @@ -51,6 +77,7 @@ impl ResultCollector { Self { results: Arc::new(ArrayQueue::new(capacity)), success_flag: Arc::new(AtomicBool::new(false)), + landed_failed_flag: Arc::new(AtomicBool::new(false)), completed_count: Arc::new(AtomicUsize::new(0)), total_tasks: capacity, } @@ -59,11 +86,15 @@ impl ResultCollector { fn submit(&self, result: TaskResult) { // 🚀 优化：ArrayQueue 内部已保证同步，无需额外 fence let is_success = result.success; + let is_landed_failed = result.landed_on_chain && !result.success; let _ = self.results.push(result); if is_success { self.success_flag.store(true, Ordering::Release); // Release 确保 push 可见 + } else if is_landed_failed { + // 🔧 Tx landed but failed (e.g., ExceededSlippage) - nonce is consumed, no point waiting + self.landed_failed_flag.store(true, Ordering::Release); } self.completed_count.fetch_add(1, Ordering::Release); @@ -90,6 +121,23 @@ impl ResultCollector { } } + // 🔧 Early exit: if a tx landed but failed (e.g., ExceededSlippage), + // nonce is consumed and other channels can't succeed - return immediately + if self.landed_failed_flag.load(Ordering::Acquire) { + let mut signatures = Vec::new(); + let mut landed_error = None; + while let Some(result) = self.results.pop() { + signatures.push(result.signature); + // Prefer the error from the tx that actually landed + if result.landed_on_chain && result.error.is_some() { + landed_error = result.error; + } + } + if !signatures.is_empty() { + return Some((false, signatures, landed_error)); + } + } + let completed = self.completed_count.load(Ordering::Acquire); if completed >= self.total_tasks { // 🔧 修复：收集所有签名 @@ -289,6 +337,7 @@ pub async fn execute_parallel( signature: Signature::default(), error: Some(e), swqos_type, // 🔧 记录SWQOS类型 + landed_on_chain: false, // Build failed, tx never sent }); return; } @@ -297,7 +346,8 @@ pub async fn execute_parallel( // Transaction built let _send_start = Instant::now(); - let mut err = None; + let mut err: Option = None; + let mut landed_on_chain = false; let success = match swqos_client .send_transaction( if is_buy { TradeType::Buy } else { TradeType::Sell }, @@ -305,8 +355,13 @@ pub async fn execute_parallel( ) .await { - Ok(()) => true, + 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 @@ -316,11 +371,12 @@ pub async fn execute_parallel( // Transaction sent if let Some(signature) = transaction.signatures.first() { - collector.submit(TaskResult { - success, - signature: *signature, + collector.submit(TaskResult { + success, + signature: *signature, error: err, swqos_type, // 🔧 记录SWQOS类型 + landed_on_chain, // 🔧 Whether tx landed (even if it failed) }); } });