From c02817ff792c1f5e4683e754cf6e793668c17a86 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Thu, 30 Jul 2026 06:34:31 -0500 Subject: [PATCH 01/25] PYTHON-5898 Make TopologyDescription.apply_selector() non-mutating _filter_servers() cached its per-call filtered list on self._candidate_servers, and apply_selector() read it back through the candidate_servers property. TopologyDescription is a shared, publicly exposed *immutable* snapshot, so this left candidate_servers permanently stale after any selection that deprioritized servers: the cached value from the last call, not the true set of known servers. This is a regression from PYTHON-5662 (4.16.0, commit 0cfba499), which changed Selection.from_topology_description() to default to candidate_servers instead of known_servers. On current main the stale cache makes Topology.get_primary() (and client.primary) raise IndexError on the unguarded selection[0] after any retryable operation deprioritized the primary, and makes client.secondaries/client.arbiters silently return stale results. All of these are single-threaded, user-facing symptoms, not a concurrency issue. Return the candidate list instead and pass it explicitly to Selection.from_topology_description(), restoring TopologyDescription to a genuinely immutable snapshot after construction and restoring candidate_servers to its pre-4.16.0 value. --- pymongo/server_selectors.py | 19 ++++++- pymongo/topology_description.py | 50 ++++++++++------- test/test_topology.py | 97 +++++++++++++++++++++++++++++++++ 3 files changed, 144 insertions(+), 22 deletions(-) diff --git a/pymongo/server_selectors.py b/pymongo/server_selectors.py index b43272f7f3..f1f0c81272 100644 --- a/pymongo/server_selectors.py +++ b/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 ``topology_description.candidate_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.candidate_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, ) diff --git a/pymongo/topology_description.py b/pymongo/topology_description.py index 87966aca45..352e8a7647 100644 --- a/pymongo/topology_description.py +++ b/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 @@ -240,8 +239,13 @@ def readable_servers(self) -> list[ServerDescription]: @property def candidate_servers(self) -> list[ServerDescription]: - """List of Servers excluding deprioritized servers.""" - return self._candidate_servers + """List of Servers eligible for selection when nothing is deprioritized. + + Deprioritization is per server-selection call, so it is applied by + :meth:`apply_selector` (via :meth:`_filter_servers`) and deliberately + not cached on the (immutable) TopologyDescription. + """ + return self.known_servers @property def common_wire_version(self) -> Optional[int]: @@ -280,18 +284,27 @@ 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``: server selection runs it while + holding only a shared read lock, so concurrent callers with different + ``deprioritized_servers`` would otherwise clobber each other's results. + + :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 +348,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,14 +368,13 @@ 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) - selection = Selection.from_topology_description(self) + selection = Selection.from_topology_description(self, self._filter_servers(None)) selection = selector(selection) # Apply custom selector followed by localThresholdMS. diff --git a/test/test_topology.py b/test/test_topology.py index 47e670b20d..7ddbffed0d 100644 --- a/test/test_topology.py +++ b/test/test_topology.py @@ -17,6 +17,7 @@ from __future__ import annotations import sys +import threading from pymongo.operations import _Op @@ -907,5 +908,101 @@ def test_no_mongoses(self): self.assertMessage("No mongoses available", t) +def create_mock_replica_set_topology(hosts=("a", "b", "c")): + """A ReplicaSetWithPrimary topology: hosts[0] is primary, the rest secondary.""" + t = create_mock_topology(seeds=list(hosts), replica_set_name="rs") + got_hello( + t, + (hosts[0], 27017), + { + "ok": 1, + HelloCompat.LEGACY_CMD: True, + "setName": "rs", + "hosts": list(hosts), + "maxWireVersion": common.MIN_SUPPORTED_WIRE_VERSION, + }, + ) + for host in hosts[1:]: + got_hello( + t, + (host, 27017), + { + "ok": 1, + HelloCompat.LEGACY_CMD: False, + "secondary": True, + "setName": "rs", + "hosts": list(hosts), + "maxWireVersion": common.MIN_SUPPORTED_WIRE_VERSION, + }, + ) + return t + + +class TestTopologyDescriptionConcurrency(TopologyTest): + """apply_selector() runs under a shared read lock (PYTHON-5898), so it must + not mutate the TopologyDescription: concurrent callers passing different + deprioritized_servers would otherwise clobber each other's candidate lists. + """ + + def test_concurrent_apply_selector_with_deprioritized_servers(self): + t = create_mock_replica_set_topology() + self.addCleanup(t.close) + description = t.description + self.assertEqual(TOPOLOGY_TYPE.ReplicaSetWithPrimary, description.topology_type) + primary_sd = description.server_descriptions()[("a", 27017)] + self.assertEqual(SERVER_TYPE.RSPrimary, primary_sd.server_type) + + # Make the interpreter switch threads aggressively so a mutation race + # inside apply_selector() is caught reliably rather than occasionally. + old_interval = sys.getswitchinterval() + sys.setswitchinterval(1e-6) + self.addCleanup(sys.setswitchinterval, old_interval) + + iterations = 500 + errors: list[str] = [] + barrier = threading.Barrier(6) + + def deprioritizing_worker(): + barrier.wait() + for _ in range(iterations): + # A retryable write retrying away from the primary must never + # be handed the deprioritized primary back. + sds = description.apply_selector( + ReadPreference.PRIMARY_PREFERRED, deprioritized_servers=[primary_sd] + ) + if any(sd.address == primary_sd.address for sd in sds): + errors.append(f"deprioritized primary was selected: {sds}") + + def plain_worker(): + barrier.wait() + for _ in range(iterations): + # An ordinary operation must always find the healthy primary. + sds = description.apply_selector(Primary()) + if [sd.address for sd in sds] != [primary_sd.address]: + errors.append(f"Primary() selected {sds} instead of the primary") + + threads = [threading.Thread(target=deprioritizing_worker) for _ in range(3)] + threads += [threading.Thread(target=plain_worker) for _ in range(3)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + self.assertEqual([], errors[:5], f"{len(errors)} racy selections") + + def test_apply_selector_does_not_mutate_description(self): + t = create_mock_replica_set_topology() + self.addCleanup(t.close) + description = t.description + primary_sd = description.server_descriptions()[("a", 27017)] + + before = list(description.candidate_servers) + description.apply_selector( + ReadPreference.PRIMARY_PREFERRED, deprioritized_servers=[primary_sd] + ) + self.assertEqual(before, description.candidate_servers) + self.assertIn(primary_sd, description.candidate_servers) + + if __name__ == "__main__": unittest.main() From d0e29570fe0adecfbc4305f3e2bebe86af886047 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Thu, 30 Jul 2026 06:34:31 -0500 Subject: [PATCH 02/25] PYTHON-5898 Merge pool checkout lock acquisitions on the fast path Pool._get_conn() used to take self.lock, then self.size_cond, then self.lock again just to bump operation_count and requests/active_sockets on an uncontended checkout. self.lock, size_cond, and _max_connecting_cond all wrap the same underlying mutex, so those three acquisitions were serializing on one lock for no reason. On the fast (uncontended) path, do the operation_count, requests, and active_sockets bookkeeping in a single critical section under self.lock, falling through to the old size_cond wait loop only when no pool slot is immediately available. The contended (slow) path is otherwise unchanged. Also wrap the whole checkout body in try/except so operation_count is always decremented on failure, fixing a pre-existing leak where a failed checkout left operation_count incremented forever, skewing server selection among mongoses toward the affected server. --- pymongo/asynchronous/pool.py | 74 ++++++++++++++++++++++++------------ pymongo/synchronous/pool.py | 74 ++++++++++++++++++++++++------------ 2 files changed, 98 insertions(+), 50 deletions(-) diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py index fdf3b1d816..3e3296b974 100644 --- a/pymongo/asynchronous/pool.py +++ b/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,50 @@ 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 + requests_incremented = False + active_sockets_incremented = False emitted_event = False is_new_conn = False + try: + slot_acquired = False async with self.lock: - self.active_sockets += 1 - incremented = True + self.operation_count += 1 + op_count_incremented = True + self._raise_if_not_ready(checkout_started_time, emit_event=True) + if self.requests < self.max_pool_size: + # Fast path: a slot is immediately available, so do all + # of the counter bookkeeping in this single critical + # section instead of taking the lock multiple times. + self.requests += 1 + requests_incremented = True + self.active_sockets += 1 + active_sockets_incremented = True + slot_acquired = True + + if not slot_acquired: + # Slow path: no slot was free. Fall back to waiting on the + # requests semaphore, exactly as before. + 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 + requests_incremented = True + + async with self.lock: + 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 +1103,16 @@ 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 requests_incremented or active_sockets_incremented: + async with self.size_cond: + if requests_incremented: + self.requests -= 1 + if active_sockets_incremented: + self.active_sockets -= 1 + self.size_cond.notify() + if op_count_incremented: + async with self.lock: + self.operation_count -= 1 if not emitted_event: self._telemetry.checkout_failed( diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py index 1304921781..ce1fd7aac2 100644 --- a/pymongo/synchronous/pool.py +++ b/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,50 @@ 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 + requests_incremented = False + active_sockets_incremented = False emitted_event = False is_new_conn = False + try: + slot_acquired = False with self.lock: - self.active_sockets += 1 - incremented = True + self.operation_count += 1 + op_count_incremented = True + self._raise_if_not_ready(checkout_started_time, emit_event=True) + if self.requests < self.max_pool_size: + # Fast path: a slot is immediately available, so do all + # of the counter bookkeeping in this single critical + # section instead of taking the lock multiple times. + self.requests += 1 + requests_incremented = True + self.active_sockets += 1 + active_sockets_incremented = True + slot_acquired = True + + if not slot_acquired: + # Slow path: no slot was free. Fall back to waiting on the + # requests semaphore, exactly as before. + 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 + requests_incremented = True + + with self.lock: + 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 +1099,16 @@ 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 requests_incremented or active_sockets_incremented: + with self.size_cond: + if requests_incremented: + self.requests -= 1 + if active_sockets_incremented: + self.active_sockets -= 1 + self.size_cond.notify() + if op_count_incremented: + with self.lock: + self.operation_count -= 1 if not emitted_event: self._telemetry.checkout_failed( From e79d6a77037ad3f5712f14b332659135e07f2183 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Thu, 30 Jul 2026 06:40:42 -0500 Subject: [PATCH 03/25] PYTHON-5898 Add regression test for operation_count leak on checkout failure Adds test_wait_queue_timeout_does_not_leak_operation_count to test/test_pooling.py and test/asynchronous/test_pooling.py, covering the "Merge pool checkout lock acquisitions on the fast path" fix: a checkout that fails while waiting for a free pool slot (wait queue timeout) must not leave Pool.operation_count permanently incremented for the failed attempt. Each test was verified to fail against the pre-fix code (operation_count left incremented instead of returning to its prior value), and passes after the fix. --- test/asynchronous/test_pooling.py | 18 ++++++++++++++++++ test/test_pooling.py | 18 ++++++++++++++++++ 2 files changed, 36 insertions(+) diff --git a/test/asynchronous/test_pooling.py b/test/asynchronous/test_pooling.py index 063f5f06ec..e37bab5648 100644 --- a/test/asynchronous/test_pooling.py +++ b/test/asynchronous/test_pooling.py @@ -399,6 +399,24 @@ async def test_wait_queue_timeout(self): f"Waited {duration:.2f} seconds for a socket, expected {wait_queue_timeout:f}", ) + async def test_wait_queue_timeout_does_not_leak_operation_count(self): + # Regression test: a checkout that fails while waiting for a pool + # slot must not leave operation_count permanently incremented. + wait_queue_timeout = 1 # Seconds + pool = await self.create_pool(max_pool_size=1, wait_queue_timeout=wait_queue_timeout) + self.addAsyncCleanup(pool.close) + + async with pool.checkout(): + self.assertEqual(pool.operation_count, 1) + with self.assertRaises(ConnectionFailure): + async with pool.checkout(): + pass + # The failed second checkout must not have left operation_count + # incremented for its own (failed) attempt. + self.assertEqual(pool.operation_count, 1) + + self.assertEqual(pool.operation_count, 0) + async def test_no_wait_queue_timeout(self): # Verify get_socket() with no wait_queue_timeout blocks forever. pool = await self.create_pool(max_pool_size=1) diff --git a/test/test_pooling.py b/test/test_pooling.py index 64146a0e13..a23e95b081 100644 --- a/test/test_pooling.py +++ b/test/test_pooling.py @@ -399,6 +399,24 @@ def test_wait_queue_timeout(self): f"Waited {duration:.2f} seconds for a socket, expected {wait_queue_timeout:f}", ) + def test_wait_queue_timeout_does_not_leak_operation_count(self): + # Regression test: a checkout that fails while waiting for a pool + # slot must not leave operation_count permanently incremented. + wait_queue_timeout = 1 # Seconds + pool = self.create_pool(max_pool_size=1, wait_queue_timeout=wait_queue_timeout) + self.addCleanup(pool.close) + + with pool.checkout(): + self.assertEqual(pool.operation_count, 1) + with self.assertRaises(ConnectionFailure): + with pool.checkout(): + pass + # The failed second checkout must not have left operation_count + # incremented for its own (failed) attempt. + self.assertEqual(pool.operation_count, 1) + + self.assertEqual(pool.operation_count, 0) + def test_no_wait_queue_timeout(self): # Verify get_socket() with no wait_queue_timeout blocks forever. pool = self.create_pool(max_pool_size=1) From adac148028c2aba93e3c89bdf2ac0599912c4738 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Thu, 30 Jul 2026 06:40:42 -0500 Subject: [PATCH 04/25] PYTHON-5898 Fix duplicate ConnectionCheckOutFailed event in _get_conn The lock-merge refactor moved the pool-not-ready and wait-queue-timeout checks inside the outer try/except (needed for the operation_count leak fix), but didn't account for the outer except also running for those exceptions now. That caused a second, incorrect ConnectionCheckOutFailed event (reason connectionError) to follow the correct one whenever a checkout failed because the pool was paused/closed or because the wait queue timed out. Fix: the three _raise_if_not_ready call sites in the fast/slow path now pass emit_event=False and let the outer except emit once, and the wait-queue-timeout branch now sets emitted_event=True before raising, mirroring the existing max_connecting_cond pattern. Also removed the redundant slot_acquired flag (identical to requests_incremented at that point) and folded operation_count's decrement into the size_cond critical section in the except block, gating notify() on whether a slot was actually released. Extends the operation_count leak regression test to also assert on requests/active_sockets and on the emitted event count/reason, and adds a second regression test covering the pool-paused checkout-failure path, both of which fail against the pre-fix code with the double emission this commit fixes. --- pymongo/asynchronous/pool.py | 25 ++++++++------ pymongo/synchronous/pool.py | 25 ++++++++------ test/asynchronous/test_pooling.py | 57 ++++++++++++++++++++++++++++--- test/test_pooling.py | 57 ++++++++++++++++++++++++++++--- 4 files changed, 134 insertions(+), 30 deletions(-) diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py index 3e3296b974..b2a518e658 100644 --- a/pymongo/asynchronous/pool.py +++ b/pymongo/asynchronous/pool.py @@ -1023,11 +1023,14 @@ async def _get_conn( is_new_conn = False try: - slot_acquired = False async with self.lock: self.operation_count += 1 op_count_incremented = True - self._raise_if_not_ready(checkout_started_time, emit_event=True) + # Emission is deferred to the outer `except` below (which + # runs for any exception raised inside this `try`, unlike + # the pre-refactor code where this check lived outside the + # `try`); emitting here too would double-emit the event. + self._raise_if_not_ready(checkout_started_time, emit_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 single critical @@ -1036,13 +1039,12 @@ async def _get_conn( requests_incremented = True self.active_sockets += 1 active_sockets_incremented = True - slot_acquired = True - if not slot_acquired: + if not requests_incremented: # Slow path: no slot was free. Fall back to waiting on the # requests semaphore, exactly as before. async with self.size_cond: - self._raise_if_not_ready(checkout_started_time, emit_event=True) + self._raise_if_not_ready(checkout_started_time, emit_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): @@ -1050,8 +1052,9 @@ async def _get_conn( # 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) - self._raise_if_not_ready(checkout_started_time, emit_event=True) + self._raise_if_not_ready(checkout_started_time, emit_event=False) self.requests += 1 requests_incremented = True @@ -1103,16 +1106,16 @@ async def _get_conn( if conn: # We checked out a socket but authentication failed. await conn.close_conn(ConnectionClosedReason.ERROR) - if requests_incremented or active_sockets_incremented: + if requests_incremented or active_sockets_incremented or op_count_incremented: async with self.size_cond: if requests_incremented: self.requests -= 1 if active_sockets_incremented: self.active_sockets -= 1 - self.size_cond.notify() - if op_count_incremented: - async with self.lock: - self.operation_count -= 1 + if op_count_incremented: + self.operation_count -= 1 + if requests_incremented or active_sockets_incremented: + self.size_cond.notify() if not emitted_event: self._telemetry.checkout_failed( diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py index ce1fd7aac2..4ffec426a8 100644 --- a/pymongo/synchronous/pool.py +++ b/pymongo/synchronous/pool.py @@ -1019,11 +1019,14 @@ def _get_conn( is_new_conn = False try: - slot_acquired = False with self.lock: self.operation_count += 1 op_count_incremented = True - self._raise_if_not_ready(checkout_started_time, emit_event=True) + # Emission is deferred to the outer `except` below (which + # runs for any exception raised inside this `try`, unlike + # the pre-refactor code where this check lived outside the + # `try`); emitting here too would double-emit the event. + self._raise_if_not_ready(checkout_started_time, emit_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 single critical @@ -1032,13 +1035,12 @@ def _get_conn( requests_incremented = True self.active_sockets += 1 active_sockets_incremented = True - slot_acquired = True - if not slot_acquired: + if not requests_incremented: # Slow path: no slot was free. Fall back to waiting on the # requests semaphore, exactly as before. with self.size_cond: - self._raise_if_not_ready(checkout_started_time, emit_event=True) + self._raise_if_not_ready(checkout_started_time, emit_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): @@ -1046,8 +1048,9 @@ def _get_conn( # 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) - self._raise_if_not_ready(checkout_started_time, emit_event=True) + self._raise_if_not_ready(checkout_started_time, emit_event=False) self.requests += 1 requests_incremented = True @@ -1099,16 +1102,16 @@ def _get_conn( if conn: # We checked out a socket but authentication failed. conn.close_conn(ConnectionClosedReason.ERROR) - if requests_incremented or active_sockets_incremented: + if requests_incremented or active_sockets_incremented or op_count_incremented: with self.size_cond: if requests_incremented: self.requests -= 1 if active_sockets_incremented: self.active_sockets -= 1 - self.size_cond.notify() - if op_count_incremented: - with self.lock: - self.operation_count -= 1 + if op_count_incremented: + self.operation_count -= 1 + if requests_incremented or active_sockets_incremented: + self.size_cond.notify() if not emitted_event: self._telemetry.checkout_failed( diff --git a/test/asynchronous/test_pooling.py b/test/asynchronous/test_pooling.py index e37bab5648..b18f5e0fcc 100644 --- a/test/asynchronous/test_pooling.py +++ b/test/asynchronous/test_pooling.py @@ -33,7 +33,11 @@ from pymongo.errors import AutoReconnect, ConnectionFailure, DuplicateKeyError from pymongo.hello import HelloCompat from pymongo.lock import _async_create_lock -from pymongo.monitoring import _EventListeners +from pymongo.monitoring import ( + ConnectionCheckOutFailedEvent, + ConnectionCheckOutFailedReason, + _EventListeners, +) from test.asynchronous.utils import async_get_pool, async_joinall, flaky sys.path[0:0] = [""] @@ -401,21 +405,66 @@ async def test_wait_queue_timeout(self): async def test_wait_queue_timeout_does_not_leak_operation_count(self): # Regression test: a checkout that fails while waiting for a pool - # slot must not leave operation_count permanently incremented. + # slot must not leave operation_count, requests, or active_sockets + # permanently incremented, and must emit exactly one + # ConnectionCheckOutFailedEvent with reason TIMEOUT (not a second, + # bogus CONN_ERROR event from the outer except block). wait_queue_timeout = 1 # Seconds - pool = await self.create_pool(max_pool_size=1, wait_queue_timeout=wait_queue_timeout) + listener = CMAPListener() + pool = await self.create_pool( + max_pool_size=1, + wait_queue_timeout=wait_queue_timeout, + event_listeners=_EventListeners([listener]), + ) self.addAsyncCleanup(pool.close) async with pool.checkout(): self.assertEqual(pool.operation_count, 1) + self.assertEqual(pool.requests, 1) + self.assertEqual(pool.active_sockets, 1) + listener.reset() with self.assertRaises(ConnectionFailure): async with pool.checkout(): pass - # The failed second checkout must not have left operation_count + # The failed second checkout must not have left any counter # incremented for its own (failed) attempt. self.assertEqual(pool.operation_count, 1) + self.assertEqual(pool.requests, 1) + self.assertEqual(pool.active_sockets, 1) + + failed_events = listener.events_by_type(ConnectionCheckOutFailedEvent) + self.assertEqual(len(failed_events), 1, [e.reason for e in failed_events]) + self.assertEqual(failed_events[0].reason, ConnectionCheckOutFailedReason.TIMEOUT) self.assertEqual(pool.operation_count, 0) + self.assertEqual(pool.requests, 0) + self.assertEqual(pool.active_sockets, 0) + + async def test_paused_pool_checkout_failure_does_not_leak_or_double_emit(self): + # Regression test: a checkout that fails because the pool is paused + # (not ready) must not leave operation_count, requests, or + # active_sockets permanently incremented, and must emit exactly one + # ConnectionCheckOutFailedEvent (not a second, bogus one from the + # outer except block). With no outstanding checkouts, a slot is + # immediately available, so this exercises the fast path's + # _raise_if_not_ready check. + listener = CMAPListener() + pool = await self.create_pool(max_pool_size=1, event_listeners=_EventListeners([listener])) + self.addAsyncCleanup(pool.close) + + await pool.reset() # Pause the pool. + listener.reset() + with self.assertRaises(AutoReconnect): + async with pool.checkout(): + pass + + self.assertEqual(pool.operation_count, 0) + self.assertEqual(pool.requests, 0) + self.assertEqual(pool.active_sockets, 0) + + failed_events = listener.events_by_type(ConnectionCheckOutFailedEvent) + self.assertEqual(len(failed_events), 1, [e.reason for e in failed_events]) + self.assertEqual(failed_events[0].reason, ConnectionCheckOutFailedReason.CONN_ERROR) async def test_no_wait_queue_timeout(self): # Verify get_socket() with no wait_queue_timeout blocks forever. diff --git a/test/test_pooling.py b/test/test_pooling.py index a23e95b081..54357d45bd 100644 --- a/test/test_pooling.py +++ b/test/test_pooling.py @@ -33,7 +33,11 @@ from pymongo.errors import AutoReconnect, ConnectionFailure, DuplicateKeyError from pymongo.hello import HelloCompat from pymongo.lock import _create_lock -from pymongo.monitoring import _EventListeners +from pymongo.monitoring import ( + ConnectionCheckOutFailedEvent, + ConnectionCheckOutFailedReason, + _EventListeners, +) from test.utils import flaky, get_pool, joinall sys.path[0:0] = [""] @@ -401,21 +405,66 @@ def test_wait_queue_timeout(self): def test_wait_queue_timeout_does_not_leak_operation_count(self): # Regression test: a checkout that fails while waiting for a pool - # slot must not leave operation_count permanently incremented. + # slot must not leave operation_count, requests, or active_sockets + # permanently incremented, and must emit exactly one + # ConnectionCheckOutFailedEvent with reason TIMEOUT (not a second, + # bogus CONN_ERROR event from the outer except block). wait_queue_timeout = 1 # Seconds - pool = self.create_pool(max_pool_size=1, wait_queue_timeout=wait_queue_timeout) + listener = CMAPListener() + pool = self.create_pool( + max_pool_size=1, + wait_queue_timeout=wait_queue_timeout, + event_listeners=_EventListeners([listener]), + ) self.addCleanup(pool.close) with pool.checkout(): self.assertEqual(pool.operation_count, 1) + self.assertEqual(pool.requests, 1) + self.assertEqual(pool.active_sockets, 1) + listener.reset() with self.assertRaises(ConnectionFailure): with pool.checkout(): pass - # The failed second checkout must not have left operation_count + # The failed second checkout must not have left any counter # incremented for its own (failed) attempt. self.assertEqual(pool.operation_count, 1) + self.assertEqual(pool.requests, 1) + self.assertEqual(pool.active_sockets, 1) + + failed_events = listener.events_by_type(ConnectionCheckOutFailedEvent) + self.assertEqual(len(failed_events), 1, [e.reason for e in failed_events]) + self.assertEqual(failed_events[0].reason, ConnectionCheckOutFailedReason.TIMEOUT) self.assertEqual(pool.operation_count, 0) + self.assertEqual(pool.requests, 0) + self.assertEqual(pool.active_sockets, 0) + + def test_paused_pool_checkout_failure_does_not_leak_or_double_emit(self): + # Regression test: a checkout that fails because the pool is paused + # (not ready) must not leave operation_count, requests, or + # active_sockets permanently incremented, and must emit exactly one + # ConnectionCheckOutFailedEvent (not a second, bogus one from the + # outer except block). With no outstanding checkouts, a slot is + # immediately available, so this exercises the fast path's + # _raise_if_not_ready check. + listener = CMAPListener() + pool = self.create_pool(max_pool_size=1, event_listeners=_EventListeners([listener])) + self.addCleanup(pool.close) + + pool.reset() # Pause the pool. + listener.reset() + with self.assertRaises(AutoReconnect): + with pool.checkout(): + pass + + self.assertEqual(pool.operation_count, 0) + self.assertEqual(pool.requests, 0) + self.assertEqual(pool.active_sockets, 0) + + failed_events = listener.events_by_type(ConnectionCheckOutFailedEvent) + self.assertEqual(len(failed_events), 1, [e.reason for e in failed_events]) + self.assertEqual(failed_events[0].reason, ConnectionCheckOutFailedReason.CONN_ERROR) def test_no_wait_queue_timeout(self): # Verify get_socket() with no wait_queue_timeout blocks forever. From 930219fe2d82a00c379632d5a77499ae5a66146e Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Thu, 30 Jul 2026 08:36:01 -0500 Subject: [PATCH 05/25] PYTHON-5898 Correct stale-lock wording and add get_primary regression test The Topology reader/writer lock explored on this branch was benchmarked ~30% slower and fully reverted, but three places still described apply_selector()/_filter_servers() as running under a "shared read lock" that no longer exists. Rewrite those docstrings/comments around the actual, still-true justification: TopologyDescription is a shared, immutable snapshot, so caching a per-call filtered list on it leaves the public candidate_servers property stale and can make Topology.get_primary() raise IndexError. Also renames TestTopologyDescriptionConcurrency to TestTopologyDescriptionImmutability to match, adds a regression test for the concrete get_primary()/IndexError symptom, adds a timeout to the existing racy test's barrier.wait() so a dead worker can't hang CI, and documents all three fixes in the changelog. --- doc/changelog.rst | 13 +++++++++++ pymongo/topology_description.py | 11 ++++++--- test/test_topology.py | 41 ++++++++++++++++++++++++++++----- 3 files changed, 56 insertions(+), 9 deletions(-) diff --git a/doc/changelog.rst b/doc/changelog.rst index cda288c575..5342d2c694 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -27,6 +27,19 @@ 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 :attr:`~pymongo.topology_description.TopologyDescription.candidate_servers` + to once again return all known servers, matching its behavior before + PyMongo 4.16.0. Deprioritization is now applied per server-selection call + instead of being cached on the (shared, immutable) + :class:`~pymongo.topology_description.TopologyDescription`. This fixes + ``client.primary`` raising ``IndexError``, and ``client.secondaries`` and + ``client.arbiters`` returning stale results, after a retryable operation + deprioritized the primary. +- Fixed a leak where every failed connection checkout permanently + incremented a pool's ``operation_count``, biasing server selection among + mongoses toward the affected server. +- Reduced the number of lock acquisitions on the connection checkout fast + path. Changes in Version 4.17.0 (2026/04/20) -------------------------------------- diff --git a/pymongo/topology_description.py b/pymongo/topology_description.py index 352e8a7647..90fd40c8b0 100644 --- a/pymongo/topology_description.py +++ b/pymongo/topology_description.py @@ -290,9 +290,14 @@ def _filter_servers( If every known server is deprioritized, all known servers are returned so that selection can still make progress. - This method must not mutate ``self``: server selection runs it while - holding only a shared read lock, so concurrent callers with different - ``deprioritized_servers`` would otherwise clobber each other's results. + This method must not mutate ``self``: TopologyDescription is a shared, + publicly-exposed immutable snapshot, and deprioritization is specific + to a single selection call. Caching the filtered list on ``self`` + would leave the public :attr:`candidate_servers` property stale after + any selection that deprioritized servers, which is what used to make + :meth:`~pymongo.synchronous.topology.Topology.get_primary` (and + ``client.primary``) raise ``IndexError`` after a retryable operation + deprioritized the primary. :param deprioritized_servers: servers to exclude, or None. """ diff --git a/test/test_topology.py b/test/test_topology.py index 7ddbffed0d..fc956bd142 100644 --- a/test/test_topology.py +++ b/test/test_topology.py @@ -938,10 +938,12 @@ def create_mock_replica_set_topology(hosts=("a", "b", "c")): return t -class TestTopologyDescriptionConcurrency(TopologyTest): - """apply_selector() runs under a shared read lock (PYTHON-5898), so it must - not mutate the TopologyDescription: concurrent callers passing different - deprioritized_servers would otherwise clobber each other's candidate lists. +class TestTopologyDescriptionImmutability(TopologyTest): + """TopologyDescription is a shared, publicly-exposed immutable snapshot + (PYTHON-5898), so apply_selector() must not mutate it: caching a per-call + filtered candidate list on the description would leave the public + candidate_servers property permanently stale after any selection that + deprioritized servers. """ def test_concurrent_apply_selector_with_deprioritized_servers(self): @@ -963,7 +965,7 @@ def test_concurrent_apply_selector_with_deprioritized_servers(self): barrier = threading.Barrier(6) def deprioritizing_worker(): - barrier.wait() + barrier.wait(timeout=30) for _ in range(iterations): # A retryable write retrying away from the primary must never # be handed the deprioritized primary back. @@ -974,7 +976,7 @@ def deprioritizing_worker(): errors.append(f"deprioritized primary was selected: {sds}") def plain_worker(): - barrier.wait() + barrier.wait(timeout=30) for _ in range(iterations): # An ordinary operation must always find the healthy primary. sds = description.apply_selector(Primary()) @@ -990,6 +992,33 @@ def plain_worker(): self.assertEqual([], errors[:5], f"{len(errors)} racy selections") + def test_get_primary_after_deprioritized_selection(self): + # Regression test for PYTHON-5898 / PYTHON-5662: apply_selector() + # used to cache its filtered candidate list on the (shared, publicly + # exposed) TopologyDescription, so candidate_servers stayed stale + # after any selection that deprioritized a server. get_primary() + # (and client.primary) build their selection from candidate_servers, + # so a stale, primary-less candidate list made + # writable_server_selector(...)[0] raise IndexError even though the + # primary was still known and healthy. + t = create_mock_replica_set_topology() + self.addCleanup(t.close) + description = t.description + primary_sd = description.server_descriptions()[("a", 27017)] + self.assertEqual(SERVER_TYPE.RSPrimary, primary_sd.server_type) + + self.assertEqual(("a", 27017), t.get_primary()) + + # Simulate a retryable operation deprioritizing the primary for one + # selection call. + description.apply_selector( + ReadPreference.PRIMARY_PREFERRED, deprioritized_servers=[primary_sd] + ) + + # get_primary() must still find the primary afterwards; it must not + # raise IndexError because of a stale, primary-less candidate list. + self.assertEqual(("a", 27017), t.get_primary()) + def test_apply_selector_does_not_mutate_description(self): t = create_mock_replica_set_topology() self.addCleanup(t.close) From 0bf12ae0cb269190256080ad65df2654a471e882 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Thu, 30 Jul 2026 11:13:23 -0500 Subject: [PATCH 06/25] PYTHON-5898 Remove the candidate_servers property PYTHON-5662 added candidate_servers to plumb its deprioritization-filtered list into Selection.from_topology_description(), which took only a TopologyDescription. The list is now passed explicitly, so the property has no remaining consumer. Its documented contract ("servers excluding deprioritized servers") could not be satisfied anyway: deprioritization is an argument to a single selection call, so there is no single correct answer for a description that serves many calls. Selection now defaults to known_servers. --- doc/changelog.rst | 9 ++------- pymongo/server_selectors.py | 8 ++++---- pymongo/topology_description.py | 10 ---------- test/test_topology.py | 11 ++++++++--- 4 files changed, 14 insertions(+), 24 deletions(-) diff --git a/doc/changelog.rst b/doc/changelog.rst index 5342d2c694..03b6a6ae48 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -27,13 +27,8 @@ 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 :attr:`~pymongo.topology_description.TopologyDescription.candidate_servers` - to once again return all known servers, matching its behavior before - PyMongo 4.16.0. Deprioritization is now applied per server-selection call - instead of being cached on the (shared, immutable) - :class:`~pymongo.topology_description.TopologyDescription`. This fixes - ``client.primary`` raising ``IndexError``, and ``client.secondaries`` and - ``client.arbiters`` returning stale results, after a retryable operation +- Fixed ``client.primary`` raising ``IndexError``, and ``client.secondaries`` + and ``client.arbiters`` returning stale results, after a retryable operation deprioritized the primary. - Fixed a leak where every failed connection checkout permanently incremented a pool's ``operation_count``, biasing server selection among diff --git a/pymongo/server_selectors.py b/pymongo/server_selectors.py index f1f0c81272..74c10ff431 100644 --- a/pymongo/server_selectors.py +++ b/pymongo/server_selectors.py @@ -44,12 +44,12 @@ def from_topology_description( :param topology_description: the TopologyDescription to select from. :param candidate_servers: the servers eligible for selection. Defaults - to ``topology_description.candidate_servers``. Server selection - passes its own (per-call, deprioritization-filtered) list here so - that no state has to be cached on the TopologyDescription. + 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.candidate_servers + candidate_servers = topology_description.known_servers primary = None for sd in candidate_servers: if sd.server_type == SERVER_TYPE.RSPrimary: diff --git a/pymongo/topology_description.py b/pymongo/topology_description.py index 90fd40c8b0..f98c0e6ef7 100644 --- a/pymongo/topology_description.py +++ b/pymongo/topology_description.py @@ -237,16 +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 eligible for selection when nothing is deprioritized. - - Deprioritization is per server-selection call, so it is applied by - :meth:`apply_selector` (via :meth:`_filter_servers`) and deliberately - not cached on the (immutable) TopologyDescription. - """ - return self.known_servers - @property def common_wire_version(self) -> Optional[int]: """Minimum of all servers' max wire versions, or None.""" diff --git a/test/test_topology.py b/test/test_topology.py index fc956bd142..a53cac371e 100644 --- a/test/test_topology.py +++ b/test/test_topology.py @@ -1025,12 +1025,17 @@ def test_apply_selector_does_not_mutate_description(self): description = t.description primary_sd = description.server_descriptions()[("a", 27017)] - before = list(description.candidate_servers) + # A selection that deprioritizes the primary must not leave that + # filtering behind on the description: a later selection with nothing + # deprioritized has to see the primary again. + before = list(description.known_servers) description.apply_selector( ReadPreference.PRIMARY_PREFERRED, deprioritized_servers=[primary_sd] ) - self.assertEqual(before, description.candidate_servers) - self.assertIn(primary_sd, description.candidate_servers) + self.assertEqual(before, description.known_servers) + + after = description.apply_selector(ReadPreference.PRIMARY_PREFERRED) + self.assertIn(primary_sd, after) if __name__ == "__main__": From 6aff28f5f7106df5ee5834a73f83e65a81ae7c2a Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Thu, 30 Jul 2026 15:07:46 -0500 Subject: [PATCH 07/25] PYTHON-5898 Drop a test that could not fail and fix stale wording test_apply_selector_does_not_mutate_description asserted on the candidate_servers property; once that property was removed the assertion reduced to a tautology and passed against the unfixed code, so it was carrying no weight. The two remaining tests in the class both fail against the unfixed code and cover the same ground. Also drop the remaining references to the removed property, and stop describing TopologyDescription as immutable while explaining that it used to be mutated. --- pymongo/topology_description.py | 10 ++++----- test/test_topology.py | 40 +++++++++------------------------ 2 files changed, 15 insertions(+), 35 deletions(-) diff --git a/pymongo/topology_description.py b/pymongo/topology_description.py index f98c0e6ef7..ea759b633e 100644 --- a/pymongo/topology_description.py +++ b/pymongo/topology_description.py @@ -280,11 +280,11 @@ def _filter_servers( If every known server is deprioritized, all known servers are returned so that selection can still make progress. - This method must not mutate ``self``: TopologyDescription is a shared, - publicly-exposed immutable snapshot, and deprioritization is specific - to a single selection call. Caching the filtered list on ``self`` - would leave the public :attr:`candidate_servers` property stale after - any selection that deprioritized servers, which is what used to make + This method must not mutate ``self``: a TopologyDescription is shared + by every concurrent selection call and is replaced, not edited, when + the topology changes. Deprioritization is specific to a single call, + so caching the filtered list on ``self`` outlived the call that + produced it, which is what used to make :meth:`~pymongo.synchronous.topology.Topology.get_primary` (and ``client.primary``) raise ``IndexError`` after a retryable operation deprioritized the primary. diff --git a/test/test_topology.py b/test/test_topology.py index a53cac371e..ed2c454d89 100644 --- a/test/test_topology.py +++ b/test/test_topology.py @@ -939,11 +939,10 @@ def create_mock_replica_set_topology(hosts=("a", "b", "c")): class TestTopologyDescriptionImmutability(TopologyTest): - """TopologyDescription is a shared, publicly-exposed immutable snapshot - (PYTHON-5898), so apply_selector() must not mutate it: caching a per-call - filtered candidate list on the description would leave the public - candidate_servers property permanently stale after any selection that - deprioritized servers. + """A TopologyDescription is shared by every concurrent selection call and + is replaced, not edited, when the topology changes (PYTHON-5898). So + apply_selector() must not mutate it: caching a per-call filtered candidate + list on the description outlived the call that produced it. """ def test_concurrent_apply_selector_with_deprioritized_servers(self): @@ -994,13 +993,12 @@ def plain_worker(): def test_get_primary_after_deprioritized_selection(self): # Regression test for PYTHON-5898 / PYTHON-5662: apply_selector() - # used to cache its filtered candidate list on the (shared, publicly - # exposed) TopologyDescription, so candidate_servers stayed stale - # after any selection that deprioritized a server. get_primary() - # (and client.primary) build their selection from candidate_servers, - # so a stale, primary-less candidate list made - # writable_server_selector(...)[0] raise IndexError even though the - # primary was still known and healthy. + # used to cache its filtered candidate list on the shared + # TopologyDescription, so the filtering outlived the call that + # produced it. get_primary() (and client.primary) build their + # selection straight from the description, so a stale, primary-less + # candidate list made writable_server_selector(...)[0] raise + # IndexError even though the primary was still known and healthy. t = create_mock_replica_set_topology() self.addCleanup(t.close) description = t.description @@ -1019,24 +1017,6 @@ def test_get_primary_after_deprioritized_selection(self): # raise IndexError because of a stale, primary-less candidate list. self.assertEqual(("a", 27017), t.get_primary()) - def test_apply_selector_does_not_mutate_description(self): - t = create_mock_replica_set_topology() - self.addCleanup(t.close) - description = t.description - primary_sd = description.server_descriptions()[("a", 27017)] - - # A selection that deprioritizes the primary must not leave that - # filtering behind on the description: a later selection with nothing - # deprioritized has to see the primary again. - before = list(description.known_servers) - description.apply_selector( - ReadPreference.PRIMARY_PREFERRED, deprioritized_servers=[primary_sd] - ) - self.assertEqual(before, description.known_servers) - - after = description.apply_selector(ReadPreference.PRIMARY_PREFERRED) - self.assertIn(primary_sd, after) - if __name__ == "__main__": unittest.main() From 0ea114f7d7b51f4b296c5b1ebb6475d4aa205783 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Thu, 30 Jul 2026 15:15:14 -0500 Subject: [PATCH 08/25] PYTHON-5898 Note the candidate_servers removal in the changelog --- doc/changelog.rst | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/doc/changelog.rst b/doc/changelog.rst index 03b6a6ae48..c06bed7039 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -30,6 +30,11 @@ PyMongo 4.18 brings a number of changes including: - Fixed ``client.primary`` raising ``IndexError``, and ``client.secondaries`` and ``client.arbiters`` returning stale results, after a retryable operation deprioritized the primary. +- Removed ``TopologyDescription.candidate_servers``, added in PyMongo 4.16.0. + Its value depended on which server-selection call happened to run last, so it + could not be relied on. Use + :meth:`~pymongo.topology_description.TopologyDescription.server_descriptions` + instead. - Fixed a leak where every failed connection checkout permanently incremented a pool's ``operation_count``, biasing server selection among mongoses toward the affected server. From ebb92f252624358d2e675fe215552ab0955adafd Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Thu, 30 Jul 2026 20:58:21 -0500 Subject: [PATCH 09/25] PYTHON-5898 Describe required behavior in comments, not prior behavior --- pymongo/topology_description.py | 13 ++++++------- test/asynchronous/test_pooling.py | 20 ++++++++------------ test/test_pooling.py | 20 ++++++++------------ test/test_topology.py | 18 ++++++------------ 4 files changed, 28 insertions(+), 43 deletions(-) diff --git a/pymongo/topology_description.py b/pymongo/topology_description.py index ea759b633e..0dd815f48f 100644 --- a/pymongo/topology_description.py +++ b/pymongo/topology_description.py @@ -280,14 +280,13 @@ def _filter_servers( If every known server is deprioritized, all known servers are returned so that selection can still make progress. - This method must not mutate ``self``: a TopologyDescription is shared + This method must not mutate ``self``. A TopologyDescription is shared by every concurrent selection call and is replaced, not edited, when - the topology changes. Deprioritization is specific to a single call, - so caching the filtered list on ``self`` outlived the call that - produced it, which is what used to make - :meth:`~pymongo.synchronous.topology.Topology.get_primary` (and - ``client.primary``) raise ``IndexError`` after a retryable operation - deprioritized the primary. + the topology changes, while deprioritization is specific to a single + call. Callers such as + :meth:`~pymongo.synchronous.topology.Topology.get_primary` build their + selection straight from the description, so any filtering left on it + would be applied to selections that never asked for it. :param deprioritized_servers: servers to exclude, or None. """ diff --git a/test/asynchronous/test_pooling.py b/test/asynchronous/test_pooling.py index b18f5e0fcc..9a9b7a7381 100644 --- a/test/asynchronous/test_pooling.py +++ b/test/asynchronous/test_pooling.py @@ -404,11 +404,9 @@ async def test_wait_queue_timeout(self): ) async def test_wait_queue_timeout_does_not_leak_operation_count(self): - # Regression test: a checkout that fails while waiting for a pool - # slot must not leave operation_count, requests, or active_sockets - # permanently incremented, and must emit exactly one - # ConnectionCheckOutFailedEvent with reason TIMEOUT (not a second, - # bogus CONN_ERROR event from the outer except block). + # A checkout that fails while waiting for a pool slot must not leave + # operation_count, requests, or active_sockets incremented, and must + # emit exactly one ConnectionCheckOutFailedEvent, with reason TIMEOUT. wait_queue_timeout = 1 # Seconds listener = CMAPListener() pool = await self.create_pool( @@ -441,13 +439,11 @@ async def test_wait_queue_timeout_does_not_leak_operation_count(self): self.assertEqual(pool.active_sockets, 0) async def test_paused_pool_checkout_failure_does_not_leak_or_double_emit(self): - # Regression test: a checkout that fails because the pool is paused - # (not ready) must not leave operation_count, requests, or - # active_sockets permanently incremented, and must emit exactly one - # ConnectionCheckOutFailedEvent (not a second, bogus one from the - # outer except block). With no outstanding checkouts, a slot is - # immediately available, so this exercises the fast path's - # _raise_if_not_ready check. + # A checkout that fails because the pool is paused must not leave + # operation_count, requests, or active_sockets incremented, and must + # emit exactly one ConnectionCheckOutFailedEvent. With no outstanding + # checkouts a slot is immediately available, so this exercises the + # fast path's readiness check. listener = CMAPListener() pool = await self.create_pool(max_pool_size=1, event_listeners=_EventListeners([listener])) self.addAsyncCleanup(pool.close) diff --git a/test/test_pooling.py b/test/test_pooling.py index 54357d45bd..dd2868bfcd 100644 --- a/test/test_pooling.py +++ b/test/test_pooling.py @@ -404,11 +404,9 @@ def test_wait_queue_timeout(self): ) def test_wait_queue_timeout_does_not_leak_operation_count(self): - # Regression test: a checkout that fails while waiting for a pool - # slot must not leave operation_count, requests, or active_sockets - # permanently incremented, and must emit exactly one - # ConnectionCheckOutFailedEvent with reason TIMEOUT (not a second, - # bogus CONN_ERROR event from the outer except block). + # A checkout that fails while waiting for a pool slot must not leave + # operation_count, requests, or active_sockets incremented, and must + # emit exactly one ConnectionCheckOutFailedEvent, with reason TIMEOUT. wait_queue_timeout = 1 # Seconds listener = CMAPListener() pool = self.create_pool( @@ -441,13 +439,11 @@ def test_wait_queue_timeout_does_not_leak_operation_count(self): self.assertEqual(pool.active_sockets, 0) def test_paused_pool_checkout_failure_does_not_leak_or_double_emit(self): - # Regression test: a checkout that fails because the pool is paused - # (not ready) must not leave operation_count, requests, or - # active_sockets permanently incremented, and must emit exactly one - # ConnectionCheckOutFailedEvent (not a second, bogus one from the - # outer except block). With no outstanding checkouts, a slot is - # immediately available, so this exercises the fast path's - # _raise_if_not_ready check. + # A checkout that fails because the pool is paused must not leave + # operation_count, requests, or active_sockets incremented, and must + # emit exactly one ConnectionCheckOutFailedEvent. With no outstanding + # checkouts a slot is immediately available, so this exercises the + # fast path's readiness check. listener = CMAPListener() pool = self.create_pool(max_pool_size=1, event_listeners=_EventListeners([listener])) self.addCleanup(pool.close) diff --git a/test/test_topology.py b/test/test_topology.py index ed2c454d89..bb5b1c24b9 100644 --- a/test/test_topology.py +++ b/test/test_topology.py @@ -940,9 +940,8 @@ def create_mock_replica_set_topology(hosts=("a", "b", "c")): class TestTopologyDescriptionImmutability(TopologyTest): """A TopologyDescription is shared by every concurrent selection call and - is replaced, not edited, when the topology changes (PYTHON-5898). So - apply_selector() must not mutate it: caching a per-call filtered candidate - list on the description outlived the call that produced it. + is replaced, not edited, when the topology changes (PYTHON-5898), so + apply_selector() must not mutate it. """ def test_concurrent_apply_selector_with_deprioritized_servers(self): @@ -992,13 +991,9 @@ def plain_worker(): self.assertEqual([], errors[:5], f"{len(errors)} racy selections") def test_get_primary_after_deprioritized_selection(self): - # Regression test for PYTHON-5898 / PYTHON-5662: apply_selector() - # used to cache its filtered candidate list on the shared - # TopologyDescription, so the filtering outlived the call that - # produced it. get_primary() (and client.primary) build their - # selection straight from the description, so a stale, primary-less - # candidate list made writable_server_selector(...)[0] raise - # IndexError even though the primary was still known and healthy. + # get_primary() (and client.primary) build their selection straight + # from the description, so a selection that deprioritizes the primary + # must not stop a later get_primary() from finding it. t = create_mock_replica_set_topology() self.addCleanup(t.close) description = t.description @@ -1013,8 +1008,7 @@ def test_get_primary_after_deprioritized_selection(self): ReadPreference.PRIMARY_PREFERRED, deprioritized_servers=[primary_sd] ) - # get_primary() must still find the primary afterwards; it must not - # raise IndexError because of a stale, primary-less candidate list. + # get_primary() must still find the primary afterwards. self.assertEqual(("a", 27017), t.get_primary()) From 1d36c3666ed621e49e245f52dd6e2249cdaeebb8 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 31 Jul 2026 05:44:20 -0500 Subject: [PATCH 10/25] PYTHON-5898 Drop remaining references to prior behavior in pool comments --- pymongo/asynchronous/pool.py | 14 ++++++-------- pymongo/synchronous/pool.py | 14 ++++++-------- 2 files changed, 12 insertions(+), 16 deletions(-) diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py index b2a518e658..71e84614fb 100644 --- a/pymongo/asynchronous/pool.py +++ b/pymongo/asynchronous/pool.py @@ -1026,23 +1026,21 @@ async def _get_conn( async with self.lock: self.operation_count += 1 op_count_incremented = True - # Emission is deferred to the outer `except` below (which - # runs for any exception raised inside this `try`, unlike - # the pre-refactor code where this check lived outside the - # `try`); emitting here too would double-emit the event. + # The outer `except` below emits for any exception raised in + # this `try`, so emitting here too would double-emit. self._raise_if_not_ready(checkout_started_time, emit_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 single critical - # section instead of taking the lock multiple times. + # 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. Fall back to waiting on the - # requests semaphore, exactly as before. + # Slow path: no slot was free, so wait on the requests + # semaphore for one to be released. async with self.size_cond: self._raise_if_not_ready(checkout_started_time, emit_event=False) while not (self.requests < self.max_pool_size): diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py index 4ffec426a8..df9d1bf08c 100644 --- a/pymongo/synchronous/pool.py +++ b/pymongo/synchronous/pool.py @@ -1022,23 +1022,21 @@ def _get_conn( with self.lock: self.operation_count += 1 op_count_incremented = True - # Emission is deferred to the outer `except` below (which - # runs for any exception raised inside this `try`, unlike - # the pre-refactor code where this check lived outside the - # `try`); emitting here too would double-emit the event. + # The outer `except` below emits for any exception raised in + # this `try`, so emitting here too would double-emit. self._raise_if_not_ready(checkout_started_time, emit_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 single critical - # section instead of taking the lock multiple times. + # 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. Fall back to waiting on the - # requests semaphore, exactly as before. + # Slow path: no slot was free, so wait on the requests + # semaphore for one to be released. with self.size_cond: self._raise_if_not_ready(checkout_started_time, emit_event=False) while not (self.requests < self.max_pool_size): From 576f31402c23191fc1acf6eba0c99a2186b4d10c Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 31 Jul 2026 06:15:54 -0500 Subject: [PATCH 11/25] PYTHON-5898 Comment what the concurrency assertion checks --- test/test_topology.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/test/test_topology.py b/test/test_topology.py index bb5b1c24b9..bad952f74b 100644 --- a/test/test_topology.py +++ b/test/test_topology.py @@ -988,6 +988,9 @@ def plain_worker(): for thread in threads: thread.join() + # Every selection must have returned its own correct result: no + # worker was handed a server that another worker's concurrent call + # had filtered out, or should have filtered out. self.assertEqual([], errors[:5], f"{len(errors)} racy selections") def test_get_primary_after_deprioritized_selection(self): From 78dfe14ba5450ae2ef385f0084677107c9db8342 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 31 Jul 2026 06:21:51 -0500 Subject: [PATCH 12/25] PYTHON-5898 Explain the assertion's failure output --- test/test_topology.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/test/test_topology.py b/test/test_topology.py index bad952f74b..af44e39eab 100644 --- a/test/test_topology.py +++ b/test/test_topology.py @@ -990,7 +990,9 @@ def plain_worker(): # Every selection must have returned its own correct result: no # worker was handed a server that another worker's concurrent call - # had filtered out, or should have filtered out. + # had filtered out, or should have filtered out. On failure the + # assertion output includes the first five error messages, and the + # custom message carries the true total. self.assertEqual([], errors[:5], f"{len(errors)} racy selections") def test_get_primary_after_deprioritized_selection(self): From 59bce8f8dff58a7910b72ac79e9e58b1b86e17ec Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 31 Jul 2026 10:26:39 -0500 Subject: [PATCH 13/25] PYTHON-5898 Emit checkout-failed under the pool lock and fold the slow-path counters Restore upstream's ordering guarantee (PYTHON-3519) by having the three size_cond-region readiness checks publish ConnectionCheckOutFailed while holding the mutex, bracketed so the outer handler does not double-emit. Fold the slow path's active_sockets increment into the size_cond block, collapsing the two counter flags into one. --- pymongo/asynchronous/pool.py | 38 +++++++++++++++++------------------- pymongo/synchronous/pool.py | 38 +++++++++++++++++------------------- 2 files changed, 36 insertions(+), 40 deletions(-) diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py index 71e84614fb..601d32f7d6 100644 --- a/pymongo/asynchronous/pool.py +++ b/pymongo/asynchronous/pool.py @@ -1017,8 +1017,9 @@ async def _get_conn( conn = None op_count_incremented = False - requests_incremented = False - active_sockets_incremented = False + slot_acquired = 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 @@ -1026,23 +1027,24 @@ async def _get_conn( async with self.lock: self.operation_count += 1 op_count_incremented = True - # The outer `except` below emits for any exception raised in - # this `try`, so emitting here too would double-emit. - self._raise_if_not_ready(checkout_started_time, emit_event=False) + 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 + slot_acquired = True - if not requests_incremented: + if not slot_acquired: # Slow path: no slot was free, so wait on the requests # semaphore for one to be released. async with self.size_cond: - self._raise_if_not_ready(checkout_started_time, emit_event=False) + 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): @@ -1052,13 +1054,12 @@ async def _get_conn( self.size_cond.notify() emitted_event = True self._raise_wait_queue_timeout(checkout_started_time) - self._raise_if_not_ready(checkout_started_time, emit_event=False) + emitted_event = True + self._raise_if_not_ready(checkout_started_time, emit_event=True) + emitted_event = False self.requests += 1 - requests_incremented = True - - async with self.lock: self.active_sockets += 1 - active_sockets_incremented = True + slot_acquired = True while conn is None: # CMAP: we MUST wait for either maxConnecting OR for a socket @@ -1104,15 +1105,12 @@ async def _get_conn( if conn: # We checked out a socket but authentication failed. await conn.close_conn(ConnectionClosedReason.ERROR) - if requests_incremented or active_sockets_incremented or op_count_incremented: + if op_count_incremented: async with self.size_cond: - if requests_incremented: + self.operation_count -= 1 + if slot_acquired: self.requests -= 1 - if active_sockets_incremented: self.active_sockets -= 1 - if op_count_incremented: - self.operation_count -= 1 - if requests_incremented or active_sockets_incremented: self.size_cond.notify() if not emitted_event: diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py index df9d1bf08c..118ecd195e 100644 --- a/pymongo/synchronous/pool.py +++ b/pymongo/synchronous/pool.py @@ -1013,8 +1013,9 @@ def _get_conn( conn = None op_count_incremented = False - requests_incremented = False - active_sockets_incremented = False + slot_acquired = 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 @@ -1022,23 +1023,24 @@ def _get_conn( with self.lock: self.operation_count += 1 op_count_incremented = True - # The outer `except` below emits for any exception raised in - # this `try`, so emitting here too would double-emit. - self._raise_if_not_ready(checkout_started_time, emit_event=False) + 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 + slot_acquired = True - if not requests_incremented: + if not slot_acquired: # Slow path: no slot was free, so wait on the requests # semaphore for one to be released. with self.size_cond: - self._raise_if_not_ready(checkout_started_time, emit_event=False) + 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): @@ -1048,13 +1050,12 @@ def _get_conn( self.size_cond.notify() emitted_event = True self._raise_wait_queue_timeout(checkout_started_time) - self._raise_if_not_ready(checkout_started_time, emit_event=False) + emitted_event = True + self._raise_if_not_ready(checkout_started_time, emit_event=True) + emitted_event = False self.requests += 1 - requests_incremented = True - - with self.lock: self.active_sockets += 1 - active_sockets_incremented = True + slot_acquired = True while conn is None: # CMAP: we MUST wait for either maxConnecting OR for a socket @@ -1100,15 +1101,12 @@ def _get_conn( if conn: # We checked out a socket but authentication failed. conn.close_conn(ConnectionClosedReason.ERROR) - if requests_incremented or active_sockets_incremented or op_count_incremented: + if op_count_incremented: with self.size_cond: - if requests_incremented: + self.operation_count -= 1 + if slot_acquired: self.requests -= 1 - if active_sockets_incremented: self.active_sockets -= 1 - if op_count_incremented: - self.operation_count -= 1 - if requests_incremented or active_sockets_incremented: self.size_cond.notify() if not emitted_event: From a2943a4ec0b73b2474ed14922925a245b3e174d9 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 31 Jul 2026 10:43:20 -0500 Subject: [PATCH 14/25] PYTHON-5898 Assert checkout-failed is emitted under the pool lock Register a CMAP listener that records pool.lock.locked() at the moment ConnectionCheckOutFailedEvent fires. Listeners run synchronously inside the publish call, so this observes directly whether the emitting code still holds the mutex, pinning the PYTHON-3519 ordering guarantee. --- test/asynchronous/test_pooling.py | 33 +++++++++++++++++++++++++++++++ test/test_pooling.py | 33 +++++++++++++++++++++++++++++++ 2 files changed, 66 insertions(+) diff --git a/test/asynchronous/test_pooling.py b/test/asynchronous/test_pooling.py index 9a9b7a7381..d9b9d26d6a 100644 --- a/test/asynchronous/test_pooling.py +++ b/test/asynchronous/test_pooling.py @@ -462,6 +462,39 @@ async def test_paused_pool_checkout_failure_does_not_leak_or_double_emit(self): self.assertEqual(len(failed_events), 1, [e.reason for e in failed_events]) self.assertEqual(failed_events[0].reason, ConnectionCheckOutFailedReason.CONN_ERROR) + async def test_checkout_failed_event_is_emitted_under_the_pool_lock(self): + # A readiness check that fails must publish its + # ConnectionCheckOutFailedEvent while still holding the pool mutex. + # _reset() publishes PoolClearedEvent under that same mutex precisely + # so that it is always recorded first; deferring the checkout failure + # to after the mutex is released loses that ordering (PYTHON-3519). + # Listeners are invoked synchronously inside the publish call, so this + # one observes directly whether the emitting code holds the mutex. + locked_while_emitting = [] + pool_ref: list = [] + + class LockObservingListener(CMAPListener): + def connection_check_out_failed(self, event): + locked_while_emitting.append(pool_ref[0].lock.locked()) + super().connection_check_out_failed(event) + + listener = LockObservingListener() + pool = await self.create_pool(max_pool_size=1, event_listeners=_EventListeners([listener])) + self.addAsyncCleanup(pool.close) + pool_ref.append(pool) + + await pool.reset() # Pause the pool. + listener.reset() + with self.assertRaises(AutoReconnect): + async with pool.checkout(): + pass + + self.assertEqual( + [True], + locked_while_emitting, + "ConnectionCheckOutFailedEvent must be published while the pool mutex is held", + ) + async def test_no_wait_queue_timeout(self): # Verify get_socket() with no wait_queue_timeout blocks forever. pool = await self.create_pool(max_pool_size=1) diff --git a/test/test_pooling.py b/test/test_pooling.py index dd2868bfcd..41b2c39345 100644 --- a/test/test_pooling.py +++ b/test/test_pooling.py @@ -462,6 +462,39 @@ def test_paused_pool_checkout_failure_does_not_leak_or_double_emit(self): self.assertEqual(len(failed_events), 1, [e.reason for e in failed_events]) self.assertEqual(failed_events[0].reason, ConnectionCheckOutFailedReason.CONN_ERROR) + def test_checkout_failed_event_is_emitted_under_the_pool_lock(self): + # A readiness check that fails must publish its + # ConnectionCheckOutFailedEvent while still holding the pool mutex. + # _reset() publishes PoolClearedEvent under that same mutex precisely + # so that it is always recorded first; deferring the checkout failure + # to after the mutex is released loses that ordering (PYTHON-3519). + # Listeners are invoked synchronously inside the publish call, so this + # one observes directly whether the emitting code holds the mutex. + locked_while_emitting = [] + pool_ref: list = [] + + class LockObservingListener(CMAPListener): + def connection_check_out_failed(self, event): + locked_while_emitting.append(pool_ref[0].lock.locked()) + super().connection_check_out_failed(event) + + listener = LockObservingListener() + pool = self.create_pool(max_pool_size=1, event_listeners=_EventListeners([listener])) + self.addCleanup(pool.close) + pool_ref.append(pool) + + pool.reset() # Pause the pool. + listener.reset() + with self.assertRaises(AutoReconnect): + with pool.checkout(): + pass + + self.assertEqual( + [True], + locked_while_emitting, + "ConnectionCheckOutFailedEvent must be published while the pool mutex is held", + ) + def test_no_wait_queue_timeout(self): # Verify get_socket() with no wait_queue_timeout blocks forever. pool = self.create_pool(max_pool_size=1) From afe01b1616ec3b30458a2c508b47af84c483747a Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 31 Jul 2026 11:13:12 -0500 Subject: [PATCH 15/25] PYTHON-5898 Point the changelog at known_servers, not server_descriptions candidate_servers returned a list of known servers; server_descriptions() returns a dict of every server including Unknown. known_servers is the exact analogue and is what selection now falls back to. --- doc/changelog.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/doc/changelog.rst b/doc/changelog.rst index c06bed7039..190f8501a2 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -33,7 +33,7 @@ PyMongo 4.18 brings a number of changes including: - Removed ``TopologyDescription.candidate_servers``, added in PyMongo 4.16.0. Its value depended on which server-selection call happened to run last, so it could not be relied on. Use - :meth:`~pymongo.topology_description.TopologyDescription.server_descriptions` + :attr:`~pymongo.topology_description.TopologyDescription.known_servers` instead. - Fixed a leak where every failed connection checkout permanently incremented a pool's ``operation_count``, biasing server selection among From a2e7ed5b39eca5088e9b07c402dda13bd139e184 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 31 Jul 2026 11:13:41 -0500 Subject: [PATCH 16/25] PYTHON-5898 Give the non-mutation requirement its real rationale The old wording blamed concurrent selection calls, but select_servers(), get_primary() and _get_replica_set_members() all hold Topology._lock for the whole call, so those apply_selector() calls never overlap. The actual hazards are sequential state leakage into later readers of the same description, and genuinely concurrent reads from external callers such as topology event listeners and holders of client.topology_description. --- pymongo/topology_description.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/pymongo/topology_description.py b/pymongo/topology_description.py index 0dd815f48f..e559b357df 100644 --- a/pymongo/topology_description.py +++ b/pymongo/topology_description.py @@ -280,13 +280,16 @@ def _filter_servers( If every known server is deprioritized, all known servers are returned so that selection can still make progress. - This method must not mutate ``self``. A TopologyDescription is shared - by every concurrent selection call and is replaced, not edited, when - the topology changes, while deprioritization is specific to a single - call. Callers such as + 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, so any filtering left on it - would be applied to selections that never asked for it. + 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. """ From 239c085df29165e2e1b9ccfbca588997ef1630ab Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 31 Jul 2026 11:13:43 -0500 Subject: [PATCH 17/25] PYTHON-5898 Harden the immutability tests Correct the class docstring's rationale to match topology_description. Surface unexpected worker exceptions (a BrokenBarrierError from the barrier timeout, say) through errors instead of letting threading print and discard them, which left errors empty and passed vacuously. Add a get_secondaries() regression test, since the stale-membership bug the changelog describes covered client.secondaries/arbiters too. --- test/test_topology.py | 75 ++++++++++++++++++++++++++++++++----------- 1 file changed, 57 insertions(+), 18 deletions(-) diff --git a/test/test_topology.py b/test/test_topology.py index af44e39eab..b0cabe290e 100644 --- a/test/test_topology.py +++ b/test/test_topology.py @@ -939,9 +939,14 @@ def create_mock_replica_set_topology(hosts=("a", "b", "c")): class TestTopologyDescriptionImmutability(TopologyTest): - """A TopologyDescription is shared by every concurrent selection call and - is replaced, not edited, when the topology changes (PYTHON-5898), so - apply_selector() must not mutate it. + """apply_selector() must not mutate the TopologyDescription (PYTHON-5898). + + Deprioritization is specific to a single call, but the description outlives + it, so filtering left behind would leak into later readers such as + get_primary(), which builds its selection straight from the description. + The same object is also handed to code on other threads -- topology event + listeners and anything holding client.topology_description -- which may + read it concurrently with a selection in progress. """ def test_concurrent_apply_selector_with_deprioritized_servers(self): @@ -963,23 +968,33 @@ def test_concurrent_apply_selector_with_deprioritized_servers(self): barrier = threading.Barrier(6) def deprioritizing_worker(): - barrier.wait(timeout=30) - for _ in range(iterations): - # A retryable write retrying away from the primary must never - # be handed the deprioritized primary back. - sds = description.apply_selector( - ReadPreference.PRIMARY_PREFERRED, deprioritized_servers=[primary_sd] - ) - if any(sd.address == primary_sd.address for sd in sds): - errors.append(f"deprioritized primary was selected: {sds}") + # Without this, threading would print an unexpected exception (a + # BrokenBarrierError from the barrier timeout, say) and discard it, + # leaving errors empty and passing the test vacuously. + try: + barrier.wait(timeout=30) + for _ in range(iterations): + # A retryable write retrying away from the primary must + # never be handed the deprioritized primary back. + sds = description.apply_selector( + ReadPreference.PRIMARY_PREFERRED, deprioritized_servers=[primary_sd] + ) + if any(sd.address == primary_sd.address for sd in sds): + errors.append(f"deprioritized primary was selected: {sds}") + except BaseException as exc: + errors.append(repr(exc)) def plain_worker(): - barrier.wait(timeout=30) - for _ in range(iterations): - # An ordinary operation must always find the healthy primary. - sds = description.apply_selector(Primary()) - if [sd.address for sd in sds] != [primary_sd.address]: - errors.append(f"Primary() selected {sds} instead of the primary") + try: + barrier.wait(timeout=30) + for _ in range(iterations): + # An ordinary operation must always find the healthy + # primary. + sds = description.apply_selector(Primary()) + if [sd.address for sd in sds] != [primary_sd.address]: + errors.append(f"Primary() selected {sds} instead of the primary") + except BaseException as exc: + errors.append(repr(exc)) threads = [threading.Thread(target=deprioritizing_worker) for _ in range(3)] threads += [threading.Thread(target=plain_worker) for _ in range(3)] @@ -1016,6 +1031,30 @@ def test_get_primary_after_deprioritized_selection(self): # get_primary() must still find the primary afterwards. self.assertEqual(("a", 27017), t.get_primary()) + def test_get_secondaries_after_deprioritized_selection(self): + # get_secondaries()/get_arbiters() (and client.secondaries/arbiters) + # also build their selection straight from the description, so a + # selection that deprioritizes a secondary must not drop it from a + # later membership listing. + t = create_mock_replica_set_topology() + self.addCleanup(t.close) + description = t.description + secondaries = {("b", 27017), ("c", 27017)} + secondary_sd = description.server_descriptions()[("b", 27017)] + self.assertEqual(SERVER_TYPE.RSSecondary, secondary_sd.server_type) + + self.assertEqual(secondaries, t.get_secondaries()) + + # Simulate a retryable operation deprioritizing one secondary for a + # single selection call. + description.apply_selector( + ReadPreference.SECONDARY_PREFERRED, deprioritized_servers=[secondary_sd] + ) + + # Both secondaries must still be reported afterwards. + self.assertEqual(secondaries, t.get_secondaries()) + self.assertEqual(set(), t.get_arbiters()) + if __name__ == "__main__": unittest.main() From eb68fb942e4d931c98c46faa90f54471948ad3df Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 31 Jul 2026 13:35:38 -0500 Subject: [PATCH 18/25] PYTHON-5898 Track requests and active_sockets with separate flags Set each flag immediately after its own increment, with no statement in between, matching upstream. A single combined flag left a window where an interrupt delivered between the two increments and the flag assignment would leak both counters. --- pymongo/asynchronous/pool.py | 20 ++++++++++++++------ pymongo/synchronous/pool.py | 20 ++++++++++++++------ 2 files changed, 28 insertions(+), 12 deletions(-) diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py index 601d32f7d6..fe4675f75d 100644 --- a/pymongo/asynchronous/pool.py +++ b/pymongo/asynchronous/pool.py @@ -1017,7 +1017,11 @@ async def _get_conn( conn = None op_count_incremented = False - slot_acquired = False + # Each flag is set immediately after its own increment, with no + # statement in between, so an interrupt cannot leave a counter + # incremented but unrecorded. + 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 @@ -1035,10 +1039,11 @@ async def _get_conn( # 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 - slot_acquired = True + active_sockets_incremented = True - if not slot_acquired: + if not requests_incremented: # Slow path: no slot was free, so wait on the requests # semaphore for one to be released. async with self.size_cond: @@ -1058,8 +1063,9 @@ async def _get_conn( self._raise_if_not_ready(checkout_started_time, emit_event=True) emitted_event = False self.requests += 1 + requests_incremented = True self.active_sockets += 1 - slot_acquired = True + active_sockets_incremented = True while conn is None: # CMAP: we MUST wait for either maxConnecting OR for a socket @@ -1108,9 +1114,11 @@ async def _get_conn( if op_count_incremented: async with self.size_cond: self.operation_count -= 1 - if slot_acquired: - self.requests -= 1 + if active_sockets_incremented: self.active_sockets -= 1 + if requests_incremented: + # Notify last, so a waiter wakes to consistent counters. + self.requests -= 1 self.size_cond.notify() if not emitted_event: diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py index 118ecd195e..a271431799 100644 --- a/pymongo/synchronous/pool.py +++ b/pymongo/synchronous/pool.py @@ -1013,7 +1013,11 @@ def _get_conn( conn = None op_count_incremented = False - slot_acquired = False + # Each flag is set immediately after its own increment, with no + # statement in between, so an interrupt cannot leave a counter + # incremented but unrecorded. + 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 @@ -1031,10 +1035,11 @@ def _get_conn( # 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 - slot_acquired = True + active_sockets_incremented = True - if not slot_acquired: + if not requests_incremented: # Slow path: no slot was free, so wait on the requests # semaphore for one to be released. with self.size_cond: @@ -1054,8 +1059,9 @@ def _get_conn( self._raise_if_not_ready(checkout_started_time, emit_event=True) emitted_event = False self.requests += 1 + requests_incremented = True self.active_sockets += 1 - slot_acquired = True + active_sockets_incremented = True while conn is None: # CMAP: we MUST wait for either maxConnecting OR for a socket @@ -1104,9 +1110,11 @@ def _get_conn( if op_count_incremented: with self.size_cond: self.operation_count -= 1 - if slot_acquired: - self.requests -= 1 + if active_sockets_incremented: self.active_sockets -= 1 + if requests_incremented: + # Notify last, so a witer wakes to consistent counters. + self.requests -= 1 self.size_cond.notify() if not emitted_event: From d9de328df3b3eae5c02542a1acfa3c794b5f0d3a Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 3 Aug 2026 06:17:45 -0500 Subject: [PATCH 19/25] PYTHON-5898 Correct operation_count leak changelog and flag candidate_servers removal as breaking --- doc/changelog.rst | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/doc/changelog.rst b/doc/changelog.rst index 190f8501a2..9bd52fdd71 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -30,14 +30,18 @@ PyMongo 4.18 brings a number of changes including: - Fixed ``client.primary`` raising ``IndexError``, and ``client.secondaries`` and ``client.arbiters`` returning stale results, after a retryable operation deprioritized the primary. -- Removed ``TopologyDescription.candidate_servers``, added in PyMongo 4.16.0. - Its value depended on which server-selection call happened to run last, so it - could not be relied on. Use +- **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``, biasing server selection among - mongoses toward the affected server. +- 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. From 932632ccd769e3fb45c17a9267fbd2addcecd2f4 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 3 Aug 2026 06:17:45 -0500 Subject: [PATCH 20/25] PYTHON-5898 Fix inaccurate checkout comments in pool --- pymongo/asynchronous/pool.py | 16 +++++++++++----- pymongo/synchronous/pool.py | 16 +++++++++++----- 2 files changed, 22 insertions(+), 10 deletions(-) diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py index fe4675f75d..5478dcac02 100644 --- a/pymongo/asynchronous/pool.py +++ b/pymongo/asynchronous/pool.py @@ -1018,8 +1018,10 @@ async def _get_conn( conn = None op_count_incremented = False # Each flag is set immediately after its own increment, with no - # statement in between, so an interrupt cannot leave a counter - # incremented but unrecorded. + # 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 @@ -1044,8 +1046,10 @@ async def _get_conn( active_sockets_incremented = True if not requests_incremented: - # Slow path: no slot was free, so wait on the requests - # semaphore for one to be released. + # 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) @@ -1117,8 +1121,10 @@ async def _get_conn( if active_sockets_incremented: self.active_sockets -= 1 if requests_incremented: - # Notify last, so a waiter wakes to consistent counters. 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: diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py index a271431799..21f8ced18a 100644 --- a/pymongo/synchronous/pool.py +++ b/pymongo/synchronous/pool.py @@ -1014,8 +1014,10 @@ def _get_conn( conn = None op_count_incremented = False # Each flag is set immediately after its own increment, with no - # statement in between, so an interrupt cannot leave a counter - # incremented but unrecorded. + # 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 @@ -1040,8 +1042,10 @@ def _get_conn( active_sockets_incremented = True if not requests_incremented: - # Slow path: no slot was free, so wait on the requests - # semaphore for one to be released. + # 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) @@ -1113,8 +1117,10 @@ def _get_conn( if active_sockets_incremented: self.active_sockets -= 1 if requests_incremented: - # Notify last, so a witer wakes to consistent counters. 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: From fc30cfe1b32cf5f0638ba56a280eaeb4ee438ecc Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 3 Aug 2026 06:17:45 -0500 Subject: [PATCH 21/25] PYTHON-5898 Drop _filter_servers(None) indirection in apply_selector --- pymongo/topology_description.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/pymongo/topology_description.py b/pymongo/topology_description.py index e559b357df..32d27fedb4 100644 --- a/pymongo/topology_description.py +++ b/pymongo/topology_description.py @@ -371,7 +371,9 @@ def apply_selector( selection = selector(selection) # No suitable servers found, apply preference again but include deprioritized servers. if not selection and deprioritized_servers: - selection = Selection.from_topology_description(self, self._filter_servers(None)) + # No candidate filtering: from_topology_description() defaults + # to all known servers. + selection = Selection.from_topology_description(self) selection = selector(selection) # Apply custom selector followed by localThresholdMS. From 2042961bcfe46877aad6f5a2eb395bae677078bc Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 3 Aug 2026 06:17:46 -0500 Subject: [PATCH 22/25] PYTHON-5898 Make the arbiter assertion load-bearing in test_topology --- test/test_topology.py | 45 ++++++++++++++++++++++++++++++++----------- 1 file changed, 34 insertions(+), 11 deletions(-) diff --git a/test/test_topology.py b/test/test_topology.py index b0cabe290e..c2772dd87b 100644 --- a/test/test_topology.py +++ b/test/test_topology.py @@ -908,9 +908,14 @@ def test_no_mongoses(self): self.assertMessage("No mongoses available", t) -def create_mock_replica_set_topology(hosts=("a", "b", "c")): - """A ReplicaSetWithPrimary topology: hosts[0] is primary, the rest secondary.""" - t = create_mock_topology(seeds=list(hosts), replica_set_name="rs") +def create_mock_replica_set_topology(hosts=("a", "b", "c"), arbiters=()): + """A ReplicaSetWithPrimary topology: hosts[0] is primary, the rest secondary. + + Any names in ``arbiters`` join the set as arbiters. + """ + arbiters = list(arbiters) + members = {"hosts": list(hosts), "arbiters": arbiters} + t = create_mock_topology(seeds=list(hosts) + arbiters, replica_set_name="rs") got_hello( t, (hosts[0], 27017), @@ -918,8 +923,8 @@ def create_mock_replica_set_topology(hosts=("a", "b", "c")): "ok": 1, HelloCompat.LEGACY_CMD: True, "setName": "rs", - "hosts": list(hosts), "maxWireVersion": common.MIN_SUPPORTED_WIRE_VERSION, + **members, }, ) for host in hosts[1:]: @@ -931,8 +936,21 @@ def create_mock_replica_set_topology(hosts=("a", "b", "c")): HelloCompat.LEGACY_CMD: False, "secondary": True, "setName": "rs", - "hosts": list(hosts), "maxWireVersion": common.MIN_SUPPORTED_WIRE_VERSION, + **members, + }, + ) + for arbiter in arbiters: + got_hello( + t, + (arbiter, 27017), + { + "ok": 1, + HelloCompat.LEGACY_CMD: False, + "arbiterOnly": True, + "setName": "rs", + "maxWireVersion": common.MIN_SUPPORTED_WIRE_VERSION, + **members, }, ) return t @@ -1036,24 +1054,29 @@ def test_get_secondaries_after_deprioritized_selection(self): # also build their selection straight from the description, so a # selection that deprioritizes a secondary must not drop it from a # later membership listing. - t = create_mock_replica_set_topology() + t = create_mock_replica_set_topology(arbiters=("d",)) self.addCleanup(t.close) description = t.description secondaries = {("b", 27017), ("c", 27017)} + arbiters = {("d", 27017)} secondary_sd = description.server_descriptions()[("b", 27017)] self.assertEqual(SERVER_TYPE.RSSecondary, secondary_sd.server_type) + arbiter_sd = description.server_descriptions()[("d", 27017)] + self.assertEqual(SERVER_TYPE.RSArbiter, arbiter_sd.server_type) self.assertEqual(secondaries, t.get_secondaries()) + self.assertEqual(arbiters, t.get_arbiters()) - # Simulate a retryable operation deprioritizing one secondary for a - # single selection call. + # Simulate a retryable operation deprioritizing one secondary and the + # arbiter for a single selection call. description.apply_selector( - ReadPreference.SECONDARY_PREFERRED, deprioritized_servers=[secondary_sd] + ReadPreference.SECONDARY_PREFERRED, + deprioritized_servers=[secondary_sd, arbiter_sd], ) - # Both secondaries must still be reported afterwards. + # Both secondaries and the arbiter must still be reported afterwards. self.assertEqual(secondaries, t.get_secondaries()) - self.assertEqual(set(), t.get_arbiters()) + self.assertEqual(arbiters, t.get_arbiters()) if __name__ == "__main__": From c437addf4d370372212b6cde0d66f5a1ccacbea2 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 3 Aug 2026 06:17:46 -0500 Subject: [PATCH 23/25] PYTHON-5898 Cover slow-path checkout failure emission under the pool lock --- test/asynchronous/test_pooling.py | 87 ++++++++++++++++++++++++++++++- test/test_pooling.py | 87 ++++++++++++++++++++++++++++++- 2 files changed, 172 insertions(+), 2 deletions(-) diff --git a/test/asynchronous/test_pooling.py b/test/asynchronous/test_pooling.py index d9b9d26d6a..316bc49715 100644 --- a/test/asynchronous/test_pooling.py +++ b/test/asynchronous/test_pooling.py @@ -32,16 +32,18 @@ from pymongo import AsyncMongoClient, message, timeout from pymongo.errors import AutoReconnect, ConnectionFailure, DuplicateKeyError from pymongo.hello import HelloCompat -from pymongo.lock import _async_create_lock +from pymongo.lock import _async_cond_wait, _async_create_lock from pymongo.monitoring import ( ConnectionCheckOutFailedEvent, ConnectionCheckOutFailedReason, + PoolClearedEvent, _EventListeners, ) from test.asynchronous.utils import async_get_pool, async_joinall, flaky sys.path[0:0] = [""] +from pymongo.asynchronous import pool as pool_module from pymongo.asynchronous.pool import Pool, PoolOptions from pymongo.socket_checker import SocketChecker from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest @@ -280,6 +282,7 @@ def add(self, item): # Bookkeeping must be rolled back, not left half-updated. self.assertEqual(0, cx_pool.active_sockets) self.assertEqual(0, cx_pool.requests) + self.assertEqual(0, cx_pool.operation_count) async def test_pool_removes_closed_socket(self): # Test that Pool removes explicitly closed socket. @@ -495,6 +498,88 @@ def connection_check_out_failed(self, event): "ConnectionCheckOutFailedEvent must be published while the pool mutex is held", ) + async def test_checkout_failed_event_is_emitted_under_the_pool_lock_slow_path(self): + # Same guarantee as the test above, but for the readiness checks on + # the slow path. That test pauses an idle pool, so a slot is free and + # only the fast path runs; moving emission out from under the mutex in + # the slow path alone would not fail it. Here the pool's only slot is + # already taken, so the checkout blocks on size_cond, and reset() + # wakes it. This is exactly the PYTHON-3519 scenario: _reset() + # publishes PoolClearedEvent and calls notify_all() while holding the + # mutex, so the woken checkout can only publish its + # ConnectionCheckOutFailedEvent afterwards -- but only if it, too, + # publishes while still holding the mutex. + locked_while_emitting = [] + pool_ref: list = [] + + class LockObservingListener(CMAPListener): + def connection_check_out_failed(self, event): + locked_while_emitting.append(pool_ref[0].lock.locked()) + super().connection_check_out_failed(event) + + listener = LockObservingListener() + pool = await self.create_pool(max_pool_size=1, event_listeners=_EventListeners([listener])) + self.addAsyncCleanup(pool.close) + pool_ref.append(pool) + + errors: list = [] + + async def blocked_checkout(): + try: + async with pool.checkout(): + pass + except BaseException as exc: + errors.append(exc) + + # The checkout must be parked on size_cond before the pool is reset, + # otherwise it could still fail on the fast path and this test would + # silently duplicate the one above. Flag it from inside the condition + # wait itself: the flag is set while size_cond is still held, so + # reset() cannot acquire the mutex until the checkout has released it + # by blocking. + parked: list = [] + real_cond_wait = _async_cond_wait + + async def flagging_cond_wait(condition, timeout): + if condition is pool.size_cond: + parked.append(True) + return await real_cond_wait(condition, timeout) + + with patch.object(pool_module, "_async_cond_wait", flagging_cond_wait): + async with pool.checkout(): + listener.reset() + task = ConcurrentRunner(target=blocked_checkout, name="blocked_checkout") + await task.start() + + start = time.monotonic() + while not parked: # noqa: ASYNC110, RUF100 + self.assertLess( + time.monotonic() - start, 30, "checkout never blocked on size_cond" + ) + await asyncio.sleep(0.01) + + await pool.reset() # Pause the pool and wake the blocked checkout. + await task.join(30) + self.assertFalse(task.is_alive(), "blocked checkout never finished") + + self.assertEqual(1, len(errors), f"expected exactly one failed checkout, got {errors}") + self.assertIsInstance(errors[0], AutoReconnect) + self.assertEqual( + [True], + locked_while_emitting, + "ConnectionCheckOutFailedEvent must be published while the pool mutex is held", + ) + # PYTHON-3519: the clear that caused the failure must be recorded first. + self.assertEqual( + [PoolClearedEvent, ConnectionCheckOutFailedEvent], + [ + type(event) + for event in listener.events_by_type( + (PoolClearedEvent, ConnectionCheckOutFailedEvent) + ) + ], + ) + async def test_no_wait_queue_timeout(self): # Verify get_socket() with no wait_queue_timeout blocks forever. pool = await self.create_pool(max_pool_size=1) diff --git a/test/test_pooling.py b/test/test_pooling.py index 41b2c39345..09a9c66739 100644 --- a/test/test_pooling.py +++ b/test/test_pooling.py @@ -32,10 +32,11 @@ from pymongo import MongoClient, message, timeout from pymongo.errors import AutoReconnect, ConnectionFailure, DuplicateKeyError from pymongo.hello import HelloCompat -from pymongo.lock import _create_lock +from pymongo.lock import _cond_wait, _create_lock from pymongo.monitoring import ( ConnectionCheckOutFailedEvent, ConnectionCheckOutFailedReason, + PoolClearedEvent, _EventListeners, ) from test.utils import flaky, get_pool, joinall @@ -43,6 +44,7 @@ sys.path[0:0] = [""] from pymongo.socket_checker import SocketChecker +from pymongo.synchronous import pool as pool_module from pymongo.synchronous.pool import Pool, PoolOptions from test import IntegrationTest, client_context, unittest from test.helpers import ConcurrentRunner @@ -280,6 +282,7 @@ def add(self, item): # Bookkeeping must be rolled back, not left half-updated. self.assertEqual(0, cx_pool.active_sockets) self.assertEqual(0, cx_pool.requests) + self.assertEqual(0, cx_pool.operation_count) def test_pool_removes_closed_socket(self): # Test that Pool removes explicitly closed socket. @@ -495,6 +498,88 @@ def connection_check_out_failed(self, event): "ConnectionCheckOutFailedEvent must be published while the pool mutex is held", ) + def test_checkout_failed_event_is_emitted_under_the_pool_lock_slow_path(self): + # Same guarantee as the test above, but for the readiness checks on + # the slow path. That test pauses an idle pool, so a slot is free and + # only the fast path runs; moving emission out from under the mutex in + # the slow path alone would not fail it. Here the pool's only slot is + # already taken, so the checkout blocks on size_cond, and reset() + # wakes it. This is exactly the PYTHON-3519 scenario: _reset() + # publishes PoolClearedEvent and calls notify_all() while holding the + # mutex, so the woken checkout can only publish its + # ConnectionCheckOutFailedEvent afterwards -- but only if it, too, + # publishes while still holding the mutex. + locked_while_emitting = [] + pool_ref: list = [] + + class LockObservingListener(CMAPListener): + def connection_check_out_failed(self, event): + locked_while_emitting.append(pool_ref[0].lock.locked()) + super().connection_check_out_failed(event) + + listener = LockObservingListener() + pool = self.create_pool(max_pool_size=1, event_listeners=_EventListeners([listener])) + self.addCleanup(pool.close) + pool_ref.append(pool) + + errors: list = [] + + def blocked_checkout(): + try: + with pool.checkout(): + pass + except BaseException as exc: + errors.append(exc) + + # The checkout must be parked on size_cond before the pool is reset, + # otherwise it could still fail on the fast path and this test would + # silently duplicate the one above. Flag it from inside the condition + # wait itself: the flag is set while size_cond is still held, so + # reset() cannot acquire the mutex until the checkout has released it + # by blocking. + parked: list = [] + real_cond_wait = _cond_wait + + def flagging_cond_wait(condition, timeout): + if condition is pool.size_cond: + parked.append(True) + return real_cond_wait(condition, timeout) + + with patch.object(pool_module, "_cond_wait", flagging_cond_wait): + with pool.checkout(): + listener.reset() + task = ConcurrentRunner(target=blocked_checkout, name="blocked_checkout") + task.start() + + start = time.monotonic() + while not parked: # noqa: ASYNC110, RUF100 + self.assertLess( + time.monotonic() - start, 30, "checkout never blocked on size_cond" + ) + time.sleep(0.01) + + pool.reset() # Pause the pool and wake the blocked checkout. + task.join(30) + self.assertFalse(task.is_alive(), "blocked checkout never finished") + + self.assertEqual(1, len(errors), f"expected exactly one failed checkout, got {errors}") + self.assertIsInstance(errors[0], AutoReconnect) + self.assertEqual( + [True], + locked_while_emitting, + "ConnectionCheckOutFailedEvent must be published while the pool mutex is held", + ) + # PYTHON-3519: the clear that caused the failure must be recorded first. + self.assertEqual( + [PoolClearedEvent, ConnectionCheckOutFailedEvent], + [ + type(event) + for event in listener.events_by_type( + (PoolClearedEvent, ConnectionCheckOutFailedEvent) + ) + ], + ) + def test_no_wait_queue_timeout(self): # Verify get_socket() with no wait_queue_timeout blocks forever. pool = self.create_pool(max_pool_size=1) From 69ca9dfe4c500d9c320c9af464b69735feb1fe72 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 3 Aug 2026 07:20:20 -0500 Subject: [PATCH 24/25] PYTHON-5898 Pin the pool lock acquisition count on the checkout fast path Nothing asserted the number of lock acquisitions, so splitting the merged counter bookkeeping back apart would have passed every other test in the file and silently undone the change this ticket is for. --- test/asynchronous/test_pooling.py | 50 +++++++++++++++++++++++++++++++ test/test_pooling.py | 50 +++++++++++++++++++++++++++++++ 2 files changed, 100 insertions(+) diff --git a/test/asynchronous/test_pooling.py b/test/asynchronous/test_pooling.py index 316bc49715..5b4ae5842c 100644 --- a/test/asynchronous/test_pooling.py +++ b/test/asynchronous/test_pooling.py @@ -580,6 +580,56 @@ async def flagging_cond_wait(condition, timeout): ], ) + async def test_uncontended_checkout_pool_lock_acquisitions(self): + # An uncontended checkout must do all of its operation_count, + # requests and active_sockets bookkeeping in a single critical + # section. Pinning the acquisition count keeps that from silently + # regressing: splitting the bookkeeping back apart would still pass + # every other test in this file. + # + # Two acquisitions are expected for a warm checkout that reuses an + # idle connection: one for the merged bookkeeping, and one to + # register the connection's cancel context. size_cond and + # _max_connecting_cond are distinct objects wrapping the same mutex, + # so their blocks are not counted here. + pool = await self.create_pool(max_pool_size=1) + self.addAsyncCleanup(pool.close) + + # Check a connection out and back in first, so this measurement + # covers a warm pool and does not include connection establishment. + async with pool.checkout(): + pass + + acquires = 0 + real_lock = pool.lock + + class CountingLock: + async def __aenter__(self): + nonlocal acquires + acquires += 1 + return await real_lock.__aenter__() + + async def __aexit__(self, *args): + return await real_lock.__aexit__(*args) + + def __getattr__(self, name): + return getattr(real_lock, name) + + pool.lock = CountingLock() # type: ignore[assignment] + try: + async with pool.checkout(): + checkout_acquires = acquires + finally: + pool.lock = real_lock + + self.assertEqual( + 2, + checkout_acquires, + f"an uncontended checkout should acquire the pool lock twice -- once for " + f"counter bookkeeping and once to register the cancel context -- got " + f"{checkout_acquires}", + ) + async def test_no_wait_queue_timeout(self): # Verify get_socket() with no wait_queue_timeout blocks forever. pool = await self.create_pool(max_pool_size=1) diff --git a/test/test_pooling.py b/test/test_pooling.py index 09a9c66739..22c3deb74b 100644 --- a/test/test_pooling.py +++ b/test/test_pooling.py @@ -580,6 +580,56 @@ def flagging_cond_wait(condition, timeout): ], ) + def test_uncontended_checkout_pool_lock_acquisitions(self): + # An uncontended checkout must do all of its operation_count, + # requests and active_sockets bookkeeping in a single critical + # section. Pinning the acquisition count keeps that from silently + # regressing: splitting the bookkeeping back apart would still pass + # every other test in this file. + # + # Two acquisitions are expected for a warm checkout that reuses an + # idle connection: one for the merged bookkeeping, and one to + # register the connection's cancel context. size_cond and + # _max_connecting_cond are distinct objects wrapping the same mutex, + # so their blocks are not counted here. + pool = self.create_pool(max_pool_size=1) + self.addCleanup(pool.close) + + # Check a connection out and back in first, so this measurement + # covers a warm pool and does not include connection establishment. + with pool.checkout(): + pass + + acquires = 0 + real_lock = pool.lock + + class CountingLock: + def __enter__(self): + nonlocal acquires + acquires += 1 + return real_lock.__enter__() + + def __exit__(self, *args): + return real_lock.__exit__(*args) + + def __getattr__(self, name): + return getattr(real_lock, name) + + pool.lock = CountingLock() # type: ignore[assignment] + try: + with pool.checkout(): + checkout_acquires = acquires + finally: + pool.lock = real_lock + + self.assertEqual( + 2, + checkout_acquires, + f"an uncontended checkout should acquire the pool lock twice -- once for " + f"counter bookkeeping and once to register the cancel context -- got " + f"{checkout_acquires}", + ) + def test_no_wait_queue_timeout(self): # Verify get_socket() with no wait_queue_timeout blocks forever. pool = self.create_pool(max_pool_size=1) From 0815f42e8838c981c5ae27da30d904e390fd8e6d Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 3 Aug 2026 09:10:43 -0500 Subject: [PATCH 25/25] PYTHON-5898 Pin the pool lock acquisition count on the contended path Counterpart to the fast-path test. Folding the slot bookkeeping into the size_cond critical section removed an acquisition from the contended path, and nothing asserted it, so splitting it back apart would have gone unnoticed. --- test/asynchronous/test_pooling.py | 58 +++++++++++++++++++++++++++++++ test/test_pooling.py | 58 +++++++++++++++++++++++++++++++ 2 files changed, 116 insertions(+) diff --git a/test/asynchronous/test_pooling.py b/test/asynchronous/test_pooling.py index 5b4ae5842c..96a24b4ef2 100644 --- a/test/asynchronous/test_pooling.py +++ b/test/asynchronous/test_pooling.py @@ -630,6 +630,64 @@ def __getattr__(self, name): f"{checkout_acquires}", ) + async def test_contended_checkout_pool_lock_acquisitions(self): + # The same guarantee as the test above, for a checkout that has to + # wait for a slot: its requests and active_sockets bookkeeping belongs + # in the one size_cond critical section, not in a further acquisition + # afterwards. + # + # Driven from a single task so the count is not polluted by the + # releasing side -- checkin() acquires the pool lock twice itself. The + # slot is freed from inside the condition wait, which is where a real + # waiter would be woken. + pool = await self.create_pool(max_pool_size=1) + self.addAsyncCleanup(pool.close) + + async with pool.checkout(): + pass + + # Make the fast path's slot check fail so the slow path runs. + pool.requests = pool.max_pool_size + + real_cond_wait = _async_cond_wait + + async def releasing_cond_wait(condition, timeout): + if condition is pool.size_cond: + pool.requests = 0 + return True + return await real_cond_wait(condition, timeout) + + acquires = 0 + real_lock = pool.lock + + class CountingLock: + async def __aenter__(self): + nonlocal acquires + acquires += 1 + return await real_lock.__aenter__() + + async def __aexit__(self, *args): + return await real_lock.__aexit__(*args) + + def __getattr__(self, name): + return getattr(real_lock, name) + + pool.lock = CountingLock() # type: ignore[assignment] + try: + with patch.object(pool_module, "_async_cond_wait", releasing_cond_wait): + async with pool.checkout(): + checkout_acquires = acquires + finally: + pool.lock = real_lock + + self.assertEqual( + 2, + checkout_acquires, + f"a checkout that waited for a slot should acquire the pool lock twice -- " + f"once for counter bookkeeping and once to register the cancel context -- " + f"got {checkout_acquires}", + ) + async def test_no_wait_queue_timeout(self): # Verify get_socket() with no wait_queue_timeout blocks forever. pool = await self.create_pool(max_pool_size=1) diff --git a/test/test_pooling.py b/test/test_pooling.py index 22c3deb74b..2829fe6ef3 100644 --- a/test/test_pooling.py +++ b/test/test_pooling.py @@ -630,6 +630,64 @@ def __getattr__(self, name): f"{checkout_acquires}", ) + def test_contended_checkout_pool_lock_acquisitions(self): + # The same guarantee as the test above, for a checkout that has to + # wait for a slot: its requests and active_sockets bookkeeping belongs + # in the one size_cond critical section, not in a further acquisition + # afterwards. + # + # Driven from a single task so the count is not polluted by the + # releasing side -- checkin() acquires the pool lock twice itself. The + # slot is freed from inside the condition wait, which is where a real + # witer would be woken. + pool = self.create_pool(max_pool_size=1) + self.addCleanup(pool.close) + + with pool.checkout(): + pass + + # Make the fast path's slot check fail so the slow path runs. + pool.requests = pool.max_pool_size + + real_cond_wait = _cond_wait + + def releasing_cond_wait(condition, timeout): + if condition is pool.size_cond: + pool.requests = 0 + return True + return real_cond_wait(condition, timeout) + + acquires = 0 + real_lock = pool.lock + + class CountingLock: + def __enter__(self): + nonlocal acquires + acquires += 1 + return real_lock.__enter__() + + def __exit__(self, *args): + return real_lock.__exit__(*args) + + def __getattr__(self, name): + return getattr(real_lock, name) + + pool.lock = CountingLock() # type: ignore[assignment] + try: + with patch.object(pool_module, "_cond_wait", releasing_cond_wait): + with pool.checkout(): + checkout_acquires = acquires + finally: + pool.lock = real_lock + + self.assertEqual( + 2, + checkout_acquires, + f"a checkout that waited for a slot should acquire the pool lock twice -- " + f"once for counter bookkeeping and once to register the cancel context -- " + f"got {checkout_acquires}", + ) + def test_no_wait_queue_timeout(self): # Verify get_socket() with no wait_queue_timeout blocks forever. pool = self.create_pool(max_pool_size=1)