diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 2ae3aee83..5dd4872b2 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -204,14 +204,18 @@ jobs:
# canonical, so this leg re-runs it with the queue-storage engine active on a
# fresh database (a separate DB avoids cross-engine job_unique_claims carryover).
rust-test-queue-storage:
- name: Rust tests (queue_storage)
+ name: Rust tests (queue_storage, pg${{ matrix.postgres }})
needs: [rust-build]
if: github.event_name != 'pull_request' || contains(github.event.pull_request.labels.*.name, 'full-ci')
runs-on: ubuntu-latest
timeout-minutes: 30
+ strategy:
+ fail-fast: false
+ matrix:
+ postgres: ["17", "18"]
services:
postgres:
- image: postgres:17-alpine
+ image: postgres:${{ matrix.postgres }}-alpine
env:
POSTGRES_PASSWORD: postgres
POSTGRES_DB: awa_test
@@ -314,10 +318,14 @@ jobs:
# ─── Python (awa-python) ────────────────────────────────
python-build-test:
- name: Python build + test
+ name: Python build + test (${{ matrix.python }})
if: github.event_name != 'pull_request' || contains(github.event.pull_request.labels.*.name, 'full-ci')
runs-on: ubuntu-latest
timeout-minutes: 60
+ strategy:
+ fail-fast: false
+ matrix:
+ python: ["3.14", "3.15"]
services:
postgres:
image: postgres:17-alpine
@@ -342,7 +350,7 @@ jobs:
workspaces: "awa-python -> awa-python/target"
- uses: actions/setup-python@v6
with:
- python-version: "3.12"
+ python-version: ${{ matrix.python }}
- name: Install uv
uses: astral-sh/setup-uv@v8.2.0
- name: Create venv and install deps
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1db9c1170..5f19bd2ff 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,185 +4,86 @@ Notable changes between releases. Detailed migration notes for storage transitio
## [Unreleased]
-### Added
+### Fixed
+
+- Bound outstanding worker finalization to the larger of runtime worker capacity
+ and twice the completion batch size times the shard count. Finished handlers
+ retain their worker slot while waiting for finalizer capacity, preventing an
+ unbounded backlog of tasks retaining job payloads when durable completion falls
+ behind. Attempt heartbeats remain active until finalization settles.
+- Complete retried queue-storage jobs and jobs carrying metadata by compact claim
+ batch identity and item index, avoiding a scan of every retained claim batch.
+
+## [0.6.14] — 2026-11-08
-- **The queue-storage prune `lock_timeout` is configurable, and prune
- attempts are counted by outcome for every ring.** Prune (queue, lease and
- claim rings) and terminal-delta rollup take their exclusive locks under a
- 10 ms `lock_timeout` that could only be changed through the hidden
- `QueueStorage::with_prune_lock_timeout`. When short readers of the ring
- partitions held a conflicting lock at that instant, prune reported
- `blocked` every tick and the rings stopped recycling. Set it with
- `ClientBuilder::prune_lock_timeout` (Rust) or
- `queue_storage_prune_lock_timeout_ms` on `client.start()` (Python); the
- timeout applies separately to each lock acquisition, not to the whole prune;
- queue prune can wait on up to eight partition locks. The default stays 10 ms and zero or a value above PostgreSQL's `lock_timeout`
- maximum is rejected. Start logs a warning when the value is at or above the
- connection's `deadlock_timeout` (prune waiting that long can make deadlock
- detection cancel a claim instead) or `statement_timeout`. `with_prune_lock_timeout` is now
- public, with `set_prune_lock_timeout` and `prune_lock_timeout()`.
- `awa_maintenance_prune_attempts_total{awa_ring,awa_ring_outcome,awa_ring_reason}`
- gains `backed_off` (prune skipped by the backoff that follows consecutive
- `skipped_active` or `blocked` results) and
- `error` (prune failed with a database error) alongside the existing
- `pruned`, `noop`, `blocked` and `skipped_active`. See
- [Configuration](docs/configuration.md#what-the-knobs-mean).
+Recommended for everyone on 0.6.13 using queue storage. A busy queue no longer stalls promotion of the others, prune backs off cleanly under lock contention, new queue lanes are usable by every runtime role, and jobs no longer stay running, run again or get misreported after priority aging.
+
+### Upgrading
+
+- The schema stays at v40. Run `awa migrate` once as the schema owner while workers keep running. It applies the repeatable `definer_lane_provisioning` patch and repairs existing lane sequences. Then roll workers, producers and admin processes in any order. Binaries can roll back to 0.6.13 with the patch left in place.
+- `awa storage lane-sequences` checks lane-sequence ownership and grants and exits with code 2 while problems remain; `--repair` fixes what the connected role may.
+- External migration runners: apply `R__definer_lane_provisioning.sql` after `V40` (`awa migrate --extract-to
--pending`), replacing all `R__` files together, then run `awa storage lane-sequences --repair` as the migration role. See the [upgrade guide](docs/upgrade-0.6.13-to-0.6.14.md).
### Fixed
-- Queue-storage completion of retried jobs or jobs carrying metadata now
- addresses compact claims by their batch identity and item index, as on main.
- It no longer expands every retained claim batch for each completion flush,
- so bounded finalization can drain large recovery backlogs without that scan.
-
-- **A completion backlog no longer gets finished jobs rescued and run
- again.** An attempt stops heartbeating when its handler returns, and its
- finalize then waits for the completion batcher. The runtime released the
- worker slot at the same moment and kept claiming, so when finalization was
- slower than claiming (for example, retried jobs, whose completions take
- the slower per-row path) the queue of finished, unheartbeated attempts grew
- without bound. Attempts that waited longer than `heartbeat_staleness` were
- rescued and executed again, some of them twice. The bound on attempts waiting
- for finalization is `max(worker count, 2 × batch size × completion shards)`.
- Past that, a finished attempt keeps its
- worker slot and its heartbeat until there is room, so the dispatcher stops
- claiming until finalization catches up.
-- **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
- only that `LOCK TABLE` and the final `TRUNCATE` turned a timeout into
- `PruneOutcome::Blocked`. The timeout covers every later statement in the
- transaction too: the post-lock liveness checks read the lease and claim
- partitions, the failed-row carry inserts into the current slot's
- `done_entries` and terminal-count deltas, and the fold writes
- `queue_terminal_rollups` and deletes `queue_terminal_live_counts` rows. When
- one of those waited out the timeout, for example behind autovacuum
- truncating a relation or a hosted Postgres's own background workers, prune
- failed with a `lock_not_available` (`55P03`) database error, was logged as
- an error and skipped the prune backoff. A lock timeout anywhere in a queue,
- lease or claim prune, or in a terminal-count rollup slot, now rolls the
- whole transaction back and reports `Blocked` (`blocked_slots` for rollup),
- so maintenance backs off and retries; nothing the attempt wrote survives
- the rollback. A rollup pass stops at its first blocked slot rather than
- waiting out the timeout again on each remaining one.
-- **On queue storage, a queue whose enqueue lane is busy no longer holds
- back promotion of other queues, nor the maintenance leader's rescue,
- rotation and prune.** Promotion re-enqueues due scheduled and retryable
- rows into their lane, which takes the lane's reservation lock that
- producers hold until they commit. A promotion batch was one transaction
- across every queue it touched, and the leader runs its branches one after
- another, so a lane kept busy by producers delayed due jobs of unrelated
- queues in the same batch, and rescue started seconds late. Promotion now
- gives each `(queue, priority)` lane its own transaction under a 50 ms
- `lock_timeout`. A lane that times out keeps its rows deferred and is
- handed to a dedicated task that waits up to 30 s for the lock, queued
- behind the producers already waiting, so the lane is still promoted in
- turn however busy it stays; the leader skips it meanwhile. Each waiter
- holds a pool connection, so the leader runs one per eight pool
- connections, at least one and at most four; a further contended lane is
- retried on the next tick. Batch selection skips rows another transaction
- has locked, so locked due rows cannot fill every batch of a tick. `awa.maintenance.promote.lane_contention`
- (`awa_maintenance_promote_lane_contention_total{awa_job_state,awa_job_queue,awa_promote_outcome}`)
- counts `handed_off`, `waiter_limit` and `waiter_timed_out`. Promoted rows
- now take lane sequence numbers in due order within their lane.
- `QueueStorage::promote_due` keeps its signature and promotes lane by lane
- without a lock timeout; `promote_due_lanes` and `promote_due_lane` expose
- the per-lane form.
-- **A queue lane is usable by every runtime role, whichever role creates it
- first.** A lane's enqueue and claim sequences belonged to the role that
- first enqueued into, claimed from or promoted into it, so with several
- runtime roles sharing a group role, any role other than the creator whose
- default privileges did not cover it failed with `permission denied for
- sequence`. That included the maintenance leader promoting into a shard
- created after `enqueue_shards` was raised. The new schema patch
- `definer_lane_provisioning` creates lanes through two `SECURITY DEFINER`
- functions per queue-storage schema, `provision_lane` and
- `provision_lane_sequences`, owned by the schema owner. Each new sequence gets
- the privileges of the head table it backs, explicitly rather than through
- default privileges. A runtime without schema `CREATE` can therefore create
- lanes itself, and `awa storage prepare-queue` is optional and may run as any
- runtime role. Once the patch commits, `awa migrate` gives existing lane
- sequences to the schema owner and adds their missing grants, one sequence
- per transaction with a 300 ms lock wait, and only where it is allowed to,
- trying the effective role and then the login role. A sequence an open
- transaction holds is skipped and reported as busy; one it may not change is
- reported with the roles that can repair it. Neither fails the run. The new read-only `awa storage lane-sequences` (or
- `awa.lane_sequence_problems(schema)`) lists what remains, with the
- statements to run and the roles to run them as, and exits 2 while any
- remain; `--repair` (or `awa.repair_lane_sequence(schema, sequence)` per
- sequence) repairs what the connected role may. `awa migrate` prints a summary of the leftovers. A binary
- running before the patch detects that `provision_lane` is missing and
- creates lanes as before, and older binaries create correctly owned lanes once
- the patch has refreshed the helpers they call. `awa migrate` applies the
- patch on any v040 database; external runners apply the exported
- `R__definer_lane_provisioning.sql` after `V40`. See
- [`docs/upgrade-0.6.13-to-0.6.14.md`](docs/upgrade-0.6.13-to-0.6.14.md).
-- **A graceful shutdown no longer leaves a finished attempt for stale-heartbeat
- rescue to run again.** Once a handler returned, its finalize (completion,
- retry, cancel or failure, and any ADR-029 follow-up inserts) ran in a task
- that `Client::shutdown` did not wait for. When the last handler returned
- just as the drain ended, shutdown stopped the completion batcher before
- that finalize reached it, and the direct fallback raced the process exit.
- The attempt had already left the heartbeat set, so it went stale and was
- rescued and executed again `heartbeat_staleness` later; a deploy could
- re-run a few jobs on every rolling restart. Shutdown now waits for
- pending finalization, within the same drain timeout, before it stops the
- completion batcher, heartbeat and maintenance services. A finalize that
- outlives the timeout is still left to rescue, as before.
-- **Point reads show the live queue and priority after a lane move.** Retained
- ready rows from before a priority or queue change could win a read's tie
- after the ring wrapped, returning the old fields even though the batch
- operation succeeded. Point reads now exclude tombstoned source rows.
-
-- **Callback completion, `get_job`, lifecycle events and the admin reads
- find running jobs that priority aging promoted on queue storage** (backport
- of #600). Claims and leases record a job's aged effective priority while its
- ready row keeps the lane priority, and eight reads joined the two on
- priority. With aging on (the default, every 60 s), a job that waited past a
- minute before it was claimed could not be completed by callback with an
- expected kind, failed the mixed-transition callback routing check, read back
- from `get_job` in the wrong state, was missing from the job list, the kinds
- page and the dashboard's recent-jobs counts, and was not found by
- `cancel_by_unique_key`. They now match the attempt (`job_id` and `run_lease`). No
- migration.
-- **`get_job` no longer misses a job that moves while it reads.** It read
- ready, deferred, leased, claimed, completed and DLQ rows in separate
- statements, so a job promoted, claimed or completed between two of them
- could read as missing. It now reads all of them from one read-only snapshot.
-- **`admin::job_kind_overviews` on queue storage no longer depends on the
- planner's row estimates.** Behind the admin UI's kinds page, it counted
- every job through the `current_jobs` view. Two nodes of that view are
- estimated at one row whatever the statistics say: the join from completion
- batches to `ready_entries`, and the `UNION ALL` over every place a job can
- live. Postgres therefore chose nested loops that visit each retained row,
- and the join on `queue_claim_heads` was only as good as that table's
- estimate, which autovacuum never refreshes. It now aggregates each place to
- `(kind, queue)` on its own and combines the results: available jobs are
- found per lane from the claim cursor, and a completion batch's members are
- unnested and probed along the ready primary key. The rows are identical,
- and a test compares them with the former definition across every job
- state. With 2.6 million retained ready rows (1.8 million of them available)
- and 860,000 retained completions it takes about 6 to 8 s instead of 20 s on
- PostgreSQL 17 and 18. It still reads each available job and each retained
- completion's ready row, because a job's kind lives there. JIT is off for
- the read.
-- **`admin::list_jobs` on queue storage reads only the places that can hold
- the requested state.** It selected candidate ids from the `current_jobs`
- view, so every listing, whatever its filter, planned around the same
- single-row estimates and joined each retained completion to a scan of the
- ready rows. On PostgreSQL 18 a kind filter ran past a four-minute statement
- timeout. Each place a job can live is now read and limited on its own,
- only those that can hold the state asked for, with a predicate for each
- filter that is set rather than `$n IS NULL OR` terms, which a prepared
- statement's generic plan can't drop. Completion batches are ordered by job
- id before their ready rows are probed, so a page probes only as many as it
- returns. The ids are identical, and a test compares them with the former
- definition for every state, kind, queue, limit and page. On the dataset
- above, `state=failed&limit=10` drops from 12 s to 0.7 s,
- `state=completed&limit=100` from 20 s to 6 s, and an unfiltered or
- kind-filtered page of 100 from 23 s to 11 s (a kind filter on PostgreSQL 18
- from more than four minutes). Available jobs and completions are still read
- in full when the state allows them. JIT is off for the read.
+**Lane ownership**
+
+- A queue lane created by one runtime role is usable by every other runtime role. Before, its sequences belonged to whichever role created it, so other roles could fail with `permission denied for sequence`, for example after `enqueue_shards` was raised. New lanes now belong to the schema owner with the head table's grants, and migration repairs existing ones, including when the migration role can only `SET ROLE` to the schema owner.
+
+**Promotion, prune and lock contention**
+
+- A busy queue lane no longer delays promotion of other queues or maintenance. Promotion waits 50 ms for a lane's locks and hands a busy lane to a dedicated waiter.
+- A lock timeout anywhere in prune or terminal rollup rolls the attempt back and reports `Blocked` instead of an error, and prune backs off.
+- Admin job details, lists and kind counts release partition locks between read phases, so they block prune less, and lists of retained history do less repeated work.
+- A read cancelled while starting its transaction closes its connection, so the next caller cannot inherit an open transaction and its partition locks.
+
+**Job correctness after priority aging and recovery**
+
+- Physical priority aging no longer moves a ready row that is being claimed. Before, it could tombstone the body of a just-claimed job, after which completion, cancellation and rescue failed and the claim ring stayed pinned. Bodies left tombstoned by older releases are recovered when the job is read, completed, cancelled or rescued.
+- Reads and callbacks find attempts that priority aging or a priority change moved, and admin lists and kind counts include running jobs whose claims have not yet become leases.
+- Failed jobs keep their arguments, attempt limit and lane in terminal history; manual retries and DLQ moves keep the correct priority and lane.
+- Schema-only admin clients work with ready rings larger than the default.
+
+**Workers**
+
+- Workers retry lock contention during finalization for up to 120 s with the handler's saved result, keeping the job's heartbeat until the outcome settles.
+- Graceful shutdown waits for pending finalization before stopping heartbeat and maintenance, within the existing drain timeout.
+- A completion or callback wait that has committed is no longer reported as failed if re-reading the job for its lifecycle event fails.
+
+**CLI**
+
+- `awa migrate` rejects `--sql` combined with `--extract-to`. Before, `--sql` silently won and nothing was extracted.
+
+### Since 0.6.14-alpha.1
+
+For testers of the prerelease, these fix behaviour that alpha.1 introduced:
+
+- Promotion no longer holds a lane's locks while waiting for PostgreSQL's cluster-wide NOTIFY lock at commit, where under concurrent enqueues the 50 ms lane timeout could expire and roll the batch back. Claimers are woken once after the lanes commit.
+- With lifecycle hooks registered, a completion builds its event from the job the worker holds instead of re-reading it, which cost about 24 round trips per completion.
+- Rescue skips an attempt with ambiguous retained bodies and logs an error on each pass, instead of failing the whole rescue.
+
+### Added
+
+- `ClientBuilder::prune_lock_timeout` in Rust and `queue_storage_prune_lock_timeout_ms` in Python configure prune's lock wait. The timeout applies to each acquisition; queue prune can wait on up to eight partition locks. The default remains 10 ms. `awa.maintenance.prune.attempts` includes the outcomes `backed_off` and `error`.
+- `awa.maintenance.promote.lane_contention` records promotion lane contention and waiter outcomes by queue.
+- Multi-runtime soak coverage for killed workers, shard changes, mixed Rust and Python workloads, durable recovery evidence and ring-lock diagnostics.
+
+### Changed
+
+- Python CI tests 3.14 and 3.15. Older interpreter versions remain best-effort; package classifiers list 3.14 and 3.15.
+- Operator guidance explains that longer rotation intervals reduce churn while increasing rows and index size per slot and extending retained history.
+
+### Known limitations
+
+- A job-detail read can race prune and briefly show a completed job as available or missing; reading again returns the current state.
+- If an older release's aging race left two equally plausible bodies for one running attempt, reads and direct transitions of that attempt fail with an ambiguity error, and rescue skips it until the bodies are repaired.
+- A worker completing a job with the body it already holds does not retire an unclaimed copy left by older aging, so that copy can run again (at-least-once).
+- Direct API calls that recover a body left by older aging (external callback complete, fail and retry; admin cancel and retry; completion by ID) can fail with SQLSTATE 55P03 under contention and must be retried. Workers retry automatically.
+- The installed `terminal_jobs` SQL view still reports the stored priority for older rows whose priority changed.
+
+## [0.6.14-alpha.1] — 2026-11-07
+
+Prerelease of 0.6.14; every change is listed under 0.6.14.
## [0.6.13] — 2026-11-06
diff --git a/Cargo.lock b/Cargo.lock
index 4386517ed..656f4a77d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -383,7 +383,7 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
[[package]]
name = "awa"
-version = "0.6.13"
+version = "0.6.14"
dependencies = [
"async-trait",
"awa-macros",
@@ -409,7 +409,7 @@ dependencies = [
[[package]]
name = "awa-cli"
-version = "0.6.13"
+version = "0.6.14"
dependencies = [
"assert_cmd",
"awa-model",
@@ -431,7 +431,7 @@ dependencies = [
[[package]]
name = "awa-macros"
-version = "0.6.13"
+version = "0.6.14"
dependencies = [
"proc-macro-crate",
"proc-macro2",
@@ -441,7 +441,7 @@ dependencies = [
[[package]]
name = "awa-metrics"
-version = "0.6.13"
+version = "0.6.14"
dependencies = [
"awa-model",
"opentelemetry",
@@ -450,7 +450,7 @@ dependencies = [
[[package]]
name = "awa-model"
-version = "0.6.13"
+version = "0.6.14"
dependencies = [
"awa-macros",
"blake3",
@@ -471,7 +471,7 @@ dependencies = [
[[package]]
name = "awa-seaorm"
-version = "0.6.13"
+version = "0.6.14"
dependencies = [
"awa",
"awa-testing",
@@ -484,7 +484,7 @@ dependencies = [
[[package]]
name = "awa-testing"
-version = "0.6.13"
+version = "0.6.14"
dependencies = [
"async-trait",
"awa-model",
@@ -499,7 +499,7 @@ dependencies = [
[[package]]
name = "awa-ui"
-version = "0.6.13"
+version = "0.6.14"
dependencies = [
"awa-metrics",
"awa-model",
@@ -522,7 +522,7 @@ dependencies = [
[[package]]
name = "awa-worker"
-version = "0.6.13"
+version = "0.6.14"
dependencies = [
"anyhow",
"async-trait",
diff --git a/Cargo.toml b/Cargo.toml
index eba7da12b..2121e4041 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -14,7 +14,7 @@ members = [
exclude = ["awa-python", "spike"]
[workspace.package]
-version = "0.6.13"
+version = "0.6.14"
edition = "2021"
license = "MIT OR Apache-2.0"
description = "Postgres-native background job queue — transactional enqueue, heartbeat crash recovery, SKIP LOCKED dispatch"
@@ -25,11 +25,11 @@ categories = ["database", "asynchronous", "web-programming"]
[workspace.dependencies]
# Internal crates
-awa-model = { path = "awa-model", version = "0.6.13" }
-awa-macros = { path = "awa-macros", version = "0.6.13" }
-awa-metrics = { path = "awa-metrics", version = "0.6.13" }
-awa-worker = { path = "awa-worker", version = "0.6.13" }
-awa = { path = "awa", version = "0.6.13" }
+awa-model = { path = "awa-model", version = "0.6.14" }
+awa-macros = { path = "awa-macros", version = "0.6.14" }
+awa-metrics = { path = "awa-metrics", version = "0.6.14" }
+awa-worker = { path = "awa-worker", version = "0.6.14" }
+awa = { path = "awa", version = "0.6.14" }
# Database
sea-orm = { version = "=2.0.0-rc.38", default-features = false, features = ["sqlx-postgres", "runtime-tokio-rustls", "with-chrono", "with-json"] }
diff --git a/awa-cli/Cargo.toml b/awa-cli/Cargo.toml
index 14f235f7b..da4141211 100644
--- a/awa-cli/Cargo.toml
+++ b/awa-cli/Cargo.toml
@@ -14,7 +14,7 @@ path = "src/main.rs"
[dependencies]
awa-model.workspace = true
awa-worker.workspace = true
-awa-ui = { path = "../awa-ui", version = "0.6.13" }
+awa-ui = { path = "../awa-ui", version = "0.6.14" }
axum.workspace = true
sqlx.workspace = true
tokio.workspace = true
@@ -28,5 +28,5 @@ hex.workspace = true
uuid.workspace = true
[dev-dependencies]
-awa-testing = { path = "../awa-testing", version = "0.6.13" }
+awa-testing = { path = "../awa-testing", version = "0.6.14" }
assert_cmd = "2"
diff --git a/awa-cli/pyproject.toml b/awa-cli/pyproject.toml
index 7e4f95bbf..8b3be9b01 100644
--- a/awa-cli/pyproject.toml
+++ b/awa-cli/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "maturin"
[project]
name = "awa-cli"
-version = "0.6.13"
+version = "0.6.14"
description = "CLI for the Awa Postgres-native job queue (migrations, admin, serve)"
readme = "README.md"
requires-python = ">=3.10"
diff --git a/awa-cli/src/main.rs b/awa-cli/src/main.rs
index 1e347be53..47ee732a4 100644
--- a/awa-cli/src/main.rs
+++ b/awa-cli/src/main.rs
@@ -29,7 +29,7 @@ enum Commands {
#[arg(long)]
extract_to: Option,
/// Print migration SQL to stdout instead of applying
- #[arg(long)]
+ #[arg(long, conflicts_with = "extract_to")]
sql: bool,
/// Only include migrations after this version (exclusive)
#[arg(long)]
diff --git a/awa-model/migrations/v023_install_queue_storage_substrate.sql b/awa-model/migrations/v023_install_queue_storage_substrate.sql
index 4625ebada..57f56c7f7 100644
--- a/awa-model/migrations/v023_install_queue_storage_substrate.sql
+++ b/awa-model/migrations/v023_install_queue_storage_substrate.sql
@@ -511,13 +511,12 @@ AS $problems$
AND NOT candidate.rolsuper
AND pg_has_role(candidate.oid, lane_sequence.relowner, 'USAGE')
AND (lane_sequence.relowner = lane_sequence.nspowner
- OR (pg_has_role(candidate.oid, lane_sequence.nspowner, 'USAGE')
- AND pg_has_role(
+ OR pg_has_role(
candidate.oid,
lane_sequence.nspowner,
CASE WHEN current_setting('server_version_num')::int >= 160000
THEN 'SET' ELSE 'MEMBER' END
- )))
+ ))
ORDER BY candidate.rolname
)
FROM lane_sequence
@@ -555,6 +554,9 @@ AS $repair$
DECLARE
v_problem RECORD;
v_remaining RECORD;
+ v_original_role TEXT := current_setting('role');
+ v_reassigned BOOLEAN := false;
+ v_role_changed BOOLEAN := false;
v_busy BOOLEAN := false;
v_set_mode TEXT := CASE WHEN current_setting('server_version_num')::int >= 160000
THEN 'SET' ELSE 'MEMBER' END;
@@ -576,6 +578,7 @@ BEGIN
AND pg_has_role(current_user, to_regrole(quote_ident(v_problem.expected_owner)), v_set_mode) THEN
BEGIN
EXECUTE v_problem.reassign_sql;
+ v_reassigned := true;
EXCEPTION WHEN insufficient_privilege THEN
NULL;
END;
@@ -585,14 +588,29 @@ BEGIN
FROM awa.lane_sequence_problems(p_schema, p_sequence) AS problem;
IF FOUND
AND cardinality(v_remaining.missing_grants) > 0
- AND pg_has_role(current_user, to_regrole(quote_ident(v_remaining.owner)), 'USAGE') THEN
- FOREACH v_statement IN ARRAY v_remaining.missing_grants LOOP
- BEGIN
- EXECUTE v_statement;
- EXCEPTION WHEN insufficient_privilege THEN
- NULL;
- END;
- END LOOP;
+ AND (pg_has_role(current_user, to_regrole(quote_ident(v_remaining.owner)), 'USAGE')
+ OR v_reassigned) THEN
+ BEGIN
+ IF v_reassigned
+ AND NOT pg_has_role(current_user, to_regrole(quote_ident(v_remaining.owner)), 'USAGE') THEN
+ PERFORM set_config('role', v_remaining.owner, true);
+ v_role_changed := true;
+ END IF;
+
+ FOREACH v_statement IN ARRAY v_remaining.missing_grants LOOP
+ BEGIN
+ EXECUTE v_statement;
+ EXCEPTION WHEN insufficient_privilege THEN
+ NULL;
+ END;
+ END LOOP;
+
+ IF v_role_changed THEN
+ PERFORM set_config('role', v_original_role, true);
+ END IF;
+ EXCEPTION WHEN insufficient_privilege THEN
+ NULL;
+ END;
END IF;
EXCEPTION WHEN lock_not_available THEN
v_busy := true;
diff --git a/awa-model/src/admin.rs b/awa-model/src/admin.rs
index a40efb4e5..296e533de 100644
--- a/awa-model/src/admin.rs
+++ b/awa-model/src/admin.rs
@@ -9,7 +9,7 @@ use sqlx::Acquire;
use sqlx::PgExecutor;
use sqlx::PgPool;
use std::cmp::max;
-use std::collections::HashMap;
+use std::collections::{BTreeSet, HashMap};
use std::fmt;
use std::str::FromStr;
use uuid::Uuid;
@@ -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
@@ -802,48 +961,42 @@ fn queue_storage_recent_jobs_cte(schema: &str, since: &str) -> String {
)
}
-/// The per-kind, per-queue counts behind [`job_kind_overviews`], as a
-/// `kind_counts` CTE of `kind`, `job_count`, `queue_count` and
-/// `completed_last_hour`.
-///
-/// It counts what `current_jobs` holds without expanding it. Each place a job
-/// can live is aggregated to `(kind, queue)` on its own, as a `MATERIALIZED`
-/// CTE, and only those few rows are combined. Read through `current_jobs`,
-/// the receipt-batch join to `ready_entries` and the `UNION ALL` are both
-/// estimated at one row whatever the statistics say, so Postgres chooses
-/// nested loops that probe once per retained row, and a join on
-/// `queue_claim_heads` is only as good as that table's estimate.
-///
-/// Available rows are found per lane as a claim finds them (see
-/// [`queue_storage_available_by_queue_sql`]). A completion batch
-/// member's kind lives on its ready row, so that leg unnests each batch and
-/// probes the ready primary key.
-fn queue_storage_kind_counts_cte(schema: &str) -> String {
+// Per-source kind/queue counts keep the same joins and predicates as the
+// combined projection, while allowing each source's relation locks to end
+// before another source opens a different set of retained partitions.
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum KindCountSource {
+ Available,
+ Leased,
+ Deferred,
+ Done,
+ Receipts,
+ Dlq,
+}
+
+// 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");
- format!(
- r#"
- heads AS MATERIALIZED (
- 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
- ),
- available_by_kind AS MATERIALIZED (
+ match source {
+ KindCountSource::Available => format!(
+ r#"
+ WITH heads AS MATERIALIZED (
+ SELECT queue, priority, enqueue_shard, claim_seq, next_seq
+ FROM unnest(
+ $1::text[], $2::smallint[], $3::smallint[], $4::bigint[], $5::bigint[]
+ ) AS captured(queue, priority, enqueue_shard, claim_seq, next_seq)
+ )
SELECT lane.kind, heads.queue, count(*)::bigint AS n, 0::bigint AS n_recent
FROM heads
CROSS JOIN LATERAL (
{available}
) AS lane
GROUP BY lane.kind, heads.queue
+ "#
),
- leased_by_kind AS MATERIALIZED (
+ KindCountSource::Leased => format!(
+ r#"
SELECT ready.kind, leases.queue, count(*)::bigint AS n, 0::bigint AS n_recent
FROM {schema}.leases AS leases
JOIN {schema}.ready_entries AS ready
@@ -856,14 +1009,27 @@ 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 (
+ KindCountSource::Deferred => format!(
+ r#"
SELECT kind, queue, count(*)::bigint AS n, 0::bigint AS n_recent
FROM {schema}.deferred_jobs
GROUP BY kind, queue
+ "#
),
- done_by_kind AS MATERIALIZED (
+ KindCountSource::Done => format!(
+ r#"
SELECT
kind,
queue,
@@ -874,8 +1040,10 @@ fn queue_storage_kind_counts_cte(schema: &str) -> String {
)::bigint AS n_recent
FROM {schema}.done_entries
GROUP BY kind, queue
+ "#
),
- receipts_by_kind AS MATERIALIZED (
+ KindCountSource::Receipts => format!(
+ r#"
SELECT
ready.kind,
batch.queue,
@@ -889,14 +1057,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
@@ -905,22 +1079,25 @@ fn queue_storage_kind_counts_cte(schema: &str) -> String {
AND tomb.run_lease = item.run_lease
)
GROUP BY ready.kind, batch.queue
+ "#
),
- dlq_by_kind AS MATERIALIZED (
+ KindCountSource::Dlq => format!(
+ r#"
SELECT kind, queue, count(*)::bigint AS n, 0::bigint AS n_recent
FROM {schema}.dlq_entries
GROUP BY kind, queue
+ "#
),
+ }
+}
+
+// Merge small (kind, queue) aggregates from separately locked read sources.
+fn queue_storage_kind_counts_cte() -> &'static str {
+ r#"
kind_queue AS (
SELECT kind, queue, sum(n)::bigint AS n, sum(n_recent)::bigint AS n_recent
- FROM (
- SELECT * FROM available_by_kind
- UNION ALL SELECT * FROM leased_by_kind
- UNION ALL SELECT * FROM deferred_by_kind
- UNION ALL SELECT * FROM done_by_kind
- UNION ALL SELECT * FROM receipts_by_kind
- UNION ALL SELECT * FROM dlq_by_kind
- ) AS legs
+ FROM unnest($1::text[], $2::text[], $3::bigint[], $4::bigint[])
+ AS sources(kind, queue, n, n_recent)
GROUP BY kind, queue
),
kind_counts AS (
@@ -932,8 +1109,7 @@ fn queue_storage_kind_counts_cte(schema: &str) -> String {
FROM kind_queue
GROUP BY kind
)
- "#
- )
+ "#
}
async fn notify_cancellation_tx<'a>(
@@ -949,14 +1125,21 @@ async fn notify_cancellation_tx<'a>(
Ok(())
}
-/// Candidate job ids for [`list_jobs`]: the newest `limit` ids below `cursor`
-/// matching `filter`'s state, kind and queue, newest first, with the binds the
-/// statement takes in order.
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum JobCandidateSource {
+ States,
+ Receipts(usize),
+}
+
+/// Candidate job ids for [`list_jobs`] from one source: the newest `limit`
+/// distinct ids below `cursor`, with the statement's binds in order. State
+/// sources apply kind/state/queue filters. Receipt sources enumerate open
+/// claims in one physical slot and apply kind/queue before their limit.
///
-/// It selects what `current_jobs` would with the same filter, but each place a
-/// job can live is read and limited on its own, with only the legs that can
-/// hold the requested state, and the filter written into the statement rather
-/// than as `$n IS NULL OR` terms that a generic plan can't drop. Read through
+/// Each place a job can live is read and limited on its own, with only the
+/// state legs that can hold the requested state. Known filters are written
+/// into the statement rather than as `$n IS NULL OR` terms that a generic plan
+/// can't drop. Read through
/// `current_jobs`, Postgres joins every retained completion to a scan of the
/// ready rows, because that join and the `UNION ALL` are each estimated at one
/// row. Available rows are found per lane as a claim finds them, and a
@@ -968,13 +1151,17 @@ fn queue_storage_job_id_candidates_sql(
filter: &ListJobsFilter,
cursor: Option,
limit: i64,
+ source: JobCandidateSource,
) -> (String, Vec) {
let mut binds = Vec::new();
let mut bind = |value: CandidateBind| {
binds.push(value);
format!("${}", binds.len())
};
- let state = filter.state.map(|state| bind(CandidateBind::State(state)));
+ let state = filter
+ .state
+ .filter(|_| source == JobCandidateSource::States)
+ .map(|state| bind(CandidateBind::State(state)));
let kind = filter
.kind
.as_ref()
@@ -1008,9 +1195,10 @@ fn queue_storage_job_id_candidates_sql(
sql
};
let holds = |states: &[JobState]| {
- filter
- .state
- .is_none_or(|requested| states.contains(&requested))
+ source == JobCandidateSource::States
+ && filter
+ .state
+ .is_none_or(|requested| states.contains(&requested))
};
let mut legs: Vec = Vec::new();
@@ -1049,7 +1237,7 @@ fn queue_storage_job_id_candidates_sql(
);
legs.push(format!(
r#"(
- SELECT lane.job_id
+ SELECT DISTINCT lane.job_id
FROM heads
CROSS JOIN LATERAL ({lane_rows}) AS lane
ORDER BY lane.job_id DESC
@@ -1061,7 +1249,7 @@ fn queue_storage_job_id_candidates_sql(
let predicates = predicates(Some("state"), Some("kind"), "queue", "job_id");
legs.push(format!(
r#"(
- SELECT job_id
+ SELECT DISTINCT job_id
FROM {schema}.deferred_jobs
WHERE TRUE{predicates}
ORDER BY job_id DESC
@@ -1078,7 +1266,7 @@ fn queue_storage_job_id_candidates_sql(
);
legs.push(format!(
r#"(
- SELECT leases.job_id
+ SELECT DISTINCT leases.job_id
FROM {schema}.leases AS leases
JOIN {schema}.ready_entries AS ready
ON ready.ready_slot = leases.ready_slot
@@ -1090,17 +1278,62 @@ 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 let JobCandidateSource::Receipts(slot) = source {
+ let open_running =
+ crate::queue_storage::open_receipt_running_claims_sql_in_slot(schema, slot);
+ let predicates = predicates(
+ None,
+ Some("created.kind"),
+ "open_claims.queue",
+ "open_claims.job_id",
+ );
+ legs.push(format!(
+ r#"(
+ SELECT DISTINCT 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!(
r#"(
- SELECT job_id
+ SELECT DISTINCT job_id
FROM {schema}.done_entries
WHERE TRUE{predicates}
ORDER BY job_id DESC
@@ -1116,7 +1349,7 @@ fn queue_storage_job_id_candidates_sql(
.unwrap_or_default();
legs.push(format!(
r#"(
- SELECT member.job_id
+ SELECT DISTINCT member.job_id
FROM (
SELECT
batch.ready_slot,
@@ -1165,7 +1398,7 @@ fn queue_storage_job_id_candidates_sql(
let predicates = predicates(None, Some("kind"), "queue", "job_id");
legs.push(format!(
r#"(
- SELECT job_id
+ SELECT DISTINCT job_id
FROM {schema}.dlq_entries
WHERE TRUE{predicates}
ORDER BY job_id DESC
@@ -1177,7 +1410,7 @@ fn queue_storage_job_id_candidates_sql(
let sql = format!(
r#"
{with}
- SELECT job_id
+ SELECT DISTINCT job_id
FROM ({legs}) AS candidates
ORDER BY job_id DESC
LIMIT {limit}
@@ -1196,6 +1429,34 @@ enum CandidateBind {
Int(i64),
}
+async fn fetch_queue_storage_job_candidates(
+ tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+ sql: &str,
+ binds: Vec,
+) -> Result, sqlx::Error> {
+ sqlx::query("SAVEPOINT awa_list_candidates")
+ .execute(tx.as_mut())
+ .await?;
+ let mut query = sqlx::query_scalar::<_, i64>(sql);
+ for value in binds {
+ query = match value {
+ CandidateBind::State(state) => query.bind(state),
+ CandidateBind::Text(text) => query.bind(text),
+ CandidateBind::Int(int) => query.bind(int),
+ };
+ }
+ let ids = query.fetch_all(tx.as_mut()).await?;
+ // Keep the outer repeatable-read snapshot, but release this read phase's
+ // relation locks. Destroy the savepoint too rather than stacking names.
+ sqlx::query("ROLLBACK TO SAVEPOINT awa_list_candidates")
+ .execute(tx.as_mut())
+ .await?;
+ sqlx::query("RELEASE SAVEPOINT awa_list_candidates")
+ .execute(tx.as_mut())
+ .await?;
+ Ok(ids)
+}
+
async fn list_queue_storage_jobs(
store: &QueueStorage,
pool: &PgPool,
@@ -1212,23 +1473,65 @@ async fn list_queue_storage_jobs(
let mut cursor = filter.before_id;
loop {
- let (sql, binds) =
- queue_storage_job_id_candidates_sql(store.schema(), filter, cursor, candidate_limit);
- let mut query = sqlx::query_scalar::<_, i64>(&sql);
- for value in binds {
- query = match value {
- CandidateBind::State(state) => query.bind(state),
- CandidateBind::Text(text) => query.bind(text),
- CandidateBind::Int(int) => query.bind(int),
- };
- }
- // As in `queue_overviews`: compiling costs more than running.
- let mut tx = pool.begin().await?;
+ // Every candidate phase sees the same MVCC snapshot. Each phase
+ // releases its own read locks before the next one opens more tables.
+ let mut begin_guard = crate::read_transaction::BeginGuard::acquire(pool).await?;
+ let mut tx = sqlx::Connection::begin_with(
+ &mut *begin_guard.connection,
+ "BEGIN ISOLATION LEVEL REPEATABLE READ, READ ONLY",
+ )
+ .await?;
+ begin_guard.begin_pending = false;
sqlx::query("SET LOCAL jit = off")
.execute(tx.as_mut())
.await?;
- let ids: Vec = query.fetch_all(tx.as_mut()).await?;
+ let states = filter.state.map_or_else(
+ || {
+ vec![
+ JobState::Available,
+ JobState::Scheduled,
+ JobState::Retryable,
+ JobState::Running,
+ JobState::WaitingExternal,
+ JobState::Completed,
+ JobState::Failed,
+ JobState::Cancelled,
+ ]
+ },
+ |state| vec![state],
+ );
+ let mut ids = BTreeSet::new();
+ for state in states {
+ let mut phase_filter = filter.clone();
+ phase_filter.state = Some(state);
+ let (sql, binds) = queue_storage_job_id_candidates_sql(
+ store.schema(),
+ &phase_filter,
+ cursor,
+ candidate_limit,
+ JobCandidateSource::States,
+ );
+ ids.extend(fetch_queue_storage_job_candidates(&mut tx, &sql, binds).await?);
+ }
+ if filter.state.is_none_or(|state| state == JobState::Running) {
+ for slot in queue_storage_populated_claim_slots(&mut tx, store).await? {
+ let (sql, binds) = queue_storage_job_id_candidates_sql(
+ store.schema(),
+ filter,
+ cursor,
+ candidate_limit,
+ JobCandidateSource::Receipts(slot),
+ );
+ ids.extend(fetch_queue_storage_job_candidates(&mut tx, &sql, binds).await?);
+ }
+ }
tx.commit().await?;
+ drop(begin_guard);
+ let ids: Vec = ids
+ .into_iter()
+ .rev()
+ .take(candidate_limit as usize)
+ .collect();
if ids.is_empty() {
break;
}
@@ -1552,6 +1855,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
@@ -3121,7 +3433,9 @@ pub async fn queue_overviews(pool: &PgPool) -> Result, AwaErr
);
// JIT compiles this statement for longer than it runs.
- let mut tx = pool.begin().await?;
+ let mut begin_guard = crate::read_transaction::BeginGuard::acquire(pool).await?;
+ let mut tx = sqlx::Connection::begin(&mut *begin_guard.connection).await?;
+ begin_guard.begin_pending = false;
sqlx::query("SET LOCAL jit = off")
.execute(tx.as_mut())
.await?;
@@ -3129,6 +3443,7 @@ pub async fn queue_overviews(pool: &PgPool) -> Result, AwaErr
.fetch_all(tx.as_mut())
.await?;
tx.commit().await?;
+ drop(begin_guard);
return Ok(rows);
}
@@ -3599,12 +3914,15 @@ pub async fn state_counts(pool: &PgPool) -> Result, AwaEr
waiting_external,
): (i64, i64, i64, i64, i64, i64, i64, i64) = {
// As in `queue_overviews`: compiling costs more than running.
- let mut tx = pool.begin().await?;
+ let mut begin_guard = crate::read_transaction::BeginGuard::acquire(pool).await?;
+ let mut tx = sqlx::Connection::begin(&mut *begin_guard.connection).await?;
+ begin_guard.begin_pending = false;
sqlx::query("SET LOCAL jit = off")
.execute(tx.as_mut())
.await?;
let counts = sqlx::query_as(&sql).fetch_one(tx.as_mut()).await?;
tx.commit().await?;
+ drop(begin_guard);
counts
};
@@ -3654,6 +3972,185 @@ pub async fn state_counts(pool: &PgPool) -> Result, AwaEr
Ok(rows.into_iter().collect())
}
+#[derive(sqlx::FromRow)]
+struct KindCountHead {
+ queue: String,
+ priority: i16,
+ enqueue_shard: i16,
+ claim_seq: i64,
+ next_seq: i64,
+}
+
+// Sequence values do not obey MVCC. Capture them with the first read's
+// snapshot instead of letting later claims advance them during receipt phases.
+async fn queue_storage_kind_heads(
+ tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+ store: &QueueStorage,
+) -> Result, sqlx::Error> {
+ let schema = store.schema();
+ sqlx::query("SAVEPOINT awa_kind_heads")
+ .execute(tx.as_mut())
+ .await?;
+ let heads = sqlx::query_as::<_, KindCountHead>(&format!(
+ r#"
+ 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
+ "#
+ ))
+ .fetch_all(tx.as_mut())
+ .await?;
+ sqlx::query("ROLLBACK TO SAVEPOINT awa_kind_heads")
+ .execute(tx.as_mut())
+ .await?;
+ sqlx::query("RELEASE SAVEPOINT awa_kind_heads")
+ .execute(tx.as_mut())
+ .await?;
+ Ok(heads)
+}
+
+async fn queue_storage_base_kind_counts(
+ tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+ store: &QueueStorage,
+ heads: Vec,
+) -> Result, sqlx::Error> {
+ let mut head_queues = Vec::with_capacity(heads.len());
+ let mut head_priorities = Vec::with_capacity(heads.len());
+ let mut head_shards = Vec::with_capacity(heads.len());
+ let mut claim_seqs = Vec::with_capacity(heads.len());
+ let mut next_seqs = Vec::with_capacity(heads.len());
+ for head in heads {
+ head_queues.push(head.queue);
+ head_priorities.push(head.priority);
+ head_shards.push(head.enqueue_shard);
+ claim_seqs.push(head.claim_seq);
+ next_seqs.push(head.next_seq);
+ }
+ let mut counts = Vec::new();
+ for source in [
+ KindCountSource::Available,
+ KindCountSource::Leased,
+ KindCountSource::Deferred,
+ KindCountSource::Done,
+ KindCountSource::Receipts,
+ KindCountSource::Dlq,
+ ] {
+ let sql = queue_storage_kind_count_source_sql(store.schema(), source);
+ sqlx::query("SAVEPOINT awa_kind_source")
+ .execute(tx.as_mut())
+ .await?;
+ let mut query = sqlx::query_as::<_, (String, String, i64, i64)>(&sql);
+ if source == KindCountSource::Available {
+ query = query
+ .bind(&head_queues)
+ .bind(&head_priorities)
+ .bind(&head_shards)
+ .bind(&claim_seqs)
+ .bind(&next_seqs);
+ }
+ counts.extend(query.fetch_all(tx.as_mut()).await?);
+ sqlx::query("ROLLBACK TO SAVEPOINT awa_kind_source")
+ .execute(tx.as_mut())
+ .await?;
+ sqlx::query("RELEASE SAVEPOINT awa_kind_source")
+ .execute(tx.as_mut())
+ .await?;
+ }
+ Ok(counts)
+}
+
+// The result is bounded by the configured claim-ring slot count, not by
+// retained job volume. Include all populated slots, including old generations.
+async fn queue_storage_populated_claim_slots(
+ tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+ store: &QueueStorage,
+) -> Result, AwaError> {
+ let schema = store.schema();
+ sqlx::query("SAVEPOINT awa_claim_slots")
+ .execute(tx.as_mut())
+ .await?;
+ let slots: Vec = sqlx::query_scalar(&format!(
+ "SELECT claim_slot FROM {schema}.lease_claims WHERE closed_at IS NULL \
+ UNION SELECT claim_slot FROM {schema}.lease_claim_batches ORDER BY claim_slot"
+ ))
+ .fetch_all(tx.as_mut())
+ .await?;
+ sqlx::query("ROLLBACK TO SAVEPOINT awa_claim_slots")
+ .execute(tx.as_mut())
+ .await?;
+ sqlx::query("RELEASE SAVEPOINT awa_claim_slots")
+ .execute(tx.as_mut())
+ .await?;
+ slots
+ .into_iter()
+ .map(|slot| {
+ usize::try_from(slot)
+ .ok()
+ .filter(|slot| *slot < store.claim_slot_count())
+ .ok_or_else(|| AwaError::Validation(format!("invalid physical claim slot {slot}")))
+ })
+ .collect()
+}
+
+// Keep only per-kind, per-queue aggregates in Rust, not every claim identity.
+async fn queue_storage_running_kind_counts(
+ tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+ store: &QueueStorage,
+) -> Result, AwaError> {
+ let schema = store.schema();
+ let slots = queue_storage_populated_claim_slots(tx, store).await?;
+ let mut counts = Vec::new();
+ for slot in slots {
+ let open_running =
+ crate::queue_storage::open_receipt_running_claims_sql_in_slot(schema, slot);
+ sqlx::query("SAVEPOINT awa_kind_counts")
+ .execute(tx.as_mut())
+ .await?;
+ let rows: Vec<(String, String, i64)> = sqlx::query_as(&format!(
+ r#"
+ SELECT created.kind, open_claims.queue, count(*)::bigint
+ 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
+ "#
+ ))
+ .fetch_all(tx.as_mut())
+ .await?;
+ sqlx::query("ROLLBACK TO SAVEPOINT awa_kind_counts")
+ .execute(tx.as_mut())
+ .await?;
+ sqlx::query("RELEASE SAVEPOINT awa_kind_counts")
+ .execute(tx.as_mut())
+ .await?;
+ counts.extend(rows);
+ }
+ Ok(counts)
+}
+
/// Get job-kind overviews for all known kinds.
pub async fn job_kind_overviews(pool: &PgPool) -> Result, AwaError> {
if let Some(store) = active_queue_storage(pool).await? {
@@ -3705,18 +4202,47 @@ pub async fn job_kind_overviews(pool: &PgPool) -> Result, A
LEFT JOIN live_kind_descriptor_variants kdv ON kdv.kind = k.kind
ORDER BY k.kind
"#,
- kind_counts_cte = queue_storage_kind_counts_cte(store.schema())
+ kind_counts_cte = queue_storage_kind_counts_cte()
);
// As in `queue_overviews`: compiling costs more than running.
- let mut tx = pool.begin().await?;
+ let mut begin_guard = crate::read_transaction::BeginGuard::acquire(pool).await?;
+ let mut tx = sqlx::Connection::begin_with(
+ &mut *begin_guard.connection,
+ "BEGIN ISOLATION LEVEL REPEATABLE READ, READ ONLY",
+ )
+ .await?;
+ begin_guard.begin_pending = false;
sqlx::query("SET LOCAL jit = off")
.execute(tx.as_mut())
.await?;
+ let heads = queue_storage_kind_heads(&mut tx, &store).await?;
+ let mut sources = queue_storage_base_kind_counts(&mut tx, &store, heads).await?;
+ sources.extend(
+ queue_storage_running_kind_counts(&mut tx, &store)
+ .await?
+ .into_iter()
+ .map(|(kind, queue, n)| (kind, queue, n, 0)),
+ );
+ let mut kinds = Vec::with_capacity(sources.len());
+ let mut queues = Vec::with_capacity(sources.len());
+ let mut counts = Vec::with_capacity(sources.len());
+ let mut recent_counts = Vec::with_capacity(sources.len());
+ for (kind, queue, count, recent) in sources {
+ kinds.push(kind);
+ queues.push(queue);
+ counts.push(count);
+ recent_counts.push(recent);
+ }
let rows = sqlx::query_as::<_, JobKindOverview>(&sql)
+ .bind(kinds)
+ .bind(queues)
+ .bind(counts)
+ .bind(recent_counts)
.fetch_all(tx.as_mut())
.await?;
tx.commit().await?;
+ drop(begin_guard);
return Ok(rows);
}
@@ -3995,7 +4521,9 @@ pub async fn state_timeseries(
ORDER BY bucket",
queue_storage_recent_jobs_cte(store.schema(), "$1")
);
- let mut tx = pool.begin().await?;
+ let mut begin_guard = crate::read_transaction::BeginGuard::acquire(pool).await?;
+ let mut tx = sqlx::Connection::begin(&mut *begin_guard.connection).await?;
+ begin_guard.begin_pending = false;
sqlx::query("SET LOCAL jit = off")
.execute(tx.as_mut())
.await?;
@@ -4012,6 +4540,7 @@ pub async fn state_timeseries(
.fetch_all(tx.as_mut())
.await?;
tx.commit().await?;
+ drop(begin_guard);
return Ok(rows
.into_iter()
diff --git a/awa-model/src/lib.rs b/awa-model/src/lib.rs
index 88d658d05..b2e69c45b 100644
--- a/awa-model/src/lib.rs
+++ b/awa-model/src/lib.rs
@@ -12,6 +12,7 @@ pub mod kind;
pub mod migrations;
pub mod partitioned_queue;
pub mod queue_storage;
+mod read_transaction;
pub mod reschedule;
pub mod storage;
pub mod unique;
diff --git a/awa-model/src/migrations.rs b/awa-model/src/migrations.rs
index d6b4a6734..549dcabef 100644
--- a/awa-model/src/migrations.rs
+++ b/awa-model/src/migrations.rs
@@ -307,7 +307,10 @@ macro_rules! definer_lane_provisioning_pending_sql {
WHERE to_regprocedure('awa.lane_sequence_problems(text,text)') IS NULL \
UNION ALL \
SELECT 'awa.repair_lane_sequence' \
- WHERE to_regprocedure('awa.repair_lane_sequence(text,text)') IS NULL \
+ WHERE NOT EXISTS ( \
+ SELECT 1 FROM pg_proc AS p \
+ WHERE p.oid = to_regprocedure('awa.repair_lane_sequence(text,text)') \
+ AND p.prosrc LIKE '%set_config(''role'', v_original_role, true)%') \
UNION ALL \
SELECT 'awa.refresh_lane_sequence_helpers' \
WHERE EXISTS ( \
diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs
index b253338d3..746cd60cf 100644
--- a/awa-model/src/queue_storage.rs
+++ b/awa-model/src/queue_storage.rs
@@ -6,7 +6,7 @@ use crate::{InsertParams, JobRow, JobState};
use chrono::TimeDelta;
use chrono::{DateTime, Utc};
use sqlx::Acquire;
-use sqlx::{PgPool, Postgres};
+use sqlx::{FromRow, PgPool, Postgres, Row};
use std::collections::hash_map::DefaultHasher;
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::hash::{Hash, Hasher};
@@ -380,8 +380,8 @@ pub struct PromoteOutcome {
/// contended lane. A batch that selected its full size may have left
/// further due rows behind.
pub selected: usize,
- /// Lanes whose promotion transaction hit `lock_timeout` and rolled back.
- /// Their rows stay deferred.
+ /// Lanes whose locks were held past `lock_timeout`, so their promotion
+ /// transaction rolled back. Their rows stay deferred.
pub contended: Vec,
}
@@ -539,20 +539,84 @@ fn terminal_delta_child_name(schema: &str, slot: usize) -> String {
}
fn done_ready_join(schema: &str, done_alias: &str, ready_alias: &str) -> String {
+ done_ready_join_from(
+ done_alias,
+ ready_alias,
+ &format!("{schema}.ready_entries"),
+ &format!("{schema}.ready_tombstones"),
+ )
+}
+
+// Old writers could store effective priority in a narrow terminal row. Prefer
+// the live body of that exact claimed attempt, while retaining same-lease,
+// exact-lane bodies tombstoned by an available-job cancellation. Choose one
+// body so moved source rows cannot multiply terminal mutations or counters.
+fn done_ready_join_from(
+ done_alias: &str,
+ ready_alias: &str,
+ ready_relation: &str,
+ tombstone_relation: &str,
+) -> String {
format!(
r#"
- LEFT JOIN {schema}.ready_entries AS {ready_alias}
- ON {ready_alias}.ready_slot = {done_alias}.ready_slot
- AND {ready_alias}.ready_generation = {done_alias}.ready_generation
- AND {ready_alias}.queue = {done_alias}.queue
- AND {ready_alias}.priority = {done_alias}.priority
- AND {ready_alias}.enqueue_shard = {done_alias}.enqueue_shard
- AND {ready_alias}.lane_seq = {done_alias}.lane_seq
+ LEFT JOIN LATERAL (
+ SELECT candidate.*
+ FROM (
+ SELECT body.*,
+ (
+ body.run_lease + 1 = {done_alias}.run_lease
+ AND NOT EXISTS (
+ SELECT 1 FROM {tombstone_relation} AS tomb
+ WHERE tomb.ready_slot = body.ready_slot
+ AND tomb.ready_generation = body.ready_generation
+ AND tomb.queue = body.queue
+ AND tomb.priority = body.priority
+ AND tomb.enqueue_shard = body.enqueue_shard
+ AND tomb.lane_seq = body.lane_seq
+ AND tomb.job_id = body.job_id
+ )
+ ) AS live_attempt_body
+ FROM {ready_relation} AS body
+ WHERE body.ready_slot = {done_alias}.ready_slot
+ AND body.ready_generation = {done_alias}.ready_generation
+ AND body.queue = {done_alias}.queue
+ AND body.enqueue_shard = {done_alias}.enqueue_shard
+ AND body.lane_seq = {done_alias}.lane_seq
+ AND body.job_id = {done_alias}.job_id
+ AND (body.run_lease = {done_alias}.run_lease
+ OR body.run_lease + 1 = {done_alias}.run_lease)
+ ) AS candidate
+ WHERE candidate.priority = {done_alias}.priority
+ OR candidate.live_attempt_body
+ ORDER BY candidate.live_attempt_body DESC,
+ (candidate.priority = {done_alias}.priority) DESC,
+ candidate.priority
+ LIMIT 1
+ ) AS {ready_alias} ON TRUE
"#
)
}
+fn done_payload_projection(done_alias: &str, ready_alias: &str) -> String {
+ // A narrow old terminal payload can lack metadata from its live moved
+ // body. Supply only metadata defaults; terminal errors/tags/progress and
+ // explicit metadata overrides remain authoritative. Keep default {} elision.
+ format!(
+ r#"CASE
+ WHEN {done_alias}.args IS NULL
+ AND {ready_alias}.live_attempt_body
+ AND COALESCE({ready_alias}.payload -> 'metadata', '{{}}'::jsonb) <> '{{}}'::jsonb
+ THEN COALESCE({done_alias}.payload, {ready_alias}.payload, '{{}}'::jsonb)
+ || jsonb_build_object('metadata',
+ COALESCE({ready_alias}.payload -> 'metadata', '{{}}'::jsonb)
+ || COALESCE({done_alias}.payload -> 'metadata', '{{}}'::jsonb))
+ ELSE COALESCE({done_alias}.payload, {ready_alias}.payload, '{{}}'::jsonb)
+ END"#
+ )
+}
+
fn done_row_projection(done_alias: &str, ready_alias: &str) -> String {
+ let payload = done_payload_projection(done_alias, ready_alias);
format!(
r#"
{done_alias}.ready_slot,
@@ -574,7 +638,7 @@ fn done_row_projection(done_alias: &str, ready_alias: &str) -> String {
COALESCE({done_alias}.created_at, {ready_alias}.created_at, {done_alias}.finalized_at) AS created_at,
COALESCE({done_alias}.unique_key, {ready_alias}.unique_key) AS unique_key,
COALESCE({done_alias}.unique_states, {ready_alias}.unique_states) AS unique_states,
- COALESCE({done_alias}.payload, {ready_alias}.payload, '{{}}'::jsonb) AS payload
+ {payload} AS payload
"#
)
}
@@ -609,12 +673,16 @@ fn claim_closure_batch_child_name(schema: &str, slot: usize) -> String {
format!("{schema}.lease_claim_closure_batches_{slot}")
}
-fn ring_slot_index(slot: i32, slot_count: usize, ring: &str) -> Result {
- let index = usize::try_from(slot).map_err(|_| {
+fn ring_slot_number(slot: i32, ring: &str) -> Result {
+ usize::try_from(slot).map_err(|_| {
AwaError::Validation(format!(
"invalid {ring} ring slot {slot}: slot must be non-negative"
))
- })?;
+ })
+}
+
+fn ring_slot_index(slot: i32, slot_count: usize, ring: &str) -> Result {
+ let index = ring_slot_number(slot, ring)?;
if index >= slot_count {
return Err(AwaError::Validation(format!(
"invalid {ring} ring slot {slot}: configured slot count is {slot_count}"
@@ -623,42 +691,76 @@ fn ring_slot_index(slot: i32, slot_count: usize, ring: &str) -> Result 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_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,
+ // but hot membership checks use the per-child GiST index on the multirange.
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.receipt_ranges @> {claims_alias}.receipt_id
+ WHERE closure_batches.claim_slot = {claims_alias}.claim_slot
+ 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
@@ -670,12 +772,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
@@ -699,10 +800,41 @@ pub(crate) fn open_receipt_running_claims_sql(schema: &str) -> String {
/// retained batch header is read, but only batches holding one of the jobs
/// are unnested and checked.
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");
+ open_receipt_running_claims_sql_from(schema, job_ids, None)
+}
+
+/// The same open-claim projection, scoped to one validated physical claim slot.
+/// Ready and terminal history are deliberately not restricted to that slot.
+pub(crate) fn open_receipt_running_claims_sql_in_slot(schema: &str, slot: usize) -> String {
+ open_receipt_running_claims_sql_from(schema, None, Some(slot))
+}
+
+fn open_receipt_running_claims_sql_from(
+ schema: &str,
+ job_ids: Option<&str>,
+ slot: Option,
+) -> String {
+ let (claims_rel, batches_rel, closure_rel, closure_batch_rel) = match slot {
+ Some(slot) => (
+ claim_child_name(schema, slot),
+ claim_batch_child_name(schema, slot),
+ closure_child_name(schema, slot),
+ claim_closure_batch_child_name(schema, slot),
+ ),
+ None => (
+ format!("{schema}.lease_claims"),
+ format!("{schema}.lease_claim_batches"),
+ format!("{schema}.lease_claim_closures"),
+ format!("{schema}.lease_claim_closure_batches"),
+ ),
+ };
+ 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 "),
@@ -713,8 +845,8 @@ pub(crate) fn open_receipt_running_claims_sql_for(schema: &str, job_ids: Option<
format!(
r#"
SELECT claims.job_id, claims.run_lease, claims.queue
- FROM {schema}.lease_claims AS claims
- WHERE {row_filter}NOT {row_closed}
+ FROM {claims_rel} AS claims
+ WHERE {row_filter}{row_open}
AND NOT EXISTS (
SELECT 1 FROM {schema}.leases AS lease
WHERE lease.job_id = claims.job_id
@@ -722,7 +854,7 @@ pub(crate) fn open_receipt_running_claims_sql_for(schema: &str, job_ids: Option<
)
UNION ALL
SELECT items.job_id, items.run_lease, batches.queue
- FROM {schema}.lease_claim_batches AS batches
+ FROM {batches_rel} AS batches
CROSS JOIN LATERAL unnest(
batches.job_ids,
batches.run_leases,
@@ -749,6 +881,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
@@ -1608,6 +1743,58 @@ impl ReadyJobRow {
}
}
+#[derive(Debug, PartialEq, Eq, sqlx::FromRow)]
+struct ReadyRowIdentity {
+ ready_slot: i32,
+ ready_generation: i64,
+ queue: String,
+ priority: i16,
+ enqueue_shard: i16,
+ lane_seq: i64,
+ job_id: i64,
+ run_lease: i64,
+}
+
+/// Attempts a rescue pass leaves for a later pass, as `(job_id, run_lease)`.
+struct BusyRetained(Vec<(i64, i64)>);
+
+impl BusyRetained {
+ fn exclude(self, excluded: &mut BTreeSet<(i64, i64)>) -> Result<(), AwaError> {
+ let previous = excluded.len();
+ excluded.extend(self.0);
+ if excluded.len() == previous {
+ return Err(AwaError::Validation(
+ "queue storage retained attempt retry made no progress".to_string(),
+ ));
+ }
+ Ok(())
+ }
+}
+
+/// A failure hydrating a deleted lease. `Ambiguous` names an attempt whose
+/// retained ready bodies rank equally, so none can be chosen for it.
+enum HydrateError {
+ Ambiguous { job_id: i64, run_lease: i64 },
+ Awa(AwaError),
+}
+
+impl From for HydrateError {
+ fn from(err: AwaError) -> Self {
+ Self::Awa(err)
+ }
+}
+
+impl From for AwaError {
+ fn from(err: HydrateError) -> Self {
+ match err {
+ HydrateError::Ambiguous { job_id, run_lease } => AwaError::Validation(format!(
+ "queue storage ready body ambiguous for deleted lease job {job_id} run_lease {run_lease}"
+ )),
+ HydrateError::Awa(err) => err,
+ }
+ }
+}
+
#[derive(Debug, Clone, sqlx::FromRow)]
struct ReadyTransitionRow {
ready_slot: i32,
@@ -1989,6 +2176,15 @@ struct DoneJobRow {
payload: serde_json::Value,
}
+// Keep the stored terminal priority for decrementing its old counter key.
+// Manual retry and terminal-to-DLQ moves recover the retained physical lane.
+#[derive(Debug, sqlx::FromRow)]
+struct HydratedTerminalRow {
+ #[sqlx(flatten)]
+ terminal: DoneJobRow,
+ body_priority: i16,
+}
+
impl DoneJobRow {
fn into_job_row(self) -> Result {
let payload = RuntimePayload::from_json(self.payload)?;
@@ -2209,6 +2405,7 @@ struct ReadySnapshotRow {
kind: String,
queue: String,
args: serde_json::Value,
+ priority: i16,
lane_seq: i64,
enqueue_shard: i16,
run_at: DateTime,
@@ -2216,6 +2413,55 @@ struct ReadySnapshotRow {
unique_key: Option>,
unique_states: Option,
payload: serde_json::Value,
+ run_lease: i64,
+ live_body: bool,
+}
+
+type ReadyBodyKey = (i32, i64, String, i16, i64, i64, i64);
+type ReadyBodyLaneKey = (i32, i64, String, i16, i64, i64, i64, i16);
+
+fn select_claimed_ready_body<'a>(
+ candidates: &'a [ReadySnapshotRow],
+ claimed_lanes: &BTreeSet,
+ reference: &DeletedLeaseRow,
+) -> Result<(&'a ReadySnapshotRow, bool), HydrateError> {
+ // Emitted physical-lane evidence wins over a live replacement.
+ // Without it, retain the live-body compatibility fallback; it
+ // cannot identify a legacy claim's original colliding lane.
+ let emitted = |row: &ReadySnapshotRow| {
+ claimed_lanes.contains(&(
+ row.ready_slot,
+ row.ready_generation,
+ row.queue.clone(),
+ row.enqueue_shard,
+ row.lane_seq,
+ row.job_id,
+ row.run_lease + 1,
+ row.priority,
+ ))
+ };
+ let has_emitted = candidates.iter().any(&emitted);
+ let rank = |row: &ReadySnapshotRow| {
+ if has_emitted {
+ emitted(row)
+ } else {
+ row.live_body
+ }
+ };
+ let strongest = candidates
+ .iter()
+ .map(&rank)
+ .max()
+ .expect("ready body group is nonempty");
+ let mut best = candidates.iter().filter(|row| rank(row) == strongest);
+ let ready = best.next().expect("ready body rank has a candidate");
+ if best.next().is_some() {
+ return Err(HydrateError::Ambiguous {
+ job_id: reference.job_id,
+ run_lease: reference.run_lease,
+ });
+ }
+ Ok((ready, has_emitted))
}
#[derive(Debug, Clone, sqlx::FromRow)]
@@ -2240,6 +2486,7 @@ struct LeaseTransitionRow {
args: serde_json::Value,
state: JobState,
priority: i16,
+ ready_priority: i16,
attempt: i16,
run_lease: i64,
max_attempts: i16,
@@ -2252,6 +2499,7 @@ struct LeaseTransitionRow {
unique_states: Option,
payload: serde_json::Value,
progress: Option,
+ self_contained_terminal: bool,
}
impl LeaseTransitionRow {
@@ -2269,11 +2517,19 @@ impl LeaseTransitionRow {
queue: self.queue,
args: self.args,
state,
- priority: self.priority,
+ // Terminal rows retain their ready-lane identity for body hydration.
+ // The effective attempt priority is still used by deferred/DLQ rows.
+ priority: self.ready_priority,
attempt: self.attempt,
run_lease: self.run_lease,
max_attempts: self.max_attempts,
- lane_seq: self.lane_seq,
+ // A competing live copy must not replace this recovered body
+ // when later terminal reads resolve retained ready rows.
+ lane_seq: if self.self_contained_terminal {
+ -self.job_id
+ } else {
+ self.lane_seq
+ },
enqueue_shard: self.enqueue_shard,
run_at: self.run_at,
attempted_at: self.attempted_at,
@@ -7747,7 +8003,7 @@ impl QueueStorage {
let done_projection = done_row_projection("done", "ready");
let ready_join = done_ready_join(schema, "done", "ready");
- let terminal: Option = sqlx::query_as(&format!(
+ let terminal: Option = sqlx::query_as(&format!(
r#"
WITH deleted AS (
DELETE FROM {schema}.done_entries
@@ -7762,7 +8018,7 @@ impl QueueStorage {
)
RETURNING *
)
- SELECT {done_projection}
+ SELECT {done_projection}, COALESCE(ready.priority, done.priority) AS body_priority
FROM deleted AS done
{ready_join}
"#
@@ -7772,7 +8028,9 @@ impl QueueStorage {
.await
.map_err(map_sqlx_error)?;
- if let Some(terminal) = terminal {
+ if let Some(hydrated) = terminal {
+ let body_priority = hydrated.body_priority;
+ let terminal = hydrated.terminal;
self.ensure_terminal_removed_receipt_closures_tx(tx, std::slice::from_ref(&terminal))
.await?;
// The DELETE FROM done_entries above removes one terminal row;
@@ -7787,7 +8045,7 @@ impl QueueStorage {
kind: terminal.kind,
queue: terminal.queue.clone(),
args: terminal.args,
- priority: terminal.priority,
+ priority: body_priority,
attempt: 0,
run_lease: terminal.run_lease,
max_attempts: terminal.max_attempts,
@@ -8268,6 +8526,72 @@ impl QueueStorage {
Ok(newest)
}
+ async fn cancel_materialized_job_lease_tx<'a>(
+ &self,
+ tx: &mut sqlx::Transaction<'a, sqlx::Postgres>,
+ job_id: i64,
+ ) -> Result