Writing a Solana QUIC TPU client from scratch

1use anyhow::Result;
2use rcgen::{CertificateParams, DistinguishedName, KeyPair as RcgenKeyPair, PKCS_ED25519};
3use rustls_pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
4use solana_keypair::Keypair;
5use solana_signer::Signer;
6 
7/// Build the self-signed certificate a TPU client presents.
8///
9/// This is the whole authentication story. There is no header and no token.
10/// The leader reads the Ed25519 public key out of this certificate during the
11/// handshake and looks it up against the current stake map, which is what
12/// decides your connection's QoS treatment before you have sent a byte.
13///
14/// The certificate is deliberately minimal: no SANs, no meaningful validity
15/// window, no chain. It is a carrier for a public key, not a trust statement.
16pub fn client_identity(identity: &Keypair) -> Result<(CertificateDer<'static>, PrivateKeyDer<'static>)> {
17    // rcgen wants the PKCS#8 encoding of the Ed25519 key.
18    let pkcs8 = identity.to_bytes();
19    let key_pair = RcgenKeyPair::from_pkcs8_der_and_sign_algo(
20        &PrivatePkcs8KeyDer::from(pkcs8.as_slice()),
21        &PKCS_ED25519,
22    )?;
23 
24    let mut params = CertificateParams::default();
25    params.distinguished_name = DistinguishedName::new();
26    // Solana's own client uses the identity pubkey as the common name. Nothing
27    // reads it, but matching the convention avoids surprises.
28    params
29        .distinguished_name
30        .push(rcgen::DnType::CommonName, identity.pubkey().to_string());
31 
32    let cert = params.self_signed(&key_pair)?;
33 
34    Ok((
35        cert.der().clone(),
36        PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(key_pair.serialize_der())),
37    ))
38}

1use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
2use rustls::pki_types::{CertificateDer, ServerName, UnixTime};
3use rustls::{DigitallySignedStruct, Error as TlsError, SignatureScheme};
4 
5/// A validator presents a self-signed certificate too: there is no CA in this
6/// system on either side. So the client cannot do conventional chain
7/// validation, and skipping it is the documented, intended behaviour rather
8/// than a shortcut.
9///
10/// This is worth being clear-eyed about. You are NOT authenticating the
11/// server. You are relying on having the correct address for the leader, which
12/// you got from the leader schedule via an RPC you do trust. The security
13/// model lives at that layer, not this one.
14#[derive(Debug)]
15pub struct SkipServerVerification;
16 
17impl ServerCertVerifier for SkipServerVerification {
18    fn verify_server_cert(
19        &self,
20        _end_entity: &CertificateDer<'_>,
21        _intermediates: &[CertificateDer<'_>],
22        _server_name: &ServerName<'_>,
23        _ocsp: &[u8],
24        _now: UnixTime,
25    ) -> Result<ServerCertVerified, TlsError> {
26        Ok(ServerCertVerified::assertion())
27    }
28 
29    fn verify_tls12_signature(
30        &self, _m: &[u8], _c: &CertificateDer<'_>, _d: &DigitallySignedStruct,
31    ) -> Result<HandshakeSignatureValid, TlsError> {
32        Ok(HandshakeSignatureValid::assertion())
33    }
34 
35    fn verify_tls13_signature(
36        &self, _m: &[u8], _c: &CertificateDer<'_>, _d: &DigitallySignedStruct,
37    ) -> Result<HandshakeSignatureValid, TlsError> {
38        Ok(HandshakeSignatureValid::assertion())
39    }
40 
41    fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
42        vec![SignatureScheme::ED25519]
43    }
44}

1[dependencies]
2quinn = "0.11"
3rustls = { version = "0.23", default-features = false, features = ["ring"] }
4rcgen = "0.13"
5rustls-pki-types = "1"
6tokio = { version = "1", features = ["full"] }
7anyhow = "1"
8bincode = "1"
9bs58 = "0.5"
10ed25519-dalek = "2"
11 
12# The Solana crates you need are small and specific. Pulling the whole
13# solana-sdk is a large compile for three types.
14solana-keypair = "3"
15solana-signer = "3"
16solana-transaction = "3"
17solana-pubkey = "3"

1use anyhow::Result;
2use quinn::{ClientConfig, Endpoint, TransportConfig};
3use std::{net::SocketAddr, sync::Arc, time::Duration};
4 
5/// The ALPN the leader expects. Get this wrong and the handshake completes
6/// TLS and then the connection closes with no useful error.
7const ALPN_TPU: &[u8] = b"solana-tpu";
8 
9pub fn client_config(identity: &Keypair) -> Result<ClientConfig> {
10    let (cert, key) = client_identity(identity)?;
11 
12    let mut crypto = rustls::ClientConfig::builder()
13        .dangerous()
14        .with_custom_certificate_verifier(Arc::new(SkipServerVerification))
15        .with_client_auth_cert(vec![cert], key)?;
16 
17    crypto.alpn_protocols = vec![ALPN_TPU.to_vec()];
18    // Session resumption is not useful here and adds handshake surface.
19    crypto.enable_early_data = false;
20 
21    let mut config = ClientConfig::new(Arc::new(
22        quinn::crypto::rustls::QuicClientConfig::try_from(crypto)?,
23    ));
24 
25    let mut transport = TransportConfig::default();
26    // Leaders rotate every four slots, about 1.6 seconds. Keepalive well under
27    // the idle timeout so a connection you are about to need is still warm.
28    transport.keep_alive_interval(Some(Duration::from_secs(2)));
29    transport.max_idle_timeout(Some(Duration::from_secs(30).try_into()?));
30    // A transaction is at most 1232 bytes. There is no reason for large
31    // windows, and small ones fail faster when something is wrong.
32    transport.datagram_receive_buffer_size(None);
33    config.transport_config(Arc::new(transport));
34 
35    Ok(config)
36}
37 
38pub fn make_endpoint(identity: &Keypair) -> Result<Endpoint> {
39    let bind: SocketAddr = "0.0.0.0:0".parse()?;
40    let mut endpoint = Endpoint::client(bind)?;
41    endpoint.set_default_client_config(client_config(identity)?);
42    Ok(endpoint)
43}

