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"