Sending transaction via QUIC (ConnectionError(TimedOut))

Question

I am developing a Rust application that sends transactions on the Solana blockchain using QUIC. My code works perfectly on the devnet, but when I switch to the mainnet, the transaction sending fails with a ConnectionError(TimedOut) error.

The most interesting thing is that sometimes the transaction is successful.

Cargo.toml

[package]
name = "quic"
version = "0.1.0"
edition = "2021"

[dependencies]
solana-sdk = "1.18.13"
solana-client = "1.18.13"
solana-quic-client = "1.18.13"
solana-connection-cache = "1.18.13"
solana-transaction-status = "1.18.13"

tokio = {version = "1.37.0", features = ["full"]}
bincode = "1.3.3"
bs58 = "0.5.1"
clap = { version = "4.5.4", features = ["derive"] }

dashmap = "5.5.3"
futures-util = "0.3.30"
indexmap = "2.2.6"
serde_json = "1.0.117"
tokio-tungstenite = { version = "0.21.0", features = ["native-tls"] }
tracing = "0.1.40"
tracing-subscriber = "0.3.18"

Here is my code:

main.rs

use std::sync::Arc;
use tokio::time::{sleep, Duration};

use crate::leader_tracker::LeaderTracker;
use solana_client::nonblocking::rpc_client::RpcClient;

use tracing::{error, info, Level};
use tracing_subscriber::FmtSubscriber;

use self::{
    config::{Config, Network},
    leader_tracker::LeaderTrackerImpl,
    quic_manager::QuicManager,
};

mod config;
mod leader_tracker;
mod quic_manager;

use clap::{ArgGroup, Parser};

#[derive(Debug, Parser)]
#[command(name = "Solana Transaction")]
#[command(group(
    ArgGroup::new("network")
        .required(true)
        .args(&["mainnet", "devnet"]),
))]
pub struct Cli {
    #[arg(long)]
    pub mainnet: bool,
    #[arg(long)]
    pub devnet: bool,
    #[arg(long, default_value_t = 1)]
    pub retry: u8,
}

#[tokio::main]
async fn main() {
    // Initialize the tracing subscriber for logging
    let subscriber = FmtSubscriber::builder()
        .with_max_level(Level::INFO)
        .finish();
    tracing::subscriber::set_global_default(subscriber).expect("setting default subscriber failed");

    let cli = Cli::parse();

    let network = if cli.mainnet {
        Network::Mainnet
    } else {
        Network::Devnet
    };

    let config = Config::new(network, cli.retry);

    let rpc_client = Arc::new(RpcClient::new_with_commitment(
        config.rpc_url.clone(),
        config.commitment_level,
    ));
    info!("CONFIG {:#?}", config);

    let tracker =
        Arc::new(LeaderTrackerImpl::new(rpc_client.clone(), 4, 0, config.ws_url.clone()).await);
    tracker.poll_slot_leaders_once().await.unwrap();

    let mut attempts = 0;
    while attempts < config.retry {
        let leaders = tracker.get_leaders();

        if let Some(leader) = leaders.last() {
            info!("LEADER: {:#?}", leader);
            // берем первого лидера из списка с учетом смещения
            if let Some(tpu_quic) = &leader.tpu_quic {
                let manager = QuicManager::new(rpc_client.clone(), *tpu_quic).await;
                info!("QUIC: {:#?}", tpu_quic);

                match manager.send_transaction(&config).await {
                    Ok(signature) => {
                        info!("Transaction sent. Confirmation...");
                        match manager.check_confirm_transaction(&signature).await {
                            Ok(_) => {
                                info!("Transaction confirmed successfully.");
                                let full_url = config.generate_url(&signature.to_string());
                                info!("{}", full_url);
                                break;
                            }
                            Err(e) => error!("Error confirming transaction: {:#?}", e),
                        }
                    }
                    Err(e) => error!("Error sending transaction: {:#?}", e),
                }
            } else {
                error!("No QUIC address available for the current leader.");
            }
        } else {
            error!("No current leader available. Searching...");
        }
        attempts += 1;
        sleep(Duration::from_secs(1)).await;
    }

    if attempts >= config.retry {
        info!("Maximum number of attempts reached, stopping the application.");
    }
}

leader_tracker.rs

use std::{
    collections::HashMap,
    sync::{
        atomic::{AtomicU64, Ordering},
        Arc,
    },
    time::Duration,
};

