gh-159999: Fix needs_input for bz2 and zlib decompressors

Lib/gzip.py

@@ -612,6 +612,9 @@ def read(self, size=-1):
             # Read a chunk of data from the file
             if self._decompressor.needs_input:
                 buf = self._fp.read(READ_BUFFER_SIZE)
+                if buf == b"":
+                    raise EOFError("Compressed file ended before the "
+                                   "end-of-stream marker was reached")
             else:
                 buf = b""
 
@@ -623,9 +626,6 @@ def read(self, size=-1):
 
             if uncompress != b"":
                 break
-            if buf == b"":
-                raise EOFError("Compressed file ended before the "
-                               "end-of-stream marker was reached")
 
         self._crc = zlib.crc32(uncompress, self._crc)
         self._stream_size += len(uncompress)

Lib/test/test_bz2.py

@@ -967,6 +967,34 @@ def testDecompressorChunksMaxsize(self):
         self.assertEqual(out, self.BIG_TEXT)
         self.assertEqual(bzd.unused_data, b"")
 
+    def test_decompressor_needs_input_with_buffered_output(self):
+        # gh-159999: needs_input must be false if all input was consumed
+        # but max_length prevented returning all output.
+        data = b"\x00\x10 0@" * 2
+        bzd = BZ2Decompressor()
+        self.assertEqual(bzd.decompress(bz2.compress(data)[:40], max_length=0),
+                         b"")
+        self.assertFalse(bzd.needs_input)
+        self.assertEqual(bzd.decompress(b""), data)
+
+        # Without max_length, decompress() must return all output available
+        # so far, and with max_length, needs_input must stay false until it
+        # has been returned.
+        data = bytes(100_000)
+        compressed = bz2.compress(data)
+        for max_length in (-1, 1000):
+            with self.subTest(max_length=max_length):
+                bzd = BZ2Decompressor()
+                out = []
+                for i in range(len(compressed)):
+                    out.append(bzd.decompress(compressed[i:i+1], max_length))
+                    while not bzd.needs_input and not bzd.eof:
+                        out.append(bzd.decompress(b"", max_length))
+                    if not bzd.eof:
+                        self.assertEqual(bzd.decompress(b""), b"")
+                self.assertTrue(bzd.eof)
+                self.assertEqual(b"".join(out), data)
+
     def test_decompressor_inputbuf_1(self):
         # Test reusing input buffer after moving existing
         # contents to beginning

Lib/test/test_gzip.py

@@ -870,6 +870,25 @@ def test_read_truncated(self):
             with gzip.GzipFile(fileobj=io.BytesIO(truncated[:i])) as f:
                 self.assertRaises(EOFError, f.read, 1)
 
