Reap expired connections on drop

bb8/src/inner.rs

@@ -149,12 +149,18 @@ where
             "handled in caller"
         );
 
+        let max_lifetime = self.inner.statics.max_lifetime;
+        let is_expired = max_lifetime.map_or(false, |lt| conn.is_expired(Instant::now() - lt));
+        let is_broken = self.inner.manager.has_broken(&mut conn.conn);
+
         let mut locked = self.inner.internals.lock();
-        match (state, self.inner.manager.has_broken(&mut conn.conn)) {
+        match (state, is_broken || is_expired) {
             (ConnectionState::Present, false) => locked.put(conn, None, self.inner.clone()),
-            (_, is_broken) => {
+            _ => {
                 if is_broken {
                     self.inner.statistics.record(StatsKind::ClosedBroken);
+                } else if is_expired {
+                    self.inner.statistics.record_connections_reaped(0, 1);
                 }
                 let approvals = locked.dropped(1, &self.inner.statics);
                 self.spawn_replenishing_approvals(approvals);

bb8/src/internals.rs

@@ -169,7 +169,7 @@ where
                 }
             }
             if let Some(lifetime) = config.max_lifetime {
-                if now - conn.conn.birth >= lifetime {
+                if conn.conn.is_expired(now - lifetime) {
                     closed_max_lifetime += 1;
                     keep &= false;
                 }
@@ -252,6 +252,10 @@ impl<C: Send> Conn<C> {
             birth: Instant::now(),
         }
     }
+
+    pub(crate) fn is_expired(&self, cutoff: Instant) -> bool {
+        self.birth <= cutoff
+    }
 }
 
 impl<C: Send> From<IdleConn<C>> for Conn<C> {
@@ -357,3 +361,28 @@ pub(crate) enum StatsKind {
     ClosedBroken,
     ClosedInvalid,
 }
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use crate::internals::Conn;
+
+    #[test]
+    fn test_conn_is_expired() {
+        let conn = Conn::new(0);
+
+        assert!(
+            conn.is_expired(conn.birth),
+            "conn is expired for same cuttoff"
+        );
+        assert!(
+            !conn.is_expired(conn.birth - Duration::from_nanos(1)),
+            "conn is not expired for earlier cuttoff"
+        );
+        assert!(
+            conn.is_expired(conn.birth + Duration::from_nanos(1)),
+            "conn is expired for later cuttoff"
+        );
+    }
+}

bb8/tests/test.rs

@@ -1,4 +1,5 @@
 use bb8::*;
+use tokio::time::sleep;
 
 use std::future::Future;
 use std::marker::PhantomData;
@@ -585,6 +586,40 @@ async fn test_max_lifetime() {
     assert_eq!(pool.state().statistics.connections_closed_max_lifetime, 5);
 }
 
+#[tokio::test]
+async fn test_max_lifetime_reap_on_drop() {
+    static DROPPED: AtomicUsize = AtomicUsize::new(0);
+
+    #[derive(Default)]
+    struct Connection;
+
+    impl Drop for Connection {
+        fn drop(&mut self) {
+            DROPPED.fetch_add(1, Ordering::SeqCst);
+        }
+    }
+
+    let manager = OkManager::<Connection>::new();
+    let pool = Pool::builder()
+        .max_lifetime(Some(Duration::from_secs(1)))
+        .connection_timeout(Duration::from_secs(1))
+        .reaper_rate(Duration::from_secs(999))
+        .build(manager)
+        .await
+        .unwrap();
+
+    let conn = pool.get().await;
+
+    // And wait.
+    sleep(Duration::from_secs(2)).await;
+    assert_eq!(DROPPED.load(Ordering::SeqCst), 0);
+
+    // Connection is reaped on drop.
+    drop(conn);
+    assert_eq!(DROPPED.load(Ordering::SeqCst), 1);
+    assert_eq!(pool.state().statistics.connections_closed_max_lifetime, 1);
+}
+
 #[tokio::test]
 async fn test_min_idle() {
     let pool = Pool::builder()