diff --git a/actix-http/CHANGES.md b/actix-http/CHANGES.md index 5775c17d344..6c7f8ef8c1d 100644 --- a/actix-http/CHANGES.md +++ b/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. diff --git a/actix-http/src/encoding/encoder.rs b/actix-http/src/encoding/encoder.rs index aa730cfaf8a..094b300a234 100644 --- a/actix-http/src/encoding/encoder.rs +++ b/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")] diff --git a/actix-web/tests/compression.rs b/actix-web/tests/compression.rs index 61ff1bff54a..ffdac8dafd7 100644 --- a/actix-web/tests/compression.rs +++ b/actix-web/tests/compression.rs @@ -326,3 +326,129 @@ async fn deny_identity_for_manual_coding() { 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; +}