fix(core): do not reuse an HTTP/1 connection after a 204 that frames a body

pingora-core/src/connectors/http/v1.rs

@@ -97,6 +97,7 @@ mod tests {
     use crate::upstreams::peer::HttpPeer;
     use crate::upstreams::peer::Peer;
     use pingora_http::RequestHeader;
+    use rstest::rstest;
     use std::fmt::{Display, Formatter, Result as FmtResult};
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
@@ -170,6 +171,71 @@ ZZ\r\n",
         server.abort();
     }
 
+    #[rstest]
+    #[case("Content-Length: 128")]
+    #[case("Transfer-Encoding: chunked")]
+    #[tokio::test]
+    async fn no_content_response_with_body_framing_is_not_returned_to_pool(#[case] framing: &str) {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let address = listener.local_addr().unwrap();
+        let response = format!("HTTP/1.1 204 No Content\r\n{framing}\r\n\r\n");
+        let server = tokio::spawn(async move {
+            let (mut stream, _) = listener.accept().await.unwrap();
+            let mut request_byte = [0];
+            stream.read_exact(&mut request_byte).await.unwrap();
+            // Only the header section is sent. The body it frames, which a 204 must
+            // not have, could still follow once the connection has been released.
+            stream.write_all(response.as_bytes()).await.unwrap();
+            std::future::pending::<()>().await;
+        });
+
+        let connector = Connector::new(None);
+        let peer = HttpPeer::new(address, false, "".into());
+        let (mut http, reused) = connector.get_http_session(&peer).await.unwrap();
+        assert!(!reused);
+
+        let request = RequestHeader::build("GET", b"/", None).unwrap();
+        http.write_request_header(Box::new(request)).await.unwrap();
+        http.read_response().await.unwrap();
+        assert_eq!(http.get_status().unwrap(), 204);
+        assert!(http.read_body_bytes().await.unwrap().is_none());
+
+        connector.release_http_session(http, &peer, None).await;
+        assert!(connector.reused_http_session(&peer).await.is_none());
+        server.abort();
+    }
+
+    #[tokio::test]
+    async fn no_content_response_with_zero_content_length_is_returned_to_pool() {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let address = listener.local_addr().unwrap();
+        let server = tokio::spawn(async move {
+            let (mut stream, _) = listener.accept().await.unwrap();
+            let mut request_byte = [0];
+            stream.read_exact(&mut request_byte).await.unwrap();
+            stream
+                .write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n")
+                .await
+                .unwrap();
+            std::future::pending::<()>().await;
+        });
+
+        let connector = Connector::new(None);
+        let peer = HttpPeer::new(address, false, "".into());
+        let (mut http, reused) = connector.get_http_session(&peer).await.unwrap();
+        assert!(!reused);
+
+        let request = RequestHeader::build("GET", b"/", None).unwrap();
+        http.write_request_header(Box::new(request)).await.unwrap();
+        http.read_response().await.unwrap();
+        assert_eq!(http.get_status().unwrap(), 204);
+        assert!(http.read_body_bytes().await.unwrap().is_none());
+
+        connector.release_http_session(http, &peer, None).await;
+        assert!(connector.reused_http_session(&peer).await.is_some());
+        server.abort();
+    }
+
     #[tokio::test]
     async fn response_parser_config_is_applied_to_new_sessions() {
         let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();

pingora-core/src/protocols/http/v1/client.rs

@@ -602,6 +602,18 @@ impl HttpSession {
             self.set_keepalive(None);
             return;
         }
+        // Per [RFC 9112 Section 6.3](https://datatracker.ietf.org/doc/html/rfc9112#section-6.3),
+        // a 204 response ends at the first empty line after the header fields regardless of
+        // the fields present, and a server MUST NOT send Transfer-Encoding or Content-Length
+        // with it ([RFC 9112 Section 6.1](https://datatracker.ietf.org/doc/html/rfc9112#section-6.1),
+        // [RFC 9110 Section 8.6](https://datatracker.ietf.org/doc/html/rfc9110#section-8.6)).
+        // A server that frames a body on a 204 anyway may send it after the header section.
+        // Bytes that arrive once the connection is back in the pool would be read as the
+        // response to the next request, so such a connection must not be reused.
+        if self.get_status() == Some(StatusCode::NO_CONTENT) && self.resp_declares_body() {
+            self.set_keepalive(None);
+            return;
+        }
         if let Some(keepalive) = self.is_connection_keepalive() {
             if keepalive {
                 let (timeout, _max_use) = self.get_keepalive_values();
@@ -788,6 +800,17 @@ impl HttpSession {
             .unwrap_or(false)
     }
 
+    /// Whether the response header fields announce a body: Transfer-Encoding is present, or
+    /// Content-Length is present with a value other than zero.
+    fn resp_declares_body(&self) -> bool {
+        let Some(headers) = self.resp_header().map(|h| &h.headers) else {
+            return false;
+        };
+        headers.contains_key(header::TRANSFER_ENCODING)
+            || (headers.contains_key(header::CONTENT_LENGTH)
+                && content_length_parse_all(headers) != Some(0))
+    }
+
     fn init_req_body_writer(&mut self, header: &RequestHeader) {
         self.init_body_writer_comm(&header.headers)
     }
@@ -2429,6 +2452,45 @@ hello\r\n\
         assert!(http_stream.will_keepalive());
     }
 
+    #[rstest]
+    #[case("Content-Length: 128")]
+    #[case("Content-Length: 1")]
+    #[case("Transfer-Encoding: chunked")]
+    #[case("Transfer-Encoding: chunked\r\nContent-Length: 0")]
+    #[tokio::test]
+    async fn no_content_response_with_body_framing_disables_keepalive(#[case] framing: &str) {
+        init_log();
+        // A 204 has no body, but this server frames one anyway. Its bytes may only
+        // show up after the connection is released, so the connection is not reusable.
+        let input =
+            format!("HTTP/1.1 204 No Content\r\n{framing}\r\nConnection: keep-alive\r\n\r\n");
+        let mock_io = Builder::new().read(input.as_bytes()).build();
+        let mut http_stream = HttpSession::new(Box::new(mock_io));
+        http_stream.read_response().await.unwrap();
+        assert!(http_stream.is_body_done());
+
+        http_stream.respect_keepalive();
+        assert!(!http_stream.will_keepalive());
+        assert!(http_stream.reuse().await.is_none());
+    }
+
+    #[rstest]
+    #[case("")]
+    #[case("Content-Length: 0\r\n")]
+    #[tokio::test]
+    async fn no_content_response_without_body_framing_allows_keepalive(#[case] framing: &str) {
+        init_log();
+        let input = format!("HTTP/1.1 204 No Content\r\n{framing}\r\n");
+        let mock_io = Builder::new().read(input.as_bytes()).build();
+        let mut http_stream = HttpSession::new(Box::new(mock_io));
+        http_stream.read_response().await.unwrap();
+        assert!(http_stream.is_body_done());
+
+        http_stream.respect_keepalive();
+        assert!(http_stream.will_keepalive());
+        assert!(http_stream.reuse().await.is_some());
+    }
+
     #[tokio::test]
     async fn test_malformed_chunked_response_is_not_reusable() {
         let input = b"HTTP/1.1 503 Service Unavailable\r\nTransfer-Encoding: chunked\r\n\r\nZZ\r\n";