diff --git a/.env.example b/.env.example index b9bfcada0e..94b7bf5d12 100644 --- a/.env.example +++ b/.env.example @@ -38,6 +38,14 @@ 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. Postgres stores both as int milliseconds, so the +# ceiling is 2147483647ms (~24 days); anything above it, or otherwise malformed, +# falls back to the default. 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..4019d97d3a 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -65,6 +65,39 @@ 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"; +/// `statement_timeout` and `lock_timeout` are `int` GUCs measured in +/// milliseconds, so Postgres refuses anything larger regardless of the unit it +/// is spelled with. Callers building a [`DbConfig`] from operator input must +/// range-check against this: [`apply_runtime_connection_timeouts`] runs on every +/// pooled connection, so an unusable value fails all database access. +/// `pg_timeout_max_millis_matches_postgres` pins it to the live server. +pub const PG_TIMEOUT_MAX_MILLIS: u128 = i32::MAX as u128; + +/// 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 +560,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 +585,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 +725,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 +769,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 +6569,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 +8434,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 +8443,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 @@ -8388,6 +8529,67 @@ mod tests { db.pool.close().await; } + /// [`PG_TIMEOUT_MAX_MILLIS`] is the bound callers range-check operator input + /// against, so it must be the server's real bound: too high and an accepted + /// value still fails every `after_connect`; too low and we reject settings + /// Postgres would have taken. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn pg_timeout_max_millis_matches_postgres() { + let mut conn = PgConnection::connect(&admin_url().await) + .await + .expect("connect"); + + let boundary = PG_TIMEOUT_MAX_MILLIS.to_string(); + apply_runtime_connection_timeouts(&mut conn, &boundary, &boundary) + .await + .expect("PG_TIMEOUT_MAX_MILLIS must be settable"); + + // Same instant spelled in a coarser unit — the conversion the caller's + // range check performs must land inside the range too. + let in_seconds = (PG_TIMEOUT_MAX_MILLIS / 1_000).to_string(); + apply_runtime_connection_timeouts( + &mut conn, + &format!("{in_seconds}s"), + &format!("{in_seconds}s"), + ) + .await + .expect("the boundary in seconds must be settable"); + + for over in [ + (PG_TIMEOUT_MAX_MILLIS + 1).to_string(), + format!("{}ms", PG_TIMEOUT_MAX_MILLIS + 1), + format!("{}s", PG_TIMEOUT_MAX_MILLIS / 1_000 + 1), + "999999999999999999999999999999999999999999d".to_string(), + ] { + let err = apply_runtime_connection_timeouts(&mut conn, &over, RUNTIME_LOCK_TIMEOUT) + .await + .expect_err(&format!("{over} must be rejected by Postgres")); + let code = match &err { + sqlx::Error::Database(db) => db.code().map(|c| c.to_string()), + other => panic!("expected a database error, got {other:?}"), + }; + assert_eq!( + code.as_deref(), + Some("22023"), + "{over}: expected invalid_parameter_value" + ); + } + + // A rejected `set_config` leaves the session usable, so the failure mode + // is a poisoned pool of unbounded sessions only if the caller ignores it. + let statement_timeout: String = sqlx::query_scalar("SHOW statement_timeout") + .fetch_one(&mut conn) + .await + .expect("SHOW statement_timeout"); + assert_ne!( + statement_timeout, "0", + "a rejected value must not silently disable the limit" + ); + + conn.close().await.expect("close"); + } + /// `spawn_fence_probe` must verify the floor guard before letting the /// probe run — catalog shape AND observed behavior — and refuse on /// sabotage. This is the production gate for a relay running with diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 37f54d0fa2..806742fad7 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -4,42 +4,104 @@ //! 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. +/// +/// Everything that can be slow on a populated database — the legacy NIP-RS +/// preflight scan and the migrator itself — runs on one dedicated connection +/// with the runtime `statement_timeout` and `lock_timeout` lifted. An index +/// build on a populated table, an `ACCESS EXCLUSIVE` wait behind live traffic, +/// or a full scan of `events` with per-row JSON expansion 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 detached from the pool before its limits are lifted and +/// closed afterwards: a session with no caps must never serve traffic, and +/// detaching first means even a cancellation mid-migration drops it rather than +/// releasing it back. pub async fn run_migrations(pool: &PgPool) -> Result<()> { - reject_legacy_nip_rs_cardinality_ambiguity(pool).await?; - MIGRATOR.run(pool).await?; + let mut connection = exempt_migration_connection(pool).await?; + let migrated = migrate_on_exempt_connection(&mut connection).await; + // Close the connection either way; report the migration outcome first so a + // close failure cannot mask it. + let closed = connection.close().await; + migrated?; + closed?; // 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` // clones parent triggers, but a partition attached with `ATTACH // PARTITION` or created by an older code path would silently escape the // guard, so migration fails closed if any is missing. (The fence probe - // re-runs this same check at startup on non-migrating relays.) + // re-runs this same check at startup on non-migrating relays.) Catalog-only + // and bounded by the number of partitions, so the runtime caps are fine. crate::replica_fence::verify_floor_guard_catalog(pool).await?; Ok(()) } +/// Take a connection out of the pool for good, then lift its runtime limits. +/// +/// Detaching before lifting is what makes the exemption structurally +/// cancellation-safe: a [`sqlx::pool::PoolConnection`] returns itself to the +/// pool on drop, so a future cancelled after the lift would hand an unbounded +/// session to runtime traffic. A detached [`sqlx::PgConnection`] closes on drop +/// instead. +async fn exempt_migration_connection(pool: &PgPool) -> Result { + let mut connection = pool.acquire().await?.detach(); + lift_runtime_timeouts(&mut connection).await?; + Ok(connection) +} + +/// The preflight *and* the migrator, both on the exempt connection. The +/// preflight stays ahead of sqlx's migration transaction so an operator can +/// still inspect and repair before any DDL runs. +async fn migrate_on_exempt_connection(connection: &mut sqlx::PgConnection) -> Result<()> { + reject_legacy_nip_rs_cardinality_ambiguity(connection).await?; + MIGRATOR.run(connection).await?; + 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(()) +} + /// 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 /// migration transaction so an operator can inspect and repair those rows. -async fn reject_legacy_nip_rs_cardinality_ambiguity(pool: &PgPool) -> Result<()> { +/// +/// Takes a connection, not the pool: the scan below is a full pass over +/// `events` with per-row JSON expansion on exactly the databases that are large +/// enough to matter, so it must run on the timeout-exempt migration connection +/// (see [`run_migrations`]) or a slow scan becomes a fatal startup failure. +async fn reject_legacy_nip_rs_cardinality_ambiguity( + connection: &mut sqlx::PgConnection, +) -> Result<()> { let migrations_table: Option = sqlx::query_scalar("SELECT to_regclass('_sqlx_migrations')::text") - .fetch_one(pool) + .fetch_one(&mut *connection) .await?; if migrations_table.is_none() { return Ok(()); } let applied: Option = sqlx::query_scalar("SELECT max(version) FROM _sqlx_migrations WHERE success") - .fetch_one(pool) + .fetch_one(&mut *connection) .await?; if applied.is_none_or(|version| version >= 7) { return Ok(()); @@ -83,7 +145,7 @@ async fn reject_legacy_nip_rs_cardinality_ambiguity(pool: &PgPool) -> Result<()> )\ )", ) - .fetch_one(pool) + .fetch_one(&mut *connection) .await?; if ambiguous { @@ -1121,6 +1183,67 @@ 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 pooled = pool.acquire().await.expect("acquire"); + assert_eq!(show_timeout(&mut pooled, "statement_timeout").await, TIGHT); + drop(pooled); + + let mut connection = exempt_migration_connection(&pool) + .await + .expect("acquire exempt migration connection"); + for setting in ["statement_timeout", "lock_timeout"] { + assert_eq!( + show_timeout(&mut connection, setting).await, + "0", + "{setting} must be lifted for the migrator" + ); + } + + // Dropping without closing stands in for a cancelled migration future: + // the connection is detached, so the pool must still not see it. + drop(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() { @@ -1191,6 +1314,131 @@ mod tests { assert_eq!(applied_versions(&pool).await.last().copied(), Some(27)); } + /// A pool whose connections carry `timeout` for both runtime limits. + async fn connect_capped_pool(timeout: &'static str) -> PgPool { + 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()); + + sqlx::postgres::PgPoolOptions::new() + .max_connections(2) + .after_connect(move |connection, _meta| { + Box::pin(crate::apply_runtime_connection_timeouts( + connection, timeout, timeout, + )) + }) + .connect(&database_url) + .await + .expect("connect capped test pool") + } + + fn is_statement_timeout(error: &crate::DbError) -> bool { + // 57014 = query_canceled, which is what `statement_timeout` raises. + matches!(error, crate::DbError::Sqlx(sqlx::Error::Database(db)) + if db.code().as_deref() == Some("57014")) + } + + /// Conforming (never ambiguous) kind-30078 read-state rows, so the preflight + /// scan cannot short-circuit on an early match and has to read them all. + async fn seed_conforming_read_state_rows( + pool: &PgPool, + community_id: uuid::Uuid, + offset: i64, + count: i64, + ) { + sqlx::query( + "INSERT INTO events \ + (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, d_tag) \ + SELECT $1, \ + decode(md5('id' || i::text) || md5('id2' || i::text), 'hex'), \ + decode(md5('pk' || i::text) || md5('pk2' || i::text), 'hex'), \ + NOW(), 30078, \ + jsonb_build_array( \ + jsonb_build_array('d', 'read-state:' || md5(i::text)), \ + jsonb_build_array('t', 'read-state')), \ + 'conforming', \ + decode(repeat(md5(i::text), 4), 'hex'), \ + NOW(), \ + 'read-state:' || md5(i::text) \ + FROM generate_series($2::bigint, $3::bigint) AS i", + ) + .bind(community_id) + .bind(offset + 1) + .bind(offset + count) + .execute(pool) + .await + .expect("seed conforming read-state rows"); + } + + /// The legacy NIP-RS preflight must run on the timeout-exempt migration + /// connection, not on a pooled one. It scans all of `events` with per-row + /// JSON expansion on exactly the databases big enough for that to be slow, + /// and startup treats the error as fatal — so inheriting the runtime + /// `statement_timeout` there is a relay that cannot boot. The fresh-database + /// migration test returns before this scan and cannot catch the ordering. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn legacy_preflight_scan_outlives_the_runtime_statement_timeout() { + const TIGHT: &str = "50ms"; + const SEED_BATCH: i64 = 20_000; + const MAX_SEEDED: i64 = 200_000; + + let pool = connect_test_pool().await; + reset_public_schema(&pool).await; + MIGRATOR + .run_to(6, &pool) + .await + .expect("apply migrations 1-6"); + + let community_id = uuid::Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_id) + .bind(format!("preflight-{}.example", community_id.simple())) + .execute(&pool) + .await + .expect("insert community"); + + let capped = connect_capped_pool(TIGHT).await; + + // Grow the table until the preflight genuinely exceeds the cap on a + // pooled connection. Scan cost is hardware-dependent, so calibrate + // instead of hardcoding a row count that is slow on one machine only. + let mut seeded = 0; + loop { + seed_conforming_read_state_rows(&pool, community_id, seeded, SEED_BATCH).await; + seeded += SEED_BATCH; + + let mut capped_connection = capped.acquire().await.expect("acquire capped connection"); + match reject_legacy_nip_rs_cardinality_ambiguity(&mut capped_connection).await { + Err(error) if is_statement_timeout(&error) => break, + Err(error) => panic!("preflight failed for an unrelated reason: {error}"), + Ok(()) => assert!( + seeded < MAX_SEEDED, + "preflight still finished inside {TIGHT} with {seeded} rows; \ + the test can no longer prove the exemption is load-bearing" + ), + } + } + + // Same cap, same data — the exemption is the only difference. + run_migrations(&capped) + .await + .expect("migrations must not inherit the runtime caps during the legacy preflight"); + assert_eq!(applied_versions(&pool).await.last().copied(), Some(26)); + + // And the caps are still in force for traffic afterwards. + let mut runtime = capped.acquire().await.expect("acquire after migration"); + for setting in ["statement_timeout", "lock_timeout"] { + assert_eq!( + show_timeout(&mut runtime, setting).await, + TIGHT, + "the pool must not hand out a relaxed session after migrating" + ); + } + drop(runtime); + capped.close().await; + } + #[tokio::test] #[ignore = "requires Postgres"] async fn populated_upgrade_preserves_search_policy_except_for_push_leases() { diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index dd50973d03..41ea4e22cf 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -99,6 +99,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. @@ -302,6 +310,65 @@ fn parse_bind_addr(raw: &str) -> Result { .map_err(|e| ConfigError::InvalidBindAddr(e.to_string())) } +/// Convert a Postgres timeout magnitude and unit to milliseconds, mirroring the +/// rounding Postgres applies to sub-millisecond `us` values. `None` means the +/// spelling is not something Postgres would accept. +fn pg_timeout_millis(magnitude: &str, unit: &str) -> Option { + let value = magnitude.parse::().ok()?; + match unit { + // Round half up without the `value + 500` intermediate, which overflows + // for the top 500 representable microsecond values. + "us" => Some(value / 1_000 + u128::from(value % 1_000 >= 500)), + "" | "ms" => Some(value), + "s" => value.checked_mul(1_000), + "min" => value.checked_mul(60_000), + "h" => value.checked_mul(3_600_000), + "d" => value.checked_mul(86_400_000), + _ => None, + } +} + +/// Postgres accepts a timeout as an integer (milliseconds) with an optional +/// unit, bounded by `i32::MAX` ms once converted. Anything else is refused in +/// favor of the default rather than failing the config: an unusable 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. Magnitude is range-checked, not just shape-checked — `999...9d` is a +/// well-formed spelling that Postgres still rejects. +fn pg_timeout_or_default(raw: Option<&str>, default: &str) -> String { + 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().to_ascii_lowercase(); + + match pg_timeout_millis(magnitude, &unit) { + Some(millis) if millis <= buzz_db::PG_TIMEOUT_MAX_MILLIS => candidate.to_string(), + Some(millis) => { + tracing::warn!( + value = candidate, + millis = %millis, + max_millis = %buzz_db::PG_TIMEOUT_MAX_MILLIS, + default, + "ignoring out-of-range Postgres timeout — Postgres stores it as int milliseconds" + ); + default.to_string() + } + None => { + 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 @@ -510,6 +577,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()); @@ -976,6 +1052,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, @@ -1196,6 +1274,129 @@ 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 pg_timeout_refuses_magnitudes_postgres_cannot_store() { + // Well-formed spellings whose millisecond value exceeds the int GUC + // range. Postgres rejects these in `set_config`, which would fail every + // pool's `after_connect`. + for rejected in [ + "2147483648", + "2147483648ms", + "2147484s", + "35792min", + "597h", + "25d", + // Wider than any integer type — must fall back, not overflow. + "999999999999999999999999999999999999999999d", + "99999999999999999999999999999999999999999999999999", + // Parses as u128, so unlike the two above it reaches the unit + // conversion — where rounding must not overflow on the way to the + // range check. + &format!("{}us", u128::MAX), + &format!("{}us", u128::MAX - 499), + ] { + assert_eq!( + pg_timeout_or_default(Some(rejected), "30s"), + "30s", + "{rejected:?} is out of range for Postgres and must fall back" + ); + } + + // The boundary itself, and the same instant in every unit, stay valid. + for accepted in [ + "2147483647", + "2147483647ms", + "2147483647000us", + "2147483s", + "35791min", + "596h", + "24d", + ] { + assert_eq!( + pg_timeout_or_default(Some(accepted), "30s"), + accepted, + "{accepted} is inside the Postgres range" + ); + } + } + + #[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" + ); + + // Shape-valid but far outside the int millisecond GUC range: Postgres + // would reject it in every pool's `after_connect`. + std::env::set_var( + "BUZZ_DB_STATEMENT_TIMEOUT", + "999999999999999999999999999999999999999999d", + ); + std::env::set_var("BUZZ_DB_LOCK_TIMEOUT", "25d"); + let out_of_range = Config::from_env().expect("config"); + assert_eq!( + out_of_range.db_statement_timeout, + buzz_db::RUNTIME_STATEMENT_TIMEOUT, + "an out-of-range value must not be handed to Postgres" + ); + assert_eq!(out_of_range.db_lock_timeout, buzz_db::RUNTIME_LOCK_TIMEOUT); + + 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 34dc2dfcf8..c962b549e0 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}"))?;