diff --git a/crates/tranquil-config/src/lib.rs b/crates/tranquil-config/src/lib.rs index 661e5e3..976ecbe 100644 --- a/crates/tranquil-config/src/lib.rs +++ b/crates/tranquil-config/src/lib.rs @@ -176,6 +176,9 @@ impl TranquilConfig { .map(|(var, guidance)| format!("{var} is no longer supported: {guidance}")), ); + // -- database -------------------------------------------------------- + self.database.validate(&mut errors); + // -- secrets ---------------------------------------------------------- if !ignore_secrets && !self.secrets.allow_insecure && !cfg!(test) { if let Some(ref s) = self.secrets.jwt_secret { @@ -615,10 +618,20 @@ pub struct FrontendConfig { #[derive(Debug, Config)] #[config(layer_attr(serde(deny_unknown_fields)))] pub struct DatabaseConfig { - /// PostgreSQL connection URL. + /// PostgreSQL connection URL. When PgBouncer support is enabled, this + /// should point to a transaction-pooled PgBouncer endpoint. #[config(env = "DATABASE_URL")] pub url: String, + /// Direct PostgreSQL URL used for migrations and LISTEN notifications when + /// normal queries go through transaction-pooled PgBouncer. + #[config(env = "DATABASE_DIRECT_URL")] + pub direct_url: Option, + + /// Enable compatibility with PgBouncer transaction pooling. + #[config(env = "DATABASE_PGBOUNCER", default = false)] + pub pgbouncer: bool, + /// Maximum number of connections in the pool. #[config(env = "DATABASE_MAX_CONNECTIONS", default = 100)] pub max_connections: u32, @@ -632,6 +645,35 @@ pub struct DatabaseConfig { pub acquire_timeout_secs: u64, } +impl DatabaseConfig { + fn validate(&self, errors: &mut Vec) { + if self.pgbouncer + && self + .direct_url + .as_deref() + .is_none_or(|url| url.trim().is_empty()) + { + errors.push( + "database.pgbouncer is enabled but database.direct_url \ + (DATABASE_DIRECT_URL) is not set; migrations and LISTEN require a direct \ + PostgreSQL connection" + .to_string(), + ); + } + if self.max_connections == 0 { + errors.push("database.max_connections must be greater than zero".to_string()); + } + if self.min_connections > self.max_connections { + errors.push( + "database.min_connections cannot exceed database.max_connections".to_string(), + ); + } + if self.acquire_timeout_secs == 0 { + errors.push("database.acquire_timeout_secs must be greater than zero".to_string()); + } + } +} + #[derive(Config)] #[config(layer_attr(serde(deny_unknown_fields)))] pub struct SecretsConfig { @@ -1648,6 +1690,63 @@ pub fn template() -> String { mod tests { use super::*; + #[test] + fn pgbouncer_requires_a_direct_postgres_url() { + let config = DatabaseConfig { + url: "postgres://pgbouncer/pds".to_string(), + direct_url: None, + pgbouncer: true, + max_connections: 20, + min_connections: 2, + acquire_timeout_secs: 10, + }; + let mut errors = Vec::new(); + + config.validate(&mut errors); + + assert!( + errors.iter().any(|error| error.contains("direct_url")), + "expected a direct_url error, got {errors:?}" + ); + } + + #[test] + fn pgbouncer_accepts_a_direct_postgres_url() { + let config = DatabaseConfig { + url: "postgres://pgbouncer/pds".to_string(), + direct_url: Some("postgres://postgres/pds".to_string()), + pgbouncer: true, + max_connections: 20, + min_connections: 2, + acquire_timeout_secs: 10, + }; + let mut errors = Vec::new(); + + config.validate(&mut errors); + + assert!(errors.is_empty(), "expected no errors, got {errors:?}"); + } + + #[test] + fn database_pool_minimum_cannot_exceed_maximum() { + let config = DatabaseConfig { + url: "postgres://postgres/pds".to_string(), + direct_url: None, + pgbouncer: false, + max_connections: 2, + min_connections: 3, + acquire_timeout_secs: 10, + }; + let mut errors = Vec::new(); + + config.validate(&mut errors); + + assert!( + errors.iter().any(|error| error.contains("min_connections")), + "expected a pool size error, got {errors:?}" + ); + } + fn seed_required_env() { let required = [ ("PDS_HOSTNAME", "test.local"), diff --git a/crates/tranquil-db/src/postgres/mod.rs b/crates/tranquil-db/src/postgres/mod.rs index 1f97998..9674e7c 100644 --- a/crates/tranquil-db/src/postgres/mod.rs +++ b/crates/tranquil-db/src/postgres/mod.rs @@ -142,6 +142,10 @@ pub struct PostgresRepositories { impl PostgresRepositories { pub fn new(pool: PgPool) -> Self { + Self::new_with_event_pool(pool.clone(), pool) + } + + pub fn new_with_event_pool(pool: PgPool, event_pool: PgPool) -> Self { Self { pool: Some(pool.clone()), user: Arc::new(PostgresUserRepository::new(pool.clone())), @@ -153,7 +157,7 @@ impl PostgresRepositories { infra: Arc::new(PostgresInfraRepository::new(pool.clone())), backlink: Arc::new(PostgresBacklinkRepository::new(pool.clone())), sso: Arc::new(PostgresSsoRepository::new(pool.clone())), - event_notifier: Arc::new(PostgresRepoEventNotifier::new(pool)), + event_notifier: Arc::new(PostgresRepoEventNotifier::new(event_pool)), } } } diff --git a/crates/tranquil-pds/src/state.rs b/crates/tranquil-pds/src/state.rs index 66f5fd1..c20362a 100644 --- a/crates/tranquil-pds/src/state.rs +++ b/crates/tranquil-pds/src/state.rs @@ -11,8 +11,10 @@ use crate::repo_write_lock::RepoWriteLocks; use crate::sso::{SsoConfig, SsoManager}; use crate::storage::{BlobStorage, create_blob_storage}; use sqlx::PgPool; +use sqlx::postgres::{PgConnectOptions, PgPoolOptions}; use std::error::Error; use std::path::PathBuf; +use std::str::FromStr; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use tokio::sync::broadcast; @@ -22,6 +24,18 @@ use tranquil_db_traits::SequencedEvent; static RATE_LIMITING_DISABLED: AtomicBool = AtomicBool::new(false); +fn postgres_connect_options( + database_url: &str, + pgbouncer: bool, +) -> Result { + let options = PgConnectOptions::from_str(database_url)?; + Ok(if pgbouncer { + options.statement_cache_capacity(0) + } else { + options + }) +} + pub fn init_rate_limit_override() { let disabled = tranquil_config::get().server.disable_rate_limiting; RATE_LIMITING_DISABLED.store(disabled, Ordering::Relaxed); @@ -224,34 +238,56 @@ impl AppState { Self::from_store(shutdown).await } tranquil_config::RepoBackend::Postgres => { - let database_url = &cfg.database.url; let max_connections = cfg.database.max_connections; let min_connections = cfg.database.min_connections; let acquire_timeout_secs = cfg.database.acquire_timeout_secs; + let pgbouncer = cfg.database.pgbouncer; tracing::info!( + pgbouncer, "Configuring database pool: max={}, min={}, acquire_timeout={}s", max_connections, min_connections, acquire_timeout_secs ); - let db = sqlx::postgres::PgPoolOptions::new() + let db_options = postgres_connect_options(&cfg.database.url, pgbouncer) + .map_err(|e| format!("Invalid database URL: {e}"))?; + let db = PgPoolOptions::new() .max_connections(max_connections) .min_connections(min_connections) .acquire_timeout(std::time::Duration::from_secs(acquire_timeout_secs)) .idle_timeout(std::time::Duration::from_secs(300)) .max_lifetime(std::time::Duration::from_secs(1800)) - .connect(database_url) + .connect_with(db_options) .await .map_err(|e| format!("Failed to connect to Postgres: {}", e))?; + let event_pool = if pgbouncer { + let direct_url = cfg.database.direct_url.as_deref().ok_or( + "DATABASE_DIRECT_URL is required when DATABASE_PGBOUNCER is enabled", + )?; + let direct_options = postgres_connect_options(direct_url, false) + .map_err(|e| format!("Invalid direct database URL: {e}"))?; + PgPoolOptions::new() + .max_connections(1) + .min_connections(0) + .acquire_timeout(std::time::Duration::from_secs(acquire_timeout_secs)) + .idle_timeout(std::time::Duration::from_secs(300)) + .max_lifetime(std::time::Duration::from_secs(1800)) + .connect_with(direct_options) + .await + .map_err(|e| format!("Failed to connect directly to Postgres: {e}"))? + } else { + db.clone() + }; + sqlx::migrate!("./migrations") - .run(&db) + .run(&event_pool) .await .map_err(|e| format!("Failed to run migrations: {}", e))?; - Self::from_db(db, shutdown).await + Self::from_db_with_event_pool(db, event_pool, shutdown).await } }; @@ -269,6 +305,14 @@ impl AppState { } pub async fn from_db(db: PgPool, shutdown: CancellationToken) -> Self { + Self::from_db_with_event_pool(db.clone(), db, shutdown).await + } + + async fn from_db_with_event_pool( + db: PgPool, + event_pool: PgPool, + shutdown: CancellationToken, + ) -> Self { let cfg = tranquil_config::get(); let (repos, block_store, signal_store_provider, eventlog_segments_dir): ( PostgresRepositories, @@ -286,7 +330,7 @@ impl AppState { ) } false => { - let repos = PostgresRepositories::new(db.clone()); + let repos = PostgresRepositories::new_with_event_pool(db.clone(), event_pool); let provider: Arc = Arc::new(tranquil_signal::PgSignalStoreProvider { pool: db.clone() }); ( @@ -725,3 +769,37 @@ fn wire_tranquil_store( segments_dir: eventlog_segments_dir, } } + +#[cfg(test)] +mod tests { + use super::postgres_connect_options; + use sqlx::ConnectOptions; + + #[test] + fn pgbouncer_connections_disable_the_statement_cache() { + let options = postgres_connect_options("postgres://localhost/pds", true) + .expect("valid PostgreSQL URL"); + let url = options.to_url_lossy(); + + assert_eq!( + url.query_pairs() + .find(|(key, _)| key == "statement-cache-capacity") + .map(|(_, value)| value.into_owned()), + Some("0".to_string()) + ); + } + + #[test] + fn direct_connections_keep_the_default_statement_cache() { + let options = postgres_connect_options("postgres://localhost/pds", false) + .expect("valid PostgreSQL URL"); + let url = options.to_url_lossy(); + + assert_eq!( + url.query_pairs() + .find(|(key, _)| key == "statement-cache-capacity") + .map(|(_, value)| value.into_owned()), + Some("100".to_string()) + ); + } +} diff --git a/example.toml b/example.toml index 8f5771e..21a4a84 100644 --- a/example.toml +++ b/example.toml @@ -166,13 +166,30 @@ #dir = "/var/lib/tranquil-pds/frontend" [database] -# PostgreSQL connection URL. +# PostgreSQL connection URL. When PgBouncer support is enabled, point this at +# the transaction-pooled PgBouncer service. # # Can also be specified via environment variable `DATABASE_URL`. # # Required! This value must be specified. #url = +# Direct PostgreSQL connection URL used only for startup migrations and the +# firehose LISTEN connection. Required when PgBouncer support is enabled. +# Normal application queries do not use this URL. +# +# Can also be specified via environment variable `DATABASE_DIRECT_URL`. +#direct_url = + +# Enable compatibility with PgBouncer transaction pooling. This disables +# SQLx's per-connection prepared-statement cache on the normal query pool and +# routes session-bound operations through direct_url. +# +# Can also be specified via environment variable `DATABASE_PGBOUNCER`. +# +# Default value: false +#pgbouncer = false + # Maximum number of connections in the pool. # # Can also be specified via environment variable `DATABASE_MAX_CONNECTIONS`.