1use anyhow::{Context, Result};
2use quinn::Connection;
3use solana_transaction::versioned::VersionedTransaction;
4use std::time::Duration;
5 
6/// Send one transaction on an established connection.
7///
8/// The TPU protocol is deliberately minimal: open a UNIdirectional stream,
9/// write the serialized transaction, finish. There is no response. The leader
10/// either processes it or does not, and you find out by querying the chain.
11///
12/// That asymmetry is the point. A response would mean a round trip, and the
13/// entire design goal is to avoid one.
14pub async fn send_transaction(conn: &Connection, tx: &VersionedTransaction) -> Result<()> {
15    let bytes = bincode::serialize(tx).context("serialize transaction")?;
16    anyhow::ensure!(bytes.len() <= 1232, "transaction exceeds the packet limit");
17 
18    let mut stream = conn.open_uni().await.context("open stream")?;
19    stream.write_all(&bytes).await.context("write transaction")?;
20    stream.finish().context("finish stream")?;
21 
22    // finish() only queues the FIN. Waiting for the peer to acknowledge the
23    // stream is what tells you the bytes actually left. Skip this and you
24    // will report successes for transactions that never went anywhere.
25    tokio::time::timeout(Duration::from_millis(500), stream.stopped())
26        .await
27        .context("timed out waiting for stream acknowledgement")??;
28 
29    Ok(())
30}

1use anyhow::Result;
2use quinn::{Connection, Endpoint};
3use std::{collections::HashMap, net::SocketAddr, sync::Arc, time::{Duration, Instant}};
4use tokio::sync::RwLock;
5 
6/// Connections, kept warm, keyed by leader address.
7///
8/// This struct is the difference between a fast client and a slow one. A QUIC
9/// handshake costs a round trip; doing it per transaction hands back everything
10/// the protocol was meant to save. Connect ahead of the leader you are about to
11/// need, and hold what you have.
12pub struct ConnectionPool {
13    endpoint: Endpoint,
14    conns: Arc<RwLock<HashMap<SocketAddr, (Connection, Instant)>>>,
15    idle_ttl: Duration,
16}
17 
18impl ConnectionPool {
19    pub fn new(endpoint: Endpoint) -> Self {
20        Self {
21            endpoint,
22            conns: Arc::new(RwLock::new(HashMap::new())),
23            idle_ttl: Duration::from_secs(60),
24        }
25    }
26 
27    pub async fn get(&self, addr: SocketAddr) -> Result<Connection> {
28        // Fast path: a live connection we already hold.
29        if let Some((conn, _)) = self.conns.read().await.get(&addr) {
30            if conn.close_reason().is_none() {
31                return Ok(conn.clone());
32            }
33        }
34 
35        // Slow path. Note this can race: two callers may both connect to the
36        // same leader. That is deliberate. Serialising behind a write lock
37        // would make every caller wait out one handshake, which is worse than
38        // occasionally opening a connection twice and dropping one.
39        let conn = self.endpoint.connect(addr, "solana")?.await?;
40        self.conns.write().await.insert(addr, (conn.clone(), Instant::now()));
41        Ok(conn)
42    }
43 
44    /// Open connections to leaders you are about to need, before you need them.
45    pub async fn warm(&self, addrs: &[SocketAddr]) {
46        for addr in addrs {
47            let _ = self.get(*addr).await;
48        }
49    }
50 
51    pub async fn evict_idle(&self) {
52        let now = Instant::now();
53        self.conns.write().await.retain(|_, (conn, last)| {
54            conn.close_reason().is_none() && now.duration_since(*last) < self.idle_ttl
55        });
56    }
57}

1use anyhow::Result;
2use solana_client::nonblocking::rpc_client::RpcClient;
3use std::net::SocketAddr;
4 
5/// Resolve the TPU addresses of the current and upcoming leaders.
6///
7/// Leaders serve four consecutive slots, so looking ahead by a handful of
8/// slots gives you the next two or three validators. Warm connections to them
9/// and your send path never pays a handshake.
10pub async fn upcoming_tpu_addrs(rpc: &RpcClient, lookahead_slots: u64) -> Result<Vec<SocketAddr>> {
11    let slot = rpc.get_slot().await?;
12    let leaders = rpc.get_slot_leaders(slot, lookahead_slots).await?;
13    let nodes = rpc.get_cluster_nodes().await?;
14 
15    let mut out = Vec::new();
16    let mut seen = std::collections::HashSet::new();
17 
18    for leader in leaders {
19        if !seen.insert(leader) {
20            continue; // the same validator across its four slots
21        }
22        if let Some(node) = nodes.iter().find(|n| n.pubkey == leader.to_string()) {
23            // tpu_quic is the QUIC port; tpu is the legacy UDP one.
24            if let Some(addr) = node.tpu_quic {
25                out.push(addr);
26            }
27        }
28    }
29    Ok(out)
30}
31 
32// A sensible loop: refresh every slot or two, warm anything new, evict the
33// leaders that have passed.
34//
35//   loop {
36//       let addrs = upcoming_tpu_addrs(&rpc, 8).await?;
37//       pool.warm(&addrs).await;
38//       pool.evict_idle().await;
39//       tokio::time::sleep(Duration::from_millis(400)).await;
40//   }