set a minimal size in the compression output buffer

.changesets/fix_geal_compression_minimal_output_size.md

@@ -0,0 +1,7 @@
+### Fix hang and high CPU usage when compressing small responses ([PR #3961](https://github.com/apollographql/router/pull/3961))
+
+When returning small responses (less than 10 bytes) and compressing them using gzip, the router could go into an infinite loop
+
+---
+
+By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/pull/3961

apollo-router/src/axum_factory/compression/mod.rs

@@ -19,6 +19,8 @@ pub(crate) mod codec;
 pub(crate) mod unshared;
 pub(crate) mod util;
 
+const GZIP_HEADER_LEN: usize = 10;
+
 pub(crate) enum Compressor {
     Deflate(DeflateEncoder),
     Gzip(GzipEncoder),
@@ -79,7 +81,9 @@ where {
                         }
                     }
                     Ok(data) => {
-                        let mut buf = BytesMut::zeroed(data.len());
+                        // the buffer needs at least 10 bytes for a gzip header if we use gzip, then more
+                        // room to store the data itself
+                        let mut buf = BytesMut::zeroed(GZIP_HEADER_LEN + data.len());
 
                         let mut partial_input = PartialBuffer::new(&*data);
                         let mut partial_output = PartialBuffer::new(&mut buf);
@@ -230,6 +234,27 @@ mod tests {
         assert!(stream.next().await.is_none());
     }
 
+    #[tokio::test]
+    async fn small_input() {
+        let compressor = Compressor::new(["gzip"].into_iter()).unwrap();
+
+        let body: Body = vec![0u8, 1, 2, 3].into();
+
+        let mut stream = compressor.process(body);
+        let mut decoder = GzipDecoder::new(Vec::new());
+
+        while let Some(buf) = stream.next().await {
+            let b = buf.unwrap();
+            decoder.write_all(&b).await.unwrap();
+        }
+
+        decoder.shutdown().await.unwrap();
+        let response = decoder.into_inner();
+        assert_eq!(response, [0u8, 1, 2, 3]);
+
+        assert!(stream.next().await.is_none());
+    }
+
     #[tokio::test]
     async fn gzip_header_writing() {
         let compressor = Compressor::new(["gzip"].into_iter()).unwrap();