trillium_server_common/
h3.rs

1//! HTTP/3 specific exports
2
3pub mod web_transport;
4use crate::{
5    ArcHandler, QuicConnection, QuicConnectionTrait, QuicEndpoint, QuicTransportReceive,
6    QuicTransportSend, RuntimeTrait,
7};
8use std::sync::Arc;
9use trillium::{Handler, Upgrade};
10use trillium_http::{
11    HttpContext,
12    h3::{H3Connection, H3Error, H3ErrorCode, H3StreamResult, UniStreamResult},
13};
14use web_transport::{WebTransportDispatcher, WebTransportStream};
15
16/// A QUIC stream identifier
17#[derive(Clone, Copy, Debug)]
18pub struct StreamId(u64);
19impl From<StreamId> for u64 {
20    fn from(val: StreamId) -> Self {
21        val.0
22    }
23}
24
25impl From<u64> for StreamId {
26    fn from(value: u64) -> Self {
27        Self(value)
28    }
29}
30
31pub(crate) async fn run_h3<QE: QuicEndpoint>(
32    quic_binding: QE,
33    context: Arc<HttpContext>,
34    handler: ArcHandler<impl Handler>,
35    runtime: impl RuntimeTrait,
36) {
37    let swansong = context.swansong();
38    while let Some(connection) = swansong.interrupt(quic_binding.accept()).await.flatten() {
39        let h3 = H3Connection::new(context.clone());
40        let handler = handler.clone();
41        let runtime = runtime.clone();
42        runtime
43            .clone()
44            .spawn(run_h3_connection(connection, h3, handler, runtime));
45    }
46}
47
48async fn run_h3_connection<QC: QuicConnectionTrait>(
49    connection: QC,
50    h3: Arc<H3Connection>,
51    handler: ArcHandler<impl Handler>,
52    runtime: impl RuntimeTrait,
53) {
54    let wt_dispatcher = h3
55        .context()
56        .config()
57        .webtransport_enabled()
58        .then(WebTransportDispatcher::new);
59
60    log::trace!("new quic connection from {}", connection.remote_address());
61
62    spawn_outbound_control_stream(&connection, &h3, &runtime);
63    spawn_qpack_encoder_stream(&connection, &h3, &runtime);
64    spawn_qpack_decoder_stream(&connection, &h3, &runtime);
65    spawn_inbound_uni_streams(&connection, &h3, &runtime, &wt_dispatcher);
66    handle_inbound_bidi_streams(connection, h3.clone(), handler, runtime, wt_dispatcher).await;
67}
68
69async fn handle_inbound_bidi_streams<QC: QuicConnectionTrait>(
70    connection: QC,
71    h3: Arc<H3Connection>,
72    handler: ArcHandler<impl Handler>,
73    runtime: impl RuntimeTrait,
74    wt_dispatcher: Option<WebTransportDispatcher>,
75) {
76    loop {
77        match h3.swansong().interrupt(connection.accept_bidi()).await {
78            None => {
79                log::trace!("H3 bidi accept loop: interrupted by swansong shutdown");
80                break;
81            }
82            Some(Err(e)) => {
83                log::debug!("H3 bidi accept loop: accept_bidi error: {e}");
84                break;
85            }
86            Some(Ok((stream_id, transport))) => {
87                handle_bidi_stream(
88                    stream_id,
89                    transport,
90                    &h3,
91                    &handler,
92                    &connection,
93                    &runtime,
94                    &wt_dispatcher,
95                );
96            }
97        }
98    }
99
100    h3.shut_down();
101}
102
103fn handle_bidi_stream<QC: QuicConnectionTrait>(
104    stream_id: u64,
105    transport: QC::BidiStream,
106    h3: &Arc<H3Connection>,
107    handler: &ArcHandler<impl Handler>,
108    connection: &QC,
109    runtime: &impl RuntimeTrait,
110    wt_dispatcher: &Option<WebTransportDispatcher>,
111) {
112    log::trace!("H3 bidi stream {stream_id}: spawning handler task");
113    let (h3, handler, connection, wt_dispatcher) = (
114        h3.clone(),
115        handler.clone(),
116        connection.clone(),
117        wt_dispatcher.clone(),
118    );
119
120    runtime.spawn(async move {
121        let handler = &handler;
122        let peer_ip = connection.remote_address().ip();
123        let quic_connection = connection.clone();
124        let wt_dispatcher = wt_dispatcher.clone();
125
126        let handler_fn = {
127            let wt_dispatcher = wt_dispatcher.clone();
128            |mut conn: trillium_http::Conn<_>| async move {
129                conn.set_peer_ip(Some(peer_ip));
130                conn.set_secure(true);
131
132                let state = conn.state_mut();
133                state.insert(quic_connection.clone());
134                state.insert(QuicConnection::from(quic_connection));
135                state.insert(StreamId(stream_id));
136                if let Some(dispatcher) = wt_dispatcher {
137                    state.insert(dispatcher);
138                }
139
140                let conn = handler.run(conn.into()).await;
141                let conn = handler.before_send(conn).await;
142
143                conn.into_inner()
144            }
145        };
146
147        let result = h3
148            .clone()
149            .process_inbound_bidi(transport, handler_fn, stream_id)
150            .await;
151
152        match result {
153            Ok(H3StreamResult::Request(conn)) if conn.should_upgrade() => {
154                let upgrade = Upgrade::from(conn);
155                if handler.has_upgrade(&upgrade) {
156                    log::debug!("upgrading h3 stream");
157                    handler.upgrade(upgrade).await;
158                } else {
159                    log::error!("h3 upgrade specified but no upgrade handler provided");
160                }
161            }
162
163            Ok(H3StreamResult::Request(_)) => {}
164
165            Ok(H3StreamResult::WebTransport {
166                session_id,
167                mut transport,
168                buffer,
169            }) => {
170                if let Some(dispatcher) = &wt_dispatcher {
171                    dispatcher.dispatch(WebTransportStream::Bidi {
172                        session_id,
173                        stream: Box::new(transport),
174                        buffer: buffer.into(),
175                    });
176                } else {
177                    transport.stop(H3ErrorCode::StreamCreationError.into());
178                    transport.reset(H3ErrorCode::StreamCreationError.into());
179                }
180            }
181
182            Err(error) => {
183                log::debug!("H3 bidi stream {stream_id}: error: {error}");
184                handle_h3_error(error, &connection, &h3);
185            }
186        }
187    });
188}
189
190fn spawn_inbound_uni_streams<QC: QuicConnectionTrait>(
191    connection: &QC,
192    h3: &Arc<H3Connection>,
193    runtime: &impl RuntimeTrait,
194    wt_dispatcher: &Option<WebTransportDispatcher>,
195) {
196    let (connection, h3, runtime, wt_dispatcher) = (
197        connection.clone(),
198        h3.clone(),
199        runtime.clone(),
200        wt_dispatcher.clone(),
201    );
202    runtime.clone().spawn(async move {
203        while let Some(Ok((_stream_id, recv))) =
204            h3.swansong().interrupt(connection.accept_uni()).await
205        {
206            let (connection, h3, wt_dispatcher) =
207                (connection.clone(), h3.clone(), wt_dispatcher.clone());
208
209            runtime.spawn(async move {
210                let result = h3.process_inbound_uni(recv).await;
211
212                match result {
213                    Ok(UniStreamResult::Handled) => {}
214                    Ok(UniStreamResult::WebTransport {
215                        session_id,
216                        mut stream,
217                        buffer,
218                    }) => {
219                        if let Some(dispatcher) = &wt_dispatcher {
220                            dispatcher.dispatch(WebTransportStream::Uni {
221                                session_id,
222                                stream: Box::new(stream),
223                                buffer: buffer.into(),
224                            });
225                        } else {
226                            stream.stop(H3ErrorCode::StreamCreationError.into());
227                        }
228                    }
229
230                    Ok(UniStreamResult::Unknown { mut stream, .. }) => {
231                        stream.stop(H3ErrorCode::StreamCreationError.into());
232                    }
233
234                    Err(error) => {
235                        handle_h3_error(error, &connection, &h3);
236                    }
237                }
238            });
239        }
240
241        h3.shut_down();
242    });
243}
244
245fn spawn_qpack_decoder_stream<QC: QuicConnectionTrait>(
246    connection: &QC,
247    h3: &Arc<H3Connection>,
248    runtime: &impl RuntimeTrait,
249) {
250    let (connection, h3) = (connection.clone(), h3.clone());
251
252    runtime.spawn(async move {
253        log::trace!("H3: opening outbound QPACK decoder stream");
254        let stream = match connection.open_uni().await {
255            Ok((_stream_id, stream)) => stream,
256            Err(err) => {
257                log::error!("H3: open_uni for QPACK decoder stream failed: {err:?}");
258                h3.shut_down();
259                return;
260            }
261        };
262
263        let result = h3.run_decoder(stream).await;
264
265        if let Err(error) = result {
266            handle_h3_error(error, &connection, &h3);
267        }
268
269        h3.shut_down();
270    });
271}
272
273fn spawn_qpack_encoder_stream<QC: QuicConnectionTrait>(
274    connection: &QC,
275    h3: &Arc<H3Connection>,
276    runtime: &impl RuntimeTrait,
277) {
278    let (connection, h3) = (connection.clone(), h3.clone());
279    runtime.spawn(async move {
280        log::trace!("H3: opening outbound QPACK encoder stream");
281        let stream = match connection.open_uni().await {
282            Ok((_stream_id, stream)) => stream,
283            Err(err) => {
284                log::error!("H3: open_uni for QPACK encoder stream failed: {err:?}");
285                h3.shut_down();
286                return;
287            }
288        };
289
290        let result = h3.run_encoder(stream).await;
291
292        if let Err(error) = result {
293            handle_h3_error(error, &connection, &h3);
294        }
295
296        h3.shut_down();
297    });
298}
299
300fn spawn_outbound_control_stream<QC: QuicConnectionTrait>(
301    connection: &QC,
302    h3: &Arc<H3Connection>,
303    runtime: &impl RuntimeTrait,
304) {
305    let (connection, h3) = (connection.clone(), h3.clone());
306    runtime.spawn(async move {
307        log::trace!("H3: opening outbound control stream");
308        let stream = match connection.open_uni().await {
309            Ok((_stream_id, stream)) => stream,
310            Err(err) => {
311                log::error!("H3: open_uni for outbound control stream failed: {err:?}");
312                h3.shut_down();
313                return;
314            }
315        };
316
317        let result = h3.run_outbound_control(stream).await;
318
319        if let Err(error) = result {
320            handle_h3_error(error, &connection, &h3);
321        }
322
323        h3.shut_down();
324    });
325}
326
327fn handle_h3_error(error: H3Error, connection: &impl QuicConnectionTrait, h3: &H3Connection) {
328    log::debug!("H3 error: {error}");
329    if let H3Error::Protocol(code) = error
330        && code.is_connection_error()
331    {
332        connection.close(code.into(), code.reason().as_bytes());
333        h3.shut_down();
334    }
335}
trillium_server_common/h3.rs

trillium_server_common/
h3.rs
