From 5d921f23ff998287f531aa78e4b76c733fd8db47 Mon Sep 17 00:00:00 2001 From: Wood Date: Wed, 18 Mar 2026 17:29:14 +0800 Subject: [PATCH] Improve ZeroSlot client: add HTTP keep-alive ping and switch to Binary-Tx - Add HTTP keep-alive ping task (30s interval) using free getHealth method - Switch from JSON-RPC to faster Binary-Tx endpoint (/txb) for transaction submission - Send raw binary transaction bytes directly to avoid encoding/decoding overhead - Update response handling for Binary-Tx plain text status codes (200/403/419/500) - Follow same keep-alive pattern as BlockRazor, Stellium, and Astralane clients These changes reduce first-submit cold start latency and overall transaction submission time. Co-Authored-By: Claude Sonnet 4.6 --- src/swqos/zeroslot.rs | 176 ++++++++++++++++++++++++++++++++++-------- 1 file changed, 143 insertions(+), 33 deletions(-) diff --git a/src/swqos/zeroslot.rs b/src/swqos/zeroslot.rs index 42dc306..cb93d6d 100755 --- a/src/swqos/zeroslot.rs +++ b/src/swqos/zeroslot.rs @@ -1,12 +1,12 @@ use crate::swqos::common::{ - default_http_client_builder, poll_transaction_confirmation, serialize_transaction_and_encode, + default_http_client_builder, poll_transaction_confirmation, }; use rand::seq::IndexedRandom; use reqwest::Client; -use serde_json::json; -use std::{sync::Arc, time::Instant}; - -use solana_transaction_status::UiTransactionEncoding; +use std::{sync::Arc, time::Instant, time::Duration}; +use std::sync::atomic::{AtomicBool, Ordering}; +use tokio::task::JoinHandle; +use bincode; use crate::swqos::SwqosClientTrait; use crate::swqos::{SwqosType, TradeType}; @@ -21,6 +21,8 @@ pub struct ZeroSlotClient { pub auth_token: String, pub rpc_client: Arc, pub http_client: Client, + pub ping_handle: Arc>>>, + pub stop_ping: Arc, } #[async_trait::async_trait] @@ -60,7 +62,79 @@ impl ZeroSlotClient { 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(); - Self { rpc_client: Arc::new(rpc_client), endpoint, auth_token, http_client } + + 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)), + }; + + // 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 + 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 { + // Immediate first ping to warm connection and reduce first-submit cold start latency + if let Err(e) = Self::send_ping_request(&http_client, &endpoint, &auth_token).await { + if crate::common::sdk_log::sdk_log_enabled() { + eprintln!("0slot ping request failed: {}", e); + } + } + let mut interval = tokio::time::interval(Duration::from_secs(30)); // 30s keepalive under 65s server timeout + 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 { + if crate::common::sdk_log::sdk_log_enabled() { + eprintln!("0slot ping request failed: {}", e); + } + } + } + }); + + // 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(); + } + *ping_guard = Some(handle); + } + } + + /// Send ping request: POST with getHealth method (Keep Alive). Free operation, not counted toward TPS. + async fn send_ping_request( + http_client: &Client, + endpoint: &str, + auth_token: &str, + ) -> Result<()> { + let url = format!("{}/?api-key={}", endpoint, auth_token); + let response = http_client + .post(&url) + .header("Content-Type", "application/json") + .timeout(Duration::from_millis(1500)) + .body(r#"{"jsonrpc":"2.0","id":1,"method":"getHealth"}"#) + .send() + .await?; + let _ = response.bytes().await; + Ok(()) } pub async fn send_transaction( @@ -70,47 +144,65 @@ impl ZeroSlotClient { wait_confirmation: bool, ) -> Result<()> { let start_time = Instant::now(); - let (content, signature) = - serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64)?; - let request_body = serde_json::to_string(&json!({ - "jsonrpc": "2.0", - "id": 1, - "method": "sendTransaction", - "params": [ - content, - { "encoding": "base64", "skipPreflight": true } - ] - }))?; + // Binary-Tx: Send raw binary transaction bytes directly + // This is faster than JSON-RPC as it avoids unnecessary encoding/decoding + let tx_bytes = bincode::serialize(transaction)?; + // Build URL for Binary-Tx endpoint: {endpoint}/txb?api-key={auth_token} let mut url = String::with_capacity(self.endpoint.len() + self.auth_token.len() + 20); url.push_str(&self.endpoint); - url.push_str("/?api-key="); + url.push_str("/txb?api-key="); url.push_str(&self.auth_token); - // 4. Use `text().await?` directly, avoiding async JSON parsing from `json().await?` - let response_text = self + // Send binary transaction directly + let response = self .http_client .post(&url) - .body(request_body) // Pass string directly, avoiding `json()` overhead - .header("Content-Type", "application/json") // Explicitly specify JSON header + .header("User-Agent", "") // Optional: 0slot recommends empty User-Agent + .body(tx_bytes) .send() - .await? - .text() .await?; - // 5. Use `serde_json::from_str()` to parse JSON, reducing extra wait from `.json().await?` - if let Ok(response_json) = serde_json::from_str::(&response_text) { - if response_json.get("result").is_some() { - crate::common::sdk_log::log_swqos_submitted("0slot", trade_type, start_time.elapsed()); - } else if let Some(_error) = response_json.get("error") { - eprintln!(" [0slot] {} submission failed after {:?}: {:?}", trade_type, start_time.elapsed(), _error); + let status = response.status(); + let response_text = response.text().await?; + + // Binary-Tx returns plain text responses with specific status codes + // 200: ok - transaction submitted successfully + // 403: api-key error (null, doesn't exist, or expired) + // 419: rate limit exceeded + // 500: submission failed + match status.as_u16() { + 200 => { + if response_text.trim() == "ok" { + crate::common::sdk_log::log_swqos_submitted("0slot", trade_type, start_time.elapsed()); + } else { + crate::common::sdk_log::log_swqos_submission_failed("0slot", trade_type, start_time.elapsed(), format!("unexpected response: {}", response_text)); + return Err(anyhow::anyhow!("0slot Binary-Tx unexpected response: {}", response_text)); + } + } + 403 => { + crate::common::sdk_log::log_swqos_submission_failed("0slot", trade_type, start_time.elapsed(), response_text.clone()); + return Err(anyhow::anyhow!("0slot API key error: {}", response_text)); + } + 419 => { + crate::common::sdk_log::log_swqos_submission_failed("0slot", trade_type, start_time.elapsed(), response_text.clone()); + return Err(anyhow::anyhow!("0slot rate limit exceeded")); + } + 500 => { + crate::common::sdk_log::log_swqos_submission_failed("0slot", trade_type, start_time.elapsed(), "submission failed".to_string()); + return Err(anyhow::anyhow!("0slot transaction submission failed")); + } + _ => { + crate::common::sdk_log::log_swqos_submission_failed("0slot", trade_type, start_time.elapsed(), format!("status {} body: {}", status, response_text)); + return Err(anyhow::anyhow!("0slot Binary-Tx failed with status {}: {}", status, response_text)); } - } else { - crate::common::sdk_log::log_swqos_submission_failed("0slot", trade_type, start_time.elapsed(), response_text); } - let start_time: Instant = Instant::now(); + // Get transaction signature from the transaction for confirmation polling + let signature = transaction.signatures[0]; + + let start_time = Instant::now(); match poll_transaction_confirmation(&self.rpc_client, signature, wait_confirmation).await { Ok(_) => (), Err(e) => { @@ -139,3 +231,21 @@ impl ZeroSlotClient { Ok(()) } } + +impl Drop for ZeroSlotClient { + 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(); + } + *ping_guard = None; + }); + } +}