From a6680d8e42bf0d3f402db0059ab4e69f07317976 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 4 Nov 2026 14:37:13 +0000 Subject: [PATCH] Find aged running jobs and read get_job from one snapshot (0.6) Backport of #600. Claims and leases record the aged effective priority while the ready row keeps the lane priority; load_job, load_active_lease_in_tx and callback_job_in_tx joined the two on priority, so with priority aging on a job that had aged before its claim could not be found by callback or read back as running. Match the attempt by job_id and run_lease instead, and read load_job's relations from one REPEATABLE READ, READ ONLY snapshot. --- CHANGELOG.md | 14 ++++ awa-model/src/queue_storage.rs | 42 ++++++++--- awa/tests/queue_storage_runtime_test.rs | 96 +++++++++++++++++++++++++ 3 files changed, 141 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 49866ca92..4db148edf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,20 @@ Notable changes between releases. Detailed migration notes for storage transitio ## [Unreleased] +- **Callback completion, `get_job` and lifecycle events find running jobs + that priority aging promoted on queue storage** (backport of #600). Claims + and leases record a job's aged effective priority while its ready row keeps + the lane priority, and three reads joined the two on priority. With aging on + (the default, every 60 s), a job that waited past a minute before it was + claimed could not be completed by callback with an expected kind, failed the + mixed-transition callback routing check, and read back from `get_job` in the + wrong state. They now match the attempt (`job_id` and `run_lease`). No + migration. +- **`get_job` no longer misses a job that moves while it reads.** It read + ready, deferred, leased, claimed, completed and DLQ rows in separate + statements, so a job promoted, claimed or completed between two of them + could read as missing. It now reads all of them from one read-only snapshot. + ## [0.6.11] — 2026-11-04 Patch release: promotion of due scheduled and retryable jobs on queue storage no longer stalls once a backlog builds. No migration, no schema patch and no public API change. Upgrading from 0.6.10 is a rolling upgrade of processes; promotion runs on the maintenance leader, so a stalled deployment recovers once the leader runs 0.6.11, and any runtime can become leader, so roll all of them. diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 1834cf21f..47d3afbda 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -10807,6 +10807,13 @@ impl QueueStorage { let closed_evidence = receipt_closed_evidence_sql(schema, &closure_rel, &closure_batch_rel, "claims"); let mut candidates = Vec::new(); + // A job moves between these relations as it is promoted, claimed, + // completed or dead-lettered. Read them all from one snapshot so a + // move between two reads cannot hide the job or surface a stale row. + let mut tx = pool + .begin_with("BEGIN ISOLATION LEVEL REPEATABLE READ, READ ONLY") + .await + .map_err(map_sqlx_error)?; let ready_rows: Vec = sqlx::query_as(&format!( r#" @@ -10831,7 +10838,7 @@ impl QueueStorage { "#, )) .bind(job_id) - .fetch_all(pool) + .fetch_all(tx.as_mut()) .await .map_err(map_sqlx_error)?; for row in ready_rows { @@ -10862,7 +10869,7 @@ impl QueueStorage { "#, )) .bind(job_id) - .fetch_all(pool) + .fetch_all(tx.as_mut()) .await .map_err(map_sqlx_error)?; for row in deferred_rows { @@ -10906,7 +10913,10 @@ impl QueueStorage { ON ready.ready_slot = lease.ready_slot AND ready.ready_generation = lease.ready_generation AND ready.queue = lease.queue - AND ready.priority = lease.priority + -- Claims and leases record the aged effective priority while the + -- ready row keeps its lane priority, so match the attempt instead. + AND ready.job_id = lease.job_id + AND ready.run_lease + 1 = lease.run_lease AND ready.enqueue_shard = lease.enqueue_shard AND ready.lane_seq = lease.lane_seq LEFT JOIN {schema}.attempt_state AS attempt @@ -10917,7 +10927,7 @@ impl QueueStorage { "#, )) .bind(job_id) - .fetch_all(pool) + .fetch_all(tx.as_mut()) .await .map_err(map_sqlx_error)?; for row in lease_rows { @@ -10963,7 +10973,10 @@ impl QueueStorage { ON ready.ready_slot = claims.ready_slot AND ready.ready_generation = claims.ready_generation AND ready.queue = claims.queue - AND ready.priority = claims.priority + -- Claims and leases record the aged effective priority while the + -- ready row keeps its lane priority, so match the attempt instead. + AND ready.job_id = claims.job_id + AND ready.run_lease + 1 = claims.run_lease AND ready.enqueue_shard = claims.enqueue_shard AND ready.lane_seq = claims.lane_seq LEFT JOIN {schema}.attempt_state AS attempt @@ -10983,7 +10996,7 @@ impl QueueStorage { "#, )) .bind(job_id) - .fetch_all(pool) + .fetch_all(tx.as_mut()) .await .map_err(map_sqlx_error)?; for row in lease_claim_rows { @@ -11084,7 +11097,7 @@ impl QueueStorage { "#, )) .bind(job_id) - .fetch_all(pool) + .fetch_all(tx.as_mut()) .await .map_err(map_sqlx_error)?; for row in lease_claim_batch_rows { @@ -11120,7 +11133,7 @@ impl QueueStorage { "#, )) .bind(job_id) - .fetch_all(pool) + .fetch_all(tx.as_mut()) .await .map_err(map_sqlx_error)?; for row in done_rows { @@ -11155,13 +11168,14 @@ impl QueueStorage { "#, )) .bind(job_id) - .fetch_all(pool) + .fetch_all(tx.as_mut()) .await .map_err(map_sqlx_error)?; for row in dlq_rows { candidates.push(row.into_job_row()?); } + tx.commit().await.map_err(map_sqlx_error)?; Ok(candidates.into_iter().max_by_key(|job| { ( job.run_lease, @@ -11487,7 +11501,10 @@ impl QueueStorage { ON ready.ready_slot = lease.ready_slot AND ready.ready_generation = lease.ready_generation AND ready.queue = lease.queue - AND ready.priority = lease.priority + -- Claims and leases record the aged effective priority while the + -- ready row keeps its lane priority, so match the attempt instead. + AND ready.job_id = lease.job_id + AND ready.run_lease + 1 = lease.run_lease AND ready.enqueue_shard = lease.enqueue_shard AND ready.lane_seq = lease.lane_seq LEFT JOIN {schema}.attempt_state AS attempt @@ -11683,7 +11700,10 @@ impl QueueStorage { ON ready.ready_slot = lease.ready_slot AND ready.ready_generation = lease.ready_generation AND ready.queue = lease.queue - AND ready.priority = lease.priority + -- Claims and leases record the aged effective priority while the + -- ready row keeps its lane priority, so match the attempt instead. + AND ready.job_id = lease.job_id + AND ready.run_lease + 1 = lease.run_lease AND ready.enqueue_shard = lease.enqueue_shard AND ready.lane_seq = lease.lane_seq LEFT JOIN {schema}.attempt_state AS attempt diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index 712d079a9..8f2f4e04c 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -13638,3 +13638,99 @@ async fn test_queue_storage_cancel_by_unique_key_reads_live_work_only() { {scoped_finished_claims} with retained claims" ); } + +/// Claims and leases record a job's aged effective priority while its ready +/// row keeps the lane priority. Reads that join the two must still find the +/// running attempt: `load_job` (behind `get_job` and lifecycle events) and +/// the callback lookup behind external completion. +async fn aged_claim_is_visible_to_job_and_callback_reads(receipts: bool) { + let (_db_guard, pool) = setup_pool(6).await; + let queue = "qs_aged_claim_reads"; + let schema = if receipts { + "awa_qs_aged_claim_reads_receipts" + } else { + "awa_qs_aged_claim_reads_leases" + }; + let store = create_store_with_config( + &pool, + QueueStorageConfig { + schema: schema.to_string(), + queue_slot_count: 4, + lease_slot_count: 2, + claim_slot_count: 2, + lease_claim_receipts: receipts, + ..Default::default() + }, + ) + .await; + let job_id = enqueue_job( + &pool, + &store, + &CompleteJob { id: 902 }, + InsertOpts { + queue: queue.to_string(), + priority: 4, + ..Default::default() + }, + ) + .await; + backdate_one_lane_ready_job_run_at( + &pool, + schema, + job_id, + Utc::now() - chrono::Duration::minutes(5), + ) + .await; + let claimed = store + .claim_runtime_batch_with_aging_for_instance( + &pool, + queue, + 1, + Duration::ZERO, + Duration::from_secs(60), + Uuid::new_v4(), + 4, + Duration::from_secs(30), + Duration::from_secs(1), + ) + .await + .expect("claim with aging"); + assert_eq!(claimed.len(), 1); + let run_lease = claimed[0].job.run_lease; + assert!( + claimed[0].job.priority < 4, + "the claim must have aged the job for this test to mean anything" + ); + + let loaded = store + .load_job(&pool, job_id) + .await + .expect("load aged job") + .expect("aged job exists"); + assert_eq!(loaded.state, JobState::Running); + assert_eq!(loaded.run_lease, run_lease); + + let callback_id = store + .register_callback(&pool, job_id, run_lease, Duration::from_secs(300)) + .await + .expect("register callback on aged job"); + let mut tx = pool.begin().await.unwrap(); + let by_callback = store + .callback_job_in_tx(&mut tx, callback_id, Some(run_lease), false) + .await + .expect("callback lookup") + .expect("callback lookup finds the aged job"); + tx.rollback().await.unwrap(); + assert_eq!(by_callback.id, job_id); + assert_eq!(by_callback.run_lease, run_lease); +} + +#[tokio::test] +async fn test_aged_receipt_claim_is_visible_to_job_and_callback_reads() { + aged_claim_is_visible_to_job_and_callback_reads(true).await; +} + +#[tokio::test] +async fn test_aged_materialized_lease_is_visible_to_job_and_callback_reads() { + aged_claim_is_visible_to_job_and_callback_reads(false).await; +}