[pull] master from aio-libs:master

CHANGES/13504.bugfix.rst

@@ -0,0 +1 @@
+Fixed two edge cases where flow control could get stuck -- by :user:`Dreamsorcerer`.

aiohttp/_http_parser.pyx

@@ -552,6 +552,9 @@ cdef class HttpParser:
             self._payload = DeflateBuffer(payload, encoding, max_decompress_size=self._limit)
 
         self._messages.append((msg, payload))
+        if self._max_msg_queue_size:
+            # Count the message where it is handed over, not where its body completes.
+            self._msg_in_flight += 1
 
     cdef _on_message_complete(self):
         # The payload is None when feed_eof() already completed a fully
@@ -940,12 +943,12 @@ cdef int cb_on_message_complete(cparser.llhttp_t* parser) except? -1:
         pyparser._last_error = exc
         return -1
     else:
-        if pyparser._max_msg_queue_size:
-            pyparser._msg_in_flight += 1
-            if pyparser._msg_in_flight >= pyparser._max_msg_queue_size:
-                # Queue full: pause llhttp between messages. feed_data() buffers
-                # the remainder as tail; resumes once the queue drains.
-                return cparser.HPE_PAUSED
+        if (
+            pyparser._max_msg_queue_size
+            and pyparser._msg_in_flight >= pyparser._max_msg_queue_size
+        ):
+            # Queue full: pause llhttp between messages.
+            return cparser.HPE_PAUSED
         return 0
 
 

aiohttp/_websocket/writer.py

