feat(db): configurable writer session timeouts (lock, idle-txn, statement) (#6229)

## Why

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. We hit exactly this in
production — ~1,400 sessions queued behind one crash-looping pod's boot
transaction for ~20 minutes until kubelet killed the container.

## What

Applies session-level Postgres timeouts to every **writer** connection
inside the existing single `after_connect` hook in `buzz-db`, all
env-tunable through the same `Config::from_env → DbConfig` path as the
existing pool-size knobs:

| Env var | GUC | Default | Effect |
|---|---|---|---|
| `BUZZ_DB_LOCK_TIMEOUT_MS` | `lock_timeout` | 5000 | statements waiting
on any lock fail fast instead of parking behind a wedged holder |
| `BUZZ_DB_IDLE_TXN_TIMEOUT_MS` | `idle_in_transaction_session_timeout`
| 60000 | reaps wedged clients idling inside an open transaction while
holding locks |
| `BUZZ_DB_STATEMENT_TIMEOUT_MS` | `statement_timeout` | 0 (off) |
opt-in runaway-statement cap; off by default because startup
migrations/backfills legitimately run long statements |

`0` disables a timeout (Postgres semantics) and deliberately passes
through the env parsing — unlike the pool-size knobs where `0` falls
back to the default. The reader pool is untouched: replica sessions
never take contended locks and already fail acquire in 150 ms.

Deployers tune these via plain env vars (`.env`, or `relay.extraEnv` in
the Helm chart) — no code changes needed.

## Behavior change to note

With the 5 s default `lock_timeout`, a boot-time migration or backfill
that waits >5 s on a lock now errors (surfacing in logs / crash-looping
the pod) instead of stalling silently. That is the intended
visible-failure-over-fleet-stall tradeoff; deployers with slow contended
migrations can set `BUZZ_DB_LOCK_TIMEOUT_MS=0`.

## Testing

- `cargo test -p buzz-db -p buzz-relay` — buzz-db green; buzz-relay has
9 failures that also fail on clean `main` in this environment
(api::admin/api::media/mesh_demo — unrelated, pre-existing).
- New config test covers override / `0`-passthrough / invalid-fallback
for all three env vars.
- Extended the existing `writer_pool_safety_hook_is_single_and_composed`
source-shape test so the timeouts can't drift out of the single
`after_connect` hook (SQLx replaces hooks — a second hook would silently
disarm the floor guard).
- `cargo fmt --check` and `cargo clippy --all-targets` clean for the
touched crates.

Closest existing PR/issue: none found.

---
**Update Aug 28, 17:06 EDT:** Rebased onto `main` at `a3730784fc` and
addressed the latest correctness review.

- Ported the timeout policy onto the refactored `buzz-db::runtime` pool
constructor and kept the shared env overlay for relay, admin, deletion,
and audit writers.
- Migration/schema-destruction connections now disable `lock_timeout`
and `statement_timeout` for their intentional long wait/DDL path. This
supersedes the earlier “Behavior change to note”: contended boot
migrations wait for the current migration owner rather than
crash-looping after five seconds.
- The audit worker now preserves and retries the same entry on
PostgreSQL `55P03` lock timeouts, using exponential backoff capped at
one second. Other database errors retain the existing terminal error
behavior, and retries emit `buzz_audit_log_lock_retries_total`.
- Added CI-backed PostgreSQL regressions for writer GUC
installation/migration exemption, audit-pool lock timeouts, and worker
recovery. The worker regression holds the real audit advisory lock past
`lock_timeout`, observes a retry, releases the lock, and proves the
original entry is appended exactly once.

Current verification supersedes the earlier testing notes: workspace
Rust clippy passed with warnings denied; all nine infrastructure-free
backend unit-test lanes passed; all three focused PostgreSQL regressions
passed against PostgreSQL 17; formatting, diff checks, file-size guards,
and desktop frontend checks passed. The Linux Blox workstation could not
run the unrelated Tauri native lane because `glib-2.0` is absent, so
that platform check is left to PR CI.


---
**Update Aug 31, 11:09 EDT:** Rebased onto current `main` at
`c3132c3ee9` and reran the requested audit-lock contention scenario on
Blox at head `896c3fe9ed`.

- `git range-diff` reports both PR commits unchanged by the rebase; the
branch remains two commits and the worktree is clean.
- `cargo fmt --all -- --check` and clippy with warnings denied passed
for `buzz-db`, `buzz-relay`, `buzz-admin`, and `buzz-deletion`.
- All three focused PostgreSQL 17 regressions passed: writer session
timeout/migration exemption, audit writer timeout bounds, and audit
worker recovery of the original entry exactly once.
- Live protocol verification used a head-built relay and CLI, native
PostgreSQL 17/Redis, `BUZZ_DB_LOCK_TIMEOUT_MS=300`, and an eight-second
hold on the community audit advisory lock. The real message was accepted
and persisted once while the lock was held; its audit-row count remained
zero during contention while retries accumulated. After release, exactly
one `event_created` audit row appeared and remained exactly one after an
additional two-second duplicate check. The run recorded nine
lock-timeout retries, zero audit failures, and event ID
`7f8c4ffae28e78555fcf2d56396d6e6c01b3712e5411288dc79e9a54af9d9444`.

Generated with Codex

---------

Signed-off-by: Luke Tornquist <tornquist@squareup.com>

.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)
 # -----------------------------------------------------------------------------

.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

crates/buzz-admin/src/main.rs

@@ -433,10 +433,13 @@ async fn connect_member_services() -> Result<(Db, Arc<PubSubManager>, Keys)> {
 async fn connect_db() -> Result<Db> {
     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)
 }

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)")

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<u64> {
+            std::env::var(key)
+                .ok()
+                .and_then(|value| value.parse::<u64>().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<Self> {
-        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<PgPool> {
+    /// 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<PgPool> {
+        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

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.

crates/buzz-deletion/src/lib.rs

@@ -531,11 +531,14 @@ fn resolve_submit_host(host: Option<&str>, relay_url: Option<&str>) -> Result<St
 
 async fn connect_store() -> Result<DeletionStore> {
     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))
 }

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<sqlx::PgPool> {
+    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<nostr::Keys> {
     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));

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);

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.