PLPeeters · Nov 6, 2026 · Nov 10, 2026
@@ -0,0 +1,2 @@
+Offloaded client response decompression to the event loop's default executor to
+keep decompression work from blocking the event loop -- by :user:`PLPeeters`.
@@ -253,6 +253,7 @@ into `StreamReader`) is then handed to `web_protocol.RequestHandler` and
 | 1.13 | Vendored llhttp version drift | S / T | An upstream llhttp CVE not picked up by aiohttp's vendoring cadence remains exploitable until `make generate-llhttp` is re-run and released. | Medium |
 | 1.14 | Build/regen of llhttp (`make generate-llhttp`) | S / T | Local tampering or supply-chain compromise of the npm `llhttp` package gets baked into the vendored C. Covered in [§5.19](#519-build--release-supply-chain) but originates here. | Medium |
 | 1.15 | `Content-Encoding` coding chain | D | RFC 9110 §8.4 allows several nested codings; each layer multiplies the decompression amplification available from a small wire body (~1000× per gzip layer), so unbounded chains make client-side zip bombs cheap. | Low–Med |
+| 1.15 | Client response decompression scheduling | D | Enabled client response decompression uses shared executor workers; server-side request decompression still runs on the event loop. | Medium |
 
 **Mitigations.**
 
@@ -273,9 +274,14 @@ into `StreamReader`) is then handed to `web_protocol.RequestHandler` and
 | 1.13 | llhttp version drift | Manual upgrade via `make generate-llhttp`; vendor pinned in `vendor/llhttp/package.json`. | Track upstream releases (e.g. via Dependabot rule for `vendor/llhttp/package.json`), bump on every llhttp release, regenerate in CI. |
 | 1.14 | npm-side compromise of `llhttp` | The vendored output is checked into git, so a compromise during a future regen would be detectable in PR review. See [§5.19](#519-build--release-supply-chain). | **Make the llhttp build reproducible: pin Node.js version, commit the npm lockfile, and on every bump verify the regenerated C against upstream's release tarballs before committing.** |
 | 1.15 | Nested coding amplification | `http_parser.py:parse_content_encoding` decodes at most `MAX_CONTENT_CODINGS = 2` codings and raises `ContentEncodingError` beyond that instead of decoding or silently passing compressed bytes through. Unknown codings anywhere in the chain disable decoding. `DeflateBuffer` drains the stage nearest the output first, keeping each stage's intermediate buffer to a small multiple of `max_length` rather than the full decompressed stream. At EOF every stage must end exactly on a stream/frame/member boundary (`stream_complete` on each decompressor; gzip/zstd keep the spent decompressor observable by deferring its replacement until the next member's data arrives): a gzip stream cut before its 8-byte trailer inflates in full and previously passed through silently. | None. |
+| 1.15 | Client response decompression scheduling | When `auto_decompress=True`, each decode, including empty-input drains, runs in the default thread pool. Each response has at most one operation in flight; parsers pause while it runs, preserve existing output limits and stream backpressure, and finish decoding before payload EOF permits connection reuse. Closing discards late output; it cannot interrupt a running worker. | Offloading protects the event loop from codec work but does not cap aggregate CPU usage or the shared executor queue across responses. |
 
 **Past advisories / hardening (recap).**
 
+- **Response decompression offloading** — Python and Cython parser
+  continuation tests cover output backpressure, chunk boundaries, queued input,
+  EOF, decoding/framing errors, and response cleanup while a worker is pending.
+
 - **GHSA-xx9p-xxvh-7g8j (CVE-2023-47641)** (3.8.0) — CL-vs-TE divergence
   between the Cython and pure-Python parsers, allowing request smuggling
   against deployments that switched backends.

@@ -366,6 +366,7 @@ cdef class HttpParser:
         object  _payload_exception
         object  _last_error
         bint    _auto_decompress
+        bint    _offload_decompression
         int     _limit
 
         str     _content_encoding
@@ -397,6 +398,7 @@ cdef class HttpParser:
         bint response_with_body=True, bint read_until_eof=False,
         bint auto_decompress=True,
         Py_ssize_t max_msg_queue_size=0,
