CDRIVER-2883 Support sharded transactions recovery token

Author: bazile-clyde

.gdbinit

@@ -16,11 +16,11 @@ define __printbson
 	set $bson = ((uint32_t*)$arg0)
 	printf " (len=%d)\n", $bson[0]
 	printf "{\n"
-	__printelements ($bson+1) 0 1
+	__printElements ($bson+1) 0 1
 	printf "\n}\n"
 end
 
-define __printelements
+define __printElements
 	set $data = ((uint8_t*)$arg0)
 	set $isDocument = $arg1
 	set $depth = $arg2

src/libmongoc/doc/mongoc_read_prefs_t.rst

@@ -8,7 +8,7 @@ A read preference abstraction
 Synopsis
 --------
 
-:symbol:`mongoc_read_prefs_t` provides an abstraction on top of the MongoDB connection read prefences. It allows for hinting to the driver which nodes in a replica set should be accessed first.
+:symbol:`mongoc_read_prefs_t` provides an abstraction on top of the MongoDB connection read preferences. It allows for hinting to the driver which nodes in a replica set should be accessed first.
 
 You can specify a read preference mode on connection objects, database objects, collection objects, or per-operation.  Generally, it makes the most sense to stick with the global default, ``MONGOC_READ_PRIMARY``.  All of the other modes come with caveats that won't be covered in great detail here.
 

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

