CDRIVER-3969 error when creating uncompleted cursor with no server ID (#1321)

* fix type of BCON_INT32 value

* require serverID for open cursor with `mongoc_cursor_new_from_command_reply_with_opts`

* document expectations of server ID option

* assert `found` after `mongoc_cursor_error`

If an error occurs, `found` is false. Assert on the error first to print the error.

* remove unnecessary nested `if`

Author: kevinAlbs

src/libmongoc/doc/mongoc_cursor_new_from_command_reply.rst

@@ -55,7 +55,7 @@ The server replies:
 
 ``mongoc_cursor_new_from_command_reply`` is a low-level function that initializes a :symbol:`mongoc_cursor_t` from such a reply. Additional options such as "tailable" or "awaitData" can be included in the reply.
 
-When synthesizing a completed cursor response that has no more batches (i.e. with cursor id 0), set ``server_id`` to 0 as well.
+When synthesizing a completed cursor response that has no more batches (i.e. with cursor id 0), ``server_id`` may be 0. If the cursor response is not completed (i.e. with non-zero cursor id), pass the ``server_id`` of the server used to create the cursor.
 
 Use this function only for building a language driver that wraps the C Driver. When writing applications in C, higher-level functions such as :symbol:`mongoc_collection_aggregate` are more appropriate, and ensure compatibility with a range of MongoDB versions.
 

src/libmongoc/doc/mongoc_cursor_new_from_command_reply_with_opts.rst

@@ -55,6 +55,8 @@ The server replies:
 
 ``mongoc_cursor_new_from_command_reply_with_opts`` is a low-level function that initializes a :symbol:`mongoc_cursor_t` from such a reply.
 
+When synthesizing a completed cursor response that has no more batches (i.e. with cursor id 0), ``serverId`` may be 0. If the cursor response is not completed (i.e. with non-zero cursor id), pass the ``serverId`` of the server used to create the cursor.
+
 Use this function only for building a language driver that wraps the C Driver. When writing applications in C, higher-level functions such as :symbol:`mongoc_collection_aggregate` are more appropriate, and ensure compatibility with a range of MongoDB versions.
 
 Returns

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

@@ -216,5 +216,26 @@ _mongoc_cursor_cmd_new_from_reply (mongoc_client_t *client,
                       MONGOC_ERROR_CURSOR_INVALID_CURSOR,
                       "Couldn't parse cursor document");
    }
+
+   if (0 != cursor->cursor_id && 0 == cursor->server_id) {
+      // A non-zero cursor_id means the cursor is still open on the server.
+      // Expect the "serverId" option to have been passed. The "serverId" option
+      // identifies the server with the cursor.
+      // The server with the cursor is required to send a "getMore" or
+      // "killCursors" command.
+      bson_set_error (
+         &cursor->error,
+         MONGOC_ERROR_CURSOR,
+         MONGOC_ERROR_CURSOR_INVALID_CURSOR,
+         "Expected `serverId` option to identify server with open cursor "
+         "(cursor ID is %" PRId64 "). "
+         "Consider using `mongoc_client_select_server` and using the "
+         "resulting server ID to create the cursor.",
+         cursor->cursor_id);
+      // Reset cursor_id to 0 to avoid an assertion error in
+      // `mongoc_cursor_destroy` when attempting to send "killCursors".
+      cursor->cursor_id = 0;
+   }
+
    return cursor;
 }

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

@@ -2519,7 +2519,7 @@ test_cursor_batchsize_override_range_warning (void)
 {
    mongoc_client_t *client;
    mongoc_collection_t *coll;
-   bson_t *findopts = BCON_NEW ("batchSize", BCON_INT32 (1.0));
+   bson_t *findopts = BCON_NEW ("batchSize", BCON_INT32 (1));
 
    client = test_framework_new_default_client ();
    coll = mongoc_client_get_collection (client, "db", "coll");
@@ -2547,6 +2547,186 @@ test_cursor_batchsize_override_range_warning (void)
    mongoc_client_destroy (client);
 }
 
