CDRIVER-789 crash on network err from removed node

If the scanner knows about A, a primary, and B, a secondary:

* B is removed from the replica set and shut down
* The scanner begins, launching async commands to check A and B
* A responds first and tells the scanner to remove B
* mongoc_topology_reconcile removes B's mongoc_topology_scanner_node_t and
  destroys it, stream and all
* But the mongoc_async_cmd_t for B is still active in the scanner, with the
  same stream.
* B's mongoc_async_cmd_t fails with a network error, and passes the error into
  the command callback.
* The command callback sees the error and destroys the same stream *again*,
  crashing.

The solution is to not destroy removed nodes in mongoc_topology_reconcile, but
close their streams, and mark them "retired" for the remainder of the
scan. At the end of the scan destroy retired nodes.

Author: ajdavis

CMakeLists.txt

@@ -242,6 +242,7 @@ function(mongoc_add_example example use_shared)
 endfunction()
 
 set(test-libmongoc-sources
+   ${SOURCE_DIR}/tests/debug-stream.c
    ${SOURCE_DIR}/tests/ha-test.c
    ${SOURCE_DIR}/tests/json-test.c
    ${SOURCE_DIR}/tests/mock_server/future.c

src/mongoc/mongoc-async-cmd-private.h

@@ -39,6 +39,7 @@ typedef enum
    MONGOC_ASYNC_CMD_RECV_LEN,
    MONGOC_ASYNC_CMD_RECV_RPC,
    MONGOC_ASYNC_CMD_ERROR_STATE,
+   MONGOC_ASYNC_CMD_CANCELED_STATE,
 } mongoc_async_cmd_state_t;
 
 typedef struct _mongoc_async_cmd

src/mongoc/mongoc-async-cmd.c

@@ -44,15 +44,14 @@ mongoc_async_cmd_result_t
 _mongoc_async_cmd_phase_recv_len (mongoc_async_cmd_t *cmd);
 mongoc_async_cmd_result_t
 _mongoc_async_cmd_phase_recv_rpc (mongoc_async_cmd_t *cmd);
-mongoc_async_cmd_result_t
-_mongoc_async_cmd_phase_error (mongoc_async_cmd_t *cmd);
 
 static const _monogc_async_cmd_phase_t gMongocCMDPhases[] = {
    _mongoc_async_cmd_phase_setup,
    _mongoc_async_cmd_phase_send,
    _mongoc_async_cmd_phase_recv_len,
    _mongoc_async_cmd_phase_recv_rpc,
-   _mongoc_async_cmd_phase_error,
+   NULL,  /* no callback for MONGOC_ASYNC_CMD_ERROR_STATE    */
+   NULL,  /* no callback for MONGOC_ASYNC_CMD_CANCELED_STATE */
 };
 
 #ifdef MONGOC_ENABLE_SSL
