PYTHON-2143 Use a whitelist to determine resumable change stream errors

Author: prashantmital

pymongo/change_stream.py

@@ -32,11 +32,25 @@
 
 # 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
 ])
 
 
@@ -287,8 +301,13 @@ def try_next(self):
             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")):
+            if exc.max_wire_version is None:
+                raise
+            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

@@ -136,18 +136,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):
@@ -167,6 +170,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
+
 
 class CursorNotFound(OperationFailure):
     """Raised while iterating query results if the cursor is

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
         }
       }
     }

test/test_change_stream.py

@@ -37,7 +37,7 @@
 from bson.raw_bson import DEFAULT_RAW_BSON_OPTIONS, RawBSONDocument
 
 from pymongo import MongoClient
-from pymongo.change_stream import _NON_RESUMABLE_GETMORE_ERRORS
+# from pymongo.change_stream import _NON_RESUMABLE_GETMORE_ERRORS
 from pymongo.command_cursor import CommandCursor
 from pymongo.errors import (InvalidOperation, OperationFailure,
                             ServerSelectionTimeoutError)
@@ -1075,7 +1075,7 @@ class TestAllScenarios(unittest.TestCase):
     @classmethod
     @client_context.require_connection
     def setUpClass(cls):
-        cls.listener = WhiteListEventListener("aggregate")
+        cls.listener = WhiteListEventListener("aggregate", "getMore")
         cls.client = rs_or_single_client(event_listeners=[cls.listener])
 
     @classmethod
@@ -1086,14 +1086,26 @@ def setUp(self):
         self.listener.results.clear()
 
     def setUpCluster(self, scenario_dict):
-        assets = [
-            (scenario_dict["database_name"], scenario_dict["collection_name"]),
-            (scenario_dict["database2_name"], scenario_dict["collection2_name"]),
-        ]
+        assets = [(scenario_dict["database_name"],
+                   scenario_dict["collection_name"]),
+                  (scenario_dict.get("database2_name", "db2"),
+                   scenario_dict.get("collection2_name", "coll2"))]
         for db, coll in assets:
             self.client.drop_database(db)
             self.client[db].create_collection(coll)
 
+    def setFailPoint(self, scenario_dict):
+        fail_point = scenario_dict.get("failPoint")
+        if fail_point is None:
+            return
+
+        fail_cmd = SON([('configureFailPoint', 'failCommand')])
+        fail_cmd.update(fail_point)
+        client_context.client.admin.command(fail_cmd)
+        self.addCleanup(
+            client_context.client.admin.command,
+            'configureFailPoint', fail_cmd['configureFailPoint'], mode='off')
+
     def tearDown(self):
         self.listener.results.clear()
 
@@ -1147,18 +1159,35 @@ def run_operation(client, operation):
     return cmd(**arguments)
 
 
+def assert_list_contents_are_subset(superlist, sublist):
+    assert len(superlist) == len(sublist)
+    for super, sub in zip(superlist, sublist):
+        if isinstance(sub, dict):
+            assert_dict_is_subset(super, sub)
+            continue
+        if isinstance(sub, (list, tuple)):
+            assert_list_contents_are_subset(super, sub)
+            continue
+        assert super == sub
+
+
 def assert_dict_is_subset(superdict, subdict):
     """Check that subdict is a subset of superdict."""
-    exempt_fields = ["documentKey", "_id"]
+    exempt_fields = ["documentKey", "_id", "getMore"]
     for key, value in iteritems(subdict):
         if key not in superdict:
             assert False
         if isinstance(value, dict):
             assert_dict_is_subset(superdict[key], value)
             continue
+        if isinstance(value, (list, tuple)):
+            assert_list_contents_are_subset(superdict[key], value)
+            continue
         if key in exempt_fields:
-            superdict[key] = "42"
-        assert superdict[key] == value
+            # Only check for presence of these exempt fields, but not value.
+            assert key in superdict
+        else:
+            assert superdict[key] == value
 
 
 def check_event(event, expectation_dict):
@@ -1177,6 +1206,7 @@ def create_test(scenario_def, test):
     def run_scenario(self):
         # Set up
         self.setUpCluster(scenario_def)
+        self.setFailPoint(test)
         is_error = test["result"].get("error", False)
         try:
             with get_change_stream(
@@ -1208,8 +1238,8 @@ def run_scenario(self):
         finally:
             # Check for expected events
             results = self.listener.results
-            for expectation in test.get("expectations", []):
-                for idx, (event_type, event_desc) in enumerate(iteritems(expectation)):
+            for idx, expectation in enumerate(test.get("expectations", [])):
+                for event_type, event_desc in iteritems(expectation):
                     results_key = event_type.split("_")[1]
                     event = results[results_key][idx] if len(results[results_key]) > idx else None
                     check_event(event, event_desc)