scuffle_http/backend/h3/
mod.rs

1//! HTTP3 backend.
2use std::fmt::Debug;
3use std::io;
4use std::net::SocketAddr;
5use std::sync::Arc;
6
7use body::QuicIncomingBody;
8use scuffle_context::ContextFutExt;
9#[cfg(feature = "tracing")]
10use tracing::Instrument;
11use utils::copy_response_body;
12
13use crate::error::HttpError;
14use crate::service::{HttpService, HttpServiceFactory};
15
16pub mod body;
17mod utils;
18
19/// A backend that handles incoming HTTP3 connections.
20///
21/// This is used internally by the [`HttpServer`](crate::server::HttpServer) but can be used directly if preferred.
22///
23/// Call [`run`](Http3Backend::run) to start the server.
24#[derive(bon::Builder, Debug, Clone)]
25pub struct Http3Backend<F> {
26    /// The [`scuffle_context::Context`] this server will live by.
27    #[builder(default = scuffle_context::Context::global())]
28    ctx: scuffle_context::Context,
29    /// The number of worker tasks to spawn for each server backend.
30    #[builder(default = 1)]
31    worker_tasks: usize,
32    /// The service factory that will be used to create new services.
33    service_factory: F,
34    /// The address to bind to.
35    ///
36    /// Use `[::]` for a dual-stack listener.
37    /// For example, use `[::]:80` to bind to port 80 on both IPv4 and IPv6.
38    bind: SocketAddr,
39    /// rustls config.
40    ///
41    /// Use this field to set the server into TLS mode.
42    /// It will only accept TLS connections when this is set.
43    rustls_config: rustls::ServerConfig,
44}
45
46impl<F> Http3Backend<F>
47where
48    F: HttpServiceFactory + Clone + Send + 'static,
49    F::Error: std::error::Error + Send,
50    F::Service: Clone + Send + 'static,
51    <F::Service as HttpService>::Error: std::error::Error + Send + Sync,
52    <F::Service as HttpService>::ResBody: Send,
53    <<F::Service as HttpService>::ResBody as http_body::Body>::Data: Send,
54    <<F::Service as HttpService>::ResBody as http_body::Body>::Error: std::error::Error + Send + Sync,
55{
56    /// Run the HTTP3 server
57    ///
58    /// This function will bind to the address specified in `bind`, listen for incoming connections and handle requests.
59    #[cfg_attr(feature = "tracing", tracing::instrument(skip_all, fields(bind = %self.bind)))]
60    pub async fn run(mut self) -> Result<(), HttpError<F>> {
61        #[cfg(feature = "tracing")]
62        tracing::debug!("starting server");
63
64        // not quite sure why this is necessary but it is
65        self.rustls_config.max_early_data_size = u32::MAX;
66        let crypto = h3_quinn::quinn::crypto::rustls::QuicServerConfig::try_from(self.rustls_config)?;
67        let server_config = h3_quinn::quinn::ServerConfig::with_crypto(Arc::new(crypto));
68
69        // Bind the UDP socket
70        let socket = std::net::UdpSocket::bind(self.bind)?;
71
72        // Runtime for the quinn endpoint
73        let runtime = h3_quinn::quinn::default_runtime().ok_or_else(|| io::Error::other("no async runtime found"))?;
74
75        // Create a child context for the workers so we can shut them down if one of them fails without shutting down the main context
76        let (worker_ctx, worker_handler) = self.ctx.new_child();
77
78        let workers = (0..self.worker_tasks).map(|_n| {
79            let ctx = worker_ctx.clone();
80            let service_factory = self.service_factory.clone();
81            let server_config = server_config.clone();
82            let socket = socket.try_clone().expect("failed to clone socket");
83            let runtime = Arc::clone(&runtime);
84
85            let worker_fut = async move {
86                let endpoint = h3_quinn::quinn::Endpoint::new(
87                    h3_quinn::quinn::EndpointConfig::default(),
88                    Some(server_config),
89                    socket,
90                    runtime,
91                )?;
92
93                #[cfg(feature = "tracing")]
94                tracing::trace!("waiting for connections");
95
96                while let Some(Some(new_conn)) = endpoint.accept().with_context(&ctx).await {
97                    let mut service_factory = service_factory.clone();
98                    let ctx = ctx.clone();
99
100                    tokio::spawn(async move {
101                        let _res: Result<_, HttpError<F>> = async move {
102                            let Some(conn) = new_conn.with_context(&ctx).await.transpose()? else {
103                                #[cfg(feature = "tracing")]
104                                tracing::trace!("context done while accepting connection");
105                                return Ok(());
106                            };
107                            let addr = conn.remote_address();
108
109                            #[cfg(feature = "tracing")]
110                            tracing::debug!(addr = %addr, "accepted quic connection");
111
112                            let connection_fut = async move {
113                                let Some(mut h3_conn) = h3::server::Connection::new(h3_quinn::Connection::new(conn))
114                                    .with_context(&ctx)
115                                    .await
116                                    .transpose()?
117                                else {
118                                    #[cfg(feature = "tracing")]
119                                    tracing::trace!("context done while establishing connection");
120                                    return Ok(());
121                                };
122
123                                // make a new service for this connection
124                                let http_service = service_factory
125                                    .new_service(addr)
126                                    .await
127                                    .map_err(|e| HttpError::ServiceFactoryError(e))?;
128
129                                loop {
130                                    match h3_conn.accept().with_context(&ctx).await {
131                                        Some(Ok(Some(resolver))) => {
132                                            let (req, stream) = match resolver.resolve_request().await {
133                                                Ok(r) => r,
134                                                Err(_err) => {
135                                                    #[cfg(feature = "tracing")]
136                                                    tracing::warn!("error on accept: {}", _err);
137                                                    continue;
138                                                }
139                                            };
140
141                                            #[cfg(feature = "tracing")]
142                                            tracing::debug!(method = %req.method(), uri = %req.uri(), "received request");
143
144                                            let (mut send, recv) = stream.split();
145
146                                            let size_hint = req
147                                                .headers()
148                                                .get(http::header::CONTENT_LENGTH)
149                                                .and_then(|len| len.to_str().ok().and_then(|x| x.parse().ok()));
150                                            let body = QuicIncomingBody::new(recv, size_hint);
151                                            let req = req.map(|_| crate::body::IncomingBody::from(body));
152
153                                            let ctx = ctx.clone();
154                                            let mut http_service = http_service.clone();
155                                            tokio::spawn(async move {
156                                                let _res: Result<_, HttpError<F>> = async move {
157                                                    let resp = http_service
158                                                        .call(req)
159                                                        .await
160                                                        .map_err(|e| HttpError::ServiceError(e))?;
161                                                    let (parts, body) = resp.into_parts();
162
163                                                    send.send_response(http::Response::from_parts(parts, ())).await?;
164                                                    copy_response_body(send, body).await?;
165
166                                                    Ok(())
167                                                }
168                                                .await;
169
170                                                #[cfg(feature = "tracing")]
171                                                if let Err(e) = _res {
172                                                    tracing::warn!(err = %e, "error handling request");
173                                                }
174
175                                                // This moves the context into the async block because it is dropped here
176                                                drop(ctx);
177                                            });
178                                        }
179                                        // indicating no more streams to be received
180                                        Some(Ok(None)) => {
181                                            break;
182                                        }
183                                        Some(Err(err)) => return Err(err.into()),
184                                        // context is done
185                                        None => {
186                                            #[cfg(feature = "tracing")]
187                                            tracing::trace!("context done, stopping connection loop");
188                                            break;
189                                        }
190                                    }
191                                }
192
193                                #[cfg(feature = "tracing")]
194                                tracing::trace!("connection closed");
195
196                                Ok(())
197                            };
198
199                            #[cfg(feature = "tracing")]
200                            let connection_fut = connection_fut.instrument(tracing::trace_span!("connection", addr = %addr));
201
202                            connection_fut.await
203                        }
204                        .await;
205
206                        #[cfg(feature = "tracing")]
207                        if let Err(err) = _res {
208                            tracing::warn!(err = %err, "error handling connection");
209                        }
210                    });
211                }
212
213                // shut down gracefully
214                // wait for connections to be closed before exiting
215                endpoint.wait_idle().await;
216
217                Ok::<_, crate::error::HttpError<F>>(())
218            };
219
220            #[cfg(feature = "tracing")]
221            let worker_fut = worker_fut.instrument(tracing::trace_span!("worker", n = _n));
222
223            tokio::spawn(worker_fut)
224        });
225
226        if let Err(_e) = futures::future::try_join_all(workers).await {
227            #[cfg(feature = "tracing")]
228            tracing::error!(err = %_e, "error running workers");
229        }
230
231        drop(worker_ctx);
232        worker_handler.shutdown().await;
233
234        #[cfg(feature = "tracing")]
235        tracing::debug!("all workers finished");
236
237        Ok(())
238    }
239}
scuffle_http/backend/h3/mod.rs

scuffle_http/backend/h3/
mod.rs
