From 8f5c1bda9dc2eebcfde62c326261b973d9c3a0f7 Mon Sep 17 00:00:00 2001 From: Fizz <3a9f8a30fbb462abec1e2977b2280a7ae50c7ff794433790be15bd48bfd52d0b@buzz.block.builderlab.xyz> Date: Tue, 18 Aug 2026 10:54:10 -0400 Subject: [PATCH 1/4] feat(db): configurable writer session timeouts (lock, idle-txn, statement) A wedged relay boot pod holding a relation lock can park every other writer in the fleet behind it: DB load pins at pool capacity in Lock:relation waits while CPU stays flat, and nothing server-side releases the lock until the holder dies. Apply session-level timeouts to every writer connection inside the existing single after_connect hook: - lock_timeout (BUZZ_DB_LOCK_TIMEOUT_MS, default 5000): statements waiting on any lock fail fast instead of parking, so a stuck holder produces visible per-statement errors rather than a fleet stall. - idle_in_transaction_session_timeout (BUZZ_DB_IDLE_TXN_TIMEOUT_MS, default 60000): reaps wedged clients idling inside an open transaction while holding locks. - statement_timeout (BUZZ_DB_STATEMENT_TIMEOUT_MS, default 0/off): opt-in runaway-statement cap; off by default because startup migrations and backfills legitimately run long statements. All three are env-tunable through the same Config::from_env -> DbConfig path as the pool-size knobs; 0 disables a timeout (Postgres semantics) and must pass through, so the env parsing deliberately does not filter zero like the pool sizes do. Reader pool untouched: replica sessions never take contended locks and already fail acquire in 150ms. Closest existing PR/issue: none found. Co-authored-by: Luke Tornquist Signed-off-by: Luke Tornquist --- .env.example | 11 ++++ crates/buzz-db/src/lib.rs | 56 +++++++++++++++++++- crates/buzz-relay/src/config.rs | 94 +++++++++++++++++++++++++++++++++ crates/buzz-relay/src/main.rs | 12 ++++- 4 files changed, 171 insertions(+), 2 deletions(-) diff --git a/.env.example b/.env.example index 0f7bbba6f1..6924a87f74 100644 --- a/.env.example +++ b/.env.example @@ -38,6 +38,17 @@ REDIS_URL=redis://localhost:6379 # READ_DATABASE_URL is set, reader (default 50). # BUZZ_DB_POOL_SIZE=50 +# Writer-session Postgres timeouts, all in milliseconds; 0 disables. +# 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. +# BUZZ_DB_STATEMENT_TIMEOUT_MS=0 + # ----------------------------------------------------------------------------- # Typesense (search) # ----------------------------------------------------------------------------- diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 330525d310..cfd8c0fc78 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,24 @@ 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; + /// Community host-map row returned by [`Db::lookup_community_by_host`]. #[derive(Debug, Clone, PartialEq, Eq)] pub struct CommunityRecord { @@ -701,19 +731,37 @@ impl Db { /// 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 { + 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). + // These make one wedged lock holder fail its own + // statements instead of parking every other writer + // behind it (see DbConfig docs for each knob). Bare + // integers are milliseconds for all three GUCs. + 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?; @@ -8749,6 +8797,12 @@ mod tests { ); assert!(connect_pool.contains("buzz.created_at_floor")); assert!(connect_pool.contains("SHOW transaction_isolation")); + assert!( + connect_pool.contains("'lock_timeout'") + && connect_pool.contains("'idle_in_transaction_session_timeout'") + && connect_pool.contains("'statement_timeout'"), + "session timeouts must be applied inside the single writer hook" + ); assert!(!connect_pool.contains("arm_floor_guard")); assert!(!connect_pool.contains("_arm_floor_guard")); assert!(!connect_pool.contains("allow(unused_variables)")); diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 037c6b1dd3..fa1fcf7805 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -99,6 +99,17 @@ 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, + /// Writer session `lock_timeout` in ms (`BUZZ_DB_LOCK_TIMEOUT_MS`). + /// `None` keeps the [`buzz_db::DEFAULT_LOCK_TIMEOUT_MS`] default; + /// `Some(0)` disables the timeout. + pub db_lock_timeout_ms: Option, + /// Writer session `idle_in_transaction_session_timeout` in ms + /// (`BUZZ_DB_IDLE_TXN_TIMEOUT_MS`). `None` keeps the + /// [`buzz_db::DEFAULT_IDLE_TXN_TIMEOUT_MS`] default; `Some(0)` disables. + pub db_idle_txn_timeout_ms: Option, + /// Writer session `statement_timeout` in ms + /// (`BUZZ_DB_STATEMENT_TIMEOUT_MS`). `None`/`Some(0)` disables — opt-in. + pub db_statement_timeout_ms: Option, /// 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. @@ -531,6 +542,22 @@ impl Config { .and_then(|v| v.parse::().ok()) .filter(|&v| v > 0); + // Session timeout knobs: absent or unparseable env keeps the buzz-db + // default; an explicit `0` disables the timeout (Postgres semantics), + // so 0 must pass through rather than be filtered out like the pool + // sizes above. + let db_lock_timeout_ms = std::env::var("BUZZ_DB_LOCK_TIMEOUT_MS") + .ok() + .and_then(|v| v.parse::().ok()); + + let db_idle_txn_timeout_ms = std::env::var("BUZZ_DB_IDLE_TXN_TIMEOUT_MS") + .ok() + .and_then(|v| v.parse::().ok()); + + let db_statement_timeout_ms = std::env::var("BUZZ_DB_STATEMENT_TIMEOUT_MS") + .ok() + .and_then(|v| v.parse::().ok()); + let relay_url = std::env::var("RELAY_URL").unwrap_or_else(|_| "ws://localhost:3000".to_string()); @@ -996,6 +1023,9 @@ impl Config { redis_pool_size, db_pool_size, db_read_pool_size, + db_lock_timeout_ms, + db_idle_txn_timeout_ms, + db_statement_timeout_ms, relay_url, pairing_relay_url, max_connections, @@ -1293,6 +1323,70 @@ mod tests { assert_eq!(junk, None, "unparsable value must fall back to inheriting"); } + #[test] + fn db_session_timeout_env_overrides_zero_passthrough_and_invalid_fallback() { + let _guard = ENV_MUTEX.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: &Config| { + ( + config.db_lock_timeout_ms, + config.db_idle_txn_timeout_ms, + config.db_statement_timeout_ms, + ) + }; + + for key in keys { + std::env::remove_var(key); + } + let unset = read(&Config::from_env().expect("config")); + + 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(&Config::from_env().expect("config")); + + // `0` means "disable this timeout" and must pass through, unlike + // the pool-size knobs where 0 falls back to the default. + std::env::set_var("BUZZ_DB_LOCK_TIMEOUT_MS", "0"); + std::env::set_var("BUZZ_DB_IDLE_TXN_TIMEOUT_MS", "0"); + std::env::set_var("BUZZ_DB_STATEMENT_TIMEOUT_MS", "0"); + let zero = read(&Config::from_env().expect("config")); + + std::env::set_var("BUZZ_DB_LOCK_TIMEOUT_MS", "not-a-number"); + std::env::set_var("BUZZ_DB_IDLE_TXN_TIMEOUT_MS", "not-a-number"); + std::env::set_var("BUZZ_DB_STATEMENT_TIMEOUT_MS", "not-a-number"); + let junk = read(&Config::from_env().expect("config")); + + for (key, value) in keys.iter().zip(previous) { + match value { + Some(value) => std::env::set_var(key, value), + None => std::env::remove_var(key), + } + } + + assert_eq!( + unset, + (None, None, None), + "unset must defer to buzz-db defaults" + ); + assert_eq!(overridden, (Some(2000), Some(30000), Some(10000))); + assert_eq!( + zero, + (Some(0), Some(0), Some(0)), + "explicit 0 must pass through to disable the timeout" + ); + assert_eq!( + junk, + (None, None, None), + "unparsable values must defer to buzz-db defaults" + ); + } + #[test] fn read_database_url_unset_or_blank_is_none() { let _guard = ENV_MUTEX.lock().unwrap(); diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 3584e1849d..a12b1bbf3d 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -163,13 +163,23 @@ async fn main() -> anyhow::Result<()> { "Prometheus metrics exporter started" ); + let default_db = DbConfig::default(); let db_config = DbConfig { database_url: config.database_url.clone(), read_database_url: config.read_database_url.clone(), 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, - ..DbConfig::default() + lock_timeout_ms: config + .db_lock_timeout_ms + .unwrap_or(default_db.lock_timeout_ms), + idle_txn_timeout_ms: config + .db_idle_txn_timeout_ms + .unwrap_or(default_db.idle_txn_timeout_ms), + statement_timeout_ms: config + .db_statement_timeout_ms + .unwrap_or(default_db.statement_timeout_ms), + ..default_db }; let db = Db::new(&db_config).await.map_err(|e| { error!("Failed to connect to Postgres: {e}"); From 605244ccb2e0c57fbe88f921cc7f3f951b61e767 Mon Sep 17 00:00:00 2001 From: Fizz <3a9f8a30fbb462abec1e2977b2280a7ae50c7ff794433790be15bd48bfd52d0b@buzz.block.builderlab.xyz> Date: Thu, 20 Aug 2026 10:18:46 -0400 Subject: [PATCH 2/4] =?UTF-8?q?fix(db):=20address=20session-timeout=20revi?= =?UTF-8?q?ew=20=E2=80=94=20migration=20exemption,=20shared=20env=20knobs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the four actionable review threads on the writer session timeout PR: - Exempt the migration/schema-destruction advisory-lock connection from lock_timeout and statement_timeout. Without this, a non-winning boot pod waiting out the migration winner would crash-loop on SQLSTATE 55P03 after 5s. idle_in_transaction_session_timeout deliberately stays active there: a wedged idle migration client is exactly the holder that must be reaped, and backend death releases the advisory lock. Proven by a Postgres-backed test that runs Db::migrate() while another session holds the advisory lock past lock_timeout, wired into the Backend Integration CI job. - Move the BUZZ_DB_*_TIMEOUT_MS env parsing from buzz-relay's Config into buzz-db as DbConfig::with_session_timeouts_from_env(), so buzz-admin and buzz-deletion writers honor the knobs instead of silently ignoring them. All three binaries now overlay the same env keys onto DbConfig; zero still passes through (Postgres "disabled" semantics) and invalid values fall back to defaults, covered by a unit test. - Reword the after_connect hook comment to describe each knob's actual actor: lock_timeout fails the *waiter*, idle-txn timeout reaps idle in-transaction holders, and an actively-executing wedged holder is only bounded by statement_timeout. - Warn in .env.example that a pathologically low statement_timeout can fail connection setup itself. Also adds an end-to-end Postgres test asserting the effective GUCs on pooled connections and the fast 55P03 failure of a contended lock wait. Co-authored-by: Luke Tornquist Signed-off-by: Luke Tornquist --- .env.example | 5 +- .github/workflows/ci.yml | 13 ++ crates/buzz-admin/src/main.rs | 11 +- crates/buzz-db/src/lib.rs | 217 +++++++++++++++++++++++++++++++- crates/buzz-db/src/migration.rs | 14 +++ crates/buzz-deletion/src/lib.rs | 13 +- crates/buzz-relay/src/config.rs | 94 -------------- crates/buzz-relay/src/main.rs | 15 +-- 8 files changed, 262 insertions(+), 120 deletions(-) diff --git a/.env.example b/.env.example index 6924a87f74..e16bcee2b5 100644 --- a/.env.example +++ b/.env.example @@ -46,7 +46,10 @@ REDIS_URL=redis://localhost:6379 # 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. +# 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 # ----------------------------------------------------------------------------- diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a832c0a0af..57f1ccaf9b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -744,6 +744,19 @@ 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 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 — see buzz-db tests. + 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: 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 42a7de84f7..19a3b1d9d4 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 cfd8c0fc78..9eb4c97fa5 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -604,6 +604,37 @@ pub const DEFAULT_LOCK_TIMEOUT_MS: u64 = 5_000; /// 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 writer + /// binary (`buzz-relay`, `buzz-admin`, `buzz-deletion`) honors the same + /// operator knobs; parsing it in one binary would make the documented + /// override silently no-op in the others. + /// + /// 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 { @@ -748,10 +779,18 @@ impl Db { .execute(&mut *conn) .await?; // Session timeouts (0 = disabled, Postgres semantics). - // These make one wedged lock holder fail its own - // statements instead of parking every other writer - // behind it (see DbConfig docs for each knob). Bare - // integers are milliseconds for all three GUCs. + // 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), \ @@ -8858,6 +8897,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 be87faa1ac..de73c6953d 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -67,6 +67,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 4e27b85fe9..c0d2a89e8d 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/config.rs b/crates/buzz-relay/src/config.rs index fa1fcf7805..037c6b1dd3 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -99,17 +99,6 @@ 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, - /// Writer session `lock_timeout` in ms (`BUZZ_DB_LOCK_TIMEOUT_MS`). - /// `None` keeps the [`buzz_db::DEFAULT_LOCK_TIMEOUT_MS`] default; - /// `Some(0)` disables the timeout. - pub db_lock_timeout_ms: Option, - /// Writer session `idle_in_transaction_session_timeout` in ms - /// (`BUZZ_DB_IDLE_TXN_TIMEOUT_MS`). `None` keeps the - /// [`buzz_db::DEFAULT_IDLE_TXN_TIMEOUT_MS`] default; `Some(0)` disables. - pub db_idle_txn_timeout_ms: Option, - /// Writer session `statement_timeout` in ms - /// (`BUZZ_DB_STATEMENT_TIMEOUT_MS`). `None`/`Some(0)` disables — opt-in. - pub db_statement_timeout_ms: Option, /// 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. @@ -542,22 +531,6 @@ impl Config { .and_then(|v| v.parse::().ok()) .filter(|&v| v > 0); - // Session timeout knobs: absent or unparseable env keeps the buzz-db - // default; an explicit `0` disables the timeout (Postgres semantics), - // so 0 must pass through rather than be filtered out like the pool - // sizes above. - let db_lock_timeout_ms = std::env::var("BUZZ_DB_LOCK_TIMEOUT_MS") - .ok() - .and_then(|v| v.parse::().ok()); - - let db_idle_txn_timeout_ms = std::env::var("BUZZ_DB_IDLE_TXN_TIMEOUT_MS") - .ok() - .and_then(|v| v.parse::().ok()); - - let db_statement_timeout_ms = std::env::var("BUZZ_DB_STATEMENT_TIMEOUT_MS") - .ok() - .and_then(|v| v.parse::().ok()); - let relay_url = std::env::var("RELAY_URL").unwrap_or_else(|_| "ws://localhost:3000".to_string()); @@ -1023,9 +996,6 @@ impl Config { redis_pool_size, db_pool_size, db_read_pool_size, - db_lock_timeout_ms, - db_idle_txn_timeout_ms, - db_statement_timeout_ms, relay_url, pairing_relay_url, max_connections, @@ -1323,70 +1293,6 @@ mod tests { assert_eq!(junk, None, "unparsable value must fall back to inheriting"); } - #[test] - fn db_session_timeout_env_overrides_zero_passthrough_and_invalid_fallback() { - let _guard = ENV_MUTEX.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: &Config| { - ( - config.db_lock_timeout_ms, - config.db_idle_txn_timeout_ms, - config.db_statement_timeout_ms, - ) - }; - - for key in keys { - std::env::remove_var(key); - } - let unset = read(&Config::from_env().expect("config")); - - 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(&Config::from_env().expect("config")); - - // `0` means "disable this timeout" and must pass through, unlike - // the pool-size knobs where 0 falls back to the default. - std::env::set_var("BUZZ_DB_LOCK_TIMEOUT_MS", "0"); - std::env::set_var("BUZZ_DB_IDLE_TXN_TIMEOUT_MS", "0"); - std::env::set_var("BUZZ_DB_STATEMENT_TIMEOUT_MS", "0"); - let zero = read(&Config::from_env().expect("config")); - - std::env::set_var("BUZZ_DB_LOCK_TIMEOUT_MS", "not-a-number"); - std::env::set_var("BUZZ_DB_IDLE_TXN_TIMEOUT_MS", "not-a-number"); - std::env::set_var("BUZZ_DB_STATEMENT_TIMEOUT_MS", "not-a-number"); - let junk = read(&Config::from_env().expect("config")); - - for (key, value) in keys.iter().zip(previous) { - match value { - Some(value) => std::env::set_var(key, value), - None => std::env::remove_var(key), - } - } - - assert_eq!( - unset, - (None, None, None), - "unset must defer to buzz-db defaults" - ); - assert_eq!(overridden, (Some(2000), Some(30000), Some(10000))); - assert_eq!( - zero, - (Some(0), Some(0), Some(0)), - "explicit 0 must pass through to disable the timeout" - ); - assert_eq!( - junk, - (None, None, None), - "unparsable values must defer to buzz-db defaults" - ); - } - #[test] fn read_database_url_unset_or_blank_is_none() { let _guard = ENV_MUTEX.lock().unwrap(); diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index a12b1bbf3d..411e931b21 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -163,24 +163,15 @@ async fn main() -> anyhow::Result<()> { "Prometheus metrics exporter started" ); - let default_db = DbConfig::default(); let db_config = DbConfig { database_url: config.database_url.clone(), read_database_url: config.read_database_url.clone(), 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, - lock_timeout_ms: config - .db_lock_timeout_ms - .unwrap_or(default_db.lock_timeout_ms), - idle_txn_timeout_ms: config - .db_idle_txn_timeout_ms - .unwrap_or(default_db.idle_txn_timeout_ms), - statement_timeout_ms: config - .db_statement_timeout_ms - .unwrap_or(default_db.statement_timeout_ms), - ..default_db - }; + ..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}") From 9c86ce2bad772d6a673887a6c14bfe36afe883f8 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Fri, 21 Aug 2026 10:16:23 -0400 Subject: [PATCH 3/4] fix(db): apply writer timeouts to audit pool Signed-off-by: Luke Tornquist --- .env.example | 4 +- .github/workflows/ci.yml | 8 ++-- crates/buzz-db/src/lib.rs | 47 ++++++++++--------- crates/buzz-relay/src/main.rs | 81 +++++++++++++++++++++++++++++++-- docs/push-gateway-deployment.md | 2 + 5 files changed, 111 insertions(+), 31 deletions(-) diff --git a/.env.example b/.env.example index e16bcee2b5..f5210c31e2 100644 --- a/.env.example +++ b/.env.example @@ -38,7 +38,9 @@ REDIS_URL=redis://localhost:6379 # READ_DATABASE_URL is set, reader (default 50). # BUZZ_DB_POOL_SIZE=50 -# Writer-session Postgres timeouts, all in milliseconds; 0 disables. +# 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 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 57f1ccaf9b..b6fc227ec7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -747,13 +747,13 @@ jobs: - name: Writer session timeout guardrails # Proves the 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 — see buzz-db tests. + # through Db::new() and the relay audit writer, 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)' \ + -E '(package(buzz-db) and test(session_timeouts_install_through_db_new_and_bound_lock_waits)) or (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 diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9eb4c97fa5..091e608faf 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -609,10 +609,12 @@ impl DbConfig { /// `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 writer - /// binary (`buzz-relay`, `buzz-admin`, `buzz-deletion`) honors the same - /// operator knobs; parsing it in one binary would make the documented - /// override silently no-op in the others. + /// 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 @@ -736,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); @@ -756,12 +758,15 @@ 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; @@ -815,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 @@ -8824,27 +8829,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_writer_pool.contains("buzz.created_at_floor")); + assert!(connect_writer_pool.contains("SHOW transaction_isolation")); assert!( - connect_pool.contains("'lock_timeout'") - && connect_pool.contains("'idle_in_transaction_session_timeout'") - && connect_pool.contains("'statement_timeout'"), + 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_pool.contains("arm_floor_guard")); - assert!(!connect_pool.contains("_arm_floor_guard")); - assert!(!connect_pool.contains("allow(unused_variables)")); + 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") @@ -8853,7 +8858,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] diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 411e931b21..837cc13406 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 @@ -354,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"); @@ -2012,10 +2021,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, @@ -2047,6 +2057,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 63c63355a1..3e1908b719 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. From 9bbcade722b7a680e764feabcdb5acf738347ed3 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Fri, 21 Aug 2026 11:02:55 -0400 Subject: [PATCH 4/4] fix(ci): archive audit pool regression test Signed-off-by: Luke Tornquist --- .github/workflows/ci.yml | 22 +++++++++++++++++----- 1 file changed, 17 insertions(+), 5 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b6fc227ec7..31eece1f4f 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 @@ -745,15 +746,26 @@ jobs: env: DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz - name: Writer session timeout guardrails - # Proves the writer-pool session timeouts (lock_timeout / + # Proves the buzz-db writer-pool session timeouts (lock_timeout / # idle_in_transaction_session_timeout / statement_timeout) install - # through Db::new() and the relay audit writer, bound contended lock - # waits with SQLSTATE 55P03, and that Db::migrate()'s schema-destruction - # advisory lock is exempt from lock_timeout. + # 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)) or (package(buzz-relay) and test(audit_writer_pool_installs_timeouts_and_bounds_advisory_lock_waits))' \ + -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