From e34211ab77e5aeed9464ed24132ad83eb6656b4b Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 1 Nov 2026 22:42:19 +0000 Subject: [PATCH 1/3] Read queue overviews from live work on queue storage admin::queue_overviews counted every current job through terminal_jobs, which expands each completion batch and joins it back to ready_entries, so its cost grew with retained history: 22 s in one production deployment, polled by an open admin dashboard. queue_state_counts is no answer: the maintenance leader stops maintaining it once queue storage is active. The workers' queue gauges already read only live-work tables. That query moves into awa-model as queue_signals_sql and serves both. The admin variant also counts receipt-backed running claims, counts available rows exactly along each lane's primary key, and takes completions in the last hour from completion-batch sizes and done_entries. With 100k retained completions and 100k queued jobs, the overview drops from 7.7 s to 0.27 s; JIT is off for it, since compiling the statement cost 1.2 s. --- CHANGELOG.md | 11 ++ awa-model/src/admin.rs | 83 ++++---- awa-model/src/queue_storage.rs | 247 ++++++++++++++++++++++++ awa-worker/src/maintenance.rs | 154 +-------------- awa/tests/queue_storage_runtime_test.rs | 98 ++++++++++ 5 files changed, 396 insertions(+), 197 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 54759f505..a9819d5a1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,17 @@ Notable changes between releases. Detailed migration notes for storage transitio ## [Unreleased] +### Fixed + +- **Queue overviews on queue storage no longer read retained history.** + `admin::queue_overviews`, behind the admin UI's queues and dashboard pages + and `awa queue stats`, counted every current job through `terminal_jobs`, + which expands each completion batch and joins it back to `ready_entries`. + It now reads the live-work tables, as the workers' queue gauges do, and + counts completions in the last hour from batch sizes. With 100,000 + retained completions and 100,000 queued jobs it drops from 7.7 s to + 0.27 s. JIT is off for the read; compiling it cost more than running it. + ## [0.7.0-alpha.1] — 2026-10-29 ### Upgrading diff --git a/awa-model/src/admin.rs b/awa-model/src/admin.rs index daab789a9..fd9b1969d 100644 --- a/awa-model/src/admin.rs +++ b/awa-model/src/admin.rs @@ -2288,9 +2288,13 @@ where /// Get queue overviews for all known queues. /// -/// Hybrid read: per-state counts come from the `queue_state_counts` -/// cache table (eventually consistent, ~2s lag), while `lag_seconds` -/// and `completed_last_hour` are computed live from `jobs_hot`. +/// On queue storage, counts come from the live-work tables through +/// [`crate::queue_storage::queue_signals_sql`]. +/// +/// On the canonical engine it is a hybrid read: per-state counts come from +/// the `queue_state_counts` cache table (eventually consistent, ~2s lag), +/// while `lag_seconds` and `completed_last_hour` are computed live from +/// `jobs_hot`. /// /// The cache is kept fresh by the maintenance leader's dirty-key /// recompute (~2s) and full reconciliation (~60s). Also warmed during @@ -2300,11 +2304,14 @@ where /// leader, call `flush_dirty_admin_metadata()` first. pub async fn queue_overviews(pool: &PgPool) -> Result, AwaError> { if let Some(store) = active_queue_storage(pool).await? { + // `queue_state_counts` isn't maintained on queue storage, and the + // `current_jobs` view expands every retained completion, so counts + // come from the live-work tables instead. let sql = format!( r#" - {current_jobs_cte}, + {signals_cte}, all_queues AS ( - SELECT queue FROM current_jobs + SELECT queue FROM signals UNION SELECT queue FROM awa.queue_descriptors UNION @@ -2324,35 +2331,6 @@ pub async fn queue_overviews(pool: &PgPool) -> Result, AwaErr secs => GREATEST(((GREATEST(runtime.snapshot_interval_ms, 1000) / 1000) * 3)::int, 30) ) >= now() GROUP BY descriptor.key - ), - queue_counts AS ( - SELECT - queue, - count(*) FILTER (WHERE state = 'scheduled')::bigint AS scheduled, - count(*) FILTER (WHERE state = 'available')::bigint AS available, - count(*) FILTER (WHERE state = 'retryable')::bigint AS retryable, - count(*) FILTER (WHERE state = 'running')::bigint AS running, - count(*) FILTER (WHERE state = 'failed')::bigint AS failed, - count(*) FILTER (WHERE state = 'waiting_external')::bigint AS waiting_external - FROM current_jobs - GROUP BY queue - ), - available_lag AS ( - SELECT - queue, - EXTRACT(EPOCH FROM (clock_timestamp() - min(run_at)))::float8 AS lag_seconds - FROM current_jobs - WHERE state = 'available' - GROUP BY queue - ), - completed_recent AS ( - SELECT - queue, - count(*)::bigint AS completed_last_hour - FROM current_jobs - WHERE state = 'completed' - AND finalized_at > clock_timestamp() - interval '1 hour' - GROUP BY queue ) SELECT q.queue, @@ -2370,31 +2348,40 @@ pub async fn queue_overviews(pool: &PgPool) -> Result, AwaErr ) < now() END AS descriptor_stale, COALESCE(qdv.descriptor_variant_count, 0) > 1 AS descriptor_mismatch, - COALESCE(qc.scheduled + qc.available + qc.running + qc.retryable + qc.waiting_external, 0) AS total_queued, - COALESCE(qc.scheduled, 0) AS scheduled, - COALESCE(qc.available, 0) AS available, - COALESCE(qc.retryable, 0) AS retryable, - COALESCE(qc.running, 0) AS running, - COALESCE(qc.failed, 0) AS failed, - COALESCE(qc.waiting_external, 0) AS waiting_external, - COALESCE(cr.completed_last_hour, 0) AS completed_last_hour, - al.lag_seconds, + COALESCE(sg.scheduled + sg.available + sg.running + sg.retryable + sg.waiting_external, 0) + AS total_queued, + COALESCE(sg.scheduled, 0) AS scheduled, + COALESCE(sg.available, 0) AS available, + COALESCE(sg.retryable, 0) AS retryable, + COALESCE(sg.running, 0) AS running, + COALESCE(sg.failed_done + sg.failed_dlq, 0) AS failed, + COALESCE(sg.waiting_external, 0) AS waiting_external, + COALESCE(sg.completed_last_hour, 0) AS completed_last_hour, + sg.lag_seconds, COALESCE(qm.paused, FALSE) AS paused FROM all_queues q - LEFT JOIN queue_counts qc ON qc.queue = q.queue + LEFT JOIN signals sg ON sg.queue = q.queue LEFT JOIN awa.queue_descriptors qd ON qd.queue = q.queue LEFT JOIN live_queue_descriptor_variants qdv ON qdv.queue = q.queue - LEFT JOIN available_lag al ON al.queue = q.queue - LEFT JOIN completed_recent cr ON cr.queue = q.queue LEFT JOIN awa.queue_meta qm ON qm.queue = q.queue ORDER BY q.queue "#, - current_jobs_cte = queue_storage_current_jobs_cte(store.schema()) + signals_cte = crate::queue_storage::queue_signals_sql( + store.schema(), + crate::queue_storage::QueueSignalReads::Admin, + ) ); + // JIT compiles this statement for longer than it runs (about 1.2 s + // against 0.35 s with 100k retained completions). + let mut tx = pool.begin().await?; + sqlx::query("SET LOCAL jit = off") + .execute(tx.as_mut()) + .await?; let rows = sqlx::query_as::<_, QueueOverview>(audited_sql(sql)) - .fetch_all(pool) + .fetch_all(tx.as_mut()) .await?; + tx.commit().await?; return Ok(rows); } diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index aac2ec1ea..27b8120b9 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -892,6 +892,253 @@ fn receipt_closed_evidence_sql( ) } +/// What [`queue_signals_sql`] reads. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum QueueSignalReads { + /// No receipt-claim table, so claim-ring rotation's `ACCESS EXCLUSIVE` + /// locks never block it: `running` counts materialised leases only, and + /// `completed_last_hour` is 0. For gauges polled by the maintenance loop. + Gauge, + /// Counts available rows exactly, and also counts receipt-backed running + /// claims and completions in the last hour from completion batches and + /// `done_entries`. For the admin queue views. + Admin, +} + +/// Per-queue counts that never expand terminal history, ending in a +/// `signals` CTE with `queue`, `available`, `running`, `waiting_external`, +/// `scheduled`, `retryable`, `failed_done`, `failed_dlq`, `lag_seconds` and +/// `completed_last_hour`. +/// +/// With [`QueueSignalReads::Gauge`], `available` is the sequence-head signal, +/// an upper bound as in [`QueueStorage::queue_counts_fast`]; with +/// [`QueueSignalReads::Admin`] it is exact. Nothing here reads `terminal_jobs`, +/// whose completion batches expand and join back to `ready_entries`, so the +/// cost follows live work rather than retained history. +pub fn queue_signals_sql(schema: &str, reads: QueueSignalReads) -> String { + let (ready, receipt_running, completed_recent) = match reads { + QueueSignalReads::Gauge => ( + // Enqueued minus claimed: an upper bound, since a tombstoned + // row or a sequence gap still counts until the claimer passes. + r#" + SELECT + head_signal.queue, + COALESCE( + sum(GREATEST(head_signal.next_seq - head_signal.claim_seq, 0)), + 0 + )::bigint AS available + FROM head_signal + GROUP BY head_signal.queue + "# + .to_string(), + "SELECT NULL::text AS queue, 0::bigint AS running WHERE false".to_string(), + "SELECT NULL::text AS queue, 0::bigint AS completed_last_hour WHERE false".to_string(), + ), + QueueSignalReads::Admin => ( + // Exact: each lane's unclaimed rows, read from its claim cursor + // up to its enqueue head along the primary key, so the cost + // follows the backlog rather than the partition. + format!( + r#" + SELECT head_signal.queue, COALESCE(sum(lane.available), 0)::bigint AS available + FROM head_signal + CROSS JOIN LATERAL ( + SELECT count(*)::bigint AS available + FROM {schema}.ready_entries AS ready + WHERE ready.queue = head_signal.queue + AND ready.priority = head_signal.priority + AND ready.enqueue_shard = head_signal.enqueue_shard + AND ready.lane_seq >= head_signal.claim_seq + AND ready.lane_seq < head_signal.next_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 + ) + ) AS lane + GROUP BY head_signal.queue + "# + ), + format!( + "SELECT queue, count(*)::bigint AS running FROM ({}) AS open_claims GROUP BY queue", + open_receipt_running_claims_sql(schema) + ), + format!( + r#" + SELECT queue, sum(completed)::bigint AS completed_last_hour + FROM ( + -- A batch carries its queue and size, so it is counted + -- without unnesting its members. + SELECT queue, completed_count::bigint AS completed + FROM {schema}.receipt_completion_batches + WHERE finalized_at > clock_timestamp() - interval '1 hour' + UNION ALL + SELECT queue, 1::bigint + FROM {schema}.done_entries + WHERE state = 'completed' + AND finalized_at > clock_timestamp() - interval '1 hour' + ) AS recent + GROUP BY queue + "# + ), + ), + }; + format!( + r#" + WITH head_signal AS ( + SELECT + enqueues.queue, + enqueues.priority, + enqueues.enqueue_shard, + {schema}.sequence_next_value(enqueues.seq_name) AS next_seq, + {schema}.sequence_next_value(claims.seq_name) AS claim_seq + FROM {schema}.queue_enqueue_heads AS enqueues + JOIN {schema}.queue_claim_heads AS claims + ON claims.queue = enqueues.queue + AND claims.priority = enqueues.priority + AND claims.enqueue_shard = enqueues.enqueue_shard + ), + queues AS ( + SELECT DISTINCT queue + FROM ( + SELECT queue FROM awa.queue_meta + UNION ALL + SELECT queue FROM head_signal + UNION ALL + SELECT queue FROM {schema}.leases + UNION ALL + SELECT queue FROM {schema}.deferred_jobs + UNION ALL + SELECT queue FROM {schema}.queue_terminal_live_counts + UNION ALL + SELECT queue FROM {schema}.queue_terminal_rollups + UNION ALL + SELECT queue FROM {schema}.dlq_entries + ) queues + ), + ready AS ( + {ready} + ), + lag AS ( + SELECT + head_signal.queue, + EXTRACT(EPOCH FROM clock_timestamp() - min(next_ready.run_at))::double precision + AS lag_seconds + FROM head_signal + JOIN LATERAL ( + SELECT next_in_segment.run_at + FROM {schema}.ready_segments AS segment + CROSS JOIN LATERAL ( + SELECT ready.run_at + FROM {schema}.ready_entries AS ready + WHERE ready.ready_slot = segment.ready_slot + AND ready.ready_generation = segment.ready_generation + AND ready.queue = segment.queue + AND ready.priority = segment.priority + AND ready.enqueue_shard = segment.enqueue_shard + AND ready.lane_seq >= GREATEST(segment.first_lane_seq, head_signal.claim_seq) + AND ready.lane_seq < segment.next_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 + ) + ORDER BY ready.lane_seq + LIMIT 1 + ) AS next_in_segment + WHERE segment.queue = head_signal.queue + AND segment.priority = head_signal.priority + AND segment.enqueue_shard = head_signal.enqueue_shard + AND segment.next_lane_seq > head_signal.claim_seq + ORDER BY segment.next_lane_seq + LIMIT 1 + ) AS next_ready ON TRUE + GROUP BY head_signal.queue + ), + leases AS ( + -- Both legs are `leases`-only because this gauge must not touch + -- the receipt tables (see the function-level lock note). The + -- `running` leg is therefore an approximation (misses compact + -- lease_claim_batches claims, #246). `waiting_external` is exact + -- even so: a receipt claim cannot be `waiting_external` off the + -- `leases` plane — QueueStorage::register_callback runs + -- ensure_mutable_running_attempt_tx (materialises the claim into + -- `leases`) before enter_callback_wait_in_tx flips + -- `leases.state`. (Same waiting_external invariant, exact form, + -- at admin::state_counts.) + SELECT + queue, + count(*) FILTER (WHERE state = 'running')::bigint AS running, + count(*) FILTER (WHERE state = 'waiting_external')::bigint + AS waiting_external + FROM {schema}.leases + GROUP BY queue + ), + deferred AS ( + SELECT + queue, + count(*) FILTER (WHERE state = 'scheduled')::bigint AS scheduled, + count(*) FILTER (WHERE state = 'retryable')::bigint AS retryable + FROM {schema}.deferred_jobs + GROUP BY queue + ), + terminal AS ( + SELECT + queue, + count(*)::bigint AS failed_done + FROM {schema}.done_entries + WHERE state = 'failed' + GROUP BY queue + ), + dlq AS ( + SELECT + queue, + count(*)::bigint AS failed_dlq + FROM {schema}.dlq_entries + GROUP BY queue + ), + receipt_running AS ( + {receipt_running} + ), + completed_recent AS ( + {completed_recent} + ), + signals AS ( + SELECT + queues.queue, + COALESCE(ready.available, 0)::bigint AS available, + (COALESCE(leases.running, 0) + COALESCE(receipt_running.running, 0))::bigint + AS running, + COALESCE(leases.waiting_external, 0)::bigint AS waiting_external, + COALESCE(deferred.scheduled, 0)::bigint AS scheduled, + COALESCE(deferred.retryable, 0)::bigint AS retryable, + COALESCE(terminal.failed_done, 0)::bigint AS failed_done, + COALESCE(dlq.failed_dlq, 0)::bigint AS failed_dlq, + lag.lag_seconds, + COALESCE(completed_recent.completed_last_hour, 0)::bigint AS completed_last_hour + FROM queues + LEFT JOIN ready ON ready.queue = queues.queue + LEFT JOIN lag ON lag.queue = queues.queue + LEFT JOIN leases ON leases.queue = queues.queue + LEFT JOIN receipt_running ON receipt_running.queue = queues.queue + LEFT JOIN deferred ON deferred.queue = queues.queue + LEFT JOIN terminal ON terminal.queue = queues.queue + LEFT JOIN dlq ON dlq.queue = queues.queue + LEFT JOIN completed_recent ON completed_recent.queue = queues.queue + ) + "# + ) +} + /// 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 diff --git a/awa-worker/src/maintenance.rs b/awa-worker/src/maintenance.rs index 98ecf9495..78db4f9b0 100644 --- a/awa-worker/src/maintenance.rs +++ b/awa-worker/src/maintenance.rs @@ -3610,156 +3610,12 @@ impl MaintenanceService { // exact running count (which DOES expand lease_claim_batches) go // through QueueStorage::queue_counts_exact() / admin::state_counts(). let rows: Vec = match sqlx::query_as(audited_sql(format!( - r#" - WITH head_signal AS ( - SELECT - enqueues.queue, - enqueues.priority, - enqueues.enqueue_shard, - {schema}.sequence_next_value(enqueues.seq_name) AS next_seq, - {schema}.sequence_next_value(claims.seq_name) AS claim_seq - FROM {schema}.queue_enqueue_heads AS enqueues - JOIN {schema}.queue_claim_heads AS claims - ON claims.queue = enqueues.queue - AND claims.priority = enqueues.priority - AND claims.enqueue_shard = enqueues.enqueue_shard - ), - queues AS ( - SELECT DISTINCT queue - FROM ( - SELECT queue FROM awa.queue_meta - UNION ALL - SELECT queue FROM head_signal - UNION ALL - SELECT queue FROM {schema}.leases - UNION ALL - SELECT queue FROM {schema}.deferred_jobs - UNION ALL - SELECT queue FROM {schema}.queue_terminal_live_counts - UNION ALL - SELECT queue FROM {schema}.queue_terminal_rollups - UNION ALL - SELECT queue FROM {schema}.dlq_entries - ) queues - ), - ready AS ( - SELECT - head_signal.queue, - COALESCE( - sum(GREATEST(head_signal.next_seq - head_signal.claim_seq, 0)), - 0 - )::bigint AS available - FROM head_signal - GROUP BY head_signal.queue - ), - lag AS ( - SELECT - head_signal.queue, - EXTRACT(EPOCH FROM clock_timestamp() - min(next_ready.run_at))::double precision - AS lag_seconds - FROM head_signal - JOIN LATERAL ( - SELECT next_in_segment.run_at - FROM {schema}.ready_segments AS segment - CROSS JOIN LATERAL ( - SELECT ready.run_at - FROM {schema}.ready_entries AS ready - WHERE ready.ready_slot = segment.ready_slot - AND ready.ready_generation = segment.ready_generation - AND ready.queue = segment.queue - AND ready.priority = segment.priority - AND ready.enqueue_shard = segment.enqueue_shard - AND ready.lane_seq >= GREATEST(segment.first_lane_seq, head_signal.claim_seq) - AND ready.lane_seq < segment.next_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 - ) - ORDER BY ready.lane_seq - LIMIT 1 - ) AS next_in_segment - WHERE segment.queue = head_signal.queue - AND segment.priority = head_signal.priority - AND segment.enqueue_shard = head_signal.enqueue_shard - AND segment.next_lane_seq > head_signal.claim_seq - ORDER BY segment.next_lane_seq - LIMIT 1 - ) AS next_ready ON TRUE - GROUP BY head_signal.queue - ), - leases AS ( - -- Both legs are `leases`-only because this gauge must not touch - -- the receipt tables (see the function-level lock note). The - -- `running` leg is therefore an approximation (misses compact - -- lease_claim_batches claims, #246). `waiting_external` is exact - -- even so: a receipt claim cannot be `waiting_external` off the - -- `leases` plane — QueueStorage::register_callback runs - -- ensure_mutable_running_attempt_tx (materialises the claim into - -- `leases`) before enter_callback_wait_in_tx flips - -- `leases.state`. (Same waiting_external invariant, exact form, - -- at admin::state_counts.) - SELECT - queue, - count(*) FILTER (WHERE state = 'running')::bigint AS running, - count(*) FILTER (WHERE state = 'waiting_external')::bigint - AS waiting_external - FROM {schema}.leases - GROUP BY queue - ), - deferred AS ( - SELECT - queue, - count(*) FILTER (WHERE state = 'scheduled')::bigint AS scheduled, - count(*) FILTER (WHERE state = 'retryable')::bigint AS retryable - FROM {schema}.deferred_jobs - GROUP BY queue - ), - terminal AS ( - SELECT - queue, - count(*)::bigint AS failed_done - FROM {schema}.done_entries - WHERE state = 'failed' - GROUP BY queue - ), - dlq AS ( - SELECT - queue, - count(*)::bigint AS failed_dlq - FROM {schema}.dlq_entries - GROUP BY queue + "{} SELECT queue, available, running, waiting_external, scheduled, retryable, \ + failed_done, failed_dlq, lag_seconds FROM signals ORDER BY queue", + awa_model::queue_storage::queue_signals_sql( + schema, + awa_model::queue_storage::QueueSignalReads::Gauge, ) - SELECT - queues.queue, - COALESCE(ready.available, 0)::bigint AS available, - COALESCE(leases.running, 0)::bigint AS running, - COALESCE(leases.waiting_external, 0)::bigint AS waiting_external, - COALESCE(deferred.scheduled, 0)::bigint AS scheduled, - COALESCE(deferred.retryable, 0)::bigint AS retryable, - COALESCE(terminal.failed_done, 0)::bigint AS failed_done, - COALESCE(dlq.failed_dlq, 0)::bigint AS failed_dlq, - lag.lag_seconds - FROM queues - LEFT JOIN ready - ON ready.queue = queues.queue - LEFT JOIN lag - ON lag.queue = queues.queue - LEFT JOIN leases - ON leases.queue = queues.queue - LEFT JOIN deferred - ON deferred.queue = queues.queue - LEFT JOIN terminal - ON terminal.queue = queues.queue - LEFT JOIN dlq - ON dlq.queue = queues.queue - ORDER BY queues.queue - "# ))) .fetch_all(&self.pool) .await diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index 69453d02f..ddb79d0c8 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -17217,6 +17217,104 @@ async fn test_queue_storage_multi_shard_public_available_counts_are_exact() { client.shutdown(Duration::from_secs(1)).await; } +/// `queue_overviews` must not read `queue_state_counts`, which queue storage +/// never maintains, nor expand retained completions: completions in the last +/// hour come from batch sizes, and `available` stays exact past a +/// tombstoned ready row, which the sequence-head signal would still count. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_queue_storage_queue_overview_counts_without_terminal_history() { + let (_db_guard, pool) = setup_pool(8).await; + let queue = "qs_overview_counts"; + let schema = "awa_qs_overview_counts"; + let store_config = QueueStorageConfig { + schema: schema.to_string(), + queue_slot_count: 4, + lease_slot_count: 2, + queue_stripe_count: 1, + lease_claim_receipts: true, + claim_slot_count: 2, + }; + let store = create_store_with_config(&pool, store_config.clone()).await; + // A row left behind by the canonical engine before the cutover. + sqlx::query( + "INSERT INTO awa.queue_state_counts (queue, available, completed) VALUES ($1, 999, 999) \ + ON CONFLICT (queue) DO UPDATE SET available = 999, completed = 999", + ) + .bind(queue) + .execute(&pool) + .await + .expect("seed a stale queue_state_counts row"); + + let mut completed = Vec::new(); + for i in 0..5 { + completed.push( + enqueue_job( + &pool, + &store, + &CompleteJob { id: i }, + InsertOpts { + queue: queue.into(), + ..Default::default() + }, + ) + .await, + ); + } + let client = queue_storage_client(&pool, queue, store_config, CompleteWorker); + client.start().await.expect("client start"); + for job_id in &completed { + wait_for_job_state( + &store, + &pool, + *job_id, + &[JobState::Completed], + Duration::from_secs(10), + ) + .await; + } + client.shutdown(Duration::from_secs(5)).await; + let batched: i64 = sqlx::query_scalar(audited_sql(format!( + "SELECT COALESCE(sum(completed_count), 0)::bigint \ + FROM {schema}.receipt_completion_batches WHERE queue = $1" + ))) + .bind(queue) + .fetch_one(&pool) + .await + .expect("read completion batches"); + assert!( + batched > 0, + "completions should take the compact batch path" + ); + + let mut waiting = Vec::new(); + for i in 5..8 { + waiting.push( + enqueue_job( + &pool, + &store, + &CompleteJob { id: i }, + InsertOpts { + queue: queue.into(), + ..Default::default() + }, + ) + .await, + ); + } + awa::model::admin::cancel(&pool, waiting[0]) + .await + .expect("cancel an available job"); + + let overview = admin::queue_overview(&pool, queue) + .await + .expect("read queue overview") + .expect("queue overview should include the queue"); + assert_eq!(overview.completed_last_hour, 5); + assert_eq!(overview.available, 2, "the cancelled job is not available"); + assert_eq!(overview.total_queued, 2); + assert_eq!(overview.running, 0); +} + /// `ordering_key` pins a job to a deterministic shard based on a /// portable hash of the key bytes. Producers can use it to keep jobs /// for the same logical partition (customer, order, account) on the From 993dc19b0f7d2c65e4688100235eefee878bfeee Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 1 Nov 2026 23:09:23 +0000 Subject: [PATCH 2/3] Bound the dashboard's state counts and timeseries on queue storage The admin dashboard polls state_counts and state_timeseries beside the queue overviews, and on queue storage both read retained history: - state_counts counted completions through terminal_jobs, expanding every completion batch and joining it back to ready_entries, and counted available jobs by scanning every ready row. It now takes completions from batch sizes less their tombstones, and available jobs per lane from its claim cursor (available_by_queue_sql, shared with the queue signals). 6.5 s to 0.24 s with 100k retained completions and 100k queued jobs. - state_timeseries filtered current_jobs by created_at after building it, so it expanded every retained completion and sorted the lot. Against the same data it ran past 10 minutes and filled the disk with temporary files. A recent_jobs CTE now bounds each source first; terminal rows come only from done rows and batches finalised within the window, since a job finishes after it is created. The bound is a bind so Postgres pushes it into the terminal_jobs view's batch scans. 0.6 s with every completion inside the window. JIT is off for both reads. --- CHANGELOG.md | 9 ++ awa-model/src/admin.rs | 146 +++++++++++++++++++----- awa-model/src/queue_storage.rs | 73 +++++++----- awa/tests/queue_storage_runtime_test.rs | 53 ++++++++- 4 files changed, 222 insertions(+), 59 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a9819d5a1..7f9bee4cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,15 @@ Notable changes between releases. Detailed migration notes for storage transitio counts completions in the last hour from batch sizes. With 100,000 retained completions and 100,000 queued jobs it drops from 7.7 s to 0.27 s. JIT is off for the read; compiling it cost more than running it. +- **The dashboard's state counts and timeseries no longer read retained + history on queue storage.** `admin::state_counts` counts completions from + batch sizes and available jobs per lane from its claim cursor (6.5 s to + 0.24 s on the same data). `admin::state_timeseries` reads only rows + finalised within its window, since a job finishes after it is created. It + previously expanded every retained completion and sorted the result: + against that data it ran for over 10 minutes and filled the disk with + temporary files. It now takes 0.6 s, with all 100,000 completions inside + the window. ## [0.7.0-alpha.1] — 2026-10-29 diff --git a/awa-model/src/admin.rs b/awa-model/src/admin.rs index fd9b1969d..44e2f7cee 100644 --- a/awa-model/src/admin.rs +++ b/awa-model/src/admin.rs @@ -511,6 +511,95 @@ fn queue_storage_current_jobs_cte(schema: &str) -> String { ) } +/// Jobs created at or after `since` (a bind such as `$1`), each in its +/// current state, as a `recent_jobs` CTE of `created_at` and `state`. +/// +/// It selects what `current_jobs` would with the same bound, but bounds each +/// source before any join: available rows are read per lane from its claim +/// cursor, and terminal rows only from done rows and completion batches +/// finalised since `since`, as a job finishes after it is created. `since` +/// must be a bind rather than an expression such as `clock_timestamp()`, +/// which Postgres won't push into the `terminal_jobs` view. +fn queue_storage_recent_jobs_cte(schema: &str, since: &str) -> String { + let open_running = crate::queue_storage::open_receipt_running_claims_sql(schema); + format!( + r#" + WITH head_signal AS ( + SELECT + claims.queue, + claims.priority, + claims.enqueue_shard, + {schema}.sequence_next_value(claims.seq_name) AS claim_seq, + {schema}.sequence_next_value(enqueues.seq_name) AS next_seq + FROM {schema}.queue_claim_heads AS claims + JOIN {schema}.queue_enqueue_heads AS enqueues + ON enqueues.queue = claims.queue + AND enqueues.priority = claims.priority + AND enqueues.enqueue_shard = claims.enqueue_shard + ), + recent_jobs AS ( + SELECT ready.created_at, 'available'::awa.job_state AS state + FROM head_signal + CROSS JOIN LATERAL ( + SELECT ready.created_at + FROM {schema}.ready_entries AS ready + WHERE ready.queue = head_signal.queue + AND ready.priority = head_signal.priority + AND ready.enqueue_shard = head_signal.enqueue_shard + AND ready.lane_seq >= head_signal.claim_seq + AND ready.lane_seq < head_signal.next_seq + AND ready.created_at >= {since} + 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 + ) + ) AS ready + UNION ALL + SELECT created_at, state + FROM {schema}.deferred_jobs + WHERE created_at >= {since} + UNION ALL + SELECT ready.created_at, leases.state + FROM {schema}.leases AS leases + JOIN {schema}.ready_entries AS ready + ON ready.ready_slot = leases.ready_slot + AND ready.ready_generation = leases.ready_generation + AND ready.queue = leases.queue + AND ready.priority = leases.priority + AND ready.enqueue_shard = leases.enqueue_shard + AND ready.lane_seq = leases.lane_seq + WHERE ready.created_at >= {since} + UNION ALL + -- A queue or priority move copies the row with its times, so + -- any one of a job's ready rows gives its creation time. + SELECT created.created_at, 'running'::awa.job_state + FROM ({open_running}) AS open_claims + CROSS JOIN LATERAL ( + SELECT ready.created_at + FROM {schema}.ready_entries AS ready + WHERE ready.job_id = open_claims.job_id + LIMIT 1 + ) AS created + WHERE created.created_at >= {since} + UNION ALL + SELECT created_at, state + FROM {schema}.terminal_jobs + WHERE finalized_at >= {since} + AND created_at >= {since} + UNION ALL + SELECT created_at, 'failed'::awa.job_state + FROM {schema}.dlq_entries + WHERE created_at >= {since} + ) + "# + ) +} + async fn notify_cancellation_tx<'a>( tx: &mut sqlx::Transaction<'a, sqlx::Postgres>, job_id: i64, @@ -2825,27 +2914,16 @@ pub async fn state_counts(pool: &PgPool) -> Result, AwaEr r#" SELECT COALESCE((SELECT count(*)::bigint FROM {schema}.deferred_jobs WHERE state = 'scheduled'), 0) AS scheduled, - COALESCE(( - SELECT count(*)::bigint - FROM {schema}.ready_entries AS ready - JOIN {schema}.queue_claim_heads AS claims - ON claims.queue = ready.queue - AND claims.priority = ready.priority - AND claims.enqueue_shard = ready.enqueue_shard - WHERE ready.lane_seq >= {schema}.sequence_next_value(claims.seq_name) - AND NOT EXISTS ( - SELECT 1 FROM {schema}.ready_tombstones AS tomb - WHERE tomb.queue = ready.queue - AND tomb.priority = ready.priority - AND tomb.enqueue_shard = ready.enqueue_shard - AND tomb.lane_seq = ready.lane_seq - AND tomb.ready_slot = ready.ready_slot - AND tomb.ready_generation = ready.ready_generation - ) - ), 0) AS available, + COALESCE((SELECT sum(available)::bigint FROM ({available}) AS lanes), 0) AS available, COALESCE((SELECT count(*)::bigint FROM {schema}.leases WHERE state = 'running'), 0) + COALESCE((SELECT count(*)::bigint FROM ({open_running}) AS open_running), 0) AS running, - COALESCE((SELECT count(*)::bigint FROM {schema}.terminal_jobs WHERE state = 'completed'), 0) AS completed, + -- A completion batch carries its size, so completed jobs are + -- counted without expanding batches through `terminal_jobs`. + -- Tombstones mark batch members deleted or retried since. + COALESCE((SELECT sum(completed_count)::bigint FROM {schema}.receipt_completion_batches), 0) + - COALESCE((SELECT count(*)::bigint FROM {schema}.receipt_completion_tombstones), 0) + + COALESCE((SELECT count(*)::bigint FROM {schema}.done_entries WHERE state = 'completed'), 0) + AS completed, COALESCE((SELECT count(*)::bigint FROM {schema}.deferred_jobs WHERE state = 'retryable'), 0) AS retryable, COALESCE((SELECT count(*)::bigint FROM {schema}.done_entries WHERE state = 'failed'), 0) + COALESCE((SELECT count(*)::bigint FROM {schema}.dlq_entries), 0) AS failed, @@ -2863,6 +2941,7 @@ pub async fn state_counts(pool: &PgPool) -> Result, AwaEr -- publish_queue_storage_health_metrics.) COALESCE((SELECT count(*)::bigint FROM {schema}.leases WHERE state = 'waiting_external'), 0) AS waiting_external "#, + available = crate::queue_storage::available_by_queue_sql(store.schema()), schema = store.schema(), // Compact receipt claims (#246) live in `lease_claim_batches` until // materialised; count them as running alongside the `leases` leg. @@ -2878,8 +2957,18 @@ pub async fn state_counts(pool: &PgPool) -> Result, AwaEr failed, cancelled, waiting_external, - ): (i64, i64, i64, i64, i64, i64, i64, i64) = - sqlx::query_as(audited_sql(sql)).fetch_one(pool).await?; + ): (i64, i64, i64, i64, i64, i64, i64, i64) = { + // As in `queue_overviews`: compiling costs more than running. + let mut tx = pool.begin().await?; + sqlx::query("SET LOCAL jit = off") + .execute(tx.as_mut()) + .await?; + let counts = sqlx::query_as(audited_sql(sql)) + .fetch_one(tx.as_mut()) + .await?; + tx.commit().await?; + counts + }; return Ok(HashMap::from([ (JobState::Scheduled, scheduled), @@ -3271,23 +3360,28 @@ pub async fn state_timeseries( minutes: i32, ) -> Result, AwaError> { if let Some(store) = active_queue_storage(pool).await? { + let since = chrono::Utc::now() - chrono::Duration::minutes(i64::from(minutes)); let sql = format!( "{} \ SELECT \ date_trunc('minute', created_at) AS bucket, \ state, \ count(*) AS count \ - FROM current_jobs \ - WHERE created_at >= clock_timestamp() - make_interval(mins => $1) \ + FROM recent_jobs \ GROUP BY bucket, state \ ORDER BY bucket", - queue_storage_current_jobs_cte(store.schema()) + queue_storage_recent_jobs_cte(store.schema(), "$1") ); + let mut tx = pool.begin().await?; + sqlx::query("SET LOCAL jit = off") + .execute(tx.as_mut()) + .await?; let rows = sqlx::query_as::<_, (chrono::DateTime, JobState, i64)>(audited_sql(sql)) - .bind(minutes) - .fetch_all(pool) + .bind(since) + .fetch_all(tx.as_mut()) .await?; + tx.commit().await?; return Ok(rows .into_iter() diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 27b8120b9..26c81be76 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -892,6 +892,50 @@ fn receipt_closed_evidence_sql( ) } +/// Each queue's available rows, exactly, as `(queue, available)`: every +/// lane's unclaimed rows, read along the primary key from its claim cursor to +/// its enqueue head, so the cost follows the backlog rather than the +/// partition. +pub fn available_by_queue_sql(schema: &str) -> String { + format!( + r#" + SELECT heads.queue, COALESCE(sum(lane.available), 0)::bigint AS available + FROM ( + SELECT + claims.queue, + claims.priority, + claims.enqueue_shard, + {schema}.sequence_next_value(claims.seq_name) AS claim_seq, + {schema}.sequence_next_value(enqueues.seq_name) AS next_seq + FROM {schema}.queue_claim_heads AS claims + JOIN {schema}.queue_enqueue_heads AS enqueues + ON enqueues.queue = claims.queue + AND enqueues.priority = claims.priority + AND enqueues.enqueue_shard = claims.enqueue_shard + ) AS heads + CROSS JOIN LATERAL ( + SELECT count(*)::bigint AS available + FROM {schema}.ready_entries AS ready + WHERE ready.queue = heads.queue + AND ready.priority = heads.priority + AND ready.enqueue_shard = heads.enqueue_shard + AND ready.lane_seq >= heads.claim_seq + AND ready.lane_seq < heads.next_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 + ) + ) AS lane + GROUP BY heads.queue + "# + ) +} + /// What [`queue_signals_sql`] reads. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum QueueSignalReads { @@ -935,34 +979,7 @@ pub fn queue_signals_sql(schema: &str, reads: QueueSignalReads) -> String { "SELECT NULL::text AS queue, 0::bigint AS completed_last_hour WHERE false".to_string(), ), QueueSignalReads::Admin => ( - // Exact: each lane's unclaimed rows, read from its claim cursor - // up to its enqueue head along the primary key, so the cost - // follows the backlog rather than the partition. - format!( - r#" - SELECT head_signal.queue, COALESCE(sum(lane.available), 0)::bigint AS available - FROM head_signal - CROSS JOIN LATERAL ( - SELECT count(*)::bigint AS available - FROM {schema}.ready_entries AS ready - WHERE ready.queue = head_signal.queue - AND ready.priority = head_signal.priority - AND ready.enqueue_shard = head_signal.enqueue_shard - AND ready.lane_seq >= head_signal.claim_seq - AND ready.lane_seq < head_signal.next_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 - ) - ) AS lane - GROUP BY head_signal.queue - "# - ), + available_by_queue_sql(schema), format!( "SELECT queue, count(*)::bigint AS running FROM ({}) AS open_claims GROUP BY queue", open_receipt_running_claims_sql(schema) diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index ddb79d0c8..d4c4079c4 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -21,7 +21,7 @@ use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData}; use opentelemetry_sdk::metrics::{InMemoryMetricExporter, SdkMeterProvider}; use serde::{Deserialize, Serialize}; use sqlx::postgres::PgPoolOptions; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -17217,10 +17217,11 @@ async fn test_queue_storage_multi_shard_public_available_counts_are_exact() { client.shutdown(Duration::from_secs(1)).await; } -/// `queue_overviews` must not read `queue_state_counts`, which queue storage -/// never maintains, nor expand retained completions: completions in the last -/// hour come from batch sizes, and `available` stays exact past a -/// tombstoned ready row, which the sequence-head signal would still count. +/// The dashboard's reads must not read `queue_state_counts`, which queue +/// storage never maintains, nor expand every retained completion: completions +/// are counted from batch sizes, the timeseries reads only batches finalised +/// within its window, and `available` stays exact past a tombstoned ready +/// row, which the sequence-head signal would still count. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_queue_storage_queue_overview_counts_without_terminal_history() { let (_db_guard, pool) = setup_pool(8).await; @@ -17313,6 +17314,48 @@ async fn test_queue_storage_queue_overview_counts_without_terminal_history() { assert_eq!(overview.available, 2, "the cancelled job is not available"); assert_eq!(overview.total_queued, 2); assert_eq!(overview.running, 0); + + // The dashboard's other two reads, over the same jobs. + let counts = admin::state_counts(&pool) + .await + .expect("read queue-storage state counts"); + assert_eq!(counts.get(&JobState::Completed).copied(), Some(5)); + assert_eq!(counts.get(&JobState::Available).copied(), Some(2)); + assert_eq!(counts.get(&JobState::Cancelled).copied(), Some(1)); + + let mut by_state: HashMap = HashMap::new(); + for bucket in admin::state_timeseries(&pool, 60) + .await + .expect("read queue-storage state timeseries") + { + *by_state.entry(bucket.state).or_default() += bucket.count; + } + assert_eq!( + by_state, + HashMap::from([ + (JobState::Completed, 5), + (JobState::Available, 2), + (JobState::Cancelled, 1), + ]) + ); + // A window that ends before these jobs were created holds none of them. + sqlx::query(audited_sql(format!( + "UPDATE {schema}.receipt_completion_batches SET finalized_at = now() - interval '2 hours'" + ))) + .execute(&pool) + .await + .expect("age the completion batches"); + let recent_completed: i64 = admin::state_timeseries(&pool, 60) + .await + .expect("read queue-storage state timeseries") + .into_iter() + .filter(|bucket| bucket.state == JobState::Completed) + .map(|bucket| bucket.count) + .sum(); + assert_eq!( + recent_completed, 0, + "a job finalised before the window began cannot have been created in it" + ); } /// `ordering_key` pins a job to a deterministic shard based on a From 90c4783ffa6c6f5294585e52e3f8d6e6cd216187 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 1 Nov 2026 23:30:42 +0000 Subject: [PATCH 3/3] Leave deleted and retried completions out of completed_last_hour A deleted or retried compact completion keeps its batch unchanged and gains a tombstone, which terminal_jobs honours. Subtract tombstones on members of batches finalised within the hour; such a tombstone is itself newer than the hour, so only recent tombstones are matched. --- awa-model/src/queue_storage.rs | 17 +++++++++++++++++ awa/tests/queue_storage_runtime_test.rs | 14 +++++++++++--- 2 files changed, 28 insertions(+), 3 deletions(-) diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 27b8120b9..7be6e6cbb 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -977,6 +977,23 @@ pub fn queue_signals_sql(schema: &str, reads: QueueSignalReads) -> String { FROM {schema}.receipt_completion_batches WHERE finalized_at > clock_timestamp() - interval '1 hour' UNION ALL + -- Less its members deleted or retried since, as + -- `terminal_jobs` drops them. A tombstone on a member + -- finalised within the hour is itself newer than that. + SELECT tomb.queue, -1::bigint + FROM {schema}.receipt_completion_tombstones AS tomb + WHERE tomb.deleted_at > clock_timestamp() - interval '1 hour' + AND EXISTS ( + SELECT 1 + FROM {schema}.receipt_completion_batches AS batch + WHERE batch.ready_slot = tomb.ready_slot + AND batch.ready_generation = tomb.ready_generation + AND batch.queue = tomb.queue + AND batch.finalized_at > clock_timestamp() - interval '1 hour' + AND batch.run_leases[array_position(batch.job_ids, tomb.job_id)] + = tomb.run_lease + ) + UNION ALL SELECT queue, 1::bigint FROM {schema}.done_entries WHERE state = 'completed' diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index ddb79d0c8..46820c6ce 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -17219,8 +17219,9 @@ async fn test_queue_storage_multi_shard_public_available_counts_are_exact() { /// `queue_overviews` must not read `queue_state_counts`, which queue storage /// never maintains, nor expand retained completions: completions in the last -/// hour come from batch sizes, and `available` stays exact past a -/// tombstoned ready row, which the sequence-head signal would still count. +/// hour come from batch sizes less tombstoned members, and `available` stays +/// exact past a tombstoned ready row, which the sequence-head signal would +/// still count. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_queue_storage_queue_overview_counts_without_terminal_history() { let (_db_guard, pool) = setup_pool(8).await; @@ -17305,11 +17306,18 @@ async fn test_queue_storage_queue_overview_counts_without_terminal_history() { .await .expect("cancel an available job"); + // A deleted completion leaves its batch as it was, with a tombstone. + sqlx::query("DELETE FROM awa.jobs WHERE id = $1") + .bind(completed[0]) + .execute(&pool) + .await + .expect("delete a completed job"); + let overview = admin::queue_overview(&pool, queue) .await .expect("read queue overview") .expect("queue overview should include the queue"); - assert_eq!(overview.completed_last_hour, 5); + assert_eq!(overview.completed_last_hour, 4); assert_eq!(overview.available, 2, "the cancelled job is not available"); assert_eq!(overview.total_queued, 2); assert_eq!(overview.running, 0);