CDRIVER-879 test cluster node reconnect

Author: ajdavis

tests/mock_server/mock-server.c

@@ -1114,6 +1114,47 @@ mock_server_hangs_up (request_t *request)
 }
 
 
+/*--------------------------------------------------------------------------
+ *
+ * mock_server_resets --
+ *
+ *       Forcefully reset a connection from the client.
+ *
+ * Returns:
+ *       None.
+ *
+ * Side effects:
+ *       Causes ECONNRESET on the client side.
+ *
+ *--------------------------------------------------------------------------
+ */
+
+void
+mock_server_resets (request_t *request)
+{
+   struct linger no_linger;
+   no_linger.l_onoff = 1;
+   no_linger.l_linger = 0;
+
+   if (mock_server_get_verbose (request->server)) {
+      printf ("%5.2f  %hu <- %hu \treset!\n",
+              mock_server_get_uptime_sec (request->server),
+              request->client_port,
+              request_get_server_port (request));
+      fflush (stdout);
+   }
+
+   /* send RST packet to client */
+   mongoc_stream_setsockopt (request->client,
+                             SOL_SOCKET,
+                             SO_LINGER,
+                             &no_linger,
+                             sizeof no_linger);
+
+   mongoc_stream_close (request->client);
+}
+
+
 /*--------------------------------------------------------------------------
  *
  * mock_server_replies --

tests/mock_server/mock-server.h

@@ -131,6 +131,8 @@ request_t *mock_server_receives_kill_cursors (mock_server_t *server,
 
 void mock_server_hangs_up (request_t *request);
 
+void mock_server_resets (request_t *request);
+
 void mock_server_replies (request_t *request,
                           uint32_t flags,
                           int64_t cursor_id,

tests/mock_server/request.c

@@ -461,6 +461,28 @@ request_get_server_port (request_t *request)
 }
 
 
+/*--------------------------------------------------------------------------
+ *
+ * request_get_client_port --
+ *
+ *       Get the client port this request was sent from.
+ *
+ * Returns:
+ *       A port number.
+ *
+ * Side effects:
+ *       None.
+ *
+ *--------------------------------------------------------------------------
+ */
+
+uint16_t
+request_get_client_port (request_t *request)
+{
+   return request->client_port;
+}
+
+
 /*--------------------------------------------------------------------------
  *
  * request_destroy --

tests/mock_server/request.h

@@ -91,6 +91,8 @@ bool request_matches_kill_cursors (const request_t *request,
 
 uint16_t request_get_server_port (request_t *request);
 
+uint16_t request_get_client_port (request_t *request);
+
 void request_destroy (request_t *request);
 
 #endif //REQUEST_H

tests/test-mongoc-cluster.c

@@ -1,10 +1,15 @@
 #include <mongoc.h>
 
 #include "mongoc-client-private.h"
+#include "mongoc-uri-private.h"
 
+#include "mock_server/mock-server.h"
+#include "mock_server/future.h"
+#include "mock_server/future-functions.h"
 #include "mongoc-tests.h"
 #include "TestSuite.h"
 #include "test-libmongoc.h"
+#include "test-conveniences.h"
 
 
 #undef MONGOC_LOG_DOMAIN
@@ -91,9 +96,125 @@ test_get_max_msg_size (void)
    mongoc_client_pool_destroy (pool);
 }
 
+
+#define ASSERT_CURSOR_ERR() do { \
+      char *error_message = bson_strdup_printf ( \
+         "Failed to read 4 bytes from socket within %d milliseconds.", \
+         socket_timeout_ms); \
+      BSON_ASSERT (!future_get_bool (future)); \
+      BSON_ASSERT (mongoc_cursor_error (cursor, &error)); \
+      ASSERT_CMPINT (error.domain, ==, MONGOC_ERROR_STREAM); \
+      ASSERT_CMPINT (error.code, ==, MONGOC_ERROR_STREAM_SOCKET); \
+      ASSERT_CMPSTR (error.message, error_message); \
+      bson_free (error_message); \
+   } while (0)
+
+
+#define START_QUERY(client_port_variable) do { \
+      cursor = mongoc_collection_find (collection, \
+                                        MONGOC_QUERY_NONE, \
+                                        0, 0, 0, tmp_bson ("{}"), \
+                                        NULL, NULL); \
+      future = future_cursor_next (cursor, &doc); \
+      request = mock_server_receives_query (server, "test.test", \
+                                             MONGOC_QUERY_SLAVE_OK, 0, 0, \
+                                             "{}", NULL); \
+      BSON_ASSERT (request); \
+      client_port_variable = request_get_client_port (request); \
+   } while (0)
+
+
+#define CLEANUP_QUERY() do { \
+      request_destroy (request); \
+      future_destroy (future); \
+      mongoc_cursor_destroy (cursor); \
+   } while (0)
+
+
+/* test that we reconnect a cluster node after disconnect */
+static void
+_test_cluster_node_disconnect (bool pooled)
+{
+   mock_server_t *server;
+   const int32_t socket_timeout_ms = 100;
+   mongoc_uri_t *uri;
+   mongoc_client_pool_t *pool = NULL;
+   mongoc_client_t *client;
+   mongoc_collection_t *collection;
+   const bson_t *doc;
+   mongoc_cursor_t *cursor;
+   future_t *future;
+   request_t *request;
+   uint16_t client_port_0, client_port_1;
+   bson_error_t error;
+
+   server = mock_server_with_autoismaster (0);
+   mock_server_run (server);
+
+   uri = mongoc_uri_copy (mock_server_get_uri (server));
+   mongoc_uri_set_option_as_int32 (uri, "socketTimeoutMS", socket_timeout_ms);
+
+   if (pooled) {
+      pool = mongoc_client_pool_new (uri);
+      client = mongoc_client_pool_pop (pool);
+   } else {
+      client = mongoc_client_new_from_uri (uri);
+   }
+
+   collection = mongoc_client_get_collection (client, "test", "test");
+
+   /* query 0 fails. set client_port_0 to the port used by the query. */
+   START_QUERY (client_port_0);
+   if (pooled) {
+      suppress_one_message ();
+   }
+
+   mock_server_resets (request);
+   ASSERT_CURSOR_ERR ();
+   CLEANUP_QUERY ();
+
+   /* query 1 opens a new socket. set client_port_1 to the new port. */
+   START_QUERY (client_port_1);
+   ASSERT_CMPINT (client_port_1, !=, client_port_0);
+   mock_server_replies_simple (request, "{'a': 1}");
+
+   /* success! */
+   BSON_ASSERT (future_get_bool (future));
+
+   CLEANUP_QUERY ();
+   mongoc_collection_destroy (collection);
+
+   if (pooled) {
+      mongoc_client_pool_push (pool, client);
+      mongoc_client_pool_destroy (pool);
+   } else {
+      mongoc_client_destroy (client);
+   }
+
+   mongoc_uri_destroy (uri);
+   mock_server_destroy (server);
+}
+
+
+static void
+test_cluster_node_disconnect_single (void)
+{
+   _test_cluster_node_disconnect (false);
+}
+
+
+static void
+test_cluster_node_disconnect_pooled (void)
+{
+   _test_cluster_node_disconnect (true);
+}
+
+
 void
 test_cluster_install (TestSuite *suite)
 {
    TestSuite_Add (suite, "/Cluster/test_get_max_bson_obj_size", test_get_max_bson_obj_size);
    TestSuite_Add (suite, "/Cluster/test_get_max_msg_size", test_get_max_msg_size);
+   TestSuite_Add (suite, "/Cluster/disconnect/single", test_cluster_node_disconnect_single);
+   TestSuite_Add (suite, "/Cluster/disconnect/pooled", test_cluster_node_disconnect_pooled);
 }