Merge branch 'main' into fix/listing-denial-text

Author: kvlonge

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<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"

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::<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();

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"]);

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);

crates/sqlbuild-orchestration/tests/scheduler/ticks/helpers.rs

@@ -619,9 +619,12 @@ pub(crate) async fn fault_run(
             .0;
     }
     let settled = Utc::now();
-    scheduler_cycle(&second, survivor)
-        .await
-        .expect("final recovery");
+    for _ in 0..FAULT_TRIGGERS.div_ceil(100) {
+        survivor = scheduler_cycle(&second, survivor)
+            .await
+            .expect("final recovery")
+            .0;
+    }
     let first = started
         .duration_trunc(TimeDelta::minutes(1))
         .expect("minute")

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::<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

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
+    }
+}