h3: add support for streaming HEADERS sends

Calls to send_request(), send_response(), send_response_with_priority(),
and send_additional_headers() can fail with a StreamBlocked
error indicating lack of underlying transport capacity. When this
occurs, applications are expected to retry the operation when
the stream is later reported as writable.

However, certain conditions could mean that sufficient capacity
might never be made available, effectively permenantly blocking
header sends. The root cause of this problem was the choice to enforce
that a HEADERS frame is always sent (buffered into a quiche stream)
in whole.

This change adds new functions related to stream management and header
sending to support a streaming send design. These will produce a HEADERS
frame that can be sent in whole or in part, depending on available
capacity. When a frame is only partly sent, applications are notified and
can resume sending using the new continue_partial_headers()
method, once the stream is writable.

The new functions unify how clients and servers send headers.
For a client, the expected sequence is now something like:

0. Decide to initate a request
1. reserve_request_stream() - reserves a stream if limits allow
2. stream_headers() - inititates streaming of a HEADERS frame
3. continue_partial_headers() - if stream_headers() returned Error::PartialHeader

For a server, the expected sequence is now something like:

0. Receive a request via the poll() function and decide to respond
1. stream_priority() - set the stream's sending priority per RFC 9218
3. stream_headers() - inititates streaming of a HEADERS frame
4. continue_partial_headers() - if stream_headers return Error::PartialHeader

While headers are being streamed, other operations that
would cause an HTTP/3 frame to be sent on the stream are
prevented. HEADERS frames must be sent completely before
other operations are successful.

Applications do not need to manage the partial HEADERS
buffer, this is dealt with inside quiche.

Co-authored-by: Gregor Maier <[email protected]>

Author: LPardue

apps/src/common.rs

@@ -334,6 +334,8 @@ pub trait HttpConn {
         &mut self, conn: &mut quiche::Connection, target_path: &Option<String>,
     );
 
