fix: review fixes from the 0.6.14 forward-port (0.6)

.github/workflows/ci.yml

@@ -108,6 +108,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s
@@ -153,6 +154,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s
@@ -222,6 +224,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s
@@ -262,6 +265,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s
@@ -295,6 +299,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s
@@ -335,6 +340,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s
@@ -417,6 +423,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s
@@ -488,6 +495,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s
@@ -586,6 +594,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s

CHANGELOG.md

@@ -11,6 +11,9 @@ Notable changes between releases. Detailed migration notes for storage transitio
 ### Fixed
 
 - Admin job lists and kind counts work on a queue-storage schema installed with more claim slots than the default, instead of failing with `invalid physical claim slot`.
+- Heartbeat, deadline and callback rescue of lease-backed attempts reach the stale attempts behind a full batch of attempts whose retained ready bodies are ambiguous. Before, every pass selected the same ambiguous attempts first, left them running, and rescued nothing else. A pass now sets all of a batch's ambiguous attempts aside at once.
+- Admin job lists and kind counts no longer read other claim slots' receipt closure partitions when they read one slot's open claims.
+- A `Completed` lifecycle event, and the job a `Completed` follow-up enqueue spec receives, no longer carry the finished attempt's `heartbeat_at` and `deadline_at`; both are `None`, as a read of the completed job returns them.
 
 ## [0.6.14] — 2026-11-08
 

awa-model/src/queue_storage.rs

@@ -865,20 +865,20 @@ fn open_receipt_running_claims_sql_from(
         WHERE {batch_filter}NOT (
                   batches.receipt_ranges <@ COALESCE((
                       SELECT range_agg(closed)
-                      FROM {schema}.lease_claim_closure_batches AS closure_batches
+                      FROM {closure_batch_rel} AS closure_batches
                       CROSS JOIN LATERAL unnest(closure_batches.receipt_ranges) AS closed
                       WHERE closure_batches.claim_slot = batches.claim_slot
                         AND closure_batches.receipt_ranges && batches.receipt_ranges
                   ), '{{}}'::int8multirange)
               )
           AND NOT EXISTS (
-                  SELECT 1 FROM {schema}.lease_claim_closures AS closures
+                  SELECT 1 FROM {closure_rel} AS closures
                   WHERE closures.claim_slot = batches.claim_slot
                     AND closures.job_id = items.job_id
                     AND closures.run_lease = items.run_lease
               )
           AND NOT EXISTS (
-                  SELECT 1 FROM {schema}.lease_claim_closure_batches AS closure_batches
+                  SELECT 1 FROM {closure_batch_rel} 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
@@ -1756,25 +1756,49 @@ struct ReadyRowIdentity {
 }
 
 /// Attempts a rescue pass leaves for a later pass, as `(job_id, run_lease)`.
-struct BusyRetained(Vec<(i64, i64)>);
+struct BusyRetained {
+    attempts: Vec<(i64, i64)>,
+    /// The attempts stay unresolvable until their retained bodies are
+    /// repaired, rather than until a lock is released.
+    unresolvable: bool,
+}
 
 impl BusyRetained {
-    fn exclude(self, excluded: &mut BTreeSet<(i64, i64)>) -> Result<(), AwaError> {
+    fn busy(attempts: Vec<(i64, i64)>) -> Self {
+        Self {
+            attempts,
+            unresolvable: false,
+        }
+    }
+
+    /// Leaves the attempts out of the rest of the pass. Busy attempts shrink
+    /// the pass's selection. Unresolvable attempts are replaced in it, so a
+    /// backlog of them, which every pass meets first, cannot fill each batch
+    /// and keep rescue from the attempts behind it.
+    fn exclude(
+        self,
+        excluded: &mut BTreeSet<(i64, i64)>,
+        selected: &mut Option<BTreeSet<(i64, i64)>>,
+    ) -> Result<(), AwaError> {
         let previous = excluded.len();
-        excluded.extend(self.0);
+        excluded.extend(self.attempts);
         if excluded.len() == previous {
             return Err(AwaError::Validation(
                 "queue storage retained attempt retry made no progress".to_string(),
             ));
         }
+        if self.unresolvable {
+            *selected = None;
+        }
         Ok(())
     }
 }
 
-/// A failure hydrating a deleted lease. `Ambiguous` names an attempt whose
-/// retained ready bodies rank equally, so none can be chosen for it.
+/// A failure hydrating deleted leases. `Ambiguous` names the attempts, as
+/// `(job_id, run_lease)`, whose retained ready bodies rank equally, so none
+/// can be chosen for them.
 enum HydrateError {
-    Ambiguous { job_id: i64, run_lease: i64 },
+    Ambiguous(Vec<(i64, i64)>),
     Awa(AwaError),
 }
 
@@ -1787,9 +1811,16 @@ impl From<AwaError> for HydrateError {
 impl From<HydrateError> 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::Ambiguous(attempts) => AwaError::Validation(match attempts.as_slice() {
+                [] => "queue storage ready body ambiguous for deleted lease".to_string(),
+                [(job_id, run_lease)] => format!(
+                    "queue storage ready body ambiguous for deleted lease job {job_id} run_lease {run_lease}"
+                ),
+                [(job_id, run_lease), rest @ ..] => format!(
+                    "queue storage ready body ambiguous for deleted lease job {job_id} run_lease {run_lease} and {} more",
+                    rest.len()
+                ),
+            }),
             HydrateError::Awa(err) => err,
         }
     }
@@ -2456,10 +2487,10 @@ fn select_claimed_ready_body<'a>(
     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,
-        });
+        return Err(HydrateError::Ambiguous(vec![(
+            reference.job_id,
+            reference.run_lease,
+        )]));
     }
     Ok((ready, has_emitted))
 }
