CDRIVER-3553 use whitelist/labels for resume

Author: kevinAlbs

src/libmongoc/src/mongoc/mongoc-change-stream-private.h

@@ -62,6 +62,9 @@ struct _mongoc_change_stream_t {
    bool resumed;
 
    mongoc_client_session_t *implicit_session;
+
+   /* The max_wire_version of the server the change stream is tied to. */
+   uint32_t max_wire_version;
 };
 
 mongoc_change_stream_t *

src/libmongoc/src/mongoc/mongoc-change-stream.c

@@ -32,7 +32,7 @@
 /* the caller knows either a client or server error has occurred.
  * `reply` contains the server reply or an empty document. */
 static bool
-_is_resumable_error (const bson_t *reply)
+_is_resumable_error (mongoc_change_stream_t *stream, const bson_t *reply)
 {
    bson_error_t error = {0};
 
@@ -46,28 +46,32 @@ _is_resumable_error (const bson_t *reply)
       return true;
    }
 
-   if (mongoc_error_has_label (reply, "NonResumableChangeStreamError")) {
-      return false;
-   }
-
-   /* Change Streams Spec resumable criteria: "a server error response with an
-    * error message containing the substring 'not master' or 'node is
-    * recovering' */
-   if (strstr (error.message, "not master") ||
-       strstr (error.message, "node is recovering")) {
-      return true;
+   if (stream->max_wire_version >= WIRE_VERSION_4_4) {
+      return mongoc_error_has_label (reply, "ResumableChangeStreamError");
    }
 
-   /* Change Streams Spec resumable criteria: "any server error response from a
-    * getMore command excluding those containing the following error codes" */
    switch (error.code) {
-   case 11601:                      /* Interrupted */
-   case 136:                        /* CappedPositionLost */
-   case 237:                        /* CursorKilled */
-   case MONGOC_ERROR_QUERY_FAILURE: /* error code omitted */
-      return false;
-   default:
+   case 6:     /* HostUnreachable */
+   case 7:     /* HostNotFound */
+   case 89:    /* NetworkTimeout */
+   case 91:    /* ShutdownInProgress */
+   case 189:   /* PrimarySteppedDown */
+   case 262:   /* ExceededTimeLimit */
+   case 9001:  /* SocketException */
+   case 10107: /* NotMaster */
+   case 11600: /* InterruptedAtShutdown */
+   case 11602: /* InterruptedDueToReplStateChange */
+   case 13435: /* NotMasterNoSlaveOk */
+   case 13436: /* NotMasterOrSecondary */
+   case 63:    /* StaleShardVersion */
+   case 150:   /* StaleEpoch */
+   case 13388: /* StaleConfig */
+   case 234:   /* RetryChangeStream */
+   case 133:   /* FailedToSatisfyReadPreference */
+   case 216:   /* ElectionInProgress */
       return true;
+   default:
+      return false;
    }
 }
 
@@ -93,9 +97,7 @@ _set_resume_token (mongoc_change_stream_t *stream, const bson_t *resume_token)
  *     cursor: { batchSize: x } }
  */
 static void
