From 44260b15da7d6cf15c241295a2d27917c157f4c8 Mon Sep 17 00:00:00 2001 From: zaidoon Date: Tue, 13 Oct 2026 11:52:07 -0400 Subject: [PATCH] fix(client): ignore frames on forgotten streams When a client cancels a request, the server has often already sent some or all of the response. h2 remembers locally reset streams for a while and drops their frames, but once a stream falls out of that window, each late HEADERS or DATA frame got a RST_STREAM(STREAM_CLOSED) back. Every one of those counts against max_local_error_reset_streams, which never goes down, so a long-lived client connection eventually sends itself GOAWAY(ENHANCE_YOUR_CALM, "too_many_internal_resets") and fails every request in flight on it. RFC 9113 section 5.1 lets an endpoint minimally process and discard frames on any closed stream, so the client now does that for HEADERS and DATA, like it already did for RST_STREAM and WINDOW_UPDATE. DATA still gives back its connection window and counts against the DATA frame budget the way it would on an open stream, and the codec has already decoded the header block. Frames on streams the client never opened are still a connection error, and the server side is unchanged. New test client_discards_late_frames_on_forgotten_streams fails on master with GoAway(too_many_internal_resets). Five tests that expected the STREAM_CLOSED reply now check that nothing is sent instead. client_counts_late_data_frames_on_forgotten_streams checks the budget. --- src/proto/streams/streams.rs | 47 +++++++-- tests/h2-tests/tests/client_request.rs | 3 +- tests/h2-tests/tests/flow_control.rs | 3 +- tests/h2-tests/tests/stream_states.rs | 128 +++++++++++++++++++++++-- 4 files changed, 162 insertions(+), 19 deletions(-) diff --git a/src/proto/streams/streams.rs b/src/proto/streams/streams.rs index ea81ee0d7..e34092836 100644 --- a/src/proto/streams/streams.rs +++ b/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,8 +632,6 @@ 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. @@ -638,6 +639,31 @@ impl Inner { 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; diff --git a/tests/h2-tests/tests/client_request.rs b/tests/h2-tests/tests/client_request.rs index 189f6917c..f563df539 100644 --- a/tests/h2-tests/tests/client_request.rs +++ b/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 { diff --git a/tests/h2-tests/tests/flow_control.rs b/tests/h2-tests/tests/flow_control.rs index 11d1207e1..59ae6d8b7 100644 --- a/tests/h2-tests/tests/flow_control.rs +++ b/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 { diff --git a/tests/h2-tests/tests/stream_states.rs b/tests/h2-tests/tests/stream_states.rs index 6f9e31de5..95ef3b781 100644 --- a/tests/h2-tests/tests/stream_states.rs +++ b/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!();