seyd_transport/
endpoint.rs

1// Copyright 2026 Anton Gravestam
2// SPDX-License-Identifier: Apache-2.0
3use crate::cert::Cert;
4use crate::prober::Prober;
5use crate::session::Session;
6use crate::webtransport;
7use std::sync::atomic::{AtomicU64, Ordering};
8use std::sync::Arc;
9use std::time::Duration;
10use tokio::sync::mpsc;
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub enum Congestion {
14    /// Model-based; keeps the modem queue shallow. The default for media.
15    Bbr,
16    /// Loss-based; selectable for A/B on netem.
17    Cubic,
18}
19
20#[derive(Debug, Clone)]
21pub struct TransportConfig {
22    pub congestion: Congestion,
23    /// Bytes quinn may hold in its outgoing datagram buffer.
24    ///
25    /// This is a ceiling, not the latency bound: the engine's admission control
26    /// measures what is *queued* here against the QoS profile's backlog
27    /// threshold (`Session::send_buffer_queued`) and drops delta frames long
28    /// before the buffer fills. What the size must guarantee is that a keyframe
29    /// always fits — keyframes are never dropped, and a keyframe chunk that
30    /// finds no room is retried with 1 ms sleeps and then abandoned, tearing
31    /// the frame. So: the largest backlog a profile tolerates (3 frames at
32    /// 6 Mbps / 30 fps is 75 KB) plus the largest keyframe with parity (a 2 s
33    /// GOP at 6 Mbps can reach ~170 KB), with headroom.
34    pub datagram_send_buffer: usize,
35    pub datagram_receive_buffer: usize,
36    pub max_idle: Duration,
37    pub keep_alive: Duration,
38    /// Starting MTU assumption; DPLPMTUD probes upward from here.
39    pub initial_mtu: u16,
40}
41
42impl Default for TransportConfig {
43    fn default() -> Self {
44        Self {
45            congestion: Congestion::Bbr,
46            datagram_send_buffer: 256 * 1024,
47            datagram_receive_buffer: 2 * 1024 * 1024,
48            max_idle: Duration::from_secs(10),
49            keep_alive: Duration::from_secs(2),
50            initial_mtu: 1200,
51        }
52    }
53}
54
55/// One quinn endpoint per pre-bound socket, all feeding one session queue.
56pub struct Endpoint {
57    endpoints: Vec<quinn::Endpoint>,
58    probers: Vec<Prober>,
59    sessions: tokio::sync::Mutex<mpsc::Receiver<Session>>,
60    local_addrs: Vec<std::net::SocketAddr>,
61    cfg: TransportConfig,
62}
63
64fn install_crypto_provider() {
65    // Idempotent; a second install returns Err, which is fine.
66    let _ = rustls::crypto::ring::default_provider().install_default();
67}
68
69/// Build a full quinn `ServerConfig` (TLS + transport policy) for `cert`.
70/// Used at bind time and again on rotation.
71fn make_server_config(cert: &Cert, cfg: &TransportConfig) -> anyhow::Result<quinn::ServerConfig> {
72    let cert_chain = vec![rustls::pki_types::CertificateDer::from(
73        cert.cert_der.clone(),
74    )];
75    let key = rustls::pki_types::PrivateKeyDer::Pkcs8(rustls::pki_types::PrivatePkcs8KeyDer::from(
76        cert.key_der.clone(),
77    ));
78    let mut tls = rustls::ServerConfig::builder_with_provider(Arc::new(
79        rustls::crypto::ring::default_provider(),
80    ))
81    .with_protocol_versions(&[&rustls::version::TLS13])?
82    .with_no_client_auth()
83    .with_single_cert(cert_chain, key)?;
84    tls.alpn_protocols = vec![b"h3".to_vec()];
85    tls.max_early_data_size = 0;
86
87    let mut transport = quinn::TransportConfig::default();
88    transport
89        .max_idle_timeout(Some(cfg.max_idle.try_into()?))
90        .keep_alive_interval(Some(cfg.keep_alive))
91        .datagram_send_buffer_size(cfg.datagram_send_buffer)
92        .datagram_receive_buffer_size(Some(cfg.datagram_receive_buffer))
93        .initial_mtu(cfg.initial_mtu)
94        .mtu_discovery_config(Some(quinn::MtuDiscoveryConfig::default()))
95        // Streams are only the control channel and, later, reliable
96        // channels; datagrams carry media.
97        .max_concurrent_bidi_streams(16u32.into())
98        .max_concurrent_uni_streams(16u32.into());
99    match cfg.congestion {
100        Congestion::Bbr => {
101            transport
102                .congestion_controller_factory(Arc::new(quinn::congestion::BbrConfig::default()));
103        }
104        Congestion::Cubic => {
105            transport
106                .congestion_controller_factory(Arc::new(quinn::congestion::CubicConfig::default()));
107        }
108    }
109
110    let quic_server = quinn::crypto::rustls::QuicServerConfig::try_from(tls)?;
111    let mut server_config = quinn::ServerConfig::with_crypto(Arc::new(quic_server));
112    server_config.transport_config(Arc::new(transport));
113    Ok(server_config)
114}
115
116impl Endpoint {
117    /// Serve on already-bound UDP sockets (the caller binds them, runs STUN on
118    /// them, and hands them over; they are never rebound).
119    pub fn bind(
120        socks: Vec<std::net::UdpSocket>,
121        cert: &Cert,
122        cfg: TransportConfig,
123    ) -> anyhow::Result<Endpoint> {
124        install_crypto_provider();
125        anyhow::ensure!(!socks.is_empty(), "at least one socket is required");
126        let server_config = make_server_config(cert, &cfg)?;
127        let (tx, rx) = mpsc::channel(16);
128        let next_id = Arc::new(AtomicU64::new(1));
129        let mut endpoints = Vec::new();
130        let mut probers = Vec::new();
131        let mut local_addrs = Vec::new();
132        for sock in socks {
133            sock.set_nonblocking(true)?;
134            let probe_sock = sock.try_clone()?;
135            local_addrs.push(sock.local_addr()?);
136            let ep = quinn::Endpoint::new(
137                quinn::EndpointConfig::default(),
138                Some(server_config.clone()),
139                sock,
140                Arc::new(quinn::TokioRuntime),
141            )?;
142            probers.push(Prober::new(probe_sock));
143            let accept_ep = ep.clone();
144            let tx = tx.clone();
145            let next_id = next_id.clone();
146            let send_buffer = cfg.datagram_send_buffer;
147            tokio::spawn(async move {
148                while let Some(incoming) = accept_ep.accept().await {
149                    let tx = tx.clone();
150                    let id = next_id.fetch_add(1, Ordering::Relaxed);
151                    tokio::spawn(async move {
152                        let conn = match incoming.await {
153                            Ok(c) => c,
154                            Err(e) => {
155                                tracing::debug!("handshake failed: {e}");
156                                return;
157                            }
158                        };
159                        tracing::info!(session = id, peer = %conn.remote_address(), "quic connection");
160                        if let Err(e) = webtransport::serve(conn, id, tx, send_buffer).await {
161                            tracing::debug!(session = id, "h3 connection ended: {e:#}");
162                        }
163                    });
164                }
165            });
166            endpoints.push(ep);
167        }
168        Ok(Endpoint {
169            endpoints,
170            probers,
171            sessions: tokio::sync::Mutex::new(rx),
172            local_addrs,
173            cfg,
174        })
175    }
176
177    /// Next accepted WebTransport session, or `None` once the endpoint is closed.
178    pub async fn accept(&self) -> Option<Session> {
179        self.sessions.lock().await.recv().await
180    }
181
182    /// Install a new certificate for **future** connections. Existing QUIC
183    /// sessions carry their own completed TLS state and are untouched
184    /// (`quinn::Endpoint::set_server_config` applies to new handshakes only).
185    pub fn rotate(&self, cert: &Cert) -> anyhow::Result<()> {
186        let sc = make_server_config(cert, &self.cfg)?;
187        for ep in &self.endpoints {
188            ep.set_server_config(Some(sc.clone()));
189        }
190        Ok(())
191    }
192
193    pub fn local_addrs(&self) -> &[std::net::SocketAddr] {
194        &self.local_addrs
195    }
196
197    /// Hole-punch prober bound to socket `index` (same order as `bind`).
198    pub fn prober(&self, index: usize) -> Option<&Prober> {
199        self.probers.get(index)
200    }
201
202    /// Index of the socket whose address family matches `ip`.
203    pub fn socket_index_for(&self, ip: std::net::IpAddr) -> Option<usize> {
204        self.local_addrs
205            .iter()
206            .position(|a| a.is_ipv6() == ip.is_ipv6())
207    }
208
209    pub fn close(&self) {
210        for ep in &self.endpoints {
211            ep.close(0u32.into(), b"shutdown");
212        }
213    }
214}
seyd_transport/endpoint.rs

seyd_transport/
endpoint.rs
