Option to set an absolute TTL for connections

* The TTL, if set, forces connections to be disconnected, even
  overriding the minimum pool size constraint.
* The TTL can be combined with a random jitter value to prevent
  all connections closing at the same time.
* This enables gradual connection migration and regular connection
  recycling when CONN_RESET on return to pool is not desired.

Cargo.toml

@@ -28,6 +28,7 @@ once_cell = "1.7.2"
 pem = "3.0"
 percent-encoding = "2.1.0"
 pin-project = "1.0.2"
+rand = "0.8.5"
 serde = "1"
 serde_json = "1"
 socket2 = "0.5.2"
@@ -74,7 +75,6 @@ optional = true
 tempfile = "3.1.0"
 socket2 = { version = "0.5.2", features = ["all"] }
 tokio = { version = "1.0", features = ["macros", "rt", "rt-multi-thread"] }
-rand = "0.8.0"
 
 [features]
 default = [

src/conn/mod.rs

@@ -107,6 +107,7 @@ struct ConnInner {
     tx_status: TxStatus,
     reset_upon_returning_to_a_pool: bool,
     opts: Opts,
+    ttl_deadline: Option<Instant>,
     last_io: Instant,
     wait_timeout: Duration,
     stmt_cache: StmtCache,
@@ -157,6 +158,7 @@ impl ConnInner {
             stmt_cache: StmtCache::new(opts.stmt_cache_size()),
             socket: opts.socket().map(Into::into),
             opts,
+            ttl_deadline: None,
             nonce: Vec::default(),
             auth_plugin: AuthPlugin::MysqlNativePassword,
             auth_switched: false,
@@ -1088,6 +1090,11 @@ impl Conn {
     /// Returns true if time since last IO exceeds `wait_timeout`
     /// (or `conn_ttl` if specified in opts).
     fn expired(&self) -> bool {
+        if let Some(deadline) = self.inner.ttl_deadline {
+            if Instant::now() > deadline {
+                return true;
+            }
+        }
         let ttl = self
             .inner
             .opts

src/conn/pool/futures/get_conn.rs

@@ -142,6 +142,8 @@ impl Future for GetConn {
 
                     return match result {
                         Ok(mut c) => {
+                            c.inner.ttl_deadline =
+                                pool.opts.pool_opts().new_connection_ttl_deadline();
                             c.inner.pool = Some(pool);
                             c.inner.reset_upon_returning_to_a_pool =
                                 self.reset_upon_returning_to_a_pool;
@@ -160,6 +162,8 @@ impl Future for GetConn {
                             self.inner = GetConnInner::Done;
 
                             let pool = self.pool_take();
+                            c.inner.ttl_deadline =
+                                pool.opts.pool_opts().new_connection_ttl_deadline();
                             c.inner.pool = Some(pool);
                             c.inner.reset_upon_returning_to_a_pool =
                                 self.reset_upon_returning_to_a_pool;

src/conn/pool/mod.rs

@@ -44,6 +44,15 @@ struct IdlingConn {
 }
 
 impl IdlingConn {
+    /// Returns true when this connection has a TTL and it elapsed.
+    fn expired(&self) -> bool {
+        self.conn
+            .inner
+            .ttl_deadline
+            .map(|t| Instant::now() > t)
+            .unwrap_or_default()
+    }
+
     /// Returns duration elapsed since this connection is idling.
     fn elapsed(&self) -> Duration {
         self.since.elapsed()
@@ -82,8 +91,11 @@ impl Exchange {
             // Spawn the Recycler.
             tokio::spawn(Recycler::new(pool_opts.clone(), inner.clone(), dropped));
 
-            // Spawn the ttl check interval if `inactive_connection_ttl` isn't `0`
-            if pool_opts.inactive_connection_ttl() > Duration::from_secs(0) {
+            // Spawn the ttl check interval if `inactive_connection_ttl` isn't `0` or
+            // connections have an absolute TTL.
+            if pool_opts.inactive_connection_ttl() > Duration::ZERO
+                || pool_opts.connection_ttl().0.is_some()
+            {
                 tokio::spawn(TtlCheckInterval::new(pool_opts, inner.clone()));
             }
         }
@@ -1012,6 +1024,41 @@ mod test {
         assert_eq!(0, waitlist.queue.len());
     }
 
+    #[tokio::test]
+    async fn check_absolute_connection_ttl() -> super::Result<()> {
+        let constraints = PoolConstraints::new(1, 3).unwrap();
+        let pool_opts = PoolOpts::default()
+            .with_constraints(constraints)
+            .with_inactive_connection_ttl(Duration::from_secs(99))
+            .with_ttl_check_interval(Duration::from_secs(1))
+            .with_connection_ttl(Some(Duration::from_secs(2)), None);
+
+        let pool = Pool::new(get_opts().pool_opts(pool_opts));
+
+        let conn_ttl0 = pool.get_conn().await?;
+        sleep(Duration::from_millis(1000)).await;
+        let conn_ttl1 = pool.get_conn().await?;
+        sleep(Duration::from_millis(1000)).await;
+        let conn_ttl2 = pool.get_conn().await?;
+
+        drop(conn_ttl0);
+        drop(conn_ttl1);
+        drop(conn_ttl2);
+        assert_eq!(ex_field!(pool, exist), 3);
+
+        sleep(Duration::from_millis(1500)).await;
+        assert_eq!(ex_field!(pool, exist), 2);
+
+        sleep(Duration::from_millis(1000)).await;
+        assert_eq!(ex_field!(pool, exist), 1);
+
+        // Go even below min pool size.
+        sleep(Duration::from_millis(1000)).await;
+        assert_eq!(ex_field!(pool, exist), 0);
+
+        Ok(())
+    }
+
     #[cfg(feature = "nightly")]
     mod bench {
         use futures_util::future::{FutureExt, TryFutureExt};

src/conn/pool/ttl_check_inerval.rs

@@ -11,6 +11,7 @@ use pin_project::pin_project;
 use tokio::time::{self, Interval};
 
 use std::{
+    collections::VecDeque,
     future::Future,
     sync::{atomic::Ordering, Arc},
 };
@@ -46,24 +47,41 @@ impl TtlCheckInterval {
 
     /// Perform the check.
     pub fn check_ttl(&self) {
-        let mut exchange = self.inner.exchange.lock().unwrap();
+        let to_be_dropped = {
+            let mut exchange = self.inner.exchange.lock().unwrap();
 
-        let num_idling = exchange.available.len();
-        let num_to_drop = num_idling.saturating_sub(self.pool_opts.constraints().min());
+            let num_to_drop = exchange
+                .available
+                .len()
+                .saturating_sub(self.pool_opts.constraints().min());
 
-        for _ in 0..num_to_drop {
-            let idling_conn = exchange.available.pop_front().unwrap();
-            if idling_conn.elapsed() > self.pool_opts.inactive_connection_ttl() {
-                assert!(idling_conn.conn.inner.pool.is_none());
-                let inner = self.inner.clone();
-                tokio::spawn(idling_conn.conn.disconnect().then(move |_| {
-                    let mut exchange = inner.exchange.lock().unwrap();
-                    exchange.exist -= 1;
-                    ok::<_, ()>(())
-                }));
-            } else {
-                exchange.available.push_back(idling_conn);
+            let mut to_be_dropped = Vec::<_>::with_capacity(exchange.available.len());
+            let mut kept_available =
+                VecDeque::<_>::with_capacity(self.pool_opts.constraints().max());
+
+            while let Some(conn) = exchange.available.pop_front() {
+                if conn.expired() {
+                    to_be_dropped.push(conn);
+                } else if to_be_dropped.len() < num_to_drop
+                    && conn.elapsed() > self.pool_opts.inactive_connection_ttl()
+                {
+                    to_be_dropped.push(conn);
+                } else {
+                    kept_available.push_back(conn);
+                }
             }
+            exchange.available = kept_available;
+            to_be_dropped
+        };
+
+        for idling_conn in to_be_dropped {
+            assert!(idling_conn.conn.inner.pool.is_none());
+            let inner = self.inner.clone();
+            tokio::spawn(idling_conn.conn.disconnect().then(move |_| {
+                let mut exchange = inner.exchange.lock().unwrap();
+                exchange.exist -= 1;
+                ok::<_, ()>(())
+            }));
         }
     }
 }

src/opts/mod.rs

@@ -16,6 +16,7 @@ pub use native_tls_opts::ClientIdentity;
 pub use rustls_opts::ClientIdentity;
 
 use percent_encoding::percent_decode;
+use rand::Rng;
 use url::{Host, Url};
 
 use std::{
@@ -26,7 +27,7 @@ use std::{
     path::Path,
     str::FromStr,
     sync::Arc,
-    time::Duration,
+    time::{Duration, Instant},
     vec,
 };
 
@@ -209,6 +210,10 @@ pub struct PoolOpts {
     constraints: PoolConstraints,
     inactive_connection_ttl: Duration,
     ttl_check_interval: Duration,
+    /// Absolute TTL after which a connection expires, if set.
+    connection_ttl: Option<Duration>,
+    /// Upper bound of random value added to `connection_ttl`, if set.
+    connection_ttl_jitter: Option<Duration>,
     reset_connection: bool,
 }
 
@@ -273,6 +278,37 @@ impl PoolOpts {
         self.reset_connection
     }
 
+    /// Sets an absolute TTL after which a connection is removed from the pool.
+    /// This may push the pool below the requested minimum pool size and is indepedent of the
+    /// idle TTL. Optionally, the TTL can be extended by a per-connection random amount upper bound
+    /// set by `jitter`.
+    /// The absolute TTL is disabled by default. Setting `jitter` without `ttl` has no effect.
+    /// Fractions of seconds are ignored.
+    pub fn with_connection_ttl(mut self, ttl: Option<Duration>, jitter: Option<Duration>) -> Self {
+        self.connection_ttl = ttl;
+        self.connection_ttl_jitter = jitter;
+        self
+    }
+
+    /// Returns the absolute TTL, if set, and the jitter bound, if set.
+    pub fn connection_ttl(&self) -> (Option<Duration>, Option<Duration>) {
+        (self.connection_ttl, self.connection_ttl_jitter)
+    }
+
+    /// Returns a new deadline that's TTL (+ random jitter) in the future.
+    pub(crate) fn new_connection_ttl_deadline(&self) -> Option<Instant> {
+        if let Some(ttl) = self.connection_ttl {
+            let jitter = if let Some(jitter) = self.connection_ttl_jitter {
+                Duration::from_secs(rand::thread_rng().gen_range(0..=jitter.as_secs()))
+            } else {
+                Duration::ZERO
+            };
+            Some(Instant::now() + ttl + jitter)
+        } else {
+            None
+        }
+    }
+
     /// Pool will recycle inactive connection if it is outside of the lower bound of the pool
     /// and if it is idling longer than this value (defaults to
     /// [`DEFAULT_INACTIVE_CONNECTION_TTL`]).
@@ -359,6 +395,8 @@ impl Default for PoolOpts {
             constraints: DEFAULT_POOL_CONSTRAINTS,
             inactive_connection_ttl: DEFAULT_INACTIVE_CONNECTION_TTL,
             ttl_check_interval: DEFAULT_TTL_CHECK_INTERVAL,
+            connection_ttl: None,
+            connection_ttl_jitter: None,
             reset_connection: true,
         }
     }