@@ -49,7 +49,7 @@ class WebSocketWriter:
     def __init__(
         self,
         protocol: BaseProtocol,
-        transport: asyncio.Transport,
+        transport: asyncio.WriteTransport,
         *,
         use_mask: bool = False,
         limit: int = DEFAULT_CHUNK_SIZE,

aiohttp/web_protocol.py

@@ -458,6 +458,12 @@ def set_parser(
         self._payload_parser = parser
         self._data_received_cb = data_received_cb
 
+        if self._reading_paused:
+            # After upgrade nothing will read the stream again, so we need
+            # to resume here before feeding the tail, so a pause in the
+            # upgraded protocol still takes effect.
+            self.resume_reading(resume_parser=False)
+
         if self._message_tail:
             self._payload_parser.feed_data(self._message_tail)
             self._message_tail = b""
@@ -556,6 +562,67 @@ def _resume_msg_queue_reading(self) -> None:
                 # ignored (see PAUSE_RESUME_READING_ERRORS; do not use suppress).
                 pass
 
+    def _replay_message_tail(self) -> None:
+        """Re-feed the bytes buffered behind a rejected upgrade.
+
+        The parser stops at an upgrade boundary and holds everything after it in
+        ``_message_tail``. If the upgrade is rejected those bytes are
+        pipelined requests and have to go back through the parser.
+        """
+        if (
+            not self._upgraded
+            # The upgrade request is the last request before the parser paused,
+            # so wait for messages to be empty.
+            or self._messages
+            # payload_parser is not None if the upgrade was accepted.
+            or self._payload_parser is not None
+            or self._parser is None
+        ):
+            return
+
+        self._parser.set_upgraded(False)
+        self._upgraded = False
+        if not self._message_tail:
+            return
+
+        messages: Sequence[_MsgType]
+        try:
+            messages, upgraded, tail = self._parser.feed_data(self._message_tail)
+        except HttpProcessingError as parse_exc:
+            # Garbage (or an oversized request line) buffered behind the
+            # upgrade: answer 400 instead of letting the error escape
+            # and lose this response, like data_received() does.
+            messages = [
+                (
+                    _ErrInfo(
+                        status=400,
+                        exc=parse_exc,
+                        message=parse_exc.message,
+                    ),
+                    EMPTY_PAYLOAD,
+                )
+            ]
+            upgraded = False
+            tail = b""
+
+        # A further upgrade request in the tail buffers its own remainder.
+        self._upgraded = upgraded
+        self._message_tail = tail
+        for msg, payload in messages:
+            self._request_count += 1
+            self._messages.append((msg, payload))
+
+        if len(self._messages) >= self._max_msg_queue_size:
+            # Pause the transport, like in data_received().
+            self._pause_msg_queue_reading()
+        elif self._msg_queue_paused:
+            # Resume reading now the tail has been parsed.
+            self._resume_msg_queue_reading()
+
+        # This shouldn't be possible. If a future refactor results in this
+        # failing, then the code may need to be updated to set the waiter.
+        assert self._waiter is None
+
     def keep_alive(self, val: bool) -> None:
         """Set keep-alive connection mode.
 
@@ -789,6 +856,10 @@ async def start(self) -> None:
 
                 payload.set_exception(_PAYLOAD_ACCESS_ERROR)
 
+                # Draining the body above can have been what finally settled a
+                # deferred upgrade, seating a tail that finish_response() was
+                # too early to see.
+                self._replay_message_tail()
             except asyncio.CancelledError:
                 self.log_debug("Ignored premature client disconnection")
                 self.force_close()
@@ -833,56 +904,7 @@ async def finish_response(
         """
         request._finish()
 
-        # Handle feeding the message tail following an upgrade request that
-        # was declined.
-        # The upgrade request is the last request before the parser paused,
-        # so wait for self._messages to be empty.
-        # payload_parser is not None if the upgrade was accepted.
-        if (
-            self._upgraded
-            and not self._messages
-            and self._payload_parser is None
-            and self._parser is not None
-        ):
-            self._parser.set_upgraded(False)
-            self._upgraded = False
-            if self._message_tail:
-                messages: Sequence[_MsgType]
-                try:
-                    messages, upgraded, tail = self._parser.feed_data(
-                        self._message_tail
-                    )
-                except HttpProcessingError as parse_exc:
-                    # Garbage (or an oversized request line) buffered behind the
-                    # upgrade: answer 400 instead of letting the error escape
-                    # and lose this response, like data_received() does.
-                    messages = [
-                        (
-                            _ErrInfo(
-                                status=400,
-                                exc=parse_exc,
-                                message=parse_exc.message,
-                            ),
-                            EMPTY_PAYLOAD,
-                        )
-                    ]
-                    upgraded = False
-                    tail = b""
-                # A further upgrade request in the tail buffers its own remainder.
-                self._upgraded = upgraded
-                self._message_tail = tail
-                for msg, payload in messages:
-                    self._request_count += 1
-                    self._messages.append((msg, payload))
-                if len(self._messages) >= self._max_msg_queue_size:
-                    # Pause the transport, like in data_received().
-                    self._pause_msg_queue_reading()
-                elif self._msg_queue_paused:
-                    # Resume reading now the tail has been parsed.
-                    self._resume_msg_queue_reading()
-                # This shouldn't be possible. If a future refactor results in this
-                # failing, then the code may need to be updated to set the waiter.
-                assert self._waiter is None
+        self._replay_message_tail()
         try:
             prepare_meth = resp.prepare
         except AttributeError:

requirements/constraints.txt

@@ -65,11 +65,11 @@ click==8.5.0
     # via
     #   pip-tools
     #   towncrier
-coverage==7.15.4
+coverage==7.16.0
     # via
     #   -r requirements/test-common.in
     #   pytest-cov
-cryptography==50.0.0
+cryptography==50.0.1
     # via trustme
 cython==3.3.0
     # via -r requirements/cython.in
@@ -187,9 +187,9 @@ pycares==5.0.1
     # via aiodns
 pycparser==3.0
     # via cffi
-pydantic==2.13.4
+pydantic==2.13.5
     # via python-on-whales
-pydantic-core==2.46.4
+pydantic-core==2.46.5
     # via pydantic
 pyenchant==3.3.0
     # via sphinxcontrib-spelling
@@ -235,7 +235,7 @@ pytest-xdist==3.8.0
     # via -r requirements/test-common.in
 python-dateutil==2.9.0.post0
     # via freezegun
-python-discovery==1.5.3
+python-discovery==1.6.0
     # via virtualenv
 python-on-whales==0.81.0
     # via
@@ -331,7 +331,7 @@ uvloop==0.22.1 ; platform_system != "Windows"
     #   -r requirements/lint.in
 valkey==6.1.1
     # via -r requirements/lint.in
-virtualenv==21.7.5
+virtualenv==21.7.7
     # via pre-commit
 wheel==0.48.0
     # via pip-tools

requirements/dev.txt

@@ -65,11 +65,11 @@ click==8.5.0
     # via
     #   pip-tools
     #   towncrier
-coverage==7.15.4
+coverage==7.16.0
     # via
     #   -r requirements/test-common.in
     #   pytest-cov
-cryptography==50.0.0
+cryptography==50.0.1
     # via trustme
 distlib==0.4.3
     # via virtualenv
@@ -184,9 +184,9 @@ pycares==5.0.1
     # via aiodns
 pycparser==3.0
     # via cffi
-pydantic==2.13.4
+pydantic==2.13.5
     # via python-on-whales
-pydantic-core==2.46.4
+pydantic-core==2.46.5
     # via pydantic
 pygments==2.21.0
     # via
@@ -230,7 +230,7 @@ pytest-xdist==3.8.0
     # via -r requirements/test-common.in
 python-dateutil==2.9.0.post0
     # via freezegun
-python-discovery==1.5.3
+python-discovery==1.6.0
     # via virtualenv
 python-on-whales==0.81.0
     # via
@@ -321,7 +321,7 @@ uvloop==0.22.1 ; platform_system != "Windows" and implementation_name == "cpytho
     #   -r requirements/lint.in
 valkey==6.1.1
     # via -r requirements/lint.in
-virtualenv==21.7.5
+virtualenv==21.7.7
     # via pre-commit
 wheel==0.48.0
     # via pip-tools

requirements/lint.txt

@@ -34,7 +34,7 @@ cffi==2.1.1
     #   pycares
 cfgv==3.5.0
     # via pre-commit
-cryptography==50.0.0
+cryptography==50.0.1
     # via trustme
 distlib==0.4.3
     # via virtualenv
@@ -100,9 +100,9 @@ pycares==5.0.1
     # via aiodns
 pycparser==3.0
     # via cffi
-pydantic==2.13.4
+pydantic==2.13.5
     # via python-on-whales
-pydantic-core==2.46.4
+pydantic-core==2.46.5
     # via pydantic
 pygments==2.21.0
     # via
@@ -125,7 +125,7 @@ pytest-mock==3.15.1
     # via -r requirements/lint.in
 python-dateutil==2.9.0.post0
     # via freezegun
-python-discovery==1.5.3
+python-discovery==1.6.0
     # via virtualenv
 python-on-whales==0.81.0
     # via -r requirements/lint.in
@@ -164,7 +164,7 @@ uvloop==0.22.1 ; platform_system != "Windows"
     # via -r requirements/lint.in
 valkey==6.1.1
     # via -r requirements/lint.in
-virtualenv==21.7.5
+virtualenv==21.7.7
     # via pre-commit
 yarl==1.24.5
     # via aiohttp

requirements/test-common-base.txt

@@ -14,7 +14,7 @@ attrs==26.1.0
     # via aiohttp
 backports-asyncio-runner==1.2.0
     # via pytest-asyncio
-coverage==7.15.4
+coverage==7.16.0
     # via pytest-cov
 exceptiongroup==1.3.1
     # via pytest

requirements/test-common.txt

@@ -22,11 +22,11 @@ blockbuster==1.5.27
     # via -r requirements/test-common.in
 cffi==2.1.1
     # via cryptography
-coverage==7.15.4
+coverage==7.16.0
     # via
     #   -r requirements/test-common.in
     #   pytest-cov
-cryptography==50.0.0
+cryptography==50.0.1
     # via trustme
 exceptiongroup==1.3.1
     # via pytest
@@ -80,9 +80,9 @@ proxy-py==2.4.10
     # via -r requirements/test-common-base.in
 pycparser==3.0
     # via cffi
-pydantic==2.13.4
+pydantic==2.13.5
     # via python-on-whales
-pydantic-core==2.46.4
+pydantic-core==2.46.5
     # via pydantic
 pygments==2.21.0
     # via

requirements/test-ft.txt

@@ -38,11 +38,11 @@ cffi==2.1.1
     # via
     #   cryptography
     #   pycares
-coverage==7.15.4
+coverage==7.16.0
     # via
     #   -r requirements/test-common.in
     #   pytest-cov
-cryptography==50.0.0
+cryptography==50.0.1
     # via trustme
 exceptiongroup==1.3.1
     # via
@@ -105,9 +105,9 @@ pycares==5.0.1
     # via aiodns
 pycparser==3.0
     # via cffi
-pydantic==2.13.4
+pydantic==2.13.5
     # via python-on-whales
-pydantic-core==2.46.4
+pydantic-core==2.46.5
     # via pydantic
 pygments==2.21.0
     # via

requirements/test-mobile.txt

@@ -34,7 +34,7 @@ cffi==2.1.1 ; sys_platform != "android" and sys_platform != "ios"
     # via
     #   -r requirements/test-mobile.in
     #   pycares
-coverage==7.15.4
+coverage==7.16.0
     # via pytest-cov
 exceptiongroup==1.3.1
     # via