From 2f6dc843a0204ebf18648919a9850813a076af8a Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 20:41:58 +1300 Subject: [PATCH 1/2] test(rescue): run the backlog tests one at a time and report rescued heartbeat ages The 50,000-claim test requires that no claim of the live doomed runtime is rescued before it is killed. That holds only while the runtime heartbeats every claim within the staleness window, 18 s under CI scaling. Its heartbeat pass over 50,000 claims is sequential, and with the row-local backlog test loading the same database in parallel the pass took 17 to 19 s, so claims went stale and rescue took them. Instrumented runs showed every rescued claim had gone 18.4 s or more without a heartbeat, most of them never heartbeated since being claimed 21 to 25 s earlier; none was within the staleness window. The two backlog tests now share a lock so they never run together. A trigger records, as rescue closes each compact claim, how long it had gone without a heartbeat. The test first requires that no rescued claim was within the staleness window, and otherwise reports how stale the rescued claims were, so a future failure says whether rescue misjudged a claim or the live runtime's heartbeats fell behind. --- awa/tests/bulk_rescue_test.rs | 72 ++++++++++++++++++++++++++++++++++- 1 file changed, 71 insertions(+), 1 deletion(-) diff --git a/awa/tests/bulk_rescue_test.rs b/awa/tests/bulk_rescue_test.rs index 2fc103c02..880017e9c 100644 --- a/awa/tests/bulk_rescue_test.rs +++ b/awa/tests/bulk_rescue_test.rs @@ -27,6 +27,10 @@ const EXHAUSTED_EVERY: i64 = 10; const ABANDONED_QUEUE: &str = "bulk_rescue_abandoned"; const LIVE_QUEUE: &str = "bulk_rescue_live"; +/// The backlog tests each load the database heavily enough to slow the other's +/// heartbeat passes past their staleness window, so they run one at a time. +static BACKLOG_TEST: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + /// What each run of a job saw, by job id. type RunsByJob = HashMap>; @@ -225,6 +229,62 @@ async fn rescued_rows(pool: &PgPool, queue: &str) -> i64 { .expect("count rescued rows") } +/// Record, as each compact claim is closed as rescued, how long it had gone +/// without a heartbeat. The trigger runs inside the closing statement, after +/// rescue holds the attempt's lock, so no later heartbeat can land. +async fn record_compact_rescue_heartbeats(pool: &PgPool) { + sqlx::raw_sql( + r#" + CREATE TABLE public.rescued_claim_heartbeats ( + job_id bigint NOT NULL, + queue text NOT NULL, + unheartbeated_for interval NOT NULL + ); + CREATE FUNCTION public.record_rescued_claim_heartbeat() RETURNS trigger + LANGUAGE plpgsql AS $$ + BEGIN + INSERT INTO public.rescued_claim_heartbeats + SELECT items.job_id, batches.queue, + NEW.closed_at - COALESCE(attempt.heartbeat_at, batches.claimed_at) + FROM awa.lease_claim_batches AS batches + CROSS JOIN LATERAL unnest(batches.receipt_ids, batches.job_ids, batches.run_leases) + AS items(receipt_id, job_id, run_lease) + LEFT JOIN awa.attempt_state AS attempt + ON attempt.job_id = items.job_id AND attempt.run_lease = items.run_lease + WHERE batches.claim_slot = NEW.claim_slot + AND items.receipt_id = ANY(NEW.receipt_ids); + RETURN NEW; + END + $$; + CREATE TRIGGER record_rescued_claim_heartbeat + AFTER INSERT ON awa.lease_claim_closure_batches + FOR EACH ROW WHEN (NEW.outcome = 'rescued') + EXECUTE FUNCTION public.record_rescued_claim_heartbeat(); + "#, + ) + .execute(pool) + .await + .expect("record rescued claim heartbeats"); +} + +/// Abandoned-queue claims rescued so far: how many had a heartbeat (or claim) +/// younger than `staleness` when closed, how many there were, and the shortest +/// time any of them had gone without a heartbeat. +async fn rescued_heartbeat_ages(pool: &PgPool, staleness: Duration) -> (i64, i64, String) { + sqlx::query_as( + "SELECT count(*) FILTER (WHERE unheartbeated_for < $2 * interval '1 millisecond'), + count(*), + COALESCE(min(unheartbeated_for)::text, 'n/a') + FROM public.rescued_claim_heartbeats + WHERE queue = $1", + ) + .bind(ABANDONED_QUEUE) + .bind(staleness.as_millis() as f64) + .fetch_one(pool) + .await + .expect("rescued claim heartbeat ages") +} + async fn wait_until(timeout: Duration, what: &str, mut done: F) -> Duration where F: FnMut() -> Fut, @@ -420,6 +480,7 @@ impl DoomedClient { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn rescue_drains_an_abandoned_backlog_without_rescuing_live_claims() { + let _backlog_test = BACKLOG_TEST.lock().await; let _ = tracing_subscriber::fmt() .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) .with_test_writer() @@ -557,6 +618,7 @@ async fn rescue_drains_an_abandoned_backlog_without_rescuing_live_claims() { // The doomed runtime claims the backlog in chunks, each after the claim // ring has rotated, so the abandoned claims span several slots. let generation_before = claim_ring_generation(&pool).await; + record_compact_rescue_heartbeats(&pool).await; let doomed = DoomedClient::start(db.url(), staleness, heartbeat_interval); let claim_started = Instant::now(); for chunk_start in (0..ABANDONED).step_by(ABANDONED_CHUNK as usize) { @@ -595,10 +657,17 @@ async fn rescue_drains_an_abandoned_backlog_without_rescuing_live_claims() { } } let claimed_in = claim_started.elapsed(); + let (fresh, rescued, freshest) = rescued_heartbeat_ages(&pool, staleness).await; + assert_eq!( + fresh, 0, + "rescue closed {fresh} claims last heartbeated within the {staleness:?} staleness window" + ); assert_eq!( rescued_rows(&pool, ABANDONED_QUEUE).await, 0, - "rescue took a claim while its runtime was alive and heartbeating" + "rescue took a claim while its runtime was alive: {rescued} claims had gone {freshest} \ + or longer without a heartbeat, beyond the {staleness:?} staleness, so the live \ + runtime's heartbeat passes fell behind" ); let generation_at_kill = claim_ring_generation(&pool).await; live_hold_ms.store(live_job_duration.as_millis() as u64, Ordering::SeqCst); @@ -849,6 +918,7 @@ async fn open_first_attempt_claims(pool: &PgPool) -> i64 { /// the drain rate after that long instead of requiring a full drain. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn rescue_drains_abandoned_row_local_claims_and_unpins_both_rings() { + let _backlog_test = BACKLOG_TEST.lock().await; let _ = tracing_subscriber::fmt() .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) .with_test_writer() From 819c9c3c869e6feda176f95a88686bd139a0285e Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 20:54:54 +1300 Subject: [PATCH 2/2] test(rescue): serialise the bulk rescue backlog tests under nextest too --- .config/nextest.toml | 8 ++++++++ awa/tests/bulk_rescue_test.rs | 4 +++- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/.config/nextest.toml b/.config/nextest.toml index 8b0ad8e3e..04b07483c 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -15,7 +15,15 @@ slow-timeout = { period = "60s" } [test-groups] migrations = { max-threads = 1 } +bulk-rescue-backlog = { max-threads = 1 } [[profile.default.overrides]] filter = 'binary(migration_test)' test-group = 'migrations' + +# The two bulk rescue backlog tests each load the same database heavily; run +# concurrently, the live runtime in one cannot heartbeat its claims inside the +# staleness window. +[[profile.default.overrides]] +filter = 'binary(bulk_rescue_test) and test(/^rescue_drains_/)' +test-group = 'bulk-rescue-backlog' diff --git a/awa/tests/bulk_rescue_test.rs b/awa/tests/bulk_rescue_test.rs index 880017e9c..8ce927567 100644 --- a/awa/tests/bulk_rescue_test.rs +++ b/awa/tests/bulk_rescue_test.rs @@ -28,7 +28,9 @@ const ABANDONED_QUEUE: &str = "bulk_rescue_abandoned"; const LIVE_QUEUE: &str = "bulk_rescue_live"; /// The backlog tests each load the database heavily enough to slow the other's -/// heartbeat passes past their staleness window, so they run one at a time. +/// heartbeat passes past their staleness window, so they run one at a time: +/// this lock serialises them under `cargo test`, and a nextest test group in +/// `.config/nextest.toml` does so across processes. static BACKLOG_TEST: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); /// What each run of a job saw, by job id.