PYTHON-3923 - Handle socket closures in tests to avoid ResourceWarning

pymongo/asynchronous/pool.py

@@ -181,6 +181,11 @@ def __init__(
         # For gossiping $clusterTime from the connection handshake to the client.
         self._cluster_time = None
 
+    def __del__(self) -> None:
+        # Ensure all async connections are properly cleaned up on GC
+        if not _IS_SYNC and not getattr(self, "closed", True):
+            self.conn.abort()
+
     def set_conn_timeout(self, timeout: Optional[float]) -> None:
         """Cache last timeout to avoid duplicate calls to conn.settimeout."""
         if timeout == self.last_timeout:
@@ -1064,7 +1069,16 @@ async def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> A
                 _raise_connection_failure(self.address, error, timeout_details=details)
             raise
 
-        conn = AsyncConnection(networking_interface, self, self.address, conn_id, self.is_sdam)  # type: ignore[arg-type]
+        try:
+            conn = AsyncConnection(networking_interface, self, self.address, conn_id, self.is_sdam)  # type: ignore[arg-type]
+        except BaseException:
+            # Release the networking_interface's transport if AsyncConnection
+            # construction failed, since no AsyncConnection exists yet to
+            # close_conn() through the outer cleanup path.
+            networking_interface.abort()
+            async with self.lock:
+                self.active_contexts.discard(tmp_context)
+            raise
         async with self.lock:
             self.active_contexts.add(conn.cancel_context)
             self.active_contexts.discard(tmp_context)
@@ -1129,21 +1143,21 @@ async def checkout(
 
         conn = await self._get_conn(checkout_started_time, handler=handler)
 
-        duration = time.monotonic() - checkout_started_time
-        if self.enabled_for_cmap:
-            assert listeners is not None
-            listeners.publish_connection_checked_out(self.address, conn.id, duration)
-        if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _CONNECTION_LOGGER,
-                message=_ConnectionStatusMessage.CHECKOUT_SUCCEEDED,
-                clientId=self._client_id,
-                serverHost=self.address[0],
-                serverPort=self.address[1],
-                driverConnectionId=conn.id,
-                durationMS=duration,
-            )
         try:
+            duration = time.monotonic() - checkout_started_time
+            if self.enabled_for_cmap:
+                assert listeners is not None
+                listeners.publish_connection_checked_out(self.address, conn.id, duration)
+            if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
+                _debug_log(
+                    _CONNECTION_LOGGER,
+                    message=_ConnectionStatusMessage.CHECKOUT_SUCCEEDED,
+                    clientId=self._client_id,
+                    serverHost=self.address[0],
+                    serverPort=self.address[1],
+                    driverConnectionId=conn.id,
+                    durationMS=duration,
+                )
             async with self.lock:
                 self.active_contexts.add(conn.cancel_context)
             yield conn
@@ -1160,7 +1174,12 @@ async def checkout(
                 exc_type, exc_val, _ = sys.exc_info()
                 await handler.handle(exc_type, exc_val)
             if not pinned and conn.active:
-                await self.checkin(conn)
+                try:
+                    await self.checkin(conn)
+                except BaseException:
+                    # If checkin is interrupted, the conn is orphaned. Abort to avoid leaking the socket
+                    conn.conn.abort()
+                    raise
             raise
         if conn.pinned_txn:
             async with self.lock:

pymongo/asynchronous/server.py

@@ -114,8 +114,14 @@ async def close(self) -> None:
                 serverPort=self._description.address[1],
             )
 
-        await self._monitor.close()
+        monitor_error: Optional[BaseException] = None
+        try:
+            await self._monitor.close()
+        except BaseException as exc:
+            monitor_error = exc
         await self._pool.close()
+        if monitor_error is not None:
+            raise monitor_error
 
     def request_check(self) -> None:
         """Check the server's state soon."""

pymongo/asynchronous/topology.py

@@ -722,8 +722,15 @@ async def close(self) -> None:
         """
         async with self._lock:
             old_td = self._description
+            first_close_error: Optional[BaseException] = None
             for server in self._servers.values():
-                await server.close()
+                # Each server.close must run independently. A failure on one
+                # server must not skip the remaining servers
+                try:
+                    await server.close()
+                except BaseException as exc:
+                    if first_close_error is None:
+                        first_close_error = exc
                 if not _IS_SYNC:
                     self._monitor_tasks.append(server._monitor)
 
@@ -783,6 +790,9 @@ async def close(self) -> None:
             await self.__events_executor.join(1)
             process_events_queue(weakref.ref(self._events))  # type: ignore[arg-type]
 
+        if first_close_error is not None:
+            raise first_close_error
+
     @property
     def description(self) -> TopologyDescription:
         return self._description
@@ -979,13 +989,24 @@ async def _update_servers(self) -> None:
                 if was_writable != sd.is_writable:
                     await self._servers[address].pool.update_is_writable(sd.is_writable)
 
+        first_close_error: Optional[BaseException] = None
         for address, server in list(self._servers.items()):
             if not self._description.has_server(address):
-                await server.close()
+                # Each server.close must run independently. If one server's
+                # cleanup raises, we still need to remove it from _servers and
+                # add its monitor to _monitor_tasks, and we must process the
+                # remaining servers, otherwise their pool conns leak.
+                try:
+                    await server.close()
+                except BaseException as exc:
+                    first_close_error = exc
                 if not _IS_SYNC:
                     self._monitor_tasks.append(server._monitor)
                 self._servers.pop(address)
 
+        if first_close_error is not None:
+            raise first_close_error
+
     def _create_pool_for_server(self, address: _Address) -> Pool:
         return self._settings.pool_class(
             address, self._settings.pool_options, client_id=self._topology_id

pymongo/network_layer.py

@@ -402,6 +402,9 @@ def close(self) -> Any:
     def is_closing(self) -> bool:
         raise NotImplementedError
 
+    def abort(self) -> None:
+        raise NotImplementedError
+
     @property
     def get_conn(self) -> Any:
         raise NotImplementedError
@@ -429,6 +432,9 @@ async def close(self) -> None:
     def is_closing(self) -> bool:
         return self.conn[0].is_closing()
 
+    def abort(self) -> None:
+        self.conn[1].abort()
+
     @property
     def get_conn(self) -> PyMongoProtocol:
         return self.conn[1]
@@ -454,6 +460,9 @@ def close(self) -> None:
     def is_closing(self) -> bool:
         return self.conn.is_closing()
 
+    def abort(self) -> None:
+        return
+
     @property
     def get_conn(self) -> Union[socket.socket, _sslConn]:
         return self.conn
@@ -675,6 +684,21 @@ def close(self, exc: Optional[Exception] = None) -> None:
         self._resolve_pending_messages(exc)
         self._connection_lost = True
 
+    def abort(self) -> None:
+        """Synchronously abort the underlying transport. Idempotent, best-effort.
+
+        Unlike :meth:`close`, this does not resolve pending messages and is
+        safe to call during cleanup paths where the protocol may be
+        partially initialized or the event loop already closed.
+        """
+        transport = self.transport
+        if transport is None:
+            return
+        try:
+            transport.abort()
+        except Exception:  # noqa: S110
+            pass
+
     def connection_lost(self, exc: Optional[Exception] = None) -> None:
         self._resolve_pending_messages(exc)
         if not self._closed.done():

pymongo/pool_shared.py

@@ -207,6 +207,7 @@ async def _async_create_connection(address: _Address, options: PoolOptions) -> s
             sock = socket.socket(af, socktype, proto)
         # Fallback when SOCK_CLOEXEC isn't available.
         _set_non_inheritable_non_atomic(sock.fileno())
+        sock_returned = False
         try:
             sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
             # CSOT: apply timeout to socket connect.
@@ -223,14 +224,18 @@ async def _async_create_connection(address: _Address, options: PoolOptions) -> s
                 asyncio.get_running_loop().sock_connect(sock, sa), timeout=timeout
             )
             sock.settimeout(timeout)
+            # Set immediately before return. Do not insert an await between this and the return
+            sock_returned = True
             return sock
         except asyncio.TimeoutError as e:
-            sock.close()
             err = socket.timeout("timed out")
             err.__cause__ = e
         except OSError as e:
-            sock.close()
             err = e  # type: ignore[assignment]
+        finally:
+            # Always close the socket if it wasn't returned to avoid leaks.
+            if not sock_returned:
+                sock.close()
 
     if err is not None:
         raise err
@@ -309,46 +314,58 @@ async def _configured_protocol_interface(
     sock = await _async_create_connection(address, options)
     ssl_context = options._ssl_context
     timeout = options.socket_timeout
-
-    if ssl_context is None:
-        return AsyncNetworkingInterface(
-            await asyncio.get_running_loop().create_connection(
-                lambda: PyMongoProtocol(timeout=timeout), sock=sock
-            )
-        )
-
-    host = address[0]
+    # Create the Protocol early to prevent asyncio resource leaks during cleanup path
+    protocol = PyMongoProtocol(timeout=timeout)
+    sock_adopted = False
     try:
-        # We have to pass hostname / ip address to wrap_socket
-        # to use SSLContext.check_hostname.
-        transport, protocol = await asyncio.get_running_loop().create_connection(  # type: ignore[call-overload]
-            lambda: PyMongoProtocol(timeout=timeout),
-            sock=sock,
-            server_hostname=host,
-            ssl=ssl_context,
-        )
-    except _CertificateError:
-        # Raise _CertificateError directly like we do after match_hostname
-        # below.
-        raise
-    except (OSError, *SSLErrors) as exc:
-        # We raise AutoReconnect for transient and permanent SSL handshake
-        # failures alike. Permanent handshake failures, like protocol
-        # mismatch, will be turned into ServerSelectionTimeoutErrors later.
-        details = _get_timeout_details(options)
-        _raise_connection_failure(address, exc, "SSL handshake failed: ", timeout_details=details)
-    if (
-        ssl_context.verify_mode
-        and not ssl_context.check_hostname
-        and not options.tls_allow_invalid_hostnames
-    ):
+        if ssl_context is None:
+            result = await asyncio.get_running_loop().create_connection(lambda: protocol, sock=sock)
+            sock_adopted = True
+            return AsyncNetworkingInterface(result)
+
+        host = address[0]
         try:
-            ssl.match_hostname(transport.get_extra_info("peercert"), hostname=host)  # type:ignore[attr-defined,unused-ignore]
+            # We have to pass hostname / ip address to wrap_socket
+            # to use SSLContext.check_hostname.
+            transport, _ = await asyncio.get_running_loop().create_connection(  # type: ignore[call-overload]
+                lambda: protocol,
+                sock=sock,
+                server_hostname=host,
+                ssl=ssl_context,
+            )
+            sock_adopted = True
         except _CertificateError:
-            transport.abort()
+            # Raise _CertificateError directly like we do after match_hostname
+            # below.
             raise
-
-    return AsyncNetworkingInterface((transport, protocol))
+        except (OSError, *SSLErrors) as exc:
+            # We raise AutoReconnect for transient and permanent SSL handshake
+            # failures alike. Permanent handshake failures, like protocol
+            # mismatch, will be turned into ServerSelectionTimeoutErrors later.
+            details = _get_timeout_details(options)
+            _raise_connection_failure(
+                address, exc, "SSL handshake failed: ", timeout_details=details
+            )
+        if (
+            ssl_context.verify_mode
+            and not ssl_context.check_hostname
+            and not options.tls_allow_invalid_hostnames
+        ):
+            try:
+                ssl.match_hostname(transport.get_extra_info("peercert"), hostname=host)  # type:ignore[attr-defined,unused-ignore]
+            except _CertificateError:
+                transport.abort()
+                raise
+
+        return AsyncNetworkingInterface((transport, protocol))
+    finally:
+        if not sock_adopted:
+            # If the protocol owns the transport, it also adopted the socket and needs to be cleaned up from the transport
+            if protocol.transport is not None:
+                protocol.abort()
+            # Otherwise the socket was never adopted, close it directly
+            else:
+                sock.close()
 
 
 def _create_connection(address: _Address, options: PoolOptions) -> socket.socket: