Don't re-arm read timeout on a connection returned to the pool

CHANGES/12953.bugfix.rst

@@ -0,0 +1,6 @@
+Fixed the ``sock_read`` timeout being re-armed on a keep-alive connection after
+it had been returned to the pool. An idle pooled connection could be left with a
+pending read timeout that fired and poisoned it, so the next request reusing the
+connection failed immediately with :exc:`aiohttp.SocketTimeoutError`. The read
+timeout is now only rescheduled when resuming a transport that was actually
+paused -- by :user:`daragok`.

CHANGES/12954.bugfix.rst

@@ -0,0 +1 @@
+12953.bugfix.rst

CONTRIBUTORS.txt

@@ -343,6 +343,7 @@ Sebastian Hanula
 Sebastian Hüther
 Sebastien Geffroy
 SeongSoo Cho
+Sergei Grachev
 Sergey Ninua
 Sergey Skripnick
 Serhii Charykov

aiohttp/client_proto.py

@@ -196,8 +196,10 @@ def pause_reading(self) -> None:
         self._drop_timeout()
 
     def resume_reading(self, resume_parser: bool = True) -> None:
+        was_paused = self._reading_paused
         super().resume_reading(resume_parser)
-        self._reschedule_timeout()
+        if was_paused:
+            self._reschedule_timeout()
 
     def set_exception(
         self,

tests/test_client_functional.py

@@ -1255,6 +1255,43 @@ async def handler(request: web.Request) -> web.Response:
     assert result == b"foo"
 
 
+async def test_sock_read_timeout_not_rearmed_on_pooled_connection(
+    aiohttp_client: AiohttpClient,
+) -> None:
+    # Reading the buffered body of a completed response must not re-arm the
+    # sock_read timeout on a connection that has already been released to the
+    # keep-alive pool. Otherwise the timer fires while the connection sits idle
+    # in the pool, stamps SocketTimeoutError on it, and the next request that
+    # reuses it fails immediately (with no real read having stalled).
+    async def handler(request: web.Request) -> web.Response:
+        return web.json_response({"ok": True})
+
+    app = web.Application()
+    app.router.add_get("/", handler)
+
+    timeout = aiohttp.ClientTimeout(total=30, sock_read=0.1)
+    client = await aiohttp_client(app, timeout=timeout)
+
+    async with client.get("/") as resp:
+        assert resp.status == 200
+        await resp.read()
+
+    assert client.session.connector is not None
+    pooled = next(iter(client.session.connector._conns.values()))
+    proto = pooled[0][0]
+    # The pooled connection must carry no read-timeout handle, otherwise
+    # it could trigger an exception on the next request.
+    assert proto._read_timeout_handle is None
+    assert proto.exception() is None
+
+    # The connection is still reusable.
+    async with client.get("/") as resp:
+        assert resp.status == 200
+        assert await resp.json() == {"ok": True}
+
+    assert next(iter(client.session.connector._conns.values()))[0][0] is proto
+
+
 async def test_request_exception_cleanup_with_no_total_timeout(
     aiohttp_client: AiohttpClient,
 ) -> None: