feat(worker): configurable queue-storage prune lock timeout and prune outcome metrics

CHANGELOG.md

@@ -4,6 +4,31 @@ Notable changes between releases. Detailed migration notes for storage transitio
 
 ## [Unreleased]
 
+### Added
+
+- **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 takes its other initial partition locks with `NOWAIT`. Later
+  statements also run under the timeout. 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 post-`skipped_active` backoff) and
+  `error` (prune failed with a database error) alongside the existing
+  `pruned`, `noop`, `blocked`, `skipped_active` and `already_pruned`. See
+  [Configuration](docs/configuration.md#what-the-knobs-mean).
+
 ### Fixed
 
 - Retry queue-storage worker finalization after transient lock contention within

awa-metrics/src/lib.rs

@@ -251,9 +251,12 @@ pub struct AwaMetrics {
     /// pinned by handfuls of stragglers or by mountains of unfinished work.
     pub maintenance_rotate_skipped_rows: Histogram<u64>,
     /// Maintenance ring prune attempts, attributed to (ring, outcome, reason).
-    /// reason="none" for Pruned/Noop/Blocked; otherwise carries the
-    /// SkipReason discriminator (e.g. "queue.active_leases",
-    /// "queue.pending_ready", "claim.open").
+    /// Outcomes: `pruned`, `noop`, `blocked` (the exclusive locks were not
+    /// available within the prune lock timeout), `skipped_active`,
+    /// `already_pruned`, `backed_off` (the tick skipped prune after repeated
+    /// `skipped_active` results) and `error`. reason="none" except for
+    /// `skipped_active`, which carries the SkipReason discriminator (e.g.
+    /// "queue.active_leases", "queue.pending_ready", "claim.open").
     pub maintenance_prune_attempts: Counter<u64>,
     /// Magnitude of the reason count when prune returns SkippedActive.
     pub maintenance_prune_skipped_rows: Histogram<u64>,
@@ -1315,16 +1318,32 @@ impl AwaMetrics {
         }
     }
 
-    /// Record an idle maintenance tick that skipped destructive prune because
-    /// the exact sealed generation was already reclaimed by this leader.
-    pub fn record_prune_already_pruned(&self, ring: &'static str) {
+    /// Record a maintenance tick that skipped prune because the ring's
+    /// backoff after repeated `skipped_active` results had not yet elapsed.
+    pub fn record_prune_backed_off(&self, ring: &'static str) {
+        self.record_prune_attempt_without_reason(ring, "backed_off");
+    }
+
+    /// Record a prune attempt that failed with a database error rather than
+    /// returning an outcome.
+    pub fn record_prune_error(&self, ring: &'static str) {
+        self.record_prune_attempt_without_reason(ring, "error");
+    }
+
+    fn record_prune_attempt_without_reason(&self, ring: &'static str, outcome: &'static str) {
         let attrs = [
             opentelemetry::KeyValue::new("awa.ring", ring),
-            opentelemetry::KeyValue::new("awa.ring.outcome", "already_pruned"),
+            opentelemetry::KeyValue::new("awa.ring.outcome", outcome),
             opentelemetry::KeyValue::new("awa.ring.reason", "none"),
         ];
         self.maintenance_prune_attempts.add(1, &attrs);
     }
+
+    /// Record an idle maintenance tick that skipped destructive prune because
+    /// the exact sealed generation was already reclaimed by this leader.
+    pub fn record_prune_already_pruned(&self, ring: &'static str) {
+        self.record_prune_attempt_without_reason(ring, "already_pruned");
+    }
 }
 
 /// No-op metrics for when OTel is not configured.
@@ -1443,6 +1462,97 @@ mod tests {
         assert!(saw_already_pruned);
     }
 
+    #[test]
+    fn prune_attempts_count_every_outcome_per_ring() {
+        use awa_model::{PruneDurations, PruneOutcome, SkipReason};
+
+        let exporter = InMemoryMetricExporter::default();
+        let provider = SdkMeterProvider::builder()
+            .with_periodic_exporter(exporter.clone())
+            .build();
+        let metrics = AwaMetrics::new(&provider.meter("awa-metrics-test"));
+
+        for ring in ["queue", "lease", "claim"] {
+            metrics.record_prune_outcome(
+                ring,
+                &PruneOutcome::Pruned {
+                    slot: 0,
+                    generation: 1,
+                    carried_failed_rows: 0,
+                    durations: PruneDurations::default(),
+                },
+            );
+            metrics.record_prune_outcome(ring, &PruneOutcome::Blocked { slot: 0 });
+            metrics.record_prune_outcome(ring, &PruneOutcome::Noop);
+            metrics.record_prune_already_pruned(ring);
+            metrics.record_prune_backed_off(ring);
+            metrics.record_prune_error(ring);
+        }
+        metrics.record_prune_outcome("claim", &PruneOutcome::Blocked { slot: 1 });
+        metrics.record_prune_outcome(
+            "claim",
+            &PruneOutcome::SkippedActive {
+                slot: 0,
+                reason: SkipReason::ClaimOpen,
+                count: 3,
+            },
+        );
+        provider.force_flush().expect("flush metrics");
+        let resource_metrics = exporter.get_finished_metrics().expect("read metrics");
+
+        let mut counts: std::collections::BTreeMap<(String, String, String), u64> =
+            std::collections::BTreeMap::new();
+        for resource in &resource_metrics {
+            for scope in resource.scope_metrics() {
+                for metric in scope.metrics() {
+                    let (
+                        names::MAINTENANCE_PRUNE_ATTEMPTS,
+                        AggregatedMetrics::U64(MetricData::Sum(sum)),
+                    ) = (metric.name(), metric.data())
+                    else {
+                        continue;
+                    };
+                    for point in sum.data_points() {
+                        let label = |key: &str| {
+                            point
+                                .attributes()
+                                .find(|attribute| attribute.key.as_str() == key)
+                                .map(|attribute| attribute.value.as_str().into_owned())
+                                .unwrap_or_default()
+                        };
+                        counts.insert(
+                            (
+                                label("awa.ring"),
+                                label("awa.ring.outcome"),
+                                label("awa.ring.reason"),
+                            ),
+                            point.value(),
+                        );
+                    }
+                }
+            }
+        }
+
+        let count = |ring: &str, outcome: &str, reason: &str| {
+            counts
+                .get(&(ring.to_string(), outcome.to_string(), reason.to_string()))
+                .copied()
+        };
+        for ring in ["queue", "lease", "claim"] {
+            for outcome in ["pruned", "noop", "already_pruned", "backed_off", "error"] {
+                assert_eq!(count(ring, outcome, "none"), Some(1), "{ring} {outcome}");
+            }
+            let expected_blocked = if ring == "claim" { 2 } else { 1 };
+            assert_eq!(
+                count(ring, "blocked", "none"),
+                Some(expected_blocked),
+                "{ring} blocked"
+            );
+        }
+        assert_eq!(count("claim", "skipped_active", "claim.open"), Some(1));
+        assert_eq!(count("queue", "skipped_active", "claim.open"), None);
+    }
+
     /// A second-scale observation must not fall in the SDK's default
     /// millisecond-scale buckets (upper bounds 5, 10, 25, ...), and every
     /// histogram must export the intended explicit boundaries.

awa-model/src/queue_storage.rs

@@ -3784,15 +3784,54 @@ impl QueueStorage {
         }
     }
 