+        bint offload_decompression=False,
     ):
         cparser.llhttp_settings_init(self._csettings)
         cparser.llhttp_init(self._cparser, mode, self._csettings)
@@ -433,6 +435,7 @@ cdef class HttpParser:
         self._upgraded = False
         self._pending_upgrade = False
         self._auto_decompress = auto_decompress
+        self._offload_decompression = offload_decompression
         self._content_encoding = None
         self._content_encoding_repeated = False
         self._lax = False
@@ -579,7 +582,9 @@ cdef class HttpParser:
         self._payload = payload
         self._content_length_expected = self._cparser.content_length
         if encoding is not None and self._auto_decompress:
-            self._payload = DeflateBuffer(payload, encoding, max_decompress_size=self._limit)
+            self._payload = DeflateBuffer(
+                payload, encoding, max_decompress_size=self._limit,
+                in_executor=self._offload_decompression)
 
         self._messages.append((msg, payload))
         if self._max_msg_queue_size:
@@ -638,6 +643,8 @@ cdef class HttpParser:
     def feed_eof(self):
         cdef bytes desc
 
+        if self._payload is not None and self._tail:
+            return None
         if self._payload is not None:
             if self._cparser.flags & cparser.F_CHUNKED:
                 raise TransferEncodingError(
@@ -858,12 +865,12 @@ cdef class HttpResponseParser(HttpParser):
             size_t max_line_size=8190, size_t max_headers=128,
             size_t max_field_size=8190, payload_exception=None,
             bint response_with_body=True, bint read_until_eof=False,
-            bint auto_decompress=True
+            bint auto_decompress=True, bint offload_decompression=False
     ):
         self._init(cparser.HTTP_RESPONSE, protocol, loop, limit, timer,
                    max_line_size, max_headers, max_field_size,
                    payload_exception, response_with_body, read_until_eof,
-                   auto_decompress)
+                   auto_decompress, 0, offload_decompression)
         # Use strict parsing on dev mode, so users are warned about broken servers.
         if not DEBUG:
             cparser.llhttp_set_lenient_headers(self._cparser, 1)

@@ -47,6 +47,11 @@ def connected(self) -> bool:
         """Return True if the connection is open."""
         return self.transport is not None
 
+    @property
+    def can_receive_data(self) -> bool:
+        """Return True if a stream reader can still await more data."""
+        return self.connected
+
     @property
     def writing_paused(self) -> bool:
         return self._paused

@@ -56,6 +56,8 @@ def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
 
         self._closed: None | asyncio.Future[None] = None
         self._connection_lost_called = False
+        self._deferred_eof = False
+        self._offload_decompression = False
 
     @property
     def closed(self) -> None | asyncio.Future[None]:
@@ -96,6 +98,10 @@ def force_close(self) -> None:
 
     def close(self) -> None:
         self._exception = None  # Break cyclic references
+        if self._deferred_eof:
+            self._fail_deferred_payload(
+                ClientPayloadError("Response payload was closed")
+            )
         transport = self.transport
         if transport is not None:
             transport.close()
@@ -105,6 +111,10 @@ def close(self) -> None:
 
     def abort(self) -> None:
         self._exception = None  # Break cyclic references
+        if self._deferred_eof:
+            self._fail_deferred_payload(
+                ClientPayloadError("Response payload was aborted")
+            )
         transport = self.transport
         if transport is not None:
             transport.abort()
@@ -115,6 +125,11 @@ def abort(self) -> None:
     def is_connected(self) -> bool:
         return self.transport is not None and not self.transport.is_closing()
 
+    @property
+    def can_receive_data(self) -> bool:
+        # StreamReader can still await decoded data after the transport reaches EOF.
+        return self.transport is not None or self._deferred_eof
+
     def connection_lost(self, exc: BaseException | None) -> None:
         self._connection_lost_called = True
         self._drop_timeout()
@@ -147,6 +162,12 @@ def connection_lost(self, exc: BaseException | None) -> None:
                 self._payload_parser.feed_eof()
 
         uncompleted = None
+        if (
+            self._offload_decompression
+            and self._parser is not None
+            and not self._upgraded
+        ):
+            self.data_received(b"")
         if self._parser is not None:
             try:
                 uncompleted = self._parser.feed_eof()
