add h2 server session tests

 - test content-length=0 header
 - test HEADERS frames without EOS follows and an empty DATA frame with EOS

3 files changed, 148 insertions, 3 deletions

diff --git a/.bleep b/.bleep
index f999f12..265906b 100644
--- a/.bleep
+++ b/.bleep
@@ -1 +1 @@
-32635fba2c86e17defba5601199d11373dce0bea
\ No newline at end of file
+0875924524a7338c15784029df5306dd08e981dd
\ No newline at end of file
diff --git a/pingora-core/src/protocols/http/v2/server.rs b/pingora-core/src/protocols/http/v2/server.rs
index 718ddde..5f38542 100644
--- a/pingora-core/src/protocols/http/v2/server.rs
+++ b/pingora-core/src/protocols/http/v2/server.rs
@@ -579,4 +579,149 @@ mod test {
             assert!(handle.await.is_ok());
         }
     }
+
+    #[tokio::test]
+    async fn test_req_content_length_eq_0_and_no_header_eos() {
+        let (client, server) = duplex(65536);
+
+        let server_body = "test server body";
+
+        let mut handles = vec![];
+
+        handles.push(tokio::spawn(async move {
+            let (h2, connection) = h2::client::handshake(client).await.unwrap();
+            tokio::spawn(async move {
+                connection.await.unwrap();
+            });
+
+            let mut h2 = h2.ready().await.unwrap();
+
+            let request = Request::builder()
+                .method(Method::POST)
+                .uri("https://www.example.com/")
+                .header("content-length", "0") // explicitly set
+                .body(())
+                .unwrap();
+
+            let (response, mut req_body) = h2.send_request(request, false).unwrap(); // no EOS
+
+            let (head, mut body) = response.await.unwrap().into_parts();
+
+            assert_eq!(head.status, 200);
+            let data = body.data().await.unwrap().unwrap();
+            assert_eq!(data, server_body);
+
+            req_body.send_data("".into(), true).unwrap(); // set EOS after read the resp body
+        }));
+
+        let mut connection = handshake(Box::new(server), None).await.unwrap();
+        let digest = Arc::new(Digest::default());
+
+        while let Some(mut http) = HttpSession::from_h2_conn(&mut connection, digest.clone())
+            .await
+            .unwrap()
+        {
+            handles.push(tokio::spawn(async move {
+                let req = http.req_header();
+                assert_eq!(req.method, Method::POST);
+                assert_eq!(req.uri, "https://www.example.com/");
+
+                // 1. Check body related methods
+                http.enable_retry_buffering();
+                assert!(http.is_body_empty());
+                assert!(http.is_body_done());
+                let retry_body = http.get_retry_buffer();
+                assert!(retry_body.is_none());
+
+                // 2. Send response
+                let response_header = Box::new(ResponseHeader::build(200, None).unwrap());
+                assert!(http
+                    .write_response_header(response_header.clone(), false)
+                    .is_ok());
+
+                http.write_body(server_body.into(), false).unwrap();
+                assert_eq!(http.body_bytes_sent(), 16);
+
+                // 3. Waiting for the reset from the client
+                assert!(http.read_body_or_idle(http.is_body_done()).await.is_err());
+            }));
+        }
+
+        for handle in handles {
+            // ensure no panics
+            assert!(handle.await.is_ok());
+        }
+    }
+
+    #[tokio::test]
+    async fn test_req_header_no_eos_empty_data_with_eos() {
+        let (client, server) = duplex(65536);
+
+        let server_body = "test server body";
+
+        let mut handles = vec![];
+
+        handles.push(tokio::spawn(async move {
+            let (h2, connection) = h2::client::handshake(client).await.unwrap();
+            tokio::spawn(async move {
+                connection.await.unwrap();
+            });
+
+            let mut h2 = h2.ready().await.unwrap();
+
+            let request = Request::builder()
+                .method(Method::POST)
+                .uri("https://www.example.com/")
+                .body(())
+                .unwrap();
+
+            let (response, mut req_body) = h2.send_request(request, false).unwrap(); // no EOS
+
+            let (head, mut body) = response.await.unwrap().into_parts();
+
+            assert_eq!(head.status, 200);
+            let data = body.data().await.unwrap().unwrap();
+            assert_eq!(data, server_body);
+
+            req_body.send_data("".into(), true).unwrap(); // set EOS after read the resp body
+        }));
+
+        let mut connection = handshake(Box::new(server), None).await.unwrap();
+        let digest = Arc::new(Digest::default());
+
+        while let Some(mut http) = HttpSession::from_h2_conn(&mut connection, digest.clone())
+            .await
+            .unwrap()
+        {
+            handles.push(tokio::spawn(async move {
+                let req = http.req_header();
+                assert_eq!(req.method, Method::POST);
+                assert_eq!(req.uri, "https://www.example.com/");
+
+                // 1. Check body related methods
+                http.enable_retry_buffering();
+                assert!(!http.is_body_empty());
+                assert!(!http.is_body_done());
+                let retry_body = http.get_retry_buffer();
+                assert!(retry_body.is_none());
+
+                // 2. Send response
+                let response_header = Box::new(ResponseHeader::build(200, None).unwrap());
+                assert!(http
+                    .write_response_header(response_header.clone(), false)
+                    .is_ok());
+
+                http.write_body(server_body.into(), false).unwrap();
+                assert_eq!(http.body_bytes_sent(), 16);
+
+                // 3. Waiting for the client to close stream.
+                http.read_body_or_idle(http.is_body_done()).await.unwrap();
+            }));
+        }
+
+        for handle in handles {
+            // ensure no panics
+            assert!(handle.await.is_ok());
+        }
+    }
 }
diff --git a/pingora-proxy/src/proxy_h1.rs b/pingora-proxy/src/proxy_h1.rs
index e13d4c1..d323562 100644
--- a/pingora-proxy/src/proxy_h1.rs
+++ b/pingora-proxy/src/proxy_h1.rs
@@ -311,9 +311,9 @@ impl<SV> HttpProxy<SV> {
                 },
 
                 _ = tx.reserve(), if downstream_state.is_reading() && send_permit.is_err() => {
-                    // If tx is closed, downstream already finish its job.
+                    // If tx is closed, the upstream has already finished its job.
                     downstream_state.maybe_finished(tx.is_closed());
-                    debug!("waiting for permit {send_permit:?}, downstream closed {}", tx.is_closed());
+                    debug!("waiting for permit {send_permit:?}, upstream closed {}", tx.is_closed());
                     /* No permit, wait on more capacity to avoid starving.
                      * Otherwise this select only blocks on rx, which might send no data
                      * before the entire body is uploaded.