+// Test using an open cursor created by
+// `mongoc_cursor_new_from_command_reply_with_opts`.
+// This is a regression test for CDRIVER-3969.
+static void
+test_open_cursor_from_reply (void)
+{
+   mongoc_client_t *client;
+   mongoc_collection_t *coll;
+   bson_error_t error;
+   bool ok;
+
+   client = test_framework_new_default_client ();
+   coll = get_test_collection (client, "test_open_cursor_from_reply");
+
+   // Drop collection to remove data from prior runs.
+   // Ignore errors. Dropping a non-existing collection may return an "ns not
+   // found" error.
+   mongoc_collection_drop (coll, &error);
+
+   // Insert two documents.
+   {
+      ok = mongoc_collection_insert_one (coll,
+                                         tmp_bson ("{'_id': 0}"),
+                                         NULL /* opts */,
+                                         NULL /* reply */,
+                                         &error);
+      ASSERT_OR_PRINT (ok, error);
+      ok = mongoc_collection_insert_one (coll,
+                                         tmp_bson ("{'_id': 1}"),
+                                         NULL /* opts */,
+                                         NULL /* reply */,
+                                         &error);
+      ASSERT_OR_PRINT (ok, error);
+   }
+
+   // Test creating an open cursor created without a serverId. Expect error.
+   {
+      mongoc_cursor_t *cursor;
+      bson_t reply;
+      // Use a smaller batchSize than the number of documents. The smaller
+      // batchSize will result in the cursor being left open on the server.
+      bson_t *cmd = tmp_bson ("{'find': '%s', 'batchSize': 1}",
+                              mongoc_collection_get_name (coll));
+      ok = mongoc_collection_command_simple (
+         coll, cmd, NULL /* read_prefs */, &reply, &error);
+      ASSERT_OR_PRINT (ok, error);
+
+      // Assert that the cursor has a non-zero cursorId. A non-zero cursorId
+      // means the cursor is open on the server.
+      {
+         bson_iter_t iter;
+         ASSERT (bson_iter_init (&iter, &reply));
+         ASSERT (bson_iter_find_descendant (&iter, "cursor.id", &iter));
+         ASSERT (BSON_ITER_HOLDS_INT64 (&iter));
+         ASSERT_CMPINT64 (bson_iter_int64 (&iter), >, 0);
+      }
+
+      // `reply` is destroyed by
+      // `mongoc_cursor_new_from_command_reply_with_opts`.
+      cursor = mongoc_cursor_new_from_command_reply_with_opts (
+         client, &reply, NULL /* opts */);
+
+      // Expect an error to be returned.
+      ASSERT (mongoc_cursor_error (cursor, &error));
+      ASSERT_ERROR_CONTAINS (error,
+                             MONGOC_ERROR_CURSOR,
+                             MONGOC_ERROR_CURSOR_INVALID_CURSOR,
+                             "Expected `serverId` option");
+
+      mongoc_cursor_destroy (cursor);
+   }
+
+   // Test iterating an open cursor created with a serverId. Expect no error.
+   {
+      // Get a serverID.
+      uint32_t server_id;
+      {
+         mongoc_server_description_t *sd = mongoc_client_select_server (
+            client, true /* for_writes */, NULL /* read prefs */, &error);
+         ASSERT_OR_PRINT (sd, error);
+         server_id = mongoc_server_description_id (sd);
+         mongoc_server_description_destroy (sd);
+      }
+      mongoc_cursor_t *cursor;
+      bson_t reply;
+      // Use a smaller batchSize than the number of documents. The smaller
+      // batchSize will result in the cursor being left open on the server.
+      bson_t *cmd =
+         tmp_bson ("{'find': '%s', 'batchSize': 1, 'sort': {'_id': 1}}",
+                   mongoc_collection_get_name (coll));
+      ok = mongoc_collection_command_with_opts (
+         coll,
+         cmd,
+         NULL /* read_prefs */,
+         tmp_bson ("{'serverId': %" PRIu32 "}", server_id),
+         &reply,
+         &error);
+      ASSERT_OR_PRINT (ok, error);
+
+      // Assert that the cursor has a non-zero cursorId. A non-zero cursorId
+      // means the cursor is open on the server.
+      {
+         bson_iter_t iter;
+         ASSERT (bson_iter_init (&iter, &reply));
+         ASSERT (bson_iter_find_descendant (&iter, "cursor.id", &iter));
+         ASSERT (BSON_ITER_HOLDS_INT64 (&iter));
+         ASSERT_CMPINT64 (bson_iter_int64 (&iter), >, 0);
+      }
+
+      // `reply` is destroyed by
+      // `mongoc_cursor_new_from_command_reply_with_opts`.
+      cursor = mongoc_cursor_new_from_command_reply_with_opts (
+         client, &reply, tmp_bson ("{'serverId': %" PRIu32 "}", server_id));
+
+      ASSERT_OR_PRINT (!mongoc_cursor_error (cursor, &error), error);
+      const bson_t *got;
+      bool found = mongoc_cursor_next (cursor, &got);
+      ASSERT_OR_PRINT (!mongoc_cursor_error (cursor, &error), error);
+      ASSERT (found);
+      ASSERT_MATCH (got, "{'_id': 0}");
+      found = mongoc_cursor_next (cursor, &got);
+      ASSERT_OR_PRINT (!mongoc_cursor_error (cursor, &error), error);
+      ASSERT (found);
+      ASSERT_MATCH (got, "{'_id': 1}");
+      found = mongoc_cursor_next (cursor, &got);
+      ASSERT_OR_PRINT (!mongoc_cursor_error (cursor, &error), error);
+      ASSERT (!found);
+
+      mongoc_cursor_destroy (cursor);
+   }
+
+   // Test destroying an open cursor created with a serverId. Expect no error.
+   {
+      // Get a serverID.
+      uint32_t server_id;
+      {
+         mongoc_server_description_t *sd = mongoc_client_select_server (
+            client, true /* for_writes */, NULL /* read prefs */, &error);
+         ASSERT_OR_PRINT (sd, error);
+         server_id = mongoc_server_description_id (sd);
+         mongoc_server_description_destroy (sd);
+      }
+      mongoc_cursor_t *cursor;
+      bson_t reply;
+      // Use a smaller batchSize than the number of documents. The smaller
+      // batchSize will result in the cursor being left open on the server.
+      bson_t *cmd = tmp_bson ("{'find': '%s', 'batchSize': 1}",
+                              mongoc_collection_get_name (coll));
+      ok = mongoc_collection_command_with_opts (
+         coll,
+         cmd,
+         NULL /* read_prefs */,
+         tmp_bson ("{'serverId': %" PRIu32 "}", server_id),
+         &reply,
+         &error);
+      ASSERT_OR_PRINT (ok, error);
+
+      // Assert that the cursor has a non-zero cursorId. A non-zero cursorId
+      // means the cursor is open on the server.
+      {
+         bson_iter_t iter;
+         ASSERT (bson_iter_init (&iter, &reply));
+         ASSERT (bson_iter_find_descendant (&iter, "cursor.id", &iter));
+         ASSERT (BSON_ITER_HOLDS_INT64 (&iter));
+         ASSERT_CMPINT64 (bson_iter_int64 (&iter), >, 0);
+      }
+
+      // `reply` is destroyed by
+      // `mongoc_cursor_new_from_command_reply_with_opts`.
+      cursor = mongoc_cursor_new_from_command_reply_with_opts (
+         client, &reply, tmp_bson ("{'serverId': %" PRIu32 "}", server_id));
+
+      ASSERT_OR_PRINT (!mongoc_cursor_error (cursor, &error), error);
+      mongoc_cursor_destroy (cursor);
+   }
+
+   mongoc_collection_destroy (coll);
+   mongoc_client_destroy (client);
+}
+
 void
 test_cursor_install (TestSuite *suite)
 {
@@ -2647,4 +2827,6 @@ test_cursor_install (TestSuite *suite)
    TestSuite_AddLive (suite,
                       "/Cursor/batchsize_override_range_warning",
                       test_cursor_batchsize_override_range_warning);
+   TestSuite_AddLive (
+      suite, "/Cursor/open_cursor_from_reply", test_open_cursor_from_reply);
 }