From d684d2adcefd26df1d50e46731075ea247c1679b Mon Sep 17 00:00:00 2001 From: Mustafa Senoglu Date: Sun, 9 Aug 2026 21:34:15 +0300 Subject: [PATCH 1/2] feat(pool): add connect_timeout option for spawned connection tasks When connect_timeout is set, Pool::acquire() spawns the connection attempt as a separate task instead of running it inline. This ensures that if acquire() is cancelled or times out, the connection attempt continues in the background. If it succeeds, the connection is returned to the pool's idle queue. This addresses the issue where acquire() cancellation would abort in-flight connection attempts, causing connection churn under high contention (see #3315, #3132, #2848). Closes #3513 --- sqlx-core/src/pool/inner.rs | 32 +++++++++++++++++++++++++++++++- sqlx-core/src/pool/options.rs | 30 +++++++++++++++++++++++++++++- 2 files changed, 60 insertions(+), 2 deletions(-) diff --git a/sqlx-core/src/pool/inner.rs b/sqlx-core/src/pool/inner.rs index b698dc9df0..6c6a869aaf 100644 --- a/sqlx-core/src/pool/inner.rs +++ b/sqlx-core/src/pool/inner.rs @@ -289,7 +289,37 @@ impl PoolInner { }; // Attempt to connect... - return self.connect(deadline, guard).await; + // + // If `connect_timeout` is set, spawn the connection as a separate task + // so that cancellation of `acquire()` doesn't abort the connection attempt. + // If the connection succeeds but no one picks it up, it's returned to the pool. + if let Some(connect_timeout) = self.options.connect_timeout { + let connect_deadline = acquire_started_at + self.options.acquire_timeout + connect_timeout; + let pool = (*self).clone(); + + let (tx, rx) = futures_intrusive::channel::oneshot_channel(); + + crate::rt::spawn(async move { + let result = pool.connect(connect_deadline, guard).await; + + // If `acquire()` is still waiting, send the result. + // If `acquire()` was cancelled (rx dropped), send() will fail + // and we return the connection to the pool. + match tx.send(result) { + Ok(()) => {} + Err(result) => { + // acquire() was cancelled, return connection to idle queue + if let Ok(live) = result { + pool.release(live); + } + } + } + }); + + return rx.receive().await.map_err(|_| Error::PoolTimedOut)?; + } else { + return self.connect(deadline, guard).await; + } } } ) diff --git a/sqlx-core/src/pool/options.rs b/sqlx-core/src/pool/options.rs index 3d048f1795..40beb4ff93 100644 --- a/sqlx-core/src/pool/options.rs +++ b/sqlx-core/src/pool/options.rs @@ -83,6 +83,7 @@ pub struct PoolOptions { pub(crate) max_lifetime: Option, pub(crate) idle_timeout: Option, pub(crate) fair: bool, + pub(crate) connect_timeout: Option, pub(crate) parent_pool: Option>, } @@ -106,6 +107,7 @@ impl Clone for PoolOptions { max_lifetime: self.max_lifetime, idle_timeout: self.idle_timeout, fair: self.fair, + connect_timeout: self.connect_timeout, parent_pool: self.parent_pool.clone(), } } @@ -161,6 +163,7 @@ impl PoolOptions { idle_timeout: Some(Duration::from_secs(10 * 60)), max_lifetime: Some(Duration::from_secs(30 * 60)), fair: true, + connect_timeout: None, parent_pool: None, } } @@ -307,6 +310,30 @@ impl PoolOptions { self.idle_timeout } + /// Set a separate timeout for establishing new connections. + /// + /// When set, this timeout is used specifically for the connection establishment phase + /// of [`Pool::acquire()`], separate from [`acquire_timeout`][Self::acquire_timeout]. + /// This allows the connection to have its own timeout that doesn't share the budget + /// with semaphore acquisition and idle connection checks. + /// + /// If not set, the connection establishment uses the remaining time from `acquire_timeout`. + /// + /// This is useful when: + /// * You want a longer timeout for establishing connections than for acquiring from the pool. + /// * You want connection attempts to survive cancellation of the `acquire()` future. + /// When `acquire()` is cancelled, the spawned connection task continues in the background. + /// If it succeeds, the connection is returned to the pool's idle queue. + pub fn connect_timeout(mut self, timeout: Duration) -> Self { + self.connect_timeout = Some(timeout); + self + } + + /// Get the connect timeout, if set. + pub fn get_connect_timeout(&self) -> Option { + self.connect_timeout + } + /// If true, the health of a connection will be verified by a call to [`Connection::ping`] /// before returning the connection. /// @@ -587,9 +614,10 @@ impl Debug for PoolOptions { f.debug_struct("PoolOptions") .field("max_connections", &self.max_connections) .field("min_connections", &self.min_connections) - .field("connect_timeout", &self.acquire_timeout) + .field("acquire_timeout", &self.acquire_timeout) .field("max_lifetime", &self.max_lifetime) .field("idle_timeout", &self.idle_timeout) + .field("connect_timeout", &self.connect_timeout) .field("test_before_acquire", &self.test_before_acquire) .finish() } From bc4c076a830c450e788b1814d4770dbafa51590b Mon Sep 17 00:00:00 2001 From: mmustafasenoglu Date: Sun, 9 Aug 2026 22:15:48 +0300 Subject: [PATCH 2/2] fix: use correct futures_intrusive API for oneshot channel - Import path: futures_intrusive::channel::shared::oneshot_channel (not ::channel::oneshot_channel) - ChannelSendError wrapping: destructure with Err(ChannelSendError(result)) - rx.receive().await returns Option, use ok_or(Error::PoolTimedOut) Fixes CI failures in sqlx #4369. --- sqlx-core/src/pool/inner.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/sqlx-core/src/pool/inner.rs b/sqlx-core/src/pool/inner.rs index 6c6a869aaf..1e42721b27 100644 --- a/sqlx-core/src/pool/inner.rs +++ b/sqlx-core/src/pool/inner.rs @@ -297,7 +297,7 @@ impl PoolInner { let connect_deadline = acquire_started_at + self.options.acquire_timeout + connect_timeout; let pool = (*self).clone(); - let (tx, rx) = futures_intrusive::channel::oneshot_channel(); + let (tx, rx) = futures_intrusive::channel::shared::oneshot_channel(); crate::rt::spawn(async move { let result = pool.connect(connect_deadline, guard).await; @@ -307,7 +307,7 @@ impl PoolInner { // and we return the connection to the pool. match tx.send(result) { Ok(()) => {} - Err(result) => { + Err(futures_intrusive::channel::ChannelSendError(result)) => { // acquire() was cancelled, return connection to idle queue if let Ok(live) = result { pool.release(live); @@ -316,7 +316,7 @@ impl PoolInner { } }); - return rx.receive().await.map_err(|_| Error::PoolTimedOut)?; + return rx.receive().await.ok_or(Error::PoolTimedOut)?; } else { return self.connect(deadline, guard).await; }