Forward-port 0.6.16 and 0.6.17 to main, step 1: bounded rescue drain

.config/nextest.toml

@@ -15,7 +15,15 @@ slow-timeout = { period = "60s" }
 
 [test-groups]
 migrations = { max-threads = 1 }
+bulk-rescue-backlog = { max-threads = 1 }
 
 [[profile.default.overrides]]
 filter = 'binary(migration_test)'
 test-group = 'migrations'
+
+# The two bulk rescue backlog tests each load the same database heavily; run
+# concurrently, the live runtime in one cannot heartbeat its claims inside the
+# staleness window.
+[[profile.default.overrides]]
+filter = 'binary(bulk_rescue_test) and test(/^rescue_drains_/)'
+test-group = 'bulk-rescue-backlog'

CHANGELOG.md

@@ -12,6 +12,10 @@ Notable changes between releases. Detailed migration notes for storage transitio
 
 - Everything [0.6.15](https://github.com/hardbyte/awa/releases/tag/v0.6.15) shipped that this branch did not already have; see its release notes for the details. Lifecycle hooks hold their attempt's finalizer slot until they return, so hooks slower than job throughput slow claiming instead of growing worker memory, and maintenance keeps at most 1,024 detached `Rescued` and duplicate `Cancelled` hook tasks per worker. By-job lookups of compact receipt claims use a GIN index on `lease_claim_batches.job_ids` that claim prune builds on each partition as it empties it (migration v054). Compatibility COPY inserts go through `awa.insert_jobs_compat_batch` (migration v055), which takes each queue lane's enqueue lock only when it writes that lane rather than at the batch's first ready row (the lock is still held until the transaction ends), and the new `insert_many_skip_unique_conflicts` inserts a batch through it while skipping claimed unique keys. Rescue reaches stale attempts behind a full batch of ambiguous ones, `Completed` events and follow-ups carry the completed job without the attempt's heartbeat and deadline, slot-scoped admin receipt reads stay off other slots' closure partitions, and unfiltered receipt job lists select candidate IDs before loading bodies. v054 and v055 add functions only and take no table locks; a database coming from 0.6.15 already has both functions from its schema patches. See [v054](docs/upgrade-0.6-to-0.7.md#v054-claim-batch-job-index) and [v055](docs/upgrade-0.6-to-0.7.md#v055-batch-compatibility-insert) in the upgrade guide.
 
+### Forward-ported from 0.6.16
+
+- 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 and claim prune skips their claim slots, so neither ring can wrap. A rescue pass 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. `ClientBuilder::rescue_pass_budget` and `rescue_pass_max_rows` (Python `rescue_pass_budget_ms` and `rescue_pass_max_rows`) set both; zero for either keeps one batch per interval. 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. Rescue's cursor scans seek to their cursor through the index, including the compact deadline-batch scan that 0.6 does not have, and rescue transactions run without JIT. `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. Heartbeats no longer queue behind a completion: an attempt whose closer holds its lock is skipped, lease and progress writes skip rows a finalizer has locked and retry them on the next pass, and a finished attempt stops flushing progress once its handler returns. A point read of a failed job that starts while queue prune moves its row out of the slot being pruned is covered by a test; on this branch the read already locks the ring before it takes its snapshot. See [draining a backlog of abandoned claims](docs/configuration.md#rescue-backlog).
+
 ### Added
 
 - **The queue-storage prune `lock_timeout` is configurable, and prune

awa-metrics/src/lib.rs

@@ -67,6 +67,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";
@@ -204,6 +206,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.
@@ -508,6 +518,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")
@@ -908,6 +931,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(
@@ -1701,7 +1746,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,
@@ -1716,6 +1761,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

@@ -56,8 +56,10 @@ pub use partitioned_queue::{
 };
 pub use queue_storage::{
     ClaimedEntry, ClaimedRuntimeJob, PromoteOutcome, PromotionLane, PruneDurations, PruneOutcome,
-    QueueCounts, QueueStorage, QueueStorageConfig, RingLedgerFoldOutcome, RingRotationCadence,
-    RotateOutcome, SkipReason, TerminalDeltaRollupOutcome, TerminalRollupFoldOutcome,
+    QueueCounts, QueueStorage, QueueStorageConfig, RescueBatch, RescueBatchOptions,
+    RescueScanCursor, RescueScanProgress, RescueSlotOrder, RingLedgerFoldOutcome,
+    RingRotationCadence, RotateOutcome, SkipReason, TerminalDeltaRollupOutcome,
+    TerminalRollupFoldOutcome,
 };
 pub use storage::StorageStatus;