[PR #10171/5185f93 backport][3.11] Stream unpauses protocol before re…

…leasing connection (#10179)

Co-authored-by: Javier Torres <javier@javiertorres.eu>

CHANGES/10169.bugfix.rst

@@ -0,0 +1,3 @@
+Fixed a hang where a connection previously used for a streaming
+download could be returned to the pool in a paused state.
+-- by :user:`javitonino`.

CONTRIBUTORS.txt

@@ -171,6 +171,7 @@ Jan Buchar
 Jan Gosmann
 Jarno Elonen
 Jashandeep Sohi
+Javier Torres
 Jean-Baptiste Estival
 Jens Steinhauser
 Jeonghun Lee

aiohttp/streams.py

@@ -220,6 +220,9 @@ def feed_eof(self) -> None:
             self._eof_waiter = None
             set_result(waiter, None)
 
+        if self._protocol._reading_paused:
+            self._protocol.resume_reading()
+
         for cb in self._eof_callbacks:
             try:
                 cb()

tests/test_flowcontrol_streams.py

@@ -4,6 +4,7 @@
 import pytest
 
 from aiohttp import streams
+from aiohttp.base_protocol import BaseProtocol
 
 
 @pytest.fixture
@@ -112,6 +113,15 @@ async def test_read_nowait(self, stream) -> None:
         assert res == b""
         assert stream._protocol.resume_reading.call_count == 1  # type: ignore[attr-defined]
 
+    async def test_resumed_on_eof(self, stream: streams.StreamReader) -> None:
+        stream.feed_data(b"data")
+        assert stream._protocol.pause_reading.call_count == 1  # type: ignore[attr-defined]
+        assert stream._protocol.resume_reading.call_count == 0  # type: ignore[attr-defined]
+        stream._protocol._reading_paused = True
+
+        stream.feed_eof()
+        assert stream._protocol.resume_reading.call_count == 1  # type: ignore[attr-defined]
+
 
 async def test_flow_control_data_queue_waiter_cancelled(
     buffer: streams.FlowControlDataQueue,
@@ -180,3 +190,16 @@ async def test_flow_control_data_queue_read_eof(
     buffer.feed_eof()
     with pytest.raises(streams.EofStream):
         await buffer.read()
+
+
+async def test_stream_reader_eof_when_full() -> None:
+    loop = asyncio.get_event_loop()
+    protocol = BaseProtocol(loop=loop)
+    protocol.transport = asyncio.Transport()
+    stream = streams.StreamReader(protocol, 1024, loop=loop)
+
+    data_len = stream._high_water + 1
+    stream.feed_data(b"0" * data_len)
+    assert protocol._reading_paused
+    stream.feed_eof()
+    assert not protocol._reading_paused