Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 15 additions & 6 deletions crates/sqlbuild-orchestration/tests/platform/helpers.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<Result<(), ControlError>>,
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"
Expand Down
13 changes: 5 additions & 8 deletions crates/sqlbuild-orchestration/tests/runs/test_claims.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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::<Vec<_>>().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();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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"]);
Expand Down
9 changes: 3 additions & 6 deletions crates/sqlbuild-orchestration/tests/runs/test_monitor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;

Expand Down Expand Up @@ -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),
Expand All @@ -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);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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;

Expand Down Expand Up @@ -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] {
Expand Down Expand Up @@ -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::<Vec<_>>();
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
Expand Down
48 changes: 47 additions & 1 deletion crates/sqlbuild-orchestration/tests/store/leases/helpers.rs
Original file line number Diff line number Diff line change
@@ -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) {
Expand All @@ -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
}
}
Loading