[pull] master from aio-libs:master

CHANGES/13356.bugfix.rst

@@ -0,0 +1,4 @@
+Fixed requests pipelined behind a request whose upgrade the handler declined
+going unanswered once there were more of them than the per-connection queue
+holds. With the pure-Python parser the same requests were also served more
+than once -- by :user:`rodrigobnogueira`.

aiohttp/http_parser.py

@@ -358,6 +358,8 @@ def feed_data(
                     # any preceding body is consumed before the next request
                     # line. Resumes via feed_data(b"") when the queue drains.
                     self._tail = data[start_pos:]
+                    # The remainder now lives in self._tail only. Don't return it.
+                    data = EMPTY
                     break
                 pos = data.find(SEP, start_pos)
                 # consume \r\n

aiohttp/web_protocol.py

@@ -816,6 +816,12 @@ async def finish_response(
                 for msg, payload in messages:
                     self._request_count += 1
                     self._messages.append((msg, payload))
+                # Pause the transport, like in data_received().
+                if (
+                    not self._msg_queue_paused
+                    and len(self._messages) >= self._max_msg_queue_size
+                ):
+                    self._pause_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

tests/test_http_parser.py

@@ -184,6 +184,24 @@ def test_max_msg_queue_size_caps_emitted_messages(
     assert not upgraded
 
 
+def test_max_msg_queue_size_keeps_tail_to_itself(
+    request_cls: type[HttpRequestParser],
+    protocol: BaseProtocol,
+    event_loop: asyncio.AbstractEventLoop,
+) -> None:
+    """The remainder is buffered for the next feed, so it must not be returned.
+
+    Handing it back as well gives the caller a second copy of bytes the parser
+    is already holding, and both copies get parsed.
+    """
+    parser = _build_request_parser(request_cls, protocol, event_loop, 4)
+
+    messages, _upgraded, tail = parser.feed_data(_PIPELINED_GET * 10)
+
+    assert len(messages) == 4
+    assert tail == b""
+
+
 def test_max_msg_queue_size_resumes_after_consume(
     request_cls: type[HttpRequestParser],
     protocol: BaseProtocol,

tests/test_web_functional.py

@@ -1847,6 +1847,57 @@ def raw_get(path: str) -> bytes:
     assert len(handled) == pipelined_requests + 1
 
 
+async def test_http1_pipelined_behind_declined_upgrade_served_once(
+    aiohttp_server: AiohttpServer,
+) -> None:
+    """Requests pipelined behind a declined upgrade are each served once.
+
+    The bytes following an upgrade request are buffered whole, then re-fed once
+    the handler answers it normally. More of them than the queue holds must
+    still be served, and none of them twice.
+    """
+    pipelined_requests = MAX_MSG_QUEUE_SIZE + 8
+    handled: list[str] = []
+    all_handled = asyncio.Event()
+
+    async def handler(request: web.Request) -> web.Response:
+        handled.append(request.path)
+        if len(handled) == pipelined_requests + 1:
+            all_handled.set()
+        return web.Response()
+
+    app = web.Application()
+    app.router.add_get("/{tail:.*}", handler)
+    server = await aiohttp_server(app)
+
+    def raw_get(path: str) -> bytes:
+        return (
+            f"GET {path} HTTP/1.1\r\nHost: localhost\r\n"
+            "Connection: keep-alive\r\n\r\n"
+        ).encode("ascii")
+
+    # An upgrade the handler answers normally, then the pipeline, in one write.
+    upgrade = (
+        b"GET /upgrade HTTP/1.1\r\nHost: localhost\r\n"
+        b"Connection: Upgrade\r\nUpgrade: websocket\r\n\r\n"
+    )
+
+    reader, writer = await asyncio.open_connection(server.host, server.port)
+    try:
+        writer.write(
+            upgrade + b"".join(raw_get(f"/r{i}") for i in range(pipelined_requests))
+        )
+        await writer.drain()
+        await asyncio.wait_for(all_handled.wait(), 10)
+    finally:
+        writer.close()
+        with suppress(ConnectionResetError, BrokenPipeError):
+            await writer.wait_closed()
+
+    assert len(handled) == pipelined_requests + 1
+    assert len(set(handled)) == len(handled)
+
+
 async def test_declined_websocket_upgrade_reads_body(
     aiohttp_server: AiohttpServer,
 ) -> None: