Find aged running jobs and read get_job from one snapshot (0.6)

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

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

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<ReadyJobRow> = 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

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<_>>(),
+        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);
+}