tokio-quiche: add ServerH3Event::Headers

Priority signals take the form of HTTP headers and/or frames. Some clients
only send frames, while some clients can both. At the point in time a server
reads request HEADERS, a client may sent a PRIORITY_UPDATE frame that
should take precedence.

With this change, we read take the last PRIORITY_UPDATE processed by quiche,
and pass it alongside IncomingH3Headers in a new ServerH3Event:Headers event
so that Tokio-Quiche-based servers can use it if they wish.

Author: LPardue

tokio-quiche/examples/async_http3_server/server.rs

@@ -44,6 +44,7 @@ use tokio_quiche::http3::driver::H3Event;
 use tokio_quiche::http3::driver::IncomingH3Headers;
 use tokio_quiche::http3::driver::OutboundFrame;
 use tokio_quiche::http3::driver::OutboundFrameSender;
+use tokio_quiche::http3::driver::RawPriorityValue;
 use tokio_quiche::http3::driver::ServerEventStream;
 use tokio_quiche::http3::driver::ServerH3Event;
 use tokio_quiche::BoxError;
@@ -102,20 +103,14 @@ where
     ) -> QuicResult<()> {
         loop {
             match h3_event_receiver.recv().await {
-                Some(event) => self.handle_h3_event(event).await?,
+                Some(event) => self.handle_server_h3_event(event).await?,
                 None => return Ok(()), /* The sender was dropped, implying
                                         * connection was terminated */
             }
         }
     }
 
