h3: ignore priority updates for closed streams

Prevent late PRIORITY_UPDATE frames from recreating H3 stream state
after the request stream has completed but before transport collection.

Add a regression test covering H3 collection before QUIC stream collection.

Author: ghedo

quiche/src/h3/mod.rs

@@ -3202,13 +3202,11 @@ impl Connection {
                     return Err(Error::IdError);
                 }
 
-                // If the PRIORITY_UPDATE is valid, consider storing the latest
-                // contents. Due to reordering, it is possible that we might
-                // receive frames that reference streams that have not yet to
-                // been opened and that's OK because it's within our concurrency
-                // limit. However, we discard PRIORITY_UPDATE that refers to
-                // streams that we know have been collected.
-                if conn.streams.is_collected(prioritized_element_id) {
+                // PRIORITY_UPDATE can arrive before the request stream exists,
+                // so a missing transport stream is allowed. Ignore updates only
+                // once the transport stream was collected or both transport
+                // directions are finished.
+                if conn.stream_closed(prioritized_element_id) {
                     return Err(Error::Done);
                 }
 
@@ -5236,6 +5234,64 @@ mod tests {
         assert_eq!(s.poll_server(), Err(Error::Done));
     }
 
+    #[test]
+    /// Send a PRIORITY_UPDATE for a request stream after H3 has collected it,
+    /// but before the transport stream has been collected.
+    fn priority_update_request_after_h3_collection() {
+        let mut s = Session::new().unwrap();
+        s.handshake().unwrap();
+
+        let init_streams_server = s.server.streams.len();
+
+        let (stream, req) = s.send_request(true).unwrap();
+        let ev_headers = Event::Headers {
+            list: req,
+            more_frames: false,
+        };
+
+        assert_eq!(s.poll_server(), Ok((stream, ev_headers)));
+        assert_eq!(s.poll_server(), Ok((stream, Event::Finished)));
+        assert_eq!(s.poll_server(), Err(Error::Done));
+
+        let resp = vec![
+            Header::new(b":status", b"200"),
+            Header::new(b"server", b"quiche-test"),
+        ];
+
+        s.server
+            .send_response(&mut s.pipe.server, stream, &resp, true)
+            .unwrap();
+
+        // H3 no longer needs its stream state once it has sent the response
+        // FIN and consumed the request FIN. The QUIC stream remains until the
+        // response FIN is acknowledged by the peer.
+        assert_eq!(s.server.streams.len(), init_streams_server);
+        assert!(s.pipe.server.stream_finished(stream));
+        assert!(s.pipe.server.stream_closed(stream));
+
+        let stream_state = s.pipe.server.streams.get(stream).unwrap();
+        assert!(stream_state.recv.is_fin());
+        assert!(stream_state.send.is_fin());
+        assert!(!s.pipe.server.streams.is_collected(stream));
+
+        s.client
+            .send_priority_update_for_request(
+                &mut s.pipe.client,
+                stream,
+                &Priority {
+                    urgency: 3,
+                    incremental: false,
+                },
+            )
+            .unwrap();
+
+        let flight = crate::test_utils::emit_flight(&mut s.pipe.client).unwrap();
+        crate::test_utils::process_flight(&mut s.pipe.server, flight).unwrap();
+
+        assert_eq!(s.poll_server(), Err(Error::Done));
+        assert_eq!(s.server.streams.len(), init_streams_server);
+    }
+
     #[test]
     /// Send a PRIORITY_UPDATE for a request stream, before and after the stream
     /// has been stopped.

quiche/src/lib.rs

@@ -6540,6 +6540,36 @@ impl<F: BufFactory> Connection<F> {
         stream.recv.is_fin()
     }
 
+    /// Returns true if the specified stream is closed.
+    ///
+    /// For bidirectional streams this happens when both the receive and send
+    /// sides have signaled `fin`. For unidirectional streams only the
+    /// relevant direction is checked, depending on whether the stream was
+    /// created locally or not.
+    ///
+    /// This also returns true if the stream has already been collected, but
+    /// returns false if the stream was never opened.
+    #[inline]
+    pub fn stream_closed(&self, stream_id: u64) -> bool {
+        let Some(stream) = self.streams.get(stream_id) else {
+            return self.streams.is_collected(stream_id);
+        };
+
+        match (stream.bidi, stream.local) {
+            // For bidirectional streams both directions must have signaled
+            // FIN.
+            (true, _) => stream.recv.is_fin() && stream.send.is_fin(),
+
+            // For unidirectional streams created locally, only the send side
+            // is checked.
+            (false, true) => stream.send.is_fin(),
+
+            // For unidirectional streams created by the peer, only the
+            // receive side is checked.
+            (false, false) => stream.recv.is_fin(),
+        }
+    }
+
     /// Returns the number of bidirectional streams that can be created
     /// before the peer's stream count limit is reached.
     ///

quiche/src/tests.rs

@@ -1013,6 +1013,61 @@ fn streamio(#[values("cubic", "bbr2_gcongestion")] cc_algorithm_name: &str) {
     assert!(pipe.server.stream_finished(4));
 }
 
+#[rstest]
+fn stream_closed_bidi(
+    #[values("cubic", "bbr2_gcongestion")] cc_algorithm_name: &str,
+) {
+    let mut pipe = test_utils::Pipe::new(cc_algorithm_name).unwrap();
+    assert_eq!(pipe.handshake(), Ok(()));
+
+    assert!(!pipe.client.stream_closed(0));
+    assert!(!pipe.server.stream_closed(0));
+
+    assert_eq!(pipe.client.stream_send(0, b"hello", true), Ok(5));
+    assert_eq!(pipe.advance(), Ok(()));
+
+    assert!(!pipe.client.stream_closed(0));
+    assert!(!pipe.server.stream_closed(0));
+
+    let mut buf = [0; 5];
+    assert_eq!(pipe.server.stream_recv(0, &mut buf), Ok((5, true)));
+
+    assert!(!pipe.server.stream_closed(0));
+
+    assert_eq!(pipe.server.stream_send(0, b"world", true), Ok(5));
+    assert!(pipe.server.stream_closed(0));
+
+    assert_eq!(pipe.advance(), Ok(()));
+
+    assert!(!pipe.client.stream_closed(0));
+    assert_eq!(pipe.client.stream_recv(0, &mut buf), Ok((5, true)));
+    assert!(pipe.client.stream_closed(0));
+}
+
+#[rstest]
+fn stream_closed_uni(
+    #[values("cubic", "bbr2_gcongestion")] cc_algorithm_name: &str,
+) {
+    let mut pipe = test_utils::Pipe::new(cc_algorithm_name).unwrap();
+    assert_eq!(pipe.handshake(), Ok(()));
+
+    assert!(!pipe.client.stream_closed(2));
+    assert!(!pipe.server.stream_closed(2));
+
+    assert_eq!(pipe.client.stream_send(2, b"hello", true), Ok(5));
+
+    assert!(pipe.client.stream_closed(2));
+
+    assert_eq!(pipe.advance(), Ok(()));
+
+    assert!(!pipe.server.stream_closed(2));
+
+    let mut buf = [0; 5];
+    assert_eq!(pipe.server.stream_recv(2, &mut buf), Ok((5, true)));
+
+    assert!(pipe.server.stream_closed(2));
+}
+
 /// Test receiving into `BufMut`
 #[rstest]
 fn stream_recv_buf(