RUST-757 Remove wait_queue_timeout option (#338)

Author: patrickfreed

src/client/options/mod.rs

@@ -443,14 +443,6 @@ pub struct ClientOptions {
     #[builder(default, setter(strip_option))]
     pub tls: Option<Tls>,
 
-    /// The amount of time a thread should block while waiting to check out a connection before
-    /// returning an error. Note that if there are fewer than `max_pool_size` connections checked
-    /// out or if a connection is available in the pool, checking out a connection will not block.
-    ///
-    /// By default, threads will wait indefinitely for a connection to become available.
-    #[builder(default, setter(strip_option))]
-    pub wait_queue_timeout: Option<Duration>,
-
     /// Specifies the default write concern for operations performed on the Client. See the
     /// WriteConcern type documentation for more details.
     #[builder(default, setter(strip_option))]
@@ -689,7 +681,6 @@ impl From<ClientOptionsParser> for ClientOptions {
             max_pool_size: parser.max_pool_size,
             min_pool_size: parser.min_pool_size,
             max_idle_time: parser.max_idle_time,
-            wait_queue_timeout: parser.wait_queue_timeout,
             server_selection_timeout: parser.server_selection_timeout,
             compressors: parser.compressors,
             connect_timeout: parser.connect_timeout,
@@ -723,8 +714,8 @@ impl ClientOptions {
         }
     }
 
-    /// Parses a MongoDB connection string into a ClientOptions struct. If the string is malformed
-    /// or one of the options has an invalid value, an error will be returned.
+    /// Parses a MongoDB connection string into a [`ClientOptions`] struct. If the string is
+    /// malformed or one of the options has an invalid value, an error will be returned.
     ///
     /// In the case that "mongodb+srv" is used, SRV and TXT record lookups will be done as
     /// part of this method.
@@ -759,7 +750,7 @@ impl ClientOptions {
     ///   * `retryWrites`: not yet implemented
     ///   * `retryReads`: maps to the `retry_reads` field
     ///   * `serverSelectionTimeoutMS`: maps to the `server_selection_timeout` field
-    ///   * `socketTimeoutMS`: maps to the `socket_timeout` field
+    ///   * `socketTimeoutMS`: unsupported, does not map to any field
     ///   * `ssl`: an alias of the `tls` option
     ///   * `tls`: maps to the TLS variant of the `tls` field`.
     ///   * `tlsInsecure`: relaxes the TLS constraints on connections being made; currently is just
@@ -770,7 +761,7 @@ impl ClientOptions {
     ///   * `tlsCAFile`: maps to the `ca_file_path` field of the `tls` field
     ///   * `tlsCertificateKeyFile`: maps to the `cert_key_file_path` field of the `tls` field
     ///   * `w`: maps to the `w` field of the `write_concern` field
-    ///   * `waitQueueTimeoutMS`: maps to the `wait_queue_timeout` field
+    ///   * `waitQueueTimeoutMS`: unsupported, does not map to any field
     ///   * `wTimeoutMS`: maps to the `w_timeout` field of the `write_concern` field
     ///   * `zlibCompressionLevel`: not yet implemented
     ///
@@ -939,7 +930,6 @@ impl ClientOptions {
                 server_selection_timeout,
                 socket_timeout,
                 tls,
-                wait_queue_timeout,
                 write_concern,
                 zlib_compression,
                 original_srv_hostname,

src/cmap/connection_requester.rs

@@ -1,8 +1,7 @@
 use tokio::sync::{mpsc, oneshot};
 
 use super::{worker::PoolWorkerHandle, Connection};
-use crate::{error::Result, options::StreamAddress, runtime::AsyncJoinHandle, RUNTIME};
-use std::time::Duration;
+use crate::{error::Result, options::StreamAddress, runtime::AsyncJoinHandle};
 
 /// Returns a new requester/receiver pair.
 pub(super) fn channel(
@@ -32,28 +31,16 @@ pub(super) struct ConnectionRequester {
 
 impl ConnectionRequester {
     /// Request a connection from the pool that owns the receiver end of this requester.
-    /// Returns None if it takes longer than wait_queue_timeout before the pool returns a result.
-    pub(super) async fn request(
-        &self,
-        wait_queue_timeout: Option<Duration>,
-    ) -> Option<ConnectionRequestResult> {
+    pub(super) async fn request(&self) -> ConnectionRequestResult {
         let (sender, receiver) = oneshot::channel();
 
         // this only errors if the receiver end is dropped, which can't happen because
         // we own a handle to the worker, keeping it alive.
         self.sender.send(sender).unwrap();
 
-        match wait_queue_timeout {
-            Some(timeout) => RUNTIME
-                .timeout(timeout, receiver)
-                .await
-                .map(|r| r.unwrap()) // see comment below as to why this is safe
-                .ok(),
-
-            // similarly, the receiver only returns an error if the sender is dropped, which
-            // can't happen due to the handle.
-            None => Some(receiver.await.unwrap()),
-        }
+        // similarly, the receiver only returns an error if the sender is dropped, which
+        // can't happen due to the handle.
+        receiver.await.unwrap()
     }
 }
 

src/cmap/mod.rs

@@ -9,7 +9,7 @@ pub(crate) mod options;
 mod status;
 mod worker;
 
-use std::{sync::Arc, time::Duration};
+use std::sync::Arc;
 
 use derivative::Derivative;
 
@@ -21,7 +21,7 @@ pub(crate) use self::{
 };
 use self::{connection_requester::ConnectionRequestResult, options::ConnectionPoolOptions};
 use crate::{
-    error::{Error, ErrorKind, Result},
+    error::{Error, Result},
     event::cmap::{
         CmapEventHandler,
         ConnectionCheckoutFailedEvent,
@@ -52,8 +52,6 @@ pub(crate) struct ConnectionPool {
     connection_requester: ConnectionRequester,
     generation_subscriber: PoolGenerationSubscriber,
 
-    wait_queue_timeout: Option<Duration>,
-
     #[derivative(Debug = "ignore")]
     event_handler: Option<Arc<dyn CmapEventHandler>>,
 }
@@ -73,7 +71,6 @@ impl ConnectionPool {
         );
 
         let event_handler = options.as_ref().and_then(|opts| opts.event_handler.clone());
-        let wait_queue_timeout = options.as_ref().and_then(|opts| opts.wait_queue_timeout);
 
         if let Some(ref handler) = event_handler {
             handler.handle_pool_created_event(PoolCreatedEvent {
@@ -87,7 +84,6 @@ impl ConnectionPool {
             manager,
             connection_requester,
             generation_subscriber,
-            wait_queue_timeout,
             event_handler,
         }
     }
@@ -104,7 +100,6 @@ impl ConnectionPool {
             manager,
             connection_requester,
             generation_subscriber,
-            wait_queue_timeout: None,
             event_handler: None,
         }
     }
@@ -120,9 +115,7 @@ impl ConnectionPool {
 
     /// Checks out a connection from the pool. This method will yield until this thread is at the
     /// front of the wait queue, and then will block again if no available connections are in the
-    /// pool and the total number of connections is not less than the max pool size. If the method
-    /// blocks for longer than `wait_queue_timeout` waiting for an available connection or to
-    /// start establishing a new one, a `WaitQueueTimeoutError` will be returned.
+    /// pool and the total number of connections is not less than the max pool size.
     pub(crate) async fn check_out(&self) -> Result<Connection> {
         self.emit_event(|handler| {
             let event = ConnectionCheckoutStartedEvent {
@@ -132,21 +125,12 @@ impl ConnectionPool {
             handler.handle_connection_checkout_started_event(event);
         });
 
-        let response = self
-            .connection_requester
-            .request(self.wait_queue_timeout)
-            .await;
+        let response = self.connection_requester.request().await;
 
         let conn = match response {
-            Some(ConnectionRequestResult::Pooled(c)) => Ok(c),
-            Some(ConnectionRequestResult::Establishing(task)) => task.await,
-            Some(ConnectionRequestResult::PoolCleared) => {
-                Err(Error::pool_cleared_error(&self.address))
-            }
-            None => Err(ErrorKind::WaitQueueTimeout {
-                address: self.address.clone(),
-            }
-            .into()),
+            ConnectionRequestResult::Pooled(c) => Ok(c),
+            ConnectionRequestResult::Establishing(task) => task.await,
+            ConnectionRequestResult::PoolCleared => Err(Error::pool_cleared_error(&self.address)),
         };
 
         match conn {
@@ -155,17 +139,11 @@ impl ConnectionPool {
                     handler.handle_connection_checked_out_event(conn.checked_out_event());
                 });
             }
-            Err(ref e) => {
-                let failure_reason = if let ErrorKind::WaitQueueTimeout { .. } = *e.kind {
-                    ConnectionCheckoutFailedReason::Timeout
-                } else {
-                    ConnectionCheckoutFailedReason::ConnectionError
-                };
-
+            Err(_) => {
                 self.emit_event(|handler| {
                     handler.handle_connection_checkout_failed_event(ConnectionCheckoutFailedEvent {
                         address: self.address.clone(),
-                        reason: failure_reason,
+                        reason: ConnectionCheckoutFailedReason::ConnectionError,
                     })
                 });
             }

src/cmap/options.rs

@@ -85,15 +85,6 @@ pub(crate) struct ConnectionPoolOptions {
     /// The default is not to use TLS for connections.
     #[serde(skip)]
     pub(crate) tls_options: Option<TlsOptions>,
-
-    /// Rather than wait indefinitely for a connection to become available, instead return an error
-    /// after the given duration.
-    ///
-    /// The default is to block indefinitely until a connection becomes available.
-    #[serde(rename = "waitQueueTimeoutMS")]
-    #[serde(default)]
-    #[serde(deserialize_with = "bson_util::deserialize_duration_from_u64_millis")]
-    pub(crate) wait_queue_timeout: Option<Duration>,
 }
 
 impl ConnectionPoolOptions {
@@ -107,7 +98,6 @@ impl ConnectionPoolOptions {
             max_pool_size: options.max_pool_size,
             server_api: options.server_api.clone(),
             tls_options: options.tls_options(),
-            wait_queue_timeout: options.wait_queue_timeout,
             credential: options.credential.clone(),
             event_handler: options.cmap_event_handler.clone(),
             #[cfg(test)]
@@ -127,7 +117,6 @@ impl ConnectionPoolOptions {
             max_pool_size: self.max_pool_size,
             server_api: self.server_api.clone(),
             tls_options: self.tls_options.clone(),
-            wait_queue_timeout: self.wait_queue_timeout,
         }
     }
 }

src/cmap/test/file.rs

@@ -3,12 +3,7 @@ use std::{sync::Arc, time::Duration};
 use serde::Deserialize;
 
 use super::{event::Event, State};
-use crate::{
-    bson_util,
-    cmap::options::ConnectionPoolOptions,
-    error::{ErrorKind, Result},
-    test::RunOn,
-};
+use crate::{bson_util, cmap::options::ConnectionPoolOptions, error::Result, test::RunOn};
 use bson::Document;
 
 #[derive(Debug, Deserialize)]
@@ -94,16 +89,3 @@ pub struct Error {
     message: String,
     address: Option<String>,
 }
-
-impl Error {
-    pub fn assert_matches(&self, error: &crate::error::Error, description: &str) {
-        match error.kind.as_ref() {
-            ErrorKind::WaitQueueTimeout { .. } => {
-                assert_eq!(self.type_, "WaitQueueTimeoutError", "{}", description);
-            }
-            _ => {
-                panic!("Expected {}, but got {:?}", self.type_, error);
-            }
-        }
-    }
-}

src/cmap/test/mod.rs

@@ -34,6 +34,9 @@ use bson::doc;
 const TEST_DESCRIPTIONS_TO_SKIP: &[&str] = &[
     "must destroy checked in connection if pool has been closed",
     "must throw error if checkOut is called on a closed pool",
+    // WaitQueueTimeoutMS is not supported
+    "must aggressively timeout threads enqueued longer than waitQueueTimeoutMS",
+    "waiting on maxConnecting is limited by WaitQueueTimeoutMS",
 ];
 
 /// Many different types of CMAP events are emitted from tasks spawned in the drop
@@ -185,17 +188,14 @@ impl Executor {
         }
 
         match (self.error, error) {
-            (Some(ref expected), Some(ref actual)) => {
-                expected.assert_matches(actual, self.description.as_str())
-            }
             (Some(ref expected), None) => {
                 panic!("Expected {}, but no error occurred", expected.type_)
             }
             (None, Some(ref actual)) => panic!(
                 "Expected no error to occur, but the following error was returned: {:?}",
                 actual
             ),
-            (None, None) => {}
+            (None, None) | (Some(_), Some(_)) => {}
         }
 
         let ignored_event_names = self.ignored_event_names;
@@ -354,9 +354,6 @@ impl Matchable for EventOptions {
             && self.max_pool_size.matches(&expected.max_pool_size)
             && self.min_pool_size.matches(&expected.min_pool_size)
             && self.tls_options.matches(&expected.tls_options)
-            && self
-                .wait_queue_timeout
-                .matches(&expected.wait_queue_timeout)
     }
 }
 

src/error.rs

@@ -392,13 +392,6 @@ pub enum ErrorKind {
     #[non_exhaustive]
     InvalidTlsConfig { message: String },
 
-    /// The Client timed out while checking out a connection from connection pool.
-    #[error(
-        "Timed out while checking out a connection from connection pool with address {address}"
-    )]
-    #[non_exhaustive]
-    WaitQueueTimeout { address: StreamAddress },
-
     /// An error occurred when trying to execute a write operation
     #[error("An error occurred when trying to execute a write operation: {0:?}")]
     Write(WriteFailure),

src/event/cmap.rs

@@ -88,15 +88,6 @@ pub struct ConnectionPoolOptions {
     /// The default is not to use TLS for connections.
     #[serde(skip)]
     pub tls_options: Option<TlsOptions>,
-
-    /// Rather than wait indefinitely for a connection to become available, instead return an error
-    /// after the given duration.
-    ///
-    /// The default is to block indefinitely until a connection becomes available.
-    #[serde(rename = "waitQueueTimeoutMS")]
-    #[serde(default)]
-    #[serde(deserialize_with = "crate::bson_util::deserialize_duration_from_u64_millis")]
-    pub wait_queue_timeout: Option<Duration>,
 }
 
 /// Event emitted when a connection pool becomes ready.