WIP pool changes

Cargo.toml

@@ -35,8 +35,7 @@ authors = [
     "Chloe Ross <[email protected]>",
     "Daniel Akhterov <[email protected]>",
 ]
-# TODO: enable this for 0.9.0
-# rust-version = "1.80.0"
+rust-version = "1.80.0"
 
 [package]
 name = "sqlx"
@@ -48,6 +47,7 @@ license.workspace = true
 edition.workspace = true
 authors.workspace = true
 repository.workspace = true
+rust-version.workspace = true
 
 [package.metadata.docs.rs]
 features = ["all-databases", "_unstable-all-types"]
@@ -147,6 +147,7 @@ uuid = "1.1.2"
 
 # Common utility crates
 dotenvy = { version = "0.15.0", default-features = false }
+ease-off = "0.1.4"
 
 # Runtimes
 [workspace.dependencies.async-std]

rust-toolchain.toml

@@ -1,4 +1,4 @@
 # Note: should NOT increase during a minor/patch release cycle
 [toolchain]
-channel = "1.78"
+channel = "1.80"
 profile = "minimal"

sqlx-core/Cargo.toml

@@ -19,8 +19,8 @@ any = []
 json = ["serde", "serde_json"]
 
 # for conditional compilation
-_rt-async-std = ["async-std", "async-io"]
-_rt-tokio = ["tokio", "tokio-stream"]
+_rt-async-std = ["async-std", "async-io", "ease-off/async-io-2"]
+_rt-tokio = ["tokio", "tokio-stream", "ease-off/tokio"]
 _tls-native-tls = ["native-tls"]
 _tls-rustls-aws-lc-rs = ["_tls-rustls", "rustls/aws-lc-rs"]
 _tls-rustls-ring = ["_tls-rustls", "rustls/ring"]
@@ -59,7 +59,6 @@ crossbeam-queue = "0.3.2"
 either = "1.6.1"
 futures-core = { version = "0.3.19", default-features = false }
 futures-io = "0.3.24"
-futures-intrusive = "0.5.0"
 futures-util = { version = "0.3.19", default-features = false, features = ["alloc", "sink", "io"] }
 log = { version = "0.4.18", default-features = false }
 memchr = { version = "2.4.1", default-features = false }
@@ -81,6 +80,9 @@ indexmap = "2.0"
 event-listener = "5.2.0"
 hashbrown = "0.14.5"
 
+ease-off = { workspace = true, features = ["futures"] }
+pin-project-lite = "0.2.14"
+
 [dev-dependencies]
 sqlx = { workspace = true, features = ["postgres", "sqlite", "mysql", "migrate", "macros", "time", "uuid"] }
 tokio = { version = "1", features = ["rt"] }

sqlx-core/src/error.rs

@@ -11,6 +11,9 @@ use crate::database::Database;
 use crate::type_info::TypeInfo;
 use crate::types::Type;
 
+#[cfg(doc)]
+use crate::pool::{PoolConnector, PoolOptions};
+
 /// A specialized `Result` type for SQLx.
 pub type Result<T, E = Error> = ::std::result::Result<T, E>;
 
@@ -104,6 +107,19 @@ pub enum Error {
     #[error("attempted to acquire a connection on a closed pool")]
     PoolClosed,
 
+    /// A custom error that may be returned from a [`PoolConnector`] implementation.
+    #[error("error returned from pool connector")]
+    PoolConnector {
+        #[source]
+        source: BoxDynError,
+
+        /// If `true`, `PoolConnector::connect()` is called again in an exponential backoff loop
+        /// up to [`PoolOptions::connect_timeout`].
+        ///
+        /// See [`PoolConnector::connect()`] for details.
+        retryable: bool,
+    },
+
     /// A background worker has crashed.
     #[error("attempted to communicate with a crashed background worker")]
     WorkerCrashed,
@@ -202,11 +218,6 @@ pub trait DatabaseError: 'static + Send + Sync + StdError {
     #[doc(hidden)]
     fn into_error(self: Box<Self>) -> Box<dyn StdError + Send + Sync + 'static>;
 
-    #[doc(hidden)]
-    fn is_transient_in_connect_phase(&self) -> bool {
-        false
-    }
-
     /// Returns the name of the constraint that triggered the error, if applicable.
     /// If the error was caused by a conflict of a unique index, this will be the index name.
     ///
@@ -244,6 +255,24 @@ pub trait DatabaseError: 'static + Send + Sync + StdError {
     fn is_check_violation(&self) -> bool {
         matches!(self.kind(), ErrorKind::CheckViolation)
     }
+
+    /// Returns `true` if this error can be retried when connecting to the database.
+    ///
+    /// Defaults to `false`.
+    ///
+    /// For example, the Postgres driver overrides this to return `true` for the following error codes:
+    ///
+    /// * `53300 too_many_connections`: returned when the maximum connections are exceeded
+    /// on the server. Assumed to be the result of a temporary overcommit
+    /// (e.g. an extra application replica being spun up to replace one that is going down).
+    ///   * This error being consistently logged or returned is a likely indicator of a misconfiguration;
+    ///     the sum of [`PoolOptions::max_connections`] for all replicas should not exceed
+    ///     the maximum connections allowed by the server.
+    /// * `57P03 cannot_connect_now`: returned when the database server is still starting up
+    /// and the tcop component is not ready to accept connections yet.
+    fn is_retryable_connect_error(&self) -> bool {
+        false
+    }
 }
 
 impl dyn DatabaseError {