+    fn retry_partial_headers(&mut self, conn: &mut quiche::Connection);
+
     fn handle_responses(
         &mut self, conn: &mut quiche::Connection, buf: &mut [u8],
         req_start: &std::time::Instant,
@@ -369,12 +371,19 @@ pub struct Http09Request {
     response_writer: Option<std::io::BufWriter<std::fs::File>>,
 }
 
+enum HeadersSendStatus {
+    NotStarted,
+    InProgess,
+    Completed,
+}
+
 /// Represents an HTTP/3 formatted request.
 struct Http3Request {
     url: url::Url,
     cardinal: u64,
     stream_id: Option<u64>,
     hdrs: Vec<quiche::h3::Header>,
+    hdrs_send_status: HeadersSendStatus,
     priority: Option<Priority>,
     response_hdrs: Vec<quiche::h3::Header>,
     response_body: Vec<u8>,
@@ -477,6 +486,8 @@ impl HttpConn for Http09Conn {
         self.reqs_sent += reqs_done;
     }
 
+    fn retry_partial_headers(&mut self, _conn: &mut quiche::Connection) {}
+
     fn handle_responses(
         &mut self, conn: &mut quiche::Connection, buf: &mut [u8],
         req_start: &std::time::Instant,
@@ -829,6 +840,7 @@ impl Http3Conn {
                     url: url.clone(),
                     cardinal: i,
                     hdrs,
+                    hdrs_send_status: HeadersSendStatus::NotStarted,
                     priority,
                     response_hdrs: Vec::new(),
                     response_body: Vec::new(),
@@ -1120,19 +1132,78 @@ impl Http3Conn {
 }
 
 impl HttpConn for Http3Conn {
+    fn retry_partial_headers(&mut self, conn: &mut quiche::Connection) {
+        for req in self.reqs.iter_mut().filter(|r| {
+            matches!(r.hdrs_send_status, HeadersSendStatus::InProgess)
+        }) {
+            let stream_id = req
+                .stream_id
+                .expect("partial header resend but no stream ID");
+
+            debug!("retrying stream id={}", stream_id);
+            match self.h3_conn.continue_partial_headers(conn, stream_id) {
+                Ok(_) => {
+                    req.hdrs_send_status = HeadersSendStatus::Completed;
+                    debug!("Completed HTTP request {:?}", &req.hdrs);
+                    self.reqs_hdrs_sent += 1;
+                },
+
+                Err(e) => {
+                    error!("retry failed to send request {:?}", e);
+                    break;
+                },
+            }
+        }
+    }
+
     fn send_requests(
         &mut self, conn: &mut quiche::Connection, target_path: &Option<String>,
     ) {
-        let mut reqs_done = 0;
+        // First retry partial headers
+        self.retry_partial_headers(conn);
+
+        // Then send new headers.
+        for req in self.reqs.iter_mut().filter(|r| {
+            matches!(r.hdrs_send_status, HeadersSendStatus::NotStarted)
+        }) {
+            let stream_id = match self.h3_conn.reserve_request_stream(conn) {
+                Ok(v) => v,
 
-        // First send headers.
-        for req in self.reqs.iter_mut().skip(self.reqs_hdrs_sent) {
-            let s = match self.h3_conn.send_request(
+                Err(quiche::h3::Error::TransportError(
+                    quiche::Error::StreamLimit,
+                )) => {
+                    debug!("not enough stream credits, retry later...");
+                    break;
+                },
+
+                Err(quiche::h3::Error::StreamBlocked) => {
+                    debug!("stream is blocked, retry later...");
+                    break;
+                },
+
+                Err(e) => {
+                    error!("failed to reserve stream {e:?}");
+                    break;
+                },
+            };
+
+            match self.h3_conn.stream_headers(
                 conn,
+                stream_id,
                 &req.hdrs,
+                false,
                 self.body.is_none(),
             ) {
-                Ok(v) => v,
+                Ok(()) => {
+                        req.hdrs_send_status = HeadersSendStatus::Completed;
+                        debug!("Completed HTTP request {:?}", &req.hdrs);
+                        self.reqs_hdrs_sent += 1;
+                    },
+
+                Err(quiche::h3::Error::PartialHeader) => {
+                    debug!("sent partial headers, marking for later continuation");
+                    req.hdrs_send_status = HeadersSendStatus::InProgess;
+                },
 
                 Err(quiche::h3::Error::TransportError(
                     quiche::Error::StreamLimit,
@@ -1152,23 +1223,18 @@ impl HttpConn for Http3Conn {
                 },
             };
 
-            debug!("Sent HTTP request {:?}", &req.hdrs);
-
             if let Some(priority) = &req.priority {
                 // If sending the priority fails, don't try again.
                 self.h3_conn
-                    .send_priority_update_for_request(conn, s, priority)
+                    .send_priority_update_for_request(conn, stream_id, priority)
                     .ok();
             }
 
-            req.stream_id = Some(s);
+            req.stream_id = Some(stream_id);
             req.response_writer =
                 make_resource_writer(&req.url, target_path, req.cardinal);
-            self.sent_body_bytes.insert(s, 0);
-
-            reqs_done += 1;
+            self.sent_body_bytes.insert(stream_id, 0);
         }
-        self.reqs_hdrs_sent += reqs_done;
 
         // Then send any remaining body.
         if let Some(body) = &self.body {

quiche/include/quiche.h

@@ -1117,6 +1117,9 @@ int64_t quiche_h3_send_request(quiche_h3_conn *conn, quiche_conn *quic_conn,
                                const quiche_h3_header *headers, size_t headers_len,
                                bool fin);
 
+// Reserves an HTTP/3 request stream.
+int64_t reserve_request_stream(quiche_h3_conn *conn, quiche_conn *quic_conn);
+
 // Sends an HTTP/3 response on the specified stream with default priority.
 int quiche_h3_send_response(quiche_h3_conn *conn, quiche_conn *quic_conn,
                             uint64_t stream_id, const quiche_h3_header *headers,
@@ -1134,6 +1137,16 @@ int quiche_h3_send_additional_headers(quiche_h3_conn *conn,
                             quiche_h3_header *headers, size_t headers_len,
                             bool is_trailer_section, bool fin);
 
+// Initiates streaming of a new HTTP/3 HEADERS frame.
+int quiche_h3_stream_headers(quiche_h3_conn *conn,
+                            quiche_conn *quic_conn, uint64_t stream_id,
+                            quiche_h3_header *headers, size_t headers_len,
+                            bool is_trailer_section, bool fin);
+
+// Continues sending headers on the given stream.
+int quiche_h3_continue_partial_headers(quiche_h3_conn *conn,
+                            quiche_conn *quic_conn, uint64_t stream_id);
+
 // Sends an HTTP/3 body chunk on the given stream.
 ssize_t quiche_h3_send_body(quiche_h3_conn *conn, quiche_conn *quic_conn,
                             uint64_t stream_id, const uint8_t *body, size_t body_len,
@@ -1147,6 +1160,10 @@ ssize_t quiche_h3_recv_body(quiche_h3_conn *conn, quiche_conn *quic_conn,
 int quiche_h3_send_goaway(quiche_h3_conn *conn, quiche_conn *quic_conn,
                           uint64_t id);
 
+// Sets the HTTP/3 priority for a stream.
+int quiche_h3_stream_priority(quiche_conn *conn, uint64_t stream_id,
+                                   quiche_h3_priority *priority);
+
 // Try to parse an Extensible Priority field value.
 int quiche_h3_parse_extensible_priority(uint8_t *priority,
                                         size_t priority_len,

quiche/src/h3/ffi.rs

@@ -288,6 +288,37 @@ pub extern "C" fn quiche_h3_send_additional_headers(
     }
 }
 
+#[no_mangle]
+pub extern "C" fn quiche_h3_stream_headers(
+    conn: &mut h3::Connection, quic_conn: &mut Connection, stream_id: u64,
+    headers: *const Header, headers_len: size_t, is_trailer_section: bool,
+    fin: bool,
+) -> c_int {
+    let headers = headers_from_ptr(headers, headers_len);
+
+    match conn.stream_headers(
+        quic_conn,
+        stream_id,
+        &headers,
+        is_trailer_section,
+        fin,
+    ) {
+        Ok(_) => 0,
+
+        Err(e) => e.to_c() as c_int,
+    }
+}
+
+#[no_mangle]
+pub extern "C" fn quiche_h3_continue_partial_headers(
+    conn: &mut h3::Connection, quic_conn: &mut Connection, stream_id: u64,
+) -> c_int {
+    match conn.continue_partial_headers(quic_conn, stream_id) {
+        Ok(_) => 0,
+        Err(e) => e.to_c() as c_int,
+    }
+}
+
 #[no_mangle]
 pub extern "C" fn quiche_h3_send_body(
     conn: &mut h3::Connection, quic_conn: &mut Connection, stream_id: u64,
@@ -335,6 +366,17 @@ pub extern "C" fn quiche_h3_send_goaway(
     }
 }
 
+#[no_mangle]
+pub extern "C" fn quiche_h3_stream_priority(
+    quic_conn: &mut Connection, id: u64, priority: &Priority,
+) -> c_int {
+    match h3::Connection::stream_priority(quic_conn, id, priority) {
+        Ok(()) => 0,
+
+        Err(e) => e.to_c() as c_int,
+    }
+}
+
 #[no_mangle]
 #[cfg(feature = "sfv")]
 pub extern "C" fn quiche_h3_parse_extensible_priority(