Fix/http encoder stream flush

actix-http/CHANGES.md

@@ -2,6 +2,8 @@
 
 ## Unreleased
 
+- Flush compression encoders when a write does not produce output so that streaming response bodies (SSE, for example) are delivered incrementally instead of being buffered until the stream ends.
+
 ## 3.13.5
 
 - Reject invalid WebSocket close-frame status codes, malformed payloads, and invalid UTF-8 close reasons.
@@ -11,7 +13,6 @@
 
 - Reject WebSocket frames with reserved bits.
 - Reject new WebSocket text or binary frames during a continuation.
-
 ## 3.13.3
 
 - Close idle HTTP/1 keep-alive connections during graceful server shutdown and close active connections after their current request finishes.

actix-http/src/encoding/encoder.rs

@@ -208,7 +208,18 @@ where
                     if let Some(mut encoder) = this.encoder.take() {
                         if chunk.len() < MAX_CHUNK_SIZE_ENCODE_IN_PLACE {
                             encoder.write(&chunk).map_err(EncoderError::Io)?;
-                            let chunk = encoder.take();
+                            let mut chunk = encoder.take();
+
+                            if chunk.is_empty() {
+                                // Small chunks (SSE events, for example) do not
+                                // produce compressed output on their own; the
+                                // encoder buffers them internally. Flush so the
+                                // client receives each chunk as it is produced
+                                // instead of only when the stream ends.
+                                encoder.flush().map_err(EncoderError::Io)?;
+                                chunk = encoder.take();
+                            }
+
                             *this.encoder = Some(encoder);
 
                             if !chunk.is_empty() {
@@ -217,6 +228,11 @@ where
                         } else {
                             *this.fut = Some(spawn_blocking(move || {
                                 encoder.write(&chunk)?;
+
+                                if encoder.output_is_empty() {
+                                    encoder.flush()?;
+                                }
+
                                 Ok(encoder)
                             }));
                         }
@@ -360,6 +376,46 @@ impl ContentEncoder {
         }
     }
 
+    /// Flushes the encoder so compressed output for already written data is
+    /// made available to `take()`.
+    ///
+    /// Stream encoders buffer internally and small writes do not produce
+    /// output; without an explicit flush, a streaming body's chunks only
+    /// reach the client when `finish()` is called at stream end.
+    fn flush(&mut self) -> Result<(), io::Error> {
+        match *self {
+            #[cfg(feature = "compress-brotli")]
+            ContentEncoder::Brotli(ref mut encoder) => encoder.flush(),
+
+            #[cfg(feature = "compress-gzip")]
+            ContentEncoder::Deflate(ref mut encoder) => encoder.flush(),
+
+            #[cfg(feature = "compress-gzip")]
+            ContentEncoder::Gzip(ref mut encoder) => encoder.flush(),
+
+            #[cfg(feature = "compress-zstd")]
+            ContentEncoder::Zstd(ref mut encoder) => encoder.flush(),
+        }
+    }
+
+    /// Returns `true` if the encoder has produced no output yet, i.e. all
+    /// data written so far is still buffered internally.
+    fn output_is_empty(&self) -> bool {
+        match *self {
+            #[cfg(feature = "compress-brotli")]
+            ContentEncoder::Brotli(ref encoder) => encoder.get_ref().buf.is_empty(),
+
+            #[cfg(feature = "compress-gzip")]
+            ContentEncoder::Deflate(ref encoder) => encoder.get_ref().buf.is_empty(),
+
+            #[cfg(feature = "compress-gzip")]
+            ContentEncoder::Gzip(ref encoder) => encoder.get_ref().buf.is_empty(),
+
+            #[cfg(feature = "compress-zstd")]
+            ContentEncoder::Zstd(ref encoder) => encoder.get_ref().buf.is_empty(),
+        }
+    }
+
     fn write(&mut self, data: &[u8]) -> Result<(), io::Error> {
         match *self {
             #[cfg(feature = "compress-brotli")]

actix-web/tests/compression.rs

@@ -326,3 +326,129 @@
 
     srv.stop().await;
 }
+
+#[actix_rt::test]
+async fn gzip_flushes_small_streaming_chunks() {
+    // Regression: small chunks (SSE events, for example) produce no compressed
+    // output on their own; the encoders buffer them internally. Without a
+    // flush when a write yields no output, the whole body only reaches the
+    // client when the stream ends.
+    use std::{
+        convert::Infallible,
+        sync::{Arc, Mutex},
+    };
+
+    use futures_util::{stream, StreamExt};
+
+    let (tx, rx) = tokio::sync::oneshot::channel::<()>();
+    // The server factory must be Clone; only the (single) worker's handler
+    // takes the receiver out of the shared slot.
+    let rx = Arc::new(Mutex::new(Some(rx)));
+
+    // The first two events are produced back to back; the third one is held
+    // back behind the gate. The encoder's first write only emits the gzip
+    // header, so streaming starts with the second write's flush.
+    let srv = actix_test::start(move || {
+        let rx = Arc::clone(&rx);
+        App::new().wrap(Compress::default()).route(
+            "/gated",
+            web::to(move || {
+                let rx = Arc::clone(&rx);
+                async move {
+                    let rx = rx.lock().unwrap().take();
+
+                    let body = stream::unfold((rx, 0u8), |(rx, step)| async move {
+                        match step {
+                            0 => Some((
+                                Ok::<_, Infallible>(Bytes::from_static(b"data: first\n\n")),
+                                (rx, 1),
+                            )),
+                            1 => Some((
+                                Ok::<_, Infallible>(Bytes::from_static(b"data: second\n\n")),
+                                (rx, 2),
+                            )),
+                            2 => {
+                                let _ = rx.unwrap().await;
+                                Some((
+                                    Ok::<_, Infallible>(Bytes::from_static(b"data: third\n\n")),
+                                    (None, 3),
+                                ))
+                            }
+                            _ => None,
+                        }
+                    });
+
+                    HttpResponse::Ok()
+                        .content_type("text/event-stream")
+                        .streaming(Box::pin(body))
+                }
+            }),
+        )
+    });
+
+    let mut res = srv
+        .post("/gated")
+        .no_decompress()
+        .insert_header((header::ACCEPT_ENCODING, "gzip"))
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(res.status(), StatusCode::OK);
+    assert_eq!(res.headers().get(header::CONTENT_ENCODING).unwrap(), "gzip");
+
+    // The gate has not been released yet. Keep reading with a timeout: with
+    // the flush in place the first two events arrive as compressed data; a
+    // buffering encoder only ever sends the bare 10-byte gzip header here.
+    use std::io::Read as _;
+
+    let mut raw = Vec::new();
+    let first_events: &[u8] = b"data: first\n\ndata: second\n\n";
+    let mut decompressed = Vec::new();
+
+    loop {
+        let chunk = actix_rt::time::timeout(std::time::Duration::from_secs(3), res.next())
+            .await
+            .expect("compressed data for the first events was not flushed before the stream ended")
+            .unwrap()
+            .unwrap();
+        raw.extend_from_slice(&chunk);
+
+        let mut decoder = flate2::read::GzDecoder::new(&raw[..]);
+        let mut buf = [0u8; 128];
+        loop {
+            let n = decoder.read(&mut buf).unwrap_or(0);
+            if n == 0 {
+                break;
+            }
+            decompressed.extend_from_slice(&buf[..n]);
+            if decompressed.len() >= first_events.len() {
+                break;
+            }
+        }
+
+        if !decompressed.is_empty() {
+            break;
+        }
+    }
+
+    assert_eq!(
+        &decompressed[..],
+        &first_events[..],
+        "first flushed chunk did not decompress to the first two events"
+    );
+
+    // Release the last event and collect the rest of the body.
+    tx.send(()).unwrap();
+
+    while let Some(chunk) = res.next().await {
+        raw.extend_from_slice(&chunk.unwrap());
+    }
+
+    assert_eq!(
+        utils::gzip::decode(raw),
+        &b"data: first\n\ndata: second\n\ndata: third\n\n"[..]
+    );
+
+    srv.stop().await;
+}