From 7b3c71773c6801a7b3bddc0d7b073f78e7de1d47 Mon Sep 17 00:00:00 2001 From: John Howard Date: Fri, 17 Jul 2026 11:52:28 -0700 Subject: [PATCH 1/2] fix: preserve received END_STREAM across HTTP/2 stream resets Keep resets received after EOS distinct from ordinary stream errors. This allows the receive half to remain complete while the send half still observes the reset and its reason. This fixes header-only responses followed by RST_STREAM(NO_ERROR), which could otherwise lose their END_STREAM state when passed through a proxy. --- src/proto/streams/state.rs | 65 ++++++++++++++++++++++++++++++++------ 1 file changed, 56 insertions(+), 9 deletions(-) diff --git a/src/proto/streams/state.rs b/src/proto/streams/state.rs index c9e709ac..d0009a5a 100644 --- a/src/proto/streams/state.rs +++ b/src/proto/streams/state.rs @@ -76,6 +76,8 @@ enum Peer { enum Cause { EndStream, Error(Error), + /// The stream was reset after the receive half had already reached EOS. + ErrorAfterEndStream(Error), /// This indicates to the connection that a reset frame must be sent out /// once the send queue has been flushed. @@ -256,6 +258,7 @@ impl State { /// - `frame`: the received RST_STREAM frame. /// - `queued`: true if this stream has frames in the pending send queue. pub fn recv_reset(&mut self, frame: frame::Reset, queued: bool) { + let recv_end_stream = self.is_recv_end_stream(); match self.inner { // If the stream is already in a `Closed` state, do nothing, // provided that there are no frames still in the send queue. @@ -281,10 +284,13 @@ impl State { state, queued ); - self.inner = Closed(Cause::Error(Error::remote_reset( - frame.stream_id(), - frame.reason(), - ))); + let error = Error::remote_reset(frame.stream_id(), frame.reason()); + // Preserve the received EOS while retaining the reset for the send half. + self.inner = Closed(if recv_end_stream { + Cause::ErrorAfterEndStream(error) + } else { + Cause::Error(error) + }); } } } @@ -356,7 +362,7 @@ impl State { pub fn is_local_error(&self) -> bool { match self.inner { - Closed(Cause::Error(ref e)) => e.is_local(), + Closed(Cause::Error(ref e) | Cause::ErrorAfterEndStream(ref e)) => e.is_local(), Closed(Cause::ScheduledLibraryReset(..)) => true, _ => false, } @@ -366,6 +372,11 @@ impl State { matches!( self.inner, Closed(Cause::Error(Error::Reset(_, _, Initiator::Remote))) + | Closed(Cause::ErrorAfterEndStream(Error::Reset( + _, + _, + Initiator::Remote + ))) ) } @@ -411,8 +422,11 @@ impl State { } pub fn is_recv_end_stream(&self) -> bool { - // In either case END_STREAM has been received - matches!(self.inner, Closed(Cause::EndStream) | HalfClosedRemote(..)) + // In each case END_STREAM has been received. + matches!( + self.inner, + Closed(Cause::EndStream | Cause::ErrorAfterEndStream(_)) | HalfClosedRemote(..) + ) } pub fn is_closed(&self) -> bool { @@ -437,7 +451,9 @@ impl State { Closed(Cause::ScheduledLibraryReset(reason)) => { Err(proto::Error::library_go_away(reason)) } - Closed(Cause::EndStream) | HalfClosedRemote(..) | ReservedLocal => Ok(false), + Closed(Cause::EndStream | Cause::ErrorAfterEndStream(_)) + | HalfClosedRemote(..) + | ReservedLocal => Ok(false), _ => Ok(true), } } @@ -446,9 +462,13 @@ impl State { pub(super) fn ensure_reason(&self, mode: PollReset) -> Result, crate::Error> { match self.inner { Closed(Cause::Error(Error::Reset(_, reason, _))) + | Closed(Cause::ErrorAfterEndStream(Error::Reset(_, reason, _))) | Closed(Cause::Error(Error::GoAway(_, reason, _))) + | Closed(Cause::ErrorAfterEndStream(Error::GoAway(_, reason, _))) | Closed(Cause::ScheduledLibraryReset(reason)) => Ok(Some(reason)), - Closed(Cause::Error(ref e)) => Err(e.clone().into()), + Closed(Cause::Error(ref e) | Cause::ErrorAfterEndStream(ref e)) => { + Err(e.clone().into()) + } Open { local: Streaming, .. } @@ -473,3 +493,30 @@ impl fmt::Debug for State { self.inner.fmt(f) } } + +#[cfg(test)] +mod tests { + use super::*; + use http::HeaderMap; + + #[test] + fn recv_reset_preserves_received_end_stream() { + let stream_id = StreamId::from(1); + let mut state = State::default(); + state.send_open(false).unwrap(); + + let mut headers = frame::Headers::new(stream_id, Default::default(), HeaderMap::new()); + headers.set_end_stream(); + state.recv_open(&headers).unwrap(); + assert!(state.is_recv_end_stream()); + + state.recv_reset(frame::Reset::new(stream_id, Reason::NO_ERROR), true); + + assert!(state.is_recv_end_stream()); + assert_eq!(state.ensure_recv_open().unwrap(), false); + assert_eq!( + state.ensure_reason(PollReset::Streaming).unwrap(), + Some(Reason::NO_ERROR) + ); + } +} From 8b46b233ddd21604036bc09442719b56853f6cd1 Mon Sep 17 00:00:00 2001 From: John Howard Date: Fri, 17 Jul 2026 13:22:45 -0700 Subject: [PATCH 2/2] add integration test --- tests/h2-tests/tests/stream_states.rs | 42 +++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/tests/h2-tests/tests/stream_states.rs b/tests/h2-tests/tests/stream_states.rs index c01b6daf..ab1603ef 100644 --- a/tests/h2-tests/tests/stream_states.rs +++ b/tests/h2-tests/tests/stream_states.rs @@ -876,6 +876,48 @@ fn exceed_max_streams() { } */ +#[tokio::test] +async fn recv_end_stream_survives_reset() { + 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("POST", "https://example.com/")) + .await; + srv.send_frame(frames::headers(1).response(200).eos()).await; + srv.send_frame(frames::reset(1).reason(Reason::NO_ERROR)) + .await; + }; + + let client = async move { + let (mut client, mut conn) = client::handshake(io).await.expect("handshake"); + let request = Request::builder() + .method(Method::POST) + .uri("https://example.com/") + .body(()) + .unwrap(); + let (response, mut send_stream) = client.send_request(request, false).unwrap(); + + // Process the reset before polling the response future. + let reason = conn + .drive(poll_fn(move |cx| send_stream.poll_reset(cx))) + .await + .unwrap(); + assert_eq!(reason, Reason::NO_ERROR); + + let response = response.await.unwrap(); + assert!(response.body().is_end_stream()); + + let mut body = response.into_body(); + assert!(body.data().await.is_none()); + assert!(body.trailers().await.unwrap().is_none()); + }; + + join(srv, client).await; +} + #[tokio::test] async fn rst_while_closing() { // Test to reproduce panic in issue #246 --- receipt of a RST_STREAM frame