@@ -77,6 +77,7 @@ struct _mongoc_client_session_t {
    uint32_t operation_increment;
    uint32_t client_generation;
    uint32_t server_id;
+   bson_t *recovery_token;
 
    /* For testing only */
    int64_t with_txn_timeout_ms;

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

@@ -209,6 +209,9 @@ txn_commit (mongoc_client_session_t *session,
    _mongoc_bson_init_if_set (reply);
 
    BSON_APPEND_INT32 (&cmd, "commitTransaction", 1);
+   if (session->recovery_token) {
+      BSON_APPEND_DOCUMENT (&cmd, "recoveryToken", session->recovery_token);
+   }
 
 retry:
    if (!mongoc_client_session_append (session, &opts, err_ptr)) {
@@ -1423,6 +1426,7 @@ mongoc_client_session_destroy (mongoc_client_session_t *session)
    txn_opts_cleanup (&session->txn.opts);
 
    bson_destroy (&session->cluster_time);
+   bson_destroy (session->recovery_token);
    bson_free (session);
 
    EXIT;

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

@@ -506,6 +506,14 @@ handle_not_master_error (mongoc_cluster_t *cluster,
    }
 }
 
+bool
+_in_sharded_txn (const mongoc_client_session_t *session)
+{
+   return session && _mongoc_client_session_in_txn (session) &&
+          _mongoc_topology_get_type (session->client->topology) ==
+             MONGOC_TOPOLOGY_SHARDED;
+}
+
 /*
  *--------------------------------------------------------------------------
  *
@@ -542,6 +550,7 @@ mongoc_cluster_run_command_monitored (mongoc_cluster_t *cluster,
    bson_t reply_local;
    bson_error_t error_local;
    int32_t compressor_id;
+   bson_iter_t iter;
 
    server_stream = cmd->server_stream;
    server_id = server_stream->sd->id;
@@ -612,6 +621,19 @@ mongoc_cluster_run_command_monitored (mongoc_cluster_t *cluster,
 
    handle_not_master_error (cluster, server_id, reply);
 
+   if (retval && _in_sharded_txn (cmd->session) &&
+       bson_iter_init_find (&iter, reply, "recoveryToken")) {
+      bson_free (cmd->session->recovery_token);
+      if (BSON_ITER_HOLDS_DOCUMENT (&iter)) {
+         cmd->session->recovery_token =
+            bson_new_from_data (bson_iter_value (&iter)->value.v_doc.data,
+                                bson_iter_value (&iter)->value.v_doc.data_len);
+      } else {
+         MONGOC_ERROR ("Malformed recovery token from server");
+         cmd->session->recovery_token = NULL;
+      }
+   }
+
    if (reply == &reply_local) {
       bson_destroy (&reply_local);
    }
@@ -2153,11 +2175,8 @@ _mongoc_cluster_select_server_id (mongoc_client_session_t *cs,
                                   bson_error_t *error)
 {
    uint32_t server_id;
-   bool in_txn;
 
-   in_txn = _mongoc_client_session_in_txn (cs);
-   if (in_txn && _mongoc_topology_get_type (cs->client->topology) ==
-                    MONGOC_TOPOLOGY_SHARDED) {
+   if (_in_sharded_txn (cs)) {
       server_id = cs->server_id;
       if (!server_id) {
          server_id = mongoc_topology_select_server_id (
@@ -2170,7 +2189,7 @@ _mongoc_cluster_select_server_id (mongoc_client_session_t *cs,
       /* Transactions Spec: Additionally, any non-transaction operation using a
        * pinned ClientSession MUST unpin the session and the operation MUST
        * perform normal server selection. */
-      if (cs && !in_txn) {
+      if (cs && !_mongoc_client_session_in_txn (cs)) {
          _mongoc_client_session_unpin (cs);
       }
    }

src/libmongoc/tests/json-test-operations.c

@@ -945,6 +945,8 @@ find_and_modify (mongoc_collection_t *collection,
     */
    if (r) {
       bson_lookup_value (reply, "value", &value);
+   } else {
+      value_init_from_doc (&value, reply);
    }
 
    check_result (test, operation, r, &value, &error);
@@ -1658,9 +1660,20 @@ json_test_operation (json_test_ctx_t *ctx,
          test_error ("unrecognized session operation name %s", op_name);
       }
    } else if (!strcmp (obj_name, "testRunner")) {
+      /* We don't use reply, but we need to initialize it for the test runner */
+      bson_init (reply);
       if (!strcmp (op_name, "assertSessionPinned")) {
-         bson_init (reply);
          res = (0 != mongoc_client_session_get_server_id (session));
+      } else if (!strcmp (op_name, "assertSessionUnpinned")) {
+         res = (0 == mongoc_client_session_get_server_id (session));
+      } else if (!strcmp (op_name, "targetedFailPoint")) {
+         mongoc_client_t *client;
+
+         client = mongoc_client_new_from_uri (ctx->client->uri);
+         activate_fail_point (
+            client, session->server_id, operation, "arguments.failPoint");
+
+         mongoc_client_destroy (client);
       } else {
          test_error ("unrecognized session operation name %s", op_name);
       }

src/libmongoc/tests/json-test.c

@@ -991,14 +991,7 @@ execute_test (const json_test_config_t *config,
    ASSERT_OR_PRINT (server_id, error);
 
    if (bson_has_field (test, "failPoint")) {
-      bson_t command;
-      bool r;
-
-      bson_lookup_doc (test, "failPoint", &command);
-      ASSERT_CMPSTR (_mongoc_get_command_name (&command), "configureFailPoint");
-      r = mongoc_client_command_simple_with_server_id (
-         client, "admin", &command, NULL, server_id, NULL, &error);
-      ASSERT_OR_PRINT (r, error);
+      activate_fail_point (client, server_id, test, "failPoint");
    }
 
    json_test_ctx_init (&ctx, test, client, db, collection, config);
@@ -1043,6 +1036,27 @@ execute_test (const json_test_config_t *config,
 }
 
 
+void
+activate_fail_point (mongoc_client_t *client,
+                     const uint32_t server_id,
+                     const bson_t *test,
+                     const char *key)
+{
+   bson_t command;
+   bson_error_t error;
+   bool r;
+
+   BSON_ASSERT (client);
+   BSON_ASSERT (server_id);
+
+   bson_lookup_doc (test, key, &command);
+
+   ASSERT_CMPSTR (_mongoc_get_command_name (&command), "configureFailPoint");
+   r = mongoc_client_command_simple_with_server_id (
+      client, "admin", &command, NULL, server_id, NULL, &error);
+   ASSERT_OR_PRINT (r, error);
+}
+
 /*
  *-----------------------------------------------------------------------
  *
@@ -1186,6 +1200,7 @@ run_json_general_test (const json_test_config_t *config)
       uint32_t server_id;
       bson_error_t error;
       bool r;
+      bson_iter_t uri_iter;
 
       ASSERT (BSON_ITER_HOLDS_DOCUMENT (&tests_iter));
       bson_iter_bson (&tests_iter, &test);
@@ -1209,7 +1224,10 @@ run_json_general_test (const json_test_config_t *config)
 
       bson_free (selected_test);
 
-      uri = test_framework_get_uri ();
+      uri = bson_iter_init_find (&uri_iter, &test, "useMultipleMongoses")
+               ? mongoc_uri_new ("mongodb://localhost:27017,localhost:27018/")
+               : test_framework_get_uri ();
+
       if (bson_iter_init_find (&client_opts_iter, &test, "clientOptions")) {
          bson_t client_opts;
 

src/libmongoc/tests/json-test.h

@@ -40,7 +40,10 @@ typedef struct _json_test_config_t {
    bool command_monitoring_allow_subset;
 } json_test_config_t;
 
-#define JSON_TEST_CONFIG_INIT {NULL, NULL, NULL, NULL, false}
+#define JSON_TEST_CONFIG_INIT       \
+   {                                \
+      NULL, NULL, NULL, NULL, false \
+   }
 
 bson_t *
 get_bson_from_json_file (char *filename);
@@ -75,8 +78,9 @@ server_type_from_test (const char *type);
 
 void
 activate_fail_point (mongoc_client_t *client,
-                     uint32_t server_id,
-                     const bson_t *opts);
+                     const uint32_t server_id,
+                     const bson_t *test,
+                     const char *key);
 
 void
 deactivate_fail_points (mongoc_client_t *client, uint32_t server_id);