Merge branch 'master' of github.com:mongodb/mongo-python-driver into cleanup_task_combos

Author: blink1073

.evergreen/generated_configs/variants.yml

@@ -626,17 +626,19 @@ buildvariants:
       - macos-14
     batchtime: 10080
     expansions:
+      TEST_NAME: default
       SUB_TEST_NAME: pyopenssl
       PYTHON_BINARY: /Library/Frameworks/Python.Framework/Versions/3.9/bin/python3
   - name: pyopenssl-rhel8-python3.10
     tasks:
-      - name: .replica_set .auth .ssl .sync
-      - name: .7.0 .auth .ssl .sync
+      - name: .replica_set .auth .ssl .sync_async
+      - name: .7.0 .auth .ssl .sync_async
     display_name: PyOpenSSL RHEL8 Python3.10
     run_on:
       - rhel87-small
     batchtime: 10080
     expansions:
+      TEST_NAME: default
       SUB_TEST_NAME: pyopenssl
       PYTHON_BINARY: /opt/python/3.10/bin/python3
   - name: pyopenssl-rhel8-python3.11
@@ -648,6 +650,7 @@ buildvariants:
       - rhel87-small
     batchtime: 10080
     expansions:
+      TEST_NAME: default
       SUB_TEST_NAME: pyopenssl
       PYTHON_BINARY: /opt/python/3.11/bin/python3
   - name: pyopenssl-rhel8-python3.12
@@ -659,17 +662,19 @@ buildvariants:
       - rhel87-small
     batchtime: 10080
     expansions:
+      TEST_NAME: default
       SUB_TEST_NAME: pyopenssl
       PYTHON_BINARY: /opt/python/3.12/bin/python3
   - name: pyopenssl-win64-python3.13
     tasks:
-      - name: .replica_set .auth .ssl .sync
-      - name: .7.0 .auth .ssl .sync
+      - name: .replica_set .auth .ssl .sync_async
+      - name: .7.0 .auth .ssl .sync_async
     display_name: PyOpenSSL Win64 Python3.13
     run_on:
       - windows-64-vsMulti-small
     batchtime: 10080
     expansions:
+      TEST_NAME: default
       SUB_TEST_NAME: pyopenssl
       PYTHON_BINARY: C:/python/Python313/python.exe
   - name: pyopenssl-rhel8-pypy3.10
@@ -681,6 +686,7 @@ buildvariants:
       - rhel87-small
     batchtime: 10080
     expansions:
+      TEST_NAME: default
       SUB_TEST_NAME: pyopenssl
       PYTHON_BINARY: /opt/python/pypy3.10/bin/python3
 

.evergreen/scripts/generate_config.py

@@ -253,7 +253,7 @@ def create_enterprise_auth_variants():
 def create_pyopenssl_variants():
     base_name = "PyOpenSSL"
     batchtime = BATCHTIME_WEEK
-    expansions = dict(SUB_TEST_NAME="pyopenssl")
+    expansions = dict(TEST_NAME="default", SUB_TEST_NAME="pyopenssl")
     variants = []
 
     for python in ALL_PYTHONS:
@@ -268,14 +268,25 @@ def create_pyopenssl_variants():
             host = DEFAULT_HOST
 
         display_name = get_variant_name(base_name, host, python=python)
-        variant = create_variant(
-            [f".replica_set .{auth} .{ssl} .sync", f".7.0 .{auth} .{ssl} .sync"],
-            display_name,
-            python=python,
-            host=host,
-            expansions=expansions,
-            batchtime=batchtime,
-        )
+        # only need to run some on async
+        if python in (CPYTHONS[1], CPYTHONS[-1]):
+            variant = create_variant(
+                [f".replica_set .{auth} .{ssl} .sync_async", f".7.0 .{auth} .{ssl} .sync_async"],
+                display_name,
+                python=python,
+                host=host,
+                expansions=expansions,
+                batchtime=batchtime,
+            )
+        else:
+            variant = create_variant(
+                [f".replica_set .{auth} .{ssl} .sync", f".7.0 .{auth} .{ssl} .sync"],
+                display_name,
+                python=python,
+                host=host,
+                expansions=expansions,
+                batchtime=batchtime,
+            )
         variants.append(variant)
 
     return variants

