From dae720ed982f0360fba95e3dc5038eb32f626edd Mon Sep 17 00:00:00 2001 From: Matt Toohey Date: Thu, 29 Oct 2026 16:24:55 +1100 Subject: [PATCH] fix(db): keep lease releases durable and deflake PostgreSQL lane tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes the PostgreSQL-lane flakes from the 21-28 Oct 2026 CI audit: families B, C, D, E, F1, F2 and the §A residual. G (one hermit pgschema 504, a CI workflow change) is out of scope. Nothing on origin/main (eda26b942) or in an open or merged PR covers any of these. Local runs: macOS, OrbStack Postgres, cargo-nextest 0.9.136, Rust 1.98.1. Baselines ran from a nextest archive built before any fix (HEAD plus the B regression test). "Load" means one `yes` per core (16). B. PRODUCT FIX: a dropped lease heartbeat stranded the release. Tests: buzz-deletion user::worker::postgres_tests dummy_executor_failure_retries_same_community_stage (:47) and next_step_advances_one_handler_at_a_time_through_completion (:131) Runs: 40439041234 (main), 40399635380#3, 40413732817#1 Cause: run_stage_with_heartbeat drops the renewal future whenever the stage handler wins its select!. EntityDeletionStore::heartbeat opened a transaction. sqlx 0.9 PgTransactionManager::begin bumps transaction_depth only after BEGIN returns, so a begin() dropped mid-BEGIN queues no ROLLBACK. The release ping drains the BEGIN, and the connection goes back to the pool inside an open transaction. The worker's final autocommit `release` then ran inside it and reported success, but nothing committed. user-deletions run shares this path: the release is lost at exit, and reruns see "leased by another worker" until the lease expires (up to 60 s). Change: heartbeat is one autocommit UPDATE guarded by lock_owned's ownership/status/expiry predicate. The only callers are worker.rs and transition_postgres_tests.rs; the community loop's heartbeat is a different store. The worker comment now states why heartbeat must not open a transaction. Locking guarantee, as the leases.rs doc now states it: a superseded lease is never renewed; an expired lease can be renewed only while nobody has reclaimed it. Postgres re-checks the WHERE after a lock wait only when the holder committed an update of the row, so a holder that only took lock_owned's FOR UPDATE (save_stage_success writes entity_deletion_stage_history, never the parent row) or rolled back leaves the pre-wait expiry check in force. That is safe: claim takes the row with FOR UPDATE SKIP LOCKED while the waiting UPDATE holds the tuple lock, so no one can reclaim during the wait, and a reclaim that commits first bumps the generation, which the predicate re-checks. Only claim acts on expiry. Regression test: store::entity_deletion::postgres_tests:: dropped_heartbeat_leaves_no_open_transaction_for_the_release. It drops the production heartbeat with its BEGIN held by PausingProxy (1-connection pool, test_before_acquire(false)), then releases and asserts on a direct connection. Before -> after: regression test 3/3 FAIL (lease_owner Some("worker") after release() returned true) -> 3/3 PASS. Worker stress under load, before: 2/138 iterations FAIL at load average ~200, then 1/150 at ~25, all at :131 (3/288). After, at ~25: 300/300 iterations (2100 runs) PASS. Out of scope: the shutdown/renewal-failure branches can still drop a handler's begin() mid-BEGIN. The general fix belongs in sqlx. C. session_statement_timeouts_bound_writer_and_reader_sessions (buzz-db runtime/tests.rs:3138 "reader timeout: PoolTimedOut") Runs: 40128463987, 40314950864#1, 40374286094 (main), 40394821009#1, 40413732817#1 Cause: each config's lazy reader pool dialed cold inside the 150 ms READER_ACQUIRE_TIMEOUT. Db::new never runs the boot ping. Change: warm the reader the way the relay boots it (read_pool_boot_ping_for_tests + await_idle_floor). await_idle_floor moved from reader_pool_postgres_tests.rs to runtime/tests.rs. Before -> after, 30% of sessions delayed 250 ms by proxy, 50 iterations: 44/50 FAIL at :3138 -> 50/50 PASS. D. startup_owns_the_dependency_sampler (buzz-relay boot_lifecycle.rs) Runs: torn scrape 39865872891 (main), 39624074244#2, 40385799750#1, 40413293801#1; early exit 39704588623 Cause: metrics-exporter-prometheus 0.18.3 snapshots counters, then gauges, with no single snapshot. The scrape that first shows the completion gauge can predate the checks_total counters registered just before it. Change: assert on a second scrape taken after the gauge is seen. The counters were written before the gauge and never idle out. Fix the doc comment that claimed no window. When the relay exits early, wait_for_scraped_metric now panics with its exit status, stdout and stderr. Before -> after, 200 iterations under load: 2/200 FAIL at :857 (the scrape had the gauge and histograms but no checks_total) -> 200/200 PASS. Not reproduced: the early exit (:173). Only diagnostics were added. E. sample_writer_waits_for_shared_floor_lock_before_sampling_time (replica_fence.rs:1086) and replica_floor_probe_waits_for_shared_writer_and_records_after_release (runtime/tests.rs:3452), both "MaskedActivity { masked: 1 }" Runs: 39601763602 (main), 40391445284#1; 39825846222, 40301667418 Cause: both scan cluster-wide pg_stat_activity but were outside the serialized postgres-cluster-global group, so other lane tests' connection churn masked activity. Change: rename both with a cluster_global_ prefix (each migrates its own scratch DB, so not migration_schema_). `cargo nextest show-config test-groups --profile postgres-ci` lists both in postgres-cluster-global. Before, 50 iterations with pgbench -C -c 8 on the cluster: 2/50 iterations FAIL (3 MaskedActivity panics). After: the buzz-db lane passes 3x (613/613). Not re-run under pgbench; the group serializes lane tests only. F1. stalled_reader_sessions_release_their_slots_for_new_connections (reader_pool_postgres_tests.rs:354 "second warm connection: PoolTimedOut") Run: 40399635380#2 Cause: the floor warms only one connection, so the second warm-up acquire dials cold inside the 150 ms budget. Change: retry only that warm-up acquire on PoolTimedOut within 5 s (#8472's precedent). The stall under test keeps the production budget. Before -> after, session-delay proxy, 30 iterations: 10/30 FAIL at :354 -> 30/30 PASS. F2. readiness_check_cancellation_balances_waiter_and_inflight_connection (runtime/tests.rs:391 "cancelled in-flight query must return or replace its connection: Elapsed(())") Run: 39785370353 Cause: the test aborted pg_sleep(5) after a fixed 25 ms sleep. sqlx's release ping waits for the query, so recovery lands at server start + 5 s, against a bound of abort + 5 s (~22 ms margin locally). Change: poll pg_stat_activity until the query is active, then abort. Use pg_sleep(1), so recovery takes about 1 s inside the unchanged 5 s bound. Before -> after, a proxy delaying only the Bind/Execute chunk after pg_sleep's Parse by 40 ms: 10/10 FAIL at :391 -> 10/10 PASS. §A residual. upgrade_would_deny_line_is_written_at_auth_while_open (buzz-relay connection.rs) used the 100 ms-acquire test_state_with_database_url against the real test database. No failure observed yet; not reproduced. It now uses test_state_with_database_url_and_pool_options with sqlx's default timeout. The 100 ms helper is renamed test_state_with_unreachable_database_url; all of its remaining callers use closed ports. Co-Authored-By: Claude Opus 5.5 Co-Authored-By: Claude Fable 5.1 Signed-off-by: Matt Toohey --- crates/buzz-db/src/runtime/replica_fence.rs | 2 +- crates/buzz-db/src/runtime/tests.rs | 50 +++++++++++- .../tests/reader_pool_postgres_tests.rs | 32 ++++---- .../src/store/entity_deletion/leases.rs | 38 ++++++--- .../store/entity_deletion_postgres_tests.rs | 80 +++++++++++++++++++ crates/buzz-deletion/src/user/worker.rs | 2 + crates/buzz-relay/src/connection.rs | 7 +- crates/buzz-relay/src/handlers/auth.rs | 2 +- .../handlers/group_state_postgres_tests.rs | 6 +- crates/buzz-relay/src/handlers/ingest.rs | 2 +- crates/buzz-relay/src/nip11.rs | 6 +- .../src/nip_fi_resolver/tests/router_tests.rs | 6 +- .../buzz-relay/src/nip_fi_shadow_session.rs | 9 ++- crates/buzz-relay/src/readiness.rs | 2 +- crates/buzz-relay/src/state.rs | 17 ++-- crates/buzz-relay/tests/boot_lifecycle.rs | 19 +++-- 16 files changed, 220 insertions(+), 60 deletions(-) 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 c9e1c6dc36a..76a77d2a43c 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -3,6 +3,7 @@ use crate::{relay_members, thread}; use buzz_core::CommunityId; use metrics_util::debugging::{DebugValue, DebuggingRecorder, Snapshotter}; use sqlx::{Connection, PgPool}; +use std::time::Instant; use uuid::Uuid; const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 -- local test-only credentials @@ -360,11 +361,34 @@ async fn readiness_check_cancellation_balances_waiter_and_inflight_connection() querying_db .readiness_check_sql( tokio::time::Instant::now() + std::time::Duration::from_secs(5), - "SELECT pg_sleep(5)", + "SELECT pg_sleep(1)", ) .await }); - tokio::time::sleep(std::time::Duration::from_millis(25)).await; + // Abort only once the server is running the query. sqlx's release ping + // waits for it to finish, so the connection returns about 1 s later, + // well inside the 5 s recovery bound below. + let observer = PgPool::connect(&database_url) + .await + .expect("connect activity observer"); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let running: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pg_stat_activity \ + WHERE datname = current_database() AND state = 'active' \ + AND query = 'SELECT pg_sleep(1)')", + ) + .fetch_one(&observer) + .await + .expect("inspect readiness query"); + if running { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + }) + .await + .expect("readiness query reached the server"); querying.abort(); assert!(querying .await @@ -3102,6 +3126,22 @@ fn writer_statement_timeout_is_always_bounded() { assert_eq!(MAX_STATEMENT_TIMEOUT_MS, 300_000); } +/// Poll until the reader pool holds exactly `target` connections, all idle. +/// Returned and floor-maintenance connections land asynchronously, from +/// tasks sqlx spawns on drop. +async fn await_idle_floor(read_pool: &PgPool, target: u32) { + let deadline = Instant::now() + Duration::from_secs(10); + while read_pool.size() != target || read_pool.num_idle() != target as usize { + assert!( + Instant::now() < deadline, + "reader pool never settled at {target} idle; size={} idle={}", + read_pool.size(), + read_pool.num_idle() + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } +} + /// Through the production constructors, every writer session carries the /// clamped statement timeout and every replica session the 20 s read /// deadline. Mutation oracles: dropping the clamp leaves `0`/`10min` on the @@ -3131,7 +3171,11 @@ async fn session_statement_timeouts_bound_writer_and_reader_sessions() { .await .expect("writer timeout"); assert_eq!(writer, writer_expected, "writer, configured {configured}"); + // Warm the reader the way the relay boots it: the routed-read acquire + // budget is 150 ms, and a cold dial on a busy server can exceed it. let reader_pool = db.read_pool.clone().expect("reader configured"); + db.read_pool_boot_ping_for_tests().await; + await_idle_floor(&reader_pool, 1).await; let reader: String = sqlx::query_scalar("SHOW statement_timeout") .fetch_one(&reader_pool) .await @@ -3405,7 +3449,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/runtime/tests/reader_pool_postgres_tests.rs b/crates/buzz-db/src/runtime/tests/reader_pool_postgres_tests.rs index a2549715e08..2360b40d325 100644 --- a/crates/buzz-db/src/runtime/tests/reader_pool_postgres_tests.rs +++ b/crates/buzz-db/src/runtime/tests/reader_pool_postgres_tests.rs @@ -3,7 +3,6 @@ //! reader). use super::*; -use std::time::Instant; /// A `Db` whose writer and lazy reader both point at one scratch database, /// with the fence open so routed reads attempt the reader. @@ -32,22 +31,6 @@ async fn same_db_reader(prefix: &str, read_max: u32) -> (PgPool, Db, String, Str (admin, db, wname, scratch_url) } -/// Poll until the reader pool holds exactly `target` connections, all idle. -/// Returned and floor-maintenance connections land asynchronously, from -/// tasks sqlx spawns on drop. -async fn await_idle_floor(read_pool: &PgPool, target: u32) { - let deadline = Instant::now() + Duration::from_secs(10); - while read_pool.size() != target || read_pool.num_idle() != target as usize { - assert!( - Instant::now() < deadline, - "reader pool never settled at {target} idle; size={} idle={}", - read_pool.size(), - read_pool.num_idle() - ); - tokio::time::sleep(Duration::from_millis(10)).await; - } -} - /// Counters keyed by `(name, decision, reason)` labels, drained: each /// snapshot resets the recorder's counters to zero. fn drain_counters( @@ -351,7 +334,20 @@ async fn stalled_reader_sessions_release_their_slots_for_new_connections() { await_idle_floor(&read_pool, 1).await; { let first = read_pool.acquire().await.expect("first warm connection"); - let second = read_pool.acquire().await.expect("second warm connection"); + // The floor maintenance stops at one, so the second slot is a cold + // dial inside the 150ms acquire budget, which a loaded runner can + // outlast. Warm-up is not under test; the stall below is. + let second = tokio::time::timeout(Duration::from_secs(5), async { + loop { + match read_pool.acquire().await { + Err(sqlx::Error::PoolTimedOut) => continue, + connection => break connection, + } + } + }) + .await + .expect("second reader connects within 5s") + .expect("second warm connection"); drop((first, second)); } await_idle_floor(&read_pool, 2).await; diff --git a/crates/buzz-db/src/store/entity_deletion/leases.rs b/crates/buzz-db/src/store/entity_deletion/leases.rs index bffb6e45791..42ceeb55990 100644 --- a/crates/buzz-db/src/store/entity_deletion/leases.rs +++ b/crates/buzz-db/src/store/entity_deletion/leases.rs @@ -101,23 +101,41 @@ impl EntityDeletionStore { })) } - /// Renew an unexpired lease. An expired or superseded worker cannot revive it. + /// Renew this lease. A superseded lease is never renewed; an expired lease + /// can be renewed only while nobody has reclaimed it. + /// + /// One autocommit statement: the worker drops this future whenever its + /// stage handler finishes first, and sqlx 0.9 queues no `ROLLBACK` for a + /// `begin()` dropped while its `BEGIN` is in flight, so the pooled + /// connection would carry an open transaction into the next write. pub async fn heartbeat(&self, lease: &EntityLease, duration: Duration) -> Result<()> { let millis = duration_millis(duration)?; - let mut tx = self.pool.begin().await?; - lock_owned(&mut tx, lease).await?; - sqlx::query( + // `lock_owned`'s predicate. After a row-lock wait, Postgres re-checks it + // only if the holder committed an update of this row, so a reclaim's new + // generation always wins. A holder that only locked the row (e.g. + // `save_stage_success`) or rolled back leaves the pre-wait expiry check + // in force, so a heartbeat that waits past expiry still renews. No one + // can reclaim in that window: `claim` skips locked rows. + let renewed = sqlx::query( "UPDATE entity_deletion_requests - SET lease_expires_at = clock_timestamp() + $2 * interval '1 millisecond', + SET lease_expires_at = clock_timestamp() + $4 * interval '1 millisecond', heartbeat_at = clock_timestamp() - WHERE id = $1", + WHERE id = $1 AND lease_owner = $2 AND lease_generation = $3 + AND status IN ('approved', 'running', 'retry_wait', 'completed') + AND lease_expires_at > clock_timestamp()", ) .bind(lease.request_id) + .bind(&lease.owner) + .bind(lease.generation) .bind(millis) - .execute(&mut *tx) - .await?; - tx.commit().await?; - Ok(()) + .execute(&self.pool) + .await? + .rows_affected(); + if renewed == 1 { + Ok(()) + } else { + Err(stale_lease(lease)) + } } /// Release this live claim after a pause or shutdown, without advancing its stage. diff --git a/crates/buzz-db/src/store/entity_deletion_postgres_tests.rs b/crates/buzz-db/src/store/entity_deletion_postgres_tests.rs index c93c460ce69..787cd24e5b5 100644 --- a/crates/buzz-db/src/store/entity_deletion_postgres_tests.rs +++ b/crates/buzz-db/src/store/entity_deletion_postgres_tests.rs @@ -1,4 +1,5 @@ use super::*; +use std::time::Duration; async fn store() -> EntityDeletionStore { EntityDeletionStore { @@ -446,6 +447,85 @@ async fn approval_refuses_to_repair_missing_evidence() { ); } +#[tokio::test] +#[ignore = "requires Postgres"] +async fn dropped_heartbeat_leaves_no_open_transaction_for_the_release() { + let store = store().await; + let request = users(&store) + .submit("demo:dropped-heartbeat", "operator-a", None) + .await + .unwrap(); + store + .approve( + request.id, + "operator-b", + request.inventory_digest.as_deref().unwrap(), + ) + .await + .unwrap(); + let lease = store + .claim(request.id, "worker", Duration::from_secs(60)) + .await + .unwrap() + .expect("claim approved request"); + + let direct = store.pool.connect_options(); + let proxy = + crate::test_support::PausingProxy::spawn(direct.get_host().to_owned(), direct.get_port()) + .await; + // One connection, so the release reuses the heartbeat's. No acquire ping, + // so the heartbeat's first statement is the one in flight when it drops. + let proxied = EntityDeletionStore { + pool: sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .test_before_acquire(false) + .connect_with((*direct).clone().host("127.0.0.1").port(proxy.port)) + .await + .expect("connect through proxy"), + }; + sqlx::query("SELECT 1") + .execute(&proxied.pool) + .await + .expect("warm proxied connection"); + // sqlx returns the connection from a spawned task after a ping; pausing + // before it lands would stall the heartbeat in `acquire` instead. + tokio::time::timeout(Duration::from_secs(5), async { + while proxied.pool.num_idle() != 1 { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("warm connection returned to the pool"); + + // The worker's stage `select!` drops the heartbeat whenever the handler + // wins. Holding the bytes drops it with its first round trip in flight. + proxy.set_paused(true); + assert!(tokio::time::timeout( + Duration::from_millis(100), + proxied.heartbeat(&lease, Duration::from_secs(60)), + ) + .await + .is_err()); + proxy.set_paused(false); + assert!(proxied.release(&lease).await.unwrap()); + + let lease_owner: Option = + sqlx::query_scalar("SELECT lease_owner FROM entity_deletion_requests WHERE id = $1") + .bind(request.id) + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!(lease_owner, None, "the reported release must be committed"); + let open: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_stat_activity + WHERE datname = current_database() AND state = 'idle in transaction'", + ) + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!(open, 0, "no pooled session may sit inside a transaction"); +} + fn users(store: &EntityDeletionStore) -> crate::user_deletion::UserDeletionStore { Db::from_pool(store.pool.clone()).user_deletion_store() } diff --git a/crates/buzz-deletion/src/user/worker.rs b/crates/buzz-deletion/src/user/worker.rs index 3cf6afe6500..f8df167da97 100644 --- a/crates/buzz-deletion/src/user/worker.rs +++ b/crates/buzz-deletion/src/user/worker.rs @@ -121,6 +121,8 @@ async fn run_stage_with_heartbeat( // Poll renewal alongside the handler. Awaiting renewal inside a selected // timer branch would stop polling a handler that may hold the parent lock. // Both futures live here, so cancellation leaves no detached heartbeat task. + // A handler can finish mid-heartbeat; `heartbeat` is one autocommit + // statement so that drop cannot strand an open transaction on the pool. let renewal = async { loop { interval.tick().await; diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 291705493ae..8384345671f 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -3798,8 +3798,11 @@ pub(crate) mod terminal_close_delivery_tests { #[ignore = "requires Postgres"] async fn upgrade_would_deny_line_is_written_at_auth_while_open() { let log = crate::nip_fi_shadow::tests::WouldDenyLog::capture(); - let state = crate::state::tests::test_state_with_database_url( + // sqlx's default acquire timeout: a first connect to a busy CI + // database can outlast the unreachable-target helper's 100 ms. + let state = crate::state::tests::test_state_with_database_url_and_pool_options( &crate::test_support::database_url(), + sqlx::postgres::PgPoolOptions::new(), ) .await; let community = Uuid::new_v4(); @@ -4158,7 +4161,7 @@ pub(crate) mod terminal_close_delivery_tests { .unwrap(); let ((), lines) = crate::nip_fi_shadow::tests::would_deny_lines(|| { rt.block_on(async { - let state = crate::state::tests::test_state_with_database_url( + let state = crate::state::tests::test_state_with_unreachable_database_url( "postgres://127.0.0.1:1/none", ) .await; diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index 57e77af36a2..9ec99a56336 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/crates/buzz-relay/src/handlers/auth.rs @@ -3939,7 +3939,7 @@ mod tests { let recorder = metrics_util::debugging::DebuggingRecorder::new(); let snapshotter = recorder.snapshotter(); let _recorder_guard = metrics::set_default_local_recorder(&recorder); - let state = crate::state::tests::test_state_with_database_url( + let state = crate::state::tests::test_state_with_unreachable_database_url( "postgres://buzz:buzz_dev@127.0.0.1:1/buzz", ) .await; diff --git a/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs b/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs index f0586181d9b..0e3859c05b9 100644 --- a/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs @@ -13,9 +13,9 @@ use crate::state::AppState; use buzz_auth::Scope; /// Test state on the shared test database. The pool keeps sqlx's default -/// acquire timeout: `test_state_with_database_url` waits only 100 ms so -/// closed-port tests fail fast, and a first connect to a busy CI database can -/// take longer than that (`PoolTimedOut`). +/// acquire timeout: `test_state_with_unreachable_database_url` waits only +/// 100 ms so closed-port tests fail fast, and a first connect to a busy CI +/// database can take longer than that (`PoolTimedOut`). pub(crate) async fn state() -> Arc { crate::state::tests::test_state_with_database_url_and_pool_options( &crate::test_support::database_url(), diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 9c6d6f206e7..3da4880f39e 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -3818,7 +3818,7 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn check_channel_write_denies_when_channel_lookup_fails() { - let state = crate::state::tests::test_state_with_database_url( + let state = crate::state::tests::test_state_with_unreachable_database_url( "postgres://buzz:buzz_dev@127.0.0.1:1/buzz", // sadscan:disable np.postgres.1 ) .await; diff --git a/crates/buzz-relay/src/nip11.rs b/crates/buzz-relay/src/nip11.rs index 761c9e8d9e6..390e73d0a3d 100644 --- a/crates/buzz-relay/src/nip11.rs +++ b/crates/buzz-relay/src/nip11.rs @@ -524,8 +524,10 @@ mod tests { #[tokio::test] async fn shadow_serves_the_same_nip11_document_as_enforce() { use buzz_auth::NipFiMode; - let base = - crate::state::tests::test_state_with_database_url("postgres://127.0.0.1:1/none").await; + let base = crate::state::tests::test_state_with_unreachable_database_url( + "postgres://127.0.0.1:1/none", + ) + .await; let mut docs = Vec::new(); for mode in [NipFiMode::Off, NipFiMode::Enforce, NipFiMode::Shadow] { let mut state = (*base).clone(); diff --git a/crates/buzz-relay/src/nip_fi_resolver/tests/router_tests.rs b/crates/buzz-relay/src/nip_fi_resolver/tests/router_tests.rs index 55284cc6381..eaa8974b016 100644 --- a/crates/buzz-relay/src/nip_fi_resolver/tests/router_tests.rs +++ b/crates/buzz-relay/src/nip_fi_resolver/tests/router_tests.rs @@ -146,8 +146,10 @@ struct Pod { } async fn pod(store: &Arc, clock: &Arc, mode: NipFiMode) -> Pod { - let base = - crate::state::tests::test_state_with_database_url("postgres://127.0.0.1:1/none").await; + let base = crate::state::tests::test_state_with_unreachable_database_url( + "postgres://127.0.0.1:1/none", + ) + .await; pod_on(base, store, clock, mode).await } diff --git a/crates/buzz-relay/src/nip_fi_shadow_session.rs b/crates/buzz-relay/src/nip_fi_shadow_session.rs index 0dfa5472ca6..c8599db8265 100644 --- a/crates/buzz-relay/src/nip_fi_shadow_session.rs +++ b/crates/buzz-relay/src/nip_fi_shadow_session.rs @@ -393,10 +393,11 @@ pub(crate) mod tests { /// timeout, so a path that reaches Postgres fails fast whether or not /// one runs locally. pub(crate) async fn shadow_state(denied: Option) -> AppState { - let mut state = - (*crate::state::tests::test_state_with_database_url("postgres://127.0.0.1:1/none") - .await) - .clone(); + let mut state = (*crate::state::tests::test_state_with_unreachable_database_url( + "postgres://127.0.0.1:1/none", + ) + .await) + .clone(); let capacity = vec![IssuerCapacity { issuer: "test-issuer".to_owned(), capacity: 8, diff --git a/crates/buzz-relay/src/readiness.rs b/crates/buzz-relay/src/readiness.rs index f39689f20ac..f3afffde795 100644 --- a/crates/buzz-relay/src/readiness.rs +++ b/crates/buzz-relay/src/readiness.rs @@ -1258,7 +1258,7 @@ mod tests { metrics::with_local_recorder(&recorder, || { runtime.block_on(async { - let mut state = crate::state::tests::test_state_with_database_url( + let mut state = crate::state::tests::test_state_with_unreachable_database_url( "postgres://127.0.0.1:1/buzz", ) .await; diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 4fe9e7b1e77..6d2264efaf0 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -3013,12 +3013,15 @@ pub(crate) mod tests { build_test_state(config, pool).await } - /// The same test state with an explicit database target. This lets handler - /// tests deterministically exercise fail-closed database seams without - /// depending on whether a developer has the normal test database running. - /// Its pool waits only 100 ms for a connection, so closed-port targets - /// fail fast. - pub(crate) async fn test_state_with_database_url(database_url: &str) -> Arc { + /// The same test state with an unreachable database target. This lets + /// handler tests deterministically exercise fail-closed database seams + /// without depending on whether a developer has the normal test database + /// running. Its pool waits only 100 ms for a connection, so closed-port + /// targets fail fast; a real database can miss that budget, so tests that + /// need one use [`test_state_with_database_url_and_pool_options`]. + pub(crate) async fn test_state_with_unreachable_database_url( + database_url: &str, + ) -> Arc { test_state_with_database_url_and_pool_options( database_url, sqlx::postgres::PgPoolOptions::new() @@ -4085,7 +4088,7 @@ pub(crate) mod tests { /// archive, so it retains sockets for lifecycle revalidation instead. #[tokio::test] async fn bare_community_disconnect_fails_closed_when_the_fence_is_unavailable() { - let state = test_state_with_database_url( + let state = test_state_with_unreachable_database_url( "postgres://unused:unused@127.0.0.1:1/unused", // sadscan:disable np.postgres.1 ) .await; diff --git a/crates/buzz-relay/tests/boot_lifecycle.rs b/crates/buzz-relay/tests/boot_lifecycle.rs index 0a63a83be8c..f784dd15e79 100644 --- a/crates/buzz-relay/tests/boot_lifecycle.rs +++ b/crates/buzz-relay/tests/boot_lifecycle.rs @@ -82,6 +82,11 @@ impl RelayProcess { } thread::sleep(Duration::from_millis(10)); }; + self.collect(status) + } + + /// The captured output of a child that has already exited as `status`. + fn collect(&mut self, status: ExitStatus) -> Output { self.child.take(); let output = Output { status, @@ -166,7 +171,7 @@ fn wait_for_relay_metrics(process: &mut RelayProcess, port: u16) -> String { /// Polls the relay's own `/metrics` until `needle` appears, bounded by /// [`METRICS_SCRAPE_DEADLINE`]. A relay that exits first is a failure, not a -/// timeout, so the panic names the real cause. +/// timeout, so the panic names the real cause and carries the relay's output. fn wait_for_scraped_metric(process: &mut RelayProcess, port: u16, needle: &str) -> String { wait_for_scraped_metrics(process, port, &[needle]) } @@ -180,10 +185,14 @@ fn wait_for_scraped_metrics(process: &mut RelayProcess, port: u16, needles: &[&s let mut last_error: Option; let deadline = Instant::now() + METRICS_SCRAPE_DEADLINE; loop { - assert!( - process.try_wait().is_none(), - "relay exited before exporting {requested}" - ); + if let Some(status) = process.try_wait() { + let output = process.collect(status); + panic!( + "relay exited ({status}) before exporting {requested}\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + } match scrape_metrics(port) { Ok(response) => { if needles