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) + ); + } +} 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