From a720e249b245e11dc8c763b251515146bbf896e5 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Fri, 28 Aug 2026 17:03:21 -0400 Subject: [PATCH 1/2] feat(db): configure writer session timeouts Signed-off-by: Luke Tornquist --- .env.example | 15 +++ crates/buzz-admin/src/main.rs | 11 +- crates/buzz-db/src/runtime/migration.rs | 9 ++ crates/buzz-db/src/runtime/mod.rs | 79 ++++++++++++- crates/buzz-db/src/runtime/tests.rs | 151 +++++++++++++++++++++++- crates/buzz-deletion/src/lib.rs | 13 +- crates/buzz-relay/src/main.rs | 84 ++++++++++++- docs/push-gateway-deployment.md | 2 + 8 files changed, 341 insertions(+), 23 deletions(-) diff --git a/.env.example b/.env.example index 02e907b8cfd..c66d9c26c7a 100644 --- a/.env.example +++ b/.env.example @@ -38,6 +38,21 @@ 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 connection setup and can +# prevent any DB connection from establishing. +# BUZZ_DB_STATEMENT_TIMEOUT_MS=0 + # ----------------------------------------------------------------------------- # Typesense (search) # ----------------------------------------------------------------------------- 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/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 66251563cbd..00cd81c6940 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -85,6 +85,15 @@ where let mut lock_conn = crate::observability::acquire(pool, crate::observability::PoolRole::Writer) .await? .detach(); + // This dedicated connection intentionally waits for the current migration + // or schema-destruction owner and may then run long DDL. Exempt those two + // phases from runtime lock/statement budgets. Keep the idle-in-transaction + // timeout: a client wedged idle mid-migration is still a lock holder that + // should be reaped. The detached connection is closed below and never + // returns these session settings to the pool. + sqlx::raw_sql("SET lock_timeout = 0; SET statement_timeout = 0") + .execute(&mut lock_conn) + .await?; crate::observability::observe_advisory_lock( crate::observability::LockType::MigrationSchemaSafety, sqlx::query("SELECT pg_advisory_lock($1)") diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 29eef884024..73e5e4fe569 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -454,6 +454,16 @@ 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`). `0` disables the timeout. + pub lock_timeout_ms: u64, + /// Session `idle_in_transaction_session_timeout` in milliseconds for + /// writer connections (env `BUZZ_DB_IDLE_TXN_TIMEOUT_MS`). `0` disables. + pub idle_txn_timeout_ms: u64, + /// Session `statement_timeout` in milliseconds for writer connections + /// (env `BUZZ_DB_STATEMENT_TIMEOUT_MS`). `0` disables it and is the + /// default because migrations and backfills may legitimately run long. + pub statement_timeout_ms: u64, } impl Default for DbConfig { @@ -471,10 +481,47 @@ 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` in milliseconds. +pub const DEFAULT_LOCK_TIMEOUT_MS: u64 = 5_000; + +/// Default writer `idle_in_transaction_session_timeout` in milliseconds. +pub const DEFAULT_IDLE_TXN_TIMEOUT_MS: u64 = 60_000; + +impl DbConfig { + /// Overlay writer session timeouts from the shared `BUZZ_DB_*_TIMEOUT_MS` + /// environment variables. Missing or invalid values retain the existing + /// configuration; explicit zeroes pass through to disable a timeout. + /// + /// This belongs in `buzz-db` so relay, admin, deletion, and audit writers + /// share one policy. The separately deployed push gateway owns its own + /// database and session policy. + pub fn with_session_timeouts_from_env(mut self) -> Self { + fn parse(key: &str) -> Option { + std::env::var(key) + .ok() + .and_then(|value| value.parse::().ok()) + } + + if let Some(value) = parse("BUZZ_DB_LOCK_TIMEOUT_MS") { + self.lock_timeout_ms = value; + } + if let Some(value) = parse("BUZZ_DB_IDLE_TXN_TIMEOUT_MS") { + self.idle_txn_timeout_ms = value; + } + if let Some(value) = parse("BUZZ_DB_STATEMENT_TIMEOUT_MS") { + self.statement_timeout_ms = value; + } + self + } +} + impl Db { /// Creates a new `Db` by connecting a Postgres pool with the given config. /// @@ -486,7 +533,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); @@ -511,20 +558,44 @@ impl Db { /// 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, including the relay audit pool, must use this + /// constructor so they inherit the timeout, floor-guard, and isolation + /// policy installed by [`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?; + // `lock_timeout` fails the waiting statement; it does not + // cancel the holder. `idle_in_transaction_session_timeout` + // reaps only holders idling inside an open transaction, + // while actively executing holders are bounded only by + // `statement_timeout` (off by default). Bare values are + // milliseconds. Migration/schema-destruction connections + // reset lock and statement timeouts before their intentional + // long wait (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?; @@ -539,7 +610,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 diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index ecdc983a4ac..8349fb53874 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -1,7 +1,7 @@ use super::*; use crate::{relay_members, thread}; use buzz_core::CommunityId; -use sqlx::PgPool; +use sqlx::{Connection, PgPool}; use uuid::Uuid; const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; @@ -2180,10 +2180,10 @@ async fn created_at_floor_guard_aborts_old_channel_rows_at_commit() { fn writer_pool_safety_hook_is_single_and_composed() { let source = include_str!("mod.rs"); let connect_pool = source - .split("async fn connect_pool") + .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(), 1, @@ -2191,6 +2191,9 @@ fn writer_pool_safety_hook_is_single_and_composed() { ); assert!(connect_pool.contains("buzz.created_at_floor")); assert!(connect_pool.contains("SHOW transaction_isolation")); + assert!(connect_pool.contains("'lock_timeout'")); + assert!(connect_pool.contains("'idle_in_transaction_session_timeout'")); + assert!(connect_pool.contains("'statement_timeout'")); assert!(!connect_pool.contains("arm_floor_guard")); assert!(!connect_pool.contains("_arm_floor_guard")); assert!(!connect_pool.contains("allow(unused_variables)")); @@ -2202,7 +2205,7 @@ fn writer_pool_safety_hook_is_single_and_composed() { .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] @@ -2246,6 +2249,146 @@ async fn writer_pool_rejects_non_read_committed_database_default() { .expect("drop isolation test database"); } +/// Session-timeout environment overrides retain PostgreSQL's `0 = disabled` +/// semantics and ignore invalid values. +#[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 disable each timeout"); + assert_eq!(junk, defaults, "junk env must keep the defaults"); +} + +/// The production writer constructor installs all three timeout GUCs, bounds +/// ordinary lock waits, and exempts the intentional migration lock wait. +#[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"); + + 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(&db.pool) + .await + .expect("read effective GUCs"); + assert_eq!(lock, "500ms"); + assert_eq!(idle, "1min"); + assert_eq!(statement, "0"); + + 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 error = 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 &error { + sqlx::Error::Database(db_error) => db_error.code().map(|code| code.to_string()), + other => panic!("expected database error, got {other:?}"), + }; + assert_eq!(code.as_deref(), Some("55P03")); + assert!(waited.elapsed() < std::time::Duration::from_secs(5)); + + 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"); + release.await.expect("release task"); + + 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-deletion/src/lib.rs b/crates/buzz-deletion/src/lib.rs index f13b7d507ac..d963a7f261f 100644 --- a/crates/buzz-deletion/src/lib.rs +++ b/crates/buzz-deletion/src/lib.rs @@ -531,11 +531,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 d9432589f46..d81602e2019 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) +} + fn relay_keypair_from_config(relay_private_key: Option<&str>) -> anyhow::Result { let hex = relay_private_key.ok_or_else(|| { anyhow::anyhow!( @@ -183,7 +195,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}") @@ -366,10 +379,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"); @@ -2052,10 +2062,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, relay_keypair_from_config, run_periodic_until_cancelled, EmissionScope, InMemoryMetricKey, }; + use buzz_db::DbConfig; use metrics::GaugeFn; use metrics_util::{ debugging::DebugValue, @@ -2087,6 +2098,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 error = 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 &error { + sqlx::Error::Database(db_error) => db_error.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 e9a9ae16055..b2b66f5b5b4 100644 --- a/docs/push-gateway-deployment.md +++ b/docs/push-gateway-deployment.md @@ -53,6 +53,8 @@ The gateway stores APNs tokens encrypted in PostgreSQL. Database backups therefo ## PostgreSQL and replicas +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. + 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 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. From 896c3fe9edd3098be067c1cd1d347cf7c44e45db Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Fri, 28 Aug 2026 17:03:33 -0400 Subject: [PATCH 2/2] fix(audit): retry lock-timeout entries Signed-off-by: Luke Tornquist --- .github/workflows/ci.yml | 26 ++++++ crates/buzz-relay/src/state.rs | 157 +++++++++++++++++++++++++++++++-- 2 files changed, 178 insertions(+), 5 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 25a59c32432..44966c28de6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -380,6 +380,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 @@ -709,6 +710,31 @@ jobs: env: DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz TEST_DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz + - name: Writer session timeout guardrails + 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 + TEST_DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz + - name: Audit writer session timeout guardrails + 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: Audit worker lock-timeout recovery + run: | + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E 'package(buzz-relay) and test(audit_worker_retries_lock_timeout_until_original_entry_is_appended_once)' \ + --run-ignored ignored-only + env: + DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz - name: Start relay run: | chmod +x ./target/ci/buzz-relay diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index ab3f1d8c7eb..efdb2846148 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -1346,11 +1346,33 @@ impl AuditShutdownHandle { /// and the post-cancel drain share the same logic. async fn log_audit_entry(audit: &buzz_audit::AuditService, entry: buzz_audit::NewAuditEntry) { let t = std::time::Instant::now(); - if let Err(e) = audit.log(entry).await { - metrics::counter!("buzz_audit_log_errors_total").increment(1); - tracing::error!("Audit log failed: {e}"); - } else { - metrics::histogram!("buzz_audit_log_seconds").record(t.elapsed().as_secs_f64()); + let mut retry_delay_ms = 50u64; + let mut retries = 0u64; + loop { + match audit.log(entry.clone()).await { + Ok(_) => { + metrics::histogram!("buzz_audit_log_seconds").record(t.elapsed().as_secs_f64()); + return; + } + Err(buzz_audit::AuditError::Database(sqlx::Error::Database(database_error))) + if database_error.code().as_deref() == Some("55P03") => + { + retries += 1; + metrics::counter!("buzz_audit_log_lock_retries_total").increment(1); + tracing::warn!( + retries, + retry_delay_ms, + "Audit advisory lock timed out; preserving entry for retry" + ); + tokio::time::sleep(std::time::Duration::from_millis(retry_delay_ms)).await; + retry_delay_ms = (retry_delay_ms * 2).min(1_000); + } + Err(error) => { + metrics::counter!("buzz_audit_log_errors_total").increment(1); + tracing::error!("Audit log failed: {error}"); + return; + } + } } } @@ -1440,6 +1462,131 @@ mod tests { Arc::new(state) } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn audit_worker_retries_lock_timeout_until_original_entry_is_appended_once() { + let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL"); + let observer = sqlx::PgPool::connect(&database_url) + .await + .expect("connect observer pool"); + let application_name = format!("audit-retry-test-{}", Uuid::new_v4()); + let hook_application_name = application_name.clone(); + let audit_pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .after_connect(move |conn, _meta| { + let application_name = hook_application_name.clone(); + Box::pin(async move { + sqlx::query( + "SELECT set_config('application_name', $1, false), \ + set_config('lock_timeout', '100', false)", + ) + .bind(application_name) + .execute(&mut *conn) + .await?; + Ok(()) + }) + }) + .connect(&database_url) + .await + .expect("connect audit pool"); + + let community_id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_id) + .bind(format!("audit-retry-{community_id}.example")) + .execute(&observer) + .await + .expect("insert test community"); + let object_id = format!("audit-retry-object-{}", Uuid::new_v4()); + let entry = buzz_audit::NewAuditEntry { + community_id: CommunityId::from_uuid(community_id), + action: buzz_audit::AuditAction::EventCreated, + actor_pubkey: Some(vec![0xab; 32]), + object_id: Some(object_id.clone()), + detail: serde_json::json!({"test": "lock-timeout-retry"}), + }; + + // Mirrors buzz_audit::service::AUDIT_LOCK_NAMESPACE. + let lock_key = format!("buzz_audit:{community_id}"); + let mut holder = observer.acquire().await.expect("acquire lock holder"); + sqlx::query("SELECT pg_advisory_lock(hashtextextended($1, 0))") + .bind(&lock_key) + .execute(&mut *holder) + .await + .expect("hold community audit lock"); + + let audit = Arc::new(AuditService::new(audit_pool)); + let worker = tokio::spawn({ + let audit = Arc::clone(&audit); + async move { log_audit_entry(&audit, entry).await } + }); + + // Observe one timed-out advisory-lock attempt and then a second wait. + // Releasing during the first wait would not prove that the worker + // preserved and retried the original queue entry. + tokio::time::timeout(std::time::Duration::from_secs(3), async { + let mut saw_first_wait = false; + let mut saw_retry_gap = false; + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS (\ + SELECT 1 FROM pg_stat_activity \ + WHERE application_name = $1 \ + AND query LIKE 'SELECT pg_advisory_lock%' \ + AND wait_event = 'advisory'\ + )", + ) + .bind(&application_name) + .fetch_one(&observer) + .await + .expect("inspect audit lock waiter"); + if waiting { + if saw_retry_gap { + break; + } + saw_first_wait = true; + } else if saw_first_wait { + saw_retry_gap = true; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + }) + .await + .expect("audit worker never retried after lock_timeout"); + + sqlx::query("SELECT pg_advisory_unlock(hashtextextended($1, 0))") + .bind(&lock_key) + .execute(&mut *holder) + .await + .expect("release community audit lock"); + tokio::time::timeout(std::time::Duration::from_secs(3), worker) + .await + .expect("audit worker did not finish after lock release") + .expect("audit worker task panicked"); + + let rows: i64 = sqlx::query_scalar( + "SELECT count(*) FROM audit_log WHERE community_id = $1 AND object_id = $2", + ) + .bind(community_id) + .bind(&object_id) + .fetch_one(&observer) + .await + .expect("count retried audit rows"); + assert_eq!(rows, 1, "the preserved entry must be appended exactly once"); + + sqlx::query("DELETE FROM audit_log WHERE community_id = $1 AND object_id = $2") + .bind(community_id) + .bind(&object_id) + .execute(&observer) + .await + .expect("remove test audit row"); + sqlx::query("DELETE FROM communities WHERE id = $1") + .bind(community_id) + .execute(&observer) + .await + .expect("remove test community"); + } + #[test] fn send_to_resets_grace_counter_on_success() { let (mgr, id, _rx, _ctrl_rx, _cancel, bp) = setup_conn(16);