WIP: fixing connection logic

Author: kevinAlbs

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

@@ -2256,6 +2256,7 @@ _mongoc_cluster_stream_for_server (mongoc_cluster_t *cluster,
        * into account.
        */
 
+      // LBTODO: do not invalidate the server!
       mongoc_topology_invalidate_server (topology, server_id, err_ptr);
       mongoc_cluster_disconnect_node (cluster, server_id);
       bson_mutex_lock (&topology->mutex);
@@ -2382,13 +2383,15 @@ mongoc_cluster_fetch_stream_single (mongoc_cluster_t *cluster,
       }
    } else {
       if (!reconnect_ok) {
+         MONGOC_DEBUG ("mongoc_cluster_fetch_stream_single reconnect_ok==false, erroring");
          stream_not_found (
             topology, server_id, scanner_node->host.host_and_port, error);
          return NULL;
       }
 
       /* save the scanner node address in case it is removed during the scan. */
       address = bson_strdup (scanner_node->host.host_and_port);
+      MONGOC_DEBUG ("mongoc_cluster_fetch_stream_single is doing a blocking scan");
       _mongoc_topology_do_blocking_scan (topology, error);
       if (error->code) {
          bson_free (address);
@@ -3040,6 +3043,7 @@ mongoc_cluster_check_interval (mongoc_cluster_t *cluster, uint32_t server_id)
                          MONGOC_ERROR_STREAM_SOCKET,
                          "connection closed");
          mongoc_cluster_disconnect_node (cluster, server_id);
+         // LBTODO (error handing)
          mongoc_topology_invalidate_server (topology, server_id, &error);
          return false;
       }
@@ -3066,6 +3070,7 @@ mongoc_cluster_check_interval (mongoc_cluster_t *cluster, uint32_t server_id)
 
       if (!r) {
          mongoc_cluster_disconnect_node (cluster, server_id);
+         // LBTODO (error handling): do not invalidate the server!
          mongoc_topology_invalidate_server (topology, server_id, &error);
       }
    }

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

@@ -226,6 +226,9 @@ _mongoc_topology_get_handshake_cmd (mongoc_topology_t *topology);
 void
 _mongoc_topology_request_scan (mongoc_topology_t *topology);
 
+// LBTODO: call this when in load balancer mode.
+// If a network error occurs on a server, there is no reason to assume
+// we should wait for a cooldown period before attempting subsequent connection.
 void
 _mongoc_topology_bypass_cooldown (mongoc_topology_t *topology);
 

src/libmongoc/src/mongoc/mongoc-topology-scanner.c

@@ -654,6 +654,7 @@ _async_success (mongoc_async_cmd_t *acmd,
          hello_response, &node->speculative_auth_response);
    }
 
