[PR #11107/21d640d backport][3.12] Avoid creating closed futures that will never be awaited (#11110)

Author: bdraco

CHANGES/11107.misc.rst

@@ -0,0 +1 @@
+Avoided creating closed futures in ``ResponseHandler`` that will never be awaited -- by :user:`bdraco`.

aiohttp/client_proto.py

@@ -1,6 +1,6 @@
 import asyncio
 from contextlib import suppress
-from typing import Any, Optional, Tuple
+from typing import Any, Optional, Tuple, Union
 
 from .base_protocol import BaseProtocol
 from .client_exceptions import (
@@ -45,7 +45,27 @@ def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
         self._read_timeout_handle: Optional[asyncio.TimerHandle] = None
 
         self._timeout_ceil_threshold: Optional[float] = 5
-        self.closed: asyncio.Future[None] = self._loop.create_future()
+
+        self._closed: Union[None, asyncio.Future[None]] = None
+        self._connection_lost_called = False
+
+    @property
+    def closed(self) -> Union[None, asyncio.Future[None]]:
+        """Future that is set when the connection is closed.
+
+        This property returns a Future that will be completed when the connection
+        is closed. The Future is created lazily on first access to avoid creating
+        futures that will never be awaited.
+
+        Returns:
+            - A Future[None] if the connection is still open or was closed after
+              this property was accessed
+            - None if connection_lost() was already called before this property
+              was ever accessed (indicating no one is waiting for the closure)
+        """
+        if self._closed is None and not self._connection_lost_called:
+            self._closed = self._loop.create_future()
+        return self._closed
 
     @property
     def upgraded(self) -> bool:
@@ -79,30 +99,31 @@ def is_connected(self) -> bool:
         return self.transport is not None and not self.transport.is_closing()
 
     def connection_lost(self, exc: Optional[BaseException]) -> None:
+        self._connection_lost_called = True
         self._drop_timeout()
 
         original_connection_error = exc
         reraised_exc = original_connection_error
 
         connection_closed_cleanly = original_connection_error is None
 
-        if connection_closed_cleanly:
-            set_result(self.closed, None)
-        else:
-            assert original_connection_error is not None
-            set_exception(
-                self.closed,
-                ClientConnectionError(
-                    f"Connection lost: {original_connection_error !s}",
-                ),
-                original_connection_error,
-            )
-            # Mark the exception as retrieved to prevent
-            # "Future exception was never retrieved" warnings
-            # The exception is always passed on through
-            # other means, so this is safe
-            with suppress(Exception):
-                self.closed.exception()
+        if self._closed is not None:
+            # If someone is waiting for the closed future,
+            # we should set it to None or an exception. If
+            # self._closed is None, it means that
+            # connection_lost() was called already
+            # or nobody is waiting for it.
+            if connection_closed_cleanly:
+                set_result(self._closed, None)
+            else:
+                assert original_connection_error is not None
+                set_exception(
+                    self._closed,
+                    ClientConnectionError(
+                        f"Connection lost: {original_connection_error !s}",
+                    ),
+                    original_connection_error,
+                )
 
         if self._payload_parser is not None:
             with suppress(Exception):  # FIXME: log this somehow?

aiohttp/connector.py

@@ -507,11 +507,13 @@ def _close(self) -> List[Awaitable[object]]:
             for data in self._conns.values():
                 for proto, _ in data:
                     proto.close()
-                    waiters.append(proto.closed)
+                    if closed := proto.closed:
+                        waiters.append(closed)
 
             for proto in self._acquired:
                 proto.close()
-                waiters.append(proto.closed)
+                if closed := proto.closed:
+                    waiters.append(closed)
 
             for transport in self._cleanup_closed_transports:
                 if transport is not None:

tests/test_client_proto.py

@@ -256,13 +256,50 @@ async def test_connection_lost_exception_is_marked_retrieved(
     proto = ResponseHandler(loop=loop)
     proto.connection_made(mock.Mock())
 
+    # Access closed property before connection_lost to ensure future is created
+    closed_future = proto.closed
+    assert closed_future is not None
+
     # Simulate an SSL shutdown timeout error
     ssl_error = TimeoutError("SSL shutdown timed out")
     proto.connection_lost(ssl_error)
 
     # Verify the exception was set on the closed future
-    assert proto.closed.done()
-    exc = proto.closed.exception()
+    assert closed_future.done()
+    exc = closed_future.exception()
     assert exc is not None
     assert "Connection lost: SSL shutdown timed out" in str(exc)
     assert exc.__cause__ is ssl_error
+
+
+async def test_closed_property_lazy_creation(
+    loop: asyncio.AbstractEventLoop,
+) -> None:
+    """Test that closed future is created lazily."""
+    proto = ResponseHandler(loop=loop)
+
+    # Initially, the closed future should not be created
+    assert proto._closed is None
+
+    # Accessing the property should create the future
+    closed_future = proto.closed
+    assert closed_future is not None
+    assert isinstance(closed_future, asyncio.Future)
+    assert not closed_future.done()
+
+    # Subsequent access should return the same future
+    assert proto.closed is closed_future
+
+
+async def test_closed_property_after_connection_lost(
+    loop: asyncio.AbstractEventLoop,
+) -> None:
+    """Test that closed property returns None after connection_lost if never accessed."""
+    proto = ResponseHandler(loop=loop)
+    proto.connection_made(mock.Mock())
+
+    # Don't access proto.closed before connection_lost
+    proto.connection_lost(None)
+
+    # After connection_lost, closed should return None if it was never accessed
+    assert proto.closed is None

tests/test_connector.py

@@ -331,6 +331,28 @@ async def test_close_with_exception_during_closing(
     assert "Error while closing connector" in caplog.records[0].message
     assert "RuntimeError('Connection close failed')" in caplog.records[0].message
 
+
+async def test_close_with_proto_closed_none(key: ConnectionKey) -> None:
+    """Test close when protocol.closed is None."""
+    # Create protocols where closed property returns None
+    proto1 = mock.create_autospec(ResponseHandler, instance=True)
+    proto1.closed = None
+    proto1.close = mock.Mock()
+
+    proto2 = mock.create_autospec(ResponseHandler, instance=True)
+    proto2.closed = None
+    proto2.close = mock.Mock()
+
+    conn = aiohttp.BaseConnector()
+    conn._conns[key] = deque([(proto1, 0)])
+    conn._acquired.add(proto2)
+
+    # Close the connector - this should handle the case where proto.closed is None
+    await conn.close()
+
+    # Verify close was called on both protocols
+    assert proto1.close.called
+    assert proto2.close.called
     assert conn.closed