PYTHON-2143 Use an allow-list to determine resumable change stream errors

Author: prashantmital

pymongo/change_stream.py

@@ -25,18 +25,33 @@
 from pymongo.collation import validate_collation_or_none
 from pymongo.command_cursor import CommandCursor
 from pymongo.errors import (ConnectionFailure,
+                            CursorNotFound,
                             InvalidOperation,
                             OperationFailure,
                             PyMongoError)
 
 
 # The change streams spec considers the following server errors from the
 # getMore command non-resumable. All other getMore errors are resumable.
-_NON_RESUMABLE_GETMORE_ERRORS = frozenset([
-    11601,  # Interrupted
-    136,    # CappedPositionLost
-    237,    # CursorKilled
-    None,   # No error code was returned.
+_RESUMABLE_GETMORE_ERRORS = frozenset([
+    6,      # HostUnreachable
+    7,      # HostNotFound
+    89,     # NetworkTimeout
+    91,     # ShutdownInProgress
+    189,    # PrimarySteppedDown
+    262,    # ExceededTimeLimit
+    9001,   # SocketException
+    10107,  # NotMaster
+    11600,  # InterruptedAtShutdown
+    11602,  # InterruptedDueToReplStateChange
+    13435,  # NotMasterNoSlaveOk
+    13436,  # NotMasterOrSecondary
+    63,     # StaleShardVersion
+    150,    # StaleEpoch
+    13388,  # StaleConfig
+    234,    # RetryChangeStream
+    133,    # FailedToSatisfyReadPreference
+    216,    # ElectionInProgress
 ])
 
 
@@ -283,12 +298,15 @@ def try_next(self):
         # one resume attempt.
         try:
             change = self._cursor._try_next(True)
-        except ConnectionFailure:
+        except (ConnectionFailure, CursorNotFound):
             self._resume()
             change = self._cursor._try_next(False)
         except OperationFailure as exc:
-            if (exc.code in _NON_RESUMABLE_GETMORE_ERRORS or
-                    exc.has_error_label("NonResumableChangeStreamError")):
+            is_resumable = ((exc.max_wire_version >= 9 and
+                             exc.has_error_label("ResumableChangeStreamError")) or
+                            (exc.max_wire_version < 9 and
+                             exc.code in _RESUMABLE_GETMORE_ERRORS))
+            if not is_resumable:
                 raise
             self._resume()
             change = self._cursor._try_next(False)

pymongo/errors.py

@@ -140,18 +140,21 @@ class ConfigurationError(PyMongoError):
 class OperationFailure(PyMongoError):
     """Raised when a database operation fails.
 
+    .. versionadded:: 3.11
+       The :attr:`max_wire_version` attribute.
     .. versionadded:: 2.7
        The :attr:`details` attribute.
     """
 
-    def __init__(self, error, code=None, details=None):
+    def __init__(self, error, code=None, details=None, max_wire_version=None):
         error_labels = None
         if details is not None:
             error_labels = details.get('errorLabels')
         super(OperationFailure, self).__init__(
             error, error_labels=error_labels)
         self.__code = code
         self.__details = details
+        self.__max_wire_version = max_wire_version
 
     @property
     def code(self):
@@ -171,6 +174,15 @@ def details(self):
         """
         return self.__details
 
+    @property
+    def max_wire_version(self):
+        """The latest version of the wire protocol supported by the socket
+        that was used to run the operation that raised this exception.
+
+        PyMongo does not always record this value and it may be None.
+        """
+        return self.__max_wire_version
+
     def __str__(self):
         output_str = "%s, full error: %s" % (self._message, self.__details)
         if sys.version_info[0] == 2 and isinstance(output_str, unicode):

pymongo/helpers.py

@@ -103,14 +103,16 @@ def _index_document(index_list):
 
 
 def _check_command_response(response, msg=None, allowable_errors=None,
-                            parse_write_concern_error=False):
+                            parse_write_concern_error=False,
+                            max_wire_version=None):
     """Check the response to a command for errors.
     """
     if "ok" not in response:
         # Server didn't recognize our message as a command.
         raise OperationFailure(response.get("$err"),
                                response.get("code"),
-                               response)
+                               response,
+                               max_wire_version)
 
     if parse_write_concern_error and 'writeConcernError' in response:
         _raise_write_concern_error(response['writeConcernError'])
@@ -146,19 +148,24 @@ def _check_command_response(response, msg=None, allowable_errors=None,
                           details.get("assertion", ""))
                 raise OperationFailure(errmsg,
                                        details.get("assertionCode"),
-                                       response)
+                                       response,
+                                       max_wire_version)
 
             # Other errors
             # findAndModify with upsert can raise duplicate key error
             if code in (11000, 11001, 12582):
-                raise DuplicateKeyError(errmsg, code, response)
+                raise DuplicateKeyError(errmsg, code, response,
+                                        max_wire_version)
             elif code == 50:
-                raise ExecutionTimeout(errmsg, code, response)
+                raise ExecutionTimeout(errmsg, code, response,
+                                       max_wire_version)
             elif code == 43:
-                raise CursorNotFound(errmsg, code, response)
+                raise CursorNotFound(errmsg, code, response,
+                                     max_wire_version)
 
             msg = msg or "%s"
-            raise OperationFailure(msg % errmsg, code, response)
+            raise OperationFailure(msg % errmsg, code, response,
+                                   max_wire_version)
 
 
 def _check_gle_response(result):

pymongo/server.py

@@ -133,7 +133,8 @@ def run_operation_with_response(
                 first = docs[0]
                 operation.client._process_response(
                     first, operation.session)
-                _check_command_response(first)
+                _check_command_response(
+                    first, max_wire_version=sock_info.max_wire_version)
         except Exception as exc:
             if publish:
                 duration = datetime.now() - start

test/change_streams/change-streams-errors.json

@@ -75,7 +75,6 @@
     {
       "description": "Change Stream should error when _id is projected out",
       "minServerVersion": "4.1.11",
-      "maxServerVersion": "4.3.3",
       "target": "collection",
       "topology": [
         "replicaset",
@@ -103,10 +102,54 @@
       ],
       "result": {
         "error": {
-          "code": 280,
-          "errorLabels": [
-            "NonResumableChangeStreamError"
-          ]
+          "code": 280
+        }
+      }
+    },
+    {
+      "description": "change stream errors on MaxTimeMSExpired",
+      "minServerVersion": "4.2",
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 1
+        },
+        "data": {
+          "failCommands": [
+            "getMore"
+          ],
+          "errorCode": 50,
+          "closeConnection": false
+        }
+      },
+      "target": "collection",
+      "topology": [
+        "replicaset",
+        "sharded"
+      ],
+      "changeStreamPipeline": [
+        {
+          "$project": {
+            "_id": 0
+          }
+        }
+      ],
+      "changeStreamOptions": {},
+      "operations": [
+        {
+          "database": "change-stream-tests",
+          "collection": "test",
+          "name": "insertOne",
+          "arguments": {
+            "document": {
+              "z": 3
+            }
+          }
+        }
+      ],
+      "result": {
+        "error": {
+          "code": 50
         }
       }
     }