Break cyclic references when there is an exception handling a request (#10569)

<!-- Thank you for your contribution! -->

## What do these changes do?

This is a partial fix for #10548

- There is still another case for `SystemRoute`s that needs to be
addressed. No reproducer available yet.
- There is also another case on the client side on connection refused
that still needs to be addressed
#10548 (comment)

## Are there changes in behavior for the user?

fixes memory leak

## Is it a substantial burden for the maintainers to support this?
no

Author: bdraco

CHANGES/10569.bugfix.rst

@@ -0,0 +1 @@
+Break cyclic references when there is an exception handling a request -- by :user:`bdraco`.

aiohttp/web_protocol.py

@@ -543,8 +543,6 @@ async def start(self) -> None:
         keep_alive(True) specified.
         """
         loop = self._loop
-        handler = asyncio.current_task(loop)
-        assert handler is not None
         manager = self._manager
         assert manager is not None
         keepalive_timeout = self._keepalive_timeout
@@ -574,7 +572,16 @@ async def start(self) -> None:
                 request_handler = self._make_error_handler(message)
                 message = ERROR
 
-            request = self._request_factory(message, payload, self, writer, handler)
+            # Important don't hold a reference to the current task
+            # as on traceback it will prevent the task from being
+            # collected and will cause a memory leak.
+            request = self._request_factory(
+                message,
+                payload,
+                self,
+                writer,
+                self._task_handler or asyncio.current_task(loop),  # type: ignore[arg-type]
+            )
             try:
                 # a new task is used for copy context vars (#3406)
                 coro = self._handle_request(request, start, request_handler)
@@ -642,6 +649,7 @@ async def start(self) -> None:
                 self.force_close()
                 raise
             finally:
+                request._task = None  # type: ignore[assignment] # Break reference cycle in case of exception
                 if self.transport is None and resp is not None:
                     self.log_debug("Ignored premature client disconnection.")
 

tests/isolated/check_for_request_leak.py

@@ -0,0 +1,41 @@
+import asyncio
+import gc
+import sys
+from typing import NoReturn
+
+from aiohttp import ClientSession, web
+from aiohttp.test_utils import get_unused_port_socket
+
+gc.set_debug(gc.DEBUG_LEAK)
+
+
+async def main() -> None:
+    app = web.Application()
+
+    async def handler(request: web.Request) -> NoReturn:
+        await request.json()
+        assert False
+
+    app.router.add_route("GET", "/json", handler)
+    sock = get_unused_port_socket("127.0.0.1")
+    port = sock.getsockname()[1]
+
+    runner = web.AppRunner(app)
+    await runner.setup()
+    site = web.SockSite(runner, sock)
+    await site.start()
+
+    async with ClientSession() as session:
+        async with session.get(f"http://127.0.0.1:{port}/json") as resp:
+            await resp.read()
+
+    # Give time for the cancelled task to be collected
+    await asyncio.sleep(0.5)
+    gc.collect()
+    request_present = any(type(obj).__name__ == "Request" for obj in gc.garbage)
+    await session.close()
+    await runner.cleanup()
+    sys.exit(1 if request_present else 0)
+
+
+asyncio.run(main())

tests/test_leaks.py

@@ -23,3 +23,20 @@ def test_client_response_does_not_leak_on_server_disconnected_error() -> None:
         stdout=subprocess.PIPE,
     ) as proc:
         assert proc.wait() == 0, "ClientResponse leaked"
+
+
+@pytest.mark.skipif(IS_PYPY, reason="gc.DEBUG_LEAK not available on PyPy")
+def test_request_does_not_leak_when_request_handler_raises() -> None:
+    """Test that the Request object is collected when the handler raises.
+
+    https://github.com/aio-libs/aiohttp/issues/10548
+    """
+    leak_test_script = pathlib.Path(__file__).parent.joinpath(
+        "isolated", "check_for_request_leak.py"
+    )
+
+    with subprocess.Popen(
+        [sys.executable, "-u", str(leak_test_script)],
+        stdout=subprocess.PIPE,
+    ) as proc:
+        assert proc.wait() == 0, "Request leaked"