+   // LBTODO: skip this if load balanced?
    /* mongoc_topology_scanner_cb_t takes rtt_msec, not usec */
    ts->cb (node->id,
            hello_response,
@@ -715,6 +716,7 @@ _async_error_or_timeout (mongoc_async_cmd_t *acmd,
 
       /* call the topology scanner callback. cannot connect to this node.
        * callback takes rtt_msec, not usec. */
+      // LBTODO: skip this if load balanced?
       ts->cb (node->id, NULL, duration_usec / 1000, ts->cb_data, error);
    } else {
       /* there are still more commands left for this node or it succeeded
@@ -978,6 +980,8 @@ mongoc_topology_scanner_node_setup (mongoc_topology_scanner_node_t *node,
    _mongoc_topology_scanner_monitor_heartbeat_started (node->ts, &node->host);
    start = bson_get_monotonic_time ();
 
+   // LBTODO assert that a node already has a stream.
+   // A load balanced cluster only uses the topology scanner for connection establishment. Not monitoring.
    /* if there is already a working stream, push it back to be re-scanned. */
    if (node->stream) {
       _begin_hello_cmd (node, node->stream, true /* is_setup_done */, NULL, 0);
@@ -1117,6 +1121,7 @@ mongoc_topology_scanner_start (mongoc_topology_scanner_t *ts,
 
    DL_FOREACH_SAFE (ts->nodes, node, tmp)
    {
+      // LBTODO: Should we skip the cooldown period?
       skip =
          obey_cooldown && mongoc_topology_scanner_node_in_cooldown (node, now);
 

src/libmongoc/src/mongoc/mongoc-topology.c

@@ -1010,7 +1010,10 @@ mongoc_topology_select_server_id (mongoc_topology_t *topology,
       _mongoc_topology_description_monitor_opening (&topology->description);
 
       if (topology->description.type == MONGOC_TOPOLOGY_LOADBALANCED) {
+         MONGOC_DEBUG ("mongoc_topology_select_server_id is returning single load balancer, not proceeding with monitoring");
          /* Bypass server selection loop. Always select the only server. */
+         // LBTODO: by skipping topology scanning, no connection is created
+         // to the load balancer here. It will be created in mongoc_cluster_fetch_stream_singled.
          selected_server = mongoc_topology_description_select (
             &topology->description, optype, read_prefs, local_threshold_ms);
 
@@ -1021,6 +1024,7 @@ mongoc_topology_select_server_id (mongoc_topology_t *topology,
                error);
             return 0;
          }
+         return selected_server->id;
       }
 
       tried_once = false;
@@ -1130,6 +1134,8 @@ mongoc_topology_select_server_id (mongoc_topology_t *topology,
                error);
             return 0;
          }
+         bson_mutex_unlock (&topology->mutex);
+         return selected_server->id;
       }
 
       if (!selected_server) {
@@ -1329,6 +1335,12 @@ _mongoc_topology_update_from_handshake (mongoc_topology_t *topology,
 
    bson_mutex_lock (&topology->mutex);
 
+   if (topology->description.type == MONGOC_TOPOLOGY_LOADBALANCED) {
+      /* Do not update the topology from a connection handshake. */
+      bson_mutex_unlock (&topology->mutex);
+      return true;
+   }
+
    /* return false if server was removed from topology */
    has_server = _mongoc_topology_update_no_lock (sd->id,
                                                  &sd->last_hello_response,

src/libmongoc/tests/test-mongoc-loadbalanced.c

@@ -21,7 +21,7 @@
 #include "TestSuite.h"
 
 static char *
-loadbalanced_uri (void)
+loadbalanced_single_uri (void)
 {
    /* TODO (CDRIVER-4062): This will need to add TLS and auth to the URI when
     * run in evergreen. */
@@ -33,7 +33,7 @@ test_loadbalanced_sessions_supported (void *unused)
 {
    mongoc_client_t *client;
    mongoc_client_session_t *session;
-   char *uristr = loadbalanced_uri ();
+   char *uristr = loadbalanced_single_uri ();
    bson_error_t error;
 
    client = mongoc_client_new (uristr);
@@ -50,7 +50,7 @@ test_loadbalanced_sessions_do_not_expire (void *unused)
 {
    mongoc_client_t *client;
    mongoc_client_session_t *session;
-   char *uristr = loadbalanced_uri ();
+   char *uristr = loadbalanced_single_uri ();
    bson_error_t error;
    bson_t *lsid1;
    bson_t *lsid2;
@@ -79,6 +79,102 @@ test_loadbalanced_sessions_do_not_expire (void *unused)
    mongoc_client_destroy (client);
 }
 
+typedef struct {
+   int topology_changed;
+   int server_changed;
+   int heartbeat_started;
+   int heartbeat_succeeded;
+   int heartbeat_failed;
+} sdam_event_counters_t;
+
+static void server_changed(mongoc_apm_server_changed_t *event) {
+   sdam_event_counters_t *counters = (sdam_event_counters_t*) mongoc_apm_server_changed_get_context (event);
+   counters->server_changed++;
+}
+static void topology_changed(mongoc_apm_topology_changed_t *event) {
+   sdam_event_counters_t *counters = (sdam_event_counters_t*) mongoc_apm_topology_changed_get_context (event);
+   counters->topology_changed++;
+}
+static void heartbeat_started(mongoc_apm_server_heartbeat_started_t *event) {
+   sdam_event_counters_t *counters = (sdam_event_counters_t*) mongoc_apm_server_heartbeat_started_get_context (event);
+   counters->heartbeat_started++;
+}
+static void heartbeat_failed(mongoc_apm_server_heartbeat_failed_t *event) {
+   sdam_event_counters_t *counters = (sdam_event_counters_t*) mongoc_apm_server_heartbeat_failed_get_context (event);
+   counters->heartbeat_failed++;
+}
+static void heartbeat_succeeded(mongoc_apm_server_heartbeat_succeeded_t *event) {
+   sdam_event_counters_t *counters = (sdam_event_counters_t*) mongoc_apm_server_heartbeat_succeeded_get_context (event);
+   counters->heartbeat_succeeded++;
+}
+
+static mongoc_apm_callbacks_t * make_callbacks (void) {
+   mongoc_apm_callbacks_t *callbacks;
+
+   callbacks = mongoc_apm_callbacks_new ();
+   mongoc_apm_set_server_changed_cb (callbacks, server_changed);
+   mongoc_apm_set_topology_changed_cb (callbacks, topology_changed);
+   mongoc_apm_set_server_heartbeat_started_cb (callbacks, heartbeat_started);
+   mongoc_apm_set_server_heartbeat_failed_cb (callbacks, heartbeat_failed);
+   mongoc_apm_set_server_heartbeat_succeeded_cb (callbacks, heartbeat_succeeded);
+   return callbacks;
+}
+
+/* Test that connection to a load balanced cluster is possible, and the topology does not change after a connection is created. */
+static void
+test_loadbalanced_ping_single (void *unused) {
+   mongoc_client_t *client;
+   mongoc_client_session_t *session;
+   char *uristr = loadbalanced_single_uri ();
+   bson_error_t error;
+   mongoc_server_description_t *sd;
+   mongoc_apm_callbacks_t *callbacks;
+   sdam_event_counters_t counters = {0};
+   bool ret;
+
+   client = mongoc_client_new (uristr);
+   callbacks = make_callbacks();
+   mongoc_client_set_apm_callbacks (client, callbacks, &counters);
+
+   sd = mongoc_client_select_server (client, true, NULL /* read prefs */, &error);
+   ASSERT_OR_PRINT (sd != NULL, error);
+
+   /* There are two topology changed events:
+    * - The initial changed event
+    * - The server transition of Unknown => LoadBalancer */
+   ASSERT_CMPINT (counters.topology_changed, ==, 2);
+   ASSERT_CMPINT (counters.server_changed, ==, 1);
+   /* Selecting a server does NOT open a new connection in load balanced mode. */
+   ASSERT_CMPINT (counters.heartbeat_started, ==, 0);
+
+   // LBTODO: this is where I left off.
+   // The server description is updated by monitoring and includes the wire version.
+   // If we skip monitoring and force the server description, how do we update the wire version from the handshake?
+   /* The first operation requiring I/O opens a connection. */
+   ret = mongoc_client_command_simple (client, "admin", tmp_bson ("{'ping': 1}"), NULL /* read prefs */, NULL /* reply */, &error);
+   ASSERT_OR_PRINT (ret, error);
+
+   /* No topology changes should have occurred */
+   ASSERT_CMPINT (counters.topology_changed, ==, 2);
+   ASSERT_CMPINT (counters.server_changed, ==, 1);
+   /* The new connection should have produced a heartbeat event. */
+   ASSERT_CMPINT (counters.heartbeat_started, ==, 1);
+   ASSERT_CMPINT (counters.heartbeat_succeeded, ==, 1);
+   
+   bson_free (uristr);
+   mongoc_server_description_destroy (sd);
+   mongoc_apm_callbacks_destroy (callbacks);
+   mongoc_client_destroy (client);
+}
+
+/* Test that the handshake reply does not update the topology in load balanced mode. */
+static void
+test_loadbalanced_handshake_does_not_update (void *unused);
+
+/* Test that re-establishing a connection after a network error does not wait for the cooldown period in single threaded mode. */
+static void
+test_loadbalanced_no_cooldown_on_reconnect (void *unused);
+
 static int
 skip_if_not_loadbalanced (void)
 {
@@ -104,4 +200,10 @@ test_loadbalanced_install (TestSuite *suite)
                       NULL /* ctx */,
                       NULL /* dtor */,
                       skip_if_not_loadbalanced);
+   TestSuite_AddFull (suite,
+                      "/loadbalanced/ping/single",
+                      test_loadbalanced_ping_single,
+                      NULL /* ctx */,
+                      NULL /* dtor */,
+                      skip_if_not_loadbalanced);
 }