CDRIVER-2873 retryable reads

Co-authored-by: Haris Sheikh <[email protected]>

Author: isabelatkinson

src/libmongoc/CMakeLists.txt

@@ -787,6 +787,7 @@ set (test-libmongoc-sources
    ${PROJECT_SOURCE_DIR}/tests/test-mongoc-read-write-concern.c
    ${PROJECT_SOURCE_DIR}/tests/test-mongoc-read-prefs.c
    ${PROJECT_SOURCE_DIR}/tests/test-mongoc-retryable-writes.c
+   ${PROJECT_SOURCE_DIR}/tests/test-mongoc-retryable-reads.c
    ${PROJECT_SOURCE_DIR}/tests/test-mongoc-rpc.c
    ${PROJECT_SOURCE_DIR}/tests/test-mongoc-sample-commands.c
    ${PROJECT_SOURCE_DIR}/tests/test-mongoc-scram.c

src/libmongoc/doc/mongoc_uri_t.rst

@@ -90,6 +90,7 @@ Connection Options
 ========================================== ================================= ============================================================================================================================================================================================================================================
 Constant                                   Key                               Description
 ========================================== ================================= ============================================================================================================================================================================================================================================
+MONGOC_URI_RETRYREADS                      retryreads                        If "true" and the server is a MongoDB 3.6+ standalone, replica set, or sharded cluster, the driver safely retries a read that failed due to a network error or replica set failover.
 MONGOC_URI_RETRYWRITES                     retrywrites                       If "true" and the server is a MongoDB 3.6+ replica set or sharded cluster, the driver safely retries a write that failed due to a network error or replica set failover. Only inserts, updates of single documents, or deletes of single
                                                                              documents are retried.
 MONGOC_URI_APPNAME                         appname                           The client application name. This value is used by MongoDB when it logs connection information and profile information, such as slow queries.

src/libmongoc/src/mongoc/CMakeLists.txt

@@ -99,6 +99,7 @@ set (src_libmongoc_src_mongoc_DIST_noinst_hs
    mongoc-cyrus-private.h
    mongoc-database-private.h
    mongoc-errno-private.h
+   mongoc-error-private.h
    mongoc-find-and-modify-private.h
    mongoc-gridfs-bucket-file-private.h
    mongoc-gridfs-bucket-private.h

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

@@ -42,6 +42,8 @@ _mongoc_aggregate (mongoc_client_t *client,
                    const mongoc_read_concern_t *default_rc,
                    const mongoc_write_concern_t *default_wc);
 
+bool
+_has_write_key (bson_iter_t *iter);
 
 BSON_END_DECLS
 

src/libmongoc/src/mongoc/mongoc-aggregate.c

@@ -37,7 +37,7 @@
  *--------------------------------------------------------------------------
  */
 
-static bool
+bool
 _has_write_key (bson_iter_t *iter)
 {
    bson_iter_t stage;

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

@@ -61,6 +61,8 @@ BSON_BEGIN_DECLS
 #define WIRE_VERSION_OP_MSG 6
 /* first version to support array filters for "update" command */
 #define WIRE_VERSION_ARRAY_FILTERS 6
+/* first version to support retryable reads  */
+#define WIRE_VERSION_RETRY_READS 6
 /* first version to support retryable writes  */
 #define WIRE_VERSION_RETRY_WRITES 6
 /* version corresponding to server 4.0 release */

src/libmongoc/src/mongoc/mongoc-client.c

@@ -41,6 +41,7 @@
 #include "mongoc/mongoc-database-private.h"
 #include "mongoc/mongoc-gridfs-private.h"
 #include "mongoc/mongoc-error.h"
+#include "mongoc/mongoc-error-private.h"
 #include "mongoc/mongoc-log.h"
 #include "mongoc/mongoc-queue-private.h"
 #include "mongoc/mongoc-socket.h"
@@ -1603,7 +1604,8 @@ _mongoc_client_retryable_write_command_with_stream (
       &client->cluster, &parts->assembled, reply, error);
 
    if (is_retryable) {
-      _mongoc_write_error_update_if_unsupported_storage_engine (ret, error, reply);
+      _mongoc_write_error_update_if_unsupported_storage_engine (
+         ret, error, reply);
    }
 
    /* If a retryable error is encountered and the write is retryable, select
@@ -1642,9 +1644,75 @@ _mongoc_client_retryable_write_command_with_stream (
 }
 
 
+static bool
+_mongoc_client_retryable_read_command_with_stream (
+   mongoc_client_t *client,
+   mongoc_cmd_parts_t *parts,
+   mongoc_server_stream_t *server_stream,
+   bson_t *reply,
+   bson_error_t *error)
+{
+   mongoc_server_stream_t *retry_server_stream = NULL;
+   bool is_retryable = true;
+   bool ret;
+   bson_t reply_local;
+
+   if (reply == NULL) {
+      reply = &reply_local;
+   }
+
+   ENTRY;
+
+   BSON_ASSERT (parts->is_retryable_read);
+
+retry:
+   ret = mongoc_cluster_run_command_monitored (
+      &client->cluster, &parts->assembled, reply, error);
+
+   /* If a retryable error is encountered and the read is retryable, select
+    * a new readable stream and retry. If server selection fails or the selected
+    * server does not support retryable reads, fall through and allow the
+    * original error to be reported. */
+   if (is_retryable &&
+       _mongoc_read_error_get_type (ret, error, reply) ==
+          MONGOC_READ_ERR_RETRY) {
+      bson_error_t ignored_error;
+
+      /* each read command may be retried at most once */
+      is_retryable = false;
+
+      if (retry_server_stream) {
+         mongoc_server_stream_cleanup (retry_server_stream);
+      }
+
+      retry_server_stream =
+         mongoc_cluster_stream_for_reads (&client->cluster,
+                                          parts->read_prefs,
+                                          parts->assembled.session,
+                                          NULL,
+                                          &ignored_error);
+
+      if (retry_server_stream &&
+          retry_server_stream->sd->max_wire_version >=
+             WIRE_VERSION_RETRY_READS) {
+         parts->assembled.server_stream = retry_server_stream;
+         bson_destroy (reply);
+         GOTO (retry);
+      }
+   }
+
+   if (retry_server_stream) {
+      mongoc_server_stream_cleanup (retry_server_stream);
+   }
+
+   RETURN (ret);
+}
+
+
 static bool
 _mongoc_client_command_with_stream (mongoc_client_t *client,
                                     mongoc_cmd_parts_t *parts,
+                                    const mongoc_read_prefs_t *read_prefs,
                                     mongoc_server_stream_t *server_stream,
                                     bson_t *reply,
                                     bson_error_t *error)
@@ -1662,6 +1730,11 @@ _mongoc_client_command_with_stream (mongoc_client_t *client,
          client, parts, server_stream, reply, error));
    }
 
