CDRIVER-6182 break resume loop after two timeouts

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

@@ -507,6 +507,7 @@ mongoc_change_stream_next(mongoc_change_stream_t *stream, const bson_t **bson)
       }
 
       resumable = _is_resumable_error(stream, err_doc);
+      int iteration_timeout_count = 0;
       while (resumable) {
          /* recreate the cursor. */
          mongoc_cursor_destroy(stream->cursor);
@@ -521,10 +522,17 @@ mongoc_change_stream_next(mongoc_change_stream_t *stream, const bson_t **bson)
          if (!mongoc_cursor_error_document(stream->cursor, &err, &err_doc)) {
             goto end;
          }
-         if (err_doc) {
-            resumable = _is_resumable_error(stream, err_doc);
-         } else {
+         BSON_ASSERT(err_doc);
+         if (stream->cursor->had_stream_timeout) {
+            iteration_timeout_count++;
+         }
+         if (iteration_timeout_count >= 2) {
+            // CDRIVER-6182: Do not resume if two iteration timeouts occur. Intended to avoid a possible resume loop
+            // when `aggregate` succeeds but `getMore` consistently times out.
+            MONGOC_WARNING("Breaking change stream resume loop after two timeouts");
             resumable = false;
+         } else {
+            resumable = _is_resumable_error(stream, err_doc);
          }
       }
 

src/libmongoc/src/mongoc/mongoc-cluster.c

@@ -130,6 +130,7 @@ _handle_network_error(mongoc_cluster_t *cluster, mongoc_server_stream_t *server_
    type = MONGOC_SDAM_APP_ERROR_NETWORK;
    if (mongoc_stream_timed_out(server_stream->stream)) {
       type = MONGOC_SDAM_APP_ERROR_TIMEOUT;
+      server_stream->timed_out = true;
    }
 
    _mongoc_topology_handle_app_error(topology,

src/libmongoc/src/mongoc/mongoc-cursor-private.h

@@ -149,7 +149,8 @@ struct _mongoc_cursor_t {
    uint32_t dblen;
 
    bson_error_t error;
-   bson_t error_doc; /* always initialized, and set with server errors. */
+   bson_t error_doc;        // Always initialized, and set with server errors.
+   bool had_stream_timeout; // True if previous command run resulted in a timeout on the mongoc_stream_t.
 
    const bson_t *current;
 

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

@@ -741,6 +741,8 @@ _mongoc_cursor_run_command(
       }
    }
 
+   cursor->had_stream_timeout = false; // Reset before running next command.
+
    if (parts.assembled.session) {
       /* initial query/aggregate/etc, and opts contains "sessionId" */
       BSON_ASSERT(!cursor->client_session);
@@ -835,6 +837,8 @@ _mongoc_cursor_run_command(
       memset(&cursor->error, 0, sizeof(bson_error_t));
    }
 
+   cursor->had_stream_timeout = server_stream->timed_out;
+
    if (is_retryable && _mongoc_read_error_get_type(ret, &cursor->error, reply) == MONGOC_READ_ERR_RETRY) {
       is_retryable = false;
 

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

@@ -41,6 +41,7 @@ typedef struct _mongoc_server_stream_t {
    // by a network error establishing an initial connection. Used to avoid
    // further retry attempts.
    bool retry_attempted;
+   bool timed_out; // True if an operation on `stream` timed out.
 } mongoc_server_stream_t;
 
 

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

@@ -36,6 +36,7 @@ mongoc_server_stream_new(const mongoc_topology_description_t *td,
    server_stream->stream = stream; /* merely borrowed */
    server_stream->must_use_primary = false;
    server_stream->retry_attempted = false;
+   server_stream->timed_out = false;
 
    return server_stream;
 }

src/libmongoc/tests/test-conveniences.c

@@ -721,7 +721,8 @@ match_json(const bson_t *doc,
    }
 
    ctx.is_command = is_command;
-   matches = match_bson_with_ctx(doc, pattern, &ctx);
+   bson_t empty = BSON_INITIALIZER;
+   matches = match_bson_with_ctx(doc ? doc : &empty, pattern, &ctx);
 
    if (!matches) {
       char *as_string = doc ? bson_as_canonical_extended_json(doc, NULL) : NULL;