.github/workflows/codeql.yml

@@ -54,7 +54,6 @@ jobs:
         queries: security-extended
         config: |
           paths-ignore:
-            - '.github/**'
             - 'doc/**'
             - 'tools/**'
             - 'test/**'

doc/changelog.rst

@@ -16,6 +16,10 @@ Version 4.12.1 is a bug fix release.
   errors such as: "NotImplementedError: Database objects do not implement truth value testing or bool()".
 - Removed Eventlet testing against Python versions newer than 3.9 since
   Eventlet is actively being sunset by its maintainers and has compatibility issues with PyMongo's dnspython dependency.
+- Fixed a bug where MongoDB cluster topology changes could cause asynchronous operations to take much longer to complete
+  due to holding the Topology lock while closing stale connections.
+- Fixed a bug that would cause AsyncMongoClient to attempt to use PyOpenSSL when available, resulting in errors such as
+  "pymongo.errors.ServerSelectionTimeoutError: 'SSLContext' object has no attribute 'wrap_bio'".
 
 Issues Resolved
 ...............

pymongo/asynchronous/encryption.py

@@ -87,7 +87,7 @@
 from pymongo.results import BulkWriteResult, DeleteResult
 from pymongo.ssl_support import BLOCKING_IO_ERRORS, get_ssl_context
 from pymongo.typings import _DocumentType, _DocumentTypeArg
-from pymongo.uri_parser_shared import parse_host
+from pymongo.uri_parser_shared import _parse_kms_tls_options, parse_host
 from pymongo.write_concern import WriteConcern
 
 if TYPE_CHECKING:
