From 2ba5dc636857e112beaab84fd5fdf19750f58912 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Fri, 6 Nov 2026 15:34:40 +0000 Subject: [PATCH 1/3] test: stop scheduler tests depending on runner speed Four orchestration tests held a real 30-second scheduler lease across setup that is not about lease expiry, so a slow runner fenced them before the code under test ran. They now hold the lease for the whole test through a helper beside expire_lease, replacing per-chunk renewals. The platform webhook test waited a fixed ten seconds for 107 deliveries. It now fails only when the running dispatcher makes no progress for a minute; the delivered count and dispatcher assertions are unchanged. Validated locally on Linux with SQLite and PostgreSQL: the five affected test modules pass, and with injected stalls (31 s in setup, 120 ms per webhook delivery) the previous tests fail with the CI errors while these pass. --- .../tests/platform/helpers.rs | 19 +++++--- .../tests/runs/test_claims.rs | 13 +++--- .../tests/runs/test_group_admission.rs | 4 +- .../tests/runs/test_monitor.rs | 9 ++-- .../tests/sensors/test_execution_bounds.rs | 43 ++++++++---------- .../tests/store/leases/helpers.rs | 45 ++++++++++++++++++- 6 files changed, 86 insertions(+), 47 deletions(-) diff --git a/crates/sqlbuild-orchestration/tests/platform/helpers.rs b/crates/sqlbuild-orchestration/tests/platform/helpers.rs index eae020ac84..b88a36774d 100644 --- a/crates/sqlbuild-orchestration/tests/platform/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/platform/helpers.rs @@ -141,19 +141,24 @@ pub(crate) async fn receive_webhook( ) } +/// Wait for `count` deliveries; only a running dispatcher that delivers nothing for a minute fails. 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), - ) - .await - .expect("dispatcher drains more than one bounded page") - .expect("webhook receiver remains available"); + 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"); + } + 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..f821d9cf34 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::hold_lease; 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 = hold_lease(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..3268b71874 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::hold_lease; 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 = hold_lease(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..87eff1e1c2 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, hold_lease}; 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 = hold_lease(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..0a9dbd1131 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::hold_lease; 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 = hold_lease(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..a3a81eff26 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,43 @@ pub(crate) async fn expire_lease(store: &ControlStore) { .await .expect("expire PostgreSQL scheduler lease"); } + +/// Keep `lease` live for a day, so slow setup cannot fence a test that is not about lease expiry. +pub(crate) async fn hold_lease(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 + } +} From 2bd8f7a9b8ea1aa07bc1f575e5e98b522f74966d Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Fri, 6 Nov 2026 15:42:59 +0000 Subject: [PATCH 2/3] test: name the test lease extension apart from the background renewer Validated locally: the five affected test modules on SQLite and PostgreSQL, clippy and fensu. --- crates/sqlbuild-orchestration/tests/runs/test_claims.rs | 4 ++-- .../tests/runs/test_group_admission.rs | 4 ++-- crates/sqlbuild-orchestration/tests/runs/test_monitor.rs | 4 ++-- .../tests/sensors/test_execution_bounds.rs | 4 ++-- .../sqlbuild-orchestration/tests/store/leases/helpers.rs | 7 +++++-- 5 files changed, 13 insertions(+), 10 deletions(-) diff --git a/crates/sqlbuild-orchestration/tests/runs/test_claims.rs b/crates/sqlbuild-orchestration/tests/runs/test_claims.rs index f821d9cf34..b9aefe9a36 100644 --- a/crates/sqlbuild-orchestration/tests/runs/test_claims.rs +++ b/crates/sqlbuild-orchestration/tests/runs/test_claims.rs @@ -18,7 +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::hold_lease; +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; @@ -404,7 +404,7 @@ fn given_more_than_one_page_of_blocked_runs_when_claiming_then_lower_priority_in .await .expect("elect") .expect("leader"); - let lease = hold_lease(store, elected).await; + 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) diff --git a/crates/sqlbuild-orchestration/tests/runs/test_group_admission.rs b/crates/sqlbuild-orchestration/tests/runs/test_group_admission.rs index 3268b71874..8e0e670966 100644 --- a/crates/sqlbuild-orchestration/tests/runs/test_group_admission.rs +++ b/crates/sqlbuild-orchestration/tests/runs/test_group_admission.rs @@ -22,7 +22,7 @@ use crate::runs::tests::test_types::{ AdmissionParent, GroupAdmissionTestCase, GroupChildLimitTestCase, GroupQueueTestCase, MixedAdmissionTestCase, RunLifecycleTestCase, }; -use crate::store::tests::lease_helpers::hold_lease; +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; @@ -807,7 +807,7 @@ fn given_a_large_healthy_project_with_blocked_groups_when_claiming_then_unrelate .await .expect("elect") .expect("leader"); - let lease = hold_lease(store, elected).await; + 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 87eff1e1c2..1f874b411e 100644 --- a/crates/sqlbuild-orchestration/tests/runs/test_monitor.rs +++ b/crates/sqlbuild-orchestration/tests/runs/test_monitor.rs @@ -24,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, hold_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; @@ -290,7 +290,7 @@ fn given_a_silent_worker_backlog_when_monitoring_then_bounded_transactions_conti .await .expect("elect") .expect("leader"); - let lease = hold_lease(store, elected).await; + 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), diff --git a/crates/sqlbuild-orchestration/tests/sensors/test_execution_bounds.rs b/crates/sqlbuild-orchestration/tests/sensors/test_execution_bounds.rs index 0a9dbd1131..472d2b5e20 100644 --- a/crates/sqlbuild-orchestration/tests/sensors/test_execution_bounds.rs +++ b/crates/sqlbuild-orchestration/tests/sensors/test_execution_bounds.rs @@ -24,7 +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::hold_lease; +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; @@ -83,7 +83,7 @@ fn given_a_wide_group_when_only_the_last_node_has_not_started_then_all_chunks_mu .await .expect("elect") .expect("leader"); - let lease = hold_lease(store, elected).await; + 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] { diff --git a/crates/sqlbuild-orchestration/tests/store/leases/helpers.rs b/crates/sqlbuild-orchestration/tests/store/leases/helpers.rs index a3a81eff26..a5b9945b67 100644 --- a/crates/sqlbuild-orchestration/tests/store/leases/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/store/leases/helpers.rs @@ -26,8 +26,11 @@ pub(crate) async fn expire_lease(store: &ControlStore) { .expect("expire PostgreSQL scheduler lease"); } -/// Keep `lease` live for a day, so slow setup cannot fence a test that is not about lease expiry. -pub(crate) async fn hold_lease(store: &ControlStore, lease: SchedulerLease) -> SchedulerLease { +/// 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 { From c878dd1fd7d7f0cfd9c4579298a05615b8d146c4 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Fri, 6 Nov 2026 15:45:12 +0000 Subject: [PATCH 3/3] test: cap the webhook wait at ten minutes in total A dispatcher that keeps delivering but trickles now fails with its own message instead of reaching the job timeout. The one-minute progress bound is unchanged. Validated locally: the webhook test on SQLite and PostgreSQL, clippy and fensu. --- .../tests/platform/helpers.rs | 26 +++++++++++-------- 1 file changed, 15 insertions(+), 11 deletions(-) diff --git a/crates/sqlbuild-orchestration/tests/platform/helpers.rs b/crates/sqlbuild-orchestration/tests/platform/helpers.rs index b88a36774d..b5a4e301a4 100644 --- a/crates/sqlbuild-orchestration/tests/platform/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/platform/helpers.rs @@ -141,23 +141,27 @@ pub(crate) async fn receive_webhook( ) } -/// Wait for `count` deliveries; only a running dispatcher that delivers nothing for a minute fails. +/// 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(); - 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"); - } + 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 delivers every event within ten minutes"); assert!(*delivered.borrow() >= count, "every event delivered"); assert!( !dispatcher.is_finished(),