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((req, stream)))) => {
132                                            #[cfg(feature = "tracing")]
133                                            tracing::debug!(method = %req.method(), uri = %req.uri(), "received request");
134
135                                            let (mut send, recv) = stream.split();
136
137                                            let size_hint = req
138                                                .headers()
139                                                .get(http::header::CONTENT_LENGTH)
140                                                .and_then(|len| len.to_str().ok().and_then(|x| x.parse().ok()));
141                                            let body = QuicIncomingBody::new(recv, size_hint);
142                                            let req = req.map(|_| crate::body::IncomingBody::from(body));
143
144                                            let ctx = ctx.clone();
145                                            let mut http_service = http_service.clone();
146                                            tokio::spawn(async move {
147                                                let _res: Result<_, HttpError<F>> = async move {
148                                                    let resp = http_service
149                                                        .call(req)
150                                                        .await
151                                                        .map_err(|e| HttpError::ServiceError(e))?;
152                                                    let (parts, body) = resp.into_parts();
153
154                                                    send.send_response(http::Response::from_parts(parts, ())).await?;
155                                                    copy_response_body(send, body).await?;
156
157                                                    Ok(())
158                                                }
159                                                .await;
160
161                                                #[cfg(feature = "tracing")]
162                                                if let Err(e) = _res {
163                                                    tracing::warn!(err = %e, "error handling request");
164                                                }
165
166                                                // This moves the context into the async block because it is dropped here
167                                                drop(ctx);
168                                            });
169                                        }
170                                        // indicating no more streams to be received
171                                        Some(Ok(None)) => {
172                                            break;
173                                        }
174                                        Some(Err(err)) => match err.get_error_level() {
175                                            h3::error::ErrorLevel::ConnectionError => return Err(err.into()),
176                                            h3::error::ErrorLevel::StreamError => {
177                                                #[cfg(feature = "tracing")]
178                                                tracing::warn!("error on accept: {}", err);
179                                                continue;
180                                            }
181                                        },
182                                        // context is done
183                                        None => {
184                                            #[cfg(feature = "tracing")]
185                                            tracing::trace!("context done, stopping connection loop");
186                                            break;
187                                        }
188                                    }
189                                }
190
191                                #[cfg(feature = "tracing")]
192                                tracing::trace!("connection closed");
193
194                                Ok(())
195                            };
196
197                            #[cfg(feature = "tracing")]
198                            let connection_fut = connection_fut.instrument(tracing::trace_span!("connection", addr = %addr));
199
200                            connection_fut.await
201                        }
202                        .await;
203
204                        #[cfg(feature = "tracing")]
205                        if let Err(err) = _res {
206                            tracing::warn!(err = %err, "error handling connection");
207                        }
208                    });
209                }
210
211                // shut down gracefully
212                // wait for connections to be closed before exiting
213                endpoint.wait_idle().await;
214
215                Ok::<_, crate::error::HttpError<F>>(())
216            };
217
218            #[cfg(feature = "tracing")]
219            let worker_fut = worker_fut.instrument(tracing::trace_span!("worker", n = _n));
220
221            tokio::spawn(worker_fut)
222        });
223
224        if let Err(_e) = futures::future::try_join_all(workers).await {
225            #[cfg(feature = "tracing")]
226            tracing::error!(err = %_e, "error running workers");
227        }
228
229        drop(worker_ctx);
230        worker_handler.shutdown().await;
231
232        #[cfg(feature = "tracing")]
233        tracing::debug!("all workers finished");
234
235        Ok(())
236    }
237}
scuffle_http/backend/h3/mod.rs

scuffle_http/backend/h3/
mod.rs
