CDRIVER-751 check discovered servers in same scan

Bug in unreleased Server Discovery And Monitoring implementation.

The spec requires a single-threaded scan to keep going until all nodes
have been checked, *including* nodes discovered during the scan. But
_mongoc_topology_do_blocking_scan quit once all servers known at the
*start* of the scan were checked. Discovered nodes are added to the
topology description, but not checked until the next scan.

With this patch, a block scan keeps going until all servers are checked,
including those discovered during this scan.

TODO: test_topology_scanner_oscillate currently fails, revealing that
in a pathological scenario the driver now errs on the other side, and
scans forever.

Author: ajdavis

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

@@ -88,6 +88,12 @@ mongoc_topology_scanner_add (mongoc_topology_scanner_t *ts,
                              const mongoc_host_list_t  *host,
                              uint32_t                   id);
 
+void
+mongoc_topology_scanner_add_and_scan (mongoc_topology_scanner_t *ts,
+                                      const mongoc_host_list_t  *host,
+                                      uint32_t                   id,
+                                      int64_t                    timeout_msec);
+
 void
 mongoc_topology_scanner_node_destroy (mongoc_topology_scanner_node_t *node,
                                       bool failed);

src/mongoc/mongoc-topology-scanner.c

@@ -26,14 +26,22 @@
 #endif
 
 #include "mongoc-counters-private.h"
-#include "mongoc-async-private.h"
-#include "mongoc-async-cmd-private.h"
 #include "utlist.h"
 #include "mongoc-topology-private.h"
 
 #undef MONGOC_LOG_DOMAIN
 #define MONGOC_LOG_DOMAIN "topology_scanner"
 
+static bool
+mongoc_topology_scanner_node_setup (mongoc_topology_scanner_node_t *node);
+
+static void
+mongoc_topology_scanner_ismaster_handler (mongoc_async_cmd_result_t async_status,
+                                          const bson_t             *ismaster_response,
+                                          int64_t                   rtt_msec,
+                                          void                     *data,
+                                          bson_error_t             *error);
+
 mongoc_topology_scanner_t *
 mongoc_topology_scanner_new (const mongoc_uri_t          *uri,
                              mongoc_topology_scanner_cb_t cb,
@@ -87,10 +95,10 @@ mongoc_topology_scanner_destroy (mongoc_topology_scanner_t *ts)
    bson_free (ts);
 }
 
-void
-mongoc_topology_scanner_add (mongoc_topology_scanner_t *ts,
-                             const mongoc_host_list_t  *host,
-                             uint32_t                   id)
+static mongoc_topology_scanner_node_t *
+_add_node (mongoc_topology_scanner_t *ts,
+           const mongoc_host_list_t  *host,
+           uint32_t                   id)
 {
    mongoc_topology_scanner_node_t *node = (mongoc_topology_scanner_node_t *)bson_malloc0 (sizeof (*node));
 
@@ -102,6 +110,37 @@ mongoc_topology_scanner_add (mongoc_topology_scanner_t *ts,
 
    DL_APPEND(ts->nodes, node);
 
+   return node;
+}
+
+void
+mongoc_topology_scanner_add (mongoc_topology_scanner_t *ts,
+                             const mongoc_host_list_t  *host,
+                             uint32_t                   id)
+{
+   _add_node (ts, host, id);
+}
+
+
+void
+mongoc_topology_scanner_add_and_scan (mongoc_topology_scanner_t *ts,
+                                      const mongoc_host_list_t  *host,
+                                      uint32_t                   id,
+                                      int64_t                    timeout_msec)
+{
+   mongoc_topology_scanner_node_t *node;
+
+   node = _add_node (ts, host, id);
+
+   if (mongoc_topology_scanner_node_setup (node)) {
+      node->cmd = mongoc_async_cmd (
+         ts->async, node->stream, ts->setup,
+         node->host.host, "admin",
+         &ts->ismaster_cmd,
+         &mongoc_topology_scanner_ismaster_handler,
+         node, (int32_t) timeout_msec);
+   }
+
    return;
 }
 

src/mongoc/mongoc-topology.c

@@ -32,25 +32,33 @@ _mongoc_topology_request_scan (mongoc_topology_t *topology);
 
 static bool
 _mongoc_topology_reconcile_add_nodes (void *item,
-                            void *ctx)
+                                      void *ctx)
 {
-   mongoc_server_description_t *sd = (mongoc_server_description_t *)item;
-   mongoc_topology_scanner_t *scanner = (mongoc_topology_scanner_t *)ctx;
+   mongoc_server_description_t *sd = item;
+   mongoc_topology_t *topology = (mongoc_topology_t *)ctx;
+   mongoc_topology_scanner_t *scanner = topology->scanner;
 
    if (! mongoc_topology_scanner_get_node (scanner, sd->id)) {
-      mongoc_topology_scanner_add (scanner, &sd->host, sd->id);
+      mongoc_topology_scanner_add_and_scan (scanner, &sd->host, sd->id,
+                                            topology->timeout_msec);
    }
 
    return true;
 }
 
 void
