diff --git a/CHANGES/13504.bugfix.rst b/CHANGES/13504.bugfix.rst
new file mode 100644
index 00000000000..e3a4a73d533
--- /dev/null
+++ b/CHANGES/13504.bugfix.rst
@@ -0,0 +1 @@
+Fixed two edge cases where flow control could get stuck -- by :user:`Dreamsorcerer`.
diff --git a/aiohttp/_http_parser.pyx b/aiohttp/_http_parser.pyx
index c78d9a30be4..25218645263 100644
--- a/aiohttp/_http_parser.pyx
+++ b/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
 
 
diff --git a/aiohttp/_websocket/writer.py b/aiohttp/_websocket/writer.py
index 6bcc2867f4e..591c3d224c5 100644
--- a/aiohttp/_websocket/writer.py
+++ b/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,
diff --git a/aiohttp/web_protocol.py b/aiohttp/web_protocol.py
index f17db73b3f5..0b591ad128d 100644
--- a/aiohttp/web_protocol.py
+++ b/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:
diff --git a/requirements/constraints.txt b/requirements/constraints.txt
index a8d17be70ab..346df184588 100644
--- a/requirements/constraints.txt
+++ b/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
diff --git a/requirements/dev.txt b/requirements/dev.txt
index 7b16b93d2bc..ba443755289 100644
--- a/requirements/dev.txt
+++ b/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
diff --git a/requirements/lint.txt b/requirements/lint.txt
index 041f3c5094c..9aedc9f8871 100644
--- a/requirements/lint.txt
+++ b/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
diff --git a/requirements/test-common-base.txt b/requirements/test-common-base.txt
index 709eaa1b642..2fc5b13681e 100644
--- a/requirements/test-common-base.txt
+++ b/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
diff --git a/requirements/test-common.txt b/requirements/test-common.txt
index 4070c75cc96..7471fe9aee6 100644
--- a/requirements/test-common.txt
+++ b/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
diff --git a/requirements/test-ft.txt b/requirements/test-ft.txt
index 304735580de..1df6989293e 100644
--- a/requirements/test-ft.txt
+++ b/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
diff --git a/requirements/test-mobile.txt b/requirements/test-mobile.txt
index 1cb597f14b4..3364b7fd15d 100644
--- a/requirements/test-mobile.txt
+++ b/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
diff --git a/requirements/test.txt b/requirements/test.txt
index 934575f3dd6..cce5aaf44ff 100644
--- a/requirements/test.txt
+++ b/requirements/test.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
diff --git a/tests/test_web_functional.py b/tests/test_web_functional.py
index f129cbe90df..eaba0f79caa 100644
--- a/tests/test_web_functional.py
+++ b/tests/test_web_functional.py
@@ -2120,6 +2120,118 @@ async def upgrade_handler(request: web.Request) -> web.Response:
     assert b" 400 " in response, response
 
 
+async def test_pipelined_requests_after_deferred_upgrade_are_served(
+    aiohttp_server: AiohttpServer,
+) -> None:
+    pipelined_requests = MAX_MSG_QUEUE_SIZE + 8
+    body = b"b" * 8192
+    handled: list[str] = []
+
+    async def upgrade_handler(request: web.Request) -> web.Response:
+        # Deliberately never reads request.content, so the upgrade stays pending.
+        handled.append(request.path)
+        return web.Response(text="declined")
+
+    async def plain_handler(request: web.Request) -> web.Response:
+        handled.append(request.path)
+        return web.Response(text=f"ok:{request.path}")
+
+    app = web.Application()
+    app.router.add_post("/up", upgrade_handler)
+    app.router.add_get("/{tail:.*}", plain_handler)
+    # Small enough that the unread body pauses the parser mid-message.
+    server = await aiohttp_server(app, read_bufsize=1024, lingering_time=10.0)
+
+    reader, writer = await asyncio.open_connection(server.host, server.port)
+    try:
+        writer.write(
+            b"POST /up HTTP/1.1\r\nHost: localhost\r\n"
+            b"Connection: Upgrade\r\nUpgrade: websocket\r\n"
+            b"Content-Length: "
+            + str(len(body)).encode()
+            + b"\r\n\r\n"
+            + body
+            + b"".join(
+                f"GET /r{i} HTTP/1.1\r\nHost: localhost\r\n\r\n".encode()
+                for i in range(pipelined_requests)
+            )
+        )
+        await writer.drain()
+
+        # Only ever dispatched if the drained body's tail was replayed.
+        first = await asyncio.wait_for(reader.readuntil(b"declined"), 5)
+        last = f"ok:/r{pipelined_requests - 1}".encode()
+        responses = first + await asyncio.wait_for(reader.readuntil(last), 10)
+    finally:
+        writer.close()
+        with suppress(ConnectionResetError, BrokenPipeError):
+            await writer.wait_closed()
+
+    expected = ["/up"] + [f"/r{i}" for i in range(pipelined_requests)]
+    assert handled == expected
+    # One response per request, so a duplicate dispatch cannot hide behind a
+    # readuntil() that already found what it wanted.
+    assert responses.count(b"HTTP/1.1 ") == len(expected), responses[:200]
+
+
+async def test_websocket_prepared_with_unread_body_does_not_stall(
+    aiohttp_server: AiohttpServer,
+) -> None:
+    """Upgrading with the request body unread must leave the socket readable.
+
+    The body's stream pauses reading when nobody drains it, and with the parser
+    stopped mid-message that hold is never lifted by the stream itself. Handing
+    the connection to the websocket has to release it, or the transport stays
+    paused and the websocket never receives a frame -- with keep-alive disabled
+    for the upgrade, nothing would ever reap the connection either.
+    """
+    body = b"b" * 8192
+    echoed: list[str] = []
+
+    async def ws_handler(request: web.Request) -> web.WebSocketResponse:
+        # Deliberately never reads request.content.
+        ws = web.WebSocketResponse()
+        await ws.prepare(request)
+        async for msg in ws:  # pragma: no branch
+            assert isinstance(msg.data, str)
+            echoed.append(msg.data)
+            await ws.send_str(f"echo:{msg.data}")
+            break
+        return ws
+
+    app = web.Application()
+    app.router.add_post("/ws", ws_handler)
+    server = await aiohttp_server(app, read_bufsize=1024)
+
+    reader, writer = await asyncio.open_connection(server.host, server.port)
+    try:
+        writer.write(
+            b"POST /ws HTTP/1.1\r\nHost: localhost\r\n"
+            b"Connection: Upgrade\r\nUpgrade: websocket\r\n"
+            b"Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n"
+            b"Sec-WebSocket-Version: 13\r\n"
+            b"Content-Length: " + str(len(body)).encode() + b"\r\n\r\n" + body
+        )
+        await writer.drain()
+        await asyncio.wait_for(reader.readuntil(b"\r\n\r\n"), 5)
+
+        # Sent after the handshake, so it is only read if the hold was released.
+        ws_writer = WebSocketWriter(
+            mock.Mock(_paused=False),
+            writer.transport,
+            use_mask=True,
+        )
+        await ws_writer.send_frame(b"hi", WSMsgType.TEXT)
+        await writer.drain()
+        await asyncio.wait_for(reader.readuntil(b"echo:hi"), 5)
+    finally:
+        writer.close()
+        with suppress(ConnectionResetError, BrokenPipeError):
+            await writer.wait_closed()
+
+    assert echoed == ["hi"]
+
+
 async def test_declined_websocket_upgrade_reads_body(
     aiohttp_server: AiohttpServer,
 ) -> None: