fix(queue-storage): drain a large abandoned-claim backlog in bounded time (0.6)

CHANGELOG.md

@@ -12,10 +12,13 @@ Notable changes between releases. Detailed migration notes for storage transitio
 
 ### Added
 
+- `ClientBuilder::rescue_pass_budget` and `rescue_pass_max_rows` (Python `rescue_pass_budget_ms` and `rescue_pass_max_rows`) control how long and how far one queue-storage rescue pass drains. `awa.maintenance.rescue.pass_size` (by `awa.rescue.kind` and `awa.rescue.pass_end`) and `awa.maintenance.rescue.oldest_attempt_age` show a drain in progress.
 - Durable operator halt/resume for `maintenance.prune`, in drain mode. Compatible leaders observe it on their existing session; dispatch, cron, promotion, rescue and enqueue continue. Status lists live runtimes without supported, recent reports and cannot report the fleet halted while any remain. Older binaries ignore the control. See [runtime controls](docs/runtime-controls.md).
 
 ### Fixed
 
+- Queue-storage rescue drains a large backlog of abandoned claims in bounded time. A killed runtime leaves every attempt it claimed open, and until they close, queue prune skips the queue slots they reference (`queue.unclosed_claim_refs`) and claim prune skips their claim slots, so neither ring can wrap. Rescue closed at most 500 attempts per kind per 30-second interval: about 33 a second, so 200,000 abandoned row-local claims took about 100 minutes. A rescue pass now takes 500-attempt batches, each in its own transaction, until a batch comes back short, `rescue_pass_budget` (default 1 s, at most half the rescue interval) is spent, or it has rescued `rescue_pass_max_rows` (default 5,000), and while a backlog remains the next pass runs a second later. Rescue's cursor scans also seek to their cursor through the index instead of rereading every claim already swept, so a batch costs the same at the end of a drain as at the start, and they run without JIT, whose compilation took seconds per statement. With the defaults, 1.6 million abandoned row-local claims past their deadlines drained in about 17 minutes (1,600 a second), and both rings moved past the pinned slots about two minutes later. Deadline and callback rescue act on their expired deadlines; stale-heartbeat rescue takes more than one batch per pass only once runtime snapshots show the claims' owners gone, about 3 to 3.5 minutes after a runtime stops reporting with default settings. Retried attempts keep their backoffs, so their due times can cluster. Zero for either control keeps one batch per interval. See [draining a backlog of abandoned claims](docs/configuration.md#rescue-backlog).
+- Heartbeats no longer queue behind a completion. Compact receipt claims are heartbeated by their claim batch and item index, an attempt whose closer holds its lock is skipped rather than waited for, and lease and progress writes skip rows a finalizer has locked and retry them on the next pass. One stalled completion could otherwise hold up a whole heartbeat pass until other attempts went stale and were rescued. A finished attempt stops flushing progress once its handler returns.
 - `insert_many_copy` and `insert_many_skip_unique_conflicts` no longer deadlock with per-row producers (generic `insert_many`, single-job inserts, 0.6.14 and earlier binaries) that share their unique keys or lanes. Since 0.6.15, when a per-row transaction held a lane and then inserted a key a waiting batch had claimed, PostgreSQL aborted one of the two with `deadlock detected` (`40P01`). The batch function now waits at most 500 ms (or half of its session's `deadlock_timeout`, if lower) while it holds claims or lanes: on a longer wait it releases them and tries again, so the per-row transaction proceeds and the batch then treats the keys it committed as already claimed. The same applies to a transaction that enqueues lanes in separate statements in the opposite order to a batch. This covers per-row sessions whose `deadlock_timeout` is above 500 ms, as PostgreSQL's default of 1 s is, and batches whose unique claims insert within that budget (10,000 to 20,000 keys on a fast server, fewer on a slow or busy one); larger batches behave as in 0.6.15. The protection lasts for 2 s of retries; then the batch makes one last attempt with ordinary lock waits, as in 0.6.15. A per-row transaction that holds a lane for longer than that before it inserts one of the batch's keys can therefore still deadlock with it, as in 0.6.15. A cycle through a lock the batch's own transaction took in an earlier statement also ends in `deadlock detected`, after about 3 to 4 s instead of 1 to 2 s, and usually aborts the batch's transaction. The caller's `lock_timeout` applies to each lock wait, as before. Results and the lane lock window are unchanged.
 - Claim prune no longer treats a partial GIN index on `lease_claim_batches.job_ids`, or a session with `quote_all_identifiers` on, as the claim batch index. It recognises only a valid, non-partial GIN index on `job_ids` alone. When another index or relation already has the index's name, `ensure_claim_batch_job_index` leaves it in place, returns false and raises a warning naming it, instead of reporting the slot indexed; the slot keeps scanning until an operator drops or renames it.
 - Refreshing `ensure_claim_batch_job_index` as a migration role that owns it but may only `SET ROLE` to the table owner (PostgreSQL 16+ membership `WITH INHERIT FALSE`) no longer leaves `EXECUTE` with `PUBLIC`. The refresh used to hand the function to the table owner first, after which that role could no longer revoke or grant on it, and PostgreSQL only warned. It now resets `EXECUTE` after the transfer, acting as the new owner when it cannot as itself, so exactly the roles holding `TRUNCATE` on `lease_claim_batches` can execute it, the former owner included. Migrations run with the owner as the effective role, as documented, were not affected.

awa-metrics/src/lib.rs

@@ -61,6 +61,8 @@ pub mod names {
     pub const COMPLETION_FLUSH_DURATION: &str = "awa.completion.flush_duration";
     pub const HEARTBEAT_BATCHES: &str = "awa.heartbeat.batches";
     pub const MAINTENANCE_RESCUES: &str = "awa.maintenance.rescues";
+    pub const MAINTENANCE_RESCUE_PASS_SIZE: &str = "awa.maintenance.rescue.pass_size";
+    pub const MAINTENANCE_RESCUE_OLDEST_AGE: &str = "awa.maintenance.rescue.oldest_attempt_age";
     pub const MAINTENANCE_PROMOTE_BATCHES: &str = "awa.maintenance.promote_batches";
     pub const MAINTENANCE_PROMOTE_BATCH_SIZE: &str = "awa.maintenance.promote_batch_size";
     pub const MAINTENANCE_PROMOTE_DURATION: &str = "awa.maintenance.promote_duration";
@@ -174,6 +176,14 @@ pub struct AwaMetrics {
     pub heartbeat_batches: Counter<u64>,
     /// Total maintenance rescue operations.
     pub maintenance_rescues: Counter<u64>,
+    /// Attempts rescued per queue-storage rescue pass, attributed by
+    /// `awa.rescue.kind` and why the pass stopped (`awa.rescue.pass_end`).
+    pub maintenance_rescue_pass_size: Histogram<u64>,
+    /// Age of the oldest attempt the latest queue-storage rescue pass
+    /// rescued, per `awa.rescue.kind`; 0 when it rescued nothing. Rescue
+    /// works oldest claim first, so a value well above the heartbeat
+    /// staleness or deadline means a backlog of abandoned claims.
+    pub maintenance_rescue_oldest_attempt_age_seconds: Gauge<f64>,
     /// Total jobs parked for external callback.
     pub jobs_waiting_external: Counter<u64>,
     /// Current queue depth per state — how many jobs are in each state per queue.
@@ -433,6 +443,19 @@ impl AwaMetrics {
                 .with_description("Number of jobs rescued by maintenance")
                 .with_unit("{job}")
                 .build(),
+            maintenance_rescue_pass_size: meter
+                .u64_histogram(names::MAINTENANCE_RESCUE_PASS_SIZE)
+                .with_description("Attempts rescued per queue-storage rescue pass")
+                .with_unit("{job}")
+                .with_boundaries(COUNT_BUCKETS.to_vec())
+                .build(),
+            maintenance_rescue_oldest_attempt_age_seconds: meter
+                .f64_gauge(names::MAINTENANCE_RESCUE_OLDEST_AGE)
+                .with_description(
+                    "Age of the oldest attempt rescued by the latest queue-storage rescue pass",
+                )
+                .with_unit("s")
+                .build(),
             jobs_waiting_external: meter
                 .u64_counter(names::JOB_WAITING_EXTERNAL)
                 .with_description("Number of jobs parked for external callback")
@@ -751,6 +774,28 @@ impl AwaMetrics {
             .record(duration.as_secs_f64(), &attrs);
     }
 
+    /// Record one queue-storage rescue pass: how many attempts it rescued,
+    /// why it stopped, and the age of the oldest attempt it rescued.
+    pub fn record_rescue_pass(
+        &self,
+        kind: &'static str,
+        pass_end: &'static str,
+        rescued: u64,
+        oldest_attempt_age: Duration,
+    ) {
+        self.maintenance_rescue_pass_size.record(
+            rescued,
+            &[
+                opentelemetry::KeyValue::new("awa.rescue.kind", kind),
+                opentelemetry::KeyValue::new("awa.rescue.pass_end", pass_end),
+            ],
+        );
+        self.maintenance_rescue_oldest_attempt_age_seconds.record(
+            oldest_attempt_age.as_secs_f64(),
+            &[opentelemetry::KeyValue::new("awa.rescue.kind", kind)],
+        );
+    }
+
     /// Record a scheduled/retryable promotion batch.
     pub fn record_promotion_batch(&self, state: &str, batch_size: u64, duration: Duration) {
         let attrs = [opentelemetry::KeyValue::new(
@@ -1418,7 +1463,7 @@ mod tests {
                 &JOB_DURATION_BUCKETS_SECONDS,
             ),
         ];
-        let counts: [(&str, &Histogram<u64>); 7] = [
+        let counts: [(&str, &Histogram<u64>); 8] = [
             (names::ENQUEUE_BATCH_SIZE, &metrics.enqueue_batch_size),
             (
                 names::DISPATCH_CAPACITY_AVAILABLE,
@@ -1433,6 +1478,10 @@ mod tests {
                 names::MAINTENANCE_PROMOTE_BATCH_SIZE,
                 &metrics.promotion_batch_size,
             ),
+            (
+                names::MAINTENANCE_RESCUE_PASS_SIZE,
+                &metrics.maintenance_rescue_pass_size,
+            ),
             (
                 names::MAINTENANCE_ROTATE_SKIPPED_ROWS,
                 &metrics.maintenance_rotate_skipped_rows,

awa-model/src/lib.rs

@@ -51,7 +51,8 @@ pub use partitioned_queue::{
 };
 pub use queue_storage::{
     ClaimedEntry, ClaimedRuntimeJob, PromoteOutcome, PromotionLane, PruneOutcome, QueueCounts,
-    QueueStorage, QueueStorageConfig, RotateOutcome, SkipReason, TerminalDeltaRollupOutcome,
+    QueueStorage, QueueStorageConfig, RescueBatch, RescueBatchOptions, RescueScanCursor,
+    RescueScanProgress, RescueSlotOrder, RotateOutcome, SkipReason, TerminalDeltaRollupOutcome,
 };
 pub use storage::StorageStatus;