CDRIVER-4738 do not resume after change stream cursor is closed

Author: Julia-Garland

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

@@ -497,6 +497,15 @@ mongoc_change_stream_next(mongoc_change_stream_t *stream, const bson_t **bson)
          goto end;
       }
 
+      /* the cursor is closed. */
+      if (stream->cursor->cursor_id == 0) {
+         _mongoc_set_error(&stream->err,
+                           MONGOC_ERROR_CURSOR,
+                           MONGOC_ERROR_CURSOR_INVALID_CURSOR,
+                           "Cannot advance a closed change stream.");
+         goto end;
+      }
+
       resumable = _is_resumable_error(stream, err_doc);
       while (resumable) {
          /* recreate the cursor. */

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

@@ -1930,6 +1930,78 @@ prose_test_18(void)
    mock_server_destroy(server);
 }
 
+// Test that a resume does not occur after an "invalidate" event.
+static void
+iterate_after_invalidate(void *test_ctx)
+{
+   mongoc_client_t *client = test_framework_new_default_client();
+   mongoc_collection_t *coll = mongoc_client_get_collection(client, "db", "coll");
+   bson_error_t error;
+   int64_t start_time = bson_get_monotonic_time();
+
+   BSON_UNUSED(test_ctx);
+
+   // Insert a document into the collection to ensure the collection is created.
+   bool ok = mongoc_collection_insert_one(coll, tmp_bson("{'foo': 'bar'}"), NULL /* opts */, NULL /* reply */, &error);
+   ASSERT_OR_PRINT(ok, error);
+
+   mongoc_change_stream_t *cs = mongoc_collection_watch(coll, tmp_bson("{}"), NULL /* opts */);
+
+   ASSERT_OR_PRINT(mongoc_collection_drop(coll, &error), error);
+
+   // Iterate until the next event. Expect "drop" event.
+   {
+      const bson_t *event;
+      bool found_event = false;
+      while (!found_event) {
+         found_event = mongoc_change_stream_next(cs, &event);
+         if (!found_event) {
+            ASSERT_OR_PRINT(!mongoc_change_stream_error_document(cs, &error, NULL /* document */), error);
+         }
+
+         int64_t delta = bson_get_monotonic_time() - start_time;
+         if (delta > 10 * 1000 * 1000) {
+            test_error("test exceeded 10 seconds");
+         }
+      }
+      ASSERT_MATCH(event, "{'operationType': 'drop'}");
+   }
+
+   // Iterate until the next event. Expect "invalidate" event.
+   {
+      const bson_t *event;
+      bool found_event = false;
+      while (!found_event) {
+         found_event = mongoc_change_stream_next(cs, &event);
+         if (!found_event) {
+            ASSERT_OR_PRINT(!mongoc_change_stream_error_document(cs, &error, NULL /* document */), error);
+         }
+
+         int64_t delta = bson_get_monotonic_time() - start_time;
+         if (delta > 10 * 1000 * 1000) {
+            test_error("test exceeded 10 seconds");
+         }
+      }
+      ASSERT_MATCH(event, "{'operationType': 'invalidate'}");
+   }
+
+   // Iterate. Expect error suggesting failure to iterate a closed cursor.
+   {
+      const bson_t *event;
+      bool found_error = false;
+      while (!found_error) {
+         ASSERT_WITH_MSG(!mongoc_change_stream_next(cs, &event), "expected no event, got: %s", tmp_json(event));
+         found_error = mongoc_change_stream_error_document(cs, &error, NULL /* document */);
+      }
+      ASSERT_ERROR_CONTAINS(
+         error, MONGOC_ERROR_CURSOR, MONGOC_ERROR_CURSOR_INVALID_CURSOR, "Cannot advance a closed change stream");
+   }
+
+   mongoc_change_stream_destroy(cs);
+   mongoc_collection_destroy(coll);
+   mongoc_client_destroy(client);
+}
+
 typedef struct {
    bson_t *commands[6];
    size_t commands_len;
@@ -2176,6 +2248,12 @@ test_change_stream_install(TestSuite *suite)
                      test_framework_skip_if_not_replset);
    TestSuite_AddMockServerTest(suite, "/change_streams/prose_test_17", prose_test_17);
    TestSuite_AddMockServerTest(suite, "/change_streams/prose_test_18", prose_test_18);
+   TestSuite_AddFull(suite,
+                     "/change_streams/iterate_after_invalidate",
+                     iterate_after_invalidate,
+                     NULL,
+                     NULL,
+                     test_framework_skip_if_not_replset);
    TestSuite_AddFull(suite,
                      "/change_stream/batchSize0",
                      test_change_stream_batchSize0,