Proactively reap idle connections past max_lifetime

src/lib.rs

@@ -325,11 +325,24 @@ impl<M: Manager> Pool<M> {
         gauge!(IDLE_CONNECTIONS).set(0.0);
 
         let (share_config, internal_config) = config.split();
+        let max_lifetime = internal_config.max_lifetime;
+        let clean_rate = share_config.clean_rate;
+
+        // Start the periodic cleaner up-front when a `max_lifetime` is
+        // configured, so idle connections are proactively reaped instead of
+        // only being checked lazily on checkout.
+        let (cleaner_ch_sender, cleaner_ch_receiver) = if max_lifetime.is_some() {
+            let (tx, rx) = mpsc::channel(1);
+            (Some(tx), Some(rx))
+        } else {
+            (None, None)
+        };
+
         let internals = Mutex::new(PoolInternals {
             config: internal_config,
             free_conns: Vec::new(),
             wait_duration: Duration::from_secs(0),
-            cleaner_ch: None,
+            cleaner_ch: cleaner_ch_sender,
         });
 
         let pool_state = PoolState {
@@ -347,6 +360,14 @@ impl<M: Manager> Pool<M> {
             state: pool_state,
         });
 
+        if let Some(cleaner_ch) = cleaner_ch_receiver {
+            log::debug!("run connection cleaner");
+            let shared1 = Arc::downgrade(&shared);
+            shared.manager.spawn_task(async move {
+                connection_cleaner(shared1, cleaner_ch, clean_rate).await;
+            });
+        }
+
         Pool(shared)
     }
 

tests/mobc.rs

@@ -573,7 +573,68 @@ fn test_max_lifetime_lazy() {
         }
         assert_eq!(5, DROPPED.load(Ordering::SeqCst));
         pool.get().await.unwrap();
+        // The connection above is returned to the pool as idle and then
+        // exceeds `max_lifetime` while sitting unused. The periodic cleaner
+        // now proactively reaps it even though nothing checks it out again.
         delay_for(Duration::from_secs(2)).await;
+        assert_eq!(6, DROPPED.load(Ordering::SeqCst));
+        Ok::<(), Error<TestError>>(())
+    })
+    .unwrap();
+}
+
+#[test]
+fn test_max_lifetime_proactive_reap() {
+    static DROPPED: AtomicUsize = AtomicUsize::new(0);
+    let mut rt: Runtime = Runtime::new().unwrap();
+
+    struct Connection;
+
+    impl Drop for Connection {
+        fn drop(&mut self) {
+            DROPPED.fetch_add(1, Ordering::SeqCst);
+        }
+    }
+
+    struct Handler;
+
+    #[async_trait]
+    impl Manager for Handler {
+        type Connection = Connection;
+        type Error = TestError;
+
+        async fn connect(&self) -> Result<Self::Connection, Self::Error> {
+            Ok(Connection)
+        }
+
+        async fn check(&self, conn: Self::Connection) -> Result<Self::Connection, Self::Error> {
+            Ok(conn)
+        }
+    }
+    let handler = Handler;
+    rt.block_on(async {
+        let pool = Pool::builder()
+            .max_open(5)
+            .max_idle(5)
+            .max_lifetime(Some(Duration::from_millis(300)))
+            .get_timeout(Some(Duration::from_secs(1)))
+            .clean_rate(Duration::from_secs(1))
+            .build(handler);
+
+        let mut v = vec![];
+        for _ in 0..5 {
+            v.push(pool.get().await.unwrap());
+        }
+        // Return all connections to the pool as idle, but never check any
+        // connection back out again.
+        drop(v);
+        assert_eq!(0, DROPPED.load(Ordering::SeqCst));
+
+        // Wait past both `max_lifetime` and `clean_rate` without performing
+        // any checkout. The periodic cleaner should proactively reap the
+        // now-expired idle connections on its own.
+        delay_for(Duration::from_secs(2)).await;
+
         assert_eq!(5, DROPPED.load(Ordering::SeqCst));
         Ok::<(), Error<TestError>>(())
     })