-    #[doc(hidden)]
+    /// Set the PostgreSQL `lock_timeout` applied to the lock acquisitions of
+    /// ring prune (queue, lease and claim rings) and terminal-delta rollup.
+    ///
+    /// Among its initial partition locks, queue ring prune waits only on
+    /// `ready_segments`; the others use `NOWAIT`. Later reads and writes also
+    /// run under the timeout. A reader holding a `NOWAIT` partition makes the
+    /// prune `Blocked` immediately whatever the timeout.
+    ///
+    /// Each lock acquisition gets its own timeout; this does not bound the
+    /// whole prune. A lock timeout rolls back the prune and reports
+    /// [`PruneOutcome::Blocked`]; the maintenance loop retries on a later
+    /// tick. The default is 10 ms, which keeps prune from queueing behind
+    /// readers and so from stalling claims and enqueues behind its pending
+    /// `ACCESS EXCLUSIVE` request. Raise it
+    /// when prune is persistently `Blocked` by short-lived readers of the
+    /// ring partitions. Keep it well below `deadlock_timeout` (default 1 s)
+    /// and any `statement_timeout`: a prune that waits as long as
+    /// `deadlock_timeout` triggers deadlock detection, which may cancel a
+    /// concurrent claim instead of the prune. The timeout is applied at
+    /// millisecond granularity, with a minimum of 1 ms.
+    ///
+    /// Returns a validation error for a zero timeout or one above PostgreSQL's
+    /// `lock_timeout` maximum (`i32::MAX` milliseconds, about 24.8 days).
     pub fn with_prune_lock_timeout(mut self, timeout: Duration) -> Result<Self, AwaError> {
+        self.set_prune_lock_timeout(timeout)?;
+        Ok(self)
+    }
+
+    /// In-place form of [`Self::with_prune_lock_timeout`].
+    pub fn set_prune_lock_timeout(&mut self, timeout: Duration) -> Result<(), AwaError> {
         if timeout.is_zero() {
             return Err(AwaError::Validation(
                 "queue storage prune lock timeout must be greater than zero".into(),
             ));
         }
+        if timeout.as_millis() > i32::MAX as u128 {
+            return Err(AwaError::Validation(format!(
+                "queue storage prune lock timeout must not exceed {} ms",
+                i32::MAX
+            )));
+        }
         self.prune_lock_timeout = timeout;
-        Ok(self)
+        Ok(())
+    }
+
+    /// The `lock_timeout` applied to prune's lock acquisitions.
+    pub fn prune_lock_timeout(&self) -> Duration {
+        self.prune_lock_timeout
     }
 
     pub fn from_existing_schema(schema: impl Into<String>) -> Result<Self, AwaError> {

awa-python/python/awa/_awa.pyi

@@ -613,6 +613,7 @@ class Client:
         queue_storage_queue_rotate_interval_ms: int = 1000,
         queue_storage_lease_rotate_interval_ms: int = 250,
         queue_storage_claim_rotate_interval_ms: int | None = None,
+        queue_storage_prune_lock_timeout_ms: int | None = None,
         queue_storage_rotate_min_slot_age_ms: int | None = None,
         queue_storage_rotate_target_slot_bytes: int | None = None,
         queue_storage_rotate_min_slot_rows: int | None = None,

awa-python/python/awa/client.py

@@ -930,6 +930,7 @@ async def start(
         queue_storage_queue_rotate_interval_ms: int = 1000,
         queue_storage_lease_rotate_interval_ms: int = 250,
         queue_storage_claim_rotate_interval_ms: int | None = None,
+        queue_storage_prune_lock_timeout_ms: int | None = None,
         queue_storage_rotate_min_slot_age_ms: int | None = None,
         queue_storage_rotate_target_slot_bytes: int | None = None,
         queue_storage_rotate_min_slot_rows: int | None = None,
@@ -967,6 +968,18 @@ async def start(
         counters. Exact reads include pending deltas, so the default favors
         fewer hot counter updates over immediate folding.
 
+        ``queue_storage_prune_lock_timeout_ms`` sets the PostgreSQL
+        ``lock_timeout`` the maintenance leader uses when pruning the queue,
+        lease and claim rings (default 10 ms). Each lock acquisition gets its
+        own timeout; this does not bound the whole prune. Among queue prune's
+        initial partition locks, only ``ready_segments`` waits; the others use
+        ``NOWAIT``. Later reads and writes also run under the timeout. A lock
+        timeout rolls back the prune and counts it as ``blocked``; the
+        maintenance loop retries on a later tick. Raise it when ``blocked``
+        outcomes persist, keeping it well below the server's
+        ``deadlock_timeout`` and ``statement_timeout`` (a warning is logged at
+        start otherwise). Must be > 0.
+
         Queue dicts also accept ``claimers`` and ``claim_batch_size`` for hot
         queue-storage queues. ``claimers`` adds dispatcher/claimer loops that
         share the queue's worker permits; benchmark ``2`` or ``4`` before
@@ -996,6 +1009,7 @@ async def start(
             queue_storage_queue_rotate_interval_ms=queue_storage_queue_rotate_interval_ms,
             queue_storage_lease_rotate_interval_ms=queue_storage_lease_rotate_interval_ms,
             queue_storage_claim_rotate_interval_ms=queue_storage_claim_rotate_interval_ms,
+            queue_storage_prune_lock_timeout_ms=queue_storage_prune_lock_timeout_ms,
             queue_storage_rotate_min_slot_age_ms=queue_storage_rotate_min_slot_age_ms,
             queue_storage_rotate_target_slot_bytes=queue_storage_rotate_target_slot_bytes,
             queue_storage_rotate_min_slot_rows=queue_storage_rotate_min_slot_rows,

awa-python/src/client.rs

@@ -1796,7 +1796,7 @@ impl PyClient {
         })
     }
 
-    #[pyo3(signature = (queues=None, *, poll_interval_ms=200, global_max_workers=None, completed_retention_hours=None, failed_retention_hours=None, descriptor_retention_days=None, cleanup_batch_size=None, leader_election_interval_ms=None, heartbeat_interval_ms=None, promote_interval_ms=None, heartbeat_rescue_interval_ms=None, heartbeat_staleness_ms=None, deadline_rescue_interval_ms=None, callback_rescue_interval_ms=None, terminal_count_rollup_interval_ms=None, queue_storage_schema=None, queue_storage_queue_slot_count=16, queue_storage_lease_slot_count=8, queue_storage_claim_slot_count=8, queue_storage_queue_stripe_count=1, queue_storage_queue_rotate_interval_ms=1000, queue_storage_lease_rotate_interval_ms=1000, queue_storage_claim_rotate_interval_ms=None, queue_storage_rotate_min_slot_age_ms=None, queue_storage_rotate_target_slot_bytes=None, queue_storage_rotate_min_slot_rows=None, queue_storage_rotate_max_slot_age_ms=None, storage_transition_role=None))]
+    #[pyo3(signature = (queues=None, *, poll_interval_ms=200, global_max_workers=None, completed_retention_hours=None, failed_retention_hours=None, descriptor_retention_days=None, cleanup_batch_size=None, leader_election_interval_ms=None, heartbeat_interval_ms=None, promote_interval_ms=None, heartbeat_rescue_interval_ms=None, heartbeat_staleness_ms=None, deadline_rescue_interval_ms=None, callback_rescue_interval_ms=None, terminal_count_rollup_interval_ms=None, queue_storage_schema=None, queue_storage_queue_slot_count=16, queue_storage_lease_slot_count=8, queue_storage_claim_slot_count=8, queue_storage_queue_stripe_count=1, queue_storage_queue_rotate_interval_ms=1000, queue_storage_lease_rotate_interval_ms=1000, queue_storage_claim_rotate_interval_ms=None, queue_storage_prune_lock_timeout_ms=None, queue_storage_rotate_min_slot_age_ms=None, queue_storage_rotate_target_slot_bytes=None, queue_storage_rotate_min_slot_rows=None, queue_storage_rotate_max_slot_age_ms=None, storage_transition_role=None))]
     #[allow(clippy::too_many_arguments)]
     fn start<'py>(
         &self,
@@ -1824,6 +1824,7 @@ impl PyClient {
         queue_storage_queue_rotate_interval_ms: u64,
         queue_storage_lease_rotate_interval_ms: u64,
         queue_storage_claim_rotate_interval_ms: Option<u64>,
+        queue_storage_prune_lock_timeout_ms: Option<u64>,
         queue_storage_rotate_min_slot_age_ms: Option<u64>,
         queue_storage_rotate_target_slot_bytes: Option<u64>,
         queue_storage_rotate_min_slot_rows: Option<u64>,
@@ -1988,6 +1989,12 @@ impl PyClient {
             }
         }
 
+        if queue_storage_prune_lock_timeout_ms == Some(0) {
+            return Err(pyo3::exceptions::PyValueError::new_err(
+                "queue_storage_prune_lock_timeout_ms must be > 0",
+            ));
+        }
+
         builder = builder.queue_storage(
             QueueStorageConfig {
                 schema: queue_storage_schema
@@ -2009,6 +2016,9 @@ impl PyClient {
         if let Some(ms) = queue_storage_claim_rotate_interval_ms {
             builder = builder.claim_rotate_interval(Duration::from_millis(ms));
         }
+        if let Some(ms) = queue_storage_prune_lock_timeout_ms {
+            builder = builder.prune_lock_timeout(Duration::from_millis(ms));
+        }
         if queue_storage_rotate_min_slot_age_ms.is_some()
             || queue_storage_rotate_target_slot_bytes.is_some()
             || queue_storage_rotate_min_slot_rows.is_some()

awa-python/tests/test_start_config.py

@@ -747,6 +747,13 @@ async def handle(job):
             queue_storage_claim_rotate_interval_ms=0,
         )
 
+    with pytest.raises(ValueError, match="queue_storage_prune_lock_timeout_ms must be > 0"):
+        await client.start(
+            [(queue, 1)],
+            poll_interval_ms=25,
+            queue_storage_prune_lock_timeout_ms=0,
+        )
+
     with pytest.raises(ValueError, match="heartbeat_interval_ms must be > 0"):
         await client.start(
             [(queue, 1)],