@@ -10849,14 +10880,19 @@ impl QueueStorage {
         match hydrated {
             Ok(hydrated) => Ok(hydrated),
             // One unresolvable attempt must not stop the rescue of the others.
-            Err(HydrateError::Ambiguous { job_id, run_lease }) => {
-                tracing::error!(
-                    schema = self.schema(),
-                    job_id,
-                    run_lease,
-                    "Rescue skipped an attempt: queue storage ready body ambiguous for deleted lease; it stays running until its retained bodies are repaired"
-                );
-                Ok(Err(BusyRetained(vec![(job_id, run_lease)])))
+            Err(HydrateError::Ambiguous(attempts)) => {
+                for (job_id, run_lease) in &attempts {
+                    tracing::error!(
+                        schema = self.schema(),
+                        job_id,
+                        run_lease,
+                        "Rescue skipped an attempt: queue storage ready body ambiguous for deleted lease; it stays running until its retained bodies are repaired"
+                    );
+                }
+                Ok(Err(BusyRetained {
+                    attempts,
+                    unresolvable: true,
+                }))
             }
             Err(HydrateError::Awa(err)) => Err(err),
         }
@@ -10876,6 +10912,33 @@ impl QueueStorage {
         let run_leases: Vec<i64> = deleted.iter().map(|row| row.run_lease).collect();
         let (ready_map, claimed_lanes) =
             self.load_ready_snapshot_candidates_tx(tx, &deleted).await?;
+        // Report every ambiguous attempt at once, before any write, so a
+        // caller that sets them aside needs one retry for all of them.
+        let ambiguous: Vec<(i64, i64)> = deleted
+            .iter()
+            .filter(|row| {
+                ready_map
+                    .get(&(
+                        row.ready_slot,
+                        row.ready_generation,
+                        row.queue.clone(),
+                        row.enqueue_shard,
+                        row.lane_seq,
+                        row.job_id,
+                        row.run_lease,
+                    ))
+                    .is_some_and(|candidates| {
+                        matches!(
+                            select_claimed_ready_body(candidates, &claimed_lanes, row),
+                            Err(HydrateError::Ambiguous(_))
+                        )
+                    })
+            })
+            .map(|row| (row.job_id, row.run_lease))
+            .collect();
+        if !ambiguous.is_empty() {
+            return Err(HydrateError::Ambiguous(ambiguous));
+        }
 
         let attempt_rows: Vec<AttemptStateRow> = sqlx::query_as(&format!(
             r#"
@@ -11130,10 +11193,10 @@ impl QueueStorage {
                 .insert((source.job_id, source.run_lease), source)
                 .is_some()
             {
-                return Err(HydrateError::Ambiguous {
-                    job_id: source.job_id,
-                    run_lease: source.run_lease + 1,
-                });
+                return Err(HydrateError::Ambiguous(vec![(
+                    source.job_id,
+                    source.run_lease + 1,
+                )]));
             }
         }
         let job_ids: Vec<i64> = sources.iter().map(|row| row.job_id).collect();
@@ -11181,7 +11244,7 @@ impl QueueStorage {
             .await
             .map_err(map_sqlx_error)?;
             if !busy.is_empty() {
-                return Ok(Some(BusyRetained(busy)));
+                return Ok(Some(BusyRetained::busy(busy)));
             }
         } else {
             self.lock_claim_heads_of_jobs_with_wait_tx(tx, &job_ids, true)
@@ -11266,10 +11329,10 @@ impl QueueStorage {
         for row in &emitted {
             let key = (row.job_id, row.run_lease);
             if !seen.insert(key) || expected.get(&key).copied() != Some(row) {
-                return Err(HydrateError::Ambiguous {
-                    job_id: row.job_id,
-                    run_lease: row.run_lease + 1,
-                });
+                return Err(HydrateError::Ambiguous(vec![(
+                    row.job_id,
+                    row.run_lease + 1,
+                )]));
             }
         }
         sqlx::query(&format!(
@@ -15408,7 +15471,7 @@ impl QueueStorage {
                 Ok(moved) => moved,
                 Err(busy) => {
                     attempt.rollback().await.map_err(map_sqlx_error)?;
-                    busy.exclude(&mut excluded)?;
+                    busy.exclude(&mut excluded, &mut selected)?;
                     continue;
                 }
             };
@@ -15419,7 +15482,7 @@ impl QueueStorage {
                 Ok(moved) => moved,
                 Err(busy) => {
                     attempt.rollback().await.map_err(map_sqlx_error)?;
-                    busy.exclude(&mut excluded)?;
+                    busy.exclude(&mut excluded, &mut selected)?;
                     continue;
                 }
             };
@@ -15578,7 +15641,7 @@ impl QueueStorage {
                 Ok(moved) => moved,
                 Err(busy) => {
                     attempt.rollback().await.map_err(map_sqlx_error)?;
-                    busy.exclude(&mut excluded)?;
+                    busy.exclude(&mut excluded, &mut selected)?;
                     continue;
                 }
             };
@@ -15590,7 +15653,7 @@ impl QueueStorage {
                     Ok(moved) => moved,
                     Err(busy) => {
                         attempt.rollback().await.map_err(map_sqlx_error)?;
-                        busy.exclude(&mut excluded)?;
+                        busy.exclude(&mut excluded, &mut selected)?;
                         continue;
                     }
                 },
@@ -15716,7 +15779,7 @@ impl QueueStorage {
                 Ok(moved) => moved,
                 Err(busy) => {
                     attempt.rollback().await.map_err(map_sqlx_error)?;
-                    busy.exclude(&mut excluded)?;
+                    busy.exclude(&mut excluded, &mut selected)?;
                     continue;
                 }
             };

awa-worker/src/executor.rs

@@ -1723,6 +1723,17 @@ fn rescheduled_event_row(
     row
 }
 
+/// The job as a read returns it after this attempt completes.
+fn completed_event_row(job: &JobRow) -> JobRow {
+    let mut row = job.clone();
+    row.state = JobState::Completed;
+    row.finalized_at = Some(chrono::Utc::now());
+    row.progress = None;
+    row.heartbeat_at = None;
+    row.deadline_at = None;
+    row
+}
+
 #[allow(clippy::too_many_arguments)]
 async fn complete_job_queue_storage(
     runtime: &QueueStorageRuntime,
@@ -1833,15 +1844,11 @@ async fn complete_job_queue_storage(
                 return Ok(CompletionOutcome::IgnoredStale);
             }
             if needs_event {
-                // A completed attempt changes only these fields, so the event is
-                // built from the worker's copy rather than re-reading the job.
-                let mut updated_job = job.clone();
-                updated_job.state = JobState::Completed;
-                updated_job.finalized_at = Some(chrono::Utc::now());
-                updated_job.progress = None;
+                // The event is built from the worker's copy rather than
+                // re-reading the job.
                 Ok(CompletionOutcome::Applied {
                     event: Some(UntypedJobEvent::Completed {
-                        job: updated_job,
+                        job: completed_event_row(job),
                         duration,
                     }),
                     terminal: false,
@@ -2459,12 +2466,7 @@ async fn complete_queue_storage_with_followups(
 
     // Synthesise the post-completion snapshot — the store has just moved the
     // lease into `done_entries` so a live SELECT would race with rotation.
-    // The happy-path fields are exactly those `complete_job_queue_storage`
-    // reconstructs when `load_job` returns None.
-    let mut updated_job = job.clone();
-    updated_job.state = JobState::Completed;
-    updated_job.finalized_at = Some(chrono::Utc::now());
-    updated_job.progress = None;
+    let updated_job = completed_event_row(job);
 
     crate::enqueue_specs::dispatch_specs_in_tx(&mut tx, &updated_job, specs, None).await?;