diff --git a/.env.example b/.env.example index b9bfcada0e..e81ea15895 100644 --- a/.env.example +++ b/.env.example @@ -38,6 +38,12 @@ REDIS_URL=redis://localhost:6379 # READ_DATABASE_URL is set, reader (default 50). # BUZZ_DB_POOL_SIZE=50 +# Postgres statement_timeout and lock_timeout applied to every runtime +# connection. Accepts an integer (milliseconds) with an optional us/ms/s/min/h/d +# unit; `0` disables the limit. Schema migrations always run with both lifted. +# BUZZ_DB_STATEMENT_TIMEOUT=30s +# BUZZ_DB_LOCK_TIMEOUT=5s + # ----------------------------------------------------------------------------- # Typesense (search) # ----------------------------------------------------------------------------- diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9b26876747..37977a4541 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -65,6 +65,32 @@ use uuid::Uuid; use buzz_core::{CommunityId, StoredEvent}; +/// Default maximum time a runtime query may execute before Postgres cancels it. +pub const RUNTIME_STATEMENT_TIMEOUT: &str = "30s"; +/// Default maximum time a runtime query may wait to acquire a lock. +pub const RUNTIME_LOCK_TIMEOUT: &str = "5s"; +/// Postgres spelling of "no limit", used for schema migrations. +pub const TIMEOUT_DISABLED: &str = "0"; + +/// Apply the runtime safety limits shared by writer, reader, audit, and search +/// pools. Values are Postgres interval strings (`"30s"`, `"500ms"`), with +/// [`TIMEOUT_DISABLED`] lifting a limit entirely. +pub async fn apply_runtime_connection_timeouts( + connection: &mut PgConnection, + statement_timeout: &str, + lock_timeout: &str, +) -> std::result::Result<(), sqlx::Error> { + sqlx::query( + "SELECT set_config('statement_timeout', $1, false), \ + set_config('lock_timeout', $2, false)", + ) + .bind(statement_timeout) + .bind(lock_timeout) + .execute(connection) + .await?; + Ok(()) +} + fn event_replacement_lock_key( community_id: CommunityId, kind: i32, @@ -527,6 +553,14 @@ pub struct DbConfig { /// than the staleness gate never routes anyway, so a larger budget /// would only misrepresent the config. pub replica_read_max_age_ms: u64, + /// Postgres `statement_timeout` applied to every runtime connection. An + /// operator running a backfill or working an incident can widen this without + /// a code change; [`TIMEOUT_DISABLED`] removes the cap. + pub statement_timeout: String, + /// Postgres `lock_timeout` applied to every runtime connection. Bounds + /// heavyweight and row lock waits only — advisory-lock waits are bounded by + /// [`Self::statement_timeout`] instead. + pub lock_timeout: String, } impl Default for DbConfig { @@ -544,6 +578,8 @@ impl Default for DbConfig { max_lifetime_secs: 1800, idle_timeout_secs: 600, replica_read_max_age_ms: 0, + statement_timeout: RUNTIME_STATEMENT_TIMEOUT.to_string(), + lock_timeout: RUNTIME_LOCK_TIMEOUT.to_string(), } } } @@ -682,18 +718,23 @@ impl Db { .acquire_timeout(Duration::from_secs(config.acquire_timeout_secs)) .max_lifetime(Duration::from_secs(config.max_lifetime_secs)) .idle_timeout(Duration::from_secs(config.idle_timeout_secs)); - if arm_floor_guard { - options = options.after_connect(|conn, _meta| { - Box::pin(async move { + let statement_timeout = config.statement_timeout.clone(); + let lock_timeout = config.lock_timeout.clone(); + options = options.after_connect(move |conn, _meta| { + let statement_timeout = statement_timeout.clone(); + let lock_timeout = lock_timeout.clone(); + Box::pin(async move { + apply_runtime_connection_timeouts(conn, &statement_timeout, &lock_timeout).await?; + if arm_floor_guard { // `SET` cannot take bind parameters; `set_config` can. sqlx::query("SELECT set_config('buzz.created_at_floor', $1, false)") .bind(replica_fence::CREATED_AT_FLOOR_SECS.to_string()) .execute(conn) .await?; - Ok(()) - }) - }); - } + } + Ok(()) + }) + }); Ok(options.connect(url).await?) } @@ -721,12 +762,22 @@ impl Db { /// No floor guard: replica sessions are read-only, the trigger never /// fires there (see [`Db::connect_pool`]). fn connect_read_pool(config: &DbConfig, url: &str, max_connections: u32) -> Result { + let statement_timeout = config.statement_timeout.clone(); + let lock_timeout = config.lock_timeout.clone(); Ok(PgPoolOptions::new() .max_connections(max_connections) .min_connections(0) .acquire_timeout(Self::READER_ACQUIRE_TIMEOUT) .max_lifetime(Duration::from_secs(config.max_lifetime_secs)) .idle_timeout(Duration::from_secs(config.idle_timeout_secs)) + .after_connect(move |connection, _meta| { + let statement_timeout = statement_timeout.clone(); + let lock_timeout = lock_timeout.clone(); + Box::pin(async move { + apply_runtime_connection_timeouts(connection, &statement_timeout, &lock_timeout) + .await + }) + }) .connect_lazy(url)?) } @@ -6511,6 +6562,65 @@ mod tests { .await; } + /// Migrations must outlive the runtime caps — an index build or an + /// `ACCESS EXCLUSIVE` wait routinely exceeds them, and startup treats a + /// migration failure as fatal — and the relaxed session must not survive + /// into the pool afterwards. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn migrations_ignore_runtime_timeouts_and_leak_no_relaxed_session() { + const TIGHT: &str = "50ms"; + + let admin = PgPool::connect(&admin_url().await) + .await + .expect("connect admin"); + let name = format!("migration_timeouts_{}", Uuid::new_v4().simple()); + sqlx::query(sqlx::AssertSqlSafe(format!("CREATE DATABASE {name}"))) + .execute(&admin) + .await + .expect("create scratch db"); + let base = admin_url().await; + let idx = base.rfind('/').expect("db url has a path segment"); + let scratch_url = format!("{}/{}", &base[..idx], name); + + // Far shorter than the migration suite needs, ample for a pooled query. + let db = Db::new(&DbConfig { + database_url: scratch_url, + max_connections: 2, + min_connections: 2, + statement_timeout: TIGHT.to_string(), + lock_timeout: TIGHT.to_string(), + ..DbConfig::default() + }) + .await + .expect("connect Db against the unmigrated scratch db"); + + db.migrate() + .await + .expect("migrations must not inherit the runtime caps"); + + // Hold every connection at once so a leaked relaxed session cannot hide + // behind a freshly dialed one. + let mut held = Vec::new(); + for _ in 0..2 { + let mut connection = db.pool.acquire().await.expect("acquire pooled connection"); + let statement_timeout: String = sqlx::query_scalar("SHOW statement_timeout") + .fetch_one(&mut *connection) + .await + .expect("SHOW statement_timeout"); + let lock_timeout: String = sqlx::query_scalar("SHOW lock_timeout") + .fetch_one(&mut *connection) + .await + .expect("SHOW lock_timeout"); + assert_eq!(statement_timeout, TIGHT); + assert_eq!(lock_timeout, TIGHT); + held.push(connection); + } + drop(held); + + drop_scratch_db(&admin, db.pool.clone(), &name).await; + } + /// Insert identical community + channel rows into a database so the same /// (community, channel) ids resolve in both writer and replica. async fn seed_community_channel( @@ -8317,7 +8427,8 @@ mod tests { let idx = base.rfind('/').expect("db url has a path segment"); let scratch_url = format!("{}/{}", &base[..idx], name); let db = Db::new(&DbConfig { - database_url: scratch_url, + database_url: scratch_url.clone(), + read_database_url: Some(scratch_url), max_connections: 2, ..DbConfig::default() }) @@ -8325,7 +8436,30 @@ mod tests { .expect("connect armed Db"); let cid = CommunityId::from_uuid(community); - // Perci nit: assert the effective session value, not the intent. + // Assert the effective session values, not only pool-builder intent. + let statement_timeout: String = sqlx::query_scalar("SHOW statement_timeout") + .fetch_one(&db.pool) + .await + .expect("SHOW statement_timeout"); + let lock_timeout: String = sqlx::query_scalar("SHOW lock_timeout") + .fetch_one(&db.pool) + .await + .expect("SHOW lock_timeout"); + assert_eq!(statement_timeout, RUNTIME_STATEMENT_TIMEOUT); + assert_eq!(lock_timeout, RUNTIME_LOCK_TIMEOUT); + + let read_pool = db.read_pool.as_ref().expect("read pool configured"); + let reader_statement_timeout: String = sqlx::query_scalar("SHOW statement_timeout") + .fetch_one(read_pool) + .await + .expect("SHOW reader statement_timeout"); + let reader_lock_timeout: String = sqlx::query_scalar("SHOW lock_timeout") + .fetch_one(read_pool) + .await + .expect("SHOW reader lock_timeout"); + assert_eq!(reader_statement_timeout, RUNTIME_STATEMENT_TIMEOUT); + assert_eq!(reader_lock_timeout, RUNTIME_LOCK_TIMEOUT); + let effective: String = sqlx::query_scalar("SHOW buzz.created_at_floor") .fetch_one(&db.pool) .await diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 6985916bba..c8cc935334 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -4,16 +4,27 @@ //! multi-tenant rewrite owns a clean consolidated `0001`; legacy single-tenant //! cutover/backfill is a separate operator script, not startup migration state. -use sqlx::PgPool; +use sqlx::{Connection, PgPool}; use crate::Result; static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("../../migrations"); /// Run all pending Buzz database migrations. +/// +/// DDL runs with the runtime `statement_timeout` and `lock_timeout` lifted. An +/// index build on a populated table, or an `ACCESS EXCLUSIVE` wait behind live +/// traffic, routinely outlasts the runtime caps — and because startup treats a +/// migration failure as fatal, inheriting them would turn a slow migration into +/// a relay that cannot boot. sqlx also takes its migration advisory lock as a +/// single waiting statement, so a second replica rolling out would be canceled +/// mid-wait rather than queueing behind the first. +/// +/// The connection is closed instead of returned to the pool: its session still +/// carries the lifted limits and must never serve traffic. pub async fn run_migrations(pool: &PgPool) -> Result<()> { reject_legacy_nip_rs_cardinality_ambiguity(pool).await?; - MIGRATOR.run(pool).await?; + run_migrator_without_runtime_timeouts(pool).await?; // The replica-fence proof (see `replica_fence`) requires the commit-time // `created_at` floor trigger from migration 0021 — correctly shaped — on // the `events` parent and every partition. `CREATE TABLE .. PARTITION OF` @@ -25,6 +36,36 @@ pub async fn run_migrations(pool: &PgPool) -> Result<()> { Ok(()) } +async fn run_migrator_without_runtime_timeouts(pool: &PgPool) -> Result<()> { + let mut connection = pool.acquire().await?; + lift_runtime_timeouts(&mut connection).await?; + let migrated = MIGRATOR.run(&mut *connection).await; + // Retire the connection either way; report the migration outcome first so a + // close failure cannot mask it. + let retired = retire_connection(connection).await; + migrated?; + retired?; + Ok(()) +} + +/// Remove both runtime limits from one connection's session. +async fn lift_runtime_timeouts(connection: &mut sqlx::PgConnection) -> Result<()> { + crate::apply_runtime_connection_timeouts( + connection, + crate::TIMEOUT_DISABLED, + crate::TIMEOUT_DISABLED, + ) + .await?; + Ok(()) +} + +/// Close a connection instead of returning it to the pool, so a session that +/// carries lifted limits can never serve traffic. +async fn retire_connection(connection: sqlx::pool::PoolConnection) -> Result<()> { + connection.detach().close().await?; + Ok(()) +} + /// Migration 0007 is checksum-frozen and predates exact NIP-RS tag-cardinality /// enforcement. A populated database still on 0001-0006 must not let 0007 /// irreversibly purge duplicate-tag history. Fail before sqlx starts its @@ -1094,6 +1135,69 @@ mod tests { .expect("read applied migrations") } + /// The migration connection must have both limits lifted, and it must not + /// come back to the pool afterwards — a session with no statement timeout + /// serving traffic is the failure this exemption trades against. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn migration_connection_is_unbounded_and_is_retired_not_reused() { + const TIGHT: &str = "50ms"; + + let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| TEST_DB_URL.to_owned()); + // One slot: a reused connection would be handed straight back below. + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .after_connect(|connection, _meta| { + Box::pin(crate::apply_runtime_connection_timeouts( + connection, TIGHT, TIGHT, + )) + }) + .connect(&database_url) + .await + .expect("connect to test DB"); + + let mut connection = pool.acquire().await.expect("acquire"); + assert_eq!( + show_timeout(&mut connection, "statement_timeout").await, + TIGHT + ); + + lift_runtime_timeouts(&mut connection) + .await + .expect("lift runtime timeouts"); + for setting in ["statement_timeout", "lock_timeout"] { + assert_eq!( + show_timeout(&mut connection, setting).await, + "0", + "{setting} must be lifted for the migrator" + ); + } + + retire_connection(connection) + .await + .expect("retire migration connection"); + + let mut fresh = pool.acquire().await.expect("re-acquire"); + for setting in ["statement_timeout", "lock_timeout"] { + assert_eq!( + show_timeout(&mut fresh, setting).await, + TIGHT, + "the pool must not hand out the relaxed migration session" + ); + } + drop(fresh); + pool.close().await; + } + + async fn show_timeout(connection: &mut sqlx::PgConnection, setting: &str) -> String { + sqlx::query_scalar(sqlx::AssertSqlSafe(format!("SHOW {setting}"))) + .fetch_one(connection) + .await + .unwrap_or_else(|error| panic!("SHOW {setting}: {error}")) + } + #[tokio::test] #[ignore = "requires Postgres"] async fn pre_0007_ambiguous_nip_rs_data_blocks_without_mutation_and_allows_retry() { diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 85a0ca2efe..fb03e6187e 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -81,6 +81,14 @@ pub struct Config { /// independently so reader capacity can be tuned against the replica's /// headroom without touching the writer pool. pub db_read_pool_size: Option, + /// Postgres `statement_timeout` for every runtime connection + /// (`BUZZ_DB_STATEMENT_TIMEOUT`, e.g. `45s`, `500ms`, `0` to disable). + /// Tunable so a backfill or an incident does not need a code change. + pub db_statement_timeout: String, + /// Postgres `lock_timeout` for every runtime connection + /// (`BUZZ_DB_LOCK_TIMEOUT`). Schema migrations always run with both limits + /// lifted — see `buzz_db::migration::run_migrations`. + pub db_lock_timeout: String, /// Public WebSocket URL of this relay, advertised in NIP-11. pub relay_url: String, /// Public WebSocket URL of the dedicated device-pairing relay, when configured. @@ -284,6 +292,37 @@ fn parse_bind_addr(raw: &str) -> Result { .map_err(|e| ConfigError::InvalidBindAddr(e.to_string())) } +/// Postgres accepts a timeout as an integer (milliseconds) with an optional +/// unit. Anything else is refused in favor of the default rather than failing +/// the config: a malformed value would break every `after_connect`, taking all +/// Postgres access with it, and a relay that keeps its documented default is a +/// better outcome than one that will not start. +fn pg_timeout_or_default(raw: Option<&str>, default: &str) -> String { + const UNITS: [&str; 6] = ["us", "ms", "s", "min", "h", "d"]; + + let candidate = raw.map(str::trim).filter(|value| !value.is_empty()); + let Some(candidate) = candidate else { + return default.to_string(); + }; + + let digits = candidate.chars().take_while(char::is_ascii_digit).count(); + let (magnitude, unit) = candidate.split_at(digits); + let unit = unit.trim(); + let valid = !magnitude.is_empty() + && (unit.is_empty() || UNITS.iter().any(|known| unit.eq_ignore_ascii_case(known))); + if valid { + candidate.to_string() + } else { + tracing::warn!( + value = candidate, + default, + "ignoring malformed Postgres timeout — expected an integer with an optional \ + us/ms/s/min/h/d unit" + ); + default.to_string() + } +} + fn positive_u64_from_env(name: &str, default: u64) -> Result { match std::env::var(name) { Ok(raw) => raw @@ -473,6 +512,15 @@ impl Config { .and_then(|v| v.parse::().ok()) .filter(|&v| v > 0); + let db_statement_timeout = pg_timeout_or_default( + std::env::var("BUZZ_DB_STATEMENT_TIMEOUT").ok().as_deref(), + buzz_db::RUNTIME_STATEMENT_TIMEOUT, + ); + let db_lock_timeout = pg_timeout_or_default( + std::env::var("BUZZ_DB_LOCK_TIMEOUT").ok().as_deref(), + buzz_db::RUNTIME_LOCK_TIMEOUT, + ); + let relay_url = std::env::var("RELAY_URL").unwrap_or_else(|_| "ws://localhost:3000".to_string()); @@ -938,6 +986,8 @@ impl Config { redis_pool_size, db_pool_size, db_read_pool_size, + db_statement_timeout, + db_lock_timeout, relay_url, pairing_relay_url, max_connections, @@ -1158,6 +1208,68 @@ mod tests { assert_eq!(junk, 50, "unparsable value must fall back to the default"); } + #[test] + fn pg_timeout_accepts_postgres_spellings_and_refuses_the_rest() { + for accepted in ["30s", "500ms", "0", "45S", "2min", " 10s "] { + assert_eq!( + pg_timeout_or_default(Some(accepted), "30s"), + accepted.trim(), + "{accepted} is a valid Postgres timeout" + ); + } + + // A rejected value must not reach Postgres: `after_connect` would fail + // for every connection, which is worse than the documented default. + for rejected in ["", " ", "soon", "30 seconds", "s30", "-5s", "30s;DROP"] { + assert_eq!( + pg_timeout_or_default(Some(rejected), "30s"), + "30s", + "{rejected:?} must fall back to the default" + ); + } + + assert_eq!(pg_timeout_or_default(None, "5s"), "5s"); + } + + #[test] + fn db_timeout_env_overrides_and_invalid_fallback() { + let _guard = ENV_MUTEX.lock().unwrap(); + let previous_statement = std::env::var_os("BUZZ_DB_STATEMENT_TIMEOUT"); + let previous_lock = std::env::var_os("BUZZ_DB_LOCK_TIMEOUT"); + + std::env::remove_var("BUZZ_DB_STATEMENT_TIMEOUT"); + std::env::remove_var("BUZZ_DB_LOCK_TIMEOUT"); + let defaults = Config::from_env().expect("config"); + assert_eq!( + defaults.db_statement_timeout, + buzz_db::RUNTIME_STATEMENT_TIMEOUT + ); + assert_eq!(defaults.db_lock_timeout, buzz_db::RUNTIME_LOCK_TIMEOUT); + + std::env::set_var("BUZZ_DB_STATEMENT_TIMEOUT", "90s"); + std::env::set_var("BUZZ_DB_LOCK_TIMEOUT", "250ms"); + let overridden = Config::from_env().expect("config"); + assert_eq!(overridden.db_statement_timeout, "90s"); + assert_eq!(overridden.db_lock_timeout, "250ms"); + + std::env::set_var("BUZZ_DB_STATEMENT_TIMEOUT", "half a minute"); + let junk = Config::from_env().expect("config"); + assert_eq!( + junk.db_statement_timeout, + buzz_db::RUNTIME_STATEMENT_TIMEOUT, + "a malformed value must not be handed to Postgres" + ); + + match previous_statement { + Some(value) => std::env::set_var("BUZZ_DB_STATEMENT_TIMEOUT", value), + None => std::env::remove_var("BUZZ_DB_STATEMENT_TIMEOUT"), + } + match previous_lock { + Some(value) => std::env::set_var("BUZZ_DB_LOCK_TIMEOUT", value), + None => std::env::remove_var("BUZZ_DB_LOCK_TIMEOUT"), + } + } + #[test] fn db_read_pool_size_env_override_and_invalid_fallback() { let _guard = ENV_MUTEX.lock().unwrap(); diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 799cf9cf60..9cc56cb0ad 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -35,6 +35,35 @@ fn buzz_auto_migrate_enabled(value: Option<&str>) -> bool { }) } +/// `after_connect` hook applying the configured runtime timeouts to the pools +/// the relay owns directly. The writer and replica pools get theirs from +/// `buzz_db::Db::new`; the audit and search pools are built here, so they need +/// the same treatment from the same config. +fn runtime_timeout_hook( + db_config: &DbConfig, +) -> impl for<'a> Fn( + &'a mut sqlx::PgConnection, + sqlx::pool::PoolConnectionMetadata, +) -> futures_util::future::BoxFuture<'a, Result<(), sqlx::Error>> + + Send + + Sync + + 'static { + let statement_timeout = db_config.statement_timeout.clone(); + let lock_timeout = db_config.lock_timeout.clone(); + move |connection, _meta| { + let statement_timeout = statement_timeout.clone(); + let lock_timeout = lock_timeout.clone(); + Box::pin(async move { + buzz_db::apply_runtime_connection_timeouts( + connection, + &statement_timeout, + &lock_timeout, + ) + .await + }) + } +} + /// Controls how many per-community gauge series the usage poller emits. /// /// Datadog cost is proportional to the number of unique time-series. With ~25 @@ -169,6 +198,8 @@ async fn main() -> anyhow::Result<()> { replica_read_max_age_ms: config.replica_read_max_age_ms, max_connections: config.db_pool_size, read_max_connections: config.db_read_pool_size, + statement_timeout: config.db_statement_timeout.clone(), + lock_timeout: config.db_lock_timeout.clone(), ..DbConfig::default() }; let db = Db::new(&db_config).await.map_err(|e| { @@ -350,6 +381,7 @@ async fn main() -> anyhow::Result<()> { let audit_pool = sqlx::postgres::PgPoolOptions::new() .max_connections(5) .min_connections(1) + .after_connect(runtime_timeout_hook(&db_config)) .connect(&config.database_url) .await .map_err(|e| anyhow::anyhow!("Audit DB connection failed: {e}"))?; @@ -404,6 +436,7 @@ async fn main() -> anyhow::Result<()> { .as_deref() .unwrap_or(&config.database_url); let search_pool = sqlx::postgres::PgPoolOptions::new() + .after_connect(runtime_timeout_hook(&db_config)) .connect(search_db_url) .await .map_err(|e| anyhow::anyhow!("Search DB connection failed: {e}"))?;