fix(client): ignore frames on forgotten streams

src/proto/streams/streams.rs

@@ -496,13 +496,16 @@ impl Inner {
                 // the request headers, so don't allow.
                 if !peer.is_server() {
                     // This may be response headers for a stream we've already
-                    // forgotten about...
+                    // forgotten about. RFC 9113 section 5.1 lets an endpoint
+                    // discard frames on any closed stream, and the codec has
+                    // already updated the HPACK state. Answering with a reset
+                    // would count against `max_local_error_reset_streams` and
+                    // eventually close the connection, although the server did
+                    // nothing wrong when it sent the response before it read
+                    // our RST_STREAM.
                     if self.actions.may_have_forgotten_stream(peer, id) {
-                        tracing::debug!(
-                            "recv_headers for old stream={:?}, sending STREAM_CLOSED",
-                            id,
-                        );
-                        return Err(Error::library_reset(id, Reason::STREAM_CLOSED));
+                        tracing::debug!("recv_headers for old stream={:?}, ignoring", id);
+                        return Ok(RecvOutcome::Processed);
                     }
                 }
 
@@ -629,15 +632,38 @@ impl Inner {
                 }
 
                 if self.actions.may_have_forgotten_stream(peer, id) {
-                    tracing::debug!("recv_data for old stream={:?}, sending STREAM_CLOSED", id,);
-
                     let sz = frame.flow_controlled_len();
                     // This should have been enforced at the codec::FramedRead layer, so
                     // this is just a sanity check.
                     assert!(sz <= super::MAX_WINDOW_SIZE as usize);
                     let sz = sz as WindowSize;
                     self.actions.recv.ignore_data(sz)?;
 
+                    // A client discards the rest of a response it no longer
+                    // tracks, for the same reason as in `recv_headers`. The
+                    // frame counts against the DATA frame budget as it would
+                    // on an open stream, so a final frame doesn't count.
+                    // Servers often end a response with an empty one, which
+                    // would otherwise count against
+                    // `MAX_RECV_EMPTY_DATA_FRAMES` for every cancelled stream.
+                    if !peer.is_server() {
+                        tracing::debug!("recv_data for old stream={:?}, ignoring", id);
+                        if frame.is_end_stream() {
+                            return Ok(());
+                        }
+                        return self
+                            .counts
+                            .record_data_frame(frame.payload().len())
+                            .map_err(|_| {
+                                tracing::debug!("too many small DATA frames");
+                                Error::library_go_away_data(
+                                    Reason::ENHANCE_YOUR_CALM,
+                                    "too_many_data_frames",
+                                )
+                            });
+                    }
+
+                    tracing::debug!("recv_data for old stream={:?}, sending STREAM_CLOSED", id);
                     return Err(Error::library_reset(id, Reason::STREAM_CLOSED));
                 }
 
@@ -1848,8 +1874,9 @@ impl Actions {
     /// frames in-flight, and by the time we receive them, our own state is
     /// gone. We *could* tear everything down by sending a GOAWAY, but it
     /// is more likely to be latency/memory constraints that caused this,
-    /// and not a bad actor. So be less catastrophic, the spec allows
-    /// us to send another RST_STREAM of STREAM_CLOSED.
+    /// and not a bad actor. So be less catastrophic: a client discards
+    /// these frames, and a server answers with a RST_STREAM of
+    /// STREAM_CLOSED, both of which the spec allows.
     fn may_have_forgotten_stream(&self, peer: peer::Dyn, id: StreamId) -> bool {
         if id.is_zero() {
             return false;

tests/h2-tests/tests/client_request.rs

@@ -1826,7 +1826,8 @@ async fn rogue_server_reused_headers() {
         .await;
         srv.send_frame(frames::headers(1).response(200).eos()).await;
         srv.send_frame(frames::headers(1)).await;
-        srv.recv_frame(frames::reset(1).stream_closed()).await;
+        // the client discards HEADERS on a stream it no longer tracks
+        srv.ping_pong([1; 8]).await;
     };
 
     let h2 = async move {

tests/h2-tests/tests/flow_control.rs

@@ -715,7 +715,8 @@ async fn padded_data_on_forgotten_stream_releases_connection_capacity() {
         // flow_controlled_len. There used to be a bug where the padded
         // frame would release only 16378 (payload) instead of 16384.
         srv.recv_frame(frames::window_update(0, 16_384 * 2)).await;
-        srv.recv_frame(frames::reset(1).stream_closed()).await;
+        // and the frame is discarded without a reset
+        srv.ping_pong([2; 8]).await;
     };
 
     let client = async move {

tests/h2-tests/tests/stream_states.rs

@@ -872,9 +872,9 @@ async fn skipped_stream_ids_are_implicitly_closed() {
         // send the response on a lower-numbered stream, which should be
         // implicitly closed.
         srv.send_frame(frames::headers(3).response(299)).await;
-        // however, our client choose to send a RST_STREAM because it
-        // can't tell if it had previously reset '3'.
-        srv.recv_frame(frames::reset(3).stream_closed()).await;
+        // our client discards it, because it can't tell if it had
+        // previously reset '3'.
+        srv.ping_pong([1; 8]).await;
         srv.send_frame(frames::headers(5).response(200).eos()).await;
     };
 
@@ -1006,8 +1006,8 @@ async fn rst_stream_expires() {
         srv.send_frame(frames::data(1, vec![0; 16_384]).eos()).await;
         // window capacity is returned
         srv.recv_frame(frames::window_update(0, 16_384 * 2)).await;
-        // and then stream error
-        srv.recv_frame(frames::reset(1).stream_closed()).await;
+        // and the frame is discarded without a stream error
+        srv.ping_pong([2; 8]).await;
     };
 
     let client = async move {
@@ -1065,9 +1065,9 @@ async fn rst_stream_max() {
         srv.send_frame(frames::data(1, vec![0; 16]).eos()).await;
         // ping pong to be sure of no goaway
         srv.ping_pong([1; 8]).await;
-        // 3 has been evicted, will get a reset
+        // 3 has been evicted, but is discarded too
         srv.send_frame(frames::data(3, vec![0; 16]).eos()).await;
-        srv.recv_frame(frames::reset(3).stream_closed()).await;
+        srv.ping_pong([2; 8]).await;
     };
 
     let client = async move {
@@ -1104,6 +1104,120 @@ async fn rst_stream_max() {
     join(srv, client).await;
 }
 
+#[tokio::test]
+async fn client_discards_late_frames_on_forgotten_streams() {
+    h2_support::trace_init!();
+    let (io, mut srv) = mock::new();
+
+    let srv = async move {
+        let settings = srv.assert_client_handshake().await;
+        assert_default_settings!(settings);
+        for id in [1, 3, 5] {
+            srv.recv_frame(
+                frames::headers(id)
+                    .request("GET", "https://example.com/")
+                    .eos(),
+            )
+            .await;
+        }
+        srv.recv_frame(frames::reset(1).cancel()).await;
+        srv.recv_frame(frames::reset(3).cancel()).await;
+        // sending the responses that were in flight when the client cancelled
+        srv.send_frame(frames::headers(1).response(200)).await;
+        srv.send_frame(frames::data(1, vec![0; 16_384]).eos()).await;
+        srv.send_frame(frames::headers(3).response(200)).await;
+        srv.send_frame(frames::data(3, vec![0; 16_384]).eos()).await;
+        // the client returns the connection window and sends nothing else
+        srv.recv_frame(frames::window_update(0, 16_384 * 2)).await;
+        srv.ping_pong([1; 8]).await;
+        // the request still in flight completes
+        srv.send_frame(frames::headers(5).response(200).eos()).await;
+    };
+
+    let client = async move {
+        let (mut client, mut conn) = client::Builder::new()
+            // forget reset streams as soon as the RST_STREAM is sent
+            .max_concurrent_reset_streams(0)
+            // a second local reset would close the connection
+            .max_local_error_reset_streams(Some(1))
+            .handshake::<_, Bytes>(io)
+            .await
+            .expect("handshake");
+
+        for _ in 0..2 {
+            let request = Request::builder()
+                .uri("https://example.com/")
+                .body(())
+                .unwrap();
+            let (resp, _) = client.send_request(request, true).unwrap();
+            // drop resp will send a reset
+            drop(resp);
+        }
+
+        let resp = conn
+            .drive(client.get("https://example.com/"))
+            .await
+            .expect("response");
+        assert_eq!(resp.status(), StatusCode::OK);
+        conn.await.expect("client");
+    };
+
+    join(srv, client).await;
+}
+
+#[tokio::test]
+async fn client_counts_late_data_frames_on_forgotten_streams() {
+    h2_support::trace_init!();
+    let (io, mut srv) = mock::new();
+
+    let srv = async move {
+        let settings = srv.assert_client_handshake().await;
+        assert_default_settings!(settings);
+        srv.recv_frame(
+            frames::headers(1)
+                .request("GET", "https://example.com/")
+                .eos(),
+        )
+        .await;
+        srv.recv_frame(frames::reset(1).cancel()).await;
+        srv.send_frame(frames::headers(1).response(200)).await;
+        // final frames don't count, as on an open stream
+        for _ in 0..101 {
+            srv.send_frame(frames::data(1, "").eos()).await;
+        }
+        srv.ping_pong([1; 8]).await;
+        // the client discards these, but they still exhaust the budget
+        for _ in 0..101 {
+            srv.send_frame(frames::data(1, "")).await;
+        }
+        srv.recv_frame(frames::go_away(0).calm().data("too_many_data_frames"))
+            .await;
+    };
+
+    let client = async move {
+        let (mut client, conn) = client::Builder::new()
+            // forget reset streams as soon as the RST_STREAM is sent
+            .max_concurrent_reset_streams(0)
+            .handshake::<_, Bytes>(io)
+            .await
+            .expect("handshake");
+
+        let request = Request::builder()
+            .uri("https://example.com/")
+            .body(())
+            .unwrap();
+        let (resp, _) = client.send_request(request, true).unwrap();
+        // drop resp will send a reset
+        drop(resp);
+
+        let err = conn.await.unwrap_err();
+        assert_eq!(err.reason(), Some(Reason::ENHANCE_YOUR_CALM));
+        drop(client);
+    };
+
+    join(srv, client).await;
+}
+
 #[tokio::test]
 async fn reserved_state_recv_window_update() {
     h2_support::trace_init!();