CDRIVER-3587 do not reuse invalid server stream

Author: rcsanchez97

src/libmongoc/src/mongoc/mongoc-client.c

@@ -2962,8 +2962,15 @@ _mongoc_client_end_sessions (mongoc_client_t *client)
             }
          }
 
-         bson_destroy (&cmd);
          mongoc_cmd_parts_cleanup (&parts);
+
+         if (!stream->stream) {
+            /* The stream was invalidated as a result of a network error, so we
+             * stop sending commands. */
+            break;
+         }
+
+         bson_destroy (&cmd);
       }
 
       bson_destroy (&cmd);

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

@@ -2895,7 +2895,7 @@ mongoc_cluster_run_opmsg (mongoc_cluster_t *cluster,
    mongoc_rpc_t rpc;
    int32_t msg_len;
    bool ok;
-   const mongoc_server_stream_t *server_stream;
+   mongoc_server_stream_t *server_stream;
 
    server_stream = cmd->server_stream;
    if (!cmd->command_name) {
@@ -2974,6 +2974,7 @@ mongoc_cluster_run_opmsg (mongoc_cluster_t *cluster,
       RUN_CMD_ERR_DECORATE;
       mongoc_cluster_disconnect_node (
          cluster, server_stream->sd->id, true, error);
+      server_stream->stream = NULL;
       bson_free (output);
       network_error_reply (reply, cmd);
       _mongoc_buffer_destroy (&buffer);
@@ -2988,6 +2989,7 @@ mongoc_cluster_run_opmsg (mongoc_cluster_t *cluster,
          RUN_CMD_ERR_DECORATE;
          mongoc_cluster_disconnect_node (
             cluster, server_stream->sd->id, true, error);
+         server_stream->stream = NULL;
          bson_free (output);
          network_error_reply (reply, cmd);
          _mongoc_buffer_destroy (&buffer);
@@ -3006,6 +3008,7 @@ mongoc_cluster_run_opmsg (mongoc_cluster_t *cluster,
             server_stream->sd->max_msg_size);
          mongoc_cluster_disconnect_node (
             cluster, server_stream->sd->id, true, error);
+         server_stream->stream = NULL;
          bson_free (output);
          network_error_reply (reply, cmd);
          _mongoc_buffer_destroy (&buffer);
@@ -3021,6 +3024,7 @@ mongoc_cluster_run_opmsg (mongoc_cluster_t *cluster,
          RUN_CMD_ERR_DECORATE;
          mongoc_cluster_disconnect_node (
             cluster, server_stream->sd->id, true, error);
+         server_stream->stream = NULL;
          bson_free (output);
          network_error_reply (reply, cmd);
          _mongoc_buffer_destroy (&buffer);
@@ -3048,6 +3052,7 @@ mongoc_cluster_run_opmsg (mongoc_cluster_t *cluster,
                          "Could not decompress message from server");
             mongoc_cluster_disconnect_node (
                cluster, server_stream->sd->id, true, error);
+            server_stream->stream = NULL;
             bson_free (output);
             network_error_reply (reply, cmd);
             _mongoc_buffer_destroy (&buffer);

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

@@ -58,7 +58,7 @@ typedef struct _mongoc_cmd_t {
    const uint8_t *payload;
    int32_t payload_size;
    const char *payload_identifier;
-   const mongoc_server_stream_t *server_stream;
+   mongoc_server_stream_t *server_stream;
    int64_t operation_id;
    mongoc_client_session_t *session;
    bool is_acknowledged;
@@ -123,7 +123,7 @@ mongoc_cmd_parts_append_read_write (mongoc_cmd_parts_t *parts,
 
 bool
 mongoc_cmd_parts_assemble (mongoc_cmd_parts_t *parts,
-                           const mongoc_server_stream_t *server_stream,
+                           mongoc_server_stream_t *server_stream,
                            bson_error_t *error);
 
 bool

src/libmongoc/src/mongoc/mongoc-cmd.c

@@ -786,7 +786,7 @@ _is_retryable_read (const mongoc_cmd_parts_t *parts,
 
 bool
 mongoc_cmd_parts_assemble (mongoc_cmd_parts_t *parts,
-                           const mongoc_server_stream_t *server_stream,
+                           mongoc_server_stream_t *server_stream,
                            bson_error_t *error)
 {
    mongoc_server_description_type_t server_type;
@@ -1086,4 +1086,4 @@ _mongoc_cmd_append_payload_as_array (const mongoc_cmd_t *cmd, bson_t *out)
    }
 
    bson_append_array_end (out, &bson);
-}
+}

src/libmongoc/tests/test-mongoc-client-session.c

@@ -497,6 +497,55 @@ test_mock_end_sessions_pooled (void)
    _test_mock_end_sessions (true);
 }
 
+/* Test for CDRIVER-3587 - Do not reuse server stream that becomes invalid on
+ * failure to end session */
+static void
+test_mock_end_sessions_server_disconnect (void)
+{
+   mock_server_t *server;
+   mongoc_client_t *client;
+   bson_error_t error;
+   mongoc_client_session_t *session[12000];
+   future_t *future;
+   uint16_t i;
+
+   server = mock_mongos_new (WIRE_VERSION_OP_MSG);
+   mock_server_run (server);
+
+   client = mongoc_client_new_from_uri (mock_server_get_uri (server));
+
+   for (i = 0; i < 12000; i++) {
+      session[i] = mongoc_client_start_session (client, NULL, &error);
+      ASSERT_OR_PRINT (session[i], error);
+   }
+
+   /* Simulate server failure or network failure.  Destroy the mock server here
+    * rather than at the end of the test so that the 'endSessions' commands fail
+    * to reach the mock server. */
+   mock_server_destroy (server);
+
+   /* The below calls to mongoc_client_session_destroy () will produce a warning
+    * regarding the inability to send the 'endSessions' command. */
+   capture_logs (true);
+
+   for (i = 0; i < 12000; i++) {
+      mongoc_client_session_destroy (session[i]);
+   }
+
+   /* The above loop will add each session back to the session pool.  If
+    * CDRIVER-3587 has not been fixed, the mongoc_client_destroy () call below
+    * will create 'endSessions' commands which will be sent but fail to reach
+    * the server; the associated server stream will not be correctly
+    * invalidated.  Subsequent reuse of the stream, as in the attempt to send
+    * the second batch of 10,000 during the attempt to destroy the client, will
+    * trigger a segfault. */
+
+   future = future_client_destroy (client);
+
+   future_wait (future);
+   future_destroy (future);
+}
+
 typedef struct {
    int started_calls;
    int succeeded_calls;
@@ -2687,6 +2736,10 @@ test_session_install (TestSuite *suite)
                                 "/Session/end/mock/pooled",
                                 test_mock_end_sessions_pooled,
                                 test_framework_skip_if_no_crypto);
+   TestSuite_AddMockServerTest (suite,
+                                "/Session/end/mock/disconnected",
+                                test_mock_end_sessions_server_disconnect,
+                                test_framework_skip_if_no_crypto);
    TestSuite_AddFull (suite,
                       "/Session/end/single",
                       test_end_sessions_single,