tests: handle h2 reset timing after partial response body

Allow the h2 proxy test to pass whether the response body DATA frame is
observed before the downstream stream reset or the stream reset arrives first.
Both orderings are valid for this error path, and newer h2 behavior can surface
RST_STREAM(CANCEL) before the buffered body chunk is delivered.

Also keep the h2 drain-loop cleanup explicit by using clearer expect messages
and releasing flow-control capacity while draining the response.

Author: areyia

pingora-core/src/protocols/http/v2/server.rs

@@ -994,6 +994,17 @@ mod test {
             assert_eq!(data, server_body);
 
             req_body.send_data("".into(), true).unwrap(); // set EOS after read the resp body
+
+            // Drain the response to EOS before dropping the stream. Newer h2
+            // sends RST_STREAM(CANCEL) when a still-open recv stream is dropped,
+            // which would race with the server reading the request EOS and turn
+            // the server-side read into a stream-reset error.
+            while let Some(chunk) = body.data().await {
+                let chunk = chunk.expect("response body error");
+                body.flow_control()
+                    .release_capacity(chunk.len())
+                    .expect("release capacity");
+            }
         }));
 
         let mut connection = handshake(Box::new(server), None).await.unwrap();
@@ -1024,8 +1035,12 @@ mod test {
                 http.write_body(server_body.into(), false).await.unwrap();
                 assert_eq!(http.body_bytes_sent(), 16);
 
-                // 3. Waiting for the client to close stream.
+                // 3. Read the empty DATA frame carrying the request EOS.
                 http.read_body_or_idle(http.is_body_done()).await.unwrap();
+
+                // 4. Finish the response so the client can drain it to EOS and
+                //    close the stream cleanly instead of cancelling it.
+                http.finish().unwrap();
             }));
         }
 

pingora-proxy/tests/test_basic.rs

@@ -1108,11 +1108,21 @@ async fn test_error_after_headers_sent_rst_received() {
     let response = response.await.unwrap();
     let mut body = response.into_body();
 
-    let chunk = body.data().await.unwrap();
-    assert_eq!(chunk.unwrap(), Bytes::from_static(b"AAAAA"));
-
-    let err = body.data().await.unwrap().err().unwrap();
-    assert_eq!(err.reason().unwrap(), h2::Reason::CANCEL);
+    match body.data().await.expect("response body frame or reset") {
+        Ok(chunk) => {
+            assert_eq!(chunk, Bytes::from_static(b"AAAAA"));
+
+            let err = body
+                .data()
+                .await
+                .expect("response body reset")
+                .expect_err("expected stream reset");
+            assert_eq!(err.reason().expect("reset reason"), h2::Reason::CANCEL);
+        }
+        Err(err) => {
+            assert_eq!(err.reason().expect("reset reason"), h2::Reason::CANCEL);
+        }
+    }
 }
 
 #[tokio::test]