PYTHON-2199 Reduce race conditions in SDAM error handling

Use Pool.generation and topologyVersion to reduce race conditions
SDAM error handling.
Implement SDAM error handling spec tests.

Author: ShaneHarvey

pymongo/database.py

@@ -29,7 +29,6 @@
 from pymongo.errors import (CollectionInvalid,
                             ConfigurationError,
                             InvalidName,
-                            NotMasterError,
                             OperationFailure)
 from pymongo.message import _first_batch
 from pymongo.read_preferences import ReadPreference
@@ -1158,8 +1157,7 @@ def error(self):
             # doing so already.
             primary = self.__client.primary
             if primary:
-                self.__client._reset_server_and_request_check(
-                    primary, NotMasterError(error_msg, error))
+                self.__client._handle_getlasterror(primary, error_msg)
         return error
 
     def last_status(self):

pymongo/ismaster.py

@@ -168,3 +168,7 @@ def sasl_supported_mechs(self):
 
         """
         return self._doc.get('saslSupportedMechs', [])
+
+    @property
+    def topology_version(self):
+        return self._doc.get('topologyVersion')

pymongo/mongo_client.py

@@ -59,16 +59,15 @@
                             ConfigurationError,
                             ConnectionFailure,
                             InvalidOperation,
-                            NetworkTimeout,
-                            NotMasterError,
                             OperationFailure,
                             PyMongoError,
                             ServerSelectionTimeoutError)
 from pymongo.read_preferences import ReadPreference
 from pymongo.server_selectors import (writable_preferred_server_selector,
                                       writable_server_selector)
 from pymongo.server_type import SERVER_TYPE
-from pymongo.topology import Topology
+from pymongo.topology import (Topology,
+                              _ErrorContext)
 from pymongo.topology_description import TOPOLOGY_TYPE
 from pymongo.settings import TopologySettings
 from pymongo.uri_parser import (_handle_option_deprecations,
@@ -1225,8 +1224,7 @@ def _get_topology(self):
 
     @contextlib.contextmanager
     def _get_socket(self, server, session, exhaust=False):
-        with _MongoClientErrorHandler(
-                self, server.description.address, session) as err_handler:
+        with _MongoClientErrorHandler(self, server, session) as err_handler:
             with server.get_socket(
                     self.__all_credentials, checkout=exhaust) as sock_info:
                 err_handler.contribute_socket(sock_info)
@@ -1328,8 +1326,7 @@ def _run_operation_with_response(self, operation, unpack_res,
                 operation.read_preference, operation.session, address=address)
 
             with _MongoClientErrorHandler(
-                    self, server.description.address,
-                    operation.session) as err_handler:
+                    self, server, operation.session) as err_handler:
                 err_handler.contribute_socket(operation.exhaust_mgr.sock)
                 return server.run_operation_with_response(
                     operation.exhaust_mgr.sock,
@@ -1499,9 +1496,9 @@ def _retryable_write(self, retryable, func, session):
         with self._tmp_session(session) as s:
             return self._retry_with_session(retryable, func, s, None)
 
-    def _reset_server_and_request_check(self, address, error):
+    def _handle_getlasterror(self, address, error_msg):
         """Clear our pool for a server, mark it Unknown, and check it soon."""
-        self._topology.reset_server_and_request_check(address, error)
+        self._topology.handle_getlasterror(address, error_msg)
 
     def __eq__(self, other):
         if isinstance(other, self.__class__):
@@ -2167,18 +2164,24 @@ def __next__(self):
 
 class _MongoClientErrorHandler(object):
     """Error handler for MongoClient."""
-    __slots__ = ('_client', '_server_address', '_session', '_max_wire_version')
+    __slots__ = ('_client', '_server_address', '_session',
+                 '_max_wire_version', '_sock_generation')
 
-    def __init__(self, client, server_address, session):
+    def __init__(self, client, server, session):
         self._client = client
-        self._server_address = server_address
+        self._server_address = server.description.address
         self._session = session
         self._max_wire_version = common.MIN_WIRE_VERSION
+        # XXX: When get_socket fails, this generation could be out of date:
+        # "Note that when a network error occurs before the handshake
+        # completes then the error's generation number is the generation
+        # of the pool at the time the connection attempt was started."
+        self._sock_generation = server.pool.generation
 
     def contribute_socket(self, sock_info):
         """Provide socket information to the error handler."""
-        # Currently, we only extract the max_wire_version information.
         self._max_wire_version = sock_info.max_wire_version
+        self._sock_generation = sock_info.generation
 
     def __enter__(self):
         return self
@@ -2187,45 +2190,15 @@ def __exit__(self, exc_type, exc_val, exc_tb):
         if exc_type is None:
             return
 
+        err_ctx = _ErrorContext(
+            exc_val, self._max_wire_version, self._sock_generation)
+        self._client._topology.handle_error(self._server_address, err_ctx)
+
         if issubclass(exc_type, PyMongoError):
             if self._session and exc_val.has_error_label(
                     "TransientTransactionError"):
                 self._session._unpin_mongos()
 
-        if issubclass(exc_type, NetworkTimeout):
-            # 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...."
-            if self._session:
-                self._session._server_session.mark_dirty()
-        elif issubclass(exc_type, NotMasterError):
-            # As per the SDAM spec if:
-            #   - the server sees a "not master" 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.
-            err_code = exc_val.details.get('code', -1)
-            is_shutting_down = err_code in helpers._SHUTDOWN_CODES
-            if is_shutting_down or (self._max_wire_version <= 7):
-                # Clear the pool, mark server Unknown and request check.
-                self._client._reset_server_and_request_check(
-                    self._server_address, exc_val)
-            else:
-                self._client._topology.mark_server_unknown_and_request_check(
-                    self._server_address, exc_val)
-        elif issubclass(exc_type, ConnectionFailure):
-            # "Client MUST replace the server's description with type Unknown
-            # ... MUST NOT request an immediate check of the server."
-            self._client._topology.reset_server(self._server_address, exc_val)
+        if issubclass(exc_type, ConnectionFailure):
             if self._session:
                 self._session._server_session.mark_dirty()
-        elif issubclass(exc_type, OperationFailure):
-            # Do not request an immediate check since the server is likely
-            # shutting down.
-            if exc_val.code in helpers._RETRYABLE_ERROR_CODES:
-                self._client._topology.reset_server(
-                    self._server_address, exc_val)

pymongo/server_description.py

@@ -36,7 +36,8 @@ class ServerDescription(object):
         '_max_write_batch_size', '_min_wire_version', '_max_wire_version',
         '_round_trip_time', '_me', '_is_writable', '_is_readable',
         '_ls_timeout_minutes', '_error', '_set_version', '_election_id',
-        '_cluster_time', '_last_write_date', '_last_update_time')
+        '_cluster_time', '_last_write_date', '_last_update_time',
+        '_topology_version')
 
     def __init__(
             self,
@@ -68,6 +69,10 @@ def __init__(
         self._me = ismaster.me
         self._last_update_time = _time()
         self._error = error
+        self._topology_version = ismaster.topology_version
+        if error:
+            if hasattr(error, 'details') and isinstance(error.details, dict):
+                self._topology_version = error.details.get('topologyVersion')
 
         if ismaster.last_write_date:
             # Convert from datetime to seconds.
@@ -207,6 +212,15 @@ def retryable_reads_supported(self):
         """Checks if this server supports retryable writes."""
         return self._max_wire_version >= 6
 
+    @property
+    def topology_version(self):
+        return self._topology_version
+
+    def to_unknown(self):
+        unknown = ServerDescription(self.address)
+        unknown._topology_version = self.topology_version
+        return unknown
+
     def __eq__(self, other):
         if isinstance(other, ServerDescription):
             return ((self._address == other.address) and

pymongo/topology.py

@@ -26,14 +26,20 @@
 else:
     import Queue
 
-from pymongo import common
-from pymongo import periodic_executor
+from pymongo import (common,
+                     helpers,
+                     periodic_executor)
 from pymongo.pool import PoolOptions
 from pymongo.topology_description import (updated_topology_description,
                                           _updated_topology_description_srv_polling,
                                           TopologyDescription,
                                           SRV_POLLING_TOPOLOGIES, TOPOLOGY_TYPE)
-from pymongo.errors import ServerSelectionTimeoutError, ConfigurationError
+from pymongo.errors import (ConnectionFailure,
+                            ConfigurationError,
+                            NetworkTimeout,
+                            NotMasterError,
+                            OperationFailure,
+                            ServerSelectionTimeoutError)
 from pymongo.monitor import SrvMonitor
 from pymongo.monotonic import time as _time
 from pymongo.server import Server
@@ -264,14 +270,17 @@ def _process_change(self, server_description):
         Hold the lock when calling this.
         """
         td_old = self._description
