Reap expired connections on drop

The reaper only runs against the connections in its idle pool. This is
fine for reaping idle connections, but for hotly contested connections
beyond their maximum lifetime this can prove problematic.

Consider an active connection beyond its lifetime and a reaper that runs
every 3 seconds:
- [t0] Connection is idle
- [t1] Connection is active
- [t2] Reaper runs, does not see connection
- [t3] Connection is idle

This pattern can repeat infinitely with the connection never being reaped.

By checking the max lifetime on drop, we can ensure that expired
connections are reaped in a reason amount of time (assuming they
eventually do get dropped).

Author: tneely

bb8/src/inner.rs

@@ -149,12 +149,19 @@ 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 mut locked = self.inner.internals.lock();
         match (state, self.inner.manager.has_broken(&mut conn.conn)) {
-            (ConnectionState::Present, false) => locked.put(conn, None, self.inner.clone()),
+            (ConnectionState::Present, false) if !is_expired => {
+                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

@@ -585,6 +585,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.
+    tokio::time::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()