From 213fc64f62d872e42e37acb44e5995c2da07479a Mon Sep 17 00:00:00 2001 From: Junyao Dong <1071307515@qq.com> Date: Thu, 13 Aug 2026 12:39:05 +0800 Subject: [PATCH 1/2] PYTHON-4008 Support exhaust cursors on mongos 7.1+ Gate the refusal on wire version instead of on server type. mongos has served exhaust getMore since 7.1 (SERVER-57297); the categorical check dates to 2.6 (4d422586, 2013), when exhaust was an OP_QUERY flag. The check is made against the connection in use, in _Query.use_command and _GetMore.use_command, since a sharded topology can hold mongoses of different versions. --- doc/changelog.rst | 8 +++ doc/contributors.rst | 1 + pymongo/asynchronous/collection.py | 16 ++++- pymongo/asynchronous/cursor.py | 26 +++----- pymongo/common.py | 3 + pymongo/message.py | 38 ++++++------ pymongo/synchronous/collection.py | 16 ++++- pymongo/synchronous/cursor.py | 26 +++----- test/__init__.py | 14 +++++ test/asynchronous/__init__.py | 14 +++++ test/asynchronous/test_client.py | 4 +- test/asynchronous/test_collection.py | 11 +++- test/asynchronous/test_cursor.py | 8 +-- test/asynchronous/test_encryption.py | 7 +-- test/asynchronous/test_monitoring.py | 2 +- test/asynchronous/utils.py | 5 -- test/test_client.py | 4 +- test/test_collection.py | 11 +++- test/test_cursor.py | 8 +-- test/test_encryption.py | 7 +-- test/test_message.py | 90 +++++++++++++++++++++++++++- test/test_monitoring.py | 2 +- test/utils.py | 5 -- tools/synchro.py | 1 - 24 files changed, 226 insertions(+), 101 deletions(-) diff --git a/doc/changelog.rst b/doc/changelog.rst index 6b6c379eba..6b3259f89d 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -11,6 +11,14 @@ PyMongo 4.18 brings a number of changes including: Session resumption is supported on all Python versions for synchronous clients and on Python 3.11+ for async clients. - Improved performance for MongoDB 9.0's Intelligent Workload Management (IWM) by only retrying overload errors when doing so is expected to not worsen server conditions. +- Added support for exhaust cursors (:attr:`~pymongo.cursor.CursorType.EXHAUST`) + against mongos 7.1+. An older mongos still raises + :class:`~pymongo.errors.InvalidOperation`, now on the first iteration of the + cursor rather than from + :meth:`~pymongo.synchronous.collection.Collection.find`, since the requirement + is checked against the connection in use. Separately, async cursors combining + ``limit`` with :attr:`~pymongo.cursor.CursorType.EXHAUST` now raise at ``find`` + rather than on first iteration, matching the synchronous API. - Redacted potentially sensitive authentication mechanism properties, including AWS session tokens, from the representations of :class:`~pymongo.synchronous.mongo_client.MongoClient` and diff --git a/doc/contributors.rst b/doc/contributors.rst index 6c2cff68a3..4caa90eebe 100644 --- a/doc/contributors.rst +++ b/doc/contributors.rst @@ -110,3 +110,4 @@ The following is a list of people who have contributed to - Cal Jacobson (cj81499) - Sophia Yang (sophiayangDB) - Madan Kumar (winklemad) +- Junyao Dong (carsontung666) diff --git a/pymongo/asynchronous/collection.py b/pymongo/asynchronous/collection.py index ddffc46632..ff2df1be82 100644 --- a/pymongo/asynchronous/collection.py +++ b/pymongo/asynchronous/collection.py @@ -1881,8 +1881,14 @@ def find(self, *args: Any, **kwargs: Any) -> AsyncCursor[_DocumentType]: - The `limit` option can not be used with an exhaust cursor. - - Exhaust cursors are not supported by mongos and can not be - used with a sharded cluster. + - When connected to mongos, exhaust cursors require MongoDB 7.1 or + newer; an older one raises + :class:`~pymongo.errors.InvalidOperation` on the first iteration. + On a cluster midway through an upgrade the same call may succeed + or raise depending on which mongos it reaches. Behind a load + balancer no version check applies; a pre-7.1 mongos there ignores + the exhaust request and the cursor falls back to ordinary getMore + batches. - A :class:`~pymongo.cursor.AsyncCursor` instance created with the :attr:`~pymongo.cursor.CursorType.EXHAUST` cursor_type requires an @@ -1892,6 +1898,12 @@ def find(self, *args: Any, **kwargs: Any) -> AsyncCursor[_DocumentType]: connection will be closed and discarded without being returned to the connection pool. + .. versionchanged:: 4.18 + :attr:`~pymongo.cursor.CursorType.EXHAUST` is now permitted when + connected to mongos 7.1 or newer. Against an older mongos the + resulting :class:`~pymongo.errors.InvalidOperation` is now raised on + the first iteration of the cursor rather than by this method. + .. versionchanged:: 4.0 Removed the ``modifiers`` option. Empty projections (eg {} or []) are passed to the server as-is, diff --git a/pymongo/asynchronous/cursor.py b/pymongo/asynchronous/cursor.py index 51c00a0b36..1b2428a6d8 100644 --- a/pymongo/asynchronous/cursor.py +++ b/pymongo/asynchronous/cursor.py @@ -213,19 +213,15 @@ def __init__( self._dbname = collection.database.name self._collname = collection.name - # Checking exhaust cursor support requires network IO - if _IS_SYNC: - self._exhaust_checked = True - self._supports_exhaust() # type: ignore[unused-coroutine] - else: - self._exhaust = cursor_type == CursorType.EXHAUST - self._exhaust_checked = False + self._validate_exhaust_handling() - async def _supports_exhaust(self) -> None: - # Exhaust cursor support + def _validate_exhaust_handling(self) -> None: + """Reject option combinations an exhaust cursor cannot serve. + + Server support is checked against the connection in use, in + _check_exhaust_supported. + """ if self._cursor_type == CursorType.EXHAUST: - if await self._collection.database.client.is_mongos: - raise InvalidOperation("Exhaust cursors are not supported by mongos") if self._limit: raise InvalidOperation("Can't use limit and exhaust together.") self._exhaust = True @@ -369,8 +365,6 @@ async def add_option(self, mask: int) -> AsyncCursor[_DocumentType]: if mask & _QUERY_OPTIONS["exhaust"]: if self._limit: raise InvalidOperation("Can't use limit and exhaust together.") - if await self._collection.database.client.is_mongos: - raise InvalidOperation("Exhaust cursors are not supported by mongos") self._exhaust = True self._query_flags |= mask @@ -1125,9 +1119,6 @@ async def rewind(self) -> AsyncCursor[_DocumentType]: async def next(self) -> _DocumentType: """Advance the cursor.""" - if not self._exhaust_checked: - self._exhaust_checked = True - await self._supports_exhaust() if self._empty: raise StopAsyncIteration if len(self._data) or await self._refresh(): @@ -1137,9 +1128,6 @@ async def next(self) -> _DocumentType: async def _next_batch(self, result: list, total: Optional[int] = None) -> bool: # type: ignore[type-arg] """Get all or some documents from the cursor.""" - if not self._exhaust_checked: - self._exhaust_checked = True - await self._supports_exhaust() if self._empty: return False if len(self._data) or await self._refresh(): diff --git a/pymongo/common.py b/pymongo/common.py index 57606a6317..174d99c512 100644 --- a/pymongo/common.py +++ b/pymongo/common.py @@ -69,6 +69,9 @@ # MongoDB 9.0 MAX_SUPPORTED_WIRE_VERSION = 29 +# MongoDB 7.1, the first release whose mongos continues an exhaust getMore stream. +MONGOS_EXHAUST_WIRE_VERSION = 22 + # Frequency to call hello on servers, in seconds. HEARTBEAT_FREQUENCY = 10 diff --git a/pymongo/message.py b/pymongo/message.py index a41dab2506..b696ba2302 100644 --- a/pymongo/message.py +++ b/pymongo/message.py @@ -43,6 +43,7 @@ RawBSONDocument, _inflate_bson, ) +from pymongo.common import MONGOS_EXHAUST_WIRE_VERSION from pymongo.monitoring import _EventListeners try: @@ -52,7 +53,6 @@ except ImportError: _use_c = False from pymongo.errors import ( - ConfigurationError, DocumentTooLarge, InvalidOperation, ProtocolError, @@ -1220,6 +1220,17 @@ def unpack(cls, msg: bytes | memoryview) -> _OpMsg: } +def _check_exhaust_supported(conn: _AgnosticConnection) -> None: + """Raise if this connection's server will not continue an exhaust stream. + + mongos gained exhaust getMore in 7.1 (SERVER-57297). Load-balanced deployments + are left alone: every cursor pins its connection there anyway, so an older mongos + costs nothing extra, and refusing would break clients that work today. + """ + if conn.is_mongos and conn.max_wire_version < MONGOS_EXHAUST_WIRE_VERSION: + raise InvalidOperation("Exhaust cursors require MongoDB 7.1+ when connected to mongos.") + + class _Query: """A query operation.""" @@ -1293,20 +1304,11 @@ def namespace(self) -> str: return f"{self.db}.{self.coll}" def use_command(self, conn: _AgnosticConnection) -> bool: - use_find_cmd = False - if not self.exhaust: - use_find_cmd = True - elif conn.max_wire_version >= 8: - # OP_MSG supports exhaust on MongoDB 4.2+ - use_find_cmd = True - elif not self.read_concern.ok_for_legacy: - raise ConfigurationError( - f"read concern level of {self.read_concern.level} is not valid " - f"with a max wire version of {conn.max_wire_version}." - ) + if self.exhaust: + _check_exhaust_supported(conn) conn.validate_session(self.client, self.session) # type: ignore[arg-type] - return use_find_cmd + return True def update_command(self, cmd: dict[str, Any]) -> None: self._as_command = cmd, self.db @@ -1426,15 +1428,11 @@ def namespace(self) -> str: return f"{self.db}.{self.coll}" def use_command(self, conn: _AgnosticConnection) -> bool: - use_cmd = False - if not self.exhaust: - use_cmd = True - elif conn.max_wire_version >= 8: - # OP_MSG supports exhaust on MongoDB 4.2+ - use_cmd = True + if self.exhaust: + _check_exhaust_supported(conn) conn.validate_session(self.client, self.session) # type: ignore[arg-type] - return use_cmd + return True def update_command(self, cmd: dict[str, Any]) -> None: self._as_command = cmd, self.db diff --git a/pymongo/synchronous/collection.py b/pymongo/synchronous/collection.py index 15a38b2f3e..3051ac9839 100644 --- a/pymongo/synchronous/collection.py +++ b/pymongo/synchronous/collection.py @@ -1881,8 +1881,14 @@ def find(self, *args: Any, **kwargs: Any) -> Cursor[_DocumentType]: - The `limit` option can not be used with an exhaust cursor. - - Exhaust cursors are not supported by mongos and can not be - used with a sharded cluster. + - When connected to mongos, exhaust cursors require MongoDB 7.1 or + newer; an older one raises + :class:`~pymongo.errors.InvalidOperation` on the first iteration. + On a cluster midway through an upgrade the same call may succeed + or raise depending on which mongos it reaches. Behind a load + balancer no version check applies; a pre-7.1 mongos there ignores + the exhaust request and the cursor falls back to ordinary getMore + batches. - A :class:`~pymongo.cursor.Cursor` instance created with the :attr:`~pymongo.cursor.CursorType.EXHAUST` cursor_type requires an @@ -1892,6 +1898,12 @@ def find(self, *args: Any, **kwargs: Any) -> Cursor[_DocumentType]: connection will be closed and discarded without being returned to the connection pool. + .. versionchanged:: 4.18 + :attr:`~pymongo.cursor.CursorType.EXHAUST` is now permitted when + connected to mongos 7.1 or newer. Against an older mongos the + resulting :class:`~pymongo.errors.InvalidOperation` is now raised on + the first iteration of the cursor rather than by this method. + .. versionchanged:: 4.0 Removed the ``modifiers`` option. Empty projections (eg {} or []) are passed to the server as-is, diff --git a/pymongo/synchronous/cursor.py b/pymongo/synchronous/cursor.py index 7ea81ea9ac..fddbcc520f 100644 --- a/pymongo/synchronous/cursor.py +++ b/pymongo/synchronous/cursor.py @@ -213,19 +213,15 @@ def __init__( self._dbname = collection.database.name self._collname = collection.name - # Checking exhaust cursor support requires network IO - if _IS_SYNC: - self._exhaust_checked = True - self._supports_exhaust() # type: ignore[unused-coroutine] - else: - self._exhaust = cursor_type == CursorType.EXHAUST - self._exhaust_checked = False + self._validate_exhaust_handling() - def _supports_exhaust(self) -> None: - # Exhaust cursor support + def _validate_exhaust_handling(self) -> None: + """Reject option combinations an exhaust cursor cannot serve. + + Server support is checked against the connection in use, in + _check_exhaust_supported. + """ if self._cursor_type == CursorType.EXHAUST: - if self._collection.database.client.is_mongos: - raise InvalidOperation("Exhaust cursors are not supported by mongos") if self._limit: raise InvalidOperation("Can't use limit and exhaust together.") self._exhaust = True @@ -369,8 +365,6 @@ def add_option(self, mask: int) -> Cursor[_DocumentType]: if mask & _QUERY_OPTIONS["exhaust"]: if self._limit: raise InvalidOperation("Can't use limit and exhaust together.") - if self._collection.database.client.is_mongos: - raise InvalidOperation("Exhaust cursors are not supported by mongos") self._exhaust = True self._query_flags |= mask @@ -1121,9 +1115,6 @@ def rewind(self) -> Cursor[_DocumentType]: def next(self) -> _DocumentType: """Advance the cursor.""" - if not self._exhaust_checked: - self._exhaust_checked = True - self._supports_exhaust() if self._empty: raise StopIteration if len(self._data) or self._refresh(): @@ -1133,9 +1124,6 @@ def next(self) -> _DocumentType: def _next_batch(self, result: list, total: Optional[int] = None) -> bool: # type: ignore[type-arg] """Get all or some documents from the cursor.""" - if not self._exhaust_checked: - self._exhaust_checked = True - self._supports_exhaust() if self._empty: return False if len(self._data) or self._refresh(): diff --git a/test/__init__.py b/test/__init__.py index f4ae7fe948..6aa7454f07 100644 --- a/test/__init__.py +++ b/test/__init__.py @@ -763,6 +763,20 @@ def require_retryable_writes(self, func): func=func, ) + def supports_exhaust_cursors(self): + """Whether this deployment supports exhaust cursors.""" + if self.is_mongos: + return self.version.at_least(7, 1) + return True + + def require_exhaust_cursors(self, func): + """Run a test only if the deployment supports exhaust cursors.""" + return self._require( + self.supports_exhaust_cursors, + "This server does not support exhaust cursors", + func=func, + ) + def supports_transactions(self): if self.version.at_least(4, 1, 8): return self.is_mongos or self.is_rs diff --git a/test/asynchronous/__init__.py b/test/asynchronous/__init__.py index 0699451d7e..4e361c1b1e 100644 --- a/test/asynchronous/__init__.py +++ b/test/asynchronous/__init__.py @@ -763,6 +763,20 @@ def require_retryable_writes(self, func): func=func, ) + def supports_exhaust_cursors(self): + """Whether this deployment supports exhaust cursors.""" + if self.is_mongos: + return self.version.at_least(7, 1) + return True + + def require_exhaust_cursors(self, func): + """Run a test only if the deployment supports exhaust cursors.""" + return self._require( + self.supports_exhaust_cursors, + "This server does not support exhaust cursors", + func=func, + ) + def supports_transactions(self): if self.version.at_least(4, 1, 8): return self.is_mongos or self.is_rs diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py index 5606c7f8c1..92da29145c 100644 --- a/test/asynchronous/test_client.py +++ b/test/asynchronous/test_client.py @@ -2417,8 +2417,8 @@ class TestExhaustCursor(AsyncIntegrationTest): def setUp(self): super().setUp() - if async_client_context.is_mongos: - raise SkipTest("mongos doesn't support exhaust, SERVER-2627") + if not async_client_context.supports_exhaust_cursors(): + raise SkipTest("mongos serves exhaust cursors only from 7.1, SERVER-57297") async def test_exhaust_query_server_error(self): # When doing an exhaust query, the socket stays checked out on success diff --git a/test/asynchronous/test_collection.py b/test/asynchronous/test_collection.py index 568d5a32e0..d787080d02 100644 --- a/test/asynchronous/test_collection.py +++ b/test/asynchronous/test_collection.py @@ -27,7 +27,7 @@ from pymongo.asynchronous.database import AsyncDatabase from pymongo.asynchronous.helpers import anext -from test.asynchronous.utils import async_get_pool, async_is_mongos +from test.asynchronous.utils import async_get_pool sys.path[0:0] = [""] @@ -1796,8 +1796,15 @@ async def test_cursor_timeout(self): await self.db.test.find(no_cursor_timeout=True).to_list() await self.db.test.find(no_cursor_timeout=False).to_list() + async def test_exhaust_limit_raises_without_iterating(self): + # The limit conflict is settled at find(); the mongos wire version is not. + with self.assertRaises(InvalidOperation): + self.db.test.find(cursor_type=CursorType.EXHAUST, limit=5) + self.db.test.find(cursor_type=CursorType.EXHAUST) + async def test_exhaust(self): - if await async_is_mongos(self.db.client): + # mongos only serves exhaust cursors from 7.1 onwards (SERVER-57297). + if not async_client_context.supports_exhaust_cursors(): with self.assertRaises(InvalidOperation): await anext(self.db.test.find(cursor_type=CursorType.EXHAUST)) return diff --git a/test/asynchronous/test_cursor.py b/test/asynchronous/test_cursor.py index 354a975d19..1fe023720a 100644 --- a/test/asynchronous/test_cursor.py +++ b/test/asynchronous/test_cursor.py @@ -130,8 +130,8 @@ async def test_add_remove_option(self): self.assertEqual(0, cursor._query_flags) async def test_add_remove_option_exhaust(self): - # Exhaust - which mongos doesn't support - if async_client_context.is_mongos: + # mongos only serves exhaust cursors from 7.1 onwards (SERVER-57297). + if not async_client_context.supports_exhaust_cursors(): with self.assertRaises(InvalidOperation): await anext(self.db.test.find(cursor_type=CursorType.EXHAUST)) else: @@ -1615,7 +1615,7 @@ async def test_clone(self): self.assertIsInstance(await anext(cursor.clone()), bytes) self.assertIsInstance(await anext(copy.copy(cursor)), bytes) - @async_client_context.require_no_mongos + @async_client_context.require_exhaust_cursors async def test_exhaust(self): c = self.db.test await c.insert_many({"_id": i} for i in range(200)) @@ -1849,7 +1849,7 @@ async def test_monitoring(self): listener.reset() @async_client_context.require_version_min(5, 0, -1) - @async_client_context.require_no_mongos + @async_client_context.require_exhaust_cursors @async_client_context.require_sync async def test_exhaust_cursor_db_set(self): listener = OvertCommandListener() diff --git a/test/asynchronous/test_encryption.py b/test/asynchronous/test_encryption.py index bbd324df32..daeb18607a 100644 --- a/test/asynchronous/test_encryption.py +++ b/test/asynchronous/test_encryption.py @@ -461,10 +461,9 @@ async def test_raise_unsupported_error(self): with self.assertRaisesRegex(InvalidOperation, msg): await client.test.test.aggregate_raw_batches([]) - if async_client_context.is_mongos: - msg = "Exhaust cursors are not supported by mongos" - else: - msg = "exhaust cursors do not support auto encryption" + # The auto-encryption guard runs at cursor iteration, before the wire-version + # check in _Query.use_command, so it is the error regardless of deployment. + msg = "exhaust cursors do not support auto encryption" with self.assertRaisesRegex(InvalidOperation, msg): await anext(client.test.test.find(cursor_type=CursorType.EXHAUST)) diff --git a/test/asynchronous/test_monitoring.py b/test/asynchronous/test_monitoring.py index 19dc1abcc7..7d6075ad48 100644 --- a/test/asynchronous/test_monitoring.py +++ b/test/asynchronous/test_monitoring.py @@ -422,7 +422,7 @@ async def test_not_primary_error(self): self.assertIsInstance(failed.duration_micros, int) self.assertEqual(error, failed.failure) - @async_client_context.require_no_mongos + @async_client_context.require_exhaust_cursors async def test_exhaust(self): await self.client.pymongo_test.test.drop() await self.client.pymongo_test.test.insert_many([{} for _ in range(11)]) diff --git a/test/asynchronous/utils.py b/test/asynchronous/utils.py index 32e8c89502..2b59d151eb 100644 --- a/test/asynchronous/utils.py +++ b/test/asynchronous/utils.py @@ -82,11 +82,6 @@ async def async_wait_until(predicate, success_description, timeout=10): await asyncio.sleep(interval) -async def async_is_mongos(client): - res = await client.admin.command(HelloCompat.LEGACY_CMD) - return res.get("msg", "") == "isdbgrid" - - async def async_ensure_all_connected(client: AsyncMongoClient) -> None: """Ensure that the client's connection pool has socket connections to all members of a replica set. Raises ConfigurationError when called with a diff --git a/test/test_client.py b/test/test_client.py index 9df392e64f..e0a3b3dcfe 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -2370,8 +2370,8 @@ class TestExhaustCursor(IntegrationTest): def setUp(self): super().setUp() - if client_context.is_mongos: - raise SkipTest("mongos doesn't support exhaust, SERVER-2627") + if not client_context.supports_exhaust_cursors(): + raise SkipTest("mongos serves exhaust cursors only from 7.1, SERVER-57297") def test_exhaust_query_server_error(self): # When doing an exhaust query, the socket stays checked out on success diff --git a/test/test_collection.py b/test/test_collection.py index 0b7080d115..7a52f94081 100644 --- a/test/test_collection.py +++ b/test/test_collection.py @@ -27,7 +27,7 @@ from pymongo.synchronous.database import Database from pymongo.synchronous.helpers import next -from test.utils import get_pool, is_mongos +from test.utils import get_pool sys.path[0:0] = [""] @@ -1778,8 +1778,15 @@ def test_cursor_timeout(self): self.db.test.find(no_cursor_timeout=True).to_list() self.db.test.find(no_cursor_timeout=False).to_list() + def test_exhaust_limit_raises_without_iterating(self): + # The limit conflict is settled at find(); the mongos wire version is not. + with self.assertRaises(InvalidOperation): + self.db.test.find(cursor_type=CursorType.EXHAUST, limit=5) + self.db.test.find(cursor_type=CursorType.EXHAUST) + def test_exhaust(self): - if is_mongos(self.db.client): + # mongos only serves exhaust cursors from 7.1 onwards (SERVER-57297). + if not client_context.supports_exhaust_cursors(): with self.assertRaises(InvalidOperation): next(self.db.test.find(cursor_type=CursorType.EXHAUST)) return diff --git a/test/test_cursor.py b/test/test_cursor.py index 8f32554624..58aae5f0f9 100644 --- a/test/test_cursor.py +++ b/test/test_cursor.py @@ -130,8 +130,8 @@ def test_add_remove_option(self): self.assertEqual(0, cursor._query_flags) def test_add_remove_option_exhaust(self): - # Exhaust - which mongos doesn't support - if client_context.is_mongos: + # mongos only serves exhaust cursors from 7.1 onwards (SERVER-57297). + if not client_context.supports_exhaust_cursors(): with self.assertRaises(InvalidOperation): next(self.db.test.find(cursor_type=CursorType.EXHAUST)) else: @@ -1606,7 +1606,7 @@ def test_clone(self): self.assertIsInstance(next(cursor.clone()), bytes) self.assertIsInstance(next(copy.copy(cursor)), bytes) - @client_context.require_no_mongos + @client_context.require_exhaust_cursors def test_exhaust(self): c = self.db.test c.insert_many({"_id": i} for i in range(200)) @@ -1838,7 +1838,7 @@ def test_monitoring(self): listener.reset() @client_context.require_version_min(5, 0, -1) - @client_context.require_no_mongos + @client_context.require_exhaust_cursors @client_context.require_sync def test_exhaust_cursor_db_set(self): listener = OvertCommandListener() diff --git a/test/test_encryption.py b/test/test_encryption.py index e567826f2a..744db01b1b 100644 --- a/test/test_encryption.py +++ b/test/test_encryption.py @@ -461,10 +461,9 @@ def test_raise_unsupported_error(self): with self.assertRaisesRegex(InvalidOperation, msg): client.test.test.aggregate_raw_batches([]) - if client_context.is_mongos: - msg = "Exhaust cursors are not supported by mongos" - else: - msg = "exhaust cursors do not support auto encryption" + # The auto-encryption guard runs at cursor iteration, before the wire-version + # check in _Query.use_command, so it is the error regardless of deployment. + msg = "exhaust cursors do not support auto encryption" with self.assertRaisesRegex(InvalidOperation, msg): next(client.test.test.find(cursor_type=CursorType.EXHAUST)) diff --git a/test/test_message.py b/test/test_message.py index e3094d1e27..9c11ffef5b 100644 --- a/test/test_message.py +++ b/test/test_message.py @@ -24,19 +24,26 @@ sys.path[0:0] = [""] from bson import CodecOptions, encode +from bson.objectid import ObjectId +from pymongo.common import MIN_SUPPORTED_WIRE_VERSION, MONGOS_EXHAUST_WIRE_VERSION from pymongo.compression_support import ZlibContext, _have_zlib -from pymongo.errors import DocumentTooLarge, OperationFailure +from pymongo.errors import DocumentTooLarge, InvalidOperation, OperationFailure +from pymongo.hello import _get_server_type from pymongo.message import ( + _check_exhaust_supported, _convert_client_bulk_exception, _convert_exception, _gen_find_command, _gen_get_more_command, + _GetMore, _maybe_add_read_preference, _op_msg, + _Query, _raise_document_too_large, ) from pymongo.read_concern import ReadConcern from pymongo.read_preferences import ReadPreference, SecondaryPreferred +from pymongo.server_type import SERVER_TYPE from test import unittest _OPTS = CodecOptions() @@ -44,11 +51,90 @@ class TestMessage(unittest.TestCase): # _gen_get_more_command helper - def _make_conn(self, max_wire_version=9): + def _make_conn(self, max_wire_version=9, is_mongos=False): conn = MagicMock() conn.max_wire_version = max_wire_version + conn.is_mongos = is_mongos return conn + # _check_exhaust_supported + + def test_exhaust_allowed_on_mongos_71(self): + _check_exhaust_supported(self._make_conn(MONGOS_EXHAUST_WIRE_VERSION, is_mongos=True)) + + def test_exhaust_refused_on_older_mongos(self): + with self.assertRaisesRegex(InvalidOperation, "MongoDB 7.1"): + _check_exhaust_supported( + self._make_conn(MONGOS_EXHAUST_WIRE_VERSION - 1, is_mongos=True) + ) + + def test_exhaust_not_gated_behind_a_load_balancer(self): + # A load-balanced hello carries serviceId and is typed LoadBalancer, + # not Mongos, so these connections take the non-mongos path. + doc = {"ok": 1, "msg": "isdbgrid", "serviceId": ObjectId(), "maxWireVersion": 21} + self.assertEqual(_get_server_type(doc), SERVER_TYPE.LoadBalancer) + _check_exhaust_supported(self._make_conn(MONGOS_EXHAUST_WIRE_VERSION - 1)) + + def test_exhaust_allowed_on_any_supported_non_mongos(self): + # Every other server type has served exhaust since 4.2. + _check_exhaust_supported(self._make_conn(MIN_SUPPORTED_WIRE_VERSION)) + + def _exhaust_query(self): + return _Query( + 0, + "db", + "coll", + 0, + {}, + None, + CodecOptions(), + ReadPreference.PRIMARY, + 0, + 0, + ReadConcern(), + None, + None, + MagicMock(), + None, + True, + ) + + def _exhaust_get_more(self): + return _GetMore( + "db", + "coll", + 0, + 12345, + CodecOptions(), + ReadPreference.PRIMARY, + None, + MagicMock(), + None, + MagicMock(), + True, + None, + ) + + # use_command is where the check is consulted. + + def test_find_on_older_mongos_refuses_exhaust(self): + conn = self._make_conn(MONGOS_EXHAUST_WIRE_VERSION - 1, is_mongos=True) + with self.assertRaisesRegex(InvalidOperation, "MongoDB 7.1"): + self._exhaust_query().use_command(conn) + + def test_get_more_on_older_mongos_refuses_exhaust(self): + conn = self._make_conn(MONGOS_EXHAUST_WIRE_VERSION - 1, is_mongos=True) + with self.assertRaisesRegex(InvalidOperation, "MongoDB 7.1"): + self._exhaust_get_more().use_command(conn) + + def test_find_on_71_mongos_allows_exhaust(self): + conn = self._make_conn(MONGOS_EXHAUST_WIRE_VERSION, is_mongos=True) + self.assertTrue(self._exhaust_query().use_command(conn)) + + def test_get_more_on_71_mongos_allows_exhaust(self): + conn = self._make_conn(MONGOS_EXHAUST_WIRE_VERSION, is_mongos=True) + self.assertTrue(self._exhaust_get_more().use_command(conn)) + # _maybe_add_read_preference def test_primary_no_read_preference_added(self): diff --git a/test/test_monitoring.py b/test/test_monitoring.py index cb9417787b..ec24acb4a0 100644 --- a/test/test_monitoring.py +++ b/test/test_monitoring.py @@ -420,7 +420,7 @@ def test_not_primary_error(self): self.assertIsInstance(failed.duration_micros, int) self.assertEqual(error, failed.failure) - @client_context.require_no_mongos + @client_context.require_exhaust_cursors def test_exhaust(self): self.client.pymongo_test.test.drop() self.client.pymongo_test.test.insert_many([{} for _ in range(11)]) diff --git a/test/utils.py b/test/utils.py index 2e67008431..1526e2a828 100644 --- a/test/utils.py +++ b/test/utils.py @@ -82,11 +82,6 @@ def wait_until(predicate, success_description, timeout=10): time.sleep(interval) -def is_mongos(client): - res = client.admin.command(HelloCompat.LEGACY_CMD) - return res.get("msg", "") == "isdbgrid" - - def ensure_all_connected(client: MongoClient) -> None: """Ensure that the client's connection pool has socket connections to all members of a replica set. Raises ConfigurationError when called with a diff --git a/tools/synchro.py b/tools/synchro.py index 51bef8ed1a..bebf92c005 100644 --- a/tools/synchro.py +++ b/tools/synchro.py @@ -95,7 +95,6 @@ "AsyncSpecRunner": "SpecRunner", "AsyncTransactionsBase": "TransactionsBase", "async_get_pool": "get_pool", - "async_is_mongos": "is_mongos", "async_rs_or_single_client": "rs_or_single_client", "async_rs_or_single_client_noauth": "rs_or_single_client_noauth", "async_rs_client": "rs_client", From 3cb5da1093b96509ac53d529e0076db3ec516e32 Mon Sep 17 00:00:00 2001 From: Junyao Dong <1071307515@qq.com> Date: Sat, 15 Aug 2026 12:43:39 +0800 Subject: [PATCH 2/2] PYTHON-4008 Run the exhaust tests in the load balancer suite --- test/__init__.py | 2 ++ test/asynchronous/__init__.py | 2 ++ test/asynchronous/test_load_balancer.py | 12 ++++++++++++ test/test_load_balancer.py | 12 ++++++++++++ 4 files changed, 28 insertions(+) diff --git a/test/__init__.py b/test/__init__.py index 6aa7454f07..2816edb835 100644 --- a/test/__init__.py +++ b/test/__init__.py @@ -765,6 +765,8 @@ def require_retryable_writes(self, func): def supports_exhaust_cursors(self): """Whether this deployment supports exhaust cursors.""" + if self.load_balancer: + return True if self.is_mongos: return self.version.at_least(7, 1) return True diff --git a/test/asynchronous/__init__.py b/test/asynchronous/__init__.py index 4e361c1b1e..5160529ddc 100644 --- a/test/asynchronous/__init__.py +++ b/test/asynchronous/__init__.py @@ -765,6 +765,8 @@ def require_retryable_writes(self, func): def supports_exhaust_cursors(self): """Whether this deployment supports exhaust cursors.""" + if self.load_balancer: + return True if self.is_mongos: return self.version.at_least(7, 1) return True diff --git a/test/asynchronous/test_load_balancer.py b/test/asynchronous/test_load_balancer.py index 15b048f5b0..bfc278f152 100644 --- a/test/asynchronous/test_load_balancer.py +++ b/test/asynchronous/test_load_balancer.py @@ -32,6 +32,7 @@ sys.path[0:0] = [""] from pymongo.asynchronous.helpers import anext +from pymongo.cursor_shared import CursorType from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest from test.asynchronous.unified_format import generate_test_classes, get_test_path from test.utils_shared import ( @@ -50,6 +51,17 @@ class TestLB(AsyncIntegrationTest): RUN_ON_LOAD_BALANCER = True + async def test_exhaust_cursor(self): + coll = self.db.test + await coll.drop() + await coll.insert_many([{} for _ in range(150)]) + pool = await async_get_pool(self.client) + n_conns = len(pool.conns) + docs = await coll.find(cursor_type=CursorType.EXHAUST, batch_size=10).to_list() + self.assertEqual(len(docs), 150) + # The pinned connection is returned once the stream is exhausted. + await async_wait_until(lambda: len(pool.conns) == n_conns, "return the exhaust connection") + async def test_connections_are_only_returned_once(self): pool = await async_get_pool(self.client) n_conns = len(pool.conns) diff --git a/test/test_load_balancer.py b/test/test_load_balancer.py index 7edaec9ebb..ab549f2288 100644 --- a/test/test_load_balancer.py +++ b/test/test_load_balancer.py @@ -31,6 +31,7 @@ sys.path[0:0] = [""] +from pymongo.cursor_shared import CursorType from pymongo.synchronous.helpers import next from test import IntegrationTest, client_context, unittest from test.unified_format import generate_test_classes, get_test_path @@ -50,6 +51,17 @@ class TestLB(IntegrationTest): RUN_ON_LOAD_BALANCER = True + def test_exhaust_cursor(self): + coll = self.db.test + coll.drop() + coll.insert_many([{} for _ in range(150)]) + pool = get_pool(self.client) + n_conns = len(pool.conns) + docs = coll.find(cursor_type=CursorType.EXHAUST, batch_size=10).to_list() + self.assertEqual(len(docs), 150) + # The pinned connection is returned once the stream is exhausted. + wait_until(lambda: len(pool.conns) == n_conns, "return the exhaust connection") + def test_connections_are_only_returned_once(self): pool = get_pool(self.client) n_conns = len(pool.conns)