diff --git a/CHANGELOG.md b/CHANGELOG.md index ec6c0e663..3003c068f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,7 @@ Notable changes between releases. Detailed migration notes for storage transitio - The embedded admin UI now works when its axum router is nested under a path such as `/awa` (#502). Assets, API requests, navigation and direct deep links automatically use the mount path, including outer nested routers, without rebuilding the frontend or changing existing router calls. - `admin::list_jobs` and `admin::job_kind_overviews` no longer omit a failed job, or leave it out of its kind's count, while queue prune moves its row out of the slot being pruned. Both read several queries from one snapshot and release partition locks between them, so the queries after a prune committed saw neither the pruned slot's rows nor the moved ones. These reads and the point reads such as `admin::get_job` now compare the relfilenode of every ring partition before their snapshot and after it ends, and read again in a new snapshot when prune or the terminal-count rollup truncated one in between, where 0.6.16 read a single job again only when it found nothing. After three reads that each overlapped a truncation, the fourth locks the ring partitions before taking its snapshot and holds them until it commits, so a read never returns a result a truncation could have changed; prune skips its pass if it cannot get its locks within its lock timeout. The comparison adds two catalog lookups to each read and takes no partition locks, so prune is blocked no more often than before. - Claiming from a ready slot that holds tombstones no longer scans the slot's whole tombstone partition on every claim (#713, reported and fixed by @Wallace99). Each claim checks its candidate rows against `ready_tombstones`, and for a small batch PostgreSQL could plan that check as a hashed subplan over the partition instead of one index probe per row: a 32-row claim read 10,000 tombstone rows in the regression test. The check now carries `OFFSET 0`, which keeps it correlated. Tombstones are left by cancelled, retried and moved jobs, so the cost grew with churn in the slot. Applied by the `correlated_claim_tombstones` schema patch; no configuration or protocol change. +- The admin queue views no longer time out on a large backlog with many ready tombstones. Tombstone checks in the admin SQL are per-row index probes, and the per-queue `available` count reads at most 5,000 unclaimed rows per lane (`admin::AVAILABLE_LANE_EXACT_COUNT_CAP`). A deeper lane reports its head-sequence depth, as the worker queue-depth gauge does, and sets `QueueOverview::available_is_estimate`; the UI shows `≈N` and `awa queue stats` shows `~N`. While a queue's count is an estimate its `lag_seconds` is a lower bound. The state timeseries reads each lane's newest 5,000 available rows, so a lane that gained more than that inside the window is undercounted there. No schema change. `admin::state_counts` now returns `StateCounts { counts, available_is_estimate }` and `/api/stats` returns `{ "counts": {...}, "available_is_estimate": bool }` instead of a bare state map. - Cleanup, cron sync, batch operations, metadata refresh, rollup and ring rotation no longer hold back promotion and rescue (#303). The maintenance leader ran every branch from one loop, so while one branch's statement waited on a lock or ran a long scan, scheduled and retryable jobs were not promoted and stale, overdue and timed-out attempts were not rescued. With one `awa.cron_jobs` row locked by another session, the leader's cron sync stalled both for as long as the lock was held, and stopped cron from firing. Promotion and the three rescues now run in their own loop within the leader term, and the cron evaluator starts before the startup cron sync. A rescue pass still delays promotion, and the other branches still delay one another: a stuck cleanup or cron sync still holds back ring rotation and prune. Leadership and the shutdown grace are unchanged, and no configuration or schema change is needed. A leader holds one more pool connection at a time. While a rescue pass runs, lease prune may report more `blocked` outcomes and claim prune may wait out a rescue batch. See [ADR-051](docs/adr/051-maintenance-recovery-loop.md). - One completion waiting on another transaction's lock no longer stalls every other completion in the runtime. A transaction that changes a running attempt, such as `admin::cancel_tx` inside an application's transaction, holds the attempt's completion lock until it commits. If the attempt's handler returned `Completed` meanwhile, the completion batch waited on that lock, so no other job's completion committed. Once the in-flight completion allowance filled, the runtime stopped claiming in every queue until the transaction ended. With receipts on (the default), a completion batch now leaves an attempt whose lock is held out of the batch, and the worker retries that attempt with backoff for up to two minutes without holding a pool connection, then leaves it to stale-heartbeat rescue as before. Non-success outcomes and kinds with follow-up enqueues already finalized one job at a time and still wait on the lock. With receipts off, a batch still waits. No configuration or schema change. See [a completion waits on an application's transaction](docs/troubleshooting.md#a-completion-waits-on-an-applications-transaction). - Python handlers receive an asyncio cancellation request when the Rust executor abandons their attempt after its deadline grace or shutdown exceeds its worker timeout (#697). Previously, dropping the Rust result receiver left the Python coroutine running, so repeated abandoned attempts leaked live handlers and could produce late effects. Shutdown also waits up to one additional second for that client's Python handlers to finish cancellation cleanup, then warns if any remain. Cancellable handlers receive `asyncio.CancelledError` and can release resources in `finally`; cancellation is cooperative and cannot stop blocking code, spawned threads, child tasks or handlers that suppress cancellation. No schema change. diff --git a/awa-cli/src/main.rs b/awa-cli/src/main.rs index 16d9b2872..6bf2086ba 100644 --- a/awa-cli/src/main.rs +++ b/awa-cli/src/main.rs @@ -1439,7 +1439,11 @@ async fn main() -> Result<(), Box> { println!( "{:<15} {:<10} {:<10} {:<10} {:<15} {:<10} {:<8}", stat.queue, - stat.available, + if stat.available_is_estimate { + format!("~{}", stat.available) + } else { + stat.available.to_string() + }, stat.running, stat.failed, stat.completed_last_hour, diff --git a/awa-model/src/admin.rs b/awa-model/src/admin.rs index c31bfda83..b7c452857 100644 --- a/awa-model/src/admin.rs +++ b/awa-model/src/admin.rs @@ -355,6 +355,7 @@ pub fn queue_storage_current_jobs_cte(schema: &str) -> String { AND tomb.priority = ready.priority AND tomb.enqueue_shard = ready.enqueue_shard AND tomb.lane_seq = ready.lane_seq + OFFSET 0 ) ), current_jobs AS ( @@ -391,6 +392,7 @@ pub fn queue_storage_current_jobs_cte(schema: &str) -> String { AND tomb.priority = ready.priority AND tomb.enqueue_shard = ready.enqueue_shard AND tomb.lane_seq = ready.lane_seq + OFFSET 0 ) UNION ALL -- Receipt claims that have not materialised into `leases` are @@ -432,6 +434,7 @@ pub fn queue_storage_current_jobs_cte(schema: &str) -> String { AND tomb.priority = ready.priority AND tomb.enqueue_shard = ready.enqueue_shard AND tomb.lane_seq = ready.lane_seq + OFFSET 0 ) WHERE claims.closed_at IS NULL AND NOT EXISTS ( @@ -501,6 +504,7 @@ pub fn queue_storage_current_jobs_cte(schema: &str) -> String { AND tomb.priority = ready.priority AND tomb.enqueue_shard = ready.enqueue_shard AND tomb.lane_seq = ready.lane_seq + OFFSET 0 ) WHERE NOT EXISTS ( SELECT 1 FROM {schema}.lease_claim_closures AS cx @@ -551,24 +555,43 @@ pub fn queue_storage_current_jobs_cte(schema: &str) -> String { ) } -/// Each queue's available rows, exactly, and the oldest of them, as -/// `(queue, available, oldest_run_at)`. +/// The most unclaimed rows [`queue_overviews`] and [`state_counts`] walk in one +/// lane to count it exactly. A lane holding more is counted from its head +/// sequences instead, so the cost of the admin views is bounded per lane +/// rather than following the backlog. +pub const AVAILABLE_LANE_EXACT_COUNT_CAP: i64 = 5_000; + +/// Each queue's available rows and the oldest of them, as +/// `(queue, available, available_is_estimate, oldest_run_at)`. /// /// Each lane's unclaimed rows are found the way a claim finds them: through /// `ready_segments` from the lane's claim cursor, then along the ready /// primary key within each segment's ring slot. The ready key leads with the /// ring slot, so a lane range without it cannot use the index and would scan -/// every retained ready row. The cost follows the backlog. +/// every retained ready row. /// -/// `oldest_run_at` is the earliest `run_at` among those rows, not the first -/// row's: a queue move or priority aging appends an older job behind newer -/// ones. +/// A lane is walked for at most [`AVAILABLE_LANE_EXACT_COUNT_CAP`] rows, one +/// more to detect overflow. Within the cap its count is exact. Past it the +/// lane contributes `next_seq - claim_seq`, the head-sequence depth the worker +/// queue gauges publish, which also counts tombstoned rows, and the queue's +/// `available_is_estimate` is set. +/// +/// `oldest_run_at` is the earliest `run_at` among the rows walked, not the +/// first row's: a queue move or priority aging appends an older job behind +/// newer ones. Beyond the cap it is the oldest among the first rows in lane +/// order. fn queue_storage_available_by_queue_sql(schema: &str) -> String { format!( r#" SELECT heads.queue, - COALESCE(sum(lane.available), 0)::bigint AS available, + COALESCE(sum( + CASE WHEN lane.walked > {cap} + THEN GREATEST(heads.next_seq - heads.claim_seq, 0) + ELSE lane.walked + END + ), 0)::bigint AS available, + COALESCE(bool_or(lane.walked > {cap}), FALSE) AS available_is_estimate, min(lane.oldest_run_at) AS oldest_run_at FROM ( SELECT @@ -588,26 +611,87 @@ fn queue_storage_available_by_queue_sql(schema: &str) -> String { ) AS lane GROUP BY heads.queue "#, + cap = AVAILABLE_LANE_EXACT_COUNT_CAP, lane_rows = queue_storage_available_lane_rows_sql( schema, "heads", - "count(*)::bigint AS available, min(ready.run_at) AS oldest_run_at", + "count(*)::bigint AS walked, min(ready.run_at) AS oldest_run_at", "TRUE", + Some(LaneWalk { + direction: LaneWalkDirection::FromClaimCursor, + limit: AVAILABLE_LANE_EXACT_COUNT_CAP + 1, + }), ), ) } +/// Which end of a lane's backlog a bounded walk starts from. +#[derive(Clone, Copy)] +enum LaneWalkDirection { + /// Oldest first, from the lane's claim cursor. + FromClaimCursor, + /// Newest first, from the lane's enqueue head. + FromTail, +} + +/// A bound on how many unclaimed rows a lane walk reads. +#[derive(Clone, Copy)] +struct LaneWalk { + direction: LaneWalkDirection, + limit: i64, +} + /// A `SELECT {columns} FROM ready_segments JOIN ready_entries` over the /// unclaimed, untombstoned rows of the lane in `heads` (which must carry /// `queue`, `priority`, `enqueue_shard`, `claim_seq` and `next_seq`) that /// satisfy `condition` on `ready`. See [`queue_storage_available_by_queue_sql`] for why it /// walks segments. +/// +/// With a `walk`, only the first `limit` rows from its end of the lane are +/// read, and `condition` applies to those rows rather than choosing them. fn queue_storage_available_lane_rows_sql( schema: &str, heads: &str, columns: &str, condition: &str, + walk: Option, +) -> String { + match walk { + None => queue_storage_lane_walk_sql(schema, heads, columns, condition, None), + Some(walk) => { + let walked = queue_storage_lane_walk_sql(schema, heads, "ready.*", "TRUE", Some(walk)); + format!( + r#" + SELECT {columns} + FROM ({walked}) AS ready + WHERE ({condition}) + "# + ) + } + } +} + +fn queue_storage_lane_walk_sql( + schema: &str, + heads: &str, + columns: &str, + condition: &str, + walk: Option, ) -> String { + let (direction, ready_order, limit) = match walk { + None => ("", String::new(), String::new()), + Some(LaneWalk { direction, limit }) => { + let keyword = match direction { + LaneWalkDirection::FromClaimCursor => "ASC", + LaneWalkDirection::FromTail => "DESC", + }; + ( + keyword, + format!("ORDER BY ready.lane_seq {keyword}"), + format!("ORDER BY segment.next_lane_seq {keyword}, ready.lane_seq {keyword} LIMIT {limit}"), + ) + } + }; format!( r#" SELECT {columns} @@ -618,6 +702,7 @@ fn queue_storage_available_lane_rows_sql( SELECT segment.ready_slot, segment.ready_generation, + segment.next_lane_seq, GREATEST(segment.first_lane_seq, {heads}.claim_seq) AS from_seq, LEAST(segment.next_lane_seq, {heads}.next_seq) AS to_seq FROM {schema}.ready_segments AS segment @@ -626,7 +711,7 @@ fn queue_storage_available_lane_rows_sql( AND segment.enqueue_shard = {heads}.enqueue_shard AND segment.next_lane_seq > {heads}.claim_seq AND segment.first_lane_seq < {heads}.next_seq - ORDER BY segment.next_lane_seq + ORDER BY segment.next_lane_seq {direction} ) AS segment -- Each segment's rows along the ready key within its ring slot. -- OFFSET 0 keeps this a per-segment probe: flattened into a @@ -642,9 +727,14 @@ fn queue_storage_available_lane_rows_sql( AND ready.enqueue_shard = {heads}.enqueue_shard AND ready.lane_seq >= segment.from_seq AND ready.lane_seq < segment.to_seq + {ready_order} OFFSET 0 ) AS ready WHERE ({condition}) + -- OFFSET 0 keeps the tombstone check a probe per row checked. + -- Flattened into an anti join, Postgres can sort and merge + -- the whole tombstone table against each lane's rows, so the + -- cost follows the tombstones rather than the rows walked. AND NOT EXISTS ( SELECT 1 FROM {schema}.ready_tombstones AS tomb WHERE tomb.ready_slot = ready.ready_slot @@ -653,7 +743,9 @@ fn queue_storage_available_lane_rows_sql( AND tomb.priority = ready.priority AND tomb.enqueue_shard = ready.enqueue_shard AND tomb.lane_seq = ready.lane_seq + OFFSET 0 ) + {limit} "# ) } @@ -662,7 +754,8 @@ fn queue_storage_available_lane_rows_sql( /// They come from the live-work tables the worker queue gauges read, never /// from `terminal_jobs`, whose completion batches expand and join back to /// `ready_entries`: the cost follows live work rather than retained history. -/// `available` is exact, and `completed_last_hour` comes from completion-batch +/// `available` is exact up to a per-lane cap (see +/// [`queue_storage_available_by_queue_sql`]), and `completed_last_hour` comes from completion-batch /// sizes and completed `done_entries` rows. fn queue_storage_signals_cte(schema: &str) -> String { format!( @@ -809,6 +902,7 @@ fn queue_storage_signals_cte(schema: &str) -> String { SELECT queues.queue, COALESCE(ready.available, 0)::bigint AS available, + COALESCE(ready.available_is_estimate, FALSE) AS available_is_estimate, COALESCE(leases.running, 0)::bigint AS running, COALESCE(leases.waiting_external, 0)::bigint AS waiting_external, COALESCE(deferred.scheduled, 0)::bigint AS scheduled, @@ -846,6 +940,10 @@ fn queue_storage_recent_jobs_cte(schema: &str, since: &str) -> String { "head_signal", "ready.created_at", &format!("ready.created_at >= {since}"), + Some(LaneWalk { + direction: LaneWalkDirection::FromTail, + limit: AVAILABLE_LANE_EXACT_COUNT_CAP, + }), ); let open_running = crate::queue_storage::open_receipt_running_claims_sql(schema); format!( @@ -894,6 +992,7 @@ fn queue_storage_recent_jobs_cte(schema: &str, since: &str) -> String { AND tomb.priority = ready.priority AND tomb.enqueue_shard = ready.enqueue_shard AND tomb.lane_seq = ready.lane_seq + OFFSET 0 ) WHERE ready.created_at >= {since} UNION ALL @@ -953,6 +1052,7 @@ fn queue_storage_recent_jobs_cte(schema: &str, since: &str) -> String { AND tomb.ready_generation = batch.ready_generation AND tomb.job_id = item.job_id AND tomb.run_lease = item.run_lease + OFFSET 0 ) UNION ALL SELECT created_at, 'failed'::awa.job_state @@ -979,7 +1079,8 @@ enum KindCountSource { // Generate only this source's relations: unused CTEs would still acquire // relation locks during parse/plan before execution could discard them. fn queue_storage_kind_count_source_sql(schema: &str, source: KindCountSource) -> String { - let available = queue_storage_available_lane_rows_sql(schema, "heads", "ready.kind", "TRUE"); + let available = + queue_storage_available_lane_rows_sql(schema, "heads", "ready.kind", "TRUE", None); match source { KindCountSource::Available => format!( r#" @@ -1019,6 +1120,7 @@ fn queue_storage_kind_count_source_sql(schema: &str, source: KindCountSource) -> AND tomb.priority = ready.priority AND tomb.enqueue_shard = ready.enqueue_shard AND tomb.lane_seq = ready.lane_seq + OFFSET 0 ) GROUP BY ready.kind, leases.queue "# @@ -1079,6 +1181,7 @@ fn queue_storage_kind_count_source_sql(schema: &str, source: KindCountSource) -> AND tomb.ready_generation = batch.ready_generation AND tomb.job_id = item.job_id AND tomb.run_lease = item.run_lease + OFFSET 0 ) GROUP BY ready.kind, batch.queue "# @@ -1214,6 +1317,7 @@ fn queue_storage_job_id_candidates_sql( "TRUE{}", predicates(None, Some("ready.kind"), "ready.queue", "ready.job_id") ), + None, ); let head_queue = queue .as_ref() @@ -1288,6 +1392,7 @@ fn queue_storage_job_id_candidates_sql( AND tomb.priority = ready.priority AND tomb.enqueue_shard = ready.enqueue_shard AND tomb.lane_seq = ready.lane_seq + OFFSET 0 ) WHERE TRUE{predicates} ORDER BY leases.job_id DESC @@ -1315,6 +1420,7 @@ fn queue_storage_job_id_candidates_sql( AND tomb.priority = ready.priority AND tomb.enqueue_shard = ready.enqueue_shard AND tomb.lane_seq = ready.lane_seq + OFFSET 0 ) LIMIT 1 ) AS created @@ -1397,6 +1503,7 @@ fn queue_storage_job_id_candidates_sql( AND tomb.ready_generation = member.ready_generation AND tomb.job_id = member.job_id AND tomb.run_lease = member.run_lease + OFFSET 0 ) LIMIT 1 ) AS live @@ -1867,6 +1974,7 @@ fn unique_key_candidate_sql(schema: &str, scoped: bool) -> String { AND tomb.priority = ready.priority AND tomb.enqueue_shard = ready.enqueue_shard AND tomb.lane_seq = ready.lane_seq + OFFSET 0 ) UNION ALL SELECT job_id @@ -1897,6 +2005,7 @@ fn unique_key_candidate_sql(schema: &str, scoped: bool) -> String { AND tomb.priority = ready.priority AND tomb.enqueue_shard = ready.enqueue_shard AND tomb.lane_seq = ready.lane_seq + OFFSET 0 ) AND ready.unique_key = $1 OFFSET 0 @@ -1971,6 +2080,7 @@ fn scoped_unique_key_candidate_sql(schema: &str) -> String { AND tomb.priority = keyed.priority AND tomb.enqueue_shard = keyed.enqueue_shard AND tomb.lane_seq = keyed.lane_seq + OFFSET 0 ) UNION ALL SELECT job_id @@ -2773,11 +2883,18 @@ pub struct QueueOverview { pub total_queued: i64, pub scheduled: i64, pub available: i64, + /// `available` is a head-sequence estimate for at least one lane whose + /// backlog exceeds [`AVAILABLE_LANE_EXACT_COUNT_CAP`]; otherwise it is exact. + pub available_is_estimate: bool, pub retryable: i64, pub running: i64, pub failed: i64, pub waiting_external: i64, pub completed_last_hour: i64, + /// Seconds since the earliest `run_at` among the available rows counted. + /// While `available_is_estimate` is set this is a lower bound: a job + /// moved or aged into a lane behind its first + /// [`AVAILABLE_LANE_EXACT_COUNT_CAP`] rows is not seen. pub lag_seconds: Option, pub paused: bool, } @@ -3597,6 +3714,7 @@ pub async fn queue_overviews(pool: &PgPool) -> Result, AwaErr AS total_queued, COALESCE(sg.scheduled, 0) AS scheduled, COALESCE(sg.available, 0) AS available, + COALESCE(sg.available_is_estimate, FALSE) AS available_is_estimate, COALESCE(sg.retryable, 0) AS retryable, COALESCE(sg.running, 0) AS running, COALESCE(sg.failed, 0) AS failed, @@ -3699,6 +3817,7 @@ pub async fn queue_overviews(pool: &PgPool) -> Result, AwaErr COALESCE(qs.scheduled + qs.available + qs.running + qs.retryable + qs.waiting_external, 0) AS total_queued, COALESCE(qs.scheduled, 0) AS scheduled, COALESCE(qs.available, 0) AS available, + FALSE AS available_is_estimate, COALESCE(qs.retryable, 0) AS retryable, COALESCE(qs.running, 0) AS running, COALESCE(qs.failed, 0) AS failed, @@ -4059,16 +4178,36 @@ pub async fn dump_run( }) } +/// Job counts by state, as [`state_counts`] returns them. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct StateCounts { + pub counts: HashMap, + /// `counts[Available]` is a head-sequence estimate for at least one lane + /// holding more than [`AVAILABLE_LANE_EXACT_COUNT_CAP`] unclaimed rows; + /// otherwise it is exact. + pub available_is_estimate: bool, +} + +impl StateCounts { + /// The count for `state`, zero if the state has no row. + pub fn count(&self, state: JobState) -> i64 { + self.counts.get(&state).copied().unwrap_or(0) + } +} + /// Count jobs grouped by state. /// -/// Reads from the `queue_state_counts` cache table. -pub async fn state_counts(pool: &PgPool) -> Result, AwaError> { +/// Reads from the `queue_state_counts` cache table. On queue storage, +/// `available` is exact up to [`AVAILABLE_LANE_EXACT_COUNT_CAP`] unclaimed +/// rows per lane, and `available_is_estimate` says when it is not. +pub async fn state_counts(pool: &PgPool) -> Result { if let Some(store) = active_queue_storage(pool).await? { let sql = format!( r#" SELECT COALESCE((SELECT count(*)::bigint FROM {schema}.deferred_jobs WHERE state = 'scheduled'), 0) AS scheduled, - COALESCE((SELECT sum(available)::bigint FROM ({available}) AS lanes), 0) AS available, + COALESCE(avail.available, 0) AS available, + COALESCE(avail.available_is_estimate, FALSE) AS available_is_estimate, COALESCE((SELECT count(*)::bigint FROM {schema}.leases WHERE state = 'running'), 0) + COALESCE((SELECT count(*)::bigint FROM ({open_running}) AS open_running), 0) AS running, -- A completion batch carries its size, so completed jobs are @@ -4083,6 +4222,12 @@ pub async fn state_counts(pool: &PgPool) -> Result, AwaEr + COALESCE((SELECT count(*)::bigint FROM {schema}.dlq_entries), 0) AS failed, COALESCE((SELECT count(*)::bigint FROM {schema}.done_entries WHERE state = 'cancelled'), 0) AS cancelled, COALESCE((SELECT count(*)::bigint FROM {schema}.leases WHERE state = 'waiting_external'), 0) AS waiting_external + FROM ( + SELECT + sum(available)::bigint AS available, + bool_or(available_is_estimate) AS available_is_estimate + FROM ({available}) AS lanes + ) AS avail "#, available = queue_storage_available_by_queue_sql(store.schema()), open_running = crate::queue_storage::open_receipt_running_claims_sql(store.schema()), @@ -4092,13 +4237,14 @@ pub async fn state_counts(pool: &PgPool) -> Result, AwaEr let ( scheduled, available, + available_is_estimate, running, completed, retryable, failed, cancelled, waiting_external, - ): (i64, i64, i64, i64, i64, i64, i64, i64) = { + ): (i64, i64, bool, i64, i64, i64, i64, i64, i64) = { // As in `queue_overviews`: compiling costs more than running. let mut begin_guard = crate::read_transaction::BeginGuard::acquire(pool).await?; let mut tx = sqlx::Connection::begin_with( @@ -4116,16 +4262,19 @@ pub async fn state_counts(pool: &PgPool) -> Result, AwaEr counts }; - return Ok(HashMap::from([ - (JobState::Scheduled, scheduled), - (JobState::Available, available), - (JobState::Running, running), - (JobState::Completed, completed), - (JobState::Retryable, retryable), - (JobState::Failed, failed), - (JobState::Cancelled, cancelled), - (JobState::WaitingExternal, waiting_external), - ])); + return Ok(StateCounts { + counts: HashMap::from([ + (JobState::Scheduled, scheduled), + (JobState::Available, available), + (JobState::Running, running), + (JobState::Completed, completed), + (JobState::Retryable, retryable), + (JobState::Failed, failed), + (JobState::Cancelled, cancelled), + (JobState::WaitingExternal, waiting_external), + ]), + available_is_estimate, + }); } // Single scan of queue_state_counts — sums all columns in one pass @@ -4159,7 +4308,10 @@ pub async fn state_counts(pool: &PgPool) -> Result, AwaEr .fetch_all(pool) .await?; - Ok(rows.into_iter().collect()) + Ok(StateCounts { + counts: rows.into_iter().collect(), + available_is_estimate: false, + }) } #[derive(sqlx::FromRow)] @@ -4319,6 +4471,7 @@ async fn queue_storage_running_kind_counts( AND tomb.priority = ready.priority AND tomb.enqueue_shard = ready.enqueue_shard AND tomb.lane_seq = ready.lane_seq + OFFSET 0 ) LIMIT 1 ) AS created @@ -4712,6 +4865,11 @@ pub struct StateTimeseriesBucket { } /// Return time-bucketed state counts over the last N minutes. +/// +/// On queue storage the available buckets read at most +/// [`AVAILABLE_LANE_EXACT_COUNT_CAP`] rows per lane, newest first. A lane +/// that gained more available jobs than that inside the window is +/// undercounted, with nothing in the result to mark it. pub async fn state_timeseries( pool: &PgPool, minutes: i32, @@ -5883,3 +6041,79 @@ fn apply_transform(job: &JobRow, payload_value: &serde_json::Value) -> serde_jso } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{migrations, QueueStorageConfig}; + + /// Tombstone rows read by sequential scans, per relation. A scan of a + /// partition that was planned but never run, or is empty, reads none. + fn tombstone_rows_seq_scanned(plan: &serde_json::Value, found: &mut Vec<(String, u64)>) { + if plan["Node Type"] == "Seq Scan" + && plan["Relation Name"] + .as_str() + .is_some_and(|name| name.starts_with("ready_tombstones")) + { + let read = plan["Actual Loops"].as_u64().unwrap_or(0) + * (plan["Actual Rows"].as_u64().unwrap_or(0) + + plan["Rows Removed by Filter"].as_u64().unwrap_or(0)); + if read > 0 { + found.push((plan["Relation Name"].to_string(), read)); + } + } + for child in plan["Plans"].as_array().into_iter().flatten() { + tombstone_rows_seq_scanned(child, found); + } + } + + /// With many tombstones, an anti join over them sorts and merges the whole + /// table against each lane's rows. The count must probe them per row. + #[sqlx::test(migrations = false)] + async fn available_count_probes_tombstones_per_row( + pool: PgPool, + ) -> Result<(), Box> { + migrations::run(&pool).await?; + let schema = "tombstone_probe"; + let store = QueueStorage::new(QueueStorageConfig { + schema: schema.into(), + queue_slot_count: 4, + lease_slot_count: 2, + claim_slot_count: 2, + ..Default::default() + })?; + store.install(&pool).await?; + store + .enqueue_batch(&pool, "probe", 1, 4 * AVAILABLE_LANE_EXACT_COUNT_CAP) + .await?; + sqlx::raw_sql(&format!( + "INSERT INTO {schema}.ready_tombstones + (ready_slot, ready_generation, queue, priority, enqueue_shard, lane_seq, job_id) + SELECT ready_slot, ready_generation, queue, priority, enqueue_shard, lane_seq, job_id + FROM {schema}.ready_entries WHERE lane_seq % 2 = 0; + ANALYZE {schema}.ready_entries; + ANALYZE {schema}.ready_tombstones;" + )) + .execute(&pool) + .await?; + + let mut tx = pool.begin().await?; + sqlx::raw_sql("SET LOCAL work_mem = '125MB'; SET LOCAL random_page_cost = 1.2;") + .execute(&mut *tx) + .await?; + let sql = queue_storage_available_by_queue_sql(schema); + let explained: serde_json::Value = + sqlx::query_scalar(&format!("EXPLAIN (ANALYZE, FORMAT JSON) {sql}")) + .fetch_one(&mut *tx) + .await?; + tx.rollback().await?; + + let mut scanned = Vec::new(); + tombstone_rows_seq_scanned(&explained[0]["Plan"], &mut scanned); + assert!( + scanned.is_empty(), + "tombstones are read by sequential scans: {scanned:?}\n{explained:#}" + ); + Ok(()) + } +} diff --git a/awa-python/python/awa/_awa.pyi b/awa-python/python/awa/_awa.pyi index 8842384fd..4618749a8 100644 --- a/awa-python/python/awa/_awa.pyi +++ b/awa-python/python/awa/_awa.pyi @@ -62,6 +62,8 @@ class QueueStat: @property def available(self) -> int: ... @property + def available_is_estimate(self) -> bool: ... + @property def retryable(self) -> int: ... @property def running(self) -> int: ... diff --git a/awa-python/src/client.rs b/awa-python/src/client.rs index 9645c76a6..d13f04136 100644 --- a/awa-python/src/client.rs +++ b/awa-python/src/client.rs @@ -326,6 +326,8 @@ pub struct PyQueueStat { #[pyo3(get)] pub available: i64, #[pyo3(get)] + pub available_is_estimate: bool, + #[pyo3(get)] pub retryable: i64, #[pyo3(get)] pub running: i64, @@ -1233,6 +1235,7 @@ impl PyClient { total_queued: stat.total_queued, scheduled: stat.scheduled, available: stat.available, + available_is_estimate: stat.available_is_estimate, retryable: stat.retryable, running: stat.running, failed: stat.failed, @@ -2825,6 +2828,7 @@ impl PyClient { total_queued: s.total_queued, scheduled: s.scheduled, available: s.available, + available_is_estimate: s.available_is_estimate, retryable: s.retryable, running: s.running, failed: s.failed, diff --git a/awa-python/tests/test_adversarial_api.py b/awa-python/tests/test_adversarial_api.py index 6ac385d6d..934af26b1 100644 --- a/awa-python/tests/test_adversarial_api.py +++ b/awa-python/tests/test_adversarial_api.py @@ -175,6 +175,7 @@ async def test_queue_stats_returns_typed_objects(async_client): assert hasattr(stat, "total_queued") assert hasattr(stat, "scheduled") assert hasattr(stat, "available") + assert hasattr(stat, "available_is_estimate") assert hasattr(stat, "retryable") assert hasattr(stat, "running") assert hasattr(stat, "failed") diff --git a/awa-ui/frontend/src/components/AvailableCount.tsx b/awa-ui/frontend/src/components/AvailableCount.tsx new file mode 100644 index 000000000..902b9c3e7 --- /dev/null +++ b/awa-ui/frontend/src/components/AvailableCount.tsx @@ -0,0 +1,15 @@ +/** Available count; prefixed with "≈" when the backlog exceeds the exact-count limit. */ +export function AvailableCount({ + count, + isEstimate, +}: { + count: number; + isEstimate?: boolean; +}) { + if (!isEstimate) return <>{count.toLocaleString()}; + return ( + + ≈{count.toLocaleString()} + + ); +} diff --git a/awa-ui/frontend/src/lib/api.ts b/awa-ui/frontend/src/lib/api.ts index d615464d4..4bfe2d305 100644 --- a/awa-ui/frontend/src/lib/api.ts +++ b/awa-ui/frontend/src/lib/api.ts @@ -66,6 +66,8 @@ export interface QueueOverview extends DescriptorFields { total_queued: number; scheduled: number; available: number; + /** Absent on servers that predate the exact-count limit. */ + available_is_estimate?: boolean; retryable: number; running: number; failed: number; @@ -189,7 +191,11 @@ export interface CronJobRow { paused_by: string | null; } -export type StateCounts = Record; +export interface StateCounts { + counts: Record; + /** `counts.available` is a head-sequence estimate for an over-deep lane. */ + available_is_estimate: boolean; +} export interface TimeseriesBucket { bucket: string; diff --git a/awa-ui/frontend/src/routes/index.tsx b/awa-ui/frontend/src/routes/index.tsx index 6400e1d91..8d3672c2b 100644 --- a/awa-ui/frontend/src/routes/index.tsx +++ b/awa-ui/frontend/src/routes/index.tsx @@ -29,6 +29,7 @@ import { } from "@/components/ui/table"; import { Badge } from "@/components/ui/badge"; import { LagValue } from "@/components/LagValue"; +import { AvailableCount } from "@/components/AvailableCount"; import { timeAgo } from "@/lib/time"; import { DASHBOARD_QUEUE_LIMIT } from "@/lib/constants"; import { usePollInterval } from "@/hooks/use-poll-interval"; @@ -120,7 +121,7 @@ export function DashboardPage() { : null; const totalJobs = statsQuery.data - ? Object.values(statsQuery.data).reduce((a, b) => a + b, 0) + ? Object.values(statsQuery.data.counts).reduce((a, b) => a + b, 0) : null; // Sort queues by visible workload so deferred-heavy queues stay prominent. @@ -142,7 +143,7 @@ export function DashboardPage() { {/* Headline counter cards */}
{COUNTER_KEYS.map((key) => { - const count = statsQuery.data?.[key]; + const count = statsQuery.data?.counts[key]; const bg = STATE_CARD_BG[key] ?? ""; const series = seriesByState[key] ?? []; return ( @@ -160,7 +161,12 @@ export function DashboardPage() { {statsQuery.isLoading ? ( ) : ( - (count ?? 0).toLocaleString() + )}
@@ -459,7 +465,7 @@ export function DashboardPage() { {q.scheduled.toLocaleString()} - {q.available.toLocaleString()} + {q.retryable.toLocaleString()} diff --git a/awa-ui/frontend/src/routes/jobs.tsx b/awa-ui/frontend/src/routes/jobs.tsx index 57202d6bf..271de2125 100644 --- a/awa-ui/frontend/src/routes/jobs.tsx +++ b/awa-ui/frontend/src/routes/jobs.tsx @@ -24,6 +24,7 @@ import { Heading } from "@/components/ui/heading"; import { Badge } from "@/components/ui/badge"; import { Button } from "@/components/ui/button"; import { ConfirmDialog } from "@/components/ConfirmDialog"; +import { AvailableCount } from "@/components/AvailableCount"; import { Table, TableHeader, @@ -258,7 +259,7 @@ export function JobsPage() { }; const total = statsQuery.data - ? Object.values(statsQuery.data).reduce((a, b) => a + b, 0) + ? Object.values(statsQuery.data.counts).reduce((a, b) => a + b, 0) : undefined; const lastJob = jobs[jobs.length - 1]; @@ -291,7 +292,7 @@ export function JobsPage() { {queueStats.description} )} - {queueStats.available} available ·{" "} + available ·{" "} {queueStats.running} running ·{" "} 0 ? "text-danger" : ""}> {queueStats.failed} failed @@ -334,7 +335,7 @@ export function JobsPage() { {/* Compact state filter pills + search on same row */}
{STATES.map((s) => { - const count = s === "all" ? total : statsQuery.data?.[s]; + const count = s === "all" ? total : statsQuery.data?.counts[s]; const isActive = filters.state === s; return (