[PR #8611/1fcef940 backport][3.10] Fix handler waiting on shutdown (#8627)

Co-authored-by: Sam Bull <[email protected]>

Author: patchback[bot]

CHANGES/8611.bugfix.rst

@@ -0,0 +1 @@
+Fixed an edge case where shutdown would wait for timeout when handler was already completed -- by :user:`Dreamsorcerer`.

aiohttp/web_protocol.py

@@ -148,6 +148,7 @@ class RequestHandler(BaseProtocol):
         "_lingering_time",
         "_messages",
         "_message_tail",
+        "_handler_waiter",
         "_waiter",
         "_task_handler",
         "_upgrade",
@@ -204,6 +205,7 @@ def __init__(
         self._message_tail = b""
 
         self._waiter: Optional[asyncio.Future[None]] = None
+        self._handler_waiter: Optional[asyncio.Future[None]] = None
         self._task_handler: Optional[asyncio.Task[None]] = None
 
         self._upgrade = False
@@ -262,11 +264,11 @@ async def shutdown(self, timeout: Optional[float] = 15.0) -> None:
         if self._waiter:
             self._waiter.cancel()
 
-        # Wait for graceful disconnection
-        if self._current_request is not None:
+        # Wait for graceful handler completion
+        if self._handler_waiter is not None:
             with suppress(asyncio.CancelledError, asyncio.TimeoutError):
                 async with ceil_timeout(timeout):
-                    await self._current_request.wait_for_disconnection()
+                    await self._handler_waiter
         # Then cancel handler and wait
         with suppress(asyncio.CancelledError, asyncio.TimeoutError):
             async with ceil_timeout(timeout):
@@ -450,6 +452,7 @@ async def _handle_request(
         start_time: float,
         request_handler: Callable[[BaseRequest], Awaitable[StreamResponse]],
     ) -> Tuple[StreamResponse, bool]:
+        self._handler_waiter = self._loop.create_future()
         try:
             try:
                 self._current_request = request
@@ -479,6 +482,8 @@ async def _handle_request(
                 )
 
             reset = await self.finish_response(request, resp, start_time)
+        finally:
+            self._handler_waiter.set_result(None)
 
         return resp, reset