use dashmap::DashMap;
use futures_util::{SinkExt, StreamExt};
use indexmap::IndexMap;
use solana_client::{nonblocking::rpc_client::RpcClient, rpc_response::RpcContactInfo};
use solana_sdk::clock::Slot;
use tokio::time::sleep;
use tokio_tungstenite::tungstenite::protocol::Message;
use tracing::{error, info};

pub trait LeaderTracker: Send + Sync {
    /// get_leaders returns the next slot leaders in order
    fn get_leaders(&self) -> Vec<RpcContactInfo>;
}

const NUM_LEADERS_PER_SLOT: usize = 4;

#[derive(Clone)]
pub struct LeaderTrackerImpl {
    rpc_client: Arc<RpcClient>,
    cur_slot: Arc<AtomicU64>,
    cur_leaders: Arc<DashMap<Slot, RpcContactInfo>>,
    num_leaders: usize,
    leader_offset: i64,
}

impl LeaderTrackerImpl {
    pub async fn new(
        rpc_client: Arc<RpcClient>,
        num_leaders: usize,
        leader_offset: i64,
        ws_url: String,
    ) -> Self {
        let cur_slot = Arc::new(AtomicU64::new(0));

        let initial_slot = rpc_client.get_slot().await.unwrap_or(0);
        cur_slot.store(initial_slot, Ordering::Relaxed);

        let leader_tracker = Self {
            rpc_client,
            cur_slot,
            cur_leaders: Arc::new(DashMap::new()),
            num_leaders,
            leader_offset,
        };
        leader_tracker.start_websocket_listener(ws_url);
        leader_tracker.poll_slot_leaders();
        leader_tracker
    }

    /// Start WebSocket listener for slot updates
    fn start_websocket_listener(&self, ws_url: String) {
        let cur_slot = self.cur_slot.clone();
        tokio::spawn(async move {
            info!("Starting WebSocket listener...");
            let (ws_stream, _) = match tokio_tungstenite::connect_async(ws_url).await {
                Ok(stream) => stream,
                Err(e) => {
                    error!("Failed to connect: {}", e);
                    return;
                }
            };

            let (mut write, mut read) = ws_stream.split();

            // Subscribe to slot updates
            match write
                .send(Message::Text(
                    r#"{"jsonrpc":"2.0","id":1,"method":"slotSubscribe"}"#.to_string(),
                ))
                .await
            {
                Ok(_) => info!("WebSocket subscribed to slot updates"),
                Err(e) => {
                    error!("Failed to send subscribe message: {:#?}", e);
                    return;
                }
            };

            while let Some(Ok(message)) = read.next().await {
                if let Message::Text(text) = message {
                    if let Ok(response) = serde_json::from_str::<serde_json::Value>(&text) {
                        if let Some(slot) = response["params"]["result"]["slot"].as_u64() {
                            cur_slot.store(slot, Ordering::Relaxed);
                        }
                    }
                }
            }
        });
    }

    /// poll_slot_leaders polls every minute for the next 1000 slot leaders and populates the cur_leaders map with the slot and ContactInfo of each leader
    fn poll_slot_leaders(&self) {
        let self_clone = self.clone();
        tokio::spawn(async move {
            loop {
                let start = std::time::Instant::now();
                if let Err(e) = self_clone.poll_slot_leaders_once().await {
                    error!("Error polling slot leaders: {}", e);
                    sleep(Duration::from_secs(1)).await;
                    continue;
                }
                let duration = start.elapsed();
                info!("poll_slot_leaders took {:?}", duration);
                sleep(Duration::from_secs(60)).await;
            }
        });
    }

    pub async fn poll_slot_leaders_once(&self) -> Result<(), String> {
        let next_slot = self.cur_slot.load(Ordering::Relaxed);

        // polling 1000 slots ahead is more than enough
        let slot_leaders = self
            .rpc_client
            .get_slot_leaders(next_slot, 1000)
            .await
            .map_err(|e| format!("Error getting slot leaders: {}", e))?;

        let new_cluster_nodes = self
            .rpc_client
            .get_cluster_nodes()
            .await
            .map_err(|e| format!("Error getting cluster nodes: {}", e))?;

        let cluster_node_map: HashMap<_, _> = new_cluster_nodes
            .into_iter()
            .map(|node| (node.pubkey.clone(), node))
            .collect();

        for (i, leader) in slot_leaders.iter().enumerate() {
            if let Some(contact_info) = cluster_node_map.get(&leader.to_string()) {
                self.cur_leaders
                    .insert(next_slot + i as u64, contact_info.clone());
            } else {
                error!("Leader {} not found in cluster nodes", leader);
            }
        }

        self.clean_up_slot_leaders();
        Ok(())
    }