-_make_command (mongoc_change_stream_t *stream,
-               bson_t *command,
-               int32_t max_wire_version)
+_make_command (mongoc_change_stream_t *stream, bson_t *command)
 {
    bson_iter_t iter;
    bson_t change_stream_stage; /* { $changeStream: <change_stream_doc> } */
@@ -135,7 +137,7 @@ _make_command (mongoc_change_stream_t *stream,
                &change_stream_doc, "resumeAfter", &stream->resume_token);
          }
       } else if (!_mongoc_timestamp_empty (&stream->operation_time) &&
-                 max_wire_version >= 7) {
+                 stream->max_wire_version >= WIRE_VERSION_4_0) {
          /* Else if there is no cached resumeToken and the ChangeStream
             has a saved operation time and the max wire version is >= 7,
             the driver MUST set startAtOperationTime */
@@ -231,7 +233,6 @@ _make_cursor (mongoc_change_stream_t *stream)
    bson_iter_t iter;
    mongoc_server_description_t *sd;
    uint32_t server_id;
-   int32_t max_wire_version = -1;
 
    BSON_ASSERT (stream);
    BSON_ASSERT (!stream->cursor);
@@ -245,7 +246,7 @@ _make_cursor (mongoc_change_stream_t *stream)
    server_id = mongoc_server_description_id (sd);
    bson_append_int32 (&command_opts, "serverId", 8, server_id);
    bson_append_int32 (&getmore_opts, "serverId", 8, server_id);
-   max_wire_version = sd->max_wire_version;
+   stream->max_wire_version = sd->max_wire_version;
    mongoc_server_description_destroy (sd);
 
    if (bson_iter_init_find (&iter, &command_opts, "sessionId")) {
@@ -288,7 +289,7 @@ _make_cursor (mongoc_change_stream_t *stream)
       mongoc_read_concern_append (stream->read_concern, &command_opts);
    }
 
-   _make_command (stream, &command, max_wire_version);
+   _make_command (stream, &command);
 
    /* even though serverId has already been set, still pass the read prefs.
     * they are necessary for OP_MSG if sending to a secondary. */
@@ -350,7 +351,8 @@ _make_cursor (mongoc_change_stream_t *stream)
    if (bson_empty (&stream->opts.resumeAfter) &&
        bson_empty (&stream->opts.startAfter) &&
        _mongoc_timestamp_empty (&stream->operation_time) &&
-       max_wire_version >= 7 && bson_empty (&stream->resume_token) &&
+       stream->max_wire_version >= WIRE_VERSION_4_0 &&
+       bson_empty (&stream->resume_token) &&
        bson_iter_init_find (
           &iter,
           _mongoc_cursor_change_stream_get_reply (stream->cursor),
@@ -529,7 +531,7 @@ mongoc_change_stream_next (mongoc_change_stream_t *stream, const bson_t **bson)
          goto end;
       }
 
-      resumable = _is_resumable_error (err_doc);
+      resumable = _is_resumable_error (stream, err_doc);
       while (resumable) {
          /* recreate the cursor. */
          mongoc_cursor_destroy (stream->cursor);
@@ -545,7 +547,7 @@ mongoc_change_stream_next (mongoc_change_stream_t *stream, const bson_t **bson)
             goto end;
          }
          if (err_doc) {
-            resumable = _is_resumable_error (err_doc);
+            resumable = _is_resumable_error (stream, err_doc);
          } else {
             resumable = false;
          }

src/libmongoc/tests/json-test.c

@@ -869,6 +869,28 @@ check_topology_type (const bson_t *test)
    return can_proceed;
 }
 
+static void
+_recreate (mongoc_client_t *client,
+           const char *db_name,
+           const char *collection_name)
+{
+   mongoc_collection_t *collection;
+   mongoc_database_t *db;
+
+   if (!db_name || !collection_name) {
+      return;
+   }
+   collection = mongoc_client_get_collection (client, db_name, collection_name);
+   mongoc_collection_drop (collection, NULL);
+   mongoc_collection_destroy (collection);
+
+   db = mongoc_client_get_database (client, db_name);
+   collection = mongoc_database_create_collection (
+      db, collection_name, NULL /* options */, NULL);
+   mongoc_collection_destroy (collection);
+   mongoc_database_destroy (db);
+}
+
 
 static void
 _insert_data (mongoc_collection_t *collection, bson_t *documents)
@@ -1174,6 +1196,15 @@ deactivate_fail_points (mongoc_client_t *client, uint32_t server_id)
       }
    }
 
+   if (sd->max_wire_version >= WIRE_VERSION_4_4) {
+      /* failGetMoreAfterCursorCheckout added in 4.4 */
+      command = tmp_bson ("{'configureFailPoint': "
+                          "'failGetMoreAfterCursorCheckout', 'mode': 'off'}");
+      r = mongoc_client_command_simple_with_server_id (
+         client, "admin", command, NULL, server_id, NULL, &error);
+      ASSERT_OR_PRINT (r, error);
+   }
+
    mongoc_server_description_destroy (sd);
 }
 
@@ -1435,8 +1466,8 @@ run_json_general_test (const json_test_config_t *config)
    const bson_t *scenario = config->scenario;
    bson_iter_t scenario_iter;
    bson_iter_t tests_iter;
-   const char *db_name;
-   const char *collection_name;
+   const char *db_name, *db2_name;
+   const char *collection_name, *collection2_name;
 
    ASSERT (scenario);
 
@@ -1451,6 +1482,15 @@ run_json_general_test (const json_test_config_t *config)
                         ? bson_lookup_utf8 (scenario, "collection_name")
                         : "test";
 
+   /* database2_name/collection2_name are optional. */
+   db2_name = bson_has_field (scenario, "database2_name")
+                 ? bson_lookup_utf8 (scenario, "database2_name")
+                 : NULL;
+   collection2_name = bson_has_field (scenario, "collection2_name")
+                         ? bson_lookup_utf8 (scenario, "collection2_name")
+                         : NULL;
+
+
    ASSERT (bson_iter_init_find (&scenario_iter, scenario, "tests"));
    ASSERT (BSON_ITER_HOLDS_ARRAY (&scenario_iter));
    ASSERT (bson_iter_recurse (&scenario_iter, &tests_iter));
@@ -1557,6 +1597,9 @@ run_json_general_test (const json_test_config_t *config)
       }
 
       set_auto_encryption_opts (client, &test);
+      /* Drop and recreate test database/collection if necessary. */
+      _recreate (client, db_name, collection_name);
+      _recreate (client, db2_name, collection2_name);
       insert_data (db_name, collection_name, scenario);
 
       db = mongoc_client_get_database (client, db_name);

src/libmongoc/tests/json/change_streams/change-streams-errors.json

@@ -71,6 +71,90 @@
           "code": 40324
         }
       }
+    },
+    {
+      "description": "Change Stream should error when _id is projected out",
+      "minServerVersion": "4.1.11",
+      "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": 280,
+          "errorLabels": [
+            "NonResumableChangeStreamError"
+          ]
+        }
+      }
+    },
+    {
+      "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
+        }
+      }
     }
   ]
 }