test(buzz-db): fix two PostgreSQL-lane timing flakes (#8472)

## Summary

Two timing flakes in the PostgreSQL lane, one fix each.

### Deletion lock order

`abort_serializes_before_fence_without_deadlock` failed on main ([run
39330176879](https://github.com/block/buzz/actions/runs/39330176879)):
`post-lock lease verification rejects aborted request: 1`.

The test slept 50ms and assumed `abort` had queued on the community
advisory lock by then. On a slow runner, `fence` queued first, won the
lock, and succeeded.

The test now waits until `pg_locks` shows each contender waiting on that
community's lock, `abort` first and then `fence`. Postgres grants queued
advisory locks in arrival order, so `abort` always wins. The sleeps and
their timing guess are gone.

### Saturated-reader fixture

`routed_fallback_spends_one_acquire_budget_when_aurora_cache_is_cold`
failed this PR's first run ([run
39335038111](https://github.com/block/buzz/actions/runs/39335038111)):
`establish the reader's sole connection: PoolTimedOut`. Setup opens the
reader's only connection through the production lazy pool, which has a
150ms acquire timeout. On a loaded runner that first connect took
longer, so the test failed before reaching the budget it measures. Setup
now retries on `PoolTimedOut` within 5s. The pool and its 150ms budget
are unchanged for the routed read under test.

Test-only. Production code is unchanged.

## Validation

- Deletion test against an isolated local Postgres 17 database: 40/40
passes. Saturated-reader test: 15/15.
- `cargo fmt --check`, `cargo clippy -p buzz-db --tests -D warnings`,
pre-push Rust unit tests: pass.
- Three other `store::deletion::postgres_tests` fail locally on
unmodified `main` as well (local environment). CI's PostgreSQL lane is
the authoritative check for them.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Wes <wesbillman@users.noreply.github.com>
Co-authored-by: Carl <32a2e2c9d428ee08902cab75d956da2c1d235a22d4766b0dd4138bf6e2e5db1d@buzz.block.builderlab.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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

@@ -1982,11 +1982,20 @@ async fn routed_fallback_spends_one_acquire_budget_when_aurora_cache_is_cold() {
     db.set_replica_read_max_age_for_tests(Some(Duration::from_secs(5)));
 
     let read_pool = db.read_pool.clone().expect("reader pool configured");
-    // Establish and hold the reader's only connection: saturated.
-    let held = read_pool
-        .acquire()
-        .await
-        .expect("establish the reader's sole connection");
+    // Establish and hold the reader's only connection: saturated. A fresh
+    // connection can outlast the 150ms acquire budget on a loaded runner, and
+    // the budget is under test only once the connection is held.
+    let held = tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            match read_pool.acquire().await {
+                Err(sqlx::Error::PoolTimedOut) => continue,
+                connection => break connection,
+            }
+        }
+    })
+    .await
+    .expect("reader connects within 5s")
+    .expect("establish the reader's sole connection");
     assert_eq!(
         db.read_max_connections, 1,
         "reader max must report 1 for this fixture to test saturation"

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

@@ -4833,6 +4833,29 @@ DROP FUNCTION community_write_fence_excluded_table(NAME);
         .expect("contender reached community row lock");
     }
 
+    async fn wait_for_community_lock_waiters(db: &Db, community: CommunityId, waiters: i64) {
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            loop {
+                let queued: i64 = sqlx::query_scalar(
+                    "SELECT count(*) FROM pg_locks \
+                     WHERE locktype = 'advisory' AND objsubid = 1 AND NOT granted \
+                       AND ((classid::bigint << 32) | objid::bigint) \
+                           = community_deletion_lock_key($1)",
+                )
+                .bind(community.as_uuid())
+                .fetch_one(&db.pool)
+                .await
+                .expect("inspect community lock queue");
+                if queued >= waiters {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("contenders queued on community lock");
+    }
+
     async fn membership_roles(db: &Db, community: CommunityId) -> Vec<(String, String)> {
         sqlx::query_as(
             "SELECT pubkey, role FROM relay_members WHERE community_id = $1 ORDER BY pubkey",
@@ -6778,7 +6801,9 @@ DROP FUNCTION community_write_fence_excluded_table(NAME);
                 .abort(request.id, "operator", "race recovery")
                 .await
         });
-        tokio::time::sleep(Duration::from_millis(50)).await;
+        // Postgres grants queued advisory locks in arrival order, so abort must
+        // be queued before fence is spawned for the outcome to be deterministic.
+        wait_for_community_lock_waiters(&db, request.community_id, 1).await;
         assert!(
             !aborting.is_finished(),
             "abort must wait for the community lock"
@@ -6786,7 +6811,7 @@ DROP FUNCTION community_write_fence_excluded_table(NAME);
         let forward_store = store.clone();
         let lease = claim.lease.clone();
         let forwarding = tokio::spawn(async move { forward_store.fence(&lease).await });
-        tokio::time::sleep(Duration::from_millis(50)).await;
+        wait_for_community_lock_waiters(&db, request.community_id, 2).await;
         assert!(
             !forwarding.is_finished(),
             "fence must queue on the same lock"