    fn clean_up_slot_leaders(&self) {
        let cur_slot = self.cur_slot.load(Ordering::Relaxed);
        let slots_to_remove: Vec<_> = self
            .cur_leaders
            .iter()
            .filter(|leader| *leader.key() < cur_slot)
            .map(|leader| *leader.key())
            .collect();

        for slot in slots_to_remove {
            self.cur_leaders.remove(&slot);
        }
    }
}

impl LeaderTracker for LeaderTrackerImpl {
    fn get_leaders(&self) -> Vec<RpcContactInfo> {
        let start_slot = self.cur_slot.load(Ordering::Relaxed) + self.leader_offset as u64;
        let end_slot = start_slot + (self.num_leaders * NUM_LEADERS_PER_SLOT) as u64;
        let mut leaders = IndexMap::new();

        for slot in start_slot..end_slot {
            if let Some(leader) = self.cur_leaders.get(&slot) {
                leaders.insert(leader.pubkey.clone(), leader.value().clone());
            }
            if leaders.len() >= self.num_leaders {
                break;
            }
        }

        info!(
            "leaders: {:#?}, start_slot: {:#?}",
            leaders.keys(),
            start_slot
        );

        leaders.values().cloned().collect()
    }
}

quic_manager.rs

use solana_client::{
    nonblocking::{
        quic_client::{QuicLazyInitializedEndpoint, QuicTpuConnection},
        rpc_client::RpcClient,
        tpu_connection::TpuConnection,
    },
    tpu_connection::ClientStats,
};
use solana_connection_cache::connection_cache_stats::ConnectionCacheStats;
use solana_sdk::signature::Signature;
use std::{net::SocketAddr, sync::Arc};
use tracing::{error, info};

use crate::config::Config;

pub struct QuicManager {
    pub connection: Arc<QuicTpuConnection>,
    pub stats: Arc<ClientStats>,
    pub rpc_client: Arc<RpcClient>,
}

impl QuicManager {
    pub async fn new(rpc_client: Arc<RpcClient>, socket_addr: SocketAddr) -> Self {
        let endpoint = Arc::new(QuicLazyInitializedEndpoint::default());
        let stats = Arc::new(ClientStats::default());
        let connection_stats = Arc::new(ConnectionCacheStats::default());

        let quic_tpu_connection = QuicTpuConnection::new(endpoint, socket_addr, connection_stats);

        QuicManager {
            connection: Arc::new(quic_tpu_connection),
            stats,
            rpc_client,
        }
    }

    pub async fn send_transaction(&self, config: &Config) -> Result<Signature, String> {
        let max_attempts = 3; // Увеличение числа попыток
        for attempt in 0..max_attempts {
            let blockhash = self
                .rpc_client
                .get_latest_blockhash()
                .await
                .map_err(|e| format!("Failed to get blockhash: {}", e))?;
            info!("[ BLOCKHASH ] - {:#?}", blockhash);

            let transaction = config.create_transaction(blockhash);

            info!(
            "[ TRANSACTION\n\tSENDER: {:?}\n\tRECEIVER: {:?}\n\tBLOCKHASH: {:?}\n\tSIGNATURE: {:?}\n]",
            transaction.message.account_keys[0],
            transaction.message.account_keys[1],
            transaction.message.recent_blockhash,
            transaction.signatures
        );

            let serialized_tx = bincode::serialize(&transaction).unwrap();

            let send_result = tokio::time::timeout(
                std::time::Duration::from_secs(60), // Увеличение таймаута до 60 секунд
                self.connection.send_data(&serialized_tx),
            )
            .await;

            match send_result {
                Ok(Ok(_)) => {
                    if let Some(signature) = transaction.signatures.first() {
                        return Ok(*signature);
                    } else {
                        return Err("No signature found in the transaction".to_string());
                    }
                }
                Ok(Err(e)) => {
                    error!(
                        "Attempt {}: Failed to send transaction via QUIC: {:#?}",
                        attempt + 1,
                        e
                    );
                    if attempt + 1 < max_attempts {
                        tokio::time::sleep(std::time::Duration::from_secs(2)).await;
                    }
                }
                Err(_) => {
                    error!(
                        "Attempt {}: Timed out while sending transaction via QUIC",
                        attempt + 1
                    );
                    if attempt + 1 < max_attempts {
                        tokio::time::sleep(std::time::Duration::from_secs(2)).await;
                    }
                }
            }
        }

        Err("Failed to send transaction via QUIC after multiple attempts".to_string())
    }

