CDRIVER-2172 robust mongoc_client_server_select

If a server is selected and its socket has not been used in 5 seconds,
check if the socket is still good and retry once if not.

Author: ajdavis

build/generate-future-functions.py

@@ -253,6 +253,13 @@
                     [param("mongoc_client_ptr", "client"),
                      param("bson_error_ptr", "error")]),
 
+    future_function("mongoc_server_description_ptr",
+                    "mongoc_client_select_server",
+                    [param("mongoc_client_ptr", "client"),
+                     param("bool", "for_writes"),
+                     param("const_mongoc_read_prefs_ptr", "prefs"),
+                     param("bson_error_ptr", "error")]),
+
     future_function("bool",
                     "mongoc_database_command_simple",
                     [param("mongoc_database_ptr", "database"),

src/mongoc/mongoc-client.c

@@ -2167,6 +2167,10 @@ mongoc_client_select_server (mongoc_client_t *client,
                              const mongoc_read_prefs_t *prefs,
                              bson_error_t *error)
 {
+   mongoc_ss_optype_t optype = for_writes ? MONGOC_SS_WRITE : MONGOC_SS_READ;
+   mongoc_server_description_t *sd;
+   bson_error_t tmp_err;
+
    if (for_writes && prefs) {
       bson_set_error (error,
                       MONGOC_ERROR_SERVER_SELECTION,
@@ -2179,10 +2183,25 @@ mongoc_client_select_server (mongoc_client_t *client,
       return NULL;
    }
 
-   return mongoc_topology_select (client->topology,
-                                  for_writes ? MONGOC_SS_WRITE : MONGOC_SS_READ,
-                                  prefs,
-                                  error);
+   sd = mongoc_topology_select (client->topology, optype, prefs, error);
+   if (!sd) {
+      return NULL;
+   }
+
+   if (mongoc_cluster_check_interval (&client->cluster, sd->id, &tmp_err)) {
+      /* check not required, or it succeeded */
+      return sd;
+   }
+
+   /* check failed, retry once */
+   mongoc_server_description_destroy (sd);
+   sd = mongoc_topology_select (client->topology, optype, prefs, error);
+   if (sd) {
+      return sd;
+   }
+
+   memcpy (error, &tmp_err, sizeof tmp_err);
+   return NULL;
 }
 
 bool

src/mongoc/mongoc-cluster-private.h

@@ -100,6 +100,11 @@ _mongoc_cluster_buffer_iovec (mongoc_iovec_t *iov,
                               int skip,
                               char *buffer);
 
+bool
+mongoc_cluster_check_interval (mongoc_cluster_t *cluster,
+                               uint32_t server_id,
+                               bson_error_t *error);
+
 bool
 mongoc_cluster_sendv_to_server (mongoc_cluster_t *cluster,
                                 mongoc_rpc_t *rpcs,

src/mongoc/mongoc-cluster.c

@@ -2162,10 +2162,34 @@ mongoc_cluster_get_max_msg_size (mongoc_cluster_t *cluster)
 }
 
 
-static bool
-_mongoc_cluster_check_interval (mongoc_cluster_t *cluster,
-                                uint32_t server_id,
-                                bson_error_t *error)
+/*
+ *--------------------------------------------------------------------------
+ *
+ * mongoc_cluster_check_interval --
+ *
+ *      Server Discovery And Monitoring Spec:
+ *
+ *      "Only for single-threaded clients. If a server is selected that has an
+ *      existing connection that has been idle for socketCheckIntervalMS, the
+ *      driver MUST check it, by using it for an ismaster call. Whether
+ *      ismaster succeeds or fails, the driver MUST update its topology
+ *      description with the outcome. Then, if the check failed, the driver
+ *      MUST retry server selection once."
+ *
+ * Returns:
+ *      True if the check succeeded or no check was required, false if the
+ *      check failed.
+ *
+ * Side effects:
+ *      If a check is needed, updates the topology with its outcome.
+ *
+ *--------------------------------------------------------------------------
+ */
+
+bool
+mongoc_cluster_check_interval (mongoc_cluster_t *cluster,
+                               uint32_t server_id,
+                               bson_error_t *error)
 {
    mongoc_topology_t *topology;
    mongoc_topology_scanner_node_t *scanner_node;
@@ -2223,7 +2247,7 @@ _mongoc_cluster_check_interval (mongoc_cluster_t *cluster,
                                               "admin",
                                               &command,
                                               &reply,
-                                              error);
+                                              error /* OUT */);
 
       now = bson_get_monotonic_time ();
 
@@ -2234,7 +2258,7 @@ _mongoc_cluster_check_interval (mongoc_cluster_t *cluster,
                                                    &reply,
                                                    (now - before_ismaster) /
                                                       1000, /* RTT_MS */
-                                                   error);
+                                                   error /* IN */);
 
       bson_destroy (&reply);
    }
@@ -2302,23 +2326,11 @@ mongoc_cluster_sendv_to_server (mongoc_cluster_t *cluster,
       write_concern = cluster->client->write_concern;
    }
 
-   if (!_mongoc_cluster_check_interval (
-          cluster, server_stream->sd->id, error)) {
-      GOTO (done);
-   }
-
    _mongoc_array_clear (&cluster->iov);
 #ifdef MONGOC_ENABLE_COMPRESSION
    compressor_id = mongoc_server_description_compressor_id (server_stream->sd);
 #endif
 
-   /*
-    * TODO: We can probably remove the need for sendv and just do send since
-    * we support write concerns now. Also, we clobber our getlasterror on
-    * each subsequent mutation. It's okay, since it comes out correct anyway,
-    * just useless work (and technically the request_id changes).
-    */
-
    need_gle = _mongoc_rpc_needs_gle (rpc, write_concern);
    _mongoc_cluster_inc_egress_rpc (rpc);
    _mongoc_rpc_gather (rpc, &cluster->iov);

tests/mock_server/future-functions.c

@@ -456,6 +456,28 @@ background_mongoc_client_get_database_names (void *data)
    return NULL;
 }
 
+static void *
+background_mongoc_client_select_server (void *data)
+{
+   future_t *future = (future_t *) data;
+   future_value_t return_value;
+
+   return_value.type = future_value_mongoc_server_description_ptr_type;
+
+   future_value_set_mongoc_server_description_ptr (
+      &return_value,
+      mongoc_client_select_server (
+         future_value_get_mongoc_client_ptr (future_get_param (future, 0)),
+         future_value_get_bool (future_get_param (future, 1)),
+         future_value_get_const_mongoc_read_prefs_ptr (future_get_param (future, 2)),
+         future_value_get_bson_error_ptr (future_get_param (future, 3))
+      ));
+
+   future_resolve (future, return_value);
+
+   return NULL;
+}
+
 static void *
 background_mongoc_database_command_simple (void *data)
 {
@@ -1240,6 +1262,32 @@ future_client_get_database_names (
    return future;
 }
 
+future_t *
+future_client_select_server (
+   mongoc_client_ptr client,
+   bool for_writes,
+   const_mongoc_read_prefs_ptr prefs,
+   bson_error_ptr error)
+{
+   future_t *future = future_new (future_value_mongoc_server_description_ptr_type,
+                                  4);
+   
+   future_value_set_mongoc_client_ptr (
+      future_get_param (future, 0), client);
+   
+   future_value_set_bool (
+      future_get_param (future, 1), for_writes);
+   
+   future_value_set_const_mongoc_read_prefs_ptr (
+      future_get_param (future, 2), prefs);
+   
+   future_value_set_bson_error_ptr (
+      future_get_param (future, 3), error);
+   
+   future_start (future, background_mongoc_client_select_server);
+   return future;
+}
+
 future_t *
 future_database_command_simple (
    mongoc_database_ptr database,

tests/mock_server/future-functions.h

@@ -223,6 +223,16 @@ future_client_get_database_names (
 );
 
 
+future_t *
+future_client_select_server (
+
+   mongoc_client_ptr client,
+   bool for_writes,
+   const_mongoc_read_prefs_ptr prefs,
+   bson_error_ptr error
+);
+
+
 future_t *
 future_database_command_simple (
 

tests/test-mongoc-client.c

@@ -2228,6 +2228,89 @@ test_mongoc_client_select_server_error_pooled (void)
 }
 
 
+/* CDRIVER-2172: in single mode, if the selected server has a socket that's been
+ * idle for socketCheckIntervalMS, check it with an ismaster call. If that
+ * fails, retry once.
+ */
+static void
+_test_mongoc_client_select_server_retry (bool retry_succeeds)
+{
+   char *ismaster;
+   mock_server_t *server;
+   mongoc_uri_t *uri;
+   mongoc_client_t *client;
+   bson_error_t error;
+   request_t *request;
+   future_t *future;
+   mongoc_server_description_t *sd;
+
+   server = mock_server_new ();
+   mock_server_run (server);
+   ismaster = bson_strdup_printf ("{'ok': 1, 'ismaster': true,"
+                                  " 'secondary': false,"
+                                  " 'setName': 'rs', 'hosts': ['%s']}",
+                                  mock_server_get_host_and_port (server));
+
+   uri = mongoc_uri_copy (mock_server_get_uri (server));
+   mongoc_uri_set_option_as_utf8 (uri, "replicaSet", "rs");
+   mongoc_uri_set_option_as_int32 (uri, "socketCheckIntervalMS", 50);
+   client = mongoc_client_new_from_uri (uri);
+
+   /* first selection succeeds */
+   future = future_client_select_server (client, true, NULL, &error);
+   request = mock_server_receives_ismaster (server);
+   mock_server_replies_simple (request, ismaster);
+   request_destroy (request);
+   sd = future_get_mongoc_server_description_ptr (future);
+   ASSERT_OR_PRINT (sd, error);
+
+   future_destroy (future);
+   mongoc_server_description_destroy (sd);
+
+   /* let socketCheckIntervalMS pass */
+   _mongoc_usleep (100 * 1000);
+
+   /* second selection requires ismaster check, which fails */
+   future = future_client_select_server (client, true, NULL, &error);
+   request = mock_server_receives_ismaster (server);
+   mock_server_hangs_up (request);
+   request_destroy (request);
+
+   /* mongoc_client_select_server retries once */
+   request = mock_server_receives_ismaster (server);
+   if (retry_succeeds) {
+      mock_server_replies_simple (request, ismaster);
+      sd = future_get_mongoc_server_description_ptr (future);
+      ASSERT_OR_PRINT (sd, error);
+      mongoc_server_description_destroy (sd);
+   } else {
+      mock_server_hangs_up (request);
+      sd = future_get_mongoc_server_description_ptr (future);
+      BSON_ASSERT (sd == NULL);
+   }
+
+   future_destroy (future);
+   request_destroy (request);
+   mongoc_client_destroy (client);
+   mongoc_uri_destroy (uri);
+   bson_free (ismaster);
+   mock_server_destroy (server);
+}
+
+
+static void
+test_mongoc_client_select_server_retry_succeed (void)
+{
+   _test_mongoc_client_select_server_retry (true);
+}
+
+static void
+test_mongoc_client_select_server_retry_fail (void)
+{
+   _test_mongoc_client_select_server_retry (false);
+}
+
+
 #if defined(MONGOC_ENABLE_SSL_OPENSSL) || \
    defined(MONGOC_ENABLE_SSL_SECURE_TRANSPORT)
 static bool
@@ -3011,6 +3094,12 @@ test_client_install (TestSuite *suite)
    TestSuite_AddLive (suite,
                       "/Client/select_server/err/pooled",
                       test_mongoc_client_select_server_error_pooled);
+   TestSuite_AddMockServerTest (suite,
+                                "/Client/select_server/retry/succeed",
+                                test_mongoc_client_select_server_retry_succeed);
+   TestSuite_AddMockServerTest (suite,
+                                "/Client/select_server/retry/fail",
+                                test_mongoc_client_select_server_retry_fail);
    TestSuite_AddFull (suite,
                       "/Client/null_error_pointer/single",
                       test_null_error_pointer_single,