diff --git a/crates/common/src/compression/zlib.rs b/crates/common/src/compression/zlib.rs
index 6c34d7cbe55..5c90c41d8cb 100644
--- a/crates/common/src/compression/zlib.rs
+++ b/crates/common/src/compression/zlib.rs
@@ -350,9 +350,6 @@ fn decompress_chunks(
     max_length: Option<usize>,
     calc_flush: impl Fn(bool) -> InflateFlush,
 ) -> Result<(Vec<u8>, bool), String> {
-    if data.is_empty() {
-        return Ok((Vec::new(), false));
-    }
     let max_length = max_length.unwrap_or(usize::MAX);
     let mut buf = Vec::new();
 
@@ -378,7 +375,9 @@ fn decompress_chunks(
             match res {
                 Ok(status) => {
                     let stream_end = status == Status::StreamEnd;
-                    if stream_end || data.is_empty() {
+                    // A full output buffer can leave output pending in inflate,
+                    // even with no input left: only a partial buffer ends it.
+                    if stream_end || (data.is_empty() && produced < additional) {
                         buf.shrink_to_fit();
                         return Ok((buf, stream_end));
                     } else if !chunk.is_empty() && consumed == 0 {
@@ -1022,4 +1021,56 @@ mod tests {
         assert_eq!(d.unconsumed_tail(), b"");
         assert_eq!(d.unused_data(), b"NEXT");
     }
+
+    /// Decode raw deflate of `b"x" * 1168` in 100-byte steps until all
+    /// input is consumed. Inflate then still holds the last 68 bytes.
+    fn drained_raw_deflate() -> (Decompressor, Vec<u8>) {
+        let mut co = Compressor::new(6, 8, -MAX_WBITS, 8, 0, None).unwrap();
+        let mut raw = co.compress(&[b'x'; 1168]).unwrap();
+        raw.extend(co.flush(Z_FINISH).unwrap());
+        let mut d = Decompressor::new(-MAX_WBITS, None).unwrap();
+        let mut out = d.decompress(&raw, Some(100)).unwrap();
+        while !d.unconsumed_tail().is_empty() {
+            let tail = d.unconsumed_tail().to_vec();
+            out.extend(d.decompress(&tail, Some(100)).unwrap());
+        }
+        assert_eq!(out.len(), 1100);
+        assert!(!d.eof());
+        (d, out)
+    }
+
+    #[test]
+    fn empty_input_returns_pending_output() {
+        let (mut d, mut out) = drained_raw_deflate();
+        out.extend(d.decompress(b"", Some(100)).unwrap());
+        assert_eq!(out, [b'x'; 1168]);
+        assert!(d.eof());
+    }
+
+    #[test]
+    fn flush_returns_pending_output() {
+        let (mut d, mut out) = drained_raw_deflate();
+        out.extend(d.flush(DEF_BUF_SIZE).unwrap());
+        assert_eq!(out, [b'x'; 1168]);
+        assert!(d.eof());
+    }
+
+    #[test]
+    fn small_bufsize_returns_pending_output() {
+        let data = [b'x'; 1168];
+        let raw = compress(&data, 6, -MAX_WBITS).unwrap();
+        for bufsize in [1, 100] {
+            assert_eq!(decompress(&raw, -MAX_WBITS, bufsize).unwrap(), data);
+        }
+    }
+
+    #[test]
+    fn empty_flush_does_not_finish_a_stream() {
+        let encoded = compress(b"later input", -1, MAX_WBITS).unwrap();
+        let mut d = Decompressor::new(MAX_WBITS, None).unwrap();
+        assert_eq!(d.flush(DEF_BUF_SIZE).unwrap(), b"");
+        assert!(!d.eof());
+        assert_eq!(d.decompress(&encoded, None).unwrap(), b"later input");
+        assert!(d.eof());
+    }
 }
diff --git a/extra_tests/snippets/stdlib_zlib.py b/extra_tests/snippets/stdlib_zlib.py
index dfadf7eebdc..cd3f91b61d5 100644
--- a/extra_tests/snippets/stdlib_zlib.py
+++ b/extra_tests/snippets/stdlib_zlib.py
@@ -79,3 +79,45 @@
 assert d.flush() == b""
 assert d.unconsumed_tail == tail
 assert d.unused_data == b"NEXT" + tail
+
+
+# Raw deflate has no trailer, so inflate can consume the last input byte and
+# still hold output. decompress(b"", n) and flush() must return that output.
+def drained_raw_deflate():
+    c = zlib.compressobj(6, zlib.DEFLATED, -15)
+    raw = c.compress(b"x" * 1168) + c.flush()
+    d = zlib.decompressobj(-15)
+    out = d.decompress(raw, 100)
+    while d.unconsumed_tail:
+        out += d.decompress(d.unconsumed_tail, 100)
+    assert len(out) == 1100
+    assert not d.eof
+    return d, out
+
+
+d, out = drained_raw_deflate()
+more = d.decompress(b"", 100)
+assert len(more) == 68
+assert d.eof
+assert d.flush() == b""
+assert out + more == b"x" * 1168
+
+d, out = drained_raw_deflate()
+rest = d.flush()
+assert len(rest) == 68
+assert d.eof
+assert out + rest == b"x" * 1168
+
+# An empty input before any compressed data does not end the stream.
+d = zlib.decompressobj()
+assert d.decompress(b"") == b""
+assert not d.eof
+assert d.flush() == b""
+assert not d.eof
+assert d.decompress(zlib.compress(b"later input")) == b"later input"
+assert d.eof
+
+# The one-shot decompress drains the same pending output with a small bufsize.
+raw = zlib.compress(b"x" * 1168, 6, -15)
+for bufsize in (1, 100):
+    assert zlib.decompress(raw, -15, bufsize) == b"x" * 1168