diff --git a/CHANGELOG.md b/CHANGELOG.md index f7be9872f..4df2e6521 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/awa-metrics/src/lib.rs b/awa-metrics/src/lib.rs index 1a1c9097b..5241d6740 100644 --- a/awa-metrics/src/lib.rs +++ b/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, /// 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, /// Magnitude of the reason count when prune returns SkippedActive. pub maintenance_prune_skipped_rows: Histogram, @@ -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. diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index e1d203fb1..ef7226acd 100644 --- a/awa-model/src/queue_storage.rs +++ b/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.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) -> Result { diff --git a/awa-python/python/awa/_awa.pyi b/awa-python/python/awa/_awa.pyi index 2219bfc3b..aa1cd7872 100644 --- a/awa-python/python/awa/_awa.pyi +++ b/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, diff --git a/awa-python/python/awa/client.py b/awa-python/python/awa/client.py index f734ca679..733114388 100644 --- a/awa-python/python/awa/client.py +++ b/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, diff --git a/awa-python/src/client.rs b/awa-python/src/client.rs index f141508a7..cf3bc70fd 100644 --- a/awa-python/src/client.rs +++ b/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, + queue_storage_prune_lock_timeout_ms: Option, queue_storage_rotate_min_slot_age_ms: Option, queue_storage_rotate_target_slot_bytes: Option, queue_storage_rotate_min_slot_rows: Option, @@ -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() diff --git a/awa-python/tests/test_start_config.py b/awa-python/tests/test_start_config.py index 302df0fc2..889b72327 100644 --- a/awa-python/tests/test_start_config.py +++ b/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)], diff --git a/awa-worker/src/client.rs b/awa-worker/src/client.rs index 253bbde7a..06e4ecd27 100644 --- a/awa-worker/src/client.rs +++ b/awa-worker/src/client.rs @@ -32,7 +32,7 @@ use tokio::sync::{Mutex, RwLock}; use tokio::task::JoinSet; use tokio_util::sync::CancellationToken; use tokio_util::task::TaskTracker; -use tracing::{info, warn}; +use tracing::{debug, info, warn}; use uuid::Uuid; /// Errors returned when building a worker client. @@ -179,6 +179,7 @@ pub struct ClientBuilder { storage: RuntimeStorage, ring_rotation_cadence: RingRotationCadence, claim_rotate_interval: Option, + prune_lock_timeout: Option, transition_role: TransitionWorkerRole, storage_error: Option, health_addr: Option, @@ -242,6 +243,7 @@ impl ClientBuilder { storage, ring_rotation_cadence: RingRotationCadence::DEFAULT, claim_rotate_interval: None, + prune_lock_timeout: None, transition_role: TransitionWorkerRole::Auto, storage_error, health_addr: None, @@ -978,6 +980,35 @@ impl ClientBuilder { self } + /// Override the PostgreSQL `lock_timeout` the maintenance leader applies to + /// the exclusive-lock steps of queue, lease and claim ring prune and of + /// terminal-delta rollup (default 10 ms). + /// + /// Queue ring prune waits only on its `ready_segments` lock; its other + /// partition locks use `NOWAIT`, so a reader holding one of them makes it + /// `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 records `blocked` + /// in `awa.maintenance.prune.attempts`; the maintenance loop retries on a + /// later tick. The short default keeps the pending `ACCESS EXCLUSIVE` + /// request from stalling claims and enqueues behind it. + /// Raise it when `blocked` outcomes persist and the rings stop recycling + /// because short-lived readers (admin queries, dashboards) repeatedly hold + /// the ring partitions at the moment prune tries. Keep it well below + /// PostgreSQL's `deadlock_timeout` (default 1 s), so that a waiting prune + /// is abandoned before deadlock detection can pick a claim as the victim, + /// and below any `statement_timeout`; [`Self::start`] logs a warning when + /// it is not. A zero timeout, or one above + /// PostgreSQL's `lock_timeout` maximum (`i32::MAX` ms), makes + /// [`Self::build`] fail. Only takes effect when queue storage is active. + /// May be set before or after [`Self::queue_storage`]; retained across + /// storage configuration changes. + pub fn prune_lock_timeout(mut self, timeout: Duration) -> Self { + self.prune_lock_timeout = Some(timeout); + self + } + /// Override when a maintenance tick may seal the current slot of the /// queue, lease and claim rings (default /// [`RingRotationCadence::DEFAULT`]: five seconds of observed filling @@ -1091,6 +1122,11 @@ impl ClientBuilder { if let Some(interval) = self.claim_rotate_interval { runtime.claim_rotate_interval = interval; } + if let Some(timeout) = self.prune_lock_timeout { + runtime + .set_prune_lock_timeout(timeout) + .map_err(|err| BuildError::InvalidQueueStorage(err.to_string()))?; + } } let mut queue_names = HashSet::with_capacity(self.queues.len()); @@ -1680,6 +1716,38 @@ impl Client { reporter.publish_snapshot().await; } + async fn warn_on_unsafe_prune_lock_timeout(&self) { + let Some(runtime) = self.storage.queue_storage() else { + return; + }; + let timeout = runtime.store.prune_lock_timeout(); + let timeout_ms = timeout.as_millis().max(1) as u64; + let risks = match crate::storage::check_prune_lock_timeout(&self.pool, timeout).await { + Ok(risks) => risks, + Err(err) => { + debug!(error = %err, "Could not read lock settings to check the prune lock timeout"); + return; + } + }; + for risk in risks { + match risk { + crate::storage::PruneLockTimeoutRisk::DeadlockTimeout { limit_ms } => warn!( + prune_lock_timeout_ms = timeout_ms, + deadlock_timeout_ms = limit_ms, + "Prune lock timeout is at or above deadlock_timeout; a waiting prune can \ + make PostgreSQL run deadlock detection and cancel a concurrent claim. \ + Keep the timeout well below deadlock_timeout" + ), + crate::storage::PruneLockTimeoutRisk::StatementTimeout { limit_ms } => warn!( + prune_lock_timeout_ms = timeout_ms, + statement_timeout_ms = limit_ms, + "Prune lock timeout is at or above statement_timeout; prune is cancelled \ + by the statement timeout and fails instead of reporting blocked" + ), + } + } + } + async fn log_transition_startup_status( &self, effective_storage: &RuntimeStorage, @@ -1753,6 +1821,7 @@ impl Client { self.log_transition_startup_status(&effective_storage) .await?; + self.warn_on_unsafe_prune_lock_timeout().await; admin::sync_queue_descriptors( &self.pool, @@ -2971,6 +3040,115 @@ mod tests { assert!(client.storage.queue_storage().is_none()); } + #[tokio::test] + async fn prune_lock_timeout_reaches_the_store_in_either_builder_order() { + let timeout = Duration::from_millis(750); + for configure_first in [true, false] { + let mut builder = Client::builder(lazy_pool()).queue("q", QueueConfig::default()); + if configure_first { + builder = builder.prune_lock_timeout(timeout); + } + builder = builder.queue_storage( + QueueStorageConfig::default(), + Duration::from_secs(3), + Duration::from_secs(7), + ); + if !configure_first { + builder = builder.prune_lock_timeout(timeout); + } + let client = builder.build().unwrap(); + let store = client.storage.queue_storage_store().unwrap(); + assert_eq!(store.prune_lock_timeout(), timeout); + } + + let client = Client::builder(lazy_pool()) + .queue("q", QueueConfig::default()) + .build() + .unwrap(); + assert_eq!( + client + .storage + .queue_storage_store() + .unwrap() + .prune_lock_timeout(), + Duration::from_millis(10) + ); + } + + #[tokio::test] + async fn prune_lock_timeout_check_reads_the_session_limits() { + use crate::storage::{check_prune_lock_timeout, PruneLockTimeoutRisk}; + + let url = database_url(); + ensure_database_exists(&url).await; + let pool = PgPoolOptions::new() + .max_connections(1) + .after_connect(|conn, _meta| { + Box::pin(async move { + sqlx::query("SET deadlock_timeout = '200ms'") + .execute(&mut *conn) + .await?; + sqlx::query("SET statement_timeout = '3s'") + .execute(&mut *conn) + .await?; + Ok(()) + }) + }) + .connect(&url) + .await + .expect("connect with session limits"); + + assert!(check_prune_lock_timeout(&pool, Duration::from_millis(100)) + .await + .unwrap() + .is_empty()); + assert_eq!( + check_prune_lock_timeout(&pool, Duration::from_millis(300)) + .await + .unwrap(), + vec![PruneLockTimeoutRisk::DeadlockTimeout { limit_ms: 200 }] + ); + assert_eq!( + check_prune_lock_timeout(&pool, Duration::from_secs(5)) + .await + .unwrap(), + vec![ + PruneLockTimeoutRisk::DeadlockTimeout { limit_ms: 200 }, + PruneLockTimeoutRisk::StatementTimeout { limit_ms: 3000 }, + ] + ); + } + + #[tokio::test] + async fn out_of_range_prune_lock_timeout_is_rejected_at_build() { + for timeout in [Duration::ZERO, Duration::from_millis(i32::MAX as u64 + 1)] { + let result = Client::builder(lazy_pool()) + .queue("q", QueueConfig::default()) + .prune_lock_timeout(timeout) + .build(); + assert!( + matches!(result, Err(BuildError::InvalidQueueStorage(ref message)) if message.contains("prune lock timeout")), + "{timeout:?}: {:?}", + result.err() + ); + } + + let at_maximum = Duration::from_millis(i32::MAX as u64); + let client = Client::builder(lazy_pool()) + .queue("q", QueueConfig::default()) + .prune_lock_timeout(at_maximum) + .build() + .unwrap(); + assert_eq!( + client + .storage + .queue_storage_store() + .unwrap() + .prune_lock_timeout(), + at_maximum + ); + } + #[tokio::test] async fn register_accepts_send_non_sync_handler_future() { #[derive(serde::Serialize, serde::Deserialize, awa_macros::JobArgs)] diff --git a/awa-worker/src/maintenance.rs b/awa-worker/src/maintenance.rs index d84960d37..f1dfe9f79 100644 --- a/awa-worker/src/maintenance.rs +++ b/awa-worker/src/maintenance.rs @@ -2863,6 +2863,7 @@ impl MaintenanceService { if prune_tracker.should_skip(PRUNE_BRANCH_QUEUE) { debug!(branch = PRUNE_BRANCH_QUEUE, "Prune backed off this tick"); + self.metrics.record_prune_backed_off("queue"); return; } @@ -2910,6 +2911,7 @@ impl MaintenanceService { } } Err(err) => { + self.metrics.record_prune_error("queue"); error!(error = %err, "Failed to prune queue storage queue segments"); } } @@ -2978,6 +2980,7 @@ impl MaintenanceService { if prune_tracker.should_skip(PRUNE_BRANCH_LEASE) { debug!(branch = PRUNE_BRANCH_LEASE, "Prune backed off this tick"); + self.metrics.record_prune_backed_off("lease"); return; } @@ -3008,6 +3011,7 @@ impl MaintenanceService { } } Err(err) => { + self.metrics.record_prune_error("lease"); error!(error = %err, "Failed to prune queue storage lease segments"); } } @@ -3081,6 +3085,7 @@ impl MaintenanceService { if prune_tracker.should_skip(PRUNE_BRANCH_CLAIM) { debug!(branch = PRUNE_BRANCH_CLAIM, "Prune backed off this tick"); + self.metrics.record_prune_backed_off("claim"); return; } @@ -3111,6 +3116,7 @@ impl MaintenanceService { } } Err(err) => { + self.metrics.record_prune_error("claim"); error!(error = %err, "Failed to prune queue storage claim segments"); } } diff --git a/awa-worker/src/storage.rs b/awa-worker/src/storage.rs index ac0f4520d..ac9601618 100644 --- a/awa-worker/src/storage.rs +++ b/awa-worker/src/storage.rs @@ -30,6 +30,19 @@ impl QueueStorageRuntime { rotation_cadence: RingRotationCadence::default(), }) } + + /// Set the prune `lock_timeout` on the store. Only valid before the store + /// is shared, which holds for a runtime still owned by the builder. + pub fn set_prune_lock_timeout(&mut self, timeout: Duration) -> Result<(), awa_model::AwaError> { + Arc::get_mut(&mut self.store) + .ok_or_else(|| { + awa_model::AwaError::Validation( + "queue storage is already shared; set the prune lock timeout before build" + .into(), + ) + })? + .set_prune_lock_timeout(timeout) + } } #[derive(Debug, Clone, Default)] @@ -51,3 +64,92 @@ impl RuntimeStorage { self.queue_storage().map(|runtime| runtime.store.clone()) } } + +/// Server settings that a prune `lock_timeout` must stay below. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum PruneLockTimeoutRisk { + /// A prune waiting this long runs PostgreSQL's deadlock detector, which + /// may cancel a concurrent claim or enqueue instead of the prune. + DeadlockTimeout { limit_ms: u64 }, + /// `statement_timeout` cancels the statement first, so the prune fails + /// with an error instead of reporting `blocked`. + StatementTimeout { limit_ms: u64 }, +} + +/// Compare a prune lock timeout with the session's `deadlock_timeout` and +/// `statement_timeout` (milliseconds; a zero `statement_timeout` is disabled). +pub(crate) fn prune_lock_timeout_risks( + timeout: Duration, + deadlock_timeout_ms: Option, + statement_timeout_ms: Option, +) -> Vec { + let timeout_ms = timeout.as_millis().max(1); + let mut risks = Vec::new(); + if let Some(limit_ms) = deadlock_timeout_ms.filter(|limit| *limit > 0) { + if timeout_ms >= u128::from(limit_ms) { + risks.push(PruneLockTimeoutRisk::DeadlockTimeout { limit_ms }); + } + } + if let Some(limit_ms) = statement_timeout_ms.filter(|limit| *limit > 0) { + if timeout_ms >= u128::from(limit_ms) { + risks.push(PruneLockTimeoutRisk::StatementTimeout { limit_ms }); + } + } + risks +} + +/// Read the connection's effective `deadlock_timeout` and `statement_timeout` +/// and report the ones the prune lock timeout reaches. +pub(crate) async fn check_prune_lock_timeout( + pool: &sqlx::PgPool, + timeout: Duration, +) -> Result, sqlx::Error> { + let settings: Vec<(String, i64)> = sqlx::query_as( + "SELECT name, setting::bigint FROM pg_settings \ + WHERE name IN ('deadlock_timeout', 'statement_timeout')", + ) + .fetch_all(pool) + .await?; + let setting = |name: &str| { + settings + .iter() + .find(|(setting_name, _)| setting_name == name) + .and_then(|(_, value)| u64::try_from(*value).ok()) + }; + Ok(prune_lock_timeout_risks( + timeout, + setting("deadlock_timeout"), + setting("statement_timeout"), + )) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn prune_lock_timeout_risks_flag_timeouts_at_or_above_server_limits() { + let below = Duration::from_millis(999); + let at = Duration::from_millis(1000); + assert!(prune_lock_timeout_risks(below, Some(1000), Some(0)).is_empty()); + assert_eq!( + prune_lock_timeout_risks(at, Some(1000), Some(0)), + vec![PruneLockTimeoutRisk::DeadlockTimeout { limit_ms: 1000 }] + ); + assert_eq!( + prune_lock_timeout_risks(Duration::from_secs(30), Some(1000), Some(5000)), + vec![ + PruneLockTimeoutRisk::DeadlockTimeout { limit_ms: 1000 }, + PruneLockTimeoutRisk::StatementTimeout { limit_ms: 5000 }, + ] + ); + assert!(prune_lock_timeout_risks(Duration::from_secs(30), None, None).is_empty()); + assert_eq!( + prune_lock_timeout_risks(Duration::from_nanos(1), Some(1), Some(1)), + vec![ + PruneLockTimeoutRisk::DeadlockTimeout { limit_ms: 1 }, + PruneLockTimeoutRisk::StatementTimeout { limit_ms: 1 }, + ] + ); + } +} diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index b66ad8335..c693f4e30 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -41,6 +41,10 @@ struct RawReceiptClaimRow { claim_slot: i32, } +/// `AwaMetrics::from_global` reads the process-wide meter provider when a +/// client starts, so tests that install their own provider must not overlap. +static GLOBAL_METRICS_TEST_LOCK: Mutex<()> = Mutex::const_new(()); + fn install_in_memory_metrics() -> (InMemoryMetricExporter, SdkMeterProvider) { let exporter = InMemoryMetricExporter::default(); let meter_provider = SdkMeterProvider::builder() @@ -7626,6 +7630,7 @@ async fn test_queue_storage_cancelled_compact_claim_closes_via_batch_and_queue_p #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn test_queue_storage_capacity_wake_drains_after_partial_drain() { + let _metrics_guard = GLOBAL_METRICS_TEST_LOCK.lock().await; let (exporter, meter_provider) = install_in_memory_metrics(); let (_db_guard, pool) = setup_pool(10).await; let queue = "qs_capacity_wake_partial_drain"; @@ -15600,6 +15605,335 @@ async fn test_queue_storage_claimer_heartbeat_skips_fresh_lease() { ); } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_runtime_counts_blocked_then_pruned_lease_prune_attempts() { + let (_db_guard, pool) = setup_pool(10).await; + let queue = "qs_prune_attempt_metric"; + let schema = "awa_qs_prune_attempt_metric"; + let store_config = QueueStorageConfig { + schema: schema.to_string(), + queue_slot_count: 4, + lease_slot_count: 2, + lease_claim_receipts: false, + ..Default::default() + }; + let store = create_store_with_config(&pool, store_config.clone()).await; + + let mut reader_tx = pool.begin().await.expect("begin reader tx"); + sqlx::query(audited_sql(format!( + "LOCK TABLE {schema}.leases_0, {schema}.leases_1 IN ACCESS SHARE MODE" + ))) + .execute(reader_tx.as_mut()) + .await + .expect("reader lock"); + + let _metrics_guard = GLOBAL_METRICS_TEST_LOCK.lock().await; + let (exporter, meter_provider) = install_in_memory_metrics(); + let client = Client::builder(pool.clone()) + .queue( + queue, + QueueConfig { + max_workers: 4, + poll_interval: Duration::from_millis(25), + ..QueueConfig::default() + }, + ) + .queue_storage( + store_config, + Duration::from_secs(60), + Duration::from_millis(50), + ) + .claim_rotate_interval(Duration::from_secs(60)) + .ring_rotation_cadence(RingRotationCadence::EAGER) + .prune_lock_timeout(Duration::from_millis(20)) + .register_worker(CompleteWorker) + .promote_interval(Duration::from_millis(25)) + .leader_election_interval(Duration::from_millis(100)) + .leader_check_interval(Duration::from_millis(50)) + .heartbeat_rescue_interval(Duration::from_millis(100)) + .deadline_rescue_interval(Duration::from_millis(100)) + .callback_rescue_interval(Duration::from_millis(25)) + .build() + .expect("build prune metric client"); + client.start().await.expect("start prune metric client"); + + let job_id = enqueue_job( + &pool, + &store, + &CompleteJob { id: 4001 }, + InsertOpts { + queue: queue.to_string(), + ..Default::default() + }, + ) + .await; + wait_for_job_state( + &store, + &pool, + job_id, + &[JobState::Completed], + Duration::from_secs(15), + ) + .await; + + let lease_attempts = |outcome: &str| { + meter_provider.force_flush().expect("flush metrics"); + let resource_metrics = exporter.get_finished_metrics().expect("read metrics"); + let mut total = 0; + for resource in &resource_metrics { + for scope in resource.scope_metrics() { + for metric in scope.metrics() { + if metric.name() != "awa.maintenance.prune.attempts" { + continue; + } + let AggregatedMetrics::U64(MetricData::Sum(sum)) = metric.data() else { + continue; + }; + total += sum + .data_points() + .filter(|point| { + let has = |key: &str, value: &str| { + point + .attributes() + .any(|kv| kv.key.as_str() == key && kv.value.as_str() == value) + }; + has("awa.ring", "lease") && has("awa.ring.outcome", outcome) + }) + .map(|point| point.value()) + .sum::(); + } + } + } + total + }; + + let deadline = Instant::now() + Duration::from_secs(20); + while lease_attempts("blocked") == 0 { + assert!( + Instant::now() < deadline, + "lease prune never reported blocked while a reader held the lease partition" + ); + tokio::time::sleep(Duration::from_millis(100)).await; + } + assert_eq!(lease_attempts("pruned"), 0); + + reader_tx.rollback().await.expect("release reader lock"); + let deadline = Instant::now() + Duration::from_secs(20); + while lease_attempts("pruned") == 0 { + assert!( + Instant::now() < deadline, + "lease prune never completed after the reader released" + ); + tokio::time::sleep(Duration::from_millis(100)).await; + } + + client.shutdown(Duration::from_secs(5)).await; +} + +#[derive(Clone, Copy, Debug)] +enum PruneRing { + Queue, + Lease, + Claim, +} + +async fn prune_ring( + store: &QueueStorage, + pool: &sqlx::PgPool, + ring: PruneRing, +) -> Result { + match ring { + PruneRing::Queue => store.prune_oldest(pool, Duration::ZERO).await, + PruneRing::Lease => store.prune_oldest_leases(pool).await, + PruneRing::Claim => store.prune_oldest_claims(pool).await, + } +} + +/// A reader holds `locked_relation` so prune must wait for its exclusive lock. +/// With a short timeout prune gives up as `Blocked` only after that timeout; +/// with a long timeout it waits out a reader that releases part-way and then +/// completes. +async fn assert_prune_waits_for_configured_lock_timeout( + pool: &sqlx::PgPool, + config: &QueueStorageConfig, + ring: PruneRing, + locked_relation: &str, +) { + let short_timeout = Duration::from_millis(300); + let short_store = QueueStorage::new(config.clone()) + .expect("short-timeout store") + .with_prune_lock_timeout(short_timeout) + .expect("short prune lock timeout"); + assert_eq!(short_store.prune_lock_timeout(), short_timeout); + + let mut reader_tx = pool.begin().await.expect("begin reader tx"); + sqlx::query(audited_sql(format!( + "LOCK TABLE {locked_relation} IN ACCESS SHARE MODE" + ))) + .execute(reader_tx.as_mut()) + .await + .expect("reader lock"); + + let started = Instant::now(); + let blocked = prune_ring(&short_store, pool, ring) + .await + .expect("prune under reader lock"); + let elapsed = started.elapsed(); + assert!( + matches!(blocked, PruneOutcome::Blocked { .. }), + "{ring:?} prune must report Blocked while the reader holds its lock, got {blocked:?}" + ); + assert!( + elapsed >= short_timeout, + "{ring:?} prune gave up after {elapsed:?}, before the {short_timeout:?} lock timeout" + ); + assert!( + elapsed < Duration::from_secs(10), + "{ring:?} prune waited {elapsed:?} for a {short_timeout:?} lock timeout" + ); + + let long_store = QueueStorage::new(config.clone()) + .expect("long-timeout store") + .with_prune_lock_timeout(Duration::from_secs(60)) + .expect("long prune lock timeout"); + let reader_hold = Duration::from_millis(600); + let release = tokio::spawn(async move { + tokio::time::sleep(reader_hold).await; + reader_tx.rollback().await.expect("release reader lock"); + }); + let started = Instant::now(); + let outcome = prune_ring(&long_store, pool, ring) + .await + .expect("prune after reader release"); + let elapsed = started.elapsed(); + release.await.expect("release task"); + assert!( + matches!(outcome, PruneOutcome::Pruned { .. }), + "{ring:?} prune with a long lock timeout must outwait the reader and finish, got {outcome:?}" + ); + assert!( + elapsed >= short_timeout, + "{ring:?} prune returned after {elapsed:?} instead of waiting for the reader" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_queue_prune_lock_timeout_is_configurable() { + let (_db_guard, pool) = setup_pool(8).await; + let schema = "awa_qs_prune_lt_queue"; + let store = create_store(&pool, schema).await; + store.enqueue_batch(&pool, "q", 1, 1).await.unwrap(); + let claims = store.claim_batch(&pool, "q", 1).await.unwrap(); + assert_eq!(store.complete_batch(&pool, &claims).await.unwrap(), 1); + assert!(matches!( + store.rotate(&pool).await.unwrap(), + RotateOutcome::Rotated { .. } + )); + + let config = QueueStorageConfig { + schema: schema.to_string(), + queue_slot_count: 4, + lease_slot_count: 2, + lease_claim_receipts: false, + ..Default::default() + }; + assert_prune_waits_for_configured_lock_timeout( + &pool, + &config, + PruneRing::Queue, + &format!("{schema}.ready_segments_0"), + ) + .await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_lease_prune_lock_timeout_is_configurable() { + let (_db_guard, pool) = setup_pool(8).await; + let schema = "awa_qs_prune_lt_lease"; + let store = create_store(&pool, schema).await; + store.enqueue_batch(&pool, "q", 1, 2).await.unwrap(); + let claimed = store + .claim_runtime_batch(&pool, "q", 2, Duration::from_secs(60)) + .await + .unwrap(); + assert_eq!(claimed.len(), 2); + store.complete_runtime_batch(&pool, &claimed).await.unwrap(); + let cadence = RingRotationCadence::EAGER + .with_min_slot_age(Duration::from_secs(3600)) + .with_target_slot_bytes(1 << 30) + .with_min_slot_rows(0) + .with_max_slot_age(Duration::ZERO); + assert!(matches!( + store + .rotate_leases_with_cadence(&pool, cadence) + .await + .unwrap(), + RotateOutcome::Rotated { slot: 1, .. } + )); + + let config = QueueStorageConfig { + schema: schema.to_string(), + queue_slot_count: 4, + lease_slot_count: 2, + lease_claim_receipts: false, + ..Default::default() + }; + assert_prune_waits_for_configured_lock_timeout( + &pool, + &config, + PruneRing::Lease, + &format!("{schema}.leases_0"), + ) + .await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_claim_prune_lock_timeout_is_configurable() { + let (_db_guard, pool) = setup_pool(8).await; + let schema = "awa_qs_prune_lt_claim"; + let config = QueueStorageConfig { + schema: schema.to_string(), + queue_slot_count: 4, + lease_slot_count: 2, + claim_slot_count: 4, + lease_claim_receipts: true, + ..Default::default() + }; + create_store_with_config(&pool, config.clone()).await; + sqlx::query(audited_sql(format!( + r#" + INSERT INTO {schema}.lease_claims ( + claim_slot, job_id, run_lease, ready_slot, ready_generation, + queue, priority, attempt, max_attempts, lane_seq + ) VALUES (0, 1001, 1, 0, 0, 'synthetic', 2, 1, 25, 1001) + "# + ))) + .execute(&pool) + .await + .expect("seed claim"); + sqlx::query(audited_sql(format!( + "INSERT INTO {schema}.lease_claim_closures (claim_slot, job_id, run_lease, outcome) \ + VALUES (0, 1001, 1, 'completed')" + ))) + .execute(&pool) + .await + .expect("seed closure"); + sqlx::query(audited_sql(format!( + "INSERT INTO {schema}.claim_ring_rotations (generation, slot) VALUES (1, 1)" + ))) + .execute(&pool) + .await + .expect("advance claim ring cursor away from slot 0"); + + assert_prune_waits_for_configured_lock_timeout( + &pool, + &config, + PruneRing::Claim, + &format!("{schema}.lease_claims_0"), + ) + .await; +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn test_queue_storage_prune_oldest_blocks_on_reader_lock() { let (_db_guard, pool) = setup_pool(10).await; diff --git a/docs/configuration.md b/docs/configuration.md index 930bc6bde..f4e36389f 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -548,6 +548,7 @@ let client = Client::builder(pool.clone()) Duration::from_millis(1_000), ) .claim_rotate_interval(Duration::from_millis(1_000)) + .prune_lock_timeout(Duration::from_millis(10)) .build()?; ``` @@ -564,6 +565,7 @@ await client.start( queue_storage_queue_rotate_interval_ms=1000, queue_storage_lease_rotate_interval_ms=1000, queue_storage_claim_rotate_interval_ms=1000, + queue_storage_prune_lock_timeout_ms=10, ) ``` @@ -579,11 +581,12 @@ await client.start( | `queue_rotate_interval` | `1000ms` | How often ready/terminal segments rotate | | `lease_rotate_interval` | `1000ms` | How often lease segments rotate | | `claim_rotate_interval` | matches `queue_rotate_interval` | How often the ADR-023 claim-ring rotates. Set with `ClientBuilder::claim_rotate_interval` (Rust) or `queue_storage_claim_rotate_interval_ms` (Python). Test harnesses sometimes set this to a long interval to pin claim-ring layout for deterministic count assertions. | +| `prune_lock_timeout` | `10ms` | The PostgreSQL `lock_timeout` the maintenance leader applies when it takes the exclusive locks to reclaim a queue, lease or claim ring partition (and when it folds terminal-count deltas); among queue prune's initial partition locks, only `ready_segments` waits under it; later reads and writes also run under it. The timeout applies separately to each lock acquisition, not to the whole prune. A lock timeout rolls the prune back, is counted as `blocked`, and is retried on a later maintenance tick. Set with `ClientBuilder::prune_lock_timeout` (Rust) or `queue_storage_prune_lock_timeout_ms` (Python). Must be greater than zero and at most `2147483647` ms; keep it well below `deadlock_timeout` and `statement_timeout`. The short default keeps prune's pending `ACCESS EXCLUSIVE` request from stalling claims and enqueues queued behind it. See below for when to raise it. | | `ring_rotation_cadence` | 5 s / 2 MiB target / 256 rows / 60 s | A non-empty slot becomes eligible after `min_slot_age` since the first non-empty observation once it holds `min_slot_rows`, when `target_slot_bytes` is reached, or after `max_slot_age` regardless of rows (so a trickle of writes neither seals a nearly empty slot every window nor keeps one open indefinitely). Python: `queue_storage_rotate_min_slot_rows`, `queue_storage_rotate_max_slot_age_ms`. Counts total physical storage, including indexes and TOAST. Empty slots bypass the cadence. The minimum age uses a monotonic clock and resets on an empty observation or leader change; `max_slot_age` also counts from when the slot opened on the database clock, so a leader change does not restart it, except under compat (pre-flip) authority when an older rotator left no ledger row for the slot. `max_slot_age` makes a slot eligible to seal; the busy and reuse checks still decide when it is reclaimed. Configure with `ClientBuilder::ring_rotation_cadence` (Rust) or `queue_storage_rotate_min_slot_age_ms` / `queue_storage_rotate_target_slot_bytes` (Python). A zero byte target disables the size trigger, a zero `min_slot_rows` disables the row floor, and a zero `max_slot_age` disables the ceiling. A row floor needs a size target or a ceiling, and a queue-storage client refuses to build without one, because lease rows are deleted as attempts finish and a lease slot may never reach the floor. A zero `min_slot_age`, or `RingRotationCadence::EAGER`, permits sealing on every tick whatever the other settings. The target is sampled and does not bound retained storage. | The benchmark harness in [postgresql-job-queue-benchmarking](https://github.com/hardbyte/postgresql-job-queue-benchmarking) reads `QUEUE_SLOT_COUNT`, `LEASE_SLOT_COUNT`, `CLAIM_SLOT_COUNT`, `QUEUE_STRIPE_COUNT`, and `LEASE_CLAIM_RECEIPTS` from the environment. Those env vars are benchmark configuration, not general worker-runtime configuration. -Rust builder overrides for cadence and claim rotation interval may be set before +Rust builder overrides for cadence, claim rotation interval and prune lock timeout may be set before or after `queue_storage`; reconfiguring storage preserves those overrides. With no claim interval override, the claim ring uses the queue decision interval. @@ -593,6 +596,7 @@ Use the defaults unless you have a reason not to: - Increase `lease_slot_count` if lease churn is high enough that dead tuples in the lease ring stop collapsing promptly. - Increase `claim_slot_count` if the rotation cadence (`claim_rotate_interval`) plus the slot count combine to a partition retention window shorter than your longest in-flight zero-deadline short job; running out of empty slots forces `rotate_claims` to return `SkippedBusy` and the receipt-plane churn falls back onto a smaller working set of partitions. - Increase `queue_stripe_count` only for measured workloads where many small replicas contend on the same logical queue. This is queue-storage configuration, not a per-queue `QueueConfig` field: it gives each logical queue the same number of physical stripes. Striping spreads a hot logical queue over `queue#N` physical coordination paths, but it weakens perfect global ordering and can regress calmer shapes if overused. For single hot-queue workloads, benchmark `2` stripes first; higher values should be treated as workload-specific tuning rather than a general default. +- Increase `prune_lock_timeout` when `increase(awa_maintenance_prune_attempts_total{awa_ring_outcome="blocked"}[15m])` stays above zero and the rings stop recycling, typically because frequent short reads of the ring partitions (admin or dashboard queries over the queue-storage tables) hold a conflicting lock at the instant prune tries. A stalled ring grows its current slot, which makes those queries slower and the contention worse. Among its initial partition locks, queue ring prune waits under the timeout only on `ready_segments` and takes the others with `NOWAIT`, so a longer timeout does not help when readers hold those. Later reads and writes also run under the timeout. Raise it in small steps (for example `50ms`, then `100ms`, then `250ms`) and keep it well below PostgreSQL's `deadlock_timeout` (default `1s`) and below any `statement_timeout` on the pool's connections. A single lock wait that reaches `deadlock_timeout` makes PostgreSQL run its deadlock detector, and if claims or enqueues are part of a lock cycle with it, the detector can cancel them rather than the prune; a short `lock_timeout` abandons the prune first, so the prune is always the one that gives way. A `statement_timeout` at or below the lock timeout cancels the prune statement before the lock timeout fires, so it fails with an error instead of reporting `blocked`. Startup logs a warning when the configured timeout is at or above either setting. While prune waits, new accesses to the same partition also queue behind its lock request, so a longer timeout trades claim latency for reclamation. Shortening the readers is the more durable fix when they run longer than a few hundred milliseconds. Watch `awa_maintenance_prune_attempts_total` by `awa_ring` and `awa_ring_outcome`: `pruned` should advance steadily, `blocked` means the lock timeout expired, `skipped_active` (with `awa_ring_reason`) means live rows kept the slot, `backed_off` means prune was skipped for a few ticks after repeated `skipped_active`, and `error` is a database failure. - Increase rotation intervals to reduce partition churn and metadata activity. - Decrease rotation intervals to tighten dead-tuple bounds at the cost of more frequent rotate/prune work. diff --git a/docs/deployment.md b/docs/deployment.md index 5d1123cfa..dd26f6da5 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -82,6 +82,7 @@ Recommended practice: - monitor `pg_stat_activity` for long transactions - monitor `pg_stat_user_tables` on the active queue-storage schema; ready, tombstone, done, and terminal-delta segments should stay near zero dead tuples, lease segments may spike within the rotation window but should collapse after prune, and `attempt_state` should roughly track live long-running attempts - tune autovacuum for the database if the lease tables churn heavily +- monitor `increase(awa_maintenance_prune_attempts_total{awa_ring_outcome=~"blocked|error"}[15m])` for contention or failures, alongside successful `pruned` outcomes, ring generations and accumulated rows. A positive increase can persist while recycling succeeds. If recycling has stalled, inspect long transactions and conflicting locks; for short readers, consider [`prune_lock_timeout`](configuration.md#what-the-knobs-mean) and its effect on waiting queries Queue-storage depth metrics are designed for high-cadence monitoring. The worker records `available` from lane-head cursor differences and probes queue lag from the next claimable row per lane, instead of running the exact `queue_counts` ready-row scan on every metrics tick. Use the admin API or CLI when you need exact queue counts for an operator action. diff --git a/skills/awa-operations/SKILL.md b/skills/awa-operations/SKILL.md index 4cfb039b0..25975c603 100644 --- a/skills/awa-operations/SKILL.md +++ b/skills/awa-operations/SKILL.md @@ -290,6 +290,25 @@ Rings rotate only when active — a frozen generation on an idle queue is health The condition to alert on is a frozen generation *with* accumulating rows (`skipped_busy`): a pinned ring, almost always caused by a long-held snapshot. +## Prune Lock Timeout + +Ring prune installs its `lock_timeout` after the initial slot-row locks and +pre-lock checks. From then on, each lock acquisition has its own timeout of +10 ms by default; this does not bound the whole prune. A lock timeout rolls it +back and counts `blocked` in +`awa_maintenance_prune_attempts_total{awa_ring,awa_ring_outcome}`. Sustained +`increase()` of `blocked` (or `error`) warrants checking `pruned` outcomes, +ring generations and accumulating rows; successful prunes can occur between +blocked attempts. Raise +`ClientBuilder::prune_lock_timeout` (Python +`queue_storage_prune_lock_timeout_ms`) in small steps (`50ms`, `100ms`, `250ms`) when short readers of the ring +partitions are the cause, and keep it well below `deadlock_timeout` (default 1 s) +and any enabled `statement_timeout`. Startup warns when it is at or above either +enabled limit. Of its initial partition locks, queue-ring prune waits under the +timeout only on `ready_segments` and takes its other partitions with `NOWAIT`. +`backed_off` means prune was skipped for a few ticks +after repeated `skipped_active`. + ## Ring Rotation Cadence For queue storage, `ClientBuilder::ring_rotation_cadence` seals a slot after a