diff --git a/doc/changelog.rst b/doc/changelog.rst index cda288c575..9bd52fdd71 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -27,6 +27,23 @@ PyMongo 4.18 brings a number of changes including: buffer. - Fixed :func:`bson.json_util.loads` to reject ``$timestamp`` values containing fields other than ``t`` and ``i``. +- Fixed ``client.primary`` raising ``IndexError``, and ``client.secondaries`` + and ``client.arbiters`` returning stale results, after a retryable operation + deprioritized the primary. +- **Breaking change**: removed the public ``TopologyDescription.candidate_servers`` + attribute, which was added in PyMongo 4.16.0 and appeared in the 4.16 and + 4.17 API documentation. Its value depended on which server-selection call + happened to run last, so it could not be relied on. Any code reading it must + be updated to use + :attr:`~pymongo.topology_description.TopologyDescription.known_servers` + instead. +- Fixed a leak where every failed connection checkout permanently incremented + a pool's ``operation_count``. Because nothing short of a fork reset that + counter, a mongos that suffered a burst of checkout failures looked + permanently busier than its peers and was progressively avoided by server + selection for the remaining life of the client. +- Reduced the number of lock acquisitions on the connection checkout fast + path. Changes in Version 4.17.0 (2026/04/20) -------------------------------------- diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py index fdf3b1d816..5478dcac02 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,62 @@ async def _get_conn( else: deadline = None - async with self.size_cond: - self._raise_if_not_ready(checkout_started_time, emit_event=True) - while not (self.requests < self.max_pool_size): - timeout = deadline - time.monotonic() if deadline else None - if not await _async_cond_wait(self.size_cond, timeout): - # Timed out, notify the next thread to ensure a - # timeout doesn't consume the condition. - if self.requests < self.max_pool_size: - self.size_cond.notify() - self._raise_wait_queue_timeout(checkout_started_time) - self._raise_if_not_ready(checkout_started_time, emit_event=True) - self.requests += 1 - - # We've now acquired the semaphore and must release it on error. conn = None - incremented = False + op_count_incremented = False + # Each flag is set immediately after its own increment, with no + # statement in between, which narrows the window in which an interrupt + # could leave a counter incremented but unrecorded to a single + # statement boundary. It does not close the window entirely: the + # increment and the flag assignment are still separate statements. + requests_incremented = False + active_sockets_incremented = False + # Invariant: any site inside the `try` below that emits a checkout + # failed event must set this so the outer handler does not re-emit. emitted_event = False is_new_conn = False + try: async with self.lock: - self.active_sockets += 1 - incremented = True + self.operation_count += 1 + op_count_incremented = True + emitted_event = True + self._raise_if_not_ready(checkout_started_time, emit_event=True) + emitted_event = False + if self.requests < self.max_pool_size: + # Fast path: a slot is immediately available, so do all + # of the counter bookkeeping in this one critical section + # and keep the checkout to a single lock acquisition. + self.requests += 1 + requests_incremented = True + self.active_sockets += 1 + active_sockets_incremented = True + + if not requests_incremented: + # Slow path: no slot was free under the pool mutex. Re-check + # under size_cond -- a slot may have been released in the + # meantime, in which case the loop below never waits -- and + # otherwise wait for one to be released. + async with self.size_cond: + emitted_event = True + self._raise_if_not_ready(checkout_started_time, emit_event=True) + emitted_event = False + while not (self.requests < self.max_pool_size): + timeout = deadline - time.monotonic() if deadline else None + if not await _async_cond_wait(self.size_cond, timeout): + # Timed out, notify the next thread to ensure a + # timeout doesn't consume the condition. + if self.requests < self.max_pool_size: + self.size_cond.notify() + emitted_event = True + self._raise_wait_queue_timeout(checkout_started_time) + emitted_event = True + self._raise_if_not_ready(checkout_started_time, emit_event=True) + emitted_event = False + self.requests += 1 + requests_incremented = True + self.active_sockets += 1 + active_sockets_incremented = True + while conn is None: # CMAP: we MUST wait for either maxConnecting OR for a socket # to be checked back into the pool. @@ -1084,11 +1115,17 @@ async def _get_conn( if conn: # We checked out a socket but authentication failed. await conn.close_conn(ConnectionClosedReason.ERROR) - async with self.size_cond: - self.requests -= 1 - if incremented: - self.active_sockets -= 1 - self.size_cond.notify() + if op_count_incremented: + async with self.size_cond: + self.operation_count -= 1 + if active_sockets_incremented: + self.active_sockets -= 1 + if requests_incremented: + self.requests -= 1 + # Notify only when a slot was actually released; + # otherwise there is nothing for a blocked checkout to + # wake up for. + self.size_cond.notify() if not emitted_event: self._telemetry.checkout_failed( diff --git a/pymongo/server_selectors.py b/pymongo/server_selectors.py index b43272f7f3..74c10ff431 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 all known servers. Server selection passes its own (per-call, + deprioritization-filtered) list here so that no state has to be + cached on the TopologyDescription. + """ + if candidate_servers is None: + candidate_servers = topology_description.known_servers primary = None for sd in candidate_servers: if sd.server_type == SERVER_TYPE.RSPrimary: @@ -45,7 +58,7 @@ def from_topology_description(cls, topology_description: TopologyDescription) -> return Selection( topology_description, - topology_description.candidate_servers, + candidate_servers, topology_description.common_wire_version, primary, ) diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py index 1304921781..21f8ced18a 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,62 @@ def _get_conn( else: deadline = None - with self.size_cond: - self._raise_if_not_ready(checkout_started_time, emit_event=True) - while not (self.requests < self.max_pool_size): - timeout = deadline - time.monotonic() if deadline else None - if not _cond_wait(self.size_cond, timeout): - # Timed out, notify the next thread to ensure a - # timeout doesn't consume the condition. - if self.requests < self.max_pool_size: - self.size_cond.notify() - self._raise_wait_queue_timeout(checkout_started_time) - self._raise_if_not_ready(checkout_started_time, emit_event=True) - self.requests += 1 - - # We've now acquired the semaphore and must release it on error. conn = None - incremented = False + op_count_incremented = False + # Each flag is set immediately after its own increment, with no + # statement in between, which narrows the window in which an interrupt + # could leave a counter incremented but unrecorded to a single + # statement boundary. It does not close the window entirely: the + # increment and the flag assignment are still separate statements. + requests_incremented = False + active_sockets_incremented = False + # Invariant: any site inside the `try` below that emits a checkout + # failed event must set this so the outer handler does not re-emit. emitted_event = False is_new_conn = False + try: with self.lock: - self.active_sockets += 1 - incremented = True + self.operation_count += 1 + op_count_incremented = True + emitted_event = True + self._raise_if_not_ready(checkout_started_time, emit_event=True) + emitted_event = False + if self.requests < self.max_pool_size: + # Fast path: a slot is immediately available, so do all + # of the counter bookkeeping in this one critical section + # and keep the checkout to a single lock acquisition. + self.requests += 1 + requests_incremented = True + self.active_sockets += 1 + active_sockets_incremented = True + + if not requests_incremented: + # Slow path: no slot was free under the pool mutex. Re-check + # under size_cond -- a slot may have been released in the + # meantime, in which case the loop below never waits -- and + # otherwise wait for one to be released. + with self.size_cond: + emitted_event = True + self._raise_if_not_ready(checkout_started_time, emit_event=True) + emitted_event = False + while not (self.requests < self.max_pool_size): + timeout = deadline - time.monotonic() if deadline else None + if not _cond_wait(self.size_cond, timeout): + # Timed out, notify the next thread to ensure a + # timeout doesn't consume the condition. + if self.requests < self.max_pool_size: + self.size_cond.notify() + emitted_event = True + self._raise_wait_queue_timeout(checkout_started_time) + emitted_event = True + self._raise_if_not_ready(checkout_started_time, emit_event=True) + emitted_event = False + self.requests += 1 + requests_incremented = True + self.active_sockets += 1 + active_sockets_incremented = True + while conn is None: # CMAP: we MUST wait for either maxConnecting OR for a socket # to be checked back into the pool. @@ -1080,11 +1111,17 @@ def _get_conn( if conn: # We checked out a socket but authentication failed. conn.close_conn(ConnectionClosedReason.ERROR) - with self.size_cond: - self.requests -= 1 - if incremented: - self.active_sockets -= 1 - self.size_cond.notify() + if op_count_incremented: + with self.size_cond: + self.operation_count -= 1 + if active_sockets_incremented: + self.active_sockets -= 1 + if requests_incremented: + self.requests -= 1 + # Notify only when a slot was actually released; + # otherwise there is nothing for a blocked checkout to + # wake up for. + self.size_cond.notify() if not emitted_event: self._telemetry.checkout_failed( diff --git a/pymongo/topology_description.py b/pymongo/topology_description.py index 87966aca45..32d27fedb4 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 @@ -238,11 +237,6 @@ def readable_servers(self) -> list[ServerDescription]: """List of readable Servers.""" return [s for s in self._server_descriptions.values() if s.is_readable] - @property - def candidate_servers(self) -> list[ServerDescription]: - """List of Servers excluding deprioritized servers.""" - return self._candidate_servers - @property def common_wire_version(self) -> Optional[int]: """Minimum of all servers' max wire versions, or None.""" @@ -280,18 +274,34 @@ def _apply_local_threshold(self, selection: Optional[Selection]) -> list[ServerD def _filter_servers( self, deprioritized_servers: Optional[list[ServerDescription]] = None - ) -> None: - """Filter out deprioritized servers from a list of server candidates.""" + ) -> list[ServerDescription]: + """Return the known servers with any deprioritized servers filtered out. + + If every known server is deprioritized, all known servers are returned + so that selection can still make progress. + + This method must not mutate ``self``. Deprioritization is specific to + a single call, but a TopologyDescription outlives it and is replaced, + not edited, when the topology changes. Filtering left behind on the + description would therefore leak into later readers: callers such as + :meth:`~pymongo.synchronous.topology.Topology.get_primary` build their + selection straight from the description and would see a view narrowed + by a selection they never asked for. The same object is also handed to + code running on other threads -- topology event listeners and anything + holding :attr:`~pymongo.mongo_client.MongoClient.topology_description` + -- which may read it concurrently with a selection in progress. + + :param deprioritized_servers: servers to exclude, or None. + """ + known_servers = self.known_servers if not deprioritized_servers: - self._candidate_servers = self.known_servers - else: - deprioritized_addresses = {sd.address for sd in deprioritized_servers} - filtered = [ - server - for server in self.known_servers - if server.address not in deprioritized_addresses - ] - self._candidate_servers = filtered or self.known_servers + return known_servers + + deprioritized_addresses = {sd.address for sd in deprioritized_servers} + filtered = [ + server for server in known_servers if server.address not in deprioritized_addresses + ] + return filtered or known_servers def apply_selector( self, @@ -335,10 +345,10 @@ def apply_selector( description = self.server_descriptions().get(address) return [description] if description and description.is_server_type_known else [] - self._filter_servers(deprioritized_servers) + candidate_servers = self._filter_servers(deprioritized_servers) # Primary selection fast path. if self.topology_type == TOPOLOGY_TYPE.ReplicaSetWithPrimary and type(selector) is Primary: - for sd in self._candidate_servers: + for sd in candidate_servers: if sd.server_type == SERVER_TYPE.RSPrimary: sds = [sd] if custom_selector: @@ -355,13 +365,14 @@ def apply_selector( # No primary found, return an empty list. return [] - selection = Selection.from_topology_description(self) + selection = Selection.from_topology_description(self, candidate_servers) # Ignore read preference for sharded clusters. if self.topology_type != TOPOLOGY_TYPE.Sharded: selection = selector(selection) # No suitable servers found, apply preference again but include deprioritized servers. if not selection and deprioritized_servers: - self._filter_servers(None) + # No candidate filtering: from_topology_description() defaults + # to all known servers. selection = Selection.from_topology_description(self) selection = selector(selection) diff --git a/test/asynchronous/test_pooling.py b/test/asynchronous/test_pooling.py index 063f5f06ec..96a24b4ef2 100644 --- a/test/asynchronous/test_pooling.py +++ b/test/asynchronous/test_pooling.py @@ -32,12 +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.monitoring import _EventListeners +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 @@ -276,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. @@ -399,6 +406,288 @@ 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): + # 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( + 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 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): + # 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) + + 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_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_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_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_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 64146a0e13..2829fe6ef3 100644 --- a/test/test_pooling.py +++ b/test/test_pooling.py @@ -32,13 +32,19 @@ 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.monitoring import _EventListeners +from pymongo.lock import _cond_wait, _create_lock +from pymongo.monitoring import ( + ConnectionCheckOutFailedEvent, + ConnectionCheckOutFailedReason, + PoolClearedEvent, + _EventListeners, +) from test.utils import flaky, get_pool, joinall 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 @@ -276,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. @@ -399,6 +406,288 @@ 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): + # 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( + 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 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): + # 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) + + 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_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_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_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_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) diff --git a/test/test_topology.py b/test/test_topology.py index 47e670b20d..c2772dd87b 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,176 @@ def test_no_mongoses(self): self.assertMessage("No mongoses available", t) +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), + { + "ok": 1, + HelloCompat.LEGACY_CMD: True, + "setName": "rs", + "maxWireVersion": common.MIN_SUPPORTED_WIRE_VERSION, + **members, + }, + ) + for host in hosts[1:]: + got_hello( + t, + (host, 27017), + { + "ok": 1, + HelloCompat.LEGACY_CMD: False, + "secondary": True, + "setName": "rs", + "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 + + +class TestTopologyDescriptionImmutability(TopologyTest): + """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): + 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(): + # 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(): + 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)] + for thread in threads: + thread.start() + 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. 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): + # 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 + 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. + 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(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 and the + # arbiter for a single selection call. + description.apply_selector( + ReadPreference.SECONDARY_PREFERRED, + deprioritized_servers=[secondary_sd, arbiter_sd], + ) + + # Both secondaries and the arbiter must still be reported afterwards. + self.assertEqual(secondaries, t.get_secondaries()) + self.assertEqual(arbiters, t.get_arbiters()) + + if __name__ == "__main__": unittest.main()