From 8b3b49470c6199f52aea451517226ce43f8750a2 Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Mon, 6 Jul 2026 12:08:00 -0400 Subject: [PATCH 01/11] [SPIKE] sans-I/O: extract wire-header + SDAM error cores Move pure protocol/decision logic out of the async/sync-duplicated layer into shared sans-I/O modules, as a proof of concept for reducing unasync's footprint. - network_layer.parse_wire_header: shared wire-message framing/validation used by both receive_message and PyMongoProtocol.process_header. Fixes the missing OP_COMPRESSED length<=25 guard on the sync path (PYTHON-5906): the sync path previously raised ValueError instead of ProtocolError. - pymongo/_sdam_error.py: pure SDAM application-error decision core (decide_error_action + _SDAMAction) plus the topologyVersion staleness helpers, testable without a live server. - topology._handle_error / _is_stale_error reduced to thin executors around the shared core. - test/test_sdam_error.py: server-free unit tests for the new core. --- pymongo/_sdam_error.py | 173 +++++++++++++++++++++++++++++ pymongo/asynchronous/topology.py | 113 +++++-------------- pymongo/network_layer.py | 83 ++++++++------ pymongo/synchronous/topology.py | 113 +++++-------------- test/test_sdam_error.py | 183 +++++++++++++++++++++++++++++++ 5 files changed, 457 insertions(+), 208 deletions(-) create mode 100644 pymongo/_sdam_error.py create mode 100644 test/test_sdam_error.py diff --git a/pymongo/_sdam_error.py b/pymongo/_sdam_error.py new file mode 100644 index 0000000000..83ac66259c --- /dev/null +++ b/pymongo/_sdam_error.py @@ -0,0 +1,173 @@ +# Copyright 2026-present MongoDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Sans-I/O core for SDAM application-error handling. + +This module contains the *pure* decision logic that classifies an application +error and decides which Server Discovery And Monitoring (SDAM) actions to take, +independent of any I/O or concurrency. Because it performs no ``await`` and +acquires no locks, both the synchronous and asynchronous topologies import and +call it directly -- there is a single implementation of the SDAM error +state-transition rules, and it can be unit-tested without a live server. + +The caller (``Topology._handle_error``) is responsible only for *executing* the +returned :class:`_SDAMAction` (marking the server Unknown, resetting the pool, +requesting a check, cancelling the monitor check), each of which is inherently +I/O- or concurrency-colored and therefore stays in the topology layer. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any, NamedTuple, Optional + +from pymongo import helpers_shared +from pymongo.errors import ( + ConnectionFailure, + NetworkTimeout, + NotPrimaryError, + OperationFailure, + WaitQueueTimeoutError, + WriteError, +) + + +def error_topology_version(error: Optional[BaseException]) -> Optional[Mapping[str, Any]]: + """Extract the ``topologyVersion`` from an error's details, if present. + + Pure helper: returns the ``topologyVersion`` mapping embedded in a command + error's ``details``, or ``None`` when the error carries none. + """ + if error and hasattr(error, "details") and isinstance(error.details, dict): + return error.details.get("topologyVersion") + return None + + +def is_stale_error_topology_version( + current_tv: Optional[Mapping[str, Any]], error_tv: Optional[Mapping[str, Any]] +) -> bool: + """Return True if the error's topologyVersion is <= the current one. + + Pure comparison of two ``topologyVersion`` mappings; an error whose + topologyVersion is not newer than what we already know about the server is + stale and should be ignored. + """ + if current_tv is None or error_tv is None: + return False + if current_tv["processId"] != error_tv["processId"]: + return False + return current_tv["counter"] >= error_tv["counter"] + + +class _SDAMAction(NamedTuple): + """The set of SDAM actions to take in response to an application error. + + This is the pure, immutable result of classifying an error; it contains no + I/O. All fields default to ``False`` (i.e. "do nothing"). + """ + + # Replace the server's description with type Unknown. + mark_unknown: bool = False + # Clear the server's connection pool. + reset_pool: bool = False + # Request an immediate check of the server. + request_check: bool = False + # Cancel the in-progress monitor hello check. + cancel_check: bool = False + + +# Shared singleton for the common "ignore this error" outcome. +_NO_ACTION = _SDAMAction() + + +def decide_error_action( + error: BaseException, + *, + load_balanced: bool, + has_service_id: bool, + completed_handshake: bool, + max_wire_version: int, +) -> _SDAMAction: + """Classify an application error into the SDAM actions it requires. + + This is the sans-I/O core of ``Topology._handle_error``. It implements the + Server Discovery And Monitoring specification's application-error rules and + performs no I/O. + + :param error: The error raised by the failed operation. + :param load_balanced: Whether the client is in load-balanced mode. + :param has_service_id: Whether a service ID is known for the connection. + :param completed_handshake: Whether the connection finished its handshake + before the error occurred. + :param max_wire_version: The connection's negotiated max wire version. + + :return: An :class:`_SDAMAction` describing which state transitions the + topology should perform. Returns :data:`_NO_ACTION` when the error + should be ignored. + """ + # Ignore a handshake error if the server is behind a load balancer but the + # service ID is unknown. This indicates that the error happened when dialing + # the connection or during the MongoDB handshake, so we don't know the + # service ID to use for clearing the pool. + if load_balanced and not has_service_id and not completed_handshake: + return _NO_ACTION + + if isinstance(error, NetworkTimeout) and completed_handshake: + # The socket has been closed. Don't reset the server. SDAM Spec: "When + # an application operation fails because of any network error besides a + # socket timeout...." + return _NO_ACTION + elif isinstance(error, WriteError): + # Ignore writeErrors. + return _NO_ACTION + elif isinstance(error, (NotPrimaryError, OperationFailure)): + # As per the SDAM spec if: + # - the server sees a "not primary" error, and + # - the server is not shutting down, and + # - the server version is >= 4.2, then + # we keep the existing connection pool, but mark the server type as + # Unknown and request an immediate check of the server. Otherwise, we + # clear the connection pool, mark the server as Unknown and request an + # immediate check of the server. + if hasattr(error, "code"): + err_code = error.code + else: + # Default error code if one does not exist. + default = 10107 if isinstance(error, NotPrimaryError) else None + err_code = error.details.get("code", default) # type: ignore[union-attr] + if err_code in helpers_shared._NOT_PRIMARY_CODES: + is_shutting_down = err_code in helpers_shared._SHUTDOWN_CODES + # Mark server Unknown, clear the pool, and request check. + return _SDAMAction( + mark_unknown=not load_balanced, + reset_pool=is_shutting_down or (max_wire_version <= 7), + request_check=True, + ) + elif not completed_handshake: + # Unknown command error during the connection handshake. + return _SDAMAction(mark_unknown=not load_balanced, reset_pool=True) + return _NO_ACTION + elif isinstance(error, ConnectionFailure): + if isinstance(error, WaitQueueTimeoutError) or error.has_error_label( + "SystemOverloadedError" + ): + return _NO_ACTION + # "Client MUST replace the server's description with type Unknown ... + # MUST NOT request an immediate check of the server." When a client + # marks a server Unknown from a network error when reading or writing, + # it MUST cancel the hello check on that server and close the current + # monitoring connection. + return _SDAMAction(mark_unknown=not load_balanced, reset_pool=True, cancel_check=True) + + return _NO_ACTION diff --git a/pymongo/asynchronous/topology.py b/pymongo/asynchronous/topology.py index 6af075cdc2..ae3df920d6 100644 --- a/pymongo/asynchronous/topology.py +++ b/pymongo/asynchronous/topology.py @@ -29,21 +29,20 @@ from pathlib import Path from typing import TYPE_CHECKING, Any, Callable, Optional, cast -from pymongo import _csot, common, helpers_shared, periodic_executor +from pymongo import _csot, common, periodic_executor +from pymongo._sdam_error import ( + decide_error_action, + error_topology_version, + is_stale_error_topology_version, +) from pymongo.asynchronous.client_session import _ServerSession, _ServerSessionPool from pymongo.asynchronous.monitor import MonitorBase, SrvMonitor from pymongo.asynchronous.pool import Pool from pymongo.asynchronous.server import Server from pymongo.errors import ( - ConnectionFailure, InvalidOperation, - NetworkTimeout, - NotPrimaryError, - OperationFailure, PyMongoError, ServerSelectionTimeoutError, - WaitQueueTimeoutError, - WriteError, ) from pymongo.hello import Hello from pymongo.lock import ( @@ -849,14 +848,8 @@ def _is_stale_error(self, address: _Address, err_ctx: _ErrorContext) -> bool: return True # topologyVersion check, ignore error when cur_tv >= error_tv: - cur_tv = server.description.topology_version - error = err_ctx.error - error_tv = None - if error and hasattr(error, "details"): - if isinstance(error.details, dict): - error_tv = error.details.get("topologyVersion") - - return _is_stale_error_topology_version(cur_tv, error_tv) + error_tv = error_topology_version(err_ctx.error) + return is_stale_error_topology_version(server.description.topology_version, error_tv) async def _handle_error(self, address: _Address, err_ctx: _ErrorContext) -> None: if self._is_stale_error(address, err_ctx): @@ -864,68 +857,25 @@ async def _handle_error(self, address: _Address, err_ctx: _ErrorContext) -> None server = self._servers[address] error = err_ctx.error - service_id = err_ctx.service_id - - # Ignore a handshake error if the server is behind a load balancer but - # the service ID is unknown. This indicates that the error happened - # when dialing the connection or during the MongoDB handshake, so we - # don't know the service ID to use for clearing the pool. - if self._settings.load_balanced and not service_id and not err_ctx.completed_handshake: - return - if isinstance(error, NetworkTimeout) and err_ctx.completed_handshake: - # The socket has been closed. Don't reset the server. - # Server Discovery And Monitoring Spec: "When an application - # operation fails because of any network error besides a socket - # timeout...." - return - elif isinstance(error, WriteError): - # Ignore writeErrors. - return - elif isinstance(error, (NotPrimaryError, OperationFailure)): - # As per the SDAM spec if: - # - the server sees a "not primary" error, and - # - the server is not shutting down, and - # - the server version is >= 4.2, then - # we keep the existing connection pool, but mark the server type - # as Unknown and request an immediate check of the server. - # Otherwise, we clear the connection pool, mark the server as - # Unknown and request an immediate check of the server. - if hasattr(error, "code"): - err_code = error.code - else: - # Default error code if one does not exist. - default = 10107 if isinstance(error, NotPrimaryError) else None - err_code = error.details.get("code", default) # type: ignore[union-attr] - if err_code in helpers_shared._NOT_PRIMARY_CODES: - is_shutting_down = err_code in helpers_shared._SHUTDOWN_CODES - # Mark server Unknown, clear the pool, and request check. - if not self._settings.load_balanced: - await self._process_change(ServerDescription(address, error=error)) - if is_shutting_down or (err_ctx.max_wire_version <= 7): - # Clear the pool. - await server.reset(service_id) - server.request_check() - elif not err_ctx.completed_handshake: - # Unknown command error during the connection handshake. - if not self._settings.load_balanced: - await self._process_change(ServerDescription(address, error=error)) - # Clear the pool. - await server.reset(service_id) - elif isinstance(error, ConnectionFailure): - if isinstance(error, WaitQueueTimeoutError) or ( - error.has_error_label("SystemOverloadedError") - ): - return - # "Client MUST replace the server's description with type Unknown - # ... MUST NOT request an immediate check of the server." - if not self._settings.load_balanced: - await self._process_change(ServerDescription(address, error=error)) - # Clear the pool. - await server.reset(service_id) - # "When a client marks a server Unknown from `Network error when - # reading or writing`_, clients MUST cancel the hello check on - # that server and close the current monitoring connection." + # Classify the error using the shared sans-I/O SDAM core, then execute + # the resulting actions here (the only I/O-colored part). + action = decide_error_action( + error, + load_balanced=bool(self._settings.load_balanced), + has_service_id=bool(err_ctx.service_id), + completed_handshake=err_ctx.completed_handshake, + max_wire_version=err_ctx.max_wire_version, + ) + if action.mark_unknown: + # Only error types that map to mark_unknown reach here, so error is + # a concrete Exception (see decide_error_action). + await self._process_change(ServerDescription(address, error=cast(Exception, error))) + if action.reset_pool: + await server.reset(err_ctx.service_id) + if action.request_check: + server.request_check() + if action.cancel_check: server._monitor.cancel_check() async def handle_error(self, address: _Address, err_ctx: _ErrorContext) -> None: @@ -1115,17 +1065,6 @@ def __init__( self.service_id = service_id -def _is_stale_error_topology_version( - current_tv: Optional[Mapping[str, Any]], error_tv: Optional[Mapping[str, Any]] -) -> bool: - """Return True if the error's topologyVersion is <= current.""" - if current_tv is None or error_tv is None: - return False - if current_tv["processId"] != error_tv["processId"]: - return False - return current_tv["counter"] >= error_tv["counter"] - - def _is_stale_server_description(current_sd: ServerDescription, new_sd: ServerDescription) -> bool: """Return True if the new topologyVersion is < current.""" current_tv, new_tv = current_sd.topology_version, new_sd.topology_version diff --git a/pymongo/network_layer.py b/pymongo/network_layer.py index 2295af7198..054fb6a35b 100644 --- a/pymongo/network_layer.py +++ b/pymongo/network_layer.py @@ -66,6 +66,45 @@ _UNPACK_HEADER = struct.Struct(" tuple[int, int, int, bool]: + """Parse and validate a MongoDB Wire Protocol message header. + + This is the sans-I/O core shared by the synchronous and asynchronous read + paths: it performs no I/O and contains no ``await``/blocking calls, so both + transports can reuse the exact same framing and validation logic. + + Given the 16 header bytes, returns a tuple of + ``(body_length, op_code, response_to, compressed)`` where ``body_length`` is + the number of bytes to read *after* the 16-byte header (and, when + ``compressed`` is True, also after the 9-byte OP_COMPRESSED sub-header). + Raises :class:`~pymongo.errors.ProtocolError` for malformed headers. + """ + length, _, response_to, op_code = _UNPACK_HEADER(header) + compressed = False + if op_code == 2012: # OP_COMPRESSED + if length <= 25: + raise ProtocolError( + f"Message length ({length!r}) not longer than standard OP_COMPRESSED " + f"message header size (25)" + ) + compressed = True + length -= 9 + if length <= 16: + raise ProtocolError( + f"Message length ({length!r}) not longer than standard message header size (16)" + ) + if length > max_message_size: + raise ProtocolError( + f"Message length ({length!r}) is larger than server max " + f"message size ({max_message_size!r})" + ) + return length - 16, op_code, response_to, compressed + + _POLL_TIMEOUT = 0.5 # Errors raised by sockets (and TLS sockets) when in non-blocking mode. BLOCKING_IO_ERRORS = (BlockingIOError, *BLOCKING_IO_LOOKUP_ERROR, *ssl_support.BLOCKING_IO_ERRORS) @@ -637,26 +676,7 @@ def buffer_updated(self, nbytes: int) -> None: def process_header(self) -> tuple[int, int, int, bool]: """Unpack a MongoDB Wire Protocol header.""" - length, _, response_to, op_code = _UNPACK_HEADER(self._header) - expecting_compression = False - if op_code == 2012: # OP_COMPRESSED - if length <= 25: - raise ProtocolError( - f"Message length ({length!r}) not longer than standard OP_COMPRESSED message header size (25)" - ) - expecting_compression = True - length -= 9 - if length <= 16: - raise ProtocolError( - f"Message length ({length!r}) not longer than standard message header size (16)" - ) - if length > self._max_message_size: - raise ProtocolError( - f"Message length ({length!r}) is larger than server max " - f"message size ({self._max_message_size!r})" - ) - - return length - 16, op_code, response_to, expecting_compression + return parse_wire_header(self._header, self._max_message_size) def process_compression_header(self) -> tuple[int, int]: """Unpack a MongoDB Wire Protocol compression header.""" @@ -758,27 +778,22 @@ def receive_message( deadline = time.monotonic() + timeout else: deadline = None - # Ignore the response's request id. - length, _, response_to, op_code = _UNPACK_HEADER(receive_data(conn, 16, deadline)) + # Parse and validate the header using the shared sans-I/O core. body_length + # is the number of bytes to read after the 16-byte header (and after the + # 9-byte OP_COMPRESSED sub-header when compressed). + body_length, op_code, response_to, compressed = parse_wire_header( + receive_data(conn, 16, deadline), max_message_size + ) # No request_id for exhaust cursor "getMore". if request_id is not None: if request_id != response_to: raise ProtocolError(f"Got response id {response_to!r} but expected {request_id!r}") - if length <= 16: - raise ProtocolError( - f"Message length ({length!r}) not longer than standard message header size (16)" - ) - if length > max_message_size: - raise ProtocolError( - f"Message length ({length!r}) is larger than server max " - f"message size ({max_message_size!r})" - ) data: memoryview | bytes - if op_code == 2012: + if compressed: op_code, _, compressor_id = _UNPACK_COMPRESSION_HEADER(receive_data(conn, 9, deadline)) - data = decompress(receive_data(conn, length - 25, deadline), compressor_id) + data = decompress(receive_data(conn, body_length, deadline), compressor_id) else: - data = receive_data(conn, length - 16, deadline) + data = receive_data(conn, body_length, deadline) try: unpack_reply = _UNPACK_REPLY[op_code] diff --git a/pymongo/synchronous/topology.py b/pymongo/synchronous/topology.py index b419833256..304ba6a164 100644 --- a/pymongo/synchronous/topology.py +++ b/pymongo/synchronous/topology.py @@ -29,17 +29,16 @@ from pathlib import Path from typing import TYPE_CHECKING, Any, Callable, Optional, cast -from pymongo import _csot, common, helpers_shared, periodic_executor +from pymongo import _csot, common, periodic_executor +from pymongo._sdam_error import ( + decide_error_action, + error_topology_version, + is_stale_error_topology_version, +) from pymongo.errors import ( - ConnectionFailure, InvalidOperation, - NetworkTimeout, - NotPrimaryError, - OperationFailure, PyMongoError, ServerSelectionTimeoutError, - WaitQueueTimeoutError, - WriteError, ) from pymongo.hello import Hello from pymongo.lock import ( @@ -847,14 +846,8 @@ def _is_stale_error(self, address: _Address, err_ctx: _ErrorContext) -> bool: return True # topologyVersion check, ignore error when cur_tv >= error_tv: - cur_tv = server.description.topology_version - error = err_ctx.error - error_tv = None - if error and hasattr(error, "details"): - if isinstance(error.details, dict): - error_tv = error.details.get("topologyVersion") - - return _is_stale_error_topology_version(cur_tv, error_tv) + error_tv = error_topology_version(err_ctx.error) + return is_stale_error_topology_version(server.description.topology_version, error_tv) def _handle_error(self, address: _Address, err_ctx: _ErrorContext) -> None: if self._is_stale_error(address, err_ctx): @@ -862,68 +855,25 @@ def _handle_error(self, address: _Address, err_ctx: _ErrorContext) -> None: server = self._servers[address] error = err_ctx.error - service_id = err_ctx.service_id - - # Ignore a handshake error if the server is behind a load balancer but - # the service ID is unknown. This indicates that the error happened - # when dialing the connection or during the MongoDB handshake, so we - # don't know the service ID to use for clearing the pool. - if self._settings.load_balanced and not service_id and not err_ctx.completed_handshake: - return - if isinstance(error, NetworkTimeout) and err_ctx.completed_handshake: - # The socket has been closed. Don't reset the server. - # Server Discovery And Monitoring Spec: "When an application - # operation fails because of any network error besides a socket - # timeout...." - return - elif isinstance(error, WriteError): - # Ignore writeErrors. - return - elif isinstance(error, (NotPrimaryError, OperationFailure)): - # As per the SDAM spec if: - # - the server sees a "not primary" error, and - # - the server is not shutting down, and - # - the server version is >= 4.2, then - # we keep the existing connection pool, but mark the server type - # as Unknown and request an immediate check of the server. - # Otherwise, we clear the connection pool, mark the server as - # Unknown and request an immediate check of the server. - if hasattr(error, "code"): - err_code = error.code - else: - # Default error code if one does not exist. - default = 10107 if isinstance(error, NotPrimaryError) else None - err_code = error.details.get("code", default) # type: ignore[union-attr] - if err_code in helpers_shared._NOT_PRIMARY_CODES: - is_shutting_down = err_code in helpers_shared._SHUTDOWN_CODES - # Mark server Unknown, clear the pool, and request check. - if not self._settings.load_balanced: - self._process_change(ServerDescription(address, error=error)) - if is_shutting_down or (err_ctx.max_wire_version <= 7): - # Clear the pool. - server.reset(service_id) - server.request_check() - elif not err_ctx.completed_handshake: - # Unknown command error during the connection handshake. - if not self._settings.load_balanced: - self._process_change(ServerDescription(address, error=error)) - # Clear the pool. - server.reset(service_id) - elif isinstance(error, ConnectionFailure): - if isinstance(error, WaitQueueTimeoutError) or ( - error.has_error_label("SystemOverloadedError") - ): - return - # "Client MUST replace the server's description with type Unknown - # ... MUST NOT request an immediate check of the server." - if not self._settings.load_balanced: - self._process_change(ServerDescription(address, error=error)) - # Clear the pool. - server.reset(service_id) - # "When a client marks a server Unknown from `Network error when - # reading or writing`_, clients MUST cancel the hello check on - # that server and close the current monitoring connection." + # Classify the error using the shared sans-I/O SDAM core, then execute + # the resulting actions here (the only I/O-colored part). + action = decide_error_action( + error, + load_balanced=bool(self._settings.load_balanced), + has_service_id=bool(err_ctx.service_id), + completed_handshake=err_ctx.completed_handshake, + max_wire_version=err_ctx.max_wire_version, + ) + if action.mark_unknown: + # Only error types that map to mark_unknown reach here, so error is + # a concrete Exception (see decide_error_action). + self._process_change(ServerDescription(address, error=cast(Exception, error))) + if action.reset_pool: + server.reset(err_ctx.service_id) + if action.request_check: + server.request_check() + if action.cancel_check: server._monitor.cancel_check() def handle_error(self, address: _Address, err_ctx: _ErrorContext) -> None: @@ -1113,17 +1063,6 @@ def __init__( self.service_id = service_id -def _is_stale_error_topology_version( - current_tv: Optional[Mapping[str, Any]], error_tv: Optional[Mapping[str, Any]] -) -> bool: - """Return True if the error's topologyVersion is <= current.""" - if current_tv is None or error_tv is None: - return False - if current_tv["processId"] != error_tv["processId"]: - return False - return current_tv["counter"] >= error_tv["counter"] - - def _is_stale_server_description(current_sd: ServerDescription, new_sd: ServerDescription) -> bool: """Return True if the new topologyVersion is < current.""" current_tv, new_tv = current_sd.topology_version, new_sd.topology_version diff --git a/test/test_sdam_error.py b/test/test_sdam_error.py new file mode 100644 index 0000000000..e11aec7931 --- /dev/null +++ b/test/test_sdam_error.py @@ -0,0 +1,183 @@ +# Copyright 2026-present MongoDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Test the sans-I/O SDAM error-handling core. + +These tests exercise pymongo._sdam_error directly. They perform no I/O and +require no MongoDB server: the module is pure decision logic. +""" + +from __future__ import annotations + +import sys + +sys.path[0:0] = [""] + +from pymongo._sdam_error import ( + _NO_ACTION, + _SDAMAction, + decide_error_action, + error_topology_version, + is_stale_error_topology_version, +) +from pymongo.errors import ( + ConnectionFailure, + NetworkTimeout, + NotPrimaryError, + OperationFailure, + WaitQueueTimeoutError, + WriteError, +) +from test import unittest + +# Baseline context: an error that occurred after a completed handshake on a +# modern (non-load-balanced) server. +_BASE = dict( + load_balanced=False, + has_service_id=False, + completed_handshake=True, + max_wire_version=21, +) + + +def _decide(error, **overrides): + return decide_error_action(error, **{**_BASE, **overrides}) + + +class TestDecideErrorAction(unittest.TestCase): + def test_network_timeout_after_handshake_is_ignored(self): + # SDAM: a network timeout after handshake does not reset the server. + self.assertEqual(_decide(NetworkTimeout("x")), _NO_ACTION) + + def test_network_timeout_during_handshake_is_connection_failure(self): + # NetworkTimeout is a ConnectionFailure; before handshake it is not the + # "ignore" case, so it marks Unknown, resets, and cancels the check. + self.assertEqual( + _decide(NetworkTimeout("x"), completed_handshake=False), + _SDAMAction(mark_unknown=True, reset_pool=True, cancel_check=True), + ) + + def test_write_error_is_ignored(self): + self.assertEqual(_decide(WriteError("x", 1)), _NO_ACTION) + + def test_not_primary_keeps_pool_and_requests_check(self): + # Code 10107 is a "not primary" code; on a modern server we keep the + # pool, mark Unknown, and request an immediate check. + err = NotPrimaryError("not primary", {"code": 10107, "errmsg": "not primary"}) + self.assertEqual( + _decide(err), + _SDAMAction(mark_unknown=True, request_check=True), + ) + + def test_shutdown_code_also_resets_pool(self): + # Code 91 (ShutdownInProgress) additionally clears the pool. + err = NotPrimaryError("shutting down", {"code": 91, "errmsg": "shutting down"}) + self.assertEqual( + _decide(err), + _SDAMAction(mark_unknown=True, reset_pool=True, request_check=True), + ) + + def test_not_primary_on_old_server_resets_pool(self): + # max_wire_version <= 7 forces a pool reset even for non-shutdown codes. + err = NotPrimaryError("not primary", {"code": 10107, "errmsg": "not primary"}) + self.assertEqual( + _decide(err, max_wire_version=7), + _SDAMAction(mark_unknown=True, reset_pool=True, request_check=True), + ) + + def test_operation_failure_during_handshake_resets_pool(self): + # A non-"not primary" command error during handshake marks Unknown and + # resets the pool, but does not request a check. + err = OperationFailure("boom", 59, {"code": 59}) + self.assertEqual( + _decide(err, completed_handshake=False), + _SDAMAction(mark_unknown=True, reset_pool=True), + ) + + def test_operation_failure_after_handshake_is_ignored(self): + # A non-"not primary" command error after handshake is ignored. + err = OperationFailure("boom", 59, {"code": 59}) + self.assertEqual(_decide(err), _NO_ACTION) + + def test_connection_failure_marks_unknown_resets_and_cancels(self): + self.assertEqual( + _decide(ConnectionFailure("x")), + _SDAMAction(mark_unknown=True, reset_pool=True, cancel_check=True), + ) + + def test_wait_queue_timeout_is_ignored(self): + self.assertEqual(_decide(WaitQueueTimeoutError("x")), _NO_ACTION) + + def test_system_overloaded_label_is_ignored(self): + err = ConnectionFailure("x") + err._add_error_label("SystemOverloadedError") + self.assertEqual(_decide(err), _NO_ACTION) + + def test_load_balanced_handshake_without_service_id_is_ignored(self): + # Behind a load balancer with no known service ID, a handshake error + # cannot be attributed to a pool and is ignored. + self.assertEqual( + _decide( + ConnectionFailure("x"), + load_balanced=True, + has_service_id=False, + completed_handshake=False, + ), + _NO_ACTION, + ) + + def test_load_balanced_suppresses_mark_unknown(self): + # In load-balanced mode we never mark the server Unknown, but we still + # reset the pool and cancel the check for a connection failure. + self.assertEqual( + _decide( + ConnectionFailure("x"), + load_balanced=True, + has_service_id=True, + completed_handshake=True, + ), + _SDAMAction(mark_unknown=False, reset_pool=True, cancel_check=True), + ) + + +class TestStaleErrorHelpers(unittest.TestCase): + def test_error_topology_version_absent(self): + self.assertIsNone(error_topology_version(None)) + self.assertIsNone(error_topology_version(ConnectionFailure("x"))) + self.assertIsNone(error_topology_version(OperationFailure("x", 1, {"code": 1}))) + + def test_error_topology_version_present(self): + tv = {"processId": "abc", "counter": 3} + err = OperationFailure("x", 1, {"code": 1, "topologyVersion": tv}) + self.assertEqual(error_topology_version(err), tv) + + def test_stale_when_current_counter_not_older(self): + cur = {"processId": "p", "counter": 5} + older = {"processId": "p", "counter": 4} + same = {"processId": "p", "counter": 5} + newer = {"processId": "p", "counter": 6} + self.assertTrue(is_stale_error_topology_version(cur, older)) + self.assertTrue(is_stale_error_topology_version(cur, same)) + self.assertFalse(is_stale_error_topology_version(cur, newer)) + + def test_not_stale_when_process_differs_or_missing(self): + cur = {"processId": "p", "counter": 5} + other = {"processId": "q", "counter": 1} + self.assertFalse(is_stale_error_topology_version(cur, other)) + self.assertFalse(is_stale_error_topology_version(None, cur)) + self.assertFalse(is_stale_error_topology_version(cur, None)) + + +if __name__ == "__main__": + unittest.main() From 8ad339192eb2896d38718c8e3f1bcb9b72807e03 Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Mon, 6 Jul 2026 12:16:05 -0400 Subject: [PATCH 02/11] [SPIKE] sans-I/O: extract monitor streaming-check predicate The predicate 'streaming enabled and server type + topologyVersion known' was duplicated at three sites in monitor.py (the streaming-continuation gate and two 'awaited' computations). Extract it into a shared sans-I/O helper is_streaming_check() in pymongo/_sdam_monitor.py, depending on a minimal structural Protocol rather than the full ServerDescription. - Deduplicates the predicate within monitor.py and removes it from unasync's scope (shared module, single implementation). - test/test_sdam_monitor.py: server-free unit tests. Verified: mypy+pyright clean, ruff clean, test_discovery_and_monitoring (180 mocked SDAM spec tests) pass. --- pymongo/_sdam_monitor.py | 46 ++++++++++++++++++++++++ pymongo/asynchronous/monitor.py | 12 +++---- pymongo/synchronous/monitor.py | 12 +++---- test/test_sdam_monitor.py | 63 +++++++++++++++++++++++++++++++++ 4 files changed, 117 insertions(+), 16 deletions(-) create mode 100644 pymongo/_sdam_monitor.py create mode 100644 test/test_sdam_monitor.py diff --git a/pymongo/_sdam_monitor.py b/pymongo/_sdam_monitor.py new file mode 100644 index 0000000000..3b2854699a --- /dev/null +++ b/pymongo/_sdam_monitor.py @@ -0,0 +1,46 @@ +# Copyright 2026-present MongoDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Sans-I/O decision helpers for SDAM server monitoring. + +Pure predicates over a :class:`~pymongo.server_description.ServerDescription` +shared by the synchronous and asynchronous monitors. They perform no I/O and +hold no locks, so a single implementation serves both flavors and can be +unit-tested without a live server. +""" + +from __future__ import annotations + +from typing import Any, Optional, Protocol + + +class _StreamableServer(Protocol): + """The minimal view of a ServerDescription these predicates depend on.""" + + @property + def is_server_type_known(self) -> bool: ... + + @property + def topology_version(self) -> Optional[Any]: ... + + +def is_streaming_check(stream: bool, sd: _StreamableServer) -> bool: + """Return True if the monitor will use the streaming protocol for ``sd``. + + A monitor uses the streaming (awaitable ``hello``) protocol when streaming + is enabled and the server's type and topologyVersion are already known. + This same condition decides whether an in-flight check is "awaited" and + whether to continue streaming immediately after a successful check. + """ + return bool(stream and sd.is_server_type_known and sd.topology_version) diff --git a/pymongo/asynchronous/monitor.py b/pymongo/asynchronous/monitor.py index 45c12b219f..9e5199d885 100644 --- a/pymongo/asynchronous/monitor.py +++ b/pymongo/asynchronous/monitor.py @@ -25,6 +25,7 @@ from pymongo import common, periodic_executor from pymongo._csot import MovingMinimum +from pymongo._sdam_monitor import is_streaming_check from pymongo.asynchronous.srv_resolver import _SrvResolver from pymongo.errors import NetworkTimeout, _OperationCancelled from pymongo.hello import Hello @@ -233,10 +234,7 @@ async def _run(self) -> None: interrupt_connections=isinstance(self._server_description.error, NetworkTimeout), ) - if self._stream and ( - self._server_description.is_server_type_known - and self._server_description.topology_version - ): + if is_streaming_check(self._stream, self._server_description): await self._start_rtt_monitor() # Immediately check for the next streaming response. self._executor.skip_sleep() @@ -267,7 +265,7 @@ async def _check_server(self) -> ServerDescription: sd = self._server_description address = sd.address duration = _monotonic_duration(start) - awaited = bool(self._stream and sd.is_server_type_known and sd.topology_version) + awaited = is_streaming_check(self._stream, sd) if self._publish: assert self._listeners is not None self._listeners.publish_server_heartbeat_failed(address, duration, error, awaited) @@ -300,9 +298,7 @@ async def _check_once(self) -> ServerDescription: # XXX: "awaited" could be incorrectly set to True in the rare case # the pool checkout closes and recreates a connection. - awaited = bool( - self._pool.conns and self._stream and sd.is_server_type_known and sd.topology_version - ) + awaited = bool(self._pool.conns) and is_streaming_check(self._stream, sd) if self._publish: assert self._listeners is not None self._listeners.publish_server_heartbeat_started(address, awaited) diff --git a/pymongo/synchronous/monitor.py b/pymongo/synchronous/monitor.py index f395588814..4d462f5c52 100644 --- a/pymongo/synchronous/monitor.py +++ b/pymongo/synchronous/monitor.py @@ -25,6 +25,7 @@ from pymongo import common, periodic_executor from pymongo._csot import MovingMinimum +from pymongo._sdam_monitor import is_streaming_check from pymongo.errors import NetworkTimeout, _OperationCancelled from pymongo.hello import Hello from pymongo.lock import _create_lock @@ -231,10 +232,7 @@ def _run(self) -> None: interrupt_connections=isinstance(self._server_description.error, NetworkTimeout), ) - if self._stream and ( - self._server_description.is_server_type_known - and self._server_description.topology_version - ): + if is_streaming_check(self._stream, self._server_description): self._start_rtt_monitor() # Immediately check for the next streaming response. self._executor.skip_sleep() @@ -265,7 +263,7 @@ def _check_server(self) -> ServerDescription: sd = self._server_description address = sd.address duration = _monotonic_duration(start) - awaited = bool(self._stream and sd.is_server_type_known and sd.topology_version) + awaited = is_streaming_check(self._stream, sd) if self._publish: assert self._listeners is not None self._listeners.publish_server_heartbeat_failed(address, duration, error, awaited) @@ -298,9 +296,7 @@ def _check_once(self) -> ServerDescription: # XXX: "awaited" could be incorrectly set to True in the rare case # the pool checkout closes and recreates a connection. - awaited = bool( - self._pool.conns and self._stream and sd.is_server_type_known and sd.topology_version - ) + awaited = bool(self._pool.conns) and is_streaming_check(self._stream, sd) if self._publish: assert self._listeners is not None self._listeners.publish_server_heartbeat_started(address, awaited) diff --git a/test/test_sdam_monitor.py b/test/test_sdam_monitor.py new file mode 100644 index 0000000000..1c78958502 --- /dev/null +++ b/test/test_sdam_monitor.py @@ -0,0 +1,63 @@ +# Copyright 2026-present MongoDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Test the sans-I/O SDAM monitor decision helpers. + +These tests exercise pymongo._sdam_monitor directly. They perform no I/O and +require no MongoDB server. +""" + +from __future__ import annotations + +import sys + +sys.path[0:0] = [""] + +from pymongo._sdam_monitor import is_streaming_check +from test import unittest + + +class _FakeSD: + """Minimal stand-in for ServerDescription; is_streaming_check only reads + is_server_type_known and topology_version.""" + + def __init__(self, is_server_type_known, topology_version): + self.is_server_type_known = is_server_type_known + self.topology_version = topology_version + + +_TV = {"processId": "p", "counter": 1} + + +class TestIsStreamingCheck(unittest.TestCase): + def test_streaming_when_enabled_and_type_and_tv_known(self): + self.assertTrue(is_streaming_check(True, _FakeSD(True, _TV))) + + def test_not_streaming_when_disabled(self): + self.assertFalse(is_streaming_check(False, _FakeSD(True, _TV))) + + def test_not_streaming_when_type_unknown(self): + self.assertFalse(is_streaming_check(True, _FakeSD(False, _TV))) + + def test_not_streaming_without_topology_version(self): + self.assertFalse(is_streaming_check(True, _FakeSD(True, None))) + + def test_returns_bool(self): + # Guard against leaking the truthy topology_version mapping. + result = is_streaming_check(True, _FakeSD(True, _TV)) + self.assertIsInstance(result, bool) + + +if __name__ == "__main__": + unittest.main() From b0cbc05444919f9371c9efe526bc5eae12248140 Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Mon, 6 Jul 2026 12:21:39 -0400 Subject: [PATCH 03/11] [SPIKE] sans-I/O: extract power-of-two server selection tie-break Move the SDAM 'power of two random choices' load-balancing tie-break out of Topology._select_server into a shared, generic select_least_loaded() in pymongo/_sdam_selection.py. It is generic over the server type via a 'load' callable, so it does not depend on the flavor-specific Server class, and takes an injectable sampler for deterministic testing. - Removes the decision from unasync's scope; drops the now-unused 'random' import from topology.py. - test/test_sdam_selection.py: server-free unit tests. --- pymongo/_sdam_selection.py | 52 ++++++++++++++++++++++++++++++ pymongo/asynchronous/topology.py | 10 ++---- pymongo/synchronous/topology.py | 10 ++---- test/test_sdam_selection.py | 55 ++++++++++++++++++++++++++++++++ 4 files changed, 111 insertions(+), 16 deletions(-) create mode 100644 pymongo/_sdam_selection.py create mode 100644 test/test_sdam_selection.py diff --git a/pymongo/_sdam_selection.py b/pymongo/_sdam_selection.py new file mode 100644 index 0000000000..2a8702570f --- /dev/null +++ b/pymongo/_sdam_selection.py @@ -0,0 +1,52 @@ +# Copyright 2026-present MongoDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Sans-I/O server-selection tie-breaking. + +Pure implementation of the SDAM "power of two random choices" load balancing +used to pick among equally-suitable servers. It is generic over the server type +(via a ``load`` callable), performs no I/O, and is shared by the synchronous and +asynchronous topologies. +""" + +from __future__ import annotations + +import random +from collections.abc import Sequence +from typing import Callable, TypeVar + +_T = TypeVar("_T") + + +def select_least_loaded( + candidates: Sequence[_T], + load: Callable[[_T], int], + *, + sample: Callable[[Sequence[_T], int], Sequence[_T]] = random.sample, +) -> _T: + """Choose a server from ``candidates`` using power-of-two random choices. + + Per the SDAM spec, when more than one server is suitable we sample two at + random and pick the one with the lower in-progress operation count, which + spreads load without a full scan. + + :param candidates: The suitable servers; must be non-empty. + :param load: Returns a candidate's current operation count. + :param sample: Injectable sampler (defaults to :func:`random.sample`), + used to make selection deterministic in tests. + """ + if len(candidates) == 1: + return candidates[0] + first, second = sample(candidates, 2) + return first if load(first) <= load(second) else second diff --git a/pymongo/asynchronous/topology.py b/pymongo/asynchronous/topology.py index ae3df920d6..a6a8075c37 100644 --- a/pymongo/asynchronous/topology.py +++ b/pymongo/asynchronous/topology.py @@ -20,7 +20,6 @@ import logging import os import queue -import random import sys import time import warnings @@ -35,6 +34,7 @@ error_topology_version, is_stale_error_topology_version, ) +from pymongo._sdam_selection import select_least_loaded from pymongo.asynchronous.client_session import _ServerSession, _ServerSessionPool from pymongo.asynchronous.monitor import MonitorBase, SrvMonitor from pymongo.asynchronous.pool import Pool @@ -406,13 +406,7 @@ async def _select_server( operation_id, deprioritized_servers, ) - if len(servers) == 1: - return servers[0] - server1, server2 = random.sample(servers, 2) - if server1.pool.operation_count <= server2.pool.operation_count: - return server1 - else: - return server2 + return select_least_loaded(servers, lambda server: server.pool.operation_count) async def select_server( self, diff --git a/pymongo/synchronous/topology.py b/pymongo/synchronous/topology.py index 304ba6a164..5667f044d9 100644 --- a/pymongo/synchronous/topology.py +++ b/pymongo/synchronous/topology.py @@ -20,7 +20,6 @@ import logging import os import queue -import random import sys import time import warnings @@ -35,6 +34,7 @@ error_topology_version, is_stale_error_topology_version, ) +from pymongo._sdam_selection import select_least_loaded from pymongo.errors import ( InvalidOperation, PyMongoError, @@ -406,13 +406,7 @@ def _select_server( operation_id, deprioritized_servers, ) - if len(servers) == 1: - return servers[0] - server1, server2 = random.sample(servers, 2) - if server1.pool.operation_count <= server2.pool.operation_count: - return server1 - else: - return server2 + return select_least_loaded(servers, lambda server: server.pool.operation_count) def select_server( self, diff --git a/test/test_sdam_selection.py b/test/test_sdam_selection.py new file mode 100644 index 0000000000..4964aa6a1e --- /dev/null +++ b/test/test_sdam_selection.py @@ -0,0 +1,55 @@ +# Copyright 2026-present MongoDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Test the sans-I/O server-selection tie-breaking. No server required.""" + +from __future__ import annotations + +import sys + +sys.path[0:0] = [""] + +from pymongo._sdam_selection import select_least_loaded +from test import unittest + + +class TestSelectLeastLoaded(unittest.TestCase): + def test_single_candidate_returned_directly(self): + # A lone candidate is returned without sampling. + def boom(_seq, _k): + raise AssertionError("sample must not be called for one candidate") + + self.assertEqual(select_least_loaded(["a"], lambda s: 0, sample=boom), "a") + + def test_picks_lower_load_of_the_two_sampled(self): + load = {"a": 5, "b": 2, "c": 9} + # Force the sampler to pick a and c; c is more loaded, so a wins. + chosen = select_least_loaded( + ["a", "b", "c"], load.__getitem__, sample=lambda seq, k: ["a", "c"] + ) + self.assertEqual(chosen, "a") + + def test_ties_prefer_first_sampled(self): + chosen = select_least_loaded(["a", "b"], lambda s: 7, sample=lambda seq, k: ["b", "a"]) + self.assertEqual(chosen, "b") + + def test_default_sampler_returns_a_candidate(self): + # With the real random sampler, the result is always one of the inputs. + candidates = ["a", "b", "c", "d"] + for _ in range(50): + self.assertIn(select_least_loaded(candidates, lambda s: 1), candidates) + + +if __name__ == "__main__": + unittest.main() From d29e422db94871ee9f0cc9b2e3e03c1949d63c40 Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Mon, 6 Jul 2026 12:31:56 -0400 Subject: [PATCH 04/11] [SPIKE] sans-I/O: make settings.py a shared, flavor-agnostic module Prove that a whole generated module can leave unasync. TopologySettings is pure configuration, but it was duplicated only because it defaulted pool_class/monitor_class to the flavor-specific pool.Pool/monitor.Monitor. - Move settings to a single shared pymongo/settings.py; delete the asynchronous/ and synchronous/ copies. - Relocate the flavor default from settings into topology.py (the layer that legitimately knows its flavor): 'self._settings.pool_class or Pool' etc. settings now stores pool_class/monitor_class as given (Optional). - Update all importers to 'from pymongo.settings import TopologySettings'. Because the default moved rather than being removed, every existing caller (including tests that omit the classes and rely on the default) keeps working with no behavior change. Cost of removing one module from unasync: ~17 files touched (mostly mechanical import swaps + regenerated twins), no shared Pool/Monitor base needed thanks to the default relocation. --- pymongo/asynchronous/mongo_client.py | 2 +- pymongo/asynchronous/monitor.py | 2 +- pymongo/asynchronous/topology.py | 17 +- pymongo/{asynchronous => }/settings.py | 32 ++-- pymongo/synchronous/mongo_client.py | 2 +- pymongo/synchronous/monitor.py | 2 +- pymongo/synchronous/settings.py | 176 ------------------ pymongo/synchronous/topology.py | 17 +- test/asynchronous/test_client.py | 2 +- .../test_discovery_and_monitoring.py | 2 +- test/asynchronous/test_server_selection.py | 2 +- test/asynchronous/utils_selection_tests.py | 2 +- test/test_client.py | 2 +- test/test_discovery_and_monitoring.py | 2 +- test/test_server_selection.py | 2 +- test/test_topology.py | 2 +- test/utils_selection_tests.py | 2 +- 17 files changed, 50 insertions(+), 218 deletions(-) rename pymongo/{asynchronous => }/settings.py (83%) delete mode 100644 pymongo/synchronous/settings.py diff --git a/pymongo/asynchronous/mongo_client.py b/pymongo/asynchronous/mongo_client.py index a558f96356..bdb2579a7b 100644 --- a/pymongo/asynchronous/mongo_client.py +++ b/pymongo/asynchronous/mongo_client.py @@ -65,7 +65,6 @@ from pymongo.asynchronous.helpers import ( _RetryPolicy, ) -from pymongo.asynchronous.settings import TopologySettings from pymongo.asynchronous.topology import Topology, _ErrorContext from pymongo.client_options import ClientOptions from pymongo.driver_info import DriverInfo @@ -111,6 +110,7 @@ from pymongo.server_description import ServerDescription from pymongo.server_selectors import writable_server_selector from pymongo.server_type import SERVER_TYPE +from pymongo.settings import TopologySettings from pymongo.topology_description import TOPOLOGY_TYPE, TopologyDescription from pymongo.typings import ( ClusterTime, diff --git a/pymongo/asynchronous/monitor.py b/pymongo/asynchronous/monitor.py index 9e5199d885..258c7eaeb0 100644 --- a/pymongo/asynchronous/monitor.py +++ b/pymongo/asynchronous/monitor.py @@ -42,8 +42,8 @@ Pool, _CancellationContext, ) - from pymongo.asynchronous.settings import TopologySettings from pymongo.asynchronous.topology import Topology + from pymongo.settings import TopologySettings _IS_SYNC = False diff --git a/pymongo/asynchronous/topology.py b/pymongo/asynchronous/topology.py index a6a8075c37..818b73fe3b 100644 --- a/pymongo/asynchronous/topology.py +++ b/pymongo/asynchronous/topology.py @@ -36,7 +36,7 @@ ) from pymongo._sdam_selection import select_least_loaded from pymongo.asynchronous.client_session import _ServerSession, _ServerSessionPool -from pymongo.asynchronous.monitor import MonitorBase, SrvMonitor +from pymongo.asynchronous.monitor import Monitor, MonitorBase, SrvMonitor from pymongo.asynchronous.pool import Pool from pymongo.asynchronous.server import Server from pymongo.errors import ( @@ -76,7 +76,7 @@ if TYPE_CHECKING: from bson import ObjectId - from pymongo.asynchronous.settings import TopologySettings + from pymongo.settings import TopologySettings from pymongo.typings import ClusterTime, _Address _IS_SYNC = False @@ -893,7 +893,8 @@ async def _update_servers(self) -> None: """ for address, sd in self._description.server_descriptions().items(): if address not in self._servers: - monitor = self._settings.monitor_class( + monitor_class = self._settings.monitor_class or Monitor + monitor = monitor_class( server_description=sd, topology=self, pool=self._create_pool_for_monitor(address), @@ -931,9 +932,8 @@ async def _update_servers(self) -> None: self._servers.pop(address) def _create_pool_for_server(self, address: _Address) -> Pool: - return self._settings.pool_class( - address, self._settings.pool_options, client_id=self._topology_id - ) + pool_class = self._settings.pool_class or Pool + return pool_class(address, self._settings.pool_options, client_id=self._topology_id) def _create_pool_for_monitor(self, address: _Address) -> Pool: options = self._settings.pool_options @@ -953,9 +953,8 @@ def _create_pool_for_monitor(self, address: _Address) -> Pool: server_api=options.server_api, ) - return self._settings.pool_class( - address, monitor_pool_options, is_sdam=True, client_id=self._topology_id - ) + pool_class = self._settings.pool_class or Pool + return pool_class(address, monitor_pool_options, is_sdam=True, client_id=self._topology_id) def _error_message(self, selector: Callable[[Selection], Selection]) -> str: """Format an error message if server selection fails. diff --git a/pymongo/asynchronous/settings.py b/pymongo/settings.py similarity index 83% rename from pymongo/asynchronous/settings.py rename to pymongo/settings.py index e3c2ee7fb3..2e984b4b4e 100644 --- a/pymongo/asynchronous/settings.py +++ b/pymongo/settings.py @@ -12,7 +12,14 @@ # implied. See the License for the specific language governing # permissions and limitations under the License. -"""Represent MongoClient's configuration.""" +"""Represent MongoClient's configuration. + +This module is flavor-agnostic (shared by the synchronous and asynchronous +drivers) and performs no I/O. It intentionally does not import the +flavor-specific ``pool``/``monitor`` modules: ``pool_class`` and +``monitor_class`` are stored as given (possibly ``None``), and the flavor's +default classes are supplied by the (flavor-specific) topology layer. +""" from __future__ import annotations @@ -23,25 +30,21 @@ from bson.objectid import ObjectId from pymongo import common -from pymongo.asynchronous import monitor, pool -from pymongo.asynchronous.pool import Pool from pymongo.common import LOCAL_THRESHOLD_MS, SERVER_SELECTION_TIMEOUT from pymongo.errors import ConfigurationError from pymongo.pool_options import PoolOptions from pymongo.server_description import ServerDescription from pymongo.topology_description import TOPOLOGY_TYPE, _ServerSelector -_IS_SYNC = False - class TopologySettings: def __init__( self, seeds: Optional[Collection[tuple[str, int]]] = None, replica_set_name: Optional[str] = None, - pool_class: Optional[type[Pool]] = None, + pool_class: Optional[type[Any]] = None, pool_options: Optional[PoolOptions] = None, - monitor_class: Optional[type[monitor.Monitor]] = None, + monitor_class: Optional[type[Any]] = None, condition_class: Optional[type[threading.Condition]] = None, local_threshold_ms: int = LOCAL_THRESHOLD_MS, server_selection_timeout: int = SERVER_SELECTION_TIMEOUT, @@ -58,6 +61,11 @@ def __init__( """Represent MongoClient's configuration. Take a list of (host, port) pairs and optional replica set name. + + ``pool_class`` and ``monitor_class`` are stored as given; when omitted + the topology layer fills in the flavor-specific default class, so this + module never needs to know whether it is serving the sync or async + driver. """ if heartbeat_frequency < common.MIN_HEARTBEAT_INTERVAL: raise ConfigurationError( @@ -66,9 +74,9 @@ def __init__( self._seeds: Collection[tuple[str, int]] = seeds or [("localhost", 27017)] self._replica_set_name = replica_set_name - self._pool_class: type[Pool] = pool_class or pool.Pool + self._pool_class: Optional[type[Any]] = pool_class self._pool_options: PoolOptions = pool_options or PoolOptions() - self._monitor_class: type[monitor.Monitor] = monitor_class or monitor.Monitor + self._monitor_class: Optional[type[Any]] = monitor_class self._condition_class: type[threading.Condition] = condition_class or threading.Condition self._local_threshold_ms = local_threshold_ms self._server_selection_timeout = server_selection_timeout @@ -98,7 +106,8 @@ def replica_set_name(self) -> Optional[str]: return self._replica_set_name @property - def pool_class(self) -> type[Pool]: + def pool_class(self) -> Optional[type[Any]]: + """The configured pool class, or None to use the flavor default.""" return self._pool_class @property @@ -106,7 +115,8 @@ def pool_options(self) -> PoolOptions: return self._pool_options @property - def monitor_class(self) -> type[monitor.Monitor]: + def monitor_class(self) -> Optional[type[Any]]: + """The configured monitor class, or None to use the flavor default.""" return self._monitor_class @property diff --git a/pymongo/synchronous/mongo_client.py b/pymongo/synchronous/mongo_client.py index 5f321afe5c..f5f273fa81 100644 --- a/pymongo/synchronous/mongo_client.py +++ b/pymongo/synchronous/mongo_client.py @@ -101,6 +101,7 @@ from pymongo.server_description import ServerDescription from pymongo.server_selectors import writable_server_selector from pymongo.server_type import SERVER_TYPE +from pymongo.settings import TopologySettings from pymongo.synchronous import client_session, database, uri_parser from pymongo.synchronous.change_stream import ChangeStream, ClusterChangeStream from pymongo.synchronous.client_bulk import _ClientBulk @@ -109,7 +110,6 @@ from pymongo.synchronous.helpers import ( _RetryPolicy, ) -from pymongo.synchronous.settings import TopologySettings from pymongo.synchronous.topology import Topology, _ErrorContext from pymongo.topology_description import TOPOLOGY_TYPE, TopologyDescription from pymongo.typings import ( diff --git a/pymongo/synchronous/monitor.py b/pymongo/synchronous/monitor.py index 4d462f5c52..aa71b78197 100644 --- a/pymongo/synchronous/monitor.py +++ b/pymongo/synchronous/monitor.py @@ -37,12 +37,12 @@ from pymongo.synchronous.srv_resolver import _SrvResolver if TYPE_CHECKING: + from pymongo.settings import TopologySettings from pymongo.synchronous.pool import ( # type: ignore[attr-defined] Connection, Pool, _CancellationContext, ) - from pymongo.synchronous.settings import TopologySettings from pymongo.synchronous.topology import Topology _IS_SYNC = True diff --git a/pymongo/synchronous/settings.py b/pymongo/synchronous/settings.py deleted file mode 100644 index bc664e8421..0000000000 --- a/pymongo/synchronous/settings.py +++ /dev/null @@ -1,176 +0,0 @@ -# Copyright 2014-present MongoDB, Inc. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you -# may not use this file except in compliance with the License. You -# may obtain a copy of the License at -# -# https://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or -# implied. See the License for the specific language governing -# permissions and limitations under the License. - -"""Represent MongoClient's configuration.""" - -from __future__ import annotations - -import threading -import traceback -from collections.abc import Collection -from typing import Any, Optional, Union - -from bson.objectid import ObjectId -from pymongo import common -from pymongo.common import LOCAL_THRESHOLD_MS, SERVER_SELECTION_TIMEOUT -from pymongo.errors import ConfigurationError -from pymongo.pool_options import PoolOptions -from pymongo.server_description import ServerDescription -from pymongo.synchronous import monitor, pool -from pymongo.synchronous.pool import Pool -from pymongo.topology_description import TOPOLOGY_TYPE, _ServerSelector - -_IS_SYNC = True - - -class TopologySettings: - def __init__( - self, - seeds: Optional[Collection[tuple[str, int]]] = None, - replica_set_name: Optional[str] = None, - pool_class: Optional[type[Pool]] = None, - pool_options: Optional[PoolOptions] = None, - monitor_class: Optional[type[monitor.Monitor]] = None, - condition_class: Optional[type[threading.Condition]] = None, - local_threshold_ms: int = LOCAL_THRESHOLD_MS, - server_selection_timeout: int = SERVER_SELECTION_TIMEOUT, - heartbeat_frequency: int = common.HEARTBEAT_FREQUENCY, - server_selector: Optional[_ServerSelector] = None, - fqdn: Optional[str] = None, - direct_connection: Optional[bool] = False, - load_balanced: Optional[bool] = None, - srv_service_name: str = common.SRV_SERVICE_NAME, - srv_max_hosts: int = 0, - server_monitoring_mode: str = common.SERVER_MONITORING_MODE, - topology_id: Optional[ObjectId] = None, - ): - """Represent MongoClient's configuration. - - Take a list of (host, port) pairs and optional replica set name. - """ - if heartbeat_frequency < common.MIN_HEARTBEAT_INTERVAL: - raise ConfigurationError( - f"heartbeatFrequencyMS cannot be less than {common.MIN_HEARTBEAT_INTERVAL * 1000}" - ) - - self._seeds: Collection[tuple[str, int]] = seeds or [("localhost", 27017)] - self._replica_set_name = replica_set_name - self._pool_class: type[Pool] = pool_class or pool.Pool - self._pool_options: PoolOptions = pool_options or PoolOptions() - self._monitor_class: type[monitor.Monitor] = monitor_class or monitor.Monitor - self._condition_class: type[threading.Condition] = condition_class or threading.Condition - self._local_threshold_ms = local_threshold_ms - self._server_selection_timeout = server_selection_timeout - self._server_selector = server_selector - self._fqdn = fqdn - self._heartbeat_frequency = heartbeat_frequency - self._direct = direct_connection - self._load_balanced = load_balanced - self._srv_service_name = srv_service_name - self._srv_max_hosts = srv_max_hosts or 0 - self._server_monitoring_mode = server_monitoring_mode - if topology_id is not None: - self._topology_id = topology_id - else: - self._topology_id = ObjectId() - # Store the allocation traceback to catch unclosed clients in the - # test suite. - self._stack = "".join(traceback.format_stack()[:-2]) - - @property - def seeds(self) -> Collection[tuple[str, int]]: - """List of server addresses.""" - return self._seeds - - @property - def replica_set_name(self) -> Optional[str]: - return self._replica_set_name - - @property - def pool_class(self) -> type[Pool]: - return self._pool_class - - @property - def pool_options(self) -> PoolOptions: - return self._pool_options - - @property - def monitor_class(self) -> type[monitor.Monitor]: - return self._monitor_class - - @property - def condition_class(self) -> type[threading.Condition]: - return self._condition_class - - @property - def local_threshold_ms(self) -> int: - return self._local_threshold_ms - - @property - def server_selection_timeout(self) -> int: - return self._server_selection_timeout - - @property - def server_selector(self) -> Optional[_ServerSelector]: - return self._server_selector - - @property - def heartbeat_frequency(self) -> int: - return self._heartbeat_frequency - - @property - def fqdn(self) -> Optional[str]: - return self._fqdn - - @property - def direct(self) -> Optional[bool]: - """Connect directly to a single server, or use a set of servers? - - True if there is one seed and no replica_set_name. - """ - return self._direct - - @property - def load_balanced(self) -> Optional[bool]: - """True if the client was configured to connect to a load balancer.""" - return self._load_balanced - - @property - def srv_service_name(self) -> str: - """The srvServiceName.""" - return self._srv_service_name - - @property - def srv_max_hosts(self) -> int: - """The srvMaxHosts.""" - return self._srv_max_hosts - - @property - def server_monitoring_mode(self) -> str: - """The serverMonitoringMode.""" - return self._server_monitoring_mode - - def get_topology_type(self) -> int: - if self.load_balanced: - return TOPOLOGY_TYPE.LoadBalanced - elif self.direct: - return TOPOLOGY_TYPE.Single - elif self.replica_set_name is not None: - return TOPOLOGY_TYPE.ReplicaSetNoPrimary - else: - return TOPOLOGY_TYPE.Unknown - - def get_server_descriptions(self) -> dict[Union[tuple[str, int], Any], ServerDescription]: - """Initial dict of (address, ServerDescription) for all seeds.""" - return {address: ServerDescription(address) for address in self.seeds} diff --git a/pymongo/synchronous/topology.py b/pymongo/synchronous/topology.py index 5667f044d9..f0c61f0a39 100644 --- a/pymongo/synchronous/topology.py +++ b/pymongo/synchronous/topology.py @@ -63,7 +63,7 @@ writable_server_selector, ) from pymongo.synchronous.client_session import _ServerSession, _ServerSessionPool -from pymongo.synchronous.monitor import MonitorBase, SrvMonitor +from pymongo.synchronous.monitor import Monitor, MonitorBase, SrvMonitor from pymongo.synchronous.pool import Pool from pymongo.synchronous.server import Server from pymongo.topology_description import ( @@ -76,7 +76,7 @@ if TYPE_CHECKING: from bson import ObjectId - from pymongo.synchronous.settings import TopologySettings + from pymongo.settings import TopologySettings from pymongo.typings import ClusterTime, _Address _IS_SYNC = True @@ -891,7 +891,8 @@ def _update_servers(self) -> None: """ for address, sd in self._description.server_descriptions().items(): if address not in self._servers: - monitor = self._settings.monitor_class( + monitor_class = self._settings.monitor_class or Monitor + monitor = monitor_class( server_description=sd, topology=self, pool=self._create_pool_for_monitor(address), @@ -929,9 +930,8 @@ def _update_servers(self) -> None: self._servers.pop(address) def _create_pool_for_server(self, address: _Address) -> Pool: - return self._settings.pool_class( - address, self._settings.pool_options, client_id=self._topology_id - ) + pool_class = self._settings.pool_class or Pool + return pool_class(address, self._settings.pool_options, client_id=self._topology_id) def _create_pool_for_monitor(self, address: _Address) -> Pool: options = self._settings.pool_options @@ -951,9 +951,8 @@ def _create_pool_for_monitor(self, address: _Address) -> Pool: server_api=options.server_api, ) - return self._settings.pool_class( - address, monitor_pool_options, is_sdam=True, client_id=self._topology_id - ) + pool_class = self._settings.pool_class or Pool + return pool_class(address, monitor_pool_options, is_sdam=True, client_id=self._topology_id) def _error_message(self, selector: Callable[[Selection], Selection]) -> str: """Format an error message if server selection fails. diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py index 5da186931a..1e258a5cf1 100644 --- a/test/asynchronous/test_client.py +++ b/test/asynchronous/test_client.py @@ -67,7 +67,6 @@ from pymongo.asynchronous.pool import ( AsyncConnection, ) -from pymongo.asynchronous.settings import TOPOLOGY_TYPE from pymongo.asynchronous.topology import _ErrorContext from pymongo.client_options import ClientOptions from pymongo.common import _UUID_REPRESENTATIONS, CONNECT_TIMEOUT, MIN_SUPPORTED_WIRE_VERSION, has_c @@ -92,6 +91,7 @@ from pymongo.server_description import ServerDescription from pymongo.server_selectors import readable_server_selector, writable_server_selector from pymongo.server_type import SERVER_TYPE +from pymongo.settings import TOPOLOGY_TYPE from pymongo.topology_description import TopologyDescription from pymongo.write_concern import WriteConcern from test.asynchronous import ( diff --git a/test/asynchronous/test_discovery_and_monitoring.py b/test/asynchronous/test_discovery_and_monitoring.py index 6fc0e09a19..9dfc2eccb6 100644 --- a/test/asynchronous/test_discovery_and_monitoring.py +++ b/test/asynchronous/test_discovery_and_monitoring.py @@ -40,7 +40,6 @@ from bson import Timestamp, json_util from pymongo import common, monitoring -from pymongo.asynchronous.settings import TopologySettings from pymongo.asynchronous.topology import Topology, _ErrorContext from pymongo.asynchronous.uri_parser import parse_uri from pymongo.errors import ( @@ -59,6 +58,7 @@ ServerHeartbeatStartedEvent, ) from pymongo.server_description import SERVER_TYPE, ServerDescription +from pymongo.settings import TopologySettings from pymongo.topology_description import TOPOLOGY_TYPE from test.asynchronous import ( AsyncIntegrationTest, diff --git a/test/asynchronous/test_server_selection.py b/test/asynchronous/test_server_selection.py index b9c84778b2..b06b8404aa 100644 --- a/test/asynchronous/test_server_selection.py +++ b/test/asynchronous/test_server_selection.py @@ -22,12 +22,12 @@ from pathlib import Path from pymongo import AsyncMongoClient, ReadPreference, monitoring -from pymongo.asynchronous.settings import TopologySettings from pymongo.asynchronous.topology import Topology from pymongo.errors import ServerSelectionTimeoutError from pymongo.hello import HelloCompat from pymongo.operations import _Op from pymongo.server_selectors import writable_server_selector +from pymongo.settings import TopologySettings from pymongo.typings import strip_optional sys.path[0:0] = [""] diff --git a/test/asynchronous/utils_selection_tests.py b/test/asynchronous/utils_selection_tests.py index eec6dc9719..39aec93ca5 100644 --- a/test/asynchronous/utils_selection_tests.py +++ b/test/asynchronous/utils_selection_tests.py @@ -26,12 +26,12 @@ sys.path[0:0] = [""] from bson import json_util -from pymongo.asynchronous.settings import TopologySettings from pymongo.asynchronous.topology import Topology from pymongo.common import HEARTBEAT_FREQUENCY, clean_node from pymongo.errors import AutoReconnect, ConfigurationError from pymongo.operations import _Op from pymongo.server_selectors import writable_server_selector +from pymongo.settings import TopologySettings from test import unittest from test.pymongo_mocks import DummyMonitor from test.utils_selection_tests_shared import ( diff --git a/test/test_client.py b/test/test_client.py index b37b5e57ac..6d6bdf394c 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -81,6 +81,7 @@ from pymongo.server_description import ServerDescription from pymongo.server_selectors import readable_server_selector, writable_server_selector from pymongo.server_type import SERVER_TYPE +from pymongo.settings import TOPOLOGY_TYPE from pymongo.synchronous.command_cursor import CommandCursor from pymongo.synchronous.cursor import Cursor, CursorType from pymongo.synchronous.database import Database @@ -89,7 +90,6 @@ from pymongo.synchronous.pool import ( Connection, ) -from pymongo.synchronous.settings import TOPOLOGY_TYPE from pymongo.synchronous.topology import _ErrorContext from pymongo.topology_description import TopologyDescription from pymongo.write_concern import WriteConcern diff --git a/test/test_discovery_and_monitoring.py b/test/test_discovery_and_monitoring.py index 612d52afbf..8df2fb7bf0 100644 --- a/test/test_discovery_and_monitoring.py +++ b/test/test_discovery_and_monitoring.py @@ -56,7 +56,7 @@ ServerHeartbeatStartedEvent, ) from pymongo.server_description import SERVER_TYPE, ServerDescription -from pymongo.synchronous.settings import TopologySettings +from pymongo.settings import TopologySettings from pymongo.synchronous.topology import Topology, _ErrorContext from pymongo.synchronous.uri_parser import parse_uri from pymongo.topology_description import TOPOLOGY_TYPE diff --git a/test/test_server_selection.py b/test/test_server_selection.py index 50076a8024..f2844eb4cf 100644 --- a/test/test_server_selection.py +++ b/test/test_server_selection.py @@ -26,7 +26,7 @@ from pymongo.hello import HelloCompat from pymongo.operations import _Op from pymongo.server_selectors import writable_server_selector -from pymongo.synchronous.settings import TopologySettings +from pymongo.settings import TopologySettings from pymongo.synchronous.topology import Topology from pymongo.typings import strip_optional diff --git a/test/test_topology.py b/test/test_topology.py index 47e670b20d..4d521a3d39 100644 --- a/test/test_topology.py +++ b/test/test_topology.py @@ -30,10 +30,10 @@ from pymongo.server_description import ServerDescription from pymongo.server_selectors import any_server_selector, writable_server_selector from pymongo.server_type import SERVER_TYPE +from pymongo.settings import TopologySettings from pymongo.synchronous.monitor import Monitor from pymongo.synchronous.pool import PoolOptions from pymongo.synchronous.server import Server -from pymongo.synchronous.settings import TopologySettings from pymongo.synchronous.topology import Topology, _ErrorContext from pymongo.topology_description import TOPOLOGY_TYPE from test import client_knobs, unittest diff --git a/test/utils_selection_tests.py b/test/utils_selection_tests.py index 3754c74674..f617f47047 100644 --- a/test/utils_selection_tests.py +++ b/test/utils_selection_tests.py @@ -30,7 +30,7 @@ from pymongo.errors import AutoReconnect, ConfigurationError from pymongo.operations import _Op from pymongo.server_selectors import writable_server_selector -from pymongo.synchronous.settings import TopologySettings +from pymongo.settings import TopologySettings from pymongo.synchronous.topology import Topology from test import unittest from test.pymongo_mocks import DummyMonitor From 944bc0faff6bd776e57d7371b7b0ebd4baf2d50e Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Mon, 6 Jul 2026 12:46:39 -0400 Subject: [PATCH 05/11] [SPIKE] sans-I/O: extract server-selection error message formatting Topology._error_message was ~55 lines of pure formatting over a TopologyDescription, duplicated verbatim in both flavors. Move it into a shared format_selection_error() in pymongo/_sdam_selection.py; topology now passes (description, selector, replica_set_name, seed_addresses) and delegates. - Removes the duplicated logic from unasync's scope. - test/test_sdam_selection.py: 6 server-free tests covering the no-servers / discovering / shared-error / distinct-errors / unreachable-seeds branches. Verified: mypy+pyright clean, ruff clean, test_topology + test_server_selection (sync and async) pass. --- pymongo/_sdam_selection.py | 72 +++++++++++++++++++++++++++++++- pymongo/asynchronous/topology.py | 55 +++--------------------- pymongo/synchronous/topology.py | 55 +++--------------------- test/test_sdam_selection.py | 62 ++++++++++++++++++++++++++- 4 files changed, 143 insertions(+), 101 deletions(-) diff --git a/pymongo/_sdam_selection.py b/pymongo/_sdam_selection.py index 2a8702570f..b70c13c1ef 100644 --- a/pymongo/_sdam_selection.py +++ b/pymongo/_sdam_selection.py @@ -24,11 +24,81 @@ import random from collections.abc import Sequence -from typing import Callable, TypeVar +from typing import TYPE_CHECKING, Callable, Optional, TypeVar + +from pymongo.server_selectors import Selection, writable_server_selector +from pymongo.topology_description import TOPOLOGY_TYPE + +if TYPE_CHECKING: + from pymongo.topology_description import TopologyDescription + from pymongo.typings import _Address _T = TypeVar("_T") +def format_selection_error( + description: TopologyDescription, + selector: Callable[[Selection], Selection], + replica_set_name: Optional[str], + seed_addresses: Sequence[_Address], +) -> str: + """Build the human-readable reason server selection failed. + + Pure formatting over a :class:`TopologyDescription`; performs no I/O. Shared + by the synchronous and asynchronous topologies so the diagnostic text has a + single source of truth. + """ + is_replica_set = description.topology_type in ( + TOPOLOGY_TYPE.ReplicaSetWithPrimary, + TOPOLOGY_TYPE.ReplicaSetNoPrimary, + ) + + if is_replica_set: + server_plural = "replica set members" + elif description.topology_type == TOPOLOGY_TYPE.Sharded: + server_plural = "mongoses" + else: + server_plural = "servers" + + if description.known_servers: + # We've connected, but no servers match the selector. + if selector is writable_server_selector: + if is_replica_set: + return "No primary available for writes" + else: + return f"No {server_plural} available for writes" + else: + return f'No {server_plural} match selector "{selector}"' + else: + addresses = list(description.server_descriptions()) + servers = list(description.server_descriptions().values()) + if not servers: + if is_replica_set: + # We removed all servers because of the wrong setName? + return f'No {server_plural} available for replica set name "{replica_set_name}"' + else: + return f"No {server_plural} available" + + # 1 or more servers, all Unknown. Are they unknown for one reason? + error = servers[0].error + same = all(server.error == error for server in servers[1:]) + if same: + if error is None: + # We're still discovering. + return f"No {server_plural} found yet" + + if is_replica_set and not set(addresses).intersection(seed_addresses): + # We replaced our seeds with new hosts but can't reach any. + return ( + f"Could not reach any servers in {addresses}. Replica set is" + " configured with internal hostnames or IPs?" + ) + + return str(error) + else: + return ",".join(str(server.error) for server in servers if server.error) + + def select_least_loaded( candidates: Sequence[_T], load: Callable[[_T], int], diff --git a/pymongo/asynchronous/topology.py b/pymongo/asynchronous/topology.py index 818b73fe3b..37ee5423e6 100644 --- a/pymongo/asynchronous/topology.py +++ b/pymongo/asynchronous/topology.py @@ -34,7 +34,7 @@ error_topology_version, is_stale_error_topology_version, ) -from pymongo._sdam_selection import select_least_loaded +from pymongo._sdam_selection import format_selection_error, select_least_loaded from pymongo.asynchronous.client_session import _ServerSession, _ServerSessionPool from pymongo.asynchronous.monitor import Monitor, MonitorBase, SrvMonitor from pymongo.asynchronous.pool import Pool @@ -961,56 +961,13 @@ def _error_message(self, selector: Callable[[Selection], Selection]) -> str: Hold the lock when calling this. """ - is_replica_set = self._description.topology_type in ( - TOPOLOGY_TYPE.ReplicaSetWithPrimary, - TOPOLOGY_TYPE.ReplicaSetNoPrimary, + return format_selection_error( + self._description, + selector, + self._settings.replica_set_name, + self._seed_addresses, ) - if is_replica_set: - server_plural = "replica set members" - elif self._description.topology_type == TOPOLOGY_TYPE.Sharded: - server_plural = "mongoses" - else: - server_plural = "servers" - - if self._description.known_servers: - # We've connected, but no servers match the selector. - if selector is writable_server_selector: - if is_replica_set: - return "No primary available for writes" - else: - return f"No {server_plural} available for writes" - else: - return f'No {server_plural} match selector "{selector}"' - else: - addresses = list(self._description.server_descriptions()) - servers = list(self._description.server_descriptions().values()) - if not servers: - if is_replica_set: - # We removed all servers because of the wrong setName? - return f'No {server_plural} available for replica set name "{self._settings.replica_set_name}"' - else: - return f"No {server_plural} available" - - # 1 or more servers, all Unknown. Are they unknown for one reason? - error = servers[0].error - same = all(server.error == error for server in servers[1:]) - if same: - if error is None: - # We're still discovering. - return f"No {server_plural} found yet" - - if is_replica_set and not set(addresses).intersection(self._seed_addresses): - # We replaced our seeds with new hosts but can't reach any. - return ( - f"Could not reach any servers in {addresses}. Replica set is" - " configured with internal hostnames or IPs?" - ) - - return str(error) - else: - return ",".join(str(server.error) for server in servers if server.error) - async def cleanup_monitors(self) -> None: tasks = [] try: diff --git a/pymongo/synchronous/topology.py b/pymongo/synchronous/topology.py index f0c61f0a39..b312207d65 100644 --- a/pymongo/synchronous/topology.py +++ b/pymongo/synchronous/topology.py @@ -34,7 +34,7 @@ error_topology_version, is_stale_error_topology_version, ) -from pymongo._sdam_selection import select_least_loaded +from pymongo._sdam_selection import format_selection_error, select_least_loaded from pymongo.errors import ( InvalidOperation, PyMongoError, @@ -959,56 +959,13 @@ def _error_message(self, selector: Callable[[Selection], Selection]) -> str: Hold the lock when calling this. """ - is_replica_set = self._description.topology_type in ( - TOPOLOGY_TYPE.ReplicaSetWithPrimary, - TOPOLOGY_TYPE.ReplicaSetNoPrimary, + return format_selection_error( + self._description, + selector, + self._settings.replica_set_name, + self._seed_addresses, ) - if is_replica_set: - server_plural = "replica set members" - elif self._description.topology_type == TOPOLOGY_TYPE.Sharded: - server_plural = "mongoses" - else: - server_plural = "servers" - - if self._description.known_servers: - # We've connected, but no servers match the selector. - if selector is writable_server_selector: - if is_replica_set: - return "No primary available for writes" - else: - return f"No {server_plural} available for writes" - else: - return f'No {server_plural} match selector "{selector}"' - else: - addresses = list(self._description.server_descriptions()) - servers = list(self._description.server_descriptions().values()) - if not servers: - if is_replica_set: - # We removed all servers because of the wrong setName? - return f'No {server_plural} available for replica set name "{self._settings.replica_set_name}"' - else: - return f"No {server_plural} available" - - # 1 or more servers, all Unknown. Are they unknown for one reason? - error = servers[0].error - same = all(server.error == error for server in servers[1:]) - if same: - if error is None: - # We're still discovering. - return f"No {server_plural} found yet" - - if is_replica_set and not set(addresses).intersection(self._seed_addresses): - # We replaced our seeds with new hosts but can't reach any. - return ( - f"Could not reach any servers in {addresses}. Replica set is" - " configured with internal hostnames or IPs?" - ) - - return str(error) - else: - return ",".join(str(server.error) for server in servers if server.error) - def cleanup_monitors(self) -> None: tasks = [] try: diff --git a/test/test_sdam_selection.py b/test/test_sdam_selection.py index 4964aa6a1e..d0b829e3e7 100644 --- a/test/test_sdam_selection.py +++ b/test/test_sdam_selection.py @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Test the sans-I/O server-selection tie-breaking. No server required.""" +"""Test the sans-I/O server-selection helpers. No server required.""" from __future__ import annotations @@ -20,10 +20,25 @@ sys.path[0:0] = [""] -from pymongo._sdam_selection import select_least_loaded +from pymongo._sdam_selection import format_selection_error, select_least_loaded +from pymongo.errors import ConnectionFailure +from pymongo.server_description import ServerDescription +from pymongo.server_selectors import any_server_selector, writable_server_selector +from pymongo.settings import TopologySettings +from pymongo.topology_description import TOPOLOGY_TYPE, TopologyDescription from test import unittest +def _td(topology_type, servers): + """Build a real TopologyDescription from {address: ServerDescription}.""" + settings = TopologySettings(seeds=list(servers) or [("a", 27017)]) + return TopologyDescription(topology_type, dict(servers), None, None, None, settings) + + +_A = ("a", 27017) +_B = ("b", 27017) + + class TestSelectLeastLoaded(unittest.TestCase): def test_single_candidate_returned_directly(self): # A lone candidate is returned without sampling. @@ -51,5 +66,48 @@ def test_default_sampler_returns_a_candidate(self): self.assertIn(select_least_loaded(candidates, lambda s: 1), candidates) +class TestFormatSelectionError(unittest.TestCase): + def test_no_servers_available(self): + td = _td(TOPOLOGY_TYPE.Unknown, {}) + msg = format_selection_error(td, any_server_selector, None, []) + self.assertEqual(msg, "No servers available") + + def test_no_replica_set_members_for_set_name(self): + td = _td(TOPOLOGY_TYPE.ReplicaSetNoPrimary, {}) + msg = format_selection_error(td, any_server_selector, "rs0", []) + self.assertEqual(msg, 'No replica set members available for replica set name "rs0"') + + def test_still_discovering(self): + # A single Unknown server with no error yet. + td = _td(TOPOLOGY_TYPE.Unknown, {_A: ServerDescription(_A)}) + msg = format_selection_error(td, any_server_selector, None, [_A]) + self.assertEqual(msg, "No servers found yet") + + def test_all_servers_share_one_error(self): + err = ConnectionFailure("boom") + td = _td(TOPOLOGY_TYPE.Unknown, {_A: ServerDescription(_A, error=err)}) + msg = format_selection_error(td, any_server_selector, None, [_A]) + self.assertEqual(msg, "boom") + + def test_distinct_errors_are_joined(self): + td = _td( + TOPOLOGY_TYPE.Unknown, + { + _A: ServerDescription(_A, error=ConnectionFailure("err-a")), + _B: ServerDescription(_B, error=ConnectionFailure("err-b")), + }, + ) + msg = format_selection_error(td, any_server_selector, None, [_A, _B]) + self.assertEqual(sorted(msg.split(",")), ["err-a", "err-b"]) + + def test_replica_set_seeds_unreachable(self): + # Servers whose addresses do not intersect the original seeds. + err = ConnectionFailure("unreachable") + td = _td(TOPOLOGY_TYPE.ReplicaSetNoPrimary, {_A: ServerDescription(_A, error=err)}) + msg = format_selection_error(td, writable_server_selector, "rs0", seed_addresses=[_B]) + self.assertIn("Could not reach any servers", msg) + self.assertIn("internal hostnames or IPs", msg) + + if __name__ == "__main__": unittest.main() From 8ccc7076b86c92bc0e0a7a7f885c199704a90873 Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Mon, 6 Jul 2026 12:52:56 -0400 Subject: [PATCH 06/11] [SPIKE] sans-I/O: extract staleness + data-bearing-servers decisions Two more pure decisions that lived in the generated topology.py: - is_stale_server_description: move next to its sibling staleness helpers in pymongo/_sdam_error.py (topologyVersion comparison; pure). - data_bearing_servers: extract the 'which servers can be selected' decision into pymongo/_sdam_selection.py; Topology keeps a thin delegating wrapper. Both removed from unasync's scope. Added server-free tests for each (building ServerDescriptions from Hello docs). Verified: mypy+pyright clean, ruff clean, test_topology passes. --- pymongo/_sdam_error.py | 19 +++++++++++- pymongo/_sdam_selection.py | 12 ++++++++ pymongo/asynchronous/topology.py | 23 +++++--------- pymongo/synchronous/topology.py | 23 +++++--------- test/test_sdam_error.py | 51 ++++++++++++++++++++++++++++++++ test/test_sdam_selection.py | 39 +++++++++++++++++++++++- 6 files changed, 135 insertions(+), 32 deletions(-) diff --git a/pymongo/_sdam_error.py b/pymongo/_sdam_error.py index 83ac66259c..31d5edd18d 100644 --- a/pymongo/_sdam_error.py +++ b/pymongo/_sdam_error.py @@ -30,9 +30,12 @@ from __future__ import annotations from collections.abc import Mapping -from typing import Any, NamedTuple, Optional +from typing import TYPE_CHECKING, Any, NamedTuple, Optional from pymongo import helpers_shared + +if TYPE_CHECKING: + from pymongo.server_description import ServerDescription from pymongo.errors import ( ConnectionFailure, NetworkTimeout, @@ -70,6 +73,20 @@ def is_stale_error_topology_version( return current_tv["counter"] >= error_tv["counter"] +def is_stale_server_description(current_sd: ServerDescription, new_sd: ServerDescription) -> bool: + """Return True if ``new_sd``'s topologyVersion is older than ``current_sd``'s. + + Pure comparison used to discard a server description update that is stale + relative to what the topology already knows. + """ + current_tv, new_tv = current_sd.topology_version, new_sd.topology_version + if current_tv is None or new_tv is None: + return False + if current_tv["processId"] != new_tv["processId"]: + return False + return current_tv["counter"] > new_tv["counter"] + + class _SDAMAction(NamedTuple): """The set of SDAM actions to take in response to an application error. diff --git a/pymongo/_sdam_selection.py b/pymongo/_sdam_selection.py index b70c13c1ef..250eaaf93a 100644 --- a/pymongo/_sdam_selection.py +++ b/pymongo/_sdam_selection.py @@ -30,12 +30,24 @@ from pymongo.topology_description import TOPOLOGY_TYPE if TYPE_CHECKING: + from pymongo.server_description import ServerDescription from pymongo.topology_description import TopologyDescription from pymongo.typings import _Address _T = TypeVar("_T") +def data_bearing_servers(description: TopologyDescription) -> list[ServerDescription]: + """Return all servers that might be selected for an operation. + + Pure decision over a :class:`TopologyDescription`: for a single-server + topology any known server is data-bearing; otherwise only readable servers. + """ + if description.topology_type == TOPOLOGY_TYPE.Single: + return description.known_servers + return description.readable_servers + + def format_selection_error( description: TopologyDescription, selector: Callable[[Selection], Selection], diff --git a/pymongo/asynchronous/topology.py b/pymongo/asynchronous/topology.py index 37ee5423e6..454c31348f 100644 --- a/pymongo/asynchronous/topology.py +++ b/pymongo/asynchronous/topology.py @@ -33,8 +33,13 @@ decide_error_action, error_topology_version, is_stale_error_topology_version, + is_stale_server_description, +) +from pymongo._sdam_selection import ( + data_bearing_servers, + format_selection_error, + select_least_loaded, ) -from pymongo._sdam_selection import format_selection_error, select_least_loaded from pymongo.asynchronous.client_session import _ServerSession, _ServerSessionPool from pymongo.asynchronous.monitor import Monitor, MonitorBase, SrvMonitor from pymongo.asynchronous.pool import Pool @@ -487,7 +492,7 @@ async def _process_change( """ td_old = self._description sd_old = td_old._server_descriptions[server_description.address] - if _is_stale_server_description(sd_old, server_description): + if is_stale_server_description(sd_old, server_description): # This is a stale hello response. Ignore it. return @@ -687,9 +692,7 @@ def data_bearing_servers(self) -> list[ServerDescription]: This includes any server that might be selected for an operation. """ - if self._description.topology_type == TOPOLOGY_TYPE.Single: - return self._description.known_servers - return self._description.readable_servers + return data_bearing_servers(self._description) async def update_pool(self) -> None: # Remove any stale sockets and add new sockets if pool is too small. @@ -1013,13 +1016,3 @@ def __init__( self.sock_generation = sock_generation self.completed_handshake = completed_handshake self.service_id = service_id - - -def _is_stale_server_description(current_sd: ServerDescription, new_sd: ServerDescription) -> bool: - """Return True if the new topologyVersion is < current.""" - current_tv, new_tv = current_sd.topology_version, new_sd.topology_version - if current_tv is None or new_tv is None: - return False - if current_tv["processId"] != new_tv["processId"]: - return False - return current_tv["counter"] > new_tv["counter"] diff --git a/pymongo/synchronous/topology.py b/pymongo/synchronous/topology.py index b312207d65..68777ef9b1 100644 --- a/pymongo/synchronous/topology.py +++ b/pymongo/synchronous/topology.py @@ -33,8 +33,13 @@ decide_error_action, error_topology_version, is_stale_error_topology_version, + is_stale_server_description, +) +from pymongo._sdam_selection import ( + data_bearing_servers, + format_selection_error, + select_least_loaded, ) -from pymongo._sdam_selection import format_selection_error, select_least_loaded from pymongo.errors import ( InvalidOperation, PyMongoError, @@ -487,7 +492,7 @@ def _process_change( """ td_old = self._description sd_old = td_old._server_descriptions[server_description.address] - if _is_stale_server_description(sd_old, server_description): + if is_stale_server_description(sd_old, server_description): # This is a stale hello response. Ignore it. return @@ -685,9 +690,7 @@ def data_bearing_servers(self) -> list[ServerDescription]: This includes any server that might be selected for an operation. """ - if self._description.topology_type == TOPOLOGY_TYPE.Single: - return self._description.known_servers - return self._description.readable_servers + return data_bearing_servers(self._description) def update_pool(self) -> None: # Remove any stale sockets and add new sockets if pool is too small. @@ -1011,13 +1014,3 @@ def __init__( self.sock_generation = sock_generation self.completed_handshake = completed_handshake self.service_id = service_id - - -def _is_stale_server_description(current_sd: ServerDescription, new_sd: ServerDescription) -> bool: - """Return True if the new topologyVersion is < current.""" - current_tv, new_tv = current_sd.topology_version, new_sd.topology_version - if current_tv is None or new_tv is None: - return False - if current_tv["processId"] != new_tv["processId"]: - return False - return current_tv["counter"] > new_tv["counter"] diff --git a/test/test_sdam_error.py b/test/test_sdam_error.py index e11aec7931..1ab641acb7 100644 --- a/test/test_sdam_error.py +++ b/test/test_sdam_error.py @@ -24,12 +24,15 @@ sys.path[0:0] = [""] +from bson import ObjectId +from bson.int64 import Int64 from pymongo._sdam_error import ( _NO_ACTION, _SDAMAction, decide_error_action, error_topology_version, is_stale_error_topology_version, + is_stale_server_description, ) from pymongo.errors import ( ConnectionFailure, @@ -39,6 +42,24 @@ WaitQueueTimeoutError, WriteError, ) +from pymongo.hello import Hello +from pymongo.server_description import ServerDescription + +_PID = ObjectId() + + +def _sd_with_topology_version(counter, process_id=_PID): + hello = Hello( + { + "ok": 1, + "isWritablePrimary": True, + "maxWireVersion": 21, + "topologyVersion": {"processId": process_id, "counter": Int64(counter)}, + } + ) + return ServerDescription(("a", 27017), hello, 0) + + from test import unittest # Baseline context: an error that occurred after a completed handshake on a @@ -179,5 +200,35 @@ def test_not_stale_when_process_differs_or_missing(self): self.assertFalse(is_stale_error_topology_version(cur, None)) +class TestIsStaleServerDescription(unittest.TestCase): + def test_older_counter_is_stale(self): + current = _sd_with_topology_version(5) + older = _sd_with_topology_version(3) + self.assertTrue(is_stale_server_description(current, older)) + + def test_newer_counter_is_not_stale(self): + current = _sd_with_topology_version(5) + newer = _sd_with_topology_version(7) + self.assertFalse(is_stale_server_description(current, newer)) + + def test_equal_counter_is_not_stale(self): + # Only strictly-older is stale (">" not ">="). + current = _sd_with_topology_version(5) + same = _sd_with_topology_version(5) + self.assertFalse(is_stale_server_description(current, same)) + + def test_different_process_is_not_stale(self): + current = _sd_with_topology_version(5) + other = _sd_with_topology_version(1, process_id=ObjectId()) + self.assertFalse(is_stale_server_description(current, other)) + + def test_missing_topology_version_is_not_stale(self): + current = _sd_with_topology_version(5) + # A plain Unknown ServerDescription has no topologyVersion. + unknown = ServerDescription(("a", 27017)) + self.assertFalse(is_stale_server_description(current, unknown)) + self.assertFalse(is_stale_server_description(unknown, current)) + + if __name__ == "__main__": unittest.main() diff --git a/test/test_sdam_selection.py b/test/test_sdam_selection.py index d0b829e3e7..3c25d8ca6a 100644 --- a/test/test_sdam_selection.py +++ b/test/test_sdam_selection.py @@ -20,8 +20,13 @@ sys.path[0:0] = [""] -from pymongo._sdam_selection import format_selection_error, select_least_loaded +from pymongo._sdam_selection import ( + data_bearing_servers, + format_selection_error, + select_least_loaded, +) from pymongo.errors import ConnectionFailure +from pymongo.hello import Hello from pymongo.server_description import ServerDescription from pymongo.server_selectors import any_server_selector, writable_server_selector from pymongo.settings import TopologySettings @@ -29,6 +34,16 @@ from test import unittest +def _standalone_sd(address): + hello = Hello({"ok": 1, "isWritablePrimary": True, "maxWireVersion": 21}) + return ServerDescription(address, hello, 0) + + +def _mongos_sd(address): + hello = Hello({"ok": 1, "msg": "isdbgrid", "maxWireVersion": 21}) + return ServerDescription(address, hello, 0) + + def _td(topology_type, servers): """Build a real TopologyDescription from {address: ServerDescription}.""" settings = TopologySettings(seeds=list(servers) or [("a", 27017)]) @@ -109,5 +124,27 @@ def test_replica_set_seeds_unreachable(self): self.assertIn("internal hostnames or IPs", msg) +class TestDataBearingServers(unittest.TestCase): + def test_single_returns_known_servers(self): + # For a Single topology, any known server is data-bearing. + td = _td(TOPOLOGY_TYPE.Single, {_A: _standalone_sd(_A)}) + result = data_bearing_servers(td) + self.assertEqual([sd.address for sd in result], [_A]) + + def test_single_excludes_unknown_servers(self): + td = _td(TOPOLOGY_TYPE.Single, {_A: ServerDescription(_A)}) + self.assertEqual(data_bearing_servers(td), []) + + def test_sharded_returns_readable_servers(self): + td = _td(TOPOLOGY_TYPE.Sharded, {_A: _mongos_sd(_A), _B: _mongos_sd(_B)}) + result = data_bearing_servers(td) + self.assertEqual({sd.address for sd in result}, {_A, _B}) + + def test_sharded_excludes_unknown_servers(self): + td = _td(TOPOLOGY_TYPE.Sharded, {_A: _mongos_sd(_A), _B: ServerDescription(_B)}) + result = data_bearing_servers(td) + self.assertEqual({sd.address for sd in result}, {_A}) + + if __name__ == "__main__": unittest.main() From a1e9ff87067befe2f70630225e07751d102740b3 Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Mon, 6 Jul 2026 12:55:24 -0400 Subject: [PATCH 07/11] [SPIKE] sans-I/O: tidy import ordering in test_sdam_error --- test/test_sdam_error.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/test/test_sdam_error.py b/test/test_sdam_error.py index 1ab641acb7..89e9da2c10 100644 --- a/test/test_sdam_error.py +++ b/test/test_sdam_error.py @@ -44,6 +44,7 @@ ) from pymongo.hello import Hello from pymongo.server_description import ServerDescription +from test import unittest _PID = ObjectId() @@ -60,8 +61,6 @@ def _sd_with_topology_version(counter, process_id=_PID): return ServerDescription(("a", 27017), hello, 0) -from test import unittest - # Baseline context: an error that occurred after a completed handshake on a # modern (non-load-balanced) server. _BASE = dict( From 744b614ab51b28f70e192a08ed87534cb2732f47 Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Mon, 6 Jul 2026 15:52:37 -0400 Subject: [PATCH 08/11] [SPIKE] sans-I/O: add findings write-up (SANS_IO_SPIKE.md) --- SANS_IO_SPIKE.md | 67 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 67 insertions(+) create mode 100644 SANS_IO_SPIKE.md diff --git a/SANS_IO_SPIKE.md b/SANS_IO_SPIKE.md new file mode 100644 index 0000000000..5ec0c76c80 --- /dev/null +++ b/SANS_IO_SPIKE.md @@ -0,0 +1,67 @@ +# Spike: sans-I/O extraction to reduce `unasync` duplication + +**Status:** exploratory spike (branch `state-machines`, draft PR #2918). Not intended to merge as-is. + +**Question:** How far can the sans-I/O (state-machine) pattern reduce the async/sync code that `unasync` must generate — and what does it cost? + +## Background + +PyMongo ships two drivers for the same functionality: + +- `pymongo/asynchronous/` — hand-written (`AsyncMongoClient`, `AsyncCollection`, …). +- `pymongo/synchronous/` — generated from the async source by `unasync` (token replacement: `await x` → `x`, `AsyncFoo` → `Foo`, `async def` → `def`, …). + +This duplicates a lot of code that is not actually async/sync-specific — pure protocol/decision logic that happens to sit next to an `await`. Two costs follow: + +1. **Maintenance:** every change must survive naive text substitution and is effectively written twice. +2. **Divergence risk:** the two copies can drift. This spike found a live example — see PYTHON-5906 below. + +## The idea: separate "colorless" logic from the "colored" shell + +Borrowing the *sans-I/O* framing (and the "what color is your function" metaphor): + +- **Colored** code is tied to one driver: it contains `await`, an `asyncio`/`threading` lock or condition, a blocking or awaited socket call, a background loop. It *cannot* be shared and must be generated. +- **Colorless / sans-I/O** code performs no I/O and holds no locks — pure decisions over data. It can be written once in a shared top-level module and imported by both drivers. + +The method for each extraction: + +1. Find a pure decision embedded in a colored method (e.g. "given this error, should we reset the pool?"). +2. Lift it into a shared module as a pure function. +3. Leave a thin call site in the colored method that executes the decision. + +## What was extracted + +| Area | Shared module / function | Was in | +| --- | --- | --- | +| Wire-message framing + validation | `network_layer.parse_wire_header()` | `network_layer.py` (both `receive_message` and `PyMongoProtocol.process_header`) | +| SDAM application-error decision tree | `pymongo/_sdam_error.py` — `decide_error_action()`, `_SDAMAction` | `topology._handle_error` | +| topologyVersion staleness | `_sdam_error.py` — `error_topology_version`, `is_stale_error_topology_version`, `is_stale_server_description` | `topology` | +| Monitor streaming predicate | `pymongo/_sdam_monitor.py` — `is_streaming_check()` | `monitor.py` (deduped 3 call sites) | +| Server-selection tie-break | `pymongo/_sdam_selection.py` — `select_least_loaded()` | `topology._select_server` | +| Selection-failure message | `_sdam_selection.py` — `format_selection_error()` | `topology._error_message` | +| Data-bearing servers | `_sdam_selection.py` — `data_bearing_servers()` | `topology` | +| **Whole config module** | `pymongo/settings.py` (shared; async/sync copies deleted) | `pymongo/{asynchronous,synchronous}/settings.py` | + +~41 new unit tests were added for the extracted logic. All are **server-free** — they exercise SDAM/wire-protocol/selection decisions that previously could only be tested against a live `mongod`. + +## Bug found and fixed: PYTHON-5906 + +Unifying the wire-header framing surfaced a real divergence. The sync `receive_message` was missing the `OP_COMPRESSED` minimum-length guard (`length <= 25`) that the async `process_header` had. A malformed compressed header with `16 < length <= 25` made the sync path compute a negative body length and raise `ValueError` instead of `ProtocolError`. Sharing `parse_wire_header()` fixes both paths from one place. (Blocked on triage; can ship as its own small PR.) + +## Findings + +1. **Leaf extraction is cheap and high-value.** Lifting a pure decision into a shared module + thin call site is low-risk, deletes duplicated logic, and unlocks server-free tests. This is the bulk of the branch (6 of 7 commits) and would be safe to do incrementally on `master`. + +2. **A whole module *can* leave `unasync` — but the cost is the import graph, not the code.** `settings.py` is pure config, yet it was split only because it defaulted `pool_class`/`monitor_class` to the flavor-specific `pool.Pool`/`monitor.Monitor`. Relocating that one default into `topology.py` (`self._settings.pool_class or Pool`) let the module become shared with **no behavior change** — but the move touched ~17 files, because every `from pymongo.{flavor}.settings import` caller (many themselves generated) had to be rewritten. **The barrier to de-duplicating a module is how deeply its flavor-specific import path is woven through the generated layer, not how pure it is.** + +3. **Zero `unasync` is not reachable — and that's fine.** The irreducibly colored shell stays: lock/condition acquisition, condition-variable waits (`asyncio` vs `threading`), background monitor loops, task lifecycle, socket I/O. No trick shares those. But essentially all the *logic* can move out of the shell, leaving generated code that contains nothing worth testing or worth risking divergence over. + +## Validation + +- Local: `just typing` (mypy + pyright) clean, `ruff`/`synchro` clean, full topology/SDAM/selection suites pass (sync and async twins). +- Evergreen targeted patch (server-free + `mongodb-latest`, both drivers, all topologies): green. +- Evergreen broad stress patch (server v4.2→rapid, all compression codecs, load-balancer, mockupdb, gevent, no-c-ext, encryption, pyopenssl, macOS/Win64): 113 tasks, 81 passed; all 32 failures triaged to pre-existing/unrelated causes (encryption `prefix` feature gap, mongod-start infra flakiness, timing-sensitive backpressure/handshake tests that passed under the identical diff, a Windows infra glitch). No failure attributable to this diff. + +## Open question / possible follow-up + +The extractions above pull *decisions* out of the shell but leave *control flow* (retry loops, waits) colored. The remaining question is whether a colored control-flow shell — e.g. `topology._select_servers_loop` — can be modeled as an explicit sans-I/O state machine that **emits intents** ("wait for a topology change", "re-apply the selector") consumed by a tiny colored driver. That would shrink even the shell, but it is a redesign rather than a leaf extraction and was left out of this spike. From 52775ea18e5a1302d30502fb460b2dc40920f94c Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Mon, 6 Jul 2026 16:02:17 -0400 Subject: [PATCH 09/11] [SPIKE] sans-I/O: replace color metaphor with async/sync terminology in write-up --- SANS_IO_SPIKE.md | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/SANS_IO_SPIKE.md b/SANS_IO_SPIKE.md index 5ec0c76c80..20e95a5ff0 100644 --- a/SANS_IO_SPIKE.md +++ b/SANS_IO_SPIKE.md @@ -16,18 +16,18 @@ This duplicates a lot of code that is not actually async/sync-specific — pure 1. **Maintenance:** every change must survive naive text substitution and is effectively written twice. 2. **Divergence risk:** the two copies can drift. This spike found a live example — see PYTHON-5906 below. -## The idea: separate "colorless" logic from the "colored" shell +## The idea: separate shared logic from the async/sync-specific wrapper -Borrowing the *sans-I/O* framing (and the "what color is your function" metaphor): +Every method mixes two kinds of code: -- **Colored** code is tied to one driver: it contains `await`, an `asyncio`/`threading` lock or condition, a blocking or awaited socket call, a background loop. It *cannot* be shared and must be generated. -- **Colorless / sans-I/O** code performs no I/O and holds no locks — pure decisions over data. It can be written once in a shared top-level module and imported by both drivers. +- **async/sync-specific** code that is tied to one driver: it contains `await`, an `asyncio`/`threading` lock or condition, a blocking or awaited socket call, or a background loop. It *cannot* be shared and must be generated by `unasync`. +- **Shared (sans-I/O)** code that performs no I/O and holds no locks — pure decisions over data. It can be written once in a shared top-level module and imported by both drivers. The method for each extraction: -1. Find a pure decision embedded in a colored method (e.g. "given this error, should we reset the pool?"). +1. Find a pure decision embedded in an async/sync-specific method (e.g. "given this error, should we reset the pool?"). 2. Lift it into a shared module as a pure function. -3. Leave a thin call site in the colored method that executes the decision. +3. Leave a thin call site in the async/sync method that executes the decision. ## What was extracted @@ -52,9 +52,9 @@ Unifying the wire-header framing surfaced a real divergence. The sync `receive_m 1. **Leaf extraction is cheap and high-value.** Lifting a pure decision into a shared module + thin call site is low-risk, deletes duplicated logic, and unlocks server-free tests. This is the bulk of the branch (6 of 7 commits) and would be safe to do incrementally on `master`. -2. **A whole module *can* leave `unasync` — but the cost is the import graph, not the code.** `settings.py` is pure config, yet it was split only because it defaulted `pool_class`/`monitor_class` to the flavor-specific `pool.Pool`/`monitor.Monitor`. Relocating that one default into `topology.py` (`self._settings.pool_class or Pool`) let the module become shared with **no behavior change** — but the move touched ~17 files, because every `from pymongo.{flavor}.settings import` caller (many themselves generated) had to be rewritten. **The barrier to de-duplicating a module is how deeply its flavor-specific import path is woven through the generated layer, not how pure it is.** +2. **A whole module *can* leave `unasync` — but the cost is the import graph, not the code.** `settings.py` is pure config, yet it was split only because it defaulted `pool_class`/`monitor_class` to the async/sync-specific `pool.Pool`/`monitor.Monitor`. Relocating that one default into `topology.py` (`self._settings.pool_class or Pool`) let the module become shared with **no behavior change** — but the move touched ~17 files, because every `from pymongo.{asynchronous,synchronous}.settings import` caller (many themselves generated) had to be rewritten. **The barrier to de-duplicating a module is how deeply its async/sync-specific import path is woven through the generated layer, not how pure it is.** -3. **Zero `unasync` is not reachable — and that's fine.** The irreducibly colored shell stays: lock/condition acquisition, condition-variable waits (`asyncio` vs `threading`), background monitor loops, task lifecycle, socket I/O. No trick shares those. But essentially all the *logic* can move out of the shell, leaving generated code that contains nothing worth testing or worth risking divergence over. +3. **Zero `unasync` is not reachable — and that's fine.** The async/sync-specific wrapper stays: lock/condition acquisition, condition-variable waits (`asyncio` vs `threading`), background monitor loops, task lifecycle, socket I/O. No trick shares those. But essentially all the *logic* can move out of the wrapper, leaving generated code that contains nothing worth testing or worth risking divergence over. ## Validation @@ -64,4 +64,4 @@ Unifying the wire-header framing surfaced a real divergence. The sync `receive_m ## Open question / possible follow-up -The extractions above pull *decisions* out of the shell but leave *control flow* (retry loops, waits) colored. The remaining question is whether a colored control-flow shell — e.g. `topology._select_servers_loop` — can be modeled as an explicit sans-I/O state machine that **emits intents** ("wait for a topology change", "re-apply the selector") consumed by a tiny colored driver. That would shrink even the shell, but it is a redesign rather than a leaf extraction and was left out of this spike. +The extractions above pull *decisions* out of the async/sync wrapper but leave *control flow* (retry loops, waits) driver-specific. The remaining question is whether such a wrapper — e.g. `topology._select_servers_loop` — can be modeled as an explicit sans-I/O state machine that **emits intents** ("wait for a topology change", "re-apply the selector") consumed by a tiny async/sync-specific driver. That would shrink even the wrapper, but it is a redesign rather than a leaf extraction and was left out of this spike. From db8e99004380a653aa0110e574f5875fd7d8c3b2 Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Mon, 6 Jul 2026 16:07:27 -0400 Subject: [PATCH 10/11] [SPIKE] sans-I/O: drop PYTHON-5906 section, reword 'leaf extraction' --- SANS_IO_SPIKE.md | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/SANS_IO_SPIKE.md b/SANS_IO_SPIKE.md index 20e95a5ff0..ea0a58a682 100644 --- a/SANS_IO_SPIKE.md +++ b/SANS_IO_SPIKE.md @@ -44,13 +44,9 @@ The method for each extraction: ~41 new unit tests were added for the extracted logic. All are **server-free** — they exercise SDAM/wire-protocol/selection decisions that previously could only be tested against a live `mongod`. -## Bug found and fixed: PYTHON-5906 - -Unifying the wire-header framing surfaced a real divergence. The sync `receive_message` was missing the `OP_COMPRESSED` minimum-length guard (`length <= 25`) that the async `process_header` had. A malformed compressed header with `16 < length <= 25` made the sync path compute a negative body length and raise `ValueError` instead of `ProtocolError`. Sharing `parse_wire_header()` fixes both paths from one place. (Blocked on triage; can ship as its own small PR.) - ## Findings -1. **Leaf extraction is cheap and high-value.** Lifting a pure decision into a shared module + thin call site is low-risk, deletes duplicated logic, and unlocks server-free tests. This is the bulk of the branch (6 of 7 commits) and would be safe to do incrementally on `master`. +1. **Extracting individual functions is cheap and high-value.** Lifting a single self-contained decision into a shared module + thin call site is low-risk, deletes duplicated logic, and unlocks server-free tests. This is the bulk of the branch (6 of 7 commits) and would be safe to do incrementally on `master`. 2. **A whole module *can* leave `unasync` — but the cost is the import graph, not the code.** `settings.py` is pure config, yet it was split only because it defaulted `pool_class`/`monitor_class` to the async/sync-specific `pool.Pool`/`monitor.Monitor`. Relocating that one default into `topology.py` (`self._settings.pool_class or Pool`) let the module become shared with **no behavior change** — but the move touched ~17 files, because every `from pymongo.{asynchronous,synchronous}.settings import` caller (many themselves generated) had to be rewritten. **The barrier to de-duplicating a module is how deeply its async/sync-specific import path is woven through the generated layer, not how pure it is.** @@ -64,4 +60,4 @@ Unifying the wire-header framing surfaced a real divergence. The sync `receive_m ## Open question / possible follow-up -The extractions above pull *decisions* out of the async/sync wrapper but leave *control flow* (retry loops, waits) driver-specific. The remaining question is whether such a wrapper — e.g. `topology._select_servers_loop` — can be modeled as an explicit sans-I/O state machine that **emits intents** ("wait for a topology change", "re-apply the selector") consumed by a tiny async/sync-specific driver. That would shrink even the wrapper, but it is a redesign rather than a leaf extraction and was left out of this spike. +The extractions above pull *decisions* out of the async/sync wrapper but leave *control flow* (retry loops, waits) driver-specific. The remaining question is whether such a wrapper — e.g. `topology._select_servers_loop` — can be modeled as an explicit sans-I/O state machine that **emits intents** ("wait for a topology change", "re-apply the selector") consumed by a tiny async/sync-specific driver. That would shrink even the wrapper, but it is a redesign rather than a single-function extraction and was left out of this spike. From 4bab64dd284f77274721c41b61c513734966226d Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Mon, 6 Jul 2026 16:17:18 -0400 Subject: [PATCH 11/11] Fix wording in divergence risk section Clarified the mention of a live example in the divergence risk section. --- SANS_IO_SPIKE.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/SANS_IO_SPIKE.md b/SANS_IO_SPIKE.md index ea0a58a682..0134c1eb33 100644 --- a/SANS_IO_SPIKE.md +++ b/SANS_IO_SPIKE.md @@ -14,7 +14,7 @@ PyMongo ships two drivers for the same functionality: This duplicates a lot of code that is not actually async/sync-specific — pure protocol/decision logic that happens to sit next to an `await`. Two costs follow: 1. **Maintenance:** every change must survive naive text substitution and is effectively written twice. -2. **Divergence risk:** the two copies can drift. This spike found a live example — see PYTHON-5906 below. +2. **Divergence risk:** the two copies can drift. This spike found a live example — see PYTHON-5906. ## The idea: separate shared logic from the async/sync-specific wrapper