feat(core): expose acquired pool connections

sqlx-core/src/pool/inner.rs

@@ -84,6 +84,11 @@ impl<DB: Database> PoolInner<DB> {
         self.num_idle.load(Ordering::Acquire)
     }
 
+    pub(super) fn num_acquired(&self) -> u32 {
+        self.size()
+            .saturating_sub(u32::try_from(self.num_idle()).unwrap_or(u32::MAX))
+    }
+
     pub(super) fn is_closed(&self) -> bool {
         self.is_closed.load(Ordering::Acquire)
     }

sqlx-core/src/pool/mod.rs

@@ -541,6 +541,14 @@ impl<DB: Database> Pool<DB> {
         self.0.num_idle()
     }
 
+    /// Returns the number of connections currently checked out from the pool.
+    ///
+    /// This is an instantaneous snapshot. The value may change immediately as
+    /// tasks acquire or release connections.
+    pub fn num_acquired(&self) -> u32 {
+        self.0.num_acquired()
+    }
+
     /// Gets a clone of the connection options for this pool
     pub fn connect_options(&self) -> Arc<<DB::Connection as Connection>::Options> {
         self.0
@@ -581,6 +589,7 @@ impl<DB: Database> fmt::Debug for Pool<DB> {
         fmt.debug_struct("Pool")
             .field("size", &self.0.size())
             .field("num_idle", &self.0.num_idle())
+            .field("num_acquired", &self.0.num_acquired())
             .field("is_closed", &self.0.is_closed())
             .field("options", &self.0.options)
             .finish()

tests/any/pool.rs

@@ -71,6 +71,43 @@ async fn pool_should_be_returned_failed_transactions() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[sqlx_macros::test]
+async fn pool_should_report_acquired_connections() -> anyhow::Result<()> {
+    sqlx::any::install_default_drivers();
+
+    let pool = AnyPoolOptions::new()
+        .max_connections(2)
+        .connect(&dotenvy::var("DATABASE_URL")?)
+        .await?;
+
+    assert_eq!(pool.size(), 1);
+    assert_eq!(pool.num_idle(), 1);
+    assert_eq!(pool.num_acquired(), 0);
+
+    let mut conn1 = pool.acquire().await?;
+
+    assert_eq!(pool.size(), 1);
+    assert_eq!(pool.num_idle(), 0);
+    assert_eq!(pool.num_acquired(), 1);
+
+    let mut conn2 = pool.acquire().await?;
+
+    assert_eq!(pool.size(), 2);
+    assert_eq!(pool.num_idle(), 0);
+    assert_eq!(pool.num_acquired(), 2);
+
+    conn2.return_to_pool().await;
+    conn1.return_to_pool().await;
+
+    assert_eq!(pool.size(), 2);
+    assert_eq!(pool.num_idle(), 2);
+    assert_eq!(pool.num_acquired(), 0);
+
+    pool.close().await;
+
+    Ok(())
+}
+
 #[sqlx_macros::test]
 async fn test_pool_callbacks() -> anyhow::Result<()> {
     sqlx::any::install_default_drivers();

tests/sqlite/sqlcipher.rs

@@ -56,6 +56,7 @@ async fn it_encrypts() -> anyhow::Result<()> {
         .await?;
 
     fill_db(&mut conn).await?;
+    conn.close().await?;
 
     // Create another connection without key, query should fail
     let mut conn = SqliteConnectOptions::from_str(&url)?.connect().await?;
@@ -67,6 +68,8 @@ async fn it_encrypts() -> anyhow::Result<()> {
         .await
         .is_err());
 
+    conn.close().await?;
+
     Ok(())
 }
 
@@ -81,6 +84,7 @@ async fn it_can_store_and_read_encrypted_data() -> anyhow::Result<()> {
         .await?;
 
     fill_db(&mut conn).await?;
+    conn.close().await?;
 
     // Create another connection with valid key
     let mut conn = SqliteConnectOptions::from_str(&url)?
@@ -96,6 +100,8 @@ async fn it_can_store_and_read_encrypted_data() -> anyhow::Result<()> {
 
     assert!(result.len() > 0);
 
+    conn.close().await?;
+
     Ok(())
 }
 
@@ -110,6 +116,7 @@ async fn it_fails_if_password_is_incorrect() -> anyhow::Result<()> {
         .await?;
 
     fill_db(&mut conn).await?;
+    conn.close().await?;
 
     // Connection with invalid key should not allow to execute queries
     let mut conn = SqliteConnectOptions::from_str(&url)?
@@ -124,6 +131,8 @@ async fn it_fails_if_password_is_incorrect() -> anyhow::Result<()> {
         .await
         .is_err());
 
+    conn.close().await?;
+
     Ok(())
 }
 
@@ -148,6 +157,7 @@ async fn it_honors_order_of_encryption_pragmas() -> anyhow::Result<()> {
         .await?;
 
     fill_db(&mut conn).await?;
+    conn.close().await?;
 
     let mut conn = SqliteConnectOptions::from_str(&url)?
         .pragma("dummy", "pragma")
@@ -167,6 +177,8 @@ async fn it_honors_order_of_encryption_pragmas() -> anyhow::Result<()> {
 
     assert!(result.len() > 0);
 
+    conn.close().await?;
+
     Ok(())
 }
 
@@ -186,6 +198,7 @@ async fn it_allows_to_rekey_the_db() -> anyhow::Result<()> {
     query("PRAGMA rekey = new_password;")
         .execute(&mut conn)
         .await?;
+    conn.close().await?;
 
     let mut conn = SqliteConnectOptions::from_str(&url)?
         .pragma("dummy", "pragma")
@@ -201,5 +214,7 @@ async fn it_allows_to_rekey_the_db() -> anyhow::Result<()> {
 
     assert!(result.len() > 0);
 
+    conn.close().await?;
+
     Ok(())
 }