From d7b09858444ac24e99f2cc3fa11181afb9f29de0 Mon Sep 17 00:00:00 2001 From: Wood Date: Thu, 12 Mar 2026 15:40:56 +0800 Subject: [PATCH] Release v3.6.2: Node1 QUIC, Speedlanding reliability, version bump - Add Node1 QUIC transport (UUID auth, bincode tx, region endpoints) - Speedlanding: ensure_connected, connect/send timeouts, lock on reconnect, SNI from endpoint host - Bump version to 3.6.2; update README(s) and add release notes Made-with: Cursor --- Cargo.toml | 3 +- README.md | 4 +- README_CN.md | 4 +- RELEASE_NOTES_v3.6.2.md | 16 ++ examples/trading_client/src/main.rs | 2 +- src/constants/swqos.rs | 13 ++ src/swqos/mod.rs | 38 +++- src/swqos/node1_quic.rs | 316 ++++++++++++++++++++++++++++ src/swqos/speedlanding.rs | 87 ++++++-- 9 files changed, 445 insertions(+), 38 deletions(-) create mode 100644 RELEASE_NOTES_v3.6.2.md create mode 100644 src/swqos/node1_quic.rs diff --git a/Cargo.toml b/Cargo.toml index d5bed05..b189a9e 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "sol-trade-sdk" -version = "3.6.1" +version = "3.6.2" edition = "2021" authors = [ "William ", @@ -109,6 +109,7 @@ sha2 = "0.10" tonic-prost = "0.14.2" quinn = { version = "0.11", default-features = false, features = ["rustls"] } rcgen = "0.13" +uuid = "1.11" # Performance optimization dependencies crossbeam-queue = "0.3" diff --git a/README.md b/README.md index da9ead9..fb90f1e 100644 --- a/README.md +++ b/README.md @@ -90,14 +90,14 @@ Add the dependency to your `Cargo.toml`: ```toml # Add to your Cargo.toml -sol-trade-sdk = { path = "./sol-trade-sdk", version = "3.6.1" } +sol-trade-sdk = { path = "./sol-trade-sdk", version = "3.6.2" } ``` ### Use crates.io ```toml # Add to your Cargo.toml -sol-trade-sdk = "3.6.1" +sol-trade-sdk = "3.6.2" ``` ## 🛠️ Usage Examples diff --git a/README_CN.md b/README_CN.md index e1039d2..bc96bc5 100755 --- a/README_CN.md +++ b/README_CN.md @@ -90,14 +90,14 @@ git clone https://github.com/0xfnzero/sol-trade-sdk ```toml # 添加到您的 Cargo.toml -sol-trade-sdk = { path = "./sol-trade-sdk", version = "3.5.7" } +sol-trade-sdk = { path = "./sol-trade-sdk", version = "3.6.2" } ``` ### 使用 crates.io ```toml # 添加到您的 Cargo.toml -sol-trade-sdk = "3.5.7" +sol-trade-sdk = "3.6.2" ``` ## 🛠️ 使用示例 diff --git a/RELEASE_NOTES_v3.6.2.md b/RELEASE_NOTES_v3.6.2.md new file mode 100644 index 0000000..ae2c767 --- /dev/null +++ b/RELEASE_NOTES_v3.6.2.md @@ -0,0 +1,16 @@ +## sol-trade-sdk v3.6.2 + +### Changes + +- **Node1 QUIC support** — Node1 SWQOS can use QUIC transport: `SwqosConfig::Node1(api_token, region, custom_url, Some(SwqosTransport::Quic))`. Uses UUID auth on first bi stream, one bi stream per transaction, bincode-serialized `VersionedTransaction`. Region endpoints: `SWQOS_ENDPOINTS_NODE1_QUIC` (ny, fra, ams, lon, tk). New dependency: `uuid`. +- **Speedlanding QUIC reliability** — Proactive connection check before send (`ensure_connected()`); 5s connect and send timeouts; reconnect uses `lock().await` so concurrent senders wait for the new connection instead of failing on `try_lock()`; TLS SNI is derived from the endpoint host (e.g. `nyc.speedlanding.trade`) with fallback to `speed-landing` for IP or unknown host. Addresses user reports of transactions failing to send. + +### Crates.io + +```toml +sol-trade-sdk = "3.6.2" +``` + +### Repository + +- **Tag:** [v3.6.2](https://github.com/0xfnzero/sol-trade-sdk/releases/tag/v3.6.2) diff --git a/examples/trading_client/src/main.rs b/examples/trading_client/src/main.rs index 77e3e03..c840562 100644 --- a/examples/trading_client/src/main.rs +++ b/examples/trading_client/src/main.rs @@ -46,7 +46,7 @@ async fn create_trading_client_simple() -> AnyResult { SwqosConfig::ZeroSlot("your_api_token".to_string(), SwqosRegion::Frankfurt, None), SwqosConfig::Temporal("your_api_token".to_string(), SwqosRegion::Frankfurt, None), SwqosConfig::FlashBlock("your_api_token".to_string(), SwqosRegion::Frankfurt, None), - SwqosConfig::Node1("your_api_token".to_string(), SwqosRegion::Frankfurt, None), + SwqosConfig::Node1("your_api_token".to_string(), SwqosRegion::Frankfurt, None, None), SwqosConfig::BlockRazor("your_api_token".to_string(), SwqosRegion::Frankfurt, None), SwqosConfig::Astralane("your_api_token".to_string(), SwqosRegion::Frankfurt, None, Some(SwqosTransport::Quic)), // QUIC; use None for HTTP // Helius Sender: 4th param swqos_only Some(true) => min tip 0.000005 SOL; None => 0.0002 SOL diff --git a/src/constants/swqos.rs b/src/constants/swqos.rs index 3499355..40d08ee 100755 --- a/src/constants/swqos.rs +++ b/src/constants/swqos.rs @@ -230,6 +230,19 @@ pub const SWQOS_ENDPOINTS_NODE1: [&str; 8] = [ "http://fra.node1.me", ]; +/// Node1 QUIC: port 16666. Region order: NewYork, Frankfurt, Amsterdam, SLC→ny, Tokyo, London, LosAngeles→ny, Default→ny. +/// server_name = host part (e.g. ny.node1.me). Auth: first bi stream = 16-byte UUID; each tx = new bi stream, bincode body. +pub const SWQOS_ENDPOINTS_NODE1_QUIC: [&str; 8] = [ + "ny.node1.me:16666", + "fra.node1.me:16666", + "ams.node1.me:16666", + "ny.node1.me:16666", // SLC → ny + "tk.node1.me:16666", + "lon.node1.me:16666", + "ny.node1.me:16666", // LA → ny + "ny.node1.me:16666", // Default → ny +]; + pub const SWQOS_ENDPOINTS_FLASHBLOCK: [&str; 8] = [ "http://ny.flashblock.trade", "http://fra.flashblock.trade", diff --git a/src/swqos/mod.rs b/src/swqos/mod.rs index e151e49..05b8306 100755 --- a/src/swqos/mod.rs +++ b/src/swqos/mod.rs @@ -8,6 +8,7 @@ pub mod zeroslot; pub mod temporal; pub mod bloxroute; pub mod node1; +pub mod node1_quic; pub mod flashblock; pub mod blockrazor; pub mod astralane; @@ -34,6 +35,7 @@ use crate::{ SWQOS_ENDPOINTS_TEMPORAL, SWQOS_ENDPOINTS_ZERO_SLOT, SWQOS_ENDPOINTS_NODE1, + SWQOS_ENDPOINTS_NODE1_QUIC, SWQOS_ENDPOINTS_FLASHBLOCK, SWQOS_ENDPOINTS_BLOCKRAZOR, SWQOS_ENDPOINTS_ASTRALANE, @@ -66,6 +68,7 @@ use crate::{ temporal::TemporalClient, zeroslot::ZeroSlotClient, node1::Node1Client, + node1_quic::Node1QuicClient, flashblock::FlashBlockClient, blockrazor::BlockRazorClient, astralane::AstralaneClient, @@ -215,8 +218,8 @@ pub enum SwqosConfig { Temporal(String, SwqosRegion, Option), /// ZeroSlot(api_token, region, custom_url) ZeroSlot(String, SwqosRegion, Option), - /// Node1(api_token, region, custom_url) - Node1(String, SwqosRegion, Option), + /// Node1(api_token, region, custom_url, transport). transport=None => HTTP; Some(Quic) => QUIC (port 16666, UUID auth). + Node1(String, SwqosRegion, Option, Option), /// FlashBlock(api_token, region, custom_url) FlashBlock(String, SwqosRegion, Option), /// BlockRazor(api_token, region, custom_url) @@ -248,7 +251,7 @@ impl SwqosConfig { SwqosConfig::Bloxroute(_, _, _) => SwqosType::Bloxroute, SwqosConfig::Temporal(_, _, _) => SwqosType::Temporal, SwqosConfig::ZeroSlot(_, _, _) => SwqosType::ZeroSlot, - SwqosConfig::Node1(_, _, _) => SwqosType::Node1, + SwqosConfig::Node1(_, _, _, _) => SwqosType::Node1, SwqosConfig::FlashBlock(_, _, _) => SwqosType::FlashBlock, SwqosConfig::BlockRazor(_, _, _) => SwqosType::BlockRazor, SwqosConfig::Astralane(_, _, _, _) => SwqosType::Astralane, @@ -336,14 +339,27 @@ impl SwqosConfig { ); Ok(Arc::new(bloxroute_client)) }, - SwqosConfig::Node1(auth_token, region, url) => { - let endpoint = SwqosConfig::get_endpoint(SwqosType::Node1, region, url); - let node1_client = Node1Client::new( - rpc_url.clone(), - endpoint.to_string(), - auth_token - ); - Ok(Arc::new(node1_client)) + SwqosConfig::Node1(auth_token, region, url, transport) => { + let use_quic = transport.map_or(false, |t| t == SwqosTransport::Quic); + if use_quic { + let quic_endpoint = url + .unwrap_or_else(|| SWQOS_ENDPOINTS_NODE1_QUIC[region as usize].to_string()); + let node1_quic = Node1QuicClient::connect( + &quic_endpoint, + &auth_token, + rpc_url.clone(), + ) + .await?; + Ok(Arc::new(node1_quic)) + } else { + let endpoint = SwqosConfig::get_endpoint(SwqosType::Node1, region, url); + let node1_client = Node1Client::new( + rpc_url.clone(), + endpoint.to_string(), + auth_token, + ); + Ok(Arc::new(node1_client)) + } }, SwqosConfig::FlashBlock(auth_token, region, url) => { let endpoint = SwqosConfig::get_endpoint(SwqosType::FlashBlock, region, url); diff --git a/src/swqos/node1_quic.rs b/src/swqos/node1_quic.rs new file mode 100644 index 0000000..45f78bf --- /dev/null +++ b/src/swqos/node1_quic.rs @@ -0,0 +1,316 @@ +//! Node1 QUIC SWQOS client. +//! +//! Protocol: first bi stream = auth (16-byte UUID); each transaction uses a new bi stream. +//! Request body = bincode(VersionedTransaction); response = 2 bytes status (BE) + 4 bytes msg_len (BE) + msg. +//! Reuses a single authenticated connection; reconnects and re-auth when connection is closed. + +use anyhow::{Context, Result}; +use quinn::crypto::rustls::QuicClientConfig; +use quinn::{ClientConfig, Connection, Endpoint, IdleTimeout, RecvStream, TransportConfig}; +use std::net::ToSocketAddrs; +use std::sync::Arc; +use std::time::Duration; +use tokio::time::timeout; +use tokio::sync::Mutex; +use uuid::Uuid; + +use crate::common::SolanaRpcClient; +use crate::constants::swqos::NODE1_TIP_ACCOUNTS; +use crate::swqos::common::poll_transaction_confirmation; +use crate::swqos::{SwqosClientTrait, SwqosType, TradeType}; +use rand::seq::IndexedRandom; +use solana_sdk::transaction::VersionedTransaction; +use std::time::Instant; + +const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); +const AUTH_TIMEOUT: Duration = Duration::from_secs(5); +const SEND_TIMEOUT: Duration = Duration::from_secs(5); +const KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(15); +const MAX_IDLE_TIMEOUT: Duration = Duration::from_secs(60); +const MAX_TX_SIZE: usize = 1232; + +/// Node1 QUIC client: one authenticated connection, reuse for all transactions. +pub struct Node1QuicClient { + endpoint: Endpoint, + connection: Mutex, + server_addr: String, + server_name: String, + api_key_uuid: [u8; 16], + rpc_client: Arc, +} + +impl Node1QuicClient { + /// Connect and authenticate. Reuse the returned client for all subsequent sends. + pub async fn connect( + server_addr: &str, + api_key: &str, + rpc_url: String, + ) -> Result { + let socket_addr = server_addr + .to_socket_addrs() + .context("resolve Node1 QUIC server address")? + .next() + .context("no socket address for Node1 QUIC")?; + + let api_key_uuid = Uuid::parse_str(api_key).context("Node1 API key must be a valid UUID")?; + let api_key_bytes: [u8; 16] = *api_key_uuid.as_bytes(); + + let server_name = server_addr + .split(':') + .next() + .unwrap_or(server_addr); + + let client_config = Self::build_client_config()?; + let mut endpoint = Endpoint::client("0.0.0.0:0".parse()?) + .context("create QUIC endpoint")?; + endpoint.set_default_client_config(client_config); + + let connecting = endpoint + .connect(socket_addr, server_name) + .context("Node1 QUIC connect failed")?; + let connection = timeout(CONNECT_TIMEOUT, connecting) + .await + .context("Node1 QUIC connect timeout")? + .context("Node1 QUIC handshake failed")?; + + timeout( + AUTH_TIMEOUT, + Self::authenticate(&connection, &api_key_bytes), + ) + .await + .context("Node1 QUIC auth timeout")??; + + Ok(Self { + endpoint, + connection: Mutex::new(connection), + server_addr: server_addr.to_string(), + server_name: server_name.to_string(), + api_key_uuid: api_key_bytes, + rpc_client: Arc::new(SolanaRpcClient::new(rpc_url)), + }) + } + + fn build_client_config() -> Result { + let crypto = rustls::ClientConfig::builder() + .dangerous() + .with_custom_certificate_verifier(Arc::new(SkipServerVerification)) + .with_no_client_auth(); + + let client_crypto = QuicClientConfig::try_from(crypto) + .context("build QUIC TLS config")?; + let mut client_config = ClientConfig::new(Arc::new(client_crypto)); + + let mut transport = TransportConfig::default(); + transport.max_idle_timeout(Some(IdleTimeout::try_from(MAX_IDLE_TIMEOUT).unwrap())); + transport.keep_alive_interval(Some(KEEP_ALIVE_INTERVAL)); + client_config.transport_config(Arc::new(transport)); + + Ok(client_config) + } + + async fn authenticate(connection: &Connection, api_key_bytes: &[u8; 16]) -> Result<()> { + let (mut send, mut recv) = connection.open_bi().await.context("open_bi for auth")?; + send.write_all(api_key_bytes).await.context("write auth bytes")?; + send.finish().context("finish auth stream")?; + + let mut reply = [0u8; 1]; + recv.read_exact(&mut reply).await.context("read auth reply")?; + match reply[0] { + 0 => Ok(()), + code => anyhow::bail!("Node1 QUIC auth rejected, reply={}", code), + } + } + + async fn ensure_connected(&self) -> Result { + let guard = self.connection.lock().await; + if let Some(_reason) = guard.close_reason() { + drop(guard); + let socket_addr = self + .server_addr + .to_socket_addrs() + .context("resolve Node1 QUIC server address")? + .next() + .context("no socket address")?; + let connecting = self + .endpoint + .connect(socket_addr, &self.server_name) + .context("Node1 QUIC reconnect failed")?; + let connection = timeout(CONNECT_TIMEOUT, connecting) + .await + .context("Node1 QUIC reconnect timeout")? + .context("Node1 QUIC re-handshake failed")?; + + timeout( + AUTH_TIMEOUT, + Self::authenticate(&connection, &self.api_key_uuid), + ) + .await + .context("Node1 QUIC re-auth timeout")??; + + let mut g = self.connection.lock().await; + *g = connection.clone(); + Ok(connection) + } else { + Ok(guard.clone()) + } + } + + async fn read_response(recv: &mut RecvStream) -> Result<(u16, String)> { + let mut header = [0u8; 6]; + recv.read_exact(&mut header) + .await + .map_err(|e| anyhow::anyhow!("read response header: {:?}", e))?; + let status = u16::from_be_bytes(header[0..2].try_into().unwrap()); + let msg_len = u32::from_be_bytes(header[2..6].try_into().unwrap()) as usize; + let mut msg = vec![0u8; msg_len]; + if msg_len > 0 { + recv.read_exact(&mut msg) + .await + .map_err(|e| anyhow::anyhow!("read response body: {:?}", e))?; + } + Ok((status, String::from_utf8_lossy(&msg).into_owned())) + } + + /// Send one transaction over QUIC (opens new bi stream, writes bincode tx, reads status+msg). + pub async fn send_transaction_bytes(&self, tx_bytes: &[u8]) -> Result<(u16, String)> { + if tx_bytes.len() > MAX_TX_SIZE { + anyhow::bail!( + "Node1 QUIC: transaction too large ({} > {})", + tx_bytes.len(), + MAX_TX_SIZE + ); + } + + let conn = self.ensure_connected().await?; + let (mut send, mut recv) = conn.open_bi().await.context("open_bi for tx")?; + send.write_all(tx_bytes).await.context("write tx")?; + send.finish().context("finish tx stream")?; + Self::read_response(&mut recv).await + } +} + +#[async_trait::async_trait] +impl SwqosClientTrait for Node1QuicClient { + async fn send_transaction( + &self, + trade_type: TradeType, + transaction: &VersionedTransaction, + wait_confirmation: bool, + ) -> Result<()> { + let start = Instant::now(); + let signature = transaction.signatures.first().copied().unwrap_or_default(); + let tx_bytes = bincode::serialize(transaction).context("Node1 QUIC: bincode serialize")?; + + let (status, msg) = timeout( + SEND_TIMEOUT, + self.send_transaction_bytes(&tx_bytes), + ) + .await + .context("Node1 QUIC send timeout")??; + + if status != 200 { + if crate::common::sdk_log::sdk_log_enabled() { + eprintln!(" [node1-quic] {} submit failed: status={} msg={}", trade_type, status, msg); + } + anyhow::bail!("Node1 QUIC submit failed: status={} msg={}", status, msg); + } + + if crate::common::sdk_log::sdk_log_enabled() { + println!(" [node1-quic] {} submitted: {:?}", trade_type, start.elapsed()); + } + + let start = Instant::now(); + match poll_transaction_confirmation(&self.rpc_client, signature, wait_confirmation).await { + Ok(_) => { + if wait_confirmation && crate::common::sdk_log::sdk_log_enabled() { + println!(" [node1-quic] {} confirmed: {:?}", trade_type, start.elapsed()); + } + Ok(()) + } + Err(e) => { + if crate::common::sdk_log::sdk_log_enabled() { + eprintln!(" [node1-quic] {} confirmation failed: {:?}", trade_type, start.elapsed()); + } + Err(e) + } + } + } + + async fn send_transactions( + &self, + trade_type: TradeType, + transactions: &Vec, + wait_confirmation: bool, + ) -> Result<()> { + for tx in transactions { + self.send_transaction(trade_type, tx, wait_confirmation).await?; + } + Ok(()) + } + + fn get_tip_account(&self) -> Result { + let tip = *NODE1_TIP_ACCOUNTS + .choose(&mut rand::rng()) + .or_else(|| NODE1_TIP_ACCOUNTS.first()) + .unwrap(); + Ok(tip.to_string()) + } + + fn get_swqos_type(&self) -> SwqosType { + SwqosType::Node1 + } +} + +impl Drop for Node1QuicClient { + fn drop(&mut self) { + self.connection.get_mut().close(0u32.into(), b"client closing"); + } +} + +#[derive(Debug)] +struct SkipServerVerification; + +impl rustls::client::danger::ServerCertVerifier for SkipServerVerification { + fn verify_server_cert( + &self, + _: &rustls::pki_types::CertificateDer<'_>, + _: &[rustls::pki_types::CertificateDer<'_>], + _: &rustls::pki_types::ServerName<'_>, + _: &[u8], + _: rustls::pki_types::UnixTime, + ) -> Result { + Ok(rustls::client::danger::ServerCertVerified::assertion()) + } + + fn verify_tls12_signature( + &self, + _: &[u8], + _: &rustls::pki_types::CertificateDer<'_>, + _: &rustls::DigitallySignedStruct, + ) -> Result { + Ok(rustls::client::danger::HandshakeSignatureValid::assertion()) + } + + fn verify_tls13_signature( + &self, + _: &[u8], + _: &rustls::pki_types::CertificateDer<'_>, + _: &rustls::DigitallySignedStruct, + ) -> Result { + Ok(rustls::client::danger::HandshakeSignatureValid::assertion()) + } + + fn supported_verify_schemes(&self) -> Vec { + vec![ + rustls::SignatureScheme::ECDSA_NISTP256_SHA256, + rustls::SignatureScheme::ECDSA_NISTP384_SHA384, + rustls::SignatureScheme::RSA_PSS_SHA256, + rustls::SignatureScheme::RSA_PSS_SHA384, + rustls::SignatureScheme::RSA_PSS_SHA512, + rustls::SignatureScheme::RSA_PKCS1_SHA256, + rustls::SignatureScheme::RSA_PKCS1_SHA384, + rustls::SignatureScheme::RSA_PKCS1_SHA512, + rustls::SignatureScheme::ED25519, + ] + } +} diff --git a/src/swqos/speedlanding.rs b/src/swqos/speedlanding.rs index a791dc2..1a52a36 100644 --- a/src/swqos/speedlanding.rs +++ b/src/swqos/speedlanding.rs @@ -15,6 +15,7 @@ use std::{ time::Duration, }; use tokio::sync::Mutex; +use tokio::time::timeout; use crate::common::SolanaRpcClient; use crate::swqos::common::poll_transaction_confirmation; @@ -26,22 +27,44 @@ use crate::{ }; const ALPN_TPU_PROTOCOL_ID: &[u8] = b"solana-tpu"; -const SPEED_SERVER: &str = "speed-landing"; +/// Fallback SNI when endpoint is IP or cannot extract host (keeps legacy behavior). +const SPEED_SERVER_FALLBACK: &str = "speed-landing"; const KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(25); const MAX_IDLE_TIMEOUT: Duration = Duration::from_secs(5 * 60); +const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); +const SEND_TIMEOUT: Duration = Duration::from_secs(5); pub struct SpeedlandingClient { pub rpc_client: Arc, endpoint: Endpoint, client_config: ClientConfig, addr: SocketAddr, + /// TLS SNI: host from endpoint URL so server presents the right cert (e.g. nyc.speedlanding.trade). + server_name: String, connection: ArcSwap, reconnect: Mutex<()>, } impl SpeedlandingClient { + /// Extract TLS SNI (host) from endpoint URL. Uses fallback "speed-landing" for IP or when host cannot be determined. + fn server_name_from_endpoint(endpoint: &str) -> String { + let without_scheme = endpoint + .strip_prefix("https://") + .or_else(|| endpoint.strip_prefix("http://")) + .unwrap_or(endpoint); + let host = without_scheme.split(':').next().unwrap_or("").trim(); + if host.is_empty() { + return SPEED_SERVER_FALLBACK.to_string(); + } + if !host.chars().any(|c| c.is_ascii_alphabetic()) { + return SPEED_SERVER_FALLBACK.to_string(); + } + host.to_string() + } + pub async fn new(rpc_url: String, endpoint_string: String, api_key: String) -> Result { let rpc_client = SolanaRpcClient::new(rpc_url); + let server_name = Self::server_name_from_endpoint(&endpoint_string); let keypair = Keypair::from_base58_string(&api_key); let (cert, key) = new_dummy_x509_certificate(&keypair); let mut crypto = rustls::ClientConfig::builder() @@ -66,26 +89,47 @@ impl SpeedlandingClient { .to_socket_addrs()? .next() .ok_or_else(|| anyhow::anyhow!("Address not resolved"))?; - let connection = endpoint.connect(addr, SPEED_SERVER)?.await?; + let connecting = endpoint.connect(addr, &server_name)?; + let connection = timeout(CONNECT_TIMEOUT, connecting) + .await + .context("Speedlanding QUIC connect timeout")? + .context("Speedlanding QUIC handshake failed")?; Ok(Self { rpc_client: Arc::new(rpc_client), endpoint, client_config, addr, + server_name, connection: ArcSwap::from_pointee(connection), reconnect: Mutex::new(()), }) } - async fn reconnect(&self) -> Result<()> { - let _guard = self.reconnect.try_lock()?; - let connection = self - .endpoint - .connect_with(self.client_config.clone(), self.addr, SPEED_SERVER)? - .await?; - self.connection.store(Arc::new(connection)); - Ok(()) + /// Ensure we have a live connection: if current one is closed, reconnect under lock so + /// concurrent senders wait and then all use the new connection. Uses blocking lock so + /// waiters get the updated connection. + async fn ensure_connected(&self) -> Result> { + let guard = self.reconnect.lock().await; + let current = self.connection.load_full(); + if current.close_reason().is_none() { + return Ok(current); + } + drop(guard); + let _guard = self.reconnect.lock().await; + let current = self.connection.load_full(); + if current.close_reason().is_some() { + let connecting = self + .endpoint + .connect_with(self.client_config.clone(), self.addr, self.server_name.as_str())?; + let connection = timeout(CONNECT_TIMEOUT, connecting) + .await + .context("Speedlanding QUIC reconnect timeout")? + .context("Speedlanding QUIC re-handshake failed")?; + self.connection.store(Arc::new(connection)); + return Ok(self.connection.load_full()); + } + Ok(current) } async fn try_send_bytes(connection: &Connection, payload: &[u8]) -> Result<()> { @@ -106,20 +150,21 @@ impl SwqosClientTrait for SpeedlandingClient { ) -> Result<()> { let start_time = Instant::now(); let (buf_guard, signature) = serialize_transaction_bincode_sync(transaction)?; - let connection = self.connection.load_full(); - if Self::try_send_bytes(&connection, &*buf_guard).await.is_err() { + let connection = self.ensure_connected().await?; + let mut send_result = timeout(SEND_TIMEOUT, Self::try_send_bytes(&connection, &*buf_guard)).await; + let need_retry = match &send_result { + Ok(Ok(())) => false, + Ok(Err(_)) | Err(_) => true, + }; + if need_retry { if crate::common::sdk_log::sdk_log_enabled() { - eprintln!(" [speedlanding] {} submission failed, reconnecting", trade_type); - } - self.reconnect().await?; - let connection = self.connection.load_full(); - if let Err(e) = Self::try_send_bytes(&connection, &*buf_guard).await { - if crate::common::sdk_log::sdk_log_enabled() { - eprintln!(" [speedlanding] {} submission failed: {:?}", trade_type, e); - } - return Err(e.into()); + eprintln!(" [speedlanding] {} send failed or timeout, reconnecting", trade_type); } + let connection = self.ensure_connected().await?; + send_result = timeout(SEND_TIMEOUT, Self::try_send_bytes(&connection, &*buf_guard)).await; } + send_result + .context("Speedlanding QUIC send timeout")??; match poll_transaction_confirmation(&self.rpc_client, signature, wait_confirmation).await { Ok(_) => (), Err(e) => {