From d6ee3a89f1ca38fe11c7e2002f012e06ffb855d3 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Sun, 8 Nov 2026 22:56:57 +1300 Subject: [PATCH] fix(admin): read claims in installed slots beyond the default ring size Admin job lists and kind counts open the active queue-storage schema with the default configuration, so they carry the default claim slot count. Populated claim slots read from the database were rejected when at or above that count, failing lists and kind counts with "invalid physical claim slot N" on a schema installed with a larger claim ring, even though the partition exists. Take the non-negative slot numbers read from stored rows as installed, as is already done for ready slots. --- CHANGELOG.md | 4 ++ awa-model/src/admin.rs | 11 ++--- awa-model/src/queue_storage.rs | 2 +- awa/tests/queue_storage_runtime_test.rs | 61 +++++++++++++++++++++++++ 4 files changed, 70 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a2fb1494b..832a6558d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ Notable changes between releases. Detailed migration notes for storage transitio ## [Unreleased] +### Fixed + +- Admin job lists and kind counts work on a queue-storage schema installed with more claim slots than the default, instead of failing with `invalid physical claim slot`. + ## [0.6.14] — 2026-11-08 Recommended for everyone on 0.6.13 using queue storage. A busy queue no longer stalls promotion of the others, prune backs off cleanly under lock contention, new queue lanes are usable by every runtime role, and jobs no longer stay running, run again or get misreported after priority aging. diff --git a/awa-model/src/admin.rs b/awa-model/src/admin.rs index 296e533de..f68c9228e 100644 --- a/awa-model/src/admin.rs +++ b/awa-model/src/admin.rs @@ -4067,7 +4067,7 @@ async fn queue_storage_base_kind_counts( Ok(counts) } -// The result is bounded by the configured claim-ring slot count, not by +// The result is bounded by the installed claim-ring slot count, not by // retained job volume. Include all populated slots, including old generations. async fn queue_storage_populated_claim_slots( tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, @@ -4089,14 +4089,11 @@ async fn queue_storage_populated_claim_slots( sqlx::query("RELEASE SAVEPOINT awa_claim_slots") .execute(tx.as_mut()) .await?; + // A schema-only admin store carries the default slot count, so take the + // installed slot numbers as they are rather than bounding them by it. slots .into_iter() - .map(|slot| { - usize::try_from(slot) - .ok() - .filter(|slot| *slot < store.claim_slot_count()) - .ok_or_else(|| AwaError::Validation(format!("invalid physical claim slot {slot}"))) - }) + .map(|slot| crate::queue_storage::ring_slot_number(slot, "claim")) .collect() } diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 457e7fb96..90a774b0b 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -673,7 +673,7 @@ fn claim_closure_batch_child_name(schema: &str, slot: usize) -> String { format!("{schema}.lease_claim_closure_batches_{slot}") } -fn ring_slot_number(slot: i32, ring: &str) -> Result { +pub(crate) fn ring_slot_number(slot: i32, ring: &str) -> Result { usize::try_from(slot).map_err(|_| { AwaError::Validation(format!( "invalid {ring} ring slot {slot}: slot must be non-negative" diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index 41a574599..53998d25b 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -17552,6 +17552,67 @@ async fn test_receipt_admin_scope_materialized_control() { receipt_admin_reads_preserve_scope_and_disposition(false, false).await; } +#[tokio::test] +async fn test_schema_only_admin_reads_claims_in_installed_slot() { + let (_db_guard, pool) = setup_pool(6).await; + let schema = "awa_qs_admin_wide_claims"; + let queue = "admin_wide_claims"; + let default_claim_slots = QueueStorageConfig::default().claim_slot_count; + let store = create_store_with_config( + &pool, + QueueStorageConfig { + claim_slot_count: default_claim_slots + 2, + ..unique_cancel_store_config(schema) + }, + ) + .await; + for _ in 0..default_claim_slots { + assert!(matches!( + store.rotate_claims(&pool).await.unwrap(), + RotateOutcome::Rotated { .. } + )); + } + let job_id = enqueue_job( + &pool, + &store, + &CompleteJob { id: 1 }, + InsertOpts { + queue: queue.into(), + ..Default::default() + }, + ) + .await; + let claims = store + .claim_runtime_batch(&pool, queue, 1, Duration::ZERO) + .await + .unwrap(); + assert_eq!(claims.len(), 1); + assert_eq!(claims[0].claim.claim_slot, default_claim_slots as i32); + + let running = admin::list_jobs( + &pool, + &admin::ListJobsFilter { + state: Some(JobState::Running), + queue: Some(queue.into()), + ..Default::default() + }, + ) + .await + .expect("list running jobs through a schema-only store"); + assert_eq!( + running.iter().map(|job| job.id).collect::>(), + vec![job_id] + ); + let kinds = admin::job_kind_overviews(&pool) + .await + .expect("count kinds through a schema-only store"); + let complete = kinds + .iter() + .find(|kind| kind.kind == CompleteJob::kind()) + .unwrap(); + assert_eq!((complete.job_count, complete.queue_count), (1, 1)); +} + async fn claim_aged_rescue_body_job( schema: &str, receipts: bool,