CDRIVER-4130 Unconditionally use OP_MSG for server 5.1 and newer (#912)

* Warn and ignore exhaust cursor option given server 5.1 or newer
* Unconditionally use OP_MSG for find command given server 5.1 or newer
* Unconditionally use OP_MSG for getMore command given server 5.1 or newer
* Avoid unnecessary exhaust cursor in /counters/op_msg test
* Document exhaust option behavior for find command
* Add /Client/exhaust_cursor/fallback test for server 5.1 and newer

Author: eramongodb

src/libmongoc/doc/mongoc_collection_find_with_opts.rst

@@ -132,7 +132,7 @@ Option                   BSON type           Option               BSON type
 ``allowDiskUse``         bool
 =======================  ==================  ===================  ==================
 
-All options are documented in the reference page for `the "find" command`_ in the MongoDB server manual, except for "maxAwaitTimeMS" and "sessionId".
+All options are documented in the reference page for `the "find" command`_ in the MongoDB server manual, except for "maxAwaitTimeMS", "sessionId", and "exhaust".
 
 "maxAwaitTimeMS" is the maximum amount of time for the server to wait on new documents to satisfy a query, if "tailable" and "awaitData" are both true.
 If no new documents are found, the tailable cursor receives an empty batch. The "maxAwaitTimeMS" option is ignored for MongoDB older than 3.4.
@@ -141,6 +141,8 @@ To add a "sessionId", construct a :symbol:`mongoc_client_session_t` with :symbol
 
 To add a "readConcern", construct a :symbol:`mongoc_read_concern_t` with :symbol:`mongoc_read_concern_new` and configure it with :symbol:`mongoc_read_concern_set_level`. Then use :symbol:`mongoc_read_concern_append` to add the read concern to ``opts``.
 
+"exhaust" requests the construction of an exhaust cursor. For MongoDB servers before 5.1, this option converts the command into a legacy OP_QUERY message. For MongoDB servers 5.1 and newer, this option is ignored and a normal cursor is constructed instead.
+
 For some options like "collation", the driver returns an error if the server version is too old to support the feature.
 Any fields in ``opts`` that are not listed here are passed to the server unmodified.
 
@@ -153,7 +155,7 @@ The ``snapshot`` boolean option is removed in MongoDB 4.0. The ``maxScan`` optio
 
 .. seealso::
 
-  | `The "find" command`_ in the MongoDB Manual. All options listed there are supported by the C Driver.  For MongoDB servers before 3.2, or for exhaust queries, the driver transparently converts the query to a legacy OP_QUERY message.
+  | `The "find" command`_ in the MongoDB Manual. All options listed there are supported by the C Driver.  For MongoDB servers before 3.2, the driver transparently converts the query to a legacy OP_QUERY message.
 
 .. _the "find" command: https://docs.mongodb.org/master/reference/command/find/
 

src/libmongoc/src/mongoc/mongoc-cmd.c

@@ -181,7 +181,8 @@ mongoc_cmd_parts_append_opts (mongoc_cmd_parts_t *parts,
          parts->assembled.session = cs;
          continue;
       } else if (BSON_ITER_IS_KEY (iter, "serverId") ||
-                 BSON_ITER_IS_KEY (iter, "maxAwaitTimeMS")) {
+                 BSON_ITER_IS_KEY (iter, "maxAwaitTimeMS") ||
+                 BSON_ITER_IS_KEY (iter, "exhaust")) {
          continue;
       }
 
@@ -629,7 +630,8 @@ _mongoc_cmd_parts_assemble_mongod (mongoc_cmd_parts_t *parts,
       case MONGOC_TOPOLOGY_LOAD_BALANCED:
       case MONGOC_TOPOLOGY_DESCRIPTION_TYPES:
       default:
-         /* must not call this function w/ sharded, load balanced, or unknown topology type */
+         /* must not call this function w/ sharded, load balanced, or unknown
+          * topology type */
          BSON_ASSERT (false);
       }
    } /* if (!parts->is_write_command) */
