store: Reap idle connections according to configuration

Author: lutter

store/postgres/src/pool/manager.rs

@@ -8,6 +8,7 @@ use diesel::IntoSql;
 
 use diesel_async::pooled_connection::{PoolError as DieselPoolError, PoolableConnection};
 use diesel_async::{AsyncConnection, RunQueryDsl};
+use graph::env::ENV_VARS;
 use graph::prelude::error;
 use graph::prelude::Counter;
 use graph::prelude::Gauge;
@@ -237,6 +238,33 @@ pub(crate) fn spawn_size_stat_collector(
     });
 }
 
+/// Reap connections that are too old (older than 30 minutes) or if there
+/// are more than `connection_min_idle` connections in the pool that have
+/// been idle for longer than `idle_timeout`
+pub(crate) fn spawn_connection_reaper(pool: AsyncPool, idle_timeout: Duration) {
+    const MAX_LIFETIME: Duration = Duration::from_secs(30 * 60);
+    let Some(min_idle) = ENV_VARS.store.connection_min_idle else {
+        // If this is None, we will never reap anything
+        return;
+    };
+    tokio::task::spawn(async move {
+        loop {
+            let mut idle_count = 0;
+            pool.retain(|_, metrics| {
+                if metrics.age() > MAX_LIFETIME {
+                    return false;
+                }
+                if metrics.last_used() > idle_timeout {
+                    idle_count += 1;
+                    return idle_count <= min_idle;
+                }
+                true
+            });
+            tokio::time::sleep(Duration::from_secs(30)).await;
+        }
+    });
+}
+
 pub(crate) struct WaitMeter {
     wait_gauge: Gauge,
     pub(crate) wait_stats: PoolWaitStats,

store/postgres/src/pool/mod.rs

@@ -438,15 +438,6 @@ impl PoolInner {
             const_labels.clone(),
         );
 
-        // Note: deadpool does not support min_idle configuration
-        if let Some(min_idle) = ENV_VARS.store.connection_min_idle {
-            warn!(
-                logger_pool,
-                "min_idle configuration ({}) is not supported by deadpool and will be ignored",
-                min_idle
-            );
-        }
-
         let timeouts = Timeouts {
             wait: Some(ENV_VARS.store.connection_timeout),
             create: Some(ENV_VARS.store.connection_timeout),
@@ -467,6 +458,8 @@ impl PoolInner {
 
         manager::spawn_size_stat_collector(pool.clone(), &registry, const_labels.clone());
 
+        manager::spawn_connection_reaper(pool.clone(), ENV_VARS.store.connection_idle_timeout);
+
         let wait_meter = WaitMeter::new(&registry, const_labels.clone());
 
         let fdw_pool = fdw_pool_size.map(|pool_size| {
@@ -476,14 +469,17 @@ impl PoolInner {
                 recycle: Some(FDW_IDLE_TIMEOUT),
             };
 
-            AsyncPool::builder(conn_manager)
+            let fdw_pool = AsyncPool::builder(conn_manager)
                 .max_size(pool_size as usize)
                 .timeouts(fdw_timeouts)
                 .runtime(Runtime::Tokio1)
                 .post_create(state_tracker.mark_available_hook())
                 .post_recycle(state_tracker.mark_available_hook())
                 .build()
-                .expect("failed to create fdw connection pool")
+                .expect("failed to create fdw connection pool");
+
+            manager::spawn_connection_reaper(fdw_pool.clone(), FDW_IDLE_TIMEOUT);
+            fdw_pool
         });
 
         info!(logger_store, "Pool successfully connected to Postgres");