From c08c24c43f7ff4bf55d7f9ce24f0856a31d5132e Mon Sep 17 00:00:00 2001 From: Iris Date: Wed, 7 Oct 2026 17:43:30 -0700 Subject: [PATCH 1/3] PYTHON-6027 Consolidate non-IO code in Topology into shared base class --- pymongo/asynchronous/topology.py | 270 ++----------------------------- pymongo/synchronous/topology.py | 270 ++----------------------------- pymongo/topology_shared.py | 265 +++++++++++++++++++++++++++++- pymongo/typings.py | 4 +- 4 files changed, 291 insertions(+), 518 deletions(-) diff --git a/pymongo/asynchronous/topology.py b/pymongo/asynchronous/topology.py index c4f1a3fa96..268873cd32 100644 --- a/pymongo/asynchronous/topology.py +++ b/pymongo/asynchronous/topology.py @@ -30,14 +30,12 @@ from pymongo import _csot, common, helpers_shared, periodic_executor from pymongo._telemetry import ( - _SdamTelemetry, _ServerSelectionTelemetry, log_server_selection_succeeded, ) -from pymongo.asynchronous.monitor import MonitorBase, SrvMonitor +from pymongo.asynchronous.monitor import SrvMonitor from pymongo.asynchronous.pool import Pool from pymongo.asynchronous.server import Server -from pymongo.client_session_shared import _ServerSession, _ServerSessionPool from pymongo.errors import ( ConnectionFailure, InvalidOperation, @@ -50,13 +48,8 @@ WriteError, ) from pymongo.hello import Hello -from pymongo.lock import ( - _async_cond_wait, - _async_create_condition, - _async_create_lock, -) +from pymongo.lock import _async_cond_wait, _async_create_condition, _async_create_lock from pymongo.logger import _SERVER_SELECTION_LOGGER, _is_debug_enabled -from pymongo.pool_options import PoolOptions from pymongo.server_description import ServerDescription from pymongo.server_selectors import ( Selection, @@ -73,71 +66,32 @@ updated_topology_description, ) from pymongo.topology_shared import ( + _AgnosticTopologyBase, _ErrorContext, - _is_stale_error_topology_version, _is_stale_server_description, process_events_queue, ) if TYPE_CHECKING: from pymongo.asynchronous.settings import TopologySettings - from pymongo.typings import ClusterTime, _Address + from pymongo.typings import _Address _IS_SYNC = False _pymongo_dir = str(Path(__file__).parent) -class Topology: +class Topology(_AgnosticTopologyBase[Pool, Server]): """Monitor a topology of one or more servers.""" - def __init__(self, topology_settings: TopologySettings): - self._topology_id = topology_settings._topology_id - self._listeners = topology_settings._pool_options._event_listeners - - # Create events queue if there are publishers. - self._events: queue.Queue[Any] | None = None - self.__events_executor: Any = None - - publish_server = self._listeners is not None and self._listeners.enabled_for_server - publish_tp = self._listeners is not None and self._listeners.enabled_for_topology - if publish_server or publish_tp: - self._events = queue.Queue(maxsize=100) - - self._sdam = _SdamTelemetry(self._topology_id, self._listeners, self._events) - self._sdam.topology_opened() - - self._settings = topology_settings - topology_description = TopologyDescription( - topology_settings.get_topology_type(), - topology_settings.get_server_descriptions(), - topology_settings.replica_set_name, - None, - None, - topology_settings, - ) + _settings: TopologySettings - self._description = topology_description - initial_td = TopologyDescription( - TOPOLOGY_TYPE.Unknown, {}, None, None, None, self._settings - ) - self._sdam.topology_description_changed(initial_td, self._description) - - for seed in topology_settings.seeds: - self._sdam.server_opened(seed) - - # Store the seed list to help diagnose errors in _error_message(). - self._seed_addresses = list(topology_description.server_descriptions()) - self._opened = False - self._closed = False + def __init__(self, topology_settings: TopologySettings): + super().__init__(topology_settings) self._lock = _async_create_lock() self._condition = _async_create_condition( self._lock, self._settings.condition_class if _IS_SYNC else None ) - self._servers: dict[_Address, Server] = {} - self._pid: Optional[int] = None - self._max_cluster_time: Optional[ClusterTime] = None - self._session_pool = _ServerSessionPool() if self._sdam._publish_server or self._sdam._publish_tp: assert self._events is not None @@ -157,16 +111,12 @@ async def target() -> bool: # the queue via this closure. When the topology is freed, stop # the executor soon. weak = weakref.ref(self._events, executor.close) - self.__events_executor = executor + self._events_executor = executor executor.open() - self._srv_monitor = None if self._settings.fqdn is not None and not self._settings.load_balanced: self._srv_monitor = SrvMonitor(self, self._settings) - # Stores all monitor tasks that need to be joined on close or server selection - self._monitor_tasks: list[MonitorBase] = [] - async def open(self) -> None: """Start monitoring, or restart after a fork. @@ -205,13 +155,6 @@ async def open(self) -> None: async with self._lock: await self._ensure_opened() - def get_server_selection_timeout(self) -> float: - # CSOT: use remaining timeout when set. - timeout = _csot.remaining() - if timeout is None: - return self._settings.server_selection_timeout - return timeout - async def select_servers( self, selector: Callable[[Selection], Selection], @@ -504,19 +447,6 @@ async def on_srv_update(self, seedlist: list[tuple[str, Any]]) -> None: if self._opened: await self._process_srv_update(seedlist) - def get_server_by_address(self, address: _Address) -> Optional[Server]: - """Get a Server or None. - - Returns the current version of the server immediately, even if it's - Unknown or absent from the topology. Only use this in unittests. - In driver code, use select_server_by_address, since then you're - assured a recent view of the server's type and wire protocol version. - """ - return self._servers.get(address) - - def has_server(self, address: _Address) -> bool: - return address in self._servers - async def get_primary(self) -> Optional[_Address]: """Return primary's address or None.""" # Implemented here in Topology instead of AsyncMongoClient, so it can lock. @@ -550,25 +480,6 @@ async def get_arbiters(self) -> set[_Address]: """Return set of arbiter addresses.""" return await self._get_replica_set_members(arbiter_server_selector) - def max_cluster_time(self) -> Optional[ClusterTime]: - """Return a document, the highest seen $clusterTime.""" - return self._max_cluster_time - - def _receive_cluster_time_no_lock(self, cluster_time: Optional[Mapping[str, Any]]) -> None: - # Driver Sessions Spec: "Whenever a driver receives a cluster time from - # a server it MUST compare it to the current highest seen cluster time - # for the deployment. If the new cluster time is higher than the - # highest seen cluster time it MUST become the new highest seen cluster - # time. Two cluster times are compared using only the BsonTimestamp - # value of the clusterTime embedded field." - if cluster_time: - # ">" uses bson.timestamp.Timestamp's comparison operator. - if ( - not self._max_cluster_time - or cluster_time["clusterTime"] > self._max_cluster_time["clusterTime"] - ): - self._max_cluster_time = cluster_time - async def receive_cluster_time(self, cluster_time: Optional[Mapping[str, Any]]) -> None: async with self._lock: self._receive_cluster_time_no_lock(cluster_time) @@ -579,15 +490,6 @@ async def request_check_all(self, wait_time: int = 5) -> None: self._request_check_all() await _async_cond_wait(self._condition, wait_time) - def data_bearing_servers(self) -> list[ServerDescription]: - """Return a list of all data-bearing servers. - - 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 - async def update_pool(self) -> None: # Remove any stale sockets and add new sockets if pool is too small. servers = [] @@ -646,32 +548,10 @@ async def close(self) -> None: if self._sdam._publish_server or self._sdam._publish_tp: # Make sure the events executor thread is fully closed before publishing the remaining events - self.__events_executor.close() - await self.__events_executor.join(1) + self._events_executor.close() + await self._events_executor.join(1) process_events_queue(weakref.ref(self._events)) # type: ignore[arg-type] - @property - def description(self) -> TopologyDescription: - return self._description - - def pop_all_sessions(self) -> list[_ServerSession]: - """Pop all session ids from the pool.""" - return self._session_pool.pop_all() - - def get_server_session(self, session_timeout_minutes: Optional[int]) -> _ServerSession: - """Start or resume a server session, or raise ConfigurationError.""" - return self._session_pool.get_server_session(session_timeout_minutes) - - def return_server_session(self, server_session: _ServerSession) -> None: - self._session_pool.return_server_session(server_session) - - def _new_selection(self) -> Selection: - """A Selection object, initially including all known servers. - - Hold the lock when calling this. - """ - return Selection.from_topology_description(self._description) - async def _ensure_opened(self) -> None: """Start monitors, or restart after a fork. @@ -686,7 +566,7 @@ async def _ensure_opened(self) -> None: # Start or restart the events publishing thread. if self._sdam._publish_tp or self._sdam._publish_server: - self.__events_executor.open() + self._events_executor.open() # Start the SRV polling thread. if self._srv_monitor and (self.description.topology_type in SRV_POLLING_TOPOLOGIES): @@ -705,26 +585,6 @@ async def _ensure_opened(self) -> None: for server in self._servers.values(): await server.open() - def _is_stale_error(self, address: _Address, err_ctx: _ErrorContext) -> bool: - server = self._servers.get(address) - if server is None: - # Another thread removed this server from the topology. - return True - - if server._pool.stale_generation(err_ctx.sock_generation, err_ctx.service_id): - # This is an outdated error from a previous pool version. - 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) - async def _handle_error(self, address: _Address, err_ctx: _ErrorContext) -> None: if self._is_stale_error(address, err_ctx): return @@ -802,11 +662,6 @@ async def handle_error(self, address: _Address, err_ctx: _ErrorContext) -> None: async with self._lock: await self._handle_error(address, err_ctx) - def _request_check_all(self) -> None: - """Wake all monitors. Hold the lock when calling this.""" - for server in self._servers.values(): - server.request_check() - async def _update_servers(self) -> None: """Sync our Servers from TopologyDescription.server_descriptions. @@ -851,88 +706,6 @@ async def _update_servers(self) -> None: self._monitor_tasks.append(server._monitor) 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 - ) - - def _create_pool_for_monitor(self, address: _Address) -> Pool: - options = self._settings.pool_options - - # According to the Server Discovery And Monitoring Spec, monitors use - # connect_timeout for both connect_timeout and socket_timeout. The - # pool only has one socket so maxPoolSize and so on aren't needed. - monitor_pool_options = PoolOptions( - connect_timeout=options.connect_timeout, - socket_timeout=options.connect_timeout, - ssl_context=options._ssl_context, - tls_allow_invalid_hostnames=options.tls_allow_invalid_hostnames, - event_listeners=options._event_listeners, - appname=options.appname, - driver=options.driver, - pause_enabled=False, - server_api=options.server_api, - ) - - return self._settings.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, - ) - - 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: @@ -941,22 +714,3 @@ async def cleanup_monitors(self) -> None: except IndexError: pass await asyncio.gather(*[t.join() for t in tasks], return_exceptions=True) # type: ignore[func-returns-value] - - def __repr__(self) -> str: - msg = "" - if not self._opened: - msg = "CLOSED " - return f"<{self.__class__.__name__} {msg}{self._description!r}>" - - def eq_props(self) -> tuple[tuple[_Address, ...], Optional[str], Optional[str], str]: - """The properties to use for AsyncMongoClient/Topology equality checks.""" - ts = self._settings - return (tuple(sorted(ts.seeds)), ts.replica_set_name, ts.fqdn, ts.srv_service_name) - - def __eq__(self, other: object) -> bool: - if isinstance(other, self.__class__): - return self.eq_props() == other.eq_props() - return NotImplemented - - def __hash__(self) -> int: - return hash(self.eq_props()) diff --git a/pymongo/synchronous/topology.py b/pymongo/synchronous/topology.py index c6468c8912..dc52b36726 100644 --- a/pymongo/synchronous/topology.py +++ b/pymongo/synchronous/topology.py @@ -30,11 +30,9 @@ from pymongo import _csot, common, helpers_shared, periodic_executor from pymongo._telemetry import ( - _SdamTelemetry, _ServerSelectionTelemetry, log_server_selection_succeeded, ) -from pymongo.client_session_shared import _ServerSession, _ServerSessionPool from pymongo.errors import ( ConnectionFailure, InvalidOperation, @@ -47,13 +45,8 @@ WriteError, ) from pymongo.hello import Hello -from pymongo.lock import ( - _cond_wait, - _create_condition, - _create_lock, -) +from pymongo.lock import _cond_wait, _create_condition, _create_lock from pymongo.logger import _SERVER_SELECTION_LOGGER, _is_debug_enabled -from pymongo.pool_options import PoolOptions from pymongo.server_description import ServerDescription from pymongo.server_selectors import ( Selection, @@ -62,7 +55,7 @@ secondary_server_selector, writable_server_selector, ) -from pymongo.synchronous.monitor import MonitorBase, SrvMonitor +from pymongo.synchronous.monitor import SrvMonitor from pymongo.synchronous.pool import Pool from pymongo.synchronous.server import Server from pymongo.topology_description import ( @@ -73,71 +66,32 @@ updated_topology_description, ) from pymongo.topology_shared import ( + _AgnosticTopologyBase, _ErrorContext, - _is_stale_error_topology_version, _is_stale_server_description, process_events_queue, ) if TYPE_CHECKING: from pymongo.synchronous.settings import TopologySettings - from pymongo.typings import ClusterTime, _Address + from pymongo.typings import _Address _IS_SYNC = True _pymongo_dir = str(Path(__file__).parent) -class Topology: +class Topology(_AgnosticTopologyBase[Pool, Server]): """Monitor a topology of one or more servers.""" - def __init__(self, topology_settings: TopologySettings): - self._topology_id = topology_settings._topology_id - self._listeners = topology_settings._pool_options._event_listeners - - # Create events queue if there are publishers. - self._events: queue.Queue[Any] | None = None - self.__events_executor: Any = None - - publish_server = self._listeners is not None and self._listeners.enabled_for_server - publish_tp = self._listeners is not None and self._listeners.enabled_for_topology - if publish_server or publish_tp: - self._events = queue.Queue(maxsize=100) - - self._sdam = _SdamTelemetry(self._topology_id, self._listeners, self._events) - self._sdam.topology_opened() - - self._settings = topology_settings - topology_description = TopologyDescription( - topology_settings.get_topology_type(), - topology_settings.get_server_descriptions(), - topology_settings.replica_set_name, - None, - None, - topology_settings, - ) + _settings: TopologySettings - self._description = topology_description - initial_td = TopologyDescription( - TOPOLOGY_TYPE.Unknown, {}, None, None, None, self._settings - ) - self._sdam.topology_description_changed(initial_td, self._description) - - for seed in topology_settings.seeds: - self._sdam.server_opened(seed) - - # Store the seed list to help diagnose errors in _error_message(). - self._seed_addresses = list(topology_description.server_descriptions()) - self._opened = False - self._closed = False + def __init__(self, topology_settings: TopologySettings): + super().__init__(topology_settings) self._lock = _create_lock() self._condition = _create_condition( self._lock, self._settings.condition_class if _IS_SYNC else None ) - self._servers: dict[_Address, Server] = {} - self._pid: Optional[int] = None - self._max_cluster_time: Optional[ClusterTime] = None - self._session_pool = _ServerSessionPool() if self._sdam._publish_server or self._sdam._publish_tp: assert self._events is not None @@ -157,16 +111,12 @@ def target() -> bool: # the queue via this closure. When the topology is freed, stop # the executor soon. weak = weakref.ref(self._events, executor.close) - self.__events_executor = executor + self._events_executor = executor executor.open() - self._srv_monitor = None if self._settings.fqdn is not None and not self._settings.load_balanced: self._srv_monitor = SrvMonitor(self, self._settings) - # Stores all monitor tasks that need to be joined on close or server selection - self._monitor_tasks: list[MonitorBase] = [] - def open(self) -> None: """Start monitoring, or restart after a fork. @@ -205,13 +155,6 @@ def open(self) -> None: with self._lock: self._ensure_opened() - def get_server_selection_timeout(self) -> float: - # CSOT: use remaining timeout when set. - timeout = _csot.remaining() - if timeout is None: - return self._settings.server_selection_timeout - return timeout - def select_servers( self, selector: Callable[[Selection], Selection], @@ -504,19 +447,6 @@ def on_srv_update(self, seedlist: list[tuple[str, Any]]) -> None: if self._opened: self._process_srv_update(seedlist) - def get_server_by_address(self, address: _Address) -> Optional[Server]: - """Get a Server or None. - - Returns the current version of the server immediately, even if it's - Unknown or absent from the topology. Only use this in unittests. - In driver code, use select_server_by_address, since then you're - assured a recent view of the server's type and wire protocol version. - """ - return self._servers.get(address) - - def has_server(self, address: _Address) -> bool: - return address in self._servers - def get_primary(self) -> Optional[_Address]: """Return primary's address or None.""" # Implemented here in Topology instead of MongoClient, so it can lock. @@ -548,25 +478,6 @@ def get_arbiters(self) -> set[_Address]: """Return set of arbiter addresses.""" return self._get_replica_set_members(arbiter_server_selector) - def max_cluster_time(self) -> Optional[ClusterTime]: - """Return a document, the highest seen $clusterTime.""" - return self._max_cluster_time - - def _receive_cluster_time_no_lock(self, cluster_time: Optional[Mapping[str, Any]]) -> None: - # Driver Sessions Spec: "Whenever a driver receives a cluster time from - # a server it MUST compare it to the current highest seen cluster time - # for the deployment. If the new cluster time is higher than the - # highest seen cluster time it MUST become the new highest seen cluster - # time. Two cluster times are compared using only the BsonTimestamp - # value of the clusterTime embedded field." - if cluster_time: - # ">" uses bson.timestamp.Timestamp's comparison operator. - if ( - not self._max_cluster_time - or cluster_time["clusterTime"] > self._max_cluster_time["clusterTime"] - ): - self._max_cluster_time = cluster_time - def receive_cluster_time(self, cluster_time: Optional[Mapping[str, Any]]) -> None: with self._lock: self._receive_cluster_time_no_lock(cluster_time) @@ -577,15 +488,6 @@ def request_check_all(self, wait_time: int = 5) -> None: self._request_check_all() _cond_wait(self._condition, wait_time) - def data_bearing_servers(self) -> list[ServerDescription]: - """Return a list of all data-bearing servers. - - 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 - def update_pool(self) -> None: # Remove any stale sockets and add new sockets if pool is too small. servers = [] @@ -644,32 +546,10 @@ def close(self) -> None: if self._sdam._publish_server or self._sdam._publish_tp: # Make sure the events executor thread is fully closed before publishing the remaining events - self.__events_executor.close() - self.__events_executor.join(1) + self._events_executor.close() + self._events_executor.join(1) process_events_queue(weakref.ref(self._events)) # type: ignore[arg-type] - @property - def description(self) -> TopologyDescription: - return self._description - - def pop_all_sessions(self) -> list[_ServerSession]: - """Pop all session ids from the pool.""" - return self._session_pool.pop_all() - - def get_server_session(self, session_timeout_minutes: Optional[int]) -> _ServerSession: - """Start or resume a server session, or raise ConfigurationError.""" - return self._session_pool.get_server_session(session_timeout_minutes) - - def return_server_session(self, server_session: _ServerSession) -> None: - self._session_pool.return_server_session(server_session) - - def _new_selection(self) -> Selection: - """A Selection object, initially including all known servers. - - Hold the lock when calling this. - """ - return Selection.from_topology_description(self._description) - def _ensure_opened(self) -> None: """Start monitors, or restart after a fork. @@ -684,7 +564,7 @@ def _ensure_opened(self) -> None: # Start or restart the events publishing thread. if self._sdam._publish_tp or self._sdam._publish_server: - self.__events_executor.open() + self._events_executor.open() # Start the SRV polling thread. if self._srv_monitor and (self.description.topology_type in SRV_POLLING_TOPOLOGIES): @@ -703,26 +583,6 @@ def _ensure_opened(self) -> None: for server in self._servers.values(): server.open() - def _is_stale_error(self, address: _Address, err_ctx: _ErrorContext) -> bool: - server = self._servers.get(address) - if server is None: - # Another thread removed this server from the topology. - return True - - if server._pool.stale_generation(err_ctx.sock_generation, err_ctx.service_id): - # This is an outdated error from a previous pool version. - 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) - def _handle_error(self, address: _Address, err_ctx: _ErrorContext) -> None: if self._is_stale_error(address, err_ctx): return @@ -800,11 +660,6 @@ def handle_error(self, address: _Address, err_ctx: _ErrorContext) -> None: with self._lock: self._handle_error(address, err_ctx) - def _request_check_all(self) -> None: - """Wake all monitors. Hold the lock when calling this.""" - for server in self._servers.values(): - server.request_check() - def _update_servers(self) -> None: """Sync our Servers from TopologyDescription.server_descriptions. @@ -849,88 +704,6 @@ def _update_servers(self) -> None: self._monitor_tasks.append(server._monitor) 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 - ) - - def _create_pool_for_monitor(self, address: _Address) -> Pool: - options = self._settings.pool_options - - # According to the Server Discovery And Monitoring Spec, monitors use - # connect_timeout for both connect_timeout and socket_timeout. The - # pool only has one socket so maxPoolSize and so on aren't needed. - monitor_pool_options = PoolOptions( - connect_timeout=options.connect_timeout, - socket_timeout=options.connect_timeout, - ssl_context=options._ssl_context, - tls_allow_invalid_hostnames=options.tls_allow_invalid_hostnames, - event_listeners=options._event_listeners, - appname=options.appname, - driver=options.driver, - pause_enabled=False, - server_api=options.server_api, - ) - - return self._settings.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, - ) - - 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: @@ -939,22 +712,3 @@ def cleanup_monitors(self) -> None: except IndexError: pass asyncio.gather(*[t.join() for t in tasks], return_exceptions=True) # type: ignore[func-returns-value] - - def __repr__(self) -> str: - msg = "" - if not self._opened: - msg = "CLOSED " - return f"<{self.__class__.__name__} {msg}{self._description!r}>" - - def eq_props(self) -> tuple[tuple[_Address, ...], Optional[str], Optional[str], str]: - """The properties to use for MongoClient/Topology equality checks.""" - ts = self._settings - return (tuple(sorted(ts.seeds)), ts.replica_set_name, ts.fqdn, ts.srv_service_name) - - def __eq__(self, other: object) -> bool: - if isinstance(other, self.__class__): - return self.eq_props() == other.eq_props() - return NotImplemented - - def __hash__(self) -> int: - return hash(self.eq_props()) diff --git a/pymongo/topology_shared.py b/pymongo/topology_shared.py index 23ef6a8f91..095399ed31 100644 --- a/pymongo/topology_shared.py +++ b/pymongo/topology_shared.py @@ -19,11 +19,23 @@ import queue import weakref from collections.abc import Mapping -from typing import TYPE_CHECKING, Any, Optional +from typing import TYPE_CHECKING, Any, Callable, Generic, Optional, TypeVar + +from pymongo import _csot +from pymongo._telemetry import _SdamTelemetry +from pymongo.client_session_shared import _ServerSession, _ServerSessionPool +from pymongo.pool_options import PoolOptions +from pymongo.server_selectors import Selection, writable_server_selector +from pymongo.topology_description import TOPOLOGY_TYPE, TopologyDescription if TYPE_CHECKING: from bson import ObjectId from pymongo.server_description import ServerDescription + from pymongo.settings_shared import _BaseTopologySettings + from pymongo.typings import ClusterTime, _Address, _AgnosticPool, _AgnosticServer + +_PoolT = TypeVar("_PoolT", bound="_AgnosticPool") +_ServerT = TypeVar("_ServerT", bound="_AgnosticServer") def process_events_queue(queue_ref: weakref.ReferenceType[queue.Queue]) -> bool: # type: ignore[type-arg] @@ -80,3 +92,254 @@ def _is_stale_server_description(current_sd: ServerDescription, new_sd: ServerDe if current_tv["processId"] != new_tv["processId"]: return False return current_tv["counter"] > new_tv["counter"] + + +class _AgnosticTopologyBase(Generic[_PoolT, _ServerT]): + """Monitor a topology of one or more servers.""" + + def __init__(self, topology_settings: _BaseTopologySettings[Any, Any]) -> None: + self._topology_id = topology_settings._topology_id + self._listeners = topology_settings._pool_options._event_listeners + + # Create events queue if there are publishers. + self._events: queue.Queue[Any] | None = None + self._events_executor: Any = None + + publish_server = self._listeners is not None and self._listeners.enabled_for_server + publish_tp = self._listeners is not None and self._listeners.enabled_for_topology + if publish_server or publish_tp: + self._events = queue.Queue(maxsize=100) + + self._sdam = _SdamTelemetry(self._topology_id, self._listeners, self._events) + self._sdam.topology_opened() + + self._settings = topology_settings + topology_description = TopologyDescription( + topology_settings.get_topology_type(), + topology_settings.get_server_descriptions(), + topology_settings.replica_set_name, + None, + None, + topology_settings, + ) + + self._description = topology_description + initial_td = TopologyDescription( + TOPOLOGY_TYPE.Unknown, {}, None, None, None, self._settings + ) + self._sdam.topology_description_changed(initial_td, self._description) + + for seed in topology_settings.seeds: + self._sdam.server_opened(seed) + + # Store the seed list to help diagnose errors in _error_message(). + self._seed_addresses = list(topology_description.server_descriptions()) + self._opened = False + self._closed = False + self._servers: dict[_Address, _ServerT] = {} + self._pid: Optional[int] = None + self._max_cluster_time: Optional[ClusterTime] = None + self._session_pool = _ServerSessionPool() + # The lock, condition, events executor, SRV monitor, and monitor tasks + # are IO-specific and are created by the async/sync subclasses. + self._srv_monitor: Any = None + # Stores all monitor tasks that need to be joined on close or server selection + self._monitor_tasks: list[Any] = [] + + def get_server_selection_timeout(self) -> float: + # CSOT: use remaining timeout when set. + timeout = _csot.remaining() + if timeout is None: + return self._settings.server_selection_timeout + return timeout + + def get_server_by_address(self, address: _Address) -> Optional[_ServerT]: + """Get a Server or None. + + Returns the current version of the server immediately, even if it's + Unknown or absent from the topology. Only use this in unittests. + In driver code, use select_server_by_address, since then you're + assured a recent view of the server's type and wire protocol version. + """ + return self._servers.get(address) + + def has_server(self, address: _Address) -> bool: + return address in self._servers + + def max_cluster_time(self) -> Optional[ClusterTime]: + """Return a document, the highest seen $clusterTime.""" + return self._max_cluster_time + + def _receive_cluster_time_no_lock(self, cluster_time: Optional[Mapping[str, Any]]) -> None: + # Driver Sessions Spec: "Whenever a driver receives a cluster time from + # a server it MUST compare it to the current highest seen cluster time + # for the deployment. If the new cluster time is higher than the + # highest seen cluster time it MUST become the new highest seen cluster + # time. Two cluster times are compared using only the BsonTimestamp + # value of the clusterTime embedded field." + if cluster_time: + # ">" uses bson.timestamp.Timestamp's comparison operator. + if ( + not self._max_cluster_time + or cluster_time["clusterTime"] > self._max_cluster_time["clusterTime"] + ): + self._max_cluster_time = cluster_time + + def data_bearing_servers(self) -> list[ServerDescription]: + """Return a list of all data-bearing servers. + + 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 + + @property + def description(self) -> TopologyDescription: + return self._description + + def pop_all_sessions(self) -> list[_ServerSession]: + """Pop all session ids from the pool.""" + return self._session_pool.pop_all() + + def get_server_session(self, session_timeout_minutes: Optional[int]) -> _ServerSession: + """Start or resume a server session, or raise ConfigurationError.""" + return self._session_pool.get_server_session(session_timeout_minutes) + + def return_server_session(self, server_session: _ServerSession) -> None: + self._session_pool.return_server_session(server_session) + + def _new_selection(self) -> Selection: + """A Selection object, initially including all known servers. + + Hold the lock when calling this. + """ + return Selection.from_topology_description(self._description) + + def _is_stale_error(self, address: _Address, err_ctx: _ErrorContext) -> bool: + server = self._servers.get(address) + if server is None: + # Another thread removed this server from the topology. + return True + + if server._pool.stale_generation(err_ctx.sock_generation, err_ctx.service_id): + # This is an outdated error from a previous pool version. + 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) + + def _request_check_all(self) -> None: + """Wake all monitors. Hold the lock when calling this.""" + for server in self._servers.values(): + server.request_check() + + def _create_pool_for_server(self, address: _Address) -> _PoolT: + return self._settings.pool_class( + address, self._settings.pool_options, client_id=self._topology_id + ) + + def _create_pool_for_monitor(self, address: _Address) -> _PoolT: + options = self._settings.pool_options + + # According to the Server Discovery And Monitoring Spec, monitors use + # connect_timeout for both connect_timeout and socket_timeout. The + # pool only has one socket so maxPoolSize and so on aren't needed. + monitor_pool_options = PoolOptions( + connect_timeout=options.connect_timeout, + socket_timeout=options.connect_timeout, + ssl_context=options._ssl_context, + tls_allow_invalid_hostnames=options.tls_allow_invalid_hostnames, + event_listeners=options._event_listeners, + appname=options.appname, + driver=options.driver, + pause_enabled=False, + server_api=options.server_api, + ) + + return self._settings.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, + ) + + 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 __repr__(self) -> str: + msg = "" + if not self._opened: + msg = "CLOSED " + return f"<{self.__class__.__name__} {msg}{self._description!r}>" + + def eq_props(self) -> tuple[tuple[_Address, ...], Optional[str], Optional[str], str]: + """The properties to use for MongoClient/AsyncMongoClient and Topology + equality checks. + """ + ts = self._settings + return (tuple(sorted(ts.seeds)), ts.replica_set_name, ts.fqdn, ts.srv_service_name) + + def __eq__(self, other: object) -> bool: + if isinstance(other, self.__class__): + return self.eq_props() == other.eq_props() + return NotImplemented + + def __hash__(self) -> int: + return hash(self.eq_props()) diff --git a/pymongo/typings.py b/pymongo/typings.py index f90cf4b768..33e07b356b 100644 --- a/pymongo/typings.py +++ b/pymongo/typings.py @@ -34,6 +34,7 @@ from pymongo.asynchronous.collection import AsyncCollection from pymongo.asynchronous.mongo_client import AsyncMongoClient from pymongo.asynchronous.pool import AsyncConnection + from pymongo.asynchronous.pool import Pool as AsyncPool from pymongo.asynchronous.server import Server as AsyncServer from pymongo.collation import Collation from pymongo.synchronous.bulk import _Bulk @@ -41,7 +42,7 @@ from pymongo.synchronous.client_session import ClientSession from pymongo.synchronous.collection import Collection from pymongo.synchronous.mongo_client import MongoClient - from pymongo.synchronous.pool import Connection + from pymongo.synchronous.pool import Connection, Pool from pymongo.synchronous.server import Server @@ -58,6 +59,7 @@ _AgnosticCollection = Union["AsyncCollection[_DocumentType]", "Collection[_DocumentType]"] _AgnosticConnection = Union["AsyncConnection", "Connection"] _AgnosticServer = Union["AsyncServer", "Server"] +_AgnosticPool = Union["AsyncPool", "Pool"] _AgnosticClientSession = Union["AsyncClientSession", "ClientSession"] _AgnosticBulk = Union["_AsyncBulk", "_Bulk"] _AgnosticClientBulk = Union["_AsyncClientBulk", "_ClientBulk"] From af744a9cf14db6549706d16550223645fa5d065a Mon Sep 17 00:00:00 2001 From: Iris Date: Thu, 8 Oct 2026 12:10:17 -0700 Subject: [PATCH 2/3] better typing and fix tests --- pymongo/asynchronous/topology.py | 5 ++++- pymongo/synchronous/topology.py | 5 ++++- test/asynchronous/test_monitor.py | 2 +- test/helpers_shared.py | 2 +- test/test_monitor.py | 2 +- 5 files changed, 11 insertions(+), 5 deletions(-) diff --git a/pymongo/asynchronous/topology.py b/pymongo/asynchronous/topology.py index 268873cd32..8f1b4b1a21 100644 --- a/pymongo/asynchronous/topology.py +++ b/pymongo/asynchronous/topology.py @@ -33,7 +33,7 @@ _ServerSelectionTelemetry, log_server_selection_succeeded, ) -from pymongo.asynchronous.monitor import SrvMonitor +from pymongo.asynchronous.monitor import MonitorBase, SrvMonitor from pymongo.asynchronous.pool import Pool from pymongo.asynchronous.server import Server from pymongo.errors import ( @@ -85,6 +85,9 @@ class Topology(_AgnosticTopologyBase[Pool, Server]): """Monitor a topology of one or more servers.""" _settings: TopologySettings + _events_executor: periodic_executor.AsyncPeriodicExecutor + _srv_monitor: Optional[SrvMonitor] + _monitor_tasks: list[MonitorBase] def __init__(self, topology_settings: TopologySettings): super().__init__(topology_settings) diff --git a/pymongo/synchronous/topology.py b/pymongo/synchronous/topology.py index dc52b36726..9accc15dc1 100644 --- a/pymongo/synchronous/topology.py +++ b/pymongo/synchronous/topology.py @@ -55,7 +55,7 @@ secondary_server_selector, writable_server_selector, ) -from pymongo.synchronous.monitor import SrvMonitor +from pymongo.synchronous.monitor import MonitorBase, SrvMonitor from pymongo.synchronous.pool import Pool from pymongo.synchronous.server import Server from pymongo.topology_description import ( @@ -85,6 +85,9 @@ class Topology(_AgnosticTopologyBase[Pool, Server]): """Monitor a topology of one or more servers.""" _settings: TopologySettings + _events_executor: periodic_executor.PeriodicExecutor + _srv_monitor: Optional[SrvMonitor] + _monitor_tasks: list[MonitorBase] def __init__(self, topology_settings: TopologySettings): super().__init__(topology_settings) diff --git a/test/asynchronous/test_monitor.py b/test/asynchronous/test_monitor.py index ed4a025580..12aed6e899 100644 --- a/test/asynchronous/test_monitor.py +++ b/test/asynchronous/test_monitor.py @@ -50,7 +50,7 @@ def get_executors(client): executors.append(server._monitor._executor) executors.append(server._monitor._rtt_monitor._executor) executors.append(client._kill_cursors_executor) - executors.append(client._topology._Topology__events_executor) + executors.append(client._topology._events_executor) return [e for e in executors if e is not None] diff --git a/test/helpers_shared.py b/test/helpers_shared.py index 49cf131808..9b8c49cd6f 100644 --- a/test/helpers_shared.py +++ b/test/helpers_shared.py @@ -212,7 +212,7 @@ def _get_executors(topology): executors.append(server._monitor._executor) if hasattr(server._monitor, "_rtt_monitor"): executors.append(server._monitor._rtt_monitor._executor) - executors.append(topology._Topology__events_executor) + executors.append(topology._events_executor) if topology._srv_monitor: executors.append(topology._srv_monitor._executor) diff --git a/test/test_monitor.py b/test/test_monitor.py index 8fd427cb3a..6369dafc77 100644 --- a/test/test_monitor.py +++ b/test/test_monitor.py @@ -50,7 +50,7 @@ def get_executors(client): executors.append(server._monitor._executor) executors.append(server._monitor._rtt_monitor._executor) executors.append(client._kill_cursors_executor) - executors.append(client._topology._Topology__events_executor) + executors.append(client._topology._events_executor) return [e for e in executors if e is not None] From 47d6c9cf9972f077d4c5063b9fbd1b0c88e09df4 Mon Sep 17 00:00:00 2001 From: Iris Date: Wed, 14 Oct 2026 14:20:39 -0700 Subject: [PATCH 3/3] AC feedback --- pymongo/topology_shared.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pymongo/topology_shared.py b/pymongo/topology_shared.py index 095399ed31..50bf0c8ad1 100644 --- a/pymongo/topology_shared.py +++ b/pymongo/topology_shared.py @@ -95,7 +95,7 @@ def _is_stale_server_description(current_sd: ServerDescription, new_sd: ServerDe class _AgnosticTopologyBase(Generic[_PoolT, _ServerT]): - """Monitor a topology of one or more servers.""" + """Shared base for the sync and async classes.""" def __init__(self, topology_settings: _BaseTopologySettings[Any, Any]) -> None: self._topology_id = topology_settings._topology_id