@@ -1024,7 +1026,8 @@ mongoc_cmd_parts_assemble (mongoc_cmd_parts_t *parts,
       ret = true;
    } else if (server_type == MONGOC_SERVER_MONGOS ||
               server_stream->topology_type == MONGOC_TOPOLOGY_LOAD_BALANCED) {
-      /* TODO (CDRIVER-4117) remove the check of the topology description type. */
+      /* TODO (CDRIVER-4117) remove the check of the topology description type.
+       */
       _mongoc_cmd_parts_assemble_mongos (parts, server_stream);
       ret = true;
    } else {

src/libmongoc/src/mongoc/mongoc-cursor-cmd.c

@@ -37,7 +37,7 @@ static getmore_type_t
 _getmore_type (mongoc_cursor_t *cursor)
 {
    mongoc_server_stream_t *server_stream;
-   bool use_cmd;
+   int32_t wire_version;
    data_cmd_t *data = (data_cmd_t *) cursor->impl.data;
    if (data->getmore_type != UNKNOWN) {
       return data->getmore_type;
@@ -46,10 +46,21 @@ _getmore_type (mongoc_cursor_t *cursor)
    if (!server_stream) {
       return UNKNOWN;
    }
-   use_cmd = server_stream->sd->max_wire_version >= WIRE_VERSION_FIND_CMD &&
-             !_mongoc_cursor_get_opt_bool (cursor, MONGOC_CURSOR_EXHAUST);
-   data->getmore_type = use_cmd ? GETMORE_CMD : OP_GETMORE;
+   wire_version = server_stream->sd->max_wire_version;
    mongoc_server_stream_cleanup (server_stream);
+
+   if (
+      /* Server version 5.1 and newer do not support OP_GETMORE. */
+      wire_version > WIRE_VERSION_5_0 ||
+      /* Fallback to legacy OP_GETMORE wire protocol messages if exhaust cursor
+         requested with server version 3.6 or newer . */
+      (wire_version >= WIRE_VERSION_FIND_CMD &&
+       !_mongoc_cursor_get_opt_bool (cursor, MONGOC_CURSOR_EXHAUST))) {
+      data->getmore_type = GETMORE_CMD;
+   } else {
+      data->getmore_type = OP_GETMORE;
+   }
+
    return data->getmore_type;
 }
 

src/libmongoc/src/mongoc/mongoc-cursor-find.c

@@ -32,7 +32,7 @@ _mongoc_cursor_impl_find_opquery_init (mongoc_cursor_t *cursor, bson_t *filter);
 static mongoc_cursor_state_t
 _prime (mongoc_cursor_t *cursor)
 {
-   bool use_find_command;
+   int32_t wire_version;
    mongoc_server_stream_t *server_stream;
    data_find_t *data = (data_find_t *) cursor->impl.data;
 
@@ -41,15 +41,17 @@ _prime (mongoc_cursor_t *cursor)
    if (!server_stream) {
       return DONE;
    }
-   /* find_getmore_killcursors spec:
-    * "The find command does not support the exhaust flag from OP_QUERY." */
-   use_find_command =
-      server_stream->sd->max_wire_version >= WIRE_VERSION_FIND_CMD &&
-      !_mongoc_cursor_get_opt_bool (cursor, MONGOC_CURSOR_EXHAUST);
+   wire_version = server_stream->sd->max_wire_version;
    mongoc_server_stream_cleanup (server_stream);
 
    /* set all mongoc_impl_t function pointers. */
-   if (use_find_command) {
+   if (
+      /* Server version 5.1 and newer do not support OP_QUERY. */
+      wire_version > WIRE_VERSION_5_0 ||
+      /* Fallback to legacy OP_QUERY wire protocol messages if exhaust cursor
+         requested with server version 3.6 or newer. */
+      (wire_version >= WIRE_VERSION_FIND_CMD &&
+       !_mongoc_cursor_get_opt_bool (cursor, MONGOC_CURSOR_EXHAUST))) {
       _mongoc_cursor_impl_find_cmd_init (cursor, &data->filter /* stolen */);
    } else {
       _mongoc_cursor_impl_find_opquery_init (cursor,

src/libmongoc/src/mongoc/mongoc-cursor.c

@@ -963,6 +963,11 @@ _mongoc_cursor_run_command (mongoc_cursor_t *cursor,
          _mongoc_bson_init_if_set (reply);
          GOTO (done);
       }
+      if (_mongoc_cursor_get_opt_bool (cursor, MONGOC_CURSOR_EXHAUST)) {
+         MONGOC_WARNING (
+            "exhaust cursors not supported with OP_MSG, using normal "
+            "cursor instead");
+      }
    }
 
    if (parts.assembled.session) {
@@ -1005,6 +1010,14 @@ _mongoc_cursor_run_command (mongoc_cursor_t *cursor,
       GOTO (done);
    }
 
+   /* Exhaust cursors with OP_MSG not yet supported; fallback to normal cursor.
+    * user_query_flags is unused in OP_MSG, so this technically has no effect,
+    * but is done anyways to ensure the query flags match handling of options.
+    */
+   if (parts.user_query_flags & MONGOC_QUERY_EXHAUST) {
+      parts.user_query_flags ^= MONGOC_QUERY_EXHAUST;
+   }
+
    /* we might use mongoc_cursor_set_hint to target a secondary but have no
     * read preference, so the secondary rejects the read. same if we have a
     * direct connection to a secondary (topology type "single"). with

src/libmongoc/tests/test-mongoc-counters.c

@@ -129,7 +129,7 @@ static void
 test_counters_op_msg (void *ctx)
 {
    mongoc_collection_t *coll;
-   mongoc_cursor_t *exhaust_cursor;
+   mongoc_cursor_t *cursor;
    const bson_t *bson;
    mongoc_client_t *client;
 
@@ -144,22 +144,15 @@ test_counters_op_msg (void *ctx)
    DIFF_AND_RESET (op_egress_total, ==, 4);
    DIFF_AND_RESET (op_ingress_msg, ==, 4);
    DIFF_AND_RESET (op_ingress_total, ==, 4);
-   /* an exhaust cursor still must use an OP_QUERY find. */
-   exhaust_cursor = mongoc_collection_find (coll,
-                                            MONGOC_QUERY_EXHAUST,
-                                            0 /* skip */,
-                                            0 /* limit */,
-                                            1 /* batch size */,
-                                            tmp_bson ("{}"),
-                                            NULL /* fields */,
-                                            NULL /* read prefs */);
-   while (mongoc_cursor_next (exhaust_cursor, &bson))
+   cursor =
+      mongoc_collection_find_with_opts (coll, tmp_bson ("{}"), NULL, NULL);
+   while (mongoc_cursor_next (cursor, &bson))
       ;
-   mongoc_cursor_destroy (exhaust_cursor);
-   DIFF_AND_RESET (op_egress_msg, ==, 0);
-   DIFF_AND_RESET (op_ingress_msg, ==, 0);
-   DIFF_AND_RESET (op_egress_query, ==, 1);
-   DIFF_AND_RESET (op_ingress_reply, >, 0);
+   mongoc_cursor_destroy (cursor);
+   DIFF_AND_RESET (op_egress_msg, >, 0);
+   DIFF_AND_RESET (op_ingress_msg, >, 0);
+   DIFF_AND_RESET (op_egress_query, ==, 0);
+   DIFF_AND_RESET (op_ingress_reply, ==, 0);
    DIFF_AND_RESET (op_egress_total, >, 0);
    DIFF_AND_RESET (op_ingress_total, >, 0);
    mongoc_collection_destroy (coll);

src/libmongoc/tests/test-mongoc-exhaust.c

@@ -300,6 +300,55 @@ test_exhaust_cursor_pool (void *context)
    test_exhaust_cursor (true);
 }
 
+static void
+test_exhaust_cursor_fallback (void *unused)
+{
+   mongoc_client_t *client;
+   mongoc_collection_t *collection;
+   bson_error_t error;
+   mongoc_cursor_t *cursor;
+   const bson_t *doc;
+
+   client = test_framework_new_default_client ();
+   ASSERT (client);
+
+   collection = get_test_collection (client, "test_exhaust_cursor_fallback");
+   ASSERT (collection);
+
+   (void) mongoc_collection_drop (collection, &error);
+   ASSERT_OR_PRINT (mongoc_collection_insert_one (
+                       collection, tmp_bson ("{'a': 1}"), NULL, NULL, &error),
+                    error);
+
+
+   cursor = mongoc_collection_find_with_opts (
+      collection, tmp_bson ("{}"), tmp_bson ("{'exhaust': true}"), NULL);
+   ASSERT (cursor);
+
+   /* Cursor should be a normal cursor despite exhaust option. */
+   ASSERT (!cursor->in_exhaust);
+   ASSERT (!cursor->client->in_exhaust);
+
+   /* Warning message is generated on call to mongoc_cursor_next() during which
+    * server wire version is discovered, not on call to
+    * mongoc_collection_find_with_opts(). */
+   capture_logs (true);
+   ASSERT_OR_PRINT (mongoc_cursor_next (cursor, &doc),
+                    (mongoc_cursor_error (cursor, &error), error));
+   ASSERT_CAPTURED_LOG (
+      "cursor",
+      MONGOC_LOG_LEVEL_WARNING,
+      "exhaust cursors not supported with OP_MSG, using normal cursor instead");
+   capture_logs (false);
+
+   ASSERT_MATCH (doc, "{'a': 1}");
+   ASSERT (!mongoc_cursor_next (cursor, &doc));
+
+   mongoc_cursor_destroy (cursor);
+   mongoc_collection_destroy (collection);
+   mongoc_client_destroy (client);
+}
+
 static void
 test_exhaust_cursor_multi_batch (void *context)
 {
@@ -682,6 +731,13 @@ test_exhaust_install (TestSuite *suite)
                       NULL,
                       skip_if_mongos,
                       test_framework_skip_if_no_legacy_opcodes);
+   TestSuite_AddFull (suite,
+                      "/Client/exhaust_cursor/fallback",
+                      test_exhaust_cursor_fallback,
+                      NULL,
+                      NULL,
+                      skip_if_mongos,
+                      test_framework_skip_if_max_wire_version_less_than_14);
    TestSuite_AddLive (suite,
                       "/Client/set_max_await_time_ms",
                       test_cursor_set_max_await_time_ms);