@@ -157,6 +157,7 @@ def __init__(
         self.mongocryptd_client = mongocryptd_client
         self.opts = opts
         self._spawned = False
+        self._kms_ssl_contexts = opts._kms_ssl_contexts(_IS_SYNC)
 
     async def kms_request(self, kms_context: MongoCryptKmsContext) -> None:
         """Complete a KMS request.
@@ -168,7 +169,7 @@ async def kms_request(self, kms_context: MongoCryptKmsContext) -> None:
         endpoint = kms_context.endpoint
         message = kms_context.message
         provider = kms_context.kms_provider
-        ctx = self.opts._kms_ssl_contexts.get(provider)
+        ctx = self._kms_ssl_contexts.get(provider)
         if ctx is None:
             # Enable strict certificate verification, OCSP, match hostname, and
             # SNI using the system default CA certificates.
@@ -180,6 +181,7 @@ async def kms_request(self, kms_context: MongoCryptKmsContext) -> None:
                 False,  # allow_invalid_certificates
                 False,  # allow_invalid_hostnames
                 False,  # disable_ocsp_endpoint_check
+                _IS_SYNC,
             )
         # CSOT: set timeout for socket creation.
         connect_timeout = max(_csot.clamp_remaining(_KMS_CONNECT_TIMEOUT), 0.001)
@@ -396,6 +398,8 @@ def __init__(self, client: AsyncMongoClient[_DocumentTypeArg], opts: AutoEncrypt
             encrypted_fields_map = _dict_to_bson(opts._encrypted_fields_map, False, _DATA_KEY_OPTS)
         self._bypass_auto_encryption = opts._bypass_auto_encryption
         self._internal_client = None
+        # parsing kms_ssl_contexts here so that parsing errors will be raised before internal clients are created
+        opts._kms_ssl_contexts(_IS_SYNC)
 
         def _get_internal_client(
             encrypter: _Encrypter, mongo_client: AsyncMongoClient[_DocumentTypeArg]
@@ -675,6 +679,7 @@ def __init__(
             kms_tls_options=kms_tls_options,
             key_expiration_ms=key_expiration_ms,
         )
+        self._kms_ssl_contexts = _parse_kms_tls_options(opts._kms_tls_options, _IS_SYNC)
         self._io_callbacks: Optional[_EncryptionIO] = _EncryptionIO(
             None, key_vault_coll, None, opts
         )

pymongo/asynchronous/pool.py

@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import collections
 import contextlib
 import logging
@@ -75,6 +76,7 @@
 from pymongo.network_layer import AsyncNetworkingInterface, async_receive_message, async_sendall
 from pymongo.pool_options import PoolOptions
 from pymongo.pool_shared import (
+    SSLErrors,
     _CancellationContext,
     _configured_protocol_interface,
     _get_timeout_details,
@@ -85,7 +87,6 @@
 from pymongo.server_api import _add_to_command
 from pymongo.server_type import SERVER_TYPE
 from pymongo.socket_checker import SocketChecker
-from pymongo.ssl_support import SSLError
 
 if TYPE_CHECKING:
     from bson import CodecOptions
@@ -637,7 +638,7 @@ async def _raise_connection_failure(self, error: BaseException) -> NoReturn:
             reason = ConnectionClosedReason.ERROR
         await self.close_conn(reason)
         # SSLError from PyOpenSSL inherits directly from Exception.
-        if isinstance(error, (IOError, OSError, SSLError)):
+        if isinstance(error, (IOError, OSError, *SSLErrors)):
             details = _get_timeout_details(self.opts)
             _raise_connection_failure(self.address, error, timeout_details=details)
         else:
@@ -860,8 +861,14 @@ async def _reset(
         # PoolClosedEvent but that reset() SHOULD close sockets *after*
         # publishing the PoolClearedEvent.
         if close:
-            for conn in sockets:
-                await conn.close_conn(ConnectionClosedReason.POOL_CLOSED)
+            if not _IS_SYNC:
+                await asyncio.gather(
+                    *[conn.close_conn(ConnectionClosedReason.POOL_CLOSED) for conn in sockets],
+                    return_exceptions=True,
+                )
+            else:
+                for conn in sockets:
+                    await conn.close_conn(ConnectionClosedReason.POOL_CLOSED)
             if self.enabled_for_cmap:
                 assert listeners is not None
                 listeners.publish_pool_closed(self.address)
@@ -891,8 +898,14 @@ async def _reset(
                         serverPort=self.address[1],
                         serviceId=service_id,
                     )
-            for conn in sockets:
-                await conn.close_conn(ConnectionClosedReason.STALE)
+            if not _IS_SYNC:
+                await asyncio.gather(
+                    *[conn.close_conn(ConnectionClosedReason.STALE) for conn in sockets],
+                    return_exceptions=True,
+                )
+            else:
+                for conn in sockets:
+                    await conn.close_conn(ConnectionClosedReason.STALE)
 
     async def update_is_writable(self, is_writable: Optional[bool]) -> None:
         """Updates the is_writable attribute on all sockets currently in the
@@ -938,8 +951,14 @@ async def remove_stale_sockets(self, reference_generation: int) -> None:
                     and self.conns[-1].idle_time_seconds() > self.opts.max_idle_time_seconds
                 ):
                     close_conns.append(self.conns.pop())
-            for conn in close_conns:
-                await conn.close_conn(ConnectionClosedReason.IDLE)
+            if not _IS_SYNC:
+                await asyncio.gather(
+                    *[conn.close_conn(ConnectionClosedReason.IDLE) for conn in close_conns],
+                    return_exceptions=True,
+                )
+            else:
+                for conn in close_conns:
+                    await conn.close_conn(ConnectionClosedReason.IDLE)
 
         while True:
             async with self.size_cond:
@@ -1033,7 +1052,7 @@ async def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> A
                     reason=_verbose_connection_error_reason(ConnectionClosedReason.ERROR),
                     error=ConnectionClosedReason.ERROR,
                 )
-            if isinstance(error, (IOError, OSError, SSLError)):
+            if isinstance(error, (IOError, OSError, *SSLErrors)):
                 details = _get_timeout_details(self.opts)
                 _raise_connection_failure(self.address, error, timeout_details=details)
 

pymongo/asynchronous/topology.py

