fix(ci): stabilize PostgreSQL concurrency tests

.config/nextest.toml

@@ -24,14 +24,11 @@ fail-fast = false
 # wrapper retained one database per concurrently running test process.
 test-threads = 8
 
-[test-groups.postgres-cluster-global]
-# These tests inspect cluster-wide activity or create least-privilege sessions,
-# so database-per-test isolation alone cannot make them independent.
-max-threads = 1
-
 [[profile.postgres-ci.overrides]]
 filter = "test(/cluster_global_/)"
-test-group = "postgres-cluster-global"
+# Activity scans also observe ordinary tests connecting to the same cluster.
+# Reserve every worker so those tests and their wrappers cannot overlap a scan.
+threads-required = "num-test-threads"
 
 [[profile.postgres-ci.scripts]]
 # Script filters are separate from default-filter: they attach the setup and

crates/buzz-db/TESTING.md

@@ -41,8 +41,9 @@ migration-applied coverage rather than making the bootstrap implicit.
 
 Tests that inspect cluster-wide PostgreSQL state or open least-privilege
 sessions include `cluster_global_` in the function name. Migration-backed cases
-use `migration_schema_cluster_global_`. Nextest serializes this small group
-because separate databases still share `pg_stat_activity` and roles.
+use `migration_schema_cluster_global_`. Nextest reserves all test workers for
+each of these tests, preventing overlap with ordinary database tests and their
+wrappers. Separate databases still share `pg_stat_activity` and roles.
 
 ## Run the lane locally
 

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

@@ -316,24 +316,21 @@ async fn readiness_check_cancellation_balances_waiter_and_inflight_connection()
         .connect(&database_url)
         .await
         .expect("connect cancellation readiness test pool");
-    let held = pool
+    let mut held = pool
         .acquire()
         .await
         .expect("hold sole connection before waiter cancellation");
+    let backend_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()")
+        .fetch_one(&mut *held)
+        .await
+        .expect("identify readiness connection");
     let db = Db::from_pool(pool);
 
-    let waiting_db = db.clone();
-    let waiting = tokio::spawn(async move {
-        waiting_db
-            .readiness_check(tokio::time::Instant::now() + std::time::Duration::from_secs(5))
-            .await
-    });
-    tokio::time::sleep(std::time::Duration::from_millis(25)).await;
-    waiting.abort();
-    assert!(waiting
-        .await
-        .expect_err("waiting check must be cancelled")
-        .is_cancelled());
+    let mut waiting = Box::pin(
+        db.readiness_check(tokio::time::Instant::now() + std::time::Duration::from_secs(5)),
+    );
+    assert!(futures_util::poll!(waiting.as_mut()).is_pending());
+    drop(waiting);
     drop(held);
 
     assert_eq!(
@@ -343,21 +340,52 @@ async fn readiness_check_cancellation_balances_waiter_and_inflight_connection()
         "cancelled pool waiter must not consume the released connection"
     );
 
+    // Hold the server query until cancellation is observed. A sleep with the
+    // same duration as the recovery deadline races SQLx's asynchronous drain
+    // of the cancelled query before it returns the connection to the pool.
+    let mut blocker = sqlx::PgConnection::connect(&database_url)
+        .await
+        .expect("connect readiness query blocker");
+    sqlx::query("SELECT pg_advisory_lock(86299171)")
+        .execute(&mut blocker)
+        .await
+        .expect("block readiness query");
     let querying_db = db.clone();
     let querying = tokio::spawn(async move {
         querying_db
             .readiness_check_sql(
-                tokio::time::Instant::now() + std::time::Duration::from_secs(5),
-                "SELECT pg_sleep(5)",
+                tokio::time::Instant::now() + std::time::Duration::from_secs(30),
+                "SELECT pg_advisory_xact_lock(86299171)",
             )
             .await
     });
