RUST-1712 Support User Configuration for max_connecting (#923)

Author: LuisOsta

src/client/options.rs

@@ -62,6 +62,7 @@ const URI_OPTIONS: &[&str] = &[
     "maxstalenessseconds",
     "maxpoolsize",
     "minpoolsize",
+    "maxconnecting",
     "readconcernlevel",
     "readpreference",
     "readpreferencetags",
@@ -491,6 +492,12 @@ pub struct ClientOptions {
     #[builder(default)]
     pub min_pool_size: Option<u32>,
 
+    /// The maximum number of new connections that can be created concurrently.
+    ///
+    /// If specified, this value must be greater than 0. The default is 2.
+    #[builder(default)]
+    pub max_connecting: Option<u32>,
+
     /// Specifies the default read concern for operations performed on the Client. See the
     /// ReadConcern type documentation for more details.
     #[builder(default)]
@@ -669,6 +676,8 @@ impl Serialize for ClientOptions {
 
             minpoolsize: &'a Option<u32>,
 
+            maxconnecting: &'a Option<u32>,
+
             #[serde(flatten, serialize_with = "ReadConcern::serialize_for_client_options")]
             readconcern: &'a Option<ReadConcern>,
 
@@ -711,6 +720,7 @@ impl Serialize for ClientOptions {
             maxidletimems: &self.max_idle_time,
             maxpoolsize: &self.max_pool_size,
             minpoolsize: &self.min_pool_size,
+            maxconnecting: &self.max_connecting,
             readconcern: &self.read_concern,
             replicaset: &self.repl_set_name,
             retryreads: &self.retry_reads,
@@ -802,6 +812,11 @@ pub struct ConnectionString {
     /// The default value is 0.
     pub min_pool_size: Option<u32>,
 
+    /// The maximum number of new connections that can be created concurrently.
+    ///
+    /// If specified, this value must be greater than 0. The default is 2.
+    pub max_connecting: Option<u32>,
+
     /// The amount of time that a connection can remain idle in a connection pool before being
     /// closed. A value of zero indicates that connections should not be closed due to being idle.
     ///
@@ -1285,6 +1300,7 @@ impl ClientOptions {
                 };
             }
         }
+
         Self {
             hosts: vec![],
             app_name: conn_str.app_name,
@@ -1298,6 +1314,7 @@ impl ClientOptions {
             max_pool_size: conn_str.max_pool_size,
             min_pool_size: conn_str.min_pool_size,
             max_idle_time: conn_str.max_idle_time,
+            max_connecting: conn_str.max_connecting,
             server_selection_timeout: conn_str.server_selection_timeout,
             compressors: conn_str.compressors,
             connect_timeout: conn_str.connect_timeout,
@@ -1378,6 +1395,10 @@ impl ClientOptions {
             return Err(Error::invalid_argument("cannot specify maxPoolSize=0"));
         }
 
+        if let Some(0) = self.max_connecting {
+            return Err(Error::invalid_argument("cannot specify maxConnecting=0"));
+        }
+
         if let Some(SelectionCriteria::ReadPreference(ref rp)) = self.selection_criteria {
             if let Some(max_staleness) = rp.max_staleness() {
                 verify_max_staleness(
@@ -2028,6 +2049,9 @@ impl ConnectionString {
             k @ "minpoolsize" => {
                 self.min_pool_size = Some(get_u32!(value, k));
             }
+            k @ "maxconnecting" => {
+                self.max_connecting = Some(get_u32!(value, k));
+            }
             "readconcernlevel" => {
                 self.read_concern = Some(ReadConcernLevel::from_str(value).into());
             }

src/client/options/test.rs

@@ -161,6 +161,9 @@ async fn run_test(test_file: TestFile) {
                     if let Ok(max) = json_options.get_i32("maxpoolsize") {
                         json_options.insert("maxpoolsize", Bson::Int64(max.into()));
                     }
+                    if let Ok(max_connecting) = json_options.get_i32("maxconnecting") {
+                        json_options.insert("maxconnecting", Bson::Int64(max_connecting.into()));
+                    }
 
                     options_doc = options_doc
                         .into_iter()
@@ -185,7 +188,6 @@ async fn run_test(test_file: TestFile) {
                             }
                         }
                     }
-
                     assert_eq!(options_doc, json_options, "{}", test_case.description)
                 }
 

src/cmap/options.rs

@@ -62,6 +62,11 @@ pub(crate) struct ConnectionPoolOptions {
 
     /// Whether or not the client is connecting to a MongoDB cluster through a load balancer.
     pub(crate) load_balanced: Option<bool>,
+
+    /// The maximum number of new connections that can be created concurrently.
+    ///
+    /// The default is 2.
+    pub(crate) max_connecting: Option<u32>,
 }
 
 impl ConnectionPoolOptions {
@@ -77,6 +82,7 @@ impl ConnectionPoolOptions {
             ready: None,
             load_balanced: options.load_balanced,
             credential: options.credential.clone(),
+            max_connecting: options.max_connecting,
         }
     }
 

src/cmap/worker.rs

@@ -41,7 +41,7 @@ use std::{
     time::Duration,
 };
 
-const MAX_CONNECTING: u32 = 2;
+const DEFAULT_MAX_CONNECTING: u32 = 2;
 const MAINTENACE_FREQUENCY: Duration = Duration::from_millis(500);
 
 /// A worker task that manages the shared state of the pool.
@@ -134,6 +134,9 @@ pub(crate) struct ConnectionPoolWorker {
     /// A handle used to notify SDAM that a connection establishment error happened. This will
     /// allow the server to transition to Unknown and clear the pool as necessary.
     server_updater: TopologyUpdater,
+
+    /// The maximum number of new connections that can be created concurrently.
+    max_connecting: u32,
 }
 
 impl ConnectionPoolWorker {
@@ -158,6 +161,10 @@ impl ConnectionPoolWorker {
             .as_ref()
             .and_then(|opts| opts.max_pool_size)
             .unwrap_or(DEFAULT_MAX_POOL_SIZE);
+        let max_connecting = options
+            .as_ref()
+            .and_then(|opts| opts.max_connecting)
+            .unwrap_or(DEFAULT_MAX_CONNECTING);
 
         let min_pool_size = options.as_ref().and_then(|opts| opts.min_pool_size);
 
@@ -233,6 +240,7 @@ impl ConnectionPoolWorker {
             generation_publisher,
             maintenance_frequency,
             server_updater,
+            max_connecting,
         };
 
         runtime::execute(async move {
@@ -371,7 +379,7 @@ impl ConnectionPoolWorker {
             return true;
         }
 
-        self.below_max_connections() && self.pending_connection_count < MAX_CONNECTING
+        self.below_max_connections() && self.pending_connection_count < self.max_connecting
     }
 
     fn check_out(&mut self, request: ConnectionRequest) {
@@ -612,7 +620,7 @@ impl ConnectionPoolWorker {
     fn ensure_min_connections(&mut self) {
         if let Some(min_pool_size) = self.min_pool_size {
             while self.total_connection_count < min_pool_size
-                && self.pending_connection_count < MAX_CONNECTING
+                && self.pending_connection_count < self.max_connecting
             {
                 let pending_connection = self.create_pending_connection();
                 let event_handler = self.event_emitter.clone();

src/test/spec/json/connection-monitoring-and-pooling/cmap-format/pool-checkout-custom-maxConnecting-is-enforced.json

@@ -0,0 +1,81 @@
+{
+  "version": 1,
+  "style": "integration",
+  "description": "custom maxConnecting is enforced",
+  "runOn": [
+    {
+      "minServerVersion": "4.4.0"
+    }
+  ],
+  "failPoint": {
+    "configureFailPoint": "failCommand",
+    "mode": "alwaysOn",
+    "data": {
+      "failCommands": [
+        "isMaster",
+        "hello"
+      ],
+      "closeConnection": false,
+      "blockConnection": true,
+      "blockTimeMS": 500
+    }
+  },
+  "poolOptions": {
+    "maxConnecting": 1,
+    "maxPoolSize": 2,
+    "waitQueueTimeoutMS": 5000
+  },
+  "operations": [
+    {
+      "name": "ready"
+    },
+    {
+      "name": "start",
+      "target": "thread1"
+    },
+    {
+      "name": "start",
+      "target": "thread2"
+    },
+    {
+      "name": "checkOut",
+      "thread": "thread1"
+    },
+    {
+      "name": "waitForEvent",
+      "event": "ConnectionCreated",
+      "count": 1
+    },
+    {
+      "name": "checkOut",
+      "thread": "thread2"
+    },
+    {
+      "name": "waitForEvent",
+      "event": "ConnectionReady",
+      "count": 2
+    }
+  ],
+  "events": [
+    {
+      "type": "ConnectionCreated"
+    },
+    {
+      "type": "ConnectionReady"
+    },
+    {
+      "type": "ConnectionCreated"
+    },
+    {
+      "type": "ConnectionReady"
+    }
+  ],
+  "ignore": [
+    "ConnectionCheckOutStarted",
+    "ConnectionCheckedIn",
+    "ConnectionCheckedOut",
+    "ConnectionClosed",
+    "ConnectionPoolCreated",
+    "ConnectionPoolReady"
+  ]
+}

src/test/spec/json/uri-options/connection-pool-options.json

@@ -2,15 +2,16 @@
   "tests": [
     {
       "description": "Valid connection pool options are parsed correctly",
-      "uri": "mongodb://example.com/?maxIdleTimeMS=50000&maxPoolSize=5&minPoolSize=3",
+      "uri": "mongodb://example.com/?maxIdleTimeMS=50000&maxPoolSize=5&minPoolSize=3&maxConnecting=5",
       "valid": true,
       "warning": false,
       "hosts": null,
       "auth": null,
       "options": {
         "maxIdleTimeMS": 50000,
         "maxPoolSize": 5,
-        "minPoolSize": 3
+        "minPoolSize": 3,
+        "maxConnecting": 5
       }
     },
     {