diff --git a/rs/web-transport-quiche/Cargo.toml b/rs/web-transport-quiche/Cargo.toml index c751061..fd4c072 100644 --- a/rs/web-transport-quiche/Cargo.toml +++ b/rs/web-transport-quiche/Cargo.toml @@ -33,6 +33,7 @@ tokio = { version = "1", default-features = false, features = [ tokio-quiche = "0.19" tracing = "0.1" +url = "2" web-transport-proto = { workspace = true } web-transport-trait = { workspace = true } @@ -44,4 +45,3 @@ rustls-pemfile = "2" sha2 = "0.10" tokio = { version = "1", features = ["full"] } tracing-subscriber = { version = "0.3", features = ["env-filter"] } -url = "2" diff --git a/rs/web-transport-quiche/src/client.rs b/rs/web-transport-quiche/src/client.rs index 56a4135..0b25064 100644 --- a/rs/web-transport-quiche/src/client.rs +++ b/rs/web-transport-quiche/src/client.rs @@ -104,6 +104,26 @@ impl ClientBuilder { Self(self.0.with_server_certificate_hashes(hashes)) } + /// Send a PING on this interval, keeping an idle connection alive. + /// + /// Disabled by default. This must be shorter than the peer's + /// [Settings::max_idle_timeout] to have any effect; a third of it is a + /// reasonable choice. + pub fn with_keep_alive(self, interval: std::time::Duration) -> Self { + Self(self.0.with_keep_alive(interval)) + } + + /// Enable UDP generic segmentation offload (GSO), on by default. + /// + /// GSO cuts syscall overhead at high throughput by handing the kernel + /// several packets at once, but some NICs and virtual network stacks + /// mishandle it. Turn it off if large sends are being dropped. + /// + /// Only Linux supports GSO; elsewhere this does nothing. + pub fn with_gso(self, enabled: bool) -> Self { + Self(self.0.with_gso(enabled)) + } + /// Connect to the WebTransport server at the given URL. /// /// DNS resolution and socket setup happen eagerly. The returned [Connecting] @@ -116,13 +136,7 @@ impl ClientBuilder { request: impl Into, ) -> Result { let request = request.into(); - - let port = request.url.port().unwrap_or(443); - - let host = match request.url.host() { - Some(host) => host.to_string(), - None => return Err(ClientError::InvalidUrl(request.url.to_string())), - }; + let (host, port) = Self::target(&request)?; let connecting = self.0.connect(&host, port).await?; @@ -131,6 +145,20 @@ impl ClientBuilder { request, }) } + + /// The host and port to dial for a request. + fn target(request: &ConnectRequest) -> Result<(String, u16), ClientError> { + // `Host` renders IPv6 in URL form, bracketed, which is not what a + // resolver or a TLS server name wants. + let host = match request.url.host() { + Some(url::Host::Domain(host)) => host.to_string(), + Some(url::Host::Ipv4(ip)) => ip.to_string(), + Some(url::Host::Ipv6(ip)) => ip.to_string(), + None => return Err(ClientError::InvalidUrl(request.url.to_string())), + }; + + Ok((host, request.url.port().unwrap_or(443))) + } } /// A WebTransport connection that is still completing the handshake. diff --git a/rs/web-transport-quiche/src/ez/client.rs b/rs/web-transport-quiche/src/ez/client.rs index 684c250..955b41d 100644 --- a/rs/web-transport-quiche/src/ez/client.rs +++ b/rs/web-transport-quiche/src/ez/client.rs @@ -1,9 +1,11 @@ use std::io; use std::sync::Arc; +use std::time::Duration; use tokio_quiche::settings::{CertificateKind, Hooks, TlsCertificatePaths}; use rustls_pki_types::{CertificateDer, PrivateKeyDer}; +use crate::ez::socket::capabilities; use crate::ez::tls::{ClientHook, ClientVerify}; use crate::ez::DriverState; @@ -30,6 +32,8 @@ pub struct ClientBuilder { tls: Option<(Vec>, PrivateKeyDer<'static>)>, verify: ClientVerify, server_name: Option, + keep_alive: Option, + gso: bool, } impl Default for ClientBuilder { @@ -50,9 +54,33 @@ impl ClientBuilder { tls: None, verify: ClientVerify::Default, server_name: None, + keep_alive: None, + gso: true, } } + /// Send a PING on this interval, keeping an idle connection alive. + /// + /// Disabled by default. This must be shorter than the peer's + /// [Settings::max_idle_timeout] to have any effect; a third of it is a + /// reasonable choice. + pub fn with_keep_alive(mut self, interval: Duration) -> Self { + self.keep_alive = Some(interval); + self + } + + /// Enable UDP generic segmentation offload (GSO), on by default. + /// + /// GSO cuts syscall overhead at high throughput by handing the kernel + /// several packets at once, but some NICs and virtual network stacks + /// mishandle it. Turn it off if large sends are being dropped. + /// + /// Only Linux supports GSO; elsewhere this does nothing. + pub fn with_gso(mut self, enabled: bool) -> Self { + self.gso = enabled; + self + } + /// Listen for incoming packets on the given socket. /// /// Defaults to an ephemeral port if not specified. @@ -161,18 +189,16 @@ impl ClientBuilder { socket.connect(remote).await?; + // Enable the offloads the kernel supports before the socket is wrapped; + // `from_udp` starts with everything disabled. + let capabilities = capabilities(&socket, self.gso); + // Connect to the server using the addr we just resolved. - #[cfg_attr(not(target_os = "linux"), allow(unused_mut))] let mut socket = tokio_quiche::socket::Socket::< Arc, Arc, >::from_udp(socket)?; - - // Enable UDP GSO/GRO offload where the kernel supports it (Linux only). - // This mirrors the server listener and cuts syscall overhead at high - // throughput; it's a no-op if send/recv don't share one FD. - #[cfg(target_os = "linux")] - socket.apply_max_capabilities(); + socket.capabilities = capabilities; // Only the fully-insecure path (no verification of any kind) deserves a // warning; hash- and root-based verification still authenticate the peer. @@ -222,6 +248,7 @@ impl ClientBuilder { dgram_in.0, dgram_out.1, dgram_max.clone(), + self.keep_alive, ); let conn = tokio_quiche::quic::connect_with_config(socket, Some(server_name), ¶ms, app) diff --git a/rs/web-transport-quiche/src/ez/driver.rs b/rs/web-transport-quiche/src/ez/driver.rs index 51cfea0..6626a23 100644 --- a/rs/web-transport-quiche/src/ez/driver.rs +++ b/rs/web-transport-quiche/src/ez/driver.rs @@ -7,7 +7,8 @@ use std::{ atomic::{AtomicUsize, Ordering}, Arc, }, - task::{Poll, Waker}, + task::{Context, Poll, Waker}, + time::Duration, }; use tokio_quiche::{ buf_factory::BufFactory, @@ -236,6 +237,42 @@ impl DriverState { } } +/// Periodically asks quiche to make the next packet ack-eliciting, keeping an +/// otherwise silent connection out of the peer's idle timeout and holding NAT +/// bindings open. +struct KeepAlive { + period: Duration, + /// Created on the first poll so the timer registers with the runtime that + /// actually drives the connection, not whoever built the endpoint. + ticker: Option, +} + +impl KeepAlive { + fn new(period: Duration) -> Self { + Self { + period, + ticker: None, + } + } + + /// Returns true when a keep-alive is due. + fn poll(&mut self, cx: &mut Context) -> bool { + let period = self.period; + let ticker = self.ticker.get_or_insert_with(|| { + // The first tick is one period out; `interval` would instead fire + // immediately and ping a connection that just finished handshaking. + let start = tokio::time::Instant::now() + period; + let mut ticker = tokio::time::interval_at(start, period); + // A late tick means the connection was busy, which is exactly when a + // keep-alive is unnecessary. Don't replay the backlog. + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + ticker + }); + + ticker.poll_tick(cx).is_ready() + } +} + pub(super) struct Driver { state: Lock, @@ -253,6 +290,8 @@ pub(super) struct Driver { // Writable datagram size in bytes, published once at handshake. 0 means the // peer didn't negotiate the datagram extension. dgram_max: Arc, + + keep_alive: Option, } impl Driver { @@ -263,6 +302,7 @@ impl Driver { dgram_in: flume::Sender, dgram_out: flume::Receiver, dgram_max: Arc, + keep_alive: Option, ) -> Self { Self { state, @@ -274,6 +314,7 @@ impl Driver { dgram_in, dgram_out, dgram_max, + keep_alive: keep_alive.map(KeepAlive::new), } } @@ -486,6 +527,16 @@ impl Driver { return Poll::Pending; } + // quiche only adds a PING if the next packet wouldn't already be + // ack-eliciting, so a tick on a busy connection costs nothing. + let mut keep_alive = false; + if let Some(k) = self.keep_alive.as_mut() { + if k.poll(&mut Context::from_waker(waker)) { + qconn.send_ack_eliciting()?; + keep_alive = true; + } + } + // Snapshot stats while we hold an immutable view; stored under the lock below. let stats = ConnectionStats::from_quiche(qconn); @@ -549,7 +600,9 @@ impl Driver { self.flush_send(qconn, stream_id)?; } - if sleep { + // Returning Ready hands control back to the io loop, which flushes the + // scheduled PING to the socket. + if sleep && !keep_alive { Poll::Pending } else { Poll::Ready(Ok(())) diff --git a/rs/web-transport-quiche/src/ez/mod.rs b/rs/web-transport-quiche/src/ez/mod.rs index 8f69fa9..1b92945 100644 --- a/rs/web-transport-quiche/src/ez/mod.rs +++ b/rs/web-transport-quiche/src/ez/mod.rs @@ -11,6 +11,7 @@ mod lock; mod recv; mod send; mod server; +mod socket; mod stream; pub mod tls; diff --git a/rs/web-transport-quiche/src/ez/server.rs b/rs/web-transport-quiche/src/ez/server.rs index 806cd30..2f34db0 100644 --- a/rs/web-transport-quiche/src/ez/server.rs +++ b/rs/web-transport-quiche/src/ez/server.rs @@ -1,14 +1,16 @@ use std::net::SocketAddr; use std::sync::Arc; +use std::time::Duration; use std::{io, marker::PhantomData}; use tokio::sync::mpsc; use tokio::task::JoinSet; use tokio_quiche::quic::SimpleConnectionIdGenerator; use tokio_quiche::settings::{CertificateKind, Hooks, TlsCertificatePaths}; -use tokio_quiche::socket::{QuicListener, SocketCapabilities}; +use tokio_quiche::socket::QuicListener; use rustls_pki_types::{CertificateDer, PrivateKeyDer}; +use crate::ez::socket::capabilities; use crate::ez::tls::{DynamicCertHook, StaticCertHook}; use crate::ez::DriverState; @@ -25,7 +27,17 @@ pub struct ServerInit {} /// Used with [ServerBuilder] to require at least one listener. #[derive(Default)] pub struct ServerWithListener { - listeners: Vec, + listeners: Vec, +} + +/// A listener, in insertion order so [Server::local_addrs] matches the order the +/// caller added them. +enum Listener { + /// Supplied by the caller and used as-is. + Ready(QuicListener), + /// Opened by us. It stays a bare socket until the build step, the first + /// point where every option affecting it is known. + Socket(tokio::net::UdpSocket), } /// Construct a QUIC server using sane defaults. @@ -34,6 +46,8 @@ pub struct ServerBuilder { metrics: M, state: S, alpn: Vec>, + keep_alive: Option, + gso: bool, client_auth: ClientAuth, } @@ -53,6 +67,8 @@ impl ServerBuilder { metrics: m, state: ServerInit {}, alpn: Vec::new(), + keep_alive: None, + gso: true, client_auth: ClientAuth::None, } } @@ -63,8 +79,10 @@ impl ServerBuilder { ServerBuilder { settings: self.settings, metrics: self.metrics, - state: ServerWithListener { listeners: vec![] }, + state: ServerWithListener::default(), alpn: self.alpn, + keep_alive: self.keep_alive, + gso: self.gso, client_auth: self.client_auth, } } @@ -96,6 +114,22 @@ impl ServerBuilder { self } + /// Send a PING to each client on this interval, keeping idle connections alive. + /// + /// See [ServerBuilder::with_keep_alive](ServerBuilder::::with_keep_alive). + pub fn with_keep_alive(mut self, interval: Duration) -> Self { + self.keep_alive = Some(interval); + self + } + + /// Enable UDP generic segmentation offload (GSO), on by default. + /// + /// See [ServerBuilder::with_gso](ServerBuilder::::with_gso). + pub fn with_gso(mut self, enabled: bool) -> Self { + self.gso = enabled; + self + } + /// Authenticate clients with mTLS. /// /// Defaults to [ClientAuth::None]. @@ -107,29 +141,20 @@ impl ServerBuilder { impl ServerBuilder { /// Configure the server to use the provided QUIC listener. + /// + /// The listener is used as-is: it carries its own capabilities and + /// connection ID generator, so [ServerBuilder::with_gso] does not apply. pub fn with_listener(mut self, listener: QuicListener) -> Self { - self.state.listeners.push(listener); + self.state.listeners.push(Listener::Ready(listener)); self } /// Listen for incoming packets on the given socket. - pub fn with_socket(self, socket: std::net::UdpSocket) -> io::Result { + pub fn with_socket(mut self, socket: std::net::UdpSocket) -> io::Result { socket.set_nonblocking(true)?; let socket = tokio::net::UdpSocket::from_std(socket)?; - - // TODO Modify quiche to add other platform support. - #[cfg(target_os = "linux")] - let capabilities = SocketCapabilities::apply_all_and_get_compatibility(&socket); - #[cfg(not(target_os = "linux"))] - let capabilities = SocketCapabilities::default(); - - let listener = QuicListener { - socket, - cid_generator: Arc::new(SimpleConnectionIdGenerator), - capabilities, - }; - - Ok(self.with_listener(listener)) + self.state.listeners.push(Listener::Socket(socket)); + Ok(self) } /// Listen for incoming packets on the given address. @@ -148,6 +173,31 @@ impl ServerBuilder { self } + /// Send a PING to each client on this interval, keeping idle connections alive. + /// + /// Disabled by default. A server usually wants to let idle clients time out + /// rather than hold them open, so reach for this only when something in the + /// path (a NAT or load balancer) drops silent flows sooner than + /// [Settings::max_idle_timeout] would. + pub fn with_keep_alive(mut self, interval: Duration) -> Self { + self.keep_alive = Some(interval); + self + } + + /// Enable UDP generic segmentation offload (GSO), on by default. + /// + /// GSO cuts syscall overhead at high throughput by handing the kernel + /// several packets at once, but some NICs and virtual network stacks + /// mishandle it. Turn it off if large sends are being dropped. + /// + /// This applies to sockets from [ServerBuilder::with_socket] and + /// [ServerBuilder::with_bind] only, not to a [ServerBuilder::with_listener] + /// listener. Only Linux supports GSO; elsewhere this does nothing. + pub fn with_gso(mut self, enabled: bool) -> Self { + self.gso = enabled; + self + } + /// Authenticate clients with mTLS. /// /// Defaults to [ClientAuth::None]. @@ -212,18 +262,29 @@ impl ServerBuilder { connection_hook: Some(hook), }; - // Capture local addresses before the listeners are consumed. - let local_addrs: Vec = self + let listeners: Vec = self .state .listeners - .iter() - .map(|l| l.socket.local_addr().unwrap()) + .into_iter() + .map(|listener| match listener { + Listener::Ready(listener) => listener, + Listener::Socket(socket) => QuicListener { + capabilities: capabilities(&socket, self.gso), + socket, + cid_generator: Arc::new(SimpleConnectionIdGenerator), + }, + }) .collect(); + // Capture local addresses before the listeners are consumed. + let local_addrs: Vec = listeners + .iter() + .map(|l| l.socket.local_addr()) + .collect::>()?; + let params = tokio_quiche::ConnectionParams::new_server(self.settings, dummy_tls, hooks); - let server = - tokio_quiche::listen_with_capabilities(self.state.listeners, params, self.metrics)?; - Ok(Server::new(server, local_addrs)) + let server = tokio_quiche::listen_with_capabilities(listeners, params, self.metrics)?; + Ok(Server::new(server, local_addrs, self.keep_alive)) } } @@ -289,6 +350,7 @@ impl Server { fn new( sockets: Vec>, local_addrs: Vec, + keep_alive: Option, ) -> Self { let mut tasks = JoinSet::default(); @@ -297,7 +359,7 @@ impl Server { for socket in sockets { let accept = accept.0.clone(); // TODO close all when one errors - tasks.spawn(Self::run_socket(socket, accept)); + tasks.spawn(Self::run_socket(socket, accept, keep_alive)); } Self { @@ -311,6 +373,7 @@ impl Server { async fn run_socket( socket: tokio_quiche::QuicConnectionStream, accept: mpsc::Sender, + keep_alive: Option, ) -> io::Result<()> { let mut rx = socket.into_inner(); while let Some(initial) = rx.recv().await { @@ -330,6 +393,7 @@ impl Server { dgram_in.0, dgram_out.1, dgram_max.clone(), + keep_alive, ); let inner = initial.start(session); diff --git a/rs/web-transport-quiche/src/ez/socket.rs b/rs/web-transport-quiche/src/ez/socket.rs new file mode 100644 index 0000000..84787ed --- /dev/null +++ b/rs/web-transport-quiche/src/ez/socket.rs @@ -0,0 +1,41 @@ +use tokio_quiche::socket::SocketCapabilities; + +/// Enable the socket options tokio-quiche knows how to use, optionally leaving +/// `UDP_SEGMENT` (GSO) off. +/// +/// All of these are Linux-only; every other platform reports no capabilities. +#[cfg(target_os = "linux")] +pub(super) fn capabilities(socket: &S, gso: bool) -> SocketCapabilities { + use tokio_quiche::socket::SocketCapabilitiesBuilder; + + if gso { + return SocketCapabilities::apply_all_and_get_compatibility(socket); + } + + // `apply_all_and_get_compatibility` always turns GSO on, so opting out means + // reproducing it here minus the `gso()` call. Each option is independently + // best-effort: a kernel that rejects one still gets the rest. + let mut builder = SocketCapabilitiesBuilder::new(socket); + let _ = builder.check_udp_drop(); + let _ = builder.txtime(); + let _ = builder.gro(); + let _ = builder.rcvmark(); + let _ = builder.ip_mtu_discover_probe(); + let _ = builder.ipv6_mtu_discover_probe(); + + // Source-address control is only useful when the socket may send from an + // address it isn't bound to. + if let Ok(true) = builder.allows_nonlocal_source() { + let _ = builder.ipv4_pktinfo(); + let _ = builder.ipv4_recvorigdstaddr(); + let _ = builder.ipv6_pktinfo(); + let _ = builder.ipv6_recvorigdstaddr(); + } + + builder.finish() +} + +#[cfg(not(target_os = "linux"))] +pub(super) fn capabilities(_socket: &S, _gso: bool) -> SocketCapabilities { + SocketCapabilities::default() +} diff --git a/rs/web-transport-quiche/src/server.rs b/rs/web-transport-quiche/src/server.rs index 6f5cdac..b833ea1 100644 --- a/rs/web-transport-quiche/src/server.rs +++ b/rs/web-transport-quiche/src/server.rs @@ -82,6 +82,20 @@ impl ServerBuilder { Self(self.0.with_settings(settings)) } + /// Send a PING to each client on this interval, keeping idle connections alive. + /// + /// See [ServerBuilder::with_keep_alive](ServerBuilder::::with_keep_alive). + pub fn with_keep_alive(self, interval: std::time::Duration) -> Self { + Self(self.0.with_keep_alive(interval)) + } + + /// Enable UDP generic segmentation offload (GSO), on by default. + /// + /// See [ServerBuilder::with_gso](ServerBuilder::::with_gso). + pub fn with_gso(self, enabled: bool) -> Self { + Self(self.0.with_gso(enabled)) + } + /// Authenticate clients with mTLS. /// /// Defaults to [ez::ClientAuth::None]. @@ -92,6 +106,9 @@ impl ServerBuilder { impl ServerBuilder { /// Configure the server to use the provided QUIC listener. + /// + /// The listener is used as-is: it carries its own capabilities and + /// connection ID generator, so [ServerBuilder::with_gso] does not apply. pub fn with_listener(self, listener: tokio_quiche::socket::QuicListener) -> Self { Self(self.0.with_listener(listener)) } @@ -114,6 +131,29 @@ impl ServerBuilder { Self(self.0.with_settings(settings)) } + /// Send a PING to each client on this interval, keeping idle connections alive. + /// + /// Disabled by default. A server usually wants to let idle clients time out + /// rather than hold them open, so reach for this only when something in the + /// path (a NAT or load balancer) drops silent flows sooner than + /// [Settings::max_idle_timeout](ez::Settings::max_idle_timeout) would. + pub fn with_keep_alive(self, interval: std::time::Duration) -> Self { + Self(self.0.with_keep_alive(interval)) + } + + /// Enable UDP generic segmentation offload (GSO), on by default. + /// + /// GSO cuts syscall overhead at high throughput by handing the kernel + /// several packets at once, but some NICs and virtual network stacks + /// mishandle it. Turn it off if large sends are being dropped. + /// + /// This applies to sockets from [ServerBuilder::with_socket] and + /// [ServerBuilder::with_bind] only, not to a [ServerBuilder::with_listener] + /// listener. Only Linux supports GSO; elsewhere this does nothing. + pub fn with_gso(self, enabled: bool) -> Self { + Self(self.0.with_gso(enabled)) + } + /// Authenticate clients with mTLS. /// /// Defaults to [ez::ClientAuth::None]. diff --git a/rs/web-transport-quiche/tests/connect.rs b/rs/web-transport-quiche/tests/connect.rs new file mode 100644 index 0000000..6406980 --- /dev/null +++ b/rs/web-transport-quiche/tests/connect.rs @@ -0,0 +1,115 @@ +//! Dialing by address literal. +//! +//! `Url::host()` renders an IPv6 host in URL form, bracketed, so handing it +//! straight to a resolver fails. Hostname URLs never exercise that path, which +//! is why it went unnoticed. + +use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr}; + +use anyhow::{Context, Result}; +use rcgen::{CertifiedKey, KeyPair}; +use rustls_pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer}; +use url::Url; +use web_transport_quiche::{ClientBuilder, ServerBuilder, Settings}; + +fn make_self_signed() -> Result<(Vec>, PrivateKeyDer<'static>)> { + let CertifiedKey { cert, signing_key } = rcgen::generate_simple_self_signed(vec![ + "localhost".into(), + "::1".into(), + "127.0.0.1".into(), + ]) + .context("rcgen self-signed")?; + + let cert_der = CertificateDer::from(cert.der().to_vec()); + let key_bytes = KeyPair::serialize_der(&signing_key); + let key_der = PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(key_bytes)); + + Ok((vec![cert_der], key_der)) +} + +fn init_tracing() { + let _ = tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("warn")), + ) + .with_test_writer() + .try_init(); +} + +async fn spawn_server(bind: SocketAddr) -> Result<(SocketAddr, tokio::task::JoinHandle<()>)> { + let (chain, key) = make_self_signed()?; + + let mut server = ServerBuilder::default() + .with_bind(bind)? + .with_single_cert(chain, key)?; + + let addr = *server + .local_addrs() + .first() + .context("server has no local address")?; + + let handle = tokio::spawn(async move { + if let Some(request) = server.accept().await { + if let Ok(session) = request.ok().await { + let _ = session.closed().await; + } + } + }); + + Ok((addr, handle)) +} + +fn client() -> ClientBuilder { + // The certificate is self-signed and the subject here is address handling, + // which verify.rs already covers from the other side. + let mut settings = Settings::default(); + settings.verify_peer = false; + + ClientBuilder::default().with_settings(settings) +} + +/// `https://[::1]:port/` must reach the server. Before the fix the bracketed +/// form reached the resolver verbatim and failed to resolve. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn connect_ipv6_literal_url() -> Result<()> { + init_tracing(); + + let (addr, server) = spawn_server((Ipv6Addr::LOCALHOST, 0).into()).await?; + let url = Url::parse(&format!("https://[::1]:{}/", addr.port()))?; + + let session = client() + .with_bind((Ipv6Addr::LOCALHOST, 0))? + .connect(url) + .await + .context("an IPv6 literal URL should resolve to the address it names")? + .established() + .await?; + + session.close(0, "bye"); + session.closed().await; + server.abort(); + Ok(()) +} + +/// The IPv4 counterpart, which already worked: `Host`'s IPv4 form needs no +/// unwrapping. It guards against a "fix" that breaks the common case. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn connect_ipv4_literal_url() -> Result<()> { + init_tracing(); + + let (addr, server) = spawn_server((Ipv4Addr::LOCALHOST, 0).into()).await?; + let url = Url::parse(&format!("https://127.0.0.1:{}/", addr.port()))?; + + let session = client() + .with_bind((Ipv4Addr::LOCALHOST, 0))? + .connect(url) + .await? + .established() + .await?; + + session.close(0, "bye"); + session.closed().await; + server.abort(); + Ok(()) +} diff --git a/rs/web-transport-quiche/tests/transport.rs b/rs/web-transport-quiche/tests/transport.rs new file mode 100644 index 0000000..1060835 --- /dev/null +++ b/rs/web-transport-quiche/tests/transport.rs @@ -0,0 +1,166 @@ +//! QUIC transport controls: keep-alive and the GSO opt-out. +//! +//! The load-bearing assertion for keep-alive is the negative one: the same +//! connection *without* it must actually die of the idle timeout, otherwise the +//! positive test would pass no matter what the driver does. + +use std::{ + net::{Ipv4Addr, SocketAddr}, + time::Duration, +}; + +use anyhow::{Context, Result}; +use rcgen::{CertifiedKey, KeyPair}; +use rustls_pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer}; +use url::Url; +use web_transport_quiche::{ClientBuilder, ServerBuilder, Settings}; + +/// Short enough to keep the test quick, long enough to survive a loaded CI +/// machine stalling the driver task between ticks. +const IDLE_TIMEOUT: Duration = Duration::from_secs(1); +const KEEP_ALIVE: Duration = Duration::from_millis(200); + +/// Long enough that an un-kept connection is certainly gone. +const IDLE_WAIT: Duration = Duration::from_secs(3); + +fn make_self_signed() -> Result<(Vec>, PrivateKeyDer<'static>)> { + let CertifiedKey { cert, signing_key } = + rcgen::generate_simple_self_signed(vec!["localhost".into(), "127.0.0.1".into()]) + .context("rcgen self-signed")?; + + let cert_der = CertificateDer::from(cert.der().to_vec()); + let key_bytes = KeyPair::serialize_der(&signing_key); + let key_der = PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(key_bytes)); + + Ok((vec![cert_der], key_der)) +} + +fn init_tracing() { + let _ = tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("warn")), + ) + .with_test_writer() + .try_init(); +} + +fn idle_settings() -> Settings { + let mut settings = Settings::default(); + settings.max_idle_timeout = Some(IDLE_TIMEOUT); + settings +} + +/// Spawn a server that accepts one session and holds it open without sending +/// anything, so the only traffic on the wire is whatever keep-alive produces. +async fn spawn_server(gso: bool) -> Result<(SocketAddr, tokio::task::JoinHandle<()>)> { + let (chain, key) = make_self_signed()?; + + let bind: SocketAddr = (Ipv4Addr::LOCALHOST, 0).into(); + let mut server = ServerBuilder::default() + .with_bind(bind)? + .with_settings(idle_settings()) + .with_gso(gso) + .with_single_cert(chain, key)?; + + let addr = *server + .local_addrs() + .first() + .context("server has no local address")?; + + let handle = tokio::spawn(async move { + if let Some(request) = server.accept().await { + if let Ok(session) = request.ok().await { + let _ = session.closed().await; + } + } + }); + + Ok((addr, handle)) +} + +fn url_for(addr: SocketAddr) -> Result { + Ok(Url::parse(&format!("https://127.0.0.1:{}/", addr.port()))?) +} + +fn client() -> ClientBuilder { + // The cert is self-signed, and the point here is transport behavior rather + // than verification, which verify.rs already covers. + let mut settings = idle_settings(); + settings.verify_peer = false; + + ClientBuilder::default().with_settings(settings) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn keep_alive_outlives_idle_timeout() -> Result<()> { + init_tracing(); + + let (addr, server) = spawn_server(true).await?; + + let session = client() + .with_bind((Ipv4Addr::LOCALHOST, 0))? + .with_keep_alive(KEEP_ALIVE) + .connect(url_for(addr)?) + .await? + .established() + .await?; + + // Nothing is sent on the session, so only the keep-alive PINGs can hold the + // idle timer open on either end. + if let Ok(err) = tokio::time::timeout(IDLE_WAIT, session.closed()).await { + anyhow::bail!("keep-alive connection closed after {IDLE_WAIT:?}: {err}"); + } + + session.close(0, "bye"); + session.closed().await; + server.abort(); + Ok(()) +} + +/// The control for `keep_alive_outlives_idle_timeout`. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn idle_timeout_without_keep_alive() -> Result<()> { + init_tracing(); + + let (addr, server) = spawn_server(true).await?; + + let session = client() + .with_bind((Ipv4Addr::LOCALHOST, 0))? + .connect(url_for(addr)?) + .await? + .established() + .await?; + + tokio::time::timeout(IDLE_WAIT, session.closed()) + .await + .map_err(|_| { + anyhow::anyhow!("idle connection survived {IDLE_WAIT:?} with no keep-alive") + })?; + + server.abort(); + Ok(()) +} + +/// GSO is Linux-only, so elsewhere this just pins that the option is accepted +/// rather than rejected at initialization. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn handshake_without_gso() -> Result<()> { + init_tracing(); + + let (addr, server) = spawn_server(false).await?; + + let session = client() + .with_bind((Ipv4Addr::LOCALHOST, 0))? + .with_gso(false) + .connect(url_for(addr)?) + .await? + .established() + .await + .context("handshake should succeed with GSO disabled")?; + + session.close(0, "bye"); + session.closed().await; + server.abort(); + Ok(()) +}