@@ -97,8 +96,14 @@ mongoc_async_cmd_run (mongoc_async_cmd_t *acmd)
 {
    mongoc_async_cmd_result_t result;
    int64_t rtt;
+   _monogc_async_cmd_phase_t phase_callback;
 
-   result = gMongocCMDPhases[acmd->state](acmd);
+   phase_callback = gMongocCMDPhases[acmd->state];
+   if (phase_callback) {
+      result = phase_callback (acmd);
+   } else {
+      result = MONGOC_ASYNC_CMD_ERROR;
+   }
 
    if (result == MONGOC_ASYNC_CMD_IN_PROGRESS) {
       return true;
@@ -109,7 +114,7 @@ mongoc_async_cmd_run (mongoc_async_cmd_t *acmd)
    if (result == MONGOC_ASYNC_CMD_SUCCESS) {
       acmd->cb (result, &acmd->reply, rtt, acmd->data, &acmd->error);
    } else {
-      /* we're either in ERROR or TIMEOUT */
+      /* we're in ERROR, TIMEOUT, or CANCELED */
       acmd->cb (result, NULL, rtt, acmd->data, &acmd->error);
    }
 
@@ -385,9 +390,3 @@ _mongoc_async_cmd_phase_recv_rpc (mongoc_async_cmd_t *acmd)
 
    return MONGOC_ASYNC_CMD_IN_PROGRESS;
 }
-
-mongoc_async_cmd_result_t
-_mongoc_async_cmd_phase_error (mongoc_async_cmd_t *acmd)
-{
-   return MONGOC_ASYNC_CMD_ERROR;
-}

src/mongoc/mongoc-async.c

@@ -92,6 +92,7 @@ mongoc_async_run (mongoc_async_t *async,
 
       DL_FOREACH_SAFE (async->cmds, acmd, tmp)
       {
+         /* async commands are sorted by expire_at */
          if (now > acmd->expire_at) {
             acmd->cb (MONGOC_ASYNC_CMD_TIMEOUT, NULL, (now - acmd->start_time), acmd->data,
                       &acmd->error);

src/mongoc/mongoc-client-private.h

@@ -69,6 +69,12 @@ mongoc_client_t *
 _mongoc_client_new_from_uri (const mongoc_uri_t *uri,
                              mongoc_topology_t      *topology);
 
+mongoc_stream_t *
+mongoc_client_default_stream_initiator (const mongoc_uri_t       *uri,
+                                        const mongoc_host_list_t *host,
+                                        void                     *user_data,
+                                        bson_error_t             *error);
+
 mongoc_stream_t *
 _mongoc_client_create_stream (mongoc_client_t          *client,
                               const mongoc_host_list_t *host,

src/mongoc/mongoc-client.c

@@ -265,7 +265,7 @@ mongoc_client_connect_unix (const mongoc_uri_t       *uri,
  *--------------------------------------------------------------------------
  */
 
-static mongoc_stream_t *
+mongoc_stream_t *
 mongoc_client_default_stream_initiator (const mongoc_uri_t       *uri,
                                         const mongoc_host_list_t *host,
                                         void                     *user_data,

src/mongoc/mongoc-cluster.c

@@ -1161,6 +1161,8 @@ _mongoc_cluster_add_node (mongoc_cluster_t *cluster,
          RETURN (NULL);
       }
 
+      BSON_ASSERT (!scanner_node->retired);
+
       stream = scanner_node->stream;
       if (!stream) {
          MONGOC_WARNING ("Failed connection to %s", sd->connection_address);
@@ -1251,6 +1253,8 @@ mongoc_cluster_fetch_stream (mongoc_cluster_t *cluster,
          goto FETCH_FAIL;
       }
 
+      BSON_ASSERT (!scanner_node->retired);
+
       stream = scanner_node->stream;
       if (!stream) {
          goto FETCH_FAIL;
@@ -2047,6 +2051,8 @@ _mongoc_cluster_check_interval (mongoc_cluster_t *cluster,
       return false;
    }
 
+   BSON_ASSERT (!scanner_node->retired);
+
    stream = scanner_node->stream;
 
    if (!stream) {

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

@@ -54,6 +54,8 @@ typedef struct mongoc_topology_scanner_node
 
    struct mongoc_topology_scanner_node *next;
    struct mongoc_topology_scanner_node *prev;
+
+   bool                            retired;
 } mongoc_topology_scanner_node_t;
 
 typedef struct mongoc_topology_scanner
@@ -84,7 +86,7 @@ mongoc_topology_scanner_new (const mongoc_uri_t          *uri,
 void
 mongoc_topology_scanner_destroy (mongoc_topology_scanner_t *ts);
 
-void
+mongoc_topology_scanner_node_t *
 mongoc_topology_scanner_add (mongoc_topology_scanner_t *ts,
                              const mongoc_host_list_t  *host,
                              uint32_t                   id);
@@ -95,6 +97,9 @@ mongoc_topology_scanner_add_and_scan (mongoc_topology_scanner_t *ts,
                                       uint32_t                   id,
                                       int64_t                    timeout_msec);
 
+void
+mongoc_topology_scanner_node_retire (mongoc_topology_scanner_node_t *node);
+
 void
 mongoc_topology_scanner_node_destroy (mongoc_topology_scanner_node_t *node,
                                       bool failed);
@@ -108,6 +113,9 @@ bool
 mongoc_topology_scanner_work (mongoc_topology_scanner_t *ts,
                               int32_t                    timeout_msec);
 
+void
+mongoc_topology_scanner_reset (mongoc_topology_scanner_t *ts);
+
 mongoc_topology_scanner_node_t *
 mongoc_topology_scanner_get_node (mongoc_topology_scanner_t *ts,
                                   uint32_t                   id);

src/mongoc/mongoc-topology-scanner.c

@@ -97,10 +97,10 @@ mongoc_topology_scanner_destroy (mongoc_topology_scanner_t *ts)
    bson_free (ts);
 }
 
-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 *
+mongoc_topology_scanner_add (mongoc_topology_scanner_t *ts,
+                             const mongoc_host_list_t  *host,
+                             uint32_t                   id)
 {
    mongoc_topology_scanner_node_t *node;
 
@@ -128,15 +128,6 @@ _add_node (mongoc_topology_scanner_t *ts,
    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,
@@ -147,7 +138,7 @@ mongoc_topology_scanner_add_and_scan (mongoc_topology_scanner_t *ts,
 
    BSON_ASSERT (timeout_msec < INT32_MAX);
 
-   node = _add_node (ts, host, id);
+   node = mongoc_topology_scanner_add (ts, host, id);
 
    if (node && mongoc_topology_scanner_node_setup (node)) {
       node->cmd = mongoc_async_cmd (
@@ -158,9 +149,20 @@ mongoc_topology_scanner_add_and_scan (mongoc_topology_scanner_t *ts,
          node, (int32_t) timeout_msec);
    }
 
+   /* if setup fails the node stays in the scanner. destroyed after the scan. */
    return;
 }
 
+void
+mongoc_topology_scanner_node_retire (mongoc_topology_scanner_node_t *node)
+{
+   if (node->cmd) {
+      node->cmd->state = MONGOC_ASYNC_CMD_CANCELED_STATE;
+   }
+
+   node->retired = true;
+}
+
 void
 mongoc_topology_scanner_node_destroy (mongoc_topology_scanner_node_t *node, bool failed)
 {
@@ -236,13 +238,19 @@ mongoc_topology_scanner_ismaster_handler (mongoc_async_cmd_result_t async_status
                                           bson_error_t             *error)
 {
    mongoc_topology_scanner_node_t *node;
-   int64_t now = bson_get_monotonic_time ();
+   int64_t now;
 
    bson_return_if_fail (data);
 
    node = (mongoc_topology_scanner_node_t *)data;
    node->cmd = NULL;
 
+   if (node->retired) {
+      return;
+   }
+
+   now = bson_get_monotonic_time ();
+
    /* if no ismaster response, async cmd had an error or timed out */
    if (!ismaster_response ||
        async_status == MONGOC_ASYNC_CMD_ERROR ||
@@ -396,6 +404,8 @@ mongoc_topology_scanner_node_setup (mongoc_topology_scanner_node_t *node)
 
    if (node->stream) { return true; }
 
+   BSON_ASSERT (!node->retired);
+
    if (node->ts->initiator) {
       sock_stream = node->ts->initiator (node->ts->uri, &node->host,
                                          node->ts->initiator_context, &error);
@@ -474,6 +484,9 @@ mongoc_topology_scanner_start (mongoc_topology_scanner_t *ts,
       /* check node if it last failed before current cooldown period began */
       if (node->last_failed < cooldown) {
          if (mongoc_topology_scanner_node_setup (node)) {
+
+            BSON_ASSERT (!node->cmd);
+
             node->cmd = mongoc_async_cmd (
                ts->async, node->stream, ts->setup,
                node->host.host, "admin",
@@ -509,14 +522,37 @@ mongoc_topology_scanner_work (mongoc_topology_scanner_t *ts,
    r = mongoc_async_run (ts->async, timeout_msec);
 
    if (! r) {
-      mongoc_host_list_destroy_all (ts->seen);
-      ts->seen = NULL;
       ts->in_progress = false;
    }
 
    return r;
 }
 
+/*
+ *--------------------------------------------------------------------------
+ *
+ * mongoc_topology_scanner_reset --
+ *
+ *      Reset "retired" nodes that failed or were removed in the previous
+ *      scan.
+ *
+ *--------------------------------------------------------------------------
+ */
+
+void
+mongoc_topology_scanner_reset (mongoc_topology_scanner_t *ts)
+{
+   mongoc_topology_scanner_node_t *node, *tmp;
+
+   DL_FOREACH_SAFE (ts->nodes, node, tmp) {
+      if (node->retired) {
+         mongoc_topology_scanner_node_destroy (node, true);
+      }
+   }
+
+   mongoc_host_list_destroy_all (ts->seen);
+   ts->seen = NULL;
+}
 
 void
 mongoc_topology_scanner_set_async_cb (mongoc_topology_scanner_t *ts,

src/mongoc/mongoc-topology.c

@@ -63,7 +63,7 @@ mongoc_topology_reconcile (mongoc_topology_t *topology) {
    /* Remove removed nodes */
    DL_FOREACH_SAFE (scanner->nodes, ele, tmp) {
       if (! mongoc_topology_description_server_by_id (description, ele->id)) {
-         mongoc_topology_scanner_node_destroy (ele, true);
+         mongoc_topology_scanner_node_retire (ele);
       }
    }
 }
@@ -326,6 +326,8 @@ _mongoc_topology_do_blocking_scan (mongoc_topology_t *topology) {
    while (_mongoc_topology_run_scanner (topology,
                                         topology->connect_timeout_msec)) {}
 
+   /* "retired" nodes can be checked again in the next scan */
+   mongoc_topology_scanner_reset (topology->scanner);
    topology->last_scan = bson_get_monotonic_time ();
    topology->stale = false;
 }
@@ -697,6 +699,10 @@ void * _mongoc_topology_run_background (void *data)
                                            topology->connect_timeout_msec)) {}
 
       mongoc_mutex_lock (&topology->mutex);
+
+      /* "retired" nodes can be checked again in the next scan */
+      mongoc_topology_scanner_reset (topology->scanner);
+
       topology->last_scan = bson_get_monotonic_time ();
       topology->scanning = false;
       mongoc_mutex_unlock (&topology->mutex);