-    tokio::time::sleep(std::time::Duration::from_millis(25)).await;
+    tokio::time::timeout(std::time::Duration::from_secs(5), async {
+        loop {
+            let blocked: bool = sqlx::query_scalar(
+                "SELECT EXISTS (SELECT 1 FROM pg_stat_activity \
+                 WHERE pid = $1 AND wait_event_type = 'Lock' AND wait_event = 'advisory')",
+            )
+            .bind(backend_pid)
+            .fetch_one(&mut blocker)
+            .await
+            .expect("observe blocked readiness query");
+            if blocked {
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("readiness query must reach the server before cancellation");
     querying.abort();
     assert!(querying
         .await
         .expect_err("querying check must be cancelled")
         .is_cancelled());
+    sqlx::query("SELECT pg_advisory_unlock(86299171)")
+        .execute(&mut blocker)
+        .await
+        .expect("release cancelled readiness query");
 
     let recovered = tokio::time::timeout(std::time::Duration::from_secs(5), async {
         loop {
@@ -1906,11 +1934,10 @@ async fn is_relay_member_is_bounded_routed_and_fails_closed() {
 /// directly, because `buzz_db_route_decision` is emitted by `route_read`
 /// — a direct call would prove the timing but never emit the label.
 ///
-/// Timing uses an upper bound of 2x the budget minus a margin: it must
-/// fail for two stacked budgets (~300ms) while tolerating scheduler
-/// jitter on one (~150ms). Asserting a lower bound too would pin the
-/// budget's own value, which `reader_acquire_timeout_is_the_documented_budget`
-/// already covers.
+/// Pause Tokio time only while the reader is saturated. After advancing one
+/// acquire budget, the production route must already have selected the writer.
+/// Resume time before awaiting writer I/O so PostgreSQL scheduling cannot
+/// affect the budget assertion.
 #[tokio::test(flavor = "current_thread")]
 #[ignore = "requires Postgres"]
 async fn routed_fallback_spends_one_acquire_budget_when_aurora_cache_is_cold() {
@@ -1975,24 +2002,17 @@ async fn routed_fallback_spends_one_acquire_budget_when_aurora_cache_is_cold() {
     // a multi-thread runtime the emit could land on a worker where no
     // local recorder is installed and the label assertions would vacuously
     // see an empty snapshot.
-    let start = std::time::Instant::now();
-    let count = {
-        let _guard = metrics::set_default_local_recorder(&recorder);
-        db.count_events_routed("one_budget_probe", &query).await
-    }
-    .expect("writer fallback still answers the read");
-    let elapsed = start.elapsed();
-
-    assert_eq!(count, 0, "writer answered on an empty scratch database");
-    assert!(
-        elapsed < Duration::from_millis(250),
-        "routed fallback must spend ONE {}ms acquire budget, not two; took {}ms",
-        Db::READER_ACQUIRE_TIMEOUT.as_millis(),
-        elapsed.as_millis()
-    );
-
-    let reasons: std::collections::HashMap<(String, String), u64> = snapshotter
-        .snapshot()
+    let _guard = metrics::set_default_local_recorder(&recorder);
+    tokio::time::pause();
+    let mut read = Box::pin(db.count_events_routed("one_budget_probe", &query));
+    assert!(futures_util::poll!(read.as_mut()).is_pending());
+    // Tokio timers have millisecond granularity; cross the deadline by one tick.
+    tokio::time::advance(Db::READER_ACQUIRE_TIMEOUT + Duration::from_millis(1)).await;
+    let completion = futures_util::poll!(read.as_mut());
+    let after_one_budget = snapshotter.snapshot();
+    tokio::time::resume();
+
+    let reasons: std::collections::HashMap<(String, String), u64> = after_one_budget
         .into_vec()
         .into_iter()
         .filter(|(key, ..)| key.key().name() == "buzz_db_route_decision")
@@ -2015,7 +2035,7 @@ async fn routed_fallback_spends_one_acquire_budget_when_aurora_cache_is_cold() {
     assert_eq!(
         reasons.get(&("writer".to_owned(), "reader_acquire_timeout".to_owned())),
         Some(&1),
-        "saturated reader must fall back as writer/reader_acquire_timeout; got {reasons:?}"
+        "saturated reader must choose writer/reader_acquire_timeout after one budget; got {reasons:?}"
     );
     // `reader_validation_error` would mean we misclassified a timeout as a
     // broken reader, and `pool_busy` is the retired name — neither may
@@ -2027,6 +2047,13 @@ async fn routed_fallback_spends_one_acquire_budget_when_aurora_cache_is_cold() {
         "no reader_validation_error or retired pool_busy label may be emitted; got {reasons:?}"
     );
 
+    let count = match completion {
+        std::task::Poll::Ready(result) => result,
+        std::task::Poll::Pending => read.await,
+    }
+    .expect("writer fallback still answers the read");
+    assert_eq!(count, 0, "writer answered on an empty scratch database");
+
     drop(held);
     drop_scratch_db(&admin, db.pool.clone(), &wname).await;
 }
@@ -3256,7 +3283,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/store/deletion.rs

@@ -6243,6 +6243,11 @@ mod postgres_tests {
         assert_eq!(current_schema.scoped_tables, inventory.schema.scoped_tables);
         assert_eq!(current_schema.fenced_tables, inventory.schema.fenced_tables);
 
+        let mut rejected = db
+            .pool
+            .begin()
+            .await
+            .expect("begin rejected approval insert");
         let mismatched_insert = sqlx::query(
             "INSERT INTO community_deletion_approvals \
              (request_id, community_id, inventory_digest, approved_by) \
@@ -6251,12 +6256,16 @@ mod postgres_tests {
         .bind(request.id)
         .bind(*request.community_id.as_uuid())
         .bind(vec![0_u8; 32])
-        .execute(&db.pool)
+        .execute(&mut *rejected)
         .await;
         assert!(
             mismatched_insert.is_err(),
             "a mismatched approval must be unrepresentable"
         );
+        rejected
+            .rollback()
+            .await
+            .expect("roll back rejected approval insert");
         let approved = store
             .approve(request.id, "approver-a", Some("reviewed"))
             .await
@@ -6276,28 +6285,48 @@ mod postgres_tests {
             approved.inventory_digest,
             Some(hex::encode(inventory.digest().unwrap()))
         );
+        let mut rejected = db
+            .pool
+            .begin()
+            .await
+            .expect("begin rejected approval update");
         let mismatched_approval = sqlx::query(
             "UPDATE community_deletion_approvals SET inventory_digest = $2 WHERE request_id = $1",
         )
         .bind(request.id)
         .bind(vec![0_u8; 32])
-        .execute(&db.pool)
+        .execute(&mut *rejected)
         .await;
         assert!(
             mismatched_approval.is_err(),
             "approval digest must remain database-bound to the frozen request digest"
         );
+        rejected
+            .rollback()
+            .await
+            .expect("roll back rejected approval update");
+        let mut rejected = db
+            .pool
+            .begin()
+            .await
+            .expect("begin rejected request update");
         let mismatched_request = sqlx::query(
             "UPDATE community_deletion_requests SET inventory_digest = $2 WHERE id = $1",
         )
         .bind(request.id)
         .bind(vec![1_u8; 32])
-        .execute(&db.pool)
+        .execute(&mut *rejected)
         .await;
         assert!(
             mismatched_request.is_err(),
             "the frozen request digest must remain bound to its approval"
         );
+        // ErrorResponse can arrive before PostgreSQL releases the row lock.
+        // Await rollback before the claimant uses FOR UPDATE SKIP LOCKED.
+        rejected
+            .rollback()
+            .await
+            .expect("roll back rejected request update");
         assert!(store
             .claim_specific(request.id, "executor-a", DEFAULT_LEASE_DURATION)
             .await

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

@@ -924,16 +924,25 @@ mod postgres_tests {
         );
 
         drop(leader);
-        assert!(
-            second
-                .try_lock_usage_metrics(key)
-                .await
-                .expect("lock attempt after leader drop")
-                .is_some(),
-            "dropping the detached session releases its advisory lock"
-        );
+        // Closing the socket does not wait for PostgreSQL to process the
+        // disconnect. Observe lock release instead of racing server cleanup.
+        let successor = tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            loop {
+                if let Some(leader) = second
+                    .try_lock_usage_metrics(key)
+                    .await
+                    .expect("lock attempt after leader drop")
+                {
+                    break leader;
+                }
+                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .expect("dropping the detached session releases its advisory lock");
 
         // Release any remaining session state before DROP DATABASE.
+        drop(successor);
         drop(first);
         drop(second);
         drop_scratch_db(&admin, pool, &scratch_name).await;