    pub async fn check_confirm_transaction(&self, signature: &Signature) -> Result<bool, String> {
        let transaction_with_meta = self
            .rpc_client
            .get_transaction(
                signature,
                solana_transaction_status::UiTransactionEncoding::Json,
            )
            .await;
        info!("META {:#?}", transaction_with_meta);

        let max_attempts = 10;
        for _ in 0..max_attempts {
            let statuses = self
                .rpc_client
                .get_signature_statuses(&[*signature])
                .await
                .map_err(|e| format!("Failed to get signature statuses: {}", e))?;

            if let Some(Some(status)) = statuses.value.first() {
                if status.confirmations.is_some() {
                    return Ok(true);
                } else {
                    info!("Transaction not confirmed yet, retrying...");
                }
            }

            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
        }

        Err("Transaction failed to confirm".to_string())
    }
}

config.rs

use std::str::FromStr;

use solana_sdk::{
    commitment_config::CommitmentConfig,
    compute_budget::ComputeBudgetInstruction,
    hash::Hash,
    pubkey::Pubkey,
    signature::{read_keypair_file, Keypair},
    signer::Signer,
    system_instruction::transfer,
    transaction::Transaction,
};

#[derive(Debug)]
pub enum Network {
    Mainnet,
    Devnet,
}

#[derive(Debug)]
pub struct Config {
    pub rpc_url: String,
    pub ws_url: String,
    pub sender_key: String,
    pub receiver_key: String,
    pub amount: u64,
    pub retry: u8,
    pub network: Network,
    pub commitment_level: CommitmentConfig,
}

impl Config {
    pub fn new(network: Network, retry: u8) -> Self {
        match network {
            Network::Mainnet => Self {
                rpc_url: String::from("https://api.mainnet-beta.solana.com"),
                ws_url: String::from("wss://api.mainnet-beta.solana.com"),
                sender_key: String::from("<sender_private_key>"),
                receiver_key: String::from("<receiver_public_key>"),
                amount: 1_000,
                retry,
                network: Network::Mainnet,
                commitment_level: CommitmentConfig::finalized()
            },
            Network::Devnet => Self {
                rpc_url: String::from("https://api.devnet.solana.com"),
                ws_url: String::from("wss://api.devnet.solana.com"),
                sender_key: String::from("/home/user/.config/solana/devnet.json"),
                receiver_key: String::from("/home/user/.config/solana/test-receiver.json"),
                amount: 1_000,
                retry,
                network: Network::Devnet,
                commitment_level: CommitmentConfig::finalized()
            },
        }
    }

    fn setup_sender(&self) -> Keypair {
        match self.network {
            Network::Mainnet => Keypair::from_base58_string(&self.sender_key),
            Network::Devnet => {
                read_keypair_file(&self.sender_key).expect("Unable to read keypair file")
            }
        }
    }

    fn setup_receiver(&self) -> Pubkey {
        match self.network {
            Network::Mainnet => Pubkey::from_str(&self.receiver_key).expect("Invalid pubkey"),
            Network::Devnet => read_keypair_file(&self.receiver_key)
                .expect("Failed to read receiver keypair from file")
                .pubkey(),
        }
    }

    pub fn create_transaction(&self, blockhash: Hash) -> Transaction {
        let sender = Config::setup_sender(self);
        let receiver = Config::setup_receiver(self);

        let compute_unit_limit_instruction =
            ComputeBudgetInstruction::set_compute_unit_limit(50_000);
        let compute_unit_price_instruction =
            ComputeBudgetInstruction::set_compute_unit_price(10000);

        let transfer_instruction = transfer(&sender.pubkey(), &receiver, self.amount);

        Transaction::new_signed_with_payer(
            &[
                compute_unit_limit_instruction,
                compute_unit_price_instruction,
                transfer_instruction,
            ],
            Some(&sender.pubkey()),
            &[&sender],
            blockhash,
        )
    }

    pub fn generate_url(&self, transaction_number: &str) -> String {
        let base_url = "https://explorer.solana.com/tx/";
        let cluster = match self.network {
            Network::Mainnet => "",
            Network::Devnet => "?cluster=devnet",
        };

        format!(
            "Check transaction {}{}{}",
            base_url, transaction_number, cluster
        )
    }
}

