v0.34.3

src/http3.rs

@@ -17,7 +17,7 @@ use http_body_util::BodyExt;
 use serde::{Deserialize, Serialize};
 use std::{
     io,
-    net::SocketAddr,
+    net::{SocketAddr, UdpSocket},
     pin::Pin,
     sync::{atomic::Ordering, Arc},
     task::{Context, Poll},
@@ -63,6 +63,66 @@ pub struct Bound {
     address: SocketAddr,
     settings: Settings,
 }
+// Set the actual QUIC UDP socket buffers, not only sysctl maxima. Linux
+// doubles SO_RCVBUF/SO_SNDBUF values returned by getsockopt(2), which `ss`
+// reports as rb/tb. Target the effective buffer size shown by `ss`.
+fn requested_socket_buffer_size(effective_bytes: u32) -> usize {
+    #[cfg(target_os = "linux")]
+    {
+        (effective_bytes as usize / 2).max(1)
+    }
+    #[cfg(not(target_os = "linux"))]
+    {
+        effective_bytes as usize
+    }
+}
+
+fn configure_udp_socket(socket: &UdpSocket, address: SocketAddr, settings: Settings) {
+    let socket = socket2::SockRef::from(socket);
+    for (direction, requested, result) in [
+        (
+            "receive",
+            settings.udp_receive_buffer_bytes,
+            if settings.udp_receive_buffer_bytes == 0 {
+                Ok(())
+            } else {
+                socket.set_recv_buffer_size(requested_socket_buffer_size(
+                    settings.udp_receive_buffer_bytes,
+                ))
+            },
+        ),
+        (
+            "send",
+            settings.udp_send_buffer_bytes,
+            if settings.udp_send_buffer_bytes == 0 {
+                Ok(())
+            } else {
+                socket.set_send_buffer_size(requested_socket_buffer_size(
+                    settings.udp_send_buffer_bytes,
+                ))
+            },
+        ),
+    ] {
+        if let Err(error) = result {
+            tracing::warn!(%address, direction, requested, %error,
+                "HTTP/3 UDP buffer request failed; continuing with OS socket default");
+        }
+    }
+    let receive = socket.recv_buffer_size().ok();
+    let send = socket.send_buffer_size().ok();
+    tracing::info!(%address, ?receive, ?send,
+        "HTTP/3 UDP effective socket buffers (bytes)");
+    for (direction, requested, actual) in [
+        ("receive", settings.udp_receive_buffer_bytes, receive),
+        ("send", settings.udp_send_buffer_bytes, send),
+    ] {
+        if requested != 0 && actual.is_some_and(|size| size < requested as usize) {
+            tracing::warn!(%address, direction, requested, ?actual,
+                "HTTP/3 UDP socket buffer below configured target; check net.core.rmem_max and wmem_max on Linux");
+        }
+    }
+}
+
 pub fn bind(listener: Listener, state: &App) -> Result<Bound> {
     let settings = listener
         .http3
@@ -82,8 +142,16 @@ pub fn bind(listener: Listener, state: &App) -> Result<Bound> {
         .max_incoming(settings.max_concurrent_connections as usize)
         .incoming_buffer_size(64 * 1024)
         .incoming_buffer_size_total(1024 * 1024);
-    let endpoint = quinn::Endpoint::server(server_config.clone(), listener.address)
+    let socket = UdpSocket::bind(listener.address)
         .with_context(|| format!("cannot bind HTTP/3 UDP {}", listener.address))?;
+    configure_udp_socket(&socket, listener.address, settings);
+    let endpoint = quinn::Endpoint::new(
+        quinn::EndpointConfig::default(),
+        Some(server_config.clone()),
+        socket,
+        Arc::new(quinn::TokioRuntime),
+    )
+    .with_context(|| format!("cannot initialize HTTP/3 UDP {}", listener.address))?;
     Ok(Bound {
         endpoint,
         server_config,
@@ -91,13 +159,77 @@ pub fn bind(listener: Listener, state: &App) -> Result<Bound> {
         settings,
     })
 }
-/// Derive the advertised endpoint from ACTUALLY bound sockets, never saved intent
-/// or a backend Alt-Svc header. A pending H3 enable is not advertised before restart.
+/// An established UDP socket is immutable until restart. Use the committed
+/// *desired* transport policy for new QUIC connections on that same socket;
+/// never apply settings from a listener planned on a different IP/port.
+fn desired_for_bound(address: SocketAddr, tls: bool, desired: &[Listener]) -> Option<Settings> {
+    desired
+        .iter()
+        .find(|listener| listener.address == address && listener.tls == tls)
+        .and_then(|listener| listener.http3)
+}
+
+/// Compare immutable listener properties. H3 limits other than an increase in
+/// the startup connection capacity are hot-applied for *new* connections.
+/// UDP socket creation/removal (and H1/H2 TLS/ALPN settings) still need restart.
+pub fn listeners_require_restart(active: &[Listener], desired: &[Listener]) -> bool {
+    if active.len() != desired.len() {
+        return true;
+    }
+    active.iter().any(|bound| {
+        let Some(next) = desired.iter().find(|next| next.address == bound.address) else {
+            return true;
+        };
+        if bound.tls != next.tls
+            || bound.http2 != next.http2
+            || bound.http2_settings != next.http2_settings
+            || bound.h2c != next.h2c
+        {
+            return true;
+        }
+        match (bound.http3, next.http3) {
+            (None, None) => false,
+            (Some(initial), Some(updated)) => {
+                updated.max_concurrent_connections > initial.max_concurrent_connections
+                    || updated.udp_receive_buffer_bytes != initial.udp_receive_buffer_bytes
+                    || updated.udp_send_buffer_bytes != initial.udp_send_buffer_bytes
+            }
+            _ => true,
+        }
+    })
+}
+
+/// Origin-supplied Alt-Svc is never trusted. An HTTP/3 opt-out emits `clear`
+/// on TLS responses while the existing QUIC socket is still bound, so clients
+/// are told to discard their cached alternative service immediately.
+pub fn withdraw_advertisement(
+    headers: &mut HeaderMap,
+    local: Option<SocketAddr>,
+    tls: bool,
+    active: &[Listener],
+) {
+    headers.remove("alt-svc");
+    if tls
+        && local.is_some_and(|local| {
+            active
+                .iter()
+                .any(|binding| binding.matches(local, true) && binding.http3.is_some())
+        })
+    {
+        headers.insert("alt-svc", HeaderValue::from_static("clear"));
+    }
+}
+
+/// Advertise only an actually active H3 UDP listener, using the newest
+/// committed Alt-Svc lifetime. Disabling H3 for the listener or vhost publishes
+/// `Alt-Svc: clear` without waiting for a restart.
 pub fn advertise(
     headers: &mut HeaderMap,
     local: Option<SocketAddr>,
     tls: bool,
     active: &[Listener],
+    desired: &[Listener],
+    host_http3_enabled: bool,
 ) {
     headers.remove("alt-svc");
     if !tls {
@@ -106,11 +238,19 @@ pub fn advertise(
     let Some(local) = local else {
         return;
     };
-    if let Some(settings) = active
-        .iter()
-        .find(|binding| binding.matches(local, true))
-        .and_then(|binding| binding.http3)
-    {
+    let Some(bound) = active.iter().find(|binding| binding.matches(local, true)) else {
+        return;
+    };
+    // A desired H3 enable on a socket with no UDP listener is pending restart.
+    if bound.http3.is_none() {
+        return;
+    }
+    let settings = desired_for_bound(bound.address, bound.tls, desired);
+    if !host_http3_enabled || settings.is_none() {
+        withdraw_advertisement(headers, Some(local), true, active);
+        return;
+    }
+    if let Some(settings) = settings {
         let value = format!("h3=\":{}\"; ma={}", local.port(), settings.alt_svc_max_age);
         if let Ok(value) = HeaderValue::from_str(&value) {
             headers.insert("alt-svc", value);
@@ -131,6 +271,29 @@ pub async fn serve(bound: Bound, router: Router, state: Arc<App>, grace: Duratio
                 let Some(incoming)=incoming else {break;};
                 let peer=incoming.remote_address();
                 if !incoming.remote_address_validated() {let _=incoming.retry();continue;}
+                // Listener setting saves are already published through ArcSwap.
+                // Apply new limits to fresh QUIC handshakes on this bound UDP
+                // address, without mutating established connections.
+                let desired = {
+                    let snapshot = state.config.load();
+                    desired_for_bound(bound.address, true, &snapshot.listeners)
+                };
+                let Some(settings) = desired else { incoming.refuse(); continue; };
+                // Startup permits and Quinn's incoming capacity cannot be grown
+                // without restarting the UDP endpoint. A lower limit can be
+                // enforced dynamically without releasing live connection leases.
+                let open = bound.settings.max_concurrent_connections as usize - slots.available_permits();
+                let capacity = (settings.max_concurrent_connections as usize)
+                    .min(bound.settings.max_concurrent_connections as usize);
+                if open >= capacity {
+                    state.metrics.http3_rejected_connections.fetch_add(1,Ordering::Relaxed);
+                    state.metrics.rejected_connections.fetch_add(1,Ordering::Relaxed);
+                    state.logs.push_protocol_error(
+                        "http3_connection_limit_rejected", "HTTP/3 configured connection cap reached",
+                        peer, bound.address, "quic", "3", 0, None,
+                    );
+                    incoming.refuse();continue;
+                }
                 let Ok(permit)=slots.clone().try_acquire_owned() else {
                     state.metrics.http3_rejected_connections.fetch_add(1,Ordering::Relaxed);
                     state.metrics.rejected_connections.fetch_add(1,Ordering::Relaxed);
@@ -141,9 +304,21 @@ pub async fn serve(bound: Bound, router: Router, state: Arc<App>, grace: Duratio
                     incoming.refuse();continue;
                 };
                 let mut config=bound.server_config.clone();
+                let transport = match settings.transport() {
+                    Ok(transport) => transport,
+                    Err(error) => {
+                        state.logs.push_protocol_error(
+                            "http3_transport_config_error", error,
+                            peer, bound.address, "quic", "3", 0, None,
+                        );
+                        incoming.refuse();continue;
+                    }
+                };
+                config.transport = Arc::new(transport);
                 let Some(crypto)=state.config.load().quic_tls.clone() else {incoming.refuse();continue;};
-                // Certificate changes apply to new handshakes without changing
-                // active transport limits or regenerating the Retry token key.
+                // Existing QUIC sessions keep their negotiated transport limits.
+                // New sessions get current limits and certificate without
+                // regenerating the Retry token key.
                 config.crypto=crypto;
                 let connecting=match incoming.accept_with(Arc::new(config)) {
                     Ok(c)=>c,
@@ -155,7 +330,7 @@ pub async fn serve(bound: Bound, router: Router, state: Arc<App>, grace: Duratio
                     }
                 };
                 let router=router.clone();let state=state.clone();let address=bound.address;
-                let limit=bound.settings.max_concurrent_bidistream as usize;
+                let limit=settings.max_concurrent_bidistream as usize;
                 tasks.spawn(async move {
                     let connection=tokio::select! {
                         _=state.shutdown.cancelled()=>return,
@@ -251,6 +426,7 @@ async fn connection_loop(
                         drop(resolver);continue;
                     }
                     let router=router.clone();let state=state.clone();let peer=connection.remote_address();let tls_sni=tls_sni.clone();
+                    let quic_diagnostics = connection.clone();
                     requests.spawn(async move {
                         let resolved=tokio::time::timeout(state.header_timeout,resolver.resolve_request()).await;
                         let (mut request,stream)=match resolved {
@@ -403,6 +579,27 @@ async fn connection_loop(
                         } else {
                             log_h3_response_complete(peer, &stream_id, response_status, sent_body_bytes);
                         }
+                        // Connection-level cumulative counters (not per-stream
+                        // deltas). Useful for distinguishing QUIC loss/cwnd/MTU
+                        // limitations from application CPU when benchmarking.
+                        if sent_body_bytes >= 8 * 1024 * 1024 || expected.is_some_and(|n| n >= 8 * 1024 * 1024) {
+                            let stats = quic_diagnostics.stats();
+                            tracing::debug!(
+                                %peer, local=%local, stream_id=%stream_id,
+                                response_bytes=sent_body_bytes,
+                                request_content_length=?expected,
+                                quic_udp_tx_datagrams=stats.udp_tx.datagrams,
+                                quic_udp_tx_ios=stats.udp_tx.ios,
+                                quic_udp_rx_datagrams=stats.udp_rx.datagrams,
+                                quic_sent_packets=stats.path.sent_packets,
+                                quic_lost_packets=stats.path.lost_packets,
+                                quic_congestion_events=stats.path.congestion_events,
+                                quic_rtt_ms=stats.path.rtt.as_secs_f64() * 1000.0,
+                                quic_cwnd_bytes=stats.path.cwnd,
+                                quic_path_mtu=stats.path.current_mtu,
+                                "HTTP/3 bulk transfer transport snapshot"
+                            );
+                        }
                     });
                 },
                 Ok(None)=>break,