[SPIKE] sans-I/O extraction to reduce unasync duplication

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.

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

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)