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