From 0e01357eb4b275cd2a74bce99b53c08da9fba953 Mon Sep 17 00:00:00 2001 From: Jordan Mecom Date: Fri, 31 Jul 2026 14:14:07 -0700 Subject: [PATCH 1/7] Bound runtime Postgres statements and lock waits Set a 30-second statement timeout and 5-second lock timeout on writer, reader, audit, and search pool connections. Verify the effective session settings alongside the existing writer guard. Co-authored-by: Jordan Mecom Signed-off-by: Jordan Mecom --- crates/buzz-db/src/lib.rs | 49 +++++++++++++++++++++++++++++------ crates/buzz-relay/src/main.rs | 6 +++++ 2 files changed, 47 insertions(+), 8 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9b26876747..4b2010b869 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -65,6 +65,27 @@ use uuid::Uuid; use buzz_core::{CommunityId, StoredEvent}; +/// Maximum time a runtime query may execute before Postgres cancels it. +pub const RUNTIME_STATEMENT_TIMEOUT: &str = "30s"; +/// Maximum time a runtime query may wait to acquire a lock. +pub const RUNTIME_LOCK_TIMEOUT: &str = "5s"; + +/// Apply the runtime safety limits shared by writer, reader, audit, and search +/// pools. +pub async fn apply_runtime_connection_timeouts( + connection: &mut PgConnection, +) -> std::result::Result<(), sqlx::Error> { + sqlx::query( + "SELECT set_config('statement_timeout', $1, false), \ + set_config('lock_timeout', $2, false)", + ) + .bind(RUNTIME_STATEMENT_TIMEOUT) + .bind(RUNTIME_LOCK_TIMEOUT) + .execute(connection) + .await?; + Ok(()) +} + fn event_replacement_lock_key( community_id: CommunityId, kind: i32, @@ -682,18 +703,19 @@ 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 { + options = options.after_connect(move |conn, _meta| { + Box::pin(async move { + apply_runtime_connection_timeouts(conn).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?) } @@ -8325,7 +8347,18 @@ 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 effective: String = sqlx::query_scalar("SHOW buzz.created_at_floor") .fetch_one(&db.pool) .await diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 34dc2dfcf8..987848c4de 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -350,6 +350,9 @@ async fn main() -> anyhow::Result<()> { let audit_pool = sqlx::postgres::PgPoolOptions::new() .max_connections(5) .min_connections(1) + .after_connect(|connection, _meta| { + Box::pin(buzz_db::apply_runtime_connection_timeouts(connection)) + }) .connect(&config.database_url) .await .map_err(|e| anyhow::anyhow!("Audit DB connection failed: {e}"))?; @@ -404,6 +407,9 @@ async fn main() -> anyhow::Result<()> { .as_deref() .unwrap_or(&config.database_url); let search_pool = sqlx::postgres::PgPoolOptions::new() + .after_connect(|connection, _meta| { + Box::pin(buzz_db::apply_runtime_connection_timeouts(connection)) + }) .connect(search_db_url) .await .map_err(|e| anyhow::anyhow!("Search DB connection failed: {e}"))?; From b1daf4d9ee85ad16b5c75dcb72f7e42b496f9680 Mon Sep 17 00:00:00 2001 From: Jordan Mecom Date: Fri, 31 Jul 2026 14:55:20 -0700 Subject: [PATCH 2/7] Apply Postgres timeouts to replica reads Co-authored-by: Jordan Mecom Signed-off-by: Jordan Mecom --- crates/buzz-db/src/lib.rs | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 4b2010b869..99af5da1fe 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -749,6 +749,9 @@ impl Db { .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(|connection, _meta| { + Box::pin(apply_runtime_connection_timeouts(connection)) + }) .connect_lazy(url)?) } @@ -8339,7 +8342,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() }) @@ -8359,6 +8363,18 @@ mod tests { 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 From f83ac1b48a4e796baadce8b96d048a8df71a4630 Mon Sep 17 00:00:00 2001 From: Eli Foster Date: Tue, 4 Aug 2026 12:07:14 -0700 Subject: [PATCH 3/7] Exempt migrations from the runtime timeouts and make them tunable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Schema migrations now run on a connection with both limits lifted. An index build on a populated table, or an ACCESS EXCLUSIVE wait behind live traffic, routinely outlasts the runtime caps, and startup treats a migration failure as fatal — so inheriting them turned a slow migration into a relay that cannot boot. sqlx also takes its migration advisory lock as one waiting statement, so a second replica rolling out would be canceled mid-wait instead of queueing. The connection is closed rather than returned to the pool, since its session still carries no limits. The two values move to DbConfig, wired to BUZZ_DB_STATEMENT_TIMEOUT and BUZZ_DB_LOCK_TIMEOUT with the previous constants as defaults, so a backfill or an incident does not need a code change. A malformed value falls back to the default with a warning: handing it to Postgres would fail every after_connect and take all database access with it. Co-Authored-By: Claude Opus 5 Signed-off-by: Eli Foster --- .env.example | 6 ++ crates/buzz-db/src/lib.rs | 42 +++++++++--- crates/buzz-db/src/migration.rs | 32 ++++++++- crates/buzz-relay/src/config.rs | 112 ++++++++++++++++++++++++++++++++ crates/buzz-relay/src/main.rs | 39 +++++++++-- 5 files changed, 215 insertions(+), 16 deletions(-) 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 99af5da1fe..55bff9759e 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -65,22 +65,27 @@ use uuid::Uuid; use buzz_core::{CommunityId, StoredEvent}; -/// Maximum time a runtime query may execute before Postgres cancels it. +/// Default maximum time a runtime query may execute before Postgres cancels it. pub const RUNTIME_STATEMENT_TIMEOUT: &str = "30s"; -/// Maximum time a runtime query may wait to acquire a lock. +/// 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. +/// 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(RUNTIME_STATEMENT_TIMEOUT) - .bind(RUNTIME_LOCK_TIMEOUT) + .bind(statement_timeout) + .bind(lock_timeout) .execute(connection) .await?; Ok(()) @@ -548,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 { @@ -565,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(), } } } @@ -703,9 +718,13 @@ 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)); + 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).await?; + 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)") @@ -743,14 +762,21 @@ 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(|connection, _meta| { - Box::pin(apply_runtime_connection_timeouts(connection)) + .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)?) } diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 37f54d0fa2..30eee9df35 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,23 @@ 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?; + crate::apply_runtime_connection_timeouts( + &mut connection, + crate::TIMEOUT_DISABLED, + crate::TIMEOUT_DISABLED, + ) + .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 closed = connection.detach().close().await; + migrated?; + closed?; + 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 diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index dd50973d03..6e2a9725aa 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,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 @@ -510,6 +549,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 +1024,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 +1246,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 987848c4de..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,9 +381,7 @@ async fn main() -> anyhow::Result<()> { let audit_pool = sqlx::postgres::PgPoolOptions::new() .max_connections(5) .min_connections(1) - .after_connect(|connection, _meta| { - Box::pin(buzz_db::apply_runtime_connection_timeouts(connection)) - }) + .after_connect(runtime_timeout_hook(&db_config)) .connect(&config.database_url) .await .map_err(|e| anyhow::anyhow!("Audit DB connection failed: {e}"))?; @@ -407,9 +436,7 @@ async fn main() -> anyhow::Result<()> { .as_deref() .unwrap_or(&config.database_url); let search_pool = sqlx::postgres::PgPoolOptions::new() - .after_connect(|connection, _meta| { - Box::pin(buzz_db::apply_runtime_connection_timeouts(connection)) - }) + .after_connect(runtime_timeout_hook(&db_config)) .connect(search_db_url) .await .map_err(|e| anyhow::anyhow!("Search DB connection failed: {e}"))?; From 821fad71bd2c83b5def7c85f72fb8e8927aca284 Mon Sep 17 00:00:00 2001 From: Eli Foster Date: Tue, 4 Aug 2026 12:23:50 -0700 Subject: [PATCH 4/7] Verify the migration exemption against Postgres MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Split the exemption into `lift_runtime_timeouts` and `retire_connection` so both halves are observable, and cover them: the migrator's connection reports both limits as `0`, and a single-slot pool hands out a freshly configured connection afterwards rather than the relaxed session. Both assertions were checked against a neutered implementation — a no-op lift and a `drop` instead of a close each fail the test — because a timing-based test does not discriminate here: on an empty database every migration statement finishes well inside the runtime cap, so it would pass with or without the exemption. Co-Authored-By: Claude Opus 5 Signed-off-by: Eli Foster --- crates/buzz-db/src/lib.rs | 59 +++++++++++++++++++++ crates/buzz-db/src/migration.rs | 90 ++++++++++++++++++++++++++++++--- 2 files changed, 142 insertions(+), 7 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 55bff9759e..37977a4541 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -6562,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( diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 30eee9df35..09a993518c 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -38,18 +38,31 @@ pub async fn run_migrations(pool: &PgPool) -> Result<()> { 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( - &mut connection, + connection, crate::TIMEOUT_DISABLED, crate::TIMEOUT_DISABLED, ) .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 closed = connection.detach().close().await; - migrated?; - closed?; + 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(()) } @@ -1149,6 +1162,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() { From 3c1a09a6c8a82e3137e3405b2bfc855aa60ca3d8 Mon Sep 17 00:00:00 2001 From: Eli Foster Date: Wed, 5 Aug 2026 14:20:27 -0700 Subject: [PATCH 5/7] fix(db): exempt the legacy NIP-RS preflight from runtime timeouts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `run_migrations` ran `reject_legacy_nip_rs_cardinality_ambiguity` on a pooled connection before acquiring the timeout-exempt migration connection. On a pre-0007 database that preflight is a full scan of `events` with per-row JSON expansion — exactly the databases large enough for it to be slow — so it inherited the runtime `statement_timeout`, and startup treats the error as fatal. The relay could not boot: the regression the migration exemption exists to prevent. Acquire and relax the connection first, then run the preflight and the migrator on it. The preflight still precedes sqlx's migration transaction, so an operator can inspect and repair before any DDL. Detach the connection from the pool before lifting its limits, which makes the exemption structurally cancellation-safe. A `PoolConnection` returns itself to the pool on drop, so a migration future cancelled after the lift handed an unbounded session to runtime traffic; a detached `PgConnection` closes on drop instead. Add a Postgres-backed regression test that seeds conforming kind-30078 rows until the preflight exceeds a tight cap on a pooled connection, then asserts `run_migrations` completes against the same capped pool and that the pool still hands out capped sessions afterwards. Row count is calibrated at runtime rather than hardcoded, so the test stays discriminating across hardware. Verified against a neutered implementation: moving the preflight back to the pool fails it with 57014. Signed-off-by: Eli Foster --- crates/buzz-db/src/migration.rs | 226 ++++++++++++++++++++++++++------ 1 file changed, 185 insertions(+), 41 deletions(-) diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 09a993518c..806742fad7 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -12,39 +12,60 @@ 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. +/// 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 closed instead of returned to the pool: its session still -/// carries the lifted limits and must never serve traffic. +/// 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?; - run_migrator_without_runtime_timeouts(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(()) } -async fn run_migrator_without_runtime_timeouts(pool: &PgPool) -> Result<()> { - let mut connection = pool.acquire().await?; +/// 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?; - 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(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(()) } @@ -59,28 +80,28 @@ async fn lift_runtime_timeouts(connection: &mut sqlx::PgConnection) -> Result<() 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 /// 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(()); @@ -124,7 +145,7 @@ async fn reject_legacy_nip_rs_cardinality_ambiguity(pool: &PgPool) -> Result<()> )\ )", ) - .fetch_one(pool) + .fetch_one(&mut *connection) .await?; if ambiguous { @@ -1185,15 +1206,13 @@ mod tests { .await .expect("connect to test DB"); - let mut connection = pool.acquire().await.expect("acquire"); - assert_eq!( - show_timeout(&mut connection, "statement_timeout").await, - TIGHT - ); + let mut pooled = pool.acquire().await.expect("acquire"); + assert_eq!(show_timeout(&mut pooled, "statement_timeout").await, TIGHT); + drop(pooled); - lift_runtime_timeouts(&mut connection) + let mut connection = exempt_migration_connection(&pool) .await - .expect("lift runtime timeouts"); + .expect("acquire exempt migration connection"); for setting in ["statement_timeout", "lock_timeout"] { assert_eq!( show_timeout(&mut connection, setting).await, @@ -1202,9 +1221,9 @@ mod tests { ); } - retire_connection(connection) - .await - .expect("retire migration connection"); + // 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"] { @@ -1295,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() { From b46f28cc036348631bba4a2381a28dbaed9cb279 Mon Sep 17 00:00:00 2001 From: Eli Foster Date: Wed, 5 Aug 2026 16:16:58 -0700 Subject: [PATCH 6/7] fix(db): range-check operator timeout values against Postgres MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pg_timeout_or_default only checked shape, so a well-formed but unstorable value ("999...9d") reached every pool's after_connect and Postgres rejected it there — failing all database access, the exact outcome the fallback exists to avoid. Parse the magnitude, convert it to milliseconds by unit, and fall back on overflow or anything above the int GUC ceiling. The ceiling lives in buzz-db as PG_TIMEOUT_MAX_MILLIS next to the timeouts it bounds, pinned to the live server by a Postgres-backed test so it can drift in neither direction. Co-Authored-By: Claude Opus 5 Signed-off-by: Eli Foster --- .env.example | 4 +- crates/buzz-db/src/lib.rs | 68 ++++++++++++++++++ crates/buzz-relay/src/config.rs | 120 +++++++++++++++++++++++++++----- 3 files changed, 172 insertions(+), 20 deletions(-) diff --git a/.env.example b/.env.example index e81ea15895..94b7bf5d12 100644 --- a/.env.example +++ b/.env.example @@ -40,7 +40,9 @@ REDIS_URL=redis://localhost:6379 # 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. +# 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 diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 37977a4541..4019d97d3a 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -71,6 +71,13 @@ pub const RUNTIME_STATEMENT_TIMEOUT: &str = "30s"; 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 @@ -8522,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-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 6e2a9725aa..b0d105a3c1 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -310,14 +310,30 @@ 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 { + "us" => Some((value + 500) / 1_000), + "" | "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. 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. +/// 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 { - 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(); @@ -325,19 +341,29 @@ fn pg_timeout_or_default(raw: Option<&str>, default: &str) -> 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() + 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() + } } } @@ -1269,6 +1295,47 @@ mod tests { 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", + ] { + 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(); @@ -1298,6 +1365,21 @@ mod tests { "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"), From 637a3231c5684d482aefb10999dbdc276db30f14 Mon Sep 17 00:00:00 2001 From: Eli Foster Date: Wed, 5 Aug 2026 16:43:12 -0700 Subject: [PATCH 7/7] fix(relay): avoid overflow when rounding microsecond timeouts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pg_timeout_millis rounded `us` values with `(value + 500) / 1_000`, which overflows for the top 500 representable u128 magnitudes — a debug panic during config load, or in release a wrapped near-zero millisecond value that slips past the range check and hands the original gigantic string to every pool's after_connect. Round without the intermediate, and cover u128::MAX and the first overflowing value in the fallback test. Co-Authored-By: Claude Opus 5 Signed-off-by: Eli Foster --- crates/buzz-relay/src/config.rs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index b0d105a3c1..41ea4e22cf 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -316,7 +316,9 @@ fn parse_bind_addr(raw: &str) -> Result { fn pg_timeout_millis(magnitude: &str, unit: &str) -> Option { let value = magnitude.parse::().ok()?; match unit { - "us" => Some((value + 500) / 1_000), + // 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), @@ -1310,6 +1312,11 @@ mod tests { // 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"),