PYTHON-5898 Reduce pool checkout lock overhead and stop mutating TopologyDescription during selection

doc/changelog.rst

@@ -27,6 +27,23 @@ PyMongo 4.18 brings a number of changes including:
   buffer.
 - Fixed :func:`bson.json_util.loads` to reject ``$timestamp`` values containing
   fields other than ``t`` and ``i``.
+- Fixed ``client.primary`` raising ``IndexError``, and ``client.secondaries``
+  and ``client.arbiters`` returning stale results, after a retryable operation
+  deprioritized the primary.
+- **Breaking change**: removed the public ``TopologyDescription.candidate_servers``
+  attribute, which was added in PyMongo 4.16.0 and appeared in the 4.16 and
+  4.17 API documentation. Its value depended on which server-selection call
+  happened to run last, so it could not be relied on. Any code reading it must
+  be updated to use
+  :attr:`~pymongo.topology_description.TopologyDescription.known_servers`
+  instead.
+- Fixed a leak where every failed connection checkout permanently incremented
+  a pool's ``operation_count``. Because nothing short of a fork reset that
+  counter, a mongos that suffered a burst of checkout failures looked
+  permanently busier than its peers and was progressively avoided by server
+  selection for the remaining life of the client.
+- Reduced the number of lock acquisitions on the connection checkout fast
+  path.
 
 Changes in Version 4.17.0 (2026/04/20)
 --------------------------------------

pymongo/asynchronous/pool.py

@@ -1007,9 +1007,6 @@ async def _get_conn(
                 "Attempted to check out a connection from closed connection pool"
             )
 
-        async with self.lock:
-            self.operation_count += 1
-
         # Get a free socket or create one.
         if _csot.get_timeout():
             deadline = _csot.get_deadline()
