mechanics_http_server/
server.rs

1//! HTTP/3 QUIC server lifecycle.
2
3use std::future::Future;
4use std::net::SocketAddr;
5use std::pin::Pin;
6use std::sync::Arc;
7use std::task::{Context, Poll};
8
9use bytes::Bytes;
10use http::{Request, Response};
11use http_body::Body;
12use quinn::crypto::rustls::QuicServerConfig;
13use tokio::sync::watch;
14use tokio::task::{JoinHandle, JoinSet};
15use tower::{Service, ServiceExt};
16
17use crate::body::H3RequestBody;
18use crate::error::{Error, Result};
19use crate::zero_rtt::{default_zero_rtt_methods, is_zero_rtt_safe};
20
21/// Optional HTTP/3 listener that runs alongside an existing TCP+TLS server.
22#[derive(Clone, Debug)]
23pub struct Http3Server {
24    config: Http3ServerConfig,
25}
26
27/// Configuration for the HTTP/3 listener.
28#[derive(Clone, Debug)]
29pub struct Http3ServerConfig {
30    /// UDP bind address. `None` makes [`Http3Server::start`] fully inert.
31    pub bind_h3: Option<SocketAddr>,
32    /// Default `Alt-Svc` max-age value, in seconds.
33    pub alt_svc_max_age_secs: u64,
34    /// Methods allowed by the 0-RTT replay-safety policy.
35    pub zero_rtt_idempotent_methods: Vec<http::Method>,
36}
37
38impl Default for Http3ServerConfig {
39    fn default() -> Self {
40        Self {
41            bind_h3: None,
42            alt_svc_max_age_secs: 86_400,
43            zero_rtt_idempotent_methods: default_zero_rtt_methods(),
44        }
45    }
46}
47
48impl Http3Server {
49    /// Build a server value without starting it.
50    pub fn new(config: Http3ServerConfig) -> Self {
51        Self { config }
52    }
53
54    /// Start the HTTP/3 listener and route requests into `service`.
55    ///
56    /// The service receives HTTP/3 request headers plus a streaming
57    /// [`H3RequestBody`] and returns response headers plus any streaming
58    /// `http_body` response body whose data chunks are [`Bytes`].
59    pub fn start<S, RespBody>(
60        self,
61        service: S,
62        tls_cert_chain: Vec<rustls::pki_types::CertificateDer<'static>>,
63        tls_private_key: rustls::pki_types::PrivateKeyDer<'static>,
64    ) -> Result<Http3Handle>
65    where
66        S: Service<Request<H3RequestBody>, Response = Response<RespBody>> + Clone + Send + 'static,
67        S::Future: Send + 'static,
68        S::Error: std::fmt::Display + Send + Sync + 'static,
69        RespBody: Body<Data = Bytes> + Send + 'static,
70        RespBody::Error: std::fmt::Display + Send + Sync + 'static,
71    {
72        let Some(bind_addr) = self.config.bind_h3 else {
73            return Ok(Http3Handle::inert());
74        };
75
76        let server_config = quic_server_config(tls_cert_chain, tls_private_key)?;
77        let endpoint = quinn::Endpoint::server(server_config, bind_addr)
78            .map_err(|e| Error::BindFailed(e.to_string()))?;
79        let local_addr = endpoint
80            .local_addr()
81            .map_err(|e| Error::BindFailed(e.to_string()))?;
82
83        let (shutdown_tx, shutdown_rx) = watch::channel(false);
84        let allowed_zero_rtt_methods = Arc::new(self.config.zero_rtt_idempotent_methods);
85        let task_endpoint = endpoint.clone();
86        let accept_task = tokio::spawn(async move {
87            accept_loop(
88                task_endpoint,
89                shutdown_rx,
90                service,
91                allowed_zero_rtt_methods,
92            )
93            .await
94        });
95
96        Ok(Http3Handle::running(
97            endpoint,
98            local_addr,
99            shutdown_tx,
100            accept_task,
101        ))
102    }
103}
104
105/// Handle returned by [`Http3Server::start`].
106///
107/// Awaiting the handle waits for the accept loop to finish. Dropping
108/// it closes the QUIC endpoint and signals the accept loop to stop
109/// accepting new connections while existing connection tasks drain.
110pub struct Http3Handle {
111    state: Http3HandleState,
112}
113
114enum Http3HandleState {
115    Inert,
116    Running {
117        endpoint: quinn::Endpoint,
118        local_addr: SocketAddr,
119        shutdown_tx: watch::Sender<bool>,
120        accept_task: JoinHandle<Result<()>>,
121    },
122    Done,
123}
124
125impl Http3Handle {
126    fn inert() -> Self {
127        Self {
128            state: Http3HandleState::Inert,
129        }
130    }
131
132    fn running(
133        endpoint: quinn::Endpoint,
134        local_addr: SocketAddr,
135        shutdown_tx: watch::Sender<bool>,
136        accept_task: JoinHandle<Result<()>>,
137    ) -> Self {
138        Self {
139            state: Http3HandleState::Running {
140                endpoint,
141                local_addr,
142                shutdown_tx,
143                accept_task,
144            },
145        }
146    }
147
148    /// Return whether this handle is the no-op `bind_h3 = None` variant.
149    pub fn is_inert(&self) -> bool {
150        matches!(self.state, Http3HandleState::Inert | Http3HandleState::Done)
151    }
152
153    /// Return the bound UDP address for a running HTTP/3 listener.
154    pub fn local_addr(&self) -> Option<SocketAddr> {
155        match &self.state {
156            Http3HandleState::Running { local_addr, .. } => Some(*local_addr),
157            Http3HandleState::Inert | Http3HandleState::Done => None,
158        }
159    }
160
161    /// Return whether the accept-loop task has finished.
162    pub fn is_finished(&self) -> bool {
163        match &self.state {
164            Http3HandleState::Inert | Http3HandleState::Done => true,
165            Http3HandleState::Running { accept_task, .. } => accept_task.is_finished(),
166        }
167    }
168
169    /// Signal shutdown without consuming the handle.
170    pub fn shutdown(&self) {
171        if let Http3HandleState::Running {
172            endpoint,
173            shutdown_tx,
174            ..
175        } = &self.state
176        {
177            let _ = shutdown_tx.send(true);
178            endpoint.close(0_u32.into(), b"shutdown");
179        }
180    }
181}
182
183impl Future for Http3Handle {
184    type Output = Result<()>;
185
186    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
187        let this = self.get_mut();
188        match &mut this.state {
189            Http3HandleState::Inert | Http3HandleState::Done => Poll::Ready(Ok(())),
190            Http3HandleState::Running { accept_task, .. } => match Pin::new(accept_task).poll(cx) {
191                Poll::Pending => Poll::Pending,
192                Poll::Ready(joined) => {
193                    this.state = Http3HandleState::Done;
194                    Poll::Ready(
195                        joined.map_err(|e| Error::Internal(format!("accept task failed: {e}")))?,
196                    )
197                }
198            },
199        }
200    }
201}
202
203impl Drop for Http3Handle {
204    fn drop(&mut self) {
205        self.shutdown();
206    }
207}
208
209fn quic_server_config(
210    tls_cert_chain: Vec<rustls::pki_types::CertificateDer<'static>>,
211    tls_private_key: rustls::pki_types::PrivateKeyDer<'static>,
212) -> Result<quinn::ServerConfig> {
213    if tls_cert_chain.is_empty() {
214        return Err(Error::InvalidTls("empty certificate chain".to_owned()));
215    }
216
217    let provider = Arc::new(rustls::crypto::aws_lc_rs::default_provider());
218    let mut tls_config = rustls::ServerConfig::builder_with_provider(provider)
219        .with_safe_default_protocol_versions()
220        .map_err(|e| Error::InvalidTls(e.to_string()))?
221        .with_no_client_auth()
222        .with_single_cert(tls_cert_chain, tls_private_key)
223        .map_err(|e| Error::InvalidTls(e.to_string()))?;
224    tls_config.alpn_protocols = vec![b"h3".to_vec()];
225    tls_config.max_early_data_size = 0;
226
227    let quic_config =
228        QuicServerConfig::try_from(tls_config).map_err(|e| Error::InvalidTls(e.to_string()))?;
229    Ok(quinn::ServerConfig::with_crypto(Arc::new(quic_config)))
230}
231
232async fn accept_loop<S, RespBody>(
233    endpoint: quinn::Endpoint,
234    mut shutdown_rx: watch::Receiver<bool>,
235    service: S,
236    allowed_zero_rtt_methods: Arc<Vec<http::Method>>,
237) -> Result<()>
238where
239    S: Service<Request<H3RequestBody>, Response = Response<RespBody>> + Clone + Send + 'static,
240    S::Future: Send + 'static,
241    S::Error: std::fmt::Display + Send + Sync + 'static,
242    RespBody: Body<Data = Bytes> + Send + 'static,
243    RespBody::Error: std::fmt::Display + Send + Sync + 'static,
244{
245    let mut connections = JoinSet::new();
246
247    loop {
248        tokio::select! {
249            changed = shutdown_rx.changed() => {
250                if changed.is_ok() && *shutdown_rx.borrow() {
251                    break;
252                }
253                if changed.is_err() {
254                    break;
255                }
256            }
257            incoming = endpoint.accept() => {
258                let Some(incoming) = incoming else {
259                    break;
260                };
261                let connecting = incoming
262                    .accept()
263                    .map_err(|e| Error::Internal(format!("incoming connection rejected: {e}")))?;
264                let connection_service = service.clone();
265                let connection_allowed = allowed_zero_rtt_methods.clone();
266                connections.spawn(async move {
267                    handle_connection(connecting, connection_service, connection_allowed).await
268                });
269            }
270            joined = connections.join_next(), if !connections.is_empty() => {
271                match joined {
272                    Some(Ok(Ok(()))) | None => {}
273                    Some(Ok(Err(e))) => tracing::warn!("HTTP/3 connection failed: {e}"),
274                    Some(Err(e)) => tracing::warn!("HTTP/3 connection task failed: {e}"),
275                }
276            }
277        }
278    }
279
280    while let Some(joined) = connections.join_next().await {
281        match joined {
282            Ok(Ok(())) => {}
283            Ok(Err(e)) => tracing::warn!("HTTP/3 connection failed during shutdown: {e}"),
284            Err(e) => tracing::warn!("HTTP/3 connection task failed during shutdown: {e}"),
285        }
286    }
287
288    Ok(())
289}
290
291async fn handle_connection<S, RespBody>(
292    connecting: quinn::Connecting,
293    service: S,
294    allowed_zero_rtt_methods: Arc<Vec<http::Method>>,
295) -> Result<()>
296where
297    S: Service<Request<H3RequestBody>, Response = Response<RespBody>> + Clone + Send + 'static,
298    S::Future: Send + 'static,
299    S::Error: std::fmt::Display + Send + Sync + 'static,
300    RespBody: Body<Data = Bytes> + Send + 'static,
301    RespBody::Error: std::fmt::Display + Send + Sync + 'static,
302{
303    let connection = connecting
304        .await
305        .map_err(|e| Error::Internal(format!("QUIC handshake failed: {e}")))?;
306    let quic = h3_quinn::Connection::new(connection);
307    let mut h3_conn = h3::server::Connection::new(quic)
308        .await
309        .map_err(|e| Error::Internal(format!("HTTP/3 connection failed: {e}")))?;
310    let mut streams = JoinSet::new();
311
312    loop {
313        match h3_conn.accept().await {
314            Ok(Some(resolver)) => {
315                let stream_service = service.clone();
316                let stream_allowed = allowed_zero_rtt_methods.clone();
317                streams.spawn(async move {
318                    if let Err(e) =
319                        handle_request(resolver, stream_service, stream_allowed.as_slice()).await
320                    {
321                        tracing::warn!("HTTP/3 request failed: {e}");
322                    }
323                });
324            }
325            Ok(None) => break,
326            Err(e) => return Err(Error::Internal(format!("HTTP/3 accept failed: {e}"))),
327        }
328
329        while let Some(joined) = streams.try_join_next() {
330            if let Err(e) = joined {
331                tracing::warn!("HTTP/3 request task failed: {e}");
332            }
333        }
334    }
335
336    while let Some(joined) = streams.join_next().await {
337        if let Err(e) = joined {
338            tracing::warn!("HTTP/3 request task failed during drain: {e}");
339        }
340    }
341
342    Ok(())
343}
344
345async fn handle_request<S, RespBody>(
346    resolver: h3::server::RequestResolver<h3_quinn::Connection, Bytes>,
347    mut service: S,
348    allowed_zero_rtt_methods: &[http::Method],
349) -> Result<()>
350where
351    S: Service<Request<H3RequestBody>, Response = Response<RespBody>> + Send + 'static,
352    S::Future: Send + 'static,
353    S::Error: std::fmt::Display + Send + Sync + 'static,
354    RespBody: Body<Data = Bytes> + Send + 'static,
355    RespBody::Error: std::fmt::Display + Send + Sync + 'static,
356{
357    let (request, stream) = resolver
358        .resolve_request()
359        .await
360        .map_err(|e| Error::Internal(format!("HTTP/3 request resolution failed: {e}")))?;
361    let (mut send_stream, recv_stream) = stream.split();
362
363    if !is_zero_rtt_safe(request.method(), allowed_zero_rtt_methods) {
364        tracing::trace!(
365            method = %request.method(),
366            "HTTP/3 request method is outside the configured 0-RTT allow-list"
367        );
368    }
369
370    let (parts, ()) = request.into_parts();
371    let request = Request::from_parts(parts, H3RequestBody::new(recv_stream));
372    let response = service
373        .ready()
374        .await
375        .map_err(|e| Error::Internal(format!("HTTP/3 service was not ready: {e}")))?
376        .call(request)
377        .await
378        .map_err(|e| Error::Internal(format!("HTTP/3 service failed: {e}")))?;
379
380    let (parts, body) = response.into_parts();
381    send_stream
382        .send_response(Response::from_parts(parts, ()))
383        .await
384        .map_err(|e| Error::Internal(format!("HTTP/3 response headers failed: {e}")))?;
385    send_response_body(&mut send_stream, body).await?;
386    send_stream
387        .finish()
388        .await
389        .map_err(|e| Error::Internal(format!("HTTP/3 response finish failed: {e}")))?;
390
391    Ok(())
392}
393
394async fn send_response_body<RespBody>(
395    stream: &mut h3::server::RequestStream<h3_quinn::SendStream<Bytes>, Bytes>,
396    body: RespBody,
397) -> Result<()>
398where
399    RespBody: Body<Data = Bytes> + Send + 'static,
400    RespBody::Error: std::fmt::Display + Send + Sync + 'static,
401{
402    let mut body = std::pin::pin!(body);
403    while let Some(frame) = std::future::poll_fn(|cx| body.as_mut().poll_frame(cx)).await {
404        let frame =
405            frame.map_err(|e| Error::Internal(format!("HTTP/3 response body failed: {e}")))?;
406        match frame.into_data() {
407            Ok(bytes) => {
408                if !bytes.is_empty() {
409                    stream.send_data(bytes).await.map_err(|e| {
410                        Error::Internal(format!("HTTP/3 response body failed: {e}"))
411                    })?;
412                }
413            }
414            Err(frame) => {
415                let trailers = frame.into_trailers().map_err(|_| {
416                    Error::Internal("HTTP/3 response body yielded an unknown frame".to_owned())
417                })?;
418                stream.send_trailers(trailers).await.map_err(|e| {
419                    Error::Internal(format!("HTTP/3 response trailers failed: {e}"))
420                })?;
421            }
422        }
423    }
424
425    Ok(())
426}
mechanics_http_server/server.rs

mechanics_http_server/
server.rs
