fix(db): keep lease releases durable and deflake PostgreSQL lane tests

crates/buzz-db/src/runtime/replica_fence.rs

@@ -1050,7 +1050,7 @@ mod postgres_tests {
     /// still held. It must block until release, then sample.
     #[tokio::test]
     #[ignore = "requires Postgres"]
-    async fn sample_writer_waits_for_shared_floor_lock_before_sampling_time() {
+    async fn cluster_global_sample_writer_waits_for_shared_floor_lock_before_sampling_time() {
         let (admin, pool, name) = scratch_db().await;
 
         let mut blocker = pool.begin().await.expect("begin shared-lock blocker");

crates/buzz-db/src/runtime/tests.rs

@@ -3,6 +3,7 @@ use crate::{relay_members, thread};
 use buzz_core::CommunityId;
 use metrics_util::debugging::{DebugValue, DebuggingRecorder, Snapshotter};
 use sqlx::{Connection, PgPool};
+use std::time::Instant;
 use uuid::Uuid;
 
 const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 -- local test-only credentials
@@ -360,11 +361,34 @@ async fn readiness_check_cancellation_balances_waiter_and_inflight_connection()
         querying_db
             .readiness_check_sql(
                 tokio::time::Instant::now() + std::time::Duration::from_secs(5),
-                "SELECT pg_sleep(5)",
+                "SELECT pg_sleep(1)",
             )
             .await
     });
-    tokio::time::sleep(std::time::Duration::from_millis(25)).await;
+    // Abort only once the server is running the query. sqlx's release ping
+    // waits for it to finish, so the connection returns about 1 s later,
+    // well inside the 5 s recovery bound below.
+    let observer = PgPool::connect(&database_url)
+        .await
+        .expect("connect activity observer");
+    tokio::time::timeout(std::time::Duration::from_secs(5), async {
+        loop {
+            let running: bool = sqlx::query_scalar(
+                "SELECT EXISTS(SELECT 1 FROM pg_stat_activity \
+                 WHERE datname = current_database() AND state = 'active' \
+                   AND query = 'SELECT pg_sleep(1)')",
+            )
+            .fetch_one(&observer)
+            .await
+            .expect("inspect readiness query");
+            if running {
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+        }
+    })
+    .await
+    .expect("readiness query reached the server");
     querying.abort();
     assert!(querying
         .await
@@ -3102,6 +3126,22 @@ fn writer_statement_timeout_is_always_bounded() {
     assert_eq!(MAX_STATEMENT_TIMEOUT_MS, 300_000);
 }
 
+/// Poll until the reader pool holds exactly `target` connections, all idle.
+/// Returned and floor-maintenance connections land asynchronously, from
+/// tasks sqlx spawns on drop.
+async fn await_idle_floor(read_pool: &PgPool, target: u32) {
+    let deadline = Instant::now() + Duration::from_secs(10);
+    while read_pool.size() != target || read_pool.num_idle() != target as usize {
+        assert!(
+            Instant::now() < deadline,
+            "reader pool never settled at {target} idle; size={} idle={}",
+            read_pool.size(),
+            read_pool.num_idle()
+        );
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+}
+
 /// Through the production constructors, every writer session carries the
 /// clamped statement timeout and every replica session the 20 s read
 /// deadline. Mutation oracles: dropping the clamp leaves `0`/`10min` on the
@@ -3131,7 +3171,11 @@ async fn session_statement_timeouts_bound_writer_and_reader_sessions() {
             .await
             .expect("writer timeout");
         assert_eq!(writer, writer_expected, "writer, configured {configured}");
+        // Warm the reader the way the relay boots it: the routed-read acquire
+        // budget is 150 ms, and a cold dial on a busy server can exceed it.
         let reader_pool = db.read_pool.clone().expect("reader configured");
+        db.read_pool_boot_ping_for_tests().await;
+        await_idle_floor(&reader_pool, 1).await;
         let reader: String = sqlx::query_scalar("SHOW statement_timeout")
             .fetch_one(&reader_pool)
             .await
@@ -3405,7 +3449,7 @@ async fn replica_floor_writer_transaction_holds_shared_lock() {
 
 #[tokio::test]
 #[ignore = "requires Postgres"]
-async fn replica_floor_probe_waits_for_shared_writer_and_records_after_release() {
+async fn cluster_global_replica_floor_probe_waits_for_shared_writer_and_records_after_release() {
     let admin = PgPool::connect(&admin_url().await)
         .await
         .expect("connect admin");

crates/buzz-db/src/runtime/tests/reader_pool_postgres_tests.rs

@@ -3,7 +3,6 @@
 //! reader).
 
 use super::*;
-use std::time::Instant;
 
 /// A `Db` whose writer and lazy reader both point at one scratch database,
 /// with the fence open so routed reads attempt the reader.
@@ -32,22 +31,6 @@ async fn same_db_reader(prefix: &str, read_max: u32) -> (PgPool, Db, String, Str
     (admin, db, wname, scratch_url)
 }
 
-/// Poll until the reader pool holds exactly `target` connections, all idle.
-/// Returned and floor-maintenance connections land asynchronously, from
-/// tasks sqlx spawns on drop.
-async fn await_idle_floor(read_pool: &PgPool, target: u32) {
-    let deadline = Instant::now() + Duration::from_secs(10);
-    while read_pool.size() != target || read_pool.num_idle() != target as usize {
-        assert!(
-            Instant::now() < deadline,
-            "reader pool never settled at {target} idle; size={} idle={}",
-            read_pool.size(),
-            read_pool.num_idle()
-        );
-        tokio::time::sleep(Duration::from_millis(10)).await;
-    }
-}
-
 /// Counters keyed by `(name, decision, reason)` labels, drained: each
 /// snapshot resets the recorder's counters to zero.
 fn drain_counters(
@@ -351,7 +334,20 @@ async fn stalled_reader_sessions_release_their_slots_for_new_connections() {
     await_idle_floor(&read_pool, 1).await;
     {
         let first = read_pool.acquire().await.expect("first warm connection");
-        let second = read_pool.acquire().await.expect("second warm connection");
+        // The floor maintenance stops at one, so the second slot is a cold
+        // dial inside the 150ms acquire budget, which a loaded runner can
+        // outlast. Warm-up is not under test; the stall below is.
+        let second = tokio::time::timeout(Duration::from_secs(5), async {
+            loop {
+                match read_pool.acquire().await {
+                    Err(sqlx::Error::PoolTimedOut) => continue,
+                    connection => break connection,
+                }
+            }
+        })
+        .await
+        .expect("second reader connects within 5s")
+        .expect("second warm connection");
         drop((first, second));
     }
     await_idle_floor(&read_pool, 2).await;

crates/buzz-db/src/store/entity_deletion/leases.rs

@@ -101,23 +101,41 @@ impl EntityDeletionStore {
         }))
     }
 
-    /// Renew an unexpired lease. An expired or superseded worker cannot revive it.
+    /// Renew this lease. A superseded lease is never renewed; an expired lease
+    /// can be renewed only while nobody has reclaimed it.
+    ///
+    /// One autocommit statement: the worker drops this future whenever its
+    /// stage handler finishes first, and sqlx 0.9 queues no `ROLLBACK` for a
+    /// `begin()` dropped while its `BEGIN` is in flight, so the pooled
+    /// connection would carry an open transaction into the next write.
     pub async fn heartbeat(&self, lease: &EntityLease, duration: Duration) -> Result<()> {
         let millis = duration_millis(duration)?;
-        let mut tx = self.pool.begin().await?;
-        lock_owned(&mut tx, lease).await?;
-        sqlx::query(
+        // `lock_owned`'s predicate. After a row-lock wait, Postgres re-checks it
+        // only if the holder committed an update of this row, so a reclaim's new
+        // generation always wins. A holder that only locked the row (e.g.
+        // `save_stage_success`) or rolled back leaves the pre-wait expiry check
+        // in force, so a heartbeat that waits past expiry still renews. No one
+        // can reclaim in that window: `claim` skips locked rows.
+        let renewed = sqlx::query(
             "UPDATE entity_deletion_requests
-             SET lease_expires_at = clock_timestamp() + $2 * interval '1 millisecond',
+             SET lease_expires_at = clock_timestamp() + $4 * interval '1 millisecond',
                  heartbeat_at = clock_timestamp()
-             WHERE id = $1",
+             WHERE id = $1 AND lease_owner = $2 AND lease_generation = $3
+               AND status IN ('approved', 'running', 'retry_wait', 'completed')
+               AND lease_expires_at > clock_timestamp()",
         )
         .bind(lease.request_id)
+        .bind(&lease.owner)
+        .bind(lease.generation)
         .bind(millis)
-        .execute(&mut *tx)
-        .await?;
-        tx.commit().await?;
-        Ok(())
+        .execute(&self.pool)
+        .await?
+        .rows_affected();
+        if renewed == 1 {
+            Ok(())
+        } else {
+            Err(stale_lease(lease))
+        }
     }
 
     /// Release this live claim after a pause or shutdown, without advancing its stage.

crates/buzz-db/src/store/entity_deletion_postgres_tests.rs

@@ -1,4 +1,5 @@
 use super::*;
+use std::time::Duration;
 
 async fn store() -> EntityDeletionStore {
     EntityDeletionStore {
@@ -446,6 +447,85 @@ async fn approval_refuses_to_repair_missing_evidence() {
     );
 }
 
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn dropped_heartbeat_leaves_no_open_transaction_for_the_release() {
+    let store = store().await;
+    let request = users(&store)
+        .submit("demo:dropped-heartbeat", "operator-a", None)
+        .await
+        .unwrap();
+    store
+        .approve(
+            request.id,
+            "operator-b",
+            request.inventory_digest.as_deref().unwrap(),
+        )
+        .await
+        .unwrap();
+    let lease = store
+        .claim(request.id, "worker", Duration::from_secs(60))
+        .await
+        .unwrap()
+        .expect("claim approved request");
+
+    let direct = store.pool.connect_options();
+    let proxy =
+        crate::test_support::PausingProxy::spawn(direct.get_host().to_owned(), direct.get_port())
+            .await;
+    // One connection, so the release reuses the heartbeat's. No acquire ping,
+    // so the heartbeat's first statement is the one in flight when it drops.
+    let proxied = EntityDeletionStore {
+        pool: sqlx::postgres::PgPoolOptions::new()
+            .max_connections(1)
+            .test_before_acquire(false)
+            .connect_with((*direct).clone().host("127.0.0.1").port(proxy.port))
+            .await
+            .expect("connect through proxy"),
+    };
+    sqlx::query("SELECT 1")
+        .execute(&proxied.pool)
+        .await
+        .expect("warm proxied connection");
+    // sqlx returns the connection from a spawned task after a ping; pausing
+    // before it lands would stall the heartbeat in `acquire` instead.
+    tokio::time::timeout(Duration::from_secs(5), async {
+        while proxied.pool.num_idle() != 1 {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("warm connection returned to the pool");
+
+    // The worker's stage `select!` drops the heartbeat whenever the handler
+    // wins. Holding the bytes drops it with its first round trip in flight.
+    proxy.set_paused(true);
+    assert!(tokio::time::timeout(
+        Duration::from_millis(100),
+        proxied.heartbeat(&lease, Duration::from_secs(60)),
+    )
+    .await
+    .is_err());
+    proxy.set_paused(false);
+    assert!(proxied.release(&lease).await.unwrap());
+
+    let lease_owner: Option<String> =
+        sqlx::query_scalar("SELECT lease_owner FROM entity_deletion_requests WHERE id = $1")
+            .bind(request.id)
+            .fetch_one(&store.pool)
+            .await
+            .unwrap();
+    assert_eq!(lease_owner, None, "the reported release must be committed");
+    let open: i64 = sqlx::query_scalar(
+        "SELECT count(*) FROM pg_stat_activity
+         WHERE datname = current_database() AND state = 'idle in transaction'",
+    )
+    .fetch_one(&store.pool)
+    .await
+    .unwrap();
+    assert_eq!(open, 0, "no pooled session may sit inside a transaction");
+}
+
 fn users(store: &EntityDeletionStore) -> crate::user_deletion::UserDeletionStore {
     Db::from_pool(store.pool.clone()).user_deletion_store()
 }

crates/buzz-deletion/src/user/worker.rs

@@ -121,6 +121,8 @@ async fn run_stage_with_heartbeat(
     // Poll renewal alongside the handler. Awaiting renewal inside a selected
     // timer branch would stop polling a handler that may hold the parent lock.
     // Both futures live here, so cancellation leaves no detached heartbeat task.
+    // A handler can finish mid-heartbeat; `heartbeat` is one autocommit
+    // statement so that drop cannot strand an open transaction on the pool.
     let renewal = async {
         loop {
             interval.tick().await;

crates/buzz-relay/src/connection.rs

@@ -3798,8 +3798,11 @@ pub(crate) mod terminal_close_delivery_tests {
         #[ignore = "requires Postgres"]
         async fn upgrade_would_deny_line_is_written_at_auth_while_open() {
             let log = crate::nip_fi_shadow::tests::WouldDenyLog::capture();
-            let state = crate::state::tests::test_state_with_database_url(
+            // sqlx's default acquire timeout: a first connect to a busy CI
+            // database can outlast the unreachable-target helper's 100 ms.
+            let state = crate::state::tests::test_state_with_database_url_and_pool_options(
                 &crate::test_support::database_url(),
+                sqlx::postgres::PgPoolOptions::new(),
             )
             .await;
             let community = Uuid::new_v4();
@@ -4158,7 +4161,7 @@ pub(crate) mod terminal_close_delivery_tests {
             .unwrap();
         let ((), lines) = crate::nip_fi_shadow::tests::would_deny_lines(|| {
             rt.block_on(async {
-                let state = crate::state::tests::test_state_with_database_url(
+                let state = crate::state::tests::test_state_with_unreachable_database_url(
                     "postgres://127.0.0.1:1/none",
                 )
                 .await;

crates/buzz-relay/src/handlers/auth.rs

@@ -3939,7 +3939,7 @@ mod tests {
         let recorder = metrics_util::debugging::DebuggingRecorder::new();
         let snapshotter = recorder.snapshotter();
         let _recorder_guard = metrics::set_default_local_recorder(&recorder);
-        let state = crate::state::tests::test_state_with_database_url(
+        let state = crate::state::tests::test_state_with_unreachable_database_url(
             "postgres://buzz:buzz_dev@127.0.0.1:1/buzz",
         )
         .await;

crates/buzz-relay/src/handlers/group_state_postgres_tests.rs

@@ -13,9 +13,9 @@ use crate::state::AppState;
 use buzz_auth::Scope;
 
 /// Test state on the shared test database. The pool keeps sqlx's default
-/// acquire timeout: `test_state_with_database_url` waits only 100 ms so
-/// closed-port tests fail fast, and a first connect to a busy CI database can
-/// take longer than that (`PoolTimedOut`).
+/// acquire timeout: `test_state_with_unreachable_database_url` waits only
+/// 100 ms so closed-port tests fail fast, and a first connect to a busy CI
+/// database can take longer than that (`PoolTimedOut`).
 pub(crate) async fn state() -> Arc<AppState> {
     crate::state::tests::test_state_with_database_url_and_pool_options(
         &crate::test_support::database_url(),

crates/buzz-relay/src/handlers/ingest.rs

@@ -3818,7 +3818,7 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn check_channel_write_denies_when_channel_lookup_fails() {
-        let state = crate::state::tests::test_state_with_database_url(
+        let state = crate::state::tests::test_state_with_unreachable_database_url(
             "postgres://buzz:buzz_dev@127.0.0.1:1/buzz", // sadscan:disable np.postgres.1
         )
         .await;

crates/buzz-relay/src/nip11.rs

@@ -524,8 +524,10 @@ mod tests {
     #[tokio::test]
     async fn shadow_serves_the_same_nip11_document_as_enforce() {
         use buzz_auth::NipFiMode;
-        let base =
-            crate::state::tests::test_state_with_database_url("postgres://127.0.0.1:1/none").await;
+        let base = crate::state::tests::test_state_with_unreachable_database_url(
+            "postgres://127.0.0.1:1/none",
+        )
+        .await;
         let mut docs = Vec::new();
         for mode in [NipFiMode::Off, NipFiMode::Enforce, NipFiMode::Shadow] {
             let mut state = (*base).clone();

crates/buzz-relay/src/nip_fi_resolver/tests/router_tests.rs

@@ -146,8 +146,10 @@ struct Pod {
 }
 
 async fn pod(store: &Arc<Bindings>, clock: &Arc<ManualClock>, mode: NipFiMode) -> Pod {
-    let base =
-        crate::state::tests::test_state_with_database_url("postgres://127.0.0.1:1/none").await;
+    let base = crate::state::tests::test_state_with_unreachable_database_url(
+        "postgres://127.0.0.1:1/none",
+    )
+    .await;
     pod_on(base, store, clock, mode).await
 }
 

crates/buzz-relay/src/nip_fi_shadow_session.rs

@@ -393,10 +393,11 @@ pub(crate) mod tests {
     /// timeout, so a path that reaches Postgres fails fast whether or not
     /// one runs locally.
     pub(crate) async fn shadow_state(denied: Option<PublicKey>) -> AppState {
-        let mut state =
-            (*crate::state::tests::test_state_with_database_url("postgres://127.0.0.1:1/none")
-                .await)
-                .clone();
+        let mut state = (*crate::state::tests::test_state_with_unreachable_database_url(
+            "postgres://127.0.0.1:1/none",
+        )
+        .await)
+            .clone();
         let capacity = vec![IssuerCapacity {
             issuer: "test-issuer".to_owned(),
             capacity: 8,