Keep admin reads correct across moves, aging and request cancellation

CHANGELOG.md

@@ -6,6 +6,75 @@ Notable changes between releases. Detailed migration notes for storage transitio
 
 ### Fixed
 
+- Preserve recovered bodies and physical lane priority when manually moving
+  historical aged terminal failures to the DLQ, so later DLQ retry retains
+  arguments, attempt limits and priority provenance. Terminal counters are
+  decremented at their original stored priority; automatic rescue-to-DLQ
+  policy keeps the effective priority.
+
+- Recover retained terminal bodies from the exact prior attempt when older
+  workers stored an aged effective priority in a narrow terminal key. Live
+  moved-lane metadata supplies defaults without replacing stored terminal
+  metadata, errors, tags or progress. Manual terminal retry restarts aging from
+  the physical lane; counter removal keeps the stored terminal key and failed
+  carry derives replacement counters from its inserted rows. The installed
+  `terminal_jobs` SQL view is unchanged for older mismatched-priority rows;
+  bodies whose ready backing was already pruned cannot be reconstructed.
+- Exclude tombstoned source lanes and require the prior run lease when
+  hydrating rescued attempts, retaining physical priority only for terminal
+  identity. Deferred retry and DLQ policy keep effective priority. The lookup
+  reads only validated referenced tombstone children; this adds read locks.
+
+- Preserve the original job body when manually retrying a priority-aged attempt
+  that exhausted its retries during rescue. Retained failed reads and failed-row
+  carry through pruning now use the physical ready-lane identity.
+
+- Close read connections if their transaction start is cancelled or fails before
+  SQLx acknowledges it, so later reads cannot inherit an untracked transaction.
+
+- **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` and lifecycle events find running jobs
+  that priority aging promoted** (the Rust side of
+  [#569](https://github.com/hardbyte/awa/issues/569)). Claims and leases
+  record a job's aged effective priority while its ready row keeps the lane
+  priority, and the job and admin reads joined them 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, and read back from `get_job` in
+  the wrong state. Admin job lists, kind counts and cancellation by unique
+  key also missed those running jobs. They now match the attempt (`job_id` and `run_lease`),
+  skipping the tombstoned source row a lane move leaves behind.
+- **`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; the weekly mixed-version rehearsal hit this on
+  2026-11-01. It now reads all of them from one read-only snapshot, after
+  locking the ready, lease and claim rings so prune cannot truncate a partition the
+  snapshot still needs.
+- **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 those `LOCK TABLE` statements 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 appends
+  `queue_terminal_rollup_deltas` 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 in any
+  statement after a queue, lease or claim prune, or a terminal-count rollup
+  slot, sets its `lock_timeout` 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.
 - **Batch enqueues take their lane locks in one order, so they no longer
   deadlock with each other.** On queue storage every enqueue holds a
   transaction-scoped advisory lock on its `(queue, priority, shard)` lane

awa-model/src/admin.rs