+   if (parts->is_retryable_read) {
+      RETURN (_mongoc_client_retryable_read_command_with_stream (
+         client, parts, server_stream, reply, error));
+   }
+
    RETURN (mongoc_cluster_run_command_monitored (
       &client->cluster, &parts->assembled, reply, error));
 }
@@ -1705,7 +1778,7 @@ mongoc_client_command_simple (mongoc_client_t *client,
 
    if (server_stream) {
       ret = _mongoc_client_command_with_stream (
-         client, &parts, server_stream, reply, error);
+         client, &parts, read_prefs, server_stream, reply, error);
    } else {
       /* reply initialized by mongoc_cluster_stream_for_reads */
       ret = false;
@@ -1914,7 +1987,7 @@ _mongoc_client_command_with_opts (mongoc_client_t *client,
    }
 
    ret = _mongoc_client_command_with_stream (
-      client, &parts, server_stream, reply_ptr, error);
+      client, &parts, user_prefs, server_stream, reply_ptr, error);
 
    reply_initialized = true;
 
@@ -2071,7 +2144,7 @@ mongoc_client_command_simple_with_server_id (
       parts.read_prefs = read_prefs;
 
       ret = _mongoc_client_command_with_stream (
-         client, &parts, server_stream, reply, error);
+         client, &parts, read_prefs, server_stream, reply, error);
 
       mongoc_cmd_parts_cleanup (&parts);
       mongoc_server_stream_cleanup (server_stream);

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

@@ -36,6 +36,7 @@
 
 BSON_BEGIN_DECLS
 
+#define MONGOC_DEFAULT_RETRYREADS true
 /* retryWrites requires sessions, which require crypto */
 #ifdef MONGOC_ENABLE_CRYPTO
 #define MONGOC_DEFAULT_RETRYWRITES true