-mongoc_topology_reconcile (mongoc_topology_description_t *description,
-                           mongoc_topology_scanner_t     *scanner) {
+mongoc_topology_reconcile (mongoc_topology_t *topology) {
    mongoc_topology_scanner_node_t *ele, *tmp;
+   mongoc_topology_description_t *description;
+   mongoc_topology_scanner_t *scanner;
+
+   description = &topology->description;
+   scanner = topology->scanner;
 
    /* Add newly discovered nodes */
-   mongoc_set_for_each(description->servers, _mongoc_topology_reconcile_add_nodes, scanner);
+   mongoc_set_for_each(description->servers,
+                       _mongoc_topology_reconcile_add_nodes,
+                       topology);
 
    /* Remove removed nodes */
    DL_FOREACH_SAFE (scanner->nodes, ele, tmp) {
@@ -105,7 +113,7 @@ _mongoc_topology_scanner_cb (uint32_t      id,
        * server descriptions. We need to reconcile that with our monitoring agents
        */
 
-      mongoc_topology_reconcile(&topology->description, topology->scanner);
+      mongoc_topology_reconcile(topology);
 
       /* TODO only wake up all clients if we found any topology changes */
       mongoc_cond_broadcast (&topology->cond_client);

tests/TestSuite.h

@@ -98,6 +98,18 @@ extern "C" {
       } \
    } while (0)
 
+#define AWAIT(_condition) \
+   do { \
+      int64_t _start = bson_get_monotonic_time (); \
+      while (! _condition) { \
+         if (bson_get_monotonic_time() - _start > 1000 * 1000) { \
+            fprintf (stderr, \
+                     "%s:%d %s(): \"%s\" still false after 1 second\n", \
+                     __FILE__, __LINE__, __FUNCTION__, #_condition); \
+            abort (); \
+         } \
+      }  \
+   } while (0)
 
 #define MAX_TEST_NAME_LENGTH 500
 

tests/test-mongoc-topology-reconcile.c

@@ -131,10 +131,8 @@ _test_topology_reconcile_rs (bool pooled)
    RS_RESPONSE_TO_ISMASTER (server1, true, false, server0, server1);
 
    /* provide secondary in seed list */
-   /* TODO: CDRIVER-751 serverSelectionTryOnce=false shouldn't be needed */
    uri_str = bson_strdup_printf (
-      "mongodb://%s/?replicaSet=rs&connectTimeoutMS=10"
-         "&serverSelectionTryOnce=false",
+      "mongodb://%s/?replicaSet=rs",
       mock_server_get_host_and_port (server0));
 
    uri = mongoc_uri_new (uri_str);
@@ -153,7 +151,6 @@ _test_topology_reconcile_rs (bool pooled)
 
    /*
     * server0 is selected, server1 is discovered and added to scanner.
-    * TODO: CDRIVER-751 actually it isn't.
     */
    assert (selects_server (client, secondary_read_prefs, server0));
    assert (get_node (client->topology->scanner,
@@ -168,7 +165,16 @@ _test_topology_reconcile_rs (bool pooled)
     * remove server0 from set. tag primary, select w/ tags to trigger re-scan.
     */
    RS_RESPONSE_TO_ISMASTER (server1, true, true, server1);  /* server0 absent */
+
+   if (!pooled) {
+      /* TODO: SPEC-217 this selection should trigger one rescan but doesn't */
+      assert (!client->topology->stale);
+      assert (!selects_server (client, tag_read_prefs, server1));
+      assert (client->topology->stale);
+   }
+
    assert (selects_server (client, tag_read_prefs, server1));
+   assert (!client->topology->stale);
 
    assert (!get_node (client->topology->scanner,
                       mock_server_get_host_and_port (server0)));

tests/test-mongoc-topology-scanner.c

@@ -1,9 +1,13 @@
 #include <mongoc.h>
 
-#include "mongoc-topology-scanner-private.h"
+#include "mongoc-util-private.h"
+#include "mongoc-client-private.h"
+
 #include "mongoc-tests.h"
 #include "TestSuite.h"
 #include "mock_server/mock-server.h"
+#include "mock_server/future.h"
+#include "mock_server/future-functions.h"
 
 #undef MONGOC_LOG_DOMAIN
 #define MONGOC_LOG_DOMAIN "topology-scanner-test"
@@ -125,11 +129,172 @@ test_topology_scanner_ssl ()
 #endif
 
 
+/* 
+ * Servers discovered by a scan should be checked during that scan, CDRIVER-751. 
+ */
+void
+test_topology_scanner_discovery ()
+{
+   mock_server_t *primary;
+   mock_server_t *secondary;
+   char *primary_response;
+   char *secondary_response;
+   mongoc_client_t *client;
+   mongoc_topology_scanner_t *scanner;
+   char *uri_str;
+   mongoc_read_prefs_t *primary_pref;
+   bson_error_t error;
+   future_t *future;
+   request_t *request;
+   mongoc_server_description_t *sd;
+
+   primary = mock_server_new ();
+   secondary = mock_server_new ();
+   mock_server_run (primary);
+   mock_server_run (secondary);
+
+   primary_response = bson_strdup_printf (
+      "{'ok': 1, "
+      " 'ismaster': true,"
+      " 'setName': 'rs',"
+      " 'hosts': ['%s', '%s']}",
+      mock_server_get_host_and_port (primary),
+      mock_server_get_host_and_port (secondary));
+
+   secondary_response = bson_strdup_printf (
+      "{'ok': 1, "
+      " 'ismaster': false,"
+      " 'secondary': true,"
+      " 'setName': 'rs',"
+      " 'hosts': ['%s', '%s']}",
+      mock_server_get_host_and_port (primary),
+      mock_server_get_host_and_port (secondary));
+
+   uri_str = bson_strdup_printf ("mongodb://%s/?replicaSet=rs",
+                                 mock_server_get_host_and_port (primary));
+   client = mongoc_client_new (uri_str);
+   scanner = client->topology->scanner;
+   primary_pref = mongoc_read_prefs_new (MONGOC_READ_PRIMARY);
+
+   future = future_topology_select (client->topology, MONGOC_SS_READ,
+                                    primary_pref, 15, &error);
+
+   /* a single scan discovers *and* checks the secondary */
+   request = mock_server_receives_ismaster (primary);
+   mock_server_replies_simple (request, primary_response);
+   request_destroy (request);
+
+   /* let client process that response */
+   _mongoc_usleep (250 * 1000);
+
+   /* a check of the secondary is scheduled in this scan */
+   AWAIT (scanner->async->ncmds);
+
+   request = mock_server_receives_ismaster (secondary);
+   mock_server_replies_simple (request, secondary_response);
+   /* scan completes */
+   AWAIT (!scanner->async->ncmds);
+   ASSERT_OR_PRINT ((sd = future_get_mongoc_server_description_ptr (future)),
+                    error);
+
+   mongoc_server_description_destroy (sd);
+   request_destroy (request);
+   mongoc_read_prefs_destroy (primary_pref);
+   bson_free (secondary_response);
+   bson_free (primary_response);
+   bson_free (uri_str);
+   mongoc_client_destroy (client);
+   mock_server_destroy (secondary);
+   mock_server_destroy (primary);
+}
+
+
+/*
+ * Check that scanner doesn't spinlock if two primaries point at each other.
+ */
+void
+test_topology_scanner_oscillate ()
+{
+   mock_server_t *server0;
+   mock_server_t *server1;
+   char *server0_response;
+   char *server1_response;
+   mongoc_client_t *client;
+   mongoc_topology_scanner_t *scanner;
+   char *uri_str;
+   mongoc_read_prefs_t *primary_pref;
+   bson_error_t error;
+   future_t *future;
+   request_t *request;
+
+   server0 = mock_server_new ();
+   server1 = mock_server_new ();
+   mock_server_run (server0);
+   mock_server_run (server1);
+
+   /* server 0 says it's primary, but only server 1 is in the set */
+   server0_response = bson_strdup_printf (
+      "{'ok': 1, "
+      " 'ismaster': true,"
+      " 'setName': 'rs',"
+      " 'hosts': ['%s']}",
+      mock_server_get_host_and_port (server1));
+
+   /* the opposite */
+   server1_response = bson_strdup_printf (
+      "{'ok': 1, "
+      " 'ismaster': true,"
+      " 'setName': 'rs',"
+      " 'hosts': ['%s']}",
+      mock_server_get_host_and_port (server0));
+
+   /* start with server 0 */
+   uri_str = bson_strdup_printf ("mongodb://%s/?replicaSet=rs",
+                                 mock_server_get_host_and_port (server0));
+   client = mongoc_client_new (uri_str);
+   scanner = client->topology->scanner;
+   primary_pref = mongoc_read_prefs_new (MONGOC_READ_PRIMARY);
+
+   assert (!scanner->async->ncmds);
+   future = future_topology_select (client->topology, MONGOC_SS_READ,
+                                    primary_pref, 15, &error);
+
+   /* a single scan discovers servers 0 and 1 */
+   request = mock_server_receives_ismaster (server0);
+   mock_server_replies_simple (request, server0_response);
+   request_destroy (request);
+
+   /* let client process that response */
+   _mongoc_usleep (250 * 1000);
+   AWAIT (scanner->async->ncmds);
+
+   request = mock_server_receives_ismaster (server1);
+   mock_server_replies_simple (request, server1_response);
+   AWAIT (!scanner->async->ncmds);
+
+   assert (!future_get_mongoc_server_description_ptr (future));
+   request_destroy (request);
+   mongoc_read_prefs_destroy (primary_pref);
+   bson_free (server1_response);
+   bson_free (server0_response);
+   bson_free (uri_str);
+   mongoc_client_destroy (client);
+   mock_server_destroy (server1);
+   mock_server_destroy (server0);
+}
+
+
 void
 test_topology_scanner_install (TestSuite *suite)
 {
    TestSuite_Add (suite, "/TOPOLOGY/scanner", test_topology_scanner);
 #ifdef MONGOC_ENABLE_SSL
    TestSuite_Add (suite, "/TOPOLOGY/scanner_ssl", test_topology_scanner_ssl);
 #endif
+   TestSuite_Add (suite, "/TOPOLOGY/scanner_discovery",
+                  test_topology_scanner_discovery);
+/*
+   TestSuite_Add (suite, "/TOPOLOGY/scanner_oscillate",
+                  test_topology_scanner_oscillate);
+*/
 }