From 6e6173acd58029bac800583c4cc7482dafed693a Mon Sep 17 00:00:00 2001 From: Wes Date: Mon, 19 Oct 2026 16:40:43 -0700 Subject: [PATCH] test(buzz-db): fix two PostgreSQL-lane timing flakes (#8472) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 Co-authored-by: Carl <32a2e2c9d428ee08902cab75d956da2c1d235a22d4766b0dd4138bf6e2e5db1d@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 --- crates/buzz-db/src/runtime/tests.rs | 19 +++++++++++++----- crates/buzz-db/src/store/deletion.rs | 29 ++++++++++++++++++++++++++-- 2 files changed, 41 insertions(+), 7 deletions(-) diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index d35923623d..a1bfc3bd26 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/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" diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 1e1e68cb05..fe8ad4adf9 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/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"