Here is output:

 Running `target/debug/quic --mainnet --retry 1`
2024-05-16T23:09:38.423426Z  INFO quic: CONFIG Config {
    rpc_url: "https://api.mainnet-beta.solana.com",
    ws_url: "wss://api.mainnet-beta.solana.com",
    sender_key: "<sender_private_key>",
    receiver_key: "<receiver_public_key>",
    amount: 1000,
    retry: 1,
    network: Mainnet,
    commitment_level: CommitmentConfig {
        commitment: Finalized,
    },
}
2024-05-16T23:09:38.669906Z  INFO quic::leader_tracker: Starting WebSocket listener...
2024-05-16T23:09:38.910773Z  INFO quic::leader_tracker: WebSocket subscribed to slot updates
2024-05-16T23:09:39.034071Z  INFO quic::leader_tracker: leaders: [
    "krakeNd6ednDPEXxHAmoBs1qKVM8kLg79PvWF2mhXV1",
    "FBKFWadXZJahGtFitAsBvbqh5968gLY7dMBBJUoUjeNi",
    "EgxVyTgh2Msg781wt9EsqYx4fW8wSvfFAHGLaJQjghiL",
    "CwyVpfmfSiMeCexi3JgUNvaiDfYN14cLDjzT99zcBuD2",
], start_slot: 266208226
2024-05-16T23:09:39.034111Z  INFO quic: LEADER: RpcContactInfo {
    pubkey: "CwyVpfmfSiMeCexi3JgUNvaiDfYN14cLDjzT99zcBuD2",
    gossip: Some(
        93.115.25.113:12000,
    ),
    tpu: Some(
        93.115.25.113:12003,
    ),
    tpu_quic: Some(
        93.115.25.113:12009,
    ),
    rpc: None,
    pubsub: None,
    version: Some(
        "1.17.33",
    ),
    feature_set: Some(
        3746964731,
    ),
    shred_version: Some(
        50093,
    ),
}
2024-05-16T23:09:39.034708Z  INFO quic: QUIC: 93.115.25.113:12009
2024-05-16T23:09:39.084071Z  INFO quic::leader_tracker: poll_slot_leaders took 414.067368ms
2024-05-16T23:09:39.126373Z  INFO quic::quic_manager: [ BLOCKHASH ] - 6F6wb44ZbMGAfWpbhdEJAdn7t52Ysi2fUU2ohDSSZJq8
2024-05-16T23:09:39.127037Z  INFO quic::quic_manager: [ TRANSACTION
    SENDER: <sender_key>
    RECEIVER: <receiver_key>
    BLOCKHASH: 6F6wb44ZbMGAfWpbhdEJAdn7t52Ysi2fUU2ohDSSZJq8
    SIGNATURE: [<signature>]
]
2024-05-16T23:09:42.127766Z ERROR quic::quic_manager: Attempt 1: Failed to send transaction via QUIC: Custom(
    "\"ConnectionError(TimedOut)\"",
)

Unfortunately, this is a result of congestion on mainnet. Without stake, you're allocated very few connections to the validators in the network, which means that you'll often get timeouts when trying to send transactions. — Jon C, Commented May 19 at 20:42
@JonC, the problem persists even when I use WSS and RPC addresses from Helios. Moreover, if transactions sometimes go through using the regular mainnet, they always time out when using Helios. Thank you for your response. By the way, do you know how this problem can be solved? — RAprogramm, Commented May 20 at 21:40
There are improvements in flight on the server side to better handle connection spam, but it will take some time for it to be rolled out. And if Helius isn't working, you may want to ask their support team — Jon C, Commented May 21 at 10:18

Stack Exchange Network

Sending transaction via QUIC (ConnectionError(TimedOut))

0

Your Answer

Browse other questions tagged
rpc
send-transaction
mainnet
solana-client
or ask your own question.

Hot Network Questions

Sending transaction via QUIC (ConnectionError(TimedOut))

0

Know someone who can answer? Share a link to this question via email, Twitter, or Facebook.

Your Answer

Sign up or log in

Post as a guest

Browse other questions tagged rpcsend-transactionmainnetsolana-client or ask your own question.

Related

Hot Network Questions

Browse other questions tagged
rpc
send-transaction
mainnet
solana-client
or ask your own question.