@@ -529,12 +529,6 @@ async def _process_change(
             if not _IS_SYNC:
                 self._monitor_tasks.append(self._srv_monitor)
 
-        # Clear the pool from a failed heartbeat.
-        if reset_pool:
-            server = self._servers.get(server_description.address)
-            if server:
-                await server.pool.reset(interrupt_connections=interrupt_connections)
-
         # Wake anything waiting in select_servers().
         self._condition.notify_all()
 
@@ -557,6 +551,11 @@ async def on_change(
             # that didn't include this server.
             if self._opened and self._description.has_server(server_description.address):
                 await self._process_change(server_description, reset_pool, interrupt_connections)
+        # Clear the pool from a failed heartbeat, done outside the lock to avoid blocking on connection close.
+        if reset_pool:
+            server = self._servers.get(server_description.address)
+            if server:
+                await server.pool.reset(interrupt_connections=interrupt_connections)
 
     async def _process_srv_update(self, seedlist: list[tuple[str, Any]]) -> None:
         """Process a new seedlist on an opened topology.

pymongo/client_options.py

@@ -84,7 +84,9 @@ def _parse_read_concern(options: Mapping[str, Any]) -> ReadConcern:
     return ReadConcern(concern)
 
 
-def _parse_ssl_options(options: Mapping[str, Any]) -> tuple[Optional[SSLContext], bool]:
+def _parse_ssl_options(
+    options: Mapping[str, Any], is_sync: bool
+) -> tuple[Optional[SSLContext], bool]:
     """Parse ssl options."""
     use_tls = options.get("tls")
     if use_tls is not None:
@@ -138,6 +140,7 @@ def _parse_ssl_options(options: Mapping[str, Any]) -> tuple[Optional[SSLContext]
             allow_invalid_certificates,
             allow_invalid_hostnames,
             disable_ocsp_endpoint_check,
+            is_sync,
         )
         return ctx, allow_invalid_hostnames
     return None, allow_invalid_hostnames
@@ -167,7 +170,7 @@ def _parse_pool_options(
     compression_settings = CompressionSettings(
         options.get("compressors", []), options.get("zlibcompressionlevel", -1)
     )
-    ssl_context, tls_allow_invalid_hostnames = _parse_ssl_options(options)
+    ssl_context, tls_allow_invalid_hostnames = _parse_ssl_options(options, is_sync)
     load_balanced = options.get("loadbalanced")
     max_connecting = options.get("maxconnecting", common.MAX_CONNECTING)
     return PoolOptions(

pymongo/encryption_options.py

@@ -20,6 +20,8 @@
 
 from typing import TYPE_CHECKING, Any, Mapping, Optional
 
+from pymongo.uri_parser_shared import _parse_kms_tls_options
+
 try:
     import pymongocrypt  # type:ignore[import-untyped] # noqa: F401
 
@@ -32,9 +34,9 @@
 from bson import int64
 from pymongo.common import validate_is_mapping
 from pymongo.errors import ConfigurationError
-from pymongo.uri_parser_shared import _parse_kms_tls_options
 
 if TYPE_CHECKING:
+    from pymongo.pyopenssl_context import SSLContext
     from pymongo.typings import _AgnosticMongoClient, _DocumentTypeArg
 
 
@@ -236,10 +238,22 @@ def __init__(
         if not any("idleShutdownTimeoutSecs" in s for s in self._mongocryptd_spawn_args):
             self._mongocryptd_spawn_args.append("--idleShutdownTimeoutSecs=60")
         # Maps KMS provider name to a SSLContext.
-        self._kms_ssl_contexts = _parse_kms_tls_options(kms_tls_options)
+        self._kms_tls_options = kms_tls_options
+        self._sync_kms_ssl_contexts: Optional[dict[str, SSLContext]] = None
+        self._async_kms_ssl_contexts: Optional[dict[str, SSLContext]] = None
         self._bypass_query_analysis = bypass_query_analysis
         self._key_expiration_ms = key_expiration_ms
 
+    def _kms_ssl_contexts(self, is_sync: bool) -> dict[str, SSLContext]:
+        if is_sync:
+            if self._sync_kms_ssl_contexts is None:
+                self._sync_kms_ssl_contexts = _parse_kms_tls_options(self._kms_tls_options, True)
+            return self._sync_kms_ssl_contexts
+        else:
+            if self._async_kms_ssl_contexts is None:
+                self._async_kms_ssl_contexts = _parse_kms_tls_options(self._kms_tls_options, False)
+            return self._async_kms_ssl_contexts
+
 
 class RangeOpts:
     """Options to configure encrypted queries using the range algorithm."""