Feat: add task to release unused db conns (#1640)

Feat: add task to release unused db conns

Cargo.lock

@@ -3006,6 +3006,7 @@ dependencies = [
 name = "syncstorage-spanner"
 version = "0.17.15"
 dependencies = [
+ "actix-web",
  "async-trait",
  "backtrace",
  "cadence",

syncserver/src/server/mod.rs

@@ -273,6 +273,13 @@ impl Server {
             &Metrics::from(&metrics),
             blocking_threadpool.clone(),
         )?;
+        // Spawns sweeper that calls Deadpool `retain` method, clearing unused connections.
+        db_pool.spawn_sweeper(Duration::from_secs(
+            settings
+                .syncstorage
+                .database_pool_sweeper_task_interval
+                .into(),
+        ));
         let glean_logger = Arc::new(GleanEventsLogger {
             // app_id corresponds to probe-scraper entry.
             // https://github.com/mozilla/probe-scraper/blob/main/repositories.yaml

syncstorage-mysql/src/pool.rs

@@ -99,6 +99,13 @@ impl MysqlDbPool {
         })
     }
 
+    /// Spawn a task to periodically evict idle connections. Calls wrapper sweeper fn
+    ///  to use pool.retain, retaining objects only if they are shorter in duration than
+    ///  defined max_idle. Noop for mysql impl.
+    pub fn spawn_sweeper(&self, _interval: Duration) {
+        sweeper()
+    }
+
     pub fn get_sync(&self) -> DbResult<MysqlDb> {
         Ok(MysqlDb::new(
             self.pool.get()?,
@@ -110,6 +117,13 @@ impl MysqlDbPool {
     }
 }
 
+/// Sweeper to retain only the objects specified within the closure.
+/// In this context, if a Spanner connection is unutilized, we want it
+/// to release the given connections.
+/// See: https://docs.rs/deadpool/latest/deadpool/managed/struct.Pool.html#method.retain
+/// Noop for mysql impl
+fn sweeper() {}
+
 #[async_trait]
 impl DbPool for MysqlDbPool {
     type Error = DbError;

syncstorage-settings/src/lib.rs

@@ -78,6 +78,8 @@ pub struct Settings {
     pub database_pool_connection_lifespan: Option<u32>,
     /// Max time a connection should sit idle before being dropped.
     pub database_pool_connection_max_idle: Option<u32>,
+    /// Interval for sweeper task releasing unused connections.
+    pub database_pool_sweeper_task_interval: u32,
     #[cfg(debug_assertions)]
     pub database_use_test_transactions: bool,
     #[cfg(debug_assertions)]
@@ -115,6 +117,7 @@ impl Default for Settings {
             database_pool_min_idle: None,
             database_pool_connection_lifespan: None,
             database_pool_connection_max_idle: None,
+            database_pool_sweeper_task_interval: 30,
             database_pool_connection_timeout: Some(30),
             #[cfg(debug_assertions)]
             database_use_test_transactions: false,

syncstorage-spanner/Cargo.toml

@@ -6,6 +6,7 @@ authors.workspace = true
 edition.workspace = true
 
 [dependencies]
+actix-web.workspace = true
 backtrace.workspace = true
 cadence.workspace = true
 deadpool.workspace = true

syncstorage-spanner/src/manager/deadpool.rs

@@ -13,7 +13,7 @@ use crate::error::DbError;
 pub(crate) type Conn = deadpool::managed::Object<SpannerSessionManager>;
 
 pub(crate) struct SpannerSessionManager {
-    settings: SpannerSessionSettings,
+    pub settings: SpannerSessionSettings,
     /// The gRPC environment
     env: Arc<Environment>,
     metrics: Metrics,

syncstorage-spanner/src/pool.rs

@@ -1,5 +1,6 @@
 use std::{collections::HashMap, fmt, sync::Arc, time::Duration};
 
+use actix_web::rt;
 use async_trait::async_trait;
 use syncserver_common::{BlockingThreadpool, Metrics};
 use syncserver_db_common::{GetPoolState, PoolState};
@@ -49,7 +50,9 @@ impl SpannerDbPool {
         let config = deadpool::managed::PoolConfig {
             max_size,
             timeouts,
-            ..Default::default()
+            // Prefer LIFO to allow the sweeper task to evict least frequently
+            // used connections.
+            queue_mode: deadpool::managed::QueueMode::Lifo,
         };
         let pool = deadpool::managed::Pool::builder(manager)
             .config(config)
@@ -84,6 +87,30 @@ impl SpannerDbPool {
             self.quota,
         ))
     }
+
+    /// Spawn a task to periodically evict idle connections. Calls wrapper sweeper fn
+    ///  to use pool.retain, retaining objects only if they are shorter in duration than
+    ///  defined max_idle.
+    pub fn spawn_sweeper(&self, interval: Duration) {
+        let Some(max_idle) = self.pool.manager().settings.max_idle else {
+            return;
+        };
+        let pool = self.pool.clone();
+        rt::spawn(async move {
+            loop {
+                sweeper(&pool, Duration::from_secs(max_idle.into()));
+                rt::time::sleep(interval).await;
+            }
+        });
+    }
+}
+
+/// Sweeper to retain only the objects specified within the closure.
+/// In this context, if a Spanner connection is unutilized, we want it
+/// to release the given connection.
+/// See: https://docs.rs/deadpool/latest/deadpool/managed/struct.Pool.html#method.retain
+fn sweeper(pool: &deadpool::managed::Pool<SpannerSessionManager>, max_idle: Duration) {
+    pool.retain(|_, metrics| metrics.last_used() < max_idle);
 }
 
 #[async_trait]