@@ -78,6 +79,7 @@ typedef struct _mongoc_cmd_parts_t {
    bool is_write_command;
    bool prohibit_lsid;
    mongoc_cmd_parts_allow_txn_number_t allow_txn_number;
+   bool is_retryable_read;
    bool is_retryable_write;
    bool has_temp_session;
    mongoc_client_t *client;
@@ -130,6 +132,10 @@ mongoc_cmd_is_compressible (mongoc_cmd_t *cmd);
 void
 mongoc_cmd_parts_cleanup (mongoc_cmd_parts_t *op);
 
+bool
+_is_retryable_read (const mongoc_cmd_parts_t *parts,
+                    const mongoc_server_stream_t *server_stream);
+
 BSON_END_DECLS
 
 

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

@@ -39,6 +39,7 @@ mongoc_cmd_parts_init (mongoc_cmd_parts_t *parts,
    parts->is_write_command = false;
    parts->prohibit_lsid = false;
    parts->allow_txn_number = MONGOC_CMD_PARTS_ALLOW_TXN_NUMBER_UNKNOWN;
+   parts->is_retryable_read = false;
    parts->is_retryable_write = false;
    parts->has_temp_session = false;
    parts->client = client;
@@ -730,6 +731,39 @@ _is_retryable_write (const mongoc_cmd_parts_t *parts,
 }
 
 
+/* Check if the read command should support retryable behavior. */
+bool
+_is_retryable_read (const mongoc_cmd_parts_t *parts,
+                    const mongoc_server_stream_t *server_stream)
+{
+   if (!parts->is_read_command) {
+      return false;
+   }
+
+   /* Commands that go through read_write_command helpers are also write
+    * commands. Prohibit from read retry. */
+   if (parts->is_write_command) {
+      return false;
+   }
+
+   if (server_stream->sd->max_wire_version < WIRE_VERSION_RETRY_READS) {
+      return false;
+   }
+
+   if (_mongoc_client_session_in_txn (parts->assembled.session)) {
+      return false;
+   }
+
+   if (!mongoc_uri_get_option_as_bool (parts->client->uri,
+                                       MONGOC_URI_RETRYREADS,
+                                       MONGOC_DEFAULT_RETRYREADS)) {
+      return false;
+   }
+
+   return true;
+}
+
+
 /*
  *--------------------------------------------------------------------------
  *
@@ -888,6 +922,11 @@ mongoc_cmd_parts_assemble (mongoc_cmd_parts_t *parts,
          parts->is_retryable_write = true;
       }
 
+      /* Conversely, check if the command is retryable if it is a read. */
+      if (_is_retryable_read (parts, server_stream) && !is_get_more) {
+         parts->is_retryable_read = true;
+      }
+
       if (!bson_empty (&server_stream->cluster_time)) {
          cluster_time =
             _largest_cluster_time (&server_stream->cluster_time, cluster_time);

src/libmongoc/src/mongoc/mongoc-collection.c

@@ -652,7 +652,7 @@ mongoc_collection_command_simple (mongoc_collection_t *collection,
    return _mongoc_client_command_with_opts (collection->client,
                                             collection->db,
                                             command,
-                                            MONGOC_CMD_READ,
+                                            MONGOC_CMD_RAW,
                                             NULL /* opts */,
                                             MONGOC_QUERY_NONE,
                                             read_prefs,

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

@@ -36,7 +36,7 @@ _prime (mongoc_cursor_t *cursor)
    /* this cursor is only used with the listDatabases command. it iterates
     * over the array in the response's "databases" field. */
    if (_mongoc_cursor_run_command (
-          cursor, &data->cmd, &cursor->opts, &data->array) &&
+          cursor, &data->cmd, &cursor->opts, &data->array, false) &&
        bson_iter_init_find (&iter, &data->array, data->field_name) &&
        BSON_ITER_HOLDS_ARRAY (&iter) &&
        bson_iter_recurse (&iter, &data->iter)) {

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

@@ -32,7 +32,7 @@ _prime (mongoc_cursor_t *cursor)
    data_cmd_deprecated_t *data = (data_cmd_deprecated_t *) cursor->impl.data;
    bson_destroy (&data->reply);
    if (_mongoc_cursor_run_command (
-          cursor, &data->cmd, &cursor->opts, &data->reply)) {
+          cursor, &data->cmd, &cursor->opts, &data->reply, true)) {
       return IN_BATCH;
    } else {
       return DONE;

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

@@ -176,7 +176,8 @@ bool
 _mongoc_cursor_run_command (mongoc_cursor_t *cursor,
                             const bson_t *command,
                             const bson_t *opts,
-                            bson_t *reply);
+                            bson_t *reply,
+                            bool retry_prohibited);
 bool
 _mongoc_cursor_more (mongoc_cursor_t *cursor);