@@ -376,9 +376,22 @@ pub fn queue_storage_current_jobs_cte(schema: &str) -> String {
               ON ready.ready_slot = leases.ready_slot
              AND ready.ready_generation = leases.ready_generation
              AND ready.queue = leases.queue
-             AND ready.priority = leases.priority
+             -- Leases keep the aged effective priority; match the ready attempt.
+             AND ready.job_id = leases.job_id
+             AND ready.run_lease + 1 = leases.run_lease
              AND ready.enqueue_shard = leases.enqueue_shard
              AND ready.lane_seq = leases.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
+             )
             UNION ALL
             -- Receipt claims that have not materialised into `leases` are
             -- running too (#246 / #416). They live on legacy row-local
@@ -404,9 +417,22 @@ pub fn queue_storage_current_jobs_cte(schema: &str) -> String {
               ON ready.ready_slot = claims.ready_slot
              AND ready.ready_generation = claims.ready_generation
              AND ready.queue = claims.queue
-             AND ready.priority = claims.priority
+             -- 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
@@ -466,6 +492,16 @@ pub fn queue_storage_current_jobs_cte(schema: &str) -> String {
              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
@@ -564,9 +600,22 @@ fn queue_storage_recent_jobs_cte(schema: &str, since: &str) -> String {
               ON ready.ready_slot = leases.ready_slot
              AND ready.ready_generation = leases.ready_generation
              AND ready.queue = leases.queue
-             AND ready.priority = leases.priority
+             -- Leases keep the aged effective priority; match the ready attempt.
+             AND ready.job_id = leases.job_id
+             AND ready.run_lease + 1 = leases.run_lease
              AND ready.enqueue_shard = leases.enqueue_shard
              AND ready.lane_seq = leases.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 ready.created_at >= {since}
             UNION ALL
             -- A queue or priority move copies the row with its times, so
@@ -685,9 +734,22 @@ fn queue_storage_kind_counts_cte(schema: &str) -> String {
               ON ready.ready_slot = leases.ready_slot
              AND ready.ready_generation = leases.ready_generation
              AND ready.queue = leases.queue
-             AND ready.priority = leases.priority
+             -- Leases keep the aged effective priority; match the ready attempt.
+             AND ready.job_id = leases.job_id
+             AND ready.run_lease + 1 = leases.run_lease
              AND ready.enqueue_shard = leases.enqueue_shard
              AND ready.lane_seq = leases.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
+             )
             GROUP BY ready.kind, leases.queue
         ),
         deferred_by_kind AS MATERIALIZED (
@@ -931,9 +993,22 @@ fn queue_storage_job_id_candidates_sql(
                   ON ready.ready_slot = leases.ready_slot
                  AND ready.ready_generation = leases.ready_generation
                  AND ready.queue = leases.queue
-                 AND ready.priority = leases.priority
+                 -- Leases keep the aged effective priority; match the ready attempt.
+                 AND ready.job_id = leases.job_id
+                 AND ready.run_lease + 1 = leases.run_lease
                  AND ready.enqueue_shard = leases.enqueue_shard
                  AND ready.lane_seq = leases.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 TRUE{predicates}
                 ORDER BY leases.job_id DESC
                 LIMIT {limit}
@@ -1091,12 +1166,15 @@ async fn list_queue_storage_jobs(
             };
         }
         // 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 ids: Vec<i64> = query.fetch_all(tx.as_mut()).await?;
         tx.commit().await?;
+        drop(begin_guard);
         if ids.is_empty() {
             break;
         }
@@ -1408,9 +1486,22 @@ fn unique_key_candidate_sql(schema: &str, scoped: bool) -> String {
                 WHERE ready.ready_slot = leases.ready_slot
                   AND ready.ready_generation = leases.ready_generation
                   AND ready.queue = leases.queue
-                  AND ready.priority = leases.priority
+                  -- Leases keep the aged effective priority; match the ready attempt.
+                  AND ready.job_id = leases.job_id
+                  AND ready.run_lease + 1 = leases.run_lease
                   AND ready.enqueue_shard = leases.enqueue_shard
                   AND ready.lane_seq = leases.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
+                  )
                   AND ready.unique_key = $1
                 OFFSET 0
             ) AS ready ON TRUE
@@ -3141,14 +3232,17 @@ pub async fn queue_overviews(pool: &PgPool) -> Result<Vec<QueueOverview>, AwaErr
 
         // JIT compiles this statement for longer than it runs (about 1.2 s
         // against 0.35 s with 100k retained completions).
-        let mut tx = pool.begin().await?;
+        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 rows = sqlx::query_as::<_, QueueOverview>(audited_sql(sql))
             .fetch_all(tx.as_mut())
             .await?;
         tx.commit().await?;
+        drop(begin_guard);
         return Ok(rows);
     }
 
@@ -3637,14 +3731,17 @@ pub async fn state_counts(pool: &PgPool) -> Result<HashMap<JobState, i64>, 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(audited_sql(sql))
                 .fetch_one(tx.as_mut())
                 .await?;
             tx.commit().await?;
+            drop(begin_guard);
             counts
         };
 
@@ -3750,14 +3847,17 @@ pub async fn job_kind_overviews(pool: &PgPool) -> Result<Vec<JobKindOverview>, A
         );
 
         // 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 rows = sqlx::query_as::<_, JobKindOverview>(audited_sql(sql))
             .fetch_all(tx.as_mut())
             .await?;
         tx.commit().await?;
+        drop(begin_guard);
         return Ok(rows);
     }
 
@@ -4037,7 +4137,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?;
@@ -4055,6 +4157,7 @@ pub async fn state_timeseries(
                 .fetch_all(tx.as_mut())
                 .await?;
         tx.commit().await?;
+        drop(begin_guard);
 
         return Ok(rows
             .into_iter()

awa-model/src/lib.rs

@@ -15,6 +15,7 @@ pub mod migrations;
 pub mod output;
 pub mod partitioned_queue;
 pub mod queue_storage;
+mod read_transaction;
 pub mod reschedule;
 pub mod storage;
 mod storage_footprint_sql;