Fix keepalive race condition (#9140) (#9143)

(cherry picked from commit 37e3aa4)

CHANGES/9140.bugfix.rst

@@ -0,0 +1 @@
+Fixed race condition that could cause server to close connection incorrectly at keepalive timeout -- by :user:`Dreamosorcerer`.

aiohttp/web_protocol.py

@@ -437,7 +437,7 @@ def _process_keepalive(self) -> None:
             return
 
         # handler in idle state
-        if self._waiter:
+        if self._waiter and not self._waiter.done():
             self.force_close()
 
     async def _handle_request(

tests/test_web_functional.py

@@ -24,6 +24,7 @@
 from aiohttp.hdrs import CONTENT_LENGTH, CONTENT_TYPE, TRANSFER_ENCODING
 from aiohttp.test_utils import make_mocked_coro
 from aiohttp.typedefs import Handler
+from aiohttp.web_protocol import RequestHandler
 
 try:
     import brotlicffi as brotli
@@ -2246,3 +2247,49 @@ async def handler(_):
     assert TRANSFER_ENCODING not in resp.headers
     await resp.read() == b""
     await resp.release()
+
+
+async def test_keepalive_race_condition(aiohttp_client: Any) -> None:
+    protocol = None
+    orig_data_received = RequestHandler.data_received
+
+    def delay_received(self, data: bytes) -> None:
+        """Emulate race condition.
+
+        The keepalive callback needs to be called between data_received() and
+        when start() resumes from the waiter set within data_received().
+        """
+        data = orig_data_received(self, data)
+        if protocol is None:  # First request creating the keepalive connection.
+            return data
+
+        assert self is protocol
+        assert protocol._keepalive_handle is not None
+        # Cancel existing callback that would run at some point in future.
+        protocol._keepalive_handle.cancel()
+        protocol._keepalive_handle = None
+
+        # Set next run time into the past and run callback manually.
+        protocol._next_keepalive_close_time = asyncio.get_running_loop().time() - 1
+        protocol._process_keepalive()
+
+        return data
+
+    async def handler(request: web.Request) -> web.Response:
+        nonlocal protocol
+        protocol = request.protocol
+        return web.Response()
+
+    target = "aiohttp.web_protocol.RequestHandler.data_received"
+    with mock.patch(target, delay_received):
+        app = web.Application()
+        app.router.add_get("/", handler)
+        client = await aiohttp_client(app)
+
+        # Open connection, so we have a keepalive connection and reference to protocol.
+        async with client.get("/") as resp:
+            assert resp.status == 200
+        assert protocol is not None
+        # Make 2nd request which will hit the race condition.
+        async with client.get("/") as resp:
+            assert resp.status == 200