From 1b2437bb8459fdba2ad1da473cb0f35811df46ea Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Sat, 7 Nov 2026 02:58:57 +1300 Subject: [PATCH 1/5] Keep compact receipt closure membership correlated Prevent the compact-member range membership check from becoming a broad anti-join over retained closure batches. Preserve every identity and closure predicate and the existing result semantics. --- CHANGELOG.md | 2 ++ awa-model/src/queue_storage.rs | 3 +++ 2 files changed, 5 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 32d39a6dd..d0488b5df 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,8 @@ Notable changes between releases. Detailed migration notes for storage transitio - Close read connections if their transaction start is cancelled or fails before SQLx acknowledges it, so later reads cannot inherit an untracked transaction. +- Reduce running-count query work for partially closed compact receipt batches + by keeping closure membership lookups correlated. - **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/queue_storage.rs b/awa-model/src/queue_storage.rs index 751081d37..b715b3428 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -749,6 +749,9 @@ pub(crate) fn open_receipt_running_claims_sql_for(schema: &str, job_ids: Option< SELECT 1 FROM {schema}.lease_claim_closure_batches AS closure_batches WHERE closure_batches.claim_slot = batches.claim_slot AND closure_batches.receipt_ranges @> items.receipt_id + -- Keep the range lookup correlated: joining on claim_slot can + -- multiply receipt members by retained closure batches. + OFFSET 0 ) AND NOT EXISTS ( SELECT 1 FROM {schema}.leases AS lease From 42b7197a1f6acb203ada553149aa0211c5d6ed8f Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Sat, 7 Nov 2026 03:44:52 +1300 Subject: [PATCH 2/5] Prune unrelated receipt closure partitions Closure writers preserve the original claim slot. Match that slot in row-local receipt closure checks while retaining global receipt identity. --- CHANGELOG.md | 2 ++ awa-model/src/queue_storage.rs | 11 ++++++----- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d0488b5df..858013022 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,8 @@ Notable changes between releases. Detailed migration notes for storage transitio SQLx acknowledges it, so later reads cannot inherit an untracked transaction. - Reduce running-count query work for partially closed compact receipt batches by keeping closure membership lookups correlated. +- Reduce running-count query work by keeping compact closure membership + correlated and limiting row-local closure lookups to their claim partition. - **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/queue_storage.rs b/awa-model/src/queue_storage.rs index b715b3428..d84af2860 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -629,10 +629,10 @@ fn receipt_closed_evidence_sql( closure_batch_rel: &str, claims_alias: &str, ) -> String { - // `receipt_id` is allocated from a global sequence, so it identifies a - // claim without a `claim_slot` predicate. Compact closure batches retain - // the raw receipt array for audit/debugging, but hot membership checks use - // the per-child GiST index on the derived multirange. + // `receipt_id` is allocated from a global sequence and identifies a claim. + // Closures retain that claim's slot; match it to prune unrelated partitions. + // Compact closure batches retain the raw receipt array for audit/debugging, + // but hot membership checks use the per-child GiST index on the multirange. format!( r#" ( @@ -646,7 +646,8 @@ fn receipt_closed_evidence_sql( OR EXISTS ( SELECT 1 FROM {closure_batch_rel} AS closure_batches - WHERE closure_batches.receipt_ranges @> {claims_alias}.receipt_id + WHERE closure_batches.claim_slot = {claims_alias}.claim_slot + AND closure_batches.receipt_ranges @> {claims_alias}.receipt_id ) OR EXISTS ( SELECT 1 FROM {schema}.done_entries AS done From d88980676faf42fd09202d77b4721642c8718bc0 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Sat, 7 Nov 2026 05:52:27 +1300 Subject: [PATCH 3/5] perf: parameterize completed receipt kind lookups --- CHANGELOG.md | 2 ++ awa-model/src/admin.rs | 22 ++++++++++++++-------- awa/tests/queue_storage_runtime_test.rs | 11 +++++++++++ 3 files changed, 27 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 858013022..2bd89e2a9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,6 +36,8 @@ Notable changes between releases. Detailed migration notes for storage transitio by keeping closure membership lookups correlated. - Reduce running-count query work by keeping compact closure membership correlated and limiting row-local closure lookups to their claim partition. +- Hydrate completed job-kind counts through each member's full ready-row key, + avoiding repeated scans of other retained rows in the same enqueue lane. - **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 8984a85f0..a0fa1aec0 100644 --- a/awa-model/src/admin.rs +++ b/awa-model/src/admin.rs @@ -889,14 +889,20 @@ fn queue_storage_kind_counts_cte(schema: &str) -> String { batch.run_leases, batch.lane_seqs ) AS item(job_id, run_lease, lane_seq) - JOIN {schema}.ready_entries AS ready - ON ready.ready_slot = batch.ready_slot - AND ready.ready_generation = batch.ready_generation - AND ready.queue = batch.queue - AND ready.priority = batch.priority - AND ready.enqueue_shard = batch.enqueue_shard - AND ready.lane_seq = item.lane_seq - AND ready.job_id = item.job_id + CROSS JOIN LATERAL ( + SELECT ready.kind + FROM {schema}.ready_entries AS ready + WHERE ready.ready_slot = batch.ready_slot + AND ready.ready_generation = batch.ready_generation + AND ready.queue = batch.queue + AND ready.priority = batch.priority + AND ready.enqueue_shard = batch.enqueue_shard + AND ready.lane_seq = item.lane_seq + AND ready.job_id = item.job_id + -- Keep the full member key in the parameterized lookup, + -- rather than scanning its retained lane for every member. + OFFSET 0 + ) AS ready WHERE NOT EXISTS ( SELECT 1 FROM {schema}.receipt_completion_tombstones AS tomb WHERE tomb.ready_slot = batch.ready_slot diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index e80c74c9e..caba171a1 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -14618,6 +14618,17 @@ async fn test_queue_storage_queue_overview_counts_without_terminal_history() { assert_eq!(overview.total_queued, 2); assert_eq!(overview.running, 0); + let kind = admin::job_kind_overview(&pool, "complete_job") + .await + .expect("read kind overview after reusing the retained lane") + .expect("the completed and newly enqueued jobs have a kind"); + assert_eq!( + kind.job_count, + (completed.len() + waiting.len() - 1) as i64, + "retained completions and later jobs in the same lane count once" + ); + assert_eq!(kind.completed_last_hour, (completed.len() - 1) as i64); + // The dashboard's other two reads, over the same jobs. let counts = admin::state_counts(&pool) .await From b3a850f8feeca9a9ad6b99a9056f217c8c3b09c0 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Sat, 7 Nov 2026 11:15:35 +1300 Subject: [PATCH 4/5] fix: include open receipt jobs in admin reads without prelocks --- CHANGELOG.md | 4 + awa-model/src/admin.rs | 247 +++++++++++++ awa-model/src/queue_storage.rs | 46 +++ awa/tests/admin_read_equivalence_test.rs | 22 +- awa/tests/queue_storage_runtime_test.rs | 426 +++++++++++++++++++++++ docs/troubleshooting.md | 8 + 6 files changed, 734 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 80fda9ffd..60038116d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,10 @@ 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. Point reads retain the [concurrent-prune limitation](docs/troubleshooting.md#transient-admin-reads-during-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..8737f364d 100644 --- a/awa-model/src/admin.rs +++ b/awa-model/src/admin.rs @@ -381,6 +381,156 @@ 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 + -- 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 @@ -734,6 +884,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 @@ -820,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 ( @@ -856,6 +1016,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 ( @@ -863,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, @@ -916,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 @@ -1090,12 +1284,56 @@ 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} )"# )); } + 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!( @@ -1552,6 +1790,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..83ef62966 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -11979,6 +11979,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 +12048,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 +12133,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 +12595,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 +12803,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/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 diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index e80c74c9e..21a59f0be 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -16848,3 +16848,429 @@ 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)); + + // 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"); + + 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; +} + +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; +} diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index bfa095cd5..9f60299f5 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -85,6 +85,14 @@ Interpretation: - `heartbeat_age_s` well above `heartbeat_staleness` (default `90s`): stale-heartbeat rescue should pick it up on the next scan - very negative `deadline_remaining_s`: deadline rescue should pick it up soon (lease side or receipt side, depending on which table the row lives in) +### Transient Admin Reads During Partition Pruning + +In 0.6, a job-detail read, including list-item hydration, uses several queries. +If prune truncates a partition during the read, a completed job can briefly +appear available or missing. Retry with a fresh read before treating that result +as missing durable work. Filtering tombstoned source rows after a queue or +priority move does not close this separate window. + ### Inspect Runtime Health ```sql From 7be41b513b69e60eeba87b6b58ade996b6a0ce71 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Sat, 7 Nov 2026 07:37:50 +1300 Subject: [PATCH 5/5] perf: avoid repeated terminal probes for open receipts --- CHANGELOG.md | 5 +- awa-model/src/queue_storage.rs | 69 ++++++++++++---- awa/tests/queue_storage_runtime_test.rs | 103 ++++++++++++++++++++++++ 3 files changed, 159 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2bd89e2a9..676f4d347 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,8 +32,9 @@ Notable changes between releases. Detailed migration notes for storage transitio - Close read connections if their transaction start is cancelled or fails before SQLx acknowledges it, so later reads cannot inherit an untracked transaction. -- Reduce running-count query work for partially closed compact receipt batches - by keeping closure membership lookups correlated. + +- Avoid repeated terminal-ledger probes when counting open deadline-backed + receipts, while preserving all closure and materialization exclusions. - Reduce running-count query work by keeping compact closure membership correlated and limiting row-local closure lookups to their claim partition. - Hydrate completed job-kind counts through each member's full ready-row key, diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index d84af2860..048d50c4d 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -623,12 +623,45 @@ fn ring_slot_index(slot: i32, slot_count: usize, ring: &str) -> Result String { + receipt_evidence_sql( + schema, + closure_rel, + closure_batch_rel, + claims_alias, + ReceiptEvidence::Closed, + ) +} + +fn receipt_evidence_sql( + schema: &str, + closure_rel: &str, + closure_batch_rel: &str, + claims_alias: &str, + evidence: ReceiptEvidence, +) -> String { + let (null_test, join, exists, range_fence) = match evidence { + ReceiptEvidence::Closed => ("NOT NULL", "OR", "EXISTS", ""), + ReceiptEvidence::Open => ( + "NULL", + "AND", + "NOT EXISTS", + " + -- Keep receipt membership correlated while the other closure + -- exclusions can use full-identity anti joins. + OFFSET 0", + ), + }; // `receipt_id` is allocated from a global sequence and identifies a claim. // Closures retain that claim's slot; match it to prune unrelated partitions. // Compact closure batches retain the raw receipt array for audit/debugging, @@ -636,30 +669,30 @@ fn receipt_closed_evidence_sql( format!( r#" ( - {claims_alias}.closed_at IS NOT NULL - OR EXISTS ( + {claims_alias}.closed_at IS {null_test} + {join} {exists} ( SELECT 1 FROM {closure_rel} AS closures WHERE closures.claim_slot = {claims_alias}.claim_slot AND closures.job_id = {claims_alias}.job_id AND closures.run_lease = {claims_alias}.run_lease ) - OR EXISTS ( + {join} {exists} ( SELECT 1 FROM {closure_batch_rel} AS closure_batches WHERE closure_batches.claim_slot = {claims_alias}.claim_slot - AND closure_batches.receipt_ranges @> {claims_alias}.receipt_id + AND closure_batches.receipt_ranges @> {claims_alias}.receipt_id{range_fence} ) - OR EXISTS ( + {join} {exists} ( SELECT 1 FROM {schema}.done_entries AS done WHERE done.job_id = {claims_alias}.job_id AND done.run_lease = {claims_alias}.run_lease ) - OR EXISTS ( + {join} {exists} ( SELECT 1 FROM {schema}.deferred_jobs AS deferred WHERE deferred.job_id = {claims_alias}.job_id AND deferred.run_lease = {claims_alias}.run_lease ) - OR EXISTS ( + {join} {exists} ( SELECT 1 FROM {schema}.dlq_entries AS dlq WHERE dlq.job_id = {claims_alias}.job_id AND dlq.run_lease = {claims_alias}.run_lease @@ -671,12 +704,11 @@ fn receipt_closed_evidence_sql( /// A subquery that yields the identity `(job_id, run_lease, queue)` of every /// receipt claim currently open (i.e. `running`) but not yet materialised -/// into `leases`. Since #246 receipt claims — zero-deadline and -/// deadline-backed alike — are written as compact `lease_claim_batches` rows -/// rather than one `lease_claims` row per job, so the "open receipt claims" -/// set is the union of legacy row-local `lease_claims` and the unnested -/// members of `lease_claim_batches`, each anti-joined against every durable -/// closure/terminal/materialised-lease evidence shape. +/// into `leases`. Zero-deadline receipts use compact `lease_claim_batches`; +/// deadline-backed receipts retain row-local `lease_claims` for indexed +/// deadline rescue. The open set unions those row-local claims with the +/// unnested compact members, excluding every durable closure, terminal, and +/// materialised-lease evidence shape. /// /// This is the canonical open-receipt-claim shape (mirrors `queue_counts`'s /// `live_running` derivation). Compose it wherever a running enumeration must @@ -702,8 +734,13 @@ pub(crate) fn open_receipt_running_claims_sql(schema: &str) -> String { pub(crate) fn open_receipt_running_claims_sql_for(schema: &str, job_ids: Option<&str>) -> String { let closure_rel = format!("{schema}.lease_claim_closures"); let closure_batch_rel = format!("{schema}.lease_claim_closure_batches"); - let row_closed = - receipt_closed_evidence_sql(schema, &closure_rel, &closure_batch_rel, "claims"); + let row_open = receipt_evidence_sql( + schema, + &closure_rel, + &closure_batch_rel, + "claims", + ReceiptEvidence::Open, + ); let (row_filter, batch_filter) = match job_ids { Some(ids) => ( format!("claims.job_id = ANY({ids}) AND "), @@ -715,7 +752,7 @@ pub(crate) fn open_receipt_running_claims_sql_for(schema: &str, job_ids: Option< r#" SELECT claims.job_id, claims.run_lease, claims.queue FROM {schema}.lease_claims AS claims - WHERE {row_filter}NOT {row_closed} + WHERE {row_filter}{row_open} AND NOT EXISTS ( SELECT 1 FROM {schema}.leases AS lease WHERE lease.job_id = claims.job_id diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index caba171a1..fec229ef3 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -8840,6 +8840,109 @@ async fn test_queue_storage_admin_queries_cover_running_and_failed_rows() { client.shutdown(Duration::from_secs(5)).await; } +#[tokio::test] +async fn test_row_local_receipt_admin_counts_follow_durable_dispositions() { + let (_db_guard, pool) = setup_pool(6).await; + let queue = "qs_row_receipt_dispositions"; + let schema = "awa_qs_row_receipt_dispositions"; + let store = create_store_with_config( + &pool, + QueueStorageConfig { + schema: schema.to_string(), + lease_claim_receipts: true, + ..Default::default() + }, + ) + .await; + + async fn counts(pool: &sqlx::PgPool, queue: &str) -> (i64, i64, i64) { + let overview = admin::queue_overview(pool, queue) + .await + .expect("read receipt disposition counts") + .expect("receipt queue exists"); + ( + overview.running, + overview.retryable, + overview.completed_last_hour, + ) + } + + store.enqueue_batch(&pool, queue, 2, 3).await.unwrap(); + let claimed = store + .claim_runtime_batch(&pool, queue, 3, Duration::from_secs(300)) + .await + .expect("claim three deadline-backed receipts"); + assert_eq!(claimed.len(), 3); + assert_eq!(lease_claim_count(&pool, &store).await, 3); + assert_eq!(lease_count(&pool, &store).await, 0); + assert_eq!(counts(&pool, queue).await, (3, 0, 0)); + + let completed = store + .complete_runtime_batch(&pool, &claimed[..1]) + .await + .unwrap(); + assert_eq!( + completed, + vec![(claimed[0].job.id, claimed[0].job.run_lease)] + ); + let null_fences: i64 = sqlx::query_scalar(&format!("SELECT count(*) FROM {schema}.lease_claims WHERE job_id = $1 AND run_lease = $2 AND closed_at IS NULL")) + .bind(claimed[0].job.id) + .bind(claimed[0].job.run_lease) + .fetch_one(&pool) + .await + .expect("completed row-local claim remains behind compact closure evidence"); + assert_eq!(null_fences, 1); + assert_eq!(lease_claim_closure_batch_count(&pool, &store).await, 1); + assert_eq!(counts(&pool, queue).await, (2, 0, 1)); + + let retried = store + .fail_retryable( + &pool, + claimed[1].job.id, + claimed[1].job.run_lease, + "retry", + None, + ) + .await + .unwrap() + .expect("second attempt moves to retryable"); + assert_eq!(retried.state, JobState::Retryable); + assert_eq!(counts(&pool, queue).await, (1, 1, 1)); + + let callback = store + .register_callback( + &pool, + claimed[2].job.id, + claimed[2].job.run_lease, + Duration::from_secs(300), + ) + .await + .unwrap(); + assert_eq!(lease_count(&pool, &store).await, 1); + assert_eq!( + counts(&pool, queue).await, + (1, 1, 1), + "materialization counts the attempt once" + ); + let finished = store + .complete_external(&pool, callback, None, Some(claimed[2].job.run_lease), false) + .await + .unwrap(); + assert_eq!(finished.state, JobState::Completed); + assert_eq!(counts(&pool, queue).await, (0, 1, 2)); + for (claim, expected) in claimed.iter().zip([ + JobState::Completed, + JobState::Retryable, + JobState::Completed, + ]) { + let job = store.load_job(&pool, claim.job.id).await.unwrap().unwrap(); + assert_eq!( + job.state, expected, + "point reads and aggregate counts agree" + ); + } +} + /// With lease-claim receipts on, a running job is an open receipt claim and /// `leases` stays empty, so every running count must read the claims too. #[tokio::test(flavor = "multi_thread", worker_threads = 4)]