feat(prepared_statements): add new ttl config

.schema/pgdog.schema.json

@@ -83,6 +83,8 @@
         "port": 6432,
         "prepared_statements": "extended",
         "prepared_statements_limit": 9223372036854775807,
+        "prepared_statements_ttl": null,
+        "prepared_statements_ttl_jitter": 30000,
         "pub_sub_channel_size": 0,
         "query_cache_limit": 1000,
         "query_log": null,
@@ -1028,6 +1030,23 @@
           "default": 9223372036854775807,
           "minimum": 0
         },
+        "prepared_statements_ttl": {
+          "description": "How long a prepared statement is allowed to stay prepared on a server connection, in milliseconds.\n\n**Note:** Expired statements are closed and prepared again the next time they are used. This stops stale execution plans from staying in Postgres. Omit or set to `0` to let statements stay prepared forever.\n\n_Default:_ `None` (disabled)\n\n<https://docs.pgdog.dev/configuration/pgdog.toml/general/#prepared_statements_ttl>",
+          "type": [
+            "integer",
+            "null"
+          ],
+          "format": "uint64",
+          "default": null,
+          "minimum": 0
+        },
+        "prepared_statements_ttl_jitter": {
+          "description": "Maximum random adjustment applied to `prepared_statements_ttl` per prepared\nstatement, in milliseconds. Each statement expires at a point sampled uniformly\nfrom `[ttl - jitter, ttl + jitter]`, chosen once when the statement is prepared.\n\n**Note:** Clients run their whole statement set together, so a small jitter\nleaves them expiring in lockstep for hours. Keep this at a meaningful fraction\nof the TTL.\n\n_Default:_ `30000` (30 seconds)\n\n<https://docs.pgdog.dev/configuration/pgdog.toml/general/#prepared_statements_ttl_jitter>",
+          "type": "integer",
+          "format": "uint64",
+          "default": 30000,
+          "minimum": 0
+        },
         "pub_sub_channel_size": {
           "description": "Enables support for pub/sub and configures the size of the background task queue.\n\n<https://docs.pgdog.dev/configuration/pgdog.toml/general/#pub_sub_channel_size>",
           "type": "integer",

example.pgdog.toml

@@ -197,6 +197,22 @@ query_parser = "on"
 # Default: unlimited
 #
 prepared_statements_limit = 1_000
+# How long a prepared statement can stay prepared on a Postgres server
+# connection. Expired statements are closed and prepared again the next
+# time they are used, so execution plans don't get stale.
+#
+# Omit or set to 0 to let statements stay prepared forever.
+#
+# Default: none (disabled)
+#
+prepared_statements_ttl = 300_000
+# Random spread applied to prepared_statements_ttl, per statement. Clients
+# usually prepare their whole statement set at once, so without this they
+# would all expire at the same moment, over and over.
+#
+# Default: 30_000 (30 seconds)
+#
+prepared_statements_ttl_jitter = 30_000
 # Limit on the number of queries cached in the Abstract Syntax Tree
 # cache used for query routing and sharding.
 #

integration/pgdog.toml

@@ -12,6 +12,8 @@ openmetrics_port = 9090
 openmetrics_namespace = "pgdog_"
 prepared_statements_limit = 500
 prepared_statements = "extended"
+prepared_statements_ttl = 5_000
+prepared_statements_ttl_jitter = 1_000
 expanded_explain = true
 dns_ttl = 1_000
 query_cache_limit = 500

pgdog-config/src/general.rs

@@ -382,6 +382,30 @@ pub struct General {
     #[serde(default = "General::prepared_statements_limit")]
     pub prepared_statements_limit: usize,
 
+    /// How long a prepared statement is allowed to stay prepared on a server connection, in milliseconds.
+    ///
+    /// **Note:** Expired statements are closed and prepared again the next time they are used. This stops stale execution plans from staying in Postgres. Omit or set to `0` to let statements stay prepared forever.
+    ///
+    /// _Default:_ `None` (disabled)
+    ///
+    /// <https://docs.pgdog.dev/configuration/pgdog.toml/general/#prepared_statements_ttl>
+    #[serde(default = "General::default_prepared_statements_ttl")]
+    pub prepared_statements_ttl: Option<u64>,
+
+    /// Maximum random adjustment applied to `prepared_statements_ttl` per prepared
+    /// statement, in milliseconds. Each statement expires at a point sampled uniformly
+    /// from `[ttl - jitter, ttl + jitter]`, chosen once when the statement is prepared.
+    ///
+    /// **Note:** Clients run their whole statement set together, so a small jitter
+    /// leaves them expiring in lockstep for hours. Keep this at a meaningful fraction
+    /// of the TTL.
+    ///
+    /// _Default:_ `30000` (30 seconds)
+    ///
+    /// <https://docs.pgdog.dev/configuration/pgdog.toml/general/#prepared_statements_ttl_jitter>
+    #[serde(default = "General::default_prepared_statements_ttl_jitter")]
+    pub prepared_statements_ttl_jitter: u64,
+
     /// Limit on the number of statements saved in the statement cache used to accelerate query parsing.
     ///
     /// _Default:_ `50000`
@@ -887,6 +911,8 @@ impl Default for General {
             regex_parser_limit: Self::regex_parser_limit(),
             query_parser_engine: QueryParserEngine::default(),
             prepared_statements_limit: Self::prepared_statements_limit(),
+            prepared_statements_ttl: Self::default_prepared_statements_ttl(),
+            prepared_statements_ttl_jitter: Self::default_prepared_statements_ttl_jitter(),
             query_cache_limit: Self::query_cache_limit(),
             passthrough_auth: Self::default_passthrough_auth(),
             connect_timeout: Self::default_connect_timeout(),
@@ -1142,6 +1168,28 @@ impl General {
         self.dns_ttl.map(Duration::from_millis)
     }
 
+    /// How long a statement can stay prepared on a server connection.
+    ///
+    /// `None` means statements never expire.
+    pub fn prepared_statements_ttl(&self) -> Option<Duration> {
+        self.prepared_statements_ttl
+            .filter(|ttl| *ttl > 0 && *ttl < crate::MAX_DURATION.as_millis() as u64)
+            .map(Duration::from_millis)
+    }
+
+    /// Random spread applied to [`Self::prepared_statements_ttl`].
+    ///
+    /// Clamped to just below the TTL.
+    pub fn prepared_statements_ttl_jitter(&self) -> Duration {
+        let ttl = self
+            .prepared_statements_ttl
+            .unwrap_or(0)
+            .min(crate::MAX_DURATION.as_millis() as u64)
+            .saturating_sub(1);
+
+        Duration::from_millis(self.prepared_statements_ttl_jitter.min(ttl))
+    }
+
     pub fn client_idle_timeout(&self) -> Duration {
         Duration::from_millis(self.client_idle_timeout)
     }
@@ -1381,6 +1429,17 @@ impl General {
         Self::env_or_default("PGDOG_PREPARED_STATEMENTS_LIMIT", i64::MAX as usize)
     }
 
+    fn default_prepared_statements_ttl() -> Option<u64> {
+        Self::env_option("PGDOG_PREPARED_STATEMENTS_TTL")
+    }
+
+    pub fn default_prepared_statements_ttl_jitter() -> u64 {
+        Self::env_or_default(
+            "PGDOG_PREPARED_STATEMENTS_TTL_JITTER",
+            Duration::from_secs(30).as_millis() as u64,
+        )
+    }
+
     pub fn query_cache_limit() -> usize {
         Self::env_or_default("PGDOG_QUERY_CACHE_LIMIT", 1_000)
     }
@@ -1494,6 +1553,82 @@ mod tests {
     use super::*;
     use crate::test_utils::*;
 
+    #[test]
+    fn test_prepared_statements_ttl_defaults() {
+        let general = General::default();
+
+        assert_eq!(general.prepared_statements_ttl(), None);
+        assert_eq!(
+            general.prepared_statements_ttl_jitter(),
+            Duration::from_millis(0)
+        );
+    }
+
+    #[test]
+    fn test_prepared_statements_ttl_is_read_in_millis() {
+        let general = General {
+            prepared_statements_ttl: Some(3_600_000),
+            prepared_statements_ttl_jitter: 5_000,
+            ..Default::default()
+        };
+
+        assert_eq!(
+            general.prepared_statements_ttl(),
+            Some(Duration::from_millis(3_600_000))
+        );
+        assert_eq!(
+            general.prepared_statements_ttl_jitter(),
+            Duration::from_millis(5_000)
+        );
+    }
+
+    #[test]
+    fn test_prepared_statements_ttl_disabled() {
+        for ttl in [None, Some(0), Some(crate::MAX_DURATION.as_millis() as u64)] {
+            let general = General {
+                prepared_statements_ttl: ttl,
+                ..Default::default()
+            };
+
+            assert_eq!(
+                general.prepared_statements_ttl(),
+                None,
+                "ttl {ttl:?} should disable expiration"
+            );
+        }
+    }
+
+    #[test]
+    fn test_prepared_statements_ttl_jitter_is_clamped_below_ttl() {
+        let general = General {
+            prepared_statements_ttl: Some(10_000),
+            prepared_statements_ttl_jitter: u64::MAX,
+            ..Default::default()
+        };
+
+        assert_eq!(
+            general.prepared_statements_ttl_jitter(),
+            Duration::from_millis(9_999)
+        );
+    }
+
+    #[test]
+    fn test_prepared_statements_ttl_jitter_never_reaches_the_ttl() {
+        for ttl in [0, 1, 2, 300_000] {
+            let general = General {
+                prepared_statements_ttl: Some(ttl),
+                prepared_statements_ttl_jitter: u64::MAX,
+                ..Default::default()
+            };
+
+            let jitter = general.prepared_statements_ttl_jitter().as_millis() as u64;
+            assert!(
+                jitter < ttl.max(1),
+                "jitter {jitter} must stay below ttl {ttl}"
+            );
+        }
+    }
+
     #[test]
     fn test_sharding_lookup_cache_size() {
         let general = General::default();
@@ -1820,6 +1955,8 @@ mod tests {
         let _guard = set_env_var("PGDOG_BROADCAST_PORT", "7432");
         let _guard = set_env_var("PGDOG_OPENMETRICS_PORT", "9090");
         let _guard = set_env_var("PGDOG_PREPARED_STATEMENTS_LIMIT", "1000");
+        let _guard = set_env_var("PGDOG_PREPARED_STATEMENTS_TTL", "3600000");
+        let _guard = set_env_var("PGDOG_PREPARED_STATEMENTS_TTL_JITTER", "5000");
         let _guard = set_env_var("PGDOG_QUERY_CACHE_LIMIT", "500");
         let _guard = set_env_var("PGDOG_CONNECT_ATTEMPTS", "3");
         let _guard = set_env_var("PGDOG_MIRROR_QUEUE", "256");
@@ -1832,6 +1969,8 @@ mod tests {
         assert_eq!(General::broadcast_port(), 7432);
         assert_eq!(General::openmetrics_port(), Some(9090));
         assert_eq!(General::prepared_statements_limit(), 1000);
+        assert_eq!(General::default_prepared_statements_ttl(), Some(3600000));
+        assert_eq!(General::default_prepared_statements_ttl_jitter(), 5000);
         assert_eq!(General::query_cache_limit(), 500);
         assert_eq!(General::connect_attempts(), 3);
         assert_eq!(General::mirror_queue(), 256);
@@ -1844,6 +1983,8 @@ mod tests {
         let _guard = remove_env_var("PGDOG_BROADCAST_PORT");
         let _guard = remove_env_var("PGDOG_OPENMETRICS_PORT");
         let _guard = remove_env_var("PGDOG_PREPARED_STATEMENTS_LIMIT");
+        let _guard = remove_env_var("PGDOG_PREPARED_STATEMENTS_TTL");
+        let _guard = remove_env_var("PGDOG_PREPARED_STATEMENTS_TTL_JITTER");
         let _guard = remove_env_var("PGDOG_QUERY_CACHE_LIMIT");
         let _guard = remove_env_var("PGDOG_CONNECT_ATTEMPTS");
         let _guard = remove_env_var("PGDOG_MIRROR_QUEUE");
@@ -1856,6 +1997,8 @@ mod tests {
         assert_eq!(General::broadcast_port(), General::port() + 1);
         assert_eq!(General::openmetrics_port(), None);
         assert_eq!(General::prepared_statements_limit(), i64::MAX as usize);
+        assert_eq!(General::default_prepared_statements_ttl(), None);
+        assert_eq!(General::default_prepared_statements_ttl_jitter(), 30_000);
         assert_eq!(General::query_cache_limit(), 1_000);
         assert_eq!(General::connect_attempts(), 1);
         assert_eq!(General::mirror_queue(), 128);

pgdog-stats/src/pool.rs

@@ -285,6 +285,30 @@ pub struct State {
     pub lsn_stats: LsnStats,
 }
 
+/// How a server connection handles prepared statements.
+#[derive(Debug, Serialize, Deserialize, Clone, Copy, PartialEq)]
+pub struct PreparedStatementsConfig {
+    /// Which statements PgDog keeps prepared on the connection.
+    pub level: PreparedStatements,
+    /// Maximum prepared statements per connection.
+    pub limit: usize,
+    /// How long a statement can keep a cached plan. `None` never expires.
+    pub ttl: Option<Duration>,
+    /// Random spread applied to `ttl`, per statement.
+    pub ttl_jitter: Duration,
+}
+
+impl Default for PreparedStatementsConfig {
+    fn default() -> Self {
+        Self {
+            level: PreparedStatements::default(),
+            limit: usize::MAX,
+            ttl: None,
+            ttl_jitter: Duration::ZERO,
+        }
+    }
+}
+
 /// Pool configuration.
 #[derive(Debug, Serialize, Deserialize, Clone, Copy, PartialEq)]
 pub struct Config {
@@ -353,8 +377,8 @@ pub struct Config {
     pub pooler_mode: PoolerMode,
     /// Read only mode.
     pub read_only: bool,
-    /// Maximum prepared statements per connection.
-    pub prepared_statements_limit: usize,
+    /// Prepared statements config.
+    pub prepared_statements: PreparedStatementsConfig,
     /// Stats averaging period.
     pub stats_period: Duration,
     /// Recovery algo.
@@ -371,8 +395,6 @@ pub struct Config {
     pub resharding_only: bool,
     /// LB weight.
     pub lb_weight: u8,
-    /// Prepared statements level.
-    pub prepared_statements_level: PreparedStatements,
 }
 
 pub struct RoleSpecificConfig<T: Copy> {
@@ -424,7 +446,7 @@ impl Default for Config {
             replication_mode: false,
             pooler_mode: PoolerMode::default(),
             read_only: false,
-            prepared_statements_limit: usize::MAX,
+            prepared_statements: PreparedStatementsConfig::default(),
             stats_period: Duration::from_millis(15_000),
             dns_ttl: Duration::from_millis(60_000),
             connection_recovery: ConnectionRecovery::Recover,
@@ -434,7 +456,6 @@ impl Default for Config {
             role_detection: false,
             resharding_only: false,
             lb_weight: 255,
-            prepared_statements_level: PreparedStatements::default(),
         }
     }
 }

pgdog/src/backend/pool/config.rs

@@ -122,7 +122,12 @@ impl Config {
                 read_only: user
                     .read_only
                     .unwrap_or(database.read_only.unwrap_or_default()),
-                prepared_statements_limit: general.prepared_statements_limit,
+                prepared_statements: pgdog_stats::PreparedStatementsConfig {
+                    level: general.prepared_statements,
+                    limit: general.prepared_statements_limit,
+                    ttl: general.prepared_statements_ttl(),
+                    ttl_jitter: general.prepared_statements_ttl_jitter(),
+                },
                 stats_period: Duration::from_millis(general.stats_period),
                 bannable: !is_only_replica,
                 connection_recovery: general.connection_recovery,
@@ -132,7 +137,6 @@ impl Config {
                 role_detection: database.role == Role::Auto,
                 resharding_only: database.resharding_only,
                 lb_weight: database.lb_weight,
-                prepared_statements_level: general.prepared_statements,
                 ..Default::default()
             },
         }
@@ -165,6 +169,32 @@ mod test {
         assert!(config.role_detection);
     }
 
+    #[test]
+    fn test_prepared_statements_config_from_general() {
+        let general = General {
+            prepared_statements_ttl: Some(60_000),
+            prepared_statements_limit: 10,
+            ..Default::default()
+        };
+
+        let config = Config::new(
+            &general,
+            &create_database(Role::Primary),
+            &User::default(),
+            false,
+        );
+
+        assert_eq!(
+            config.prepared_statements.ttl,
+            Some(Duration::from_millis(60_000))
+        );
+        assert_eq!(config.prepared_statements.limit, 10);
+        assert_eq!(
+            config.prepared_statements.level,
+            general.prepared_statements
+        );
+    }
+
     #[test]
     fn test_user_takes_precedence_over_database() {
         let general = General::default();