From a41c85b59b411e3bfd78fb77a060017d2b8cfb55 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Sat, 7 Nov 2026 01:30:36 +1300 Subject: [PATCH 1/3] Fix moved-attempt hydration and protect point-read snapshots --- CHANGELOG.md | 1 + awa-model/src/admin.rs | 45 +++ awa-model/src/queue_storage.rs | 58 ++++ awa/tests/queue_storage_runtime_test.rs | 383 ++++++++++++++++++++++++ 4 files changed, 487 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 80fda9ffd..1eeedca4d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,7 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Fixed +- Exclude tombstoned source rows when hydrating moved running jobs, including compact receipts, and preserve point-read snapshots against concurrent claim/lease partition pruning. - **Ring prune and terminal-count rollup report `Blocked` when any of their statements times out on a lock.** Prune sets a short `lock_timeout` (10 ms by default) before it takes `ACCESS EXCLUSIVE` on a slot's partitions, but diff --git a/awa-model/src/admin.rs b/awa-model/src/admin.rs index a40efb4e5..b45b2b8d3 100644 --- a/awa-model/src/admin.rs +++ b/awa-model/src/admin.rs @@ -381,6 +381,15 @@ pub fn queue_storage_current_jobs_cte(schema: &str) -> String { AND ready.run_lease + 1 = leases.run_lease AND ready.enqueue_shard = leases.enqueue_shard AND ready.lane_seq = leases.lane_seq + AND NOT EXISTS ( + SELECT 1 FROM {schema}.ready_tombstones AS tomb + WHERE tomb.ready_slot = ready.ready_slot + AND tomb.ready_generation = ready.ready_generation + AND tomb.queue = ready.queue + AND tomb.priority = ready.priority + AND tomb.enqueue_shard = ready.enqueue_shard + AND tomb.lane_seq = ready.lane_seq + ) UNION ALL SELECT job_id, kind, queue, state, created_at, run_at, finalized_at FROM {schema}.terminal_jobs @@ -734,6 +743,15 @@ fn queue_storage_recent_jobs_cte(schema: &str, since: &str) -> String { AND ready.run_lease + 1 = leases.run_lease AND ready.enqueue_shard = leases.enqueue_shard AND ready.lane_seq = leases.lane_seq + AND NOT EXISTS ( + SELECT 1 FROM {schema}.ready_tombstones AS tomb + WHERE tomb.ready_slot = ready.ready_slot + AND tomb.ready_generation = ready.ready_generation + AND tomb.queue = ready.queue + AND tomb.priority = ready.priority + AND tomb.enqueue_shard = ready.enqueue_shard + AND tomb.lane_seq = ready.lane_seq + ) WHERE ready.created_at >= {since} UNION ALL -- A queue or priority move copies the row with its times, so @@ -856,6 +874,15 @@ fn queue_storage_kind_counts_cte(schema: &str) -> String { AND ready.run_lease + 1 = leases.run_lease AND ready.enqueue_shard = leases.enqueue_shard AND ready.lane_seq = leases.lane_seq + AND NOT EXISTS ( + SELECT 1 FROM {schema}.ready_tombstones AS tomb + WHERE tomb.ready_slot = ready.ready_slot + AND tomb.ready_generation = ready.ready_generation + AND tomb.queue = ready.queue + AND tomb.priority = ready.priority + AND tomb.enqueue_shard = ready.enqueue_shard + AND tomb.lane_seq = ready.lane_seq + ) GROUP BY ready.kind, leases.queue ), deferred_by_kind AS MATERIALIZED ( @@ -1090,6 +1117,15 @@ fn queue_storage_job_id_candidates_sql( AND ready.run_lease + 1 = leases.run_lease AND ready.enqueue_shard = leases.enqueue_shard AND ready.lane_seq = leases.lane_seq + AND NOT EXISTS ( + SELECT 1 FROM {schema}.ready_tombstones AS tomb + WHERE tomb.ready_slot = ready.ready_slot + AND tomb.ready_generation = ready.ready_generation + AND tomb.queue = ready.queue + AND tomb.priority = ready.priority + AND tomb.enqueue_shard = ready.enqueue_shard + AND tomb.lane_seq = ready.lane_seq + ) WHERE TRUE{predicates} ORDER BY leases.job_id DESC LIMIT {limit} @@ -1552,6 +1588,15 @@ fn unique_key_candidate_sql(schema: &str, scoped: bool) -> String { AND ready.run_lease + 1 = leases.run_lease AND ready.enqueue_shard = leases.enqueue_shard AND ready.lane_seq = leases.lane_seq + AND NOT EXISTS ( + SELECT 1 FROM {schema}.ready_tombstones AS tomb + WHERE tomb.ready_slot = ready.ready_slot + AND tomb.ready_generation = ready.ready_generation + AND tomb.queue = ready.queue + AND tomb.priority = ready.priority + AND tomb.enqueue_shard = ready.enqueue_shard + AND tomb.lane_seq = ready.lane_seq + ) AND ready.unique_key = $1 OFFSET 0 ) AS ready ON TRUE diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 0fb6aee36..b8c11dee5 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -11866,6 +11866,18 @@ impl QueueStorage { .await .map_err(map_sqlx_error)?; + // Protect running evidence from non-MVCC-safe TRUNCATE before + // the first data query establishes this repeatable-read snapshot. + // Claim-family order agrees with claim prune. + sqlx::query(&format!( + "LOCK TABLE {schema}.leases, {schema}.lease_claims, \ + {schema}.lease_claim_batches, {schema}.lease_claim_closures, \ + {schema}.lease_claim_closure_batches IN ACCESS SHARE MODE" + )) + .execute(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + let ready_rows: Vec = sqlx::query_as(&format!( r#" SELECT @@ -11979,6 +11991,15 @@ impl QueueStorage { AND ready.run_lease + 1 = lease.run_lease AND ready.enqueue_shard = lease.enqueue_shard AND ready.lane_seq = lease.lane_seq + AND NOT EXISTS ( + SELECT 1 FROM {schema}.ready_tombstones AS tomb + WHERE tomb.ready_slot = ready.ready_slot + AND tomb.ready_generation = ready.ready_generation + AND tomb.queue = ready.queue + AND tomb.priority = ready.priority + AND tomb.enqueue_shard = ready.enqueue_shard + AND tomb.lane_seq = ready.lane_seq + ) LEFT JOIN {schema}.attempt_state AS attempt ON attempt.job_id = lease.job_id AND attempt.run_lease = lease.run_lease @@ -12039,6 +12060,15 @@ impl QueueStorage { AND ready.run_lease + 1 = claims.run_lease AND ready.enqueue_shard = claims.enqueue_shard AND ready.lane_seq = claims.lane_seq + AND NOT EXISTS ( + SELECT 1 FROM {schema}.ready_tombstones AS tomb + WHERE tomb.ready_slot = ready.ready_slot + AND tomb.ready_generation = ready.ready_generation + AND tomb.queue = ready.queue + AND tomb.priority = ready.priority + AND tomb.enqueue_shard = ready.enqueue_shard + AND tomb.lane_seq = ready.lane_seq + ) LEFT JOIN {schema}.attempt_state AS attempt ON attempt.job_id = claims.job_id AND attempt.run_lease = claims.run_lease @@ -12115,6 +12145,16 @@ impl QueueStorage { AND ready.enqueue_shard = claim_batches.enqueue_shard AND ready.lane_seq = items.lane_seq AND ready.job_id = items.job_id + AND ready.run_lease + 1 = items.run_lease + AND NOT EXISTS ( + SELECT 1 FROM {schema}.ready_tombstones AS tomb + WHERE tomb.ready_slot = ready.ready_slot + AND tomb.ready_generation = ready.ready_generation + AND tomb.queue = ready.queue + AND tomb.priority = ready.priority + AND tomb.enqueue_shard = ready.enqueue_shard + AND tomb.lane_seq = ready.lane_seq + ) LEFT JOIN {schema}.attempt_state AS attempt ON attempt.job_id = items.job_id AND attempt.run_lease = items.run_lease @@ -12567,6 +12607,15 @@ impl QueueStorage { AND ready.run_lease + 1 = lease.run_lease AND ready.enqueue_shard = lease.enqueue_shard AND ready.lane_seq = lease.lane_seq + AND NOT EXISTS ( + SELECT 1 FROM {schema}.ready_tombstones AS tomb + WHERE tomb.ready_slot = ready.ready_slot + AND tomb.ready_generation = ready.ready_generation + AND tomb.queue = ready.queue + AND tomb.priority = ready.priority + AND tomb.enqueue_shard = ready.enqueue_shard + AND tomb.lane_seq = ready.lane_seq + ) LEFT JOIN {schema}.attempt_state AS attempt ON attempt.job_id = lease.job_id AND attempt.run_lease = lease.run_lease @@ -12766,6 +12815,15 @@ impl QueueStorage { AND ready.run_lease + 1 = lease.run_lease AND ready.enqueue_shard = lease.enqueue_shard AND ready.lane_seq = lease.lane_seq + AND NOT EXISTS ( + SELECT 1 FROM {schema}.ready_tombstones AS tomb + WHERE tomb.ready_slot = ready.ready_slot + AND tomb.ready_generation = ready.ready_generation + AND tomb.queue = ready.queue + AND tomb.priority = ready.priority + AND tomb.enqueue_shard = ready.enqueue_shard + AND tomb.lane_seq = ready.lane_seq + ) LEFT JOIN {schema}.attempt_state AS attempt ON attempt.job_id = lease.job_id AND attempt.run_lease = lease.run_lease diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index e80c74c9e..681da1608 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -16848,3 +16848,386 @@ async fn test_point_read_uses_live_ready_queue_after_ring_wrap() { "point read must ignore the tombstoned source" ); } + +/// Aging moves a waiting job by tombstoning its ready row and re-inserting the +/// same attempt on the next lane. In a quiet queue the new lane hands out the +/// same `lane_seq`, so a lease joined to its ready row by attempt alone matched +/// both rows and the admin reads counted the running job twice. +async fn moved_ready_row_is_joined_once_by_attempt_reads(receipts: bool, row_local: bool) { + let (_db_guard, pool) = setup_pool(6).await; + let queue = "qs_moved_ready_lease_reads"; + let schema = if row_local { + "awa_qs_moved_ready_row_claim_reads" + } else if receipts { + "awa_qs_moved_ready_compact_reads" + } else { + "awa_qs_moved_ready_lease_reads" + }; + let store = create_store_with_config( + &pool, + QueueStorageConfig { + lease_claim_receipts: receipts, + ..unique_cancel_store_config(schema) + }, + ) + .await; + let job_id = enqueue_job( + &pool, + &store, + &unique_cancel_job("moved"), + InsertOpts { + priority: 2, + ..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 aged = store + .age_waiting_priorities(&pool, Duration::from_secs(60), 10) + .await + .expect("age waiting priorities"); + assert_eq!(aged, vec![job_id]); + + let same_position: i64 = sqlx::query_scalar(&format!( + "SELECT count(DISTINCT priority)::bigint + FROM {schema}.ready_entries + WHERE job_id = $1 + GROUP BY ready_slot, ready_generation, enqueue_shard, lane_seq" + )) + .bind(job_id) + .fetch_one(&pool) + .await + .expect("ready rows for the moved job"); + assert_eq!( + same_position, 2, + "the source and destination rows must share a position for this test to mean anything" + ); + + let claimed = store + .claim_runtime_batch( + &pool, + queue, + 1, + if receipts && !row_local { + Duration::ZERO + } else { + Duration::from_secs(30) + }, + ) + .await + .expect("claim moved job"); + assert_eq!(claimed.len(), 1); + assert_eq!(claimed[0].job.id, job_id); + assert_eq!(claimed[0].claim.lease_claim_receipt, receipts); + assert_eq!( + claimed[0].claim.claim_batch_id.is_some(), + receipts && !row_local + ); + let legacy_rows: i64 = sqlx::query_scalar(&format!( + "SELECT count(*) FROM {schema}.lease_claims WHERE job_id = $1" + )) + .bind(job_id) + .fetch_one(&pool) + .await + .expect("verify the claim representation"); + assert_eq!(legacy_rows, i64::from(row_local)); + + // The release CTE projects materialized leases; receipt views use their + // existing open-claim sources. Do not add a new projection feature here. + if !receipts { + // This is the production CTE used by current-job projections, not only + // the point-read candidate list. The lease must hydrate exactly once. + let projected: Vec = sqlx::query_scalar(&format!( + "{} SELECT job_id FROM current_jobs WHERE queue = $1 AND state = 'running'", + admin::queue_storage_current_jobs_cte(schema) + )) + .bind(queue) + .fetch_all(&pool) + .await + .expect("project current running jobs"); + assert_eq!(projected, vec![job_id], "hydrate only the live ready row"); + } + + if !receipts { + 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 moved running job is listed once" + ); + + let overview = admin::job_kind_overviews(&pool) + .await + .expect("job kind overviews"); + let kind = overview + .iter() + .find(|row| row.kind == UniqueCancelJob::kind()) + .expect("the moved job's kind is listed"); + assert_eq!(kind.job_count, 1, "the moved running job is counted once"); + } + + let loaded = store + .load_job(&pool, job_id) + .await + .expect("load moved job") + .expect("moved job exists"); + assert_eq!(loaded.state, JobState::Running); + assert_eq!(loaded.priority, 1, "reads the live destination row"); + assert_eq!( + loaded.metadata.get("_awa_original_priority"), + Some(&serde_json::json!(2)), + "payload comes from the aged destination, not its tombstoned source" + ); +} + +#[tokio::test] +async fn test_moved_ready_row_is_joined_once_by_lease_reads() { + moved_ready_row_is_joined_once_by_attempt_reads(false, false).await; +} + +#[tokio::test] +async fn test_moved_ready_row_is_joined_once_by_compact_receipt_reads() { + moved_ready_row_is_joined_once_by_attempt_reads(true, false).await; +} + +#[tokio::test] +async fn test_moved_ready_row_is_joined_once_by_row_local_receipt_reads() { + moved_ready_row_is_joined_once_by_attempt_reads(true, true).await; +} + +/// A point read's snapshot can outlive completion. TRUNCATE is not MVCC-safe: +/// pruning its claim/lease partition before the reader visits it would erase +/// the old running evidence, while the new completion is outside that snapshot. +async fn point_read_snapshot_survives_ring_prune(receipts: bool) { + let (_db_guard, pool) = setup_pool(6).await; + let schema = if receipts { + "awa_qs_snapshot_claim_prune" + } else { + "awa_qs_snapshot_lease_prune" + }; + let store = Arc::new( + create_store_with_config( + &pool, + QueueStorageConfig { + schema: schema.into(), + queue_slot_count: 4, + lease_slot_count: 2, + claim_slot_count: 2, + lease_claim_receipts: receipts, + ..Default::default() + }, + ) + .await, + ); + let queue = "snapshot_prune"; + let job_id = enqueue_job( + &pool, + &store, + &CompleteJob { id: 1 }, + InsertOpts { + queue: queue.into(), + ..Default::default() + }, + ) + .await; + let claimed = store + .claim_runtime_batch( + &pool, + queue, + 1, + if receipts { + Duration::ZERO + } else { + Duration::from_secs(30) + }, + ) + .await + .expect("claim the snapshot probe"); + assert_eq!(claimed.len(), 1); + assert_eq!(claimed[0].claim.claim_batch_id.is_some(), receipts); + assert_eq!( + store.load_job(&pool, job_id).await.unwrap().unwrap().state, + JobState::Running + ); + + // Gate the reader's second query without blocking completion's own + // deferred-state checks. Only this reader's application name takes the + // advisory lock; the view's scalar initplan runs even with no deferred rows. + sqlx::raw_sql(&format!( + r#"CREATE FUNCTION {schema}.snapshot_read_gate() RETURNS boolean + LANGUAGE plpgsql VOLATILE AS $$ + BEGIN + IF current_setting('application_name') = 'awa_snapshot_reader' THEN + PERFORM pg_advisory_xact_lock(60006, 1); + END IF; + RETURN TRUE; + END; + $$; + ALTER TABLE {schema}.deferred_jobs RENAME TO deferred_rows; + CREATE VIEW {schema}.deferred_jobs AS + SELECT * FROM {schema}.deferred_rows + WHERE (SELECT {schema}.snapshot_read_gate())"# + )) + .execute(&pool) + .await + .expect("install reader-only query gate"); + let mut gate = pool.begin().await.expect("begin read gate"); + sqlx::query("SELECT pg_advisory_xact_lock(60006, 1)") + .execute(gate.as_mut()) + .await + .expect("hold the second read"); + let reader_pool = PgPoolOptions::new() + .max_connections(1) + .connect_with( + (*pool.connect_options()) + .clone() + .application_name("awa_snapshot_reader"), + ) + .await + .expect("connect isolated snapshot reader"); + let reader_store = store.clone(); + let reader_connection = reader_pool.clone(); + let reader = + tokio::spawn(async move { reader_store.load_job(&reader_connection, job_id).await }); + let deadline = Instant::now() + ci_timing::scaled_timeout(Duration::from_secs(10)); + loop { + let at_gate: bool = sqlx::query_scalar( + "SELECT EXISTS ( + SELECT 1 FROM pg_locks + WHERE locktype = 'advisory' + AND database = (SELECT oid FROM pg_database WHERE datname = current_database()) + AND classid = 60006 AND objid = 1 AND NOT granted + )", + ) + .fetch_one(&pool) + .await + .expect("observe the reader-only gate"); + if at_gate { + break; + } + assert!( + Instant::now() < deadline, + "point read did not reach its second query" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } + + let during_read = + tokio::time::timeout(ci_timing::scaled_timeout(Duration::from_secs(10)), async { + assert_eq!( + store + .complete_runtime_batch(&pool, &claimed) + .await + .expect("complete outside the snapshot") + .len(), + 1 + ); + let rotation = if receipts { + store.rotate_claims(&pool).await + } else { + store.rotate_leases(&pool).await + } + .expect("seal the completed attempt's slot"); + assert!(matches!(rotation, RotateOutcome::Rotated { slot: 1, .. })); + if receipts { + store + .prune_oldest_claims(&pool) + .await + .expect("prune while the snapshot is live") + } else { + // The release lease-pruner skips an already empty child. Use + // the actual destructive SQL here to exercise its MVCC hazard. + let mut prune = pool.begin().await.expect("begin lease truncate"); + sqlx::query("SET LOCAL lock_timeout = '10ms'") + .execute(prune.as_mut()) + .await + .expect("bound lease truncate"); + let result = sqlx::query(&format!("TRUNCATE TABLE {schema}.leases_0")) + .execute(prune.as_mut()) + .await; + match result { + Ok(_) => { + prune.commit().await.expect("commit lease truncate"); + PruneOutcome::Pruned { + slot: 0, + carried_failed_rows: 0, + } + } + Err(sqlx::Error::Database(error)) + if error.code().as_deref() == Some("55P03") => + { + prune + .rollback() + .await + .expect("release blocked lease truncate"); + PruneOutcome::Blocked { slot: 0 } + } + Err(error) => panic!("lease truncate failed: {error}"), + } + } + }) + .await; + + // Always release the gate and finish the reader before asserting the + // outcome, so the negative control also reports the inconsistent state. + gate.rollback().await.expect("release the read gate"); + let snapshot = tokio::time::timeout(ci_timing::scaled_timeout(Duration::from_secs(10)), reader) + .await + .expect("point read finished") + .expect("join point read") + .expect("point read succeeds") + .expect("the job remains in its snapshot"); + let during_read = + during_read.expect("completion/prune must not wait on the deferred read gate"); + assert_eq!( + ( + matches!(during_read, PruneOutcome::Blocked { slot: 0 }), + snapshot.state + ), + (true, JobState::Running), + "prune must retain the snapshot's running evidence: {during_read:?}" + ); + assert_eq!(snapshot.run_lease, claimed[0].job.run_lease); + reader_pool.close().await; + if receipts { + let after_read = store + .prune_oldest_claims(&pool) + .await + .expect("prune after the reader releases its snapshot"); + assert!(matches!(after_read, PruneOutcome::Pruned { slot: 0, .. })); + } else { + sqlx::query(&format!("TRUNCATE TABLE {schema}.leases_0")) + .execute(&pool) + .await + .expect("truncate after the reader releases its snapshot"); + } + assert_eq!( + store.load_job(&pool, job_id).await.unwrap().unwrap().state, + JobState::Completed + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_point_read_snapshot_blocks_compact_claim_prune() { + point_read_snapshot_survives_ring_prune(true).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_point_read_snapshot_blocks_lease_truncate() { + point_read_snapshot_survives_ring_prune(false).await; +} From d43db1138753b72026efebca1c9ca46ac8056738 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Sat, 7 Nov 2026 02:08:36 +1300 Subject: [PATCH 2/3] Include running receipt claims in release admin reads Backport receipt-aware job list projections and kind counts for compact and row-local claims. Preserve exact live attempt hydration and closure, terminal and materialization exclusions. Cover scope, paging and disposition for both receipt formats and the materialized lease control. --- CHANGELOG.md | 2 + awa-model/src/admin.rs | 202 ++++++++++++++ awa/tests/queue_storage_runtime_test.rs | 345 +++++++++++++++++++++--- 3 files changed, 509 insertions(+), 40 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1eeedca4d..ed0019fb3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,8 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Fixed +- Admin job lists and kind counts include running receipt claims before + they materialize into leases, including compact and legacy row-local claims. - Exclude tombstoned source rows when hydrating moved running jobs, including compact receipts, and preserve point-read snapshots against concurrent claim/lease partition pruning. - **Ring prune and terminal-count rollup report `Blocked` when any of their statements times out on a lock.** Prune sets a short `lock_timeout` (10 ms diff --git a/awa-model/src/admin.rs b/awa-model/src/admin.rs index b45b2b8d3..8737f364d 100644 --- a/awa-model/src/admin.rs +++ b/awa-model/src/admin.rs @@ -391,6 +391,147 @@ pub fn queue_storage_current_jobs_cte(schema: &str) -> String { AND tomb.lane_seq = ready.lane_seq ) UNION ALL + -- Receipt claims that have not materialised into `leases` are + -- running too (#246 / #416). They live on legacy row-local + -- `lease_claims` or, by default, compact `lease_claim_batches`. + -- Report both as `running`, anti-joined against every durable + -- closure/terminal/materialised-lease shape. This mirrors + -- QueueStorage::load_job (the batch-aware single-job admin read) + -- and the canonical open-receipt shape in + -- queue_storage::open_receipt_running_claims_sql; it is spelled + -- out here because this view projects the full job row + -- (kind/created_at/run_at from `ready_entries`), not just claim + -- identity. + SELECT + claims.job_id, + ready.kind, + claims.queue, + 'running'::awa.job_state AS state, + ready.created_at, + ready.run_at, + NULL::timestamptz AS finalized_at + FROM {schema}.lease_claims AS claims + JOIN {schema}.ready_entries AS ready + ON ready.ready_slot = claims.ready_slot + AND ready.ready_generation = claims.ready_generation + AND ready.queue = claims.queue + -- Claims keep the aged effective priority; match the ready attempt. + 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 + -- Moving a ready row tombstones the old one and re-inserts the same + -- attempt, which can reuse the old lane_seq on the new lane. + AND NOT EXISTS ( + SELECT 1 FROM {schema}.ready_tombstones AS tomb + WHERE tomb.ready_slot = ready.ready_slot + AND tomb.ready_generation = ready.ready_generation + AND tomb.queue = ready.queue + AND tomb.priority = ready.priority + AND tomb.enqueue_shard = ready.enqueue_shard + AND tomb.lane_seq = ready.lane_seq + ) + WHERE claims.closed_at IS NULL + AND NOT EXISTS ( + SELECT 1 FROM {schema}.lease_claim_closures AS cx + WHERE cx.claim_slot = claims.claim_slot + AND cx.job_id = claims.job_id + AND cx.run_lease = claims.run_lease + ) + AND NOT EXISTS ( + SELECT 1 FROM {schema}.lease_claim_closure_batches AS cb + WHERE cb.receipt_ranges @> claims.receipt_id + ) + AND NOT EXISTS ( + SELECT 1 FROM {schema}.leases AS lease + WHERE lease.job_id = claims.job_id + AND lease.run_lease = claims.run_lease + ) + -- Terminal supersession counts as closed evidence even + -- without an explicit closure row (mirrors the compact + -- branch below and open_receipt_running_claims_sql); + -- without these a legacy row-local claim would be listed + -- as running AND terminal at once. + AND NOT EXISTS ( + SELECT 1 FROM {schema}.done_entries AS done + WHERE done.job_id = claims.job_id + AND done.run_lease = claims.run_lease + ) + AND NOT EXISTS ( + SELECT 1 FROM {schema}.deferred_jobs AS deferred + WHERE deferred.job_id = claims.job_id + AND deferred.run_lease = claims.run_lease + ) + AND NOT EXISTS ( + SELECT 1 FROM {schema}.dlq_entries AS dlq + WHERE dlq.job_id = claims.job_id + AND dlq.run_lease = claims.run_lease + ) + UNION ALL + SELECT + items.job_id, + ready.kind, + batches.queue, + 'running'::awa.job_state AS state, + ready.created_at, + ready.run_at, + NULL::timestamptz AS finalized_at + FROM {schema}.lease_claim_batches AS batches + CROSS JOIN LATERAL unnest( + batches.job_ids, + batches.run_leases, + batches.receipt_ids, + batches.lane_seqs + ) AS items(job_id, run_lease, receipt_id, lane_seq) + JOIN {schema}.ready_entries AS ready + ON ready.ready_slot = batches.ready_slot + AND ready.ready_generation = batches.ready_generation + AND ready.queue = batches.queue + AND ready.enqueue_shard = batches.enqueue_shard + AND ready.lane_seq = items.lane_seq + AND ready.job_id = items.job_id + AND ready.run_lease + 1 = items.run_lease + AND NOT EXISTS ( + SELECT 1 FROM {schema}.ready_tombstones AS tomb + WHERE tomb.ready_slot = ready.ready_slot + AND tomb.ready_generation = ready.ready_generation + AND tomb.queue = ready.queue + AND tomb.priority = ready.priority + AND tomb.enqueue_shard = ready.enqueue_shard + AND tomb.lane_seq = ready.lane_seq + ) + WHERE NOT EXISTS ( + SELECT 1 FROM {schema}.lease_claim_closures AS cx + WHERE cx.claim_slot = batches.claim_slot + AND cx.job_id = items.job_id + AND cx.run_lease = items.run_lease + ) + AND NOT EXISTS ( + SELECT 1 FROM {schema}.lease_claim_closure_batches AS cb + WHERE cb.claim_slot = batches.claim_slot + AND cb.receipt_ranges @> items.receipt_id + ) + AND NOT EXISTS ( + SELECT 1 FROM {schema}.leases AS lease + WHERE lease.job_id = items.job_id + AND lease.run_lease = items.run_lease + ) + AND NOT EXISTS ( + SELECT 1 FROM {schema}.done_entries AS done + WHERE done.job_id = items.job_id + AND done.run_lease = items.run_lease + ) + AND NOT EXISTS ( + SELECT 1 FROM {schema}.deferred_jobs AS deferred + WHERE deferred.job_id = items.job_id + AND deferred.run_lease = items.run_lease + ) + AND NOT EXISTS ( + SELECT 1 FROM {schema}.dlq_entries AS dlq + WHERE dlq.job_id = items.job_id + AND dlq.run_lease = items.run_lease + ) + UNION ALL SELECT job_id, kind, queue, state, created_at, run_at, finalized_at FROM {schema}.terminal_jobs UNION ALL @@ -838,6 +979,7 @@ fn queue_storage_recent_jobs_cte(schema: &str, since: &str) -> String { /// probes the ready primary key. fn queue_storage_kind_counts_cte(schema: &str) -> String { let available = queue_storage_available_lane_rows_sql(schema, "heads", "ready.kind", "TRUE"); + let open_running = crate::queue_storage::open_receipt_running_claims_sql(schema); format!( r#" heads AS MATERIALIZED ( @@ -890,6 +1032,30 @@ fn queue_storage_kind_counts_cte(schema: &str) -> String { FROM {schema}.deferred_jobs GROUP BY kind, queue ), + -- Count each open claim once using its live ready row. Materialized + -- leases and closed claims are excluded by open_running. + running_by_kind AS MATERIALIZED ( + SELECT created.kind, open_claims.queue, count(*)::bigint AS n, 0::bigint AS n_recent + FROM ({open_running}) AS open_claims + CROSS JOIN LATERAL ( + SELECT ready.kind + FROM {schema}.ready_entries AS ready + WHERE ready.job_id = open_claims.job_id + AND ready.run_lease + 1 = open_claims.run_lease + AND ready.queue = open_claims.queue + AND NOT EXISTS ( + SELECT 1 FROM {schema}.ready_tombstones AS tomb + WHERE tomb.ready_slot = ready.ready_slot + AND tomb.ready_generation = ready.ready_generation + AND tomb.queue = ready.queue + AND tomb.priority = ready.priority + AND tomb.enqueue_shard = ready.enqueue_shard + AND tomb.lane_seq = ready.lane_seq + ) + LIMIT 1 + ) AS created + GROUP BY created.kind, open_claims.queue + ), done_by_kind AS MATERIALIZED ( SELECT kind, @@ -943,6 +1109,7 @@ fn queue_storage_kind_counts_cte(schema: &str) -> String { FROM ( SELECT * FROM available_by_kind UNION ALL SELECT * FROM leased_by_kind + UNION ALL SELECT * FROM running_by_kind UNION ALL SELECT * FROM deferred_by_kind UNION ALL SELECT * FROM done_by_kind UNION ALL SELECT * FROM receipts_by_kind @@ -1132,6 +1299,41 @@ fn queue_storage_job_id_candidates_sql( )"# )); } + if holds(&[JobState::Running]) { + let open_running = crate::queue_storage::open_receipt_running_claims_sql(schema); + let predicates = predicates( + None, + Some("created.kind"), + "open_claims.queue", + "open_claims.job_id", + ); + legs.push(format!( + r#"( + SELECT open_claims.job_id + FROM ({open_running}) AS open_claims + CROSS JOIN LATERAL ( + SELECT ready.kind + FROM {schema}.ready_entries AS ready + WHERE ready.job_id = open_claims.job_id + AND ready.run_lease + 1 = open_claims.run_lease + AND ready.queue = open_claims.queue + AND NOT EXISTS ( + SELECT 1 FROM {schema}.ready_tombstones AS tomb + WHERE tomb.ready_slot = ready.ready_slot + AND tomb.ready_generation = ready.ready_generation + AND tomb.queue = ready.queue + AND tomb.priority = ready.priority + AND tomb.enqueue_shard = ready.enqueue_shard + AND tomb.lane_seq = ready.lane_seq + ) + LIMIT 1 + ) AS created + WHERE TRUE{predicates} + ORDER BY open_claims.job_id DESC + LIMIT {limit} + )"# + )); + } if holds(&[JobState::Completed, JobState::Failed, JobState::Cancelled]) { let predicates = predicates(Some("state"), Some("kind"), "queue", "job_id"); legs.push(format!( diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index 681da1608..573ed3f4d 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -16938,48 +16938,42 @@ async fn moved_ready_row_is_joined_once_by_attempt_reads(receipts: bool, row_loc .expect("verify the claim representation"); assert_eq!(legacy_rows, i64::from(row_local)); - // The release CTE projects materialized leases; receipt views use their - // existing open-claim sources. Do not add a new projection feature here. - if !receipts { - // This is the production CTE used by current-job projections, not only - // the point-read candidate list. The lease must hydrate exactly once. - let projected: Vec = sqlx::query_scalar(&format!( - "{} SELECT job_id FROM current_jobs WHERE queue = $1 AND state = 'running'", - admin::queue_storage_current_jobs_cte(schema) - )) - .bind(queue) - .fetch_all(&pool) - .await - .expect("project current running jobs"); - assert_eq!(projected, vec![job_id], "hydrate only the live ready row"); - } + // This is the production CTE used by current-job projections, not only + // the point-read candidate list. Every attempt must hydrate exactly once. + let projected: Vec = sqlx::query_scalar(&format!( + "{} SELECT job_id FROM current_jobs WHERE queue = $1 AND state = 'running'", + admin::queue_storage_current_jobs_cte(schema) + )) + .bind(queue) + .fetch_all(&pool) + .await + .expect("project current running jobs"); + assert_eq!(projected, vec![job_id], "hydrate only the live ready row"); - if !receipts { - 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 moved running job is listed once" - ); + 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 moved running job is listed once" + ); - let overview = admin::job_kind_overviews(&pool) - .await - .expect("job kind overviews"); - let kind = overview - .iter() - .find(|row| row.kind == UniqueCancelJob::kind()) - .expect("the moved job's kind is listed"); - assert_eq!(kind.job_count, 1, "the moved running job is counted once"); - } + let overview = admin::job_kind_overviews(&pool) + .await + .expect("job kind overviews"); + let kind = overview + .iter() + .find(|row| row.kind == UniqueCancelJob::kind()) + .expect("the moved job's kind is listed"); + assert_eq!(kind.job_count, 1, "the moved running job is counted once"); let loaded = store .load_job(&pool, job_id) @@ -17231,3 +17225,274 @@ async fn test_point_read_snapshot_blocks_compact_claim_prune() { async fn test_point_read_snapshot_blocks_lease_truncate() { point_read_snapshot_survives_ring_prune(false).await; } + +async fn receipt_admin_ids(pool: &sqlx::PgPool, filter: admin::ListJobsFilter) -> Vec { + admin::list_jobs(pool, &filter) + .await + .expect("list receipt-held jobs") + .into_iter() + .map(|job| job.id) + .collect() +} + +async fn receipt_admin_reads_preserve_scope_and_disposition(receipts: bool, row_local: bool) { + let (_db_guard, pool) = setup_pool(6).await; + let schema = match (receipts, row_local) { + (false, _) => "awa_qs_admin_scope_leases", + (true, false) => "awa_qs_admin_scope_compact", + (true, true) => "awa_qs_admin_scope_row_claims", + }; + let store = create_store_with_config( + &pool, + QueueStorageConfig { + lease_claim_receipts: receipts, + ..unique_cancel_store_config(schema) + }, + ) + .await; + let queue = "receipt_admin_scope"; + let other_queue = "receipt_admin_other"; + let first = enqueue_job( + &pool, + &store, + &CompleteJob { id: 1 }, + InsertOpts { + queue: queue.into(), + tags: vec!["selected".into()], + ..Default::default() + }, + ) + .await; + let second = enqueue_job( + &pool, + &store, + &CompleteJob { id: 2 }, + InsertOpts { + queue: queue.into(), + tags: vec!["selected".into()], + ..Default::default() + }, + ) + .await; + let other_kind = enqueue_job( + &pool, + &store, + &DlqJob { id: 3 }, + InsertOpts { + queue: queue.into(), + tags: vec!["selected".into()], + ..Default::default() + }, + ) + .await; + let other = enqueue_job( + &pool, + &store, + &CompleteJob { id: 4 }, + InsertOpts { + queue: other_queue.into(), + tags: vec!["selected".into()], + ..Default::default() + }, + ) + .await; + let deadline = if receipts && !row_local { + Duration::ZERO + } else { + Duration::from_secs(30) + }; + let mut claims = store + .claim_runtime_batch(&pool, queue, 3, deadline) + .await + .unwrap(); + claims.extend( + store + .claim_runtime_batch(&pool, other_queue, 1, deadline) + .await + .unwrap(), + ); + assert_eq!(claims.len(), 4); + assert!(claims + .iter() + .all(|claim| claim.claim.lease_claim_receipt == receipts)); + assert!(claims + .iter() + .all(|claim| claim.claim.claim_batch_id.is_some() == (receipts && !row_local))); + let row_claims: i64 = + sqlx::query_scalar(&format!("SELECT count(*) FROM {schema}.lease_claims")) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(row_claims, if row_local { 4 } else { 0 }); + + let scoped = admin::ListJobsFilter { + state: Some(JobState::Running), + kind: Some(CompleteJob::kind().into()), + queue: Some(queue.into()), + tag: Some("selected".into()), + limit: Some(1), + ..Default::default() + }; + assert_eq!(receipt_admin_ids(&pool, scoped.clone()).await, vec![second]); + assert_eq!( + receipt_admin_ids( + &pool, + admin::ListJobsFilter { + before_id: Some(second), + ..scoped.clone() + } + ) + .await, + vec![first] + ); + assert!(receipt_admin_ids( + &pool, + admin::ListJobsFilter { + before_id: Some(first), + ..scoped.clone() + } + ) + .await + .is_empty()); + assert!(receipt_admin_ids( + &pool, + admin::ListJobsFilter { + tag: Some("absent".into()), + ..scoped.clone() + } + ) + .await + .is_empty()); + assert!(receipt_admin_ids( + &pool, + admin::ListJobsFilter { + state: Some(JobState::WaitingExternal), + ..scoped.clone() + } + ) + .await + .is_empty()); + assert_eq!( + receipt_admin_ids( + &pool, + admin::ListJobsFilter { + state: None, + limit: Some(10), + ..scoped.clone() + } + ) + .await, + vec![second, first] + ); + + let projected: Vec = sqlx::query_scalar(&format!( + "{} SELECT job_id FROM current_jobs WHERE state='running' ORDER BY job_id DESC", + admin::queue_storage_current_jobs_cte(schema) + )) + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!(projected, vec![other, other_kind, second, first]); + let kinds = admin::job_kind_overviews(&pool).await.unwrap(); + let complete = kinds + .iter() + .find(|kind| kind.kind == CompleteJob::kind()) + .unwrap(); + assert_eq!((complete.job_count, complete.queue_count), (3, 2)); + let alternate = kinds + .iter() + .find(|kind| kind.kind == DlqJob::kind()) + .unwrap(); + assert_eq!((alternate.job_count, alternate.queue_count), (1, 1)); + + // Materialization keeps the same attempt running. The receipt leg must + // yield to the lease leg rather than report both representations. + let first_claim = claims.iter().find(|claim| claim.job.id == first).unwrap(); + store + .register_callback( + &pool, + first, + first_claim.job.run_lease, + Duration::from_secs(60), + ) + .await + .expect("materialize one running attempt"); + assert_eq!( + receipt_admin_ids( + &pool, + admin::ListJobsFilter { + limit: Some(10), + ..scoped.clone() + } + ) + .await, + vec![second, first] + ); + let running: i64 = sqlx::query_scalar(&format!( + "{} SELECT count(*) FROM current_jobs WHERE state='running'", + admin::queue_storage_current_jobs_cte(schema) + )) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(running, 4, "materialization does not duplicate the claim"); + + let closed = claims + .iter() + .find(|claim| claim.job.id == second) + .unwrap() + .clone(); + assert_eq!( + store + .complete_runtime_batch(&pool, &[closed]) + .await + .unwrap() + .len(), + 1 + ); + assert_eq!( + receipt_admin_ids( + &pool, + admin::ListJobsFilter { + limit: Some(10), + ..scoped + } + ) + .await, + vec![first], + "closed receipts leave running enumeration" + ); + let running: i64 = sqlx::query_scalar(&format!( + "{} SELECT count(*) FROM current_jobs WHERE state='running'", + admin::queue_storage_current_jobs_cte(schema) + )) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(running, 3); + let kinds = admin::job_kind_overviews(&pool).await.unwrap(); + let complete = kinds + .iter() + .find(|kind| kind.kind == CompleteJob::kind()) + .unwrap(); + assert_eq!( + (complete.job_count, complete.queue_count), + (3, 2), + "the completed job is counted once, not as both running and terminal" + ); +} + +#[tokio::test] +async fn test_receipt_admin_scope_compact() { + receipt_admin_reads_preserve_scope_and_disposition(true, false).await; +} + +#[tokio::test] +async fn test_receipt_admin_scope_row_local() { + receipt_admin_reads_preserve_scope_and_disposition(true, true).await; +} + +#[tokio::test] +async fn test_receipt_admin_scope_materialized_control() { + receipt_admin_reads_preserve_scope_and_disposition(false, false).await; +} From 30037c6f9365a259e7c1909ae7d134d4a538eeba Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Sat, 7 Nov 2026 04:06:40 +1300 Subject: [PATCH 3/3] test(admin): include receipts in conservation checks --- awa/tests/admin_read_equivalence_test.rs | 22 +++------------------- 1 file changed, 3 insertions(+), 19 deletions(-) diff --git a/awa/tests/admin_read_equivalence_test.rs b/awa/tests/admin_read_equivalence_test.rs index 6de0bbb74..cf122a03a 100644 --- a/awa/tests/admin_read_equivalence_test.rs +++ b/awa/tests/admin_read_equivalence_test.rs @@ -26,9 +26,6 @@ struct Fixture { store: QueueStorage, /// Jobs inserted, by kind. inserted: BTreeMap, - /// Jobs left running on open receipt claims, by kind. `current_jobs` - /// on this branch doesn't list them. - on_open_claims: BTreeMap, } impl Fixture { @@ -56,7 +53,6 @@ impl Fixture { pool, store, inserted: BTreeMap::new(), - on_open_claims: BTreeMap::new(), }) } @@ -255,9 +251,6 @@ impl Fixture { .await?; } // The rest stay running on open receipt claims. - for claim in iter { - *self.on_open_claims.entry(claim.job.kind).or_default() += 1; - } // Running on open receipt claims, with a callback registered on some // (which materialises a lease), and one waiting on its callback. @@ -275,12 +268,6 @@ impl Fixture { .await?; } let waiting = &running[2]; - for claim in &running[3..] { - *self - .on_open_claims - .entry(claim.job.kind.clone()) - .or_default() += 1; - } let callback = self .store .register_callback( @@ -299,9 +286,7 @@ impl Fixture { let cancelled = self .enqueue("epsilon", "eq_cancel", 2, 9, &[], None) .await?; - for claim in self.claim("eq_cancel", 2).await? { - *self.on_open_claims.entry(claim.job.kind).or_default() += 1; - } + self.claim("eq_cancel", 2).await?; for job in &cancelled[4..7] { admin::cancel(&self.pool, job.id).await?; } @@ -457,10 +442,9 @@ async fn job_kind_overviews_match_the_current_jobs_definition() -> TestResult { .iter() .find(|overview| &overview.kind == kind) .unwrap_or_else(|| panic!("no overview for {kind}")); - let listed = inserted - fixture.on_open_claims.get(kind).copied().unwrap_or(0); assert_eq!( - overview.job_count, listed, - "job_count for {kind}: every listed job is counted once" + overview.job_count, *inserted, + "job_count for {kind}: every inserted job is counted once, including open receipts" ); } let by_kind: BTreeMap<_, _> = actual