diff --git a/crates/sqlbuild-orchestration/tests/platform/helpers.rs b/crates/sqlbuild-orchestration/tests/platform/helpers.rs index eae020ac84..b5a4e301a4 100644 --- a/crates/sqlbuild-orchestration/tests/platform/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/platform/helpers.rs @@ -141,19 +141,28 @@ pub(crate) async fn receive_webhook( ) } +/// Wait for `count` deliveries: fail after a minute without progress or ten minutes in total. pub(crate) async fn wait_for_webhook( receiver: &WebhookReceiver, dispatcher: &JoinHandle>, count: usize, ) { let mut delivered = receiver.delivered.subscribe(); - tokio::time::timeout( - Duration::from_secs(10), - delivered.wait_for(|received| *received >= count), - ) + tokio::time::timeout(Duration::from_secs(600), async { + for _ in 0..count { + let seen = *delivered.borrow_and_update(); + tokio::time::timeout( + Duration::from_secs(60), + delivered.wait_for(|received| *received > seen || *received >= count), + ) + .await + .expect("dispatcher keeps delivering across bounded pages") + .expect("webhook receiver remains available"); + } + }) .await - .expect("dispatcher drains more than one bounded page") - .expect("webhook receiver remains available"); + .expect("dispatcher delivers every event within ten minutes"); + assert!(*delivered.borrow() >= count, "every event delivered"); assert!( !dispatcher.is_finished(), "delivery loop stopped before draining" diff --git a/crates/sqlbuild-orchestration/tests/runs/test_claims.rs b/crates/sqlbuild-orchestration/tests/runs/test_claims.rs index d8c4151143..b9aefe9a36 100644 --- a/crates/sqlbuild-orchestration/tests/runs/test_claims.rs +++ b/crates/sqlbuild-orchestration/tests/runs/test_claims.rs @@ -18,6 +18,7 @@ use crate::runs::tests::terminal_helpers::finish; use crate::runs::tests::test_types::{ BacklogClaimTestCase, GroupLimitTestCase, PriorityClaimsTestCase, ScopedClaimTestCase, }; +use crate::store::tests::lease_helpers::extend_lease_through_test; use crate::store::tests::replica_helpers::second_store; use crate::store::tests::store_helpers::{close_store, close_test_store, open_test_store, run}; use crate::store::tests::store_test_types::Backend; @@ -399,23 +400,19 @@ fn given_more_than_one_page_of_blocked_runs_when_claiming_then_lower_priority_in .await .expect("snapshot"); let scope = (location, published.snapshot_id); - let mut lease = acquire_lease(store, Uuid::now_v7()) + let elected = acquire_lease(store, Uuid::now_v7()) .await .expect("elect") .expect("leader"); + let lease = extend_lease_through_test(store, elected).await; let limits = BTreeMap::new(); queue(store, &request(scope, 1, 0, &["orders"])).await; claim_next_run(store, &lease, &limits, None) .await .expect("claim blocker") .expect("available"); - for batch in (2..=104).collect::>().chunks(20) { - for id in batch { - queue(store, &request(scope, *id, 0, &["orders"])).await; - } - lease = crate::scheduler::main::renew_lease::renew_lease(store, &lease) - .await - .expect("keep leader live while preparing backlog"); + for id in 2..=104 { + queue(store, &request(scope, id, 0, &["orders"])).await; } queue(store, &request(scope, 105, -1, &["customers"])).await; let started = std::time::Instant::now(); diff --git a/crates/sqlbuild-orchestration/tests/runs/test_group_admission.rs b/crates/sqlbuild-orchestration/tests/runs/test_group_admission.rs index b900611fa2..8e0e670966 100644 --- a/crates/sqlbuild-orchestration/tests/runs/test_group_admission.rs +++ b/crates/sqlbuild-orchestration/tests/runs/test_group_admission.rs @@ -22,6 +22,7 @@ use crate::runs::tests::test_types::{ AdmissionParent, GroupAdmissionTestCase, GroupChildLimitTestCase, GroupQueueTestCase, MixedAdmissionTestCase, RunLifecycleTestCase, }; +use crate::store::tests::lease_helpers::extend_lease_through_test; use crate::store::tests::replica_helpers::second_store; use crate::store::tests::store_helpers::{close_store, close_test_store, open_test_store, run}; use crate::store::tests::store_test_types::Backend; @@ -802,10 +803,11 @@ fn given_a_large_healthy_project_with_blocked_groups_when_claiming_then_unrelate let published = publish_snapshot(store, &input) .await .expect("publish large project"); - let lease = acquire_lease(store, Uuid::now_v7()) + let elected = acquire_lease(store, Uuid::now_v7()) .await .expect("elect") .expect("leader"); + let lease = extend_lease_through_test(store, elected).await; let limits = BTreeMap::from([("warehouse".into(), NonZeroU32::new(1).expect("one"))]); let scope = (location, "development"); let mut physical = request((location, published.snapshot_id), 1, 10, &["products"]); diff --git a/crates/sqlbuild-orchestration/tests/runs/test_monitor.rs b/crates/sqlbuild-orchestration/tests/runs/test_monitor.rs index a10b3a2a4c..1f874b411e 100644 --- a/crates/sqlbuild-orchestration/tests/runs/test_monitor.rs +++ b/crates/sqlbuild-orchestration/tests/runs/test_monitor.rs @@ -9,7 +9,6 @@ use crate::runs::main::heartbeat_run::heartbeat_run; use crate::runs::main::list_runs::list_runs; use crate::runs::main::monitor_runs::monitor_runs; use crate::runs::models::{RunAuthority, RunChange, RunChangeRequest, RunOutcome, RunStatus}; -use crate::scheduler::main::renew_lease::renew_lease; use crate::scheduler::models::SchedulerState; use crate::scheduler::tests::contracts::helpers::{acquire_lease, scheduler_cycle}; use crate::snapshots::main::publish_snapshot::publish_snapshot; @@ -25,7 +24,7 @@ use crate::runs::tests::test_types::{ MonitorBacklogTestCase, RunMonitorTestCase, TransactionFault, }; use crate::store::tests::fault_helpers::{install_fault, remove_fault}; -use crate::store::tests::lease_helpers::expire_lease; +use crate::store::tests::lease_helpers::{expire_lease, extend_lease_through_test}; use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; use crate::store::tests::store_test_types::Backend; @@ -287,10 +286,11 @@ fn given_a_silent_worker_backlog_when_monitoring_then_bounded_transactions_conti .await .expect("snapshot"); let holder = Uuid::now_v7(); - let mut lease = acquire_lease(store, holder) + let elected = acquire_lease(store, holder) .await .expect("elect") .expect("leader"); + let lease = extend_lease_through_test(store, elected).await; for index in 0..test_case.run_count { let mut input = request( (location, published.snapshot_id), @@ -301,9 +301,6 @@ fn given_a_silent_worker_backlog_when_monitoring_then_bounded_transactions_conti input.write_set.clear(); start_run(store, &lease, &input).await; age_heartbeat(store, input.run_id, 50).await; - lease = renew_lease(store, &lease) - .await - .expect("renew during setup"); } expire_lease(store).await; let mut state = SchedulerState::new(holder); diff --git a/crates/sqlbuild-orchestration/tests/sensors/test_execution_bounds.rs b/crates/sqlbuild-orchestration/tests/sensors/test_execution_bounds.rs index 87340bf331..472d2b5e20 100644 --- a/crates/sqlbuild-orchestration/tests/sensors/test_execution_bounds.rs +++ b/crates/sqlbuild-orchestration/tests/sensors/test_execution_bounds.rs @@ -15,7 +15,6 @@ use crate::runs::main::{ use crate::runs::models::{ JobRunRequest, NodeOutcome, RunAuthority, RunChange, RunChangeRequest, RunOutcome, }; -use crate::scheduler::main::renew_lease::renew_lease; use crate::scheduler::tests::contracts::helpers::acquire_lease; use crate::sensors::main::{ begin_sensor_evaluation::begin_sensor_evaluation, @@ -25,6 +24,7 @@ use crate::sensors::tests::helpers::snapshot; use crate::sensors::tests::test_types::SensorTestCase; use crate::snapshots::main::publish_snapshot::publish_snapshot; use crate::store::main::migrate_store::migrate_store; +use crate::store::tests::lease_helpers::extend_lease_through_test; use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; use crate::store::tests::store_test_types::Backend; @@ -79,10 +79,11 @@ fn given_a_wide_group_when_only_the_last_node_has_not_started_then_all_chunks_mu publish_snapshot(store, &input) .await .expect("publish wide predicate"); - let lease = acquire_lease(store, Uuid::now_v7()) + let elected = acquire_lease(store, Uuid::now_v7()) .await .expect("elect") .expect("leader"); + let lease = extend_lease_through_test(store, elected).await; let limits = std::collections::BTreeMap::new(); let scope = (location, "development"); for include_last in [true, false] { @@ -138,28 +139,22 @@ fn given_a_wide_group_when_only_the_last_node_has_not_started_then_all_chunks_mu exit_code: None, })) .collect::>(); - for batch in changes.chunks(100) { - // Evidence setup must keep its scheduler alive, just like the real loop. - renew_lease(store, &lease) - .await - .expect("renew during setup"); - for change in batch { - change_run( - store, - RunAuthority::Worker { - run_id: claim.request.run_id, - claim_id: claim.claim_id, - }, - &RunChangeRequest { - event_id: Uuid::now_v7(), - run_id: claim.request.run_id, - claim_id: Some(claim.claim_id), - change: change.clone(), - }, - ) - .await - .expect("real node evidence"); - } + for change in &changes { + change_run( + store, + RunAuthority::Worker { + run_id: claim.request.run_id, + claim_id: claim.claim_id, + }, + &RunChangeRequest { + event_id: Uuid::now_v7(), + run_id: claim.request.run_id, + claim_id: Some(claim.claim_id), + change: change.clone(), + }, + ) + .await + .expect("real node evidence"); } advance_groups(store, &lease, (&limits, scope)) .await diff --git a/crates/sqlbuild-orchestration/tests/store/leases/helpers.rs b/crates/sqlbuild-orchestration/tests/store/leases/helpers.rs index 59f067dc23..a5b9945b67 100644 --- a/crates/sqlbuild-orchestration/tests/store/leases/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/store/leases/helpers.rs @@ -1,5 +1,8 @@ -//! Expire the persisted lease without introducing a fake control-store clock. +//! Expire or hold the persisted lease without introducing a fake control-store clock. +use chrono::{SecondsFormat, TimeDelta, Utc}; + +use crate::scheduler::models::SchedulerLease; use crate::store::models::ControlStore; pub(crate) async fn expire_lease(store: &ControlStore) { @@ -22,3 +25,46 @@ pub(crate) async fn expire_lease(store: &ControlStore) { .await .expect("expire PostgreSQL scheduler lease"); } + +/// Keep `lease` live for a day without a renewer task, so slow setup cannot fence the test. +pub(crate) async fn extend_lease_through_test( + store: &ControlStore, + lease: SchedulerLease, +) -> SchedulerLease { + let expires_at = Utc::now() + TimeDelta::days(1); + let ControlStore::Postgres(pool) = store else { + let ControlStore::Sqlite(path) = store else { + unreachable!() + }; + let held = rusqlite::Connection::open(path) + .expect("open lease hold connection") + .execute( + "UPDATE sqb_scheduler_lease SET expires_at=?1 WHERE name='scheduler' AND holder=?2 AND epoch=?3", + rusqlite::params![ + expires_at.to_rfc3339_opts(SecondsFormat::Micros, true), + lease.holder.to_string(), + lease.epoch + ], + ) + .expect("hold SQLite scheduler lease"); + assert_eq!(held, 1, "the held lease is the current one"); + return SchedulerLease { + expires_at, + ..lease + }; + }; + let held = sqlx_core::query::query( + "UPDATE sqb_scheduler_lease SET expires_at=$1 WHERE name='scheduler' AND holder=$2 AND epoch=$3", + ) + .bind(expires_at) + .bind(lease.holder) + .bind(lease.epoch) + .execute(pool) + .await + .expect("hold PostgreSQL scheduler lease"); + assert_eq!(held.rows_affected(), 1, "the held lease is the current one"); + SchedulerLease { + expires_at, + ..lease + } +}