fix: preserve received END_STREAM across HTTP/2 stream resets

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

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