-        old_server_description = td_old._server_descriptions[
-            server_description.address]
+        sd_old = td_old._server_descriptions[server_description.address]
+        if _is_stale_server_description(sd_old, server_description):
+            # This is a stale isMaster response. Ignore it.
+            return
+
         suppress_event = ((self._publish_server or self._publish_tp)
-                          and old_server_description == server_description)
+                          and sd_old == server_description)
         if self._publish_server and not suppress_event:
             self._events.put((
                 self._listeners.publish_server_description_changed,
-                (old_server_description, server_description,
+                (sd_old, server_description,
                  server_description.address, self._topology_id)))
 
         self._description = updated_topology_description(
@@ -410,25 +419,15 @@ def reset_pool(self, address):
             if server:
                 server.pool.reset()
 
-    def reset_server(self, address, error):
-        """Clear our pool for a server and mark it Unknown.
-
-        Do *not* request an immediate check.
-        """
-        with self._lock:
-            self._reset_server(address, reset_pool=True, error=error)
-
-    def reset_server_and_request_check(self, address, error):
+    def handle_getlasterror(self, address, error_msg):
         """Clear our pool for a server, mark it Unknown, and check it soon."""
+        error = NotMasterError(error_msg, {'code': 10107, 'errmsg': error_msg})
         with self._lock:
-            self._reset_server(address, reset_pool=True, error=error)
-            self._request_check(address)
-
-    def mark_server_unknown_and_request_check(self, address, error):
-        """Mark a server Unknown, and check it soon."""
-        with self._lock:
-            self._reset_server(address, reset_pool=False, error=error)
-            self._request_check(address)
+            server = self._servers.get(address)
+            if server:
+                self._process_change(ServerDescription(address, error=error))
+                server.pool.reset()
+                server.request_check()
 
     def update_pool(self, all_credentials):
         # Remove any stale sockets and add new sockets if pool is too small.
@@ -540,28 +539,78 @@ def _ensure_opened(self):
         for server in itervalues(self._servers):
             server.open()
 
-    def _reset_server(self, address, reset_pool, error):
-        """Mark a server Unknown and optionally reset it's pool.
-
-        Hold the lock when calling this. Does *not* request an immediate check.
-        """
+    def _is_stale_error(self, address, err_ctx):
         server = self._servers.get(address)
-
-        # "server" is None if another thread removed it from the topology.
-        if server:
-            if reset_pool:
+        if server is None:
+            # Another thread removed this server from the topology.
+            return True
+
+        if err_ctx.sock_generation != server._pool.generation:
+            # 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, err_ctx):
+        if self._is_stale_error(address, err_ctx):
+            return
+
+        server = self._servers[address]
+        error = err_ctx.error
+        exc_type = type(error)
+        if issubclass(exc_type, NetworkTimeout):
+            # 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 issubclass(exc_type, NotMasterError):
+            # As per the SDAM spec if:
+            #   - the server sees a "not master" 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.
+            err_code = error.details.get('code', -1)
+            is_shutting_down = err_code in helpers._SHUTDOWN_CODES
+            # Mark server Unknown, clear the pool, and request check.
+            self._process_change(ServerDescription(address, error=error))
+            if is_shutting_down or (err_ctx.max_wire_version <= 7):
+                # Clear the pool.
                 server.reset()
-
-            # Mark this server Unknown.
+            server.request_check()
+        elif issubclass(exc_type, ConnectionFailure):
+            # "Client MUST replace the server's description with type Unknown
+            # ... MUST NOT request an immediate check of the server."
             self._process_change(ServerDescription(address, error=error))
+            # Clear the pool.
+            server.reset()
+        elif issubclass(exc_type, OperationFailure):
+            # Do not request an immediate check since the server is likely
+            # shutting down.
+            if error.code in helpers._NOT_MASTER_CODES:
+                self._process_change(ServerDescription(address, error=error))
+                # Clear the pool.
+                server.reset()
 
-    def _request_check(self, address):
-        """Wake one monitor. Hold the lock when calling this."""
-        server = self._servers.get(address)
+    def handle_error(self, address, err_ctx):
+        """Handle an application error.
 
-        # "server" is None if another thread removed it from the topology.
-        if server:
-            server.request_check()
+        May reset the server to Unknown, clear the pool, and request an
+        immediate check depending on the error and the context.
+        """
+        with self._lock:
+            self._handle_error(address, err_ctx)
 
     def _request_check_all(self):
         """Wake all monitors. Hold the lock when calling this."""
@@ -692,3 +741,30 @@ def __repr__(self):
         if not self._opened:
             msg = 'CLOSED '
         return '<%s %s%r>' % (self.__class__.__name__, msg, self._description)
+
+
+class _ErrorContext(object):
+    """An error with context for SDAM error handling."""
+    def __init__(self, error, max_wire_version, sock_generation):
+        self.error = error
+        self.max_wire_version = max_wire_version
+        self.sock_generation = sock_generation
+
+
+def _is_stale_error_topology_version(current_tv, error_tv):
+    """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, new_sd):
+    """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']