@@ -1018,28 +1015,62 @@ async def _get_conn(
         else:
             deadline = None
 
-        async with self.size_cond:
-            self._raise_if_not_ready(checkout_started_time, emit_event=True)
-            while not (self.requests < self.max_pool_size):
-                timeout = deadline - time.monotonic() if deadline else None
-                if not await _async_cond_wait(self.size_cond, timeout):
-                    # Timed out, notify the next thread to ensure a
-                    # timeout doesn't consume the condition.
-                    if self.requests < self.max_pool_size:
-                        self.size_cond.notify()
-                    self._raise_wait_queue_timeout(checkout_started_time)
-                self._raise_if_not_ready(checkout_started_time, emit_event=True)
-            self.requests += 1
-
-        # We've now acquired the semaphore and must release it on error.
         conn = None
-        incremented = False
+        op_count_incremented = False
+        # Each flag is set immediately after its own increment, with no
+        # statement in between, which narrows the window in which an interrupt
+        # could leave a counter incremented but unrecorded to a single
+        # statement boundary. It does not close the window entirely: the
+        # increment and the flag assignment are still separate statements.
+        requests_incremented = False
+        active_sockets_incremented = False
+        # Invariant: any site inside the `try` below that emits a checkout
+        # failed event must set this so the outer handler does not re-emit.
         emitted_event = False
         is_new_conn = False
+
         try:
             async with self.lock:
-                self.active_sockets += 1
-                incremented = True
+                self.operation_count += 1
+                op_count_incremented = True
+                emitted_event = True
+                self._raise_if_not_ready(checkout_started_time, emit_event=True)
+                emitted_event = False
+                if self.requests < self.max_pool_size:
+                    # Fast path: a slot is immediately available, so do all
+                    # of the counter bookkeeping in this one critical section
+                    # and keep the checkout to a single lock acquisition.
+                    self.requests += 1
+                    requests_incremented = True
+                    self.active_sockets += 1
+                    active_sockets_incremented = True
+
+            if not requests_incremented:
+                # Slow path: no slot was free under the pool mutex. Re-check
+                # under size_cond -- a slot may have been released in the
+                # meantime, in which case the loop below never waits -- and
+                # otherwise wait for one to be released.
+                async with self.size_cond:
+                    emitted_event = True
+                    self._raise_if_not_ready(checkout_started_time, emit_event=True)
+                    emitted_event = False
+                    while not (self.requests < self.max_pool_size):
+                        timeout = deadline - time.monotonic() if deadline else None
+                        if not await _async_cond_wait(self.size_cond, timeout):
+                            # Timed out, notify the next thread to ensure a
+                            # timeout doesn't consume the condition.
+                            if self.requests < self.max_pool_size:
+                                self.size_cond.notify()
+                            emitted_event = True
+                            self._raise_wait_queue_timeout(checkout_started_time)
+                        emitted_event = True
+                        self._raise_if_not_ready(checkout_started_time, emit_event=True)
+                        emitted_event = False
+                    self.requests += 1
+                    requests_incremented = True
+                    self.active_sockets += 1
+                    active_sockets_incremented = True
+
             while conn is None:
                 # CMAP: we MUST wait for either maxConnecting OR for a socket
                 # to be checked back into the pool.
@@ -1084,11 +1115,17 @@ async def _get_conn(
             if conn:
                 # We checked out a socket but authentication failed.
                 await conn.close_conn(ConnectionClosedReason.ERROR)
-            async with self.size_cond:
-                self.requests -= 1
-                if incremented:
-                    self.active_sockets -= 1
-                self.size_cond.notify()
+            if op_count_incremented:
+                async with self.size_cond:
+                    self.operation_count -= 1
+                    if active_sockets_incremented:
+                        self.active_sockets -= 1
+                    if requests_incremented:
+                        self.requests -= 1
+                        # Notify only when a slot was actually released;
+                        # otherwise there is nothing for a blocked checkout to
+                        # wake up for.
+                        self.size_cond.notify()
 
             if not emitted_event:
                 self._telemetry.checkout_failed(

pymongo/server_selectors.py

@@ -35,8 +35,21 @@ class Selection:
     """Input or output of a server selector function."""
 
     @classmethod
-    def from_topology_description(cls, topology_description: TopologyDescription) -> Selection:
-        candidate_servers = topology_description.candidate_servers
+    def from_topology_description(
+        cls,
+        topology_description: TopologyDescription,
+        candidate_servers: Optional[list[ServerDescription]] = None,
+    ) -> Selection:
+        """Build a Selection from a TopologyDescription.
+
+        :param topology_description: the TopologyDescription to select from.
+        :param candidate_servers: the servers eligible for selection. Defaults
+            to all known servers. Server selection passes its own (per-call,
+            deprioritization-filtered) list here so that no state has to be
+            cached on the TopologyDescription.
+        """
+        if candidate_servers is None:
+            candidate_servers = topology_description.known_servers
         primary = None
         for sd in candidate_servers:
             if sd.server_type == SERVER_TYPE.RSPrimary:
@@ -45,7 +58,7 @@ def from_topology_description(cls, topology_description: TopologyDescription) ->
 
         return Selection(
             topology_description,
-            topology_description.candidate_servers,
+            candidate_servers,
             topology_description.common_wire_version,
             primary,
         )

pymongo/synchronous/pool.py

@@ -1003,9 +1003,6 @@ def _get_conn(
                 "Attempted to check out a connection from closed connection pool"
             )
 
-        with self.lock:
-            self.operation_count += 1
-
         # Get a free socket or create one.
         if _csot.get_timeout():
             deadline = _csot.get_deadline()
@@ -1014,28 +1011,62 @@ def _get_conn(
         else:
             deadline = None
 
-        with self.size_cond:
-            self._raise_if_not_ready(checkout_started_time, emit_event=True)
-            while not (self.requests < self.max_pool_size):
-                timeout = deadline - time.monotonic() if deadline else None
-                if not _cond_wait(self.size_cond, timeout):
-                    # Timed out, notify the next thread to ensure a
-                    # timeout doesn't consume the condition.
-                    if self.requests < self.max_pool_size:
-                        self.size_cond.notify()
-                    self._raise_wait_queue_timeout(checkout_started_time)
-                self._raise_if_not_ready(checkout_started_time, emit_event=True)
-            self.requests += 1
-
-        # We've now acquired the semaphore and must release it on error.
         conn = None
-        incremented = False
+        op_count_incremented = False
+        # Each flag is set immediately after its own increment, with no
+        # statement in between, which narrows the window in which an interrupt
+        # could leave a counter incremented but unrecorded to a single
+        # statement boundary. It does not close the window entirely: the
+        # increment and the flag assignment are still separate statements.
+        requests_incremented = False
+        active_sockets_incremented = False
+        # Invariant: any site inside the `try` below that emits a checkout
+        # failed event must set this so the outer handler does not re-emit.
         emitted_event = False
         is_new_conn = False
+
         try:
             with self.lock:
-                self.active_sockets += 1
-                incremented = True
+                self.operation_count += 1
+                op_count_incremented = True
+                emitted_event = True
+                self._raise_if_not_ready(checkout_started_time, emit_event=True)
+                emitted_event = False
+                if self.requests < self.max_pool_size:
+                    # Fast path: a slot is immediately available, so do all
+                    # of the counter bookkeeping in this one critical section
+                    # and keep the checkout to a single lock acquisition.
+                    self.requests += 1
+                    requests_incremented = True
+                    self.active_sockets += 1
+                    active_sockets_incremented = True
+
+            if not requests_incremented:
+                # Slow path: no slot was free under the pool mutex. Re-check
+                # under size_cond -- a slot may have been released in the
+                # meantime, in which case the loop below never waits -- and
+                # otherwise wait for one to be released.
+                with self.size_cond:
+                    emitted_event = True
+                    self._raise_if_not_ready(checkout_started_time, emit_event=True)
+                    emitted_event = False
+                    while not (self.requests < self.max_pool_size):
+                        timeout = deadline - time.monotonic() if deadline else None
+                        if not _cond_wait(self.size_cond, timeout):
+                            # Timed out, notify the next thread to ensure a
+                            # timeout doesn't consume the condition.
+                            if self.requests < self.max_pool_size:
+                                self.size_cond.notify()
+                            emitted_event = True
+                            self._raise_wait_queue_timeout(checkout_started_time)
+                        emitted_event = True
+                        self._raise_if_not_ready(checkout_started_time, emit_event=True)
+                        emitted_event = False
+                    self.requests += 1
+                    requests_incremented = True
+                    self.active_sockets += 1
+                    active_sockets_incremented = True
+
             while conn is None:
                 # CMAP: we MUST wait for either maxConnecting OR for a socket
                 # to be checked back into the pool.
@@ -1080,11 +1111,17 @@ def _get_conn(
             if conn:
                 # We checked out a socket but authentication failed.
                 conn.close_conn(ConnectionClosedReason.ERROR)
-            with self.size_cond:
-                self.requests -= 1
-                if incremented:
-                    self.active_sockets -= 1
-                self.size_cond.notify()
+            if op_count_incremented:
+                with self.size_cond:
+                    self.operation_count -= 1
+                    if active_sockets_incremented:
+                        self.active_sockets -= 1
+                    if requests_incremented:
+                        self.requests -= 1
+                        # Notify only when a slot was actually released;
+                        # otherwise there is nothing for a blocked checkout to
+                        # wake up for.
+                        self.size_cond.notify()
 
             if not emitted_event:
                 self._telemetry.checkout_failed(

pymongo/topology_description.py

@@ -84,7 +84,6 @@ def __init__(
         self._server_descriptions = server_descriptions
         self._max_set_version = max_set_version
         self._max_election_id = max_election_id
-        self._candidate_servers = list(self._server_descriptions.values())
 
         # The heartbeat_frequency is used in staleness estimates.
         self._topology_settings = topology_settings
@@ -238,11 +237,6 @@ def readable_servers(self) -> list[ServerDescription]:
         """List of readable Servers."""
         return [s for s in self._server_descriptions.values() if s.is_readable]
 
-    @property
-    def candidate_servers(self) -> list[ServerDescription]:
-        """List of Servers excluding deprioritized servers."""
-        return self._candidate_servers
-
     @property
     def common_wire_version(self) -> Optional[int]:
         """Minimum of all servers' max wire versions, or None."""
@@ -280,18 +274,34 @@ def _apply_local_threshold(self, selection: Optional[Selection]) -> list[ServerD
 
     def _filter_servers(
         self, deprioritized_servers: Optional[list[ServerDescription]] = None
-    ) -> None:
-        """Filter out deprioritized servers from a list of server candidates."""
+    ) -> list[ServerDescription]:
+        """Return the known servers with any deprioritized servers filtered out.
+
+        If every known server is deprioritized, all known servers are returned
+        so that selection can still make progress.
+
+        This method must not mutate ``self``. Deprioritization is specific to
+        a single call, but a TopologyDescription outlives it and is replaced,
+        not edited, when the topology changes. Filtering left behind on the
+        description would therefore leak into later readers: callers such as
+        :meth:`~pymongo.synchronous.topology.Topology.get_primary` build their
+        selection straight from the description and would see a view narrowed
+        by a selection they never asked for. The same object is also handed to
+        code running on other threads -- topology event listeners and anything
+        holding :attr:`~pymongo.mongo_client.MongoClient.topology_description`
+        -- which may read it concurrently with a selection in progress.
+
+        :param deprioritized_servers: servers to exclude, or None.
+        """
+        known_servers = self.known_servers
         if not deprioritized_servers:
-            self._candidate_servers = self.known_servers
-        else:
-            deprioritized_addresses = {sd.address for sd in deprioritized_servers}
-            filtered = [
-                server
-                for server in self.known_servers
-                if server.address not in deprioritized_addresses
-            ]
-            self._candidate_servers = filtered or self.known_servers
+            return known_servers
+
+        deprioritized_addresses = {sd.address for sd in deprioritized_servers}
+        filtered = [
+            server for server in known_servers if server.address not in deprioritized_addresses
+        ]
+        return filtered or known_servers
 
     def apply_selector(
         self,
@@ -335,10 +345,10 @@ def apply_selector(
             description = self.server_descriptions().get(address)
             return [description] if description and description.is_server_type_known else []
 
-        self._filter_servers(deprioritized_servers)
+        candidate_servers = self._filter_servers(deprioritized_servers)
         # Primary selection fast path.
         if self.topology_type == TOPOLOGY_TYPE.ReplicaSetWithPrimary and type(selector) is Primary:
-            for sd in self._candidate_servers:
+            for sd in candidate_servers:
                 if sd.server_type == SERVER_TYPE.RSPrimary:
                     sds = [sd]
                     if custom_selector:
@@ -355,13 +365,14 @@ def apply_selector(
             # No primary found, return an empty list.
             return []
 
-        selection = Selection.from_topology_description(self)
+        selection = Selection.from_topology_description(self, candidate_servers)
         # Ignore read preference for sharded clusters.
         if self.topology_type != TOPOLOGY_TYPE.Sharded:
             selection = selector(selection)
             # No suitable servers found, apply preference again but include deprioritized servers.
             if not selection and deprioritized_servers:
-                self._filter_servers(None)
+                # No candidate filtering: from_topology_description() defaults
+                # to all known servers.
                 selection = Selection.from_topology_description(self)
                 selection = selector(selection)