+    def test_read_short_reads(self):
+        # gh-159999: The decompressor can consume all input and fill the
+        # output buffer at the same time. This must not be reported as
+        # EOF before more input has been read.
+        class ShortReadIO(io.BytesIO):
+            # Return one byte per call after the 10-byte header.
+            def read(self, size=-1):
+                return super().read(size if self.tell() < 10 else 1)
+
+        data = data1*50
+        compressed = gzip.compress(data)
+        for size in (1, 2, 3, 100):
+            with self.subTest(size=size):
+                reader = gzip._GzipReader(ShortReadIO(compressed))
+                out = []
+                while chunk := reader.read(size):
+                    out.append(chunk)
+                self.assertEqual(b"".join(out), data)
+
     def test_read_with_extra(self):
         # Gzip data with an extra field
         gzdata = (b'\x1f\x8b\x08\x04\xb2\x17cQ\x02\xff'

Lib/test/test_zlib.py

@@ -1231,6 +1231,30 @@ def testDecompressorChunksMaxsize(self):
         self.assertEqual(out, self.BIG_TEXT)
         self.assertEqual(zlibd.unused_data, b"")
 
+    def test_decompressor_needs_input_with_buffered_output(self):
+        # gh-159999: needs_input must be false if all input was consumed
+        # but max_length prevented returning all output.
+        zlibd = zlib._ZlibDecompressor()
+        self.assertEqual(zlibd.decompress(zlib.compress(b"\x00")[:4],
+                                          max_length=0), b"")
+        self.assertFalse(zlibd.needs_input)
+        self.assertEqual(zlibd.decompress(b""), b"\x00")
+
+        data = bytes(100_000)
+        compressed = zlib.compress(data)
+        for max_length in (-1, 1000):
+            with self.subTest(max_length=max_length):
+                zlibd = zlib._ZlibDecompressor()
+                out = []
+                for i in range(len(compressed)):
+                    out.append(zlibd.decompress(compressed[i:i+1], max_length))
+                    while not zlibd.needs_input and not zlibd.eof:
+                        out.append(zlibd.decompress(b"", max_length))
+                    if not zlibd.eof:
+                        self.assertEqual(zlibd.decompress(b""), b"")
+                self.assertTrue(zlibd.eof)
+                self.assertEqual(b"".join(out), data)
+
     def test_decompressor_inputbuf_1(self):
         # Test reusing input buffer after moving existing
         # contents to beginning

Misc/NEWS.d/next/Library/2026-10-26-06-55-31.gh-issue-159999.cPp3S0.rst

@@ -0,0 +1,4 @@
+Fix :meth:`bz2.BZ2Decompressor.decompress` sometimes returning only part of
+the available output, and setting :attr:`~bz2.BZ2Decompressor.needs_input` to
+true while output was still buffered. Fix the same :attr:`!needs_input` bug in
+the decompressor used by :mod:`gzip`.

Modules/_bz2module.c

@@ -444,15 +444,19 @@ decompress_buf(BZ2Decompressor *d, Py_ssize_t max_length)
         if (bzret == BZ_STREAM_END) {
             FT_ATOMIC_STORE_CHAR_RELAXED(d->eof, 1);
             break;
-        } else if (d->bzs_avail_in_real == 0) {
-            break;
         } else if (bzs->avail_out == 0) {
+            /* Need to check bzs->avail_out before d->bzs_avail_in_real.
+               bzs's internal state may still have output pending after
+               all input was consumed, so grow the output buffer and
+               continue unless max_length has been reached. */
             if (OutputBuffer_GetDataSize(&buffer, bzs->avail_out) == max_length) {
                 break;
             }
             if (OutputBuffer_Grow(&buffer, &bzs->next_out, &bzs->avail_out) < 0) {
                 goto error;
             }
+        } else if (d->bzs_avail_in_real == 0) {
+            break;
         }
     }
 
@@ -538,7 +542,19 @@ decompress(BZ2Decompressor *d, char *data, size_t len, Py_ssize_t max_length)
     }
     else if (d->bzs_avail_in_real == 0) {
         bzs->next_in = NULL;
-        FT_ATOMIC_STORE_CHAR_RELAXED(d->needs_input, 1);
+
+        if (bzs->avail_out == 0) {
+            /* (avail_in==0 && avail_out==0)
+               bzs's internal state may still have output pending,
+               try to output it next time. */
+            FT_ATOMIC_STORE_CHAR_RELAXED(d->needs_input, 0);
+
+            /* If max_length < 0, bzs->avail_out always > 0 */
+            assert(max_length >= 0);
+        } else {
+            /* Input buffer exhausted, output buffer has space. */
+            FT_ATOMIC_STORE_CHAR_RELAXED(d->needs_input, 1);
+        }
     }
     else {
         FT_ATOMIC_STORE_CHAR_RELAXED(d->needs_input, 0);

Modules/zlibmodule.c

@@ -1645,7 +1645,19 @@ decompress(ZlibDecompressor *self, uint8_t *data,
     }
     else if (self->avail_in_real == 0) {
         self->zst.next_in = NULL;
-        FT_ATOMIC_STORE_CHAR_RELAXED(self->needs_input, 1);
+
+        if (self->zst.avail_out == 0) {
+            /* (avail_in==0 && avail_out==0)
+               zst's internal state may still have output pending,
+               try to output it next time. */
+            FT_ATOMIC_STORE_CHAR_RELAXED(self->needs_input, 0);
+
+            /* If max_length < 0, zst.avail_out always > 0 */
+            assert(max_length >= 0);
+        } else {
+            /* Input buffer exhausted, output buffer has space. */
+            FT_ATOMIC_STORE_CHAR_RELAXED(self->needs_input, 1);
+        }
     }
     else {
         FT_ATOMIC_STORE_CHAR_RELAXED(self->needs_input, 0);