h3: earlier check for control stream finished

A peer can FIN a control stream at any point during the stream lifetime. While
doing so is an error in HTTP/3, our handling before this change was a bit odd.

Previously, depending on the sequencing, an unexpected FIN could trigger an
InvalidStreamState error when attempting to read from a control stream. This
would bubble up to an app via poll(), which is weird and unexpected because the
quiche h3 layer is supposed to deal with all the control stream details.
A subsequent poll() would then cause the finished detection to kick in and
enforce the RFC rules to close the connection with CloseCriticialStream.
However, that's a bit late and clunky.

This change adds additional checks for the stream finished event in order
to make sure we close connections at the earliest time and prevent useless
errors getting bubbled up to apps.

quiche/src/h3/mod.rs

@@ -931,6 +931,26 @@ pub struct Stats {
     pub qpack_decoder_stream_recv_bytes: u64,
 }
 
+fn close_conn_critical_stream(conn: &mut super::Connection) -> Result<()> {
+    conn.close(
+        true,
+        Error::ClosedCriticalStream.to_wire(),
+        b"Critical stream closed.",
+    )?;
+
+    Err(Error::ClosedCriticalStream)
+}
+
+fn check_closed_critical_stream(
+    conn: &mut super::Connection, stream_id: u64,
+) -> Result<()> {
+    if conn.stream_finished(stream_id) {
+        close_conn_critical_stream(conn)?;
+    }
+
+    Ok(())
+}
+
 /// An HTTP/3 connection.
 pub struct Connection {
     is_server: bool,
@@ -2129,15 +2149,7 @@ impl Connection {
     fn process_control_stream(
         &mut self, conn: &mut super::Connection, stream_id: u64,
     ) -> Result<(u64, Event)> {
-        if conn.stream_finished(stream_id) {
-            conn.close(
-                true,
-                Error::ClosedCriticalStream.to_wire(),
-                b"Critical stream closed.",
-            )?;
-
-            return Err(Error::ClosedCriticalStream);
-        }
+        check_closed_critical_stream(conn, stream_id)?;
 
         if !conn.stream_readable(stream_id) {
             return Err(Error::Done);
@@ -2151,15 +2163,7 @@ impl Connection {
             Err(e) => return Err(e),
         };
 
-        if conn.stream_finished(stream_id) {
-            conn.close(
-                true,
-                Error::ClosedCriticalStream.to_wire(),
-                b"Critical stream closed.",
-            )?;
-
-            return Err(Error::ClosedCriticalStream);
-        }
+        check_closed_critical_stream(conn, stream_id)?;
 
         Err(Error::Done)
     }
@@ -2231,6 +2235,8 @@ impl Connection {
                                 stream_id
                             );
 
+                            check_closed_critical_stream(conn, stream_id)?;
+
                             self.peer_control_stream_id = Some(stream_id);
                         },
 
@@ -2308,6 +2314,10 @@ impl Connection {
                 stream::State::FrameType => {
                     stream.try_fill_buffer(conn)?;
 
+                    if stream.ty() == Some(stream::Type::Control) {
+                        check_closed_critical_stream(conn, stream_id)?;
+                    }
+
                     let varint = match stream.try_consume_varint() {
                         Ok(v) => v,
 
@@ -2350,6 +2360,10 @@ impl Connection {
                         Err(_) => continue,
                     };
 
+                    if stream.ty() == Some(stream::Type::Control) {
+                        check_closed_critical_stream(conn, stream_id)?;
+                    }
+
                     // DATA frames are handled uniquely. After this point we lose
                     // visibility of DATA framing, so just log here.
                     if Some(frame::DATA_FRAME_TYPE_ID) == stream.frame_type() {
@@ -2389,6 +2403,10 @@ impl Connection {
 
                     stream.try_fill_buffer(conn)?;
 
+                    if stream.ty() == Some(stream::Type::Control) {
+                        check_closed_critical_stream(conn, stream_id)?;
+                    }
+
                     let (frame, payload_len) = match stream.try_consume_frame() {
                         Ok(frame) => frame,
 
@@ -4581,7 +4599,43 @@ mod tests {
 
     #[test]
     /// Client closes the control stream, which is forbidden.
-    fn close_control_stream() {
+    fn close_control_stream_after_type() {
+        let mut s = Session::new().unwrap();
+        s.handshake().unwrap();
+
+        let mut control_stream_closed = false;
+
+        s.pipe
+            .client
+            .stream_send(s.client.control_stream_id.unwrap(), &vec![], true)
+            .unwrap();
+
+        s.advance().ok();
+
+        loop {
+            match s.server.poll(&mut s.pipe.server) {
+                Ok(_) => (),
+
+                Err(Error::Done) => {
+                    break;
+                },
+
+                Err(Error::ClosedCriticalStream) => {
+                    control_stream_closed = true;
+                    break;
+                },
+
+                Err(_) => panic!("unexpected control stream error bubbling"),
+            }
+        }
+
+        assert!(control_stream_closed);
+    }
+
+    #[test]
+    /// Client closes the control stream after a frame is sent, which is
+    /// forbidden.
+    fn close_control_stream_after_frame() {
         let mut s = Session::new().unwrap();
         s.handshake().unwrap();
 
@@ -4607,7 +4661,7 @@ mod tests {
                     break;
                 },
 
-                Err(_) => (),
+                Err(_) => panic!("unexpected control stream error bubbling"),
             }
         }