From 37e3aa4639193c6423562a4a4dfbf3310772b7a6 Mon Sep 17 00:00:00 2001
From: Sam Bull <git@sambull.org>
Date: Fri, 13 Sep 2024 21:52:40 +0100
Subject: [PATCH] Fix keepalive race condition (#9140)

---
 CHANGES/9140.bugfix.rst      |  1 +
 aiohttp/web_protocol.py      |  2 +-
 tests/test_web_functional.py | 47 ++++++++++++++++++++++++++++++++++++
 3 files changed, 49 insertions(+), 1 deletion(-)
 create mode 100644 CHANGES/9140.bugfix.rst

diff --git a/CHANGES/9140.bugfix.rst b/CHANGES/9140.bugfix.rst
new file mode 100644
index 00000000000..c9b8f7bf4ea
--- /dev/null
+++ b/CHANGES/9140.bugfix.rst
@@ -0,0 +1 @@
+Fixed race condition that could cause server to close connection incorrectly at keepalive timeout -- by :user:`Dreamosorcerer`.
diff --git a/aiohttp/web_protocol.py b/aiohttp/web_protocol.py
index 4ea0706e3ac..038a7d7047d 100644
--- a/aiohttp/web_protocol.py
+++ b/aiohttp/web_protocol.py
@@ -460,7 +460,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(
diff --git a/tests/test_web_functional.py b/tests/test_web_functional.py
index ec9279ccbf1..aad386f0757 100644
--- a/tests/test_web_functional.py
+++ b/tests/test_web_functional.py
@@ -25,6 +25,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
@@ -2180,3 +2181,49 @@ async def handler(_):
     assert TRANSFER_ENCODING not in resp.headers
     await resp.read() == b""
     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