fix: pending-accept remotely-reset streams pattern was checking is_local

src/proto/streams/state.rs

@@ -362,7 +362,7 @@ impl State {
 
     pub fn is_remote_reset(&self) -> bool {
         match self.inner {
-            Closed(Cause::Error(ref e)) => e.is_local(),
+            Closed(Cause::Error(ref e)) => !e.is_local(),
             _ => false,
         }
     }

tests/h2-tests/tests/server.rs

@@ -879,6 +879,40 @@ async fn too_big_headers_sends_reset_after_431_if_not_eos() {
     join(client, srv).await;
 }
 
+#[tokio::test]
+async fn pending_accept_recv_illegal_content_length_data() {
+    h2_support::trace_init!();
+    let (io, mut client) = mock::new();
+
+    let client = async move {
+        let settings = client.assert_server_handshake().await;
+        assert_default_settings!(settings);
+        client
+            .send_frame(
+                frames::headers(1)
+                    .request("POST", "https://a.b")
+                    .field("content-length", "1"),
+            )
+            .await;
+        client
+            .send_frame(frames::data(1, &b"hello"[..]).eos())
+            .await;
+        client.recv_frame(frames::reset(1).protocol_error()).await;
+        idle_ms(10).await;
+    };
+
+    let srv = async move {
+        let mut srv = server::Builder::new()
+            .handshake::<_, Bytes>(io)
+            .await
+            .expect("handshake");
+
+        let _req = srv.next().await.expect("req").expect("is_ok");
+    };
+
+    join(client, srv).await;
+}
+
 #[tokio::test]
 async fn poll_reset() {
     h2_support::trace_init!();

tests/h2-tests/tests/stream_states.rs

@@ -235,6 +235,47 @@ async fn reset_streams_dont_grow_memory_continuously() {
     join(srv, client).await;
 }
 
+#[tokio::test]
+async fn pending_accept_reset_streams_decrement_too() {
+    h2_support::trace_init!();
+    let (io, mut client) = mock::new();
+
+    // If it didn't decrement internally, this would eventually get
+    // the count over MAX.
+    const M: usize = 2;
+    const N: usize = 5;
+    const MAX: usize = 6;
+
+    let client = async move {
+        let settings = client.assert_server_handshake().await;
+        assert_default_settings!(settings);
+        let mut id = 1;
+        for _ in 0..M {
+            for _ in 0..N {
+                client
+                    .send_frame(frames::headers(id).request("GET", "https://a.b/").eos())
+                    .await;
+                client.send_frame(frames::reset(id).protocol_error()).await;
+                id += 2;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        }
+    };
+
+    let srv = async move {
+        let mut srv = server::Builder::new()
+            .max_pending_accept_reset_streams(MAX)
+            .handshake::<_, Bytes>(io)
+            .await
+            .expect("handshake");
+
+        while let Some(Ok(_)) = srv.accept().await {}
+
+        poll_fn(|cx| srv.poll_closed(cx)).await.expect("server");
+    };
+    join(srv, client).await;
+}
+
 #[tokio::test]
 async fn errors_if_recv_frame_exceeds_max_frame_size() {
     h2_support::trace_init!();