diff --git a/src/proto/streams/streams.rs b/src/proto/streams/streams.rs index ea81ee0d..e3409283 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 189f6917..f563df53 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 11d1207e..59ae6d8b 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 6f9e31de..95ef3b78 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!();