[PR #10556/9d4e1161 backport][3.11] Break cyclic references at connection close when there was a traceback (#10566)

**This is a backport of PR #10556 as merged into master
(9d4e116).**


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

## What do these changes do?

Clears the exception on the `DataQueue` and `WebSocketDataQueue` when
the connection is closed to break cyclic references.

## Are there changes in behavior for the user?

bugfix

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

no

## Related issue number

fixes #10535

## Checklist

- [ ] I think the code is well written
- [ ] Unit tests for the changes exist
- [ ] Documentation reflects the changes
- [ ] If you provide code modification, please add yourself to
`CONTRIBUTORS.txt`
  * The format is &lt;Name&gt; &lt;Surname&gt;.
  * Please keep alphabetical order, the file is sorted by names.
- [ ] Add a new news fragment into the `CHANGES/` folder
  * name it `<issue_or_pr_num>.<type>.rst` (e.g. `588.bugfix.rst`)
  * if you don't have an issue number, change it to the pull request
    number after creating the PR
    * `.bugfix`: A bug fix for something the maintainers deemed an
      improper undesired behavior that got corrected to match
      pre-agreed expectations.
    * `.feature`: A new behavior, public APIs. That sort of stuff.
    * `.deprecation`: A declaration of future API removals and breaking
      changes in behavior.
    * `.breaking`: When something public is removed in a breaking way.
      Could be deprecated in an earlier release.
    * `.doc`: Notable updates to the documentation structure or build
      process.
    * `.packaging`: Notes for downstreams about unobvious side effects
      and tooling. Changes in the test invocation considerations and
      runtime assumptions.
    * `.contrib`: Stuff that affects the contributor experience. e.g.
      Running tests, building the docs, setting up the development
      environment.
    * `.misc`: Changes that are hard to assign to any of the above
      categories.
  * Make sure to use full sentences with correct case and punctuation,
    for example:
    ```rst
    Fixed issue with non-ascii contents in doctest text files
    -- by :user:`contributor-gh-handle`.
    ```

    Use the past tense or the present tense a non-imperative mood,
    referring to what's changed compared to the last released version
    of this project.

Co-authored-by: J. Nick Koston <[email protected]>

Author: patchback[bot]

CHANGES/10556.bugfix.rst

@@ -0,0 +1,3 @@
+Break cyclic references at connection close when there was a traceback -- by :user:`bdraco`.
+
+Special thanks to :user:`availov` for reporting the issue.

aiohttp/_websocket/reader_py.py

@@ -93,6 +93,7 @@ def _release_waiter(self) -> None:
     def feed_eof(self) -> None:
         self._eof = True
         self._release_waiter()
+        self._exception = None  # Break cyclic references
 
     def feed_data(self, data: "WSMessage", size: "int_") -> None:
         self._size += size

aiohttp/client_proto.py

@@ -64,6 +64,7 @@ def force_close(self) -> None:
         self._should_close = True
 
     def close(self) -> None:
+        self._exception = None  # Break cyclic references
         transport = self.transport
         if transport is not None:
             transport.close()

tests/isolated/check_for_client_response_leak.py

@@ -0,0 +1,47 @@
+import asyncio
+import contextlib
+import gc
+import sys
+
+from aiohttp import ClientError, 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 stream_handler(request: web.Request) -> web.Response:
+        assert request.transport is not None
+        request.transport.close()  # Forcefully closing connection
+        return web.Response()
+
+    app.router.add_get("/stream", stream_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()
+
+    session = ClientSession()
+
+    async def fetch_stream(url: str) -> None:
+        """Fetch a stream and read a few bytes from it."""
+        with contextlib.suppress(ClientError):
+            await session.get(url)
+
+    client_task = asyncio.create_task(fetch_stream(f"http://localhost:{port}/stream"))
+    await client_task
+    gc.collect()
+    client_response_present = any(
+        type(obj).__name__ == "ClientResponse" for obj in gc.garbage
+    )
+    await session.close()
+    await runner.cleanup()
+    sys.exit(1 if client_response_present else 0)
+
+
+asyncio.run(main())

tests/test_leaks.py

@@ -0,0 +1,25 @@
+import pathlib
+import platform
+import subprocess
+import sys
+
+import pytest
+
+IS_PYPY = platform.python_implementation() == "PyPy"
+
+
+@pytest.mark.skipif(IS_PYPY, reason="gc.DEBUG_LEAK not available on PyPy")
+def test_client_response_does_not_leak_on_server_disconnected_error() -> None:
+    """Test that ClientResponse is collected after server disconnects.
+
+    https://github.com/aio-libs/aiohttp/issues/10535
+    """
+    leak_test_script = pathlib.Path(__file__).parent.joinpath(
+        "isolated", "check_for_client_response_leak.py"
+    )
+
+    with subprocess.Popen(
+        [sys.executable, "-u", str(leak_test_script)],
+        stdout=subprocess.PIPE,
+    ) as proc:
+        assert proc.wait() == 0, "ClientResponse leaked"