-    /// Handle a [`ServerH3Event`].
-    ///
-    /// For simplicity's sake, we only handle a couple of events here.
-    // TODO(evanrittenhouse): support POST requests
-    async fn handle_h3_event(&mut self, event: ServerH3Event) -> QuicResult<()> {
-        let ServerH3Event::Core(event) = event;
-
+    fn handle_h3_event(event: H3Event) -> QuicResult<()> {
         match event {
             // Received an explicit connection level error. Not much to do here.
             H3Event::ConnectionError(err) => QuicResult::Err(Box::new(err)),
@@ -130,15 +125,30 @@ where
                 QuicResult::Err(err)
             },
 
-            // Received headers for a new stream from the H3Driver.
-            H3Event::IncomingHeaders(headers) => {
-                self.handle_incoming_headers(headers).await;
-
+            _ => {
+                log::info!("received unhandled event: {event:?}");
                 Ok(())
             },
+        }
+    }
 
-            _ => {
-                log::info!("received unhandled event: {event:?}");
+    /// Handle a [`ServerH3Event`].
+    ///
+    /// For simplicity's sake, we only handle a couple of events here.
+    // TODO(evanrittenhouse): support POST requests
+    async fn handle_server_h3_event(
+        &mut self, event: ServerH3Event,
+    ) -> QuicResult<()> {
+        match event {
+            ServerH3Event::Core(event) => Self::handle_h3_event(event),
+
+            ServerH3Event::Headers {
+                incoming_headers,
+                priority,
+            } => {
+                // Received headers for a new stream from the H3Driver.
+                self.handle_incoming_headers(incoming_headers, priority)
+                    .await;
                 Ok(())
             },
         }
@@ -149,7 +159,10 @@ where
     /// This function transforms the incoming headers into a [`Request`],
     /// creating the proper response body if requested. It then spawns a
     /// Tokio task which calls the `service_fn` on the [`Request`].
-    async fn handle_incoming_headers(&mut self, headers: IncomingH3Headers) {
+    async fn handle_incoming_headers(
+        &mut self, headers: IncomingH3Headers,
+        _priority: Option<RawPriorityValue>,
+    ) {
         log::info!("received headers: {:?}", &headers);
 
         let IncomingH3Headers {
@@ -168,6 +181,7 @@ where
 
         let service_fn = Arc::clone(&self.service_fn);
 
+        // TODO: use the _priority input parameter in request handling
         tokio::spawn(async move {
             Self::handle_request(service_fn, req, frame_sender).await;
         });

tokio-quiche/src/http3/driver/mod.rs

@@ -84,6 +84,7 @@ pub use self::client::ClientH3Driver;
 pub use self::client::ClientH3Event;
 pub use self::client::ClientRequestSender;
 pub use self::client::NewClientRequest;
+pub use self::server::RawPriorityValue;
 pub use self::server::ServerEventStream;
 pub use self::server::ServerH3Command;
 pub use self::server::ServerH3Controller;

tokio-quiche/src/http3/driver/server.rs

@@ -24,6 +24,7 @@
 // NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 // SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
+use std::ops::Deref;
 use std::sync::Arc;
 
 use tokio::sync::mpsc;
@@ -60,15 +61,44 @@ pub type ServerH3Controller = H3Controller<ServerHooks>;
 /// transfer data.
 pub type ServerEventStream = mpsc::UnboundedReceiver<ServerH3Event>;
 
+#[derive(Clone, Debug)]
+pub struct RawPriorityValue(Vec<u8>);
+
+impl From<Vec<u8>> for RawPriorityValue {
+    fn from(value: Vec<u8>) -> Self {
+        RawPriorityValue(value)
+    }
+}
+
+impl Deref for RawPriorityValue {
+    type Target = [u8];
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
 /// Events produced by [ServerH3Driver].
 #[derive(Debug)]
 pub enum ServerH3Event {
     Core(H3Event),
+
+    Headers {
+        incoming_headers: IncomingH3Headers,
+        /// The latest PRIORITY_UPDATE frame value, if any.
+        priority: Option<RawPriorityValue>,
+    },
 }
 
 impl From<H3Event> for ServerH3Event {
     fn from(ev: H3Event) -> Self {
-        Self::Core(ev)
+        match ev {
+            H3Event::IncomingHeaders(incoming_headers) => Self::Headers {
+                incoming_headers,
+                priority: None,
+            },
+            _ => Self::Core(ev),
+        }
     }
 }
 
@@ -130,6 +160,12 @@ impl ServerHooks {
             stream_ctx.associated_dgram_flow_id = Some(flow_id);
         }
 
+        let latest_priority_update: Option<RawPriorityValue> = driver
+            .conn_mut()?
+            .take_last_priority_update(stream_id)
+            .ok()
+            .map(|v| v.into());
+
         let headers = IncomingH3Headers {
             stream_id,
             headers,
@@ -146,7 +182,10 @@ impl ServerHooks {
 
         driver
             .h3_event_sender
-            .send(H3Event::IncomingHeaders(headers).into())
+            .send(ServerH3Event::Headers {
+                incoming_headers: headers,
+                priority: latest_priority_update,
+            })
             .map_err(|_| H3ConnectionError::ControllerWentAway)?;
         driver.hooks.requests += 1;
 

tokio-quiche/tests/fixtures/mod.rs

@@ -133,21 +133,25 @@ pub async fn serve_connection_details(
 
     loop {
         select! {
-            Some(frame) = event_rx.recv() => {
-                let ServerH3Event::Core(frame) = frame;
-                match frame {
-                    H3Event::IncomingSettings {..} | H3Event::BodyBytesReceived { .. } | H3Event::StreamClosed { .. } => {},
-                    H3Event::IncomingHeaders(headers) => {
+            Some(event) = event_rx.recv() => {
+                match event {
+                    ServerH3Event::Core(event) => {
+                        match event {
+                            H3Event::IncomingSettings {..} | H3Event::BodyBytesReceived { .. } | H3Event::StreamClosed { .. } | H3Event::IncomingHeaders(..) => {},
+                            H3Event::ConnectionError(err) => { break Err(err.into()); }
+                            H3Event::ConnectionShutdown(Some(err)) => { break Err(err.into()); }
+                            _ => unreachable!()
+                        }
+                    }
+
+                    ServerH3Event::Headers{ incoming_headers, ..} => {
                         let IncomingH3Headers {
-                            stream_id, headers, send, recv, ..
-                        } = headers;
+                                stream_id, headers, send, recv, ..
+                            } = incoming_headers;
 
-                        request_counter.fetch_add(1, Ordering::SeqCst);
-                        request_futs.push(handle_forwarded_headers_frame(stream_id, headers, send, recv));
+                            request_counter.fetch_add(1, Ordering::SeqCst);
+                            request_futs.push(handle_forwarded_headers_frame(stream_id, headers, send, recv));
                     }
-                    H3Event::ConnectionError(err) => { break Err(err.into()); }
-                    H3Event::ConnectionShutdown(Some(err)) => { break Err(err.into()); }
-                    _ => unreachable!()
                 }
             }
             Some(_) = request_futs.next() => {}

tokio-quiche/tests/integration_tests/headers.rs

@@ -45,12 +45,18 @@ async fn test_additional_headers() {
         move |mut h3_conn| async move {
             let event_rx = h3_conn.h3_controller.event_receiver_mut();
 
-            while let Some(frame) = event_rx.recv().await {
-                let ServerH3Event::Core(frame) = frame;
+            while let Some(event) = event_rx.recv().await {
+                match event {
+                    ServerH3Event::Core(event) => match event {
+                        H3Event::ConnectionShutdown(_) => break,
 
-                match frame {
-                    H3Event::IncomingHeaders(headers) => {
-                        let IncomingH3Headers { mut send, .. } = headers;
+                        _ => (),
+                    },
+
+                    ServerH3Event::Headers {
+                        incoming_headers, ..
+                    } => {
+                        let IncomingH3Headers { mut send, .. } = incoming_headers;
 
                         // Send initial headers.
                         send.send(OutboundFrame::Headers(
@@ -73,10 +79,6 @@ async fn test_additional_headers() {
                         .await
                         .unwrap();
                     },
-
-                    H3Event::ConnectionShutdown(_) => break,
-
-                    _ => (),
                 }
             }
         },