From e3f8f9a218f113d9d11849736640b909383faf18 Mon Sep 17 00:00:00 2001 From: Tee8z Date: Tue, 20 Oct 2026 16:31:30 -0400 Subject: [PATCH] fix(ci): stabilize PostgreSQL concurrency tests Reserve all nextest workers for cluster-wide activity scans, including the floor-lock probes. Wait for server-side cleanup when the usage-metrics leader disconnects. Synchronize readiness cancellation with a blocked server query and await rollback after rejected deletion updates before claiming rows. Drive the reader acquisition budget with Tokio's controlled clock instead of a wall-clock upper bound. Preserve exclusive ownership, cancellation recovery, exact inventory approval, and single-budget fallback assertions. Signed-off-by: Tee8z --- .config/nextest.toml | 9 +- crates/buzz-db/TESTING.md | 5 +- crates/buzz-db/src/runtime/replica_fence.rs | 2 +- crates/buzz-db/src/runtime/tests.rs | 109 ++++++++++++-------- crates/buzz-db/src/store/deletion.rs | 35 ++++++- crates/buzz-db/src/store/usage.rs | 25 +++-- 6 files changed, 124 insertions(+), 61 deletions(-) diff --git a/.config/nextest.toml b/.config/nextest.toml index 4481ebc6ba3..14596e0d880 100644 --- a/.config/nextest.toml +++ b/.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 diff --git a/crates/buzz-db/TESTING.md b/crates/buzz-db/TESTING.md index bebe09dd98e..fb65c36af64 100644 --- a/crates/buzz-db/TESTING.md +++ b/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 diff --git a/crates/buzz-db/src/runtime/replica_fence.rs b/crates/buzz-db/src/runtime/replica_fence.rs index 520ff9e8698..c812856c1f7 100644 --- a/crates/buzz-db/src/runtime/replica_fence.rs +++ b/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"); diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 3998e10b3b0..1222705afb8 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/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"); diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index eef0ac8b0b9..379784dc3b5 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/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 diff --git a/crates/buzz-db/src/store/usage.rs b/crates/buzz-db/src/store/usage.rs index f2d7f996aae..ca0d350e310 100644 --- a/crates/buzz-db/src/store/usage.rs +++ b/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;