diff --git a/.env.example b/.env.example index 0f7bbba6f13..f5210c31e2c 100644 --- a/.env.example +++ b/.env.example @@ -38,6 +38,22 @@ REDIS_URL=redis://localhost:6379 # READ_DATABASE_URL is set, reader (default 50). # BUZZ_DB_POOL_SIZE=50 +# Writer-session Postgres timeouts for buzz-db-backed pools and the relay audit +# pool, all in milliseconds; 0 disables. The separately deployed push gateway +# owns its own database and session policy and does not consume these knobs. +# lock_timeout: fail a statement that waits this long on any lock instead of +# parking behind a wedged holder (default 5000). +# BUZZ_DB_LOCK_TIMEOUT_MS=5000 +# idle_in_transaction_session_timeout: reap sessions idle inside an open +# transaction — bounds how long a wedged client can hold locks (default 60000). +# BUZZ_DB_IDLE_TXN_TIMEOUT_MS=60000 +# statement_timeout: cap any single statement's runtime. Off by default — +# startup migrations/backfills legitimately run long statements. Warning: a +# pathologically low value (e.g. 1) also times out the relay's own +# connection-setup statements and can prevent any DB connection from +# establishing; keep it comfortably above normal query latency. +# BUZZ_DB_STATEMENT_TIMEOUT_MS=0 + # ----------------------------------------------------------------------------- # Typesense (search) # ----------------------------------------------------------------------------- diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a832c0a0aff..31eece1f4f9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -371,6 +371,7 @@ jobs: -p buzz-relay \ -p buzz-test-client \ --lib \ + --bin buzz-relay \ --test e2e_event_reminder \ --archive-file target/ci/backend-integration-tests.tar.zst - name: Save relay artifacts cache @@ -744,6 +745,30 @@ jobs: --run-ignored ignored-only env: DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz + - name: Writer session timeout guardrails + # Proves the buzz-db writer-pool session timeouts (lock_timeout / + # idle_in_transaction_session_timeout / statement_timeout) install + # through Db::new(), bound contended lock waits with SQLSTATE 55P03, + # and that Db::migrate()'s schema-destruction advisory lock is exempt + # from lock_timeout. + run: | + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E 'package(buzz-db) and test(session_timeouts_install_through_db_new_and_bound_lock_waits)' \ + --run-ignored ignored-only + env: + DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz + - name: Audit writer session timeout guardrails + # Keep this separate from the buzz-db guard so nextest fails when the + # relay binary test is absent from the archive instead of silently + # succeeding because the buzz-db half of a combined filter matched. + run: | + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E 'package(buzz-relay) and test(audit_writer_pool_installs_timeouts_and_bounds_advisory_lock_waits)' \ + --run-ignored ignored-only + env: + DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz - name: Upload relay log if: failure() uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 diff --git a/crates/buzz-admin/src/main.rs b/crates/buzz-admin/src/main.rs index 42a7de84f7c..19a3b1d9d48 100644 --- a/crates/buzz-admin/src/main.rs +++ b/crates/buzz-admin/src/main.rs @@ -433,10 +433,13 @@ async fn connect_member_services() -> Result<(Db, Arc, Keys)> { async fn connect_db() -> Result { let db_url = std::env::var("DATABASE_URL") .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string()); - let db = Db::new(&DbConfig { - database_url: db_url, - ..DbConfig::default() - }) + let db = Db::new( + &DbConfig { + database_url: db_url, + ..DbConfig::default() + } + .with_session_timeouts_from_env(), + ) .await?; Ok(db) } diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 3ff230f9503..d2b9fbd504a 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -553,6 +553,22 @@ 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, + /// Session `lock_timeout` in milliseconds for writer connections (env + /// `BUZZ_DB_LOCK_TIMEOUT_MS`). A statement that waits longer than this + /// on any lock errors out instead of parking — the fast-fail that keeps + /// one wedged lock holder from queueing the whole fleet behind it. + /// `0` disables the timeout (Postgres semantics). + pub lock_timeout_ms: u64, + /// Session `idle_in_transaction_session_timeout` in milliseconds for + /// writer connections (env `BUZZ_DB_IDLE_TXN_TIMEOUT_MS`). Reaps + /// sessions that opened a transaction (possibly holding locks) and then + /// went idle — e.g. a client wedged mid-boot. `0` disables. + pub idle_txn_timeout_ms: u64, + /// Session `statement_timeout` in milliseconds for writer connections + /// (env `BUZZ_DB_STATEMENT_TIMEOUT_MS`). `0` (the default) disables it: + /// startup migrations and backfills legitimately run long statements, + /// so this is opt-in for deployers who know their workload. + pub statement_timeout_ms: u64, } impl Default for DbConfig { @@ -570,10 +586,57 @@ impl Default for DbConfig { max_lifetime_secs: 1800, idle_timeout_secs: 600, replica_read_max_age_ms: 0, + lock_timeout_ms: DEFAULT_LOCK_TIMEOUT_MS, + idle_txn_timeout_ms: DEFAULT_IDLE_TXN_TIMEOUT_MS, + statement_timeout_ms: 0, } } } +/// Default writer `lock_timeout` (ms). Five seconds is far above any healthy +/// lock wait on the relay's hot paths, yet turns a wedged relation-lock +/// holder from a fleet-wide stall into per-statement errors that surface in +/// logs and retry naturally. +pub const DEFAULT_LOCK_TIMEOUT_MS: u64 = 5_000; + +/// Default writer `idle_in_transaction_session_timeout` (ms). One minute: +/// no legitimate relay transaction idles anywhere near this long, and it +/// bounds how long a wedged client can hold locks from an open transaction. +pub const DEFAULT_IDLE_TXN_TIMEOUT_MS: u64 = 60_000; + +impl DbConfig { + /// Overlay the writer session-timeout knobs from the environment: + /// `BUZZ_DB_LOCK_TIMEOUT_MS`, `BUZZ_DB_IDLE_TXN_TIMEOUT_MS`, + /// `BUZZ_DB_STATEMENT_TIMEOUT_MS`. + /// + /// This lives in `buzz-db` — not per-binary config — so every + /// `buzz-db`-backed writer (`buzz-relay`, `buzz-admin`, `buzz-deletion`, + /// and the relay audit pool) honors the same operator knobs; parsing it + /// in one binary would make the documented override silently no-op in + /// the others. The separately deployed push gateway owns a dedicated + /// database and session policy and is outside this configuration's scope. + /// + /// Absent or unparseable values keep the current (default) settings. + /// An explicit `0` disables that timeout (Postgres semantics), so zero + /// must pass through rather than being filtered like the pool-size env + /// knobs. + pub fn with_session_timeouts_from_env(mut self) -> Self { + fn parse(key: &str) -> Option { + std::env::var(key).ok().and_then(|v| v.parse::().ok()) + } + if let Some(v) = parse("BUZZ_DB_LOCK_TIMEOUT_MS") { + self.lock_timeout_ms = v; + } + if let Some(v) = parse("BUZZ_DB_IDLE_TXN_TIMEOUT_MS") { + self.idle_txn_timeout_ms = v; + } + if let Some(v) = parse("BUZZ_DB_STATEMENT_TIMEOUT_MS") { + self.statement_timeout_ms = v; + } + self + } +} + /// Community host-map row returned by [`Db::lookup_community_by_host`]. #[derive(Debug, Clone, PartialEq, Eq)] pub struct CommunityRecord { @@ -675,7 +738,7 @@ impl Db { /// `buzz.created_at_floor` GUC — this is what makes the replica fence /// proof hold for every insert path that goes through this pool. pub async fn new(config: &DbConfig) -> Result { - let pool = Self::connect_pool(config, &config.database_url).await?; + let pool = Self::connect_writer_pool(config).await?; let read_max_connections = config .read_max_connections .unwrap_or(config.max_connections); @@ -695,25 +758,54 @@ impl Db { }) } - /// Connect the writer pool with all session-level safety premises. + /// Connect a writer pool with all session-level safety premises. /// /// SQLx stores one `after_connect` hook, so the floor guard and transaction /// isolation assertion must remain in this single closure. Registering a /// second hook replaces the first and silently disarms the floor trigger. - async fn connect_pool(config: &DbConfig, url: &str) -> Result { + /// Additional writer pools, such as the relay audit pool, must use this + /// constructor rather than raw [`PgPoolOptions`] so they inherit the same + /// timeout, floor-guard, and isolation policy as [`Db::new`]. + pub async fn connect_writer_pool(config: &DbConfig) -> Result { + let lock_timeout_ms = config.lock_timeout_ms; + let idle_txn_timeout_ms = config.idle_txn_timeout_ms; + let statement_timeout_ms = config.statement_timeout_ms; let options = PgPoolOptions::new() .max_connections(config.max_connections) .min_connections(config.min_connections) .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)) - .after_connect(|conn, _meta| { + .after_connect(move |conn, _meta| { Box::pin(async move { // `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(&mut *conn) .await?; + // Session timeouts (0 = disabled, Postgres semantics). + // Scope is per-knob and worth being precise about: + // `lock_timeout` fails the *waiting* statement fast so a + // wedged holder produces visible errors instead of an + // unbounded queue; it never cancels the holder itself. + // `idle_in_transaction_session_timeout` is what reaps a + // wedged holder, and only while it sits idle inside an + // open transaction. An actively-executing wedged holder + // is bounded only by `statement_timeout` (off by + // default). Bare integers are milliseconds for all three + // GUCs. The migration/schema-destruction path resets + // lock/statement timeouts on its dedicated connection — + // see `with_exclusive_schema_destruction_lock`. + sqlx::query( + "SELECT set_config('lock_timeout', $1, false), \ + set_config('idle_in_transaction_session_timeout', $2, false), \ + set_config('statement_timeout', $3, false)", + ) + .bind(lock_timeout_ms.to_string()) + .bind(idle_txn_timeout_ms.to_string()) + .bind(statement_timeout_ms.to_string()) + .execute(&mut *conn) + .await?; let isolation: String = sqlx::query_scalar("SHOW transaction_isolation") .fetch_one(&mut *conn) .await?; @@ -728,7 +820,7 @@ impl Db { Ok(()) }) }); - Ok(options.connect(url).await?) + Ok(options.connect(&config.database_url).await?) } /// Reader acquire timeout — deliberately far below the writer's @@ -9127,21 +9219,27 @@ mod tests { #[test] fn writer_pool_safety_hook_is_single_and_composed() { let source = include_str!("lib.rs"); - let connect_pool = source - .split("async fn connect_pool") + let connect_writer_pool = source + .split("async fn connect_writer_pool") .nth(1) .and_then(|tail| tail.split("const READER_ACQUIRE_TIMEOUT").next()) - .expect("connect_pool source block"); + .expect("connect_writer_pool source block"); assert_eq!( - connect_pool.matches(".after_connect(").count(), + connect_writer_pool.matches(".after_connect(").count(), 1, "SQLx replaces after_connect hooks; writer safety must use exactly one" ); - assert!(connect_pool.contains("buzz.created_at_floor")); - assert!(connect_pool.contains("SHOW transaction_isolation")); - assert!(!connect_pool.contains("arm_floor_guard")); - assert!(!connect_pool.contains("_arm_floor_guard")); - assert!(!connect_pool.contains("allow(unused_variables)")); + assert!(connect_writer_pool.contains("buzz.created_at_floor")); + assert!(connect_writer_pool.contains("SHOW transaction_isolation")); + assert!( + connect_writer_pool.contains("'lock_timeout'") + && connect_writer_pool.contains("'idle_in_transaction_session_timeout'") + && connect_writer_pool.contains("'statement_timeout'"), + "session timeouts must be applied inside the single writer hook" + ); + assert!(!connect_writer_pool.contains("arm_floor_guard")); + assert!(!connect_writer_pool.contains("_arm_floor_guard")); + assert!(!connect_writer_pool.contains("allow(unused_variables)")); let reader_doc = source .split("fn connect_read_pool") @@ -9150,7 +9248,7 @@ mod tests { .expect("reader pool documentation"); assert!(reader_doc.contains("replica sessions are")); assert!(reader_doc.contains("read-only")); - assert!(!reader_doc.contains("Db::connect_pool")); + assert!(!reader_doc.contains("Db::connect_writer_pool")); } #[tokio::test] @@ -9194,6 +9292,176 @@ mod tests { .expect("drop isolation test database"); } + /// `with_session_timeouts_from_env` must apply present values (including + /// an explicit `0` = disable) and keep defaults for absent or junk input. + /// Serialized against other env-mutating tests via a process-wide lock. + #[test] + fn session_timeout_env_overlay_zero_passthrough_and_invalid_fallback() { + static ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + let _guard = ENV_LOCK.lock().unwrap(); + let keys = [ + "BUZZ_DB_LOCK_TIMEOUT_MS", + "BUZZ_DB_IDLE_TXN_TIMEOUT_MS", + "BUZZ_DB_STATEMENT_TIMEOUT_MS", + ]; + let previous: Vec<_> = keys.iter().map(std::env::var_os).collect(); + let read = |config: DbConfig| { + ( + config.lock_timeout_ms, + config.idle_txn_timeout_ms, + config.statement_timeout_ms, + ) + }; + + for key in keys { + std::env::remove_var(key); + } + let unset = read(DbConfig::default().with_session_timeouts_from_env()); + + std::env::set_var("BUZZ_DB_LOCK_TIMEOUT_MS", "2000"); + std::env::set_var("BUZZ_DB_IDLE_TXN_TIMEOUT_MS", "30000"); + std::env::set_var("BUZZ_DB_STATEMENT_TIMEOUT_MS", "10000"); + let overridden = read(DbConfig::default().with_session_timeouts_from_env()); + + for key in keys { + std::env::set_var(key, "0"); + } + let zero = read(DbConfig::default().with_session_timeouts_from_env()); + + for key in keys { + std::env::set_var(key, "not-a-number"); + } + let junk = read(DbConfig::default().with_session_timeouts_from_env()); + + for (key, value) in keys.iter().zip(previous) { + match value { + Some(value) => std::env::set_var(key, value), + None => std::env::remove_var(key), + } + } + + let defaults = (DEFAULT_LOCK_TIMEOUT_MS, DEFAULT_IDLE_TXN_TIMEOUT_MS, 0); + assert_eq!(unset, defaults, "unset env must keep the defaults"); + assert_eq!(overridden, (2000, 30000, 10000)); + assert_eq!( + zero, + (0, 0, 0), + "explicit 0 must pass through to disable the timeout" + ); + assert_eq!(junk, defaults, "junk env must keep the defaults"); + } + + /// End-to-end proof that the session-timeout GUCs actually install + /// through `Db::new()` and behave under contention: + /// + /// 1. The three effective GUC values on a pooled connection match the + /// configured `DbConfig` values. + /// 2. A statement waiting on a held relation lock fails with SQLSTATE + /// `55P03` (lock_not_available) at the configured `lock_timeout` + /// instead of parking — the incident shape this PR exists for. + /// 3. `Db::migrate()` (the advisory-lock path) is exempt: it succeeds + /// even while another session holds the schema-destruction advisory + /// lock longer than `lock_timeout`. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn session_timeouts_install_through_db_new_and_bound_lock_waits() { + let admin = PgPool::connect(&admin_url().await) + .await + .expect("connect admin"); + let (seed_pool, name) = create_scratch_db(&admin, "session_timeouts").await; + seed_pool.close().await; + + let base = admin_url().await; + 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.clone(), + max_connections: 2, + lock_timeout_ms: 500, + idle_txn_timeout_ms: 60_000, + statement_timeout_ms: 0, + ..DbConfig::default() + }) + .await + .expect("connect Db with session timeouts"); + + // 1. Effective GUCs, not intent. + let (lock, idle, stmt): (String, String, String) = sqlx::query_as( + "SELECT current_setting('lock_timeout'), \ + current_setting('idle_in_transaction_session_timeout'), \ + current_setting('statement_timeout')", + ) + .fetch_one(&db.pool) + .await + .expect("read effective GUCs"); + assert_eq!(lock, "500ms", "lock_timeout must install through Db::new"); + assert_eq!( + idle, "1min", + "idle txn timeout must install through Db::new" + ); + assert_eq!(stmt, "0", "statement_timeout must stay disabled by default"); + + // 2. Contended relation lock fails fast with 55P03. + let mut holder = db.pool.acquire().await.expect("holder connection"); + sqlx::raw_sql("BEGIN; LOCK TABLE events IN ACCESS EXCLUSIVE MODE") + .execute(&mut *holder) + .await + .expect("hold relation lock"); + let waited = std::time::Instant::now(); + let mut waiter_txn = db.pool.begin().await.expect("waiter transaction"); + let err = sqlx::query("LOCK TABLE events IN ACCESS SHARE MODE") + .execute(&mut *waiter_txn) + .await + .expect_err("waiter must time out, not park"); + drop(waiter_txn); + let code = match &err { + sqlx::Error::Database(db_err) => db_err.code().map(|c| c.to_string()), + other => panic!("expected database error, got {other:?}"), + }; + assert_eq!( + code.as_deref(), + Some("55P03"), + "contended lock wait must fail with lock_not_available" + ); + assert!( + waited.elapsed() < std::time::Duration::from_secs(5), + "waiter must fail at the configured timeout, not park indefinitely" + ); + + // 3. Migration path is exempt: hold the schema-destruction advisory + // lock on a separate session, then release it after well over the + // 500ms lock_timeout. `Db::migrate()` must wait it out and succeed. + let mut advisory_holder = PgPool::connect(&scratch_url) + .await + .expect("advisory holder pool") + .acquire() + .await + .expect("advisory holder conn") + .detach(); + sqlx::query("SELECT pg_advisory_lock($1)") + .bind(crate::deletion::SCHEMA_DESTRUCTION_LOCK_KEY) + .execute(&mut advisory_holder) + .await + .expect("hold schema advisory lock"); + let release = tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(1_500)).await; + let _ = sqlx::query("SELECT pg_advisory_unlock($1)") + .bind(crate::deletion::SCHEMA_DESTRUCTION_LOCK_KEY) + .execute(&mut advisory_holder) + .await; + let _ = advisory_holder.close().await; + }); + db.migrate() + .await + .expect("migrate must wait out the advisory holder, not fail at lock_timeout"); + release.await.expect("release task"); + + // Cleanup: end the holder's transaction before dropping the database. + let _ = sqlx::query("ROLLBACK").execute(&mut *holder).await; + drop(holder); + drop_scratch_db(&admin, db.pool.clone(), &name).await; + } + /// The armed writer pool (`Db::new`) must enforce the floor end-to-end /// through the public insert APIs, and the session GUC must be verifiably /// set on pooled connections. diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 94c7aea2faf..f20edf2f54f 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -82,6 +82,20 @@ where Fut: Future)>, { let mut lock_conn = pool.acquire().await?.detach(); + // Exempt this connection from the lock/statement writer-session timeouts. + // Migrations and schema-destruction ops are the two legitimate long-lock + // paths: the advisory-lock acquisition below is *designed* to park until + // the current holder finishes (a non-winning boot pod waiting out the + // migration winner must wait, not crash-loop on SQLSTATE 55P03), and the + // DDL that runs on this connection may legitimately exceed lock/statement + // budgets sized for runtime traffic. `idle_in_transaction_session_timeout` + // is deliberately NOT reset: a migration client wedged idle mid-transaction + // is exactly the holder that must be reaped — backend death also releases + // this advisory lock. Session-scoped: the connection is detached and + // closed below, never returned to the pool. + sqlx::raw_sql("SET lock_timeout = 0; SET statement_timeout = 0") + .execute(&mut lock_conn) + .await?; sqlx::query("SELECT pg_advisory_lock($1)") .bind(SCHEMA_DESTRUCTION_LOCK_KEY) .execute(&mut lock_conn) diff --git a/crates/buzz-deletion/src/lib.rs b/crates/buzz-deletion/src/lib.rs index 4e27b85fe9f..c0d2a89e8d1 100644 --- a/crates/buzz-deletion/src/lib.rs +++ b/crates/buzz-deletion/src/lib.rs @@ -526,11 +526,14 @@ fn resolve_submit_host(host: Option<&str>, relay_url: Option<&str>) -> Result Result { let database_url = required_env("DATABASE_URL")?; - let db = Db::new(&DbConfig { - database_url, - max_connections: env_parse("BUZZ_DB_POOL_SIZE", 20), - ..DbConfig::default() - }) + let db = Db::new( + &DbConfig { + database_url, + max_connections: env_parse("BUZZ_DB_POOL_SIZE", 20), + ..DbConfig::default() + } + .with_session_timeouts_from_env(), + ) .await?; Ok(store(&db)) } diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 566b684f830..e36dda25aa9 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -35,6 +35,18 @@ fn buzz_auto_migrate_enabled(value: Option<&str>) -> bool { }) } +async fn connect_audit_pool(config: &DbConfig) -> anyhow::Result { + let audit_config = DbConfig { + read_database_url: None, + max_connections: 5, + min_connections: 1, + ..config.clone() + }; + Db::connect_writer_pool(&audit_config) + .await + .map_err(Into::into) +} + /// Controls how many per-community gauge series the usage poller emits. /// /// Datadog cost is proportional to the number of unique time-series. With ~25 @@ -170,7 +182,8 @@ async fn main() -> anyhow::Result<()> { max_connections: config.db_pool_size, read_max_connections: config.db_read_pool_size, ..DbConfig::default() - }; + } + .with_session_timeouts_from_env(); let db = Db::new(&db_config).await.map_err(|e| { error!("Failed to connect to Postgres: {e}"); anyhow::anyhow!("DB connection failed: {e}") @@ -353,10 +366,7 @@ async fn main() -> anyhow::Result<()> { } let audit = if config.audit_enabled { - let audit_pool = sqlx::postgres::PgPoolOptions::new() - .max_connections(5) - .min_connections(1) - .connect(&config.database_url) + let audit_pool = connect_audit_pool(&db_config) .await .map_err(|e| anyhow::anyhow!("Audit DB connection failed: {e}"))?; info!("Audit service ready"); @@ -2036,10 +2046,11 @@ mod tests { use uuid::Uuid; use super::{ - buzz_auto_migrate_enabled, dropped_in_memory_keys, idle_timeout_secs, + buzz_auto_migrate_enabled, connect_audit_pool, dropped_in_memory_keys, idle_timeout_secs, refresh_legacy_active_gauge_recency, run_periodic_until_cancelled, EmissionScope, InMemoryMetricKey, }; + use buzz_db::DbConfig; use metrics::GaugeFn; use metrics_util::{ debugging::DebugValue, @@ -2071,6 +2082,67 @@ mod tests { assert!(tick_count.load(std::sync::atomic::Ordering::Relaxed) <= 1); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn audit_writer_pool_installs_timeouts_and_bounds_advisory_lock_waits() { + let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL"); + let pool = connect_audit_pool(&DbConfig { + database_url, + max_connections: 2, + min_connections: 0, + lock_timeout_ms: 500, + idle_txn_timeout_ms: 60_000, + statement_timeout_ms: 0, + ..DbConfig::default() + }) + .await + .expect("connect audit writer pool"); + + let (lock, idle, statement): (String, String, String) = sqlx::query_as( + "SELECT current_setting('lock_timeout'), \ + current_setting('idle_in_transaction_session_timeout'), \ + current_setting('statement_timeout')", + ) + .fetch_one(&pool) + .await + .expect("read effective audit writer GUCs"); + assert_eq!(lock, "500ms"); + assert_eq!(idle, "1min"); + assert_eq!(statement, "0"); + + let lock_key = i64::from_be_bytes( + Uuid::new_v4().as_bytes()[..8] + .try_into() + .expect("eight UUID bytes"), + ); + let mut holder = pool.acquire().await.expect("audit lock holder"); + sqlx::query("SELECT pg_advisory_lock($1)") + .bind(lock_key) + .execute(&mut *holder) + .await + .expect("hold audit advisory lock"); + + let started = std::time::Instant::now(); + let mut waiter = pool.acquire().await.expect("audit lock waiter"); + let err = sqlx::query("SELECT pg_advisory_lock($1)") + .bind(lock_key) + .execute(&mut *waiter) + .await + .expect_err("audit advisory-lock waiter must time out"); + let code = match &err { + sqlx::Error::Database(db_err) => db_err.code().map(|code| code.to_string()), + other => panic!("expected database error, got {other:?}"), + }; + assert_eq!(code.as_deref(), Some("55P03")); + assert!(started.elapsed() < Duration::from_secs(5)); + + sqlx::query("SELECT pg_advisory_unlock($1)") + .bind(lock_key) + .execute(&mut *holder) + .await + .expect("release audit advisory lock"); + } + #[test] fn buzz_auto_migrate_is_opt_in() { assert!(!buzz_auto_migrate_enabled(None)); diff --git a/docs/push-gateway-deployment.md b/docs/push-gateway-deployment.md index 63c63355a11..3e1908b7195 100644 --- a/docs/push-gateway-deployment.md +++ b/docs/push-gateway-deployment.md @@ -38,6 +38,8 @@ The gateway stores APNs tokens encrypted in PostgreSQL. Database backups therefo All replicas must share one PostgreSQL database. Delivery authority, replay admission, and endpoint quota reservation are transactional there, so replica count does not multiply the abuse ceiling. The gateway owns a scoped migration history under `crates/buzz-push-gateway/migrations`; it creates only the six `push_gateway_*` authority tables plus SQLx's migration-history table and never runs relay migrations. +The gateway's dedicated pool does not consume the relay-oriented `BUZZ_DB_LOCK_TIMEOUT_MS`, `BUZZ_DB_IDLE_TXN_TIMEOUT_MS`, or `BUZZ_DB_STATEMENT_TIMEOUT_MS` settings. Its session-timeout policy remains separate from the `buzz-db` writer policy and must be designed and rolled out independently. + The Helm chart runs a single pre-install/pre-upgrade migration Job using `migration.existingSecret`; that secret contains a DDL-capable `DATABASE_URL`. The URL MUST name a dedicated gateway database, not the relay database: SQLx stores its `_sqlx_migrations` history in `public`, so sharing a database would collide with another application's migration history. `migration.runtimeDatabaseRole` names an existing LOGIN role (the default is `buzz_push_gateway_runtime`) used by runtime `DATABASE_URL`. After scoped migrations, the Job revokes database `CREATE` from that role and schema `CREATE` from both `PUBLIC` and the role, then grants only database `CONNECT`, schema `USAGE`, and `SELECT, INSERT, UPDATE, DELETE` on the six gateway tables. The migration role must own the database/schema objects or otherwise be allowed to issue those grants; it is never provided to runtime replicas. Readiness rejects an empty/partial schema, missing DML, or a runtime role that retains database/schema `CREATE`. Helm waits for the migration hook before updating replicas, so rolling deployments never race unconditional startup migration. Readiness must be removed from load-balancer service endpoints before terminating a pod. The service reaps expired challenges and replay rows, idle quota rows, expired/revoked delegations, and retention-eligible installations (including their encrypted token ciphertext) at startup and every five minutes. Monitor reaper failures and table growth; retention does not depend on process restarts.