diff --git a/CHANGELOG.md b/CHANGELOG.md index 879a17deb..a1b0a6839 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,21 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Fixed +- **Callback completion, `get_job`, lifecycle events and the admin reads + 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 eight 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, read back + from `get_job` in the wrong state, was missing from the job list, the kinds + page and the dashboard's recent-jobs counts, and was not found by + `cancel_by_unique_key`. 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. - **`admin::job_kind_overviews` on queue storage no longer depends on the planner's row estimates.** Behind the admin UI's kinds page, it counted every job through the `current_jobs` view. Two nodes of that view are diff --git a/awa-model/src/admin.rs b/awa-model/src/admin.rs index ca3250567..a40efb4e5 100644 --- a/awa-model/src/admin.rs +++ b/awa-model/src/admin.rs @@ -375,7 +375,10 @@ pub fn queue_storage_current_jobs_cte(schema: &str) -> String { ON ready.ready_slot = leases.ready_slot AND ready.ready_generation = leases.ready_generation AND ready.queue = leases.queue - AND ready.priority = leases.priority + -- Leases record the aged effective priority while the ready row + -- keeps its lane priority, so match the attempt instead. + AND ready.job_id = leases.job_id + AND ready.run_lease + 1 = leases.run_lease AND ready.enqueue_shard = leases.enqueue_shard AND ready.lane_seq = leases.lane_seq UNION ALL @@ -725,7 +728,10 @@ fn queue_storage_recent_jobs_cte(schema: &str, since: &str) -> String { ON ready.ready_slot = leases.ready_slot AND ready.ready_generation = leases.ready_generation AND ready.queue = leases.queue - AND ready.priority = leases.priority + -- Leases record the aged effective priority while the ready row + -- keeps its lane priority, so match the attempt instead. + AND ready.job_id = leases.job_id + AND ready.run_lease + 1 = leases.run_lease AND ready.enqueue_shard = leases.enqueue_shard AND ready.lane_seq = leases.lane_seq WHERE ready.created_at >= {since} @@ -844,7 +850,10 @@ fn queue_storage_kind_counts_cte(schema: &str) -> String { ON ready.ready_slot = leases.ready_slot AND ready.ready_generation = leases.ready_generation AND ready.queue = leases.queue - AND ready.priority = leases.priority + -- Leases record the aged effective priority while the ready row + -- keeps its lane priority, so match the attempt instead. + AND ready.job_id = leases.job_id + AND ready.run_lease + 1 = leases.run_lease AND ready.enqueue_shard = leases.enqueue_shard AND ready.lane_seq = leases.lane_seq GROUP BY ready.kind, leases.queue @@ -1075,7 +1084,10 @@ fn queue_storage_job_id_candidates_sql( ON ready.ready_slot = leases.ready_slot AND ready.ready_generation = leases.ready_generation AND ready.queue = leases.queue - AND ready.priority = leases.priority + -- Leases record the aged effective priority while the ready row + -- keeps its lane priority, so match the attempt instead. + AND ready.job_id = leases.job_id + AND ready.run_lease + 1 = leases.run_lease AND ready.enqueue_shard = leases.enqueue_shard AND ready.lane_seq = leases.lane_seq WHERE TRUE{predicates} @@ -1534,7 +1546,10 @@ fn unique_key_candidate_sql(schema: &str, scoped: bool) -> String { WHERE ready.ready_slot = leases.ready_slot AND ready.ready_generation = leases.ready_generation AND ready.queue = leases.queue - AND ready.priority = leases.priority + -- Leases record the aged effective priority while the ready row + -- keeps its lane priority, so match the attempt instead. + AND ready.job_id = leases.job_id + AND ready.run_lease + 1 = leases.run_lease AND ready.enqueue_shard = leases.enqueue_shard AND ready.lane_seq = leases.lane_seq AND ready.unique_key = $1 diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 930aab715..3b0b1503f 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -11670,6 +11670,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#" @@ -11694,7 +11701,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 { @@ -11725,7 +11732,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 { @@ -11769,7 +11776,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 @@ -11780,7 +11790,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 { @@ -11826,7 +11836,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 @@ -11846,7 +11859,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 { @@ -11947,7 +11960,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 { @@ -11983,7 +11996,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 { @@ -12018,13 +12031,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, @@ -12350,7 +12364,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 @@ -12546,7 +12563,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 a060369f8..376548e7f 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -15367,3 +15367,183 @@ async fn test_queue_storage_receipt_claims_are_locked_before_the_heartbeat_state .expect("join heartbeat") .expect("heartbeat after the rewrite"); } + +/// 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; +} + +/// The admin reads join a materialized lease to its ready row. A lease records +/// the aged effective priority, so joining on priority hid a job that aged +/// before its claim from the job list, the kinds page and unique-key cancel. +#[tokio::test] +async fn test_aged_materialized_lease_is_visible_to_admin_reads() { + let (_db_guard, pool) = setup_pool(6).await; + let queue = "qs_aged_lease_admin_reads"; + let schema = "awa_qs_aged_lease_admin_reads"; + let store = create_store_with_config( + &pool, + QueueStorageConfig { + lease_claim_receipts: false, + ..unique_cancel_store_config(schema) + }, + ) + .await; + let job_id = enqueue_job( + &pool, + &store, + &unique_cancel_job("aged"), + InsertOpts { + priority: 4, + ..unique_cancel_opts(queue, None) + }, + ) + .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); + assert!( + claimed[0].job.priority < 4, + "the claim must have aged the job for this test to mean anything" + ); + + let running = admin::list_jobs( + &pool, + &admin::ListJobsFilter { + state: Some(JobState::Running), + queue: Some(queue.to_string()), + ..Default::default() + }, + ) + .await + .expect("list running jobs"); + assert_eq!( + running.iter().map(|job| job.id).collect::>(), + vec![job_id], + "the aged running job must be listed" + ); + + let overview = admin::job_kind_overviews(&pool) + .await + .expect("job kind overviews"); + let kind = overview + .iter() + .find(|row| row.kind == UniqueCancelJob::kind()) + .expect("the aged job's kind is listed"); + assert_eq!(kind.job_count, 1, "the aged running job must be counted"); + + let cancelled = cancel_unique(&pool, Some(queue), "aged") + .await + .expect("cancel by unique key finds the aged running job"); + assert_eq!(cancelled.id, job_id); +}