@@ -183,11 +204,19 @@ def connection_lost(self, exc: BaseException | None) -> None:
             self.set_exception(reraised_exc, underlying_non_eof_exc)
 
         self._should_close = True
-        self._parser = None
-        self._payload = None
+        self._deferred_eof = (
+            self._offload_decompression
+            and isinstance(self._payload, StreamReader)
+            and not self._payload.is_eof()
+            and self._payload.exception() is None
+        )
+        if not self._deferred_eof:
+            self._parser = None
+            self._payload = None
         self._payload_parser = None
         self._payload_parser_failed = False
-        self._reading_paused = False
+        if not self._deferred_eof:
+            self._reading_paused = False
         self._buffer_paused = False
 
         super().connection_lost(reraised_exc)
@@ -203,11 +232,37 @@ def pause_reading(self) -> None:
     def resume_reading(self, resume_parser: bool = True) -> None:
         was_paused = self._reading_paused
         super().resume_reading(resume_parser)
-        # sock_read measures the peer, so it stays stopped while the tail
-        # bound is still holding the transport.
-        if was_paused and not self._buffer_paused:
+        if resume_parser and self._deferred_eof:
+            self._finish_deferred_eof()
+        # sock_read measures the peer, so it stays stopped while a local pause
+        # still holds the transport.
+        if (
+            was_paused
+            and not self._reading_paused
+            and not self._buffer_paused
+            and not self._connection_lost_called
+        ):
             self._reschedule_timeout()
 
+    def _finish_deferred_eof(self) -> None:
+        assert self._parser is not None
+        assert isinstance(self._payload, StreamReader)
+        payload = self._payload
+        if not payload.is_eof() and payload.exception() is None:
+            try:
+                self._parser.feed_eof()
+            except Exception as exc:
+                set_exception(
+                    payload,
+                    ClientPayloadError(f"Response payload is not completed: {exc !r}"),
+                    exc,
+                )
+        if payload.is_eof() or payload.exception() is not None:
+            self._deferred_eof = False
+            self._parser = None
+            self._payload = None
+            self._reading_paused = False
+
     def _drain_tail(self) -> None:
         if self._tail:
             data, self._tail = self._tail, b""
@@ -259,6 +314,7 @@ def set_response_params(
         max_headers: int = 128,
     ) -> None:
         self._skip_payload = skip_payload
+        self._offload_decompression = auto_decompress
 
         self._read_bufsize = read_bufsize
         self._read_timeout = read_timeout
@@ -274,6 +330,7 @@ def set_response_params(
             response_with_body=not skip_payload,
             read_until_eof=read_until_eof,
             auto_decompress=auto_decompress,
+            offload_decompression=auto_decompress,
             max_line_size=max_line_size,
             max_field_size=max_field_size,
             max_headers=max_headers,
@@ -365,6 +422,14 @@ def data_received(self, data: bytes) -> None:
                 # data_received(), the transport is already
                 # closed in this case
                 self.transport.close()
+            if (
+                isinstance(underlying_exc, asyncio.CancelledError)
+                and self._deferred_eof
+            ):
+                self._fail_deferred_payload(underlying_exc)
+                return
+            if self._deferred_eof:
+                self._fail_deferred_payload(underlying_exc)
             if not isinstance(underlying_exc, Exception):
                 raise
             # should_close is True after the call
@@ -405,3 +470,18 @@ def data_received(self, data: bytes) -> None:
 
         if upgraded and tail:
             self.data_received(tail)
+
+    def _fail_deferred_payload(self, exc: BaseException) -> None:
+        payload = self._payload
+        if isinstance(payload, StreamReader) and payload.exception() is None:
+            message = (
+                str(exc)
+                if isinstance(exc, ClientPayloadError)
+                else f"Response payload is not completed: {exc!r}"
+            )
+            set_exception(payload, ClientPayloadError(message), exc)
+
+        self._deferred_eof = False
+        self._parser = None
+        self._payload = None
+        self._reading_paused = False
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,2 @@
		Offloaded client response decompression to the event loop's default executor to
		keep decompression work from blocking the event loop -- by :user:`PLPeeters`.