diff --git a/SANS_IO_SPIKE.md b/SANS_IO_SPIKE.md new file mode 100644 index 0000000000..0134c1eb33 --- /dev/null +++ b/SANS_IO_SPIKE.md @@ -0,0 +1,63 @@ +# 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. + +## The idea: separate shared logic from the async/sync-specific wrapper + +Every method mixes two kinds of code: + +- **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 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 async/sync 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`. + +## Findings + +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.** + +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 + +- 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 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. diff --git a/pymongo/_sdam_error.py b/pymongo/_sdam_error.py new file mode 100644 index 0000000000..31d5edd18d --- /dev/null +++ b/pymongo/_sdam_error.py @@ -0,0 +1,190 @@ +# 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 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, + 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"] + + +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. + + 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/_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/_sdam_selection.py b/pymongo/_sdam_selection.py new file mode 100644 index 0000000000..250eaaf93a --- /dev/null +++ b/pymongo/_sdam_selection.py @@ -0,0 +1,134 @@ +# 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 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.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], + 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], + *, + 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/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 45c12b219f..258c7eaeb0 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 @@ -41,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 @@ -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/asynchronous/topology.py b/pymongo/asynchronous/topology.py index 6af075cdc2..454c31348f 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 @@ -29,21 +28,26 @@ 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, + is_stale_server_description, +) +from pymongo._sdam_selection import ( + data_bearing_servers, + format_selection_error, + 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 ( - ConnectionFailure, InvalidOperation, - NetworkTimeout, - NotPrimaryError, - OperationFailure, PyMongoError, ServerSelectionTimeoutError, - WaitQueueTimeoutError, - WriteError, ) from pymongo.hello import Hello from pymongo.lock import ( @@ -77,7 +81,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 @@ -407,13 +411,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, @@ -494,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 @@ -694,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. @@ -849,14 +845,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 +854,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: @@ -949,7 +896,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), @@ -987,9 +935,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 @@ -1009,65 +956,21 @@ 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. 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: @@ -1113,24 +1016,3 @@ def __init__( self.sock_generation = sock_generation self.completed_handshake = completed_handshake 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 - 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/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/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 f395588814..aa71b78197 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 @@ -36,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 @@ -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/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 b419833256..68777ef9b1 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 @@ -29,17 +28,22 @@ 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, + is_stale_server_description, +) +from pymongo._sdam_selection import ( + data_bearing_servers, + format_selection_error, + select_least_loaded, +) from pymongo.errors import ( - ConnectionFailure, InvalidOperation, - NetworkTimeout, - NotPrimaryError, - OperationFailure, PyMongoError, ServerSelectionTimeoutError, - WaitQueueTimeoutError, - WriteError, ) from pymongo.hello import Hello from pymongo.lock import ( @@ -64,7 +68,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 ( @@ -77,7 +81,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 @@ -407,13 +411,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, @@ -494,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 @@ -692,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. @@ -847,14 +843,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 +852,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: @@ -947,7 +894,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), @@ -985,9 +933,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 @@ -1007,65 +954,21 @@ 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. 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: @@ -1111,24 +1014,3 @@ def __init__( self.sock_generation = sock_generation self.completed_handshake = completed_handshake 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 - 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/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_sdam_error.py b/test/test_sdam_error.py new file mode 100644 index 0000000000..89e9da2c10 --- /dev/null +++ b/test/test_sdam_error.py @@ -0,0 +1,233 @@ +# 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 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, + NetworkTimeout, + NotPrimaryError, + OperationFailure, + WaitQueueTimeoutError, + WriteError, +) +from pymongo.hello import Hello +from pymongo.server_description import ServerDescription +from test import unittest + +_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) + + +# 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)) + + +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_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() diff --git a/test/test_sdam_selection.py b/test/test_sdam_selection.py new file mode 100644 index 0000000000..3c25d8ca6a --- /dev/null +++ b/test/test_sdam_selection.py @@ -0,0 +1,150 @@ +# 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 helpers. No server required.""" + +from __future__ import annotations + +import sys + +sys.path[0:0] = [""] + +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 +from pymongo.topology_description import TOPOLOGY_TYPE, TopologyDescription +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)]) + 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. + 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) + + +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) + + +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() 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