Add connection timeout, TCP keepalive, and pool health features (#283)

* feat: add connection timeout, TCP keepalive, and pool health features

- Add connection_timeout (default 30s) wrapping TcpStream::connect and recv
- Add tcp_keepalive (default 60s) via socket2 on new connections
- Add idle_timeout and max_lifetime config options wired to deadpool
- Add 5s timeout on pool recycle() to prevent broken connection poisoning
- Add ConnectionTimedOut error variant

Fixes indefinite hangs when TCP connections become half-open (container
restarts, network interruptions), which previously exhausted the
deadpool connection pool permanently.

* test/docs: add tests and documentation for timeout/keepalive features

- Add 7 unit tests covering config defaults, custom timeouts, pool
  creation, and error display
- Update lib.rs Configurations doc section with new options
- All 290 tests pass

* style: `cargo fmt`

* fix: change function definitions to requested

* fix: update MSRV lockfiles and add missing Config fields in tests

Rebase onto upstream/main which includes MSRV pin updates (#284).
Re-ran `cargo xtask msrv min` to regenerate lockfiles and added
missing timeout/keepalive fields to Config initializers in tests.

Author: ajmeese7

ci/Cargo.lock.min

@@ -61,7 +61,7 @@ features = ["std", "serde"]
 [dependencies.deadpool]
 version = "0.12.0"       # 0.13 requires bumping MSRV to 1.85
 default-features = false
-features = ["managed"]
+features = ["managed", "rt_tokio_1"]
 
 [dependencies.rustls]
 version = "0.23.29"
@@ -77,6 +77,10 @@ version = "0.26.0"
 default-features = false
 features = ["tls12", "ring"]
 
+[dependencies.socket2]
+version = "0.5"
+features = ["all"]
+
 [dev-dependencies]
 pretty_env_logger = "0.5.0"
 serde_bytes = "0.11.0"

ci/Cargo.lock.msrv

@@ -3,10 +3,12 @@ use crate::errors::{Error, Result};
 #[cfg(feature = "unstable-bolt-protocol-impl-v2")]
 use serde::{Deserialize, Deserializer, Serialize};
 use std::path::Path;
-use std::{ops::Deref, sync::Arc};
+use std::{ops::Deref, sync::Arc, time::Duration};
 
 const DEFAULT_FETCH_SIZE: usize = 200;
 const DEFAULT_MAX_CONNECTIONS: usize = 16;
+const DEFAULT_CONNECTION_TIMEOUT: Duration = Duration::from_secs(30);
+const DEFAULT_TCP_KEEPALIVE: Option<Duration> = Some(Duration::from_secs(60));
 
 /// Newtype for the name of the database.
 /// Stores the name as an `Arc<str>` to avoid cloning the name around.
@@ -131,6 +133,14 @@ pub struct Config {
     pub(crate) fetch_size: usize,
     pub(crate) imp_user: Option<ImpersonateUser>,
     pub(crate) tls_config: ConnectionTLSConfig,
+    /// Timeout for establishing a new connection and for read operations.
+    pub(crate) connection_timeout: Duration,
+    /// TCP keepalive interval. If `Some`, TCP keepalive is enabled on the socket.
+    pub(crate) tcp_keepalive: Option<Duration>,
+    /// Maximum idle time for a connection in the pool before it is discarded.
+    pub(crate) idle_timeout: Option<Duration>,
+    /// Maximum lifetime of a connection in the pool before it is discarded.
+    pub(crate) max_lifetime: Option<Duration>,
 }
 
 impl Config {
@@ -153,6 +163,10 @@ pub struct ConfigBuilder {
     max_connections: usize,
     imp_user: Option<ImpersonateUser>,
     tls_config: ConnectionTLSConfig,
+    connection_timeout: Duration,
+    tcp_keepalive: Option<Duration>,
+    idle_timeout: Option<Duration>,
+    max_lifetime: Option<Duration>,
 }
 
 impl ConfigBuilder {
@@ -240,6 +254,39 @@ impl ConfigBuilder {
         self
     }
 
+    /// The timeout for establishing a new connection and for read operations.
+    ///
+    /// Defaults to 30 seconds if not set.
+    pub fn connection_timeout(mut self, timeout: Duration) -> Self {
+        self.connection_timeout = timeout;
+        self
+    }
+
+    /// The TCP keepalive interval. Set to `Some(duration)` to enable TCP keepalive
+    /// on the underlying socket, or `None` to disable it.
+    ///
+    /// Defaults to `Some(60 seconds)` if not set.
+    pub fn tcp_keepalive(mut self, interval: impl Into<Option<Duration>>) -> Self {
+        self.tcp_keepalive = interval.into();
+        self
+    }
+
+    /// The maximum idle time for a connection in the pool before it is discarded.
+    ///
+    /// Defaults to `None` (no idle timeout) if not set.
+    pub fn idle_timeout(mut self, timeout: impl Into<Option<Duration>>) -> Self {
+        self.idle_timeout = timeout.into();
+        self
+    }
+
+    /// The maximum lifetime of a connection in the pool before it is discarded.
+    ///
+    /// Defaults to `None` (no maximum lifetime) if not set.
+    pub fn max_lifetime(mut self, lifetime: impl Into<Option<Duration>>) -> Self {
+        self.max_lifetime = lifetime.into();
+        self
+    }
+
     pub fn build(self) -> Result<Config> {
         if let (Some(uri), Some(user), Some(password)) = (self.uri, self.user, self.password) {
             Ok(Config {
@@ -251,6 +298,10 @@ impl ConfigBuilder {
                 db: self.db,
                 imp_user: self.imp_user,
                 tls_config: self.tls_config,
+                connection_timeout: self.connection_timeout,
+                tcp_keepalive: self.tcp_keepalive,
+                idle_timeout: self.idle_timeout,
+                max_lifetime: self.max_lifetime,
             })
         } else {
             Err(Error::InvalidConfig)
@@ -269,6 +320,10 @@ impl Default for ConfigBuilder {
             imp_user: None,
             fetch_size: DEFAULT_FETCH_SIZE,
             tls_config: ConnectionTLSConfig::None,
+            connection_timeout: DEFAULT_CONNECTION_TIMEOUT,
+            tcp_keepalive: DEFAULT_TCP_KEEPALIVE,
+            idle_timeout: None,
+            max_lifetime: None,
         }
     }
 }
@@ -332,6 +387,64 @@ mod tests {
         assert_eq!(config.tls_config, ConnectionTLSConfig::NoSSLValidation);
     }
 
+    #[test]
+    fn should_build_with_timeout_defaults() {
+        let config = ConfigBuilder::default()
+            .uri("127.0.0.1:7687")
+            .user("some_user")
+            .password("some_password")
+            .build()
+            .unwrap();
+        assert_eq!(config.connection_timeout, Duration::from_secs(30));
+        assert_eq!(config.tcp_keepalive, Some(Duration::from_secs(60)));
+        assert_eq!(config.idle_timeout, None);
+        assert_eq!(config.max_lifetime, None);
+    }
+
+    #[test]
+    fn should_build_with_custom_timeouts() {
+        let config = ConfigBuilder::default()
+            .uri("127.0.0.1:7687")
+            .user("some_user")
+            .password("some_password")
+            .connection_timeout(Duration::from_secs(10))
+            .tcp_keepalive(Some(Duration::from_secs(120)))
+            .idle_timeout(Some(Duration::from_secs(300)))
+            .max_lifetime(Some(Duration::from_secs(3600)))
+            .build()
+            .unwrap();
+        assert_eq!(config.connection_timeout, Duration::from_secs(10));
+        assert_eq!(config.tcp_keepalive, Some(Duration::from_secs(120)));
+        assert_eq!(config.idle_timeout, Some(Duration::from_secs(300)));
+        assert_eq!(config.max_lifetime, Some(Duration::from_secs(3600)));
+    }
+
+    #[test]
+    fn should_disable_tcp_keepalive() {
+        let config = ConfigBuilder::default()
+            .uri("127.0.0.1:7687")
+            .user("some_user")
+            .password("some_password")
+            .tcp_keepalive(None)
+            .build()
+            .unwrap();
+        assert_eq!(config.tcp_keepalive, None);
+    }
+
+    #[test]
+    fn should_set_idle_and_max_lifetime() {
+        let config = ConfigBuilder::default()
+            .uri("127.0.0.1:7687")
+            .user("some_user")
+            .password("some_password")
+            .idle_timeout(Some(Duration::from_secs(600)))
+            .max_lifetime(Some(Duration::from_secs(1800)))
+            .build()
+            .unwrap();
+        assert_eq!(config.idle_timeout, Some(Duration::from_secs(600)));
+        assert_eq!(config.max_lifetime, Some(Duration::from_secs(1800)));
+    }
+
     #[test]
     fn should_reject_invalid_config() {
         assert!(ConfigBuilder::default()